Add bounded Earn accounting diagnosis

.github/workflows/main.yml

@@ -52,7 +52,7 @@ on:
         type: choice
         required: false
         default: none
-        options: [none, inspect, quiesce, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply]
+        options: [none, inspect, quiesce, earn-forward-diagnose, audit, preview, scope-preview, cash-flow-preview, rebase-proposal, rebase-apply, prospective-rebase-apply, apply]
       proposal_recipient_certificate:
         description: "Public X.509 encryption certificate for rebase-proposal only; never provide a private key"
         type: string
@@ -132,7 +132,7 @@ jobs:
                 exit 1
               fi
               ;;
-            inspect|quiesce|audit|preview|scope-preview|cash-flow-preview|rebase-proposal|rebase-apply|prospective-rebase-apply|apply)
+            inspect|quiesce|earn-forward-diagnose|audit|preview|scope-preview|cash-flow-preview|rebase-proposal|rebase-apply|prospective-rebase-apply|apply)
               if [ "${RECOVERY_ACTION_INPUT:-none}" != "none" ] || [ "${GITHUB_REF:-}" != "refs/heads/main" ] || [ "${RUNTIME_TARGET_ENABLED:-}" != "false" ] || [ "$RECONCILE_ONLY_INPUT" != "true" ] || [ "$VALIDATE_ONLY_INPUT" = "true" ] || [ "$DIAGNOSE_BALANCES_INPUT" = "true" ] || [ "$RECONCILE_PERSIST_INPUT" = "true" ]; then
                 echo "::error::Accounting migration requires disabled main runtime, reconcile_only, and no other recovery or diagnostic mode."
                 exit 1

README.md

@@ -65,6 +65,11 @@ uv run --no-sync python -m unittest discover -s tests -v
 - [`docs/binance_platform_rename_checklist.md`](docs/binance_platform_rename_checklist.md)
 - [`docs/operator_runbook.md`](docs/operator_runbook.md)
 
+For a forward Earn accounting failure, the disabled `main` workflow has one
+read-only `accounting_migration_action=earn-forward-diagnose` mode. It samples
+from the current ledger checkpoint and reports bounded, redacted diagnostics;
+it never clears an owner, writes accounting state, or grants execution.
+
 ## Community and security
 
 - See [CONTRIBUTING.md](CONTRIBUTING.md) for pull request scope, local verification, and documentation expectations.

application/cycle_service.py

@@ -9,6 +9,7 @@
 from quant_platform_kit.common.runtime_reports import persist_runtime_report
 from quant_platform_kit.strategy_lifecycle.performance_monitor import try_record_platform_execution
 from application.execution_receipt_adapter import attach_execution_receipt_from_report
