fix: Unify report and candidate execution facts

application/runtime_report_summary.py

@@ -30,7 +30,9 @@ def _counter_by_key(items: Iterable[Mapping[str, object]], key: str) -> dict[str
     return dict(counter)
 
 
-def summarize_execution_cycle_result(result: object, *, dry_run: bool) -> dict[str, object]:
+def summarize_execution_cycle_result(
+    result: object, *, dry_run: bool, existing_summary: Mapping | None = None,
+) -> dict[str, object]:
     execution = _as_mapping(getattr(result, "execution", {}))
     submitted_orders = _normalize_submitted_orders(result)
     trade_logs = tuple(str(item) for item in (getattr(result, "trade_logs", ()) or ()) if str(item).strip())
@@ -72,11 +74,31 @@ def summarize_execution_cycle_result(result: object, *, dry_run: bool) -> dict[s
     from notifications.accepted_orders import cycle_notification_evidence
     raw_orders = tuple(getattr(result, "submitted_orders", ()) or ())
     summary["notification_digest"] = cycle_notification_evidence(execution, raw_orders, dry_run=dry_run)
+    if not execution_status and not raw_orders:
+        # Validate completion independently of dry-run digest eligibility. Keep
+        # the original preview evidence above: no live orders or fills are proved.
+        completion = cycle_notification_evidence(execution, raw_orders)
+        if completion["routine"] is True:
+            status = "dry_run_completed" if dry_run else "no_action"
+            summary.update(result=status, execution_status=status)
     if unified_enabled() and execution.get("notification_approved_gates"):
         from notifications.approved_gates import clean_gate_reasons
         summary["notification_approved_gates"] = clean_gate_reasons(execution["notification_approved_gates"])
         if execution.get("notification_attention_reason_codes"):
             summary["notification_attention_reason_codes"] = list(execution["notification_attention_reason_codes"])
     if execution.get("notification_events"):
         summary["notification_events"] = execution["notification_events"]
-    return summary
+    return {**dict(existing_summary or {}), **summary}
+
+
+def execution_report_status(summary: Mapping) -> str:
+    """Report completion must not claim success for blocked or unknown execution.
+
+    Keep the runtime_report.v1 outer vocabulary (ok/error); precise execution
+    outcomes remain in summary.execution_status. This grants no trading authority.
+    """
+    status = str(summary.get("execution_status") or "").strip().lower()
+    healthy = {"no_action", "no_op", "completed", "executed", "submitted",
+               "broker_acknowledged", "pending", "partially_filled", "pending_reconciliation",
+               "dry_run", "dry_run_completed"}
+    return "ok" if status in healthy else "error"

docs/unified-daily-notification.md

@@ -250,3 +250,29 @@ Candidate merging normalizes documented optional null defaults, count-status
 spelling and equivalent timestamp offsets. Identity, counters, execution state
 and other factual differences still fail before the generation-guarded upload.
 This projection does not change reconciliation, broker calls or alert routing.
+
+Execution cycles use the same summary builder for early notification candidates
+and the completed archive. The outer `runtime_report.v1` status remains `ok` or
+`error`: blocked, failed, missing and unknown execution outcomes use `error`;
+recognized completed, preview and broker-submission outcomes use `ok`.
+A positively completed zero-order cycle without a submission status is normalized
+to `no_action` when the existing completion and failure checks pass. An empty
+dry-run preview uses the same checks and reports `dry_run_completed`; its digest
+remains ineligible and live order and fill counts remain unknown. Missing
+completion evidence and conflicting failures still use `error`.
+Verified consecutive-loss and daily-loss blocks retain their structured reasons
+with an outer `error`. They remain rejected by the execution heartbeat, while
+their notification is deferred to the digest only after receipt validation and
+confirmed candidate publication. Other failures still alert immediately.
+These validated gate-only errors can still publish the same cycle's account
+observation through the account-facts projector. Receipt, source identity,
+currency and freshness validation remain required; this does not accept execution.
+`summary.execution_status` retains the precise outcome. An outer `ok` still does
+not prove a fill or reconciliation. Probe and skipped-report semantics are unchanged.
+
+Plugin `routine_observation_counts` are carried from the current run's summary
+into both projections. An absent field stays unknown; an explicitly empty map
+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.

main.py

@@ -720,8 +720,8 @@ def _has_signal_snapshot_details(snapshot: dict[str, object]) -> bool:
     )
 
 
-def _summarize_cycle_result_for_report(result, *, dry_run: bool) -> dict[str, object]:
-    return summarize_execution_cycle_result(result, dry_run=dry_run)
+def _summarize_cycle_result_for_report(result, *, dry_run: bool, existing_summary=None) -> dict[str, object]:
+    return summarize_execution_cycle_result(result, dry_run=dry_run, existing_summary=existing_summary)
 
 
 def persist_execution_report(report, *, dry_run_only_override: bool | None = None):
@@ -949,6 +949,7 @@ def _handle_schwab_cycle(*, dry_run_only_override: bool | None = None, response_
         execution_summary = _summarize_cycle_result_for_report(
             execution_result,
             dry_run=bool(report.get("dry_run")),
+            existing_summary=report.get("summary"),
         )
         account_observation = getattr(execution_result, "account_observation", None)
         if isinstance(account_observation, dict):
@@ -970,9 +971,10 @@ def _handle_schwab_cycle(*, dry_run_only_override: bool | None = None, response_
             # execution result intact and let the control plane show missing
             # evidence instead of changing a completed cycle into an error.
             pass
+        from application.runtime_report_summary import execution_report_status
         finalize_runtime_report(
             report,
-            status="ok",
+            status=execution_report_status(execution_summary),
             summary=execution_summary,
             diagnostics=_runtime_revision_diagnostics(
                 signal=signal_diagnostics,

notifications/approved_gates.py

@@ -105,11 +105,15 @@ def render_gate_reasons(reasons, locale):
 
 
 def report_gate_reasons(report, *, observed_at=None):
-    """Require the existing receipt validator; never make a report acceptable."""
+    """Classify gate-only outer errors without making execution acceptable.
+
+    Older archives use a successful outer status. Both forms require the same
+    validated risk-blocked receipt and absence of independent failures.
+    """
     summary = report.get("summary")
     if not isinstance(summary, Mapping) or report.get("dry_run") is not False:
         return []
-    if (report.get("status") not in {"ok", "success", "completed"}
+    if (report.get("status") not in {"ok", "success", "completed", "error"}
             or any(report.get(k) for k in ("errors", "error", "error_summary"))
             or type(summary.get("orders_pending_count")) is not int or summary["orders_pending_count"] != 0):
         return []

notifications/unified.py

@@ -277,13 +277,11 @@ def _cycle_projection(result, fact, config):
         raise CandidatePersistenceError("notification_target_unknown")
     # Normalize only the notification copy of the established Schwab aliases.
     target["platform_id"] = "schwab"
-    report.update(platform="schwab", status="ok", dry_run=bool(config.dry_run_only))
-    from .accepted_orders import cycle_notification_evidence
-    summary = dict(report.get("summary") or {})
-    summary.update(execution_status=result.execution.get("execution_status"),
-        notification_digest=cycle_notification_evidence(result.execution, result.submitted_orders,
-            dry_run=bool(config.dry_run_only)))
-    report["summary"] = summary
+    from application.runtime_report_summary import execution_report_status, summarize_execution_cycle_result
+    summary = summarize_execution_cycle_result(result, dry_run=bool(config.dry_run_only),
+        existing_summary=report.get("summary"))
+    report.update(platform="schwab", status=execution_report_status(summary),
+        dry_run=bool(config.dry_run_only), summary=summary)
     if report["dry_run"]:
         # Existing live archive acceptance is unchanged; previews are facts only.
         return {"schema_version": "qsl.digest_candidates.v1", "runs": [], "notification_events": [fact]}

scripts/execution_report_heartbeat.py

@@ -1063,8 +1063,12 @@ def main(now: dt.datetime | None = None) -> int:
         if is_latest_report:
             latest_report_seen.add(service_name)
             if not ok:
-                if unified_enabled() and reason == "rejected inner execution status" and report_gate_reasons(payload, observed_at=now):
-                    approved_latest[service_name] = payload
+                if unified_enabled() and report_gate_reasons(payload, observed_at=now):
+                    # Gate-only reports now carry outer error. Classify a copy
+                    # with the legacy outer status so unrelated stage, error and
+                    # delivery rejections still prevent notification deferral.
+                    if _is_accepted_report({**payload, "status": "ok"}) == (False, "rejected inner execution status"):
+                        approved_latest[service_name] = payload
                 latest_report_issues.append(f"{service_name or name}: {reason}")
                 if _report_execution_status(payload).lower() in {"blocked", "risk_blocked"}:
                     summary = payload.get("summary") or {}

scripts/publish_account_facts_from_reports.py

@@ -272,7 +272,7 @@ def project_schwab_account_facts_history(
             or report.get("service_name") != expected_service_name
             or report.get("strategy_profile") != STRATEGY_PROFILE
             or report.get("account_scope") not in (None, ACCOUNT_SCOPE)
-            or report.get("status") != "ok"
+            or report.get("status") not in ("ok", "error")
             or report.get("errors") != []
         ):
             raise _ProjectionError("report_identity_mismatch")
@@ -298,6 +298,12 @@ def project_schwab_account_facts_history(
         if not isinstance(now, datetime) or now.tzinfo is None or now.utcoffset() is None:
             raise _ProjectionError("invalid_observation_time")
         current = now.astimezone(timezone.utc)
+        if report.get("status") == "error":
+            # A validated gate-only error still carries the broker observation.
+            # Validate identity/time first and never relabel execution as healthy.
+            from notifications.approved_gates import report_gate_reasons
+            if not report_gate_reasons(report, observed_at=current):
+                raise _ProjectionError("report_identity_mismatch")
 
         summary = report.get("summary")
         observation = summary.get("account_observation") if isinstance(summary, Mapping) else None

tests/test_notification_approved_gates.py

@@ -69,25 +69,37 @@ def test_approved_gates_are_digest_candidates_with_every_reason_and_no_order(mon
     assert summary["notification_approved_gates"] == reasons
 
 
+@pytest.mark.parametrize("outer_status", ["ok", "error"])
 @pytest.mark.parametrize("reasons", [[CONSECUTIVE], [DAILY], [CONSECUTIVE, DAILY]])
-def test_approved_block_receipt_suppresses_alert_only_after_candidate_and_still_fails(monkeypatch, reasons):
+def test_approved_block_receipt_suppresses_alert_only_after_candidate_and_still_fails(monkeypatch, reasons, outer_status):
     store = configure(monkeypatch)
     source = source_report(reasons)
-    assert report_gate_reasons(source) == reasons
+    source["status"] = outer_status
+    assert report_gate_reasons(source, observed_at=NOW) == reasons
+    projection = unified._project_runs([source], expected_target=source["runtime_target"],
+        opaque_account_uid="fake-opaque", target_id="fake-target", now=NOW)
+    assert projection["runs"][0]["approved_gate_reasons"] == reasons
     before = copy.deepcopy(source)
     sent = mock_heartbeat(monkeypatch, source)
     monkeypatch.setenv("RUNTIME_TARGET_JSON", json.dumps(source["runtime_target"]))
     monkeypatch.setattr(unified, "emit_completed_candidates", lambda env, **kwargs: {"uploaded": True})
     assert heartbeat.main(NOW) == 1
     assert not sent and source == before
-    assert heartbeat._is_accepted_report(source) == (False, "rejected inner execution status")
+    expected_reason = "rejected inner execution status" if outer_status == "ok" else "rejected status=error stage=-"
+    assert heartbeat._is_accepted_report(source) == (False, expected_reason)
     assert store.facts == {}  # Archive coverage owns this receipt; no second cycle fact.
 
 
-@pytest.mark.parametrize("change", ["missing_receipt", "wrong_receipt", "order", "error", "pending", "other_reason", "bad_limit", "producer_failed"])
-def test_other_receipt_failures_stay_immediate(monkeypatch, change):
+@pytest.mark.parametrize("outer_status", ["ok", "error"])
+@pytest.mark.parametrize("change", ["missing_receipt", "wrong_receipt", "order", "error", "pending", "other_reason", "bad_limit", "producer_failed", "stage", "notification_failure", "stale_receipt", "unknown_outer"])
+def test_other_receipt_failures_stay_immediate(monkeypatch, change, outer_status):
     configure(monkeypatch)
     source = source_report([CONSECUTIVE])
+    source["status"] = outer_status
+    if change == "stage": source["stage"] = "FUNDING_BLOCKED"
+    if change == "notification_failure": source["notification_error"] = "fake-error"
+    if change == "stale_receipt": source["execution_receipt"]["observed_at"] = "2026-10-05T11:00:00Z"
+    if change == "unknown_outer": source["status"] = "unknown"
     if change == "missing_receipt": source.pop("execution_receipt")
     if change == "wrong_receipt": source["execution_receipt"]["strategy_profile"] = "another-strategy"
     if change == "order": source["summary"].update(orders_submitted_count=1, broker_submission_done=True)