fix: close 11 correctness findings in the merged runtime

agent_core/components/middleware/llm/proxy.py

@@ -106,7 +106,13 @@ def _make_ctx(self, call_index: int) -> LLMCallContext:
                 refresh_prompt_id=True,
             )
             metadata["step_id"] = default_step_id
-            scope.metadata.update(metadata)
+            # Only the scope-stable identifier goes back to the scope.
+            # ``step_id`` / ``prompt_id`` identify THIS call, and an
+            # execution scope is shared by concurrent ``chat`` / ``stream``
+            # invocations (see ``_next_call_index``) — writing them onto the
+            # scope let two calls overwrite each other, so a failure was
+            # reported against the other call's step.
+            scope.metadata.setdefault("session_id", metadata["session_id"])
         # Stash the inner model id so a cost-accounting middleware can record
         # spend even when the provider's response_metadata doesn't echo
         # model_name (some models behind an OpenAI-compat gateway don't).
@@ -176,79 +182,89 @@ async def stream(
         ctx = self._make_ctx(self._next_call_index())
         messages = await self.chain.run_before(ctx, messages)
         attempt = 0
-        while True:
-            start_time = time.time()
-            full_content = ""
-            full_reasoning = ""
-            stream_error: Exception | None = None
-            any_chunk_yielded = False
-            try:
-                async for delta in self.inner.stream(
-                    messages,
-                    tools=tools,
-                    temperature=temperature,
-                    max_tokens=max_tokens,
-                    extra_headers=extra_headers,
-                    timeout=timeout,
-                ):
-                    full_content += delta.content or ""
-                    full_reasoning += delta.reasoning_content or ""
-                    any_chunk_yielded = True
-                    # Per-chunk middleware hook. A middleware returning
-                    # True (e.g. StreamRepetitionDetector noticing a
-                    # degenerate loop) tells us to stop consuming the
-                    # inner stream and exit cleanly — the partial
-                    # response still flows through ``after_llm`` in the
-                    # finally block so observers see the truncated
-                    # content rather than nothing. The delta that
-                    # triggered the abort IS still yielded so the
-                    # consumer's accumulator stays consistent with the
-                    # LLMResponse we'll synthesise.
-                    yield delta
-                    if await self.chain.run_on_chunk(
-                        ctx,
-                        delta,
-                        full_content,
+        start_time = time.time()
+        full_content = ""
+        full_reasoning = ""
+        stream_error: Exception | None = None
+        try:
+            while True:
+                start_time = time.time()
+                full_content = ""
+                full_reasoning = ""
+                stream_error = None
+                any_chunk_yielded = False
+                try:
+                    async for delta in self.inner.stream(
+                        messages,
+                        tools=tools,
+                        temperature=temperature,
+                        max_tokens=max_tokens,
+                        extra_headers=extra_headers,
+                        timeout=timeout,
                     ):
-                        ctx.metadata["stream_aborted_by_middleware"] = True
-                        logger.info(
-                            "LLMProxy: stream aborted by middleware after %d chars",
-                            len(full_content),
-                        )
-                        break
-                break
-            except Exception as e:
-                stream_error = e
-                await _log_llm_exception(ctx, e)
-                if not any_chunk_yielded:
-                    should_retry = await self.chain.run_on_llm_error(
-                        ctx,
-                        e,
-                        attempt,
-                    )
-                    if should_retry:
-                        attempt += 1
-                        logger.info(
-                            "LLMProxy: retrying stream after attempt %d (%s)",
+                        full_content += delta.content or ""
+                        full_reasoning += delta.reasoning_content or ""
+                        any_chunk_yielded = True
+                        # Per-chunk middleware hook. A middleware returning
+                        # True (e.g. StreamRepetitionDetector noticing a
+                        # degenerate loop) tells us to stop consuming the
+                        # inner stream and exit cleanly — the partial
+                        # response still flows through ``after_llm`` in the
+                        # finally block so observers see the truncated
+                        # content rather than nothing. The delta that
+                        # triggered the abort IS still yielded so the
+                        # consumer's accumulator stays consistent with the
+                        # LLMResponse we'll synthesise.
+                        yield delta
+                        if await self.chain.run_on_chunk(
+                            ctx,
+                            delta,
+                            full_content,
+                        ):
+                            ctx.metadata["stream_aborted_by_middleware"] = True
+                            logger.info(
+                                "LLMProxy: stream aborted by middleware after %d chars",
+                                len(full_content),
+                            )
+                            break
+                    break
+                except Exception as e:
+                    stream_error = e
+                    await _log_llm_exception(ctx, e)
+                    if not any_chunk_yielded:
+                        should_retry = await self.chain.run_on_llm_error(
+                            ctx,
+                            e,
                             attempt,
-                            type(e).__name__,
                         )
-                        continue
-                raise
-            finally:
-                duration_ms = int(
-                    (time.time() - start_time) * 1000,
-                )
-                ctx.metadata["duration_ms"] = duration_ms
-                if stream_error:
-                    ctx.metadata["error"] = str(stream_error)
-                await self.chain.run_after(
-                    ctx,
-                    LLMResponse(
-                        content=full_content,
-                        reasoning_content=full_reasoning,
-                    ),
-                )
+                        if should_retry:
+                            attempt += 1
+                            logger.info(
+                                "LLMProxy: retrying stream after attempt %d (%s)",
+                                attempt,
+                                type(e).__name__,
+                            )
+                            continue
+                    raise
+        finally:
+            # Once per ``stream()`` call, not once per retry attempt. This
+            # block used to sit inside ``while True``, so a stream that failed
+            # before its first chunk and retried handed every middleware an
+            # empty ``LLMResponse`` and then the real one — any usage/cost
+            # accounting or history-appending middleware double-counted.
+            duration_ms = int(
+                (time.time() - start_time) * 1000,
+            )
+            ctx.metadata["duration_ms"] = duration_ms
+            if stream_error:
+                ctx.metadata["error"] = str(stream_error)
+            await self.chain.run_after(
+                ctx,
+                LLMResponse(
+                    content=full_content,
+                    reasoning_content=full_reasoning,
+                ),
+            )
 
     def __getattr__(self, name: str) -> Any:
         # Defer unknown attributes to the inner client so callers that

agent_core/components/middleware/llm/summarization.py

@@ -111,7 +111,15 @@ async def before_llm(self, ctx: LLMCallContext, messages: list[Message]) -> list
         # return HTTP 400 "No tool call found for function call output
         # with call_id ...".
         split_idx = len(rest) - keep_recent
-        while split_idx < len(rest) - 1 and rest[split_idx].get("role") == "tool":
+        # ``len(rest)``, not ``len(rest) - 1``: stopping one short left the
+        # window starting on a tool message whenever the tail was all tool
+        # results — e.g. keep_recent=3 over
+        # ``[assistant(tool_calls), tool, tool, tool]`` kept ``[tool]``, the
+        # exact orphan this loop exists to prevent. Running off the end is
+        # the safe outcome: everything gets summarised and the kept window is
+        # empty, rather than shipping a tool result whose ``tool_calls`` was
+        # summarised away.
+        while split_idx < len(rest) and rest[split_idx].get("role") == "tool":
             split_idx += 1
         to_summarize = rest[:split_idx]
         keep = rest[split_idx:]

agent_core/components/observers/budget_observer.py

@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 from agent_core.loop_types import Intervention, LoopConfig, TurnContext
+from agent_core.runtime.loop import usage_input_tokens, usage_output_tokens
 
 
 class BudgetObserver:
@@ -26,10 +27,21 @@ async def on_loop_start(self, config: LoopConfig) -> None:
         self._warned = False
 
     async def on_llm_response(self, ctx: TurnContext) -> None:
-        """Accumulate token counts from the LLM response usage dict."""
+        """Accumulate token counts from the LLM response usage dict.
+
+        Reads both spellings through the shared accessors: the loop
+        normalises usage to ``prompt_tokens`` / ``completion_tokens``
+        (``extract_usage``), while hosts and test doubles may hand in the
+        ``input_tokens`` / ``output_tokens`` aliases. Reading only the
+        aliases left ``tokens_used`` at 0 against normalised usage, so
+        neither the ``budget_exhausted`` intervention nor the warning at
+        ``warn_ratio`` could ever fire in production.
+        """
         if ctx.usage is None:
             return
-        self.tokens_used += ctx.usage.get("input_tokens", 0) + ctx.usage.get("output_tokens", 0)
+        self.tokens_used += (
+            usage_input_tokens(ctx.usage) + usage_output_tokens(ctx.usage)
+        )
         ctx.metadata["budget_tokens_used"] = self.tokens_used
         ctx.metadata["budget_tokens_limit"] = self.max_tokens
 

agent_core/components/observers/context_size_guard.py

@@ -10,6 +10,7 @@
     LoopConfig,
     TurnContext,
 )
+from agent_core.runtime.loop import usage_input_tokens
 from agent_core.runtime.loop.compact import (
     COMPACTION_SEQ_KEY,
     FORCE_COMPACTION_KEY,
@@ -51,10 +52,7 @@ async def on_loop_start(self, config: LoopConfig) -> None:
     async def on_llm_response(self, ctx: TurnContext) -> Intervention | None:
         if self._tripped or ctx.usage is None:
             return None
-        # Normalised usage (infra/usage.py) exposes ``prompt_tokens``; raw
-        # Anthropic payloads use ``input_tokens``. Read prompt_tokens first —
-        # reading only ``input_tokens`` meant this guard saw 0 and never tripped.
-        used = int(ctx.usage.get("prompt_tokens") or ctx.usage.get("input_tokens") or 0)
+        used = usage_input_tokens(ctx.usage)
         if used <= self._limit:
             self._requested_at_seq = None
             return None

agent_core/components/observers/leaked_tool_call_retry.py

@@ -151,6 +151,18 @@ async def on_llm_response(self, ctx: TurnContext) -> Intervention | None:
             ctx.metadata.pop(_METADATA_TEMP_KEY, None)
             return None
 
+        # This turn ran with no tools bound at all — ``LastTurnForcer`` strips
+        # them for the landing turn. Nothing could be parsed, so
+        # ``blocked_tool_calls`` is empty and the guard above does not cover
+        # it, yet a prose answer that merely mentions ``finalize_answer(...)``
+        # trips ``_looks_like_leak``. Replaying would hand tools back on the
+        # one turn that was supposed to have none, because
+        # ``_llm_strip_tools`` is one-shot and already consumed.
+        if ctx.metadata.get("_llm_tools_stripped"):
+            ctx.metadata.pop(_METADATA_COUNT_KEY, None)
+            ctx.metadata.pop(_METADATA_TEMP_KEY, None)
+            return None
+
         # A structured tool call was parsed — nothing to retry. Also reset
         # the escalation counter so a future leak starts at 0.3 again.
         if ctx.tool_calls:

agent_core/components/observers/trajectory.py

@@ -346,8 +346,16 @@ def _flush_json(self, *, force: bool = False) -> None:
         if not force:
             self._pending_flushes += 1
             now = time.monotonic()
+            # "0 disables the respective bound" (module header). With every
+            # bound disabled there is nothing left to coalesce, so every event
+            # flushes — treating a disabled bound as "never due" instead meant
+            # ``SWARM_TRAJECTORY_COALESCE_N=0`` froze the JSON snapshot after
+            # the first flush, and a process killed before ``on_loop_end``
+            # left only the ``start`` state on disk.
+            coalescing = _COALESCE_N > 0 or _COALESCE_MS > 0
             due = (
-                self._last_flush_at == 0.0
+                not coalescing
+                or self._last_flush_at == 0.0
                 or (
                     _COALESCE_N > 0
                     and self._pending_flushes >= _COALESCE_N

agent_core/components/skills/extensions_config.py

@@ -63,7 +63,16 @@ class ExtensionsConfig(BaseModel):
     @classmethod
     def from_file(cls, config_path: str | Path | None = None) -> ExtensionsConfig:
         """Load config from JSON file with environment variable resolution."""
-        resolved = Path(config_path) if config_path else _find_config_file()
+        if config_path:
+            resolved = Path(config_path)
+            # Keep the source stable across later ``chdir`` calls.  Do not use
+            # ``resolve()`` here: retaining a configured symlink path lets an
+            # operator atomically repoint that symlink and have reload observe
+            # the new target.
+            if not resolved.is_absolute():
+                resolved = Path.cwd() / resolved
+        else:
+            resolved = _find_config_file()
 
         if resolved is None or not resolved.is_file():
             logger.debug("No extensions config found — using empty defaults")
@@ -83,6 +92,16 @@ def from_file(cls, config_path: str | Path | None = None) -> ExtensionsConfig:
             logger.warning("Failed to load extensions config %s: %s", resolved, e)
             return cls()
 
+    @property
+    def source_path(self) -> Path | None:
+        """File this config was loaded from; ``None`` for in-memory defaults.
+
+        A reloader must pass this back to :meth:`from_file` — calling it with
+        no argument restarts the cwd/env search, which silently returns empty
+        defaults for a config that was loaded from an explicit path.
+        """
+        return self._file_path
+
     def has_changed(self) -> bool:
         """Return True if the backing file has been modified since load."""
         if self._file_path is None or not self._file_path.is_file():

agent_core/components/skills/file_system_loader.py

@@ -188,7 +188,7 @@ def get_enabled_skills(self) -> list[SkillConfig]:
         """
         if self._extensions_config.has_changed():
             logger.info("Extensions config changed on disk — reloading skill state")
-            self._extensions_config = ExtensionsConfig.from_file()
+            self._extensions_config = self._reload_extensions_config()
             # Re-apply enabled state to already-discovered skills
             for skill in self._skills.values():
                 skill.enabled = self._extensions_config.is_skill_enabled(skill.skill_id)
@@ -197,7 +197,14 @@ def get_enabled_skills(self) -> list[SkillConfig]:
     # ── Mutations ─────────────────────────────────────────────────────
 
     def toggle_skill(self, skill_id: str, enabled: bool) -> bool:
-        """Enable or disable a skill. Returns True if skill exists."""
+        """Enable or disable a skill. Returns True if skill exists.
+
+        Discovers lazily like every other accessor: without this, toggling on
+        a freshly built loader read an empty ``_skills`` and reported False for
+        a skill that exists on disk — a silent no-op.
+        """
+        if not self._discovered:
+            self.discover()
         skill = self._skills.get(skill_id)
         if skill is None:
             return False
@@ -208,5 +215,27 @@ def reload(self) -> None:
         """Force re-discovery of skills."""
         self._skills.clear()
         self._discovered = False
-        self._extensions_config = ExtensionsConfig.from_file()
+        self._extensions_config = self._reload_extensions_config()
         self.discover()
+
+    def _reload_extensions_config(self) -> ExtensionsConfig:
+        """Re-read the config from the file it came from.
+
+        Calling ``from_file()`` with no argument restarts the cwd/env search.
+        For a loader built with a config loaded from an explicit path outside
+        those locations, that search finds nothing and returns empty defaults —
+        and because ``is_skill_enabled`` defaults to True for anything unlisted,
+        an operator disabling a skill got every skill re-enabled instead. The
+        empty config also carries no ``_file_path``, so ``has_changed`` returned
+        False from then on and the state could never recover.
+        """
+        source = self._extensions_config.source_path
+        reloaded = ExtensionsConfig.from_file(source)
+        if source is not None and reloaded.source_path is None:
+            logger.warning(
+                "Extensions config %s could not be re-read; keeping the "
+                "previously loaded skill state",
+                source,
+            )
+            return self._extensions_config
+        return reloaded