Move a retired Claude Sonnet 4 API application to Sonnet 5.5 and check the result

"""Reader reference for one owned, nonstreaming direct Messages call path.

Documentation-only. BIG CHANGE did not send a request or run a reader case.
Supply real, approved messages, case rules, client-tool handler and output validator.
This module deliberately makes one HTTP attempt per turn and never retries a write.
"""

import copy
import base64
import hashlib
import json
import urllib.error
import urllib.request


MODEL = "claude-sonnet-5-5"
ENDPOINT = "https://api.anthropic.com/v1/messages"


class Hold(Exception):
    """A migration case needs an owner decision or a corrected revision."""


class TransportHold(Hold):
    def __init__(self, reason, metadata):
        super().__init__(reason)
        self.metadata = metadata


def _raw_record(raw):
    return {"raw_body_base64": base64.b64encode(raw[:65536]).decode("ascii"),
            "raw_body_sha256": hashlib.sha256(raw).hexdigest(),
            "raw_body_truncated": len(raw) > 65536}


def _case_authority(old_request, case_policy):
    if not isinstance(case_policy, dict) or case_policy.get("authorized") is not True or not case_policy.get("case_id"):
        raise Hold("Frozen owner case authority and ID required")
    tools = old_request.get("tools", [])
    if not isinstance(tools, list) or any(not isinstance(t, dict) or not t.get("name") for t in tools):
        raise Hold("Malformed captured tool declarations")
    names = [t["name"] for t in tools]
    if len(set(names)) != len(names):
        raise Hold("Duplicate declared tool names")
    allowed = case_policy.get("allowed_tool_names")
    required = case_policy.get("required_tool_names")
    if not isinstance(allowed, list) or not isinstance(required, list) or len(set(allowed)) != len(allowed) or len(set(required)) != len(required):
        raise Hold("Freeze unique allowed/required tool names")
    allowed, required = set(allowed), set(required)
    if not allowed <= set(names) or not required <= allowed:
        raise Hold("Case authority exceeds captured declarations")
    minimum, maximum = case_policy.get("min_tool_calls"), case_policy.get("max_tool_calls")
    if type(minimum) is not int or type(maximum) is not int or minimum < 0 or maximum < minimum:
        raise Hold("Freeze bounded integer tool call count")
    if type(case_policy.get("allow_parallel")) is not bool:
        raise Hold("Freeze parallel-call authority")
    choice = old_request.get("tool_choice", {"type": "auto"} if names else {"type": "none"})
    if not isinstance(choice, dict) or choice.get("type") not in ("none", "auto", "any", "tool"):
        raise Hold("Unknown captured tool choice")
    kind = choice["type"]
    if kind == "none" and (allowed or required or minimum or maximum):
        raise Hold("Captured none forbids tool calls")
    if kind == "tool" and (choice.get("name") not in names or allowed != {choice["name"]} or choice["name"] not in required or minimum < 1):
        raise Hold("Specific-tool case must require exact captured name")
    if kind == "any" and (not required or minimum < 1):
        raise Hold("Forced-any completion needs explicit required name(s)")
    if kind in ("any", "tool") and case_policy.get("approve_forced_to_auto") is not True:
        raise Hold("Owner must approve forced-choice completion change")
    if choice.get("disable_parallel_tool_use") is True and case_policy["allow_parallel"]:
        raise Hold("Cannot widen captured parallel restriction")
    return choice, allowed, required, minimum, maximum


def _object_schema_is_strict(schema):
    if not isinstance(schema, dict):
        return False
    if schema.get("type") == "object" and schema.get("additionalProperties") is not False:
        return False
    children = list(schema.get("properties", {}).values())
    if "items" in schema:
        children.append(schema["items"])
    for key in ("allOf", "anyOf", "oneOf"):
        children.extend(schema.get(key, []))
    return all(_object_schema_is_strict(v) for v in children)


