"""Private-pilot fictional lifecycle. Default preflight makes no network request. Run from brain-platform after installing the checkout with: pip install -e '.[api]'. Use one dedicated, operator-authorized test connection and the same run ID for all three stages. If a request is unresolved, explicitly repeat that exact stage and run ID; never invent a replacement ID to get past an uncertain write. """ from __future__ import annotations import argparse from dataclasses import dataclass, field import hashlib import json import os import re from urllib.parse import urlsplit from uuid import UUID INITIAL_TEXT = "Fictional caller: please call after 6 pm. Café visit pending." CORRECTED_TEXT = "Fictional caller: correction — please call after 7 pm. Café visit pending." REQUIRED_ENV = ( "BRAIN_API_BASE_URL", "BRAIN_ID", "BRAIN_APPLICATION_CONNECTION_ID", "BRAIN_APPLICATION_CREDENTIAL", "BRAIN_APPLICATION_TEST_CONNECTION", ) class DemoStopped(ValueError): """Content-free stop reason; source bodies and credentials never enter output.""" @dataclass(frozen=True) class Config: base_url: str brain_id: str connection_id: UUID credential: str = field(repr=False) def configuration(environ): if any(not environ.get(name) for name in REQUIRED_ENV): raise DemoStopped("missing_configuration") if environ["BRAIN_APPLICATION_TEST_CONNECTION"] != "1": raise DemoStopped("dedicated_test_connection_required") base = urlsplit(environ["BRAIN_API_BASE_URL"]) if (base.username is not None or base.password is not None or base.query or base.fragment or base.path not in ("", "/") or not base.hostname or not (base.scheme == "https" or base.scheme == "http" and base.hostname in {"localhost", "127.0.0.1", "::1"})): raise DemoStopped("https_or_loopback_origin_required") if re.fullmatch(r"[A-Za-z0-9._:-]{1,64}", environ["BRAIN_ID"]) is None: raise DemoStopped("invalid_brain_id") try: connection = UUID(environ["BRAIN_APPLICATION_CONNECTION_ID"]) except ValueError: raise DemoStopped("invalid_connection_id") from None return Config(environ["BRAIN_API_BASE_URL"], environ["BRAIN_ID"], connection, environ["BRAIN_APPLICATION_CREDENTIAL"]) def _export(run_id, *, text, expected_version_id=None, correction=False): return json.dumps({"format": "brain-application-conversation-export-v1", "export_id": "quickstart:" + run_id + (":correct" if correction else ":ingest"), "messages": [{"external_id": "quickstart:" + run_id + ":message", "thread_id": "quickstart:" + run_id + ":thread", "author_id": None, "role": "human", "occurred_at": None, "text": text, "expected_version_id": expected_version_id}]}, ensure_ascii=False).encode("utf-8") def _saved(client, raw): manifest = client.ingest(raw) if (manifest["counts"]["saved"] != 1 or manifest["counts"]["received"] != 1 or manifest["entries"][0]["status"] != "saved"): raise DemoStopped("ingest_unresolved_or_refused") receipt = manifest["entries"][0]["receipt"] if client.receipt(UUID(receipt["receipt_id"])) != receipt: raise DemoStopped("receipt_readback_mismatch") return receipt def _evidence(client, thread_id, version_id, expected_text): context = client.recall(thread_id) if (context["contract"] != "application-conversation-recall-v1" or context["publication_status"] != "held" or context["semantic_ready"] is not False or context["accepted_memory"] is not False or context["conversation_complete"] is not None or context["context_status"] != "prepared" or context["selected_count"] != 1 or context["has_more"] is not False or context["omission_reasons"]): raise DemoStopped("current_context_incomplete") prepared = context["prepared"] if len(prepared["originals"]) != 1 or len(prepared["spans"]) != 1: raise DemoStopped("evidence_count_mismatch") original, span = prepared["originals"][0], prepared["spans"][0] rendered = prepared["unit"]["deterministic_text"] if (original["source_record_version_id"] != version_id or original["text"] != expected_text or original["role"] != "human" or original["occurred_at"] is not None or span["source_record_version_id"] != version_id or span["text_sha256"] != "sha256:" + hashlib.sha256(expected_text.encode()).hexdigest() or rendered[span["rendered_start"]:span["rendered_end"]] != expected_text or rendered.encode()[span["rendered_utf8_start"]:span["rendered_utf8_end"]] != expected_text.encode() or context["context_bytes"] != len(rendered.encode())): raise DemoStopped("exact_evidence_mismatch") if client.original(UUID(version_id))["original"]["text"] != expected_text: raise DemoStopped("original_readback_mismatch") return {"context_status": "prepared", "selected_count": 1, "version_id": version_id, "exact_evidence_verified": True} def run_demo(transport, config, run_id, step): """Execute only the requested stage; earlier exports replay identical keys. Inject a request-compatible transport for tests. There are no automatic retries. Repeating a completed earlier stage after a later stage changes current state will fail its current-context check; repeat only the interrupted stage. """ from brain.client.application_conversations import ApplicationConversations from brain.ingest.application_conversation_contract import ( CONTRACT, ApplicationWithdrawalSubmission, canonical_hash, source_payload, ) if re.fullmatch(r"[A-Za-z0-9._:-]{1,32}", run_id) is None: raise DemoStopped("invalid_run_id") if step not in {"ingest", "correct", "withdraw"}: raise DemoStopped("invalid_step") client = ApplicationConversations(transport, brain_id=config.brain_id, connection_id=config.connection_id, credential=config.credential) thread_id = "quickstart:" + run_id + ":thread" first = _saved(client, _export(run_id, text=INITIAL_TEXT)) if step == "ingest": return {"step": step, **_evidence(client, thread_id, first["version_id"], INITIAL_TEXT)} corrected = _saved(client, _export(run_id, text=CORRECTED_TEXT, expected_version_id=first["version_id"], correction=True)) if corrected["version_id"] == first["version_id"]: raise DemoStopped("correction_did_not_advance_version") if step == "correct": return {"step": step, **_evidence(client, thread_id, corrected["version_id"], CORRECTED_TEXT)} # The SDK intentionally has no withdrawal helper. Use the existing typed route. body = ApplicationWithdrawalSubmission(external_id="quickstart:" + run_id + ":message", expected_version_id=UUID(corrected["version_id"])) response = transport.request("POST", client.prefix + "/withdrawals", headers={**client.headers, "Idempotency-Key": "quickstart:" + run_id + ":withdraw"}, json=body.model_dump(mode="json")) response.raise_for_status() receipt = response.json() if (receipt["contract"] != CONTRACT or receipt["outcome"] != "deleted" or receipt["custody_status"] != "saved" or receipt["body_hash"] is not None or receipt["context_status"] != "not_integrated" or receipt["search_ready"] is not False or receipt["semantic_ready"] is not False or receipt["payload_hash"] != canonical_hash(source_payload(body)) or receipt["request_digest"] != canonical_hash({"contract": CONTRACT, "kind": "ApplicationWithdrawalSubmission", "submission": body.model_dump(mode="json")}) or client.receipt(UUID(receipt["receipt_id"])) != receipt): raise DemoStopped("withdrawal_receipt_mismatch") context = client.recall(thread_id) if (context["context_status"] != "empty" or context["selected_count"] != 0 or context["prepared"] is not None or context["has_more"] is not False or context["omission_reasons"] or context["publication_status"] != "held" or context["semantic_ready"] is not False or context["accepted_memory"] is not False): raise DemoStopped("withdrawal_not_reflected") return {"step": step, "context_status": "empty", "selected_count": 0, "immutable_history": "retained; withdrawal is not erasure"} def main(argv=None, *, environ=None, transport_factory=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--execute", action="store_true", help="write fictional data to the authorized test connection") parser.add_argument("--run-id", help="stable identity, 1–32 ASCII letters/digits or . _ : -; reuse on explicit replay") parser.add_argument("--step", choices=("ingest", "correct", "withdraw"), help="required with --execute") args = parser.parse_args(argv) env = os.environ if environ is None else environ if not args.execute: missing = [name for name in REQUIRED_ENV if not env.get(name)] problem = None try: configuration(env) except (DemoStopped, ValueError) as error: problem = str(error) if isinstance(error, DemoStopped) else "invalid_configuration" print(json.dumps({"mode": "preflight", "network_requests": 0, "writes": 0, "configuration_valid": problem is None, "missing_environment": missing, "configuration_problem": problem, "authorization_verified": False})) return 0 if args.run_id is None or args.step is None: parser.error("--execute requires --run-id and --step") try: config = configuration(env) if transport_factory is None: import httpx transport_factory = httpx.Client with transport_factory(base_url=config.base_url, timeout=10, follow_redirects=False) as transport: result = run_demo(transport, config, args.run_id, args.step) print(json.dumps({"mode": "executed", "run_id": args.run_id, **result})) return 0 except Exception as error: # HTTP exceptions may carry a private origin or body. Stop without logging them. print(json.dumps({"mode": "stopped", "step": args.step, "run_id": args.run_id, "code": str(error) if isinstance(error, DemoStopped) else "request_or_response_unresolved", "action": "Inspect the private server receipt/error. If unresolved, explicitly repeat this exact stage and run ID; no automatic retry occurred."})) return 1 if __name__ == "__main__": raise SystemExit(main())