714 lines
29 KiB
Python
714 lines
29 KiB
Python
#!/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
|