212 lines
10 KiB
Python
212 lines
10 KiB
Python
"""Incremental process telemetry and minute-labelled execution reports."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import shlex
|
|
from collections import defaultdict
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
LOCAL_ZONE = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def minute_label(value):
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(LOCAL_ZONE).strftime("%Y-%m-%d %H:%M")
|
|
except (ValueError, TypeError):
|
|
return value
|
|
|
|
|
|
def display_times(value):
|
|
if isinstance(value, list):
|
|
return [display_times(item) for item in value]
|
|
if not isinstance(value, dict):
|
|
return value
|
|
return {key: minute_label(item) if key.endswith("At") and isinstance(item, str) else display_times(item) for key, item in value.items()}
|
|
|
|
|
|
def command_phase(argv):
|
|
if not argv:
|
|
return "other"
|
|
if argv[0] == "codex":
|
|
return "model_request"
|
|
if argv[0] == "git":
|
|
return "git_" + (argv[1] if len(argv) > 1 else "other")
|
|
if any("task-context" in arg or "context-governance" in arg for arg in argv):
|
|
return "context_check"
|
|
if argv[0] in {"cmake", "ninja"}:
|
|
return "build"
|
|
return "command"
|
|
|
|
|
|
def item_phase(item):
|
|
kind = item.get("type", "unknown")
|
|
if kind == "file_change":
|
|
return "editing"
|
|
if kind != "command_execution":
|
|
return "tool" if kind != "agent_message" else "model_response"
|
|
try:
|
|
argv = shlex.split(item.get("command", ""))
|
|
if len(argv) >= 3 and argv[0].endswith(("bash", "sh")) and argv[1] in {"-lc", "-c"}:
|
|
argv = shlex.split(argv[2])
|
|
except ValueError:
|
|
return "command"
|
|
if not argv:
|
|
return "command"
|
|
if argv[0] in {"cat", "sed", "rg", "ls", "head", "tail", "wc", "find", "stat"} or any("search-wbv2" in arg or "print-task-context" in arg for arg in argv):
|
|
return "reading"
|
|
if "apply_patch" in argv[0]:
|
|
return "editing"
|
|
if any("test" in arg or "lint" in arg or "typecheck" in arg for arg in argv):
|
|
return "testing"
|
|
return command_phase(argv)
|
|
|
|
|
|
class WorkerTrace:
|
|
"""Times observed JSONL events; never infers model CPU time from silence."""
|
|
def __init__(self):
|
|
self.offset = 0
|
|
self.pending = b""
|
|
self.turn_started = False
|
|
self.actions_seen = False
|
|
self.unknown_output = False
|
|
self.first_event = None
|
|
self.first_action = None
|
|
self.last_progress = 0.0
|
|
self.last_output = 0.0
|
|
self.provider_error = None
|
|
self.errors = 0
|
|
self.active_items = {}
|
|
self.items = []
|
|
self.counts = defaultdict(int)
|
|
|
|
def feed(self, event, elapsed):
|
|
kind = event.get("type", "unknown")
|
|
if not isinstance(kind, str):
|
|
self.unknown_output = True
|
|
return
|
|
self.counts[kind] += 1
|
|
self.first_event = elapsed if self.first_event is None else self.first_event
|
|
if kind == "turn.started":
|
|
self.turn_started = True
|
|
if kind in {"error", "turn.failed"}:
|
|
self.errors += 1
|
|
error = event.get("error")
|
|
message = event.get("message") or (error.get("message", "") if isinstance(error, dict) else error) or ""
|
|
if not isinstance(message, str):
|
|
message = str(message)
|
|
if re.search(r"concurrency limit|overload|rate.?limit|too many requests|\b429\b|\b503\b|service unavailable", message, re.I):
|
|
self.provider_error = re.sub(r"sk-[A-Za-z0-9_-]+", "[redacted]", message)[:600]
|
|
if kind.startswith("item."):
|
|
self.actions_seen = True
|
|
self.first_action = elapsed if self.first_action is None else self.first_action
|
|
self.last_progress = elapsed
|
|
item = event.get("item") or {}
|
|
if not isinstance(item, dict):
|
|
self.unknown_output = True
|
|
return
|
|
key = item.get("id")
|
|
if kind == "item.started" and key:
|
|
self.active_items[key] = {"id": key, "phase": item_phase(item), "startedSeconds": elapsed}
|
|
elif kind == "item.completed":
|
|
record = self.active_items.pop(key, {"id": key, "phase": item_phase(item), "startedSeconds": elapsed})
|
|
record.update(finishedSeconds=elapsed, durationSeconds=max(0, elapsed - record["startedSeconds"]), status=item.get("status"), exitCode=item.get("exit_code"))
|
|
self.items.append(record)
|
|
if kind == "turn.completed":
|
|
self.last_progress = elapsed
|
|
|
|
def read(self, path, elapsed):
|
|
with path.open("rb") as stream:
|
|
stream.seek(self.offset)
|
|
chunk = stream.read()
|
|
self.offset = stream.tell()
|
|
if not chunk:
|
|
return
|
|
self.last_output = elapsed
|
|
lines = (self.pending + chunk).split(b"\n")
|
|
self.pending = lines.pop()
|
|
for line in lines:
|
|
if not line.strip():
|
|
continue
|
|
try:
|
|
event = json.loads(line)
|
|
if not isinstance(event, dict):
|
|
raise ValueError("non-object event")
|
|
self.feed(event, elapsed)
|
|
except (ValueError, UnicodeDecodeError):
|
|
self.unknown_output = True
|
|
|
|
@property
|
|
def safe_to_retry(self):
|
|
return self.turn_started and not self.actions_seen and not self.unknown_output and not self.pending.strip()
|
|
|
|
def snapshot(self, elapsed):
|
|
return {
|
|
"firstEventSeconds": self.first_event,
|
|
"firstActionSeconds": self.first_action,
|
|
"providerWaitSeconds": self.first_action if self.first_action is not None else elapsed,
|
|
"idleSeconds": max(0, elapsed - self.last_progress),
|
|
"actionsSeen": self.actions_seen,
|
|
"safeToRetry": self.safe_to_retry,
|
|
"providerError": self.provider_error,
|
|
"errorCount": self.errors,
|
|
"eventCounts": dict(self.counts),
|
|
"activeItems": list(self.active_items.values()),
|
|
"items": self.items,
|
|
}
|
|
|
|
|
|
def timing_report(directory, state=None):
|
|
rows, worker_details, seen, intervals = [], [], set(), []
|
|
path = directory / "events.jsonl"
|
|
if path.exists():
|
|
with path.open(encoding="utf-8") as stream:
|
|
for line in stream:
|
|
try:
|
|
event = json.loads(line)
|
|
except ValueError:
|
|
continue
|
|
if event.get("event") not in {"process.finished", "timing.span"}:
|
|
continue
|
|
key = (event.get("log"), event.get("startedAt"), event.get("phase"))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
phase = event.get("phase") or command_phase(event.get("argv", []))
|
|
seconds = max(0, event.get("durationSeconds", 0))
|
|
try:
|
|
intervals.append((datetime.fromisoformat(event["startedAt"]), datetime.fromisoformat(event.get("finishedAt") or event["time"])))
|
|
except (KeyError, ValueError):
|
|
pass
|
|
rows.append({"phase": phase, "start": minute_label(event.get("startedAt")), "finish": minute_label(event.get("finishedAt") or event.get("time")), "seconds": seconds, "minutes": round(seconds / 60, 4), "exitCode": event.get("exitCode"), "log": event.get("log"), "providerAbort": event.get("providerAbort")})
|
|
if event.get("workerTrace"):
|
|
worker_details.append({"log": event.get("log"), **event["workerTrace"]})
|
|
totals = defaultdict(lambda: {"count": 0, "seconds": 0.0, "maxSeconds": 0.0})
|
|
for row in rows:
|
|
total = totals[row["phase"]]
|
|
total["count"] += 1
|
|
total["seconds"] += row["seconds"]
|
|
total["maxSeconds"] = max(total["maxSeconds"], row["seconds"])
|
|
measured = sum(row["seconds"] for row in rows)
|
|
wall = (max(end for _, end in intervals) - min(start for start, _ in intervals)).total_seconds() if intervals else 0
|
|
summary = [{"phase": phase, **{k: round(v, 6) if isinstance(v, float) else v for k, v in total.items()}, "minutes": round(total["seconds"] / 60, 4), "sharePercent": round(total["seconds"] / measured * 100, 2) if measured else 0} for phase, total in totals.items()]
|
|
summary.sort(key=lambda row: row["seconds"], reverse=True)
|
|
current = dict((state or {}).get("currentProcess") or {})
|
|
if current.get("status") == "running" and current.get("startedAt"):
|
|
current["elapsedMinutes"] = round((datetime.now(timezone.utc) - datetime.fromisoformat(current["startedAt"])).total_seconds() / 60, 4)
|
|
return {"schemaVersion": 1, "observedAt": datetime.now(LOCAL_ZONE).strftime("%Y-%m-%d %H:%M"), "timezone": "Asia/Shanghai", "labelPrecision": "minute", "totalMeasuredMinutes": round(measured / 60, 4), "wallClockMinutes": round(wall / 60, 4), "outsideMeasuredStepsMinutes": round(max(0, wall - measured) / 60, 4), "summary": summary, "steps": rows, "workers": worker_details, "current": display_times(current), "accounting": "Top-level measured spans only; worker item durations are nested and must not be added again. Outside measured steps includes unrecorded retry, scheduling or manual pause gaps; legacy logs cannot attribute them further."}
|
|
|
|
|
|
def format_report(report):
|
|
lines = [f"Observed: {report['observedAt']} (Asia/Shanghai)", "Phase | Count | Minutes | Share", "--- | ---: | ---: | ---:"]
|
|
lines += [f"{r['phase']} | {r['count']} | {r['minutes']:.4f} | {r['sharePercent']:.2f}%" for r in report["summary"]]
|
|
lines += [f"Measured: {report['totalMeasuredMinutes']:.4f} minutes; outside measured steps: {report['outsideMeasuredStepsMinutes']:.4f} minutes."]
|
|
lines += ["", "Start | Finish | Phase | Minutes | Exit", "--- | --- | --- | ---: | ---:"]
|
|
lines += [f"{r['start']} | {r['finish']} | {r['phase']} | {r['minutes']:.4f} | {r['exitCode']}" for r in report["steps"]]
|
|
if report["current"]:
|
|
lines += ["", "Current: " + json.dumps(report["current"], ensure_ascii=False)]
|
|
return "\n".join(lines)
|