fix(gateway): settle a queued steer by its admission, not by searching its text

Settlement withdrew a transfer by finding its text between newlines in the
agent's combined steer buffer. Those newlines also occur inside another
submission, so withdrawing a consumed 'alpha' stripped it out of a later
'alpha\nbeta': the consumed prompt was queued again and ran twice, the unread
one was recorded steered, and 'beta' leaked into a later turn. Identical text
on both sides of a drain had the same swap.

The agent's steer buffer now keeps each submission as a (ticket, text) part,
with _pending_steer still their newline join. The gateway hands a transfer
over with its admission id as the ticket. A drain records each ticket it took
as consumed, a hard stop records it as dropped unread, and a drained text put
back undelivered (_requeue_pending_steer) gets its tickets back.
withdraw_steer(ticket) settles exactly that submission: True while it is still
buffered (only its part is removed) or was dropped unread, False once a drain
consumed it.

Refs #134889 (RT-35)
Refs #140208

agent/agent_runtime_helpers.py

@@ -3703,19 +3703,8 @@ def extract_api_error_context(error: Exception) -> dict[str, Any]:
 
 def _requeue_pending_steer(agent, steer_text: str) -> None:
     """Put drained steer text back so the caller's fallback delivers it as a next-turn user message."""
-    # Under the lock the slot is read directly: an initialized agent always has both attributes, so a
-    # missing ``_pending_steer`` there is a real bug and must fail loud. The lock-less branch only
-    # exists for test stubs built via ``object.__new__`` that skipped ``__init__``.
-    _lock = getattr(agent, "_pending_steer_lock", None)
-    if _lock is not None:
-        with _lock:
-            if agent._pending_steer:
-                agent._pending_steer = agent._pending_steer + "\n" + steer_text
-            else:
-                agent._pending_steer = steer_text
-    else:
-        existing = getattr(agent, "_pending_steer", None)
-        agent._pending_steer = (existing + "\n" + steer_text) if existing else steer_text
+    from agent.interrupt_control import requeue_drained_steer
+    requeue_drained_steer(agent, steer_text)
 
 
 def apply_pending_steer_to_tool_results(agent, messages: list, num_tool_msgs: int) -> None:

agent/interrupt_control.py

@@ -85,6 +85,39 @@ def _ic_slot(agent, lock_attr: str, slot: str):
     return getattr(agent, slot)
 
 
+def _steer_parts(agent) -> list:
+    """The buffered steers as ``(ticket, text)`` pairs; ``_pending_steer`` is their newline join. A ticket
+    names one submission (a gateway transfer). A buffer another writer set directly is one ticketless part."""
+    pending = _ic_slot(agent, "_pending_steer_lock", "_pending_steer")
+    parts = vars(agent).get("_pending_steer_parts") or []
+    if "\n".join(text for _, text in parts) != (pending or ""):
+        return [(None, pending)] if pending else []
+    return list(parts)
+
+
+def _set_steer_parts(agent, parts: list) -> None:
+    agent._pending_steer_parts = parts
+    agent._pending_steer = "\n".join(text for _, text in parts) or None
+
+
+def _steer_fates(agent) -> dict:
+    """``{ticket: consumed}`` for ticketed steers that left the buffer: True when a drain handed the text
+    to the turn, False when it was dropped unread. ``withdraw_steer`` reads each one once."""
+    return vars(agent).setdefault("_steer_ticket_fates", {})
+
+
+def requeue_drained_steer(agent, steer_text: str) -> None:
+    """Put drained steer text back (nothing took it). When it is the last drain's text, its parts go back
+    with their tickets, so a transfer in it reads unconsumed again."""
+    with _ic_lock(agent, "_pending_steer_lock"):
+        drained, parts = vars(agent).pop("_steer_last_drain", (None, None))
+        if drained != steer_text or not parts:
+            parts = [(None, steer_text)]
+        for ticket, _ in parts:
+            _steer_fates(agent).pop(ticket, None)
+        _set_steer_parts(agent, _steer_parts(agent) + parts)
+
+
 def _ic_codex_method(agent, name: str):
     """Codex app-server owns its model/tool loop; return its ``name`` hook or None."""
     if getattr(agent, "api_mode", None) != "codex_app_server":
@@ -261,18 +294,21 @@ class InterruptControlMixin:
             # claimed a hard interrupt supersedes the steer — but nothing gated this wipe on
             # hard_cancel, so a soft clear dropped a live user message with no trace.
             with _ic_lock(self, "_pending_steer_lock"):
-                self._pending_steer = None
+                for ticket, _ in _steer_parts(self):
+                    if ticket is not None:
+                        _steer_fates(self)[ticket] = False
+                _set_steer_parts(self, [])
         return True
 
-    def steer(self, text: str) -> bool:
+    def steer(self, text: str, *, ticket=None) -> bool:
         """Queue user text for delivery as its own user row after the current tool batch finishes (no
-        interrupt); multiple calls concatenate with newlines. Returns False for empty text."""
+        interrupt); multiple calls concatenate with newlines. Returns False for empty text. ``ticket``
+        names the submission, so ``withdraw_steer`` can settle exactly it, whatever text others carry."""
         if not text or not text.strip():
             return False
         cleaned = text.strip()
         with _ic_lock(self, "_pending_steer_lock"):
-            existing = _ic_slot(self, "_pending_steer_lock", "_pending_steer")
-            self._pending_steer = (existing + "\n" + cleaned) if existing else cleaned
+            _set_steer_parts(self, _steer_parts(self) + [(ticket, cleaned)])
         # The turn no longer ran on its own (first-task-done detection reads this off the result).
         self._turn_user_intervened = True
         return True
@@ -356,19 +392,25 @@ class InterruptControlMixin:
         """Return the pending steer text (if any) and clear the slot; None when nothing is pending."""
         with _ic_lock(self, "_pending_steer_lock"):
             text = _ic_slot(self, "_pending_steer_lock", "_pending_steer")
-            self._pending_steer = None
+            parts = _steer_parts(self)
+            _set_steer_parts(self, [])
+            for ticket, _ in parts:
+                if ticket is not None:
+                    _steer_fates(self)[ticket] = True
+            self._steer_last_drain = (text, parts)  # requeue_drained_steer restores its tickets
         return text
 
-    def withdraw_steer(self, text: str) -> bool:
-        """Take one earlier ``steer(text)`` back out of the buffer while no drain has consumed it (the
-        gateway returns a queued prompt's unconsumed steer to its queue). Other buffered steers stay."""
-        cleaned = (text or "").strip()
+    def withdraw_steer(self, ticket) -> bool:
+        """Settle one ``steer(text, ticket=...)`` by identity (the gateway's transfer of a queued prompt):
+        True when no drain handed it to the turn, either still buffered (taken out now; every other steer
+        stays, whatever its text) or dropped unread; False once a drain consumed it, or for an unknown one."""
+        if ticket is None:
+            return False
         with _ic_lock(self, "_pending_steer_lock"):
-            pending = _ic_slot(self, "_pending_steer_lock", "_pending_steer")
-            framed = "\n" + (pending or "") + "\n"
-            index = framed.find("\n" + cleaned + "\n") if cleaned else -1
-            if index < 0:
-                return False
-            rest = (framed[:index] + framed[index + 1 + len(cleaned):]).strip("\n")
-            self._pending_steer = rest or None
+            parts = _steer_parts(self)
+            kept = [part for part in parts if part[0] != ticket]
+            consumed = _steer_fates(self).pop(ticket, True)
+            if len(kept) == len(parts):
+                return consumed is False
+            _set_steer_parts(self, kept)
         return True

gateway/session_queued_controls.py

@@ -82,21 +82,21 @@ async def _queued(connection, ref, admission_id):
 def steer_transfers(authority, session_id, generation):
     """``{admission_id: delivered}`` handed to this claim; the caller holds the stream lock. Read by
     the claim's settlement after its execution returned, when no transfer can still be delivered.
-    A text its agent still buffers (accepted after the turn's last drain) never reached the claim: it
-    is withdrawn and reads undelivered, so its admission goes back to its queue place."""
+    The agent settles each one by its ticket (the admission id), never by its text: one no drain of
+    the turn consumed (still buffered, or dropped unread) reads undelivered and goes back to its queue."""
     entry = authority.steer_transfers.get(session_id)
     if entry is None or entry[0] != generation:
         return None
     adopted_generation, agent = authority.adopted.get(session_id, (None, None))
     withdraw = getattr(agent, 'withdraw_steer', None) if adopted_generation == generation else None
     for admission_id, delivered in entry[1].items():
-        if delivered and callable(withdraw) and withdraw(delivered):
+        if delivered and callable(withdraw) and withdraw(admission_id):
             entry[1][admission_id] = False
-    return {admission_id: bool(delivered) for admission_id, delivered in entry[1].items()}
+    return dict(entry[1])
 
 
 def _register(authority, session_id, generation, admission_id):
-    """``{admission_id: False | accepted text}`` for the claim ``generation``."""
+    """``{admission_id: delivered}`` for the claim ``generation``."""
     entry = authority.steer_transfers.get(session_id)
     if entry is None or entry[0] != generation:
         entry = authority.steer_transfers[session_id] = (generation, {})
@@ -127,9 +127,10 @@ def _deliver(authority, session_id, generation, agent, admission_id, text):
             or _receiving_agent(authority, session_id, generation) is not agent):
         return False
     if not entry[1][admission_id]:
-        # The accepted text itself: settlement withdraws it if no drain of the turn consumed it.
-        entry[1][admission_id] = text if agent.steer(text) else False
-    return bool(entry[1][admission_id])
+        # Ticketed by its admission id: settlement withdraws exactly it if no drain of the turn consumed it.
+        ticketed = callable(getattr(agent, 'withdraw_steer', None))
+        entry[1][admission_id] = bool(agent.steer(text, ticket=admission_id) if ticketed else agent.steer(text))
+    return entry[1][admission_id]
 
 
 async def steer_queued(connection, ref, params):

tests/agent/test_steer.py

@@ -80,22 +80,44 @@ class TestSteerDrain:
         assert agent._drain_pending_steer() == "hello"
         assert agent._pending_steer is None
 
-    def test_withdraw_takes_back_one_unconsumed_steer_and_keeps_the_rest(self):
+    def test_withdraw_takes_back_one_unconsumed_ticket_and_keeps_the_rest(self):
         agent = _bare_agent()
-        for text in ("first", "  two\nlines  ", "last"):
-            agent.steer(text)
-        assert agent.withdraw_steer("two\nlines") is True
-        assert agent._pending_steer == "first\nlast"
+        agent.steer("first")
+        agent.steer("  two\nlines  ", ticket="t2")
+        agent.steer("two")  # the same text as a line of t2, without a ticket
+        agent.steer("last", ticket="t4")
+        assert agent.withdraw_steer("t2") is True
+        assert agent._pending_steer == "first\ntwo\nlast"
         assert agent.withdraw_steer("missing") is False
-        assert agent.withdraw_steer("first") is True
-        assert agent.withdraw_steer("last") is True
+        assert agent.withdraw_steer("t4") is True
+        assert agent._pending_steer == "first\ntwo"
+
+    def test_withdraw_after_a_drain_took_the_ticket_is_refused_even_if_its_text_is_buffered_again(self):
+        agent = _bare_agent()
+        agent.steer("alpha", ticket="a")
+        assert agent._drain_pending_steer() == "alpha"
+        agent.steer("alpha", ticket="b")
+        assert agent.withdraw_steer("a") is False, "a drain consumed a"
+        assert agent.withdraw_steer("b") is True
         assert agent._pending_steer is None
 
-    def test_withdraw_after_a_drain_took_the_text_is_refused(self):
+    def test_a_drained_steer_put_back_undelivered_keeps_its_ticket(self):
+        from agent.agent_runtime_helpers import _requeue_pending_steer
         agent = _bare_agent()
-        agent.steer("hello")
-        assert agent._drain_pending_steer() == "hello"
-        assert agent.withdraw_steer("hello") is False
+        agent.steer("note")
+        agent.steer("queued prompt", ticket="t")
+        drained = agent._drain_pending_steer() or ""
+        _requeue_pending_steer(agent, drained)  # no tool result took it
+        assert agent._pending_steer == "note\nqueued prompt"
+        assert agent.withdraw_steer("t") is True
+        assert agent._pending_steer == "note"
+
+    def test_a_hard_stop_drops_a_ticket_unread(self):
+        agent = _bare_agent()
+        agent.steer("queued prompt", ticket="t")
+        agent.clear_interrupt(hard_cancel=True)
+        assert agent._pending_steer is None
+        assert agent.withdraw_steer("t") is True, "dropped unread: never delivered"
 
 
 

tests/gateway/test_prompt_queued_transfer.py

@@ -130,6 +130,50 @@ async def test_a_steer_settles_steered_only_when_a_drain_of_the_turn_took_it(tmp
         assert ran == ['running', 'follow-up'], 'the returned prompt runs once, as its own turn'
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize('later', ['alpha\nbeta', 'alpha'])
+async def test_each_transfer_settles_by_its_own_identity_whatever_text_another_carries(tmp_path, monkeypatch, later):
+    """Reviewer repro (round 3): A ('alpha') is consumed by a drain of the turn; B arrives after that last
+    drain carrying A's text inside a multiline prompt, or exactly A's text. A is steered and never runs
+    again; B, which the turn never read, goes back to its queue place and runs once as its own turn."""
+    db, authority = _authority(tmp_path, monkeypatch)
+    interactive(db)
+    agent, ran, taken = BufferedAgent(), [], []
+    claimed, drain, drained, release = (asyncio.Event() for _ in range(4))
+
+    async def execute(authority, ref, row):
+        ran.append(row['payload']['text'])
+        authority.adopt_agent(ref.session_id, row['generation'], agent)
+        if len(ran) > 1:
+            return 'done'
+        claimed.set()
+        await asyncio.wait_for(drain.wait(), 30)
+        taken.append(agent._drain_pending_steer())  # a drain of the turn consumes A
+        drained.set()
+        await asyncio.wait_for(release.wait(), 30)
+        return 'done'  # B was accepted after that last drain
+    monkeypatch.setattr(session_finite, 'execute_finite_admission', execute)
+    with db:
+        await _submit(authority, 'running')
+        await asyncio.wait_for(claimed.wait(), 30)
+        conn = _connection(authority)
+        a = await _submit(authority, 'a', 'alpha')
+        assert 'error' not in await _call(conn, 'prompt.steer_queued', admission_id=a.admission_id)
+        drain.set()
+        await asyncio.wait_for(drained.wait(), 30)
+        b = await _submit(authority, 'b', later)
+        assert 'error' not in await _call(conn, 'prompt.steer_queued', admission_id=b.admission_id)
+        release.set()
+        await _drained(authority)
+        await async_wait_until(lambda: len(ran) == 2, timeout=30, what='the returned admission ran')
+        await _drained(authority)
+    assert taken == ['alpha']
+    assert ran == ['running', later], 'the consumed prompt never runs again; the unread one runs once'
+    assert _ledger(db) == [('running', 'terminal', 'completed'), ('a', 'terminal', 'steered'),
+                           ('b', 'terminal', 'completed')]
+    assert agent._pending_steer is None
+
+
 @pytest.mark.asyncio
 async def test_a_transfer_stays_listed_as_steering_until_its_claim_settles(tmp_path, monkeypatch):
     """Reviewer repro (returned prompt stays hidden): a pending snapshot taken while a steer is in