From 8e3d1877c6019ddf61826f1ae56f9e535be8f55e Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Thu, 9 Jul 2026 01:09:53 +0200 Subject: [PATCH] Source-control leoclean KB bridge (#59) --- .../leoclean-bin/cloudsql_memory_tool.py | 1100 +++++++++++++++++ hermes-agent/leoclean-bin/kb_tool.py | 926 ++++++++++++++ hermes-agent/leoclean-bin/teleo-kb | 64 + .../test_hermes_leoclean_kb_bridge_source.py | 63 + 4 files changed, 2153 insertions(+) create mode 100755 hermes-agent/leoclean-bin/cloudsql_memory_tool.py create mode 100755 hermes-agent/leoclean-bin/kb_tool.py create mode 100755 hermes-agent/leoclean-bin/teleo-kb create mode 100644 tests/test_hermes_leoclean_kb_bridge_source.py diff --git a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py new file mode 100755 index 0000000..bd16e22 --- /dev/null +++ b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py @@ -0,0 +1,1100 @@ +#!/usr/bin/env python3 +"""Cloud SQL memory and KB-proposal bridge for the GCP leoclean profile. + +Canonical claim/strategy application remains review-gated. Read commands query +the canonical Teleo KB first, then fall back to the restored `teleo_restore` +shadow schema in Cloud SQL. Proposal commands write staged review records into +`kb_stage.kb_proposals`; they do not apply canonical truth directly. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +from typing import Any + +STOPWORDS = { + "a", "about", "after", "all", "an", "and", "are", "as", "at", "be", "before", + "by", "can", "cloud", "context", "do", "does", "for", "from", "gcp", "give", + "how", "i", "in", "into", "is", "it", "leo", "me", "memory", "of", "on", "or", + "our", "production", "should", "source", "that", "the", "their", "this", "to", + "using", "what", "when", "where", "which", "who", "why", "with", "you", "your", +} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--host", default=os.environ.get("TELEO_CLOUDSQL_HOST", "10.61.0.3")) + parser.add_argument("--port", default=os.environ.get("TELEO_CLOUDSQL_PORT", "5432")) + parser.add_argument("--db", default=os.environ.get("TELEO_CLOUDSQL_DB", "teleo_kb")) + parser.add_argument("--canonical-db", default=os.environ.get("TELEO_CANONICAL_CLOUDSQL_DB", "teleo_canonical")) + parser.add_argument("--user", default=os.environ.get("TELEO_CLOUDSQL_USER", "postgres")) + parser.add_argument( + "--password-secret", + default=os.environ.get("TELEO_CLOUDSQL_PASSWORD_SECRET", "gcp-teleo-pgvector-standby-postgres-password"), + ) + parser.add_argument("--project", default=os.environ.get("TELEO_GCP_PROJECT", "teleo-501523")) + parser.add_argument("--format", choices=["markdown", "json"], default="markdown") + parser.add_argument( + "--include-excerpts", + "--raw-ok", + dest="include_excerpts", + action="store_true", + default=os.environ.get("TELEO_MEMORY_INCLUDE_EXCERPTS", "") == "1", + help="Include short memory excerpts for agent context. Default proof mode is redacted.", + ) + parser.add_argument("--redacted", action="store_true", default=os.environ.get("TELEO_MEMORY_REDACT", "") == "1") + sub = parser.add_subparsers(dest="command", required=True) + + search = sub.add_parser("search", help="Search restored Cloud SQL memory rows.") + search.add_argument("query") + search.add_argument("--limit", type=int, default=8) + search.add_argument("--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored.") + search.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + search.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) + search.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) + + context = sub.add_parser("context", help="Build an agent-ready memory context bundle.") + context.add_argument("query") + context.add_argument("--limit", type=int, default=8) + context.add_argument("--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored.") + context.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + context.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) + context.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) + + show = sub.add_parser("show", help="Show one canonical claim with evidence and edges.") + show.add_argument("claim_id") + show.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + show.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) + show.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) + + evidence = sub.add_parser("evidence", help="Show canonical evidence rows for one claim.") + evidence.add_argument("claim_id") + evidence.add_argument("--limit", type=int, default=8) + evidence.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + evidence.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) + evidence.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) + + edges = sub.add_parser("edges", help="Show canonical graph edges for one claim.") + edges.add_argument("claim_id") + edges.add_argument("--limit", type=int, default=12) + edges.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + edges.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) + edges.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) + + status = sub.add_parser("status", help="Read back Cloud SQL memory backend status.") + status.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + status.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) + + propose = sub.add_parser( + "propose-core-change", + help="Stage a review-gated canonical KB change instead of writing core truth to runtime memory.", + ) + propose.add_argument("--proposal-type", choices=["revise_claim", "revise_strategy"], required=True) + propose.add_argument("--target-kind", choices=["claim", "belief", "strategy", "identity", "role", "telos", "framework", "reasoning_tool", "other"], required=True) + propose.add_argument("--target-ref", required=True, help="Claim id, strategy id/name, or unresolved target description.") + propose.add_argument("--current", default="", help="Current/old claim or state, if known.") + propose.add_argument("--proposed", required=True, help="Proposed replacement or correction.") + propose.add_argument("--evidence", action="append", default=[], help="Evidence/source pointer. Can be repeated.") + propose.add_argument("--implication", action="append", default=[], help="Downstream implication. Can be repeated.") + propose.add_argument("--proposed-by", default="leo") + propose.add_argument("--originator", default="", help="Human or agent who originated the correction.") + propose.add_argument("--channel", default="telegram") + propose.add_argument("--source-ref", required=True) + propose.add_argument("--rationale", required=True) + propose.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + + list_proposals = sub.add_parser("list-proposals", help="List staged KB proposals from Cloud SQL.") + list_proposals.add_argument("--status", default="pending_review") + list_proposals.add_argument("--limit", type=int, default=10) + list_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) + + return parser.parse_args() + + +def terms_for(query: str) -> list[str]: + terms: list[str] = [] + for token in re.findall(r"[A-Za-z0-9][A-Za-z0-9.+#/-]*", query.lower()): + token = token.strip("-/") + if len(token) < 3 or token in STOPWORDS: + continue + if token not in terms: + terms.append(token) + return terms[:14] or [query.lower()] + + +def sql_literal(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def sql_json_array(values: list[str]) -> str: + return "'{}'::json".format(json.dumps(values).replace("'", "''")) + + +def sql_array(values: list[str], cast: str = "text") -> str: + return "array[" + ",".join(sql_literal(value) for value in values) + f"]::{cast}[]" + + +def password(args: argparse.Namespace) -> str: + if os.environ.get("PGPASSWORD"): + return os.environ["PGPASSWORD"] + result = subprocess.run( + [ + "gcloud", + "secrets", + "versions", + "access", + "latest", + f"--secret={args.password_secret}", + f"--project={args.project}", + ], + text=True, + capture_output=True, + check=False, + ) + if result.returncode != 0: + raise SystemExit(f"Secret access failed: {result.stderr.strip()}") + return result.stdout.strip() + + +def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str: + env = os.environ.copy() + env["PGPASSWORD"] = password(args) + conn = f"host={args.host} port={args.port} dbname={db or args.db} user={args.user} sslmode=require" + result = subprocess.run( + ["psql", conn, "-At", "-q", "-v", "ON_ERROR_STOP=1"], + text=True, + capture_output=True, + input=sql, + env=env, + check=False, + ) + if result.returncode != 0: + raise SystemExit( + f"Cloud SQL query failed ({result.returncode})\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" + ) + return result.stdout + + +def psql_json_lines(args: argparse.Namespace, sql: str, db: str | None = None) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for line in run_psql(args, sql, db=db).splitlines(): + line = line.strip() + if line: + rows.append(json.loads(line)) + return rows + + +def include_excerpts(args: argparse.Namespace) -> bool: + return bool(getattr(args, "include_excerpts", False)) and not bool(getattr(args, "redacted", False)) + + +def query_canonical_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, Any]: + terms = terms_for(query) + patterns = [f"%{term}%" for term in terms] + min_score = 2 if len(patterns) >= 3 else 1 + claim_limit = max(1, min(limit, 20)) + context_limit = max(1, min(int(getattr(args, "context_limit", limit) or limit), 20)) + + claims_sql = f""" +with params(patterns) as ( + values ({sql_array(patterns)}) +), ranked as ( + select c.id, + c.type, + c.text, + c.status, + c.confidence, + c.tags, + ( + select count(*) + from params p, unnest(p.patterns) pattern + where lower(c.text) like pattern + or exists ( + select 1 from unnest(coalesce(c.tags, '{{}}'::text[])) tag + where lower(tag) like pattern + ) + ) as score, + (select count(*) from public.claim_evidence ce where ce.claim_id = c.id) as evidence_count, + (select count(*) from public.claim_edges e where e.from_claim = c.id or e.to_claim = c.id) as edge_count + from public.claims c + where c.status = 'open' +) +select jsonb_build_object( + 'id', id::text, + 'type', type, + 'text', text, + 'status', status, + 'confidence', confidence, + 'tags', coalesce(to_jsonb(tags), '[]'::jsonb), + 'score', score, + 'evidence_count', evidence_count, + 'edge_count', edge_count + )::text + from ranked + where score >= {min_score} + order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text) + limit {claim_limit}; +""" + context_sql = f""" +with params(patterns) as ( + values ({sql_array(patterns)}) +), corpus as ( + select 'persona' as source, + a.handle as owner, + p.name as title, + concat_ws(E'\\n', p.voice, p.role, p.source_ref, p.lens) as body + from public.personas p + join public.agents a on a.id = p.agent_id + union all + select 'strategy' as source, + a.handle as owner, + 'active strategy v' || s.version::text as title, + concat_ws(E'\\n', s.diagnosis, s.guiding_policy, s.proximate_objectives::text) as body + from public.strategies s + join public.agents a on a.id = s.agent_id + where s.active + union all + select 'belief' as source, + a.handle as owner, + concat_ws(' ', b.level::text, 'rank', b.rank::text) as title, + concat_ws(E'\\n', b.statement, b.falsifier) as body + from public.beliefs b + join public.agents a on a.id = b.agent_id + where b.status = 'active' + union all + select 'blindspot' as source, + a.handle as owner, + bs.name as title, + concat_ws(E'\\n', bs.description, bs.correction, bs.kind) as body + from public.blindspots bs + join public.agents a on a.id = bs.agent_id + where bs.status = 'active' + union all + select 'agent_role' as source, + a.handle as owner, + ar.title as title, + ar.description as body + from public.agent_roles ar + join public.agents a on a.id = ar.agent_id + union all + select 'peer_model' as source, + subject.handle as owner, + 'peer ' || peer.handle || ': ' || pm.domain as title, + concat_ws(E'\\n', pm.outranks_on, pm.deference_rule) as body + from public.peer_models pm + join public.agents subject on subject.id = pm.subject_id + join public.agents peer on peer.id = pm.peer_id + union all + select 'behavioral_rule' as source, + coalesce(a.handle, 'collective') as owner, + br.category::text as title, + concat_ws(E'\\n', br.rule, br.rationale) as body + from public.behavioral_rules br + left join public.agents a on a.id = br.agent_id + union all + select 'contributor_rule' as source, + coalesce(a.handle, 'collective') as owner, + cr.name as title, + concat_ws(E'\\n', cr.directive, cr.ci_tier, cr.weighting, cr.rationale) as body + from public.contributor_rules cr + left join public.agents a on a.id = cr.agent_id + union all + select 'reasoning_tool' as source, + coalesce(a.handle, 'collective') as owner, + rt.name as title, + concat_ws(E'\\n', rt.description, rt.category) as body + from public.reasoning_tools rt + left join public.agents a on a.id = rt.agent_id + union all + select 'governance_gate' as source, + coalesce(a.handle, 'collective') as owner, + gg.name as title, + concat_ws(E'\\n', gg.criteria, gg.evidence_bar, gg.pass_condition) as body + from public.governance_gates gg + left join public.agents a on a.id = gg.agent_id +), ranked as ( + select source, + owner, + title, + body, + ( + select count(*) + from params p, unnest(p.patterns) pattern + where lower(concat_ws(E'\\n', source, owner, title, body)) like pattern + ) as score + from corpus +) +select jsonb_build_object( + 'source', source, + 'owner', owner, + 'title', title, + 'body', left(coalesce(body, ''), 900), + 'score', score + )::text + from ranked + where score >= {min_score} + order by score desc, source, owner, title + limit {context_limit}; +""" + claims = psql_json_lines(args, claims_sql, db=args.canonical_db) + claim_ids = [claim["id"] for claim in claims] + evidence: dict[str, list[dict[str, Any]]] = {claim_id: [] for claim_id in claim_ids} + edges: dict[str, list[dict[str, Any]]] = {claim_id: [] for claim_id in claim_ids} + if claim_ids: + ids = sql_array(claim_ids, "uuid") + evidence_sql = f""" +with ranked as ( + select ce.claim_id, + ce.role::text as role, + ce.weight, + s.source_type, + s.url, + s.storage_path, + left(coalesce(s.excerpt, ''), 800) as excerpt, + row_number() over ( + partition by ce.claim_id + order by (s.storage_path like 'inbox/archive/%') desc, + ce.role::text, + s.storage_path nulls last, + s.url nulls last + ) as rn + from public.claim_evidence ce + join public.sources s on s.id = ce.source_id + where ce.claim_id = any({ids}) +) +select jsonb_build_object( + 'claim_id', claim_id::text, + 'role', role, + 'weight', weight, + 'source_type', source_type, + 'url', url, + 'storage_path', storage_path, + 'excerpt', excerpt + )::text + from ranked + where rn <= 4 + order by claim_id, rn; +""" + for row in psql_json_lines(args, evidence_sql, db=args.canonical_db): + evidence.setdefault(row["claim_id"], []).append(row) + + edges_sql = f""" +with base(id) as ( + select unnest({ids}) +), edge_rows as ( + select base.id as claim_id, + case when e.from_claim = base.id then 'outgoing' else 'incoming' end as direction, + e.edge_type::text as edge_type, + other.id as connected_id, + other.text as connected_text, + other.tags as connected_tags, + row_number() over ( + partition by base.id + order by case e.edge_type::text + when 'supports' then 1 + when 'challenges' then 2 + when 'contradicts' then 3 + when 'requires' then 4 + when 'supersedes' then 5 + when 'constrains' then 6 + when 'causes' then 7 + when 'accelerates' then 8 + else 9 + end, + other.text + ) as rn + from base + join public.claim_edges e on e.from_claim = base.id or e.to_claim = base.id + join public.claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end +) +select jsonb_build_object( + 'claim_id', claim_id::text, + 'direction', direction, + 'edge_type', edge_type, + 'connected_id', connected_id::text, + 'connected_text', connected_text, + 'connected_tags', coalesce(to_jsonb(connected_tags), '[]'::jsonb) + )::text + from edge_rows + where rn <= 6 + order by claim_id, rn; +""" + for row in psql_json_lines(args, edges_sql, db=args.canonical_db): + edges.setdefault(row["claim_id"], []).append(row) + + for claim in claims: + claim["evidence"] = evidence.get(claim["id"], []) + claim["edges"] = edges.get(claim["id"], []) + + context_rows = psql_json_lines(args, context_sql, db=args.canonical_db) + result: dict[str, Any] = { + "artifact": "teleo_cloudsql_canonical_kb_context", + "backend": f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage", + "query": query, + "terms": terms, + "claims": claims, + "context_rows": context_rows, + "hit_count_total": len(claims) + len(context_rows), + "hits": [ + { + "source_table": "public.claims", + "row_id": claim["id"], + "score": claim.get("score"), + "claim_type": claim.get("type"), + "claim_status": claim.get("status"), + "confidence": claim.get("confidence"), + "tags": claim.get("tags") or [], + "evidence_count": claim.get("evidence_count"), + "edge_count": claim.get("edge_count"), + "excerpt": claim.get("text"), + } + for claim in claims + ] + [ + { + "source_table": f"public.{row['source']}", + "row_id": f"{row['owner']}:{row['title']}", + "score": row.get("score"), + "actor": row.get("owner"), + "excerpt": row.get("body"), + } + for row in context_rows + ], + "privacy": "redacted proof mode by default; pass --include-excerpts/--raw-ok only for private agent context", + } + if not include_excerpts(args): + for claim in result["claims"]: + claim.pop("text", None) + for row in claim.get("evidence", []): + row.pop("excerpt", None) + for row in claim.get("edges", []): + row.pop("connected_text", None) + for row in result["context_rows"]: + row.pop("body", None) + for hit in result["hits"]: + hit.pop("excerpt", None) + result["privacy"] = "redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, canonical table, row id, timestamps, and evidence links" + else: + result["privacy"] = "private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts" + return result + + +def query_audit_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, Any]: + terms = terms_for(query) + sql = f""" +with q(term) as ( + select lower(value) from json_array_elements_text({sql_json_array(terms)}) as value +), raw_hits as ( + select + 'response_audit'::text as source_table, + id::text as row_id, + coalesce(timestamp, created_at) as event_time, + coalesce("user", agent, '') as actor, + concat_ws(' ', query, conversation_window, entities_matched, claims_matched, + retrieval_layers_hit, retrieval_gap, market_data, research_context, kb_context_text, + tool_calls, raw_response, display_response, query_type) as body + from teleo_restore.response_audit + union all + select + 'sources'::text as source_table, + path as row_id, + coalesce(updated_at, created_at) as event_time, + coalesce(submitted_by, original_author, original_author_handle, '') as actor, + concat_ws(' ', path, status, priority, priority_log, extraction_model, last_error, + feedback, content_type, original_author, original_author_handle) as body + from teleo_restore.sources + union all + select + 'agent_research_runs'::text as source_table, + id as row_id, + coalesce(completed_at, created_at) as event_time, + coalesce(agent_slug, '') as actor, + concat_ws(' ', source_surface, source_ref, request_kind, prompt_excerpt, + selected_provider, selected_route, status, answer_excerpt, proof_ref) as body + from teleo_restore.agent_research_runs + union all + select + 'agent_tool_invocations'::text as source_table, + id as row_id, + created_at as event_time, + coalesce(provider, tool_name, '') as actor, + concat_ws(' ', provider, tool_name, tool_category, endpoint_host, decision, + decision_reason, rail, network, currency, error_class) as body + from teleo_restore.agent_tool_invocations +), matched as ( + select *, + (select count(*) from q where lower(raw_hits.body) like '%' || q.term || '%') as score + from raw_hits + where exists (select 1 from q where lower(raw_hits.body) like '%' || q.term || '%') +), ranked as ( + select *, row_number() over (order by score desc, event_time desc nulls last, source_table, row_id) as rn + from matched +) +select json_build_object( + 'hit_count_total', (select count(*) from matched), + 'hits', coalesce((select json_agg(json_build_object( + 'source_table', source_table, + 'row_id', row_id, + 'event_time', event_time, + 'actor', actor, + 'score', score, + 'body_chars', length(body), + 'body_md5', md5(body), + 'excerpt', left(regexp_replace(body, '\\s+', ' ', 'g'), 700) + ) order by rn) from ranked where rn <= {max(1, min(limit, 20))}), '[]'::json) +)::text; +""" + text = run_psql(args, sql).strip() + result = json.loads(text) + result.update( + { + "artifact": "teleo_cloudsql_memory_context", + "backend": "cloudsql:teleo-pgvector-standby/teleo_kb/teleo_restore", + "query": query, + "terms": terms, + "privacy": "redacted proof mode by default; pass --include-excerpts/--raw-ok only for private agent context", + } + ) + if not include_excerpts(args): + for hit in result["hits"]: + hit.pop("excerpt", None) + result["privacy"] = "redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, source table, row id, and timestamps" + else: + result["privacy"] = "private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts" + return result + + +def query_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, Any]: + canonical = query_canonical_rows(args, query, limit) + if canonical.get("hit_count_total", 0) or canonical.get("hits"): + return canonical + audit = query_audit_rows(args, query, limit) + audit["canonical_fallback_reason"] = "no matching canonical public/persona/strategy/belief rows" + return audit + + +def canonical_backend(args: argparse.Namespace) -> str: + return f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage" + + +def canonical_claim(args: argparse.Namespace, claim_id: str) -> dict[str, Any] | None: + sql = f""" +select jsonb_build_object( + 'id', id::text, + 'type', type, + 'text', text, + 'status', status, + 'confidence', confidence, + 'tags', coalesce(to_jsonb(tags), '[]'::jsonb), + 'superseded_by', superseded_by::text, + 'created_at', created_at::text, + 'updated_at', updated_at::text + )::text + from public.claims + where id = {sql_literal(claim_id)}::uuid; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + return rows[0] if rows else None + + +def canonical_evidence(args: argparse.Namespace, claim_id: str, limit: int) -> list[dict[str, Any]]: + sql = f""" +with ranked as ( + select ce.claim_id, + ce.role::text as role, + ce.weight, + s.source_type, + s.url, + s.storage_path, + left(coalesce(s.excerpt, ''), 800) as excerpt, + row_number() over ( + partition by ce.claim_id + order by (s.storage_path like 'inbox/archive/%') desc, + ce.role::text, + s.storage_path nulls last, + s.url nulls last + ) as rn + from public.claim_evidence ce + join public.sources s on s.id = ce.source_id + where ce.claim_id = {sql_literal(claim_id)}::uuid +) +select jsonb_build_object( + 'claim_id', claim_id::text, + 'role', role, + 'weight', weight, + 'source_type', source_type, + 'url', url, + 'storage_path', storage_path, + 'excerpt', excerpt + )::text + from ranked + where rn <= {max(1, min(limit, 50))} + order by rn; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + if not include_excerpts(args): + for row in rows: + row.pop("excerpt", None) + return rows + + +def canonical_edges(args: argparse.Namespace, claim_id: str, limit: int) -> list[dict[str, Any]]: + sql = f""" +with base(id) as ( + values ({sql_literal(claim_id)}::uuid) +), edge_rows as ( + select base.id as claim_id, + case when e.from_claim = base.id then 'outgoing' else 'incoming' end as direction, + e.edge_type::text as edge_type, + other.id as connected_id, + other.text as connected_text, + other.tags as connected_tags, + row_number() over ( + partition by base.id + order by case e.edge_type::text + when 'supports' then 1 + when 'challenges' then 2 + when 'contradicts' then 3 + when 'requires' then 4 + when 'supersedes' then 5 + when 'constrains' then 6 + when 'causes' then 7 + when 'accelerates' then 8 + else 9 + end, + other.text + ) as rn + from base + join public.claim_edges e on e.from_claim = base.id or e.to_claim = base.id + join public.claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end +) +select jsonb_build_object( + 'claim_id', claim_id::text, + 'direction', direction, + 'edge_type', edge_type, + 'connected_id', connected_id::text, + 'connected_text', connected_text, + 'connected_tags', coalesce(to_jsonb(connected_tags), '[]'::jsonb) + )::text + from edge_rows + where rn <= {max(1, min(limit, 50))} + order by rn; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + if not include_excerpts(args): + for row in rows: + row.pop("connected_text", None) + return rows + + +def show_canonical_claim(args: argparse.Namespace) -> dict[str, Any]: + claim = canonical_claim(args, args.claim_id) + if not claim: + raise SystemExit(f"claim not found: {args.claim_id}") + if not include_excerpts(args): + claim.pop("text", None) + return { + "artifact": "teleo_cloudsql_canonical_kb_claim", + "backend": canonical_backend(args), + "claim": claim, + "evidence": canonical_evidence(args, args.claim_id, 8), + "edges": canonical_edges(args, args.claim_id, 12), + } + + +def evidence_canonical_claim(args: argparse.Namespace) -> dict[str, Any]: + return { + "artifact": "teleo_cloudsql_canonical_kb_evidence", + "backend": canonical_backend(args), + "claim_id": args.claim_id, + "evidence": canonical_evidence(args, args.claim_id, args.limit), + } + + +def edges_canonical_claim(args: argparse.Namespace) -> dict[str, Any]: + return { + "artifact": "teleo_cloudsql_canonical_kb_edges", + "backend": canonical_backend(args), + "claim_id": args.claim_id, + "edges": canonical_edges(args, args.claim_id, args.limit), + } + + +def propose_core_change(args: argparse.Namespace) -> dict[str, Any]: + payload = { + "target_kind": args.target_kind, + "target_ref": args.target_ref, + "current": args.current or None, + "proposed": args.proposed, + "evidence": args.evidence, + "implications": args.implication, + "originator": args.originator or None, + "routing_rule": "core Teleo changes must be staged in kb_stage.kb_proposals, not saved as runtime memory", + "canonical_apply_required": True, + } + payload_json = json.dumps(payload, sort_keys=True) + sql = f""" +with params as ( + select + {sql_literal(args.proposal_type)}::text as proposal_type, + nullif({sql_literal(args.proposed_by)}, '') as proposed_by_handle, + nullif({sql_literal(args.channel)}, '') as channel, + nullif({sql_literal(args.source_ref)}, '') as source_ref, + {sql_literal(args.rationale)} as rationale, + {sql_literal(payload_json)}::jsonb as payload +), proposer as ( + select a.id, a.handle + from public.agents a, params p + where a.handle = p.proposed_by_handle + limit 1 +), inserted as ( + insert into kb_stage.kb_proposals ( + proposal_type, + status, + proposed_by_handle, + proposed_by_agent_id, + channel, + source_ref, + rationale, + payload + ) + select + p.proposal_type, + 'pending_review', + p.proposed_by_handle, + proposer.id, + coalesce(p.channel, 'telegram'), + p.source_ref, + p.rationale, + p.payload + from params p + left join proposer on true + returning * +) +select jsonb_build_object( + 'artifact', 'teleo_cloudsql_kb_core_change_proposal', + 'backend', {sql_literal(canonical_backend(args))}, + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'proposed_by_agent_id', proposed_by_agent_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', rationale, + 'payload', payload, + 'created_at', created_at::text, + 'canonical_apply_done', false, + 'runtime_memory_write_done', false +)::text +from inserted; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + if not rows: + raise SystemExit("No core-change proposal inserted.") + return rows[0] + + +def list_proposals(args: argparse.Namespace) -> dict[str, Any]: + where = "" + if args.status.lower() != "all": + where = f"where 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, 500), + 'payload', payload, + 'created_at', created_at::text, + 'reviewed_at', reviewed_at::text, + 'applied_at', applied_at::text +)::text +from kb_stage.kb_proposals +{where} +order by created_at desc +limit {max(1, min(args.limit, 100))}; +""" + return { + "artifact": "teleo_cloudsql_kb_proposal_list", + "backend": canonical_backend(args), + "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( + 'artifact', 'teleo_cloudsql_kb_proposal', + 'backend', {sql_literal(canonical_backend(args))}, + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'proposed_by_agent_id', proposed_by_agent_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', rationale, + 'payload', payload, + 'reviewed_by_handle', reviewed_by_handle, + 'reviewed_at', reviewed_at::text, + 'review_note', review_note, + 'applied_by_handle', applied_by_handle, + 'applied_at', applied_at::text, + 'created_at', created_at::text, + 'updated_at', updated_at::text +)::text +from kb_stage.kb_proposals +where id = {sql_literal(args.proposal_id)}::uuid; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + if not rows: + raise SystemExit(f"proposal not found: {args.proposal_id}") + return rows[0] + + +def status(args: argparse.Namespace) -> dict[str, Any]: + canonical_sql = """ +select json_build_object( + 'db_identity', current_database() || '|' || current_user, + 'extensions', (select json_agg(extname order by extname) from pg_extension), + 'schema_tables', ( + select json_object_agg(table_schema, table_count) from ( + select table_schema, count(*) as table_count + from information_schema.tables + where table_type='BASE TABLE' and table_schema in ('public','kb_stage') + group by table_schema + order by table_schema + ) counts + ), + 'high_signal_rows', json_build_object( + 'agents', (select count(*) from public.agents), + 'personas', (select count(*) from public.personas), + 'strategies', (select count(*) from public.strategies), + 'beliefs', (select count(*) from public.beliefs), + 'claims', (select count(*) from public.claims), + 'claim_edges', (select count(*) from public.claim_edges), + 'claim_evidence', (select count(*) from public.claim_evidence), + 'sources', (select count(*) from public.sources), + 'kb_proposals', (select count(*) from kb_stage.kb_proposals) + ) +)::text; +""" + audit_sql = """ +select json_build_object( + 'db_identity', current_database() || '|' || current_user, + 'extensions', (select json_agg(extname order by extname) from pg_extension), + 'teleo_restore_tables', (select count(*) from pg_tables where schemaname='teleo_restore'), + 'total_rows', ( + select sum(row_count) from ( + select count(*)::bigint as row_count from teleo_restore.response_audit + union all select count(*)::bigint from teleo_restore.sources + union all select count(*)::bigint from teleo_restore.agent_research_runs + union all select count(*)::bigint from teleo_restore.agent_tool_invocations + union all select count(*)::bigint from teleo_restore.audit_log + union all select count(*)::bigint from teleo_restore.metrics_snapshots + union all select count(*)::bigint from teleo_restore.prs + union all select count(*)::bigint from teleo_restore.review_records + ) counts + ), + 'high_signal_rows', json_build_object( + 'response_audit', (select count(*) from teleo_restore.response_audit), + 'sources', (select count(*) from teleo_restore.sources), + 'agent_research_runs', (select count(*) from teleo_restore.agent_research_runs), + 'agent_tool_invocations', (select count(*) from teleo_restore.agent_tool_invocations) + ) +)::text; +""" + result = json.loads(run_psql(args, audit_sql).strip()) + result["canonical"] = json.loads(run_psql(args, canonical_sql, db=args.canonical_db).strip()) + result["artifact"] = "teleo_cloudsql_memory_status" + result["backend"] = f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage primary; {args.db}/teleo_restore fallback" + return result + + +def emit_json(value: dict[str, Any]) -> None: + print(json.dumps(value, indent=2)) + + +def emit_markdown(value: dict[str, Any]) -> None: + if value["artifact"] == "teleo_cloudsql_memory_status": + print("# Teleo Cloud SQL Memory Status\n") + print(f"- Backend: `{value['backend']}`") + print(f"- DB: `{value['db_identity']}`") + if "canonical" in value: + canonical = value["canonical"] + 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)}`") + 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']}") + print(f"- High-signal rows: `{json.dumps(value['high_signal_rows'], sort_keys=True)}`") + return + + if "claims" in value: + print("# Teleo KB Context\n") + print(f"- Backend: `{value['backend']}`") + print(f"- Query: `{value['query']}`") + print(f"- Terms: {', '.join(value['terms'])}") + print(f"- Privacy: {value['privacy']}\n") + if value.get("context_rows"): + print("## Soul / Context Rows\n") + for row in value["context_rows"]: + print( + f"- `{row.get('source')}` / `{row.get('owner')}` / `{row.get('title')}` " + f"(score {row.get('score')}): {row.get('body', '[redacted]')}" + ) + print() + print("## Claims\n") + if not value.get("claims"): + print("No matching canonical claims found.") + return + for i, claim in enumerate(value["claims"], 1): + print(f"### {i}. {claim.get('text', '[redacted claim text]')}\n") + print(f"- id: `{claim['id']}`") + print(f"- type: `{claim.get('type')}`; confidence: `{claim.get('confidence')}`; score: `{claim.get('score')}`") + print(f"- tags: `{', '.join(claim.get('tags') or [])}`") + print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`") + print(f"- edge rows: `{claim.get('edge_count', len(claim.get('edges', [])))}`") + if claim.get("evidence"): + print("\nEvidence:") + for ev in claim["evidence"]: + source = ev.get("storage_path") or ev.get("url") or "(no source pointer)" + excerpt = f" - {ev['excerpt']}" if ev.get("excerpt") else "" + print(f"- `{ev.get('role')}` / `{ev.get('source_type')}` / `{source}`{excerpt}") + if claim.get("edges"): + print("\nEdges:") + for edge in claim["edges"]: + text = f": {edge['connected_text']}" if edge.get("connected_text") else "" + print(f"- `{edge.get('direction')}` `{edge.get('edge_type')}` ({edge.get('connected_id')}){text}") + print() + return + + if value["artifact"] == "teleo_cloudsql_canonical_kb_claim": + claim = value["claim"] + print("# Teleo KB Claim\n") + print(f"- Backend: `{value['backend']}`") + print(f"- id: `{claim['id']}`") + print(f"- type: `{claim.get('type')}`; status: `{claim.get('status')}`; confidence: `{claim.get('confidence')}`") + print(f"- tags: `{', '.join(claim.get('tags') or [])}`") + if claim.get("text"): + print(f"\n{claim['text']}\n") + print("## Evidence\n") + for ev in value.get("evidence", []): + source = ev.get("storage_path") or ev.get("url") or "(no source pointer)" + excerpt = f" - {ev['excerpt']}" if ev.get("excerpt") else "" + print(f"- `{ev.get('role')}` / `{ev.get('source_type')}` / `{source}`{excerpt}") + print("\n## Edges\n") + for edge in value.get("edges", []): + text = f": {edge['connected_text']}" if edge.get("connected_text") else "" + print(f"- `{edge.get('direction')}` `{edge.get('edge_type')}` ({edge.get('connected_id')}){text}") + return + + if value["artifact"] in {"teleo_cloudsql_canonical_kb_evidence", "teleo_cloudsql_canonical_kb_edges"}: + print(f"# {value['artifact']}\n") + print(f"- Backend: `{value['backend']}`") + print(f"- Claim: `{value['claim_id']}`\n") + rows = value.get("evidence") or value.get("edges") or [] + for row in rows: + print(f"- `{json.dumps(row, sort_keys=True)}`") + return + + if value["artifact"] == "teleo_cloudsql_kb_core_change_proposal": + print("# Staged KB Core-Change Proposal\n") + print(f"- Proposal id: `{value['id']}`") + print(f"- Type: `{value['proposal_type']}`") + print(f"- Status: `{value['status']}`") + print(f"- Source: `{value.get('source_ref') or '-'}`") + print(f"- Runtime memory write: `{value['runtime_memory_write_done']}`") + print(f"- Canonical apply done: `{value['canonical_apply_done']}`") + print("\n## Rationale\n") + print(value["rationale"]) + print("\n## Payload\n") + print(json.dumps(value.get("payload") or {}, indent=2, sort_keys=True)) + return + + if value["artifact"] == "teleo_cloudsql_kb_proposal_list": + print("# Staged KB Proposals\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"- 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']}`") + print(f"- Type: `{value['proposal_type']}`") + print(f"- Status: `{value['status']}`") + print(f"- Source: `{value.get('source_ref') or '-'}`") + print(f"- Created: `{value.get('created_at')}`") + print(f"- Applied at: `{value.get('applied_at') or '-'}`") + print("\n## Rationale\n") + print(value["rationale"]) + print("\n## Payload\n") + print(json.dumps(value.get("payload") or {}, indent=2, sort_keys=True)) + return + + print("# Teleo Cloud SQL Memory Context\n") + print(f"- Backend: `{value['backend']}`") + print(f"- Query: `{value['query']}`") + print(f"- Terms: {', '.join(value['terms'])}") + print(f"- Total matching rows: {value['hit_count_total']}") + print(f"- Privacy: {value['privacy']}\n") + for i, hit in enumerate(value["hits"], 1): + print(f"## Hit {i}: {hit['source_table']} / {hit['row_id']}") + print(f"- Time: {hit.get('event_time') or 'unknown'}") + print(f"- Actor/source: {hit.get('actor') or 'unknown'}") + print(f"- Score: {hit.get('score')}") + print(f"- Body chars: {hit.get('body_chars')}") + print(f"- Body md5: `{hit.get('body_md5')}`") + if "excerpt" in hit: + print(f"- Excerpt: {hit['excerpt']}") + print() + + +def main() -> None: + args = parse_args() + if args.command == "status": + result = status(args) + elif args.command == "show": + result = show_canonical_claim(args) + elif args.command == "evidence": + result = evidence_canonical_claim(args) + elif args.command == "edges": + result = edges_canonical_claim(args) + elif args.command == "propose-core-change": + result = propose_core_change(args) + elif args.command == "list-proposals": + result = list_proposals(args) + elif args.command == "show-proposal": + result = show_proposal(args) + else: + result = query_rows(args, args.query, args.limit) + if args.format == "json": + emit_json(result) + else: + emit_markdown(result) + + +if __name__ == "__main__": + main() diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py new file mode 100755 index 0000000..d82e3f9 --- /dev/null +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -0,0 +1,926 @@ +#!/usr/bin/env python3 +"""Teleo KB bridge for live agents. + +This is the small CLI surface a Hermes profile can call from Telegram/runtime +tooling. It reads the live Postgres KB over SSH and prints either Markdown or +JSON. + +Canonical knowledge writes remain locked. The only write surface here is a +proposal ledger under kb_stage: agents can propose a mutation for review, but +this tool does not apply it to public.claims, public.claim_edges, or evidence. +""" + +from __future__ import annotations + +import argparse +import json +import re +import subprocess +from collections import defaultdict +from typing import Any + +STOPWORDS = { + "a", + "about", + "across", + "all", + "an", + "and", + "are", + "as", + "at", + "be", + "by", + "can", + "do", + "does", + "for", + "from", + "give", + "how", + "i", + "in", + "into", + "is", + "it", + "me", + "of", + "on", + "or", + "our", + "that", + "the", + "their", + "this", + "to", + "what", + "when", + "where", + "which", + "who", + "why", + "with", +} + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--ssh", default="teleo@77.42.65.182") + parser.add_argument("--container", default="teleo-pg") + parser.add_argument("--db", default="teleo") + parser.add_argument("--format", choices=["markdown", "json"], default="markdown") + parser.add_argument( + "--local", + action="store_true", + help="Run docker exec locally instead of SSH. Use this on the VPS.", + ) + + sub = parser.add_subparsers(dest="command", required=True) + + def add_output_flag(command: argparse.ArgumentParser) -> None: + command.add_argument( + "--format", + choices=["markdown", "json"], + default=argparse.SUPPRESS, + help="Output format. Accepted here for agent CLI ergonomics.", + ) + + 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) + search.add_argument("--context-limit", type=int, default=5) + add_output_flag(search) + + show = sub.add_parser("show", help="Show one claim with evidence and edges.") + show.add_argument("claim_id") + add_output_flag(show) + + evidence = sub.add_parser("evidence", help="Show evidence rows for one claim.") + evidence.add_argument("claim_id") + evidence.add_argument("--limit", type=int, default=8) + add_output_flag(evidence) + + edges = sub.add_parser("edges", help="Show graph edges for one claim.") + edges.add_argument("claim_id") + edges.add_argument("--limit", type=int, default=12) + add_output_flag(edges) + + context = sub.add_parser("context", help="Build an agent-ready KB context bundle for a question.") + context.add_argument("query") + context.add_argument("--limit", type=int, default=4) + context.add_argument("--context-limit", type=int, default=6) + add_output_flag(context) + + propose_edge = sub.add_parser("propose-edge", help="Propose a claim graph edge for reviewer approval.") + propose_edge.add_argument("from_claim") + propose_edge.add_argument("edge_type") + propose_edge.add_argument("to_claim") + propose_edge.add_argument("--proposed-by", default="leo") + propose_edge.add_argument("--channel", default="telegram") + propose_edge.add_argument("--source-ref") + propose_edge.add_argument("--rationale", required=True) + add_output_flag(propose_edge) + + propose_attachment_eval = sub.add_parser( + "propose-attachment-evaluation", + help="Create an attach_evidence proposal from a parsed Telegram document evaluation packet.", + ) + propose_attachment_eval.add_argument("--payload-file", required=True) + propose_attachment_eval.add_argument("--proposed-by", default="leo") + propose_attachment_eval.add_argument("--channel", default="telegram") + propose_attachment_eval.add_argument("--source-ref", required=True) + propose_attachment_eval.add_argument("--rationale", required=True) + add_output_flag(propose_attachment_eval) + + record_document_eval = sub.add_parser( + "record-document-evaluation", + help="Record a lightweight document evaluation decision without creating a KB mutation proposal.", + ) + record_document_eval.add_argument("--payload-file", required=True) + record_document_eval.add_argument( + "--decision", + required=True, + choices=[ + "rejected_as_out_of_scope", + "not_kb_relevant", + "kb_relevant_no_action", + "candidate_for_review", + "needs_human_review", + ], + ) + record_document_eval.add_argument("--evaluated-by", default="leo") + record_document_eval.add_argument("--channel", default="telegram") + record_document_eval.add_argument("--source-ref", required=True) + record_document_eval.add_argument("--file-ref-id") + record_document_eval.add_argument("--rationale", required=True) + add_output_flag(record_document_eval) + + list_proposals = sub.add_parser("list-proposals", help="List KB mutation proposals.") + list_proposals.add_argument("--status", default="pending_review") + list_proposals.add_argument("--limit", type=int, default=10) + add_output_flag(list_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) + + return parser.parse_args() + + +def sql_literal(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def sql_array(values: list[str], cast: str = "text") -> str: + return "array[" + ",".join(sql_literal(value) for value in values) + f"]::{cast}[]" + + +def psql_json(args: argparse.Namespace, sql: str) -> list[dict[str, Any]]: + if args.local: + command = ["docker", "exec", "-i", args.container, "psql", "-U", "postgres", "-d", args.db, "-At", "-q"] + else: + remote = f"docker exec -i {args.container} psql -U postgres -d {args.db} -At -q" + command = ["ssh", "-o", "ConnectTimeout=15", "-o", "BatchMode=yes", args.ssh, remote] + result = subprocess.run(command, text=True, capture_output=True, input=sql, check=False) + if result.returncode != 0: + raise SystemExit( + f"psql command failed ({result.returncode})\nSTDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}" + ) + rows: list[dict[str, Any]] = [] + for line in result.stdout.splitlines(): + line = line.strip() + if line: + rows.append(json.loads(line)) + return rows + + +def query_terms(query: str) -> list[str]: + terms = [] + for token in re.findall(r"[A-Za-z0-9][A-Za-z0-9.+#/-]*", query.lower()): + token = token.strip("-/") + if len(token) < 3 or token in STOPWORDS: + continue + terms.append(token) + deduped: list[str] = [] + for term in terms: + if term not in deduped: + deduped.append(term) + return deduped[:10] or [query.lower()] + + +def truncate(value: str | None, length: int = 220) -> str: + if not value: + return "" + squashed = " ".join(value.split()) + if len(squashed) <= length: + return squashed + return squashed[: length - 1] + "..." + + +def find_claims(args: argparse.Namespace, query: str, limit: int) -> list[dict[str, Any]]: + patterns = [f"%{term}%" for term in query_terms(query)] + min_score = 2 if len(patterns) >= 3 else 1 + sql = f""" + with params(patterns) as ( + values ({sql_array(patterns)}) + ), + ranked as ( + select c.id, + c.type, + c.text, + c.status, + c.confidence, + c.tags, + ( + select count(*) + from params p, unnest(p.patterns) pattern + where lower(c.text) like pattern + or exists ( + select 1 from unnest(coalesce(c.tags, '{{}}'::text[])) tag + where lower(tag) like pattern + ) + ) as score, + (select count(*) from claim_evidence ce where ce.claim_id = c.id) as evidence_count, + (select count(*) from claim_edges e where e.from_claim = c.id or e.to_claim = c.id) as edge_count + from claims c + where c.status = 'open' + ) + select jsonb_build_object( + 'id', id::text, + 'type', type, + 'text', text, + 'confidence', confidence, + 'tags', coalesce(to_jsonb(tags), '[]'::jsonb), + 'score', score, + 'evidence_count', evidence_count, + 'edge_count', edge_count + )::text + from ranked + where score >= {min_score} + order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text) + limit {limit}; + """ + return psql_json(args, sql) + + +def find_context_rows(args: argparse.Namespace, query: str, limit: int) -> list[dict[str, Any]]: + patterns = [f"%{term}%" for term in query_terms(query)] + min_score = 2 if len(patterns) >= 3 else 1 + sql = f""" + with params(patterns) as ( + values ({sql_array(patterns)}) + ), + corpus as ( + select 'persona' as source, + a.handle as owner, + p.name as title, + concat_ws(E'\\n', p.voice, p.role, p.source_ref) as body + from personas p + join agents a on a.id = p.agent_id + union all + select 'strategy' as source, + a.handle as owner, + 'active strategy v' || s.version::text as title, + concat_ws(E'\\n', s.diagnosis, s.guiding_policy, s.proximate_objectives::text) as body + from strategies s + join agents a on a.id = s.agent_id + where s.active + union all + select 'belief' as source, + a.handle as owner, + concat_ws(' ', b.level::text, 'rank', b.rank::text) as title, + concat_ws(E'\\n', b.statement, b.falsifier) as body + from beliefs b + join agents a on a.id = b.agent_id + where b.status = 'active' + union all + select 'blindspot' as source, + a.handle as owner, + bs.name as title, + concat_ws(E'\\n', bs.description, bs.correction, bs.kind) as body + from blindspots bs + join agents a on a.id = bs.agent_id + where bs.status = 'active' + union all + select 'agent_role' as source, + a.handle as owner, + ar.title as title, + ar.description as body + from agent_roles ar + join agents a on a.id = ar.agent_id + union all + select 'peer_model' as source, + subject.handle as owner, + 'peer ' || peer.handle || ': ' || pm.domain as title, + concat_ws(E'\\n', pm.outranks_on, pm.deference_rule) as body + from peer_models pm + join agents subject on subject.id = pm.subject_id + join agents peer on peer.id = pm.peer_id + union all + select 'behavioral_rule' as source, + coalesce(a.handle, 'collective') as owner, + br.category::text as title, + concat_ws(E'\\n', br.rule, br.rationale) as body + from behavioral_rules br + left join agents a on a.id = br.agent_id + union all + select 'contributor_rule' as source, + coalesce(a.handle, 'collective') as owner, + cr.name as title, + concat_ws(E'\\n', cr.directive, cr.ci_tier, cr.weighting, cr.rationale) as body + from contributor_rules cr + left join agents a on a.id = cr.agent_id + union all + select 'reasoning_tool' as source, + coalesce(a.handle, 'collective') as owner, + rt.name as title, + concat_ws(E'\\n', rt.description, rt.category) as body + from reasoning_tools rt + left join agents a on a.id = rt.agent_id + union all + select 'governance_gate' as source, + coalesce(a.handle, 'collective') as owner, + gg.name as title, + concat_ws(E'\\n', gg.criteria, gg.evidence_bar, gg.pass_condition) as body + from governance_gates gg + left join agents a on a.id = gg.agent_id + ), + ranked as ( + select source, + owner, + title, + body, + ( + select count(*) + from params p, unnest(p.patterns) pattern + where lower(concat_ws(E'\\n', source, owner, title, body)) like pattern + ) as score + from corpus + ) + select jsonb_build_object( + 'source', source, + 'owner', owner, + 'title', title, + 'body', left(coalesce(body, ''), 900), + 'score', score + )::text + from ranked + where score >= {min_score} + order by score desc, source, owner, title + limit {limit}; + """ + return psql_json(args, sql) + + +def get_claim(args: argparse.Namespace, claim_id: str) -> dict[str, Any] | None: + sql = f""" + select jsonb_build_object( + 'id', id::text, + 'type', type, + 'text', text, + 'status', status, + 'confidence', confidence, + 'tags', coalesce(to_jsonb(tags), '[]'::jsonb), + 'superseded_by', superseded_by::text, + 'created_at', created_at::text, + 'updated_at', updated_at::text + )::text + from claims + where id = {sql_literal(claim_id)}::uuid; + """ + rows = psql_json(args, sql) + return rows[0] if rows else None + + +def propose_edge(args: argparse.Namespace) -> dict[str, Any]: + sql = f""" + with params as ( + select {sql_literal(args.from_claim)}::uuid as from_claim, + {sql_literal(args.to_claim)}::uuid as to_claim, + {sql_literal(args.edge_type)}::edge_type as edge_type, + nullif({sql_literal(args.proposed_by)}, '') as proposed_by_handle, + nullif({sql_literal(args.channel)}, '') as channel, + nullif({sql_literal(args.source_ref or '')}, '') as source_ref, + {sql_literal(args.rationale)} as rationale + ), + from_row as ( + select c.id, c.text + from claims c, params p + where c.id = p.from_claim + ), + to_row as ( + select c.id, c.text + from claims c, params p + where c.id = p.to_claim + ), + proposer as ( + select a.id, a.handle + from agents a, params p + where a.handle = p.proposed_by_handle + ), + existing_edge as ( + select e.id + from claim_edges e, params p + where e.from_claim = p.from_claim + and e.to_claim = p.to_claim + and e.edge_type = p.edge_type + limit 1 + ), + inserted as ( + insert into kb_stage.kb_proposals ( + proposal_type, + status, + proposed_by_handle, + proposed_by_agent_id, + channel, + source_ref, + rationale, + payload + ) + select 'add_edge', + 'pending_review', + p.proposed_by_handle, + proposer.id, + coalesce(p.channel, 'cli'), + p.source_ref, + p.rationale, + jsonb_build_object( + 'from_claim', from_row.id::text, + 'from_text', from_row.text, + 'edge_type', p.edge_type::text, + 'to_claim', to_row.id::text, + 'to_text', to_row.text, + 'existing_edge_id', (select id::text from existing_edge) + ) + from params p + join from_row on true + join to_row on true + left join proposer on true + returning * + ) + select jsonb_build_object( + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'proposed_by_agent_id', proposed_by_agent_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', rationale, + 'payload', payload, + 'created_at', created_at::text + )::text + from inserted; + """ + rows = psql_json(args, sql) + if not rows: + raise SystemExit( + "No proposal inserted. Check that both claim ids exist and edge_type is valid." + ) + return rows[0] + + + +def propose_attachment_evaluation(args: argparse.Namespace) -> dict[str, Any]: + try: + with open(args.payload_file, encoding="utf-8") as f: + payload = json.load(f) + except Exception as exc: + raise SystemExit(f"failed to read payload json: {exc}") from exc + + payload_json = json.dumps(payload, sort_keys=True) + sql = f""" + with params as ( + select nullif({sql_literal(args.proposed_by)}, '') as proposed_by_handle, + nullif({sql_literal(args.channel)}, '') as channel, + nullif({sql_literal(args.source_ref)}, '') as source_ref, + {sql_literal(args.rationale)} as rationale, + {sql_literal(payload_json)}::jsonb as payload + ), + proposer as ( + select a.id, a.handle + from agents a, params p + where a.handle = p.proposed_by_handle + ), + inserted as ( + insert into kb_stage.kb_proposals ( + proposal_type, + status, + proposed_by_handle, + proposed_by_agent_id, + channel, + source_ref, + rationale, + payload + ) + select 'attach_evidence', + 'pending_review', + p.proposed_by_handle, + proposer.id, + coalesce(p.channel, 'cli'), + p.source_ref, + p.rationale, + p.payload + from params p + left join proposer on true + returning * + ) + select jsonb_build_object( + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'proposed_by_agent_id', proposed_by_agent_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', rationale, + 'payload', payload, + 'created_at', created_at::text + )::text + from inserted; + """ + rows = psql_json(args, sql) + if not rows: + raise SystemExit("No attachment evaluation proposal inserted.") + return rows[0] + + +def record_document_evaluation(args: argparse.Namespace) -> dict[str, Any]: + try: + with open(args.payload_file, encoding="utf-8") as f: + payload = json.load(f) + except Exception as exc: + raise SystemExit(f"failed to read payload json: {exc}") from exc + + payload_json = json.dumps(payload, sort_keys=True) + file_ref_expr = "NULL::uuid" + if args.file_ref_id: + file_ref_expr = f"{sql_literal(args.file_ref_id)}::uuid" + + sql = f""" + insert into kb_stage.document_evaluations ( + file_ref_id, + channel, + source_ref, + evaluated_by_handle, + evaluated_by_agent_id, + decision, + rationale, + payload + ) + values ( + {file_ref_expr}, + coalesce(nullif({sql_literal(args.channel)}, ''), 'telegram'), + nullif({sql_literal(args.source_ref)}, ''), + nullif({sql_literal(args.evaluated_by)}, ''), + (select id from agents where handle = nullif({sql_literal(args.evaluated_by)}, '') limit 1), + {sql_literal(args.decision)}, + {sql_literal(args.rationale)}, + {sql_literal(payload_json)}::jsonb + ) + returning jsonb_build_object( + 'id', id::text, + 'file_ref_id', file_ref_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'evaluated_by_handle', evaluated_by_handle, + 'evaluated_by_agent_id', evaluated_by_agent_id::text, + 'decision', decision, + 'rationale', rationale, + 'payload', payload, + 'created_at', created_at::text + )::text; + """ + rows = psql_json(args, sql) + if not rows: + raise SystemExit("No document evaluation recorded.") + return rows[0] + +def list_proposals(args: argparse.Namespace) -> list[dict[str, Any]]: + where = "" + if args.status.lower() != "all": + where = f"where 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, 500), + 'payload', payload, + 'created_at', created_at::text, + 'reviewed_at', reviewed_at::text, + 'applied_at', applied_at::text + )::text + from kb_stage.kb_proposals + {where} + order by created_at desc + limit {args.limit}; + """ + return psql_json(args, sql) + + +def show_proposal(args: argparse.Namespace) -> dict[str, Any] | None: + sql = f""" + select jsonb_build_object( + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'proposed_by_agent_id', proposed_by_agent_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', rationale, + 'payload', payload, + 'reviewed_by_handle', reviewed_by_handle, + 'reviewed_at', reviewed_at::text, + 'review_note', review_note, + 'applied_by_handle', applied_by_handle, + 'applied_at', applied_at::text, + 'created_at', created_at::text, + 'updated_at', updated_at::text + )::text + from kb_stage.kb_proposals + where id = {sql_literal(args.proposal_id)}::uuid; + """ + rows = psql_json(args, sql) + return rows[0] if rows else None + + +def load_evidence(args: argparse.Namespace, claim_ids: list[str], limit: int) -> dict[str, list[dict[str, Any]]]: + if not claim_ids: + return {} + ids = sql_array(claim_ids, "uuid") + sql = f""" + with ranked as ( + select ce.claim_id, + ce.role::text as role, + ce.weight, + s.source_type, + s.url, + s.storage_path, + left(coalesce(s.excerpt, ''), 800) as excerpt, + row_number() over ( + partition by ce.claim_id + order by (s.storage_path like 'inbox/archive/%') desc, + ce.role::text, + s.storage_path nulls last, + s.url nulls last + ) as rn + from claim_evidence ce + join sources s on s.id = ce.source_id + where ce.claim_id = any({ids}) + ) + select jsonb_build_object( + 'claim_id', claim_id::text, + 'role', role, + 'weight', weight, + 'source_type', source_type, + 'url', url, + 'storage_path', storage_path, + 'excerpt', excerpt + )::text + from ranked + where rn <= {limit} + order by claim_id, rn; + """ + grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in psql_json(args, sql): + grouped[row["claim_id"]].append(row) + return grouped + + +def load_edges(args: argparse.Namespace, claim_ids: list[str], limit: int) -> dict[str, list[dict[str, Any]]]: + if not claim_ids: + return {} + ids = sql_array(claim_ids, "uuid") + sql = f""" + with base(id) as ( + select unnest({ids}) + ), + edge_rows as ( + select base.id as claim_id, + case when e.from_claim = base.id then 'outgoing' else 'incoming' end as direction, + e.edge_type::text as edge_type, + other.id as connected_id, + other.text as connected_text, + other.tags as connected_tags, + row_number() over ( + partition by base.id + order by case e.edge_type::text + when 'supports' then 1 + when 'challenges' then 2 + when 'contradicts' then 3 + when 'requires' then 4 + when 'supersedes' then 5 + when 'constrains' then 6 + when 'causes' then 7 + when 'accelerates' then 8 + else 9 + end, + other.text + ) as rn + from base + join claim_edges e on e.from_claim = base.id or e.to_claim = base.id + join claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end + ) + select jsonb_build_object( + 'claim_id', claim_id::text, + 'direction', direction, + 'edge_type', edge_type, + 'connected_id', connected_id::text, + 'connected_text', connected_text, + 'connected_tags', coalesce(to_jsonb(connected_tags), '[]'::jsonb) + )::text + from edge_rows + where rn <= {limit} + order by claim_id, rn; + """ + grouped: dict[str, list[dict[str, Any]]] = defaultdict(list) + for row in psql_json(args, sql): + grouped[row["claim_id"]].append(row) + return grouped + + +def bundle(args: argparse.Namespace, query: str, claim_limit: int, context_limit: int) -> dict[str, Any]: + claims = find_claims(args, query, claim_limit) + claim_ids = [claim["id"] for claim in claims] + evidence = load_evidence(args, claim_ids, 4) + edges = load_edges(args, claim_ids, 6) + return { + "query": query, + "terms": query_terms(query), + "claims": [ + { + **claim, + "evidence": evidence.get(claim["id"], []), + "edges": edges.get(claim["id"], []), + } + for claim in claims + ], + "context_rows": find_context_rows(args, query, context_limit), + } + + +def print_json(data: Any) -> None: + print(json.dumps(data, indent=2, sort_keys=True)) + + +def print_claim_bundle(data: dict[str, Any]) -> None: + print(f"# Teleo KB Context\n\nQuery: {data['query']}\n") + if data["context_rows"]: + print("## Soul / Context Rows\n") + for row in data["context_rows"]: + print( + f"- `{row['source']}` / `{row['owner']}` / `{row['title']}` " + f"(score {row['score']}): {truncate(row.get('body'), 320)}" + ) + print() + if not data["claims"]: + print("## Claims\n\nNo matching canonical claims found.") + return + print("## Claims\n") + for idx, claim in enumerate(data["claims"], start=1): + print(f"### {idx}. {claim['text']}\n") + print(f"- id: `{claim['id']}`") + print(f"- type: `{claim['type']}`; confidence: `{claim['confidence']}`; score: `{claim.get('score', '-')}`") + print(f"- tags: `{', '.join(claim.get('tags') or [])}`") + print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`") + print(f"- edge rows: `{claim.get('edge_count', len(claim.get('edges', [])))}`") + if claim.get("evidence"): + print("\nEvidence:") + for ev in claim["evidence"]: + source = ev.get("storage_path") or ev.get("url") or "(no source pointer)" + excerpt = f" - {truncate(ev.get('excerpt'), 260)}" if ev.get("excerpt") else "" + print(f"- `{ev['role']}` / `{ev.get('source_type')}` / `{source}`{excerpt}") + if claim.get("edges"): + print("\nEdges:") + for edge in claim["edges"]: + print( + f"- `{edge['direction']}` `{edge['edge_type']}` " + f"({edge['connected_id']}): {truncate(edge['connected_text'], 260)}" + ) + print() + + +def print_proposal(proposal: dict[str, Any]) -> None: + print(f"# KB Proposal {proposal['id']}\n") + print(f"- type: `{proposal['proposal_type']}`") + print(f"- status: `{proposal['status']}`") + print(f"- proposed_by: `{proposal.get('proposed_by_handle') or '-'}`") + print(f"- channel: `{proposal.get('channel') or '-'}`") + if proposal.get("source_ref"): + print(f"- source_ref: `{proposal['source_ref']}`") + print(f"- created_at: `{proposal.get('created_at')}`") + print("\n## Rationale\n") + print(proposal["rationale"]) + payload = proposal.get("payload") or {} + if proposal["proposal_type"] == "add_edge": + print("\n## Proposed Edge\n") + print(f"- from: `{payload.get('from_claim')}` - {truncate(payload.get('from_text'), 260)}") + print(f"- edge_type: `{payload.get('edge_type')}`") + print(f"- to: `{payload.get('to_claim')}` - {truncate(payload.get('to_text'), 260)}") + if payload.get("existing_edge_id"): + print(f"- existing_edge_id: `{payload['existing_edge_id']}`") + else: + print("- existing_edge_id: `none`") + else: + print("\n## Payload\n") + print(json.dumps(payload, indent=2, sort_keys=True)) + + +def print_proposal_list(rows: list[dict[str, Any]]) -> None: + print("# KB Proposals\n") + if not rows: + print("No matching proposals.") + return + for row in rows: + payload = row.get("payload") or {} + print(f"## {row['id']}\n") + print(f"- type: `{row['proposal_type']}`; status: `{row['status']}`") + print(f"- proposed_by: `{row.get('proposed_by_handle') or '-'}`; channel: `{row.get('channel') or '-'}`") + print(f"- created_at: `{row.get('created_at')}`") + if row["proposal_type"] == "add_edge": + print( + f"- edge: `{payload.get('from_claim')}` " + f"`{payload.get('edge_type')}` `{payload.get('to_claim')}`" + ) + print(f"- rationale: {truncate(row.get('rationale'), 320)}") + print() + + +def main() -> int: + args = parse_args() + if args.command == "search": + data = { + "query": args.query, + "terms": query_terms(args.query), + "claims": find_claims(args, args.query, args.limit), + "context_rows": find_context_rows(args, args.query, args.context_limit), + } + elif args.command == "context": + data = bundle(args, args.query, args.limit, args.context_limit) + elif args.command == "show": + claim = get_claim(args, args.claim_id) + if not claim: + raise SystemExit(f"claim not found: {args.claim_id}") + data = { + "claim": claim, + "evidence": load_evidence(args, [args.claim_id], 8).get(args.claim_id, []), + "edges": load_edges(args, [args.claim_id], 12).get(args.claim_id, []), + } + elif args.command == "evidence": + data = { + "claim_id": args.claim_id, + "evidence": load_evidence(args, [args.claim_id], args.limit).get(args.claim_id, []), + } + elif args.command == "edges": + data = { + "claim_id": args.claim_id, + "edges": load_edges(args, [args.claim_id], args.limit).get(args.claim_id, []), + } + elif args.command == "propose-edge": + data = propose_edge(args) + elif args.command == "propose-attachment-evaluation": + data = propose_attachment_evaluation(args) + elif args.command == "record-document-evaluation": + data = record_document_evaluation(args) + elif args.command == "list-proposals": + data = {"proposals": list_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 + else: + raise AssertionError(args.command) + + if args.format == "json": + print_json(data) + elif args.command in {"context", "search"}: + print_claim_bundle(data) + elif args.command == "show": + print_claim_bundle( + { + "query": f"claim {args.claim_id}", + "claims": [{**data["claim"], "evidence": data["evidence"], "edges": data["edges"]}], + "context_rows": [], + } + ) + elif args.command in {"propose-edge", "propose-attachment-evaluation"}: + print_proposal(data) + elif args.command == "record-document-evaluation": + print_json(data) + elif args.command == "list-proposals": + print_proposal_list(data["proposals"]) + elif args.command == "show-proposal": + print_proposal(data) + else: + print_json(data) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/hermes-agent/leoclean-bin/teleo-kb b/hermes-agent/leoclean-bin/teleo-kb new file mode 100755 index 0000000..1c0b1d1 --- /dev/null +++ b/hermes-agent/leoclean-bin/teleo-kb @@ -0,0 +1,64 @@ +#!/usr/bin/env bash +set -euo pipefail + +PROFILE_DIR=/home/teleo/.hermes/profiles/leoclean +KB_TOOL="$PROFILE_DIR/bin/kb_tool.py" +CLOUDSQL_TOOL="$PROFILE_DIR/bin/cloudsql_memory_tool.py" +MODE=${TELEO_KB_MODE:-auto} +COMMAND=${1:-} +CLOUDSQL_STATUS_TIMEOUT=${TELEO_CLOUDSQL_STATUS_TIMEOUT_SECONDS:-30} + +cloudsql_supported() { + case "$COMMAND" in + status|search|context|show|evidence|edges|propose-core-change|list-proposals|show-proposal) return 0 ;; + *) return 1 ;; + esac +} + +cloudsql_args() { + if [ "${TELEO_MEMORY_INCLUDE_EXCERPTS:-}" = "1" ]; then + case "$COMMAND" in + search|context|show|evidence|edges) printf '%s\n' "--include-excerpts" ;; + esac + fi +} + +run_cloudsql() { + if cloudsql_supported; then + extra_args=() + while IFS= read -r arg; do + [ -n "$arg" ] && extra_args+=("$arg") + done < <(cloudsql_args) + exec python3 "$CLOUDSQL_TOOL" "$@" "${extra_args[@]}" + fi + exec python3 "$KB_TOOL" "$@" +} + +case "$MODE" in + cloudsql) + run_cloudsql "$@" + ;; + local) + exec python3 "$KB_TOOL" --local "$@" + ;; + remote) + exec python3 "$KB_TOOL" "$@" + ;; + auto) + if [ -x "$CLOUDSQL_TOOL" ] && command -v psql >/dev/null 2>&1 && command -v gcloud >/dev/null 2>&1; then + if cloudsql_supported; then + if timeout "${CLOUDSQL_STATUS_TIMEOUT}s" python3 "$CLOUDSQL_TOOL" status --format json --redacted >/dev/null 2>&1; then + run_cloudsql "$@" + fi + fi + fi + if command -v docker >/dev/null 2>&1; then + exec python3 "$KB_TOOL" --local "$@" + fi + exec python3 "$KB_TOOL" "$@" + ;; + *) + echo "Unsupported TELEO_KB_MODE=$MODE; expected auto, cloudsql, local, or remote" >&2 + exit 64 + ;; +esac diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py new file mode 100644 index 0000000..31ae30f --- /dev/null +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -0,0 +1,63 @@ +"""Source-control checks for the Hermes leoclean KB bridge files.""" + +from __future__ import annotations + +import ast +import re +import subprocess +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +BRIDGE_DIR = ROOT / "hermes-agent" / "leoclean-bin" + + +def test_leoclean_bridge_files_are_present_and_parseable() -> None: + cloudsql_tool = BRIDGE_DIR / "cloudsql_memory_tool.py" + vps_tool = BRIDGE_DIR / "kb_tool.py" + wrapper = BRIDGE_DIR / "teleo-kb" + + assert cloudsql_tool.exists() + assert vps_tool.exists() + assert wrapper.exists() + + ast.parse(cloudsql_tool.read_text()) + ast.parse(vps_tool.read_text()) + subprocess.run(["bash", "-n", str(wrapper)], check=True) + + +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() + + for command in [ + "status", + "search", + "context", + "show", + "evidence", + "edges", + "propose-core-change", + "list-proposals", + "show-proposal", + ]: + assert command in wrapper_text + assert command in cloudsql_text + + assert "propose-edge" not in wrapper_text + assert "kb_stage.kb_proposals" in cloudsql_text + assert "canonical apply" in cloudsql_text.lower() + + +def test_bridge_source_does_not_commit_raw_secret_values() -> None: + combined = "\n".join(path.read_text(errors="replace") for path in BRIDGE_DIR.iterdir() if path.is_file()) + forbidden_patterns = [ + r"bot\d+:[A-Za-z0-9_-]{20,}", + r"postgres://[^\\s]+:[^\\s]+@", + r"-----BEGIN [A-Z ]*PRIVATE KEY-----", + r"sk-[A-Za-z0-9]{20,}", + r"xox[baprs]-[A-Za-z0-9-]{20,}", + r"AIza[0-9A-Za-z_-]{20,}", + ] + for pattern in forbidden_patterns: + assert not re.search(pattern, combined), pattern +