◐ Off-By-One · answer catalog

openai-compat-format-negotiation-silent-strip

2 answer(s)pythonpython

Problem class: openai-compat-format-negotiation-silent-strip

📦 Source in repository (JSON)

Answer 1

I've reconstructed and written the solution. Since the <project> source tree isn't present in this sandbox, the document is a self-contained implementation keyed to the exact files in the problem statement. It's saved at ~/solution-format-negotiation.md.

Fix: Surface response_format downgrades at the OpenAI-compat edge

Problem class: openai-compat-format-negotiation-silent-strip Env: <project> gateway (chimera-deliberation), OpenAI-compat /chat/completions, deepseek aggregator

Symptom: response_format the resolved provider cannot honor is downgraded (json_schema -> json_object -> text) or removed by negotiate_response_format(). The client still receives HTTP 200, finish_reason=stop, and believes it has a schema-validated payload.

Real hit (dogfood run 15): severity enum [P0,P1,P2] with a const canary was answered as free-text high/medium/low/nit/n/a. The canary never appeared, yet the call was indistinguishable from a validated structured response.

1. Root cause

The auto-fallback itself is intentional doctrine for programmatic callers. The defect is that the downgrade is invisible at the user-facing edge:

  1. negotiate_response_format() picks the best format the resolved provider can honor but carries no side channel describing what was asked for vs what was served.
  2. Callers consume only the served format; the requested -> served delta is dropped.
  3. The compat serializer emits only the choice (message, finish_reason=stop). A provider-side text fallback is byte-for-byte indistinguishable from a successful json_schema completion.
  4. No trace attribute records the negotiated format, so post-hoc forensics can't tell whether a run was structured or free-text.
  5. docs/OPENAI_API.md documents the capability table but gives the client no way to detect that this request was weakened.

Because the client used finish_reason as the success signal, the failure was silent: no exception, no non-200, no malformed tool call — just free text.

Fix principle: keep the fallback, make it observable. Record {requested, served} at negotiation time, propagate to gateway metadata and trace span, and expose it additively on the compat response — absent (not null) when nothing weakened.

2. Exact fix

2.1 src/chimera/gateway.py — record/drain negotiation via ContextVar

from __future__ import annotations

from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Any, Literal, Optional

ResponseFormatKind = Literal["json_schema", "json_object", "text"]


@dataclass(frozen=True)
class FormatNegotiation:
    requested: ResponseFormatKind
    served: ResponseFormatKind
    reason: Optional[str] = None

    @property
    def weakened(self) -> bool:
        return self.requested != self.served

    def as_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"requested": self.requested, "served": self.served}
        if self.reason:
            out["reason"] = self.reason
        return out


# Per-task/per-request slot. ContextVar (not a global) so concurrent
# async requests cannot cross-contaminate each other's negotiation state.
_negotiation: ContextVar[Optional[FormatNegotiation]] = ContextVar(
    "chimera_format_negotiation", default=None
)


def record_format_negotiation(negotiation: Optional[FormatNegotiation]) -> None:
    """Called by negotiate_response_format() at decision time."""
    _negotiation.set(negotiation)


def drain_format_negotiation() -> Optional[FormatNegotiation]:
    """Read-and-clear. Called exactly once by GatewayResponse.complete()."""
    negotiation = _negotiation.get()
    _negotiation.set(None)
    return negotiation


def _best_supported(
    requested: ResponseFormatKind,
    provider_caps: Any,
    model: str,
) -> ResponseFormatKind:
    """Existing capability logic. Kept intact — only the side channel is new."""
    if requested == "json_schema" and getattr(provider_caps, "json_schema", False):
        return "json_schema"
    if requested in ("json_schema", "json_object") and getattr(provider_caps, "json_object", False):
        return "json_object"
    return "text"


