2015 lines
78 KiB
Python
Executable file
2015 lines
78 KiB
Python
Executable file
#!/usr/bin/python3 -I
|
|
"""Cloud SQL memory and KB-proposal bridge for the GCP leoclean profile.
|
|
|
|
Canonical claim/strategy application remains review-gated. The production-like
|
|
Leo runtime is pinned to the private canonical Cloud SQL database and never
|
|
falls back to the restored `teleo_restore` shadow schema. Proposal commands
|
|
write staged review records into `kb_stage.kb_proposals`; they do not apply
|
|
canonical truth directly.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import subprocess
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
STOPWORDS = {
|
|
"a",
|
|
"about",
|
|
"after",
|
|
"all",
|
|
"an",
|
|
"and",
|
|
"are",
|
|
"as",
|
|
"at",
|
|
"be",
|
|
"before",
|
|
"by",
|
|
"can",
|
|
"cloud",
|
|
"context",
|
|
"do",
|
|
"does",
|
|
"for",
|
|
"from",
|
|
"gcp",
|
|
"give",
|
|
"how",
|
|
"i",
|
|
"in",
|
|
"into",
|
|
"is",
|
|
"it",
|
|
"leo",
|
|
"me",
|
|
"memory",
|
|
"of",
|
|
"on",
|
|
"or",
|
|
"our",
|
|
"production",
|
|
"should",
|
|
"source",
|
|
"that",
|
|
"the",
|
|
"their",
|
|
"this",
|
|
"to",
|
|
"using",
|
|
"what",
|
|
"when",
|
|
"where",
|
|
"which",
|
|
"who",
|
|
"why",
|
|
"with",
|
|
"you",
|
|
"your",
|
|
}
|
|
|
|
DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz"
|
|
RETRIEVAL_RECEIPT_SCHEMA = "livingip.teleoKbRetrievalReceipt.v1"
|
|
RUNTIME_DB_USER = "leoclean_kb_runtime"
|
|
RUNTIME_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-leoclean-kb-runtime-password"
|
|
RUNTIME_GCP_PROJECT = "teleo-501523"
|
|
RUNTIME_CLOUDSQL_HOST = "10.61.0.3"
|
|
RUNTIME_CLOUDSQL_PORT = "5432"
|
|
RUNTIME_CLOUDSQL_DB = "teleo_canonical"
|
|
RUNTIME_SYSTEM_IDENTIFIER = "7659718422914359312"
|
|
RUNTIME_CLOUDSDK_CONFIG = "/usr/local/libexec/livingip/leoclean-kb/gcloud-config"
|
|
RUNTIME_PSQL_BIN = "/usr/bin/psql"
|
|
RUNTIME_SSL_MODE = "verify-ca"
|
|
RUNTIME_SSL_ROOT_CERT = "/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem"
|
|
RUNTIME_SSL_ROOT_CERT_SHA256 = "80701e768f0e1f6b9d621aa0b53f6e851daaa276c6d9a8e51a300fbc015539cb"
|
|
RUNTIME_SERVICE_ACCOUNT = "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com"
|
|
GCE_METADATA_EMAIL_URL = "http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/email"
|
|
GCE_METADATA_TOKEN_URL = "http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/token"
|
|
RUNTIME_SECRET_ACCESS_URL = (
|
|
"https://secretmanager.googleapis.com/v1/projects/teleo-501523/secrets/"
|
|
"gcp-teleo-pgvector-standby-leoclean-kb-runtime-password/versions/latest:access"
|
|
)
|
|
METADATA_HEADERS = {"Metadata-Flavor": "Google"}
|
|
HTTP_TIMEOUT_SECONDS = 3.0
|
|
MAX_METADATA_EMAIL_BYTES = 256
|
|
MAX_METADATA_TOKEN_BYTES = 4096
|
|
MAX_SECRET_RESPONSE_BYTES = 8192
|
|
MAX_RUNTIME_SECRET_BYTES = 4096
|
|
RUNTIME_TLS_TRUST_ENVIRONMENT = frozenset(
|
|
{
|
|
"CURL_CA_BUNDLE",
|
|
"PYTHONHTTPSVERIFY",
|
|
"REQUESTS_CA_BUNDLE",
|
|
"SSL_CERT_DIR",
|
|
"SSL_CERT_FILE",
|
|
"SSLKEYLOGFILE",
|
|
"OPENSSL_CONF",
|
|
"OPENSSL_ENGINES",
|
|
"OPENSSL_MODULES",
|
|
"GCONV_PATH",
|
|
}
|
|
)
|
|
RUNTIME_PSQL_ENV_BASE = {
|
|
"PATH": "/usr/bin:/bin",
|
|
"PGSSLMODE": RUNTIME_SSL_MODE,
|
|
"PGCONNECT_TIMEOUT": "8",
|
|
}
|
|
|
|
|
|
def runtime_ssl_root_cert() -> str:
|
|
"""Return the reviewed installed or ephemeral preflight CA path."""
|
|
|
|
raw = os.environ.get("TELEO_GCP_PREFLIGHT_SSL_ROOT_CERT", RUNTIME_SSL_ROOT_CERT)
|
|
candidate = Path(raw)
|
|
try:
|
|
if not candidate.is_absolute() or candidate.is_symlink():
|
|
raise OSError
|
|
resolved = candidate.resolve(strict=True)
|
|
if resolved != candidate or not resolved.is_file():
|
|
raise OSError
|
|
for part in (resolved, *resolved.parents):
|
|
metadata = part.stat()
|
|
if metadata.st_uid != 0 or metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
|
|
raise OSError
|
|
certificate = resolved.read_bytes()
|
|
except OSError:
|
|
raise SystemExit("Reviewed Cloud SQL server CA path is unavailable or untrusted.") from None
|
|
if hashlib.sha256(certificate).hexdigest() != RUNTIME_SSL_ROOT_CERT_SHA256:
|
|
raise SystemExit("Reviewed Cloud SQL server CA does not match the pinned certificate.")
|
|
return str(resolved)
|
|
|
|
|
|
def runtime_cloudsdk_config() -> str:
|
|
"""Return a root-owned empty SDK configuration directory."""
|
|
|
|
raw = os.environ.get("TELEO_GCP_PREFLIGHT_CLOUDSDK_CONFIG", RUNTIME_CLOUDSDK_CONFIG)
|
|
candidate = Path(raw)
|
|
try:
|
|
if not candidate.is_absolute() or candidate.is_symlink():
|
|
raise OSError
|
|
resolved = candidate.resolve(strict=True)
|
|
if resolved != candidate or not resolved.is_dir() or any(resolved.iterdir()):
|
|
raise OSError
|
|
for part in (resolved, *resolved.parents):
|
|
metadata = part.stat()
|
|
if metadata.st_uid != 0 or metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
|
|
raise OSError
|
|
except OSError:
|
|
raise SystemExit("Reviewed empty Cloud SDK configuration is unavailable or untrusted.") from None
|
|
return str(resolved)
|
|
CLONE_READ_ONLY_PGOPTIONS = "-c default_transaction_read_only=on"
|
|
RECEIPTED_READ_COMMANDS = frozenset(
|
|
{
|
|
"search",
|
|
"context",
|
|
"show",
|
|
"evidence",
|
|
"edges",
|
|
"status",
|
|
"list-proposals",
|
|
"search-proposals",
|
|
"show-proposal",
|
|
"decision-matrix-status",
|
|
}
|
|
)
|
|
|
|
|
|
class _RejectRedirectHandler(urllib.request.HTTPRedirectHandler):
|
|
def redirect_request(self, *_args: Any, **_kwargs: Any) -> None:
|
|
return None
|
|
|
|
|
|
RUNTIME_PROXY_HANDLER = urllib.request.ProxyHandler({})
|
|
RUNTIME_HTTP_OPENER = urllib.request.build_opener(
|
|
RUNTIME_PROXY_HANDLER,
|
|
_RejectRedirectHandler(),
|
|
)
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--host", default=os.environ.get("TELEO_CLOUDSQL_HOST", "10.61.0.3"))
|
|
parser.add_argument("--port", default=os.environ.get("TELEO_CLOUDSQL_PORT", "5432"))
|
|
parser.add_argument("--db", default=os.environ.get("TELEO_CLOUDSQL_DB", "teleo_canonical"))
|
|
parser.add_argument("--canonical-db", default=os.environ.get("TELEO_CANONICAL_CLOUDSQL_DB", "teleo_canonical"))
|
|
parser.add_argument("--user", default=os.environ.get("TELEO_CLOUDSQL_USER"))
|
|
parser.add_argument(
|
|
"--password-secret",
|
|
default=os.environ.get("TELEO_CLOUDSQL_PASSWORD_SECRET"),
|
|
)
|
|
parser.add_argument("--project", default=os.environ.get("TELEO_GCP_PROJECT", "teleo-501523"))
|
|
parser.add_argument(
|
|
"--credential-mode",
|
|
choices=["runtime", "clone-readonly"],
|
|
default=os.environ.get("TELEO_CLOUDSQL_CREDENTIAL_MODE", "runtime"),
|
|
help="Runtime mode requires the exact scoped Leo identity. Clone mode is limited to read-only teleo_clone_* tests.",
|
|
)
|
|
parser.add_argument("--format", choices=["markdown", "json"], default="markdown")
|
|
parser.add_argument(
|
|
"--include-excerpts",
|
|
"--raw-ok",
|
|
dest="include_excerpts",
|
|
action="store_true",
|
|
default=os.environ.get("TELEO_MEMORY_INCLUDE_EXCERPTS", "") == "1",
|
|
help="Include short memory excerpts for agent context. Default proof mode is redacted.",
|
|
)
|
|
parser.add_argument("--redacted", action="store_true", default=os.environ.get("TELEO_MEMORY_REDACT", "") == "1")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
search = sub.add_parser("search", help="Search restored Cloud SQL memory rows.")
|
|
search.add_argument("query")
|
|
search.add_argument("--limit", type=int, default=8)
|
|
search.add_argument(
|
|
"--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored."
|
|
)
|
|
search.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
search.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
|
|
search.add_argument(
|
|
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
|
|
)
|
|
|
|
context = sub.add_parser("context", help="Build an agent-ready memory context bundle.")
|
|
context.add_argument("query")
|
|
context.add_argument("--limit", type=int, default=8)
|
|
context.add_argument(
|
|
"--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored."
|
|
)
|
|
context.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
context.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
|
|
context.add_argument(
|
|
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
|
|
)
|
|
|
|
show = sub.add_parser("show", help="Show one canonical claim with evidence and edges.")
|
|
show.add_argument("claim_id")
|
|
show.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
show.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
|
|
show.add_argument(
|
|
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
|
|
)
|
|
|
|
evidence = sub.add_parser("evidence", help="Show canonical evidence rows for one claim.")
|
|
evidence.add_argument("claim_id")
|
|
evidence.add_argument("--limit", type=int, default=8)
|
|
evidence.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
evidence.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
|
|
evidence.add_argument(
|
|
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
|
|
)
|
|
|
|
edges = sub.add_parser("edges", help="Show canonical graph edges for one claim.")
|
|
edges.add_argument("claim_id")
|
|
edges.add_argument("--limit", type=int, default=12)
|
|
edges.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
edges.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
|
|
edges.add_argument(
|
|
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
|
|
)
|
|
|
|
status = sub.add_parser("status", help="Read back Cloud SQL memory backend status.")
|
|
status.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
status.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
|
|
|
|
propose = sub.add_parser(
|
|
"propose-core-change",
|
|
help="Stage a review-gated canonical KB change instead of writing core truth to runtime memory.",
|
|
)
|
|
propose.add_argument("--proposal-type", choices=["revise_claim", "revise_strategy"], required=True)
|
|
propose.add_argument(
|
|
"--target-kind",
|
|
choices=["claim", "belief", "strategy", "identity", "role", "telos", "framework", "reasoning_tool", "other"],
|
|
required=True,
|
|
)
|
|
propose.add_argument(
|
|
"--target-ref", required=True, help="Claim id, strategy id/name, or unresolved target description."
|
|
)
|
|
propose.add_argument("--current", default="", help="Current/old claim or state, if known.")
|
|
propose.add_argument("--proposed", required=True, help="Proposed replacement or correction.")
|
|
propose.add_argument("--evidence", action="append", default=[], help="Evidence/source pointer. Can be repeated.")
|
|
propose.add_argument("--implication", action="append", default=[], help="Downstream implication. Can be repeated.")
|
|
propose.add_argument("--proposed-by", choices=["leo"], default="leo")
|
|
propose.add_argument("--originator", default="", help="Human or agent who originated the correction.")
|
|
propose.add_argument("--channel", default="telegram")
|
|
propose.add_argument("--source-ref", required=True)
|
|
propose.add_argument("--rationale", required=True)
|
|
propose.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
|
|
propose_edge = sub.add_parser("propose-edge", help="Stage a strict add_edge proposal 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", choices=["leo"], default="leo")
|
|
propose_edge.add_argument("--channel", default="telegram")
|
|
propose_edge.add_argument("--source-ref", required=True)
|
|
propose_edge.add_argument("--rationale", required=True)
|
|
propose_edge.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
|
|
list_proposals = sub.add_parser("list-proposals", help="List staged KB proposals from Cloud SQL.")
|
|
list_proposals.add_argument("--status", default="pending_review")
|
|
list_proposals.add_argument("--limit", type=int, default=10)
|
|
list_proposals.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
|
|
search_proposals = sub.add_parser(
|
|
"search-proposals", help="Search staged KB proposals across statuses from Cloud SQL."
|
|
)
|
|
search_proposals.add_argument("query")
|
|
search_proposals.add_argument("--status", default="all")
|
|
search_proposals.add_argument("--limit", type=int, default=10)
|
|
search_proposals.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
|
|
show_proposal = sub.add_parser("show-proposal", help="Show one staged KB proposal from Cloud SQL.")
|
|
show_proposal.add_argument("proposal_id")
|
|
show_proposal.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
|
|
decision_matrix_status = sub.add_parser(
|
|
"decision-matrix-status",
|
|
help="Read whether decision-matrix approval tables exist before claiming matrix approval.",
|
|
)
|
|
decision_matrix_status.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
def validate_runtime_connection(args: argparse.Namespace) -> None:
|
|
user = str(args.user or "")
|
|
password_secret = str(args.password_secret or "")
|
|
mode = str(getattr(args, "credential_mode", "runtime"))
|
|
if mode == "runtime":
|
|
ssl_root_cert = runtime_ssl_root_cert()
|
|
cloudsdk_config = runtime_cloudsdk_config()
|
|
if user != RUNTIME_DB_USER:
|
|
raise SystemExit(f"Leo runtime requires database user {RUNTIME_DB_USER}.")
|
|
if password_secret != RUNTIME_PASSWORD_SECRET:
|
|
raise SystemExit(f"Leo runtime requires scoped password secret {RUNTIME_PASSWORD_SECRET}.")
|
|
expected_fields = {
|
|
"project": RUNTIME_GCP_PROJECT,
|
|
"host": RUNTIME_CLOUDSQL_HOST,
|
|
"port": RUNTIME_CLOUDSQL_PORT,
|
|
"db": RUNTIME_CLOUDSQL_DB,
|
|
"canonical_db": RUNTIME_CLOUDSQL_DB,
|
|
}
|
|
for field, expected in expected_fields.items():
|
|
actual = str(getattr(args, field, "") or "")
|
|
if actual != expected:
|
|
raise SystemExit(f"Leo runtime requires {field}={expected}.")
|
|
if os.environ.get("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "0") != "0":
|
|
raise SystemExit("Leo runtime requires TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK=0.")
|
|
if os.environ.get("TELEO_GCP_METADATA_ONLY") != "1":
|
|
raise SystemExit("Leo runtime requires TELEO_GCP_METADATA_ONLY=1.")
|
|
forbidden_environment = any(
|
|
(
|
|
normalized_key.startswith("PG")
|
|
and not (
|
|
(normalized_key == "PGSSLMODE" and value == RUNTIME_SSL_MODE)
|
|
or (normalized_key == "PGSSLROOTCERT" and value == ssl_root_cert)
|
|
)
|
|
)
|
|
or (
|
|
normalized_key.startswith("CLOUDSDK_")
|
|
and not (
|
|
normalized_key == "CLOUDSDK_CONFIG"
|
|
and value == cloudsdk_config
|
|
)
|
|
)
|
|
or normalized_key == "GOOGLE_APPLICATION_CREDENTIALS"
|
|
or normalized_key.startswith("LD_")
|
|
or normalized_key.startswith("BASH_FUNC_")
|
|
or (
|
|
normalized_key.startswith("PYTHON")
|
|
and not (
|
|
(normalized_key == "PYTHONNOUSERSITE" and value == "1")
|
|
or (normalized_key == "PYTHONSAFEPATH" and value == "1")
|
|
or normalized_key == "PYTHONUNBUFFERED"
|
|
)
|
|
)
|
|
or normalized_key in {
|
|
"BASH_ENV",
|
|
"BASHOPTS",
|
|
"CDPATH",
|
|
"ENV",
|
|
"GLOBIGNORE",
|
|
"IFS",
|
|
"PS4",
|
|
"SHELLOPTS",
|
|
}
|
|
or normalized_key == "PROXY"
|
|
or normalized_key.endswith("_PROXY")
|
|
or normalized_key in RUNTIME_TLS_TRUST_ENVIRONMENT
|
|
for key, value in os.environ.items()
|
|
for normalized_key in (key.upper(),)
|
|
)
|
|
if forbidden_environment:
|
|
raise SystemExit(
|
|
"Refusing inherited credential, proxy, TLS, or PostgreSQL environment in Leo runtime mode."
|
|
)
|
|
return
|
|
|
|
if mode != "clone-readonly":
|
|
raise SystemExit(f"Unsupported Cloud SQL credential mode: {mode}.")
|
|
canonical_db = str(args.canonical_db or "")
|
|
audit_db = str(args.db or "")
|
|
pgoptions = os.environ.get("PGOPTIONS", "")
|
|
if not re.fullmatch(r"teleo_clone_[a-z0-9_]+", canonical_db) or audit_db != canonical_db:
|
|
raise SystemExit("Clone credential mode is limited to one teleo_clone_* database.")
|
|
if args.command not in RECEIPTED_READ_COMMANDS:
|
|
raise SystemExit("Clone credential mode permits read commands only.")
|
|
if pgoptions != CLONE_READ_ONLY_PGOPTIONS:
|
|
raise SystemExit("Clone credential mode requires server-enforced default_transaction_read_only=on.")
|
|
if not user or not os.environ.get("PGPASSWORD"):
|
|
raise SystemExit("Clone credential mode requires an explicit user and inherited test credential.")
|
|
|
|
|
|
def terms_for(query: str) -> list[str]:
|
|
terms: list[str] = []
|
|
for token in re.findall(r"[A-Za-z0-9][A-Za-z0-9.+#/-]*", query.lower()):
|
|
token = token.strip("-/")
|
|
if len(token) < 3 or token in STOPWORDS:
|
|
continue
|
|
if token not in terms:
|
|
terms.append(token)
|
|
return terms[:14] or [query.lower()]
|
|
|
|
|
|
def 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 sql_literal(value: str) -> str:
|
|
return "'" + value.replace("'", "''") + "'"
|
|
|
|
|
|
def sql_json_array(values: list[str]) -> str:
|
|
return "'{}'::json".format(json.dumps(values).replace("'", "''"))
|
|
|
|
|
|
def sql_array(values: list[str], cast: str = "text") -> str:
|
|
return "array[" + ",".join(sql_literal(value) for value in values) + f"]::{cast}[]"
|
|
|
|
|
|
def _read_http_response(request: urllib.request.Request, *, max_bytes: int, purpose: str) -> bytes:
|
|
try:
|
|
with RUNTIME_HTTP_OPENER.open(request, timeout=HTTP_TIMEOUT_SECONDS) as response:
|
|
if response.getcode() != 200:
|
|
raise SystemExit(f"{purpose} request failed.")
|
|
body = response.read(max_bytes + 1)
|
|
except SystemExit:
|
|
raise
|
|
# This is a credential-bearing boundary. Never propagate provider, proxy,
|
|
# TLS, or test-double exception text because it can contain bearer data.
|
|
except Exception:
|
|
raise SystemExit(f"{purpose} request failed.") from None
|
|
if not isinstance(body, bytes) or len(body) > max_bytes:
|
|
raise SystemExit(f"{purpose} response was invalid.")
|
|
return body
|
|
|
|
|
|
def _strict_json_object(body: bytes, *, purpose: str) -> dict[str, Any]:
|
|
def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
|
|
value: dict[str, Any] = {}
|
|
for key, item in pairs:
|
|
if key in value:
|
|
raise ValueError("duplicate JSON key")
|
|
value[key] = item
|
|
return value
|
|
|
|
try:
|
|
decoded = body.decode("utf-8", errors="strict")
|
|
value = json.loads(
|
|
decoded,
|
|
object_pairs_hook=reject_duplicate_keys,
|
|
parse_constant=lambda _constant: (_ for _ in ()).throw(ValueError("non-finite JSON value")),
|
|
)
|
|
except (UnicodeDecodeError, ValueError, TypeError):
|
|
raise SystemExit(f"{purpose} response was invalid.") from None
|
|
if not isinstance(value, dict):
|
|
raise SystemExit(f"{purpose} response was invalid.")
|
|
return value
|
|
|
|
|
|
def _metadata_request(url: str, *, max_bytes: int, purpose: str) -> bytes:
|
|
request = urllib.request.Request(url, headers=METADATA_HEADERS, method="GET")
|
|
return _read_http_response(request, max_bytes=max_bytes, purpose=purpose)
|
|
|
|
|
|
def runtime_password_from_metadata() -> str:
|
|
email_body = _metadata_request(
|
|
GCE_METADATA_EMAIL_URL,
|
|
max_bytes=MAX_METADATA_EMAIL_BYTES,
|
|
purpose="Runtime metadata identity",
|
|
)
|
|
try:
|
|
service_account = email_body.decode("ascii", errors="strict")
|
|
except UnicodeDecodeError:
|
|
raise SystemExit("Runtime metadata identity response was invalid.") from None
|
|
if service_account != RUNTIME_SERVICE_ACCOUNT:
|
|
raise SystemExit("Runtime metadata identity did not match the required service account.")
|
|
|
|
token_value = _strict_json_object(
|
|
_metadata_request(
|
|
GCE_METADATA_TOKEN_URL,
|
|
max_bytes=MAX_METADATA_TOKEN_BYTES,
|
|
purpose="Runtime metadata token",
|
|
),
|
|
purpose="Runtime metadata token",
|
|
)
|
|
access_token = token_value.get("access_token")
|
|
token_type = token_value.get("token_type")
|
|
expires_in = token_value.get("expires_in")
|
|
if (
|
|
token_type != "Bearer"
|
|
or not isinstance(access_token, str)
|
|
or not re.fullmatch(r"[!-~]{1,4096}", access_token)
|
|
or not isinstance(expires_in, int)
|
|
or isinstance(expires_in, bool)
|
|
or expires_in <= 0
|
|
):
|
|
raise SystemExit("Runtime metadata token response was invalid.")
|
|
|
|
secret_request = urllib.request.Request(
|
|
RUNTIME_SECRET_ACCESS_URL,
|
|
headers={"Accept": "application/json", "Authorization": f"Bearer {access_token}"},
|
|
method="GET",
|
|
)
|
|
secret_value = _strict_json_object(
|
|
_read_http_response(
|
|
secret_request,
|
|
max_bytes=MAX_SECRET_RESPONSE_BYTES,
|
|
purpose="Scoped runtime secret access",
|
|
),
|
|
purpose="Scoped runtime secret access",
|
|
)
|
|
payload = secret_value.get("payload")
|
|
encoded_secret = payload.get("data") if isinstance(payload, dict) else None
|
|
if not isinstance(encoded_secret, str):
|
|
raise SystemExit("Scoped runtime secret response was invalid.")
|
|
try:
|
|
credential_bytes = base64.b64decode(encoded_secret, validate=True)
|
|
except (binascii.Error, ValueError):
|
|
raise SystemExit("Scoped runtime secret response was invalid.") from None
|
|
if not credential_bytes or len(credential_bytes) > MAX_RUNTIME_SECRET_BYTES:
|
|
raise SystemExit("Scoped runtime secret response was invalid.")
|
|
if any(byte in credential_bytes for byte in (0, 10, 13)):
|
|
raise SystemExit("Scoped runtime secret response was invalid.")
|
|
try:
|
|
return credential_bytes.decode("utf-8", errors="strict")
|
|
except UnicodeDecodeError:
|
|
raise SystemExit("Scoped runtime secret response was invalid.") from None
|
|
|
|
|
|
def password(args: argparse.Namespace) -> str:
|
|
validate_runtime_connection(args)
|
|
if args.credential_mode == "runtime":
|
|
return runtime_password_from_metadata()
|
|
if os.environ.get("PGPASSWORD"):
|
|
return os.environ["PGPASSWORD"]
|
|
raise SystemExit("Clone credential mode requires an inherited test credential.")
|
|
|
|
|
|
def run_psql(args: argparse.Namespace, sql: str, db: str | None = None) -> str:
|
|
credential = password(args)
|
|
if args.credential_mode == "runtime":
|
|
env = dict(RUNTIME_PSQL_ENV_BASE)
|
|
env["PGSSLROOTCERT"] = runtime_ssl_root_cert()
|
|
else:
|
|
env = {key: value for key, value in os.environ.items() if not key.startswith("PG")}
|
|
env.update(
|
|
{
|
|
"PGHOST": str(args.host),
|
|
"PGPORT": str(args.port),
|
|
"PGDATABASE": str(db or args.db),
|
|
"PGUSER": str(args.user),
|
|
"PGPASSWORD": credential,
|
|
}
|
|
)
|
|
if args.credential_mode == "clone-readonly":
|
|
env["PGOPTIONS"] = CLONE_READ_ONLY_PGOPTIONS
|
|
result = subprocess.run(
|
|
[RUNTIME_PSQL_BIN, "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"],
|
|
text=True,
|
|
capture_output=True,
|
|
input=sql,
|
|
env=env,
|
|
check=False,
|
|
)
|
|
if result.returncode != 0:
|
|
raise SystemExit(f"Cloud SQL query failed with exit status {result.returncode}.")
|
|
return result.stdout
|
|
|
|
|
|
def psql_json_lines(args: argparse.Namespace, sql: str, db: str | None = None) -> list[dict[str, Any]]:
|
|
rows: list[dict[str, Any]] = []
|
|
for line in run_psql(args, sql, db=db).splitlines():
|
|
line = line.strip()
|
|
if line:
|
|
rows.append(json.loads(line))
|
|
return rows
|
|
|
|
|
|
def canonical_json_sha256(value: Any) -> str:
|
|
encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
|
|
return hashlib.sha256(encoded).hexdigest()
|
|
|
|
|
|
def database_read_marker(args: argparse.Namespace) -> dict[str, Any]:
|
|
sql = """
|
|
select jsonb_build_object(
|
|
'database', current_database(),
|
|
'database_user', current_user,
|
|
'server_addr', inet_server_addr()::text,
|
|
'server_port', inet_server_port(),
|
|
'ssl', coalesce((select ssl from pg_catalog.pg_stat_ssl where pid = pg_backend_pid()), false),
|
|
'ssl_version', (select version from pg_catalog.pg_stat_ssl where pid = pg_backend_pid()),
|
|
'system_identifier', (select system_identifier::text from pg_control_system()),
|
|
'wal_lsn', pg_current_wal_lsn()::text
|
|
)::text;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
if len(rows) != 1:
|
|
raise SystemExit("Canonical database read marker returned an invalid row count.")
|
|
marker = rows[0]
|
|
expected_database = str(args.canonical_db or "")
|
|
expected_user = str(args.user or "")
|
|
if marker.get("database") != expected_database or marker.get("database_user") != expected_user:
|
|
raise SystemExit("Canonical database read marker returned an unexpected database identity.")
|
|
if getattr(args, "credential_mode", "runtime") == "runtime":
|
|
if marker.get("server_addr") != RUNTIME_CLOUDSQL_HOST:
|
|
raise SystemExit("Canonical database read marker returned an unexpected server address.")
|
|
if str(marker.get("server_port")) != RUNTIME_CLOUDSQL_PORT:
|
|
raise SystemExit("Canonical database read marker returned an unexpected server port.")
|
|
if marker.get("ssl") is not True:
|
|
raise SystemExit("Canonical database read marker did not prove an encrypted connection.")
|
|
if marker.get("system_identifier") != RUNTIME_SYSTEM_IDENTIFIER:
|
|
raise SystemExit("Canonical database read marker returned an unexpected system identifier.")
|
|
return marker
|
|
|
|
|
|
def _read_marker_stable(before: dict[str, Any], after: dict[str, Any]) -> bool:
|
|
return all(
|
|
before.get(key) == after.get(key)
|
|
for key in (
|
|
"database",
|
|
"database_user",
|
|
"server_addr",
|
|
"server_port",
|
|
"ssl",
|
|
"ssl_version",
|
|
"system_identifier",
|
|
"wal_lsn",
|
|
)
|
|
)
|
|
|
|
|
|
def _stable_receipt_value(value: Any) -> Any:
|
|
if isinstance(value, dict):
|
|
return {
|
|
key: _stable_receipt_value(item)
|
|
for key, item in sorted(value.items())
|
|
if key not in {"claim_url", "connected_url", "retrieval_receipt"}
|
|
}
|
|
if isinstance(value, list):
|
|
return [_stable_receipt_value(item) for item in value]
|
|
return value
|
|
|
|
|
|
def _retrieved_claim_ids(data: dict[str, Any]) -> list[str]:
|
|
values = []
|
|
if data.get("claim_id"):
|
|
values.append(str(data["claim_id"]))
|
|
claim = data.get("claim") or {}
|
|
if claim.get("id"):
|
|
values.append(str(claim["id"]))
|
|
values.extend(str(row["id"]) for row in data.get("claims", []) if row.get("id"))
|
|
return sorted(set(values))
|
|
|
|
|
|
def _retrieved_evidence(data: dict[str, Any]) -> list[dict[str, Any]]:
|
|
rows = [row for row in data.get("evidence", []) if isinstance(row, dict)]
|
|
for claim in data.get("claims", []):
|
|
rows.extend(row for row in claim.get("evidence", []) if isinstance(row, dict))
|
|
return rows
|
|
|
|
|
|
def build_retrieval_receipt(
|
|
data: dict[str, Any],
|
|
*,
|
|
query_descriptor: dict[str, Any],
|
|
before: dict[str, Any],
|
|
after: dict[str, Any],
|
|
attempts: int,
|
|
consistency_status: str,
|
|
) -> dict[str, Any]:
|
|
evidence = _retrieved_evidence(data)
|
|
artifact_states = sorted(
|
|
(
|
|
{
|
|
"source_id": row.get("source_id"),
|
|
"source_hash": row.get("source_hash"),
|
|
"storage_path": row.get("storage_path"),
|
|
}
|
|
for row in evidence
|
|
),
|
|
key=lambda row: (str(row.get("source_id") or ""), str(row.get("storage_path") or "")),
|
|
)
|
|
return {
|
|
"schema": RETRIEVAL_RECEIPT_SCHEMA,
|
|
"query_sha256": canonical_json_sha256(query_descriptor),
|
|
"semantic_context_sha256": canonical_json_sha256(_stable_receipt_value(data)),
|
|
"artifact_state_sha256": canonical_json_sha256(artifact_states),
|
|
"claim_ids": _retrieved_claim_ids(data),
|
|
"source_ids": sorted({str(row["source_id"]) for row in evidence if row.get("source_id")}),
|
|
"counts": {
|
|
"claims": len(_retrieved_claim_ids(data)),
|
|
"evidence_rows": len(evidence),
|
|
"edge_rows": len(data.get("edges", []))
|
|
+ sum(len(claim.get("edges", [])) for claim in data.get("claims", [])),
|
|
},
|
|
"read_consistency": {
|
|
"status": consistency_status,
|
|
"attempts": attempts,
|
|
"database": before.get("database"),
|
|
"database_user": before.get("database_user"),
|
|
"server_addr": before.get("server_addr"),
|
|
"server_port": before.get("server_port"),
|
|
"ssl": before.get("ssl"),
|
|
"ssl_version": before.get("ssl_version"),
|
|
"system_identifier": before.get("system_identifier"),
|
|
"wal_lsn_before": before.get("wal_lsn"),
|
|
"wal_lsn_after": after.get("wal_lsn"),
|
|
},
|
|
}
|
|
|
|
|
|
def receipt_query_descriptor(args: argparse.Namespace) -> dict[str, Any]:
|
|
return {
|
|
"command": args.command,
|
|
"query": getattr(args, "query", None),
|
|
"claim_id": getattr(args, "claim_id", None),
|
|
"limit": getattr(args, "limit", None),
|
|
"context_limit": getattr(args, "context_limit", None),
|
|
}
|
|
|
|
|
|
def read_with_retrieval_receipt(args: argparse.Namespace, loader: Any) -> dict[str, Any]:
|
|
try:
|
|
max_attempts = max(1, min(int(os.environ.get("TELEO_KB_READ_ATTEMPTS", "3")), 5))
|
|
except ValueError as exc:
|
|
raise SystemExit("TELEO_KB_READ_ATTEMPTS must be an integer") from exc
|
|
previous_semantic_sha256: str | None = None
|
|
last_markers: tuple[dict[str, Any], dict[str, Any]] | None = None
|
|
descriptor = receipt_query_descriptor(args)
|
|
for attempt in range(1, max_attempts + 1):
|
|
before = database_read_marker(args)
|
|
data = loader()
|
|
after = database_read_marker(args)
|
|
last_markers = (before, after)
|
|
consistency_status = "stable_wal_marker" if _read_marker_stable(before, after) else "wal_changed_during_read"
|
|
receipt = build_retrieval_receipt(
|
|
data,
|
|
query_descriptor=descriptor,
|
|
before=before,
|
|
after=after,
|
|
attempts=attempt,
|
|
consistency_status=consistency_status,
|
|
)
|
|
if consistency_status == "stable_wal_marker":
|
|
data["retrieval_receipt"] = receipt
|
|
return data
|
|
semantic_sha256 = receipt["semantic_context_sha256"]
|
|
if previous_semantic_sha256 == semantic_sha256:
|
|
receipt["read_consistency"]["status"] = "stable_content_across_wal_change_retry"
|
|
data["retrieval_receipt"] = receipt
|
|
return data
|
|
previous_semantic_sha256 = semantic_sha256
|
|
assert last_markers is not None
|
|
before, after = last_markers
|
|
raise SystemExit(
|
|
"Canonical DB changed during retrieval and no identical retry was observed "
|
|
f"(wal_lsn {before.get('wal_lsn')} -> {after.get('wal_lsn')})."
|
|
)
|
|
|
|
|
|
def include_excerpts(args: argparse.Namespace) -> bool:
|
|
return bool(getattr(args, "include_excerpts", False)) and not bool(getattr(args, "redacted", False))
|
|
|
|
|
|
def query_canonical_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, Any]:
|
|
terms = terms_for(query)
|
|
patterns = [f"%{term}%" for term in terms]
|
|
min_score = 2 if len(patterns) >= 3 else 1
|
|
claim_limit = max(1, min(limit, 20))
|
|
context_limit = max(1, min(int(getattr(args, "context_limit", limit) or limit), 20))
|
|
|
|
claims_sql = f"""
|
|
with params(patterns) as (
|
|
values ({sql_array(patterns)})
|
|
), ranked as (
|
|
select c.id,
|
|
c.type,
|
|
c.text,
|
|
c.status,
|
|
c.confidence,
|
|
c.tags,
|
|
(
|
|
select count(*)
|
|
from params p, unnest(p.patterns) pattern
|
|
where lower(c.text) like pattern
|
|
or exists (
|
|
select 1 from unnest(coalesce(c.tags, '{{}}'::text[])) tag
|
|
where lower(tag) like pattern
|
|
)
|
|
) as score,
|
|
(select count(*) from public.claim_evidence ce where ce.claim_id = c.id) as evidence_count,
|
|
(select count(*) from public.claim_edges e where e.from_claim = c.id or e.to_claim = c.id) as edge_count
|
|
from public.claims c
|
|
where c.status = 'open'
|
|
)
|
|
select jsonb_build_object(
|
|
'id', id::text,
|
|
'type', type,
|
|
'text', text,
|
|
'status', status,
|
|
'confidence', confidence,
|
|
'tags', coalesce(to_jsonb(tags), '[]'::jsonb),
|
|
'score', score,
|
|
'evidence_count', evidence_count,
|
|
'edge_count', edge_count
|
|
)::text
|
|
from ranked
|
|
where score >= {min_score}
|
|
order by score desc, evidence_count desc, edge_count desc, coalesce(confidence, 0) desc, length(text)
|
|
limit {claim_limit};
|
|
"""
|
|
context_sql = f"""
|
|
with params(patterns) as (
|
|
values ({sql_array(patterns)})
|
|
), corpus as (
|
|
select 'persona' as source,
|
|
a.handle as owner,
|
|
p.name as title,
|
|
concat_ws(E'\\n', p.voice, p.role, p.source_ref, p.lens) as body
|
|
from public.personas p
|
|
join public.agents a on a.id = p.agent_id
|
|
union all
|
|
select 'strategy' as source,
|
|
a.handle as owner,
|
|
'active strategy v' || s.version::text as title,
|
|
concat_ws(E'\\n', s.diagnosis, s.guiding_policy, s.proximate_objectives::text) as body
|
|
from public.strategies s
|
|
join public.agents a on a.id = s.agent_id
|
|
where s.active
|
|
union all
|
|
select 'belief' as source,
|
|
a.handle as owner,
|
|
concat_ws(' ', b.level::text, 'rank', b.rank::text) as title,
|
|
concat_ws(E'\\n', b.statement, b.falsifier) as body
|
|
from public.beliefs b
|
|
join public.agents a on a.id = b.agent_id
|
|
where b.status = 'active'
|
|
union all
|
|
select 'blindspot' as source,
|
|
a.handle as owner,
|
|
bs.name as title,
|
|
concat_ws(E'\\n', bs.description, bs.correction, bs.kind) as body
|
|
from public.blindspots bs
|
|
join public.agents a on a.id = bs.agent_id
|
|
where bs.status = 'active'
|
|
union all
|
|
select 'agent_role' as source,
|
|
a.handle as owner,
|
|
ar.title as title,
|
|
ar.description as body
|
|
from public.agent_roles ar
|
|
join public.agents a on a.id = ar.agent_id
|
|
union all
|
|
select 'peer_model' as source,
|
|
subject.handle as owner,
|
|
'peer ' || peer.handle || ': ' || pm.domain as title,
|
|
concat_ws(E'\\n', pm.outranks_on, pm.deference_rule) as body
|
|
from public.peer_models pm
|
|
join public.agents subject on subject.id = pm.subject_id
|
|
join public.agents peer on peer.id = pm.peer_id
|
|
union all
|
|
select 'behavioral_rule' as source,
|
|
coalesce(a.handle, 'collective') as owner,
|
|
br.category::text as title,
|
|
concat_ws(E'\\n', br.rule, br.rationale) as body
|
|
from public.behavioral_rules br
|
|
left join public.agents a on a.id = br.agent_id
|
|
union all
|
|
select 'contributor_rule' as source,
|
|
coalesce(a.handle, 'collective') as owner,
|
|
cr.name as title,
|
|
concat_ws(E'\\n', cr.directive, cr.ci_tier, cr.weighting, cr.rationale) as body
|
|
from public.contributor_rules cr
|
|
left join public.agents a on a.id = cr.agent_id
|
|
union all
|
|
select 'reasoning_tool' as source,
|
|
coalesce(a.handle, 'collective') as owner,
|
|
rt.name as title,
|
|
concat_ws(E'\\n', rt.description, rt.category) as body
|
|
from public.reasoning_tools rt
|
|
left join public.agents a on a.id = rt.agent_id
|
|
union all
|
|
select 'governance_gate' as source,
|
|
coalesce(a.handle, 'collective') as owner,
|
|
gg.name as title,
|
|
concat_ws(E'\\n', gg.criteria, gg.evidence_bar, gg.pass_condition) as body
|
|
from public.governance_gates gg
|
|
left join public.agents a on a.id = gg.agent_id
|
|
), ranked as (
|
|
select source,
|
|
owner,
|
|
title,
|
|
body,
|
|
(
|
|
select count(*)
|
|
from params p, unnest(p.patterns) pattern
|
|
where lower(concat_ws(E'\\n', source, owner, title, body)) like pattern
|
|
) as score
|
|
from corpus
|
|
)
|
|
select jsonb_build_object(
|
|
'source', source,
|
|
'owner', owner,
|
|
'title', title,
|
|
'body', left(coalesce(body, ''), 900),
|
|
'score', score
|
|
)::text
|
|
from ranked
|
|
where score >= {min_score}
|
|
order by score desc, source, owner, title
|
|
limit {context_limit};
|
|
"""
|
|
claims = psql_json_lines(args, claims_sql, db=args.canonical_db)
|
|
claim_ids = [claim["id"] for claim in claims]
|
|
evidence: dict[str, list[dict[str, Any]]] = {claim_id: [] for claim_id in claim_ids}
|
|
edges: dict[str, list[dict[str, Any]]] = {claim_id: [] for claim_id in claim_ids}
|
|
if claim_ids:
|
|
ids = sql_array(claim_ids, "uuid")
|
|
evidence_sql = f"""
|
|
with ranked as (
|
|
select ce.claim_id,
|
|
s.id as source_id,
|
|
ce.role::text as role,
|
|
ce.weight,
|
|
s.source_type,
|
|
s.url,
|
|
s.storage_path,
|
|
s.hash as source_hash,
|
|
left(coalesce(s.excerpt, ''), 800) as excerpt,
|
|
row_number() over (
|
|
partition by ce.claim_id
|
|
order by (s.storage_path like 'inbox/archive/%') desc,
|
|
ce.role::text,
|
|
s.storage_path nulls last,
|
|
s.url nulls last
|
|
) as rn
|
|
from public.claim_evidence ce
|
|
join public.sources s on s.id = ce.source_id
|
|
where ce.claim_id = any({ids})
|
|
)
|
|
select jsonb_build_object(
|
|
'claim_id', claim_id::text,
|
|
'source_id', source_id::text,
|
|
'role', role,
|
|
'weight', weight,
|
|
'source_type', source_type,
|
|
'url', url,
|
|
'storage_path', storage_path,
|
|
'source_hash', source_hash,
|
|
'excerpt', excerpt
|
|
)::text
|
|
from ranked
|
|
where rn <= 4
|
|
order by claim_id, rn;
|
|
"""
|
|
for row in psql_json_lines(args, evidence_sql, db=args.canonical_db):
|
|
evidence.setdefault(row["claim_id"], []).append(row)
|
|
|
|
edges_sql = f"""
|
|
with base(id) as (
|
|
select unnest({ids})
|
|
), edge_rows as (
|
|
select base.id as claim_id,
|
|
case when e.from_claim = base.id then 'outgoing' else 'incoming' end as direction,
|
|
e.edge_type::text as edge_type,
|
|
other.id as connected_id,
|
|
other.text as connected_text,
|
|
other.tags as connected_tags,
|
|
row_number() over (
|
|
partition by base.id
|
|
order by case e.edge_type::text
|
|
when 'supports' then 1
|
|
when 'challenges' then 2
|
|
when 'contradicts' then 3
|
|
when 'requires' then 4
|
|
when 'supersedes' then 5
|
|
when 'constrains' then 6
|
|
when 'causes' then 7
|
|
when 'accelerates' then 8
|
|
else 9
|
|
end,
|
|
other.text
|
|
) as rn
|
|
from base
|
|
join public.claim_edges e on e.from_claim = base.id or e.to_claim = base.id
|
|
join public.claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end
|
|
)
|
|
select jsonb_build_object(
|
|
'claim_id', claim_id::text,
|
|
'direction', direction,
|
|
'edge_type', edge_type,
|
|
'connected_id', connected_id::text,
|
|
'connected_text', connected_text,
|
|
'connected_tags', coalesce(to_jsonb(connected_tags), '[]'::jsonb)
|
|
)::text
|
|
from edge_rows
|
|
where rn <= 6
|
|
order by claim_id, rn;
|
|
"""
|
|
for row in psql_json_lines(args, edges_sql, db=args.canonical_db):
|
|
edges.setdefault(row["claim_id"], []).append(row)
|
|
|
|
for claim in claims:
|
|
claim["claim_url"] = claim_url(claim["id"])
|
|
claim["evidence"] = evidence.get(claim["id"], [])
|
|
claim["edges"] = edges.get(claim["id"], [])
|
|
for edge in claim["edges"]:
|
|
edge["connected_url"] = claim_url(edge.get("connected_id"))
|
|
|
|
context_rows = psql_json_lines(args, context_sql, db=args.canonical_db)
|
|
result: dict[str, Any] = {
|
|
"artifact": "teleo_cloudsql_canonical_kb_context",
|
|
"backend": f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage",
|
|
"query": query,
|
|
"terms": terms,
|
|
"claims": claims,
|
|
"context_rows": context_rows,
|
|
"hit_count_total": len(claims) + len(context_rows),
|
|
"hits": [
|
|
{
|
|
"source_table": "public.claims",
|
|
"row_id": claim["id"],
|
|
"score": claim.get("score"),
|
|
"claim_type": claim.get("type"),
|
|
"claim_status": claim.get("status"),
|
|
"confidence": claim.get("confidence"),
|
|
"tags": claim.get("tags") or [],
|
|
"evidence_count": claim.get("evidence_count"),
|
|
"edge_count": claim.get("edge_count"),
|
|
"excerpt": claim.get("text"),
|
|
}
|
|
for claim in claims
|
|
]
|
|
+ [
|
|
{
|
|
"source_table": f"public.{row['source']}",
|
|
"row_id": f"{row['owner']}:{row['title']}",
|
|
"score": row.get("score"),
|
|
"actor": row.get("owner"),
|
|
"excerpt": row.get("body"),
|
|
}
|
|
for row in context_rows
|
|
],
|
|
"privacy": "redacted proof mode by default; pass --include-excerpts/--raw-ok only for private agent context",
|
|
}
|
|
if not include_excerpts(args):
|
|
for claim in result["claims"]:
|
|
claim.pop("text", None)
|
|
for row in claim.get("evidence", []):
|
|
row.pop("excerpt", None)
|
|
for row in claim.get("edges", []):
|
|
row.pop("connected_text", None)
|
|
for row in result["context_rows"]:
|
|
row.pop("body", None)
|
|
for hit in result["hits"]:
|
|
hit.pop("excerpt", None)
|
|
result["privacy"] = (
|
|
"redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, canonical table, row id, timestamps, and evidence links"
|
|
)
|
|
else:
|
|
result["privacy"] = (
|
|
"private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts"
|
|
)
|
|
return result
|
|
|
|
|
|
def query_audit_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, Any]:
|
|
terms = terms_for(query)
|
|
sql = f"""
|
|
with q(term) as (
|
|
select lower(value) from json_array_elements_text({sql_json_array(terms)}) as value
|
|
), raw_hits as (
|
|
select
|
|
'response_audit'::text as source_table,
|
|
id::text as row_id,
|
|
coalesce(timestamp, created_at) as event_time,
|
|
coalesce("user", agent, '') as actor,
|
|
concat_ws(' ', query, conversation_window, entities_matched, claims_matched,
|
|
retrieval_layers_hit, retrieval_gap, market_data, research_context, kb_context_text,
|
|
tool_calls, raw_response, display_response, query_type) as body
|
|
from teleo_restore.response_audit
|
|
union all
|
|
select
|
|
'sources'::text as source_table,
|
|
path as row_id,
|
|
coalesce(updated_at, created_at) as event_time,
|
|
coalesce(submitted_by, original_author, original_author_handle, '') as actor,
|
|
concat_ws(' ', path, status, priority, priority_log, extraction_model, last_error,
|
|
feedback, content_type, original_author, original_author_handle) as body
|
|
from teleo_restore.sources
|
|
union all
|
|
select
|
|
'agent_research_runs'::text as source_table,
|
|
id as row_id,
|
|
coalesce(completed_at, created_at) as event_time,
|
|
coalesce(agent_slug, '') as actor,
|
|
concat_ws(' ', source_surface, source_ref, request_kind, prompt_excerpt,
|
|
selected_provider, selected_route, status, answer_excerpt, proof_ref) as body
|
|
from teleo_restore.agent_research_runs
|
|
union all
|
|
select
|
|
'agent_tool_invocations'::text as source_table,
|
|
id as row_id,
|
|
created_at as event_time,
|
|
coalesce(provider, tool_name, '') as actor,
|
|
concat_ws(' ', provider, tool_name, tool_category, endpoint_host, decision,
|
|
decision_reason, rail, network, currency, error_class) as body
|
|
from teleo_restore.agent_tool_invocations
|
|
), matched as (
|
|
select *,
|
|
(select count(*) from q where lower(raw_hits.body) like '%' || q.term || '%') as score
|
|
from raw_hits
|
|
where exists (select 1 from q where lower(raw_hits.body) like '%' || q.term || '%')
|
|
), ranked as (
|
|
select *, row_number() over (order by score desc, event_time desc nulls last, source_table, row_id) as rn
|
|
from matched
|
|
)
|
|
select json_build_object(
|
|
'hit_count_total', (select count(*) from matched),
|
|
'hits', coalesce((select json_agg(json_build_object(
|
|
'source_table', source_table,
|
|
'row_id', row_id,
|
|
'event_time', event_time,
|
|
'actor', actor,
|
|
'score', score,
|
|
'body_chars', length(body),
|
|
'body_md5', md5(body),
|
|
'excerpt', left(regexp_replace(body, '\\s+', ' ', 'g'), 700)
|
|
) order by rn) from ranked where rn <= {max(1, min(limit, 20))}), '[]'::json)
|
|
)::text;
|
|
"""
|
|
text = run_psql(args, sql).strip()
|
|
result = json.loads(text)
|
|
result.update(
|
|
{
|
|
"artifact": "teleo_cloudsql_memory_context",
|
|
"backend": "cloudsql:teleo-pgvector-standby/teleo_kb/teleo_restore",
|
|
"query": query,
|
|
"terms": terms,
|
|
"privacy": "redacted proof mode by default; pass --include-excerpts/--raw-ok only for private agent context",
|
|
}
|
|
)
|
|
if not include_excerpts(args):
|
|
for hit in result["hits"]:
|
|
hit.pop("excerpt", None)
|
|
result["privacy"] = (
|
|
"redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, source table, row id, and timestamps"
|
|
)
|
|
else:
|
|
result["privacy"] = (
|
|
"private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts"
|
|
)
|
|
return result
|
|
|
|
|
|
def query_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, Any]:
|
|
canonical = query_canonical_rows(args, query, limit)
|
|
if canonical.get("hit_count_total", 0) or canonical.get("hits"):
|
|
return canonical
|
|
if os.environ.get("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "") != "1":
|
|
canonical["canonical_fallback_reason"] = (
|
|
"no matching canonical public/persona/strategy/belief rows; audit fallback disabled"
|
|
)
|
|
return canonical
|
|
if not audit_restore_available(args):
|
|
canonical["canonical_fallback_reason"] = (
|
|
"no matching canonical public/persona/strategy/belief rows; teleo_restore audit fallback unavailable"
|
|
)
|
|
return canonical
|
|
audit = query_audit_rows(args, query, limit)
|
|
audit["canonical_fallback_reason"] = "no matching canonical public/persona/strategy/belief rows"
|
|
return audit
|
|
|
|
|
|
def canonical_backend(args: argparse.Namespace) -> str:
|
|
return f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage"
|
|
|
|
|
|
def canonical_claim(args: argparse.Namespace, claim_id: str) -> dict[str, Any] | None:
|
|
sql = f"""
|
|
select jsonb_build_object(
|
|
'id', id::text,
|
|
'type', type,
|
|
'text', text,
|
|
'status', status,
|
|
'confidence', confidence,
|
|
'tags', coalesce(to_jsonb(tags), '[]'::jsonb),
|
|
'superseded_by', superseded_by::text,
|
|
'created_at', created_at::text,
|
|
'updated_at', updated_at::text
|
|
)::text
|
|
from public.claims
|
|
where id = {sql_literal(claim_id)}::uuid;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
return rows[0] if rows else None
|
|
|
|
|
|
def canonical_evidence(args: argparse.Namespace, claim_id: str, limit: int) -> list[dict[str, Any]]:
|
|
sql = f"""
|
|
with ranked as (
|
|
select ce.claim_id,
|
|
s.id as source_id,
|
|
ce.role::text as role,
|
|
ce.weight,
|
|
s.source_type,
|
|
s.url,
|
|
s.storage_path,
|
|
s.hash as source_hash,
|
|
left(coalesce(s.excerpt, ''), 800) as excerpt,
|
|
row_number() over (
|
|
partition by ce.claim_id
|
|
order by (s.storage_path like 'inbox/archive/%') desc,
|
|
ce.role::text,
|
|
s.storage_path nulls last,
|
|
s.url nulls last
|
|
) as rn
|
|
from public.claim_evidence ce
|
|
join public.sources s on s.id = ce.source_id
|
|
where ce.claim_id = {sql_literal(claim_id)}::uuid
|
|
)
|
|
select jsonb_build_object(
|
|
'claim_id', claim_id::text,
|
|
'source_id', source_id::text,
|
|
'role', role,
|
|
'weight', weight,
|
|
'source_type', source_type,
|
|
'url', url,
|
|
'storage_path', storage_path,
|
|
'source_hash', source_hash,
|
|
'excerpt', excerpt
|
|
)::text
|
|
from ranked
|
|
where rn <= {max(1, min(limit, 50))}
|
|
order by rn;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
if not include_excerpts(args):
|
|
for row in rows:
|
|
row.pop("excerpt", None)
|
|
return rows
|
|
|
|
|
|
def canonical_edges(args: argparse.Namespace, claim_id: str, limit: int) -> list[dict[str, Any]]:
|
|
sql = f"""
|
|
with base(id) as (
|
|
values ({sql_literal(claim_id)}::uuid)
|
|
), edge_rows as (
|
|
select base.id as claim_id,
|
|
case when e.from_claim = base.id then 'outgoing' else 'incoming' end as direction,
|
|
e.edge_type::text as edge_type,
|
|
other.id as connected_id,
|
|
other.text as connected_text,
|
|
other.tags as connected_tags,
|
|
row_number() over (
|
|
partition by base.id
|
|
order by case e.edge_type::text
|
|
when 'supports' then 1
|
|
when 'challenges' then 2
|
|
when 'contradicts' then 3
|
|
when 'requires' then 4
|
|
when 'supersedes' then 5
|
|
when 'constrains' then 6
|
|
when 'causes' then 7
|
|
when 'accelerates' then 8
|
|
else 9
|
|
end,
|
|
other.text
|
|
) as rn
|
|
from base
|
|
join public.claim_edges e on e.from_claim = base.id or e.to_claim = base.id
|
|
join public.claims other on other.id = case when e.from_claim = base.id then e.to_claim else e.from_claim end
|
|
)
|
|
select jsonb_build_object(
|
|
'claim_id', claim_id::text,
|
|
'direction', direction,
|
|
'edge_type', edge_type,
|
|
'connected_id', connected_id::text,
|
|
'connected_text', connected_text,
|
|
'connected_tags', coalesce(to_jsonb(connected_tags), '[]'::jsonb)
|
|
)::text
|
|
from edge_rows
|
|
where rn <= {max(1, min(limit, 50))}
|
|
order by rn;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
if not include_excerpts(args):
|
|
for row in rows:
|
|
row.pop("connected_text", None)
|
|
return rows
|
|
|
|
|
|
def show_canonical_claim(args: argparse.Namespace) -> dict[str, Any]:
|
|
claim = canonical_claim(args, args.claim_id)
|
|
if not claim:
|
|
raise SystemExit(f"claim not found: {args.claim_id}")
|
|
if not include_excerpts(args):
|
|
claim.pop("text", None)
|
|
claim["claim_url"] = claim_url(args.claim_id)
|
|
edges = canonical_edges(args, args.claim_id, 12)
|
|
for edge in edges:
|
|
edge["connected_url"] = claim_url(edge.get("connected_id"))
|
|
return {
|
|
"artifact": "teleo_cloudsql_canonical_kb_claim",
|
|
"backend": canonical_backend(args),
|
|
"claim": claim,
|
|
"evidence": canonical_evidence(args, args.claim_id, 8),
|
|
"edges": edges,
|
|
}
|
|
|
|
|
|
def evidence_canonical_claim(args: argparse.Namespace) -> dict[str, Any]:
|
|
return {
|
|
"artifact": "teleo_cloudsql_canonical_kb_evidence",
|
|
"backend": canonical_backend(args),
|
|
"claim_id": args.claim_id,
|
|
"evidence": canonical_evidence(args, args.claim_id, args.limit),
|
|
}
|
|
|
|
|
|
def edges_canonical_claim(args: argparse.Namespace) -> dict[str, Any]:
|
|
edges = canonical_edges(args, args.claim_id, args.limit)
|
|
for edge in edges:
|
|
edge["connected_url"] = claim_url(edge.get("connected_id"))
|
|
return {
|
|
"artifact": "teleo_cloudsql_canonical_kb_edges",
|
|
"backend": canonical_backend(args),
|
|
"claim_id": args.claim_id,
|
|
"claim_url": claim_url(args.claim_id),
|
|
"edges": edges,
|
|
}
|
|
|
|
|
|
def propose_core_change(args: argparse.Namespace) -> dict[str, Any]:
|
|
payload = {
|
|
"target_kind": args.target_kind,
|
|
"target_ref": args.target_ref,
|
|
"current": args.current or None,
|
|
"proposed": args.proposed,
|
|
"evidence": args.evidence,
|
|
"implications": args.implication,
|
|
"originator": args.originator or None,
|
|
"routing_rule": "core Teleo changes must be staged in kb_stage.kb_proposals, not saved as runtime memory",
|
|
"canonical_apply_required": True,
|
|
}
|
|
payload_json = json.dumps(payload, sort_keys=True)
|
|
sql = f"""
|
|
with params as (
|
|
select
|
|
{sql_literal(args.proposal_type)}::text as proposal_type,
|
|
nullif({sql_literal(args.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 kb_stage.stage_leoclean_proposal(
|
|
p.proposal_type,
|
|
p.channel,
|
|
p.source_ref,
|
|
p.rationale,
|
|
p.payload
|
|
) as proposal
|
|
from params p
|
|
)
|
|
select jsonb_build_object(
|
|
'artifact', 'teleo_cloudsql_kb_core_change_proposal',
|
|
'backend', {sql_literal(canonical_backend(args))},
|
|
'id', proposal->>'id',
|
|
'proposal_type', proposal->>'proposal_type',
|
|
'status', proposal->>'status',
|
|
'proposed_by_handle', proposal->>'proposed_by_handle',
|
|
'proposed_by_agent_id', proposal->>'proposed_by_agent_id',
|
|
'channel', proposal->>'channel',
|
|
'source_ref', proposal->>'source_ref',
|
|
'rationale', proposal->>'rationale',
|
|
'payload', proposal->'payload',
|
|
'created_at', proposal->>'created_at',
|
|
'canonical_apply_done', false,
|
|
'runtime_memory_write_done', false
|
|
)::text
|
|
from proposer;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
if not rows:
|
|
raise SystemExit("No core-change proposal inserted.")
|
|
return rows[0]
|
|
|
|
|
|
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.channel)}, '') as channel,
|
|
nullif({sql_literal(args.source_ref)}, '') as source_ref,
|
|
{sql_literal(args.rationale)} as rationale
|
|
), from_row as (
|
|
select c.id, c.text
|
|
from public.claims c, params p
|
|
where c.id = p.from_claim
|
|
), to_row as (
|
|
select c.id, c.text
|
|
from public.claims c, params p
|
|
where c.id = p.to_claim
|
|
), existing_edge as (
|
|
select e.id
|
|
from public.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
|
|
), staged as (
|
|
select kb_stage.stage_leoclean_proposal(
|
|
'add_edge',
|
|
p.channel,
|
|
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
|
|
)
|
|
)
|
|
) as proposal
|
|
from params p
|
|
join from_row on true
|
|
join to_row on true
|
|
)
|
|
select jsonb_build_object(
|
|
'artifact', 'teleo_cloudsql_kb_edge_proposal',
|
|
'backend', {sql_literal(canonical_backend(args))},
|
|
'id', proposal->>'id',
|
|
'proposal_type', proposal->>'proposal_type',
|
|
'status', proposal->>'status',
|
|
'proposed_by_handle', proposal->>'proposed_by_handle',
|
|
'proposed_by_agent_id', proposal->>'proposed_by_agent_id',
|
|
'channel', proposal->>'channel',
|
|
'source_ref', proposal->>'source_ref',
|
|
'rationale', proposal->>'rationale',
|
|
'payload', proposal->'payload',
|
|
'created_at', proposal->>'created_at',
|
|
'canonical_apply_done', false,
|
|
'runtime_memory_write_done', false
|
|
)::text
|
|
from staged;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
if not rows:
|
|
raise SystemExit("No edge proposal inserted. Check that both claim ids exist and edge_type is valid.")
|
|
return rows[0]
|
|
|
|
|
|
def list_proposals(args: argparse.Namespace) -> dict[str, Any]:
|
|
where = ""
|
|
if args.status.lower() != "all":
|
|
where = f"where status = {sql_literal(args.status)}"
|
|
sql = f"""
|
|
select jsonb_build_object(
|
|
'id', id::text,
|
|
'proposal_type', proposal_type,
|
|
'status', status,
|
|
'proposed_by_handle', proposed_by_handle,
|
|
'channel', channel,
|
|
'source_ref', source_ref,
|
|
'rationale', left(rationale, 500),
|
|
'payload', payload,
|
|
'created_at', created_at::text,
|
|
'reviewed_at', reviewed_at::text,
|
|
'applied_at', applied_at::text
|
|
)::text
|
|
from kb_stage.kb_proposals
|
|
{where}
|
|
order by created_at desc
|
|
limit {max(1, min(args.limit, 100))};
|
|
"""
|
|
return {
|
|
"artifact": "teleo_cloudsql_kb_proposal_list",
|
|
"backend": canonical_backend(args),
|
|
"status_filter": args.status,
|
|
"proposals": psql_json_lines(args, sql, db=args.canonical_db),
|
|
}
|
|
|
|
|
|
def search_proposals(args: argparse.Namespace) -> dict[str, Any]:
|
|
terms = terms_for(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 {
|
|
"artifact": "teleo_cloudsql_kb_proposal_search",
|
|
"backend": canonical_backend(args),
|
|
"query": args.query,
|
|
"terms": terms,
|
|
"status_filter": args.status,
|
|
"proposals": psql_json_lines(args, sql, db=args.canonical_db),
|
|
}
|
|
|
|
|
|
def show_proposal(args: argparse.Namespace) -> dict[str, Any]:
|
|
sql = f"""
|
|
select jsonb_build_object(
|
|
'artifact', 'teleo_cloudsql_kb_proposal',
|
|
'backend', {sql_literal(canonical_backend(args))},
|
|
'id', id::text,
|
|
'proposal_type', proposal_type,
|
|
'status', status,
|
|
'proposed_by_handle', proposed_by_handle,
|
|
'proposed_by_agent_id', proposed_by_agent_id::text,
|
|
'channel', channel,
|
|
'source_ref', source_ref,
|
|
'rationale', rationale,
|
|
'payload', payload,
|
|
'reviewed_by_handle', reviewed_by_handle,
|
|
'reviewed_at', reviewed_at::text,
|
|
'review_note', review_note,
|
|
'applied_by_handle', applied_by_handle,
|
|
'applied_at', applied_at::text,
|
|
'created_at', created_at::text,
|
|
'updated_at', updated_at::text
|
|
)::text
|
|
from kb_stage.kb_proposals
|
|
where id = {sql_literal(args.proposal_id)}::uuid;
|
|
"""
|
|
rows = psql_json_lines(args, sql, db=args.canonical_db)
|
|
if not rows:
|
|
raise SystemExit(f"proposal not found: {args.proposal_id}")
|
|
return rows[0]
|
|
|
|
|
|
def decision_matrix_status(args: argparse.Namespace) -> dict[str, Any]:
|
|
sql = """
|
|
with table_status(schema_name, table_name, exists_in_cloudsql) 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_cloudsql_kb_decision_matrix_status',
|
|
'backend', current_database() || '|' || current_user,
|
|
'decision_matrix_tables', (
|
|
select jsonb_object_agg(schema_name || '.' || table_name, exists_in_cloudsql order by schema_name, table_name)
|
|
from table_status
|
|
),
|
|
'all_required_tables_present', (
|
|
select bool_and(exists_in_cloudsql) from table_status
|
|
),
|
|
'any_decision_matrix_table_present', (
|
|
select bool_or(exists_in_cloudsql) from table_status
|
|
),
|
|
'proposal_status_counts', (select counts from status_counts),
|
|
'guidance', case
|
|
when (select bool_and(exists_in_cloudsql) 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_lines(args, sql, db=args.canonical_db)
|
|
if not rows:
|
|
raise SystemExit("No decision matrix status returned.")
|
|
result = rows[0]
|
|
result["backend"] = canonical_backend(args)
|
|
return result
|
|
|
|
|
|
def audit_restore_available(args: argparse.Namespace) -> bool:
|
|
if os.environ.get("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "") != "1":
|
|
return False
|
|
sql = """
|
|
select bool_and(to_regclass(table_name) is not null)
|
|
from (values
|
|
('teleo_restore.response_audit'),
|
|
('teleo_restore.sources'),
|
|
('teleo_restore.agent_research_runs'),
|
|
('teleo_restore.agent_tool_invocations'),
|
|
('teleo_restore.audit_log'),
|
|
('teleo_restore.metrics_snapshots'),
|
|
('teleo_restore.prs'),
|
|
('teleo_restore.review_records')
|
|
) required(table_name);
|
|
"""
|
|
return run_psql(args, sql).strip().lower() in {"t", "true"}
|
|
|
|
|
|
def status(args: argparse.Namespace) -> dict[str, Any]:
|
|
canonical_sql = """
|
|
select json_build_object(
|
|
'db_identity', current_database() || '|' || current_user,
|
|
'extensions', (select json_agg(extname order by extname) from pg_extension),
|
|
'schema_tables', (
|
|
select json_object_agg(table_schema, table_count) from (
|
|
select table_schema, count(*) as table_count
|
|
from information_schema.tables
|
|
where table_type='BASE TABLE' and table_schema in ('public','kb_stage')
|
|
group by table_schema
|
|
order by table_schema
|
|
) counts
|
|
),
|
|
'high_signal_rows', json_build_object(
|
|
'agents', (select count(*) from public.agents),
|
|
'personas', (select count(*) from public.personas),
|
|
'strategies', (select count(*) from public.strategies),
|
|
'beliefs', (select count(*) from public.beliefs),
|
|
'claims', (select count(*) from public.claims),
|
|
'claim_edges', (select count(*) from public.claim_edges),
|
|
'claim_evidence', (select count(*) from public.claim_evidence),
|
|
'sources', (select count(*) from public.sources),
|
|
'kb_proposals', (select count(*) from kb_stage.kb_proposals)
|
|
)
|
|
)::text;
|
|
"""
|
|
audit_sql = """
|
|
select json_build_object(
|
|
'db_identity', current_database() || '|' || current_user,
|
|
'extensions', (select json_agg(extname order by extname) from pg_extension),
|
|
'teleo_restore_tables', (select count(*) from pg_tables where schemaname='teleo_restore'),
|
|
'total_rows', (
|
|
select sum(row_count) from (
|
|
select count(*)::bigint as row_count from teleo_restore.response_audit
|
|
union all select count(*)::bigint from teleo_restore.sources
|
|
union all select count(*)::bigint from teleo_restore.agent_research_runs
|
|
union all select count(*)::bigint from teleo_restore.agent_tool_invocations
|
|
union all select count(*)::bigint from teleo_restore.audit_log
|
|
union all select count(*)::bigint from teleo_restore.metrics_snapshots
|
|
union all select count(*)::bigint from teleo_restore.prs
|
|
union all select count(*)::bigint from teleo_restore.review_records
|
|
) counts
|
|
),
|
|
'high_signal_rows', json_build_object(
|
|
'response_audit', (select count(*) from teleo_restore.response_audit),
|
|
'sources', (select count(*) from teleo_restore.sources),
|
|
'agent_research_runs', (select count(*) from teleo_restore.agent_research_runs),
|
|
'agent_tool_invocations', (select count(*) from teleo_restore.agent_tool_invocations)
|
|
)
|
|
)::text;
|
|
"""
|
|
canonical = json.loads(run_psql(args, canonical_sql, db=args.canonical_db).strip())
|
|
audit_available = audit_restore_available(args)
|
|
if audit_available:
|
|
result = json.loads(run_psql(args, audit_sql).strip())
|
|
else:
|
|
result = {
|
|
"db_identity": canonical["db_identity"],
|
|
"extensions": canonical.get("extensions") or [],
|
|
"teleo_restore_tables": 0,
|
|
"total_rows": 0,
|
|
"high_signal_rows": {},
|
|
}
|
|
result["canonical"] = canonical
|
|
result["audit_fallback_available"] = audit_available
|
|
result["artifact"] = "teleo_cloudsql_memory_status"
|
|
result["backend"] = f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage primary; " + (
|
|
f"{args.db}/teleo_restore fallback" if audit_available else "teleo_restore fallback unavailable"
|
|
)
|
|
return result
|
|
|
|
|
|
def emit_json(value: dict[str, Any]) -> None:
|
|
print(json.dumps(value, indent=2))
|
|
|
|
|
|
def emit_markdown(value: dict[str, Any]) -> None:
|
|
receipt = value.get("retrieval_receipt") or {}
|
|
if receipt:
|
|
consistency = receipt.get("read_consistency") or {}
|
|
print("# Teleo KB Retrieval Receipt\n")
|
|
print(f"- schema: `{receipt.get('schema')}`")
|
|
print(f"- semantic context SHA-256: `{receipt.get('semantic_context_sha256')}`")
|
|
print(f"- artifact state SHA-256: `{receipt.get('artifact_state_sha256')}`")
|
|
print(f"- DB read consistency: `{consistency.get('status')}`; attempts: `{consistency.get('attempts')}`\n")
|
|
|
|
if value["artifact"] == "teleo_cloudsql_memory_status":
|
|
print("# Teleo Cloud SQL Memory Status\n")
|
|
print(f"- Backend: `{value['backend']}`")
|
|
print(f"- DB: `{value['db_identity']}`")
|
|
if "canonical" in value:
|
|
canonical = value["canonical"]
|
|
print(f"- Canonical DB: `{canonical['db_identity']}`")
|
|
print(f"- Canonical tables: `{json.dumps(canonical.get('schema_tables'), sort_keys=True)}`")
|
|
print(f"- Canonical high-signal rows: `{json.dumps(canonical.get('high_signal_rows'), sort_keys=True)}`")
|
|
counts = canonical.get("high_signal_rows") or {}
|
|
if all(key in counts for key in ("claims", "sources", "claim_edges", "claim_evidence", "kb_proposals")):
|
|
print(
|
|
f"\nDB readback: claims: `{counts['claims']}`; sources: `{counts['sources']}`; "
|
|
f"claim_edges: `{counts['claim_edges']}`; claim_evidence: `{counts['claim_evidence']}`; "
|
|
f"kb_proposals: `{counts['kb_proposals']}`."
|
|
)
|
|
print(f"- Extensions: {', '.join(value.get('extensions') or [])}")
|
|
print(f"- Restored tables: {value['teleo_restore_tables']}")
|
|
print(f"- Approx restored rows checked: {value['total_rows']}")
|
|
print(f"- High-signal rows: `{json.dumps(value['high_signal_rows'], sort_keys=True)}`")
|
|
return
|
|
|
|
if "claims" in value:
|
|
print("# Teleo KB Context\n")
|
|
print(f"- Backend: `{value['backend']}`")
|
|
print(f"- Query: `{value['query']}`")
|
|
print(f"- Terms: {', '.join(value['terms'])}")
|
|
print(f"- Privacy: {value['privacy']}\n")
|
|
if value.get("context_rows"):
|
|
print("## Soul / Context Rows\n")
|
|
for row in value["context_rows"]:
|
|
print(
|
|
f"- `{row.get('source')}` / `{row.get('owner')}` / `{row.get('title')}` "
|
|
f"(score {row.get('score')}): {row.get('body', '[redacted]')}"
|
|
)
|
|
print()
|
|
print("## Claims\n")
|
|
if not value.get("claims"):
|
|
print("No matching canonical claims found.")
|
|
return
|
|
for i, claim in enumerate(value["claims"], 1):
|
|
claim_label = claim.get("text") or "[redacted claim text]"
|
|
print(f"### {i}. {markdown_claim_link(claim['id'], claim_label[: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.get('type')}`; confidence: `{claim.get('confidence')}`; score: `{claim.get('score')}`"
|
|
)
|
|
print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
|
|
print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`")
|
|
print(f"- edge rows: `{claim.get('edge_count', len(claim.get('edges', [])))}`")
|
|
if claim.get("evidence"):
|
|
print("\nEvidence:")
|
|
for ev in claim["evidence"]:
|
|
source = ev.get("storage_path") or ev.get("url") or "(no source pointer)"
|
|
excerpt = f" - {ev['excerpt']}" if ev.get("excerpt") else ""
|
|
print(f"- `{ev.get('role')}` / `{ev.get('source_type')}` / `{source}`{excerpt}")
|
|
if claim.get("edges"):
|
|
print("\nEdges:")
|
|
for edge in claim["edges"]:
|
|
connected = edge.get("connected_id")
|
|
connected_label = (edge.get("connected_text") or str(connected or ""))[:120]
|
|
print(
|
|
f"- `{edge.get('direction')}` `{edge.get('edge_type')}` "
|
|
f"{markdown_claim_link(connected, connected_label)}"
|
|
)
|
|
print()
|
|
return
|
|
|
|
if value["artifact"] == "teleo_cloudsql_canonical_kb_claim":
|
|
claim = value["claim"]
|
|
print("# Teleo KB Claim\n")
|
|
print(f"- Backend: `{value['backend']}`")
|
|
print(f"- 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.get('type')}`; status: `{claim.get('status')}`; confidence: `{claim.get('confidence')}`"
|
|
)
|
|
print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
|
|
if claim.get("text"):
|
|
print(f"\n{markdown_claim_link(claim['id'], claim['text'][:180])}\n")
|
|
print("## Evidence\n")
|
|
for ev in value.get("evidence", []):
|
|
source = ev.get("storage_path") or ev.get("url") or "(no source pointer)"
|
|
excerpt = f" - {ev['excerpt']}" if ev.get("excerpt") else ""
|
|
print(f"- `{ev.get('role')}` / `{ev.get('source_type')}` / `{source}`{excerpt}")
|
|
print("\n## Edges\n")
|
|
for edge in value.get("edges", []):
|
|
connected = edge.get("connected_id")
|
|
connected_label = (edge.get("connected_text") or str(connected or ""))[:120]
|
|
print(
|
|
f"- `{edge.get('direction')}` `{edge.get('edge_type')}` "
|
|
f"{markdown_claim_link(connected, connected_label)}"
|
|
)
|
|
return
|
|
|
|
if value["artifact"] in {"teleo_cloudsql_canonical_kb_evidence", "teleo_cloudsql_canonical_kb_edges"}:
|
|
print(f"# {value['artifact']}\n")
|
|
print(f"- Backend: `{value['backend']}`")
|
|
print(f"- Claim: {markdown_claim_link(value['claim_id'], value['claim_id'])}\n")
|
|
print(f"- Open full claim/body/edges: {markdown_claim_link(value['claim_id'], 'claim page')}\n")
|
|
rows = value.get("evidence") or value.get("edges") or []
|
|
for row in rows:
|
|
print(f"- `{json.dumps(row, sort_keys=True)}`")
|
|
return
|
|
|
|
if value["artifact"] in {"teleo_cloudsql_kb_core_change_proposal", "teleo_cloudsql_kb_edge_proposal"}:
|
|
title = (
|
|
"Staged KB Edge Proposal"
|
|
if value["artifact"] == "teleo_cloudsql_kb_edge_proposal"
|
|
else "Staged KB Core-Change Proposal"
|
|
)
|
|
print(f"# {title}\n")
|
|
print(f"- Proposal id: `{value['id']}`")
|
|
print(f"- Type: `{value['proposal_type']}`")
|
|
print(f"- Status: `{value['status']}`")
|
|
print(f"- Source: `{value.get('source_ref') or '-'}`")
|
|
print(f"- Runtime memory write: `{value['runtime_memory_write_done']}`")
|
|
print(f"- Canonical apply done: `{value['canonical_apply_done']}`")
|
|
print("\n## Rationale\n")
|
|
print(value["rationale"])
|
|
print("\n## Payload\n")
|
|
print(json.dumps(value.get("payload") or {}, indent=2, sort_keys=True))
|
|
return
|
|
|
|
if value["artifact"] == "teleo_cloudsql_kb_proposal_list":
|
|
print("# Staged KB Proposals\n")
|
|
print(f"- Status filter: `{value.get('status_filter')}`\n")
|
|
rows = value.get("proposals") or []
|
|
if not rows:
|
|
print("No matching proposals.")
|
|
for row in rows:
|
|
print(f"## {row['id']}\n")
|
|
print(f"- Type: `{row['proposal_type']}`")
|
|
print(f"- Status: `{row['status']}`")
|
|
print(f"- Source: `{row.get('source_ref') or '-'}`")
|
|
print(f"- Created: `{row.get('created_at')}`")
|
|
print(f"- Rationale: {row.get('rationale') or ''}")
|
|
print()
|
|
return
|
|
|
|
if value["artifact"] == "teleo_cloudsql_kb_proposal_search":
|
|
print("# Staged KB Proposal Search\n")
|
|
print(f"- Query: `{value['query']}`")
|
|
print(f"- Terms: `{', '.join(value['terms'])}`")
|
|
print(f"- Status filter: `{value['status_filter']}`\n")
|
|
rows = value.get("proposals") or []
|
|
if not rows:
|
|
print("No matching proposals.")
|
|
for row in rows:
|
|
print(f"## {row['id']}\n")
|
|
print(f"- Type: `{row['proposal_type']}`")
|
|
print(f"- Status: `{row['status']}`")
|
|
print(f"- Source: `{row.get('source_ref') or '-'}`")
|
|
print(f"- Created: `{row.get('created_at')}`")
|
|
print(f"- Applied at: `{row.get('applied_at') or '-'}`")
|
|
print(f"- Rationale: {row.get('rationale') or ''}")
|
|
print()
|
|
return
|
|
|
|
if value["artifact"] == "teleo_cloudsql_kb_proposal":
|
|
print("# Staged KB Proposal\n")
|
|
print(f"- Proposal id: `{value['id']}`")
|
|
print(f"- Type: `{value['proposal_type']}`")
|
|
print(f"- Status: `{value['status']}`")
|
|
print(f"- Source: `{value.get('source_ref') or '-'}`")
|
|
print(f"- Created: `{value.get('created_at')}`")
|
|
print(f"- Applied at: `{value.get('applied_at') or '-'}`")
|
|
print("\n## Rationale\n")
|
|
print(value["rationale"])
|
|
print("\n## Payload\n")
|
|
print(json.dumps(value.get("payload") or {}, indent=2, sort_keys=True))
|
|
return
|
|
|
|
if value["artifact"] == "teleo_cloudsql_kb_decision_matrix_status":
|
|
print("# Decision Matrix Status\n")
|
|
print(f"- Backend: `{value['backend']}`")
|
|
print(f"- all required tables present: `{value['all_required_tables_present']}`")
|
|
print(f"- any decision-matrix table present: `{value['any_decision_matrix_table_present']}`")
|
|
print(f"- proposal status counts: `{json.dumps(value.get('proposal_status_counts') or {}, sort_keys=True)}`")
|
|
print(f"- guidance: {value['guidance']}")
|
|
print("\n## Tables\n")
|
|
for table, exists in sorted((value.get("decision_matrix_tables") or {}).items()):
|
|
print(f"- `{table}`: `{exists}`")
|
|
print()
|
|
return
|
|
|
|
print("# Teleo Cloud SQL Memory Context\n")
|
|
print(f"- Backend: `{value['backend']}`")
|
|
print(f"- Query: `{value['query']}`")
|
|
print(f"- Terms: {', '.join(value['terms'])}")
|
|
print(f"- Total matching rows: {value['hit_count_total']}")
|
|
print(f"- Privacy: {value['privacy']}\n")
|
|
for i, hit in enumerate(value["hits"], 1):
|
|
print(f"## Hit {i}: {hit['source_table']} / {hit['row_id']}")
|
|
print(f"- Time: {hit.get('event_time') or 'unknown'}")
|
|
print(f"- Actor/source: {hit.get('actor') or 'unknown'}")
|
|
print(f"- Score: {hit.get('score')}")
|
|
print(f"- Body chars: {hit.get('body_chars')}")
|
|
print(f"- Body md5: `{hit.get('body_md5')}`")
|
|
if "excerpt" in hit:
|
|
print(f"- Excerpt: {hit['excerpt']}")
|
|
print()
|
|
|
|
|
|
def main() -> None:
|
|
args = parse_args()
|
|
validate_runtime_connection(args)
|
|
read_loaders = {
|
|
"search": lambda: query_rows(args, args.query, args.limit),
|
|
"context": lambda: query_rows(args, args.query, args.limit),
|
|
"show": lambda: show_canonical_claim(args),
|
|
"evidence": lambda: evidence_canonical_claim(args),
|
|
"edges": lambda: edges_canonical_claim(args),
|
|
"status": lambda: status(args),
|
|
"list-proposals": lambda: list_proposals(args),
|
|
"search-proposals": lambda: search_proposals(args),
|
|
"show-proposal": lambda: show_proposal(args),
|
|
"decision-matrix-status": lambda: decision_matrix_status(args),
|
|
}
|
|
if args.command in read_loaders:
|
|
result = read_with_retrieval_receipt(args, read_loaders[args.command])
|
|
elif args.command == "propose-core-change":
|
|
result = propose_core_change(args)
|
|
elif args.command == "propose-edge":
|
|
result = propose_edge(args)
|
|
else:
|
|
raise SystemExit(f"Unsupported command: {args.command}")
|
|
if args.format == "json":
|
|
emit_json(result)
|
|
else:
|
|
emit_markdown(result)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|