test(managed-agent): cover coordinator-only restart (#14155)

Co-authored-by: Qwen-Coder <[email protected]>

integration-tests/helpers/hosted-coordinator-restart-driver.ts

@@ -0,0 +1,188 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import assert from 'node:assert/strict';
+import { readFile, rename, writeFile } from 'node:fs/promises';
+import {
+  createServer,
+  request as createRequest,
+  type ServerResponse,
+} from 'node:http';
+import { HostedHarnessProcess } from './hosted-harness-process.js';
+
+const configPath = process.argv[2];
+const config = JSON.parse(await readFile(configPath, 'utf8')) as {
+  readyPath: string;
+};
+
+let primaryRequests = 0;
+let followupRequests = 0;
+let loadResponses = 0;
+let firstChunk = false;
+let released = false;
+let releasePrimary!: () => void;
+const primaryGate = new Promise<void>((resolve) => {
+  releasePrimary = resolve;
+});
+let stop!: () => void;
+const stopped = new Promise<void>((resolve) => {
+  stop = resolve;
+});
+
+function writeEvent(response: ServerResponse, value: unknown) {
+  response.write(`data: ${JSON.stringify(value)}\n\n`);
+}
+
+function chunk(content: string, finishReason: string | null = null) {
+  return {
+    id: 'coordinator-restart-fixture',
+    object: 'chat.completion.chunk',
+    created: 0,
+    model: 'hosted-fixture',
+    choices: [
+      {
+        index: 0,
+        delta: content ? { role: 'assistant', content } : {},
+        finish_reason: finishReason,
+      },
+    ],
+  };
+}
+
+const model = createServer(async (request, response) => {
+  try {
+    assert.equal(request.method, 'POST');
+    assert.equal(request.url, '/v1/chat/completions');
+    const body: Buffer[] = [];
+    for await (const part of request) body.push(Buffer.from(part));
+    const parsed = JSON.parse(Buffer.concat(body).toString()) as {
+      messages?: unknown;
+    };
+    const history = JSON.stringify(parsed.messages ?? []);
+    const followup = history.includes('FOLLOW_UP');
+    if (followup) followupRequests++;
+    else primaryRequests++;
+    assert.equal(followup ? followupRequests : primaryRequests, 1, history);
+
+    response.writeHead(200, {
+      'Content-Type': 'text/event-stream',
+      'Cache-Control': 'no-cache',
+      Connection: 'keep-alive',
+    });
+    if (followup) {
+      writeEvent(response, chunk('FOLLOW_UP_OK'));
+    } else {
+      writeEvent(response, chunk('RESTART_'));
+      firstChunk = true;
+      await primaryGate;
+      writeEvent(response, chunk('OK'));
+    }
+    writeEvent(response, chunk('', 'stop'));
+    response.end('data: [DONE]\n\n');
+  } catch (error) {
+    response.destroy(error as Error);
+  }
+});
+await new Promise<void>((resolve) => model.listen(0, '127.0.0.1', resolve));
+const modelAddress = model.address();
+assert(modelAddress && typeof modelAddress !== 'string');
+
+const harness = new HostedHarnessProcess();
+let harnessProxy: ReturnType<typeof createServer> | undefined;
+let control: ReturnType<typeof createServer> | undefined;
+try {
+  await harness.start(`http://127.0.0.1:${modelAddress.port}/v1`);
+  harnessProxy = createServer((request, response) => {
+    const target = new URL(request.url ?? '/', harness.baseUrl);
+    const upstream = createRequest(
+      target,
+      {
+        method: request.method,
+        headers: { ...request.headers, host: target.host },
+      },
+      (upstreamResponse) => {
+        if (
+          request.method === 'POST' &&
+          target.pathname.endsWith('/load') &&
+          upstreamResponse.statusCode === 200
+        ) {
+          loadResponses++;
+        }
+        response.writeHead(
+          upstreamResponse.statusCode ?? 502,
+          upstreamResponse.headers,
+        );
+        upstreamResponse.pipe(response);
+      },
+    );
+    upstream.on('error', (error) => response.destroy(error));
+    request.pipe(upstream);
+  });
+  await new Promise<void>((resolve) =>
+    harnessProxy!.listen(0, '127.0.0.1', resolve),
+  );
+  const harnessProxyAddress = harnessProxy.address();
+  assert(harnessProxyAddress && typeof harnessProxyAddress !== 'string');
+  control = createServer((request, response) => {
+    const state = {
+      primaryRequests,
+      followupRequests,
+      loadResponses,
+      firstChunk,
+      released,
+      harnessAlive:
+        harness.child?.exitCode === null && harness.child.signalCode === null,
+    };
+    response.setHeader('Content-Type', 'application/json');
+    if (request.method === 'GET' && request.url === '/state') {
+      response.end(JSON.stringify(state));
+      return;
+    }
+    if (request.method === 'POST' && request.url === '/release') {
+      if (!released) {
+        released = true;
+        releasePrimary();
+      }
+      response.end(JSON.stringify({ ...state, released: true }));
+      return;
+    }
+    if (request.method === 'POST' && request.url === '/stop') {
+      if (!released) releasePrimary();
+      response.end(JSON.stringify(state));
+      stop();
+      return;
+    }
+    response.statusCode = 404;
+    response.end(JSON.stringify({ error: 'not_found' }));
+  });
+  await new Promise<void>((resolve) =>
+    control!.listen(0, '127.0.0.1', resolve),
+  );
+  const controlAddress = control.address();
+  assert(controlAddress && typeof controlAddress !== 'string');
+  const temporary = `${config.readyPath}.tmp`;
+  await writeFile(
+    temporary,
+    JSON.stringify({
+      harnessUrl: `http://127.0.0.1:${harnessProxyAddress.port}`,
+      controlUrl: `http://127.0.0.1:${controlAddress.port}`,
+    }),
+  );
+  await rename(temporary, config.readyPath);
+  await stopped;
+} finally {
+  if (harnessProxy?.listening) {
+    harnessProxy.closeAllConnections();
+    await new Promise<void>((resolve) => harnessProxy!.close(() => resolve()));
+  }
+  await harness.close();
+  model.closeAllConnections();
+  await new Promise<void>((resolve) => model.close(() => resolve()));
+  if (control?.listening) {
+    control.closeAllConnections();
+    await new Promise<void>((resolve) => control!.close(() => resolve()));
+  }
+}

packages/cli/src/serve/hosted-harness-session.test.ts

@@ -1272,6 +1272,84 @@ describe('Hosted Harness no-tool session', () => {
     await rm(state.root, { recursive: true, force: true });
   });
 
+  it('re-answers a drive takeover while the resident model turn is active', async () => {
+    let releaseModel!: () => void;
+    let markFirstChunk!: () => void;
+    const modelHeld = new Promise<void>((resolve) => {
+      releaseModel = resolve;
+    });
+    const firstChunk = new Promise<void>((resolve) => {
+      markFirstChunk = resolve;
+    });
+    const text = 'COORDINATOR_RECOVERED';
+    state.model.mockImplementationOnce(async ({ textDeltas }) => {
+      await textDeltas!.delta('COORDINATOR_');
+      markFirstChunk();
+      await modelHeld;
+      await textDeltas!.delta('RECOVERED');
+      return { text, model: 'test-model' };
+    });
+
+    const server = await app();
+    const created = await headers(supertest(server).post('/session')).send({
+      sessionId: SESSION_ID,
+      sessionScope: 'thread',
+      managedSessionStore: store(),
+    });
+    expect(created.status).toBe(200);
+    const prompt = [{ type: 'text', text: 'COORDINATOR_ONLY' }];
+    await headers(supertest(server).post(`/session/${SESSION_ID}/prompt`))
+      .set('X-Qwen-Client-Id', created.body.clientId as string)
+      .send({
+        prompt,
+        promptId: PROMPT_ID,
+        payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(prompt)).digest('hex')}`,
+      })
+      .expect(202);
+    await firstChunk;
+
+    const load = () =>
+      headers(supertest(server).post(`/session/${SESSION_ID}/load`)).send({
+        managedSessionStore: store(),
+        driveRuntimeRecovery: true,
+      });
+    const takeover = await load();
+    expect(takeover.status).toBe(200);
+    expect(takeover.body.clientId).toBe(created.body.clientId);
+    const redriven = await load();
+    expect(redriven.status).toBe(200);
+    expect(redriven.body).toMatchObject({
+      clientId: takeover.body.clientId,
+      eventEpoch: takeover.body.eventEpoch,
+      lastEventId: takeover.body.lastEventId,
+    });
+    expect(state.model).toHaveBeenCalledTimes(1);
+
+    releaseModel();
+    await vi.waitFor(
+      async () => {
+        const status = await headers(
+          supertest(server).get(`/session/${SESSION_ID}/status`),
+        ).set('X-Qwen-Client-Id', created.body.clientId as string);
+        expect(status.body.hasActivePrompt).toBe(false);
+      },
+      { timeout: 10_000 },
+    );
+    const transcript = await headers(
+      supertest(server).get(`/session/${SESSION_ID}/transcript?limit=256`),
+    ).set('X-Qwen-Client-Id', created.body.clientId as string);
+    expectFullStreamedAnswer(transcript.body, text);
+    expect(
+      transcript.body.events.filter((event: { type: string }) =>
+        ['turn_complete', 'turn_error'].includes(event.type),
+      ),
+    ).toHaveLength(1);
+    expect(state.model).toHaveBeenCalledTimes(1);
+    await headers(supertest(server).delete(`/session/${SESSION_ID}`)).expect(
+      204,
+    );
+  });
+
   it.each(['close', 'delete'] as const)(
     'settles %s lifecycle Hooks once and retains the attachment until detach',
     async (kind) => {

packages/sdk-java/managed-agent-server/pom.xml

@@ -315,7 +315,7 @@
                 <configuration>
                     <includes><include>**/HostedProcessCrashIT.java</include></includes>
                     <failIfNoTests>true</failIfNoTests>
-                    <forkedProcessTimeoutInSeconds>480</forkedProcessTimeoutInSeconds>
+                    <forkedProcessTimeoutInSeconds>720</forkedProcessTimeoutInSeconds>
                 </configuration>
                 <executions><execution><goals><goal>integration-test</goal><goal>verify</goal></goals></execution></executions>
             </plugin></plugins></build>

packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedProcessCrashFixtureMain.java

@@ -9,6 +9,7 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.StandardCopyOption;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -25,23 +26,41 @@ public final class HostedProcessCrashFixtureMain {
         String cli = args[2];
         String tenant = args[3];
         String fault = args[4];
+        boolean coordinatorRestart = "coordinator-only".equals(fault);
         Path workspace = root.resolve("workspace");
-        Path worker = root.resolve("fixture-worker.mjs");
-        Path cliPath = Path.of(cli).toAbsolutePath().normalize();
         ObjectMapper mapper = new ObjectMapper();
-        Files.writeString(worker, "await import("
-                + mapper.writeValueAsString(cliPath.getParent().getParent()
-                        .resolve("integration-tests/helpers/hosted-file-read-gate.mjs").toUri().toString())
-                + ");\nprocess.argv[1] = " + mapper.writeValueAsString(cliPath.toString())
-                + ";\nawait import(" + mapper.writeValueAsString(cliPath.toUri().toString()) + ");\n");
-        var spring = (ServletWebServerApplicationContext) new SpringApplicationBuilder(
-                ManagedAgentServerApplication.class).run(
-                "--server.address=127.0.0.1", "--server.port=0",
+        List<String> arguments = new ArrayList<>(List.of(
+                "--server.address=127.0.0.1",
                 "--spring.datasource.url=" + System.getenv("FG6C_MYSQL_URL"),
                 "--spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
                 "--spring.datasource.username=" + System.getenv("FG6C_MYSQL_USER"),
                 "--spring.datasource.password=" + System.getenv().getOrDefault("FG6C_MYSQL_PASSWORD", ""),
-                "--qwen.managed-agent.session-store.enabled=true",
+                "--qwen.managed-agent.session-store.enabled=true"));
+        if (coordinatorRestart) {
+            String port = args[5];
+            arguments.addAll(List.of(
+                    "--server.port=" + port,
+                    "--qwen.managed-agent.session-store.base-url=http://127.0.0.1:" + port,
+                    "--qwen.managed-agent.session-store.workspace-id=coordinator-restart",
+                    "--qwen.managed-agent.trusted-actor-header=X-Qwen-Actor-Id",
+                    "--qwen.managed-agent.harness.enabled=true",
+                    "--qwen.managed-agent.harness.base-url=" + args[6],
+                    "--qwen.managed-agent.harness.token=hosted-process-fixture-token",
+                    "--qwen.managed-agent.harness.capability-digest=sha256:" + "a".repeat(64),
+                    "--qwen.managed-agent.runtime-broker.enabled=false",
+                    "--qwen.managed-agent.dispatch.scan-delay=100ms",
+                    "--qwen.managed-agent.dispatch.lease-duration=3s",
+                    "--qwen.managed-agent.dispatch.lease-renew-interval=1s"));
+        } else {
+            Path worker = root.resolve("fixture-worker.mjs");
+            Path cliPath = Path.of(cli).toAbsolutePath().normalize();
+            Files.writeString(worker, "await import("
+                    + mapper.writeValueAsString(cliPath.getParent().getParent()
+                            .resolve("integration-tests/helpers/hosted-file-read-gate.mjs").toUri().toString())
+                    + ");\nprocess.argv[1] = " + mapper.writeValueAsString(cliPath.toString())
+                    + ";\nawait import(" + mapper.writeValueAsString(cliPath.toUri().toString()) + ");\n");
+            arguments.addAll(List.of(
+                "--server.port=0",
                 "--qwen.managed-agent.harness.enabled=false",
                 "--qwen.managed-agent.harness.capability-digest=sha256:" + "a".repeat(64),
                 "--qwen.managed-agent.runtime-broker.enabled=true",
@@ -59,7 +78,18 @@ public final class HostedProcessCrashFixtureMain {
                 "--qwen.managed-agent.runtime-broker.worker-entry=" + worker,
                 "--qwen.managed-agent.runtime-broker.workspace-mounts[0].tenant-id=" + tenant,
                 "--qwen.managed-agent.runtime-broker.workspace-mounts[0].storage-id=storage",
-                "--qwen.managed-agent.runtime-broker.workspace-mounts[0].root=" + workspace);
+                "--qwen.managed-agent.runtime-broker.workspace-mounts[0].root=" + workspace));
+        }
+        var spring = (ServletWebServerApplicationContext) new SpringApplicationBuilder(
+                ManagedAgentServerApplication.class).run(arguments.toArray(String[]::new));
+        if (coordinatorRestart) {
+            Path ready = root.resolve("ready.tmp");
+            mapper.writeValue(ready.toFile(), Map.of(
+                    "storeUrl", "http://127.0.0.1:" + spring.getWebServer().getPort()));
+            Files.move(ready, root.resolve("ready.json"), StandardCopyOption.ATOMIC_MOVE,
+                    StandardCopyOption.REPLACE_EXISTING);
+            return;
+        }
         Path saved = root.resolve("session-id");
         String sessionId;
         if (Files.exists(saved)) {

packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedProcessCrashIT.java

@@ -6,6 +6,11 @@ import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.sun.net.httpserver.HttpServer;
+import java.net.ServerSocket;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -59,6 +64,9 @@ class HostedProcessCrashIT {
         }
         System.out.println("FG6C_DATABASE " + JSON.writeValueAsString(database));
         String selected = System.getProperty("qwen.fg6c.case");
+        if ("coordinator-only".equals(selected)) {
+            return;
+        }
         if (selected != null) {
             assertThat(CASES).contains(selected);
         }
@@ -69,6 +77,256 @@ class HostedProcessCrashIT {
         }
     }
 
+    @Test
+    @Timeout(180)
+    void coordinatorRestartReattachesTheResidentTurnOnMySql() throws Exception {
+        String selected = System.getProperty("qwen.fg6c.case");
+        if (selected != null && !"coordinator-only".equals(selected)) {
+            return;
+        }
+        assertThat(required("mysql.url")).startsWith("jdbc:mysql:");
+        assertThat(System.getProperty("os.name").toLowerCase()).doesNotContain("windows");
+        var source = new DriverManagerDataSource(required("mysql.url"), required("mysql.user"),
+                System.getProperty("mysql.password", ""));
+        Properties timeouts = new Properties();
+        timeouts.setProperty("connectTimeout", "5000");
+        timeouts.setProperty("socketTimeout", "10000");
+        source.setConnectionProperties(timeouts);
+        JdbcTemplate jdbc = new com.alibaba.qwen.code.managedagent.store.ManagedSchemaJdbcTemplate(source);
+        jdbc.setQueryTimeout(10);
+        try (var fixture = new CoordinatorRestartFixture(jdbc)) {
+            fixture.run();
+        }
+    }
+
+    private final class CoordinatorRestartFixture implements AutoCloseable {
+        private final JdbcTemplate jdbc;
+        private final Path root;
+        private final Path cli = Path.of(required("qwen.cli.entry")).toAbsolutePath().normalize();
+        private final String node = required("node.executable");
+        private final String tenant = "hosted-coordinator-" + UUID.randomUUID();
+        private final int port;
+        private final Set<ProcessHandle> processes = new LinkedHashSet<>();
+        private final HttpClient http = HttpClient.newBuilder()
+                .connectTimeout(java.time.Duration.ofSeconds(5)).build();
+        private Process spring;
+        private Process driver;
+        private ObjectNode driverReady;
+        private int boots;
+
+        private CoordinatorRestartFixture(JdbcTemplate jdbc) throws Exception {
+            this.jdbc = jdbc;
+            root = Files.createDirectory(temporary.resolve("coordinator-only")).toRealPath();
+            assertThat(cli).isRegularFile();
+            try (var socket = new ServerSocket(0)) {
+                socket.setReuseAddress(true);
+                port = socket.getLocalPort();
+            }
+        }
+
+        private void run() throws Exception {
+            startDriver();
+            startSpring();
+            JsonNode created = api("POST", "/v1/agents/sessions", "create", """
+                    {"agent_id":"qwen-code","input":[]}
+                    """, 202);
+            String session = created.path("id").asText();
+            JsonNode admitted = api("POST", "/v1/agents/sessions/" + session + "/events", "primary", """
+                    {"type":"agent.session.input.message",
+                     "input":[{"type":"input_text","text":"COORDINATOR_ONLY"}]}
+                    """, 202);
+            String turn = admitted.path("turn_id").asText();
+            await(() -> {
+                JsonNode state = driverState();
+                return state.path("firstChunk").asBoolean()
+                        && state.path("primaryRequests").asInt() == 1;
+            }, 30, "first model chunk");
+            await(() -> jdbc.queryForObject("SELECT COUNT(*) FROM ma_public_event"
+                            + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?"
+                            + " AND event_type = 'item.output_text.delta'", Integer.class,
+                    tenant, session, turn) == 1, 30, "first persisted delta");
+            int loadResponsesBeforeRestart = driverState().path("loadResponses").asInt();
+            String originalOwner = jdbc.queryForObject("SELECT dispatch_owner FROM ma_turn"
+                    + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?", String.class,
+                    tenant, session, turn);
+
+            long killedPid = spring.pid();
+            spring.destroyForcibly();
+            assertThat(spring.waitFor(10, TimeUnit.SECONDS)).isTrue();
+            assertThat(spring.exitValue()).isEqualTo(137);
+            JsonNode during = driverState();
+            assertThat(during.path("harnessAlive").asBoolean()).isTrue();
+            assertThat(during.path("primaryRequests").asInt()).isEqualTo(1);
+            assertThat(during.path("released").asBoolean()).isFalse();
+
+            startSpring();
+            await(() -> {
+                String owner = jdbc.queryForObject("SELECT dispatch_owner FROM ma_turn"
+                                + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?",
+                        String.class, tenant, session, turn);
+                return owner != null && !owner.equals(originalOwner);
+            }, 30, "replacement coordinator claim");
+            await(() -> driverState().path("loadResponses").asInt() > loadResponsesBeforeRestart,
+                    30, "replacement coordinator Harness load");
+            control("POST", "/release");
+            await(() -> "COMPLETED".equals(jdbc.queryForObject("SELECT status FROM ma_turn"
+                            + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?",
+                    String.class, tenant, session, turn)), 45, "recovered Turn completion");
+            assertThat(api("GET", "/v1/agents/sessions/" + session + "/turns/" + turn,
+                    null, null, 200).path("status").asText()).isEqualTo("completed");
+            await(() -> api("GET", "/v1/agents/sessions/" + session + "/items?limit=100",
+                    null, null, 200).toString().contains("RESTART_OK"), 15,
+                    "materialized recovered output");
+
+            JsonNode state = driverState();
+            assertThat(state.path("primaryRequests").asInt()).isEqualTo(1);
+            assertThat(state.path("harnessAlive").asBoolean()).isTrue();
+            List<String> deltas = jdbc.queryForList("SELECT data_json FROM ma_public_event"
+                            + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?"
+                            + " AND event_type = 'item.output_text.delta' ORDER BY sequence_id",
+                    String.class, tenant, session, turn);
+            String output = deltas.stream().map(value -> {
+                try {
+                    return JSON.readTree(value).path("text").asText();
+                } catch (Exception error) {
+                    throw new IllegalStateException(error);
+                }
+            }).reduce("", String::concat);
+            assertThat(output).isEqualTo("RESTART_OK");
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_public_event"
+                            + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?"
+                            + " AND event_type IN ('turn.completed','turn.failed','turn.cancelled')",
+                    Integer.class, tenant, session, turn)).isEqualTo(1);
+            assertThat(api("GET", "/v1/agents/sessions/" + session + "/events", null,
+                    null, 200).path("data").toString()).contains("RESTART_").contains("OK")
+                    .contains("turn.completed");
+
+            JsonNode followup = api("POST", "/v1/agents/sessions/" + session + "/events", "follow-up", """
+                    {"type":"agent.session.input.message",
+                     "input":[{"type":"input_text","text":"FOLLOW_UP"}]}
+                    """, 202);
+            String followupTurn = followup.path("turn_id").asText();
+            await(() -> "COMPLETED".equals(jdbc.queryForObject("SELECT status FROM ma_turn"
+                            + " WHERE tenant_id = ? AND session_id = ? AND turn_id = ?",
+                    String.class, tenant, session, followupTurn)), 30, "following Turn completion");
+            await(() -> api("GET", "/v1/agents/sessions/" + session + "/items?limit=100",
+                    null, null, 200).toString().contains("FOLLOW_UP_OK"), 15,
+                    "materialized following output");
+            state = driverState();
+            assertThat(state.path("primaryRequests").asInt()).isEqualTo(1);
+            assertThat(state.path("followupRequests").asInt()).isEqualTo(1);
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_turn WHERE tenant_id = ?"
+                            + " AND session_id = ? AND status = 'COMPLETED'", Integer.class,
+                    tenant, session)).isEqualTo(2);
+            System.out.printf("ISSUE13327 coordinatorPid=%d replacementPid=%d session=%s turn=%s"
+                            + " primaryModelRequests=1 output=%s terminal=1 followup=COMPLETED%n",
+                    killedPid, spring.pid(), session, turn, output);
+        }
+
+        private void startDriver() throws Exception {
+            Path config = root.resolve("coordinator-driver.json");
+            Path ready = root.resolve("coordinator-driver-ready.json");
+            JSON.writeValue(config.toFile(), Map.of("readyPath", ready.toString()));
+            Path log = root.resolve("coordinator-driver.log");
+            driver = new ProcessBuilder(node, "--import", "tsx",
+                    "integration-tests/helpers/hosted-coordinator-restart-driver.ts", config.toString())
+                    .directory(cli.getParent().getParent().toFile()).redirectErrorStream(true)
+                    .redirectOutput(log.toFile()).start();
+            processes.add(driver.toHandle());
+            await(() -> Files.exists(ready) || !driver.isAlive(), 30, "coordinator driver readiness");
+            assertThat(driver.isAlive()).as("Driver failed: %s", Files.readString(log)).isTrue();
+            driverReady = (ObjectNode) JSON.readTree(Files.readString(ready));
+        }
+
+        private void startSpring() throws Exception {
+            Path readiness = root.resolve("ready.json");
+            Files.deleteIfExists(readiness);
+            String classpath = System.getProperty("surefire.test.class.path", System.getProperty("java.class.path"));
+            Path log = root.resolve("coordinator-spring-" + ++boots + ".log");
+            ProcessBuilder builder = new ProcessBuilder(Path.of(System.getProperty("java.home"), "bin", "java").toString(),
+                    "-cp", classpath, HostedProcessCrashFixtureMain.class.getName(), root.toString(), node,
+                    cli.toString(), tenant, "coordinator-only", Integer.toString(port),
+                    driverReady.path("harnessUrl").asText()).redirectErrorStream(true).redirectOutput(log.toFile());
+            builder.environment().put("FG6C_MYSQL_URL", required("mysql.url"));
+            builder.environment().put("FG6C_MYSQL_USER", required("mysql.user"));
+            builder.environment().put("FG6C_MYSQL_PASSWORD", System.getProperty("mysql.password", ""));
+            spring = builder.start();
+            processes.add(spring.toHandle());
+            await(() -> Files.exists(readiness) || !spring.isAlive(), 30, "coordinator Spring readiness");
+            assertThat(spring.isAlive()).as("Spring failed: %s", Files.readString(log)).isTrue();
+            assertThat(JSON.readTree(Files.readString(readiness)).path("storeUrl").asText())
+                    .isEqualTo("http://127.0.0.1:" + port);
+        }
+
+        private JsonNode api(String method, String route, String idempotencyKey,
+                String body, int expected) {
+            try {
+                HttpRequest.Builder request = HttpRequest.newBuilder(URI.create("http://127.0.0.1:" + port + route))
+                        .timeout(java.time.Duration.ofSeconds(10))
+                        .header("X-Qwen-Tenant-Id", tenant).header("X-Qwen-Actor-Id", "actor")
+                        .header("Accept", "application/json")
+                        .method(method, body == null ? HttpRequest.BodyPublishers.noBody()
+                                : HttpRequest.BodyPublishers.ofString(body));
+                if (body != null) {
+                    request.header("Content-Type", "application/json");
+                }
+                if (idempotencyKey != null) {
+                    request.header("Idempotency-Key", idempotencyKey);
+                }
+                HttpResponse<String> response = http.send(request.build(), HttpResponse.BodyHandlers.ofString());
+                assertThat(response.statusCode()).as("%s %s: %s", method, route, response.body())
+                        .isEqualTo(expected);
+                return JSON.readTree(response.body());
+            } catch (Exception error) {
+                throw new IllegalStateException(error);
+            }
+        }
+
+        private JsonNode driverState() {
+            return control("GET", "/state");
+        }
+
+        private JsonNode control(String method, String route) {
+            try {
+                HttpRequest request = HttpRequest.newBuilder(URI.create(driverReady.path("controlUrl").asText() + route))
+                        .timeout(java.time.Duration.ofSeconds(10))
+                        .method(method, HttpRequest.BodyPublishers.noBody()).build();
+                HttpResponse<String> response = http.send(request, HttpResponse.BodyHandlers.ofString());
+                assertThat(response.statusCode()).as("driver %s %s: %s", method, route, response.body())
+                        .isEqualTo(200);
+                return JSON.readTree(response.body());
+            } catch (Exception error) {
+                throw new IllegalStateException(error);
+            }
+        }
+
+        @Override
+        public void close() throws Exception {
+            if (driver != null && driver.isAlive()) {
+                processes.addAll(driver.descendants().toList());
+                if (driverReady == null) {
+                    driver.destroyForcibly();
+                } else {
+                    try {
+                        control("POST", "/stop");
+                        if (!driver.waitFor(10, TimeUnit.SECONDS)) {
+                            driver.destroyForcibly();
+                        }
+                    } catch (RuntimeException ignored) {
+                        driver.destroyForcibly();
+                    }
+                }
+            }
+            if (spring != null) {
+                processes.addAll(spring.descendants().toList());
+            }
+            processes.forEach(ProcessHandle::destroyForcibly);
+            for (ProcessHandle process : processes) {
+                await(() -> !running(process), 10, "process cleanup " + process.pid());
+            }
+        }
+    }
+
     private final class Fixture implements AutoCloseable {
         private final JdbcTemplate jdbc;
         private final String fault;