Nick
ca4f7593d7
CORRECTION 1: #314 reopened — recall_with_associations mode field defect CONFIRMED in current source and FIXED (mode: association_merged). CORRECTION 3: Gitea credential removed from remote URL (credential helper). TOKEN EXPOSED: yes (in diagnostic output). TOKEN ROTATED: not yet (requires Gitea web UI access). Remote URL now shows username only, no secret. CORRECTION 4: duplicate-release-commit defect ROOT CAUSE: convergence computed warranted bump from commits-since-tag without checking if master version already satisfies the bump. FIX: already-converged guard in version_convergence.py (master version > last tag → no commit). Repo-aware (args.repo passed to read_version/latest_tag). CORRECTION 6: lifecycle terminalization — execution exception now guarantees task terminalizes as failed with evidence (no immortal RUNNING). CORRECTION 7: route instruction task-scoped on RuneTask.route_instruction (NOT shared operator state). Read-only leading-verb override. Isolation tests prove no cross-contamination between interleaved tasks. PL1-01/02/03/05: pl1_records.py (route resolution, proof level, failure class, candidate status), metrics with honest NOT_YET_MEASURABLE semantics, route-instruction learning loop (PE-1789102658-0269 SEED). 1253 tests green; ratchet 352/80 (shrink); henxels clean.
222 lines
9.2 KiB
Python
222 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
|
"""stage_telemetry.py — StageAttempt telemetry (OBS-1, epic #284 / #287).
|
|
|
|
OBSERVATIONAL ONLY. This module records what happened; it must never change
|
|
stage authorization, retry behavior, model prompts, runner selection, or gate
|
|
outcomes. Timing code is not a process authority.
|
|
|
|
Design decisions (ratified in #287 tightening):
|
|
- EVERY attempt persists its own lifecycle record — successful, gate-failed,
|
|
failed, timed-out, interrupted. Dropping expensive failures would bias
|
|
ADAPT-1's baseline toward cheap success.
|
|
- Timing categories are OBSERVED, not forced additive: wall_time is not
|
|
model_time + tool_time when MiMo waits on a tool or relay overhead overlaps.
|
|
Records carry the raw timestamps so reports can distinguish exclusive /
|
|
overlapping / unmeasured-residual.
|
|
- Records survive restart: each attempt is appended to
|
|
<artifact_dir>/telemetry/attempts.jsonl (append-only, one JSON object per
|
|
line) AND mirrored into the task-tree state at persist time.
|
|
- UNKNOWN_NOT_EXPOSED is recorded verbatim when a counter (e.g. provider
|
|
tokens) is unavailable. Estimates are never presented as measurements.
|
|
|
|
Attempt lifecycle (from the executor's perspective):
|
|
begin(attempt_id, event_id, stage) → returns a handle
|
|
handle.mark_model_start/end() → model invocation windows
|
|
handle.mark_tool_start/end() → tool execution windows
|
|
handle.finish(terminal_status) → COMPLETED | GATE_FAILED | FAILED |
|
|
TIMEOUT | INTERRUPTED
|
|
persist(handle) → append + mirror
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import time
|
|
from datetime import datetime, timezone
|
|
|
|
TERMINAL_STATUSES = {"COMPLETED", "GATE_FAILED", "FAILED", "TIMEOUT", "INTERRUPTED"}
|
|
UNKNOWN = "UNKNOWN_NOT_EXPOSED"
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
class StageAttempt:
|
|
"""One attempt's observed lifecycle. Pure data — no control flow."""
|
|
|
|
def __init__(self, attempt_id: str, event_id: str, stage: str, issue: int, process_id: str):
|
|
self.attempt_id = attempt_id
|
|
self.event_id = event_id
|
|
self.stage = stage
|
|
self.issue = issue
|
|
self.process_id = process_id
|
|
self.started_at = _now_iso()
|
|
self._t0 = time.monotonic()
|
|
self.finished_at: str | None = None
|
|
self.terminal_status: str | None = None
|
|
# Observed windows (monotonic pairs, converted at finish)
|
|
self.model_windows: list[tuple[float, float]] = []
|
|
self.tool_windows: list[tuple[float, float]] = []
|
|
self.gate_windows: list[tuple[float, float]] = []
|
|
self.persistence_windows: list[tuple[float, float]] = []
|
|
# Counters
|
|
self.model_calls = 0
|
|
self.tool_calls = 0
|
|
self.repo_read_calls = 0
|
|
self.external_queries = 0
|
|
self.sources_discovered = 0
|
|
self.sources_read = 0
|
|
self.retries = 0
|
|
self.reconciliation_actions = 0
|
|
self.implementation_corrections = 0
|
|
self.ci_failures = 0
|
|
# Provider counters — recorded as UNKNOWN unless actually exposed
|
|
self.input_tokens: int | str = UNKNOWN
|
|
self.output_tokens: int | str = UNKNOWN
|
|
# Queue wait: measured by the caller if it knows the dispatch time
|
|
self.queue_wait_s: float | None = None
|
|
|
|
# -- window markers (observational; no control flow) ---------------------
|
|
|
|
def mark_model_start(self) -> None:
|
|
self.model_windows.append((time.monotonic(), 0.0))
|
|
|
|
def mark_model_end(self) -> None:
|
|
if self.model_windows and self.model_windows[-1][1] == 0.0:
|
|
s, _ = self.model_windows[-1]
|
|
self.model_windows[-1] = (s, time.monotonic())
|
|
self.model_calls += 1
|
|
|
|
def mark_tool_start(self) -> None:
|
|
self.tool_windows.append((time.monotonic(), 0.0))
|
|
|
|
def mark_tool_end(self) -> None:
|
|
if self.tool_windows and self.tool_windows[-1][1] == 0.0:
|
|
s, _ = self.tool_windows[-1]
|
|
self.tool_windows[-1] = (s, time.monotonic())
|
|
self.tool_calls += 1
|
|
|
|
def mark_gate_start(self) -> None:
|
|
self.gate_windows.append((time.monotonic(), 0.0))
|
|
|
|
def mark_gate_end(self) -> None:
|
|
if self.gate_windows and self.gate_windows[-1][1] == 0.0:
|
|
s, _ = self.gate_windows[-1]
|
|
self.gate_windows[-1] = (s, time.monotonic())
|
|
|
|
def mark_persistence_start(self) -> None:
|
|
self.persistence_windows.append((time.monotonic(), 0.0))
|
|
|
|
def mark_persistence_end(self) -> None:
|
|
if self.persistence_windows and self.persistence_windows[-1][1] == 0.0:
|
|
s, _ = self.persistence_windows[-1]
|
|
self.persistence_windows[-1] = (s, time.monotonic())
|
|
|
|
# -- terminal ------------------------------------------------------------
|
|
|
|
def finish(self, terminal_status: str) -> None:
|
|
assert terminal_status in TERMINAL_STATUSES, terminal_status
|
|
self.terminal_status = terminal_status
|
|
self.finished_at = _now_iso()
|
|
# Close any unclosed windows (INTERRUPTED/TIMEOUT paths)
|
|
for lst in (self.model_windows, self.tool_windows, self.gate_windows, self.persistence_windows):
|
|
if lst and lst[-1][1] == 0.0:
|
|
s, _ = lst[-1]
|
|
lst[-1] = (s, time.monotonic())
|
|
|
|
# -- derived (observed, explicitly non-additive) --------------------------
|
|
|
|
def _sum_windows(self, windows: list[tuple[float, float]]) -> float:
|
|
return round(sum(e - s for s, e in windows), 4)
|
|
|
|
def summary(self) -> dict:
|
|
# wall_time: monotonic delta while running; after finish, derived from
|
|
# the started/finished ISO pair is NOT monotonic-safe — keep the
|
|
# monotonic delta captured at finish time instead.
|
|
wall = round(time.monotonic() - self._t0, 4)
|
|
model_time = self._sum_windows(self.model_windows)
|
|
tool_time = self._sum_windows(self.tool_windows)
|
|
gate_time = self._sum_windows(self.gate_windows)
|
|
persistence_time = self._sum_windows(self.persistence_windows)
|
|
return {
|
|
"attempt_id": self.attempt_id,
|
|
"event_id": self.event_id,
|
|
"stage": self.stage,
|
|
"issue": self.issue,
|
|
"process_id": self.process_id,
|
|
"started_at": self.started_at,
|
|
"finished_at": self.finished_at,
|
|
"terminal_status": self.terminal_status,
|
|
"wall_time_s": wall,
|
|
# Observed categories — NOT asserted to sum to wall_time.
|
|
"model_time_s": model_time,
|
|
"tool_time_s": tool_time,
|
|
"gate_time_s": gate_time,
|
|
"persistence_time_s": persistence_time,
|
|
# Overlap analysis: tool time inside model windows is overlap,
|
|
# not double-counted work.
|
|
"overlap_note": (
|
|
"categories are observed windows and may overlap; "
|
|
"wall_time is authoritative, category sums are not additive"
|
|
),
|
|
"model_calls": self.model_calls,
|
|
"tool_calls": self.tool_calls,
|
|
"repo_read_calls": self.repo_read_calls,
|
|
"external_queries": self.external_queries,
|
|
"sources_discovered": self.sources_discovered,
|
|
"sources_read": self.sources_read,
|
|
"retries": self.retries,
|
|
"reconciliation_actions": self.reconciliation_actions,
|
|
"implementation_corrections": self.implementation_corrections,
|
|
"ci_failures": self.ci_failures,
|
|
"input_tokens": self.input_tokens,
|
|
"output_tokens": self.output_tokens,
|
|
"queue_wait_s": self.queue_wait_s,
|
|
"window_counts": {
|
|
"model": len(self.model_windows),
|
|
"tool": len(self.tool_windows),
|
|
"gate": len(self.gate_windows),
|
|
"persistence": len(self.persistence_windows),
|
|
},
|
|
}
|
|
|
|
|
|
class TelemetryLedger:
|
|
"""Append-only attempt ledger. Survives restart (JSONL on disk).
|
|
Appending is observational: a failed append must never break the stage."""
|
|
|
|
def __init__(self, artifact_dir: str):
|
|
self.dir = os.path.join(artifact_dir, "telemetry")
|
|
self.path = os.path.join(self.dir, "attempts.jsonl")
|
|
|
|
def persist(self, attempt: StageAttempt) -> bool:
|
|
"""Append one attempt record. Returns True on success; NEVER raises
|
|
into the stage path (observational invariant)."""
|
|
try:
|
|
os.makedirs(self.dir, exist_ok=True)
|
|
with open(self.path, "a") as f:
|
|
f.write(json.dumps(attempt.summary(), default=str) + "\n")
|
|
return True
|
|
except (OSError, TypeError, ValueError):
|
|
# Observational invariant: telemetry failures (unwritable path,
|
|
# serialization problems) are swallowed, never raised into the
|
|
# stage path.
|
|
return False
|
|
|
|
def load_all(self) -> list[dict]:
|
|
if not os.path.exists(self.path):
|
|
return []
|
|
out = []
|
|
with open(self.path) as f:
|
|
for line in f:
|
|
try:
|
|
out.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return out
|
|
|
|
def attempts_for(self, stage: str) -> list[dict]:
|
|
return [a for a in self.load_all() if a.get("stage") == stage]
|