import importlib.util import json import os import subprocess import sys import tempfile import unittest from datetime import datetime, timezone from pathlib import Path from types import SimpleNamespace from unittest import mock HERE = Path(__file__).parent sys.path.insert(0, str(HERE)) def load_module(name, filename): spec = importlib.util.spec_from_file_location(name, HERE / filename) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module worker = load_module("board_agent_worker", "worker.py") failure_handler = load_module("board_agent_failure_handler", "failure_handler.py") trigger = load_module("board_agent_trigger", "trigger.py") completion_watcher = load_module("board_agent_completion_watcher", "completion_watcher.py") # worker.py, trigger.py and completion_watcher.py each do a plain `import # fanout`, which Python caches in sys.modules by name -- so all three (and # this alias) share the exact same module object, unlike load_module()'s # separately-registered copies above. Tests below call fanout functions # and mock.patch.object() through this one shared reference. fanout = worker.fanout metrics = worker.metrics orchestration = worker.orchestration class ReviewerPhaseVisibilityTests(unittest.TestCase): def test_runtime_is_visible_and_validated_in_frontend(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('input.dataset.field = field', board_html) self.assertIn('"codex"', board_html) self.assertIn('"claude"', board_html) self.assertIn('"openrouter"', board_html) def test_frontend_lists_only_the_selected_openrouter_models(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") for model in ( "moonshotai/kimi-k2-thinking", "deepseek/deepseek-v3.2", "cohere/north-mini-code:free", "nvidia/nemotron-3-super-120b-a12b:free", ): self.assertIn(model, board_html) def test_frontend_rebuilds_model_options_when_runtime_changes(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("function modelOptionsForRuntime(runtime)", board_html) self.assertIn("function syncModelOptions(modelSelect, runtime, value)", board_html) self.assertIn("modelSelect.replaceChildren()", board_html) self.assertIn('field === "runtime"', board_html) self.assertIn('modelSelect, input.value, ""', board_html) def test_reviewer_outcome_phases_are_valid_backend_phases(self): self.assertEqual( trigger.PHASES, ( "idea", "geplant", "in_umsetzung", "in_umsetzung_leicht", "review", "bereit_fuer_deployment", "planung_mit_feedback", "deployed", "archiviert", "aufteilen_angefordert", "operator_angefordert", "operator_laeuft", "wartet_auf_kinder", "durch_kinder_blockiert", "operator_klaerung_noetig", "fixer_angefordert", "fixer_laeuft", "fixer_rueckfrage", "fixer_fehlgeschlagen", "klaerung_reparatur", "inspector_laeuft", "inspector_rueckfrage", "inspector_fehlgeschlagen", ), ) def test_frontend_has_clarification_config_and_answer_form(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") for phase in ("fixer_rueckfrage", "inspector_rueckfrage", "klaerung_reparatur"): self.assertIn(phase, board_html) self.assertIn("submitFixerAnswer", board_html) self.assertIn("Rückfrage (", board_html) self.assertIn("boardClarificationDrafts", board_html) def test_frontend_projects_operator_clarification_and_reason_block(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('"operator_klaerung_noetig"', board_html) # appendOperatorReason() must not live inside appendOrchestrationBadges() # -- that one bails out for a card with no parent/children, which is # exactly the needs_human/operator-crash case this column is for. self.assertIn("function appendOperatorReason(el, card)", board_html) self.assertIn("appendOperatorReason(el, card);", board_html) self.assertIn("card.human_action", board_html) self.assertIn("fixerAnswerDrafts", board_html) self.assertIn("answer.addEventListener(\"input\"", board_html) self.assertIn("fixerAnswerFocus", board_html) self.assertIn("rememberFixerAnswerFocus", board_html) self.assertIn("restoreFixerAnswerFocus", board_html) self.assertIn("setSelectionRange", board_html) def test_frontend_has_lifecycle_groups_for_reviewer_outcomes(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") for label in ("Backlog", "Ready", "Running", "Needs attention", "Deploying", "Done", "Archived"): self.assertIn(f'label: "{label}"', board_html) self.assertIn('["bereit_fuer_deployment"]', board_html) self.assertIn('"planung_mit_feedback"', board_html) self.assertIn('card.phase === "archiviert"', board_html) def test_frontend_projects_split_escalation_phases(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('"aufgeteilt"', board_html) self.assertIn('action.id === "repair"', board_html) self.assertIn("action.phase", board_html) def test_frontend_invokes_agents_through_contextual_buttons(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("function appendCardActions(el, card)", board_html) self.assertIn("card.available_actions", board_html) self.assertNotIn('window.prompt("Auftragstyp:', board_html) self.assertNotIn("dragstart", board_html) def test_frontend_displays_fanout_children(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("card.depends_on", board_html) self.assertIn("Fan-out-Kinder", board_html) def test_frontend_exposes_accessible_review_dialog_for_report_cards(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('', board_html) self.assertIn('detail.setAttribute("aria-haspopup", "dialog")', board_html) self.assertIn("if(card.review_report)", board_html) self.assertIn("/review`", board_html) class NestedChildCardTests(unittest.TestCase): """docs/intent/board-nested-child-intents.md: a card with a parent_id that still exists is rendered inside that parent's card instead of in its own column.""" def setUp(self): self.board_html = (HERE / "board.html").read_text(encoding="utf-8") def test_column_filter_excludes_cards_whose_parent_still_exists(self): self.assertIn("const ids = new Set(cachedCards.map(c => c.id))", self.board_html) self.assertIn( "|| !c.parent_id || !ids.has(c.parent_id)", self.board_html, ) def test_render_card_embeds_direct_children(self): self.assertIn( "cards.filter(c => c.parent_id === card.id)", self.board_html, ) self.assertIn('children.className = "children"', self.board_html) self.assertIn("renderCard(child, cards, chain)", self.board_html) def test_render_card_guards_against_parent_id_cycles(self): # A -> parent B, B -> parent A must not infinite-loop the recursion. self.assertIn("ancestry.has(card.id)", self.board_html) self.assertIn("function renderCard(card, cards, ancestry = new Set())", self.board_html) def test_orphaned_child_falls_back_to_its_own_column(self): # If the parent card has vanished, isNestedChild() must be false so # the child is not silently dropped off the board. self.assertIn("ids.has(c.parent_id)", self.board_html) def test_parent_offers_navigation_to_blocking_child(self): self.assertIn('button.textContent = `Zum Kind: ${childId}`', self.board_html) self.assertIn("card.blocking_children", self.board_html) class ContextualActionTests(unittest.TestCase): def action_ids(self, source, stored, card_id="card"): return [action["id"] for action in trigger.available_actions(card_id, source, stored)] def test_ready_intent_offers_operator_and_builder_actions(self): actions = self.action_ids("intent", {"phase": "geplant"}) self.assertEqual(actions[:3], ["operator", "builder", "builder-light"]) self.assertIn("diagnose", actions) self.assertNotIn("repair", actions) def test_in_progress_limit_escalation_never_offers_repair_or_split(self): stored = { "phase": "in_umsetzung", "last_run_status": "limit_reached", "auto_retry_attempts": 1, "escalation_attempts": [{"model": "one"}, {"model": "two"}], } self.assertFalse(trigger.escalation_exhausted(stored)) self.assertNotIn("repair", self.action_ids("intent", stored)) self.assertNotIn("split", self.action_ids("intent", stored)) def test_recovered_escalation_never_offers_repair(self): stored = { "phase": "review", "last_run_status": "done", "failure_detail": "old limit", "auto_retry_attempts": 1, "escalation_attempts": [{"model": "one"}], } self.assertNotIn("repair", self.action_ids("intent", stored)) def test_terminal_child_offers_diagnose_repair_but_parent_does_not_repair(self): child = { "phase": "planung_mit_feedback", "last_run_status": "limit_reached", "auto_retry_attempts": 1, "escalation_attempts": [{"model": "one"}], "operator_generated": True, "parent_id": "parent", } child_actions = self.action_ids("intent", child, "child") self.assertLess(child_actions.index("diagnose"), child_actions.index("repair")) self.assertNotIn("split", child_actions) parent = {"phase": "durch_kinder_blockiert", "blocking_children": ["child"]} parent_actions = self.action_ids("intent", parent, "parent") self.assertIn("diagnose", parent_actions) self.assertNotIn("repair", parent_actions) def test_healthy_active_and_aggregated_parents_offer_no_diagnose_or_repair(self): for phase in ( "operator_laeuft", "wartet_auf_kinder", "aufgeteilt", "bereit_fuer_deployment", "deployed", ): with self.subTest(phase=phase): actions = [a["id"] for a in trigger.available_actions("parent", "intent", {"phase": phase}, job_active=True)] self.assertNotIn("diagnose", actions) self.assertNotIn("repair", actions) def test_failed_operator_parent_with_cleared_blockers_has_no_child_actions(self): parent = { "phase": "fixer_fehlgeschlagen", "last_run_status": "failed", "orchestration_kind": "operator", "child_slugs": ["escalation-smoke-control", "escalation-smoke-many-files"], "depends_on": ["escalation-smoke-control", "escalation-smoke-many-files"], "blocking_children": None, } self.assertTrue(trigger.is_family_parent(parent)) actions = self.action_ids("intent", parent, "board-escalation-smoke-test") self.assertIn("diagnose", actions) self.assertNotIn("repair", actions) def test_research_and_archive_actions_match_backend_guards(self): self.assertIn("research", self.action_ids("research", {"phase": "recherche_angefordert"})) self.assertEqual(self.action_ids("research", {"phase": "archiviert"}), ["reactivate"]) class ReviewReportApiTests(unittest.TestCase): def _environment(self, repo_dir, state_path): return { "BOARD_REPO_DIR": str(repo_dir), "BOARD_INTENT_DIR": str(repo_dir / "docs" / "intent"), "BOARD_STATE_FILE": str(state_path), } def _prepare_report(self, root, state_report="docs/review/my-slug.md"): repo_dir = root / "repo" intent_dir = repo_dir / "docs" / "intent" review_dir = repo_dir / "docs" / "review" intent_dir.mkdir(parents=True) review_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n", encoding="utf-8") (review_dir / "my-slug.md").write_text("Review body\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps({"cards": {"my-slug": {"review_report": state_report}}}), encoding="utf-8", ) return repo_dir, state_path def test_read_review_report_returns_only_fixed_slug_report(self): with tempfile.TemporaryDirectory() as temp_dir: repo_dir, state_path = self._prepare_report(Path(temp_dir)) with mock.patch.dict( os.environ, self._environment(repo_dir, state_path), clear=False ): report = trigger.read_review_report("my-slug") self.assertEqual( report, {"path": "docs/review/my-slug.md", "content": "Review body\n"}, ) def test_merged_board_exposes_review_metadata_for_frontend(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) repo_dir, state_path = self._prepare_report(root) state_path.write_text( json.dumps( { "cards": { "my-slug": { "phase": "review", "review_decision": "planung_mit_feedback", "review_report": "docs/review/my-slug.md", } } } ), encoding="utf-8", ) with mock.patch.dict( os.environ, self._environment(repo_dir, state_path), clear=False ): card = trigger.merged_board()[0] self.assertEqual(card["review_decision"], "planung_mit_feedback") self.assertEqual(card["review_report"], "docs/review/my-slug.md") def test_merged_board_exposes_builder_report_for_frontend(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) repo_dir, state_path = self._prepare_report(root) state_path.write_text( json.dumps( { "cards": { "my-slug": { "phase": "in_umsetzung", "last_run_status": "no_changes", "builder_report": "Kein Diff noetig, weil ...", } } } ), encoding="utf-8", ) with mock.patch.dict( os.environ, self._environment(repo_dir, state_path), clear=False ): card = trigger.merged_board()[0] self.assertEqual(card["builder_report"], "Kein Diff noetig, weil ...") def test_read_review_report_rejects_state_supplied_path(self): with tempfile.TemporaryDirectory() as temp_dir: repo_dir, state_path = self._prepare_report( Path(temp_dir), state_report="../../etc/passwd" ) with mock.patch.dict( os.environ, self._environment(repo_dir, state_path), clear=False ): with self.assertRaises(LookupError): trigger.read_review_report("my-slug") def test_read_review_report_rejects_symlink_outside_review_directory(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) repo_dir, state_path = self._prepare_report(root) report_path = repo_dir / "docs" / "review" / "my-slug.md" report_path.unlink() outside_report = root / "outside.md" outside_report.write_text("must not be read\n", encoding="utf-8") report_path.symlink_to(outside_report) with mock.patch.dict( os.environ, self._environment(repo_dir, state_path), clear=False ): with self.assertRaises(LookupError): trigger.read_review_report("my-slug") def test_get_review_route_returns_report_json(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/my-slug/review" handler._send_json = mock.Mock() report = {"path": "docs/review/my-slug.md", "content": "Review body\n"} with mock.patch.object(trigger, "read_review_report", return_value=report): handler.do_GET() handler._send_json.assert_called_once_with(200, report) def test_get_review_route_rejects_encoded_path_components(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/my-slug%2F..%2Fetc/review" handler._send_json = mock.Mock() with mock.patch.object(trigger, "read_review_report") as read_report: handler.do_GET() read_report.assert_not_called() handler._send_json.assert_called_once_with(404, {"error": "not found"}) class IntentApiTests(ReviewReportApiTests): def _prepare_intent(self, root): repo_dir = root / "repo" intent_dir = repo_dir / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n\nIntent body\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {}}), encoding="utf-8") return repo_dir, state_path def test_read_intent_returns_document(self): with tempfile.TemporaryDirectory() as temp_dir: repo_dir, state_path = self._prepare_intent(Path(temp_dir)) with mock.patch.dict(os.environ, self._environment(repo_dir, state_path), clear=False): intent = trigger.read_intent("my-slug") self.assertEqual(intent, {"path": "docs/intent/my-slug.md", "content": "# My Slug\n\nIntent body\n"}) def test_get_intent_route_returns_document_json(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/my-slug/intent" handler._send_json = mock.Mock() intent = {"path": "docs/intent/my-slug.md", "content": "Intent body\n"} with mock.patch.object(trigger, "read_intent", return_value=intent): handler.do_GET() handler._send_json.assert_called_once_with(200, intent) class AgentPromptApiTests(unittest.TestCase): """The Agents column's per-phase 'view role prompt' links, backed by read_agent_prompt() and GET /api/board/agents/.""" def _prepare(self, root): role_dir = root / "agents" role_dir.mkdir(parents=True) (role_dir / "reviewer.md").write_text("Du bist die Reviewer-Rolle.\n", encoding="utf-8") return role_dir def test_read_agent_prompt_returns_known_role_file(self): with tempfile.TemporaryDirectory() as temp_dir: role_dir = self._prepare(Path(temp_dir)) with mock.patch.dict(os.environ, {"BOARD_ROLE_DIR": str(role_dir)}, clear=False): prompt = trigger.read_agent_prompt("reviewer") self.assertEqual( prompt, {"path": "agents/reviewer.md", "content": "Du bist die Reviewer-Rolle.\n"}, ) def test_read_agent_prompt_rejects_unknown_role(self): with tempfile.TemporaryDirectory() as temp_dir: role_dir = self._prepare(Path(temp_dir)) with mock.patch.dict(os.environ, {"BOARD_ROLE_DIR": str(role_dir)}, clear=False): with self.assertRaises(LookupError): trigger.read_agent_prompt("../../etc/passwd") def test_get_agent_route_returns_prompt_json(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/agents/reviewer" handler._send_json = mock.Mock() prompt = {"path": "agents/reviewer.md", "content": "Du bist die Reviewer-Rolle.\n"} with mock.patch.object(trigger, "read_agent_prompt", return_value=prompt) as read_prompt: handler.do_GET() read_prompt.assert_called_once_with("reviewer") handler._send_json.assert_called_once_with(200, prompt) def test_get_agent_route_404s_for_unknown_role(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/agents/bogus" handler._send_json = mock.Mock() with mock.patch.object(trigger, "read_agent_prompt", side_effect=LookupError("unknown agent role")): handler.do_GET() handler._send_json.assert_called_once_with(404, {"error": "agent prompt not found"}) def test_frontend_has_agent_prompt_links(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("showAgentPrompt(role, label)", board_html) self.assertIn("/api/board/agents/", board_html) self.assertIn("PHASE_AGENT_ROLES", board_html) class ResearchApiTests(unittest.TestCase): def _prepare(self, root): repo_dir = root / "repo" research_dir = repo_dir / "docs" / "research-intent" report_dir = repo_dir / "docs" / "research" research_dir.mkdir(parents=True) report_dir.mkdir(parents=True) (research_dir / "my-slug.md").write_text("# Research\n", encoding="utf-8") (report_dir / "my-slug.md").write_text("Report\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {"my-slug": { "source": "research", "research_report": "docs/research/my-slug.md" }}}), encoding="utf-8") return repo_dir, state_path def test_read_research_intent_and_report(self): with tempfile.TemporaryDirectory() as temp_dir: repo_dir, state_path = self._prepare(Path(temp_dir)) environment = { "BOARD_REPO_DIR": str(repo_dir), "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(repo_dir / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(repo_dir / "docs" / "research-intent"), } with mock.patch.dict(os.environ, environment, clear=False): self.assertEqual(trigger.read_research_intent("my-slug")["path"], "docs/research-intent/my-slug.md") self.assertEqual(trigger.read_research_report("my-slug")["content"], "Report\n") def test_research_intent_rejects_outside_symlink(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) repo_dir, state_path = self._prepare(root) intent = repo_dir / "docs" / "research-intent" / "my-slug.md" intent.unlink() outside = root / "outside.md" outside.write_text("secret\n", encoding="utf-8") intent.symlink_to(outside) environment = { "BOARD_REPO_DIR": str(repo_dir), "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(repo_dir / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(repo_dir / "docs" / "research-intent"), } with mock.patch.dict(os.environ, environment, clear=False): with self.assertRaises(LookupError): trigger.read_research_intent("my-slug") class PhaseGuardTests(unittest.TestCase): def _intent_environment(self, root, state): return { "BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), } def _write_intent_state(self, root, phase): (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"slug": {"phase": phase}}}), encoding="utf-8") return state def test_research_transition_rejects_invalid_stored_phase(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) research_dir = root / "docs" / "research-intent" research_dir.mkdir(parents=True) (research_dir / "slug.md").write_text("# Research\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"slug": {"phase": "corrupt"}}}), encoding="utf-8") environment = { "BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(research_dir), } with mock.patch.dict(os.environ, environment, clear=False): with self.assertRaises(PermissionError): trigger.update_phase("slug", "recherche_angefordert") def test_archiving_from_pipeline_phases_is_bookkeeping_only(self): for phase in ("review", "bereit_fuer_deployment", "planung_mit_feedback", "deployed"): with self.subTest(phase=phase), tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, phase) with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False), \ mock.patch.object(trigger, "write_job") as write_job: self.assertIsNone(trigger.update_phase("slug", "archiviert")) write_job.assert_not_called() self.assertEqual(json.loads(state.read_text())["cards"]["slug"]["phase"], "archiviert") def test_repair_is_restricted_to_real_failures_or_negative_reviews(self): for phase in set(trigger.PHASE_RANK) | {"archiviert", "idea", "deployed"}: with self.subTest(phase=phase), tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, phase) environment = self._intent_environment(root, state) | { "BOARD_BUILDER_OWNER_JID": "builder@example", "AGENT_BUILDER_QUEUE_DIR": str(root / "builder-inbox"), } with mock.patch.dict(os.environ, environment, clear=False), \ mock.patch.object(trigger, "repo_job_active", return_value=False), \ mock.patch.object(trigger, "write_job", return_value="fixer-1"): if phase == "planung_mit_feedback": self.assertEqual( trigger.update_phase("slug", "klaerung_reparatur", "repair"), "fixer-1", ) else: with self.assertRaises(PermissionError): trigger.update_phase("slug", "klaerung_reparatur", "repair") def test_fixer_question_cannot_enter_review_without_a_commit(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, "fixer_rueckfrage") payload = json.loads(state.read_text()) payload["cards"]["slug"].update( {"fixer_question": "Welche Datei?", "fixer_attempts": 1} ) state.write_text(json.dumps(payload), encoding="utf-8") with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False), \ mock.patch.object(trigger, "write_job") as write_job: with self.assertRaises(LookupError): trigger.update_phase("slug", "review") saved = json.loads(state.read_text())["cards"]["slug"] self.assertEqual(saved["phase"], "fixer_rueckfrage") self.assertEqual(saved["fixer_question"], "Welche Datei?") write_job.assert_not_called() def test_archived_to_planned_is_bookkeeping_only(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, "archiviert") with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False), \ mock.patch.object(trigger, "write_job") as write_job: self.assertIsNone(trigger.update_phase("slug", "geplant")) write_job.assert_not_called() def test_archived_to_in_progress_is_rejected_without_job(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, "archiviert") with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False), \ mock.patch.object(trigger, "write_job") as write_job: with self.assertRaises(PermissionError): trigger.update_phase("slug", "in_umsetzung") write_job.assert_not_called() def test_operator_trigger_clears_outcomes_from_previous_pipeline(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, "geplant") data = json.loads(state.read_text(encoding="utf-8")) data["cards"]["slug"].update( { "lane": "schwer", "builder_report": "Altes Iterationslimit erreicht", "auto_retry_attempts": 1, "commit_hash": "deadbeef", "review_decision": "planung_mit_feedback", "review_report": "Alter Review", "failure_stage": "worker", "failure_detail": "Alter Fehler", "last_run_at": "2026-09-01T00:00:00Z", "operator_decision": "needs_human", "operator_reason": "Alter Grund", "human_action": "Alte Aktion", "model": "old-model", "reasoning_effort": "old-effort", } ) state.write_text(json.dumps(data), encoding="utf-8") environment = self._intent_environment(root, state) | { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False), \ mock.patch.object(trigger, "write_job", return_value="operator-1"): job_id = trigger.update_phase("slug", "operator_angefordert") self.assertEqual(job_id, "operator-1") saved = json.loads(state.read_text(encoding="utf-8"))["cards"]["slug"] self.assertEqual(saved["phase"], "operator_angefordert") self.assertEqual(saved["last_run_status"], "queued_operator") self.assertEqual(saved["model"], "") self.assertEqual(saved["reasoning_effort"], "") for stale_field in ( "lane", "builder_report", "auto_retry_attempts", "commit_hash", "review_decision", "review_report", "failure_stage", "failure_detail", "last_run_at", "operator_decision", "operator_reason", "human_action", ): self.assertNotIn(stale_field, saved) def test_forward_step_to_terminal_phase_calls_maybe_synthesize(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"slug": {"phase": "review"}}}), encoding="utf-8") with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False), \ mock.patch.object(trigger.fanout, "maybe_synthesize") as maybe_synthesize: with self.assertRaises(PermissionError): trigger.update_phase("slug", "bereit_fuer_deployment") maybe_synthesize.assert_not_called() def test_forward_step_is_allowed(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"slug": {"phase": "review"}}}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(root / "docs" / "intent")}, clear=False), mock.patch.object(trigger, "write_job"): with self.assertRaises(PermissionError): trigger.update_phase("slug", "bereit_fuer_deployment") def test_skipped_forward_step_is_rejected(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"slug": {"phase": "geplant"}}}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(root / "docs" / "intent")}, clear=False): with self.assertRaises(PermissionError): trigger.update_phase("slug", "deployed") def test_backward_step_is_allowed(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"slug": {"phase": "review"}}}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(root / "docs" / "intent")}, clear=False): trigger.update_phase("slug", "geplant") def test_split_move_rejected_without_limit_reached(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text( json.dumps({"cards": {"slug": {"phase": "in_umsetzung", "last_run_status": "done"}}}), encoding="utf-8", ) env = self._intent_environment(root, state) | { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, env, clear=False), \ mock.patch.object(trigger, "write_job") as write_job: with self.assertRaises(PermissionError): trigger.update_phase("slug", "aufteilen_angefordert") write_job.assert_not_called() def test_split_move_rejected_before_automatic_retry_is_spent(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text( json.dumps( { "cards": { "slug": { "phase": "in_umsetzung", "last_run_status": "limit_reached", "auto_retry_attempts": 0, } } } ), encoding="utf-8", ) env = self._intent_environment(root, state) | { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, env, clear=False), \ mock.patch.object(trigger, "write_job") as write_job: with self.assertRaises(PermissionError): trigger.update_phase("slug", "aufteilen_angefordert") write_job.assert_not_called() def test_split_move_allowed_after_retry_is_spent_and_queues_split_job(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text( json.dumps( { "cards": { "slug": { "phase": "in_umsetzung", "last_run_status": "limit_reached", "auto_retry_attempts": 1, "builder_report": "Zwischenstand ...", } } } ), encoding="utf-8", ) env = self._intent_environment(root, state) | { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, env, clear=False), \ mock.patch.object(trigger, "write_job", return_value="split-1") as write_job: job_id = trigger.update_phase("slug", "aufteilen_angefordert") self.assertEqual(job_id, "split-1") write_job.assert_called_once() self.assertEqual(write_job.call_args.kwargs["job_type"], "split") self.assertEqual(write_job.call_args.kwargs["builder_report"], "Zwischenstand ...") self.assertEqual(write_job.call_args.kwargs["runtime"], "codex") self.assertEqual(write_job.call_args.kwargs["iteration_limit"], 20) saved = json.loads(state.read_text())["cards"]["slug"] self.assertEqual(saved["phase"], "aufteilen_angefordert") self.assertEqual(saved["last_run_status"], "queued_split") def test_split_move_rejected_before_full_escalation_ladder_is_spent(self): # docs/intent/board-operator-escalation-ladder.md: a card with an # Operator-authored two-step ladder must exhaust both automatic # escalation attempts, not just the first, before "Aufteilen # angefordert" becomes legal -- auto_retry_attempts=1 alone (which # the flat pre-ladder check used to accept) is still mid-ladder. with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text( json.dumps( { "cards": { "slug": { "phase": "in_umsetzung", "last_run_status": "limit_reached", "auto_retry_attempts": 1, "escalation_attempts": [ {"model": "opus", "reasoning_effort": "high", "iteration_limit": 30}, {"model": "opus", "reasoning_effort": "high", "iteration_limit": 40}, ], } } } ), encoding="utf-8", ) env = self._intent_environment(root, state) | { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, env, clear=False), \ mock.patch.object(trigger, "write_job") as write_job: with self.assertRaises(PermissionError): trigger.update_phase("slug", "aufteilen_angefordert") write_job.assert_not_called() def test_split_move_allowed_once_full_escalation_ladder_is_spent(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "slug.md").write_text("# Slug\n", encoding="utf-8") state = root / "state.json" state.write_text( json.dumps( { "cards": { "slug": { "phase": "in_umsetzung", "last_run_status": "limit_reached", "auto_retry_attempts": 2, "escalation_attempts": [ {"model": "opus", "reasoning_effort": "high", "iteration_limit": 30}, {"model": "opus", "reasoning_effort": "high", "iteration_limit": 40}, ], "builder_report": "Zwischenstand ...", } } } ), encoding="utf-8", ) env = self._intent_environment(root, state) | { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, env, clear=False), \ mock.patch.object(trigger, "write_job", return_value="split-1") as write_job: job_id = trigger.update_phase("slug", "aufteilen_angefordert") self.assertEqual(job_id, "split-1") write_job.assert_called_once() def test_aufgeteilt_is_not_a_legal_patch_target(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, "in_umsetzung") with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False): with self.assertRaises(ValueError): trigger.update_phase("slug", "aufgeteilt") def test_leaving_aufteilen_angefordert_is_bookkeeping_only(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = self._write_intent_state(root, "aufteilen_angefordert") with mock.patch.dict(os.environ, self._intent_environment(root, state), clear=False), \ mock.patch.object(trigger, "write_job") as write_job: self.assertIsNone(trigger.update_phase("slug", "geplant")) write_job.assert_not_called() self.assertEqual(json.loads(state.read_text())["cards"]["slug"]["phase"], "geplant") def test_intent_remains_visible_from_validated_state_while_file_is_missing(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) intent_path = intent_dir / "slug.md" intent_path.write_text("# Stored title\n", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {}}), encoding="utf-8") env = { "BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(intent_dir), "BOARD_REPO_DIR": str(root), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), "BOARD_RUNTIME_DEFAULT": "codex", } with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(trigger, "write_job", return_value="slug-1"): trigger.update_phase("slug", "in_umsetzung") intent_path.unlink() card = trigger.merged_board()[0] self.assertEqual(card["title"], "Stored title") self.assertEqual(card["intent_path"], "docs/intent/slug.md") self.assertTrue(card["source_missing"]) intent_path.write_text("# Returned title\n", encoding="utf-8") with mock.patch.dict(os.environ, env, clear=False): card = trigger.merged_board()[0] self.assertEqual(card["title"], "Returned title") self.assertFalse(card["source_missing"]) class TriggerJobTypeTests(unittest.TestCase): def test_column_config_defaults_and_validation(self): with tempfile.TemporaryDirectory() as temp_dir: state_path = Path(temp_dir) / "state.json" with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): self.assertEqual(trigger.load_column_config()["in_umsetzung"]["iteration_limit"], 20) self.assertEqual(trigger.set_column_config("in_umsetzung", {"model": "gpt-5.6", "iteration_limit": 30})["model"], "gpt-5.6") self.assertEqual( trigger.set_column_config( "review", {"model": "deepseek/deepseek-v3.2"} )["model"], "deepseek/deepseek-v3.2", ) self.assertEqual(trigger.load_column_config()["in_umsetzung"]["iteration_limit"], 30) with self.assertRaises(ValueError): trigger.validate_column_config_value("review", {"iteration_limit": 10}) with self.assertRaises(ValueError): trigger.validate_column_config_value("review", {"model": "bad model"}) def test_column_config_routes(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/column-config" handler._send_json = mock.Mock() with mock.patch.object(trigger, "load_column_config", return_value={"review": {}}): handler.do_GET() handler._send_json.assert_called_once_with(200, {"review": {}}) handler = object.__new__(trigger.Handler) handler.path = "/api/board/column-config/nope" handler._send_json = mock.Mock() handler.do_POST() handler._send_json.assert_called_once_with(404, {"error": "not found"}) def test_runtime_validation_and_state_persistence(self): with tempfile.TemporaryDirectory() as temp_dir: state_path = Path(temp_dir) / "state.json" with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): self.assertEqual(trigger.set_board_runtime("claude"), "claude") self.assertEqual(trigger.board_runtime(), "claude") with self.assertRaises(ValueError): trigger.set_board_runtime("codex-or-claude") def test_write_job_defaults_to_builder_job_type(self): with tempfile.TemporaryDirectory() as inbox_dir: environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(inbox_dir) / "state.json"), } with mock.patch.dict(os.environ, environment, clear=False): job_id = trigger.write_job("my-slug", "My Title", "docs/intent/my-slug.md") payload = json.loads((Path(inbox_dir) / f"{job_id}.json").read_text(encoding="utf-8")) self.assertEqual(payload["job_type"], "builder") self.assertEqual(payload["slug"], "my-slug") def test_write_job_accepts_explicit_job_type_and_extra_fields(self): with tempfile.TemporaryDirectory() as inbox_dir: with mock.patch.dict(os.environ, {"BOARD_INBOX_DIR": inbox_dir}, clear=False): job_id = trigger.write_job( "my-slug", "", "", job_type="review", runtime="claude", commit_hash="abc1234", ) payload = json.loads((Path(inbox_dir) / f"{job_id}.json").read_text(encoding="utf-8")) self.assertEqual(payload["job_type"], "review") self.assertEqual(payload["commit_hash"], "abc1234") self.assertEqual(payload["runtime"], "claude") def test_research_transition_snapshots_selected_board_runtime(self): for runtime in ("codex", "claude", "openrouter"): with self.subTest(runtime=runtime), tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) research_dir = root / "docs" / "research-intent" research_dir.mkdir() (research_dir / "research-slug.md").write_text( "# Research\n\nQuestion\n", encoding="utf-8" ) state_path = root / "state.json" environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(research_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): trigger.set_column_config("recherche_laeuft", {"runtime": runtime}) job_id = trigger.update_phase("research-slug", "recherche_laeuft") payload = json.loads( (root / "inbox" / f"{job_id}.json").read_text(encoding="utf-8") ) state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(payload["job_type"], "research") self.assertEqual(payload["runtime"], runtime) self.assertEqual( payload["research_intent_path"], "docs/research-intent/research-slug.md", ) self.assertEqual(state["cards"]["research-slug"]["runtime"], runtime) def test_research_transition_snapshots_column_config_overrides(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) research_dir = root / "docs" / "research-intent" research_dir.mkdir() (research_dir / "research-slug.md").write_text("# Research\n", encoding="utf-8") state_path = root / "state.json" environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(research_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): trigger.set_column_config( "recherche_laeuft", {"runtime": "claude", "model": "claude-test", "reasoning_effort": "high"}, ) job_id = trigger.update_phase("research-slug", "recherche_laeuft") payload = json.loads((root / "inbox" / f"{job_id}.json").read_text(encoding="utf-8")) self.assertEqual( {key: payload[key] for key in ("runtime", "model", "reasoning_effort")}, {"runtime": "claude", "model": "claude-test", "reasoning_effort": "high"}, ) def test_research_transition_rejects_invalid_persisted_runtime(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) research_dir = root / "docs" / "research-intent" research_dir.mkdir() (research_dir / "research-slug.md").write_text("# Research\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps( { "cards": {}, "column_config": {"recherche_laeuft": {"runtime": "invalid"}}, } ), encoding="utf-8", ) environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(research_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): with self.assertRaises(ValueError): trigger.update_phase("research-slug", "recherche_laeuft") self.assertEqual(list((root / "inbox").glob("*.json")), []) class EnqueueBuilderJobTests(unittest.TestCase): def test_builder_transition_snapshots_column_config_overrides(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "slug.md").write_text("# Slug\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps({"cards": {"slug": {"phase": "geplant"}}}), encoding="utf-8", ) environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(intent_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): trigger.set_column_config( "in_umsetzung", { "runtime": "claude", "model": "claude-test", "reasoning_effort": "high", "iteration_limit": 30, }, ) job_id = trigger.update_phase("slug", "in_umsetzung") payload = json.loads( (root / "inbox" / f"{job_id}.json").read_text(encoding="utf-8") ) self.assertEqual( {key: payload[key] for key in ("runtime", "model", "reasoning_effort", "iteration_limit")}, { "runtime": "claude", "model": "claude-test", "reasoning_effort": "high", "iteration_limit": 30, }, ) def test_builder_transition_omits_empty_column_config_overrides(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "slug.md").write_text("# Slug\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps({"cards": {"slug": {"phase": "geplant"}}}), encoding="utf-8", ) environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(intent_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): job_id = trigger.update_phase("slug", "in_umsetzung") payload = json.loads( (root / "inbox" / f"{job_id}.json").read_text(encoding="utf-8") ) self.assertEqual(payload["runtime"], "codex") self.assertEqual(payload["iteration_limit"], 20) def test_leicht_column_trigger_defaults_to_openrouter_and_tags_lane(self): # See docs/intent/board-builder-weighted-lanes.md: the "leicht" # column is a second, independent Builder trigger with its own # column config default and its own lane tag on both the queued # job and the card's persisted state. with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "slug.md").write_text("# Slug\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps({"cards": {"slug": {"phase": "geplant"}}}), encoding="utf-8", ) environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(intent_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): job_id = trigger.update_phase("slug", "in_umsetzung_leicht") payload = json.loads( (root / "inbox" / f"{job_id}.json").read_text(encoding="utf-8") ) self.assertEqual(payload["runtime"], "openrouter") self.assertEqual(payload["model"], "deepseek/deepseek-v3.2") self.assertEqual(payload["lane"], "leicht") state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(state["cards"]["slug"]["lane"], "leicht") self.assertEqual(state["cards"]["slug"]["phase"], "in_umsetzung_leicht") def test_schwer_and_leicht_lanes_can_both_be_triggered_at_once(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "card-a.md").write_text("# Card A\n", encoding="utf-8") (intent_dir / "card-b.md").write_text("# Card B\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps( { "cards": { "card-a": {"phase": "geplant"}, "card-b": {"phase": "geplant"}, } } ), encoding="utf-8", ) environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(intent_dir), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } with mock.patch.dict(os.environ, environment, clear=False): trigger.update_phase("card-a", "in_umsetzung") # A second "schwer" trigger while the first is still queued # would be rejected -- but a "leicht" trigger for a # different card is unaffected, since each lane has its # own dedicated instance. job_id_b = trigger.update_phase("card-b", "in_umsetzung_leicht") self.assertTrue((root / "inbox" / f"{job_id_b}.json").exists()) def test_builder_trigger_rejects_same_card_and_busy_lane(self): # Each Builder column now maps to exactly one dedicated # agent-codex-builder instance/inbox (see # docs/intent/board-builder-weighted-lanes.md), so the cap is one # in-flight job per lane rather than one shared number -- a # "schwer" job blocks another "schwer" trigger but never a # "leicht" one. with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) inbox = root / "inbox" processing = root / "processing" inbox.mkdir() processing.mkdir() (inbox / "same.json").write_text( json.dumps({"job_type": "builder", "slug": "same", "lane": "schwer"}), encoding="utf-8", ) (processing / "other.json").write_text( json.dumps({"job_type": "builder", "slug": "other", "lane": "schwer"}), encoding="utf-8", ) environment = { "BOARD_INBOX_DIR": str(inbox), "BOARD_PROCESSING_DIR": str(processing), } with mock.patch.dict(os.environ, environment, clear=False): self.assertIn("for this card", trigger.builder_trigger_blocked("same", "schwer")) self.assertIn( "lane 'schwer'", trigger.builder_trigger_blocked("new", "schwer") ) self.assertIsNone(trigger.builder_trigger_blocked("new", "leicht")) def test_builder_trigger_rejects_non_builder_job_on_shared_checkout(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) inbox = root / "inbox" processing = root / "processing" inbox.mkdir() processing.mkdir() (inbox / "research.json").write_text( json.dumps({"job_type": "research", "slug": "research"}), encoding="utf-8", ) environment = { "BOARD_INBOX_DIR": str(inbox), "BOARD_PROCESSING_DIR": str(processing), } with mock.patch.dict(os.environ, environment, clear=False): self.assertIn("already queued or running", trigger.builder_trigger_blocked("new")) def test_enqueue_builder_job_translates_into_agent_codex_queue(self): with tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir: intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n\nDetails.\n", encoding="utf-8") environment = { "BOARD_REPO_DIR": repo_dir, "BOARD_BUILDER_OWNER_JID": "lukas@0ff.space", "AGENT_BUILDER_QUEUE_DIR": queue_dir, } with mock.patch.dict(os.environ, environment, clear=False): job_id = worker.enqueue_builder_job( "my-slug", "docs/intent/my-slug.md", "/var/lib/board-agent/builder-results" ) payload = json.loads((Path(queue_dir) / f"{job_id}.json").read_text(encoding="utf-8")) self.assertEqual(payload["sender"], "lukas@0ff.space") self.assertEqual(payload["requested_by"], "lukas@0ff.space") self.assertIn("Details.", payload["body"]) self.assertIn("my-slug", payload["body"]) self.assertEqual(payload["on_complete_dir"], "/var/lib/board-agent/builder-results") self.assertEqual(payload["board_slug"], "my-slug") self.assertEqual(payload["runtime"], "codex") self.assertTrue(payload["defer_deploy_until_review"]) def test_enqueue_builder_job_forwards_optional_payload_fields(self): with tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir: intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n", encoding="utf-8") environment = { "BOARD_REPO_DIR": repo_dir, "BOARD_BUILDER_OWNER_JID": "lukas@0ff.space", "AGENT_BUILDER_QUEUE_DIR": queue_dir, } with mock.patch.dict(os.environ, environment, clear=False): job_id = worker.enqueue_builder_job( "my-slug", "docs/intent/my-slug.md", "", "codex", "gpt-5.6", "high", 30, ) payload = json.loads((Path(queue_dir) / f"{job_id}.json").read_text(encoding="utf-8")) self.assertEqual( {key: payload[key] for key in ("model", "reasoning_effort", "iteration_limit")}, {"model": "gpt-5.6", "reasoning_effort": "high", "iteration_limit": 30}, ) def test_enqueue_builder_job_omits_none_optional_payload_fields(self): with tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir: intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n", encoding="utf-8") environment = { "BOARD_REPO_DIR": repo_dir, "BOARD_BUILDER_OWNER_JID": "lukas@0ff.space", "AGENT_BUILDER_QUEUE_DIR": queue_dir, } with mock.patch.dict(os.environ, environment, clear=False): job_id = worker.enqueue_builder_job("my-slug", "docs/intent/my-slug.md", "") payload = json.loads((Path(queue_dir) / f"{job_id}.json").read_text(encoding="utf-8")) self.assertNotIn("model", payload) self.assertNotIn("reasoning_effort", payload) self.assertNotIn("iteration_limit", payload) def test_enqueue_builder_job_routes_to_queue_dir_env_override(self): # The "leicht" lane's dedicated inbox -- see LANE_QUEUE_ENV and # docs/intent/board-builder-weighted-lanes.md. with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as queue_dir_leicht, ): intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n", encoding="utf-8") environment = { "BOARD_REPO_DIR": repo_dir, "BOARD_BUILDER_OWNER_JID": "lukas@0ff.space", "AGENT_BUILDER_QUEUE_DIR": queue_dir, "AGENT_BUILDER_QUEUE_DIR_LEICHT": queue_dir_leicht, } with mock.patch.dict(os.environ, environment, clear=False): job_id = worker.enqueue_builder_job( "my-slug", "docs/intent/my-slug.md", "", queue_dir_env="AGENT_BUILDER_QUEUE_DIR_LEICHT", ) self.assertTrue((Path(queue_dir_leicht) / f"{job_id}.json").exists()) self.assertFalse((Path(queue_dir) / f"{job_id}.json").exists()) def test_enqueue_builder_job_rejects_unexpected_intent_path(self): with mock.patch.dict(os.environ, {"BOARD_REPO_DIR": "/tmp"}, clear=False): with self.assertRaises(ValueError): worker.enqueue_builder_job("x", "../../etc/passwd", "") class ProcessJobDispatchTests(unittest.TestCase): def test_builder_job_type_hands_off_without_running_codex(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "slug-1.json" worker.write_payload( job_path, {"job_id": "slug-1", "job_type": "builder", "slug": "slug", "intent_path": "docs/intent/slug.md"}, ) environment = { "BOARD_BUILDER_RESULTS_DIR": "/var/lib/board-agent/builder-results", "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with mock.patch.object(worker, "enqueue_builder_job", return_value="board-slug-1") as enqueue, \ mock.patch.object(worker, "run_review") as run_review, \ mock.patch.dict(os.environ, environment, clear=False): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) enqueue.assert_called_once() run_review.assert_not_called() self.assertTrue((Path(done_dir) / "slug-1.json").exists()) def test_builder_job_type_forwards_optional_payload_fields(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "slug-1.json" worker.write_payload(job_path, { "job_id": "slug-1", "job_type": "builder", "slug": "slug", "intent_path": "docs/intent/slug.md", "model": "gpt-5.6", "reasoning_effort": "high", "iteration_limit": 30, }) environment = { "BOARD_BUILDER_RESULTS_DIR": "/var/lib/board-agent/builder-results", "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with mock.patch.dict(os.environ, environment, clear=False), \ mock.patch.object(worker, "enqueue_builder_job", return_value="board-slug-1") as enqueue: worker.process_job(job_path, Path(failed_dir), Path(done_dir)) enqueue.assert_called_once_with( "slug", "docs/intent/slug.md", "/var/lib/board-agent/builder-results", "codex", "gpt-5.6", "high", 30, queue_dir_env="AGENT_BUILDER_QUEUE_DIR", ) def test_builder_job_type_routes_leicht_lane_to_its_own_queue_dir_env(self): # See docs/intent/board-builder-weighted-lanes.md: a card queued # from the "leicht" column carries lane="leicht" all the way # through to enqueue_builder_job()'s queue_dir_env. with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "slug-1.json" worker.write_payload( job_path, { "job_id": "slug-1", "job_type": "builder", "slug": "slug", "intent_path": "docs/intent/slug.md", "lane": "leicht", }, ) environment = { "BOARD_BUILDER_RESULTS_DIR": "/var/lib/board-agent/builder-results", "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with ( mock.patch.object(worker, "enqueue_builder_job", return_value="board-slug-1") as enqueue, mock.patch.dict(os.environ, environment, clear=False), ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) self.assertEqual(enqueue.call_args.kwargs["queue_dir_env"], "AGENT_BUILDER_QUEUE_DIR_LEICHT") def test_review_job_routes_each_decision_and_invokes_synthesis_after_update(self): for decision in ("bereit_fuer_deployment", "planung_mit_feedback"): with self.subTest(decision=decision): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "slug-2.json" worker.write_payload( job_path, { "job_id": "slug-2", "job_type": "review", "slug": "slug", "commit_hash": "abc1234", }, ) environment = { "BOARD_LOCK_FILE": str(Path(lock_dir) / "lock"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), "AGENT_BUILDER_DEPLOY_QUEUE_DIR": str(Path(state_dir) / "deploy"), "BOARD_REPO_DIR": "/home/drone/nixos-config", "BOARD_BUILDER_RESULTS_DIR": "/var/lib/board-agent/builder-results", } def assert_updated_before_synthesis(slug): self.assertEqual(slug, "slug") self.assertEqual(worker.load_card(slug)["phase"], decision) with mock.patch.dict(os.environ, environment, clear=False), \ mock.patch.object(worker, "enqueue_builder_job") as enqueue, \ mock.patch.object( worker.fanout, "maybe_synthesize", side_effect=assert_updated_before_synthesis, ) as maybe_synthesize, \ mock.patch.object( worker, "run_review", return_value=("Antwort", decision, "docs/review/slug.md"), ) as run_review: worker.process_job(job_path, Path(failed_dir), Path(done_dir)) enqueue.assert_not_called() run_review.assert_called_once_with("slug", "abc1234", "codex", None, None) maybe_synthesize.assert_called_once_with("slug") state = json.loads( (Path(state_dir) / "state.json").read_text(encoding="utf-8") ) self.assertEqual(state["cards"]["slug"]["phase"], decision) self.assertEqual(state["cards"]["slug"]["last_run_status"], "done") self.assertEqual(state["cards"]["slug"]["review_decision"], decision) self.assertEqual( state["cards"]["slug"]["review_report"], "docs/review/slug.md" ) self.assertEqual(state["cards"]["slug"]["commit_hash"], "abc1234") deploy_jobs = list((Path(state_dir) / "deploy").glob("*.json")) if decision == "bereit_fuer_deployment": self.assertEqual(len(deploy_jobs), 1) deploy = json.loads(deploy_jobs[0].read_text(encoding="utf-8")) self.assertEqual(deploy["commit_hash"], "abc1234") self.assertEqual(state["cards"]["slug"]["deployment_status"], "queued") else: self.assertEqual(deploy_jobs, []) def test_research_job_dispatches_its_snapshotted_runtime(self): for runtime in ("codex", "claude", "openrouter"): with ( self.subTest(runtime=runtime), tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "research-slug-1.json" worker.write_payload( job_path, { "job_id": "research-slug-1", "job_type": "research", "slug": "research-slug", "runtime": runtime, "research_intent_path": "docs/research-intent/research-slug.md", }, ) environment = { "BOARD_LOCK_FILE": str(Path(lock_dir) / "lock"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( worker, "run_research", return_value=("Research response", "docs/research/research-slug.md"), ) as run_research, ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) run_research.assert_called_once_with( "research-slug", "docs/research-intent/research-slug.md", runtime, None, None, ) completed_payload = json.loads( (Path(done_dir) / "research-slug-1.json").read_text(encoding="utf-8") ) self.assertEqual(completed_payload["runtime"], runtime) state = json.loads( (Path(state_dir) / "state.json").read_text(encoding="utf-8") ) self.assertEqual(state["cards"]["research-slug"]["phase"], "recherche_fertig") def test_research_job_rejects_invalid_runtime_before_dispatch(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "research-slug-1.json" worker.write_payload( job_path, { "job_id": "research-slug-1", "job_type": "research", "slug": "research-slug", "runtime": "invalid", "research_intent_path": "docs/research-intent/research-slug.md", }, ) environment = {"BOARD_STATE_FILE": str(Path(state_dir) / "state.json")} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker, "run_research") as run_research, ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) run_research.assert_not_called() self.assertTrue((Path(failed_dir) / "research-slug-1.json").exists()) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["research-slug"]["phase"], "recherche_fehlgeschlagen") class FailureHandlerTests(unittest.TestCase): def _env(self, root): return { "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), "BOARD_FAILED_DIR": str(root / "failed"), "BOARD_STATE_FILE": str(root / "state.json"), } def test_unique_candidate_is_failed_and_detail_is_cleaned(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) inbox = root / "inbox" inbox.mkdir() job = inbox / "job.json" worker.write_payload(job, { "job_id": "job-1", "slug": "card", "source": "intent", "intent_path": "docs/intent/card.md", }) (root / "state.json").write_text(json.dumps({"cards": {"card": {"last_job_id": "job-1"}}})) with mock.patch.dict(os.environ, self._env(root) | { "BOARD_FAILURE_STAGE": "pre_start", "BOARD_FAILURE_DETAIL": "bad\n" + "x" * 400, }, clear=False): self.assertEqual(failure_handler.main(), 0) state = json.loads((root / "state.json").read_text()) self.assertEqual(state["cards"]["card"]["last_run_status"], "failed") self.assertEqual(state["cards"]["card"]["failure_stage"], "pre_start") self.assertLessEqual(len(state["cards"]["card"]["failure_detail"]), 240) self.assertTrue((root / "failed" / "job.json").exists()) def test_empty_or_ambiguous_queue_is_left_untouched(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "inbox").mkdir() with mock.patch.dict(os.environ, self._env(root), clear=False): self.assertEqual(failure_handler.main(), 0) for name, slug in (("one", "one"), ("two", "two")): worker.write_payload(root / "inbox" / f"{name}.json", { "job_id": name, "slug": slug, "source": "intent", "intent_path": f"docs/intent/{slug}.md", }) with mock.patch.dict(os.environ, self._env(root), clear=False): self.assertEqual(failure_handler.main(), 0) self.assertEqual(len(list((root / "inbox").glob("*.json"))), 2) def test_late_hook_does_not_overwrite_success(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) inbox = root / "inbox" inbox.mkdir() worker.write_payload(inbox / "job.json", { "job_id": "job-1", "slug": "card", "source": "research", "intent_path": "docs/research-intent/card.md", }) (root / "state.json").write_text(json.dumps({"cards": {"card": { "last_job_id": "job-1", "last_run_status": "done" }}})) with mock.patch.dict(os.environ, self._env(root), clear=False): failure_handler.main() state = json.loads((root / "state.json").read_text()) self.assertEqual(state["cards"]["card"]["last_run_status"], "done") self.assertTrue((root / "inbox" / "job.json").exists()) def test_successful_service_does_not_claim_new_queue_entry(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) inbox = root / "inbox" inbox.mkdir() worker.write_payload(inbox / "next.json", { "job_id": "job-2", "slug": "next", "source": "intent", "intent_path": "docs/intent/next.md", }) (root / "state.json").write_text(json.dumps({"cards": {"next": { "last_job_id": "job-2", "last_run_status": "queued" }}})) with mock.patch.dict(os.environ, self._env(root) | { "SERVICE_RESULT": "success", }, clear=False): self.assertEqual(failure_handler.main(), 0) state = json.loads((root / "state.json").read_text()) self.assertEqual(state["cards"]["next"]["last_run_status"], "queued") self.assertTrue((inbox / "next.json").exists()) def test_default_detail_contains_bounded_systemd_result(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) inbox = root / "inbox" inbox.mkdir() worker.write_payload(inbox / "job.json", { "job_id": "job-1", "slug": "card", "source": "intent", "intent_path": "docs/intent/card.md", }) (root / "state.json").write_text(json.dumps({"cards": {"card": { "last_job_id": "job-1", "last_run_status": "queued" }}})) with mock.patch.dict(os.environ, self._env(root) | { "SERVICE_RESULT": "exit-code", "EXIT_CODE": "exited", "EXIT_STATUS": "1", }, clear=False): self.assertEqual(failure_handler.main(), 0) state = json.loads((root / "state.json").read_text()) self.assertIn("exit-code", state["cards"]["card"]["failure_detail"]) self.assertLessEqual(len(state["cards"]["card"]["failure_detail"]), 240) def test_frontend_exposes_failure_details_as_text(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('card.failure_detail', board_html) self.assertIn('textSection(content,"Failure",card.failure_detail)', board_html) self.assertIn('body.textContent=String(value ?? "Not recorded")', board_html) class ReviewReportExtractionTests(unittest.TestCase): def test_extract_review_report_returns_none_without_markers(self): self.assertIsNone(worker.extract_review_report("Nur eine Antwort.")) def test_extract_review_report_strips_markers(self): response = ( "Vorspann\n" "===REVIEW-REPORT===\n" "Geprüfter Commit: abc1234\n" "Entscheidung: bereit für Deployment\n" "===END-REVIEW-REPORT===\n" "Nachspann" ) report = worker.extract_review_report(response) self.assertIn("Entscheidung: bereit für Deployment", report) self.assertNotIn("===REVIEW-REPORT===", report) self.assertNotIn("Vorspann", report) def test_extract_review_report_falls_back_without_closing_marker(self): response = ( "Vorspann\n" "===REVIEW-REPORT===\n" "Geprüfter Commit: abc1234\n" "Entscheidung: bereit für Deployment" ) report = worker.extract_review_report(response) self.assertIn("Entscheidung: bereit für Deployment", report) self.assertNotIn("===REVIEW-REPORT===", report) self.assertNotIn("Vorspann", report) class CodexResponseParsingTests(unittest.TestCase): def test_tolerates_malformed_intermediate_event_when_final_answer_is_valid(self): output = ( '{"type":"item.completed","item":{"text":"abgeschnitten}\n' '{"type":"item.completed","item":{"type":"agent_message",' '"text":"===REVIEW-REPORT===\\nEntscheidung: bereit für Deployment\\n' '===END-REVIEW-REPORT==="}}\n' ) response = worker.parse_codex_response(output) self.assertIn("Entscheidung: bereit für Deployment", response) def test_rejects_malformed_output_without_final_answer(self): output = '{"type":"item.completed","item":{"text":"abgeschnitten}\n' with self.assertRaisesRegex(RuntimeError, "Ungültige Codex-JSONL-Ausgabe"): worker.parse_codex_response(output) class RunReviewTests(unittest.TestCase): def _environment(self, repo_dir): return { "BOARD_CODEX_BIN": "/bin/codex", "BOARD_CODEX_HOME": "/var/lib/board-agent/codex-home", "BOARD_HOME": "/home/board-agent", "BOARD_REVIEWER_CODEX_MODEL": "gpt-5.6-luna", "BOARD_REVIEWER_CODEX_REASONING_EFFORT": "none", "BOARD_CLAUDE_BIN": "/bin/claude", "BOARD_REVIEWER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/reviewer-settings.json", "BOARD_REVIEWER_CLAUDE_MODEL": "opus", "BOARD_REVIEWER_CLAUDE_REASONING_EFFORT": "high", "BOARD_ROLE_DIR": f"{repo_dir}/agents", "BOARD_REPO_DIR": repo_dir, "BOARD_SCREENSHOT_SCRIPT": "/repo/board-agent/screenshot.py", } def _claude_result(self, report_body): decision = "reject" if any(word in report_body.lower() for word in ("zurück", "back to planning", "return to planning")) else "approve" verdict = "\n===REVIEW-RESULT===\n" + json.dumps({"schema_version": 1, "commit_hash": "abc1234", "decision": decision}) + "\n===END-REVIEW-RESULT===" stdout = json.dumps( { "type": "result", "subtype": "success", "is_error": False, "result": f"===REVIEW-REPORT===\n{report_body}\n===END-REVIEW-REPORT===" + verdict, "session_id": "session-1", } ) return SimpleNamespace(returncode=0, stdout=stdout, stderr="") def test_run_review_writes_report_and_returns_ready_for_deployment_decision(self): with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("Entscheidung: bereit für Deployment"), ), ): response, decision, report_path = worker.run_review( "my-slug", "abc1234", "claude" ) self.assertEqual(decision, "bereit_fuer_deployment") self.assertEqual(report_path, "docs/review/my-slug.md") written = (Path(repo_dir) / "docs" / "review" / "my-slug.md").read_text(encoding="utf-8") self.assertIn("bereit für Deployment", written) def test_run_review_matches_decision_regardless_of_case(self): # Regression test for the 2026-09-09 live failure of # operator-smoke-split-docs: the Reviewer put the decision on its # own line after a bold "**Entscheidung:**" heading rather than # inline after "Entscheidung: " -- capitalizing "Bereit" the way # any German sentence/heading start would, even though # docs/review/operator-smoke-split-docs.md's content was otherwise # exactly right. with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("**Entscheidung:**\nBereit für Deployment"), ), ): _response, decision, _report_path = worker.run_review( "my-slug", "abc1234", "claude" ) self.assertEqual(decision, "bereit_fuer_deployment") def test_run_review_matches_an_english_decision_rendering(self): # Regression test for the 2026-09-09 live failure of # operator-smoke-split-agent-codex-2: the Reviewer wrote an # otherwise-correct report entirely in English, closing with # "Decision: ready for deployment" instead of the German phrase # reviewer.md asks for. with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("Decision: ready for deployment"), ), ): _response, decision, _report_path = worker.run_review( "my-slug", "abc1234", "claude" ) self.assertEqual(decision, "bereit_fuer_deployment") def test_run_review_returns_planung_mit_feedback_decision(self): with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("Entscheidung: zurück zur Planung"), ), ): _response, decision, _report_path = worker.run_review( "my-slug", "abc1234", "claude" ) self.assertEqual(decision, "planung_mit_feedback") def test_run_review_uses_disallowed_tools_and_add_dir(self): with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("Entscheidung: bereit für Deployment"), ) as run, ): worker.run_review("my-slug", "abc1234", "claude") command = run.call_args.args[0] # CLI-level pre-grant for read-only git access -- see # run_review()'s comment: the settings.json permission system # intermittently denied Bash entirely under dontAsk, live # 2026-09-09, even for commands its own allow list covered. # Also pins the exact resolved screenshot.py invocation (see # docs/intent/board-reviewer-visual-verification.md) so the # model can only ever run that one fixed command via Bash. allowed_tools = command[command.index("--allowedTools") + 1] self.assertIn("Bash(git *)", allowed_tools) self.assertIn(f"Bash({sys.executable} /repo/board-agent/screenshot.py abc1234 *)", allowed_tools) self.assertEqual( command[command.index("--disallowed-tools") + 1], "Edit,Write,MultiEdit,NotebookEdit", ) # Unlike Codex's --sandbox read-only, Claude needs --add-dir to # even reach repoDir from cwd=BOARD_ROLE_DIR (agents/, a # subdirectory of repoDir) at all. self.assertEqual(command[command.index("--add-dir") + 1], repo_dir) self.assertEqual(run.call_args.kwargs["cwd"], f"{repo_dir}/agents") def test_claude_review_prompt_includes_visual_verification_instructions(self): with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("Entscheidung: bereit für Deployment"), ) as run, ): worker.run_review("my-slug", "abc1234", "claude") prompt = run.call_args.kwargs["input"] self.assertIn("Screenshot-Werkzeug", prompt) self.assertIn(f"{sys.executable} /repo/board-agent/screenshot.py abc1234", prompt) def test_codex_review_prompt_omits_visual_verification_instructions(self): # docs/intent/board-reviewer-visual-verification.md: Codex's # --sandbox read-only has nowhere to write a screenshot (verified, # see this directory's README), so the tool is never offered there. with tempfile.TemporaryDirectory() as repo_dir: codex_result = SimpleNamespace( returncode=0, stdout=( '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps('===REVIEW-REPORT===\nEntscheidung: bereit für Deployment\n===END-REVIEW-REPORT===\n===REVIEW-RESULT===\n{"schema_version":1,"commit_hash":"abc1234","decision":"approve"}\n===END-REVIEW-RESULT===') + "}}\n" ), stderr="", ) with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=codex_result) as run, ): worker.run_review("my-slug", "abc1234", "codex") command = run.call_args.args[0] self.assertNotIn("Screenshot-Werkzeug", command[-1]) def test_run_review_uses_luna_none_and_read_only_for_codex(self): with tempfile.TemporaryDirectory() as repo_dir: codex_result = SimpleNamespace( returncode=0, stdout=( '{"type":"thread.started","thread_id":"session-1"}\n' '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps( "===REVIEW-REPORT===\nEntscheidung: bereit für Deployment\n" "===END-REVIEW-REPORT===\n" '===REVIEW-RESULT===\n{"schema_version":1,"commit_hash":"abc1234","decision":"approve"}\n===END-REVIEW-RESULT===' ) + "}}\n" ), stderr="", ) with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=codex_result) as run, ): worker.run_review("my-slug", "abc1234", "codex") command = run.call_args.args[0] self.assertEqual(command[command.index("--sandbox") + 1], "read-only") self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-luna") reasoning = next(value for value in command if value.startswith("model_reasoning_effort=")) self.assertEqual(reasoning, 'model_reasoning_effort="none"') self.assertEqual(run.call_args.kwargs["env"]["CODEX_HOME"], "/var/lib/board-agent/codex-home") def test_run_review_uses_explicit_codex_model_options(self): with tempfile.TemporaryDirectory() as repo_dir: codex_result = SimpleNamespace( returncode=0, stdout=( '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps( "===REVIEW-REPORT===\nEntscheidung: bereit für Deployment\n" "===END-REVIEW-REPORT===\n" '===REVIEW-RESULT===\n{"schema_version":1,"commit_hash":"abc1234","decision":"approve"}\n===END-REVIEW-RESULT===' ) + '}}\n' ), stderr="", ) with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=codex_result) as run, ): worker.run_review("my-slug", "abc1234", "codex", "gpt-explicit", "high") command = run.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "gpt-explicit") self.assertIn('model_reasoning_effort="high"', command) def test_run_review_uses_openrouter_default_and_child_only_secret(self): with tempfile.TemporaryDirectory() as repo_dir: secret_path = Path(repo_dir) / "openrouter-key" secret_path.write_text("review-secret\n", encoding="utf-8") environment = self._environment(repo_dir) | { "BOARD_OPENROUTER_API_KEY_FILE": str(secret_path), "BOARD_REVIEWER_OPENROUTER_MODEL": "deepseek/deepseek-v3.2", "BOARD_REVIEWER_OPENROUTER_REASONING_EFFORT": "", } codex_result = SimpleNamespace( returncode=0, stdout=( '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps( "===REVIEW-REPORT===\nEntscheidung: bereit für Deployment\n" "===END-REVIEW-REPORT===\n" '===REVIEW-RESULT===\n{"schema_version":1,"commit_hash":"abc1234","decision":"approve"}\n===END-REVIEW-RESULT===' ) + "}}\n" ), stderr="", ) with ( mock.patch.dict(os.environ, environment, clear=True), mock.patch.object(worker.subprocess, "run", return_value=codex_result) as run, ): worker.run_review("my-slug", "abc1234", "openrouter", "") self.assertNotIn("OPENROUTER_API_KEY", os.environ) command = run.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "deepseek/deepseek-v3.2") self.assertIn('model_provider="openrouter"', command) self.assertEqual(run.call_args.kwargs["env"]["OPENROUTER_API_KEY"], "review-secret") self.assertNotIn("review-secret", " ".join(command)) def test_run_review_uses_explicit_claude_model_options(self): with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=self._claude_result("Entscheidung: bereit für Deployment"), ) as run, ): worker.run_review("my-slug", "abc1234", "claude", "claude-explicit", "low") command = run.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "claude-explicit") self.assertEqual(command[command.index("--effort") + 1], "low") def test_run_review_raises_without_a_decision_literal(self): with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object( worker.subprocess, "run", return_value=SimpleNamespace(returncode=0, stderr="", stdout=json.dumps({"result": "===REVIEW-REPORT===\nKein klares Ergebnis.\n===END-REVIEW-REPORT==="})), ), ): with self.assertRaises(RuntimeError): worker.run_review("my-slug", "abc1234", "claude") class ScreenshotScriptTests(unittest.TestCase): """docs/intent/board-reviewer-visual-verification.md: screenshot.py's own argument validation -- the --allowedTools grant only pins the script path, so the script must reject a bad argument on its own rather than trusting the Reviewer's allowlisted-but-still-arbitrary trailing arguments. These exercise only the fast-fail paths (before any git/chromium subprocess call), so they run without either dependency installed. """ SCRIPT = HERE / "screenshot.py" def _run(self, *args, env_extra=None): env = {**os.environ, "BOARD_REPO_DIR": "/nonexistent"} env.update(env_extra or {}) return subprocess.run( [sys.executable, str(self.SCRIPT), *args], capture_output=True, text=True, timeout=10, env=env, check=False, ) def test_rejects_wrong_argument_count(self): result = self._run("only-one-arg") self.assertEqual(result.returncode, 2) self.assertIn("usage:", result.stderr) def test_rejects_non_hex_commit_hash(self): result = self._run("not-a-hash", "page.html") self.assertEqual(result.returncode, 2) self.assertIn("commit hash", result.stderr) def test_rejects_non_html_path(self): result = self._run("abc1234", "notes.txt") self.assertEqual(result.returncode, 2) self.assertIn("repo-relative .html path", result.stderr) def test_rejects_absolute_path(self): result = self._run("abc1234", "/etc/passwd.html") self.assertEqual(result.returncode, 2) self.assertIn("repo-relative .html path", result.stderr) def test_rejects_path_traversal(self): result = self._run("abc1234", "../../etc/page.html") self.assertEqual(result.returncode, 2) self.assertIn("repo-relative .html path", result.stderr) class RunResearchTests(unittest.TestCase): def _environment(self, repo_dir): return { "BOARD_CODEX_BIN": "/bin/codex", "BOARD_CODEX_HOME": "/var/lib/board-agent/codex-home", "BOARD_HOME": "/home/board-agent", "BOARD_RESEARCHER_CODEX_MODEL": "gpt-5.6-terra", "BOARD_RESEARCHER_CODEX_REASONING_EFFORT": "medium", "BOARD_CLAUDE_BIN": "/bin/claude", "BOARD_RESEARCHER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/researcher-settings.json", "BOARD_RESEARCHER_CLAUDE_MODEL": "researcher-sonnet", "BOARD_RESEARCHER_CLAUDE_REASONING_EFFORT": "medium", "BOARD_REVIEWER_CLAUDE_MODEL": "sonnet", "BOARD_REVIEWER_CLAUDE_REASONING_EFFORT": "low", "BOARD_ROLE_DIR": f"{repo_dir}/agents", "BOARD_REPO_DIR": repo_dir, } def _research_intent(self, repo_dir): intent_dir = Path(repo_dir) / "docs" / "research-intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# Research\n\nQuestion\n", encoding="utf-8") def _claude_result(self): return SimpleNamespace( returncode=0, stdout=json.dumps( { "type": "result", "subtype": "success", "is_error": False, "result": "===RESEARCH-REPORT===\nResearch body\n===END-RESEARCH-REPORT===", } ), stderr="", ) def _codex_result(self): return SimpleNamespace( returncode=0, stdout=( '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps("===RESEARCH-REPORT===\nResearch body\n===END-RESEARCH-REPORT===") + "}}\n" ), stderr="", ) def test_run_research_uses_read_only_codex_and_writes_report(self): with tempfile.TemporaryDirectory() as repo_dir: self._research_intent(repo_dir) with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result()) as run, ): _response, report_path = worker.run_research( "my-slug", "docs/research-intent/my-slug.md", "codex" ) command = run.call_args.args[0] self.assertEqual(command[0], "/bin/codex") self.assertEqual(command[command.index("--sandbox") + 1], "read-only") self.assertEqual(command[command.index("-C") + 1], f"{repo_dir}/agents") self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-terra") self.assertEqual(run.call_args.kwargs["cwd"], None) self.assertEqual(run.call_args.kwargs["env"]["CODEX_HOME"], "/var/lib/board-agent/codex-home") self.assertEqual(report_path, "docs/research/my-slug.md") self.assertEqual( (Path(repo_dir) / report_path).read_text(encoding="utf-8"), "Research body\n" ) def test_run_research_uses_researcher_claude_settings_and_writes_report(self): with tempfile.TemporaryDirectory() as repo_dir: self._research_intent(repo_dir) with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._claude_result()) as run, ): _response, report_path = worker.run_research( "my-slug", "docs/research-intent/my-slug.md", "claude" ) command = run.call_args.args[0] self.assertEqual(command[0], "/bin/claude") self.assertEqual( command[command.index("--settings") + 1], "/repo/agents/claude-settings/researcher-settings.json", ) self.assertEqual(command[command.index("--add-dir") + 1], repo_dir) self.assertEqual(command[command.index("--model") + 1], "researcher-sonnet") self.assertEqual(command[command.index("--effort") + 1], "medium") self.assertEqual(run.call_args.kwargs["cwd"], f"{repo_dir}/agents") self.assertEqual(report_path, "docs/research/my-slug.md") self.assertEqual( (Path(repo_dir) / report_path).read_text(encoding="utf-8"), "Research body\n" ) def test_run_research_uses_explicit_model_options(self): with tempfile.TemporaryDirectory() as repo_dir: self._research_intent(repo_dir) environment = self._environment(repo_dir) environment.update({ "BOARD_RESEARCHER_CODEX_MODEL": "env-model", "BOARD_RESEARCHER_CODEX_REASONING_EFFORT": "env-effort", }) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result()) as run, ): worker.run_research( "my-slug", "docs/research-intent/my-slug.md", "codex", "explicit-model", "explicit-effort" ) command = run.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "explicit-model") self.assertIn("model_reasoning_effort=\"explicit-effort\"", command) def test_run_research_uses_explicit_claude_options(self): with tempfile.TemporaryDirectory() as repo_dir: self._research_intent(repo_dir) with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._claude_result()) as run, ): worker.run_research( "my-slug", "docs/research-intent/my-slug.md", "claude", "explicit-model", "explicit-effort" ) command = run.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "explicit-model") self.assertEqual(command[command.index("--effort") + 1], "explicit-effort") def test_run_research_uses_explicit_openrouter_model(self): with tempfile.TemporaryDirectory() as repo_dir: self._research_intent(repo_dir) secret_path = Path(repo_dir) / "openrouter-key" secret_path.write_text("research-secret\n", encoding="utf-8") environment = self._environment(repo_dir) | { "BOARD_OPENROUTER_API_KEY_FILE": str(secret_path), "BOARD_RESEARCHER_OPENROUTER_MODEL": "deepseek/deepseek-v3.2", "BOARD_RESEARCHER_OPENROUTER_REASONING_EFFORT": "", } with ( mock.patch.dict(os.environ, environment, clear=True), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result()) as run, ): worker.run_research( "my-slug", "docs/research-intent/my-slug.md", "openrouter", "moonshotai/kimi-k2-thinking", ) command = run.call_args.args[0] self.assertEqual( command[command.index("--model") + 1], "moonshotai/kimi-k2-thinking", ) self.assertEqual(run.call_args.kwargs["env"]["OPENROUTER_API_KEY"], "research-secret") class SplitChildExtractionTests(unittest.TestCase): def _split_report(self, children): blocks = "\n".join( f"===CHILD===\nslug: {slug}\n---\n{content}\n===END-CHILD===" for slug, content in children ) return f"===SPLIT-REPORT===\n{blocks}\n===END-SPLIT-REPORT===" def test_parses_multiple_children_in_order(self): response = self._split_report( [("child-one", "# Intent: One\n\nBody one."), ("child-two", "# Intent: Two\n\nBody two.")] ) children = worker.extract_split_children(response, "parent-slug") self.assertEqual([slug for slug, _content in children], ["child-one", "child-two"]) self.assertIn("Body one.", children[0][1]) def test_raises_without_outer_markers(self): with self.assertRaises(ValueError): worker.extract_split_children("Nur eine Antwort ohne Marker.", "parent-slug") def test_raises_without_any_child_block(self): with self.assertRaises(ValueError): worker.extract_split_children("===SPLIT-REPORT===\n===END-SPLIT-REPORT===", "parent-slug") def test_raises_on_more_than_ten_children(self): response = self._split_report([(f"child-{i}", f"Body {i}") for i in range(11)]) with self.assertRaises(ValueError): worker.extract_split_children(response, "parent-slug") def test_raises_on_duplicate_child_slug(self): response = self._split_report([("child-one", "Body a"), ("child-one", "Body b")]) with self.assertRaises(ValueError): worker.extract_split_children(response, "parent-slug") def test_raises_when_child_slug_equals_parent_slug(self): response = self._split_report([("parent-slug", "Body")]) with self.assertRaises(ValueError): worker.extract_split_children(response, "parent-slug") def test_raises_on_malformed_child_header(self): response = "===SPLIT-REPORT===\n===CHILD===\nkein gueltiger header\n===END-CHILD===\n===END-SPLIT-REPORT===" with self.assertRaises(ValueError): worker.extract_split_children(response, "parent-slug") class DraftResponseExtractionTests(unittest.TestCase): """docs/intent/board-interactive-planner-intake.md: the interactive Planner's per-turn reply is either a follow-up question or a finished intent, parsed by worker.extract_draft_response().""" def test_parses_question_with_options(self): response = ( "===DRAFT-QUESTION===\n" "Welche Hosts sind betroffen?\n" "optionen: tnix | pinix | blunix | alle\n" "===END-DRAFT-QUESTION===" ) parsed = worker.extract_draft_response(response) self.assertEqual(parsed, { "kind": "question", "text": "Welche Hosts sind betroffen?", "options": ["tnix", "pinix", "blunix", "alle"], }) def test_parses_question_with_no_options(self): response = ( "===DRAFT-QUESTION===\nWas ist das Ziel?\noptionen: keine\n===END-DRAFT-QUESTION===" ) parsed = worker.extract_draft_response(response) self.assertEqual(parsed["options"], []) def test_caps_options_at_four(self): response = ( "===DRAFT-QUESTION===\nWelche?\noptionen: a | b | c | d | e\n===END-DRAFT-QUESTION===" ) parsed = worker.extract_draft_response(response) self.assertEqual(parsed["options"], ["a", "b", "c", "d"]) def test_parses_finished_intent(self): response = ( "===DRAFT-INTENT===\n" "slug: install-foo\n" "---\n" "# Intent: Foo installieren\n\nKlarer Auftrag.\n" "===END-DRAFT-INTENT===" ) parsed = worker.extract_draft_response(response) self.assertEqual(parsed["kind"], "intent") self.assertEqual(parsed["slug"], "install-foo") self.assertIn("Foo installieren", parsed["body"]) def test_raises_when_neither_block_present(self): with self.assertRaises(ValueError): worker.extract_draft_response("Nur Fließtext, keine Marker.") def test_raises_when_both_blocks_present(self): response = ( "===DRAFT-QUESTION===\nFrage?\noptionen: keine\n===END-DRAFT-QUESTION===\n" "===DRAFT-INTENT===\nslug: x\n---\nBody\n===END-DRAFT-INTENT===" ) with self.assertRaises(ValueError): worker.extract_draft_response(response) def test_raises_on_malformed_intent_header(self): response = "===DRAFT-INTENT===\nkein gueltiger header\n===END-DRAFT-INTENT===" with self.assertRaises(ValueError): worker.extract_draft_response(response) def test_format_draft_conversation_labels_speakers(self): conversation = [ {"role": "user", "text": "Ripgrep entfernen"}, {"role": "assistant", "text": "Von welchem Host?"}, ] formatted = worker.format_draft_conversation(conversation) self.assertEqual(formatted, "Operator: Ripgrep entfernen\nPlanner: Von welchem Host?") def test_format_draft_conversation_empty(self): self.assertEqual(worker.format_draft_conversation([]), "(noch keine Konversation)") class RunSplitTests(unittest.TestCase): def _environment(self, repo_dir): return { "BOARD_CODEX_BIN": "/bin/codex", "BOARD_CODEX_HOME": "/var/lib/board-agent/codex-home", "BOARD_HOME": "/home/board-agent", "BOARD_PLANNER_CODEX_MODEL": "gpt-5.6-terra", "BOARD_PLANNER_CODEX_REASONING_EFFORT": "high", "BOARD_CLAUDE_BIN": "/bin/claude", "BOARD_PLANNER_CLAUDE_SETTINGS": "/repo/agents/claude-settings/planner-settings.json", "BOARD_PLANNER_CLAUDE_MODEL": "sonnet", "BOARD_PLANNER_CLAUDE_REASONING_EFFORT": "high", "BOARD_ROLE_DIR": f"{repo_dir}/agents", "BOARD_REPO_DIR": repo_dir, } def _intent(self, repo_dir): intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "parent-slug.md").write_text("# Intent: Parent\n\nToo big.\n", encoding="utf-8") return intent_dir def _codex_result(self, response_body): return SimpleNamespace( returncode=0, stdout=( '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps(response_body) + "}}\n" ), stderr="", ) def test_run_split_writes_child_files_and_records_state(self): with tempfile.TemporaryDirectory() as repo_dir: self._intent(repo_dir) state_path = Path(repo_dir) / "state.json" response = ( "===SPLIT-REPORT===\n" "===CHILD===\n" "slug: child-one\n" "---\n" "# Intent: Child One\n\nBody one.\n" "===END-CHILD===\n" "===CHILD===\n" "slug: child-two\n" "---\n" "# Intent: Child Two\n\nBody two.\n" "===END-CHILD===\n" "===END-SPLIT-REPORT===" ) env = self._environment(repo_dir) | {"BOARD_STATE_FILE": str(state_path)} with ( mock.patch.dict(os.environ, env, clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result(response)) as run, ): _response, child_slugs = worker.run_split( "parent-slug", "docs/intent/parent-slug.md", "Zwischenstand ...", "codex" ) self.assertEqual(child_slugs, ["child-one", "child-two"]) command = run.call_args.args[0] self.assertEqual(command[command.index("--sandbox") + 1], "read-only") self.assertEqual(command[command.index("--model") + 1], "gpt-5.6-terra") for slug in child_slugs: written = (Path(repo_dir) / "docs" / "intent" / f"{slug}.md").read_text(encoding="utf-8") self.assertIn("Intent:", written) state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(state["cards"]["child-one"]["parent_id"], "parent-slug") self.assertEqual(state["cards"]["child-one"]["phase"], "geplant") self.assertEqual(state["cards"]["child-two"]["parent_id"], "parent-slug") def test_run_split_writes_nothing_when_a_child_slug_already_has_a_file(self): with tempfile.TemporaryDirectory() as repo_dir: intent_dir = self._intent(repo_dir) (intent_dir / "child-one.md").write_text("# Existing\n", encoding="utf-8") state_path = Path(repo_dir) / "state.json" response = ( "===SPLIT-REPORT===\n" "===CHILD===\n" "slug: child-one\n" "---\n" "# Intent: Child One\n\nBody.\n" "===END-CHILD===\n" "===END-SPLIT-REPORT===" ) env = self._environment(repo_dir) | {"BOARD_STATE_FILE": str(state_path)} with ( mock.patch.dict(os.environ, env, clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result(response)), ): with self.assertRaises(RuntimeError): worker.run_split("parent-slug", "docs/intent/parent-slug.md", "", "codex") self.assertEqual( (intent_dir / "child-one.md").read_text(encoding="utf-8"), "# Existing\n" ) self.assertFalse(state_path.exists()) def test_run_split_uses_openrouter_planner_default(self): with tempfile.TemporaryDirectory() as repo_dir: self._intent(repo_dir) state_path = Path(repo_dir) / "state.json" secret_path = Path(repo_dir) / "openrouter-key" secret_path.write_text("planner-secret\n", encoding="utf-8") response = ( "===SPLIT-REPORT===\n===CHILD===\nslug: child-one\n---\n" "# Intent: Child One\n\nBody.\n===END-CHILD===\n" "===END-SPLIT-REPORT===" ) environment = self._environment(repo_dir) | { "BOARD_STATE_FILE": str(state_path), "BOARD_OPENROUTER_API_KEY_FILE": str(secret_path), "BOARD_PLANNER_OPENROUTER_MODEL": "deepseek/deepseek-v3.2", "BOARD_PLANNER_OPENROUTER_REASONING_EFFORT": "", } with ( mock.patch.dict(os.environ, environment, clear=True), mock.patch.object( worker.subprocess, "run", return_value=self._codex_result(response) ) as run, ): _response, children = worker.run_split( "parent-slug", "docs/intent/parent-slug.md", "Zwischenstand", "openrouter", "", ) self.assertEqual(children, ["child-one"]) command = run.call_args.args[0] self.assertEqual(command[command.index("--model") + 1], "deepseek/deepseek-v3.2") self.assertEqual(run.call_args.kwargs["env"]["OPENROUTER_API_KEY"], "planner-secret") class RunDraftTests(unittest.TestCase): """run_draft() never touches the repo checkout -- no BOARD_REPO_DIR, no --add-dir, no intent file to set up, unlike RunSplitTests above.""" def _environment(self, role_dir): return { "BOARD_CODEX_BIN": "/bin/codex", "BOARD_CODEX_HOME": "/var/lib/board-agent/codex-home", "BOARD_HOME": "/home/board-agent", "BOARD_PLANNER_CODEX_MODEL": "gpt-5.6-terra", "BOARD_PLANNER_CODEX_REASONING_EFFORT": "high", "BOARD_ROLE_DIR": role_dir, } def _codex_result(self, text): return SimpleNamespace( returncode=0, stdout='{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps(text) + "}}\n", stderr="", ) def test_run_draft_returns_parsed_question(self): with tempfile.TemporaryDirectory() as role_dir: response = "===DRAFT-QUESTION===\nWas ist das Ziel?\noptionen: keine\n===END-DRAFT-QUESTION===" with ( mock.patch.dict(os.environ, self._environment(role_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result(response)) as run, ): raw, parsed = worker.run_draft("draft-1", [{"role": "user", "text": "Ripgrep entfernen"}], "codex") self.assertEqual(parsed, {"kind": "question", "text": "Was ist das Ziel?", "options": []}) self.assertEqual(raw, response) command = run.call_args.args[0] self.assertEqual(command[command.index("--sandbox") + 1], "read-only") self.assertNotIn("--add-dir", command) def test_run_draft_returns_parsed_intent(self): with tempfile.TemporaryDirectory() as role_dir: response = ( "===DRAFT-INTENT===\nslug: install-foo\n---\n" "# Intent: Foo installieren\n\nKlar.\n===END-DRAFT-INTENT===" ) with ( mock.patch.dict(os.environ, self._environment(role_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result(response)), ): _raw, parsed = worker.run_draft("draft-1", [], "codex") self.assertEqual(parsed["kind"], "intent") self.assertEqual(parsed["slug"], "install-foo") def test_run_draft_raises_on_malformed_response(self): with tempfile.TemporaryDirectory() as role_dir: with ( mock.patch.dict(os.environ, self._environment(role_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=self._codex_result("Fließtext ohne Marker.")), ): with self.assertRaises(ValueError): worker.run_draft("draft-1", [], "codex") class ProcessJobSplitDispatchTests(unittest.TestCase): def test_split_job_type_writes_children_and_sets_terminal_phase(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "split-1.json" worker.write_payload( job_path, { "job_id": "split-1", "job_type": "split", "slug": "parent-slug", "intent_path": "docs/intent/parent-slug.md", "builder_report": "Zwischenstand ...", }, ) environment = { "BOARD_LOCK_FILE": str(Path(lock_dir) / "lock"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( worker, "run_split", return_value=("Antwort", ["child-one", "child-two"]) ) as run_split, ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) run_split.assert_called_once_with( "parent-slug", "docs/intent/parent-slug.md", "Zwischenstand ...", "codex", None, None, ) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["parent-slug"]["phase"], "aufgeteilt") self.assertEqual(state["cards"]["parent-slug"]["last_run_status"], "done") self.assertEqual(state["cards"]["parent-slug"]["depends_on"], ["child-one", "child-two"]) self.assertEqual(state["cards"]["parent-slug"]["child_slugs"], ["child-one", "child-two"]) def test_split_job_failure_leaves_phase_untouched(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = Path(processing_dir) / "split-1.json" worker.write_payload( job_path, { "job_id": "split-1", "job_type": "split", "slug": "parent-slug", "intent_path": "docs/intent/parent-slug.md", }, ) environment = { "BOARD_LOCK_FILE": str(Path(lock_dir) / "lock"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with mock.patch.dict(os.environ, environment, clear=False): worker.set_phase("parent-slug", "aufteilen_angefordert") with mock.patch.object(worker, "run_split", side_effect=RuntimeError("kaputt")): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["parent-slug"]["phase"], "aufteilen_angefordert") self.assertEqual(state["cards"]["parent-slug"]["last_run_status"], "failed") class ProcessJobDraftDispatchTests(unittest.TestCase): """docs/intent/board-interactive-planner-intake.md: job_type == "draft" either records a follow-up question (no repo touched at all) or, once the Planner is done, writes the new intent file and replaces the draft card's own state entry -- see worker.run_draft()/record_draft_intent().""" def _job(self, processing_dir, conversation): job_path = Path(processing_dir) / "draft-1.json" worker.write_payload( job_path, {"job_id": "draft-1", "job_type": "draft", "slug": "draft-abc123", "conversation": conversation}, ) return job_path def test_question_turn_appends_conversation_and_sets_phase(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as state_dir, ): conversation = [{"role": "user", "text": "Ripgrep entfernen"}] job_path = self._job(processing_dir, conversation) environment = {"BOARD_STATE_FILE": str(Path(state_dir) / "state.json")} parsed = {"kind": "question", "text": "Von welchem Host?", "options": ["blunix", "pinix"]} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker, "run_draft", return_value=("Antwort", parsed)) as run_draft, ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) run_draft.assert_called_once_with("draft-abc123", conversation, "codex", None, None, telemetry_context=None) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) card = state["cards"]["draft-abc123"] self.assertEqual(card["phase"], "entwurf_frage") self.assertEqual(card["draft_conversation"][-1], { "role": "assistant", "text": "Von welchem Host?", "options": ["blunix", "pinix"], }) self.assertTrue((Path(done_dir) / "draft-1.json").exists()) def test_finished_intent_writes_file_and_replaces_draft_card(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, tempfile.TemporaryDirectory() as repo_dir, ): conversation = [{"role": "user", "text": "Ripgrep von blunix entfernen"}] job_path = self._job(processing_dir, conversation) environment = { "BOARD_LOCK_FILE": str(Path(lock_dir) / "lock"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), "BOARD_REPO_DIR": repo_dir, } parsed = {"kind": "intent", "slug": "remove-ripgrep-blunix", "body": "# Intent: Ripgrep entfernen\n\nKlar.\n"} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker, "run_draft", return_value=("Antwort", parsed)), ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) written = (Path(repo_dir) / "docs" / "intent" / "remove-ripgrep-blunix.md").read_text(encoding="utf-8") self.assertIn("Ripgrep entfernen", written) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertNotIn("draft-abc123", state["cards"]) new_card = state["cards"]["remove-ripgrep-blunix"] self.assertEqual(new_card["phase"], "geplant") self.assertEqual(new_card["draft_origin"], "draft-abc123") def test_finished_intent_slug_collision_fails_without_overwriting(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as lock_dir, tempfile.TemporaryDirectory() as state_dir, tempfile.TemporaryDirectory() as repo_dir, ): intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "remove-ripgrep-blunix.md").write_text("# Existing\n", encoding="utf-8") job_path = self._job(processing_dir, []) environment = { "BOARD_LOCK_FILE": str(Path(lock_dir) / "lock"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), "BOARD_REPO_DIR": repo_dir, } parsed = {"kind": "intent", "slug": "remove-ripgrep-blunix", "body": "# Intent: Neu\n\nKlar.\n"} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker, "run_draft", return_value=("Antwort", parsed)), ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) self.assertEqual( (intent_dir / "remove-ripgrep-blunix.md").read_text(encoding="utf-8"), "# Existing\n" ) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["draft-abc123"]["phase"], "entwurf_fehlgeschlagen") self.assertTrue((Path(failed_dir) / "draft-1.json").exists()) def test_runtime_error_sets_failed_phase(self): with ( tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as state_dir, ): job_path = self._job(processing_dir, []) environment = {"BOARD_STATE_FILE": str(Path(state_dir) / "state.json")} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(worker, "run_draft", side_effect=RuntimeError("kaputt")), ): worker.process_job(job_path, Path(failed_dir), Path(done_dir)) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["draft-abc123"]["phase"], "entwurf_fehlgeschlagen") self.assertEqual(state["cards"]["draft-abc123"]["last_run_status"], "failed") class FanoutTests(unittest.TestCase): def _env(self, root): return { "BOARD_STATE_FILE": str(root / "state.json"), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(root / "docs" / "research-intent"), } def test_child_terminal_builder_phases(self): for phase in ("bereit_fuer_deployment", "planung_mit_feedback", "deployed", "archiviert"): with self.subTest(phase=phase): self.assertTrue( fanout.child_terminal("child", {"source": "intent", "phase": phase}) ) self.assertFalse(fanout.child_terminal("child", {"source": "intent", "phase": "review"})) self.assertFalse(fanout.child_terminal("child", {"source": "intent", "phase": "in_umsetzung"})) def test_child_terminal_builder_stalled_status_counts_as_terminal(self): for status in ("no_changes", "failed", "check_failed"): with self.subTest(status=status): self.assertTrue( fanout.child_terminal( "child", {"source": "intent", "phase": "in_umsetzung", "last_run_status": status} ) ) def test_child_terminal_limit_reached_only_after_retry_spent(self): stalled = {"source": "intent", "phase": "in_umsetzung", "last_run_status": "limit_reached"} self.assertFalse(fanout.child_terminal("child", stalled)) stalled["auto_retry_attempts"] = 1 self.assertTrue(fanout.child_terminal("child", stalled)) def test_child_terminal_limit_reached_waits_for_full_escalation_ladder(self): # docs/intent/board-operator-escalation-ladder.md: a card carrying # its own escalation_attempts ladder isn't terminal just because # the flat fallback threshold (1) was reached -- only once every # ladder step is spent. stalled = { "source": "intent", "phase": "in_umsetzung", "last_run_status": "limit_reached", "auto_retry_attempts": 1, "escalation_attempts": [ {"model": "opus", "reasoning_effort": "high", "iteration_limit": 30}, {"model": "opus", "reasoning_effort": "high", "iteration_limit": 40}, ], } self.assertFalse(fanout.child_terminal("child", stalled)) stalled["auto_retry_attempts"] = 2 self.assertTrue(fanout.child_terminal("child", stalled)) def test_escalation_ladder_for_falls_back_to_current_params_without_operator_ladder(self): card = {"model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20} self.assertEqual( fanout.escalation_ladder_for(card), [{"model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20}], ) def test_escalation_ladder_for_uses_operator_authored_ladder_when_present(self): ladder = [{"model": "opus", "reasoning_effort": "high", "iteration_limit": 30}] card = {"model": "gpt-5.6-luna", "escalation_attempts": ladder} self.assertEqual(fanout.escalation_ladder_for(card), ladder) def test_child_terminal_research_phases(self): for phase in ("recherche_fertig", "recherche_fehlgeschlagen", "archiviert"): with self.subTest(phase=phase): self.assertTrue( fanout.child_terminal("child", {"source": "research", "phase": phase}) ) self.assertFalse( fanout.child_terminal("child", {"source": "research", "phase": "recherche_laeuft"}) ) def test_link_child_writes_parent_id_and_depends_on(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) with mock.patch.dict(os.environ, self._env(root), clear=False): fanout.link_child("parent-slug", "child-one") fanout.link_child("parent-slug", "child-two") fanout.link_child("parent-slug", "child-one") # idempotent state = json.loads((root / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["child-one"]["parent_id"], "parent-slug") self.assertEqual(state["cards"]["parent-slug"]["depends_on"], ["child-one", "child-two"]) def test_link_child_rejects_self_parent(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(ValueError): fanout.link_child("same-slug", "same-slug") def test_maybe_generate_synthesis_waits_for_all_children(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) state_path = root / "state.json" state_path.write_text( json.dumps( { "cards": { "parent-slug": {"depends_on": ["child-one", "child-two"], "title": "Parent"}, "child-one": {"source": "intent", "phase": "bereit_fuer_deployment"}, "child-two": {"source": "intent", "phase": "in_umsetzung"}, } } ), encoding="utf-8", ) with mock.patch.dict(os.environ, self._env(root), clear=False): result = fanout.maybe_generate_synthesis("parent-slug") self.assertIsNone(result) self.assertEqual(list((root / "docs" / "intent").glob("*.md")), []) def test_maybe_generate_synthesis_writes_file_once_all_children_done(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) state_path = root / "state.json" state_path.write_text( json.dumps( { "cards": { "parent-slug": {"depends_on": ["child-one", "child-two"], "title": "Parent"}, "child-one": { "source": "intent", "phase": "bereit_fuer_deployment", "commit_hash": "abc1234", }, "child-two": { "source": "research", "phase": "recherche_fertig", "research_report": "docs/research/child-two.md", }, } } ), encoding="utf-8", ) with mock.patch.dict(os.environ, self._env(root), clear=False): synthesis_slug = fanout.maybe_generate_synthesis("parent-slug") self.assertEqual(synthesis_slug, "parent-slug-synthese") # Second call is a no-op: already marked, no duplicate file. self.assertIsNone(fanout.maybe_generate_synthesis("parent-slug")) written = (root / "docs" / "intent" / "parent-slug-synthese.md").read_text(encoding="utf-8") self.assertIn("child-one", written) self.assertIn("abc1234", written) self.assertIn("child-two", written) self.assertIn("docs/research/child-two.md", written) state = json.loads(state_path.read_text(encoding="utf-8")) self.assertTrue(state["cards"]["parent-slug"]["synthesis_generated"]) self.assertEqual( state["cards"]["parent-slug"]["synthesis_intent_path"], "docs/intent/parent-slug-synthese.md", ) def test_operator_synthesis_embeds_original_parent_intent(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "parent.md").write_text( "# Parent\n\nExaktes ursprüngliches Ziel.\n", encoding="utf-8" ) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": { "parent": { "title": "Parent", "orchestration_kind": "operator", "depends_on": ["child"], }, "child": { "source": "intent", "phase": "bereit_fuer_deployment", "review_decision": "bereit_fuer_deployment", "commit_hash": "abc1234", }, }}), encoding="utf-8") with mock.patch.dict(os.environ, self._env(root), clear=False): self.assertEqual(fanout.maybe_generate_synthesis("parent"), "parent-synthese") written = (intent_dir / "parent-synthese.md").read_text(encoding="utf-8") self.assertIn("## Unveränderter Original-Intent", written) self.assertIn("Exaktes ursprüngliches Ziel.", written) def test_maybe_synthesize_falls_back_to_legacy_child_slugs(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) state_path = root / "state.json" state_path.write_text( json.dumps( { "cards": { "board-builder-column-config": { "child_slugs": ["child-one", "child-two"], "title": "Legacy split parent", }, "child-one": { "source": "intent", "phase": "bereit_fuer_deployment", "parent_id": "board-builder-column-config", }, "child-two": { "source": "intent", "phase": "planung_mit_feedback", "parent_id": "board-builder-column-config", }, } } ), encoding="utf-8", ) with mock.patch.dict(os.environ, self._env(root), clear=False): synthesis_slug = fanout.maybe_synthesize("child-two") self.assertEqual(synthesis_slug, "board-builder-column-config-synthese") written = ( root / "docs" / "intent" / "board-builder-column-config-synthese.md" ).read_text(encoding="utf-8") self.assertIn("child-one", written) self.assertIn("child-two", written) def test_maybe_synthesize_is_noop_without_parent_id(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {"slug": {"phase": "review"}}}), encoding="utf-8") with mock.patch.dict(os.environ, self._env(root), clear=False): self.assertIsNone(fanout.maybe_synthesize("slug")) class LinkParentApiTests(unittest.TestCase): def _env(self, root): return { "BOARD_STATE_FILE": str(root / "state.json"), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(root / "docs" / "research-intent"), } def test_link_parent_rejects_unknown_child(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "parent-slug.md").write_text("# Parent\n", encoding="utf-8") with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(LookupError): trigger.link_parent("no-such-card", "parent-slug") def test_link_parent_rejects_unknown_parent(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "child-slug.md").write_text("# Child\n", encoding="utf-8") with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(LookupError): trigger.link_parent("child-slug", "no-such-parent") def test_link_parent_success_between_two_known_intent_cards(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) (root / "docs" / "intent" / "parent-slug.md").write_text("# Parent\n", encoding="utf-8") (root / "docs" / "intent" / "child-slug.md").write_text("# Child\n", encoding="utf-8") with mock.patch.dict(os.environ, self._env(root), clear=False): trigger.link_parent("child-slug", "parent-slug") board = trigger.merged_board() card = next(c for c in board if c["id"] == "child-slug") self.assertEqual(card["parent_id"], "parent-slug") def test_patch_parent_route_dispatches_to_link_parent(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/child-slug/parent" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(return_value={"parent_id": "parent-slug"}) with mock.patch.object(trigger, "link_parent") as link_parent: handler.do_PATCH() link_parent.assert_called_once_with("child-slug", "parent-slug") handler._send_json.assert_called_once_with(200, {"status": "ok"}) class SmokeTestButtonTests(unittest.TestCase): """The board's install/removal smoke-test buttons write fresh, numbered intents without slug collisions between test kinds.""" def test_create_smoke_test_numbers_fresh_slugs(self): with tempfile.TemporaryDirectory() as intent_dir: with mock.patch.dict(os.environ, {"BOARD_INTENT_DIR": intent_dir}, clear=False): first = trigger.create_smoke_test("direct") second = trigger.create_smoke_test("direct") self.assertEqual(first, "board-operator-smoke-test-direct-1") self.assertEqual(second, "board-operator-smoke-test-direct-2") content = (Path(intent_dir) / f"{second}.md").read_text(encoding="utf-8") self.assertIn("`ripgrep` systemweit", content) self.assertIn("`hosts/blunix/localServices/packages.nix`", content) self.assertIn("`direct_builder`", content) def test_create_smoke_test_continues_past_existing_numbers(self): with tempfile.TemporaryDirectory() as intent_dir: (Path(intent_dir) / "board-operator-smoke-test-split-2.md").write_text("x", encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_INTENT_DIR": intent_dir}, clear=False): slug = trigger.create_smoke_test("split") self.assertEqual(slug, "board-operator-smoke-test-split-3") content = (Path(intent_dir) / f"{slug}.md").read_text(encoding="utf-8") self.assertIn("exakt zwei Kinder", content) self.assertIn("`ripgrep`", content) self.assertIn("`fd`", content) self.assertIn("`hosts/blunix/localServices/packages.nix`", content) self.assertIn("`hosts/blunix/localServices/packages-stable.nix`", content) def test_create_remove_smoke_tests_have_opposite_package_actions(self): with tempfile.TemporaryDirectory() as intent_dir: with mock.patch.dict(os.environ, {"BOARD_INTENT_DIR": intent_dir}, clear=False): direct_slug = trigger.create_smoke_test("remove-direct") split_slug = trigger.create_smoke_test("remove-split") self.assertEqual(direct_slug, "board-operator-smoke-test-remove-direct-1") self.assertEqual(split_slug, "board-operator-smoke-test-remove-split-1") direct = (Path(intent_dir) / f"{direct_slug}.md").read_text(encoding="utf-8") split = (Path(intent_dir) / f"{split_slug}.md").read_text(encoding="utf-8") self.assertIn("Entferne ausschließlich `ripgrep`", direct) self.assertIn("exakt zwei Kinder", split) self.assertIn("entfernt ausschließlich `ripgrep`", split) self.assertIn("entfernt ausschließlich `fd`", split) def test_create_smoke_test_rejects_unknown_kind(self): with self.assertRaises(ValueError): trigger.create_smoke_test("bogus") def test_post_smoke_tests_route_dispatches(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/smoke-tests/remove-split" handler._send_json = mock.Mock() with mock.patch.object(trigger, "create_smoke_test", return_value="slug-1") as create: handler.do_POST() create.assert_called_once_with("remove-split") handler._send_json.assert_called_once_with(201, {"id": "slug-1"}) def test_frontend_has_smoke_test_buttons(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('id="smoke-test-direct"', board_html) self.assertIn('id="smoke-test-split"', board_html) self.assertIn('id="smoke-test-remove-direct"', board_html) self.assertIn('id="smoke-test-remove-split"', board_html) self.assertIn('["smoke-test-remove-direct", "remove-direct"]', board_html) self.assertIn('["smoke-test-remove-split", "remove-split"]', board_html) self.assertIn("createSmokeTest(kind)", board_html) def test_frontend_has_smoke_test_cleanup_button(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('id="smoke-test-cleanup"', board_html) self.assertIn("deleteSmokeTestCards()", board_html) self.assertIn("window.confirm(", board_html) class SmokeTestCleanupTests(unittest.TestCase): """DELETE /api/board/smoke-tests: bulk-remove every smoke-test card and its linked family (split/fan-out children, generated synthesis) at once, backed by delete_smoke_test_cards().""" def _environment(self, repo_dir, state_path, root): return { "BOARD_REPO_DIR": str(repo_dir), "BOARD_INTENT_DIR": str(repo_dir / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(repo_dir / "docs" / "research-intent"), "BOARD_STATE_FILE": str(state_path), "BOARD_TELEMETRY_DB": str(root / "telemetry.sqlite3"), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } def test_removes_smoke_card_its_split_child_and_backing_files(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) repo_dir = root / "repo" intent_dir = repo_dir / "docs" / "intent" review_dir = repo_dir / "docs" / "review" intent_dir.mkdir(parents=True) review_dir.mkdir(parents=True) (intent_dir / "board-operator-smoke-test-split-1.md").write_text("x", encoding="utf-8") (intent_dir / "install-fd-2.md").write_text("x", encoding="utf-8") (review_dir / "install-fd-2.md").write_text("x", encoding="utf-8") (intent_dir / "unrelated-card.md").write_text("x", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps( { "cards": { "board-operator-smoke-test-split-1": { "title": "Smoke-Test", "child_slugs": ["install-fd-2"], }, "install-fd-2": {"title": "fd entfernen", "parent_id": "board-operator-smoke-test-split-1"}, "unrelated-card": {"title": "Echte Karte"}, } } ), encoding="utf-8", ) with mock.patch.dict(os.environ, self._environment(repo_dir, state_path, root), clear=False): result = trigger.delete_smoke_test_cards() self.assertEqual(result, {"removed_cards": 2, "removed_files": 3}) remaining = json.loads(state_path.read_text(encoding="utf-8"))["cards"] self.assertEqual(list(remaining), ["unrelated-card"]) self.assertFalse((intent_dir / "board-operator-smoke-test-split-1.md").exists()) self.assertFalse((intent_dir / "install-fd-2.md").exists()) self.assertFalse((review_dir / "install-fd-2.md").exists()) self.assertTrue((intent_dir / "unrelated-card.md").exists()) def test_refuses_while_a_board_job_is_queued(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) repo_dir = root / "repo" (repo_dir / "docs" / "intent").mkdir(parents=True) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {}}), encoding="utf-8") inbox_dir = root / "inbox" inbox_dir.mkdir() (inbox_dir / "job-1.json").write_text("{}", encoding="utf-8") environment = self._environment(repo_dir, state_path, root) environment["BOARD_INBOX_DIR"] = str(inbox_dir) with mock.patch.dict(os.environ, environment, clear=False): with self.assertRaises(PermissionError): trigger.delete_smoke_test_cards() def test_delete_smoke_tests_route_dispatches(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/smoke-tests" handler._send_json = mock.Mock() with mock.patch.object( trigger, "delete_smoke_test_cards", return_value={"removed_cards": 2, "removed_files": 3} ) as delete: handler.do_DELETE() delete.assert_called_once_with() handler._send_json.assert_called_once_with( 200, {"status": "deleted", "removed_cards": 2, "removed_files": 3} ) def test_delete_smoke_tests_route_returns_409_when_job_active(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/smoke-tests" handler._send_json = mock.Mock() with mock.patch.object( trigger, "delete_smoke_test_cards", side_effect=PermissionError("Ein Board-Job läuft oder ist eingereiht") ): handler.do_DELETE() handler._send_json.assert_called_once_with( 409, {"error": "Ein Board-Job läuft oder ist eingereiht"} ) class RetryReviewApiTests(unittest.TestCase): """docs/intent/board-agent-retry-review-api.md: queue a fresh Review job against a card's own already-recorded commit_hash, without a new Builder run -- for a Review that failed for reasons unrelated to the commit itself (e.g. the 2026-09-09 case-sensitivity bug).""" def _env(self, root): return { "BOARD_STATE_FILE": str(root / "state.json"), "BOARD_INTENT_DIR": str(root / "docs" / "intent"), "BOARD_RESEARCH_INTENT_DIR": str(root / "docs" / "research-intent"), "BOARD_INBOX_DIR": str(root / "inbox"), "BOARD_PROCESSING_DIR": str(root / "processing"), } def _make_card(self, root, slug, **stored): (root / "docs" / "intent").mkdir(parents=True, exist_ok=True) (root / "docs" / "intent" / f"{slug}.md").write_text(f"# Intent: {slug}\n", encoding="utf-8") state_path = root / "state.json" state = {"cards": {slug: stored}} state_path.write_text(json.dumps(state), encoding="utf-8") def test_rejects_unknown_card(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) (root / "docs" / "intent").mkdir(parents=True) with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(LookupError): trigger.retry_review("no-such-card") def test_rejects_card_without_a_recorded_commit(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) self._make_card(root, "my-slug", phase="in_umsetzung_leicht", last_run_status="no_changes") with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(LookupError): trigger.retry_review("my-slug") def test_rejects_when_a_job_is_already_active(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) self._make_card(root, "my-slug", commit_hash="abc1234") (root / "inbox").mkdir(parents=True) (root / "inbox" / "other-job.json").write_text("{}", encoding="utf-8") with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(PermissionError): trigger.retry_review("my-slug") def test_writes_review_job_using_the_cards_own_stored_profile(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) self._make_card( root, "my-slug", commit_hash="91c5bc8", runtime="claude", model="sonnet", reasoning_effort="low", phase="in_umsetzung_leicht", last_run_status="no_changes", ) with mock.patch.dict(os.environ, self._env(root), clear=False): job_id = trigger.retry_review("my-slug") jobs = [ json.loads(path.read_text(encoding="utf-8")) for path in (root / "inbox").glob("*.json") ] state = json.loads((root / "state.json").read_text(encoding="utf-8")) self.assertEqual(len(jobs), 1) job = jobs[0] self.assertEqual(job["job_id"], job_id) self.assertEqual(job["job_type"], "review") self.assertEqual(job["commit_hash"], "91c5bc8") self.assertEqual(job["runtime"], "claude") self.assertEqual(job["model"], "sonnet") self.assertEqual(job["reasoning_effort"], "low") card = state["cards"]["my-slug"] self.assertEqual(card["phase"], "review") self.assertEqual(card["last_run_status"], "queued_review") self.assertEqual(card["last_job_id"], job_id) def test_retry_review_accepts_an_explicit_commit_hash_override(self): # Needed only for a card whose Review failed before # completion_watcher.py started recording commit_hash at the # commit itself -- see retry_review()'s docstring. with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) self._make_card(root, "my-slug", runtime="claude") with mock.patch.dict(os.environ, self._env(root), clear=False): job_id = trigger.retry_review("my-slug", "691fa38f036b19932c6568f9e9e82cd367e9ff21") jobs = [ json.loads(path.read_text(encoding="utf-8")) for path in (root / "inbox").glob("*.json") ] self.assertEqual(len(jobs), 1) self.assertEqual(jobs[0]["job_id"], job_id) self.assertEqual(jobs[0]["commit_hash"], "691fa38f036b19932c6568f9e9e82cd367e9ff21") def test_retry_review_rejects_a_malformed_commit_hash(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) self._make_card(root, "my-slug", commit_hash="abc1234") with mock.patch.dict(os.environ, self._env(root), clear=False): with self.assertRaises(ValueError): trigger.retry_review("my-slug", "not a hash; rm -rf /") def test_post_retry_review_route_dispatches_to_retry_review(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/my-slug/retry-review" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(side_effect=ValueError("invalid request body")) with mock.patch.object(trigger, "retry_review", return_value="job-1") as retry_review: handler.do_POST() retry_review.assert_called_once_with("my-slug", None) handler._send_json.assert_called_once_with(200, {"status": "ok", "job_id": "job-1"}) def test_post_retry_review_route_forwards_an_explicit_commit_hash_body(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/my-slug/retry-review" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(return_value={"commit_hash": "abc1234"}) with mock.patch.object(trigger, "retry_review", return_value="job-1") as retry_review: handler.do_POST() retry_review.assert_called_once_with("my-slug", "abc1234") def test_post_retry_review_route_surfaces_lookup_and_permission_errors(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/my-slug/retry-review" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(side_effect=ValueError("invalid request body")) with mock.patch.object(trigger, "retry_review", side_effect=LookupError("nope")): handler.do_POST() handler._send_json.assert_called_once_with(404, {"error": "nope"}) handler._send_json = mock.Mock() with mock.patch.object(trigger, "retry_review", side_effect=PermissionError("busy")): handler.do_POST() handler._send_json.assert_called_once_with(409, {"error": "busy"}) handler._send_json = mock.Mock() with mock.patch.object(trigger, "retry_review", side_effect=ValueError("bad hash")): handler.do_POST() handler._send_json.assert_called_once_with(400, {"error": "bad hash"}) class DraftDialogueApiTests(unittest.TestCase): """docs/intent/board-interactive-planner-intake.md: trigger.py's side of the interactive intake dialogue -- create_draft()/answer_draft()/ retry_draft(), delete_manual_card()'s broadened idea-or-draft check, and the routes that expose them.""" def _environment(self, state_path, inbox_dir): return {"BOARD_STATE_FILE": str(state_path), "BOARD_INBOX_DIR": str(inbox_dir)} def test_create_draft_writes_state_and_queues_opening_turn(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" inbox_dir = root / "inbox" with mock.patch.dict(os.environ, self._environment(state_path, inbox_dir), clear=False): card_id = trigger.create_draft("Ripgrep von blunix entfernen") self.assertTrue(card_id.startswith("draft-")) state = json.loads(state_path.read_text(encoding="utf-8")) card = state["cards"][card_id] self.assertEqual(card["source"], "draft") self.assertEqual(card["phase"], "entwurf_laeuft") self.assertEqual(card["draft_conversation"], [{"role": "user", "text": "Ripgrep von blunix entfernen"}]) jobs = list(inbox_dir.glob("*.json")) self.assertEqual(len(jobs), 1) job = json.loads(jobs[0].read_text(encoding="utf-8")) self.assertEqual(job["job_type"], "draft") self.assertEqual(job["slug"], card_id) def test_create_draft_rejects_empty_topic(self): with self.assertRaises(ValueError): trigger.create_draft(" ") def test_answer_draft_appends_conversation_and_requeues(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" inbox_dir = root / "inbox" state_path.write_text(json.dumps({"cards": {"draft-1": { "source": "draft", "phase": "entwurf_frage", "title": "Thema", "draft_conversation": [ {"role": "user", "text": "Thema"}, {"role": "assistant", "text": "Welcher Host?", "options": ["blunix", "pinix"]}, ], }}}), encoding="utf-8") with mock.patch.dict(os.environ, self._environment(state_path, inbox_dir), clear=False): job_id = trigger.answer_draft("draft-1", "blunix") self.assertTrue(job_id) state = json.loads(state_path.read_text(encoding="utf-8")) card = state["cards"]["draft-1"] self.assertEqual(card["phase"], "entwurf_laeuft") self.assertEqual(card["draft_conversation"][-1], {"role": "user", "text": "blunix"}) def test_answer_draft_rejects_wrong_phase(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {"draft-1": { "source": "draft", "phase": "entwurf_laeuft", "draft_conversation": [], }}}), encoding="utf-8") with mock.patch.dict(os.environ, self._environment(state_path, root / "inbox"), clear=False): with self.assertRaises(PermissionError): trigger.answer_draft("draft-1", "blunix") def test_answer_draft_rejects_unknown_card(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {}}), encoding="utf-8") with mock.patch.dict(os.environ, self._environment(state_path, root / "inbox"), clear=False): with self.assertRaises(LookupError): trigger.answer_draft("draft-nope", "blunix") def test_answer_draft_rejects_empty_answer(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {"draft-1": { "source": "draft", "phase": "entwurf_frage", "draft_conversation": [], }}}), encoding="utf-8") with mock.patch.dict(os.environ, self._environment(state_path, root / "inbox"), clear=False): with self.assertRaises(ValueError): trigger.answer_draft("draft-1", " ") def test_retry_draft_reuses_conversation_unchanged(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" inbox_dir = root / "inbox" conversation = [{"role": "user", "text": "Thema"}] state_path.write_text(json.dumps({"cards": {"draft-1": { "source": "draft", "phase": "entwurf_fehlgeschlagen", "draft_conversation": conversation, }}}), encoding="utf-8") with mock.patch.dict(os.environ, self._environment(state_path, inbox_dir), clear=False): trigger.retry_draft("draft-1") job = json.loads(next(inbox_dir.glob("*.json")).read_text(encoding="utf-8")) self.assertEqual(job["conversation"], conversation) state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(state["cards"]["draft-1"]["phase"], "entwurf_laeuft") def test_retry_draft_rejects_wrong_phase(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {"draft-1": { "source": "draft", "phase": "entwurf_frage", "draft_conversation": [], }}}), encoding="utf-8") with mock.patch.dict(os.environ, self._environment(state_path, root / "inbox"), clear=False): with self.assertRaises(PermissionError): trigger.retry_draft("draft-1") def test_delete_manual_card_accepts_idea_and_draft(self): with tempfile.TemporaryDirectory() as temp_dir: state_path = Path(temp_dir) / "state.json" state_path.write_text(json.dumps({"cards": { "idea-1": {"source": "idea"}, "draft-1": {"source": "draft"}, }}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): trigger.delete_manual_card("idea-1") trigger.delete_manual_card("draft-1") state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(state["cards"], {}) def test_delete_manual_card_rejects_intent_card(self): with tempfile.TemporaryDirectory() as temp_dir: state_path = Path(temp_dir) / "state.json" state_path.write_text(json.dumps({"cards": {"real-slug": {"source": "intent"}}}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): with self.assertRaises(LookupError): trigger.delete_manual_card("real-slug") def test_merged_board_exposes_draft_card(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": {"draft-1": { "source": "draft", "title": "Thema", "phase": "entwurf_frage", "draft_conversation": [{"role": "assistant", "text": "Welcher Host?", "options": []}], }}}), encoding="utf-8") environment = { "BOARD_REPO_DIR": str(root / "repo"), "BOARD_INTENT_DIR": str(root / "repo" / "docs" / "intent"), "BOARD_STATE_FILE": str(state_path), } with mock.patch.dict(os.environ, environment, clear=False): card = trigger.merged_board()[0] self.assertEqual(card["source"], "draft") self.assertEqual(card["phase"], "entwurf_frage") self.assertEqual(card["draft_conversation"][0]["text"], "Welcher Host?") def test_post_drafts_route_dispatches(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/drafts" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(return_value={"topic": "Ripgrep entfernen"}) with mock.patch.object(trigger, "create_draft", return_value="draft-1") as create: handler.do_POST() create.assert_called_once_with("Ripgrep entfernen") handler._send_json.assert_called_once_with(201, {"id": "draft-1"}) def test_post_draft_answer_route_dispatches(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/draft-1/draft-answer" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(return_value={"answer": "blunix"}) with mock.patch.object(trigger, "answer_draft", return_value="job-1") as answer: handler.do_POST() answer.assert_called_once_with("draft-1", "blunix") handler._send_json.assert_called_once_with(200, {"status": "ok", "job_id": "job-1"}) def test_post_draft_answer_route_returns_409_when_phase_wrong(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/draft-1/draft-answer" handler._send_json = mock.Mock() handler._read_json_body = mock.Mock(return_value={"answer": "blunix"}) with mock.patch.object(trigger, "answer_draft", side_effect=PermissionError("nope")): handler.do_POST() handler._send_json.assert_called_once_with(409, {"error": "nope"}) def test_post_draft_retry_route_dispatches(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/draft-1/draft-retry" handler._send_json = mock.Mock() with mock.patch.object(trigger, "retry_draft", return_value="job-1") as retry: handler.do_POST() retry.assert_called_once_with("draft-1") handler._send_json.assert_called_once_with(200, {"status": "ok", "job_id": "job-1"}) def test_delete_draft_route_dispatches(self): handler = object.__new__(trigger.Handler) handler.path = "/api/board/cards/draft-1" handler._send_json = mock.Mock() with mock.patch.object(trigger, "delete_manual_card", return_value=None) as delete: handler.do_DELETE() delete.assert_called_once_with("draft-1") handler._send_json.assert_called_once_with(200, {"status": "deleted"}) def test_frontend_has_interactive_draft_ui(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("createDraft(topic)", board_html) self.assertIn("answerDraft(id, answer)", board_html) self.assertIn("retryDraft(id)", board_html) self.assertIn("/api/board/drafts", board_html) self.assertIn("appendDraftDialogue", board_html) self.assertIn("+ Plan a goal", board_html) class CompletionWatcherTests(unittest.TestCase): def test_deployed_child_progresses_fanout_and_operator_families(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" intent_dir = root / "intent" intent_dir.mkdir() state_path.write_text(json.dumps({"cards": { "parent": {"phase": "wartet_auf_kinder", "depends_on": ["child"]}, "child": { "phase": "bereit_fuer_deployment", "parent_id": "parent", "review_decision": "bereit_fuer_deployment", "commit_hash": "abc1234", }, }}), encoding="utf-8") record_path = root / "board-child-1.json" worker.write_payload(record_path, { "job_id": "board-child-1", "status": "committed", "deployment_status": "deployed", "deployed_commit_hash": "abc1234" + "0" * 33, "activated_system": "/nix/store/test-system", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], }) with ( mock.patch.dict(os.environ, { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(intent_dir), }, clear=False), mock.patch.object(completion_watcher, "progress_operator_families") as progress, ): completion_watcher.process_record(record_path) progress.assert_called_once_with() state = json.loads(state_path.read_text(encoding="utf-8")) card = state["cards"]["child"] self.assertEqual(card["phase"], "deployed") self.assertTrue(state["cards"]["parent"]["synthesis_generated"]) self.assertTrue((intent_dir / "parent-synthese.md").is_file()) def test_combined_committed_and_deployed_record_cannot_skip_review(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" state_path.write_text(json.dumps({"cards": { "child": {"phase": "queued_builder", "parent_id": "parent"}, "parent": {"phase": "wartet_auf_kinder", "depends_on": ["child"]}, }}), encoding="utf-8") record_path = root / "board-child-1.json" worker.write_payload(record_path, { "job_id": "board-child-1", "status": "committed", "deployment_status": "deployed", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "fab5c07"}], }) with ( mock.patch.dict(os.environ, { "BOARD_STATE_FILE": str(state_path), "BOARD_INBOX_DIR": str(root / "inbox"), }, clear=False), mock.patch.object( completion_watcher.trigger, "load_column_config", return_value={"review": {"runtime": "codex"}}, ), mock.patch.object(completion_watcher.fanout, "maybe_synthesize") as synthesize, mock.patch.object(completion_watcher, "progress_operator_families") as progress, ): completion_watcher.process_record(record_path) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["child"] self.assertEqual(card["phase"], "review") self.assertEqual(card["commit_hash"], "fab5c07") self.assertEqual(card["deployment_status"], "deployed") self.assertTrue(card["deployment_before_review"]) self.assertIsNone(card.get("review_decision")) review_jobs = list((root / "inbox").glob("*.json")) self.assertEqual(len(review_jobs), 1) self.assertEqual(json.loads(review_jobs[0].read_text())["commit_hash"], "fab5c07") synthesize.assert_not_called() progress.assert_called_once_with() def test_failed_fixer_without_outcome_preserves_runtime_failure(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" record_path = root / "board-card-1.json" worker.write_payload(record_path, { "job_id": "board-card-1", "pipeline_role": "fixer", "status": "failed", "runtime": "claude", "fixer_outcome": None, "failure_stage": "runtime", "failure_detail": "Claude exit-code 1", "report": "Providerprozess fehlgeschlagen.", }) with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): completion_watcher.process_record(record_path) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["card"] self.assertEqual(card["last_run_status"], "failed") self.assertEqual(card["failure_stage"], "runtime") self.assertEqual(card["failure_detail"], "Claude exit-code 1") self.assertEqual(card["builder_report"], "Providerprozess fehlgeschlagen.") def test_successful_fixer_with_malformed_outcome_is_outcome_failure(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" record_path = root / "board-card-1.json" worker.write_payload(record_path, { "job_id": "board-card-1", "pipeline_role": "fixer", "status": "no_changes", "runtime": "claude", "fixer_outcome": None, "report": "Antwort ohne Ergebnisblock.", }) with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): completion_watcher.process_record(record_path) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["card"] self.assertEqual(card["last_run_status"], "failed") self.assertEqual(card["failure_stage"], "fixer_outcome") self.assertEqual(card["builder_report"], "Antwort ohne Ergebnisblock.") def test_other_underlying_fixer_failures_without_outcome_keep_status(self): for status in ("check_failed", "limit_reached"): with self.subTest(status=status), tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" record_path = root / "board-card-1.json" worker.write_payload(record_path, { "job_id": "board-card-1", "pipeline_role": "fixer", "status": status, "fixer_outcome": None, "report": f"Underlying {status}", }) with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): completion_watcher.process_record(record_path) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["card"] self.assertEqual(card["last_run_status"], status) self.assertEqual(card["failure_stage"], "fixer_runtime") self.assertEqual(card["failure_detail"], f"Underlying {status}") def test_fixer_needs_input_persists_question_without_waiting_session(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" record_path = root / "board-card-1.json" worker.write_payload(record_path, { "job_id": "board-card-1", "pipeline_role": "fixer", "status": "no_changes", "runtime": "codex", "fixer_outcome": {"outcome": "needs_input", "question": "Welche Option?", "summary": "unklar"}, }) with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): completion_watcher.process_record(record_path) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["card"] self.assertEqual(card["phase"], "inspector_rueckfrage") self.assertEqual(card["fixer_question"], "Welche Option?") def test_fixed_fixer_completion_always_queues_review_not_deploy(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" record_path = root / "board-card-1.json" worker.write_payload(record_path, { "job_id": "board-card-1", "pipeline_role": "fixer", "status": "committed", "runtime": "codex", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], "fixer_outcome": { "schema_version": 1, "outcome": "repaired", "summary": "repariert", "commit_hash": "a" * 40, "pushed_branch": "dev", "tests": [], "question": None, "required_action": None, "review_required": True, "failure": None, }, }) environment = {"BOARD_STATE_FILE": str(state_path), "BOARD_INBOX_DIR": str(root / "inbox")} with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(completion_watcher.trigger, "load_column_config", return_value={"review": {"runtime": "codex"}}), ): completion_watcher.process_record(record_path) jobs = [json.loads(path.read_text(encoding="utf-8")) for path in (root / "inbox").glob("*.json")] self.assertEqual([job["job_type"] for job in jobs], ["review"]) self.assertTrue(jobs[0]["fixer_result"]) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["card"] self.assertEqual(card["phase"], "review") self.assertEqual(card["commit_hash"], "abc1234") self.assertTrue(card["fixer_commit_discrepancy"]) def test_reviewed_fixer_deploy_request_is_pinned_to_blunix(self): with tempfile.TemporaryDirectory() as temp_dir: environment = { "AGENT_BUILDER_DEPLOY_QUEUE_DIR": temp_dir, "BOARD_REPO_DIR": "/home/drone/nixos-config", "BOARD_BUILDER_RESULTS_DIR": "/var/lib/board-agent/builder-results", } with mock.patch.dict(os.environ, environment, clear=False): job_id = worker.queue_reviewed_deployment( "card", "abc1234", {"runtime": "codex", "pipeline_run_id": "run-1"} ) request = json.loads((Path(temp_dir) / f"{job_id}.json").read_text(encoding="utf-8")) self.assertEqual(request["host"], "blunix") self.assertEqual(request["repo"], "/home/drone/nixos-config") self.assertNotIn("tnix", json.dumps(request)) self.assertNotIn("pinix", json.dumps(request)) def test_slug_from_job_id_parses_board_prefixed_ids(self): self.assertEqual( completion_watcher.slug_from_job_id("board-my-slug-1735689600"), "my-slug" ) self.assertEqual(completion_watcher.slug_from_job_id("not-a-board-job"), "") def test_committed_record_enqueues_review_job_and_sets_review_phase(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "committed", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], "runtime": "claude", }, ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( completion_watcher.trigger, "load_column_config", return_value={"review": {"runtime": "claude"}}, ), ): completion_watcher.process_record(record_path) queued = list(Path(inbox_dir).glob("*.json")) self.assertEqual(len(queued), 2) # the record itself + the new review job review_jobs = [ json.loads(p.read_text(encoding="utf-8")) for p in queued if json.loads(p.read_text(encoding="utf-8")).get("job_type") == "review" ] self.assertEqual(len(review_jobs), 1) self.assertEqual(review_jobs[0]["commit_hash"], "abc1234") self.assertEqual(review_jobs[0]["slug"], "my-slug") self.assertEqual(review_jobs[0]["runtime"], "claude") state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["my-slug"]["phase"], "review") # Recorded at the commit itself, not only on a successful # Review -- see completion_watcher.py's 2026-09-09 fix and # trigger.retry_review(), which depends on this being present # even after a Review failure. self.assertEqual(state["cards"]["my-slug"]["commit_hash"], "abc1234") def test_committed_record_uses_fresh_review_column_config(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "committed", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], "runtime": "codex", }, ) review_config = { "runtime": "claude", "model": "claude-reviewer", "reasoning_effort": "high", } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( completion_watcher.trigger, "load_column_config", return_value={"review": review_config}, ), ): completion_watcher.process_record(record_path) review_jobs = [ json.loads(path.read_text(encoding="utf-8")) for path in Path(inbox_dir).glob("*.json") if json.loads(path.read_text(encoding="utf-8")).get("job_type") == "review" ] self.assertEqual(len(review_jobs), 1) self.assertEqual( {key: review_jobs[0][key] for key in ("runtime", "model", "reasoning_effort")}, review_config, ) def test_committed_record_marks_review_config_errors_diagnostically(self): with tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir: environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "committed", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], }, ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object(completion_watcher.trigger, "load_column_config", return_value={}), ): completion_watcher.process_record(record_path) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) card = state["cards"]["my-slug"] self.assertEqual(card["last_run_status"], "failed") self.assertEqual(card["failure_stage"], "review_config") self.assertIn("review", card["failure_detail"].lower()) self.assertEqual(list(Path(inbox_dir).glob("*.json")), [record_path]) def test_no_changes_record_only_updates_status_without_phase_change(self): with tempfile.TemporaryDirectory() as state_dir: environment = {"BOARD_STATE_FILE": str(Path(state_dir) / "state.json")} with mock.patch.dict(os.environ, environment, clear=False): worker.set_phase("my-slug", "in_umsetzung") record_path = Path(state_dir) / "record.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "no_changes", "report": "Ich habe nichts geändert, weil ...", }, ) completion_watcher.process_record(record_path) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["my-slug"]["phase"], "in_umsetzung") self.assertEqual(state["cards"]["my-slug"]["last_run_status"], "no_changes") self.assertEqual( state["cards"]["my-slug"]["builder_report"], "Ich habe nichts geändert, weil ..." ) def test_no_changes_operator_synthesis_completes_it_and_parent(self): with tempfile.TemporaryDirectory() as state_dir: state_path = Path(state_dir) / "state.json" state_path.write_text(json.dumps({"cards": { "parent": { "orchestration_kind": "operator", "synthesis_generated": True, "synthesis_intent_path": "docs/intent/parent-synthese.md", "phase": "wartet_auf_kinder", }, "parent-synthese": { "source": "intent", "synthesis_for": "parent", "operator_managed": True, "phase": "in_umsetzung", }, }}), encoding="utf-8") record_path = Path(state_dir) / "board-parent-synthese-1735689600.json" worker.write_payload(record_path, { "job_id": "board-parent-synthese-1735689600", "status": "no_changes", "report": "Keine Integrationsänderung nötig.", }) environment = {"BOARD_STATE_FILE": str(state_path)} profiles = { "leicht": { "runtime": "codex", "model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20, }, "schwer": { "runtime": "codex", "model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20, }, } with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( completion_watcher.trigger, "load_column_config", return_value=profiles ), ): completion_watcher.process_record(record_path) state = json.loads(state_path.read_text(encoding="utf-8")) synthesis = state["cards"]["parent-synthese"] self.assertEqual(synthesis["phase"], "bereit_fuer_deployment") self.assertEqual(synthesis["review_decision"], "bereit_fuer_deployment") self.assertTrue(synthesis["synthesis_no_changes"]) self.assertEqual(state["cards"]["parent"]["phase"], "bereit_fuer_deployment") def test_committed_record_does_not_set_builder_report(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "committed", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], "runtime": "claude", }, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertNotIn("builder_report", state["cards"]["my-slug"]) def _limit_reached_environment(self, repo_dir, queue_dir, results_dir, state_dir): intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n\nDetails.\n", encoding="utf-8") return { "BOARD_REPO_DIR": str(repo_dir), "BOARD_BUILDER_OWNER_JID": "lukas@0ff.space", "AGENT_BUILDER_QUEUE_DIR": str(queue_dir), "BOARD_BUILDER_RESULTS_DIR": str(results_dir), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } def test_first_limit_reached_queues_one_automatic_retry(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as results_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = self._limit_reached_environment(repo_dir, queue_dir, results_dir, state_dir) record_path = Path(state_dir) / "record.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "limit_reached", "report": "Zwischenstand vor Abbruch ...", "runtime": "claude", }, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) queued = list(Path(queue_dir).glob("*.json")) self.assertEqual(len(queued), 1) retry_payload = json.loads(queued[0].read_text(encoding="utf-8")) self.assertEqual(retry_payload["board_slug"], "my-slug") self.assertEqual(retry_payload["runtime"], "claude") state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["my-slug"]["last_run_status"], "queued_builder") self.assertEqual(state["cards"]["my-slug"]["auto_retry_attempts"], 1) self.assertEqual( state["cards"]["my-slug"]["builder_report"], "Zwischenstand vor Abbruch ..." ) def test_limit_reached_retry_stays_on_the_cards_own_lane(self): # See docs/intent/board-builder-weighted-lanes.md: a "leicht" card # that hits its iteration limit must retry back onto the same # dedicated instance/inbox, not the default "schwer" one. with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as queue_dir_leicht, tempfile.TemporaryDirectory() as results_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = self._limit_reached_environment(repo_dir, queue_dir, results_dir, state_dir) environment["AGENT_BUILDER_QUEUE_DIR_LEICHT"] = str(queue_dir_leicht) state_path = Path(state_dir) / "state.json" state_path.write_text( json.dumps({"cards": {"my-slug": {"phase": "in_umsetzung_leicht", "lane": "leicht"}}}), encoding="utf-8", ) record_path = Path(state_dir) / "record.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "limit_reached", "report": "Zwischenstand vor Abbruch ...", "runtime": "openrouter", }, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) self.assertEqual(len(list(Path(queue_dir_leicht).glob("*.json"))), 1) self.assertEqual(len(list(Path(queue_dir).glob("*.json"))), 0) def test_second_limit_reached_does_not_retry_again(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as results_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = self._limit_reached_environment(repo_dir, queue_dir, results_dir, state_dir) with mock.patch.dict(os.environ, environment, clear=False): worker.update_state("my-slug", "queued_builder", "board-my-slug-1735689600", auto_retry_attempts=1) record_path = Path(state_dir) / "record.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689700", "status": "limit_reached", "report": "Zweiter Abbruch ...", "runtime": "claude", }, ) completion_watcher.process_record(record_path) self.assertEqual(list(Path(queue_dir).glob("*.json")), []) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["my-slug"]["last_run_status"], "limit_reached") self.assertEqual(state["cards"]["my-slug"]["auto_retry_attempts"], 1) def test_limit_reached_with_open_ladder_step_escalates_model_and_limit(self): # docs/intent/board-operator-escalation-ladder.md: a card carrying # an Operator-authored escalation_attempts ladder gets its *next* # step's model/reasoning_effort/iteration_limit on retry, not the # identical params of the just-failed attempt. with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as results_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = self._limit_reached_environment(repo_dir, queue_dir, results_dir, state_dir) state_path = Path(state_dir) / "state.json" state_path.write_text( json.dumps({"cards": {"my-slug": { "phase": "in_umsetzung", "model": "gpt-5.6-luna", "reasoning_effort": "low", "iteration_limit": 20, "escalation_attempts": [ {"model": "opus", "reasoning_effort": "high", "iteration_limit": 30}, {"model": "opus", "reasoning_effort": "high", "iteration_limit": 40}, ], }}}), encoding="utf-8", ) record_path = Path(state_dir) / "record.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "limit_reached", "report": "Zwischenstand vor Abbruch ...", "runtime": "claude", }, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) queued = list(Path(queue_dir).glob("*.json")) self.assertEqual(len(queued), 1) retry_payload = json.loads(queued[0].read_text(encoding="utf-8")) self.assertEqual(retry_payload["model"], "opus") self.assertEqual(retry_payload["reasoning_effort"], "high") self.assertEqual(retry_payload["iteration_limit"], 30) state = json.loads(state_path.read_text(encoding="utf-8")) card = state["cards"]["my-slug"] self.assertEqual(card["auto_retry_attempts"], 1) self.assertEqual(card["model"], "opus") self.assertEqual(card["iteration_limit"], 30) def test_limit_reached_stops_only_once_the_whole_ladder_is_spent(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as results_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = self._limit_reached_environment(repo_dir, queue_dir, results_dir, state_dir) state_path = Path(state_dir) / "state.json" state_path.write_text( json.dumps({"cards": {"my-slug": { "phase": "in_umsetzung", "model": "opus", "reasoning_effort": "high", "iteration_limit": 40, "auto_retry_attempts": 2, "escalation_attempts": [ {"model": "opus", "reasoning_effort": "high", "iteration_limit": 30}, {"model": "opus", "reasoning_effort": "high", "iteration_limit": 40}, ], }}}), encoding="utf-8", ) record_path = Path(state_dir) / "record.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689700", "status": "limit_reached", "report": "Dritter Abbruch ...", "runtime": "claude", }, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) self.assertEqual(list(Path(queue_dir).glob("*.json")), []) state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(state["cards"]["my-slug"]["last_run_status"], "limit_reached") self.assertEqual(state["cards"]["my-slug"]["auto_retry_attempts"], 2) def test_manual_builder_trigger_resets_auto_retry_counter(self): with ( tempfile.TemporaryDirectory() as repo_dir, tempfile.TemporaryDirectory() as processing_dir, tempfile.TemporaryDirectory() as failed_dir, tempfile.TemporaryDirectory() as done_dir, tempfile.TemporaryDirectory() as queue_dir, tempfile.TemporaryDirectory() as state_dir, ): intent_dir = Path(repo_dir) / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# My Slug\n\nDetails.\n", encoding="utf-8") environment = { "BOARD_REPO_DIR": str(repo_dir), "BOARD_BUILDER_OWNER_JID": "lukas@0ff.space", "AGENT_BUILDER_QUEUE_DIR": str(queue_dir), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with mock.patch.dict(os.environ, environment, clear=False): worker.update_state("my-slug", "limit_reached", "old-job", auto_retry_attempts=1) job_path = Path(processing_dir) / "board-my-slug-1735689800.json" worker.write_payload( job_path, { "job_id": "board-my-slug-1735689800", "job_type": "builder", "slug": "my-slug", "intent_path": "docs/intent/my-slug.md", }, ) worker.process_job(job_path, Path(failed_dir), Path(done_dir)) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertEqual(state["cards"]["my-slug"]["auto_retry_attempts"], 0) class OperatorOrchestrationTests(unittest.TestCase): def profiles(self): return { "schwer": {"runtime": "codex", "model": "gpt-5.6", "reasoning_effort": "high", "iteration_limit": 20}, "leicht": {"runtime": "openrouter", "model": "deepseek/deepseek-v3.2", "reasoning_effort": "", "iteration_limit": 20}, } def response(self, payload): return ( "===OPERATOR-DECISION===\n" + json.dumps(payload) + "\n===END-OPERATOR-DECISION===" ) def direct(self): return { "schema_version": 1, "decision": "direct_builder", "reason": "klein", "recommended_lane": "schwer", "runtime": "codex", "model": "gpt-5.6", "reasoning_effort": "high", "human_action": None, "children": [], } def child(self, slug, wave=1, blocked_by=None, lane="leicht", scope=None): profile = self.profiles()[lane] return { "slug": slug, "title": slug, "wave": wave, "blocked_by": blocked_by or [], "recommended_lane": lane, "runtime": profile["runtime"], "model": profile["model"], "reasoning_effort": profile["reasoning_effort"] or None, "write_scope": scope or [f"hosts/{slug}.nix"], "acceptance_criteria": ["test passes"], "intent_markdown": f"# Intent: {slug}\n\n## Ziel\n\nDo it.", } def split(self, children): return { "schema_version": 1, "decision": "split", "reason": "gross", "recommended_lane": None, "runtime": None, "model": None, "reasoning_effort": None, "human_action": None, "children": children, } def test_strict_direct_parser_accepts_allowlisted_profile(self): parsed = orchestration.extract_operator_decision( self.response(self.direct()), "parent", self.profiles() ) self.assertEqual(parsed["profile"]["iteration_limit"], 20) bad = self.direct() | {"model": "invented"} with self.assertRaisesRegex(ValueError, "serverseitigen"): orchestration.extract_operator_decision(self.response(bad), "parent", self.profiles()) def test_direct_builder_without_escalation_field_defaults_to_empty(self): # docs/intent/board-operator-escalation-ladder.md: a response # predating this schema addition (no "escalation" key at all) must # keep validating exactly as before. parsed = orchestration.extract_operator_decision( self.response(self.direct()), "parent", self.profiles() ) self.assertEqual(parsed["escalation"], []) def test_direct_builder_accepts_a_valid_escalation_ladder(self): payload = self.direct() | { "escalation": [ {"model": "gpt-5.6-luna", "reasoning_effort": "high", "iteration_limit": 30}, {"model": "gpt-5.6-luna", "reasoning_effort": "high", "iteration_limit": 40}, ] } parsed = orchestration.extract_operator_decision( self.response(payload), "parent", self.profiles() ) self.assertEqual(len(parsed["escalation"]), 2) self.assertEqual(parsed["escalation"][1]["iteration_limit"], 40) def test_direct_builder_rejects_escalation_model_outside_the_pool(self): payload = self.direct() | { "escalation": [ {"model": "invented-model", "reasoning_effort": "high", "iteration_limit": 30}, ] } with self.assertRaisesRegex(ValueError, "konfigurierten Modelle"): orchestration.extract_operator_decision(self.response(payload), "parent", self.profiles()) def test_direct_builder_rejects_more_than_two_escalation_steps(self): step = {"model": "gpt-5.6-luna", "reasoning_effort": "high", "iteration_limit": 30} payload = self.direct() | {"escalation": [step, step, step]} with self.assertRaisesRegex(ValueError, "höchstens 2"): orchestration.extract_operator_decision(self.response(payload), "parent", self.profiles()) def test_direct_builder_rejects_escalation_iteration_limit_out_of_range(self): payload = self.direct() | { "escalation": [ {"model": "gpt-5.6-luna", "reasoning_effort": "high", "iteration_limit": 41}, ] } with self.assertRaisesRegex(ValueError, "zwischen 5 und 40"): orchestration.extract_operator_decision(self.response(payload), "parent", self.profiles()) def test_needs_human_rejects_a_nonempty_escalation(self): payload = { "schema_version": 1, "decision": "needs_human", "reason": "unklar", "recommended_lane": None, "runtime": None, "model": None, "reasoning_effort": None, "human_action": "Bitte klären.", "children": [], "escalation": [{"model": "gpt-5.6-luna", "reasoning_effort": "", "iteration_limit": 30}], } with self.assertRaisesRegex(ValueError, "needs_human.escalation"): orchestration.extract_operator_decision(self.response(payload), "parent", self.profiles()) def test_split_rejects_a_nonempty_top_level_escalation(self): payload = self.split([self.child("one")]) | { "escalation": [{"model": "gpt-5.6-luna", "reasoning_effort": "", "iteration_limit": 30}], } with self.assertRaisesRegex(ValueError, "split.escalation"): orchestration.extract_operator_decision(self.response(payload), "parent", self.profiles()) def test_split_child_escalation_validated_against_the_childs_own_runtime(self): one = self.child("one", lane="leicht") | { "escalation": [ {"model": "moonshotai/kimi-k2-thinking", "reasoning_effort": "", "iteration_limit": 30}, ] } parsed = orchestration.extract_operator_decision( self.response(self.split([one])), "parent", self.profiles() ) self.assertEqual( parsed["children"][0]["escalation"][0]["model"], "moonshotai/kimi-k2-thinking" ) bad_one = self.child("one", lane="leicht") | { # gpt-5.6-luna is a codex model, not an openrouter one -- must # be rejected for a "leicht" (openrouter) child. "escalation": [{"model": "gpt-5.6-luna", "reasoning_effort": "", "iteration_limit": 30}], } with self.assertRaisesRegex(ValueError, "konfigurierten Modelle"): orchestration.extract_operator_decision( self.response(self.split([bad_one])), "parent", self.profiles() ) def test_parser_tolerates_a_rationale_sentence_before_the_block(self): # Regression test for the 2026-09-09 live failure of both # board-operator-smoke-test-direct and -split: sonnet/high prefaced # the marker block with a one-line rationale despite operator.md's # "ausschließlich... ohne Text... davor" instruction. The parser # must tolerate this the same way extract_review_report() already # tolerates surrounding text, since the real trust boundary is the # schema validation below, not the framing. prefaced = "Kurze Begründung vor dem Block.\n\n" + self.response(self.direct()) parsed = orchestration.extract_operator_decision(prefaced, "parent", self.profiles()) self.assertEqual(parsed["decision"], "direct_builder") def test_parser_accepts_the_verbatim_live_failure_response(self): # Captured verbatim from a real Claude sonnet/high Operator run # against docs/intent/board-operator-smoke-test-direct.md on # 2026-09-09 -- the exact response that made the live smoke test # fail with "Operator-Antwort darf außerhalb des Entscheidungsblocks # keinen Text enthalten" before this parser was relaxed. response = ( 'This is a clear single-file, single-commit smoke test — direct_builder ' 'with the "leicht" profile fits.\n\n' "===OPERATOR-DECISION===\n" "{\n" ' "schema_version": 1,\n' ' "decision": "direct_builder",\n' ' "reason": "Auftrag erzeugt genau eine neue Datei mit exakt vorgegebenem ' 'Inhalt, keine Konfigurationsänderung, kein Recherche- oder Zerlegungsbedarf.",\n' ' "recommended_lane": "leicht",\n' ' "runtime": "openrouter",\n' ' "model": "deepseek/deepseek-v3.2",\n' ' "reasoning_effort": null,\n' ' "human_action": null,\n' ' "children": []\n' "}\n" "===END-OPERATOR-DECISION===" ) parsed = orchestration.extract_operator_decision(response, "parent", self.profiles()) self.assertEqual(parsed["decision"], "direct_builder") self.assertEqual(parsed["recommended_lane"], "leicht") def test_parser_still_rejects_more_than_one_decision_block(self): doubled = self.response(self.direct()) + "\n" + self.response(self.direct()) with self.assertRaisesRegex(ValueError, "nicht exakt einen"): orchestration.extract_operator_decision(doubled, "parent", self.profiles()) def test_split_parser_rejects_unknown_dependency_and_overlapping_wave_scope(self): unknown = self.split([self.child("one", blocked_by=["missing"], wave=2)]) with self.assertRaisesRegex(ValueError, "verweist"): orchestration.extract_operator_decision(self.response(unknown), "parent", self.profiles()) overlap = self.split([ self.child("one", scope=["hosts/blunix"]), self.child("two", scope=["hosts/blunix/test.nix"]), ]) with self.assertRaisesRegex(ValueError, "überlappendem"): orchestration.extract_operator_decision(self.response(overlap), "parent", self.profiles()) def test_apply_split_writes_no_children_when_any_target_exists(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "two.md").write_text("existing", encoding="utf-8") state = root / "state.json" state.write_text(json.dumps({"cards": {"parent": {"title": "Parent"}}}), encoding="utf-8") decision = orchestration.extract_operator_decision( self.response(self.split([self.child("one"), self.child("two", scope=["home/two.nix"])])), "parent", self.profiles(), ) with mock.patch.dict(os.environ, {"BOARD_INTENT_DIR": str(intent_dir), "BOARD_STATE_FILE": str(state)}, clear=False): with self.assertRaises(RuntimeError): orchestration.apply_split("parent", decision) self.assertFalse((intent_dir / "one.md").exists()) self.assertEqual((intent_dir / "two.md").read_text(encoding="utf-8"), "existing") def test_apply_split_persists_each_childs_escalation_ladder(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) state = root / "state.json" state.write_text(json.dumps({"cards": {"parent": {"title": "Parent"}}}), encoding="utf-8") one = self.child("one") | { "escalation": [ {"model": "moonshotai/kimi-k2-thinking", "reasoning_effort": "", "iteration_limit": 30}, ] } decision = orchestration.extract_operator_decision( self.response(self.split([one])), "parent", self.profiles(), ) with mock.patch.dict(os.environ, {"BOARD_INTENT_DIR": str(intent_dir), "BOARD_STATE_FILE": str(state)}, clear=False): orchestration.apply_split("parent", decision) saved = json.loads(state.read_text(encoding="utf-8"))["cards"]["one"] self.assertEqual( saved["escalation_attempts"], [{"model": "moonshotai/kimi-k2-thinking", "reasoning_effort": "", "iteration_limit": 30}], ) def test_apply_direct_persists_the_cards_escalation_ladder(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state = root / "state.json" state.write_text(json.dumps({"cards": {"parent": {"title": "Parent"}}}), encoding="utf-8") payload = self.direct() | { "escalation": [ {"model": "gpt-5.6-luna", "reasoning_effort": "high", "iteration_limit": 30}, ] } decision = orchestration.extract_operator_decision( self.response(payload), "parent", self.profiles(), ) with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state)}, clear=False): orchestration.apply_direct("parent", decision) saved = json.loads(state.read_text(encoding="utf-8"))["cards"]["parent"] self.assertEqual( saved["escalation_attempts"], [{"model": "gpt-5.6-luna", "reasoning_effort": "high", "iteration_limit": 30}], ) def test_scheduler_obeys_dependency_and_counts_downstream_slot(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" inbox = root / "board-inbox" downstream = root / "builder-processing" downstream.mkdir() state_path.write_text(json.dumps({"cards": { "parent": {"orchestration_kind": "operator", "depends_on": ["one", "two"]}, "one": {"operator_generated": True, "operator_parent_id": "parent", "phase": "in_umsetzung", "lane": "schwer"}, "two": {"operator_generated": True, "operator_parent_id": "parent", "phase": "geplant", "recommended_lane": "leicht", "recommended_profile": self.profiles()["leicht"], "blocked_by": ["one"], "wave": 2}, }}), encoding="utf-8") (downstream / "one.json").write_text(json.dumps({"board_slug": "one"}), encoding="utf-8") env = { "BOARD_STATE_FILE": str(state_path), "BOARD_INBOX_DIR": str(inbox), "AGENT_BUILDER_PROCESSING_DIR": str(downstream), "BOARD_OPERATOR_MAX_CONCURRENT_BUILDER_JOBS": "1", } queued = [] with mock.patch.dict(os.environ, env, clear=False): result = orchestration.progress_all(lambda *a, **kw: queued.append(a[0]) or "job", self.profiles()) self.assertEqual(result, []) self.assertEqual(queued, []) def test_operator_parent_requires_success_not_terminal_failure_for_synthesis(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) state = root / "state.json" state.write_text(json.dumps({"cards": { "parent": {"title": "Parent", "orchestration_kind": "operator", "depends_on": ["child"]}, "child": {"source": "intent", "phase": "planung_mit_feedback", "last_run_status": "done"}, }}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state), "BOARD_INTENT_DIR": str(intent_dir)}, clear=False): self.assertIsNone(fanout.maybe_generate_synthesis("parent")) self.assertFalse((intent_dir / "parent-synthese.md").exists()) class MetricsModuleTests(unittest.TestCase): """docs/intent/board-intent-run-metrics-baseline.md.""" def test_record_run_is_a_noop_for_a_slug_not_in_board_metrics_slugs(self): with tempfile.TemporaryDirectory() as state_dir: state_path = Path(state_dir) / "state.json" state_path.write_text(json.dumps({"cards": {}}), encoding="utf-8") with mock.patch.dict( os.environ, {"BOARD_STATE_FILE": str(state_path), "BOARD_METRICS_SLUGS": "other-slug"}, clear=False, ): self.assertIsNone(metrics.record_run("my-slug", "board", status="done")) state = json.loads(state_path.read_text(encoding="utf-8")) self.assertNotIn("my-slug", state["cards"]) def test_record_run_appends_to_run_history_for_an_enrolled_slug(self): with tempfile.TemporaryDirectory() as state_dir: state_path = Path(state_dir) / "state.json" state_path.write_text(json.dumps({"cards": {}}), encoding="utf-8") with mock.patch.dict( os.environ, {"BOARD_STATE_FILE": str(state_path), "BOARD_METRICS_SLUGS": "my-slug, other-slug"}, clear=False, ): entry = metrics.record_run( "my-slug", "board", status="done", job_type="review", runtime="claude", model="opus", duration_seconds=12.5, review_decision="bereit_fuer_deployment", ) self.assertEqual(entry["run_kind"], "board") self.assertEqual(entry["job_type"], "review") self.assertNotIn("commit_hash", entry) # omitted when None, not written as null metrics.record_run("my-slug", "baseline_single_prompt", status="done", runtime="claude") state = json.loads(state_path.read_text(encoding="utf-8")) history = state["cards"]["my-slug"]["run_history"] self.assertEqual(len(history), 2) self.assertEqual(history[0]["run_kind"], "board") self.assertEqual(history[1]["run_kind"], "baseline_single_prompt") def test_record_run_caps_history_length(self): with tempfile.TemporaryDirectory() as state_dir: state_path = Path(state_dir) / "state.json" state_path.write_text(json.dumps({"cards": {}}), encoding="utf-8") with mock.patch.dict( os.environ, {"BOARD_STATE_FILE": str(state_path), "BOARD_METRICS_SLUGS": "my-slug"}, clear=False, ): for _ in range(metrics.MAX_RUN_HISTORY + 5): metrics.record_run("my-slug", "board", status="done") state = json.loads(state_path.read_text(encoding="utf-8")) self.assertEqual(len(state["cards"]["my-slug"]["run_history"]), metrics.MAX_RUN_HISTORY) def test_duration_since_parses_now_iso_format(self): started = "2026-09-09T10:00:00Z" ended = datetime(2026, 9, 9, 10, 0, 30, tzinfo=timezone.utc) self.assertEqual(metrics.duration_since(started, now=ended), 30.0) def test_duration_since_returns_none_for_missing_or_malformed_input(self): self.assertIsNone(metrics.duration_since(None)) self.assertIsNone(metrics.duration_since("")) self.assertIsNone(metrics.duration_since("not-a-timestamp")) def test_summary_is_none_without_run_history(self): self.assertIsNone(metrics.summary({})) self.assertIsNone(metrics.summary({"run_history": []})) def test_summary_counts_board_baseline_and_successful(self): card = { "run_history": [ {"run_kind": "board", "review_decision": "bereit_fuer_deployment"}, {"run_kind": "board", "review_decision": "planung_mit_feedback"}, {"run_kind": "baseline_single_prompt", "status": "done"}, {"run_kind": "baseline_single_prompt", "status": "failed"}, ] } self.assertEqual( metrics.summary(card), {"total": 4, "board": 2, "baseline": 2, "successful": 2}, ) class RunBaselineTests(unittest.TestCase): def _environment(self, repo_dir): return { "BOARD_CODEX_BIN": "/bin/codex", "BOARD_CODEX_HOME": "/var/lib/board-agent/codex-home", "BOARD_HOME": "/home/board-agent", "BOARD_CLAUDE_BIN": "/bin/claude", "BOARD_REPO_DIR": repo_dir, } def test_run_baseline_rejects_openrouter(self): with tempfile.TemporaryDirectory() as repo_dir: with mock.patch.dict(os.environ, self._environment(repo_dir), clear=False): with self.assertRaises(ValueError): worker.run_baseline("Intent-Text", "openrouter", "some-model") def test_run_baseline_codex_has_no_role_dir_and_is_read_only(self): codex_result = SimpleNamespace( returncode=0, stdout=( '{"type":"item.completed","item":{"type":"agent_message","text":' + json.dumps("Antwort des Modells") + "}}\n" ), stderr="", ) with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=codex_result) as run, ): response = worker.run_baseline("# Intent: Beispiel", "codex", "gpt-strong", "high") self.assertEqual(response, "Antwort des Modells") command = run.call_args.args[0] self.assertNotIn("-C", command) self.assertIn("--sandbox", command) self.assertEqual(command[command.index("--sandbox") + 1], "read-only") self.assertEqual(command[command.index("--model") + 1], "gpt-strong") self.assertEqual(command[-1], "# Intent: Beispiel") def test_run_baseline_claude_disallows_write_tools_and_has_no_add_dir(self): claude_result = SimpleNamespace( returncode=0, stdout=json.dumps( { "type": "result", "subtype": "success", "is_error": False, "result": "Antwort des Modells", "session_id": "session-1", } ), stderr="", ) with tempfile.TemporaryDirectory() as repo_dir: with ( mock.patch.dict(os.environ, self._environment(repo_dir), clear=False), mock.patch.object(worker.subprocess, "run", return_value=claude_result) as run, ): response = worker.run_baseline("# Intent: Beispiel", "claude", "opus") self.assertEqual(response, "Antwort des Modells") command = run.call_args.args[0] self.assertNotIn("--add-dir", command) self.assertEqual( command[command.index("--disallowed-tools") + 1], "Edit,Write,MultiEdit,NotebookEdit", ) self.assertEqual(run.call_args.kwargs["input"], "# Intent: Beispiel") self.assertIsNone(run.call_args.kwargs["cwd"]) class CompletionWatcherMetricsTests(unittest.TestCase): def test_committed_builder_run_is_recorded_when_slug_enrolled(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), "BOARD_METRICS_SLUGS": "my-slug", } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "committed", "commits": [{"repo": "nixos-config", "branch": "dev", "hash": "abc1234"}], "runtime": "claude", "completed_at": 1735689660.0, }, ) with ( mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( completion_watcher.trigger, "load_column_config", return_value={"review": {"runtime": "claude"}}, ), ): completion_watcher.process_record(record_path) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) history = state["cards"]["my-slug"]["run_history"] self.assertEqual(len(history), 1) self.assertEqual(history[0]["job_type"], "builder") self.assertEqual(history[0]["status"], "committed") self.assertEqual(history[0]["commit_hash"], "abc1234") self.assertEqual(history[0]["duration_seconds"], 60.0) def test_no_changes_builder_run_is_recorded_without_commit_hash(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), "BOARD_METRICS_SLUGS": "my-slug", } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, { "job_id": "board-my-slug-1735689600", "status": "no_changes", "runtime": "codex", "completed_at": 1735689620.0, "report": "Nichts zu tun.", }, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) history = state["cards"]["my-slug"]["run_history"] self.assertEqual(len(history), 1) self.assertEqual(history[0]["status"], "no_changes") self.assertNotIn("commit_hash", history[0]) self.assertEqual(history[0]["duration_seconds"], 20.0) def test_unenrolled_slug_gets_no_run_history_at_all(self): with ( tempfile.TemporaryDirectory() as inbox_dir, tempfile.TemporaryDirectory() as state_dir, ): environment = { "BOARD_INBOX_DIR": inbox_dir, "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } record_path = Path(inbox_dir) / "board-my-slug-1735689600.json" worker.write_payload( record_path, {"job_id": "board-my-slug-1735689600", "status": "no_changes", "runtime": "codex"}, ) with mock.patch.dict(os.environ, environment, clear=False): completion_watcher.process_record(record_path) state = json.loads((Path(state_dir) / "state.json").read_text(encoding="utf-8")) self.assertNotIn("run_history", state["cards"].get("my-slug", {})) class MergedBoardMetricsSummaryTests(unittest.TestCase): def test_merged_board_exposes_metrics_summary_for_enrolled_intent_card(self): with ( tempfile.TemporaryDirectory() as intent_dir, tempfile.TemporaryDirectory() as state_dir, ): (Path(intent_dir) / "my-slug.md").write_text("# Intent: Beispiel\n", encoding="utf-8") state_path = Path(state_dir) / "state.json" state_path.write_text( json.dumps( { "cards": { "my-slug": { "run_history": [ {"run_kind": "board", "status": "done", "review_decision": "bereit_fuer_deployment"}, {"run_kind": "baseline_single_prompt", "status": "failed"}, ] } } } ), encoding="utf-8", ) environment = { "BOARD_INTENT_DIR": intent_dir, "BOARD_RESEARCH_INTENT_DIR": str(Path(intent_dir).parent / "no-research-intent"), "BOARD_STATE_FILE": str(state_path), } with mock.patch.dict(os.environ, environment, clear=False): cards = trigger.merged_board() card = next(card for card in cards if card["id"] == "my-slug") self.assertEqual( card["metrics_summary"], {"total": 2, "board": 1, "baseline": 1, "successful": 1}, ) def test_merged_board_omits_metrics_summary_for_a_card_with_no_run_history(self): with ( tempfile.TemporaryDirectory() as intent_dir, tempfile.TemporaryDirectory() as state_dir, ): (Path(intent_dir) / "my-slug.md").write_text("# Intent: Beispiel\n", encoding="utf-8") environment = { "BOARD_INTENT_DIR": intent_dir, "BOARD_RESEARCH_INTENT_DIR": str(Path(intent_dir).parent / "no-research-intent"), "BOARD_STATE_FILE": str(Path(state_dir) / "state.json"), } with mock.patch.dict(os.environ, environment, clear=False): cards = trigger.merged_board() card = next(card for card in cards if card["id"] == "my-slug") self.assertIsNone(card["metrics_summary"]) class MetricsBoardFrontendTests(unittest.TestCase): def test_frontend_renders_metrics_badge_when_present(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("function appendMetricsBadge(el, card)", board_html) self.assertIn("card.metrics_summary", board_html) self.assertIn('textSection(content,"Current pipeline",formatPipelineMetrics(telemetry.current_pipeline))', board_html) class PipelineTelemetryIntegrationTests(unittest.TestCase): def test_trigger_job_gets_pipeline_id_and_explicit_id_is_preserved(self): with tempfile.TemporaryDirectory() as temp_dir: environment = { "BOARD_INBOX_DIR": temp_dir, "BOARD_STATE_FILE": str(Path(temp_dir) / "state.json"), } with mock.patch.dict(os.environ, environment, clear=False): generated_id = trigger.write_job("one", "One", "docs/intent/one.md") explicit_id = trigger.write_job( "two", "Two", "docs/intent/two.md", pipeline_run_id="pipeline-fixed", ) generated = json.loads( (Path(temp_dir) / f"{generated_id}.json").read_text(encoding="utf-8") ) explicit = json.loads( (Path(temp_dir) / f"{explicit_id}.json").read_text(encoding="utf-8") ) self.assertRegex(generated["pipeline_run_id"], r"^[0-9a-f]{32}$") self.assertEqual(explicit["pipeline_run_id"], "pipeline-fixed") def test_builder_millisecond_timing_is_converted_to_seconds(self): timing = completion_watcher._builder_timing( "board-my-slug-100", {"timing": {"wall_ms": 12500, "model_ms": 9000, "ttft_ms": 250}}, {}, ) self.assertEqual( timing, {"duration_seconds": 12.5, "model_duration_seconds": 9.0, "ttft_seconds": 0.25}, ) def test_card_telemetry_api_returns_current_pipeline_and_history(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) intent_dir = root / "docs" / "intent" intent_dir.mkdir(parents=True) (intent_dir / "my-slug.md").write_text("# Intent\n", encoding="utf-8") state_path = root / "state.json" state_path.write_text( json.dumps({"cards": {"my-slug": {"pipeline_run_id": "run-1"}}}), encoding="utf-8", ) environment = { "BOARD_STATE_FILE": str(state_path), "BOARD_INTENT_DIR": str(intent_dir), "BOARD_RESEARCH_INTENT_DIR": str(root / "docs" / "research-intent"), "BOARD_TELEMETRY_DB": str(root / "telemetry.sqlite3"), } with mock.patch.dict(os.environ, environment, clear=False): worker.telemetry.record_job( root / "telemetry.sqlite3", pipeline_run_id="run-1", slug="my-slug", job_id="job-1", role="builder", usage={"input_tokens": 10, "output_tokens": 2}, outcome={"status": "committed", "success": True}, ) payload = trigger.read_card_telemetry("my-slug") self.assertEqual(payload["pipeline_run_id"], "run-1") self.assertEqual(payload["current_pipeline"]["totals"]["total_tokens"], 12) self.assertEqual(payload["runs"][0]["pipeline_run_id"], "run-1") def test_frontend_has_separate_machine_metrics_dialog(self): board_html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn('id="metrics-dialog"', board_html) self.assertIn("function showTelemetry(card)", board_html) self.assertIn("/telemetry`", board_html) self.assertIn('"Overview","Attempts","Evidence","Conversation","Changes"', board_html) class ClarificationRepairArchitectureTests(unittest.TestCase): def inspection_block(self, **updates): result = { "schema_version": 1, "mode": "implementation_status", "repository_status": "partial", "blunix_deployment_status": "deployed", "checked_revision": "a" * 40, "repository_evidence": [{"source": "hosts/x.nix", "finding": "fehlt teilweise"}], "deployment_evidence": [{"source": "model claim", "finding": "behauptet deployed"}], "summary": "teilweise", "question": None, "required_action": None, } | updates return f"===INSPECTION_RESULT===\n{json.dumps(result)}\n===END_INSPECTION_RESULT===" def routing_block(self, **updates): result = { "schema_version": 1, "mode": "route_next", "outcome": "recommendation", "recommended_action": "review", "basis": [{"source": "commit_hash", "finding": "Commit vorhanden"}], "summary": "Review ist als Nächstes fällig", "question": None, "required_action": None, } | updates return f"===ROUTING_RESULT===\n{json.dumps(result)}\n===END_ROUTING_RESULT===" def test_three_distinct_result_schemas_have_no_target_phase(self): inspection = worker.extract_inspection_result( self.inspection_block(), {"status": "not_deployed", "source": "wrapper"}, ) self.assertEqual(inspection["repository_status"], "partial") self.assertEqual(inspection["blunix_deployment_status"], "not_deployed") self.assertTrue(inspection["deployment_status_discrepancy"]) routing = worker.extract_routing_result(self.routing_block()) self.assertEqual(routing["recommended_action"], "review") self.assertEqual(routing["outcome"], "recommendation") with self.assertRaises(ValueError): worker.extract_routing_result(self.routing_block(recommended_action="deployed")) with self.assertRaises(ValueError): worker.extract_routing_result( "===ROUTING_RESULT===\nACTION: review\nREASON: alt\n===END_ROUTING_RESULT===" ) def test_deployment_status_can_only_come_from_wrapper_evidence(self): response = self.inspection_block(repository_status="present") result = worker.extract_inspection_result( response, {"status": "unknown", "source": "trusted board wrapper state"} ) self.assertEqual(result["blunix_deployment_status"], "unknown") self.assertEqual(result["deployment_evidence"][0]["source"], "trusted board wrapper state") def test_json_result_tolerates_one_wrapping_code_fence(self): payload = self.inspection_block().split("\n", 1)[1].rsplit("\n", 1)[0] response = f"===INSPECTION_RESULT===\n```json\n{payload}\n```\n===END_INSPECTION_RESULT===" result = worker.extract_inspection_result( response, {"status": "not_deployed", "source": "wrapper"} ) self.assertEqual(result["repository_status"], "partial") with self.assertRaises(ValueError): worker.extract_inspection_result( f"===INSPECTION_RESULT===\n```json\n{payload}\n``` und noch Text\n===END_INSPECTION_RESULT===", {"status": "not_deployed", "source": "wrapper"}, ) def test_not_applicable_stays_visible_without_review_or_target_phase(self): with tempfile.TemporaryDirectory() as temp_dir: root = Path(temp_dir) state_path = root / "state.json" record = root / "board-card-1.json" worker.write_payload(record, { "job_id": "board-card-1", "pipeline_role": "fixer", "status": "no_changes", "runtime": "codex", "fixer_outcome": { "schema_version": 1, "outcome": "not_applicable", "summary": "Kein Defekt", "commit_hash": None, "pushed_branch": None, "tests": [], "question": None, "required_action": None, "review_required": False, "failure": None, }, }) with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state_path)}, clear=False): completion_watcher.process_record(record) card = json.loads(state_path.read_text(encoding="utf-8"))["cards"]["card"] self.assertEqual(card["phase"], "klaerung_reparatur") self.assertEqual(card["last_run_status"], "not_applicable") def test_routing_requires_explicit_user_apply(self): with tempfile.TemporaryDirectory() as temp_dir: state = Path(temp_dir) / "state.json" state.write_text(json.dumps({"cards": {"slug": { "phase": "klaerung_reparatur", "routing_pending": True, "routing_recommendation": "review", "routing_reason": "Commit vorhanden", }}}), encoding="utf-8") with mock.patch.dict(os.environ, {"BOARD_STATE_FILE": str(state)}, clear=False), mock.patch.object(trigger, "retry_review") as retry: self.assertEqual(trigger.apply_routing_recommendation("slug"), "review") retry.assert_called_once_with("slug") saved = json.loads(state.read_text(encoding="utf-8"))["cards"]["slug"] self.assertEqual(saved["phase"], "klaerung_reparatur") self.assertFalse(saved["routing_pending"]) def test_legacy_fixer_state_migrates_in_memory(self): migrated = trigger._migrate_clarification_state({ "fixer_attempts": 1, "fixer_question": "Datei?", "fixer_qa": [{"question": "A", "answer": "B"}], }) self.assertEqual(migrated["clarification_attempts"], 1) self.assertEqual(migrated["clarification_question"], "Datei?") self.assertEqual(len(migrated["clarification_qa"]), 1) def test_inspector_command_is_read_only_and_fresh(self): payload = { "runtime": "codex", "clarification_task_type": "route_next", "deployment_evidence": {"status": "unknown"}, } completed = SimpleNamespace( returncode=0, stderr="", stdout=json.dumps({"type": "item.completed", "item": { "type": "agent_message", "text": self.routing_block() }}) + "\n", ) environment = { "BOARD_REPO_DIR": "/repo", "BOARD_CODEX_BIN": "codex", "BOARD_ROLE_DIR": "/roles", "BOARD_CODEX_TIMEOUT_SECONDS": "10", "BOARD_HOME": "/tmp", "BOARD_CODEX_HOME": "/tmp/codex", } with mock.patch.dict(os.environ, environment, clear=False), mock.patch.object( worker.subprocess, "run", return_value=completed ) as run: worker.run_inspector("slug", payload) command = run.call_args.args[0] self.assertIn("read-only", command) self.assertNotIn("workspace-write", command) self.assertNotIn("resume", command) def test_frontend_persists_draft_and_selects_task_type(self): html = (HERE / "board.html").read_text(encoding="utf-8") self.assertIn("sessionStorage", html) self.assertIn("action.task_type", html) self.assertIn("card.available_actions", html) self.assertIn("Empfehlung anwenden", html) if __name__ == "__main__": unittest.main()