# Muse Responses API: protocol documentation and transpiler

Status snapshot: 2026-10-11. Research complete and archived under
`research/muse-responses-openai-transpiler/`. Transpiler module complete,
stdlib-only, 21/21 unit tests green; devplacepy gateway integration in
progress (module tests green, gateway wiring under test).

## 1. What this document is

One place for everything learned about the API that Meta's `muse` CLI
speaks, plus the complete bidirectional transpiler between that wire
format and classic OpenAI Chat Completions. Part 1 is the research
(what the protocol is, with evidence). Part 2 is the transpiler
(contract, usage, full source). Nothing here is theorized: request-side
claims come from a captured MITM log of the real CLI, schema claims
come from the official OpenAI docs fetched the same day, and every
mapping in the code is covered by a test.

## 2. The wire protocol

### 2.1 Endpoints

The CLI talks to `https://api.meta.ai/v1` (from its own
`~/.config/muse/auth.json`, `providers.meta.api_base_url`, overridable
per-run with `--base-url`):

| Method | Path | Purpose |
|---|---|---|
| `POST` | `/responses` | Model turn (OpenAI Responses API shape) |
| `GET` | `/muse-code/models` | Model listing (Meta-specific path) |

The capture (`/tmp/mitm_probe/request.txt`, 795KB, 11 requests) shows
one models GET plus ten `/responses` POSTs in two alternating shapes:
the main turn (full 19-function tool namespace, `reasoning.effort`
`max`) and a reminder turn (single `submit_reminder_decision` tool,
`effort` `low`).

### 2.2 Headers

Every request carries `Accept: text/event-stream` (streaming SSE),
`Content-Type: application/json`, `Authorization: Bearer ...`
(redacted in the capture, never logged), a client UA
(`muse-build/...`), `x-client-id: tbh:exec`, `x-tbh-session-id`, and
`x-meta-ai-gateway-session-id`. The `x-tbh-*` / `x-meta-ai-gateway-*`
trio is Meta-specific; the rest is stock OpenAI practice.

### 2.3 Request schema

All ten POSTs share these top-level keys: `model`, `instructions`,
`input`, `tools`, `stream`, `store`, `max_output_tokens`, `reasoning`,
`include`, `prompt_cache_key`. Observed values: `stream: true`,
`store: false`, `max_output_tokens: 128000`,
`reasoning: {effort, summary: "auto"}`,
`include: ["reasoning.encrypted_content"]`,
`prompt_cache_key: "tbh:skill-reminder:..."`, model
`muse-spark-1.3-contributor`. The client is stateless (`store: false`)
and replays encrypted reasoning items instead of server-side
conversation state.

Illustrative Meta-flavored request (trimmed, synthetic content):

```json
{
  "model": "muse-spark-1.3-contributor",
  "instructions": "<25-44KB system string>",
  "input": [
    {"type": "message", "role": "developer", "content": "<31KB reminder>"},
    {"type": "message", "role": "user", "content": "hi"}
  ],
  "tools": [{"type": "namespace", "name": "muse", "tools": ["...19 builtins..."]}],
  "stream": true,
  "store": false,
  "max_output_tokens": 128000,
  "reasoning": {"effort": "max", "summary": "auto"},
  "include": ["reasoning.encrypted_content"],
  "prompt_cache_key": "tbh:skill-reminder:..."
}
```

### 2.4 The tools namespace

`tools` holds exactly one entry: `{type: "namespace", name: "muse"}`
with 19 nested harness functions (`workflow`, `read_file`, `search`,
`write_file`, `edit_file`, `bash`, `web_fetch`, `web_search`,
`monitor`, `cron_*`, `subagent_*`, `read_skill`, `add_memory`,
`edit_memory`, `read_memory`, `request_user_input`, `snooze_reminder`,
`update_goal`, ...). The reminder shape swaps this for a single
`submit_reminder_decision` function. `namespace` is not in the public
reference: it is a Meta vendor extension, executed natively by the
harness in its own sandbox. tai-style tool loops do not use it at all;
they constrain the answer to `{content, tool_calls}` JSON via
`--output-schema` and execute calls themselves.

### 2.5 Meta extensions vs stock OpenAI

Only four things are Meta-specific: the `namespace` tool type, the
`x-tbh-*` / `x-meta-ai-gateway-*` headers, the `/muse-code/models`
path, and the `prompt_cache_key` values. Everything else matches the
public Responses reference field for field, which is why a generic
transpiler works.

### 2.6 Response object and SSE (unobserved, per reference)

The probe stub answered HTTP 500 to all 11 requests, so no Meta
response bytes were ever seen. The transpiler therefore implements the
documented standard shapes, which the Meta client demonstrably shares
on the request side: a Response object with `output` items of type
`message` (`output_text` / `refusal` parts) and `function_call`
(`call_id`, `name`, string `arguments`), plus `usage
{input_tokens, output_tokens, total_tokens}`; streamed as typed SSE
(`response.created`, `response.output_text.delta`, `response.output_item.added`,
`response.function_call_arguments.delta`, `response.completed`,
`response.failed` / `error`). A recording forward proxy would close
this gap.

### 2.7 CLI translation

The CLI converts SSE into newline-delimited JSON on stdout
(`run.output.delta` text events, `run.terminal.*` terminal events),
which is what `muse_run` parses. Timeouts, retries, and the
`--no-session-log` / `--session-id` session modes all live above this
layer and are unaffected by the protocol shape.

### 2.8 System messages

System guidance works through three channels, and Meta's client uses
two of them at once: top-level `instructions` (preferred, 25KB+
observed), `developer`-role input messages (31KB observed), and
`system`-role input messages (spec-allowed, equivalent for older
clients). The transpiler folds leading Chat system/developer messages
into `instructions`, keeps mid-transcript ones as role-kept Items, and
reverses `instructions` into a leading Chat `system` message.

## 3. Transpiler contract

Pure functions returning `(converted, warnings)`. Core conversions are
exact both ways; everything outside the core warns on drop. Unknown
top-level fields warn too, so a gateway dialect field can never vanish
silently.

### 3.1 Exact mappings

| Chat Completions | Responses |
|---|---|
| `messages[]` | `input[]` Items |
| system/developer (leading) | `instructions` |
| assistant `tool_calls` | `function_call` (`id` to `call_id`) |
| `tool` role | `function_call_output` |
| externally-tagged function tools | internally-tagged function tools |
| `response_format` | `text.format` |
| `max_tokens` | `max_output_tokens` |
| `reasoning_effort` | `reasoning.effort` |
| delta chunks | typed SSE events |

The `strict` default flips explicitly: Chat omits it (non-strict),
Responses omits it (attempts strict), so Chat-to-Responses pins
`strict: false`.

### 3.2 Fixed loss table

Chat to Responses drops: `n` (single generation only), `stop`,
`seed`, `logprobs`, sampling penalties, `logit_bias`, `prediction`,
`modalities`, audio, `web_search_options`, audio/file content parts,
custom tools, message `name`/`audio`. Responses to Chat drops:
`namespace`/built-in tools, `reasoning` and other native Items,
`previous_response_id` / `conversation`, `truncation`, `include`,
extra choices, non-text call-output parts.

Thinking dialects collapse into `reasoning`: explicit
`reasoning_effort` / `reasoning.effort` wins; `thinking: {type}`,
`think`, `enable_thinking` map disabled to `effort: "none"` and leave
enabled as upstream default.

### 3.3 Streaming and errors

`ResponsesToChatStream` maps text and function-argument deltas
exactly, emits `role: assistant` once, yields the terminal chunk with
the real finish reason and usage from `response.completed`, and turns
`response.failed` / `error` into Chat stream error payloads (also
stashed on `failed`). `ChatToResponsesStream` is best effort: exact
deltas, synthesized lifecycle, assembled `response.completed`.
Response failures normalize to Chat-style `{error: ...}` objects via
`responses_error_object`. Synthesized ids (`chatcmpl_transpiled`,
`resp_transpiled`, `msg_transpiled_0`) are documented fakes.

## 4. Usage

Configure a devplacepy provider row with base URL
`https://api.meta.ai/v1/responses`, protocol Responses API, and a
route whose target model is `muse-spark-1.3-contributor`; clients keep
sending normal `POST /v1/chat/completions`. Verified example (zero
warnings both ways):

```python
from responses import chat_to_responses, responses_to_chat

chat = {
    "model": "gpt-6-astra",
    "messages": [
        {"role": "system", "content": "Be terse."},
        {"role": "user", "content": "Weather in Berlin?"},
        {"role": "assistant", "content": None, "tool_calls": [
            {"id": "call_1", "type": "function",
             "function": {"name": "get_weather",
                          "arguments": '{"location": "Berlin"}'}}]},
        {"role": "tool", "tool_call_id": "call_1", "content": "12C, rain"},
    ],
    "tools": [{"type": "function", "function": {
        "name": "get_weather", "description": "City weather",
        "parameters": {"type": "object",
                       "properties": {"location": {"type": "string"}}}}}],
    "tool_choice": "auto",
}
resp, w1 = chat_to_responses(chat)   # instructions + 3 Items, flat tool
back, w2 = responses_to_chat(resp)   # [system, user, assistant, tool]
assert (w1, w2) == ([], [])
```

## 5. Complete transpiler source

Stdlib only (`typing`). Verbatim copy of the tested module:

```python
# retoor <retoor@molodetz.nl>

"""Chat Completions <-> Responses protocol transpiler.

Lets the gateway keep its normal OpenAI surface (`POST /v1/chat/completions`
in, ChatCompletion JSON or Chat chunks out) while forwarding to a
Responses-protocol upstream such as Meta's `.../v1/responses`. The provider
row selects the protocol; `handle_chat` calls into this module at the send
boundary and on the way back.

Every lossy mapping is reported in a warnings list, never dropped silently.
Core conversions (text, images, function tools/calls, structured outputs,
scalars, common stream events) are exact both ways. Known drops: Chat `n`,
`stop`, seeds/penalties, audio/file parts, custom tools; Responses
`namespace`/built-in tools, reasoning items, `previous_response_id` /
conversation state, `truncation`, `include`, extra choices.

Grounded in the official OpenAI migration guide and both API references
(retrieved 2026-10-11) plus the Meta wire capture; Meta response bytes were
never observed, so the Response-object/SSE shapes are the OpenAI standard
ones the Meta client demonstrably shares on the request side.
"""

from __future__ import annotations

from typing import Any, Optional


def _text_of_content(content: Any) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return "".join(
            str(part.get("text") or part.get("refusal") or "")
            for part in content
            if isinstance(part, dict)
        )
    return ""


def _chat_parts_to_input_parts(parts: Any, warn: list, where: str) -> list:
    out = []
    for part in parts or []:
        if not isinstance(part, dict):
            continue
        kind = part.get("type")
        if kind == "text":
            out.append({"type": "input_text", "text": part.get("text") or ""})
        elif kind == "image_url":
            image = part.get("image_url") or {}
            item = {"type": "input_image", "image_url": image.get("url") or ""}
            if image.get("detail"):
                item["detail"] = image["detail"]
            out.append(item)
        elif kind == "refusal":
            warn.append(
                "%s: refusal part has no Responses input shape; kept as text"
                % (where,)
            )
            out.append({"type": "input_text", "text": part.get("refusal") or ""})
        else:
            warn.append(
                "%s: dropped content part type %r (no verified Responses mapping)"
                % (where, kind)
            )
    return out


def _chat_messages_to_input(messages: Any) -> tuple[str, list, list]:
    """Split Chat messages into Responses `instructions` + `input` items."""
    warn: list = []
    instructions: list = []
    items: list = []
    leading = True
    for message in messages or []:
        if not isinstance(message, dict):
            continue
        role = message.get("role") or "user"
        if role in ("system", "developer") and leading:
            text = _text_of_content(message.get("content")).strip()
            if text:
                instructions.append(text)
            continue
        leading = False
        if role == "tool":
            items.append(
                {
                    "type": "function_call_output",
                    "call_id": message.get("tool_call_id") or "",
                    "output": _text_of_content(message.get("content")),
                }
            )
            continue
        content = message.get("content")
        if isinstance(content, list):
            content = _chat_parts_to_input_parts(
                content, warn, "message[%s]" % (role,)
            )
        calls = (
            list(message.get("tool_calls") or []) if role == "assistant" else []
        )
        legacy = (
            (message.get("function_call") or {}) if role == "assistant" else {}
        )
        empty = content is None or content == "" or content == []
        if not (role == "assistant" and empty and (calls or legacy.get("name"))):
            items.append(
                {
                    "type": "message",
                    "role": role,
                    "content": content if content is not None else "",
                }
            )
        if role == "assistant":
            for call in calls:
                func = (
                    (call.get("function") or {})
                    if isinstance(call, dict)
                    else {}
                )
                items.append(
                    {
                        "type": "function_call",
                        "call_id": (
                            (call.get("id") if isinstance(call, dict) else "")
                            or ""
                        ),
                        "name": func.get("name") or "",
                        "arguments": func.get("arguments") or "",
                    }
                )
            if legacy.get("name"):
                warn.append(
                    "deprecated assistant.function_call converted without a "
                    "stable id"
                )
                items.append(
                    {
                        "type": "function_call",
                        "call_id": "call_legacy",
                        "name": legacy.get("name") or "",
                        "arguments": legacy.get("arguments") or "",
                    }
                )
        for dropped in ("name", "audio"):
            if message.get(dropped) is not None:
                warn.append(
                    "message[%s]: dropped %r (no Responses input field)"
                    % (role, dropped)
                )
    return "\n\n".join(instructions), items, warn


def _chat_tool_to_responses(tool: Any, warn: list) -> Optional[dict]:
    if not isinstance(tool, dict):
        return None
    if tool.get("type") == "function":
        func = tool.get("function") or {}
        out = {
            "type": "function",
            "name": func.get("name") or "",
            "parameters": func.get("parameters")
            or {"type": "object", "properties": {}},
        }
        if func.get("description") is not None:
            out["description"] = func["description"]
        # Chat is non-strict by default; Responses attempts strict mode when
        # the key is omitted, so pin the Chat default explicitly.
        out["strict"] = bool(func.get("strict", False))
        return out
    warn.append(
        "dropped tool type %r (only function tools transpile)"
        % (tool.get("type"),)
    )
    return None


def _chat_tool_choice_to_responses(choice: Any, warn: list) -> Any:
    if choice in (None, "none", "auto", "required"):
        return choice
    if isinstance(choice, dict):
        if choice.get("type") == "function":
            return {
                "type": "function",
                "name": (choice.get("function") or {}).get("name") or "",
            }
        if choice.get("type") == "allowed_tools":
            tools = []
            for entry in choice.get("tools") or []:
                if isinstance(entry, dict) and entry.get("type") == "function":
                    tools.append(
                        {
                            "type": "function",
                            "name": (entry.get("function") or {}).get("name")
                            or "",
                        }
                    )
                else:
                    warn.append("dropped allowed_tools entry %r" % (entry,))
            return {
                "type": "allowed_tools",
                "mode": choice.get("mode") or "auto",
                "tools": tools,
            }
    warn.append("dropped tool_choice %r (no Responses equivalent)" % (choice,))
    return None


def _chat_response_format_to_text(fmt: Any, warn: list) -> Optional[dict]:
    if not isinstance(fmt, dict):
        return None
    kind = fmt.get("type") or "text"
    if kind == "text":
        return {"format": {"type": "text"}}
    if kind == "json_object":
        return {"format": {"type": "json_object"}}
    if kind == "json_schema":
        schema = fmt.get("json_schema") or {}
        out = {
            "format": {
                "type": "json_schema",
                "name": schema.get("name") or "result",
                "schema": schema.get("schema") or {},
            }
        }
        if schema.get("description") is not None:
            out["format"]["description"] = schema["description"]
        if schema.get("strict") is not None:
            out["format"]["strict"] = schema["strict"]
        return out
    warn.append("dropped response_format type %r" % (kind,))
    return None


# Chat scalar fields with a direct Responses counterpart.
_CHAT_SCALARS = (
    ("model", "model"),
    ("temperature", "temperature"),
    ("top_p", "top_p"),
    ("stream", "stream"),
    ("store", "store"),
    ("metadata", "metadata"),
    ("user", "user"),
    ("service_tier", "service_tier"),
    ("parallel_tool_calls", "parallel_tool_calls"),
    ("prompt_cache_key", "prompt_cache_key"),
    ("max_tokens", "max_output_tokens"),
    ("max_completion_tokens", "max_output_tokens"),
)

# Chat fields with no Responses equivalent (verified against the reference).
_CHAT_DROPPED = (
    "n",
    "stop",
    "seed",
    "logprobs",
    "frequency_penalty",
    "presence_penalty",
    "logit_bias",
    "prediction",
    "modalities",
    "audio",
    "web_search_options",
    "prompt_cache_options",
    "prompt_cache_retention",
    "safety_identifier",
    "stream_options",
    "verbosity",
)

# Top-level Chat fields the transpiler understands (mapped, converted, or
# explicitly dropped above). Anything else is warned about, never silently
# ignored, so a gateway dialect field can never vanish here unnoticed.
_CHAT_KNOWN = (
    frozenset(src for src, _ in _CHAT_SCALARS)
    | frozenset(_CHAT_DROPPED)
    | frozenset(
        {
            "messages",
            "tools",
            "tool_choice",
            "response_format",
            "reasoning_effort",
            "reasoning",
            "thinking",
            "think",
            "enable_thinking",
            "top_logprobs",
        }
    )
)


_THINKING_FALSE = {False, 0, "0", "false", "no", "off", "none", "disabled"}
_THINKING_TRUE = {True, 1, "1", "true", "yes", "on", "enabled"}


def _thinking_flag(value: Any) -> Optional[bool]:
    if isinstance(value, str):
        value = value.strip().lower()
    if value in _THINKING_FALSE:
        return False
    if value in _THINKING_TRUE:
        return True
    return None


def _chat_thinking_to_reasoning(chat: dict, out: dict, warn: list) -> None:
    # Explicit effort wins; a Responses-shaped reasoning dict passes its
    # effort through; chat-dialect thinking flags collapse to effort "none"
    # when disabled and to "upstream default" (unset) when enabled.
    if chat.get("reasoning_effort") is not None:
        out["reasoning"] = {"effort": chat["reasoning_effort"]}
        return
    reasoning = chat.get("reasoning")
    if isinstance(reasoning, dict):
        if reasoning.get("effort") is not None:
            out["reasoning"] = {"effort": reasoning["effort"]}
            extra = sorted(k for k in reasoning if k != "effort")
            if extra:
                warn.append(
                    "dropped reasoning keys %r (only effort transpiles)"
                    % (extra,)
                )
            return
        enabled = _thinking_flag(reasoning.get("enabled"))
        if enabled is False:
            out["reasoning"] = {"effort": "none"}
            return
        if enabled is True:
            return
    thinking = chat.get("thinking")
    if isinstance(thinking, dict):
        enabled = _thinking_flag(thinking.get("type"))
        if enabled is False:
            out["reasoning"] = {"effort": "none"}
            return
        if enabled is True:
            return
    for key in ("think", "enable_thinking"):
        enabled = _thinking_flag(chat.get(key))
        if enabled is False:
            out["reasoning"] = {"effort": "none"}
            return
        if enabled is True:
            return


def chat_to_responses(chat: dict) -> tuple[dict, list]:
    """Convert a `POST /v1/chat/completions` body to `POST /responses`."""
    warn: list = []
    out: dict = {}
    for src, dst in _CHAT_SCALARS:
        if chat.get(src) is not None and dst not in out:
            out[dst] = chat[src]
    if chat.get("top_logprobs") is not None:
        out["top_logprobs"] = chat["top_logprobs"]
    _chat_thinking_to_reasoning(chat, out, warn)
    instructions, items, converted_warnings = _chat_messages_to_input(
        chat.get("messages")
    )
    warn += converted_warnings
    if instructions:
        out["instructions"] = instructions
    out["input"] = items
    tools = []
    for tool in chat.get("tools") or []:
        converted = _chat_tool_to_responses(tool, warn)
        if converted is not None:
            tools.append(converted)
    if tools:
        out["tools"] = tools
    if chat.get("tool_choice") is not None:
        choice = _chat_tool_choice_to_responses(chat["tool_choice"], warn)
        if choice is not None:
            out["tool_choice"] = choice
    if chat.get("response_format") is not None:
        text = _chat_response_format_to_text(chat["response_format"], warn)
        if text is not None:
            out["text"] = text
    for key in _CHAT_DROPPED:
        if chat.get(key) is not None:
            warn.append("dropped chat field %r (no Responses equivalent)" % (key,))
    for key in sorted(chat):
        if key not in _CHAT_KNOWN:
            warn.append("ignored unknown chat field %r" % (key,))
    return out, warn


def _input_parts_to_chat_parts(parts: Any, warn: list, where: str) -> list:
    out = []
    for part in parts or []:
        if not isinstance(part, dict):
            continue
        kind = part.get("type")
        if kind in ("input_text", "output_text", "text"):
            out.append({"type": "text", "text": part.get("text") or ""})
        elif kind == "input_image":
            image = {"url": part.get("image_url") or ""}
            if part.get("detail"):
                image["detail"] = part["detail"]
            out.append({"type": "image_url", "image_url": image})
        elif kind == "refusal":
            out.append({"type": "refusal", "refusal": part.get("refusal") or ""})
        else:
            warn.append(
                "%s: dropped input part type %r (no Chat mapping)"
                % (where, kind)
            )
    return out


def _call_output_to_text(output: Any, warn: list) -> str:
    if isinstance(output, str):
        return output
    if isinstance(output, list):
        texts = []
        for part in output:
            if not isinstance(part, dict):
                continue
            if part.get("type") == "input_text":
                texts.append(part.get("text") or "")
            else:
                warn.append(
                    "dropped function_call_output part type %r "
                    "(tool messages carry text only)" % (part.get("type"),)
                )
        return "".join(texts)
    return ""


def _responses_input_to_chat_messages(
    input_items: Any, warn: list
) -> list:
    messages: list = []
    pending_calls: list = []
    for item in input_items or []:
        if isinstance(item, str):
            messages.append({"role": "user", "content": item})
            continue
        if not isinstance(item, dict):
            continue
        kind = item.get("type") or "message"
        if kind == "message":
            content = item.get("content")
            if isinstance(content, list):
                content = _input_parts_to_chat_parts(
                    content, warn, "input.message"
                )
            messages.append(
                {"role": item.get("role") or "user", "content": content}
            )
            continue
        if kind == "function_call":
            pending_calls.append(
                {
                    "id": item.get("call_id") or item.get("id") or "",
                    "type": "function",
                    "function": {
                        "name": item.get("name") or "",
                        "arguments": item.get("arguments") or "",
                    },
                }
            )
            continue
        if kind == "function_call_output":
            if pending_calls:
                messages.append(
                    {
                        "role": "assistant",
                        "content": None,
                        "tool_calls": pending_calls,
                    }
                )
                pending_calls = []
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": item.get("call_id") or "",
                    "content": _call_output_to_text(item.get("output"), warn),
                }
            )
            continue
        warn.append(
            "dropped input item type %r "
            "(reasoning/native-tool items do not replay in Chat)" % (kind,)
        )
    if pending_calls:
        messages.append(
            {"role": "assistant", "content": None, "tool_calls": pending_calls}
        )
    return messages


def _responses_tool_to_chat(tool: Any, warn: list) -> Optional[dict]:
    if not isinstance(tool, dict):
        return None
    if tool.get("type") == "function":
        func: dict = {"name": tool.get("name") or ""}
        if tool.get("description") is not None:
            func["description"] = tool["description"]
        if tool.get("parameters") is not None:
            func["parameters"] = tool["parameters"]
        if tool.get("strict") is not None:
            func["strict"] = tool["strict"]
        return {"type": "function", "function": func}
    warn.append(
        "dropped tool type %r (namespaces and builtins have no Chat shape)"
        % (tool.get("type"),)
    )
    return None


def _responses_tool_choice_to_chat(choice: Any, warn: list) -> Any:
    if choice in (None, "none", "auto", "required"):
        return choice
    if isinstance(choice, dict):
        if choice.get("type") == "function":
            return {
                "type": "function",
                "function": {"name": choice.get("name") or ""},
            }
        if choice.get("type") == "allowed_tools":
            tools = []
            for entry in choice.get("tools") or []:
                if isinstance(entry, dict) and entry.get("type") == "function":
                    tools.append(
                        {
                            "type": "function",
                            "function": {"name": entry.get("name") or ""},
                        }
                    )
                else:
                    warn.append("dropped allowed_tools entry %r" % (entry,))
            return {
                "type": "allowed_tools",
                "mode": choice.get("mode") or "auto",
                "tools": tools,
            }
    warn.append("dropped tool_choice %r (no Chat equivalent)" % (choice,))
    return None


def _responses_text_to_response_format(text: Any, warn: list) -> Optional[dict]:
    fmt = (text or {}).get("format") if isinstance(text, dict) else None
    if not isinstance(fmt, dict):
        return None
    kind = fmt.get("type") or "text"
    if kind == "text":
        return None
    if kind == "json_object":
        return {"type": "json_object"}
    if kind == "json_schema":
        schema: dict = {
            "name": fmt.get("name") or "result",
            "schema": fmt.get("schema") or {},
        }
        if fmt.get("description") is not None:
            schema["description"] = fmt["description"]
        if fmt.get("strict") is not None:
            schema["strict"] = fmt["strict"]
        return {"type": "json_schema", "json_schema": schema}
    warn.append("dropped text.format type %r" % (kind,))
    return None


_RESPONSES_SCALARS = (
    ("model", "model"),
    ("temperature", "temperature"),
    ("top_p", "top_p"),
    ("stream", "stream"),
    ("store", "store"),
    ("metadata", "metadata"),
    ("user", "user"),
    ("service_tier", "service_tier"),
    ("parallel_tool_calls", "parallel_tool_calls"),
    ("prompt_cache_key", "prompt_cache_key"),
    ("top_logprobs", "top_logprobs"),
    ("max_output_tokens", "max_tokens"),
)

_RESPONSES_DROPPED = (
    "previous_response_id",
    "conversation",
    "truncation",
    "include",
    "background",
    "max_tool_calls",
    "prompt",
    "prompt_cache_options",
    "prompt_cache_retention",
    "safety_identifier",
    "stream_options",
    "context_management",
    "moderation",
    "access_programs",
)

_RESPONSES_KNOWN = (
    frozenset(src for src, _ in _RESPONSES_SCALARS)
    | frozenset(_RESPONSES_DROPPED)
    | frozenset(
        {
            "input",
            "instructions",
            "tools",
            "tool_choice",
            "text",
            "reasoning",
        }
    )
)


def responses_to_chat(resp: dict) -> tuple[dict, list]:
    """Convert a `POST /responses` body to `POST /v1/chat/completions`."""
    warn: list = []
    out: dict = {}
    for src, dst in _RESPONSES_SCALARS:
        if resp.get(src) is not None:
            out[dst] = resp[src]
    reasoning = resp.get("reasoning") or {}
    if isinstance(reasoning, dict) and reasoning.get("effort") is not None:
        out["reasoning_effort"] = reasoning["effort"]
    raw_input = resp.get("input")
    items = (
        [{"type": "message", "role": "user", "content": raw_input}]
        if isinstance(raw_input, str)
        else raw_input
    )
    messages = _responses_input_to_chat_messages(items, warn)
    instructions = resp.get("instructions")
    if isinstance(instructions, str) and instructions.strip():
        messages.insert(0, {"role": "system", "content": instructions})
    elif isinstance(instructions, list):
        messages[0:0] = _responses_input_to_chat_messages(instructions, warn)
    out["messages"] = messages
    tools = []
    for tool in resp.get("tools") or []:
        converted = _responses_tool_to_chat(tool, warn)
        if converted is not None:
            tools.append(converted)
    if tools:
        out["tools"] = tools
    if resp.get("tool_choice") is not None:
        choice = _responses_tool_choice_to_chat(resp["tool_choice"], warn)
        if choice is not None:
            out["tool_choice"] = choice
    if resp.get("text") is not None:
        fmt = _responses_text_to_response_format(resp["text"], warn)
        if fmt is not None:
            out["response_format"] = fmt
    for key in _RESPONSES_DROPPED:
        if resp.get(key) is not None:
            warn.append(
                "dropped responses field %r (no Chat equivalent)" % (key,)
            )
    for key in sorted(resp):
        if key not in _RESPONSES_KNOWN:
            warn.append("ignored unknown responses field %r" % (key,))
    return out, warn


def responses_object_to_chat_completion(resp: dict) -> tuple[dict, list]:
    """Convert a Responses object to a ChatCompletion object (one choice)."""
    warn: list = []
    text: list = []
    refusal: list = []
    calls: list = []
    for item in resp.get("output") or []:
        if not isinstance(item, dict):
            continue
        if item.get("type") == "message":
            for part in item.get("content") or []:
                if not isinstance(part, dict):
                    continue
                if part.get("type") == "output_text":
                    text.append(part.get("text") or "")
                elif part.get("type") == "refusal":
                    refusal.append(part.get("refusal") or "")
                else:
                    warn.append(
                        "dropped output part type %r" % (part.get("type"),)
                    )
        elif item.get("type") == "function_call":
            calls.append(
                {
                    "id": item.get("call_id") or item.get("id") or "",
                    "type": "function",
                    "function": {
                        "name": item.get("name") or "",
                        "arguments": item.get("arguments") or "",
                    },
                }
            )
        else:
            warn.append("dropped output item type %r" % (item.get("type"),))
    message: dict = {"role": "assistant", "content": "".join(text) or None}
    if refusal:
        message["refusal"] = "".join(refusal)
    if calls:
        message["tool_calls"] = calls
    incomplete = (resp.get("incomplete_details") or {}).get("reason")
    if calls:
        finish = "tool_calls"
    elif incomplete == "max_output_tokens":
        finish = "length"
    elif incomplete == "content_filter":
        finish = "content_filter"
    else:
        finish = "stop"
    usage = resp.get("usage") or {}
    return {
        "id": resp.get("id") or "",
        "object": "chat.completion",
        "created": resp.get("created_at") or 0,
        "model": resp.get("model") or "",
        "choices": [{"index": 0, "message": message, "finish_reason": finish}],
        "usage": {
            "prompt_tokens": usage.get("input_tokens") or 0,
            "completion_tokens": usage.get("output_tokens") or 0,
            "total_tokens": usage.get("total_tokens") or 0,
        },
    }, warn


def chat_completion_to_responses_object(comp: dict) -> tuple[dict, list]:
    """Convert a ChatCompletion object to a Responses object (choice 0)."""
    warn: list = []
    choices = comp.get("choices") or []
    if len(choices) > 1:
        warn.append(
            "kept choice 0 of %d (Responses returns one generation)"
            % (len(choices),)
        )
    message = (choices[0] or {}).get("message") if choices else {}
    output: list = []
    if (message.get("content") or None) is not None or not message.get(
        "tool_calls"
    ):
        output.append(
            {
                "id": "msg_transpiled_0",
                "type": "message",
                "status": "completed",
                "role": "assistant",
                "content": [
                    {
                        "type": "output_text",
                        "text": message.get("content") or "",
                        "annotations": [],
                    }
                ],
            }
        )
    if message.get("refusal"):
        output.append(
            {
                "id": "msg_transpiled_refusal",
                "type": "message",
                "status": "completed",
                "role": "assistant",
                "content": [
                    {"type": "refusal", "refusal": message["refusal"]}
                ],
            }
        )
    for call in message.get("tool_calls") or []:
        func = (
            (call.get("function") or {}) if isinstance(call, dict) else {}
        )
        output.append(
            {
                "type": "function_call",
                "call_id": (
                    (call.get("id") if isinstance(call, dict) else "") or ""
                ),
                "name": func.get("name") or "",
                "arguments": func.get("arguments") or "",
            }
        )
    usage = comp.get("usage") or {}
    return {
        "id": comp.get("id") or "",
        "object": "response",
        "created_at": comp.get("created") or 0,
        "status": "completed",
        "model": comp.get("model") or "",
        "output": output,
        "usage": {
            "input_tokens": usage.get("prompt_tokens") or 0,
            "output_tokens": usage.get("completion_tokens") or 0,
            "total_tokens": usage.get("total_tokens") or 0,
        },
    }, warn


def responses_failure_message(data: Any) -> Optional[str]:
    """Extract a failure message from a 200 Responses body, if any."""
    if not isinstance(data, dict):
        return None
    error = data.get("error")
    if isinstance(error, dict):
        message = error.get("message") or error.get("code") or "upstream error"
        return str(message)[:300]
    if data.get("object") == "response" and data.get("status") in (
        "failed",
        "incomplete",
        "cancelled",
    ):
        details = data.get("incomplete_details") or {}
        reason = details.get("reason") or data.get("status")
        return "response %s: %s" % (data.get("status"), reason)
    return None


def responses_error_object(data: Any) -> Optional[dict]:
    """Normalize a Responses error body to a Chat-style error object."""
    if not isinstance(data, dict):
        return None
    error = data.get("error")
    if isinstance(error, dict):
        out = {"message": str(error.get("message") or "upstream error")}
        if error.get("type") is not None:
            out["type"] = error["type"]
        if error.get("code") is not None:
            out["code"] = error["code"]
        return {"error": out}
    message = responses_failure_message(data)
    if message:
        return {"error": {"message": message, "type": "upstream_error"}}
    return None


class ResponsesToChatStream:
    """Translate Responses SSE events into ChatCompletion chunks.

    Text deltas and function-call argument deltas map exactly; lifecycle
    events are consumed; `response.completed` yields the terminal chunk with
    the real finish reason and usage. Error events yield `{"error": ...}`
    payloads (the Chat stream error shape) and are also stashed on `failed`.
    """

    def __init__(self, model: str = ""):
        self.model = model
        self.calls: dict = {}
        self.order: list = []
        self.failed: Any = None
        self.role_sent = False
        self.text_sent = False

    def feed(self, event: Any) -> list:
        if not isinstance(event, dict):
            return []
        kind = event.get("type") or ""
        if kind == "response.output_text.delta":
            delta: dict = {"content": event.get("delta") or ""}
            if not self.role_sent:
                delta = {"role": "assistant", **delta}
                self.role_sent = True
            self.text_sent = True
            return [self._chunk(delta)]
        if kind == "response.output_item.added":
            item = event.get("item") or {}
            if item.get("type") == "function_call" and item.get("call_id"):
                index = self._index(item["call_id"])
                return [
                    self._chunk(
                        {
                            "tool_calls": [
                                {
                                    "index": index,
                                    "id": item["call_id"],
                                    "type": "function",
                                    "function": {
                                        "name": item.get("name") or "",
                                        "arguments": "",
                                    },
                                }
                            ]
                        }
                    )
                ]
            return []
        if kind == "response.function_call_arguments.delta":
            call_id = event.get("item_id") or ""
            return [
                self._chunk(
                    {
                        "tool_calls": [
                            {
                                "index": self._index(call_id),
                                "function": {
                                    "arguments": event.get("delta") or ""
                                },
                            }
                        ]
                    }
                )
            ]
        if kind == "response.completed":
            response = event.get("response") or {}
            converted, _ = responses_object_to_chat_completion(response)
            chunks = []
            message = (converted["choices"][0] or {}).get("message") or {}
            if message.get("refusal") and not self.text_sent:
                chunks.append(self._chunk({"refusal": message["refusal"]}))
            chunk = self._chunk({})
            chunk["choices"][0]["finish_reason"] = (
                converted["choices"][0] or {}
            ).get("finish_reason")
            chunk["usage"] = converted["usage"]
            chunks.append(chunk)
            return chunks
        if kind in ("response.failed", "response.incomplete", "error"):
            self.failed = event
            payload = responses_error_object(
                event.get("response") or event.get("error") or event
            ) or {"error": {"message": kind, "type": "upstream_error"}}
            return [payload]
        return []

    def _index(self, call_id: str) -> int:
        if call_id not in self.calls:
            self.calls[call_id] = len(self.order)
            self.order.append(call_id)
        return self.calls[call_id]

    def _chunk(self, delta: dict) -> dict:
        return {
            "id": "chatcmpl_transpiled",
            "object": "chat.completion.chunk",
            "created": 0,
            "model": self.model,
            "choices": [{"index": 0, "delta": delta, "finish_reason": None}],
        }


class ChatToResponsesStream:
    """Translate ChatCompletion chunks into Responses SSE events.

    Best effort: text and tool-call deltas map exactly, but Chat chunks
    carry no lifecycle events, so the synthesized stream skips
    `response.created` / `output_item.added` for text and emits
    `response.completed` with an assembled object. Streaming call ids are
    tracked by tool-call index, per the Chat streaming contract (the id
    arrives on the first chunk for each index).
    """

    def __init__(self):
        self.text: list = []
        self.calls: dict = {}
        self.order: list = []
        self.warnings: list = []

    def feed(self, chunk: Any) -> list:
        events = []
        choices = (chunk or {}).get("choices") or []
        delta = (choices[0] or {}).get("delta") if choices else {}
        for call in (delta or {}).get("tool_calls") or []:
            if not isinstance(call, dict):
                continue
            index = call.get("index")
            func = call.get("function") or {}
            call_id = call.get("id")
            if call_id and call_id not in self.calls:
                self.calls[call_id] = True
                self.order.append(call_id)
                events.append(
                    {
                        "type": "response.output_item.added",
                        "output_index": len(self.order) - 1,
                        "item": {
                            "type": "function_call",
                            "call_id": call_id,
                            "name": func.get("name") or "",
                            "arguments": "",
                        },
                    }
                )
            if not call_id and isinstance(index, int) and index < len(
                self.order
            ):
                call_id = self.order[index]
            if func.get("arguments"):
                if not call_id:
                    self.warnings.append(
                        "tool-call arguments arrived before any call id"
                    )
                events.append(
                    {
                        "type": "response.function_call_arguments.delta",
                        "item_id": call_id or "",
                        "output_index": 0,
                        "delta": func["arguments"],
                    }
                )
        piece = (delta or {}).get("content")
        if piece:
            self.text.append(piece)
            events.append(
                {
                    "type": "response.output_text.delta",
                    "item_id": "msg_transpiled_0",
                    "output_index": 0,
                    "content_index": 0,
                    "delta": piece,
                }
            )
        finish = (choices[0] or {}).get("finish_reason") if choices else None
        if finish:
            output = [
                {
                    "id": "msg_transpiled_0",
                    "type": "message",
                    "status": "completed",
                    "role": "assistant",
                    "content": [
                        {
                            "type": "output_text",
                            "text": "".join(self.text),
                            "annotations": [],
                        }
                    ],
                }
            ]
            events.append(
                {
                    "type": "response.completed",
                    "response": {
                        "id": "resp_transpiled",
                        "object": "response",
                        "status": "completed",
                        "model": (chunk or {}).get("model") or "",
                        "output": output,
                    },
                }
            )
        return events
```

