diff --git a/application/runtime_report_summary.py b/application/runtime_report_summary.py
index 56b710e..dc7c9b5 100644
--- a/application/runtime_report_summary.py
+++ b/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,6 +74,13 @@ 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"])
@@ -79,4 +88,17 @@ def summarize_execution_cycle_result(result: object, *, dry_run: bool) -> dict[s
             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"
diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md
index 7c20a2c..585cc83 100644
--- a/docs/unified-daily-notification.md
+++ b/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.
diff --git a/main.py b/main.py
index 8e64dba..f92b7ea 100644
--- a/main.py
+++ b/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,
diff --git a/notifications/approved_gates.py b/notifications/approved_gates.py
index 1e15f7b..1e0aba1 100644
--- a/notifications/approved_gates.py
+++ b/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 []
diff --git a/notifications/unified.py b/notifications/unified.py
index d6bd02a..9910390 100644
--- a/notifications/unified.py
+++ b/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]}
diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py
index 0ada7da..50e1d2d 100644
--- a/scripts/execution_report_heartbeat.py
+++ b/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 {}
diff --git a/scripts/publish_account_facts_from_reports.py b/scripts/publish_account_facts_from_reports.py
index a08973b..223ace0 100644
--- a/scripts/publish_account_facts_from_reports.py
+++ b/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
diff --git a/tests/test_notification_approved_gates.py b/tests/test_notification_approved_gates.py
index 0858cfc..ded5ea8 100644
--- a/tests/test_notification_approved_gates.py
+++ b/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)
diff --git a/tests/test_publish_account_facts_from_reports.py b/tests/test_publish_account_facts_from_reports.py
index 388d978..8126229 100644
--- a/tests/test_publish_account_facts_from_reports.py
+++ b/tests/test_publish_account_facts_from_reports.py
@@ -847,3 +847,101 @@ def test_external_flow_opt_in_preserves_independent_positions_opt_in():
             assert ("externalFlow" in payload) is flow
     observation["net_assets_currency_source"] = None
     assert publisher.project_schwab_account_facts_history(report, external_flow_enabled=True, **kwargs)["reason"] == "net_assets_currency_unconfirmed"
