Day 5: self-healing, interruption, and model-API retry

src/agentling/agent.py

@@ -1,16 +1,17 @@
 """The agent loop.
 
-Agent assembles the framework's primitives — a Model, a set of Tools, and a
-Memory of typed steps — into an async ReAct loop. A single async generator
-(``_run_stream``) drives everything and yields Events; ``run()`` is a thin
-dispatcher that either hands back that stream (``stream=True``) or drains it and
-returns the final answer (``stream=False``).
+Agent ties the framework's primitives (a Model, some Tools, and a Memory of
+typed steps) into an async ReAct loop. One async generator, _run_stream, does
+the real work and yields Events. run() is a thin dispatcher: it hands back that
+stream when stream is True, or drains it and returns the final answer otherwise.
 """
 
 from __future__ import annotations
 
 import asyncio
 import time
+import json
+from dataclasses import replace
 from collections.abc import AsyncIterator, Callable, Sequence
 from typing import Any
 
@@ -25,6 +26,11 @@
     "final_answer tool (or simply reply with plain text)."
 )
 
+_LOOP_NUDGE = (
+    " Note: you already made this exact tool call and got the same result. "
+    "Try a different approach."
+)
+
 
 class Agent:
     """An async tool-calling agent.
@@ -48,11 +54,11 @@ def __init__(
         self.max_steps = max_steps
         self.step_callbacks = list(step_callbacks)
         self.parallel_tools = parallel_tools
-        self.skills = list(skills)  # wired in Day 6
+        self.skills = list(skills)  # accepted but not consumed yet
 
         self.memory = Memory()
+        self._interrupt = asyncio.Event()
 
-        # final_answer is always available so the model has an explicit way to stop.
         all_tools = [*tools, FinalAnswerTool()]
         self.tools: dict[str, Tool] = {tool.name: tool for tool in all_tools}
         self._tool_schemas = [tool.to_schema() for tool in all_tools]
@@ -67,10 +73,15 @@ def run(
     ) -> Any:
         """Run the agent on a task.
 
-        With stream=False (default) returns a coroutine resolving to the final
-        answer string:  ``answer = await agent.run(task)``.
-        With stream=True returns an async iterator of Events:
-        ``async for event in agent.run(task, stream=True): ...``.
+        With stream=False (the default) this returns a coroutine that resolves
+        to the final answer string:
+
+            answer = await agent.run(task)
+
+        With stream=True it returns an async iterator of Events instead:
+
+            async for event in agent.run(task, stream=True):
+                ...
         """
         events = self._run_stream(task, reset=reset, max_steps=max_steps)
         if stream:
@@ -95,17 +106,34 @@ async def _run_stream(
         self.memory.add(TaskStep(task=task))
 
         limit = max_steps or self.max_steps
+
+        # Remember the previous step's calls so we can spot an exact repeat.
+        previous_signature: tuple[tuple[str, str], ...] | None = None
+
         for _ in range(limit):
+            if self._interrupt.is_set():
+                self._interrupt.clear()
+                yield FinalEvent(answer="Run interrupted.")
+                return
+
             started = time.monotonic()
             messages = self.memory.to_messages(self.instructions)
             response = await self.model.generate(messages, tools=self._tool_schemas)
 
-            # Forgiving termination: a turn with no tool calls IS the final answer.
+            # Forgiving termination: no tool calls means the content is the answer.
             if not response.tool_calls:
                 self.memory.add(FinalStep(answer=response.content))
                 yield FinalEvent(answer=response.content, usage=response.usage)
                 return
 
+            # Fingerprint this step's calls: (name, canonical JSON args) each.
+            signature = tuple(
+                (tc.name, json.dumps(tc.arguments, sort_keys=True))
+                for tc in response.tool_calls
+            )
+            looping = signature == previous_signature
+            previous_signature = signature
+
             # Announce every call, then run them (concurrently or in order).
             for tool_call in response.tool_calls:
                 yield ToolCallEvent(tool_call=tool_call)
@@ -117,6 +145,10 @@ async def _run_stream(
             else:
                 results = [await self._execute_tool(tc) for tc in response.tool_calls]
 
+            # Same calls as the last step: nudge the model to change approach.
+            if looping:
+                results = [replace(r, content=r.content + _LOOP_NUDGE) for r in results]
+
             for result in results:
                 yield ToolResultEvent(result=result)
 
@@ -169,8 +201,8 @@ async def _execute_tool(self, tool_call: ToolCall) -> ToolResult:
         try:
             output = await tool.call(tool_call.arguments)
         except Exception as exc:
-            # Deliberate broad catch: a failing tool becomes an observation the
-            # model can recover from, never a crash. (Day 5 refines the layers.)
+            # A failing tool becomes an observation the model can recover from
+            # rather than a crash, so catching broadly here is intentional.
             return ToolResult(
                 tool_call_id=tool_call.id,
                 name=tool_call.name,
@@ -183,3 +215,11 @@ async def _execute_tool(self, tool_call: ToolCall) -> ToolResult:
             content=str(output),
             is_error=False,
         )
+
+    def interrupt(self) -> None:
+        """Request a graceful stop before the next step.
+
+        The current run pauses rather than crashing; resume it later with
+        run(..., reset=False), which continues from the steps already in memory.
+        """
+        self._interrupt.set()

src/agentling/memory.py

@@ -1,11 +1,11 @@
 """Typed step records and the agent's memory.
 
 The agent loop records each turn as a typed Step (TaskStep, ActionStep,
-FinalStep) rather than as raw messages. Steps carry metadata — token usage,
-timing, per-tool results and errors — and know how to render themselves back
-into the ChatMessage list the model sees. Memory holds the steps, renders the
-full conversation via to_messages(), and round-trips to JSON for persistence
-and replay.
+FinalStep) rather than as raw messages. Steps carry metadata such as token
+usage, timing, and per-tool results and errors, and they know how to render
+themselves back into the ChatMessage list the model sees. Memory holds the
+steps, renders the full conversation via to_messages(), and round-trips to JSON
+for persistence and replay.
 """
 
 from __future__ import annotations

src/agentling/models.py

@@ -1,10 +1,9 @@
 """Provider-neutral model layer and the OpenAI-compatible adapter.
 
 This module defines the framework's message types (ChatMessage, ToolCall,