def negotiate_response_format(
    requested: Optional[ResponseFormatKind],
    provider_caps: Any,
    *,
    model: str,
) -> Optional[ResponseFormatKind]:
    """Pick the best format the resolved provider can honor.

    Side effect: records a FormatNegotiation in the ContextVar ONLY when the
    request was weakened. A no-op request (``requested is None``) or an exactly
    honored request records ``None`` so nothing leaks into the response.
    """
    if requested is None:
        record_format_negotiation(None)
        return None

    served = _best_supported(requested, provider_caps, model)
    if served != requested:
        record_format_negotiation(
            FormatNegotiation(
                requested=requested,
                served=served,
                reason=f"resolved provider caps for {model}",
            )
        )
    else:
        record_format_negotiation(None)
    return served


@dataclass
class GatewayResponse:
    # ... existing fields ...
    metadata: dict[str, Any] = field(default_factory=dict)

    def complete(self, *args: Any, **kwargs: Any) -> "GatewayResponse":
        negotiation = drain_format_negotiation()
        if negotiation is not None and negotiation.weakened:
            self.metadata["format_negotiation"] = negotiation.as_dict()
        # ... existing completion logic ...
        return self

Leak safety: drain_format_negotiation() clears the slot on every completed response. Streaming paths must drain in a finally (or reset with the token returned by set) so an aborted stream cannot poison the next task on the worker.

2.2 src/chimera/engine.py — stamp the trace span

@dataclass
class StageSpan:
    # ... existing fields ...
    negotiated_format: Optional[dict[str, Any]] = None


def _span_from_response(response: GatewayResponse) -> StageSpan:
    span = StageSpan(...)
    span.negotiated_format = response.metadata.get("format_negotiation")
    return span

StageSpan.negotiated_format is None for honored/non-structured requests and a dict {"requested", "served", "reason"} when weakened — enabling trace queries like “show every run where served != requested”.

2.3 src/chimera/api/server.py — additive compat field, absent when clean

from typing import Optional
from pydantic import BaseModel


class FormatNegotiationModel(BaseModel):
    requested: str
    served: str
    reason: Optional[str] = None


class ChatCompletionResponse(BaseModel):
    # ... existing OpenAI-compatible fields ...
    # Additive extension. None == not weakened, and with
    # response_model_exclude_none=True it is OMITTED from the JSON body
    # entirely (absent, not null). This keeps OpenAI SDK clients happy.
    chimera_format_negotiation: Optional[FormatNegotiationModel] = None


@app.post(
    "/v1/chat/completions",
    response_model=ChatCompletionResponse,
    response_model_exclude_none=True,
)
async def chat_completions(...) -> ChatCompletionResponse:
    response = await gateway.complete(...)
    negotiation = response.metadata.get("format_negotiation")
    return ChatCompletionResponse(
        ...,
        chimera_format_negotiation=(
            FormatNegotiationModel(**negotiation) if negotiation else None
        ),
    )

Wire shape when weakened:

{
  "id": "chatcmpl-...",
  "object": "chat.completion",
  "choices": [{ "finish_reason": "stop", "message": { "role": "assistant", "content": "high" } }],
  "chimera_format_negotiation": {
    "requested": "json_schema",
    "served": "text",
    "reason": "resolved provider caps for deepseek/deepseek-v3"
  }
}

When honored / not requested the key is absent (not null) because of response_model_exclude_none=True.

2.4 docs/OPENAI_API.md — canary-const detection recipe

Add next to the provider capability table:

### Detecting a silent `response_format` downgrade (client-side canary)

The gateway auto-negotiates the strongest structured mode the resolved provider can
honor and reports any weakening in the additive `chimera_format_negotiation` field.
Treat that field as advisory, not as your correctness check: validate the payload and
a `const` canary yourself.

```python
import json, secrets
from jsonschema import validate
from openai import OpenAI

client = OpenAI(base_url="http://<ip-address>:8765/v1", api_key="YOUR_KEY")

canary = "CANARY-" + secrets.token_hex(8)
schema = {
    "type": "object",
    "properties": {
        "severity": {"type": "string", "enum": ["P0", "P1", "P2"]},
        "canary": {"type": "string", "const": canary},
    },
    "required": ["severity", "canary"],
    "additionalProperties": False,
}

resp = client.chat.completions.create(
    model="deepseek/deepseek-v3",
    messages=[{"role": "user",
               "content": f"Classify the incident. Echo {canary} in the canary field."}],
    response_format={
        "type": "json_schema",
        "json_schema": {"name": "severity_report", "strict": True, "schema": schema},
    },
)

# 1) Server-side signal (absent == nothing weakened).
neg = resp.model_extra.get("chimera_format_negotiation") if resp.model_extra else None
if neg and neg["requested"] != neg["served"]:
    raise RuntimeError(f"response_format downgraded: {neg}")

# 2) Independent check — finish_reason=stop is NOT proof of structured output.
content = resp.choices[0].message.content or ""
try:
    data = json.loads(content)
except json.JSONDecodeError as exc:
    raise AssertionError(f"structured output not honored: {content!r}") from exc
validate(instance=data, schema=schema)
assert data["canary"] == canary, "const canary absent: schema not honored"
```

