fix(decision-context): guarded recovery for unreplayable capture batches

loopx/capabilities/decision_context/README.md

@@ -312,10 +312,9 @@ substitute capture cursors for that file or manually manufacture reviewed cursor
 If several settlements occur between ticks, their intermediate transitions may
 be unobservable; ambiguous batches are retained, not inferred to be reviewed.
 An older spool without review observations is baselined without retiring rows.
-For either hold, reconcile against actual review evidence explicitly; if starting
-a new spool after a current-source rebase, retain the old spool as a private
-checkpoint. This conservative protocol does not promise automatic queue drainage
-after skipped review transitions.
+For either hold, reconcile against actual review evidence explicitly, or use the
+guarded recovery below. This conservative protocol does not promise automatic
+queue drainage after skipped review transitions.
 
 This is a change-reference spool, **not a lossless source archive**. First-scan
 history, pagination, late edits, deletion visibility and deadlines remain provider
@@ -329,6 +328,68 @@ Existing reviewable batches remain private and can still be prepared. To roll
 back to an older release, also remove the three new automation fields; retain
 the spool as a private checkpoint rather than deleting unreviewed work.
 
+#### Recover an unreplayable source without discarding history
+
+Recovery belongs to this capability's existing private SQLite spool, not Core
+Goal lifecycle. The local host/CLI is the operator surface; no provider, model,
+remote write permission, scheduler or dashboard setting is added. A trusted
+host may call `diagnose_capture_source` / `recover_capture_source` from
+`loopx.capabilities.decision_context.capture_recovery` with its existing
+`source_provider_overrides`.
+
+1. Run `capture-diagnose` with the same `--goal-id`, `--agent-id`, `--profile`,
+   `--spool`, `--cursor-state` arguments and an explicit `--source-id`.
+   Metadata-only diagnostics never contact providers. Add `--probe` to perform
+   one bounded transient replay/exact-read check, without semantic review or
+   pending settlement. Results distinguish `replay_not_checked`, `replayable`,
+   `revision_unavailable`, `binding_changed`, `cursor_diverged`,
+   `probe_unavailable`, `state_changed`, `empty` and `acquisition_held`.
+2. Preview `capture-recovery --action hold` with that same scope. It shows
+   affected counts and an opaque `preview_token`. Apply only with explicit
+   operator authorization, `--execute --expected-token <preview_token>`.
+   All pending references for that source move atomically to **held, unresolved
+   history**, byte-for-byte, and acquisition for that source pauses. They are
+   not marked reviewed. Other sources can use the released active capacity.
+3. When ready to read current material, preview and apply `--action restart`
+   with a fresh token. This rebinds acquisition to the current profile and the
+   **unchanged settlement-owned reviewed cursor**, clears the source's interval
+   wait, and removes its acquisition hold. The next ordinary capture produces
+   a fresh batch for `prepare-captured` and normal `settle-review`. Older held
+   references remain unresolved, even after the new batch is reviewed.
+4. `--action rollback --recovery-id <applied-recovery-id>` also requires preview
+   and explicit apply. It restores that operation's prior source state only if
+   the source scope and profile/reviewed files still match its receipt and the
+   active capacity permits restoration. After new capture/review, it fails
+   closed rather than overwriting progress. The applied/rolled-back receipts
+   remain in the private spool; copy/export the spool securely for inspection.
+
+All apply tokens bind action, source, profile/binding, spool identity and full
+queue frontier, reviewed-file digest/identity and audit state. A concurrent
+capture or review, duplicate apply, rebind or file ABA requires a fresh preview.
+SQLite serializes capture/recovery; apply uses the settlement cursor lock.
+Arbitrary manual edits are unsupported. Recovery never writes reviewed cursors,
+so it does not invalidate or supersede a legitimate separate review settlement.
+It is an acquisition restart, **not a claim to have created a new review epoch**.
+
+The existing `max_pending_batches=N` still bounds active batches. Held history
+has a separate cap of N; new hold/restart operations stop after 2N audit records
+(at most one rollback per applicable receipt). No automatic eviction, compaction
+or repeated capacity increase is performed. At that bound, preserve/export the
+private spool and make an explicit retention decision; increasing the limit is
+not evidence consumption. A hold is explicit, not an automatic fairness policy.
+It can isolate a noisy source, but exhaustion can recur if other sources are
+not reviewed. Backpressured sources now retry on the next tick when capacity is
+available instead of waiting an additional scan interval.
+
+`capture-status` separates active `pending_batch_count`, unresolved
+`held_batch_count`, per-source `acquisition_held` and
+`semantic_review_completion=not_inferred_from_capture`. `last_checked_at` is
+the last attempt, not necessarily a successful scan; host service liveness and
+successful-scan timestamps remain separate. No status-only call proves historical
+replay or complete decision coverage. Disable capture using the existing profile
+switch; stop the scheduler before downgrading, since older runtimes do not honor
+recovery holds. Retain the spool/receipts rather than treating downgrade as rollback.
+
 ## Relationship To Other Capabilities
 
 | Capability | Primary question | Relationship |

