engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

journal: emit unfirehose/1.0 JSONL for queries + bench cycles

Browse source 
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=<wall>
                          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.
This commit is contained in:
russell@unturf.com 2026-05-02 15:19:39 -04:00
parent 529434ec67
commit d3ad520529
No known key found for this signature in database
4 changed files with 504 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

46
aborist/cli.py
View file

@ -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

285
aborist/journal.py Normal file
View file

@ -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()

53
scripts/bench_emergent.py
View file

@ -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 = (

120
tests/test_journal.py Normal file
View file

@ -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