#!/usr/bin/env python3
"""Path-triggered oneshot: watches BOARD_BUILDER_RESULTS_DIR for the
completion records ../agent-codex/builder.py writes when a builder-inbox
job carries on_complete_dir (see enqueue_builder_job() in worker.py,
which is the only thing that ever sets it).

For each record:
- status == "committed": enqueue a job_type="review" job in this board's
  own inbox, referencing the first commit hash in the record (Phase 1 of
  docs/intent/git-intent-board-feedback.md only ever expects one repo --
  nixos-config -- to actually change per board-triggered intent; a
  Builder run that touched more than one writable dir would only get the
  first reviewed here, a known, accepted Phase-1 simplification, not a
  silent bug). Card phase moves to "review".
- status == "limit_reached" and this card hasn't exhausted its escalation
  ladder yet (orchestration.escalation_ladder_for(), see
  docs/intent/board-operator-escalation-ladder.md): queue one more Builder
  job for the same card at the next ladder step's model/reasoning_effort/
  iteration_limit, and bump auto_retry_attempts by one. An Operator-
  authored ladder (direct_builder or a split child) supplies up to two
  such steps; a card that never went through the Operator falls back to a
  single synthetic step that repeats its current params -- today's exact
  original flat single retry (docs/intent/board-builder-limit-reached-fallback.md),
  unchanged. Relies on ../agent-codex/builder.py's existing sender-bound
  save_context()/load_context() (BOARD_BUILDER_OWNER_JID is the fixed
  sender for every board job) to hand the retry the prior attempt's
  intermediate report as context automatically -- no context is threaded
  through here. The card stays in "queued_builder", the same state a
  manual trigger produces; no new board state/badge.
- anything else (no_changes/check_failed/failed, or limit_reached with
  the ladder already exhausted): no automatic phase transition and no
  Review job -- there is no commit for a Reviewer to check. last_run_status
  plus the Builder's own final response text (record["report"], written by
  builder.py's write_completion_record()) are recorded as builder_report,
  so the reasoning behind an empty run is visible on the card instead of
  only in agent.db.

Claimed records move into a `consumed/` subdirectory rather than being
deleted, mirroring why ../agent-codex/*'s done/ directories exist.
"""
import json
import hashlib
import re
import uuid
import state_store
import os
import sys
import time
from pathlib import Path

import fanout
import metrics
import orchestration
import telemetry
import trigger
import worker


def progress_operator_families():
    try:
        orchestration.progress_all(worker.enqueue_board_job, trigger.load_column_config())
    except (OSError, ValueError, RuntimeError) as error:
        print(f"Operator-Scheduler konnte nicht fortgesetzt werden: {error}", file=sys.stderr)


def claim_records(inbox, consumed):
    consumed.mkdir(parents=True, exist_ok=True, mode=0o700)
    for source in sorted(inbox.glob("*.json")):
        target = consumed / source.name
        try:
            source.replace(target)
        except FileNotFoundError:
            continue
        yield target


def slug_from_job_id(job_id):
    # enqueue_builder_job() names jobs "board-<slug>-<timestamp>".
    if not job_id.startswith("board-"):
        return ""
    rest = job_id[len("board-") :]
    slug, _, _timestamp = rest.rpartition("-")
    return slug


def _builder_started_epoch(job_id):
    """The Unix timestamp enqueue_builder_job() embedded in job_id when it
    queued this attempt -- the closest thing to a start time this record
    carries, used only for metrics.py's opt-in duration_seconds field."""
    if not job_id.startswith("board-"):
        return None
    _slug, _, timestamp = job_id[len("board-") :].rpartition("-")
    if not re.fullmatch(r"\d{10}", timestamp):
        return None
    try:
        return float(timestamp)
    except ValueError:
        return None


def _builder_duration_seconds(job_id, record):
    started = _builder_started_epoch(job_id)
    completed = record.get("completed_at")
    if started is None or not isinstance(completed, (int, float)):
        return None
    return max(0.0, completed - started)


