#!/usr/bin/env python3
"""Tailnet-only kanban board: static frontend + trigger API for the intent
lifecycle pilot (docs/intent/kanban-board-admin-trigger.md).

Cards backed by a docs/intent/*.md file default to phase "geplant" until a
board move records a different phase in BOARD_STATE_FILE. Manually created
"idea" cards live only in that state file. Moving an intent-backed card
from "geplant" to "in_umsetzung" is the one active transition: it queues a
job for ../board-agent/worker.py (systemd-path-triggered, see
../board-agent.nix) instead of just recording the move. Every other move,
for every card, is plain bookkeeping with no side effect -- see the intent
document's explicit scope limits.
"""

import json
import state_store
import workflow
import os
import re
import sys
import threading
import time
import uuid
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import urlsplit

import fanout
import metrics
import orchestration
import telemetry

SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,60}$")
CARD_ID_RE = re.compile(r"[a-z0-9][a-z0-9-]{0,80}")
SUPPORTED_RUNTIMES = ("codex", "claude", "openrouter")
COLUMN_CONFIG_DEFAULTS = {
    "in_umsetzung": {
        "runtime": "codex",
        "model": "",
        "reasoning_effort": "",
        "iteration_limit": 20,
    },
    # The board's "leicht" lane -- its own dedicated agent-codex-builder-2
    # instance/inbox (see docs/intent/board-builder-weighted-lanes.md),
    # defaulting to a cheap OpenRouter model instead of "in_umsetzung"'s
    # Codex default. Same config shape/API, just a second column.
    "in_umsetzung_leicht": {
        "runtime": "openrouter",
        "model": "deepseek/deepseek-v3.2",
        "reasoning_effort": "",
        "iteration_limit": 20,
    },
    "review": {"runtime": "codex", "model": "", "reasoning_effort": ""},
    "recherche_laeuft": {"runtime": "codex", "model": "", "reasoning_effort": ""},
    "operator_angefordert": {"runtime": "codex", "model": "", "reasoning_effort": ""},
    "klaerung_reparatur": {
        "runtime": "codex", "model": "", "reasoning_effort": "", "iteration_limit": 20,
    },
}
MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:+/-]{0,127}$")
REASONING_EFFORT_RE = re.compile(r"^[A-Za-z0-9_-]{1,32}$")
PHASES = (
    "idea",
    "geplant",
    "in_umsetzung",
    # The board's second, dedicated Builder column ("leicht") -- see
    # COLUMN_CONFIG_DEFAULTS and docs/intent/board-builder-weighted-lanes.md.
    # Same rank as "in_umsetzung" in PHASE_RANK below: both are valid,
    # parallel successors of "geplant".
    "in_umsetzung_leicht",
    "review",
    "bereit_fuer_deployment",
    "planung_mit_feedback",
    "deployed",
    "archiviert",
    # Manual escalation column for a card stuck at the Builder iteration
    # limit even after its one automatic retry -- see
    # docs/intent/board-builder-limit-reached-auto-split.md. Deliberately
    # NOT in PHASE_RANK (see update_phase(), same treatment as archiviert):
    # it is a sibling escalation path, not a further pipeline stage.
    # "aufgeteilt" (the split outcome) is intentionally absent here -- it is
    # a worker-only terminal phase, never a legal PATCH target from the API.
    "aufteilen_angefordert",
    "operator_angefordert",
    "operator_laeuft",
    "wartet_auf_kinder",
    "durch_kinder_blockiert",
    # Distinct from "durch_kinder_blockiert": this is set when the Operator
    # step itself never produced children -- either a considered
    # needs_human decision or the Operator job raising an exception -- so
    # "blocked by children" would be a lie (there are none). See
    # orchestration.apply_needs_human() and worker.py's operator except
    # branch.
    "operator_klaerung_noetig",
    "fixer_angefordert",
    "fixer_laeuft",
    "fixer_rueckfrage",
    "fixer_fehlgeschlagen",
    "klaerung_reparatur",
    "inspector_laeuft",
    "inspector_rueckfrage",
    "inspector_fehlgeschlagen",
)
CLARIFICATION_TASK_TYPES = {"implementation_status", "route_next", "repair"}
ROUTING_ACTION_PHASES = {
    "geplant", "in_umsetzung", "in_umsetzung_leicht", "review",
    "archiviert", "recherche_angefordert", "recherche_laeuft",
}

# Pipeline rank per phase, used only to reject forward moves that skip a
# stage. bereit_fuer_deployment and planung_mit_feedback share a rank: both
# are valid successors of review and valid predecessors of deployed (see
# docs/intent/board-intent-detail-and-transition-guardrails.md). Backward
# moves (new rank <= current rank) are never guarded.
PHASE_RANK = {
    "idea": 0,
    "geplant": 1,
    "in_umsetzung": 2,
    "in_umsetzung_leicht": 2,
    "review": 3,
    "bereit_fuer_deployment": 4,
    "planung_mit_feedback": 4,
    "deployed": 5,
}
RESEARCH_PHASES = (
    "recherche_angefordert",
    "recherche_laeuft",
    "recherche_fertig",
    "recherche_fehlgeschlagen",
    "archiviert",
)
RESEARCH_PHASE_RANK = {
    "recherche_angefordert": 0,
    "recherche_laeuft": 1,
    "recherche_fertig": 2,
    "recherche_fehlgeschlagen": 2,
}

# Single-process ThreadingHTTPServer: this lock keeps a create/patch/delete
# from interleaving with another one, which is what makes the "only from
# geplant" trigger guard in update_phase() below actually safe against a
# double-drag race instead of just usually working.
STATE_LOCK = state_store.StateLock()


def validate_runtime(runtime):
    value = str(runtime or "").strip().lower()
    if value not in SUPPORTED_RUNTIMES:
        raise ValueError("Runtime muss exakt codex, claude oder openrouter sein.")
    return value


def board_runtime():
    return validate_runtime(load_state().get("runtime", os.environ.get("BOARD_RUNTIME_DEFAULT", "codex")))


def set_board_runtime(runtime):
    runtime = validate_runtime(runtime)
    with STATE_LOCK:
        state = load_state()
        state["runtime"] = runtime
        save_state(state)
    return runtime


def validate_column_config_value(column, payload):
    if column not in COLUMN_CONFIG_DEFAULTS:
        raise ValueError("unbekannte Spalte")
    if not isinstance(payload, dict):
        raise ValueError("body must be a JSON object")
    allowed = COLUMN_CONFIG_DEFAULTS[column]
    unknown = set(payload) - set(allowed)
    if unknown:
        raise ValueError(f"unbekanntes Feld: {sorted(unknown)[0]}")
    validated = {}
    if "runtime" in payload:
        validated["runtime"] = validate_runtime(payload["runtime"])
    if "model" in payload:
        value = str(payload["model"])
        if value and not MODEL_RE.fullmatch(value):
            raise ValueError("ungültiges model")
        if len(value) > 128:
            raise ValueError("ungültiges model")
        validated["model"] = value
    if "reasoning_effort" in payload:
        value = str(payload["reasoning_effort"])
        if value and not REASONING_EFFORT_RE.fullmatch(value):
            raise ValueError("ungültiges reasoning_effort")
        validated["reasoning_effort"] = value
    if "iteration_limit" in payload:
        value = payload["iteration_limit"]
        if isinstance(value, bool) or not isinstance(value, int) or not 5 <= value <= 40:
            raise ValueError("iteration_limit muss eine Ganzzahl zwischen 5 und 40 sein")
        validated["iteration_limit"] = value
    return validated


def load_column_config():
    stored = load_state().get("column_config", {})
    result = {}
    for column, defaults in COLUMN_CONFIG_DEFAULTS.items():
        values = stored.get(column, {}) if isinstance(stored, dict) else {}
        result[column] = {
            field: values.get(field, default) for field, default in defaults.items()
        }
    return result


def set_column_config(column, updates):
    validated = validate_column_config_value(column, updates)
    with STATE_LOCK:
        state = load_state()
        config = load_column_config()
        config[column].update(validated)
        state["column_config"] = config
        save_state(state)
    return config[column]


def now_iso():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def state_file():
    return Path(os.environ["BOARD_STATE_FILE"])


def load_state():
    return state_store.load()


def save_state(data):
    with state_store.transaction():
        state_store.save(data)


def intent_title(path):
    try:
        for line in path.read_text(encoding="utf-8").splitlines():
            stripped = line.strip()
            if stripped.startswith("# "):
                return stripped[2:].strip()
    except OSError:
        pass
    return path.stem


def list_intent_cards():
    intent_dir = Path(os.environ["BOARD_INTENT_DIR"])
    cards = {}
    if not intent_dir.is_dir():
        return cards
    for md in sorted(intent_dir.glob("*.md")):
        slug = md.stem
        if not SLUG_RE.match(slug):
            continue
        cards[slug] = {
            "id": slug,
            "source": "intent",
            "title": intent_title(md),
            "intent_path": f"docs/intent/{md.name}",
        }
    return cards


def list_research_cards():
    research_dir = Path(
        os.environ.get(
            "BOARD_RESEARCH_INTENT_DIR",
            str(Path(os.environ["BOARD_INTENT_DIR"]).parent / "research-intent"),
        )
    )
    cards = {}
    if not research_dir.is_dir():
        return cards
    for md in sorted(research_dir.glob("*.md")):
        slug = md.stem
        if not SLUG_RE.match(slug):
            continue
        cards[slug] = {
            "id": slug,
            "source": "research",
            "title": intent_title(md),
            "intent_path": f"docs/research-intent/{md.name}",
        }
    return cards


def stored_intent_card(card_id, stored):
    """Return a validated state-only card for an intent whose file is absent."""
    slug = card_id
    path = f"docs/intent/{slug}.md"
    if (
        stored.get("source") != "intent"
        or not SLUG_RE.fullmatch(slug)
    ):
        return None
    return {
        "id": slug,
        "source": "intent",
        "title": str(stored.get("title") or slug),
        "intent_path": path,
        "source_missing": True,
    }


def escalation_exhausted(stored):
    """True only when the most recent Builder limit is terminal."""
    return (
        stored.get("last_run_status") == "limit_reached"
        and stored.get("auto_retry_attempts", 0)
        >= len(orchestration.escalation_ladder_for(stored))
    )


def is_family_parent(stored):
    """Recognize a family parent even after clarification cleared blockers."""
    return bool(
        stored.get("orchestration_kind") == "operator"
        or stored.get("child_slugs")
        or stored.get("depends_on")
        or stored.get("blocking_children")
    )


def repair_eligible(stored):
    """Keep expected/in-flight Builder escalation out of the Fixer path."""
    failed_status = stored.get("last_run_status") in {
        "failed", "check_failed", "deploy_failed",
    }
    return (
        int(stored.get("fixer_attempts", 0)) < 2
        and not is_family_parent(stored)
        and (
            stored.get("review_decision") == "planung_mit_feedback"
            or stored.get("phase") in {
                "planung_mit_feedback", "fixer_fehlgeschlagen", "inspector_fehlgeschlagen",
            }
            or failed_status
            or escalation_exhausted(stored)
        )
    )


