Agent Advisor

scripts/openai_model_recommendation.py

"""Deterministic OpenAI-to-Bedrock model/path recommendation for agent-advisor.

Factual baseline: an internal OpenAI-to-Bedrock migration reference (2026-06 live
probes and its recorded feature-status notes). The engine is offline:
no openai/boto3/token-generator imports at module load. Rankings are only produced
where the reference provides evidence; otherwise the engine returns an explicit
decision or a provisional limitation rather than inventing a comparative order.
"""

import re

import anthropic_model_recommendation

# --- Source model-family detection (ported from the reference compatibility matrix) ---
# Reasoning generation: gpt-5 / gpt-6 variants and o1/o3/o4.
_REASONING_PATTERNS = (
    re.compile(r"(?i)(^|[^a-z0-9])gpt[-_ ]?[56](\.\d+)?([^0-9]|$)"),
    re.compile(r"(?i)(^|[^a-z0-9])o[134](-mini|-pro|-preview)?([^a-z0-9]|$)"),
)
# Legacy generation: gpt-4*, gpt-3.5 / gpt-35.
_LEGACY_PATTERNS = (
    re.compile(r"(?i)gpt[-_ ]?4o"),
    re.compile(r"(?i)gpt[-_ ]?4(?!\d)"),
    re.compile(r"(?i)gpt[-_ ]?4[-_ ]?32k"),
    re.compile(r"(?i)gpt[-_ ]?3\.?5"),
    re.compile(r"(?i)gpt[-_ ]?35"),
)
# GPT-5.2+ re-accepts sampling on Responses; earlier reasoning models differ.
_GPT5_MINOR = re.compile(r"(?i)gpt[-_ ]?5\.(\d+)")


def detect_family(model_id):
    """Return 'reasoning', 'legacy', or 'unknown'.

    Fails to 'unknown' for opaque deployment names (e.g. 'prod-default') so the
    engine never infers a model family from a deployment alias.
    """
    name = (model_id or "").strip()
    if not name:
        return "unknown"
    for pat in _REASONING_PATTERNS:
        if pat.search(name):
            return "reasoning"
    for pat in _LEGACY_PATTERNS:
        if pat.search(name):
            return "legacy"
    return "unknown"


def _gpt5_minor(model_id):
    match = _GPT5_MINOR.search(model_id or "")
    return int(match.group(1)) if match else None


def _primary_source_id(source):
    ids = source.get("model_ids") or []
    return ids[0] if ids else ""


def _source_family(source):
    """Family for the workload: reasoning if ANY id is reasoning, else legacy if
    any is legacy, else unknown (opaque deployment names stay unknown)."""
    families = [detect_family(mid) for mid in source.get("model_ids") or [""]]
    if "reasoning" in families:
        return "reasoning"
    if "legacy" in families:
        return "legacy"
    return "unknown"


def _finding(code, tag, message, remediation):
    return {"code": code, "tag": tag, "message": message, "remediation": remediation}


def _delta(code, category, description):
    return {"code": code, "category": category, "description": description}


# --- Feature vocabulary (provider-scoped; codes match scanner feature-catalog) ---
_HOSTED_TOOL_IMPACTS = {
    "web_search": (
        "Hosted web search does not execute live on Bedrock Mantle (verified passthrough: "
        "no query issued, no citations).",
        "Implement a client-side tool loop, or expose a server-side tool (MCP/Lambda) through "
        "Bedrock AgentCore Gateway; use Knowledge Bases for document retrieval.",
    ),
    "file_search_retrieval": (
        "Hosted file search / retrieval is not a drop-in on Bedrock.",
        "Re-platform retrieval onto Bedrock Knowledge Bases or application-owned retrieval.",
    ),
    "files_api": (
        "The OpenAI Files API and vector stores do not port directly.",
        "Redesign data ingestion and retrieval; use Knowledge Bases or application storage.",
    ),
    "vector_stores": (
        "OpenAI vector stores are not a Bedrock primitive.",
        "Move vectors to a Bedrock-supported store and retrieve from the application.",
    ),
    "assistants_threads": (
        "Assistants/Threads hosted state is not available on Bedrock.",
        "Redesign agent state, tools, and memory (e.g. Bedrock AgentCore).",
    ),
    "audio_modality": (
        "Audio is not a single-call text-model capability on Bedrock.",
        "Split into a verified STT/TTS service path (Amazon Transcribe / Polly).",
    ),
    "embeddings": (
        "Embeddings require a separate Bedrock embedding model, not the text model.",
        "Select a sourced Bedrock embedding model (e.g. Titan/Cohere) as a separate workload.",
    ),
    "images": (
        "Image generation/editing is a separate capability, not a text-model swap.",
        "Select a sourced Bedrock image model/service as a separate workload.",
    ),
}


# Converse tier map: a governance workload keeps its capability tier when moving
# from the (Mantle-only) GPT-5.x family to Claude on runtime Converse.
#   GPT-5.6 Sol (frontier)      -> Claude Opus 5.5
#   GPT-5.6 Terra / 5.5 / 5.4   -> Claude Sonnet 5 (also the default tier)
#   GPT-5.6 Luna (fast/low-cost)-> Claude Haiku 4.5
_CONVERSE_TIER_DEFAULT = "anthropic_claude_sonnet_5"
_CONVERSE_TIER_ORDER = (
    "anthropic_claude_sonnet_5",
    "anthropic_claude_opus_5_5",
    "anthropic_claude_haiku_4_5",
)


def _converse_tier_for_source(source):
    sid = _primary_source_id(source).lower()
    if "5.6-sol" in sid or "6-astra" in sid:
        return "anthropic_claude_opus_5_5"
    if "5.6-luna" in sid:
        return "anthropic_claude_haiku_4_5"
    # 5.6-terra, 5.5, 5.4, legacy, and unknown sources all map to the balanced tier.
    return _CONVERSE_TIER_DEFAULT


_SOURCE_MODEL_KEYS = {
    "gpt-5.6-sol": "openai_gpt_5_6_sol",
    "gpt-5.6-terra": "openai_gpt_5_6_terra",
    "gpt-5.6-luna": "openai_gpt_5_6_luna",
    "gpt-6-astra": "openai_gpt_6_astra",
    "gpt-6-sol": "openai_gpt_6_sol",
    "gpt-6-luna": "openai_gpt_6_luna",
    "gpt-6.1-sol": "openai_gpt_6_1_sol",
}


