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

Author: declan-scale

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

@@ -74,6 +74,35 @@ 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:
+        inner_aclose = getattr(inner, "aclose", None)
+        if inner_aclose is not None:
+            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 +114,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 +397,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,34 @@ 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:
+        inner_aclose = getattr(inner, "aclose", None)
+        if inner_aclose is not None:
+            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 +574,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 +624,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/adk/_modules/_langgraph_sync.py

@@ -32,6 +32,27 @@
 async def convert_langgraph_to_agentex_events(
     stream: Any,
     on_final_ai_message: Optional[Callable[..., None]] = None,
+) -> AsyncGenerator[Any, None]:
+    """Public LangGraph tap: convert events, closing the source stream on exit.
+
+    Thin wrapper over ``_convert_langgraph_impl`` that adds a cancellation-safe
+    ``finally`` so an interrupted turn tears down the LangGraph ``astream`` source
+    instead of leaking it.
+    """
+    inner = _convert_langgraph_impl(stream, on_final_ai_message=on_final_ai_message)
+    try:
+        async for event in inner:
+            yield event
+    finally:
+        for _src in (inner, stream):
+            _aclose = getattr(_src, "aclose", None)
+            if _aclose is not None:
+                await _aclose()
+
+
+async def _convert_langgraph_impl(
+    stream: Any,
+    on_final_ai_message: Optional[Callable[..., None]] = None,
 ) -> AsyncGenerator[Any, None]:
     """Convert LangGraph streaming events to Agentex TaskMessageUpdate events.
 

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

@@ -145,6 +145,25 @@ def _extract_tool_response_info(tool_map: dict[str, Any], tool_output_item: Any)
 
 
 async def convert_openai_to_agentex_events(stream_response):
+    """Public OpenAI tap: parse the event stream, closing the source on exit.
+
+    Thin wrapper over ``_convert_openai_impl`` that adds a cancellation-safe
+    ``finally`` so an interrupted turn tears down the source event stream instead
+    of leaking it. (Resume state is carried by the OpenAI Agents SDK input list,
+    not a session id, so there is no early-session_id capture like the CLI taps.)
+    """
+    inner = _convert_openai_impl(stream_response)
+    try:
+        async for event in inner:
+            yield event
+    finally:
+        for _src in (inner, stream_response):
+            _aclose = getattr(_src, "aclose", None)
+            if _aclose is not None:
+                await _aclose()
+
+
+async def _convert_openai_impl(stream_response):
     """Convert OpenAI streaming events to AgentEx TaskMessageUpdate events with reasoning support.
 
     This is an enhanced version of the base converter that includes support for:

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

@@ -120,6 +120,27 @@ def _tool_return_content(result: ToolReturnPart | Any) -> Any:
 async def convert_pydantic_ai_to_agentex_events(
     stream_response: AsyncIterator[Any],
     on_result: Callable[[AgentRunResultEvent], Any] | None = None,
+) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]:
+    """Public Pydantic AI tap: convert events, closing the source stream on exit.
+
+    Thin wrapper over ``_convert_pydantic_ai_impl`` that adds a cancellation-safe
+    ``finally`` so an interrupted turn tears down the source stream instead of
+    leaking it.
+    """
+    inner = _convert_pydantic_ai_impl(stream_response, on_result=on_result)
+    try:
+        async for event in inner:
+            yield event
+    finally:
+        for _src in (inner, stream_response):
+            _aclose = getattr(_src, "aclose", None)
+            if _aclose is not None:
+                await _aclose()
+
+
+async def _convert_pydantic_ai_impl(
+    stream_response: AsyncIterator[Any],
+    on_result: Callable[[AgentRunResultEvent], Any] | None = None,
 ) -> AsyncIterator[StreamTaskMessageStart | StreamTaskMessageDelta | StreamTaskMessageFull | StreamTaskMessageDone]:
     """Convert a Pydantic AI agent event stream into Agentex stream events.
 

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

@@ -174,6 +174,53 @@ async def cancel(
                 parent_span_id=parent_span_id,
             )
 
+    async def interrupt(
+        self,
+        *,
+        task_id: str,
+        reason: str | None = None,
+        trace_id: str | None = None,
+        parent_span_id: str | None = None,
+        start_to_close_timeout: timedelta = timedelta(seconds=5),
+        heartbeat_timeout: timedelta = timedelta(seconds=5),
+        retry_policy: RetryPolicy = DEFAULT_RETRY_POLICY,
+    ) -> Task:
+        """
+        Mark a running task as interrupted (non-terminal).
+
+        Interrupt is cooperative: an agent calls this from its interrupt handler,
+        after it has actually stopped its in-flight turn, to record INTERRUPTED.
+        The task stays continuable; the control plane resumes it to RUNNING on the
+        next turn.
+        Args:
+            task_id: The ID of the task to interrupt.
+            reason: Optional reason for the interrupt.
+        Returns:
+            The updated task entry.
+        """
+        params = TaskStatusTransitionParams(
+            task_id=task_id,
+            reason=reason,
+            trace_id=trace_id,
+            parent_span_id=parent_span_id,
+        )
+        if in_temporal_workflow():
+            return await ActivityHelpers.execute_activity(
+                activity_name=TasksActivityName.INTERRUPT_TASK,
+                request=params,
+                response_type=Task,
+                start_to_close_timeout=start_to_close_timeout,
+                retry_policy=retry_policy,
+                heartbeat_timeout=heartbeat_timeout,
+            )
+        else:
+            return await self._tasks_service.interrupt_task(
+                task_id=task_id,
+                reason=reason,
+                trace_id=trace_id,
+                parent_span_id=parent_span_id,
+            )
+
     async def complete(
         self,
         *,

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

@@ -98,6 +98,25 @@ async def cancel_task(
                 span.output = task_model.model_dump()
             return task_model
 
+    async def interrupt_task(
+        self,
+        task_id: str,
+        reason: str | None = None,
+        trace_id: str | None = None,
+        parent_span_id: str | None = None,
+    ) -> Task:
+        trace = self._tracer.trace(trace_id)
+        async with trace.span(
+            parent_id=parent_span_id,
+            name="interrupt_task",
+            input={"task_id": task_id, "reason": reason},
+        ) as span:
+            heartbeat_if_in_workflow("interrupt task")
+            task_model = await self._agentex_client.tasks.interrupt(task_id=task_id, reason=reason)
+            if span:
+                span.output = task_model.model_dump()
+            return task_model
+
     async def complete_task(
         self,
         task_id: str,