fix: signed reasoning replays after resume on Anthropic and Bedrock Converse

What users saw
- Anthropic (last-turn-only Claude models, e.g. Sonnet 4.5): after `chat --resume`
  or on the next gateway/TUI turn, the previous answer's signed thinking block was
  stripped, so its reasoning was lost.
- Bedrock Converse (non-Claude reasoning models, e.g. deepseek.v3): every resumed
  or later turn sent prior reasoning without its signature, and Bedrock rejected it
  with ValidationException "The reasoning block signature is missing or invalid".
  Streamed reasoning was also stored as two concatenated copies.

Root causes
- 33d6139ccd replaced "keep the latest assistant turn" with "keep the in-flight
  tool loop". Once a new user message closes that loop, the latest turn lands
  before loop_start and its signed thinking is dropped.
- d6f3199584 persisted Converse reasoning into reasoning_details, so the reasoning
  extractor now also read it and duplicated the streamed text. On reload, non-Claude
  turns replayed that text with the signature removed, because they could not tell
  their own signatures apart from a foreign model's.

Fix
- anthropic_message_convert: loop_start never goes past the latest assistant turn.
- bedrock_adapter + transports/bedrock: Converse turns record the model that produced them (a private
  `bedrock.native_assistant` reasoning_details record). When the same model reads
  them back, they replay verbatim with signatures. Other models keep the per-model
  policy. reasoning_content is now joined per block, not per streamed delta.
- test_chat_reasoning_variants: the long-session Portal test asserted the old
  "never replay reasoning_details to the Portal" contract that 336b11584c
  removed on purpose. It now asserts every earlier turn's blocks replay verbatim
  and every turn is answered.

agent/anthropic_message_convert.py

@@ -596,7 +596,9 @@ def _manage_thinking_signatures(result: list[dict[str, Any]], base_url: str | No
     replays as-is; DeepSeek needs unsigned blocks round-tripped but rejects signed ones.
     """
     route = anthropic_thinking_route(base_url, model)
-    loop_start = _current_tool_loop_start(result)
+    # The latest assistant turn always keeps its thinking, even once a new user message closes its loop.
+    loop_start = min(_current_tool_loop_start(result),
+                     next((i for i in range(len(result) - 1, -1, -1) if result[i].get("role") == "assistant"), len(result)))
     preserve_prior = route_replays_prior_thinking(route, model)
     for idx, m in _assistant_block_lists(result):
         if route == "kimi":

agent/bedrock_adapter.py

@@ -809,7 +809,7 @@ def _anthropic_ordered_blocks(msg: dict) -> list[dict]:
     return [block for block in converted if block is not None]
 
 
-def _reasoning_details_blocks(msg: dict, claude: bool) -> list[dict]:
+def _reasoning_details_blocks(msg: dict, claude: bool, own: bool = False) -> list[dict]:
     """Thinking from the persisted ``reasoning_details`` (Anthropic shape, as Converse captures it), in stored
     order: Claude gets signed thinking + redacted blobs, other models readable text only (falling back to
     ``reasoning_content``). This is the carrier a turn reloaded from state.db replays from, since the
@@ -819,7 +819,7 @@ def _reasoning_details_blocks(msg: dict, claude: bool) -> list[dict]:
         if not isinstance(d, dict):
             continue
         if d.get("type") == "thinking" and isinstance(d.get("thinking"), str) and d["thinking"].strip():
-            if claude and d.get("signature"):
+            if (claude or own) and d.get("signature"):
                 blocks.append(_reasoning_text_block(d["thinking"], d["signature"]))
             elif not claude:
                 blocks.append(_reasoning_text_block(d["thinking"], None))
@@ -832,7 +832,7 @@ def _reasoning_details_blocks(msg: dict, claude: bool) -> list[dict]:
     return blocks
 
 
-def _assistant_blocks(msg: dict, content, claude: bool = True) -> list[dict]:
+def _assistant_blocks(msg: dict, content, claude: bool = True, own: bool = False) -> list[dict]:
     """Assistant message → Converse blocks, from the first carrier that yields any: the ordered
     ``bedrock_content_blocks`` sidecar, (Claude only) the ordered ``anthropic_content_blocks`` sidecar,
     else thinking from ``reasoning_details`` followed by text, then tool calls. Anthropic-shaped carriers
@@ -842,7 +842,7 @@ def _assistant_blocks(msg: dict, content, claude: bool = True) -> list[dict]:
         return content_blocks
     if claude and (content_blocks := _anthropic_ordered_blocks(msg)):
         return content_blocks
-    content_blocks = _reasoning_details_blocks(msg, claude)
+    content_blocks = _reasoning_details_blocks(msg, claude, own)
     if isinstance(content, str) and content.strip():
         content_blocks.append({"text": content})
     elif isinstance(content, list):
@@ -866,6 +866,17 @@ def _assistant_blocks(msg: dict, content, claude: bool = True) -> list[dict]:
 _CONVERSE_REPLAY_STRIP_PATTERNS = ("deepseek.r1", "deepseek-r1", "moonshotai.kimi-k3")
 
 
+# reasoning_details record naming the model that minted a turn's Converse reasoning: its signatures are that
+# model's own, so it gets them back unmodified on every later turn and after a reload (AWS: "include the text and
+# its signature unmodified"); any other model still gets the per-policy prior-turn treatment.
+CONVERSE_ISSUER_TYPE = "bedrock.native_assistant"
+
+
+def _issued_by(msg: dict, model: Optional[str]) -> bool:
+    return model is not None and any(isinstance(d, dict) and d.get("type") == CONVERSE_ISSUER_TYPE
+                                     and d.get("model") == model for d in msg.get("reasoning_details") or [])
+
+
 def _converse_replay_policy(model: Optional[str]) -> str:
     if model is None:
         return "signed"
@@ -934,9 +945,9 @@ def convert_messages_to_converse(
             append_turn("user", [{"toolResult": {
                 "toolUseId": msg.get("tool_call_id", ""), "content": [{"text": _safe_text(result_content)}]}}])
         elif role == "assistant":
-            blocks = _filter_reasoning(
-                _assistant_blocks(msg, content, claude=policy == "signed"), prior_filter if idx < loop_start else loop_filter
-            )
+            own = _issued_by(msg, model)
+            keep = loop_filter if idx >= loop_start or (own and policy == "unsigned") else prior_filter
+            blocks = _filter_reasoning(_assistant_blocks(msg, content, claude=policy == "signed", own=own), keep)
             append_turn("assistant", blocks or [dict(_PLACEHOLDER_BLOCK)])
         elif role == "user":
             append_turn("user", _convert_content_to_converse(content))
@@ -988,7 +999,6 @@ class _ResponseParts:
 
     def __init__(self) -> None:
         self.text_parts: list[str] = []
-        self.reasoning_parts: list[str] = []
         self.reasoning_details: list[dict[str, Any]] = []
         self.tool_calls: list[SimpleNamespace] = []
 
@@ -1001,7 +1011,6 @@ class _ResponseParts:
             reasoning = {**reasoning, **reasoning["reasoningText"]}
         thinking_text = reasoning.get("text", "")
         if thinking_text:
-            self.reasoning_parts.append(str(thinking_text))
             if on_text:
                 on_text(thinking_text)
             block["text"] = block.get("text", "") + str(thinking_text)
@@ -1021,7 +1030,8 @@ class _ResponseParts:
             role="assistant", content="\n".join(self.text_parts) if self.text_parts else None,
             tool_calls=self.tool_calls or None,
             reasoning_details=_persisted_reasoning_details(ordered_blocks) or self.reasoning_details or None,
-            reasoning_content="\n\n".join(self.reasoning_parts) if self.reasoning_parts else None,
+            reasoning_content="\n\n".join(texts) if (texts := [b["reasoningContent"]["text"] for b in ordered_blocks
+                if isinstance(b.get("reasoningContent"), dict) and b["reasoningContent"].get("text")]) else None,
             bedrock_content_blocks=ordered_blocks or None,
         )
         cache_read_tokens, cache_write_tokens, output_tokens = (

agent/transports/bedrock.py

@@ -40,6 +40,7 @@ class BedrockTransport(ProviderTransport):
         """Build Converse kwargs, leaving the optional output limit to the provider."""
         from agent.bedrock_adapter import build_converse_kwargs
 
+        self._model = model  # stamps the turns normalize_response captures as this model's own
         kwargs = build_converse_kwargs(
             model=model, messages=messages, tools=tools, max_tokens=params.get("max_tokens"),
             temperature=params.get("temperature"), guardrail_config=params.get("guardrail_config"),
@@ -65,6 +66,10 @@ class BedrockTransport(ProviderTransport):
         provider_data = {
             key: getattr(msg, key) for key in ("reasoning_details", "bedrock_content_blocks") if getattr(msg, key, None)
         }
+        if provider_data.get("reasoning_details") and getattr(self, "_model", None):
+            from agent.bedrock_adapter import CONVERSE_ISSUER_TYPE
+            provider_data["reasoning_details"] = [*provider_data["reasoning_details"],
+                                                  {"type": CONVERSE_ISSUER_TYPE, "model": self._model}]
         return NormalizedResponse(
             content=msg.content, tool_calls=tool_calls,
             finish_reason=choice.finish_reason or "stop",

tests/e2e/core/providers/test_chat_reasoning_variants.py

@@ -9,9 +9,8 @@
   the whole request), while state.db keeps it for a later switch back.
 * DeepSeek-style ``reasoning_content`` is echoed on the assistant tool-call message,
   within the turn and after ``--resume``.
-* A long session on a route that accepts replayed reasoning only up to a cumulative
-  budget must keep answering instead of wedging on a non-retryable 400 (#118182),
-  driven through one long-lived ``tui_gateway`` backend.
+* A long Nous Portal session replays every earlier turn's ``reasoning_details`` verbatim
+  and keeps answering, driven through one long-lived ``tui_gateway`` backend.
 """
 
 from __future__ import annotations
@@ -117,16 +116,15 @@ def test_reasoning_content_echoed_on_tool_call_messages(tmp_path) -> None:
     assert _assistant_field(mains[2], "reasoning_content") == ["ds-1", "ds-2"], chat_messages(mains[2], "assistant")
 
 
-def test_long_session_does_not_wedge_on_replayed_reasoning_budget(tmp_path) -> None:
-    """Each turn mints ~1 KB of reasoning; the route 400s ("Provider returned error", not
-    retryable) once the replayed total passes 4 KB. The Portal never sees replayed
-    reasoning_details on the wire (#118182), so every turn is answered and the budget is
-    never crossed, while state.db keeps the blocks for a later switch to a replaying route."""
+def test_long_session_replays_every_turns_reasoning_to_the_portal(tmp_path) -> None:
+    """Each turn mints ~1 KB of reasoning. The Portal reads reasoning_details (the #118182
+    "replay budget" does not exist live), so every later request replays every earlier
+    turn's blocks verbatim, every turn is answered, and state.db keeps the blocks."""
     turns = 8
     script = [CText(f"ANSWER-{i}", reasoning_details=_rd(str(i), "r" * 1000)) for i in range(turns)]
     h = Home(tmp_path)
     answers: list[str] = []
-    with FakeChatVariantServer(script, reasoning_budget_chars=4000) as srv:
+    with FakeChatVariantServer(script) as srv:
         h.write(_impersonated_config("inference-api.nousresearch.com"), dotenv={"OPENAI_API_KEY": "sk-fake"})
         gw = TuiGateway(h, _proxy_env(srv))
         try:
@@ -138,11 +136,9 @@ def test_long_session_does_not_wedge_on_replayed_reasoning_budget(tmp_path) -> N
             gw.close()
         records = srv.main_records()
     assert records and {r["host"] for r in records} == {"inference-api.nousresearch.com"}, "precondition: impersonated"
-    rejected = [i for i, r in enumerate(records) if r.get("response") == "route_rejection"]
-    assert rejected == [], f"route budget crossed at requests {rejected}: replayed reasoning_details reached the wire"
-    leaked = [(i, m) for i, r in enumerate(records) for m in chat_messages(r["body"], "assistant")
-              if "reasoning_details" in m]
-    assert leaked == [], f"reasoning_details sent to the Portal wire: {leaked}"
+    replayed = [_assistant_field(r["body"], "reasoning_details") for r in records]
+    assert replayed == [[_rd(str(i), "r" * 1000) for i in range(n)] for n in range(turns)], (
+        f"Portal requests did not replay every earlier turn's reasoning_details: {replayed}")
     missing = [i for i in range(turns) if f"ANSWER-{i}" not in answers[i]]
     assert not missing, f"turns {missing} were not answered: {answers}"
     # Rows persist under the STORED session key (the runtime id only routes RPCs).