"""Tracer: every event lands in JSONL and SQLite AS IT HAPPENS. Files are the raw record; sssf.db is the queryable mirror the UI polls. No push transport — the flow is always: agents -> sqlite -> web ui. WAL mode so the UI can read while ADW processes write. """ from __future__ import annotations import json import os import re import socket import sqlite3 import sys import time import uuid from pathlib import Path from typing import Any from .data_types import AgentConfig, EventRecord, GateReport, Phase from .git_helper import derive_repo_name from .run_slug import derive_slug_base, pick_unique_slug from .utils import ensure_dir, new_id, now_iso SCHEMA = """ CREATE TABLE IF NOT EXISTS sessions ( adw_id TEXT PRIMARY KEY, adw_name TEXT, -- ADW script(s) run, e.g. "adw_plan + adw_build_test" repo TEXT, -- canonical target repository root for this run request TEXT, status TEXT, engineer TEXT, host TEXT, started_at TEXT, ended_at TEXT, total_tokens INTEGER DEFAULT 0, total_cost REAL DEFAULT 0, archived INTEGER DEFAULT 0 -- review triage, set by the UI; never by a run ); CREATE TABLE IF NOT EXISTS request_run_links ( request_id TEXT NOT NULL, adw_id TEXT NOT NULL UNIQUE REFERENCES sessions(adw_id), linked_at TEXT NOT NULL, PRIMARY KEY (request_id, adw_id) ); CREATE INDEX IF NOT EXISTS idx_request_run_links_request ON request_run_links(request_id, linked_at); CREATE TABLE IF NOT EXISTS phases ( phase_id TEXT PRIMARY KEY, adw_id TEXT REFERENCES sessions, seq INTEGER, name TEXT, kind TEXT, owner TEXT, description TEXT, status TEXT DEFAULT 'fail', attempt INTEGER DEFAULT 0, retries INTEGER DEFAULT 0, error TEXT, started_at TEXT, ended_at TEXT ); CREATE TABLE IF NOT EXISTS events ( event_id TEXT PRIMARY KEY, adw_id TEXT REFERENCES sessions, phase_id TEXT REFERENCES phases, parent_id TEXT, type TEXT, name TEXT, payload_json TEXT, tokens INTEGER, started_at TEXT, ended_at TEXT ); CREATE TABLE IF NOT EXISTS envelopes ( envelope_id TEXT PRIMARY KEY, adw_id TEXT REFERENCES sessions, phase_id TEXT REFERENCES phases, agent TEXT, output_type TEXT, payload_json TEXT, valid INTEGER, attempt INTEGER, created_at TEXT ); CREATE TABLE IF NOT EXISTS gate_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, adw_id TEXT REFERENCES sessions, phase_id TEXT REFERENCES phases, attempt INTEGER, gate TEXT, passed INTEGER, violations_json TEXT, checks_json TEXT, -- [{item, ok, note}] — WHAT the gate verified created_at TEXT ); CREATE TABLE IF NOT EXISTS phase_diffs ( id INTEGER PRIMARY KEY AUTOINCREMENT, adw_id TEXT NOT NULL, phase_id TEXT NOT NULL, attempt INTEGER, files_json TEXT, insertions INTEGER, deletions INTEGER, diff_text TEXT, truncated INTEGER, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_phase_diffs_adw_phase ON phase_diffs(adw_id, phase_id); CREATE TABLE IF NOT EXISTS processes ( id INTEGER PRIMARY KEY AUTOINCREMENT, adw_id TEXT REFERENCES sessions, kind TEXT, -- 'adw' (the workflow process) | 'agent' (a coding-agent child) name TEXT, -- '' for the adw, the agent name for a child pid INTEGER, start_ticks INTEGER, command TEXT, -- what the pid was, so a recycled pid is not killed by mistake started_at TEXT, ended_at TEXT -- ended_at NULL = believed alive ); CREATE TABLE IF NOT EXISTS agent_sessions ( adw_id TEXT REFERENCES sessions, agent TEXT, coding_agent TEXT, model TEXT, color TEXT, session_id TEXT, context_tokens INTEGER, -- window occupancy after the agent's last turn context_window INTEGER, -- the model's ceiling; 0/NULL = unknown created_at TEXT, last_used_at TEXT, PRIMARY KEY (adw_id, agent) ); CREATE TABLE IF NOT EXISTS agent_attempts ( attempt_id TEXT PRIMARY KEY, adw_id TEXT REFERENCES sessions, phase_id TEXT REFERENCES phases, parent_id TEXT, agent TEXT, session_id TEXT, command TEXT, system_prompt TEXT, user_prompt TEXT, host TEXT, account TEXT, returncode INTEGER, signal INTEGER, timed_out INTEGER NOT NULL DEFAULT 0, stderr_path TEXT, tokens INTEGER, usage_json TEXT, error TEXT, started_at TEXT NOT NULL, ended_at TEXT ); CREATE TABLE IF NOT EXISTS tool_calls ( tool_call_id TEXT PRIMARY KEY, attempt_id TEXT NOT NULL, seq INTEGER NOT NULL, tool_name TEXT, args_json TEXT, started_at TEXT, ended_at TEXT, duration_ms INTEGER, ok INTEGER, result_excerpt TEXT ); CREATE INDEX IF NOT EXISTS idx_tool_calls_attempt_seq ON tool_calls(attempt_id, seq); CREATE TABLE IF NOT EXISTS decisions ( decision_id TEXT PRIMARY KEY, adw_id TEXT NOT NULL, phase TEXT, question TEXT NOT NULL, options TEXT NOT NULL DEFAULT '[]', free_text INTEGER NOT NULL DEFAULT 0, context TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending', answer_value TEXT, answer_text TEXT, answered_by TEXT, created_at TEXT NOT NULL, answered_at TEXT ); CREATE INDEX IF NOT EXISTS idx_decisions_status ON decisions(status); """ # Columns added after a schema shipped. CREATE TABLE IF NOT EXISTS never # revisits an existing table, so additive changes need an explicit ALTER. MIGRATIONS = [("agent_sessions", "color", "TEXT"), ("gate_results", "checks_json", "TEXT"), ("sessions", "adw_name", "TEXT"), ("agent_sessions", "context_tokens", "INTEGER"), ("agent_sessions", "context_window", "INTEGER"), ("sessions", "archived", "INTEGER DEFAULT 0"), ("sessions", "repo", "TEXT"), ("agent_attempts", "host", "TEXT"), ("agent_attempts", "account", "TEXT"), ("agent_attempts", "model", "TEXT"), ("agent_attempts", "system_prompt", "TEXT"), ("agent_attempts", "user_prompt", "TEXT"), ("sessions", "host", "TEXT"), ("processes", "start_ticks", "INTEGER"), ("agent_attempts", "timeout_kind", "TEXT"), ("sessions", "run_slug", "TEXT"), ("sessions", "repo_name", "TEXT"), ("sessions", "preset", "TEXT"), ("phases", "task_id", "TEXT"), ("phase_diffs", "task_id", "TEXT"), ("phase_diffs", "attempt_id", "TEXT")] RUN_SLUG_UNIQUE_INDEX = ( "CREATE UNIQUE INDEX IF NOT EXISTS idx_sessions_run_slug" " ON sessions(run_slug) WHERE run_slug IS NOT NULL" ) RUN_SLUG_INDEX_NAME = "idx_sessions_run_slug" def verify_run_slug_index(conn: sqlite3.Connection) -> None: """Fail closed unless idx_sessions_run_slug is unique and partial on run_slug.""" row = conn.execute( "SELECT name, [unique], partial FROM pragma_index_list('sessions')" " WHERE name=?", (RUN_SLUG_INDEX_NAME,), ).fetchone() if row is None: raise RuntimeError(f"{RUN_SLUG_INDEX_NAME} missing") _name, is_unique, is_partial = row if not is_unique: raise RuntimeError(f"{RUN_SLUG_INDEX_NAME} is not unique") if not is_partial: raise RuntimeError(f"{RUN_SLUG_INDEX_NAME} is not partial") columns = [ info[2] for info in conn.execute( f"PRAGMA index_info({RUN_SLUG_INDEX_NAME})", ).fetchall() ] if columns != ["run_slug"]: raise RuntimeError( f"{RUN_SLUG_INDEX_NAME} indexes {columns!r}, expected ['run_slug']", ) sql_row = conn.execute( "SELECT sql FROM sqlite_master WHERE type='index' AND name=?", (RUN_SLUG_INDEX_NAME,), ).fetchone() sql = (sql_row[0] if sql_row else "").lower() if "run_slug is not null" not in sql: raise RuntimeError( f"{RUN_SLUG_INDEX_NAME} missing run_slug IS NOT NULL predicate", ) def _ensure_run_slug_unique_index(conn: sqlite3.Connection) -> None: existing = conn.execute( "SELECT 1 FROM pragma_index_list('sessions') WHERE name=?", (RUN_SLUG_INDEX_NAME,), ).fetchone() if existing is not None: verify_run_slug_index(conn) return deadline = time.monotonic() + 5.0 while True: try: conn.execute(RUN_SLUG_UNIQUE_INDEX) break except sqlite3.OperationalError as exc: if "database is locked" not in str(exc).lower(): raise if time.monotonic() >= deadline: raise time.sleep(0.001) verify_run_slug_index(conn) TOOL_RESULT_EXCERPT_CHARS = 2000 TOOL_RESULT_TRUNCATION_SUFFIX = "\n… [truncated]" KUBERNETES_TRACE_PREFIX = "FACTORY_K3S_TRACE_V1 " KUBERNETES_TRACE_VERSION = 1 MAX_KUBERNETES_TRACE_BYTES = 16 * 1024 _TRACE_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") _TRACE_PATH = re.compile(r"^[^\x00-\x1f\x7f]{1,500}$") _TRACE_EVENT_TYPES = frozenset({ "agent_start", "agent_end", "error", "gate_pass", "gate_fail", "handoff", "log", "phase_start", "phase_end", "tool_call_start", "tool_call", }) _TRACE_FIELDS: dict[str, frozenset[str]] = { "session_start": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "engineer", "adw_name", "repo", "repo_name", "host", "preset", "run_slug", "started_at", }), "session_finish": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "status", "ended_at", }), "session_usage": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "tokens", "cost", }), "phase": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "phase_id", "task_id", "seq", "name", "phase_kind", "owner", "status", "phase_attempt", "retries", "started_at", "ended_at", }), "event": frozenset({ "version", "kind", "attempt_id", "ts", "event_id", "adw_id", "phase_id", "parent_id", "event_type", "name", "metadata", "tokens", "started_at", "ended_at", }), "agent_attempt_start": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "phase_id", "agent_attempt_id", "parent_id", "agent", "session_id", "host", "account", "provider", "model", "started_at", }), "agent_attempt_finish": frozenset({ "version", "kind", "attempt_id", "ts", "agent_attempt_id", "returncode", "signal", "timed_out", "timeout_kind", "tokens", "input_tokens", "output_tokens", "cost", "ended_at", }), "agent_session": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "agent", "coding_agent", "model", "color", "session_id", "context_tokens", "context_window", "created_at", }), "gate": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "phase_id", "phase_attempt", "gate", "passed", "violation_count", "check_count", "checks", "created_at", }), "diff": frozenset({ "version", "kind", "attempt_id", "ts", "adw_id", "task_id", "phase_id", "agent_attempt_id", "phase_attempt", "files", "file_count", "insertions", "deletions", "truncated", "created_at", }), } def _trace_text(value: Any, *, limit: int = 200) -> str | None: if value is None: return None text = str(value) if not _TRACE_PATH.fullmatch(text) or len(text) > limit: return None return text def _trace_event_name(record: EventRecord) -> str | None: if record.type in {"tool_call_start", "tool_call"}: return None return _trace_text(record.name) def _trace_usage(usage: dict | None) -> dict[str, int | float]: if not isinstance(usage, dict): return {} output: dict[str, int | float] = {} aliases = { "input_tokens": ("input_tokens", "inputTokens"), "output_tokens": ("output_tokens", "outputTokens"), "cost": ("cost", "total_cost", "totalCost"), } for target, names in aliases.items(): value = next((usage[name] for name in names if name in usage), None) if target == "cost" and isinstance(value, (int, float)) and not isinstance(value, bool): output[target] = float(value) elif target != "cost" and isinstance(value, int) and not isinstance(value, bool) and value >= 0: output[target] = value return output def _trace_event_metadata(record: EventRecord) -> dict[str, Any]: payload = record.payload if isinstance(record.payload, dict) else {} metadata: dict[str, Any] = {} text_fields: dict[str, tuple[str, ...]] = { "phase_start": ("kind", "owner"), "phase_end": ("status",), "agent_start": ("model", "thinking", "color", "session_id", "client_session_id", "provider", "coding_agent"), "agent_end": ("provider", "model", "client_session_id", "billing_status"), "log": ("agent", "output_type", "level"), "error": ("agent",), "tool_call_start": ("tool_call_id", "attempt_id", "tool"), "tool_call": ("tool_call_id", "attempt_id"), } integer_fields: dict[str, tuple[str, ...]] = { "agent_end": ("context_tokens", "context_window", "max_tokens"), "log": ("attempt", "max_attempts"), "tool_call_start": ("seq",), "tool_call": ("duration_ms",), } for key in text_fields.get(record.type, ()): value = _trace_text(payload.get(key), limit=200) if value is not None: metadata[key] = value for key in integer_fields.get(record.type, ()): value = payload.get(key) if isinstance(value, int) and not isinstance(value, bool) and value >= 0: metadata[key] = value if record.type == "agent_end": cost = payload.get("cost") if isinstance(cost, (int, float)) and not isinstance(cost, bool) and cost >= 0: metadata["cost"] = float(cost) if record.type == "tool_call" and isinstance(payload.get("ok"), bool): metadata["ok"] = payload["ok"] if record.type in {"gate_pass", "gate_fail"}: attempt = payload.get("attempt") violations = payload.get("violations") if isinstance(attempt, int) and not isinstance(attempt, bool) and attempt >= 0: metadata["attempt"] = attempt if isinstance(violations, list): metadata["violation_count"] = len(violations) if record.type == "error": metadata["error"] = True return metadata DECISIONS_SCHEMA = """ CREATE TABLE IF NOT EXISTS decisions ( decision_id TEXT PRIMARY KEY, adw_id TEXT NOT NULL, phase TEXT, question TEXT NOT NULL, options TEXT NOT NULL DEFAULT '[]', free_text INTEGER NOT NULL DEFAULT 0, context TEXT NOT NULL DEFAULT '', status TEXT NOT NULL DEFAULT 'pending', answer_value TEXT, answer_text TEXT, answered_by TEXT, created_at TEXT NOT NULL, answered_at TEXT ); CREATE INDEX IF NOT EXISTS idx_decisions_status ON decisions(status); """ def _ensure_wal_mode(conn: sqlite3.Connection) -> None: """Switch to WAL; busy_timeout must already be set on ``conn``. First-time WAL activation takes an exclusive lock. Concurrent Tracer constructors on a legacy delete-mode file can still raise SQLITE_BUSY on this pragma even with busy_timeout, so retry only that locked case until the journal is wal or the 5s budget expires. """ deadline = time.monotonic() + 5.0 while True: try: mode = conn.execute("PRAGMA journal_mode=WAL;").fetchone()[0] except sqlite3.OperationalError as exc: if "database is locked" not in str(exc).lower(): raise else: if mode.lower() == "wal": return if time.monotonic() >= deadline: raise sqlite3.OperationalError("database is locked") time.sleep(0.001) def connect_db( db_path: str | Path, *, check_same_thread: bool = True, ) -> sqlite3.Connection: """Open SQLite with the observability WAL pragmas on every connection.""" conn = sqlite3.connect( str(db_path), isolation_level=None, timeout=5.0, check_same_thread=check_same_thread, ) conn.execute("PRAGMA busy_timeout=5000;") _ensure_wal_mode(conn) conn.execute("PRAGMA synchronous=NORMAL;") return conn class Tracer: def __init__( self, db_path: str | Path, events_jsonl: str | Path, *, emit_kubernetes_trace: bool = True, check_same_thread: bool = True, ): self.kubernetes_attempt_id = self._kubernetes_attempt_id() if emit_kubernetes_trace else None self._kubernetes_finished_sessions: set[str] = set() self._kubernetes_pending_tool_calls: dict[tuple[str, str, str], dict[str, Any]] = {} self._kubernetes_tool_call_ids: set[str] = set() ensure_dir(Path(db_path).parent) self.db_path = str(db_path) self.events_jsonl = Path(events_jsonl) ensure_dir(self.events_jsonl.parent) self.conn = connect_db(self.db_path, check_same_thread=check_same_thread) self.conn.executescript(SCHEMA) self._migrate() @staticmethod def _kubernetes_attempt_id() -> str | None: raw = os.environ.get("FACTORY_ATTEMPT_ENVELOPE") if raw is None: return None try: envelope = json.loads(raw) attempt_id = envelope["attempt_id"] except (KeyError, TypeError, json.JSONDecodeError) as exc: raise RuntimeError("FACTORY_ATTEMPT_ENVELOPE cannot identify the trace attempt") from exc if not isinstance(attempt_id, str) or not re.fullmatch(r"[0-9a-f]{24}", attempt_id): raise RuntimeError("FACTORY_ATTEMPT_ENVELOPE has an invalid trace attempt") return attempt_id def _kubernetes_trace(self, kind: str, **fields: Any) -> None: if self.kubernetes_attempt_id is None: return allowed = _TRACE_FIELDS.get(kind) if allowed is None: raise RuntimeError("unsupported Kubernetes trace record") record = { "version": KUBERNETES_TRACE_VERSION, "kind": kind, "attempt_id": self.kubernetes_attempt_id, "ts": now_iso(), **{key: value for key, value in fields.items() if value is not None}, } if not set(record) <= allowed: raise RuntimeError("Kubernetes trace record contains an unsafe field") encoded = json.dumps(record, separators=(",", ":"), ensure_ascii=True) line = f"{KUBERNETES_TRACE_PREFIX}{encoded}" if len(line.encode()) > MAX_KUBERNETES_TRACE_BYTES: raise RuntimeError("Kubernetes trace record exceeds its size limit") print(line, file=sys.stdout, flush=True) def _migrate(self) -> None: """Additive column migrations, so a db from an older SSSF still opens.""" for table, column, decl in MIGRATIONS: columns = {row[1] for row in self.conn.execute(f"PRAGMA table_info({table})")} if column not in columns: try: self.conn.execute(f"ALTER TABLE {table} ADD COLUMN {column} {decl}") except sqlite3.OperationalError as exc: # Two Tracers can open the same legacy db at once; the loser # of the race hits "duplicate column name" after the winner # already applied this migration. if "duplicate column name" not in str(exc).lower(): raise self._migrate_decisions_table() self._migrate_run_slug_index() def _migrate_run_slug_index(self) -> None: if "run_slug" not in { row[1] for row in self.conn.execute("PRAGMA table_info(sessions)") }: return _ensure_run_slug_unique_index(self.conn) def _migrate_decisions_table(self) -> None: """Ensure the decisions table exists on dbs opened before it shipped.""" try: self.conn.executescript(DECISIONS_SCHEMA) except sqlite3.OperationalError as exc: # Concurrent Tracer opens can race on first CREATE; IF NOT EXISTS # makes the winner/loser both safe except for transient locks. if "database is locked" not in str(exc).lower(): raise def _decision_row_to_dict(self, row: tuple) -> dict: return { "decision_id": row[0], "adw_id": row[1], "phase": row[2], "question": row[3], "options": json.loads(row[4]), "free_text": bool(row[5]), "context": row[6], "status": row[7], "answer_value": row[8], "answer_text": row[9], "answered_by": row[10], "created_at": row[11], "answered_at": row[12], } _DECISION_COLS = ( "decision_id, adw_id, phase, question, options, free_text, context," " status, answer_value, answer_text, answered_by, created_at, answered_at" ) # ── decisions (human-in-the-loop) ─────────────────────────────────────── def decision_request( self, adw_id: str, question: str, options: list[dict], free_text: bool = False, context: str = "", phase: str | None = None, ) -> str: if not free_text and not options: raise ValueError("options required when free_text is False") decision_id = uuid.uuid4().hex ts = now_iso() self.conn.execute( "INSERT INTO decisions (decision_id, adw_id, phase, question, options," " free_text, context, status, created_at) VALUES (?,?,?,?,?,?,?,?,?)", (decision_id, adw_id, phase, question, json.dumps(options), int(free_text), context, "pending", ts), ) self.event(EventRecord( adw_id=adw_id, type="decision_requested", name="decision", payload={"decision_id": decision_id, "question": question}, )) return decision_id def decision_get(self, decision_id: str) -> dict | None: row = self.conn.execute( f"SELECT {self._DECISION_COLS} FROM decisions WHERE decision_id=?", (decision_id,), ).fetchone() if row is None: return None return self._decision_row_to_dict(row) def decisions_pending(self, adw_id: str | None = None) -> list[dict]: if adw_id is None: rows = self.conn.execute( f"SELECT {self._DECISION_COLS} FROM decisions" " WHERE status='pending' ORDER BY created_at", ).fetchall() else: rows = self.conn.execute( f"SELECT {self._DECISION_COLS} FROM decisions" " WHERE status='pending' AND adw_id=? ORDER BY created_at", (adw_id,), ).fetchall() return [self._decision_row_to_dict(row) for row in rows] def decision_answer( self, decision_id: str, value: str | None = None, text: str | None = None, answered_by: str = "", ) -> bool: row = self.conn.execute( "SELECT options, free_text, status FROM decisions WHERE decision_id=?", (decision_id,), ).fetchone() if row is None: return False options_json, free_text_int, status = row if status != "pending": return False options: list[dict] = json.loads(options_json) free_text = bool(free_text_int) if value is None and text is None: raise ValueError("answer requires value or text") answer_value = value answer_text = text if value is not None and text is not None: answer_text = text if free_text else None if answer_text is not None and not free_text: raise ValueError("text answer not allowed when free_text is False") if answer_value is not None and options: valid = {opt["value"] for opt in options} if answer_value not in valid: raise ValueError(f"invalid choice: {answer_value!r}") if answer_value is None: if not free_text: raise ValueError("value required when free_text is False") if not answer_text: raise ValueError("text required for free_text decision") ts = now_iso() cur = self.conn.execute( "UPDATE decisions SET status='answered', answer_value=?, answer_text=?," " answered_by=?, answered_at=? WHERE decision_id=? AND status='pending'", (answer_value, answer_text, answered_by, ts, decision_id), ) if cur.rowcount == 0: return False answered = self.decision_get(decision_id) assert answered is not None self.event(EventRecord( adw_id=answered["adw_id"], type="decision_answered", name="decision", payload={ "decision_id": decision_id, "answer_value": answer_value, "answer_text": answer_text, }, )) return True def decisions_cancel(self, adw_id: str) -> int: rows = self.conn.execute( "SELECT decision_id, question FROM decisions" " WHERE adw_id=? AND status='pending'", (adw_id,), ).fetchall() if not rows: return 0 ts = now_iso() self.conn.execute( "UPDATE decisions SET status='canceled' WHERE adw_id=? AND status='pending'", (adw_id,), ) for decision_id, question in rows: self.event(EventRecord( adw_id=adw_id, type="decision_canceled", name="decision", payload={"decision_id": decision_id, "question": question}, )) return len(rows) # ── events ────────────────────────────────────────────────────────────── def event(self, record: EventRecord) -> str: event_id = f"evt_{new_id(12)}" ts = now_iso() line = {"event_id": event_id, "ts": ts, **record.model_dump()} with self.events_jsonl.open("a") as f: f.write(json.dumps(line) + "\n") if (record.type == "tool_call_start" and record.payload.get("tool_call_id") and record.payload.get("attempt_id")): self.tool_call_start( str(record.payload.get("tool_call_id") or ""), str(record.payload.get("attempt_id") or ""), int(record.payload.get("seq") or 0), record.payload.get("tool"), record.payload.get("args"), record.started_at or ts, ) elif (record.type == "tool_call" and record.payload.get("tool_call_id") and record.payload.get("attempt_id")): self.tool_call_finish( str(record.payload.get("tool_call_id") or ""), ended_at=record.ended_at, duration_ms=record.payload.get("duration_ms"), ok=record.payload.get("ok"), result=record.payload.get("result_snippet"), ) elif record.type == "agent_attempt_end": attempt = record.payload.get("attempt", {}) or {} if attempt.get("attemptId"): self.conn.execute( "UPDATE agent_attempts SET timeout_kind=? WHERE attempt_id=?", (attempt.get("timeoutKind"), attempt["attemptId"]), ) self.conn.execute( "INSERT INTO events (event_id, adw_id, phase_id, parent_id, type, name," " payload_json, tokens, started_at, ended_at) VALUES (?,?,?,?,?,?,?,?,?,?)", (event_id, record.adw_id, record.phase_id, record.parent_id, record.type, record.name, json.dumps(record.payload), record.tokens, record.started_at or ts, record.ended_at), ) if ( record.type not in _TRACE_EVENT_TYPES or record.adw_id in self._kubernetes_finished_sessions ): return event_id metadata = _trace_event_metadata(record) trace_fields = { "event_id": event_id, "adw_id": record.adw_id, "phase_id": _trace_text(record.phase_id), "parent_id": _trace_text(record.parent_id), "event_type": record.type, "name": _trace_event_name(record), "metadata": metadata, "tokens": record.tokens, "started_at": record.started_at or ts, "ended_at": record.ended_at, } tool_call_id = metadata.get("tool_call_id") attempt_id = metadata.get("attempt_id") if record.type == "tool_call_start": if ( {"tool_call_id", "attempt_id", "seq"} <= metadata.keys() and isinstance(tool_call_id, str) and isinstance(attempt_id, str) and tool_call_id not in self._kubernetes_tool_call_ids ): key = (record.adw_id, attempt_id, tool_call_id) self._kubernetes_pending_tool_calls[key] = trace_fields self._kubernetes_tool_call_ids.add(tool_call_id) elif record.type == "tool_call": if ( {"tool_call_id", "attempt_id", "ok"} <= metadata.keys() and isinstance(tool_call_id, str) and isinstance(attempt_id, str) ): key = (record.adw_id, attempt_id, tool_call_id) start = self._kubernetes_pending_tool_calls.pop(key, None) if start is not None: self._kubernetes_trace("event", **start) self._kubernetes_trace("event", **trace_fields) else: self._kubernetes_trace("event", **trace_fields) return event_id # ── sessions ──────────────────────────────────────────────────────────── def session_start(self, adw_id: str, engineer: str, adw_name: str | None = None, repo: str | Path | None = None, preset: str | None = None, slug_hint: str | None = None, slug_hint_is_path: bool = False, adw_script_stem: str | None = None, request_id: str | None = None) -> None: repo_path = str(Path(repo).resolve()) if repo is not None else None repo_name = derive_repo_name(repo_path) if repo_path is not None else None host = socket.gethostname() ts = now_iso() self.conn.execute("BEGIN IMMEDIATE") try: existing = self.conn.execute( "SELECT run_slug FROM sessions WHERE adw_id = ?", (adw_id,), ).fetchone() if existing is not None: self.conn.execute( "UPDATE sessions SET status='running'," " repo=COALESCE(repo, ?)," " host=COALESCE(host, ?)," " repo_name=COALESCE(repo_name, ?)," " preset=COALESCE(preset, ?)" " WHERE adw_id=?", (repo_path, host, repo_name, preset, adw_id), ) else: base = derive_slug_base( slug_hint, adw_script_stem, adw_id, hint_is_path=slug_hint_is_path, ) run_slug = pick_unique_slug(self.conn, base) self.conn.execute( "INSERT INTO sessions" " (adw_id, status, engineer, started_at, repo, host, run_slug," " repo_name, preset)" " VALUES (?,?,?,?,?,?,?,?,?)", (adw_id, "running", engineer, ts, repo_path, host, run_slug, repo_name, preset), ) if request_id is not None: linked = self.conn.execute( "SELECT request_id FROM request_run_links WHERE adw_id=?", (adw_id,), ).fetchone() if linked is not None and linked[0] != request_id: raise ValueError("factory run already belongs to a different request") if linked is None: self.conn.execute( "INSERT INTO request_run_links (request_id,adw_id,linked_at) VALUES (?,?,?)", (request_id, adw_id, ts), ) self.conn.execute("COMMIT") except Exception: self.conn.execute("ROLLBACK") raise if adw_name: row = self.conn.execute("SELECT adw_name FROM sessions WHERE adw_id=?", (adw_id,)).fetchone() names = row[0].split(" + ") if row and row[0] else [] if adw_name not in names: names.append(adw_name) self.conn.execute("UPDATE sessions SET adw_name=? WHERE adw_id=?", (" + ".join(names), adw_id)) row = self.conn.execute( "SELECT adw_name,repo,repo_name,host,preset,run_slug,started_at" " FROM sessions WHERE adw_id=?", (adw_id,), ).fetchone() assert row is not None self._kubernetes_finished_sessions.discard(adw_id) self._kubernetes_pending_tool_calls.clear() self._kubernetes_tool_call_ids.clear() self._kubernetes_trace( "session_start", adw_id=adw_id, engineer=_trace_text(engineer), adw_name=_trace_text(row[0]), repo=_trace_text(row[1], limit=500), repo_name=_trace_text(row[2]), host=_trace_text(row[3]), preset=_trace_text(row[4]), run_slug=_trace_text(row[5]), request_id=_trace_text(request_id), started_at=row[6], ) def session_request(self, adw_id: str, request: str) -> None: self.conn.execute("UPDATE sessions SET request=? WHERE adw_id=?", (request[:500], adw_id)) def session_finish(self, adw_id: str, ok: bool) -> None: ended_at = now_iso() status = "success" if ok else "fail" self.conn.execute( "UPDATE sessions SET status=?, ended_at=? WHERE adw_id=?", (status, ended_at, adw_id), ) self.processes_end_all(adw_id) # nothing of this run is alive any more self._kubernetes_trace( "session_finish", adw_id=adw_id, status=status, ended_at=ended_at, ) self._kubernetes_pending_tool_calls.clear() self._kubernetes_tool_call_ids.clear() self._kubernetes_finished_sessions.add(adw_id) def session_add_usage(self, adw_id: str, tokens: int, cost: float) -> None: self.conn.execute( "UPDATE sessions SET total_tokens=total_tokens+?, total_cost=total_cost+? WHERE adw_id=?", (tokens, cost, adw_id), ) self._kubernetes_trace( "session_usage", adw_id=adw_id, tokens=tokens, cost=cost, ) # ── processes (adw_id → pid, so a hung run can be found and killed) ───── def process_start(self, adw_id: str, kind: str, name: str, pid: int, command: str, *, start_ticks: int | None = None) -> None: """Record a live process for this run. A coding agent that hangs produces no events at all, which is exactly when you need its pid — and `ps` cannot tell you which adw_id it belongs to. Writing it here makes the trace the answer to "what is this run running, and how do I stop it". """ self.conn.execute( "INSERT INTO processes (adw_id, kind, name, pid, start_ticks, command, started_at)" " VALUES (?,?,?,?,?,?,?)", (adw_id, kind, name, pid, start_ticks, command, now_iso()), ) def process_end(self, adw_id: str, pid: int) -> None: """Mark the newest live row for this pid as finished.""" self.conn.execute( "UPDATE processes SET ended_at=? WHERE id = (" " SELECT id FROM processes WHERE adw_id=? AND pid=? AND ended_at IS NULL" " ORDER BY id DESC LIMIT 1)", (now_iso(), adw_id, pid), ) def processes_end_all(self, adw_id: str) -> None: """Close out every live row for a run — called when the session ends.""" self.conn.execute( "UPDATE processes SET ended_at=? WHERE adw_id=? AND ended_at IS NULL", (now_iso(), adw_id), ) # ── agent attempts (one row per subprocess invocation) ───────────────── def agent_attempt_start(self, adw_id: str, phase_id: str, agent: str, session_id: str, command: str, *, host: str, account: str | None, provider: str | None = None, model: str, system_prompt: str | None = None, user_prompt: str | None = None, parent_id: str = "") -> str: """Create durable evidence before an agent child is spawned.""" attempt_id = f"att_{new_id(12)}" started_at = now_iso() self.conn.execute( "INSERT INTO agent_attempts (attempt_id,adw_id,phase_id,parent_id,agent," "session_id,command,system_prompt,user_prompt,host,account,model,started_at)" " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (attempt_id, adw_id, phase_id, parent_id, agent, session_id, command, system_prompt, user_prompt, host, account, model, started_at), ) self._kubernetes_trace( "agent_attempt_start", adw_id=adw_id, phase_id=phase_id, agent_attempt_id=attempt_id, parent_id=_trace_text(parent_id), agent=_trace_text(agent), session_id=_trace_text(session_id), host=_trace_text(host), account=_trace_text(account), provider=_trace_text(provider), model=_trace_text(model), started_at=started_at, ) return attempt_id def agent_attempt_finish(self, attempt_id: str, *, returncode: int | None, signal: int | None, timed_out: bool, stderr_path: str | None, tokens: int | None, usage: dict | None, error: str | None = None, timeout_kind: str | None = None, provider_failure: dict | None = None) -> None: """Finalize an attempt without requiring any artifact to exist.""" usage_json = None if usage is not None or provider_failure is not None: payload = dict(usage or {}) if provider_failure is not None: payload["provider_failure"] = provider_failure usage_json = json.dumps(payload) ended_at = now_iso() self.conn.execute( "UPDATE agent_attempts SET returncode=?,signal=?,timed_out=?,timeout_kind=?," "stderr_path=?,tokens=?,usage_json=?,error=?,ended_at=? WHERE attempt_id=?", (returncode, signal, int(timed_out), timeout_kind, stderr_path, tokens, usage_json, error, ended_at, attempt_id), ) self._kubernetes_trace( "agent_attempt_finish", agent_attempt_id=attempt_id, returncode=returncode, signal=signal, timed_out=bool(timed_out), timeout_kind=_trace_text(timeout_kind), tokens=tokens, ended_at=ended_at, **_trace_usage(usage), ) # ── tool calls (one row per invocation, committed while the child runs) ─ def tool_call_start(self, tool_call_id: str, attempt_id: str, seq: int, tool_name: str | None, args: dict | None, started_at: str | None = None) -> None: self.conn.execute( "INSERT INTO tool_calls (tool_call_id,attempt_id,seq,tool_name,args_json,started_at)" " VALUES (?,?,?,?,?,?)", (tool_call_id, attempt_id, seq, tool_name, json.dumps(args) if args is not None else None, started_at or now_iso()), ) def tool_call_finish(self, tool_call_id: str, *, ended_at: str | None, duration_ms: int | None, ok: bool | None, result: str | None) -> None: excerpt = result if excerpt is not None and len(excerpt) > TOOL_RESULT_EXCERPT_CHARS: keep = TOOL_RESULT_EXCERPT_CHARS - len(TOOL_RESULT_TRUNCATION_SUFFIX) excerpt = excerpt[:keep].rstrip() + TOOL_RESULT_TRUNCATION_SUFFIX self.conn.execute( "UPDATE tool_calls SET ended_at=?,duration_ms=?,ok=?,result_excerpt=?" " WHERE tool_call_id=?", (ended_at or now_iso(), duration_ms, None if ok is None else int(ok), excerpt, tool_call_id), ) # ── phases ────────────────────────────────────────────────────────────── def max_phase_seq(self, adw_id: str) -> int: """Highest seq already recorded for this session; 0 when it is new. A joined run continues the sequence instead of restarting at 1 — which would collide with the first run's phases on both `seq` (breaking ordering) and `phase_id` (silently overwriting a row through the phase_upsert conflict clause). """ row = self.conn.execute("SELECT MAX(seq) FROM phases WHERE adw_id = ?", (adw_id,)).fetchone() return row[0] if row and row[0] is not None else 0 def phase_upsert(self, phase: Phase) -> None: p = phase.params self.conn.execute( "INSERT INTO phases (phase_id, task_id, adw_id, seq, name, kind, owner, description," " status, attempt, retries, error, started_at, ended_at)" " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)" " ON CONFLICT(phase_id) DO UPDATE SET task_id=excluded.task_id, status=excluded.status," " attempt=excluded.attempt, error=excluded.error, ended_at=excluded.ended_at", (phase.phase_id, phase.task_id, phase.adw_id, phase.seq, p.name, p.kind, p.owner, p.description, phase.status, phase.attempt, p.retries, phase.error, phase.started_at, phase.ended_at), ) self._kubernetes_trace( "phase", adw_id=phase.adw_id, phase_id=phase.phase_id, task_id=_trace_text(phase.task_id), seq=phase.seq, name=_trace_text(p.name), phase_kind=_trace_text(p.kind), owner=_trace_text(p.owner), status=_trace_text(phase.status), phase_attempt=phase.attempt, retries=p.retries, started_at=phase.started_at, ended_at=phase.ended_at, ) # ── envelopes / gates / agent sessions ────────────────────────────────── def envelope_row(self, phase: Phase, agent: str, output_type: str, payload_json: str, valid: bool, attempt: int) -> None: self.conn.execute( "INSERT INTO envelopes (envelope_id, adw_id, phase_id, agent, output_type," " payload_json, valid, attempt, created_at) VALUES (?,?,?,?,?,?,?,?,?)", (f"env_{new_id(12)}", phase.adw_id, phase.phase_id, agent, output_type, payload_json, int(valid), attempt, now_iso()), ) def gate_row(self, phase: Phase, gate: str, report: GateReport, attempt: int, *, violations: list[str] | None = None, checks: list[dict] | None = None) -> None: """The report carries both the verdict and the evidence behind it.""" created_at = now_iso() gate_checks = [c.model_dump() for c in report.checks] if checks is None else checks self.conn.execute( "INSERT INTO gate_results (adw_id, phase_id, attempt, gate, passed," " violations_json, checks_json, created_at) VALUES (?,?,?,?,?,?,?,?)", (phase.adw_id, phase.phase_id, attempt, gate, int(report.passed), json.dumps(report.violations if violations is None else violations), json.dumps(gate_checks), created_at), ) safe_checks = [] for check in gate_checks[:64]: if not isinstance(check, dict): continue item = _trace_text(check.get("item"), limit=120) ok = check.get("ok") if item is not None and isinstance(ok, bool): safe_checks.append({"item": item, "ok": ok}) gate_violations = report.violations if violations is None else violations self._kubernetes_trace( "gate", adw_id=phase.adw_id, phase_id=phase.phase_id, phase_attempt=attempt, gate=_trace_text(gate), passed=bool(report.passed), violation_count=len(gate_violations), check_count=len(gate_checks), checks=safe_checks, created_at=created_at, ) def phase_diff_row(self, phase: Phase, diff: dict) -> None: attempts = self.conn.execute( "SELECT attempt_id FROM agent_attempts WHERE adw_id=? AND phase_id=? " "ORDER BY started_at, attempt_id", (phase.adw_id, phase.phase_id), ).fetchall() attempt_id = attempts[0][0] if len(attempts) == 1 else None created_at = now_iso() self.conn.execute( "INSERT INTO phase_diffs (adw_id,task_id,phase_id,attempt_id,attempt," "files_json,insertions,deletions,diff_text,truncated,created_at) " "VALUES (?,?,?,?,?,?,?,?,?,?,?)", (phase.adw_id, phase.task_id, phase.phase_id, attempt_id, phase.attempt, json.dumps(diff["files"]), diff["insertions"], diff["deletions"], diff["diff_text"], int(diff["truncated"]), created_at), ) files = [ safe for value in diff["files"][:20] if (safe := _trace_text(value, limit=500)) is not None ] self._kubernetes_trace( "diff", adw_id=phase.adw_id, task_id=_trace_text(phase.task_id), phase_id=phase.phase_id, agent_attempt_id=attempt_id, phase_attempt=phase.attempt, files=files, file_count=len(diff["files"]), insertions=diff["insertions"], deletions=diff["deletions"], truncated=bool(diff["truncated"] or len(files) != len(diff["files"])), created_at=created_at, ) def agent_session_row(self, adw_id: str, agent: AgentConfig, session_id: str, context_tokens: int = 0, context_window: int = 0) -> None: """The agent's config row is the source of truth for its label and color. Context is carried here rather than derived from events because the lane wants one number per agent — the latest — and a session that runs the same agent twice overwrites it, exactly like model and session_id. """ ts = now_iso() self.conn.execute( "INSERT INTO agent_sessions (adw_id, agent, coding_agent, model, color," " session_id, context_tokens, context_window, created_at, last_used_at)" " VALUES (?,?,?,?,?,?,?,?,?,?)" " ON CONFLICT(adw_id, agent) DO UPDATE SET model=excluded.model," " color=excluded.color, session_id=excluded.session_id," " context_tokens=excluded.context_tokens," " context_window=excluded.context_window," " last_used_at=excluded.last_used_at", (adw_id, agent.name, agent.coding_agent, agent.model, agent.color, session_id, context_tokens, context_window, ts, ts), ) self._kubernetes_trace( "agent_session", adw_id=adw_id, agent=_trace_text(agent.name), coding_agent=_trace_text(agent.coding_agent), model=_trace_text(agent.model), color=_trace_text(agent.color), session_id=_trace_text(session_id), context_tokens=context_tokens, context_window=context_window, created_at=ts, ) class KubernetesTraceMirror: def __init__( self, db_path: str | Path, events_jsonl: str | Path, attempt_id: str, placement: dict[str, str], ): if not re.fullmatch(r"[0-9a-f]{24}", attempt_id): raise ValueError("invalid Kubernetes trace attempt") allowed_placement = {"cluster", "namespace", "job", "job_uid", "pod", "pod_uid", "node", "container"} if set(placement) - allowed_placement or any( not isinstance(value, str) or _trace_text(value, limit=253) is None for value in placement.values() ): raise ValueError("invalid Kubernetes trace placement") self.tracer = Tracer( db_path, events_jsonl, emit_kubernetes_trace=False, check_same_thread=False, ) self.attempt_id = attempt_id self.placement = dict(placement) self.adw_id: str | None = None self.started = False self.finished = False self.placement_written = False self.phase_ids: set[str] = set() self.event_ids: set[str] = set() self.agent_attempt_ids: set[str] = set() self.agent_names: set[str] = set() self.tool_call_ids: set[str] = set() @staticmethod def _integer(record: dict[str, Any], name: str, *, optional: bool = False) -> int | None: value = record.get(name) if value is None and optional: return None if not isinstance(value, int) or isinstance(value, bool) or value < 0: raise ValueError(f"Kubernetes trace field {name} is invalid") return value @staticmethod def _number(record: dict[str, Any], name: str) -> float: value = record.get(name) if not isinstance(value, (int, float)) or isinstance(value, bool) or value < 0: raise ValueError(f"Kubernetes trace field {name} is invalid") return float(value) @staticmethod def _text(record: dict[str, Any], name: str, *, optional: bool = False, limit: int = 200) -> str | None: value = record.get(name) if value is None and optional: return None if not isinstance(value, str) or _trace_text(value, limit=limit) is None: raise ValueError(f"Kubernetes trace field {name} is invalid") return value def _session(self, record: dict[str, Any]) -> str: adw_id = self._text(record, "adw_id") assert adw_id is not None if ( self.adw_id is None or adw_id != self.adw_id or self.tracer.conn.execute( "SELECT 1 FROM sessions WHERE adw_id=?", (adw_id,), ).fetchone() is None ): raise ValueError("Kubernetes trace session ownership mismatch") return adw_id def _owned_phase(self, phase_id: str) -> None: row = self.tracer.conn.execute( "SELECT adw_id FROM phases WHERE phase_id=?", (phase_id,), ).fetchone() if phase_id not in self.phase_ids or row != (self.adw_id,): raise ValueError("Kubernetes trace phase ownership mismatch") def _owned_event(self, event_id: str) -> None: row = self.tracer.conn.execute( "SELECT adw_id FROM events WHERE event_id=?", (event_id,), ).fetchone() if event_id not in self.event_ids or row != (self.adw_id,): raise ValueError("Kubernetes trace event ownership mismatch") def _owned_agent_attempt(self, attempt_id: str) -> None: row = self.tracer.conn.execute( "SELECT adw_id FROM agent_attempts WHERE attempt_id=?", (attempt_id,), ).fetchone() if attempt_id not in self.agent_attempt_ids or row != (self.adw_id,): raise ValueError("Kubernetes trace agent attempt ownership mismatch") def _controller_event(self, suffix: str, event_type: str, payload: dict[str, Any]) -> None: assert self.adw_id is not None event_id = f"k8s_{suffix}_{self.attempt_id}" existing = self.tracer.conn.execute( "SELECT adw_id,type FROM events WHERE event_id=?", (event_id,), ).fetchone() if existing is not None and existing != (self.adw_id, event_type): raise ValueError("Kubernetes controller event ownership mismatch") timestamp = now_iso() self.tracer.conn.execute( "INSERT INTO events (event_id,adw_id,type,name,payload_json,started_at)" " VALUES (?,?,?,?,?,?) ON CONFLICT(event_id) DO UPDATE SET" " payload_json=excluded.payload_json,started_at=excluded.started_at", (event_id, self.adw_id, event_type, "k3s", json.dumps(payload), timestamp), ) def consume(self, line: bytes | str) -> bool: raw = line.encode() if isinstance(line, str) else line prefix = KUBERNETES_TRACE_PREFIX.encode() if not raw.startswith(prefix): return False if len(raw) > MAX_KUBERNETES_TRACE_BYTES: raise ValueError("Kubernetes trace record exceeds its size limit") try: record = json.loads(raw[len(prefix):]) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ValueError("malformed Kubernetes trace record") from exc if not isinstance(record, dict): raise ValueError("malformed Kubernetes trace record") kind = record.get("kind") allowed = _TRACE_FIELDS.get(kind) if isinstance(kind, str) else None if ( allowed is None or set(record) - allowed or record.get("version") != KUBERNETES_TRACE_VERSION or record.get("attempt_id") != self.attempt_id or self._text(record, "ts") is None ): raise ValueError("Kubernetes trace record contract mismatch") if self.finished: raise ValueError("Kubernetes trace record followed session finish") handler = getattr(self, f"_apply_{kind}") state = ( self.adw_id, self.started, self.finished, self.placement_written, self.phase_ids.copy(), self.event_ids.copy(), self.agent_attempt_ids.copy(), self.agent_names.copy(), self.tool_call_ids.copy(), ) try: self.tracer.conn.execute("BEGIN IMMEDIATE") handler(record) self.tracer.conn.execute("COMMIT") except BaseException: if self.tracer.conn.in_transaction: self.tracer.conn.execute("ROLLBACK") ( self.adw_id, self.started, self.finished, self.placement_written, self.phase_ids, self.event_ids, self.agent_attempt_ids, self.agent_names, self.tool_call_ids, ) = state raise return True def _apply_session_start(self, record: dict[str, Any]) -> None: adw_id = self._text(record, "adw_id") assert adw_id is not None if self.adw_id is None: if self.tracer.conn.execute("SELECT 1 FROM sessions WHERE adw_id=?", (adw_id,)).fetchone(): raise ValueError("Kubernetes trace session already exists") self.adw_id = adw_id requested_slug = self._text(record, "run_slug", optional=True) run_slug = pick_unique_slug(self.tracer.conn, requested_slug or adw_id) self.tracer.conn.execute( "INSERT INTO sessions (adw_id,adw_name,repo,status,engineer,host,started_at," "run_slug,repo_name,preset) VALUES (?,?,?,?,?,?,?,?,?,?)", ( adw_id, self._text(record, "adw_name", optional=True), self._text(record, "repo", optional=True, limit=500), "running", self._text(record, "engineer", optional=True), self._text(record, "host", optional=True), self._text(record, "started_at"), run_slug, self._text(record, "repo_name", optional=True), self._text(record, "preset", optional=True), ), ) self.started = True else: if ( adw_id != self.adw_id or self.tracer.conn.execute( "SELECT 1 FROM sessions WHERE adw_id=?", (adw_id,), ).fetchone() is None ): raise ValueError("Kubernetes trace session ownership mismatch") self.tracer.conn.execute( "UPDATE sessions SET status='running',ended_at=NULL," "adw_name=COALESCE(?,adw_name),repo=COALESCE(?,repo)," "repo_name=COALESCE(?,repo_name),preset=COALESCE(?,preset) WHERE adw_id=?", ( self._text(record, "adw_name", optional=True), self._text(record, "repo", optional=True, limit=500), self._text(record, "repo_name", optional=True), self._text(record, "preset", optional=True), adw_id, ), ) self.finished = False request_id = self._text(record, "request_id", optional=True) if request_id is not None: linked = self.tracer.conn.execute( "SELECT request_id FROM request_run_links WHERE adw_id=?", (adw_id,), ).fetchone() if linked is not None and linked[0] != request_id: raise ValueError("Kubernetes trace request ownership mismatch") if linked is None: self.tracer.conn.execute( "INSERT INTO request_run_links (request_id,adw_id,linked_at) VALUES (?,?,?)", (request_id, adw_id, self._text(record, "started_at")), ) if not self.placement_written: self._controller_event( "placement", "kubernetes_placement", {"attempt_id": self.attempt_id, **self.placement}, ) self.placement_written = True def _apply_session_finish(self, record: dict[str, Any]) -> None: adw_id = self._session(record) status = self._text(record, "status") if status not in {"success", "fail"}: raise ValueError("Kubernetes trace session status is invalid") self.tracer.conn.execute( "UPDATE sessions SET status=?,ended_at=? WHERE adw_id=?", (status, self._text(record, "ended_at"), adw_id), ) self.finished = True def _apply_session_usage(self, record: dict[str, Any]) -> None: adw_id = self._session(record) self.tracer.conn.execute( "UPDATE sessions SET total_tokens=total_tokens+?,total_cost=total_cost+? WHERE adw_id=?", (self._integer(record, "tokens"), self._number(record, "cost"), adw_id), ) def _apply_phase(self, record: dict[str, Any]) -> None: adw_id = self._session(record) phase_id = self._text(record, "phase_id") assert phase_id is not None existing = self.tracer.conn.execute( "SELECT adw_id FROM phases WHERE phase_id=?", (phase_id,), ).fetchone() if phase_id in self.phase_ids: if existing != (adw_id,): raise ValueError("Kubernetes trace phase ownership mismatch") elif existing is not None: raise ValueError("Kubernetes trace phase identifier conflict") self.tracer.conn.execute( "INSERT INTO phases (phase_id,task_id,adw_id,seq,name,kind,owner,status,attempt," "retries,started_at,ended_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)" " ON CONFLICT(phase_id) DO UPDATE SET status=excluded.status,attempt=excluded.attempt," "ended_at=excluded.ended_at", ( phase_id, self._text(record, "task_id", optional=True), adw_id, self._integer(record, "seq"), self._text(record, "name", optional=True), self._text(record, "phase_kind", optional=True), self._text(record, "owner", optional=True), self._text(record, "status"), self._integer(record, "phase_attempt"), self._integer(record, "retries"), self._text(record, "started_at", optional=True), self._text(record, "ended_at", optional=True), ), ) self.phase_ids.add(phase_id) def _apply_event(self, record: dict[str, Any]) -> None: adw_id = self._session(record) event_id = self._text(record, "event_id") event_type = self._text(record, "event_type") assert event_id is not None and event_type is not None if event_type not in _TRACE_EVENT_TYPES: raise ValueError("Kubernetes trace event type is invalid") if event_id in self.event_ids or self.tracer.conn.execute( "SELECT 1 FROM events WHERE event_id=?", (event_id,), ).fetchone(): raise ValueError("Kubernetes trace event identifier conflict") phase_id = self._text(record, "phase_id", optional=True) if phase_id is not None: self._owned_phase(phase_id) parent_id = self._text(record, "parent_id", optional=True) if parent_id is not None: self._owned_event(parent_id) metadata = record.get("metadata") if not isinstance(metadata, dict): raise ValueError("Kubernetes trace event metadata is invalid") allowed_metadata = { "phase_start": {"kind", "owner"}, "phase_end": {"status"}, "agent_start": {"model", "thinking", "color", "session_id", "client_session_id", "provider", "coding_agent"}, "agent_end": {"provider", "model", "client_session_id", "billing_status", "context_tokens", "context_window", "max_tokens", "cost"}, "log": {"agent", "output_type", "level", "attempt", "max_attempts"}, "error": {"agent", "error"}, "gate_pass": {"attempt", "violation_count"}, "gate_fail": {"attempt", "violation_count"}, "handoff": set(), "tool_call_start": {"tool_call_id", "attempt_id", "tool", "seq"}, "tool_call": {"tool_call_id", "attempt_id", "duration_ms", "ok"}, }[event_type] if set(metadata) - allowed_metadata: raise ValueError("Kubernetes trace event metadata fields are invalid") for key, value in metadata.items(): if key in {"attempt", "max_attempts", "violation_count", "context_tokens", "context_window", "max_tokens", "seq", "duration_ms"}: if not isinstance(value, int) or isinstance(value, bool) or value < 0: raise ValueError("Kubernetes trace event count is invalid") elif key == "cost": if not isinstance(value, (int, float)) or isinstance(value, bool) or value < 0: raise ValueError("Kubernetes trace event cost is invalid") elif key in {"ok", "error"}: if not isinstance(value, bool): raise ValueError("Kubernetes trace event verdict is invalid") elif not isinstance(value, str) or _trace_text(value, limit=200) is None: raise ValueError("Kubernetes trace event text is invalid") tokens = self._integer(record, "tokens", optional=True) started_at = self._text(record, "started_at", optional=True) ended_at = self._text(record, "ended_at", optional=True) self.tracer.conn.execute( "INSERT INTO events (event_id,adw_id,phase_id,parent_id,type,name,payload_json,tokens," "started_at,ended_at) VALUES (?,?,?,?,?,?,?,?,?,?)", ( event_id, adw_id, phase_id, parent_id, event_type, self._text(record, "name", optional=True), json.dumps(metadata), tokens, started_at, ended_at, ), ) self.event_ids.add(event_id) if event_type == "tool_call_start": tool_call_id = self._text(metadata, "tool_call_id") attempt_id = self._text(metadata, "attempt_id") assert tool_call_id is not None and attempt_id is not None self._owned_agent_attempt(attempt_id) if tool_call_id in self.tool_call_ids or self.tracer.conn.execute( "SELECT 1 FROM tool_calls WHERE tool_call_id=?", (tool_call_id,), ).fetchone(): raise ValueError("Kubernetes trace tool call identifier conflict") self.tracer.conn.execute( "INSERT INTO tool_calls (tool_call_id,attempt_id,seq,tool_name,started_at)" " VALUES (?,?,?,?,?)", ( tool_call_id, attempt_id, self._integer(metadata, "seq"), self._text(metadata, "tool", optional=True), started_at, ), ) self.tool_call_ids.add(tool_call_id) elif event_type == "tool_call": tool_call_id = self._text(metadata, "tool_call_id") attempt_id = self._text(metadata, "attempt_id") assert tool_call_id is not None and attempt_id is not None self._owned_agent_attempt(attempt_id) row = self.tracer.conn.execute( "SELECT tc.attempt_id,aa.adw_id FROM tool_calls tc JOIN agent_attempts aa" " ON aa.attempt_id=tc.attempt_id WHERE tc.tool_call_id=?", (tool_call_id,), ).fetchone() if tool_call_id not in self.tool_call_ids or row != (attempt_id, adw_id): raise ValueError("Kubernetes trace tool call ownership mismatch") ok = metadata.get("ok") if not isinstance(ok, bool): raise ValueError("Kubernetes trace tool call verdict is invalid") self.tracer.conn.execute( "UPDATE tool_calls SET ended_at=?,duration_ms=?,ok=? WHERE tool_call_id=?", (ended_at, self._integer(metadata, "duration_ms", optional=True), int(ok), tool_call_id), ) def _apply_agent_attempt_start(self, record: dict[str, Any]) -> None: adw_id = self._session(record) attempt_id = self._text(record, "agent_attempt_id") phase_id = self._text(record, "phase_id") assert attempt_id is not None and phase_id is not None self._owned_phase(phase_id) parent_id = self._text(record, "parent_id", optional=True) if parent_id is not None: self._owned_event(parent_id) if attempt_id in self.agent_attempt_ids or self.tracer.conn.execute( "SELECT 1 FROM agent_attempts WHERE attempt_id=?", (attempt_id,), ).fetchone(): raise ValueError("Kubernetes trace agent attempt identifier conflict") self.tracer.conn.execute( "INSERT INTO agent_attempts (attempt_id,adw_id,phase_id,parent_id,agent,session_id," "host,account,model,started_at) VALUES (?,?,?,?,?,?,?,?,?,?)", ( attempt_id, adw_id, phase_id, parent_id, self._text(record, "agent"), self._text(record, "session_id", optional=True), self._text(record, "host", optional=True), self._text(record, "account", optional=True), self._text(record, "model", optional=True), self._text(record, "started_at"), ), ) self.agent_attempt_ids.add(attempt_id) def _apply_agent_attempt_finish(self, record: dict[str, Any]) -> None: attempt_id = self._text(record, "agent_attempt_id") assert attempt_id is not None self._owned_agent_attempt(attempt_id) usage = { key: record[key] for key in ("input_tokens", "output_tokens", "cost") if key in record } cursor = self.tracer.conn.execute( "UPDATE agent_attempts SET returncode=?,signal=?,timed_out=?,timeout_kind=?,tokens=?," "usage_json=?,ended_at=? WHERE attempt_id=? AND adw_id=?", ( self._integer(record, "returncode", optional=True), self._integer(record, "signal", optional=True), int(record.get("timed_out", False)), self._text(record, "timeout_kind", optional=True), self._integer(record, "tokens", optional=True), json.dumps(usage) if usage else None, self._text(record, "ended_at"), attempt_id, self.adw_id, ), ) if cursor.rowcount != 1 or not isinstance(record.get("timed_out"), bool): raise ValueError("Kubernetes trace agent attempt ownership mismatch") def _apply_agent_session(self, record: dict[str, Any]) -> None: adw_id = self._session(record) created_at = self._text(record, "created_at") agent_name = self._text(record, "agent") assert agent_name is not None existing = self.tracer.conn.execute( "SELECT adw_id FROM agent_sessions WHERE adw_id=? AND agent=?", (adw_id, agent_name), ).fetchone() if agent_name in self.agent_names: if existing != (adw_id,): raise ValueError("Kubernetes trace agent session ownership mismatch") elif existing is not None: raise ValueError("Kubernetes trace agent session identifier conflict") self.tracer.conn.execute( "INSERT INTO agent_sessions (adw_id,agent,coding_agent,model,color,session_id," "context_tokens,context_window,created_at,last_used_at) VALUES (?,?,?,?,?,?,?,?,?,?)" " ON CONFLICT(adw_id,agent) DO UPDATE SET model=excluded.model,color=excluded.color," "session_id=excluded.session_id,context_tokens=excluded.context_tokens," "context_window=excluded.context_window,last_used_at=excluded.last_used_at", ( adw_id, agent_name, self._text(record, "coding_agent"), self._text(record, "model"), self._text(record, "color", optional=True), self._text(record, "session_id"), self._integer(record, "context_tokens"), self._integer(record, "context_window"), created_at, created_at, ), ) self.agent_names.add(agent_name) def _apply_gate(self, record: dict[str, Any]) -> None: adw_id = self._session(record) phase_id = self._text(record, "phase_id") assert phase_id is not None self._owned_phase(phase_id) checks = record.get("checks") if not isinstance(checks, list) or len(checks) > 64: raise ValueError("Kubernetes trace gate checks are invalid") check_count = self._integer(record, "check_count") if check_count < len(checks): raise ValueError("Kubernetes trace gate check count is invalid") safe_checks = [] for check in checks: if set(check) != {"item", "ok"} or not isinstance(check.get("ok"), bool): raise ValueError("Kubernetes trace gate check is invalid") safe_checks.append({"item": self._text(check, "item", limit=120), "ok": check["ok"]}) count = self._integer(record, "violation_count") self.tracer.conn.execute( "INSERT INTO gate_results (adw_id,phase_id,attempt,gate,passed,violations_json," "checks_json,created_at) VALUES (?,?,?,?,?,?,?,?)", ( adw_id, phase_id, self._integer(record, "phase_attempt"), self._text(record, "gate"), int(record.get("passed") is True), json.dumps([f"{count} violation(s)"] if count else []), json.dumps(safe_checks), self._text(record, "created_at"), ), ) if not isinstance(record.get("passed"), bool): raise ValueError("Kubernetes trace gate verdict is invalid") def _apply_diff(self, record: dict[str, Any]) -> None: adw_id = self._session(record) phase_id = self._text(record, "phase_id") assert phase_id is not None self._owned_phase(phase_id) agent_attempt_id = self._text(record, "agent_attempt_id", optional=True) if agent_attempt_id is not None: self._owned_agent_attempt(agent_attempt_id) files = record.get("files") if not isinstance(files, list) or len(files) > 20: raise ValueError("Kubernetes trace diff files are invalid") file_count = self._integer(record, "file_count") if file_count < len(files): raise ValueError("Kubernetes trace diff file count is invalid") safe_files = [self._text({"file": value}, "file", limit=500) for value in files] truncated = record.get("truncated") if not isinstance(truncated, bool): raise ValueError("Kubernetes trace diff truncation is invalid") self.tracer.conn.execute( "INSERT INTO phase_diffs (adw_id,task_id,phase_id,attempt_id,attempt,files_json," "insertions,deletions,diff_text,truncated,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?)", ( adw_id, self._text(record, "task_id", optional=True), phase_id, agent_attempt_id, self._integer(record, "phase_attempt"), json.dumps(safe_files), self._integer(record, "insertions"), self._integer(record, "deletions"), "", int(truncated), self._text(record, "created_at"), ), ) def complete(self) -> None: if not self.started or not self.finished or self.adw_id is None: raise RuntimeError("Kubernetes trace is missing session start or finish") unfinished_phases = self.tracer.conn.execute( "SELECT COUNT(*) FROM phases WHERE adw_id=? AND" " (status NOT IN ('success','fail') OR ended_at IS NULL)", (self.adw_id,), ).fetchone()[0] unfinished_attempts = self.tracer.conn.execute( "SELECT COUNT(*) FROM agent_attempts WHERE adw_id=? AND ended_at IS NULL", (self.adw_id,), ).fetchone()[0] unfinished_tools = self.tracer.conn.execute( "SELECT COUNT(*) FROM tool_calls tc JOIN agent_attempts aa" " ON aa.attempt_id=tc.attempt_id WHERE aa.adw_id=? AND tc.ended_at IS NULL", (self.adw_id,), ).fetchone()[0] if unfinished_phases or unfinished_attempts or unfinished_tools: raise RuntimeError( "Kubernetes trace has unfinished phases, agent attempts, or tool calls", ) def fail(self) -> None: if self.adw_id is None: return self.tracer.conn.execute( "UPDATE sessions SET status='fail',ended_at=COALESCE(ended_at,?) WHERE adw_id=?", (now_iso(), self.adw_id), ) self._controller_event( "failure", "kubernetes_trace_failure", {"attempt_id": self.attempt_id, "status": "fail"}, ) def close(self) -> None: self.tracer.conn.close()