fix(server): show reused plugin children as running again (#8080)

* fix(server): report reused plugin children as running again

Child turn starts were discarded after attachment, leaving the previous completed status visible during later work. Translate started to running for live events and ordered history replay.

Co-authored-by: yywl <yywl@linux.0392yy.cn>

* test(server): wait for Git observation before completing fetch

The structural fallback test advanced its debounce timer before asynchronous watcher setup and fetch completion. Wait for those stages so slow filesystem setup does not turn the intended refresh assertion into a scheduling race.

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>
Co-authored-by: yywl <yywl@linux.0392yy.cn>

packages/server/src/server/agent/plugin-provider.test.ts

@@ -486,6 +486,137 @@ describe("PluginAgentClientRegistry", () => {
     }
   });
 
+  test.each(["completed", "canceled", "failed"] as const)(
+    "reports a reused plugin child as running until its next turn is %s",
+    async (outcome) => {
+      const logger = createTestLogger();
+      const harness = createProviderHarness();
+      const plugins = new PluginAgentClientRegistry(logger);
+      plugins.replace([harness.registration]);
+      const registry = buildProviderRegistry(logger, { pluginProviders: plugins.definitions() });
+      const manager = new AgentManager({
+        logger,
+        clients: { "plugin-direct": registry["plugin-direct"].createClient(logger) },
+        providerDefinitions: { "plugin-direct": registry["plugin-direct"] },
+      });
+      try {
+        const agent = await manager.createAgent(
+          { provider: "plugin-direct", cwd: tmpdir() },
+          undefined,
+          { workspaceId: undefined },
+        );
+        const open = harness.inputs.find((input) => input.type === "session.open")!;
+        harness.emit({
+          type: "session.opened",
+          sessionId: "reusable-child",
+          parentSessionId: open.sessionId,
+          capabilities: [],
+          restoration: "parent",
+          title: "Plugin child",
+          cwd: tmpdir(),
+        });
+        harness.emit({
+          type: "session.turn",
+          sessionId: "reusable-child",
+          turnId: "first-turn",
+          state: "started",
+        });
+        harness.emit({
+          type: "session.turn",
+          sessionId: "reusable-child",
+          turnId: "first-turn",
+          state: "completed",
+        });
+        await nextTurn();
+        const [child] = manager.listProviderSubagents(agent.id);
+        expect(child).toMatchObject({
+          id: "reusable-child",
+          title: "Plugin child",
+          status: "completed",
+        });
+
+        harness.emit({
+          type: "session.turn",
+          sessionId: "reusable-child",
+          turnId: "second-turn",
+          state: "started",
+        });
+        await nextTurn();
+        expect(manager.listProviderSubagents(agent.id)).toMatchObject([
+          { id: child.id, title: child.title, status: "running" },
+        ]);
+
+        harness.emit({
+          type: "session.turn",
+          sessionId: "reusable-child",
+          turnId: "second-turn",
+          state: outcome,
+        });
+        await nextTurn();
+        expect(manager.listProviderSubagents(agent.id)).toMatchObject([
+          { id: child.id, title: child.title, status: outcome },
+        ]);
+      } finally {
+        await Promise.all(manager.listAgents().map((agent) => manager.closeAgent(agent.id)));
+        await plugins.shutdown();
+      }
+    },
+  );
+
+  test.each(["started", "completed", "canceled", "failed"] as const)(
+    "replays a reused plugin child's latest turn state: %s",
+    async (state) => {
+      const harness = createProviderHarness({
+        openChildren(rootSessionId, emit) {
+          emit({
+            type: "session.opened",
+            sessionId: "reusable-child",
+            parentSessionId: rootSessionId,
+            capabilities: [],
+            restoration: "parent",
+            cwd: "/workspace",
+          });
+          for (const [turnId, turnState] of [
+            ["first-turn", "started"],
+            ["first-turn", "completed"],
+            ["second-turn", "started"],
+          ] as const) {
+            emit({ type: "session.turn", sessionId: "reusable-child", turnId, state: turnState });
+          }
+          if (state !== "started") {
+            emit({
+              type: "session.turn",
+              sessionId: "reusable-child",
+              turnId: "second-turn",
+              state,
+            });
+          }
+        },
+      });
+      const registry = new PluginAgentClientRegistry(createTestLogger());
+      registry.replace([harness.registration]);
+      try {
+        const session = await registry.clients()[harness.registration.id]!.createSession({
+          provider: harness.registration.id,
+          cwd: "/workspace",
+        });
+        const statuses = [];
+        for await (const event of session.streamHistory()) {
+          if (
+            event.type === "provider_subagent" &&
+            event.event.type === "upsert" &&
+            event.event.id === "reusable-child"
+          ) {
+            statuses.push(event.event.status);
+          }
+        }
+        expect(statuses.at(-1)).toBe(state === "started" ? "running" : state);
+      } finally {
+        await registry.shutdown();
+      }
+    },
+  );
+
   test("gates persistence operations on negotiated provider capabilities", async () => {
     const harness = createProviderHarness({ capabilities: ["session.persistence"] });
     const registry = new PluginAgentClientRegistry(createTestLogger());

packages/server/src/server/agent/plugin-provider.ts

@@ -1581,11 +1581,15 @@ class PluginAgentSession implements AgentSession {
       }
       return;
     }
-    if (event.type === "session.turn" && event.state !== "started") {
+    if (event.type === "session.turn") {
       this.publish({
         type: "provider_subagent",
         provider: this.provider,
-        event: { type: "upsert", id: childId, status: event.state },
+        event: {
+          type: "upsert",
+          id: childId,
+          status: event.state === "started" ? "running" : event.state,
+        },
       });
       return;
     }

packages/server/src/server/workspace-git-service.test.ts

@@ -1105,12 +1105,23 @@ describe("WorkspaceGitServiceImpl", () => {
       { cwd: join(REPO_CWD, "packages", "server") },
       vi.fn(),
     );
-    await vi.waitFor(() => {
-      expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2);
-    });
+    await vi.waitFor(
+      () => {
+        expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(2);
+        expect(service.getMetrics()).toMatchObject({
+          repositoryWorkspaceLinkCount: 2,
+          workspaceObservationSetupInFlightCount: 0,
+          workspaceRefreshInFlightCount: 0,
+          fetchInFlightCount: 1,
+        });
+      },
+      { timeout: 10_000 },
+    );
 
     releaseFetch.resolve();
-    await flushPromises();
+    await vi.waitFor(() => {
+      expect(service.getMetrics().fetchInFlightCount).toBe(0);
+    });
     await vi.advanceTimersByTimeAsync(1_000);
     await vi.waitFor(() => {
       expect(getCheckoutSnapshotFacts).toHaveBeenCalledTimes(4);