+from application.portfolio_service import EARN_FORWARD_REASON_CODES
 from runtime_logging import RuntimeLogContext, emit_runtime_log
 from runtime_support import (
     append_report_error, finalize_notification_delivery, acquire_runtime_state_owner,
@@ -334,9 +335,13 @@ def execute_strategy_cycle(
             KeyError: "key_error",
             RuntimeError: "runtime_error",
         }.get(type(exc), "unclassified_error")
-        report.setdefault("diagnostics", {})["cycle_failure"] = {
-            "stage": failure_stage, "error_type": error_type,
-        }
+        failure_metadata = {"stage": failure_stage, "error_type": error_type}
+        if failure_stage == "daily_state":
+            earn_diagnostics = report.get("diagnostics", {}).get("earn_accrual", {})
+            reason_code = earn_diagnostics.get("reason_code") if isinstance(earn_diagnostics, Mapping) else None
+            if reason_code in EARN_FORWARD_REASON_CODES:
+                failure_metadata["reason_code"] = reason_code
+        report.setdefault("diagnostics", {})["cycle_failure"] = failure_metadata
         log_buffer.append(f"cycle_execution_failed stage={failure_stage} error_type={error_type}")
         append_report_error(report, "cycle_execution_failed", stage="execute_cycle")
         try:

application/portfolio_service.py

@@ -11,6 +11,28 @@
 
 
 _TREND_PNL_BASIS = "trend_mark_plus_cash_flow_v1"
+EARN_FORWARD_REASON_CODES = frozenset({
+    "earn_checkpoint_invalid",
+    "earn_checkpoint_time_invalid",
+    "earn_checkpoint_scope_changed",
+    "earn_product_lifecycle_unverified",
+    "earn_counter_reset",
+    "earn_quantity_change_unexplained",
+    "earn_order_unsettled",
+    "earn_accounted_changes_missing",
+    "earn_cash_cursor_mismatch",
+    "earn_cash_flow_invalid",
+    "earn_cash_flow_unsupported",
+    "earn_cash_flow_time_unverified",
+    "earn_valuation_snapshot_mismatch",
+    "external_cash_flow_window_invalid",
+    "external_cash_flow_cursor_invalid",
+    "external_cash_flow_history_read_failed",
+    "external_cash_flow_history_incomplete",
+    "external_cash_flow_record_invalid",
+    "external_cash_flow_record_changed",
+    "external_cash_flow_cursor_capacity_exceeded",
+})
 
 
 def compute_portfolio_allocation(
@@ -71,7 +93,16 @@ def maybe_rebase_daily_state_for_balance_change(
             cash = collect_external_cash_flows_fn(runtime.client, now=_time(current['observed_at']),
                                                   cursor=state.get('external_cash_flow_cursor'))
             updated = prepare_forward_earn_state(state, current, cash)
-        except Exception:
+        except Exception as exc:
+            reason_code = (
+                str(exc)
+                if type(exc) is ValueError and str(exc) in EARN_FORWARD_REASON_CODES
+                else "earn_forward_accounting_unverified"
+            )
+            report.setdefault("diagnostics", {})["earn_accrual"] = {
+                "status": "blocked",
+                "reason_code": reason_code,
+            }
             raise ExecutionIntegrityError("earn_forward_accounting_unverified") from None
         runtime_set_trade_state_fn(runtime, report, updated, reason="earn_forward_accounting")
         state.clear()

docs/operator_runbook.md

@@ -271,6 +271,20 @@ or changing balances/ledger blocks the audit. A match
 does not reconcile the whole account, authorize migration, or restore trading;
 the existing zero-activity preview/apply checks remain unchanged.
 
+When the forward Earn consumer reports `earn_quantity_change_unexplained`, use
+the separate `accounting_migration_action=earn-forward-diagnose` once with the
+runtime disabled on `main` and `reconcile_only=true`. It starts at the ledger's
+current `earn_accrual_checkpoint`, reads current Spot/Flexible Earn, bounded
+trades, the existing cash-flow cursor, and bounded reward history, then reads
+the three Firestore documents again. It reports only asset names, directions,
+product and reward counts, matching flags, owner existence, and fixed no-write
+policy flags. An existing owner is observed and reported; it is never cleared
+or bypassed. Any ledger, control, or owner change during sampling discards the
+result. A matching BONUS or REALTIME record is diagnostic evidence only and
+never a causal reconciliation or execution permission. The bounded trade net is
+reconstructed from normalized `myTrades` quantity and price fields for
+diagnosis only; it is not a complete fill or accounting proof.
+
 The migration is a separate, one-time `Runtime` workflow mode for an old
 `trend_val` ledger. It does not activate recovery control, grant execution
 authority, clear the circuit-breaker latch, or reconstruct historical

scripts/execution_report_heartbeat.py

@@ -14,6 +14,8 @@
 
 _SCHEMA = "qsl.execution_report_heartbeat_assessment.v1"
 _ACCEPTED_STATUSES = frozenset({"ok", "skipped", "success", "completed", "no_action", "aborted"})
+_OPERATIONAL_BLOCKERS = frozenset({"state_owner_busy"})
+_OPERATIONAL_BLOCKER_REASON = "execution is blocked by an operational guard"
 
 
 def _env_bool(name: str, default: bool = False) -> bool:
@@ -136,6 +138,11 @@ def _payload_strategy(payload: dict[str, Any]) -> str:
     return str(payload.get("strategy_profile") or target_strategy or "").strip()
 
 
+def _operational_blocker(payload: dict[str, Any]) -> str | None:
+    value = payload.get("execution_blocked_reason")
+    return value if isinstance(value, str) and value in _OPERATIONAL_BLOCKERS else None
+
+
 def _accepted_payload(payload: dict[str, Any]) -> tuple[bool, str]:
     expected_platform = (os.environ.get("RUNTIME_HEARTBEAT_REPORT_PLATFORM") or "binance").strip().lower()
     expected_strategy = (os.environ.get("RUNTIME_HEARTBEAT_STRATEGY_PROFILE") or "").strip()
@@ -146,6 +153,8 @@ def _accepted_payload(payload: dict[str, Any]) -> tuple[bool, str]:
         return False, "strategy profile does not match"
     if expected_service and _payload_service_name(payload) != expected_service:
         return False, "service name does not match"
+    if _operational_blocker(payload) is not None:
+        return False, _OPERATIONAL_BLOCKER_REASON
     errors = payload.get("errors")
     if isinstance(errors, list) and errors:
         return False, "report contains errors"
@@ -242,6 +251,17 @@ def assess_execution_report_heartbeat(now: dt.datetime | None = None) -> dict[st
         if payload is None:
             continue
         accepted, reason = _accepted_payload(payload)
+        if not accepted and reason == _OPERATIONAL_BLOCKER_REASON:
+            blocker = _operational_blocker(payload)
+            return {
+                "schema": _SCHEMA,
+                "observed_at": current.isoformat().replace("+00:00", "Z"),
+                "status": "alert",
+                "reason": f"execution_blocked:{blocker}",
+                "name": name,
+                "report_updated_at": (_entry_updated_at(entry) or current).isoformat().replace("+00:00", "Z"),
+                "reports_returned": len(entries),
+            }
         if accepted:
             assessment = {
                 "schema": _SCHEMA,