601 lines
26 KiB
Python
601 lines
26 KiB
Python
#!/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()
|