def _same_model_catalog_key(source):
    """Return the exact catalog key for a recognized source model identity."""
    sid = re.sub(r"[_ ]", "-", _primary_source_id(source).lower())
    for identity in sorted(_SOURCE_MODEL_KEYS, key=len, reverse=True):
        if re.search(rf"(?<![a-z0-9]){re.escape(identity)}(?![a-z0-9])", sid):
            return _SOURCE_MODEL_KEYS[identity]
    return None


def _same_model_runtime_key(source):
    """Return the exact source model when it has a runtime_converse path."""
    return _same_model_catalog_key(source)


def _converse_candidate_order(source, requirements=None):
    tier = (
        "anthropic_claude_opus_5_5"
        if (requirements or {}).get("priority") == "quality"
        else _converse_tier_for_source(source)
    )
    order = [tier] + [k for k in _CONVERSE_TIER_ORDER if k != tier]
    same = _same_model_runtime_key(source)
    if same:
        # Same model outranks any cross-family tier: keeping the model removes
        # prompt-adaptation and behavior-delta risk entirely.
        order.insert(0, same)
    return order


def _mantle_path(source, requirements):
    same = _same_model_catalog_key(source)
    if (
        source.get("api_surface") == "chat_completions"
        and _api_continuity(requirements) in {"required", "preferred"}
        and same is not None
        and same.startswith("openai_gpt_6")
    ):
        return "mantle_openai_chat"
    return "mantle_openai_responses"


def _mantle_candidate_order(catalog, source):
    same = _same_model_catalog_key(source)
    keys = list(catalog["models"])
    return ([same] if same else []) + [key for key in keys if key != same]


def _hard_profile_requirement(requirements):
    """Whether runtime selection must prove a specific CRIS profile is callable."""
    residency = requirements.get("data_residency")
    return bool(
        requirements.get("inference_profile_id")
        or residency == "global_allowed"
        or (residency == "geo_required" and requirements.get("cris_geography"))
    )


def _catalog_model_for_path(catalog, path, detected_features=None, requirements=None,
                            candidate_order=None, region=None):
    """Pick the first available catalog model for `path` whose capability evidence
    covers the required/detected text features (evidence-driven fallback: a newer
    model without feature-level evidence is skipped in favor of an older probed
    one). Mantle candidates with cataloged region availability are skipped outside
    that footprint; runtime Converse availability remains governed by CRIS resolution.
    When no candidate covers the requirements, return the first available model together
    with its unmet-capability list so the caller can fail closed. Deterministic:
    iterates `candidate_order` when given (e.g. the Converse tier map), else
    catalog key order."""
    first = None
    first_unmet = None
    keys = candidate_order if candidate_order is not None else list(catalog["models"])
    for model_key in keys:
        model = catalog["models"].get(model_key)
        if model is None:
            continue
        path_config = model["paths"].get(path, {})
        if path_config.get("available") is not True:
            continue
        if "capabilities" in path_config:
            model = {**model, "capabilities": path_config["capabilities"]}
        availability = model.get("region_availability")
        if path.startswith("mantle_") and region and region != "unknown" and availability:
            supported_regions = set(availability.get("commercial", []))
            supported_regions.update(availability.get("govcloud", []))
            if region not in supported_regions:
                continue
        if (
            path == "runtime_converse"
            and not path_config.get("inference_profile_regions")
            and path_config.get("requires_cris")
            and _hard_profile_requirement(requirements or {})
            and _resolve_invocation_model_id(
                path_config["model_id"], True, requirements or {}, model, region
            )
            is None
        ):
            continue
        unmet = _unsupported_required_capabilities(
            model, detected_features, requirements or {}
        )
        if first is None:
            first = (model_key, model, path_config)
            first_unmet = unmet
        if not unmet:
            return (model_key, model, path_config), []
    return first, first_unmet


def _resolve_invocation_model_id(model_id, requires_cris, requirements, model=None, region=None):
    if (model or {}).get("inference_profiles"):
        return anthropic_model_recommendation._resolve_invocation_model_id(
            model_id, requires_cris, requirements, model, region
        )
    if not requires_cris:
        return model_id
    explicit = requirements.get("inference_profile_id")
    if explicit:
        return explicit
    residency = requirements.get("data_residency", "unknown")
    if residency == "global_allowed":
        return f"global.{model_id}"
    if residency == "geo_required" and requirements.get("cris_geography"):
        return f"{requirements['cris_geography']}.{model_id}"
    return None


def _api_continuity(requirements):
    """Provider-neutral continuity: required | preferred | not_required | unknown.

    Accepts the explicit `api_continuity` field; falls back to the OpenAI-specific
    `preserve_openai_api` boolean. Never reuses Anthropic's preserve_messages_api.
    """
    explicit = requirements.get("api_continuity")
    if explicit:
        return explicit
    if requirements.get("preserve_openai_api") is True:
        return "required"
    if requirements.get("preserve_openai_api") is False:
        return "not_required"
    return "unknown"


def _runtime_required(requirements):
    return bool(requirements.get("governance")) or requirements.get(
        "multi_model_converse", False
    )


def _model_identity(model_key, model, path_config, invocation_model_id=None):
    return {
        "model_key": model_key,
        "display_name": model["display_name"],
        "family": model["family"],
        "version": model["version"],
        "context_window": model["context_window"],
        "output_token_ceiling": model["output_token_ceiling"],
        "path_model_id": path_config["model_id"],
        "requires_cris": anthropic_model_recommendation._invocation_requires_cris(
            path_config, invocation_model_id
        ),
    }


def _numeric_requirement_conflict(model, requirements):
    """Fail closed: an unknown catalog limit cannot satisfy a hard numeric need."""
    conflicts = []
    for req_key, cat_key, label in (
        ("min_context_tokens", "context_window", "context window"),
        ("expected_output_tokens", "output_token_ceiling", "output ceiling"),
    ):
        need = requirements.get(req_key)
        if not need:
            continue
        have = model.get(cat_key)
        if have == "unknown" or not isinstance(have, int):
            conflicts.append(
                f"{label} for {model['display_name']} is unknown in the catalog; "
                f"a required {need}-token need cannot be confirmed"
            )
        elif have < need:
            conflicts.append(
                f"{label} for {model['display_name']} ({have}) is below the required {need}"
            )
    return conflicts