loopx/capabilities/decision_context/README.zh-CN.md

@@ -273,8 +273,8 @@ loopx decision-context prepare-captured --goal-id <goal-id> --agent-id <agent-id
 
 两次采集之间若发生多次 settlement，中间变化可能无法观察；有歧义的批次保留，
 不推断为已审阅。旧 spool 没有观察记录时，仅建立基线，不清理已有批次。
-这两类阻塞均需依据真实审阅证据显式核对；若在当前来源 rebase 后启用新 spool，
-旧 spool 仍须保留为私有检查点。本协议不保证跳过审阅变化后自动排空队列。
+这两类阻塞均需依据真实审阅证据显式核对，或使用下述受保护的恢复入口。
+本协议不保证跳过审阅变化后自动排空队列。
 
 这是变更引用队列，**不是无损历史归档**。首轮历史范围、分页、旧消息编辑/删除可见性、
 超时依然由 provider 保证。回读要求同一边界能确定性复现；历史版本已不可读时明确
@@ -289,6 +289,48 @@ loopx decision-context prepare-captured --goal-id <goal-id> --agent-id <agent-id
 python3 -m pytest -q tests/capabilities/test_decision_context_capture.py
 ```
 
+#### 无损恢复无法重放的来源
+
+恢复由现有 `decision_context` 私有 SQLite spool 负责，不改变 Core 生命周期。
+入口是本地 CLI／受信任 host，不新增 provider、模型调用、远程权限或调度器配置。
+私有 host 可从 `loopx.capabilities.decision_context.capture_recovery` 调用
+`diagnose_capture_source` / `recover_capture_source`，继续传现有 provider overrides。
+
+1. 用相同的 `--goal-id`、`--agent-id`、`--profile`、`--spool`、`--cursor-state`
+   以及明确的 `--source-id` 调用 `capture-diagnose`。默认只读元数据；加 `--probe`
+   才做一次有界、瞬态的 replay / exact read，不做语义审阅或待结算写入。
+   区分未检查、可重放、revision 不可用、binding 改变、cursor 分叉、probe 不可用、
+   状态并发改变、空队列及 acquisition held，不把 provider 原始异常写进输出。
+2. `capture-recovery --action hold` 先预览；确认影响范围后，以
+   `--execute --expected-token <preview_token>` 显式应用。该来源所有待审阅引用
+   原样转入 **held、未解决历史**，暂停其采集，释放活跃队列容量给其他来源。
+   这不是审阅完成，不修改 reviewed cursor，也不丢弃旧证据。
+3. 准备读取当前材料时，重新预览并应用 `--action restart`。它以当前 profile
+   binding 和 **未改动的审阅游标** 重启采集，清除此来源的扫描间隔等待。
+   下一次正常 capture 产生新批次，再走 `prepare-captured` / `settle-review`。
+   新批次被审阅，也不会把旧 held 引用变成已审阅。
+4. `--action rollback --recovery-id <已应用回执 ID>` 同样需要预览和显式应用。
+   只有来源状态、profile／reviewed 文件仍匹配回执且恢复后不超过容量上限时，
+   才恢复操作前状态；新采集或审阅后拒绝覆盖进展。回执保留在私有 spool 中。
+
+预览令牌绑定 action、source、profile／binding、spool 身份、完整队列前沿、
+审阅文件内容与文件身份，以及审计记录。并发采集／审阅、重复应用、重绑或文件 ABA
+必须重新预览。SQLite 串行化采集与恢复写入，恢复使用与 settlement 相同的游标锁。
+不支持手工改库绕过门禁；恢复不替代合法的独立审阅结算，也不宣称创建新审阅 epoch。
+
+`max_pending_batches=N` 继续限制活跃批次；另最多保留 N 条未解决历史，
+2N 条审计记录后停止新增 hold/restart（每条适用回执最多再 rollback 一次）。
+不会自动删除、压缩或无限扩容。达到上限需保留／导出私有 spool 后明确处理保留策略。
+这是显式来源隔离，不是默认公平调度；若其他来源长期不被审阅，仍可能再次背压。
+行为变化：曾背压的来源在容量释放后的下一 tick 可重试，不再多等一个扫描间隔。
+
+`capture-status` 分开报告 active pending、held 历史、每来源 acquisition hold，
+并明确 `semantic_review_completion=not_inferred_from_capture`。`last_checked_at`
+是尝试时间，不保证成功；服务存活和最近成功扫描时间仍由 host 独立报告。
+仅看 status 不能证明历史可重放或决策覆盖完整。停用仍使用原 profile 开关；
+降级旧版本前必须停止调度器，因为旧运行时不认识 recovery hold。
+保留 spool 与回执，不能把软件降级当成状态回滚。
+
 ## 与其他能力的关系
 
 | 能力 | 核心问题 | 与 Decision Context 的关系 |

loopx/capabilities/decision_context/capture.py

@@ -28,6 +28,14 @@
 from .sources import DecisionSourceProvider, DecisionSourceSpec
 
 
+class CaptureReplayError(ValueError):
+    """Typed recovery diagnosis; never classify provider exception prose."""
+
+    def __init__(self, reason: str, message: str):
+        self.reason = reason
+        super().__init__(message)
+
+
 def _open_spool(path: Path, *, goal_id: str, agent_id: str) -> sqlite3.Connection:
     path.parent.mkdir(parents=True, exist_ok=True)
     descriptor = os.open(
@@ -74,6 +82,10 @@ def _binding_digest(profile: DecisionContextProfile, source: DecisionSourceSpec)
 
 
 def _status(db: sqlite3.Connection, source_ids: tuple[str, ...]) -> dict[str, Any]:
+    has_recovery = (
+        db.execute("SELECT 1 FROM sqlite_master WHERE name='capture_holds'").fetchone()
+        is not None
+    )
     rows = []
     for source_id in source_ids:
         source = db.execute(
@@ -89,12 +101,30 @@ def _status(db: sqlite3.Connection, source_ids: tuple[str, ...]) -> dict[str, An
                 "status": source["status"] if source else "never_checked",
                 "pending_batch_count": pending[0],
                 "next_batch_id": pending[1],
+                "held_batch_count": db.execute(
+                    "SELECT count(*) FROM held_batches WHERE source_id=?", (source_id,)
+                ).fetchone()[0]
+                if has_recovery
+                else 0,
+                "acquisition_held": bool(
+                    db.execute(
+                        "SELECT 1 FROM capture_holds WHERE source_id=?", (source_id,)
+                    ).fetchone()
+                )
+                if has_recovery
+                else False,
             }
         )
     return {
         "schema_version": "decision_context_capture_status_v0",
         "sources": rows,
         "pending_batch_count": db.execute("SELECT count(*) FROM batches").fetchone()[0],
+        "held_batch_count": db.execute("SELECT count(*) FROM held_batches").fetchone()[
+            0
+        ]
+        if has_recovery
+        else 0,
+        "semantic_review_completion": "not_inferred_from_capture",
         "raw_content_captured": False,
         "decision_cursors_mutated": False,
         "external_writes_performed": False,
@@ -180,6 +210,15 @@ def capture_profile_sources(
     try:
         reviewed = load_private_decision_cursors(cursor_path, profile=profile)
         for source in sources:
+            if (
+                db.execute(
+                    "SELECT 1 FROM sqlite_master WHERE name='capture_holds'"
+                ).fetchone()
+                and db.execute(
+                    "SELECT 1 FROM capture_holds WHERE source_id=?", (source.source_id,)
+                ).fetchone()
+            ):
+                continue
             binding = _binding_digest(profile, source)
             row = db.execute(
                 "SELECT * FROM sources WHERE source_id=?", (source.source_id,)
@@ -217,6 +256,7 @@ def capture_profile_sources(
             if (
                 row
                 and row["checked_at"]
+                and row["status"] != "backpressure"
                 and (now - datetime.fromisoformat(row["checked_at"])).total_seconds()
                 < profile.capture_interval_seconds
             ):
@@ -333,10 +373,12 @@ def assemble_captured_decision_evidence(
         None,
     )
     if source is None or binding != _binding_digest(profile, source):
-        raise ValueError("capture source binding changed")
+        raise CaptureReplayError("binding_changed", "capture source binding changed")
     reviewed = load_private_decision_cursors(cursor_path, profile=profile)
     if reviewed.get(source.source_id) != batch["cursor_before"]:
-        raise ValueError("capture batch must follow the reviewed cursor")
+        raise CaptureReplayError(
+            "cursor_diverged", "capture batch must follow the reviewed cursor"
+        )
     expected = json.loads(batch["receipt"])
 
     def checked_rebase(collection: DecisionEvidenceCollection):
@@ -353,8 +395,9 @@ def changes(value):
             or receipt["status"] != expected["status"]
             or receipt["exact_read_count"] != expected["changed_count"]
         ):
-            raise ValueError(
-                "captured revision unavailable; explicit source rebase required"
+            raise CaptureReplayError(
+                "revision_unavailable",
+                "captured revision unavailable; explicit source rebase required",
             )
         return rebase(collection)