Resolve daily runtime handoffs and isolate notification imports

ops/quant-monitor/README.md

@@ -123,7 +123,9 @@ python3 scripts/consume_daily_briefing.py --runtime-projection projection.json -
 
 默认只预览。`--dispatch` 才发送，并按平台、业务日和投影中的目标集合复用既有逐目标送达；同一目标集合换顺序、改观察时间或改正文不会另发。这个目标集合是否等于生产固定配置，还要云端接线验收。`--dry-run` / `--send-dry-run` 不联网、不写送达状态。投影里的成交笔数尚未接通，不能读成零成交。这不是云端真实日报，也不关闭各平台原来的通知。
 
-GCS 输入默认关闭，也不改 22:30 UTC 的简报定时。只有 `QUANT_MONITOR_RUNTIME_DIGEST_ENABLED=true`，并且同时配置合法的 `QUANT_MONITOR_RUNTIME_PROJECTION_PREFIX`（末段必须是 `runtime_daily`）、IANA `QUANT_MONITOR_RUNTIME_TIMEZONE` 和 `QUANT_MONITOR_RUNTIME_EXPECTED_TARGET_KEY`（scope 末段为 `paper`）时，pipeline 才按该时区的业务日读取 `{prefix}/longbridge/paper/{day}.json`。日期和目标键来自这三项配置，不从对象内容反推。读取用现有 `gcloud storage cat`，20 秒超时、正文上限 1MiB、不重试；缺对象、无权限、超时或非法内容都不发送。发送仍是原来的 `--dispatch`、量化哨兵和 `health_cycle.json`。domain 简报照常单独执行：它失败不会跳过 runtime，runtime 失败也不会取消已跑的 domain；任一项失败则 pipeline 以非零结束。日志不写正文、对象 URI 或 chat id。本地 `--runtime-projection` 仍然可用，并与 `--runtime-projection-gcs` 互斥。VPS 是否已安装该开关、云端身份是否能读该对象、以及首份送达，都还没有验收。
+GCS 输入默认关闭，也不改 22:30 UTC 的简报定时。只有 `QUANT_MONITOR_RUNTIME_DIGEST_ENABLED=true`，并且同时配置合法的 `QUANT_MONITOR_RUNTIME_PROJECTION_PREFIX`（末段必须是 `runtime_daily`）、IANA `QUANT_MONITOR_RUNTIME_TIMEZONE` 和 `QUANT_MONITOR_RUNTIME_EXPECTED_TARGET_KEY`（scope 末段为 `paper`）时，pipeline 才按该时区的业务日读取固定指针 `{prefix}/longbridge/paper/{day}/handoff.json`。指针必须只有 `schema_version`、`business_date`、`object_uri` 三项，版本为 `runtime_daily_handoff.v1`；对象 URI 必须落在同一 approved prefix 和同一业务日目录，且文件名为 producer 的 UTC 观察时间格式。consumer 随后只读取该精确 immutable object，再复用原投影校验、`--dispatch` 和 `health_cycle.json`。不会列目录、挑 `latest`，也不会退回旧 `{day}.json`；缺失或无效指针以失败结束，不发送。已有成日对象由 producer 的 create-only handoff 保持首个指针；consumer ledger 已记录当日目标后，同一日报或同日重跑不会重复发送。只有同时提供 `QUANT_MONITOR_RUNTIME_PROJECTION_OBJECT` 与 `QUANT_MONITOR_RUNTIME_BUSINESS_DAY` 才使用显式对象覆盖模式；二者必须成对且对象须在相同 prefix/日期目录。读取使用现有 `gcloud storage cat`，20 秒超时、正文上限 1MiB、不重试。domain 简报照常单独执行：它失败不会跳过 runtime，runtime 失败也不会取消已跑的 domain；任一项失败则 pipeline 以非零结束。日志不写正文、对象 URI 或 chat id。本地 `--runtime-projection` 仍然可用，并与 GCS 输入互斥。VPS 是否已安装该版本、云端身份能否读取 handoff 与 immutable object、以及真实日报送达，都还没有验收。
+
+GCS input remains disabled by default and does not change the 22:30 UTC briefing timer. When enabled with a valid `QUANT_MONITOR_RUNTIME_PROJECTION_PREFIX` (ending in `runtime_daily`), IANA `QUANT_MONITOR_RUNTIME_TIMEZONE`, and paper-scoped `QUANT_MONITOR_RUNTIME_EXPECTED_TARGET_KEY`, the pipeline reads the fixed `{prefix}/longbridge/paper/{day}/handoff.json` for the business date in that timezone. The pointer must contain exactly `schema_version`, `business_date`, and `object_uri`, with schema `runtime_daily_handoff.v1`. Its object URI must stay under the same approved prefix and business-day directory, with the producer's UTC-observation filename format. The consumer then reads only that immutable object and reuses the existing projection validation, `--dispatch`, and `health_cycle.json` ledger. It never lists objects, selects `latest`, or falls back to the old `{day}.json`; a missing or invalid pointer fails without sending. The producer's create-only handoff preserves the first pointer for a day, and the consumer ledger prevents duplicate delivery on same-day reruns. The explicit-object mode is used only when both `QUANT_MONITOR_RUNTIME_PROJECTION_OBJECT` and `QUANT_MONITOR_RUNTIME_BUSINESS_DAY` are supplied; they must be paired and identify an object under the configured prefix and day. Reads use the existing `gcloud storage cat` client with a 20-second timeout, 1 MiB body limit, and no retry. Domain briefing continues independently; either branch failing makes the pipeline exit nonzero. Logs do not include the body, object URI, or chat ID. The local `--runtime-projection` mode remains available and is mutually exclusive with GCS input. Installation on the VPS, cloud identity access to the handoff and immutable object, and real-report delivery are not yet verified.
 
 ### 一次性固定对象日报 / One-shot immutable-object runtime briefing
 