-Usage) and streaming types (Delta, ToolCallDelta), the Model protocol that the
-agent loop depends on, and OpenAIModel — an async adapter for any
-OpenAI-compatible chat-completions endpoint, with rate-limit retries,
-streaming, and token-usage capture.
+Usage) and streaming types (Delta, ToolCallDelta), the Model protocol the agent
+loop depends on, and OpenAIModel: an async adapter for any OpenAI-compatible
+chat-completions endpoint, with retries, streaming, and token-usage capture.
 """
 
 from __future__ import annotations
@@ -190,10 +189,14 @@ async def stream(
                 yield delta
 
     async def _create_with_retry(self, **kwargs: Any) -> Any:
-        """Call the OpenAI chat-completions API with rate-limit retries.
-
-        Retries cover establishing the request, including opening a stream.
-        A 429 raised mid-stream (after chunks have been yielded) is not retried.
+        """Call the OpenAI chat-completions API, retrying transient failures.
+
+        Rate limits, connection/timeout errors, and 5xx responses are retried
+        with exponential backoff, then re-raised once max_retries is exhausted.
+        That is the agent's one fatal path. Permanent errors such as a bad
+        request or bad auth are not retried and propagate immediately. Retries
+        cover establishing the request, including opening a stream; a failure
+        that happens mid-stream is not retried.
         """
 
         total_attempts = max(1, self.max_retries + 1)
@@ -202,7 +205,11 @@ async def _create_with_retry(self, **kwargs: Any) -> Any:
             try:
                 return await self.client.chat.completions.create(**kwargs)
 
-            except openai.RateLimitError:
+            except (
+                openai.RateLimitError,
+                openai.APIConnectionError,  # also covers APITimeoutError
+                openai.InternalServerError,
+            ):
                 if attempt == total_attempts - 1:
                     raise
 

tests/test_agent.py

@@ -189,3 +189,135 @@ async def test_max_steps_forces_a_final_answer() -> None:
 
     assert await agent.run("loop forever") == "forced final"
     assert len(model.calls) == 3  # 2 loop steps + 1 forced tool-free answer
