fix: Preserve daily caller presentation and run evidence

docs/unified-daily-notification.md

@@ -70,7 +70,27 @@ Every daily report lists IBKR, Schwab, Firstrade, LongBridge, Binance and the
 management console. An explicitly unbound source says it is not included and
 its run state is unknown. A missing inventory says coverage is unknown. This
 display does not bind a new source, collect another platform, or make unknown
-orders/fills zero. Existing totals cover bound targets only.
+orders/fills zero. Existing totals cover bound targets only. The central Cloud Run caller presents
+an unbound Binance inventory as disabled under the standing owner OFF decision:
+runs, orders and fills are N/A. This display-only decision does not change any
+switch or collect data. A missing inventory remains unknown; a bound Binance
+target retains its observed coverage rather than being hidden by the OFF label.
+Other unbound platforms retain unknown run state.
+
+The central caller and its deadline fallback use the same five sections,
+including missing-source and verified-idle reports. Coverage details stay inside
+the Coverage section. The deadline fallback keeps trades and assets unknown and
+retains the existing single reservation and no-follow-up behavior.
+
+For each actual source run, a supplied `source_run_id` may already be a full
+Actions run URL. The renderer accepts that URL only when it matches the
+platform's configured repository and the canonical HTTPS GitHub Actions run
+path; query and fragment are stripped by the shared URL helper. It never
+constructs a URL from a numeric or opaque run ID. Missing or rejected links
+produce an explicit unavailable marker; optional links may be omitted under
+the existing evidence budget. Run identity, counts and receipt fields are
+unchanged. An accepted link is navigation evidence, not a new claim that the
+run succeeded, remains active or delivered a message.
 
 Verified idle reports retain the business window, independently known counts,
 source coverage and next action. Their result says “No trades in verified

python/scripts/cloud_run_daily_digest.py

@@ -21,7 +21,7 @@
 import notification_candidate_adapter as adapter
 from daily_digest_delivery_store import GCSDeliveryStore, StoreUnavailable, location
 from display_names import configured_account_label, display_name
-from unified_daily_digest import _safe_display_text, render
+from unified_daily_digest import _safe_display_text, _render_consolidated as render
 
 SGT = ZoneInfo('Asia/Singapore')
 IDENTITY = adapter.IDENTITY[:4]
@@ -213,14 +213,13 @@ def _report(configuration, rows, outcomes, day, platform_config):
             trailer.append(f'另有 {len(details) - index} 项缺口' if zh else f'{len(details) - index} more gaps')
             break
         trailer.append(detail)
-    base = render(payload, max_units=max(256, 3850 - _units('\n'.join(trailer))))
-    idle = '😴 今日全部账号无交易' if zh else '😴 No trades across all accounts today'
-    if coverage == 'complete' and (not base or base == idle):
-        return idle, coverage
-    if not base:
-        base = (f'每日交易摘要 · {day}\n未收录实际运行；订单及成交统计未知。' if zh else
-                f'Daily trading digest · {day}\nNo observed runs; order and fill counts unknown.')
-    return base + '\n' + '\n'.join(trailer), coverage
+    # Presentation-only standing owner OFF, applied solely to an unbound
+    # Binance inventory. A contradictory binding retains its real coverage.
+    text = render(payload, max_units=3850, known_disabled_platforms={"binance"},
+                  source_repositories={key: value.get("repository")
+                                       for key, value in config.get("platforms", {}).items()},
+                  coverage_details=trailer)
+    return text, coverage
 
 
 def execute(headers, configuration, store, send, *, preview=False, now=utc_now,

python/scripts/daily_digest_aggregator.py

@@ -934,7 +934,10 @@ def build_digest_payload(
         notification_summary=collected.notification_summary,
     )
     from unified_daily_digest import enabled, render
-    return payload, render(payload) if enabled() else digest.render_daily_digest(payload), collected
+    text = (render(payload, source_repositories={key: value.get("repository")
+                                               for key, value in config.get("platforms", {}).items()})
+            if enabled() else digest.render_daily_digest(payload))
+    return payload, text, collected
 
 
 def build_receipt(

python/scripts/daily_digest_deadline.py

@@ -67,6 +67,22 @@ def expired(_signal, _frame):
 
 
 def failure_text(env, *, is_late, locale='zh'):
+    from notification_mode import consolidation_enabled
+    if consolidation_enabled():
+        from notification_snapshot.presentation import render_operator_message
+        zh = not str(locale).startswith('en')
+        return render_operator_message(
+            title=(f"⚠️ QRS 日报{' late' if is_late else ''}｜{env['DIGEST_REPORT_DATE']}" if zh else
+                   f"QRS daily digest{' late' if is_late else ''} | {env['DIGEST_REPORT_DATE']}"),
+            locale='zh' if zh else 'en',
+            result=['本次日报未完成验证，已停止等待上游；成交与资产状态未知。' if zh else
+                    'The digest was not verified in time. Upstream waiting has stopped. Trades and asset state remain unknown.'],
+            scope=[f"业务日：{env['DIGEST_BUSINESS_DAY']}；截止：07:30 Asia/Singapore。" if zh else
+                   f"Business day: {env['DIGEST_BUSINESS_DAY']}; deadline: 07:30 Asia/Singapore."],
+            evidence=['运行链接：不可用' if zh else 'Run link: unavailable'],
+            coverage=['来源覆盖未核验' if zh else 'Source coverage unverified'],
+            next_step=['本窗口仅发送本条通知，不再自动补发。' if zh else
+                       'No automatic follow-up will be sent in this window.'], max_units=3900)
     title = ' late' if is_late else ''
     if str(locale).startswith('en'):
         return (f"QRS daily digest{title} | {env['DIGEST_REPORT_DATE']}\n"

python/scripts/unified_daily_digest.py

@@ -6,7 +6,7 @@
 from datetime import date, datetime, time
 from notification_mode import consolidation_enabled, routine_mode
 from notification_snapshot.unified_contract import SGT, sgt_business_window
-from notification_snapshot.presentation import next_step_text, render_operator_message
+from notification_snapshot.presentation import next_step_text, render_operator_message, safe_run_url
 from display_names import broker_account_identity, display_name, runtime_account_names, safe_display_text as _safe_display_text
 from daily_digest_notify import DailyDigestInput, _presentation_needs_attention, _trade_counts, block_reason_text, counts_not_connected, is_blocked, normalize_locale, routine_decision_text, routine_observation_text
 
@@ -27,11 +27,11 @@ def _window_text(business_day: str, locale: str) -> str:
     return f"统计区间：{interval}" if locale == "zh" else f"Reporting window: {interval}"
 
 
-def render(payload: DailyDigestInput, *, max_units: int = 3900) -> str:
+def render(payload: DailyDigestInput, *, max_units: int = 3900, source_repositories=None) -> str:
     if max_units < 256:
         raise ValueError("digest_budget_too_small")
     if consolidation_enabled():
-        return _render_consolidated(payload, max_units=max_units)
+        return _render_consolidated(payload, max_units=max_units, source_repositories=source_repositories)
     trades, problems = [], []
     for row in payload.runs:
         label = " / ".join(" ".join(str(v).split())[:70] for v in
@@ -261,7 +261,9 @@ def _account_observations(rows, locale):
     return lines
 
 
-def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
+def _render_consolidated(payload: DailyDigestInput, *, max_units: int,
+                         known_disabled_platforms=(), source_repositories=None,
+                         coverage_details=()) -> str:
     """Flat account blocks for trading activity; retain evidence diagnostics."""
     locale = normalize_locale(payload.locale)
     zh = locale == "zh"
@@ -293,6 +295,10 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
             inventory.append(f"{label}：来源覆盖未知" if zh else f"{label}: source coverage unknown")
             continue
         expected, covered, runs = source["expected_targets"], source["covered_targets"], source["runs"]
+        if not expected and platform in known_disabled_platforms:
+            inventory.append(f"{label}：已停用；运行／订单／成交不适用" if zh else
+                             f"{label}: disabled; runs/orders/fills N/A")
+            continue
         if not expected:
             inventory.append(f"{label}：未纳入本期汇总，运行状态未知" if zh else
                          f"{label}: not included in this digest; run state unknown")
@@ -376,8 +382,9 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
                        list(dict.fromkeys(diagnostic_observations)), list(dict.fromkeys(diagnostics))))
     blocks.sort(key=lambda item: not item[0])
     coverage_lines = ["统计范围：已绑定目标；缺失统计保留未知" if zh else
-                      "Counts cover bound targets; missing counts remain unknown"]
+                      "Bound targets; missing is unknown"]
     coverage_lines.extend(inventory)
+    coverage_lines.extend(coverage_details)
     evidence, optional_evidence = [], []
     for _, block, optional, required in blocks:
         context = "\n".join(required or block)
@@ -386,6 +393,24 @@ def _render_consolidated(payload: DailyDigestInput, *, max_units: int) -> str:
                                   if required else []) + optional)
         if required:
             required_diagnostics.append(context)
