fix(http-bridge): retire bridge anchors upstream has denied

app/modules/proxy/_service/http_bridge/streaming.py

@@ -3447,6 +3447,18 @@ async def rollback_pre_dispatch_recovery_claim() -> None:
                 retry_request_state.operation_attempt_generation = request_state.operation_attempt_generation
                 retry_request_state.operation_persisted_response_id = request_state.operation_persisted_response_id
                 retry_request_state.operation_rebind_required = request_state.operation_rebind_required
+                # An anchored recovery replays the proxy's own anchor, so the
+                # retry inherits its provenance. Without this the retry looks
+                # client-anchored: diagnostics report
+                # ``previous_response_source=client_supplied`` for an id no
+                # client sent, ``_http_bridge_request_state_wedged_reattach``
+                # cannot recognise the reattach, and an upstream denial of the
+                # anchor is not attributable to this proxy. The anchor-free
+                # recovery paths carry no anchor at all, so the flag stays
+                # false there and cannot describe an id the retry never sends.
+                retry_request_state.proxy_injected_previous_response_id = (
+                    request_state.proxy_injected_previous_response_id and retry_previous_response_id is not None
+                )
                 if recovery_path == "local_previous_response_error":
                     # The prior response.failed/error made the operation
                     # terminal. Re-enter record_operation so its owner fence

app/modules/proxy/_service/http_bridge/upstream_events.py

