feat(tasks): non-terminal task/interrupt + INTERRUPTED status (AGX1-391)

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

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

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}; "

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,

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):

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)

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"

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

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"

agentex/src/domain/services/agent_acp_service.py

@@ -13,6 +13,7 @@
     AgentRPCMethod,
     CancelTaskParams,
     CreateTaskParams,
+    InterruptTaskParams,
     SendEventParams,
     SendMessageParams,
 )
@@ -412,6 +413,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,