Ingest. Recall. Correct. Withdraw.

"""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())