diff --git a/application/runtime_notification_adapters.py b/application/runtime_notification_adapters.py index e1502273..bcc12210 100644 --- a/application/runtime_notification_adapters.py +++ b/application/runtime_notification_adapters.py @@ -11,10 +11,8 @@ from notifications.order_alerts import ( OrderLifecycleEvent, monitor_submitted_order_status, - publish_order_lifecycle_event, - is_filled_status, is_partial_filled_status, is_terminal_error_status, + is_filled_status, is_verified_partial_fill, is_terminal_error_status, ) -from notifications.telegram import build_issue_notifier from quant_platform_kit.common.quantity import format_quantity from quant_platform_kit.common.port_adapters import CallableNotificationPort from quant_platform_kit.common.ports import NotificationPort @@ -90,10 +88,10 @@ def send_recorded_message(message: str) -> bool: log_message=log_message or (lambda message: print(with_prefix(message), flush=True)), send_message=send_recorded_message, ) - legacy_notify_issue = build_issue_notifier( - with_prefix_fn=with_prefix, - send_tg_message_fn=send_recorded_message, - ) + def legacy_notify_issue(title, detail): + from notifications.renderers import render_runtime_error + # Raw broker detail belongs to controlled diagnostics, never the body. + return send_recorded_message(render_runtime_error(kind="real_error")) def issue_with_ack(title, detail): before = len(recorded_delivery_events) @@ -119,20 +117,22 @@ def publish_order_event(event: OrderLifecycleEvent) -> None: from notifications import unified def publish(): before = len(recorded_delivery_events) - outcome = publish_order_lifecycle_event(event, translator=translator, publisher=order_event_publisher) + from notifications.renderers import render_runtime_error + outcome = order_event_publisher.publish(RenderedNotification( + detailed_text="", compact_text=render_runtime_error(kind="order_status_unknown"))) if not unified.enabled(): return outcome events = recorded_delivery_events[before:] return bool(events) and all(item.get("transport_acknowledged") is True for item in events) - if unified.enabled() and not (is_filled_status(event.status) or is_partial_filled_status(event.status)): - from notifications.digest_snapshot.unified_contract import order_identity_digest - unified.dispatch_immediate(publish, kind="order_status_unknown", - incident=f"order_status:{order_identity_digest(event.order_id)}", state=event.status, - severity=2 if is_terminal_error_status(event.status) else 1) - else: - if unified.enabled() and is_filled_status(event.status): - unified.record_order_resolution(event) - publish() + if is_filled_status(event.status) or is_verified_partial_fill(event): + # P10 applies in either flag state. Persist before suppressing send. + if not unified.record_order_resolution(event): + notify_issue("Order notification evidence unavailable", "Candidate persistence is unconfirmed.") + return + from notifications.digest_snapshot.unified_contract import order_identity_digest + unified.dispatch_immediate(publish, kind="order_status_unknown", + incident=f"order_status:{order_identity_digest(event.order_id)}", state=event.status, + severity=2 if is_terminal_error_status(event.status) else 1) def post_submit_order(trade_context, order_intent, report) -> None: from notifications import unified diff --git a/notifications/order_alerts.py b/notifications/order_alerts.py index 6cc3d60e..6e57863c 100644 --- a/notifications/order_alerts.py +++ b/notifications/order_alerts.py @@ -5,6 +5,7 @@ from collections.abc import Callable from dataclasses import dataclass import traceback +from decimal import Decimal, InvalidOperation from notifications.events import NotificationPublisher, RenderedNotification @@ -19,6 +20,7 @@ class OrderLifecycleEvent: executed_qty: str = "0" executed_price: str = "0" reason: str = "" + status_verified: bool = False def is_filled_status(status): @@ -29,6 +31,17 @@ def is_partial_filled_status(status): return "PartialFilled" in status +def is_verified_partial_fill(event: OrderLifecycleEvent) -> bool: + """Only a broker observation with consistent quantities proves processing.""" + if not event.status_verified or not is_partial_filled_status(event.status) or event.reason: + return False + try: + executed, requested, price = map(Decimal, (event.executed_qty, str(event.quantity), event.executed_price)) + return all(value.is_finite() for value in (executed, requested, price)) and 0 < executed < requested and price > 0 + except (InvalidOperation, ValueError, TypeError): + return False + + def is_terminal_error_status(status): return any(keyword in status for keyword in ["Rejected", "Canceled", "Expired"]) @@ -47,6 +60,7 @@ def build_order_lifecycle_event( executed_qty="0", executed_price="0", reason="", + status_verified=False, ): return OrderLifecycleEvent( symbol=str(symbol), @@ -57,6 +71,7 @@ def build_order_lifecycle_event( executed_qty=str(executed_qty or "0"), executed_price=str(executed_price or "0"), reason=str(reason or ""), + status_verified=status_verified, ) @@ -163,7 +178,6 @@ def monitor_submitted_order_status( return try: - published_any_status = False for _ in range(order_poll_max_attempts): sleeper(order_poll_interval_sec) order_status = fetch_order_status(trade_context, order_id) @@ -183,6 +197,7 @@ def monitor_submitted_order_status( executed_qty=executed_qty, executed_price=executed_price, reason=reject_message, + status_verified=True, ) if is_filled_status(status): @@ -191,22 +206,21 @@ def monitor_submitted_order_status( if is_partial_filled_status(status): publish_order_event(event) - published_any_status = True if is_terminal_error_status(status): publish_order_event(event) return - if not published_any_status: - publish_order_event( - build_order_lifecycle_event( - symbol, - side_text, - quantity, - order_id, - "StatusCheckTimeout", - reason="poll_timeout", - ) + # A partial observation does not prove the eventual terminal outcome. + publish_order_event( + build_order_lifecycle_event( + symbol, + side_text, + quantity, + order_id, + "StatusCheckTimeout", + reason="poll_timeout", ) + ) except Exception: notify_issue( "Order status poll failed", diff --git a/notifications/renderers.py b/notifications/renderers.py index e49a33d2..03509100 100644 --- a/notifications/renderers.py +++ b/notifications/renderers.py @@ -536,3 +536,22 @@ def _heartbeat_outcome_key(execution, *, skip_logs, note_logs) -> str: if bool(execution.get("direct_live_routing_blocked")): return "heartbeat_execution_blocked" return "no_executable_orders" if (skip_logs or note_logs) else "no_trades" + + +def render_runtime_error(*, kind="real_error", occurred_at=None, report=None, locale=None): + """Adapt runtime facts to the step-1 shared formatter; never render diagnostics. + + The QPK step-1 revision must be pinned before deployment. Monitor time is + not a business-run observation; missing producer links/counts stay unknown. + """ + import os + from datetime import datetime, timezone + from quant_platform_kit.notifications.renderer_base import render_asset_management_error + from quant_platform_kit.notifications.unified_contract import NotificationEvent + + report = report or {} + stamp = occurred_at or datetime.now(timezone.utc) + event = NotificationEvent(kind, stamp, platform_id="longbridge", + strategy_profile=report.get("strategy_profile"), + incident_reason=kind) + return render_asset_management_error(event, locale=locale or os.environ.get("NOTIFY_LANG") or "zh-CN") diff --git a/notifications/unified.py b/notifications/unified.py index 7dd598b2..0ebfe8b5 100644 --- a/notifications/unified.py +++ b/notifications/unified.py @@ -541,11 +541,17 @@ def record_order_resolution(order_event): try: digest = order_identity_digest(order_event.order_id) fact = make_event("fill", order_execution_digest=digest) - persist_candidate(fact.to_candidate(), actually_ran=False) + if persist_candidate(fact.to_candidate(), actually_ran=False) is not True: + raise CandidatePersistenceError("notification_candidate_write_unconfirmed") + from .order_alerts import is_filled_status + if not is_filled_status(order_event.status): + return True for stream in ("order_status", "order_status_poll"): observe_incident(replace(fact, incident_reason=f"{stream}:{digest}"), state="filled", recovered=True, severity=0) + return True except Exception: record_failure("order_notification_evidence_unconfirmed") + return False def recover_runtime(*, report=None, route_label=None, occurred_at=None, run_id=None, probe=False): diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index a4b75799..e0fe6b3c 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -1186,13 +1186,8 @@ def _notify_normal_heartbeat(name: str, detail: str, *, observed_at=None, checke def _notify_heartbeat_alert(message, *, now, kind, inspected_count=0): if not unified_notifications.enabled(): return _send_telegram(message) - locale = _notification_locale() - issue = (_notice("heartbeat_no_recent_report", lookback_hours=os.environ.get("RUNTIME_HEARTBEAT_LOOKBACK_HOURS") or "36") - if kind == "receipt_missing" else _notice("heartbeat_no_acceptable_report", count=inspected_count)) - # Keep raw diagnostics in the existing logs; use the shared localized - # template without private target names, archive URIs or mixed raw reasons. - compact = format_operational_alert(locale=locale, alert_type="execution_report_heartbeat", - name="Execution receipts" if locale == "en" else "执行回执", context={}, issues=[issue]) + from notifications.renderers import render_runtime_error + compact = render_runtime_error(kind=kind, occurred_at=now, locale=_notification_locale()) return unified_notifications.dispatch_immediate(lambda: _send_telegram(compact[:3900]), kind=kind, incident="execution_receipt", occurred_at=now, run_id=os.environ.get("GITHUB_RUN_ID"), actually_ran=False) diff --git a/scripts/paper_observation/core.py b/scripts/paper_observation/core.py index 370861f6..af7578d2 100644 --- a/scripts/paper_observation/core.py +++ b/scripts/paper_observation/core.py @@ -470,9 +470,24 @@ def __init__(self, store, send, calendar, start, activation, verification=False) if verification: self.send = lambda message: send("[verification] " + message) + def record_routine(self, root, kind, incident, now): + # Observation facts are not business cycles or delivery receipts. Keep + # the existing append-only store, explicitly pending central ingestion. + record = {"event_kind": kind, "occurred_at": now.isoformat(), + "incident_reason": incident, "actually_ran": False, + "cycle_count": 0, "order_count": None, "fill_count": None, + "delivery_status": "pending_central"} + key = f"{root}/digest/{kind}.json" + if not self.store.create(key, record) and not self.store.get(key): + raise ObservationError("storage_unavailable") + def notify(self, incident, active, now): prefix = "verification/" if self.verification else "" root = f"alerts/{prefix}{incident}" + if incident in {f"calendar/coverage_{days}" for days in (90, 30, 7)}: + if active: + self.record_routine(root, "maintenance_reminder", incident, now) + return if active and not self.store.get(f"{root}/ack.json"): claim = f"{root}/claims/{slot(now)}.json" if self.store.create(claim, {"at": now.isoformat()}) and self.send( @@ -491,10 +506,8 @@ def notify(self, incident, active, now): and self.store.get(f"{root}/ack.json") and not self.store.get(f"{root}/recovered.json") ): - if self.store.create( - f"{root}/recover-claims/{slot(now)}.json", {"at": now.isoformat()} - ) and self.send(f"PAPER observation recovery: {incident}"): - self.store.create(f"{root}/recovered.json", {"at": now.isoformat()}) + self.record_routine(root, "recovery", incident, now) + self.store.create(f"{root}/recovered.json", {"at": now.isoformat()}) def run(self, now): if now < self.activation: @@ -544,8 +557,7 @@ def run(self, now): pass # Storage failure must not depend on storage-based deduplication. code = "storage_unavailable" - self.send( - f"PAPER observation alert: {now.date()}/{code}; persistent dedup unavailable" - ) + from notifications.renderers import render_runtime_error + self.send(render_runtime_error(kind="real_error", occurred_at=now)) return code return "checked" diff --git a/tests/test_order_alerts.py b/tests/test_order_alerts.py index 112d0bb0..1bda88da 100644 --- a/tests/test_order_alerts.py +++ b/tests/test_order_alerts.py @@ -68,7 +68,7 @@ def test_monitor_submitted_order_status_emits_lifecycle_events_via_callback(): sleeper=lambda _seconds: None, ) - assert len(events) == 1 + assert [event.status for event in events] == ["PartialFilled", "StatusCheckTimeout"] assert events[0].symbol == "SOXL.US" assert events[0].status == "PartialFilled" assert events[0].executed_qty == "4" @@ -91,7 +91,7 @@ def test_monitor_submitted_order_status_emits_unconfirmed_event_after_poll_windo sleeper=lambda _seconds: None, ) - assert len(events) == 1 + assert [event.status for event in events] == ["StatusCheckTimeout"] assert events[0].symbol == "SOXL.US" assert events[0].status == "StatusCheckTimeout" assert events[0].reason == "poll_timeout" diff --git a/tests/test_p10_notification_routing.py b/tests/test_p10_notification_routing.py new file mode 100644 index 00000000..a209a065 --- /dev/null +++ b/tests/test_p10_notification_routing.py @@ -0,0 +1,173 @@ +"""P10 callback and PAPER routing regressions; isolated stores and fake senders.""" +from datetime import date, datetime, timezone +from types import SimpleNamespace + +import pytest + +from application.runtime_notification_adapters import build_runtime_notification_adapters +from notifications import unified +from notifications.order_alerts import build_order_lifecycle_event, monitor_submitted_order_status +from notifications.telegram import build_translator +from scripts.paper_observation.core import Watchdog + +NOW = datetime(2026, 11, 4, 22, tzinfo=timezone.utc) + + +def adapters(fetch, sent): + return build_runtime_notification_adapters(with_prefix=lambda text: text, + send_message=lambda text: sent.append(text) or True, + translator=build_translator("en"), fetch_order_status=fetch, + order_poll_interval_sec=0, order_poll_max_attempts=2, sleeper=lambda _: None, + log_message=lambda _: None) + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_filled_callback_never_sends_but_persists(monkeypatch, enabled): + sent, facts = [], [] + monkeypatch.setattr(unified, "enabled", lambda: enabled) + monkeypatch.setattr(unified, "record_order_resolution", lambda event: facts.append(event) or True) + runner = adapters(lambda *_: dict(status="Filled", reason="", executed_qty="10", executed_price="12"), sent) + runner.post_submit_order(None, SimpleNamespace(symbol="FAKE.US", side="buy", quantity=10), + SimpleNamespace(broker_order_id="fake-order")) + assert not sent and len(facts) == 1 + + +def test_verified_partial_then_filled_never_sends(monkeypatch): + sent, facts = [], [] + statuses = iter([dict(status="PartialFilled", reason="", executed_qty="4", executed_price="12"), + dict(status="Filled", reason="", executed_qty="10", executed_price="12")]) + monkeypatch.setattr(unified, "record_order_resolution", lambda event: facts.append(event) or True) + runner = adapters(lambda *_: next(statuses), sent) + runner.post_submit_order(None, SimpleNamespace(symbol="FAKE.US", side="buy", quantity=10), + SimpleNamespace(broker_order_id="fake-order")) + assert not sent and len(facts) == 2 + + +def test_partial_does_not_hide_timeout(): + events = [] + monitor_submitted_order_status(None, "FAKE.US", "Buy", 10, "fake-order", + fetch_order_status=lambda *_: dict(status="PartialFilled", reason="", executed_qty="4", executed_price="12"), + order_poll_interval_sec=0, order_poll_max_attempts=1, publish_order_event=events.append, + notify_issue=lambda *_: pytest.fail("unexpected poll exception"), sleeper=lambda _: None) + assert [event.status for event in events] == ["PartialFilled", "StatusCheckTimeout"] + + +@pytest.mark.parametrize("quantity", ["NaN", "11", "0", "bad"]) +def test_inconsistent_partial_stays_error(monkeypatch, quantity): + sent, facts = [], [] + monkeypatch.setattr(unified, "enabled", lambda: False) + monkeypatch.setattr(unified, "record_order_resolution", lambda event: facts.append(event) or True) + runner = adapters(lambda *_: dict(status="PartialFilled", reason="", executed_qty=quantity, executed_price="12"), sent) + runner.post_submit_order(None, SimpleNamespace(symbol="FAKE.US", side="buy", quantity=10), + SimpleNamespace(broker_order_id="fake-order")) + assert sent and not facts + + +def test_candidate_failure_is_visible(monkeypatch): + sent = [] + monkeypatch.setattr(unified, "enabled", lambda: False) + monkeypatch.setattr(unified, "record_order_resolution", lambda event: False) + runner = adapters(lambda *_: dict(status="Filled", reason="", executed_qty="10", executed_price="12"), sent) + runner.post_submit_order(None, SimpleNamespace(symbol="FAKE.US", side="buy", quantity=10), + SimpleNamespace(broker_order_id="fake-order")) + assert len(sent) == 1 and "【资产管理｜运行错误】" in sent[0] + + +class Store: + def __init__(self): + self.data = {} + + def get(self, key): + return self.data.get(key) + + def create(self, key, value): + if key in self.data: + return False + self.data[key] = value + return True + + +def test_paper_recovery_and_maintenance_are_persisted_without_send(): + store, sent = Store(), [] + watch = Watchdog(store, lambda text: sent.append(text) or True, None, date(2026, 11, 4), NOW) + watch.notify("fake-incident", True, NOW) + assert len(sent) == 1 + watch.notify("fake-incident", False, NOW) + watch.notify("fake-incident", False, NOW) + watch.notify("calendar/coverage_7", True, NOW) + assert len(sent) == 1 + assert store.get("alerts/fake-incident/recovered.json") + records = [value for value in store.data.values() if value.get("delivery_status") == "pending_central"] + assert {value["event_kind"] for value in records} == {"recovery", "maintenance_reminder"} + assert all(value["actually_ran"] is False and value["cycle_count"] == 0 for value in records) + + +@pytest.mark.parametrize("locale,title", [("zh-CN", "【资产管理｜运行错误】"), ("en", "[Asset management | Runtime error]")]) +def test_fixed_error_body_omits_diagnostics_and_unknown_counts(locale, title): + from notifications.renderers import render_runtime_error + text = render_runtime_error(kind="order_status_unknown", occurred_at=NOW, + report={"strategy_profile": "fake-private-strategy", "account_id": "fake-private-account", + "error": "fake-private-diagnostic"}, locale=locale) + assert text.startswith(title) + assert "fake-private" not in text + assert ("未知" in text) if locale == "zh-CN" else ("unknown" in text) + + +def test_recovery_write_failure_does_not_claim_recovery(): + class FailingStore(Store): + def create(self, key, value): + if "/digest/" in key: + raise RuntimeError("fake write failure") + return super().create(key, value) + store = FailingStore() + watch = Watchdog(store, lambda _: True, None, NOW.date(), NOW) + watch.notify("fake-incident", True, NOW) + with pytest.raises(RuntimeError): + watch.notify("fake-incident", False, NOW) + assert not store.get("alerts/fake-incident/recovered.json") + + +class FileStore: + def __init__(self, root): + self.root = root + + def get(self, key): + import json + path = self.root / key + return json.loads(path.read_text()) if path.exists() else None + + def create(self, key, value): + import json + path = self.root / key + path.parent.mkdir(parents=True, exist_ok=True) + try: + with path.open("x") as output: + json.dump(value, output) + return True + except FileExistsError: + return False + + +def _recover_in_process(root): + def send(text): + with (root / "sent.txt").open("a") as output: + output.write(text + "\n") + return True + watch = Watchdog(FileStore(root), send, None, NOW.date(), NOW) + watch.notify("fake-incident", False, NOW) + + +def test_recovery_survives_restart_and_two_processes(tmp_path): + import multiprocessing + store = FileStore(tmp_path) + store.create("alerts/fake-incident/ack.json", {"sender_ack": True}) + context = multiprocessing.get_context("fork") + workers = [context.Process(target=_recover_in_process, args=(tmp_path,)) for _ in range(2)] + for worker in workers: + worker.start() + for worker in workers: + worker.join(10) + assert worker.exitcode == 0 + _recover_in_process(tmp_path) + assert not (tmp_path / "sent.txt").exists() + assert store.get("alerts/fake-incident/digest/recovery.json")["delivery_status"] == "pending_central" diff --git a/tests/test_paper_cloud_observation.py b/tests/test_paper_cloud_observation.py index 2e1d00e5..6b551167 100644 --- a/tests/test_paper_cloud_observation.py +++ b/tests/test_paper_cloud_observation.py @@ -199,8 +199,10 @@ def test_watchdog_missing_recovery_and_dedup(): watch.run(NOW + timedelta(minutes=91)) watch.run(NOW + timedelta(minutes=106)) assert ( - len([m for m in messages if "recovery" in m and "missing_snapshot" in m]) == 1 + len([m for m in messages if "recovery" in m and "missing_snapshot" in m]) == 0 ) + assert store.get("alerts/2026-11-02/missing_snapshot/recovered.json") + assert store.get("alerts/2026-11-02/missing_snapshot/digest/recovery.json")["delivery_status"] == "pending_central" assert store.get("alerts/watch-heartbeat/20261102T2245Z.json") diff --git a/tests/test_runtime_notification_adapters.py b/tests/test_runtime_notification_adapters.py index 568ec4c3..39011008 100644 --- a/tests/test_runtime_notification_adapters.py +++ b/tests/test_runtime_notification_adapters.py @@ -27,7 +27,10 @@ def test_runtime_notification_adapters_publish_cycle_notification_uses_log_and_s assert messages == ["compact line"] -def test_runtime_notification_adapters_post_submit_order_publishes_order_events(): +def test_runtime_notification_adapters_post_submit_order_publishes_order_events(monkeypatch): + from notifications import unified + facts = [] + monkeypatch.setattr(unified, "record_order_resolution", lambda event: facts.append(event) or True) messages = [] adapters = build_runtime_notification_adapters( with_prefix=lambda message: f"[HK] {message}", @@ -51,7 +54,8 @@ def test_runtime_notification_adapters_post_submit_order_publishes_order_events( SimpleNamespace(broker_order_id="order-1"), ) - assert messages == ["✅ Order Filled | SOXL Buy 10 shares avg $123.45 (ID: order-1)"] + assert messages == [] + assert len(facts) == 1 and facts[0].status == "Filled" def test_runtime_notification_adapter_records_negative_ack_as_failed():