Coverage for src/ai_jury/events.py: 98%

1"""Machine-readable progress events for a live jury run (``--events-file``).

3A side channel for observers that cannot sit in the jury process — a Claude Code mod,

4keel's progress view, a dashboard — so they can follow a run as it happens: the

5``on_event`` stream the orchestrator already fires for ``--live`` and ``--theater``,

6written as one JSON object per line (NDJSON).

8The records carry **metadata only**: who spoke, in which phase, whether it worked,

9how long it took and how many findings it raised. No reviewer output, diff text or

10finding body is written, so the file can be watched by another process without

11widening what the run exposes. Like ``--live``, it never touches the structured

12outcome, the report or the CI gate.

14Where the file goes: ``--events-file PATH`` names it for one run. Without the flag,

15``$JURY_EVENTS_DIR`` (when set, and not ``off``) gives every run its own file in that

16directory, ``<UTC stamp>-<pid>.ndjson``, so a watcher that knows the directory finds

17runs nobody passed a flag to — one an agent or keel started. The directory keeps the

18newest :data:`KEEP` run files. Unset, nothing is written: a run writes no file the

19operator did not ask for.

21The record builders and the naming are pure; :class:`EventsWriter` and

22:func:`open_dir_writer` are the thin I/O around them and the only places that read the

23wall clock, the process id or the directory.

24"""

26from __future__ import annotations

28import contextlib

29import json

30import os

31import re

32import time

33from collections.abc import Callable, Iterable

34from pathlib import Path

35from typing import Any, TextIO

37from .adapters import AgentResult

39#: Bumped when a record's shape changes in a way a reader has to know about.

40SCHEMA = "ai-jury.events.v1"

42#: Phases in the order a run moves through them; ``step`` records name one of these.

43PHASES = ("review", "debate", "verify", "synthesis")

45#: The environment variable naming the directory every run writes its events to.

46ENV_DIR = "JURY_EVENTS_DIR"

48#: Values of :data:`ENV_DIR` that mean "write nothing", as unset does.

49_OFF = frozenset({"", "off", "0", "false", "no"})

51#: How many run files the directory keeps; older ones go when a new run starts.

52KEEP = 20

54#: How many names a run tries before it gives up on writing events.

55_ATTEMPTS = 5

57#: The run files this module names, and the only files it ever deletes: a directory

58#: shared with anything else loses nothing else.

59_RUN_NAME = re.compile(r"^\d{8}T\d{6}\.\d{3}Z-\d+(?:-\d+)?\.ndjson$")

62def events_dir(value: str | None) -> Path | None:

63 """The directory ``$JURY_EVENTS_DIR`` names, or ``None`` when it asks for none."""

64 if value is None or value.strip().lower() in _OFF:

65 return None

66 return Path(value.strip()).expanduser()

69#: The file a watcher (the jury-progress mod) keeps in ``<cache dir>/events`` to ask for

70#: events without an environment variable, which a Claude Code mod cannot hand to the

71#: commands its session runs. It reads ``on`` while someone watches, ``off`` after.

72WATCH_MARKER = ".watched"

75def marker_says_on(text: str) -> bool:

76 """Whether a watch marker's content asks for events: its first word is ``on``."""

77 words = text.split()

78 return bool(words) and words[0].lower() == "on"

81def watched_dir(directory: Path) -> Path | None:

82 """``directory`` when its :data:`WATCH_MARKER` reads ``on``, else ``None``.

84 A marker that is missing, unreadable, too large to be one, or says anything else

85 asks for nothing: no events are written unless a watcher said so.

86 """

87 marker = directory / WATCH_MARKER

88 try:

89 if not marker.is_file() or marker.stat().st_size > 64:

90 return None

91 return directory if marker_says_on(marker.read_text(encoding="utf-8")) else None

92 except (OSError, UnicodeDecodeError):

93 return None

96def run_file_name(now: float, pid: int, attempt: int = 0) -> str:

97 """``<UTC stamp to the ms>-<pid>.ndjson``: sorts by start time.

99 ``attempt`` > 0 adds ``-<attempt>``, for a name another run of the same process

100 already took in the same millisecond (the file is created exclusively).

101 """

102 stamp = time.strftime("%Y%m%dT%H%M%S", time.gmtime(now))

103 millis = int(now * 1000) % 1000

104 suffix = f"-{attempt}" if attempt else ""

105 return f"{stamp}.{millis:03d}Z-{pid}{suffix}.ndjson"

106

107

108def _run_order(name: str) -> tuple[str, int]:

109 """Sort key: a retry name (``...-<pid>-1.ndjson``) is newer than its base name.

110

111 Plain string order puts ``-`` before ``.``, so a retry would sort as older.

112 """

113 stamp, _, rest = name[: -len(".ndjson")].partition("Z-")

114 pid, _, attempt = rest.partition("-")

