From dd75f42c985a3c4245ccedbf5107a4468ef3920f Mon Sep 17 00:00:00 2001 From: WeiCHEN Date: Wed, 23 Sep 2026 14:39:49 +0800 Subject: [PATCH] fix(agentctl): converge automation into w4 and drop fork Port deploy phase transition guards and cleanup markdown projection into agentctl_w4, delete the unused automation module, and pin illegal jumps with a regression test. --- ...2026-09-23-agentctl-automation-converge.md | 70 +++ scripts/agentctl_automation.py | 413 ------------------ scripts/agentctl_w4.py | 42 ++ scripts/tests/test_agentctl_w4.py | 15 + 4 files changed, 127 insertions(+), 413 deletions(-) create mode 100644 .doc/ops/short-term/2026-09-23-agentctl-automation-converge.md delete mode 100644 scripts/agentctl_automation.py diff --git a/.doc/ops/short-term/2026-09-23-agentctl-automation-converge.md b/.doc/ops/short-term/2026-09-23-agentctl-automation-converge.md new file mode 100644 index 00000000..b55c386b --- /dev/null +++ b/.doc/ops/short-term/2026-09-23-agentctl-automation-converge.md @@ -0,0 +1,70 @@ +--- +title: "W1-U1r：agentctl automation 归一" +task_id: agentctl-automation-converge-2026-09-23 +status: in_progress +created: 2026-09-23 +updated: 2026-09-23 +owner: Cursor +task_type: implementation +task_tier: single +policy_revision: "agent-governance-2026-09-21-r1 @ 21c1278c04566f90230c6e354121041f2c020214" +branch: fix/agentctl-automation-converge-2026-09-23 +worktree: .worktrees/agentctl-automation-converge-2026-09-23 +base_dev_sha: "21c1278c04566f90230c6e354121041f2c020214" +write_scope: ["scripts", ".doc"] +allowed_environments: [] +mutex: agentctl-automation-converge +review_state: pending +integration_state: pending +deployment: not_required +cleanup: pending +cleanup_owner: "独立维护者" +cleanup_trigger: "合入 local_dev 且无进程占用后复查" +remote_sync: not_requested +--- + +# W1-U1r：agentctl automation 归一 + +## 用户目标与授权 + +- 用户目标：删除分叉 `scripts/agentctl_automation.py`，移植部署阶段转移校验与 cleanup Markdown 投影到 `agentctl_w4.py`。 +- 授权：W1 波次实施计划 U1r；write_scope=`scripts`（+ short-term/CURRENT）；不 push。 +- 交付终点：代码合入 local_dev；五态可审。 + +## 移植与丢弃 + +### 已移植 + +- `deployment_set_phase` 合法转移边（`prepared→deploying→accepting→passed`；非法 `passed→deploying` 拒绝）。 +- `project_cleanup_markdown` → `agentctl_w4.py`（供 W2-U5）。 + +### 已丢弃（检索证据） + +| 符号 | 取代方 / 原因 | +|---|---| +| `assess_cleanup_entry` | `agentctl.py` `cleanup_status` 8 gate | +| `process_platform_ready` | W1-B macOS 进程层三值 `alive_status` | +| `build_command_for_job` | 无调用方的 `*-proxy` 占位 | +| automation `jobs` schema | W4 `jobs` 已 landed | +| `operations` / `doc_registry` 表 | 活库 `.tables` 无此二者 | + +`DEFAULT_DOC_REGISTRY` per-doc 规则不在本包移植，交 U2 策略 JSON。 + +## 步骤与验收 + +1. [x] worktree + register + CURRENT 认领 +2. [x] 移植 phase 校验与 `project_cleanup_markdown` +3. [x] 删除 `agentctl_automation.py`；`rg agentctl_automation` 为空；清理 0 字节 `.locks/state.db` +4. [x] 补 `test_deploy_phase_transition_rejects_illegal_jump`；`unittest discover` 104 ran / 2 skipped / OK +5. [ ] 原子提交 + 候选 validate/integrate + +## 验证与证据 + +| 范围 | 命令 | 结果 | +|---|---|---| +| 全套治理脚本 | `python3 -B -m unittest discover -s scripts/tests -p 'test_*.py' -v` | Ran 104 / OK (skipped=2) | + +## 相关链接 + +- [实施承接计划](../../remediation/2026-09-23-coordination-db-daemon-implementation-undertaking-plan.md) +- [W1 实施计划](file:///Users/weichen/.cursor/plans/w1_三包实施_43ec5550.plan.md) diff --git a/scripts/agentctl_automation.py b/scripts/agentctl_automation.py deleted file mode 100644 index 55501ba1..00000000 --- a/scripts/agentctl_automation.py +++ /dev/null @@ -1,413 +0,0 @@ -#!/usr/bin/env python3 -"""W4 governance automation: durable queue, doc projection, deploy ops. - -Process liveness remains in agentctl.py (Linux /proc today; macOS sibling). -This module is intentionally importable for unit tests without supervising children. -""" -from __future__ import annotations - -import json -import os -import re -import time -import uuid -from pathlib import Path - -# Job / operation terminal and active states. -JOB_PENDING = 'pending' -JOB_CLAIMED = 'claimed' -JOB_RUNNING = 'running' -JOB_COMPLETED = 'completed' -JOB_FAILED = 'failed' -JOB_CANCELLED = 'cancelled' -JOB_ACTIVE = (JOB_PENDING, JOB_CLAIMED, JOB_RUNNING) -JOB_TERMINAL = (JOB_COMPLETED, JOB_FAILED, JOB_CANCELLED) - -DEPLOY_PHASES = ( - 'planned', 'deploying', 'accepting', 'passed', 'failed', - 'recovery_required', 'recovered', -) - -# Managed shared docs plus projection-capable documents. -DEFAULT_DOC_REGISTRY = { - '.doc/ops/CURRENT.md': { - 'kind': 'shared_cas', - 'header': '| owner | task |', - 'rules': {'own_row_only': True, 'slot_protected': True}, - }, - '.doc/status/FEATURE-MATRIX.md': { - 'kind': 'shared_cas', - 'header': '| 模块 |', - 'rules': {'evidence_for_done': True}, - }, - '.doc/ops/projections/cleanup-visibility.md': { - 'kind': 'projection', - 'header': '| path | branch | reason | cleanup_owner |', - 'rules': {'generated': True}, - }, -} - -SCHEMA_SQL = ''' -CREATE TABLE IF NOT EXISTS jobs( - generation INTEGER PRIMARY KEY AUTOINCREMENT, - id TEXT UNIQUE NOT NULL, - task TEXT NOT NULL, - kind TEXT NOT NULL, - payload TEXT NOT NULL, - resources TEXT NOT NULL, - state TEXT NOT NULL, - priority INTEGER NOT NULL DEFAULT 100, - created REAL NOT NULL, - claimed_at REAL, - claimed_by TEXT, - run_id TEXT, - finished_at REAL, - result TEXT, - fingerprint TEXT NOT NULL -); -CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, priority, generation); -CREATE TABLE IF NOT EXISTS operations( - id TEXT PRIMARY KEY, - task TEXT NOT NULL, - kind TEXT NOT NULL, - phase TEXT NOT NULL, - env TEXT NOT NULL, - expected_sha TEXT, - runtime_sha TEXT, - receipt TEXT NOT NULL, - created REAL NOT NULL, - updated REAL NOT NULL -); -CREATE INDEX IF NOT EXISTS operations_task ON operations(task, updated); -CREATE TABLE IF NOT EXISTS doc_registry( - path TEXT PRIMARY KEY, - kind TEXT NOT NULL, - header TEXT, - rules TEXT NOT NULL DEFAULT '{}' -); -''' - - -def ensure_schema(db): - """Idempotent schema evolution; keeps protocol VERSION stable.""" - db.executescript(SCHEMA_SQL) - for path, meta in DEFAULT_DOC_REGISTRY.items(): - db.execute( - 'INSERT OR IGNORE INTO doc_registry(path,kind,header,rules) VALUES(?,?,?,?)', - (path, meta['kind'], meta.get('header'), json.dumps(meta.get('rules') or {})), - ) - - -def process_platform_ready(): - """True when the host can supply process start tokens for leases. - - Linux uses /proc. Darwin needs the agentctl-macos process layer; until that - lands, lease-backed dispatch stays deferred rather than TypeError. - """ - if os.environ.get('AGENTCTL_FORCE_PROC') == '1': - return True - if os.environ.get('AGENTCTL_FORCE_PROC') == '0': - return False - return Path('/proc/self/stat').is_file() - - -def job_fingerprint(kind, payload, resources): - blob = json.dumps({'kind': kind, 'payload': payload, 'resources': sorted(resources)}, - sort_keys=True, ensure_ascii=False) - return blob - - -def enqueue_job(db, task_id, kind, payload, resources, job_id=None, priority=100): - if kind not in ('command', 'validate', 'integrate', 'deploy', 'docs_apply', 'cleanup_report'): - raise ValueError(f'unsupported job kind: {kind}') - if not isinstance(payload, dict): - raise ValueError('payload must be an object') - if not isinstance(resources, (list, tuple)) or not resources: - raise ValueError('resources must be a nonempty list') - job_id = job_id or str(uuid.uuid4()) - if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', job_id): - raise ValueError('invalid job id') - fingerprint = __import__('hashlib').sha256( - job_fingerprint(kind, payload, list(resources)).encode()).hexdigest() - existing = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone() - if existing: - if existing['fingerprint'] != fingerprint or existing['task'] != task_id: - raise ValueError('job id reused with different fingerprint') - return dict(existing), True - now = time.time() - db.execute( - '''INSERT INTO jobs(id,task,kind,payload,resources,state,priority,created,fingerprint) - VALUES(?,?,?,?,?,?,?,?,?)''', - (job_id, task_id, kind, json.dumps(payload, ensure_ascii=False, sort_keys=True), - json.dumps(list(resources)), JOB_PENDING, int(priority), now, fingerprint)) - row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone() - return dict(row), False - - -def list_jobs(db, state=None, limit=50): - if state: - rows = db.execute( - 'SELECT * FROM jobs WHERE state=? ORDER BY priority ASC, generation ASC LIMIT ?', - (state, limit)).fetchall() - else: - rows = db.execute( - 'SELECT * FROM jobs ORDER BY CASE state ' - "WHEN 'pending' THEN 0 WHEN 'claimed' THEN 1 WHEN 'running' THEN 2 " - "WHEN 'failed' THEN 3 ELSE 4 END, priority ASC, generation ASC LIMIT ?", - (limit,)).fetchall() - return [dict(r) for r in rows] - - -def cancel_job(db, job_id, reason='cancelled'): - row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone() - if not row: - raise KeyError('unknown job') - if row['state'] in JOB_TERMINAL: - return dict(row), True - if row['state'] == JOB_RUNNING: - raise RuntimeError('cannot cancel a running job; recover the run first') - db.execute( - "UPDATE jobs SET state=?,finished_at=?,result=? WHERE id=?", - (JOB_CANCELLED, time.time(), json.dumps({'reason': reason}), job_id)) - return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()), False - - -def claim_next_job(db, worker_id, kinds=None): - """Atomically claim the highest-priority pending job. Returns None if empty.""" - clauses = ["state=?", JOB_PENDING] - sql = 'SELECT * FROM jobs WHERE state=? ' - args = [JOB_PENDING] - if kinds: - placeholders = ','.join('?' for _ in kinds) - sql += f'AND kind IN ({placeholders}) ' - args.extend(kinds) - sql += 'ORDER BY priority ASC, generation ASC LIMIT 1' - row = db.execute(sql, args).fetchone() - if not row: - return None - db.execute( - "UPDATE jobs SET state=?,claimed_at=?,claimed_by=? WHERE id=? AND state=?", - (JOB_CLAIMED, time.time(), worker_id, row['id'], JOB_PENDING)) - claimed = db.execute('SELECT * FROM jobs WHERE id=?', (row['id'],)).fetchone() - if not claimed or claimed['state'] != JOB_CLAIMED or claimed['claimed_by'] != worker_id: - return None - return dict(claimed) - - -def mark_job(db, job_id, state, result=None, run_id=None): - if state not in (JOB_RUNNING, JOB_COMPLETED, JOB_FAILED, JOB_PENDING, JOB_CLAIMED): - raise ValueError('invalid job state') - finished = time.time() if state in (JOB_COMPLETED, JOB_FAILED) else None - db.execute( - 'UPDATE jobs SET state=?,result=?,run_id=COALESCE(?,run_id),finished_at=COALESCE(?,finished_at) WHERE id=?', - (state, json.dumps(result) if result is not None else None, run_id, finished, job_id)) - return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()) - - -def plan_deploy(db, task_id, env, expected_sha, operation_id=None, extra=None): - if env not in ('dev', 'test', 'main'): - raise ValueError('env must be dev|test|main') - if not re.fullmatch(r'[0-9a-f]{7,40}', expected_sha): - raise ValueError('expected_sha must be a git object name') - operation_id = operation_id or str(uuid.uuid4()) - now = time.time() - receipt = { - 'task': task_id, - 'env': env, - 'expected_sha': expected_sha, - 'events': [{'at': now, 'phase': 'planned'}], - **(extra or {}), - } - existing = db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone() - if existing: - if existing['task'] != task_id or existing['expected_sha'] != expected_sha: - raise ValueError('operation id reused with different identity') - return dict(existing), True - db.execute( - '''INSERT INTO operations(id,task,kind,phase,env,expected_sha,runtime_sha,receipt,created,updated) - VALUES(?,?,?,?,?,?,?,?,?,?)''', - (operation_id, task_id, 'deploy', 'planned', env, expected_sha, None, - json.dumps(receipt, ensure_ascii=False), now, now)) - return dict(db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()), False - - -def advance_deploy(db, operation_id, phase, runtime_sha=None, detail=None): - if phase not in DEPLOY_PHASES: - raise ValueError(f'invalid phase: {phase}') - row = db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone() - if not row: - raise KeyError('unknown operation') - allowed = { - 'planned': {'deploying', 'failed', 'recovery_required'}, - 'deploying': {'accepting', 'failed', 'recovery_required'}, - 'accepting': {'passed', 'failed', 'recovery_required'}, - 'failed': {'recovery_required', 'recovered'}, - 'recovery_required': {'recovered', 'deploying'}, - 'passed': set(), - 'recovered': {'deploying'}, - } - if phase not in allowed.get(row['phase'], set()) and phase != row['phase']: - raise RuntimeError(f'illegal transition {row["phase"]} -> {phase}') - receipt = json.loads(row['receipt']) - now = time.time() - receipt.setdefault('events', []).append({'at': now, 'phase': phase, 'detail': detail}) - if runtime_sha: - receipt['runtime_sha'] = runtime_sha - db.execute( - 'UPDATE operations SET phase=?,runtime_sha=COALESCE(?,runtime_sha),receipt=?,updated=? WHERE id=?', - (phase, runtime_sha, json.dumps(receipt, ensure_ascii=False), now, operation_id)) - return dict(db.execute('SELECT * FROM operations WHERE id=?', (operation_id,)).fetchone()) - - -def list_operations(db, task=None, limit=50): - if task: - rows = db.execute( - 'SELECT * FROM operations WHERE task=? ORDER BY updated DESC LIMIT ?', - (task, limit)).fetchall() - else: - rows = db.execute( - 'SELECT * FROM operations ORDER BY updated DESC LIMIT ?', (limit,)).fetchall() - return [dict(r) for r in rows] - - -def list_doc_registry(db): - ensure_schema(db) - return [dict(r) for r in db.execute('SELECT * FROM doc_registry ORDER BY path').fetchall()] - - -def managed_doc_paths(db): - return {row['path'] for row in list_doc_registry(db)} - - -def project_cleanup_markdown(entries): - """Render cleanup visibility projection from structured entries.""" - lines = [ - '# Cleanup visibility projection', - '', - '> Generated by `agentctl docs project --kind cleanup`. Not a lock; machine `.locks` is authoritative.', - '', - '| path | branch | reason | cleanup_owner |', - '| --- | --- | --- | --- |', - ] - if not entries: - lines.append('| — | — | none_pending | — |') - else: - for item in entries: - path = str(item.get('path') or '').replace('|', '\\|') - branch = str(item.get('branch') or '').replace('|', '\\|') - reason = str(item.get('reason') or '').replace('|', '\\|') - owner = str(item.get('cleanup_owner') or 'independent-maintainer').replace('|', '\\|') - lines.append(f'| {path} | {branch} | {reason} | {owner} |') - lines.append('') - return '\n'.join(lines) - - -def assess_cleanup_entry(main, entry, control_fn=None, process_cwd_fn=None): - """Return a visibility dict; never deletes. Failures become reason strings.""" - from pathlib import Path as P - branch_ref = entry.get('branch') or '' - branch = branch_ref.replace('refs/heads/', '') if branch_ref.startswith('refs/heads/') else '' - path = P(entry.get('worktree') or '') - result = { - 'path': str(path), - 'branch': branch or None, - 'eligible': False, - 'reason': 'unknown', - 'checks': {}, - } - if not branch or not path: - result['reason'] = 'missing_branch_or_path' - return result - if 'locked' in entry: - result['reason'] = 'git_worktree_locked' - return result - if 'prunable' in entry: - result['reason'] = 'git_worktree_prunable' - return result - if not path.is_dir(): - result['reason'] = 'path_missing' - return result - - def git_c(*args): - import subprocess - return subprocess.run(['git', '-C', str(path if args[0] != '--main' else main), *([a for a in args if a != '--main'])], - capture_output=True, text=True) - - # Use explicit git -C for main vs path. - import subprocess - - def run(cwd, *args): - return subprocess.run(['git', '-C', str(cwd), *args], capture_output=True, text=True) - - actual = run(path, 'symbolic-ref', '--quiet', 'HEAD') - if actual.returncode or actual.stdout.strip() != f'refs/heads/{branch}': - result['reason'] = 'branch_identity_mismatch' - result['checks']['head_ref'] = actual.stdout.strip() - return result - head = run(path, 'rev-parse', 'HEAD').stdout.strip() - ancestor = run(main, 'merge-base', '--is-ancestor', head, 'refs/heads/dev') - if ancestor.returncode: - result['reason'] = 'head_not_ancestor_of_dev' - result['checks']['head'] = head - return result - result['checks']['head'] = head - for marker in ('MERGE_HEAD', 'CHERRY_PICK_HEAD', 'REVERT_HEAD', 'rebase-merge', 'rebase-apply', 'BISECT_START'): - marker_path = run(path, 'rev-parse', '--path-format=absolute', '--git-path', marker).stdout.strip() - if marker_path and Path(marker_path).exists(): - result['reason'] = f'unfinished_git_op:{marker}' - return result - status = run(path, 'status', '--porcelain=v1', '--untracked-files=all', '--ignored=matching') - if status.stdout.strip(): - result['reason'] = 'dirty_or_ignored_retained' - result['checks']['status_bytes'] = len(status.stdout) - return result - if control_fn is not None: - try: - control_fn(path) - result['checks']['registry'] = 'ok' - except Exception as exc: # noqa: BLE001 — visibility aggregates any refusal - result['reason'] = f'registry:{exc}' - return result - if process_cwd_fn is not None: - try: - process_cwd_fn(path) - result['checks']['process_cwd'] = 'ok' - except Exception as exc: # noqa: BLE001 - result['reason'] = f'process_cwd:{exc}' - return result - elif not process_platform_ready(): - result['reason'] = 'process_cwd:platform_deferred_macos_or_no_proc' - result['checks']['process_cwd'] = 'deferred' - return result - result['eligible'] = True - result['reason'] = 'eligible' - return result - - -def build_command_for_job(kind, payload): - """Map a durable job to an argv list relative to agentctl semantics.""" - if kind == 'command': - command = payload.get('command') - if not isinstance(command, list) or not command or any(not isinstance(c, str) for c in command): - raise ValueError('command job requires string argv list') - return command - if kind == 'validate': - command = payload.get('command') - if not isinstance(command, list) or not command: - raise ValueError('validate job requires command argv') - return ['validate-proxy', *command] # resolved by dispatcher - if kind == 'integrate': - expected = payload.get('expected_dev') - if not isinstance(expected, str) or not expected: - raise ValueError('integrate job requires expected_dev') - return ['integrate-proxy', expected] - if kind == 'deploy': - return ['deploy-proxy', payload.get('operation_id') or ''] - if kind == 'docs_apply': - request = payload.get('request_path') - if not isinstance(request, str): - raise ValueError('docs_apply requires request_path') - return ['docs-proxy', request] - if kind == 'cleanup_report': - return ['cleanup-report-proxy'] - raise ValueError(f'unsupported kind {kind}') diff --git a/scripts/agentctl_w4.py b/scripts/agentctl_w4.py index 3e4cb67a..0b30d826 100644 --- a/scripts/agentctl_w4.py +++ b/scripts/agentctl_w4.py @@ -312,6 +312,21 @@ def deployment_set_phase(db, deploy_id: str, phase: str, *, receipt: Optional[di error: Optional[str] = None) -> dict: if phase not in DEPLOY_PHASES: raise ValueError(f'invalid deploy phase: {phase}') + current = deployment_get(db, deploy_id) + if current is None: + raise RuntimeError(f'unknown deploy: {deploy_id}') + from_phase = current['phase'] + # Terminal passed has empty outbound set; recover may jump failed/recovery_required → passed. + allowed = { + 'prepared': {'deploying', 'failed', 'recovery_required'}, + 'deploying': {'accepting', 'failed', 'recovery_required'}, + 'accepting': {'passed', 'failed', 'recovery_required'}, + 'failed': {'recovery_required', 'passed'}, + 'recovery_required': {'deploying', 'passed'}, + 'passed': set(), + } + if phase != from_phase and phase not in allowed.get(from_phase, set()): + raise RuntimeError(f'illegal transition {from_phase} -> {phase}') db.execute( '''UPDATE deployments SET phase=?,receipt=?,last_error=?,updated=? WHERE id=?''', (phase, json.dumps(receipt or {}, sort_keys=True) if receipt is not None else None, @@ -532,6 +547,33 @@ def cleanup_visibility_report( } +def project_cleanup_markdown(entries): + """Render cleanup visibility projection from structured entries. + + Ported for W2-U5 projection reuse; structured report remains authoritative. + """ + lines = [ + '# Cleanup visibility projection', + '', + '> Generated by `agentctl docs project --kind cleanup`. ' + 'Not a lock; machine `.locks` is authoritative.', + '', + '| path | branch | reason | cleanup_owner |', + '| --- | --- | --- | --- |', + ] + if not entries: + lines.append('| — | — | none_pending | — |') + else: + for item in entries: + path = str(item.get('path') or '').replace('|', '\\|') + branch = str(item.get('branch') or '').replace('|', '\\|') + reason = str(item.get('reason') or '').replace('|', '\\|') + owner = str(item.get('cleanup_owner') or 'independent-maintainer').replace('|', '\\|') + lines.append(f'| {path} | {branch} | {reason} | {owner} |') + lines.append('') + return '\n'.join(lines) + + 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 diff --git a/scripts/tests/test_agentctl_w4.py b/scripts/tests/test_agentctl_w4.py index 4c493c19..1f53b70f 100644 --- a/scripts/tests/test_agentctl_w4.py +++ b/scripts/tests/test_agentctl_w4.py @@ -151,6 +151,21 @@ class AgentctlW4Tests(unittest.TestCase): self.assertEqual(receipt.get("recover_reason"), "operator cleared after fake failure") + def test_deploy_phase_transition_rejects_illegal_jump(self): + deploy = self.w4.deployment_create( + self.db, deploy_id="dep-phase", task="phase-task", + env="dev", expected_commit="abcdef4", manifest=None, dry_run=True) + self.assertEqual(deploy["phase"], "prepared") + self.w4.deployment_set_phase(self.db, "dep-phase", "deploying") + self.w4.deployment_set_phase(self.db, "dep-phase", "accepting") + passed = self.w4.deployment_set_phase(self.db, "dep-phase", "passed") + self.assertEqual(passed["phase"], "passed") + with self.assertRaises(RuntimeError) as ctx: + self.w4.deployment_set_phase(self.db, "dep-phase", "deploying") + self.assertIn("illegal transition passed -> deploying", str(ctx.exception)) + still = self.w4.deployment_get(self.db, "dep-phase") + self.assertEqual(still["phase"], "passed") + class DarwinCwdVisibilityTests(unittest.TestCase): def setUp(self):