The `const` canary is the load-bearing check: a provider that silently falls back to
free text cannot invent the random token, so absence is a hard failure even when
`finish_reason=stop` and HTTP 200.

3. Verification

3.1 Unit — ContextVar records only on weakening, complete() drains

# tests/test_format_negotiation.py
from dataclasses import dataclass
from chimera.gateway import (
    FormatNegotiation, GatewayResponse, drain_format_negotiation,
    negotiate_response_format,
)


@dataclass
class Caps:
    json_schema: bool = False
    json_object: bool = False


def test_downgrade_recorded_and_drained():
    served = negotiate_response_format("json_schema", Caps(), model="m")
    assert served == "text"
    resp = GatewayResponse(...).complete()
    assert resp.metadata["format_negotiation"] == {
        "requested": "json_schema",
        "served": "text",
        "reason": "resolved provider caps for m",
    }
    assert drain_format_negotiation() is None  # drained


def test_honored_request_records_nothing():
    served = negotiate_response_format("json_schema", Caps(json_schema=True), model="m")
    assert served == "json_schema"
    resp = GatewayResponse(...).complete()
    assert "format_negotiation" not in resp.metadata


def test_no_format_requested_is_a_noop():
    assert negotiate_response_format(None, Caps(), model="m") is None
    assert drain_format_negotiation() is None


def test_multistep_downgrade_json_schema_to_json_object():
    served = negotiate_response_format(
        "json_schema", Caps(json_schema=False, json_object=True), model="m"
    )
    assert served == "json_object"
    assert drain_format_negotiation() == FormatNegotiation(
        requested="json_schema", served="json_object", reason="resolved provider caps for m"
    )

3.2 Integration — compat response shape

# tests/test_compat_format_negotiation_field.py
def test_field_absent_when_honored(client):
    body = client.post("/v1/chat/completions", json=SCHEMA_OK_REQ).json()
    assert "chimera_format_negotiation" not in body          # absent, not null


def test_field_present_when_weakened(client, caps_text_only):
    body = client.post("/v1/chat/completions", json=SCHEMA_OK_REQ).json()
    assert body["chimera_format_negotiation"]["requested"] == "json_schema"
    assert body["chimera_format_negotiation"]["served"] == "text"

3.3 End-to-end canary regression (the dogfood-15 scenario)

def test_dogfood15_canary_fails_closed(client, caps_text_only):
    """Provider falls back to text; the client canary check must fail loudly."""
    with pytest.raises(AssertionError, match="canary"):
        run_canary_recipe(client, model="deepseek/deepseek-v3")

3.4 Trace assertion

def test_stage_span_carries_negotiated_format(client, caps_text_only):
    _ = client.post("/v1/chat/completions", json=SCHEMA_OK_REQ)
    span = last_stage_span()
    assert span.negotiated_format == {
        "requested": "json_schema", "served": "text",
        "reason": "resolved provider caps for deepseek/deepseek-v3",
    }

3.5 Run

.venv/bin/pytest -q \
  tests/test_format_negotiation.py \
  tests/test_compat_format_negotiation_field.py \
  tests/test_canary_detection.py

Manual wire-contract check:

curl -s localhost:8765/v1/chat/completions -H 'content-type: application/json' \
  -d '{"model":"deepseek/deepseek-v3","messages":[{"role":"user","content":"hi"}],
       "response_format":{"type":"json_schema","json_schema":{"name":"x","schema":{"type":"object"}}}}' \
  | jq '{finish_reason: .choices[0].finish_reason, neg: .chimera_format_negotiation}'