@@ -1011,6 +1011,48 @@ async def _abandon_durable_http_bridge_continuity(
     return True
 
 
+async def _invalidate_denied_http_bridge_anchor(
+    service: Any,
+    session: "_HTTPBridgeSession",
+    *,
+    denied_response_id: str | None,
+) -> bool:
+    """Retire an anchor upstream has explicitly denied.
+
+    ``previous_response_not_found`` against an anchor the proxy injected is a
+    verdict, not a symptom: the id came from this proxy's own durable record,
+    no client asked for it, and upstream says it does not exist. The poison
+    counter cannot act on that verdict because it only scores reader failures
+    (see ``_HTTP_BRIDGE_ANCHOR_POISON_DETAILS``), so the dead id survives at
+    any threshold and is re-injected into the following turn, where the
+    store-context trim strips the resent history against it and upstream never
+    emits ``response.created``.
+
+    Clearing costs nothing that is not already lost. The next turn simply
+    dispatches unanchored with the history the client sends, which is the
+    client's own replay rather than a server-side one, so no forked child
+    response can be created against a parent this proxy cannot see.
+    """
+    if denied_response_id is None:
+        return False
+    # Another request may have completed and advanced the anchor between the
+    # denied dispatch and this frame. Only retire the id that was refused.
+    if session.last_completed_response_id != denied_response_id:
+        return False
+    cleared = await _abandon_durable_http_bridge_continuity(
+        service,
+        session,
+        detail="upstream_denied_proxy_injected_anchor",
+    )
+    await service._unregister_http_bridge_previous_response_ids(session)
+    session.last_completed_response_id = None
+    session.last_completed_response_account_id = None
+    session.last_completed_input_count = 0
+    session.last_completed_input_prefix_fingerprint = None
+    session.last_pending_tool_calls.clear()
+    return cleared
+
+
 class _HTTPBridgeUpstreamEventsMixin:
     async def _fail_http_bridge_reader_and_maybe_retire(
         self: Any,
@@ -2289,6 +2331,12 @@ async def persist_grouped_terminal_events() -> Exception | None:
                 original_text=text,
             )
             event_block = f"data: {rewritten_text}\n\n"
+            if status_request_state.proxy_injected_previous_response_id:
+                await _invalidate_denied_http_bridge_anchor(
+                    self,
+                    session,
+                    denied_response_id=status_request_state.previous_response_id,
+                )
 
         retry_error_code = _websocket_precreated_retry_error_code(
             status_request_state,

openspec/changes/invalidate-denied-bridge-anchor/.openspec.yaml

@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-08-22

openspec/changes/invalidate-denied-bridge-anchor/proposal.md

@@ -0,0 +1,41 @@
+# Invalidate Bridge Anchors Upstream Has Denied
+
+## Why
+
+The HTTP responses session bridge keeps re-injecting a `previous_response_id` that upstream has already said does not exist (issue #1852).
+
+When upstream answers an anchored bridge request with `previous_response_not_found`, that is a verdict about the anchor. If the proxy injected the anchor itself, no client asked for the id, it came from the proxy's own durable record, and upstream has now refused it. Nothing in the bridge acts on that verdict:
+
+1. Anchor poisoning cannot see it. `_http_bridge_anchor_poison_detail` only scores reader failures whose detail is `stream_incomplete` or `stream_idle_timeout`. A denial arrives as a terminal upstream event, not a reader failure, so it contributes nothing at any value of `http_responses_session_bridge_anchor_poison_failure_threshold`. Lowering the threshold does not help.
+2. The dead id therefore survives in both carriers, the durable `latest_response_id` row and the in-memory `session.last_completed_response_id`, and the fresh-reattach path injects it into the next turn.
+3. On that next turn the store-context trim matches the stored prefix and strips it, because the trim consults the stored fingerprint and never whether the anchor is still alive. Upstream then receives a few items instead of the conversation, never emits `response.created`, and the attempt presents as an eventless failure rather than as a stale anchor.
+
+Two of those eventless failures open the retry circuit, so the client sees `503 ... cooling down`. Measured over 12.7 h on one host: 163 `continuity_fail_closed` rejections, 29 circuit opens, and a worst-case trim of `original_items=602 trimmed_to=3`.
+
+The second half of this change is why the first half currently could not fire even if it existed. The anchored recovery replays the proxy's own anchor but never copies `proxy_injected_previous_response_id` onto the retry state, so a denial of the replayed anchor is not attributable to the proxy. The same gap misreports `previous_response_source=client_supplied` for ids no client sent and keeps `_http_bridge_request_state_wedged_reattach` from recognising the reattach shape it exists to catch.
+
+## What Changes
+
+- Retire a `previous_response_id` on the first explicit upstream denial when the proxy injected it, clearing the durable continuity row and the in-memory anchor together, instead of waiting for a counter that this failure class never increments. The retirement is skipped when a concurrent request has already advanced the anchor past the denied id.
+- Carry `proxy_injected_previous_response_id` onto the anchored recovery retry state, so a denial of the replayed anchor is attributable, diagnostics report the real provenance, and the wedge classifier sees the reattach. Anchor-free recovery paths keep the flag false because they send no anchor.
+
+## Capabilities
+
+### New Capabilities
+
+None.
+
+### Modified Capabilities
+
+- `responses-api-compat`: An explicit upstream previous-response denial retires a proxy-injected anchor immediately, and anchored recovery retries retain the provenance of the anchor they replay.
+
+## Impact
+
+- HTTP bridge terminal-event handling (`app/modules/proxy/_service/http_bridge/upstream_events.py`) and anchored recovery retry state (`app/modules/proxy/_service/http_bridge/streaming.py`).
+- No API, schema, migration, dependency, configuration, or dashboard changes. The poison threshold setting and its default of seven are untouched, and the downstream error contract is unchanged: the denial is still masked to `stream_incomplete` and still surfaces as 502, so clients keep their anchor and do not resend full history (the invariant from #397).
+
+## Non-Goals
+
+- Adding another upstream dispatch. This change never resends the turn. The next turn is the client's own, with the history the client sends, so no forked child response can be created against a parent the proxy cannot observe. Retrying the turn server-side without the anchor is what #1857 and #1863 propose and is deliberately out of scope here.
+- Changing the poison threshold arithmetic that issue #1852 is titled after.
+- Exposing the stale-anchor classifier downstream on the bridge path.

openspec/changes/invalidate-denied-bridge-anchor/specs/responses-api-compat/spec.md

@@ -0,0 +1,54 @@
+# responses-api-compat Delta
+
+## ADDED Requirements
+
+### Requirement: Explicit upstream previous-response denials retire proxy-injected anchors
+
+When upstream answers an HTTP bridge request with a `previous_response_not_found` terminal frame, and the `previous_response_id` on that request was injected by the proxy rather than supplied by the client, the proxy MUST retire that anchor on the first denial rather than waiting for the eventless-failure poison threshold. Retirement MUST clear the durable continuity record under the session's owner epoch and the in-memory session anchor together, so no later turn can re-inject the denied id from either carrier. The proxy MUST NOT retire the anchor when the session's current anchor is no longer the denied id, because a concurrent request may have completed and advanced it. A client-supplied `previous_response_id` MUST NOT be retired by this path.
+
+The downstream error contract is unchanged: the denial is still reported to the client as `stream_incomplete`, so the client retains its own anchor and is not driven into a full-history resend.
+
+#### Scenario: A denied proxy-injected anchor is retired immediately
+
+- **GIVEN** an HTTP bridge session whose stored anchor was injected by the proxy
+- **WHEN** upstream answers the anchored request with `previous_response_not_found`
+- **THEN** the proxy clears the durable continuity record under the session's owner epoch
+- **AND** clears the in-memory session anchor and its stored input count and prefix fingerprint
+- **AND** the next turn on that session dispatches without a `previous_response_id`
+
+#### Scenario: The following turn is not trimmed against a denied anchor
+
+- **GIVEN** a proxy-injected anchor was denied by upstream on the previous turn
+- **WHEN** the client sends a full resend of the conversation on the next turn
+- **THEN** the request MUST NOT be trimmed against the denied anchor's stored prefix
+- **AND** upstream receives the resent conversation rather than a suffix of it
+
+#### Scenario: A concurrent completion protects the current anchor
+
+- **GIVEN** a proxy-injected anchor is denied by upstream
+- **AND** another request on the same session completed first and advanced the session anchor to a different response id
+- **WHEN** the denial is handled
+- **THEN** the proxy MUST NOT clear the session anchor
+
+#### Scenario: Client-supplied anchors are left alone
+
+- **GIVEN** an HTTP bridge request carries a `previous_response_id` the client supplied
+- **WHEN** upstream answers it with `previous_response_not_found`
+- **THEN** the proxy MUST NOT retire the anchor on the client's behalf
+
+### Requirement: Anchored recovery retries retain the provenance of the anchor they replay
+
+When the HTTP bridge dispatches an anchored recovery retry that replays a `previous_response_id` the proxy injected, the retry request state MUST record that the anchor is proxy-injected. A recovery path that dispatches without an anchor MUST leave that provenance false, because there is no anchor for it to describe.
+
+#### Scenario: An anchored recovery retry is attributable to the proxy
+
+- **GIVEN** a request whose `previous_response_id` was injected by the proxy fails and enters anchored recovery
+- **WHEN** the recovery retry replays the same anchor
+- **THEN** the retry request state records the anchor as proxy-injected
+- **AND** continuity diagnostics for the retry report `previous_response_source=proxy_injected` rather than `client_supplied`
+
+#### Scenario: Anchor-free recovery retries claim no provenance
+
+- **GIVEN** a recovery path dispatches without a `previous_response_id`
+- **WHEN** the retry request state is prepared
+- **THEN** it MUST NOT record a proxy-injected anchor

openspec/changes/invalidate-denied-bridge-anchor/tasks.md

@@ -0,0 +1,21 @@
+# Tasks
+
+## 1. Regression Coverage
+
+- [x] 1.1 Add a bridge integration regression driving a completed turn, an anchored turn denied with `previous_response_not_found`, then a following client full resend, asserting the third dispatch carries no `previous_response_id` and is not trimmed against the denied anchor.
+- [x] 1.2 Add unit coverage that a denial retires both anchor carriers on the first occurrence, and that it is skipped when a concurrent completion has already advanced the anchor.
+- [x] 1.3 Assert at the product path that no continuity diagnostic reports a proxy-injected anchor as `client_supplied`, which fails before the provenance fix.
+
+## 2. Anchor Retirement
+
+- [x] 2.1 Add `_invalidate_denied_http_bridge_anchor`, clearing durable continuity through the existing fenced `_abandon_durable_http_bridge_continuity` write and the in-memory anchor fields together.
+- [x] 2.2 Call it from the terminal `previous_response_not_found` branch when the denied anchor was proxy-injected.
+
+## 3. Recovery Provenance
+
+- [x] 3.1 Copy `proxy_injected_previous_response_id` onto the anchored recovery retry state, gated on the retry actually carrying an anchor.
+
+## 4. Verification
+
+- [x] 4.1 Run the touched bridge unit and integration suites, ruff, and type checks.
+- [x] 4.2 Run strict OpenSpec validation for this change and review the final diff for unrelated changes.

tests/integration/test_http_responses_bridge.py

@@ -4,6 +4,7 @@
 import base64
 import contextlib
 import json
+import logging
 import socket
 import time
 from collections import deque
@@ -15556,3 +15557,155 @@ async def captive_release_live_session(**kwargs):
     # on the same key keeps working instead of failing with 409.
     third = await asyncio.wait_for(async_client.post("/v1/responses", json=payload), timeout=_TEST_SYNC_TIMEOUT_SECONDS)
     assert third.status_code == 200, third.text
+
+
+class _DeniesAnchoredTurnUpstreamWebSocket(_FakeBridgeUpstreamWebSocket):
+    """Completes unanchored turns and denies any turn that arrives with an anchor."""
+
+    async def send_text(self, text: str) -> None:
+        payload = json.loads(text)
+        previous_response_id = payload.get("previous_response_id")
+        if previous_response_id is None:
+            await super().send_text(text)
+            return
+        self.sent_text.append(text)
+        await self._messages.put(
+            _FakeUpstreamMessage(
+                "text",
+                text=json.dumps(
+                    {
+                        "type": "error",
+                        "status": 400,
+                        "error": {
+                            "type": "invalid_request_error",
+                            "code": "previous_response_not_found",
+                            "message": f"Previous response with id '{previous_response_id}' not found.",
+                            "param": "previous_response_id",
+                        },
+                    },
+                    separators=(",", ":"),
+                ),
+            )
+        )
+
+
+@pytest.mark.asyncio
+async def test_v1_responses_http_bridge_stops_reinjecting_an_anchor_upstream_denied(
+    async_client,
+    app_instance,
+    monkeypatch,
+    caplog,
+):
+    """A denied proxy-injected anchor must not be re-injected into the next turn.
+
+    Regression for the amplification in issue #1852: the denial leaves the dead
+    anchor in the session, the next full resend is trimmed against its stored
+    prefix, and upstream then receives a suffix of the conversation behind an id
+    it has already refused.
+    """
+    _install_bridge_settings(monkeypatch, enabled=True)
+    account_id = await _import_account(
+        async_client,
+        "acc_http_bridge_denied_anchor",
+        "http-bridge-denied-anchor@example.com",
+    )
+    account = await _get_account(account_id)
+    upstream = _DeniesAnchoredTurnUpstreamWebSocket()
+
+    async def fake_select_account_with_budget(
+        self,
+        deadline,
+        *,
+        request_id,
+        kind,
+        request_stage="first_turn",
+        sticky_key,
+        sticky_kind,
+        reallocate_sticky,
+        sticky_max_age_seconds,
+        prefer_earlier_reset_accounts,
+        routing_strategy,
+        model,
+        exclude_account_ids=None,
+        additional_limit_name=None,
+        api_key=None,
+        preferred_account_id=None,
+    ):
+        del self, deadline, request_id, kind, request_stage, sticky_key, sticky_kind
+        del reallocate_sticky, sticky_max_age_seconds, prefer_earlier_reset_accounts
+        del routing_strategy, model, exclude_account_ids, additional_limit_name
+        del api_key, preferred_account_id
+        return AccountSelection(account=account, error_message=None, error_code=None)
+
+    async def fake_ensure_fresh_with_budget(self, target, *, force=False, timeout_seconds):
+        del self, force, timeout_seconds
+        return target
+
+    async def fake_connect_responses_websocket(
+        headers,
+        access_token,
+        account_id_header,
+        *,
+        base_url=None,
+        session=None,
+    ):
+        del headers, access_token, account_id_header, base_url, session
+        return upstream
+
+    monkeypatch.setattr(proxy_module.ProxyService, "_select_account_with_budget", fake_select_account_with_budget)
+    monkeypatch.setattr(proxy_module.ProxyService, "_ensure_fresh_with_budget", fake_ensure_fresh_with_budget)
+    monkeypatch.setattr(proxy_module, "connect_responses_websocket", fake_connect_responses_websocket)
+
+    headers = {"session_id": "http-bridge-denied-anchor-session"}
+
+    def _user_item(text: str) -> dict[str, Any]:
+        return {"role": "user", "content": [{"type": "input_text", "text": text}]}
+
+    turn_one_input = [_user_item("turn one")]
+    turn_two_input = [*turn_one_input, _user_item("turn two")]
+    turn_three_input = [*turn_two_input, _user_item("turn three")]
+
+    caplog.set_level(logging.WARNING, logger="app.modules.proxy.service")
+
+    first = await async_client.post(
+        "/backend-api/codex/responses",
+        json={"model": "gpt-5.1", "instructions": "Return exactly OK.", "input": turn_one_input},
+        headers=headers,
+    )
+    assert first.status_code == 200
+
+    second = await async_client.post(
+        "/backend-api/codex/responses",
+        json={"model": "gpt-5.1", "instructions": "Return exactly OK.", "input": turn_two_input},
+        headers=headers,
+    )
+    assert second.status_code in {200, 502}
+
+    third = await async_client.post(
+        "/backend-api/codex/responses",
+        json={"model": "gpt-5.1", "instructions": "Return exactly OK.", "input": turn_three_input},
+        headers=headers,
+    )
+    assert third.status_code == 200
+
+    dispatched = [json.loads(text) for text in upstream.sent_text]
+    anchored = [frame for frame in dispatched if frame.get("previous_response_id") is not None]
+    assert anchored, "expected the proxy to inject an anchor on the second turn"
+
+    final = dispatched[-1]
+    assert final.get("previous_response_id") is None, (
+        f"the denied anchor was re-injected into a later turn: {final.get('previous_response_id')}"
+    )
+    assert len(final["input"]) == len(turn_three_input), (
+        "the later turn was trimmed against the denied anchor's stored prefix"
+    )
+
+    # No client supplied an anchor in this test, so every anchor the diagnostics
+    # describe must be attributed to the proxy that injected it.
+    continuity_diagnostics = [
+        record.getMessage() for record in caplog.records if "continuity_fail_closed" in record.getMessage()
+    ]
+    assert continuity_diagnostics, "expected the denial to be recorded"
+    assert not [line for line in continuity_diagnostics if "previous_response_source=client_supplied" in line], (
+        "an anchored recovery retry reported a proxy-injected anchor as client-supplied"
+    )