from pathlib import Path ROOT = Path(__file__).resolve().parents[1] SKILL_ROOT = ROOT / "hermes-agent" / "leoclean-skills" def _kb_skill_bundle(surface: str) -> str: root = SKILL_ROOT / surface / "teleo-kb-bridge" parts = [(root / "SKILL.md").read_text()] reference = root / "REFERENCE.md" if reference.exists(): parts.append(reference.read_text()) return "\n".join(parts) def test_gcp_kb_skill_uses_cloudsql_bridge_not_vps_docker() -> None: text = (SKILL_ROOT / "gcp" / "teleo-kb-bridge" / "SKILL.md").read_text() squashed = " ".join(text.split()) 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 "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 proof-changing follow-up" in text assert "Always include the final line label" in text assert "Operator Direct-Claim Answer Contract" in text assert "Address `@m3taversal` only as `m3taversal`, exactly" in text assert "legacy reviewer value `m3ta`" in text assert "It is not a form of address" in squashed assert "no `body`, generic metadata, or forecast-resolution" in squashed assert "it has no author/channel/date fields" in text assert "Do not present a proposed v3 field" in text assert "Cory" not in text assert "not just pointer mismatch" in text assert "staging write to `kb_stage.kb_proposals`" in squashed assert "demo tier" in text assert "Approved is not the same as applied" in squashed assert "applied_at: NULL" in text assert "not applied" in text assert "canonical KB change not safe to demo from chat" in squashed assert "not provable from chat" in text assert "equivalent GCP execution is not yet proven" in squashed assert "false global statement that no apply tooling exists" in squashed assert "DB readback:" in text assert "full UUID plus observed `status` and `applied_at`" in squashed assert "Short eight-character IDs" in text assert "PROPOSAL_UUID_36_CHARS" in text assert "OBSERVED_STATUS" in text assert "OBSERVED_TIMESTAMP_OR_NONE" in text assert "claims: `N`; sources: `N`; claim_edges: `N`; claim_evidence: `N`; kb_proposals: `N`" in text assert "Never shorten a UUID" in text assert "Do not paraphrase a count" in text assert "`N/A` and `see public.*` are invalid" in squashed assert "canonical identity requires DB rows plus" in squashed assert "not canonical Postgres" in text assert "not the source of truth" in text assert "not a canonical commit" in text assert "not collective truth" in text assert "Row-level proof would require current readback" in text assert "new or updated row IDs" in text assert "I cannot claim canonical DB changed" in text assert "explicit operator/admin authorization" 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 = _kb_skill_bundle("vps") 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 "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 assert "Next proof-changing follow-up" in text assert "Always include the final line label" in text assert "Operator Direct-Claim Answer Contract" in text assert "Address `@m3taversal` only as `m3taversal`, exactly" in text assert "legacy reviewer value `m3ta`" in text assert "It is not a form of address" in squashed assert "no `body`, generic metadata, or forecast-resolution" in squashed assert "it has no author/channel/date fields" in text assert "Do not present a proposed v3 field" in text assert "Cory" not in text assert "not just pointer mismatch" in text assert "staging write to `kb_stage.kb_proposals`" in squashed assert "demo tier" in text assert "Approved is not the same as applied" in squashed assert "applied_at: NULL" in text assert "not applied" in text assert "canonical KB change not safe to demo from chat" in squashed 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 "run both `decision-matrix-status` and `status`" in squashed assert "DB readback:" in text assert "full UUID plus observed `status` and `applied_at`" in squashed assert "Short eight-character IDs" in text assert "PROPOSAL_UUID_36_CHARS" in text assert "OBSERVED_STATUS" in text assert "OBSERVED_TIMESTAMP_OR_NONE" in text assert "claims: `N`; sources: `N`; claim_edges: `N`; claim_evidence: `N`; kb_proposals: `N`" in text assert "Never shorten a UUID" in text assert "Do not paraphrase a count" in text assert "`N/A` and `see public.*` are invalid" in squashed assert "canonical identity requires DB rows plus" in squashed assert "not canonical Postgres" in text assert "not the source of truth" in text assert "not a canonical commit" in text assert "not collective truth" in text assert "Row-level proof would require current readback" in text assert "new or updated row IDs" in text assert "I cannot claim canonical DB changed" in text assert "explicit operator/admin authorization" in text assert 'Do not call an approved proposal "implemented"' in text assert "repository contains a live-proven strict existing-ID `add_edge` path" 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 assert "/kb/claims/" in text assert "wrap claim IDs, proposal IDs" in text assert "claim page: https://leo.livingip.xyz/kb/claims/" in text assert "search-proposals" in text assert "decision-matrix-status" in text def test_gcp_kb_skill_keeps_claim_links_and_backtick_rendering() -> None: text = (SKILL_ROOT / "gcp" / "teleo-kb-bridge" / "SKILL.md").read_text() assert "/kb/claims/" in text assert "wrap claim IDs, proposal IDs" in text assert "claim page: https://leo.livingip.xyz/kb/claims/" in text def test_leoclean_kb_skills_anchor_external_doctrine_in_target_project_language() -> None: for surface in ("vps", "gcp"): text = _kb_skill_bundle(surface) assert "External Doctrine Contributions" in text assert "Use the target project's native language first" in text assert "no single voice can own understanding" in text assert "purpose precedes" in text assert "Prefer issue-before-PR" in text assert "Consent is action-specific" in text def test_leoclean_kb_skills_distinguish_canonical_evidence_from_provenance_quality() -> None: for surface in ("vps", "gcp"): text = _kb_skill_bundle(surface) squashed = " ".join(text.split()) assert "is canonical evidence even when that source row has no `url` or `storage_path`" in squashed assert "weak or citation-only provenance" in squashed assert "not traceable to the raw artifact" in squashed assert "does not by itself prove canonical evidence from that attachment" in squashed assert "`public.sources` row representing the attachment" in squashed assert "`public.claim_evidence` link from the claim to that source row" in squashed def test_leoclean_kb_skills_use_existing_behavioral_rule_storage_and_separate_apply() -> None: for surface in ("vps", "gcp"): text = _kb_skill_bundle(surface) squashed = " ".join(text.split()) assert "existing `public.behavioral_rules` table" in squashed for field in ("`agent_id`", "`category`", "`rank`", "`rule`", "`rationale`"): assert field in text assert "does not insert `behavioral_rules` or `governance_gates`" in squashed assert "separate reviewed apply capability" in squashed assert "not a generic schema-table gap" in squashed def test_leoclean_kb_skills_separate_runtime_session_and_delivery_proof_tiers() -> None: for surface in ("vps", "gcp"): text = _kb_skill_bundle(surface) squashed = " ".join(text.split()) assert "keep the whole reply under 220 words" in squashed assert "Postgres is canonical knowledge, but it is not the only input" in squashed assert "Hermes `state.db` and session JSONL provide durable continuity" in squashed assert "not Telegram-visible delivery proof" in squashed assert "Staging is not canonical apply" in squashed assert "Never manufacture a source row from a temporary memory label" in squashed assert "claims, sources, and evidence links are shared knowledge objects" in squashed assert "no shipped forecast-resolution fields" in squashed assert "Do not overwrite historical confidence" 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 def test_vps_skills_enforce_compact_default_answers_without_losing_detail_on_request() -> None: kb_text = (SKILL_ROOT / "vps" / "teleo-kb-bridge" / "SKILL.md").read_text() telegram_text = (SKILL_ROOT / "vps" / "live-leo-telegram" / "SKILL.md").read_text() kb_squashed = " ".join(kb_text.split()) telegram_squashed = " ".join(telegram_text.split()) assert "final reply must be at most 220 words" in kb_squashed assert "smaller word or line limit is mandatory" in kb_squashed assert "at most three short bullets" in kb_squashed assert "Do not mirror this skill" in kb_squashed assert "Broad or multi-part questions still receive one compressed answer" in telegram_squashed assert "explicitly requests a detailed audit or long-form document" in telegram_squashed def test_vps_kb_skill_encodes_apply_readiness_beliefs_and_current_proof_boundaries() -> None: text = (SKILL_ROOT / "vps" / "teleo-kb-bridge" / "SKILL.md").read_text() squashed = " ".join(text.split()) assert "approved_needs_apply_payload" in text assert "normalize into a strict payload and review that new packet" in squashed assert "never tell the operator to apply the legacy row" in squashed assert "search` searches claims and identity context, not canonical source rows" in squashed assert "an existing canonical link is still canonical evidence" in squashed assert "Current agent-owned positions live in `public.beliefs`" in squashed for field in ("`agent_id`", "`level`", "`statement`", "`falsifier`", "`rank`"): assert field in text assert "has no claim-ID foreign key" in squashed assert "neither is generic stance storage" in squashed assert "source compiler is build-only and locally proven" in squashed assert "No active general DB-to-`SOUL.md` renderer automation is currently proven" in squashed assert "Unchanged table totals also do not prove unchanged rows" in squashed assert "current `public.claim_edges` has no rationale field" in squashed assert "there is no belief-edge table" in squashed assert "`public.claim_edges` connects claims, never belief rows" in squashed assert "a chat label or `source_ref` as source identity" in squashed assert "cannot insert `behavioral_rules` or `governance_gates`, update beliefs" in squashed assert "`applied_at` belongs to the proposal receipt, not each inserted row" in squashed assert "`state.db` and session JSONL can preserve continuity across restart" in squashed assert "It has no `author`, `title`, `publisher`, or publication-date column" in squashed