feat: unify digest candidate confirmation warnings

.github/workflows/execution-report-heartbeat.yml

@@ -126,6 +126,7 @@ jobs:
           uv run --no-sync python scripts/with_heartbeat_digest_binding.py candidates
 
       - name: Publish read-only runtime execution evidence
+        if: ${{ !cancelled() && steps.install_runtime_deps.outcome == 'success' && (steps.gcp_auth_primary.outcome == 'success' || steps.gcp_auth_retry.outcome == 'success') }}
         uses: QuantStrategyLab/QuantRuntimeSettings/actions/publish-runtime-execution-evidence@c97752202670fac58db797c01596d11dd105a94b
         with:
           source-id: runtime-reports-schwab

docs/runtime-daily-heartbeat-sync.zh-CN.md

@@ -9,6 +9,11 @@ daily publisher step，在既有 22:20 UTC 调度内复用同一 WIF/uv 流程
 或证据同步失败仍可记录日报；认证、依赖安装失败或运行取消不发布。日报的失败
 不会跳过已经执行的原健康通知，原步骤顺序与行为不变。
 
+`Publish read-only runtime execution evidence` 同样独立于健康检查和候选发布的
+成功状态：只要运行未取消、依赖安装成功且本次认证成功，就继续导出已有执行证据。
+因此缺少可接受回执或候选冲突不会额外造成管理站的执行证据缺口。原检查仍保留失败
+结论；导出步骤继续使用原有 action、来源和步骤级凭据，不调用策略或券商。
+
 GitHub默认条件语义：[官方表达式说明](https://docs.github.com/en/actions/reference/workflows-and-actions/expressions#status-check-functions)。
 
 ## 启用

docs/unified-daily-notification.md

@@ -276,3 +276,18 @@ means the plugin observation completed with no routine signals. These states
 remain distinct in canonical comparison. Existing contradictory candidates are
 not rewritten or normalized away. Source identity, factual conflict rejection
 and generation-conditional writes continue to apply on replay and restart.
+
+## Candidate warning display
+
+The existing candidate-persistence warning uses the shared five sections:
+Result, Scope, Evidence, Coverage and Next step. Its QPK presentation snapshot
+has a separate byte/source manifest under `notifications/digest_snapshot`;
+other snapshots and runtime dependency pins are unchanged.
+
+The warning states that inclusion is unconfirmed, with unknown orders/fills and
+an unknown business window when none was supplied. A run link uses only valid
+current Actions context; absent context remains unavailable. It does not copy
+private storage paths or raw exceptions. Existing attempt-marker identity,
+atomic publication, sender callback, strict boolean acknowledgement and
+failure behavior are unchanged. This copy change does not resolve an actual
+candidate conflict or prove central delivery.

notifications/digest_snapshot/presentation-source.json

@@ -0,0 +1,6 @@
+{
+  "repository": "QuantStrategyLab/QuantPlatformKit",
+  "commit": "fb8e687fb9a5d61694d5ad21591c52247bf8dfa1",
+  "path": "src/quant_platform_kit/notifications/presentation.py",
+  "sha256": "e3cfdadb0c138daa2e174ae1a94f43a81abacdd70b2c66b502443da4bc08e999"
+}

notifications/digest_snapshot/presentation.py

@@ -0,0 +1,80 @@
+"""Shared display sections; no execution, routing, receipt or transport policy."""
+from __future__ import annotations
+
+from collections.abc import Sequence
+from urllib.parse import urlsplit, urlunsplit
+
+SECTION_NAMES = {
+    "zh": ("结果", "范围", "证据", "覆盖", "下一步"),
+    "en": ("Result", "Scope", "Evidence", "Coverage", "Next step"),
+}
+
+
+def safe_run_url(value: object) -> str | None:
+    """Use only caller-supplied HTTPS links, dropping query and fragment."""
+    if not isinstance(value, str) or any(c.isspace() or ord(c) < 32 for c in value):
+        return None
+    try:
+        parsed = urlsplit(value)
+        if parsed.scheme == "https" and parsed.hostname and not parsed.username and not parsed.password:
+            return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
+    except ValueError:
+        pass
+    return None
+
+
+def next_step_text(locale: str, *, needs_review: bool) -> str:
+    if locale == "zh":
+        return ("核对该次运行、回执和缺失来源；保持现有执行限制。" if needs_review else
+                "无需操作；等待下一统计窗口。")
+    return ("Review the source run, receipt and missing sources; retain existing execution restrictions."
+            if needs_review else "No action required; await the next reporting window.")
+
+
+def render_operator_message(*, title: str, locale: str, result: Sequence[str],
+                            scope: Sequence[str], evidence: Sequence[str], coverage: Sequence[str],
+                            next_step: Sequence[str], max_units: int | None = None,
+                            required_evidence: Sequence[str] = ()) -> str:
+    """Keep result, scope, coverage and next step before optional evidence.
+
+    Truncation is explicit and cannot remove a section or imply full coverage.
+    Callers put urgent evidence before optional observations. Financial facts
+    must already have been validated by the producer; this helper infers none.
+    """
+    names = SECTION_NAMES["zh" if locale == "zh" else "en"]
+    unknown = "未知" if locale == "zh" else "Unknown"
+    groups = [[str(line) for line in group if line] or [unknown]
+              for group in (result, scope, evidence, coverage, next_step)]
+
+    def assemble():
+        lines = [title]
+        for name, group in zip(names, groups):
+            lines.extend([f"【{name}】" if locale == "zh" else f"[{name}]", *group])
+        return "\n".join(lines)
+
+    if max_units is None:
+        return assemble()
+    if max_units < 256:
+        raise ValueError("notification_budget_too_small")
+    omitted = 0
+    original = [line for line in groups[2] if line not in required_evidence]
+    required = list(dict.fromkeys(required_evidence))
+    groups[2] = required.copy()
+    suffix = "部分证据已省略；请查看运行记录。" if locale == "zh" else "Some evidence omitted; inspect the source run."
+    groups[2] = required + [suffix]
+    if len(assemble().encode("utf-16-le")) // 2 > max_units:
+        raise ValueError("notification_required_fields_exceed_budget")
+    groups[2] = required.copy()
+    for line in original:
+        groups[2].append(line)
+        groups[2].append(suffix)
+        fits = len(assemble().encode("utf-16-le")) // 2 <= max_units
+        groups[2].pop()
+        if not fits:
+            groups[2].pop()
+            omitted += 1
+    if omitted:
+        groups[2].append(suffix)
+    if not groups[2]:
+        groups[2] = [unknown]
+    return assemble()

scripts/digest_coverage_warning.py

@@ -4,6 +4,7 @@
 import hashlib
 import json
 import os
+import re
 import sys
 import tempfile
 
@@ -47,12 +48,37 @@ def warn_coverage_unconfirmed(send, locale):
                     pass
     elif os.environ.get("GITHUB_ACTIONS") == "true":
         print("digest_coverage_warning_marker_unavailable", file=sys.stderr)
-    message = ("⚠️ Schwab 日报候选记录未确认，日报可能缺少本次记录；请检查运行记录和摘要存储。"
-               if str(locale).lower().startswith("zh") else
-               "⚠️ Schwab daily digest candidate record is unconfirmed; the digest may omit this record. "
-               "Check the run and digest storage.")
+    message = _coverage_message(locale)
     try:
         return send(message) is True
     except Exception:  # noqa: BLE001 -- The attempt remains recorded when delivery is uncertain.
         print("digest_coverage_warning_send_unconfirmed", file=sys.stderr)
         return False
+
+
+def _coverage_message(locale):
+    """Display only; preserve the existing attempted-send marker and callback."""
+    from notifications.digest_snapshot.presentation import next_step_text, render_operator_message
+
+    lang = "zh" if str(locale).lower().startswith("zh") else "en"
+    zh = lang == "zh"
+    link = None
+    # Only the active Actions run is a valid source for this warning's run link.
+    repository = os.environ.get("GITHUB_REPOSITORY", "")
+    run_id = os.environ.get("GITHUB_RUN_ID", "")
+    if (os.environ.get("GITHUB_ACTIONS") == "true"
+            and os.environ.get("GITHUB_SERVER_URL") == "https://github.com"
+            and re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", repository)
+            and re.fullmatch(r"[0-9]{1,20}", run_id)):
+        link = f"https://github.com/{repository}/actions/runs/{run_id}"
+    return render_operator_message(
+        title="⚠️ Schwab 日报候选确认" if zh else "⚠️ Schwab digest candidate confirmation",
+        locale=lang,
+        result=["日报候选记录未确认" if zh else "Daily digest candidate record is unconfirmed"],
+        scope=["本次候选保存尝试；统计窗口未知" if zh else "This candidate persistence attempt; business window unknown"],
+        evidence=["reason_code: candidate_publication_unconfirmed",
+                  ("运行记录：" if zh else "Run record: ") + (link or ("未提供运行链接" if zh else "run link not provided"))],
+        coverage=["日报可能缺少本次记录；订单：未知；成交：未知" if zh else
+                  "The digest may omit this record; orders: unknown; fills: unknown"],
+        next_step=[next_step_text(lang, needs_review=True)],
+    )

tests/test_daily_notification_contract.py

@@ -271,8 +271,14 @@ def test_healthy_heartbeat_coverage_failure_is_urgent_and_private(locale, monkey
                                                source_reports=[{"execution_receipt": {"synthetic": True}}])
     assert result == 1
     assert len(sent) == 1
-    assert ("Schwab 日报候选记录未确认，日报可能缺少本次记录" if locale == "zh" else
-            "Schwab daily digest candidate record is unconfirmed; the digest may omit this record") in sent[0]
+    assert "Schwab" in sent[0]
+    assert ("日报候选记录未确认" if locale == "zh" else
+            "Daily digest candidate record is unconfirmed") in sent[0]
+    assert ("日报可能缺少本次记录" if locale == "zh" else "The digest may omit this record") in sent[0]
+    assert ("成交：未知" if locale == "zh" else "fills: unknown") in sent[0]
+    for section in (("结果", "范围", "证据", "覆盖", "下一步") if locale == "zh" else
+                    ("Result", "Scope", "Evidence", "Coverage", "Next step")):
+        assert (f"【{section}】" if locale == "zh" else f"[{section}]") in sent[0]
     assert "private order detail" not in sent[0]
 
 

tests/test_digest_coverage_warning.py

@@ -179,6 +179,37 @@ def test_languages_are_bounded_and_private(self):
                 self.assertNotIn('private exception', sent[0])
                 self.assertNotIn('http', sent[0].lower())
 
+    def test_unified_sections_and_unknown_counts_do_not_change_send_ack(self):
+        for locale in ("zh", "en"):
+            with self.subTest(locale=locale), patch.dict(os.environ, {}, clear=True):
+                sent = []
+                self.assertFalse(warn_coverage_unconfirmed(lambda text: sent.append(text), locale))
+                self.assertEqual(len(sent), 1)
+                sections = ("结果", "范围", "证据", "覆盖", "下一步") if locale == "zh" else ("Result", "Scope", "Evidence", "Coverage", "Next step")
+                for section in sections:
+                    self.assertIn(f"【{section}】" if locale == "zh" else f"[{section}]", sent[0])
+                self.assertIn("成交：未知" if locale == "zh" else "fills: unknown", sent[0])
+                self.assertIn("candidate_publication_unconfirmed", sent[0])
+                self.assertNotIn("成交：0", sent[0])
+
+    def test_run_link_requires_valid_current_actions_context(self):
+        environment = dict(GITHUB_ACTIONS="true", GITHUB_SERVER_URL="https://github.com",
+                           GITHUB_REPOSITORY="example/synthetic", GITHUB_RUN_ID="123")
+        for run_id in ("123", "123?private=hidden"):
+            with self.subTest(valid=run_id == "123"), patch.dict(os.environ, {**environment, "GITHUB_RUN_ID": run_id}, clear=True):
+                sent = []
+                self.assertTrue(warn_coverage_unconfirmed(lambda text: sent.append(text) or True, "en"))
+                self.assertEqual("https://github.com/example/synthetic/actions/runs/123" in sent[0], run_id == "123")
+                self.assertNotIn("private=hidden", sent[0])
+
+    def test_presentation_snapshot_matches_recorded_source_bytes(self):
+        import hashlib
+        import json
+        root = ROOT / "notifications/digest_snapshot"
+        manifest = json.loads((root / "presentation-source.json").read_text())
+        self.assertRegex(manifest["commit"], r"^[a-f0-9]{40}$")
+        self.assertEqual(hashlib.sha256((root / "presentation.py").read_bytes()).hexdigest(), manifest["sha256"])
+
     def test_workflow_uses_same_job_path_in_both_steps(self):
         workflow = (ROOT / '.github/workflows/execution-report-heartbeat.yml').read_text()
         marker = ('DIGEST_COVERAGE_ALERT_ATTEMPT_PATH: ${{ runner.temp }}/schwab-digest-coverage-'

tests/test_execution_evidence_failure_path.py

@@ -0,0 +1,20 @@
+"""Evidence export remains independent of failed heartbeat/candidate checks."""
+from test_runtime_daily_heartbeat_workflow import (
+    _condition, _evaluate, _step_block, _workflow_text,
+)
+
+
+def test_evidence_export_after_attention_uses_authenticated_installed_reader():
+    step = _step_block(_workflow_text(), "Publish read-only runtime execution evidence")
+    condition = _condition(step)
+    # A status function overrides implicit success(), including failed candidate publication.
+    assert "!cancelled()" in condition
+    assert _evaluate(condition, "", "success", "skipped")
+    assert _evaluate(condition, "", "failure", "success")
+    for primary, retry in (("failure", "failure"), ("failure", "skipped")):
+        assert not _evaluate(condition, "", primary, retry)
+    for installed in ("failure", "skipped", "cancelled"):
+        assert not _evaluate(condition, "", "success", "skipped", installed=installed)
+    assert not _evaluate(condition, "", "success", "skipped", cancelled=True)
+    assert "actions/publish-runtime-execution-evidence@" in step
+    assert "EXECUTION_EVIDENCE_SYNC_TOKEN:" in step

tests/test_unified_notification_entrypoints.py

@@ -105,7 +105,11 @@ def test_heartbeat_failed_producer_sends_coverage_error_and_reports_unknown(monk
     monkeypatch.setattr(unified, "emit_completed_candidates", lambda env, **kwargs: {"uploaded": False})
     assert heartbeat.main(NOW) == 1
     assert len(sent) == 1
-    assert "Schwab 日报候选记录未确认，日报可能缺少本次记录" in sent[0]
+    assert "Schwab" in sent[0]
+    assert "日报候选记录未确认" in sent[0]
+    assert "日报可能缺少本次记录" in sent[0]
+    assert "成交：未知" in sent[0]
+    assert "【覆盖】" in sent[0] and "【下一步】" in sent[0]
     assert "coverage=unknown" in capsys.readouterr().out