feat(interrupt): task/interrupt hook, protocol, resume-safe session capture (AGX1-391)

Hand-editable SDK support for the platform's non-terminal task/interrupt verb
(the generated client.tasks.interrupt + param/response types + RPC unions
regenerate from the scale-agentex OpenAPI change via Stainless; see the TODO in
protocol/acp.py).

- protocol/acp.py: RPCMethod.TASK_INTERRUPT, InterruptTaskParams, PARAMS_MODEL_BY_METHOD
- Temporal ACP layer: SignalName.INTERRUPT_TURN and an overridable
  BaseWorkflow.on_interrupt hook (no-op default so existing agents are unaffected),
  TemporalTaskService.interrupt() that signals interrupt_turn (never cancels/terminates),
  on_task_interrupt ACP-server decorator + handle_interrupt forwarding
- resume fix: capture session_id from the early system/init (claude) and
  thread.started (codex) envelopes with a cancellation-safe finally, so a turn
  interrupted before completion stays resumable
- unit tests

Part of the Stop-mid-stream + queue-at-boundary effort (AGX1-391). Pairs with
scaleapi/scale-agentex#365.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: declan-scale

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

@@ -74,6 +74,33 @@ def _extract_summary(text: str, max_len: int = 300) -> str:
 async def convert_claude_code_to_agentex_events(
     lines: AsyncIterator[str | dict[str, Any]],
     on_result: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
+    on_init: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
+) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]:
+    """Public tap: convert a claude-code ``stream-json`` line stream to events.
+
+    Thin wrapper over :func:`_convert_claude_code_impl` that owns the
+    cancellation backstop: a ``finally`` closes the underlying ``lines`` iterator
+    (when it exposes ``aclose``) whenever this generator is closed — including on
+    the ``GeneratorExit``/``CancelledError`` raised when a consuming task is
+    cancelled mid-turn by an interrupt. This terminates the CLI stdout handle /
+    subprocess instead of leaking it. Kept as a wrapper (rather than a
+    ``try/finally`` inside the impl) so the large parser body stays untouched.
+    """
+    inner = _convert_claude_code_impl(lines, on_result=on_result, on_init=on_init)
+    try:
+        async for event in inner:
+            yield event
+    finally:
+        await inner.aclose()
+        aclose = getattr(lines, "aclose", None)
+        if aclose is not None:
+            await aclose()
+
+
+async def _convert_claude_code_impl(
+    lines: AsyncIterator[str | dict[str, Any]],
+    on_result: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
+    on_init: Callable[[dict[str, Any]], Awaitable[None]] | None = None,
 ) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]:
     """Convert a claude-code ``stream-json`` line stream into Agentex ``StreamTaskMessage*`` events.
 
@@ -85,7 +112,20 @@ async def convert_claude_code_to_agentex_events(
     caller can capture usage and cost. It is awaited before the generator
     continues. When ``None``, the result envelope is silently dropped.
 
+    ``on_init`` is called with the ``system``/``init`` envelope the moment it
+    arrives (the FIRST envelope of a claude-code stream), so the caller can
+    capture ``session_id`` EARLY — before the turn completes. This is what makes
+    an interrupted-before-completion turn resumable: the terminal ``result``
+    envelope (which ``on_result`` reads) never arrives when a turn is cut short,
+    so relying on it alone loses the session id. Mirrors how the inline
+    ``claude_agents`` activity captures session_id from its SystemMessage init.
+    When ``None``, the init envelope's session metadata is not surfaced early.
+
     Envelope → canonical mapping is documented in this module's docstring.
+
+    A ``finally`` closes the underlying ``lines`` iterator (when it exposes
+    ``aclose``) as a backstop, so a cancellation mid-turn (e.g. an interrupt that
+    cancels the consuming task) does not leak the CLI stdout handle / subprocess.
     """
     next_index = 0
     tool_call_count = 0
