request_headers.py

"""Build upstream request headers for GitHub Copilot proxy requests."""

import hashlib
import json
import os
import uuid
from threading import Lock

from fastapi import Request

import codex_agent_compat
from constants import (
    OPENCODE_VERSION, OPENCODE_INTEGRATION_ID,
    COPILOT_CLI_INTEGRATION_ID, COPILOT_CLI_USER_AGENT,
    GITHUB_API_VERSION,
    FORWARDED_REQUEST_HEADERS,
)

_STABLE_ID_NAMESPACE = uuid.UUID("8fd22b32-4ce1-4af7-a3d6-7156a8f0ef9d")
_CODEX_ROOT_TURN_SCOPE_LOCK = Lock()
# Root observations and child pins intentionally live for the process lifetime.
# Evicting either can reattach an idle but still-running worker to the wrong
# root turn and recreate the cache-busting identity rotation these tables
# prevent.
_CODEX_ROOT_TURN_SCOPE_BY_THREAD: dict[tuple[str, str], str] = {}
_CODEX_CHILD_PARENT_TURN_SCOPE_BY_LINEAGE: dict[
    tuple[str, str, str], str
] = {}


def _stable_uuid(value: str) -> str:
    return str(uuid.uuid5(_STABLE_ID_NAMESPACE, value))


_CLIENT_SESSION_ID = os.environ.get("GHCP_COPILOT_CLIENT_SESSION_ID") or str(uuid.uuid4())
_COPILOT_CLI_CLIENT_MACHINE_ID = (
    os.environ.get("GHCP_COPILOT_CLIENT_MACHINE_ID")
    or "0d40238f-cbaa-4b91-a8d0-b46c0a95fdf6"
)
_COPILOT_CLI_EXP_ASSIGNMENT_CONTEXT = (
    os.environ.get("GHCP_COPILOT_CLIENT_EXP_ASSIGNMENT_CONTEXT")
    or (
        "cli_aa_c:1136621;h7c07110:1158940;5bb63a0f:1149772;"
        "e974i579:1132406;h0649438:1154840;be5i9337:1149385;"
        "no-gpt-default:1141892;916bj795:1153107;voting-aa-control:1146525;"
        "ibdi6602:1154657;19ji4148:1155452;voting-aa-treamtment-v2:1157379;"
        "voting-aa-control-v3:1153077;"
    )
)


def _responses_task_affinity_scope(affinity_value) -> str | None:
    """Return the parent task scope used for Responses cache affinity."""
    if not isinstance(affinity_value, str):
        return None
    normalized = affinity_value.strip()
    if not normalized:
        return None
    return normalized


def _responses_client_session_id_for_affinity(affinity_value, model=None) -> str | None:
    """Return the Copilot client-session bucket for a root Responses session.

    A real Copilot CLI process keeps one ``x-client-session-id`` for its root
    conversation, and subagents stay in that parent's bucket.  This proxy can
    multiplex multiple Codex sessions through one Python process, so a single
    process-wide client session makes unrelated roots fight in the same
    upstream cache namespace.  Derive the bucket from the root affinity only;
    callers must not use a subagent's own ``prompt_cache_key`` here.
    """
    del model
    if os.environ.get("GHCP_COPILOT_CLIENT_SESSION_ID"):
        return None
    if not isinstance(affinity_value, str):
        return None
    normalized = affinity_value.strip()
    if not normalized:
        return None
    return _copilot_uuid(f"responses-root-client-session:{normalized}")


def _responses_isolated_subagent_client_session_id(affinity_value) -> str | None:
    if not isinstance(affinity_value, str):
        return None
    normalized = affinity_value.strip()
    if not normalized:
        return None
    return _copilot_uuid(f"responses-isolated-subagent-client-session:{normalized}")


def _responses_subagent_task_id(parent_scope, subagent, affinity_value=None) -> str | None:
    if not isinstance(parent_scope, str) or not parent_scope.strip():
        return None
    if not isinstance(subagent, str) or not subagent.strip():
        return None
    if isinstance(affinity_value, str) and affinity_value.strip():
        child_scope = affinity_value.strip()
    else:
        child_scope = parent_scope.strip()
    return _copilot_uuid(
        _responses_identity_scope(
            "responses-subagent-task",
            parent_scope,
            subagent,
            child_scope,
        )
    )


def _responses_identity_scope(kind: str, *parts: str) -> str:
    """Encode typed identity fields without delimiter ambiguity."""
    return json.dumps(
        [kind.strip(), *(part.strip() for part in parts)],
        ensure_ascii=True,
        separators=(",", ":"),
    )


def _responses_subagent_affinity_scope(subagent: str, affinity_value: str) -> str:
    return _responses_identity_scope(
        "responses-subagent-affinity",
        subagent.lower(),
        affinity_value,
    )