def available_actions(card_id, source, stored, job_active=False):
    phase = stored.get("phase", "recherche_angefordert" if source == "research" else "geplant" if source == "intent" else "idea")
    if job_active:
        return []
    actions = []
    def action(id, label, phase, **extra):
        actions.append({"id": id, "label": label, "phase": phase, **extra})
    if phase == "archiviert":
        action("reactivate", "Reactivate", "recherche_angefordert" if source == "research" else "geplant" if source == "intent" else "idea")
        return actions
    if source == "idea":
        actions.append({"id": "plan", "label": "Refine with Planner", "endpoint": "plan"})
    if source == "intent" and phase == "geplant" and not stored.get("blocked_by"):
        if not stored.get("operator_generated") and SLUG_RE.fullmatch(f"{card_id}-synthese"):
            action("operator", "Ask Operator", "operator_angefordert")
        action("builder", "Start Builder", "in_umsetzung")
        action("builder-light", "Start Builder (light)", "in_umsetzung_leicht")
    if source == "research" and phase == "recherche_angefordert":
        action("research", "Start Research", "recherche_laeuft")
    if stored.get("failure_stage") == "deployment" and stored.get("review_decision") == "bereit_fuer_deployment" and stored.get("commit_hash"):
        actions.append({"id": "retry-deployment", "label": "Retry deployment", "endpoint": "retry-deployment"})
    if phase != "bereit_fuer_deployment" and not workflow.verified_deployment(stored):
        action("diagnose", "Diagnose", "klaerung_reparatur", task_type="implementation_status")
        action("route", "Choose next step", "klaerung_reparatur", task_type="route_next")
        if repair_eligible(stored) and stored.get("failure_stage") not in {"inspector", "pre_start", "review_config", "publication", "deployment"}:
            action("repair", "Repair", "klaerung_reparatur", task_type="repair")
        if source == "intent" and escalation_exhausted(stored) and not stored.get("operator_generated"):
            action("split", "Split", "aufteilen_angefordert")
        if source == "intent" and stored.get("commit_hash") and stored.get("last_run_status") in {"failed", "check_failed"}:
            actions.append({"id": "retry-review", "label": "Retry Review", "endpoint": "retry-review"})
        if source == "research" and phase != "recherche_angefordert":
            action("reset", "Return to research queue", "recherche_angefordert")
        elif source == "intent" and phase != "geplant":
            action("reset", "Return to Ready", "geplant")
    action("archive", "Archive", "archiviert")
    return actions


def merged_board():
    with STATE_LOCK:
        state = load_state()
        intent_cards = list_intent_cards()
        research_cards = list_research_cards()
        board = []
        job_active = (
            repo_job_active()
            if all(os.environ.get(name) for name in ("BOARD_INBOX_DIR", "BOARD_PROCESSING_DIR"))
            else False
        )
        seen = set()
        for slug, card in intent_cards.items():
            stored = state["cards"].get(slug, {})
            board.append(
                {
                    **card,
                    "phase": stored.get("phase", "geplant"),
                    "updated_at": stored.get("updated_at"),
                    "last_job_id": stored.get("last_job_id"),
                    "last_run_status": stored.get("last_run_status"),
                    "last_run_at": stored.get("last_run_at"),
                    "review_decision": stored.get("review_decision"),
                    "review_report": stored.get("review_report"),
                    "commit_hash": stored.get("commit_hash"),
                    "builder_report": stored.get("builder_report"),
                    "failure_stage": stored.get("failure_stage"),
                    "failure_detail": stored.get("failure_detail"),
                    "research_report": stored.get("research_report"),
                    "parent_id": stored.get("parent_id"),
                    "child_slugs": stored.get("child_slugs"),
                    "depends_on": stored.get("depends_on"),
                    "synthesis_generated": stored.get("synthesis_generated", False),
                    "operator_decision": stored.get("operator_decision"),
                    "operator_reason": stored.get("operator_reason"),
                    "human_action": stored.get("human_action"),
                    "operator_parent_id": stored.get("operator_parent_id"),
                    "blocking_children": stored.get("blocking_children"),
                    "blocked_by": stored.get("blocked_by"),
                    "wave": stored.get("wave"),
                    "recommended_lane": stored.get("recommended_lane"),
                    "recommended_profile": stored.get("recommended_profile"),
                    "lane": stored.get("lane"),
                    "runtime": stored.get("runtime"),
                    "model": stored.get("model"),
                    "reasoning_effort": stored.get("reasoning_effort"),
                    "synthesis_status": stored.get("synthesis_status"),
                    "write_scope": stored.get("write_scope"),
                    "acceptance_criteria": stored.get("acceptance_criteria"),
                    "metrics_summary": metrics.summary(stored),
                    "pipeline_run_id": stored.get("pipeline_run_id"),
                    "auto_retry_attempts": stored.get("auto_retry_attempts", 0),
                    "escalation_attempts": stored.get("escalation_attempts", []),
                    "iteration_limit": stored.get("iteration_limit"),
                    "available_actions": available_actions(slug, "intent", stored, job_active),
                    "fixer_attempts": stored.get("fixer_attempts", 0),
                    "fixer_question": stored.get("fixer_question"),
                    "fixer_qa": stored.get("fixer_qa", []),
                    "fixer_outcome": stored.get("fixer_outcome"),
                    "fixer_target_phase": stored.get("fixer_target_phase"),
                    "clarification_task_type": stored.get("clarification_task_type"),
                    "clarification_question": stored.get("clarification_question") or stored.get("fixer_question"),
                    "repo_status": stored.get("repo_status"),
                    "repo_evidence": stored.get("repo_evidence"),
                    "deployment_evidence": stored.get("deployment_evidence"),
                    "routing_recommendation": stored.get("routing_recommendation"),
                    "routing_reason": stored.get("routing_reason"),
                    "routing_pending": stored.get("routing_pending", False),
                    "deployment_status": stored.get("deployment_status"),
                    "source_missing": False,
                }
            )
            seen.add(slug)
        for slug, card in research_cards.items():
            stored = state["cards"].get(slug, {})
            board.append({
                **card,
                "phase": stored.get("phase", "recherche_angefordert"),
                "updated_at": stored.get("updated_at"),
                "last_job_id": stored.get("last_job_id"),
                "last_run_status": stored.get("last_run_status"),
                "last_run_at": stored.get("last_run_at"),
                "failure_stage": stored.get("failure_stage"),
                "failure_detail": stored.get("failure_detail"),
                "research_report": stored.get("research_report"),
                "parent_id": stored.get("parent_id"),
                "depends_on": stored.get("depends_on"),
                "synthesis_generated": stored.get("synthesis_generated", False),
                "metrics_summary": metrics.summary(stored),
                "pipeline_run_id": stored.get("pipeline_run_id"),
                "auto_retry_attempts": stored.get("auto_retry_attempts", 0),
                "escalation_attempts": stored.get("escalation_attempts", []),
                "iteration_limit": stored.get("iteration_limit"),
                "available_actions": available_actions(slug, "research", stored, job_active),
                "fixer_attempts": stored.get("fixer_attempts", 0),
                "fixer_question": stored.get("fixer_question"),
                "fixer_qa": stored.get("fixer_qa", []),
                "fixer_target_phase": stored.get("fixer_target_phase"),
                "clarification_task_type": stored.get("clarification_task_type"),
                "clarification_question": stored.get("clarification_question") or stored.get("fixer_question"),
                "routing_recommendation": stored.get("routing_recommendation"),
                "routing_reason": stored.get("routing_reason"),
                "routing_pending": stored.get("routing_pending", False),
                "source_missing": False,
            })
            seen.add(slug)
        for card_id, stored in state["cards"].items():
            if card_id in seen:
                continue
            fallback = stored_intent_card(card_id, stored)
            if fallback:
                board.append(
                    {
                        **fallback,
                        "phase": stored.get("phase", "geplant"),
                        "updated_at": stored.get("updated_at"),
                        "last_job_id": stored.get("last_job_id"),
                        "last_run_status": stored.get("last_run_status"),
                        "last_run_at": stored.get("last_run_at"),
                        "review_decision": stored.get("review_decision"),
                        "review_report": stored.get("review_report"),
                        "commit_hash": stored.get("commit_hash"),
                        "builder_report": stored.get("builder_report"),
                        "failure_stage": stored.get("failure_stage"),
                        "failure_detail": stored.get("failure_detail"),
                        "parent_id": stored.get("parent_id"),
                        "child_slugs": stored.get("child_slugs"),
                        "depends_on": stored.get("depends_on"),
                        "synthesis_generated": stored.get("synthesis_generated", False),
                        "operator_decision": stored.get("operator_decision"),
                        "operator_reason": stored.get("operator_reason"),
                        "human_action": stored.get("human_action"),
                        "operator_parent_id": stored.get("operator_parent_id"),
                        "blocking_children": stored.get("blocking_children"),
                        "blocked_by": stored.get("blocked_by"),
                        "wave": stored.get("wave"),
                        "recommended_lane": stored.get("recommended_lane"),
                        "recommended_profile": stored.get("recommended_profile"),
                        "lane": stored.get("lane"),
                        "runtime": stored.get("runtime"),
                        "model": stored.get("model"),
                        "reasoning_effort": stored.get("reasoning_effort"),
                        "synthesis_status": stored.get("synthesis_status"),
                        "write_scope": stored.get("write_scope"),
                        "acceptance_criteria": stored.get("acceptance_criteria"),
                        "metrics_summary": metrics.summary(stored),
                        "pipeline_run_id": stored.get("pipeline_run_id"),
                        "auto_retry_attempts": stored.get("auto_retry_attempts", 0),
                        "escalation_attempts": stored.get("escalation_attempts", []),
                        "iteration_limit": stored.get("iteration_limit"),
                        "available_actions": available_actions(card_id, "intent", stored, job_active),
                        "fixer_attempts": stored.get("fixer_attempts", 0),
                        "fixer_question": stored.get("fixer_question"),
                        "fixer_qa": stored.get("fixer_qa", []),
                        "fixer_target_phase": stored.get("fixer_target_phase"),
                    }
                )
                continue
            if stored.get("source") == "draft":
                board.append(
                    {
                        "id": card_id,
                        "source": "draft",
                        "title": stored.get("title", card_id),
                        "phase": stored.get("phase", "entwurf_laeuft"),
                        "updated_at": stored.get("updated_at"),
                        "last_run_status": stored.get("last_run_status"),
                        "draft_conversation": stored.get("draft_conversation", []),
                        "failure_detail": stored.get("failure_detail"),
                        "available_actions": [],
                    }
                )
                continue
            if stored.get("source") != "idea":
                continue
            board.append(
                {
                    "id": card_id,
                    "source": "idea",
                    "title": stored.get("title", card_id),
                    "phase": stored.get("phase", "idea"),
                    "updated_at": stored.get("updated_at"),
                    "available_actions": available_actions(card_id, "idea", stored, job_active),
                    "fixer_attempts": stored.get("fixer_attempts", 0),
                    "fixer_question": stored.get("fixer_question"),
                    "fixer_qa": stored.get("fixer_qa", []),
                    "fixer_target_phase": stored.get("fixer_target_phase"),
                }
            )
        active_slugs = orchestration.active_slugs()
        for card in board:
            stored = state["cards"].get(card["id"], {})
            for key in ("inspector_result", "repo_status", "repo_evidence", "deployment_evidence",
                        "deployment_status", "deployed_commit_hash", "activated_system",
                        "draft_conversation", "draft_origin", "attempt_history", "review_history",
                        "clarification_qa", "clarification_required_action", "deployment_error", "preserved_worktree"):
                card[key] = stored.get(key)
        for card in board:
            stored = state["cards"].get(card["id"], {})
            active = card["id"] in active_slugs
            if card["source"] != "draft":
                card["available_actions"] = available_actions(card["id"], card["source"], {**stored, "phase": card["phase"]}, active)
            children = [child for child in board if child.get("parent_id") == card["id"]]
            card["display"] = workflow.presentation({**stored, **card}, children, active)
        return board