ops/quant-monitor/quant_monitor_domain/briefing_dispatch.py

@@ -12,12 +12,13 @@
 import urllib.parse
 import urllib.request
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
-from quant_platform_kit.strategy_lifecycle.watch.runner import dispatch_strategy_watch_findings
 from .briefing_consumer import BriefingAction, BriefingConsumptionResult, BriefingFinding
 from .runtime_digest import prepare_runtime_digest
-from quant_platform_kit.strategy_lifecycle.watch.strategy_watch import StrategyWatchFinding, build_strategy_monitoring_finding
+
+if TYPE_CHECKING:
+    from quant_platform_kit.strategy_lifecycle.watch.strategy_watch import StrategyWatchFinding
 
 _REPOSITORY_RE = re.compile(
     r"[A-Za-z0-9][A-Za-z0-9_-]*(?:\.[A-Za-z0-9_-]+)*/"
@@ -296,17 +297,35 @@ def shutil_which(name: str) -> str | None:
     return which(name)
 
 
+def dispatch_strategy_watch_findings(*args: Any, **kwargs: Any) -> dict[str, Any]:
+    from quant_platform_kit.strategy_lifecycle.watch.runner import (
+        dispatch_strategy_watch_findings as dispatch,
+    )
+
+    return dispatch(*args, **kwargs)
+
+
 def _strategy_monitoring_findings(
     result: BriefingConsumptionResult,
 ) -> list[StrategyWatchFinding]:
-    findings: list[StrategyWatchFinding] = []
-    for finding in result.findings:
+    candidates = [
+        finding
+        for finding in result.findings
         if (
-            finding.level != BriefingAction.GITHUB_ISSUE
-            or finding.kind != "strategy_monitoring"
-            or not finding.strategy_profile
-        ):
-            continue
+            finding.level == BriefingAction.GITHUB_ISSUE
+            and finding.kind == "strategy_monitoring"
+            and finding.strategy_profile
+        )
+    ]
+    if not candidates:
+        return []
+
+    from quant_platform_kit.strategy_lifecycle.watch.strategy_watch import (
+        build_strategy_monitoring_finding,
+    )
+
+    findings: list[StrategyWatchFinding] = []
+    for finding in candidates:
         findings.append(
             build_strategy_monitoring_finding(
                 domain=finding.domain,

ops/quant-monitor/scripts/daily_briefing_pipeline.sh

@@ -95,16 +95,17 @@ except Exception:
     echo "[briefing-pipeline] runtime digest rejected: runtime_consumer_unavailable" >&2
     runtime_status=2
   else
-    runtime_object="${trimmed}/longbridge/paper/${runtime_day}.json"
     if [[ -n "$runtime_object_override" ]]; then
       runtime_day="$runtime_day_override"
-      runtime_object="$runtime_object_override"
+      runtime_source=(--runtime-projection-gcs "$runtime_object_override")
+    else
+      runtime_source=(--runtime-handoff-prefix "$trimmed")
     fi
     set +e
     (
       cd "$AAB"
       PYTHONPATH=. python3 scripts/consume_daily_briefing.py \
-        --runtime-projection-gcs "$runtime_object" \
+        "${runtime_source[@]}" \
         --day "$runtime_day" \
         --expected-target-key "$runtime_key" \
         --dispatch >/dev/null

ops/quant-monitor/tests/test_runtime_daily_pipeline.py

@@ -9,7 +9,7 @@
 import os
 import subprocess
 import sys
-from datetime import datetime, timezone
+from datetime import date, datetime, timedelta, timezone
 from pathlib import Path
 from unittest.mock import patch
 from zoneinfo import ZoneInfo
@@ -33,7 +33,7 @@ def _layout(tmp_path: Path) -> tuple[Path, Path, Path]:
             "import os, sys",
             "with open(os.environ['CONSUME_LOG'], 'a', encoding='utf-8') as handle:",
             "    handle.write('\\t'.join(sys.argv[1:]) + '\\n')",
-            "if '--runtime-projection-gcs' in sys.argv:",
+            "if '--runtime-projection-gcs' in sys.argv or '--runtime-handoff-prefix' in sys.argv:",
             "    raise SystemExit(int(os.environ.get('RUNTIME_CONSUME_EXIT', '0')))",
             "raise SystemExit(int(os.environ.get('DOMAIN_CONSUME_EXIT', '0')))",
             "",
@@ -90,7 +90,7 @@ def test_runtime_switch_off_does_not_call_gcs(tmp_path: Path) -> None:
     assert "gs://" not in result.stdout + result.stderr
 
 
-def test_enabled_pipeline_uses_timezone_day_and_fixed_paper_object(tmp_path: Path) -> None:
+def test_enabled_pipeline_uses_timezone_day_and_fixed_handoff_prefix(tmp_path: Path) -> None:
     result = _run(
         tmp_path,
         QUANT_MONITOR_RUNTIME_DIGEST_ENABLED="true",
@@ -106,8 +106,8 @@ def test_enabled_pipeline_uses_timezone_day_and_fixed_paper_object(tmp_path: Pat
     domain, runtime = calls
     assert "--runtime-projection-gcs" not in domain
     assert domain[domain.index("--day") + 1] == utc
-    object_uri = runtime[runtime.index("--runtime-projection-gcs") + 1]
-    assert object_uri == f"gs://proj-bucket/runtime_daily/longbridge/paper/{zoned}.json"
+    handoff_prefix = runtime[runtime.index("--runtime-handoff-prefix") + 1]
+    assert handoff_prefix == "gs://proj-bucket/runtime_daily"
     assert runtime[runtime.index("--day") + 1] == zoned
     assert runtime[runtime.index("--expected-target-key") + 1] == "lb-paper|rot|paper"
     assert "--dispatch" in runtime
@@ -163,7 +163,7 @@ def test_domain_failure_still_runs_runtime_and_returns_nonzero(tmp_path: Path) -
     )
     assert result.returncode == 1
     assert len(_calls(result)) == 2
-    assert "--runtime-projection-gcs" in _calls(result)[1]
+    assert "--runtime-handoff-prefix" in _calls(result)[1]
     assert "domain_exit=4 runtime_exit=0" in result.stderr
     assert "gs://" not in result.stderr
 
@@ -180,7 +180,7 @@ def test_runtime_failure_does_not_skip_domain(tmp_path: Path) -> None:
     assert result.returncode == 1
     calls = _calls(result)
     assert "--report-dir" in calls[0]
-    assert "--runtime-projection-gcs" in calls[1]
+    assert "--runtime-handoff-prefix" in calls[1]
     assert "domain_exit=0 runtime_exit=5" in result.stderr
 
 
@@ -256,6 +256,7 @@ def _install_real_main_fixture(root: Path, aab: Path) -> None:
     }), encoding="utf-8")
     real_root = SCRIPT.parents[3]
     wrapper = r'''import contextlib, json, os, sys
+from datetime import datetime
 from unittest.mock import patch
 def guard(event, args):
     if event in {"socket.connect", "socket.getaddrinfo", "subprocess.Popen", "os.system", "os.posix_spawn", "os.exec"}:
@@ -273,10 +274,30 @@ def read_fixture(uri):
     args = sys.argv[1:]
     day = args[args.index("--day") + 1]
     key = args[args.index("--expected-target-key") + 1]
+    with open(os.environ['CONSUME_LOG'] + '.reads', 'a', encoding='utf-8') as handle:
+        handle.write(uri + '\n')
+    if uri.endswith('/handoff.json'):
+        prefix = args[args.index("--runtime-handoff-prefix") + 1]
+        observed = os.environ.get("RUNTIME_FIXTURE_OBSERVED_AT", day + "T00:00:00Z")
+        stamp = datetime.fromisoformat(observed.replace("Z", "+00:00")).strftime("%Y%m%dT%H%M%S%fZ")
+        object_uri = f"{prefix}/longbridge/paper/{day}/{stamp}.json"
+        handoff = {"schema_version": "runtime_daily_handoff.v1", "business_date": day, "object_uri": object_uri}
+        bad_handoff = os.environ.get("RUNTIME_FIXTURE_BAD_HANDOFF", "")
+        if bad_handoff == "wrong_day":
+            handoff["business_date"] = "2000-01-01"
+        elif bad_handoff == "wrong_prefix":
+            handoff["object_uri"] = object_uri.replace("synthetic", "other")
+        elif bad_handoff == "legacy_object":
+            handoff["object_uri"] = f"{prefix}/longbridge/paper/{day}.json"
+        elif bad_handoff == "nested_path":
+            handoff["object_uri"] = f"{prefix}/longbridge/paper/{day}/runtime_daily/longbridge/paper/{day}/{stamp}.json"
+        elif bad_handoff == "extra_field":
+            handoff["extra"] = "ignored fields are forbidden"
+        return json.dumps(handoff).encode(), None
     service, strategy, scope = key.split("|")
     with open(os.environ['CONSUME_LOG'], 'a', encoding='utf-8') as handle:
         handle.write('\t'.join(sys.argv[1:]) + '\n')
-    payload = {"platform": "longbridge", "observed_at": os.environ.get("RUNTIME_FIXTURE_OBSERVED_AT", day + "T00:00:00+00:00"), "completeness": "complete", "records": [{
+    payload = {"platform": "longbridge", "observed_at": os.environ.get("RUNTIME_FIXTURE_OBSERVED_AT", day + "T00:00:00Z"), "completeness": "complete", "records": [{
         "target_key": key, "target": {"service": service, "strategy_profile": strategy, "account_scope": scope},
         "business_date": day, "timezone": "UTC", "status": "market_closed", "completeness": "complete",
         "execution_lane": "paper", "runs": [], "conflicts": [], "fills": {"source": "not_connected", "records": [], "count": None},
@@ -352,6 +373,52 @@ def test_immutable_pipeline_uses_real_cli_and_ledger_after_process_restart(tmp_p
     assert "gs://" not in first.stdout + first.stderr + second.stdout + second.stderr
 
 
+def test_timezone_day_handoff_consumes_exact_immutable_object_and_reuses_ledger(tmp_path: Path) -> None:
+    import json
+    business_day = datetime.now(ZoneInfo(ZONE)).date().isoformat()
+    observed_day = (date.fromisoformat(business_day) - timedelta(days=1)).isoformat()
+    env = dict(QUANT_MONITOR_RUNTIME_DIGEST_ENABLED="true",
+               QUANT_MONITOR_RUNTIME_PROJECTION_PREFIX="gs://synthetic/runtime_daily",
+               QUANT_MONITOR_RUNTIME_TIMEZONE=ZONE,
+               QUANT_MONITOR_RUNTIME_EXPECTED_TARGET_KEY="synthetic-service|rot|paper",
+               RUNTIME_FIXTURE_OBSERVED_AT=f"{observed_day}T23:40:00Z", RUNTIME_REAL_LEDGER="true",
+               TELEGRAM_TOKEN="synthetic", GLOBAL_TELEGRAM_CHAT_ID="synthetic-chat")
+    first = _run(tmp_path, real_main=True, **env)
+    assert "runtime_exit=0" in first.stderr, first.stderr
+    reads = (tmp_path / "consume.log.reads").read_text().splitlines()
+    assert reads == [
+        f"gs://synthetic/runtime_daily/longbridge/paper/{business_day}/handoff.json",
+        f"gs://synthetic/runtime_daily/longbridge/paper/{business_day}/{observed_day.replace('-', '')}T234000000000Z.json",
+    ]
+    state = tmp_path / "monitor/data/alert-state/health_cycle.json"
+    before = state.read_bytes()
+    second = _run(tmp_path, real_main=True, **env)
+    assert "runtime_exit=0" in second.stderr, second.stderr
+    assert state.read_bytes() == before
+    assert (tmp_path / "consume.log.send").read_text().splitlines() == ["synthetic-send"]
+    assert (tmp_path / "consume.log.reads").read_text().splitlines() == reads + reads
+
+
+def test_invalid_handoff_fails_closed_without_legacy_day_fallback(tmp_path: Path) -> None:
+    business_day = datetime.now(ZoneInfo("UTC")).date().isoformat()
+    for index, failure in enumerate(("wrong_day", "wrong_prefix", "legacy_object", "nested_path", "extra_field")):
+        case = tmp_path / str(index)
+        result = _run(case, real_main=True,
+                      QUANT_MONITOR_RUNTIME_DIGEST_ENABLED="true",
+                      QUANT_MONITOR_RUNTIME_PROJECTION_PREFIX="gs://synthetic/runtime_daily",
+                      QUANT_MONITOR_RUNTIME_TIMEZONE="UTC",
+                      QUANT_MONITOR_RUNTIME_EXPECTED_TARGET_KEY="synthetic-service|rot|paper",
+                      RUNTIME_FIXTURE_BAD_HANDOFF=failure, RUNTIME_REAL_LEDGER="true",
+                      TELEGRAM_TOKEN="synthetic", GLOBAL_TELEGRAM_CHAT_ID="synthetic-chat")
+        assert result.returncode == 1
+        assert "reason=invalid_runtime_handoff" in result.stderr
+        assert (case / "consume.log.reads").read_text().splitlines() == [
+            f"gs://synthetic/runtime_daily/longbridge/paper/{business_day}/handoff.json",
+        ]
+        assert not (case / "consume.log.send").exists()
+        assert not (case / "monitor/data/alert-state/health_cycle.json").exists()
+
+
 def test_immutable_pipeline_wrong_body_observation_rejects_before_send(tmp_path: Path) -> None:
     result = _run(tmp_path, real_main=True,
                   QUANT_MONITOR_RUNTIME_DIGEST_ENABLED="true",