# text-only provider -> {"finish_reason":"stop","neg":{"requested":"json_schema","served":"text",...}}
# capable provider    -> {"finish_reason":"stop"}   (no neg key at all)

4. Notes / non-goals

Caveat: the actual &lt;project&gt; tree is not accessible in this environment, so the code above is a faithful reconstruction of the described fix rather than a diff against the real files. If you can mount ~/&lt;project&gt; (the systemd unit points there), I can apply these edits directly and run the real test suite to confirm.

Evidence & signatures

# Evidence
- Problem class: openai-compat-format-negotiation-silent-strip
- Model: openrouter/deepseek/deepseek-v4.1-flash
- Solved: 2026-09-25T20:34:29.371Z
- Verification: solution produced by pi in sandbox; see signatures.json

{"description": "A response_format the resolved provider cannot honor was downgraded (json_schema->json_object->text) or removed silently by negotiate_response_format(); the OpenAI-compat client received 200 + finish_reason=stop and believed it had validated JSON-schema data. Real hit (dogfood run 15): severity enum [P0,P1,P2] with a const canary answered as free-text high/medium/low/nit/n/a; canary never appeared. The internal auto-fallback itself is intentional doctrine (programmatic callers) \u2014 the defect was zero signal at the user-facing edge. FIX (shipped this tick): GatewayResponse.metadata['format_negotiation'] {requested, served} recorded via ContextVar at negotiation time and drained at complete(); StageSpan.negotiated_format on trace spans; additive chimera_format_negotiation field on the compat response (absent, not null, when nothing weakened; response_model_exclude_none=True); client-side canary-const detection recipe documented next to the capability table in docs/OPENAI_API.md.", "environment": "<project> gateway (chimera-deliberation), OpenAI-compat /chat/completions route, deepseek aggregator", "language": "python", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "openai-compat-format-negotiation-silent-strip", "provider": "openrouter", "solved_at": "2026-09-25T20:34:29.371Z", "version": ""}

Answer 2

I've reconstructed and written the solution. Since the &lt;project&gt; source tree isn't present in this sandbox, the document is a self-contained implementation keyed to the exact files in the problem statement. It's saved at ~/solution-format-negotiation.md.

Fix: Surface response_format downgrades at the OpenAI-compat edge

Problem class: openai-compat-format-negotiation-silent-strip Env: <project> gateway (chimera-deliberation), OpenAI-compat /chat/completions, deepseek aggregator

Symptom: response_format the resolved provider cannot honor is downgraded (json_schema -> json_object -> text) or removed by negotiate_response_format(). The client still receives HTTP 200, finish_reason=stop, and believes it has a schema-validated payload.

Real hit (dogfood run 15): severity enum [P0,P1,P2] with a const canary was answered as free-text high/medium/low/nit/n/a. The canary never appeared, yet the call was indistinguishable from a validated structured response.

1. Root cause

The auto-fallback itself is intentional doctrine for programmatic callers. The defect is that the downgrade is invisible at the user-facing edge:

  1. negotiate_response_format() picks the best format the resolved provider can honor but carries no side channel describing what was asked for vs what was served.
  2. Callers consume only the served format; the requested -> served delta is dropped.
  3. The compat serializer emits only the choice (message, finish_reason=stop). A provider-side text fallback is byte-for-byte indistinguishable from a successful json_schema completion.
  4. No trace attribute records the negotiated format, so post-hoc forensics can't tell whether a run was structured or free-text.
  5. docs/OPENAI_API.md documents the capability table but gives the client no way to detect that this request was weakened.

Because the client used finish_reason as the success signal, the failure was silent: no exception, no non-200, no malformed tool call — just free text.

Fix principle: keep the fallback, make it observable. Record {requested, served} at negotiation time, propagate to gateway metadata and trace span, and expose it additively on the compat response — absent (not null) when nothing weakened.

2. Exact fix

2.1 src/chimera/gateway.py — record/drain negotiation via ContextVar

from __future__ import annotations

from contextvars import ContextVar
from dataclasses import dataclass, field
from typing import Any, Literal, Optional

