Stage source artifacts as reviewable KB proposals (#123)

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":

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(

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

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 <id>` 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:

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,
         ),
     ),

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},

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:

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

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

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)