def build_request(old_request, approved_messages, *, case_policy, effort="medium", mode=None):
    """Return a separate candidate request and a record of deliberate field edits.

    The caller first captures the old serialized request and effective API version.
    A real application's remaining request fields, cases and output contract survive.
    Unsupported branches hold instead of silently changing application behavior.
    """
    choice, _, _, _, _ = _case_authority(old_request, case_policy)
    if old_request.get("model") != "claude-sonnet-4-20250514":
        raise Hold("Expected the actual retired Sonnet 4 direct API call path")
    if old_request.get("stream") or old_request.get("streaming"):
        raise Hold("Streaming path requires its own terminal-event and partial-input adapter")
    if not approved_messages or approved_messages[-1].get("role") != "user":
        raise Hold("Replace a last-assistant prefill under owner review; end with a user turn")
    for message in approved_messages:
        content = message.get("content")
        blocks = content if isinstance(content, list) else []
        if any(isinstance(b, dict) and b.get("type") in ("image", "document") for b in blocks):
            raise Hold("Image/document path needs separate token and parser review")
    request = copy.deepcopy(old_request)
    changes = []
    request["model"] = MODEL
    request["messages"] = copy.deepcopy(approved_messages)
    request.pop("stream", None)
    for key in ("temperature", "top_p", "top_k"):
        if key in request:
            del request[key]
            changes.append(f"removed {key}, including any default-valued serialization")
    if "output_format" in request:
        old_format = request.pop("output_format")
        request.setdefault("output_config", {})["format"] = old_format
        changes.append("moved output_format to output_config.format; validate real schema")
    previous_thinking = request.get("thinking", {})
    if mode is None:
        mode = "adaptive" if previous_thinking.get("type") == "enabled" else "between_tools"
    if mode not in ("adaptive", "between_tools") or effort not in ("low", "medium", "high", "xhigh", "max"):
        raise Hold("Choose a documented thinking mode and effort from owner cases")
    if mode == "between_tools" and effort not in ("low", "medium", "high"):
        raise Hold("between_tools is supported at low, medium or high effort")
    request["thinking"] = {"type": mode}
    request.setdefault("output_config", {})["effort"] = effort
    changes.append("replaced disabled/omitted or enabled-budget thinking with explicit mode and effort")
    for tool in request.get("tools", []):
        if "type" in tool or "input_schema" not in tool:
            raise Hold("Server/built-in or unknown tool requires its exact current toolset route")
        if not _object_schema_is_strict(tool["input_schema"]):
            raise Hold("Client tool schema needs supported strict JSON Schema and additionalProperties:false")
        tool["strict"] = True
    if choice["type"] in ("any", "tool"):
        request["tool_choice"] = {**choice, "type": "auto"}
        request["tool_choice"].pop("name", None)
        changes.append("owner-approved forced choice to auto with frozen tool case authority")
    elif "tool_choice" in old_request:
        request["tool_choice"] = copy.deepcopy(choice)
    for key in ("anthropic_beta", "betas"):
        if key in request:
            raise Hold("Review and remove legacy beta headers in the actual HTTP header layer")
    if type(request.get("max_tokens")) is not int or request["max_tokens"] <= 0:
        raise Hold("Owner must set an actual bounded max_tokens after token recount")
    return request, changes