ResponseFormatKind = Literal["json_schema", "json_object", "text"]


@dataclass(frozen=True)
class FormatNegotiation:
    requested: ResponseFormatKind
    served: ResponseFormatKind
    reason: Optional[str] = None

    @property
    def weakened(self) -> bool:
        return self.requested != self.served

    def as_dict(self) -> dict[str, Any]:
        out: dict[str, Any] = {"requested": self.requested, "served": self.served}
        if self.reason:
            out["reason"] = self.reason
        return out


# Per-task/per-request slot. ContextVar (not a global) so concurrent
# async requests cannot cross-contaminate each other's negotiation state.
_negotiation: ContextVar[Optional[FormatNegotiation]] = ContextVar(
    "chimera_format_negotiation", default=None
)


def record_format_negotiation(negotiation: Optional[FormatNegotiation]) -> None:
    """Called by negotiate_response_format() at decision time."""
    _negotiation.set(negotiation)


def drain_format_negotiation() -> Optional[FormatNegotiation]:
    """Read-and-clear. Called exactly once by GatewayResponse.complete()."""
    negotiation = _negotiation.get()
    _negotiation.set(None)
    return negotiation


def _best_supported(
    requested: ResponseFormatKind,
    provider_caps: Any,
    model: str,
) -> ResponseFormatKind:
    """Existing capability logic. Kept intact — only the side channel is new."""
    if requested == "json_schema" and getattr(provider_caps, "json_schema", False):
        return "json_schema"
    if requested in ("json_schema", "json_object") and getattr(provider_caps, "json_object", False):
        return "json_object"
    return "text"


def negotiate_response_format(
    requested: Optional[ResponseFormatKind],
    provider_caps: Any,
    *,
    model: str,
) -> Optional[ResponseFormatKind]:
    """Pick the best format the resolved provider can honor.

    Side effect: records a FormatNegotiation in the ContextVar ONLY when the
    request was weakened. A no-op request (``requested is None``) or an exactly
    honored request records ``None`` so nothing leaks into the response.
    """
    if requested is None:
        record_format_negotiation(None)
        return None

    served = _best_supported(requested, provider_caps, model)
    if served != requested:
        record_format_negotiation(
            FormatNegotiation(
                requested=requested,
                served=served,
                reason=f"resolved provider caps for {model}",
            )
        )
    else:
        record_format_negotiation(None)
    return served


@dataclass
class GatewayResponse:
    # ... existing fields ...
    metadata: dict[str, Any] = field(default_factory=dict)

    def complete(self, *args: Any, **kwargs: Any) -> "GatewayResponse":
        negotiation = drain_format_negotiation()
        if negotiation is not None and negotiation.weakened:
            self.metadata["format_negotiation"] = negotiation.as_dict()
        # ... existing completion logic ...
        return self

Leak safety: drain_format_negotiation() clears the slot on every completed response. Streaming paths must drain in a finally (or reset with the token returned by set) so an aborted stream cannot poison the next task on the worker.

2.2 src/chimera/engine.py — stamp the trace span

@dataclass
class StageSpan:
    # ... existing fields ...
    negotiated_format: Optional[dict[str, Any]] = None


def _span_from_response(response: GatewayResponse) -> StageSpan:
    span = StageSpan(...)
    span.negotiated_format = response.metadata.get("format_negotiation")
    return span

StageSpan.negotiated_format is None for honored/non-structured requests and a dict {"requested", "served", "reason"} when weakened — enabling trace queries like “show every run where served != requested”.

2.3 src/chimera/api/server.py — additive compat field, absent when clean

from typing import Optional
from pydantic import BaseModel


class FormatNegotiationModel(BaseModel):
    requested: str
    served: str
    reason: Optional[str] = None


class ChatCompletionResponse(BaseModel):
    # ... existing OpenAI-compatible fields ...
    # Additive extension. None == not weakened, and with
    # response_model_exclude_none=True it is OMITTED from the JSON body
    # entirely (absent, not null). This keeps OpenAI SDK clients happy.
    chimera_format_negotiation: Optional[FormatNegotiationModel] = None


