Expose declared stop categories for missing account snapshot investigation

scripts/README_account_facts_projection.md

@@ -23,3 +23,7 @@ Each stage is only `verified`, `invalid` or `unknown`: report envelope, report-p
 `snapshot_issue` classifies the same existing snapshot-schema stage using only fixed codes. Before scope proof, preflight or an unavailable report, it is `unknown`; a verified schema/currency stage yields `none`. An invalid stage is classified in predicate order as `summary_absent`, `summary_not_object`, `account_facts_absent`, `account_facts_not_object`, `snapshot_schema_missing`, `snapshot_schema_invalid`, `currency_missing` or `currency_invalid`. Explicit null or a wrong type is invalid/not-object rather than an absent field; unexpected classification errors remain unknown. The classifier reads the same report in memory and leaves all eight stage meanings and the original rejection/CLI exit unchanged.
 
 `none` proves only the existing snapshot schema and USD label predicates, even if account, observation, balances or positions later fail. A fixed issue identifies the observed shape, not why the producer omitted or malformed it. Current source can omit facts before portfolio work or when native facts/account-selector checks fail; the diagnostic does not choose among those causes, copy report status/error text, synthesize missing facts, convert currencies, select an older report or claim the serving source is approved.
+
+`report_stop_reason` exposes only an allowlisted category already declared by the same report, after existing envelope, provenance and scope checks pass. A skipped report may declare `already_running`, `market_closed` or `ibkr_fractional_equity_api_unsupported`; an error may declare `strategy_execution_blocked`, `strategy_deadline`, `ibkr_gateway_unavailable`, `ibkr_trading_permission` or `ibkr_portfolio_snapshot`. Unknown categories become fixed `unclassified_skip`/`unclassified_error`; contradictory status/reason fields become `invalid_stop_context`. Preflight, failed reads, unproven scope, pending/unrecognized status and unexpected optional errors remain `unknown`. `no_declared_stop` means only an `ok` report has neither stop field.
+
+The category is a report declaration, not independent proof of the running process, the reason account facts were omitted, gateway credentials, account ownership or financial accuracy. It exposes no free text, error bodies, counts or private values. The same report is reused without additional cloud or broker reads; original stages, projection guard and CLI exit remain unchanged.

scripts/diagnose_account_facts_identity.py

@@ -91,6 +91,7 @@ def _result(*, source_sha: object = None, reason: str, status: str = "unavailabl
         "serving_traffic": _traffic_result(),
         "account_facts_stages": dict.fromkeys(_ACCOUNT_FACTS_STAGES, "unknown"),
         "snapshot_issue": "unknown",
+        "report_stop_reason": "unknown",
     }
 
 
@@ -275,6 +276,40 @@ def _snapshot_issue(report: object, stages: Mapping[str, str]) -> str:
     return "unknown"
 
 
