[Feature/Robustness] Complete agentic workflow support in Responses API: tolerant input items & compaction output handling #695

Summary

When using Splash as an inference backend for long-running agentic workflows and multi-turn orchestrators implementing the OpenAI Responses API (/v1/responses), two architectural incompatibilities occur:

1. Input rejection (HTTP 400): Agents send execution history items such as commandExecution, mcpToolCall, and contextCompaction that are strictly rejected by normalize_responses_input().
2. Context compaction output mismatch: When conversations approach context limits and the orchestrator requests history compaction/checkpointing, Splash emits regular reasoning + message items instead of recognizing the compaction turn and producing an item of type "compaction".

Problem 1: Ingestion of Agentic Input Items (Input)

In server/api_shapes.py, normalize_responses_input() enforces a strict whitelist of item kinds (message, reasoning, function_call, function_call_output, custom_tool_call, and custom_tool_call_output).

Modern agent harnesses frequently include:

• commandExecution: Terminal/shell command executions with output and call_id / id.
• mcpToolCall: Model Context Protocol (MCP) tool execution outputs.
• contextCompaction / item_reference: Historical checkpoints and references.

Solution 1: Tolerant Input Normalization (server/api_shapes.py)

Instead of raising APIError(400) for non-whitelisted items:

• Items containing a call_id (or id) and an output field are mapped to role "tool" (function_call_output).
• Items with textual content are added to the conversational context.
• Checkpoints and metadata-only items are safely passed through without failing the request.

        elif kind in ("function_call_output", "custom_tool_call_output"):
            call_id = item.get("call_id")
            if not isinstance(call_id, str) or not call_id:
                raise APIError(400, "call_id must be a non-empty string")
            output = item.get("output")
            flush()
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": call_id,
                    "content": _responses_content(output),
                }
            )
        else:
            # Tolerant fallback for agentic input items (MCP tools, commandExecution, checkpoints)
            call_id = item.get("call_id") or item.get("id")
            content = item.get("content")
            if call_id and "output" in item:
                flush()
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": str(call_id),
                        "content": _responses_content(item.get("output")),
                    }
                )
            elif content:
                flush()
                extracted = (
                    _responses_text(content)
                    if not isinstance(content, str)
                    else content
                )
                if extracted:
                    messages.append(
                        {
                            "role": "user",
                            "content": extracted,
                        }
                    )

Problem 2: Context Compaction Output Handling (Output)

In long-running agent workflows, when sessions approach token limits, orchestrators invoke compaction requests to summarize and prune previous turns.

The client expects the response to return a compaction output item (type: "compaction" with id prefix cmp_...). Because Splash returns standard separate output items (reasoning + message), clients fail with:

Fatal error: remote compaction expected exactly one compaction output item, got 0 from N output items

Solution 2: Compaction Output Support (server/api_shapes.py & server/server.py)

1. Register "compaction": "cmp" in _RESPONSES_ITEM_PREFIXES.
2. When a compaction request is processed (detected via instructions or compaction markers in input), coalesce the generated summary into a single compaction item:

_RESPONSES_ITEM_PREFIXES = {
    "reasoning": "rs",
    "text": "msg",
    "tool": "fc",
    "custom": "ctc",
    "compaction": "cmp",
}

def responses_output(job, blocks):
    if getattr(job, "is_compaction", False):
        full_text = "".join(b.text for b in blocks if getattr(b, "text", None))
        return [{
            "id": responses_item_id(job, "compaction", 0),
            "type": "compaction",
            "status": "completed",
            "summary": [{"type": "summary_text", "text": full_text}],
            "content": [{"type": "output_text", "text": full_text, "annotations": []}],
            "encrypted_content": None,
        }]
    return [responses_item(job, block, index) for index, block in enumerate(blocks)]

Benefits

• Full agent lifecycle compatibility: Allows any agent orchestrator using Responses API to run indefinitely on Apple Silicon via Splash, executing shell commands, MCP tools, and automatic context pruning.
• Strictly backward-compatible: Standard chat and tool calls remain 100% untouched.
• Tested locally on Splash 1.5.0 with mlx-community/Qwen3.6-35B-A3B-4bit: multi-turn agent sessions execute seamlessly across dozens of tool calls.