SMOKE_TEST_DIRECT_TEMPLATE = """\
# Intent: Operator Smoke-Test {n} (klein, direct_builder erwartet)

## Ziel

Installiere `ripgrep` systemweit auf dem NixOS-Host `blunix`. Dies ist ein
kleiner, präziser Auftrag, den der Operator als `direct_builder` einstufen
und automatisch in die empfohlene Builder-Pipeline schicken soll.

Diese Karte auf „Bereit für Builder" anlegen und dann in die
Operator-Spalte ziehen (`operator_angefordert`).

## Auftrag an den Builder

- Ergänze ausschließlich `ripgrep` in der bestehenden
  `environment.systemPackages`-Liste in
  `hosts/blunix/localServices/packages.nix`.
- Sortiere den neuen Eintrag alphabetisch in die Liste ein.
- Ändere keine andere Datei und nimm keine weiteren Paket- oder
  Konfigurationsänderungen vor.

## Erwartetes Ergebnis

Operator liefert `direct_builder`, ein Builder-Lauf erzeugt genau einen
Commit, der ausschließlich `ripgrep` zu den systemweiten Paketen von
`blunix` hinzufügt, danach normaler Review-Übergang.

Dieser Intent ist ausschließlich ein technischer Pipeline-Test.
"""

SMOKE_TEST_SPLIT_TEMPLATE = """\
# Intent: Operator Smoke-Test {n} (groß, split erwartet)

## Ziel

Installiere `ripgrep` und `fd` systemweit auf dem NixOS-Host `blunix`.
Der Operator soll diesen absichtlich aus zwei unabhängigen Paketänderungen
bestehenden Auftrag als `split` einstufen, exakt zwei Kinder automatisch
starten und anschließend die Synthese durchlaufen lassen.

Diese Karte auf „Bereit für Builder" anlegen und dann in die
Operator-Spalte ziehen (`operator_angefordert`).

## Auftrag (an den Operator, zur Zerlegung in Builder-Kinder)

Zerlege den Parent in exakt zwei vollständige Builder-Kinder mit
unterschiedlichen, um `-{n}` ergänzten Slugs. Beide Kinder dürfen in
derselben Welle (`W1`) laufen, da ihre Schreibbereiche nicht überlappen:

1. Das erste Kind ergänzt ausschließlich `ripgrep`, alphabetisch
   einsortiert, in der bestehenden `environment.systemPackages`-Liste in
   `hosts/blunix/localServices/packages.nix`. Sein `write_scope` enthält
   genau diesen Pfad.
2. Das zweite Kind ergänzt ausschließlich `fd`, alphabetisch einsortiert,
   in der bestehenden `environment.systemPackages`-Liste in
   `hosts/blunix/localServices/packages-stable.nix`. Sein `write_scope`
   enthält genau diesen Pfad.

Jedes Kind ändert nur seine genannte Datei und nimmt keine weiteren
Paket- oder Konfigurationsänderungen vor. Zwischen den Kindern besteht keine
Abhängigkeit (`blocked_by` jeweils leer).

## Erwartetes Ergebnis

Operator liefert `split` mit exakt zwei Kindern. Beide starten automatisch;
nach erfolgreichem Review beider Kinder erzeugt das Board die Synthese, die
selbst automatisch durchläuft. Das resultierende System installiert sowohl
`ripgrep` als auch `fd` systemweit auf `blunix`.

Dieser Intent ist ausschließlich ein technischer Pipeline-Test.
"""

SMOKE_TEST_REMOVE_DIRECT_TEMPLATE = """\
# Intent: Operator Deinstallations-Smoke-Test {n} (klein, direct_builder erwartet)

## Ziel

Deinstalliere `ripgrep` systemweit vom NixOS-Host `blunix`. Dies ist das
genaue Gegenstück zum kleinen Installations-Smoke-Test und soll vom Operator
als `direct_builder` eingestuft und automatisch in die empfohlene
Builder-Pipeline geschickt werden.

Diese Karte auf „Bereit für Builder" anlegen und dann in die
Operator-Spalte ziehen (`operator_angefordert`).

## Auftrag an den Builder

- Entferne ausschließlich `ripgrep` aus der bestehenden
  `environment.systemPackages`-Liste in
  `hosts/blunix/localServices/packages.nix`.
- Ändere keine andere Datei und nimm keine weiteren Paket- oder
  Konfigurationsänderungen vor.

## Erwartetes Ergebnis

Operator liefert `direct_builder`, ein Builder-Lauf erzeugt genau einen
Commit, der ausschließlich `ripgrep` aus den systemweiten Paketen von
`blunix` entfernt, danach normaler Review-Übergang.

Dieser Intent ist ausschließlich ein technischer Pipeline-Test.
"""

SMOKE_TEST_REMOVE_SPLIT_TEMPLATE = """\
# Intent: Operator Deinstallations-Smoke-Test {n} (groß, split erwartet)

## Ziel

Deinstalliere `ripgrep` und `fd` systemweit vom NixOS-Host `blunix`. Dies ist
das genaue Gegenstück zum großen Installations-Smoke-Test. Der Operator soll
den Auftrag als `split` einstufen, exakt zwei Kinder automatisch starten und
anschließend die Synthese durchlaufen lassen.

Diese Karte auf „Bereit für Builder" anlegen und dann in die
Operator-Spalte ziehen (`operator_angefordert`).

## Auftrag (an den Operator, zur Zerlegung in Builder-Kinder)

Zerlege den Parent in exakt zwei vollständige Builder-Kinder mit
unterschiedlichen, um `-{n}` ergänzten Slugs. Beide Kinder dürfen in
derselben Welle (`W1`) laufen, da ihre Schreibbereiche nicht überlappen:

1. Das erste Kind entfernt ausschließlich `ripgrep` aus der bestehenden
   `environment.systemPackages`-Liste in
   `hosts/blunix/localServices/packages.nix`. Sein `write_scope` enthält
   genau diesen Pfad.
2. Das zweite Kind entfernt ausschließlich `fd` aus der bestehenden
   `environment.systemPackages`-Liste in
   `hosts/blunix/localServices/packages-stable.nix`. Sein `write_scope`
   enthält genau diesen Pfad.

Jedes Kind ändert nur seine genannte Datei und nimmt keine weiteren
Paket- oder Konfigurationsänderungen vor. Zwischen den Kindern besteht keine
Abhängigkeit (`blocked_by` jeweils leer).

## Erwartetes Ergebnis

Operator liefert `split` mit exakt zwei Kindern. Beide starten automatisch;
nach erfolgreichem Review beider Kinder erzeugt das Board die Synthese, die
selbst automatisch durchläuft. Das resultierende System installiert weder
`ripgrep` noch `fd` systemweit auf `blunix`.

Dieser Intent ist ausschließlich ein technischer Pipeline-Test.
"""

SMOKE_TEST_KINDS = {
    "direct": ("board-operator-smoke-test-direct", SMOKE_TEST_DIRECT_TEMPLATE),
    "split": ("board-operator-smoke-test-split", SMOKE_TEST_SPLIT_TEMPLATE),
    "remove-direct": (
        "board-operator-smoke-test-remove-direct",
        SMOKE_TEST_REMOVE_DIRECT_TEMPLATE,
    ),
    "remove-split": (
        "board-operator-smoke-test-remove-split",
        SMOKE_TEST_REMOVE_SPLIT_TEMPLATE,
    ),
}


def create_smoke_test(kind):
    """Write a fresh, numbered install or removal smoke-test intent.

    The board exposes direct_builder and split variants of both operations.
    Separate prefixes prevent collisions between kinds; for split tests, the
    Operator derives fresh child slugs from the numbered intent text.
    """
    if kind not in SMOKE_TEST_KINDS:
        raise ValueError("unbekannte Smoke-Test-Art")
    prefix, template = SMOKE_TEST_KINDS[kind]
    intent_dir = Path(os.environ["BOARD_INTENT_DIR"])
    intent_dir.mkdir(parents=True, exist_ok=True)
    existing = {int(m.group(1)) for name in intent_dir.glob(f"{prefix}-*.md") if (m := re.fullmatch(rf"{prefix}-(\d+)\.md", name.name))}
    n = max(existing, default=0) + 1
    slug = f"{prefix}-{n}"
    (intent_dir / f"{slug}.md").write_text(template.format(n=n), encoding="utf-8")
    return slug


SMOKE_TEST_MARKER_RE = re.compile(r"smoke", re.IGNORECASE)


def _smoke_test_directories():
    repo_root = Path(os.environ["BOARD_REPO_DIR"])
    return {
        "docs/intent": Path(os.environ["BOARD_INTENT_DIR"]),
        "docs/research-intent": Path(os.environ["BOARD_RESEARCH_INTENT_DIR"]),
        "docs/review": repo_root / "docs" / "review",
        "docs/research": repo_root / "docs" / "research",
    }


def _expand_card_family(seeds, cards):
    """Follow parent/child/synthesis links outward from seeds until the
    selection stops growing -- a smoke test's split children, fan-out
    children and generated synthesis card never carry "smoke" in their own
    slug or title, so a name match alone would leave them behind.
    """
    selected = set(seeds)
    changed = True
    while changed:
        changed = False
        for card_id, card in cards.items():
            refs = set(card.get("depends_on") or []) | set(card.get("child_slugs") or [])
            refs.update(
                ref for ref in (
                    card.get("parent_id"),
                    card.get("operator_parent_id"),
                    card.get("synthesis_for"),
                )
                if ref
            )
            if card_id in selected or refs & selected:
                before = len(selected)
                selected.add(card_id)
                selected.update(ref for ref in refs if ref in cards)
                if len(selected) > before:
                    changed = True
    return selected


