feat(code): add Fireworks session settings (#4360)

Fireworks-backed model calls now receive stable per-thread session
identity from Deep Agents Code runtime context.

## Changes
- Add `thread_id` to `CLIContext` so model-call middleware can see the
active LangGraph session without reading global config.
- Inject Fireworks session settings from the thread ID, adding
`prompt_cache_key` and `x-multi-turn-session-id` only when callers have
not already supplied equivalent values.
- Preserve existing model settings and headers, including
case-insensitive handling for legacy Fireworks affinity headers and
malformed `extra_headers`.
- Pass runtime context through non-interactive streaming and Textual
task execution so Fireworks session identity is available consistently.

libs/code/deepagents_code/_cli_context.py

@@ -41,6 +41,8 @@ class CLIContextSchema:
 
     approval_mode_key: str | None = None
 
+    thread_id: str | None = None
+
 
 class CLIContext(TypedDict, total=False):
     """Client-facing builder for the per-run graph context payload.
@@ -87,3 +89,11 @@ class CLIContext(TypedDict, total=False):
     each gated tool call so auto-to-manual changes can take effect before the
     current stream returns.
     """
+
+    thread_id: str | None
+    """LangGraph thread ID for the active conversation.
+
+    Mirrors `config.configurable.thread_id` into runtime context for model-call
+    middleware that needs per-request session identity, including Fireworks
+    session-affinity headers.
+    """

libs/code/deepagents_code/configurable_model.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import asyncio
 import logging
+from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
 
 from deepagents._models import model_matches_spec  # noqa: PLC2701
@@ -29,6 +30,26 @@ if TYPE_CHECKING:
 logger = logging.getLogger(__name__)
 
 
+def _get_ls_provider(model: object) -> str | None:
+    """Return the LangSmith provider name reported by a chat model.
+
+    Returns:
+        The `ls_provider` string when the model reports one, otherwise `None`
+            (including when `_get_ls_params` is missing, raises, or yields a
+            non-string provider).
+    """
+    try:
+        ls_params = model._get_ls_params()  # ty: ignore[unresolved-attribute]
+    except (AttributeError, TypeError, RuntimeError):
+        logger.debug("_get_ls_params raised for %s", type(model).__name__)
+        return None
+    if isinstance(ls_params, dict):
+        provider = ls_params.get("ls_provider")
+        if isinstance(provider, str):
+            return provider
+    return None
+
+
 def _is_anthropic_model(model: object) -> bool:
     """Check whether a resolved model reports `'anthropic'` as its provider.
 
@@ -44,15 +65,16 @@ def _is_anthropic_model(model: object) -> bool:
     Returns:
         `True` if the model's `ls_provider` is `'anthropic'`.
     """
-    try:
-        ls_params = model._get_ls_params()  # ty: ignore[unresolved-attribute]
-    except (AttributeError, TypeError, RuntimeError):
-        logger.debug(
-            "_get_ls_params raised for %s; assuming non-Anthropic",
-            type(model).__name__,
-        )
-        return False
-    return isinstance(ls_params, dict) and ls_params.get("ls_provider") == "anthropic"
+    return _get_ls_provider(model) == "anthropic"
+
+
+def _is_fireworks_model(model: object) -> bool:
+    """Check whether a resolved model reports `'fireworks'` as its provider.
+
+    Returns:
+        `True` if the model's `ls_provider` is `'fireworks'`.
+    """
+    return _get_ls_provider(model) == "fireworks"
 
 
 _ANTHROPIC_ONLY_SETTINGS: set[str] = {"cache_control"}
@@ -60,6 +82,64 @@ _ANTHROPIC_ONLY_SETTINGS: set[str] = {"cache_control"}
 `AnthropicPromptCachingMiddleware`) that are not accepted by other providers and
 must be stripped on cross-provider swap."""
 
+_FIREWORKS_MULTI_TURN_SESSION_HEADER = "x-multi-turn-session-id"
+"""Fireworks header populated from the active Deep Agents Code thread ID."""
+
+_FIREWORKS_SESSION_AFFINITY_HEADER = "x-session-affinity"
+"""Legacy Fireworks prompt-cache affinity header."""
+
+
+def _has_header(headers: Mapping[object, object], target: str) -> bool:
+    """Return whether a headers mapping already includes `target`.
+
+    Comparison is case-insensitive; `target` must be supplied in lowercase.
+
+    Returns:
+        `True` if a string key case-insensitively equal to `target` is present.
+    """
+    return any(isinstance(key, str) and key.lower() == target for key in headers)
+
+
+def _with_fireworks_session_settings(
+    model_settings: dict[str, Any], thread_id: str
+) -> dict[str, Any] | None:
+    """Return model settings with Fireworks session settings added if needed.
+
+    Existing settings are preserved and never overwritten. `prompt_cache_key` is
+    the preferred typed form for prompt-cache affinity; the raw
+    `x-session-affinity` header is also treated as caller-provided affinity.
+
+    Returns:
+        A new `model_settings` dict with the missing session settings added, or
+            `None` when nothing needed adding or `extra_headers` is present but
+            not a mapping (leaving the request untouched).
+    """
+    raw_headers = model_settings.get("extra_headers")
+    if raw_headers is None:
+        headers: dict[object, object] = {}
+    elif isinstance(raw_headers, Mapping):
+        headers = dict(raw_headers)
+    else:
+        logger.warning(
+            "Cannot inject Fireworks session settings because extra_headers is %s",
+            type(raw_headers).__name__,
+        )
+        return None
+
+    updated: dict[str, Any] = {}
+    if "prompt_cache_key" not in model_settings and not _has_header(
+        headers, _FIREWORKS_SESSION_AFFINITY_HEADER
+    ):
+        updated["prompt_cache_key"] = thread_id
+
+    if not _has_header(headers, _FIREWORKS_MULTI_TURN_SESSION_HEADER):
+        headers[_FIREWORKS_MULTI_TURN_SESSION_HEADER] = thread_id
+        updated["extra_headers"] = headers
+
+    if not updated:
+        return None
+    return {**model_settings, **updated}
+
 
 def _get_context(request: ModelRequest) -> CLIContextSchema | None:
     """Return runtime context when it matches the CLI context shape."""
@@ -72,12 +152,14 @@ def _get_context(request: ModelRequest) -> CLIContextSchema | None:
         return ctx
     if isinstance(ctx, dict):
         raw_key = ctx.get("approval_mode_key")
+        raw_thread_id = ctx.get("thread_id")
         return CLIContextSchema(
             model=ctx.get("model"),
             model_params=ctx.get("model_params") or {},
             effective_model=ctx.get("effective_model"),
             auto_approve=bool(ctx.get("auto_approve", False)),
             approval_mode_key=raw_key if isinstance(raw_key, str) else None,
+            thread_id=raw_thread_id if isinstance(raw_thread_id, str) else None,
         )
     return None
 
@@ -115,6 +197,18 @@ def _build_overrides(
     if model_params:
         overrides["model_settings"] = {**request.model_settings, **model_params}
 
+    effective_model = new_model if new_model is not None else request.model
+    if ctx.thread_id and _is_fireworks_model(effective_model):
+        settings = overrides.get("model_settings", request.model_settings)
+        settings_with_session = _with_fireworks_session_settings(
+            settings, ctx.thread_id
+        )
+        if settings_with_session is not None:
+            overrides["model_settings"] = settings_with_session
+            # No thread ID in the message: it is treated as a sensitive session
+            # identifier. The line's presence alone confirms injection ran.
+            logger.debug("Injected Fireworks session settings")
+
     if not overrides:
         return request
 

libs/code/deepagents_code/non_interactive.py

@@ -13,8 +13,9 @@ Shell commands are gated by an optional allow-list (`--shell-allow-list`):
     against the list; non-shell tools approved unconditionally.
 - `all` → shell enabled, any command allowed, all tools auto-approved.
 
-An optional quiet mode (`--quiet` / `-q`) redirects all console output to
-stderr, leaving stdout exclusively for the agent's response text.
+An optional quiet mode (`--quiet` / `-q`) suppresses stream-time diagnostics
+(the tool-call and file-operation notifications) so stdout carries only the
+agent's response text.
 """
 
 from __future__ import annotations
@@ -37,6 +38,7 @@ from rich.spinner import Spinner as RichSpinner
 from rich.style import Style
 from rich.text import Text
 
+from deepagents_code._cli_context import CLIContext
 from deepagents_code._version import __version__
 from deepagents_code.agent import DEFAULT_AGENT_NAME
 from deepagents_code.config import (
@@ -233,9 +235,9 @@ class StreamState:
     """Mutable state accumulated while iterating over the agent stream."""
 
     quiet: bool = False
-    """When `True`, diagnostic formatting that would otherwise go to stdout
-    (e.g. separator newlines before tool notifications) is suppressed so that
-    stdout contains only agent response text."""
+    """When `True`, stream-time diagnostics (the tool-call and file-operation
+    notifications, plus the stdout separator newline preceding a tool call) are
+    suppressed, so stdout carries only agent response text."""
 
     stream: bool = True
     """When `True` (default), text chunks are written to stdout as they arrive.
@@ -421,7 +423,9 @@ def _process_ai_message(
                 state.tool_call_buffers[buffer_key]["name"] = chunk_name
                 if state.spinner:
                     state.spinner.stop()
-                if state.full_response and not state.quiet:
+                if state.quiet:
+                    continue
+                if state.full_response:
                     _write_newline()
                 console.print(
                     f"[dim]🔧 Calling tool: {escape_markup(chunk_name)}[/dim]",
@@ -468,10 +472,11 @@ def _process_message_chunk(
         if record and record.diff:
             if state.spinner:
                 state.spinner.stop()
-            console.print(
-                f"[dim]📝 {escape_markup(record.display_path)}[/dim]",
-                highlight=False,
-            )
+            if not state.quiet:
+                console.print(
+                    f"[dim]📝 {escape_markup(record.display_path)}[/dim]",
+                    highlight=False,
+                )
         if state.spinner:
             state.spinner.start()
 
@@ -652,6 +657,7 @@ async def _stream_agent(
     state: StreamState,
     console: Console,
     file_op_tracker: FileOpTracker,
+    context: CLIContext,
 ) -> None:
     """Consume the full agent stream and update *state* with results.
 
@@ -663,6 +669,7 @@ async def _stream_agent(
         state: Shared stream state.
         console: Rich console for formatted output.
         file_op_tracker: Tracker for file-operation diffs.
+        context: Runtime context for model-call middleware.
     """
     if state.spinner:
         state.spinner.start()
@@ -672,6 +679,7 @@ async def _stream_agent(
             stream_mode=["messages", "updates"],
             subgraphs=True,
             config=config,
+            context=context,
             durability="exit",
         ):
             _process_stream_chunk(chunk, state, console, file_op_tracker)
@@ -730,12 +738,18 @@ async def _run_agent_loop(
     stream_input: dict[str, Any] | Command = {"messages": [user_msg]}
 
     thread_id = config.get("configurable", {}).get("thread_id", "")
+    # An empty or missing thread ID carries no session identity, so leave it
+    # unset in context rather than passing a blank string to model middleware.
+    context_thread_id = thread_id if isinstance(thread_id, str) and thread_id else None
+    context = CLIContext(thread_id=context_thread_id)
     await dispatch_hook("session.start", {"thread_id": thread_id})
 
     start_time = time.monotonic()
 
     # Initial stream
-    await _stream_agent(agent, stream_input, config, state, console, file_op_tracker)
+    await _stream_agent(
+        agent, stream_input, config, state, console, file_op_tracker, context
+    )
 
     # The internal default applies when --max-turns is omitted, guarding
     # against unbounded runaway loops in scripts that forgot to set one.
@@ -764,7 +778,7 @@ async def _run_agent_loop(
         _process_hitl_interrupts(state, console)
         stream_input = Command(resume=state.hitl_response)
         await _stream_agent(
-            agent, stream_input, config, state, console, file_op_tracker
+            agent, stream_input, config, state, console, file_op_tracker, context
         )
 
     wall_time = time.monotonic() - start_time

libs/code/deepagents_code/textual_adapter.py

@@ -575,6 +575,7 @@ async def execute_task_textual(
             # direction, but the same store write also propagates turning it on.
             if context is None:
                 context = CLIContext()
+            context["thread_id"] = thread_id
             auto_approve = bool(session_state.auto_approve)
             context["auto_approve"] = auto_approve
             try:

libs/code/tests/unit_tests/test_configurable_model.py

@@ -1,6 +1,7 @@
 """Tests for ConfigurableModelMiddleware."""
 
 import asyncio
+import logging
 from collections.abc import Callable
 from types import SimpleNamespace
 from typing import Any, cast
@@ -17,6 +18,7 @@ from deepagents_code.configurable_model import (
     ConfigurableModelMiddleware,
     _get_context,
     _is_anthropic_model,
+    _is_fireworks_model,
 )
 
 
@@ -102,6 +104,7 @@ class TestNoOverride:
             context={
                 "auto_approve": True,
                 "approval_mode_key": "approval-key",
+                "thread_id": "thread-123",
             },
         )
 
@@ -110,6 +113,7 @@ class TestNoOverride:
         assert ctx is not None
         assert ctx.auto_approve is True
         assert ctx.approval_mode_key == "approval-key"
+        assert ctx.thread_id == "thread-123"
 
     @pytest.mark.parametrize("key", [None, 1, object()])
     def test_dict_context_coerces_non_string_approval_key(self, key: object) -> None:
@@ -127,6 +131,18 @@ class TestNoOverride:
         assert ctx.auto_approve is True
         assert ctx.approval_mode_key is None
 
+    @pytest.mark.parametrize("thread_id", [None, 1, object()])
+    def test_dict_context_coerces_non_string_thread_id(self, thread_id: object) -> None:
+        request = _make_request(
+            _make_model("claude-sonnet-4-6"),
+            context={"thread_id": thread_id},
+        )
+
+        ctx = _get_context(request)
+
+        assert ctx is not None
+        assert ctx.thread_id is None
+
     def test_same_model_spec(self) -> None:
         request = _make_request(
             _make_model("claude-sonnet-4-6"),
@@ -445,6 +461,229 @@ class TestAnthropicSettingsStripped:
         assert captured[0].model_settings == {}
 
 
+class TestFireworksSessionSettings:
+    """Fireworks model calls receive session settings from the thread ID."""
+
+    def _fireworks_model(self) -> MagicMock:
+        model = _make_model("accounts/fireworks/models/kimi-k2p7-code")
+        model._get_ls_params.return_value = {"ls_provider": "fireworks"}
+        return model
+
+    def test_fireworks_model_gets_session_settings(self) -> None:
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert captured[0].model is request.model
+        assert captured[0].model_settings == {
+            "prompt_cache_key": "thread-123",
+            "extra_headers": {"x-multi-turn-session-id": "thread-123"},
+        }
+
+    def test_existing_headers_preserved_and_session_affinity_not_overwritten(
+        self,
+    ) -> None:
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+            model_settings={
+                "extra_headers": {
+                    "Authorization": "Bearer custom",
+                    "X-Session-Affinity": "custom-session",
+                }
+            },
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert captured[0].model_settings == {
+            "extra_headers": {
+                "Authorization": "Bearer custom",
+                "X-Session-Affinity": "custom-session",
+                "x-multi-turn-session-id": "thread-123",
+            }
+        }
+
+    def test_non_fireworks_model_unchanged_with_thread_id(self) -> None:
+        request = _make_request(
+            _make_model("gpt-5.5"),
+            context=CLIContext(thread_id="thread-123"),
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert captured[0] is request
+
+    def test_fireworks_swap_gets_session_settings(self) -> None:
+        override = self._fireworks_model()
+        request = _make_request(
+            _make_model("gpt-5.5"),
+            context=CLIContext(
+                model="fireworks:accounts/fireworks/models/kimi-k2p7-code",
+                thread_id="thread-123",
+            ),
+        )
+        captured: list[ModelRequest] = []
+
+        with patch(_PATCH_CREATE, return_value=_make_model_result(override)):
+            _mw.wrap_model_call(
+                request, lambda r: (captured.append(r), _make_response())[1]
+            )
+
+        assert captured[0].model is override
+        assert captured[0].model_settings == {
+            "prompt_cache_key": "thread-123",
+            "extra_headers": {"x-multi-turn-session-id": "thread-123"},
+        }
+
+    async def test_async_fireworks_model_gets_session_settings(self) -> None:
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+        )
+        captured: list[ModelRequest] = []
+
+        async def handler(r: ModelRequest) -> ModelResponse[Any]:  # noqa: RUF029
+            captured.append(r)
+            return _make_response()
+
+        await _mw.awrap_model_call(request, handler)
+
+        assert captured[0].model_settings == {
+            "prompt_cache_key": "thread-123",
+            "extra_headers": {"x-multi-turn-session-id": "thread-123"},
+        }
+
+    def test_empty_thread_id_skips_session_settings(self) -> None:
+        """A blank thread ID must not inject empty session settings."""
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id=""),
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert captured[0] is request
+
+    def test_non_mapping_extra_headers_skips_injection(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        """Malformed `extra_headers` leaves the request untouched and warns."""
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+            model_settings={"extra_headers": ["not", "a", "mapping"]},
+        )
+        captured: list[ModelRequest] = []
+
+        with caplog.at_level(
+            logging.WARNING, logger="deepagents_code.configurable_model"
+        ):
+            _mw.wrap_model_call(
+                request, lambda r: (captured.append(r), _make_response())[1]
+            )
+
+        assert captured[0] is request
+        assert captured[0].model_settings == {"extra_headers": ["not", "a", "mapping"]}
+        assert "extra_headers" in caplog.text
+
+    def test_existing_prompt_cache_key_not_overwritten(self) -> None:
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+            model_settings={"prompt_cache_key": "custom-cache"},
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert captured[0].model_settings == {
+            "prompt_cache_key": "custom-cache",
+            "extra_headers": {"x-multi-turn-session-id": "thread-123"},
+        }
+
+    def test_existing_multi_turn_header_case_insensitive(self) -> None:
+        """A differently-cased multi-turn header is not duplicated."""
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+            model_settings={
+                "extra_headers": {"X-Multi-Turn-Session-ID": "custom-multi"}
+            },
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert captured[0].model_settings == {
+            "prompt_cache_key": "thread-123",
+            "extra_headers": {"X-Multi-Turn-Session-ID": "custom-multi"},
+        }
+
+    def test_caller_model_settings_not_mutated(self) -> None:
+        """Injection copies the caller's dicts instead of mutating in place."""
+        original_headers = {"Authorization": "Bearer token"}
+        model_settings = {"extra_headers": original_headers}
+        request = _make_request(
+            self._fireworks_model(),
+            context=CLIContext(thread_id="thread-123"),
+            model_settings=model_settings,
+        )
+        captured: list[ModelRequest] = []
+
+        _mw.wrap_model_call(
+            request, lambda r: (captured.append(r), _make_response())[1]
+        )
+
+        assert original_headers == {"Authorization": "Bearer token"}
+        assert model_settings == {"extra_headers": {"Authorization": "Bearer token"}}
+        assert captured[0].model_settings["extra_headers"] is not original_headers
+
+
+class TestIsFireworksModel:
+    """Direct tests for the `_is_fireworks_model` helper."""
+
+    def test_returns_true_for_fireworks(self) -> None:
+        model = _make_model("accounts/fireworks/models/kimi-k2p7-code")
+        model._get_ls_params.return_value = {"ls_provider": "fireworks"}
+        assert _is_fireworks_model(model) is True
+
+    def test_returns_false_for_non_fireworks(self) -> None:
+        assert _is_fireworks_model(_make_model("gpt-5.5")) is False
+
+    def test_returns_false_for_plain_object(self) -> None:
+        assert _is_fireworks_model(object()) is False
+
+    def test_returns_false_when_ls_params_returns_none(self) -> None:
+        model = MagicMock(spec=BaseChatModel)
+        model._get_ls_params.return_value = None
+        assert _is_fireworks_model(model) is False
+
+    def test_returns_false_when_ls_provider_not_str(self) -> None:
+        model = MagicMock(spec=BaseChatModel)
+        model._get_ls_params.return_value = {"ls_provider": 123}
+        assert _is_fireworks_model(model) is False
+
+
 class TestIsAnthropicModel:
     """Direct tests for the `_is_anthropic_model` helper."""
 

libs/code/tests/unit_tests/test_non_interactive.py

@@ -5,11 +5,12 @@ import io
 import signal
 import sys
 from collections.abc import AsyncIterator, Iterator, Sequence
+from types import SimpleNamespace
 from typing import TYPE_CHECKING, Any
 from unittest.mock import AsyncMock, MagicMock, call, patch
 
 import pytest
-from langchain_core.messages import AIMessage
+from langchain_core.messages import AIMessage, ToolMessage
 from rich.console import Console
 
 if TYPE_CHECKING:
@@ -27,6 +28,7 @@ from deepagents_code.non_interactive import (
     _collect_action_request_warnings,
     _make_hitl_decision,
     _process_ai_message,
+    _process_message_chunk,
     _run_agent_loop,
     _run_startup_command,
     _start_langsmith_thread_url_lookup,
@@ -472,13 +474,42 @@ class TestQuietMode:
         assert "Calling tool" not in stdout
         assert "Task completed" not in stdout
         assert "Running task" not in stdout
-        # Tool notifications still go to stderr
-        assert "Calling tool" in stderr or "read_file" in stderr
-        # Header and completion messages are fully suppressed in quiet mode
+        # Quiet mode suppresses diagnostics on stderr too.
+        assert "Calling tool" not in stderr
+        assert "read_file" not in stderr
         assert "Task completed" not in stderr
         assert "Running task" not in stderr
 
 
+class TestQuietFileOpNotification:
+    """The file-operation (📝) notification honors quiet mode."""
+
+    @staticmethod
+    def _run(*, quiet: bool) -> str:
+        """Drive a file-op `ToolMessage` chunk and return captured stderr."""
+        record = SimpleNamespace(diff="--- a\n+++ b", display_path="src/foo.py")
+        tracker = MagicMock()
+        tracker.complete_with_message.return_value = record
+
+        stderr_buf = io.StringIO()
+        console = Console(file=stderr_buf, width=200)
+        state = StreamState(quiet=quiet, stream=True, spinner=None)
+
+        _process_message_chunk(
+            (ToolMessage(content="ok", tool_call_id="tc1"), {}),
+            state,
+            console,
+            tracker,
+        )
+        return stderr_buf.getvalue()
+
+    def test_quiet_suppresses_file_op_notification(self) -> None:
+        assert self._run(quiet=True) == ""
+
+    def test_non_quiet_emits_file_op_notification(self) -> None:
+        assert "foo.py" in self._run(quiet=False)
+
+
 class TestNoStreamMode:
     """Tests for --no-stream flag in run_non_interactive."""
 
@@ -1122,6 +1153,29 @@ def _make_looping_agent() -> MagicMock:
 class TestMaxTurns:
     """Tests for max_turns parameter in _run_agent_loop."""
 
+    async def test_run_agent_loop_passes_thread_id_context(self) -> None:
+        """Non-interactive runs mirror `configurable.thread_id` into context."""
+        agent = MagicMock()
+        agent.astream = MagicMock(return_value=_async_iter([]))
+        console = Console(quiet=True)
+        file_op_tracker = MagicMock()
+        config: RunnableConfig = {"configurable": {"thread_id": "t1"}}
+
+        with patch(
+            "deepagents_code.non_interactive.dispatch_hook", new_callable=AsyncMock
+        ):
+            await _run_agent_loop(
+                agent,
+                "task",
+                config,
+                console,
+                file_op_tracker,
+                quiet=True,
+            )
+
+        _, kwargs = agent.astream.call_args
+        assert kwargs["context"]["thread_id"] == "t1"
+
     async def test_raises_after_user_limit(self) -> None:
         """HITLIterationLimitError is raised after max_turns HITL iterations."""
         agent = _make_looping_agent()

libs/code/tests/unit_tests/test_textual_adapter.py

@@ -982,6 +982,7 @@ class TestExecuteTaskTextualAutoApproveInput:
         assert not isinstance(stream_input, Command)
         assert stream_input == {"messages": [{"role": "user", "content": "hi"}]}
         assert agent.contexts[0]["auto_approve"] is True
+        assert agent.contexts[0]["thread_id"] == "thread-1"
         key = approval_mode_key("thread-1")
         assert agent.contexts[0]["approval_mode_key"] == key
         assert agent.store_items == [
@@ -1113,6 +1114,8 @@ class TestExecuteTaskTextualAutoApproveInput:
         assert len(agent.contexts) == 2
         assert agent.contexts[0]["auto_approve"] is False
         assert agent.contexts[1]["auto_approve"] is True
+        assert agent.contexts[0]["thread_id"] == "thread-1"
+        assert agent.contexts[1]["thread_id"] == "thread-1"
         key = approval_mode_key("thread-1")
         assert agent.contexts[0]["approval_mode_key"] == key
         assert agent.contexts[1]["approval_mode_key"] == key