753 lines
27 KiB
Python
753 lines
27 KiB
Python
"""Shared job-queue plumbing for worker.py and builder.py.
|
|
|
|
Deliberately excludes most role-specific model launchers: Worker always
|
|
invokes Codex read-only, while Builder dispatches to Codex's workspace-write
|
|
sandbox, the same harness with an OpenRouter provider, or Claude's
|
|
settings-based permission scope. The shared module provides validated
|
|
options, parsers, secret-file loading and streaming loops.
|
|
"""
|
|
import json
|
|
import fcntl
|
|
import os
|
|
import queue
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import db
|
|
|
|
SUPPORTED_RUNTIMES = ("codex", "claude", "openrouter")
|
|
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"
|
|
|
|
|
|
def validate_runtime(runtime):
|
|
"""Return the only runtime values accepted by board jobs."""
|
|
value = str(runtime or "").strip().lower()
|
|
if value not in SUPPORTED_RUNTIMES:
|
|
raise ValueError("Runtime muss exakt codex, claude oder openrouter sein.")
|
|
return value
|
|
|
|
|
|
def load_secret_file(variable):
|
|
"""Read a secret from the file named by ``variable`` without logging it."""
|
|
path = os.environ.get(variable, "").strip()
|
|
if not path:
|
|
raise RuntimeError(f"{variable} ist nicht gesetzt.")
|
|
try:
|
|
secret = Path(path).read_text(encoding="utf-8").strip()
|
|
except OSError as error:
|
|
raise RuntimeError(f"Secret-Datei aus {variable} ist nicht lesbar.") from error
|
|
if not secret:
|
|
raise RuntimeError(f"Secret-Datei aus {variable} ist leer.")
|
|
return secret
|
|
|
|
|
|
def append_openrouter_options(
|
|
command,
|
|
model_env,
|
|
model=None,
|
|
reasoning_env=None,
|
|
reasoning=None,
|
|
):
|
|
"""Configure Codex's Responses harness for the OpenRouter provider."""
|
|
command.extend(
|
|
[
|
|
"-c",
|
|
'model_provider="openrouter"',
|
|
"-c",
|
|
'model_providers.openrouter.name="OpenRouter"',
|
|
"-c",
|
|
f"model_providers.openrouter.base_url={json.dumps(OPENROUTER_BASE_URL)}",
|
|
"-c",
|
|
'model_providers.openrouter.env_key="OPENROUTER_API_KEY"',
|
|
"-c",
|
|
'model_providers.openrouter.wire_api="responses"',
|
|
"-c",
|
|
"model_providers.openrouter.requires_openai_auth=false",
|
|
]
|
|
)
|
|
selected_model = str(model or "").strip() or os.environ.get(model_env, "").strip()
|
|
if not selected_model:
|
|
raise ValueError(f"{model_env} ist nicht gesetzt.")
|
|
if len(selected_model) > 128 or not re.fullmatch(
|
|
r"[A-Za-z0-9][A-Za-z0-9._:+/-]*", selected_model
|
|
):
|
|
raise ValueError("OpenRouter-Modell enthält ungültige Zeichen.")
|
|
command.extend(["--model", selected_model])
|
|
|
|
selected_reasoning = (
|
|
os.environ.get(reasoning_env, "")
|
|
if reasoning is None and reasoning_env
|
|
else reasoning or ""
|
|
).strip()
|
|
if selected_reasoning:
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", selected_reasoning):
|
|
raise ValueError("OpenRouter-Reasoning-Effort enthält ungültige Zeichen.")
|
|
command.extend(
|
|
["-c", f"model_reasoning_effort={json.dumps(selected_reasoning)}"]
|
|
)
|
|
|
|
|
|
def write_payload(path, payload):
|
|
temporary = path.with_suffix(".tmp")
|
|
temporary.write_text(
|
|
json.dumps(payload, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
temporary.chmod(0o600)
|
|
temporary.replace(path)
|
|
|
|
|
|
def claim_jobs(inbox, processing):
|
|
processing.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
for source in sorted(inbox.glob("*.json")):
|
|
target = processing / source.name
|
|
with (inbox / '.delivery.lock').open('a+') as handle:
|
|
fcntl.flock(handle, fcntl.LOCK_EX)
|
|
if not source.exists():
|
|
continue
|
|
claims = inbox / '.claimed'
|
|
claims.mkdir(exist_ok=True, mode=0o700)
|
|
with (claims / source.name).open('w') as marker:
|
|
marker.write('{}\n')
|
|
marker.flush()
|
|
os.fsync(marker.fileno())
|
|
fd = os.open(claims, os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
source.replace(target)
|
|
yield target
|
|
|
|
|
|
def add_first_reply_notice(role, sender, response, notice_text):
|
|
if db.is_introduced(role, sender):
|
|
return response, None
|
|
return f"{notice_text.strip()}\n\n{response}", sender
|
|
|
|
|
|
def mark_introduced(role, sender):
|
|
db.mark_introduced(role, sender)
|
|
|
|
|
|
def load_session(role, sender, now=time.time):
|
|
state = db.load_session(role, sender)
|
|
if state is None:
|
|
return None
|
|
maximum_idle = int(os.environ.get("AGENT_SESSION_IDLE_SECONDS", "1800"))
|
|
session_id, updated_at = state
|
|
if (
|
|
not isinstance(session_id, str)
|
|
or not session_id
|
|
or not isinstance(updated_at, (int, float))
|
|
or now() - updated_at > maximum_idle
|
|
):
|
|
db.reset_session(role, sender)
|
|
return None
|
|
return session_id
|
|
|
|
|
|
def save_session(role, sender, session_id):
|
|
db.save_session(role, sender, session_id)
|
|
|
|
|
|
def reset_session(role, sender):
|
|
db.reset_session(role, sender)
|
|
|
|
|
|
def load_context(role, sender, now=time.time, scope=""):
|
|
"""Counterpart to load_session() for roles that carry a previous
|
|
reply forward as plain-text context instead of resuming a Codex
|
|
session (see docs/intent/builder-iterationslimit.md) -- same
|
|
idle-expiry rule, separate table (db.context, not db.sessions).
|
|
|
|
`scope` isolates this from an unrelated concurrent job for the same
|
|
(role, sender) -- see db.py's context table comment. Board jobs pass
|
|
their board_slug; plain XMPP chat leaves it at the default "" for
|
|
ordinary per-sender conversational continuity."""
|
|
state = db.load_context(role, sender, scope)
|
|
if state is None:
|
|
return None
|
|
maximum_idle = int(os.environ.get("AGENT_SESSION_IDLE_SECONDS", "1800"))
|
|
context_text, updated_at = state
|
|
if (
|
|
not isinstance(context_text, str)
|
|
or not context_text
|
|
or not isinstance(updated_at, (int, float))
|
|
or now() - updated_at > maximum_idle
|
|
):
|
|
db.reset_context(role, sender, scope)
|
|
return None
|
|
return context_text
|
|
|
|
|
|
def save_context(role, sender, context_text, scope=""):
|
|
db.save_context(role, sender, context_text, scope=scope)
|
|
|
|
|
|
def reset_context(role, sender, scope=""):
|
|
db.reset_context(role, sender, scope)
|
|
|
|
|
|
def parse_codex_events(output):
|
|
session_id = None
|
|
# Codex can emit more than one item.completed/agent_message per turn
|
|
# (e.g. an intermediate update followed by the final reply) -- keeping
|
|
# only the last one silently dropped everything before it, which
|
|
# looked from the outside like a long reply getting cut off, with no
|
|
# error and well under AGENT_MAX_RESPONSE_CHARS (found 2026-08-01 after
|
|
# a user report of truncated replies). Concatenate all of them instead.
|
|
response_parts = []
|
|
json_errors = []
|
|
for line in output.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
event = json.loads(line)
|
|
except ValueError as error:
|
|
json_errors.append(error)
|
|
continue
|
|
if event.get("type") == "thread.started":
|
|
session_id = event.get("thread_id")
|
|
if event.get("type") == "item.completed":
|
|
item = event.get("item", {})
|
|
if item.get("type") == "agent_message" and item.get("text"):
|
|
response_parts.append(item["text"])
|
|
|
|
if not session_id:
|
|
raise RuntimeError("Codex lieferte keine Session-ID.")
|
|
response = "\n\n".join(response_parts).strip()
|
|
if not response:
|
|
if json_errors:
|
|
raise RuntimeError(
|
|
f"Ungültige Codex-JSONL-Ausgabe: {json_errors[-1]}"
|
|
) from json_errors[-1]
|
|
raise RuntimeError("Codex lieferte keine Antwort.")
|
|
return response, session_id
|
|
|
|
|
|
def _integer(value):
|
|
"""Return a non-negative integer for provider usage fields."""
|
|
return value if isinstance(value, int) and not isinstance(value, bool) and value >= 0 else 0
|
|
|
|
|
|
def parse_codex_telemetry(output):
|
|
"""Extract prompt-free telemetry from Codex/OpenRouter ``--json`` output.
|
|
|
|
``turn.completed.usage`` is the authoritative aggregate emitted by the
|
|
harness. There is normally one completed turn per ``codex exec`` call;
|
|
summing also does the right thing if a future harness emits several.
|
|
"""
|
|
session_id = None
|
|
usage = {
|
|
"input_tokens": 0,
|
|
"cached_input_tokens": 0,
|
|
"output_tokens": 0,
|
|
"reasoning_tokens": 0,
|
|
"total_tokens": 0,
|
|
}
|
|
iterations = 0
|
|
for line in output.splitlines():
|
|
try:
|
|
event = json.loads(line)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if event.get("type") == "thread.started" and event.get("thread_id"):
|
|
session_id = event["thread_id"]
|
|
if event.get("type") == "item.completed":
|
|
iterations += 1
|
|
if event.get("type") != "turn.completed" or not isinstance(event.get("usage"), dict):
|
|
continue
|
|
source = event["usage"]
|
|
input_tokens = _integer(source.get("input_tokens"))
|
|
output_tokens = _integer(source.get("output_tokens"))
|
|
details = source.get("output_tokens_details") or {}
|
|
reasoning_tokens = _integer(
|
|
source.get("reasoning_tokens", details.get("reasoning_tokens"))
|
|
)
|
|
usage["input_tokens"] += input_tokens
|
|
usage["cached_input_tokens"] += _integer(source.get("cached_input_tokens"))
|
|
usage["output_tokens"] += output_tokens
|
|
usage["reasoning_tokens"] += reasoning_tokens
|
|
usage["total_tokens"] += _integer(source.get("total_tokens")) or input_tokens + output_tokens
|
|
return {
|
|
"session_id": session_id,
|
|
"actual_profile": {},
|
|
"usage": usage,
|
|
"timing": {},
|
|
"iterations": iterations,
|
|
}
|
|
|
|
|
|
def build_partial_response(output):
|
|
"""Best-effort counterpart to parse_codex_events() for an
|
|
iteration-limit cutoff (see run_streaming() below): the JSONL stream
|
|
may end mid-turn with no thread.started/complete turn, so this never
|
|
raises -- it just concatenates whatever agent_message text arrived
|
|
before the process was terminated, or "" if none did."""
|
|
parts = []
|
|
for line in output.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
event = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if event.get("type") == "item.completed":
|
|
item = event.get("item", {})
|
|
if item.get("type") == "agent_message" and item.get("text"):
|
|
parts.append(item["text"])
|
|
return "\n\n".join(parts).strip()
|
|
|
|
|
|
def _drain_lines(pipe, sink):
|
|
try:
|
|
for line in iter(pipe.readline, ""):
|
|
sink.put(line)
|
|
finally:
|
|
sink.put(None)
|
|
pipe.close()
|
|
|
|
|
|
def run_streaming(command, body, environment, timeout, iteration_limit):
|
|
"""Run `command` with `body` on stdin, counting Codex `item.completed`
|
|
JSONL events (the finest step granularity Codex's own --json output
|
|
exposes -- verified live: even a multi-tool-call turn is one single
|
|
turn.started/turn.completed pair) as iterations. Once the count
|
|
exceeds `iteration_limit`, the process is terminated (SIGTERM, then
|
|
SIGKILL if it doesn't exit promptly) instead of being allowed to run
|
|
unbounded -- see docs/intent/builder-iterationslimit.md. The overall
|
|
wall-clock `timeout` is enforced the same way subprocess.run's
|
|
`timeout` was (raises subprocess.TimeoutExpired) when the iteration
|
|
limit itself is never hit.
|
|
|
|
Returns (stdout_text, limit_reached, returncode).
|
|
"""
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
env=environment,
|
|
)
|
|
process.stdin.write(body)
|
|
process.stdin.close()
|
|
|
|
output_queue = queue.Queue()
|
|
reader = threading.Thread(
|
|
target=_drain_lines, args=(process.stdout, output_queue), daemon=True
|
|
)
|
|
reader.start()
|
|
|
|
lines = []
|
|
iterations = 0
|
|
limit_reached = False
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
process.kill()
|
|
process.wait()
|
|
raise subprocess.TimeoutExpired(command, timeout)
|
|
try:
|
|
line = output_queue.get(timeout=min(remaining, 1.0))
|
|
except queue.Empty:
|
|
continue
|
|
if line is None:
|
|
break
|
|
lines.append(line)
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
try:
|
|
event = json.loads(stripped)
|
|
except ValueError:
|
|
continue
|
|
if event.get("type") == "item.completed":
|
|
iterations += 1
|
|
if iterations > iteration_limit:
|
|
limit_reached = True
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
break
|
|
|
|
returncode = process.wait()
|
|
stderr = process.stderr.read()
|
|
if stderr:
|
|
print(stderr.rstrip(), file=sys.stderr)
|
|
return "".join(lines), limit_reached, returncode
|
|
|
|
|
|
def append_model_options(
|
|
command,
|
|
model_env="AGENT_CODEX_MODEL",
|
|
reasoning_env="AGENT_CODEX_REASONING_EFFORT",
|
|
model=None,
|
|
reasoning=None,
|
|
):
|
|
"""Add validated model controls before the ``exec`` command.
|
|
|
|
The conversational Agent uses the defaults; the Builder passes its
|
|
dedicated environment names so both roles can be tuned independently.
|
|
"""
|
|
model = (os.environ.get(model_env, "") if model is None else model).strip()
|
|
if model:
|
|
if len(model) > 128 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:+-]*", model):
|
|
raise ValueError("AGENT_CODEX_MODEL enthält ungültige Zeichen.")
|
|
command.extend(["--model", model])
|
|
|
|
reasoning = (os.environ.get(reasoning_env, "") if reasoning is None else reasoning).strip()
|
|
if reasoning:
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", reasoning):
|
|
raise ValueError("AGENT_CODEX_REASONING_EFFORT enthält ungültige Zeichen.")
|
|
# The CLI has no dedicated reasoning flag; -c is the supported
|
|
# config override and must be supplied before the `exec` subcommand.
|
|
command.extend(["-c", f"model_reasoning_effort={json.dumps(reasoning)}"])
|
|
|
|
|
|
def parse_claude_events(output):
|
|
"""Claude Code's --output-format stream-json counterpart to
|
|
parse_codex_events(): each line is one JSONL event; the run ends with
|
|
exactly one type=="result" event carrying the final plain-text answer
|
|
(result["result"]), the session id, and an is_error flag -- unlike
|
|
Codex's thread.started/item.completed shape, there's no separate
|
|
session-start event to watch for.
|
|
|
|
The is_error branch below is verified only against the *shape* of a
|
|
successful run (is_error/subtype/api_error_status all present and
|
|
populated in a success case) -- it has not yet been exercised against
|
|
a genuine Claude-side failure (e.g. a rate limit) in production.
|
|
Revisit this once one is observed live.
|
|
"""
|
|
result_event = None
|
|
for line in output.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
event = json.loads(line)
|
|
except ValueError as error:
|
|
raise RuntimeError(f"Ungültige Claude-JSONL-Ausgabe: {error}") from error
|
|
if event.get("type") == "result":
|
|
result_event = event
|
|
|
|
if result_event is None or not result_event.get("session_id"):
|
|
raise RuntimeError("Claude lieferte keine Session-ID.")
|
|
if result_event.get("is_error"):
|
|
detail = result_event.get("result") or result_event.get("subtype") or "unbekannter Fehler"
|
|
raise RuntimeError(f"Claude endete mit einem Fehler: {detail}")
|
|
response = (result_event.get("result") or "").strip()
|
|
if not response:
|
|
raise RuntimeError("Claude lieferte keine Antwort.")
|
|
return response, result_event["session_id"]
|
|
|
|
|
|
def parse_claude_telemetry(output):
|
|
"""Extract Claude's final aggregate usage/result metadata.
|
|
|
|
Assistant events repeat per-request usage while streaming. The final
|
|
``result`` event is therefore the sole source for totals, avoiding double
|
|
counting and keeping message/prompt contents out of the completion record.
|
|
"""
|
|
result = None
|
|
streamed_tool_calls = 0
|
|
for line in output.splitlines():
|
|
try:
|
|
event = json.loads(line)
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if event.get("type") == "assistant":
|
|
streamed_tool_calls += sum(
|
|
1
|
|
for block in event.get("message", {}).get("content", [])
|
|
if block.get("type") == "tool_use"
|
|
)
|
|
elif event.get("type") == "result":
|
|
result = event
|
|
|
|
result = result or {}
|
|
source = result.get("usage") if isinstance(result.get("usage"), dict) else {}
|
|
direct_input = _integer(source.get("input_tokens"))
|
|
cache_creation = _integer(source.get("cache_creation_input_tokens"))
|
|
cache_read = _integer(source.get("cache_read_input_tokens"))
|
|
input_tokens = direct_input + cache_creation + cache_read
|
|
output_tokens = _integer(source.get("output_tokens"))
|
|
details = source.get("output_tokens_details") or {}
|
|
reasoning_tokens = _integer(
|
|
source.get("reasoning_tokens", details.get("thinking_tokens"))
|
|
)
|
|
|
|
model_usage = result.get("modelUsage") or {}
|
|
actual_model = None
|
|
if isinstance(model_usage, dict) and model_usage:
|
|
def model_weight(item):
|
|
values = item[1] if isinstance(item[1], dict) else {}
|
|
return (
|
|
float(values.get("costUSD") or 0),
|
|
_integer(values.get("inputTokens"))
|
|
+ _integer(values.get("cacheReadInputTokens"))
|
|
+ _integer(values.get("cacheCreationInputTokens"))
|
|
+ _integer(values.get("outputTokens")),
|
|
)
|
|
|
|
actual_model = max(model_usage.items(), key=model_weight)[0]
|
|
|
|
timing = {}
|
|
for source_key, target_key in (
|
|
("duration_ms", "duration_seconds"),
|
|
("duration_api_ms", "model_duration_seconds"),
|
|
("ttft_ms", "ttft_seconds"),
|
|
):
|
|
value = result.get(source_key)
|
|
if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0:
|
|
timing[target_key] = value / 1000
|
|
|
|
iterations = _integer(result.get("num_turns")) or streamed_tool_calls
|
|
return {
|
|
"session_id": result.get("session_id"),
|
|
"actual_profile": {"model": actual_model} if actual_model else {},
|
|
"usage": {
|
|
"input_tokens": input_tokens,
|
|
"cached_input_tokens": cache_read,
|
|
"cache_creation_input_tokens": cache_creation,
|
|
"output_tokens": output_tokens,
|
|
"reasoning_tokens": reasoning_tokens,
|
|
"total_tokens": input_tokens + output_tokens,
|
|
},
|
|
"timing": timing,
|
|
"iterations": iterations,
|
|
}
|
|
|
|
|
|
def build_claude_partial_response(output):
|
|
"""Best-effort counterpart to build_partial_response() for Claude's
|
|
event shape -- never raises. Concatenates every text content block
|
|
from every assistant event seen before an iteration-limit kill, or ""
|
|
if none arrived."""
|
|
parts = []
|
|
for line in output.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
event = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if event.get("type") != "assistant":
|
|
continue
|
|
for block in event.get("message", {}).get("content", []):
|
|
if block.get("type") == "text" and block.get("text"):
|
|
parts.append(block["text"])
|
|
return "\n\n".join(parts).strip()
|
|
|
|
|
|
def run_claude_streaming(command, body, environment, cwd, timeout, iteration_limit):
|
|
"""Claude Code counterpart to run_streaming(): same Popen/drain-thread/
|
|
deadline/SIGTERM-then-SIGKILL structure, two differences forced by
|
|
Claude's own --output-format stream-json shape rather than Codex's:
|
|
|
|
- `cwd` is an explicit parameter and passed to subprocess.Popen
|
|
directly -- Claude has no -C-equivalent CLI flag to set it via argv.
|
|
- Iterations are counted as tool_use *content blocks*, not JSONL
|
|
lines: a single type=="assistant" event's message.content list can
|
|
carry more than one tool_use block (parallel tool calls), each one
|
|
a separate unit of work, so each is counted individually. This is
|
|
not a 1:1 semantic match for Codex's item.completed counting (which
|
|
also counts pure-text turns with no tool call at all) -- see
|
|
docs/intent/builder-iterationslimit.md's note on this being a known
|
|
per-runtime difference.
|
|
|
|
Returns (stdout_text, limit_reached, returncode).
|
|
"""
|
|
process = subprocess.Popen(
|
|
command,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
env=environment,
|
|
cwd=cwd,
|
|
)
|
|
process.stdin.write(body)
|
|
process.stdin.close()
|
|
|
|
output_queue = queue.Queue()
|
|
reader = threading.Thread(
|
|
target=_drain_lines, args=(process.stdout, output_queue), daemon=True
|
|
)
|
|
reader.start()
|
|
|
|
lines = []
|
|
iterations = 0
|
|
limit_reached = False
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
process.kill()
|
|
process.wait()
|
|
raise subprocess.TimeoutExpired(command, timeout)
|
|
try:
|
|
line = output_queue.get(timeout=min(remaining, 1.0))
|
|
except queue.Empty:
|
|
continue
|
|
if line is None:
|
|
break
|
|
lines.append(line)
|
|
stripped = line.strip()
|
|
if not stripped:
|
|
continue
|
|
try:
|
|
event = json.loads(stripped)
|
|
except ValueError:
|
|
continue
|
|
if event.get("type") == "assistant":
|
|
tool_calls = sum(
|
|
1
|
|
for block in event.get("message", {}).get("content", [])
|
|
if block.get("type") == "tool_use"
|
|
)
|
|
if tool_calls:
|
|
iterations += tool_calls
|
|
if iterations > iteration_limit:
|
|
limit_reached = True
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
break
|
|
|
|
returncode = process.wait()
|
|
stderr = process.stderr.read()
|
|
if stderr:
|
|
print(stderr.rstrip(), file=sys.stderr)
|
|
return "".join(lines), limit_reached, returncode
|
|
|
|
|
|
def append_claude_model_options(command, model=None, effort=None):
|
|
"""Claude counterpart to append_model_options(): real --model/--effort
|
|
flags rather than Codex's -c model_reasoning_effort= config override,
|
|
so both are plain trailing flags with no exec-subcommand-boundary
|
|
concern."""
|
|
model = (os.environ.get("AGENT_BUILDER_CLAUDE_MODEL", "") if model is None else model).strip()
|
|
if model:
|
|
if len(model) > 128 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:+-]*", model):
|
|
raise ValueError("AGENT_BUILDER_CLAUDE_MODEL enthält ungültige Zeichen.")
|
|
command.extend(["--model", model])
|
|
|
|
effort = (os.environ.get("AGENT_BUILDER_CLAUDE_REASONING_EFFORT", "") if effort is None else effort).strip()
|
|
if effort:
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", effort):
|
|
raise ValueError("AGENT_BUILDER_CLAUDE_REASONING_EFFORT enthält ungültige Zeichen.")
|
|
command.extend(["--effort", effort])
|
|
|
|
|
|
def extract_commit_message(response, start_marker, end_marker):
|
|
start = response.find(start_marker)
|
|
end = response.find(end_marker)
|
|
if start == -1 or end == -1 or end < start:
|
|
return response, None
|
|
|
|
message = response[start + len(start_marker) : end].strip()
|
|
before = response[:start].strip()
|
|
after = response[end + len(end_marker) :].strip()
|
|
visible = "\n\n".join(part for part in (before, after) if part)
|
|
if not message:
|
|
return visible or response, None
|
|
return visible or response[:start].strip() or "Änderung committed.", message
|
|
|
|
|
|
def repo_has_changes(repo_dir):
|
|
result = subprocess.run(
|
|
["git", "-C", str(repo_dir), "status", "--porcelain"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
return bool(result.stdout.strip())
|
|
|
|
|
|
def commit_and_push(repo_dir, message, branch=None):
|
|
"""Push to `branch`, or -- if not given -- whatever branch is
|
|
currently checked out (nixos-config and website use dev, but
|
|
dashboard and launcher only have main, and that's what their deploy
|
|
webhooks actually pull). An explicit `branch` is for isolated builder
|
|
worktrees (see docs/intent/board-builder-worktree-parallelism.md):
|
|
their own checked-out branch is a disposable per-job branch, never
|
|
the real target, so the caller must name the target explicitly
|
|
instead of relying on the worktree's own HEAD."""
|
|
if branch is None:
|
|
branch_result = subprocess.run(
|
|
["git", "-C", str(repo_dir), "rev-parse", "--abbrev-ref", "HEAD"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
check=False,
|
|
)
|
|
branch = branch_result.stdout.strip()
|
|
if branch_result.returncode != 0 or not branch:
|
|
return False, (branch_result.stdout + branch_result.stderr).strip(), None
|
|
|
|
for command in (
|
|
["git", "-C", str(repo_dir), "add", "-A"],
|
|
["git", "-C", str(repo_dir), "commit", "-m", message],
|
|
["git", "-C", str(repo_dir), "push", "origin", f"HEAD:{branch}"],
|
|
):
|
|
result = subprocess.run(
|
|
command,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
return False, (result.stdout + result.stderr).strip(), None
|
|
log = subprocess.run(
|
|
["git", "-C", str(repo_dir), "log", "-1", "--format=%h"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
check=False,
|
|
)
|
|
return True, log.stdout.strip(), branch
|
|
|
|
|
|
def send_reply(config_env_var, sender, response):
|
|
command = [
|
|
os.environ["AGENT_SENDXMPP_BIN"],
|
|
"--file",
|
|
os.environ[config_env_var],
|
|
sender,
|
|
]
|
|
result = subprocess.run(
|
|
command,
|
|
input=response,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=30,
|
|
check=False,
|
|
)
|
|
if result.stderr:
|
|
print(result.stderr.rstrip(), file=sys.stderr)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"XMPP-Antwort endete mit Status {result.returncode}.")
|
|
|
|
|
|
def fail_job(path, failed, payload, error):
|
|
payload["failed_at"] = time.time()
|
|
payload["error"] = str(error)
|
|
write_payload(path, payload)
|
|
failed.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
path.replace(failed / path.name)
|