def delete_smoke_test_cards():
    """Remove every smoke-test card, and its whole linked family (split
    children, fan-out children, a generated synthesis card), from the
    board at once.

    This is the same cleanup an operator previously had to script by hand
    after a round of pipeline smoke testing, now reachable from the board
    itself. Refuses while any board job is queued or running -- the same
    repo_job_active() guard research transitions already use -- so this
    can never delete an intent file out from under a job that is still
    reading it. Backing docs/intent, docs/research-intent, docs/review and
    docs/research files are deleted outright: these are throwaway
    pipeline-test artifacts, not files anyone edits by hand, and the
    calling UI confirms before sending this request.
    """
    if repo_job_active():
        raise PermissionError("Ein Board-Job läuft oder ist eingereiht")

    directories = _smoke_test_directories()
    with STATE_LOCK:
        state = load_state()
        cards = state["cards"]
        seeds = {
            card_id for card_id, card in cards.items()
            if SMOKE_TEST_MARKER_RE.search(f"{card_id} {card.get('title', '')}")
        }
        for directory in directories.values():
            if directory.is_dir():
                seeds.update(
                    path.stem for path in directory.glob("*.md")
                    if SMOKE_TEST_MARKER_RE.search(path.stem)
                )

        selected = _expand_card_family(seeds, cards)

        removed_files = 0
        for directory in directories.values():
            if not directory.is_dir():
                continue
            directory_resolved = directory.resolve(strict=True)
            for slug in selected:
                if not SLUG_RE.fullmatch(slug):
                    continue
                candidate = directory / f"{slug}.md"
                if not candidate.is_file():
                    continue
                resolved = candidate.resolve(strict=True)
                try:
                    resolved.relative_to(directory_resolved)
                except ValueError:
                    continue
                if resolved != directory_resolved / f"{slug}.md":
                    continue
                try:
                    resolved.unlink()
                    removed_files += 1
                except OSError as error:
                    # A single unwritable file (e.g. a sandbox directory
                    # missing from this service's ReadWritePaths) must not
                    # take the whole request down -- verified live
                    # 2026-09-10: an uncaught OSError here killed the
                    # connection mid-response, which Traefik then surfaced
                    # as a 502. State/telemetry cleanup below still runs
                    # for every selected slug regardless.
                    print(f"Smoke-Test-Datei konnte nicht entfernt werden ({resolved}): {error}", file=sys.stderr)

        removed_cards = sum(1 for slug in selected if cards.pop(slug, None) is not None)
        save_state(state)

    try:
        telemetry.delete_slugs(Path(os.environ["BOARD_TELEMETRY_DB"]), selected)
    except (OSError, ValueError) as error:
        print(f"Telemetrie-Bereinigung nach Smoke-Test-Löschung fehlgeschlagen: {error}", file=sys.stderr)

    return {"removed_cards": removed_cards, "removed_files": removed_files}


def create_idea(title):
    title = title.strip()
    if not title or len(title) > 200:
        raise ValueError("invalid title")
    card_id = "idea-" + uuid.uuid4().hex[:12]
    with STATE_LOCK:
        state = load_state()
        state["cards"][card_id] = {
            "source": "idea",
            "title": title,
            "phase": "idea",
            "updated_at": now_iso(),
        }
        save_state(state)
    return card_id


def delete_manual_card(card_id):
    """Delete a card that lives only in state.json -- a manual "idea"
    backlog card, or an abandoned interactive-intake "draft" that never
    finished. A *finished* draft is no longer a "draft" source by the
    time this could run: record_draft_intent() replaces it with a normal
    "geplant" intent card, so this path never touches it."""
    with STATE_LOCK:
        state = load_state()
        card = state["cards"].get(card_id)
        if not card or card.get("source") not in ("idea", "draft"):
            raise LookupError("not a manually created card")
        del state["cards"][card_id]
        save_state(state)


MAX_DRAFT_MESSAGE_LENGTH = 4000
MAX_DRAFT_CONVERSATION_TURNS = 40


def _queue_draft_job(state, card_id, conversation):
    """Write a job_type="draft" job and update the card's own state entry
    in place. Assumes the caller already holds STATE_LOCK and has already
    called load_state() into `state` -- same pure state-mutation-helper
    pattern as _queue_clarification() above. Shared by create_draft()
    (the opening turn), answer_draft() and retry_draft() (a follow-up
    turn) below.
    """
    stored = state["cards"].setdefault(card_id, {})
    job_id = write_job(
        card_id, stored.get("title", card_id), "", job_type="draft",
        conversation=conversation,
    )
    stored.update(
        source="draft", phase="entwurf_laeuft", last_job_id=job_id,
        last_run_status="queued_draft", draft_conversation=conversation,
        updated_at=now_iso(),
    )
    return job_id


def create_draft(topic):
    """Start a new interactive Planner intake dialogue: a draft card with
    no backing file yet, its opening turn queued immediately.

    Deliberately no repo_job_active() guard, unlike research/fixer
    transitions: an ordinary draft turn never touches the shared checkout
    (see worker.py's run_draft()), so it can run alongside a Builder/
    Review/Research/Split job without contending for anything. Only the
    single narrow moment where a finished dialogue writes its new intent
    file takes the repo lock -- see worker.py's job_type == "draft"
    handling.
    """
    topic = str(topic or "").strip()
    if not topic or len(topic) > 200:
        raise ValueError("ein Thema mit maximal 200 Zeichen ist erforderlich")
    card_id = "draft-" + uuid.uuid4().hex[:12]
    conversation = [{"role": "user", "text": topic}]
    with STATE_LOCK:
        state = load_state()
        state["cards"][card_id] = {"title": topic}
        _queue_draft_job(state, card_id, conversation)
        save_state(state)
    return card_id


def plan_card(card_id):
    """Refine an idea in place, preserving its identity."""
    with STATE_LOCK:
        state = load_state()
        stored = state["cards"].get(card_id, {})
        if stored.get("source") != "idea" or stored.get("phase") == "archiviert":
            raise PermissionError("Only active ideas can start this planning conversation.")
        _queue_draft_job(state, card_id, [{"role": "user", "text": stored["title"]}])
        save_state(state)
        return card_id


def retry_deployment(card_id):
    with STATE_LOCK:
        state = load_state()
        card = state["cards"].get(card_id, {})
        if (card.get("phase") == "archiviert" or card.get("failure_stage") != "deployment"
                or card.get("review_decision") != "bereit_fuer_deployment" or not card.get("commit_hash")
                or card_id in orchestration.active_slugs()):
            raise PermissionError("Only an inactive, failed deployment of an approved commit can be retried.")
        job_id = write_job(card_id, card.get("title", card_id), card.get("intent_path", ""),
                           job_type="retry_deployment", commit_hash=card["commit_hash"],
                           pipeline_run_id=card.get("pipeline_run_id"))
        card.update(last_job_id=job_id, last_run_status="queued", updated_at=now_iso())
        save_state(state)
        return job_id


def answer_draft(card_id, answer):
    """Answer the Planner's latest draft question and queue the next turn."""
    with STATE_LOCK:
        state = load_state()
        stored = state["cards"].get(card_id)
        if not stored or stored.get("source") != "draft":
            raise LookupError("unknown draft card")
        if stored.get("phase") != "entwurf_frage":
            raise PermissionError("Antwort ist nur bei einer offenen Rückfrage möglich")
        answer = str(answer or "").strip()
        if not answer or len(answer) > MAX_DRAFT_MESSAGE_LENGTH:
            raise ValueError(f"eine Antwort mit maximal {MAX_DRAFT_MESSAGE_LENGTH} Zeichen ist erforderlich")
        conversation = [*(stored.get("draft_conversation") or []), {"role": "user", "text": answer}]
        if len(conversation) > MAX_DRAFT_CONVERSATION_TURNS:
            raise PermissionError(
                "Konversation ist zu lang -- als Intent per Hand anlegen oder neu beginnen"
            )
        job_id = _queue_draft_job(state, card_id, conversation)
        save_state(state)
    return job_id


def retry_draft(card_id):
    """Re-run the last Planner turn unchanged -- for a draft whose last
    turn failed (a runtime error, a malformed response, a slug collision
    at finalization), since a retry needs no new operator input."""
    with STATE_LOCK:
        state = load_state()
        stored = state["cards"].get(card_id)
        if not stored or stored.get("source") != "draft":
            raise LookupError("unknown draft card")
        if stored.get("phase") != "entwurf_fehlgeschlagen":
            raise PermissionError("Erneuter Versuch ist nur nach einem fehlgeschlagenen Lauf möglich")
        conversation = stored.get("draft_conversation") or []
        job_id = _queue_draft_job(state, card_id, conversation)
        save_state(state)
    return job_id


def read_review_report(slug):
    """Read only the fixed report belonging to an existing intent card.

    The request contributes a validated slug, never a filesystem path. The
    state entry must contain the exact path the worker creates, and resolving
    that path must keep it inside this repository's docs/review directory.
    """
    if not SLUG_RE.fullmatch(slug) or slug not in list_intent_cards():
        raise LookupError("unknown intent card")

    expected_path = f"docs/review/{slug}.md"
    with STATE_LOCK:
        stored = load_state()["cards"].get(slug, {})
        if stored.get("review_report") != expected_path:
            raise LookupError("review report not found")

    repo_root = Path(os.environ["BOARD_REPO_DIR"]).resolve(strict=True)
    review_root = (repo_root / "docs" / "review").resolve(strict=True)
    report_path = repo_root / expected_path
    resolved_report = report_path.resolve(strict=True)
    try:
        review_root.relative_to(repo_root)
        resolved_report.relative_to(review_root)
    except ValueError as error:
        raise LookupError("invalid review report path") from error
    if resolved_report != review_root / f"{slug}.md":
        raise LookupError("invalid review report path")

    return {"path": expected_path, "content": resolved_report.read_text(encoding="utf-8")}


def read_research_report(slug):
    """Read only the fixed research report belonging to a research card."""
    if not SLUG_RE.fullmatch(slug) or slug not in list_research_cards():
        raise LookupError("unknown research card")

    expected_path = f"docs/research/{slug}.md"
    with STATE_LOCK:
        stored = load_state()["cards"].get(slug, {})
        if stored.get("research_report") != expected_path:
            raise LookupError("research report not found")

    repo_root = Path(os.environ["BOARD_REPO_DIR"]).resolve(strict=True)
    research_root = (repo_root / "docs" / "research").resolve(strict=True)
    resolved_report = (repo_root / expected_path).resolve(strict=True)
    try:
        research_root.relative_to(repo_root)
        resolved_report.relative_to(research_root)
    except ValueError as error:
        raise LookupError("invalid research report path") from error
    if resolved_report != research_root / f"{slug}.md":
        raise LookupError("invalid research report path")

    return {"path": expected_path, "content": resolved_report.read_text(encoding="utf-8")}


