nixos-config/hosts/blunix/localServices/board-agent/fanout.py

#!/usr/bin/env python3
"""Parent-/Child fan-out helpers shared by trigger.py, worker.py and
completion_watcher.py -- docs/intent/board-subintent-fanout-synthesis.md.

A child card's stored state carries parent_id (the parent's slug). The
parent's stored state carries depends_on, the canonical list of child
slugs linked to it via link_child(). Automatically split parents also
retain child_slugs for compatibility/display; synthesis accepts that field
as a fallback for state written before automatic splitting populated
depends_on. These are lightweight state.json overlay additions, same as
every other per-card field this board already tracks -- no board-UI change.

maybe_synthesize(slug) is meant to be called right after slug's stored
state has just been updated to what might be its terminal state (see
child_terminal() for the exact per-kind definition). It looks up slug's
parent_id and, once every sibling child of that parent has reached a
terminal state too -- successfully or not, a failed child still counts,
see the intent's "Annahmen" -- writes a new synthesis intent document
under docs/intent/ summarizing each child's outcome and result link. The
synthesis card starts in phase "geplant" purely by not having a
state.json entry yet (list_intent_cards()'s default), so the normal
manual "Geplant -> In Umsetzung" trigger is what starts its own Builder
run -- this module never queues one itself.
"""

import json
import state_store
import os
import re
from datetime import datetime, timezone
from pathlib import Path

SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,60}$")

# "geprüft" (bereit_fuer_deployment) and archiviert are the two success/
# bookkeeping terminal phases a Builder-kind (intent) child can reach;
# planung_mit_feedback is also terminal -- a completed review verdict,
# not a pending step -- even though it names a further round of manual
# planning. A Builder failure (no commit ever produced) never changes the
# card's phase at all, hence the separate status-based check below.
BUILDER_TERMINAL_PHASES = (
    "bereit_fuer_deployment",
    "planung_mit_feedback",
    "deployed",
    "archiviert",
)
# completion_watcher.py's process_record() writes exactly one of these
# literal strings as last_run_status when a Builder run ends without a
# commit and therefore without ever reaching a terminal phase.
BUILDER_TERMINAL_FAILURE_STATUSES = ("failed", "no_changes", "check_failed")
RESEARCH_TERMINAL_PHASES = ("recherche_fertig", "recherche_fehlgeschlagen", "archiviert")

SYNTHESIS_TEMPLATE = """\
# Synthese: {parent_title}

Automatisch erzeugt, nachdem alle Unter-Intents der Parent-Karte
`{parent_id}` einen Endzustand erreicht haben
(docs/intent/board-subintent-fanout-synthesis.md). Diese Karte startet
absichtlich in Phase "Geplant" -- kein automatischer Start der Umsetzung
und keine automatische Qualitätsbewertung: eine Person prüft und
überarbeitet diesen Entwurf, bevor sie ihn selbst manuell nach "In
Umsetzung" verschiebt.

Erzeugt: {generated_at}

## Ergebnisse der Unter-Intents

{child_lines}

## Ziel

<Von Hand ausfüllen: was aus den obigen Ergebnissen folgt. Bei
Teilfehlschlägen insbesondere prüfen, ob das Parent-Ziel trotzdem über
die erfolgreichen Kinder erreichbar ist oder ob zuerst ein
fehlgeschlagenes Kind wiederholt werden muss.>
"""

OPERATOR_SYNTHESIS_TEMPLATE = """\
# Intent: Synthese für {parent_title}

## Ziel

Führe die bereits erfolgreich reviewten Ergebnisse aller Unter-Intents des
Operator-Parents `{parent_id}` zu einem vollständigen Ergebnis zusammen.
Prüfe das kombinierte Ergebnis ausdrücklich gegen den unten eingebetteten,
unveränderten Original-Intent; eine reine Zusammenfassung der Kindberichte
reicht nicht aus.

Erzeugt: {generated_at}

## Erfolgreiche Unter-Intents

{child_lines}

## Unveränderter Original-Intent

{parent_intent}

## Umfang

- Fehlende Integrationsarbeit zwischen den Kind-Ergebnissen umsetzen.
- Ziel und Akzeptanzkriterien des Original-Intents vollständig validieren.
- Vorhandene Kind-Commits und Reviewberichte als Ausgangspunkt verwenden.

## Erfolgskriterien

- Das kombinierte Ergebnis erfüllt den ursprünglichen Parent-Intent.
- Alle relevanten Prüfungen laufen erfolgreich.
- Das Ergebnis durchläuft anschließend den normalen Reviewpfad.
"""


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


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


def _load_state():
    return state_store.load()


def _save_state(data):
    state_store.save(data)


def _intent_dir():
    return Path(os.environ["BOARD_INTENT_DIR"])


def _research_intent_dir():
    return Path(
        os.environ.get(
            "BOARD_RESEARCH_INTENT_DIR",
            str(_intent_dir().parent / "research-intent"),
        )
    )


def is_research_card(card_id, stored):
    if stored.get("source") == "research":
        return True
    if stored.get("source") == "intent":
        return False
    return (_research_intent_dir() / f"{card_id}.md").is_file()


def escalation_ladder_for(card):
    """Every Builder attempt profile this card is entitled to beyond the
    first, in order -- see docs/intent/board-operator-escalation-ladder.md.
    An Operator-authored ladder (direct_builder or a split child, stored as
    "escalation_attempts" by orchestration.py) is used as-is. A card that
    never went through the Operator (including a plain manually-fanned-out
    child) has no such field, so this falls back to a single synthetic
    step repeating the card's own current model/reasoning_effort/
    iteration_limit -- exactly the original flat single automatic retry
    (docs/intent/board-builder-limit-reached-fallback.md), unchanged.
    Lives here rather than in orchestration.py (which already imports this
    module) so both this module's own child_terminal() and
    orchestration.py's child_failed() share one implementation without a
    circular import."""
    ladder = card.get("escalation_attempts")
    if ladder:
        return ladder
    return [
        {
            "model": card.get("model", ""),
            "reasoning_effort": card.get("reasoning_effort", ""),
            "iteration_limit": card.get("iteration_limit"),
        }
    ]


def child_terminal(card_id, stored):
    """Whether stored (a child card's stored state) has reached an end
    state, per docs/intent/board-subintent-fanout-synthesis.md: any
    combination of success or failure counts, so a permanently failed
    child does not block synthesis forever."""
    phase = stored.get("phase", "")
    if is_research_card(card_id, stored):
        return phase in RESEARCH_TERMINAL_PHASES
    if phase in BUILDER_TERMINAL_PHASES:
        return True
    status = stored.get("last_run_status", "")
    if status in BUILDER_TERMINAL_FAILURE_STATUSES:
        return True
    # limit_reached is only terminal once every automatic escalation
    # attempt (see docs/intent/board-operator-escalation-ladder.md, or the
    # single-step fallback for a card with no Operator-authored ladder)
    # has already been used -- otherwise a retry is still queued or about
    # to run.
    if status == "limit_reached" and stored.get("auto_retry_attempts", 0) >= len(
        escalation_ladder_for(stored)
    ):
        return True
    return False


def child_result_ref(card_id, stored):
    if is_research_card(card_id, stored):
        return stored.get("research_report") or None
    return stored.get("commit_hash") or None


def child_success(stored):
    """The stricter success predicate used only by Operator families."""
    return (
        stored.get("phase") in ("bereit_fuer_deployment", "deployed")
        and stored.get("review_decision") == "bereit_fuer_deployment"
    )


def child_summary_line(card_id, stored):
    kind = "Research" if is_research_card(card_id, stored) else "Builder"
    phase = stored.get("phase", "?")
    status = stored.get("last_run_status", "?")
    ref = child_result_ref(card_id, stored)
    ref_text = f", Ergebnis: `{ref}`" if ref else ", kein Ergebnislink (Fehlschlag)"
    return f"- `{card_id}` ({kind}): Phase `{phase}`, letzter Lauf-Status `{status}`{ref_text}"