+    # A run ID is not a URL. Only retain a supplied canonical Actions link
+    # belonging to this platform's configured repository; never build one.
+    links = []
+    for row, (platform, account, strategy) in zip(payload.runs, _run_display_contexts(payload.runs, locale, runtime_accounts=True)):
+        url = safe_run_url(row.source_run_id)
+        repository = (source_repositories or {}).get(row.platform_id)
+        if (url and isinstance(repository, str)
+                and re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository)
+                and re.fullmatch(r"https://github\.com/" + re.escape(repository) + r"/actions/runs/[1-9][0-9]*", url)):
+            links.append(f"{platform} / {account} / {strategy}: {url}")
+    link_status = (("运行链接：部分不可用" if links else "运行链接：不可用") if zh else
+                   ("Run link: partially unavailable" if links else "Run link: unavailable"))
+    if not payload.runs or len(links) < len(payload.runs):
+        evidence.append(link_status)
+        optional_evidence.append(())
+        required_diagnostics.append(link_status)
+    evidence.extend(links)
+    optional_evidence.extend(() for _ in links)
     needs_review = (not complete or any(priority for priority, _, _, _ in blocks) or bool(immediate)
                     or any(not _known_count(row, name) for row in payload.runs for name in ("order", "fill")))
     result = [("证据需核对" if needs_review else "已收录窗口记录") if zh else

python/tests/fixtures/daily-caller-en.txt

@@ -0,0 +1,21 @@
+📊 Daily trading digest · <PROJECT_ID>
+[Result]
+😴 No trades in verified targets
+Window observations recorded
+Observed runs in window: 1
+Orders: 0 · Fills: 0
+[Scope]
+🕒 Reporting window: <PROJECT_ID> 06:00 – <PROJECT_ID> 06:00 (UTC+8)
+[Evidence]
+Run link: unavailable
+[Coverage]
+Bound targets; missing is unknown
+IBKR: not included in this digest; run state unknown
+Schwab: Included runs: 1; expected target coverage 1/1
+Firstrade: not included in this digest; run state unknown
+LongBridge: not included in this digest; run state unknown
+Binance: disabled; runs/orders/fills N/A
+Management console: not included in this digest; run state unknown
+Data coverage: complete
+[Next step]
+No action required; await the next reporting window.

python/tests/fixtures/daily-caller-zh.txt

@@ -0,0 +1,21 @@
+📊 每日交易摘要 · <PROJECT_ID>
+【结果】
+😴 已核验目标无交易
+已收录窗口记录
+已收录运行 1 次
+订单：0 · 成交：0
+【范围】
+🕒 统计区间：<PROJECT_ID> 06:00 – <PROJECT_ID> 06:00 (UTC+8)
+【证据】
+运行链接：不可用
+【覆盖】
+统计范围：已绑定目标；缺失统计保留未知
+IBKR：未纳入本期汇总，运行状态未知
+Schwab：已收录 1 次运行，目标覆盖 1/1
+Firstrade：未纳入本期汇总，运行状态未知
+LongBridge：未纳入本期汇总，运行状态未知
+Binance：已停用；运行／订单／成交不适用
+管理站：未纳入本期汇总，运行状态未知
+数据覆盖：完整
+【下一步】
+无需操作；等待下一统计窗口。

python/tests/test_cloud_run_daily_digest.py

@@ -182,7 +182,7 @@ def test_one_missing_target_keeps_valid_target_without_a_fake_global_zero(self):
         self.assertIn('全窗口统计未知', self.sent[0])
         self.assertIn('未找到当日报告', self.sent[0])
         self.assertNotIn('Schwab·', self.sent[0])
-        self.assertIn('Binance：未纳入本期汇总，运行状态未知', self.sent[0])
+        self.assertIn('Binance：已停用；运行／订单／成交不适用', self.sent[0])
         self.assertNotIn('Binance：运行 0', self.sent[0])
 
     def test_invalid_or_unreadable_target_does_not_discard_another_platform(self):

python/tests/test_daily_caller_presentation.py

@@ -0,0 +1,108 @@
+"""Source-only caller regressions; no network, credentials or real sends."""
+from dataclasses import replace
+import unittest
+import re
+from pathlib import Path
+
+import test_cloud_run_daily_digest as caller_fixture
+from test_cloud_run_daily_digest import EXPECTED, ROW
+from test_daily_digest_ux import config
+from test_notification_consolidation import candidate
+import cloud_run_daily_digest as app
+from notification_snapshot.presentation import SECTION_NAMES
+from unified_daily_digest import _render_consolidated
+
+
+class DailyCallerPresentationTests(unittest.TestCase):
+    setUp = caller_fixture.CloudRunDigestTests.setUp
+    run_request = caller_fixture.CloudRunDigestTests.run_request
+    send = caller_fixture.CloudRunDigestTests.send
+    enable = caller_fixture.CloudRunDigestTests.enable
+
+    def assert_sections(self, text, locale="zh"):
+        marks = [f"【{name}】" if locale == "zh" else f"[{name}]" for name in SECTION_NAMES[locale]]
+        self.assertEqual([text.count(mark) for mark in marks], [1] * 5)
+        self.assertEqual(sorted(text.index(mark) for mark in marks), [text.index(mark) for mark in marks])
+        self.assertLessEqual(app._units(text), 3900)
+
+    def test_central_caller_preserves_sections_and_disabled_scope(self):
+        for locale in ("zh", "en"):
+            with self.subTest(locale=locale):
+                text, coverage = app._report(replace(self.configuration, locale=locale), [ROW],
+                                             ["present"], ROW["business_day"], config(EXPECTED))
+                self.assert_sections(text, locale)
+                self.assertIn("Binance：已停用" if locale == "zh" else "Binance: disabled", text)
+                self.assertIn("运行／订单／成交不适用" if locale == "zh" else "runs/orders/fills N/A", text)
+                self.assertIn("未纳入本期汇总，运行状态未知" if locale == "zh" else "not included in this digest; run state unknown", text)
+                self.assertIn("运行链接：不可用" if locale == "zh" else "Run link: unavailable", text)
+
+    def test_central_caller_regression_snapshots(self):
+        for locale in ("zh", "en"):
+            text, _ = app._report(replace(self.configuration, locale=locale), [ROW],
+                                  ["present"], ROW["business_day"], config(EXPECTED))
+            normalized = re.sub(r"\d{4}-\d{2}-\d{2}", "<PROJECT_ID>", text)
+            snapshot = Path(__file__).parent / "fixtures" / f"daily-caller-{locale}.txt"
+            self.assertEqual(normalized + "\n", snapshot.read_text())
+
+    def test_disabled_claim_requires_inventory_and_no_bound_binance(self):
+        import daily_digest_notify as digest
+        for source in (None, dict(expected_targets=1, covered_targets=0, runs=0)):
+            inventory = {} if source is None else {"binance": source}
+            payload = digest.DailyDigestInput(business_day=ROW["business_day"], locale="en",
+                                              notification_summary={"source_inventory": inventory})
+            text = _render_consolidated(payload, max_units=3900, known_disabled_platforms={"binance"})
+            self.assertNotIn("Binance: disabled", text)
+            self.assertIn("source coverage unknown" if source is None else "expected target coverage 0/1", text)
+
+    def test_run_link_is_supplied_bound_and_sanitized_or_unavailable(self):
+        repository = "example-org/example-repo"
+        url = f"https://github.com/{repository}/actions/runs/42"
+        cfg = config(EXPECTED)
+        cfg["platforms"] = {"schwab": {"repository": repository}}
+        for supplied, accepted in ((url + "?token=fake#private", True),
+                                   (url.replace(repository, "other/repo"), False),
+                                   (url.replace("github.com", "github.com.evil.test"), False),
+                                   (url.replace("https:", "http:"), False),
+                                   (url.replace("github.com", "user@github.com"), False),
+                                   (url + "/private", False), ("private-run-id", False)):
+            with self.subTest(accepted=accepted, supplied=supplied):
+                row = candidate(run=supplied)
+                text, _ = app._report(self.configuration, [row], ["present"], ROW["business_day"], cfg)
+                self.assert_sections(text)
+                self.assertNotIn("fake", text)
+                self.assertNotIn("private", text)
+                self.assertNotIn("evil.test", text)
+                self.assertIn(url if accepted else "运行链接：不可用", text)
+
+    def test_late_fallback_keeps_sections_unknown_and_single_reservation(self):
+        self.now += app.timedelta(minutes=31)
+        self.run_request(configuration=self.enable())
+        self.run_request(configuration=self.enable())
+        self.assertEqual(len(self.sent), 1)
+        self.assert_sections(self.sent[0])
+        self.assertIn("成交与资产状态未知", self.sent[0])
+        self.assertIn("运行链接：不可用", self.sent[0])
+
+    def test_missing_all_sources_stays_five_sections_without_global_zero(self):
+        self.store.rows = {}
+        self.run_request(configuration=self.enable())
+        self.assert_sections(self.sent[0])
+        self.assertIn("全窗口统计未知", self.sent[0])
+        self.assertNotIn("今日全部账号无交易", self.sent[0])
+
+    def test_nonrun_coverage_and_verified_zero_keep_different_scopes(self):
+        ft = candidate(platform_id="firstrade", actually_ran=False, notification_event=None)
+        ft.pop("notification_event")
+        expected = [{**{key: ft[key] for key in app.IDENTITY}, "candidate_slot": "fake-slot"}]
+        expected += EXPECTED + [{**EXPECTED[0], "platform_id": "longbridge"}]
+        configuration = replace(self.configuration, expected=expected)
+        text, _ = app._report(configuration, [ft, ROW], ["not_run", "present", "missing"], ROW["business_day"], config(expected))
+        self.assertIn("已收录 0 次运行，目标覆盖 1/1", text)
+        self.assertIn("全窗口统计未知", text)
+        self.assertIn("Schwab：已收录 1 次运行，目标覆盖 1/1", text)
+        self.assertIn("报告注明未运行；不推断成交统计", text)
+        self.assert_sections(text)
+
+
+if __name__ == "__main__":
+    unittest.main()