weichen
71fd4486f5
Why: Codex CLI is permitted only after every normal execution profile exhausts three safe attempts, with no host implementation fallback. Evidence: The emergency RED test lacked the fourth Codex attempt before implementation; 32 controller/history executions now pass. Impact: Adds one emergency attempt for implementation, repair, and test roles while preserving fail-stop statuses and review isolation. Refs: feature/agent-cli-enforcement
2122 lines
71 KiB
Python
2122 lines
71 KiB
Python
"""Recoverable agent-exec controller: validate, isolate, supervise, receipt, transport."""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import select
|
|
import shutil
|
|
import signal
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
OUTPUT_TAIL_BYTES = 65536
|
|
|
|
_REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
if str(_REPO_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(_REPO_ROOT))
|
|
|
|
from scripts.agent_exec_adapters import build_invocation, probe_adapter
|
|
from scripts.agent_exec_config import (
|
|
DEFAULT_IDLE_TIMEOUT_SECONDS,
|
|
SECRET_VALUE_RE,
|
|
ConfigError,
|
|
ExecutionConfig,
|
|
Profile,
|
|
load_config,
|
|
validate_config,
|
|
)
|
|
|
|
RECEIPT_SCHEMA = "agent-exec-receipt/v1"
|
|
HISTORY_SCHEMA = "agent-exec-history/v1"
|
|
|
|
# Distinct stable exit codes.
|
|
EXIT_SUCCESS = 0
|
|
EXIT_INVALID_CONFIG = 2
|
|
EXIT_UNAVAILABLE_EXHAUSTION = 3
|
|
EXIT_BLOCKED = 4
|
|
EXIT_SCOPE_EXPANSION = 5
|
|
EXIT_CANCELLED = 6
|
|
EXIT_HOST_FALLBACK = 10
|
|
|
|
_EMERGENCY_ROLES = frozenset({"implementation", "repair", "test"})
|
|
_SAFE_EXHAUST_STATUSES = frozenset({"unavailable", "failed-clean"})
|
|
|
|
# Transport commits must not depend on machine-global Git identity.
|
|
TRANSPORT_AUTHOR_NAME = "agent-exec-transport"
|
|
TRANSPORT_AUTHOR_EMAIL = "agent-exec-transport@localhost.invalid"
|
|
|
|
DEFAULT_BRANCHES = frozenset({"main", "master"})
|
|
SECRET_PATH_MARKERS = (
|
|
".secrets",
|
|
"credentials",
|
|
"id_rsa",
|
|
"id_ed25519",
|
|
".env",
|
|
"ACCESS_TOKEN",
|
|
)
|
|
ENV_ALLOW_PREFIXES = ("LC_",)
|
|
ENV_ALLOW_EXACT = frozenset(
|
|
{
|
|
"PATH",
|
|
"HOME",
|
|
"USER",
|
|
"LOGNAME",
|
|
"LANG",
|
|
"LANGUAGE",
|
|
"TMPDIR",
|
|
"TMP",
|
|
"TEMP",
|
|
"TERM",
|
|
"TZ",
|
|
"XDG_RUNTIME_DIR",
|
|
"XDG_CACHE_HOME",
|
|
"XDG_CONFIG_HOME",
|
|
"XDG_DATA_HOME",
|
|
"SSL_CERT_FILE",
|
|
"SSL_CERT_DIR",
|
|
"REQUESTS_CA_BUNDLE",
|
|
"CURL_CA_BUNDLE",
|
|
"HTTP_PROXY",
|
|
"HTTPS_PROXY",
|
|
"NO_PROXY",
|
|
"http_proxy",
|
|
"https_proxy",
|
|
"no_proxy",
|
|
"ALL_PROXY",
|
|
"all_proxy",
|
|
# Provider discovery names only — values never enter receipts.
|
|
"OPENAI_API_KEY",
|
|
"ANTHROPIC_API_KEY",
|
|
"CURSOR_API_KEY",
|
|
"AGENT_EXEC_TEST_SECRET",
|
|
}
|
|
)
|
|
|
|
UNIT_REQUIRED = (
|
|
"unit_id",
|
|
"goal",
|
|
"allowed_paths",
|
|
"verification",
|
|
"constraints",
|
|
"network",
|
|
"external_cost",
|
|
"timeout_seconds",
|
|
)
|
|
|
|
|
|
class UnitPacketError(ValueError):
|
|
"""Raised when a unit packet fails contract checks."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class UnitPacket:
|
|
unit_id: str
|
|
goal: str
|
|
allowed_paths: tuple[str, ...]
|
|
verification: tuple[Any, ...]
|
|
constraints: tuple[Any, ...]
|
|
network: bool
|
|
external_cost: bool
|
|
timeout_seconds: int
|
|
raw: dict[str, Any]
|
|
digest: str
|
|
|
|
|
|
@dataclass
|
|
class AttemptRecord:
|
|
attempt_id: str
|
|
profile: str
|
|
adapter: str
|
|
terminal_status: str
|
|
reason: str | None = None
|
|
worktree: str | None = None
|
|
pid: int | None = None
|
|
exit_code: int | None = None
|
|
timed_out: bool = False
|
|
idle_timed_out: bool = False
|
|
signal: int | None = None
|
|
changed_paths: list[str] = field(default_factory=list)
|
|
inventory_paths: list[str] = field(default_factory=list)
|
|
raw_log: str | None = None
|
|
summary: str = ""
|
|
cleanup: dict[str, Any] = field(default_factory=dict)
|
|
workspace_preserved: bool = False
|
|
transport_ref: str | None = None
|
|
observed_version: str | None = None
|
|
attempt_ordinal: int = 1
|
|
emergency_fallback: bool = False
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="agent_exec")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
p_val = sub.add_parser("validate", help="Validate AGENT-EXECUTION.md (no provider calls)")
|
|
p_val.add_argument("--repo", required=True)
|
|
|
|
p_run = sub.add_parser("run", help="Run one bounded unit through configured profiles")
|
|
p_run.add_argument("--repo", required=True)
|
|
p_run.add_argument("--role", required=True)
|
|
p_run.add_argument("--unit", required=True)
|
|
p_run.add_argument("--receipt", required=True)
|
|
|
|
args = parser.parse_args(argv)
|
|
if args.command == "validate":
|
|
return cmd_validate(Path(args.repo))
|
|
if args.command == "run":
|
|
return cmd_run(
|
|
Path(args.repo),
|
|
role=args.role,
|
|
unit_path=Path(args.unit),
|
|
receipt_path=Path(args.receipt),
|
|
)
|
|
return EXIT_INVALID_CONFIG
|
|
|
|
|
|
def cmd_validate(repo: Path) -> int:
|
|
try:
|
|
cfg = load_config(repo)
|
|
validate_config(repo)
|
|
except ConfigError as exc:
|
|
print(f"invalid-config: {exc}", file=sys.stderr)
|
|
return EXIT_INVALID_CONFIG
|
|
# Local capability probes only — never contact a provider.
|
|
for name, profile in cfg.profiles.items():
|
|
if not profile.enabled:
|
|
continue
|
|
probe_adapter(profile, cfg)
|
|
print("validate: ok")
|
|
return EXIT_SUCCESS
|
|
|
|
|
|
def _apply_attempt_route(
|
|
attempt: AttemptRecord,
|
|
*,
|
|
attempt_id: str,
|
|
parent: str,
|
|
terminal: str,
|
|
reason: str | None,
|
|
changed_paths: list[str],
|
|
transport: dict[str, Any] | None,
|
|
) -> tuple[str, str, str | None, list[str], dict[str, Any] | None]:
|
|
status = attempt.terminal_status
|
|
if status == "success":
|
|
return (
|
|
"stop",
|
|
"success",
|
|
None,
|
|
list(attempt.changed_paths),
|
|
{
|
|
"ref": attempt.transport_ref,
|
|
"parent": parent,
|
|
"attempt_id": attempt_id,
|
|
},
|
|
)
|
|
if status in _SAFE_EXHAUST_STATUSES:
|
|
return ("continue", "unavailable", attempt.reason, changed_paths, transport)
|
|
if status == "scope-expansion":
|
|
return (
|
|
"stop",
|
|
"scope-expansion",
|
|
attempt.reason,
|
|
list(attempt.changed_paths),
|
|
transport,
|
|
)
|
|
if status in {"failed-unknown", "blocked"}:
|
|
return ("stop", status, attempt.reason, changed_paths, transport)
|
|
if status == "cancelled":
|
|
return ("stop", "cancelled", attempt.reason, changed_paths, transport)
|
|
return ("next-profile", terminal, reason, changed_paths, transport)
|
|
|
|
|
|
def cmd_run(
|
|
repo: Path,
|
|
*,
|
|
role: str,
|
|
unit_path: Path,
|
|
receipt_path: Path,
|
|
) -> int:
|
|
started_monotonic = time.monotonic()
|
|
run_id = uuid.uuid4().hex
|
|
attempts: list[AttemptRecord] = []
|
|
secret_values: set[str] = set()
|
|
_collect_secret_values(secret_values)
|
|
|
|
try:
|
|
cfg = load_config(repo)
|
|
except ConfigError as exc:
|
|
receipt = _base_receipt(
|
|
run_id=run_id,
|
|
repo=repo,
|
|
role=role,
|
|
unit_id=None,
|
|
config=None,
|
|
terminal_status="invalid-config",
|
|
reason=str(exc),
|
|
attempts=[],
|
|
secret_values=secret_values,
|
|
)
|
|
return _finish_run(
|
|
repo=repo,
|
|
receipt_path=receipt_path,
|
|
receipt=receipt,
|
|
secret_values=secret_values,
|
|
summary_bytes=256,
|
|
exit_code=EXIT_INVALID_CONFIG,
|
|
started_monotonic=started_monotonic,
|
|
)
|
|
|
|
try:
|
|
base = _require_canonical_ready(repo)
|
|
unit = load_unit_packet(unit_path, cfg)
|
|
profiles = _select_profiles(cfg, role)
|
|
except (UnitPacketError, ConfigError, RuntimeError, ValueError) as exc:
|
|
receipt = _base_receipt(
|
|
run_id=run_id,
|
|
repo=repo,
|
|
role=role,
|
|
unit_id=None,
|
|
config=cfg,
|
|
terminal_status="invalid-config",
|
|
reason=str(exc),
|
|
attempts=[],
|
|
secret_values=secret_values,
|
|
)
|
|
return _finish_run(
|
|
repo=repo,
|
|
receipt_path=receipt_path,
|
|
receipt=receipt,
|
|
secret_values=secret_values,
|
|
summary_bytes=cfg.execution.summary_bytes,
|
|
exit_code=EXIT_INVALID_CONFIG,
|
|
started_monotonic=started_monotonic,
|
|
)
|
|
|
|
try:
|
|
_validate_allowed_path_symlinks(repo, unit.allowed_paths)
|
|
except UnitPacketError as exc:
|
|
# Symlink escape is a safety failure: stop routing, never authorize fallback.
|
|
receipt = _base_receipt(
|
|
run_id=run_id,
|
|
repo=repo,
|
|
role=role,
|
|
unit_id=unit.unit_id,
|
|
config=cfg,
|
|
terminal_status="failed-unknown",
|
|
reason=str(exc),
|
|
attempts=[],
|
|
secret_values=secret_values,
|
|
)
|
|
return _finish_run(
|
|
repo=repo,
|
|
receipt_path=receipt_path,
|
|
receipt=receipt,
|
|
secret_values=secret_values,
|
|
summary_bytes=cfg.execution.summary_bytes,
|
|
exit_code=EXIT_BLOCKED,
|
|
started_monotonic=started_monotonic,
|
|
)
|
|
|
|
run_dir = _create_private_run_dir(run_id)
|
|
config_digest = _file_digest(repo / "AGENT-EXECUTION.md")
|
|
|
|
terminal = "unavailable"
|
|
reason: str | None = None
|
|
transport: dict[str, Any] | None = None
|
|
changed_paths: list[str] = []
|
|
host_scope: dict[str, Any] | None = None
|
|
host_fallback_authorizations = 0
|
|
stop = False
|
|
|
|
max_attempts = cfg.execution.max_attempts_per_profile
|
|
for profile_name in profiles:
|
|
if stop:
|
|
break
|
|
profile = cfg.profiles[profile_name]
|
|
for attempt_ordinal in range(1, max_attempts + 1):
|
|
attempt_id = uuid.uuid4().hex[:12]
|
|
attempt = _run_attempt(
|
|
repo=repo,
|
|
cfg=cfg,
|
|
profile=profile,
|
|
unit=unit,
|
|
unit_path=unit_path,
|
|
base=base,
|
|
run_id=run_id,
|
|
attempt_id=attempt_id,
|
|
run_dir=run_dir,
|
|
secret_values=secret_values,
|
|
)
|
|
attempt.attempt_ordinal = attempt_ordinal
|
|
attempt.emergency_fallback = False
|
|
attempts.append(attempt)
|
|
action, terminal, reason, changed_paths, transport = _apply_attempt_route(
|
|
attempt,
|
|
attempt_id=attempt_id,
|
|
parent=base["head"],
|
|
terminal=terminal,
|
|
reason=reason,
|
|
changed_paths=changed_paths,
|
|
transport=transport,
|
|
)
|
|
if action == "continue":
|
|
continue
|
|
stop = action == "stop"
|
|
break
|
|
|
|
exhausted_ok = (
|
|
not stop
|
|
and attempts
|
|
and all(a.terminal_status in _SAFE_EXHAUST_STATUSES for a in attempts)
|
|
and len(attempts) == len(profiles) * max_attempts
|
|
)
|
|
emergency_name = cfg.execution.emergency_profile
|
|
if exhausted_ok and emergency_name and role in _EMERGENCY_ROLES:
|
|
emergency_profile = cfg.profiles.get(emergency_name)
|
|
if emergency_profile is None:
|
|
terminal = "blocked"
|
|
reason = f"emergency profile {emergency_name!r} is not configured"
|
|
else:
|
|
attempt_id = uuid.uuid4().hex[:12]
|
|
attempt = _run_attempt(
|
|
repo=repo,
|
|
cfg=cfg,
|
|
profile=emergency_profile,
|
|
unit=unit,
|
|
unit_path=unit_path,
|
|
base=base,
|
|
run_id=run_id,
|
|
attempt_id=attempt_id,
|
|
run_dir=run_dir,
|
|
secret_values=secret_values,
|
|
)
|
|
attempt.attempt_ordinal = 1
|
|
attempt.emergency_fallback = True
|
|
attempts.append(attempt)
|
|
action, terminal, reason, changed_paths, transport = _apply_attempt_route(
|
|
attempt,
|
|
attempt_id=attempt_id,
|
|
parent=base["head"],
|
|
terminal=terminal,
|
|
reason=reason,
|
|
changed_paths=changed_paths,
|
|
transport=transport,
|
|
)
|
|
if action == "continue":
|
|
terminal = "blocked"
|
|
reason = attempt.reason
|
|
stop = True
|
|
|
|
if (
|
|
exhausted_ok
|
|
and not stop
|
|
and not emergency_name
|
|
and cfg.execution.fallback_to_host == "all-unavailable"
|
|
):
|
|
terminal = "host-fallback-authorized"
|
|
host_fallback_authorizations = 1
|
|
host_scope = {
|
|
"allowed_paths": list(unit.allowed_paths),
|
|
"unit_id": unit.unit_id,
|
|
"goal": unit.goal,
|
|
"verification": list(unit.verification),
|
|
"constraints": list(unit.constraints),
|
|
"network": unit.network,
|
|
"external_cost": unit.external_cost,
|
|
}
|
|
reason = "all configured profiles exhausted with unavailable or failed-clean"
|
|
|
|
summary = _bounded_summary(
|
|
_build_run_summary(terminal, attempts, reason),
|
|
cfg.execution.summary_bytes,
|
|
secret_values,
|
|
)
|
|
receipt = _base_receipt(
|
|
run_id=run_id,
|
|
repo=repo,
|
|
role=role,
|
|
unit_id=unit.unit_id,
|
|
config=cfg,
|
|
terminal_status=terminal,
|
|
reason=reason,
|
|
attempts=attempts,
|
|
secret_values=secret_values,
|
|
)
|
|
receipt.update(
|
|
{
|
|
"unit_digest": unit.digest,
|
|
"config_digest": config_digest,
|
|
"base": {"head": base["head"], "tree": base["tree"], "branch": base["branch"]},
|
|
"summary": summary,
|
|
"changed_paths": changed_paths,
|
|
"transport": transport,
|
|
"host_scope": host_scope,
|
|
"host_fallback_authorizations": host_fallback_authorizations,
|
|
"run_dir": str(run_dir),
|
|
"private_root_mode": oct(run_dir.parent.stat().st_mode & 0o777),
|
|
}
|
|
)
|
|
return _finish_run(
|
|
repo=repo,
|
|
receipt_path=receipt_path,
|
|
receipt=receipt,
|
|
secret_values=secret_values,
|
|
summary_bytes=cfg.execution.summary_bytes,
|
|
exit_code=_exit_for_terminal(terminal),
|
|
started_monotonic=started_monotonic,
|
|
)
|
|
|
|
|
|
def load_unit_packet(path: Path, cfg: ExecutionConfig) -> UnitPacket:
|
|
try:
|
|
raw_text = path.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise UnitPacketError(f"unable to read unit packet: {exc}") from exc
|
|
try:
|
|
raw = json.loads(raw_text)
|
|
except json.JSONDecodeError as exc:
|
|
raise UnitPacketError(f"unit packet is not JSON: {exc}") from exc
|
|
if not isinstance(raw, dict):
|
|
raise UnitPacketError("unit packet must be an object")
|
|
missing = [k for k in UNIT_REQUIRED if k not in raw]
|
|
if missing:
|
|
raise UnitPacketError(f"missing unit packet fields: {missing}")
|
|
|
|
unit_id = raw["unit_id"]
|
|
goal = raw["goal"]
|
|
if not isinstance(unit_id, str) or not unit_id.strip():
|
|
raise UnitPacketError("unit_id must be a non-empty string")
|
|
if not isinstance(goal, str) or not goal.strip():
|
|
raise UnitPacketError("goal must be a non-empty string")
|
|
|
|
allowed = raw["allowed_paths"]
|
|
if not isinstance(allowed, list) or not allowed:
|
|
raise UnitPacketError("allowed_paths must be a non-empty list")
|
|
paths: list[str] = []
|
|
for item in allowed:
|
|
if not isinstance(item, str) or not item.strip():
|
|
raise UnitPacketError("allowed_paths entries must be non-empty strings")
|
|
_validate_allowed_path(item)
|
|
paths.append(item)
|
|
|
|
for key in ("verification", "constraints"):
|
|
if not isinstance(raw[key], list):
|
|
raise UnitPacketError(f"{key} must be a list")
|
|
for key in ("network", "external_cost"):
|
|
if type(raw[key]) is not bool:
|
|
raise UnitPacketError(f"{key} must be a boolean")
|
|
|
|
timeout = raw["timeout_seconds"]
|
|
if type(timeout) is not int or isinstance(timeout, bool) or timeout < 1:
|
|
raise UnitPacketError("timeout_seconds must be a positive integer")
|
|
if timeout > cfg.execution.timeout_seconds:
|
|
raise UnitPacketError(
|
|
f"timeout_seconds {timeout} exceeds project policy "
|
|
f"{cfg.execution.timeout_seconds}"
|
|
)
|
|
|
|
digest = hashlib.sha256(raw_text.encode("utf-8")).hexdigest()
|
|
return UnitPacket(
|
|
unit_id=unit_id.strip(),
|
|
goal=goal.strip(),
|
|
allowed_paths=tuple(paths),
|
|
verification=tuple(raw["verification"]),
|
|
constraints=tuple(raw["constraints"]),
|
|
network=bool(raw["network"]),
|
|
external_cost=bool(raw["external_cost"]),
|
|
timeout_seconds=timeout,
|
|
raw=raw,
|
|
digest=digest,
|
|
)
|
|
|
|
|
|
def _validate_allowed_path(path: str) -> None:
|
|
if path.startswith("/") or path.startswith("~"):
|
|
raise UnitPacketError(f"absolute allowed path rejected: {path!r}")
|
|
parts = Path(path).parts
|
|
if ".." in parts:
|
|
raise UnitPacketError(f"parent traversal rejected: {path!r}")
|
|
if ".git" in parts or path == ".git" or path.startswith(".git/"):
|
|
raise UnitPacketError(f".git path rejected: {path!r}")
|
|
lowered = path.lower()
|
|
for marker in SECRET_PATH_MARKERS:
|
|
if marker.lower() in lowered:
|
|
raise UnitPacketError(f"secret-looking path rejected: {path!r}")
|
|
|
|
|
|
def _select_profiles(cfg: ExecutionConfig, role: str) -> list[str]:
|
|
if role not in cfg.roles:
|
|
raise ConfigError(f"unknown role: {role!r}")
|
|
names: list[str] = []
|
|
for name in cfg.roles[role]:
|
|
profile = cfg.profiles[name]
|
|
if not profile.enabled:
|
|
continue
|
|
names.append(name)
|
|
if not names:
|
|
raise ConfigError(f"role {role!r} has no enabled profiles")
|
|
return names
|
|
|
|
|
|
def _require_canonical_ready(repo: Path) -> dict[str, str]:
|
|
repo = repo.resolve()
|
|
if not (repo / ".git").exists():
|
|
raise RuntimeError(f"not a git repository: {repo}")
|
|
status = _git(repo, "status", "--porcelain")
|
|
if status.stdout.strip():
|
|
raise RuntimeError("canonical checkout is dirty")
|
|
branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
|
|
if branch in DEFAULT_BRANCHES:
|
|
raise RuntimeError(f"canonical checkout must not be on default branch {branch}")
|
|
if branch == "HEAD":
|
|
# Detached HEAD is acceptable only as a fixed base; still ok for tests that
|
|
# check out a branch. Prefer named non-default branches.
|
|
pass
|
|
head = _git(repo, "rev-parse", "HEAD").stdout.strip()
|
|
tree = _git(repo, "rev-parse", "HEAD^{tree}").stdout.strip()
|
|
return {"head": head, "tree": tree, "branch": branch}
|
|
|
|
|
|
def _create_private_run_dir(run_id: str) -> Path:
|
|
root = Path(f"/tmp/agent-exec-{os.getuid()}")
|
|
if root.exists():
|
|
if root.is_symlink():
|
|
raise RuntimeError(f"private root must not be a symlink: {root}")
|
|
if not root.is_dir():
|
|
raise RuntimeError(f"private root is not a directory: {root}")
|
|
st = root.stat()
|
|
if st.st_uid != os.getuid():
|
|
raise RuntimeError(f"private root owned by another user: {root}")
|
|
os.chmod(root, 0o700)
|
|
else:
|
|
root.mkdir(mode=0o700)
|
|
os.chmod(root, 0o700)
|
|
run_dir = root / run_id
|
|
run_dir.mkdir(mode=0o700)
|
|
os.chmod(run_dir, 0o700)
|
|
if run_dir.is_symlink():
|
|
raise RuntimeError(f"run dir must not be a symlink: {run_dir}")
|
|
return run_dir
|
|
|
|
|
|
def _create_private_attempt_clone(repo: Path, dest: Path, base_head: str) -> None:
|
|
"""Create a private local shared clone with isolated Git administration.
|
|
|
|
Objects are reused read-only via ``git clone --shared`` alternates. The clone
|
|
owns its own ``.git`` config, refs, index, HEAD, and worktree. Never contacts
|
|
a remote.
|
|
"""
|
|
if dest.exists():
|
|
shutil.rmtree(dest)
|
|
proc = subprocess.run(
|
|
[
|
|
"git",
|
|
"clone",
|
|
"--shared",
|
|
"--no-checkout",
|
|
"--",
|
|
str(repo.resolve()),
|
|
str(dest),
|
|
],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
if proc.returncode != 0:
|
|
raise subprocess.CalledProcessError(
|
|
proc.returncode, proc.args, proc.stdout, proc.stderr
|
|
)
|
|
# Drop the auto-configured origin so the attempt cannot reach a remote URL.
|
|
_git(dest, "remote", "remove", "origin", check=False)
|
|
_git(dest, "checkout", "--detach", base_head)
|
|
|
|
|
|
def _run_attempt(
|
|
*,
|
|
repo: Path,
|
|
cfg: ExecutionConfig,
|
|
profile: Profile,
|
|
unit: UnitPacket,
|
|
unit_path: Path,
|
|
base: dict[str, str],
|
|
run_id: str,
|
|
attempt_id: str,
|
|
run_dir: Path,
|
|
secret_values: set[str],
|
|
) -> AttemptRecord:
|
|
probe = probe_adapter(profile, cfg)
|
|
if not probe.available:
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="unavailable",
|
|
reason=probe.reason or "adapter unavailable",
|
|
observed_version=probe.observed_version,
|
|
cleanup={"exact_restore": True, "skipped": True},
|
|
)
|
|
|
|
worktree = run_dir / f"wt-{attempt_id}"
|
|
try:
|
|
_create_private_attempt_clone(repo, worktree, base["head"])
|
|
except subprocess.CalledProcessError as exc:
|
|
err = (exc.stderr or str(exc)).strip()
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="unavailable",
|
|
reason=f"private clone create failed: {err}",
|
|
cleanup={"exact_restore": True, "skipped": True},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
except OSError as exc:
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="unavailable",
|
|
reason=f"private clone create failed: {exc}",
|
|
cleanup={"exact_restore": True, "skipped": True},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
result_path = run_dir / f"result-{attempt_id}.json"
|
|
packet_copy = run_dir / f"unit-{attempt_id}.json"
|
|
shutil.copy2(unit_path, packet_copy)
|
|
raw_log = run_dir / f"raw-{attempt_id}.log"
|
|
|
|
try:
|
|
invocation = build_invocation(
|
|
profile, cfg, worktree, packet_copy, result_path
|
|
)
|
|
except ValueError as exc:
|
|
_safe_remove_worktree(repo, worktree)
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="unavailable",
|
|
reason=str(exc),
|
|
worktree=str(worktree),
|
|
cleanup={"exact_restore": True},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
try:
|
|
_validate_allowed_path_symlinks(worktree, unit.allowed_paths)
|
|
except UnitPacketError as exc:
|
|
_safe_remove_worktree(repo, worktree)
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="failed-unknown",
|
|
reason=str(exc),
|
|
worktree=str(worktree),
|
|
cleanup={"exact_restore": True, "reason": "symlink escape before launch"},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
admin_before = _git_admin_snapshot(worktree, repo)
|
|
pointer_before = _worktree_pointer_snapshot(worktree, require_git=True)
|
|
canonical_before = _canonical_checkout_snapshot(repo)
|
|
timeout = min(unit.timeout_seconds, cfg.execution.timeout_seconds)
|
|
idle_timeout = min(cfg.execution.idle_timeout_seconds, timeout)
|
|
proc_meta = _supervise(
|
|
invocation.argv,
|
|
cwd=invocation.cwd,
|
|
stdin=invocation.stdin,
|
|
timeout=timeout,
|
|
idle_timeout=idle_timeout,
|
|
raw_log=raw_log,
|
|
raw_cap=cfg.execution.raw_output_bytes,
|
|
secret_values=secret_values,
|
|
)
|
|
|
|
changed: list[str] = []
|
|
inventory: dict[str, Any] = {"paths": [], "details": []}
|
|
try:
|
|
pointer_after = _worktree_pointer_snapshot(worktree, require_git=True)
|
|
admin_after = _git_admin_snapshot(worktree, repo)
|
|
canonical_after = _canonical_checkout_snapshot(repo)
|
|
inventory = _inventory_changes(worktree)
|
|
changed = list(inventory["paths"])
|
|
except Exception as exc:
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="failed-unknown",
|
|
reason=f"post-launch git inspection failed: {exc}",
|
|
worktree=str(worktree),
|
|
pid=proc_meta["pid"],
|
|
exit_code=proc_meta["exit_code"],
|
|
timed_out=proc_meta["timed_out"],
|
|
idle_timed_out=bool(proc_meta.get("idle_timed_out")),
|
|
signal=proc_meta["signal"],
|
|
changed_paths=[],
|
|
inventory_paths=[],
|
|
raw_log=str(raw_log),
|
|
summary=_bounded_summary(
|
|
proc_meta.get("summary_source") or str(exc),
|
|
cfg.execution.summary_bytes,
|
|
secret_values,
|
|
),
|
|
workspace_preserved=True,
|
|
cleanup={
|
|
"exact_restore": False,
|
|
"reason": "post-launch snapshot or inventory failure",
|
|
"error": str(exc),
|
|
"stdout_tail_bytes": len(proc_meta.get("stdout_tail") or ""),
|
|
"stderr_tail_bytes": len(proc_meta.get("stderr_tail") or ""),
|
|
},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
admin_delta = _git_admin_delta(admin_before, admin_after)
|
|
pointer_delta = _snapshot_delta(pointer_before, pointer_after)
|
|
canonical_delta = _snapshot_delta(canonical_before, canonical_after)
|
|
|
|
worker_result = _load_worker_result(result_path, proc_meta["stdout_tail"])
|
|
summary = _bounded_summary(
|
|
proc_meta.get("summary_source") or (worker_result or {}).get("summary") or "",
|
|
cfg.execution.summary_bytes,
|
|
secret_values,
|
|
)
|
|
|
|
def _preserved(
|
|
status: str,
|
|
reason: str,
|
|
*,
|
|
exact_restore: bool = False,
|
|
cleanup_extra: dict[str, Any] | None = None,
|
|
) -> AttemptRecord:
|
|
cleanup: dict[str, Any] = {"exact_restore": exact_restore}
|
|
if cleanup_extra:
|
|
cleanup.update(cleanup_extra)
|
|
cleanup.setdefault(
|
|
"stdout_tail_bytes", len(proc_meta.get("stdout_tail") or "")
|
|
)
|
|
cleanup.setdefault(
|
|
"stderr_tail_bytes", len(proc_meta.get("stderr_tail") or "")
|
|
)
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status=status,
|
|
reason=reason,
|
|
worktree=str(worktree),
|
|
pid=proc_meta["pid"],
|
|
exit_code=proc_meta["exit_code"],
|
|
timed_out=proc_meta["timed_out"],
|
|
idle_timed_out=bool(proc_meta.get("idle_timed_out")),
|
|
signal=proc_meta["signal"],
|
|
changed_paths=list(changed),
|
|
inventory_paths=list(changed),
|
|
raw_log=str(raw_log),
|
|
summary=summary,
|
|
workspace_preserved=not exact_restore,
|
|
cleanup=cleanup,
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
if pointer_delta:
|
|
return _preserved(
|
|
"failed-unknown",
|
|
f"worktree .git pointer/admin identity changed: {', '.join(pointer_delta)}",
|
|
cleanup_extra={
|
|
"reason": "worktree git pointer mismatch",
|
|
"pointer_delta": pointer_delta,
|
|
},
|
|
)
|
|
|
|
# Git administration snapshot dominates worker prose.
|
|
if admin_delta:
|
|
return _preserved(
|
|
"failed-unknown",
|
|
f"unexplained git administration change: {', '.join(admin_delta)}",
|
|
cleanup_extra={"reason": "git admin snapshot mismatch", "admin_delta": admin_delta},
|
|
)
|
|
|
|
if canonical_delta:
|
|
return _preserved(
|
|
"failed-unknown",
|
|
f"canonical checkout mutated: {', '.join(canonical_delta)}",
|
|
cleanup_extra={
|
|
"reason": "canonical checkout snapshot mismatch",
|
|
"canonical_delta": canonical_delta,
|
|
},
|
|
)
|
|
|
|
escaped = [p for p in changed if _path_symlink_escapes(worktree, p)]
|
|
if escaped:
|
|
return _preserved(
|
|
"failed-unknown",
|
|
f"changed path escapes worktree via symlink: {escaped}",
|
|
cleanup_extra={"reason": "symlink escape on changed paths"},
|
|
)
|
|
|
|
out_of_scope = [p for p in changed if not _path_allowed(p, unit.allowed_paths)]
|
|
worker_requests_expand = bool(
|
|
worker_result
|
|
and worker_result.get("terminal_status") in {"scope_expansion", "scope-expansion"}
|
|
)
|
|
if out_of_scope or worker_requests_expand:
|
|
return _preserved(
|
|
"scope-expansion",
|
|
(
|
|
f"paths outside allowed_paths: {out_of_scope}"
|
|
if out_of_scope
|
|
else "worker requested scope expansion"
|
|
),
|
|
cleanup_extra={"preserved": True},
|
|
)
|
|
|
|
if proc_meta.get("idle_timed_out") and not proc_meta["timed_out"]:
|
|
# Idle expiry is not a proven-clean failure and must not continue routing.
|
|
return _preserved(
|
|
"failed-unknown",
|
|
"idle timeout; process group terminated",
|
|
cleanup_extra={
|
|
"reason": "idle timeout",
|
|
"idle_timed_out": True,
|
|
"timed_out": False,
|
|
},
|
|
)
|
|
|
|
if proc_meta["timed_out"] or proc_meta["signal"] is not None:
|
|
# Timeout/disconnect cannot prove absence of external side effects.
|
|
return _preserved(
|
|
"failed-unknown",
|
|
(
|
|
"timed out; process group terminated"
|
|
if proc_meta["timed_out"]
|
|
else f"abnormal termination signal={proc_meta['signal']}"
|
|
),
|
|
cleanup_extra={"reason": "unproven external side effects"},
|
|
)
|
|
|
|
clean_exit = proc_meta["exit_code"] == 0
|
|
# Never report success from worker prose alone — require exit + admin + in-scope diff + transport.
|
|
if clean_exit and changed and not out_of_scope:
|
|
# Recheck path bounds immediately before host-owned transport.
|
|
try:
|
|
_validate_allowed_path_symlinks(worktree, tuple(changed))
|
|
except UnitPacketError as exc:
|
|
return _preserved("failed-unknown", str(exc))
|
|
try:
|
|
transport_ref = _create_transport_commit(
|
|
repo=repo,
|
|
worktree=worktree,
|
|
base_head=base["head"],
|
|
run_id=run_id,
|
|
attempt_id=attempt_id,
|
|
inventory=inventory,
|
|
)
|
|
except Exception as exc:
|
|
# Staging / commit-tree / update-ref failure after a clean worker exit
|
|
# must stay recoverable: retained failed-unknown, no routing, no fallback.
|
|
return _preserved(
|
|
"failed-unknown",
|
|
f"transport construction failed: {exc}",
|
|
cleanup_extra={
|
|
"exact_restore": False,
|
|
"transport_created": False,
|
|
"reason": "transport construction failure",
|
|
"error": str(exc),
|
|
},
|
|
)
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="success",
|
|
reason=None,
|
|
worktree=str(worktree),
|
|
pid=proc_meta["pid"],
|
|
exit_code=proc_meta["exit_code"],
|
|
changed_paths=changed,
|
|
inventory_paths=list(changed),
|
|
raw_log=str(raw_log),
|
|
summary=summary,
|
|
transport_ref=transport_ref,
|
|
workspace_preserved=True,
|
|
cleanup={
|
|
"exact_restore": False,
|
|
"transport_created": True,
|
|
"stdout_tail_bytes": len(proc_meta.get("stdout_tail") or ""),
|
|
"stderr_tail_bytes": len(proc_meta.get("stderr_tail") or ""),
|
|
},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
if clean_exit and not changed:
|
|
restored = _restore_worktree(repo, worktree, base["head"])
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="failed-clean",
|
|
reason="worker exited 0 without in-scope changes",
|
|
worktree=str(worktree),
|
|
pid=proc_meta["pid"],
|
|
exit_code=proc_meta["exit_code"],
|
|
changed_paths=[],
|
|
inventory_paths=[],
|
|
raw_log=str(raw_log),
|
|
summary=summary,
|
|
cleanup={"exact_restore": restored},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
# Nonzero / aborted without trustworthy external reconciliation.
|
|
if _external_side_effects_unproven(unit, worker_result):
|
|
return _preserved(
|
|
"failed-unknown",
|
|
(
|
|
f"worker failed with exit {proc_meta['exit_code']}; "
|
|
"external side effects cannot be reconciled"
|
|
),
|
|
cleanup_extra={
|
|
"reason": "network or external_cost authorized without trustworthy receipt",
|
|
"network": unit.network,
|
|
"external_cost": unit.external_cost,
|
|
},
|
|
)
|
|
|
|
# Controlled case: network=false and external_cost=false may become failed-clean
|
|
# only after exact worktree restoration proves no unknown authorized external object.
|
|
if changed:
|
|
restored = _restore_worktree(repo, worktree, base["head"])
|
|
if restored:
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="failed-clean",
|
|
reason=f"worker failed with exit {proc_meta['exit_code']}; restored",
|
|
worktree=str(worktree),
|
|
pid=proc_meta["pid"],
|
|
exit_code=proc_meta["exit_code"],
|
|
changed_paths=changed,
|
|
inventory_paths=list(changed),
|
|
raw_log=str(raw_log),
|
|
summary=summary,
|
|
cleanup={
|
|
"exact_restore": True,
|
|
"external_objects_reconciled": True,
|
|
"network": False,
|
|
"external_cost": False,
|
|
},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
return _preserved(
|
|
"failed-unknown",
|
|
"worker failed; exact restore not proven",
|
|
)
|
|
|
|
restored = _restore_worktree(repo, worktree, base["head"])
|
|
return AttemptRecord(
|
|
attempt_id=attempt_id,
|
|
profile=profile.name,
|
|
adapter=profile.adapter,
|
|
terminal_status="failed-clean",
|
|
reason=f"worker failed with exit {proc_meta['exit_code']}",
|
|
worktree=str(worktree),
|
|
pid=proc_meta["pid"],
|
|
exit_code=proc_meta["exit_code"],
|
|
changed_paths=[],
|
|
inventory_paths=[],
|
|
raw_log=str(raw_log),
|
|
summary=summary,
|
|
cleanup={
|
|
"exact_restore": restored,
|
|
"external_objects_reconciled": True,
|
|
"network": False,
|
|
"external_cost": False,
|
|
},
|
|
observed_version=probe.observed_version,
|
|
)
|
|
|
|
|
|
def _supervise(
|
|
argv: tuple[str, ...],
|
|
*,
|
|
cwd: Path,
|
|
stdin: bytes | None,
|
|
timeout: int,
|
|
raw_log: Path,
|
|
raw_cap: int,
|
|
secret_values: set[str],
|
|
idle_timeout: int | None = None,
|
|
) -> dict[str, Any]:
|
|
env = _filtered_env()
|
|
# Capture for redaction only — never serialize into receipts.
|
|
for key, value in list(env.items()):
|
|
if SECRET_VALUE_RE.match(value.strip()) or "SECRET" in key.upper() or "TOKEN" in key.upper() or "KEY" in key.upper():
|
|
if value:
|
|
secret_values.add(value)
|
|
|
|
timed_out = False
|
|
idle_timed_out = False
|
|
exit_code: int | None = None
|
|
sig: int | None = None
|
|
pid: int | None = None
|
|
stdout_tail = _ByteTail(OUTPUT_TAIL_BYTES)
|
|
stderr_tail = _ByteTail(OUTPUT_TAIL_BYTES)
|
|
# Bounded prefix: raw_cap + exact-secret lookahead. Redact the complete
|
|
# prefix once, write at most raw_cap, then discard further drained bytes.
|
|
secret_lookahead = max((len(s) for s in secret_values), default=0)
|
|
secret_lookahead = min(max(secret_lookahead, 64), 8192)
|
|
prefix_limit = max(raw_cap, 0) + secret_lookahead
|
|
raw_prefix = bytearray()
|
|
raw_written = 0
|
|
prefix_sealed = False
|
|
|
|
def _preexec() -> None:
|
|
os.setsid()
|
|
|
|
try:
|
|
proc = subprocess.Popen(
|
|
list(argv),
|
|
cwd=str(cwd),
|
|
stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
env=env,
|
|
preexec_fn=_preexec,
|
|
bufsize=0,
|
|
)
|
|
except OSError as exc:
|
|
raw_log.write_bytes(_cap_bytes(str(exc).encode("utf-8", "replace"), raw_cap))
|
|
return {
|
|
"pid": None,
|
|
"exit_code": 127,
|
|
"timed_out": False,
|
|
"idle_timed_out": False,
|
|
"signal": None,
|
|
"stdout_tail": "",
|
|
"stderr_tail": "",
|
|
"summary_source": str(exc),
|
|
}
|
|
|
|
assert proc.stdout is not None and proc.stderr is not None
|
|
pid = proc.pid
|
|
|
|
def _set_nonblocking(fh: Any) -> None:
|
|
fd = fh.fileno()
|
|
flags = fcntl.fcntl(fd, fcntl.F_GETFL)
|
|
fcntl.fcntl(fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
|
|
|
|
_set_nonblocking(proc.stdout)
|
|
_set_nonblocking(proc.stderr)
|
|
stdin_fh = proc.stdin
|
|
stdin_view: memoryview | None = None
|
|
stdin_offset = 0
|
|
if stdin is not None and stdin_fh is not None:
|
|
_set_nonblocking(stdin_fh)
|
|
stdin_view = memoryview(stdin)
|
|
elif stdin_fh is not None:
|
|
try:
|
|
stdin_fh.close()
|
|
except OSError:
|
|
pass
|
|
stdin_fh = None
|
|
|
|
streams = {proc.stdout: stdout_tail, proc.stderr: stderr_tail}
|
|
bounded_timeout = max(int(timeout), 1)
|
|
if idle_timeout is None:
|
|
idle_limit = min(DEFAULT_IDLE_TIMEOUT_SECONDS, bounded_timeout)
|
|
else:
|
|
idle_limit = min(max(int(idle_timeout), 1), bounded_timeout)
|
|
started = time.monotonic()
|
|
deadline = started + bounded_timeout
|
|
idle_deadline = started + idle_limit
|
|
grace_deadline: float | None = None
|
|
|
|
with raw_log.open("wb") as raw_fh:
|
|
|
|
def _note_progress() -> None:
|
|
"""Reset idle only after a successful stdin write or stdout/stderr read."""
|
|
nonlocal idle_deadline
|
|
if grace_deadline is not None:
|
|
return
|
|
idle_deadline = time.monotonic() + idle_limit
|
|
|
|
def _seal_raw_prefix() -> None:
|
|
nonlocal raw_written, prefix_sealed, raw_prefix
|
|
if prefix_sealed:
|
|
return
|
|
prefix_sealed = True
|
|
if not raw_prefix:
|
|
return
|
|
redacted = _redact_bytes(bytes(raw_prefix), secret_values)
|
|
raw_prefix.clear()
|
|
if raw_cap > 0:
|
|
take = redacted[:raw_cap]
|
|
if take:
|
|
raw_fh.write(take)
|
|
raw_written = len(take)
|
|
raw_fh.flush()
|
|
|
|
def _emit_raw(chunk: bytes) -> None:
|
|
nonlocal raw_prefix
|
|
if not chunk:
|
|
return
|
|
if prefix_sealed:
|
|
# Cap reached earlier; keep draining but discard excess.
|
|
return
|
|
remaining = prefix_limit - len(raw_prefix)
|
|
if remaining <= 0:
|
|
_seal_raw_prefix()
|
|
return
|
|
if len(chunk) <= remaining:
|
|
raw_prefix.extend(chunk)
|
|
if len(raw_prefix) >= prefix_limit:
|
|
_seal_raw_prefix()
|
|
return
|
|
raw_prefix.extend(chunk[:remaining])
|
|
_seal_raw_prefix()
|
|
|
|
def _drain_fd(fh: Any, tail: _ByteTail) -> bool:
|
|
"""Read one chunk. True = keep watching; False = EOF/error (drop stream)."""
|
|
try:
|
|
chunk = os.read(fh.fileno(), 65536)
|
|
except BlockingIOError:
|
|
return True
|
|
except OSError:
|
|
return False
|
|
if not chunk:
|
|
return False
|
|
tail.extend(chunk)
|
|
_emit_raw(chunk)
|
|
_note_progress()
|
|
return True
|
|
|
|
def _drain_fd_available(fh: Any, tail: _ByteTail) -> None:
|
|
"""Drain until EOF, error, or would-block (final post-exit sweep)."""
|
|
while True:
|
|
try:
|
|
chunk = os.read(fh.fileno(), 65536)
|
|
except BlockingIOError:
|
|
return
|
|
except OSError:
|
|
return
|
|
if not chunk:
|
|
return
|
|
tail.extend(chunk)
|
|
_emit_raw(chunk)
|
|
|
|
def _write_stdin() -> bool:
|
|
"""Write a nonblocking stdin slice. Return False when stdin is done/closed."""
|
|
nonlocal stdin_fh, stdin_offset
|
|
if stdin_fh is None or stdin_view is None:
|
|
return False
|
|
if stdin_offset >= len(stdin_view):
|
|
try:
|
|
stdin_fh.close()
|
|
except OSError:
|
|
pass
|
|
stdin_fh = None
|
|
return False
|
|
try:
|
|
written = os.write(
|
|
stdin_fh.fileno(),
|
|
stdin_view[stdin_offset : stdin_offset + 65536],
|
|
)
|
|
except BrokenPipeError:
|
|
try:
|
|
stdin_fh.close()
|
|
except OSError:
|
|
pass
|
|
stdin_fh = None
|
|
return False
|
|
except BlockingIOError:
|
|
return True
|
|
except OSError:
|
|
try:
|
|
stdin_fh.close()
|
|
except OSError:
|
|
pass
|
|
stdin_fh = None
|
|
return False
|
|
stdin_offset += written
|
|
if written > 0:
|
|
_note_progress()
|
|
if stdin_offset >= len(stdin_view):
|
|
try:
|
|
stdin_fh.close()
|
|
except OSError:
|
|
pass
|
|
stdin_fh = None
|
|
return False
|
|
return True
|
|
|
|
while streams or proc.poll() is None or stdin_fh is not None:
|
|
now = time.monotonic()
|
|
if grace_deadline is None and (now >= deadline or now >= idle_deadline):
|
|
# Total timeout stays the hard bound when both deadlines are due.
|
|
if now >= deadline:
|
|
timed_out = True
|
|
else:
|
|
idle_timed_out = True
|
|
_kill_process_group(proc.pid)
|
|
grace_deadline = now + 5
|
|
if stdin_fh is not None:
|
|
try:
|
|
stdin_fh.close()
|
|
except OSError:
|
|
pass
|
|
stdin_fh = None
|
|
if grace_deadline is not None and now >= grace_deadline:
|
|
break
|
|
wait = 0.2
|
|
if grace_deadline is None:
|
|
wait = max(
|
|
0.0, min(wait, deadline - now, idle_deadline - now)
|
|
)
|
|
else:
|
|
wait = max(0.0, min(wait, grace_deadline - now))
|
|
readers = list(streams.keys())
|
|
writers: list[Any] = []
|
|
if stdin_fh is not None and stdin_view is not None and stdin_offset < len(stdin_view):
|
|
writers.append(stdin_fh)
|
|
if not readers and not writers:
|
|
# Process still alive but pipes already closed; wait for exit/timeout.
|
|
time.sleep(min(wait, 0.05))
|
|
continue
|
|
readable, writable, _ = select.select(readers, writers, [], wait)
|
|
for fh in writable:
|
|
if fh is stdin_fh:
|
|
_write_stdin()
|
|
for fh in readable:
|
|
alive = _drain_fd(fh, streams[fh])
|
|
if not alive:
|
|
streams.pop(fh, None)
|
|
try:
|
|
fh.close()
|
|
except OSError:
|
|
pass
|
|
if proc.poll() is not None and not readable and stdin_fh is None:
|
|
# Final drain after exit when select reports idle pipes.
|
|
for fh in list(streams.keys()):
|
|
_drain_fd_available(fh, streams[fh])
|
|
streams.pop(fh, None)
|
|
try:
|
|
fh.close()
|
|
except OSError:
|
|
pass
|
|
break
|
|
|
|
_seal_raw_prefix()
|
|
|
|
if proc.poll() is None:
|
|
if not idle_timed_out:
|
|
timed_out = True
|
|
_kill_process_group(proc.pid)
|
|
try:
|
|
proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
pass
|
|
exit_code = proc.returncode
|
|
|
|
if exit_code is not None and exit_code < 0:
|
|
sig = -exit_code
|
|
exit_code = 128 + sig
|
|
|
|
stdout_text = stdout_tail.text()
|
|
stderr_text = stderr_tail.text()
|
|
summary_source = _redact_text(stdout_text[-2000:], secret_values)
|
|
return {
|
|
"pid": pid,
|
|
"exit_code": exit_code,
|
|
"timed_out": timed_out,
|
|
"idle_timed_out": idle_timed_out,
|
|
"signal": sig,
|
|
"stdout_tail": stdout_text,
|
|
"stderr_tail": stderr_text,
|
|
"summary_source": summary_source,
|
|
}
|
|
|
|
|
|
class _ByteTail:
|
|
"""Bounded in-memory byte tail used while draining child output."""
|
|
|
|
def __init__(self, limit: int) -> None:
|
|
self.limit = max(1, limit)
|
|
self._data = bytearray()
|
|
|
|
def extend(self, chunk: bytes) -> None:
|
|
if not chunk:
|
|
return
|
|
self._data.extend(chunk)
|
|
if len(self._data) > self.limit:
|
|
self._data = self._data[-self.limit :]
|
|
|
|
def text(self) -> str:
|
|
return bytes(self._data).decode("utf-8", "replace")
|
|
|
|
|
|
def _kill_process_group(pid: int) -> None:
|
|
try:
|
|
os.killpg(pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
except PermissionError:
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
|
|
def _filtered_env() -> dict[str, str]:
|
|
out: dict[str, str] = {}
|
|
for key, value in os.environ.items():
|
|
if key in ENV_ALLOW_EXACT or key.startswith(ENV_ALLOW_PREFIXES):
|
|
out[key] = value
|
|
# Ensure PATH/HOME exist for executables.
|
|
if "PATH" not in out and "PATH" in os.environ:
|
|
out["PATH"] = os.environ["PATH"]
|
|
if "HOME" not in out and "HOME" in os.environ:
|
|
out["HOME"] = os.environ["HOME"]
|
|
return out
|
|
|
|
|
|
def _inventory_changes(worktree: Path) -> dict[str, Any]:
|
|
"""Inventory tracked, staged, untracked, deleted, renamed, mode, and binary changes.
|
|
|
|
Staged index entries are collected explicitly so they cannot be silently dropped.
|
|
"""
|
|
paths: list[str] = []
|
|
details: list[dict[str, Any]] = []
|
|
|
|
def _consume_name_status(text: str, *, prefix: str) -> None:
|
|
for line in text.splitlines():
|
|
if not line.strip():
|
|
continue
|
|
parts = line.split("\t")
|
|
status = parts[0]
|
|
if status.startswith("R") and len(parts) >= 3:
|
|
paths.append(parts[2])
|
|
details.append(
|
|
{
|
|
"status": f"{prefix}renamed",
|
|
"from": parts[1],
|
|
"path": parts[2],
|
|
}
|
|
)
|
|
elif len(parts) >= 2:
|
|
paths.append(parts[1])
|
|
details.append({"status": f"{prefix}{status[:1]}", "path": parts[1]})
|
|
|
|
# Explicit staged inventory (index vs HEAD) — fail-visible, never silent.
|
|
staged = _git(
|
|
worktree,
|
|
"diff",
|
|
"--cached",
|
|
"--name-status",
|
|
"--find-renames",
|
|
check=False,
|
|
)
|
|
if staged.returncode != 0:
|
|
raise RuntimeError(f"staged inventory failed: {staged.stderr.strip()}")
|
|
_consume_name_status(staged.stdout, prefix="staged:")
|
|
|
|
# Working tree vs HEAD (covers unstaged tracked modifications / deletions).
|
|
named = _git(
|
|
worktree,
|
|
"diff",
|
|
"--name-status",
|
|
"--find-renames",
|
|
"HEAD",
|
|
check=False,
|
|
)
|
|
if named.returncode != 0:
|
|
raise RuntimeError(f"worktree inventory failed: {named.stderr.strip()}")
|
|
_consume_name_status(named.stdout, prefix="")
|
|
|
|
# Untracked (including binary); ignore ignored files.
|
|
untracked = _git(
|
|
worktree,
|
|
"ls-files",
|
|
"--others",
|
|
"--exclude-standard",
|
|
check=False,
|
|
)
|
|
if untracked.returncode != 0:
|
|
raise RuntimeError(f"untracked inventory failed: {untracked.stderr.strip()}")
|
|
for line in untracked.stdout.splitlines():
|
|
path = line.strip()
|
|
if not path:
|
|
continue
|
|
if path.startswith(".git/") or "/.git/" in path:
|
|
continue
|
|
paths.append(path)
|
|
details.append({"status": "untracked", "path": path})
|
|
|
|
# Executable mode bits among tracked diffs.
|
|
mode = _git(worktree, "diff", "--summary", "HEAD", check=False)
|
|
if mode.returncode != 0:
|
|
raise RuntimeError(f"mode inventory failed: {mode.stderr.strip()}")
|
|
for line in mode.stdout.splitlines():
|
|
if "mode change" in line or "change from" in line:
|
|
details.append({"status": "mode", "raw": line.strip()})
|
|
|
|
# Deduplicate while preserving order.
|
|
seen: set[str] = set()
|
|
ordered: list[str] = []
|
|
for p in paths:
|
|
if p not in seen:
|
|
seen.add(p)
|
|
ordered.append(p)
|
|
return {"paths": ordered, "details": details}
|
|
|
|
def _path_allowed(path: str, allowed: tuple[str, ...]) -> bool:
|
|
norm = path.replace("\\", "/").lstrip("./")
|
|
if ".." in Path(norm).parts:
|
|
return False
|
|
for rule in allowed:
|
|
rule_n = rule.replace("\\", "/").lstrip("./")
|
|
if norm == rule_n or norm.startswith(rule_n.rstrip("/") + "/"):
|
|
return True
|
|
# Exact file rule: also allow if rule is a prefix directory.
|
|
if rule_n.endswith("/") and norm.startswith(rule_n):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _git_admin_snapshot(worktree: Path, repo: Path) -> dict[str, str]:
|
|
"""Snapshot the private attempt clone's Git administration state.
|
|
|
|
Canonical checkout identity is snapshotted separately. ``repo`` is unused now
|
|
that each attempt owns private config/refs; kept for call-site stability.
|
|
"""
|
|
del repo
|
|
head = _git(worktree, "rev-parse", "HEAD").stdout.strip()
|
|
index_tree = _git(worktree, "write-tree").stdout.strip()
|
|
staged_proc = _git(
|
|
worktree,
|
|
"diff",
|
|
"--cached",
|
|
"--name-status",
|
|
"--find-renames",
|
|
check=False,
|
|
)
|
|
if staged_proc.returncode != 0:
|
|
raise RuntimeError(f"admin staged snapshot failed: {staged_proc.stderr.strip()}")
|
|
staged = staged_proc.stdout
|
|
local_config_proc = _git(worktree, "config", "--local", "--list", check=False)
|
|
if local_config_proc.returncode != 0:
|
|
raise RuntimeError(
|
|
f"admin local-config snapshot failed: {local_config_proc.stderr.strip()}"
|
|
)
|
|
local_config = "\n".join(
|
|
sorted(line for line in local_config_proc.stdout.splitlines() if line)
|
|
)
|
|
refs_proc = _git(
|
|
worktree,
|
|
"for-each-ref",
|
|
"--format=%(refname) %(objectname)",
|
|
check=False,
|
|
)
|
|
if refs_proc.returncode != 0:
|
|
raise RuntimeError(f"admin refs snapshot failed: {refs_proc.stderr.strip()}")
|
|
refs = "\n".join(sorted(line for line in refs_proc.stdout.splitlines() if line))
|
|
abs_git = _git(worktree, "rev-parse", "--absolute-git-dir").stdout.strip()
|
|
common_dir = _git(
|
|
worktree, "rev-parse", "--path-format=absolute", "--git-common-dir"
|
|
).stdout.strip()
|
|
return {
|
|
"head": head,
|
|
"index_tree": index_tree,
|
|
"staged": staged,
|
|
"local_config": local_config,
|
|
"refs": refs,
|
|
"abs_git": abs_git,
|
|
"common_dir": common_dir,
|
|
}
|
|
|
|
|
|
def _git_admin_delta(before: dict[str, str], after: dict[str, str]) -> list[str]:
|
|
return _snapshot_delta(before, after)
|
|
|
|
|
|
def _snapshot_delta(before: dict[str, str], after: dict[str, str]) -> list[str]:
|
|
changed: list[str] = []
|
|
for key in sorted(set(before) | set(after)):
|
|
if before.get(key) != after.get(key):
|
|
changed.append(key)
|
|
return changed
|
|
|
|
|
|
def _canonical_checkout_snapshot(repo: Path) -> dict[str, str]:
|
|
"""Snapshot the canonical checkout identity (not the attempt workspace).
|
|
|
|
Protects HEAD, current branch, index, tracked/untracked status, and
|
|
git/common-dir identity. Unrelated sibling branch refs, branch config, and
|
|
worktree-list registration are intentionally excluded — worker Git is
|
|
isolated in a private clone, so those host-side delivery signals are not
|
|
worker mutation.
|
|
"""
|
|
repo = repo.resolve()
|
|
head = _git(repo, "rev-parse", "HEAD").stdout.strip()
|
|
branch = _git(repo, "rev-parse", "--abbrev-ref", "HEAD").stdout.strip()
|
|
index_tree = _git(repo, "write-tree").stdout.strip()
|
|
status = _git(repo, "status", "--porcelain=v2", "-uall").stdout
|
|
git_dir = _git(repo, "rev-parse", "--absolute-git-dir").stdout.strip()
|
|
common = _git(
|
|
repo, "rev-parse", "--path-format=absolute", "--git-common-dir"
|
|
).stdout.strip()
|
|
return {
|
|
"head": head,
|
|
"branch": branch,
|
|
"index_tree": index_tree,
|
|
"status": status,
|
|
"git_dir": git_dir,
|
|
"common_dir": common,
|
|
}
|
|
|
|
|
|
def _worktree_pointer_snapshot(
|
|
worktree: Path, *, require_git: bool = True
|
|
) -> dict[str, str]:
|
|
"""Snapshot the private attempt `.git` directory / admin identity."""
|
|
git_path = worktree / ".git"
|
|
if git_path.is_symlink():
|
|
kind = "symlink"
|
|
try:
|
|
content = os.readlink(git_path)
|
|
except OSError as exc:
|
|
raise RuntimeError(f"unable to read worktree .git symlink: {exc}") from exc
|
|
elif git_path.is_file():
|
|
kind = "file"
|
|
try:
|
|
content = git_path.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise RuntimeError(f"unable to read worktree .git file: {exc}") from exc
|
|
elif git_path.is_dir():
|
|
kind = "dir"
|
|
content = "dir"
|
|
else:
|
|
kind = "missing"
|
|
content = ""
|
|
snap = {"kind": kind, "content": content, "abs_git": "", "common_dir": ""}
|
|
if require_git:
|
|
snap["abs_git"] = _git(
|
|
worktree, "rev-parse", "--absolute-git-dir"
|
|
).stdout.strip()
|
|
snap["common_dir"] = _git(
|
|
worktree, "rev-parse", "--path-format=absolute", "--git-common-dir"
|
|
).stdout.strip()
|
|
return snap
|
|
|
|
|
|
def _external_side_effects_unproven(
|
|
unit: UnitPacket, worker_result: dict[str, Any] | None
|
|
) -> bool:
|
|
"""Nonzero/aborted workers cannot be failed-clean when external effects are open.
|
|
|
|
failed-clean requires exact restore plus affirmative evidence that no authorized
|
|
external object remains unknown. Units with network/external_cost, or without a
|
|
trustworthy structured terminal receipt proving reconciliation, stay unknown.
|
|
"""
|
|
del worker_result # Worker prose alone never proves external reconciliation.
|
|
return bool(unit.network or unit.external_cost)
|
|
|
|
|
|
def _is_within_root(path: Path, root: Path) -> bool:
|
|
try:
|
|
path.resolve().relative_to(root.resolve())
|
|
return True
|
|
except (ValueError, OSError):
|
|
return False
|
|
|
|
|
|
def _path_symlink_escapes(root: Path, rel: str) -> bool:
|
|
"""True when any existing ancestor/target of rel resolves outside root."""
|
|
root = root.resolve()
|
|
cur = root
|
|
for part in Path(rel).parts:
|
|
if part in ("", "."):
|
|
continue
|
|
if part == "..":
|
|
return True
|
|
nxt = cur / part
|
|
if nxt.is_symlink() or nxt.exists():
|
|
try:
|
|
resolved = nxt.resolve()
|
|
except OSError:
|
|
return True
|
|
if not _is_within_root(resolved, root):
|
|
return True
|
|
cur = resolved if nxt.is_symlink() else nxt
|
|
else:
|
|
# Remaining path does not exist yet; parent must already be inside.
|
|
return not _is_within_root(cur, root)
|
|
return False
|
|
|
|
|
|
def _validate_allowed_path_symlinks(root: Path, paths: tuple[str, ...]) -> None:
|
|
for rel in paths:
|
|
if _path_symlink_escapes(root, rel):
|
|
raise UnitPacketError(
|
|
f"allowed path escapes worktree via symlink: {rel!r}"
|
|
)
|
|
|
|
|
|
def _create_transport_commit(
|
|
*,
|
|
repo: Path,
|
|
worktree: Path,
|
|
base_head: str,
|
|
run_id: str,
|
|
attempt_id: str,
|
|
inventory: dict[str, Any],
|
|
) -> str:
|
|
"""Create a base-parented transport commit in the private clone and copy it back.
|
|
|
|
Staging and ``commit-tree`` run only inside the attempt clone. Exact
|
|
commit/tree/blob objects are fetched into the canonical repository, which
|
|
receives solely the host-owned ``refs/agent-exec/<run>/<attempt>`` ref.
|
|
"""
|
|
del inventory # Inventory already validated; staging uses the full private tree.
|
|
# Stage all tracked/untracked changes in the private clone only.
|
|
_git(worktree, "add", "-A", "--", ".")
|
|
# Include deletions.
|
|
_git(worktree, "add", "-u", check=False)
|
|
tree = _git(worktree, "write-tree").stdout.strip()
|
|
msg = f"agent-exec transport {run_id}/{attempt_id}"
|
|
commit = _git(
|
|
worktree,
|
|
"-c",
|
|
f"user.name={TRANSPORT_AUTHOR_NAME}",
|
|
"-c",
|
|
f"user.email={TRANSPORT_AUTHOR_EMAIL}",
|
|
"commit-tree",
|
|
tree,
|
|
"-p",
|
|
base_head,
|
|
"-m",
|
|
msg,
|
|
).stdout.strip()
|
|
ref = f"refs/agent-exec/{run_id}/{attempt_id}"
|
|
# Park the commit on a private ref so a local fetch can name it.
|
|
_git(worktree, "update-ref", ref, commit)
|
|
# Copy objects into the canonical store and create only the host-owned ref.
|
|
fetch = _git(
|
|
repo,
|
|
"fetch",
|
|
"--no-tags",
|
|
"--no-write-fetch-head",
|
|
str(worktree.resolve()),
|
|
f"+{ref}:{ref}",
|
|
check=False,
|
|
)
|
|
if fetch.returncode != 0:
|
|
raise RuntimeError(
|
|
f"transport object transfer failed: {fetch.stderr.strip() or fetch.stdout.strip()}"
|
|
)
|
|
resolved = _git(repo, "rev-parse", ref).stdout.strip()
|
|
if resolved != commit:
|
|
raise RuntimeError(
|
|
f"transport ref resolved to {resolved}, expected {commit}"
|
|
)
|
|
parents = (
|
|
_git(repo, "rev-list", "--parents", "-n", "1", commit)
|
|
.stdout.strip()
|
|
.split()
|
|
)
|
|
if len(parents) != 2 or parents[0] != commit or parents[1] != base_head:
|
|
raise RuntimeError(
|
|
f"transport commit parent check failed: {' '.join(parents)}"
|
|
)
|
|
return ref
|
|
|
|
|
|
def _restore_worktree(repo: Path, worktree: Path, base_head: str) -> bool:
|
|
"""Exact-restore the private clone then remove it after verification."""
|
|
try:
|
|
_git(worktree, "reset", "--hard", base_head)
|
|
_git(worktree, "clean", "-fdx")
|
|
head = _git(worktree, "rev-parse", "HEAD").stdout.strip()
|
|
dirty = _git(worktree, "status", "--porcelain").stdout.strip()
|
|
ok = head == base_head and not dirty
|
|
_safe_remove_worktree(repo, worktree)
|
|
return ok
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def _safe_remove_worktree(repo: Path, worktree: Path) -> None:
|
|
"""Remove a private attempt clone directory (not a linked worktree)."""
|
|
del repo # Private clones are not registered linked worktrees.
|
|
if worktree.exists():
|
|
shutil.rmtree(worktree, ignore_errors=True)
|
|
|
|
|
|
def _load_worker_result(
|
|
result_path: Path, stdout_tail: str
|
|
) -> dict[str, Any] | None:
|
|
if result_path.is_file():
|
|
try:
|
|
data = json.loads(result_path.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
return data
|
|
except (OSError, json.JSONDecodeError):
|
|
pass
|
|
# Try last JSON object in stdout.
|
|
text = stdout_tail.strip()
|
|
if not text:
|
|
return None
|
|
for candidate in (text, text[text.rfind("{") :] if "{" in text else ""):
|
|
if not candidate.startswith("{"):
|
|
continue
|
|
try:
|
|
data = json.loads(candidate)
|
|
if isinstance(data, dict):
|
|
return data
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _base_receipt(
|
|
*,
|
|
run_id: str,
|
|
repo: Path,
|
|
role: str,
|
|
unit_id: str | None,
|
|
config: ExecutionConfig | None,
|
|
terminal_status: str,
|
|
reason: str | None,
|
|
attempts: list[AttemptRecord],
|
|
secret_values: set[str],
|
|
) -> dict[str, Any]:
|
|
attempt_dicts = []
|
|
for a in attempts:
|
|
attempt_dicts.append(
|
|
{
|
|
"attempt_id": a.attempt_id,
|
|
"profile": a.profile,
|
|
"adapter": a.adapter,
|
|
"attempt_ordinal": a.attempt_ordinal,
|
|
"emergency_fallback": a.emergency_fallback,
|
|
"terminal_status": a.terminal_status,
|
|
"reason": a.reason,
|
|
"worktree": a.worktree,
|
|
"pid": a.pid,
|
|
"exit_code": a.exit_code,
|
|
"timed_out": a.timed_out,
|
|
"idle_timed_out": a.idle_timed_out,
|
|
"signal": a.signal,
|
|
"changed_paths": a.changed_paths,
|
|
"inventory_paths": a.inventory_paths or a.changed_paths,
|
|
"raw_log": a.raw_log,
|
|
"summary": a.summary,
|
|
"cleanup": a.cleanup,
|
|
"workspace_preserved": a.workspace_preserved,
|
|
"transport_ref": a.transport_ref,
|
|
"observed_version": a.observed_version,
|
|
}
|
|
)
|
|
return {
|
|
"schema": RECEIPT_SCHEMA,
|
|
"run_id": run_id,
|
|
"repo": str(repo),
|
|
"role": role,
|
|
"unit_id": unit_id,
|
|
"terminal_status": terminal_status,
|
|
"reason": reason,
|
|
"attempts": attempt_dicts,
|
|
"config_summary": _config_summary(config),
|
|
"created_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
}
|
|
|
|
|
|
def _config_summary(config: ExecutionConfig | None) -> dict[str, Any] | None:
|
|
if config is None:
|
|
return None
|
|
return {
|
|
"schema": config.schema,
|
|
"permission": config.execution.permission,
|
|
"fallback_to_host": config.execution.fallback_to_host,
|
|
"timeout_seconds": config.execution.timeout_seconds,
|
|
"idle_timeout_seconds": config.execution.idle_timeout_seconds,
|
|
"summary_bytes": config.execution.summary_bytes,
|
|
"raw_output_bytes": config.execution.raw_output_bytes,
|
|
"roles": {k: list(v) for k, v in config.roles.items()},
|
|
"profiles": {
|
|
name: {
|
|
"adapter": p.adapter,
|
|
"executable": p.executable,
|
|
"model": p.model,
|
|
"enabled": p.enabled,
|
|
}
|
|
for name, p in config.profiles.items()
|
|
},
|
|
}
|
|
|
|
|
|
def _finish_run(
|
|
*,
|
|
repo: Path,
|
|
receipt_path: Path,
|
|
receipt: dict[str, Any],
|
|
secret_values: set[str],
|
|
summary_bytes: int,
|
|
exit_code: int,
|
|
started_monotonic: float,
|
|
) -> int:
|
|
"""Persist one history line and the normal receipt without changing outcome."""
|
|
receipt["elapsed_seconds"] = round(
|
|
max(0.0, time.monotonic() - started_monotonic), 3
|
|
)
|
|
history_path: Path | None = None
|
|
try:
|
|
history_path = _append_execution_history(repo, receipt, secret_values)
|
|
receipt["history"] = {
|
|
"path": str(history_path),
|
|
"status": "recorded",
|
|
}
|
|
except Exception as exc:
|
|
receipt["history"] = {
|
|
"path": str(history_path) if history_path is not None else None,
|
|
"status": "failed",
|
|
"reason": _redact_text(str(exc), secret_values),
|
|
}
|
|
_write_receipt(
|
|
receipt_path,
|
|
receipt,
|
|
secret_values=secret_values,
|
|
summary_bytes=summary_bytes,
|
|
)
|
|
return exit_code
|
|
|
|
|
|
def _append_execution_history(
|
|
repo: Path,
|
|
receipt: dict[str, Any],
|
|
secret_values: set[str],
|
|
) -> Path:
|
|
common = _git(
|
|
repo.resolve(),
|
|
"rev-parse",
|
|
"--path-format=absolute",
|
|
"--git-common-dir",
|
|
check=False,
|
|
)
|
|
if common.returncode != 0 or not common.stdout.strip():
|
|
raise RuntimeError(
|
|
"unable to resolve git common dir: "
|
|
+ (common.stderr.strip() or common.stdout.strip())
|
|
)
|
|
common_dir = Path(common.stdout.strip())
|
|
if not common_dir.is_absolute():
|
|
common_dir = (repo.resolve() / common_dir).resolve()
|
|
history_dir = common_dir / "agent-exec"
|
|
if history_dir.is_symlink():
|
|
raise RuntimeError("agent-exec history directory must not be a symlink")
|
|
history_dir.mkdir(mode=0o700, parents=True, exist_ok=True)
|
|
os.chmod(history_dir, 0o700)
|
|
history_path = history_dir / "execution-history.jsonl"
|
|
record = _execution_history_record(repo, receipt)
|
|
line = json.dumps(record, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
|
|
line = _redact_text(line, secret_values) + "\n"
|
|
fd = os.open(history_path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "a", encoding="utf-8", closefd=False) as handle:
|
|
fcntl.flock(fd, fcntl.LOCK_EX)
|
|
try:
|
|
handle.write(line)
|
|
handle.flush()
|
|
os.fsync(fd)
|
|
finally:
|
|
fcntl.flock(fd, fcntl.LOCK_UN)
|
|
finally:
|
|
os.close(fd)
|
|
return history_path
|
|
|
|
|
|
def _execution_history_record(repo: Path, receipt: dict[str, Any]) -> dict[str, Any]:
|
|
config = receipt.get("config_summary")
|
|
profiles = config.get("profiles", {}) if isinstance(config, dict) else {}
|
|
attempt_records: list[dict[str, Any]] = []
|
|
for attempt in receipt.get("attempts", []):
|
|
if not isinstance(attempt, dict):
|
|
continue
|
|
profile_name = attempt.get("profile")
|
|
profile = profiles.get(profile_name, {}) if isinstance(profiles, dict) else {}
|
|
requested = profile.get("model") if isinstance(profile, dict) else None
|
|
attempt_records.append(
|
|
{
|
|
"attempt_id": attempt.get("attempt_id"),
|
|
"profile": profile_name,
|
|
"adapter": attempt.get("adapter"),
|
|
"attempt_ordinal": attempt.get("attempt_ordinal"),
|
|
"emergency_fallback": bool(attempt.get("emergency_fallback")),
|
|
"observed_version": attempt.get("observed_version"),
|
|
"model_requested": requested or "auto",
|
|
"model_observed": "not-observable",
|
|
"fresh_session": True,
|
|
"session_id": "not-observable",
|
|
"terminal_status": attempt.get("terminal_status"),
|
|
"exit_code": attempt.get("exit_code"),
|
|
"timed_out": bool(attempt.get("timed_out")),
|
|
"idle_timed_out": bool(attempt.get("idle_timed_out")),
|
|
"changed_paths": list(attempt.get("changed_paths") or []),
|
|
"transport_ref": attempt.get("transport_ref"),
|
|
}
|
|
)
|
|
transport = receipt.get("transport")
|
|
safe_transport = None
|
|
if isinstance(transport, dict):
|
|
safe_transport = {
|
|
"ref": transport.get("ref"),
|
|
"parent": transport.get("parent"),
|
|
}
|
|
terminal = receipt.get("terminal_status")
|
|
return {
|
|
"schema": HISTORY_SCHEMA,
|
|
"recorded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"record_id": receipt.get("run_id"),
|
|
"run_id": receipt.get("run_id"),
|
|
"repo": str(repo.resolve()),
|
|
"unit_id": receipt.get("unit_id"),
|
|
"role": receipt.get("role"),
|
|
"stage": receipt.get("role"),
|
|
"config_digest": receipt.get("config_digest"),
|
|
"elapsed_seconds": receipt.get("elapsed_seconds"),
|
|
"terminal_status": terminal,
|
|
"reason": receipt.get("reason"),
|
|
"failure_category": None if terminal == "success" else terminal,
|
|
"changed_paths": list(receipt.get("changed_paths") or []),
|
|
"transport": safe_transport,
|
|
"host_fallback_authorizations": receipt.get(
|
|
"host_fallback_authorizations", 0
|
|
),
|
|
"attempts": attempt_records,
|
|
"verification_result": "NOT_RUN",
|
|
"review_result": "NOT_RUN",
|
|
"final_result": "NOT_RUN",
|
|
"tokens": "not-observable",
|
|
"cost": "not-observable",
|
|
}
|
|
|
|
|
|
def _write_receipt(
|
|
path: Path,
|
|
receipt: dict[str, Any],
|
|
*,
|
|
secret_values: set[str],
|
|
summary_bytes: int,
|
|
) -> None:
|
|
if "summary" in receipt and isinstance(receipt["summary"], str):
|
|
receipt["summary"] = _bounded_summary(
|
|
receipt["summary"], summary_bytes, secret_values
|
|
)
|
|
text = json.dumps(receipt, indent=2, sort_keys=True) + "\n"
|
|
text = _redact_text(text, secret_values)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|
|
|
|
|
|
def _build_run_summary(
|
|
terminal: str, attempts: list[AttemptRecord], reason: str | None
|
|
) -> str:
|
|
parts = [f"terminal={terminal}"]
|
|
if reason:
|
|
parts.append(f"reason={reason}")
|
|
parts.append(f"attempts={len(attempts)}")
|
|
for a in attempts:
|
|
parts.append(f"{a.profile}:{a.terminal_status}")
|
|
return "; ".join(parts)
|
|
|
|
|
|
def _exit_for_terminal(terminal: str) -> int:
|
|
return {
|
|
"success": EXIT_SUCCESS,
|
|
"host-fallback-authorized": EXIT_HOST_FALLBACK,
|
|
"invalid-config": EXIT_INVALID_CONFIG,
|
|
"unavailable": EXIT_UNAVAILABLE_EXHAUSTION,
|
|
"failed-clean": EXIT_UNAVAILABLE_EXHAUSTION,
|
|
"failed-unknown": EXIT_BLOCKED,
|
|
"blocked": EXIT_BLOCKED,
|
|
"scope-expansion": EXIT_SCOPE_EXPANSION,
|
|
"cancelled": EXIT_CANCELLED,
|
|
}.get(terminal, EXIT_BLOCKED)
|
|
|
|
|
|
def _git(
|
|
repo: Path, *args: str, check: bool = True
|
|
) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["git", "-C", str(repo), *args],
|
|
check=check,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
|
|
def _file_digest(path: Path) -> str:
|
|
try:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _collect_secret_values(out: set[str]) -> None:
|
|
for key, value in os.environ.items():
|
|
if not value:
|
|
continue
|
|
if SECRET_VALUE_RE.match(value.strip()):
|
|
out.add(value)
|
|
if re.search(r"(SECRET|TOKEN|PASSWORD|API_KEY|CREDENTIAL)", key, re.I):
|
|
out.add(value)
|
|
|
|
|
|
def _redact_text(text: str, secrets: set[str]) -> str:
|
|
out = text
|
|
for secret in sorted(secrets, key=len, reverse=True):
|
|
if secret and secret in out:
|
|
out = out.replace(secret, "***")
|
|
out = SECRET_VALUE_RE.sub("***", out)
|
|
return out
|
|
|
|
|
|
def _redact_bytes(data: bytes, secrets: set[str]) -> bytes:
|
|
try:
|
|
text = data.decode("utf-8", "replace")
|
|
except Exception:
|
|
return data[: len(data)]
|
|
return _redact_text(text, secrets).encode("utf-8", "replace")
|
|
|
|
|
|
def _bounded_summary(text: str, limit: int, secrets: set[str]) -> str:
|
|
text = _redact_text(text or "", secrets)
|
|
encoded = text.encode("utf-8")
|
|
if len(encoded) <= limit:
|
|
return text
|
|
return encoded[:limit].decode("utf-8", "ignore")
|
|
|
|
|
|
def _cap_bytes(data: bytes, limit: int) -> bytes:
|
|
if len(data) <= limit:
|
|
return data
|
|
return data[:limit]
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|