Skip to content

Commit db293d5

Browse files
declan-scaleclaude
andcommitted
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) <noreply@anthropic.com>
1 parent 2913fe0 commit db293d5

17 files changed

Lines changed: 670 additions & 14 deletions

File tree

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
"""add interrupted task status
2+
3+
Revision ID: a1b2c3d4e5f6
4+
Revises: 9a4b8c7d6e5f
5+
Create Date: 2026-07-16 12:00:00.000000
6+
7+
"""
8+
from typing import Sequence, Union
9+
10+
from alembic import op
11+
import sqlalchemy as sa
12+
13+
14+
# revision identifiers, used by Alembic.
15+
revision: str = 'a1b2c3d4e5f6'
16+
down_revision: Union[str, None] = '9a4b8c7d6e5f'
17+
branch_labels: Union[str, Sequence[str], None] = None
18+
depends_on: Union[str, Sequence[str], None] = None
19+
20+
21+
def upgrade() -> None:
22+
# New non-terminal task status: the current turn was interrupted by the user
23+
# and the task is waiting for the next message (see task/interrupt).
24+
op.execute("""
25+
ALTER TYPE taskstatus ADD VALUE IF NOT EXISTS 'INTERRUPTED';
26+
""")
27+
28+
29+
def downgrade() -> None:
30+
# Postgres does not support removing a value from an enum type, so there is
31+
# nothing to do on downgrade (mirrors the soft_delete_status migration).
32+
pass

agentex/openapi.yaml

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -754,6 +754,43 @@ paths:
754754
application/json:
755755
schema:
756756
$ref: '#/components/schemas/HTTPValidationError'
757+
/tasks/{task_id}/interrupt:
758+
post:
759+
tags:
760+
- Tasks
761+
summary: Interrupt Task
762+
description: Stop the in-flight turn without terminating the task. Transitions
763+
a running task to the non-terminal INTERRUPTED status; the task stays continuable
764+
and the next message or event resumes it.
765+
operationId: interrupt_task_tasks__task_id__interrupt_post
766+
parameters:
767+
- name: task_id
768+
in: path
769+
required: true
770+
schema:
771+
type: string
772+
title: Task Id
773+
requestBody:
774+
content:
775+
application/json:
776+
schema:
777+
anyOf:
778+
- $ref: '#/components/schemas/TaskStatusReasonRequest'
779+
- type: 'null'
780+
title: Request
781+
responses:
782+
'200':
783+
description: Successful Response
784+
content:
785+
application/json:
786+
schema:
787+
$ref: '#/components/schemas/Task'
788+
'422':
789+
description: Validation Error
790+
content:
791+
application/json:
792+
schema:
793+
$ref: '#/components/schemas/HTTPValidationError'
757794
/tasks/{task_id}/terminate:
758795
post:
759796
tags:
@@ -4247,11 +4284,13 @@ components:
42474284
- task/create
42484285
- message/send
42494286
- task/cancel
4287+
- task/interrupt
42504288
title: AgentRPCMethod
42514289
AgentRPCParams:
42524290
anyOf:
42534291
- $ref: '#/components/schemas/CreateTaskRequest'
42544292
- $ref: '#/components/schemas/CancelTaskRequest'
4293+
- $ref: '#/components/schemas/InterruptTaskRequest'
42554294
- $ref: '#/components/schemas/SendMessageRequest'
42564295
- $ref: '#/components/schemas/SendEventRequest'
42574296
title: AgentRPCParams
@@ -5569,6 +5608,24 @@ components:
55695608
title: Detail
55705609
type: object
55715610
title: HTTPValidationError
5611+
InterruptTaskRequest:
5612+
properties:
5613+
task_id:
5614+
anyOf:
5615+
- type: string
5616+
- type: 'null'
5617+
title: Task Id
5618+
description: The ID of the task to interrupt. Either this or task_name must
5619+
be provided.
5620+
task_name:
5621+
anyOf:
5622+
- type: string
5623+
- type: 'null'
5624+
title: Task Name
5625+
description: The name of the task to interrupt. Either this or task_id must
5626+
be provided.
5627+
type: object
5628+
title: InterruptTaskRequest
55725629
ListCheckpointsRequest:
55735630
properties:
55745631
thread_id:
@@ -6743,6 +6800,7 @@ components:
67436800
- COMPLETED
67446801
- FAILED
67456802
- RUNNING
6803+
- INTERRUPTED
67466804
- TERMINATED
67476805
- TIMED_OUT
67486806
- DELETED

