providence_query: emit stage progress to stderr (legacy parity)

Restores the per-phase stderr streaming that legacy query() had —
operator watches `arborist query` and sees which phase is currently
running, in real time, before the JSON / human render lands on
stdout. Fox flagged this regression after the CLI flip:
"it used to stream what arborist was doing to stderr before
emitting the json or quote."

Pattern: providence_query gains a `progress` kwarg (defaults to a
no-op shim so library callers without a Progress object don't
suddenly start writing to stderr). `_cmd_query` already builds a
Progress via _progress_from_env(); now threads it through.

Stages emitted, mirroring legacy query()'s vocabulary:
  search.start top_k=N
  search.done hits=N ms=N
  cache.lookup primary=<8-hex>
  cache.burn removed=N        (only when --burn)
  cache.hit lookup_path=primary ms=N
    -- OR --
  cache.miss lookup_path=primary
  llm.start model=<id> top_k=N
  llm.done audit_mode=X n_verified=N n_quotes=N ms=N
  persist.start
  persist.done ms=N

Operator gets visibility into which phase is paying the time;
silent ARBORIST_PROGRESS=0 still no-ops as before.

4 providence_query tests still pass.

arborist/cli.py

@@ -723,6 +723,7 @@ def _cmd_query(args: argparse.Namespace) -> int:
                 burn_existing=bool(getattr(args, "burn", False)),
                 top_k=args.top_k,
                 max_context_chars=mcc,
+                progress=progress,
             )
         finally:
             try:

arborist/qa/providence_query.py

@@ -281,6 +281,7 @@ def providence_query(
     max_context_chars: int = 24_000,
     temperature: float = 0.1,
     max_tokens: int = 512,
+    progress: Any = None,
 ) -> dict[str, Any]:
     """Cache-aware claim-lattice query.
 
@@ -300,12 +301,20 @@ def providence_query(
     t_total = _time.time()
     burned_existing = 0
 
+    # Disabled no-op shim when caller doesn't pass a progress object —
+    # keeps emit-or-not logic local to the call sites.
+    if progress is None:
+        class _NoProgress:
+            def emit(self, *a, **kw): pass
+        progress = _NoProgress()
+
     # 1. RETRIEVAL ONLY — fast phase. Run corpus.fts_body +
     # apply_title_boost to surface the top-K source docs. Do NOT
     # call the LLM yet. Building the cache_key needs to know the
     # context_root over retrieved doc roots; we get that here for
     # ~sub-second on local shards.
     from arborist.qa.corpus import NotSupportedError, apply_title_boost
+    progress.emit("search.start", top_k=top_k)
     t_ret = _time.time()
     try:
         hits = corpus.fts_body(question, limit=top_k * 4)
@@ -316,6 +325,10 @@ def providence_query(
         higher_is_better=getattr(corpus, "higher_is_better", False),
     )[:top_k]
     retrieval_s = _time.time() - t_ret
+    progress.emit(
+        "search.done", hits=len(hits),
+        ms=int(retrieval_s * 1000),
+    )
 
     # Provisional sources shape for cache_key + result. The richer
     # form (with used/used_pointer_ids) gets stamped on the miss
@@ -376,15 +389,22 @@ def providence_query(
     # 4. Cache lookup. Burn first if requested (so a re-lookup after
     # burn correctly misses and re-runs the LLM in the SAME call —
     # legacy convention).
+    progress.emit("cache.lookup", primary=ckey[:8])
     qa_conn = _qa_conn(qa_db)
     audit_event_hash = None
     run_dag_root = None
     try:
         if burn_existing:
             burned_existing = _burn(qa_conn, ckey)
+            progress.emit("cache.burn", removed=burned_existing)
+        t_lookup = _time.time()
         cached = _lookup(qa_conn, ckey)
         if cached is not None and not burn_existing:
             _bump_hit(qa_conn, ckey)
+            progress.emit(
+                "cache.hit", lookup_path="primary",
+                ms=int((_time.time() - t_lookup) * 1000),
+            )
             # Cache hit — return cached answer WITHOUT running the
             # LLM. retrieval already happened (we needed it for the
             # cache_key); skipping LLM is the real win.
@@ -431,6 +451,9 @@ def providence_query(
         # keeps the orchestrator's prompt/verify path single-sourced;
         # a future refactor can pass `precomputed_hits=hits` in to
         # eliminate the redundant retrieval.
+        progress.emit("cache.miss", lookup_path="primary")
+        progress.emit("llm.start", model=model_id, top_k=top_k)
+        t_llm = _time.time()
         run_result = run_query(
             corpus, question, chat_client,
             model_id=model_id, top_k=top_k,
@@ -438,9 +461,18 @@ def providence_query(
             temperature=temperature, max_tokens=max_tokens,
             policy=policy,
         )
+        progress.emit(
+            "llm.done",
+            audit_mode=run_result.get("audit_mode"),
+            n_verified=run_result.get("n_verified"),
+            n_quotes=run_result.get("n_quotes"),
+            ms=int((_time.time() - t_llm) * 1000),
+        )
         # Use the LLM-run sources (with used/used_pointer_ids) for
         # the persist path; fall back to provisional if absent.
         sources = run_result.get("sources") or sources
+        progress.emit("persist.start")
+        t_persist = _time.time()
         audit_event_hash, run_dag = _persist_miss(
             qa_conn,
             ckey=ckey, question=question, run_result=run_result,
@@ -451,6 +483,10 @@ def providence_query(
             chunking_version=chunking_version,
         )
         run_dag_root = run_dag["root"]
+        progress.emit(
+            "persist.done",
+            ms=int((_time.time() - t_persist) * 1000),
+        )
     finally:
         qa_conn.close()