diff --git a/.doc/ops/short-term/2026-09-22-governance-automation.md b/.doc/ops/short-term/2026-09-22-governance-automation.md index 06731f00..00f66561 100644 --- a/.doc/ops/short-term/2026-09-22-governance-automation.md +++ b/.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。 diff --git a/.doc/ops/short-term/2026-09-23-modular-decouple-residual.md b/.doc/ops/short-term/2026-09-23-modular-decouple-residual.md index 1cabd031..2ad5b135 100644 --- a/.doc/ops/short-term/2026-09-23-modular-decouple-residual.md +++ b/.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 | | ## 每批验收 diff --git a/scripts/agentctl_w4.py b/scripts/agentctl_w4.py index 9607a59b..3e4cb67a 100644 --- a/scripts/agentctl_w4.py +++ b/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 ( diff --git a/scripts/tests/test_agentctl_w4.py b/scripts/tests/test_agentctl_w4.py new file mode 100644 index 00000000..5e180a1f --- /dev/null +++ b/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()