engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

qa: math/logic π* reach the user — arborist canon + query preflight

Browse source 
`make query Q="0.1 + 0.2"` used to return `no_sources` because
arithmetic-shaped input has no FTS5 hits in any text shard. Two
surfaces close that gap.

**`arborist canon <key> "<input>"`** — direct π* call, no shards,
no LLM, no audit chain. Pure projection:

    $ arborist canon arithmetic@v1 "0.1 + 0.2"      → 3/10
    $ arborist canon logic-kernel@v1 "A IMPL B"     → (NOT A OR B)
    $ arborist canon --list                         → registry contents
    $ arborist canon --json arithmetic@v1 "0.1+0.2" → SHA-256 envelope

**Math/logic preflight in `arborist query`** — pure-arithmetic and
pure-propositional questions short-circuit RAG and answer through
arithmetic@v1 / logic-kernel@v1 directly. Synthetic
`audit_mode=CANONICAL_PROJECTION`, renders as
`CANONICAL · via <pi_star_ref>`:

    $ arborist query "0.1 + 0.2"
    0.1 + 0.2
      CANONICAL · via arithmetic@v1   0.0s   (projected)
    3/10

    $ arborist query "(NOT B) IMPL (NOT A)"
    (NOT B) IMPL (NOT A)
      CANONICAL · via logic-kernel@v1   0.0s   (projected)
    (NOT A OR B)

Sniff is conservative: pure-arithmetic shape (digits + ops, no
letters) or pure-propositional shape (uppercase atoms + reserved
keywords only). Natural-language wrapping ("what is 0.1+0.2?")
falls through to RAG. PiStarError on a shape match also falls
through — preflight is best-effort, never blocking.

Disable per-call: `--no-canonical-preflight` flag,
`policy["canonical_projection_preflight"]=False`.

No schema changes: CANONICAL_PROJECTION is a render-layer audit_mode
token. No providence_cache writes, no audit_events, no
governance_policy_hash bump. The canonical bytes ARE the answer;
SHA-256 of the bytes is the equivalence-class identity (already
committed via the π* registry).

Side housekeeping: arborist/pi_star/__init__.py docstring caught up
with reality — six concrete π*'s ship today, only tabular-pinned@v1
remains as a stub.

31 new tests (preflight sniff + dispatch, query short-circuit,
contrapositive equivalence-class collapse, CLI subcommand exit codes
and JSON envelope, --no-canonical-preflight policy gate). Full
suite: 1300 passed, 36 skipped.
This commit is contained in:
russell@unturf.com 2026-05-08 12:18:28 -04:00
parent 508a25076d
commit 0a2e347f9c
No known key found for this signature in database
5 changed files with 658 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

151
arborist/cli.py
View file

@ -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 <key> "<input>"`` — 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",

18
arborist/pi_star/__init__.py
View file

@ -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``.
"""

189
arborist/qa/query.py
View file

@ -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

64
docs/_source/pi-star.rst
View file

@ -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 <key> "<input>"`` 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 <pi_star_ref>``:
.. 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
-------------------

248
tests/test_canonical_projection.py Normal file
View file

@ -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()