diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d0ebda..179027a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,15 @@ All notable changes to Ledger are documented here. timestamp window; prebind v2 receipts now carry `request_hash`/`nonce`/ `epoch` in the hash-covered payload (backward compatible). See `docs/cva-contract-spec.md`. +- **Tool-execution receipts + epistemic-source classification** (#251). + Runtime-issued HMAC-SHA256 tool receipts (unforgeable by the model, + arXiv:2603.10060) with a pramāṇa claim classifier + (pratyakṣa/anumāna/upamāna/śabda/abhāva/ungrounded), six hallucination-type + flags, five trust levels, and omitted-call completeness detection. Ships a + deterministic 1,800-scenario NyayaVerifyBench adaptation + (`benchmark/nyaya_verify_bench.py`) gated at ≥90% fabricated-reference + detection and <20 ms/response verification overhead. See + `docs/tool-receipts.md`. - **Evidence levels for receipts** (#235). Receipts now state what they prove: a four-level ladder (`structural` → `attested` → `replay` → `inclusion`) under `verification.evidence`, with stable per-level reason codes. A signed diff --git a/benchmark/nyaya_verify_bench.py b/benchmark/nyaya_verify_bench.py new file mode 100644 index 0000000..bd90673 --- /dev/null +++ b/benchmark/nyaya_verify_bench.py @@ -0,0 +1,400 @@ +#!/usr/bin/env python3 +"""Deterministic NyayaVerifyBench harness for Ledger issue #251. + +Run from the repository root with:: + + python benchmark/nyaya_verify_bench.py + +The generator intentionally uses no LLM or network calls. It creates exactly +1,800 structured scenarios: four languages, six injected hallucination types +at 50 cases each, and 150 clean controls per language. +""" +from __future__ import annotations + +import argparse +import json +import random +import statistics +import sys +import time +from collections import Counter +from pathlib import Path +from typing import Any + +# Make direct ``python benchmark/...`` execution work from any cwd while still +# keeping the harness itself stdlib-only. +_REPO_ROOT = Path(__file__).resolve().parents[1] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +from ledger_agent.tool_receipts import ( # noqa: E402 + HALLUCINATION_TYPES, + ToolReceiptLedger, + build_tool_receipt, + verify_response, +) + +LANGUAGES = ("en", "es", "fr", "hi") +BENCHMARK_KEY_ID = "bench-key" +BENCHMARK_KEY = b"nyaya-verify-bench-key" +BASE_TIMESTAMP_MS = 1_700_000_000_000 + +_SENDERS = ("Alice", "Bob", "Carol", "Dev") +_SUBJECTS = ("Deadline update", "Budget review", "Project launch", "Meeting notes") +_URL_ROOT = "https://example.test/nyaya" + +_TEMPLATES = { + "en": { + "direct": "{sender} sent {count} emails about {subject}.", + "inference": "{sender} seems worried about the {subject}.", + "absence": "No emails were found for this search.", + "source": "According to the fetched article at {url}, the report is available.", + "comparison": "The message is comparable to the {subject} message.", + }, + "es": { + "direct": "{sender} envió {count} correos sobre {subject}.", + "inference": "{sender} parece preocupado por {subject}.", + "absence": "No se encontraron correos para esta búsqueda.", + "source": "Según el artículo descargado en {url}, el informe está disponible.", + "comparison": "El mensaje es comparable al mensaje sobre {subject}.", + }, + "fr": { + "direct": "{sender} a envoyé {count} e-mails au sujet de {subject}.", + "inference": "{sender} semble inquiet au sujet de {subject}.", + "absence": "Aucun e-mail n'a été trouvé pour cette recherche.", + "source": "Selon l'article récupéré à {url}, le rapport est disponible.", + "comparison": "Le message est comparable au message sur {subject}.", + }, + "hi": { + "direct": "{sender} ने {subject} के बारे में {count} ईमेल भेजे।", + "inference": "ऐसा लगता है कि {sender} {subject} को लेकर चिंतित हैं।", + "absence": "इस खोज के लिए कोई ईमेल नहीं मिला।", + "source": "{url} पर प्राप्त लेख के अनुसार रिपोर्ट उपलब्ध है।", + "comparison": "यह संदेश {subject} वाले संदेश के समान है।", + }, +} + + +def _email_receipt( + *, language: str, case_index: int, sender: str, subject: str, count: int +) -> dict[str, Any]: + output = json.dumps( + {"sender": sender, "subject": subject, "count": count}, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ) + return build_tool_receipt( + tool_name="email_search", + input_params={"query": sender, "language": language}, + raw_output=output, + result_count=count, + facts={"sender": sender, "subject": subject, "count": count}, + duration_ms=8 + (case_index % 17), + key_id=BENCHMARK_KEY_ID, + key=BENCHMARK_KEY, + timestamp_ms=BASE_TIMESTAMP_MS + case_index, + id=f"{language}-{case_index:04d}-email", + ) + + +def _web_receipt(*, language: str, case_index: int, url: str) -> dict[str, Any]: + output = json.dumps( + {"body": "A deterministic benchmark article.", "url": url}, + sort_keys=True, + separators=(",", ":"), + ) + return build_tool_receipt( + tool_name="web_fetch", + input_params={"url": url, "language": language}, + raw_output=output, + result_count=1, + facts={"source_url": url, "fetched_urls": [url]}, + duration_ms=11 + (case_index % 13), + key_id=BENCHMARK_KEY_ID, + key=BENCHMARK_KEY, + timestamp_ms=BASE_TIMESTAMP_MS + case_index, + id=f"{language}-{case_index:04d}-web", + ) + + +def _claim_for_type( + *, + language: str, + kind: str, + email: dict[str, Any], + web: dict[str, Any], + sender: str, + subject: str, + count: int, + clean_variant: int | None = None, +) -> dict[str, Any]: + templates = _TEMPLATES[language] + direct_text = templates["direct"].format(sender=sender, subject=subject, count=count) + if kind == "fabricated_call": + return { + "text": direct_text, + "pramana": "pratyaksha", + "receipt_id": f"{email['id']}-missing", + "expected_count": count, + "expected_facts": {"sender": sender}, + } + if kind == "count_mismatch": + return { + "text": templates["direct"].format(sender=sender, subject=subject, count=count + 1), + "pramana": "pratyaksha", + "receipt_id": email["id"], + "expected_count": count + 1, + } + if kind == "fact_mismatch": + wrong_sender = "Mallory" if sender != "Mallory" else "Eve" + return { + "text": templates["direct"].format(sender=wrong_sender, subject=subject, count=count), + "pramana": "pratyaksha", + "receipt_id": email["id"], + "expected_count": count, + "expected_facts": {"sender": wrong_sender}, + } + if kind == "inference_as_fact": + return { + "text": templates["inference"].format(sender=sender, subject=subject), + "pramana": "pratyaksha", + "receipt_id": email["id"], + "premise_facts": ["sender", "subject"], + } + if kind == "false_absence": + return { + "text": templates["absence"], + "pramana": "abhava", + "receipt_id": email["id"], + } + if kind == "source_fabrication": + missing_url = f"{_URL_ROOT}/{language}/{web['id']}/never-fetched" + return { + "text": templates["source"].format(url=missing_url), + "pramana": "shabda", + "receipt_id": web["id"], + "cited_source_url": missing_url, + } + + # Clean controls rotate across all six epistemic paths that can be + # grounded. The fifth variant uses an empty email receipt for abhava. + assert kind == "clean" + variant = 0 if clean_variant is None else clean_variant + if variant % 5 == 0: + return { + "text": direct_text, + "pramana": "pratyaksha", + "receipt_id": email["id"], + "expected_count": count, + "expected_facts": {"sender": sender, "subject": subject}, + } + if variant % 5 == 1: + return { + "text": templates["inference"].format(sender=sender, subject=subject), + "pramana": "anumana", + "receipt_id": email["id"], + "premise_facts": [{"sender": sender}, {"subject": subject}], + } + if variant % 5 == 2: + return { + "text": templates["comparison"].format(subject=subject), + "pramana": "upamana", + "receipt_id": email["id"], + "premise_facts": ["sender", "subject"], + } + if variant % 5 == 3: + return { + "text": templates["source"].format(url=web["facts"]["source_url"]), + "pramana": "shabda", + "receipt_id": web["id"], + "cited_source_url": web["facts"]["source_url"], + } + return { + "text": templates["absence"], + "pramana": "abhava", + "receipt_id": email["id"], + } + + +def generate_benchmark_scenarios(seed: int = 251) -> list[dict[str, Any]]: + """Generate the exact, JSON-serializable 1,800-case benchmark corpus.""" + rng = random.Random(seed) + scenarios: list[dict[str, Any]] = [] + case_index = 0 + + for language in LANGUAGES: + for kind in HALLUCINATION_TYPES: + for _ in range(50): + sender = rng.choice(_SENDERS) + subject = rng.choice(_SUBJECTS) + count = rng.randint(1, 5) + email = _email_receipt( + language=language, + case_index=case_index, + sender=sender, + subject=subject, + count=count, + ) + url = f"{_URL_ROOT}/{language}/{case_index:04d}" + web = _web_receipt(language=language, case_index=case_index, url=url) + claim = _claim_for_type( + language=language, + kind=kind, + email=email, + web=web, + sender=sender, + subject=subject, + count=count, + ) + scenarios.append( + { + "scenario_id": f"{language}-{case_index:04d}", + "language": language, + "hallucination_type": kind, + "receipts": [email, web], + "claims": [claim], + "ground_truth": {"status": "flagged", "hallucination_type": kind}, + } + ) + case_index += 1 + + for clean_index in range(150): + sender = rng.choice(_SENDERS) + subject = rng.choice(_SUBJECTS) + count = rng.randint(1, 5) + # Every fifth control is a true absence claim. Its receipt is + # empty while all other controls use the non-empty email receipt. + email_count = 0 if clean_index % 5 == 4 else count + email = _email_receipt( + language=language, + case_index=case_index, + sender=sender, + subject=subject, + count=email_count, + ) + url = f"{_URL_ROOT}/{language}/{case_index:04d}" + web = _web_receipt(language=language, case_index=case_index, url=url) + claim = _claim_for_type( + language=language, + kind="clean", + email=email, + web=web, + sender=sender, + subject=subject, + count=email_count, + clean_variant=clean_index, + ) + scenarios.append( + { + "scenario_id": f"{language}-{case_index:04d}", + "language": language, + "hallucination_type": "clean", + "receipts": [email, web], + "claims": [claim], + "ground_truth": {"status": "verified", "hallucination_type": None}, + } + ) + case_index += 1 + + assert len(scenarios) == 1800 + return scenarios + + +def run_benchmark(seed: int = 251) -> dict[str, Any]: + """Run the deterministic verifier and return metrics for the CLI/tests.""" + scenarios = generate_benchmark_scenarios(seed=seed) + detections: Counter[str] = Counter() + totals: Counter[str] = Counter() + clean_flagged = 0 + clean_total = 0 + response_times_ns: list[int] = [] + claim_times_ns: list[int] = [] + + for scenario in scenarios: + ledger = ToolReceiptLedger(key_registry={BENCHMARK_KEY_ID: BENCHMARK_KEY}) + for receipt in scenario["receipts"]: + ledger.register(receipt) + started = time.perf_counter_ns() + result = verify_response(scenario["claims"], ledger) + elapsed = time.perf_counter_ns() - started + response_times_ns.append(elapsed) + claim_times_ns.extend([elapsed // max(1, len(scenario["claims"]))] * len(scenario["claims"])) + + kind = scenario["hallucination_type"] + verdicts = result["claims"] + if kind == "clean": + clean_total += 1 + if any(verdict["status"] == "flagged" for verdict in verdicts): + clean_flagged += 1 + else: + totals[kind] += 1 + if any(verdict["status"] == "flagged" for verdict in verdicts): + detections[kind] += 1 + + per_type = { + kind: { + "detected": detections[kind], + "total": totals[kind], + "rate": detections[kind] / totals[kind] if totals[kind] else 0.0, + } + for kind in HALLUCINATION_TYPES + } + total_ns = sum(response_times_ns) + fabricated_rate = per_type["fabricated_call"]["rate"] + return { + "seed": seed, + "scenarios": len(scenarios), + "per_type": per_type, + "fabricated_tool_reference_detection_rate": fabricated_rate, + "false_positive_rate": clean_flagged / clean_total if clean_total else 0.0, + "clean_flagged": clean_flagged, + "clean_total": clean_total, + "total_verify_ms": total_ns / 1_000_000, + "mean_response_ms": total_ns / max(1, len(response_times_ns)) / 1_000_000, + "median_response_ms": statistics.median(response_times_ns) / 1_000_000, + "median_claim_ms": statistics.median(claim_times_ns) / 1_000_000, + } + + +def _pct(value: float) -> str: + return f"{value * 100:6.2f}%" + + +def print_report(metrics: dict[str, Any]) -> None: + print(f"NyayaVerifyBench seed={metrics['seed']} scenarios={metrics['scenarios']}") + print("\nHallucination type Detected/total Detection rate") + print("----------------------------------------- --------------- --------------") + for kind, values in metrics["per_type"].items(): + print(f"{kind:41s} {values['detected']:7d}/{values['total']:<7d} {_pct(values['rate'])}") + print( + "\nOVERALL fabricated-tool-reference detection rate: " + f"{_pct(metrics['fabricated_tool_reference_detection_rate'])}" + ) + print( + "False-positive rate on clean claims: " + f"{metrics['clean_flagged']}/{metrics['clean_total']} ({_pct(metrics['false_positive_rate'])})" + ) + print( + "Verification overhead: " + f"{metrics['mean_response_ms']:.4f} ms/response average; " + f"{metrics['median_response_ms']:.4f} ms/response median; " + f"{metrics['median_claim_ms']:.4f} ms/claim median" + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--seed", type=int, default=251) + args = parser.parse_args(argv) + metrics = run_benchmark(seed=args.seed) + print_report(metrics) + if metrics["fabricated_tool_reference_detection_rate"] < 0.90: + return 2 + if metrics["median_response_ms"] > 20.0: + return 3 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/docs/tool-receipts.md b/docs/tool-receipts.md new file mode 100644 index 0000000..ef27606 --- /dev/null +++ b/docs/tool-receipts.md @@ -0,0 +1,200 @@ +# Tool-execution receipts and epistemic verification + +Ledger's `ledger_agent.tool_receipts` module is an additive bridge for +NabaOS-style claim verification ([Basu, *Tool Receipts, Not Zero-Knowledge +Proofs*, arXiv:2603.10060](https://arxiv.org/abs/2603.10060)). It does not change +Ledger's existing evidence-receipt schemas. A tool adapter can issue one +small signed receipt per call, add `receipt_to_evidence_hash(receipt)` to an +existing `evidence_hashes` list, and verify response claims at the end of a +session. + +## Receipt schema + +`build_tool_receipt` returns a dictionary with schema +`perseus-ledger-tool-receipt/v1`. The field order below is also the builder's +insertion order (canonical hashing remains key-sorted): + +| Field | Type | Meaning | +| --- | --- | --- | +| `schema` | string | Versioned schema identifier. | +| `id` | string | Caller-supplied identifier or `uuid4().hex`. | +| `tool_name` | string | Runtime tool/adapter name. | +| `input_hash` | 64-hex string | SHA-256 of compact canonical JSON for `input_params`. | +| `output_hash` | 64-hex string | SHA-256 of `raw_output` encoded as UTF-8. | +| `result_count` | non-negative integer | Number of results returned by the tool. | +| `facts` | object | Deterministically extracted ground-truth facts. | +| `timestamp_ms` | non-negative integer | Runtime issue time, in milliseconds. | +| `duration_ms` | non-negative integer | Tool execution duration. | +| `key_id` | non-empty string | Declared key-registry identifier. | +| `signature` | 64-hex string | HMAC-SHA256 signature. | + +Canonical JSON uses `json.dumps(value, sort_keys=True, separators=(",", ":"), +ensure_ascii=False)`. The signature payload is **not JSON**; it is the UTF-8 +encoding of: + +```text +id|tool_name|input_hash|output_hash|result_count|canonical(facts)|timestamp_ms +``` + +The paper's illustrative payload ends at `timestamp_ms`. Ledger deliberately +extends it with `result_count` and `canonical(facts)`: these are the fields +used to decide count and fact mismatches, so leaving them unsigned would make +the cross-check meaningless. `duration_ms` is recorded but is not in the +paper's signed field list; the schema and signature still make it visible for +audit, while the core grounding fields remain explicitly committed. + +A receipt contains commitments rather than raw input/output preimages. The +runtime or tool adapter that retains those preimages can independently +recompute both hashes; a standalone verifier validates their shape and the +HMAC binding without disclosing sensitive tool data. + +The verifier resolves keys through Ledger's existing +`evidence_levels.resolve_key`: registries may contain raw bytes or labeled +entries with `{"key_material": bytes, ...}`. The signing key is never placed +in the receipt or passed to the LLM. + +## Pramāṇa classification + +Every factual claim is tagged with one of the six lowercase labels below. +`verify_claim` returns a stable verdict with `status`, `hallucination_type`, +`trust_level`, a reason, and the cited receipt ID. + +| Pramāṇa | Source | Verification method | Trust when grounded | +| --- | --- | --- | --- | +| `pratyaksha` | Direct tool output | Verify the cited receipt; compare `expected_count` and `expected_facts`; reject an inferential marker mislabelled as direct output. | `fully_verified` | +| `anumana` | Inference from tool data | Verify that every declared premise exists as a key/value in receipt `facts`. | `mostly_verified` | +| `upamana` | Comparison or analogy | Verify that comparison subjects are present among receipt facts. | `partial` | +| `shabda` | External testimony/source | Find a verified fetch-type receipt whose `source_url` or `fetched_urls` contains `cited_source_url`. | `mostly_verified` | +| `abhava` | Knowledge from absence | Verify the cited receipt and require `result_count == 0`. | `mostly_verified` | +| `ungrounded` | No declared evidence | Cannot verify; emit an unverifiable verdict. | `ungrounded` | + +A missing cited receipt for any label except `ungrounded` is +`fabricated_call`. A valid receipt with a wrong expected count is +`count_mismatch`; a wrong expected fact is `fact_mismatch`; an inference +labelled `pratyaksha` or an inference with absent premises is +`inference_as_fact`; a non-empty receipt used for `abhava` is `false_absence`; +and a source URL absent from all verified fetch receipts is +`source_fabrication`. These six names are stable lowercase error/metric +labels used by the benchmark. + +## Six-stage protocol + +1. **User request.** The user asks the agent to retrieve, search, or act. +2. **Tool execution.** The runtime—not the LLM—executes the tool and issues + exactly one HMAC-signed receipt containing hashes, count, extracted facts, + timing, and an ID. +3. **LLM call.** The runtime supplies the raw tool result and receipt + identifier, but never the signing key. +4. **Self-tagged response.** The LLM emits each factual claim with a pramāṇa + label and the receipt ID or source URL it believes supports the claim. +5. **Verification engine.** Ledger verifies signatures, cross-checks counts and + facts, checks inference premises, validates fetched sources, and detects + fabricated or omitted calls. +6. **Trust-annotated output.** The original response can be decorated with + `render_claim` lines and a Markdown `render_verification_block` containing + summary counts and omitted receipts. + +`ToolReceiptLedger` is an insertion-ordered in-memory session registry. Its +`unreferenced(referenced_ids)` result is a completeness audit: a receipt that +exists but is never cited is an omitted-call suspect. A production runtime can +persist the receipts separately; this module intentionally does not replace +Ledger's durable event chain. + +## Trust and threat model + +The trust mapping is deliberately calibrated rather than binary: + +- **Fully verified** means a direct claim agrees with a verified receipt's + count and selected facts. It does not mean the tool itself returned true + data. +- **Mostly verified** means the evidence supports an inference, an external + fetch, or an absence claim, but the epistemic step is not direct observation. +- **Partial** means comparison subjects are grounded, while the analogy's + semantic conclusion remains weaker. +- **Unreliable** is used for a cryptographically invalid receipt or a flagged + mismatch/fabrication. +- **Ungrounded** is reserved for claims without verifiable evidence, including + explicit opinion/unknown-comparison paths. + +The threat model protects against fabricated calls, tampered receipt fields, +misstated counts/facts, fabricated source URLs, and some inference-as-fact +claims. It relies on three conditions: + +1. The LLM never sees the HMAC signing key. Under HMAC unforgeability, it + cannot manufacture a valid receipt for a call the runtime did not execute. +2. Receipt integrity is checked before epistemic comparison. Any change to + `tool_name`, `input_hash`, `output_hash`, `result_count`, `facts`, or + `timestamp_ms` invalidates the signature (the latter two are the deliberate + extension to the paper's field list). +3. The runtime maintains completeness: one receipt is emitted for each tool + call. The session ledger catches both a nonexistent cited ID and receipts + that never appear in the response's references. + +This does not protect against a compromised or incorrect tool, a stolen HMAC +key, semantic errors in a valid inference, or a model that systematically lies +in its self-tags. HMAC proves binding to the runtime's recorded data, not the +truth of an untrusted backend. + +## Benchmark + +`benchmark/nyaya_verify_bench.py` is a stdlib-only deterministic CLI. It uses +`random.Random(seed)` and generates exactly 1,800 JSON-serializable scenarios: + +- `en`, `es`, `fr`, and `hi`; +- 50 each of `fabricated_call`, `count_mismatch`, `fact_mismatch`, + `inference_as_fact`, `false_absence`, and `source_fabrication` per language; +- 150 clean controls per language. + +The corpus uses deterministic `email_search` and `web_fetch` receipts. Clean +claims use correct pramāṇa labels; injected cases change only the targeted +reference, count, fact, label/premise relation, absence assertion, or URL. No +LLM or network call is involved. The harness reports per-type detection, +overall fabricated-tool-reference detection, clean false-positive rate, total +verification time divided by responses, and median per-claim time. It exits 2 +if fabricated-reference detection is below 90% and 3 if median response +verification exceeds 20 ms. + +Measured run in this worktree (`python benchmark/nyaya_verify_bench.py`): + +```text +NyayaVerifyBench seed=251 scenarios=1800 + +Hallucination type Detected/total Detection rate +----------------------------------------- --------------- -------------- +fabricated_call 200/200 100.00% +count_mismatch 200/200 100.00% +fact_mismatch 200/200 100.00% +inference_as_fact 200/200 100.00% +false_absence 200/200 100.00% +source_fabrication 200/200 100.00% + +OVERALL fabricated-tool-reference detection rate: 100.00% +False-positive rate on clean claims: 0/600 ( 0.00%) +Verification overhead: 0.0247 ms/response average; 0.0249 ms/response median; 0.0249 ms/claim median +``` + +This synthetic implementation is intentionally deterministic and explicit, so +its 100% rates are an engineering sanity check rather than a replacement for +the paper's model-generated evaluation. For comparison, arXiv:2603.10060 +reports 94.2% fabricated-tool-reference detection, 87.6% count-mismatch +detection, and 91.3% false-absence detection, with less than 15 ms per +response. The local harness is below both the 20 ms gate and the paper's +reported latency target, but its clean controls do not measure LLM self-tagging +compliance or multilingual paraphrase difficulty. + +## Why receipts instead of ZK proofs here? + +Section 6.1 of arXiv:2603.10060 distinguishes the guarantees: ZK proofs show +that a model computation ran, while receipts show that an agent's claims are +grounded in tool execution evidence. For an interactive assistant, the latter +is the load-bearing question; a model can correctly execute a computation that +produces a hallucinated answer. The paper characterizes ZK proving as minutes +per query; using the paper's cited `zkLLM` comparison of roughly **180 s/query** +versus a receipt verification budget of **<20 ms** (and the paper's measured +`<15 ms` target), the latency ratio is several orders of magnitude. Receipts +also require only ordinary CPU/HMAC primitives and produce user-facing trust +levels, while ZK proving generally requires specialized hardware and yields a +computational-integrity result rather than semantic grounding. The approaches +are complementary for high-assurance systems: a deployment can use ZK for +model-execution integrity and receipts for claim grounding, but receipts are +the practical default for an interactive Ledger session. diff --git a/ledger_agent/tool_receipts.py b/ledger_agent/tool_receipts.py new file mode 100644 index 0000000..f7b6141 --- /dev/null +++ b/ledger_agent/tool_receipts.py @@ -0,0 +1,719 @@ +"""Signed tool-execution receipts and Nyaya (pramana) claim verification. + +This module implements the additive tool-receipt bridge described by +arXiv:2603.10060 (NabaOS). A receipt commits to a tool call and its output, +while the in-memory ledger lets a verifier cross-check the claims an agent +makes about that call. + +The paper's illustrative signature covers ``id|tool_name|input_hash| +output_hash|timestamp_ms``. This implementation deliberately extends that +payload with ``result_count`` and canonical ``facts``: those fields are the +cross-check ground truth, so leaving them unsigned would make count and fact +mismatch detection meaningless. +""" +from __future__ import annotations + +import copy +import hashlib +import hmac +import json +import time +import uuid +from collections.abc import Mapping +from typing import Any, Optional + +from .evidence_levels import resolve_key + +TOOL_RECEIPT_SCHEMA = "perseus-ledger-tool-receipt/v1" + +PRAMANA_TYPES = ( + "pratyaksha", + "anumana", + "upamana", + "shabda", + "abhava", + "ungrounded", +) + +HALLUCINATION_TYPES = ( + "fabricated_call", + "count_mismatch", + "fact_mismatch", + "inference_as_fact", + "false_absence", + "source_fabrication", +) + +_RECEIPT_FIELDS = frozenset( + { + "schema", + "id", + "tool_name", + "input_hash", + "output_hash", + "result_count", + "facts", + "timestamp_ms", + "duration_ms", + "key_id", + "signature", + } +) + + +# --------------------------------------------------------------------------- +# Canonical hashing and receipt construction + + +def _canonical_json(value: Any) -> str: + """Return the package's stable, compact JSON representation.""" + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + ) + + +def _sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _sha256_json(value: Any) -> str: + return _sha256_bytes(_canonical_json(value).encode("utf-8")) + + +def _output_bytes(raw_output: Any) -> bytes: + if isinstance(raw_output, str): + return raw_output.encode("utf-8") + if isinstance(raw_output, (bytes, bytearray)): + return bytes(raw_output) + raise TypeError("raw_output must be str or bytes") + + +def _nonnegative_int(value: Any, field: str) -> None: + if type(value) is not int or value < 0: # bool is intentionally not an int here. + raise ValueError(f"{field} must be a non-negative integer") + + +def _nonempty_string(value: Any, field: str) -> None: + if not isinstance(value, str) or not value: + raise ValueError(f"{field} must be a non-empty string") + + +def _signature_payload(receipt: Mapping[str, Any]) -> bytes: + """Build the pipe-delimited, non-JSON signature payload. + + ``result_count`` and ``facts`` are the deliberate extension to the paper's + field list. The facts object is canonicalized before it is inserted into + the pipe-delimited payload; no separator is added around that JSON value. + """ + return "|".join( + ( + str(receipt["id"]), + str(receipt["tool_name"]), + str(receipt["input_hash"]), + str(receipt["output_hash"]), + str(receipt["result_count"]), + _canonical_json(receipt["facts"]), + str(receipt["timestamp_ms"]), + ) + ).encode("utf-8") + + +def _sign_tool_receipt(receipt: Mapping[str, Any], key: bytes) -> str: + return hmac.new(key, _signature_payload(receipt), hashlib.sha256).hexdigest() + + +def build_tool_receipt( + *, + tool_name: str, + input_params: Any, + raw_output: Any, + result_count: int, + facts: dict[str, Any], + duration_ms: int, + key_id: str, + key: bytes, + timestamp_ms: Optional[int] = None, + id: Optional[str] = None, +) -> dict[str, Any]: + """Build and sign one tool-execution receipt. + + ``input_params`` and ``raw_output`` are intentionally committed as hashes; + the receipt does not disclose the potentially sensitive preimages. The + signature is HMAC-SHA256 over ``id|tool_name|input_hash|output_hash| + result_count|canonical(facts)|timestamp_ms``. Including the last two + ground-truth fields is a deliberate extension of the paper's example + payload so a verifier can trust count and fact cross-checks. + """ + _nonempty_string(tool_name, "tool_name") + _nonempty_string(key_id, "key_id") + if not isinstance(key, (bytes, bytearray)) or not key: + raise ValueError("key must be non-empty bytes") + if not isinstance(facts, dict): + raise TypeError("facts must be a dict") + _nonnegative_int(result_count, "result_count") + _nonnegative_int(duration_ms, "duration_ms") + + if timestamp_ms is None: + timestamp_ms = int(time.time() * 1000) + _nonnegative_int(timestamp_ms, "timestamp_ms") + + if id is None: + id = uuid.uuid4().hex + _nonempty_string(id, "id") + + # Force JSON serialization now so a malformed input/facts value cannot + # produce a receipt whose commitment cannot be reproduced later. + input_hash = _sha256_json(input_params) + output_hash = _sha256_bytes(_output_bytes(raw_output)) + facts_copy = copy.deepcopy(facts) + _canonical_json(facts_copy) + + receipt: dict[str, Any] = { + "schema": TOOL_RECEIPT_SCHEMA, + "id": id, + "tool_name": tool_name, + "input_hash": input_hash, + "output_hash": output_hash, + "result_count": result_count, + "facts": facts_copy, + "timestamp_ms": timestamp_ms, + "duration_ms": duration_ms, + "key_id": key_id, + } + receipt["signature"] = _sign_tool_receipt(receipt, bytes(key)) + return receipt + + +# --------------------------------------------------------------------------- +# Receipt validation and session ledger + + +def _is_sha256_hex(value: Any) -> bool: + if not isinstance(value, str) or len(value) != 64: + return False + try: + int(value, 16) + except ValueError: + return False + return True + + +def verify_tool_receipt( + receipt: Mapping[str, Any], + key_registry: Optional[Mapping[str, Any]] = None, +) -> tuple[bool, list[str]]: + """Validate receipt shape and its HMAC under the declared registry key. + + A receipt contains commitments rather than the raw input/output, so a + standalone verifier can validate their shape and the signature binding; + the preimages remain with the tool adapter. If a caller retains the + preimages, it can independently recompute the two hashes before passing + the receipt to this function. This preserves the paper's privacy + property while making every exposed field cryptographically tamper-evident. + """ + errors: list[str] = [] + + def add(code: str) -> None: + if code not in errors: + errors.append(code) + + if not isinstance(receipt, Mapping): + return False, ["not_object"] + + unknown = set(receipt) - _RECEIPT_FIELDS + if unknown: + add("unknown_field") + + if receipt.get("schema") != TOOL_RECEIPT_SCHEMA: + add("schema") + if not isinstance(receipt.get("id"), str) or not receipt.get("id"): + add("id") + if not isinstance(receipt.get("tool_name"), str) or not receipt.get("tool_name"): + add("tool_name") + for field in ("input_hash", "output_hash"): + if not _is_sha256_hex(receipt.get(field)): + add(field) + if type(receipt.get("result_count")) is not int or receipt.get("result_count", -1) < 0: + add("result_count") + if not isinstance(receipt.get("facts"), dict): + add("facts") + for field in ("timestamp_ms", "duration_ms"): + if type(receipt.get(field)) is not int or receipt.get(field, -1) < 0: + add(field) + if not isinstance(receipt.get("key_id"), str) or not receipt.get("key_id"): + add("key_id") + if not _is_sha256_hex(receipt.get("signature")): + add("signature") + + key_id = receipt.get("key_id") + key = None + if isinstance(key_id, str) and key_id: + # resolve_key is the canonical key-registry adapter for this package. + # There is intentionally no fallback key argument in this API. + key = resolve_key(key_registry, key_id, None) + if key is None: + add("unknown_key") + + if key is not None and _is_sha256_hex(receipt.get("signature")): + try: + expected = _sign_tool_receipt(receipt, key) + except (KeyError, TypeError, ValueError): + expected = None + if expected is None or not hmac.compare_digest( + expected, str(receipt["signature"]).lower() + ): + add("signature_invalid") + + return not errors, errors + + +class ToolReceiptLedger: + """An in-memory, insertion-ordered registry for one agent session.""" + + def __init__(self, *, key_registry: Optional[Mapping[str, Any]] = None): + self._receipts: dict[str, dict[str, Any]] = {} + self._key_registry: dict[str, Any] = dict(key_registry or {}) + + def issue(self, **receipt_kwargs: Any) -> dict[str, Any]: + """Build, remember, and return exactly one new receipt.""" + receipt = build_tool_receipt(**receipt_kwargs) + key_id = receipt["key_id"] + if key_id not in self._key_registry: + self._key_registry[key_id] = bytes(receipt_kwargs["key"]) + self.register(receipt) + return copy.deepcopy(receipt) + + def register(self, receipt: Mapping[str, Any]) -> dict[str, Any]: + """Register an already-built receipt, preserving session order. + + This is useful when a runtime persists receipts before constructing a + verifier. The caller supplies the corresponding key registry at + ledger construction time. + """ + if not isinstance(receipt, Mapping) or not isinstance(receipt.get("id"), str): + raise ValueError("receipt must contain a string id") + receipt_id = receipt["id"] + if receipt_id in self._receipts: + raise ValueError("duplicate_receipt_id") + self._receipts[receipt_id] = copy.deepcopy(dict(receipt)) + return copy.deepcopy(self._receipts[receipt_id]) + + # A small alias keeps adapter code readable without changing the required + # issue/verify/get/all_ids/unreferenced interface. + add = register + + def verify(self, receipt_id: str) -> tuple[bool, list[str]]: + if receipt_id not in self._receipts: + return False, ["unknown_receipt"] + return verify_tool_receipt(self._receipts[receipt_id], self._key_registry) + + def get(self, receipt_id: str) -> Optional[dict[str, Any]]: + receipt = self._receipts.get(receipt_id) + return copy.deepcopy(receipt) if receipt is not None else None + + def all_ids(self) -> list[str]: + return list(self._receipts) + + def unreferenced(self, referenced_ids: set[str]) -> list[str]: + referenced = set(referenced_ids or set()) + return [receipt_id for receipt_id in self._receipts if receipt_id not in referenced] + + +# --------------------------------------------------------------------------- +# Pramana claim verification + + +def _verdict( + claim: Mapping[str, Any], + *, + status: str, + hallucination_type: Optional[str], + trust_level: str, + reason: str, +) -> dict[str, Any]: + return { + "claim_text": claim.get("text") if isinstance(claim.get("text"), str) else "", + "pramana": claim.get("pramana") if isinstance(claim.get("pramana"), str) else "", + "status": status, + "hallucination_type": hallucination_type, + "trust_level": trust_level, + "reason": reason, + "receipt_id": claim.get("receipt_id") if isinstance(claim.get("receipt_id"), str) else None, + } + + +def _missing_receipt(ledger: ToolReceiptLedger, receipt_id: Any) -> bool: + return not isinstance(receipt_id, str) or not receipt_id or ledger.get(receipt_id) is None + + +def _receipt_or_flag( + claim: Mapping[str, Any], ledger: ToolReceiptLedger +) -> tuple[Optional[dict[str, Any]], Optional[dict[str, Any]]]: + receipt_id = claim.get("receipt_id") + if _missing_receipt(ledger, receipt_id): + return None, _verdict( + claim, + status="flagged", + hallucination_type="fabricated_call", + trust_level="unreliable", + reason="receipt_id does not exist in the session ledger", + ) + ok, errors = ledger.verify(receipt_id) + if not ok: + return None, _verdict( + claim, + status="flagged", + hallucination_type=None, + trust_level="unreliable", + reason="receipt verification failed: " + ",".join(errors), + ) + return ledger.get(receipt_id), None + + +def _fact_present(facts: Mapping[str, Any], premise: Any) -> bool: + if isinstance(premise, Mapping): + return all(key in facts and facts[key] == value for key, value in premise.items()) + if isinstance(premise, str): + return premise in facts or premise in facts.values() + return False + + +def _looks_inferential(text: str) -> bool: + lower = text.lower() + markers = ( + " seems ", + " appears ", + " likely ", + " probably ", + " may be ", + " might be ", + " suggests ", + " parece ", + " semble ", + ) + padded = f" {lower} " + return any(marker in padded for marker in markers) + + +def _is_fetch_tool(tool_name: Any) -> bool: + if not isinstance(tool_name, str): + return False + name = tool_name.lower() + return name in {"fetch", "web_fetch", "http_fetch", "url_fetch", "browser_fetch"} or "fetch" in name + + +def _url_in_facts(facts: Mapping[str, Any], url: str) -> bool: + if facts.get("source_url") == url: + return True + fetched = facts.get("fetched_urls") + if isinstance(fetched, str): + return fetched == url + if isinstance(fetched, (list, tuple, set)): + return url in fetched + return False + + +def verify_claim(claim: Mapping[str, Any], ledger: ToolReceiptLedger) -> dict[str, Any]: + """Cross-check one self-tagged claim against the session receipt ledger.""" + if not isinstance(claim, Mapping): + return { + "claim_text": "", + "pramana": "", + "status": "unverifiable", + "hallucination_type": None, + "trust_level": "ungrounded", + "reason": "claim is not an object", + "receipt_id": None, + } + + pramana = claim.get("pramana") + if not isinstance(pramana, str): + return _verdict( + claim, + status="unverifiable", + hallucination_type=None, + trust_level="ungrounded", + reason="missing pramana label", + ) + pramana = pramana.lower() + + if pramana == "ungrounded": + return _verdict( + claim, + status="unverifiable", + hallucination_type=None, + trust_level="ungrounded", + reason="no tool or source evidence was declared", + ) + if pramana not in PRAMANA_TYPES: + return _verdict( + claim, + status="unverifiable", + hallucination_type=None, + trust_level="ungrounded", + reason="unknown pramana label", + ) + + if pramana == "shabda": + receipt_id = claim.get("receipt_id") + if receipt_id is not None and _missing_receipt(ledger, receipt_id): + return _verdict( + claim, + status="flagged", + hallucination_type="fabricated_call", + trust_level="unreliable", + reason="receipt_id does not exist in the session ledger", + ) + if receipt_id is not None: + ok, errors = ledger.verify(receipt_id) + if not ok: + return _verdict( + claim, + status="flagged", + hallucination_type=None, + trust_level="unreliable", + reason="receipt verification failed: " + ",".join(errors), + ) + cited_url = claim.get("cited_source_url") + if isinstance(cited_url, str) and cited_url: + for candidate_id in ledger.all_ids(): + candidate = ledger.get(candidate_id) + if candidate is None or not _is_fetch_tool(candidate.get("tool_name")): + continue + valid, _ = ledger.verify(candidate_id) + if valid and _url_in_facts(candidate.get("facts", {}), cited_url): + return _verdict( + claim, + status="verified", + hallucination_type=None, + trust_level="mostly_verified", + reason="cited URL is present in a verified fetch receipt", + ) + return _verdict( + claim, + status="flagged", + hallucination_type="source_fabrication", + trust_level="unreliable", + reason="no verified fetch receipt contains the cited source URL", + ) + + receipt, early = _receipt_or_flag(claim, ledger) + if early is not None: + return early + assert receipt is not None + + if pramana == "pratyaksha": + premise_facts = claim.get("premise_facts") + if (isinstance(premise_facts, list) and premise_facts) or _looks_inferential( + claim.get("text", "") + ): + return _verdict( + claim, + status="flagged", + hallucination_type="inference_as_fact", + trust_level="unreliable", + reason="an inferential claim was labelled as direct tool output", + ) + expected_count = claim.get("expected_count") + if expected_count is not None and ( + type(expected_count) is not int or expected_count != receipt.get("result_count") + ): + return _verdict( + claim, + status="flagged", + hallucination_type="count_mismatch", + trust_level="unreliable", + reason="expected_count differs from receipt result_count", + ) + expected_facts = claim.get("expected_facts") + if expected_facts is not None and ( + not isinstance(expected_facts, Mapping) + or not all( + key in receipt.get("facts", {}) and receipt["facts"][key] == value + for key, value in expected_facts.items() + ) + ): + return _verdict( + claim, + status="flagged", + hallucination_type="fact_mismatch", + trust_level="unreliable", + reason="expected_facts are not a subset of receipt facts", + ) + return _verdict( + claim, + status="verified", + hallucination_type=None, + trust_level="fully_verified", + reason="claim count and facts agree with a verified receipt", + ) + + if pramana == "anumana": + premises = claim.get("premise_facts") + if not isinstance(premises, list) or not premises: + return _verdict( + claim, + status="flagged", + hallucination_type="inference_as_fact", + trust_level="unreliable", + reason="inference has no declared premises in receipt facts", + ) + facts = receipt.get("facts", {}) + if all(_fact_present(facts, premise) for premise in premises): + return _verdict( + claim, + status="verified", + hallucination_type=None, + trust_level="mostly_verified", + reason="all inference premises are present in receipt facts", + ) + return _verdict( + claim, + status="flagged", + hallucination_type="inference_as_fact", + trust_level="unreliable", + reason="one or more inference premises are absent from receipt facts", + ) + + if pramana == "upamana": + subjects = claim.get("premise_facts") + if not isinstance(subjects, list) or not subjects: + subjects = list((claim.get("expected_facts") or {}).keys()) if isinstance( + claim.get("expected_facts"), Mapping + ) else [] + if subjects and all(_fact_present(receipt.get("facts", {}), subject) for subject in subjects): + return _verdict( + claim, + status="verified", + hallucination_type=None, + trust_level="partial", + reason="comparison subjects are present in receipt facts", + ) + return _verdict( + claim, + status="unverifiable", + hallucination_type=None, + trust_level="ungrounded", + reason="comparison subjects are not grounded in receipt facts", + ) + + # The only remaining pramana is abhava. + if receipt.get("result_count") == 0: + return _verdict( + claim, + status="verified", + hallucination_type=None, + trust_level="mostly_verified", + reason="verified receipt records an empty result set", + ) + return _verdict( + claim, + status="flagged", + hallucination_type="false_absence", + trust_level="unreliable", + reason="receipt records a non-empty result set", + ) + + +def verify_response( + claims: list[Mapping[str, Any]], + ledger: ToolReceiptLedger, + referenced_receipt_ids: Optional[set[str]] = None, +) -> dict[str, Any]: + """Verify all claims and report status counts plus omitted receipts.""" + verdicts = [verify_claim(claim, ledger) for claim in claims] + by_type = {kind: 0 for kind in HALLUCINATION_TYPES} + for verdict in verdicts: + kind = verdict.get("hallucination_type") + if kind in by_type: + by_type[kind] += 1 + + if referenced_receipt_ids is None: + referenced = { + claim.get("receipt_id") + for claim in claims + if isinstance(claim, Mapping) and isinstance(claim.get("receipt_id"), str) + } + else: + referenced = set(referenced_receipt_ids) + + summary = { + "total": len(verdicts), + "verified": sum(v["status"] == "verified" for v in verdicts), + "flagged": sum(v["status"] == "flagged" for v in verdicts), + "unverifiable": sum(v["status"] == "unverifiable" for v in verdicts), + "by_type": by_type, + } + return { + "claims": verdicts, + "summary": summary, + "omitted_receipts": ledger.unreferenced(referenced), + } + + +# --------------------------------------------------------------------------- +# Markdown and evidence-hash bridges + + +def render_claim(verdict: Mapping[str, Any]) -> str: + """Render one verdict as the single-line trust annotation used in replies.""" + return ( + f"— {verdict.get('claim_text', '')} " + f"[{verdict.get('pramana', '')} · {verdict.get('trust_level', 'ungrounded')}]" + ) + + +def render_verification_block(result: Mapping[str, Any]) -> str: + """Render a complete verification result as a compact Markdown block.""" + summary = result.get("summary", {}) + lines = ["### Verification", ""] + claims = result.get("claims", []) + lines.extend(render_claim(claim) for claim in claims) + lines.extend( + [ + "", + ( + "**Summary:** " + f"{summary.get('total', 0)} total · " + f"{summary.get('verified', 0)} verified · " + f"{summary.get('flagged', 0)} flagged · " + f"{summary.get('unverifiable', 0)} unverifiable" + ), + ] + ) + omitted = result.get("omitted_receipts", []) + if omitted: + lines.append("**Omitted receipts:** " + ", ".join(str(item) for item in omitted)) + return "\n".join(lines) + + +def receipt_to_evidence_hash(receipt: Mapping[str, Any]) -> str: + """Hash the complete canonical receipt for ``evidence_hashes`` bridges.""" + return _sha256_json(dict(receipt)) + + +# Kept as a lazy compatibility hook for callers that used the first draft of +# the benchmark API. The executable harness owns the generator implementation +# so importing this core module never imports benchmark code. +def generate_benchmark_scenarios(seed: int = 251) -> list[dict[str, Any]]: + from benchmark.nyaya_verify_bench import generate_benchmark_scenarios as _generate + + return _generate(seed=seed) + + +__all__ = [ + "TOOL_RECEIPT_SCHEMA", + "PRAMANA_TYPES", + "HALLUCINATION_TYPES", + "build_tool_receipt", + "verify_tool_receipt", + "ToolReceiptLedger", + "verify_claim", + "verify_response", + "render_claim", + "render_verification_block", + "receipt_to_evidence_hash", +] diff --git a/tests/test_tool_receipts.py b/tests/test_tool_receipts.py new file mode 100644 index 0000000..6d9770d --- /dev/null +++ b/tests/test_tool_receipts.py @@ -0,0 +1,308 @@ +import copy +import hashlib +import json +from collections import Counter + +from ledger_agent.tool_receipts import ( + HALLUCINATION_TYPES, + ToolReceiptLedger, + build_tool_receipt, + generate_benchmark_scenarios, + receipt_to_evidence_hash, + render_claim, + render_verification_block, + verify_claim, + verify_response, + verify_tool_receipt, +) + + +KEY = b"tool-receipt-test-secret" +REGISTRY = {"tool-key": KEY} + + +def make_receipt(**overrides): + values = { + "tool_name": "email_search", + "input_params": {"query": "Alice"}, + "raw_output": '{"sender":"Alice","subject":"Deadline update"}', + "result_count": 2, + "facts": {"sender": "Alice", "subject": "Deadline update", "count": 2}, + "duration_ms": 12, + "key_id": "tool-key", + "key": KEY, + "timestamp_ms": 1708300000000, + "id": "receipt-1", + } + values.update(overrides) + return build_tool_receipt(**values) + + +def make_ledger(*receipts): + ledger = ToolReceiptLedger(key_registry=REGISTRY) + for receipt in receipts: + ledger.register(receipt) + return ledger + + +def test_build_and_verify_tool_receipt_signature(): + receipt = make_receipt() + assert list(receipt) == [ + "schema", + "id", + "tool_name", + "input_hash", + "output_hash", + "result_count", + "facts", + "timestamp_ms", + "duration_ms", + "key_id", + "signature", + ] + ok, errors = verify_tool_receipt(receipt, REGISTRY) + assert ok is True + assert errors == [] + facts_json = json.dumps(receipt["facts"], sort_keys=True, separators=(",", ":"), ensure_ascii=False) + payload = "|".join( + [ + receipt["id"], + receipt["tool_name"], + receipt["input_hash"], + receipt["output_hash"], + str(receipt["result_count"]), + facts_json, + str(receipt["timestamp_ms"]), + ] + ).encode() + assert receipt["signature"] == __import__("hmac").new(KEY, payload, hashlib.sha256).hexdigest() + + +def test_wrong_key_fails_verification(): + ok, errors = verify_tool_receipt(make_receipt(), {"tool-key": b"wrong"}) + assert ok is False + assert "signature_invalid" in errors + + +def test_tampering_any_signed_field_fails_verification(): + for field, value in ( + ("tool_name", "calendar_search"), + ("input_hash", "0" * 64), + ("output_hash", "1" * 64), + ("timestamp_ms", 1708300000001), + ("result_count", 99), + ("facts", {"sender": "Mallory"}), + ): + tampered = copy.deepcopy(make_receipt()) + tampered[field] = value + ok, errors = verify_tool_receipt(tampered, REGISTRY) + assert ok is False, field + assert "signature_invalid" in errors, (field, errors) + + +def test_ledger_issue_get_ids_and_unknown_verification(): + ledger = ToolReceiptLedger(key_registry=REGISTRY) + receipt = ledger.issue( + tool_name="email_search", + input_params={"query": "Alice"}, + raw_output="[]", + result_count=0, + facts={}, + duration_ms=1, + key_id="tool-key", + key=KEY, + timestamp_ms=1, + id="issue-1", + ) + assert ledger.get("issue-1") == receipt + assert ledger.all_ids() == ["issue-1"] + assert ledger.verify("issue-1") == (True, []) + assert ledger.verify("missing") == (False, ["unknown_receipt"]) + + +def test_pratyaksha_claim_is_verified(): + receipt = make_receipt() + verdict = verify_claim( + { + "text": "Alice sent two emails.", + "pramana": "pratyaksha", + "receipt_id": receipt["id"], + "expected_count": 2, + "expected_facts": {"sender": "Alice"}, + }, + make_ledger(receipt), + ) + assert verdict["status"] == "verified" + assert verdict["trust_level"] == "fully_verified" + assert verdict["hallucination_type"] is None + + +def test_fabricated_call_detection(): + verdict = verify_claim( + {"text": "The tool returned results.", "pramana": "pratyaksha", "receipt_id": "does-not-exist"}, + make_ledger(make_receipt()), + ) + assert verdict["status"] == "flagged" + assert verdict["hallucination_type"] == "fabricated_call" + assert verdict["trust_level"] == "unreliable" + + +def test_count_mismatch_detection(): + receipt = make_receipt() + verdict = verify_claim( + {"text": "Alice sent five emails.", "pramana": "pratyaksha", "receipt_id": receipt["id"], "expected_count": 5}, + make_ledger(receipt), + ) + assert verdict["hallucination_type"] == "count_mismatch" + assert verdict["status"] == "flagged" + + +def test_fact_mismatch_detection(): + receipt = make_receipt() + verdict = verify_claim( + {"text": "Bob sent the email.", "pramana": "pratyaksha", "receipt_id": receipt["id"], "expected_facts": {"sender": "Bob"}}, + make_ledger(receipt), + ) + assert verdict["hallucination_type"] == "fact_mismatch" + assert verdict["status"] == "flagged" + + +def test_inference_as_fact_detection(): + receipt = make_receipt() + verdict = verify_claim( + { + "text": "Alice seems worried.", + "pramana": "pratyaksha", + "receipt_id": receipt["id"], + "premise_facts": ["sender", "subject"], + }, + make_ledger(receipt), + ) + assert verdict["hallucination_type"] == "inference_as_fact" + assert verdict["status"] == "flagged" + + +def test_false_absence_detection(): + receipt = make_receipt() + verdict = verify_claim( + {"text": "No emails were found.", "pramana": "abhava", "receipt_id": receipt["id"]}, + make_ledger(receipt), + ) + assert verdict["hallucination_type"] == "false_absence" + assert verdict["status"] == "flagged" + + +def test_shabda_source_fabrication_and_fetched_source(): + receipt = make_receipt( + tool_name="web_fetch", + input_params={"url": "https://example.com/article"}, + raw_output="article", + result_count=1, + facts={"source_url": "https://example.com/article"}, + id="web-1", + ) + ledger = make_ledger(receipt) + good = verify_claim( + { + "text": "According to the article, ...", + "pramana": "shabda", + "receipt_id": receipt["id"], + "cited_source_url": "https://example.com/article", + }, + ledger, + ) + bad = verify_claim( + { + "text": "According to a fabricated article, ...", + "pramana": "shabda", + "receipt_id": receipt["id"], + "cited_source_url": "https://example.com/missing", + }, + ledger, + ) + assert good["status"] == "verified" + assert bad["hallucination_type"] == "source_fabrication" + assert bad["status"] == "flagged" + + +def test_anumana_premises_and_upamana_comparison(): + receipt = make_receipt() + ledger = make_ledger(receipt) + inference = verify_claim( + { + "text": "Alice appears to be discussing a deadline.", + "pramana": "anumana", + "receipt_id": receipt["id"], + "premise_facts": [{"sender": "Alice"}, {"subject": "Deadline update"}], + }, + ledger, + ) + comparison = verify_claim( + { + "text": "Alice's message is like the deadline update.", + "pramana": "upamana", + "receipt_id": receipt["id"], + "premise_facts": ["sender", "subject"], + }, + ledger, + ) + missing = verify_claim( + {"text": "This inference has no basis.", "pramana": "anumana", "receipt_id": receipt["id"], "premise_facts": ["missing"]}, + ledger, + ) + assert inference["status"] == "verified" + assert inference["trust_level"] == "mostly_verified" + assert comparison["status"] == "verified" + assert comparison["trust_level"] == "partial" + assert missing["hallucination_type"] == "inference_as_fact" + + +def test_ungrounded_claim_is_unverifiable(): + verdict = verify_claim({"text": "It will rain tomorrow.", "pramana": "ungrounded", "receipt_id": None}, make_ledger()) + assert verdict["status"] == "unverifiable" + assert verdict["trust_level"] == "ungrounded" + assert verdict["hallucination_type"] is None + + +def test_verify_response_summary_and_omitted_receipts(): + used = make_receipt(id="used") + omitted = make_receipt(id="omitted", input_params={"query": "Bob"}) + ledger = make_ledger(used, omitted) + result = verify_response( + [{"text": "Alice sent two emails.", "pramana": "pratyaksha", "receipt_id": "used", "expected_count": 2}], + ledger, + ) + assert result["summary"] == { + "total": 1, + "verified": 1, + "flagged": 0, + "unverifiable": 0, + "by_type": {key: 0 for key in HALLUCINATION_TYPES}, + } + assert result["omitted_receipts"] == ["omitted"] + + +def test_rendering_and_evidence_hash_are_stable(): + receipt = make_receipt() + verdict = verify_claim( + {"text": "Alice sent two emails.", "pramana": "pratyaksha", "receipt_id": receipt["id"], "expected_count": 2}, + make_ledger(receipt), + ) + line = render_claim(verdict) + block = render_verification_block({"claims": [verdict], "summary": {"total": 1, "verified": 1, "flagged": 0, "unverifiable": 0, "by_type": {}}, "omitted_receipts": []}) + assert line == "— Alice sent two emails. [pratyaksha · fully_verified]" + assert "Verification" in block + assert receipt_to_evidence_hash(receipt) == receipt_to_evidence_hash(copy.deepcopy(receipt)) + assert len(receipt_to_evidence_hash(receipt)) == 64 + + +def test_benchmark_generator_is_deterministic_and_balanced(): + first = generate_benchmark_scenarios(seed=251) + second = generate_benchmark_scenarios(seed=251) + assert first == second + assert len(first) == 1800 + counts = Counter(item["hallucination_type"] for item in first) + assert counts["clean"] == 600 + for kind in HALLUCINATION_TYPES: + assert counts[kind] == 200 + assert Counter(item["language"] for item in first) == {"en": 450, "es": 450, "fr": 450, "hi": 450}