agentex/src/api/routes/agents.py

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -478,6 +478,30 @@ async def _authorize_rpc_request(
478478
raise ValueError(
479479
"Either task_id or task_name must be provided for task/cancel"
480480
)
481+
482+
case AgentRPCMethod.TASK_INTERRUPT:
483+
# Interrupt is non-terminal (the task stays continuable), so it maps
484+
# to ``update`` rather than the owner-only ``cancel`` used above.
485+
task_id = request.params.task_id
486+
task_name = request.params.task_name
487+
488+
if task_id is not None:
489+
await check_task_or_collapse_to_404(
490+
authorization_service,
491+
task_id,
492+
AuthorizedOperationType.update,
493+
)
494+
elif task_name is not None:
495+
existing_task = await task_service.get_task(name=task_name)
496+
await check_task_or_collapse_to_404(
497+
authorization_service,
498+
existing_task.id,
499+
AuthorizedOperationType.update,
500+
)
501+
else:
502+
raise ValueError(
503+
"Either task_id or task_name must be provided for task/interrupt"
504+
)
481505
case _:
482506
raise NotImplementedError(
483507
f"_authorize_rpc_request has no case for {request.method}; "

agentex/src/api/routes/tasks.py

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,30 @@ async def cancel_task(
275275
return Task.model_validate(updated)
276276

277277

278+
@router.post(
279+
"/{task_id}/interrupt",
280+
response_model=Task,
281+
summary="Interrupt Task",
282+
description=(
283+
"Stop the in-flight turn without terminating the task. Transitions a "
284+
"running task to the non-terminal INTERRUPTED status; the task stays "
285+
"continuable and the next message or event resumes it."
286+
),
287+
)
288+
async def interrupt_task(
289+
# Interrupt is non-terminal (the task remains continuable), so it authorizes
290+
# the editor-allowed ``update`` action — matching the RPC task/interrupt
291+
# path — rather than the owner-only ``cancel`` used by the sibling cancel route.
292+
task_id: DAuthorizedId(AgentexResourceType.task, AuthorizedOperationType.update),
293+
task_use_case: DTaskUseCase,
294+
request: TaskStatusReasonRequest | None = None,
295+
) -> Task:
296+
updated = await task_use_case.interrupt_task(
297+
id=task_id, reason=request.reason if request else None
298+
)
299+
return Task.model_validate(updated)
300+
301+
278302
@router.post(
279303
"/{task_id}/terminate",
280304
response_model=Task,

agentex/src/api/schemas/agents.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@ class AgentRPCMethod(str, Enum):
3030
TASK_CREATE = "task/create"
3131
MESSAGE_SEND = "message/send"
3232
TASK_CANCEL = "task/cancel"
33+
TASK_INTERRUPT = "task/interrupt"
3334

3435

3536
class AgentInputType(str, Enum):

agentex/src/api/schemas/agents_rpc.py

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ class AgentRPCMethod(str, Enum):
1919
TASK_CREATE = "task/create"
2020
MESSAGE_SEND = "message/send"
2121
TASK_CANCEL = "task/cancel"
22+
TASK_INTERRUPT = "task/interrupt"
2223

2324

2425
class CreateTaskRequest(BaseModel):
@@ -66,6 +67,23 @@ def validate_task_identifiers(self):
6667
return self
6768

6869

70+
class InterruptTaskRequest(BaseModel):
71+
task_id: str | None = Field(
72+
None,
73+
description="The ID of the task to interrupt. Either this or task_name must be provided.",
74+
)
75+
task_name: str | None = Field(
76+
None,
77+
description="The name of the task to interrupt. Either this or task_id must be provided.",
78+
)
79+
80+
@model_validator(mode="after")
81+
def validate_task_identifiers(self):
82+
if self.task_id is not None and self.task_name is not None:
83+
raise ValueError("Cannot provide both task_id and task_name - use only one")
84+
return self
85+
86+
6987
class SendMessageRequest(BaseModel):
7088
task_id: str | None = Field(
7189
None, description="The ID of the task that the message was sent to"
@@ -111,7 +129,11 @@ def validate_task_identifiers(self):
111129

112130
class AgentRPCParams(RootModel):
113131
root: (
114-
CreateTaskRequest | CancelTaskRequest | SendMessageRequest | SendEventRequest
132+
CreateTaskRequest
133+
| CancelTaskRequest
134+
| InterruptTaskRequest
135+
| SendMessageRequest
136+
| SendEventRequest
115137
) = Field(..., description="The parameters for the agent RPC request")
116138

117139

@@ -137,6 +159,10 @@ def validate_params_based_on_method(cls, data):
137159
data["params"] = AgentRPCParams(
138160
root=CancelTaskRequest(**params_data)
139161
)
162+
elif method == AgentRPCMethod.TASK_INTERRUPT:
163+
data["params"] = AgentRPCParams(
164+
root=InterruptTaskRequest(**params_data)
165+
)
140166
elif method == AgentRPCMethod.MESSAGE_SEND:
141167
data["params"] = AgentRPCParams(
142168
root=SendMessageRequest(**params_data)

agentex/src/api/schemas/tasks.py

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ class TaskStatus(str, Enum):
1919
COMPLETED = "COMPLETED"
2020
FAILED = "FAILED"
2121
RUNNING = "RUNNING"
22+
# Non-terminal: current turn stopped by the user, task still continuable.
23+
INTERRUPTED = "INTERRUPTED"
2224
TERMINATED = "TERMINATED"
2325
TIMED_OUT = "TIMED_OUT"
2426
DELETED = "DELETED"

agentex/src/domain/entities/agents_rpc.py

Lines changed: 44 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
AgentRPCRequest,
88
CancelTaskRequest,
99
CreateTaskRequest,
10+
InterruptTaskRequest,
1011
)
1112
from src.domain.entities.agents import ACPType, AgentEntity
1213
from src.domain.entities.events import EventEntity
@@ -26,6 +27,7 @@ class AgentRPCMethod(str, Enum):
2627

2728
TASK_CREATE = "task/create"
2829
TASK_CANCEL = "task/cancel"
30+
TASK_INTERRUPT = "task/interrupt"
2931
MESSAGE_SEND = "message/send"
3032
EVENT_SEND = "event/send"
3133

@@ -85,16 +87,32 @@ class CancelTaskParams(BaseModel):
8587
task: TaskEntity = Field(..., description="The task that was cancelled")
8688

8789

90+
class InterruptTaskParams(BaseModel):
91+
"""Parameters for task/interrupt method"""
92+
93+
agent: AgentEntity = Field(
94+
...,
95+
description="The agent that the task was sent to",
96+
)
97+
task: TaskEntity = Field(..., description="The task that was interrupted")
98+
99+
88100
ACP_TYPE_TO_ALLOWED_RPC_METHODS = {
89-
ACPType.SYNC: [AgentRPCMethod.MESSAGE_SEND, AgentRPCMethod.TASK_CREATE],
101+
ACPType.SYNC: [
102+
AgentRPCMethod.MESSAGE_SEND,
103+
AgentRPCMethod.TASK_CREATE,
104+
AgentRPCMethod.TASK_INTERRUPT,
105+
],
90106
ACPType.AGENTIC: [
91107
AgentRPCMethod.TASK_CREATE,
92108
AgentRPCMethod.TASK_CANCEL,
109+
AgentRPCMethod.TASK_INTERRUPT,
93110
AgentRPCMethod.EVENT_SEND,
94111
],
95112
ACPType.ASYNC: [
96113
AgentRPCMethod.TASK_CREATE,
97114
AgentRPCMethod.TASK_CANCEL,
115+
AgentRPCMethod.TASK_INTERRUPT,
98116
AgentRPCMethod.EVENT_SEND,
99117
],
100118
}
@@ -132,6 +150,23 @@ def validate_task_identifiers(self):
132150
return self
133151

134152

153+
class InterruptTaskRequestEntity(BaseModel):
154+
task_id: str | None = Field(
155+
None,
156+
description="The ID of the task to interrupt. Either this or task_name must be provided.",
157+
)
158+
task_name: str | None = Field(
159+
None,
160+
description="The name of the task to interrupt. Either this or task_id must be provided.",
161+
)
162+
163+
@model_validator(mode="after")
164+
def validate_task_identifiers(self):
165+
if self.task_id is not None and self.task_name is not None:
166+
raise ValueError("Cannot provide both task_id and task_name - use only one")
167+
return self
168+
169+
135170
class SendMessageRequestEntity(BaseModel):
136171
task_id: str | None = Field(
137172
None, description="The ID of the task that the message was sent to"
@@ -180,6 +215,7 @@ class AgentRPCRequestEntity(JSONRPCRequest):
180215
params: (
181216
CreateTaskRequestEntity
182217
| CancelTaskRequestEntity
218+
| InterruptTaskRequestEntity
183219
| SendMessageRequestEntity
184220
| SendEventRequestEntity
185221
) = Field(..., description="The parameters for the agent RPC request")
@@ -201,6 +237,13 @@ def from_api_request(cls, request: AgentRPCRequest) -> Self:
201237
task_id=request.params.root.task_id,
202238
task_name=request.params.root.task_name,
203239
)
240+
elif request.method == AgentRPCMethod.TASK_INTERRUPT and isinstance(
241+
request.params.root, InterruptTaskRequest
242+
):
243+
params = InterruptTaskRequestEntity(
244+
task_id=request.params.root.task_id,
245+
task_name=request.params.root.task_name,
246+
)
204247
elif request.method == AgentRPCMethod.MESSAGE_SEND:
205248
content_entity = convert_task_message_content_to_entity(
206249
content=request.params.root.content.root

agentex/src/domain/entities/tasks.py

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,10 @@ class TaskStatus(str, Enum):
2020
COMPLETED = "COMPLETED"
2121
FAILED = "FAILED"
2222
RUNNING = "RUNNING"
23+
# Non-terminal resting state: the current turn was stopped by the user and the
24+
# task is waiting for the next message. Distinct from RUNNING (a turn is
25+
# actively in flight) and from the terminal statuses below.
26+
INTERRUPTED = "INTERRUPTED"
2327
TERMINATED = "TERMINATED"
2428
TIMED_OUT = "TIMED_OUT"
2529
DELETED = "DELETED"

0 commit comments

Comments
 (0)