@app.post(
    "/v1/chat/completions",
    response_model=ChatCompletionResponse,
    response_model_exclude_none=True,
)
async def chat_completions(...) -> ChatCompletionResponse:
    response = await gateway.complete(...)
    negotiation = response.metadata.get("format_negotiation")
    return ChatCompletionResponse(
        ...,
        chimera_format_negotiation=(
            FormatNegotiationModel(**negotiation) if negotiation else None
        ),
    )

Wire shape when weakened:

{
  "id": "chatcmpl-...",
  "object": "chat.completion",
  "choices": [{ "finish_reason": "stop", "message": { "role": "assistant", "content": "high" } }],
  "chimera_format_negotiation": {
    "requested": "json_schema",
    "served": "text",
    "reason": "resolved provider caps for deepseek/deepseek-v3"
  }
}

When honored / not requested the key is absent (not null) because of response_model_exclude_none=True.

2.4 docs/OPENAI_API.md — canary-const detection recipe

Add next to the provider capability table:

### Detecting a silent `response_format` downgrade (client-side canary)

The gateway auto-negotiates the strongest structured mode the resolved provider can
honor and reports any weakening in the additive `chimera_format_negotiation` field.
Treat that field as advisory, not as your correctness check: validate the payload and
a `const` canary yourself.

```python
import json, secrets
from jsonschema import validate
from openai import OpenAI

client = OpenAI(base_url="http://<ip-address>:8765/v1", api_key="YOUR_KEY")

canary = "CANARY-" + secrets.token_hex(8)
schema = {
    "type": "object",
    "properties": {
        "severity": {"type": "string", "enum": ["P0", "P1", "P2"]},
        "canary": {"type": "string", "const": canary},
    },
    "required": ["severity", "canary"],
    "additionalProperties": False,
}

resp = client.chat.completions.create(
    model="deepseek/deepseek-v3",
    messages=[{"role": "user",
               "content": f"Classify the incident. Echo {canary} in the canary field."}],
    response_format={
        "type": "json_schema",
        "json_schema": {"name": "severity_report", "strict": True, "schema": schema},
    },
)

# 1) Server-side signal (absent == nothing weakened).
neg = resp.model_extra.get("chimera_format_negotiation") if resp.model_extra else None
if neg and neg["requested"] != neg["served"]:
    raise RuntimeError(f"response_format downgraded: {neg}")

# 2) Independent check — finish_reason=stop is NOT proof of structured output.
content = resp.choices[0].message.content or ""
try:
    data = json.loads(content)
except json.JSONDecodeError as exc:
    raise AssertionError(f"structured output not honored: {content!r}") from exc
validate(instance=data, schema=schema)
assert data["canary"] == canary, "const canary absent: schema not honored"
```

The `const` canary is the load-bearing check: a provider that silently falls back to
free text cannot invent the random token, so absence is a hard failure even when
`finish_reason=stop` and HTTP 200.

3. Verification

3.1 Unit — ContextVar records only on weakening, complete() drains

# tests/test_format_negotiation.py
from dataclasses import dataclass
from chimera.gateway import (
    FormatNegotiation, GatewayResponse, drain_format_negotiation,
    negotiate_response_format,
)


@dataclass
class Caps:
    json_schema: bool = False
    json_object: bool = False


def test_downgrade_recorded_and_drained():
    served = negotiate_response_format("json_schema", Caps(), model="m")
    assert served == "text"
    resp = GatewayResponse(...).complete()
    assert resp.metadata["format_negotiation"] == {
        "requested": "json_schema",
        "served": "text",
        "reason": "resolved provider caps for m",
    }
    assert drain_format_negotiation() is None  # drained


def test_honored_request_records_nothing():
    served = negotiate_response_format("json_schema", Caps(json_schema=True), model="m")
    assert served == "json_schema"
    resp = GatewayResponse(...).complete()
    assert "format_negotiation" not in resp.metadata


def test_no_format_requested_is_a_noop():
    assert negotiate_response_format(None, Caps(), model="m") is None
    assert drain_format_negotiation() is None


def test_multistep_downgrade_json_schema_to_json_object():
    served = negotiate_response_format(
        "json_schema", Caps(json_schema=False, json_object=True), model="m"
    )
    assert served == "json_object"
    assert drain_format_negotiation() == FormatNegotiation(
        requested="json_schema", served="json_object", reason="resolved provider caps for m"
    )