@@ -355,9 +395,14 @@ async def convert_claude_code_to_agentex_events(
         # system / init — session metadata (ignored at this layer)
         # -----------------------------------------------------------------------
         elif evt_type == "system":
-            # Session ID tracking and MCP status logging are provider concerns.
-            # This pure parser layer intentionally emits nothing for system events.
-            pass
+            # Session ID tracking and MCP status logging are provider concerns:
+            # this pure parser layer emits no StreamTaskMessage for system events.
+            # It DOES surface the init envelope's session metadata early via
+            # on_init so a caller can capture session_id before the turn's
+            # terminal ``result`` arrives — required for resuming a turn that was
+            # interrupted before completion (no ``result`` is ever emitted then).
+            if on_init is not None and evt.get("subtype") == "init":
+                await on_init(evt)
 
         # -----------------------------------------------------------------------
         # result — carries usage + cost; fired to on_result, not emitted as msgs

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

@@ -118,31 +118,45 @@ class ClaudeCodeTurn:
     def __init__(self, lines: AsyncIterator[str | dict[str, Any]]) -> None:
         self._lines = lines
         self._result_envelope: dict[str, Any] | None = None
+        # session_id captured from the early system/init envelope. This is what
+        # keeps an interrupted-before-completion turn resumable: the terminal
+        # ``result`` envelope never arrives when a turn is cut short, so we must
+        # capture the session id up front (from init) rather than only at the end.
+        self._init_session_id: str | None = None
         self._events_stream: AsyncIterator[StreamTaskMessage] | None = None
 
     async def _on_result(self, envelope: dict[str, Any]) -> None:
         self._result_envelope = envelope
 
+    async def _on_init(self, envelope: dict[str, Any]) -> None:
+        sid = envelope.get("session_id")
+        if sid:
+            self._init_session_id = sid
+
     @property
     def events(self) -> AsyncIterator[StreamTaskMessage]:
         if self._events_stream is None:
             self._events_stream = convert_claude_code_to_agentex_events(
                 self._lines,
                 on_result=self._on_result,
+                on_init=self._on_init,
             )
         return self._events_stream
 
     @property
     def session_id(self) -> str | None:
         """The Claude Code session id, for resuming a multi-turn session.
 
-        Valid only after ``events`` has been fully consumed (populated by the
-        ``result`` envelope). Returns ``None`` if the stream was truncated or
-        Claude Code reported no session id.
+        Prefers the id from the terminal ``result`` envelope (fully-complete
+        turn), then falls back to the id captured from the early ``system/init``
+        envelope. The init fallback is what makes a turn that was interrupted
+        before completion still resumable (no ``result`` is emitted then).
+        Returns ``None`` only if neither envelope was seen (e.g. the stream was
+        truncated before init) or Claude Code reported no session id.
         """
-        if not self._result_envelope:
-            return None
-        return self._result_envelope.get("session_id")
+        if self._result_envelope and self._result_envelope.get("session_id"):
+            return self._result_envelope.get("session_id")
+        return self._init_session_id
 
     def usage(self) -> TurnUsage:
         """Return normalised usage for this turn.

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

@@ -527,6 +527,32 @@ def _handle_item(self, evt_type: str, item: dict[str, Any]) -> list[StreamTaskMe
 async def convert_codex_to_agentex_events(
     events: AsyncIterator[str | dict[str, Any]],
     on_result: Callable[[dict[str, Any]], None] | None = None,
+    on_init: Callable[[dict[str, Any]], None] | None = None,
+) -> AsyncIterator[StreamTaskMessage]:
+    """Public tap: convert a ``codex exec --json`` event stream to events.
+
+    Thin wrapper over :func:`_convert_codex_impl` that owns the cancellation
+    backstop: a ``finally`` closes the underlying ``events`` iterator (when it
+    exposes ``aclose``) whenever this generator is closed — including on the
+    ``GeneratorExit``/``CancelledError`` raised when a consuming task is
+    cancelled mid-turn by an interrupt. This terminates the CLI stdout handle /
+    subprocess instead of leaking it.
+    """
+    inner = _convert_codex_impl(events, on_result=on_result, on_init=on_init)
+    try:
+        async for event in inner:
+            yield event
+    finally:
+        await inner.aclose()
+        aclose = getattr(events, "aclose", None)
+        if aclose is not None:
+            await aclose()
+
+
+async def _convert_codex_impl(
+    events: AsyncIterator[str | dict[str, Any]],
+    on_result: Callable[[dict[str, Any]], None] | None = None,
+    on_init: Callable[[dict[str, Any]], None] | None = None,
 ) -> AsyncIterator[StreamTaskMessage]:
     """Convert a ``codex exec --json`` event stream into Agentex stream events.
 
@@ -546,6 +572,12 @@ async def convert_codex_to_agentex_events(
                 ``reasoning_count`` — int
             Use this to record turn-level metrics / usage in the caller's span
             without coupling this module to span/tracing APIs.
+        on_init: Optional callback invoked once when the ``thread.started`` event
+            is seen (the first event of a codex stream). Receives ``{"session_id":
+            <thread_id or None>}``. This surfaces the session id EARLY — before
+            the turn completes — so a turn interrupted before completion is still
+            resumable (``turn.completed`` / ``on_result`` never fires then). The
+            codex counterpart of the claude-code ``system/init`` early capture.
 
     Yields:
         Canonical ``StreamTaskMessage*`` events (Start/Delta/Full/Done) with
@@ -590,6 +622,11 @@ async def convert_codex_to_agentex_events(
         for msg in messages:
             yield msg
 
+        # Surface session_id early (processor sets it while handling
+        # thread.started) so an interrupted turn is still resumable.
+        if on_init is not None and evt.get("type") == "thread.started":
+            on_init({"session_id": processor.session_id})
+
     if on_result is not None:
         on_result(
             {

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

@@ -160,6 +160,10 @@ def __init__(
 
         # Populated by the on_result callback once the stream is exhausted.
         self._result: dict[str, Any] | None = None
+        # Populated by the on_init callback when thread.started arrives (early),
+        # so session_id is available even if the turn is interrupted before
+        # completion (turn.completed / on_result never fires then).
+        self._init_session_id: str | None = None
         # The events generator is created at most once: ``_raw_events`` is a
         # single-consumption AsyncIterator, so re-wrapping it would yield an
         # already-exhausted stream that fires on_result with zeros and clobbers
@@ -179,21 +183,31 @@ def events(self) -> AsyncIterator[StreamTaskMessage]:
             self._events_gen = convert_codex_to_agentex_events(
                 self._raw_events,
                 on_result=self._on_result,
+                on_init=self._on_init,
             )
         return self._events_gen
 
     def _on_result(self, result: dict[str, Any]) -> None:
         self._result = result
 
+    def _on_init(self, init: dict[str, Any]) -> None:
+        sid = init.get("session_id")
+        if sid:
+            self._init_session_id = sid
+
     @property
     def session_id(self) -> str | None:
         """The codex session id, for resuming a multi-turn session.
 
-        Valid only after ``events`` has been fully consumed (populated by the
-        ``on_result`` callback). Returns ``None`` if the stream is not yet
-        exhausted or codex reported no session id.
+        Prefers the id from the terminal ``turn.completed`` result (fully-complete
+        turn), then falls back to the id captured early from ``thread.started``.
+        The early fallback keeps a turn interrupted before completion resumable
+        (no ``turn.completed`` / ``on_result`` fires then). Returns ``None`` only
+        if neither was seen or codex reported no session id.
         """
-        return self._result.get("session_id") if self._result else None
+        if self._result and self._result.get("session_id"):
+            return self._result.get("session_id")
+        return self._init_session_id
 
     def usage(self) -> TurnUsage:
         """Return normalized ``TurnUsage`` for this turn.

src/agentex/lib/core/temporal/services/temporal_task_service.py

@@ -6,7 +6,7 @@
 from agentex.types.task import Task
 from agentex.types.agent import Agent
 from agentex.types.event import Event
-from agentex.protocol.acp import SendEventParams, CreateTaskParams
+from agentex.protocol.acp import SendEventParams, CreateTaskParams, InterruptTaskParams
 from agentex.lib.environment_variables import EnvironmentVariables
 from agentex.lib.core.clients.temporal.types import WorkflowState
 from agentex.lib.core.temporal.types.workflow import SignalName
@@ -74,6 +74,23 @@ async def send_event(self, agent: Agent, task: Task, event: Event, request: dict
             ).model_dump(),
         )
 
+    async def interrupt(self, agent: Agent, task: Task, request: dict | None = None) -> None:
+        """Forward a task/interrupt to the running workflow as a dedicated signal.
+
+        Non-terminal: unlike ``cancel``/``terminate`` this does NOT tear down the
+        workflow. It signals ``interrupt_turn`` so the workflow's ``on_interrupt``
+        hook can stop the in-flight turn while leaving the task continuable.
+        """
+        return await self._temporal_client.send_signal(
+            workflow_id=task.id,
+            signal=SignalName.INTERRUPT_TURN.value,
+            payload=InterruptTaskParams(
+                agent=agent,
+                task=task,
+                request=request,
+            ).model_dump(),
+        )
+
     async def cancel(self, task_id: str) -> None:
         return await self._temporal_client.cancel_workflow(
             workflow_id=task_id,

src/agentex/lib/core/temporal/types/workflow.py

@@ -3,3 +3,8 @@
 
 class SignalName(str, Enum):
     RECEIVE_EVENT = "receive_event"
+    # Dedicated non-terminal "stop the current turn" signal (design doc section 7).
+    # Routed to the overridable BaseWorkflow.on_interrupt hook. Kept separate from
+    # RECEIVE_EVENT so the interrupt handler can interleave with (and cancel) the
+    # in-flight turn WITHOUT waiting on the turn lock the running turn holds.
+    INTERRUPT_TURN = "interrupt_turn"

src/agentex/lib/core/temporal/workflows/workflow.py

@@ -6,7 +6,7 @@
 
 from temporalio import workflow
 
-from agentex.protocol.acp import SendEventParams, CreateTaskParams
+from agentex.protocol.acp import SendEventParams, CreateTaskParams, InterruptTaskParams
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.core.temporal.types.workflow import SignalName
 
@@ -160,3 +160,32 @@ def is_continued_run(self) -> bool:
         gated here runs again on each history rollover.
         """
         return workflow.info().continued_run_id is not None
+
+    @workflow.signal(name=SignalName.INTERRUPT_TURN)
+    async def on_interrupt(self, params: InterruptTaskParams) -> None:
+        """Handle a task/interrupt: stop the in-flight turn, keep the task continuable.
+
+        This is the durable transport for the platform's ``task/interrupt`` verb
+        (design doc section 7). The control plane forwards ``task/interrupt`` to the
+        agent pod over HTTP JSON-RPC; the ACP/Temporal layer routes it to the
+        ``interrupt_turn`` signal, which invokes this hook.
+
+        Temporal runs signal handlers on the same deterministic event loop as the
+        workflow ``run`` method, interleaving at ``await`` points. This handler runs
+        CONCURRENTLY with the in-flight turn coroutine, so it MUST NOT acquire the
+        turn lock (the running turn already holds it) — doing so would deadlock: the
+        interrupt could never fire because it would be waiting on the very turn it is
+        meant to stop.
+
+        The base implementation is a no-op so existing agents keep working (and the
+        signal is still accepted, avoiding "unhandled signal" warnings). Interruptible
+        agents (the golden agent is the reference) override this to actually stop the
+        in-flight model / CLI subprocess — e.g. SIGINT the leased sandbox CLI by pid,
+        or cancel the inline model ``asyncio.Task`` — while preserving partial output
+        and resume state so the next turn continues the conversation.
+        """
+        logger.info(
+            "on_interrupt (no-op base impl) for task %s; override to make this agent "
+            "interruptible",
+            params.task.id,
+        )

src/agentex/lib/sdk/fastacp/base/base_acp_server.py

@@ -22,6 +22,7 @@
     CancelTaskParams,
     CreateTaskParams,
     SendMessageParams,
+    InterruptTaskParams,
 )
 from agentex.lib.utils.logging import make_logger, ctx_var_request_id
 from agentex.protocol.json_rpc import JSONRPCError, JSONRPCRequest, JSONRPCResponse
@@ -68,6 +69,7 @@ class BaseACPServer(FastAPI):
     Available methods:
     - event/send → Send a message to a task
     - task/cancel → Cancel a task
+    - task/interrupt → Interrupt the in-flight turn (non-terminal; task stays continuable)
     - task/approve → Approve a task
     """
 
@@ -324,6 +326,7 @@ async def _process_request(
     - on_task_create
     - on_task_event_send
     - on_task_cancel
+    - on_task_interrupt
 
     ACP Type: Sync
     Decorators:
@@ -362,6 +365,18 @@ def on_task_cancel(self, fn: Callable[[CancelTaskParams], Awaitable[Any]]):
         self._handlers[RPCMethod.TASK_CANCEL] = wrapped
         return fn
 
+    # Type: Async
+    def on_task_interrupt(self, fn: Callable[[InterruptTaskParams], Awaitable[Any]]):
+        """Handle task/interrupt method.
+
+        Non-terminal counterpart to ``on_task_cancel``: forwards the interrupt to
+        the agent so it can stop the in-flight turn while leaving the task
+        continuable. See the interrupt-and-queue design doc, section 7.
+        """
+        wrapped = self._wrap_handler(fn)
+        self._handlers[RPCMethod.TASK_INTERRUPT] = wrapped
+        return fn
+
     # Type: Sync
     def on_message_send(
         self,

src/agentex/lib/sdk/fastacp/impl/temporal_acp.py

@@ -10,6 +10,7 @@
     SendEventParams,
     CancelTaskParams,
     CreateTaskParams,
+    InterruptTaskParams,
 )
 from agentex.lib.utils.logging import make_logger
 from agentex.lib.environment_variables import EnvironmentVariables
@@ -132,3 +133,21 @@ async def handle_cancel(params: CancelTaskParams) -> None:
             except Exception as e:
                 logger.error(f"Failed to cancel task: {e}")
                 raise
+
+        @self.on_task_interrupt
+        async def handle_interrupt(params: InterruptTaskParams) -> None:
+            """Forward task/interrupt to the running workflow via TaskService.
+
+            Non-terminal: signals the workflow's ``interrupt_turn`` handler rather
+            than tearing the workflow down, so the task stays continuable.
+            """
+            try:
+                if self._temporal_task_service is not None:
+                    await self._temporal_task_service.interrupt(
+                        agent=params.agent,
+                        task=params.task,
+                        request=params.request,
+                    )
+            except Exception as e:
+                logger.error(f"Failed to interrupt task: {e}")
+                raise

src/agentex/lib/types/acp.py

@@ -12,4 +12,5 @@
     CancelTaskParams,
     CreateTaskParams,
     SendMessageParams,
+    InterruptTaskParams,
 )