feat(agentctl): W4 projection apply and deploy-recover loop

Add CAS projection helpers and a real enqueue→claim→command→receipt
deploy path for temp-repo tests; PI image compose stays deferred.

.doc/ops/short-term/2026-09-22-governance-automation.md

@@ -3,7 +3,7 @@ title: "W4 治理自动化剩余：持久队列与闭环"
 task_id: governance-automation-2026-09-22
 status: in_progress
 created: 2026-09-22
-updated: 2026-09-22
+updated: 2026-09-23
 owner: Cursor
 task_type: implementation
 task_tier: multi_stage
@@ -26,6 +26,10 @@ remote_sync: not_requested
 
 # W4 治理自动化剩余：持久队列与闭环
 
+> **续作说明（2026-09-23）**：本任务旧 worktree 基线不再续写；剩余 W4 工作已并入
+> [modular-decouple-residual-2026-09-23](./2026-09-23-modular-decouple-residual.md)
+>（`.worktrees/modular-decouple-residual-2026-09-23`）。下文 landed/deferred 以该续作收口为准。
+
 ## 用户目标与授权
 
 - 用户目标：推进 agent-workflow-governance 的 W1–W4 遗留——后台持久队列与自动调度、通用文档投影、业务任务部署/恢复闭环、产品 PI 源码镜像交付、清理可见性。
@@ -35,9 +39,9 @@ remote_sync: not_requested
 ## 步骤与验收
 
 1. [x] Durable queue schema + enqueue/list/cancel/tick（`scripts/agentctl_w4.py` / automation 增量）
-2. [~] Generic doc projection（registry 骨架；完整投影仍待）
-3. [~] Deploy/recover 状态机 + 收据（表与 helper 落地；真实 env 闭环未跑）
-4. [ ] PI 源码镜像交付 compose profile + 检查入口（deferred）
+2. [x] Generic doc projection（CAS row/@document apply helpers + 黑盒测；完整 docs:admin 投影仍待）
+3. [x] Deploy/recover 状态机 + 收据（enqueue→claim→command→receipt 闭环 + 黑盒测；真实共享 env 未跑）
+4. [ ] PI 源码镜像交付 compose profile + 检查入口（**explicitly deferred**；超出 temp-repo 测试范围）
 5. [~] Cleanup visibility report（fail-closed 文案；Darwin 需 agentctl-macos 进程层）
 6. [x] short-term 写明 landed vs deferred；测试夹具复制 `agentctl_w4.py`
 
@@ -45,25 +49,29 @@ remote_sync: not_requested
 
 | 范围 | 命令 | 实际 |
 |---|---|---|
-| ops_guards 抽样 | `python3 -B -m unittest scripts.tests.test_ops_guards -v` | 14 跑；12 PASS；2 FAIL（`Lease`/`boot` 在 Darwin 仍依赖 `/proc`，本分支未含已合入的 agentctl-macos 进程层） |
-| 夹具回归 | 临时仓缺 `agentctl_w4.py` → `ModuleNotFoundError` | 已修：tests 复制 sibling 模块 |
+| W4 投影/部署黑盒 | `python3 -B -m unittest scripts.tests.test_agentctl_w4 -v` | 5 PASS（CAS 行插入/整文拒绝、deploy 成败、recover） |
+| agentctl 冒烟 | `test_uninitialized_status…` / `test_init_preserves…` | 导入 `agentctl_w4` 夹具路径未破 |
+| ops_guards 抽样（旧基线） | `python3 -B -m unittest scripts.tests.test_ops_guards -v` | 14 跑；12 PASS；2 FAIL（`Lease`/`boot` 在 Darwin 仍依赖 `/proc`，本分支未含已合入的 agentctl-macos 进程层） |
 
 ## Landed vs deferred
 
-**Landed（本分支 tip）**
+**Landed（续作 modular-decouple-residual-2026-09-23）**
 
 - `scripts/agentctl_w4.py`：jobs / deployments / projections schema + enqueue/list/cancel/tick helpers
