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

[declan-scale]

Summary

The agent-runtime (hand-authored) half of AGX1-391 (stop an agent run mid-stream). scale-agentex#365 is merged, so the generated client method client.tasks.interrupt(...) and the Task INTERRUPTED status already regenerated via Stainless; this PR adds the agentex.lib runtime support that makes an agent actually interruptible.

What's included (all in the hand-editable runtime, not generated code)

• Signal + hook: SignalName.INTERRUPT_TURN and an overridable BaseWorkflow.on_interrupt(params) hook (no-op default, so existing agents are unaffected and the signal is still accepted).
• ACP forwarding: an on_task_interrupt ACP-server decorator + handle_interrupt; TemporalTaskService.interrupt() signals interrupt_turn (never cancels/terminates the workflow).
• Agent records status: adk.tasks.interrupt(task_id=...) across the facade → activity → service → worker-registration layers (mirrors adk.tasks.cancel), calling the generated client.tasks.interrupt. Agents call this from their interrupt handler, after stopping the turn, to record the non-terminal INTERRUPTED status.
• Protocol: RPCMethod.TASK_INTERRUPT + InterruptTaskParams in protocol/acp.py.
• Resume-safe session_id: captured from the early system/init (claude) / thread.started (codex) envelopes, with a cancellation-safe finally, so a turn interrupted before completion is still resumable.

Model

Interrupt is cooperative: the control plane forwards task/interrupt (courier) and owns RUNNING (task creation + auto-resume); the agent stops its own turn and records INTERRUPTED via adk.tasks.interrupt. An agent that does not implement the hook simply stays RUNNING. Reference consumer: the golden agent (agentex-agents).

Base / testing

Targets next (Stainless convention). lint / build / conformance and the agent-SDK matrix (claude-code / codex / openai / langgraph, sync+async+temporal) are green.

Greptile Summary

This PR adds the agent-runtime half of the cooperative interrupt feature (AGX1-391): a dedicated interrupt_turn Temporal signal, an overridable BaseWorkflow.on_interrupt hook, ACP routing via task/interrupt, and the adk.tasks.interrupt(...) call chain (facade → activity → service → generated client). It also fixes resume-safety by capturing session_id from the early system/init (claude-code) and thread.started (codex) envelopes via a new on_init callback, so a turn interrupted before its terminal result envelope arrives remains resumable.

• Stream converters (_claude_code_sync.py, _codex_sync.py, _langgraph_sync.py, _openai_sync.py, _pydantic_ai_sync.py) are each split into a public wrapper + private impl, where the wrapper owns a cancellation-safe finally that calls aclose() on both the inner generator and the underlying source iterator to prevent subprocess/handle leaks on mid-turn cancellation.
• Protocol + transport: RPCMethod.TASK_INTERRUPT, InterruptTaskParams, SignalName.INTERRUPT_TURN, and TemporalTaskService.interrupt() are added — the signal path deliberately avoids cancel_workflow/terminate_workflow so the task stays continuable.
• Backward compatibility: BaseWorkflow.on_interrupt is non-abstract with a logging no-op default, so agents that don't implement the hook keep working and the unhandled-signal warning is suppressed.

Confidence Score: 5/5

Safe to merge. The interrupt path is non-terminal (signal only, never cancel/terminate), the no-op base implementation preserves backward compatibility, and all new call-chain layers mirror the well-tested cancel pattern.

The implementation is internally consistent across all five stream-converter modules and all four call-chain layers (facade → activity → service → client). The cooperative-interrupt model correctly avoids the turn-lock deadlock, the session_id early-capture logic is sound, and the cleanup backstops correctly propagate aclose() through wrapper → impl → source. The only gap is that the codex interrupted-before-completion unit test accidentally validates its assertion via the on_result path rather than the _init_session_id fallback it claims to exercise.

tests/lib/adk/test_codex_turn.py — the session-id fallback test for the interrupted path does not isolate the _init_session_id mechanism.