# Extracts a numeric OpenAI version (major or major.minor) from an id/version.
_OPENAI_VERSION = re.compile(r"(?i)gpt[-_ ]?(\d+(?:\.\d+)?)")


def _openai_version(text):
    match = _OPENAI_VERSION.search(text or "")
    return match.group(1) if match else None


def _version_tuple(value):
    if not value:
        return None
    parts = value.split(".")
    try:
        return (int(parts[0]), int(parts[1]) if len(parts) > 1 else 0)
    except (ValueError, IndexError):
        return None


def _target_accepts_sampling(target_model):
    """GPT-5.2+ re-accepts sampling params on Responses (verified for GPT-5.4).

    Returns True/False for an OpenAI target of known version, None when the
    target is not an OpenAI model or its version can't be parsed.
    """
    if target_model is None or target_model.get("generation") == "bedrock_native":
        return None
    if target_model.get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)":
        return None
    version = _version_tuple(target_model.get("version"))
    if version is None:
        return None
    return version >= (5, 2)


def _source_analysis(source, target_model=None):
    """Source-side facts plus, when a target has been selected, target-derived
    version/generation change flags. Target fields stay null before selection."""
    source_id = _primary_source_id(source)
    family = _source_family(source)
    surface = source.get("api_surface")
    if target_model is None:
        target_version = None
        version_changed = None
        generation_changes = None
    else:
        target_version = target_model.get("version")
        target_generation = target_model.get("generation")
        if target_generation == "bedrock_native":
            # OpenAI source -> a Bedrock-native model is always a model change.
            version_changed = True
            generation_changes = True
        else:
            source_version = _openai_version(source_id)
            if source_version and target_version:
                version_changed = source_version != target_version
            else:
                # o-series / opaque source vs a specific GPT-5.x target: a move,
                # but we cannot pin the exact source version.
                version_changed = family in {"legacy", "reasoning"} or None
            generation_changes = family == "legacy" and target_generation == "reasoning"
    return {
        "detected_version": source_id or None,
        "target_version": target_version,
        "version_changed": version_changed,
        "source_family": family,
        "source_api_surface": surface,
        "model_generation_changes": generation_changes,
    }


def _reasoning_findings(source, requirements, target_model, path, detected_features=()):
    """Version- and surface-specific parameter findings, derived from the SELECTED
    target model and path — never from the source id, and never Anthropic's blanket
    sampling-removal rule."""
    blocks, tuning, deltas = [], [], []
    src_family = _source_family(source)
    src_id = _primary_source_id(source)
    target_generation = target_model.get("generation") if target_model else None
    target_name = target_model.get("display_name") if target_model else "the target model"

    if target_generation == "reasoning":
        tuning.append(
            _finding(
                "reasoning_token_headroom",
                "[TUNE]",
                "Reasoning models spend hidden thinking tokens against the output budget; "
                "too-small caps return status:incomplete.",
                "Size the output cap for reasoning output. The reference uses ~2.5x the legacy "
                "budget plus a 4096 floor as a STARTING heuristic, not a guaranteed value; "
                "tune from measured output distributions.",
            )
        )
    if src_family == "legacy" and target_generation == "reasoning":
        deltas.append(
            _delta(
                "model_generation_hop",
                "version",
                f"Source '{src_id}' is a GPT-4.x/legacy model migrating to a reasoning target "
                f"({target_name}): reasoning controls, token budget, and prompt behavior change.",
            )
        )

    # Sampling acceptance is a property of the TARGET model and the selected path.
    if target_model and target_model.get("sampling_parameters_supported") is False:
        remediation = (
            "Remove temperature/top_p/top_k from requests to this target; do not rescale them "
            "or disable thinking. Route user-visible controls through the existing "
            "parameter_removed confirmation before changing them."
        )
        finding = _finding(
            "sampling_parameters_removed",
            "[BLOCKS]" if "sampling_params" in detected_features else "[TUNE]",
            f"{target_name} does not support sampling controls with its required thinking mode.",
            remediation,
        )
        (blocks if finding["tag"] == "[BLOCKS]" else tuning).append(finding)
        if "sampling_params" in detected_features:
            deltas.append(_delta("sampling_parameters_removed", "feature", remediation))
    elif path == "runtime_converse":
        tuning.append(
            _finding(
                "sampling_via_converse",
                "[TUNE]",
                f"On runtime Converse, sampling is set through inferenceConfig "
                f"(temperature/topP) on {target_name}, not OpenAI SDK kwargs.",
                "Move temperature/top_p into Converse inferenceConfig; penalties/logprobs/stop "
                "are not universally supported and must be verified.",
            )
        )
    else:
        accepts = _target_accepts_sampling(target_model)
        if accepts is True:
            tuning.append(
                _finding(
                    "sampling_params_accepted",
                    "[TUNE]",
                    f"{target_name} (GPT-5.2+) accepts temperature and top_p on the Responses "
                    "surface; do not strip them.",
                    "Keep temperature/top_p and calibrate against a golden set.",
                )
            )
            tuning.append(
                _finding(
                    "sampling_penalties_rejected",
                    "[TUNE]",
                    "frequency_penalty and presence_penalty were rejected as SDK kwargs on the "
                    "verified GPT-5.4 Responses probe.",
                    "Remove frequency_penalty/presence_penalty. logprobs, logit_bias, and stop "
                    "were NOT part of that probe and are version/endpoint-specific — verify each "
                    "for the exact target model rather than assuming acceptance or rejection.",
                )
            )
        elif accepts is False:
            tuning.append(
                _finding(
                    "sampling_params_version_specific",
                    "[TUNE]",
                    f"{target_name} predates GPT-5.2 and may reject sampling parameters on "
                    "Responses; behavior is version-specific.",
                    "Verify sampling acceptance for the exact target model and surface.",
                )
            )
        elif target_model and target_model.get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)":
            tuning.append(
                _finding(
                    "sampling_params_unverified", "[TUNE]",
                    "Astra's model card does not specify accepted sampling parameters.",
                    "Probe sampling and reasoning settings on the selected API; do not "
                    "inherit GPT-5.x parameter acceptance or rejection rules.",
                )
            )
    return blocks, tuning, deltas


