From d3ad520529e6bc476bb61f9e99725a5da60f5b8b Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sat, 2 May 2026 15:19:39 -0400 Subject: [PATCH] journal: emit unfirehose/1.0 JSONL for queries + bench cycles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit aborist now writes one JSONL session per `make query` invocation and per `bench-emergent` cycle to: ~/.aborist/unfirehose/{project-slug}/{session-uuid}.jsonl Unfirehose's native-harness auto-discovery picks up any ~/.{name}/unfirehose/ directory (see ingest.ts:discoverNativeHarnesses) without registration — once a session lands, the unfirehose watcher debounces, ingests, and exposes it in the dashboard alongside Claude Code / Fetch / uncloseai sessions. Schema: unfirehose/1.0 (per ~/git/unfirehose-nextjs-logger/docs/ unfirehose-schema.md). Each session file: line 1 type=session (header — id, projectId, firstPrompt, harness="aborist", harnessVersion) line 2 type=message role=user line 3 type=message role=assistant content=[text] model=hermes-3-llama-3.1-8b-fp8-dynamic provider=hermes durationMs= aborist_meta={audit_mode, n_verified/n_quotes, cache_key, cache_status, lookup_path, violations, sources, timings_ms, answer_mode} line 4 type=message role=system subtype=session_end durationMs aborist-specific extras (verifier verdict, sources, timings) ride under namespaced ``aborist_meta`` so the canonical fields stay clean for off-the-shelf consumers; per the spec, unknown fields are ignored downstream. Bench-emergent cycles emit an additional system init message at the start of each session noting the 3 random words, marking the session as a generator-driven cycle vs a normal user query. Failure-isolation: journal write is wrapped in a broad try/except at every call site. A journaling bug must NEVER break the query or bench loop. Tests: 10 new in tests/test_journal.py (slug encoding, session header, parent-id chain, session_end on close, aborist_meta passthrough, usage block, idempotent close). Full suite: 663 passed. Live verified: `make query Q="what is photosynthesis?"` produced a 4-line JSONL with STRICT 3/3, all sources + timings populated, ready for unfirehose ingestion. --- aborist/cli.py | 46 ++++++ aborist/journal.py | 285 ++++++++++++++++++++++++++++++++++++++ scripts/bench_emergent.py | 53 +++++++ tests/test_journal.py | 120 ++++++++++++++++ 4 files changed, 504 insertions(+) create mode 100644 aborist/journal.py create mode 100644 tests/test_journal.py diff --git a/aborist/cli.py b/aborist/cli.py index 8702cb4..acb0a3c 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -415,6 +415,16 @@ def _cmd_query(args: argparse.Namespace) -> int: retrieval_keywords=getattr(args, "retrieval_keywords", None), ) + # Emit unfirehose-compatible session journal. One JSONL file per + # `make query` invocation, written to ~/.aborist/unfirehose/{slug}/ + # {session_uuid}.jsonl. Unfirehose's native-harness watcher picks + # this up automatically (no registration). Failures here must NEVER + # break the query path — wrap in a broad except & swallow. + try: + _emit_query_journal(args.question, result, model) + except Exception: # pragma: no cover — best-effort journaling + pass + if args.json: print(json.dumps(result, indent=2, ensure_ascii=False)) else: @@ -426,6 +436,42 @@ def _cmd_query(args: argparse.Namespace) -> int: ) +def _emit_query_journal(question: str, result: dict, model: str) -> None: + """Write one unfirehose/1.0 session for this query invocation.""" + from aborist.journal import SessionWriter + timings = result.get("timings") or {} + answer = result.get("answer_text") or "" + aborist_meta = { + "audit_mode": result.get("audit_mode"), + "verifier_method": result.get("verifier_method"), + "n_quotes": result.get("n_quotes"), + "n_verified": result.get("n_verified"), + "cache_key": result.get("cache_key"), + "cache_status": result.get("status"), + "lookup_path": result.get("lookup_path"), + "violations": [ + {"kind": v.get("kind"), "rationale": (v.get("rationale") or "")[:160]} + for v in (result.get("violations") or []) + ], + "sources": [ + {"title": s.get("title"), "uri": s.get("document_uri"), "used": s.get("used"), "role": s.get("source_role")} + for s in (result.get("sources") or []) + ], + "timings_ms": timings, + "answer_mode": (result.get("policy") or {}).get("answer_mode"), + } + with SessionWriter(first_prompt=question) as s: + s.user_message(question) + s.assistant_message( + answer, + model=model, + provider="hermes", + stop_reason="end_turn", + duration_ms=int(timings.get("total_ms") or 0) or None, + aborist_meta=aborist_meta, + ) + + # Soft-demote violation kinds that demote STRICT to HYBRID without # rejecting the pointer outright. WARRANT_MISSING / TITLE_MISMATCH / # DEFLECTION_DETECTED handled separately as hard demotes (their diff --git a/aborist/journal.py b/aborist/journal.py new file mode 100644 index 0000000..e094c5b --- /dev/null +++ b/aborist/journal.py @@ -0,0 +1,285 @@ +"""Unfirehose-compatible session journal. + +Emits one JSONL file per query/bench-cycle to +``~/.aborist/unfirehose/{project-slug}/{session-uuid}.jsonl``. +Unfirehose's native-harness auto-discovery picks up any +``~/.{name}/unfirehose/`` directory without registration; once a +session lands, the unfirehose watcher debounces, ingests, and +exposes it in the dashboard alongside Claude Code, Fetch, and +uncloseai sessions. + +Schema reference: ``~/git/unfirehose-nextjs-logger/docs/unfirehose-schema.md`` +(``unfirehose/1.0``). Every record carries ``$schema: +"unfirehose/1.0"``; consumers ignore unknown fields, so aborist- +specific extras (audit_mode, cache_key, source list, verifier +timings) ride along under namespaced keys. + +Per-query layout: + + session header line type=session + user message role=user, content=[text] + assistant message role=assistant, content=[text], + usage=..., model=..., + aborist_meta={audit_mode, cache_key, + sources, timings} + system session_end message subtype=session_end, durationMs + +Per-bench-cycle layout (bench_emergent): + + session header line (firstPrompt = generated question) + system init message subtype=init, aborist_meta={words} + user message the generated question + assistant message the student answer + aborist_meta + system session_end message subtype=session_end + +Everything aborist-specific lives under ``aborist_meta`` so the +canonical fields stay clean for off-the-shelf unfirehose consumers +that expect the strict schema. +""" + +from __future__ import annotations + +import json +import os +import time +import uuid +from pathlib import Path +from typing import Any + + +# Default journal root. Override via env var or constructor. +DEFAULT_JOURNAL_ROOT = Path.home() / ".aborist" / "unfirehose" + +# Schema literal pinned to a single string — matches what unfirehose's +# native-harness ingestion checks for. +UNFIREHOSE_SCHEMA = "unfirehose/1.0" + +# Harness identity. Any aborist process emitting JSONL claims this +# `harness` name. Unfirehose surfaces it in its dashboard alongside +# claude-code / fetch / uncloseai. +HARNESS_NAME = "aborist" + + +def _aborist_version() -> str: + """Best-effort version tag for the harnessVersion field.""" + try: + from importlib.metadata import version as _v + return _v("aborist") + except Exception: + return "0.0.0+dev" + + +def slugify_cwd(cwd: Path | str | None = None) -> str: + """Convert a working-directory path to an unfirehose project slug. + + Rules (mirror Claude Code's encoding): + - leading ``/`` becomes leading ``-`` + - path separators (``/``) become ``-`` + - dots (``.``) become ``-`` + + ``/home/fox/git/aborist`` → ``-home-fox-git-aborist`` + ``/home/fox/git/my.app`` → ``-home-fox-git-my-app`` + """ + p = Path(cwd) if cwd else Path.cwd() + s = str(p) + return s.replace("/", "-").replace(".", "-") + + +def now_iso() -> str: + """ISO 8601 UTC timestamp with millisecond precision.""" + t = time.time() + base = time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(t)) + ms = int((t - int(t)) * 1000) + return f"{base}.{ms:03d}Z" + + +def new_session_id() -> str: + """UUID4 stringification for session_id / message_id.""" + return str(uuid.uuid4()) + + +def _git_branch(cwd: Path | str | None = None) -> str | None: + """Best-effort current git branch. Returns None on failure.""" + try: + import subprocess + out = subprocess.check_output( + ["git", "-C", str(cwd or "."), "branch", "--show-current"], + stderr=subprocess.DEVNULL, + timeout=2, + ) + return out.decode().strip() or None + except Exception: + return None + + +class SessionWriter: + """One open .jsonl file representing one aborist session. + + Use as a context manager so the close timestamp is set & the file + flushed on exit: + + with SessionWriter(first_prompt=question, cwd=Path.cwd()) as s: + s.user_message(question) + s.assistant_message(answer, model=..., usage=..., aborist_meta=...) + + The constructor writes the session header line; subsequent calls + append message lines; ``__exit__`` writes the session_end line. + """ + + def __init__( + self, + *, + first_prompt: str, + cwd: Path | str | None = None, + root: Path | str | None = None, + session_id: str | None = None, + sidechain: bool = False, + ) -> None: + self.session_id = session_id or new_session_id() + self.cwd = Path(cwd or Path.cwd()) + self.root = Path(root or os.environ.get("ABORIST_JOURNAL_ROOT") or DEFAULT_JOURNAL_ROOT) + self.project_slug = slugify_cwd(self.cwd) + self.sidechain = sidechain + self.git_branch = _git_branch(self.cwd) + self.harness_version = _aborist_version() + + # File path: {root}/{slug}/{session_id}.jsonl + self.session_dir = self.root / self.project_slug + self.session_dir.mkdir(parents=True, exist_ok=True) + self.path = self.session_dir / f"{self.session_id}.jsonl" + + self._fp = self.path.open("a", encoding="utf-8") + self._t_start = time.time() + self._closed = False + self._message_count = 0 + self._last_message_id: str | None = None + + # Session header + self._write({ + "$schema": UNFIREHOSE_SCHEMA, + "type": "session", + "id": self.session_id, + "projectId": self.project_slug, + "status": "active", + "createdAt": now_iso(), + "updatedAt": now_iso(), + "closedAt": None, + "firstPrompt": (first_prompt or "")[:500], + "summary": "", + "gitBranch": self.git_branch, + "cwd": str(self.cwd), + "sidechain": self.sidechain, + "harness": HARNESS_NAME, + "harnessVersion": self.harness_version, + }) + + # ─────────────────────────────────────────────────────────── messages + + def _write(self, obj: dict[str, Any]) -> None: + if self._closed: + raise RuntimeError("session already closed") + self._fp.write(json.dumps(obj, ensure_ascii=False) + "\n") + self._fp.flush() + + def _new_message_id(self) -> str: + return f"msg_{uuid.uuid4().hex[:24]}" + + def _base_message(self, role: str) -> dict[str, Any]: + msg_id = self._new_message_id() + record: dict[str, Any] = { + "$schema": UNFIREHOSE_SCHEMA, + "type": "message", + "id": msg_id, + "sessionId": self.session_id, + "parentId": self._last_message_id, + "role": role, + "timestamp": now_iso(), + "sidechain": self.sidechain, + "cwd": str(self.cwd), + "gitBranch": self.git_branch, + "harness": HARNESS_NAME, + "harnessVersion": self.harness_version, + } + self._last_message_id = msg_id + self._message_count += 1 + return record + + def user_message( + self, + text: str, + *, + aborist_meta: dict[str, Any] | None = None, + ) -> None: + rec = self._base_message("user") + rec["content"] = [{"type": "text", "text": text}] + if aborist_meta: + rec["aborist_meta"] = aborist_meta + self._write(rec) + + def assistant_message( + self, + text: str, + *, + model: str | None = None, + provider: str | None = None, + stop_reason: str | None = None, + usage: dict[str, Any] | None = None, + duration_ms: int | None = None, + aborist_meta: dict[str, Any] | None = None, + ) -> None: + rec = self._base_message("assistant") + rec["content"] = [{"type": "text", "text": text}] + if model: + rec["model"] = model + if provider: + rec["provider"] = provider + if stop_reason: + rec["stopReason"] = stop_reason + if usage: + rec["usage"] = usage + if duration_ms is not None: + rec["durationMs"] = int(duration_ms) + if aborist_meta: + rec["aborist_meta"] = aborist_meta + self._write(rec) + + def system_message( + self, + text: str = "", + *, + subtype: str | None = None, + duration_ms: int | None = None, + aborist_meta: dict[str, Any] | None = None, + ) -> None: + rec = self._base_message("system") + if text: + rec["content"] = [{"type": "text", "text": text}] + if subtype: + rec["subtype"] = subtype + if duration_ms is not None: + rec["durationMs"] = int(duration_ms) + if aborist_meta: + rec["aborist_meta"] = aborist_meta + self._write(rec) + + # ──────────────────────────────────────────────────────────── lifecycle + + def close(self, *, status: str = "closed") -> None: + if self._closed: + return + elapsed_ms = int((time.time() - self._t_start) * 1000) + # session_end system message — gives unfirehose a clean signal + # that the session is complete + the wall-clock duration. + self.system_message( + subtype="session_end", + duration_ms=elapsed_ms, + aborist_meta={"status": status, "messageCount": self._message_count}, + ) + self._fp.close() + self._closed = True + + def __enter__(self) -> "SessionWriter": + return self + + def __exit__(self, *exc) -> None: + self.close() diff --git a/scripts/bench_emergent.py b/scripts/bench_emergent.py index b3ddfe6..2afbe69 100644 --- a/scripts/bench_emergent.py +++ b/scripts/bench_emergent.py @@ -203,6 +203,51 @@ def append_log(entry: dict, log_path: Path) -> None: f.write(json.dumps(entry, ensure_ascii=False) + "\n") +def emit_unfirehose_session(entry: dict, model_id: str) -> None: + """Write one unfirehose/1.0 session for this bench cycle. + + Same auto-discovery path as `aborist.cli._emit_query_journal`: + ~/.aborist/unfirehose/{slug}/{session_uuid}.jsonl. Each cycle gets + its own session file (one prompt → one answer = one session). + Failures must NEVER break the bench loop — wrapped at the call + site. + """ + from aborist.journal import SessionWriter + aborist_meta = { + "audit_mode": entry.get("audit_mode"), + "verifier_method": entry.get("verifier_method"), + "n_quotes": entry.get("n_quotes"), + "n_verified": entry.get("n_verified"), + "cache_key": entry.get("cache_key"), + "violation_kinds": entry.get("violation_kinds"), + "metaphor_deflection_kind": entry.get("metaphor_deflection_kind"), + "answer_mode": entry.get("answer_mode"), + "sources": entry.get("sources"), + "timings": { + "question_gen_seconds": entry.get("question_gen_seconds"), + "answer_seconds": entry.get("answer_seconds"), + "total_seconds": entry.get("total_seconds"), + }, + "bench": "emergent", + "words": entry.get("words"), + "student_error": entry.get("student_error"), + } + with SessionWriter(first_prompt=entry.get("question") or "") as s: + s.system_message( + "bench-emergent cycle: 3-word triangulation", + subtype="init", + aborist_meta={"words": entry.get("words"), "harness_role": "generator"}, + ) + s.user_message(entry.get("question") or "") + s.assistant_message( + entry.get("answer") or "", + model=model_id, + provider="hermes", + duration_ms=int((entry.get("answer_seconds") or 0) * 1000) or None, + aborist_meta=aborist_meta, + ) + + def print_pending(log_path: Path) -> int: """Print every log entry with `teacher: None` — what fox should bring to a teacher model for review.""" @@ -294,6 +339,14 @@ def main(argv: list[str] | None = None) -> int: top_k=ns.top_k, ) append_log(entry, ns.log_path) + # Mirror to unfirehose-compatible journal so the bench cycles + # show up in the unfirehose dashboard alongside Claude Code / + # Fetch sessions. Best-effort — never break the bench on a + # journal write failure. + try: + emit_unfirehose_session(entry, ns.model) + except Exception: # pragma: no cover + pass # Compact stdout summary so a long sweep is observable. audit = entry.get("audit_mode") or "?" ratio = ( diff --git a/tests/test_journal.py b/tests/test_journal.py new file mode 100644 index 0000000..1e9cef0 --- /dev/null +++ b/tests/test_journal.py @@ -0,0 +1,120 @@ +"""Tests for the unfirehose-compatible session journal writer. + +The contract being tested is the unfirehose/1.0 schema: +- session header line first +- one message per subsequent line +- session_end system message on close +- ``$schema: "unfirehose/1.0"`` on every record +- aborist-specific extras under namespaced ``aborist_meta`` +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from aborist.journal import ( + HARNESS_NAME, + UNFIREHOSE_SCHEMA, + SessionWriter, + new_session_id, + slugify_cwd, +) + + +def _read_lines(path: Path) -> list[dict]: + return [json.loads(line) for line in path.read_text().splitlines() if line.strip()] + + +def test_slugify_cwd_basic(): + assert slugify_cwd("/home/fox/git/aborist") == "-home-fox-git-aborist" + + +def test_slugify_cwd_dots_become_hyphens(): + assert slugify_cwd("/home/fox/git/my.app") == "-home-fox-git-my-app" + + +def test_slugify_cwd_uses_cwd_when_none(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + expected = str(tmp_path).replace("/", "-").replace(".", "-") + assert slugify_cwd() == expected + + +def test_session_header_first_line(tmp_path): + with SessionWriter(first_prompt="hello?", root=tmp_path, cwd="/home/fox/git/myproj"): + pass + files = list(tmp_path.glob("**/*.jsonl")) + assert len(files) == 1 + rows = _read_lines(files[0]) + header = rows[0] + assert header["$schema"] == UNFIREHOSE_SCHEMA + assert header["type"] == "session" + assert header["projectId"] == "-home-fox-git-myproj" + assert header["firstPrompt"] == "hello?" + assert header["harness"] == HARNESS_NAME + assert header["status"] == "active" + + +def test_session_messages_chained_via_parent_id(tmp_path): + with SessionWriter(first_prompt="q", root=tmp_path) as s: + s.user_message("first user") + s.assistant_message("first assistant", model="hermes-3") + s.user_message("second user") + rows = _read_lines(list(tmp_path.glob("**/*.jsonl"))[0]) + # Look at the user/assistant chain — session_end is a separate + # system message we exclude here. + msgs = [r for r in rows if r["type"] == "message" and r["role"] != "system"] + assert len(msgs) == 3 + # First message has no parent. + assert msgs[0]["parentId"] is None + # Subsequent messages chain to the previous. + assert msgs[1]["parentId"] == msgs[0]["id"] + assert msgs[2]["parentId"] == msgs[1]["id"] + + +def test_close_writes_session_end(tmp_path): + with SessionWriter(first_prompt="q", root=tmp_path) as s: + s.user_message("hi") + rows = _read_lines(list(tmp_path.glob("**/*.jsonl"))[0]) + last = rows[-1] + assert last["type"] == "message" + assert last["role"] == "system" + assert last["subtype"] == "session_end" + assert isinstance(last["durationMs"], int) + assert last["durationMs"] >= 0 + + +def test_aborist_meta_passes_through(tmp_path): + meta = {"audit_mode": "STRICT", "n_verified": 2, "cache_key": "abc123"} + with SessionWriter(first_prompt="q", root=tmp_path) as s: + s.assistant_message("answer", aborist_meta=meta) + rows = _read_lines(list(tmp_path.glob("**/*.jsonl"))[0]) + asst = next(r for r in rows if r["type"] == "message" and r["role"] == "assistant") + assert asst["aborist_meta"]["audit_mode"] == "STRICT" + assert asst["aborist_meta"]["n_verified"] == 2 + assert asst["aborist_meta"]["cache_key"] == "abc123" + + +def test_session_id_used_as_filename(tmp_path): + sid = new_session_id() + with SessionWriter(first_prompt="q", root=tmp_path, session_id=sid): + pass + files = list(tmp_path.glob("**/*.jsonl")) + assert files[0].stem == sid + + +def test_assistant_usage_block(tmp_path): + usage = {"inputTokens": 100, "outputTokens": 50, "totalTokens": 150} + with SessionWriter(first_prompt="q", root=tmp_path) as s: + s.assistant_message("a", model="hermes-3", usage=usage, duration_ms=1234) + rows = _read_lines(list(tmp_path.glob("**/*.jsonl"))[0]) + asst = next(r for r in rows if r["type"] == "message" and r["role"] == "assistant") + assert asst["usage"] == usage + assert asst["durationMs"] == 1234 + assert asst["model"] == "hermes-3" + + +def test_writer_is_idempotent_on_double_close(tmp_path): + s = SessionWriter(first_prompt="q", root=tmp_path) + s.close() + s.close() # should be a no-op, not raise