diff --git a/Makefile b/Makefile
index 31eab8d..594364f 100644
--- a/Makefile
+++ b/Makefile
@@ -1572,6 +1572,14 @@ cloud-fetch-chunk: bootstrap ## fetch + hash-verify one chunk from a bucket [LEA
 	@test -n "$(BLOB_BASE)" || { echo 'BLOB_BASE required (per-chunk blobs/<hash> base URL)'; exit 2; }
 	@$(ARBORIST) cloud fetch-chunk "$(LEAF_HASH)" --blob-base "$(BLOB_BASE)"
 
+cloud-ask: bootstrap ## bucket-direct end-to-end: FTS → LLM → verify [Q="..." SHARD_URL=... TOP_K=N MAX_CONTEXT=N CACHE_MB=N]
+	@test -n "$(Q)" || { echo 'usage: make cloud-ask Q="your question" [SHARD_URL=...] [TOP_K=4] [MAX_CONTEXT=24000] [CACHE_MB=64]'; exit 2; }
+	@echo "# shard: $(SHARD_URL)" >&2
+	@$(ARBORIST) cloud ask '$(Q)' --shard-url "$(SHARD_URL)" \
+	    --top-k $(or $(TOP_K),4) \
+	    --max-context-chars $(or $(MAX_CONTEXT),24000) \
+	    --cache-mb $(or $(CACHE_MB),64)
+
 CLOUD_DEMO_PORT ?= 18785
 
 cloud-demo: bootstrap ## bucket-direct end-to-end proof on tiny in-process corpus [CLOUD_DEMO_PORT=N]
diff --git a/arborist/cli.py b/arborist/cli.py
index 3a8271a..5af5ce2 100644
--- a/arborist/cli.py
+++ b/arborist/cli.py
@@ -7264,6 +7264,34 @@ def build_parser() -> argparse.ArgumentParser:
     cloud_fetch.add_argument("--blob-base", required=True)
     cloud_fetch.set_defaults(func=_cmd_cloud_fetch_chunk)
 
+    cloud_ask = cloud_sub.add_parser(
+        "ask",
+        help=(
+            "bucket-direct end-to-end ask: FTS → pull chunks via SQL → "
+            "LLM call → verify quotes locally. Same audited-answer shape "
+            "as `arborist query` but no local DB."
+        ),
+    )
+    cloud_ask.add_argument("question", type=str)
+    cloud_ask.add_argument("--shard-url", required=True)
+    cloud_ask.add_argument("--top-k", type=int, default=4)
+    cloud_ask.add_argument("--max-context-chars", type=int, default=24_000)
+    cloud_ask.add_argument(
+        "--cache-mb", type=int, default=64,
+        help="LRU page cache size in MB (default 64 — bigger than search "
+        "default; chunk-content reads dominate).",
+    )
+    cloud_ask.add_argument(
+        "--endpoint", default=None,
+        help="LLM endpoint (default: $ARBORIST_LLM_ENDPOINT or "
+        "https://hermes.ai.unturf.com/v1).",
+    )
+    cloud_ask.add_argument(
+        "--model", default=None,
+        help="LLM model id (default: $ARBORIST_LLM_MODEL or Hermes-3-8B).",
+    )
+    cloud_ask.set_defaults(func=_cmd_cloud_ask)
+
     return p
 
 