## 6. Verification status and known gaps

Module tests: 21/21 green
(`tests/unit/services/openai_gateway/responses.py` in devplacepy),
covering both request directions, both object converters, both
stream translators, error/failure shapes, thinking-dialect collapse,
and every documented drop. A real captured Meta request round-trips
through `responses_to_chat` and back with only the expected
`namespace` and `include` warnings. The weather example in section 4
was executed with zero warnings both ways.

Known gaps, stated plainly:

- Meta response and SSE bytes were never observed; that side rests on
  the documented standard shapes. A recording forward proxy closes it.
- Audio and file content parts warn and drop; their exact cross-API
  shapes were not verified.
- Synthesized stream/object ids are documented fakes, not upstream ids.
- The devplacepy gateway wiring (provider protocol setting, send-time
  transpile, Responses stream relay) was mid-integration at snapshot
  time with gateway unit failures under diagnosis; only the module
  itself is green.

## 7. Sources

Retrieved 2026-10-11. Full dossiers live in
`research/muse-responses-openai-transpiler/`.

- OpenAI, Migrate to the Responses API:
  https://developers.openai.com/api/docs/guides/migrate-to-responses
- OpenAI, Responses API reference:
  https://developers.openai.com/api/reference/resources/responses
- OpenAI, Chat Completions API reference:
  https://developers.openai.com/api/reference/resources/chat
- Local MITM capture of the real `muse` CLI:
  `/tmp/mitm_probe/request.txt` (internal evidence, 11 requests)

## 8. Changelog

| Date | Change |
|---|---|
| 2026-10-11 | Consolidated research + verbatim transpiler source into one document. |