def direct_http_once(request, *, api_key, api_version, timeout=60):
    """One paid HTTP attempt; caller supplies a secret from private storage."""
    if not api_key or not api_version:
        raise Hold("Missing authorized API key or recorded effective anthropic-version")
    body = json.dumps(request, separators=(",", ":")).encode("utf-8")
    call = urllib.request.Request(
        ENDPOINT,
        data=body,
        headers={"x-api-key": api_key, "anthropic-version": api_version, "content-type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(call, timeout=timeout) as result:
            raw = result.read()
            metadata = {"status": result.status, "request_id": result.headers.get("request-id") or result.headers.get("x-request-id"), **_raw_record(raw)}
            if result.status != 200:
                raise TransportHold("Messages non-200 response", metadata)
            try:
                message = json.loads(raw)
                return {"message": message, "http": {"status": result.status,
                        "request_id": metadata["request_id"],
                        "raw_body_sha256": metadata["raw_body_sha256"]}}
            except (UnicodeError, json.JSONDecodeError) as exc:
                raise TransportHold("Messages invalid JSON", metadata) from exc
    except urllib.error.HTTPError as exc:
        raw = exc.read()
        raise TransportHold("Messages HTTP error", {"status": exc.code, "request_id": exc.headers.get("request-id") or exc.headers.get("x-request-id"), **_raw_record(raw)}) from exc
    except (urllib.error.URLError, TimeoutError) as exc:
        raise TransportHold("Transport outcome unknown", {"status": None, "request_id": None, "error_type": type(exc).__name__}) from exc


def _check_usage(usage):
    if not isinstance(usage, dict) or any(k not in usage for k in ("input_tokens", "output_tokens")):
        raise Hold("Mandatory raw usage absent")
    for key, value in usage.items():
        if key.endswith("_tokens") and (type(value) is not int or value < 0):
            raise Hold(f"Invalid raw usage counter: {key}")
        if key == "cache_creation" and isinstance(value, dict):
            for subkey, subvalue in value.items():
                if subkey.endswith("_tokens") and (type(subvalue) is not int or subvalue < 0):
                    raise Hold(f"Invalid raw usage counter: {subkey}")


def run_case(request, *, case_policy, identity, reconcile_prior, transport, execute_tool, validate_result, journal, max_turns=4):
    """Inspect a complete owner case; callbacks belong to the real application.

    transport(request) returns {message: parsed Messages JSON, http: status,
    request_id and raw_body_sha256}. A custom parsed-JSON-only transport is accepted
    but its HTTP status and request ID remain unknown. execute_tool(case_id, tool_use_id, name, input)
    returns a dict with content, effect_state (none/confirmed), and an owner receipt.
    validate_result(text, effects) is the owner's frozen schema/task/effect oracle.
    journal(event) durably commits private attempts, responses, intents and receipts.
    reconcile_prior(binding) returns True only when no earlier attempt or intent is
    unresolved after authoritative readback. It must not assume idempotency.
    Unknown effects, absent usage or incomplete turns hold. No automatic retry.
    """
    _, allowed, required, minimum, maximum = _case_authority(request, case_policy)
    if not isinstance(identity, dict) or identity.get("case_id") != case_policy["case_id"] or not all(identity.get(k) for k in ("code_revision", "original_request_sha256", "api_version")):
        raise Hold("Bind case to exact original request, code revision and API version")
    current = copy.deepcopy(request)
    observations = []
    effects = []
    called_names = []
    seen_tool_ids = set()
    if journal is None or type(max_turns) is not int or max_turns < 1:
        raise Hold("Provide a durable private case journal and positive turn bound")
    binding = {**identity, "candidate_request_sha256": hashlib.sha256(json.dumps(request, sort_keys=True, separators=(",", ":")).encode()).hexdigest(), "model": request.get("model")}

    def emit(kind, **fields):
        try:
            journal({"event": kind, "binding": copy.deepcopy(binding), **fields})
        except Exception as exc:
            raise Hold("Durable journal failed; effect/attempt state uncertain; authoritative readback required") from exc

    try:
      try:
        reconciled = reconcile_prior is not None and reconcile_prior(copy.deepcopy(binding)) is True
      except Exception as exc:
        raise Hold("Prior-intent reconciliation failed; authoritative readback required") from exc
      if not reconciled:
        raise Hold("Earlier attempt/intent unresolved; authoritative readback before rerun")
      for turn in range(1, max_turns + 1):
        request_hash = hashlib.sha256(json.dumps(current, sort_keys=True, separators=(",", ":")).encode()).hexdigest()
        emit("request_attempt", turn=turn, request_sha256=request_hash, request_raw=copy.deepcopy(current))
        try:
            transported = transport(current)
        except Exception as exc:
            emit("request_error", turn=turn, request_sha256=request_hash, reason=str(exc), metadata=getattr(exc, "metadata", {"error_type": type(exc).__name__}))
            raise Hold("Transport outcome/error requires exact-case reconciliation") from exc
        if isinstance(transported, dict) and "message" in transported and "http" in transported:
            response = transported["message"]
            http = copy.deepcopy(transported["http"])
            invalid_http = not isinstance(http, dict) or type(http.get("status")) is not int or http["status"] != 200
        else:
            response = transported
            http = {"status": None, "request_id": None, "raw_body_sha256": None}
            invalid_http = False
        entry = {
            "turn": turn,
            "http": copy.deepcopy(http),
            "response_id": response.get("id") if isinstance(response, dict) else None,
            "stop_reason": response.get("stop_reason") if isinstance(response, dict) else None,
            "usage_raw": copy.deepcopy(response.get("usage")) if isinstance(response, dict) else None,
            "response_raw": copy.deepcopy(response),
        }
        emit("response", turn=turn, request_sha256=request_hash, http=copy.deepcopy(http),
             response_id=entry["response_id"], response_raw=copy.deepcopy(response))
        observations.append(entry)
        if invalid_http:
            raise Hold("Successful transport envelope needs observed HTTP 200 metadata")
        if not isinstance(response, dict):
            raise Hold("Messages response is not a JSON object")
        if response.get("model") != MODEL:
            raise Hold("Returned model differs from the candidate ID")
        usage = response.get("usage")
        _check_usage(usage)
        content = response.get("content")
        if not isinstance(content, list):
            raise Hold("Missing typed content blocks")
        text = []
        calls = []
        for block in content:
            if not isinstance(block, dict):
                raise Hold("Malformed typed content block")
            kind = block.get("type")
            if kind == "text":
                if not isinstance(block.get("text"), str):
                    raise Hold("Malformed text block")
                text.append(block.get("text", ""))
            elif kind == "tool_use":
                if not block.get("id") or not block.get("name"):
                    raise Hold("Malformed client tool call")
                calls.append(block)
            elif kind in ("thinking", "redacted_thinking"):
                pass  # Preserve complete signed block in assistant history below.
            else:
                raise Hold(f"Unexpected content block type: {kind}")
        stop = response.get("stop_reason")
        if stop == "tool_use":
            if not calls or execute_tool is None:
                raise Hold("Required client tool continuation unavailable")
            ids = [b.get("id") for b in calls]
            names = [b.get("name") for b in calls]
            if any(not isinstance(i, str) or not i for i in ids) or len(ids) != len(set(ids)) or set(ids) & seen_tool_ids or any(n not in allowed for n in names):
                raise Hold("Duplicate/malformed tool IDs or undeclared/unauthorized tool names")
            if not case_policy["allow_parallel"] and len(calls) > 1:
                raise Hold("Parallel tool calls exceed frozen case authority")
            if len(called_names) + len(calls) > maximum:
                raise Hold("Tool call count exceeds frozen case authority")
            parsed = []
            for block in calls:
                value = block.get("input")
                if isinstance(value, str):
                    try:
                        value = json.loads(value)
                    except json.JSONDecodeError as exc:
                        raise Hold("Client tool input is invalid JSON") from exc
                if not isinstance(value, dict):
                    raise Hold("Client tool input is not a JSON object")
                parsed.append(value)
            current["messages"].append({"role": "assistant", "content": copy.deepcopy(content)})
            results = []
            for block, value in zip(calls, parsed):
                emit("tool_intent", turn=turn, tool_use_id=block["id"], name=block["name"], input=copy.deepcopy(value))
                try:
                    outcome = execute_tool(identity["case_id"], block["id"], block["name"], value)
                except Exception as exc:
                    emit("tool_unknown", turn=turn, tool_use_id=block["id"], name=block["name"], error_type=type(exc).__name__)
                    raise Hold("Tool effect unknown; reconcile exact call by readback before any retry") from exc
                if not isinstance(outcome, dict):
                    emit("tool_unknown", turn=turn, tool_use_id=block["id"], name=block["name"], reason="missing receipt object")
                    raise Hold("Tool effect unknown; handler supplied no receipt object")
                if outcome.get("effect_state") not in ("none", "confirmed") or not outcome.get("receipt") or "content" not in outcome:
                    emit("tool_unknown", turn=turn, tool_use_id=block["id"], name=block["name"], reason="uncertain receipt")
                    raise Hold("Tool effect uncertain; reconcile by tool readback before any retry")
                emit("tool_receipt", turn=turn, tool_use_id=block["id"], name=block["name"], effect_state=outcome["effect_state"], receipt=copy.deepcopy(outcome["receipt"]))
                effects.append(outcome["receipt"])
                seen_tool_ids.add(block["id"])
                results.append({"type": "tool_result", "tool_use_id": block["id"], "content": outcome["content"]})
                called_names.append(block["name"])
            current["messages"].append({"role": "user", "content": results})
            continue
        if stop != "end_turn":
            raise Hold(f"Nonterminal or failed stop_reason={stop}; inspect and repair before rerun")
        if calls or len(called_names) < minimum or not required <= set(called_names):
            raise Hold("A required tool was not completed; auto tool choice can make no call")
        answer = "".join(text)
        if not answer.strip() or validate_result is None:
            raise Hold("Empty answer or missing owner result validator")
        try:
            accepted = validate_result(answer, effects)
        except Exception as exc:
            raise Hold("Owner result/schema validator failed") from exc
        if accepted is not True:
            raise Hold("Empty, schema-invalid, task-failing or effect-failing result")
        emit("case_decision", decision="task_checks_passed", numerical_review="pending",
             release_acceptance="pending_owner_review", effects=copy.deepcopy(effects), answer=answer)
        return {"state": "task_checks_passed_numerical_pending", "answer": answer,
                "effects": effects, "observations": observations}
      raise Hold("Bounded tool turns exhausted; record partial effects and hold")
    except Exception as exc:
        try:
            emit("case_decision", decision="hold", reason=str(exc), effects=copy.deepcopy(effects))
        except Hold:
            pass  # Journal outage is uncertain: reconcile before rerun.
        if isinstance(exc, Hold):
            raise
        raise Hold("Unexpected case failure; reconcile exact journal/intents") from exc

import os
import hashlib
import json
import time
from direct_messages_path import build_request, direct_http_once, run_case

candidate, field_changes = build_request(
    captured_old_request,
    approved_case["messages"],
    case_policy=approved_case["tool_policy"],
    effort="medium",  # owner evaluates this level against its frozen cases
)
case_identity = {
    "case_id": approved_case["tool_policy"]["case_id"],
    "code_revision": captured_code_revision,
    "original_request_sha256": captured_old_request_sha256,
    "api_version": captured_effective_api_version,
}
started_ns = time.monotonic_ns()
observed = run_case(
    candidate,
    case_policy=approved_case["tool_policy"],
    identity=case_identity,
    reconcile_prior=the_app_authoritative_prior_intent_reconciler,
    transport=lambda req: direct_http_once(
        req,
        api_key=os.environ["ANTHROPIC_API_KEY"],
        api_version=captured_effective_api_version,
    ),
    execute_tool=the_app_authorized_tool_handler,  # (case_id, tool_use_id, name, input)
    validate_result=the_app_frozen_case_validator,
    journal=the_app_private_durable_case_journal,
)
elapsed_ms = (time.monotonic_ns() - started_ns) / 1_000_000
handoff = {
    "event": "numerical_review_pending",
    "binding": {
        **case_identity,
        "candidate_request_sha256": hashlib.sha256(
            json.dumps(candidate, sort_keys=True, separators=(",", ":")).encode()
        ).hexdigest(),
    },
    "task_state": observed["state"],  # task_checks_passed_numerical_pending
    "observed_elapsed_ms": elapsed_ms,
    "turns": [
        {"http": turn["http"], "response_id": turn["response_id"],
         "usage_raw": turn["usage_raw"]}
        for turn in observed["observations"]
    ],
    "effect_receipts": observed["effects"],
    "frozen_thresholds": approved_case["thresholds"],
    "cost_and_latency_decision": "pending_owner_review",
}
the_app_private_durable_case_journal(handoff)