From ee48f82f06f584254840a6f34d9a2d965b06113f Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Mon, 13 Jul 2026 20:38:18 +0200 Subject: [PATCH] Stage source artifacts as reviewable KB proposals (#123) --- hermes-agent/leoclean-bin/kb_tool.py | 343 +++++++++++++++++- .../vps/leo-db-context/__init__.py | 7 +- .../vps/teleo-kb-bridge/REFERENCE.md | 28 +- .../vps/teleo-kb-bridge/SKILL.md | 32 +- .../working_leo_m3taversal_oos_benchmark.py | 5 +- .../test_hermes_leoclean_db_context_plugin.py | 4 +- .../test_hermes_leoclean_kb_bridge_source.py | 9 +- tests/test_hermes_leoclean_skill_surfaces.py | 6 +- tests/test_kb_tool_propose_source.py | 246 +++++++++++++ ...st_working_leo_m3taversal_oos_benchmark.py | 12 +- 10 files changed, 663 insertions(+), 29 deletions(-) create mode 100644 tests/test_kb_tool_propose_source.py diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index 2049ebc..35b8975 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -13,14 +13,24 @@ this tool does not apply it to public.claims, public.claim_edges, or evidence. from __future__ import annotations import argparse +import hashlib import json import os import re import subprocess +import sys +import tempfile +import uuid from collections import defaultdict +from pathlib import Path from typing import Any DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz" +SOURCE_COMPILER_PATH = Path("/opt/teleo-eval/workspaces/deploy-infra/scripts/compile_kb_source_packet.py") +SOURCE_PROPOSAL_RECEIPT_SCHEMA = "livingip.teleoKbSourceProposalReceipt.v1" +SOURCE_COMPILER_BUNDLE_SCHEMA = "livingip.kbSourceProposalBundle.v1" +CANONICAL_COUNT_LABELS = ("claims", "sources", "claim_edges", "claim_evidence") +SHA256_RE = re.compile(r"^[0-9a-f]{64}$") WORKER_SUPPORTED_PROPOSAL_TYPES = frozenset({"revise_strategy", "add_edge", "attach_evidence", "approve_claim"}) DIRECT_READBACK_CONTRACTS = frozenset( { @@ -205,6 +215,16 @@ def parse_args() -> argparse.Namespace: propose_attachment_eval.add_argument("--rationale", required=True) add_output_flag(propose_attachment_eval) + propose_source = sub.add_parser( + "propose-source", + help="Compile a hash-bound source artifact and stage one idempotent pending-review proposal.", + ) + propose_source.add_argument("--artifact", required=True, type=Path) + propose_source.add_argument("--text", required=True, type=Path) + propose_source.add_argument("--manifest", required=True, type=Path) + propose_source.add_argument("--receipt", required=True, type=Path) + add_output_flag(propose_source) + record_document_eval = sub.add_parser( "record-document-evaluation", help="Record a lightweight document evaluation decision without creating a KB mutation proposal.", @@ -488,8 +508,11 @@ def operational_contracts( contracts.append( { "id": "source_intake", - "required_lead": "Local compiler proven; live VPS/chat intake not shipped.", - "capability_tier": "build-local compiler only", + "required_lead": ( + "VPS source-to-proposal staging is shipped for prepared inputs; autonomous chat attachment " + "extraction is not shipped." + ), + "capability_tier": "VPS-local prepared-input compiler and pending-review staging", "safe_without_apply_authorization": [ "retain artifact plus URL or storage_path plus content hash", "extract candidates into a pending_review proposal packet", @@ -497,7 +520,7 @@ def operational_contracts( "approval_begins": "before guarded canonical public.* apply, not before mechanical capture/staging", "current_public_sources_columns": list(schema.get("sources", ())), "must_not_claim": [ - "live VPS Leo can invoke the compiler from chat", + "Leo automatically captures or extracts arbitrary Telegram attachments or tweets", "a chat label, filename, attachment ref, or source_ref is source identity", "author/title/publisher/publication date are current public.sources columns", ], @@ -1615,6 +1638,303 @@ def get_claim(args: argparse.Namespace, claim_id: str) -> dict[str, Any] | None: return rows[0] if rows else None +def canonical_json_sha256(value: Any) -> str: + encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def _validated_uuid(value: Any, label: str) -> str: + try: + return str(uuid.UUID(str(value))) + except (ValueError, AttributeError) as exc: + raise SystemExit(f"{label} must be a full UUID") from exc + + +def _validated_source_bundle(bundle: Any) -> tuple[dict[str, Any], dict[str, Any]]: + if not isinstance(bundle, dict) or bundle.get("schema") != SOURCE_COMPILER_BUNDLE_SCHEMA: + raise SystemExit(f"source compiler must return {SOURCE_COMPILER_BUNDLE_SCHEMA}") + if ( + bundle.get("status") != "pending_review" + or bundle.get("build_only") is not True + or bundle.get("database_write_performed") is not False + or bundle.get("canonical_apply_performed") is not False + ): + raise SystemExit("source compiler bundle violated its build-only pending-review contract") + + child = bundle.get("strict_child_proposal") + if not isinstance(child, dict): + raise SystemExit("source compiler bundle has no strict_child_proposal object") + _validated_uuid(child.get("id"), "strict child proposal id") + if child.get("proposal_type") != "approve_claim" or child.get("status") != "pending_review": + raise SystemExit("strict child must be one pending_review approve_claim proposal") + if child.get("proposed_by_handle") != "leo": + raise SystemExit("strict child proposed_by_handle must be leo") + for key in ("channel", "source_ref", "rationale", "payload_sha256"): + if not isinstance(child.get(key), str) or not child[key].strip(): + raise SystemExit(f"strict child {key} must be a non-empty string") + if child["channel"] != "source_document_compiler" or not child["source_ref"].startswith("normalized:"): + raise SystemExit("strict child is not bound to the source-document compiler contract") + if not SHA256_RE.fullmatch(child["payload_sha256"]): + raise SystemExit("strict child payload_sha256 must be a lowercase SHA-256 digest") + if child.get("proposed_by_agent_id") is not None: + _validated_uuid(child["proposed_by_agent_id"], "strict child proposed_by_agent_id") + payload = child.get("payload") + if not isinstance(payload, dict): + raise SystemExit("strict child payload must be an object") + apply_payload = payload.get("apply_payload") + if not isinstance(apply_payload, dict) or apply_payload.get("contract_version") != 2: + raise SystemExit("strict child must contain a schema-valid approve_claim v2 apply_payload") + if canonical_json_sha256(payload) != child["payload_sha256"]: + raise SystemExit("strict child payload hash does not match its canonical JSON") + + hashes = bundle.get("hashes") + if not isinstance(hashes, dict): + raise SystemExit("source compiler bundle has no hashes object") + for key in ( + "artifact_sha256", + "extracted_text_sha256", + "manifest_canonical_sha256", + "strict_child_payload_sha256", + "bundle_content_sha256", + ): + if not isinstance(hashes.get(key), str) or not SHA256_RE.fullmatch(hashes[key]): + raise SystemExit(f"source compiler hash {key} is missing or invalid") + if hashes["strict_child_payload_sha256"] != child["payload_sha256"]: + raise SystemExit("source compiler child hash disagrees with the strict child") + + bundle_without_content_hash = json.loads(json.dumps(bundle)) + bundle_without_content_hash["hashes"].pop("bundle_content_sha256") + if canonical_json_sha256(bundle_without_content_hash) != hashes["bundle_content_sha256"]: + raise SystemExit("source compiler bundle content hash does not match its canonical JSON") + + manifest = bundle.get("validated_manifest") + if not isinstance(manifest, dict) or not isinstance(manifest.get("source"), dict): + raise SystemExit("source compiler bundle has no validated source manifest") + if manifest.get("artifact_sha256") != hashes["artifact_sha256"]: + raise SystemExit("validated manifest artifact hash disagrees with the compiler receipt") + if manifest.get("extracted_text_sha256") != hashes["extracted_text_sha256"]: + raise SystemExit("validated manifest text hash disagrees with the compiler receipt") + return child, manifest + + +def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]: + if not SOURCE_COMPILER_PATH.is_file(): + raise SystemExit(f"source compiler is not deployed: {SOURCE_COMPILER_PATH}") + command = [ + sys.executable, + str(SOURCE_COMPILER_PATH), + "--artifact", + str(args.artifact), + "--text", + str(args.text), + "--manifest", + str(args.manifest), + ] + try: + completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise SystemExit(f"source compiler could not run: {exc}") from exc + if completed.returncode != 0: + detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}" + raise SystemExit(f"source compiler rejected the artifact: {detail}") + try: + bundle = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise SystemExit(f"source compiler emitted invalid JSON: {exc}") from exc + _validated_source_bundle(bundle) + return bundle + + +def build_source_stage_sql(child: dict[str, Any]) -> str: + proposal_id = sql_literal(_validated_uuid(child.get("id"), "strict child proposal id")) + proposal_type = sql_literal(str(child["proposal_type"])) + proposed_by_handle = sql_literal(str(child.get("proposed_by_handle") or "")) + proposed_by_agent_id = child.get("proposed_by_agent_id") + agent_sql = ( + f"{sql_literal(_validated_uuid(proposed_by_agent_id, 'strict child proposed_by_agent_id'))}::uuid" + if proposed_by_agent_id + else "null::uuid" + ) + channel = sql_literal(str(child["channel"])) + source_ref = sql_literal(str(child["source_ref"])) + rationale = sql_literal(str(child["rationale"])) + payload = sql_literal(json.dumps(child["payload"], sort_keys=True, separators=(",", ":"), ensure_ascii=True)) + payload_sha256 = sql_literal(str(child["payload_sha256"])) + return f"""\\set ON_ERROR_STOP on +begin; +create temporary table teleo_source_stage_result (created boolean not null) on commit drop; +with inserted as ( + insert into kb_stage.kb_proposals + (id, proposal_type, status, proposed_by_handle, proposed_by_agent_id, + channel, source_ref, rationale, payload) + values + ({proposal_id}::uuid, {proposal_type}, 'pending_review', nullif({proposed_by_handle}, ''), + {agent_sql}, {channel}, {source_ref}, {rationale}, {payload}::jsonb) + on conflict (id) do nothing + returning id +) +insert into teleo_source_stage_result(created) +select exists(select 1 from inserted); + +do $teleo_source_stage$ +declare + marker_count integer; + exact_count integer; +begin + select count(*) into marker_count + from kb_stage.kb_proposals + where source_ref = {source_ref}; + select count(*) into exact_count + from kb_stage.kb_proposals + where id = {proposal_id}::uuid + and proposal_type = {proposal_type} + and proposed_by_handle is not distinct from nullif({proposed_by_handle}, '') + and proposed_by_agent_id is not distinct from {agent_sql} + and channel = {channel} + and source_ref = {source_ref} + and rationale = {rationale} + and payload = {payload}::jsonb; + if marker_count <> 1 or exact_count <> 1 then + raise exception 'source stage validation failed: marker_count=%, exact_count=%', marker_count, exact_count; + end if; +end +$teleo_source_stage$; + +select jsonb_build_object( + 'created', (select created from teleo_source_stage_result), + 'proposal_id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'source_ref', source_ref, + 'payload', payload, + 'payload_sha256', {payload_sha256}, + 'created_at', created_at::text, + 'reviewed_at', reviewed_at::text, + 'applied_at', applied_at::text +)::text +from kb_stage.kb_proposals +where id = {proposal_id}::uuid; +commit; +""" + + +def stage_source_child(args: argparse.Namespace, child: dict[str, Any]) -> dict[str, Any]: + rows = psql_json(args, build_source_stage_sql(child)) + if len(rows) != 1: + raise SystemExit("source staging did not return exactly one proposal receipt") + row = rows[0] + expected = { + "proposal_id": child["id"], + "proposal_type": child["proposal_type"], + "source_ref": child["source_ref"], + "payload_sha256": child["payload_sha256"], + "payload": child["payload"], + } + for key, value in expected.items(): + if row.get(key) != value: + raise SystemExit(f"source staging receipt mismatch for {key}") + if row.get("status") not in {"pending_review", "approved", "applied", "canceled"}: + raise SystemExit("source staging returned an unknown proposal status") + if not isinstance(row.get("created"), bool): + raise SystemExit("source staging receipt has no created boolean") + return row + + +def _write_private_json(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, raw_temp_path = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent) + temp_path = Path(raw_temp_path) + try: + os.fchmod(fd, 0o600) + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(value, handle, indent=2, sort_keys=True, ensure_ascii=True) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_path, path) + finally: + if temp_path.exists(): + temp_path.unlink() + + +def propose_source(args: argparse.Namespace) -> dict[str, Any]: + if not args.local: + raise SystemExit( + "propose-source is VPS-local only; run it through the deployed teleo-kb wrapper on the VPS" + ) + receipt_path = args.receipt.expanduser().resolve() + input_paths = {args.artifact.expanduser().resolve(), args.text.expanduser().resolve(), args.manifest.expanduser().resolve()} + if receipt_path in input_paths: + raise SystemExit("--receipt must not overwrite an input file") + + bundle = compile_source_bundle(args) + child, manifest = _validated_source_bundle(bundle) + try: + compiler_sha256 = hashlib.sha256(SOURCE_COMPILER_PATH.read_bytes()).hexdigest() + receipt_path.parent.mkdir(parents=True, exist_ok=True) + fd, receipt_preflight = tempfile.mkstemp(prefix=f".{receipt_path.name}.preflight.", dir=receipt_path.parent) + os.close(fd) + Path(receipt_preflight).unlink() + except OSError as exc: + raise SystemExit(f"receipt path or compiler cannot be readied before staging: {exc}") from exc + before = status(args) + staged = stage_source_child(args, child) + after = status(args) + before_counts = dict(before.get("high_signal_rows") or {}) + after_counts = dict(after.get("high_signal_rows") or {}) + canonical_counts_unchanged = all(before_counts.get(key) == after_counts.get(key) for key in CANONICAL_COUNT_LABELS) + proposal_delta = after_counts.get("kb_proposals", 0) - before_counts.get("kb_proposals", 0) + created = bool(staged["created"]) + source = manifest["source"] + hashes = bundle["hashes"] + receipt = { + "schema": SOURCE_PROPOSAL_RECEIPT_SCHEMA, + "artifact": "teleo_kb_source_proposal", + "status": staged["status"], + "created": created, + "proposal": { + "id": staged["proposal_id"], + "proposal_type": staged["proposal_type"], + "status": staged["status"], + "source_ref": staged["source_ref"], + "payload_sha256": staged["payload_sha256"], + "created_at": staged.get("created_at"), + "reviewed_at": staged.get("reviewed_at"), + "applied_at": staged.get("applied_at"), + }, + "source": { + "identity": source.get("identity"), + "locator": source.get("locator"), + "artifact_sha256": hashes["artifact_sha256"], + "extracted_text_sha256": hashes["extracted_text_sha256"], + "manifest_canonical_sha256": hashes["manifest_canonical_sha256"], + }, + "compiler": { + "path": str(SOURCE_COMPILER_PATH), + "sha256": compiler_sha256, + "bundle_content_sha256": hashes["bundle_content_sha256"], + }, + "database": { + "backend": after.get("backend"), + "before_counts": before_counts, + "after_counts": after_counts, + "canonical_counts_unchanged": canonical_counts_unchanged, + "proposal_count_delta": proposal_delta, + "proposal_count_matches_command": proposal_delta == (1 if created else 0), + }, + "write_contract": { + "target_table": "kb_stage.kb_proposals", + "database_write_performed_by_command": created, + "canonical_apply_performed": False, + "production_db_apply_ran": False, + "public_table_write_path_present": False, + }, + "receipt_path": str(receipt_path), + } + _write_private_json(receipt_path, receipt) + return receipt + + def propose_edge(args: argparse.Namespace) -> dict[str, Any]: sql = f""" with params as ( @@ -2259,6 +2579,19 @@ def print_decision_matrix_status(data: dict[str, Any]) -> None: print() +def print_source_proposal_receipt(data: dict[str, Any]) -> None: + proposal = data["proposal"] + database = data["database"] + print("# Source Proposal Receipt\n") + print(f"- proposal: `{proposal['id']}`") + print(f"- status: `{proposal['status']}`; created: `{data['created']}`") + print(f"- source_ref: `{proposal['source_ref']}`") + print(f"- canonical counts unchanged: `{database['canonical_counts_unchanged']}`") + print(f"- proposal count delta: `{database['proposal_count_delta']}`") + print("- canonical apply performed: `false`") + print(f"- receipt: `{data['receipt_path']}`") + + def main() -> int: args = parse_args() if args.command == "status": @@ -2296,6 +2629,8 @@ def main() -> int: data = propose_edge(args) elif args.command == "propose-attachment-evaluation": data = propose_attachment_evaluation(args) + elif args.command == "propose-source": + data = propose_source(args) elif args.command == "record-document-evaluation": data = record_document_evaluation(args) elif args.command == "list-proposals": @@ -2328,6 +2663,8 @@ def main() -> int: ) elif args.command in {"propose-edge", "propose-attachment-evaluation"}: print_proposal(data) + elif args.command == "propose-source": + print_source_proposal_receipt(data) elif args.command == "record-document-evaluation": print_json(data) elif args.command == "list-proposals": diff --git a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py index f0369d7..3915fa2 100644 --- a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py +++ b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py @@ -338,7 +338,12 @@ def _source_intake_issues(response: str) -> list[str]: issues.append("missing_staging") if not re.search(r"approval|authoriz", response, re.I) or "apply" not in lowered: issues.append("missing_apply_boundary") - if not re.search(r"live.{0,80}(?:not shipped|not live|unavailable)", response, re.I | re.S): + if not re.search( + r"live.{0,80}(?:not shipped|not live|unavailable)|" + r"autonomous.{0,80}(?:not shipped|not live|unavailable)", + response, + re.I | re.S, + ): issues.append("missing_live_intake_ceiling") for segment in re.split(r"(?<=[.!?])\s+|\n+", response): negative = re.search( diff --git a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/REFERENCE.md b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/REFERENCE.md index f4c78d4..6cde87d 100644 --- a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/REFERENCE.md +++ b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/REFERENCE.md @@ -65,11 +65,13 @@ Apply these gates before drafting the answer or proposing a next action: 4. **Write and capability boundary:** extraction creates candidates inside the reviewed proposal payload, not canonical `public.sources` or `public.claims`. Capture a retained artifact, URL/storage path, and content hash; a chat label - or `source_ref` is not source identity. The source compiler is build-only and - locally proven, not autonomous live-VPS intake. It can create a - `pending_review` packet without apply authorization. Every current-capability - intake answer must lead with: local compiler proven; live VPS/chat intake not - shipped. Current `approve_claim` cannot insert + or `source_ref` is not source identity. `teleo-kb propose-source` is shipped + on the VPS for prepared artifact/text/manifest inputs and can stage one + deterministic `pending_review` packet without apply authorization. It is not + an autonomous Telegram attachment or tweet extractor. Every current-capability + intake answer must lead with: VPS source-to-proposal staging is shipped for + prepared inputs; autonomous chat attachment extraction is not shipped. + Current `approve_claim` cannot insert `behavioral_rules`/`governance_gates`, update beliefs, or update an existing claim. Keep unsupported changes out of runnable apply sequences. `applied_at` belongs to the proposal receipt, not each inserted row. @@ -530,6 +532,22 @@ reached through `teleo-kb`: Canonical KB writes are locked. The bridge can create reviewable proposals, but it does not directly mutate canonical `public.*` rows from normal chat. +For prepared source inputs on the VPS, use the bounded staging surface: + +```bash +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb propose-source \ + --artifact /path/to/source.bin \ + --text /path/to/source.txt \ + --manifest /path/to/source-manifest.json \ + --receipt /home/teleo/.hermes/profiles/leoclean/state/kb-source-receipts/source.json +``` + +The compiler validates content hashes, exact quotes, current taxonomies, and a +strict `approve_claim` v2 child before the bridge inserts an idempotent +`pending_review` row. Its SQL contains no canonical table write path. Read the +receipt and proposal back; do not describe this as Telegram extraction, +approval, or canonical apply. + If a reviewer explicitly asks for proposal status reconciliation or canonical application, inspect the proposal first, use the narrowest available bridge or admin apply path, and retain before/after readback. The normal chat bridge does diff --git a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md index 32515bf..2fe864f 100644 --- a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md +++ b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md @@ -1,7 +1,7 @@ --- name: teleo-kb-bridge description: Use the VPS Postgres KB bridge before answering questions about claims, evidence, agent beliefs, schema-backed identity, proposal readiness, or KB changes. -version: 1.1.0 +version: 1.2.0 author: m3taversal license: MIT metadata: @@ -58,11 +58,13 @@ Apply these before drafting the answer or proposing a next action: 4. **Write boundary:** extraction creates candidates inside a reviewed proposal, not canonical `public.sources` or `public.claims`. Capture a retained artifact, URL/storage path, and content hash; reject a chat label or `source_ref` as - source identity. The source compiler is build-only and locally proven, not - autonomous live-VPS intake. It can create a `pending_review` packet without - apply authorization. Every current-capability intake answer must lead with: - local compiler proven; live VPS/chat intake not shipped. Approval begins - before guarded canonical apply. + source identity. `teleo-kb propose-source` is shipped on the VPS for prepared + artifact/text/manifest inputs and can stage one deterministic `pending_review` + packet without apply authorization. It is not an autonomous Telegram + attachment or tweet extractor. Every current-capability intake answer must + lead with: VPS source-to-proposal staging is shipped for prepared inputs; + autonomous chat attachment extraction is not shipped. Approval begins before + guarded canonical apply. 5. **Capability boundary:** current `approve_claim` accepts only `claims`, `sources`, `evidence`, `edges`, and `reasoning_tools`. It cannot insert `behavioral_rules` or `governance_gates`, update beliefs or an existing claim, @@ -124,6 +126,24 @@ Use only the bounded follow-ups relevant to the question: /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb decision-matrix-status ``` +## Source To Proposal + +For a retained artifact with a strict UTF-8 extraction and hash-matched manifest, +stage exactly one reviewable proposal and retain its private receipt: + +```bash +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb propose-source \ + --artifact /path/to/source.bin \ + --text /path/to/source.txt \ + --manifest /path/to/source-manifest.json \ + --receipt /home/teleo/.hermes/profiles/leoclean/state/kb-source-receipts/source.json +``` + +The command is VPS-local, idempotent by deterministic proposal ID, and may write +only `kb_stage.kb_proposals`. Verify the receipt says `canonical counts +unchanged: true`, then use `show-proposal ` for readback. Never call this +proof of extraction from Telegram, reviewer approval, or canonical apply. + For direct state questions, run `status` plus the relevant proposal read. Copy a full proposal UUID when one row answers the question; otherwise copy all five observed counts: diff --git a/scripts/working_leo_m3taversal_oos_benchmark.py b/scripts/working_leo_m3taversal_oos_benchmark.py index 6593d5f..fb9e834 100755 --- a/scripts/working_leo_m3taversal_oos_benchmark.py +++ b/scripts/working_leo_m3taversal_oos_benchmark.py @@ -428,10 +428,11 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = { re.compile(r"not.{0,40}(?:a )?(?:source|provenance)|must not.{0,40}source|do not manufacture", re.I | re.S), ), "bounded_intake_tier": ( - re.compile(r"build-only|local(?:ly)?|clone", re.I), + re.compile(r"build-only|local(?:ly)?|clone|prepared inputs?|prepared artifact", re.I), re.compile( r"not yet.{0,80}(?:live|autonomous|production)|not.{0,80}live-VPS|" - r"live.{0,80}(?:not shipped|not live|unavailable)", + r"live.{0,80}(?:not shipped|not live|unavailable)|" + r"autonomous.{0,80}(?:not shipped|not live|unavailable)", re.I | re.S, ), ), diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index 3e3f6c8..d5efb37 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/tests/test_hermes_leoclean_db_context_plugin.py @@ -159,8 +159,8 @@ def test_source_validator_accepts_explicit_schema_and_canonical_write_prohibitio response = ( "Capture and hash the artifact without authorization, retain its URL or storage_path, and stage candidates " "in a pending_review proposal. Author, title, and publisher are not current public.sources columns. Do not " - "create public.sources or other canonical rows during staging. The local compiler is proven; live VPS/chat " - "intake is not shipped. Approval begins before canonical apply." + "create public.sources or other canonical rows during staging. VPS staging is shipped for prepared inputs; " + "autonomous chat attachment extraction is not shipped. Approval begins before canonical apply." ) contracts = [ {"id": "reply_budget", "hard_max_words": 200}, diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 5e6a58f..0c340f3 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -212,8 +212,11 @@ def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact() for item in module.operational_contracts("I hand Leo a document. What can it absorb and stage before approval?") } source = source_contracts["source_intake"] - assert source["required_lead"] == "Local compiler proven; live VPS/chat intake not shipped." - assert source["capability_tier"] == "build-local compiler only" + assert source["required_lead"] == ( + "VPS source-to-proposal staging is shipped for prepared inputs; autonomous chat attachment extraction " + "is not shipped." + ) + assert source["capability_tier"] == "VPS-local prepared-input compiler and pending-review staging" assert source["current_public_sources_columns"] == list(module.CURRENT_PUBLIC_SCHEMA["sources"]) assert "author/title/publisher/publication date are current public.sources columns" in source["must_not_claim"] @@ -570,7 +573,7 @@ def test_vps_bridge_context_markdown_surfaces_runtime_contracts(capsys) -> None: assert "## Current Runtime Contracts" in output assert '"id": "source_intake"' in output - assert "live VPS/chat intake not shipped" in output + assert "autonomous chat attachment extraction is not shipped" in output def test_vps_bridge_context_contract_reads_current_schema_from_postgres(monkeypatch) -> None: diff --git a/tests/test_hermes_leoclean_skill_surfaces.py b/tests/test_hermes_leoclean_skill_surfaces.py index 373e4f6..3e5c4c7 100644 --- a/tests/test_hermes_leoclean_skill_surfaces.py +++ b/tests/test_hermes_leoclean_skill_surfaces.py @@ -243,7 +243,7 @@ def test_vps_kb_skill_encodes_apply_readiness_beliefs_and_current_proof_boundari assert field in text assert "has no claim-ID foreign key" in squashed assert "neither is generic stance storage" in squashed - assert "source compiler is build-only and locally proven" in squashed + assert "`teleo-kb propose-source` is shipped on the VPS for prepared" in squashed assert "No active general DB-to-`SOUL.md` renderer automation is currently proven" in squashed assert "Unchanged table totals also do not prove unchanged rows" in squashed assert "current `public.claim_edges` has no rationale field" in squashed @@ -254,7 +254,9 @@ def test_vps_kb_skill_encodes_apply_readiness_beliefs_and_current_proof_boundari assert "`applied_at` belongs to the proposal receipt, not each inserted row" in squashed assert "`state.db` and session JSONL can preserve continuity across restart" in squashed assert "It has no `author`, `title`, `publisher`, or publication-date column" in squashed - assert "local compiler proven; live VPS/chat intake not shipped" in squashed + assert "autonomous chat attachment extraction is not shipped" in squashed + assert "--artifact /path/to/source.bin" in text + assert "--receipt /home/teleo/.hermes/profiles/leoclean/state/kb-source-receipts/source.json" in text assert "a reusable framework maps to `reasoning_tools`, not a claim graph" in squashed assert "never list them in the supported apply order" in squashed assert "Treat the emitted `Current Runtime Contracts` / `operational_contracts` block as binding" in squashed diff --git a/tests/test_kb_tool_propose_source.py b/tests/test_kb_tool_propose_source.py new file mode 100644 index 0000000..b2f83a3 --- /dev/null +++ b/tests/test_kb_tool_propose_source.py @@ -0,0 +1,246 @@ +from __future__ import annotations + +import hashlib +import importlib.util +import json +import re +import stat +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +KB_TOOL = ROOT / "hermes-agent" / "leoclean-bin" / "kb_tool.py" + + +def load_module(): + spec = importlib.util.spec_from_file_location("kb_tool_propose_source", KB_TOOL) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def source_bundle(module) -> dict: + payload = { + "apply_payload": { + "contract_version": 2, + "claims": [], + "sources": [], + "evidence": [], + "edges": [], + "reasoning_tools": [], + } + } + payload_sha256 = module.canonical_json_sha256(payload) + child = { + "id": "3cf4f2de-ea69-4e3b-a7ec-128b604d4f08", + "proposal_type": "approve_claim", + "status": "pending_review", + "proposed_by_handle": "leo", + "proposed_by_agent_id": None, + "channel": "source_document_compiler", + "source_ref": "normalized:source-parent:0123456789abcdef", + "rationale": "Strict normalized child of one hash-bound source packet.", + "payload": payload, + "payload_sha256": payload_sha256, + } + bundle = { + "schema": module.SOURCE_COMPILER_BUNDLE_SCHEMA, + "status": "pending_review", + "build_only": True, + "database_write_performed": False, + "canonical_apply_performed": False, + "strict_child_proposal": child, + "validated_manifest": { + "artifact_sha256": "a" * 64, + "extracted_text_sha256": "b" * 64, + "source": { + "identity": "document:sha256:" + "a" * 64, + "locator": "artifact://sha256/" + "a" * 64, + } + }, + "hashes": { + "artifact_sha256": "a" * 64, + "extracted_text_sha256": "b" * 64, + "manifest_canonical_sha256": "c" * 64, + "strict_child_payload_sha256": payload_sha256, + }, + } + bundle["hashes"]["bundle_content_sha256"] = module.canonical_json_sha256(bundle) + return bundle + + +def status_row(proposal_count: int) -> dict: + return { + "artifact": "teleo_kb_status", + "backend": "teleo|postgres", + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": proposal_count, + }, + "proposal_status_counts": {"pending_review": 14}, + } + + +def test_propose_source_stages_only_pending_review_and_writes_private_receipt( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + module = load_module() + bundle = source_bundle(module) + child = bundle["strict_child_proposal"] + compiler = tmp_path / "compile_kb_source_packet.py" + compiler.write_text("# compiler fixture\n", encoding="utf-8") + monkeypatch.setattr(module, "SOURCE_COMPILER_PATH", compiler) + monkeypatch.setattr(module, "compile_source_bundle", lambda _args: bundle) + + calls: list[str] = [] + status_reads = iter((status_row(26), status_row(27))) + + def fake_psql_json(_args, sql): + calls.append(sql) + if "insert into kb_stage.kb_proposals" not in sql: + return [next(status_reads)] + return [ + { + "created": True, + "proposal_id": child["id"], + "proposal_type": child["proposal_type"], + "status": "pending_review", + "source_ref": child["source_ref"], + "payload": child["payload"], + "payload_sha256": child["payload_sha256"], + "created_at": "2026-07-13T18:00:00+00:00", + "reviewed_at": None, + "applied_at": None, + } + ] + + monkeypatch.setattr(module, "psql_json", fake_psql_json) + receipt_path = tmp_path / "receipts" / "source.json" + args = SimpleNamespace( + local=True, + artifact=tmp_path / "source.pdf", + text=tmp_path / "source.txt", + manifest=tmp_path / "source-manifest.json", + receipt=receipt_path, + ) + + receipt = module.propose_source(args) + + assert receipt["status"] == "pending_review" + assert receipt["created"] is True + assert receipt["proposal"]["id"] == child["id"] + assert receipt["database"]["canonical_counts_unchanged"] is True + assert receipt["database"]["proposal_count_delta"] == 1 + assert receipt["database"]["proposal_count_matches_command"] is True + assert receipt["write_contract"] == { + "target_table": "kb_stage.kb_proposals", + "database_write_performed_by_command": True, + "canonical_apply_performed": False, + "production_db_apply_ran": False, + "public_table_write_path_present": False, + } + assert json.loads(receipt_path.read_text(encoding="utf-8")) == receipt + assert stat.S_IMODE(receipt_path.stat().st_mode) == 0o600 + + stage_sql = next(sql for sql in calls if "insert into kb_stage.kb_proposals" in sql) + assert "on conflict (id) do nothing" in stage_sql + assert "marker_count <> 1 or exact_count <> 1" in stage_sql + assert not re.search(r"\b(?:insert|update|delete)\s+(?:into\s+|from\s+)?public\.", stage_sql, re.I) + assert len(calls) == 3 + + +def test_source_bundle_rejects_tampered_child_payload() -> None: + module = load_module() + bundle = source_bundle(module) + bundle["strict_child_proposal"]["payload"]["apply_payload"]["claims"].append({"id": "tampered"}) + + with pytest.raises(SystemExit, match="payload hash does not match"): + module._validated_source_bundle(bundle) + + +def test_deployed_bridge_accepts_real_source_compiler_bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + module = load_module() + artifact_bytes = b"%PDF-1.7\nbridge compatibility fixture\n%%EOF\n" + extracted_text = ( + "Bridge compatibility source.\n\n" + "Prepared source inputs may become a pending review proposal.\n\n" + "Canonical rows still require review and guarded apply." + ) + artifact = tmp_path / "source.pdf" + text = tmp_path / "source.txt" + manifest_path = tmp_path / "source-manifest.json" + artifact.write_bytes(artifact_bytes) + text.write_text(extracted_text, encoding="utf-8") + artifact_sha256 = hashlib.sha256(artifact_bytes).hexdigest() + text_sha256 = hashlib.sha256(extracted_text.encode()).hexdigest() + manifest = { + "schema": "livingip.kbSourceExtractionManifest.v1", + "artifact_sha256": artifact_sha256, + "extracted_text_sha256": text_sha256, + "extractor": {"name": "bridge-test", "version": "1"}, + "source": { + "identity": f"document:sha256:{artifact_sha256}", + "source_key": "bridge_compatibility_source", + "source_type": "article", + "title": "Bridge compatibility source", + "locator": f"artifact://sha256/{artifact_sha256}", + }, + "claims": [ + { + "claim_key": "prepared_input_staging", + "type": "structural", + "text": "Prepared inputs can be staged for review without becoming canonical.", + "quote": "Prepared source inputs may become a pending review proposal.", + "confidence": 0.99, + "tags": ["source-intake"], + "evidence": [ + { + "quote": "Canonical rows still require review and guarded apply.", + "role": "grounds", + } + ], + } + ], + } + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + monkeypatch.setattr(module, "SOURCE_COMPILER_PATH", ROOT / "scripts" / "compile_kb_source_packet.py") + + bundle = module.compile_source_bundle(SimpleNamespace(artifact=artifact, text=text, manifest=manifest_path)) + child, validated_manifest = module._validated_source_bundle(bundle) + + assert child["status"] == "pending_review" + assert child["proposal_type"] == "approve_claim" + assert validated_manifest["artifact_sha256"] == artifact_sha256 + + +def test_propose_source_refuses_nonlocal_database_route(tmp_path: Path) -> None: + module = load_module() + args = SimpleNamespace( + local=False, + artifact=tmp_path / "source.pdf", + text=tmp_path / "source.txt", + manifest=tmp_path / "source-manifest.json", + receipt=tmp_path / "receipt.json", + ) + + with pytest.raises(SystemExit, match="VPS-local only"): + module.propose_source(args) + + +def test_source_stage_sql_allows_idempotent_readback_after_review_state_advances() -> None: + module = load_module() + child, _manifest = module._validated_source_bundle(source_bundle(module)) + + sql = module.build_source_stage_sql(child) + + assert "on conflict (id) do nothing" in sql + exact_clause = sql.split("select count(*) into exact_count", 1)[1].split("if marker_count", 1)[0] + assert "status = 'pending_review'" not in exact_clause + assert "reviewed_at is null" not in exact_clause + assert "applied_at is null" not in exact_clause diff --git a/tests/test_working_leo_m3taversal_oos_benchmark.py b/tests/test_working_leo_m3taversal_oos_benchmark.py index 4b2a712..c0dc6a8 100644 --- a/tests/test_working_leo_m3taversal_oos_benchmark.py +++ b/tests/test_working_leo_m3taversal_oos_benchmark.py @@ -615,9 +615,12 @@ def test_oos_schema_gap_accepts_explicitly_absent_edge_type() -> None: assert benchmark.current_schema_overclaims(reply) == [] -def test_oos_source_intake_accepts_live_not_shipped_wording() -> None: +def test_oos_source_intake_accepts_prepared_input_staging_boundary() -> None: patterns = benchmark.CONCEPT_PATTERNS["bounded_intake_tier"] - reply = "The local compiler is proven; live VPS/chat intake is not shipped." + reply = ( + "VPS source-to-proposal staging is shipped for prepared inputs; autonomous chat attachment extraction " + "is not shipped." + ) assert all(pattern.search(reply) for pattern in patterns) @@ -628,9 +631,8 @@ def test_oos_source_intake_accepts_without_authorization_and_created_by_column() "Immediately capture and hash the artifact without authorization, retain its URL or storage path, and stage " "candidates in a pending_review proposal. The real source identity is the URL, storage path, and content hash; " "a chat label or source_ref is not a source. created_by is metadata, not an instruction to create a chat label. " - "Approval begins at canonical apply where reviewed rows would write to public.*. The local compiler is " - "proven; live VPS/chat " - "intake is not shipped." + "Approval begins at canonical apply where reviewed rows would write to public.*. VPS source-to-proposal " + "staging is shipped for prepared inputs; autonomous chat attachment extraction is not shipped." ) score = benchmark.score_reply(prompt, reply, memory_token=token)