feat(agents): human-in-the-loop interrupts and resume (#296)

CHANGELOG.md

@@ -5,6 +5,13 @@ All notable changes to dynavec are documented here. This project adheres to
 
 ## [Unreleased]
 
+### Added
+- **Human-in-the-loop interrupts** (#296) — call `dynavec.agents.interrupt(node_id, payload)`
+  inside a tool to pause a `ReActAgent` run (`termination_reason="interrupted"`, details in
+  `AgentResult.interrupt_payload`). `resume()` / `aresume()` re-run the paused tool and
+  `interrupt()` returns the human decision. With a checkpointer, a paused run resumes from
+  `thread_id` alone in a new process.
+
 ## [0.7.0] - 2026-10-01
 
 Concurrency and scale: a native async client, a lexical/hybrid retrieval path,

src/dynavec/agents/__init__.py

@@ -8,12 +8,14 @@
     DynamoDBCheckpointer,
     MemoryCheckpointer,
 )
+from ..exceptions import NodeInterrupt
 from .base import (
     AgentResult,
     AgentStep,
     AgentTool,
     Plan,
     PlanStep,
+    interrupt,
     tool,
 )
 from .connectors import (
@@ -70,6 +72,7 @@
     "HandoffRequest",
     "MaxIterationsExceededError",
     "MemoryCheckpointer",
+    "NodeInterrupt",
     "NodeTrace",
     "Plan",
     "PlanStep",
@@ -82,6 +85,7 @@
     "create_python_code_tool",
     "default_reducer",
     "default_registry",
+    "interrupt",
     "mcp_tools_from_session",
     "mcp_tools_from_session_async",
     "tool",

src/dynavec/agents/base.py

@@ -4,11 +4,14 @@
 
 import inspect
 import json
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
 from dataclasses import dataclass, field
 from typing import Any
 
 from ..chat.base import Tool, ToolCall
+from ..exceptions import NodeInterrupt
 
 
 @dataclass
@@ -28,9 +31,12 @@ class AgentResult:
     output: str
     steps: list[AgentStep] = field(default_factory=list)
     finished: bool = True
-    termination_reason: str = "completed"  # "completed", "max_steps_reached", "error"
+    termination_reason: str = "completed"  # "completed", "max_steps_reached", "interrupted"
     total_steps: int = 0
     tool_calls_count: int = 0
+    # Set when termination_reason == "interrupted": node_id, payload, and what
+    # ReActAgent.resume needs (messages, steps, total_tool_calls, start_step).
+    interrupt_payload: dict[str, Any] | None = None
 
 
 @dataclass
@@ -191,6 +197,8 @@ def execute(self, arguments: dict[str, Any] | str | None = None) -> str:
             if isinstance(result, str):
                 return result
             return json.dumps(result, ensure_ascii=False)
+        except NodeInterrupt:
+            raise
         except Exception as exc:  # noqa: BLE001
             return f"Error executing tool {self.name!r}: {exc}"
 
@@ -209,6 +217,8 @@ async def aexecute(self, arguments: dict[str, Any] | str | None = None) -> str:
             if isinstance(result, str):
                 return result
             return json.dumps(result, ensure_ascii=False)
+        except NodeInterrupt:
+            raise
         except Exception as exc:  # noqa: BLE001
             return f"Error executing tool {self.name!r}: {exc}"
 
@@ -234,3 +244,34 @@ def decorator(fn: Callable[..., Any]) -> AgentTool:
         )
 
     return decorator
+
+
+_UNSET: Any = object()
+_RESUME_VALUE: ContextVar[Any] = ContextVar("dynavec_resume_value", default=_UNSET)
+
+
+def interrupt(node_id: str, payload: Any = None) -> Any:
+    """Pause the agent run for human input, or return the human's decision on resume.
+
+    Call this inside a tool. On the first run it raises :class:`NodeInterrupt`, and the
+    agent returns ``termination_reason="interrupted"`` with ``payload`` in
+    ``AgentResult.interrupt_payload``. ``ReActAgent.resume(..., human_decision=...)``
+    re-runs the paused tool, and this call then returns ``human_decision``.
+
+    Code in the tool before ``interrupt()`` runs again on resume, so keep side effects
+    after it.
+    """
+    value = _RESUME_VALUE.get()
+    if value is not _UNSET:
+        return value
+    raise NodeInterrupt(node_id, payload)
+
+
+@contextmanager
+def resume_with(decision: Any) -> Iterator[None]:
+    """Make :func:`interrupt` return ``decision`` while re-running a paused tool."""
+    token = _RESUME_VALUE.set(decision)
+    try:
+        yield
+    finally:
+        _RESUME_VALUE.reset(token)