Fix Leo direct-claim KB proposal readback

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

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

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
 }

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 <claim_id>
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb edges <claim_id>
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb list-proposals
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search-proposals "<terms>"
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show-proposal <proposal_id>
+/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 "<claim id or description>" --proposed "<replacement>" --source-ref "<telegram/chat/source ref>" --rationale "<why this should be reviewed>"
 ```
 
@@ -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 "<entity/framework/claim terms>" --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:

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 <claim_id>
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb edges <claim_id>
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb list-proposals
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search-proposals "<terms>"
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show-proposal <proposal_id>
+/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 "<entity/framework/claim terms>" --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:

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

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/<claim_id>" in text
     assert "wrap claim IDs, proposal IDs" in text
     assert "claim page: https://leo.livingip.xyz/kb/claims/<claim_id>" 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: