import importlib.util import os import subprocess import sys import tempfile import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts" / "main_loop.py" _MAIN_LOOP_SPEC = importlib.util.spec_from_file_location( "playbook_main_loop", ROOT / "scripts" / "main_loop.py" ) assert _MAIN_LOOP_SPEC and _MAIN_LOOP_SPEC.loader MAIN_LOOP = importlib.util.module_from_spec(_MAIN_LOOP_SPEC) sys.modules[_MAIN_LOOP_SPEC.name] = MAIN_LOOP _MAIN_LOOP_SPEC.loader.exec_module(MAIN_LOOP) def run_cli(*args: str, cwd: Path) -> subprocess.CompletedProcess[str]: return subprocess.run( [sys.executable, str(SCRIPT), *args], cwd=cwd, capture_output=True, text=True, ) def start_cli(*args: str, cwd: Path) -> subprocess.Popen[str]: return subprocess.Popen( [sys.executable, str(SCRIPT), *args], cwd=cwd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) def run_git(root: Path, *args: str) -> subprocess.CompletedProcess[str]: return subprocess.run( ["git", *args], cwd=root, capture_output=True, text=True, ) def parse_assignments(output: str) -> dict[str, str]: return dict(line.split("=", 1) for line in output.splitlines() if "=" in line) def verification_evidence(commit: str, detail: str) -> str: return f"commit={commit}; result=pass; {detail}" def review_evidence(commit: str, base: str) -> str: return f"commit={commit}; base={base}; standards=pass; spec=pass" def init_repo(root: Path) -> None: initialized = run_git(root, "init", "-b", "main") if initialized.returncode != 0: initialized = run_git(root, "init") if initialized.returncode == 0: run_git(root, "checkout", "-b", "main") if initialized.returncode != 0: raise AssertionError(initialized.stderr) run_git(root, "config", "user.name", "Test User") run_git(root, "config", "user.email", "test@example.invalid") (root / ".gitignore").write_text(".scratch/\n", encoding="utf-8", newline="\n") (root / "app.txt").write_text("base\n", encoding="utf-8", newline="\n") run_git(root, "add", ".gitignore", "app.txt") committed = run_git(root, "commit", "-m", "base") if committed.returncode != 0: raise AssertionError(committed.stderr) def write_ticket( issues: Path, number: str, slug: str, title: str, blocked_by: str = "None - can start immediately.", status: str = "ready-for-agent", ) -> Path: path = issues / f"{number}-{slug}.md" path.write_text( "\n".join( [ f"# {number} - {title}", "", f"**What to build:** Deliver {title}.", "", f"**Blocked by:** {blocked_by}", "", f"**Status:** {status}", "", "- [ ] The behavior is observable.", "", ] ), encoding="utf-8", newline="\n", ) return path def write_feature(root: Path, slug: str) -> Path: feature = root / ".scratch" / slug issues = feature / "issues" issues.mkdir(parents=True) (feature / "spec.md").write_text( f"# {slug} spec\n", encoding="utf-8", newline="\n" ) return issues class MainLoopCliTests(unittest.TestCase): def test_entrypoint_exposes_only_ticket_native_commands(self): with tempfile.TemporaryDirectory() as tmp_dir: result = run_cli("--help", cwd=Path(tmp_dir)) self.assertEqual(result.returncode, 0, msg=result.stderr) for command in ( "enqueue", "status", "claim", "finish", "heartbeat", "reclaim", "integrate", ): self.assertIn(command, result.stdout) def test_enqueue_writes_a_prettier_stable_queue(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual( (root / ".scratch" / "queue.md").read_text(encoding="utf-8"), "# Feature Queue\n\n" "\n\n" "- `alpha`\n\n" "\n", ) def test_cli_rejects_a_caller_supplied_clock(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) result = run_cli( "status", "--state-root", ".scratch", "--now", "2099-01-01T00:00:00Z", cwd=root, ) self.assertEqual(result.returncode, 2) self.assertIn("unrecognized arguments: --now", result.stderr) def test_in_place_claim_resumes_with_uncommitted_work(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) original = parse_assignments(claim.stdout) (root / "unfinished.txt").write_text( "preserve me\n", encoding="utf-8", newline="\n" ) resumed = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) self.assertEqual(resumed.returncode, 0, msg=resumed.stderr) context = parse_assignments(resumed.stdout) self.assertEqual(context["WORKSPACE"], original["WORKSPACE"]) self.assertEqual(context["BRANCH"], original["BRANCH"]) self.assertEqual( (root / "unfinished.txt").read_text(encoding="utf-8"), "preserve me\n", ) def test_released_in_place_ticket_preserves_uncommitted_work(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) original = parse_assignments(claim.stdout) (root / "unfinished.txt").write_text( "preserve me\n", encoding="utf-8", newline="\n" ) released = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "released", cwd=root, ) self.assertEqual(released.returncode, 0, msg=released.stderr) reclaimed = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-2", "--isolation", "in-place", cwd=root, ) self.assertEqual(reclaimed.returncode, 0, msg=reclaimed.stderr) context = parse_assignments(reclaimed.stdout) self.assertEqual(context["WORKSPACE"], original["WORKSPACE"]) self.assertEqual(context["BRANCH"], original["BRANCH"]) self.assertEqual( (root / "unfinished.txt").read_text(encoding="utf-8"), "preserve me\n", ) def test_worktree_claim_refuses_to_resume_on_the_wrong_branch(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "worktree", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) context = parse_assignments(claim.stdout) workspace = Path(context["WORKSPACE"]) wrong_branch = run_git(workspace, "checkout", "-b", "wrong-branch") self.assertEqual(wrong_branch.returncode, 0, msg=wrong_branch.stderr) resumed = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "worktree", cwd=root, ) self.assertEqual(resumed.returncode, 2) self.assertIn("claim workspace is not on", resumed.stderr) def test_blocked_feature_allows_later_development_but_integration_stays_ordered(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) alpha_issues = write_feature(root, "alpha") beta_issues = write_feature(root, "beta") write_ticket(alpha_issues, "01", "alpha", "Alpha") write_ticket(beta_issues, "01", "beta", "Beta") for feature in ("alpha", "beta"): enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", feature, cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) alpha_claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "alpha-owner", "--isolation", "worktree", cwd=root, ) busy = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "waiting-owner", "--isolation", "worktree", cwd=root, ) blocked = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "alpha-owner", "--result", "blocked", "--reason", "waiting for decision", cwd=root, ) beta_claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "beta-owner", "--isolation", "worktree", cwd=root, ) self.assertEqual(alpha_claim.returncode, 0, msg=alpha_claim.stderr) self.assertEqual(busy.stdout.strip(), "BUSY") self.assertEqual(blocked.returncode, 0, msg=blocked.stderr) self.assertEqual(beta_claim.returncode, 0, msg=beta_claim.stderr) self.assertEqual(parse_assignments(beta_claim.stdout)["FEATURE"], "beta") released = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "alpha-owner", "--result", "released", cwd=root, ) resumed_alpha = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "alpha-owner-2", "--isolation", "worktree", cwd=root, ) self.assertEqual(released.returncode, 0, msg=released.stderr) self.assertEqual(resumed_alpha.returncode, 0, msg=resumed_alpha.stderr) self.assertEqual(parse_assignments(resumed_alpha.stdout)["FEATURE"], "alpha") contexts = { "alpha": parse_assignments(resumed_alpha.stdout), "beta": parse_assignments(beta_claim.stdout), } owners = {"alpha": "alpha-owner-2", "beta": "beta-owner"} for feature in ("alpha", "beta"): workspace = Path(contexts[feature]["WORKSPACE"]) (workspace / f"{feature}.txt").write_text( f"{feature}\n", encoding="utf-8", newline="\n" ) run_git(workspace, "add", f"{feature}.txt") committed = run_git(workspace, "commit", "-m", f"implement {feature}") self.assertEqual(committed.returncode, 0, msg=committed.stderr) commit = run_git(workspace, "rev-parse", "HEAD").stdout.strip() finished = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", feature, "--ticket", "01", "--owner", owners[feature], "--result", "resolved", "--implementation-commit", commit, "--feature-head", contexts[feature]["BASE"], "--review-base", contexts[feature]["BASE"], "--verified", verification_evidence(commit, f"{feature} ticket tests"), "--reviewed", review_evidence(commit, contexts[feature]["BASE"]), cwd=root, ) self.assertEqual(finished.returncode, 0, msg=finished.stderr) barrier = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "new-owner", "--isolation", "worktree", cwd=root, ) beta_head = run_git( root, "rev-parse", "feature/beta" ).stdout.strip() beta_out_of_order = run_cli( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "beta", "--feature-head", beta_head, "--verified", verification_evidence(beta_head, "beta feature tests"), "--main-verified", verification_evidence(beta_head, "beta main candidate tests"), "--reviewed", review_evidence( beta_head, run_git(root, "rev-parse", "main").stdout.strip(), ), cwd=root, ) self.assertIn("INTEGRATION_REQUIRED=alpha", barrier.stdout) self.assertEqual(beta_out_of_order.returncode, 2) self.assertIn("earlier feature alpha", beta_out_of_order.stderr) alpha_head = run_git(root, "rev-parse", "feature/alpha").stdout.strip() main_before_alpha = run_git(root, "rev-parse", "main").stdout.strip() alpha_integrated = run_cli( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--feature-head", alpha_head, "--verified", verification_evidence(alpha_head, "alpha feature tests"), "--main-verified", verification_evidence(alpha_head, "alpha main candidate tests"), "--reviewed", review_evidence(alpha_head, main_before_alpha), cwd=root, ) self.assertEqual(alpha_integrated.returncode, 0, msg=alpha_integrated.stderr) beta_head = run_git(root, "rev-parse", "feature/beta").stdout.strip() main_after_alpha = run_git(root, "rev-parse", "main").stdout.strip() beta_needs_sync = run_cli( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "beta", "--feature-head", beta_head, "--verified", verification_evidence(beta_head, "beta feature tests"), "--main-verified", verification_evidence(beta_head, "beta main candidate tests"), "--reviewed", review_evidence(beta_head, main_after_alpha), cwd=root, ) self.assertEqual(beta_needs_sync.returncode, 0, msg=beta_needs_sync.stderr) self.assertIn("RETRY: feature needs main sync", beta_needs_sync.stdout) beta_integration = root / ".scratch" / "worktrees" / "_integration" / "beta" synced = run_git(beta_integration, "merge", "--no-edit", "main") self.assertEqual(synced.returncode, 0, msg=synced.stderr) synced_beta_head = run_git(root, "rev-parse", "feature/beta").stdout.strip() beta_integrated = run_cli( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "beta", "--feature-head", synced_beta_head, "--verified", verification_evidence( synced_beta_head, "beta feature tests after sync" ), "--main-verified", verification_evidence( synced_beta_head, "beta main candidate tests after sync" ), "--reviewed", review_evidence(synced_beta_head, main_after_alpha), cwd=root, ) self.assertEqual(beta_integrated.returncode, 0, msg=beta_integrated.stderr) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertIn("FEATURE=alpha STATE=integrated", status.stdout) self.assertIn("FEATURE=beta STATE=integrated", status.stdout) def test_integration_block_allows_later_development_until_released(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) alpha_issues = write_feature(root, "alpha") beta_issues = write_feature(root, "beta") write_ticket( alpha_issues, "01", "alpha", "Alpha", status="resolved", ) write_ticket(beta_issues, "01", "beta", "Beta") for feature in ("alpha", "beta"): enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", feature, cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) blocked = run_cli( "block-feature", "--state-root", ".scratch", "--feature", "alpha", "--reason", "main sync conflict", cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) beta_claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "beta-owner", "--isolation", "worktree", cwd=root, ) released = run_cli( "release-feature", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) barrier = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "next-owner", "--isolation", "worktree", cwd=root, ) self.assertEqual(blocked.returncode, 0, msg=blocked.stderr) self.assertIn("FEATURE_BLOCKED=alpha", blocked.stdout) self.assertIn("FEATURE=alpha STATE=blocked", status.stdout) self.assertIn("FEATURE_BLOCKED=alpha REASON=main sync conflict", status.stdout) self.assertEqual(beta_claim.returncode, 0, msg=beta_claim.stderr) self.assertEqual(parse_assignments(beta_claim.stdout)["FEATURE"], "beta") self.assertEqual(released.returncode, 0, msg=released.stderr) self.assertIn("FEATURE_RELEASED=alpha", released.stdout) self.assertEqual(barrier.stdout.strip(), "INTEGRATION_REQUIRED=alpha") def test_reclaim_does_not_allow_the_caller_to_shorten_the_lease(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1-20260810-a1b2", "--isolation", "worktree", cwd=root, ) reclaim = run_cli( "reclaim", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-2-20260810-c3d4", "--stale-after-seconds", "0", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) self.assertEqual(reclaim.returncode, 2) self.assertIn("unrecognized arguments: --stale-after-seconds 0", reclaim.stderr) def test_heartbeat_stale_reclaim_release_blocked_and_skipped(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") write_ticket(issues, "02", "second", "Second", "01 - First.") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "worktree", cwd=root, ) self.assertEqual(claim.returncode, 0, msg=claim.stderr) original = parse_assignments(claim.stdout) heartbeat = run_cli( "heartbeat", "--state-root", ".scratch", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", cwd=root, ) wrong_owner = run_cli( "heartbeat", "--state-root", ".scratch", "--feature", "alpha", "--ticket", "01", "--owner", "session-2", cwd=root, ) feature = MAIN_LOOP.load_feature(root / ".scratch", "alpha") ticket = feature.tickets["01"] metadata = dict(ticket.metadata) metadata["heartbeat_at"] = "2000-01-01T00:00:00Z" MAIN_LOOP.update_ticket_state(ticket, metadata=metadata) status = run_cli( "status", "--state-root", ".scratch", cwd=root, ) reclaim = run_cli( "reclaim", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-2", cwd=root, ) self.assertEqual(heartbeat.returncode, 0, msg=heartbeat.stderr) self.assertEqual(wrong_owner.returncode, 2) self.assertIn("owned by another session", wrong_owner.stderr) self.assertEqual(status.returncode, 0, msg=status.stderr) self.assertIn("CLAIM=alpha/01", status.stdout) self.assertIn("STALE=yes", status.stdout) self.assertEqual(reclaim.returncode, 0, msg=reclaim.stderr) reclaimed = parse_assignments(reclaim.stdout) self.assertEqual(reclaimed["WORKSPACE"], original["WORKSPACE"]) self.assertEqual(reclaimed["BRANCH"], original["BRANCH"]) blocked = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-2", "--result", "blocked", "--reason", "environment unavailable", cwd=root, ) blocked_status = run_cli( "status", "--state-root", ".scratch", cwd=root ) released = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-2", "--result", "released", cwd=root, ) resumed = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-3", "--isolation", "worktree", cwd=root, ) skipped = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-3", "--result", "skipped", "--reason", "explicit product decision", cwd=root, ) final_status = run_cli( "status", "--state-root", ".scratch", cwd=root ) self.assertEqual(blocked.returncode, 0, msg=blocked.stderr) self.assertIn("FEATURE=alpha STATE=blocked", blocked_status.stdout) self.assertIn("REASON=environment unavailable", blocked_status.stdout) self.assertEqual(released.returncode, 0, msg=released.stderr) self.assertEqual(resumed.returncode, 0, msg=resumed.stderr) self.assertEqual( parse_assignments(resumed.stdout)["WORKSPACE"], original["WORKSPACE"], ) self.assertEqual(skipped.returncode, 0, msg=skipped.stderr) self.assertIn("FEATURE=alpha STATE=active PARTIAL=yes", final_status.stdout) self.assertIn("FRONTIER=02", final_status.stdout) def test_parallel_processes_claim_distinct_frontier_tickets(self): if os.name == "nt": self.assertIsNotNone(MAIN_LOOP.msvcrt) else: self.assertIsNotNone(MAIN_LOOP.fcntl) with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") for number in ("01", "02", "03", "04"): write_ticket(issues, number, f"ticket-{number}", f"Ticket {number}") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) processes = [ start_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", f"session-{index}", "--isolation", "worktree", cwd=root, ) for index in range(4) ] results = [] for process in processes: stdout, stderr = process.communicate(timeout=30) results.append((process.returncode, stdout, stderr)) for returncode, _stdout, stderr in results: self.assertEqual(returncode, 0, msg=stderr) contexts = [parse_assignments(stdout) for _, stdout, _ in results] self.assertEqual( {context["TICKET"] for context in contexts}, {"01", "02", "03", "04"}, ) self.assertEqual( len({context["WORKSPACE"] for context in contexts}), 4, ) def test_in_place_claim_is_busy_while_a_worktree_claim_is_active(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") write_ticket(issues, "02", "second", "Second") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) worktree_claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "worktree", cwd=root, ) in_place_claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-2", "--isolation", "in-place", cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(worktree_claim.returncode, 0, msg=worktree_claim.stderr) self.assertEqual(in_place_claim.returncode, 0, msg=in_place_claim.stderr) self.assertEqual(in_place_claim.stdout.strip(), "BUSY") self.assertEqual(status.stdout.count("CLAIM=alpha/"), 1) def test_parallel_worktree_claims_retry_when_feature_head_advances(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") write_ticket(issues, "02", "second", "Second") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) first = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "worktree", cwd=root, ) second = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-2", "--isolation", "worktree", cwd=root, ) self.assertEqual(first.returncode, 0, msg=first.stderr) self.assertEqual(second.returncode, 0, msg=second.stderr) first_context = parse_assignments(first.stdout) second_context = parse_assignments(second.stdout) self.assertEqual(first_context["TICKET"], "01") self.assertEqual(second_context["TICKET"], "02") self.assertNotEqual(first_context["WORKSPACE"], second_context["WORKSPACE"]) first_workspace = Path(first_context["WORKSPACE"]) second_workspace = Path(second_context["WORKSPACE"]) self.assertTrue(first_workspace.is_dir()) self.assertTrue(second_workspace.is_dir()) (first_workspace / "first.txt").write_text( "first\n", encoding="utf-8", newline="\n" ) run_git(first_workspace, "add", "first.txt") committed = run_git(first_workspace, "commit", "-m", "implement first") self.assertEqual(committed.returncode, 0, msg=committed.stderr) first_commit = run_git(first_workspace, "rev-parse", "HEAD").stdout.strip() (second_workspace / "second.txt").write_text( "second\n", encoding="utf-8", newline="\n" ) run_git(second_workspace, "add", "second.txt") committed = run_git(second_workspace, "commit", "-m", "implement second") self.assertEqual(committed.returncode, 0, msg=committed.stderr) second_commit = run_git(second_workspace, "rev-parse", "HEAD").stdout.strip() first_finish = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", first_commit, "--feature-head", first_context["BASE"], "--review-base", first_context["BASE"], "--verified", verification_evidence(first_commit, "tests"), "--reviewed", review_evidence(first_commit, first_context["BASE"]), cwd=root, ) stale_finish = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "02", "--owner", "session-2", "--result", "resolved", "--implementation-commit", second_commit, "--feature-head", second_context["BASE"], "--review-base", second_context["BASE"], "--verified", verification_evidence(second_commit, "tests"), "--reviewed", review_evidence(second_commit, second_context["BASE"]), cwd=root, ) self.assertEqual(first_finish.returncode, 0, msg=first_finish.stderr) self.assertEqual(stale_finish.returncode, 0, msg=stale_finish.stderr) self.assertIn("RETRY: feature advanced", stale_finish.stdout) advanced_head = run_git(root, "rev-parse", "feature/alpha").stdout.strip() shortcut_finish = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "02", "--owner", "session-2", "--result", "resolved", "--implementation-commit", second_commit, "--feature-head", advanced_head, "--review-base", advanced_head, "--verified", verification_evidence(second_commit, "stale tests"), "--reviewed", review_evidence(second_commit, advanced_head), cwd=root, ) self.assertEqual(shortcut_finish.returncode, 2) self.assertIn( "ticket branch must include the verified feature head", shortcut_finish.stderr, ) synced = run_git(second_workspace, "merge", "--no-edit", "feature/alpha") self.assertEqual(synced.returncode, 0, msg=synced.stderr) synced_commit = run_git(second_workspace, "rev-parse", "HEAD").stdout.strip() stale_review = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "02", "--owner", "session-2", "--result", "resolved", "--implementation-commit", synced_commit, "--feature-head", advanced_head, "--review-base", second_context["BASE"], "--verified", verification_evidence(synced_commit, "tests after sync"), "--reviewed", review_evidence(synced_commit, second_context["BASE"]), cwd=root, ) self.assertEqual(stale_review.returncode, 2) self.assertIn( "review base must equal the verified feature head", stale_review.stderr, ) second_finish = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "02", "--owner", "session-2", "--result", "resolved", "--implementation-commit", synced_commit, "--feature-head", advanced_head, "--review-base", advanced_head, "--verified", verification_evidence(synced_commit, "tests after sync"), "--reviewed", review_evidence(synced_commit, advanced_head), cwd=root, ) self.assertEqual(second_finish.returncode, 0, msg=second_finish.stderr) self.assertIn("RESOLVED=alpha/02", second_finish.stdout) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertIn("FEATURE=alpha STATE=ready-to-integrate", status.stdout) def test_finish_recovers_the_actual_ticket_merge_and_review_base(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "worktree", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) context = parse_assignments(claim.stdout) ticket_workspace = Path(context["WORKSPACE"]) integration_workspace = ( root / ".scratch" / "worktrees" / "_integration" / "alpha" ) (ticket_workspace / "feature.txt").write_text( "delivered\n", encoding="utf-8", newline="\n" ) run_git(ticket_workspace, "add", "feature.txt") committed = run_git( ticket_workspace, "commit", "-m", "deliver ticket" ) self.assertEqual(committed.returncode, 0, msg=committed.stderr) ticket_head = run_git( ticket_workspace, "rev-parse", "HEAD" ).stdout.strip() merged = run_git( integration_workspace, "merge", "--no-ff", "-m", "integrate before ticket state write", context["BRANCH"], ) self.assertEqual(merged.returncode, 0, msg=merged.stderr) ticket_merge = run_git( integration_workspace, "rev-parse", "HEAD" ).stdout.strip() (integration_workspace / "later.txt").write_text( "later feature work\n", encoding="utf-8", newline="\n" ) run_git(integration_workspace, "add", "later.txt") advanced = run_git( integration_workspace, "commit", "-m", "advance feature" ) self.assertEqual(advanced.returncode, 0, msg=advanced.stderr) common_args = ( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", ticket_head, ) wrong_base = run_cli( *common_args, "--feature-head", ticket_head, "--review-base", ticket_head, "--verified", verification_evidence(ticket_head, "ticket tests"), "--reviewed", review_evidence(ticket_head, ticket_head), cwd=root, ) self.assertEqual(wrong_base.returncode, 2) self.assertIn( "verified feature head must equal the ticket integration base", wrong_base.stderr, ) recovered = run_cli( *common_args, "--feature-head", context["BASE"], "--review-base", context["BASE"], "--verified", verification_evidence(ticket_head, "ticket tests"), "--reviewed", review_evidence(ticket_head, context["BASE"]), cwd=root, ) self.assertEqual(recovered.returncode, 0, msg=recovered.stderr) self.assertIn(f"INTEGRATION_COMMIT={ticket_merge}", recovered.stdout) def test_serial_in_place_claim_resume_and_finish(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") write_ticket(issues, "02", "second", "Second", "01 - First.") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) resumed = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) busy = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-2", "--isolation", "in-place", cwd=root, ) self.assertEqual(claim.returncode, 0, msg=claim.stderr) context = parse_assignments(claim.stdout) self.assertEqual(context["FEATURE"], "alpha") self.assertEqual(context["TICKET"], "01") self.assertEqual(context["BRANCH"], "ticket/alpha/01-first") self.assertEqual( Path(context["STATE_ROOT"]), (root / ".scratch").resolve(), ) self.assertEqual( run_git(root, "branch", "--show-current").stdout.strip(), "ticket/alpha/01-first", ) self.assertEqual(resumed.returncode, 0, msg=resumed.stderr) self.assertEqual(parse_assignments(resumed.stdout), context) self.assertEqual(busy.returncode, 0, msg=busy.stderr) self.assertEqual(busy.stdout.strip(), "BUSY") (root / "app.txt").write_text( "base\nimplemented\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "app.txt") implementation = run_git(root, "commit", "-m", "implement first") self.assertEqual(implementation.returncode, 0, msg=implementation.stderr) implementation_commit = run_git(root, "rev-parse", "HEAD").stdout.strip() finish = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", implementation_commit, "--feature-head", context["BASE"], "--review-base", context["BASE"], "--verified", verification_evidence( implementation_commit, "python -m unittest" ), "--reviewed", review_evidence(implementation_commit, context["BASE"]), cwd=root, ) status = run_cli( "status", "--state-root", ".scratch", cwd=root ) self.assertEqual(finish.returncode, 0, msg=finish.stderr) self.assertIn("RESOLVED=alpha/01", finish.stdout) self.assertIn("INTEGRATION_COMMIT=", finish.stdout) self.assertEqual( run_git(root, "branch", "--show-current").stdout.strip(), "feature/alpha", ) self.assertIn("FEATURE=alpha STATE=active PARTIAL=no", status.stdout) self.assertIn("FRONTIER=02", status.stdout) def test_enqueue_and_status_report_all_frontier_tickets(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") write_ticket(issues, "02", "second", "Second") write_ticket( issues, "03", "combine", "Combine", "01 - First; 02 - Second.", ) enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) status = run_cli( "status", "--state-root", ".scratch", cwd=root ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(enqueue.stdout.strip(), "ENQUEUED=alpha") self.assertEqual(status.returncode, 0, msg=status.stderr) self.assertIn("FEATURE=alpha STATE=queued PARTIAL=no", status.stdout) self.assertIn("FRONTIER=01,02", status.stdout) queue = (root / ".scratch" / "queue.md").read_text(encoding="utf-8") self.assertIn("- `alpha`", queue) def test_status_reports_a_claimed_ticket_integration_error(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "alpha") ticket_path = write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) ticket = MAIN_LOOP.parse_ticket(ticket_path) MAIN_LOOP.update_ticket_state( ticket, status="claimed", metadata={ "claimed_by": "session-1", "heartbeat_at": "2099-01-01T00:00:00Z", "isolation": "worktree", "workspace": "ticket-workspace", "last_error": "merge failed: conflict", }, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(status.returncode, 0, msg=status.stderr) self.assertIn( "TICKET_ERROR=alpha/01 REASON=merge failed: conflict", status.stdout, ) def test_enqueue_rejects_missing_blocker_before_writing_queue(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "broken") write_ticket(issues, "01", "broken", "Broken", "99 - Missing.") result = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "broken", cwd=root, ) self.assertEqual(result.returncode, 2) self.assertIn("unknown blocker 99", result.stderr) self.assertFalse((root / ".scratch" / "queue.md").exists()) def test_enqueue_rejects_partially_parsed_blockers(self): cases = ( "None; 99 - Hidden.", "01 - First; garbage", ) for blocked_by in cases: with self.subTest(blocked_by=blocked_by), tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "broken") write_ticket(issues, "01", "first", "First") write_ticket( issues, "02", "second", "Second", blocked_by, ) result = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "broken", cwd=root, ) self.assertEqual(result.returncode, 2) self.assertIn("malformed Blocked by field", result.stderr) self.assertFalse((root / ".scratch" / "queue.md").exists()) def test_enqueue_rejects_cycles_and_invalid_status_without_writing_queue(self): cases = ("cycle", "invalid-status") for case in cases: with self.subTest(case=case), tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "broken") if case == "cycle": write_ticket(issues, "01", "first", "First", "02 - Second.") write_ticket(issues, "02", "second", "Second", "01 - First.") expected = "ticket dependency cycle" else: write_ticket( issues, "01", "first", "First", status="done", ) expected = "invalid status done" result = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "broken", cwd=root, ) self.assertEqual(result.returncode, 2) self.assertIn(expected, result.stderr) self.assertFalse((root / ".scratch" / "queue.md").exists()) def test_enqueue_rejects_duplicate_ticket_machine_blocks(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "broken") ticket = write_ticket(issues, "01", "first", "First") original = ticket.read_text(encoding="utf-8") block = ( "\n" "{}\n" "\n" ) ticket.write_text( f"{original}\n{block}\n{block}", encoding="utf-8", newline="\n", ) result = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "broken", cwd=root, ) self.assertEqual(result.returncode, 2) self.assertIn("multiple ticket state blocks", result.stderr) self.assertFalse((root / ".scratch" / "queue.md").exists()) def test_status_rejects_duplicate_queue_machine_blocks(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) queue = root / ".scratch" / "queue.md" queue.write_text( queue.read_text(encoding="utf-8") + "\n\n" + "- `alpha`\n" + "\n", encoding="utf-8", newline="\n", ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(status.returncode, 2) self.assertIn("queue.md has an invalid managed block", status.stderr) def test_claim_records_environment_preparation_failures_as_blocked(self): cases = ( ("dirty", "in-place", "in-place checkout is dirty"), ("detached", "in-place", "in-place claim requires an attached HEAD"), ("worktree-path", "worktree", "worktree path is not empty"), ) for case, isolation, expected in cases: with self.subTest(case=case), tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) if case == "dirty": (root / "app.txt").write_text( "dirty\n", encoding="utf-8", newline="\n" ) elif case == "detached": detached = run_git(root, "checkout", "--detach", "HEAD") self.assertEqual(detached.returncode, 0, msg=detached.stderr) else: occupied = ( root / ".scratch" / "worktrees" / "_integration" / "alpha" ) occupied.mkdir(parents=True) (occupied / "keep.txt").write_text( "occupied\n", encoding="utf-8", newline="\n" ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", isolation, cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(claim.returncode, 2) self.assertIn(expected, claim.stderr) self.assertEqual(status.returncode, 0, msg=status.stderr) self.assertIn("FEATURE=alpha STATE=blocked", status.stdout) self.assertIn("BLOCKED=alpha/01", status.stdout) self.assertIn(expected, status.stdout) released = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "released", cwd=root, ) self.assertEqual(released.returncode, 0, msg=released.stderr) def test_finish_binds_review_to_ticket_head_and_verified_feature_head(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) self.assertEqual(claim.returncode, 0, msg=claim.stderr) context = parse_assignments(claim.stdout) (root / "reviewed.txt").write_text( "reviewed\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "reviewed.txt") first_commit = run_git(root, "commit", "-m", "reviewed implementation") self.assertEqual(first_commit.returncode, 0, msg=first_commit.stderr) reviewed_commit = run_git(root, "rev-parse", "HEAD").stdout.strip() (root / "unreviewed.txt").write_text( "unreviewed\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "unreviewed.txt") second_commit = run_git(root, "commit", "-m", "unreviewed follow-up") self.assertEqual(second_commit.returncode, 0, msg=second_commit.stderr) ticket_head = run_git(root, "rev-parse", "HEAD").stdout.strip() stale_commit_review = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", ticket_head, "--feature-head", context["BASE"], "--review-base", context["BASE"], "--verified", verification_evidence(ticket_head, "ticket tests"), "--reviewed", review_evidence(reviewed_commit, context["BASE"]), cwd=root, ) wrong_review_base = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", ticket_head, "--feature-head", context["BASE"], "--review-base", reviewed_commit, "--verified", verification_evidence(ticket_head, "ticket tests"), "--reviewed", review_evidence(ticket_head, reviewed_commit), cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(stale_commit_review.returncode, 2) self.assertIn( "review commit must equal ticket branch HEAD", stale_commit_review.stderr, ) self.assertEqual(wrong_review_base.returncode, 2) self.assertIn( "review base must equal the verified feature head", wrong_review_base.stderr, ) self.assertIn("CLAIM=alpha/01", status.stdout) def test_finish_rejects_unbound_verification_evidence(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) context = parse_assignments(claim.stdout) (root / "feature.txt").write_text( "delivered\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "feature.txt") committed = run_git(root, "commit", "-m", "deliver alpha") self.assertEqual(committed.returncode, 0, msg=committed.stderr) implementation_commit = run_git(root, "rev-parse", "HEAD").stdout.strip() finish = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", implementation_commit, "--feature-head", context["BASE"], "--review-base", context["BASE"], "--verified", "result=pass; ticket tests", "--reviewed", review_evidence(implementation_commit, context["BASE"]), cwd=root, ) conflicting = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", implementation_commit, "--feature-head", context["BASE"], "--review-base", context["BASE"], "--verified", ( f"commit={implementation_commit}; result=fail; " "result=pass; ticket tests" ), "--reviewed", review_evidence(implementation_commit, context["BASE"]), cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(finish.returncode, 2) self.assertIn("verification evidence must include commit", finish.stderr) self.assertEqual(conflicting.returncode, 2) self.assertIn( "verification evidence must include exactly one result", conflicting.stderr, ) self.assertIn("CLAIM=alpha/01", status.stdout) def test_partial_integration_requires_explicit_authorization(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "optional", "Optional") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) skipped = run_cli( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "skipped", "--reason", "explicit product decision", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) self.assertEqual(skipped.returncode, 0, msg=skipped.stderr) feature_head = run_git(root, "rev-parse", "feature/alpha").stdout.strip() main_head = run_git(root, "rev-parse", "main").stdout.strip() common_args = ( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--feature-head", feature_head, "--verified", verification_evidence(feature_head, "feature tests"), "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", review_evidence(feature_head, main_head), ) rejected = run_cli(*common_args, cwd=root) accepted = run_cli(*common_args, "--allow-partial", cwd=root) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(rejected.returncode, 2) self.assertIn("pass --allow-partial explicitly", rejected.stderr) self.assertEqual(accepted.returncode, 0, msg=accepted.stderr) self.assertIn("INTEGRATED=alpha", accepted.stdout) self.assertIn("FEATURE=alpha STATE=integrated PARTIAL=yes", status.stdout) def test_integrate_recovers_when_feature_is_already_on_main(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket( issues, "01", "delivered", "Delivered", status="resolved", ) enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) created = run_git(root, "checkout", "-b", "feature/alpha") self.assertEqual(created.returncode, 0, msg=created.stderr) (root / "feature.txt").write_text( "delivered\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "feature.txt") committed = run_git(root, "commit", "-m", "deliver alpha") self.assertEqual(committed.returncode, 0, msg=committed.stderr) feature_head = run_git(root, "rev-parse", "HEAD").stdout.strip() checkout_main = run_git(root, "checkout", "main") self.assertEqual(checkout_main.returncode, 0, msg=checkout_main.stderr) review_base = run_git(root, "rev-parse", "main").stdout.strip() merged = run_git( root, "merge", "--no-ff", "-m", "integrate before state write", "feature/alpha", ) self.assertEqual(merged.returncode, 0, msg=merged.stderr) integration_commit = run_git( root, "rev-parse", "main" ).stdout.strip() (root / "later.txt").write_text( "later main work\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "later.txt") advanced = run_git(root, "commit", "-m", "advance main") self.assertEqual(advanced.returncode, 0, msg=advanced.stderr) main_head = run_git(root, "rev-parse", "main").stdout.strip() recovered = run_cli( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--feature-head", feature_head, "--verified", verification_evidence(feature_head, "feature tests"), "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", review_evidence(feature_head, review_base), cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(recovered.returncode, 0, msg=recovered.stderr) self.assertIn("INTEGRATED=alpha", recovered.stdout) self.assertIn( f"MAIN_INTEGRATION_COMMIT={integration_commit}", recovered.stdout, ) self.assertEqual(run_git(root, "rev-parse", "main").stdout.strip(), main_head) self.assertIn("FEATURE=alpha STATE=integrated", status.stdout) def test_integrate_retries_cleanup_after_integration_state_is_written(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket( issues, "01", "delivered", "Delivered", status="resolved", ) enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) state_root = root / ".scratch" feature = MAIN_LOOP.load_feature(state_root, "alpha") main_head = run_git(root, "rev-parse", "main").stdout.strip() MAIN_LOOP.write_feature_state( feature, { "feature_branch": "feature/alpha", "integration_commit": main_head, }, ) with mock.patch.object( MAIN_LOOP, "cleanup_integrated_feature_worktrees", return_value=["cleanup retried"], ) as cleanup: recovered = MAIN_LOOP.integrate_feature( state_root, root, "alpha", main_head, verification_evidence(main_head, "feature tests"), verification_evidence(main_head, "main candidate tests"), review_evidence(main_head, main_head), "main", False, ) cleanup.assert_called_once() self.assertIn("INTEGRATED=alpha", recovered) self.assertIn("WARNING=cleanup retried", recovered) def test_atomic_replace_failure_preserves_state_and_next_claim_is_unique(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") ticket_path = write_ticket(issues, "01", "first", "First") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) state_root = root / ".scratch" queue_path = state_root / "queue.md" queue_before = queue_path.read_text(encoding="utf-8") ticket_before = ticket_path.read_text(encoding="utf-8") with mock.patch.object( MAIN_LOOP.os, "replace", side_effect=OSError("injected replace failure"), ): with self.assertRaisesRegex(OSError, "injected replace failure"): MAIN_LOOP.atomic_write_text(queue_path, "broken\n") self.assertEqual(queue_path.read_text(encoding="utf-8"), queue_before) self.assertEqual(MAIN_LOOP.load_queue(queue_path), ["alpha"]) feature = MAIN_LOOP.load_feature(state_root, "alpha") ticket = feature.tickets["01"] with mock.patch.object( MAIN_LOOP.os, "replace", side_effect=OSError("injected replace failure"), ): with self.assertRaisesRegex(OSError, "injected replace failure"): MAIN_LOOP.update_ticket_state( ticket, status="claimed", metadata={"claimed_by": "failed-owner"}, ) self.assertEqual(ticket_path.read_text(encoding="utf-8"), ticket_before) self.assertEqual( MAIN_LOOP.parse_ticket(ticket_path).status, "ready-for-agent", ) self.assertEqual(list(state_root.rglob("*.tmp")), []) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(claim.returncode, 0, msg=claim.stderr) self.assertEqual(status.stdout.count("CLAIM=alpha/01"), 1) self.assertIn("OWNER=session-1", status.stdout) def test_full_serial_workflow_enforces_all_evidence_gates(self): with tempfile.TemporaryDirectory() as tmp_dir: root = Path(tmp_dir) init_repo(root) issues = write_feature(root, "alpha") write_ticket(issues, "01", "deliver", "Deliver") enqueue = run_cli( "enqueue", "--state-root", ".scratch", "--feature", "alpha", cwd=root, ) claim = run_cli( "claim", "--state-root", ".scratch", "--repo-root", ".", "--owner", "session-1", "--isolation", "in-place", cwd=root, ) self.assertEqual(enqueue.returncode, 0, msg=enqueue.stderr) self.assertEqual(claim.returncode, 0, msg=claim.stderr) context = parse_assignments(claim.stdout) (root / "feature.txt").write_text( "delivered\n", encoding="utf-8", newline="\n" ) run_git(root, "add", "feature.txt") committed = run_git(root, "commit", "-m", "deliver alpha") self.assertEqual(committed.returncode, 0, msg=committed.stderr) implementation_commit = run_git(root, "rev-parse", "HEAD").stdout.strip() finish_args = ( "finish", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--ticket", "01", "--owner", "session-1", "--result", "resolved", "--implementation-commit", implementation_commit, "--feature-head", context["BASE"], "--review-base", context["BASE"], ) missing_ticket_verification = run_cli( *finish_args, "--reviewed", review_evidence(implementation_commit, context["BASE"]), cwd=root, ) incomplete_ticket_review = run_cli( *finish_args, "--verified", verification_evidence(implementation_commit, "ticket tests"), "--reviewed", ( f"commit={implementation_commit}; base={context['BASE']}; " "standards=pass" ), cwd=root, ) resolved = run_cli( *finish_args, "--verified", verification_evidence(implementation_commit, "ticket tests"), "--reviewed", review_evidence(implementation_commit, context["BASE"]), cwd=root, ) self.assertEqual(missing_ticket_verification.returncode, 2) self.assertIn( "verification evidence is required", missing_ticket_verification.stderr, ) self.assertEqual(incomplete_ticket_review.returncode, 2) self.assertIn("spec=pass", incomplete_ticket_review.stderr) self.assertEqual(resolved.returncode, 0, msg=resolved.stderr) feature_head = run_git(root, "rev-parse", "feature/alpha").stdout.strip() main_before = run_git(root, "rev-parse", "main").stdout.strip() integrate_args = ( "integrate", "--state-root", ".scratch", "--repo-root", ".", "--feature", "alpha", "--feature-head", feature_head, ) missing_feature_verification = run_cli( *integrate_args, "--verified", "", "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", review_evidence(feature_head, main_before), cwd=root, ) missing_main_verification = run_cli( *integrate_args, "--verified", verification_evidence(feature_head, "feature tests"), "--main-verified", "", "--reviewed", review_evidence(feature_head, main_before), cwd=root, ) incomplete_final_review = run_cli( *integrate_args, "--verified", verification_evidence(feature_head, "feature tests"), "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", f"commit={feature_head}; base={main_before}; standards=pass", cwd=root, ) unbound_feature_verification = run_cli( *integrate_args, "--verified", "result=pass; feature tests", "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", review_evidence(feature_head, main_before), cwd=root, ) wrong_final_review_base = run_cli( *integrate_args, "--verified", verification_evidence(feature_head, "feature tests"), "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", review_evidence(feature_head, feature_head), cwd=root, ) self.assertEqual(missing_feature_verification.returncode, 2) self.assertIn( "feature verification evidence is required", missing_feature_verification.stderr, ) self.assertEqual(missing_main_verification.returncode, 2) self.assertIn( "main candidate verification evidence is required", missing_main_verification.stderr, ) self.assertEqual(incomplete_final_review.returncode, 2) self.assertIn("spec=pass", incomplete_final_review.stderr) self.assertEqual(unbound_feature_verification.returncode, 2) self.assertIn( "feature verification evidence must include commit", unbound_feature_verification.stderr, ) self.assertEqual(wrong_final_review_base.returncode, 2) self.assertIn( "review base must equal the latest main HEAD", wrong_final_review_base.stderr, ) self.assertEqual( run_git(root, "rev-parse", "main").stdout.strip(), main_before, ) integrated = run_cli( *integrate_args, "--verified", verification_evidence(feature_head, "feature tests"), "--main-verified", verification_evidence(feature_head, "main candidate tests"), "--reviewed", review_evidence(feature_head, main_before), cwd=root, ) status = run_cli("status", "--state-root", ".scratch", cwd=root) self.assertEqual(integrated.returncode, 0, msg=integrated.stderr) self.assertIn("INTEGRATED=alpha", integrated.stdout) self.assertIn("FEATURE=alpha STATE=integrated PARTIAL=no", status.stdout) self.assertTrue((root / "feature.txt").is_file()) if __name__ == "__main__": unittest.main()