diff --git a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py index 3b20567..e3d8fa6 100755 --- a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py +++ b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py @@ -124,10 +124,22 @@ def parse_args() -> argparse.Namespace: list_proposals.add_argument("--limit", type=int, default=10) list_proposals.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + search_proposals = sub.add_parser("search-proposals", help="Search staged KB proposals across statuses from Cloud SQL.") + search_proposals.add_argument("query") + search_proposals.add_argument("--status", default="all") + search_proposals.add_argument("--limit", type=int, default=10) + search_proposals.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + show_proposal = sub.add_parser("show-proposal", help="Show one staged KB proposal from Cloud SQL.") show_proposal.add_argument("proposal_id") show_proposal.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + decision_matrix_status = sub.add_parser( + "decision-matrix-status", + help="Read whether decision-matrix approval tables exist before claiming matrix approval.", + ) + decision_matrix_status.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + return parser.parse_args() @@ -966,6 +978,55 @@ limit {max(1, min(args.limit, 100))}; } +def search_proposals(args: argparse.Namespace) -> dict[str, Any]: + terms = terms_for(args.query) + patterns = sql_array([f"%{term}%" for term in terms], "text") + status_clause = "" + if args.status.lower() != "all": + status_clause = f"and status = {sql_literal(args.status)}" + sql = f""" +select jsonb_build_object( + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', left(rationale, 700), + 'payload', payload, + 'created_at', created_at::text, + 'reviewed_at', reviewed_at::text, + 'applied_at', applied_at::text +)::text +from kb_stage.kb_proposals +where ( + coalesce(rationale, '') ilike any({patterns}) + or coalesce(source_ref, '') ilike any({patterns}) + or coalesce(proposal_type, '') ilike any({patterns}) + or coalesce(proposed_by_handle, '') ilike any({patterns}) + or payload::text ilike any({patterns}) +) +{status_clause} +order by + case status + when 'approved' then 0 + when 'pending_review' then 1 + when 'applied' then 2 + else 3 + end, + created_at desc +limit {max(1, min(args.limit, 100))}; +""" + return { + "artifact": "teleo_cloudsql_kb_proposal_search", + "backend": canonical_backend(args), + "query": args.query, + "terms": terms, + "status_filter": args.status, + "proposals": psql_json_lines(args, sql, db=args.canonical_db), + } + + def show_proposal(args: argparse.Namespace) -> dict[str, Any]: sql = f""" select jsonb_build_object( @@ -997,6 +1058,54 @@ where id = {sql_literal(args.proposal_id)}::uuid; return rows[0] +def decision_matrix_status(args: argparse.Namespace) -> dict[str, Any]: + sql = """ +with table_status(schema_name, table_name, exists_in_cloudsql) as ( + values + ('public', 'matrix_voters', to_regclass('public.matrix_voters') is not null), + ('public', 'proposal_votes', to_regclass('public.proposal_votes') is not null), + ('public', 'proposal_decisions', to_regclass('public.proposal_decisions') is not null), + ('kb_stage', 'matrix_voters', to_regclass('kb_stage.matrix_voters') is not null), + ('kb_stage', 'proposal_votes', to_regclass('kb_stage.proposal_votes') is not null), + ('kb_stage', 'proposal_decisions', to_regclass('kb_stage.proposal_decisions') is not null) +), +status_counts as ( + select coalesce(jsonb_object_agg(status, proposal_count order by status), '{}'::jsonb) as counts + from ( + select status, count(*) as proposal_count + from kb_stage.kb_proposals + group by status + ) grouped +) +select jsonb_build_object( + 'artifact', 'teleo_cloudsql_kb_decision_matrix_status', + 'backend', current_database() || '|' || current_user, + 'decision_matrix_tables', ( + select jsonb_object_agg(schema_name || '.' || table_name, exists_in_cloudsql order by schema_name, table_name) + from table_status + ), + 'all_required_tables_present', ( + select bool_and(exists_in_cloudsql) from table_status + ), + 'any_decision_matrix_table_present', ( + select bool_or(exists_in_cloudsql) from table_status + ), + 'proposal_status_counts', (select counts from status_counts), + 'guidance', case + when (select bool_and(exists_in_cloudsql) from table_status) + then 'Decision-matrix tables exist; inspect proposal_decisions/proposal_votes for the specific proposal before claiming approval.' + else 'Decision-matrix approval schema is absent or incomplete; do not infer matrix approval from kb_stage proposal rationale/status alone.' + end +)::text; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + if not rows: + raise SystemExit("No decision matrix status returned.") + result = rows[0] + result["backend"] = canonical_backend(args) + return result + + def status(args: argparse.Namespace) -> dict[str, Any]: canonical_sql = """ select json_build_object( @@ -1173,6 +1282,7 @@ def emit_markdown(value: dict[str, Any]) -> None: if value["artifact"] == "teleo_cloudsql_kb_proposal_list": print("# Staged KB Proposals\n") + print(f"- Status filter: `{value.get('status_filter')}`\n") rows = value.get("proposals") or [] if not rows: print("No matching proposals.") @@ -1186,6 +1296,25 @@ def emit_markdown(value: dict[str, Any]) -> None: print() return + if value["artifact"] == "teleo_cloudsql_kb_proposal_search": + print("# Staged KB Proposal Search\n") + print(f"- Query: `{value['query']}`") + print(f"- Terms: `{', '.join(value['terms'])}`") + print(f"- Status filter: `{value['status_filter']}`\n") + rows = value.get("proposals") or [] + if not rows: + print("No matching proposals.") + for row in rows: + print(f"## {row['id']}\n") + print(f"- Type: `{row['proposal_type']}`") + print(f"- Status: `{row['status']}`") + print(f"- Source: `{row.get('source_ref') or '-'}`") + print(f"- Created: `{row.get('created_at')}`") + print(f"- Applied at: `{row.get('applied_at') or '-'}`") + print(f"- Rationale: {row.get('rationale') or ''}") + print() + return + if value["artifact"] == "teleo_cloudsql_kb_proposal": print("# Staged KB Proposal\n") print(f"- Proposal id: `{value['id']}`") @@ -1200,6 +1329,19 @@ def emit_markdown(value: dict[str, Any]) -> None: print(json.dumps(value.get("payload") or {}, indent=2, sort_keys=True)) return + if value["artifact"] == "teleo_cloudsql_kb_decision_matrix_status": + print("# Decision Matrix Status\n") + print(f"- Backend: `{value['backend']}`") + print(f"- all required tables present: `{value['all_required_tables_present']}`") + print(f"- any decision-matrix table present: `{value['any_decision_matrix_table_present']}`") + print(f"- proposal status counts: `{json.dumps(value.get('proposal_status_counts') or {}, sort_keys=True)}`") + print(f"- guidance: {value['guidance']}") + print("\n## Tables\n") + for table, exists in sorted((value.get("decision_matrix_tables") or {}).items()): + print(f"- `{table}`: `{exists}`") + print() + return + print("# Teleo Cloud SQL Memory Context\n") print(f"- Backend: `{value['backend']}`") print(f"- Query: `{value['query']}`") @@ -1234,8 +1376,12 @@ def main() -> None: result = propose_edge(args) elif args.command == "list-proposals": result = list_proposals(args) + elif args.command == "search-proposals": + result = search_proposals(args) elif args.command == "show-proposal": result = show_proposal(args) + elif args.command == "decision-matrix-status": + result = decision_matrix_status(args) else: result = query_rows(args, args.query, args.limit) if args.format == "json": diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index 0cee32d..652bd9c 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -163,10 +163,22 @@ def parse_args() -> argparse.Namespace: list_proposals.add_argument("--limit", type=int, default=10) add_output_flag(list_proposals) + search_proposals = sub.add_parser("search-proposals", help="Search KB mutation proposals across statuses.") + search_proposals.add_argument("query") + search_proposals.add_argument("--status", default="all") + search_proposals.add_argument("--limit", type=int, default=10) + add_output_flag(search_proposals) + show_proposal = sub.add_parser("show-proposal", help="Show one KB mutation proposal.") show_proposal.add_argument("proposal_id") add_output_flag(show_proposal) + decision_matrix_status = sub.add_parser( + "decision-matrix-status", + help="Read whether decision-matrix approval tables exist before claiming matrix approval.", + ) + add_output_flag(decision_matrix_status) + return parser.parse_args() @@ -647,6 +659,53 @@ def list_proposals(args: argparse.Namespace) -> list[dict[str, Any]]: return psql_json(args, sql) +def search_proposals(args: argparse.Namespace) -> dict[str, Any]: + terms = query_terms(args.query) + patterns = sql_array([f"%{term}%" for term in terms], "text") + status_clause = "" + if args.status.lower() != "all": + status_clause = f"and status = {sql_literal(args.status)}" + sql = f""" + select jsonb_build_object( + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', left(rationale, 700), + 'payload', payload, + 'created_at', created_at::text, + 'reviewed_at', reviewed_at::text, + 'applied_at', applied_at::text + )::text + from kb_stage.kb_proposals + where ( + coalesce(rationale, '') ilike any({patterns}) + or coalesce(source_ref, '') ilike any({patterns}) + or coalesce(proposal_type, '') ilike any({patterns}) + or coalesce(proposed_by_handle, '') ilike any({patterns}) + or payload::text ilike any({patterns}) + ) + {status_clause} + order by + case status + when 'approved' then 0 + when 'pending_review' then 1 + when 'applied' then 2 + else 3 + end, + created_at desc + limit {max(1, min(args.limit, 100))}; + """ + return { + "query": args.query, + "terms": terms, + "status_filter": args.status, + "proposals": psql_json(args, sql), + } + + def show_proposal(args: argparse.Namespace) -> dict[str, Any] | None: sql = f""" select jsonb_build_object( @@ -674,6 +733,51 @@ def show_proposal(args: argparse.Namespace) -> dict[str, Any] | None: return rows[0] if rows else None +def decision_matrix_status(args: argparse.Namespace) -> dict[str, Any]: + sql = """ + with table_status(schema_name, table_name, exists_on_vps) as ( + values + ('public', 'matrix_voters', to_regclass('public.matrix_voters') is not null), + ('public', 'proposal_votes', to_regclass('public.proposal_votes') is not null), + ('public', 'proposal_decisions', to_regclass('public.proposal_decisions') is not null), + ('kb_stage', 'matrix_voters', to_regclass('kb_stage.matrix_voters') is not null), + ('kb_stage', 'proposal_votes', to_regclass('kb_stage.proposal_votes') is not null), + ('kb_stage', 'proposal_decisions', to_regclass('kb_stage.proposal_decisions') is not null) + ), + status_counts as ( + select coalesce(jsonb_object_agg(status, proposal_count order by status), '{}'::jsonb) as counts + from ( + select status, count(*) as proposal_count + from kb_stage.kb_proposals + group by status + ) grouped + ) + select jsonb_build_object( + 'artifact', 'teleo_kb_decision_matrix_status', + 'decision_matrix_tables', ( + select jsonb_object_agg(schema_name || '.' || table_name, exists_on_vps order by schema_name, table_name) + from table_status + ), + 'all_required_tables_present', ( + select bool_and(exists_on_vps) from table_status + ), + 'any_decision_matrix_table_present', ( + select bool_or(exists_on_vps) from table_status + ), + 'proposal_status_counts', (select counts from status_counts), + 'guidance', case + when (select bool_and(exists_on_vps) from table_status) + then 'Decision-matrix tables exist; inspect proposal_decisions/proposal_votes for the specific proposal before claiming approval.' + else 'Decision-matrix approval schema is absent or incomplete; do not infer matrix approval from kb_stage proposal rationale/status alone.' + end + )::text; + """ + rows = psql_json(args, sql) + if not rows: + raise SystemExit("No decision matrix status returned.") + return rows[0] + + def load_evidence(args: argparse.Namespace, claim_ids: list[str], limit: int) -> dict[str, list[dict[str, Any]]]: if not claim_ids: return {} @@ -882,6 +986,27 @@ def print_proposal_list(rows: list[dict[str, Any]]) -> None: print() +def print_proposal_search(data: dict[str, Any]) -> None: + print("# KB Proposal Search\n") + print(f"- query: `{data['query']}`") + print(f"- terms: `{', '.join(data['terms'])}`") + print(f"- status filter: `{data['status_filter']}`") + print() + print_proposal_list(data["proposals"]) + + +def print_decision_matrix_status(data: dict[str, Any]) -> None: + print("# Decision Matrix Status\n") + print(f"- all required tables present: `{data['all_required_tables_present']}`") + print(f"- any decision-matrix table present: `{data['any_decision_matrix_table_present']}`") + print(f"- proposal status counts: `{json.dumps(data.get('proposal_status_counts') or {}, sort_keys=True)}`") + print(f"- guidance: {data['guidance']}") + print("\n## Tables\n") + for table, exists in sorted((data.get("decision_matrix_tables") or {}).items()): + print(f"- `{table}`: `{exists}`") + print() + + def main() -> int: args = parse_args() if args.command == "search": @@ -920,11 +1045,15 @@ def main() -> int: data = record_document_evaluation(args) elif args.command == "list-proposals": data = {"proposals": list_proposals(args)} + elif args.command == "search-proposals": + data = search_proposals(args) elif args.command == "show-proposal": proposal = show_proposal(args) if not proposal: raise SystemExit(f"proposal not found: {args.proposal_id}") data = proposal + elif args.command == "decision-matrix-status": + data = decision_matrix_status(args) else: raise AssertionError(args.command) @@ -946,8 +1075,12 @@ def main() -> int: print_json(data) elif args.command == "list-proposals": print_proposal_list(data["proposals"]) + elif args.command == "search-proposals": + print_proposal_search(data) elif args.command == "show-proposal": print_proposal(data) + elif args.command == "decision-matrix-status": + print_decision_matrix_status(data) else: print_json(data) return 0 diff --git a/hermes-agent/leoclean-bin/teleo-kb b/hermes-agent/leoclean-bin/teleo-kb index 2548c3a..6aa00a1 100755 --- a/hermes-agent/leoclean-bin/teleo-kb +++ b/hermes-agent/leoclean-bin/teleo-kb @@ -10,7 +10,7 @@ CLOUDSQL_STATUS_TIMEOUT=${TELEO_CLOUDSQL_STATUS_TIMEOUT_SECONDS:-30} cloudsql_supported() { case "$COMMAND" in - status|search|context|show|evidence|edges|propose-core-change|propose-edge|list-proposals|show-proposal) return 0 ;; + status|search|context|show|evidence|edges|propose-core-change|propose-edge|list-proposals|search-proposals|show-proposal|decision-matrix-status) return 0 ;; *) return 1 ;; esac } diff --git a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md index d769c6d..a7844a1 100644 --- a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md +++ b/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md @@ -35,7 +35,9 @@ Use narrower bridge commands when needed: /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb evidence /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb edges /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb list-proposals +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search-proposals "" /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show-proposal +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb decision-matrix-status /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb propose-core-change --proposal-type revise_claim --target-kind claim --target-ref "" --proposed "" --source-ref "" --rationale "" ``` @@ -50,6 +52,25 @@ good default is: 3. final answer with what is grounded, what is weak, and what evidence would improve it. +For no-context direct claims such as "Is X in Leo now?", "did the DB change?", +"did the decision matrix approve this?", or "is it still just proposals?", do +not stop at `search` or default `list-proposals`. Run the status-specific +proposal and governance readbacks needed to avoid overclaiming: + +```bash +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search-proposals "" --status all --limit 20 +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb decision-matrix-status +``` + +If `decision-matrix-status` says the matrix tables are absent or incomplete, +do not infer matrix approval from proposal rationale, reviewer notes, or +`kb_stage.kb_proposals.status`. Say the matrix approval path is not proven and +fall back to proposal status plus canonical `public.*` readback. + +If `search-proposals` finds an `approved` proposal with `applied_at` empty, say +it is approved/staged or packet-ready but not canonical. Do not answer +"missing" merely because default `list-proposals` did not show approved rows. + ## Telegram Rendering Make KB answers easy to scan in Telegram: diff --git a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md index 7f3d988..f908cf4 100644 --- a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md +++ b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md @@ -29,7 +29,9 @@ Use narrower bridge commands when needed: /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb evidence /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb edges /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb list-proposals +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search-proposals "" /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show-proposal +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb decision-matrix-status ``` ## Answer Discipline @@ -42,6 +44,25 @@ For KB questions, prefer the bridge over raw database access. A good default is: 3. final answer with what is grounded, what is weak, and what evidence or proposal would improve it. +For no-context direct claims such as "Is X in Leo now?", "did the DB change?", +"did the decision matrix approve this?", or "is it still just proposals?", do +not stop at `search` or default `list-proposals`. Run the status-specific +proposal and governance readbacks needed to avoid overclaiming: + +```bash +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search-proposals "" --status all --limit 20 +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb decision-matrix-status +``` + +If `decision-matrix-status` says the matrix tables are absent or incomplete, +do not infer matrix approval from proposal rationale, reviewer notes, or +`kb_stage.kb_proposals.status`. Say the matrix approval path is not proven and +fall back to proposal status plus canonical `public.*` readback. + +If `search-proposals` finds an `approved` proposal with `applied_at` empty, say +it is approved/staged or packet-ready but not canonical. Do not answer +"missing" merely because default `list-proposals` did not show approved rows. + ## Telegram Rendering Make KB answers easy to scan in Telegram: diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 7826ae3..bebc4d6 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -9,6 +9,7 @@ import io import re import subprocess from pathlib import Path +from types import SimpleNamespace ROOT = Path(__file__).resolve().parents[1] BRIDGE_DIR = ROOT / "hermes-agent" / "leoclean-bin" @@ -42,7 +43,9 @@ def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None: "propose-core-change", "propose-edge", "list-proposals", + "search-proposals", "show-proposal", + "decision-matrix-status", ]: assert command in wrapper_text assert command in cloudsql_text @@ -85,6 +88,114 @@ def _load_module(path: Path): 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"}] + + 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 {"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", + "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 "Revised Helmer packet" in output + + +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_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}" diff --git a/tests/test_hermes_leoclean_skill_surfaces.py b/tests/test_hermes_leoclean_skill_surfaces.py index 28615a3..6c57ce2 100644 --- a/tests/test_hermes_leoclean_skill_surfaces.py +++ b/tests/test_hermes_leoclean_skill_surfaces.py @@ -11,6 +11,9 @@ def test_gcp_kb_skill_uses_cloudsql_bridge_not_vps_docker() -> None: assert "answer from the Cloud SQL KB after a bounded read" in text assert "Do not browse the public web just because the KB has evidence gaps" in text assert "/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb" in text + assert "search-proposals" in text + assert "decision-matrix-status" in text + assert "Do not answer" in text or "do not infer matrix approval" in text assert "docker exec" not in text assert "teleo-pg" not in text assert "on the VPS" not in text @@ -26,6 +29,10 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None: assert "prefer the bridge over raw database access" in text assert "Claim / Body / Concept Map Loop" in text assert "Status: applied | pending | missing | partially applied" in text + assert "search-proposals" in text + assert "decision-matrix-status" in text + assert "do not infer matrix approval" in text + assert "approved/staged or packet-ready but not canonical" in text assert "Do not call an approved proposal \"implemented\"" in text assert "needs reviewer/operator apply tooling rather than inviting ad hoc SQL from chat" in squashed assert "Never end a normal Telegram answer by offering to run direct `INSERT`, `UPDATE`" in text @@ -34,6 +41,8 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None: assert "/kb/claims/" in text assert "wrap claim IDs, proposal IDs" in text assert "claim page: https://leo.livingip.xyz/kb/claims/" in text + assert "search-proposals" in text + assert "decision-matrix-status" in text def test_gcp_kb_skill_keeps_claim_links_and_backtick_rendering() -> None: