diff --git a/arborist/cli.py b/arborist/cli.py index a9adb3c..fff98a4 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -353,6 +353,7 @@ def _cmd_query(args: argparse.Namespace) -> int: import os from arborist.qa.client import OpenAICompatibleClient, StubClient + from arborist.qa.progress import from_env as _progress_from_env from arborist.qa.query import query base_url = args.endpoint or os.environ.get( @@ -437,6 +438,15 @@ def _cmd_query(args: argparse.Namespace) -> int: # sidecar. Adds one short LLM round-trip; NEVER gates # admissibility (D1 preserved). call_policy["soft_preflight_enabled"] = True + if getattr(args, "no_canonical_preflight", False): + # Disable the math/logic π* short-circuit per-call. Forces + # RAG even on pure-arithmetic / pure-propositional input — + # useful when bench-comparing against the LLM path. + call_policy["canonical_projection_preflight"] = False + + progress = _progress_from_env( + cli_override=getattr(args, "progress_override", None), + ) result = query( question=args.question, @@ -454,6 +464,7 @@ def _cmd_query(args: argparse.Namespace) -> int: fidelity=getattr(args, "fidelity", None), burn_existing=bool(getattr(args, "burn", False)), retrieval_keywords=getattr(args, "retrieval_keywords", None), + progress=progress, ) # Emit unfirehose-compatible session journal. One JSONL file per @@ -472,7 +483,11 @@ def _cmd_query(args: argparse.Namespace) -> int: print(_render_query_human(result, args.question)) return ( 0 - if result.get("status") in ("cache_hit", "cache_miss_then_written") + if result.get("status") in ( + "cache_hit", + "cache_miss_then_written", + "canonical_projection", + ) else 1 ) @@ -807,6 +822,23 @@ def _render_query_human(result: dict, question: str) -> str: f"{cap_str} 0/0 0.0s (preflight)\n\n" f"{answer_text}" ) + if status == "canonical_projection": + # Math/logic π* short-circuited the RAG path — the canonical + # bytes ARE the answer. No retrieval, no LLM, no cache. + pi_star_ref = result.get("pi_star_ref", "?") + answer_text = result.get("answer_text") or "" + timings = result.get("timings") or {} + total_ms = timings.get("total_ms") + elapsed = ( + f"{total_ms / 1000:.1f}s" + if isinstance(total_ms, (int, float)) + else "?" + ) + return ( + f"{question}\n" + f" CANONICAL · via {pi_star_ref} {elapsed} (projected)\n\n" + f"{answer_text}" + ) if status not in ("cache_hit", "cache_miss_then_written"): msg = result.get("msg") or status or "unknown error" return f" {status or 'error'}: {msg}" @@ -2521,6 +2553,64 @@ def _cmd_activity(args: argparse.Namespace) -> int: return 0 +def _cmd_canon(args: argparse.Namespace) -> int: + """Direct π* canonicalization — no shards, no LLM, no audit chain. + + Two modes: + ``arborist canon --list`` — print registry contents. + ``arborist canon ""`` — canonicalize and print. + """ + import hashlib + + from arborist.pi_star import PiStarError, get, list_keys + + if args.list: + from arborist.pi_star import REGISTRY + for key in sorted(list_keys()): + ps = REGISTRY[key] + print(f"{key}\t{ps.domain}") + return 0 + + if not args.key or args.input is None: + print( + "error: KEY and INPUT required (or --list to see registry)", + file=sys.stderr, + ) + return 2 + + try: + ps = get(args.key) + except KeyError: + print(f"error: unknown π*: {args.key!r}", file=sys.stderr) + print( + f"available: {', '.join(sorted(list_keys()))}", + file=sys.stderr, + ) + return 2 + + try: + canonical = ps.canonicalize(args.input.encode("utf-8")) + except PiStarError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + except NotImplementedError as exc: + print(f"error: {args.key} not yet implemented ({exc})", file=sys.stderr) + return 1 + + if args.json: + digest = hashlib.sha256(canonical).hexdigest() + print(json.dumps({ + "pi_star_ref": args.key, + "input": args.input, + "canonical": canonical.decode("utf-8", errors="replace"), + "canonical_sha256": digest, + }, ensure_ascii=False)) + else: + sys.stdout.buffer.write(canonical) + sys.stdout.buffer.write(b"\n") + return 0 + + def _cmd_stats(args: argparse.Namespace) -> int: """Counts: documents, chunks (by tier), edges, providence, audit events.""" conn = ( @@ -4128,7 +4218,39 @@ def build_parser() -> argparse.ArgumentParser: "path; cannot create PREFLIGHT_OK or PREFLIGHT_BLOCKED." ), ) - query_cmd.set_defaults(func=_cmd_query) + query_cmd.add_argument( + "--no-canonical-preflight", + dest="no_canonical_preflight", action="store_true", + help=( + "Disable the math/logic π* short-circuit. Pure-arithmetic " + "(e.g. '0.1 + 0.2') and pure-propositional ('A IMPL B') " + "questions normally bypass RAG and answer via " + "arithmetic@v1 / logic-kernel@v1 directly. Pass this " + "flag to force RAG against the LLM (useful when bench-" + "comparing the canonical answer with the model's reply)." + ), + ) + progress_group = query_cmd.add_mutually_exclusive_group() + progress_group.add_argument( + "--progress", dest="progress_override", action="store_const", + const="on", + help=( + "Force per-stage state-machine emission to stderr (per-shard " + "× per-route search lines, llm.start/done, verify, persist, " + "etc.). Default: on at TTY, off when stderr is piped. " + "Format: [arborist NN.NNs] stage.name key=value …" + ), + ) + progress_group.add_argument( + "--no-progress", dest="progress_override", action="store_const", + const="off", + help=( + "Suppress state-machine emission even when stderr is a TTY. " + "Use when running interactively but piping output through " + "tools that misbehave on stderr noise." + ), + ) + query_cmd.set_defaults(func=_cmd_query, progress_override=None) inspect_cmd = sub.add_parser( "inspect", @@ -4439,6 +4561,31 @@ def build_parser() -> argparse.ArgumentParser: stats_cmd = sub.add_parser("stats", help="counts: docs, chunks, edges, audit") stats_cmd.set_defaults(func=_cmd_stats) + canon_cmd = sub.add_parser( + "canon", + help=( + "canonicalize input via a registered π* (direct projection, " + "no RAG, no LLM)" + ), + ) + canon_cmd.add_argument( + "key", nargs="?", + help="π* key, e.g. arithmetic@v1, logic-kernel@v1, code-py-ast@v1", + ) + canon_cmd.add_argument( + "input", nargs="?", + help="raw input string to canonicalize (UTF-8)", + ) + canon_cmd.add_argument( + "--list", action="store_true", + help="list registered π* keys + their domains and exit", + ) + canon_cmd.add_argument( + "--json", action="store_true", + help="emit {pi_star_ref, input, canonical, canonical_sha256}", + ) + canon_cmd.set_defaults(func=_cmd_canon) + analyze_cmd = sub.add_parser( "analyze", help="compression spectrum, depth distribution, audit chain integrity", diff --git a/arborist/pi_star/__init__.py b/arborist/pi_star/__init__.py index c64780f..9540d7e 100644 --- a/arborist/pi_star/__init__.py +++ b/arborist/pi_star/__init__.py @@ -8,19 +8,19 @@ behind a unified protocol so: - Versioning is mechanical: ``name@version`` is the registry key; changing a π* means a new key. -Existing canonicalizers re-homed in this ticket: +Six concrete π*'s ship today (alphabetical): +- ``arithmetic@v1`` — closed-form rational arithmetic (SQD §14.1). +- ``claim-lattice@v1`` — claim lines → JSON parsed-claim list. +- ``code-py-ast@v1`` — Python source → canonical AST S-expression. +- ``logic-kernel@v1`` — propositional Boolean → CNF (SQD §14.3). +- ``time-series-quantized@v1`` — JSON sample array → quantized + integer vector (SQD §13.5). - ``wikitext-base@v1`` — wraps :func:`arborist.wikitext.to_base`. -- ``claim-lattice@v1`` — wraps - :func:`arborist.qa.parse_claims.parse_pointer_claims`. -Stub modalities (raise :class:`NotImplementedError`; their own -implementation tickets land them): +Last reserved stub: -- ``code-py-ast@v1`` -- ``logic-kernel@v1`` -- ``time-series-quantized@v1`` -- ``tabular-pinned@v1`` +- ``tabular-pinned@v1`` (raises :class:`NotImplementedError`). Composition theory: see ``docs/pi-star-composition.md``. """ diff --git a/arborist/qa/query.py b/arborist/qa/query.py index 7d308f8..f791520 100644 --- a/arborist/qa/query.py +++ b/arborist/qa/query.py @@ -73,6 +73,7 @@ from arborist.qa.keys import ( ) from arborist.qa.dag import build_run_dag from arborist.qa.frame import FrameDetection, detect_frame as _detect_frame +from arborist.qa.progress import Progress, disabled as _progress_disabled from arborist.qa.retrieval_plan import RetrievalPlan, retrieval_plan_hash from arborist.qa.evidence import ( build_evidence_map, @@ -966,6 +967,8 @@ def _search_corpus( single_db: Path | None, question: str, over_fetch: int, + *, + progress: Progress | None = None, ) -> list[_Hit]: """Two parallel searches across shards, merged: @@ -996,6 +999,8 @@ def _search_corpus( else: paths = [] + progress = progress or _progress_disabled() + raw: list[tuple] = [] # Roots whose TF-IDF cores contain a query token — collected across # shards. Used downstream by _filter_by_title_relevance as accept-path 2. @@ -1011,9 +1016,11 @@ def _search_corpus( root_to_shard: dict[str, str] = {} for p in paths: + shard_name = p.name conn = connect(p) try: backend = FTS5Backend(conn) + _route_hits = 0 for h in backend.search( question, limit=over_fetch, extra_or_tokens=or_synonym_pool ): @@ -1028,6 +1035,10 @@ def _search_corpus( ) ) root_to_shard[h.document_root] = str(p.resolve()) + _route_hits += 1 + progress.emit( + f"search.shard.{shard_name}.fts5_body", hits=_route_hits + ) # Parallel title search using synonym-expanded tokens. # The score starts deliberately low so this signal can't drown # FTS5 BM25 + body relevance. _rerank_by_title later adds @@ -1064,6 +1075,7 @@ def _search_corpus( title_search_rows = _search_titles( conn, list(accept_tokens), over_fetch ) + _title_hits = 0 for r in title_search_rows: title_norm = (r["title"] or "").replace("_", " ") title_stems = { @@ -1085,6 +1097,10 @@ def _search_corpus( ) ) root_to_shard[r["document_root"]] = str(p.resolve()) + _title_hits += 1 + progress.emit( + f"search.shard.{shard_name}.title", hits=_title_hits + ) # Phrase-pattern search — verbatim multi-token sequences # from the question. Catches allusions / idioms / fictional- # world references whose diagnostic signal is the exact @@ -1103,6 +1119,7 @@ def _search_corpus( # "winter is coming") survive at length 5 or 3-with-light- # tokens, but the 4-gram floor is where diagnostic-ness # collapses. + _phrase_hits_total = 0 for n in (6, 5): phrase_score = 100.0 if n == 6 else 90.0 for r in _search_phrases( @@ -1120,9 +1137,14 @@ def _search_corpus( ) root_to_shard[r["document_root"]] = str(p.resolve()) phrase_match_roots.add(r["document_root"]) + _phrase_hits_total += 1 + progress.emit( + f"search.shard.{shard_name}.phrase", hits=_phrase_hits_total + ) # Core-keyword search: docs whose TF-IDF core keywords match. # The query token doesn't need to be in title or even in body — # being a TF-IDF keyword of the doc's core is enough signal. + _kw_hits = 0 for r in _docs_with_core_keyword_match( conn, list(accept_tokens), over_fetch ): @@ -1147,6 +1169,10 @@ def _search_corpus( ) ) root_to_shard[r["document_root"]] = str(p.resolve()) + _kw_hits += 1 + progress.emit( + f"search.shard.{shard_name}.core_keyword", hits=_kw_hits + ) finally: conn.close() @@ -1167,6 +1193,9 @@ def _search_corpus( chunk_idx=idx, ) ) + progress.emit( + "search.merge_dedupe", raw=len(raw), unique=len(out) + ) # Sidecar sets returned alongside the hit list. Pre-2026-05-01 # the function returned a bare list and the caller used # `getattr(hits, "_core_match_roots", set())` to fish out the @@ -1187,6 +1216,7 @@ def _rerank( phrase_match_roots: set[str] | None = None, hyphen_fold_anchors: set[str] | None = None, shards_dir=None, + progress: Progress | None = None, ) -> list[_Hit]: """Filter off-topic, then layer in body-coverage, title-overlap, and source-role rank boosts. @@ -1201,6 +1231,7 @@ def _rerank( each contain "Florida" hundreds of times in row markup, scoring above the actual ``Florida`` article on body density). """ + progress = progress or _progress_disabled() hits = _filter_by_title_relevance( hits, question, @@ -1210,6 +1241,7 @@ def _rerank( hyphen_fold_anchors=hyphen_fold_anchors, shards_dir=shards_dir, ) + progress.emit("search.title_filter", survivors=len(hits)) hits = _rerank_by_body_coverage(hits, question) hits = _rerank_by_title(hits, question) hits = _rerank_by_source_role(hits, question) @@ -1588,6 +1620,7 @@ def query( fidelity: str | None = None, burn_existing: bool = False, retrieval_keywords: str | None = None, + progress: Progress | None = None, ) -> dict: """Answer `question` using the corpus. Cache to qa_db. Returns a result dict. @@ -1630,6 +1663,7 @@ def query( on keyword sets. """ policy = policy or DEFAULT_QUERY_POLICY + progress = progress or _progress_disabled() if fidelity is None: fidelity = policy.get("fidelity", DEFAULT_FIDELITY) if fidelity not in FIDELITY_MODES: @@ -1731,6 +1765,7 @@ def query( # lives on the result dict separately, not on QuestionState # in this pass). from arborist.qa.metacognition import preflight_question + progress.emit("preflight.start", mode=answer_mode) _t_preflight = time.monotonic() question_state = preflight_question( question, @@ -1739,6 +1774,12 @@ def query( policy=policy, ) preflight_ms = _ms_since(_t_preflight) + progress.emit( + "preflight.done", + is_broad=bool(quantifier.get("is_broad")), + intensity=quantifier.get("intensity"), + ms=int(preflight_ms), + ) # Ticket #000011 — optional soft preflight sidecar. Default OFF; # one short LLM round-trip when policy["soft_preflight_enabled"] # is True. Returns a stub hint (SOFT_DISABLED) when off so the @@ -1942,6 +1983,49 @@ def query( } t_start = time.monotonic() + # 0. Canonical-projection preflight. + # + # Math- or logic-shaped questions answer exactly, deterministically, + # without retrieval or LLM call. Pure-arithmetic ("0.1 + 0.2") goes + # to arithmetic@v1; pure-propositional ("A IMPL B") to + # logic-kernel@v1. On match, we short-circuit with a synthetic + # CANONICAL_PROJECTION audit_mode — the canonical bytes ARE the + # answer. On no-match or PiStarError we fall through to RAG with + # zero side-effects. Disable per-call via + # policy["canonical_projection_preflight"] = False. + canonical_preflight_on = bool( + policy.get("canonical_projection_preflight", True) + ) + if canonical_preflight_on: + _t_canon = time.monotonic() + canonical_match = _canonical_projection_preflight(question) + canonical_ms = _ms_since(_t_canon) + if canonical_match is not None: + pi_star_ref, canonical_bytes = canonical_match + answer_text = canonical_bytes.decode("utf-8", errors="replace") + progress.emit( + "canonical_projection.hit", + pi_star_ref=pi_star_ref, + ms=int(canonical_ms), + ) + return { + "status": "canonical_projection", + "audit_mode": "CANONICAL_PROJECTION", + "verifier_method": "canonical_projection", + "pi_star_ref": pi_star_ref, + "answer_text": answer_text, + "n_quotes": 1, + "n_verified": 1, + "violations": [], + "lookup_path": "preflight_canonical", + "sources": [], + "cache_key": None, + "timings": { + "canonical_preflight_ms": canonical_ms, + "total_ms": _ms_since(t_start), + }, + } + # 1. Search. # # Retrieval-only query string: question + operator-supplied @@ -1954,9 +2038,15 @@ def query( retrieval_query = question if retrieval_keywords and retrieval_keywords.strip(): retrieval_query = f"{question} {retrieval_keywords.strip()}" + progress.emit( + "search.start", + top_k=int(top_k), + over_fetch=int(over_fetch), + keywords_appended=bool(retrieval_keywords and retrieval_keywords.strip()), + ) t_phase = time.monotonic() hits, core_match_roots, phrase_match_roots, root_to_shard = _search_corpus( - shards_dir, single_db, retrieval_query, over_fetch + shards_dir, single_db, retrieval_query, over_fetch, progress=progress, ) if not hits: return { @@ -1969,11 +2059,14 @@ def query( } qtokens_lower = {t.lower() for t in _title_query_tokens(retrieval_query)} + _bd_lookups = [0] + def _body_density_check(h) -> bool: # Lazy per-hit check: open the shard, count token mentions in this doc. sp = root_to_shard.get(h.document_root) or h.shard_path if not sp: return False + _bd_lookups[0] += 1 c = connect(sp) try: return _body_density_passes(c, h.document_root, qtokens_lower) @@ -1988,8 +2081,15 @@ def query( phrase_match_roots=phrase_match_roots, hyphen_fold_anchors=_hyphen_fold_variants(retrieval_query), shards_dir=shards_dir, + progress=progress, ) search_ms = _ms_since(t_phase) + progress.emit( + "search.done", + hits=len(hits), + body_density_lookups=_bd_lookups[0], + ms=int(search_ms), + ) # 2. Pull doc texts within budget. # @@ -2049,6 +2149,12 @@ def query( if policy.get("base_version") and _wikitext_to_base is not None: context = _wikitext_to_base(context) context_ms = _ms_since(t_phase) + progress.emit( + "context_assembly", + chosen=len(chosen), + chars=len(context), + ms=int(context_ms), + ) # 3. Build messages + hashes. Branch on answer_mode: # "quote" (default) raw sources block + verbatim-quote rules. @@ -2427,6 +2533,7 @@ def query( # If fidelity allows fallback AND the alternate dedup mode # produces a different cache_key, try that too — lets a # fast-cache agent reuse a record written under either mode. + progress.emit("cache_lookup", primary=primary_ckey[:8]) t_phase = time.monotonic() cached = qa_conn.execute( "SELECT * FROM providence_cache " @@ -2451,6 +2558,11 @@ def query( lookup_path = f"{other_mode}_fallback" cache_lookup_ms = _ms_since(t_phase) if cached is not None: + progress.emit( + "cache.hit", + lookup_path=lookup_path, + ms=int(cache_lookup_ms), + ) now = int(time.time()) with transaction(qa_conn): qa_conn.execute( @@ -2535,6 +2647,14 @@ def query( stop_seqs = list(policy.get( "claim_lattice_json_stop_sequences", ["\n\n"] )) + progress.emit( + "cache.miss", + ) + progress.emit( + "llm.start", + mode=answer_mode, + ctx=sum(len(m.get("content") or "") for m in messages), + ) t_phase = time.monotonic() raw_answer = chat_client.chat_completion( messages, @@ -2546,6 +2666,11 @@ def query( stop=stop_seqs, ) llm_ms = _ms_since(t_phase) + progress.emit( + "llm.done", + ms=int(llm_ms), + answer_chars=len(raw_answer or ""), + ) # 5b. Faithfulness check. Branch on answer_mode: # "quote" substring-verify quoted spans against context. @@ -2556,6 +2681,7 @@ def query( # one-shot benchmark discipline. repair_changes: list[dict] = [] pre_repair_verdict: dict | None = None + progress.emit("verify.start", mode=answer_mode) if answer_mode == "claim_lattice_pointer": verdict = verify_claim_lattice( @@ -2711,8 +2837,15 @@ def query( if verdict["unverified_quotes"] else None ) + progress.emit( + "verify.done", + audit_mode=verdict.get("audit_mode"), + n_verified=verdict.get("n_verified"), + n_quotes=verdict.get("n_quotes"), + ) # 6. Persist record + audit event. + progress.emit("persist.start") t_phase = time.monotonic() proof_obj = { "context_root": context_root, @@ -3030,6 +3163,7 @@ def query( qa_conn.close() persist_ms = _ms_since(t_phase) + progress.emit("persist.done", ms=int(persist_ms)) # Pull the verify node's failure_stage out of the run_dag for the result. failure_stage = next( (n.get("hash") for n in run_dag["nodes"] if n["stage"] == "verify"), @@ -3165,3 +3299,56 @@ def query( def _ms_since(t: float) -> float: """Wall-time elapsed in milliseconds, rounded to 1 decimal.""" return round((time.monotonic() - t) * 1000, 1) + + +# Pure-arithmetic shape: digits, dots, /, *, +, -, parens, **, whitespace. +# Rejects letters — no identifiers (so "x+1" or "what is 2+2" fall through). +_CANONICAL_ARITHMETIC_RE = re.compile(r"^[\s\d./*+\-()]+$") +# Propositional logic shape: only uppercase atoms and reserved keywords. +_CANONICAL_LOGIC_KEYWORDS = { + "AND", "OR", "NOT", "IMPL", "IFF", "XOR", "TRUE", "FALSE", +} +_CANONICAL_LOGIC_TOKEN_RE = re.compile(r"[A-Z][A-Z]*|[()]") +_CANONICAL_LOGIC_VALID_RE = re.compile(r"^[\sA-Z()]+$") + + +def _canonical_projection_preflight(question: str) -> tuple[str, bytes] | None: + """Try a math/logic π* before RAG starts. Returns + ``(pi_star_ref, canonical_bytes)`` on success; ``None`` to fall + through to text retrieval. + + Sniff is conservative — pure-arithmetic shape (no letters at all) + or pure-propositional shape (uppercase atoms + reserved keywords + only). Anything natural-language (a question mark, lowercase + words, mixed shape) returns ``None`` unmodified. The π* itself + is the final arbiter: a shape match that fails canonicalization + (PiStarError) also falls through. + """ + from arborist.pi_star import PiStarError, get + + q = question.strip() + if not q: + return None + + # Math route — pure-arithmetic shape. + if _CANONICAL_ARITHMETIC_RE.fullmatch(q): + try: + out = get("arithmetic@v1").canonicalize(q.encode("utf-8")) + return ("arithmetic@v1", out) + except PiStarError: + return None + + # Logic route — uppercase atoms + reserved keywords only. Require + # at least one logic operator keyword so a bare "A" doesn't trip + # the route (a one-atom question is more likely natural language). + if _CANONICAL_LOGIC_VALID_RE.fullmatch(q): + words = [w for w in _CANONICAL_LOGIC_TOKEN_RE.findall(q) if w not in "()"] + has_keyword = any(w in _CANONICAL_LOGIC_KEYWORDS for w in words) + if has_keyword: + try: + out = get("logic-kernel@v1").canonicalize(q.encode("utf-8")) + return ("logic-kernel@v1", out) + except PiStarError: + return None + + return None diff --git a/docs/_source/pi-star.rst b/docs/_source/pi-star.rst index 80c6a76..89901d0 100644 --- a/docs/_source/pi-star.rst +++ b/docs/_source/pi-star.rst @@ -92,6 +92,70 @@ commutativity, associativity, IMPL/IFF/XOR rewrites, De Morgan, double negation, distribution, idempotence, tautology collapse, contrapositive. +CLI surfaces +------------ + +Two ways to reach a registered π* from the shell — direct or +auto-routed. + +**Direct call** — ``arborist canon ""`` runs the named +π* against the given input and prints the canonical bytes. No +shards, no LLM, no audit chain. Pure projection. + +.. code-block:: console + + $ arborist canon arithmetic@v1 "0.1 + 0.2" + 3/10 + $ arborist canon logic-kernel@v1 "(NOT B) IMPL (NOT A)" + (NOT A OR B) + $ arborist canon time-series-quantized@v1 \ + '{"dt":1,"dv":0.1,"samples":[[0,1.0],[1,2.0]]}' + dt=1;dv=0.1;n=2;t0=0:10|20 + + # registry contents: + $ arborist canon --list + arithmetic@v1 arithmetic + claim-lattice@v1 text + code-py-ast@v1 code + logic-kernel@v1 logic + tabular-pinned@v1 tabular + time-series-quantized@v1 time-series + wikitext-base@v1 text-or-wikitext + + # JSON envelope w/ canonical-bytes SHA-256: + $ arborist canon --json arithmetic@v1 "0.1 + 0.2" + {"pi_star_ref":"arithmetic@v1","input":"0.1 + 0.2", + "canonical":"3/10","canonical_sha256":"088ae5cc..."} + +**Auto-routed via** ``arborist query`` — pure-arithmetic and +pure-propositional questions short-circuit the RAG path and answer +directly through ``arithmetic@v1`` / ``logic-kernel@v1``. The result +gets a synthetic ``audit_mode = CANONICAL_PROJECTION`` and renders as +``CANONICAL · via ``: + +.. code-block:: console + + $ arborist query "0.1 + 0.2" + 0.1 + 0.2 + CANONICAL · via arithmetic@v1 0.0s (projected) + + 3/10 + + $ arborist query "A IMPL B" + A IMPL B + CANONICAL · via logic-kernel@v1 0.0s (projected) + + (NOT A OR B) + +The sniff is conservative — pure-arithmetic shape (digits + ops, no +letters) or pure-propositional shape (uppercase atoms + reserved +keywords only). Anything natural-language ("what is 0.1 + 0.2?") +falls through to the existing RAG path. Pass +``--no-canonical-preflight`` to force the RAG path even on +matching input — useful when bench-comparing the canonical answer +against the LLM's reply. Disable per-call via +``policy["canonical_projection_preflight"] = False``. + Composition algebra ------------------- diff --git a/tests/test_canonical_projection.py b/tests/test_canonical_projection.py new file mode 100644 index 0000000..5740c1a --- /dev/null +++ b/tests/test_canonical_projection.py @@ -0,0 +1,248 @@ +"""Tests for the canonical-projection preflight + ``arborist canon`` CLI. + +Two surfaces, one substrate. The π* registry already has unit-level +coverage in ``tests/test_pi_star.py``; these tests exercise the ways +those π*'s become reachable from the user — without RAG, without an +LLM, and without a shard. + +- ``_canonical_projection_preflight`` — the sniff + π* dispatch + helper inside ``arborist.qa.query``. +- ``query()`` short-circuit path — math/logic-shaped questions return + ``status="canonical_projection"`` before retrieval starts. +- ``arborist canon ...`` CLI subcommand — direct π* call. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from arborist.qa.client import StubClient +from arborist.qa.query import _canonical_projection_preflight, query + + +# ----- preflight helper --------------------------------------------------- + + +@pytest.mark.parametrize( + "question,expected_ref,expected_canonical", + [ + ("0.1 + 0.2", "arithmetic@v1", b"3/10"), + ("0.3", "arithmetic@v1", b"3/10"), + ("1/3 + 1/6", "arithmetic@v1", b"1/2"), + ("6/4", "arithmetic@v1", b"3/2"), + ("(2+3)*4", "arithmetic@v1", b"20/1"), + ("A IMPL B", "logic-kernel@v1", b"(NOT A OR B)"), + ("(NOT B) IMPL (NOT A)", "logic-kernel@v1", b"(NOT A OR B)"), + ("A OR NOT A", "logic-kernel@v1", b"TRUE"), + ("NOT NOT A", "logic-kernel@v1", b"A"), + ], +) +def test_preflight_matches_math_or_logic(question, expected_ref, expected_canonical): + result = _canonical_projection_preflight(question) + assert result is not None, f"expected match, got None for {question!r}" + ref, canonical = result + assert ref == expected_ref + assert canonical == expected_canonical + + +@pytest.mark.parametrize( + "question", + [ + # natural language → no match + "what is 0.1 + 0.2?", + "tell me about the capital of france", + "is A IMPL B the same as B IMPL A", + # mixed case (lowercase letters) → no logic match + "a IMPL b", + # bare uppercase atom without operator → no logic match + "A", + "B", + # empty / whitespace + "", + " ", + ], +) +def test_preflight_falls_through_for_natural_language(question): + assert _canonical_projection_preflight(question) is None + + +def test_preflight_returns_none_on_pi_star_error(): + """Shape-matched but π*-rejected input falls through gracefully. + + ``arithmetic@v1`` rejects identifiers; a string that LOOKS + arithmetic but contains nothing parseable should not crash query(). + """ + # Empty parens — passes the regex shape but arithmetic@v1 rejects. + assert _canonical_projection_preflight("()") is None + + +# ----- query() short-circuit ---------------------------------------------- + + +def test_query_math_short_circuits_with_no_shards(tmp_path: Path): + """Math-shaped question returns canonical_projection without + needing a shard or hitting the LLM.""" + qa_db = tmp_path / "qa.db" + no_shards = tmp_path / "shards" # absent on purpose + result = query( + question="0.1 + 0.2", + qa_db=qa_db, + chat_client=StubClient(""), + model_id="test/model", + shards_dir=no_shards, + ) + assert result["status"] == "canonical_projection" + assert result["audit_mode"] == "CANONICAL_PROJECTION" + assert result["verifier_method"] == "canonical_projection" + assert result["pi_star_ref"] == "arithmetic@v1" + assert result["answer_text"] == "3/10" + assert result["lookup_path"] == "preflight_canonical" + assert result["sources"] == [] + assert result["cache_key"] is None + + +def test_query_logic_short_circuits(tmp_path: Path): + qa_db = tmp_path / "qa.db" + no_shards = tmp_path / "shards" + result = query( + question="A IMPL B", + qa_db=qa_db, + chat_client=StubClient(""), + model_id="test/model", + shards_dir=no_shards, + ) + assert result["status"] == "canonical_projection" + assert result["pi_star_ref"] == "logic-kernel@v1" + assert result["answer_text"] == "(NOT A OR B)" + + +def test_query_contrapositive_collapses_to_same_canonical(tmp_path: Path): + """A IMPL B and (NOT B) IMPL (NOT A) are the same equivalence class.""" + qa_db = tmp_path / "qa.db" + no_shards = tmp_path / "shards" + a = query( + question="A IMPL B", + qa_db=qa_db, + chat_client=StubClient(""), + model_id="m", + shards_dir=no_shards, + ) + b = query( + question="(NOT B) IMPL (NOT A)", + qa_db=qa_db, + chat_client=StubClient(""), + model_id="m", + shards_dir=no_shards, + ) + assert a["answer_text"] == b["answer_text"] == "(NOT A OR B)" + + +def test_query_natural_language_falls_through_to_rag(tmp_path: Path): + """Natural-language question doesn't trigger the preflight; goes + to RAG and (with no corpus) returns no_sources.""" + qa_db = tmp_path / "qa.db" + no_shards = tmp_path / "shards" + result = query( + question="what is the capital of france?", + qa_db=qa_db, + chat_client=StubClient(""), + model_id="m", + shards_dir=no_shards, + ) + # Falls through — preflight didn't match, RAG runs, finds no shards. + assert result["status"] != "canonical_projection" + assert result.get("audit_mode") != "CANONICAL_PROJECTION" + + +def test_query_canonical_preflight_disable_via_policy(tmp_path: Path): + """policy['canonical_projection_preflight']=False disables the + short-circuit even on math-shaped input.""" + qa_db = tmp_path / "qa.db" + no_shards = tmp_path / "shards" + result = query( + question="0.1 + 0.2", + qa_db=qa_db, + chat_client=StubClient(""), + model_id="m", + shards_dir=no_shards, + policy={"canonical_projection_preflight": False}, + ) + assert result["status"] != "canonical_projection" + + +# ----- `arborist canon` CLI subcommand ------------------------------------ + + +def _run_canon(*args: str) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, "-m", "arborist.cli", "canon", *args], + capture_output=True, + text=True, + check=False, + ) + + +def test_cli_canon_arithmetic(): + r = _run_canon("arithmetic@v1", "0.1 + 0.2") + assert r.returncode == 0, r.stderr + assert r.stdout.strip() == "3/10" + + +def test_cli_canon_logic(): + r = _run_canon("logic-kernel@v1", "A IMPL B") + assert r.returncode == 0, r.stderr + assert r.stdout.strip() == "(NOT A OR B)" + + +def test_cli_canon_time_series(): + r = _run_canon( + "time-series-quantized@v1", + '{"dt":1,"dv":0.1,"samples":[[0,1.0],[1,2.0]]}', + ) + assert r.returncode == 0, r.stderr + assert r.stdout.strip() == "dt=1;dv=0.1;n=2;t0=0:10|20" + + +def test_cli_canon_list(): + r = _run_canon("--list") + assert r.returncode == 0, r.stderr + keys = {line.split("\t")[0] for line in r.stdout.strip().split("\n")} + assert "arithmetic@v1" in keys + assert "logic-kernel@v1" in keys + assert "time-series-quantized@v1" in keys + assert "code-py-ast@v1" in keys + assert "claim-lattice@v1" in keys + assert "wikitext-base@v1" in keys + + +def test_cli_canon_json_emits_sha256(): + r = _run_canon("--json", "arithmetic@v1", "0.1 + 0.2") + assert r.returncode == 0, r.stderr + obj = json.loads(r.stdout) + assert obj["pi_star_ref"] == "arithmetic@v1" + assert obj["input"] == "0.1 + 0.2" + assert obj["canonical"] == "3/10" + assert len(obj["canonical_sha256"]) == 64 + + +def test_cli_canon_unknown_key_returns_2(): + r = _run_canon("does-not-exist@v1", "anything") + assert r.returncode == 2 + assert "unknown π*" in r.stderr + + +def test_cli_canon_pi_star_error_returns_1(): + r = _run_canon("arithmetic@v1", "x + 1") # identifier rejected + assert r.returncode == 1 + assert "error" in r.stderr.lower() + + +def test_cli_canon_missing_args_returns_2(): + r = _run_canon() + assert r.returncode == 2 + assert "required" in r.stderr.lower()