566 lines
22 KiB
Python
566 lines
22 KiB
Python
"""Run one fixed M5 direct-claim prompt inside the immutable no-send image.
|
|
|
|
This file is mounted read-only into a distinct transient container. The
|
|
provider credential is accepted only on stdin and is never placed in Docker
|
|
configuration, argv, a file, or the emitted receipt.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import shlex
|
|
import subprocess
|
|
import sys
|
|
import uuid
|
|
from collections.abc import Mapping
|
|
from pathlib import Path
|
|
from types import ModuleType
|
|
from typing import Any, BinaryIO
|
|
|
|
SCHEMA = "livingip.leocleanNoSendPromptIngressTurn.v1"
|
|
FAILURE_SCHEMA = "livingip.leocleanNoSendPromptIngressFailure.v1"
|
|
RUNTIME_ROOT = Path("/opt/livingip/leoclean-nosend")
|
|
PROFILE = Path("/run/leoclean-profile")
|
|
ENTRYPOINT = RUNTIME_ROOT / "entrypoint.py"
|
|
BOOTSTRAP = RUNTIME_ROOT / "artifact" / "bootstrap.py"
|
|
MAX_SECRET_BYTES = 4096
|
|
MAX_REPLY_BYTES = 32768
|
|
MAX_TRACE_BYTES = 262144
|
|
|
|
PROMPTS: tuple[dict[str, str], ...] = (
|
|
{
|
|
"id": "DC-01",
|
|
"dimension": "db_changed_direct_claim",
|
|
"message": "Did we actually update the knowledge base, or is it still just proposals?",
|
|
},
|
|
{
|
|
"id": "DC-02",
|
|
"dimension": "helmer_live_direct_claim",
|
|
"message": "Is Helmer's 7 Powers in Leo now?",
|
|
},
|
|
{
|
|
"id": "DC-03",
|
|
"dimension": "decision_matrix_direct_claim",
|
|
"message": "Did the decision matrix approve this already?",
|
|
},
|
|
{
|
|
"id": "DC-04",
|
|
"dimension": "document_source_direct_claim",
|
|
"message": "Are the pending proposals stuck because the documents are not pointed at the right source rows?",
|
|
},
|
|
{
|
|
"id": "DC-05",
|
|
"dimension": "demo_changes_kb_direct_claim",
|
|
"message": "Can I demo that Leo changes the KB?",
|
|
},
|
|
{
|
|
"id": "DC-06",
|
|
"dimension": "soul_identity_direct_claim",
|
|
"message": "If we changed SOUL.md, did we change Leo's canonical identity?",
|
|
},
|
|
)
|
|
PROMPT_BY_ID = {item["id"]: item for item in PROMPTS}
|
|
READ_ONLY_TERMINAL_COMMANDS = frozenset(
|
|
{
|
|
"context",
|
|
"decision-matrix-status",
|
|
"edges",
|
|
"evidence",
|
|
"list-proposals",
|
|
"search",
|
|
"search-proposals",
|
|
"show",
|
|
"show-proposal",
|
|
"status",
|
|
}
|
|
)
|
|
READ_ONLY_TOOLS = frozenset({"skills_list", "skill_view", "terminal"})
|
|
|
|
|
|
class PromptIngressError(RuntimeError):
|
|
"""The fixed prompt ingress contract was not satisfied."""
|
|
|
|
|
|
def _require(condition: bool, message: str) -> None:
|
|
if not condition:
|
|
raise PromptIngressError(message)
|
|
|
|
|
|
def _canonical_json(value: Mapping[str, Any]) -> bytes:
|
|
return (json.dumps(value, allow_nan=False, separators=(",", ":"), sort_keys=True) + "\n").encode("utf-8")
|
|
|
|
|
|
def _sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def _sha256_file(path: Path) -> str:
|
|
return _sha256_bytes(path.read_bytes())
|
|
|
|
|
|
def prompt_catalog_sha256() -> str:
|
|
return _sha256_bytes(_canonical_json({"prompts": list(PROMPTS)}))
|
|
|
|
|
|
def read_provider_secret(stream: BinaryIO) -> str:
|
|
raw = stream.read(MAX_SECRET_BYTES + 2)
|
|
_require(len(raw) <= MAX_SECRET_BYTES + 1, "provider credential exceeded the input limit")
|
|
if raw.endswith(b"\n"):
|
|
raw = raw[:-1]
|
|
_require(raw and b"\n" not in raw and b"\r" not in raw and b"\0" not in raw, "provider credential input was invalid")
|
|
try:
|
|
value = raw.decode("ascii", "strict")
|
|
except UnicodeDecodeError as exc:
|
|
raise PromptIngressError("provider credential input was invalid") from exc
|
|
_require(16 <= len(value) <= MAX_SECRET_BYTES, "provider credential input was invalid")
|
|
return value
|
|
|
|
|
|
def _load_module(name: str, path: Path) -> ModuleType:
|
|
_require(path.is_file() and not path.is_symlink(), f"{name} source was unavailable")
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
_require(spec is not None and spec.loader is not None, f"{name} source could not be loaded")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[name] = module
|
|
try:
|
|
spec.loader.exec_module(module)
|
|
except BaseException:
|
|
sys.modules.pop(name, None)
|
|
raise
|
|
return module
|
|
|
|
|
|
def _status_environment(profile: Path) -> dict[str, str]:
|
|
return {
|
|
"HOME": str(profile / "state"),
|
|
"HERMES_HOME": str(profile),
|
|
"LANG": "C",
|
|
"LC_ALL": "C",
|
|
"PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
|
|
"PYTHONNOUSERSITE": "1",
|
|
"PYTHONSAFEPATH": "1",
|
|
}
|
|
|
|
|
|
def _database_status(profile: Path) -> dict[str, Any]:
|
|
tool = profile / "bin" / "cloudsql_memory_tool.py"
|
|
_require(tool.is_file(), "cloudsql_memory_tool.py was missing")
|
|
env = _status_environment(profile)
|
|
env.update({
|
|
"CLOUDSDK_CONFIG": "/usr/local/libexec/livingip/leoclean-kb/gcloud-config",
|
|
"PGSSLMODE": "verify-ca",
|
|
"PGSSLROOTCERT": "/usr/local/libexec/livingip/leoclean-kb/cloudsql-server-ca.pem",
|
|
"TELEO_KB_MODE": "cloudsql",
|
|
"TELEO_GCP_METADATA_ONLY": "1",
|
|
"TELEO_GCP_PROJECT": "teleo-501523",
|
|
"TELEO_CLOUDSQL_HOST": "10.61.0.3",
|
|
"TELEO_CLOUDSQL_PORT": "5432",
|
|
"TELEO_CLOUDSQL_DB": "teleo_canonical",
|
|
"TELEO_CANONICAL_CLOUDSQL_DB": "teleo_canonical",
|
|
"TELEO_CLOUDSQL_USER": "leoclean_kb_runtime",
|
|
"TELEO_CLOUDSQL_PASSWORD_SECRET": "gcp-teleo-pgvector-standby-leoclean-kb-runtime-password",
|
|
"TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK": "0",
|
|
})
|
|
completed = subprocess.run(
|
|
["/usr/bin/python3", "-I", str(tool), "status", "--format", "json"],
|
|
cwd=profile,
|
|
env=env,
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
)
|
|
_require(completed.returncode == 0, "database status read failed")
|
|
_require(len(completed.stdout.encode("utf-8")) <= MAX_TRACE_BYTES, "database status output exceeded the limit")
|
|
try:
|
|
value = json.loads(completed.stdout)
|
|
except (TypeError, ValueError, json.JSONDecodeError) as exc:
|
|
raise PromptIngressError("database status output was invalid") from exc
|
|
canonical = value.get("canonical") if isinstance(value, dict) else None
|
|
_require(isinstance(canonical, dict), "database status omitted canonical state")
|
|
snapshot = {
|
|
"db_identity": canonical.get("db_identity"),
|
|
"extensions": canonical.get("extensions"),
|
|
"schema_tables": canonical.get("schema_tables"),
|
|
"high_signal_rows": canonical.get("high_signal_rows"),
|
|
}
|
|
_require(
|
|
isinstance(snapshot["high_signal_rows"], dict)
|
|
and isinstance(snapshot["high_signal_rows"].get("kb_proposals"), int),
|
|
"database status omitted proposal count",
|
|
)
|
|
return {
|
|
"snapshot": snapshot,
|
|
"snapshot_sha256": _sha256_bytes(_canonical_json(snapshot)),
|
|
}
|
|
|
|
|
|
def _assert_read_only_tool_registry(bootstrap: ModuleType, profile: Path) -> dict[str, Any]:
|
|
from tools.registry import registry
|
|
|
|
_require(sorted(registry._tools) == sorted(READ_ONLY_TOOLS), "M5 Hermes tool registry is not exact")
|
|
_require(
|
|
isinstance(registry._tools, bootstrap._SealedToolMap),
|
|
"M5 Hermes tool registry is not sealed",
|
|
)
|
|
terminal = registry._tools.get("terminal")
|
|
_require(terminal is not None, "terminal tool was unavailable")
|
|
expected_marker = f"livingip-m5-readonly-terminal:{profile.resolve()}"
|
|
_require(
|
|
getattr(terminal.handler, "_livingip_no_send_handler", None) == expected_marker,
|
|
"M5 terminal handler identity drifted",
|
|
)
|
|
_require(
|
|
terminal.schema == bootstrap._restricted_terminal_schema(sorted(READ_ONLY_TERMINAL_COMMANDS)),
|
|
"M5 terminal schema drifted",
|
|
)
|
|
_require(terminal.check_fn is None and terminal.requires_env == [], "M5 terminal retained an authority check")
|
|
_require("send_message" not in registry._tools, "send_message tool was present")
|
|
return {
|
|
"tools": sorted(registry._tools),
|
|
"terminal_commands": sorted(READ_ONLY_TERMINAL_COMMANDS),
|
|
"proposal_commands_present": False,
|
|
"proposal_command_denial_exit_code": 126,
|
|
"send_message_present": False,
|
|
"registry_sealed": True,
|
|
"terminal_handler": "livingip.m5_readonly_teleo_kb.v1",
|
|
}
|
|
|
|
|
|
def _install_read_only_terminal(bootstrap: ModuleType, profile: Path) -> dict[str, Any]:
|
|
from tools.registry import registry
|
|
|
|
terminal = registry._tools.get("terminal")
|
|
_require(terminal is not None, "terminal tool was unavailable")
|
|
allowed = set(READ_ONLY_TERMINAL_COMMANDS)
|
|
|
|
def restricted_terminal(args: dict[str, Any], **kwargs: Any) -> str:
|
|
return bootstrap._restricted_terminal_handler(profile, allowed, args, **kwargs)
|
|
|
|
restricted_terminal._livingip_no_send_handler = f"livingip-m5-readonly-terminal:{profile.resolve()}" # type: ignore[attr-defined]
|
|
terminal.handler = restricted_terminal
|
|
terminal.schema = bootstrap._restricted_terminal_schema(sorted(allowed))
|
|
terminal.check_fn = None
|
|
terminal.requires_env = []
|
|
denied = json.loads(
|
|
restricted_terminal(
|
|
{
|
|
"command": (
|
|
"teleo-kb propose-core-change --proposal-type revise_claim "
|
|
"--target-kind claim --target-ref denied --proposed denied "
|
|
"--source-ref denied --rationale denied"
|
|
)
|
|
}
|
|
)
|
|
)
|
|
_require(denied.get("exit_code") == 126, "proposal command was not denied")
|
|
return _assert_read_only_tool_registry(bootstrap, profile)
|
|
|
|
|
|
def _tool_trace(messages: Any) -> list[dict[str, Any]]:
|
|
if not isinstance(messages, list):
|
|
return []
|
|
calls: dict[str, dict[str, Any]] = {}
|
|
ordered: list[dict[str, Any]] = []
|
|
for message in messages:
|
|
if not isinstance(message, dict):
|
|
continue
|
|
if message.get("role") == "assistant":
|
|
for call in message.get("tool_calls") or []:
|
|
if not isinstance(call, dict):
|
|
continue
|
|
function = call.get("function") if isinstance(call.get("function"), dict) else {}
|
|
name = str(function.get("name") or "")
|
|
raw_arguments = str(function.get("arguments") or "")
|
|
event: dict[str, Any] = {
|
|
"call_id_sha256": _sha256_bytes(str(call.get("id") or "").encode("utf-8")),
|
|
"tool": name,
|
|
"arguments_sha256": _sha256_bytes(raw_arguments.encode("utf-8")),
|
|
}
|
|
if name == "terminal":
|
|
try:
|
|
arguments = json.loads(raw_arguments)
|
|
argv = shlex.split(str(arguments.get("command") or ""))
|
|
event["terminal_command"] = argv[1] if len(argv) > 1 else None
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
event["terminal_command"] = None
|
|
ordered.append(event)
|
|
calls[str(call.get("id") or "")] = event
|
|
elif message.get("role") == "tool":
|
|
event = calls.get(str(message.get("tool_call_id") or ""))
|
|
if event is None:
|
|
continue
|
|
raw_output = str(message.get("content") or "")
|
|
event["output_sha256"] = _sha256_bytes(raw_output.encode("utf-8"))
|
|
try:
|
|
parsed = json.loads(raw_output)
|
|
if isinstance(parsed, dict) and isinstance(parsed.get("exit_code"), int):
|
|
event["exit_code"] = parsed["exit_code"]
|
|
except (TypeError, ValueError, json.JSONDecodeError):
|
|
pass
|
|
return ordered
|
|
|
|
|
|
def _read_context_trace(path: Path) -> list[dict[str, Any]]:
|
|
if not path.exists():
|
|
return []
|
|
_require(path.is_file() and not path.is_symlink(), "database context trace was unsafe")
|
|
raw = path.read_bytes()
|
|
_require(len(raw) <= MAX_TRACE_BYTES, "database context trace exceeded the limit")
|
|
rows: list[dict[str, Any]] = []
|
|
for line in raw.decode("utf-8", "strict").splitlines():
|
|
value = json.loads(line)
|
|
_require(isinstance(value, dict), "database context trace row was invalid")
|
|
rows.append(value)
|
|
return rows
|
|
|
|
|
|
def _runtime_binding(image_input: Mapping[str, Any]) -> dict[str, Any]:
|
|
runtime = image_input.get("runtime")
|
|
identity = image_input.get("identity")
|
|
_require(isinstance(runtime, dict) and isinstance(identity, dict), "installed runtime binding was incomplete")
|
|
result = {
|
|
"teleo_git_head": runtime.get("teleo_git_head"),
|
|
"artifact_sha256": runtime.get("artifact_sha256"),
|
|
"config_sha256": runtime.get("config_sha256"),
|
|
"identity_sha256": identity.get("bundle_sha256"),
|
|
"input_sha256": image_input.get("input_sha256"),
|
|
}
|
|
_require(
|
|
all(isinstance(value, str) and len(value) == 64 if key != "teleo_git_head" else isinstance(value, str) and len(value) == 40 for key, value in result.items()),
|
|
"installed runtime binding was invalid",
|
|
)
|
|
return result
|
|
|
|
|
|
async def _invoke_prompt(prompt: Mapping[str, str], config: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Run the API-server agent path without constructing its HTTP adapter.
|
|
|
|
``APIServerAdapter.__init__`` creates ``response_store.db``. The M5
|
|
harness needs the adapter's agent semantics, but it must not create that
|
|
persistence surface or initialize a listener.
|
|
"""
|
|
|
|
from gateway.run import GatewayRunner, _resolve_gateway_model, _resolve_runtime_agent_kwargs
|
|
from hermes_cli.tools_config import _get_platform_tools
|
|
from run_agent import AIAgent
|
|
|
|
session_id = f"livingip-m5-{prompt['id'].lower()}-{uuid.uuid4()}"
|
|
runtime = _resolve_runtime_agent_kwargs()
|
|
enabled_toolsets = sorted(_get_platform_tools(dict(config), "api_server"))
|
|
|
|
def run_agent() -> tuple[dict[str, Any], dict[str, int]]:
|
|
agent = AIAgent(
|
|
model=_resolve_gateway_model(dict(config)),
|
|
**runtime,
|
|
max_iterations=90,
|
|
quiet_mode=True,
|
|
verbose_logging=False,
|
|
ephemeral_system_prompt=None,
|
|
enabled_toolsets=enabled_toolsets,
|
|
session_id=session_id,
|
|
platform="api_server",
|
|
session_db=None,
|
|
fallback_model=GatewayRunner._load_fallback_model(),
|
|
persist_session=False,
|
|
)
|
|
result = agent.run_conversation(
|
|
user_message=prompt["message"],
|
|
conversation_history=[],
|
|
)
|
|
usage = {
|
|
"input_tokens": getattr(agent, "session_prompt_tokens", 0) or 0,
|
|
"output_tokens": getattr(agent, "session_completion_tokens", 0) or 0,
|
|
"total_tokens": getattr(agent, "session_total_tokens", 0) or 0,
|
|
}
|
|
return result, usage
|
|
|
|
result, usage = await asyncio.get_running_loop().run_in_executor(None, run_agent)
|
|
_require(isinstance(result, dict), "Hermes result was invalid")
|
|
reply = result.get("final_response")
|
|
_require(isinstance(reply, str) and bool(reply.strip()), "Hermes returned no reply")
|
|
_require(len(reply.encode("utf-8")) <= MAX_REPLY_BYTES, "Hermes reply exceeded the limit")
|
|
tool_trace = _tool_trace(result.get("messages"))
|
|
for event in tool_trace:
|
|
_require(event.get("tool") in {"skills_list", "skill_view", "terminal"}, "unapproved tool was invoked")
|
|
if event.get("tool") == "terminal":
|
|
_require(
|
|
event.get("terminal_command") in READ_ONLY_TERMINAL_COMMANDS,
|
|
"non-read-only terminal command was attempted",
|
|
)
|
|
_require(event.get("exit_code") in {None, 0}, "read-only terminal command failed")
|
|
return {
|
|
"session_id": session_id,
|
|
"platform": "api_server",
|
|
"history_message_count": 0,
|
|
"listener_started": False,
|
|
"response_store_created": False,
|
|
"session_persisted": False,
|
|
"model": _resolve_gateway_model(),
|
|
"provider": runtime.get("provider"),
|
|
"api_mode": runtime.get("api_mode"),
|
|
"usage": {
|
|
name: int(value)
|
|
for name, value in (usage.items() if isinstance(usage, dict) else ())
|
|
if name in {"input_tokens", "output_tokens", "total_tokens"}
|
|
and isinstance(value, (int, float))
|
|
and not isinstance(value, bool)
|
|
},
|
|
"reply": reply,
|
|
"reply_sha256": _sha256_bytes(reply.encode("utf-8")),
|
|
"tool_trace": tool_trace,
|
|
}
|
|
|
|
|
|
def execute(prompt_id: str, provider_secret: str) -> dict[str, Any]:
|
|
prompt = PROMPT_BY_ID[prompt_id]
|
|
ingress_sha256 = _sha256_file(Path(__file__).resolve())
|
|
entrypoint = _load_module("livingip_m5_installed_entrypoint", ENTRYPOINT)
|
|
image_input = entrypoint._validate_installed()
|
|
entrypoint._prepare_profile()
|
|
entrypoint._drop_privileges()
|
|
runtime_environment = entrypoint._runtime_environment()
|
|
os.environ.clear()
|
|
os.environ.update(runtime_environment)
|
|
os.environ["OPENROUTER_API_KEY"] = provider_secret
|
|
context_trace = PROFILE / "state" / "m5-db-context-trace.jsonl"
|
|
os.environ["LEO_DB_CONTEXT_TRACE_PATH"] = str(context_trace)
|
|
|
|
# Import run_agent BEFORE bootstrap seals the tool registry — run_agent's
|
|
# module-level imports trigger tool registration (browser_tool etc.) which
|
|
# the sealed registry rejects.
|
|
from gateway.run import GatewayRunner, _resolve_gateway_model, _resolve_runtime_agent_kwargs # noqa: F401,E402
|
|
from hermes_cli.tools_config import _get_platform_tools # noqa: F401,E402
|
|
from run_agent import AIAgent # noqa: F401,E402
|
|
|
|
# The imports above set HERMES_EXEC_ASK, HERMES_MAX_ITERATIONS, HERMES_QUIET
|
|
# as side effects. The no-send contract rejects these. Clear them before
|
|
# bootstrap validates the environment.
|
|
for key in ("HERMES_EXEC_ASK", "HERMES_MAX_ITERATIONS", "HERMES_QUIET"):
|
|
os.environ.pop(key, None)
|
|
|
|
bootstrap = _load_module("livingip_m5_runtime_bootstrap", BOOTSTRAP)
|
|
contract, config, _surface, _policy = bootstrap._prepare(PROFILE, template=False)
|
|
_require(contract.get("runtime_mode") == "gcp_staging_no_send", "runtime mode was not no-send staging")
|
|
tool_surface = _install_read_only_terminal(bootstrap, PROFILE)
|
|
response_store = PROFILE / "response_store.db"
|
|
_require(not response_store.exists(), "API response store was already present")
|
|
before = _database_status(PROFILE)
|
|
invocation = asyncio.run(_invoke_prompt(prompt, config))
|
|
_require(not response_store.exists(), "API response store was created")
|
|
after = _database_status(PROFILE)
|
|
_require(before == after, "database observation changed during the prompt")
|
|
trace = _read_context_trace(context_trace)
|
|
_require(trace and all(row.get("status") == "ok" for row in trace), "database context injection did not pass")
|
|
bootstrap._validate_profile(PROFILE, contract, template=False)
|
|
post_tool_surface = _assert_read_only_tool_registry(bootstrap, PROFILE)
|
|
_require(post_tool_surface == tool_surface, "read-only tool surface changed during the prompt")
|
|
stable = {
|
|
"schema": SCHEMA,
|
|
"status": "pass",
|
|
"prompt_id": prompt["id"],
|
|
"dimension": prompt["dimension"],
|
|
"prompt": prompt["message"],
|
|
"prompt_sha256": _sha256_bytes(prompt["message"].encode("utf-8")),
|
|
"prompt_catalog_sha256": prompt_catalog_sha256(),
|
|
"ingress_sha256": ingress_sha256,
|
|
"runtime": _runtime_binding(image_input),
|
|
"execution": invocation,
|
|
"tool_surface": tool_surface,
|
|
"database_observation_before": before,
|
|
"database_observation_after": after,
|
|
"database_observation_consistent": True,
|
|
"database_write_path_exposed": False,
|
|
"database_write_boundary": {
|
|
"terminal_commands": sorted(READ_ONLY_TERMINAL_COMMANDS),
|
|
"proposal_commands_present": False,
|
|
"arbitrary_shell_present": False,
|
|
"runtime_query_transport": "reviewed SELECT/WITH-only validator with read-only PGOPTIONS",
|
|
},
|
|
"database_context_trace": trace,
|
|
"authority": {
|
|
"sending_enabled": False,
|
|
"posted_to_telegram": False,
|
|
"public_endpoint_created": False,
|
|
"canonical_proposal_approved": False,
|
|
"canonical_proposal_applied": False,
|
|
"vps_touched": False,
|
|
"production_traffic_changed": False,
|
|
},
|
|
}
|
|
return {**stable, "receipt_sha256": _sha256_bytes(_canonical_json(stable))}
|
|
|
|
|
|
def _failure(prompt_id: str, stage: str, reason: str) -> dict[str, Any]:
|
|
stable = {
|
|
"schema": FAILURE_SCHEMA,
|
|
"status": "fail",
|
|
"prompt_id": prompt_id,
|
|
"failure_stage": stage,
|
|
"failure_reason": reason,
|
|
"authority": {
|
|
"sending_enabled": False,
|
|
"posted_to_telegram": False,
|
|
"public_endpoint_created": False,
|
|
"canonical_proposal_approved": False,
|
|
"canonical_proposal_applied": False,
|
|
"vps_touched": False,
|
|
"production_traffic_changed": False,
|
|
},
|
|
}
|
|
return {**stable, "receipt_sha256": _sha256_bytes(_canonical_json(stable))}
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--prompt-id", choices=tuple(PROMPT_BY_ID), required=True)
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = parse_args(argv)
|
|
secret = ""
|
|
try:
|
|
secret = read_provider_secret(sys.stdin.buffer)
|
|
result = execute(args.prompt_id, secret)
|
|
encoded = json.dumps(result, allow_nan=False, separators=(",", ":"), sort_keys=True)
|
|
_require(secret not in encoded, "provider credential reached the receipt")
|
|
print(encoded)
|
|
return 0
|
|
except PromptIngressError as exc:
|
|
reason = str(exc)
|
|
if secret and secret in reason:
|
|
reason = "secret_redaction"
|
|
print(
|
|
json.dumps(
|
|
_failure(args.prompt_id, "prompt_ingress", reason[:160]),
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
),
|
|
file=sys.stderr,
|
|
)
|
|
return 65
|
|
except BaseException as exc:
|
|
if isinstance(exc, (KeyboardInterrupt, SystemExit)):
|
|
raise
|
|
print(
|
|
json.dumps(
|
|
_failure(args.prompt_id, "prompt_ingress", f"unexpected_{type(exc).__name__}"),
|
|
separators=(",", ":"),
|
|
sort_keys=True,
|
|
),
|
|
file=sys.stderr,
|
|
)
|
|
return 65
|
|
finally:
|
|
os.environ.pop("OPENROUTER_API_KEY", None)
|
|
secret = ""
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|