fix(muse): a resume that replays queued work keeps T3 tools

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<Record<string, unknown>>();
+      fake.queueResponse("session/resume", ack.promise);
+      const replayed = yield* Deferred.make<void>();
+      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<void>(() => {}));
+        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,27 +987,53 @@ describe("MuseAdapterV2", () => {
       sessionId: "unexpected-session",
       initialNativeThreadId: undefined,
       history: { items: [] },
+      activeTurnId: null,
     },
     {
       reason: "an unavailable projection",
       method: "session/resume",
       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);

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<ProcessExit>();
         // 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),