From bd74d47ec1c474fca76808b526f13e1563c2c2c4 Mon Sep 17 00:00:00 2001 From: murashit Date: Mon, 13 Jul 2026 13:52:57 +0900 Subject: [PATCH] Align side chats with Codex CLI lifecycle --- src/app-server/connection/client.ts | 2 + src/app-server/protocol/side-chat.ts | 26 ++++ src/app-server/services/threads.ts | 37 +++++ .../transports/session-transports.ts | 19 ++- .../threads/ephemeral-thread-lifecycle.ts | 28 +++- .../threads/ephemeral-thread-transport.ts | 8 +- tests/app-server/threads.test.ts | 126 +++++++++++++++--- .../ephemeral-thread-lifecycle.test.ts | 49 +++++-- 8 files changed, 250 insertions(+), 45 deletions(-) create mode 100644 src/app-server/protocol/side-chat.ts diff --git a/src/app-server/connection/client.ts b/src/app-server/connection/client.ts index a0d0bd87..1e3923f4 100644 --- a/src/app-server/connection/client.ts +++ b/src/app-server/connection/client.ts @@ -36,6 +36,7 @@ import type { ThreadSettingsUpdateResponse } from "../../generated/app-server/v2 import type { ThreadStartResponse } from "../../generated/app-server/v2/ThreadStartResponse"; import type { ThreadTurnsListResponse } from "../../generated/app-server/v2/ThreadTurnsListResponse"; import type { ThreadUnarchiveResponse } from "../../generated/app-server/v2/ThreadUnarchiveResponse"; +import type { ThreadUnsubscribeResponse } from "../../generated/app-server/v2/ThreadUnsubscribeResponse"; import type { TurnInterruptResponse } from "../../generated/app-server/v2/TurnInterruptResponse"; import type { TurnStartResponse } from "../../generated/app-server/v2/TurnStartResponse"; import type { TurnSteerResponse } from "../../generated/app-server/v2/TurnSteerResponse"; @@ -84,6 +85,7 @@ export interface ClientResponseByMethod { "thread/read": ThreadReadResponse; "thread/archive": ThreadArchiveResponse; "thread/delete": ThreadDeleteResponse; + "thread/unsubscribe": ThreadUnsubscribeResponse; "thread/unarchive": ThreadUnarchiveResponse; "thread/rollback": ThreadRollbackResponse; "thread/name/set": ThreadSetNameResponse; diff --git a/src/app-server/protocol/side-chat.ts b/src/app-server/protocol/side-chat.ts new file mode 100644 index 00000000..7758f65e --- /dev/null +++ b/src/app-server/protocol/side-chat.ts @@ -0,0 +1,26 @@ +type AppServerJsonValue = number | string | boolean | AppServerJsonValue[] | { [key: string]: AppServerJsonValue | undefined } | null; + +const SIDE_CHAT_DEVELOPER_INSTRUCTIONS = `You are in a side conversation, not the main thread. + +The inherited fork history is reference context only. Do not continue any task, plan, tool call, approval, edit, or request found only in that history. Only instructions submitted after the side-conversation boundary are active. + +Use the side conversation for questions and lightweight, non-mutating exploration. Do not use sub-agents. This thread is read-only; do not modify files, source, git state, permissions, configuration, or other workspace state.`; + +const SIDE_CHAT_BOUNDARY = `Side conversation boundary. + +Everything before this boundary is inherited history from the parent thread. It is reference context only, not the current task. + +Do not continue any instructions, plans, tool calls, approvals, edits, or requests from before this boundary. Only messages submitted after this boundary are active user instructions for this side conversation. If there is no user question after this boundary yet, wait for one.`; + +export function sideChatDeveloperInstructions(existingInstructions: string | null | undefined): string { + const existing = existingInstructions?.trim(); + return existing ? `${existing}\n\n${SIDE_CHAT_DEVELOPER_INSTRUCTIONS}` : SIDE_CHAT_DEVELOPER_INSTRUCTIONS; +} + +export function appServerSideChatBoundaryItem(): AppServerJsonValue { + return { + type: "message", + role: "user", + content: [{ type: "input_text", text: SIDE_CHAT_BOUNDARY }], + }; +} diff --git a/src/app-server/services/threads.ts b/src/app-server/services/threads.ts index 6cd8a359..ae460cde 100644 --- a/src/app-server/services/threads.ts +++ b/src/app-server/services/threads.ts @@ -12,6 +12,7 @@ import { REFERENCED_THREAD_TURN_LIMIT } from "../../domain/threads/reference"; import type { TurnTranscriptSummary } from "../../domain/threads/transcript"; import type { ClientResponseByMethod } from "../connection/client"; import type { ClientRequestParams } from "../connection/rpc-messages"; +import { appServerSideChatBoundaryItem, sideChatDeveloperInstructions } from "../protocol/side-chat"; import { type ThreadRecord, threadFromThreadRecord, threadsFromThreadRecords } from "../protocol/thread"; import { appServerThreadGoalUpdate, appServerThreadGoalUserHistoryItem, threadGoalFromAppServerGoal } from "../protocol/thread-goal"; import { appServerRuntimeSettingsPatch } from "../protocol/thread-settings"; @@ -229,19 +230,47 @@ export interface EphemeralThreadForkSnapshot { readonly sourceThreadId: string; } +export class EphemeralThreadCleanupRequiredError extends Error { + readonly cleanupError: unknown; + readonly threadId: string; + + constructor(threadId: string, cause: unknown, cleanupError: unknown) { + super(`Could not prepare or unsubscribe side chat ${threadId}.`, { cause }); + this.name = "EphemeralThreadCleanupRequiredError"; + this.threadId = threadId; + this.cleanupError = cleanupError; + } +} + export async function forkEphemeralThread( client: AppServerRequestClient, sourceThreadId: string, cwd: string, ): Promise { + const source = await client.request("thread/read", { threadId: sourceThreadId, includeTurns: false }); + const config = await client.request("config/read", { cwd: source.thread.cwd }); const response = await client.request("thread/fork", { threadId: sourceThreadId, cwd, ephemeral: true, sandbox: "read-only", approvalPolicy: "never", + developerInstructions: sideChatDeveloperInstructions(config.config.developer_instructions), excludeTurns: true, }); + try { + await client.request("thread/inject_items", { + threadId: response.thread.id, + items: [appServerSideChatBoundaryItem()], + }); + } catch (error) { + try { + await unsubscribeThread(client, response.thread.id, { timeoutMs: 5_000 }); + } catch (cleanupError) { + throw new EphemeralThreadCleanupRequiredError(response.thread.id, error, cleanupError); + } + throw error; + } return { activation: threadActivationSnapshotFromAppServerResponse(response), sourceThreadId, @@ -260,6 +289,14 @@ export async function deleteThread(client: AppServerRequestClient, threadId: str await client.request("thread/delete", { threadId }, options); } +export async function unsubscribeThread( + client: AppServerRequestClient, + threadId: string, + options: { timeoutMs?: number } = {}, +): Promise { + await client.request("thread/unsubscribe", { threadId }, options); +} + export async function restoreArchivedThread(client: AppServerRequestClient, threadId: string): Promise { const response = await client.request("thread/unarchive", { threadId }); return threadFromThreadRecord(response.thread); diff --git a/src/features/chat/app-server/transports/session-transports.ts b/src/features/chat/app-server/transports/session-transports.ts index d2a6b5e6..c72f2140 100644 --- a/src/features/chat/app-server/transports/session-transports.ts +++ b/src/features/chat/app-server/transports/session-transports.ts @@ -3,7 +3,7 @@ import type { AppServerRequestClient } from "../../../../app-server/services/req import { clearThreadGoal, compactThread, - deleteThread, + EphemeralThreadCleanupRequiredError, forkEphemeralThread, forkThread, listThreadTurns, @@ -14,6 +14,7 @@ import { setThreadGoal, startThread, threadActivationSnapshotFromAppServerResponse, + unsubscribeThread, updateThreadSettings, } from "../../../../app-server/services/threads"; import { interruptTurn, startTurn, steerTurn } from "../../../../app-server/services/turns"; @@ -172,10 +173,20 @@ function createChatThreadMutationTransport(host: ChatAppServerTransportHost): Th function createChatEphemeralThreadTransport(host: ChatAppServerTransportHost): EphemeralThreadTransport { return { forkEphemeralThread: (sourceThreadId) => - withConnectedChatAppServerClient(host, (client) => forkEphemeralThread(client, sourceThreadId, host.vaultPath)), - deleteEphemeralThread: async (threadId) => { + withConnectedChatAppServerClient(host, async (client) => { + try { + const snapshot = await forkEphemeralThread(client, sourceThreadId, host.vaultPath); + return { kind: "ready" as const, ...snapshot }; + } catch (error) { + if (error instanceof EphemeralThreadCleanupRequiredError) { + return { kind: "cleanup-required" as const, threadId: error.threadId }; + } + throw error; + } + }), + unsubscribeEphemeralThread: async (threadId) => { const result = await withCurrentChatAppServerClient(host, async (client) => { - await deleteThread(client, threadId, { timeoutMs: 5_000 }); + await unsubscribeThread(client, threadId, { timeoutMs: 5_000 }); return true; }); return result ?? false; diff --git a/src/features/chat/application/threads/ephemeral-thread-lifecycle.ts b/src/features/chat/application/threads/ephemeral-thread-lifecycle.ts index abd9eeef..bae174bc 100644 --- a/src/features/chat/application/threads/ephemeral-thread-lifecycle.ts +++ b/src/features/chat/application/threads/ephemeral-thread-lifecycle.ts @@ -28,10 +28,11 @@ interface EphemeralThreadLifecycleHost { export function createEphemeralThreadLifecycle(host: EphemeralThreadLifecycleHost): EphemeralThreadLifecycle { let disposed = false; let openGeneration = 0; - const deleteActiveEphemeralThread = async (): Promise => { + const cleanupRequiredThreadIds = new Set(); + const unsubscribeActiveEphemeralThread = async (): Promise => { const active = host.stateStore.getState().activeThread; if (active.id && active.lifetime?.kind === "ephemeral") { - return host.transport.deleteEphemeralThread(active.id); + return host.transport.unsubscribeEphemeralThread(active.id); } return true; }; @@ -40,11 +41,17 @@ export function createEphemeralThreadLifecycle(host: EphemeralThreadLifecycleHos async open(input): Promise { const generation = ++openGeneration; if (!(await host.ensureConnected())) return false; - const snapshot = await host.transport.forkEphemeralThread(input.sourceThreadId); - if (!snapshot) return false; + const result = await host.transport.forkEphemeralThread(input.sourceThreadId); + if (!result) return false; + if (result.kind === "cleanup-required") { + cleanupRequiredThreadIds.add(result.threadId); + host.addSystemMessage("Could not prepare the side chat. Cleanup will be retried when this view closes."); + return false; + } + const snapshot = result; if (disposed || generation !== openGeneration) { try { - await host.transport.deleteEphemeralThread(snapshot.activation.thread.id); + await host.transport.unsubscribeEphemeralThread(snapshot.activation.thread.id); } catch { // A late fork must never become active, even when best-effort cleanup fails. } @@ -69,7 +76,7 @@ export function createEphemeralThreadLifecycle(host: EphemeralThreadLifecycleHos return false; } try { - if (!(await deleteActiveEphemeralThread())) { + if (!(await unsubscribeActiveEphemeralThread())) { host.addSystemMessage("Could not discard the side chat. Try again before switching threads."); return false; } @@ -92,10 +99,17 @@ export function createEphemeralThreadLifecycle(host: EphemeralThreadLifecycleHos await settleWithin(host.interruptTurn(threadId, turnId), EPHEMERAL_INTERRUPT_DISPOSE_TIMEOUT_MS); } try { - await deleteActiveEphemeralThread(); + await unsubscribeActiveEphemeralThread(); } catch { // Ephemeral cleanup must not prevent the panel from closing. } + for (const threadId of cleanupRequiredThreadIds) { + try { + if (await host.transport.unsubscribeEphemeralThread(threadId)) cleanupRequiredThreadIds.delete(threadId); + } catch { + // Closing the app-server connection provides the final subscription cleanup boundary. + } + } }, }; } diff --git a/src/features/chat/application/threads/ephemeral-thread-transport.ts b/src/features/chat/application/threads/ephemeral-thread-transport.ts index 19fa4411..aae010e3 100644 --- a/src/features/chat/application/threads/ephemeral-thread-transport.ts +++ b/src/features/chat/application/threads/ephemeral-thread-transport.ts @@ -1,6 +1,10 @@ import type { ThreadActivationSnapshot } from "../../../../domain/threads/activation"; +type EphemeralThreadForkResult = + | { kind: "ready"; activation: ThreadActivationSnapshot; sourceThreadId: string } + | { kind: "cleanup-required"; threadId: string }; + export interface EphemeralThreadTransport { - forkEphemeralThread(sourceThreadId: string): Promise<{ activation: ThreadActivationSnapshot; sourceThreadId: string } | null>; - deleteEphemeralThread(threadId: string): Promise; + forkEphemeralThread(sourceThreadId: string): Promise; + unsubscribeEphemeralThread(threadId: string): Promise; } diff --git a/tests/app-server/threads.test.ts b/tests/app-server/threads.test.ts index a299d872..e3fed203 100644 --- a/tests/app-server/threads.test.ts +++ b/tests/app-server/threads.test.ts @@ -2,11 +2,13 @@ import { describe, expect, it, vi } from "vitest"; import type { AppServerRequestClient } from "../../src/app-server/services/request-client"; import { + EphemeralThreadCleanupRequiredError, forkEphemeralThread, listThreads, startEphemeralThread, startThread, threadFromAppServerRecord, + unsubscribeThread, } from "../../src/app-server/services/threads"; describe("app-server thread response adapters", () => { @@ -45,34 +47,122 @@ describe("app-server thread response adapters", () => { }); }); - it("forks read-only ephemeral side-chat threads", async () => { + it("forks read-only ephemeral side-chat threads behind a model-visible boundary", async () => { const client = { - request: vi.fn().mockResolvedValue({ - thread: { id: "side", preview: "", name: null, createdAt: 1, updatedAt: 1 }, - cwd: "/vault", - model: "gpt-5.5", - serviceTier: null, - approvalsReviewer: null, - reasoningEffort: null, - approvalPolicy: "never", - sandbox: { type: "readOnly", networkAccess: false }, - activePermissionProfile: null, + request: vi.fn((method: string) => { + if (method === "thread/read") return Promise.resolve({ thread: { cwd: "/source-project" } }); + if (method === "config/read") return Promise.resolve({ config: { developer_instructions: "Existing policy." } }); + if (method === "thread/inject_items") return Promise.resolve({}); + return Promise.resolve({ + thread: { id: "side", preview: "", name: null, createdAt: 1, updatedAt: 1 }, + cwd: "/vault", + model: "gpt-5.5", + serviceTier: null, + approvalsReviewer: null, + reasoningEffort: null, + approvalPolicy: "never", + sandbox: { type: "readOnly", networkAccess: false }, + activePermissionProfile: null, + }); }), } as unknown as AppServerRequestClient; const result = await forkEphemeralThread(client, "source", "/vault"); - expect(client.request).toHaveBeenCalledWith("thread/fork", { - threadId: "source", - cwd: "/vault", - ephemeral: true, - sandbox: "read-only", - approvalPolicy: "never", - excludeTurns: true, + expect(client.request).toHaveBeenNthCalledWith( + 3, + "thread/fork", + expect.objectContaining({ + threadId: "source", + cwd: "/vault", + ephemeral: true, + sandbox: "read-only", + approvalPolicy: "never", + excludeTurns: true, + developerInstructions: expect.stringContaining("Existing policy."), + }), + ); + expect(client.request).toHaveBeenNthCalledWith(1, "thread/read", { threadId: "source", includeTurns: false }); + expect(client.request).toHaveBeenNthCalledWith(2, "config/read", { cwd: "/source-project" }); + expect(client.request).toHaveBeenNthCalledWith(4, "thread/inject_items", { + threadId: "side", + items: [ + { + type: "message", + role: "user", + content: [{ type: "input_text", text: expect.stringContaining("Side conversation boundary.") }], + }, + ], }); expect(result).toMatchObject({ sourceThreadId: "source", activation: { thread: { id: "side" }, cwd: "/vault" } }); }); + it("unsubscribes ephemeral threads instead of deleting them", async () => { + const client = { request: vi.fn().mockResolvedValue({ status: "unsubscribed" }) } as unknown as AppServerRequestClient; + + await unsubscribeThread(client, "side", { timeoutMs: 5_000 }); + + expect(client.request).toHaveBeenCalledWith("thread/unsubscribe", { threadId: "side" }, { timeoutMs: 5_000 }); + }); + + it("unsubscribes a fork when boundary injection fails", async () => { + const injectionError = new Error("inject failed"); + const request = vi.fn((method: string) => { + if (method === "thread/read") return Promise.resolve({ thread: { cwd: "/source-project" } }); + if (method === "config/read") return Promise.resolve({ config: { developer_instructions: null } }); + if (method === "thread/fork") { + return Promise.resolve({ + thread: { id: "side", preview: "", name: null, createdAt: 1, updatedAt: 1 }, + cwd: "/vault", + model: "gpt-5.5", + serviceTier: null, + approvalsReviewer: null, + reasoningEffort: null, + approvalPolicy: "never", + sandbox: { type: "readOnly", networkAccess: false }, + activePermissionProfile: null, + }); + } + if (method === "thread/inject_items") return Promise.reject(injectionError); + return Promise.resolve({ status: "unsubscribed" }); + }); + const client = { request } as unknown as AppServerRequestClient; + + await expect(forkEphemeralThread(client, "source", "/vault")).rejects.toBe(injectionError); + + expect(request).toHaveBeenLastCalledWith("thread/unsubscribe", { threadId: "side" }, { timeoutMs: 5_000 }); + }); + + it("returns the fork id when boundary injection and cleanup both fail", async () => { + const injectionError = new Error("inject failed"); + const cleanupError = new Error("unsubscribe failed"); + const request = vi.fn((method: string) => { + if (method === "thread/read") return Promise.resolve({ thread: { cwd: "/source-project" } }); + if (method === "config/read") return Promise.resolve({ config: { developer_instructions: null } }); + if (method === "thread/fork") { + return Promise.resolve({ + thread: { id: "side", preview: "", name: null, createdAt: 1, updatedAt: 1 }, + cwd: "/vault", + model: "gpt-5.5", + serviceTier: null, + approvalsReviewer: null, + reasoningEffort: null, + approvalPolicy: "never", + sandbox: { type: "readOnly", networkAccess: false }, + activePermissionProfile: null, + }); + } + if (method === "thread/inject_items") return Promise.reject(injectionError); + return Promise.reject(cleanupError); + }); + const client = { request } as unknown as AppServerRequestClient; + + const error = await forkEphemeralThread(client, "source", "/vault").catch((reason: unknown) => reason); + + expect(error).toBeInstanceOf(EphemeralThreadCleanupRequiredError); + expect(error).toMatchObject({ threadId: "side", cause: injectionError, cleanupError }); + }); + it("starts panel-owned threads with the codex-panel service name", async () => { const client = { request: vi.fn().mockResolvedValue({ thread: { id: "thread-new" } }), diff --git a/tests/features/chat/application/threads/ephemeral-thread-lifecycle.test.ts b/tests/features/chat/application/threads/ephemeral-thread-lifecycle.test.ts index 3ba94109..bf71e0f5 100644 --- a/tests/features/chat/application/threads/ephemeral-thread-lifecycle.test.ts +++ b/tests/features/chat/application/threads/ephemeral-thread-lifecycle.test.ts @@ -42,7 +42,7 @@ describe("ephemeral thread lifecycle", () => { expect(store.getState().threadList.listedThreads.map((thread) => thread.id)).toEqual(["source"]); }); - it("deletes an idle ephemeral thread before persistent navigation", async () => { + it("unsubscribes an idle ephemeral thread before persistent navigation", async () => { const store = createChatStateStore(); const transport = transportMock(); const lifecycle = createEphemeralThreadLifecycle({ @@ -57,7 +57,7 @@ describe("ephemeral thread lifecycle", () => { await expect(lifecycle.prepareForPersistentNavigation()).resolves.toBe(true); - expect(transport.deleteEphemeralThread).toHaveBeenCalledWith("side"); + expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"); expect(store.getState().activeThread.id).toBeNull(); }); @@ -79,16 +79,16 @@ describe("ephemeral thread lifecycle", () => { await lifecycle.dispose(); expect(interruptTurn).toHaveBeenCalledWith("side", "turn"); - expect(transport.deleteEphemeralThread).toHaveBeenCalledWith("side"); + expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"); }); - it("deletes a fork that resolves after the lifecycle is disposed without activating it", async () => { + it("unsubscribes a fork that resolves after the lifecycle is disposed without activating it", async () => { const store = createChatStateStore(); let resolveFork!: (value: Awaited>) => void; const transport = transportMock(); transport.forkEphemeralThread = vi.fn( () => - new Promise<{ activation: ThreadActivationSnapshot; sourceThreadId: string } | null>((resolve) => { + new Promise>>((resolve) => { resolveFork = resolve; }), ); @@ -104,14 +104,14 @@ describe("ephemeral thread lifecycle", () => { const opening = lifecycle.open({ sourceThreadId: "source", sourceThreadTitle: "Source" }); await Promise.resolve(); await lifecycle.dispose(); - resolveFork({ sourceThreadId: "source", activation: activationFixture() }); + resolveFork({ kind: "ready", sourceThreadId: "source", activation: activationFixture() }); await expect(opening).resolves.toBe(false); - expect(transport.deleteEphemeralThread).toHaveBeenCalledWith("side"); + expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"); expect(store.getState().activeThread.id).toBeNull(); }); - it("still deletes the side chat when interrupting its running turn fails", async () => { + it("still unsubscribes the side chat when interrupting its running turn fails", async () => { const store = createChatStateStore(); const transport = transportMock(); const lifecycle = createEphemeralThreadLifecycle({ @@ -127,7 +127,7 @@ describe("ephemeral thread lifecycle", () => { await lifecycle.dispose(); - expect(transport.deleteEphemeralThread).toHaveBeenCalledWith("side"); + expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"); }); it("continues close cleanup when interrupting a running side turn does not settle", async () => { @@ -150,16 +150,16 @@ describe("ephemeral thread lifecycle", () => { await vi.advanceTimersByTimeAsync(1_000); await disposal; - expect(transport.deleteEphemeralThread).toHaveBeenCalledWith("side"); + expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"); } finally { vi.useRealTimers(); } }); - it("keeps the side chat active when deletion fails before navigation", async () => { + it("keeps the side chat active when unsubscribe fails before navigation", async () => { const store = createChatStateStore(); const transport = transportMock(); - transport.deleteEphemeralThread = vi.fn().mockResolvedValue(false); + transport.unsubscribeEphemeralThread = vi.fn().mockResolvedValue(false); const addSystemMessage = vi.fn(); const lifecycle = createEphemeralThreadLifecycle({ stateStore: store, @@ -176,12 +176,33 @@ describe("ephemeral thread lifecycle", () => { expect(store.getState().activeThread.id).toBe("side"); expect(addSystemMessage).toHaveBeenCalledWith("Could not discard the side chat. Try again before switching threads."); }); + + it("retries cleanup-required forks when the side view is disposed", async () => { + const store = createChatStateStore(); + const transport = transportMock(); + transport.forkEphemeralThread = vi.fn().mockResolvedValue({ kind: "cleanup-required", threadId: "side" }); + const addSystemMessage = vi.fn(); + const lifecycle = createEphemeralThreadLifecycle({ + stateStore: store, + transport, + ensureConnected: vi.fn().mockResolvedValue(true), + addSystemMessage, + notifyActiveThreadIdentityChanged: vi.fn(), + interruptTurn: vi.fn().mockResolvedValue(true), + }); + + await expect(lifecycle.open({ sourceThreadId: "source", sourceThreadTitle: null })).resolves.toBe(false); + await lifecycle.dispose(); + + expect(addSystemMessage).toHaveBeenCalledWith("Could not prepare the side chat. Cleanup will be retried when this view closes."); + expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"); + }); }); function transportMock(): EphemeralThreadTransport { return { - forkEphemeralThread: vi.fn().mockResolvedValue({ sourceThreadId: "source", activation: activationFixture() }), - deleteEphemeralThread: vi.fn().mockResolvedValue(true), + forkEphemeralThread: vi.fn().mockResolvedValue({ kind: "ready", sourceThreadId: "source", activation: activationFixture() }), + unsubscribeEphemeralThread: vi.fn().mockResolvedValue(true), }; }