Important Files Changed

Filename	Overview
src/agentex/lib/adk/_modules/_claude_code_sync.py	Refactored public function into a wrapper+impl pair; the wrapper owns cancellation-safe finally cleanup (aclose on both inner generator and underlying lines iterator) and adds the on_init callback to surface session_id from the early system/init envelope before the terminal result arrives.
src/agentex/lib/adk/_modules/_codex_sync.py	Same wrapper+impl refactor as claude_code; on_init fires synchronously after thread.started is processed (no yield between processing and on_init call, so session_id is captured before the consumer can cancel), matching the existing sync on_result pattern for codex.
src/agentex/lib/adk/_modules/tasks.py	Added interrupt() method mirroring cancel(): routes to INTERRUPT_TASK Temporal activity inside a workflow, or calls tasks_service.interrupt_task directly outside Temporal. Default timeouts and retry policy are identical to cancel.
src/agentex/lib/core/temporal/workflows/workflow.py	Adds on_interrupt as a non-abstract @workflow.signal(interrupt_turn); the no-op base implementation keeps existing agents valid while the dedicated signal avoids the turn-lock deadlock that would occur with RECEIVE_EVENT.
src/agentex/lib/core/temporal/services/temporal_task_service.py	interrupt() forwards the interrupt as a signal (not cancel/terminate) to preserve the workflow so the task stays continuable; pattern matches the existing send_event() method.
src/agentex/protocol/acp.py	Adds RPCMethod.TASK_INTERRUPT, InterruptTaskParams (mirrors CancelTaskParams shape), and registers it in PARAMS_MODEL_BY_METHOD; clean protocol extension with no impact on existing methods.
tests/test_acp_interrupt.py	Comprehensive new test file covering protocol shape, ACP routing registration, Temporal signal name/non-abstract validation, and signal-not-cancel behavior; good coverage of the integration path.
tests/lib/adk/test_codex_turn.py	Adds session-id capture and guarded-events tests; the interrupted-before-completion test drains the stream fully so on_result fires, meaning the _init_session_id fallback path is not directly exercised.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant CP as Control Plane
    participant ACP as TemporalACP
    participant TTS as TemporalTaskService
    participant TW as Temporal Workflow
    participant Agent as Agent turn coroutine
    participant ADK as adk.tasks.interrupt
    participant API as client.tasks.interrupt

    CP->>ACP: POST task/interrupt
    ACP->>TTS: interrupt(agent, task, request)
    TTS->>TW: send_signal interrupt_turn
    Note over TW: on_interrupt fires concurrently with running turn
    TW->>Agent: stop in-flight turn
    Agent-->>TW: turn stopped, partial output preserved
    Agent->>ADK: interrupt(task_id)
    ADK->>API: tasks.interrupt(task_id)
    API-->>Agent: "Task status=INTERRUPTED"
    Note over TW: Task stays continuable for next turn

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant CP as Control Plane
    participant ACP as TemporalACP
    participant TTS as TemporalTaskService
    participant TW as Temporal Workflow
    participant Agent as Agent turn coroutine
    participant ADK as adk.tasks.interrupt
    participant API as client.tasks.interrupt

    CP->>ACP: POST task/interrupt
    ACP->>TTS: interrupt(agent, task, request)
    TTS->>TW: send_signal interrupt_turn
    Note over TW: on_interrupt fires concurrently with running turn
    TW->>Agent: stop in-flight turn
    Agent-->>TW: turn stopped, partial output preserved
    Agent->>ADK: interrupt(task_id)
    ADK->>API: tasks.interrupt(task_id)
    API-->>Agent: "Task status=INTERRUPTED"
    Note over TW: Task stays continuable for next turn

Loading

_{Reviews (2): Last reviewed commit: "fix(adk): cancellation-safe teardown for..." | Re-trigger Greptile}