def read_research_intent(slug):
    """Read the validated source document belonging to a research card."""
    if not SLUG_RE.fullmatch(slug) or slug not in list_research_cards():
        raise LookupError("unknown research card")
    expected_path = f"docs/research-intent/{slug}.md"
    repo_root = Path(os.environ["BOARD_REPO_DIR"]).resolve(strict=True)
    intent_root = (repo_root / "docs" / "research-intent").resolve(strict=True)
    resolved_intent = (repo_root / expected_path).resolve(strict=True)
    try:
        intent_root.relative_to(repo_root)
        resolved_intent.relative_to(intent_root)
    except ValueError as error:
        raise LookupError("invalid research intent path") from error
    if resolved_intent != intent_root / f"{slug}.md":
        raise LookupError("invalid research intent path")
    return {"path": expected_path, "content": resolved_intent.read_text(encoding="utf-8")}


def read_intent(slug):
    """Read the intent document belonging to an existing intent card."""
    if not SLUG_RE.fullmatch(slug) or slug not in list_intent_cards():
        raise LookupError("unknown intent card")

    expected_path = f"docs/intent/{slug}.md"
    repo_root = Path(os.environ["BOARD_REPO_DIR"]).resolve(strict=True)
    intent_root = (repo_root / "docs" / "intent").resolve(strict=True)
    intent_path = repo_root / expected_path
    resolved_intent = intent_path.resolve(strict=True)
    try:
        resolved_intent.relative_to(intent_root)
    except ValueError as error:
        raise LookupError("invalid intent path") from error
    if resolved_intent != intent_root / f"{slug}.md":
        raise LookupError("invalid intent path")

    return {"path": expected_path, "content": resolved_intent.read_text(encoding="utf-8")}


# Kept in sync by hand with the role each job-triggering phase actually
# runs (worker.py's *_PROMPT_TEMPLATEs and agent-codex/builder.py's own
# fixer/builder prompts all name their role file explicitly, e.g.
# "Du bist die durch reviewer.md definierte Reviewer-Rolle") -- this is
# purely a read-only "view the role prompt" link for the Agents column,
# not itself involved in dispatching a job.
AGENT_ROLE_FILES = {
    "operator": "operator.md",
    "builder": "builder.md",
    "reviewer": "reviewer.md",
    "researcher": "researcher.md",
    "planner": "planner.md",
    "inspector": "inspector.md",
    "fixer": "fixer.md",
}


def read_agent_prompt(role):
    """Read a fixed agent role-prompt file from BOARD_ROLE_DIR (agents/).

    Same validated-allowlist pattern as read_intent() above, just rooted at
    the role directory instead of docs/intent: the request contributes a
    role name checked against AGENT_ROLE_FILES, never a filesystem path.
    """
    filename = AGENT_ROLE_FILES.get(role)
    if not filename:
        raise LookupError("unknown agent role")

    role_root = Path(os.environ["BOARD_ROLE_DIR"]).resolve(strict=True)
    prompt_path = (role_root / filename).resolve(strict=True)
    try:
        prompt_path.relative_to(role_root)
    except ValueError as error:
        raise LookupError("invalid role prompt path") from error
    if prompt_path != role_root / filename:
        raise LookupError("invalid role prompt path")

    return {"path": f"agents/{filename}", "content": prompt_path.read_text(encoding="utf-8")}


def read_card_telemetry(slug):
    if not SLUG_RE.fullmatch(slug):
        raise LookupError("unknown card")
    stored = load_state().get("cards", {}).get(slug, {})
    if slug not in list_intent_cards() and slug not in list_research_cards() and not stored:
        raise LookupError("unknown card")
    pipeline_run_id = stored.get("pipeline_run_id")
    current = (
        telemetry.get_pipeline_run(Path(os.environ["BOARD_TELEMETRY_DB"]), pipeline_run_id)
        if pipeline_run_id
        else None
    )
    return {
        "slug": slug,
        "pipeline_run_id": pipeline_run_id,
        "current_pipeline": current,
        "runs": telemetry.get_slug_runs(
            Path(os.environ["BOARD_TELEMETRY_DB"]), slug, limit=20
        ),
    }


def repo_job_active():
    """True if a job is queued or currently being run by worker.py -- the
    job file sits in BOARD_PROCESSING_DIR for the whole `codex exec`
    duration, not just while queued, so this covers both "waiting" and
    "mid-run". Used as the trigger-side guard for research jobs, which
    (per docs/intent/board-builder-worktree-parallelism.md) still run
    against the one shared checkout and therefore still need the full,
    untargeted lock against *any* other job -- builder jobs use the
    slug-aware, concurrency-limited builder_trigger_blocked() below
    instead.
    """
    for directory_env in ("BOARD_INBOX_DIR", "BOARD_PROCESSING_DIR", "AGENT_BUILDER_QUEUE_DIR", "AGENT_BUILDER_PROCESSING_DIR", "AGENT_BUILDER_QUEUE_DIR_LEICHT", "AGENT_BUILDER_PROCESSING_DIR_LEICHT"):
        if not os.environ.get(directory_env):
            continue
        directory = Path(os.environ[directory_env])
        if directory.is_dir() and any(directory.glob("*.json")):
            return True
    return False


def queued_jobs():
    """Parse every job currently queued or in flight (BOARD_INBOX_DIR and
    BOARD_PROCESSING_DIR) into a list of dicts. Corrupt or unreadable
    entries are skipped rather than raising -- a bad file here must never
    block the trigger API."""
    jobs = []
    for directory_env in ("BOARD_INBOX_DIR", "BOARD_PROCESSING_DIR", "AGENT_BUILDER_QUEUE_DIR", "AGENT_BUILDER_PROCESSING_DIR", "AGENT_BUILDER_QUEUE_DIR_LEICHT", "AGENT_BUILDER_PROCESSING_DIR_LEICHT"):
        if not os.environ.get(directory_env):
            continue
        directory = Path(os.environ[directory_env])
        if not directory.is_dir():
            continue
        for job_path in sorted(directory.glob("*.json")):
            try:
                job = json.loads(job_path.read_text(encoding="utf-8"))
                job["slug"] = job.get("slug") or job.get("board_slug")
                if directory_env.startswith("AGENT_BUILDER"):
                    job.setdefault("job_type", "builder")
                    job.setdefault("lane", "leicht" if directory_env.endswith("LEICHT") else "schwer")
                jobs.append(job)
            except (OSError, ValueError):
                continue
    return jobs


def builder_trigger_blocked(card_id, lane="schwer"):
    """Returns an error string if a new builder job for card_id must not be
    queued yet, else None.

    Builder jobs alone may be queued concurrently for *different* cards --
    see docs/intent/board-builder-worktree-parallelism.md for the isolated
    per-job git worktrees (agent-codex/builder.py's create_worktree()/
    remove_worktree(), only for the flake repo) that make that safe.
    Since docs/intent/board-builder-weighted-lanes.md gave each of the two
    Builder columns its own dedicated agent-codex-builder-1/-2 instance and
    inbox, the cap is enforced per lane (exactly one in-flight job per
    lane, matching one dedicated instance each) rather than as one shared
    number -- a "schwer" and a "leicht" job may always run at the same
    time, but two "schwer" (or two "leicht") jobs may not. Research/review
    jobs still use the one shared checkout without any lock of their own
    on the trigger side, so any of those queued or running still blocks a
    new builder job entirely, same as before."""
    jobs = queued_jobs()
    if any(job.get("job_type", "builder") != "builder" for job in jobs):
        return "a board-agent job is already queued or running against this repository"
    builder_jobs = [job for job in jobs if job.get("job_type", "builder") == "builder"]
    if any(job.get("slug") == card_id for job in builder_jobs):
        return "a board-agent job is already queued or running for this card"
    if any(job.get("lane", "schwer") == lane for job in builder_jobs):
        return f"a board-agent builder job is already queued or running for lane '{lane}'"
    return None


def write_job(slug, title, intent_path, job_type="builder", runtime=None, **extra):
    job_id = f"{slug}-{uuid.uuid4().hex[:16]}"
    inbox = Path(os.environ["BOARD_INBOX_DIR"])
    inbox.mkdir(parents=True, exist_ok=True, mode=0o700)
    pipeline_run_id = extra.pop("pipeline_run_id", None)
    if not pipeline_run_id and os.environ.get("BOARD_STATE_FILE"):
        stored = load_state().get("cards", {}).get(slug, {})
        pipeline_run_id = stored.get("pipeline_run_id")
        parent_slug = stored.get("operator_parent_id") or stored.get("synthesis_for")
        if not pipeline_run_id and parent_slug:
            pipeline_run_id = (
                load_state().get("cards", {}).get(parent_slug, {}).get("pipeline_run_id")
            )
    pipeline_run_id = pipeline_run_id or uuid.uuid4().hex
    job = {
        "job_id": job_id,
        "job_type": job_type,
        "slug": slug,
        "title": title,
        "intent_path": intent_path,
        "requested_at": now_iso(),
        "runtime": validate_runtime(runtime or board_runtime()),
        "pipeline_run_id": pipeline_run_id,
        **extra,
    }
    state_store.write_payload(inbox / f"{job_id}.json", job)
    return job_id


def builder_column_payload(column="in_umsetzung"):
    """Snapshot the Builder column settings for a queued job.

    The values are read once, at trigger time, so changing the column later
    cannot alter a job that is already waiting in the inbox.  Empty optional
    values are omitted because the worker uses its own defaults in that case.

    `column` selects which of the two Builder columns
    (docs/intent/board-builder-weighted-lanes.md) to snapshot; defaults to
    "in_umsetzung" ("schwer") for the split-escalation call site, which is
    deliberately not lane-aware.
    """
    config = load_column_config()[column]
    extra = {
        key: config[key]
        for key in ("model", "reasoning_effort")
        if config.get(key)
    }
    extra["iteration_limit"] = config["iteration_limit"]
    return config["runtime"], extra


def known_card_ids():
    """Every slug the board currently recognizes as a real card -- intent,
    research, or manually created idea -- used to validate both ends of a
    link_parent() call before touching fanout.py's state, which otherwise
    trusts its caller to have already done this (see fanout.link_child()'s
    docstring)."""
    ids = set(list_intent_cards()) | set(list_research_cards())
    for card_id, stored in load_state().get("cards", {}).items():
        if stored.get("source") == "idea":
            ids.add(card_id)
    return ids


def link_parent(child_id, parent_id):
    """Validate both slugs are real, known cards, then delegate the actual
    bookkeeping write to fanout.link_child() -- see
    docs/intent/board-subintent-fanout-synthesis.md."""
    with STATE_LOCK:
        ids = known_card_ids()
        if child_id not in ids:
            raise LookupError("unknown card")
        if parent_id not in ids:
            raise LookupError("unknown parent card")
        fanout.link_child(parent_id, child_id)