-- `scripts/agentctl_automation.py`：并行自动化增量（待与 w4 收敛）
-- `scripts/agentctl.py`：接入 w4 schema/`ensure_schema`
-- 测试夹具复制 `agentctl_w4.py`，避免黑盒仓 import 崩
+- projection apply helpers：`build_row_projection_change` / `build_document_projection_change` / `apply_projection_locally`（CAS）
+- deploy/recover loop：`run_deploy_command` / `execute_deploy_recover_loop` / `recover_deploy`（真实子进程 argv + SQLite 收据）
+- `scripts/tests/test_agentctl_w4.py`：5 项黑盒测（tempfile + sqlite3 + Path 投影）
+- 既有测试夹具复制 `agentctl_w4.py`，避免黑盒仓 import 崩
 
 **Deferred**
 
+- **PI source-image compose profile**（显式延期；超出 temp-repo 黑盒范围，不在本续作交付）
 - Rebase/merge 本地 `dev` 已合入的 `agentctl-macos` 进程层后，Darwin 上 `agentctl with`/`boot()` 才能非崩溃跑全套脚本测
-- 完整 docs projection apply、真实 deploy/recover 收据、PI 镜像 compose profile
+- 完整 docs:admin 生产路径投影、真实共享 env 部署验收
 - 清理 `agentctl_automation.py` 与 `agentctl_w4.py` 重复实现
 
 ## 收口
 
-- 状态保持 `in_progress`（禁止宣称 W3/W4 全关）。
+- 状态保持 `in_progress`（禁止宣称 W3/W4 全关；PI compose 仍 deferred）。
+- 续作进度与 tip 见 [modular-decouple-residual](./2026-09-23-modular-decouple-residual.md)。
 - `integration_state=pending`；不 push。

.doc/ops/short-term/2026-09-23-modular-decouple-residual.md

@@ -36,16 +36,16 @@ remote_sync: not_requested
 
 | 批 | 目标 | 状态 | tip |
 |---|---|---|---|
-| A | `bizdev/service.go` 拆分 | pending | |
-| B | `projectinstance/store.go` | pending | |
-| C | `intent/recognizer.go` | pending | |
-| D | `server/handlers.go` | pending | |
-| E | `projectworkflow/store.go` | pending | |
-| F | `docgov/engine.go` | pending | |
-| G | `docrevision/service.go` | pending | |
-| H | `squad/workspace.go` | pending | |
-| I | `freedoc/service.go` | pending | |
-| W4 | projection + deploy/recover + Darwin tests | pending | |
+| A | `bizdev/service.go` 拆分 | done | dca33c70 |
+| B | `projectinstance/store.go` | done | 943becfb |
+| C | `intent/recognizer.go` | done | dca33c70 |
+| D | `server/handlers.go` | done | 4c077169 |
+| E | `projectworkflow/store.go` | done | dca33c70 |
+| F | `docgov/engine.go` | done | 39e115b4 |
+| G | `docrevision/service.go` | done | 9da7f352 |
+| H | `squad/workspace.go` | done | dca33c70 |
+| I | `freedoc/service.go` | done | dca33c70 |
+| W4 | projection + deploy/recover + Darwin tests | in_progress | |
 | Sysadmin | `system:*` + 设置审计 | pending | |
 
 ## 每批验收

scripts/agentctl_w4.py

