torben
18ae28f918
Chemenu kompiliert Rohnotizen zu einem verlinkten, quellengebundenen Wiki: raw/ -> types/ + tools/ -> kb/ -> reports/. Was mechanisch ist, macht tools/wikitool; was Urteil braucht, macht ein Agent unter Contracts, deren Grenzen in Code durchgesetzt sind statt im Prompt. Dieser Commit ist der Startpunkt der oeffentlichen Historie. Die vorherige Entwicklung fand in einer privaten Instanz statt und ist nicht Teil dieses Repositorys; ihre Erzaehlung steht vollstaendig in CHANGES.md, das mit 44 Eintraegen von 0.1.0 bis 2.1.0 erhalten geblieben ist. Der mitgelieferte Korpus ist ein Testbett und eine Demo: 170 Seiten ueber den Stack selbst - Gates, Lint, Versionierung, Suche, das Wiki-Muster. Er dokumentiert das Werkzeug mit den eigenen Mitteln des Werkzeugs. Lizenz: AGPL-3.0 fuer den Stack (tools/, types/), CC-BY-4.0 fuer die Inhalte. Die Grenze zwischen beiden ist der Dateiplan, den dist export berechnet - siehe NOTICE.
70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""Reading a trace back.
|
|
|
|
The format's owner owns the reader: a consumer that re-derived the sort order or
|
|
the session-directory rule would drift from the writer the first time either
|
|
changed.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
from chemenu.telemetry.writer import trace_path
|
|
|
|
|
|
def sort_key(record: dict) -> tuple:
|
|
"""`seq` counts within one process only - a trace is written by the CLI in
|
|
one process and by a hook handler in another, so it orders by time first."""
|
|
return (record.get("ts", ""), record.get("pid", 0), record.get("seq", 0))
|
|
|
|
|
|
def read_trace(session: str | None = None, path: Path | None = None) -> list[dict]:
|
|
"""Return one session's events in order. A missing trace is an empty one:
|
|
a session that never recorded anything is a normal state, not an error."""
|
|
target = path or trace_path(session)
|
|
if not target.exists():
|
|
return []
|
|
records = []
|
|
for line in target.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
record = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
# A torn last line can only happen if a writer died mid-append.
|
|
# Losing it is better than refusing to read the rest.
|
|
continue
|
|
if isinstance(record, dict):
|
|
records.append(record)
|
|
records.sort(key=sort_key)
|
|
return records
|
|
|
|
|
|
def sessions(root: Path | None = None) -> list[str]:
|
|
"""Every session that has a trace, most recently modified first."""
|
|
from chemenu.telemetry.writer import trace_root
|
|
|
|
base = root or trace_root()
|
|
if not base.exists():
|
|
return []
|
|
traces = [p for p in base.glob("*/trace.jsonl") if p.is_file()]
|
|
traces.sort(key=lambda p: p.stat().st_mtime, reverse=True)
|
|
return [p.parent.name for p in traces]
|
|
|
|
|
|
def completeness(records: list[dict]) -> list[str]:
|
|
"""What the harnesses behind this trace said they could report.
|
|
|
|
The union across sources, because a session traced by both `wikitool` and a
|
|
hook adapter can report what either of them can.
|
|
"""
|
|
seen: list[str] = []
|
|
for record in records:
|
|
if record.get("event") != "session.start":
|
|
continue
|
|
for item in record.get("attrs", {}).get("completeness", []):
|
|
if item not in seen:
|
|
seen.append(item)
|
|
return seen
|