Skip to content

Commit 2b0611b

Browse files
declan-scaleclaude
andcommitted
feat(adk): add adk.tasks.interrupt for agents to record INTERRUPTED (AGX1-391)
Agents call adk.tasks.interrupt(task_id=...) from their interrupt handler, after stopping the in-flight turn, to record the non-terminal INTERRUPTED status (the control plane forwards task/interrupt but no longer writes status). Mirrors adk.tasks.cancel across the facade / activity / service layers and registers the interrupt-task activity. The service uses a generic POST /tasks/{id}/interrupt for now; switches to client.tasks.interrupt once Stainless regenerates the typed method. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1 parent bbdbc83 commit 2b0611b

4 files changed

Lines changed: 84 additions & 0 deletions

File tree

src/agentex/lib/adk/_modules/tasks.py

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,53 @@ async def cancel(
174174
parent_span_id=parent_span_id,
175175
)
176176

177+
async def interrupt(
178+
self,
179+
*,
180+
task_id: str,
181+
reason: str | None = None,
182+
trace_id: str | None = None,
183+
parent_span_id: str | None = None,
184+
start_to_close_timeout: timedelta = timedelta(seconds=5),
185+
heartbeat_timeout: timedelta = timedelta(seconds=5),
186+
retry_policy: RetryPolicy = DEFAULT_RETRY_POLICY,
187+
) -> Task:
188+
"""
189+
Mark a running task as interrupted (non-terminal).
190+
191+
Interrupt is cooperative: an agent calls this from its interrupt handler,
192+
after it has actually stopped its in-flight turn, to record INTERRUPTED.
193+
The task stays continuable; the control plane resumes it to RUNNING on the
194+
next turn.
195+
Args:
196+
task_id: The ID of the task to interrupt.
197+
reason: Optional reason for the interrupt.
198+
Returns:
199+
The updated task entry.
200+
"""
201+
params = TaskStatusTransitionParams(
202+
task_id=task_id,
203+
reason=reason,
204+
trace_id=trace_id,
205+
parent_span_id=parent_span_id,
206+
)
207+
if in_temporal_workflow():
208+
return await ActivityHelpers.execute_activity(
209+
activity_name=TasksActivityName.INTERRUPT_TASK,
210+
request=params,
211+
response_type=Task,
212+
start_to_close_timeout=start_to_close_timeout,
213+
retry_policy=retry_policy,
214+
heartbeat_timeout=heartbeat_timeout,
215+
)
216+
else:
217+
return await self._tasks_service.interrupt_task(
218+
task_id=task_id,
219+
reason=reason,
220+
trace_id=trace_id,
221+
parent_span_id=parent_span_id,
222+
)
223+
177224
async def complete(
178225
self,
179226
*,

src/agentex/lib/core/services/adk/tasks.py

Lines changed: 26 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,32 @@ async def cancel_task(
9898
span.output = task_model.model_dump()
9999
return task_model
100100

101+
async def interrupt_task(
102+
self,
103+
task_id: str,
104+
reason: str | None = None,
105+
trace_id: str | None = None,
106+
parent_span_id: str | None = None,
107+
) -> Task:
108+
trace = self._tracer.trace(trace_id)
109+
async with trace.span(
110+
parent_id=parent_span_id,
111+
name="interrupt_task",
112+
input={"task_id": task_id, "reason": reason},
113+
) as span:
114+
heartbeat_if_in_workflow("interrupt task")
115+
# TODO(AGX1-391): switch to self._agentex_client.tasks.interrupt(...) once the
116+
# POST /tasks/{id}/interrupt endpoint is added to the Stainless config and
117+
# regenerated. Using the generic post keeps this working before regeneration.
118+
task_model = await self._agentex_client.post(
119+
f"/tasks/{task_id}/interrupt",
120+
cast_to=Task,
121+
body={"reason": reason},
122+
)
123+
if span:
124+
span.output = task_model.model_dump()
125+
return task_model
126+
101127
async def complete_task(
102128
self,
103129
task_id: str,

src/agentex/lib/core/temporal/activities/__init__.py

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,7 @@ def get_all_activities(sgp_client=None):
181181
tasks_activities.get_task,
182182
tasks_activities.delete_task,
183183
tasks_activities.cancel_task,
184+
tasks_activities.interrupt_task,
184185
tasks_activities.complete_task,
185186
tasks_activities.fail_task,
186187
tasks_activities.terminate_task,

src/agentex/lib/core/temporal/activities/adk/tasks_activities.py

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ class TasksActivityName(str, Enum):
1818
GET_TASK = "get-task"
1919
DELETE_TASK = "delete-task"
2020
CANCEL_TASK = "cancel-task"
21+
INTERRUPT_TASK = "interrupt-task"
2122
COMPLETE_TASK = "complete-task"
2223
FAIL_TASK = "fail-task"
2324
TERMINATE_TASK = "terminate-task"
@@ -83,6 +84,15 @@ async def cancel_task(self, params: TaskStatusTransitionParams) -> Task:
8384
parent_span_id=params.parent_span_id,
8485
)
8586

87+
@activity.defn(name=TasksActivityName.INTERRUPT_TASK)
88+
async def interrupt_task(self, params: TaskStatusTransitionParams) -> Task:
89+
return await self._tasks_service.interrupt_task(
90+
task_id=params.task_id,
91+
reason=params.reason,
92+
trace_id=params.trace_id,
93+
parent_span_id=params.parent_span_id,
94+
)
95+
8696
@activity.defn(name=TasksActivityName.COMPLETE_TASK)
8797
async def complete_task(self, params: TaskStatusTransitionParams) -> Task:
8898
return await self._tasks_service.complete_task(

0 commit comments

Comments
 (0)