#!/usr/bin/env python3 """Validated Operator decisions and deterministic Builder scheduling. The Operator is deliberately read-only. This module is the trusted side of the boundary: it validates the complete model response, creates a split as a single state transition, and dispatches only dependency-ready Builder cards. """ import fcntl import json import state_store import os import re import uuid from contextlib import contextmanager from pathlib import Path import fanout DECISION_START = "===OPERATOR-DECISION===" DECISION_END = "===END-OPERATOR-DECISION===" MAX_CHILDREN = 10 LANES = {"schwer": "in_umsetzung", "leicht": "in_umsetzung_leicht"} SUCCESS_PHASES = {"bereit_fuer_deployment", "deployed"} FAILURE_PHASES = {"planung_mit_feedback"} FAILURE_STATUSES = {"failed", "no_changes", "check_failed", "publish_failed", "needs_input", "not_applicable", "deploy_failed"} SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,60}$") MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:+/-]{0,127}$") EFFORT_RE = re.compile(r"^[A-Za-z0-9_-]{1,32}$") MAX_ESCALATION_STEPS = 2 # The models an Operator-authored escalation step may pick, keyed by # runtime. Mirrors board.html's MODEL_OPTIONS (the human-facing config # dropdown) -- kept in sync by hand, same pragmatic duplication already # used elsewhere in this repo (e.g. the BOARD_*_MODEL defaults in # board-agent.nix). Escalation never changes runtime or lane, only model/ # reasoning_effort/iteration_limit within the attempt's own runtime -- see # docs/intent/board-operator-escalation-ladder.md. MODEL_POOL = { "codex": ["gpt-6-astra", "gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna"], "claude": ["sonnet", "opus", "haiku"], "openrouter": [ "moonshotai/kimi-k2-thinking", "deepseek/deepseek-v3.2", "cohere/north-mini-code:free", "nvidia/nemotron-3-super-120b-a12b:free", ], } def _exact_keys(value, expected, context): if not isinstance(value, dict) or set(value) != set(expected): raise ValueError(f"{context} muss exakt die Felder {sorted(expected)} enthalten.") def _text(value, field, *, nullable=False): if nullable and value is None: return None if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field} muss ein nichtleerer String sein.") return value.strip() def _nullable_profile(value, field, pattern): if value is None: return None if not isinstance(value, str) or (value and not pattern.fullmatch(value)): raise ValueError(f"{field} ist ungültig.") return value def configured_profiles(column_config): """Return the only Builder profiles an Operator is allowed to select.""" if set(column_config) == set(LANES): return { lane: { "runtime": str(config.get("runtime", "")), "model": str(config.get("model", "")), "reasoning_effort": str(config.get("reasoning_effort", "")), "iteration_limit": config.get("iteration_limit"), } for lane, config in column_config.items() } profiles = {} for lane, column in LANES.items(): config = column_config[column] profiles[lane] = { "runtime": str(config.get("runtime", "")), "model": str(config.get("model", "")), "reasoning_effort": str(config.get("reasoning_effort", "")), "iteration_limit": config.get("iteration_limit"), } return profiles def _validate_profile(item, profiles, context): lane = item["recommended_lane"] if lane not in LANES: raise ValueError(f"{context}.recommended_lane muss leicht oder schwer sein.") runtime = _text(item["runtime"], f"{context}.runtime") model = _nullable_profile(item["model"], f"{context}.model", MODEL_RE) effort = _nullable_profile( item["reasoning_effort"], f"{context}.reasoning_effort", EFFORT_RE ) expected = profiles[lane] actual = { "runtime": runtime, "model": model or "", "reasoning_effort": effort or "", } for field, value in actual.items(): if value != expected[field]: raise ValueError( f"{context}: Profil passt nicht zur serverseitigen Lane-Konfiguration." ) return lane, expected.copy() def _validate_escalation(value, runtime, context): """Validate a card's or child's optional escalation ladder: up to MAX_ESCALATION_STEPS additional Builder attempts beyond the first, each picking a model from MODEL_POOL for the attempt's own runtime (escalation never changes runtime or lane -- see docs/intent/board-operator-escalation-ladder.md). Missing/omitted entirely defaults to [] before this is called, so older-shaped Operator responses without any escalation field validate exactly as before and simply get no ladder (worker.py's escalation_ladder_for() then falls back to today's single flat retry).""" if not isinstance(value, list) or len(value) > MAX_ESCALATION_STEPS: raise ValueError( f"{context}.escalation muss eine Liste mit höchstens " f"{MAX_ESCALATION_STEPS} Einträgen sein." ) pool = MODEL_POOL.get(runtime, []) normalized = [] for index, step in enumerate(value): step_context = f"{context}.escalation[{index}]" _exact_keys(step, {"model", "reasoning_effort", "iteration_limit"}, step_context) model = _text(step["model"], f"{step_context}.model") if model not in pool: raise ValueError( f"{step_context}.model muss eines der konfigurierten Modelle für " f"{runtime} sein." ) effort = _nullable_profile( step["reasoning_effort"], f"{step_context}.reasoning_effort", EFFORT_RE ) limit = step["iteration_limit"] if isinstance(limit, bool) or not isinstance(limit, int) or not 5 <= limit <= 40: raise ValueError( f"{step_context}.iteration_limit muss eine Ganzzahl zwischen 5 und 40 sein." ) normalized.append( {"model": model, "reasoning_effort": effort or "", "iteration_limit": limit} ) return normalized def _string_list(value, field, *, allow_empty=False): if not isinstance(value, list) or (not allow_empty and not value): raise ValueError(f"{field} muss eine {'ggf. leere ' if allow_empty else ''}Liste sein.") result = [] for entry in value: if not isinstance(entry, str) or not entry.strip(): raise ValueError(f"{field} enthält einen ungültigen Eintrag.") entry = entry.strip() if entry in result: raise ValueError(f"{field} enthält Duplikate.") result.append(entry) return result def extract_operator_decision(response, parent_slug, column_config): """Parse and strictly validate one marker-delimited schema-v1 decision. Marker search matches worker.py's extract_review_report()/ extract_research_report(): find the block anywhere in the response and ignore text outside it, rather than requiring the whole response to be exactly the block. This isn't a security relaxation -- the trust boundary is the exact-schema validation below (slugs, profiles, the dependency graph, write_scope overlaps), never whether the model added a stray sentence before the block. Confirmed live 2026-09-09 (board-operator-smoke-test-direct/-split both failed with "Operator- Antwort darf außerhalb des Entscheidungsblocks keinen Text enthalten"): sonnet/high reliably prefaces the block with a one-line rationale despite operator.md's explicit "ausschließlich... ohne Text... davor" instruction -- the same "opens the marker but doesn't always frame it cleanly" gap extract_review_report() already tolerates for the Reviewer. """ if response.count(DECISION_START) != 1 or response.count(DECISION_END) != 1: raise ValueError("Operator lieferte nicht exakt einen Entscheidungsblock.") start = response.index(DECISION_START) + len(DECISION_START) end = response.index(DECISION_END) if end < start: raise ValueError("Operator-Entscheidungsmarker stehen in falscher Reihenfolge.") try: raw = json.loads(response[start:end].strip()) except (TypeError, json.JSONDecodeError) as error: raise ValueError("Operator-Entscheidung enthält kein gültiges JSON.") from error # Defaulted before the exact-keys check so a response that predates # docs/intent/board-operator-escalation-ladder.md (no "escalation" key # at all) still validates exactly as before -- it just gets no ladder. raw.setdefault("escalation", []) top_keys = { "schema_version", "decision", "reason", "recommended_lane", "runtime", "model", "reasoning_effort", "children", "human_action", "escalation", } _exact_keys(raw, top_keys, "Operator-Entscheidung") if raw["schema_version"] != 1: raise ValueError("Operator schema_version muss 1 sein.") decision = raw["decision"] if decision not in {"direct_builder", "split", "needs_human"}: raise ValueError("Unbekannte Operator-Entscheidung.") _text(raw["reason"], "reason") profiles = configured_profiles(column_config) child_keys = { "slug", "title", "wave", "blocked_by", "recommended_lane", "runtime", "model", "reasoning_effort", "write_scope", "acceptance_criteria", "intent_markdown", "escalation", } children = raw["children"] if not isinstance(children, list): raise ValueError("children muss eine Liste sein.") if decision == "direct_builder": if children or raw["human_action"] is not None: raise ValueError("direct_builder darf keine Kinder oder human_action enthalten.") lane, profile = _validate_profile(raw, profiles, "direct_builder") raw["recommended_lane"] = lane raw["profile"] = profile raw["escalation"] = _validate_escalation( raw["escalation"], profile["runtime"], "direct_builder" ) return raw for field in ("recommended_lane", "runtime", "model", "reasoning_effort"): if raw[field] is not None: raise ValueError(f"{decision}.{field} muss null sein.") if decision == "needs_human": if children: raise ValueError("needs_human darf keine Kinder enthalten.") if raw["escalation"]: raise ValueError("needs_human.escalation muss leer sein.") _text(raw["human_action"], "human_action") return raw if raw["human_action"] is not None: raise ValueError("split.human_action muss null sein.") if raw["escalation"]: raise ValueError( "split.escalation muss leer sein -- Eskalation wird pro Kind angegeben." ) if not 1 <= len(children) <= MAX_CHILDREN: raise ValueError(f"split muss 1 bis {MAX_CHILDREN} Kinder enthalten.") slugs = [] normalized = [] for index, child in enumerate(children): context = f"children[{index}]" child.setdefault("escalation", []) _exact_keys(child, child_keys, context) slug = _text(child["slug"], f"{context}.slug") if not SLUG_RE.fullmatch(slug) or slug == parent_slug or slug in slugs: raise ValueError(f"{context}.slug ist ungültig oder nicht eindeutig.") slugs.append(slug) title = _text(child["title"], f"{context}.title") wave = child["wave"] if isinstance(wave, bool) or not isinstance(wave, int) or not 1 <= wave <= 99: raise ValueError(f"{context}.wave muss zwischen 1 und 99 liegen.") blocked_by = _string_list(child["blocked_by"], f"{context}.blocked_by", allow_empty=True) lane, profile = _validate_profile(child, profiles, context) escalation = _validate_escalation(child["escalation"], profile["runtime"], context) write_scope = _string_list(child["write_scope"], f"{context}.write_scope") acceptance = _string_list( child["acceptance_criteria"], f"{context}.acceptance_criteria" ) intent = _text(child["intent_markdown"], f"{context}.intent_markdown") if not intent.startswith("# Intent:"): raise ValueError(f"{context}.intent_markdown muss mit '# Intent:' beginnen.") normalized.append( { **child, "slug": slug, "title": title, "blocked_by": blocked_by, "recommended_lane": lane, "profile": profile, "escalation": escalation, "write_scope": write_scope, "acceptance_criteria": acceptance, "intent_markdown": intent, } ) by_slug = {child["slug"]: child for child in normalized} for child in normalized: for dependency in child["blocked_by"]: if dependency not in by_slug or dependency == child["slug"]: raise ValueError(f"{child['slug']}.blocked_by verweist auf kein anderes Kind.") if by_slug[dependency]["wave"] >= child["wave"]: raise ValueError("Abhängigkeiten müssen in einer früheren Welle liegen.") # Path overlap is safe only when an explicit dependency serializes it. # directory declared by one child and a file below it declared by another. for index, left in enumerate(normalized): for right in normalized[index + 1 :]: for left_path in left["write_scope"]: for right_path in right["write_scope"]: a, b = left_path.rstrip("/"), right_path.rstrip("/") if a == b or a.startswith(b + "/") or b.startswith(a + "/"): serialized = ( left["slug"] in right["blocked_by"] or right["slug"] in left["blocked_by"] ) if not serialized: raise ValueError( "Kinder mit überlappendem write_scope müssen per blocked_by serialisiert sein." ) raw["children"] = normalized return raw def _state_path(): return Path(os.environ["BOARD_STATE_FILE"]) def load_state(): return state_store.load() def save_state(state): state_store.save(state) @contextmanager def scheduler_lock(): with state_store.transaction(): yield def _apply_split_unlocked(parent_slug, decision): """Publish every child and its state, rolling back on ordinary errors.""" intent_dir = Path(os.environ["BOARD_INTENT_DIR"]) intent_dir.mkdir(parents=True, exist_ok=True) children = decision["children"] targets = [intent_dir / f"{child['slug']}.md" for child in children] synthesis_path = intent_dir / f"{parent_slug}-synthese.md" if any(path.exists() for path in targets) or synthesis_path.exists(): raise RuntimeError("Mindestens ein vorgeschlagener Kind-Intent existiert bereits.") staged = [] published = [] try: for child, target in zip(children, targets): temp = intent_dir / f".{target.name}.{uuid.uuid4().hex}.tmp" temp.write_text(child["intent_markdown"].rstrip() + "\n", encoding="utf-8") staged.append(temp) state = load_state() if any(child["slug"] in state["cards"] for child in children): raise RuntimeError("Mindestens eine vorgeschlagene Kind-Karte existiert bereits.") parent = state["cards"].setdefault(parent_slug, {}) parent.update( { "orchestration_kind": "operator", "operator_decision": "split", "operator_reason": decision["reason"], "phase": "wartet_auf_kinder", "depends_on": [child["slug"] for child in children], "child_slugs": [child["slug"] for child in children], "synthesis_generated": False, } ) for child in children: state["cards"][child["slug"]] = { "source": "intent", "title": child["title"], "phase": "geplant", "parent_id": parent_slug, "operator_parent_id": parent_slug, "operator_generated": True, "operator_managed": True, "blocked_by": child["blocked_by"], "wave": child["wave"], "recommended_lane": child["recommended_lane"], "recommended_profile": child["profile"], "escalation_attempts": child["escalation"], "write_scope": child["write_scope"], "acceptance_criteria": child["acceptance_criteria"], } for temp, target in zip(staged, targets): temp.replace(target) published.append(target) save_state(state) except Exception: for path in staged: path.unlink(missing_ok=True) for path in published: path.unlink(missing_ok=True) raise return [child["slug"] for child in children] def apply_split(parent_slug, decision): """Publish a complete split while serializing every scheduler mutation.""" with scheduler_lock(): return _apply_split_unlocked(parent_slug, decision) @state_store.atomic def apply_direct(parent_slug, decision): """Persist a validated direct decision for capacity-aware scheduling.""" with scheduler_lock(): state = load_state() card = state["cards"].setdefault(parent_slug, {}) card.update( { "operator_decision": "direct_builder", "operator_reason": decision["reason"], "operator_managed": True, "recommended_lane": decision["recommended_lane"], "recommended_profile": decision["profile"], "escalation_attempts": decision["escalation"], "phase": "geplant", "operator_dispatch_state": "ready", "last_run_status": "waiting_for_builder", } ) save_state(state) @state_store.atomic def apply_needs_human(parent_slug, decision): # "operator_klaerung_noetig" is distinct from "durch_kinder_blockiert": # this card never got split, so there are no children to be blocked by # -- the Operator itself couldn't produce a safe automatic decision. with scheduler_lock(): state = load_state() state["cards"].setdefault(parent_slug, {}).update( { "operator_decision": "needs_human", "operator_reason": decision["reason"], "human_action": decision["human_action"], "phase": "operator_klaerung_noetig", } ) save_state(state) def child_success(card): return ( card.get("phase") in SUCCESS_PHASES and card.get("review_decision") == "bereit_fuer_deployment" ) # fanout.py already has no dependency on this module (this module imports # fanout, not the reverse), so the ladder helper itself lives there and is # re-exported here -- both child_failed() below and fanout.child_terminal() # need the identical "how many automatic attempts is this card entitled # to" answer, without a circular import. escalation_ladder_for = fanout.escalation_ladder_for def child_failed(card): if card.get("phase") in FAILURE_PHASES: return True status = card.get("last_run_status") return status in FAILURE_STATUSES or ( status == "limit_reached" and card.get("auto_retry_attempts", 0) >= len(escalation_ladder_for(card)) ) def _job_directories(): names = ( "BOARD_INBOX_DIR", "BOARD_PROCESSING_DIR", "AGENT_BUILDER_QUEUE_DIR", "AGENT_BUILDER_PROCESSING_DIR", "AGENT_BUILDER_QUEUE_DIR_LEICHT", "AGENT_BUILDER_PROCESSING_DIR_LEICHT", "AGENT_BUILDER_DEPLOY_QUEUE_DIR", ) result = [] for name in names: value = os.environ.get(name) if value: result.append(Path(value)) return result def active_slugs(): slugs = set() for directory in _job_directories(): if not directory.is_dir(): continue for path in directory.glob("*.json"): try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, ValueError): continue slug = payload.get("slug") or payload.get("board_slug") if slug: slugs.add(slug) return slugs def _active_operator_builders(state, active): return sum( 1 for slug in active if state["cards"].get(slug, {}).get("operator_managed") or state["cards"].get(slug, {}).get("operator_generated") or state["cards"].get(slug, {}).get("synthesis_for") ) def _lane_busy(state, active, lane): if any(state["cards"].get(slug, {}).get("lane") == lane for slug in active): return True lane_dirs = { "schwer": ("AGENT_BUILDER_QUEUE_DIR", "AGENT_BUILDER_PROCESSING_DIR"), "leicht": ( "AGENT_BUILDER_QUEUE_DIR_LEICHT", "AGENT_BUILDER_PROCESSING_DIR_LEICHT", ), } return any( value and any(Path(value).glob("*.json")) for name in lane_dirs[lane] for value in (os.environ.get(name),) ) def _queue_card(state, slug, write_job): card = state["cards"][slug] lane = card.get("recommended_lane", "schwer") profile = card.get("recommended_profile", {}) extra = { "lane": lane, "iteration_limit": profile.get("iteration_limit", 20), "operator_parent_id": card.get("operator_parent_id") or card.get("synthesis_for"), } for key in ("model", "reasoning_effort"): if profile.get(key): extra[key] = profile[key] job_id = write_job( slug, card.get("title") or slug, f"docs/intent/{slug}.md", job_type="builder", runtime=profile.get("runtime", "codex"), **extra, ) card.update( { "phase": LANES[lane], "lane": lane, "operator_dispatch_state": "queued", "last_job_id": job_id, "last_run_status": "queued", "runtime": profile.get("runtime", "codex"), "model": profile.get("model", ""), "reasoning_effort": profile.get("reasoning_effort", ""), } ) return job_id def progress_all(write_job, column_config): """Recover/advance every Operator family and dispatch up to capacity.""" with scheduler_lock(): state = load_state() if not any( card.get("orchestration_kind") == "operator" or card.get("operator_managed") or card.get("synthesis_for") for card in state["cards"].values() ): return [] profiles = configured_profiles(column_config) # Generate a success-only synthesis and give it a normal Builder profile. for parent_slug, parent in list(state["cards"].items()): if parent.get("orchestration_kind") != "operator" or parent.get("phase") == "archiviert": continue if parent.get("synthesis_generated"): synthesis_slug = Path(parent.get("synthesis_intent_path", "")).stem if not synthesis_slug: parent["phase"] = "durch_kinder_blockiert" parent["synthesis_status"] = "blocked" parent["blocking_children"] = ["fehlender-synthese-verweis"] continue child_cards = [ state["cards"].get(slug, {}) for slug in list(parent.get("depends_on") or []) ] synthesis = state["cards"].setdefault(synthesis_slug, {}) synthesis.update( { "source": "intent", "title": synthesis.get("title") or f"Synthese: {parent.get('title') or parent_slug}", "phase": synthesis.get("phase", "geplant"), "synthesis_for": parent_slug, "operator_generated": True, "operator_managed": True, "recommended_lane": synthesis.get("recommended_lane", "schwer"), "recommended_profile": synthesis.get( "recommended_profile", profiles["schwer"] ), "write_scope": synthesis.get( "write_scope", sorted( { path for card in child_cards for path in card.get("write_scope", []) } ), ), "acceptance_criteria": synthesis.get( "acceptance_criteria", parent.get("acceptance_criteria", []) ), } ) if child_success(synthesis): parent["phase"] = "bereit_fuer_deployment" parent["synthesis_status"] = "reviewed_successfully" parent.pop("blocking_children", None) elif child_failed(synthesis): parent["phase"] = "durch_kinder_blockiert" parent["synthesis_status"] = "blocked" parent["blocking_children"] = [synthesis_slug] else: parent["phase"] = "wartet_auf_kinder" parent["synthesis_status"] = "running" continue children = list(parent.get("depends_on") or []) cards = [state["cards"].get(slug, {}) for slug in children] if children and any(child_failed(card) for card in cards): parent["phase"] = "durch_kinder_blockiert" parent["blocking_children"] = [ slug for slug, card in zip(children, cards) if child_failed(card) ] elif children and all(child_success(card) for card in cards): parent["phase"] = "wartet_auf_kinder" parent.pop("blocking_children", None) save_state(state) synthesis_slug = fanout.maybe_generate_synthesis(parent_slug) state = load_state() parent = state["cards"][parent_slug] synthesis_slug = synthesis_slug or Path( parent.get("synthesis_intent_path", "") ).stem if synthesis_slug: synthesis = state["cards"].setdefault(synthesis_slug, {}) synthesis.update( { "source": "intent", "title": f"Synthese: {parent.get('title') or parent_slug}", "phase": synthesis.get("phase", "geplant"), "synthesis_for": parent_slug, "operator_generated": True, "operator_managed": True, "recommended_lane": "schwer", "recommended_profile": profiles["schwer"], "write_scope": sorted( {path for card in cards for path in card.get("write_scope", [])} ), "acceptance_criteria": parent.get("acceptance_criteria", []), } ) else: parent["phase"] = "wartet_auf_kinder" parent.pop("blocking_children", None) save_state(state) active = active_slugs() capacity = max(0, int(os.environ.get("BOARD_OPERATOR_MAX_CONCURRENT_BUILDER_JOBS", "2"))) remaining = capacity - _active_operator_builders(state, active) queued = [] if remaining <= 0: return queued candidates = [] for slug, card in state["cards"].items(): parent_id = card.get("parent_id") or card.get("operator_parent_id") or card.get("synthesis_for") if state["cards"].get(parent_id, {}).get("phase") == "archiviert": continue if not ( card.get("operator_managed") or card.get("operator_generated") or card.get("synthesis_for") ): continue if card.get("operator_dispatch_state") == "queued" or slug in active: continue if card.get("phase", "geplant") != "geplant": continue dependencies = card.get("blocked_by", []) if not all(child_success(state["cards"].get(dep, {})) for dep in dependencies): continue candidates.append((card.get("wave", 100), slug)) for _wave, slug in sorted(candidates): if remaining <= 0: break lane = state["cards"][slug].get("recommended_lane", "schwer") if _lane_busy(state, active, lane): continue queued.append(_queue_card(state, slug, write_job)) active.add(slug) remaining -= 1 save_state(state) return queued