Expose exact Leo KB status readback

hermes-agent/leoclean-bin/cloudsql_memory_tool.py

@@ -1179,6 +1179,16 @@ def emit_markdown(value: dict[str, Any]) -> None:
             print(f"- Canonical DB: `{canonical['db_identity']}`")
             print(f"- Canonical tables: `{json.dumps(canonical.get('schema_tables'), sort_keys=True)}`")
             print(f"- Canonical high-signal rows: `{json.dumps(canonical.get('high_signal_rows'), sort_keys=True)}`")
+            counts = canonical.get("high_signal_rows") or {}
+            if all(
+                key in counts
+                for key in ("claims", "sources", "claim_edges", "claim_evidence", "kb_proposals")
+            ):
+                print(
+                    f"\nDB readback: claims: `{counts['claims']}`; sources: `{counts['sources']}`; "
+                    f"claim_edges: `{counts['claim_edges']}`; claim_evidence: `{counts['claim_evidence']}`; "
+                    f"kb_proposals: `{counts['kb_proposals']}`."
+                )
         print(f"- Extensions: {', '.join(value.get('extensions') or [])}")
         print(f"- Restored tables: {value['teleo_restore_tables']}")
         print(f"- Approx restored rows checked: {value['total_rows']}")

hermes-agent/leoclean-bin/kb_tool.py

@@ -88,6 +88,9 @@ def parse_args() -> argparse.Namespace:
             help="Output format. Accepted here for agent CLI ergonomics.",
         )
 
+    status = sub.add_parser("status", help="Read exact canonical and proposal table counts.")
+    add_output_flag(status)
+
     search = sub.add_parser("search", help="Search canonical claims and soul/context rows.")
     search.add_argument("query")
     search.add_argument("--limit", type=int, default=5)
@@ -778,6 +781,35 @@ def decision_matrix_status(args: argparse.Namespace) -> dict[str, Any]:
     return rows[0]
 
 