@state_store.atomic
def link_child(parent_id, child_id):
    """Link an existing child card to an existing parent card by writing
    parent_id onto the child and appending child_id to the parent's
    depends_on list. Callers are expected to have already validated that
    both slugs refer to real cards (trigger.py's known-card check) --
    this only writes the bookkeeping, it never invents a card."""
    if not SLUG_RE.fullmatch(parent_id) or not SLUG_RE.fullmatch(child_id):
        raise ValueError("invalid slug")
    if parent_id == child_id:
        raise ValueError("a card cannot be its own parent")
    state = _load_state()
    child = state["cards"].setdefault(child_id, {})
    child["parent_id"] = parent_id
    parent = state["cards"].setdefault(parent_id, {})
    depends_on = parent.setdefault("depends_on", [])
    if child_id not in depends_on:
        depends_on.append(child_id)
    _save_state(state)


@state_store.atomic
def maybe_generate_synthesis(parent_id):
    """If every child linked to parent_id is terminal and no synthesis
    has been generated for it yet, write the synthesis intent document
    and mark the parent so this only ever happens once. Returns the new
    synthesis slug, or None if nothing was generated (children still
    pending, no children linked, or already generated)."""
    if not parent_id or not SLUG_RE.fullmatch(parent_id):
        return None
    state = _load_state()
    parent = state["cards"].get(parent_id)
    if not parent or parent.get("synthesis_generated") or parent.get("phase") == "archiviert":
        return None
    children = list(parent.get("depends_on") or parent.get("child_slugs") or [])
    if not children:
        return None

    summaries = []
    for child_id in children:
        stored = state["cards"].get(child_id, {})
        if parent.get("orchestration_kind") == "operator":
            if not child_success(stored):
                return None
        elif not child_terminal(child_id, stored):
            return None
        summaries.append(child_summary_line(child_id, stored))

    synthesis_slug = f"{parent_id}-synthese"
    synthesis_path = _intent_dir() / f"{synthesis_slug}.md"
    if synthesis_path.exists():
        # A previous run got as far as writing the file but not marking
        # the parent (e.g. a crash between the two) -- don't overwrite an
        # existing draft a person may already be editing, just catch the
        # bookkeeping up.
        parent["synthesis_generated"] = True
        parent["synthesis_intent_path"] = f"docs/intent/{synthesis_slug}.md"
        state["cards"][parent_id] = parent
        _save_state(state)
        return None

    parent_title = parent.get("title") or parent_id
    template = (
        OPERATOR_SYNTHESIS_TEMPLATE
        if parent.get("orchestration_kind") == "operator"
        else SYNTHESIS_TEMPLATE
    )
    parent_intent = ""
    if parent.get("orchestration_kind") == "operator":
        try:
            parent_intent = (_intent_dir() / f"{parent_id}.md").read_text(
                encoding="utf-8"
            ).strip()
        except OSError:
            # An Operator synthesis cannot safely validate the parent goal
            # without the exact source intent. Leave generation pending so a
            # later scheduler pass can retry after the file is restored.
            return None
    content = template.format(
        parent_id=parent_id,
        parent_title=parent_title,
        child_lines="\n".join(summaries),
        parent_intent=parent_intent,
        generated_at=now_iso(),
    )
    _intent_dir().mkdir(parents=True, exist_ok=True)
    tmp = synthesis_path.with_suffix(".tmp")
    tmp.write_text(content, encoding="utf-8")
    tmp.replace(synthesis_path)

    parent["synthesis_generated"] = True
    parent["synthesis_intent_path"] = f"docs/intent/{synthesis_slug}.md"
    state["cards"][parent_id] = parent
    _save_state(state)
    return synthesis_slug


def maybe_synthesize(slug):
    """Convenience wrapper for the common call site: given the slug of a
    card whose stored state was just updated, look up its parent_id (if
    any) and attempt synthesis for that parent. No-op for a slug with no
    parent_id."""
    state = _load_state()
    stored = state["cards"].get(slug, {})
    parent_id = stored.get("parent_id")
    if not parent_id:
        return None
    return maybe_generate_synthesis(parent_id)