def retry_review(card_id, commit_hash=None):
    """Queue a fresh Review job against a card's own already-recorded
    commit_hash, without a new Builder run.

    Exists because a Review job can fail for reasons that have nothing to
    do with the commit itself (found live 2026-09-09,
    operator-smoke-split-docs and operator-smoke-split-agent-codex-2: two
    independent Reviewer phrasing variants run_review()'s decision match
    didn't tolerate, since fixed) -- rebuilding via the normal "geplant ->
    in_umsetzung" path is the wrong recovery in that case, since Builder
    correctly reports "no_changes" when its own acceptance criteria are
    already met by the existing commit, and no_changes never queues a new
    Review. This re-runs exactly the Review step completion_watcher.py
    would have queued after that original commit.

    An explicit commit_hash overrides the card's own stored one -- needed
    only for a card whose Review failed *before* completion_watcher.py's
    2026-09-09 fix that records commit_hash at the commit itself rather
    than only on a successful Review; every card going forward has its
    own stored value and never needs the override. Validated against the
    same hex-digest shape agent-codex/builder.py's COMMIT_RECORD_RE
    extracts, since it flows straight into run_review()'s `git show`/
    `git diff` calls.

    Reuses the runtime/model/reasoning_effort already stored on the card
    (i.e. whatever the original Builder/Operator run actually used) rather
    than the live "review" column config, so a retry reviews the same
    commit under the same profile that produced it, not whatever config
    happens to be set right now.
    """
    if not SLUG_RE.fullmatch(card_id):
        raise LookupError("unknown card")
    if commit_hash is not None and not re.fullmatch(r"[0-9a-f]{7,40}", commit_hash):
        raise ValueError("ungültiger commit_hash")
    with STATE_LOCK:
        intent_cards = list_intent_cards()
        if card_id not in intent_cards:
            raise LookupError("unknown intent card")
        state = load_state()
        stored = state["cards"].get(card_id, {})
        commit_hash = commit_hash or stored.get("commit_hash")
        if not commit_hash:
            raise LookupError("card has no recorded commit to review")
        if repo_job_active():
            raise PermissionError(
                "a board-agent job is already queued or running against this repository"
            )
        runtime = validate_runtime(stored.get("runtime") or board_runtime())
        extra = {
            key: stored[key]
            for key in ("model", "reasoning_effort")
            if stored.get(key)
        }
        job_id = write_job(
            card_id,
            intent_cards[card_id]["title"],
            intent_cards[card_id]["intent_path"],
            job_type="review",
            runtime=runtime,
            commit_hash=commit_hash,
            pipeline_run_id=stored.get("pipeline_run_id") or uuid.uuid4().hex,
            **extra,
        )
        stored.update(
            {
                "source": "intent",
                "phase": "review",
                "last_job_id": job_id,
                "last_run_status": "queued_review",
                "updated_at": now_iso(),
            }
        )
        state["cards"][card_id] = stored
        save_state(state)
    return job_id


def _migrate_clarification_state(stored):
    """Read old Fixer cards without rewriting state merely on GET."""
    migrated = dict(stored)
    if "clarification_attempts" not in migrated and "fixer_attempts" in migrated:
        migrated["clarification_attempts"] = migrated.get("fixer_attempts", 0)
    if "clarification_question" not in migrated and "fixer_question" in migrated:
        migrated["clarification_question"] = migrated.get("fixer_question")
    if "clarification_qa" not in migrated and "fixer_qa" in migrated:
        migrated["clarification_qa"] = migrated.get("fixer_qa", [])
    return migrated


def _queue_clarification(card_id, stored, card, task_type, answer=None):
    if task_type not in CLARIFICATION_TASK_TYPES:
        raise ValueError("Auftragstyp muss implementation_status, route_next oder repair sein")
    stored.update(_migrate_clarification_state(stored))
    attempts = int(stored.get("fixer_attempts", 0))
    if task_type == "repair" and answer is None:
        if not repair_eligible(stored):
            raise PermissionError("repair ist nur für echte Fehler oder negative Reviews erlaubt")
    if task_type == "repair" and attempts >= 2:
        raise PermissionError("maximal zwei Fixer-Versuche pro Karte")
    qa = stored.get("clarification_qa") or stored.get("fixer_qa") or []
    if answer is not None:
        question = str(stored.get("clarification_question") or stored.get("fixer_question") or "").strip()
        answer = str(answer or "").strip()
        if not question or not answer or len(answer) > 4000:
            raise ValueError("eine konkrete Antwort mit maximal 4000 Zeichen ist erforderlich")
        qa = [*qa, {"question": question, "answer": answer}]
    config = load_column_config()["klaerung_reparatur"]
    pipeline_run_id = stored.get("pipeline_run_id") or uuid.uuid4().hex
    job_id = write_job(
        card_id, card["title"], card["intent_path"],
        job_type="repair" if task_type == "repair" else "inspector", runtime=config["runtime"],
        clarification_task_type=task_type,
        pipeline_run_id=pipeline_run_id,
        model=config.get("model") or None,
        reasoning_effort=config.get("reasoning_effort") or None,
        iteration_limit=config["iteration_limit"],
        builder_report=stored.get("builder_report", ""),
        failure_stage=stored.get("failure_stage", ""),
        failure_detail=stored.get("failure_detail", ""),
        review_report=stored.get("review_report", ""),
        commit_hash=stored.get("commit_hash", ""),
        original_phase=stored.get("clarification_origin_phase") or stored.get("phase", ""),
        fixer_qa=qa,
        fixer_attempts=attempts + (1 if task_type == "repair" else 0),
        deployment_evidence=deployment_evidence(stored),
        checked_revision=stored.get("commit_hash") or None,
    )
    stored.update(
        phase="klaerung_reparatur", last_job_id=job_id,
        last_run_status="queued_repair" if task_type == "repair" else "queued_inspector",
        clarification_task_type=task_type,
        clarification_origin_phase=stored.get("clarification_origin_phase") or stored.get("phase", ""),
        fixer_attempts=attempts + (1 if task_type == "repair" else 0),
        clarification_attempts=attempts + (1 if task_type == "repair" else 0),
        fixer_qa=qa, clarification_qa=qa, fixer_question=None,
        clarification_question=None, pipeline_run_id=pipeline_run_id,
        runtime=config["runtime"], model=config.get("model", ""),
        reasoning_effort=config.get("reasoning_effort", ""),
        iteration_limit=config["iteration_limit"], updated_at=now_iso(),
    )
    return job_id


def deployment_evidence(stored):
    status = stored.get("deployment_status")
    if workflow.verified_deployment(stored):
        normalized = "deployed"
    elif status in {"deploy_failed", "not_deployed"}:
        normalized = "not_deployed"
    else:
        normalized = "unknown"
    return {
        "status": normalized,
        "commit_hash": stored.get("commit_hash", ""),
        "deploy_job_id": stored.get("deploy_job_id", ""),
        "source": "trusted board wrapper state",
    }


def answer_fixer(card_id, answer):
    with STATE_LOCK:
        cards = {**list_intent_cards(), **list_research_cards()}
        state = load_state()
        stored = state["cards"].get(card_id, {})
        if card_id not in cards and stored.get("source") != "idea":
            raise LookupError("unknown intent card")
        if stored.get("phase") not in {"fixer_rueckfrage", "inspector_rueckfrage"}:
            raise PermissionError("Antwort ist nur bei einer persistenten Rückfrage erlaubt")
        if repo_job_active():
            raise PermissionError("a board-agent job is already queued or running")
        card = cards.get(card_id) or {
            "id": card_id, "source": "idea", "title": stored.get("title", card_id),
            "intent_path": "",
        }
        task_type = stored.get("clarification_task_type") or "repair"
        job_id = _queue_clarification(card_id, stored, card, task_type, answer=answer)
        state["cards"][card_id] = stored
        save_state(state)
        return job_id


def apply_routing_recommendation(card_id):
    """Apply only an Inspector recommendation explicitly confirmed by the user."""
    with STATE_LOCK:
        state = load_state()
        stored = state.get("cards", {}).get(card_id, {})
        action = stored.get("routing_recommendation")
        if not stored.get("routing_pending") or action not in ROUTING_ACTION_PHASES:
            raise PermissionError("keine bestätigbare Routing-Empfehlung vorhanden")
        if action == "review":
            retry_review(card_id)
        elif action in {"in_umsetzung", "in_umsetzung_leicht"} and stored.get("phase") != "geplant":
            update_phase(card_id, "geplant")
            update_phase(card_id, action)
        else:
            update_phase(card_id, action)
        state = load_state()
        stored = state["cards"][card_id]
        stored.update(routing_pending=False, routing_applied_at=now_iso(), updated_at=now_iso())
        state["cards"][card_id] = stored
        save_state(state)
        return action