+
+
+# A completed risk pause supplies broker observations, not execution authority.
+def _completed_risk_block(reasons):
+    from quant_platform_kit.common.execution_receipts import build_execution_receipt
+    from test_notification_approved_gates import execution
+
+    source = _valid_report()
+    source.update(status="error", dry_run=False, runtime_release_receipt={
+        "attestation_state": "self_attested", "strategy_release": {"strategy_revision": "a" * 40},
+    })
+    source["runtime_target"]["execution_mode"] = "live"
+    source["summary"].update(execution(reasons), orders_submitted_count=0, orders_pending_count=0)
+    source["summary"]["account_observation"].update(
+        cash_balance="12.34", cash_balance_source="cashBalance", cash_currency="USD",
+        cash_currency_source="owner_confirmed", broker_reported_positions_scope="stocks_only",
+        broker_reported_positions_complete=True,
+        broker_reported_positions_observed_at=source["finished_at"],
+        broker_reported_positions=[{"symbol": "FAKE", "quantity": "1", "market_value": "10.00", "currency": "USD"}],
+    )
+    source["execution_receipt"] = build_execution_receipt(
+        platform="schwab", strategy_profile=source["strategy_profile"], strategy_revision="a" * 40,
+        execution_mode="live", outcome="risk_blocked", observed_at=source["finished_at"],
+    )
+    return source
+
+
+def _project_completed_risk_block(source):
+    prefix = "gs://example-bucket/execution-reports/charles_schwab/soxl_soxx_trend_income/"
+    return publisher.project_schwab_account_facts_history(
+        source, source_report_uri=prefix + "2026-09/20260930T222000Z.json", report_prefix=prefix,
+        expected_service_name=source["service_name"], expected_runtime_revision="service-00007-abc",
+        expected_target_id="fake-target", expected_cash_currency="USD",
+        now=datetime(2026, 10, 1, 1, 20, tzinfo=timezone.utc),
+    )
+
+
+@pytest.mark.parametrize("reason_codes", ["consecutive", "daily", "both"])
+def test_completed_risk_block_preserves_account_facts_and_execution_rejection(reason_codes):
+    import copy
+    from notifications.approved_gates import report_gate_reasons
+    from scripts.execution_report_heartbeat import _is_accepted_report
+    from test_notification_approved_gates import CONSECUTIVE, DAILY
+
+    reasons = {"consecutive": [CONSECUTIVE], "daily": [DAILY], "both": [CONSECUTIVE, DAILY]}[reason_codes]
+    source = _completed_risk_block(reasons)
+    before = copy.deepcopy(source)
+    body = _project_completed_risk_block(source)
+    assert body.get("schema_version") == publisher.HISTORY_SCHEMA
+    legacy = copy.deepcopy(source)
+    legacy["status"] = "ok"
+    assert body == _project_completed_risk_block(legacy)
+    assert body["broker_reported_balances"][0]["net_assets"] == "123.45"
+    assert body["cash"][0]["cash_balance"] == "12.34"
+    assert body["broker_reported_positions"][0]["symbol"] == "FAKE"
+    assert _is_accepted_report(source)[0] is False
+    assert report_gate_reasons(source, observed_at=datetime(2026, 10, 1, 1, 20, tzinfo=timezone.utc)) == reasons
+    assert source == before
+
+
+@pytest.mark.parametrize("fault", [
+    "missing_receipt", "wrong_receipt", "wrong_outcome", "stale_receipt", "future_receipt", "wrong_revision",
+    "unknown_status", "failed_status", "errors", "error", "pending", "orders", "broker", "unknown_execution",
+    "missing_reasons", "invalid_reasons", "dry_run", "profile", "selector", "revision", "run_id", "currency",
+    "stale_observation", "outside_report", "invalid_net_assets", "missing_runtime_target", "invalid_status",
+])
+def test_risk_block_account_facts_keep_receipt_identity_currency_and_freshness_gates(fault):
+    from test_notification_approved_gates import CONSECUTIVE
+
+    source = _completed_risk_block([CONSECUTIVE])
+    if fault == "missing_receipt": source.pop("execution_receipt")
+    if fault == "wrong_receipt": source["execution_receipt"]["strategy_profile"] = "fake-other"
+    if fault == "wrong_outcome": source["execution_receipt"]["outcome"] = "no_action"
+    if fault == "stale_receipt": source["execution_receipt"]["observed_at"] = "2026-09-30T22:19:00Z"
+    if fault == "future_receipt": source["execution_receipt"]["observed_at"] = "2026-10-01T02:00:00Z"
+    if fault == "wrong_revision": source["execution_receipt"]["strategy_revision"] = "b" * 40
+    if fault == "invalid_status": source["status"] = {}
+    if fault == "unknown_status": source["status"] = "unknown"
+    if fault == "failed_status": source["status"] = "failed"
+    if fault == "errors": source["errors"] = ["fake-error"]
+    if fault == "error": source["error"] = "fake-error"
+    if fault == "pending": source["summary"]["orders_pending_count"] = 1
+    if fault == "orders": source["summary"]["orders_submitted_count"] = 1
+    if fault == "broker": source["summary"]["broker_submission_done"] = True
+    if fault == "unknown_execution": source["summary"]["execution_status"] = "unknown"
+    if fault == "missing_reasons": source["summary"].pop("notification_approved_gates")
+    if fault == "invalid_reasons": source["summary"]["notification_approved_gates"][0]["limit"] = 4
+    if fault == "dry_run": source["dry_run"] = True
+    if fault == "missing_runtime_target": source.pop("runtime_target")
+    if fault == "profile": source["runtime_target"]["strategy_profile"] = "fake-other"
+    if fault == "selector": source["runtime_target"]["account_selector"] = ["fake-other"]
+    if fault == "revision": source["diagnostics"]["runtime_revision"] = "fake-other"
+    if fault == "run_id": source["run_id"] = "fake-other"
+    if fault == "currency": source["summary"]["account_observation"]["net_assets_currency"] = "EUR"
+    if fault == "stale_observation": source["summary"]["account_observation"]["observed_at"] = "2026-09-28T22:21:00Z"
+    if fault == "outside_report": source["summary"]["account_observation"]["observed_at"] = "2026-09-30T22:26:00Z"
+    if fault == "invalid_net_assets": source["summary"]["account_observation"]["net_assets"] = "NaN"
+    assert _project_completed_risk_block(source)["status"] == "skipped"
diff --git a/tests/test_report_field_consistency.py b/tests/test_report_field_consistency.py
new file mode 100644
index 0000000..8826586
--- /dev/null
+++ b/tests/test_report_field_consistency.py
@@ -0,0 +1,147 @@
+"""Early and archived producers retain the same execution facts."""
+import copy
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from application.runtime_report_summary import summarize_execution_cycle_result
+from notifications import unified
+
+
+@pytest.mark.parametrize('status', ['blocked', 'unknown', 'failed', 'no_action', 'pending_reconciliation'])
+@pytest.mark.parametrize('observations', [None, {}, {'watch': 1}])
+def test_early_and_archived_candidates_agree(status, observations):
+    from application.runtime_report_summary import execution_report_status
+    target = dict(platform_id='schwab', strategy_profile='fake-strategy', service_name='fake-service',
+                  account_scope='fake', account_selector=['fake'], deployment_selector='fake', dry_run_only=False)
+    now = datetime.now(timezone.utc)
+    report = dict(schema_version='runtime_report.v1', platform='schwab', runtime_target=target,
+                  strategy_profile='fake-strategy', service_name='fake-service', run_id='fake-run',
+                  started_at=now.isoformat(), dry_run=False, summary={})
+    if observations is not None:
+        report['summary']['routine_observation_counts'] = observations
+    result = SimpleNamespace(execution={'execution_status': status}, submitted_orders=[], trade_logs=[])
+    original = copy.deepcopy(report)
+    with unified.notification_context(report):
+        early = unified._cycle_projection(result, {'opaque_account_uid': 'fake-opaque', 'target_id': 'fake-target'}, SimpleNamespace(dry_run_only=False))
+    assert report == original
+    summary = summarize_execution_cycle_result(result, dry_run=False, existing_summary=report['summary'])
+    report.update(summary=summary, status=execution_report_status(summary))
+    assert report['status'] == ('ok' if status in {'no_action', 'pending_reconciliation'} else 'error')
+    archive = unified._project_runs([report], expected_target=target, opaque_account_uid='fake-opaque', target_id='fake-target', now=now)
+    # Events are merged separately by the real store; compare canonical facts.
+    early["runs"][0].pop("notification_events")
+    archive["runs"][0].pop("notification_events")
+    assert early == archive
+    row = archive['runs'][0]
+    assert ('routine_observation_counts' in row) == (observations is not None)
+    if observations is not None:
+        assert row['routine_observation_counts'] == observations
+    assert row['execution_status'] == ('order_submitted' if status == 'pending_reconciliation' else status)
+
+
+def test_missing_execution_fails_closed():
+    from application.runtime_report_summary import execution_report_status
+    assert execution_report_status({}) == 'error'
+
+
+def test_early_projection_uses_complete_execution_summary(monkeypatch):
+    observed = []
+    monkeypatch.setattr(unified, '_project_runs', lambda reports, **kw: observed.extend(reports) or {'runs': []})
+    result = SimpleNamespace(execution={'execution_status': 'blocked'}, submitted_orders=[], trade_logs=[])
+    source = {'runtime_target': {'platform_id': 'schwab'}, 'summary': {'routine_observation_counts': {}}}
+    with unified.notification_context(source):
+        unified._cycle_projection(result, {'opaque_account_uid': 'fake', 'target_id': 'fake'}, SimpleNamespace(dry_run_only=False))
+    assert observed[0]['status'] == 'error'
+    assert observed[0]['summary']['result'] == 'blocked'
+    assert observed[0]['summary']['routine_observation_counts'] == {}
+
+
+def test_unknown_and_observed_empty_remain_distinct():
+    missing = unified._canonical_candidate_row({'source_run_id': 'fake-run'})
+    empty = unified._canonical_candidate_row({'source_run_id': 'fake-run', 'routine_observation_counts': {}})
+    assert missing != empty
+
+
+def test_current_execution_overrides_stale_summary_without_mutating_it():
+    previous = {'execution_status': 'completed', 'result': 'completed', 'routine_observation_counts': {}}
+    result = SimpleNamespace(execution={'execution_status': 'blocked'}, submitted_orders=[], trade_logs=[])
+    summary = summarize_execution_cycle_result(result, dry_run=False, existing_summary=previous)
+    assert summary['execution_status'] == summary['result'] == 'blocked'
+    assert summary['routine_observation_counts'] == {}
+    assert previous['execution_status'] == 'completed'
+
+
+@pytest.mark.parametrize('previous', [{}, {'execution_status': 'completed', 'result': 'completed'}])
+def test_completed_zero_order_cycle_is_healthy_without_submission_status(monkeypatch, previous):
+    from application.runtime_report_summary import execution_report_status
+    from notifications.accepted_orders import cycle_notification_evidence
+    from scripts import execution_report_heartbeat as heartbeat
+    from test_notification_quiet import NOW, mock_heartbeat
+    from test_runtime_daily_report_projection import report
+
+    result = SimpleNamespace(execution={'notification_cycle_completed': True}, submitted_orders=[], trade_logs=[])
+    before = copy.deepcopy(result.execution)
+    summary = summarize_execution_cycle_result(result, dry_run=False, existing_summary=previous)
+    assert summary['notification_digest'] == cycle_notification_evidence(result.execution, [])
+    assert summary['notification_digest']['routine'] is True
+    assert execution_report_status(summary) == 'ok'
+    assert summary['execution_status'] == summary['result'] == 'no_action'
+    source = report('no_action', summary=summary, status=execution_report_status(summary),
+                    started_at='2026-10-05T12:00:00Z', finished_at='2026-10-05T12:01:00Z', diagnostics={})
+    sent = mock_heartbeat(monkeypatch, source)
+    assert heartbeat.main(NOW) == 0
+    assert sent == []
+    assert result.execution == before
+
+
+@pytest.mark.parametrize('change', [
+    {'notification_cycle_completed': False}, {'notification_cycle_completed': 1},
+    {'notification_order_issue': True}, {'errors': ['fake-error']},
+    {'signal_metadata': {'errors': ['fake-error']}}, {'orders_pending_count': 1},
+    {'broker_submission_done': True}, {'execution_status': 'unknown'},
+    {'execution_status': 'blocked'}, {'execution_status': 'failed'},
+])
+@pytest.mark.parametrize('dry_run', [False, True])
+def test_completion_marker_cannot_hide_missing_or_conflicting_evidence(change, dry_run):
+    from application.runtime_report_summary import execution_report_status
+    result = SimpleNamespace(execution={'notification_cycle_completed': True, **change},
+                             submitted_orders=[], trade_logs=[])
+    summary = summarize_execution_cycle_result(result, dry_run=dry_run)
+    assert execution_report_status(summary) == 'error'
+
+
+def test_completion_marker_does_not_establish_submission_outcome():
+    from application.runtime_report_summary import execution_report_status
+    result = SimpleNamespace(execution={'notification_cycle_completed': True},
+                             submitted_orders=[{'status': 'unknown'}], trade_logs=[])
+    assert execution_report_status(summarize_execution_cycle_result(result, dry_run=False)) == 'error'
+
+
+@pytest.mark.parametrize('previous', [{}, {'execution_status': 'completed', 'result': 'completed'}])
+def test_completed_empty_dry_run_is_healthy_without_live_digest_evidence(previous):
+    from application.runtime_report_summary import execution_report_status
+    from notifications.accepted_orders import cycle_notification_evidence
+
+    result = SimpleNamespace(execution={'notification_cycle_completed': True}, submitted_orders=[], trade_logs=[])
+    before = copy.deepcopy(result)
+    previous_before = copy.deepcopy(previous)
+    summary = summarize_execution_cycle_result(result, dry_run=True, existing_summary=previous)
+    assert execution_report_status(summary) == 'ok'
+    assert summary['execution_status'] == summary['result'] == 'dry_run_completed'
+    assert summary['orders_previewed_count'] == summary['orders_submitted_count'] == 0
+    assert summary['broker_submission_done'] is False
+    assert summary['dry_run_order_preview_available'] is False
+    assert summary['notification_digest'] == cycle_notification_evidence(result.execution, [], dry_run=True)
+    assert summary['notification_digest']['routine'] is False
+    assert summary['notification_digest']['order_count'] is None
+    assert summary['notification_digest']['fill_count'] is None
+    assert result == before and previous == previous_before
+
+
+@pytest.mark.parametrize('dry_run', [False, True])
+def test_empty_execution_without_completion_is_not_healthy(dry_run):
+    from application.runtime_report_summary import execution_report_status
+    result = SimpleNamespace(execution={}, submitted_orders=[], trade_logs=[])
+    assert execution_report_status(summarize_execution_cycle_result(result, dry_run=dry_run)) == 'error'
diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py
index 0053c20..d0746c4 100644
--- a/tests/test_request_handling.py
+++ b/tests/test_request_handling.py
@@ -1023,7 +1023,7 @@ def test_handle_schwab_emits_structured_runtime_events(self):
         )
         self.assertTrue(all(run_id == "run-001" for run_id, _event, _fields in observed))
 
-    def test_handle_schwab_persists_machine_readable_report(self):
+    def test_handle_schwab_persists_unknown_execution_as_error(self):
         module = load_module()
         observed = {}
 
@@ -1038,7 +1038,8 @@ def test_handle_schwab_persists_machine_readable_report(self):
 
         self.assertEqual(status, 200)
         self.assertEqual(body, "OK")
-        self.assertEqual(observed["report"]["status"], "ok")
+        # HTTP completion does not establish a successful missing execution.
+        self.assertEqual(observed["report"]["status"], "error")
         self.assertEqual(observed["report"]["strategy_profile"], "tqqq_growth_income")
         self.assertEqual(
             observed["report"]["summary"]["strategy_display_name"],