Harden Leo DB retrieval and behavior attribution

.github/workflows/ci.yml

@@ -53,12 +53,15 @@ jobs:
             ops/verify_gcp_cloudsql_restore_readback.py \
             ops/verify_postgres_parity_manifest.py \
             telegram/approvals.py \
+            hermes-agent/leoclean-bin/kb_tool.py \
             scripts/check_crabbox_ci_contract.py \
             scripts/check_llm_refinement_contract.py \
             scripts/build_working_leo_m3taversal_outcome_sandbox.py \
+            scripts/leo_behavior_manifest.py \
             scripts/replay_decision_engine_eval.py \
             scripts/prove_phase1b_local.py \
             scripts/run_gcp_generated_db_direct_claim_suite.py \
+            scripts/run_leo_direct_claim_handler_suite.py \
             scripts/run_leo_m3taversal_oos_handler_suite.py \
             scripts/run_leo_clone_bound_handler_checkpoint.py \
             scripts/working_leo_m3taversal_oos_benchmark.py \
@@ -80,6 +83,10 @@ jobs:
             tests/test_gcp_iam_split_plan.py \
             tests/test_gcp_readiness_workflow.py \
             tests/test_gcp_generated_db_direct_claim_suite.py \
+            tests/test_hermes_leoclean_kb_bridge_source.py \
+            tests/test_hermes_leoclean_skill_surfaces.py \
+            tests/test_leo_behavior_manifest.py \
+            tests/test_compile_kb_source_packet.py \
             tests/test_verify_postgres_parity_manifest.py \
             tests/test_working_leo_m3taversal_oos_benchmark.py \
             tests/test_working_leo_open_ended_benchmark.py \

deploy/auto-deploy.sh

@@ -103,7 +103,7 @@ fi
 
 # Syntax check all Python files before copying
 ERRORS=0
-for f in lib/*.py *.py diagnostics/*.py telegram/*.py tests/*.py hermes-agent/leoclean-plugins/vps/*/*.py hermes-agent/patches/*.py scripts/compile_kb_source_packet.py scripts/kb_proposal_normalize.py scripts/prepare_kb_source_manifest.py; do
+for f in lib/*.py *.py diagnostics/*.py telegram/*.py tests/*.py hermes-agent/leoclean-bin/*.py hermes-agent/leoclean-plugins/vps/*/*.py hermes-agent/patches/*.py scripts/compile_kb_source_packet.py scripts/kb_proposal_normalize.py scripts/prepare_kb_source_manifest.py scripts/leo_behavior_manifest.py; do
   [ -f "$f" ] || continue
   if ! python3 -c "import ast, sys; ast.parse(open(sys.argv[1]).read())" "$f" 2>&1; then
     log "SYNTAX ERROR: $f"

deploy/deploy.sh

@@ -48,7 +48,7 @@ echo ""
 # Syntax check all Python files before deploying
 echo "=== Pre-deploy syntax check ==="
 ERRORS=0
-for f in "$REPO_ROOT/lib/"*.py "$REPO_ROOT/"*.py "$REPO_ROOT/diagnostics/"*.py "$REPO_ROOT/telegram/"*.py "$REPO_ROOT/hermes-agent/leoclean-plugins/vps/"*/*.py "$REPO_ROOT/hermes-agent/patches/"*.py "$REPO_ROOT/scripts/compile_kb_source_packet.py" "$REPO_ROOT/scripts/kb_proposal_normalize.py" "$REPO_ROOT/scripts/prepare_kb_source_manifest.py"; do
+for f in "$REPO_ROOT/lib/"*.py "$REPO_ROOT/"*.py "$REPO_ROOT/diagnostics/"*.py "$REPO_ROOT/telegram/"*.py "$REPO_ROOT/hermes-agent/leoclean-bin/"*.py "$REPO_ROOT/hermes-agent/leoclean-plugins/vps/"*/*.py "$REPO_ROOT/hermes-agent/patches/"*.py "$REPO_ROOT/scripts/compile_kb_source_packet.py" "$REPO_ROOT/scripts/kb_proposal_normalize.py" "$REPO_ROOT/scripts/prepare_kb_source_manifest.py" "$REPO_ROOT/scripts/leo_behavior_manifest.py"; do
   [ -f "$f" ] || continue
   if ! python3 -c "import ast, sys; ast.parse(open(sys.argv[1]).read())" "$f" 2>/dev/null; then
     echo "SYNTAX ERROR: $f"

docs/reports/leo-working-state-20260709/db-first-source-verification-canary-20260714.json

@@ -0,0 +1,86 @@
+{
+  "schema": "livingip.leoDbFirstRetrievalCanary.v1",
+  "mode": "read_only_candidate_tool_on_live_vps",
+  "recorded_at": "2026-07-14",
+  "query": "AI sandbagging M&A liability product liability consumer protection securities fraud",
+  "candidate_tool_sha256": "73de3cc641a72c00d39de7703b139daeecb562472eaf57e5f72ff1a4cd38a76b",
+  "target_claim_id": "2a7ae257-d01d-46f4-b813-63f81bb9c7c7",
+  "canonical_counts_before": {
+    "claims": 1837,
+    "sources": 4145,
+    "claim_evidence": 4670,
+    "claim_edges": 4916,
+    "kb_proposals": 29
+  },
+  "canonical_counts_after": {
+    "claims": 1837,
+    "sources": 4145,
+    "claim_evidence": 4670,
+    "claim_edges": 4916,
+    "kb_proposals": 29
+  },
+  "first_receipt": {
+    "schema": "livingip.teleoKbRetrievalReceipt.v1",
+    "semantic_context_sha256": "8333bc7a9dc2effe7d6fd135a95d9ab7822f9332f8b4019245c67c14efa38d9c",
+    "artifact_state_sha256": "47473661b1baa182d7fa2f73516a0e64415697d7ebc4a1cdeb2b59cbeb2356fc",
+    "counts": {
+      "claims": 4,
+      "context_rows": 0,
+      "evidence_rows": 14,
+      "edge_rows": 9
+    },
+    "read_consistency": {
+      "status": "stable_wal_marker",
+      "attempts": 1,
+      "wal_lsn_before": "0/49BC4478",
+      "wal_lsn_after": "0/49BC4478"
+    }
+  },
+  "second_receipt": {
+    "schema": "livingip.teleoKbRetrievalReceipt.v1",
+    "semantic_context_sha256": "8333bc7a9dc2effe7d6fd135a95d9ab7822f9332f8b4019245c67c14efa38d9c",
+    "artifact_state_sha256": "47473661b1baa182d7fa2f73516a0e64415697d7ebc4a1cdeb2b59cbeb2356fc",
+    "counts": {
+      "claims": 4,
+      "context_rows": 0,
+      "evidence_rows": 14,
+      "edge_rows": 9
+    },
+    "read_consistency": {
+      "status": "stable_wal_marker",
+      "attempts": 1,
+      "wal_lsn_before": "0/49BC4478",
+      "wal_lsn_after": "0/49BC4478"
+    }
+  },
+  "source_verification": [
+    {
+      "source_id": "261c3532-fa32-47d8-a5b5-6cc45035c267",
+      "role": "grounds",
+      "storage_path": null,
+      "expected_sha256": "9fcd40191449fe0843178e08120fb3c86fc2b35633594aee60e747ca3918cfa5",
+      "status": "no_source_pointer",
+      "artifact_exists": null,
+      "hash_matches_db": null
+    },
+    {
+      "source_id": "15740795-ecc6-40fa-9a01-3d6bc7c54f79",
+      "role": "illustrates",
+      "storage_path": "domains/ai-alignment/ai-sandbagging-creates-m-and-a-liability-exposure-across-product-liability-consumer-protection-and-securities-fraud.md",
+      "resolved_storage_path": "/opt/teleo-eval/workspaces/main/domains/ai-alignment/ai-sandbagging-creates-m-and-a-liability-exposure-across-product-liability-consumer-protection-and-securities-fraud.md",
+      "expected_sha256": "341a7685cd2909ae821d8e464fe0d4fdfbb0cadccf546ccfd4ccaf037aa9f646",
+      "artifact_sha256": "341a7685cd2909ae821d8e464fe0d4fdfbb0cadccf546ccfd4ccaf037aa9f646",
+      "status": "verified",
+      "artifact_exists": true,
+      "artifact_bytes": 2701,
+      "hash_matches_db": true
+    }
+  ],
+  "semantic_receipt_repeated": true,
+  "artifact_receipt_repeated": true,
+  "production_mutation_performed": false,
+  "gateway_pid_unchanged": true,
+  "gateway_restart_count_before": 0,
+  "gateway_restart_count_after": 0,
+  "temp_candidate_removed": true
+}

docs/reports/leo-working-state-20260709/working-leo-db-first-behavior-model-20260714.md

@@ -0,0 +1,151 @@
+# What Makes Leo Leo: DB-First Behavior Model
+
+Date: `2026-07-14`
+
+## Definition Of Working
+
+**Working target:** Leo forms durable knowledge through a replayable database lifecycle, and answers through consistent, traceable database retrieval. A conversation may become a source candidate; it does not silently train Leo or become truth.
+
+**Operator path:** retain source -> extract candidates -> deduplicate/contrast -> stage proposal -> review -> guarded apply -> canonical readback -> retrieve with a source receipt.
+
+**Done means:** the same retained inputs, compiler version, approved proposal, database snapshot, retrieval query, and runtime manifest reproduce the same rows and retrieval receipt. The benchmark can measure the outcome without changing Leo's live profile or canonical database.
+
+**Not done:** a chat answer looks better because `SOUL.md`, `USER.md`, a skill, a plugin response template, old session state, or benchmark-specific text changed without a reviewed database change.
+
+**Required tier:** live VPS database/tool lifecycle first; GCP reproduces the same manifest and receipts before promotion.
+
+## One-Line Model
+
+Leo is an external language model run by Hermes, wrapped in the `leoclean` profile, given static identity and procedural layers, dynamic session context, and database retrieval. Only reviewed canonical database changes should count as durable learning.
+
+```mermaid
+flowchart LR
+  U["Person or agent discussion"] --> A["Immutable source artifact + stable message locator"]
+  A --> X["Deterministic extraction and canonical dedupe"]
+  X --> P["Pending review proposal"]
+  P --> R["Human or governed review"]
+  R --> C["Canonical Postgres rows"]
+  C --> T["Deterministic retrieval tool + receipt"]
+  T --> H["Hermes reasoning turn"]
+  S["SOUL + config + skills"] --> H
+  M["Session continuity"] --> H
+  G["Generic safety middleware"] --> H
+  B["External benchmark"] -. "measure only" .-> H
+  B -. "must not write" .-> C
+  B -. "must not edit" .-> S
+```
+
+## Current Live Layers
+
+Fresh readback on `2026-07-14` found the VPS service active with `NRestarts=0` and Hermes Agent `v0.7.0` at commit `b2f477a30b3c05d0f383c543af98496ae8a96070`.
+
+| Layer | What it does | Change rate | How visible/replayable it is | Current assessment |
+| --- | --- | --- | --- | --- |
+| Base-model weights | Supply general language and reasoning capability | Provider-controlled | Opaque; no local weight hash | No evidence Leo was fine-tuned. The configured model name is not a weight-level identity. |
+| Model routing/config | Selects provider/model, routing, token/reasoning defaults, tools, and memory | Deployment-static | File-hashable; actual per-turn model still needs a receipt | OpenRouter, default `anthropic/claude-sonnet-4-6`, smart routing enabled through `google/gemini-2.5-flash`. Temperature/reasoning effort are not explicitly pinned in the observed config. |
+| Hermes runtime | Builds prompts, sessions, tools, gateway calls, and plugin hooks | Deployment-static | Source and commit are inspectable | Generic upstream runtime, but locally patched to allow post-model response replacement before persistence. |
+| `SOUL.md` | Primary runtime identity text loaded into the system prompt | Static until manually edited or rendered | Fully hashable; current SHA-256 `5973a54c...32a` | Important runtime input, but not canonical truth. No active general DB-to-SOUL renderer is proven. |
+| Skills | Procedural instructions discoverable by Hermes | Deployment-static | Source-controlled and hashable | Useful for tool use, but detailed answer rules can become hidden prompt training. |
+| `leo-db-context` plugin | Injects pre-turn DB contracts, validates drafts, and can replace the final answer | Request-dynamic over static code | Source-controlled but invisible in normal chat | The most behaviorally powerful and obscure layer. Current query-specific compiled answers are a temporary guardrail, not database reasoning. |
+| KB bridge/tool code | Queries claims/context/evidence/edges and stages proposals | Deployment-static code over dynamic DB | Source-controlled; calls can emit receipts | Correct architectural control point. This branch adds allowlisted artifact resolution, hash verification, and deterministic context receipts on VPS. |
+| Canonical Postgres | Stores applied claims, sources, evidence, graph, and identity/strategy rows | Governed dynamic | Row IDs, hashes, manifests, and apply receipts | Intended source of durable knowledge. |
+| `kb_stage` proposals | Holds candidates awaiting review/apply | Dynamic, review-gated | Queryable and replayable | Correct place for discussion-derived candidates before approval. |
+| `MEMORY.md` / `USER.md` | Persistent Hermes facts/preferences injected outside Postgres | Conversation-dynamic | File-hashable but provenance-poor | Noncanonical. Live `USER.md` contains a stale naming rule that conflicts with the exact visible-handle rule. |
+| Sessions / `state.db` | Preserve conversation continuity | Turn-dynamic | Hashable but highly stateful | Appropriate for “what did we just discuss,” not durable collective knowledge. |
+| Benchmark harness | Sends test turns and scores replies | External | Fully inspectable if isolated | Must remain measurement-only. The prior temp-profile harness copied live memories and sessions, weakening causal attribution. |
+
+## What “Weights” Can Mean
+
+1. **Base-model weights:** opaque parameters hosted by the model provider. We do not inspect or modify them.
+2. **Database weights:** explicit `claim_evidence.weight`, `claim_edges.weight`, and strategy-anchor weights. These are reviewable rows and belong in database receipts.
+3. **Retrieval scores:** deterministic query-term matching currently ranks claims using term hits, evidence counts, edge counts, confidence, and text length. These are code-level ranking rules, not learning.
+4. **Benchmark scores:** regex/rubric checks outside Leo decide whether a response passed. They must never feed edits back into the runtime automatically.
+5. **Plugin contracts:** not weights at all. They are hard rules that can replace a response. They currently explain part of the measured improvement and must be separated from raw model-plus-database capability.
+
+## Desired Static/Dynamic Boundary
+
+### Keep static and small
+
+- Hermes runtime version and patches.
+- Model routing and explicit inference parameters.
+- A minimal `SOUL.md`: stable identity, mission, voice, and immutable safety boundaries.
+- Generic skills describing tools and schemas.
+- Generic middleware: fail closed when live retrieval fails, prevent unauthorized writes, validate receipts.
+
+### Keep dynamic and canonical
+
+- Claims and claim state.
+- Sources, content hashes, exact excerpts, and source classification.
+- Claim-to-evidence and claim-to-claim relationships, including explicit weights.
+- Agent beliefs, strategies, identity graph rows, governance rules, and reasoning tools where supported by schema/apply contracts.
+- Candidate proposals, reviews, apply receipts, and supersession history.
+
+### Keep dynamic but noncanonical
+
+- The current conversation and same-session references.
+- Delivery metadata and Telegram message IDs.
+- Temporary plans/tool traces.
+
+These may help the current turn but must not become evidence or durable identity without source capture and review.
+
+## Conversation Learning Contract
+
+1. A discussion receives a stable locator such as a Telegram chat/message ID plus capture time and content hash.
+2. The retained artifact is immutable. A display name or chat label alone is rejected as provenance.
+3. Extraction produces typed candidate claims, evidence excerpts, relationships, and explicit uncertainty.
+4. Existing canonical rows are retrieved first to identify duplicates, conflicts, and missing evidence.
+5. The result enters `kb_stage.kb_proposals` as `pending_review`; no canonical row and no runtime identity file changes yet.
+6. Review may approve, revise, split, or reject each candidate.
+7. Guarded apply writes the supported canonical rows and emits row-level postflight proof.
+8. Later retrieval verifies the linked artifact and database hash, then emits a deterministic receipt for the context Leo received.
+
+An agent can act as the reviewing interlocutor, but its review identity, policy, and decision must be recorded. Agent discussion is not self-authenticating approval.
+
+## Harness Contract
+
+The harness must:
+
+- start from a fresh session and empty persistent memory unless memory itself is the tested dimension;
+- record the model route, Hermes commit, config/SOUL/skill/plugin/tool hashes, and database fingerprint;
+- compare live profile and database fingerprints before and after;
+- write only inside an isolated temporary profile or disposable database;
+- never update `SOUL.md`, `MEMORY.md`, `USER.md`, skills, plugins, or live Postgres as a consequence of a score;
+- keep a frozen regression set and a separate blind set so passing known prompts cannot masquerade as general reasoning;
+- score raw reasoning separately from middleware-composed or replaced responses.
+
+## Delivery Status And Repair Order
+
+1. **Implemented in this branch - remove causal ambiguity from tests:** exclude prior memories/sessions from temporary profiles and fingerprint every behavior-bearing layer.
+2. **Implemented and live-canary proven on VPS - complete retrieval receipts:** resolve source paths, verify content hashes, stabilize row ordering, and hash the exact context bundle.
+3. **Bounded compiler behavior proven locally - make discussion ingestion real:** stable message-level locators can produce pending-review proposals without profile edits, but automatic Telegram capture is not shipped.
+4. **Remaining - reduce query-specific middleware:** keep generic truth and authorization guards, then measure raw DB reasoning separately from compiled response replacement.
+5. **Remaining - make identity composition deterministic:** implement and receipt a DB-to-`SOUL.md` renderer before calling identity updates live.
+6. **Remaining - pin the execution identity:** record the actual provider/model and explicit inference settings for each benchmark turn.
+7. **Remaining - reproduce on GCP:** restore the same canonical manifest, deploy the same behavior manifest, replay the same lifecycle, and only then consider promotion.
+
+This design makes a behavior change attributable: either a reviewed DB delta, a versioned runtime delta, a model-route delta, or current-session context. “Leo learned it somehow” is not an acceptable state.
+
+## Live VPS Read-Only Canary
+
+The candidate retrieval path was exercised twice against the live VPS database
+without deploying it, restarting Leo, or writing to Postgres. An unfamiliar
+query selected claim `2a7ae257-d01d-46f4-b813-63f81bb9c7c7` and its two evidence
+rows.
+
+- One source has a database hash but no URL or storage pointer. The receipt
+  reports `no_source_pointer`; it does not misstate that the canonical evidence
+  row or raw artifact is missing.
+- The second source uses a relative storage path. The tool resolved the path
+  under an allowlisted workspace root, found the artifact, calculated SHA-256
+  `341a7685...9f646`, and matched the database hash exactly.
+- Both reads emitted semantic context hash `8333bc7a...38d9c` and artifact-state
+  hash `47473661...356fc` with a stable WAL marker. Each read returned four
+  claims, fourteen evidence rows, and nine edges in the same order.
+- Before/after counts stayed at 1,837 claims, 4,145 sources, 4,670 evidence
+  links, 4,916 claim edges, and 29 proposals. The gateway PID and restart count
+  were unchanged, and the temporary candidate was removed.
+
+The machine-readable receipt is
+[`db-first-source-verification-canary-20260714.json`](./db-first-source-verification-canary-20260714.json).
+This proves deterministic read and artifact verification on VPS. It does not
+yet prove a Telegram-visible answer, GCP parity, or an approved canonical write.

hermes-agent/leoclean-bin/kb_tool.py

@@ -24,6 +24,7 @@ import uuid
 from collections import defaultdict
 from pathlib import Path
 from typing import Any
+from urllib.parse import urlsplit
 
 DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz"
 SOURCE_COMPILER_PATH = Path("/opt/teleo-eval/workspaces/deploy-infra/scripts/compile_kb_source_packet.py")
@@ -33,8 +34,17 @@ SOURCE_PREPARATION_ROOT = Path("/home/teleo/.hermes/profiles/leoclean/state/kb-s
 SOURCE_PROPOSAL_RECEIPT_SCHEMA = "livingip.teleoKbSourceProposalReceipt.v1"
 SOURCE_PREPARATION_RECEIPT_SCHEMA = "livingip.kbSourcePreparationReceipt.v1"
 SOURCE_COMPILER_BUNDLE_SCHEMA = "livingip.kbSourceProposalBundle.v1"
+RETRIEVAL_RECEIPT_SCHEMA = "livingip.teleoKbRetrievalReceipt.v1"
 CANONICAL_COUNT_LABELS = ("claims", "sources", "claim_edges", "claim_evidence")
 SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+DEFAULT_MAX_SOURCE_ARTIFACT_BYTES = 64 * 1024 * 1024
+DEFAULT_SOURCE_ARTIFACT_ROOTS = (
+    Path("/opt/teleo-eval/workspaces/main"),
+    Path("/opt/teleo-eval/workspaces/deploy-infra"),
+    Path("/home/teleo/.hermes/profiles/leoclean/workspace/ingested"),
+    Path("/home/teleo/.hermes/profiles/leoclean/state/kb-source-inbox"),
+    Path("/home/teleo/.hermes/profiles/leoclean/state/kb-source-preparation"),
+)
 WORKER_SUPPORTED_PROPOSAL_TYPES = frozenset({"revise_strategy", "add_edge", "attach_evidence", "approve_claim"})
 DIRECT_READBACK_CONTRACTS = frozenset(
     {
@@ -1703,7 +1713,7 @@ def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[s
            )::text
       from ranked
      where score >= {min_score}
-     order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text)
+     order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text), text, id
      limit {limit};
     """
     return psql_json(args, sql)
@@ -1812,7 +1822,7 @@ def find_context_rows(args: argparse.Namespace, query: str, limit: int) -> list[
            )::text
       from ranked
      where score >= {min_score}
-     order by score desc, source, owner, title
+     order by score desc, source, owner, title, body
      limit {limit};
     """
     return psql_json(args, sql)
@@ -1843,6 +1853,166 @@ def canonical_json_sha256(value: Any) -> str:
     return hashlib.sha256(encoded).hexdigest()
 
 
+def source_artifact_roots() -> tuple[Path, ...]:
+    raw = os.environ.get("TELEO_KB_SOURCE_ROOTS", "").strip()
+    configured = tuple(Path(item).expanduser() for item in raw.split(os.pathsep) if item.strip())
+    roots = configured or DEFAULT_SOURCE_ARTIFACT_ROOTS
+    resolved: list[Path] = []
+    for root in roots:
+        if not root.is_absolute():
+            raise SystemExit(f"TELEO_KB_SOURCE_ROOTS entries must be absolute: {root}")
+        candidate = root.resolve(strict=False)
+        if candidate not in resolved:
+            resolved.append(candidate)
+    return tuple(resolved)
+
+
+def _normalized_sha256(value: Any) -> str | None:
+    text = str(value or "").strip().lower()
+    if text.startswith("sha256:"):
+        text = text.removeprefix("sha256:")
+    return text if SHA256_RE.fullmatch(text) else None
+
+
+def source_artifact_max_bytes() -> int:
+    raw = os.environ.get("TELEO_KB_MAX_ARTIFACT_BYTES", str(DEFAULT_MAX_SOURCE_ARTIFACT_BYTES))
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise SystemExit("TELEO_KB_MAX_ARTIFACT_BYTES must be an integer") from exc
+    if not 1 <= value <= 1024 * 1024 * 1024:
+        raise SystemExit("TELEO_KB_MAX_ARTIFACT_BYTES must be between 1 byte and 1 GiB")
+    return value
+
+
+def _stable_file_sha256(path: Path) -> tuple[str, bool]:
+    before = path.stat()
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    after = path.stat()
+    stable = (before.st_ino, before.st_size, before.st_mtime_ns) == (after.st_ino, after.st_size, after.st_mtime_ns)
+    return digest.hexdigest(), stable
+
+
+def verify_source_artifact(row: dict[str, Any], roots: tuple[Path, ...] | None = None) -> dict[str, Any]:
+    """Resolve one source pointer under explicit roots and verify its content hash."""
+
+    storage_path = str(row.get("storage_path") or "").strip()
+    url = str(row.get("url") or "").strip()
+    expected_sha256 = _normalized_sha256(row.get("source_hash"))
+    base: dict[str, Any] = {
+        "path_kind": "none",
+        "artifact_exists": None,
+        "artifact_sha256": None,
+        "artifact_bytes": None,
+        "expected_sha256": expected_sha256,
+        "hash_matches_db": None,
+        "resolved_storage_path": None,
+        "matched_root": None,
+    }
+    if not storage_path:
+        return {**base, "status": "remote_locator_only" if url else "no_source_pointer"}
+
+    scheme = urlsplit(storage_path).scheme.lower()
+    if scheme:
+        return {**base, "status": "locator_only", "path_kind": "uri", "locator_scheme": scheme}
+
+    allowed_roots = tuple(path.resolve(strict=False) for path in (roots or source_artifact_roots()))
+    pointer = Path(storage_path)
+    candidates: list[tuple[Path, Path]] = []
+    if pointer.is_absolute():
+        resolved = pointer.resolve(strict=False)
+        for root in allowed_roots:
+            if resolved.is_relative_to(root):
+                candidates.append((resolved, root))
+                break
+        path_kind = "absolute"
+    else:
+        path_kind = "relative"
+        for root in allowed_roots:
+            resolved = (root / pointer).resolve(strict=False)
+            if resolved.is_relative_to(root):
+                candidates.append((resolved, root))
+
+    if not candidates:
+        return {
+            **base,
+            "status": "rejected_outside_allowlisted_roots",
+            "path_kind": path_kind,
+            "artifact_exists": False,
+        }
+
+    existing = sorted(
+        ((candidate, root) for candidate, root in candidates if candidate.is_file()), key=lambda pair: str(pair[0])
+    )
+    if not existing:
+        return {
+            **base,
+            "status": "missing",
+            "path_kind": path_kind,
+            "artifact_exists": False,
+            "candidate_count": len(candidates),
+        }
+    if len(existing) > 1:
+        return {
+            **base,
+            "status": "ambiguous_multiple_files",
+            "path_kind": path_kind,
+            "artifact_exists": True,
+            "candidate_count": len(existing),
+        }
+
+    resolved, matched_root = existing[0]
+    try:
+        artifact_bytes = resolved.stat().st_size
+        max_artifact_bytes = source_artifact_max_bytes()
+        if artifact_bytes > max_artifact_bytes:
+            return {
+                **base,
+                "status": "too_large_to_verify",
+                "path_kind": path_kind,
+                "artifact_exists": True,
+                "artifact_bytes": artifact_bytes,
+                "max_artifact_bytes": max_artifact_bytes,
+                "resolved_storage_path": str(resolved),
+                "matched_root": str(matched_root),
+            }
+        artifact_sha256, stable = _stable_file_sha256(resolved)
+    except OSError as exc:
+        return {
+            **base,
+            "status": "unreadable",
+            "path_kind": path_kind,
+            "artifact_exists": True,
+            "verification_error": type(exc).__name__,
+        }
+    if not stable:
+        status = "changed_during_verification"
+        hash_matches = None
+    elif expected_sha256 is None:
+        status = "verified_hash_unavailable"
+        hash_matches = None
+    elif artifact_sha256 == expected_sha256:
+        status = "verified"
+        hash_matches = True
+    else:
+        status = "hash_mismatch"
+        hash_matches = False
+    return {
+        **base,
+        "status": status,
+        "path_kind": path_kind,
+        "artifact_exists": True,
+        "artifact_sha256": artifact_sha256,
+        "artifact_bytes": artifact_bytes,
+        "hash_matches_db": hash_matches,
+        "resolved_storage_path": str(resolved),
+        "matched_root": str(matched_root),
+    }
+
+
 def _validated_uuid(value: Any, label: str) -> str:
     try:
         return str(uuid.UUID(str(value)))
@@ -2643,16 +2813,20 @@ def load_evidence(args: argparse.Namespace, claim_ids: list[str], limit: int) ->
       select ce.claim_id,
              ce.role::text as role,
              ce.weight,
+             s.id as source_id,
              s.source_type,
              s.url,
              s.storage_path,
+             s.hash as source_hash,
+             s.captured_at,
              left(coalesce(s.excerpt, ''), 800) as excerpt,
              row_number() over (
                partition by ce.claim_id
                order by (s.storage_path like 'inbox/archive/%') desc,
                         ce.role::text,
                         s.storage_path nulls last,
-                        s.url nulls last
+                        s.url nulls last,
+                        s.id
              ) as rn
         from claim_evidence ce
         join sources s on s.id = ce.source_id
@@ -2662,9 +2836,12 @@ def load_evidence(args: argparse.Namespace, claim_ids: list[str], limit: int) ->
              'claim_id', claim_id::text,
              'role', role,
              'weight', weight,
+             'source_id', source_id::text,
              'source_type', source_type,
              'url', url,
              'storage_path', storage_path,
+             'source_hash', source_hash,
+             'captured_at', captured_at::text,
              'excerpt', excerpt
            )::text
       from ranked
@@ -2672,7 +2849,12 @@ def load_evidence(args: argparse.Namespace, claim_ids: list[str], limit: int) ->
      order by claim_id, rn;
     """
     grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    verification_cache: dict[tuple[Any, ...], dict[str, Any]] = {}
     for row in psql_json(args, sql):
+        cache_key = (row.get("source_id"), row.get("storage_path"), row.get("url"), row.get("source_hash"))
+        if cache_key not in verification_cache:
+            verification_cache[cache_key] = verify_source_artifact(row)
+        row["artifact_verification"] = verification_cache[cache_key]
         grouped[row["claim_id"]].append(row)
     return grouped
 
@@ -2705,7 +2887,8 @@ def load_edges(args: argparse.Namespace, claim_ids: list[str], limit: int) -> di
                           when 'accelerates' then 8
                           else 9
                         end,
-                        other.text
+                        other.text,
+                        other.id
              ) as rn
         from base
         join claim_edges e on e.from_claim = base.id or e.to_claim = base.id
@@ -2729,7 +2912,95 @@ def load_edges(args: argparse.Namespace, claim_ids: list[str], limit: int) -> di
     return grouped
 
 
-def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit: int) -> dict[str, Any]:
+def database_read_marker(args: argparse.Namespace) -> dict[str, Any]:
+    sql = """
+    select jsonb_build_object(
+             'database', current_database(),
+             'database_user', current_user,
+             'system_identifier', (select system_identifier::text from pg_control_system()),
+             'wal_lsn', pg_current_wal_lsn()::text,
+             'transaction_snapshot', txid_current_snapshot()::text
+           )::text;
+    """
+    rows = psql_json(args, sql)
+    if not rows:
+        raise SystemExit("No database read marker returned.")
+    return rows[0]
+
+
+def _read_marker_stable(before: dict[str, Any], after: dict[str, Any]) -> bool:
+    keys = ("database", "database_user", "system_identifier", "wal_lsn")
+    return all(before.get(key) == after.get(key) for key in keys)
+
+
+def _stable_receipt_value(value: Any) -> Any:
+    volatile_or_environmental = {
+        "claim_url",
+        "matched_root",
+        "resolved_storage_path",
+        "snapshot_id",
+        "statement_timestamp",
+    }
+    if isinstance(value, dict):
+        return {
+            key: _stable_receipt_value(item)
+            for key, item in sorted(value.items())
+            if key not in volatile_or_environmental
+        }
+    if isinstance(value, list):
+        return [_stable_receipt_value(item) for item in value]
+    return value
+
+
+def build_retrieval_receipt(
+    data: dict[str, Any],
+    *,
+    before: dict[str, Any],
+    after: dict[str, Any],
+    attempts: int,
+    consistency_status: str,
+) -> dict[str, Any]:
+    semantic_context = _stable_receipt_value(data)
+    evidence_rows = [evidence for claim in data.get("claims", []) for evidence in claim.get("evidence", [])]
+    source_ids = sorted({str(row["source_id"]) for row in evidence_rows if row.get("source_id")})
+    artifact_states = sorted(
+        (
+            {
+                "source_id": row.get("source_id"),
+                "source_hash": row.get("source_hash"),
+                "storage_path": row.get("storage_path"),
+                "artifact_verification": _stable_receipt_value(row.get("artifact_verification") or {}),
+            }
+            for row in evidence_rows
+        ),
+        key=lambda row: (str(row.get("source_id") or ""), str(row.get("storage_path") or "")),
+    )
+    return {
+        "schema": RETRIEVAL_RECEIPT_SCHEMA,
+        "query_sha256": hashlib.sha256(str(data.get("query") or "").encode("utf-8")).hexdigest(),
+        "semantic_context_sha256": canonical_json_sha256(semantic_context),
+        "artifact_state_sha256": canonical_json_sha256(artifact_states),
+        "claim_ids": [str(claim.get("id")) for claim in data.get("claims", []) if claim.get("id")],
+        "source_ids": source_ids,
+        "counts": {
+            "claims": len(data.get("claims", [])),
+            "context_rows": len(data.get("context_rows", [])),
+            "evidence_rows": len(evidence_rows),
+            "edge_rows": sum(len(claim.get("edges", [])) for claim in data.get("claims", [])),
+        },
+        "read_consistency": {
+            "status": consistency_status,
+            "attempts": attempts,
+            "database": before.get("database"),
+            "database_user": before.get("database_user"),
+            "system_identifier": before.get("system_identifier"),
+            "wal_lsn_before": before.get("wal_lsn"),
+            "wal_lsn_after": after.get("wal_lsn"),
+        },
+    }
+
+
+def _bundle_once(args: argparse.Namespace, query: str, claim_limit: int, context_limit: int) -> dict[str, Any]:
     contracts = context_operational_contracts(args, query)
     claims = find_claims(args, query, claim_limit)
     claim_ids = [claim["id"] for claim in claims]
@@ -2753,6 +3024,43 @@ def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit
     }
 
 
+def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit: int) -> dict[str, Any]:
+    try:
+        max_attempts = max(1, min(int(os.environ.get("TELEO_KB_READ_ATTEMPTS", "3")), 5))
+    except ValueError as exc:
+        raise SystemExit("TELEO_KB_READ_ATTEMPTS must be an integer") from exc
+    previous_semantic_sha256: str | None = None
+    last_markers: tuple[dict[str, Any], dict[str, Any]] | None = None
+    for attempt in range(1, max_attempts + 1):
+        before = database_read_marker(args)
+        data = _bundle_once(args, query, claim_limit, context_limit)
+        after = database_read_marker(args)
+        last_markers = (before, after)
+        consistency_status = "stable_wal_marker" if _read_marker_stable(before, after) else "wal_changed_during_read"
+        receipt = build_retrieval_receipt(
+            data,
+            before=before,
+            after=after,
+            attempts=attempt,
+            consistency_status=consistency_status,
+        )
+        semantic_sha256 = receipt["semantic_context_sha256"]
+        if consistency_status == "stable_wal_marker":
+            data["retrieval_receipt"] = receipt
+            return data
+        if previous_semantic_sha256 == semantic_sha256:
+            receipt["read_consistency"]["status"] = "stable_content_across_wal_change_retry"
+            data["retrieval_receipt"] = receipt
+            return data
+        previous_semantic_sha256 = semantic_sha256
+    assert last_markers is not None
+    before, after = last_markers
+    raise SystemExit(
+        "Canonical DB changed during context retrieval and no identical retry was observed "
+        f"(wal_lsn {before.get('wal_lsn')} -> {after.get('wal_lsn')})."
+    )
+
+
 def print_json(data: Any) -> None:
     print(json.dumps(data, indent=2, sort_keys=True))
 
@@ -2772,6 +3080,14 @@ def print_status(data: dict[str, Any]) -> None:
 
 def print_claim_bundle(data: dict[str, Any]) -> None:
     print(f"# Teleo KB Context\n\nQuery: {data['query']}\n")
+    receipt = data.get("retrieval_receipt") or {}
+    if receipt:
+        consistency = receipt.get("read_consistency") or {}
+        print("## Retrieval Receipt\n")
+        print(f"- schema: `{receipt.get('schema')}`")
+        print(f"- semantic context SHA-256: `{receipt.get('semantic_context_sha256')}`")
+        print(f"- artifact state SHA-256: `{receipt.get('artifact_state_sha256')}`")
+        print(f"- DB read consistency: `{consistency.get('status')}`; attempts: `{consistency.get('attempts')}`\n")
     contracts = data.get("operational_contracts") or []
     if contracts:
         print("## Current Runtime Contracts\n")
@@ -2804,6 +3120,12 @@ def print_claim_bundle(data: dict[str, Any]) -> None:
                 source = ev.get("storage_path") or ev.get("url") or "(no source pointer)"
                 excerpt = f" - {truncate(ev.get('excerpt'), 260)}" if ev.get("excerpt") else ""
                 print(f"- `{ev['role']}` / `{ev.get('source_type')}` / `{source}`{excerpt}")
+                verification = ev.get("artifact_verification") or {}
+                print(
+                    f"  artifact verification: `{verification.get('status', 'not_checked')}`; "
+                    f"hash matches DB: `{verification.get('hash_matches_db')}`; "
+                    f"source id: `{ev.get('source_id') or '-'}`"
+                )
         if claim.get("edges"):
             print("\nEdges:")
             for edge in claim["edges"]:
@@ -2930,15 +3252,7 @@ def main() -> int:
     args = parse_args()
     if args.command == "status":
         data = status(args)
-    elif args.command == "search":
+    elif args.command in {"search", "context"}:
-        data = {
-            "query": args.query,
-            "terms": query_terms(args.query),
-            "operational_contracts": context_operational_contracts(args, args.query),
-            "claims": find_claims(args, args.query, args.limit),
-            "context_rows": find_context_rows(args, args.query, args.context_limit),
-        }
-    elif args.command == "context":
         data = bundle(args, args.query, args.limit, args.context_limit)
     elif args.command == "show":
         claim = get_claim(args, args.claim_id)

hermes-agent/leoclean-skills/vps/teleo-kb-bridge/REFERENCE.md

@@ -48,12 +48,16 @@ Apply these gates before drafting the answer or proposing a next action:
    payload and review that new packet; never tell the operator to apply the
    legacy row. `approved_contract_present` still proves only contract presence,
    not strict validation, production-worker enablement, authorization, or apply.
-2. **Source/evidence audit:** `teleo-kb search` searches claims and identity
+2. **Source/evidence audit:** `teleo-kb search` and `teleo-kb context` emit the
-   context, not canonical source rows. Use `evidence <claim_id>` or a bounded
+   selected claims, evidence, edges, and a deterministic retrieval receipt. Read
-   read-only `public.claim_evidence` to `public.sources` join. A missing locator
+   each evidence row's `artifact_verification` before describing provenance.
-   means weak provenance, but an existing canonical link is still canonical
+   Never infer that an artifact is missing from a raw `storage_path`; use the
-   evidence. A proposal's null `applied_at` proves only that proposal did not
+   reported status. `no_source_pointer` means the canonical source row has no
-   apply; it does not prove matching canonical rows were never written elsewhere.
+   resolvable locator, while `missing`, `hash_mismatch`, and `verified` describe
+   distinct artifact states. An existing claim-evidence link remains canonical
+   evidence when raw-artifact verification is unavailable. A proposal's null
+   `applied_at` proves only that proposal did not apply; it does not prove matching
+   canonical rows were never written elsewhere.
 3. **Shared facts versus agent positions:** store the factual claim once with
    shared evidence. Current agent-owned positions live in `public.beliefs` with
    `id`, `agent_id`, `level`, `statement`, `confidence`, `falsifier`, `rank`,
@@ -109,6 +113,13 @@ Treat the emitted `Current Runtime Contracts` / `operational_contracts` block as
 binding deployed-code readback. Follow its `required_lead`, schema guards,
 capability tier, and reply budget. It overrides recalled or generic architecture.
 
+For `search` and `context`, cite the receipt's `semantic_context_sha256` when an
+exact repeatable read matters. `read_consistency.status=stable_wal_marker` means
+the read had a stable WAL marker: the database position did not move during the
+read. If it did move, the bridge
+retries and accepts only an identical semantic context; otherwise it fails rather
+than combining rows from ambiguous database states.
+
 The profile's `leo-db-context` plugin automatically uses one immutable contract
 snapshot for generation and pre-delivery validation. When a covered draft
 violates the snapshot, `kb_tool.py` compiles the delivered response from the
@@ -255,6 +266,14 @@ Describe a missing locator as weak or citation-only provenance, or say the
 evidence is not traceable to the raw artifact. Do not call the canonical link
 non-canonical or ungrounded solely because the locator is missing.
 
+The bridge's `artifact_verification.status` distinguishes `verified`,
+`verified_hash_unavailable`, `hash_mismatch`, `missing`, `locator_only`,
+`no_source_pointer`, `remote_locator_only`,
+`rejected_outside_allowlisted_roots`, `ambiguous_multiple_files`,
+`too_large_to_verify`, `unreadable`, and `changed_during_verification`. Never infer `missing` from a raw
+`storage_path` without this verification result. A verified database link and a
+verified raw artifact are related but separate proof statements.
+
 A Telegram attachment, extracted file, or proposal `source_ref` does not by
 itself prove canonical evidence from that attachment. That proof requires a
 `public.sources` row representing the attachment and a `public.claim_evidence`

hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md

@@ -42,12 +42,16 @@ Apply these before drafting the answer or proposing a next action:
    that new packet; never tell the operator to apply the legacy row.
    `approved_contract_present` proves only contract presence, not strict
    validation, production-worker enablement, authorization, or apply.
-2. **Source/evidence audit:** `teleo-kb search` searches claims and identity
+2. **Source/evidence audit:** `teleo-kb search` and `teleo-kb context` emit the
-   context, not canonical source rows. Use `evidence <claim_id>` or a bounded
+   selected claims, evidence, edges, and a deterministic retrieval receipt. Read
-   read-only `public.claim_evidence` to `public.sources` join. A missing locator
+   each evidence row's `artifact_verification` before describing provenance.
-   means weak provenance, but an existing canonical link is still canonical
+   Never infer that an artifact is missing from a raw `storage_path`; use the
-   evidence. Null `applied_at` proves only that proposal did not apply; matching
+   reported status. `no_source_pointer` means the canonical source row has no
-   canonical rows could have another origin.
+   resolvable locator, while `missing`, `hash_mismatch`, and `verified` describe
+   distinct artifact states. An existing claim-evidence link remains canonical
+   evidence even when raw-artifact verification is unavailable. Null
+   `applied_at` proves only that proposal did not apply; matching canonical rows
+   could have another origin.
 3. **Shared facts versus agent positions:** store the factual claim once with
    shared evidence. Current agent-owned positions live in `public.beliefs` with
    `id`, `agent_id`, `level`, `statement`, `confidence`, `falsifier`, `rank`,
@@ -105,6 +109,13 @@ Treat the emitted `Current Runtime Contracts` / `operational_contracts` block as
 binding deployed-code readback. Follow its `required_lead`, schema guards,
 capability tier, and reply budget. It overrides recalled or generic architecture.
 
+For `search` and `context`, cite the receipt's `semantic_context_sha256` when an
+exact repeatable read matters. `read_consistency.status=stable_wal_marker` means
+the read had a stable WAL marker: the database position did not move during the
+read. If it did move, the bridge
+retries and accepts only an identical semantic context; otherwise it fails rather
+than combining rows from ambiguous database states.
+
 The live leoclean profile injects that exact read-only contract automatically
 before the model turn. Before delivery, the same snapshot validates the draft.
 For contract-covered composition and source-intake questions, an invalid draft
@@ -205,6 +216,12 @@ one bounded read-only Postgres inspection and label it as such.
   original probability and call resolution ambiguous when criteria were absent.
 - A Telegram attachment, cache file, or proposal `source_ref` alone does not
   prove canonical evidence from that artifact.
+- `artifact_verification.status` distinguishes `verified`,
+  `verified_hash_unavailable`, `hash_mismatch`, `missing`, `locator_only`,
+  `no_source_pointer`, `remote_locator_only`,
+  `rejected_outside_allowlisted_roots`, `ambiguous_multiple_files`,
+  `too_large_to_verify`, `unreadable`, and `changed_during_verification`. Never infer `missing` from a raw
+  `storage_path` without this verification result.
 - Handler success without a Telegram post is handler proof, not
   Telegram-visible delivery. A visible reply proves delivery, not DB mutation.
 

scripts/leo_behavior_manifest.py

@@ -0,0 +1,305 @@
+#!/usr/bin/env python3
+"""Fingerprint every local runtime layer that can make Leo answer differently.
+
+The manifest deliberately records hashes and a small allowlist of non-secret
+configuration fields. It never emits profile file contents or credentials.
+Canonical Postgres state is outside this manifest and must be fingerprinted by
+the database harness that owns the connection.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+SCHEMA = "livingip.leoBehaviorManifest.v1"
+DEFAULT_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean")
+DEFAULT_HERMES_ROOT = Path("/home/teleo/.hermes/hermes-agent")
+
+
+def sha256_bytes(value: bytes) -> str:
+    return hashlib.sha256(value).hexdigest()
+
+
+def canonical_sha256(value: Any) -> str:
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+    return sha256_bytes(encoded)
+
+
+def file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def _safe_relative(path: Path, root: Path) -> str:
+    try:
+        return path.relative_to(root).as_posix()
+    except ValueError:
+        return str(path)
+
+
+def _resolved_node_fingerprint(path: Path, seen_directories: frozenset[tuple[int, int]] = frozenset()) -> dict[str, Any]:
+    """Hash a symlink target, following nested links while stopping directory cycles."""
+
+    try:
+        if path.is_symlink():
+            target = os.readlink(path)
+            resolved = path.resolve(strict=True)
+            return {
+                "kind": "symlink",
+                "target": target,
+                "resolved": _resolved_node_fingerprint(resolved, seen_directories),
+            }
+        stat = path.stat()
+        if path.is_file():
+            return {
+                "kind": "file",
+                "bytes": stat.st_size,
+                "mode": oct(stat.st_mode & 0o777),
+                "sha256": file_sha256(path),
+            }
+        if path.is_dir():
+            identity = (stat.st_dev, stat.st_ino)
+            if identity in seen_directories:
+                return {"kind": "directory_cycle"}
+            child_seen = seen_directories | {identity}
+            children = [
+                {"name": child.name, "fingerprint": _resolved_node_fingerprint(child, child_seen)}
+                for child in sorted(path.iterdir(), key=lambda item: item.name)
+            ]
+            return {
+                "kind": "directory",
+                "entry_count": len(children),
+                "sha256": canonical_sha256(children),
+            }
+        return {"kind": "other", "mode": oct(stat.st_mode & 0o777)}
+    except (OSError, RuntimeError) as exc:
+        return {"kind": "unavailable", "error": type(exc).__name__}
+
+
+def path_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, Any]:
+    files: list[dict[str, Any]] = []
+    missing: list[str] = []
+    symlinks: list[dict[str, Any]] = []
+    for requested in paths:
+        if not requested.exists() and not requested.is_symlink():
+            missing.append(_safe_relative(requested, profile))
+            continue
+        candidates = [requested] if requested.is_file() or requested.is_symlink() else sorted(requested.rglob("*"))
+        for path in candidates:
+            relative = _safe_relative(path, profile)
+            if path.is_symlink():
+                symlinks.append(
+                    {
+                        "path": relative,
+                        "target": os.readlink(path),
+                        "target_fingerprint": _resolved_node_fingerprint(path),
+                    }
+                )
+                continue
+            if not path.is_file():
+                continue
+            try:
+                stat = path.stat()
+                files.append(
+                    {
+                        "path": relative,
+                        "bytes": stat.st_size,
+                        "mode": oct(stat.st_mode & 0o777),
+                        "sha256": file_sha256(path),
+                    }
+                )
+            except OSError as exc:
+                files.append({"path": relative, "status": "unavailable", "error": type(exc).__name__})
+    files.sort(key=lambda item: item["path"])
+    symlinks.sort(key=lambda item: item["path"])
+    stable = {"files": files, "missing": sorted(missing), "symlinks": symlinks}
+    return {
+        **stable,
+        "file_count": len(files),
+        "total_bytes": sum(int(item.get("bytes", 0)) for item in files),
+        "sha256": canonical_sha256(stable),
+    }
+
+
+def _nested(config: dict[str, Any], *path: str) -> Any:
+    value: Any = config
+    for key in path:
+        value = value.get(key) if isinstance(value, dict) else None
+    return value
+
+
+def safe_model_config(config_path: Path) -> dict[str, Any]:
+    try:
+        raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {}
+    except (OSError, TypeError, yaml.YAMLError) as exc:
+        return {"status": "unavailable", "error": type(exc).__name__}
+    return {
+        "status": "observed",
+        "provider": _nested(raw, "model", "provider"),
+        "default_model": _nested(raw, "model", "default"),
+        "smart_routing": _nested(raw, "model", "smart_routing"),
+        "smart_routing_model": _nested(raw, "model", "smart_routing_model"),
+        "max_tokens": _nested(raw, "model", "max_tokens"),
+        "reasoning_effort": _nested(raw, "agent", "reasoning_effort"),
+        "memory_enabled": _nested(raw, "memory", "enabled"),
+        "memory_search": _nested(raw, "memory", "search"),
+    }
+
+
+def git_head(repo: Path) -> str | None:
+    try:
+        completed = subprocess.run(
+            ["git", "-C", str(repo), "rev-parse", "HEAD"],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+            env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    value = completed.stdout.strip()
+    return value if completed.returncode == 0 and len(value) == 40 else None
+
+
+def component(
+    profile: Path,
+    *,
+    role: str,
+    mutability: str,
+    replayability: str,
+    paths: tuple[Path, ...],
+) -> dict[str, Any]:
+    return {
+        "role": role,
+        "mutability": mutability,
+        "replayability": replayability,
+        "content": path_manifest(profile, paths),
+    }
+
+
+def build_manifest(profile: Path = DEFAULT_PROFILE, hermes_root: Path = DEFAULT_HERMES_ROOT) -> dict[str, Any]:
+    profile = profile.resolve()
+    hermes_root = hermes_root.resolve()
+    config = safe_model_config(profile / "config.yaml")
+    components = {
+        "profile_config": component(
+            profile,
+            role="Selects model routing, tools, memory, and gateway behavior.",
+            mutability="deployment_static",
+            replayability="hashable_but_may_contain_unpinned_provider_defaults",
+            paths=(profile / "config.yaml",),
+        ),
+        "runtime_identity": component(
+            profile,
+            role="SOUL.md identity text injected into the Hermes system prompt.",
+            mutability="file_static_until_edit_or_render",
+            replayability="fully_hashable",
+            paths=(profile / "SOUL.md",),
+        ),
+        "procedural_skills": component(
+            profile,
+            role="On-demand operating instructions available to the model.",
+            mutability="deployment_static",
+            replayability="fully_hashable",
+            paths=(profile / "skills",),
+        ),
+        "runtime_middleware": component(
+            profile,
+            role="Pre-LLM context injection and post-LLM validation or response replacement.",
+            mutability="deployment_static",
+            replayability="fully_hashable",
+            paths=(profile / "plugins", hermes_root / "run_agent.py"),
+        ),
+        "database_tools": component(
+            profile,
+            role="Commands that query Postgres, stage proposals, and return receipts.",
+            mutability="deployment_static",
+            replayability="fully_hashable",
+            paths=(profile / "bin",),
+        ),
+        "persistent_memory": component(
+            profile,
+            role="Hermes MEMORY.md and USER.md content injected independently of canonical Postgres.",
+            mutability="conversation_dynamic_noncanonical",
+            replayability="hashable_but_should_not_define_durable_knowledge",
+            paths=(profile / "memories",),
+        ),
+        "conversation_sessions": component(
+            profile,
+            role="Session transcripts and local state used for conversation continuity.",
+            mutability="request_dynamic_noncanonical",
+            replayability="hashable_but_not_a_knowledge_source",
+            paths=(
+                profile / "sessions",
+                profile / "state",
+                profile / "state.db",
+                profile / "state.db-wal",
+                profile / "state.db-shm",
+            ),
+        ),
+        "generated_prompt_cache": component(
+            profile,
+            role="Hermes-generated cache of the skills prompt.",
+            mutability="generated_cache",
+            replayability="discard_and_rebuild",
+            paths=(profile / ".skills_prompt_snapshot.json",),
+        ),
+    }
+    stable = {
+        "schema": SCHEMA,
+        "model_runtime": {
+            **config,
+            "base_model_weights": "external_provider_managed_not_locally_hashable",
+            "actual_per_turn_model_must_be_recorded_by_the_call_receipt": True,
+        },
+        "hermes_runtime": {
+            "git_head": git_head(hermes_root),
+            "source_tree": path_manifest(
+                profile, (hermes_root / "run_agent.py", hermes_root / "agent" / "prompt_builder.py")
+            ),
+        },
+        "components": components,
+        "canonical_database": {
+            "included": False,
+            "reason": "Fingerprint through a read-only Postgres manifest in the database-owning harness.",
+        },
+    }
+    return {
+        **stable,
+        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+        "profile_root": str(profile),
+        "hermes_root": str(hermes_root),
+        "behavior_sha256": canonical_sha256(stable),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--profile", type=Path, default=DEFAULT_PROFILE)
+    parser.add_argument("--hermes-root", type=Path, default=DEFAULT_HERMES_ROOT)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    manifest = build_manifest(args.profile, args.hermes_root)
+    encoded = json.dumps(manifest, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(encoded, encoding="utf-8")
+    print(encoded, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_leo_direct_claim_handler_suite.py

@@ -2,9 +2,11 @@
 """Run the m3taversal-style direct-claim suite through live VPS GatewayRunner.
 
 The harness does not post to Telegram and does not write to the production KB.
-It copies the live leoclean profile to a temporary profile on the VPS, invokes
+It copies only static behavior surfaces from the live leoclean profile to a
-GatewayRunner._handle_message with Telegram-shaped MessageEvents, and removes
+temporary profile on the VPS, invokes GatewayRunner._handle_message with
-the temporary profile after the run.
+Telegram-shaped MessageEvents, and removes the temporary profile after the
+run. Prior sessions, memory, state databases, and generated prompt caches are
+excluded so the result can be attributed to the declared runtime inputs.
 """
 
 from __future__ import annotations
@@ -53,6 +55,7 @@ from pathlib import Path
 
 LIVE_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean")
 AGENT_ROOT = Path("/home/teleo/.hermes/hermes-agent")
+DEPLOY_ROOT = Path("/opt/teleo-eval/workspaces/deploy-infra")
 CHAT_ID = "__CHAT_ID__"
 USER_ID = "__USER_ID__"
 USER_NAME = "__USER_NAME__"
@@ -121,10 +124,17 @@ def copy_profile() -> Path:
             "cache",
             "image_cache",
             "memory_backups",
+            "memories",
+            "sessions",
+            "state",
             ".pytest_cache",
             "__pycache__",
             "gateway.pid",
             "auth.lock",
+            ".skills_prompt_snapshot.json",
+            "state.db",
+            "state.db-shm",
+            "state.db-wal",
         }
         return {
             name
@@ -137,6 +147,8 @@ def copy_profile() -> Path:
 
     try:
         shutil.copytree(LIVE_PROFILE, target, symlinks=True, ignore=ignore)
+        for directory in ("memories", "sessions", "state"):
+            (target / directory).mkdir(mode=0o700, exist_ok=True)
     except Exception:
         shutil.rmtree(tmp_root, ignore_errors=True)
         raise
@@ -194,8 +206,12 @@ def read_database_context_trace(path):
 
 
 async def run_suite():
+    sys.path.insert(0, str(DEPLOY_ROOT / "scripts"))
+    from leo_behavior_manifest import build_manifest as build_behavior_manifest
+
     before_service = service_state()
     before_counts = db_counts()
+    behavior_before = build_behavior_manifest(LIVE_PROFILE, AGENT_ROOT)
     temp_profile = None
     context_trace_path = None
     temp_removed = False
@@ -203,8 +219,21 @@ async def run_suite():
         "generated_at_utc": datetime.now(timezone.utc).isoformat(),
         "mode": __SUITE_MODE_JSON__,
         "posted_to_telegram": False,
-        "changed_live_profile": False,
         "mutates_kb_by_harness": False,
+        "live_behavior_manifest_before": behavior_before,
+        "temp_profile_seed": {
+            "mode": "static_runtime_surfaces_only",
+            "excluded": [
+                "memories",
+                "sessions",
+                "state",
+                "state.db",
+                "state.db-shm",
+                "state.db-wal",
+                ".skills_prompt_snapshot.json",
+            ],
+            "same_session_continuity_starts_fresh": True,
+        },
         "source": {
             "platform": "telegram",
             "chat_id": CHAT_ID,
@@ -302,8 +331,12 @@ async def run_suite():
     finally:
         after_counts = db_counts()
         after_service = service_state()
+        behavior_after = build_behavior_manifest(LIVE_PROFILE, AGENT_ROOT)
         report["db_counts_after"] = after_counts
         report["db_counts_changed"] = after_counts != before_counts
+        report["live_behavior_manifest_after"] = behavior_after
+        report["changed_live_profile"] = behavior_before["behavior_sha256"] != behavior_after["behavior_sha256"]
+        report["live_behavior_manifest_unchanged"] = not report["changed_live_profile"]
         report["service_before_after"] = {
             "before": before_service,
             "after": after_service,

scripts/run_leo_m3taversal_oos_handler_suite.py

@@ -29,6 +29,7 @@ def write_score_markdown(path: Path, report: dict[str, Any]) -> None:
         f"Pass: `{score['pass']}`",
         f"Prompts: `{score['passes']}/{score['expected_prompt_count']}`",
         f"DB counts changed: `{report['db_counts_changed']}`",
+        f"Live behavior manifest unchanged: `{report['live_behavior_manifest_unchanged']}`",
         f"Service unchanged: `{report['service_unchanged']}`",
         f"Temporary profile removed: `{report['temp_profile_removed']}`",
         f"Database context injections: `{report['database_context_injection_count']}`",
@@ -67,6 +68,8 @@ def build_score_report(report: dict[str, Any], *, memory_token: str) -> dict[str
         "memory_token": memory_token,
         "score": score,
         "db_counts_changed": report.get("db_counts_changed"),
+        "changed_live_profile": report.get("changed_live_profile"),
+        "live_behavior_manifest_unchanged": report.get("live_behavior_manifest_unchanged"),
         "service_unchanged": service_unchanged,
         "temp_profile_removed": report.get("temp_profile_removed"),
         "database_context_injection_count": report.get("database_context_injection_count"),
@@ -84,6 +87,8 @@ def score_report_passes(report: dict[str, Any], score_report: dict[str, Any]) ->
         report.get("remote_returncode") == 0
         and report.get("pass_runtime") is True
         and report.get("db_counts_changed") is False
+        and report.get("changed_live_profile") is False
+        and report.get("live_behavior_manifest_unchanged") is True
         and score_report["service_unchanged"]
         and report.get("temp_profile_removed") is True
         and report.get("database_context_all_ok") is True

tests/test_compile_kb_source_packet.py

@@ -399,6 +399,31 @@ def test_rejects_attempt_to_invent_source_from_chat_label(tmp_path: Path) -> Non
         _compile(tmp_path, manifest)
 
 
+def test_compiles_stable_telegram_message_as_review_candidate_without_learning_from_chat(tmp_path: Path) -> None:
+    manifest = _manifest()
+    locator = "telegram://chat/-5146042086/message/12345"
+    manifest["source"].update(
+        {
+            "identity": locator,
+            "source_key": "telegram_message_12345",
+            "source_type": "transcript",
+            "title": "Telegram discussion message 12345",
+            "locator": locator,
+        }
+    )
+
+    bundle = _compile(tmp_path, manifest)
+    source = bundle["parent_proposal"]["payload"]["source_candidates"][0]
+
+    assert bundle["status"] == "pending_review"
+    assert bundle["database_write_performed"] is False
+    assert bundle["canonical_apply_performed"] is False
+    assert bundle["compiler_contract"]["source_from_chat_labels"] is False
+    assert source["source_type"] == "transcript"
+    assert source["storage_path"] == locator
+    assert source["content_sha256"] == manifest["artifact_sha256"]
+
+
 def test_rejects_quotes_that_cannot_fit_the_reviewable_canonical_excerpt(tmp_path: Path) -> None:
     long_quote = "x" * 8100
     text = f"prefix {long_quote} suffix"

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -100,6 +100,191 @@ def _load_module(path: Path):
     return module
 
 
+def test_vps_bridge_verifies_relative_source_artifact_against_canonical_hash(tmp_path: Path) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    root = tmp_path / "workspace"
+    artifact = root / "domains" / "research.md"
+    artifact.parent.mkdir(parents=True)
+    artifact.write_text("hash-bound evidence\n", encoding="utf-8")
+    digest = module.hashlib.sha256(artifact.read_bytes()).hexdigest()
+
+    verification = module.verify_source_artifact(
+        {
+            "storage_path": "domains/research.md",
+            "url": None,
+            "source_hash": digest,
+        },
+        roots=(root,),
+    )
+
+    assert verification["status"] == "verified"
+    assert verification["artifact_exists"] is True
+    assert verification["artifact_sha256"] == digest
+    assert verification["hash_matches_db"] is True
+    assert verification["resolved_storage_path"] == str(artifact)
+
+
+def test_vps_bridge_does_not_invent_local_artifact_state(tmp_path: Path) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    root = tmp_path / "workspace"
+    root.mkdir()
+    outside = tmp_path / "outside.md"
+    outside.write_text("outside", encoding="utf-8")
+
+    rejected = module.verify_source_artifact(
+        {"storage_path": str(outside), "source_hash": module.hashlib.sha256(outside.read_bytes()).hexdigest()},
+        roots=(root,),
+    )
+    missing = module.verify_source_artifact(
+        {"storage_path": "domains/missing.md", "source_hash": "0" * 64},
+        roots=(root,),
+    )
+    locator = module.verify_source_artifact(
+        {"storage_path": "telegram://chat/-5146042086/message/12345", "source_hash": "0" * 64},
+        roots=(root,),
+    )
+
+    assert rejected["status"] == "rejected_outside_allowlisted_roots"
+    assert rejected["artifact_exists"] is False
+    assert missing["status"] == "missing"
+    assert missing["artifact_exists"] is False
+    assert locator["status"] == "locator_only"
+    assert locator["artifact_exists"] is None
+
+
+def test_vps_bridge_reports_source_hash_mismatch(tmp_path: Path) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    root = tmp_path / "workspace"
+    artifact = root / "evidence.txt"
+    artifact.parent.mkdir()
+    artifact.write_text("actual bytes", encoding="utf-8")
+
+    verification = module.verify_source_artifact(
+        {"storage_path": "evidence.txt", "source_hash": "0" * 64},
+        roots=(root,),
+    )
+
+    assert verification["status"] == "hash_mismatch"
+    assert verification["hash_matches_db"] is False
+    assert verification["artifact_sha256"] != verification["expected_sha256"]
+
+
+def test_vps_bridge_source_artifact_verification_is_size_bounded(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    root = tmp_path / "sources"
+    artifact = root / "large.md"
+    artifact.parent.mkdir(parents=True)
+    artifact.write_bytes(b"0123456789")
+    monkeypatch.setenv("TELEO_KB_MAX_ARTIFACT_BYTES", "4")
+
+    result = module.verify_source_artifact(
+        {"storage_path": "large.md", "source_hash": module.hashlib.sha256(artifact.read_bytes()).hexdigest()},
+        roots=(root,),
+    )
+
+    assert result["status"] == "too_large_to_verify"
+    assert result["artifact_exists"] is True
+    assert result["artifact_bytes"] == 10
+    assert result["max_artifact_bytes"] == 4
+    assert result["artifact_sha256"] is None
+
+
+def test_vps_bridge_evidence_fetch_includes_source_identity_and_verification(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    root = tmp_path / "workspace"
+    artifact = root / "source.md"
+    artifact.parent.mkdir()
+    artifact.write_text("source bytes", encoding="utf-8")
+    digest = module.hashlib.sha256(artifact.read_bytes()).hexdigest()
+    claim_id = "11111111-1111-4111-8111-111111111111"
+    source_id = "22222222-2222-4222-8222-222222222222"
+    captured_sql: list[str] = []
+
+    def fake_psql_json(_args, sql):
+        captured_sql.append(sql)
+        return [
+            {
+                "claim_id": claim_id,
+                "role": "grounds",
+                "weight": 1,
+                "source_id": source_id,
+                "source_type": "article",
+                "url": None,
+                "storage_path": "source.md",
+                "source_hash": digest,
+                "captured_at": "2026-07-14T00:00:00+00:00",
+                "excerpt": "source bytes",
+            }
+        ]
+
+    monkeypatch.setenv("TELEO_KB_SOURCE_ROOTS", str(root))
+    monkeypatch.setattr(module, "psql_json", fake_psql_json)
+
+    rows = module.load_evidence(SimpleNamespace(), [claim_id], 4)[claim_id]
+
+    assert rows[0]["source_id"] == source_id
+    assert rows[0]["source_hash"] == digest
+    assert rows[0]["artifact_verification"]["status"] == "verified"
+    assert "s.id as source_id" in captured_sql[0]
+    assert "s.hash as source_hash" in captured_sql[0]
+    assert "s.captured_at" in captured_sql[0]
+
+
+def test_vps_bridge_context_receipt_is_deterministic_and_retries_moving_wal(monkeypatch) -> None:
+    module = _load_module(BRIDGE_DIR / "kb_tool.py")
+    markers = iter(
+        [
+            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/1"},
+            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/2"},
+            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/2"},
+            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/3"},
+        ]
+    )
+    contract_calls = 0
+
+    def contracts(_args, _query):
+        nonlocal contract_calls
+        contract_calls += 1
+        return [{"id": "reply_budget", "hard_max_words": 200, "statement_timestamp": f"volatile-{contract_calls}"}]
+
+    monkeypatch.setattr(module, "database_read_marker", lambda _args: next(markers))
+    monkeypatch.setattr(module, "context_operational_contracts", contracts)
+    monkeypatch.setattr(module, "compile_operational_response", lambda _contracts: None)
+    monkeypatch.setattr(module, "find_claims", lambda _args, _query, _limit: [])
+    monkeypatch.setattr(module, "load_evidence", lambda _args, _ids, _limit: {})
+    monkeypatch.setattr(module, "load_edges", lambda _args, _ids, _limit: {})
+    monkeypatch.setattr(module, "find_context_rows", lambda _args, _query, _limit: [])
+
+    result = module.bundle(SimpleNamespace(), "same database question", 4, 6)
+    receipt = result["retrieval_receipt"]
+
+    assert contract_calls == 2
+    assert receipt["read_consistency"]["status"] == "stable_content_across_wal_change_retry"
+    assert receipt["read_consistency"]["attempts"] == 2
+    stable_copy = {**result, "operational_contracts": [{"id": "reply_budget", "hard_max_words": 200}]}
+    rebuilt = module.build_retrieval_receipt(
+        {key: value for key, value in stable_copy.items() if key != "retrieval_receipt"},
+        before={"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/9"},
+        after={"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/9"},
+        attempts=1,
+        consistency_status="stable_wal_marker",
+    )
+    assert rebuilt["semantic_context_sha256"] == receipt["semantic_context_sha256"]
+
+
+def test_vps_bridge_retrieval_queries_have_deterministic_final_tie_breakers() -> None:
+    source = (BRIDGE_DIR / "kb_tool.py").read_text(encoding="utf-8")
+
+    assert "length(text), text, id" in source
+    assert "order by score desc, source, owner, title, body" in source
+    assert "s.url nulls last,\n                        s.id" in source
+    assert "other.text,\n                        other.id" in source
+
+
 def test_vps_bridge_search_proposals_finds_approved_rows_by_payload(monkeypatch) -> None:
     module = _load_module(BRIDGE_DIR / "kb_tool.py")
     captured_sql: list[str] = []

tests/test_hermes_leoclean_skill_surfaces.py

@@ -246,8 +246,8 @@ def test_vps_kb_skill_encodes_apply_readiness_beliefs_and_current_proof_boundari
     assert "approved_needs_apply_payload" in text
     assert "normalize into a strict payload and review that new packet" in squashed
     assert "never tell the operator to apply the legacy row" in squashed
-    assert "search` searches claims and identity context, not canonical source rows" in squashed
+    assert "`teleo-kb search` and `teleo-kb context` emit the selected claims" in squashed
-    assert "an existing canonical link is still canonical evidence" in squashed
+    assert "An existing claim-evidence link remains canonical evidence" in squashed
     assert "Current agent-owned positions live in `public.beliefs`" in squashed
     for field in ("`agent_id`", "`level`", "`statement`", "`falsifier`", "`rank`"):
         assert field in text
@@ -265,6 +265,12 @@ 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 "`artifact_verification`" in text
+    assert "`semantic_context_sha256`" in text
+    assert "stable WAL marker" in text
+    assert "Never infer that an artifact is missing from a raw `storage_path`" in squashed
+    assert "Never infer `missing` from a raw `storage_path`" in squashed
+    assert "accepts only an identical semantic context" in squashed
     assert "Automatic Telegram attachment capture" in squashed
     assert "state/kb-source-inbox/source.md" in text
     assert "--receipt /home/teleo/.hermes/profiles/leoclean/state/kb-source-receipts/source.json" in text

tests/test_leo_behavior_manifest.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from scripts import leo_behavior_manifest as manifest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _write(path: Path, value: str) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(value, encoding="utf-8")
+
+
+def test_behavior_manifest_is_deterministic_and_never_emits_config_secrets(tmp_path: Path) -> None:
+    profile = tmp_path / "profile"
+    hermes = tmp_path / "hermes"
+    _write(
+        profile / "config.yaml",
+        """model:
+  provider: openrouter
+  default: anthropic/claude-sonnet-4-6
+  smart_routing: true
+  smart_routing_model: google/gemini-2.5-flash
+memory:
+  enabled: true
+  search: sqlite_fts5
+gateway:
+  telegram:
+    bot_token: super-secret-token
+""",
+    )
+    _write(profile / "SOUL.md", "# Leo\n")
+    _write(profile / "skills" / "teleo-kb-bridge" / "SKILL.md", "# Bridge\n")
+    _write(profile / "plugins" / "leo-db-context" / "__init__.py", "# plugin\n")
+    _write(profile / "bin" / "kb_tool.py", "# tool\n")
+    _write(profile / "memories" / "USER.md", "private participant fact\n")
+    _write(profile / "sessions" / "one.jsonl", '{"private":"conversation"}\n')
+    _write(profile / "state.db", "runtime-state")
+    _write(hermes / "run_agent.py", "# runtime\n")
+    _write(hermes / "agent" / "prompt_builder.py", "# prompts\n")
+
+    first = manifest.build_manifest(profile, hermes)
+    second = manifest.build_manifest(profile, hermes)
+
+    assert first["behavior_sha256"] == second["behavior_sha256"]
+    assert first["model_runtime"]["default_model"] == "anthropic/claude-sonnet-4-6"
+    assert first["components"]["persistent_memory"]["content"]["file_count"] == 1
+    encoded = json.dumps(first)
+    assert "super-secret-token" not in encoded
+    assert "private participant fact" not in encoded
+    assert '{"private":"conversation"}' not in encoded
+
+
+def test_behavior_manifest_changes_when_a_behavior_layer_changes(tmp_path: Path) -> None:
+    profile = tmp_path / "profile"
+    hermes = tmp_path / "hermes"
+    _write(profile / "config.yaml", "model:\n  provider: openrouter\n  default: model-a\n")
+    _write(profile / "SOUL.md", "identity-a\n")
+    _write(hermes / "run_agent.py", "runtime\n")
+    _write(hermes / "agent" / "prompt_builder.py", "prompts\n")
+
+    before = manifest.build_manifest(profile, hermes)
+    _write(profile / "SOUL.md", "identity-b\n")
+    after = manifest.build_manifest(profile, hermes)
+
+    assert before["behavior_sha256"] != after["behavior_sha256"]
+    assert (
+        before["components"]["runtime_identity"]["content"]["sha256"]
+        != after["components"]["runtime_identity"]["content"]["sha256"]
+    )
+
+
+def test_behavior_manifest_hashes_content_behind_symlinks(tmp_path: Path) -> None:
+    profile = tmp_path / "profile"
+    hermes = tmp_path / "hermes"
+    target = tmp_path / "deployed-skill.md"
+    _write(profile / "config.yaml", "model:\n  provider: test\n")
+    _write(profile / "SOUL.md", "identity\n")
+    _write(target, "version-a\n")
+    (profile / "skills").mkdir(parents=True)
+    (profile / "skills" / "SKILL.md").symlink_to(target)
+    _write(hermes / "run_agent.py", "runtime\n")
+    _write(hermes / "agent" / "prompt_builder.py", "prompts\n")
+
+    before = manifest.build_manifest(profile, hermes)
+    _write(target, "version-b\n")
+    after = manifest.build_manifest(profile, hermes)
+
+    assert before["behavior_sha256"] != after["behavior_sha256"]
+    symlinks = after["components"]["procedural_skills"]["content"]["symlinks"]
+    assert symlinks[0]["path"] == "skills/SKILL.md"
+    assert symlinks[0]["target_fingerprint"]["kind"] == "symlink"
+    assert symlinks[0]["target_fingerprint"]["resolved"]["kind"] == "file"
+
+
+def test_live_handler_harness_excludes_prior_dynamic_state_and_compares_live_fingerprints() -> None:
+    source = (ROOT / "scripts" / "run_leo_direct_claim_handler_suite.py").read_text(encoding="utf-8")
+
+    for directory in ("sessions", "state", "memories"):
+        assert f'"{directory}"' in source
+    assert '".skills_prompt_snapshot.json"' in source
+    assert "build_behavior_manifest" in source
+    assert 'report["changed_live_profile"] =' in source
+    assert '"changed_live_profile": False' not in source

tests/test_teleo_agent_systemd.py

@@ -124,6 +124,16 @@ def test_deploy_scripts_sync_leoclean_skills_and_restart_gateway_for_runtime_cha
     assert 'systemctl is-active --quiet leoclean-gateway.service' in manual_deploy
 
 
+def test_deploy_scripts_syntax_check_leoclean_database_runtime_before_restart():
+    auto_deploy = (REPO_ROOT / "deploy" / "auto-deploy.sh").read_text()
+    manual_deploy = (REPO_ROOT / "deploy" / "deploy.sh").read_text()
+
+    assert "hermes-agent/leoclean-bin/*.py" in auto_deploy
+    assert "scripts/leo_behavior_manifest.py" in auto_deploy
+    assert '"$REPO_ROOT/hermes-agent/leoclean-bin/"*.py' in manual_deploy
+    assert '"$REPO_ROOT/scripts/leo_behavior_manifest.py"' in manual_deploy
+
+
 def test_auto_deploy_hot_syncs_leoclean_markdown_without_gateway_restart():
     assert not _leoclean_gateway_restart_required(
         "hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md"