From 5d135d8818e8cf441fd34e52c32cfe0c2f898269 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Mon, 13 Jul 2026 19:53:25 +0200 Subject: [PATCH] Make broad Leo answers database-composed (#120) --- hermes-agent/leoclean-bin/kb_tool.py | 278 ++++++++++++++++-- .../vps/leo-db-context/__init__.py | 9 +- .../working_leo_m3taversal_oos_benchmark.py | 69 ++++- .../test_hermes_leoclean_db_context_plugin.py | 35 +++ .../test_hermes_leoclean_kb_bridge_source.py | 71 +++++ ...st_working_leo_m3taversal_oos_benchmark.py | 106 +++++++ 6 files changed, 533 insertions(+), 35 deletions(-) diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index 6f809a3..8827f0c 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -32,6 +32,7 @@ DIRECT_READBACK_CONTRACTS = frozenset( "identity_canonicality_readback", } ) +SCHEMA_COMPILED_CONTRACTS = frozenset({"forecast_resolution_schema", "claim_supersession_schema"}) DECISION_MATRIX_TABLES = tuple( f"{schema}.{table}" for schema in ("public", "kb_stage") @@ -344,6 +345,16 @@ def with_proposal_readiness(proposal: dict[str, Any]) -> dict[str, Any]: return {**proposal, "readiness": classify_proposal_readiness(proposal)} +def _has_unnegated_action(query: str, actions: tuple[str, ...]) -> bool: + for segment in re.split(r"(?<=[.!?])\s+|\n+", query.lower()): + if not any(re.search(rf"\b{re.escape(action)}\b", segment) for action in actions): + continue + if "read-only" in segment or re.search(r"\b(?:do not|don't|dont|must not|without)\b", segment): + continue + return True + return False + + def operational_contracts( query: str, *, @@ -372,11 +383,29 @@ def operational_contracts( any(term in lowered for term in ("proposal", "pending", "approved", "stuck", "backlog")) and any( term in lowered - for term in ("document", "source row", "source_ref", "source ref", "pointer", "pointed", "evidence link") + for term in ( + "document", + "attachment", + "extracted text", + "source row", + "source_ref", + "source ref", + "pointer", + "pointed", + "evidence link", + ) + ) + and any( + term in lowered + for term in ("because", "wrong", "missing", "stuck", "link", "point", "cause", "canonical evidence", "audit") ) - and any(term in lowered for term in ("because", "wrong", "missing", "stuck", "link", "point", "cause")) ) - kb_scope = "knowledge base" in lowered or bool(re.search(r"\bkb\b", lowered)) or "canonical knowledge" in lowered + kb_scope = ( + "knowledge base" in lowered + or "database" in lowered + or bool(re.search(r"\bkb\b", lowered)) + or "canonical knowledge" in lowered + ) kb_implementation_question = bool( re.search( r"\b(?:scorer|benchmark|test harness|plugin|parser|implementation|search|dashboard|api|script|" @@ -384,16 +413,19 @@ def operational_contracts( lowered, ) ) + forecast_resolution_question = "forecast" in lowered and any( + term in lowered for term in ("resolution", "criteria", "event is now over", "event is over", "60%") + ) proposal_lifecycle_question = any( - term in lowered for term in ("proposal", "proposed", "pending", "approved", "applied", "staged") + term in lowered + for term in ("proposal", "proposed", "pending", "approved", "applied", "staged", "reviewer approval") ) - broad_kb_change_question = any( - term in lowered for term in ("changed", "change", "updated", "update", "landed", "live") + broad_kb_change_question = _has_unnegated_action( + query, ("change", "changed", "update", "updated", "landed") ) - proposal_state_question = ( - "demo" not in lowered - and kb_scope - and (proposal_lifecycle_question or (broad_kb_change_question and not kb_implementation_question)) + proposal_state_question = kb_scope and not forecast_resolution_question and ( + proposal_lifecycle_question + or ("demo" not in lowered and broad_kb_change_question and not kb_implementation_question) ) named_packet_question = "helmer" in lowered and any( term in lowered for term in ("7 powers", "seven powers", "powers framework", "powers packet") @@ -414,8 +446,11 @@ def operational_contracts( ) ) or ( - (kb_scope or "database" in lowered) - and any(term in lowered for term in ("write", "apply", "mutate", "stage")) + kb_scope + and ( + any(term in lowered for term in ("learned", "updated", "update", "live", "current", "working")) + or _has_unnegated_action(query, ("write", "apply", "mutate", "stage")) + ) ) ) identity_question = any(term in lowered for term in ("soul.md", "soul file")) and any( @@ -428,9 +463,9 @@ def operational_contracts( ("decision_matrix_readback", decision_matrix_question), ("source_link_audit", source_link_audit_question), ("named_packet_readback", named_packet_question), - ("demo_capability_readback", demo_question), ("identity_canonicality_readback", identity_question), ("proposal_state_readback", proposal_state_question), + ("demo_capability_readback", demo_question), ) if enabled ), @@ -483,6 +518,7 @@ def operational_contracts( contracts.append( { "id": "proposal_state_readback", + "approval_only_claim": "reviewer approval" in lowered and broad_kb_change_question, "state_boundary": ( "proposal status, reviewer approval, and canonical apply are separate states; applied rows still " "need proposal-level applied_at plus row-level postflight for a specific change claim" @@ -517,6 +553,9 @@ def operational_contracts( contracts.append( { "id": "demo_capability_readback", + "temporal_comparison_requested": any( + term in lowered for term in ("since yesterday", "since last night", "since earlier") + ), "tiers": [ "read existing canonical and proposal receipts", "stage a safe pending_review proposal", @@ -621,6 +660,53 @@ def operational_contracts( } ) + if forecast_resolution_question: + contracts.append( + { + "id": "forecast_resolution_schema", + "current_columns": { + name: list(schema.get(name, ())) for name in ("claims", "claim_edges") + }, + "history_boundary": "preserve the original forecast and its missing resolution criteria", + "schema_boundary": ( + "do not invent forecast-resolution fields or a resolves edge; stage a reviewed schema proposal" + ), + } + ) + + claim_supersession_question = ( + "claim" in lowered + and any(term in lowered for term in ("replacement", "wrong", "retired", "supersed")) + and any(term in lowered for term in ("approve_claim", "current v1", "edge field", "old claim")) + ) + if claim_supersession_question: + contracts.append( + { + "id": "claim_supersession_schema", + "current_columns": { + name: list(schema.get(name, ())) for name in ("claims", "claim_edges") + }, + "insert_boundary": "approve_claim may insert a replacement claim plus a claim-to-claim edge", + "update_boundary": ( + "existing claim status and superseded_by updates require a separate reviewed apply capability" + ), + } + ) + + telegram_delivery_question = ( + "telegram" in lowered + and any(term in lowered for term in ("gatewayrunner", "temporary-profile", "temporary profile", "posted nothing")) + and any(term in lowered for term in ("proven", "delivery", "visible")) + ) + if telegram_delivery_question: + contracts.append( + { + "id": "telegram_delivery_proof", + "handler_boundary": "a non-posting GatewayRunner reply proves handler behavior, not Telegram delivery", + "receipt": "visible reply in the intended chat with message ID, timestamp, and chat/sender readback", + } + ) + if any(term in lowered for term in ("restart", "prior session", "erased", "answer behavior", "database totals")): contracts.append( { @@ -726,13 +812,14 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: retain_step = safe_steps[0] if safe_steps else "retain the artifact plus its locator and content hash" stage_step = safe_steps[1] if len(safe_steps) > 1 else "stage candidates in a pending_review proposal" return ( - f"- {source.get('required_lead')} Immediately {retain_step}. A filename, chat label, attachment reference, or " - "source_ref is not source identity.\n\n" - f"- {stage_step}. Deduplicate candidate claims and retain evidence excerpts and contradictions. Do not " - "create public.sources or other canonical rows during capture or staging.\n\n" - f"- Explicit approval begins {source.get('approval_begins')}. The receipt should include the retained " - "locator and hash, proposal ID and status, reviewed payload, proposal-level applied_at, created row IDs, " - "and postflight readback. Until apply succeeds, the material is staged rather than live canonical knowledge." + f"- {source.get('required_lead')} Immediately {retain_step}; retain provenance for each PDF, Telegram " + "message, and tweet. A filename, chat label, attachment reference, or source_ref is not source identity.\n\n" + f"- {stage_step}. Deduplicate overlapping candidate claims, preserve evidence excerpts, source-specific " + "caveats, uncertainty, and contradictions. Candidates are not canonical until reviewed and applied: create " + "no public.sources or other public.* rows during capture or staging.\n\n" + f"- Explicit approval begins {source.get('approval_begins')}. Receipt: locator and content hash, proposal ID/" + "status, reviewed payload, applied_at, created row IDs, and postflight readback. Next proof-changing " + "follow-up: run the retained artifact through the staging compiler, then review its typed candidates." ) named_packet = by_id.get("named_packet_readback") @@ -821,14 +908,17 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: if source_audit: audit = source_audit.get("live_link_audit") or {} return ( - f"Not proven.\n{format_database_count_readback(source_audit.get('database_status'))}\nThe live row-link " + f"No. The attachment does not itself create canonical evidence; neither does extracted text on disk.\n" + f"{format_database_count_readback(source_audit.get('database_status'))}\nThe live row-link " f"audit found {audit.get('pending_or_approved', 'unknown')} pending/approved proposals: " f"{audit.get('with_source_ref', 'unknown')} have source_ref and " f"{audit.get('exact_public_source_matches', 'unknown')} exactly match a public.sources ID, URL, or " "storage_path. Pending status therefore is not by itself proof of a bad document pointer.\n\n" "Audit telegram_file_refs/raw files -> document_evaluations -> kb_proposals.source_ref -> public.sources " - "-> claim_evidence. Next proof-changing follow-up: list each proposal UUID/status/source_ref and exact " - "source match, then stage only the missing guarded source/evidence apply contract." + "-> claim_evidence. Canonical linkage and provenance quality are separate: a canonical link can still point " + "to a weak, untraceable source with no URL/storage_path or raw artifact. Before/after receipt: proposal UUID/" + "status/source_ref, source and claim_evidence row IDs, locator/hash, applied_at, and postflight. Next proof-" + "changing follow-up: list each broken layer, then stage only the missing guarded source/evidence contract." ) identity = by_id.get("identity_canonicality_readback") @@ -836,10 +926,11 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: return ( f"No.\n{format_database_count_readback(identity.get('database_status'))}\nA direct SOUL.md edit is a " "runtime/profile change; it does not change canonical Postgres rows in personas, strategies, beliefs, " - "strategy_nodes, or strategy_node_anchors. No active general DB-to-SOUL.md renderer automation is " - "currently proven, so neither a future render nor persistence of the edit should be assumed.\n\n" + "strategy_nodes, or strategy_node_anchors. A chat correction is also not canonical until it is staged, " + "reviewed, and applied as exact identity rows. No active general DB-to-SOUL.md renderer automation is " + "currently proven, so neither surface should be assumed to survive as identity after restart.\n\n" "Next proof-changing follow-up: audit the intended canonical identity rows and proposal UUID/status/" - "applied_at, stage or apply the reviewed DB change with authorization, run or verify render/sync, then " + "applied_at, stage the exact change, review it, apply it with authorization, run or verify render/sync, then " "compare SOUL.md hash/content and restart readback." ) @@ -848,15 +939,35 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: status_data = demo.get("database_status") or {} states = status_data.get("proposal_status_counts") or {} applied_rows = list(demo.get("applied_proposals") or []) + approved_rows = [ + item for item in demo.get("applied_and_approved_proposals") or [] if item.get("status") == "approved" + ] + approved_needs_payload = sum( + (item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload" + for item in approved_rows + ) applied_types = human_join( sorted({str(item.get("proposal_type")) for item in applied_rows if item.get("proposal_type")}) ) + if demo.get("temporal_comparison_requested"): + return ( + f"Not proven.\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are " + f"applied: {states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: " + f"{states.get('pending_review', 0)}. This current snapshot does not establish what changed since " + f"yesterday. Existing receipts identify applied types {applied_types or 'none'}, but not a time-window " + f"delta. Approved is not applied; {approved_needs_payload}/{len(approved_rows)} approved rows are " + "approved_needs_apply_payload and not directly applyable.\n\n" + "Next proof-changing follow-up: build and review a strict apply_payload for the highest-impact approved " + "proposal, then seek explicit apply authorization. Until its applied_at and public.* before/after row " + "IDs or hashes exist, do not claim the canonical database changed." + ) return ( - f"Partly.\n{format_database_count_readback(status_data)}\nProposal states are applied: " + f"Partly.\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: " f"{states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: " f"{states.get('pending_review', 0)}. Existing receipts show applied proposal types: " f"{applied_types or 'none identified'}; that receipt artifact does not prove every rich packet or chat " - "apply path is live. " + f"apply path is live. Approved is not applied. {approved_needs_payload}/{len(approved_rows)} approved rows " + "are approved_needs_apply_payload and not directly applyable. " "You can safely demo canonical readback and a pending_review staging write. A new canonical apply requires " "explicit authorization, a supported strict payload, and postflight row IDs.\n\n" "Next proof-changing follow-up: choose the demo tier - existing applied receipt, safe staging canary, or " @@ -868,16 +979,99 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: if proposal_state: status_data = proposal_state.get("database_status") or {} states = status_data.get("proposal_status_counts") or {} + approved_rows = [ + item + for item in proposal_state.get("applied_and_approved_proposals") or [] + if item.get("status") == "approved" + ] + approved_needs_payload = sum( + (item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload" + for item in approved_rows + ) + lead = "No." if proposal_state.get("approval_only_claim") else "Partly." return ( - f"Partly.\n{format_database_count_readback(status_data)}\nProposal states are applied: " + f"{lead}\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: " f"{states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: " f"{states.get('pending_review', 0)}, canceled: {states.get('canceled', 0)}. Approved is not applied, and " - "an applied proposal receipt alone does not identify the exact public.* rows changed.\n\n" + f"{approved_needs_payload}/{len(approved_rows)} approved rows are approved_needs_apply_payload and not " + "directly applyable. An applied proposal receipt alone does not identify the exact public.* rows changed.\n\n" "Next proof-changing follow-up: choose the proposal or time window, return its UUID/status/applied_at, " "compare the affected public.* row IDs or before/after hashes, and for approved rows normalize and review " "the strict payload before requesting explicit guarded-apply authorization." ) + runtime = by_id.get("runtime_persistence") + if runtime: + return ( + "No. Unchanged database totals do not prove unchanged rows or unchanged answer behavior, and a restart " + "does not necessarily erase prior-session facts. Compare row IDs, timestamps, and hashes, not counts alone; " + "balanced inserts, updates, or deletes can leave totals unchanged.\n\n" + "Persisted or deployed answer inputs include Postgres rows, skills, runtime config, SOUL.md, state.db, and " + "session JSONL. state.db/session JSONL can preserve durable session continuity across a restart, while a " + "changed skill or runtime artifact can change the answer with identical database totals.\n\n" + "Proof tiers: 1) canonical: public.* row/hash readback; 2) handler: the deployed runtime answers the same " + "prompt with retained trace; 3) Telegram: a visible reply with message ID, timestamp, and chat/sender " + "readback. Only a reviewed apply plus row-level postflight proves a database mutation." + ) + + shared = by_id.get("shared_claims_agent_positions") + if shared: + return ( + "Store one factual claim once as shared knowledge with shared sources/evidence; do not duplicate the claim " + "per agent. Record each agent-specific position in public.beliefs with agent_id, statement, confidence, " + "falsifier, and status.\n\n" + "The current-schema gap matters: public.beliefs has no claim-ID foreign key and there is no belief-edge " + "table. claim_edges connects claims only, so do not invent belief-to-claim or belief-to-belief edges. " + "Disagreement remains queryable by retrieving the shared factual claim/evidence and semantically matching " + "each agent's belief rows; a future reviewed schema proposal can add an explicit link if needed." + ) + + forecast = by_id.get("forecast_resolution_schema") + if forecast: + columns = forecast.get("current_columns") or {} + claim_columns = human_join([str(item) for item in columns.get("claims") or []]) + edge_columns = human_join([str(item) for item in columns.get("claim_edges") or []]) + edge_types = [str(item) for item in forecast.get("current_edge_types") or []] + resolves_state = "`resolves` is present" if "resolves" in edge_types else "there is no `resolves` edge type" + return ( + "Preserve the original 60% forecast; do not overwrite it. Because it defined no resolution criteria, its " + "outcome remains ambiguous. Leo may add a sourced explanatory claim about the observed event, but must not " + "retroactively label the forecast resolved.\n\n" + f"Current public.claims columns are {claim_columns}; there is no forecast-resolution or resolved_at field. " + f"Current public.claim_edges columns are {edge_columns}; {resolves_state}. Stage a reviewed schema proposal " + "for explicit criteria, outcome, resolved_at, resolver/provenance, and a valid link contract. Do not apply or " + "invent those fields in v1." + ) + + delivery = by_id.get("telegram_delivery_proof") + if delivery: + return ( + "No. A temporary-profile GatewayRunner response proves handler execution, model/plugin integration, and " + "reply construction only. Because it posted nothing, it does not prove Telegram delivery or a Telegram-" + "visible reply.\n\n" + "The smallest closing test is one explicitly authorized test message in the intended chat, followed by " + "Leo's visible reply. Retain a delivery receipt with the outgoing and reply message IDs, timestamps, chat " + "ID/title, sender handle, reply linkage, and a post-send readback. Then remove the temporary profile and " + "verify no orphan process remains." + ) + + supersession = by_id.get("claim_supersession_schema") + if supersession: + columns = supersession.get("current_columns") or {} + claim_columns = human_join([str(item) for item in columns.get("claims") or []]) + edge_columns = human_join([str(item) for item in columns.get("claim_edges") or []]) + edge_types = [str(item) for item in supersession.get("current_edge_types") or []] + supersedes_state = "`supersedes` is valid" if "supersedes" in edge_types else "`supersedes` is not valid" + return ( + f"Current public.claims fields: {claim_columns}. Current claim_edges fields: {edge_columns}; the edge has no " + f"rationale field, and {supersedes_state} in the live edge_type enum.\n\n" + "approve_claim can insert the replacement claim and a claim-to-claim supersedes edge using from_claim, " + "to_claim, edge_type, weight, and created_by. Put the explanation in sourced evidence or an explanatory " + "claim. Updating the existing old claim's status or superseded_by to visibly retire it requires a separate " + "reviewed apply capability; approve_claim does not update those fields. Receipt: proposal applied_at, new " + "claim/edge IDs, old-claim before/after, and postflight readback." + ) + return None @@ -917,6 +1111,25 @@ def load_current_public_constraints(args: argparse.Namespace, table_names: set[s return {row["table_name"]: row["constraints"] for row in psql_json(args, sql)} +def load_current_enum_values(args: argparse.Namespace, type_names: set[str]) -> dict[str, list[str]]: + if not type_names: + return {} + sql = f""" + select jsonb_build_object( + 'type_name', t.typname, + 'values', jsonb_agg(e.enumlabel order by e.enumsortorder) + )::text + from pg_type t + join pg_enum e on e.enumtypid = t.oid + join pg_namespace n on n.oid = t.typnamespace + where n.nspname = 'public' + and t.typname = any({sql_array(sorted(type_names))}) + group by t.typname + order by t.typname; + """ + return {row["type_name"]: row["values"] for row in psql_json(args, sql)} + + def compact_proposal(proposal: dict[str, Any]) -> dict[str, Any]: return { key: proposal.get(key) @@ -1120,6 +1333,9 @@ def context_operational_contracts(args: argparse.Namespace, query: str) -> list[ table_names.update( {"claims", "claim_edges", "reasoning_tools", "behavioral_rules", "governance_gates", "beliefs"} ) + if contract_ids & SCHEMA_COMPILED_CONTRACTS: + table_names.update({"claims", "claim_edges"}) + enum_values = load_current_enum_values(args, {"edge_type"}) if contract_ids & SCHEMA_COMPILED_CONTRACTS else {} contracts = operational_contracts( query, current_schema=load_current_public_schema(args, table_names) if table_names else None, @@ -1152,6 +1368,8 @@ def context_operational_contracts(args: argparse.Namespace, query: str) -> list[ contract["live_link_audit"] = direct_snapshot.get("source_link_audit") or {} elif contract_id == "demo_capability_readback": contract["applied_proposals"] = [item for item in proposal_rows if item.get("status") == "applied"] + elif contract_id in SCHEMA_COMPILED_CONTRACTS: + contract["current_edge_types"] = enum_values.get("edge_type", []) return contracts diff --git a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py index 6f6955c..7548662 100644 --- a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py +++ b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py @@ -28,6 +28,11 @@ COMPILED_CONTRACT_IDS = frozenset( "source_link_audit", "demo_capability_readback", "identity_canonicality_readback", + "runtime_persistence", + "shared_claims_agent_positions", + "forecast_resolution_schema", + "telegram_delivery_proof", + "claim_supersession_schema", } ) DECISION_MATRIX_TABLES = tuple( @@ -79,9 +84,11 @@ def _requires_database_truth(query: str) -> bool: ) return bool( named_packet_question + or ("telegram" in lowered and "gatewayrunner" in lowered) or re.search( r"\b(?:databases?|db|knowledge bases?|kb|proposals?|canonical|source_ref|source rows?|" - r"decision[- ]matri(?:x|ces)|soul\.md|soul file|claim_evidence|claim_edges|reasoning_tools)\b|" + r"decision[- ]matri(?:x|ces)|soul\.md|soul file|claim_evidence|claim_edges|reasoning_tools|" + r"approve_claim|superseded_by|current[- ]schema)\b|" r"public\.|kb_stage", lowered, ) diff --git a/scripts/working_leo_m3taversal_oos_benchmark.py b/scripts/working_leo_m3taversal_oos_benchmark.py index 272b5d3..6593d5f 100755 --- a/scripts/working_leo_m3taversal_oos_benchmark.py +++ b/scripts/working_leo_m3taversal_oos_benchmark.py @@ -10,6 +10,7 @@ or Telegram-delivery harness. from __future__ import annotations import argparse +import hashlib import json import re from datetime import datetime, timezone @@ -242,7 +243,7 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = { ), "state_boundary": ( re.compile( - r"approved is not the same as applied|applied_at:\s*(?:none|null)|not canonical|" + r"approved is not (?:the same as )?applied|applied_at:\s*(?:none|null)|not canonical|" r"proposal does not commit|no receipt,? no durable knowledge|" r"approval.{0,60}not (?:a )?(?:DB|database|canonical) write|" r"(?:approval|sign-off) is intent,? not (?:apply|a canonical write)|none.{0,30}applied", @@ -258,7 +259,9 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = { re.I, ), ), - "provenance": (re.compile(r"provenance|stable (?:reference|ref)|source row|author/channel|file hash", re.I),), + "provenance": ( + re.compile(r"provenance|stable (?:reference|ref)|source row|author/channel|(?:file|content) hash", re.I), + ), "deduplication": (re.compile(r"deduplic|duplicate|existing claim|overlap", re.I),), "contradiction": (re.compile(r"contradict|divergence|competing interpretation|disagree", re.I),), "staged_review_apply": ( @@ -472,9 +475,9 @@ COUNT_INVARIANT_REJECTION_RE = re.compile( SCHEMA_GAP_QUALIFIER_RE = re.compile( r"\b(?:proposed|future|not current|not shipped|does not exist|doesn't exist|absent|" - r"has no|have no|no column|not an edge|would require|schema gap|must be added|" + r"has no|have no|there is no|no column|not an edge|would require|schema gap|must be added|" r"does not support|doesn't support|supports neither|not supported|must not|do not invent)\b|" - r"\bno\s+(?:[a-z_]+\s+){0,3}(?:column|field|table)\b", + r"\bno\s+(?:[a-z_]+\s+){0,3}(?:column|field|table|edge type)\b", re.I, ) CURRENT_SCHEMA_ASSERTION_PATTERNS: dict[str, re.Pattern[str]] = { @@ -605,6 +608,20 @@ APPLYABILITY_GAP_RE = re.compile( DEFAULT_MAX_RESPONSE_WORDS = 220 MAX_RESPONSE_WORDS = {"OOS-09": 100, "OOS-10": 180} +BLOCKER_STOPWORDS = { + "blocker", + "chat", + "demo", + "label", + "only", + "remember", + "single", + "that", + "this", + "under", + "will", + "with", +} def prompt_catalog(memory_token: str) -> list[dict[str, Any]]: @@ -617,6 +634,28 @@ def prompt_catalog(memory_token: str) -> list[dict[str, Any]]: ] +def extract_blocker_clause(reply: str) -> str | None: + for pattern in ( + re.compile(r"\bblocker\s*:\s*(?P[^\n]+)", re.I), + re.compile(r"\bblocker\s+(?:is|was)\s+(?P[^\n]+)", re.I), + ): + match = pattern.search(reply) + if match: + return match.group("value").strip() + return None + + +def blocker_terms(value: str | None, *, memory_token: str) -> set[str]: + if not value: + return set() + token_terms = set(re.findall(r"[a-z0-9_]+", memory_token.lower())) + return { + term + for term in re.findall(r"[a-z0-9_]+", value.lower()) + if len(term) >= 4 and term not in BLOCKER_STOPWORDS and term not in token_terms + } + + def matched_concept(reply: str, concept: str) -> bool: return all(pattern.search(reply) for pattern in CONCEPT_PATTERNS[concept]) @@ -803,6 +842,27 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s for prompt_id in expected_ids if prompt_id in by_result ] + memory_source_reply = str((by_result.get("OOS-07") or {}).get("reply") or "") + memory_recall_reply = str((by_result.get("OOS-08") or {}).get("reply") or "") + source_clause = extract_blocker_clause(memory_source_reply) + source_terms = blocker_terms(source_clause, memory_token=memory_token) + recall_terms = blocker_terms(extract_blocker_clause(memory_recall_reply), memory_token=memory_token) + overlap_terms = source_terms & recall_terms + required_overlap = min(3, max(1, (len(source_terms) + 2) // 3)) if source_terms else 1 + same_blocker_recalled = bool(source_clause and len(overlap_terms) >= required_overlap) + for score in scores: + if score["prompt_id"] != "OOS-08": + continue + score["custom_signals"]["same_blocker_recalled"] = same_blocker_recalled + score["pass"] = bool(score["pass"] and same_blocker_recalled) + memory_continuity = { + "source_reply_sha256": hashlib.sha256(memory_source_reply.encode("utf-8")).hexdigest(), + "source_blocker_clause": source_clause, + "source_blocker_terms": sorted(source_terms), + "recall_overlap_terms": sorted(overlap_terms), + "required_overlap": required_overlap, + "same_blocker_recalled": same_blocker_recalled, + } return { "expected_prompt_count": len(expected_ids), "expected_prompt_ids": expected_ids, @@ -812,6 +872,7 @@ def score_results(results: list[dict[str, Any]], *, memory_token: str) -> dict[s "passes": sum(1 for score in scores if score["pass"]), "failures": [score for score in scores if not score["pass"]], "scores": scores, + "memory_continuity": memory_continuity, "pass": not missing and not unexpected and len(scores) == len(expected_ids) diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index d026d3a..3e3f6c8 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/tests/test_hermes_leoclean_db_context_plugin.py @@ -279,6 +279,8 @@ def test_database_truth_fallback_covers_every_direct_readback_question() -> None "Are the pending proposals stuck because the documents are not pointed at the right source rows?", "Can I demo that Leo changes the KB?", "If we changed SOUL.md, did we change Leo's canonical identity?", + "In the current-schema, can approve_claim set superseded_by?", + "A temporary-profile GatewayRunner posted nothing to Telegram. Is delivery proven?", ) assert all(module._requires_database_truth(question) for question in direct_questions) @@ -286,6 +288,39 @@ def test_database_truth_fallback_covers_every_direct_readback_question() -> None assert module._requires_database_truth("How are you?") is False +def test_plugin_enforces_every_database_backed_oos_compiler() -> None: + module = load_plugin() + expected_ids = { + "runtime_persistence", + "shared_claims_agent_positions", + "forecast_resolution_schema", + "telegram_delivery_proof", + "claim_supersession_schema", + } + assert expected_ids <= module.COMPILED_CONTRACT_IDS + + for index, contract_id in enumerate(sorted(expected_ids)): + query = f"database-backed query {index}" + query_sha256 = module.hashlib.sha256(query.encode("utf-8")).hexdigest() + compiled = f"Database-composed response for {contract_id}." + contracts = [{"id": "reply_budget", "hard_max_words": 200}, {"id": contract_id}] + module._store_snapshot( + f"compiled-{index}", + { + "status": "ok", + "query_sha256": query_sha256, + "contracts": contracts, + "compiled_response": compiled, + "requires_database_truth": True, + "context": "fixture", + }, + ) + + assert module._post_llm_call( + session_id=f"compiled-{index}", user_message=query, assistant_response="Unconstrained draft." + ) == {"assistant_response": compiled} + + def test_plugin_fails_closed_on_missing_snapshot_or_invalid_compiler() -> None: module = load_plugin() query = "Did an approved proposal change the knowledge base?" diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index b5054cc..b17cea9 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -293,6 +293,43 @@ def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> No assert substring_ids & direct_ids == {"proposal_state_readback"} +def test_vps_bridge_oos_intents_preserve_specific_contracts_and_negated_actions() -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + cases = { + ( + "I corrected Leo's worldview in chat and someone also edited SOUL.md. Will that correction be part of " + "Leo's canonical identity tomorrow after a restart? Give me the exact truth test from database row to " + "runtime artifact. Do not change either surface." + ): {"identity_canonicality_readback", "runtime_persistence"}, + ( + "The five database totals are unchanged after a restart. Does that prove Leo's answer behavior is " + "unchanged and that every fact from the prior session was erased?" + ): {"runtime_persistence"}, + ( + "An old claim recorded a 60% forecast but never defined resolution criteria. The event is now over. " + "What needs a schema proposal?" + ): {"forecast_resolution_schema"}, + ( + "A canonical claim is wrong. I want the replacement and the old claim visibly retired. In current v1, " + "which writes fit approve_claim and which require a separate reviewed apply capability?" + ): {"claim_supersession_schema"}, + ( + "A temporary-profile GatewayRunner answered, but posted nothing to Telegram. Is delivery proven live?" + ): {"telegram_delivery_proof"}, + } + + for query, expected_ids in cases.items(): + contract_ids = {item["id"] for item in module.operational_contracts(query)} - {"reply_budget"} + assert contract_ids == expected_ids, query + + memory_query = ( + "Remember this blocker under __MEMORY_TOKEN__. This is chat memory only; do not write it to the KB." + ) + memory_ids = {item["id"] for item in module.operational_contracts(memory_query)} + assert not memory_ids & module.DIRECT_READBACK_CONTRACTS + assert memory_ids == {"reply_budget"} + + def _direct_contract_fixtures() -> tuple[dict, dict, dict, dict]: database_status = { "high_signal_rows": { @@ -581,6 +618,40 @@ def test_vps_bridge_mixed_contract_reads_live_postgres_constraints(monkeypatch) assert "proposal-level applied_at" in mixed["review_apply_sequence"][-1] +def test_vps_bridge_schema_contract_reads_live_columns_and_edge_enum(monkeypatch) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + captured_sql: list[str] = [] + + def fake_psql_json(_args, sql): + captured_sql.append(sql) + if "pg_enum" in sql: + return [{"type_name": "edge_type", "values": ["supports", "supersedes"]}] + if "information_schema.columns" in sql: + return [ + {"table_name": "claims", "columns": ["id", "text", "superseded_by"]}, + {"table_name": "claim_edges", "columns": ["id", "from_claim", "to_claim", "edge_type"]}, + ] + if "pg_constraint" in sql: + return [] + raise AssertionError(sql) + + monkeypatch.setattr(module, "psql_json", fake_psql_json) + contracts = { + item["id"]: item + for item in module.context_operational_contracts( + SimpleNamespace(), + "A canonical claim is wrong. In current v1, can approve_claim insert its replacement and supersedes edge?", + ) + } + + supersession = contracts["claim_supersession_schema"] + assert supersession["current_columns"]["claims"] == ["id", "text", "superseded_by"] + assert supersession["current_columns"]["claim_edges"] == ["id", "from_claim", "to_claim", "edge_type"] + assert supersession["current_edge_types"] == ["supports", "supersedes"] + assert any("pg_enum" in sql and "t.typname" in sql for sql in captured_sql) + assert all(not re.search(r"\b(?:insert|update|delete|alter|drop|create)\b", sql, re.I) for sql in captured_sql) + + def test_vps_bridge_decision_matrix_status_checks_schema_tables(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") captured_sql: list[str] = [] diff --git a/tests/test_working_leo_m3taversal_oos_benchmark.py b/tests/test_working_leo_m3taversal_oos_benchmark.py index 4a39c8a..b1be2ee 100644 --- a/tests/test_working_leo_m3taversal_oos_benchmark.py +++ b/tests/test_working_leo_m3taversal_oos_benchmark.py @@ -154,6 +154,10 @@ def test_oos_score_passes_complete_behavior_and_memory_pair() -> None: score = benchmark.score_results(results, memory_token=token) assert score["pass"] is True assert score["passes"] == 15 + assert score["memory_continuity"]["same_blocker_recalled"] is True + assert {"approved", "applied", "canonical"} <= set( + score["memory_continuity"]["recall_overlap_terms"] + ) def test_oos_live_check_accepts_live_readback_wording() -> None: @@ -174,6 +178,26 @@ def test_oos_score_fails_when_memory_token_is_not_recalled() -> None: assert score["failures"][0]["custom_signals"]["memory_token"] is False +def test_oos_score_fails_when_recalled_label_has_a_different_blocker() -> None: + token = "demo-ledger-deadbeef" + results = [ + {"prompt_id": prompt["id"], "reply": good_reply(prompt["id"], token)} + for prompt in benchmark.prompt_catalog(token) + ] + recall = next(result for result in results if result["prompt_id"] == "OOS-08") + recall["reply"] = ( + f"Label: {token}. Blocker: Telegram delivery has no visible message receipt. " + "Proof: send one authorized canary and retain its message ID, timestamp, and chat readback." + ) + + score = benchmark.score_results(results, memory_token=token) + + assert score["pass"] is False + recall_score = next(item for item in score["scores"] if item["prompt_id"] == "OOS-08") + assert recall_score["custom_signals"]["memory_token"] is True + assert recall_score["custom_signals"]["same_blocker_recalled"] is False + + def test_oos_identity_case_requires_exact_visible_handle_and_no_alias() -> None: token = "demo-ledger-deadbeef" prompt = benchmark.prompt_catalog(token)[8] @@ -508,6 +532,88 @@ def test_db_compiled_responses_pass_composition_and_source_intake_benchmarks() - assert benchmark.score_reply(prompt, response, memory_token=token)["pass"] is True +def test_db_compiled_responses_pass_all_database_backed_oos_cases() -> None: + kb_tool = load_kb_tool() + token = "demo-ledger-deadbeef" + database_status = { + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 26, + }, + "proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 14, "canceled": 7}, + } + approved = { + "id": "a64df080-8502-42e2-98f4-9bbdecb8da73", + "proposal_type": "approve_claim", + "status": "approved", + "applied_at": None, + "readiness": {"review_state": "approved_needs_apply_payload"}, + } + applied = { + **approved, + "id": "11111111-1111-4111-8111-111111111111", + "status": "applied", + "applied_at": "2026-07-01T00:00:00+00:00", + "readiness": {"review_state": "applied"}, + } + source_audit = { + "pending_or_approved": 17, + "with_source_ref": 16, + "without_source_ref": 1, + "exact_public_source_matches": 2, + "samples": [], + } + compiled_prompt_ids = { + "OOS-01", + "OOS-02", + "OOS-03", + "OOS-04", + "OOS-05", + "OOS-06", + "OOS-10", + "OOS-11", + "OOS-12", + "OOS-13", + "OOS-14", + "OOS-15", + } + + for prompt in benchmark.prompt_catalog(token): + if prompt["id"] not in compiled_prompt_ids: + continue + contracts = kb_tool.operational_contracts(prompt["message"]) + for contract in contracts: + contract_id = contract["id"] + if contract_id in kb_tool.DIRECT_READBACK_CONTRACTS: + contract["database_status"] = database_status + if contract_id == "demo_capability_readback": + contract["applied_proposals"] = [applied] + contract["applied_and_approved_proposals"] = [approved, applied] + elif contract_id == "proposal_state_readback": + contract["applied_and_approved_proposals"] = [approved, applied] + elif contract_id == "source_link_audit": + contract["live_link_audit"] = source_audit + elif contract_id in kb_tool.SCHEMA_COMPILED_CONTRACTS: + contract["current_edge_types"] = ["supports", "supersedes"] + + response = kb_tool.compile_operational_response(contracts) + assert response is not None, prompt["id"] + score = benchmark.score_reply(prompt, response, memory_token=token) + assert score["pass"] is True, score + + +def test_oos_state_boundary_accepts_approved_is_not_applied() -> None: + assert benchmark.matched_concept("Approved is not applied.", "state_boundary") is True + + +def test_oos_schema_gap_accepts_explicitly_absent_edge_type() -> None: + reply = "Current claim_edges has edge_type; there is no `resolves` edge type." + assert benchmark.current_schema_overclaims(reply) == [] + + def test_oos_source_intake_accepts_live_not_shipped_wording() -> None: patterns = benchmark.CONCEPT_PATTERNS["bounded_intake_tier"] reply = "The local compiler is proven; live VPS/chat intake is not shipped."