def _chat_to_responses_deltas():
    return [
        _delta(
            "chat_to_responses_request",
            "path",
            "Chat Completions -> Responses: messages->input, system message->instructions, "
            "max_tokens->max_output_tokens.",
        ),
        _delta(
            "chat_to_responses_response",
            "path",
            "Read model output from output_text instead of the chat message content.",
        ),
        _delta(
            "chat_to_responses_tools",
            "feature",
            "Tool results move to function_call_output on the Responses surface.",
        ),
        _delta(
            "chat_to_responses_state",
            "path",
            "Replayed multi-turn context uses previous_response_id where appropriate, or "
            "application-owned state.",
        ),
    ]


def _feature_findings(detected_features, requirements, path=None, target_model=None):
    """Structured output, n, tools, state, and hosted-tool/modality impacts.

    Findings are path-aware: Mantle Responses keeps verified capabilities
    (responses.parse typed output, hosted store=True continuation) that a
    Bedrock-native Converse target does not."""
    detected = set(detected_features or [])
    detected.update(requirements.get("critical_features") or [])
    is_responses = path == "mantle_openai_responses"
    is_chat = path == "mantle_openai_chat"
    blocks, tuning, deltas, impacts = [], [], [], []

    if "structured_output_json" in detected:
        if path == "runtime_converse" and (target_model or {}).get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)":
            deltas.append(
                _delta(
                    "structured_output_text_format",
                    "feature",
                    "Astra runtime JSON Schema output is documented for non-streaming calls. "
                    "Set outputConfig.textFormat.type=json_schema and put name plus schema "
                    "(encoded as a JSON string) in outputConfig.textFormat.structure.jsonSchema. "
                    "Also set additionalModelRequestFields.text.format.strict=true. Use an "
                    "object schema with all fields required and additionalProperties=false; "
                    "validate the schema and check refusal/incomplete responses before parsing.",
                )
            )
        elif is_responses:
            # Verified on Mantle: responses.parse(text_format=Model) returns a typed object.
            deltas.append(
                _delta(
                    "structured_output_text_format",
                    "feature",
                    "On Mantle Responses, keep typed structured output: "
                    "client.responses.parse(..., text_format=Model) is verified, or use raw "
                    "text.format json_schema.",
                )
            )
        else:
            deltas.append(
                _delta(
                    "structured_output_text_format",
                    "feature",
                    "For the selected non-Responses path, enforce the schema through that "
                    "path's supported request shape plus application validation; the OpenAI "
                    "Responses parse() helper is not available.",
                )
            )
    if requirements.get("uses_n") or "multiple_candidates_n" in detected:
        if path == "mantle_openai_chat":
            if target_model and target_model.get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)":
                tuning.append(
                    _finding(
                        "multiple_candidates_n_unverified", "[TUNE]",
                        "Astra Chat Completions support for n has not been verified.",
                        "Probe n on the selected model and endpoint before preserving it; "
                        "if unsupported, request multiple candidates with repeated calls.",
                    )
                )
        else:
            deltas.append(
                _delta(
                    "responses_no_n",
                    "feature",
                    "The Responses API does not support n; request multiple candidates with "
                    "repeated calls.",
                )
            )
    if "tool_or_function_calling" in detected:
        if is_chat:
            tool_description = (
                "On Mantle Chat Completions, preserve assistant tool_calls and continue each "
                "result as a role: tool message with the matching tool_call_id; the "
                "application executes the tool."
            )
        else:
            tool_description = (
                "Tool calls continue via function_call_output on Responses; the application "
                "executes the tool."
            )
        deltas.append(_delta("tool_result_shape", "feature", tool_description))
    if "conversation_state" in detected or requirements.get("uses_hosted_state"):
        if target_model and target_model.get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)":
            deltas.append(
                _delta(
                    "conversation_state_ownership",
                    "path",
                    "Astra's hosted state behavior has not been probed on this API. Verify "
                    "retention and continuation semantics; use application-owned history "
                    "until those requirements are confirmed.",
                )
            )
        elif is_responses:
            deltas.append(
                _delta(
                    "conversation_state_ownership",
                    "path",
                    "Two verified modes on Mantle Responses: server-hosted continuation "
                    "(store=True + previous_response_id) or manual replay (store=False with "
                    "application-owned history). Retention/compliance/availability are "
                    "verification questions, not a lost capability.",
                )
            )
        elif is_chat:
            deltas.append(
                _delta(
                    "conversation_state_ownership",
                    "path",
                    "Mantle Chat Completions uses application-owned message history; replay "
                    "the prior user, assistant, tool-call, and tool-result messages on each "
                    "request.",
                )
            )
        else:
            deltas.append(
                _delta(
                    "conversation_state_ownership",
                    "path",
                    "This path does not use hosted Responses state; carry "
                    "conversation history in an application-owned store.",
                )
            )

    impacted = sorted(detected.intersection(_HOSTED_TOOL_IMPACTS))
    for feature in impacted:
        impact, remediation = _HOSTED_TOOL_IMPACTS[feature]
        impacts.append(
            {"feature": feature, "impact": impact, "recommendation": remediation}
        )
    return blocks, tuning, deltas, impacts


# Text-model features that map to portable request-shape changes on any path.
_PORTABLE_FEATURES = {"max_tokens", "sampling_params", "sampling_params_accepted"}


# Separate-capability modalities that need their OWN target, not the text model.
# service = the Bedrock service family the reference names; candidate stays null
# (unresolved) because the reference does not pin a specific verified model ID.
_SEPARATE_MODALITY_TARGETS = {
    "embeddings": (
        "Amazon Bedrock embedding model (e.g. Titan/Cohere families)",
        "05_migrating_for_real.ipynb maps embeddings to a separate Bedrock embedding model",
    ),
    "images": (
        "Amazon Bedrock image model/service (e.g. Titan Image / Nova / Stability)",
        "03_reasoning_api_migration.ipynb maps images to a separate Bedrock image model",
    ),
    "audio_modality": (
        "Amazon Transcribe / Polly (STT/TTS)",
        "feature-catalog + notebooks map audio to Transcribe/Polly, not a text model",
    ),
}


