session+cloud_query: emit unfirehose JSONL

arborist/cli.py

@@ -7800,6 +7800,15 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
         "/quit                  exit"
     )
 
+    # Unfirehose journal — one JSONL session per REPL invocation, lazy-
+    # opened on the first real (non-/ command) question so firstPrompt
+    # is the actual question instead of an empty placeholder. Best-
+    # effort: any failure inside the journal path swallows so the REPL
+    # itself never breaks. The arborist session sid + per-turn bates +
+    # parent_bates ride along under arborist_meta so unfirehose ingest
+    # can cross-reference back into sessions.db.
+    _unfirehose_writer = None
+
     try:
         while True:
             try:
@@ -7975,7 +7984,66 @@ def _session_repl(args: argparse.Namespace, sess) -> int:
                 f"\n# root       {sess.session_root}"
                 f"\n# cache_key  {ckey}"
             )
+            # Unfirehose emit — same shape as _emit_query_journal but
+            # one writer reused across the REPL so the unfirehose
+            # dashboard sees the full conversation as one session
+            # instead of N independent ones. Best-effort: never break
+            # the REPL on a journal failure.
+            try:
+                if _unfirehose_writer is None:
+                    from arborist.journal import SessionWriter
+                    _unfirehose_writer = SessionWriter(
+                        first_prompt=line,
+                    )
+                timings = result.get("timings") or {}
+                arborist_meta = {
+                    "session_sid": sess.sid,
+                    "bates": n.bates,
+                    "parent_bates": n.parent_bates,
+                    "session_root": sess.session_root,
+                    "audit_mode": audit_mode,
+                    "verifier_method": result.get("verifier_method"),
+                    "n_quotes": result.get("n_quotes"),
+                    "n_verified": result.get("n_verified"),
+                    "cache_key": ckey,
+                    "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": policy.get("answer_mode"),
+                }
+                _unfirehose_writer.user_message(line)
+                _unfirehose_writer.assistant_message(
+                    ans,
+                    model=model,
+                    provider="hermes",
+                    stop_reason="end_turn",
+                    duration_ms=int(timings.get("total_ms") or 0) or None,
+                    arborist_meta=arborist_meta,
+                )
+            except Exception:  # pragma: no cover — best-effort journaling
+                pass
     finally:
+        try:
+            if _unfirehose_writer is not None:
+                _unfirehose_writer.close()
+        except Exception:
+            pass
         try:
             _closer()
         except Exception:
@@ -8382,6 +8450,14 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
         try: closer()
         except Exception: pass
 
+    # Unfirehose journal — same pattern as local `arborist query` so
+    # bucket-direct sessions surface in the dashboard next to local
+    # ones. Best-effort: a journal failure never breaks the query path.
+    try:
+        _emit_query_journal(args.question, result, model)
+    except Exception:  # pragma: no cover — best-effort journaling
+        pass
+
     if getattr(args, "json", False):
         print(json.dumps(result, indent=2, ensure_ascii=False))
     else: