From c501411ef5085873fcd21e5bbcae106a706345e0 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sat, 30 May 2026 17:36:58 -0400 Subject: [PATCH] cloud query: stage-level progress + capacity/timings tail MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirrors local 'arborist query' instrumentation so the cloud path is no less observable. Uses the existing arborist.qa.progress.Progress emitter (auto-on at TTY, override via ARBORIST_PROGRESS=0|1). Stages emitted: manifest.start/done URL + shard/sidecar counts corpus.open.start/done sidecar download + dict parse cost search.start/done per-shard FTS + RRF merge context.start/done chunk-content pulls + assembled bytes llm.start/done model + endpoint + ctx + answer chars verify.start/done audit_mode + n_verified/n_quotes Render tail adds (matches local query render): capacity: prompt N chars (sys N + evidence N + question N) → answer N chars timings: manifest Xs · corpus_open Xs · search Xs · context Xs · llm Xs · verify Xs · **total Xs** bucket: N HTTP requests · KB · endpoint / model JSON output (--json / JSON=1) gains 'timings' + 'capacity' fields with the same shape, so bench harnesses can consume them directly. Surfaces that sidecar.parse dominates a cold cloud-query (~22s/shard on the genesis 6M-term sidecar). That's a real future-opt target (ProcessPool instead of ThreadPool to escape the GIL) but the mechanism is observable now, which is the prerequisite for tuning. --- arborist/cli.py | 113 +++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 97 insertions(+), 16 deletions(-) diff --git a/arborist/cli.py b/arborist/cli.py index dcd9d5a..c93b8c5 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -7482,12 +7482,15 @@ def _cmd_cloud_snapshot_root(args: argparse.Namespace) -> int: def _cmd_cloud_query(args: argparse.Namespace) -> int: - """Bucket-direct end-to-end ask: FTS → fetch chunks → LLM (claim- + """Bucket-direct end-to-end query: FTS → fetch chunks → LLM (claim- lattice mode) → verify. Same audited-answer shape as local `query()` — EVIDENCE blocks with E1/E2 pointer tags, pointer-line answer format, four-rung audit ladder (POINTER-LINKED → ANCHOR-WARRANTED → EVIDENCE-WARRANTED) — but every byte goes through HttpRangeVFS from the bucket. + + Emits stage-level progress to stderr (mirrors `arborist query`), + auto-on at TTY, override via ARBORIST_PROGRESS=0/1. """ import os as _os import re as _re @@ -7500,12 +7503,16 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: render_claim_lattice, render_evidence_map, ) + from arborist.qa.progress import from_env as _progress_from_env from arborist.qa.prompts import ( CLAIM_LATTICE_GROUNDING_REMINDER, CLAIM_LATTICE_SYSTEM_PROMPT, ) from arborist.qa.verify import verify_claim_lattice + progress = _progress_from_env() + timings: dict[str, float] = {} + # Multi-shard via manifest, or single-shard if --shard-url given. if getattr(args, "bucket_url", None): from arborist.wallet.bucket import ( @@ -7513,16 +7520,22 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: MultiShardSidecarCorpus, load_bucket_manifest, ) - manifest_url = args.bucket_url - # Convenience: allow passing the manifest path directly. - if not manifest_url.rstrip("/").endswith(".json"): - manifest_url = manifest_url # base URL → load_bucket_manifest appends path - manifest = load_bucket_manifest(manifest_url) + progress.emit("manifest.start", url=args.bucket_url) + ts = _time.time() + manifest = load_bucket_manifest(args.bucket_url) + timings["manifest"] = _time.time() - ts + any_sidecar = any(sh.get("sidecar_url") for sh in manifest.shards) + progress.emit( + "manifest.done", shards=len(manifest.shards), + sidecars=sum(1 for sh in manifest.shards if sh.get("sidecar_url")), + ms=int(timings["manifest"] * 1000), + ) # Pick the right corpus class based on whether the manifest # advertises per-shard sidecars. Mixed (some sidecar, some not) # also uses the sidecar-aware path — it falls back to bucket- # direct FTS5 for shards without one. - any_sidecar = any(sh.get("sidecar_url") for sh in manifest.shards) + progress.emit("corpus.open.start", mode="sidecar" if any_sidecar else "bucket-direct") + ts = _time.time() if any_sidecar: client = MultiShardSidecarCorpus( manifest, cache_bytes_per_shard=args.cache_mb * 1024 * 1024, @@ -7531,6 +7544,8 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: client = MultiShardBucketCorpus( manifest, cache_bytes_per_shard=args.cache_mb * 1024 * 1024, ) + timings["corpus_open"] = _time.time() - ts + progress.emit("corpus.open.done", ms=int(timings["corpus_open"] * 1000)) multi_shard = True else: if not getattr(args, "shard_url", None): @@ -7544,7 +7559,11 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: multi_shard = False t0 = _time.time() try: + progress.emit("search.start", top_k=args.top_k) + ts = _time.time() hits = client.fts_search(args.question, limit=args.top_k) + timings["search"] = _time.time() - ts + progress.emit("search.done", hits=len(hits), ms=int(timings["search"] * 1000)) if not hits: print(json.dumps({ "answer_text": "", @@ -7563,6 +7582,11 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: # Pull the first body chunk of each hit via SQL through the # same VFS; reuses the already-warm page cache. Per-doc budget # keeps one huge article from monopolizing the context. + progress.emit( + "context.start", chunks=len(hits), + max_context_chars=args.max_context_chars, + ) + ts = _time.time() chunks_for_evidence: list[dict] = [] total_chars = 0 per_doc_budget = max( @@ -7606,6 +7630,12 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: if total_chars >= args.max_context_chars: break + timings["context"] = _time.time() - ts + progress.emit( + "context.done", sources=len(chunks_for_evidence), + assembled_chars=total_chars, ms=int(timings["context"] * 1000), + ) + if not chunks_for_evidence: print(json.dumps({ "answer_text": "", @@ -7628,24 +7658,33 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic", ) chat = OpenAICompatibleClient(base_url=endpoint) + sys_prompt = CLAIM_LATTICE_SYSTEM_PROMPT + user_prompt = ( + f"EVIDENCE:\n\n{evidence_text}\n\n" + f"QUESTION: {args.question}\n\n" + f"{CLAIM_LATTICE_GROUNDING_REMINDER}" + ) messages = [ - {"role": "system", "content": CLAIM_LATTICE_SYSTEM_PROMPT}, - { - "role": "user", - "content": ( - f"EVIDENCE:\n\n{evidence_text}\n\n" - f"QUESTION: {args.question}\n\n" - f"{CLAIM_LATTICE_GROUNDING_REMINDER}" - ), - }, + {"role": "system", "content": sys_prompt}, + {"role": "user", "content": user_prompt}, ] + prompt_chars = len(sys_prompt) + len(user_prompt) + progress.emit( + "llm.start", model=model, endpoint=endpoint, ctx_chars=prompt_chars, + ) try: + ts = _time.time() answer = chat.chat_completion( messages=messages, model=model, max_tokens=512, temperature=0.1, ) + timings["llm"] = _time.time() - ts + progress.emit( + "llm.done", answer_chars=len(answer), + ms=int(timings["llm"] * 1000), + ) except Exception as e: print(json.dumps({ "answer_text": "", @@ -7666,11 +7705,19 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: # output, resolves each [E#] back through the evidence map, # runs the textual-coverage check per claim. Warrant chain # disabled in cloud mode (no derivations table available). + progress.emit("verify.start", mode="claim_lattice") + ts = _time.time() verdict = verify_claim_lattice( answer, evidence_map, question=args.question, warrant_check_enabled=False, ) + timings["verify"] = _time.time() - ts + progress.emit( + "verify.done", audit_mode=verdict["audit_mode"], + n_verified=verdict["n_verified"], n_quotes=verdict["n_quotes"], + ms=int(timings["verify"] * 1000), + ) finally: client.close() @@ -7721,6 +7768,14 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: window=200, ) or answer # fall back to raw output if parser dropped everything + timings["total"] = elapsed + capacity = { + "sys_prompt_chars": len(sys_prompt), + "evidence_chars": len(evidence_text), + "question_chars": len(args.question), + "prompt_chars": prompt_chars, + "answer_chars": len(answer), + } result = { "answer_text": rendered_answer, "raw_answer": answer, @@ -7732,6 +7787,8 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int: "sources": sources_for_render, "stats": client.stats(), "timing_s": round(elapsed, 3), + "timings": {k: round(v, 3) for k, v in timings.items()}, + "capacity": capacity, "endpoint": endpoint, "model": model, } @@ -7804,6 +7861,30 @@ def _render_cloud_query_human(result: dict, question: str) -> str: ) lines.append("") + # Capacity tail — same shape as local query, scoped to the bucket- + # direct numbers we actually have (sys + evidence + question, no + # reminder block because the claim-lattice grounding reminder is + # folded into the user-prompt evidence section). + cap = result.get("capacity") or {} + if cap: + lines.append( + f"capacity: prompt {cap.get('prompt_chars', 0):,} chars " + f"(sys {cap.get('sys_prompt_chars', 0):,} " + f"+ evidence {cap.get('evidence_chars', 0):,} " + f"+ question {cap.get('question_chars', 0):,}) → " + f"answer {cap.get('answer_chars', 0):,} chars" + ) + + # Per-stage timings — mirrors the local query 'timings: cache X · ...' line. + t = result.get("timings") or {} + if t: + parts = [] + for k in ("manifest", "corpus_open", "search", "context", "llm", "verify"): + if k in t: + parts.append(f"{k} {t[k]:.2f}s") + parts.append(f"**total {t.get('total', elapsed_s):.2f}s**") + lines.append("timings: " + " · ".join(parts)) + bytes_kb = bytes_fetched / 1024.0 lines.append( f"bucket: {http_reqs} HTTP requests · {bytes_kb:.1f} KB · "