From 63658ee7949102e9e1e20126fde50336b11a2663 Mon Sep 17 00:00:00 2001 From: patrick-chinchill Date: Sat, 20 Jun 2026 00:06:40 -0700 Subject: [PATCH 1/2] =?UTF-8?q?feat(linear):=20agent-activity=20emit=20?= =?UTF-8?q?=E2=80=94=20post/typing/stream=20raw=20GraphQL=20(chat@4.31/#15?= =?UTF-8?q?1=20=E2=80=94=20L4/5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port the Linear agent-session EMIT path (4th of the 5-PR wave D, #151). Builds on the L1 types, L2 thread-id, and L3 webhook-parse work already on main. The Python adapter has no `@linear/sdk`, so upstream's `createAgentActivity` / `updateAgentSession` SDK calls are ported as raw GraphQL mutations over the existing `_graphql_query` helper, schema-hardened against Linear's published GraphQL schema. Surface (branch-for-branch with adapter-linear/src/index.ts): - `post_message` session branch -> `agentActivityCreate` with content `{type:"response", body}` -> `_parse_message_from_agent_activity`. - `start_typing` session branch -> ephemeral `{type:"thought", body: status ?? "Thinking..."}`; warn-and-noop for comment threads. - `stream` dispatches to `_stream_in_agent_session` for session threads (comment path left byte-identical): a `StreamingMarkdownRenderer` with `flush_markdown(type, markdown, force)` computing a JS-`.trim()` delta; `task_update` -> error/action activity (ephemeral by status); `plan_update` -> `agentSessionUpdate` plan; final force-flush -> `{type:"response"}` -> `_parse_message_from_agent_activity`. - `_parse_message_from_agent_activity` resolves `agentActivity` + `sourceComment`, raising AdapterError (exact upstream strings) on the failure/missing-comment paths, and builds the raw message off the comment (user vs botActor author resolution). Mutations confirmed against the published schema/docs: `agentActivityCreate` (content as a `JSONObject!` scalar) and `agentSessionUpdate`, with the LOWERCASE `AgentActivityType` enum (response/thought/error/action). Live-tenant verification pending — documented in UPSTREAM_SYNC.md. `initialize` now captures the viewer's `organization.id` into `_default_organization_id` for the emitted raw message's organizationId. Faithful hazards preserved: `?? "Thinking..."` -> `is not None`, `filter(Boolean)` -> truthiness, JS-`.trim()` whitespace set, `if delta or force`, `ephemeral: status != "complete"`, bare `Error` -> RuntimeError. Tests: tests/test_linear_agent_session_emit.py — post/typing/stream paths plus both AdapterError paths; each fails on a plausible mutation (enum swap, ephemeral flip, nullish->or swap, dropped force-flush). Comment-path post/stream regression stays green. --- docs/UPSTREAM_SYNC.md | 3 +- src/chat_sdk/adapters/linear/adapter.py | 452 +++++++++++++++++++- tests/test_linear_agent_session_emit.py | 540 ++++++++++++++++++++++++ 3 files changed, 987 insertions(+), 8 deletions(-) create mode 100644 tests/test_linear_agent_session_emit.py diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index 403bc9e..bdde0a9 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -685,6 +685,7 @@ stay explicit instead of being rediscovered in code review. | Teams `graph` path-segment encoding (vercel/chat#8c71411, chat@4.31) | Path segments (`team_id` / `channel_id` / `chat_id` / `message_id`) are interpolated through `quote(segment, safe='')` | Upstream `graph/{channels,messages}.ts` use `encodeURIComponent(...)` | Benign over-encoding divergence. `quote(safe='')` percent-encodes a strictly larger set than `encodeURIComponent` (notably `!`, `'`, `(`, `)`, `*` — which `encodeURIComponent` leaves literal). Graph IDs (team/channel/chat/message GUIDs and thread tokens) never contain those characters, so the encoded path is identical in practice; where they did differ, the stricter `quote` is the safer choice (no URL injection via an unescaped sub-delimiter). Documented rather than narrowed to match `encodeURIComponent` exactly. | | Teams `graph` defensive shape coercion (vercel/chat#8c71411, chat@4.31) | `to_graph_message` / channel + message readers coerce unexpected Graph payload shapes with `isinstance` guards (`x if isinstance(x, Mapping) else {}`, `value if isinstance(value, list) else []`) before reading fields | Upstream `graph/messages.ts` reads `message.from?.user` / `message.body?.content ?? ""` etc. with optional chaining — a non-object where an object is expected throws at the property access | Benign defensive divergence. Upstream's optional chaining tolerates `null`/`undefined` but throws on a wrong-typed non-null (e.g. a string where an object is expected); our `isinstance` coercion fails closed to an empty mapping/list instead of raising. For well-formed Graph responses the behavior is identical; the divergence only manifests on malformed payloads, where returning an empty-shape result is more resilient than throwing. Mirrors the repo's general "more resilient than throw" stance (cf. the `renderPostable on unknown input` row). | | `ThinkingChunk` opt-in stream-input type (Python-only, default-off; supersedes PR #39) | A **separate, opt-in** dataclass — `ThinkingChunk(type="thinking", content=str)` — surfaces AI-SDK `reasoning`/`reasoning-delta` (and pydantic-ai `part_kind == "thinking"`) parts. **`StreamChunk` is NOT widened**: the canonical union stays `StreamChunk = MarkdownTextChunk \| TaskUpdateChunk \| PlanUpdateChunk` — byte-identical to upstream's three variants, so a consumer doing an exhaustive `match` over `StreamChunk` sees zero change on upgrade. `ThinkingChunk` is accepted only at the **stream-input/output boundaries** via the public alias `StreamInput = str \| StreamChunk \| ThinkingChunk` (the `Adapter.stream()` protocol signature, `from_full_stream`/`_from_full_stream` returns, `Thread._wrapped_stream`, and each receiving adapter's `stream()` signature). A producer can yield `ThinkingChunk` (opt-in) and the adapters that receive the stream type-check; code that only references `StreamChunk` never touches it. **OPT-IN, default-off**: emitted only when a caller passes `from_full_stream(stream, emit_thinking=True)` or sets the thread-level `emit_thinking=True` config; the internal `_from_full_stream` threads the same flag. With the default (`emit_thinking=False`) the normalized stream is **byte-for-byte identical** to upstream — reasoning parts are dropped and **no** `ThinkingChunk` is produced. Consumption is graceful: `Thread._handle_stream` never accumulates a `ThinkingChunk` into the posted-message text, and every adapter's stream handler skips it (Slack/Teams expose an optional `render_thinking` hook via `chat_sdk.shared.adapter_utils.maybe_render_thinking`; the text-accumulate adapters ignore it structurally). **Streaming-only — never persisted**: `Message` has no `thinking` field, `to_json()` is unchanged, and a round-tripped `Message` is byte-identical, so cross-SDK state (Redis/Postgres shared with the TS SDK) stays compatible. | Upstream `from-full-stream.ts` forwards only `text-delta` + `finish-step`; AI-SDK `reasoning`/`reasoning-delta` parts fall through and are discarded. `StreamChunk = MarkdownTextChunk \| TaskUpdateChunk \| PlanUpdateChunk` — no reasoning variant and no stream-input alias. Upstream leaves reasoning display to the AI-SDK web UI. | chinchill actively streams agent thinking to Slack/Teams but has to intercept the model stream out-of-band today because the chat-platform SDK has no path for it. This gives the SDK a first-class, opt-in one without changing any default behavior — and crucially **without widening the public `StreamChunk` union**, so consumers referencing it are unaffected. The whole design constraint is that default-off == upstream and `StreamChunk` == upstream: separate opt-in input type, opt-in emit, graceful/skip consume, zero state pollution. Regression coverage: `tests/test_thinking_chunk.py`, `tests/test_from_full_stream.py::TestThinkingOptIn`, `tests/test_types.py::TestThinkingChunk`, plus per-adapter no-crash tests in `tests/test_slack_api.py`, `tests/test_teams_native_streaming.py`, `tests/test_twilio_adapter.py`, `tests/test_messenger_api.py`. | +| Linear agent-activity emit: raw GraphQL (chat@4.31 / #151 — L4) | The agent-session EMIT path (`post_message` session branch, `start_typing` session branch, `stream` → `_stream_in_agent_session` with its flush/`task_update`/`plan_update` logic, and `_parse_message_from_agent_activity`) is ported as **raw GraphQL mutations** over the existing `_graphql_query` helper, **schema-hardened against Linear's published GraphQL schema**. Mutation names: `agentActivityCreate(input: AgentActivityCreateInput!)` and `agentSessionUpdate(id: String!, input: AgentSessionUpdateInput!)`. `content` is sent inside the `AgentActivityCreateInput.content` **`JSONObject!`** scalar — so `type`/`body`/`action`/`parameter`/`result` are inline JSON fields, with the **lowercase** `AgentActivityType` enum values `"response"` / `"thought"` / `"error"` / `"action"` (confirmed lowercase, NOT PascalCase). `ephemeral: Boolean` is a sibling of `content` and is **included only when set** (absent — not `false` — on response/thought/error). The `agentActivityCreate` return selection requests only schema-valid fields (`success`, `agentActivity { id agentSessionId sourceComment { id body parentId createdAt updatedAt url user{…} botActor{…} } }`). `plan` items are `{content, status:"completed"}`. `initialize` additionally captures the viewer's `organization.id` into `_default_organization_id` (mirroring upstream's `defaultOrganizationId`) for the emitted raw message's `organizationId`. | Upstream `adapter-linear/src/index.ts` calls `@linear/sdk`'s `createAgentActivity({agentSessionId, content, ephemeral?})` / `updateAgentSession(id, {plan})`; the SDK owns the GraphQL document, the `AgentActivityType` enum, and the `AgentActivityPayload`/`Comment`/`sourceComment` resolution | `@linear/sdk` is TypeScript-only (no official Linear Python SDK — cf. the `linear_client` getter row), so the SDK calls are reproduced as raw GraphQL. Mutation names, the `content: JSONObject!` shape, the lowercase enum casing, and the `plan` item shape are **schema-hardened against the published schema** (`https://linear.app/developers/agent-interaction`, `https://linear.app/developers/graphql`, and the SDK's generated GraphQL documents). **Live-tenant verification pending**: the exact mutation/field names and enum casing are confirmed against the published schema/docs but have **not** been exercised against a live Linear agent-session tenant; if a future live run surfaces a casing/field mismatch (e.g. an enum the schema renders differently at runtime), update the mutation strings here. Faithful-port hazards preserved: `status ?? "Thinking..."` → `is not None` (an empty status stays `""`); `[title, output].filter(Boolean).join("\n")` → `"\n".join(x for x in [title, output] if x)` (drops `None` and `""`); `markdown.slice(...).trim()` uses the JS-`.trim()` whitespace set (`_JS_WHITESPACE`, mirroring `adapters/telegram/rich.py`), not Python's broader `str.strip()`; `if delta or force`; `ephemeral: status != "complete"`; the missing-final-flush bare `throw new Error(...)` → `RuntimeError`. Regression coverage: `tests/test_linear_agent_session_emit.py`. | ### Platform-specific gaps | Area | Python | TS | Rationale | @@ -699,7 +700,7 @@ stay explicit instead of being rediscovered in code review. | Telegram `get_user().is_bot` | Always `False` (matches upstream — `getChat` does not expose `is_bot`) | Always `false` (same caveat documented in upstream code comment) | The Telegram Bot API's `getChat` endpoint does not surface the `is_bot` field that's available on the `User` object inside incoming `Message` updates. Callers needing bot detection must use `message.author.is_bot` from webhooks instead of `chat.get_user(...).is_bot`. | | WhatsApp `get_user` | Raises `ChatNotImplementedError` (`Chat.get_user` translates to "does not support get_user") | Not implemented upstream either (no `getUser` on the WhatsApp adapter) | WhatsApp Cloud API has no user lookup endpoint — phone numbers are the only stable identifier and there's no equivalent of `users.info` exposed to business apps. Documented explicitly so callers don't expect parity with Slack/Teams/Discord. | | Messenger `get_user` | Raising stub (`ChatNotImplementedError`); a Graph-API-backed impl is tracked as issue #132 | No `getUser` method on the Messenger adapter | **Parity — upstream Messenger has no user-lookup method**; the Python raising stub matches. (Meta's Graph API *could* back a real implementation, unlike WhatsApp — hence #132 stays open as an enhancement.) | -| Linear agent sessions | Not ported — the ~120-reference agent-sessions surface (`parseMessageFromAgentSessionEvent`, `agentActivity`, `AgentSessionEventWebhookPayload`, `LinearAgentSessionThreadId`, …) is absent from the Python `LinearAdapter` | Full agent-sessions support (`adapter-linear` 4.27.0, `bc94f0a`): parses agent-session webhook events into messages, emits agent activity, and routes the agent-session thread id | Largest single gap from the 0.4.30 audit; pre-existing (present since 0.4.29). Deferred to the 4.31 wave — tracked in **#151**. | +| Linear agent sessions | **In progress** (5-PR wave, **#151**). Landed on `main`: L1 agent-session types (`LinearAgentSessionThreadId`, `LinearAgentSessionCommentRawMessage`, `mode`/`kind`), L2 the `:s:{session}` thread-id encode/decode, L3 the webhook PARSE + routing (`_parse_message_from_agent_session_event`, `_handle_agent_session_event`). **L4 (this change)**: the agent-activity EMIT path — `post_message`/`start_typing`/`stream` session branches as raw GraphQL (see the "Linear agent-activity emit" divergence row above). Still pending: **L5** the agent-session FETCH path (`get_message` / history). | Full agent-sessions support (`adapter-linear` 4.27.0, `bc94f0a`): parses agent-session webhook events into messages, emits agent activity, and routes the agent-session thread id | Largest single gap from the 0.4.30 audit; pre-existing (present since 0.4.29). Being closed across the 4.31 wave — tracked in **#151**. | | `adapter-web` (`@chat-adapter/web`) | Neither half ported. (a) The server-side `WebAdapter` is **deferred** — not yet ported; (b) the client subpaths are out of scope (see Rationale). | Two distinct things: (a) a server-side `WebAdapter` — an `Adapter` implementation serving a browser chat UI over the AI SDK UI stream protocol (`3490a8c`, vercel/chat#444); (b) React/Vue/Svelte client subpaths (`716e934`) | The `WebAdapter` (a) **is portable** to a Python server SDK (it's a standard `Adapter`) and is deferred, not excluded — a future wave can port it. The client subpaths (b) are genuinely browser-only (front-end framework bindings) and are out of scope for a Python server SDK. (Corrects the earlier over-broad "browser-only; no Python runtime" note in CHANGELOG.) | ### Serialization differences diff --git a/src/chat_sdk/adapters/linear/adapter.py b/src/chat_sdk/adapters/linear/adapter.py index 18f1a89..79c2732 100644 --- a/src/chat_sdk/adapters/linear/adapter.py +++ b/src/chat_sdk/adapters/linear/adapter.py @@ -106,6 +106,82 @@ # Linear OAuth token endpoint LINEAR_TOKEN_URL = "https://api.linear.app/oauth/token" +# JS ``String.prototype.trim()`` whitespace set. Upstream's +# ``flushMarkdown`` computes ``markdown.slice(...).trim()``; Python's bare +# ``str.strip()`` strips a broader Unicode set (e.g. the C0 separators +# ``\x1c``-``\x1f`` and NEL ``\x85``, which JS keeps) and does NOT strip the +# BOM (which JS removes). Passing this explicit string to ``strip`` matches JS +# character-for-character. Canonical definition lives alongside the Telegram +# adapter (``adapters/telegram/rich.py``); duplicated here (not imported) to +# avoid a cross-adapter import for a single shared literal. +_JS_WHITESPACE = ( + "\t\n\v\f\r " # \t \n \v \f \r and SPACE + " " # NO-BREAK SPACE + " " # OGHAM SPACE MARK + "           " # EN QUAD..HAIR SPACE + "
" # LINE SEPARATOR + "
" # PARAGRAPH SEPARATOR + " " # NARROW NO-BREAK SPACE + " " # MEDIUM MATHEMATICAL SPACE + "　" # IDEOGRAPHIC SPACE + "" # ZERO WIDTH NO-BREAK SPACE (BOM) +) + +# Agent-activity create mutation. The Python adapter has no ``@linear/sdk``; +# ``createAgentActivity`` is ported as this raw GraphQL mutation. Schema-hardened +# against Linear's published GraphQL schema (https://linear.app/developers/ +# agent-interaction , https://linear.app/developers/graphql): the mutation is +# ``agentActivityCreate(input: AgentActivityCreateInput!)`` where ``content`` is +# a ``JSONObject!`` scalar (so ``type``/``body``/``action``/... are passed inline +# as a JSON object with LOWERCASE ``type`` enum values — "response"/"thought"/ +# "error"/"action"), ``agentSessionId: String!`` and ``ephemeral: Boolean``. The +# return selection requests only schema-valid fields that +# ``_parse_message_from_agent_activity`` reads off the resolved ``sourceComment``. +_AGENT_ACTIVITY_CREATE_MUTATION = """ +mutation AgentActivityCreate($input: AgentActivityCreateInput!) { + agentActivityCreate(input: $input) { + success + agentActivity { + id + agentSessionId + sourceComment { + id + body + parentId + createdAt + updatedAt + url + user { + id + displayName + name + email + avatarUrl + } + botActor { + id + name + userDisplayName + avatarUrl + } + } + } + } +} +""" + +# Agent-session plan-update mutation. Ports ``updateAgentSession``. Schema- +# hardened: ``agentSessionUpdate(id: String!, input: AgentSessionUpdateInput!)`` +# where ``input.plan`` is an array of ``{content, status}`` items (status +# ``"completed"`` from upstream). +_AGENT_SESSION_UPDATE_MUTATION = """ +mutation AgentSessionUpdate($id: String!, $input: AgentSessionUpdateInput!) { + agentSessionUpdate(id: $id, input: $input) { + success + } +} +""" + # Emoji mapping for Linear reactions (unicode) EMOJI_MAPPING: dict[str, str] = { "thumbs_up": "\U0001f44d", @@ -200,6 +276,13 @@ def __init__(self, config: LinearAdapterConfig | None = None) -> None: self._mode: LinearAdapterMode = config_mode if config_mode is not None else "comments" self._chat: ChatInstance | None = None self._bot_user_id: str | None = None + # Default organization ID, resolved at ``initialize`` from the viewer's + # ``organization.id``. Faithful port of upstream + # ``defaultOrganizationId`` (index.ts:206/347): single-tenant fallback + # used by the agent-activity emit path when there is no per-request + # installation context (no webhook payload to read ``organizationId`` + # off). ``None`` until the bot identity resolves. + self._default_organization_id: str | None = None self._format_converter = LinearFormatConverter() # Shared aiohttp session for connection pooling @@ -309,11 +392,16 @@ async def initialize(self, chat: ChatInstance) -> None: if self._client_credentials: await self._refresh_client_credentials_token() - # Fetch the bot's user ID for self-message detection + # Fetch the bot's user ID for self-message detection. Also capture the + # viewer's ``organization.id`` (upstream resolves the same field from + # client identity at init, index.ts:347/727) so the agent-activity emit + # path has a default organization ID in single-tenant mode. try: - viewer = await self._graphql_query("query { viewer { id displayName } }") + viewer = await self._graphql_query("query { viewer { id displayName organization { id } } }") viewer_data = viewer.get("data", {}).get("viewer", {}) self._bot_user_id = viewer_data.get("id") + organization = viewer_data.get("organization") or {} + self._default_organization_id = organization.get("id") self._logger.info( "Linear auth completed", { @@ -1042,6 +1130,18 @@ async def post_message( # Convert emoji placeholders to unicode body = convert_emoji_placeholders(body, "linear") + # Agent-session branch (index.ts:1323). When the decoded thread carries a + # session, post the reply as a "response" agent activity instead of a + # comment, then resolve it back into a raw message from the activity's + # ``sourceComment``. The comment path below is byte-identical to before. + if decoded.agent_session_id: + session = assert_agent_session_thread(decoded) + activity_result = await self._create_agent_activity( + session.agent_session_id, + {"type": "response", "body": body}, + ) + return await self._parse_message_from_agent_activity(session, activity_result) + # Create comment via GraphQL API result = await self._graphql_query( """ @@ -1088,6 +1188,185 @@ async def post_message( ), ) + async def _create_agent_activity( + self, + agent_session_id: str, + content: dict[str, Any], + *, + ephemeral: bool | None = None, + ) -> dict[str, Any]: + """Create a Linear agent activity via raw GraphQL. + + Faithful port of ``@linear/sdk``'s ``createAgentActivity`` + (index.ts:1336/1522/1580/1608/1616). The TS SDK call shape is + ``createAgentActivity({agentSessionId, content, ephemeral?})`` — here the + same payload is posted as ``agentActivityCreate(input: ...)`` against the + published schema, where ``content`` is a ``JSONObject!`` scalar carrying + the lowercase ``type`` plus its body/action fields inline. ``ephemeral`` + is only included when explicitly set so the serialized input matches + upstream's conditional spread (it is absent — not ``False`` — on the + ``response``/``thought``/``error`` calls that don't pass it). + + Returns the raw ``agentActivityCreate`` payload node (``{success, + agentActivity}``) — the same object upstream's ``AgentActivityPayload`` + models — for :meth:`_parse_message_from_agent_activity` to resolve. + """ + activity_input: dict[str, Any] = { + "agentSessionId": agent_session_id, + "content": content, + } + if ephemeral is not None: + activity_input["ephemeral"] = ephemeral + + result = await self._graphql_query( + _AGENT_ACTIVITY_CREATE_MUTATION, + {"input": activity_input}, + ) + return cast("dict[str, Any]", result.get("data", {}).get("agentActivityCreate", {})) + + async def _parse_message_from_agent_activity( + self, + thread: LinearAgentSessionThreadId, + result: dict[str, Any], + ) -> RawMessage: + """Build a raw message from an ``agentActivityCreate`` mutation result. + + Faithful port of upstream ``parseMessageFromAgentActivity`` + (index.ts:1082). Resolves the created ``agentActivity`` and its + ``sourceComment`` from the mutation payload, raising ``AdapterError`` with + the exact upstream strings when the activity failed to create or its + source comment could not be resolved. The resulting message is built off + the source comment, mirroring upstream's delegation to + ``parseMessageFromComment(sourceComment, issueId, agentSessionId)``. + """ + activity = result.get("agentActivity") + if not (result.get("success") and activity): + raise AdapterError( + f"Failed to create Linear agent activity for session {thread.agent_session_id}", + "linear", + ) + + source_comment = activity.get("sourceComment") + if not source_comment: + raise AdapterError( + f"Failed to resolve source comment for Linear agent activity {activity.get('id')}", + "linear", + ) + + activity_session_id = activity.get("agentSessionId") + if not activity_session_id: + raise AdapterError( + f"Missing agentSessionId for Linear agent activity {activity.get('id')}", + "linear", + ) + + return self._raw_message_from_source_comment( + source_comment, + thread.issue_id, + cast("str", activity_session_id), + ) + + def _raw_message_from_source_comment( + self, + comment: dict[str, Any], + issue_id: str, + agent_session_id: str, + ) -> RawMessage: + """Build the agent-session raw message from a resolved source comment. + + Faithful port of upstream ``parseMessageFromComment`` (index.ts:884) for + the agent-activity emit path. Author resolution mirrors upstream: when + the comment carries a ``user`` it is a user author; otherwise the comment + was created by the app, so the ``botActor`` supplies a bot author (and, + as upstream notes, an app comment without a botActor cannot determine an + author). ``avatarUrl ?? undefined`` / ``botActor.id ?? this.botUserId`` + and the display/full-name fallbacks are nullish (``is not None`` / ``or`` + per upstream's ``??`` vs ``||``) faithful. + """ + comment_user = comment.get("user") + if comment_user: + avatar_url = comment_user.get("avatarUrl") + user: LinearActorData = { + "type": "user", + "id": cast("str", comment_user.get("id", "")), + "displayName": cast("str", comment_user.get("displayName", "")), + "fullName": cast("str", comment_user.get("name", "")), + "email": cast("str", comment_user.get("email")), + **({"avatarUrl": cast("str", avatar_url)} if avatar_url is not None else {}), + } + else: + bot_actor = comment.get("botActor") + if not bot_actor: + raise AdapterError( + f"Comment {comment.get('id')} has no userId and no botActor, cannot determine author.", + "linear", + ) + # ``botActor.id ?? this.botUserId`` — nullish. Coerce a None + # bot-user-id (not yet resolved) to "" via ``is not None``. + actor_id = bot_actor.get("id") + resolved_id = actor_id if actor_id is not None else (self._bot_user_id or "") + # ``userDisplayName ?? name ?? "unknown"`` / ``name ?? userDisplayName + # ?? "unknown"`` — chained nullish. + display_name = bot_actor.get("userDisplayName") + actor_name = bot_actor.get("name") + bot_avatar = bot_actor.get("avatarUrl") + user = { + "type": "bot", + "id": cast("str", resolved_id), + "displayName": cast( + "str", + display_name if display_name is not None else (actor_name if actor_name is not None else "unknown"), + ), + "fullName": cast( + "str", + actor_name if actor_name is not None else (display_name if display_name is not None else "unknown"), + ), + **({"avatarUrl": cast("str", bot_avatar)} if bot_avatar is not None else {}), + } + + # ``parentId ?? undefined`` / ``url ?? undefined`` — nullish. createdAt / + # updatedAt are ISO strings off the resolved comment node. + parent_id = comment.get("parentId") + comment_url = comment.get("url") + comment_data: LinearCommentData = { + "id": cast("str", comment.get("id", "")), + "body": cast("str", comment.get("body", "")), + "issueId": issue_id, + "user": user, + "createdAt": cast("str", comment.get("createdAt", "")), + "updatedAt": cast("str", comment.get("updatedAt", "")), + } + if parent_id is not None: + comment_data["parentId"] = cast("str", parent_id) + if comment_url is not None: + comment_data["url"] = cast("str", comment_url) + + raw: LinearAgentSessionCommentRawMessage = { + "kind": "agent_session_comment", + "organizationId": self._default_organization_id or "", + "comment": comment_data, + "agentSessionId": agent_session_id, + } + + # The Python ``RawMessage`` is the minimal ``{id, thread_id, raw}`` wrapper + # (no author field — unlike upstream's richer ``Message extends RawMessage`` + # return). The fully-parsed author/metadata live under ``raw`` and are + # re-derived on read via :meth:`_parse_agent_session_message`. Encode the + # routed thread id (carrying the ``:s:{session}`` segment) so callers can + # round-trip it, matching the comment branch's ``thread_id`` semantics. + thread_id = self.encode_thread_id( + LinearThreadId( + issue_id=issue_id, + comment_id=cast("str | None", comment_data.get("parentId")), + agent_session_id=agent_session_id, + ) + ) + return RawMessage( + id=cast("str", comment_data["id"]), + thread_id=thread_id, + raw=cast("Any", raw), + ) + async def edit_message( self, thread_id: str, @@ -1188,8 +1467,31 @@ async def remove_reaction( self._logger.warn("removeReaction is not fully supported on Linear - reaction ID lookup would be required") async def start_typing(self, thread_id: str, status: str | None = None) -> None: - """Start typing indicator. Not supported by Linear.""" - pass + """Start typing indicator. + + Faithful port of upstream ``startTyping`` (index.ts:1517). For + agent-session threads this emits an ephemeral "thought" activity (the + Linear-native typing equivalent); for standard comment threads it + remains a warn-and-noop. ``status ?? "Thinking..."`` is nullish — an + explicit empty-string status stays empty (NOT replaced by the default), + so the ``is not None`` guard (not truthiness) is load-bearing. + """ + await self._ensure_valid_token() + decoded = self.decode_thread_id(thread_id) + + if decoded.agent_session_id: + session = assert_agent_session_thread(decoded) + await self._create_agent_activity( + session.agent_session_id, + { + "type": "thought", + "body": status if status is not None else "Thinking...", + }, + ephemeral=True, + ) + return + + self._logger.warn("startTyping is only supported in agent session threads. Ignoring for comment thread.") async def fetch_messages( self, @@ -1523,11 +1825,20 @@ async def stream( text_stream: Any, options: StreamOptions | None = None, ) -> RawMessage: - """Stream responses by accumulating chunks and posting/editing a single comment. + """Stream responses to a thread. - Linear does not support native streaming, so this accumulates the - full text and posts (or edits) a single comment at the end. + Faithful port of upstream ``stream`` (index.ts:1542): dispatch to the + agent-session streamer when the decoded thread carries a session, else + fall through to the existing comment-path behavior (byte-identical to + before — Linear comments do not support native streaming, so the comment + path accumulates the full text and posts a single comment at the end). """ + decoded = self.decode_thread_id(thread_id) + if decoded.agent_session_id: + await self._ensure_valid_token() + session = assert_agent_session_thread(decoded) + return await self._stream_in_agent_session(session, text_stream, options) + accumulated = "" message_id: str | None = None @@ -1553,6 +1864,133 @@ async def stream( raw={"text": accumulated}, ) + async def _stream_in_agent_session( + self, + decoded: LinearAgentSessionThreadId, + text_stream: Any, + _options: StreamOptions | None = None, + ) -> RawMessage: + """Stream text/chunks into a Linear agent session. + + Faithful port of upstream ``streamInAgentSession`` (index.ts:1560). + Unlike the comment path (a single post/edit), an agent session is + append-only: committable markdown is flushed as successive "response" + activities (with a delta computed against the last appended text), + ``task_update`` chunks become "action"/"error" activities, and + ``plan_update`` chunks replace the session plan. The final + ``renderer.finish()`` is force-flushed as the closing response, then + resolved back into a raw message via + :meth:`_parse_message_from_agent_activity`. + """ + from chat_sdk.shared.streaming_markdown import StreamingMarkdownRenderer + + renderer = StreamingMarkdownRenderer() + agent_session_id = decoded.agent_session_id + + # Mutable closure state. ``last_appended`` tracks the full markdown text + # already emitted so each flush sends only the new tail (the delta). + last_appended = "" + + async def flush_markdown( + activity_type: str, + markdown: str | None = None, + force: bool = False, + ) -> dict[str, Any] | None: + """Flush the current markdown buffer into a new "response"/"thought" activity. + + ``delta = markdown[len(last_appended):].trim()`` — the JS ``.trim()`` + whitespace set (``_JS_WHITESPACE``), NOT Python's broader + ``str.strip()``. ``if delta or force`` mirrors upstream: an + empty-string delta is falsy in both languages, so a no-delta flush + is a no-op unless forced (the final flush). Returns the created + activity payload, or ``None`` when nothing was sent. + """ + nonlocal last_appended + committable = renderer.get_committable_text() if markdown is None else markdown + delta = committable[len(last_appended) :].strip(_JS_WHITESPACE) + if delta or force: + last_appended = committable + return await self._create_agent_activity( + agent_session_id, + {"type": activity_type, "body": delta}, + ) + return None + + def _read(chunk: Any, name: str) -> Any: + """Read a chunk field from either a dict or a dataclass chunk.""" + if isinstance(chunk, dict): + return chunk.get(name) + return getattr(chunk, name, None) + + async for chunk in text_stream: + if isinstance(chunk, str): + renderer.push(chunk) + continue + + chunk_type = _read(chunk, "type") + + if chunk_type == "markdown_text": + renderer.push(_read(chunk, "text") or "") + continue + + if chunk_type == "task_update": + # Flush any buffered markdown as a "thought" before sending the + # action so the action card is distinct from the response body. + await flush_markdown("thought") + + title = _read(chunk, "title") + output = _read(chunk, "output") + status = _read(chunk, "status") + + if status == "error": + # ``[title, output].filter(Boolean).join("\n")`` — drops None + # AND empty string (truthiness), faithful. + body = "\n".join(x for x in [title, output] if x) + await self._create_agent_activity( + agent_session_id, + {"type": "error", "body": body}, + ) + else: + # ``ephemeral: status !== "complete"`` — in-progress/pending + # actions are ephemeral; only a completed action persists. + await self._create_agent_activity( + agent_session_id, + { + "type": "action", + "action": title, + "parameter": "", + "result": output, + }, + ephemeral=status != "complete", + ) + continue + + if chunk_type == "plan_update": + # Replace the session plan in its entirety (agents cannot patch a + # single item). https://linear.app/developers/agent-interaction + await self._graphql_query( + _AGENT_SESSION_UPDATE_MUTATION, + { + "id": agent_session_id, + "input": { + "plan": [ + { + "content": _read(chunk, "title"), + "status": "completed", + } + ] + }, + }, + ) + + final_activity = await flush_markdown("response", renderer.finish(), True) + if not final_activity: + # Upstream throws a bare ``Error`` here (NOT an ``AdapterError``); + # ported as ``RuntimeError`` per the repo's bare-Error convention. + raise RuntimeError("Failed to flush final markdown delta for agent session stream") + + return await self._parse_message_from_agent_activity(decoded, final_activity) + async def _get_http_session(self) -> Any: """Return the shared aiohttp session, creating it lazily if needed.""" import aiohttp diff --git a/tests/test_linear_agent_session_emit.py b/tests/test_linear_agent_session_emit.py new file mode 100644 index 0000000..9653cfc --- /dev/null +++ b/tests/test_linear_agent_session_emit.py @@ -0,0 +1,540 @@ +"""Tests for the Linear agent-session EMIT path (post / typing / stream). + +Ported from packages/adapter-linear/src/index.test.ts (chat@4.31 / #151, +the L4 emit surface). The Python adapter has no ``@linear/sdk`` — upstream's +``createAgentActivity`` / ``updateAgentSession`` calls are ported as raw +GraphQL mutations against the published Linear schema: + +- ``agentActivityCreate(input: AgentActivityCreateInput!)`` where ``content`` + is a ``JSONObject!`` scalar carrying the LOWERCASE ``type`` enum + ("response" / "thought" / "error" / "action") plus body/action fields. +- ``agentSessionUpdate(id, input: AgentSessionUpdateInput!)`` for plan updates. + +Each test pins the exact mutation payload so a regression — an enum swap +(response→thought), an ephemeral flip, a ``status ?? default`` → ``status or +default`` swap, or a dropped final force-flush — fails the assertion. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from chat_sdk.adapters.linear.adapter import LinearAdapter +from chat_sdk.adapters.linear.types import LinearAdapterAPIKeyConfig +from chat_sdk.shared.errors import AdapterError +from chat_sdk.types import MarkdownTextChunk, PlanUpdateChunk, TaskUpdateChunk + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +WEBHOOK_SECRET = "test-webhook-secret" + +_SESSION_THREAD = "linear:issue-123:c:comment-root:s:session-789" + + +def _make_logger() -> MagicMock: + return MagicMock( + debug=MagicMock(), + info=MagicMock(), + warn=MagicMock(), + error=MagicMock(), + ) + + +def _make_adapter(logger: MagicMock | None = None) -> LinearAdapter: + """Agent-sessions-mode adapter with a known bot-user-id and default org.""" + if logger is None: + logger = _make_logger() + config = LinearAdapterAPIKeyConfig( + api_key="test-api-key", + webhook_secret=WEBHOOK_SECRET, + user_name="test-bot", + mode="agent-sessions", # type: ignore[arg-type] + logger=logger, + ) + adapter = LinearAdapter(config) + # ``initialize`` (a viewer query) sets these in production; pre-set the + # single-tenant defaults so the emit path resolves an author + org id. + adapter._bot_user_id = "bot-user-id" + adapter._default_organization_id = "org-123" + return adapter + + +def _source_comment( + *, + activity_id: str = "activity-123", + body: str = "Agent response", + with_bot_actor: bool = True, + with_user: bool = False, +) -> dict[str, Any]: + """A resolved ``sourceComment`` node, shaped like the raw GraphQL return. + + Mirrors upstream's ``createMockAgentActivityPayload`` ``sourceComment``: + a comment created by the app (no ``user``, a ``botActor`` fallback) maps to + a bot author with ``is_me`` true (the bot user id matches). + """ + comment: dict[str, Any] = { + "id": f"comment-{activity_id}", + "body": body, + "parentId": "comment-root", + "createdAt": "2025-06-01T12:00:01.000Z", + "updatedAt": "2025-06-01T12:00:01.000Z", + "url": f"https://linear.app/comment/{activity_id}", + } + if with_user: + comment["user"] = { + "id": "human-user-1", + "displayName": "Ada", + "name": "Ada Lovelace", + "email": "ada@example.com", + "avatarUrl": "https://linear.app/avatar/ada.png", + } + if with_bot_actor: + comment["botActor"] = { + "id": "bot-user-id", + "name": "Test Bot", + "userDisplayName": "Test Bot", + } + return comment + + +def _activity_payload( + *, + activity_id: str = "activity-123", + body: str = "Agent response", + success: bool = True, + agent_session_id: str | None = "session-789", + source_comment: Any = "default", +) -> dict[str, Any]: + """The ``agentActivityCreate`` payload node (``{success, agentActivity}``).""" + activity: dict[str, Any] | None = { + "id": activity_id, + "agentSessionId": agent_session_id, + "sourceComment": _source_comment(activity_id=activity_id, body=body) + if source_comment == "default" + else source_comment, + } + return {"success": success, "agentActivity": activity} + + +def _graphql_return(activity: dict[str, Any]) -> dict[str, Any]: + """Wrap an activity payload as a ``_graphql_query`` return value.""" + return {"data": {"agentActivityCreate": activity}} + + +async def _astream(*chunks: Any) -> Any: + """Build an async iterator from the given chunks.""" + for chunk in chunks: + yield chunk + + +# =========================================================================== +# post_message — agent-session branch +# =========================================================================== + + +class TestPostMessageAgentSession: + @pytest.mark.asyncio + async def test_uses_agent_activity_create_with_response(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload())) + + result = await adapter.post_message(_SESSION_THREAD, "Agent response") + + call_args = adapter._graphql_query.call_args + query = call_args[0][0] + variables = call_args[0][1] + assert "agentActivityCreate" in query + assert variables["input"]["agentSessionId"] == "session-789" + # Enum-swap proof: the content type MUST be the lowercase "response". + assert variables["input"]["content"] == {"type": "response", "body": "Agent response"} + # ephemeral is absent (not False) for a response activity. + assert "ephemeral" not in variables["input"] + + # The resolved message is built off the source comment. + assert result.id == "comment-activity-123" + assert result.raw["kind"] == "agent_session_comment" + assert result.raw["organizationId"] == "org-123" + assert result.raw["agentSessionId"] == "session-789" + assert result.raw["comment"]["body"] == "Agent response" + # botActor → bot author, matching bot-user-id → is_me semantics. + assert result.raw["comment"]["user"]["type"] == "bot" + assert result.raw["comment"]["user"]["id"] == "bot-user-id" + + @pytest.mark.asyncio + async def test_calls_ensure_valid_token(self) -> None: + adapter = _make_adapter() + adapter._ensure_valid_token = AsyncMock() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload())) + + await adapter.post_message(_SESSION_THREAD, "Agent response") + + assert adapter._ensure_valid_token.call_count == 1 + + @pytest.mark.asyncio + async def test_comment_branch_unchanged_for_non_session_thread(self) -> None: + """Regression: a non-session thread still uses commentCreate, NOT agentActivityCreate.""" + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value={ + "data": { + "commentCreate": { + "success": True, + "comment": { + "id": "new-comment-1", + "body": "Bot reply", + "url": "https://linear.app/test/comment/new-comment-1", + "createdAt": "2025-06-01T12:00:00.000Z", + "updatedAt": "2025-06-01T12:00:00.000Z", + }, + } + } + } + ) + + result = await adapter.post_message("linear:issue-123:c:parent-comment", "Hello") + + query = adapter._graphql_query.call_args[0][0] + assert "commentCreate" in query + assert "agentActivityCreate" not in query + assert result.raw["kind"] == "comment" + + +# =========================================================================== +# _parse_message_from_agent_activity +# =========================================================================== + + +class TestParseMessageFromAgentActivity: + @pytest.mark.asyncio + async def test_resolves_user_author_when_user_present(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value=_graphql_return( + _activity_payload(source_comment=_source_comment(with_user=True, with_bot_actor=False)) + ) + ) + + result = await adapter.post_message(_SESSION_THREAD, "Agent response") + + user = result.raw["comment"]["user"] + assert user["type"] == "user" + assert user["id"] == "human-user-1" + assert user["displayName"] == "Ada" + assert user["fullName"] == "Ada Lovelace" + + @pytest.mark.asyncio + async def test_raises_when_activity_not_successful(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload(success=False))) + + with pytest.raises(AdapterError, match="Failed to create Linear agent activity for session session-789"): + await adapter.post_message(_SESSION_THREAD, "Agent response") + + @pytest.mark.asyncio + async def test_raises_when_activity_missing(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return({"success": True, "agentActivity": None})) + + with pytest.raises(AdapterError, match="Failed to create Linear agent activity"): + await adapter.post_message(_SESSION_THREAD, "Agent response") + + @pytest.mark.asyncio + async def test_raises_when_source_comment_missing(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload(source_comment=None))) + + with pytest.raises(AdapterError, match="Failed to resolve source comment for Linear agent activity"): + await adapter.post_message(_SESSION_THREAD, "Agent response") + + +# =========================================================================== +# start_typing — agent-session branch +# =========================================================================== + + +class TestStartTyping: + @pytest.mark.asyncio + async def test_emits_ephemeral_thought_with_default_body(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload())) + + await adapter.start_typing(_SESSION_THREAD) + + variables = adapter._graphql_query.call_args[0][1] + assert variables["input"]["agentSessionId"] == "session-789" + # Enum-swap proof: typing is a "thought", not "response". + assert variables["input"]["content"] == {"type": "thought", "body": "Thinking..."} + # ephemeral-flip proof: typing activities MUST be ephemeral. + assert variables["input"]["ephemeral"] is True + + @pytest.mark.asyncio + async def test_uses_explicit_status_body(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload())) + + await adapter.start_typing(_SESSION_THREAD, "Looking things up...") + + variables = adapter._graphql_query.call_args[0][1] + assert variables["input"]["content"]["body"] == "Looking things up..." + + @pytest.mark.asyncio + async def test_empty_string_status_stays_empty(self) -> None: + """``status ?? "Thinking..."`` is nullish: an empty string is NOT replaced. + + ``status or "Thinking..."`` (truthiness) would wrongly swap "" for the + default — this asserts the ``is not None`` guard. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload())) + + await adapter.start_typing(_SESSION_THREAD, "") + + variables = adapter._graphql_query.call_args[0][1] + assert variables["input"]["content"]["body"] == "" + + @pytest.mark.asyncio + async def test_warn_and_noop_for_comment_thread(self) -> None: + logger = _make_logger() + adapter = _make_adapter(logger=logger) + adapter._graphql_query = AsyncMock() + + await adapter.start_typing("linear:issue-123:c:comment-root") + + adapter._graphql_query.assert_not_called() + logger.warn.assert_called_once() + assert "only supported in agent session threads" in logger.warn.call_args[0][0] + + +# =========================================================================== +# _stream_in_agent_session +# =========================================================================== + + +class TestStreamInAgentSession: + @pytest.mark.asyncio + async def test_task_update_creates_action_activity_with_ephemeral_by_status(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="opening", body="Hello")), + _graphql_return(_activity_payload(activity_id="task")), + _graphql_return(_activity_payload(activity_id="final", body="world")), + ] + ) + + result = await adapter.stream( + _SESSION_THREAD, + _astream( + "Hello\n", + TaskUpdateChunk(id="task-1", title="Search docs", status="in_progress", output="Started"), + MarkdownTextChunk(text="world"), + ), + ) + + inputs = [call.args[1]["input"] for call in adapter._graphql_query.call_args_list] + # 1) buffered markdown flushed as a "thought" delta (the "Hello" line). + assert inputs[0]["content"] == {"type": "thought", "body": "Hello"} + # 2) task_update → action; non-complete status → ephemeral True. + assert inputs[1]["content"] == { + "type": "action", + "action": "Search docs", + "parameter": "", + "result": "Started", + } + assert inputs[1]["ephemeral"] is True + # 3) final force-flush of "world" as a "response". + assert inputs[2]["content"] == {"type": "response", "body": "world"} + assert result.raw["kind"] == "agent_session_comment" + assert result.id == "comment-final" + + @pytest.mark.asyncio + async def test_complete_task_update_is_not_ephemeral(self) -> None: + """``ephemeral: status != "complete"`` — a completed action persists.""" + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="task")), + _graphql_return(_activity_payload(activity_id="final", body="done")), + ] + ) + + await adapter.stream( + _SESSION_THREAD, + _astream( + TaskUpdateChunk(id="t", title="Build", status="complete", output="ok"), + MarkdownTextChunk(text="done"), + ), + ) + + action_input = adapter._graphql_query.call_args_list[0].args[1]["input"] + assert action_input["content"]["type"] == "action" + assert action_input["ephemeral"] is False + + @pytest.mark.asyncio + async def test_error_task_update_creates_error_activity(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="err")), + _graphql_return(_activity_payload(activity_id="final", body="world")), + ] + ) + + await adapter.stream( + _SESSION_THREAD, + _astream( + TaskUpdateChunk(id="t", title="Search docs", status="error", output="Boom"), + MarkdownTextChunk(text="world"), + ), + ) + + error_input = adapter._graphql_query.call_args_list[0].args[1]["input"] + # Enum-swap proof: error path → "error", not "action". + assert error_input["content"]["type"] == "error" + # [title, output].filter(Boolean).join("\n"). + assert error_input["content"]["body"] == "Search docs\nBoom" + assert "ephemeral" not in error_input + + @pytest.mark.asyncio + async def test_error_task_update_drops_empty_fields_via_filter_boolean(self) -> None: + """filter(Boolean) drops None AND empty-string title/output.""" + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="err")), + _graphql_return(_activity_payload(activity_id="final", body="x")), + ] + ) + + await adapter.stream( + _SESSION_THREAD, + _astream( + TaskUpdateChunk(id="t", title="", status="error", output="Boom"), + MarkdownTextChunk(text="x"), + ), + ) + + error_input = adapter._graphql_query.call_args_list[0].args[1]["input"] + assert error_input["content"]["body"] == "Boom" + + @pytest.mark.asyncio + async def test_plan_update_calls_agent_session_update(self) -> None: + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + {"data": {"agentSessionUpdate": {"success": True}}}, + _graphql_return(_activity_payload(activity_id="final", body="Hello world")), + ] + ) + + result = await adapter.stream( + _SESSION_THREAD, + _astream( + PlanUpdateChunk(title="Search docs"), + "Hello world", + ), + ) + + plan_call = adapter._graphql_query.call_args_list[0] + assert "agentSessionUpdate" in plan_call.args[0] + assert plan_call.args[1] == { + "id": "session-789", + "input": {"plan": [{"content": "Search docs", "status": "completed"}]}, + } + # The final response still posts after the plan update. + final_call = adapter._graphql_query.call_args_list[1] + assert final_call.args[1]["input"]["content"] == {"type": "response", "body": "Hello world"} + assert result.id == "comment-final" + + @pytest.mark.asyncio + async def test_final_force_flush_posts_response_even_with_no_delta(self) -> None: + """The final flush is forced, so it emits even when the delta is empty. + + Drop-the-force-flush proof: a single string chunk is flushed once at the + end as a forced "response"; there is exactly one mutation call. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value=_graphql_return(_activity_payload(activity_id="final", body="Hello world")) + ) + + result = await adapter.stream(_SESSION_THREAD, _astream("Hello world")) + + assert adapter._graphql_query.call_count == 1 + only_input = adapter._graphql_query.call_args.args[1]["input"] + assert only_input["content"] == {"type": "response", "body": "Hello world"} + assert result.raw["kind"] == "agent_session_comment" + + @pytest.mark.asyncio + async def test_empty_delta_flush_is_noop_when_not_forced(self) -> None: + """A whitespace-only buffer produces no intermediate flush. + + Two task updates fire back-to-back with no new committable markdown + between them; the pre-action ``flush_markdown("thought")`` must NOT emit + a thought activity (empty delta, not forced) — only the two actions and + the final forced response are sent. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="a1")), + _graphql_return(_activity_payload(activity_id="a2")), + _graphql_return(_activity_payload(activity_id="final", body="")), + ] + ) + + await adapter.stream( + _SESSION_THREAD, + _astream( + TaskUpdateChunk(id="t1", title="One", status="in_progress", output="a"), + TaskUpdateChunk(id="t2", title="Two", status="in_progress", output="b"), + ), + ) + + types = [c.args[1]["input"]["content"]["type"] for c in adapter._graphql_query.call_args_list] + # No "thought" flush slipped in: action, action, response. + assert types == ["action", "action", "response"] + + @pytest.mark.asyncio + async def test_raises_when_final_activity_missing(self) -> None: + """``if not finalActivity`` → raise (the final mutation resolved nothing).""" + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value={"data": {"agentActivityCreate": {"success": False, "agentActivity": None}}} + ) + + with pytest.raises(AdapterError): + await adapter.stream(_SESSION_THREAD, _astream("Hello world")) + + @pytest.mark.asyncio + async def test_dispatches_to_comment_stream_for_non_session_thread(self) -> None: + """Regression: a non-session thread uses the comment-path stream, NOT agentActivityCreate.""" + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value={ + "data": { + "commentCreate": { + "success": True, + "comment": { + "id": "c1", + "body": "Hello world", + "url": "https://linear.app/test/comment/c1", + "createdAt": "2025-06-01T12:00:00.000Z", + "updatedAt": "2025-06-01T12:00:00.000Z", + }, + } + } + } + ) + + result = await adapter.stream("linear:issue-123:c:comment-root", _astream("Hello world")) + + query = adapter._graphql_query.call_args[0][0] + assert "commentCreate" in query + assert "agentActivityCreate" not in query + assert result.id == "c1" From c06508a764050542a3d201392a960d9f5f16baf9 Mon Sep 17 00:00:00 2001 From: patrick-chinchill Date: Sat, 20 Jun 2026 00:27:24 -0700 Subject: [PATCH 2/2] fix(linear): correct agent-activity emit GraphQL selection + thread_id + result-omit (L4 review) Apply the L4 emit review fix-list: 1. GraphQL selection: agentActivityCreate return selection requested a non-existent scalar agentActivity.agentSessionId; Linear's schema exposes only the relation agentSession: AgentSession!, so strict selection validation server-rejected the whole mutation. Select agentSession { id }. 2. Consumer read (lockstep): _parse_message_from_agent_activity now reads the session id None-safely off (activity.get("agentSession") or {}).get("id"), keeping the raise-on-missing guard + message. 3. thread_id: _raw_message_from_source_comment encoded the routed thread id with the source comment's parentId; upstream parseMessage and the adapter's own read-path encode the comment's OWN id. Use comment_data["id"]. 4. task_update action wire-shape: omit the result key when output is None (upstream passes result: chunk.output string|undefined; JSON.stringify omits undefined) instead of serializing "result": null. 5. Doc: correct the UPSTREAM_SYNC.md emit row to the fixed selection and note the organizationId "" fallback divergence; keep the live-tenant hedge. Tests (test_linear_agent_session_emit.py): fix the response fixture to emit the agentSession relation; add thread_id own-id assertions (post + stream), split the mislabeled final-flush test into the RuntimeError force-flush branch and the AdapterError parse-failure branch, add the missing-agentSession AdapterError case, bot-author nullish-precedence coverage, the _JS_WHITESPACE NEL-retain / BOM-strip deltas, and the result-omit/include pair. Each new/changed assertion verified to fail against the corresponding pre-fix code. --- docs/UPSTREAM_SYNC.md | 2 +- src/chat_sdk/adapters/linear/adapter.py | 39 +++- tests/test_linear_agent_session_emit.py | 239 +++++++++++++++++++++++- 3 files changed, 265 insertions(+), 15 deletions(-) diff --git a/docs/UPSTREAM_SYNC.md b/docs/UPSTREAM_SYNC.md index bdde0a9..c6d9c93 100644 --- a/docs/UPSTREAM_SYNC.md +++ b/docs/UPSTREAM_SYNC.md @@ -685,7 +685,7 @@ stay explicit instead of being rediscovered in code review. | Teams `graph` path-segment encoding (vercel/chat#8c71411, chat@4.31) | Path segments (`team_id` / `channel_id` / `chat_id` / `message_id`) are interpolated through `quote(segment, safe='')` | Upstream `graph/{channels,messages}.ts` use `encodeURIComponent(...)` | Benign over-encoding divergence. `quote(safe='')` percent-encodes a strictly larger set than `encodeURIComponent` (notably `!`, `'`, `(`, `)`, `*` — which `encodeURIComponent` leaves literal). Graph IDs (team/channel/chat/message GUIDs and thread tokens) never contain those characters, so the encoded path is identical in practice; where they did differ, the stricter `quote` is the safer choice (no URL injection via an unescaped sub-delimiter). Documented rather than narrowed to match `encodeURIComponent` exactly. | | Teams `graph` defensive shape coercion (vercel/chat#8c71411, chat@4.31) | `to_graph_message` / channel + message readers coerce unexpected Graph payload shapes with `isinstance` guards (`x if isinstance(x, Mapping) else {}`, `value if isinstance(value, list) else []`) before reading fields | Upstream `graph/messages.ts` reads `message.from?.user` / `message.body?.content ?? ""` etc. with optional chaining — a non-object where an object is expected throws at the property access | Benign defensive divergence. Upstream's optional chaining tolerates `null`/`undefined` but throws on a wrong-typed non-null (e.g. a string where an object is expected); our `isinstance` coercion fails closed to an empty mapping/list instead of raising. For well-formed Graph responses the behavior is identical; the divergence only manifests on malformed payloads, where returning an empty-shape result is more resilient than throwing. Mirrors the repo's general "more resilient than throw" stance (cf. the `renderPostable on unknown input` row). | | `ThinkingChunk` opt-in stream-input type (Python-only, default-off; supersedes PR #39) | A **separate, opt-in** dataclass — `ThinkingChunk(type="thinking", content=str)` — surfaces AI-SDK `reasoning`/`reasoning-delta` (and pydantic-ai `part_kind == "thinking"`) parts. **`StreamChunk` is NOT widened**: the canonical union stays `StreamChunk = MarkdownTextChunk \| TaskUpdateChunk \| PlanUpdateChunk` — byte-identical to upstream's three variants, so a consumer doing an exhaustive `match` over `StreamChunk` sees zero change on upgrade. `ThinkingChunk` is accepted only at the **stream-input/output boundaries** via the public alias `StreamInput = str \| StreamChunk \| ThinkingChunk` (the `Adapter.stream()` protocol signature, `from_full_stream`/`_from_full_stream` returns, `Thread._wrapped_stream`, and each receiving adapter's `stream()` signature). A producer can yield `ThinkingChunk` (opt-in) and the adapters that receive the stream type-check; code that only references `StreamChunk` never touches it. **OPT-IN, default-off**: emitted only when a caller passes `from_full_stream(stream, emit_thinking=True)` or sets the thread-level `emit_thinking=True` config; the internal `_from_full_stream` threads the same flag. With the default (`emit_thinking=False`) the normalized stream is **byte-for-byte identical** to upstream — reasoning parts are dropped and **no** `ThinkingChunk` is produced. Consumption is graceful: `Thread._handle_stream` never accumulates a `ThinkingChunk` into the posted-message text, and every adapter's stream handler skips it (Slack/Teams expose an optional `render_thinking` hook via `chat_sdk.shared.adapter_utils.maybe_render_thinking`; the text-accumulate adapters ignore it structurally). **Streaming-only — never persisted**: `Message` has no `thinking` field, `to_json()` is unchanged, and a round-tripped `Message` is byte-identical, so cross-SDK state (Redis/Postgres shared with the TS SDK) stays compatible. | Upstream `from-full-stream.ts` forwards only `text-delta` + `finish-step`; AI-SDK `reasoning`/`reasoning-delta` parts fall through and are discarded. `StreamChunk = MarkdownTextChunk \| TaskUpdateChunk \| PlanUpdateChunk` — no reasoning variant and no stream-input alias. Upstream leaves reasoning display to the AI-SDK web UI. | chinchill actively streams agent thinking to Slack/Teams but has to intercept the model stream out-of-band today because the chat-platform SDK has no path for it. This gives the SDK a first-class, opt-in one without changing any default behavior — and crucially **without widening the public `StreamChunk` union**, so consumers referencing it are unaffected. The whole design constraint is that default-off == upstream and `StreamChunk` == upstream: separate opt-in input type, opt-in emit, graceful/skip consume, zero state pollution. Regression coverage: `tests/test_thinking_chunk.py`, `tests/test_from_full_stream.py::TestThinkingOptIn`, `tests/test_types.py::TestThinkingChunk`, plus per-adapter no-crash tests in `tests/test_slack_api.py`, `tests/test_teams_native_streaming.py`, `tests/test_twilio_adapter.py`, `tests/test_messenger_api.py`. | -| Linear agent-activity emit: raw GraphQL (chat@4.31 / #151 — L4) | The agent-session EMIT path (`post_message` session branch, `start_typing` session branch, `stream` → `_stream_in_agent_session` with its flush/`task_update`/`plan_update` logic, and `_parse_message_from_agent_activity`) is ported as **raw GraphQL mutations** over the existing `_graphql_query` helper, **schema-hardened against Linear's published GraphQL schema**. Mutation names: `agentActivityCreate(input: AgentActivityCreateInput!)` and `agentSessionUpdate(id: String!, input: AgentSessionUpdateInput!)`. `content` is sent inside the `AgentActivityCreateInput.content` **`JSONObject!`** scalar — so `type`/`body`/`action`/`parameter`/`result` are inline JSON fields, with the **lowercase** `AgentActivityType` enum values `"response"` / `"thought"` / `"error"` / `"action"` (confirmed lowercase, NOT PascalCase). `ephemeral: Boolean` is a sibling of `content` and is **included only when set** (absent — not `false` — on response/thought/error). The `agentActivityCreate` return selection requests only schema-valid fields (`success`, `agentActivity { id agentSessionId sourceComment { id body parentId createdAt updatedAt url user{…} botActor{…} } }`). `plan` items are `{content, status:"completed"}`. `initialize` additionally captures the viewer's `organization.id` into `_default_organization_id` (mirroring upstream's `defaultOrganizationId`) for the emitted raw message's `organizationId`. | Upstream `adapter-linear/src/index.ts` calls `@linear/sdk`'s `createAgentActivity({agentSessionId, content, ephemeral?})` / `updateAgentSession(id, {plan})`; the SDK owns the GraphQL document, the `AgentActivityType` enum, and the `AgentActivityPayload`/`Comment`/`sourceComment` resolution | `@linear/sdk` is TypeScript-only (no official Linear Python SDK — cf. the `linear_client` getter row), so the SDK calls are reproduced as raw GraphQL. Mutation names, the `content: JSONObject!` shape, the lowercase enum casing, and the `plan` item shape are **schema-hardened against the published schema** (`https://linear.app/developers/agent-interaction`, `https://linear.app/developers/graphql`, and the SDK's generated GraphQL documents). **Live-tenant verification pending**: the exact mutation/field names and enum casing are confirmed against the published schema/docs but have **not** been exercised against a live Linear agent-session tenant; if a future live run surfaces a casing/field mismatch (e.g. an enum the schema renders differently at runtime), update the mutation strings here. Faithful-port hazards preserved: `status ?? "Thinking..."` → `is not None` (an empty status stays `""`); `[title, output].filter(Boolean).join("\n")` → `"\n".join(x for x in [title, output] if x)` (drops `None` and `""`); `markdown.slice(...).trim()` uses the JS-`.trim()` whitespace set (`_JS_WHITESPACE`, mirroring `adapters/telegram/rich.py`), not Python's broader `str.strip()`; `if delta or force`; `ephemeral: status != "complete"`; the missing-final-flush bare `throw new Error(...)` → `RuntimeError`. Regression coverage: `tests/test_linear_agent_session_emit.py`. | +| Linear agent-activity emit: raw GraphQL (chat@4.31 / #151 — L4) | The agent-session EMIT path (`post_message` session branch, `start_typing` session branch, `stream` → `_stream_in_agent_session` with its flush/`task_update`/`plan_update` logic, and `_parse_message_from_agent_activity`) is ported as **raw GraphQL mutations** over the existing `_graphql_query` helper, **schema-hardened against Linear's published GraphQL schema**. Mutation names: `agentActivityCreate(input: AgentActivityCreateInput!)` and `agentSessionUpdate(id: String!, input: AgentSessionUpdateInput!)`. `content` is sent inside the `AgentActivityCreateInput.content` **`JSONObject!`** scalar — so `type`/`body`/`action`/`parameter`/`result` are inline JSON fields, with the **lowercase** `AgentActivityType` enum values `"response"` / `"thought"` / `"error"` / `"action"` (confirmed lowercase, NOT PascalCase). `ephemeral: Boolean` is a sibling of `content` and is **included only when set** (absent — not `false` — on response/thought/error). The `agentActivityCreate` return selection requests only schema-valid fields (`success`, `agentActivity { id agentSession { id } sourceComment { id body parentId createdAt updatedAt url user{…} botActor{…} } }`) — `agentSessionId` is **not** a scalar field on Linear's `AgentActivity` type (the schema exposes only the relation `agentSession: AgentSession!`), so the session id is read off the nested `agentSession { id }` relation; requesting the non-existent scalar would server-reject the whole mutation under GraphQL strict selection validation. `plan` items are `{content, status:"completed"}`. `initialize` additionally captures the viewer's `organization.id` into `_default_organization_id` (mirroring upstream's `defaultOrganizationId`) for the emitted raw message's `organizationId`; on the emit path `organizationId` falls back to `""` when `_default_organization_id` is unset (no per-request installation context is plumbed — a pre-existing adapter-wide divergence from upstream, which throws `AuthenticationError` when no organization is resolvable). | Upstream `adapter-linear/src/index.ts` calls `@linear/sdk`'s `createAgentActivity({agentSessionId, content, ephemeral?})` / `updateAgentSession(id, {plan})`; the SDK owns the GraphQL document, the `AgentActivityType` enum, and the `AgentActivityPayload`/`Comment`/`sourceComment` resolution | `@linear/sdk` is TypeScript-only (no official Linear Python SDK — cf. the `linear_client` getter row), so the SDK calls are reproduced as raw GraphQL. Mutation names, the `content: JSONObject!` shape, the lowercase enum casing, and the `plan` item shape are **schema-hardened against the published schema** (`https://linear.app/developers/agent-interaction`, `https://linear.app/developers/graphql`, and the SDK's generated GraphQL documents). **Live-tenant verification pending**: the exact mutation/field names and enum casing are confirmed against the published schema/docs but have **not** been exercised against a live Linear agent-session tenant; if a future live run surfaces a casing/field mismatch (e.g. an enum the schema renders differently at runtime), update the mutation strings here. Faithful-port hazards preserved: `status ?? "Thinking..."` → `is not None` (an empty status stays `""`); `[title, output].filter(Boolean).join("\n")` → `"\n".join(x for x in [title, output] if x)` (drops `None` and `""`); `markdown.slice(...).trim()` uses the JS-`.trim()` whitespace set (`_JS_WHITESPACE`, mirroring `adapters/telegram/rich.py`), not Python's broader `str.strip()`; `if delta or force`; `ephemeral: status != "complete"`; the missing-final-flush bare `throw new Error(...)` → `RuntimeError`. Regression coverage: `tests/test_linear_agent_session_emit.py`. | ### Platform-specific gaps | Area | Python | TS | Rationale | diff --git a/src/chat_sdk/adapters/linear/adapter.py b/src/chat_sdk/adapters/linear/adapter.py index 79c2732..c9a8978 100644 --- a/src/chat_sdk/adapters/linear/adapter.py +++ b/src/chat_sdk/adapters/linear/adapter.py @@ -137,13 +137,20 @@ # "error"/"action"), ``agentSessionId: String!`` and ``ephemeral: Boolean``. The # return selection requests only schema-valid fields that # ``_parse_message_from_agent_activity`` reads off the resolved ``sourceComment``. +# NOTE: ``agentSessionId`` is NOT a scalar field on Linear's ``AgentActivity`` +# type — the schema exposes only the relation ``agentSession: AgentSession!``. +# GraphQL strict-validates selections, so requesting a non-existent +# ``agentSessionId`` field server-rejects the whole mutation; select the +# ``agentSession { id }`` relation instead. _AGENT_ACTIVITY_CREATE_MUTATION = """ mutation AgentActivityCreate($input: AgentActivityCreateInput!) { agentActivityCreate(input: $input) { success agentActivity { id - agentSessionId + agentSession { + id + } sourceComment { id body @@ -1253,7 +1260,11 @@ async def _parse_message_from_agent_activity( "linear", ) - activity_session_id = activity.get("agentSessionId") + # The mutation selects the ``agentSession { id }`` relation (NOT a + # non-existent scalar ``agentSessionId`` field — see + # ``_AGENT_ACTIVITY_CREATE_MUTATION``); read the session id None-safely + # off the nested relation node. + activity_session_id = (activity.get("agentSession") or {}).get("id") if not activity_session_id: raise AdapterError( f"Missing agentSessionId for Linear agent activity {activity.get('id')}", @@ -1354,10 +1365,13 @@ def _raw_message_from_source_comment( # re-derived on read via :meth:`_parse_agent_session_message`. Encode the # routed thread id (carrying the ``:s:{session}`` segment) so callers can # round-trip it, matching the comment branch's ``thread_id`` semantics. + # Upstream ``parseMessage`` (index.ts:2033-2038) and the adapter's own + # read-path :meth:`_parse_agent_session_message` encode the source + # comment's OWN id (``commentId: raw.comment.id``) — NOT its parentId. thread_id = self.encode_thread_id( LinearThreadId( issue_id=issue_id, - comment_id=cast("str | None", comment_data.get("parentId")), + comment_id=cast("str | None", comment_data.get("id")), agent_session_id=agent_session_id, ) ) @@ -1953,14 +1967,21 @@ def _read(chunk: Any, name: str) -> Any: else: # ``ephemeral: status !== "complete"`` — in-progress/pending # actions are ephemeral; only a completed action persists. + # Upstream passes ``result: chunk.output`` (string|undefined); + # ``JSON.stringify`` OMITS a key whose value is undefined, so + # OMIT ``result`` entirely when ``output`` is None to match + # the key-absent wire shape (rather than serializing + # ``"result": null``). + action_content: dict[str, Any] = { + "type": "action", + "action": title, + "parameter": "", + } + if output is not None: + action_content["result"] = output await self._create_agent_activity( agent_session_id, - { - "type": "action", - "action": title, - "parameter": "", - "result": output, - }, + action_content, ephemeral=status != "complete", ) continue diff --git a/tests/test_linear_agent_session_emit.py b/tests/test_linear_agent_session_emit.py index 9653cfc..929a2c9 100644 --- a/tests/test_linear_agent_session_emit.py +++ b/tests/test_linear_agent_session_emit.py @@ -110,10 +110,19 @@ def _activity_payload( agent_session_id: str | None = "session-789", source_comment: Any = "default", ) -> dict[str, Any]: - """The ``agentActivityCreate`` payload node (``{success, agentActivity}``).""" + """The ``agentActivityCreate`` payload node (``{success, agentActivity}``). + + The session id is carried under the ``agentSession { id }`` RELATION — the + real server shape. ``agentSessionId`` is NOT a scalar field on Linear's + ``AgentActivity`` type (the schema exposes only ``agentSession: + AgentSession!``), so a fixture emitting a flat ``agentSessionId`` would + fabricate a field the server never returns. ``agent_session_id=None`` + models a payload with no resolvable ``agentSession.id`` (no relation node). + """ + agent_session: dict[str, Any] | None = {"id": agent_session_id} if agent_session_id is not None else None activity: dict[str, Any] | None = { "id": activity_id, - "agentSessionId": agent_session_id, + "agentSession": agent_session, "sourceComment": _source_comment(activity_id=activity_id, body=body) if source_comment == "default" else source_comment, @@ -149,6 +158,13 @@ async def test_uses_agent_activity_create_with_response(self) -> None: query = call_args[0][0] variables = call_args[0][1] assert "agentActivityCreate" in query + # GraphQL-selection proof (fix #1): the return selection requests the + # ``agentSession { id }`` RELATION, NOT a non-existent scalar + # ``agentSessionId`` field (which would server-reject the mutation under + # strict selection validation). FAILS if the selection regresses to + # ``agentSessionId``. + assert "agentSession {" in query + assert "agentSessionId" not in query assert variables["input"]["agentSessionId"] == "session-789" # Enum-swap proof: the content type MUST be the lowercase "response". assert variables["input"]["content"] == {"type": "response", "body": "Agent response"} @@ -157,6 +173,11 @@ async def test_uses_agent_activity_create_with_response(self) -> None: # The resolved message is built off the source comment. assert result.id == "comment-activity-123" + # thread_id is encoded from the source comment's OWN id (NOT its + # parentId). Source comment id == "comment-activity-123", + # parentId == "comment-root" — the encoded ``:c:`` segment MUST be the + # own id. (Old parentId code would emit ``:c:comment-root:``.) + assert result.thread_id == "linear:issue-123:c:comment-activity-123:s:session-789" assert result.raw["kind"] == "agent_session_comment" assert result.raw["organizationId"] == "org-123" assert result.raw["agentSessionId"] == "session-789" @@ -251,6 +272,77 @@ async def test_raises_when_source_comment_missing(self) -> None: with pytest.raises(AdapterError, match="Failed to resolve source comment for Linear agent activity"): await adapter.post_message(_SESSION_THREAD, "Agent response") + @pytest.mark.asyncio + async def test_raises_when_agent_session_id_missing(self) -> None: + """No resolvable ``agentSession.id`` on the activity → ``AdapterError``. + + The session id is read None-safely off the ``agentSession { id }`` + RELATION (``(activity.get("agentSession") or {}).get("id")``). When the + relation node is absent (``agent_session_id=None`` → ``agentSession: + None``), the read yields ``None`` and the guard raises. This must FAIL on + the old code that read a flat (non-existent) ``activity["agentSessionId"]`` + scalar — with the corrected ``agentSession { id }`` fixture there is no + such key, so the old read would surface ``None`` differently / KeyError + rather than this guarded message. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload(agent_session_id=None))) + + with pytest.raises(AdapterError, match="Missing agentSessionId"): + await adapter.post_message(_SESSION_THREAD, "Agent response") + + @pytest.mark.asyncio + async def test_bot_author_nullish_fallbacks(self) -> None: + """``botActor.id ?? bot_user_id`` / display/name chained-nullish fallbacks. + + A botActor with ``id=None`` / ``userDisplayName=None`` / ``name=None`` + exercises the three ``is not None`` chains: + - ``id`` None → falls back to the adapter's ``_bot_user_id``. + - ``userDisplayName`` None then ``name`` None → "unknown". + - ``name`` None then ``userDisplayName`` None → "unknown". + + Must FAIL if any chain is swapped from ``is not None`` to ``or`` / + truthiness (which would coerce identically here only because the values + are None — but a partial fixture below pins the precedence so a swap + that drops the *first* operand is caught). + """ + adapter = _make_adapter() + bot_comment = _source_comment(with_bot_actor=False) + bot_comment["botActor"] = {"id": None, "name": None, "userDisplayName": None} + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload(source_comment=bot_comment))) + + result = await adapter.post_message(_SESSION_THREAD, "Agent response") + + user = result.raw["comment"]["user"] + assert user["type"] == "bot" + # id None → bot_user_id fallback (NOT ""). + assert user["id"] == "bot-user-id" + assert user["displayName"] == "unknown" + assert user["fullName"] == "unknown" + + @pytest.mark.asyncio + async def test_bot_author_nullish_precedence_keeps_first_present_operand(self) -> None: + """``userDisplayName ?? name`` keeps ``userDisplayName`` when it is "" (present). + + ``is not None`` keeps an EMPTY-STRING ``userDisplayName`` (it is present, + just falsy); a truthiness ``or`` swap would wrongly fall through to + ``name``. Conversely ``fullName`` = ``name ?? userDisplayName`` keeps the + empty ``name``. This pins the ``is not None`` precedence so a ``or`` swap + FAILS. + """ + adapter = _make_adapter() + bot_comment = _source_comment(with_bot_actor=False) + bot_comment["botActor"] = {"id": "ba-1", "name": "", "userDisplayName": ""} + adapter._graphql_query = AsyncMock(return_value=_graphql_return(_activity_payload(source_comment=bot_comment))) + + result = await adapter.post_message(_SESSION_THREAD, "Agent response") + + user = result.raw["comment"]["user"] + assert user["id"] == "ba-1" + # "" is present → kept (NOT replaced by name / "unknown"). + assert user["displayName"] == "" + assert user["fullName"] == "" + # =========================================================================== # start_typing — agent-session branch @@ -501,16 +593,153 @@ async def test_empty_delta_flush_is_noop_when_not_forced(self) -> None: assert types == ["action", "action", "response"] @pytest.mark.asyncio - async def test_raises_when_final_activity_missing(self) -> None: - """``if not finalActivity`` → raise (the final mutation resolved nothing).""" + async def test_raises_runtime_error_when_final_flush_returns_nothing(self) -> None: + """``if not finalActivity`` → ``RuntimeError`` (force-flush resolved an empty node). + + The final force-flush calls ``_create_agent_activity``, which returns + ``result["data"].get("agentActivityCreate", {})``. When the mutation + response LACKS the ``agentActivityCreate`` key, that read yields an empty + ``{}`` (falsy) → ``if not final_activity`` fires the bare + ``RuntimeError`` (upstream's missing-final-flush ``throw new Error``), + NOT the later ``AdapterError`` parse path. This is a DISTINCT branch from + the ``{success:False, agentActivity:None}`` parse failure below. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock(return_value={"data": {}}) + + with pytest.raises(RuntimeError, match="Failed to flush final markdown delta"): + await adapter.stream(_SESSION_THREAD, _astream("Hello world")) + + @pytest.mark.asyncio + async def test_raises_adapter_error_when_final_activity_parse_fails(self) -> None: + """``{success:False, agentActivity:None}`` is a truthy node that FAILS to parse. + + Here the force-flush DOES return a non-empty payload node + (``{success:False, agentActivity:None}`` is truthy, so ``if not + final_activity`` is False), and ``_parse_message_from_agent_activity`` + raises ``AdapterError`` on the unsuccessful activity. Separate branch + from the ``RuntimeError`` force-flush case above. + """ adapter = _make_adapter() adapter._graphql_query = AsyncMock( return_value={"data": {"agentActivityCreate": {"success": False, "agentActivity": None}}} ) - with pytest.raises(AdapterError): + with pytest.raises(AdapterError, match="Failed to create Linear agent activity"): await adapter.stream(_SESSION_THREAD, _astream("Hello world")) + @pytest.mark.asyncio + async def test_stream_thread_id_uses_source_comment_own_id(self) -> None: + """The streamed result's ``thread_id`` encodes the source comment's OWN id. + + The final-flush source comment has ``id="comment-final"`` and + ``parentId="comment-root"`` (id != parentId). The encoded ``:c:`` segment + MUST be the own id ``comment-final`` — NOT ``comment-root``. This FAILS + on the old code that encoded ``comment_id=comment_data["parentId"]``. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value=_graphql_return(_activity_payload(activity_id="final", body="Hello world")) + ) + + result = await adapter.stream(_SESSION_THREAD, _astream("Hello world")) + + assert result.thread_id == "linear:issue-123:c:comment-final:s:session-789" + + @pytest.mark.asyncio + async def test_task_update_omits_result_key_when_output_none(self) -> None: + """``result: chunk.output`` with ``output=None`` → key OMITTED on the wire. + + Upstream passes ``result: chunk.output`` (string|undefined); JSON.stringify + OMITS an undefined key. So a non-error task_update with ``output=None`` must + build an action content dict WITHOUT a ``result`` key (NOT ``"result": + None``). FAILS on the old code that always set ``"result": output``. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="task")), + _graphql_return(_activity_payload(activity_id="final", body="done")), + ] + ) + + await adapter.stream( + _SESSION_THREAD, + _astream( + TaskUpdateChunk(id="t", title="Search docs", status="in_progress", output=None), + MarkdownTextChunk(text="done"), + ), + ) + + action_content = adapter._graphql_query.call_args_list[0].args[1]["input"]["content"] + assert action_content == {"type": "action", "action": "Search docs", "parameter": ""} + assert "result" not in action_content + + @pytest.mark.asyncio + async def test_task_update_includes_result_key_when_output_present(self) -> None: + """Companion to the omit case: a present ``output`` keeps the ``result`` key.""" + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + side_effect=[ + _graphql_return(_activity_payload(activity_id="task")), + _graphql_return(_activity_payload(activity_id="final", body="done")), + ] + ) + + await adapter.stream( + _SESSION_THREAD, + _astream( + TaskUpdateChunk(id="t", title="Search docs", status="in_progress", output="Found 3"), + MarkdownTextChunk(text="done"), + ), + ) + + action_content = adapter._graphql_query.call_args_list[0].args[1]["input"]["content"] + assert action_content == { + "type": "action", + "action": "Search docs", + "parameter": "", + "result": "Found 3", + } + + @pytest.mark.asyncio + async def test_final_delta_retains_nel_per_js_trim(self) -> None: + """``.strip(_JS_WHITESPACE)`` KEEPS NEL (``\\x85``) — JS ``.trim()`` does not strip it. + + NEL is NOT in the JS-``.trim()`` whitespace set, so the trailing + ``\\x85`` survives the delta trim. Python's bare ``str.strip()`` WOULD + remove it, so this FAILS if ``.strip(_JS_WHITESPACE)`` is mutated to a + bare ``.strip()``. + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value=_graphql_return(_activity_payload(activity_id="final", body="x")) + ) + + await adapter.stream(_SESSION_THREAD, _astream("Hello world\x85")) + + only_input = adapter._graphql_query.call_args.args[1]["input"] + assert only_input["content"] == {"type": "response", "body": "Hello world\x85"} + + @pytest.mark.asyncio + async def test_final_delta_strips_bom_per_js_trim(self) -> None: + """``.strip(_JS_WHITESPACE)`` STRIPS the BOM (``\\ufeff``) — JS ``.trim()`` removes it. + + The BOM IS in the JS-``.trim()`` whitespace set, so a trailing + ``\\ufeff`` is trimmed off the delta. Python's bare ``str.strip()`` does + NOT strip the BOM, so this FAILS if ``.strip(_JS_WHITESPACE)`` is mutated + to a bare ``.strip()`` (the BOM would survive). + """ + adapter = _make_adapter() + adapter._graphql_query = AsyncMock( + return_value=_graphql_return(_activity_payload(activity_id="final", body="x")) + ) + + await adapter.stream(_SESSION_THREAD, _astream("Hello world")) + + only_input = adapter._graphql_query.call_args.args[1]["input"] + assert only_input["content"] == {"type": "response", "body": "Hello world"} + @pytest.mark.asyncio async def test_dispatches_to_comment_stream_for_non_session_thread(self) -> None: """Regression: a non-session thread uses the comment-path stream, NOT agentActivityCreate."""