fix(*): carry a live model switch to every provider holder

raven/agent/loop/main.py

@@ -553,6 +553,21 @@ def __init__(
 
         self._consolidation_tasks: set[asyncio.Task] = set()
 
+        # A switch that arrives mid-turn is parked here until no turn is
+        # running, so a turn in flight finishes on the provider it started
+        # with. A depth counter, not a flag: OriginPools gates USER and
+        # system origins on independent semaphores with no global cap
+        # (spine/scheduler.py), so a user turn and a cron turn overlap on
+        # this loop under the TUI defaults.
+        self._pending_provider: tuple[LLMProvider, str] | None = None
+        self._turns_in_flight = 0
+
+        # ``self.subagents``, ``self.context_engine`` and
+        # ``self.memory_consolidator`` were each handed ``provider`` earlier in
+        # this constructor and hold their own reference; ``_adopt_provider`` is
+        # what keeps them from outliving a live model switch. Add the call
+        # there when adding another holder.
+
         # Phase B-3: the L4 facade (``DefaultMemoryEngine`` /
         # ``MemoryEngine`` ABC) has been retired. AgentLoop now holds
         # the underlying subsystems directly:
@@ -607,6 +622,62 @@ def _apply_disabled_tools(self) -> None:
             if self.tools.has(name):
                 self.tools.unregister(name)
 
+    def set_provider(self, provider: LLMProvider, model: str) -> None:
+        """Point the loop and everything it built at a new provider/model.
+
+        ``config.set model`` builds a provider from the prospective config
+        and hands it here. Assigning ``self.provider`` alone is not enough:
+        the subagent manager, the context engine's LLM-backed segments and
+        the consolidator each captured the provider handed to them in
+        ``__init__``. Left behind, they keep calling the old endpoint for
+        the rest of the process -- which is how switching away from a dead
+        credential fixed the main loop while subagents and the skill
+        rewriter/gate went on failing to authenticate.
+
+        A switch that lands while any turn is running is parked rather than
+        applied: the loop reads ``self.provider`` at call time (eight sites
+        in this module, plus the context engine and consolidator
+        underneath), so adopting mid-turn would relay one conversation
+        across two vendors. How that surfaces depends on the path: the
+        ``chat_with_retry`` sites turn a rejected request into
+        ``finish_reason="error"`` content, so the turn reports a failure
+        with no sign that its endpoint moved, while ``_llm_call_stream``
+        (which a TUI turn takes) catches only ``TimeoutError`` and lets the
+        rejection propagate. Neither is a diagnosis the user can act on.
+
+        The park is the second line of defence, not the first: the RPC
+        rejects a switch outright when the caller's own session has a turn
+        in flight (``is_turn_active`` in ``tui_rpc.methods.config``). This
+        covers what that guard cannot see -- a caller that passes no
+        ``session_id``, and the proactive turns that run in their own lanes.
+        Note the RPC still answers ``applied: True`` and the config file is
+        already written, so a parked switch is applied on disk while the
+        loop reports the old model until the last turn drains.
+
+        Detached subagents are not covered by that park -- they outlive the
+        turn that spawned them -- so ``SubagentManager`` snapshots instead.
+        """
+        if self._turns_in_flight:
+            self._pending_provider = (provider, model)
+            return
+        self._adopt_provider(provider, model)
+
+    def _adopt_provider(self, provider: LLMProvider, model: str) -> None:
+        """Hand a provider to the loop and every subsystem holding the old one."""
+        self.provider = provider
+        self.model = model
+        self.subagents.set_provider(provider, model)
+        self.context_engine.set_provider(provider, model)
+        self.memory_consolidator.set_provider(provider, model)
+
+    def _adopt_pending_provider(self) -> None:
+        """Apply a parked switch. Callers must check that no turn is running."""
+        pending = self._pending_provider
+        if pending is None:
+            return
+        self._pending_provider = None
+        self._adopt_provider(*pending)
+
     def configure_personalization(self, enable: bool) -> None:
         """Global switch for the 4-step personalization flow (PAHF-inspired).
 
@@ -2558,14 +2629,60 @@ async def run_turn(
         inline_tool_stream: bool = False,
         usage_sink: dict[str, Any] | None = None,
         text_sink: dict[str, Any] | None = None,
+    ) -> TurnOutcome:
+        """Turn boundary for a live ``/model`` switch; see ``_run_turn``.
+
+        A parked switch is adopted here, before the turn reads
+        ``self.provider`` for the first time, and the count kept for the
+        duration is what parks the next one. Wrapping rather than
+        snapshotting because the provider is read from ``self`` at eight
+        sites in this module and by the context engine and consolidator
+        underneath them -- one boundary covers all of them, a snapshot
+        would have to be threaded through each.
+
+        Both ends gate on zero, because turns overlap: a user turn and a
+        proactive turn hold slots in separate pools. Adopting on the way in
+        would otherwise land a switch parked for a turn that is still
+        running, and clearing a flag on the way out would unpark it just as
+        wrongly. Adopting again on the last exit is what keeps a park from
+        outliving the turns it was waiting on.
+        """
+        if self._turns_in_flight == 0:
+            self._adopt_pending_provider()
+        self._turns_in_flight += 1
+        try:
+            return await self._run_turn(
+                req,
+                emit,
+                drain,
+                stream=stream,
+                inline_tool_stream=inline_tool_stream,
+                usage_sink=usage_sink,
+                text_sink=text_sink,
+            )
+        finally:
+            self._turns_in_flight -= 1
+            if self._turns_in_flight == 0:
+                self._adopt_pending_provider()
+
+    async def _run_turn(
+        self,
+        req: TurnRequest,
+        emit: Emit,
+        drain: Drain,
+        *,
+        stream: bool = True,
+        inline_tool_stream: bool = False,
+        usage_sink: dict[str, Any] | None = None,
+        text_sink: dict[str, Any] | None = None,
     ) -> TurnOutcome:
         """Spine-native turn entry: consume a TurnRequest, fan the agent's output
         onto the single ``emit``, return a TurnOutcome. Collapses the legacy
         output paths (a str return + the five callbacks) onto one boundary.
 
         Named ``run_turn`` rather than ``run``: ``run`` is the runtime keep-alive
-        (executor / debug server / MCP up, then idle). A spine runner wraps this
-        method to satisfy the TurnRunner protocol.
+        (executor / debug server / MCP up, then idle). A spine runner calls the
+        public ``run_turn`` to satisfy the TurnRunner protocol.
 
         ``stream`` is the canon Q2-D assembly switch: a streaming outlet (TUI)
         wires it True so the reply goes out as StreamDelta and dissolves (b2 — no

raven/agent/subagent/manager.py

@@ -75,6 +75,19 @@ def __init__(
         # pruned to the rolling window on access, so it self-bounds.
         self._session_spawn_times: dict[str, deque[float]] = {}
 
+    def set_provider(self, provider: LLMProvider, model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built.
+
+        Subagents run on the parent's provider, so a switch that is not
+        propagated here leaves every spawn calling the credential the loop
+        has already abandoned. Only spawns requested after this call are
+        affected: a subagent is a detached task that outlives the turn that
+        spawned it, so the loop's park cannot cover it and ``spawn``
+        snapshots the pair it was asked for.
+        """
+        self.provider = provider
+        self.model = model
+
     async def spawn(
         self,
         task: str,
@@ -107,7 +120,13 @@ async def spawn(
         display_label = label or task[:30] + ("..." if len(task) > 30 else "")
         origin = {"channel": origin_channel, "chat_id": origin_chat_id, "session_key": quota_key}
 
-        bg_task = asyncio.create_task(self._run_subagent(task_id, task, display_label, origin))
+        # Snapshot here rather than where the task starts running: it queues
+        # behind the concurrency gate and a sandbox boot first, and a switch
+        # landing in that window would hand this task an endpoint the user
+        # chose after asking for it.
+        bg_task = asyncio.create_task(
+            self._run_subagent(task_id, task, display_label, origin, self.provider, self.model)
+        )
         self._running_tasks[task_id] = bg_task
         if session_key:
             self._session_tasks.setdefault(session_key, set()).add(task_id)
@@ -131,6 +150,8 @@ async def _run_subagent(
         task: str,
         label: str,
         origin: dict[str, str],
+        provider: LLMProvider,
+        model: str,
     ) -> None:
         """Execute the subagent task and announce the result."""
         logger.info("Subagent [{}] starting task: {}", task_id, label)
@@ -141,7 +162,7 @@ async def _run_subagent(
             async with self._gate:
                 executor = build_executor(self._sandbox_config, self.workspace, self._owned_ids)
                 async with executor:
-                    await self._run_subagent_inner(task_id, task, label, origin, executor)
+                    await self._run_subagent_inner(task_id, task, label, origin, executor, provider, model)
         except Exception as e:
             error_msg = f"Error: {str(e)}"
             logger.error("Subagent [{}] failed: {}", task_id, e)
@@ -154,6 +175,8 @@ async def _run_subagent_inner(
         label: str,
         origin: dict[str, str],
         executor: Any,
+        provider: LLMProvider,
+        model: str,
     ) -> None:
         try:
             # Build subagent tools (no message tool, no spawn tool)
@@ -191,10 +214,10 @@ async def _run_subagent_inner(
             while iteration < max_iterations:
                 iteration += 1
 
-                response = await self.provider.chat_with_retry(
+                response = await provider.chat_with_retry(
                     messages=messages,
                     tools=tools.get_definitions(),
-                    model=self.model,
+                    model=model,
                 )
 
                 if response.has_tool_calls:

raven/context_engine/assembler.py

@@ -37,6 +37,7 @@
 
 if TYPE_CHECKING:
     from raven.context_engine.curator import TurnContext
+    from raven.providers.base import LLMProvider
 
 
 class ContextAssembler(ContextEngine):
@@ -64,6 +65,15 @@ def owns_compaction(self) -> bool:
         # the full append-only log and skips the host MemoryConsolidator.
         return True
 
+    def set_provider(self, provider: "LLMProvider", model: str) -> None:
+        # Duck-typed on purpose: only the builders that actually call an LLM
+        # implement it, and putting it on the SegmentBuilder protocol would
+        # force an empty override onto every purely textual builder.
+        for builder in self._builders:
+            setter = getattr(builder, "set_provider", None)
+            if callable(setter):
+                setter(provider, model)
+
     async def assemble(
         self,
         session_key: str,

raven/context_engine/base.py

@@ -35,6 +35,7 @@
     # for ``ContextEngine``, so referencing ``TurnContext`` only in type
     # hints keeps the loop unbroken.
     from raven.context_engine.curator import TurnContext
+    from raven.providers.base import LLMProvider
 
 
 # ---------------------------------------------------------------------------
@@ -142,6 +143,17 @@ def owns_compaction(self) -> bool:
         lets the engine manage history compaction itself (Curator archives
         messages out-of-band)."""
 
+    def set_provider(self, provider: "LLMProvider", model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built.
+
+        Segments that call an LLM hold the provider handed to them at
+        construction; without this they keep calling the old one for the
+        rest of the process. Concrete rather than abstract so a future
+        implementation with no LLM-backed segment is not forced to write an
+        empty override; ``ContextAssembler`` is the only one today and does
+        override it.
+        """
+
     @abstractmethod
     async def assemble(
         self,

raven/context_engine/curator.py

@@ -352,6 +352,12 @@ def __init__(
         # CuratorSegmentBuilder before any build/validate call.
         self.prefix: "AssembledPrefix | None" = None
 
+    def set_provider(self, provider: LLMProvider, model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built."""
+        self.provider = provider
+        self.model = model
+        self.trimmer.set_provider(provider, model)
+
     @staticmethod
     def working_state_segment(working_state: str | None) -> str:
         """Render segment 6 text (``# Curator Working State``) or ``""``."""

raven/context_engine/history_trimmer.py

@@ -80,6 +80,12 @@ def __init__(
         self.get_tool_definitions = get_tool_definitions
         self.context_window_tokens = context_window_tokens
 
+    def set_provider(self, provider: LLMProvider, model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built, so token
+        estimates keep matching the model actually being called."""
+        self.provider = provider
+        self.model = model
+
     # ------------------------------------------------------------------
     # Pure history-shaping helpers (no token estimation / no I/O)
     # ------------------------------------------------------------------

raven/context_engine/segments/curator.py

@@ -82,6 +82,20 @@ def __init__(
         )
         self._turn_ids: dict[str, str] = {}
 
+    def set_provider(self, provider: LLMProvider, model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built.
+
+        ``curator_model`` is re-derived with the constructor's own
+        expression, so a switch cannot make it mean something it did not
+        mean at build time. The default is non-empty, so in practice it is
+        a pin; an explicitly empty ``context.curator_model`` is the one
+        config that follows the agent model, and it follows it here too.
+        """
+        self.provider = provider
+        self.model = model
+        self.curator_model = self.config.curator_model or model
+        self.assembler.set_provider(provider, model)
+
     async def build(self, ctx: AssemblyContext) -> Segment | None:
         if ctx.prefix is None:
             raise RuntimeError("CuratorSegmentBuilder requires ctx.prefix (phase B)")

raven/context_engine/segments/skills.py

@@ -41,6 +41,7 @@
     from raven.memory_engine.skill_forge.gate import LLMGateFilter
     from raven.memory_engine.skill_forge.rewriter import QueryRewriter
     from raven.memory_engine.skill_forge.types import RouterHit
+    from raven.providers.base import LLMProvider
     from raven.skill_hub import SkillHubClient
 
 log = logging.getLogger(__name__)
@@ -73,6 +74,14 @@ def __init__(
         self._hub_client = hub_client
         self._get_tool_definitions = get_tool_definitions
 
+    def set_provider(self, provider: "LLMProvider", model: str) -> None:
+        """Hand a live ``/model`` switch down to the two LLM users in this
+        segment. The router itself holds no provider."""
+        if self._rewriter is not None:
+            self._rewriter.set_provider(provider, model)
+        if self._gate is not None:
+            self._gate.set_provider(provider, model)
+
     @trace.instrument("skill.inject", kind="skill", detached=True, extract=semconv.skill_inject_skills)
     async def build(self, ctx: AssemblyContext) -> Segment | None:
         if self._router is None:

raven/memory_engine/consolidate/consolidator.py

@@ -1723,6 +1723,11 @@ def __init__(
         self.enable_foresight = enable_foresight
         self._locks: weakref.WeakValueDictionary[str, asyncio.Lock] = weakref.WeakValueDictionary()
 
+    def set_provider(self, provider: "LLMProvider", model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built."""
+        self.provider = provider
+        self.model = model
+
     def get_lock(self, session_key: str) -> asyncio.Lock:
         """Return the shared consolidation lock for one session."""
         return self._locks.setdefault(session_key, asyncio.Lock())

raven/memory_engine/skill_forge/gate.py

@@ -61,6 +61,21 @@ def __init__(
         self._temperature = temperature
         self._max_tokens = max_tokens
 
+    def set_provider(self, provider: "LLMProvider", model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built.
+
+        Unset, ``_model`` follows whatever the new provider defaults to.
+
+        Set, it is a pin, and this leaves it pinned -- which is what a
+        restart on the new model would produce, since the gate is built
+        with the agent's provider and the pin regardless of which vendor
+        the pin names. Note that a pin is only a model id: the credential
+        comes from the provider, so a pin naming a vendor the provider does
+        not serve was already broken at boot, and stays broken here.
+        """
+        del model
+        self._provider = provider
+
     @trace.instrument("skill.gate", kind="skill", extract=semconv.skill_gate)
     async def filter(
         self,

raven/memory_engine/skill_forge/rewriter.py

@@ -74,6 +74,17 @@ def __init__(
         self._max_tokens = max_tokens
         self._temperature = temperature
 
+    def set_provider(self, provider: "LLMProvider", model: str) -> None:
+        """Adopt the provider a live ``/model`` switch just built.
+
+        Held rather than looked up per call, so without this the rewriter
+        keeps calling the provider captured at construction after the loop
+        has moved on -- a switch away from an unusable credential fixes the
+        main path and leaves this one failing.
+        """
+        del model  # the rewriter runs on the provider's default model
+        self._provider = provider
+
     @trace.instrument("skill.rewrite", kind="skill", extract=semconv.skill_rewrite)
     async def analyze(self, query: str) -> RewriteResult:
         truncated = (query or "").strip()[:_QUERY_MAX_LENGTH]