def update_phase(card_id, new_phase, task_type=None):
    if new_phase not in PHASES and new_phase not in RESEARCH_PHASES:
        raise ValueError("invalid phase")
    intent_cards = list_intent_cards()
    research_cards = list_research_cards()
    with STATE_LOCK:
        state = load_state()
        is_intent = card_id in intent_cards
        is_research = card_id in research_cards
        stored = state["cards"].get(card_id, {})
        if not is_intent and not is_research and stored.get("source") != "idea":
            raise LookupError("unknown card")
        current_phase = stored.get("phase", "recherche_angefordert" if is_research else ("geplant" if is_intent else "idea"))

        job_id = None
        if new_phase in {"deployed", "bereit_fuer_deployment", "recherche_fertig"}:
            raise PermissionError("This state requires a verified worker result, not a manual move.")
        if card_id in orchestration.active_slugs():
            raise PermissionError("This card has queued or active work; wait for its result.")
        if new_phase == "review":
            return retry_review(card_id)
        if new_phase == "archiviert":
            descendants = set(stored.get("child_slugs") or stored.get("depends_on") or [])
            if descendants & orchestration.active_slugs():
                raise PermissionError("Active children must finish before this family can be archived.")
            stored.update(phase="archiviert", updated_at=now_iso())
            state["cards"][card_id] = stored
            save_state(state)
            return None
        reset_phase = "recherche_angefordert" if is_research else "geplant" if is_intent else "idea"
        if new_phase == reset_phase:
            if is_research and current_phase not in RESEARCH_PHASES and current_phase not in {
                "klaerung_reparatur", "inspector_laeuft", "inspector_rueckfrage", "inspector_fehlgeschlagen",
            }:
                raise PermissionError("Research card has an invalid stored phase.")
            history = stored.setdefault("attempt_history", [])
            if stored.get("last_job_id"):
                history.append({key: stored.get(key) for key in ("last_job_id", "last_run_status", "last_run_at", "commit_hash", "review_decision", "builder_report", "failure_detail", "pipeline_run_id")})
                del history[:-100]
            for key in ("last_run_status", "failure_stage", "failure_detail", "clarification_origin_phase",
                        "routing_pending", "operator_dispatch_state", "review_decision", "review_report",
                        "auto_retry_attempts", "deployment_status", "deployed_commit_hash", "activated_system",
                        "deploy_job_id", "expected_builder_job_id"):
                stored.pop(key, None)
            stored.update(phase=reset_phase, updated_at=now_iso())
            state["cards"][card_id] = stored
            save_state(state)
            return None

        fixer_phases = {
            "fixer_angefordert", "fixer_laeuft", "fixer_rueckfrage", "fixer_fehlgeschlagen",
            "klaerung_reparatur", "inspector_laeuft", "inspector_rueckfrage",
            "inspector_fehlgeschlagen",
        }
        if new_phase in {"fixer_angefordert", "klaerung_reparatur"}:
            if repo_job_active():
                raise PermissionError("a board-agent job is already queued or running")
            card = intent_cards.get(card_id) or research_cards.get(card_id) or {
                "id": card_id, "source": "idea", "title": stored.get("title", card_id),
                "intent_path": "",
            }
            job_id = _queue_clarification(
                card_id, stored, card, task_type or ("repair" if new_phase == "fixer_angefordert" else "")
            )
            stored["source"] = card["source"]
            state["cards"][card_id] = stored
            save_state(state)
            return job_id
        if new_phase in fixer_phases:
            raise PermissionError("diese Fixer-Phase wird nur automatisch gesetzt")
        if current_phase in fixer_phases:
            if current_phase == "fixer_rueckfrage":
                if new_phase == "review":
                    # A human may send a question-blocked card to Review for
                    # inspection without answering or discarding the
                    # persistent Fixer question.
                    stored["phase"] = "review"
                    stored["updated_at"] = now_iso()
                    state["cards"][card_id] = stored
                    save_state(state)
                    return None
                raise PermissionError("Rückfrage über das Antwortfeld beantworten")
            if new_phase != "geplant":
                raise PermissionError("Fixer-Ergebnis muss zuerst nach 'geplant' zurückgesetzt werden")
            stored["phase"] = "geplant"
            stored["updated_at"] = now_iso()
            state["cards"][card_id] = stored
            save_state(state)
            return None

        if is_research:
            if current_phase not in RESEARCH_PHASES:
                raise PermissionError("research card has an invalid stored phase")
            if new_phase == "archiviert" or current_phase == "archiviert":
                if new_phase == "recherche_laeuft":
                    raise PermissionError("archived cards must move to 'recherche_angefordert' before 'recherche_laeuft'")
                stored["phase"] = new_phase
                stored["updated_at"] = now_iso()
                stored["source"] = "research"
                state["cards"][card_id] = stored
                save_state(state)
                fanout.maybe_synthesize(card_id)
                return None
            if new_phase not in RESEARCH_PHASES:
                raise PermissionError("research cards require research phases")
            if RESEARCH_PHASE_RANK[new_phase] > RESEARCH_PHASE_RANK[current_phase] and RESEARCH_PHASE_RANK[new_phase] - RESEARCH_PHASE_RANK[current_phase] > 1:
                raise PermissionError("forward transition skips a research stage")
            if new_phase == "recherche_laeuft":
                if current_phase != "recherche_angefordert":
                    raise PermissionError("trigger only allowed from 'recherche_angefordert'")
                if repo_job_active():
                    raise PermissionError("a board-agent job is already queued or running against this repository")
                config = load_column_config()["recherche_laeuft"]
                extra = {
                    key: config[key]
                    for key in ("model", "reasoning_effort")
                    if config.get(key)
                }
                runtime = config["runtime"]
                pipeline_run_id = uuid.uuid4().hex
                job_id = write_job(card_id, research_cards[card_id]["title"], research_cards[card_id]["intent_path"], job_type="research", runtime=runtime, research_intent_path=research_cards[card_id]["intent_path"], pipeline_run_id=pipeline_run_id, **extra)
                stored.update({"source": "research", "last_job_id": job_id, "last_run_status": "queued", "runtime": runtime, "pipeline_run_id": pipeline_run_id})
            stored["phase"] = new_phase
            stored["updated_at"] = now_iso()
            stored["source"] = "research"
            state["cards"][card_id] = stored
            save_state(state)
            # Covers a card manually dragged straight to recherche_fertig/
            # -fehlgeschlagen, not just the automatic worker.py completion
            # path -- see docs/intent/board-subintent-fanout-synthesis.md.
            fanout.maybe_synthesize(card_id)
            return job_id

        # Archiving is an explicit bookkeeping escape hatch. It is allowed
        # from every phase and never queues a pipeline job. Reactivation must
        # go through geplant before the normal builder trigger can be used.
        if current_phase == "archiviert" and new_phase in ("in_umsetzung", "in_umsetzung_leicht"):
            raise PermissionError(
                "archived cards must move to 'geplant' before 'in_umsetzung'"
            )
        if new_phase == "archiviert" or current_phase == "archiviert":
            stored["phase"] = new_phase
            stored["updated_at"] = now_iso()
            if is_intent:
                stored["source"] = "intent"
            state["cards"][card_id] = stored
            save_state(state)
            fanout.maybe_synthesize(card_id)
            return None

        # The Operator request is the only manually reachable Operator
        # transition. The remaining phases are worker/scheduler outcomes.
        operator_phases = {
            "operator_angefordert", "operator_laeuft", "wartet_auf_kinder",
            "durch_kinder_blockiert", "operator_klaerung_noetig",
        }
        if new_phase == "operator_angefordert":
            if not is_intent:
                raise PermissionError("nur Intent-Karten können vom Operator analysiert werden")
            if stored.get("operator_generated"):
                raise PermissionError("Operator-Kinder dürfen nicht rekursiv analysiert werden")
            if current_phase != "geplant":
                raise PermissionError("Operator-Trigger ist nur aus 'geplant' erlaubt")
            if not SLUG_RE.fullmatch(f"{card_id}-synthese"):
                raise PermissionError("Karten-Slug ist für einen Synthese-Intent zu lang")
            if repo_job_active():
                raise PermissionError(
                    "a board-agent job is already queued or running against this repository"
                )
            config = load_column_config()["operator_angefordert"]
            extra = {
                key: config[key]
                for key in ("model", "reasoning_effort")
                if config.get(key)
            }
            pipeline_run_id = uuid.uuid4().hex
            job_id = write_job(
                card_id,
                intent_cards[card_id]["title"],
                intent_cards[card_id]["intent_path"],
                job_type="operator",
                runtime=config["runtime"],
                builder_profiles=orchestration.configured_profiles(load_column_config()),
                pipeline_run_id=pipeline_run_id,
                **extra,
            )
            # A card may be sent to the Operator after an earlier Builder or
            # Review attempt. Those fields describe the old pipeline and must
            # not be rendered as the outcome of this newly queued Operator
            # run (notably an old iteration-limit report).
            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",
            ):
                stored.pop(stale_field, None)
            stored.update(
                {
                    "source": "intent",
                    "title": intent_cards[card_id]["title"],
                    "phase": new_phase,
                    "last_job_id": job_id,
                    "last_run_status": "queued_operator",
                    "runtime": config["runtime"],
                    "model": config["model"],
                    "reasoning_effort": config["reasoning_effort"],
                    "pipeline_run_id": pipeline_run_id,
                    "updated_at": now_iso(),
                }
            )
            state["cards"][card_id] = stored
            save_state(state)
            return job_id
        if new_phase in operator_phases:
            raise PermissionError("diese Operator-Phase wird nur automatisch gesetzt")
        if current_phase in operator_phases:
            if new_phase != "geplant":
                raise PermissionError("Operator-Ergebnis muss zuerst nach 'geplant' zurückgesetzt werden")
            stored["phase"] = "geplant"
            stored["updated_at"] = now_iso()
            state["cards"][card_id] = stored
            save_state(state)
            return None

        # Split escalation column -- see
        # docs/intent/board-builder-limit-reached-auto-split.md. A sibling
        # escalation path off "in_umsetzung", not a further pipeline stage:
        # bypasses the generic rank/skip-stage guard below entirely, the
        # same way archiviert does above. Leaving "aufteilen_angefordert" or
        # the worker-only "aufgeteilt" outcome again is unguarded
        # bookkeeping, same treatment as leaving archiviert.
        if current_phase in ("aufteilen_angefordert", "aufgeteilt"):
            stored["phase"] = new_phase
            stored["updated_at"] = now_iso()
            if is_intent:
                stored["source"] = "intent"
            state["cards"][card_id] = stored
            save_state(state)
            return None
        if new_phase == "aufteilen_angefordert":
            if not is_intent:
                raise PermissionError("nur Intent-Karten können aufgeteilt werden")
            if stored.get("operator_generated"):
                raise PermissionError("Operator-Kinder dürfen nicht rekursiv aufgeteilt werden")
            if not escalation_exhausted(stored):
                raise PermissionError(
                    "Aufteilen ist erst möglich, wenn die Karte am "
                    "Iterationslimit hängen geblieben ist und ihre "
                    "automatischen Eskalationsversuche bereits verbraucht sind"
                )
            if repo_job_active():
                raise PermissionError(
                    "a board-agent job is already queued or running against this repository"
                )
            runtime, extra = builder_column_payload()
            pipeline_run_id = stored.get("pipeline_run_id") or uuid.uuid4().hex
            job_id = write_job(
                card_id,
                intent_cards[card_id]["title"],
                intent_cards[card_id]["intent_path"],
                job_type="split",
                runtime=runtime,
                builder_report=stored.get("builder_report", ""),
                pipeline_run_id=pipeline_run_id,
                **extra,
            )
            stored["pipeline_run_id"] = pipeline_run_id
            stored["last_job_id"] = job_id
            stored["last_run_status"] = "queued_split"
            stored["runtime"] = runtime
            stored["phase"] = new_phase
            stored["updated_at"] = now_iso()
            stored["source"] = "intent"
            state["cards"][card_id] = stored
            save_state(state)
            return job_id

        current_rank = PHASE_RANK[current_phase]
        new_rank = PHASE_RANK[new_phase]
        if new_rank > current_rank and new_rank - current_rank > 1:
            raise PermissionError(
                f"forward transition from '{current_phase}' to '{new_phase}' skips a pipeline stage"
            )

        job_id = None
        if is_intent and new_phase in ("in_umsetzung", "in_umsetzung_leicht"):
            if current_phase != "geplant":
                raise PermissionError(
                    f"card is '{current_phase}', trigger only allowed from 'geplant'"
                )
            # "in_umsetzung"/"in_umsetzung_leicht" are the board's two
            # dedicated Builder columns ("schwer"/"leicht") -- see
            # docs/intent/board-builder-weighted-lanes.md.
            lane = "leicht" if new_phase == "in_umsetzung_leicht" else "schwer"
            # Guards the case the phase check alone misses: this same card
            # (dragged back to geplant and forward again), a research/review
            # job on the shared checkout, or an already-queued builder job
            # in the same lane. agent-codex/builder.py's flock is the hard
            # backstop if this check ever loses a race; this is the fast,
            # clear rejection for the normal case.
            blocked = builder_trigger_blocked(card_id, lane)
            if blocked:
                raise PermissionError(blocked)
            runtime, extra = builder_column_payload(new_phase)
            extra["lane"] = lane
            pipeline_run_id = uuid.uuid4().hex
            stored.update(
                {
                    "source": "intent",
                    "slug": card_id,
                    "title": intent_cards[card_id]["title"],
                    "intent_path": f"docs/intent/{card_id}.md",
                    "lane": lane,
                }
            )
            job_id = write_job(
                card_id,
                intent_cards[card_id]["title"],
                intent_cards[card_id]["intent_path"],
                runtime=runtime,
                pipeline_run_id=pipeline_run_id,
                **extra,
            )
            stored["pipeline_run_id"] = pipeline_run_id
            stored["last_job_id"] = job_id
            stored["last_run_status"] = "queued"
            stored["runtime"] = runtime
            stored["model"] = extra.get("model", "")
            stored["reasoning_effort"] = extra.get("reasoning_effort", "")
            stored["iteration_limit"] = extra["iteration_limit"]

        stored["phase"] = new_phase
        stored["updated_at"] = now_iso()
        if is_intent:
            stored["source"] = "intent"
        state["cards"][card_id] = stored
        save_state(state)
        # The common case for a fan-out Builder child: a human drags it from
        # "review" to "bereit_fuer_deployment"/"planung_mit_feedback"/
        # "deployed" once the reviewer's verdict is in -- see
        # docs/intent/board-subintent-fanout-synthesis.md. No-op for a card
        # without a parent_id or one whose new phase isn't terminal yet.
        fanout.maybe_synthesize(card_id)
        orchestration.progress_all(write_job, load_column_config())
    return job_id