+def _report_stop_reason(report: object, stages: Mapping[str, str]) -> str:
+    """Expose a declared fixed category, never infer an omission's cause."""
+    try:
+        if any(stages.get(key) != "verified" for key in (
+            "report_envelope", "source_report_provenance", "scope_identity",
+        )) or not isinstance(report, Mapping):
+            return "unknown"
+        diagnostics = report.get("diagnostics")
+        if not isinstance(diagnostics, Mapping):
+            return "unknown"
+        status = report.get("status")
+        skip, failure = diagnostics.get("skip_reason"), diagnostics.get("failure_category")
+        if type(status) is not str:
+            return "unknown"
+        if status == "skipped":
+            if failure not in (None, ""):
+                return "invalid_stop_context"
+            allowed = {"already_running", "market_closed", "ibkr_fractional_equity_api_unsupported"}
+            return skip if type(skip) is str and skip in allowed else "unclassified_skip"
+        if status == "error":
+            if skip not in (None, ""):
+                return "invalid_stop_context"
+            allowed = {
+                "strategy_execution_blocked", "strategy_deadline", "ibkr_gateway_unavailable",
+                "ibkr_trading_permission", "ibkr_portfolio_snapshot",
+            }
+            return failure if type(failure) is str and failure in allowed else "unclassified_error"
+        if status == "ok":
+            return "no_declared_stop" if skip in (None, "") and failure in (None, "") else "invalid_stop_context"
+    except Exception:
+        pass  # Malformed optional context cannot change the original guard.
+    return "unknown"
+
+
 def diagnose_identity(
     report: Mapping[str, Any], *, expected: Mapping[str, Any], source_report_uri: str,
     now: datetime, cloud_run_metadata: object = None, source_sha: object = None,
@@ -330,6 +365,7 @@ def diagnose_identity(
         report, expected=expected, source_report_uri=source_report_uri,
     )
     result["snapshot_issue"] = _snapshot_issue(report, result["account_facts_stages"])
+    result["report_stop_reason"] = _report_stop_reason(report, result["account_facts_stages"])
     return result
 
 

tests/test_diagnose_account_facts_identity.py

@@ -1010,3 +1010,77 @@ def cloud_read(*args, **kwargs):
     assert reads == ["report", "cloud"]
     assert PRIVATE not in captured.out + captured.err
     assert not any(field in result for field in ("report_status", "skip_reason", "failure_category"))
+
+
+@pytest.mark.parametrize("status,field,reason", [
+    ("skipped", "skip_reason", "already_running"),
+    ("skipped", "skip_reason", "market_closed"),
+    ("skipped", "skip_reason", "ibkr_fractional_equity_api_unsupported"),
+    ("error", "failure_category", "strategy_execution_blocked"),
+    ("error", "failure_category", "strategy_deadline"),
+    ("error", "failure_category", "ibkr_gateway_unavailable"),
+    ("error", "failure_category", "ibkr_trading_permission"),
+    ("error", "failure_category", "ibkr_portfolio_snapshot"),
+])
+def test_report_stop_reason_is_declared_category_not_inferred_snapshot_cause(monkeypatch, status, field, reason):
+    from copy import deepcopy
+
+    expected, report, _ = _staged_report(monkeypatch)
+    report["summary"].pop("account_facts")
+    report["status"] = status
+    report["diagnostics"][field] = reason
+    original_report, original_expected = deepcopy(report), deepcopy(expected)
+    result = _diagnose(report, expected)
+    assert result["report_stop_reason"] == reason
+    assert result["snapshot_issue"] == "account_facts_absent"
+    assert result["reason"] == "runtime_target_mismatch"
+    assert result["projection_passed"] is False
+    assert report == original_report and expected == original_expected
+    assert PRIVATE not in json.dumps(result)
+
+
+@pytest.mark.parametrize("status,skip,failure,reason", [
+    ("ok", None, None, "no_declared_stop"),
+    ("skipped", PRIVATE, None, "unclassified_skip"),
+    ("error", None, PRIVATE, "unclassified_error"),
+    ("skipped", [], None, "unclassified_skip"),
+    ("error", None, {}, "unclassified_error"),
+    ("ok", "market_closed", None, "invalid_stop_context"),
+    ("skipped", "market_closed", "strategy_deadline", "invalid_stop_context"),
+    ("error", "market_closed", "strategy_deadline", "invalid_stop_context"),
+    ("pending", None, None, "unknown"),
+    (PRIVATE, None, None, "unknown"),
+    ([], None, None, "unknown"),
+])
+def test_report_stop_reason_rejects_conflicts_and_does_not_echo_free_text(monkeypatch, status, skip, failure, reason):
+    expected, report, _ = _staged_report(monkeypatch)
+    report["status"] = status
+    report["diagnostics"].update(skip_reason=skip, failure_category=failure)
+    result = _diagnose(report, expected)
+    assert result["report_stop_reason"] == reason
+    assert PRIVATE not in json.dumps(result)
+    assert result["projection_passed"] is False
+
+
+@pytest.mark.parametrize("broken", ["envelope", "scope", "provenance"])
+def test_report_stop_reason_is_unknown_before_existing_scope_proof(monkeypatch, broken):
+    expected, report, _ = _staged_report(monkeypatch)
+    report["status"] = "skipped"
+    report["diagnostics"]["skip_reason"] = "market_closed"
+    uri = expected["expected_report_prefix"] + "/r.json"
+    if broken == "envelope":
+        report["schema_version"] = PRIVATE
+    elif broken == "scope":
+        report["service_name"] = PRIVATE + "-other"
+    else:
+        uri = "gs://other-private-marker/r.json"
+    result = diagnostic.diagnose_identity(report, expected=expected, source_report_uri=uri, now=NOW, source_sha=SHA)
+    assert result["report_stop_reason"] == "unknown"
+    assert result["projection_passed"] is False
+
+
+def test_report_stop_reason_preflight_preserves_zero_cloud_calls(monkeypatch, capsys):
+    _environment(monkeypatch)
+    assert diagnostic.main(["--preflight"]) == 0
+    result = json.loads(capsys.readouterr().out)
+    assert result["report_stop_reason"] == "unknown"