def _codex_root_turn_key(
    thread_id: str | None,
    root_session_id: str | None,
) -> tuple[str, str] | None:
    if not isinstance(thread_id, str) or not thread_id.strip():
        return None
    session_scope = (
        root_session_id.strip()
        if isinstance(root_session_id, str) and root_session_id.strip()
        else ""
    )
    return session_scope, thread_id.strip()


def _remember_codex_root_turn_scope(
    thread_id: str | None,
    turn_id: str | None,
    root_session_id: str | None,
) -> None:
    if not isinstance(turn_id, str) or not turn_id.strip():
        return
    root_key = _codex_root_turn_key(thread_id, root_session_id)
    if root_key is None:
        return
    normalized_turn = turn_id.strip()
    with _CODEX_ROOT_TURN_SCOPE_LOCK:
        _CODEX_ROOT_TURN_SCOPE_BY_THREAD[root_key] = normalized_turn


def _codex_child_lineage_key(
    *,
    parent_thread_id: str | None,
    child_thread_id: str | None,
    subagent: str | None,
    affinity_value: str | None,
    root_session_id: str | None,
) -> tuple[str, str, str] | None:
    """Return a durable key for pinning a Codex child's spawn-time parent.

    The child thread is authoritative when present.  Older request shapes may
    expose only a child prompt-cache affinity or a concrete agent identity; do
    not persist the generic ``codex:subagent`` fallback because it would merge
    otherwise unrelated workers.
    """
    if not isinstance(parent_thread_id, str) or not parent_thread_id.strip():
        return None
    parent_thread = parent_thread_id.strip()
    session_scope = (
        root_session_id.strip()
        if isinstance(root_session_id, str) and root_session_id.strip()
        else ""
    )

    lineage_scope = None
    if isinstance(child_thread_id, str) and child_thread_id.strip():
        normalized_child_thread = child_thread_id.strip()
        if normalized_child_thread != parent_thread:
            lineage_scope = f"thread:{normalized_child_thread}"
    if lineage_scope is None and isinstance(affinity_value, str) and affinity_value.strip():
        normalized_affinity = affinity_value.strip()
        if normalized_affinity != parent_thread:
            lineage_scope = f"affinity:{normalized_affinity}"
    if lineage_scope is None and isinstance(subagent, str) and subagent.strip():
        normalized_subagent = subagent.strip()
        if normalized_subagent.lower() != "codex:subagent":
            lineage_scope = f"subagent:{normalized_subagent}"
    if lineage_scope is None:
        return None
    return session_scope, parent_thread, lineage_scope


def _codex_child_parent_turn_scope(
    lineage_key: tuple[str, str, str] | None,
    parent_thread_id: str | None,
    root_session_id: str | None,
) -> str | None:
    """Pin and return the first-observed parent turn for a child lineage.

    A root can start a later turn while an already-spawned child is still
    running.  Looking up the root's latest turn on every child continuation
    changes both the parent task and the derived child task.  Snapshot the root
    turn on the child's first request and retain it for the process lifetime.
    Codex does not currently send the parent's turn id on the child request, so
    the first observation is the earliest point at which it can be pinned.
    """
    root_key = _codex_root_turn_key(parent_thread_id, root_session_id)
    if root_key is None:
        return None
    normalized_parent_thread = root_key[1]
    with _CODEX_ROOT_TURN_SCOPE_LOCK:
        if lineage_key is not None:
            pinned = _CODEX_CHILD_PARENT_TURN_SCOPE_BY_LINEAGE.get(lineage_key)
            if pinned is not None:
                return pinned

        parent_turn = _CODEX_ROOT_TURN_SCOPE_BY_THREAD.get(root_key)
        if parent_turn is None:
            # If the child arrives before this process observes its root, pin
            # the explicit parent thread rather than changing identity later.
            parent_turn = normalized_parent_thread

        if lineage_key is not None:
            _CODEX_CHILD_PARENT_TURN_SCOPE_BY_LINEAGE[lineage_key] = parent_turn
        return parent_turn


def _responses_request_local_subagent_affinity(request_id: str | None) -> str:
    """Return a safe one-request worker scope when durable affinity is absent.

    A named worker without a prompt-cache key or session must not inherit the
    active root's parent-agent family: that is exactly the shape that lets a
    concurrent subagent invalidate the root's upstream cache.  There is no
    reusable lineage to preserve in this case, so request-local isolation is
    safer than a best-effort parent lookup.
    """
    if isinstance(request_id, str) and request_id.strip():
        return f"request:{request_id.strip()}"
    return f"request:{uuid.uuid4()}"