def _builder_timing(job_id, record, run):
    source = record.get("timing") or run.get("timing") or {}
    timing = {
        key: source.get(key)
        for key in ("ttft_seconds", "model_duration_seconds", "duration_seconds")
        if source.get(key) is not None
    }
    for milliseconds, seconds in (
        ("ttft_ms", "ttft_seconds"),
        ("model_ms", "model_duration_seconds"),
        ("wall_ms", "duration_seconds"),
    ):
        if seconds not in timing and source.get(milliseconds) is not None:
            try:
                timing[seconds] = max(0.0, float(source[milliseconds]) / 1000.0)
            except (TypeError, ValueError):
                pass
    if "duration_seconds" not in timing:
        timing["duration_seconds"] = _builder_duration_seconds(job_id, record)
    return timing


def enqueue_review_job(
    slug, commit_hash, runtime="codex", model=None, reasoning_effort=None,
    pipeline_run_id=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)
    payload = {
        "job_id": job_id,
        "job_type": "review",
        "slug": slug,
        "commit_hash": commit_hash,
        "requested_at": worker.now_iso(),
        "runtime": worker.validate_runtime(runtime),
        "pipeline_run_id": pipeline_run_id or worker.pipeline_id_for_card(slug),
        **extra,
    }
    if model is not None:
        payload["model"] = model
    if reasoning_effort is not None:
        payload["reasoning_effort"] = reasoning_effort
    worker.write_payload(
        inbox / f"{job_id}.json",
        payload,
    )
    return job_id


def enqueue_builder_retry_job(
    slug, runtime="codex", model=None, reasoning_effort=None, iteration_limit=None,
):
    """Queue one more Builder attempt for the same card after a
    limit_reached completion. intent_path is derived as docs/intent/<slug>.md
    -- the same convention trigger.py's list_intent_cards() already relies
    on -- rather than carried through the completion record, which
    builder.py's write_completion_record() doesn't include. Returns None
    (instead of raising) if the intent document is gone or unreadable, so
    a vanished/renamed intent degrades to "no retry", not a crash.

    model/reasoning_effort/iteration_limit default to the card's current
    stored values when omitted (today's flat identical-params retry).
    Passed explicitly by process_record() when this attempt is one step of
    an Operator-authored escalation ladder (see
    docs/intent/board-operator-escalation-ladder.md) -- the card itself
    isn't updated with the new step's values until after this call, so it
    can't be re-derived from worker.load_card() here.

    Stays on the card's own lane (worker.load_card()'s stored "lane",
    defaulting to "schwer" for cards queued before lanes existed) so the
    retry lands back on the same dedicated builder instance as the
    original attempt -- see docs/intent/board-builder-weighted-lanes.md."""
    intent_path = f"docs/intent/{slug}.md"
    on_complete_dir = os.environ["BOARD_BUILDER_RESULTS_DIR"]
    card = worker.load_card(slug)
    lane = card.get("lane", "schwer")
    try:
        return worker.enqueue_builder_job(
            slug,
            intent_path,
            on_complete_dir,
            runtime,
            model=model if model is not None else card.get("model"),
            reasoning_effort=(
                reasoning_effort if reasoning_effort is not None
                else card.get("reasoning_effort")
            ),
            iteration_limit=(
                iteration_limit if iteration_limit is not None
                else card.get("iteration_limit")
            ),
            queue_dir_env=worker.LANE_QUEUE_ENV.get(lane, "AGENT_BUILDER_QUEUE_DIR"),
            pipeline_run_id=card.get("pipeline_run_id"),
        )
    except (OSError, ValueError, RuntimeError) as error:
        print(f"could not queue automatic limit_reached retry for {slug}: {error}", file=sys.stderr)
        return None


