5088 lines
247 KiB
Python
5088 lines
247 KiB
Python
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('<dialog id="review-dialog" aria-labelledby="review-title">', 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/<role>."""
|
|
|
|
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": "<b>bad</b>\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()
|