@@ -7,10 +7,13 @@ rewriting queue semantics.
 """
 from __future__ import annotations
 
+import hashlib
 import json
 import os
 import re
 import sqlite3
+import subprocess
+import sys
 import time
 import uuid
 from pathlib import Path
@@ -198,6 +201,92 @@ def record_projection(db, *, file: str, record_id: str, task: str, source: str,
         (file, record_id, task, source, content_hash, _now()))
 
 
+def content_digest(text: str) -> str:
+    return hashlib.sha256(text.encode()).hexdigest()
+
+
+def build_row_projection_change(*, file: str, record_id: str, current_row: Optional[str],
+                                replacement: str) -> dict:
+    """Docs-apply compatible single-row change (CAS on the existing row text)."""
+    if not isinstance(replacement, str) or len(replacement.splitlines()) != 1 \
+            or not replacement.startswith('|'):
+        raise ValueError('replacement must be one Markdown table row')
+    return {
+        'file': file,
+        'record_id': record_id,
+        'expected_sha256': content_digest(current_row) if current_row else None,
+        'replacement': replacement,
+    }
+
+
+def build_document_projection_change(*, file: str, current_text: str,
+                                     replacement: str) -> dict:
+    """Docs-apply compatible whole-document change (CAS on full file hash)."""
+    if not isinstance(replacement, str):
+        raise ValueError('replacement must be full document text')
+    return {
+        'file': file,
+        'record_id': '@document',
+        'expected_sha256': content_digest(current_text),
+        'replacement': replacement,
+    }
+
+
+def apply_projection_locally(path: Path, change: dict) -> dict:
+    """Controlled apply mirroring docs-apply CAS for temp-repo tests (no lease).
+
+    Supports row edits and @document whole-file replacement. Refuses on CAS miss.
+    """
+    if not isinstance(change, dict):
+        raise ValueError('change must be an object')
+    file_rel = change.get('file')
+    record_id = change.get('record_id')
+    replacement = change.get('replacement')
+    expected = change.get('expected_sha256')
+    if not isinstance(file_rel, str) or not file_rel:
+        raise ValueError('file required')
+    if not projectable_file(file_rel):
+        raise ValueError(f'file not projectable: {file_rel}')
+    target = (path / file_rel).resolve()
+    if path.resolve() not in target.parents and target != path.resolve():
+        raise ValueError('path escapes repo root')
+    before = target.read_text(encoding='utf-8')
+    if record_id == '@document':
+        if content_digest(before) != expected:
+            raise RuntimeError('BASE_CHANGED: document hash mismatch')
+        if not isinstance(replacement, str):
+            raise ValueError('Expected full document text')
+        target.write_text(replacement, encoding='utf-8')
+        after = replacement
+    else:
+        lines = before.splitlines()
+        matches = [i for i, line in enumerate(lines)
+                   if record_id and record_id in [c.strip() for c in line.split('|') if c.strip()]]
+        current_row = lines[matches[0]] if matches else None
+        live_hash = content_digest(current_row) if current_row else None
+        if expected != live_hash:
+            raise RuntimeError('BASE_CHANGED: record hash mismatch')
+        if not isinstance(replacement, str) or len(replacement.splitlines()) != 1:
+            raise ValueError('Replacement must be one Markdown table row')
+        if matches:
+            lines[matches[0]] = replacement
+        else:
+            # Insert after first markdown table separator when adding a new row.
+            sep = next((i for i, line in enumerate(lines) if re.match(r'^\|\s*-+', line)), None)
+            if sep is None:
+                raise ValueError('Target table header/separator absent')
+            lines.insert(sep + 1, replacement)
+        after = '\n'.join(lines) + ('\n' if before.endswith('\n') or before == '' else '')
+        target.write_text(after, encoding='utf-8')
+    return {
+        'file': file_rel,
+        'record_id': record_id,
+        'before_sha256': content_digest(before),
+        'after_sha256': content_digest(after),
+        'status': 'applied',
+    }
+
+
 def deployment_create(db, *, deploy_id: str, task: str, env: str, expected_commit: str,
                       manifest: Optional[str], dry_run: bool) -> dict:
     if env not in ('dev', 'test', 'main'):
@@ -252,6 +341,145 @@ def dry_run_deploy_steps(deploy: dict) -> dict:
     }
 
 
+def run_deploy_command(command: list, *, cwd: Optional[str] = None,
+                       timeout: float = 30.0) -> dict:
+    """Execute a real subprocess for deploy_run; used with a fake command in tests."""
+    if not command or not isinstance(command, list) or not all(isinstance(c, str) for c in command):
+        raise ValueError('command must be a nonempty argv list')
+    try:
+        result = subprocess.run(
+            command, cwd=cwd, capture_output=True, text=True, timeout=timeout)
+    except subprocess.TimeoutExpired as exc:
+        return {
+            'ok': False,
+            'exit_code': None,
+            'stdout_tail': (exc.stdout or '')[-2000:],
+            'stderr_tail': (exc.stderr or '')[-2000:],
+            'error': 'timeout',
+            'command': command,
+        }
+    except OSError as exc:
+        return {
+            'ok': False,
+            'exit_code': None,
+            'stdout_tail': '',
+            'stderr_tail': str(exc),
+            'error': 'spawn_failed',
+            'command': command,
+        }
+    return {
+        'ok': result.returncode == 0,
+        'exit_code': result.returncode,
+        'stdout_tail': (result.stdout or '')[-2000:],
+        'stderr_tail': (result.stderr or '')[-2000:],
+        'error': None if result.returncode == 0 else 'nonzero_exit',
+        'command': command,
+    }
+
+
+def execute_deploy_recover_loop(
+    db,
+    *,
+    deploy_id: str,
+    task: str,
+    env: str,
+    expected_commit: str,
+    command: list,
+    dry_run: bool = False,
+    manifest: Optional[str] = None,
+    worker: str = 'w4-deploy-worker',
+    cwd: Optional[str] = None,
+) -> dict:
+    """Real enqueue→claim→deploy(command)→receipt chain; receipts persist in SQLite.
+
+    Does not touch shared docker. Pass a no-op/fake argv for black-box tests.
+    """
+    prepare = queue_enqueue(
+        db, task=task, kind='deploy_prepare',
+        payload={
+            'deploy_id': deploy_id, 'env': env,
+            'expected_commit': expected_commit, 'dry_run': dry_run,
+        },
+        job_id=f'{deploy_id}-prepare')
+    claimed = claim_next_job(db, worker=worker)
+    if not claimed or claimed['id'] != prepare['id']:
+        raise RuntimeError('failed to claim deploy_prepare job')
+    deploy = deployment_create(
+        db, deploy_id=deploy_id, task=task, env=env,
+        expected_commit=expected_commit, manifest=manifest, dry_run=dry_run)
+    finish_job(db, prepare['id'], ok=True, result={'phase': 'prepared', 'deploy': deploy})
+
+    run_job = queue_enqueue(
+        db, task=task, kind='deploy_run',
+        payload={'deploy_id': deploy_id, 'command': command, 'dry_run': dry_run},
+        job_id=f'{deploy_id}-run')
+    claimed = claim_next_job(db, worker=worker)
+    if not claimed or claimed['id'] != run_job['id']:
+        raise RuntimeError('failed to claim deploy_run job')
+    deployment_set_phase(db, deploy_id, 'deploying')
+    if dry_run:
+        receipt = dry_run_deploy_steps(deploy)
+        ok = True
+    else:
+        receipt = run_deploy_command(command, cwd=cwd)
+        ok = bool(receipt.get('ok'))
+    if not ok:
+        deploy = deployment_set_phase(
+            db, deploy_id, 'failed', receipt=receipt,
+            error=receipt.get('error') or 'deploy_command_failed')
+        finish_job(db, run_job['id'], ok=False, result=receipt,
+                   error=receipt.get('error') or 'deploy_command_failed')
+        # recovery job leaves phase recovery_required for operator recover
+        recover_job = queue_enqueue(
+            db, task=task, kind='deploy_accept',
+            payload={'deploy_id': deploy_id, 'action': 'recover'},
+            job_id=f'{deploy_id}-recover-pending')
+        deployment_set_phase(db, deploy_id, 'recovery_required', receipt=receipt,
+                             error=receipt.get('error') or 'deploy_command_failed')
+        return {
+            'status': 'recovery_required',
+            'deploy': deployment_get(db, deploy_id),
+            'jobs': {'prepare': prepare['id'], 'run': run_job['id'],
+                     'recover_pending': recover_job['id']},
+            'receipt': receipt,
+        }
+    deployment_set_phase(db, deploy_id, 'accepting', receipt=receipt)
+    finish_job(db, run_job['id'], ok=True, result=receipt)
+
+    accept_job = queue_enqueue(
+        db, task=task, kind='deploy_accept',
+        payload={'deploy_id': deploy_id, 'action': 'accept'},
+        job_id=f'{deploy_id}-accept')
+    claimed = claim_next_job(db, worker=worker)
+    if not claimed or claimed['id'] != accept_job['id']:
+        raise RuntimeError('failed to claim deploy_accept job')
+    final_receipt = dict(receipt)
+    final_receipt['accepted'] = True
+    final_receipt['accepted_at'] = _now()
+    deploy = deployment_set_phase(db, deploy_id, 'passed', receipt=final_receipt)
+    finish_job(db, accept_job['id'], ok=True, result=final_receipt)
+    return {
+        'status': 'passed',
+        'deploy': deploy,
+        'jobs': {'prepare': prepare['id'], 'run': run_job['id'], 'accept': accept_job['id']},
+        'receipt': final_receipt,
+    }
+
+
+def recover_deploy(db, deploy_id: str, *, reason: str) -> dict:
+    """Mark a failed/recovery_required deploy as recovered without re-running env writes."""
+    row = deployment_get(db, deploy_id)
+    if not row:
+        raise ValueError('unknown deploy id')
+    if row['phase'] not in ('failed', 'recovery_required'):
+        raise ValueError(f'cannot recover deploy in phase {row["phase"]}')
+    receipt = json.loads(row['receipt']) if row.get('receipt') else {}
+    receipt['recovered'] = True
+    receipt['recover_reason'] = reason
+    receipt['recovered_at'] = _now()
+    return deployment_set_phase(db, deploy_id, 'passed', receipt=receipt, error=None)
+
+
 def cleanup_visibility_report(
     *,
     path: Path,
@@ -304,14 +532,79 @@ def cleanup_visibility_report(
     }
 
 
+def _darwin_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
+    """Fail-closed cwd scan via lsof on Darwin (agentctl-macos process layer)."""
+    uid = os.getuid() if uid is None else uid
+    try:
+        result = subprocess.run(
+            ['lsof', '-a', '-d', 'cwd', '-u', str(uid), '-F', 'pn'],
+            capture_output=True, text=True)
+    except OSError as exc:
+        return {
+            'status': 'unreadable',
+            'message': f'cannot enumerate process cwd via lsof ({exc})',
+            'platform_note': 'fail-closed',
+            'pids': [],
+            'unreadable': ['*'],
+        }
+    # lsof exits 1 when selection empty; other nonzero with empty stdout is opaque.
+    if result.returncode not in (0, 1):
+        return {
+            'status': 'unreadable',
+            'message': f'lsof exit {result.returncode}',
+            'platform_note': 'fail-closed',
+            'pids': [],
+            'unreadable': ['*'],
+        }
+    if result.stderr.strip() and result.returncode != 0 and not result.stdout.strip():
+        return {
+            'status': 'unreadable',
+            'message': 'cannot verify process cwd via lsof',
+            'platform_note': 'fail-closed',
+            'pids': [],
+            'unreadable': ['*'],
+        }
+    path = Path(path).resolve()
+    busy, unreadable = [], []
+    pid, saw_cwd = None, False
+    for line in result.stdout.splitlines():
+        if not line:
+            continue
+        code, value = line[0], line[1:]
+        if code == 'p':
+            if pid is not None and not saw_cwd:
+                unreadable.append(pid)
+            pid, saw_cwd = value, False
+        elif code == 'n' and pid is not None:
+            try:
+                cwd = Path(value).resolve()
+            except OSError:
+                unreadable.append(pid)
+                saw_cwd = True
+                continue
+            if cwd == path or path in cwd.parents:
+                busy.append(pid)
+            saw_cwd = True
+    if pid is not None and not saw_cwd:
+        unreadable.append(pid)
+    if busy:
+        return {'status': 'busy', 'pids': busy, 'unreadable': unreadable,
+                'platform_note': 'darwin-lsof'}
+    if unreadable:
+        return {'status': 'unreadable', 'pids': [], 'unreadable': unreadable,
+                'platform_note': 'fail-closed; do not treat as exited'}
+    return {'status': 'clear', 'pids': [], 'unreadable': [], 'platform_note': 'darwin-lsof'}
+
+
 def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
-    """Fail-closed cwd scan using /proc. macOS should use lsof via sibling worktree."""
-    import sys
+    """Fail-closed cwd scan: /proc on Linux, lsof on Darwin."""
+    if sys.platform == 'darwin':
+        return _darwin_cwd_visibility(path, uid=uid)
     if not sys.platform.startswith('linux'):
         return {
             'status': 'unsupported',
-            'message': f'/proc cwd scan unavailable on {sys.platform}',
-            'platform_note': 'Use agentctl-macos lsof enumerator; treat unknown as cleanup_pending',
+            'message': f'cwd scan unavailable on {sys.platform}',
+            'platform_note': 'treat unknown as cleanup_pending',
             'pids': [],
             'unreadable': [],
         }
@@ -327,6 +620,7 @@ def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
             'pids': [],
             'unreadable': ['*'],
         }
+    path = Path(path).resolve()
     for proc in entries:
         if not proc.name.isdigit():
             continue
@@ -353,6 +647,10 @@ def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
     return {'status': 'clear', 'pids': [], 'unreadable': []}
 
 
+# Back-compat alias used by cleanup reports.
+cwd_visibility = linux_cwd_visibility
+
+
 def pi_source_bind_paths() -> tuple:
     """Host paths that must not be bind-mounted when PI image delivery is enforced."""
     return (

scripts/tests/test_agentctl_w4.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+"""Black-box tests for agentctl_w4 projection apply and deploy/recover loop."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import shutil
+import sqlite3
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+SOURCE = Path(__file__).resolve().parents[1]
+W4_SOURCE = SOURCE / "agentctl_w4.py"
+
+
+def load_w4(path: Path):
+    spec = importlib.util.spec_from_file_location("agentctl_w4_under_test", path)
+    mod = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(mod)
+    return mod
+
+
+class AgentctlW4Tests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="agentctl-w4-")
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name) / "repo"
+        self.root.mkdir()
+        (self.root / "scripts").mkdir()
+        shutil.copy2(W4_SOURCE, self.root / "scripts" / "agentctl_w4.py")
+        self.w4 = load_w4(self.root / "scripts" / "agentctl_w4.py")
+        self.db_path = Path(self.tmp.name) / "w4.sqlite"
+        self.db = sqlite3.connect(self.db_path)
+        self.db.row_factory = sqlite3.Row
+        self.addCleanup(self.db.close)
+        self.w4.ensure_schema(self.db)
+
+    def _write_current(self, body: str) -> Path:
+        path = self.root / ".doc" / "ops" / "CURRENT.md"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(body, encoding="utf-8")
+        return path
+
+    def test_row_projection_cas_and_insert(self):
+        doc = (
+            "# Coordination\n\n"
+            "| Task | Owner | Branch | State |\n"
+            "| --- | --- | --- | --- |\n"
+            "| alpha | Alice | feature/alpha | active |\n"
+        )
+        target = self._write_current(doc)
+        current_row = "| alpha | Alice | feature/alpha | active |"
+        replacement = "| alpha | Alice | feature/alpha | done |"
+        change = self.w4.build_row_projection_change(
+            file=".doc/ops/CURRENT.md",
+            record_id="alpha",
+            current_row=current_row,
+            replacement=replacement,
+        )
+        result = self.w4.apply_projection_locally(self.root, change)
+        self.assertEqual(result["status"], "applied")
+        self.assertIn(replacement, target.read_text(encoding="utf-8"))
+        self.assertNotIn(current_row, target.read_text(encoding="utf-8"))
+
+        insert_row = "| gamma | Carol | feature/gamma | active |"
+        insert = self.w4.build_row_projection_change(
+            file=".doc/ops/CURRENT.md",
+            record_id="gamma",
+            current_row=None,
+            replacement=insert_row,
+        )
+        self.assertIsNone(insert["expected_sha256"])
+        inserted = self.w4.apply_projection_locally(self.root, insert)
+        self.assertEqual(inserted["status"], "applied")
+        text = target.read_text(encoding="utf-8")
+        self.assertIn(insert_row, text)
+        self.assertIn(replacement, text)
+
+    def test_document_projection_cas_mismatch_refuses(self):
+        original = "# Features\n\nplanned\n"
+        path = self.root / ".doc" / "status" / "FEATURE-MATRIX.md"
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(original, encoding="utf-8")
+        change = self.w4.build_document_projection_change(
+            file=".doc/status/FEATURE-MATRIX.md",
+            current_text=original,
+            replacement="# Features\n\ndone\n",
+        )
+        # Mutate on disk so live hash no longer matches the CAS tip.
+        path.write_text("# Features\n\nchanged-underneath\n", encoding="utf-8")
+        with self.assertRaises(RuntimeError) as ctx:
+            self.w4.apply_projection_locally(self.root, change)
+        self.assertIn("BASE_CHANGED", str(ctx.exception))
+        self.assertEqual(path.read_text(encoding="utf-8"), "# Features\n\nchanged-underneath\n")
+
+    def test_deploy_recover_loop_success_with_fake_command(self):
+        outcome = self.w4.execute_deploy_recover_loop(
+            self.db,
+            deploy_id="dep-ok",
+            task="modular-decouple-residual-2026-09-23",
+            env="dev",
+            expected_commit="abcdef1",
+            command=["/bin/echo", "ok"],
+        )
+        self.assertEqual(outcome["status"], "passed")
+        self.assertEqual(outcome["deploy"]["phase"], "passed")
+        self.assertTrue(outcome["receipt"].get("ok"))
+        self.assertTrue(outcome["receipt"].get("accepted"))
+        stored = self.w4.deployment_get(self.db, "dep-ok")
+        self.assertEqual(stored["phase"], "passed")
+        receipt = json.loads(stored["receipt"])
+        self.assertTrue(receipt.get("accepted"))
+
+    def test_deploy_recover_loop_failure_sets_recovery_required(self):
+        outcome = self.w4.execute_deploy_recover_loop(
+            self.db,
+            deploy_id="dep-fail",
+            task="modular-decouple-residual-2026-09-23",
+            env="dev",
+            expected_commit="abcdef2",
+            command=["/bin/false"],
+        )
+        self.assertEqual(outcome["status"], "recovery_required")
+        self.assertEqual(outcome["deploy"]["phase"], "recovery_required")
+        self.assertFalse(outcome["receipt"].get("ok"))
+        stored = self.w4.deployment_get(self.db, "dep-fail")
+        self.assertEqual(stored["phase"], "recovery_required")
+        pending = self.w4.queue_list(self.db, task="modular-decouple-residual-2026-09-23",
+                                     state="pending")
+        self.assertTrue(any(j["id"] == "dep-fail-recover-pending" for j in pending))
+
+    def test_recover_deploy_from_recovery_required(self):
+        self.w4.execute_deploy_recover_loop(
+            self.db,
+            deploy_id="dep-recover",
+            task="modular-decouple-residual-2026-09-23",
+            env="dev",
+            expected_commit="abcdef3",
+            command=[sys.executable, "-c", "raise SystemExit(1)"],
+        )
+        recovered = self.w4.recover_deploy(
+            self.db, "dep-recover", reason="operator cleared after fake failure")
+        self.assertEqual(recovered["phase"], "passed")
+        self.assertIsNone(recovered["last_error"])
+        receipt = json.loads(recovered["receipt"])
+        self.assertTrue(receipt.get("recovered"))
+        self.assertEqual(receipt.get("recover_reason"),
+                         "operator cleared after fake failure")
+
+
+if __name__ == "__main__":
+    unittest.main()