@@ -7406,6 +7434,138 @@ def _cmd_cloud_snapshot_root(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_cloud_ask(args: argparse.Namespace) -> int:
+    """Bucket-direct end-to-end ask: FTS → fetch chunks → LLM → verify.
+
+    Mirrors the local `query()` result shape (answer_text, audit_mode,
+    sources w/ source_role, n_quotes, n_verified, verifier_method) but
+    every byte read comes from the bucket via HttpRangeVFS. No local DB.
+    """
+    import os as _os
+    import time as _time
+    from arborist.compress import unpack_chunk
+    from arborist.qa.client import OpenAICompatibleClient
+    from arborist.qa.verify import verify_quotes
+
+    client = _make_bucket_client(args)
+    t0 = _time.time()
+    try:
+        hits = client.fts_search(args.question, limit=args.top_k)
+        if not hits:
+            print(json.dumps({
+                "answer_text": "",
+                "audit_mode": "UNGROUNDED",
+                "verifier_method": "none",
+                "n_quotes": 0,
+                "n_verified": 0,
+                "sources": [],
+                "stats": client.stats(),
+                "timing_s": round(_time.time() - t0, 3),
+                "note": "FTS5 returned no hits — sanitized query may have "
+                "lost too many tokens; try `cloud search` first.",
+            }, indent=2, ensure_ascii=False))
+            return 0
+
+        # Pull body content for each hit via SQL through the same VFS;
+        # chunks.content lives in the .db file, so this reuses the
+        # already-warm page cache. Cap total context bytes.
+        sources: list[dict] = []
+        context_parts: list[str] = []
+        total_chars = 0
+        for rank, h in enumerate(hits, 1):
+            rows = list(client.conn.execute(
+                "SELECT idx, content FROM chunks "
+                "WHERE document_root = ? AND content IS NOT NULL "
+                "ORDER BY idx ASC",
+                (h["document_root"],),
+            ))
+            body = "\n\n".join(unpack_chunk(r[1]) or "" for r in rows).strip()
+            if not body:
+                continue
+            # Truncate per-doc so one huge article doesn't monopolize the budget.
+            per_doc_budget = max(
+                1000, args.max_context_chars // max(1, len(hits))
+            )
+            if len(body) > per_doc_budget:
+                body = body[:per_doc_budget] + "…"
+            sources.append({
+                "document_root": h["document_root"],
+                "document_uri": h["document_uri"],
+                "title": h["title"],
+                "score": h["score"],
+                "source_role": (
+                    "primary_answer_source" if rank == 1 else "background_source"
+                ),
+                "n_chunks": len(rows),
+            })
+            context_parts.append(f"## [{rank}] {h['title']}\n{h['document_uri']}\n\n{body}")
+            total_chars += len(body)
+            if total_chars >= args.max_context_chars:
+                break
+
+        if not sources:
+            print(json.dumps({
+                "answer_text": "",
+                "audit_mode": "UNGROUNDED",
+                "sources": hits,
+                "note": "FTS hit found, but document had no readable chunk content",
+            }, indent=2, ensure_ascii=False))
+            return 0
+
+        context = "\n\n".join(context_parts)
+        endpoint = args.endpoint or _os.environ.get(
+            "ARBORIST_LLM_ENDPOINT", "https://hermes.ai.unturf.com/v1",
+        )
+        model = args.model or _os.environ.get(
+            "ARBORIST_LLM_MODEL",
+            "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
+        )
+        chat = OpenAICompatibleClient(base_url=endpoint)
+        prompt = (
+            "Answer the user's question using only the provided context. "
+            "When you state a fact, quote the supporting span verbatim in "
+            'double quotes ("…") so the verifier can confirm grounding. '
+            "If the context does not contain the answer, say so plainly."
+            f"\n\nContext:\n{context}\n\nQuestion: {args.question}"
+        )
+        try:
+            answer = chat.chat_completion(
+                messages=[{"role": "user", "content": prompt}],
+                model=model,
+                max_tokens=512,
+                temperature=0.1,
+            )
+        except Exception as e:
+            print(json.dumps({
+                "answer_text": "",
+                "audit_mode": "UNGROUNDED",
+                "sources": sources,
+                "error": f"LLM call failed: {type(e).__name__}: {e}",
+                "endpoint": endpoint,
+                "model": model,
+            }, indent=2, ensure_ascii=False))
+            return 4
+
+        verdict = verify_quotes(answer, context)
+    finally:
+        client.close()
+
+    print(json.dumps({
+        "answer_text": answer,
+        "audit_mode": verdict["audit_mode"],
+        "verifier_method": verdict["verifier_method"],
+        "n_quotes": verdict["n_quotes"],
+        "n_verified": verdict["n_verified"],
+        "unverified_quotes": verdict.get("unverified_quotes", []),
+        "sources": sources,
+        "stats": client.stats(),
+        "timing_s": round(_time.time() - t0, 3),
+        "endpoint": endpoint,
+        "model": model,
+    }, indent=2, ensure_ascii=False))
+    return 0
+
+
 def _cmd_cloud_fetch_chunk(args: argparse.Namespace) -> int:
     from arborist.merkle import hash_leaf
     client = _make_bucket_client(args)