def _additional_targets(detected_features, requirements):
    """Emit an explicit per-modality target contract; candidate stays null
    (unresolved) since the reference names a service family, not a verified ID (G06)."""
    detected = set(detected_features or [])
    detected.update(requirements.get("critical_features") or [])
    targets = []
    for capability in sorted(detected.intersection(_SEPARATE_MODALITY_TARGETS)):
        service, evidence = _SEPARATE_MODALITY_TARGETS[capability]
        targets.append(
            {
                "capability": capability,
                "status": "unresolved",
                "candidate": None,
                "service": service,
                "evidence": evidence
                + "; select and verify a specific model/service in the target account",
            }
        )
    return targets


def _feature_assessment(workload):
    """Merge detected_features and feature_status; feature_status is authoritative
    for detected/absent/unknown, so an explicit 'unknown' is preserved (G08)."""
    assessment = {}
    for feature in workload.get("detected_features") or []:
        assessment[feature] = "detected"
    for feature, status in (workload.get("feature_status") or {}).items():
        assessment[feature] = status
    return dict(sorted(assessment.items()))


def _unknown_required_features(feature_assessment, requirements):
    """Required features whose status is explicitly unknown (G08)."""
    required = set(requirements.get("critical_features") or [])
    return sorted(f for f in required if feature_assessment.get(f) == "unknown")


def _unsupported_required_capabilities(target_model, detected_features, requirements):
    """Required/detected text-model features NOT evidenced by the target's catalog
    capabilities. Hosted tools/modalities are handled as architecture impacts, not here."""
    if target_model is None:
        return []
    catalog_caps = set(target_model.get("capabilities") or [])
    needed = set(requirements.get("critical_features") or [])
    needed.update(detected_features or [])
    checkable = needed.intersection(
        {
            "tool_or_function_calling",
            "structured_output_json",
            "streaming",
            "image_input_vision",
            "reasoning",
        }
    )
    if target_model.get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)" and "prompt_caching" in needed:
        checkable.add("prompt_caching")
    return sorted(checkable - catalog_caps)


def _compatibility(detected_features, requirements, path, target_model=None):
    """native = only features the SELECTED target's catalog evidences; features
    needed but not evidenced go to unsupported (never derive native from the name)."""
    detected = set(detected_features or [])
    detected.update(requirements.get("critical_features") or [])
    text_features = detected.intersection(
        {"tool_or_function_calling", "structured_output_json", "streaming", "image_input_vision", "reasoning"}
    )
    if (target_model or {}).get("display_name") == "OpenAI GPT-6 Astra (on Bedrock)" and "prompt_caching" in detected:
        text_features.add("prompt_caching")
    catalog_caps = set((target_model or {}).get("capabilities") or [])
    native = sorted(text_features.intersection(catalog_caps)) if target_model else []
    unsupported = sorted(text_features - catalog_caps) if target_model else []
    rearchitecture = sorted(detected.intersection(_HOSTED_TOOL_IMPACTS))
    portable = sorted(detected.intersection(_PORTABLE_FEATURES))
    return {
        "native": native,
        "portable": portable,
        "rearchitecture": rearchitecture,
        "unsupported": unsupported,
    }


def _evaluation(detected_features, requirements):
    detected = set(detected_features or [])
    detected.update(requirements.get("critical_features") or [])
    trajectory = bool(
        detected.intersection(
            {"tool_or_function_calling", "assistants_threads", "web_search", "file_search_retrieval"}
        )
        or "agentic" in (requirements.get("critical_features") or [])
    )
    gates = [
        "Build a deterministic discovery inventory of clients, APIs, model IDs, and features.",
        "Compare viable path/model candidates on a representative golden set.",
        "Size quota for peak traffic and reasoning-output tokens.",
        "Fail on refusal mishandling, truncation, or invalid structured output.",
    ]
    if trajectory:
        gates.extend(
            [
                "Verify correct tool selection, valid tool arguments, and tool-result continuation.",
                "Verify loop termination and structured-output validity.",
            ]
        )
    return {"mode": "trajectory" if trajectory else "prompt", "gates": gates}


def _verification(region, catalog, path, requires_cris, invocation_model_id, selected,
                  model=None, requirements=None):
    if not selected:
        return {
            "region": region,
            "catalog_verified_at": catalog["verified_at"],
            "verified_at": None,
            "probe_status": "not_applicable",
            "availability_claim": "not_selected",
            "invocation_model_id": None,
            "required_checks": [
                "Resolve the model/path decision before running an availability probe."
            ],
        }
    requires_cris = requires_cris and invocation_model_id != (model or {}).get(
        "paths", {}
    ).get(path, {}).get("model_id")
    checks = [
        "Probe the selected model through the selected API path in the target account and region.",
        "Verify path-specific IAM, model access, and quota before code rewrite or POC generation.",
    ]
    if requires_cris:
        checks.insert(
            1, "Resolve and probe a Global or geography-scoped CRIS inference profile."
        )
    return {
        "region": region,
        "catalog_verified_at": catalog["verified_at"],
        "verified_at": None,
        "probe_status": "not_run",
        "availability_claim": "provisional",
        "invocation_model_id": invocation_model_id,
        "required_checks": checks,
        **anthropic_model_recommendation._profile_verification(
            (model or {}).get("paths", {}).get(path, {}).get("model_id"),
            requires_cris, model, region, requirements,
        ),
    }