class Handler(BaseHTTPRequestHandler):
    server_version = "board-agent/1"

    def _send_json(self, status, payload):
        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _read_json_body(self):
        length = int(self.headers.get("Content-Length", "0") or "0")
        if length <= 0 or length > 8192:
            raise ValueError("invalid request body")
        raw = self.rfile.read(length)
        data = json.loads(raw.decode("utf-8"))
        if not isinstance(data, dict):
            raise ValueError("body must be a JSON object")
        return data

    def log_message(self, format_string, *args):
        sys.stderr.write("%s - %s\n" % (self.address_string(), format_string % args))

    def do_GET(self):
        path = urlsplit(self.path).path
        if path in ("/", "/index.html", "/board.html"):
            self._serve_board_html()
            return
        if path == "/api/board":
            try:
                result = {"cards": merged_board(), "runtime": board_runtime(), "column_config": load_column_config()}
            except (OSError, ValueError) as error:
                print(f"Board state unavailable: {error}", file=sys.stderr)
                self._send_json(503, {"error": "Board state unavailable; no data was reset."})
                return
            self._send_json(200, result)
            return
        if path == "/api/board/column-config":
            self._send_json(200, load_column_config())
            return
        if path == "/api/board/runtime":
            self._send_json(200, {"runtime": board_runtime()})
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/telemetry", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                payload = read_card_telemetry(match.group(1))
            except (LookupError, OSError, ValueError):
                self._send_json(404, {"error": "telemetry not found"})
                return
            self._send_json(200, payload)
            return
        match = re.fullmatch(r"/api/board/pipelines/([A-Za-z0-9-]{1,128})/telemetry", path)
        if match:
            try:
                payload = telemetry.get_pipeline_run(
                    Path(os.environ["BOARD_TELEMETRY_DB"]), match.group(1)
                )
            except (OSError, ValueError):
                self._send_json(404, {"error": "telemetry not found"})
                return
            self._send_json(200, payload)
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/review", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                report = read_review_report(match.group(1))
            except (LookupError, OSError, UnicodeError):
                self._send_json(404, {"error": "review report not found"})
                return
            self._send_json(200, report)
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/research", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                report = read_research_report(match.group(1))
            except (LookupError, OSError, UnicodeError):
                self._send_json(404, {"error": "research report not found"})
                return
            self._send_json(200, report)
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/research-intent", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                intent = read_research_intent(match.group(1))
            except (LookupError, OSError, UnicodeError):
                self._send_json(404, {"error": "research intent not found"})
                return
            self._send_json(200, intent)
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/intent", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                intent = read_intent(match.group(1))
            except (LookupError, OSError, UnicodeError):
                self._send_json(404, {"error": "intent not found"})
                return
            self._send_json(200, intent)
            return
        match = re.fullmatch(r"/api/board/agents/([a-z]+)", path)
        if match:
            try:
                prompt = read_agent_prompt(match.group(1))
            except (LookupError, OSError, UnicodeError):
                self._send_json(404, {"error": "agent prompt not found"})
                return
            self._send_json(200, prompt)
            return
        self._send_json(404, {"error": "not found"})

    def _serve_board_html(self):
        html_path = Path(os.environ["BOARD_HTML_PATH"])
        body = html_path.read_bytes()
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_POST(self):
        path = urlsplit(self.path).path
        match = re.fullmatch(r"/api/board/cards/([a-z0-9-]+)/retry-deployment", path)
        if match:
            try:
                job_id = retry_deployment(match.group(1))
            except (LookupError, PermissionError, ValueError) as error:
                self._send_json(409, {"error": str(error)})
                return
            self._send_json(202, {"job_id": job_id})
            return
        match = re.fullmatch(r"/api/board/cards/([a-z0-9-]+)/plan", path)
        if match:
            try:
                card_id = plan_card(match.group(1))
            except (LookupError, PermissionError, ValueError) as error:
                self._send_json(409, {"error": str(error)})
                return
            self._send_json(201, {"id": card_id})
            return
        match = re.fullmatch(r"/api/board/column-config/([^/]+)", path)
        if match:
            column = match.group(1)
            if column not in COLUMN_CONFIG_DEFAULTS:
                self._send_json(404, {"error": "not found"})
                return
            try:
                config = set_column_config(column, self._read_json_body())
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(200, config)
            return
        if path == "/api/board/runtime":
            try:
                runtime = set_board_runtime(self._read_json_body().get("runtime"))
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(200, {"runtime": runtime})
            return
        if path == "/api/board/ideas":
            try:
                payload = self._read_json_body()
                card_id = create_idea(str(payload.get("title", "")))
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(201, {"id": card_id})
            return
        if path == "/api/board/drafts":
            try:
                payload = self._read_json_body()
                card_id = create_draft(str(payload.get("topic", "")))
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(201, {"id": card_id})
            return
        match = re.fullmatch(
            r"/api/board/smoke-tests/(direct|split|remove-direct|remove-split)",
            path,
        )
        if match:
            try:
                slug = create_smoke_test(match.group(1))
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(201, {"id": slug})
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/fixer-answer", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                job_id = answer_fixer(match.group(1), self._read_json_body().get("answer"))
            except LookupError as error:
                self._send_json(404, {"error": str(error)})
                return
            except PermissionError as error:
                self._send_json(409, {"error": str(error)})
                return
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(200, {"status": "ok", "job_id": job_id})
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/draft-answer", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                job_id = answer_draft(match.group(1), self._read_json_body().get("answer"))
            except LookupError as error:
                self._send_json(404, {"error": str(error)})
                return
            except PermissionError as error:
                self._send_json(409, {"error": str(error)})
                return
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(200, {"status": "ok", "job_id": job_id})
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/draft-retry", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                job_id = retry_draft(match.group(1))
            except LookupError as error:
                self._send_json(404, {"error": str(error)})
                return
            except PermissionError as error:
                self._send_json(409, {"error": str(error)})
                return
            self._send_json(200, {"status": "ok", "job_id": job_id})
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/apply-routing", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            try:
                phase = apply_routing_recommendation(match.group(1))
            except PermissionError as error:
                self._send_json(409, {"error": str(error)})
                return
            self._send_json(200, {"status": "ok", "phase": phase})
            return
        match = re.fullmatch(r"/api/board/cards/([^/]+)/retry-review", path)
        if match and SLUG_RE.fullmatch(match.group(1)):
            # Body is optional: an explicit commit_hash is only needed for
            # a card whose Review failed before commit_hash started being
            # recorded at commit time (see retry_review()'s docstring) --
            # every other caller just POSTs with no body at all, which
            # _read_json_body() rejects (no/zero Content-Length) the same
            # way it would for a genuinely malformed body; either way this
            # route treats that as "no override", not an error.
            try:
                commit_hash = self._read_json_body().get("commit_hash")
            except ValueError:
                commit_hash = None
            try:
                job_id = retry_review(match.group(1), commit_hash)
            except LookupError as error:
                self._send_json(404, {"error": str(error)})
                return
            except PermissionError as error:
                self._send_json(409, {"error": str(error)})
                return
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(200, {"status": "ok", "job_id": job_id})
            return
        self._send_json(404, {"error": "not found"})

    def do_PATCH(self):
        path = urlsplit(self.path).path
        parent_match = re.fullmatch(r"/api/board/cards/(" + CARD_ID_RE.pattern + r")/parent", path)
        if parent_match:
            child_id = parent_match.group(1)
            try:
                payload = self._read_json_body()
                parent_id = str(payload.get("parent_id", ""))
                link_parent(child_id, parent_id)
            except LookupError as error:
                self._send_json(404, {"error": str(error)})
                return
            except ValueError as error:
                self._send_json(400, {"error": str(error)})
                return
            self._send_json(200, {"status": "ok"})
            return

        match = re.fullmatch(r"/api/board/cards/(" + CARD_ID_RE.pattern + r")", path)
        if not match:
            self._send_json(404, {"error": "not found"})
            return
        card_id = match.group(1)
        try:
            payload = self._read_json_body()
            new_phase = str(payload.get("phase", ""))
            job_id = update_phase(card_id, new_phase, payload.get("task_type"))
        except LookupError as error:
            self._send_json(404, {"error": str(error)})
            return
        except PermissionError as error:
            self._send_json(409, {"error": str(error)})
            return
        except ValueError as error:
            self._send_json(400, {"error": str(error)})
            return
        self._send_json(200, {"status": "ok", "job_id": job_id})

    def do_DELETE(self):
        path = urlsplit(self.path).path
        if path == "/api/board/smoke-tests":
            try:
                result = delete_smoke_test_cards()
            except PermissionError as error:
                self._send_json(409, {"error": str(error)})
                return
            self._send_json(200, {"status": "deleted", **result})
            return
        match = re.fullmatch(r"/api/board/cards/(" + CARD_ID_RE.pattern + r")", path)
        if not match:
            self._send_json(404, {"error": "not found"})
            return
        try:
            delete_manual_card(match.group(1))
        except LookupError as error:
            self._send_json(404, {"error": str(error)})
            return
        self._send_json(200, {"status": "deleted"})


def main():
    host = os.environ.get("BOARD_HTTP_HOST", "127.0.0.1")
    port = int(os.environ.get("BOARD_HTTP_PORT", "8060"))
    Path(os.environ["BOARD_STATE_DIR"]).mkdir(parents=True, exist_ok=True, mode=0o700)
    try:
        orchestration.progress_all(write_job, load_column_config())
    except (OSError, ValueError, RuntimeError) as error:
        print(f"Operator-Recovery konnte nicht fortgesetzt werden: {error}", file=sys.stderr)
    server = ThreadingHTTPServer((host, port), Handler)
    server.serve_forever()


if __name__ == "__main__":
    main()