Source-control leoclean KB bridge (#59)

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())

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

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
+