def _decision_options(catalog, workload, region):
    """Two-sided option set: the selected Mantle API vs runtime Converse governance."""
    options = []
    detected = [feature for feature, status in _feature_assessment(workload).items()
                if status == "detected"]
    astra = _same_model_runtime_key(workload["source"]) == "openai_gpt_6_astra"
    mantle_path = _mantle_path(workload["source"], workload["requirements"])
    if astra:
        preferred = workload["requirements"].get("preferred_api_path")
        if preferred in {"mantle_openai_chat", "mantle_openai_responses"}:
            mantle_path = preferred
        elif workload["source"].get("api_surface") == "chat_completions":
            mantle_path = "mantle_openai_chat"
        # A decision may offer changing an incompatible Chat preference to the
        # documented Responses caching path; it must not silently select it.
        needed = set(detected) | set(workload["requirements"].get("critical_features") or [])
        if "prompt_caching" in needed:
            mantle_path = _mantle_path(workload["source"], workload["requirements"])
    mantle, _ = _catalog_model_for_path(
        catalog, mantle_path, detected, workload["requirements"],
        candidate_order=["openai_gpt_6_astra"] if astra else _mantle_candidate_order(catalog, workload["source"]),
        region=region,
    )
    mantle_requires_region_change = False
    if mantle is None and region != "unknown":
        mantle, _ = _catalog_model_for_path(
            catalog, mantle_path, detected, workload["requirements"],
            candidate_order=_mantle_candidate_order(catalog, workload["source"]),
            region="unknown",
        )
        mantle_requires_region_change = mantle is not None
    if mantle:
        model_key, model, path_config = mantle
        reason = (
            "Preserves the OpenAI SDK and Chat Completions surface; gives up "
            "runtime-only Bedrock governance."
            if mantle_path == "mantle_openai_chat"
            else "Preserves the OpenAI SDK and Responses surface; gives up "
            "runtime-only Bedrock governance."
        )
        if mantle_requires_region_change:
            availability = model.get("region_availability") or {}
            supported_regions = sorted(
                set(availability.get("commercial", []))
                | set(availability.get("govcloud", []))
            )
            region_label = ", ".join(supported_regions) or "a catalog-supported region"
            reason += (
                f" Requires changing from {region} to a catalog-supported Mantle region "
                f"({region_label})."
            )
        options.append(
            {
                "model_key": model_key,
                "model": path_config["model_id"],
                "api_path": mantle_path,
                "invocation_model_id": _resolve_invocation_model_id(
                    path_config["model_id"], path_config["requires_cris"], workload["requirements"],
                    model, region,
                ),
                "requires_cris": path_config["requires_cris"],
                "reason": reason,
            }
        )
    runtime, _ = _catalog_model_for_path(
        catalog, "runtime_converse", detected, workload["requirements"],
        candidate_order=_converse_candidate_order(workload["source"], workload["requirements"]),
        region=region,
    )
    if runtime:
        model_key, model, path_config = runtime
        invocation_model_id = _resolve_invocation_model_id(
            path_config["model_id"], path_config["requires_cris"], workload["requirements"],
            model, region,
        )
        options.append(
            {
                "model_key": model_key,
                "model": path_config["model_id"],
                "api_path": "runtime_converse",
                "invocation_model_id": invocation_model_id,
                "requires_cris": anthropic_model_recommendation._invocation_requires_cris(
                    path_config, invocation_model_id
                ),
                "reason": (
                    "SAME-MODEL governance path: this OpenAI target runs on bedrock-runtime via a "
                    "CRIS id — Guardrails (Converse API only) and invocation logging without a "
                    "model change. Check the model's region/profile matrix and dated prices."
                    if model_key == _same_model_runtime_key(workload["source"])
                    else "Uses Bedrock-native Converse request/response shapes and a "
                    "cross-family model; requires rewriting the OpenAI integration."
                ),
            }
        )
    return options


def _base(workload, decision_status):
    return {
        "workload_id": workload["workload_id"],
        "provider_module": "openai",
        "decision_status": decision_status,
        "source": workload["source"],
    }


