fix(binance): bound daily state recovery diagnostics

application/broker_reconciliation.py

@@ -832,8 +832,12 @@ def diagnose_bnb_wallet_activity(
     if start.tzinfo is None or end.tzinfo is None or not start < end or end-start > timedelta(days=7):
         raise ValueError("balance_history_window_invalid")
     bounds = {"startTime": int(start.timestamp()*1000), "endTime": int(end.timestamp()*1000)}
-    result = {"requested_surfaces_complete": False, "counts": {},
-              "complete_balance_reconciliation": False}
+    result = {
+        "requested_surfaces_complete": False,
+        "counts": {},
+        "complete_balance_reconciliation": False,
+        "surface_diagnostics": {},
+    }
     for name, path, params, rows_key, time_key, limit in (
         ("bnb_dividends", "asset/assetDividend", {"asset": "BNB", "limit": 500}, "rows", "divTime", 500),
         ("spot_dust_conversions", "asset/dribblet", {"accountType": "SPOT"}, "userAssetDribblets", "operateTime", 100),
@@ -864,17 +868,53 @@ def diagnose_bnb_wallet_activity(
             "total_matches_rows": isinstance(rows, list) and count == len(rows),
             "page_full": isinstance(rows, list) and len(rows) >= limit,
         }
-        valid_rows = isinstance(rows, list) and all(
-            isinstance(row, Mapping) and type(row.get(time_key)) is int
-            and bounds["startTime"] <= row[time_key] <= bounds["endTime"]
-            and (name != "bnb_dividends" or row.get("asset") == "BNB") for row in rows
-        )
+        rows_readable = isinstance(rows, list) and all(isinstance(row, Mapping) for row in rows)
+        visible_row_count = len(rows) if isinstance(rows, list) else 0
+        row_keys = set()
+        valid_rows = rows_readable
+        if valid_rows:
+            for row in rows:
+                timestamp = row.get(time_key)
+                if (
+                    type(timestamp) is not int
+                    or not bounds["startTime"] <= timestamp <= bounds["endTime"]
+                    or (name == "bnb_dividends" and row.get("asset") != "BNB")
+                ):
+                    valid_rows = False
+                    break
+                if name == "bnb_dividends":
+                    identity = (row.get("id"), row.get("tranId"), timestamp)
+                    if (
+                        type(identity[0]) is not int
+                        or identity[0] < 0
+                        or type(identity[1]) is not int
+                        or identity[1] < 0
+                        or identity in row_keys
+                    ):
+                        valid_rows = False
+                        break
+                else:
+                    identity = row.get("transId")
+                    if type(identity) is not int or identity < 0 or identity in row_keys:
+                        valid_rows = False
+                        break
+                row_keys.add(identity)
+        shape.update({
+            "total_valid": count is not None and count >= 0,
+            "rows_readable": rows_readable,
+            "visible_row_count": visible_row_count,
+            "window_complete": (
+                count is not None and count >= 0 and shape["total_matches_rows"]
+                and count < limit and valid_rows
+            ),
+        })
         shape["row_time_and_asset_valid"] = valid_rows
-        if (count is None or count < 0 or not shape["total_matches_rows"]
-                or count >= limit or not valid_rows):
+        result["surface_diagnostics"][name] = shape
+        if isinstance(rows, list) and rows_readable:
+            result["counts"][name] = visible_row_count
+            if include_rows:
+                result.setdefault("_private_rows", {})[name] = rows
+        if not shape["window_complete"]:
             return {**result, "reason_code": "bnb_wallet_history_unverified", "failed_surface": name,
                     "failure_stage": "response_validation", "response_shape": shape}
-        result["counts"][name] = len(rows)
-        if include_rows:
-            result.setdefault("_private_rows", {})[name] = rows
     return {**result, "requested_surfaces_complete": True}

application/cycle_service.py

@@ -29,6 +29,16 @@ def _record_risk_diagnostics(report, allocation):
     report["risk_flags"] = list(allocation.get("risk_flags") or ())
 
 
+def _settled_order_state(state):
+    if not isinstance(state, Mapping):
+        return None
+    record = state.get("order_submission")
+    if not isinstance(record, Mapping):
+        return None
+    status = record.get("state")
+    return status if status in {"RESERVED", "TERMINAL"} else None
+
+
 def execute_strategy_cycle(
     runtime,
     *,
@@ -77,10 +87,33 @@ def execute_strategy_cycle(
 
     state_healthy = False
     failure_stage = "state_owner_claim"
+    owner_claimed_this_cycle = False
+    initial_order_state = None
+    daily_state_write_intents_start = None
+    daily_funding_submission_start = None
+    daily_funding_side_effect_start = None
+    daily_funding_intent_lengths = None
+    daily_order_sequence_start = None
+    cycle_funding_submission_start = None
+    cycle_funding_side_effect_start = None
+    cycle_order_sequence_start = None
     try:
         if not acquire_runtime_state_owner(runtime):
             report["execution_blocked_reason"] = "state_owner_busy"
             return report
+        owner_claimed_this_cycle = (
+            not getattr(runtime, "dry_run", False)
+            and getattr(runtime, "standard_execution_permitted", True)
+            and getattr(runtime, "state_owner_held", False)
+        )
+        receipt_observation = report.get("execution_receipt_observation", {})
+        cycle_funding_submission_start = (
+            receipt_observation.get("submission_attempted_count", 0)
+            if isinstance(receipt_observation, Mapping)
+            else 0
+        )
+        cycle_funding_side_effect_start = len(getattr(runtime, "side_effect_log", ()))
+        cycle_order_sequence_start = getattr(runtime, "order_sequence", 0)
         failure_stage = "client_connect"
         if not ensure_runtime_client(runtime, report):
             return report
@@ -92,6 +125,7 @@ def execute_strategy_cycle(
 
         state, trend_pool_resolution, runtime_trend_universe, allow_new_trend_entries = cycle_state
         runtime.trade_state = state
+        initial_order_state = _settled_order_state(state)
         failure_stage = "funding_reconciliation"
         reconcile_pending_funding_submission(runtime)
         submission_state = state.get("order_submission", {}).get("state", "RESERVED")
@@ -148,6 +182,19 @@ def execute_strategy_cycle(
             return report
 
         failure_stage = "daily_state"
+        daily_state_write_intents_start = len(report.get("state_write_intents", ()))
+        receipt_observation = report.get("execution_receipt_observation", {})
+        daily_funding_submission_start = (
+            receipt_observation.get("submission_attempted_count", 0)
+            if isinstance(receipt_observation, Mapping)
+            else 0
+        )
+        daily_funding_side_effect_start = len(getattr(runtime, "side_effect_log", ()))
+        daily_order_sequence_start = getattr(runtime, "order_sequence", 0)
+        daily_funding_intent_lengths = tuple(
+            len(report.get(key, ())) if isinstance(report.get(key, ()), list) else None
+            for key in ("buy_sell_intents", "btc_dca_intents", "redemption_subscription_intents")
+        )
         now_utc = runtime.now_utc
         today_utc = now_utc.strftime("%Y-%m-%d")
         today_id_str = now_utc.strftime("%Y%m%d")
@@ -349,7 +396,78 @@ def execute_strategy_cycle(
         except Exception:
             pass
     finally:
-        if state_healthy and getattr(runtime, "state_owner_held", False):
+        reason_code = None
+        earn_diagnostics = report.get("diagnostics", {}).get("earn_accrual", {})
+        if isinstance(earn_diagnostics, Mapping):
+            reason_code = earn_diagnostics.get("reason_code")
+        current_order_state = _settled_order_state(getattr(runtime, "trade_state", None) or locals().get("state"))
+        current_receipt_observation = report.get("execution_receipt_observation", {})
+        funding_submission_unchanged = (
+            isinstance(current_receipt_observation, Mapping)
+            and current_receipt_observation.get("submission_attempted_count", 0)
+            == daily_funding_submission_start
+        )
+        funding_submission_unchanged_full_cycle = (
+            isinstance(current_receipt_observation, Mapping)
+            and current_receipt_observation.get("submission_attempted_count", 0)
+            == cycle_funding_submission_start
+        )
+        current_intent_lengths = tuple(
+            len(report.get(key, ())) if isinstance(report.get(key, ()), list) else None
+            for key in ("buy_sell_intents", "btc_dca_intents", "redemption_subscription_intents")
+        )
+        funding_intents_unchanged = current_intent_lengths == daily_funding_intent_lengths
+        funding_order_sequence_unchanged = getattr(runtime, "order_sequence", 0) == daily_order_sequence_start
+        funding_order_sequence_unchanged_full_cycle = (
+            getattr(runtime, "order_sequence", 0) == cycle_order_sequence_start
+        )
+        full_cycle_side_effects = list(getattr(runtime, "side_effect_log", ()))[cycle_funding_side_effect_start or 0:]
+        full_cycle_funding_side_effects_absent = all(
+            not str(entry.get("effect_type", "")).startswith(("order_", "earn_"))
+            for entry in full_cycle_side_effects
+            if isinstance(entry, Mapping)
+        )
+        new_side_effects = list(getattr(runtime, "side_effect_log", ()))[daily_funding_side_effect_start or 0:]
+        funding_side_effects_absent = all(
+            not str(entry.get("effect_type", "")).startswith(("order_", "earn_"))
+            for entry in new_side_effects
+            if isinstance(entry, Mapping)
+        )
+        persistent_side_effects_absent = all(
+            not (
+                str(entry.get("target", "")) == "firestore"
+                or str(entry.get("effect_type", "")).startswith("state_")
+            )
+            for entry in new_side_effects
+            if isinstance(entry, Mapping)
+        )
+        pure_daily_state_failure = (
+            not state_healthy
+            and failure_stage == "daily_state"
+            and reason_code in EARN_FORWARD_REASON_CODES
+            and owner_claimed_this_cycle
+            and initial_order_state in {"RESERVED", "TERMINAL"}
+            and current_order_state in {"RESERVED", "TERMINAL"}
+            and not getattr(runtime, "pending_funds", ())
+            and funding_submission_unchanged
+            and funding_submission_unchanged_full_cycle
+            and funding_intents_unchanged
+            and funding_order_sequence_unchanged
+            and funding_order_sequence_unchanged_full_cycle
+            and funding_side_effects_absent
+            and full_cycle_funding_side_effects_absent
+            and persistent_side_effects_absent
+            and isinstance(report.get("state_write_intents"), list)
+            and daily_state_write_intents_start is not None
+            and len(report["state_write_intents"]) == daily_state_write_intents_start
+        )
+        if pure_daily_state_failure:
+            try:
+                release_runtime_state_owner(runtime)
+            except ExecutionIntegrityError:
+                report["status"] = "error"
+                append_report_error(report, "state_owner_release_uncertain", stage="state_release")
+        elif state_healthy and getattr(runtime, "state_owner_held", False):
             try:
                 release_runtime_state_owner(runtime)
             except ExecutionIntegrityError:

docs/binance_reconciliation_recovery.zh-CN.md

@@ -128,15 +128,27 @@ Spot + Flexible Earn 持有资产口径，闲置理财不等于资金退出策
 ## 向前收益记账消费者（2026-09-12）
 
 仅账本已含 `earn_accrual_checkpoint` 时启用新路径；旧账本继续使用原路径，不自动迁移。
-加载时保留 checkpoint 与 `earn_accounted_net_changes`。市场估值与记账使用同一次完整 Spot/Earn
-数量采样；既有 owner-protected writer 一次保存下一检查点、流水游标、余额及清零后的已记成交净变动。
+加载时保留 checkpoint 与 `earn_accounted_net_changes`。诊断和恢复核对按
+`Spot1 → Earn1 → Spot2 → Earn2` 顺序做有限双采样，比较两次 Spot/Earn 的数量、产品和实时收益计数；
+`sampling_stable` 还要求 Spot/Earn 分量变化可由实时收益计数解释，不再仅由 Firestore owner/账本/control
+标记稳定推出。市场估值与记账使用一次完整
+Spot/Earn 数量采样；既有 owner-protected writer 一次保存下一检查点、流水游标、余额及清零后的已记成交净变动。
 验证失败不推进；写入失败不更新内存检查点，也不自动重试。
 
 数量变化按同产品实时计数增量、已持久化的完整成交净数量/手续费、已核实当日 USDT 入金核对。
 成交在原 FILLED_ACCOUNTING_PENDING → TERMINAL 记账步骤累计原始 Decimal 数量，费用从对应资产扣除；
 不通过余额差反推资金来源。中途资金核验可更新当前余额，但不单独推进收益检查点或流水游标，避免漏掉区间资金活动。
 理财奖励已体现在权益中，不再加一次利润，也不记作外部本金。
 
+`earn-forward-diagnose` 只在内存保留原始采样行，公开结果只报告资产方向、收益/资金来源残差方向、
+记录条数和匹配标志。BNB wallet 的 `dribblet` 诊断分别报告行是否可读、可见行数和窗口是否完整；
+缺失或非法 `total`、满页、重复、越界或非 BNB dividend 行都会保持未核实，不会把已见有效行静默丢弃，
+也不会将不完整窗口写成完整资金来源。
+
+若既有运行仅在 `daily_state` 的纯 Earn 准备校验失败，且本轮 owner 已成功取得、加载时订单明确为
+`RESERVED/TERMINAL` 并持续 settled、无 pending funds、准备阶段没有新增持久写入意图，则只释放本轮自己的 owner。
+未知订单、写入结果不明或旧 owner 继续保留，旧 owner 不由诊断或运行补丁自动清除。
+
 未知/未完成订单、产品消失或更换、计数回退、新提币及未支持币种入金继续拒绝。内部申赎不算收益；
 只有同产品的总数量守恒可直接通过，产品生命周期不连续时不能猜测新的累计计数起点。
 这不是所有资金活动均可自动分类的声明。

docs/operator_runbook.md

@@ -274,23 +274,37 @@ 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
+current `earn_accrual_checkpoint`, reads Spot then Flexible Earn twice in the
+bounded order `Spot1 → Earn1 → Spot2 → Earn2`, 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.
+product and reward counts, matching flags, residual breakdown directions, owner
+existence, and fixed no-write policy flags. `sampling_stable` is true only when
+the two in-memory Spot/Earn shapes and Spot/Earn component movements match,
+with quantity changes explained by realtime counters; stable Firestore markers
+alone are insufficient. 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.
 When BNB has a positive residual while the other sampled checks are stable, the
 same diagnostic may read the bounded `assetDividend` and Spot `dribblet` wallet
 surfaces. It reports record counts and exact Decimal comparisons for dividend,
 transfer, fee-adjusted, and combined values without exposing amounts or source
-rows. Binance's transfer and fee semantics remain unverified, and the separate
-one-eight-decimal-unit flag is diagnostic only; neither flag changes accounting,
-owner, control, or execution state.
+rows. Row readability, visible row count, and window completeness are separate
+fields. A missing or invalid `total`, a full page, a duplicate, an out-of-window
+row, or a non-BNB dividend row keeps the window unverified even when valid rows
+are visible. Binance's transfer and fee semantics remain unverified, and the
+separate one-eight-decimal-unit flag is diagnostic only; neither flag changes
+accounting, owner, control, or execution state.
+
+If a cycle reaches `daily_state` and fails only during the pure Earn preparation
+check, the runtime releases its own newly claimed owner only when the initial
+and current order states are explicitly `RESERVED` or `TERMINAL`, no funds are
+pending, and no state-write intent was added after the state-load metadata
+refresh. Unknown orders, uncertain writes, or an owner held before this cycle
+remain locked for manual reconciliation.
 
 The migration is a separate, one-time `Runtime` workflow mode for an old
 `trend_val` ledger. It does not activate recovery control, grant execution