From db293d55763aee020b62f6db71440897ef76af8b Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Thu, 16 Jul 2026 11:03:58 -0400 Subject: [PATCH 1/5] feat(tasks): add non-terminal task/interrupt + INTERRUPTED status (AGX1-391) Adds a `task/interrupt` RPC method and `POST /tasks/{task_id}/interrupt` route that stop an in-flight agent turn without terminating the task, plus a new non-terminal `INTERRUPTED` task status. Mirrors the `task/cancel` paths but never transitions the task to a terminal state, so the conversation stays continuable. - RPC `task/interrupt` across the api/domain method enums, params, union, validators - widen ACP_TYPE_TO_ALLOWED_RPC_METHODS to SYNC / ASYNC / AGENTIC - authorize as `update`; dispatch + service forward the RPC to the agent (async) and best-effort for sync; never calls _transition_to_terminal - INTERRUPTED status + state machine: RUNNING<->INTERRUPTED and terminal transitions accept INTERRUPTED as a source - alembic migration adding the INTERRUPTED enum value - regenerated openapi.yaml (drives SDK regeneration) - unit + integration tests mirroring task/cancel Part of the Stop-mid-stream + queue-at-boundary effort; see the design doc at teams/sgp/agents/golden_agent/docs/interrupt-and-queue-design.md in agentex-agents. Co-Authored-By: Claude Opus 4.8 (1M context) --- ...dd_interrupted_task_status_a1b2c3d4e5f6.py | 32 +++++ agentex/openapi.yaml | 58 ++++++++ agentex/src/api/routes/agents.py | 24 ++++ agentex/src/api/routes/tasks.py | 24 ++++ agentex/src/api/schemas/agents.py | 1 + agentex/src/api/schemas/agents_rpc.py | 28 +++- agentex/src/api/schemas/tasks.py | 2 + agentex/src/domain/entities/agents_rpc.py | 45 +++++- agentex/src/domain/entities/tasks.py | 4 + .../src/domain/services/agent_acp_service.py | 17 ++- agentex/src/domain/services/task_service.py | 41 ++++++ .../domain/use_cases/agents_acp_use_case.py | 57 +++++++- .../src/domain/use_cases/tasks_use_case.py | 51 ++++++- .../integration/api/tasks/test_tasks_api.py | 67 +++++++++ agentex/tests/unit/api/test_tasks_authz.py | 87 ++++++++++++ ...p_type_backwards_compatibility_use_case.py | 13 +- .../unit/use_cases/test_tasks_use_case.py | 133 ++++++++++++++++++ 17 files changed, 670 insertions(+), 14 deletions(-) create mode 100644 agentex/database/migrations/alembic/versions/2026_07_16_1200_add_interrupted_task_status_a1b2c3d4e5f6.py diff --git a/agentex/database/migrations/alembic/versions/2026_07_16_1200_add_interrupted_task_status_a1b2c3d4e5f6.py b/agentex/database/migrations/alembic/versions/2026_07_16_1200_add_interrupted_task_status_a1b2c3d4e5f6.py new file mode 100644 index 00000000..389e06a5 --- /dev/null +++ b/agentex/database/migrations/alembic/versions/2026_07_16_1200_add_interrupted_task_status_a1b2c3d4e5f6.py @@ -0,0 +1,32 @@ +"""add interrupted task status + +Revision ID: a1b2c3d4e5f6 +Revises: 9a4b8c7d6e5f +Create Date: 2026-07-16 12:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = 'a1b2c3d4e5f6' +down_revision: Union[str, None] = '9a4b8c7d6e5f' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # New non-terminal task status: the current turn was interrupted by the user + # and the task is waiting for the next message (see task/interrupt). + op.execute(""" + ALTER TYPE taskstatus ADD VALUE IF NOT EXISTS 'INTERRUPTED'; + """) + + +def downgrade() -> None: + # Postgres does not support removing a value from an enum type, so there is + # nothing to do on downgrade (mirrors the soft_delete_status migration). + pass diff --git a/agentex/openapi.yaml b/agentex/openapi.yaml index b6c01b29..e43ff873 100644 --- a/agentex/openapi.yaml +++ b/agentex/openapi.yaml @@ -754,6 +754,43 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /tasks/{task_id}/interrupt: + post: + tags: + - Tasks + summary: Interrupt Task + description: Stop the in-flight turn without terminating the task. Transitions + a running task to the non-terminal INTERRUPTED status; the task stays continuable + and the next message or event resumes it. + operationId: interrupt_task_tasks__task_id__interrupt_post + parameters: + - name: task_id + in: path + required: true + schema: + type: string + title: Task Id + requestBody: + content: + application/json: + schema: + anyOf: + - $ref: '#/components/schemas/TaskStatusReasonRequest' + - type: 'null' + title: Request + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/Task' + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /tasks/{task_id}/terminate: post: tags: @@ -4247,11 +4284,13 @@ components: - task/create - message/send - task/cancel + - task/interrupt title: AgentRPCMethod AgentRPCParams: anyOf: - $ref: '#/components/schemas/CreateTaskRequest' - $ref: '#/components/schemas/CancelTaskRequest' + - $ref: '#/components/schemas/InterruptTaskRequest' - $ref: '#/components/schemas/SendMessageRequest' - $ref: '#/components/schemas/SendEventRequest' title: AgentRPCParams @@ -5569,6 +5608,24 @@ components: title: Detail type: object title: HTTPValidationError + InterruptTaskRequest: + properties: + task_id: + anyOf: + - type: string + - type: 'null' + title: Task Id + description: The ID of the task to interrupt. Either this or task_name must + be provided. + task_name: + anyOf: + - type: string + - type: 'null' + title: Task Name + description: The name of the task to interrupt. Either this or task_id must + be provided. + type: object + title: InterruptTaskRequest ListCheckpointsRequest: properties: thread_id: @@ -6743,6 +6800,7 @@ components: - COMPLETED - FAILED - RUNNING + - INTERRUPTED - TERMINATED - TIMED_OUT - DELETED diff --git a/agentex/src/api/routes/agents.py b/agentex/src/api/routes/agents.py index b7d370d3..abd10b60 100644 --- a/agentex/src/api/routes/agents.py +++ b/agentex/src/api/routes/agents.py @@ -478,6 +478,30 @@ async def _authorize_rpc_request( raise ValueError( "Either task_id or task_name must be provided for task/cancel" ) + + case AgentRPCMethod.TASK_INTERRUPT: + # Interrupt is non-terminal (the task stays continuable), so it maps + # to ``update`` rather than the owner-only ``cancel`` used above. + task_id = request.params.task_id + task_name = request.params.task_name + + if task_id is not None: + await check_task_or_collapse_to_404( + authorization_service, + task_id, + AuthorizedOperationType.update, + ) + elif task_name is not None: + existing_task = await task_service.get_task(name=task_name) + await check_task_or_collapse_to_404( + authorization_service, + existing_task.id, + AuthorizedOperationType.update, + ) + else: + raise ValueError( + "Either task_id or task_name must be provided for task/interrupt" + ) case _: raise NotImplementedError( f"_authorize_rpc_request has no case for {request.method}; " diff --git a/agentex/src/api/routes/tasks.py b/agentex/src/api/routes/tasks.py index c4b927fe..86abb22a 100644 --- a/agentex/src/api/routes/tasks.py +++ b/agentex/src/api/routes/tasks.py @@ -275,6 +275,30 @@ async def cancel_task( return Task.model_validate(updated) +@router.post( + "/{task_id}/interrupt", + response_model=Task, + summary="Interrupt Task", + description=( + "Stop the in-flight turn without terminating the task. Transitions a " + "running task to the non-terminal INTERRUPTED status; the task stays " + "continuable and the next message or event resumes it." + ), +) +async def interrupt_task( + # Interrupt is non-terminal (the task remains continuable), so it authorizes + # the editor-allowed ``update`` action — matching the RPC task/interrupt + # path — rather than the owner-only ``cancel`` used by the sibling cancel route. + task_id: DAuthorizedId(AgentexResourceType.task, AuthorizedOperationType.update), + task_use_case: DTaskUseCase, + request: TaskStatusReasonRequest | None = None, +) -> Task: + updated = await task_use_case.interrupt_task( + id=task_id, reason=request.reason if request else None + ) + return Task.model_validate(updated) + + @router.post( "/{task_id}/terminate", response_model=Task, diff --git a/agentex/src/api/schemas/agents.py b/agentex/src/api/schemas/agents.py index 79d92532..de2c3c81 100644 --- a/agentex/src/api/schemas/agents.py +++ b/agentex/src/api/schemas/agents.py @@ -30,6 +30,7 @@ class AgentRPCMethod(str, Enum): TASK_CREATE = "task/create" MESSAGE_SEND = "message/send" TASK_CANCEL = "task/cancel" + TASK_INTERRUPT = "task/interrupt" class AgentInputType(str, Enum): diff --git a/agentex/src/api/schemas/agents_rpc.py b/agentex/src/api/schemas/agents_rpc.py index 404b9bcd..4b4768cd 100644 --- a/agentex/src/api/schemas/agents_rpc.py +++ b/agentex/src/api/schemas/agents_rpc.py @@ -19,6 +19,7 @@ class AgentRPCMethod(str, Enum): TASK_CREATE = "task/create" MESSAGE_SEND = "message/send" TASK_CANCEL = "task/cancel" + TASK_INTERRUPT = "task/interrupt" class CreateTaskRequest(BaseModel): @@ -66,6 +67,23 @@ def validate_task_identifiers(self): return self +class InterruptTaskRequest(BaseModel): + task_id: str | None = Field( + None, + description="The ID of the task to interrupt. Either this or task_name must be provided.", + ) + task_name: str | None = Field( + None, + description="The name of the task to interrupt. Either this or task_id must be provided.", + ) + + @model_validator(mode="after") + def validate_task_identifiers(self): + if self.task_id is not None and self.task_name is not None: + raise ValueError("Cannot provide both task_id and task_name - use only one") + return self + + class SendMessageRequest(BaseModel): task_id: str | None = Field( None, description="The ID of the task that the message was sent to" @@ -111,7 +129,11 @@ def validate_task_identifiers(self): class AgentRPCParams(RootModel): root: ( - CreateTaskRequest | CancelTaskRequest | SendMessageRequest | SendEventRequest + CreateTaskRequest + | CancelTaskRequest + | InterruptTaskRequest + | SendMessageRequest + | SendEventRequest ) = Field(..., description="The parameters for the agent RPC request") @@ -137,6 +159,10 @@ def validate_params_based_on_method(cls, data): data["params"] = AgentRPCParams( root=CancelTaskRequest(**params_data) ) + elif method == AgentRPCMethod.TASK_INTERRUPT: + data["params"] = AgentRPCParams( + root=InterruptTaskRequest(**params_data) + ) elif method == AgentRPCMethod.MESSAGE_SEND: data["params"] = AgentRPCParams( root=SendMessageRequest(**params_data) diff --git a/agentex/src/api/schemas/tasks.py b/agentex/src/api/schemas/tasks.py index 74c8d83d..a79d9fef 100644 --- a/agentex/src/api/schemas/tasks.py +++ b/agentex/src/api/schemas/tasks.py @@ -19,6 +19,8 @@ class TaskStatus(str, Enum): COMPLETED = "COMPLETED" FAILED = "FAILED" RUNNING = "RUNNING" + # Non-terminal: current turn stopped by the user, task still continuable. + INTERRUPTED = "INTERRUPTED" TERMINATED = "TERMINATED" TIMED_OUT = "TIMED_OUT" DELETED = "DELETED" diff --git a/agentex/src/domain/entities/agents_rpc.py b/agentex/src/domain/entities/agents_rpc.py index bdc98268..ef99fa38 100644 --- a/agentex/src/domain/entities/agents_rpc.py +++ b/agentex/src/domain/entities/agents_rpc.py @@ -7,6 +7,7 @@ AgentRPCRequest, CancelTaskRequest, CreateTaskRequest, + InterruptTaskRequest, ) from src.domain.entities.agents import ACPType, AgentEntity from src.domain.entities.events import EventEntity @@ -26,6 +27,7 @@ class AgentRPCMethod(str, Enum): TASK_CREATE = "task/create" TASK_CANCEL = "task/cancel" + TASK_INTERRUPT = "task/interrupt" MESSAGE_SEND = "message/send" EVENT_SEND = "event/send" @@ -85,16 +87,32 @@ class CancelTaskParams(BaseModel): task: TaskEntity = Field(..., description="The task that was cancelled") +class InterruptTaskParams(BaseModel): + """Parameters for task/interrupt method""" + + agent: AgentEntity = Field( + ..., + description="The agent that the task was sent to", + ) + task: TaskEntity = Field(..., description="The task that was interrupted") + + ACP_TYPE_TO_ALLOWED_RPC_METHODS = { - ACPType.SYNC: [AgentRPCMethod.MESSAGE_SEND, AgentRPCMethod.TASK_CREATE], + ACPType.SYNC: [ + AgentRPCMethod.MESSAGE_SEND, + AgentRPCMethod.TASK_CREATE, + AgentRPCMethod.TASK_INTERRUPT, + ], ACPType.AGENTIC: [ AgentRPCMethod.TASK_CREATE, AgentRPCMethod.TASK_CANCEL, + AgentRPCMethod.TASK_INTERRUPT, AgentRPCMethod.EVENT_SEND, ], ACPType.ASYNC: [ AgentRPCMethod.TASK_CREATE, AgentRPCMethod.TASK_CANCEL, + AgentRPCMethod.TASK_INTERRUPT, AgentRPCMethod.EVENT_SEND, ], } @@ -132,6 +150,23 @@ def validate_task_identifiers(self): return self +class InterruptTaskRequestEntity(BaseModel): + task_id: str | None = Field( + None, + description="The ID of the task to interrupt. Either this or task_name must be provided.", + ) + task_name: str | None = Field( + None, + description="The name of the task to interrupt. Either this or task_id must be provided.", + ) + + @model_validator(mode="after") + def validate_task_identifiers(self): + if self.task_id is not None and self.task_name is not None: + raise ValueError("Cannot provide both task_id and task_name - use only one") + return self + + class SendMessageRequestEntity(BaseModel): task_id: str | None = Field( None, description="The ID of the task that the message was sent to" @@ -180,6 +215,7 @@ class AgentRPCRequestEntity(JSONRPCRequest): params: ( CreateTaskRequestEntity | CancelTaskRequestEntity + | InterruptTaskRequestEntity | SendMessageRequestEntity | SendEventRequestEntity ) = Field(..., description="The parameters for the agent RPC request") @@ -201,6 +237,13 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self: task_id=request.params.root.task_id, task_name=request.params.root.task_name, ) + elif request.method == AgentRPCMethod.TASK_INTERRUPT and isinstance( + request.params.root, InterruptTaskRequest + ): + params = InterruptTaskRequestEntity( + task_id=request.params.root.task_id, + task_name=request.params.root.task_name, + ) elif request.method == AgentRPCMethod.MESSAGE_SEND: content_entity = convert_task_message_content_to_entity( content=request.params.root.content.root diff --git a/agentex/src/domain/entities/tasks.py b/agentex/src/domain/entities/tasks.py index 6a1ffce7..76e93cf9 100644 --- a/agentex/src/domain/entities/tasks.py +++ b/agentex/src/domain/entities/tasks.py @@ -20,6 +20,10 @@ class TaskStatus(str, Enum): COMPLETED = "COMPLETED" FAILED = "FAILED" RUNNING = "RUNNING" + # Non-terminal resting state: the current turn was stopped by the user and the + # task is waiting for the next message. Distinct from RUNNING (a turn is + # actively in flight) and from the terminal statuses below. + INTERRUPTED = "INTERRUPTED" TERMINATED = "TERMINATED" TIMED_OUT = "TIMED_OUT" DELETED = "DELETED" diff --git a/agentex/src/domain/services/agent_acp_service.py b/agentex/src/domain/services/agent_acp_service.py index db22063a..0d3495af 100644 --- a/agentex/src/domain/services/agent_acp_service.py +++ b/agentex/src/domain/services/agent_acp_service.py @@ -13,6 +13,7 @@ AgentRPCMethod, CancelTaskParams, CreateTaskParams, + InterruptTaskParams, SendEventParams, SendMessageParams, ) @@ -77,7 +78,6 @@ "x-agent-api-key", "x-acting-user-api-key", "x-acting-user-cookie", - "x-acting-user-authorization", "x-acting-as-agent", "x-selected-account-id", } @@ -412,6 +412,21 @@ async def cancel_task( default_headers=headers, ) + async def interrupt_task( + self, agent: AgentEntity, task: TaskEntity, acp_url: str + ) -> dict[str, Any]: + """Forward a task/interrupt to the agent pod (non-terminal stop of the + in-flight turn). Same HTTP JSON-RPC transport as cancel_task.""" + params = InterruptTaskParams(agent=agent, task=task) + headers = await self.get_headers(agent) + return await self._call_jsonrpc( + url=acp_url, + method=AgentRPCMethod.TASK_INTERRUPT, + params=params, + request_id=f"{AgentRPCMethod.TASK_INTERRUPT}-{task.id}", # Use interrupt-specific request ID + default_headers=headers, + ) + async def send_event( self, agent: AgentEntity, diff --git a/agentex/src/domain/services/task_service.py b/agentex/src/domain/services/task_service.py index 299494b8..d2b68edf 100644 --- a/agentex/src/domain/services/task_service.py +++ b/agentex/src/domain/services/task_service.py @@ -362,6 +362,47 @@ async def cancel_task( task.status_reason = "Task canceled by user" return await self.task_repository.update(task) + async def interrupt_task( + self, agent: AgentEntity, task: TaskEntity, acp_url: str + ) -> TaskEntity: + """Interrupt a running task without terminating it. + + Mirrors cancel_task but transitions to the NON-terminal INTERRUPTED + status so the task stays continuable. The pod-forward is best-effort: + interrupt semantics live agent-side (for async agents the pod stops the + in-flight turn; for sync agents the real interrupt is the client tearing + down the streaming connection and the agent may not implement a handler + at all), so a forwarding failure must not prevent the platform from + recording the status transition. This never calls _transition_to_terminal. + """ + try: + await self.acp_client.interrupt_task( + agent=agent, task=task, acp_url=acp_url + ) + except Exception as e: + logger.warning( + f"Best-effort interrupt forward to ACP failed for task {task.id}: {e}" + ) + + task = await self.task_repository.get(id=task.id) + task.status = TaskStatus.INTERRUPTED + task.status_reason = "Task interrupted by user" + return await self.task_repository.update(task) + + async def resume_interrupted_task(self, task_id: str) -> TaskEntity | None: + """Atomically resume an INTERRUPTED task back to RUNNING on next-turn start. + + Non-terminal transition (INTERRUPTED -> RUNNING). Returns None if the task + was not INTERRUPTED (e.g. it was concurrently canceled), so callers can + keep the task as-is rather than clobbering a terminal status. + """ + return await self.transition_task_status( + task_id=task_id, + expected_status=TaskStatus.INTERRUPTED, + new_status=TaskStatus.RUNNING, + status_reason="Task resumed on next turn", + ) + async def create_event_and_forward_to_acp( self, agent: AgentEntity, diff --git a/agentex/src/domain/use_cases/agents_acp_use_case.py b/agentex/src/domain/use_cases/agents_acp_use_case.py index bc99111a..87e0b3fb 100644 --- a/agentex/src/domain/use_cases/agents_acp_use_case.py +++ b/agentex/src/domain/use_cases/agents_acp_use_case.py @@ -18,6 +18,7 @@ AgentRPCRequestEntity, CancelTaskRequestEntity, CreateTaskRequestEntity, + InterruptTaskRequestEntity, SendEventRequestEntity, SendMessageRequestEntity, ) @@ -41,7 +42,7 @@ ToolRequestContentEntity, ToolResponseContentEntity, ) -from src.domain.entities.tasks import TaskEntity +from src.domain.entities.tasks import TaskEntity, TaskStatus from src.domain.exceptions import ClientError from src.domain.mixins.task_messages.task_message_mixin import TaskMessageMixin from src.domain.repositories.agent_repository import DAgentRepository @@ -292,6 +293,16 @@ async def _get_or_create_task( # No identifier provided - will create below task = None + # INTERRUPTED is a transient resting state ("paused, waiting for you"). + # Reaching an existing task here means a new turn is starting (message/send, + # event/send, or a get-or-create by name), so resume it back to RUNNING. + # This is NOT a terminal transition, so it does not go through + # _transition_to_terminal. + if task is not None and task.status == TaskStatus.INTERRUPTED: + resumed = await self.task_service.resume_interrupted_task(task.id) + if resumed is not None: + task = resumed + # If task exists and params provided, update them if task and task_params is not None: if task.params != task_params: @@ -346,6 +357,7 @@ async def handle_rpc_request( method: AgentRPCMethod, params: CreateTaskRequestEntity | CancelTaskRequestEntity + | InterruptTaskRequestEntity | SendMessageRequestEntity | SendEventRequestEntity, agent_id: str | None = None, @@ -372,7 +384,7 @@ async def handle_rpc_request( Returns: - list[TaskMessageEntity] for synchronous MESSAGE_SEND - AsyncIterator[TaskMessageUpdateEntity] for streaming MESSAGE_SEND - - TaskEntity for TASK_CREATE or TASK_CANCEL + - TaskEntity for TASK_CREATE, TASK_CANCEL or TASK_INTERRUPT - EventEntity for EVENT_SEND """ # Get the agent @@ -396,6 +408,8 @@ async def handle_rpc_request( return await self._handle_task_create(agent, params, acp_url) elif method == AgentRPCMethod.TASK_CANCEL: return await self._handle_task_cancel(agent, params, acp_url) + elif method == AgentRPCMethod.TASK_INTERRUPT: + return await self._handle_task_interrupt(agent, params, acp_url) elif method == AgentRPCMethod.EVENT_SEND: return await self._handle_event_send( agent, params, request_headers, acp_url @@ -806,6 +820,45 @@ async def _handle_task_cancel( acp_url=acp_url, ) + async def _handle_task_interrupt( + self, agent: AgentEntity, params: InterruptTaskRequestEntity, acp_url: str + ) -> TaskEntity: + """ + Handle task/interrupt method. + + Interrupt is a non-terminal counterpart to task/cancel: it stops the + in-flight turn and leaves the task continuable (status INTERRUPTED), + rather than transitioning to a terminal status. It never calls + _transition_to_terminal. + + For ASYNC/AGENTIC agents the interrupt is forwarded to the pod over the + same HTTP JSON-RPC transport as event/send and task/cancel; the agent is + the actor that actually stops the model/tool calls. For SYNC agents the + real interrupt is the teardown of the streaming connection on the client + side, so the forwarded RPC is best-effort (a sync agent may not implement + an interrupt handler). In both cases the platform's job is to record the + non-terminal status transition; the forward is best-effort and must not + wedge the request. + + Args: + agent: The agent to interrupt the task for + params: Parameters containing task_id or task_name + acp_url: Resolved ACP URL to route to + + Returns: + The task with its (non-terminal) INTERRUPTED status. + """ + + task = await self.task_service.get_task( + id=params.task_id, name=params.task_name + ) + + return await self.task_service.interrupt_task( + agent=agent, + task=task, + acp_url=acp_url, + ) + async def _handle_event_send( self, agent: AgentEntity, diff --git a/agentex/src/domain/use_cases/tasks_use_case.py b/agentex/src/domain/use_cases/tasks_use_case.py index db8a9de4..225f04ba 100644 --- a/agentex/src/domain/use_cases/tasks_use_case.py +++ b/agentex/src/domain/use_cases/tasks_use_case.py @@ -132,6 +132,11 @@ async def update_mutable_fields_on_task( return task_entity + # Non-terminal statuses a task can be transitioned to a terminal status from. + # RUNNING is the normal case; INTERRUPTED is also valid so an interrupted + # (paused, still-continuable) task can still be canceled/completed/etc later. + _TERMINAL_TRANSITION_SOURCES = (TaskStatus.RUNNING, TaskStatus.INTERRUPTED) + async def _transition_to_terminal( self, target_status: TaskStatus, @@ -139,23 +144,26 @@ async def _transition_to_terminal( name: str | None = None, reason: str | None = None, ) -> TaskEntity: - """Atomically transition a running task to a terminal status.""" + """Atomically transition a running or interrupted task to a terminal status.""" if not id and not name: raise ClientError("Either id or name must be provided") task_entity = await self.task_service.get_task(id=id, name=name) if task_entity.status == TaskStatus.DELETED: raise ItemDoesNotExist(f"Task {id or name} not found") - if task_entity.status != TaskStatus.RUNNING: + if task_entity.status not in self._TERMINAL_TRANSITION_SOURCES: raise ClientError( f"Task {task_entity.id} is not running (current status: {task_entity.status}). " f"Only running tasks can have their status updated." ) + # Compare-and-swap on the observed non-terminal source status (RUNNING or + # INTERRUPTED) so a concurrent modification is still detected as a lost race. + expected_status = task_entity.status status_reason = reason or f"Task {target_status.value.lower()}" updated = await self.task_service.transition_task_status( task_id=task_entity.id, - expected_status=TaskStatus.RUNNING, + expected_status=expected_status, new_status=target_status, status_reason=status_reason, ) @@ -166,6 +174,43 @@ async def _transition_to_terminal( ) return updated + async def interrupt_task( + self, id: str | None = None, name: str | None = None, reason: str | None = None + ) -> TaskEntity: + """Interrupt a running task without terminating it. + + This is the non-terminal counterpart to the terminal-transition methods + (cancel/complete/fail/...): it transitions RUNNING -> INTERRUPTED via a + compare-and-swap and deliberately does NOT go through + _transition_to_terminal. The task stays continuable; the next message or + event resumes it back to RUNNING. + """ + if not id and not name: + raise ClientError("Either id or name must be provided") + + task_entity = await self.task_service.get_task(id=id, name=name) + if task_entity.status == TaskStatus.DELETED: + raise ItemDoesNotExist(f"Task {id or name} not found") + if task_entity.status != TaskStatus.RUNNING: + raise ClientError( + f"Task {task_entity.id} is not running (current status: {task_entity.status}). " + f"Only running tasks can be interrupted." + ) + + status_reason = reason or "Task interrupted" + updated = await self.task_service.transition_task_status( + task_id=task_entity.id, + expected_status=TaskStatus.RUNNING, + new_status=TaskStatus.INTERRUPTED, + status_reason=status_reason, + ) + if updated is None: + raise ClientError( + f"Task {task_entity.id} status was concurrently modified. " + f"Please retry the request." + ) + return updated + async def complete_task( self, id: str | None = None, name: str | None = None, reason: str | None = None ) -> TaskEntity: diff --git a/agentex/tests/integration/api/tasks/test_tasks_api.py b/agentex/tests/integration/api/tasks/test_tasks_api.py index b244a843..bdf857bd 100644 --- a/agentex/tests/integration/api/tasks/test_tasks_api.py +++ b/agentex/tests/integration/api/tasks/test_tasks_api.py @@ -156,6 +156,7 @@ async def test_list_tasks_returns_valid_structure_and_schema( "COMPLETED", "FAILED", "RUNNING", + "INTERRUPTED", "TERMINATED", "TIMED_OUT", ] @@ -1608,6 +1609,72 @@ async def test_cancel_task_endpoint(self, isolated_client, test_task): assert task_data["status"] == "CANCELED" assert task_data["status_reason"] == "User requested cancellation" + async def test_interrupt_task_endpoint(self, isolated_client, test_task): + """POST /tasks/{task_id}/interrupt transitions RUNNING to the non-terminal + INTERRUPTED status (task stays continuable).""" + # When + response = await isolated_client.post( + f"/tasks/{test_task.id}/interrupt", + json={"reason": "User hit stop"}, + ) + + # Then + assert response.status_code == 200 + task_data = response.json() + assert task_data["status"] == "INTERRUPTED" + assert task_data["status_reason"] == "User hit stop" + + async def test_interrupt_task_default_reason(self, isolated_client, test_task): + """POST /tasks/{task_id}/interrupt without a reason uses the default.""" + response = await isolated_client.post( + f"/tasks/{test_task.id}/interrupt", + ) + + assert response.status_code == 200 + task_data = response.json() + assert task_data["status"] == "INTERRUPTED" + assert task_data["status_reason"] == "Task interrupted" + + async def test_interrupted_task_stays_transition_eligible( + self, isolated_client, test_task + ): + """An interrupted task is non-terminal: it can still be canceled + afterwards (terminal-from-INTERRUPTED).""" + # Given - interrupt the running task + response = await isolated_client.post( + f"/tasks/{test_task.id}/interrupt", + ) + assert response.status_code == 200 + assert response.json()["status"] == "INTERRUPTED" + + # When - cancel the interrupted task + response = await isolated_client.post( + f"/tasks/{test_task.id}/cancel", + json={"reason": "canceled after interrupt"}, + ) + + # Then - the terminal transition succeeds from INTERRUPTED + assert response.status_code == 200 + task_data = response.json() + assert task_data["status"] == "CANCELED" + assert task_data["status_reason"] == "canceled after interrupt" + + async def test_cannot_interrupt_completed_task(self, isolated_client, test_task): + """A terminal (completed) task cannot be interrupted.""" + # Given - complete the task first + response = await isolated_client.post( + f"/tasks/{test_task.id}/complete", + ) + assert response.status_code == 200 + + # When - try to interrupt the completed task + response = await isolated_client.post( + f"/tasks/{test_task.id}/interrupt", + ) + + # Then - rejected + assert response.status_code == 400 + async def test_terminate_task_endpoint(self, isolated_client, test_task): """Test POST /tasks/{task_id}/terminate transitions RUNNING to TERMINATED""" # When diff --git a/agentex/tests/unit/api/test_tasks_authz.py b/agentex/tests/unit/api/test_tasks_authz.py index ebef015c..aa6f3f2a 100644 --- a/agentex/tests/unit/api/test_tasks_authz.py +++ b/agentex/tests/unit/api/test_tasks_authz.py @@ -192,6 +192,50 @@ async def test_task_cancel_with_task_name_uses_cancel(self): (AgentexResourceType.task, "task-cancel", AuthorizedOperationType.cancel) ] + async def test_task_interrupt_with_task_id_uses_update(self): + # Interrupt is non-terminal, so it maps to ``update`` (editor-allowed), + # NOT the owner-only ``cancel`` that task/cancel uses. + from src.api.routes.agents import _authorize_rpc_request + from src.api.schemas.agents import AgentRPCMethod + + authorization = MagicMock() + calls = self._capture_check(authorization) + task_service = MagicMock() + request = MagicMock() + request.method = AgentRPCMethod.TASK_INTERRUPT + request.params = MagicMock(task_id="task-int", task_name=None) + + await _authorize_rpc_request(request, authorization, task_service) + + assert calls == [ + (AgentexResourceType.task, "task-int", AuthorizedOperationType.update) + ] + + async def test_task_interrupt_with_task_name_uses_update(self): + from src.api.routes.agents import _authorize_rpc_request + from src.api.schemas.agents import AgentRPCMethod + + authorization = MagicMock() + calls = self._capture_check(authorization) + task_service = MagicMock() + task_service.get_task = AsyncMock( + return_value=MagicMock(id="task-int-resolved") + ) + request = MagicMock() + request.method = AgentRPCMethod.TASK_INTERRUPT + request.params = MagicMock(task_id=None, task_name="interrupt-task") + + await _authorize_rpc_request(request, authorization, task_service) + + task_service.get_task.assert_awaited_once_with(name="interrupt-task") + assert calls == [ + ( + AgentexResourceType.task, + "task-int-resolved", + AuthorizedOperationType.update, + ) + ] + async def test_unwired_method_fails_closed_with_not_implemented(self): # Fail-closed default: any AgentRPCMethod not explicitly wired in # _authorize_rpc_request must raise rather than silently pass through @@ -238,6 +282,49 @@ async def test_rest_cancel_route_checks_cancel_operation(self): ) +@pytest.mark.unit +@pytest.mark.asyncio +class TestRestInterruptRouteRequiresUpdate: + """The REST ``POST /tasks/{id}/interrupt`` route authorizes the editor-allowed + ``update`` action (interrupt is non-terminal), distinct from the owner-only + ``cancel`` used by the sibling cancel route.""" + + async def test_rest_interrupt_route_checks_update_operation(self): + import inspect + + from src.api.routes.tasks import interrupt_task + + annotation = inspect.signature(interrupt_task).parameters["task_id"].annotation + dep = _dep_callable(annotation) + + authorization = MagicMock() + authorization.check = AsyncMock(return_value=True) + repos = (MagicMock(), MagicMock(), MagicMock()) + + result = await dep(authorization, *repos, "task-i") + + assert result == "task-i" + assert ( + authorization.check.await_args.kwargs["operation"] + == AuthorizedOperationType.update + ) + + +def test_sync_acp_type_permits_task_interrupt(): + """task/interrupt must be allowed for SYNC, ASYNC and AGENTIC agents. The + SYNC allow-list historically only permitted MESSAGE_SEND and TASK_CREATE.""" + from src.domain.entities.agents import ACPType + from src.domain.entities.agents_rpc import ( + ACP_TYPE_TO_ALLOWED_RPC_METHODS, + AgentRPCMethod, + ) + + for acp_type in (ACPType.SYNC, ACPType.ASYNC, ACPType.AGENTIC): + assert ( + AgentRPCMethod.TASK_INTERRUPT in ACP_TYPE_TO_ALLOWED_RPC_METHODS[acp_type] + ) + + def test_cancel_operation_wire_format_matches_agentex_auth_contract(): """Cross-repo enum contract: the literal string ``"cancel"`` is the wire format shared with agentex-auth's mirror enum. Diverging strings would diff --git a/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py b/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py index e3cb7faf..78c24ac6 100644 --- a/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py +++ b/agentex/tests/unit/use_cases/test_acp_type_backwards_compatibility_use_case.py @@ -36,14 +36,15 @@ async def test_both_agentic_and_async_have_same_allowed_methods(self): agentic_methods = set(ACP_TYPE_TO_ALLOWED_RPC_METHODS[ACPType.AGENTIC]) async_methods = set(ACP_TYPE_TO_ALLOWED_RPC_METHODS[ACPType.ASYNC]) - assert agentic_methods == async_methods, ( - "AGENTIC and ASYNC should have identical allowed RPC methods" - ) + assert ( + agentic_methods == async_methods + ), "AGENTIC and ASYNC should have identical allowed RPC methods" # Verify they include the expected methods expected_methods = { AgentRPCMethod.TASK_CREATE, AgentRPCMethod.TASK_CANCEL, + AgentRPCMethod.TASK_INTERRUPT, AgentRPCMethod.EVENT_SEND, } assert agentic_methods == expected_methods @@ -361,6 +362,6 @@ async def test_agentic_and_async_agents_both_use_not_sync_logic(self): # Both AGENTIC and ASYNC should pass the same conditional checks agentic_is_not_sync = agentic_agent.acp_type != ACPType.SYNC async_is_not_sync = async_agent.acp_type != ACPType.SYNC - assert agentic_is_not_sync == async_is_not_sync, ( - "AGENTIC and ASYNC should have identical behavior in != SYNC checks" - ) + assert ( + agentic_is_not_sync == async_is_not_sync + ), "AGENTIC and ASYNC should have identical behavior in != SYNC checks" diff --git a/agentex/tests/unit/use_cases/test_tasks_use_case.py b/agentex/tests/unit/use_cases/test_tasks_use_case.py index c2b70d1e..1588439b 100644 --- a/agentex/tests/unit/use_cases/test_tasks_use_case.py +++ b/agentex/tests/unit/use_cases/test_tasks_use_case.py @@ -155,6 +155,139 @@ async def test_timeout_running_task( assert updated.status == TaskStatus.TIMED_OUT assert updated.status_reason == "Task timed_out" + # --- Non-terminal interrupt (RUNNING <-> INTERRUPTED) --- + + async def test_interrupt_running_task_is_non_terminal( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """Interrupt transitions RUNNING -> INTERRUPTED, a non-terminal status.""" + # Given + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-test" + ) + assert task.status == TaskStatus.RUNNING + + # When + updated = await tasks_use_case.interrupt_task( + id=task.id, reason="User hit stop" + ) + + # Then + assert updated.status == TaskStatus.INTERRUPTED + assert updated.status_reason == "User hit stop" + + async def test_interrupt_default_reason( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """Interrupt without an explicit reason uses the default.""" + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-default-reason-test" + ) + + updated = await tasks_use_case.interrupt_task(id=task.id) + + assert updated.status == TaskStatus.INTERRUPTED + assert updated.status_reason == "Task interrupted" + + async def test_interrupted_task_can_resume_to_running( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """INTERRUPTED -> RUNNING on next-turn start (resume_interrupted_task).""" + # Given an interrupted task + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-resume-test" + ) + await tasks_use_case.interrupt_task(id=task.id) + + # When the next turn starts + resumed = await task_service.resume_interrupted_task(task.id) + + # Then it is RUNNING again + assert resumed is not None + assert resumed.status == TaskStatus.RUNNING + + async def test_resume_non_interrupted_task_is_noop( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """Resuming a task that is not INTERRUPTED must not clobber its status.""" + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-resume-noop-test" + ) + # RUNNING (not INTERRUPTED): the compare-and-swap misses, returns None. + resumed = await task_service.resume_interrupted_task(task.id) + assert resumed is None + + async def test_interrupted_task_can_transition_to_terminal( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """An INTERRUPTED task is still transition-eligible: it can be canceled + (terminal-from-INTERRUPTED) later.""" + # Given an interrupted task + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-then-cancel-test" + ) + interrupted = await tasks_use_case.interrupt_task(id=task.id) + assert interrupted.status == TaskStatus.INTERRUPTED + + # When the interrupted task is canceled + canceled = await tasks_use_case.cancel_task( + id=task.id, reason="User canceled after interrupt" + ) + + # Then the terminal transition succeeds from INTERRUPTED + assert canceled.status == TaskStatus.CANCELED + assert canceled.status_reason == "User canceled after interrupt" + + async def test_interrupted_task_can_be_completed( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """terminal-from-INTERRUPTED also works for complete().""" + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-then-complete-test" + ) + await tasks_use_case.interrupt_task(id=task.id) + + completed = await tasks_use_case.complete_task(id=task.id, reason="done") + + assert completed.status == TaskStatus.COMPLETED + + async def test_cannot_interrupt_non_running_task( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """A task that is not RUNNING cannot be interrupted.""" + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-non-running-test" + ) + await tasks_use_case.complete_task(id=task.id) + + with pytest.raises(ClientError, match="not running"): + await tasks_use_case.interrupt_task(id=task.id) + + async def test_cannot_interrupt_already_interrupted_task( + self, tasks_use_case, task_service, agent_repository, sample_agent + ): + """Interrupt is only valid from RUNNING; a second interrupt is rejected.""" + await create_or_get_agent(agent_repository, sample_agent) + task = await task_service.create_task( + agent=sample_agent, task_name="interrupt-twice-test" + ) + await tasks_use_case.interrupt_task(id=task.id) + + with pytest.raises(ClientError, match="not running"): + await tasks_use_case.interrupt_task(id=task.id) + + async def test_interrupt_requires_id_or_name(self, tasks_use_case): + """Interrupt with neither id nor name raises.""" + with pytest.raises(ClientError, match="Either id or name"): + await tasks_use_case.interrupt_task() + # --- Default reason for each transition --- @pytest.mark.parametrize( From a838e50b94d3626a1f13faaf3dcd90298a6b3702 Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Thu, 16 Jul 2026 11:22:58 -0400 Subject: [PATCH 2/5] fix(tasks): address review feedback on task/interrupt (AGX1-391) - interrupt_task (service): use a RUNNING->INTERRUPTED compare-and-swap instead of an unconditional update, so a task the agent terminated during the ACP call window is not un-terminated (mirrors resume_interrupted_task) - restore x-acting-user-authorization to BLOCKED_HEADERS (unintentionally dropped) - fix stale 'only running tasks' error text in _transition_to_terminal now that INTERRUPTED is a valid transition source Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/domain/services/agent_acp_service.py | 1 + agentex/src/domain/services/task_service.py | 22 +++++++++++++++---- .../src/domain/use_cases/tasks_use_case.py | 4 ++-- 3 files changed, 21 insertions(+), 6 deletions(-) diff --git a/agentex/src/domain/services/agent_acp_service.py b/agentex/src/domain/services/agent_acp_service.py index 0d3495af..3a1bfddc 100644 --- a/agentex/src/domain/services/agent_acp_service.py +++ b/agentex/src/domain/services/agent_acp_service.py @@ -78,6 +78,7 @@ "x-agent-api-key", "x-acting-user-api-key", "x-acting-user-cookie", + "x-acting-user-authorization", "x-acting-as-agent", "x-selected-account-id", } diff --git a/agentex/src/domain/services/task_service.py b/agentex/src/domain/services/task_service.py index d2b68edf..eb226880 100644 --- a/agentex/src/domain/services/task_service.py +++ b/agentex/src/domain/services/task_service.py @@ -384,10 +384,24 @@ async def interrupt_task( f"Best-effort interrupt forward to ACP failed for task {task.id}: {e}" ) - task = await self.task_repository.get(id=task.id) - task.status = TaskStatus.INTERRUPTED - task.status_reason = "Task interrupted by user" - return await self.task_repository.update(task) + # Compare-and-swap RUNNING -> INTERRUPTED (mirrors resume_interrupted_task). + # If the CAS misses, the task moved to another status during the ACP call + # (the agent completed/failed it, or it was concurrently canceled); leave + # that status intact rather than clobbering a terminal status with the + # non-terminal INTERRUPTED. + updated = await self.transition_task_status( + task_id=task.id, + expected_status=TaskStatus.RUNNING, + new_status=TaskStatus.INTERRUPTED, + status_reason="Task interrupted by user", + ) + if updated is not None: + return updated + logger.info( + f"Interrupt for task {task.id} not applied: task is no longer RUNNING; " + f"leaving its current status intact." + ) + return await self.task_repository.get(id=task.id) async def resume_interrupted_task(self, task_id: str) -> TaskEntity | None: """Atomically resume an INTERRUPTED task back to RUNNING on next-turn start. diff --git a/agentex/src/domain/use_cases/tasks_use_case.py b/agentex/src/domain/use_cases/tasks_use_case.py index 225f04ba..7c114744 100644 --- a/agentex/src/domain/use_cases/tasks_use_case.py +++ b/agentex/src/domain/use_cases/tasks_use_case.py @@ -153,8 +153,8 @@ async def _transition_to_terminal( raise ItemDoesNotExist(f"Task {id or name} not found") if task_entity.status not in self._TERMINAL_TRANSITION_SOURCES: raise ClientError( - f"Task {task_entity.id} is not running (current status: {task_entity.status}). " - f"Only running tasks can have their status updated." + f"Task {task_entity.id} cannot be transitioned (current status: {task_entity.status}). " + f"Only running or interrupted tasks can have their status updated." ) # Compare-and-swap on the observed non-terminal source status (RUNNING or From 85ae510fb4b973449bfebe0c3019f08a0ddf45c5 Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Thu, 16 Jul 2026 11:33:21 -0400 Subject: [PATCH 3/5] test(tasks): update transition-guard assertions to match new error text (AGX1-391) The _transition_to_terminal error message changed (Only running -> 'Only running or interrupted') per review; assert on the stable 'Only running' substring, which matches both the transition and interrupt guard messages. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../tests/unit/use_cases/test_tasks_use_case.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/agentex/tests/unit/use_cases/test_tasks_use_case.py b/agentex/tests/unit/use_cases/test_tasks_use_case.py index 1588439b..798cb7cd 100644 --- a/agentex/tests/unit/use_cases/test_tasks_use_case.py +++ b/agentex/tests/unit/use_cases/test_tasks_use_case.py @@ -267,7 +267,7 @@ async def test_cannot_interrupt_non_running_task( ) await tasks_use_case.complete_task(id=task.id) - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.interrupt_task(id=task.id) async def test_cannot_interrupt_already_interrupted_task( @@ -280,7 +280,7 @@ async def test_cannot_interrupt_already_interrupted_task( ) await tasks_use_case.interrupt_task(id=task.id) - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.interrupt_task(id=task.id) async def test_interrupt_requires_id_or_name(self, tasks_use_case): @@ -357,7 +357,7 @@ async def test_cannot_transition_completed_task( await tasks_use_case.complete_task(id=task.id) # When / Then - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.terminate_task(id=task.id) async def test_cannot_transition_failed_task( @@ -372,7 +372,7 @@ async def test_cannot_transition_failed_task( await tasks_use_case.fail_task(id=task.id) # When / Then - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.complete_task(id=task.id) async def test_cannot_transition_canceled_task( @@ -387,7 +387,7 @@ async def test_cannot_transition_canceled_task( await tasks_use_case.cancel_task(id=task.id) # When / Then - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.complete_task(id=task.id) async def test_cannot_transition_terminated_task( @@ -402,7 +402,7 @@ async def test_cannot_transition_terminated_task( await tasks_use_case.terminate_task(id=task.id) # When / Then - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.complete_task(id=task.id) async def test_cannot_transition_timed_out_task( @@ -417,7 +417,7 @@ async def test_cannot_transition_timed_out_task( await tasks_use_case.timeout_task(id=task.id) # When / Then - with pytest.raises(ClientError, match="not running"): + with pytest.raises(ClientError, match="Only running"): await tasks_use_case.complete_task(id=task.id) async def test_cannot_transition_deleted_task( From d8f85cde9c1082d73daadb245bf82ab979e64ed8 Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Thu, 16 Jul 2026 11:46:57 -0400 Subject: [PATCH 4/5] refactor(tasks): align cancel_task with the interrupt_task CAS pattern (AGX1-391) cancel_task (RPC path) used an unconditional repository update that could clobber a task the agent terminated during the ACP-forward window. Mirror interrupt_task: CAS to CANCELED on the observed non-terminal source (RUNNING/INTERRUPTED), no-op if the task already reached a terminal status. Addresses the Greptile 4/5 consistency note. Co-Authored-By: Claude Opus 4.8 (1M context) --- agentex/src/domain/services/task_service.py | 24 ++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/agentex/src/domain/services/task_service.py b/agentex/src/domain/services/task_service.py index eb226880..66c46dbe 100644 --- a/agentex/src/domain/services/task_service.py +++ b/agentex/src/domain/services/task_service.py @@ -354,13 +354,27 @@ async def send_message_stream( async def cancel_task( self, agent: AgentEntity, task: TaskEntity, acp_url: str ) -> TaskEntity: - """Cancel a running task""" + """Cancel a running (or interrupted) task.""" await self.acp_client.cancel_task(agent=agent, task=task, acp_url=acp_url) - task = await self.task_repository.get(id=task.id) - task.status = TaskStatus.CANCELED - task.status_reason = "Task canceled by user" - return await self.task_repository.update(task) + # Compare-and-swap to CANCELED on the observed non-terminal source status + # (mirrors interrupt_task). If the task moved to another status during the + # ACP call (the agent completed/failed it, or it was concurrently modified), + # leave that status intact rather than clobbering it with CANCELED. + current = await self.task_repository.get(id=task.id) + if current.status not in (TaskStatus.RUNNING, TaskStatus.INTERRUPTED): + logger.info( + f"Cancel for task {task.id} not applied: task is no longer running " + f"(current status: {current.status}); leaving its status intact." + ) + return current + updated = await self.transition_task_status( + task_id=task.id, + expected_status=current.status, + new_status=TaskStatus.CANCELED, + status_reason="Task canceled by user", + ) + return updated if updated is not None else await self.task_repository.get(id=task.id) async def interrupt_task( self, agent: AgentEntity, task: TaskEntity, acp_url: str From 605066e626fabd5dc95cf4047480e76b788005c7 Mon Sep 17 00:00:00 2001 From: Declan Brady Date: Thu, 16 Jul 2026 14:05:27 -0400 Subject: [PATCH 5/5] refactor(tasks): make task/interrupt RPC a pure forward; agent owns INTERRUPTED (AGX1-391) Interrupt is cooperative, not control-plane-authoritative. The task/interrupt RPC now only forwards to the agent (like event/send) and writes no status; the agent stops its own turn and records INTERRUPTED by calling the REST POST /tasks/{id}/interrupt route (like the other task-state routes). The control plane still owns RUNNING (task creation + auto-resume on the next turn). An agent that does not implement interrupt stays RUNNING, keeping status honest. Also (per @smoreinis review): in _get_or_create_task, when the INTERRUPTED->RUNNING resume CAS misses, refetch and reject with ClientError if the task is now terminal, instead of starting a turn against a terminal task. Co-Authored-By: Claude Opus 4.8 (1M context) --- agentex/src/domain/services/task_service.py | 45 +++++-------------- .../domain/use_cases/agents_acp_use_case.py | 11 +++++ 2 files changed, 21 insertions(+), 35 deletions(-) diff --git a/agentex/src/domain/services/task_service.py b/agentex/src/domain/services/task_service.py index 66c46dbe..d5d1291f 100644 --- a/agentex/src/domain/services/task_service.py +++ b/agentex/src/domain/services/task_service.py @@ -379,42 +379,17 @@ async def cancel_task( async def interrupt_task( self, agent: AgentEntity, task: TaskEntity, acp_url: str ) -> TaskEntity: - """Interrupt a running task without terminating it. - - Mirrors cancel_task but transitions to the NON-terminal INTERRUPTED - status so the task stays continuable. The pod-forward is best-effort: - interrupt semantics live agent-side (for async agents the pod stops the - in-flight turn; for sync agents the real interrupt is the client tearing - down the streaming connection and the agent may not implement a handler - at all), so a forwarding failure must not prevent the platform from - recording the status transition. This never calls _transition_to_terminal. + """Forward a task/interrupt to the agent (courier only; no status write). + + Interrupt is a COOPERATIVE action, unlike cancel. The control plane only + forwards the request to the agent pod (same as event/send); the agent is + responsible for stopping its in-flight turn and then recording INTERRUPTED + via the REST POST /tasks/{id}/interrupt route (like the other task-state + routes). The control plane never writes status here — so an agent that does + not implement interrupt simply keeps running and its status stays honest. + Resume back to RUNNING is owned by the control plane (see _get_or_create_task). """ - try: - await self.acp_client.interrupt_task( - agent=agent, task=task, acp_url=acp_url - ) - except Exception as e: - logger.warning( - f"Best-effort interrupt forward to ACP failed for task {task.id}: {e}" - ) - - # Compare-and-swap RUNNING -> INTERRUPTED (mirrors resume_interrupted_task). - # If the CAS misses, the task moved to another status during the ACP call - # (the agent completed/failed it, or it was concurrently canceled); leave - # that status intact rather than clobbering a terminal status with the - # non-terminal INTERRUPTED. - updated = await self.transition_task_status( - task_id=task.id, - expected_status=TaskStatus.RUNNING, - new_status=TaskStatus.INTERRUPTED, - status_reason="Task interrupted by user", - ) - if updated is not None: - return updated - logger.info( - f"Interrupt for task {task.id} not applied: task is no longer RUNNING; " - f"leaving its current status intact." - ) + await self.acp_client.interrupt_task(agent=agent, task=task, acp_url=acp_url) return await self.task_repository.get(id=task.id) async def resume_interrupted_task(self, task_id: str) -> TaskEntity | None: diff --git a/agentex/src/domain/use_cases/agents_acp_use_case.py b/agentex/src/domain/use_cases/agents_acp_use_case.py index 87e0b3fb..2bdeea63 100644 --- a/agentex/src/domain/use_cases/agents_acp_use_case.py +++ b/agentex/src/domain/use_cases/agents_acp_use_case.py @@ -302,6 +302,17 @@ async def _get_or_create_task( resumed = await self.task_service.resume_interrupted_task(task.id) if resumed is not None: task = resumed + else: + # The INTERRUPTED -> RUNNING CAS missed: the task moved out of + # INTERRUPTED concurrently (e.g. a cancel/complete). Refetch and + # reject if it is now terminal, rather than starting a turn against + # a terminal task. (RUNNING here means another turn already resumed + # it, which is fine to proceed with.) + task = await self.task_service.get_task(id=task.id) + if task.status not in (TaskStatus.RUNNING, TaskStatus.INTERRUPTED): + raise ClientError( + f"Task {task.id} is {task.status.value} and cannot accept new turns." + ) # If task exists and params provided, update them if task and task_params is not None: