From 1703232f2463f63a446757d03729133e44b17627 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 9 Jul 2026 02:24:07 +0200 Subject: [PATCH] Add leoclean KB proposal review packets (#60) --- .../gcp/teleo-kb-bridge/SKILL.md | 145 ++++++++++ .../vps/live-leo-telegram/SKILL.md | 44 +++ .../vps/teleo-kb-bridge/SKILL.md | 124 ++++++++ scripts/apply_proposal.py | 4 +- scripts/kb_proposal_review_packet.py | 265 ++++++++++++++++++ tests/test_apply_proposal.py | 6 + tests/test_hermes_leoclean_skill_surfaces.py | 44 +++ tests/test_kb_proposal_review_packet.py | 75 +++++ 8 files changed, 705 insertions(+), 2 deletions(-) create mode 100644 hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md create mode 100644 hermes-agent/leoclean-skills/vps/live-leo-telegram/SKILL.md create mode 100644 hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md create mode 100755 scripts/kb_proposal_review_packet.py create mode 100644 tests/test_hermes_leoclean_skill_surfaces.py create mode 100644 tests/test_kb_proposal_review_packet.py diff --git a/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/gcp/teleo-kb-bridge/SKILL.md new file mode 100644 index 0000000..1217ba4 --- /dev/null +++ b/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 "" +``` + +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 "" +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb context "" +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show +/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 show-proposal +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb propose-core-change --proposal-type revise_claim --target-kind claim --target-ref "" --proposed "" --source-ref "" --rationale "" +``` + +## Answer Discipline + +For non-live questions, answer from the Cloud SQL KB after a bounded read. A +good default is: + +1. `teleo-kb context ""`; +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 "" \ + --current "" \ + --proposed "" \ + --evidence "" \ + --originator "" \ + --channel telegram \ + --source-ref "" \ + --rationale "" +``` + +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 +``` + +A good Telegram reply after staging should say: + +```text +I staged this as a KB proposal, not runtime memory. +Proposal: +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. diff --git a/hermes-agent/leoclean-skills/vps/live-leo-telegram/SKILL.md b/hermes-agent/leoclean-skills/vps/live-leo-telegram/SKILL.md new file mode 100644 index 0000000..5d7e9be --- /dev/null +++ b/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 -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. + diff --git a/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md b/hermes-agent/leoclean-skills/vps/teleo-kb-bridge/SKILL.md new file mode 100644 index 0000000..1d21307 --- /dev/null +++ b/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 "" +``` + +Use narrower bridge commands when needed: + +```bash +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb search "" +/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb show +/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 show-proposal +``` + +## Answer Discipline + +For KB questions, prefer the bridge over raw database access. A good default is: + +1. `teleo-kb context ""`; +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 ` or `search`; +2. fetch evidence and edges with `teleo-kb evidence ` and + `teleo-kb edges `; +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: +Staged proposals: +Rows/edges/evidence needed: +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. diff --git a/scripts/apply_proposal.py b/scripts/apply_proposal.py index 8757e81..70ad409 100644 --- a/scripts/apply_proposal.py +++ b/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: diff --git a/scripts/kb_proposal_review_packet.py b/scripts/kb_proposal_review_packet.py new file mode 100755 index 0000000..4faaab5 --- /dev/null +++ b/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()) diff --git a/tests/test_apply_proposal.py b/tests/test_apply_proposal.py index 5511e48..b68572d 100644 --- a/tests/test_apply_proposal.py +++ b/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 diff --git a/tests/test_hermes_leoclean_skill_surfaces.py b/tests/test_hermes_leoclean_skill_surfaces.py new file mode 100644 index 0000000..d99aa0c --- /dev/null +++ b/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 diff --git a/tests/test_kb_proposal_review_packet.py b/tests/test_kb_proposal_review_packet.py new file mode 100644 index 0000000..1bbd126 --- /dev/null +++ b/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"]