"""Source-control checks for the Hermes leoclean KB bridge files.""" from __future__ import annotations import ast import contextlib import importlib.util import io import json import re import subprocess from pathlib import Path from types import SimpleNamespace import pytest from scripts.working_leo_open_ended_benchmark import ( M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS, score_reply, ) ROOT = Path(__file__).resolve().parents[1] BRIDGE_DIR = ROOT / "hermes-agent" / "leoclean-bin" DB_CONTEXT_PLUGIN = ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py" def test_leoclean_bridge_files_are_present_and_parseable() -> None: cloudsql_tool = BRIDGE_DIR / "cloudsql_memory_tool.py" vps_tool = BRIDGE_DIR / "kb_tool.py" wrapper = BRIDGE_DIR / "teleo-kb" assert cloudsql_tool.exists() assert vps_tool.exists() assert wrapper.exists() ast.parse(cloudsql_tool.read_text()) ast.parse(vps_tool.read_text()) subprocess.run(["bash", "-n", str(wrapper)], check=True) def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None: wrapper_text = (BRIDGE_DIR / "teleo-kb").read_text() cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text() vps_text = (BRIDGE_DIR / "kb_tool.py").read_text() for command in [ "status", "search", "context", "show", "evidence", "edges", "propose-core-change", "propose-edge", "list-proposals", "search-proposals", "show-proposal", "decision-matrix-status", ]: assert command in wrapper_text assert command in cloudsql_text assert "status" in vps_text assert "kb_stage.kb_proposals" in cloudsql_text assert "canonical apply" in cloudsql_text.lower() def test_bridge_edge_proposals_stage_strict_apply_payloads() -> None: cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text() vps_text = (BRIDGE_DIR / "kb_tool.py").read_text() for text in (cloudsql_text, vps_text): assert "'proposal_type', proposal_type" in text or "select 'add_edge'" in text assert "'apply_payload', jsonb_build_object(" in text assert "'from_claim', from_row.id::text" in text assert "'to_claim', to_row.id::text" in text assert "'edge_type', p.edge_type::text" in text def test_bridge_source_does_not_commit_raw_secret_values() -> None: combined = "\n".join(path.read_text(errors="replace") for path in BRIDGE_DIR.iterdir() if path.is_file()) forbidden_patterns = [ r"bot\d+:[A-Za-z0-9_-]{20,}", r"postgres://[^\\s]+:[^\\s]+@", r"-----BEGIN [A-Z ]*PRIVATE KEY-----", r"sk-[A-Za-z0-9]{20,}", r"xox[baprs]-[A-Za-z0-9-]{20,}", r"AIza[0-9A-Za-z_-]{20,}", ] for pattern in forbidden_patterns: assert not re.search(pattern, combined), pattern def _load_module(path: Path): spec = importlib.util.spec_from_file_location(path.stem, path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module 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] = [] def fake_psql_json(_args, sql): captured_sql.append(sql) return [ { "id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "status": "approved", "proposal_type": "attach_evidence", "payload": {}, } ] monkeypatch.setattr(module, "psql_json", fake_psql_json) args = SimpleNamespace(query="Helmer 7 Powers", status="all", limit=20) result = module.search_proposals(args) assert result["proposals"][0]["status"] == "approved" assert result["proposals"][0]["readiness"] == { "review_state": "approved_needs_apply_payload", "has_apply_payload": False, "worker_supported_type": True, "worker_contract_applyable": False, "production_worker_enabled": None, "guidance": ( "This is proposal-contract readiness only. It does not prove the production apply worker is enabled, " "the payload passes strict validation, or apply is authorized." ), } assert {"helmer", "powers"} <= set(result["terms"]) sql = captured_sql[0] assert "payload::text ilike any" in sql assert "coalesce(rationale, '') ilike any" in sql assert "status =" not in sql def test_vps_bridge_proposal_list_prints_rationale_for_non_edge_rows(capsys) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") module.print_proposal_list( [ { "id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "proposal_type": "attach_evidence", "status": "approved", "proposed_by_handle": "leo", "channel": "telegram", "created_at": "2026-07-09 00:00:00+00", "reviewed_at": "2026-07-09 00:10:00+00", "applied_at": None, "payload": {"title": "Helmer 7 Powers"}, "rationale": "Revised Helmer packet should remain visible as approved but unapplied.", } ] ) output = capsys.readouterr().out assert "attach_evidence" in output assert "approved" in output assert "applied_at: `-`" in output assert "Revised Helmer packet" in output assert "approved_needs_apply_payload" in output assert "worker_contract_applyable: `False`" in output def test_vps_bridge_proposal_readiness_separates_contract_from_live_worker_state() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") legacy_approved = module.classify_proposal_readiness( {"status": "approved", "proposal_type": "approve_claim", "payload": {}} ) strict_approved = module.classify_proposal_readiness( { "status": "approved", "proposal_type": "approve_claim", "payload": {"apply_payload": {"contract_version": 2}}, } ) pending_strict = module.classify_proposal_readiness( { "status": "pending_review", "proposal_type": "approve_claim", "payload": {"apply_payload": {"contract_version": 2}}, } ) unsupported_approved = module.classify_proposal_readiness( {"status": "approved", "proposal_type": "change_policy", "payload": {}} ) assert legacy_approved["review_state"] == "approved_needs_apply_payload" assert legacy_approved["worker_contract_applyable"] is False assert strict_approved["review_state"] == "approved_contract_present" assert strict_approved["worker_contract_applyable"] is True assert pending_strict["review_state"] == "needs_human_review" assert pending_strict["worker_contract_applyable"] is False assert unsupported_approved["review_state"] == "unsupported_by_apply_worker_contract" assert unsupported_approved["worker_contract_applyable"] is False assert strict_approved["production_worker_enabled"] is None assert "does not prove the production apply worker is enabled" in strict_approved["guidance"] def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") reply_budget = module.operational_contracts("How should Leo answer?")[0] assert reply_budget["target_words"] == 150 assert reply_budget["hard_max_words"] == 200 assert "omission is better" in reply_budget["shape"] source_contracts = { item["id"]: item for item in module.operational_contracts("I hand Leo a document. What can it absorb and stage before approval?") } source = source_contracts["source_intake"] assert source["required_lead"] == ( "VPS filesystem document preparation and source-to-proposal staging are shipped; automatic Telegram " "attachment capture is not shipped." ) assert source["capability_tier"] == ( "VPS-local document preparation, canonical dedupe retrieval, and pending-review staging" ) assert source["current_public_sources_columns"] == list(module.CURRENT_PUBLIC_SCHEMA["sources"]) assert "author/title/publisher/publication date are current public.sources columns" in source["must_not_claim"] broad_source_query = ( "If I send you a report here, can you learn it into your knowledge base and make it part of what you know? " "What actually happens right now?" ) broad_source_contracts = {item["id"]: item for item in module.operational_contracts(broad_source_query)} assert "source_intake" in broad_source_contracts broad_source_response = module.compile_operational_response(list(broad_source_contracts.values())) assert broad_source_response is not None assert ( "Sending a report in chat does not by itself create a proposal or persistent knowledge" in broad_source_response ) assert "Telegram attachment downloader is not connected" in broad_source_response assert "teleo-kb prepare-source" in broad_source_response assert "teleo-kb propose-source" in broad_source_response assert "writes only kb_stage.kb_proposals" in broad_source_response mixed_contracts = { item["id"]: item for item in module.operational_contracts( "Compose a packet with a factual observation, strategic framework, governance rule, and old belief." ) } mixed = mixed_contracts["mixed_packet_composition"] assert mixed["map"]["reusable_framework"] == "reasoning_tools" assert mixed["approve_claim_supported"] == ["claims", "sources", "evidence", "edges", "reasoning_tools"] assert "belief updates" in mixed["keep_staged_until_separate_capability"] assert "reasoning_tools has no scope field" in mixed["schema_guards"] assert mixed["current_columns"]["reasoning_tools"] == list(module.CURRENT_PUBLIC_SCHEMA["reasoning_tools"]) assert mixed["current_columns"]["behavioral_rules"] == list(module.CURRENT_PUBLIC_SCHEMA["behavioral_rules"]) runtime_ids = { item["id"] for item in module.operational_contracts( "Do unchanged database totals after restart prove answer behavior and erase the prior session?" ) } assert {"reply_budget", "runtime_persistence"} <= runtime_ids def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extraction(tmp_path: Path, monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") preparer = tmp_path / "prepare_kb_source_manifest.py" preparer.write_text("# fixture\n", encoding="utf-8") inbox = tmp_path / "source-inbox" preparation_root = tmp_path / "source-preparation" inbox.mkdir() preparation_root.mkdir() artifact = inbox / "source.md" artifact.write_text("Canonical review precedes guarded apply.", encoding="utf-8") output_dir = preparation_root / "private-output" candidate_id = "11111111-1111-4111-8111-111111111111" captured: dict = {} monkeypatch.setattr(module, "SOURCE_PREPARER_PATH", preparer) monkeypatch.setattr(module, "SOURCE_INBOX_ROOT", inbox) monkeypatch.setattr(module, "SOURCE_PREPARATION_ROOT", preparation_root) monkeypatch.setattr( module, "find_claims", lambda _args, query, limit: [ { "id": candidate_id, "text": "Canonical writes require guarded apply.", "score": 2, "evidence_count": 3, "edge_count": 1, } ], ) def fake_run(command, **_kwargs): context_path = Path(command[command.index("--kb-context") + 1]) captured["context"] = json.loads(context_path.read_text(encoding="utf-8")) captured["command"] = command receipt = { "schema": module.SOURCE_PREPARATION_RECEIPT_SCHEMA, "status": "prepared", "database_write_performed": False, "canonical_apply_performed": False, "retrieval": {"canonical_candidate_ids": [candidate_id]}, "extraction": {"model": "fixture/model", "claim_count": 1}, "outputs": {}, } return subprocess.CompletedProcess(command, 0, stdout=json.dumps(receipt), stderr="") monkeypatch.setattr(module.subprocess, "run", fake_run) args = SimpleNamespace( local=True, query="canonical review guarded apply", artifact=artifact, text=None, identity="document:test", source_key="test_source", source_type="article", title="Test source", locator="artifact://test/source", output_dir=output_dir, model="fixture/model", ) receipt = module.prepare_source(args) assert receipt["status"] == "prepared" assert captured["context"]["database_search_query"] == args.query assert captured["context"]["candidate_claims"][0]["id"] == candidate_id assert "--text" not in captured["command"] def test_vps_prepare_source_rejects_files_and_outputs_outside_private_roots(tmp_path: Path) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") inbox = tmp_path / "source-inbox" preparation = tmp_path / "source-preparation" outside = tmp_path / "outside.md" inbox.mkdir() preparation.mkdir() outside.write_text("private", encoding="utf-8") with pytest.raises(SystemExit, match="must stay under"): module._scoped_source_path(outside, inbox, "--artifact", must_exist=True) with pytest.raises(SystemExit, match="must stay under"): module._scoped_source_path(tmp_path / "outside-output", preparation, "--output-dir", must_exist=False) def test_vps_bridge_broad_db_questions_select_query_specific_live_contracts() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") cases = { "What has actually changed in the KB rather than sitting in proposed state?": "proposal_state_readback", "What changed in the KB?": "proposal_state_readback", "What has changed in the knowledge base?": "proposal_state_readback", "Has Helmer's Seven Powers framework landed as live knowledge?": "named_packet_readback", "Was that accepted through the decision-matrix?": "decision_matrix_readback", "Is the proposal backlog stuck on missing document-to-source links?": "source_link_audit", ( "Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal " "exists. What is actually true right now, what would you inspect, and what must happen before those " "claims are canonical? Do not mutate anything." ): "proposal_state_readback", "Can we demo a real database write from Leo?": "demo_capability_readback", "Does editing the SOUL file alter the source of truth for identity?": "identity_canonicality_readback", ( "The current visible Telegram sender is @m3taversal. What should Leo call this participant, which " "identity sources are allowed, and how should Leo avoid mixing identities?" ): "telegram_participant_identity", } for query, expected_id in cases.items(): contract_ids = {item["id"] for item in module.operational_contracts(query)} assert expected_id in contract_ids, query def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") direct_ids = set(module.DIRECT_READBACK_CONTRACTS) for query in ( "What changed in the KB scorer?", "Who is Helmer?", "Can I demo the database dashboard?", "Is the KB live search broken?", ): contract_ids = {item["id"] for item in module.operational_contracts(query)} assert not contract_ids & direct_ids, query overlap_ids = { item["id"] for item in module.operational_contracts( "Did the decision matrix reject the pending proposal because its document pointer is missing?" ) } assert overlap_ids & direct_ids == {"decision_matrix_readback"} assert "source_intake" not in overlap_ids lifecycle_ids = { item["id"] for item in module.operational_contracts("What changed in the KB after the benchmark proposal was approved?") } assert lifecycle_ids & direct_ids == {"proposal_state_readback"} substring_ids = { item["id"] for item in module.operational_contracts("What changed in the KB after the capital allocation update?") } assert substring_ids & direct_ids == {"proposal_state_readback"} v3_ids = { item["id"] for item in module.operational_contracts( "Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal " "exists. What is actually true right now, what would you inspect, and what must happen before those " "claims are canonical? Do not mutate anything." ) } assert v3_ids & direct_ids == {"proposal_state_readback"} assert "source_link_audit" not in v3_ids def test_vps_bridge_named_document_question_matches_exact_proposal_and_compiles_receipt(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") database_status, approved, matrix, source_audit = _direct_contract_fixtures() database_status = { **database_status, "high_signal_rows": {**database_status["high_signal_rows"], "kb_proposals": 29}, "proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 16, "canceled": 8}, } source_metadata = { "apply_payload": { "sources": [ { "excerpt": json.dumps( { "metadata": { "title": "LivingIP KB Agent Identity Architecture v3", "source_key": "livingip_kb_agent_identity_architecture_v3", } } ) } ] } } corrected = { **approved, "id": "10bc0719-1c2b-5f42-a41e-86e6478692cb", "status": "pending_review", "source_ref": "normalized:12e38cb0-b72b-557c-95e5-ffc789a29f62:72895c9b58b79d41", "payload": source_metadata, "created_at": "2026-07-13T04:00:00+00:00", "reviewed_at": None, "applied_at": None, } canceled = { **corrected, "id": "a9eb4158-5aa2-5484-b5a7-3d249588b3ec", "status": "canceled", "created_at": "2026-07-13T03:00:00+00:00", } monkeypatch.setattr( module, "load_direct_readback_snapshot", lambda _args: { "snapshot_id": "300:300:", "statement_timestamp": "2026-07-13T05:00:00+00:00", "database_status": database_status, "proposals": [approved, canceled, corrected], "named_packet_proposals": [approved], "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []}, "decision_matrix_status": matrix, "source_link_audit": source_audit, }, ) prompt = ( "Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal " "exists. What is actually true right now, what would you inspect, and what must happen before those claims " "are canonical? Do not mutate anything." ) contracts = module.context_operational_contracts(SimpleNamespace(), prompt) proposal_state = next(item for item in contracts if item["id"] == "proposal_state_readback") response = module.compile_operational_response(contracts) assert proposal_state["proposal_query_terms"] == ["v3", "architecture"] assert [item["id"] for item in proposal_state["matching_proposals"]] == [corrected["id"], canceled["id"]] assert response is not None assert response.startswith("No.\nFresh live readback.") assert f"proposal: {corrected['id']}; status: pending_review; applied_at: none" in response assert corrected["source_ref"] in response assert "proposal states are applied: 2, approved: 3, pending_review: 16, canceled: 8" in response.lower() assert "Approved is not applied." in response assert "has not reached canonical apply" in response assert "telegram_file_refs" not in response assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200 assert _load_module(DB_CONTEXT_PLUGIN).response_contract_issues(response, contracts) == [] 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" }, ( "The current visible Telegram sender is @m3taversal. An earlier answer shortened that handle. What " "should Leo call this participant, which identity sources are allowed, and how should Leo avoid mixing " "identities when another user replies? Do not write to memory or the KB." ): {"telegram_participant_identity"}, } 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": { "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", "source_ref": "telegram:helmer-7-powers", "created_at": "2026-06-29T00:00:00+00:00", "reviewed_at": "2026-06-29T00:10:00+00:00", "applied_at": None, "readiness": {"review_state": "approved_needs_apply_payload"}, } matrix = { "decision_matrix_tables": { f"{schema}.{table}": False for schema in ("public", "kb_stage") for table in ("matrix_voters", "proposal_votes", "proposal_decisions") } } source_audit = { "pending_or_approved": 17, "with_source_ref": 16, "without_source_ref": 1, "exact_public_source_matches": 2, "samples": [], } return database_status, approved, matrix, source_audit def test_vps_bridge_enriches_and_compiles_all_direct_questions_from_live_rows(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") database_status, approved, matrix, source_audit = _direct_contract_fixtures() applied = { **approved, "id": "11111111-1111-4111-8111-111111111111", "status": "applied", "applied_at": "2026-07-01", } monkeypatch.setattr( module, "load_direct_readback_snapshot", lambda _args: { "snapshot_id": "100:100:", "statement_timestamp": "2026-07-13T12:00:00+00:00", "database_status": database_status, "proposals": [approved, applied], "named_packet_proposals": [approved], "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []}, "decision_matrix_status": matrix, "source_link_audit": source_audit, }, ) for prompt in M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS: contracts = module.context_operational_contracts(SimpleNamespace(), prompt["message"]) response = module.compile_operational_response(contracts) demo_contract = next( (contract for contract in contracts if contract["id"] == "demo_capability_readback"), None, ) if demo_contract: assert [row["status"] for row in demo_contract["applied_and_approved_proposals"]] == [ "approved", "applied", ] assert "1/1 approved rows are approved_needs_apply_payload" in response assert response is not None, prompt["id"] assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200 score = score_reply(prompt, response) assert score["pass"] is True, score def test_vps_bridge_direct_readback_uses_one_statement_snapshot(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") database_status, approved, matrix, source_audit = _direct_contract_fixtures() captured_sql: list[str] = [] expected = { "snapshot_id": "200:200:", "statement_timestamp": "2026-07-13T12:00:00+00:00", "database_status": database_status, "proposals": [approved], "named_packet_proposals": [approved], "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []}, "decision_matrix_status": matrix, "source_link_audit": source_audit, } def fake_psql_json(_args, sql): captured_sql.append(sql) return [expected] monkeypatch.setattr(module, "psql_json", fake_psql_json) assert module.load_direct_readback_snapshot(SimpleNamespace()) == expected assert len(captured_sql) == 1 sql = captured_sql[0] assert "txid_current_snapshot()" in sql assert "named_packet_proposals" in sql assert "search_text like '%helmer%' and search_text like '%power%'" in sql assert "decision_matrix_status" in sql assert "source_link_audit" in sql def test_vps_bridge_named_packet_compiler_uses_observed_applied_state() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") database_status, approved, _matrix, _source_audit = _direct_contract_fixtures() applied = { **approved, "status": "applied", "applied_at": "2026-07-01T00:00:00+00:00", "readiness": {"review_state": "applied"}, } contracts = module.operational_contracts("Is Helmer's 7 Powers in Leo now?") named = next(item for item in contracts if item["id"] == "named_packet_readback") named.update( { "database_status": database_status, "matching_proposals": [applied], "canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []}, } ) response = module.compile_operational_response(contracts) assert response is not None assert "status: applied" in response assert "applied_at: 2026-07-01" in response assert "applied_at is empty" not in response assert "does not prove which canonical rows were written" in response def test_vps_bridge_matrix_compiler_handles_all_tables_present_without_overclaim() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") database_status, _approved, matrix, _source_audit = _direct_contract_fixtures() matrix["decision_matrix_tables"] = {name: True for name in matrix["decision_matrix_tables"]} contracts = module.operational_contracts("Did the decision matrix approve this already?") contract = next(item for item in contracts if item["id"] == "decision_matrix_readback") contract.update({"database_status": database_status, "live_matrix_status": matrix}) response = module.compile_operational_response(contracts) assert response is not None assert "all required matrix tables are present" in response assert "proposal-specific vote or decision rows" in response assert "matrix approval is not proven" in response assert "0 matrix tables are absent ()" not in response def test_vps_bridge_matrix_compiler_treats_empty_or_partial_table_state_as_unobserved() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") database_status, _approved, _matrix, _source_audit = _direct_contract_fixtures() for table_states, unobserved_count in (({}, 6), ({"public.matrix_voters": True}, 5)): contracts = module.operational_contracts("Did the decision matrix approve this already?") contract = next(item for item in contracts if item["id"] == "decision_matrix_readback") contract.update( { "database_status": database_status, "live_matrix_status": {"decision_matrix_tables": table_states}, } ) response = module.compile_operational_response(contracts) assert response is not None assert f"{unobserved_count} required tables were not observed" in response assert "presence or absence cannot be inferred" in response assert "matrix tables are absent" not in response assert "all required matrix tables are present" not in response def test_vps_bridge_source_link_audit_checks_exact_staging_to_canonical_links(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") _database_status, _approved, _matrix, source_audit = _direct_contract_fixtures() captured_sql: list[str] = [] def fake_psql_json(_args, sql): captured_sql.append(sql) return [source_audit] monkeypatch.setattr(module, "psql_json", fake_psql_json) assert module.load_source_link_audit(SimpleNamespace()) == source_audit sql = captured_sql[0] assert "p.source_ref" in sql assert "s.id::text = p.source_ref" in sql assert "s.url = p.source_ref" in sql assert "s.storage_path = p.source_ref" in sql assert "p.status in ('pending_review', 'approved')" in sql def test_vps_bridge_compiles_safe_mixed_packet_response_from_contracts() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") contracts = module.operational_contracts( "Compose a packet with a factual observation, strategic framework, disputed interpretation, governance " "rule, and old belief." ) response = module.compile_operational_response(contracts) assert response is not None assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200 assert "approve_claim applies only claims, sources, evidence, edges, and reasoning_tools" in response assert "approve_claim supports neither behavioral_rules nor governance_gates" in response assert "belief updates, and existing-row updates remain staged" in response assert "proposal-level applied_at" in response def test_vps_bridge_context_markdown_surfaces_runtime_contracts(capsys) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") data = { "query": "absorb this document", "operational_contracts": module.operational_contracts("absorb this document"), "context_rows": [], "claims": [], } module.print_claim_bundle(data) output = capsys.readouterr().out assert "## Current Runtime Contracts" in output assert '"id": "source_intake"' in output assert "automatic Telegram attachment capture is not shipped" in output def test_vps_bridge_context_contract_reads_current_schema_from_postgres(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 "information_schema.columns" in sql: return [{"table_name": "sources", "columns": ["id", "source_type", "url", "hash", "new_live_column"]}] return [{"table_name": "sources", "constraints": {"sources_pkey": "PRIMARY KEY (id)"}}] monkeypatch.setattr(module, "psql_json", fake_psql_json) contracts = { item["id"]: item for item in module.context_operational_contracts( SimpleNamespace(), "Absorb this document into a staged packet." ) } assert contracts["source_intake"]["current_public_sources_columns"] == [ "id", "source_type", "url", "hash", "new_live_column", ] assert len(captured_sql) == 2 assert "information_schema.columns" in captured_sql[0] assert "table_name = any" in captured_sql[0] assert "pg_constraint" in captured_sql[1] def test_vps_bridge_mixed_contract_reads_live_postgres_constraints(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") def fake_psql_json(_args, sql): if "information_schema.columns" in sql: return [{"table_name": "claims", "columns": ["id", "type", "text", "status"]}] return [ { "table_name": "claims", "constraints": { "claims_type_check": "CHECK ((type = ANY (ARRAY['empirical', 'structural'])))", "claims_status_check": "CHECK ((status = ANY (ARRAY['open', 'retired'])))", }, } ] monkeypatch.setattr(module, "psql_json", fake_psql_json) contracts = { item["id"]: item for item in module.context_operational_contracts( SimpleNamespace(), "Compose a packet with factual observations, a strategic framework, and a governance rule.", ) } mixed = contracts["mixed_packet_composition"] assert "'empirical'" in mixed["current_constraints"]["claims"]["claims_type_check"] assert "'open'" in mixed["current_constraints"]["claims"]["claims_status_check"] assert "binding operating/governance rules" in mixed["table_semantics"]["behavioral_rules"] assert "evaluative pass/fail gates" in mixed["table_semantics"]["governance_gates"] assert "supports neither behavioral_rules nor governance_gates" in mixed["unsupported_apply_boundary"] 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] = [] def fake_psql_json(_args, sql): captured_sql.append(sql) return [ { "artifact": "teleo_kb_decision_matrix_status", "all_required_tables_present": False, "any_decision_matrix_table_present": False, "decision_matrix_tables": {}, "proposal_status_counts": {}, "guidance": "Decision-matrix approval schema is absent or incomplete.", } ] monkeypatch.setattr(module, "psql_json", fake_psql_json) result = module.decision_matrix_status(SimpleNamespace()) assert result["all_required_tables_present"] is False sql = captured_sql[0] for table in ( "public.matrix_voters", "public.proposal_votes", "public.proposal_decisions", "kb_stage.matrix_voters", "kb_stage.proposal_votes", "kb_stage.proposal_decisions", ): assert table in sql def test_vps_bridge_status_reads_all_direct_claim_counts(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") captured_sql: list[str] = [] expected = { "artifact": "teleo_kb_status", "backend": "teleo|postgres", "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}, } def fake_psql_json(_args, sql): captured_sql.append(sql) return [expected] monkeypatch.setattr(module, "psql_json", fake_psql_json) assert module.status(SimpleNamespace()) == expected sql = captured_sql[0] for table in ( "public.claims", "public.sources", "public.claim_edges", "public.claim_evidence", "kb_stage.kb_proposals", ): assert table in sql assert not re.search(r"\b(?:insert|update|delete|alter|drop|create)\b", sql, re.I) def test_vps_bridge_status_prints_copyable_db_readback(capsys) -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") module.print_status( { "artifact": "teleo_kb_status", "backend": "teleo|postgres", "high_signal_rows": { "claims": 1837, "sources": 4145, "claim_edges": 4916, "claim_evidence": 4670, "kb_proposals": 26, }, "proposal_status_counts": {"applied": 2}, } ) assert ( "DB readback: claims: `1837`; sources: `4145`; claim_edges: `4916`; " "claim_evidence: `4670`; kb_proposals: `26`." in capsys.readouterr().out ) def test_cloudsql_status_prints_copyable_canonical_db_readback(capsys) -> None: module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") module.emit_markdown( { "artifact": "teleo_cloudsql_memory_status", "backend": "cloudsql:test", "db_identity": "audit|reader", "canonical": { "db_identity": "teleo|reader", "schema_tables": {"public": 33, "kb_stage": 6}, "high_signal_rows": { "claims": 1837, "sources": 4145, "claim_edges": 4916, "claim_evidence": 4670, "kb_proposals": 26, }, }, "extensions": [], "teleo_restore_tables": 8, "total_rows": 52164, "high_signal_rows": {}, } ) assert ( "DB readback: claims: `1837`; sources: `4145`; claim_edges: `4916`; " "claim_evidence: `4670`; kb_proposals: `26`." in capsys.readouterr().out ) def test_cloudsql_bridge_matches_direct_claim_readback_commands(monkeypatch) -> None: module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py") captured_sql: list[str] = [] def fake_psql_json_lines(_args, sql, db=None): captured_sql.append(sql) if "matrix_voters" in sql: return [ { "artifact": "teleo_cloudsql_kb_decision_matrix_status", "all_required_tables_present": False, "any_decision_matrix_table_present": False, "decision_matrix_tables": {}, "proposal_status_counts": {}, "guidance": "Decision-matrix approval schema is absent or incomplete.", } ] return [{"id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "status": "approved"}] monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines) args = SimpleNamespace(query="Helmer 7 Powers", status="all", limit=20, canonical_db="teleo") proposal_result = module.search_proposals(args) matrix_result = module.decision_matrix_status(SimpleNamespace(canonical_db="teleo")) assert proposal_result["artifact"] == "teleo_cloudsql_kb_proposal_search" assert proposal_result["proposals"][0]["status"] == "approved" assert matrix_result["artifact"] == "teleo_cloudsql_kb_decision_matrix_status" assert any("payload::text ilike any" in sql for sql in captured_sql) assert any("public.proposal_decisions" in sql for sql in captured_sql) def test_kb_bridges_emit_public_claim_links_for_telegram_rendering() -> None: claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1" expected = f"https://leo.livingip.xyz/kb/claims/{claim_id}" for filename in ("kb_tool.py", "cloudsql_memory_tool.py"): module = _load_module(BRIDGE_DIR / filename) assert module.claim_url(claim_id) == expected assert module.markdown_claim_link(claim_id, "d3fb892b") == (f"[`d3fb892b`]({expected})") assert module.markdown_claim_link(claim_id, "claim `with ticks`") == (f"[`claim 'with ticks'`]({expected})") def test_vps_bridge_markdown_links_claim_text_and_edges() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py") claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1" connected_id = "9d5281fe-7ee4-4fe1-b1bf-c54c4d7fb6a5" data = { "query": "strategy kernel", "context_rows": [], "claims": [ { "id": claim_id, "text": "Claims should be easy to scan in Telegram.", "type": "belief", "confidence": 0.8, "tags": ["telegram"], "score": 4, "evidence_count": 2, "edge_count": 1, "evidence": [], "edges": [ { "direction": "outgoing", "edge_type": "supports", "connected_id": connected_id, "connected_text": "Full claim pages expose body, evidence, and graph edges.", } ], } ], } buffer = io.StringIO() with contextlib.redirect_stdout(buffer): module.print_claim_bundle(data) markdown = buffer.getvalue() assert f"[`Claims should be easy to scan in Telegram.`](https://leo.livingip.xyz/kb/claims/{claim_id})" in markdown assert f"- open full claim/body/edges: [`claim page`](https://leo.livingip.xyz/kb/claims/{claim_id})" in markdown assert ( f"[`Full claim pages expose body, evidence, and graph edges.`](https://leo.livingip.xyz/kb/claims/{connected_id})" in markdown )