Add leoclean KB proposal review packets (#60)

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

@@ -0,0 +1,145 @@
+---
+name: teleo-kb-bridge
+description: Use the GCP Cloud SQL KB bridge before answering questions about claims, evidence, edges, schema-backed soul/context, KB approval, or KB edit workflow.
+version: 1.0.0
+author: m3taversal
+license: MIT
+metadata:
+  hermes:
+    tags: [teleo, kb, postgres, cloudsql, claims, evidence, governance]
+    related_skills: [leo-synthesis-methods]
+---
+
+# Teleo KB Bridge
+
+The canonical Teleo knowledge base is Postgres, not runtime memory.
+
+This is the GCP parallel leoclean surface. The local bridge command is still the
+only path Leo should use directly:
+
+```bash
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb context "<question>"
+```
+
+On GCP, `teleo-kb` routes to the Cloud SQL wrapper. Do not use raw
+container-Postgres commands from the VPS playbook; those are not valid on the
+GCP VM.
+
+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 context "<question>"
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show <claim_id>
+/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 show-proposal <proposal_id>
+/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>"
+```
+
+## Answer Discipline
+
+For non-live questions, answer from the Cloud SQL KB after a bounded read. A
+good default is:
+
+1. `teleo-kb context "<question>"`;
+2. at most three `show` / `evidence` / `edges` follow-ups for the most relevant
+   IDs;
+3. final answer with what is grounded, what is weak, and what evidence would
+   improve it.
+
+Do not browse the public web just because the KB has evidence gaps. If the user
+asks for current Twitter/X, current market activity, today's news, or another
+fresh external read, use the configured current-source path if it exists. If it
+does not exist, say the fresh external source is unavailable from this runtime
+and answer from cached/canonical context with an explicit freshness caveat.
+
+## When To Use This
+
+Use this bridge before answering when the user asks:
+
+- what the KB says;
+- whether a claim is approved, grounded, contradicted, or superseded;
+- what evidence supports a claim;
+- which claims support/challenge/relate to another claim;
+- how to approve/edit/retire/supersede a claim;
+- what the schema-backed soul/context says about roles, peers, blindspots, strategy, rules, or reasoning tools.
+
+## Memory vs KB Rule
+
+Do not treat runtime memory as canonical truth.
+
+```text
+agent memory = local/runtime continuity
+Cloud SQL / Postgres KB = canonical collective knowledge
+```
+
+If a correction changes collective truth, it belongs in the KB graph, not only
+runtime memory.
+
+## GCP DB Objects
+
+Relevant DB objects are reached through `teleo-kb` on GCP:
+
+- `kb_stage.kb_proposals` - durable proposal ledger;
+- `kb_stage.pending_kb_proposals` - proposals with `status = 'pending_review'`;
+- `kb_stage.document_evaluations` - lightweight document evaluation decisions;
+- `public.claims`, `public.sources`, `public.claim_evidence`, `public.claim_edges` - canonical tables.
+
+## Write Policy
+
+Canonical KB writes are locked. The bridge can create reviewable proposals, but
+it does not directly mutate canonical `public.*` rows from normal chat.
+
+When a user says they want to change a core Teleo claim, strategy, identity,
+role, telos, framework, belief, or reasoning tool, stage it in Cloud SQL:
+
+```bash
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb propose-core-change \
+  --proposal-type revise_claim \
+  --target-kind claim \
+  --target-ref "<claim id if known, otherwise a precise target description>" \
+  --current "<current claim if known>" \
+  --proposed "<proposed replacement/correction>" \
+  --evidence "<source, message, document, or reason>" \
+  --originator "<human handle>" \
+  --channel telegram \
+  --source-ref "<chat/message/source reference>" \
+  --rationale "<why this should be reviewed>"
+```
+
+Use `--proposal-type revise_strategy` for strategy-level changes. The command
+writes a `kb_stage.kb_proposals` row with `pending_review` status. It does not
+mutate canonical `public.*` rows and does not write runtime memory.
+
+After staging, read it back:
+
+```bash
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show-proposal <proposal_id>
+```
+
+A good Telegram reply after staging should say:
+
+```text
+I staged this as a KB proposal, not runtime memory.
+Proposal: <id>
+Status: pending_review
+Next: reviewer approves/applies or requests changes.
+```
+
+## Applying Approved Changes
+
+The default Telegram path does not apply approved proposals into canonical
+truth. Application requires a reviewer/operator workflow. On GCP, do not suggest
+raw container shell commands. Prefer one of:
+
+1. use a dedicated `teleo-kb apply-*` command if the bridge exposes one;
+2. use a reviewed migration/PR workflow;
+3. use an explicit operator-approved Cloud SQL apply script with retained
+   before/after readback.
+
+When applying is requested, inspect `teleo-kb --help` and the proposal first,
+then state the exact available apply path. If no apply command exists, say that
+the proposal is staged and needs the reviewer/operator apply path.

hermes-agent/leoclean-skills/vps/live-leo-telegram/SKILL.md

@@ -0,0 +1,44 @@
+---
+name: live-leo-telegram
+description: Live Telegram operating discipline for VPS production Leo liveness, runtime status, and user-visible replies.
+version: 1.0.0
+author: m3taversal
+license: MIT
+metadata:
+  hermes:
+    tags: [teleo, leo, telegram, live-runtime, operating-discipline]
+    related_skills: [teleo-kb-bridge]
+---
+
+# Live Leo Telegram Discipline
+
+You are live Leo in Telegram, not a generic help bot.
+
+## Live Gateway Liveness
+
+When answering a Telegram liveness, runtime, or "are you alive" question, use
+the live transport and systemd process as the source of truth.
+
+If you are currently replying in Telegram, the Telegram gateway transport is
+alive for this turn.
+
+Do not use `hermes -p leoclean gateway status` as authoritative when leoclean
+is managed by systemd. That CLI status can report `not running` even while the
+production systemd service is active and handling Telegram messages.
+
+For production gateway status, prefer these checks:
+
+```bash
+systemctl is-active leoclean-gateway.service
+systemctl show leoclean-gateway.service -p MainPID -p ActiveState -p SubState --value
+ps -p <MainPID> -o pid,cmd
+```
+
+If systemd says `active` and `ps` shows
+`hermes -p leoclean gateway run`, say the production gateway is active. Do not
+tell Telegram users the gateway is down because of a contradictory Hermes CLI
+status readback.
+
+For short liveness questions, answer compactly and avoid dumping commands or
+raw logs unless the user explicitly asks for evidence.
+

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

@@ -0,0 +1,124 @@
+---
+name: teleo-kb-bridge
+description: Use the VPS Postgres KB bridge before answering questions about claims, evidence, edges, schema-backed soul/context, KB approval, or KB edit workflow.
+version: 1.0.0
+author: m3taversal
+license: MIT
+metadata:
+  hermes:
+    tags: [teleo, kb, postgres, claims, evidence, governance]
+    related_skills: [leo-synthesis-methods]
+---
+
+# Teleo KB Bridge
+
+The canonical Teleo knowledge base is Postgres, not runtime memory.
+
+This is the VPS production leoclean surface. Before answering a KB-specific
+question, run the local bridge:
+
+```bash
+/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb context "<question>"
+```
+
+Use narrower bridge commands when needed:
+
+```bash
+/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>
+/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 show-proposal <proposal_id>
+```
+
+## Answer Discipline
+
+For KB questions, prefer the bridge over raw database access. A good default is:
+
+1. `teleo-kb context "<question>"`;
+2. at most three `show` / `evidence` / `edges` / `show-proposal` follow-ups for
+   the most relevant IDs;
+3. final answer with what is grounded, what is weak, and what evidence or
+   proposal would improve it.
+
+Use raw `docker exec ... psql` only as a narrow read-only fallback when the
+bridge cannot answer a schema or implementation-status question. If you use
+that fallback, say it was a read-only inspection. Do not present raw SQL as the
+normal user workflow.
+
+## Claim / Body / Concept Map Loop
+
+When a user challenges a claim as too broad, too light, unfalsifiable, or
+poorly linked, do this loop:
+
+1. fetch the headline claim with `teleo-kb show <claim_id>` or `search`;
+2. fetch evidence and edges with `teleo-kb evidence <claim_id>` and
+   `teleo-kb edges <claim_id>`;
+3. separate what the KB actually says from your synthesis;
+4. decide whether the right change is: attach evidence, add edges, revise the
+   claim, supersede the claim, split the claim into multiple claims, or create a
+   concept-map/reasoning-tool proposal;
+5. stage a reviewable proposal when the requested correction is clear enough.
+
+For "was this implemented?" or "did you apply that?" questions, answer in this
+shape:
+
+```text
+Status: applied | pending | missing | partially applied
+Canonical rows: <what exists in public.*>
+Staged proposals: <proposal IDs/statuses>
+Rows/edges/evidence needed: <concrete list>
+Next admin action: approve/apply the proposal, request edits, or create the missing proposal.
+```
+
+Do not call an approved proposal "implemented" until canonical `public.*` rows
+and edges show the applied state.
+
+## Memory vs KB Rule
+
+Do not treat runtime memory as canonical truth.
+
+```text
+agent memory = local/runtime continuity
+Postgres KB = canonical collective knowledge
+```
+
+If a correction changes collective truth, it belongs in the KB graph, not only
+runtime memory.
+
+## VPS DB Objects
+
+Relevant DB objects live in the VPS Postgres container and should normally be
+reached through `teleo-kb`:
+
+- `kb_stage.kb_proposals` - durable proposal ledger;
+- `kb_stage.pending_kb_proposals` - proposals with `status = 'pending_review'`;
+- `kb_stage.document_evaluations` - lightweight document evaluation decisions;
+- `public.claims`, `public.sources`, `public.claim_evidence`, `public.claim_edges` - canonical tables.
+
+## Write Policy
+
+Canonical KB writes are locked. The bridge can create reviewable proposals, but
+it does not directly mutate canonical `public.*` rows from normal chat.
+
+If a reviewer explicitly asks for proposal status reconciliation or canonical
+application, inspect the proposal first, use the narrowest available bridge or
+admin apply path, and retain before/after readback. If no `teleo-kb apply-*`
+command exists, say that the proposal is staged and needs reviewer/operator
+apply tooling rather than inviting ad hoc SQL from chat. Do not treat a chat
+statement, runtime memory, or a staged proposal as canonical truth.
+
+Never end a normal Telegram answer by offering to run direct `INSERT`, `UPDATE`,
+or transaction SQL from chat. Even if the user is authorized, the product flow is
+review-first:
+
+```text
+Next admin-panel action: show the staged proposal, dependency groups, and
+before/after rows; let a reviewer approve, reject, edit, or run a dedicated
+apply tool with retained readback.
+```
+
+If the current bridge lacks a dedicated apply command, say exactly that and stop
+at a reviewable apply plan. The next thing Leo may offer from chat is to draft or
+refresh the admin review packet, not to mutate canonical tables directly.

scripts/apply_proposal.py

@@ -340,9 +340,9 @@ def load_password(secrets_file: str) -> str:
         if not line or line.startswith("#") or "=" not in line:
             continue
         key, _, val = line.partition("=")
-        if key.strip() in ("KB_APPLY_PASSWORD", "PGPASSWORD"):
+        if key.strip() in ("KB_APPLY_PASSWORD", "KB_APPLY_DB_PASSWORD", "PGPASSWORD"):
             return val.strip().strip('"').strip("'")
-    raise SystemExit(f"no KB_APPLY_PASSWORD/PGPASSWORD entry in {secrets_file}")
+    raise SystemExit(f"no KB_APPLY_PASSWORD/KB_APPLY_DB_PASSWORD/PGPASSWORD entry in {secrets_file}")
 
 
 def run_psql(args: argparse.Namespace, sql: str, password: str) -> str:

scripts/kb_proposal_review_packet.py

@@ -0,0 +1,265 @@
+#!/usr/bin/env python3
+"""Render KB proposals as admin-review packets.
+
+This is the read-only surface between Leo's Telegram reasoning and the apply
+worker. It makes the distinction explicit:
+
+* ``approved`` means a reviewer accepted the proposal intent.
+* ``applyable`` means the proposal carries a strict ``apply_payload`` contract
+  that the narrow ``kb_apply`` worker can execute.
+* ``applied`` means canonical ``public.*`` rows have changed and the proposal
+  ledger was stamped.
+
+The script performs no writes. It can read live proposals through the same
+restricted secret path used by the apply worker, or classify proposal JSON from
+a file for tests and UI development.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(HERE))
+
+import apply_proposal as ap  # noqa: E402
+import apply_worker as aw  # noqa: E402
+
+
+def _payload_list(payload: dict[str, Any], key: str) -> list[Any]:
+    value = payload.get(key)
+    return value if isinstance(value, list) else []
+
+
+def _payload_dict(payload: dict[str, Any], key: str) -> dict[str, Any]:
+    value = payload.get(key)
+    return value if isinstance(value, dict) else {}
+
+
+def classify_proposal(proposal: dict[str, Any]) -> dict[str, Any]:
+    payload = proposal.get("payload") or {}
+    if not isinstance(payload, dict):
+        payload = {}
+
+    status = proposal.get("status")
+    proposal_type = proposal.get("proposal_type")
+    has_apply_payload = "apply_payload" in payload
+    worker_supported_type = proposal_type in aw.WORKER_TYPES
+    worker_applyable = (
+        status == "approved" and worker_supported_type and has_apply_payload
+    )
+
+    if status == "applied":
+        review_state = "applied"
+    elif status == "pending_review":
+        review_state = "needs_human_review"
+    elif status == "approved" and worker_applyable:
+        review_state = "approved_applyable"
+    elif status == "approved" and not has_apply_payload:
+        review_state = "approved_needs_apply_payload"
+    elif not worker_supported_type:
+        review_state = "unsupported_by_apply_worker"
+    else:
+        review_state = "not_ready"
+
+    claim_candidates = _payload_list(payload, "claim_candidates")
+    source_candidates = _payload_list(payload, "source_candidates")
+    supersession_edges = _payload_list(payload, "proposed_supersession_edges")
+    concept_map = _payload_dict(payload, "proposed_concept_map_dependency")
+    governance_standard = _payload_dict(payload, "proposed_governance_standard")
+    approval_request = _payload_dict(payload, "approval_request")
+    review_request = _payload_dict(payload, "review_request")
+
+    missing_contract = []
+    if not has_apply_payload:
+        missing_contract.append("strict payload.apply_payload")
+    if claim_candidates and not has_apply_payload:
+        missing_contract.append("claim-row normalization")
+    if source_candidates and not has_apply_payload:
+        missing_contract.append("source/evidence normalization")
+    if supersession_edges and not has_apply_payload:
+        missing_contract.append("edge/supersession normalization")
+    if concept_map and not has_apply_payload:
+        missing_contract.append("concept-map storage decision")
+
+    if review_state == "approved_needs_apply_payload":
+        next_admin_action = (
+            "Show the approved intent, candidate rows, sources, dependencies, "
+            "and schema gaps; then normalize into one or more strict apply_payload "
+            "proposals before enabling the apply worker."
+        )
+    elif review_state == "approved_applyable":
+        next_admin_action = (
+            "Show dry-run SQL and before/after row preview; require operator apply."
+        )
+    elif review_state == "needs_human_review":
+        next_admin_action = (
+            "Reviewer should approve, reject, or request edits; no apply is possible yet."
+        )
+    elif review_state == "applied":
+        next_admin_action = "Show canonical readback and applied_at/applied_by stamp."
+    else:
+        next_admin_action = "Split or rewrite into a supported proposal contract."
+
+    return {
+        "proposal_id": proposal.get("id"),
+        "status": status,
+        "proposal_type": proposal_type,
+        "review_state": review_state,
+        "worker_supported_type": worker_supported_type,
+        "worker_applyable": worker_applyable,
+        "has_apply_payload": has_apply_payload,
+        "identity_and_authorization": {
+            "proposed_by_handle": proposal.get("proposed_by_handle"),
+            "reviewed_by_handle": proposal.get("reviewed_by_handle"),
+            "reviewed_at": proposal.get("reviewed_at"),
+            "review_note": proposal.get("review_note"),
+            "applied_by_handle": proposal.get("applied_by_handle"),
+            "applied_at": proposal.get("applied_at"),
+            "source_ref": proposal.get("source_ref") or payload.get("source_ref"),
+            "channel": proposal.get("channel") or payload.get("channel"),
+        },
+        "payload_summary": {
+            "proposal_kind": payload.get("proposal_kind"),
+            "old_claim_id": _payload_dict(payload, "old_claim").get("id"),
+            "claim_candidate_count": len(claim_candidates),
+            "source_candidate_count": len(source_candidates),
+            "supersession_edge_count": len(supersession_edges),
+            "has_concept_map_dependency": bool(concept_map),
+            "has_governance_standard": bool(governance_standard),
+            "schema_gap_note": payload.get("schema_gap_note"),
+            "approve_now": approval_request.get("approve_now", []),
+            "do_not_approve_yet": approval_request.get("do_not_approve_yet", []),
+            "requested_decision": review_request.get("requested_decision"),
+            "not_requested": review_request.get("not_requested"),
+        },
+        "applyability": {
+            "missing_contract": missing_contract,
+            "apply_worker_types": list(aw.WORKER_TYPES),
+            "strict_apply_payload_required": True,
+        },
+        "admin_panel_sections": [
+            "identity_and_authorization",
+            "payload_summary",
+            "candidate_rows_and_sources",
+            "dependencies_and_schema_gaps",
+            "dry_run_sql_only_after_apply_payload_exists",
+        ],
+        "next_admin_action": next_admin_action,
+    }
+
+
+def load_from_input(path: str) -> list[dict[str, Any]]:
+    data = json.loads(Path(path).read_text(encoding="utf-8"))
+    if isinstance(data, list):
+        return data
+    if isinstance(data, dict) and isinstance(data.get("proposals"), list):
+        return data["proposals"]
+    if isinstance(data, dict):
+        return [data]
+    raise SystemExit(f"unsupported proposal JSON shape in {path}")
+
+
+def _psql_args(args: argparse.Namespace) -> argparse.Namespace:
+    return argparse.Namespace(
+        container=args.container, db=args.db, host=args.host, role=args.role
+    )
+
+
+def load_from_db(args: argparse.Namespace) -> list[dict[str, Any]]:
+    password = ap.load_password(args.secrets_file)
+    where = "true"
+    if args.proposal_id:
+        where = f"id = {ap.sql_literal(args.proposal_id)}::uuid"
+    elif args.status:
+        where = f"status = {ap.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', rationale,
+  'payload', payload,
+  'reviewed_by_handle', reviewed_by_handle,
+  'reviewed_at', reviewed_at,
+  'review_note', review_note,
+  'applied_by_handle', applied_by_handle,
+  'applied_at', applied_at,
+  'created_at', created_at,
+  'updated_at', updated_at
+)::text
+from kb_stage.kb_proposals
+where {where}
+order by created_at desc
+limit {int(args.limit)};
+"""
+    out = ap.run_psql(_psql_args(args), sql, password)
+    rows = []
+    for line in out.splitlines():
+        line = line.strip()
+        if line:
+            rows.append(json.loads(line))
+    return rows
+
+
+def render_markdown(packets: list[dict[str, Any]]) -> str:
+    chunks = []
+    for p in packets:
+        auth = p["identity_and_authorization"]
+        summary = p["payload_summary"]
+        applyability = p["applyability"]
+        chunks.append(
+            "\n".join(
+                [
+                    f"## {p['proposal_id']}",
+                    f"- Status: {p['status']}",
+                    f"- Review state: {p['review_state']}",
+                    f"- Worker applyable: {p['worker_applyable']}",
+                    f"- Reviewed by: {auth.get('reviewed_by_handle') or 'unreviewed'}",
+                    f"- Proposal kind: {summary.get('proposal_kind') or 'unknown'}",
+                    f"- Claim candidates: {summary['claim_candidate_count']}",
+                    f"- Source candidates: {summary['source_candidate_count']}",
+                    f"- Missing contract: {', '.join(applyability['missing_contract']) or 'none'}",
+                    f"- Next admin action: {p['next_admin_action']}",
+                ]
+            )
+        )
+    return "\n\n".join(chunks) + ("\n" if chunks else "")
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input-json", help="classify proposal JSON from a file")
+    parser.add_argument("--proposal-id", help="load one live proposal from Postgres")
+    parser.add_argument("--status", default="approved", help="load live proposals by status")
+    parser.add_argument("--limit", type=int, default=20)
+    parser.add_argument("--format", choices=["json", "markdown"], default="json")
+    parser.add_argument("--secrets-file", default=ap.DEFAULT_SECRETS_FILE)
+    parser.add_argument("--container", default=ap.DEFAULT_CONTAINER)
+    parser.add_argument("--db", default=ap.DEFAULT_DB)
+    parser.add_argument("--host", default=ap.DEFAULT_HOST)
+    parser.add_argument("--role", default=ap.DEFAULT_ROLE)
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str] | None = None) -> int:
+    args = parse_args(sys.argv[1:] if argv is None else argv)
+    proposals = load_from_input(args.input_json) if args.input_json else load_from_db(args)
+    packets = [classify_proposal(proposal) for proposal in proposals]
+    if args.format == "markdown":
+        print(render_markdown(packets), end="")
+    else:
+        print(json.dumps({"packets": packets}, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_apply_proposal.py

@@ -182,6 +182,12 @@ def test_build_apply_sql_defaults_applied_by_to_service_agent():
     assert f"applied_by_handle = '{ap.SERVICE_AGENT_HANDLE}'" in sql
 
 
+def test_load_password_accepts_live_db_password_key(tmp_path):
+    secrets_file = tmp_path / "kb-apply.env"
+    secrets_file.write_text("KB_APPLY_DB_PASSWORD=live-secret\n", encoding="utf-8")
+    assert ap.load_password(str(secrets_file)) == "live-secret"
+
+
 if __name__ == "__main__":
     import traceback
 

tests/test_hermes_leoclean_skill_surfaces.py

@@ -0,0 +1,44 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+SKILL_ROOT = ROOT / "hermes-agent" / "leoclean-skills"
+
+
+def test_gcp_kb_skill_uses_cloudsql_bridge_not_vps_docker() -> None:
+    text = (SKILL_ROOT / "gcp" / "teleo-kb-bridge" / "SKILL.md").read_text()
+
+    assert "Cloud SQL" in text
+    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 "docker exec" not in text
+    assert "teleo-pg" not in text
+    assert "on the VPS" not in text
+
+
+def test_vps_kb_skill_keeps_vps_scope_explicit() -> None:
+    text = (SKILL_ROOT / "vps" / "teleo-kb-bridge" / "SKILL.md").read_text()
+    squashed = " ".join(text.split())
+
+    assert "VPS production leoclean" in text
+    assert "Cloud SQL" not in text
+    assert "/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb" in text
+    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 "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
+    assert "Next admin-panel action" in text
+    assert "draft or refresh the admin review packet" in squashed
+
+
+def test_vps_live_telegram_skill_uses_systemd_for_gateway_liveness() -> None:
+    text = (SKILL_ROOT / "vps" / "live-leo-telegram" / "SKILL.md").read_text()
+
+    assert "systemctl is-active leoclean-gateway.service" in text
+    assert "systemctl show leoclean-gateway.service" in text
+    assert "hermes -p leoclean gateway run" in text
+    assert "Do not use `hermes -p leoclean gateway status` as authoritative" in text
+    assert "If you are currently replying in Telegram" in text

tests/test_kb_proposal_review_packet.py

@@ -0,0 +1,75 @@
+"""Tests for scripts/kb_proposal_review_packet.py."""
+
+from pathlib import Path
+import sys
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import kb_proposal_review_packet as rp  # noqa: E402
+
+
+def test_approved_freeform_packet_is_not_worker_applyable():
+    proposal = {
+        "id": "14fa5ecc-ac7a-41c1-807d-a2e85b936617",
+        "proposal_type": "attach_evidence",
+        "status": "approved",
+        "reviewed_by_handle": "m3ta",
+        "payload": {
+            "proposal_kind": "supersede_compressed_claim_with_split_claim_cluster",
+            "old_claim": {"id": "d0000000-0000-0000-0000-000000000005"},
+            "claim_candidates": [{"headline": "A"}, {"headline": "B"}],
+            "source_candidates": [{"source_key": "hayek"}],
+            "proposed_supersession_edges": [{"edge_type": "supersedes_component_of"}],
+            "proposed_concept_map_dependency": {"headline": "Distributed market intelligence"},
+            "schema_gap_note": "bridge cannot propose claims directly",
+        },
+    }
+
+    packet = rp.classify_proposal(proposal)
+
+    assert packet["review_state"] == "approved_needs_apply_payload"
+    assert packet["worker_applyable"] is False
+    assert packet["has_apply_payload"] is False
+    assert "strict payload.apply_payload" in packet["applyability"]["missing_contract"]
+    assert "claim-row normalization" in packet["applyability"]["missing_contract"]
+    assert packet["payload_summary"]["claim_candidate_count"] == 2
+    assert packet["payload_summary"]["has_concept_map_dependency"] is True
+    assert "normalize into one or more strict apply_payload" in packet["next_admin_action"]
+
+
+def test_approved_strict_packet_is_worker_applyable():
+    proposal = {
+        "id": "edge-proposal",
+        "proposal_type": "add_edge",
+        "status": "approved",
+        "payload": {
+            "apply_payload": {
+                "from_claim": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
+                "to_claim": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
+                "edge_type": "supports",
+            }
+        },
+    }
+
+    packet = rp.classify_proposal(proposal)
+
+    assert packet["review_state"] == "approved_applyable"
+    assert packet["worker_applyable"] is True
+    assert packet["applyability"]["missing_contract"] == []
+    assert "Show dry-run SQL" in packet["next_admin_action"]
+
+
+def test_pending_packet_needs_human_review():
+    packet = rp.classify_proposal(
+        {
+            "id": "pending",
+            "proposal_type": "add_edge",
+            "status": "pending_review",
+            "payload": {"apply_payload": {"from_claim": "a", "to_claim": "b", "edge_type": "supports"}},
+        }
+    )
+
+    assert packet["review_state"] == "needs_human_review"
+    assert packet["worker_applyable"] is False
+    assert "approve, reject, or request edits" in packet["next_admin_action"]