[4.41/C7] Turn cancellation: abort_turn, thread.signal, typing options, agent-session events

[patrick-chinchill]

Summary

Port the core half of upstream's Agent Sessions work (2ce2be00): a per-turn cancellation signal (thread.signal), Chat.abort_turn(thread_id) (works across processes sharing a state backend, for adapters that opt in), typing lifecycle hooks (TypingOptions.initiator_user_id, optional Adapter.end_typing, session_status), and handlers for agent-session stop/title-change events. No default behavior changes: polling runs only for adapters with supports_turn_cancellation. The Slack emitter (native stop, agents.sessions.*) is #215.

Upstream changes

• 2ce2be00 feat(slack): add Agent Sessions lifecycle and native stop (#862) — chat@4.39.0. Core pieces:
  • chat.ts: activeTurnControllers, ACTIVE_TURN_TTL_MS (1h), ABORT_POLL_INTERVAL_MS (250). abortTurn() aborts local controllers, then copies active-turn:{threadId} into abort-turn:{threadId}. Dispatch runs under a per-turn AbortController; with adapter.supportsTurnCancellation it publishes active-turn, polls abort-turn (monitorTurnAbort) and clears only its own markers (clearTurnMarkers). New onAgentSessionStopped/onAgentSessionTitleChanged + processAgentSession*, dispatched via runInConversation.
  • thread.ts: signal (never aborted by default) and a takeUntilAborted stream wrapper; startTyping passes { initiatorUserId } when the current author has a user id and sets _typingStarted; post/postable/fallback-stream call endTyping(id, options.sessionStatus ?? "active") in finally; a successful native stream clears the flag without endTyping.
  • types.ts: AgentSessionStatus ("active" | "closed" | "processing" | "suspended"), TypingOptions, AgentSessionStoppedEvent/TitleChangedEvent + handlers; Adapter.endTyping?, supportsTurnCancellation?, startTyping(threadId, status?, options?); StreamOptions.signal/sessionStatus. streaming-plan.ts: sessionStatus. index.ts: exports.

Current Python behavior

• grep -rn 'abort_turn\|supports_turn_cancellation\|end_typing\|session_status\|TypingOptions\|initiator_user_id\|active-turn' src/chat_sdk → nothing. Linear's unrelated AgentSession* types live in adapters/linear/.
• src/chat_sdk/types.py:1460: the Adapter Protocol has start_typing(thread_id, status=None). All 10 in-repo adapters use that two-argument signature (e.g. adapters/slack/adapter.py:3845, adapters/teams/adapter.py:1673), as does the mock (shared/mock_adapter.py:132).
• thread.py:998-999: start_typing just calls adapter.start_typing(self._id, status). channel.py:403 does the same.
• StreamOptions (types.py:1056-1063) has no signal/session_status. StreamingPlanOptions (plan.py:528) has no session_status.
• chat.py:2293 _dispatch_to_handlers has no per-turn token. _create_thread (chat.py:2394) takes no signal.
• The ChatInstance Protocol (types.py:1748-1795) has no abort_turn or agent-session process_* methods.

Scope

• types.py: AgentSessionStatus = Literal["active", "closed", "processing", "suspended"]; @dataclass TypingOptions(initiator_user_id: str | None = None); AgentSessionStoppedEvent (adapter, channel_id, streaming_message_ts: list[str], thread_id, thread_ts, user_id) and AgentSessionTitleChangedEvent (adds title, previous_title: str | None) + handler aliases; StreamOptions.signal/session_status.
• A cancellation signal type (see Porting notes), exposed as ThreadImpl.signal, never aborted by default. Pass it through _ThreadImplConfig and _create_thread(..., signal=None).
• thread.py: wrap _from_full_stream output in _take_until_aborted(source, signal) and set options.signal; start_typing passes TypingOptions when current_message.author.user_id is truthy (upstream truthiness) and sets _typing_started; post(), postable posting and _fallback_stream call _finish_typing(status) in finally; a successful native stream() resets the flag without end_typing, but an adapter.stream exception does call it.
• plan.py: StreamingPlanOptions.session_status, mapped into StreamOptions.
• chat.py: on_agent_session_stopped/on_agent_session_title_changed (decorator-returning like other on_*); process_agent_session_stopped/_title_changed (fire-and-forget task, errors logged, wait_until honored); async abort_turn(thread_id); _active_turn_signals: dict[str, dict[str, Signal]]; a turn wrapper around _dispatch_to_handlers (chat.py:2293) that publishes/polls only when getattr(adapter, "supports_turn_cancellation", False).
• BaseAdapter: supports_turn_cancellation property (default False) and an optional end_typing(thread_id, status=None). Leave both off the Adapter Protocol.
• Add the new process methods to ChatInstance, following the optional-member convention [4.41/C4] Core lifecycle events: message updated/deleted, installed/uninstalled, app context changed #196 sets for its new members.
• Mock adapter: record start_typing(..., options=) and add an opt-in end_typing. Export the new public names.

Out of scope

• Slack agents.sessions.*, agent_session_stopped parsing, native stop, auto titles, Slack end_typing, agent_view config: [4.41/SL9] Slack agent_view (Agent messaging experience) + declarative agent config #214/[4.41/SL10] Slack Agent Sessions lifecycle + native stop #215.
• Implementing conversation context itself (runInConversation): [4.41/C3] Conversation context + AI tool scoping (read & write guards, strict_scope) #195. Any non-mock end_typing implementation.

Porting notes

• Signal type. Python has no AbortSignal. Recommended default: a small public class (e.g. TurnSignal) with an aborted property, async wait() backed by an asyncio.Event, internal _abort(), optional add_listener(cb). A module-level "never aborted" instance must not bind an Event to a loop at import: create lazily, or make wait() block on a per-call Future.
• _take_until_aborted. Race anext(it) against signal.wait() (asyncio.wait(..., FIRST_COMPLETED)). On abort: cancel the pending anext task, await it (suppress only that task's CancelledError), then await agen.aclose() — aclose() on a still-running generator raises RuntimeError. Never swallow a CancelledError aimed at the caller.
• Monitor loop. An asyncio.Task that waits 250ms or wakes on a stop Event. In finally: set stop, cancel, await the task, then clear markers — only if they still hold this turn's id (get and compare). State errors log a warning and end polling (upstream). Cost: one GET per 250ms per active turn, opted-in adapters only.
• State keys/TTL. active-turn:{thread_id} / abort-turn:{thread_id} stay byte-identical (cross-SDK state sharing); TTL in ms (3_600_000) for StateAdapter.set; turn id str(uuid.uuid4()) (correlation id, not a secret).
• start_typing compatibility. A third argument breaks two-argument custom adapters. Recommended default: pass options= only when a cached inspect.signature probe accepts it (or **kwargs); helper shared with [4.41/C6] Thread.reply, Thread.mark_as_read, post_ephemeral options #200 (first PR adds it). Update all 10 in-repo adapters to start_typing(self, thread_id, status=None, *, options: TypingOptions | None = None).
• end_typing status. Default "active". Use is not None when reading session_status from options.
• Stopped handlers must not take the thread lock. They run while the turn being stopped holds the lock.
• Wrap process_agent_session_* dispatch in [4.41/C3] Conversation context + AI tool scoping (read & write guards, strict_scope) #195's conversation-context helper (upstream runInConversation); C3 is merged before this lands (transitively via [4.41/C4] Core lifecycle events: message updated/deleted, installed/uninstalled, app context changed #196).
• No datetimes are introduced; use loop.time() for interval bookkeeping.

Tests

• packages/chat/src/chat.test.ts → tests/test_chat_faithful.py (mapped): "aborts an active thread signal from another Chat instance". Use two Chat instances sharing one MockStateAdapter, a supports_turn_cancellation=True mock, and an asyncio.Event for "started". No sleeps: advance by awaiting the signal.
• packages/chat/src/thread.test.ts → tests/test_thread_faithful.py (mapped) [startTyping]: "passes the initiating user and clears processing after posting".
• packages/chat/src/agent-session.test.ts (not in MAPPING; [4.41/P0] Fidelity tooling for the 4.41 wave: single pin constant, SHA pin, it.each expansion, map new core test files #185 plans to map it) → new tests/test_agent_session.py: "dispatches stop events to registered handlers", "dispatches title changes to registered handlers".
• Python-specific (AsyncMock throughout; patch the poll interval to 0 rather than sleeping): abort mid-stream closes the source generator (aclose ran) and the fallback final edit happens; no monitor task left after dispatch; markers owned by a newer turn are not cleared; a two-argument custom start_typing still works; no end_typing after a successful native stream.

Acceptance criteria

• Full validation command from CLAUDE.md passes (including the audit for unawaited coroutines).
• The four mapped/new upstream tests pass. Non-strict fidelity against chat@4.41.1 reports them matched.
• docs/UPSTREAM_SYNC.md records AbortSignal → the Python signal class and the start_typing signature probe.
• CHANGELOG entry under "Unreleased (4.41 wave)".
• Consumer-visible changes called out: start_typing may receive options=; thread.signal exists; no default behavior change for Slack/Teams streaming.

Dependencies

Blocked by #199 and #196 (ChatInstance member convention; transitively #195 for conversation context). Both C4 and C7 add process_* members to chat.py/ChatInstance, so they are serialized to avoid conflicts. Blocks #215, #203.

Metadata

• Effort: L
• Consumer impact: low. It is opt-in: no polling or state writes unless an adapter sets supports_turn_cancellation. The only surface change for Slack/Teams users is start_typing gaining a keyword-only options.
• Suggested branch: sync/4.41-c7

Part of #184.