diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index f903f457af17f..121738480144d 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -411,10 +411,140 @@ describe("MuseAdapterV2", () => { }, }, }); + assert.strictEqual(fake.closeCount(), 0); } }).pipe(Effect.scoped, Effect.provide(testLayer)), ); + it.effect("resumes on a fresh host when Muse replays queued work instead of starting MCP", () => + Effect.gen(function* () { + yield* (yield* McpProviderSessions.McpProviderSessions).set({ + environmentId: EnvironmentId.make("test-environment"), + threadId: THREAD_ID, + providerSessionId: "test-session", + providerInstanceId: INSTANCE_ID, + endpoint: "http://127.0.0.1:43210/mcp", + authorizationHeader: "Bearer thread-scoped-test-token", + browserToolsAvailable: true, + }); + const fake = yield* makeFakeMuse("first"); + const recovery = yield* makeFakeMuse("second"); + for (const muse of [fake, recovery]) + muse.host.initializeResult.grantedCapabilities.push("sessionMcp"); + const ack = pendingPromise>(); + fake.queueResponse("session/resume", ack.promise); + const replayed = yield* Deferred.make(); + const opening = yield* makeHarness( + fake, + INSTANCE_ID, + "saved-session", + recovery, + undefined, + undefined, + { + nativeEventLogger: { + filePath: "/tmp/provider-native.ndjson", + write: (event) => + typeof event === "object" && + event !== null && + "method" in event && + event.method === "turn/started" + ? Deferred.succeed(replayed, undefined).pipe(Effect.asVoid) + : Effect.void, + close: () => Effect.void, + }, + }, + ).pipe(Effect.forkScoped); + const first = yield* fake.takeCall("session/resume"); + // Muse announces the replayed turn before it acknowledges the resume. + yield* fake.emit("turn/started", { turnId: "replayed-turn" }); + yield* Deferred.await(replayed); + ack.resolve({ + session: { sessionId: "saved-session", modelId: MODEL, activeTurnId: "replayed-turn" }, + history: { items: [] }, + }); + const harness = yield* Fiber.join(opening); + assert.strictEqual(fake.closeCount(), 1); + const again = yield* recovery.takeCall("session/resume"); + assert.strictEqual(again.params.sessionId, "saved-session"); + assert.deepStrictEqual(again.params.config, first.params.config); + assert.isFalse(yield* harness.runtime.hasPendingBackgroundWork!); + const { nativeId } = yield* startConversation(harness, recovery); + assert.match(nativeId, /^second-/); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + + it.effect.each(["no scope unit", "a scope unit"] as const)( + "relaunches past a replaying host whose close never settles, with %s", + (scopeUnit) => + Effect.gen(function* () { + yield* (yield* McpProviderSessions.McpProviderSessions).set({ + environmentId: EnvironmentId.make("test-environment"), + threadId: THREAD_ID, + providerSessionId: "test-session", + providerInstanceId: INSTANCE_ID, + endpoint: "http://127.0.0.1:43210/mcp", + authorizationHeader: "Bearer thread-scoped-test-token", + browserToolsAvailable: true, + }); + const fake = yield* makeFakeMuse("first"); + const recovery = yield* makeFakeMuse("second"); + for (const muse of [fake, recovery]) + muse.host.initializeResult.grantedCapabilities.push("sessionMcp"); + // A replayed background command inherits Muse's pipes and outlives it. + fake.holdClose(new Promise(() => {})); + fake.queueResponse("session/resume", { + session: { sessionId: "saved-session", modelId: MODEL, activeTurnId: "replayed-turn" }, + history: { items: [] }, + }); + const stopped: Array<{ unit: string; resumes: number }> = []; + yield* Effect.gen(function* () { + const opening = yield* makeHarness(fake, INSTANCE_ID, "saved-session", recovery).pipe( + Effect.forkScoped, + ); + yield* Effect.promise(() => fake.closing); + yield* TestClock.adjust("13 seconds"); + const harness = yield* Fiber.join(opening); + assert.deepStrictEqual( + stopped, + scopeUnit === "a scope unit" ? [{ unit: "t3code-muse-replay.scope", resumes: 0 }] : [], + ); + assert.strictEqual( + (yield* recovery.takeCall("session/resume")).params.sessionId, + "saved-session", + ); + + const { nativeId, providerTurn } = yield* startConversation(harness, recovery); + const stopping = yield* harness.runtime + .interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: providerTurn.id, + }) + .pipe(Effect.forkChild({ startImmediately: true })); + assert.strictEqual((yield* recovery.takeCall("turn/interrupt")).params.turnId, nativeId); + yield* recovery.emit("turn/completed", { turnId: nativeId, terminal: "cancelled" }); + yield* Fiber.join(stopping); + assert.strictEqual((yield* harness.takeEvent("turn.terminal")).status, "interrupted"); + assert.strictEqual(fake.closeCount(), 1); + }).pipe( + Effect.scoped, + Effect.provideService(AgentScope, { + ...AgentScope.defaultValue(), + ...(scopeUnit === "a scope unit" + ? { + wrap: ({ command, args }) => + Effect.succeed({ command, args, unit: "t3code-muse-replay.scope" }), + stop: (unit) => + Effect.sync(() => { + stopped.push({ unit, resumes: recovery.calls.length }); + }), + } + : {}), + }), + ); + }).pipe(Effect.provide(testLayer)), + ); + it.effect("selects the runtime mode's permission profile on native start and resume", () => Effect.gen(function* () { for (const [runtimeMode, profile, approvalMode] of [ @@ -857,6 +987,7 @@ describe("MuseAdapterV2", () => { sessionId: "unexpected-session", initialNativeThreadId: undefined, history: { items: [] }, + activeTurnId: null, }, { reason: "an unavailable projection", @@ -864,20 +995,45 @@ describe("MuseAdapterV2", () => { sessionId: "saved-session", initialNativeThreadId: "saved-session", history: { items: null, mode: "none", noneReason: "projectionUnavailable" }, + activeTurnId: null, + }, + { + reason: "an unavailable projection while replaying queued work", + method: "session/resume", + sessionId: "saved-session", + initialNativeThreadId: "saved-session", + history: { items: null, mode: "none", noneReason: "projectionUnavailable" }, + activeTurnId: "replayed-turn", }, ])( "closes a newly opened host when native session registration returns $reason", - ({ method, sessionId, initialNativeThreadId, history }) => + ({ method, sessionId, initialNativeThreadId, history, activeTurnId }) => Effect.gen(function* () { const fake = yield* makeFakeMuse(); + const recovery = yield* makeFakeMuse("second"); + if (activeTurnId) { + yield* (yield* McpProviderSessions.McpProviderSessions).set({ + environmentId: EnvironmentId.make("test-environment"), + threadId: THREAD_ID, + providerSessionId: "test-session", + providerInstanceId: INSTANCE_ID, + endpoint: "http://127.0.0.1:43210/mcp", + authorizationHeader: "Bearer thread-scoped-test-token", + browserToolsAvailable: true, + }); + for (const muse of [fake, recovery]) + muse.host.initializeResult.grantedCapabilities.push("sessionMcp"); + } fake.queueResponse(method, { - session: { sessionId, modelId: MODEL, activeTurnId: null }, + session: { sessionId, modelId: MODEL, activeTurnId }, history, }); - const result = yield* makeHarness(fake, INSTANCE_ID, initialNativeThreadId).pipe( + const result = yield* makeHarness(fake, INSTANCE_ID, initialNativeThreadId, recovery).pipe( Effect.exit, ); assert.strictEqual(result._tag, "Failure"); + // A session Muse cannot stream fails before a relaunch cancels its replayed work. + assert.deepStrictEqual(recovery.calls, []); yield* Effect.promise(() => fake.host.exited); assert.strictEqual(fake.closeCount(), 1); assert.strictEqual(fake.calls.filter((call) => call.method === "turn/start").length, 0); diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index 82e1adcb42277..ef52e7238d11e 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -32,6 +32,7 @@ import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; @@ -341,6 +342,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const commands = yield* Semaphore.make(1); const eventPermit = yield* Semaphore.make(1); let host: MuseSdkHost; + let hostScope: Scope.Closeable; let hostEpoch = 0; let closed = false; let broken = false; @@ -1764,6 +1766,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ); const launchHost = Effect.fnUntraced(function* () { const epoch = ++hostEpoch; + const owned = yield* Scope.fork(scope); const mcpSession = yield* mcpSessions.read(input.threadId); const environment = McpProviderSession.withAgentDeviceEnvironment( options.environment, @@ -1780,10 +1783,10 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const scopeUnit = launch.unit; const stopScope = scopeUnit === undefined ? Effect.void : yield* Effect.cached(agentScope.stop(scopeUnit)); - yield* Scope.addFinalizer(scope, stopScope); + yield* Scope.addFinalizer(owned, stopScope); const processExited = yield* Deferred.make(); // Observe raw exit before acquisition: inherited stdout can also block startup cleanup. - yield* Deferred.await(processExited).pipe(Effect.andThen(stopScope), Effect.forkIn(scope)); + yield* Deferred.await(processExited).pipe(Effect.andThen(stopScope), Effect.forkIn(owned)); const created = yield* Effect.acquireRelease( createMuseSdkHostEffect( { @@ -1809,25 +1812,22 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ), (created) => Effect.gen(function* () { - if (host === created) { + if (epoch === hostEpoch) { closed = true; hostEpoch++; } - const close = Effect.tryPromise({ + // SDK shutdown sends SIGKILL after 10s + 2s, but inherited pipes can + // keep its close promise pending. Stop waiting then; the scope unit, + // when there is one, reaps those holders. + yield* Effect.tryPromise({ try: () => created.close(), catch: (cause) => new MuseSdkError({ operation: "close", cause }), - }); - // SDK shutdown sends SIGKILL after 10s + 2s, but inherited pipes can - // keep its close promise pending. Let the scope reap those holders. - yield* ( - scopeUnit === undefined - ? close - : close.pipe(Effect.interruptible, Effect.timeoutOption("13 seconds")) - ).pipe(Effect.ignore); + }).pipe(Effect.interruptible, Effect.timeoutOption("13 seconds"), Effect.ignore); }), { interruptible: true }, - ).pipe(Effect.provideService(Scope.Scope, scope)); + ).pipe(Effect.provideService(Scope.Scope, owned)); host = created; + hostScope = owned; created.connection.onNotification((notification) => { Queue.offerUnsafe(inbox, { type: "notification", @@ -1870,7 +1870,7 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( void created.exited.then(reportExit); yield* Deferred.await(processExited).pipe( Effect.tap((exit) => Effect.sync(() => reportExit(exit))), - Effect.forkIn(scope), + Effect.forkIn(owned), ); }); yield* Effect.addFinalizer(() => @@ -1921,10 +1921,6 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( return yield* Effect.gen(function* () { nativeSessionId = requestedId ?? host.connection.mintCommandId(); const mcpSession = yield* mcpSessions.read(args.threadId); - if (mcpSession && !host.initializeResult.grantedCapabilities.includes("sessionMcp")) - return yield* protocolError( - "Update Muse Code to a version that supports session MCP servers", - ); const config = mcpSession ? { mcpServers: { @@ -1940,34 +1936,67 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( : undefined; const startModel = yield* resolveModel(args.modelSelection); const permissionProfile = musePermissionProfile(args.runtimePolicy.runtimeMode); - const result = yield* request( - requestedId ? "session/resume" : "session/start", - requestedId - ? { excludeItems: true, ...(config ? { config } : {}), permissionProfile } - : { - workspaceRoot: cwd, - ...(config ? { config } : {}), - ...(startModel ? { modelId: startModel } : {}), - providerId: "meta", - approvalMode: museApprovalMode(args.runtimePolicy.runtimeMode), - permissionProfile, - }, - ).pipe( - Effect.catch((error) => { - missingNativeSession = - requestedId !== undefined && - error.payload instanceof MspError && - (error.payload.kind === "sessionNotFound" || error.payload.kind === "notFound"); - return Effect.fail(error); - }), - Effect.flatMap((value) => decode(MuseSessionResult, value)), - ); - if (result.session.sessionId !== nativeSessionId) - return yield* protocolError("Muse returned a different session identity"); - if (result.history?.noneReason === "projectionUnavailable") - return yield* protocolError( - "Muse cannot stream this session because its live view is unavailable", + const open = Effect.gen(function* () { + if (mcpSession && !host.initializeResult.grantedCapabilities.includes("sessionMcp")) + return yield* protocolError( + "Update Muse Code to a version that supports session MCP servers", + ); + const result = yield* request( + requestedId ? "session/resume" : "session/start", + requestedId + ? { excludeItems: true, ...(config ? { config } : {}), permissionProfile } + : { + workspaceRoot: cwd, + ...(config ? { config } : {}), + ...(startModel ? { modelId: startModel } : {}), + providerId: "meta", + approvalMode: museApprovalMode(args.runtimePolicy.runtimeMode), + permissionProfile, + }, + ).pipe( + Effect.catch((error) => { + missingNativeSession = + requestedId !== undefined && + error.payload instanceof MspError && + (error.payload.kind === "sessionNotFound" || error.payload.kind === "notFound"); + return Effect.fail(error); + }), + Effect.flatMap((value) => decode(MuseSessionResult, value)), ); + if (result.session.sessionId !== nativeSessionId) + return yield* protocolError("Muse returned a different session identity"); + if (result.history?.noneReason === "projectionUnavailable") + return yield* protocolError( + "Muse cannot stream this session because its live view is unavailable", + ); + return result; + }); + let result = yield* open; + // Muse Code 1.4.12 never starts session MCP servers on a host whose resume replays + // queued work, and MSP cannot reload them. Closing the host cancels one replayed + // item, so resume on fresh hosts until one comes back idle. + if (config) + for (let relaunch = 0; result.session.activeTurnId; relaunch++) { + if (relaunch === 5) { + yield* Effect.logWarning( + "Muse kept replaying queued work; T3 tools are unavailable", + ); + break; + } + yield* Effect.logWarning("Muse replayed queued work without T3 tools; relaunching"); + const replaying = hostScope; + yield* eventPermit.withPermits(1)( + Effect.sync(() => { + hostEpoch++; + wake = undefined; + }), + ); + // Muse leases a session to one live host, so the replacement waits for + // this host's bounded shutdown before it resumes. + yield* Scope.close(replaying, Exit.void); + yield* launchHost(); + result = yield* open; + } if (requestedId) yield* request("session/setApprovalMode", { mode: museApprovalMode(args.runtimePolicy.runtimeMode),