115 return (f"{stamp}Z-{pid}", int(attempt) if attempt else 0)

116

117

118def stale_run_files(names: Iterable[str], keep: int = KEEP) -> list[str]:

119 """The run files past the newest ``keep``, oldest first; other names are never listed."""

120 runs = sorted((n for n in names if _RUN_NAME.match(n)), key=_run_order)

121 return runs[: max(0, len(runs) - keep)]

122

123

124def start_record(

125 agents: Iterable[tuple[str, ...]],

126 *,

127 chair: str | None,

128 target: str,

129 mode: str,

130 decision: str,

131 cached: bool,

132 pid: int | None = None,

133 cwd: str | None = None,

134) -> dict[str, Any]:

135 """The first record: the seated panel, so an observer can draw pending seats.

136

137 Each seat is ``(name, vendor)`` or ``(name, vendor, model)``; ``model`` is the

138 id the seat will be asked for, or ``None`` when its CLI chooses. ``pid`` lets a

139 watcher tell a run still going from one that died without an ``end`` record;

140 ``cwd`` tells it which checkout the run reviews from.

141 """

142 return {

143 "event": "start",

144 "panel": [

145 {

146 "agent": seat[0],

147 "vendor": seat[1],

148 "model": (seat[2] or None) if len(seat) > 2 else None,

149 }

150 for seat in agents

151 ],

152 "chair": chair,

153 "target": target,

154 "mode": mode,

155 "decision": decision,

156 "cached": cached,

157 "phases": list(PHASES),

158 "pid": pid,

159 "cwd": cwd,

160 }

161

162

163def step_record(kind: str, result: AgentResult, round_no: int | None = None) -> dict[str, Any]:

164 """One phase result, as the orchestrator's ``on_event`` reports it."""

165 return {

166 "event": "step",

167 "phase": kind,

168 "round": round_no,

169 "agent": result.agent,

170 "vendor": result.vendor,

171 "ok": result.ok,

172 "duration_s": round(result.duration_s, 1),

173 "findings": len(result.findings),

174 "error_code": result.error_code,

175 # The id this invocation sent (None when the CLI chose), and what it raised by

176 # severity: counts only, never a finding's text.

177 "model": getattr(result, "model", "") or None,

178 "severity": severity_counts(result.findings),

179 }

180

181

182def severity_counts(findings: Iterable[Any]) -> dict[str, int]:

183 """``{severity: count}`` over findings, in no particular order; empty for none."""

184 counts: dict[str, int] = {}

185 for f in findings:

186 sev = getattr(f, "severity", None)

187 if isinstance(sev, str) and sev:

188 counts[sev] = counts.get(sev, 0) + 1

189 return counts

190

191

192def clean_verdict(verdict: Any) -> Any:

193 """A verdict token without the markdown the chair wrapped it in (``**COMMENT**``).

194

195 Strips surrounding ``*``, ``_``, backticks and whitespace; anything that is not

196 a string, or that strips to nothing, is returned as it came.

197 """

198 if not isinstance(verdict, str):

199 return verdict

200 return verdict.strip().strip("*_` \t\r\n").strip() or verdict

201

202

203def ballot_entries(reviewers: Iterable[dict[str, Any]]) -> list[dict[str, Any]]:

204 """The panelists of a report's ``reviewers`` array, cut to what a watcher shows.

205

206 ``agent``, ``model``, ``verdict``, ``findings`` (how many) and ``review``

207 (whether the ballot counts as a review): no scope or testing prose, so the

208 record stays metadata. The chair's consensus entry is left out; the run's

209 verdict is the ``end`` record's own.

210 """

211 out: list[dict[str, Any]] = []

212 for r in reviewers:

213 if r.get("role") == "chair":

214 continue

215 found = r.get("findings")

216 out.append(

217 {

218 "agent": r.get("name"),

219 "model": r.get("model") or None,

220 "verdict": clean_verdict(r.get("verdict")),

221 "findings": len(found) if isinstance(found, list) else 0,

222 "review": bool(r.get("counts_as_review")),

223 }

224 )

225 return out

226

227

228def end_record(

229 status: str,

230 *,

231 findings: int | None = None,

232 verdict: str | None = None,

233 ballots: list[dict[str, Any]] | None = None,

234) -> dict[str, Any]:

235 """The last record. ``status`` is ``done``, ``cancelled`` or ``error``.

236

237 ``ballots`` (a done run's, from :func:`ballot_entries`) says how each seat voted.

238 """

239 rec: dict[str, Any] = {

240 "event": "end",

241 "status": status,

242 "findings": findings,

243 "verdict": clean_verdict(verdict),

244 }

245 if ballots is not None:

246 rec["ballots"] = ballots

247 return rec

248

249

250class EventsWriter:

