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.
This commit is contained in:
@@ -3,7 +3,7 @@ title: "W4 治理自动化剩余:持久队列与闭环"
|
|||||||
task_id: governance-automation-2026-09-22
|
task_id: governance-automation-2026-09-22
|
||||||
status: in_progress
|
status: in_progress
|
||||||
created: 2026-09-22
|
created: 2026-09-22
|
||||||
updated: 2026-09-22
|
updated: 2026-09-23
|
||||||
owner: Cursor
|
owner: Cursor
|
||||||
task_type: implementation
|
task_type: implementation
|
||||||
task_tier: multi_stage
|
task_tier: multi_stage
|
||||||
@@ -26,6 +26,10 @@ remote_sync: not_requested
|
|||||||
|
|
||||||
# W4 治理自动化剩余:持久队列与闭环
|
# 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 源码镜像交付、清理可见性。
|
- 用户目标:推进 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 增量)
|
1. [x] Durable queue schema + enqueue/list/cancel/tick(`scripts/agentctl_w4.py` / automation 增量)
|
||||||
2. [~] Generic doc projection(registry 骨架;完整投影仍待)
|
2. [x] Generic doc projection(CAS row/@document apply helpers + 黑盒测;完整 docs:admin 投影仍待)
|
||||||
3. [~] Deploy/recover 状态机 + 收据(表与 helper 落地;真实 env 闭环未跑)
|
3. [x] Deploy/recover 状态机 + 收据(enqueue→claim→command→receipt 闭环 + 黑盒测;真实共享 env 未跑)
|
||||||
4. [ ] PI 源码镜像交付 compose profile + 检查入口(deferred)
|
4. [ ] PI 源码镜像交付 compose profile + 检查入口(**explicitly deferred**;超出 temp-repo 测试范围)
|
||||||
5. [~] Cleanup visibility report(fail-closed 文案;Darwin 需 agentctl-macos 进程层)
|
5. [~] Cleanup visibility report(fail-closed 文案;Darwin 需 agentctl-macos 进程层)
|
||||||
6. [x] short-term 写明 landed vs deferred;测试夹具复制 `agentctl_w4.py`
|
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 进程层) |
|
| W4 投影/部署黑盒 | `python3 -B -m unittest scripts.tests.test_agentctl_w4 -v` | 5 PASS(CAS 行插入/整文拒绝、deploy 成败、recover) |
|
||||||
| 夹具回归 | 临时仓缺 `agentctl_w4.py` → `ModuleNotFoundError` | 已修:tests 复制 sibling 模块 |
|
| 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 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_w4.py`:jobs / deployments / projections schema + enqueue/list/cancel/tick helpers
|
||||||
- `scripts/agentctl_automation.py`:并行自动化增量(待与 w4 收敛)
|
- projection apply helpers:`build_row_projection_change` / `build_document_projection_change` / `apply_projection_locally`(CAS)
|
||||||
- `scripts/agentctl.py`:接入 w4 schema/`ensure_schema`
|
- deploy/recover loop:`run_deploy_command` / `execute_deploy_recover_loop` / `recover_deploy`(真实子进程 argv + SQLite 收据)
|
||||||
- 测试夹具复制 `agentctl_w4.py`,避免黑盒仓 import 崩
|
- `scripts/tests/test_agentctl_w4.py`:5 项黑盒测(tempfile + sqlite3 + Path 投影)
|
||||||
|
- 既有测试夹具复制 `agentctl_w4.py`,避免黑盒仓 import 崩
|
||||||
|
|
||||||
**Deferred**
|
**Deferred**
|
||||||
|
|
||||||
|
- **PI source-image compose profile**(显式延期;超出 temp-repo 黑盒范围,不在本续作交付)
|
||||||
- Rebase/merge 本地 `dev` 已合入的 `agentctl-macos` 进程层后,Darwin 上 `agentctl with`/`boot()` 才能非崩溃跑全套脚本测
|
- 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` 重复实现
|
- 清理 `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。
|
- `integration_state=pending`;不 push。
|
||||||
|
|||||||
@@ -36,16 +36,16 @@ remote_sync: not_requested
|
|||||||
|
|
||||||
| 批 | 目标 | 状态 | tip |
|
| 批 | 目标 | 状态 | tip |
|
||||||
|---|---|---|---|
|
|---|---|---|---|
|
||||||
| A | `bizdev/service.go` 拆分 | pending | |
|
| A | `bizdev/service.go` 拆分 | done | dca33c70 |
|
||||||
| B | `projectinstance/store.go` | pending | |
|
| B | `projectinstance/store.go` | done | 943becfb |
|
||||||
| C | `intent/recognizer.go` | pending | |
|
| C | `intent/recognizer.go` | done | dca33c70 |
|
||||||
| D | `server/handlers.go` | pending | |
|
| D | `server/handlers.go` | done | 4c077169 |
|
||||||
| E | `projectworkflow/store.go` | pending | |
|
| E | `projectworkflow/store.go` | done | dca33c70 |
|
||||||
| F | `docgov/engine.go` | pending | |
|
| F | `docgov/engine.go` | done | 39e115b4 |
|
||||||
| G | `docrevision/service.go` | pending | |
|
| G | `docrevision/service.go` | done | 9da7f352 |
|
||||||
| H | `squad/workspace.go` | pending | |
|
| H | `squad/workspace.go` | done | dca33c70 |
|
||||||
| I | `freedoc/service.go` | pending | |
|
| I | `freedoc/service.go` | done | dca33c70 |
|
||||||
| W4 | projection + deploy/recover + Darwin tests | pending | |
|
| W4 | projection + deploy/recover + Darwin tests | in_progress | |
|
||||||
| Sysadmin | `system:*` + 设置审计 | pending | |
|
| Sysadmin | `system:*` + 设置审计 | pending | |
|
||||||
|
|
||||||
## 每批验收
|
## 每批验收
|
||||||
|
|||||||
+302
-4
@@ -7,10 +7,13 @@ rewriting queue semantics.
|
|||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import hashlib
|
||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
import re
|
import re
|
||||||
import sqlite3
|
import sqlite3
|
||||||
|
import subprocess
|
||||||
|
import sys
|
||||||
import time
|
import time
|
||||||
import uuid
|
import uuid
|
||||||
from pathlib import Path
|
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()))
|
(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,
|
def deployment_create(db, *, deploy_id: str, task: str, env: str, expected_commit: str,
|
||||||
manifest: Optional[str], dry_run: bool) -> dict:
|
manifest: Optional[str], dry_run: bool) -> dict:
|
||||||
if env not in ('dev', 'test', 'main'):
|
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(
|
def cleanup_visibility_report(
|
||||||
*,
|
*,
|
||||||
path: Path,
|
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:
|
def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
|
||||||
"""Fail-closed cwd scan using /proc. macOS should use lsof via sibling worktree."""
|
"""Fail-closed cwd scan: /proc on Linux, lsof on Darwin."""
|
||||||
import sys
|
if sys.platform == 'darwin':
|
||||||
|
return _darwin_cwd_visibility(path, uid=uid)
|
||||||
if not sys.platform.startswith('linux'):
|
if not sys.platform.startswith('linux'):
|
||||||
return {
|
return {
|
||||||
'status': 'unsupported',
|
'status': 'unsupported',
|
||||||
'message': f'/proc cwd scan unavailable on {sys.platform}',
|
'message': f'cwd scan unavailable on {sys.platform}',
|
||||||
'platform_note': 'Use agentctl-macos lsof enumerator; treat unknown as cleanup_pending',
|
'platform_note': 'treat unknown as cleanup_pending',
|
||||||
'pids': [],
|
'pids': [],
|
||||||
'unreadable': [],
|
'unreadable': [],
|
||||||
}
|
}
|
||||||
@@ -327,6 +620,7 @@ def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
|
|||||||
'pids': [],
|
'pids': [],
|
||||||
'unreadable': ['*'],
|
'unreadable': ['*'],
|
||||||
}
|
}
|
||||||
|
path = Path(path).resolve()
|
||||||
for proc in entries:
|
for proc in entries:
|
||||||
if not proc.name.isdigit():
|
if not proc.name.isdigit():
|
||||||
continue
|
continue
|
||||||
@@ -353,6 +647,10 @@ def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
|
|||||||
return {'status': 'clear', 'pids': [], 'unreadable': []}
|
return {'status': 'clear', 'pids': [], 'unreadable': []}
|
||||||
|
|
||||||
|
|
||||||
|
# Back-compat alias used by cleanup reports.
|
||||||
|
cwd_visibility = linux_cwd_visibility
|
||||||
|
|
||||||
|
|
||||||
def pi_source_bind_paths() -> tuple:
|
def pi_source_bind_paths() -> tuple:
|
||||||
"""Host paths that must not be bind-mounted when PI image delivery is enforced."""
|
"""Host paths that must not be bind-mounted when PI image delivery is enforced."""
|
||||||
return (
|
return (
|
||||||
|
|||||||
@@ -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()
|
||||||
Reference in New Issue
Block a user