+
+
+def _tool_turn(call_id: str, name: str, **arguments: Any) -> ChatMessage:
+    """A model turn that requests a single tool call."""
+    return _assistant(
+        tool_calls=[ToolCall(id=call_id, name=name, arguments=arguments)]
+    )
+
+
+# --------------------------------------------------------------------------- #
+# Loop detector
+# --------------------------------------------------------------------------- #
+_REPEAT_MARKER = "already made this exact tool call"
+
+
+async def test_loop_detector_nudges_on_repeat() -> None:
+    model = FakeModel(
+        [
+            _tool_turn("c1", "add", a=2, b=3),
+            _tool_turn("c2", "add", a=2, b=3),  # identical (name, args)
+            _assistant(content="done"),
+        ]
+    )
+    agent = Agent(model=model, tools=[add])
+    await agent.run("repeat the same call")
+
+    first, second = agent.memory.steps[1], agent.memory.steps[2]
+    assert isinstance(first, ActionStep)
+    assert isinstance(second, ActionStep)
+    assert _REPEAT_MARKER not in first.tool_results[0].content
+    assert _REPEAT_MARKER in second.tool_results[0].content
+
+
+async def test_loop_detector_ignores_different_args() -> None:
+    model = FakeModel(
+        [
+            _tool_turn("c1", "add", a=2, b=3),
+            _tool_turn("c2", "add", a=4, b=5),  # different args -> not a loop
+            _assistant(content="done"),
+        ]
+    )
+    agent = Agent(model=model, tools=[add])
+    await agent.run("two different sums")
+
+    for step in agent.memory.steps:
+        if isinstance(step, ActionStep):
+            assert _REPEAT_MARKER not in step.tool_results[0].content
+
+
+# --------------------------------------------------------------------------- #
+# Self-heal: unknown tool and always-raising tool
+# --------------------------------------------------------------------------- #
+async def test_unknown_tool_recovers() -> None:
+    model = FakeModel([_tool_turn("c1", "ghost"), _assistant(content="recovered")])
+    agent = Agent(model=model)  # only final_answer is registered
+    answer = await agent.run("call a missing tool")
+
+    assert answer == "recovered"
+    action = agent.memory.steps[1]
+    assert isinstance(action, ActionStep)
+    assert action.tool_results[0].is_error is True
+    assert "Unknown tool" in action.tool_results[0].content
+
+
+async def test_always_raising_tool_recovers() -> None:
+    @tool
+    def boom() -> str:
+        """Always raises."""
+        raise RuntimeError("nope")
+
+    model = FakeModel(
+        [
+            _tool_turn("c1", "boom"),
+            _tool_turn("c2", "final_answer", answer="handled"),
+        ]
+    )
+    agent = Agent(model=model, tools=[boom])
+    answer = await agent.run("trigger boom")
+
+    assert answer == "handled"
+    action = agent.memory.steps[1]
+    assert isinstance(action, ActionStep)
+    assert action.tool_results[0].is_error is True
+    assert "RuntimeError" in action.tool_results[0].content
+
+
+# --------------------------------------------------------------------------- #
+# Interruption + resume
+# --------------------------------------------------------------------------- #
+async def test_interrupt_stops_run_gracefully() -> None:
+    model = FakeModel(
+        [_tool_turn("c1", "add", a=1, b=1), _assistant(content="unreached")]
+    )
+    agent = Agent(model=model, tools=[add])
+
+    fired: list[bool] = []
+
+    def interrupt_after_first(step: object) -> None:
+        if not fired:
+            fired.append(True)
+            agent.interrupt()
+
+    agent.step_callbacks.append(interrupt_after_first)
+
+    assert await agent.run("loop") == "Run interrupted."
+    assert isinstance(agent.memory.steps[-1], ActionStep)  # paused before a FinalStep
+    assert len(model.calls) == 1
+
+
+async def test_interrupt_then_resume_completes() -> None:
+    model = FakeModel(
+        [
+            _tool_turn("c1", "add", a=1, b=1),
+            _assistant(content="finished on resume"),
+        ]
+    )
+    agent = Agent(model=model, tools=[add])
+
+    fired: list[bool] = []
+
+    def interrupt_once(step: object) -> None:
+        if not fired:
+            fired.append(True)
+            agent.interrupt()
+
+    agent.step_callbacks.append(interrupt_once)
+
+    assert await agent.run("start") == "Run interrupted."
+    # reset=False keeps the interrupted run's memory; the loop resumes from it.
+    assert await agent.run("continue", reset=False) == "finished on resume"
+    assert len(model.calls) == 2
+    assert sum(isinstance(s, TaskStep) for s in agent.memory.steps) == 2

tests/test_models.py

@@ -288,3 +288,41 @@ async def always_fails(**kwargs: Any) -> Any:
 
     with pytest.raises(openai.RateLimitError):
         await model.generate([ChatMessage(role="user", content="hi")])
+
+
+async def test_retry_on_transient_api_error() -> None:
+    calls = 0
+    result = SimpleNamespace(
+        choices=[
+            SimpleNamespace(message=SimpleNamespace(content="ok", tool_calls=None))
+        ],
+        usage=SimpleNamespace(prompt_tokens=1, completion_tokens=1),
+    )
+
+    async def flaky(**kwargs: Any) -> Any:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise openai.APIConnectionError.__new__(openai.APIConnectionError)
+        return result
+
+    model = _make_model(flaky)
+    msg = await model.generate([ChatMessage(role="user", content="hi")])
+
+    assert msg.content == "ok"
+    assert calls == 2  # transient error retried, then succeeded
+
+
+async def test_permanent_error_is_not_retried() -> None:
+    calls = 0
+
+    async def always_bad(**kwargs: Any) -> Any:
+        nonlocal calls
+        calls += 1
+        raise openai.BadRequestError.__new__(openai.BadRequestError)
+
+    model = _make_model(always_bad)
+
+    with pytest.raises(openai.BadRequestError):
+        await model.generate([ChatMessage(role="user", content="hi")])
+    assert calls == 1  # permanent error fails fast — no retry