def recommend_openai_workload(workload, region, catalog):
    """Return a recommendation dict in the shared downstream contract shape."""
    source = workload["source"]
    requirements = workload["requirements"]
    surface = source.get("api_surface")
    continuity = _api_continuity(requirements)
    runtime_required = _runtime_required(requirements)
    continuity_required = continuity == "required"

    feature_assessment = _feature_assessment(workload)
    # feature_status is authoritative: a feature is "effectively detected" only if
    # the merged assessment marks it detected. Explicit absent/unknown are excluded,
    # so downstream compatibility/deltas/targets never contradict the assessment.
    detected = [f for f, status in feature_assessment.items() if status == "detected"]

    unknown_required = _unknown_required_features(feature_assessment, requirements)

    def _unresolved(
        reason_head, reasons, block, path_for_compat, decision_options=None
    ):
        # Shared decision_required assembly (no target selected).
        rec = _base(workload, "decision_required")
        rec.update(
            {
                "source_analysis": _source_analysis(source, None),
                "feature_assessment": feature_assessment,
                "primary_model": None,
                "model_identity": None,
                "api_path": None,
                "invocation_model_id": None,
                "decision_options": (
                    _decision_options(catalog, workload, region)
                    if decision_options is None
                    else decision_options
                ),
                "alternatives": [],
                "rationale": [reason_head] + reasons,
                "blocks": [block],
                "tuning": [],
                "compatibility": _compatibility(detected, requirements, path_for_compat),
                "architecture_impacts": _feature_findings(detected, requirements, path_for_compat)[3],
                "additional_targets": _additional_targets(detected, requirements),
                "migration_deltas": [],
                "evaluation": _evaluation(detected, requirements),
                "rollout": {
                    "strategy": "decision_required",
                    "gate": "Resolve the open decision before implementation.",
                },
                "verification": _verification(region, catalog, None, False, None, selected=False),
            }
        )
        return rec

    # --- Fail closed: a required feature with unknown status is not ready (G08) ---
    if unknown_required:
        return _unresolved(
            "Required features are unresolved (status unknown): "
            + ", ".join(unknown_required),
            [
                "The scan did not confirm these required features as detected or absent, so "
                "readiness cannot be asserted."
            ],
            _finding(
                "feature_scan_incomplete",
                "[BLOCKS]",
                "Required features are unresolved (status unknown): "
                + ", ".join(unknown_required),
                "Scan the recorded source paths and mark each required feature detected or "
                "absent, then rerun Model Recommend.",
            ),
            "",
        )

    # --- Hard conflict: OpenAI continuity required AND runtime-only governance ---
    if continuity_required and runtime_required:
        options = _decision_options(catalog, workload, region)
        if len(options) < 2:
            raise ValueError(
                f"catalog cannot supply both decision options for workload "
                f"{workload['workload_id']}"
            )
        return _unresolved(
            "Required OpenAI API continuity conflicts with a runtime-only Bedrock "
            "governance requirement.",
            [
                "Mantle preserves the OpenAI SDK and the supported source API surface; runtime "
                "Converse provides Bedrock-native governance and keeps an exact same-model "
                "runtime candidate when the catalog provides one."
            ],
            _finding(
                "model_path_decision_required",
                "[BLOCKS]",
                "OpenAI API continuity and runtime-only governance cannot both be satisfied "
                "by one path.",
                "Choose the Mantle continuity option or the runtime governance option, "
                "update requirements, and rerun Model Recommend.",
            ),
            "",
        )

    # --- Select a path ---
    if runtime_required:
        path = "runtime_converse"
        candidate_order = _converse_candidate_order(source, requirements)
        same = _same_model_runtime_key(source)
        if same:
            rationale_head = (
                "Bedrock governance or multi-model requirements select runtime Converse; "
                f"the source model itself ({catalog['models'][same]['display_name']}) runs "
                "there via a CRIS id, so the same-model candidate leads, with the Claude "
                "tier as the cross-family fallback."
            )
        else:
            tier = _converse_tier_for_source(source)
            rationale_head = (
                "Bedrock governance or multi-model requirements select runtime Converse; "
                f"the source tier maps to {catalog['models'][tier]['display_name']} "
                "(capability-evidence fallback across Claude tiers). Sources without an "
                "exact runtime catalog identity require this family switch."
            )
    else:
        path = _mantle_path(source, requirements)
        candidate_order = _mantle_candidate_order(catalog, source)
        if path == "mantle_openai_chat":
            rationale_head = (
                "GPT-6 Chat Completions API continuity requires an in-region Mantle Chat "
                "Completions model."
            )
        else:
            rationale_head = (
                "OpenAI source lands on Mantle Responses, with the exact source model "
                "evaluated before capability and numeric fallbacks."
            )

    # Astra's new paths must not inherit the older GPT-5.x defaults. Keep this
    # source on Astra and fail closed when the required path lacks evidence.
    if _same_model_runtime_key(source) == "openai_gpt_6_astra":
        candidate_order = ["openai_gpt_6_astra"]
        preferred = requirements.get("preferred_api_path")
        if preferred:
            available = catalog["models"]["openai_gpt_6_astra"]["paths"].get(preferred, {})
            if (available.get("available") is not True
                    or runtime_required and preferred != "runtime_converse"
                    or continuity_required and preferred == "runtime_converse"):
                return _unresolved(
                    "The explicit Astra API preference conflicts with the available paths "
                    "or required API/governance constraints.", [],
                    _finding(
                        "preferred_api_path_conflict", "[BLOCKS]",
                        f"The requested {preferred} path cannot satisfy this workload.",
                        "Choose a supported API or revise the conflicting requirements, "
                        "then rerun Model Recommend.",
                    ), preferred,
                )
            path = preferred
        elif not runtime_required and surface == "chat_completions":
            path = "mantle_openai_chat"
        rationale_head = (
            "Keep GPT-6 Astra on Bedrock using the compatible explicit API preference, "
            "otherwise the source-compatible API or required runtime Converse path; "
            "apply Astra's own region and feature evidence."
        )

    # Evidence-driven selection: pick the first candidate on the path whose catalog
    # capabilities cover the required/detected text features (a newer model without
    # feature-level evidence falls back to an older probed one). G02: fail closed
    # only when NO candidate on the path has the evidence.
    catalog_hit, unsupported = _catalog_model_for_path(
        catalog, path, detected, requirements, candidate_order=candidate_order,
        region=region,
    )
    if not catalog_hit:
        if path.startswith("mantle_"):
            return _unresolved(
                rationale_head,
                [
                    f"The catalog has no {path} candidate available in {region}; "
                    "select another region or migration path."
                ],
                _finding(
                    "mantle_region_unavailable",
                    "[BLOCKS]",
                    f"No in-region {path} model is available in {region}.",
                    "Choose a region in the selected model's Mantle footprint, choose "
                    "runtime Converse, or update the continuity requirement, then rerun "
                    "Model Recommend.",
                ),
                path,
            )
        raise ValueError(f"catalog has no available model for path {path}")
    model_key, model, path_config = catalog_hit
    if path == "mantle_openai_chat":
        if model_key == _same_model_catalog_key(source):
            rationale_head = (
                "GPT-6 Chat Completions source with API continuity lands on the exact "
                "same model through Mantle Chat Completions."
            )
        else:
            rationale_head = (
                f"The exact source model is unavailable in {region}; "
                f"{model['display_name']} is the in-region Mantle Chat Completions fallback."
            )
    if unsupported:
        return _unresolved(
            rationale_head,
            [
                f"No {path} candidate has catalog capability evidence for: "
                + ", ".join(unsupported)
            ],
            _finding(
                "unverified_capability",
                "[BLOCKS]",
                f"No {path} candidate is evidenced to support required features: "
                + ", ".join(unsupported),
                "Add dated capability evidence for a candidate, choose a different path/model, "
                "or drop the requirement, then rerun Model Recommend.",
            ),
            path,
        )

    needed = set(detected) | set(requirements.get("critical_features") or [])
    if (model_key == "openai_gpt_6_astra" and path == "runtime_converse"
            and {"structured_output_json", "streaming"} <= needed):
        return _unresolved(
            rationale_head, [],
            _finding(
                "structured_output_streaming_unverified", "[BLOCKS]",
                "Astra runtime JSON Schema output is documented only for non-streaming calls.",
                "Use non-streaming structured output or separate the streaming workload, "
                "then update requirements and rerun Model Recommend.",
            ), path,
        )

    numeric_conflicts = _numeric_requirement_conflict(model, requirements)
    if numeric_conflicts:
        return _unresolved(
            rationale_head,
            numeric_conflicts,
            _finding(
                "unverified_capacity",
                "[BLOCKS]",
                "; ".join(numeric_conflicts),
                "Source the missing capability data from a dated reference, or reduce the "
                "hard numeric requirement, then rerun Model Recommend.",
            ),
            path,
        )

    invocation_model_id = _resolve_invocation_model_id(
        path_config["model_id"], path_config["requires_cris"], requirements, model, region
    )
    regions = path_config.get("supported_regions")
    profile_regions = path_config.get("inference_profile_regions")
    if regions is not None and region not in regions:
        return _unresolved(
            rationale_head, [],
            _finding(
                "model_region_unavailable", "[BLOCKS]",
                f"{model['display_name']} is not listed for {path} in {region}.",
                "Choose a supported region or a supported CRIS path with explicit "
                "data-residency consent, or evaluate a different model.",
            ), path,
        )
    if profile_regions is not None and (invocation_model_id is not None or _hard_profile_requirement(requirements)):
        profile_key = invocation_model_id
        if invocation_model_id and invocation_model_id.startswith("arn:"):
            match = re.fullmatch(
                r"arn:aws:bedrock:([^:]+):[0-9]{12}:inference-profile/([^/]+)",
                invocation_model_id,
            )
            profile_key = match.group(2) if match and match.group(1) == region else None
        if region not in profile_regions.get(profile_key, []):
            return _unresolved(
                rationale_head, [],
                _finding(
                    "inference_profile_unverified", "[BLOCKS]",
                    f"{invocation_model_id} has no catalog evidence for calls from {region}.",
                    "Select a cataloged profile available from this region, or verify an "
                    "application inference profile in the target account before proceeding.",
                ), path,
            )
    source_analysis = _source_analysis(source, model)

    # --- Findings (target- and path-derived) ---
    r_blocks, r_tuning, r_deltas = _reasoning_findings(source, requirements, model, path, detected)
    f_blocks, f_tuning, f_deltas, impacts = _feature_findings(detected, requirements, path, model)
    blocks = r_blocks + f_blocks
    tuning = r_tuning + f_tuning
    deltas = list(r_deltas)
    if model.get("adaptive_thinking_only"):
        finding = _finding(
            "adaptive_thinking_required",
            "[BLOCKS]" if requirements.get("thinking_enabled") is False else "[TUNE]",
            "Opus 5.5 always uses adaptive thinking and does not support disabled thinking, manual budgets, or forced any/tool choice.",
            "Set output_config.effort as needed (default medium), and size max_tokens "
            "for thinking plus response text. Use auto/none tool choice; a hard forced-tool "
            "requirement needs a verified compatible target. Verify billed usage on the workload.",
        )
        (blocks if finding["tag"] == "[BLOCKS]" else tuning).append(finding)

    if path == "runtime_converse" and path_config["requires_cris"] and invocation_model_id is None:
        # GPT-5.6 on bedrock-runtime is CRIS-only — there is no in-region invocation
        # form. A residency posture that forbids both Global and Geo CRIS makes the
        # same-model governance path unusable; say so explicitly instead of shipping
        # a recommendation with an unresolvable invocation id.
        blocks.append(
            _finding(
                "cris_residency_unresolved",
                "[BLOCKS]",
                "This runtime path is CRIS-only, and the stated data-residency posture "
                "permits neither Global nor Geo cross-region inference, so no invocation "
                "id can be resolved.",
                "Relax residency to Global or a Geo CRIS geography, supply an explicit "
                "inference_profile_id, or take the Claude cross-family Converse path.",
            )
        )

    if path == "mantle_openai_responses" and surface == "chat_completions":
        deltas.extend(_chat_to_responses_deltas())
        blocks.append(
            _finding(
                "chat_completions_to_responses_required",
                "[BLOCKS]",
                (
                    "The selected Astra Responses path requires reshaping the Chat source."
                    if model_key == "openai_gpt_6_astra"
                    else "GPT-5.x on Mantle rejects Chat Completions; the source must reshape "
                    "to the Responses API."
                ),
                "Apply the request/response/tool/state reshape deltas before cutover.",
            )
        )
    if path == "mantle_openai_chat" and surface == "responses":
        deltas.extend([
            _delta(
                "responses_to_chat_request", "path",
                "Map Responses input/instructions to Chat messages/system content and "
                "verify the selected model's output-budget parameters.",
            ),
            _delta(
                "responses_to_chat_response", "path",
                "Read choices[0].message.content instead of output_text; preserve Chat "
                "messages and tool_call_id rather than Responses output items.",
            ),
        ])
        blocks.append(_finding(
            "responses_to_chat_completions_required", "[BLOCKS]",
            "The selected Astra Chat path requires reshaping the Responses source.",
            "Apply the request/response/tool/state mapping before cutover.",
        ))
    deltas.extend(f_deltas)

    if path == "runtime_converse":
        same_model = model_key == _same_model_runtime_key(source)
        deltas.append(
            _delta(
                "openai_sdk_to_converse",
                "path",
                "The OpenAI SDK integration is rewritten to boto3 bedrock-runtime Converse; "
                + (
                    "the selected OpenAI model identity is preserved, while streaming moves "
                    "to ConverseStream event shapes."
                    if same_model
                    else "the selected model also changes families, and streaming moves to "
                    "ConverseStream event shapes."
                ),
            )
        )

    rationale = [rationale_head]
    if model["context_window"] == "unknown" or model["output_token_ceiling"] == "unknown":
        rationale.append(
            f"{model['display_name']} path is evidenced by the reference, but its context/output "
            "limits are unknown in the catalog and must be probed."
        )
    rec = _base(workload, "recommended")
    rec.update(
        {
            "source_analysis": source_analysis,
            "feature_assessment": feature_assessment,
            "primary_model": path_config["model_id"],
            "model_identity": _model_identity(model_key, model, path_config, invocation_model_id),
            "api_path": path,
            "invocation_model_id": invocation_model_id,
            "decision_options": [],
            "alternatives": [],
            "rationale": rationale,
            "blocks": blocks,
            "tuning": tuning,
            "compatibility": _compatibility(detected, requirements, path, model),
            "architecture_impacts": impacts,
            "additional_targets": _additional_targets(detected, requirements),
            "migration_deltas": deltas,
            "evaluation": _evaluation(detected, requirements),
            "rollout": {
                "strategy": "canary",
                "gate": "Compare source and target on the golden set before percentage rollout.",
            },
            "verification": _verification(
                region, catalog, path, path_config["requires_cris"], invocation_model_id,
                selected=True, model=model, requirements=requirements,
            ),
        }
    )
    return rec