2708 lines
116 KiB
Python
Executable file
2708 lines
116 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 hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import uuid
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz"
|
|
SOURCE_COMPILER_PATH = Path("/opt/teleo-eval/workspaces/deploy-infra/scripts/compile_kb_source_packet.py")
|
|
SOURCE_PROPOSAL_RECEIPT_SCHEMA = "livingip.teleoKbSourceProposalReceipt.v1"
|
|
SOURCE_COMPILER_BUNDLE_SCHEMA = "livingip.kbSourceProposalBundle.v1"
|
|
CANONICAL_COUNT_LABELS = ("claims", "sources", "claim_edges", "claim_evidence")
|
|
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
|
|
WORKER_SUPPORTED_PROPOSAL_TYPES = frozenset({"revise_strategy", "add_edge", "attach_evidence", "approve_claim"})
|
|
DIRECT_READBACK_CONTRACTS = frozenset(
|
|
{
|
|
"proposal_state_readback",
|
|
"named_packet_readback",
|
|
"decision_matrix_readback",
|
|
"source_link_audit",
|
|
"demo_capability_readback",
|
|
"identity_canonicality_readback",
|
|
}
|
|
)
|
|
SCHEMA_COMPILED_CONTRACTS = frozenset({"forecast_resolution_schema", "claim_supersession_schema"})
|
|
DECISION_MATRIX_TABLES = tuple(
|
|
f"{schema}.{table}"
|
|
for schema in ("public", "kb_stage")
|
|
for table in ("matrix_voters", "proposal_votes", "proposal_decisions")
|
|
)
|
|
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)
|
|
|
|
propose_source = sub.add_parser(
|
|
"propose-source",
|
|
help="Compile a hash-bound source artifact and stage one idempotent pending-review proposal.",
|
|
)
|
|
propose_source.add_argument("--artifact", required=True, type=Path)
|
|
propose_source.add_argument("--text", required=True, type=Path)
|
|
propose_source.add_argument("--manifest", required=True, type=Path)
|
|
propose_source.add_argument("--receipt", required=True, type=Path)
|
|
add_output_flag(propose_source)
|
|
|
|
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 _has_unnegated_action(query: str, actions: tuple[str, ...]) -> bool:
|
|
for segment in re.split(r"(?<=[.!?])\s+|\n+", query.lower()):
|
|
if not any(re.search(rf"\b{re.escape(action)}\b", segment) for action in actions):
|
|
continue
|
|
if "read-only" in segment or re.search(r"\b(?:do not|don't|dont|must not|without)\b", segment):
|
|
continue
|
|
return True
|
|
return False
|
|
|
|
|
|
def operational_contracts(
|
|
query: str,
|
|
*,
|
|
current_schema: dict[str, list[str] | tuple[str, ...]] | None = None,
|
|
current_constraints: dict[str, dict[str, 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
|
|
constraints = current_constraints or {}
|
|
contracts: list[dict[str, Any]] = [
|
|
{
|
|
"id": "reply_budget",
|
|
"target_words": 150,
|
|
"hard_max_words": 200,
|
|
"shape": (
|
|
"no intro; at most three compact bullets covering mapping, review/apply, and receipt/limitation; "
|
|
"omission is better than exceeding the budget"
|
|
),
|
|
}
|
|
]
|
|
|
|
decision_matrix_question = "decision matrix" in lowered or "decision-matrix" in lowered
|
|
source_link_audit_question = (
|
|
any(term in lowered for term in ("proposal", "pending", "approved", "stuck", "backlog"))
|
|
and any(
|
|
term in lowered
|
|
for term in (
|
|
"document",
|
|
"attachment",
|
|
"extracted text",
|
|
"source row",
|
|
"source_ref",
|
|
"source ref",
|
|
"pointer",
|
|
"pointed",
|
|
"evidence link",
|
|
)
|
|
)
|
|
and any(
|
|
term in lowered
|
|
for term in ("because", "wrong", "missing", "stuck", "link", "point", "cause", "canonical evidence", "audit")
|
|
)
|
|
)
|
|
kb_scope = (
|
|
"knowledge base" in lowered
|
|
or "database" in lowered
|
|
or bool(re.search(r"\bkb\b", lowered))
|
|
or "canonical knowledge" in lowered
|
|
)
|
|
kb_implementation_question = bool(
|
|
re.search(
|
|
r"\b(?:scorer|benchmark|test harness|plugin|parser|implementation|search|dashboard|api|script|"
|
|
r"function|code|tool)\b",
|
|
lowered,
|
|
)
|
|
)
|
|
forecast_resolution_question = "forecast" in lowered and any(
|
|
term in lowered for term in ("resolution", "criteria", "event is now over", "event is over", "60%")
|
|
)
|
|
proposal_lifecycle_question = any(
|
|
term in lowered
|
|
for term in ("proposal", "proposed", "pending", "approved", "applied", "staged", "reviewer approval")
|
|
)
|
|
broad_kb_change_question = _has_unnegated_action(
|
|
query, ("change", "changed", "update", "updated", "landed")
|
|
)
|
|
proposal_state_question = kb_scope and not forecast_resolution_question and (
|
|
proposal_lifecycle_question
|
|
or ("demo" not in lowered and broad_kb_change_question and not kb_implementation_question)
|
|
)
|
|
named_packet_question = "helmer" in lowered and any(
|
|
term in lowered for term in ("7 powers", "seven powers", "powers framework", "powers packet")
|
|
)
|
|
demo_question = "demo" in lowered and (
|
|
any(
|
|
term in lowered
|
|
for term in (
|
|
"change the kb",
|
|
"changes the kb",
|
|
"changed the kb",
|
|
"change the knowledge base",
|
|
"changes the knowledge base",
|
|
"changed the knowledge base",
|
|
"change the database",
|
|
"changes the database",
|
|
"changed the database",
|
|
)
|
|
)
|
|
or (
|
|
kb_scope
|
|
and (
|
|
any(term in lowered for term in ("learned", "updated", "update", "live", "current", "working"))
|
|
or _has_unnegated_action(query, ("write", "apply", "mutate", "stage"))
|
|
)
|
|
)
|
|
)
|
|
identity_question = any(term in lowered for term in ("soul.md", "soul file")) and any(
|
|
term in lowered for term in ("canonical identity", "canonical", "identity", "source of truth", "database")
|
|
)
|
|
visible_sender_match = re.search(
|
|
r"(?:current visible telegram sender|visible telegram sender|current telegram sender)\s+is\s+@?([a-z0-9_]+)",
|
|
query,
|
|
re.I,
|
|
)
|
|
participant_identity_question = bool(
|
|
visible_sender_match
|
|
and any(
|
|
term in lowered
|
|
for term in ("call this participant", "identity sources", "mixing identities", "mix identities")
|
|
)
|
|
)
|
|
primary_direct_intent = next(
|
|
(
|
|
intent
|
|
for intent, enabled in (
|
|
("decision_matrix_readback", decision_matrix_question),
|
|
("source_link_audit", source_link_audit_question),
|
|
("named_packet_readback", named_packet_question),
|
|
("identity_canonicality_readback", identity_question),
|
|
("proposal_state_readback", proposal_state_question),
|
|
("demo_capability_readback", demo_question),
|
|
)
|
|
if enabled
|
|
),
|
|
None,
|
|
)
|
|
source_terms = ("document", "pdf", "tweet", "attachment", "ingest", "intake", "absorb", "extract")
|
|
broad_source_object = any(term in lowered for term in ("report", "article", "file", "url", "link"))
|
|
broad_source_action = any(
|
|
term in lowered
|
|
for term in (
|
|
"send",
|
|
"hand",
|
|
"drop",
|
|
"upload",
|
|
"learn",
|
|
"absorb",
|
|
"ingest",
|
|
"extract",
|
|
"stage",
|
|
"add",
|
|
"make it part",
|
|
)
|
|
)
|
|
source_intake_question = any(term in lowered for term in source_terms) or (
|
|
broad_source_object and broad_source_action
|
|
)
|
|
if source_intake_question and primary_direct_intent is None:
|
|
contracts.append(
|
|
{
|
|
"id": "source_intake",
|
|
"required_lead": (
|
|
"VPS source-to-proposal staging is shipped for prepared inputs; autonomous chat attachment "
|
|
"extraction is not shipped."
|
|
),
|
|
"capability_tier": "VPS-local prepared-input compiler and pending-review staging",
|
|
"safe_without_apply_authorization": [
|
|
"prepare and retain the artifact, UTF-8 extracted text, extraction manifest, URL or storage_path "
|
|
"locator, and content hash",
|
|
"run teleo-kb propose-source to validate and stage a pending_review proposal",
|
|
],
|
|
"approval_begins": "before guarded canonical public.* apply, not before mechanical capture/staging",
|
|
"current_public_sources_columns": list(schema.get("sources", ())),
|
|
"must_not_claim": [
|
|
"Leo automatically captures or extracts arbitrary Telegram attachments or tweets",
|
|
"a chat label, filename, attachment ref, or source_ref is source identity",
|
|
"author/title/publisher/publication date are current public.sources columns",
|
|
],
|
|
}
|
|
)
|
|
|
|
if primary_direct_intent == "source_link_audit":
|
|
contracts.append(
|
|
{
|
|
"id": "source_link_audit",
|
|
"causality_boundary": (
|
|
"pending status alone does not prove a document-pointer failure; audit each staging and "
|
|
"canonical link before naming the cause"
|
|
),
|
|
"layers": [
|
|
"kb_stage.telegram_file_refs or retained raw files",
|
|
"kb_stage.document_evaluations",
|
|
"kb_stage.kb_proposals.source_ref",
|
|
"public.sources",
|
|
"public.claim_evidence",
|
|
],
|
|
"next_proof": (
|
|
"list proposal UUID/status/source_ref, test exact public.sources matches, inspect evidence links, "
|
|
"then stage only the missing guarded source/evidence contract"
|
|
),
|
|
}
|
|
)
|
|
|
|
if primary_direct_intent == "proposal_state_readback":
|
|
contracts.append(
|
|
{
|
|
"id": "proposal_state_readback",
|
|
"approval_only_claim": "reviewer approval" in lowered and broad_kb_change_question,
|
|
"state_boundary": (
|
|
"proposal status, reviewer approval, and canonical apply are separate states; applied rows still "
|
|
"need proposal-level applied_at plus row-level postflight for a specific change claim"
|
|
),
|
|
}
|
|
)
|
|
|
|
if primary_direct_intent == "named_packet_readback":
|
|
contracts.append(
|
|
{
|
|
"id": "named_packet_readback",
|
|
"search_query": "Helmer 7 Powers",
|
|
"canonical_boundary": (
|
|
"packet existence or reviewer approval does not mean the framework is present in canonical "
|
|
"reasoning_tools, claims, sources, and evidence"
|
|
),
|
|
}
|
|
)
|
|
|
|
if primary_direct_intent == "decision_matrix_readback":
|
|
contracts.append(
|
|
{
|
|
"id": "decision_matrix_readback",
|
|
"approval_boundary": (
|
|
"kb_stage reviewer status is not a matrix vote; matrix approval requires the matrix schema plus "
|
|
"proposal-specific decision/vote rows"
|
|
),
|
|
}
|
|
)
|
|
|
|
if primary_direct_intent == "demo_capability_readback":
|
|
contracts.append(
|
|
{
|
|
"id": "demo_capability_readback",
|
|
"temporal_comparison_requested": any(
|
|
term in lowered for term in ("since yesterday", "since last night", "since earlier")
|
|
),
|
|
"tiers": [
|
|
"read existing canonical and proposal receipts",
|
|
"stage a safe pending_review proposal",
|
|
"run a guarded canonical apply only with explicit authorization and postflight proof",
|
|
],
|
|
"claim_ceiling": (
|
|
"do not present a prepared or clone-proven rich packet as a production canonical mutation"
|
|
),
|
|
}
|
|
)
|
|
|
|
if primary_direct_intent == "identity_canonicality_readback":
|
|
contracts.append(
|
|
{
|
|
"id": "identity_canonicality_readback",
|
|
"boundary": (
|
|
"a direct SOUL.md edit changes the runtime/profile artifact, not canonical Postgres identity rows"
|
|
),
|
|
"canonical_surfaces": ["personas", "strategies", "beliefs", "strategy_nodes", "strategy_node_anchors"],
|
|
"proof_chain": [
|
|
"read canonical identity rows",
|
|
"verify a reviewed and applied proposal for the intended change",
|
|
"run or verify DB-to-runtime render/sync",
|
|
"compare the resulting SOUL.md hash/content and restart readback",
|
|
],
|
|
"renderer_ceiling": "no active general DB-to-SOUL.md renderer automation is currently proven",
|
|
}
|
|
)
|
|
|
|
if participant_identity_question and visible_sender_match:
|
|
contracts.append(
|
|
{
|
|
"id": "telegram_participant_identity",
|
|
"current_visible_handle": visible_sender_match.group(1),
|
|
"allowed_sources": [
|
|
"the current message sender handle",
|
|
"that participant's explicit in-chat correction",
|
|
],
|
|
"disallowed_sources": [
|
|
"stale session context",
|
|
"environment labels",
|
|
"inferred personal names",
|
|
"another participant's profile or messages",
|
|
],
|
|
"cross_participant_boundary": (
|
|
"re-anchor every reply to that message's visible sender; never transfer or mix identities"
|
|
),
|
|
}
|
|
)
|
|
|
|
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",
|
|
},
|
|
"table_semantics": {
|
|
"behavioral_rules": (
|
|
"binding operating/governance rules; use rule, rationale, category, rank, and agent_id"
|
|
),
|
|
"governance_gates": (
|
|
"evaluative pass/fail gates only; use criteria, evidence_bar, and pass_condition"
|
|
),
|
|
"prompt_governance_rule": (
|
|
"map to behavioral_rules unless it actually defines evaluation criteria, an evidence bar, "
|
|
"and a pass condition"
|
|
),
|
|
},
|
|
"approve_claim_supported": ["claims", "sources", "evidence", "edges", "reasoning_tools"],
|
|
"keep_staged_until_separate_capability": [
|
|
"behavioral_rules",
|
|
"governance_gates",
|
|
"belief updates",
|
|
"existing-row updates",
|
|
],
|
|
"unsupported_apply_boundary": (
|
|
"approve_claim supports neither behavioral_rules nor governance_gates; both require a separate "
|
|
"reviewed apply capability and remain staged until that review and authorization"
|
|
),
|
|
"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",
|
|
"use only claim type/status values allowed by the current Postgres CHECK constraints",
|
|
"pending_review is a proposal state, not a public.claims status",
|
|
],
|
|
"current_columns": {
|
|
name: list(schema.get(name, ()))
|
|
for name in (
|
|
"claims",
|
|
"claim_edges",
|
|
"reasoning_tools",
|
|
"behavioral_rules",
|
|
"governance_gates",
|
|
"beliefs",
|
|
)
|
|
},
|
|
"current_constraints": {
|
|
name: constraints.get(name, {}) for name in ("claims", "claim_edges", "beliefs")
|
|
},
|
|
"review_apply_sequence": [
|
|
"stage one typed pending_review proposal packet without canonical writes",
|
|
"review mappings, evidence, unsupported changes, and exact row values",
|
|
"apply only supported collections after explicit authorization and keep unsupported changes staged",
|
|
"record proposal-level applied_at plus created/updated row IDs and postflight readback",
|
|
],
|
|
}
|
|
)
|
|
|
|
if forecast_resolution_question:
|
|
contracts.append(
|
|
{
|
|
"id": "forecast_resolution_schema",
|
|
"current_columns": {
|
|
name: list(schema.get(name, ())) for name in ("claims", "claim_edges")
|
|
},
|
|
"history_boundary": "preserve the original forecast and its missing resolution criteria",
|
|
"schema_boundary": (
|
|
"do not invent forecast-resolution fields or a resolves edge; stage a reviewed schema proposal"
|
|
),
|
|
}
|
|
)
|
|
|
|
claim_supersession_question = (
|
|
"claim" in lowered
|
|
and any(term in lowered for term in ("replacement", "wrong", "retired", "supersed"))
|
|
and any(term in lowered for term in ("approve_claim", "current v1", "edge field", "old claim"))
|
|
)
|
|
if claim_supersession_question:
|
|
contracts.append(
|
|
{
|
|
"id": "claim_supersession_schema",
|
|
"current_columns": {
|
|
name: list(schema.get(name, ())) for name in ("claims", "claim_edges")
|
|
},
|
|
"insert_boundary": "approve_claim may insert a replacement claim plus a claim-to-claim edge",
|
|
"update_boundary": (
|
|
"existing claim status and superseded_by updates require a separate reviewed apply capability"
|
|
),
|
|
}
|
|
)
|
|
|
|
telegram_delivery_question = (
|
|
"telegram" in lowered
|
|
and any(term in lowered for term in ("gatewayrunner", "temporary-profile", "temporary profile", "posted nothing"))
|
|
and any(term in lowered for term in ("proven", "delivery", "visible"))
|
|
)
|
|
if telegram_delivery_question:
|
|
contracts.append(
|
|
{
|
|
"id": "telegram_delivery_proof",
|
|
"handler_boundary": "a non-posting GatewayRunner reply proves handler behavior, not Telegram delivery",
|
|
"receipt": "visible reply in the intended chat with message ID, timestamp, and chat/sender readback",
|
|
}
|
|
)
|
|
|
|
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 human_join(values: list[str]) -> str:
|
|
if not values:
|
|
return ""
|
|
if len(values) == 1:
|
|
return values[0]
|
|
if len(values) == 2:
|
|
return f"{values[0]} and {values[1]}"
|
|
return f"{', '.join(values[:-1])}, and {values[-1]}"
|
|
|
|
|
|
def format_database_count_readback(status_data: dict[str, Any] | None) -> str:
|
|
counts = (status_data or {}).get("high_signal_rows") or {}
|
|
labels = ("claims", "sources", "claim_edges", "claim_evidence", "kb_proposals")
|
|
if not all(isinstance(counts.get(label), int) for label in labels):
|
|
return "DB readback: current five-table count receipt unavailable."
|
|
return "DB readback: " + "; ".join(f"{label}: {counts[label]}" for label in labels) + "."
|
|
|
|
|
|
def format_proposal_readback(proposal: dict[str, Any]) -> str:
|
|
applied_at = proposal.get("applied_at") or "none"
|
|
readiness = (proposal.get("readiness") or {}).get("review_state") or "unknown"
|
|
return (
|
|
f"DB readback: proposal: {proposal.get('id')}; status: {proposal.get('status')}; "
|
|
f"applied_at: {applied_at}; readiness: {readiness}."
|
|
)
|
|
|
|
|
|
def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
|
|
"""Compile a safe response when a model draft violates live DB contracts."""
|
|
|
|
by_id = {str(contract.get("id") or ""): contract for contract in contracts}
|
|
participant = by_id.get("telegram_participant_identity")
|
|
if participant:
|
|
handle = str(participant.get("current_visible_handle") or "").lstrip("@")
|
|
return (
|
|
f"Call the current Telegram sender {handle}, exactly as their visible handle identifies them.\n"
|
|
"Allowed identity sources: the current message sender handle and that participant's explicit in-chat "
|
|
"correction.\n"
|
|
"Do not use stale session context, environment labels, inferred personal names, or another "
|
|
"participant's profile.\n"
|
|
"When another user replies, never transfer or mix identities; re-anchor to that message's visible "
|
|
"sender handle."
|
|
)
|
|
|
|
mixed = by_id.get("mixed_packet_composition")
|
|
if mixed:
|
|
mapping = mixed.get("map") or {}
|
|
supported = human_join([str(item) for item in mixed.get("approve_claim_supported") or []])
|
|
staged = human_join([str(item) for item in mixed.get("keep_staged_until_separate_capability") or []])
|
|
return (
|
|
f"- Map factual observations and disputed interpretations to {mapping.get('facts_and_disputes')}; keep "
|
|
"competing claims queryable through claim-to-claim edges. Store the strategic framework in "
|
|
f"{mapping.get('reusable_framework')}. Put the governance rule in {mapping.get('operating_rule')} using "
|
|
"rule, rationale, category, rank, "
|
|
"and agent_id unless it defines criteria, an evidence_bar, and a pass_condition, in which case use "
|
|
f"{mapping.get('evaluative_gate')}. Stage the correction to the old belief as a "
|
|
f"{mapping.get('agent_position')} update; claim_edges never point to "
|
|
"reasoning_tools or beliefs.\n\n"
|
|
"- Stage one typed pending_review proposal with exact rows and no canonical writes. Review mappings, "
|
|
"evidence, claim-only edge endpoints, and unsupported changes. After explicit authorization, "
|
|
f"approve_claim applies only {supported}. approve_claim "
|
|
"supports neither behavioral_rules nor governance_gates; those tables, belief updates, and existing-row "
|
|
f"updates remain staged until a separate reviewed apply capability exists. The contract keeps {staged} "
|
|
"outside this apply.\n\n"
|
|
"- Receipt: proposal-level applied_at, created or updated row IDs, and a postflight readback. Until those "
|
|
"exist, staged or approved content is not live canonical knowledge."
|
|
)
|
|
|
|
source = by_id.get("source_intake")
|
|
if source:
|
|
safe_steps = [str(item) for item in source.get("safe_without_apply_authorization") or []]
|
|
retain_step = safe_steps[0] if safe_steps else "prepare the artifact, extracted text, and manifest"
|
|
stage_step = safe_steps[1] if len(safe_steps) > 1 else "stage a pending_review proposal"
|
|
return (
|
|
f"- {source.get('required_lead')} Sending a report in chat does not by itself create a proposal or "
|
|
"persistent knowledge. I can analyze content available in this conversation, but chat attachment "
|
|
"capture and extraction are not connected to the staging command.\n\n"
|
|
f"- The live bounded path is to {retain_step}, then {stage_step}. It writes only "
|
|
"kb_stage.kb_proposals. A chat label, filename, attachment reference, or source_ref is not provenance. "
|
|
"Before review, deduplicate overlaps and retain exact excerpts, source caveats, uncertainty, and "
|
|
"contradictory interpretations. Candidates are not canonical until reviewed and applied, and no "
|
|
"public.sources or other public.* rows are created during staging.\n\n"
|
|
f"- Explicit approval begins {source.get('approval_begins')}. Receipt: locator and content hash, proposal ID/"
|
|
"status, reviewed payload, applied_at, created row IDs, and postflight readback. Next proof-changing "
|
|
"follow-up: prepare the three inputs, run teleo-kb propose-source, then review its typed candidates."
|
|
)
|
|
|
|
named_packet = by_id.get("named_packet_readback")
|
|
if named_packet:
|
|
matches = list(named_packet.get("matching_proposals") or [])
|
|
primary = matches[0] if matches else None
|
|
canonical_matches = named_packet.get("canonical_matches") or {}
|
|
canonical_counts = {
|
|
table: len(canonical_matches.get(table) or []) for table in ("reasoning_tools", "claims", "sources")
|
|
}
|
|
canonical_total = sum(canonical_counts.values())
|
|
if primary:
|
|
status_value = str(primary.get("status") or "unknown")
|
|
if status_value == "approved":
|
|
state_sentence = (
|
|
"The exact packet is reviewer-approved in kb_stage but not applied; its readiness still requires "
|
|
"a strict apply payload."
|
|
)
|
|
elif status_value == "applied":
|
|
state_sentence = (
|
|
"kb_stage marks the exact packet applied, but that proposal receipt does not prove which canonical "
|
|
"rows were written."
|
|
)
|
|
elif status_value == "pending_review":
|
|
state_sentence = "The exact packet remains pending_review in kb_stage and is not applied."
|
|
else:
|
|
state_sentence = f"The exact packet is {status_value} in kb_stage and is not a canonical apply proof."
|
|
lead = (
|
|
"No, Helmer's 7 Powers is not proven in canonical Leo."
|
|
if canonical_total == 0
|
|
else "Partly: canonical match candidates exist, but the complete Helmer's 7 Powers packet is not proven."
|
|
)
|
|
return (
|
|
f"{lead}\n{format_proposal_readback(primary)}\nCanonical match readback: reasoning_tools: "
|
|
f"{canonical_counts['reasoning_tools']}; claims: {canonical_counts['claims']}; sources: "
|
|
f"{canonical_counts['sources']}. {state_sentence}\n\n"
|
|
"Next proof-changing follow-up: inspect the exact canonical match row IDs, normalize and renew review "
|
|
"if needed, obtain explicit apply authorization, then return proposal applied_at plus public.* "
|
|
"postflight and Telegram regression proof."
|
|
)
|
|
return (
|
|
"No, Helmer's 7 Powers is not proven in canonical Leo.\n"
|
|
f"{format_database_count_readback(named_packet.get('database_status'))}\nCanonical match readback: "
|
|
f"reasoning_tools: {canonical_counts['reasoning_tools']}; claims: {canonical_counts['claims']}; sources: "
|
|
f"{canonical_counts['sources']}. No exact Helmer plus Powers proposal is present in kb_stage.\n\n"
|
|
"Next proof-changing follow-up: retain the exact source artifact, stage a strict pending_review packet, "
|
|
"obtain explicit apply authorization after review, then return proposal applied_at and public.* postflight."
|
|
)
|
|
|
|
matrix = by_id.get("decision_matrix_readback")
|
|
if matrix:
|
|
live_matrix = matrix.get("live_matrix_status") or {}
|
|
table_states = live_matrix.get("decision_matrix_tables") or {}
|
|
unobserved = [name for name in DECISION_MATRIX_TABLES if table_states.get(name) not in (True, False)]
|
|
absent = [name for name in DECISION_MATRIX_TABLES if table_states.get(name) is False]
|
|
if unobserved:
|
|
return (
|
|
f"Not proven.\n{format_database_count_readback(matrix.get('database_status'))}\nFresh readback: the "
|
|
f"matrix table-state snapshot is incomplete; {len(unobserved)} required tables were not observed "
|
|
f"({', '.join(unobserved)}). Their presence or absence cannot be inferred. kb_stage reviewer status "
|
|
"is not a decision-matrix vote, so matrix approval is not proven.\n\n"
|
|
"Next proof-changing follow-up: rerun the complete six-table schema readback, then inspect the specific "
|
|
"proposal UUID/status/applied_at and its proposal_votes and proposal_decisions rows before claiming "
|
|
"matrix approval."
|
|
)
|
|
if not absent:
|
|
return (
|
|
f"Not proven.\n{format_database_count_readback(matrix.get('database_status'))}\nFresh readback: all "
|
|
"required matrix tables are present, but this snapshot does not inspect proposal-specific vote or "
|
|
"decision rows. kb_stage reviewer status is not a decision-matrix vote, so matrix approval is not "
|
|
"proven from proposal status alone.\n\n"
|
|
"Next proof-changing follow-up: inspect the specific proposal UUID/status/applied_at, then return its "
|
|
"proposal_votes and proposal_decisions rows before claiming matrix approval; otherwise use the human "
|
|
"review and guarded canonical apply path with row-level postflight."
|
|
)
|
|
return (
|
|
f"No.\n{format_database_count_readback(matrix.get('database_status'))}\nFresh readback: "
|
|
f"{len(absent)} matrix tables are absent ({', '.join(absent)}). kb_stage reviewer status is not a "
|
|
"decision-matrix vote, so no proposal can be called matrix-approved from proposal status alone.\n\n"
|
|
"Next proof-changing follow-up: inspect the specific proposal UUID/status/applied_at in kb_stage, then "
|
|
"either show proposal_votes plus proposal_decisions for that UUID after the schema exists, or use the "
|
|
"human review and guarded canonical apply path with row-level postflight."
|
|
)
|
|
|
|
source_audit = by_id.get("source_link_audit")
|
|
if source_audit:
|
|
audit = source_audit.get("live_link_audit") or {}
|
|
return (
|
|
f"No. The attachment does not itself create canonical evidence; neither does extracted text on disk.\n"
|
|
f"{format_database_count_readback(source_audit.get('database_status'))}\nThe live row-link "
|
|
f"audit found {audit.get('pending_or_approved', 'unknown')} pending/approved proposals: "
|
|
f"{audit.get('with_source_ref', 'unknown')} have source_ref and "
|
|
f"{audit.get('exact_public_source_matches', 'unknown')} exactly match a public.sources ID, URL, or "
|
|
"storage_path. Pending status therefore is not by itself proof of a bad document pointer.\n\n"
|
|
"Audit telegram_file_refs/raw files -> document_evaluations -> kb_proposals.source_ref -> public.sources "
|
|
"-> claim_evidence. Canonical linkage and provenance quality are separate: a canonical link can still point "
|
|
"to a weak, untraceable source with no URL/storage_path or raw artifact. Before/after receipt: proposal UUID/"
|
|
"status/source_ref, source and claim_evidence row IDs, locator/hash, applied_at, and postflight. Next proof-"
|
|
"changing follow-up: list each broken layer, then stage only the missing guarded source/evidence contract."
|
|
)
|
|
|
|
identity = by_id.get("identity_canonicality_readback")
|
|
if identity:
|
|
return (
|
|
f"No.\n{format_database_count_readback(identity.get('database_status'))}\nA direct SOUL.md edit is a "
|
|
"runtime/profile change; it does not change canonical Postgres rows in personas, strategies, beliefs, "
|
|
"strategy_nodes, or strategy_node_anchors. A chat correction is also not canonical until it is staged, "
|
|
"reviewed, and applied as exact identity rows. No active general DB-to-SOUL.md renderer automation is "
|
|
"currently proven, so neither surface should be assumed to survive as identity after restart.\n\n"
|
|
"Next proof-changing follow-up: audit the intended canonical identity rows and proposal UUID/status/"
|
|
"applied_at, stage the exact change, review it, apply it with authorization, run or verify render/sync, then "
|
|
"compare SOUL.md hash/content and restart readback."
|
|
)
|
|
|
|
demo = by_id.get("demo_capability_readback")
|
|
if demo:
|
|
status_data = demo.get("database_status") or {}
|
|
states = status_data.get("proposal_status_counts") or {}
|
|
applied_rows = list(demo.get("applied_proposals") or [])
|
|
approved_rows = [
|
|
item for item in demo.get("applied_and_approved_proposals") or [] if item.get("status") == "approved"
|
|
]
|
|
approved_needs_payload = sum(
|
|
(item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload"
|
|
for item in approved_rows
|
|
)
|
|
applied_types = human_join(
|
|
sorted({str(item.get("proposal_type")) for item in applied_rows if item.get("proposal_type")})
|
|
)
|
|
if demo.get("temporal_comparison_requested"):
|
|
return (
|
|
f"Not proven.\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are "
|
|
f"applied: {states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: "
|
|
f"{states.get('pending_review', 0)}. This current snapshot does not prove what changed since "
|
|
f"yesterday, so the demo claim is limited to current readback. Existing receipt artifacts identify "
|
|
f"applied types {applied_types or 'none'}, but not a time-window delta. Approved is not applied; "
|
|
f"{approved_needs_payload}/{len(approved_rows)} approved rows are "
|
|
"approved_needs_apply_payload and not directly applyable.\n\n"
|
|
"Next proof-changing follow-up: build and review a strict apply_payload for the highest-impact approved "
|
|
"proposal, then seek explicit apply authorization. Until its applied_at, public.* before/after row "
|
|
"IDs or hashes, and postflight readback exist, do not claim the canonical database changed."
|
|
)
|
|
return (
|
|
f"Partly.\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: "
|
|
f"{states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: "
|
|
f"{states.get('pending_review', 0)}. Existing receipts show applied proposal types: "
|
|
f"{applied_types or 'none identified'}; that receipt artifact does not prove every rich packet or chat "
|
|
f"apply path is live. Approved is not applied. {approved_needs_payload}/{len(approved_rows)} approved rows "
|
|
"are approved_needs_apply_payload and not directly applyable. "
|
|
"You can safely demo canonical readback and a pending_review staging write. A new canonical apply requires "
|
|
"explicit authorization, a supported strict payload, and postflight row IDs.\n\n"
|
|
"Next proof-changing follow-up: choose the demo tier - existing applied receipt, safe staging canary, or "
|
|
"an explicitly authorized guarded apply - then capture proposal UUID/status/applied_at, public.* before/"
|
|
"after, and Telegram-visible regression proof."
|
|
)
|
|
|
|
proposal_state = by_id.get("proposal_state_readback")
|
|
if proposal_state:
|
|
status_data = proposal_state.get("database_status") or {}
|
|
states = status_data.get("proposal_status_counts") or {}
|
|
approved_rows = [
|
|
item
|
|
for item in proposal_state.get("applied_and_approved_proposals") or []
|
|
if item.get("status") == "approved"
|
|
]
|
|
approved_needs_payload = sum(
|
|
(item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload"
|
|
for item in approved_rows
|
|
)
|
|
lead = "No." if proposal_state.get("approval_only_claim") else "Partly."
|
|
return (
|
|
f"{lead}\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: "
|
|
f"{states.get('applied', 0)}, approved: {states.get('approved', 0)}, pending_review: "
|
|
f"{states.get('pending_review', 0)}, canceled: {states.get('canceled', 0)}. Approved is not applied, and "
|
|
f"{approved_needs_payload}/{len(approved_rows)} approved rows are approved_needs_apply_payload and not "
|
|
"directly applyable. An applied proposal receipt alone does not identify the exact public.* rows changed.\n\n"
|
|
"Next proof-changing follow-up: choose the proposal or time window, return its UUID/status/applied_at, "
|
|
"compare the affected public.* row IDs or before/after hashes, and for approved rows normalize and review "
|
|
"the strict payload before requesting explicit guarded-apply authorization."
|
|
)
|
|
|
|
runtime = by_id.get("runtime_persistence")
|
|
if runtime:
|
|
return (
|
|
"No. Unchanged database totals do not prove unchanged rows or unchanged answer behavior, and a restart "
|
|
"does not necessarily erase prior-session facts. Compare row IDs, timestamps, and hashes, not counts alone; "
|
|
"balanced inserts, updates, or deletes can leave totals unchanged.\n\n"
|
|
"Persisted or deployed answer inputs include Postgres rows, skills, runtime config, SOUL.md, state.db, and "
|
|
"session JSONL. state.db/session JSONL can preserve durable session continuity across a restart, while a "
|
|
"changed skill or runtime artifact can change the answer with identical database totals.\n\n"
|
|
"Proof tiers: 1) canonical: public.* row/hash readback; 2) handler: the deployed runtime answers the same "
|
|
"prompt with retained trace; 3) Telegram: a visible reply with message ID, timestamp, and chat/sender "
|
|
"readback. Only a reviewed apply plus row-level postflight proves a database mutation."
|
|
)
|
|
|
|
shared = by_id.get("shared_claims_agent_positions")
|
|
if shared:
|
|
return (
|
|
"Store one factual claim once as shared knowledge with shared sources/evidence; do not duplicate the claim "
|
|
"per agent. Record each agent-specific position in public.beliefs with agent_id, statement, confidence, "
|
|
"falsifier, and status.\n\n"
|
|
"The current-schema gap matters: public.beliefs has no claim-ID foreign key and there is no belief-edge "
|
|
"table. claim_edges connects claims only, so do not invent belief-to-claim or belief-to-belief edges. "
|
|
"Disagreement remains queryable by retrieving the shared factual claim/evidence and semantically matching "
|
|
"each agent's belief rows; a future reviewed schema proposal can add an explicit link if needed."
|
|
)
|
|
|
|
forecast = by_id.get("forecast_resolution_schema")
|
|
if forecast:
|
|
columns = forecast.get("current_columns") or {}
|
|
claim_columns = human_join([str(item) for item in columns.get("claims") or []])
|
|
edge_columns = human_join([str(item) for item in columns.get("claim_edges") or []])
|
|
edge_types = [str(item) for item in forecast.get("current_edge_types") or []]
|
|
resolves_state = "`resolves` is present" if "resolves" in edge_types else "there is no `resolves` edge type"
|
|
return (
|
|
"Preserve the original 60% forecast; do not overwrite it. Because it defined no resolution criteria, its "
|
|
"outcome remains ambiguous. Leo may add a sourced explanatory claim about the observed event, but must not "
|
|
"retroactively label the forecast resolved.\n\n"
|
|
f"Current public.claims columns are {claim_columns}; there is no forecast-resolution or resolved_at field. "
|
|
f"Current public.claim_edges columns are {edge_columns}; {resolves_state}. Stage a reviewed schema proposal "
|
|
"for explicit criteria, outcome, resolved_at, resolver/provenance, and a valid link contract. Do not apply or "
|
|
"invent those fields in v1."
|
|
)
|
|
|
|
delivery = by_id.get("telegram_delivery_proof")
|
|
if delivery:
|
|
return (
|
|
"No. A temporary-profile GatewayRunner response proves handler execution, model/plugin integration, and "
|
|
"reply construction only. Because it posted nothing, it does not prove Telegram delivery or a Telegram-"
|
|
"visible reply.\n\n"
|
|
"The smallest closing test is one explicitly authorized test message in the intended chat, followed by "
|
|
"Leo's visible reply. Retain a delivery receipt with the outgoing and reply message IDs, timestamps, chat "
|
|
"ID/title, sender handle, reply linkage, and a post-send readback. Then remove the temporary profile and "
|
|
"verify no orphan process remains."
|
|
)
|
|
|
|
supersession = by_id.get("claim_supersession_schema")
|
|
if supersession:
|
|
columns = supersession.get("current_columns") or {}
|
|
claim_columns = human_join([str(item) for item in columns.get("claims") or []])
|
|
edge_columns = human_join([str(item) for item in columns.get("claim_edges") or []])
|
|
edge_types = [str(item) for item in supersession.get("current_edge_types") or []]
|
|
supersedes_state = "`supersedes` is valid" if "supersedes" in edge_types else "`supersedes` is not valid"
|
|
return (
|
|
f"Current public.claims fields: {claim_columns}. Current claim_edges fields: {edge_columns}; the edge has no "
|
|
f"rationale field, and {supersedes_state} in the live edge_type enum.\n\n"
|
|
"approve_claim can insert the replacement claim and a claim-to-claim supersedes edge using from_claim, "
|
|
"to_claim, edge_type, weight, and created_by. Put the explanation in sourced evidence or an explanatory "
|
|
"claim. Updating the existing old claim's status or superseded_by to visibly retire it requires a separate "
|
|
"reviewed apply capability; approve_claim does not update those fields. Receipt: proposal applied_at, new "
|
|
"claim/edge IDs, old-claim before/after, and postflight readback."
|
|
)
|
|
|
|
return None
|
|
|
|
|
|
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 load_current_public_constraints(args: argparse.Namespace, table_names: set[str]) -> dict[str, dict[str, str]]:
|
|
if not table_names:
|
|
return {}
|
|
sql = f"""
|
|
select jsonb_build_object(
|
|
'table_name', t.relname,
|
|
'constraints', jsonb_object_agg(c.conname, pg_get_constraintdef(c.oid) order by c.conname)
|
|
)::text
|
|
from pg_constraint c
|
|
join pg_class t on t.oid = c.conrelid
|
|
join pg_namespace n on n.oid = t.relnamespace
|
|
where n.nspname = 'public'
|
|
and t.relname = any({sql_array(sorted(table_names))})
|
|
group by t.relname
|
|
order by t.relname;
|
|
"""
|
|
return {row["table_name"]: row["constraints"] for row in psql_json(args, sql)}
|
|
|
|
|
|
def load_current_enum_values(args: argparse.Namespace, type_names: set[str]) -> dict[str, list[str]]:
|
|
if not type_names:
|
|
return {}
|
|
sql = f"""
|
|
select jsonb_build_object(
|
|
'type_name', t.typname,
|
|
'values', jsonb_agg(e.enumlabel order by e.enumsortorder)
|
|
)::text
|
|
from pg_type t
|
|
join pg_enum e on e.enumtypid = t.oid
|
|
join pg_namespace n on n.oid = t.typnamespace
|
|
where n.nspname = 'public'
|
|
and t.typname = any({sql_array(sorted(type_names))})
|
|
group by t.typname
|
|
order by t.typname;
|
|
"""
|
|
return {row["type_name"]: row["values"] for row in psql_json(args, sql)}
|
|
|
|
|
|
def compact_proposal(proposal: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
key: proposal.get(key)
|
|
for key in ("id", "proposal_type", "status", "source_ref", "created_at", "reviewed_at", "applied_at")
|
|
} | {"readiness": proposal.get("readiness") or {}}
|
|
|
|
|
|
def load_direct_readback_snapshot(args: argparse.Namespace) -> dict[str, Any]:
|
|
"""Read all broad direct-claim state from one PostgreSQL statement snapshot."""
|
|
|
|
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
|
|
),
|
|
proposal_rows as (
|
|
select p.status,
|
|
lower(concat_ws(' ', p.rationale, p.source_ref, p.proposal_type, p.payload::text)) as search_text,
|
|
jsonb_build_object(
|
|
'id', p.id::text,
|
|
'proposal_type', p.proposal_type,
|
|
'status', p.status,
|
|
'source_ref', p.source_ref,
|
|
'payload', p.payload,
|
|
'created_at', p.created_at::text,
|
|
'reviewed_at', p.reviewed_at::text,
|
|
'applied_at', p.applied_at::text
|
|
) as row_value,
|
|
p.created_at
|
|
from kb_stage.kb_proposals p
|
|
),
|
|
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)
|
|
),
|
|
source_scoped as (
|
|
select p.id,
|
|
p.status,
|
|
p.source_ref,
|
|
exists (
|
|
select 1
|
|
from public.sources s
|
|
where p.source_ref is not null
|
|
and (s.id::text = p.source_ref or s.url = p.source_ref or s.storage_path = p.source_ref)
|
|
) as exact_public_source_match
|
|
from kb_stage.kb_proposals p
|
|
where p.status in ('pending_review', 'approved')
|
|
)
|
|
select jsonb_build_object(
|
|
'snapshot_id', txid_current_snapshot()::text,
|
|
'statement_timestamp', statement_timestamp()::text,
|
|
'database_status', 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)
|
|
),
|
|
'proposals', coalesce((
|
|
select jsonb_agg(row_value order by created_at desc) from proposal_rows
|
|
), '[]'::jsonb),
|
|
'named_packet_proposals', coalesce((
|
|
select jsonb_agg(
|
|
row_value
|
|
order by case status when 'applied' then 0 when 'approved' then 1
|
|
when 'pending_review' then 2 else 3 end,
|
|
created_at desc
|
|
)
|
|
from proposal_rows
|
|
where search_text like '%helmer%' and search_text like '%power%'
|
|
), '[]'::jsonb),
|
|
'named_packet_canonical_matches', jsonb_build_object(
|
|
'reasoning_tools', coalesce((
|
|
select jsonb_agg(jsonb_build_object('id', id::text, 'name', name) order by name, id)
|
|
from public.reasoning_tools
|
|
where lower(concat_ws(' ', name, description)) like '%helmer%'
|
|
and lower(concat_ws(' ', name, description)) like '%power%'
|
|
), '[]'::jsonb),
|
|
'claims', coalesce((
|
|
select jsonb_agg(jsonb_build_object('id', id::text, 'status', status) order by id)
|
|
from public.claims
|
|
where lower(text) like '%helmer%' and lower(text) like '%power%'
|
|
), '[]'::jsonb),
|
|
'sources', coalesce((
|
|
select jsonb_agg(jsonb_build_object('id', id::text, 'hash', hash) order by id)
|
|
from public.sources
|
|
where lower(concat_ws(' ', url, storage_path, excerpt)) like '%helmer%'
|
|
and lower(concat_ws(' ', url, storage_path, excerpt)) like '%power%'
|
|
), '[]'::jsonb)
|
|
),
|
|
'decision_matrix_status', jsonb_build_object(
|
|
'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)
|
|
),
|
|
'source_link_audit', jsonb_build_object(
|
|
'pending_or_approved', (select count(*) from source_scoped),
|
|
'with_source_ref', (select count(*) from source_scoped where source_ref is not null),
|
|
'without_source_ref', (select count(*) from source_scoped where source_ref is null),
|
|
'exact_public_source_matches', (
|
|
select count(*) from source_scoped where exact_public_source_match
|
|
),
|
|
'samples', coalesce((
|
|
select jsonb_agg(
|
|
jsonb_build_object(
|
|
'proposal_id', id::text,
|
|
'status', status,
|
|
'source_ref', source_ref,
|
|
'exact_public_source_match', exact_public_source_match
|
|
)
|
|
order by status, id
|
|
)
|
|
from (
|
|
select * from source_scoped order by status, id limit 12
|
|
) sampled
|
|
), '[]'::jsonb)
|
|
)
|
|
)::text;
|
|
"""
|
|
rows = psql_json(args, sql)
|
|
if not rows:
|
|
raise SystemExit("No direct readback snapshot returned.")
|
|
return rows[0]
|
|
|
|
|
|
def load_source_link_audit(args: argparse.Namespace) -> dict[str, Any]:
|
|
sql = """
|
|
with scoped as (
|
|
select p.id,
|
|
p.status,
|
|
p.source_ref,
|
|
exists (
|
|
select 1
|
|
from public.sources s
|
|
where p.source_ref is not null
|
|
and (
|
|
s.id::text = p.source_ref
|
|
or s.url = p.source_ref
|
|
or s.storage_path = p.source_ref
|
|
)
|
|
) as exact_public_source_match
|
|
from kb_stage.kb_proposals p
|
|
where p.status in ('pending_review', 'approved')
|
|
),
|
|
samples as (
|
|
select id, status, source_ref, exact_public_source_match
|
|
from scoped
|
|
order by status, id
|
|
limit 12
|
|
)
|
|
select jsonb_build_object(
|
|
'pending_or_approved', (select count(*) from scoped),
|
|
'with_source_ref', (select count(*) from scoped where source_ref is not null),
|
|
'without_source_ref', (select count(*) from scoped where source_ref is null),
|
|
'exact_public_source_matches', (
|
|
select count(*) from scoped where exact_public_source_match
|
|
),
|
|
'samples', coalesce((
|
|
select jsonb_agg(jsonb_build_object(
|
|
'proposal_id', id::text,
|
|
'status', status,
|
|
'source_ref', source_ref,
|
|
'exact_public_source_match', exact_public_source_match
|
|
) order by status, id)
|
|
from samples
|
|
), '[]'::jsonb)
|
|
)::text;
|
|
"""
|
|
rows = psql_json(args, sql)
|
|
return rows[0] if rows else {}
|
|
|
|
|
|
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 contract_ids & SCHEMA_COMPILED_CONTRACTS:
|
|
table_names.update({"claims", "claim_edges"})
|
|
enum_values = load_current_enum_values(args, {"edge_type"}) if contract_ids & SCHEMA_COMPILED_CONTRACTS else {}
|
|
contracts = operational_contracts(
|
|
query,
|
|
current_schema=load_current_public_schema(args, table_names) if table_names else None,
|
|
current_constraints=load_current_public_constraints(args, table_names) if table_names else None,
|
|
)
|
|
contract_ids = {contract["id"] for contract in contracts}
|
|
direct_snapshot = load_direct_readback_snapshot(args) if contract_ids & DIRECT_READBACK_CONTRACTS else {}
|
|
status_data = direct_snapshot.get("database_status")
|
|
proposal_rows = [compact_proposal(with_proposal_readiness(item)) for item in direct_snapshot.get("proposals") or []]
|
|
|
|
for contract in contracts:
|
|
contract_id = contract["id"]
|
|
if contract_id in DIRECT_READBACK_CONTRACTS:
|
|
contract["database_status"] = status_data
|
|
contract["snapshot_id"] = direct_snapshot.get("snapshot_id")
|
|
contract["statement_timestamp"] = direct_snapshot.get("statement_timestamp")
|
|
if contract_id == "proposal_state_readback":
|
|
contract["applied_and_approved_proposals"] = [
|
|
item for item in proposal_rows if item.get("status") in {"applied", "approved"}
|
|
]
|
|
elif contract_id == "named_packet_readback":
|
|
contract["matching_proposals"] = [
|
|
compact_proposal(with_proposal_readiness(item))
|
|
for item in direct_snapshot.get("named_packet_proposals") or []
|
|
]
|
|
contract["canonical_matches"] = direct_snapshot.get("named_packet_canonical_matches") or {}
|
|
elif contract_id == "decision_matrix_readback":
|
|
contract["live_matrix_status"] = direct_snapshot.get("decision_matrix_status") or {}
|
|
elif contract_id == "source_link_audit":
|
|
contract["live_link_audit"] = direct_snapshot.get("source_link_audit") or {}
|
|
elif contract_id == "demo_capability_readback":
|
|
contract["applied_proposals"] = [item for item in proposal_rows if item.get("status") == "applied"]
|
|
contract["applied_and_approved_proposals"] = [
|
|
item for item in proposal_rows if item.get("status") in {"applied", "approved"}
|
|
]
|
|
elif contract_id in SCHEMA_COMPILED_CONTRACTS:
|
|
contract["current_edge_types"] = enum_values.get("edge_type", [])
|
|
return contracts
|
|
|
|
|
|
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 canonical_json_sha256(value: Any) -> str:
|
|
encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def _validated_uuid(value: Any, label: str) -> str:
|
|
try:
|
|
return str(uuid.UUID(str(value)))
|
|
except (ValueError, AttributeError) as exc:
|
|
raise SystemExit(f"{label} must be a full UUID") from exc
|
|
|
|
|
|
def _validated_source_bundle(bundle: Any) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
if not isinstance(bundle, dict) or bundle.get("schema") != SOURCE_COMPILER_BUNDLE_SCHEMA:
|
|
raise SystemExit(f"source compiler must return {SOURCE_COMPILER_BUNDLE_SCHEMA}")
|
|
if (
|
|
bundle.get("status") != "pending_review"
|
|
or bundle.get("build_only") is not True
|
|
or bundle.get("database_write_performed") is not False
|
|
or bundle.get("canonical_apply_performed") is not False
|
|
):
|
|
raise SystemExit("source compiler bundle violated its build-only pending-review contract")
|
|
|
|
child = bundle.get("strict_child_proposal")
|
|
if not isinstance(child, dict):
|
|
raise SystemExit("source compiler bundle has no strict_child_proposal object")
|
|
_validated_uuid(child.get("id"), "strict child proposal id")
|
|
if child.get("proposal_type") != "approve_claim" or child.get("status") != "pending_review":
|
|
raise SystemExit("strict child must be one pending_review approve_claim proposal")
|
|
if child.get("proposed_by_handle") != "leo":
|
|
raise SystemExit("strict child proposed_by_handle must be leo")
|
|
for key in ("channel", "source_ref", "rationale", "payload_sha256"):
|
|
if not isinstance(child.get(key), str) or not child[key].strip():
|
|
raise SystemExit(f"strict child {key} must be a non-empty string")
|
|
if child["channel"] != "source_document_compiler" or not child["source_ref"].startswith("normalized:"):
|
|
raise SystemExit("strict child is not bound to the source-document compiler contract")
|
|
if not SHA256_RE.fullmatch(child["payload_sha256"]):
|
|
raise SystemExit("strict child payload_sha256 must be a lowercase SHA-256 digest")
|
|
if child.get("proposed_by_agent_id") is not None:
|
|
_validated_uuid(child["proposed_by_agent_id"], "strict child proposed_by_agent_id")
|
|
payload = child.get("payload")
|
|
if not isinstance(payload, dict):
|
|
raise SystemExit("strict child payload must be an object")
|
|
apply_payload = payload.get("apply_payload")
|
|
if not isinstance(apply_payload, dict) or apply_payload.get("contract_version") != 2:
|
|
raise SystemExit("strict child must contain a schema-valid approve_claim v2 apply_payload")
|
|
if canonical_json_sha256(payload) != child["payload_sha256"]:
|
|
raise SystemExit("strict child payload hash does not match its canonical JSON")
|
|
|
|
hashes = bundle.get("hashes")
|
|
if not isinstance(hashes, dict):
|
|
raise SystemExit("source compiler bundle has no hashes object")
|
|
for key in (
|
|
"artifact_sha256",
|
|
"extracted_text_sha256",
|
|
"manifest_canonical_sha256",
|
|
"strict_child_payload_sha256",
|
|
"bundle_content_sha256",
|
|
):
|
|
if not isinstance(hashes.get(key), str) or not SHA256_RE.fullmatch(hashes[key]):
|
|
raise SystemExit(f"source compiler hash {key} is missing or invalid")
|
|
if hashes["strict_child_payload_sha256"] != child["payload_sha256"]:
|
|
raise SystemExit("source compiler child hash disagrees with the strict child")
|
|
|
|
bundle_without_content_hash = json.loads(json.dumps(bundle))
|
|
bundle_without_content_hash["hashes"].pop("bundle_content_sha256")
|
|
if canonical_json_sha256(bundle_without_content_hash) != hashes["bundle_content_sha256"]:
|
|
raise SystemExit("source compiler bundle content hash does not match its canonical JSON")
|
|
|
|
manifest = bundle.get("validated_manifest")
|
|
if not isinstance(manifest, dict) or not isinstance(manifest.get("source"), dict):
|
|
raise SystemExit("source compiler bundle has no validated source manifest")
|
|
if manifest.get("artifact_sha256") != hashes["artifact_sha256"]:
|
|
raise SystemExit("validated manifest artifact hash disagrees with the compiler receipt")
|
|
if manifest.get("extracted_text_sha256") != hashes["extracted_text_sha256"]:
|
|
raise SystemExit("validated manifest text hash disagrees with the compiler receipt")
|
|
return child, manifest
|
|
|
|
|
|
def compile_source_bundle(args: argparse.Namespace) -> dict[str, Any]:
|
|
if not SOURCE_COMPILER_PATH.is_file():
|
|
raise SystemExit(f"source compiler is not deployed: {SOURCE_COMPILER_PATH}")
|
|
command = [
|
|
sys.executable,
|
|
str(SOURCE_COMPILER_PATH),
|
|
"--artifact",
|
|
str(args.artifact),
|
|
"--text",
|
|
str(args.text),
|
|
"--manifest",
|
|
str(args.manifest),
|
|
]
|
|
try:
|
|
completed = subprocess.run(command, capture_output=True, text=True, timeout=120, check=False)
|
|
except (OSError, subprocess.TimeoutExpired) as exc:
|
|
raise SystemExit(f"source compiler could not run: {exc}") from exc
|
|
if completed.returncode != 0:
|
|
detail = completed.stdout.strip() or completed.stderr.strip() or f"exit {completed.returncode}"
|
|
raise SystemExit(f"source compiler rejected the artifact: {detail}")
|
|
try:
|
|
bundle = json.loads(completed.stdout)
|
|
except json.JSONDecodeError as exc:
|
|
raise SystemExit(f"source compiler emitted invalid JSON: {exc}") from exc
|
|
_validated_source_bundle(bundle)
|
|
return bundle
|
|
|
|
|
|
def build_source_stage_sql(child: dict[str, Any]) -> str:
|
|
proposal_id = sql_literal(_validated_uuid(child.get("id"), "strict child proposal id"))
|
|
proposal_type = sql_literal(str(child["proposal_type"]))
|
|
proposed_by_handle = sql_literal(str(child.get("proposed_by_handle") or ""))
|
|
proposed_by_agent_id = child.get("proposed_by_agent_id")
|
|
agent_sql = (
|
|
f"{sql_literal(_validated_uuid(proposed_by_agent_id, 'strict child proposed_by_agent_id'))}::uuid"
|
|
if proposed_by_agent_id
|
|
else "null::uuid"
|
|
)
|
|
channel = sql_literal(str(child["channel"]))
|
|
source_ref = sql_literal(str(child["source_ref"]))
|
|
rationale = sql_literal(str(child["rationale"]))
|
|
payload = sql_literal(json.dumps(child["payload"], sort_keys=True, separators=(",", ":"), ensure_ascii=True))
|
|
payload_sha256 = sql_literal(str(child["payload_sha256"]))
|
|
return f"""\\set ON_ERROR_STOP on
|
|
begin;
|
|
create temporary table teleo_source_stage_result (created boolean not null) on commit drop;
|
|
with inserted as (
|
|
insert into kb_stage.kb_proposals
|
|
(id, proposal_type, status, proposed_by_handle, proposed_by_agent_id,
|
|
channel, source_ref, rationale, payload)
|
|
values
|
|
({proposal_id}::uuid, {proposal_type}, 'pending_review', nullif({proposed_by_handle}, ''),
|
|
{agent_sql}, {channel}, {source_ref}, {rationale}, {payload}::jsonb)
|
|
on conflict (id) do nothing
|
|
returning id
|
|
)
|
|
insert into teleo_source_stage_result(created)
|
|
select exists(select 1 from inserted);
|
|
|
|
do $teleo_source_stage$
|
|
declare
|
|
marker_count integer;
|
|
exact_count integer;
|
|
begin
|
|
select count(*) into marker_count
|
|
from kb_stage.kb_proposals
|
|
where source_ref = {source_ref};
|
|
select count(*) into exact_count
|
|
from kb_stage.kb_proposals
|
|
where id = {proposal_id}::uuid
|
|
and proposal_type = {proposal_type}
|
|
and proposed_by_handle is not distinct from nullif({proposed_by_handle}, '')
|
|
and proposed_by_agent_id is not distinct from {agent_sql}
|
|
and channel = {channel}
|
|
and source_ref = {source_ref}
|
|
and rationale = {rationale}
|
|
and payload = {payload}::jsonb;
|
|
if marker_count <> 1 or exact_count <> 1 then
|
|
raise exception 'source stage validation failed: marker_count=%, exact_count=%', marker_count, exact_count;
|
|
end if;
|
|
end
|
|
$teleo_source_stage$;
|
|
|
|
select jsonb_build_object(
|
|
'created', (select created from teleo_source_stage_result),
|
|
'proposal_id', id::text,
|
|
'proposal_type', proposal_type,
|
|
'status', status,
|
|
'source_ref', source_ref,
|
|
'payload', payload,
|
|
'payload_sha256', {payload_sha256},
|
|
'created_at', created_at::text,
|
|
'reviewed_at', reviewed_at::text,
|
|
'applied_at', applied_at::text
|
|
)::text
|
|
from kb_stage.kb_proposals
|
|
where id = {proposal_id}::uuid;
|
|
commit;
|
|
"""
|
|
|
|
|
|
def stage_source_child(args: argparse.Namespace, child: dict[str, Any]) -> dict[str, Any]:
|
|
rows = psql_json(args, build_source_stage_sql(child))
|
|
if len(rows) != 1:
|
|
raise SystemExit("source staging did not return exactly one proposal receipt")
|
|
row = rows[0]
|
|
expected = {
|
|
"proposal_id": child["id"],
|
|
"proposal_type": child["proposal_type"],
|
|
"source_ref": child["source_ref"],
|
|
"payload_sha256": child["payload_sha256"],
|
|
"payload": child["payload"],
|
|
}
|
|
for key, value in expected.items():
|
|
if row.get(key) != value:
|
|
raise SystemExit(f"source staging receipt mismatch for {key}")
|
|
if row.get("status") not in {"pending_review", "approved", "applied", "canceled"}:
|
|
raise SystemExit("source staging returned an unknown proposal status")
|
|
if not isinstance(row.get("created"), bool):
|
|
raise SystemExit("source staging receipt has no created boolean")
|
|
return row
|
|
|
|
|
|
def _write_private_json(path: Path, value: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, raw_temp_path = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
|
temp_path = Path(raw_temp_path)
|
|
try:
|
|
os.fchmod(fd, 0o600)
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
json.dump(value, handle, indent=2, sort_keys=True, ensure_ascii=True)
|
|
handle.write("\n")
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temp_path, path)
|
|
finally:
|
|
if temp_path.exists():
|
|
temp_path.unlink()
|
|
|
|
|
|
def propose_source(args: argparse.Namespace) -> dict[str, Any]:
|
|
if not args.local:
|
|
raise SystemExit(
|
|
"propose-source is VPS-local only; run it through the deployed teleo-kb wrapper on the VPS"
|
|
)
|
|
receipt_path = args.receipt.expanduser().resolve()
|
|
input_paths = {args.artifact.expanduser().resolve(), args.text.expanduser().resolve(), args.manifest.expanduser().resolve()}
|
|
if receipt_path in input_paths:
|
|
raise SystemExit("--receipt must not overwrite an input file")
|
|
|
|
bundle = compile_source_bundle(args)
|
|
child, manifest = _validated_source_bundle(bundle)
|
|
try:
|
|
compiler_sha256 = hashlib.sha256(SOURCE_COMPILER_PATH.read_bytes()).hexdigest()
|
|
receipt_path.parent.mkdir(parents=True, exist_ok=True)
|
|
fd, receipt_preflight = tempfile.mkstemp(prefix=f".{receipt_path.name}.preflight.", dir=receipt_path.parent)
|
|
os.close(fd)
|
|
Path(receipt_preflight).unlink()
|
|
except OSError as exc:
|
|
raise SystemExit(f"receipt path or compiler cannot be readied before staging: {exc}") from exc
|
|
before = status(args)
|
|
staged = stage_source_child(args, child)
|
|
after = status(args)
|
|
before_counts = dict(before.get("high_signal_rows") or {})
|
|
after_counts = dict(after.get("high_signal_rows") or {})
|
|
canonical_counts_unchanged = all(before_counts.get(key) == after_counts.get(key) for key in CANONICAL_COUNT_LABELS)
|
|
proposal_delta = after_counts.get("kb_proposals", 0) - before_counts.get("kb_proposals", 0)
|
|
created = bool(staged["created"])
|
|
source = manifest["source"]
|
|
hashes = bundle["hashes"]
|
|
receipt = {
|
|
"schema": SOURCE_PROPOSAL_RECEIPT_SCHEMA,
|
|
"artifact": "teleo_kb_source_proposal",
|
|
"status": staged["status"],
|
|
"created": created,
|
|
"proposal": {
|
|
"id": staged["proposal_id"],
|
|
"proposal_type": staged["proposal_type"],
|
|
"status": staged["status"],
|
|
"source_ref": staged["source_ref"],
|
|
"payload_sha256": staged["payload_sha256"],
|
|
"created_at": staged.get("created_at"),
|
|
"reviewed_at": staged.get("reviewed_at"),
|
|
"applied_at": staged.get("applied_at"),
|
|
},
|
|
"source": {
|
|
"identity": source.get("identity"),
|
|
"locator": source.get("locator"),
|
|
"artifact_sha256": hashes["artifact_sha256"],
|
|
"extracted_text_sha256": hashes["extracted_text_sha256"],
|
|
"manifest_canonical_sha256": hashes["manifest_canonical_sha256"],
|
|
},
|
|
"compiler": {
|
|
"path": str(SOURCE_COMPILER_PATH),
|
|
"sha256": compiler_sha256,
|
|
"bundle_content_sha256": hashes["bundle_content_sha256"],
|
|
},
|
|
"database": {
|
|
"backend": after.get("backend"),
|
|
"before_counts": before_counts,
|
|
"after_counts": after_counts,
|
|
"canonical_counts_unchanged": canonical_counts_unchanged,
|
|
"proposal_count_delta": proposal_delta,
|
|
"proposal_count_matches_command": proposal_delta == (1 if created else 0),
|
|
},
|
|
"write_contract": {
|
|
"target_table": "kb_stage.kb_proposals",
|
|
"database_write_performed_by_command": created,
|
|
"canonical_apply_performed": False,
|
|
"production_db_apply_ran": False,
|
|
"public_table_write_path_present": False,
|
|
},
|
|
"receipt_path": str(receipt_path),
|
|
}
|
|
_write_private_json(receipt_path, receipt)
|
|
return receipt
|
|
|
|
|
|
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]:
|
|
contracts = context_operational_contracts(args, query)
|
|
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": contracts,
|
|
"compiled_response": compile_operational_response(contracts),
|
|
"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 print_source_proposal_receipt(data: dict[str, Any]) -> None:
|
|
proposal = data["proposal"]
|
|
database = data["database"]
|
|
print("# Source Proposal Receipt\n")
|
|
print(f"- proposal: `{proposal['id']}`")
|
|
print(f"- status: `{proposal['status']}`; created: `{data['created']}`")
|
|
print(f"- source_ref: `{proposal['source_ref']}`")
|
|
print(f"- canonical counts unchanged: `{database['canonical_counts_unchanged']}`")
|
|
print(f"- proposal count delta: `{database['proposal_count_delta']}`")
|
|
print("- canonical apply performed: `false`")
|
|
print(f"- receipt: `{data['receipt_path']}`")
|
|
|
|
|
|
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 == "propose-source":
|
|
data = propose_source(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 == "propose-source":
|
|
print_source_proposal_receipt(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())
|