+def status(args: argparse.Namespace) -> dict[str, Any]:
+    sql = """
+    with 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_status',
+             'backend', current_database() || '|' || current_user,
+             'high_signal_rows', jsonb_build_object(
+               'claims', (select count(*) from public.claims),
+               'sources', (select count(*) from public.sources),
+               'claim_edges', (select count(*) from public.claim_edges),
+               'claim_evidence', (select count(*) from public.claim_evidence),
+               'kb_proposals', (select count(*) from kb_stage.kb_proposals)
+             ),
+             'proposal_status_counts', (select counts from status_counts)
+           )::text;
+    """
+    rows = psql_json(args, sql)
+    if not rows:
+        raise SystemExit("No KB 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 {}
@@ -898,6 +930,19 @@ def print_json(data: Any) -> None:
     print(json.dumps(data, indent=2, sort_keys=True))
 
 
+def print_status(data: dict[str, Any]) -> None:
+    counts = data["high_signal_rows"]
+    print("# Teleo VPS KB Status\n")
+    print(f"- backend: `{data['backend']}`")
+    print(f"- high-signal rows: `{json.dumps(counts, sort_keys=True)}`")
+    print(f"- proposal status counts: `{json.dumps(data.get('proposal_status_counts') or {}, sort_keys=True)}`\n")
+    print(
+        f"DB readback: claims: `{counts['claims']}`; sources: `{counts['sources']}`; "
+        f"claim_edges: `{counts['claim_edges']}`; claim_evidence: `{counts['claim_evidence']}`; "
+        f"kb_proposals: `{counts['kb_proposals']}`."
+    )
+
+
 def print_claim_bundle(data: dict[str, Any]) -> None:
     print(f"# Teleo KB Context\n\nQuery: {data['query']}\n")
     if data["context_rows"]:
@@ -1013,7 +1058,9 @@ def print_decision_matrix_status(data: dict[str, Any]) -> None:
 
 def main() -> int:
     args = parse_args()
-    if args.command == "search":
+    if args.command == "status":
+        data = status(args)
+    elif args.command == "search":
         data = {
             "query": args.query,
             "terms": query_terms(args.query),
@@ -1063,6 +1110,8 @@ def main() -> int:
 
     if args.format == "json":
         print_json(data)
+    elif args.command == "status":
+        print_status(data)
     elif args.command in {"context", "search"}:
         print_claim_bundle(data)
     elif args.command == "show":

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

@@ -112,7 +112,9 @@ proof language below. These are behavioral examples, not feature changes.
 - "Did the decision matrix approve this?": start with current/fresh schema
   readback. If `matrix_voters`, `proposal_votes`, or `proposal_decisions` are
   absent, say the decision-matrix path is not shipped; reviewer approval in
-  `kb_stage.kb_proposals` is not a matrix vote.
+  `kb_stage.kb_proposals` is not a matrix vote. Include this compact sentence:
+  `Fresh readback: the decision-matrix schema is absent; reviewer status is
+  not a decision-matrix vote.`
 - "Are proposals stuck because documents point at the wrong source rows?":
   do not answer as a single-cause `yes`. Say `not just pointer mismatch`: raw
   files, Telegram refs, document evaluations, proposal `source_ref`/logical
@@ -121,6 +123,7 @@ proof language below. These are behavioral examples, not feature changes.
 - "Can I demo Leo changes the KB?": lead with `staging yes, canonical KB
   change not safe to demo from chat`. Include `demo tier` language. A safe demo
   can show a real staging write to `kb_stage.kb_proposals` and read it back.
+  Say exactly: `Approved is not the same as applied.`
   Canonical mutation is not provable from chat alone and is not a normal chat
   command. State the exact current tier:
   the strict existing-ID `add_edge` path is live-proven on VPS; guarded
@@ -142,8 +145,9 @@ proof language below. These are behavioral examples, not feature changes.
   postflight/render-sync proof; without those rows, canonical identity is
   unchanged.`
 
-Before every direct-claim answer, run a fresh bridge read. Use `status` for
+Before every direct-claim answer, run a fresh bridge read. Use `status` for the
-canonical counts, `search-proposals` followed by `show-proposal` for a named
+complete numeric count template, `search-proposals` followed by `show-proposal`
+for a named
 proposal, and `decision-matrix-status` for matrix questions. Never invent or
 reuse a stale count.
 

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

@@ -24,6 +24,7 @@ question, run the local bridge:
 Use narrower bridge commands when needed:
 
 ```bash
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb status
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search "<terms>"
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show <claim_id>
 /home/teleo/.hermes/profiles/leoclean/bin/teleo-kb evidence <claim_id>
@@ -104,7 +105,9 @@ proof language below. These are behavioral examples, not feature changes.
 - "Did the decision matrix approve this?": start with current/fresh schema
   readback. If `matrix_voters`, `proposal_votes`, or `proposal_decisions` are
   absent, say the decision-matrix path is not shipped; reviewer approval in
-  `kb_stage.kb_proposals` is not a matrix vote.
+  `kb_stage.kb_proposals` is not a matrix vote. Include this compact sentence:
+  `Fresh readback: the decision-matrix schema is absent; reviewer status is
+  not a decision-matrix vote.`
 - "Are proposals stuck because documents point at the wrong source rows?":
   do not answer as a single-cause `yes`. Say `not just pointer mismatch`: raw
   files, Telegram refs, document evaluations, proposal `source_ref`/logical
@@ -113,6 +116,7 @@ proof language below. These are behavioral examples, not feature changes.
 - "Can I demo Leo changes the KB?": lead with `staging yes, canonical KB
   change not safe to demo from chat`. Include `demo tier` language. A safe demo
   can show a real staging write to `kb_stage.kb_proposals` and read it back.
+  Say exactly: `Approved is not the same as applied.`
   Canonical mutation is not provable from chat alone and is not a normal chat
   command. State the exact current tier:
   the strict existing-ID `add_edge` path is live-proven; guarded `approve_claim`
@@ -134,11 +138,12 @@ proof language below. These are behavioral examples, not feature changes.
   postflight/render-sync proof; without those rows, canonical identity is
   unchanged.`
 
-Before every direct-claim answer, run a fresh bridge read. Use
+Before every direct-claim answer, run a fresh bridge read. Use `status` for the
-`search-proposals` followed by `show-proposal` for a named proposal and
+complete numeric count template, `search-proposals` followed by `show-proposal`
-`decision-matrix-status` for matrix questions. If those bridge commands do not
+for a named proposal, and `decision-matrix-status` for matrix questions. If
-return the exact canonical counts needed for the question, use the documented
+those bridge commands do not return the exact canonical counts needed for the
-read-only Postgres fallback. Never invent or reuse a stale count.
+question, use the documented read-only Postgres fallback. Never invent or reuse
+a stale count.
 
 Every direct-claim answer must contain one compact line beginning `DB readback:`
 and copy either (a) a full UUID plus observed `status` and `applied_at`, or (b)

scripts/run_gcp_generated_db_direct_claim_suite.py

@@ -31,7 +31,7 @@ DEFAULT_HOST = "10.61.0.3"
 DEFAULT_PROJECT = "teleo-501523"
 DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
 DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql"
-REVIEWED_CLOUDSQL_TOOL_SHA256 = "f54040141551a22a9a39ab020abfd897963bc2a53205f00cd49d7212cbaffff1"
+REVIEWED_CLOUDSQL_TOOL_SHA256 = "f4396298405b1cbc92590e148b6272c66bb0e8461178edc3ffcfd02bfed69ab4"
 REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a"
 
 

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -32,6 +32,7 @@ def test_leoclean_bridge_files_are_present_and_parseable() -> None:
 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",
@@ -50,6 +51,8 @@ def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None:
         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()
 
@@ -168,6 +171,97 @@ def test_vps_bridge_decision_matrix_status_checks_schema_tables(monkeypatch) ->
         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] = []

tests/test_hermes_leoclean_skill_surfaces.py

@@ -14,6 +14,7 @@ def test_gcp_kb_skill_uses_cloudsql_bridge_not_vps_docker() -> None:
     assert "/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb" in text
     assert "search-proposals" in text
     assert "decision-matrix-status" in text
+    assert "Use `status` for the complete numeric count template" in squashed
     assert "list-proposals --status all" in text
     assert "Do not answer" in text or "do not infer matrix approval" in text
     assert "Next Cory-style follow-up" in text
@@ -65,6 +66,7 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None:
     assert "Status: applied | pending | missing | partially applied" in text
     assert "search-proposals" in text
     assert "decision-matrix-status" in text
+    assert "Use `status` for the complete numeric count template" in squashed
     assert "list-proposals --status all" in text
     assert "do not infer matrix approval" in text
     assert "approved/staged or packet-ready but not canonical" in text
@@ -81,6 +83,9 @@ def test_vps_kb_skill_keeps_vps_scope_explicit() -> None:
     assert "not provable from chat" in text
     assert "production permission migration and apply worker remain disabled" in squashed
     assert "false global statement that no apply tooling exists" in squashed
+    assert "Approved is not the same as applied." in text
+    assert "Fresh readback:" in text
+    assert "not a decision-matrix vote" in text
     assert "DB readback:" in text
     assert "full UUID plus observed `status` and `applied_at`" in squashed
     assert "Short eight-character IDs" in text