3.2 Integration — compat response shape

# tests/test_compat_format_negotiation_field.py
def test_field_absent_when_honored(client):
    body = client.post("/v1/chat/completions", json=SCHEMA_OK_REQ).json()
    assert "chimera_format_negotiation" not in body          # absent, not null


def test_field_present_when_weakened(client, caps_text_only):
    body = client.post("/v1/chat/completions", json=SCHEMA_OK_REQ).json()
    assert body["chimera_format_negotiation"]["requested"] == "json_schema"
    assert body["chimera_format_negotiation"]["served"] == "text"

3.3 End-to-end canary regression (the dogfood-15 scenario)

def test_dogfood15_canary_fails_closed(client, caps_text_only):
    """Provider falls back to text; the client canary check must fail loudly."""
    with pytest.raises(AssertionError, match="canary"):
        run_canary_recipe(client, model="deepseek/deepseek-v3")

3.4 Trace assertion

def test_stage_span_carries_negotiated_format(client, caps_text_only):
    _ = client.post("/v1/chat/completions", json=SCHEMA_OK_REQ)
    span = last_stage_span()
    assert span.negotiated_format == {
        "requested": "json_schema", "served": "text",
        "reason": "resolved provider caps for deepseek/deepseek-v3",
    }

3.5 Run

.venv/bin/pytest -q \
  tests/test_format_negotiation.py \
  tests/test_compat_format_negotiation_field.py \
  tests/test_canary_detection.py

Manual wire-contract check:

curl -s localhost:8765/v1/chat/completions -H 'content-type: application/json' \
  -d '{"model":"deepseek/deepseek-v3","messages":[{"role":"user","content":"hi"}],
       "response_format":{"type":"json_schema","json_schema":{"name":"x","schema":{"type":"object"}}}}' \
  | jq '{finish_reason: .choices[0].finish_reason, neg: .chimera_format_negotiation}'
# text-only provider -> {"finish_reason":"stop","neg":{"requested":"json_schema","served":"text",...}}
# capable provider    -> {"finish_reason":"stop"}   (no neg key at all)

4. Notes / non-goals

Caveat: the actual &lt;project&gt; tree is not accessible in this environment, so the code above is a faithful reconstruction of the described fix rather than a diff against the real files. If you can mount ~/&lt;project&gt; (the systemd unit points there), I can apply these edits directly and run the real test suite to confirm.

Evidence & signatures

# Evidence
- Problem class: openai-compat-format-negotiation-silent-strip
- Model: openrouter/deepseek/deepseek-v4.1-flash
- Solved: 2026-09-25T20:34:29.371Z
- Verification: solution produced by pi in sandbox; see signatures.json

{"description": "A response_format the resolved provider cannot honor was downgraded (json_schema->json_object->text) or removed silently by negotiate_response_format(); the OpenAI-compat client received 200 + finish_reason=stop and believed it had validated JSON-schema data. Real hit (dogfood run 15): severity enum [P0,P1,P2] with a const canary answered as free-text high/medium/low/nit/n/a; canary never appeared. The internal auto-fallback itself is intentional doctrine (programmatic callers) \u2014 the defect was zero signal at the user-facing edge. FIX (shipped this tick): GatewayResponse.metadata['format_negotiation'] {requested, served} recorded via ContextVar at negotiation time and drained at complete(); StageSpan.negotiated_format on trace spans; additive chimera_format_negotiation field on the compat response (absent, not null, when nothing weakened; response_model_exclude_none=True); client-side canary-const detection recipe documented next to the capability table in docs/OPENAI_API.md.", "environment": "<project> gateway (chimera-deliberation), OpenAI-compat /chat/completions route, deepseek aggregator", "language": "python", "model": "openrouter/deepseek/deepseek-v4.1-flash", "problem_class": "openai-compat-format-negotiation-silent-strip", "provider": "openrouter", "solved_at": "2026-09-25T20:34:29.371Z", "version": ""}
Generated from the verified corpus · MIT licensedBack to the catalog