@state_store.atomic
def process_record(path):
    digest = hashlib.sha256(path.read_bytes()).hexdigest()
    state = state_store.load()
    processed = state.setdefault("processed_completions", [])
    if digest in processed:
        return
    record = json.loads(path.read_text(encoding="utf-8"))
    slug = record.get("board_slug") or slug_from_job_id(record.get("job_id", path.stem))
    current = state["cards"].get(slug, {})
    stale = (record.get("pipeline_run_id") and current.get("pipeline_run_id")
             and record["pipeline_run_id"] != current["pipeline_run_id"])
    expected = current.get("deploy_job_id") if record.get("deployment_status") else current.get("expected_builder_job_id")
    if expected and expected != record.get("job_id"):
        stale = True
    if not stale:
        _process_record(path)
    state = state_store.load()
    processed = state.setdefault("processed_completions", [])
    processed.append(digest)
    state_store.save(state)


def _process_record(path):
    try:
        record = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as error:
        print(f"invalid completion record {path.name}: {error}", file=sys.stderr)
        return

    job_id = record.get("job_id", path.stem)
    status = record.get("status", "")
    slug = record.get("board_slug") or slug_from_job_id(job_id)
    if not slug:
        print(f"completion record {job_id} has no recognizable board slug, skipping", file=sys.stderr)
        return

    # One metrics.py entry per completion record, regardless of which
    # branch below ends up handling it -- a limit_reached run that gets an
    # automatic retry queued still consumed a real Builder attempt's time
    # and (if applicable) cost, exactly like a committed or exhausted one.
    # No-op unless slug is in BOARD_METRICS_SLUGS, see metrics.record_run().
    card_before = worker.load_card(slug)
    committed_commits = record.get("commits") or []
    completion_role = (
        "deployment" if record.get("deployment_status")
        else "fixer" if record.get("pipeline_role") == "fixer"
        else "builder"
    )
    metrics.record_run(
        slug,
        "board",
        status=status,
        job_type=completion_role,
        runtime=record.get("runtime"),
        model=card_before.get("model"),
        reasoning_effort=card_before.get("reasoning_effort"),
        job_id=job_id,
        commit_hash=committed_commits[0]["hash"] if status == "committed" and committed_commits else None,
        duration_seconds=_builder_duration_seconds(job_id, record),
    )

    # Automatic, board-wide telemetry is separate from the opt-in
    # state.json baseline pilot above. Builder owns its raw runtime capture;
    # this watcher consumes only its bounded completion metadata.
    run = record.get("telemetry") if isinstance(record.get("telemetry"), dict) else {}
    pipeline_run_id = (
        record.get("pipeline_run_id")
        or card_before.get("pipeline_run_id")
        or job_id
    )
    configured = record.get("configured_profile") or {
        "runtime": record.get("runtime"),
        "model": card_before.get("model"),
        "reasoning_effort": card_before.get("reasoning_effort"),
    }
    telemetry_db = worker.telemetry_db_path()
    try:
        if telemetry_db is None:
            raise LookupError("BOARD_TELEMETRY_DB ist nicht gesetzt")
        telemetry.record_job(
            telemetry_db,
            pipeline_run_id=pipeline_run_id,
            slug=slug,
            job_id=job_id,
            role=completion_role,
            configured_profile=configured,
            actual_profile=record.get("actual_profile") or run.get("actual_profile"),
            usage=record.get("usage") or run.get("usage"),
            timing=_builder_timing(job_id, record, run),
            iterations=record.get("iterations") or run.get("iterations"),
            outcome={
                "status": status,
                "success": (record.get("deployment_status") == "deployed"
                            if record.get("deployment_status") else status == "committed"),
                "commit_hash": (
                    committed_commits[0].get("hash") if committed_commits else None
                ),
            },
            session_id=record.get("session_id") or run.get("session_id"),
        )
    except LookupError:
        pass
    except (OSError, ValueError) as error:
        print(f"Builder-Telemetrie konnte nicht gespeichert werden: {error}", file=sys.stderr)

    deployment_status = record.get("deployment_status")
    deployment_commit_hash = (
        committed_commits[0].get("hash") if committed_commits else None
    )
    deployment_is_review_released = bool(
        deployment_status
        and card_before.get("review_decision") == "bereit_fuer_deployment"
        and card_before.get("commit_hash")
        and card_before.get("commit_hash") == deployment_commit_hash
    )

    # deploy_runner completions deliberately contain both status=committed
    # and deployment_status.  Accept that combined shape as a deployment only
    # when durable card state proves that the Reviewer released this exact
    # commit.  Otherwise the committed branch below must still queue Review;
    # retain the premature deployment report as evidence without making the
    # card terminal or allowing parent synthesis.
    if deployment_status and deployment_is_review_released:
        if deployment_status == "deployed" and not (
            str(record.get("deployed_commit_hash", "")).startswith(str(deployment_commit_hash))
            and re.fullmatch(r"[0-9a-f]{40}", str(record.get("deployed_commit_hash", "")))
            and str(record.get("activated_system", "")).startswith("/nix/store/")
        ):
            worker.update_state(slug, "deploy_failed", job_id, phase="fixer_fehlgeschlagen",
                                failure_stage="deployment", failure_detail="Deployment receipt lacks verified revision/closure.")
            return
        worker.update_state(
            slug,
            deployment_status,
            job_id,
            phase="deployed" if deployment_status == "deployed" else "fixer_fehlgeschlagen",
            deployment_status=deployment_status,
            deploy_job_id=job_id,
            commit_hash=deployment_commit_hash,
            deployed_commit_hash=record.get("deployed_commit_hash"),
            activated_system=record.get("activated_system"),
            failure_stage="deployment" if deployment_status != "deployed" else None,
            failure_detail=record.get("failure_detail") if deployment_status != "deployed" else None,
        )
        if deployment_status == "deployed":
            # Deployment is also a successful terminal child state.  Advance
            # the generic fan-out family first, then let the Operator
            # scheduler reconcile its stricter success-only family state.
            # Neither helper calls process_record(), so this cannot recurse.
            fanout.maybe_synthesize(slug)
            progress_operator_families()
        print(f"reviewed deployment {job_id}: {deployment_status}", file=sys.stderr)
        return

    premature_deployment_evidence = {}
    if deployment_status:
        premature_deployment_evidence = {
            "deployment_status": deployment_status,
            "deploy_job_id": job_id,
            "deployment_commit_hash": deployment_commit_hash,
            "deployment_before_review": True,
        }
        print(
            f"deployment completion {job_id} for {deployment_commit_hash or 'unknown commit'} "
            "has no matching Reviewer release; enforcing committed -> review",
            file=sys.stderr,
        )

    fixer_outcome = record.get("fixer_outcome")
    if record.get("pipeline_role") == "fixer":
        if not isinstance(fixer_outcome, dict):
            if status in ("failed", "check_failed", "limit_reached"):
                worker.update_state(
                    slug, status, job_id, phase="fixer_fehlgeschlagen",
                    failure_stage=(
                        record.get("failure_stage")
                        or card_before.get("failure_stage")
                        or "fixer_runtime"
                    ),
                    failure_detail=(
                        record.get("failure_detail")
                        or record.get("report")
                        or card_before.get("failure_detail")
                        or f"Fixer-Lauf endete mit Status {status}."
                    ),
                    builder_report=record.get("report", ""),
                )
                return
            worker.update_state(
                slug, "failed", job_id, phase="fixer_fehlgeschlagen",
                failure_stage="fixer_outcome", failure_detail="Fixer-Outcome fehlt oder ist ungültig.",
                builder_report=record.get("report", ""),
            )
            return
        outcome = fixer_outcome.get("outcome")
        if outcome == "needs_input":
            worker.update_state(
                slug, "needs_input", job_id, phase="inspector_rueckfrage",
                fixer_question=fixer_outcome.get("question", ""),
                clarification_question=fixer_outcome.get("question", ""),
                builder_report=record.get("report", ""),
            )
            return
        if outcome == "not_applicable":
            worker.update_state(
                slug, "not_applicable", job_id, phase="klaerung_reparatur",
                fixer_outcome=outcome, builder_report=record.get("report", ""),
                clarification_question=None,
            )
            return
        if outcome == "failed" or status in ("failed", "check_failed", "limit_reached"):
            worker.update_state(
                slug, "failed", job_id, phase="fixer_fehlgeschlagen",
                failure_stage="fixer", failure_detail=fixer_outcome.get("summary", "Fixer fehlgeschlagen"),
            )
            return
        if outcome != "repaired":
            worker.update_state(slug, "failed", job_id, phase="fixer_fehlgeschlagen")
            return
        commits = record.get("commits") or []
        commit_hash = commits[0].get("hash") if commits else card_before.get("commit_hash")
        if not commit_hash:
            worker.update_state(
                slug, "failed", job_id, phase="fixer_fehlgeschlagen",
                failure_stage="fixer_review", failure_detail="Kein Commit für den zwingenden Review vorhanden.",
            )
            return
        config = trigger.load_column_config()["review"]
        review_job_id = enqueue_review_job(
            slug, commit_hash, config.get("runtime", "codex"),
            model=config.get("model") or None,
            reasoning_effort=config.get("reasoning_effort") or None,
            pipeline_run_id=pipeline_run_id,
            fixer_result=True,
            fixer_job_id=job_id,
        )
        worker.update_state(
            slug, "queued_review", job_id, phase="review",
            review_job_id=review_job_id, commit_hash=commit_hash,
            fixer_outcome=outcome, fixer_question=None,
            clarification_question=None,
            fixer_reported_commit_hash=fixer_outcome.get("commit_hash"),
            fixer_reported_pushed_branch=fixer_outcome.get("pushed_branch"),
            fixer_commit_discrepancy=fixer_outcome.get("commit_hash") != commit_hash,
        )
        return

    if status == "committed":
        commits = record.get("commits") or []
        commit_hash = commits[0]["hash"] if commits else ""
        if not commit_hash:
            print(f"completion record {job_id} is 'committed' but has no commit hash", file=sys.stderr)
            worker.update_state(slug, "failed", job_id)
            return
        try:
            config = trigger.load_column_config()["review"]
            if not isinstance(config, dict):
                raise ValueError("Review-Spaltenkonfiguration ist kein Objekt")
            runtime = config.get("runtime", "codex")
            extra = {
                key: config[key]
                for key in ("model", "reasoning_effort")
                if config.get(key)
            }
            review_job_id = enqueue_review_job(
                slug, commit_hash, runtime,
                pipeline_run_id=pipeline_run_id,
                **extra,
            )
        except (OSError, ValueError, KeyError, TypeError) as error:
            worker.update_failure_state(slug, job_id, "review_config", error)
            print(
                f"builder job {job_id} committed {commit_hash}, review config invalid: {error}",
                file=sys.stderr,
            )
            return
        # commit_hash is recorded here, at the commit itself, rather than
        # only on a successful Review (worker.py's existing behaviour) --
        # found live 2026-09-09: a Review failure for a reason unrelated
        # to the commit (e.g. the case/language-tolerance gaps just fixed
        # in run_review()) left the card with no commit_hash at all, so
        # trigger.retry_review() had nothing to retry against even though
        # a perfectly good commit existed.
        worker.update_state(
            slug, "queued_review", job_id, review_job_id=review_job_id,
            commit_hash=commit_hash, pipeline_run_id=pipeline_run_id,
            **premature_deployment_evidence,
        )
        worker.set_phase(slug, "review")
        progress_operator_families()
        print(f"builder job {job_id} committed {commit_hash}, queued review {review_job_id}", file=sys.stderr)
        return

    if status == "limit_reached":
        card_for_retry = worker.load_card(slug)
        attempts_used = card_for_retry.get("auto_retry_attempts", 0)
        ladder = orchestration.escalation_ladder_for(card_for_retry)
        if attempts_used < len(ladder):
            step = ladder[attempts_used]
            retry_job_id = enqueue_builder_retry_job(
                slug,
                record.get("runtime", "codex"),
                model=step.get("model"),
                reasoning_effort=step.get("reasoning_effort"),
                iteration_limit=step.get("iteration_limit"),
            )
            if retry_job_id:
                worker.update_state(
                    slug,
                    "queued_builder",
                    job_id,
                    expected_builder_job_id=retry_job_id,
                    builder_report=record.get("report", ""),
                    auto_retry_attempts=attempts_used + 1,
                    model=step.get("model", card_for_retry.get("model", "")),
                    reasoning_effort=step.get(
                        "reasoning_effort", card_for_retry.get("reasoning_effort", "")
                    ),
                    iteration_limit=step.get(
                        "iteration_limit", card_for_retry.get("iteration_limit")
                    ),
                )
                print(
                    f"builder job {job_id} hit limit_reached, queued escalation attempt "
                    f"{attempts_used + 1}/{len(ladder)} as {retry_job_id}",
                    file=sys.stderr,
                )
                return
            print(f"builder job {job_id} hit limit_reached, automatic retry could not be queued", file=sys.stderr)

    if status == "no_changes" and card_before.get("synthesis_for"):
        # Operator children were already committed and reviewed before this
        # synthesis was dispatched. If their combined result needs no further
        # integration diff, there is nothing new for another Reviewer to
        # inspect: mark the synthesis successful and release its parent.
        worker.update_state(
            slug,
            "done",
            job_id,
            phase="bereit_fuer_deployment",
            review_decision="bereit_fuer_deployment",
            builder_report=record.get("report", ""),
            synthesis_no_changes=True,
        )
        progress_operator_families()
        print(
            f"builder synthesis {job_id} required no integration changes, marked successful",
            file=sys.stderr,
        )
        return

    extra = {}
    if status == "publish_failed":
        extra = {"failure_stage": "publication", "failure_detail": record.get("failure_detail") or record.get("report"),
                 "preserved_worktree": record.get("preserved_worktree")}
    worker.update_state(slug, status, job_id, builder_report=record.get("report", ""), **extra)
    # A Builder child that stalled without ever producing a commit
    # (no_changes/failed/check_failed, or limit_reached with its one
    # automatic retry already spent) has reached its terminal state right
    # here, unlike a committed run which still needs a human review verdict
    # -- see docs/intent/board-subintent-fanout-synthesis.md.
    fanout.maybe_synthesize(slug)
    progress_operator_families()
    print(f"builder job {job_id} finished with status {status}, no automatic transition", file=sys.stderr)


def main():
    inbox = Path(os.environ["BOARD_BUILDER_RESULTS_DIR"])
    consumed = Path(os.environ["BOARD_BUILDER_RESULTS_CONSUMED_DIR"])
    inbox.mkdir(parents=True, exist_ok=True, mode=0o700)

    processing = inbox.parent / "results-processing"
    processing.mkdir(parents=True, exist_ok=True, mode=0o700)
    consumed.mkdir(parents=True, exist_ok=True, mode=0o700)
    paths = list(processing.glob("*.json")) + list(claim_records(inbox, processing))
    for record_path in paths:
        try:
            value = json.loads(record_path.read_text(encoding="utf-8"))
            if not isinstance(value, dict) or not isinstance(value.get("job_id", record_path.stem), str):
                raise ValueError("Completion must be a JSON object with a job ID")
        except (ValueError, UnicodeError) as error:
            rejected = inbox.parent / "results-rejected"
            rejected.mkdir(parents=True, exist_ok=True, mode=0o700)
            record_path.replace(rejected / record_path.name)
            print(f"Quarantined invalid completion {record_path.name}: {error}", file=sys.stderr)
            continue
        process_record(record_path)
        record_path.replace(consumed / record_path.name)


if __name__ == "__main__":
    main()