From 54a592c6c6c0e0554b56695e2d72361f9229d105 Mon Sep 17 00:00:00 2001 From: "abhishek.rdx" Date: Thu, 29 Oct 2026 10:43:30 +0530 Subject: [PATCH] feat(agents): human-in-the-loop interrupts and resume (#296) interrupt(node_id, payload) inside a tool pauses a ReActAgent run and returns termination_reason="interrupted" with an interrupt_payload. resume()/aresume() re-run the paused tool so interrupt() returns the human decision, record it as that tool call's result, finish the rest of the batch and continue the loop. With a checkpointer the paused state is saved and resume(thread_id=...) works from a fresh agent; run() refuses a paused thread. Ported onto the current agent loop (keeps the #334 checkpointer and #345 shared state) from Uzmaa Khan's #332, with run/arun/resume sharing one loop. Co-authored-by: Uzmaa7 <135528011+Uzmaa7@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 7 + src/dynavec/agents/__init__.py | 4 + src/dynavec/agents/base.py | 45 ++- src/dynavec/agents/react.py | 524 ++++++++++++++++++++------------- src/dynavec/exceptions.py | 16 + tests/test_hitl_interrupts.py | 385 ++++++++++++++++++++++++ 6 files changed, 778 insertions(+), 203 deletions(-) create mode 100644 tests/test_hitl_interrupts.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e924037..3e792d3 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/src/dynavec/agents/__init__.py b/src/dynavec/agents/__init__.py index 750a278..4098fbc 100644 --- a/src/dynavec/agents/__init__.py +++ b/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", diff --git a/src/dynavec/agents/base.py b/src/dynavec/agents/base.py index fe4e84a..888e92e 100644 --- a/src/dynavec/agents/base.py +++ b/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) diff --git a/src/dynavec/agents/react.py b/src/dynavec/agents/react.py index d8be15c..7bd201d 100644 --- a/src/dynavec/agents/react.py +++ b/src/dynavec/agents/react.py @@ -7,7 +7,8 @@ from typing import TYPE_CHECKING, Any from ..chat.base import ChatModel, Message, Tool, ToolCall -from .base import AgentResult, AgentStep, AgentTool +from ..exceptions import NodeInterrupt +from .base import AgentResult, AgentStep, AgentTool, resume_with from .registry import ToolRegistry from .state import AgentState @@ -68,6 +69,30 @@ def _dict_to_step(d: dict[str, Any]) -> AgentStep: ) +def _pending_tool_calls(messages: list[Message]) -> list[ToolCall]: + """Tool calls in the last assistant message that have no tool reply yet.""" + for i in range(len(messages) - 1, -1, -1): + msg = messages[i] + if msg.role == "assistant": + answered = {m.tool_call_id for m in messages[i + 1 :] if m.role == "tool"} + return [tc for tc in msg.tool_calls if tc.id not in answered] + return [] + + +def _completed_step(messages: list[Message], step_number: int) -> AgentStep: + """Build the AgentStep for the last assistant tool batch once all calls are answered.""" + for i in range(len(messages) - 1, -1, -1): + msg = messages[i] + if msg.role == "assistant": + return AgentStep( + step_number=step_number, + thought=msg.content, + tool_calls=list(msg.tool_calls), + observations=[m.content or "" for m in messages[i + 1 :] if m.role == "tool"], + ) + return AgentStep(step_number=step_number) + + class ReActAgent: """A ReAct-style agent that alternates between thinking, calling tools, and observing results.""" @@ -173,57 +198,139 @@ def _finalize_result(self, steps: list[AgentStep], total_tool_calls: int) -> Age tool_calls_count=total_tool_calls, ) - def run( + # ------------------------------------------------------------------ checkpoints + def _save( self, + thread_id: str | None, goal: str, + messages: list[Message], + steps: list[AgentStep], + total_tool_calls: int, *, - thread_id: str | None = None, - resume_from: str | None = None, - state: AgentState | None = None, - **kwargs: Any, - ) -> AgentResult: - """Execute the ReAct loop synchronously until goal completion or max_steps.""" + node_id: str, + step: int, + **extra: Any, + ) -> None: + if self.checkpointer is None or thread_id is None: + return + self.checkpointer.put( + thread_id, + state={ + "goal": goal, + "messages": [_message_to_dict(m) for m in messages], + "steps": [_step_to_dict(s) for s in steps], + "total_tool_calls": total_tool_calls, + **extra, + }, + node_id=node_id, + step=step, + ) + + def _restore( + self, thread_id: str | None, resume_from: str | None + ) -> tuple[list[Message], list[AgentStep], int, int, dict[str, Any] | None]: + """Load (messages, steps, total_tool_calls, start_step, state) from a checkpoint.""" + if self.checkpointer is None or thread_id is None: + return [], [], 0, 1, None + existing_cp = self.checkpointer.get(thread_id, resume_from) + if existing_cp is None: + return [], [], 0, 1, None + state = existing_cp.state messages: list[Message] = [] steps: list[AgentStep] = [] total_tool_calls = 0 - start_step = 1 - - if self.checkpointer is not None and thread_id is not None: - existing_cp = self.checkpointer.get(thread_id, resume_from) - - if existing_cp is not None: - checkpoint_state = existing_cp.state - if checkpoint_state is not None: - if "messages" in checkpoint_state: - messages = [ - _dict_to_message(m) - for m in checkpoint_state["messages"] - ] - if "steps" in checkpoint_state: - steps = [ - _dict_to_step(s) - for s in checkpoint_state["steps"] - ] - total_tool_calls = int( - checkpoint_state.get("total_tool_calls", 0) - ) - start_step = existing_cp.step + 1 + if state is not None: + messages = [_dict_to_message(m) for m in state.get("messages", [])] + steps = [_dict_to_step(s) for s in state.get("steps", [])] + total_tool_calls = int(state.get("total_tool_calls", 0)) + return messages, steps, total_tool_calls, existing_cp.step + 1, state + + # ------------------------------------------------------------------ interrupts + def _interrupted( + self, + exc: NodeInterrupt, + goal: str, + thread_id: str | None, + messages: list[Message], + steps: list[AgentStep], + total_tool_calls: int, + step_idx: int, + ) -> AgentResult: + self._save( + thread_id, + goal, + messages, + steps, + total_tool_calls, + node_id="interrupt", + step=step_idx, + finished=False, + interrupted={"node_id": exc.node_id, "payload": exc.payload, "step": step_idx}, + ) + return AgentResult( + output=f"Interrupted at {exc.node_id!r}: awaiting human input.", + steps=list(steps), + finished=False, + termination_reason="interrupted", + total_steps=step_idx, + tool_calls_count=total_tool_calls, + interrupt_payload={ + "thread_id": thread_id, + "node_id": exc.node_id, + "payload": exc.payload, + "messages": list(messages), + "steps": list(steps), + "total_tool_calls": total_tool_calls, + "start_step": step_idx, + }, + ) - if not messages: - messages = self._init_messages(goal, state) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [], - "total_tool_calls": 0, - }, - node_id="init", - step=0, + def _prepare_resume( + self, + messages: list[Message] | None, + start_step: int | None, + thread_id: str | None, + steps: list[AgentStep] | None, + total_tool_calls: int | None, + ) -> tuple[str, list[Message], list[AgentStep], int, int, list[ToolCall]]: + if messages is None: + if self.checkpointer is None or thread_id is None: + raise ValueError( + "resume() needs the interrupted run's messages, or a checkpointer and thread_id." ) - + cp = self.checkpointer.get(thread_id) + state = (cp.state if cp is not None else None) or {} + paused = state.get("interrupted") + if not paused: + raise ValueError(f"Thread {thread_id!r} has no paused run to resume.") + run_messages = [_dict_to_message(m) for m in state.get("messages", [])] + run_steps = [_dict_to_step(s) for s in state.get("steps", [])] + total = int(state.get("total_tool_calls", 0)) + start = int(paused["step"]) + goal = str(state.get("goal") or "") + else: + run_messages = list(messages) + run_steps = list(steps or []) + total = int(total_tool_calls or 0) + start = start_step if start_step is not None else len(run_steps) + 1 + goal = next((m.content or "" for m in run_messages if m.role == "user"), "") + + pending = _pending_tool_calls(run_messages) + if not pending: + raise ValueError("Nothing to resume: no tool call is waiting for a human decision.") + return goal, run_messages, run_steps, total, start, pending + + # ------------------------------------------------------------------ loops + def _loop( + self, + goal: str, + thread_id: str | None, + messages: list[Message], + steps: list[AgentStep], + total_tool_calls: int, + start_step: int, + kwargs: dict[str, Any], + ) -> AgentResult: for step_idx in range(start_step, self.max_steps + 1): chat_res = self.model.invoke( messages, @@ -234,27 +341,12 @@ def run( messages.append(msg) if not msg.tool_calls: - step = AgentStep( - step_number=step_idx, - thought=msg.content, - tool_calls=[], - observations=[], + steps.append(AgentStep(step_number=step_idx, thought=msg.content)) + self._save( + thread_id, goal, messages, steps, total_tool_calls, + node_id=f"step_{step_idx}", step=step_idx, + output=msg.content or "", finished=True, ) - steps.append(step) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": msg.content or "", - "finished": True, - }, - node_id=f"step_{step_idx}", - step=step_idx, - ) return AgentResult( output=msg.content or "", steps=steps, @@ -265,97 +357,44 @@ def run( ) total_tool_calls += len(msg.tool_calls) - step_observations = self._process_tool_calls(list(msg.tool_calls), messages) + try: + step_observations = self._process_tool_calls(list(msg.tool_calls), messages) + except NodeInterrupt as exc: + return self._interrupted( + exc, goal, thread_id, messages, steps, total_tool_calls, step_idx + ) - step = AgentStep( - step_number=step_idx, - thought=msg.content, - tool_calls=list(msg.tool_calls), - observations=step_observations, - ) - steps.append(step) - - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "finished": False, - }, - node_id=f"step_{step_idx}", - step=step_idx, + steps.append( + AgentStep( + step_number=step_idx, + thought=msg.content, + tool_calls=list(msg.tool_calls), + observations=step_observations, ) + ) + self._save( + thread_id, goal, messages, steps, total_tool_calls, + node_id=f"step_{step_idx}", step=step_idx, finished=False, + ) res = self._finalize_result(steps, total_tool_calls) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": res.output, - "finished": False, - }, - node_id="max_steps_reached", - step=self.max_steps, - ) + self._save( + thread_id, goal, messages, steps, total_tool_calls, + node_id="max_steps_reached", step=self.max_steps, + output=res.output, finished=False, + ) return res - async def arun( + async def _aloop( self, goal: str, - *, - thread_id: str | None = None, - resume_from: str | None = None, - state: AgentState | None = None, - **kwargs: Any, + thread_id: str | None, + messages: list[Message], + steps: list[AgentStep], + total_tool_calls: int, + start_step: int, + kwargs: dict[str, Any], ) -> AgentResult: - """Execute the ReAct loop asynchronously.""" - messages: list[Message] = [] - steps: list[AgentStep] = [] - total_tool_calls = 0 - start_step = 1 - - if self.checkpointer is not None and thread_id is not None: - existing_cp = self.checkpointer.get(thread_id, resume_from) - if existing_cp is not None: - checkpoint_state = existing_cp.state - if checkpoint_state is not None: - if "messages" in checkpoint_state: - messages = [ - _dict_to_message(m) - for m in checkpoint_state["messages"] - ] - if "steps" in checkpoint_state: - steps = [ - _dict_to_step(s) - for s in checkpoint_state["steps"] - ] - total_tool_calls = int( - checkpoint_state.get("total_tool_calls", 0) - ) - start_step = existing_cp.step + 1 - - if not messages: - messages = self._init_messages(goal, state) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [], - "total_tool_calls": 0, - }, - node_id="init", - step=0, - ) - for step_idx in range(start_step, self.max_steps + 1): chat_res = await self.model.ainvoke( messages, @@ -366,27 +405,12 @@ async def arun( messages.append(msg) if not msg.tool_calls: - step = AgentStep( - step_number=step_idx, - thought=msg.content, - tool_calls=[], - observations=[], + steps.append(AgentStep(step_number=step_idx, thought=msg.content)) + self._save( + thread_id, goal, messages, steps, total_tool_calls, + node_id=f"step_{step_idx}", step=step_idx, + output=msg.content or "", finished=True, ) - steps.append(step) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": msg.content or "", - "finished": True, - }, - node_id=f"step_{step_idx}", - step=step_idx, - ) return AgentResult( output=msg.content or "", steps=steps, @@ -397,43 +421,141 @@ async def arun( ) total_tool_calls += len(msg.tool_calls) - step_observations = await self._aprocess_tool_calls(list(msg.tool_calls), messages) + try: + step_observations = await self._aprocess_tool_calls( + list(msg.tool_calls), messages + ) + except NodeInterrupt as exc: + return self._interrupted( + exc, goal, thread_id, messages, steps, total_tool_calls, step_idx + ) - step = AgentStep( - step_number=step_idx, - thought=msg.content, - tool_calls=list(msg.tool_calls), - observations=step_observations, - ) - steps.append(step) - - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "finished": False, - }, - node_id=f"step_{step_idx}", - step=step_idx, + steps.append( + AgentStep( + step_number=step_idx, + thought=msg.content, + tool_calls=list(msg.tool_calls), + observations=step_observations, ) + ) + self._save( + thread_id, goal, messages, steps, total_tool_calls, + node_id=f"step_{step_idx}", step=step_idx, finished=False, + ) res = self._finalize_result(steps, total_tool_calls) - if self.checkpointer is not None and thread_id is not None: - self.checkpointer.put( - thread_id, - state={ - "goal": goal, - "messages": [_message_to_dict(m) for m in messages], - "steps": [_step_to_dict(s) for s in steps], - "total_tool_calls": total_tool_calls, - "output": res.output, - "finished": False, - }, - node_id="max_steps_reached", - step=self.max_steps, - ) + self._save( + thread_id, goal, messages, steps, total_tool_calls, + node_id="max_steps_reached", step=self.max_steps, + output=res.output, finished=False, + ) return res + + # ------------------------------------------------------------------ public API + def _start( + self, goal: str, thread_id: str | None, resume_from: str | None, state: AgentState | None + ) -> tuple[list[Message], list[AgentStep], int, int]: + messages, steps, total_tool_calls, start_step, cp_state = self._restore( + thread_id, resume_from + ) + if cp_state and cp_state.get("interrupted"): + raise ValueError( + f"Thread {thread_id!r} is paused waiting for human input; " + "call resume(thread_id=..., human_decision=...) instead of run()." + ) + if not messages: + messages = self._init_messages(goal, state) + self._save(thread_id, goal, messages, [], 0, node_id="init", step=0) + return messages, steps, total_tool_calls, start_step + + def run( + self, + goal: str, + *, + thread_id: str | None = None, + resume_from: str | None = None, + state: AgentState | None = None, + **kwargs: Any, + ) -> AgentResult: + """Execute the ReAct loop until it completes, hits max_steps, or a tool interrupts.""" + messages, steps, total, start = self._start(goal, thread_id, resume_from, state) + return self._loop(goal, thread_id, messages, steps, total, start, kwargs) + + async def arun( + self, + goal: str, + *, + thread_id: str | None = None, + resume_from: str | None = None, + state: AgentState | None = None, + **kwargs: Any, + ) -> AgentResult: + """Execute the ReAct loop asynchronously.""" + messages, steps, total, start = self._start(goal, thread_id, resume_from, state) + return await self._aloop(goal, thread_id, messages, steps, total, start, kwargs) + + def resume( + self, + messages: list[Message] | None = None, + human_decision: Any = None, + *, + start_step: int | None = None, + thread_id: str | None = None, + steps: list[AgentStep] | None = None, + total_tool_calls: int | None = None, + **kwargs: Any, + ) -> AgentResult: + """Continue an interrupted run with the human's decision. + + Pass the ``messages`` / ``start_step`` (and optionally ``steps``, + ``total_tool_calls``) from ``AgentResult.interrupt_payload``, or just + ``thread_id`` when the agent has a checkpointer. The paused tool is re-run and its + :func:`~dynavec.agents.interrupt` call returns ``human_decision``. + """ + goal, messages, steps, total, start, pending = self._prepare_resume( + messages, start_step, thread_id, steps, total_tool_calls + ) + try: + with resume_with(human_decision): + obs = self._dispatch_tool_call(pending[0]) + messages.append(Message(role="tool", content=obs, tool_call_id=pending[0].id)) + self._process_tool_calls(pending[1:], messages) + except NodeInterrupt as exc: + return self._interrupted(exc, goal, thread_id, messages, steps, total, start) + + steps.append(_completed_step(messages, start)) + self._save( + thread_id, goal, messages, steps, total, + node_id=f"step_{start}", step=start, finished=False, + ) + return self._loop(goal, thread_id, messages, steps, total, start + 1, kwargs) + + async def aresume( + self, + messages: list[Message] | None = None, + human_decision: Any = None, + *, + start_step: int | None = None, + thread_id: str | None = None, + steps: list[AgentStep] | None = None, + total_tool_calls: int | None = None, + **kwargs: Any, + ) -> AgentResult: + """Asynchronous :meth:`resume`.""" + goal, messages, steps, total, start, pending = self._prepare_resume( + messages, start_step, thread_id, steps, total_tool_calls + ) + try: + with resume_with(human_decision): + obs = await self._adispatch_tool_call(pending[0]) + messages.append(Message(role="tool", content=obs, tool_call_id=pending[0].id)) + await self._aprocess_tool_calls(pending[1:], messages) + except NodeInterrupt as exc: + return self._interrupted(exc, goal, thread_id, messages, steps, total, start) + + steps.append(_completed_step(messages, start)) + self._save( + thread_id, goal, messages, steps, total, + node_id=f"step_{start}", step=start, finished=False, + ) + return await self._aloop(goal, thread_id, messages, steps, total, start + 1, kwargs) diff --git a/src/dynavec/exceptions.py b/src/dynavec/exceptions.py index 5ae80ff..a8dbc00 100644 --- a/src/dynavec/exceptions.py +++ b/src/dynavec/exceptions.py @@ -2,6 +2,8 @@ from __future__ import annotations +from typing import Any + class DynavecError(Exception): """Base class for all dynavec errors.""" @@ -76,3 +78,17 @@ def __init__(self, feature: str, package: str, extra: str) -> None: self.feature = feature self.package = package self.extra = extra + + +class NodeInterrupt(DynavecError): + """Raised inside a tool (via :func:`dynavec.agents.interrupt`) to pause an agent run. + + The agent catches it and returns an ``AgentResult`` with + ``termination_reason="interrupted"``; ``ReActAgent.resume`` continues the run with + the human's decision. + """ + + def __init__(self, node_id: str, payload: Any = None) -> None: + super().__init__(f"Execution interrupted at {node_id!r}; awaiting human input.") + self.node_id = node_id + self.payload = payload diff --git a/tests/test_hitl_interrupts.py b/tests/test_hitl_interrupts.py new file mode 100644 index 0000000..9000773 --- /dev/null +++ b/tests/test_hitl_interrupts.py @@ -0,0 +1,385 @@ +"""Tests for Human-in-the-loop / interrupts functionality in dynavec agents.""" +from __future__ import annotations + +import asyncio +import json +from collections.abc import AsyncIterator, Iterator +from typing import Any + +from dynavec.agents.base import interrupt, tool +from dynavec.agents.react import ReActAgent +from dynavec.chat.base import ChatChunk, ChatModel, ChatResult, Message, ToolCall + + +class DummyChatModel(ChatModel): + """Simple mock chat model for HITL tests.""" + + def __init__(self, responses: list[ChatResult]) -> None: + super().__init__() + self.responses = responses + self.call_count = 0 + + def invoke( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> ChatResult: + res = self.responses[self.call_count] + self.call_count += 1 + return res + + async def ainvoke( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> ChatResult: + return self.invoke(messages, tools, **kwargs) + + def stream( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> Iterator[ChatChunk]: + yield ChatChunk(content="") + + async def astream( + self, messages: list[Message], tools: list[Any] | None = None, **kwargs: Any + ) -> AsyncIterator[ChatChunk]: + yield ChatChunk(content="") + + +@tool() +def sensitive_action_tool(action: str) -> str: + """Action requiring human approval.""" + interrupt("approval_node", f"Action '{action}' requires approval. Confirm?") + return f"Action '{action}' executed successfully." + + +def test_interrupt_raised_and_caught() -> None: + mock_model = DummyChatModel( + responses=[ + ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall( + id="tc_1", + name="sensitive_action_tool", + arguments=json.dumps({"action": "delete_database"}), + ) + ], + ) + ) + ] + ) + + agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + result = agent.run("Please delete the database.") + + assert result.finished is False + assert result.termination_reason == "interrupted" + assert result.interrupt_payload is not None + + +def test_resume_execution() -> None: + mock_model = DummyChatModel( + responses=[ + ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall( + id="tc_1", + name="sensitive_action_tool", + arguments=json.dumps({"action": "delete_database"}), + ) + ], + ) + ), + ChatResult( + message=Message( + role="assistant", + content="Database action completed.", + ) + ), + ] + ) + + agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + result = agent.run("Please delete the database.") + + assert result.finished is False + assert result.interrupt_payload is not None + + payload = result.interrupt_payload + saved_messages = payload["messages"] + start_step = payload["start_step"] + + resume_agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + final_result = resume_agent.resume( + saved_messages, + start_step=start_step, + human_decision="approved", + ) + + assert final_result.finished is True + assert final_result.termination_reason == "completed" + assert "Database action completed" in final_result.output + + +def test_async_interrupt_and_resume() -> None: + async def _run_async() -> None: + mock_model = DummyChatModel( + responses=[ + ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall( + id="tc_1", + name="sensitive_action_tool", + arguments=json.dumps({"action": "delete_database"}), + ) + ], + ) + ), + ChatResult( + message=Message( + role="assistant", + content="Deployment completed.", + ) + ), + ] + ) + + agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + result = await agent.arun("Deploy latest release.") + + assert result.finished is False + assert result.interrupt_payload is not None + + payload = result.interrupt_payload + assert payload is not None + + resume_agent = ReActAgent(model=mock_model, tools=[sensitive_action_tool]) + final_result = await resume_agent.aresume( + payload["messages"], + start_step=payload["start_step"], + human_decision="approved", + ) + + assert final_result.finished is True + assert final_result.termination_reason == "completed" + + asyncio.run(_run_async()) + + +# --------------------------------------------------------------------------- +# Resume semantics: the paused tool re-runs and interrupt() returns the decision. +# --------------------------------------------------------------------------- + +from dynavec.agents import MemoryCheckpointer, NodeInterrupt # noqa: E402 + +DECISIONS: list[Any] = [] + + +@tool() +def transfer_funds(amount: int) -> str: + """Transfer money; needs a human decision first.""" + decision = interrupt("approve_transfer", {"amount": amount}) + DECISIONS.append(decision) + if decision != "approved": + return f"Transfer of {amount} cancelled by reviewer." + return f"Transferred {amount}." + + +@tool() +def lookup_balance() -> str: + """Read-only lookup.""" + return "balance=500" + + +def _tool_call(call_id: str, name: str, args: dict[str, Any]) -> ChatResult: + return ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ToolCall(id=call_id, name=name, arguments=json.dumps(args))], + ) + ) + + +def _final(text: str) -> ChatResult: + return ChatResult(message=Message(role="assistant", content=text)) + + +def _assert_provider_valid(messages: list[Message]) -> None: + """Every assistant tool call is answered by a tool message before the next turn.""" + for i, m in enumerate(messages): + if m.role == "assistant" and m.tool_calls: + replies = [] + for nxt in messages[i + 1 :]: + if nxt.role != "tool": + break + replies.append(nxt.tool_call_id) + assert sorted(replies) == sorted(tc.id for tc in m.tool_calls) + + +def test_interrupt_payload_carries_node_and_payload() -> None: + model = DummyChatModel([_tool_call("t1", "transfer_funds", {"amount": 120})]) + result = ReActAgent(model=model, tools=[transfer_funds]).run("pay rent", thread_id="th-1") + + assert result.termination_reason == "interrupted" + payload = result.interrupt_payload + assert payload is not None + assert payload["node_id"] == "approve_transfer" + assert payload["payload"] == {"amount": 120} + assert payload["thread_id"] == "th-1" + assert payload["start_step"] == 1 + + +def test_resume_reruns_tool_with_decision_and_records_its_result() -> None: + DECISIONS.clear() + model = DummyChatModel( + [_tool_call("t1", "transfer_funds", {"amount": 120}), _final("Done.")] + ) + agent = ReActAgent(model=model, tools=[transfer_funds]) + paused = agent.run("pay rent") + final = agent.resume( + paused.interrupt_payload["messages"], # type: ignore[index] + human_decision="approved", + start_step=paused.interrupt_payload["start_step"], # type: ignore[index] + ) + + assert DECISIONS == ["approved"] + assert final.termination_reason == "completed" + assert final.steps[0].observations == ["Transferred 120."] + assert final.total_steps == 2 + + +def test_rejected_decision_reaches_the_tool() -> None: + DECISIONS.clear() + model = DummyChatModel( + [_tool_call("t1", "transfer_funds", {"amount": 99}), _final("Cancelled.")] + ) + agent = ReActAgent(model=model, tools=[transfer_funds]) + paused = agent.run("pay") + final = agent.resume( + paused.interrupt_payload["messages"], # type: ignore[index] + human_decision="rejected", + ) + + assert DECISIONS == ["rejected"] + assert final.steps[0].observations == ["Transfer of 99 cancelled by reviewer."] + + +def test_resume_from_checkpoint_in_a_fresh_agent() -> None: + """A paused run survives a process restart when a checkpointer is configured.""" + DECISIONS.clear() + checkpointer = MemoryCheckpointer() + first = ReActAgent( + model=DummyChatModel([_tool_call("t1", "transfer_funds", {"amount": 7})]), + tools=[transfer_funds], + checkpointer=checkpointer, + ) + assert first.run("pay", thread_id="th-x").termination_reason == "interrupted" + + # New agent instance, nothing carried over except the checkpointer. + second = ReActAgent( + model=DummyChatModel([_final("Paid.")]), + tools=[transfer_funds], + checkpointer=checkpointer, + ) + final = second.resume(thread_id="th-x", human_decision="approved") + + assert final.termination_reason == "completed" + assert final.output == "Paid." + assert DECISIONS == ["approved"] + latest = checkpointer.get("th-x") + assert latest is not None and latest.state["finished"] is True + assert "interrupted" not in latest.state + + +def test_run_refuses_a_paused_thread() -> None: + checkpointer = MemoryCheckpointer() + agent = ReActAgent( + model=DummyChatModel([_tool_call("t1", "transfer_funds", {"amount": 1})]), + tools=[transfer_funds], + checkpointer=checkpointer, + ) + agent.run("pay", thread_id="th-p") + try: + agent.run("pay", thread_id="th-p") + except ValueError as exc: + assert "resume" in str(exc) + else: # pragma: no cover + raise AssertionError("run() on a paused thread should raise") + + +def test_resume_without_paused_tool_call_raises() -> None: + agent = ReActAgent(model=DummyChatModel([]), tools=[transfer_funds]) + try: + agent.resume([Message(role="user", content="hi")], human_decision="approved") + except ValueError as exc: + assert "Nothing to resume" in str(exc) + else: # pragma: no cover + raise AssertionError("expected ValueError") + + +def test_second_interrupt_in_same_batch_pauses_again() -> None: + DECISIONS.clear() + batch = ChatResult( + message=Message( + role="assistant", + content="", + tool_calls=[ + ToolCall(id="a", name="lookup_balance", arguments="{}"), + ToolCall(id="b", name="transfer_funds", arguments=json.dumps({"amount": 1})), + ToolCall(id="c", name="transfer_funds", arguments=json.dumps({"amount": 2})), + ], + ) + ) + model = DummyChatModel([batch, _final("All done.")]) + agent = ReActAgent(model=model, tools=[lookup_balance, transfer_funds]) + + first = agent.run("move money") + assert first.termination_reason == "interrupted" + second = agent.resume(first.interrupt_payload["messages"], human_decision="approved") # type: ignore[index] + assert second.termination_reason == "interrupted" + assert second.interrupt_payload["payload"] == {"amount": 2} # type: ignore[index] + final = agent.resume(second.interrupt_payload["messages"], human_decision="rejected") # type: ignore[index] + + assert final.termination_reason == "completed" + assert DECISIONS == ["approved", "rejected"] + assert final.steps[0].observations == [ + "balance=500", + "Transferred 1.", + "Transfer of 2 cancelled by reviewer.", + ] + + +def test_message_history_stays_provider_valid_after_resume() -> None: + seen: list[list[Message]] = [] + + class RecordingModel(DummyChatModel): + def invoke(self, messages, tools=None, **kwargs): # type: ignore[no-untyped-def] + seen.append(list(messages)) + return super().invoke(messages, tools, **kwargs) + + model = RecordingModel([_tool_call("t1", "transfer_funds", {"amount": 5}), _final("ok")]) + agent = ReActAgent(model=model, tools=[transfer_funds]) + paused = agent.run("pay") + agent.resume(paused.interrupt_payload["messages"], human_decision="approved") # type: ignore[index] + + _assert_provider_valid(seen[-1]) + assert seen[-1][-1].role == "tool" + + +def test_tool_errors_still_become_observations() -> None: + @tool() + def broken() -> str: + """Always fails.""" + raise RuntimeError("boom") + + model = DummyChatModel([_tool_call("t1", "broken", {}), _final("handled")]) + result = ReActAgent(model=model, tools=[broken]).run("go") + assert result.termination_reason == "completed" + assert "boom" in result.steps[0].observations[0] + assert issubclass(NodeInterrupt, Exception)