def _apply_responses_isolated_subagent_context(
    headers: dict,
    subagent: str,
    affinity_value: str,
) -> None:
    """Isolate a legacy worker that provides no explicit parent identity.

    Modern Codex child metadata carries an authoritative parent-thread id and
    must use that parent's hierarchy instead.  This synthetic family remains
    a safe fallback for older workers that have a durable child affinity but
    no way to identify their parent.
    """
    scope = _responses_subagent_affinity_scope(subagent, affinity_value)
    client_session_id = _responses_isolated_subagent_client_session_id(scope)
    if not client_session_id:
        return
    headers["x-client-session-id"] = client_session_id
    headers["x-parent-agent-id"] = _copilot_uuid(
        f"responses-isolated-subagent-parent:{scope}"
    )
    headers["x-interaction-id"] = _copilot_uuid(
        f"responses-isolated-subagent-interaction:{scope}"
    )
    headers["x-agent-task-id"] = _copilot_uuid(
        f"responses-isolated-subagent-task:{scope}"
    )


def _apply_responses_current_subagent_parent(
    headers: dict,
    subagent: str | None,
    affinity_value: str | None = None,
    *,
    parent_affinity_value: str | None = None,
    parent_turn_scope: str | None = None,
    child_turn_scope: str | None = None,
    root_affinity_value: str | None = None,
    request_scope: str | None = None,
) -> None:
    if not isinstance(subagent, str) or not subagent.strip():
        return
    sub = subagent.strip()
    normalized_affinity = (
        affinity_value.strip()
        if isinstance(affinity_value, str) and affinity_value.strip()
        else None
    )
    normalized_parent_affinity = (
        parent_affinity_value.strip()
        if isinstance(parent_affinity_value, str) and parent_affinity_value.strip()
        else None
    )
    normalized_parent_turn = (
        parent_turn_scope.strip()
        if isinstance(parent_turn_scope, str) and parent_turn_scope.strip()
        else None
    )
    normalized_child_turn = (
        child_turn_scope.strip()
        if isinstance(child_turn_scope, str) and child_turn_scope.strip()
        else None
    )
    normalized_root_affinity = (
        root_affinity_value.strip()
        if isinstance(root_affinity_value, str) and root_affinity_value.strip()
        else None
    )

    # Current Codex multi-agent requests explicitly identify their parent
    # thread in client_metadata.  Native Copilot keeps the root client session,
    # parent task, and interaction stable across all of those children while
    # assigning each child its own task/cache lineage.  Reconstruct that same
    # hierarchy from the inbound metadata before it is stripped from the body.
    if normalized_affinity and normalized_parent_affinity:
        parent_scope = (
            _responses_task_affinity_scope(normalized_parent_turn)
            or _responses_task_affinity_scope(normalized_parent_affinity)
            or normalized_parent_affinity
        )
        parent_client_session_id = _responses_client_session_id_for_affinity(
            normalized_root_affinity or normalized_parent_affinity
        )
        if parent_client_session_id:
            headers["x-client-session-id"] = parent_client_session_id
        parent_task_id = _copilot_uuid(f"responses-task:{parent_scope}")
        headers["x-parent-agent-id"] = parent_task_id
        child_scope = normalized_child_turn or _responses_subagent_affinity_scope(
            sub,
            normalized_affinity,
        )
        # Native Copilot shares the root client session and parent task with
        # all children, but every child owns a distinct interaction.  Sharing
        # the parent's interaction here makes concurrent workers overwrite the
        # same upstream cache generation.
        headers["x-interaction-id"] = _copilot_uuid(
            f"responses-interaction:{child_scope}"
        )
        child_task_id = _responses_subagent_task_id(
            parent_scope,
            sub,
            child_scope,
        )
        if child_task_id:
            headers["x-agent-task-id"] = child_task_id
        return

    worker_scope = normalized_affinity or _responses_request_local_subagent_affinity(
        request_scope
    )
    _apply_responses_isolated_subagent_context(headers, sub, worker_scope)


_COPILOT_MACHINE_ID = hashlib.sha256(f"{uuid.getnode():012x}".encode("utf-8")).hexdigest()
_FORWARD_SESSION_HEADER_DEFAULT = True
_VISION_INPUT_INITIAL_DEPTH = 0
_VISION_INPUT_MAX_DEPTH = 10


def has_vision_input(
    value,
    depth: int = _VISION_INPUT_INITIAL_DEPTH,
    max_depth: int = _VISION_INPUT_MAX_DEPTH,
) -> bool:
    """Recursively find type='input_image' anywhere in the input tree."""
    if depth > max_depth or value is None:
        return False
    if isinstance(value, list):
        return any(has_vision_input(i, depth + 1, max_depth) for i in value)
    if not isinstance(value, dict):
        return False
    value_type = value.get("type")
    if isinstance(value_type, str) and value_type.lower() == "input_image":
        return True
    content = value.get("content")
    if isinstance(content, list):
        return any(has_vision_input(i, depth + 1, max_depth) for i in content)
    return False


def _interaction_type_for_initiator(initiator: str) -> str:
    if initiator == "user":
        return "conversation-user"
    return "conversation-agent"


def _interaction_id_for_session(session_id: str | None) -> str:
    if isinstance(session_id, str):
        normalized = session_id.strip()
        if normalized:
            return normalized
    return str(uuid.uuid4())


def _copilot_uuid(content: str) -> str:
    uuid_bytes = bytearray(hashlib.sha256(content.encode("utf-8")).digest()[:16])
    uuid_bytes[6] = (uuid_bytes[6] & 0x0F) | 0x40
    uuid_bytes[8] = (uuid_bytes[8] & 0x3F) | 0x80
    return str(uuid.UUID(bytes=bytes(uuid_bytes)))


def _json_stringify_like(value) -> str:
    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))


def _strip_cache_control(value):
    if not isinstance(value, dict):
        return value
    return {k: v for k, v in value.items() if k != "cache_control"}


def _find_last_user_content(messages) -> str | None:
    if not isinstance(messages, list):
        return None
    for message in reversed(messages):
        if not isinstance(message, dict):
            continue
        if message.get("role") != "user":
            continue
        content = message.get("content")
        if isinstance(content, str) and content:
            return content
        if isinstance(content, list):
            content_items = [
                _strip_cache_control(item)
                for item in content
                if not (isinstance(item, dict) and item.get("type") == "tool_result")
            ]
            if content_items:
                return _json_stringify_like(content_items)
    return None


def _responses_request_id_payload_messages(payload):
    if not isinstance(payload, dict):
        return None
    if "messages" in payload:
        return payload.get("messages")
    return payload.get("input")


def _responses_input_text(value) -> str | None:
    if isinstance(value, str):
        return value
    if not isinstance(value, list):
        return None
    for item in value:
        if not isinstance(item, dict):
            continue
        content = item.get("content")
        if isinstance(content, str):
            return content
        if isinstance(content, list):
            parts: list[str] = []
            for entry in content:
                if not isinstance(entry, dict):
                    continue
                for key in ("text", "input_text", "output_text"):
                    text = entry.get(key)
                    if isinstance(text, str):
                        parts.append(text)
                        break
            if parts:
                return "".join(parts)
        for key in ("text", "input_text", "output_text"):
            text = item.get(key)
            if isinstance(text, str):
                return text
    return None


def _codex_rollout_memory_affinity_value(payload, base_affinity: str) -> str | None:
    """Return an isolated affinity for Codex rollout-memory writer requests.

    Codex's background memory writer sends prompts like "Analyze this rollout..."
    using the active conversation's prompt_cache_key even though each rollout is
    unrelated to the interactive turn. If those requests reuse the interactive
    affinity, a post-interrupt rollout summary with <turn_aborted> content can
    look like a cache bust and can also churn the upstream prompt-cache bucket.
    Keep them stable for the same rollout, but isolate them from the main turn
    and from other rollout summaries.
    """
    if not isinstance(payload, dict) or not isinstance(base_affinity, str):
        return None
    text = _responses_input_text(payload.get("input"))
    if not isinstance(text, str):
        return None
    if not text.startswith("Analyze this rollout and produce JSON"):
        return None
    if "rollout_context:" not in text or "rendered conversation" not in text:
        return None
    digest = hashlib.sha256(f"{base_affinity}\n{text}".encode("utf-8")).hexdigest()[:32]
    return f"codex-rollout-memory:{digest}"


def _generate_request_id_from_payload(payload, session_id: str | None = None) -> str:
    messages = _responses_request_id_payload_messages(payload)
    if isinstance(messages, str) and messages:
        last_user_content = messages
    else:
        last_user_content = _find_last_user_content(messages)

    if last_user_content:
        return _copilot_uuid(f"{session_id or ''}{_COPILOT_MACHINE_ID}{last_user_content}")

    return str(uuid.uuid4())


def _responses_affinity_value(payload, session_id: str | None = None) -> str | None:
    if isinstance(payload, dict):
        for key in ("prompt_cache_key", "promptCacheKey", "session_id", "sessionId"):
            value = payload.get(key)
            if isinstance(value, str):
                normalized = value.strip()
                if normalized:
                    isolated = _codex_rollout_memory_affinity_value(payload, normalized)
                    if isolated:
                        return isolated
                    return normalized
        metadata = payload.get("metadata")
        if isinstance(metadata, dict):
            for key in ("session_id", "sessionId"):
                value = metadata.get(key)
                if isinstance(value, str):
                    normalized = value.strip()
                    if normalized:
                        isolated = _codex_rollout_memory_affinity_value(payload, normalized)
                        if isolated:
                            return isolated
                        return normalized
        for key in ("previous_response_id", "previousResponseId"):
            value = payload.get(key)
            if isinstance(value, str):
                normalized = value.strip()
                if normalized:
                    # A bare previous-response chain has no durable session or
                    # prompt-cache key. Keep it out of the fallback task ID
                    # namespace, where identical user text can otherwise join
                    # unrelated conversations.
                    return f"previous_response:{normalized}"
    if isinstance(session_id, str):
        normalized = session_id.strip()
        if normalized:
            return normalized
    return None


def responses_affinity_value(payload, session_id: str | None = None) -> str | None:
    """Expose the effective Responses affinity used for upstream identity."""
    return _responses_affinity_value(payload, session_id)


def responses_replay_affinity_value(
    payload,
    session_id: str | None = None,
    subagent: str | None = None,
) -> str | None:
    """Return the same scope used by upstream headers for replay-ID state."""
    affinity_value = _responses_affinity_value(payload, session_id)
    if not affinity_value:
        return None
    if isinstance(subagent, str) and subagent.strip():
        return _responses_subagent_affinity_scope(subagent, affinity_value)
    return affinity_value


def _responses_body_has_affinity_hint(payload) -> bool:
    if not isinstance(payload, dict):
        return False
    for key in (
        "prompt_cache_key",
        "promptCacheKey",
        "session_id",
        "sessionId",
        "previous_response_id",
        "previousResponseId",
    ):
        value = payload.get(key)
        if isinstance(value, str) and value.strip():
            return True
    metadata = payload.get("metadata")
    if isinstance(metadata, dict):
        for key in ("session_id", "sessionId"):
            value = metadata.get(key)
            if isinstance(value, str) and value.strip():
                return True
    return False


def _responses_copilot_identity_headers(
    payload,
    session_id: str | None = None,
    request_id: str | None = None,
    *,
    stable_affinity: bool = False,
    subagent: str | None = None,
    client_session_id: str | None = None,
    task_affinity_value: str | None = None,
) -> dict[str, str]:
    del request_id
    if stable_affinity:
        affinity_value = _responses_affinity_value(payload, session_id)
        if affinity_value:
            parent_scope = (
                _responses_task_affinity_scope(task_affinity_value)
                or _responses_task_affinity_scope(affinity_value)
                or affinity_value
            )
            interaction_id = _copilot_uuid(f"responses-interaction:{parent_scope}")
            parent_task_id = _copilot_uuid(f"responses-task:{parent_scope}")
            headers = {
                "x-agent-task-id": parent_task_id,
                "x-interaction-id": interaction_id,
            }
            normalized_subagent = subagent.strip() if isinstance(subagent, str) and subagent.strip() else None
            if normalized_subagent:
                child_task_id = _responses_subagent_task_id(
                    parent_scope,
                    normalized_subagent,
                    affinity_value,
                )
                if child_task_id:
                    headers["x-agent-task-id"] = child_task_id
                    headers["x-parent-agent-id"] = parent_task_id
            return headers

    normalized_session_id = session_id.strip() if isinstance(session_id, str) and session_id.strip() else None
    normalized_client_session_id = (
        client_session_id.strip()
        if isinstance(client_session_id, str) and client_session_id.strip()
        else None
    )
    interaction_session_id = normalized_session_id or normalized_client_session_id
    is_anthropic_messages_payload = isinstance(payload, dict) and "messages" in payload
    root_session_id = None
    if is_anthropic_messages_payload and interaction_session_id:
        root_session_id = _copilot_uuid(interaction_session_id)

    agent_task_id = _generate_request_id_from_payload(
        payload,
        session_id=root_session_id if is_anthropic_messages_payload else None,
    )
    headers = {
        "x-agent-task-id": agent_task_id,
    }
    if is_anthropic_messages_payload:
        if root_session_id:
            headers["x-interaction-id"] = root_session_id
    else:
        if interaction_session_id:
            headers["x-interaction-id"] = _copilot_uuid(
                f"responses-interaction:{interaction_session_id}"
            )
        else:
            headers["x-interaction-id"] = _copilot_uuid(agent_task_id)
    return headers


def _messages_affinity_headers(initiator: str, interaction_id: str | None, request_id: str) -> dict[str, str]:
    """Return native Messages affinity headers for a request.

    Copilot's native Anthropic /v1/messages endpoint keys prompt-cache lineage
    off the body cache breakpoints and request affinity headers. Keep the
    conversation interaction stable when Claude Code supplies a real session
    id, but leave the agent task request-scoped. Making both values stable
    over-affinitizes separate internal requests and can make cache reads look
    like one giant task-wide prefix.
    """
    del initiator

    if isinstance(interaction_id, str):
        normalized = interaction_id.strip()
    else:
        normalized = ""

    if normalized:
        return {"x-interaction-id": normalized}

    return {"x-interaction-id": request_id}


def build_copilot_headers(api_key: str) -> dict:
    return {
        "Authorization": f"Bearer {api_key}",
        "content-type": "application/json",
        "accept": "application/json",
        "User-Agent": f"opencode/{OPENCODE_VERSION}",
        "Openai-Intent": "conversation-agent",
        "Copilot-Integration-Id": OPENCODE_INTEGRATION_ID,
        "x-github-api-version": GITHUB_API_VERSION,
        "x-client-session-id": _CLIENT_SESSION_ID,
    }


def build_responses_copilot_headers(api_key: str) -> dict:
    headers = build_copilot_headers(api_key)
    headers["User-Agent"] = COPILOT_CLI_USER_AGENT
    headers["Copilot-Integration-Id"] = COPILOT_CLI_INTEGRATION_ID
    headers.update(
        {
            "x-stainless-retry-count": "0",
            "x-stainless-lang": "js",
            "x-stainless-package-version": "5.20.1",
            "x-stainless-os": "Windows",
            "x-stainless-arch": "x64",
            "x-stainless-runtime": "node",
            "x-stainless-runtime-version": "v25.6.0",
            "x-client-machine-id": _COPILOT_CLI_CLIENT_MACHINE_ID,
            "x-copilot-client-exp-assignment-context": _COPILOT_CLI_EXP_ASSIGNMENT_CONTEXT,
            "accept-language": "*",
            "sec-fetch-mode": "cors",
        }
    )
    return headers


def _apply_forwarded_request_headers(
    headers: dict,
    request: Request,
    request_body: dict | None = None,
    *,
    session_id_resolver=None,
    forward_session_header: bool = _FORWARD_SESSION_HEADER_DEFAULT,
    synthesize_client_request_id: bool = True,
):
    session_id = session_id_resolver(request, request_body) if session_id_resolver else None
    if session_id and forward_session_header:
        headers["session_id"] = session_id

    for header_name in FORWARDED_REQUEST_HEADERS:
        header_value = request.headers.get(header_name)
        if header_value:
            headers[header_name] = header_value

    if synthesize_client_request_id and "x-client-request-id" not in headers and isinstance(session_id, str):
        normalized_session_id = session_id.strip()
        if normalized_session_id:
            headers["x-client-request-id"] = normalized_session_id

    return session_id


def build_responses_headers_for_request(
    request: Request,
    body: dict,
    api_key: str,
    force_initiator: str | None = None,
    request_id: str | None = None,
    *,
    initiator_policy=None,
    session_id_resolver=None,
    verdict_sink: dict | None = None,
    affinity_body: dict | None = None,
    stable_user_affinity: bool = False,
    synthetic_subagent: str | None = None,
) -> dict:
    headers = build_responses_copilot_headers(api_key)
    identity_source = affinity_body if isinstance(affinity_body, dict) else body
    session_id = _apply_forwarded_request_headers(
        headers,
        request,
        identity_source,
        session_id_resolver=session_id_resolver,
        forward_session_header=False,
        synthesize_client_request_id=False,
    )
    headers.pop("x-client-request-id", None)
    headers.pop("x-request-id", None)
    headers.pop("x-github-request-id", None)
    inbound_subagent = request.headers.get("x-openai-subagent") if hasattr(request, "headers") else None
    inbound_subagent = (
        inbound_subagent.strip()
        if isinstance(inbound_subagent, str) and inbound_subagent.strip()
        else None
    )
    effective_subagent = inbound_subagent
    if effective_subagent is None:
        effective_subagent = codex_agent_compat.codex_subagent_identity(identity_source)
    if (
        effective_subagent is None
        and isinstance(synthetic_subagent, str)
        and synthetic_subagent.strip()
    ):
        effective_subagent = synthetic_subagent.strip()
    headers.pop("x-openai-subagent", None)
    codex_turn_id = codex_agent_compat.codex_turn_id(identity_source)
    codex_thread_id = codex_agent_compat.codex_thread_id(identity_source)
    codex_session_id = codex_agent_compat.codex_session_id(identity_source)
    if not effective_subagent:
        _remember_codex_root_turn_scope(
            codex_thread_id,
            codex_turn_id,
            codex_session_id,
        )

    had_input = "input" in body
    effective_input, initiator = initiator_policy.resolve_responses_input(
        body.get("input"),
        body.get("model"),
        subagent=effective_subagent,
        trusted_user_turn=(
            codex_turn_id is not None
            and codex_agent_compat.codex_thread_source(identity_source) == "user"
        ),
        force_initiator=force_initiator,
        request_id=request_id,
        verdict_sink=verdict_sink,
    )
    if had_input:
        body["input"] = effective_input
    if effective_subagent:
        initiator = "agent"
    headers["x-initiator"] = initiator
    headers["x-interaction-type"] = (
        "conversation-subagent"
        if effective_subagent
        else _interaction_type_for_initiator(initiator)
    )
    affinity_value = _responses_affinity_value(identity_source, session_id)
    stable_affinity = (
        stable_user_affinity
        or _responses_body_has_affinity_hint(identity_source)
        or affinity_value is not None
    )
    affinity_client_session_id = _responses_client_session_id_for_affinity(
        codex_session_id or affinity_value, model=body.get("model")
    )
    if affinity_client_session_id and not effective_subagent:
        headers["x-client-session-id"] = affinity_client_session_id
    headers.update(
        _responses_copilot_identity_headers(
            identity_source,
            session_id,
            request_id=request_id,
            stable_affinity=stable_affinity,
            subagent=effective_subagent,
            client_session_id=headers.get("x-client-session-id"),
            task_affinity_value=codex_turn_id,
        )
    )
    if effective_subagent:
        parent_affinity_value = codex_agent_compat.codex_parent_affinity(identity_source)
        child_lineage_key = _codex_child_lineage_key(
            parent_thread_id=parent_affinity_value,
            child_thread_id=codex_thread_id,
            subagent=effective_subagent,
            affinity_value=affinity_value,
            root_session_id=codex_session_id,
        )
        parent_turn_scope = _codex_child_parent_turn_scope(
            child_lineage_key,
            parent_affinity_value,
            codex_session_id,
        )
        _apply_responses_current_subagent_parent(
            headers,
            effective_subagent,
            affinity_value,
            parent_affinity_value=parent_affinity_value,
            parent_turn_scope=parent_turn_scope,
            child_turn_scope=codex_turn_id,
            root_affinity_value=codex_session_id,
            request_scope=request_id,
        )

    if has_vision_input(effective_input):
        headers["Copilot-Vision-Request"] = "true"

    return headers


def build_chat_headers_for_request(
    request: Request,
    messages,
    model_name: str,
    api_key: str,
    request_id: str | None = None,
    *,
    initiator_policy=None,
    session_id_resolver=None,
    verdict_sink: dict | None = None,
    affinity_body: dict | None = None,
    synthetic_subagent: str | None = None,
) -> dict:
    headers = build_copilot_headers(api_key)
    identity_source = affinity_body if isinstance(affinity_body, dict) else None
    session_id = _apply_forwarded_request_headers(
        headers,
        request,
        identity_source,
        session_id_resolver=session_id_resolver,
    )
    affinity_value = _responses_affinity_value(identity_source, session_id)
    inbound_subagent = (
        request.headers.get("x-openai-subagent")
        if hasattr(request, "headers")
        else None
    )
    effective_subagent = (
        inbound_subagent.strip()
        if isinstance(inbound_subagent, str) and inbound_subagent.strip()
        else None
    )
    if effective_subagent is None:
        effective_subagent = codex_agent_compat.codex_subagent_identity(identity_source)
    if (
        effective_subagent is None
        and isinstance(synthetic_subagent, str)
        and synthetic_subagent.strip()
    ):
        effective_subagent = synthetic_subagent.strip()
    headers.pop("x-openai-subagent", None)
    affinity_client_session_id = _responses_client_session_id_for_affinity(
        affinity_value, model=model_name
    )
    if affinity_client_session_id and not (
        isinstance(effective_subagent, str) and effective_subagent.strip()
    ):
        headers["x-client-session-id"] = affinity_client_session_id

    initiator = initiator_policy.resolve_chat_messages(
        messages,
        model_name,
        subagent=effective_subagent,
        request_id=request_id,
        verdict_sink=verdict_sink,
    )
    headers["x-initiator"] = initiator
    headers["x-interaction-type"] = (
        "conversation-subagent"
        if effective_subagent
        else _interaction_type_for_initiator(initiator)
    )
    headers["x-interaction-id"] = _interaction_id_for_session(session_id)
    headers["x-agent-task-id"] = str(uuid.uuid4())
    if affinity_value:
        headers["x-interaction-id"] = _copilot_uuid(f"chat-interaction:{affinity_value}")
        headers["x-agent-task-id"] = _copilot_uuid(f"chat-task:{affinity_value}")

    if effective_subagent:
        headers["x-initiator"] = "agent"
        _apply_responses_current_subagent_parent(
            headers,
            effective_subagent,
            affinity_value,
            request_scope=request_id,
        )

    if isinstance(messages, list):
        for msg in messages:
            if not isinstance(msg, dict):
                continue
            content = msg.get("content")
            if isinstance(content, list):
                for item in content:
                    if isinstance(item, dict) and ("image_url" in item or item.get("type") == "image_url"):
                        headers["Copilot-Vision-Request"] = "true"
                        break
                if headers.get("Copilot-Vision-Request") == "true":
                    break

    return headers


def _anthropic_messages_has_vision(messages) -> bool:
    if not isinstance(messages, list):
        return False
    for item in messages:
        if not isinstance(item, dict):
            continue
        content = item.get("content")
        if not isinstance(content, list):
            continue
        for part in content:
            if not isinstance(part, dict):
                continue
            part_type = str(part.get("type", "")).lower()
            if part_type == "image":
                return True
            if part_type == "tool_result" and isinstance(part.get("content"), list):
                for nested in part["content"]:
                    if isinstance(nested, dict) and str(nested.get("type", "")).lower() == "image":
                        return True
    return False


def build_anthropic_headers_for_request(
    request: Request,
    body: dict,
    api_key: str,
    request_id: str | None = None,
    *,
    initiator_policy=None,
    session_id_resolver=None,
    verdict_sink: dict | None = None,
) -> dict:
    headers = build_copilot_headers(api_key)
    session_id = _apply_forwarded_request_headers(headers, request, body, session_id_resolver=session_id_resolver)

    messages = body.get("messages")
    initiator = initiator_policy.resolve_anthropic_messages(
        messages,
        body.get("model"),
        system=body.get("system"),
        subagent=request.headers.get("x-openai-subagent"),
        request_id=request_id,
        verdict_sink=verdict_sink,
    )
    headers["x-initiator"] = initiator
    headers["x-interaction-type"] = _interaction_type_for_initiator(initiator)
    headers["x-interaction-id"] = _interaction_id_for_session(session_id)
    headers["x-agent-task-id"] = str(uuid.uuid4())

    if _anthropic_messages_has_vision(messages):
        headers["Copilot-Vision-Request"] = "true"

    return headers


# ---------------------------------------------------------------------------
# Anthropic /v1/messages native passthrough helpers
# ---------------------------------------------------------------------------

ALLOWED_ANTHROPIC_BETAS = frozenset({
    "interleaved-thinking-2025-05-14",
    "context-management-2025-06-27",
    "advanced-tool-use-2025-11-20",
})

ADVANCED_TOOL_USE_MODELS_PREFIXES = (
    "claude-sonnet-4.5",
    "claude-sonnet-4.6",
    "claude-opus-4.5",
    "claude-opus-4.6",
)

CLAUDE_AGENT_USER_AGENT = (
    "vscode_claude_code/2.1.98 (external, sdk-ts, agent-sdk/0.2.98)"
)


def _normalize_model_for_betas(model: str | None) -> str:
    norm = (model or "").strip().lower()
    if norm.startswith("anthropic/"):
        norm = norm.split("/", 1)[1]
    return norm


def derive_anthropic_betas(
    *,
    client_betas: list[str] | None,
    body: dict,
    model: str,
) -> list[str]:
    """Filter inbound ``anthropic-beta`` values against the allowlist and
    auto-inject the betas this proxy knows how to use."""

    seen: set[str] = set()
    out: list[str] = []

    def _add(name: str) -> None:
        if name in ALLOWED_ANTHROPIC_BETAS and name not in seen:
            seen.add(name)
            out.append(name)

    if isinstance(client_betas, list):
        for entry in client_betas:
            if not isinstance(entry, str):
                continue
            for piece in entry.split(","):
                token = piece.strip()
                if token:
                    _add(token)

    # Auto-inject interleaved-thinking when explicit budget_tokens are used
    # (and the request is not already adaptive).
    thinking = body.get("thinking") if isinstance(body, dict) else None
    if isinstance(thinking, dict):
        t_type = thinking.get("type")
        budget = thinking.get("budget_tokens")
        if t_type == "enabled" and isinstance(budget, int) and budget > 0:
            _add("interleaved-thinking-2025-05-14")

    norm_model = _normalize_model_for_betas(model)
    if any(norm_model.startswith(p) for p in ADVANCED_TOOL_USE_MODELS_PREFIXES):
        _add("advanced-tool-use-2025-11-20")

    return out


def build_anthropic_messages_passthrough_headers(
    *,
    request_id: str,
    initiator: str,
    interaction_id: str | None,
    interaction_type: str | None,  # noqa: ARG001 - accepted for API symmetry
    anthropic_betas: list[str],
    base_headers: dict,
) -> dict:
    """Produce the upstream header set for a native Copilot Messages proxy
    request. Mirrors copilot-api's ``prepareMessageProxyHeaders``."""

    headers: dict = dict(base_headers) if isinstance(base_headers, dict) else {}

    # Drop any header (regardless of casing) that this function is about to
    # set itself. httpx merges duplicate-cased keys into a comma-joined value
    # which corrupts user-agent / openai-intent / interaction headers and
    # triggers Copilot validation errors.
    _drop_keys = (
        "copilot-integration-id",
        "user-agent",
        "openai-intent",
        "x-interaction-type",
        "x-interaction-id",
        "x-agent-task-id",
        "x-request-id",
        "x-initiator",
        "anthropic-version",
        "anthropic-beta",
    )
    for key in [k for k in list(headers.keys()) if k.lower() in _drop_keys]:
        del headers[key]

    headers["x-agent-task-id"] = request_id
    headers["x-request-id"] = request_id
    headers["x-interaction-type"] = "messages-proxy"
    headers["openai-intent"] = "messages-proxy"
    headers["user-agent"] = CLAUDE_AGENT_USER_AGENT
    headers["anthropic-version"] = "2023-06-01"

    if isinstance(anthropic_betas, list) and anthropic_betas:
        headers["anthropic-beta"] = ",".join(anthropic_betas)

    if isinstance(initiator, str) and initiator:
        headers["x-initiator"] = initiator

    headers.update(_messages_affinity_headers(initiator, interaction_id, request_id))

    return headers