teleo-infrastructure/hermes-agent/leoclean-bin/kb_tool.py
2026-07-13 12:20:30 +02:00

1412 lines
52 KiB
Python
Executable file

#!/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 os
import re
import subprocess
from collections import defaultdict
from typing import Any
DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz"
WORKER_SUPPORTED_PROPOSAL_TYPES = frozenset({"revise_strategy", "add_edge", "attach_evidence", "approve_claim"})
CURRENT_PUBLIC_SCHEMA = {
"behavioral_rules": ("id", "agent_id", "category", "rank", "rule", "rationale", "created_at"),
"beliefs": (
"id",
"agent_id",
"level",
"statement",
"confidence",
"falsifier",
"rank",
"status",
"created_at",
"updated_at",
),
"claim_edges": ("id", "from_claim", "to_claim", "edge_type", "weight", "created_by", "created_at"),
"claims": (
"id",
"type",
"text",
"status",
"confidence",
"tags",
"created_by",
"superseded_by",
"created_at",
"updated_at",
),
"governance_gates": (
"id",
"agent_id",
"name",
"criteria",
"evidence_bar",
"pass_condition",
"rank",
"created_at",
),
"reasoning_tools": ("id", "agent_id", "name", "description", "category", "created_at"),
"sources": (
"id",
"source_type",
"url",
"storage_path",
"excerpt",
"hash",
"captured_at",
"created_by",
"created_at",
),
}
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.",
)
status = sub.add_parser("status", help="Read exact canonical and proposal table counts.")
add_output_flag(status)
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)
search_proposals = sub.add_parser("search-proposals", help="Search KB mutation proposals across statuses.")
search_proposals.add_argument("query")
search_proposals.add_argument("--status", default="all")
search_proposals.add_argument("--limit", type=int, default=10)
add_output_flag(search_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)
decision_matrix_status = sub.add_parser(
"decision-matrix-status",
help="Read whether decision-matrix approval tables exist before claiming matrix approval.",
)
add_output_flag(decision_matrix_status)
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 classify_proposal_readiness(proposal: dict[str, Any]) -> dict[str, Any]:
"""Classify contract readiness without claiming the production worker is enabled."""
payload = proposal.get("payload")
if not isinstance(payload, dict):
payload = {}
apply_payload = payload.get("apply_payload")
has_apply_payload = isinstance(apply_payload, dict)
status = proposal.get("status")
proposal_type = proposal.get("proposal_type")
worker_supported_type = proposal_type in WORKER_SUPPORTED_PROPOSAL_TYPES
worker_contract_applyable = bool(status == "approved" and worker_supported_type and has_apply_payload)
if status == "applied":
review_state = "applied"
elif status == "pending_review":
review_state = "needs_human_review"
elif not worker_supported_type:
review_state = "unsupported_by_apply_worker_contract"
elif status == "approved" and worker_contract_applyable:
review_state = "approved_contract_present"
elif status == "approved" and not has_apply_payload:
review_state = "approved_needs_apply_payload"
else:
review_state = "not_ready"
return {
"review_state": review_state,
"has_apply_payload": has_apply_payload,
"worker_supported_type": worker_supported_type,
"worker_contract_applyable": worker_contract_applyable,
"production_worker_enabled": None,
"guidance": (
"This is proposal-contract readiness only. It does not prove the production apply worker is enabled, "
"the payload passes strict validation, or apply is authorized."
),
}
def with_proposal_readiness(proposal: dict[str, Any]) -> dict[str, Any]:
return {**proposal, "readiness": classify_proposal_readiness(proposal)}
def operational_contracts(
query: str, *, current_schema: dict[str, list[str] | tuple[str, ...]] | None = None
) -> list[dict[str, Any]]:
"""Return question-specific current-runtime truth, not recalled architecture."""
lowered = query.lower()
schema = current_schema if current_schema is not None else CURRENT_PUBLIC_SCHEMA
contracts: list[dict[str, Any]] = [
{
"id": "reply_budget",
"target_words": 170,
"hard_max_words": 220,
"shape": "direct answer; at most three short bullets; at most one proof-changing action",
}
]
source_terms = ("document", "pdf", "tweet", "attachment", "ingest", "intake", "absorb", "extract")
if any(term in lowered for term in source_terms):
contracts.append(
{
"id": "source_intake",
"required_lead": "Local compiler proven; live VPS/chat intake not shipped.",
"capability_tier": "build-local compiler only",
"safe_without_apply_authorization": [
"retain artifact plus URL or storage_path plus content hash",
"extract candidates into a pending_review proposal packet",
],
"approval_begins": "before guarded canonical public.* apply, not before mechanical capture/staging",
"current_public_sources_columns": list(schema.get("sources", ())),
"must_not_claim": [
"live VPS Leo can invoke the compiler from chat",
"a chat label, filename, attachment ref, or source_ref is source identity",
"author/title/publisher/publication date are current public.sources columns",
],
}
)
mixed_markers = (
"packet",
"factual observation",
"strategic framework",
"disputed interpretation",
"governance rule",
"old belief",
"compose the database",
)
if sum(marker in lowered for marker in mixed_markers) >= 2:
contracts.append(
{
"id": "mixed_packet_composition",
"map": {
"facts_and_disputes": "claims plus sources and evidence",
"reusable_framework": "reasoning_tools",
"operating_rule": "behavioral_rules",
"evaluative_gate": "governance_gates",
"agent_position": "beliefs",
},
"approve_claim_supported": ["claims", "sources", "evidence", "edges", "reasoning_tools"],
"keep_staged_until_separate_capability": [
"behavioral_rules",
"governance_gates",
"belief updates",
"existing-row updates",
],
"schema_guards": [
"claim_edges endpoints are claim IDs; never point an edge at a reasoning_tool or belief row",
"reasoning_tools has no scope field",
"claims has no falsifier field",
"behavioral_rules has no status field",
"applied_at belongs to the proposal receipt, not each row",
],
"current_columns": {
name: list(schema.get(name, ()))
for name in (
"claims",
"claim_edges",
"reasoning_tools",
"behavioral_rules",
"governance_gates",
"beliefs",
)
},
}
)
if any(term in lowered for term in ("restart", "prior session", "erased", "answer behavior", "database totals")):
contracts.append(
{
"id": "runtime_persistence",
"count_boundary": "unchanged totals do not prove unchanged rows or unchanged answers",
"persisted_inputs": [
"Postgres rows",
"skills",
"runtime config",
"SOUL.md",
"state.db",
"session JSONL",
],
"proof": "use row IDs/timestamps/hashes; separate handler, Telegram-visible delivery, and canonical mutation",
}
)
if any(
term in lowered
for term in ("two agents", "agent-specific", "different conclusions", "duplicate the factual claim")
):
contracts.append(
{
"id": "shared_claims_agent_positions",
"fact_storage": "one shared claim plus shared sources/evidence",
"position_storage": "one public.beliefs row per agent position",
"schema_gap": "beliefs has no claim-ID foreign key and there is no belief-edge table",
"edge_boundary": "claim_edges connects claims only",
}
)
if any(term in lowered for term in ("reviewer approval", "approved", "partner demo", "database updated")):
contracts.append(
{
"id": "proposal_apply_readiness",
"boundary": "reviewer approval is intent, not canonical apply",
"required_readback": ["readiness", "applied_at", "row-level postflight"],
"legacy_rule": "approved_needs_apply_payload requires strict normalization plus renewed review; do not apply the legacy row",
}
)
return contracts
def load_current_public_schema(args: argparse.Namespace, table_names: set[str]) -> dict[str, list[str]]:
if not table_names:
return {}
sql = f"""
select jsonb_build_object(
'table_name', table_name,
'columns', jsonb_agg(column_name order by ordinal_position)
)::text
from information_schema.columns
where table_schema = 'public'
and table_name = any({sql_array(sorted(table_names))})
group by table_name
order by table_name;
"""
return {row["table_name"]: row["columns"] for row in psql_json(args, sql)}
def context_operational_contracts(args: argparse.Namespace, query: str) -> list[dict[str, Any]]:
initial = operational_contracts(query)
contract_ids = {contract["id"] for contract in initial}
table_names: set[str] = set()
if "source_intake" in contract_ids:
table_names.add("sources")
if "mixed_packet_composition" in contract_ids:
table_names.update(
{"claims", "claim_edges", "reasoning_tools", "behavioral_rules", "governance_gates", "beliefs"}
)
if not table_names:
return initial
return operational_contracts(query, current_schema=load_current_public_schema(args, table_names))
def claim_base_url() -> str:
return os.environ.get("TELEO_KB_CLAIM_BASE_URL", DEFAULT_CLAIM_BASE_URL).rstrip("/")
def claim_url(claim_id: str | None) -> str | None:
if not claim_id:
return None
return f"{claim_base_url()}/kb/claims/{claim_id}"
def markdown_claim_link(claim_id: str | None, label: str | None = None) -> str:
if not claim_id:
return "`unknown`"
text = " ".join(str(label or claim_id).replace("`", "'").split())
return f"[`{text}`]({claim_url(claim_id)})"
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),
'apply_payload', jsonb_build_object(
'from_claim', from_row.id::text,
'to_claim', to_row.id::text,
'edge_type', p.edge_type::text
)
)
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 [with_proposal_readiness(row) for row in psql_json(args, sql)]
def search_proposals(args: argparse.Namespace) -> dict[str, Any]:
terms = query_terms(args.query)
patterns = sql_array([f"%{term}%" for term in terms], "text")
status_clause = ""
if args.status.lower() != "all":
status_clause = f"and 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, 700),
'payload', payload,
'created_at', created_at::text,
'reviewed_at', reviewed_at::text,
'applied_at', applied_at::text
)::text
from kb_stage.kb_proposals
where (
coalesce(rationale, '') ilike any({patterns})
or coalesce(source_ref, '') ilike any({patterns})
or coalesce(proposal_type, '') ilike any({patterns})
or coalesce(proposed_by_handle, '') ilike any({patterns})
or payload::text ilike any({patterns})
)
{status_clause}
order by
case status
when 'approved' then 0
when 'pending_review' then 1
when 'applied' then 2
else 3
end,
created_at desc
limit {max(1, min(args.limit, 100))};
"""
return {
"query": args.query,
"terms": terms,
"status_filter": args.status,
"proposals": [with_proposal_readiness(row) for row in 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 with_proposal_readiness(rows[0]) if rows else None
def decision_matrix_status(args: argparse.Namespace) -> dict[str, Any]:
sql = """
with table_status(schema_name, table_name, exists_on_vps) as (
values
('public', 'matrix_voters', to_regclass('public.matrix_voters') is not null),
('public', 'proposal_votes', to_regclass('public.proposal_votes') is not null),
('public', 'proposal_decisions', to_regclass('public.proposal_decisions') is not null),
('kb_stage', 'matrix_voters', to_regclass('kb_stage.matrix_voters') is not null),
('kb_stage', 'proposal_votes', to_regclass('kb_stage.proposal_votes') is not null),
('kb_stage', 'proposal_decisions', to_regclass('kb_stage.proposal_decisions') is not null)
),
status_counts as (
select coalesce(jsonb_object_agg(status, proposal_count order by status), '{}'::jsonb) as counts
from (
select status, count(*) as proposal_count
from kb_stage.kb_proposals
group by status
) grouped
)
select jsonb_build_object(
'artifact', 'teleo_kb_decision_matrix_status',
'decision_matrix_tables', (
select jsonb_object_agg(schema_name || '.' || table_name, exists_on_vps order by schema_name, table_name)
from table_status
),
'all_required_tables_present', (
select bool_and(exists_on_vps) from table_status
),
'any_decision_matrix_table_present', (
select bool_or(exists_on_vps) from table_status
),
'proposal_status_counts', (select counts from status_counts),
'guidance', case
when (select bool_and(exists_on_vps) from table_status)
then 'Decision-matrix tables exist; inspect proposal_decisions/proposal_votes for the specific proposal before claiming approval.'
else 'Decision-matrix approval schema is absent or incomplete; do not infer matrix approval from kb_stage proposal rationale/status alone.'
end
)::text;
"""
rows = psql_json(args, sql)
if not rows:
raise SystemExit("No decision matrix status returned.")
return rows[0]
def status(args: argparse.Namespace) -> dict[str, Any]:
sql = """
with status_counts as (
select coalesce(jsonb_object_agg(status, proposal_count order by status), '{}'::jsonb) as counts
from (
select status, count(*) as proposal_count
from kb_stage.kb_proposals
group by status
) grouped
)
select jsonb_build_object(
'artifact', 'teleo_kb_status',
'backend', current_database() || '|' || current_user,
'high_signal_rows', jsonb_build_object(
'claims', (select count(*) from public.claims),
'sources', (select count(*) from public.sources),
'claim_edges', (select count(*) from public.claim_edges),
'claim_evidence', (select count(*) from public.claim_evidence),
'kb_proposals', (select count(*) from kb_stage.kb_proposals)
),
'proposal_status_counts', (select counts from status_counts)
)::text;
"""
rows = psql_json(args, sql)
if not rows:
raise SystemExit("No KB status returned.")
return rows[0]
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),
"operational_contracts": context_operational_contracts(args, query),
"claims": [
{
**claim,
"claim_url": claim_url(claim["id"]),
"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_status(data: dict[str, Any]) -> None:
counts = data["high_signal_rows"]
print("# Teleo VPS KB Status\n")
print(f"- backend: `{data['backend']}`")
print(f"- high-signal rows: `{json.dumps(counts, sort_keys=True)}`")
print(f"- proposal status counts: `{json.dumps(data.get('proposal_status_counts') or {}, sort_keys=True)}`\n")
print(
f"DB readback: claims: `{counts['claims']}`; sources: `{counts['sources']}`; "
f"claim_edges: `{counts['claim_edges']}`; claim_evidence: `{counts['claim_evidence']}`; "
f"kb_proposals: `{counts['kb_proposals']}`."
)
def print_claim_bundle(data: dict[str, Any]) -> None:
print(f"# Teleo KB Context\n\nQuery: {data['query']}\n")
contracts = data.get("operational_contracts") or []
if contracts:
print("## Current Runtime Contracts\n")
for contract in contracts:
print(f"- `{contract['id']}`: `{json.dumps(contract, sort_keys=True)}`")
print()
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}. {markdown_claim_link(claim['id'], truncate(claim.get('text'), 140))}\n")
print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}")
print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}")
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"]:
connected = edge["connected_id"]
connected_label = truncate(edge.get("connected_text") or connected, 120)
print(
f"- `{edge['direction']}` `{edge['edge_type']}` {markdown_claim_link(connected, connected_label)}"
)
print()
def print_proposal(proposal: dict[str, Any]) -> None:
readiness = proposal.get("readiness") or classify_proposal_readiness(proposal)
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(f"- reviewed_at: `{proposal.get('reviewed_at') or '-'}`")
print(f"- applied_at: `{proposal.get('applied_at') or '-'}`")
print(f"- review_state: `{readiness['review_state']}`")
print(f"- has_apply_payload: `{readiness['has_apply_payload']}`")
print(f"- worker_contract_applyable: `{readiness['worker_contract_applyable']}`")
print("- production_worker_enabled: `not proven by this readback`")
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: {markdown_claim_link(payload.get('from_claim'), truncate(payload.get('from_text'), 120))}")
print(f"- edge_type: `{payload.get('edge_type')}`")
print(f"- to: {markdown_claim_link(payload.get('to_claim'), truncate(payload.get('to_text'), 120))}")
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 {}
readiness = row.get("readiness") or classify_proposal_readiness(row)
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')}`")
print(f"- reviewed_at: `{row.get('reviewed_at') or '-'}`")
print(f"- applied_at: `{row.get('applied_at') or '-'}`")
print(
f"- readiness: `{readiness['review_state']}`; "
f"has_apply_payload: `{readiness['has_apply_payload']}`; "
f"worker_contract_applyable: `{readiness['worker_contract_applyable']}`"
)
if row["proposal_type"] == "add_edge":
print(
f"- edge: {markdown_claim_link(payload.get('from_claim'), truncate(payload.get('from_text'), 80))} "
f"`{payload.get('edge_type')}` {markdown_claim_link(payload.get('to_claim'), truncate(payload.get('to_text'), 80))}"
)
print(f"- rationale: {truncate(row.get('rationale'), 320)}")
print()
def print_proposal_search(data: dict[str, Any]) -> None:
print("# KB Proposal Search\n")
print(f"- query: `{data['query']}`")
print(f"- terms: `{', '.join(data['terms'])}`")
print(f"- status filter: `{data['status_filter']}`")
print()
print_proposal_list(data["proposals"])
def print_decision_matrix_status(data: dict[str, Any]) -> None:
print("# Decision Matrix Status\n")
print(f"- all required tables present: `{data['all_required_tables_present']}`")
print(f"- any decision-matrix table present: `{data['any_decision_matrix_table_present']}`")
print(f"- proposal status counts: `{json.dumps(data.get('proposal_status_counts') or {}, sort_keys=True)}`")
print(f"- guidance: {data['guidance']}")
print("\n## Tables\n")
for table, exists in sorted((data.get("decision_matrix_tables") or {}).items()):
print(f"- `{table}`: `{exists}`")
print()
def main() -> int:
args = parse_args()
if args.command == "status":
data = status(args)
elif args.command == "search":
data = {
"query": args.query,
"terms": query_terms(args.query),
"operational_contracts": context_operational_contracts(args, 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 == "search-proposals":
data = search_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
elif args.command == "decision-matrix-status":
data = decision_matrix_status(args)
else:
raise AssertionError(args.command)
if args.format == "json":
print_json(data)
elif args.command == "status":
print_status(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 == "search-proposals":
print_proposal_search(data)
elif args.command == "show-proposal":
print_proposal(data)
elif args.command == "decision-matrix-status":
print_decision_matrix_status(data)
else:
print_json(data)
return 0
if __name__ == "__main__":
raise SystemExit(main())