251 """Writes records to ``path`` as NDJSON, one flushed line each.

252

253 The file is truncated when the writer opens, so it always describes one run.

254 Each record gets the schema, a sequence number and a timestamp in seconds.

255

256 Opening is the caller's to fail on (an unwritable path is a user error, reported

257 before the run). A write that fails mid-run must not take the review down with it:

258 the file is a side channel someone is watching, not the outcome. The first failure

259 is reported once through ``on_error`` and the writer goes quiet; the run, its report

260 and its CI gate carry on.

261 """

262

263 def __init__(

264 self,

265 path: str,

266 *,

267 clock: Callable[[], float] = time.time,

268 on_error: Callable[[str], None] | None = None,

269 exclusive: bool = False,

270 source: str = "--events-file",

271 ) -> None:

272 self._path = path

273 self._source = source

274 self._fh: TextIO | None

275 if exclusive:

276 # A file of the run's own in a directory others may share: created, never

277 # reused, and never through a symlink someone placed under its name.

278 flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0)

279 # open() owns the descriptor from os.open on: it closes it itself if the

280 # text layer fails, so nothing here may close it a second time.

281 self._fh = open( # noqa: SIM115 - closed in close()

282 path, "w", encoding="utf-8", opener=lambda p, _f: os.open(p, flags, 0o600)

283 )

284 else:

285 self._fh = Path(path).open("w", encoding="utf-8") # noqa: SIM115 - closed in close()

286 self._clock = clock

287 self._seq = 0

288 self._on_error = on_error

289

290 @property

291 def failed(self) -> bool:

292 """Whether a write failed and the writer stopped."""

293 return self._fh is None

294

295 def write(self, record: dict[str, Any]) -> None:

296 if self._fh is None:

297 return

298 self._seq += 1

299 line = {"schema": SCHEMA, "seq": self._seq, "ts": round(self._clock(), 3), **record}

300 try:

301 self._fh.write(json.dumps(line, sort_keys=True) + "\n")

302 self._fh.flush()

303 except OSError as exc:

304 self._stop(exc)

305

306 def close(self) -> None:

307 if self._fh is None:

308 return

309 try:

310 self._fh.close()

311 except OSError as exc:

312 self._fh = None

313 self._report(exc)

314 else:

315 self._fh = None

316

317 def _stop(self, exc: OSError) -> None:

318 fh, self._fh = self._fh, None

319 with contextlib.suppress(OSError):

320 if fh is not None: 320 ↛ 322line 320 didn't jump to line 322

321 fh.close()

322 self._report(exc)

323

324 def _report(self, exc: OSError) -> None:

325 if self._on_error is not None:

326 self._on_error(

327 f"{self._source} {self._path}: {exc}; progress events stopped, the review continues"

328 )

329

330

331def open_dir_writer(

332 directory: Path,

333 *,

334 clock: Callable[[], float] = time.time,

335 pid: int | None = None,

336 on_error: Callable[[str], None] | None = None,

337) -> EventsWriter | None:

338 """A writer for this run's own file in ``directory``, after pruning old runs.

339

340 Nobody passed a flag for this file, so nothing about it may fail the run: a

341 directory that cannot be made or written is reported once through ``on_error``

342 and the run goes on without events. A run file that cannot be pruned is left.

343 """

344 pid = os.getpid() if pid is None else pid

345 try:

346 existed = directory.is_dir()

347 directory.mkdir(mode=0o700, parents=True, exist_ok=True)

348 if (

349 existed

350 and on_error is not None

351 and os.name == "posix"

352 and directory.stat().st_mode & 0o022

353 ):

354 # 0700 only applies to a directory jury creates; say so once when one that

355 # already exists lets others write to it (the files stay 0600, O_EXCL).

356 on_error(f"${ENV_DIR} {directory} is group/world-writable; its mode was left as is")

357 # Room for this run's own file: the directory holds KEEP runs with it. Pruning

358 # goes by name, so with KEEP runs going at once the oldest live one's file can

359 # go too; that run keeps writing to its open handle, only a watcher loses it.

360 for name in stale_run_files((p.name for p in directory.iterdir()), KEEP - 1):

361 with contextlib.suppress(OSError):

362 (directory / name).unlink()

363 now = clock()

364 for attempt in range(_ATTEMPTS): 364 ↛ 375line 364 didn't jump to line 375 because the loop on line 364 didn't complete

365 try:

366 return EventsWriter(

367 str(directory / run_file_name(now, pid, attempt)),

368 clock=clock,

369 on_error=on_error,

370 exclusive=True,

371 source=f"${ENV_DIR}",

372 )

373 except FileExistsError:

374 continue

375 raise FileExistsError(f"{_ATTEMPTS} run file names for pid {pid} are taken")

376 except OSError as exc:

377 if on_error is not None: 377 ↛ 379line 377 didn't jump to line 379 because the condition on line 377 was always true

378 on_error(f"${ENV_DIR} {directory}: {exc}; this run writes no progress events")

379 return None