test(managed-agent): verify operator recovery on real SQL engines (#13759)

docs/verification/12904-operator-recovery/README.md

@@ -0,0 +1,58 @@
+# Hosted Shell operator recovery acceptance
+
+This acceptance for [#12904](https://github.com/QwenLM/qwen-code/issues/12904) exercises the recovery already shipped by [#12977](https://github.com/QwenLM/qwen-code/pull/12977). It does not add another recovery mechanism.
+
+Run the packaged CLI, real worker, Runtime Broker, Spring store and maintenance command on Linux with Java 21, Node 22+, and an isolated MySQL or MariaDB database. Only the model endpoint is deterministic. Build the CLI and bundle first, then install the Java SDK and Broker artifacts as in the SDK Java workflow.
+
+```sh
+mvn --batch-mode --no-transfer-progress \
+  -f packages/sdk-java/managed-agent-server/pom.xml \
+  '-Dtest=HostedWorkspaceToolTurnIT#operatorRecoveryReleasesOnlyTheOriginalQuiescentShellOnMySql' \
+  -Dnode.executable="$(command -v node)" \
+  -Dqwen.cli.entry="$PWD/dist/cli.js" \
+  -Dmysql.url='jdbc:mysql://127.0.0.1:23060/qwen_12904?allowPublicKeyRetrieval=true&useSSL=false' \
+  -Dmysql.user=root -Dmysql.password=qwen-acceptance-only \
+  test
+```
+
+The example credentials belong only to the disposable acceptance database. Repeat against MariaDB using its isolated port and schema. The test fails if Linux, SQL or the packaged bundle is unavailable.
+
+The test observes three independent Workspaces:
+
+- The original compound command retains its Workspace after a `partial` / `producer_lost` result. A second Session receives `workspace_busy`.
+- A second compound command starts a real detached writer. After the original worker dies, the writer must still be alive and its file must keep growing while the exact Workspace holder remains fenced.
+- An independent Workspace can still complete a file operation while the other two are blocked.
+
+Recovery invokes the real maintenance entry point in separate JVMs: inspect the original binding and generation, prepare the exact held lease, stop the fixture's workers and potential writers, and submit a private quiescence statement. Completion is retried with the same evidence file. SQL must show the original lease removed and binding retired; a new Session must actually write a file in the recovered Workspace. The original marker must remain a single `x`, and recovery itself must not call the model again.
+
+Success includes `HOSTED_OPERATOR_RECOVERY_OK` and identity-bearing `OPERATOR_RECOVERY` records. Retain the Surefire report and process output when a check fails. This does not certify arbitrary escaped processes, external restart sources, cross-host recovery, or completion of the original uncertain turn. The operator statement remains the trust boundary described in the [recovery procedure](../../../docs/users/hosted-workspace-recovery.md).
+
+## Recorded acceptance — 2026-10-09
+
+The production baseline was `669b2f0f91b0c787f7d8a26971c7c34210b935e1`. The unchanged packaged CLI had SHA256 `11db529ca76a62e3b19c95269da45583ffb6b820b5e9619ae38bb973661c5c2d`. Both runs used Ubuntu 24.04 arm64, Linux 6.8.0-146-generic, Java 21.0.12.1 and Node 22.23.3 in an isolated local VM.
+
+| Database        | Tests / failures / errors / skipped | Test duration | Result                        |
+| --------------- | ----------------------------------- | ------------- | ----------------------------- |
+| MySQL 8.4.11    | 1 / 0 / 0 / 0                       | 21.17 s       | `HOSTED_OPERATOR_RECOVERY_OK` |
+| MariaDB 11.4.13 | 1 / 0 / 0 / 0                       | 23.84 s       | `HOSTED_OPERATOR_RECOVERY_OK` |
+
+In both runs, the detached case recorded `escapedWriterLive=true` after the original worker was killed; its holder stayed fenced. Both recoveries completed with replacement binding identities, and both replacement Sessions wrote `available` while the original marker remained `x`. The maintenance retry and no-replay assertions also passed. These runs selected the legacy saved-result Shell profile. `W1_PHYSICAL_GUARD=false`; they do not certify W1 mount/backup isolation or H3 background/Monitor publication, restart or drain behavior.
+
+<details>
+<summary>中文</summary>
+
+本验收对应 [#12904](https://github.com/QwenLM/qwen-code/issues/12904)，测试 [#12977](https://github.com/QwenLM/qwen-code/pull/12977) 已交付的恢复机制，不新增另一套恢复逻辑。
+
+使用 Linux、Java 21、Node 22+ 和独立 MySQL 或 MariaDB 数据库，运行打包后的 CLI、真实 worker、Runtime Broker、Spring 存储服务及运维命令。仅模型端点返回确定性响应。先构建 CLI 和 bundle，再按 SDK Java 工作流安装 Java SDK 和 Broker 构件。上方命令只运行新增的验收方法；示例凭据仅用于可丢弃的验收数据库。将端口和数据库名换成独立的 MariaDB 配置后再次运行。缺少 Linux、SQL 或打包产物时，测试会失败。
+
+测试观察三个独立 Workspace：原始复合命令产生 `partial` / `producer_lost` 后继续持有 Workspace，第二个 Session 收到 `workspace_busy`；另一个复合命令必须启动真实的脱离进程，原 worker 死亡后它仍存活、文件持续增长，同时原 Workspace 的精确 holder 保持封锁；第三个 Workspace 在前两者阻塞期间仍能完成文件操作。
+
+恢复通过独立 JVM 调用真实运维入口：inspect 原绑定及 generation、prepare 精确持有的租约、停止测试拥有的 worker 和潜在写入者，再提交私有的静止声明。随后使用同一证据文件重试 complete。SQL 必须证明原租约被移除、原绑定已退役，新 Session 必须在恢复的 Workspace 中实际写入文件。原标记始终只有一个 `x`，恢复本身不得再次调用模型。
+
+成功输出包含 `HOSTED_OPERATOR_RECOVERY_OK` 和带身份信息的 `OPERATOR_RECOVERY` 记录。检查失败时保留 Surefire 报告和进程输出。本验收不证明任意逃逸进程、外部重启源、跨宿主恢复或原不确定回合已经完成；运维声明仍是[恢复流程](../../../docs/users/hosted-workspace-recovery.zh-CN.md)规定的信任边界。
+
+2026-10-09 的生产基线为 `669b2f0f91b0c787f7d8a26971c7c34210b935e1`，未修改的 CLI bundle SHA256 为 `11db529ca76a62e3b19c95269da45583ffb6b820b5e9619ae38bb973661c5c2d`。两次运行都使用独立本地虚拟机中的 Ubuntu 24.04 arm64、Linux 6.8.0-146-generic、Java 21.0.12.1 和 Node 22.23.3。MySQL 8.4.11：1 项测试、0 failures、0 errors、0 skipped，21.17 秒；MariaDB 11.4.13：1 项测试、0 failures、0 errors、0 skipped，23.84 秒。两次均输出 `HOSTED_OPERATOR_RECOVERY_OK`。
+
+两次脱离进程场景均在原 worker 被杀后记录 `escapedWriterLive=true`，holder 仍保持封锁。两个恢复均完成并产生新 binding，两个替代 Session 实际写入 `available`，原标记始终为 `x`；运维重试与不重放断言也通过。此次选择 legacy saved-result Shell profile。`W1_PHYSICAL_GUARD=false`，不能据此证明 W1 挂载/备份隔离或 H3 background/Monitor publication、重启和 drain 行为。
+
+</details>

integration-tests/helpers/hosted-operator-recovery-driver.ts

@@ -0,0 +1,215 @@
+/**
+ * @license
+ * Copyright 2026 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import assert from 'node:assert/strict';
+import { createHash, randomUUID } from 'node:crypto';
+import { readFile } from 'node:fs/promises';
+import path from 'node:path';
+import { fakeToolCall, startFakeOpenAIServer } from '../fake-openai-server.js';
+import { HostedHarnessProcess, waitUntil } from './hosted-harness-process.js';
+
+const config = JSON.parse(await readFile(process.argv[2], 'utf8')) as {
+  tenantId: string;
+  storeUrl: string;
+  brokerUrl: string;
+  statusGateUrl: string;
+  sessions: Array<{
+    sessionId: string;
+    secondarySessionId: string;
+    workspaceId: string;
+    directory: string;
+    toolProfile: string;
+    fault: string;
+  }>;
+};
+let current = config.sessions[0];
+let modelCalls = 0;
+let replacement = false;
+const model = await startFakeOpenAIServer(({ body }) => {
+  modelCalls++;
+  const messages = body['messages'] as Array<{
+    role: string;
+    content: unknown;
+  }>;
+  const last = messages.findLastIndex((message) => message.role === 'user');
+  if (messages.slice(last + 1).some((message) => message.role === 'tool'))
+    return { content: 'RECOVERY_CONTROL_DONE' };
+  if (replacement || current.fault === 'independent')
+    return {
+      toolCalls: [
+        fakeToolCall('write_file', {
+          file_path: replacement ? 'replacement.txt' : 'control.txt',
+          content: 'available',
+        }),
+      ],
+    };
+  const escaped =
+    current.fault === 'detached'
+      ? "setsid sh -c 'echo $$ > escaped.pid; while :; do printf x >> escaped.txt; sleep 0.1; done' >/dev/null 2>&1 & "
+      : '';
+  return {
+    toolCalls: [
+      fakeToolCall(
+        'run_shell_command',
+        {
+          command: `printf x >> once.txt; ${escaped}cd . && sleep 20 >/dev/null 2>&1 & echo ok`,
+        },
+        'partial-shell',
+      ),
+    ],
+  };
+});
+const cli = new HostedHarnessProcess();
+let clientId = '';
+async function json(route: string, body?: unknown, expected = 200) {
+  const response = await cli.request(route, {
+    method: body === undefined ? 'GET' : 'POST',
+    headers: { ...cli.headers(clientId), 'Content-Type': 'application/json' },
+    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
+  });
+  const text = await response.text();
+  assert.equal(response.status, expected, text);
+  return text ? JSON.parse(text) : undefined;
+}
+async function probe(phase: string) {
+  const response = await fetch(
+    new URL(
+      `/operator-recovery/${current.sessionId}/${phase}`,
+      config.statusGateUrl,
+    ),
+    {
+      method: 'POST',
+      signal: AbortSignal.timeout(90_000),
+    },
+  );
+  const text = await response.text();
+  assert.equal(response.status, 200, text);
+  console.log(`OPERATOR_RECOVERY ${current.fault} ${phase} ${text}`);
+}
+try {
+  await cli.start(model.baseUrl, {
+    extraArgs: [
+      '--managed-runtime-broker-url',
+      config.brokerUrl,
+      '--managed-runtime-broker-token',
+      'hosted-tools-broker-token',
+    ],
+  });
+  // The independent Workspace remains usable while both damaged Workspaces are fenced.
+  for (const session of config.sessions) {
+    current = session;
+    const created = await json('/session', {
+      sessionId: session.sessionId,
+      sessionScope: 'thread',
+      toolProfile: session.toolProfile,
+      managedSessionStore: {
+        baseUrl: config.storeUrl,
+        tenantId: config.tenantId,
+        workspaceId: session.workspaceId,
+        writerId: cli.bootId,
+        leaseDurationMs: 60_000,
+      },
+    });
+    clientId = created.clientId;
+    const blocks = [{ type: 'text', text: `RECOVERY_${session.fault}` }];
+    await json(
+      `/session/${session.sessionId}/prompt`,
+      {
+        promptId: randomUUID(),
+        prompt: blocks,
+        payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(blocks)).digest('hex')}`,
+      },
+      202,
+    );
+    await waitUntil(
+      async () =>
+        !(await json(`/session/${session.sessionId}/status`)).hasActivePrompt,
+      60_000,
+    );
+    const status = await json(`/session/${session.sessionId}/status`);
+    assert.equal(
+      status.recoveryBlocked,
+      session.fault !== 'independent',
+      cli.output,
+    );
+    if (session.fault === 'independent') {
+      assert.equal(
+        await readFile(path.join(session.directory, 'control.txt'), 'utf8'),
+        'available',
+      );
+    } else {
+      assert.equal(
+        await readFile(path.join(session.directory, 'once.txt'), 'utf8'),
+        'x',
+      );
+      await probe('prepare');
+    }
+  }
+  const callsBeforeRecovery = modelCalls;
+  for (const session of config.sessions.slice(0, 2)) {
+    current = session;
+    await probe('complete');
+    assert.equal(
+      await readFile(path.join(session.directory, 'once.txt'), 'utf8'),
+      'x',
+    );
+  }
+  assert.equal(
+    modelCalls,
+    callsBeforeRecovery,
+    'Operator recovery must never replay inference or Shell',
+  );
+  replacement = true;
+  for (const session of config.sessions.slice(0, 2)) {
+    current = session;
+    const created = await json('/session', {
+      sessionId: session.secondarySessionId,
+      sessionScope: 'thread',
+      toolProfile: session.toolProfile,
+      managedSessionStore: {
+        baseUrl: config.storeUrl,
+        tenantId: config.tenantId,
+        workspaceId: session.workspaceId,
+        writerId: cli.bootId,
+        leaseDurationMs: 60_000,
+      },
+    });
+    clientId = created.clientId;
+    const blocks = [{ type: 'text', text: 'RECOVERY_NEW_SESSION' }];
+    await json(
+      `/session/${session.secondarySessionId}/prompt`,
+      {
+        promptId: randomUUID(),
+        prompt: blocks,
+        payloadDigest: `sha256:${createHash('sha256').update(JSON.stringify(blocks)).digest('hex')}`,
+      },
+      202,
+    );
+    await waitUntil(
+      async () =>
+        !(await json(`/session/${session.secondarySessionId}/status`))
+          .hasActivePrompt,
+      60_000,
+    );
+    assert.equal(
+      (await json(`/session/${session.secondarySessionId}/status`))
+        .recoveryBlocked,
+      false,
+    );
+    assert.equal(
+      await readFile(path.join(session.directory, 'replacement.txt'), 'utf8'),
+      'available',
+    );
+    assert.equal(
+      await readFile(path.join(session.directory, 'once.txt'), 'utf8'),
+      'x',
+    );
+  }
+  console.log('HOSTED_OPERATOR_RECOVERY_OK');
+} finally {
+  await cli.close();
+  await model.close();
+}

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

@@ -0,0 +1,267 @@
+package com.alibaba.qwen.code.managedagent;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import com.alibaba.qwen.code.managedagent.service.EmbeddedRuntimeBroker;
+import com.alibaba.qwen.code.managedagent.service.WorkspaceRecoveryCommand;
+import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
+import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerService;
+import com.alibaba.qwen.code.runtimebroker.RuntimeProvisionSeed;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpServer;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.NoSuchFileException;
+import java.nio.file.attribute.PosixFilePermissions;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.util.ReflectionTestUtils;
+
+/** Real maintenance CLI, real worker and SQL; only the model is deterministic. */
+final class HostedOperatorRecoveryProbe implements AutoCloseable {
+    private static final ObjectMapper JSON = new ObjectMapper();
+    private final ServletWebServerApplicationContext spring;
+    private final JdbcTemplate jdbc;
+    private final List<Map<String, Object>> sessions;
+    private final Path state;
+    private final RuntimeBrokerService broker;
+    private final Object local;
+    private final Set<ProcessHandle> processes = new LinkedHashSet<>();
+    private final Map<String, ProcessHandle> escapedProcesses = new HashMap<>();
+    private final Map<String, String> recoveries = new HashMap<>();
+
+    HostedOperatorRecoveryProbe(ServletWebServerApplicationContext spring, List<Map<String, Object>> sessions,
+            Path state, HttpServer server) {
+        this.spring = spring;
+        this.jdbc = spring.getBean(JdbcTemplate.class);
+        this.sessions = sessions;
+        this.state = state;
+        broker = (RuntimeBrokerService) ReflectionTestUtils.getField(spring.getBean(EmbeddedRuntimeBroker.class), "service");
+        Object provisioner = ReflectionTestUtils.getField(broker, "provisioner");
+        local = ReflectionTestUtils.getField(provisioner, "delegate");
+        server.createContext("/operator-recovery/", exchange -> {
+            try {
+                assertThat(exchange.getRequestMethod()).isEqualTo("POST");
+                String[] route = exchange.getRequestURI().getPath().split("/");
+                assertThat(route).hasSize(4);
+                var session = sessions.stream().filter(value -> value.get("sessionId").equals(route[2]))
+                        .findFirst().orElseThrow();
+                byte[] result = JSON.writeValueAsBytes(check(session, route[3]));
+                exchange.sendResponseHeaders(200, result.length);
+                exchange.getResponseBody().write(result);
+            } catch (Throwable error) {
+                byte[] result = error.toString().getBytes(StandardCharsets.UTF_8);
+                exchange.sendResponseHeaders(500, result.length);
+                exchange.getResponseBody().write(result);
+            } finally {
+                exchange.close();
+            }
+        });
+    }
+
+    private Map<String, Object> check(Map<String, Object> session, String phase) throws Exception {
+        String id = session.get("sessionId").toString();
+        var rows = jdbc.queryForList("SELECT * FROM qwen_tool_execution WHERE harness_session_id = ?", id);
+        assertThat(rows).hasSize(1);
+        var execution = rows.getFirst();
+        String binding = execution.get("binding_id").toString();
+        String generation = execution.get("runtime_generation").toString();
+        assertThat(execution.get("dispatch_generation")).isEqualTo(1L);
+        if (phase.equals("prepare")) {
+            assertBusy(session);
+            String inspection = command("inspect", binding, generation);
+            var view = JSON.readTree(inspection.lines().filter(line -> line.startsWith("{")).findFirst().orElseThrow());
+            assertThat(view.path("captureStatus").asText()).isEqualTo("partial");
+            assertThat(view.path("captureReason").asText()).isEqualTo("producer_lost");
+            String holder = view.path("holderKey").asText();
+            assertThat(holder).isNotBlank();
+            String prepared = command("prepare", binding, generation, holder, "#12904 physical acceptance");
+            String recovery = prepared.lines().filter(line -> line.matches("[0-9a-f-]{36}")).findFirst().orElseThrow();
+            assertThat(recoveries.putIfAbsent(id, recovery)).isNull();
+            ProcessHandle worker = worker(binding);
+            processes.addAll(worker.descendants().toList());
+            processes.add(worker);
+            ProcessHandle escaped = escaped(session);
+            if ("detached".equals(session.get("fault"))) {
+                assertThat(escaped).as("detached fixture must start a real escaped writer").isNotNull();
+            }
+            worker.destroyForcibly();
+            worker.onExit().get(10, TimeUnit.SECONDS);
+            assertHeld(binding, holder);
+            if (escaped != null) {
+                Path proof = Path.of(session.get("directory").toString(), "escaped.txt");
+                long before = Files.size(proof);
+                long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+                while (Files.size(proof) == before && System.nanoTime() < deadline) {
+                    Thread.sleep(20);
+                }
+                assertThat(running(escaped)).isTrue();
+                assertThat(Files.size(proof)).as("escaped writer survives worker death").isGreaterThan(before);
+                assertHeld(binding, holder);
+            }
+            return Map.of("bindingId", binding, "holderKey", holder, "recoveryId", recovery,
+                    "state", "fenced", "escapedWriterLive", escaped != null);
+        }
+        assertThat(phase).isEqualTo("complete");
+        String recovery = recoveries.get(id);
+        assertThat(recovery).isNotNull();
+        stopOwnedProcesses();
+        Path evidence = state.resolve(recovery + ".json");
+        Files.writeString(evidence, JSON.writeValueAsString(Map.of("version", 1, "recoveryId", recovery,
+                "verifiedAt", Instant.now().toString(), "method", "isolated test process inventory",
+                "actions", "Stopped the owned worker, Shell descendants and escaped writer; no external restart source exists",
+                "restartPrevention", true)));
+        Files.setPosixFilePermissions(evidence, PosixFilePermissions.fromString("rw-------"));
+        assertThat(command("complete", recovery, evidence.toString())).contains("completed");
+        assertThat(command("complete", recovery, evidence.toString())).contains("completed");
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM managed_workspace_execution_lease WHERE binding_id = ?",
+                Integer.class, binding)).isZero();
+        assertThat(jdbc.queryForObject("SELECT binding_state FROM qwen_runtime_binding WHERE binding_id = ?",
+                String.class, binding)).isEqualTo("RELEASED");
+        var next = broker.acquire(session.get("secondarySessionId").toString(), UUID.randomUUID().toString(), "bootstrap")
+                .toCompletableFuture().get(30, TimeUnit.SECONDS);
+        assertThat(next.getBindingId()).isNotEqualTo(binding);
+        assertThat(broker.release(next.getSession().getHarnessSessionId(), next.getRuntimeSessionId())
+                .toCompletableFuture().get(30, TimeUnit.SECONDS)).isTrue();
+        return Map.of("bindingId", binding, "recoveryId", recovery, "state", "completed",
+                "replacementBindingId", next.getBindingId());
+    }
+
+    private void assertBusy(Map<String, Object> session) throws Exception {
+        try {
+            broker.acquire(session.get("secondarySessionId").toString(), UUID.randomUUID().toString(), "bootstrap")
+                    .toCompletableFuture().get(10, TimeUnit.SECONDS);
+            throw new AssertionError("A second Session entered the uncertain Workspace");
+        } catch (java.util.concurrent.ExecutionException error) {
+            Throwable cause = error.getCause();
+            while (cause.getCause() != null && !(cause instanceof RuntimeBrokerException)) {
+                cause = cause.getCause();
+            }
+            assertThat(cause).isInstanceOf(RuntimeBrokerException.class);
+            assertThat(((RuntimeBrokerException) cause).getCode()).isEqualTo("workspace_busy");
+        }
+    }
+
+    private void assertHeld(String binding, String holder) {
+        assertThat(jdbc.queryForObject("SELECT holder_key FROM managed_workspace_execution_lease WHERE binding_id = ?",
+                String.class, binding)).isEqualTo(holder);
+    }
+
+    private ProcessHandle worker(String binding) {
+        var owned = (Map<?, ?>) ReflectionTestUtils.getField(local, "owned");
+        Object match = owned.values().stream().filter(value -> binding.equals(
+                ((RuntimeProvisionSeed) ReflectionTestUtils.getField(value, "seed")).getProvisionalRuntimeId()))
+                .findFirst().orElseThrow();
+        var process = registeredProcess(match);
+        assertThat(process).as("original durable worker with matching process identity").isNotNull();
+        return process;
+    }
+
+    private ProcessHandle registeredProcess(Object owned) {
+        Object registration = ReflectionTestUtils.getField(owned, "registration");
+        return ReflectionTestUtils.invokeMethod(registration, "process");
+    }
+
+    private ProcessHandle escaped(Map<String, Object> session) throws Exception {
+        String id = session.get("sessionId").toString();
+        if (escapedProcesses.containsKey(id)) {
+            return escapedProcesses.get(id);
+        }
+        Path directory = Path.of(session.get("directory").toString());
+        Path identity = directory.resolve("escaped.pid");
+        if (!Files.exists(identity)) {
+            return null;
+        }
+        var process = ProcessHandle.of(Long.parseLong(Files.readString(identity).strip())).orElse(null);
+        if (process == null || !running(process)) {
+            return null;
+        }
+        assertThat(Files.readSymbolicLink(Path.of("/proc", Long.toString(process.pid()), "cwd"))).isEqualTo(directory);
+        processes.add(process);
+        escapedProcesses.put(id, process);
+        return process;
+    }
+
+    private String command(String... arguments) throws Exception {
+        var command = new ArrayList<String>();
+        command.add(Path.of(System.getProperty("java.home"), "bin", "java").toString());
+        for (String name : List.of("spring.datasource.url", "spring.datasource.driver-class-name",
+                "spring.datasource.username", "spring.datasource.password",
+                "qwen.managed-agent.runtime-broker.credential-key-id", "qwen.managed-agent.runtime-broker.credential-key")) {
+            command.add("-D" + name + "=" + spring.getEnvironment().getProperty(name));
+        }
+        command.add("-Dqwen.managed-agent.runtime-broker.state-directory=" + state);
+        command.add("-Dqwen.managed-agent.runtime-broker.durable-local-process=true");
+        command.add("-Dqwen.managed-agent.runtime-broker.operator-recovery-enabled=true");
+        command.add("-Dqwen.managed-agent.runtime-broker.provisioner=local-process");
+        command.add("-cp");
+        command.add(System.getProperty("surefire.test.class.path", System.getProperty("java.class.path")));
+        command.add(WorkspaceRecoveryCommand.class.getName());
+        command.addAll(List.of(arguments));
+        Path log = state.resolve("maintenance-" + UUID.randomUUID() + ".log");
+        Process maintenance = new ProcessBuilder(command).redirectErrorStream(true).redirectOutput(log.toFile()).start();
+        try {
+            assertThat(maintenance.waitFor(45, TimeUnit.SECONDS)).as("Maintenance timeout: %s", Files.readString(log)).isTrue();
+            assertThat(maintenance.exitValue()).as("Maintenance output: %s", Files.readString(log)).isZero();
+            return Files.readString(log);
+        } finally {
+            if (maintenance.isAlive()) {
+                maintenance.destroyForcibly();
+            }
+        }
+    }
+
+    private static boolean running(ProcessHandle process) throws Exception {
+        if (!process.isAlive()) {
+            return false;
+        }
+        Path stat = Path.of("/proc", Long.toString(process.pid()), "stat");
+        if (!Files.exists(stat)) {
+            return false;
+        }
+        try {
+            String value = Files.readString(stat);
+            return value.charAt(value.lastIndexOf(')') + 2) != 'Z';
+        } catch (NoSuchFileException exited) {
+            return false;
+        }
+    }
+
+    private void stopOwnedProcesses() throws Exception {
+        processes.forEach(ProcessHandle::destroyForcibly);
+        long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+        for (ProcessHandle process : processes) {
+            while (running(process)) {
+                assertThat(System.nanoTime()).as("Owned producer %s", process.pid()).isLessThan(deadline);
+                Thread.sleep(20);
+            }
+        }
+    }
+
+    @Override
+    public void close() throws Exception {
+        for (var session : sessions) {
+            escaped(session);
+        }
+        var owned = (Map<?, ?>) ReflectionTestUtils.getField(local, "owned");
+        for (var value : owned.values()) {
+            ProcessHandle process = registeredProcess(value);
+            if (process == null) {
+                continue;
+            }
+            processes.addAll(process.descendants().toList());
+            processes.add(process);
+        }
+        stopOwnedProcesses();
+    }
+}

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

@@ -19,6 +19,7 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.attribute.FileTime;
+import java.nio.file.attribute.PosixFilePermissions;
 import java.security.MessageDigest;
 import java.util.ArrayList;
 import java.util.Base64;
@@ -53,6 +54,14 @@ class HostedWorkspaceToolTurnIT {
                 "workspace-tool-turn");
     }
 
+    @Test
+    @Timeout(240)
+    void operatorRecoveryReleasesOnlyTheOriginalQuiescentShellOnMySql() throws Exception {
+        assertThat(System.getProperty("os.name")).isEqualTo("Linux");
+        assertThat(System.getProperty("mysql.url")).startsWith("jdbc:mysql:");
+        runDriver(List.of("compound", "detached", "independent"), "operator-recovery");
+    }
+
     @Test
     @Timeout(180)
     void recordsLatencyWithDelayedRuntimeProvisioning() throws Exception {
@@ -151,7 +160,8 @@ class HostedWorkspaceToolTurnIT {
 
     private void runDriver(List<String> cases, String driverName) throws Exception {
         boolean latency = driverName.equals("latency");
-        boolean faults = !driverName.equals("workspace-tool-turn") && !latency;
+        boolean operatorRecovery = driverName.equals("operator-recovery");
+        boolean faults = !driverName.equals("workspace-tool-turn") && !latency && !operatorRecovery;
         boolean storeFaults = driverName.equals("store-failure");
         boolean cancellations = driverName.equals("cancellation");
         boolean sseGaps = driverName.equals("sse-gap");
@@ -163,6 +173,12 @@ class HostedWorkspaceToolTurnIT {
         assertThat(node).as("Pass -Dnode.executable with an absolute Node.js 22+ path").isNotBlank();
         temporary = temporary.toRealPath();
         Files.createDirectory(temporary.resolve("runtime"));
+        if (operatorRecovery) {
+            Files.setPosixFilePermissions(temporary.resolve("runtime"),
+                    PosixFilePermissions.fromString("rwx------"));
+        }
+        Path fallbackWorkspace = operatorRecovery
+                ? Files.createDirectory(temporary.resolve("legacy-workspace")) : temporary;
         Path root = cli.getParent().getParent();
         Path worker = cli;
         int runtimeProvisioningDelayMs = 15_000;
@@ -194,9 +210,9 @@ class HostedWorkspaceToolTurnIT {
                 "--qwen.managed-agent.runtime-broker.enabled=true",
                 "--qwen.managed-agent.runtime-broker.port=0",
                 "--qwen.managed-agent.runtime-broker.token=hosted-tools-broker-token",
-                "--qwen.managed-agent.runtime-broker.durable-local-process=false",
+                "--qwen.managed-agent.runtime-broker.durable-local-process=" + operatorRecovery,
                 "--qwen.managed-agent.runtime-broker.trusted-local-reboot-recovery=false",
-                "--qwen.managed-agent.runtime-broker.workspace-cwd=" + temporary,
+                "--qwen.managed-agent.runtime-broker.workspace-cwd=" + fallbackWorkspace,
                 "--qwen.managed-agent.runtime-broker.state-directory=" + temporary.resolve("runtime"),
                 "--qwen.managed-agent.runtime-broker.credential-key-id=test",
                 "--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
@@ -212,7 +228,7 @@ class HostedWorkspaceToolTurnIT {
             // Give initialized roots an unambiguous birth time without a timing-dependent sleep.
             Files.setLastModifiedTime(workspaces.get(index), FileTime.fromMillis(1));
         }
-        boolean verifiedRecovery = !faults && !latency && "Linux".equals(System.getProperty("os.name"));
+        boolean verifiedRecovery = !faults && !latency && !operatorRecovery && "Linux".equals(System.getProperty("os.name"));
         arguments.add("--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=" + verifiedRecovery);
         var application = new SpringApplicationBuilder(ManagedAgentServerApplication.class);
         if (sseGaps) {
@@ -242,8 +258,13 @@ class HostedWorkspaceToolTurnIT {
                 var created = store.insertWorkspaceSessionCommand(tenant, "actor", "create-" + index,
                         "sha256:" + "a".repeat(64), "qwen-code", null, null, List.of(), null,
                         new WorkspaceSelection(workspaceId, "child"));
+                String profile = operatorRecovery ? (index < 2 ? "hosted-workspace-shell/1" : "hosted-workspace-files/1")
+                        : !shellOutput && (faults || index < 2) ? "hosted-workspace-files/1" : "hosted-workspace-shell/1";
+                String secondary = operatorRecovery ? store.insertWorkspaceSessionCommand(tenant, "actor", "secondary-" + index,
+                        "sha256:" + "a".repeat(64), "qwen-code", null, null, List.of(), null,
+                        new WorkspaceSelection(workspaceId, "child")).sessionId() : "";
                 sessions.add(Map.of("sessionId", created.sessionId(), "workspaceId", workspaceId,
-                        "toolProfile", !shellOutput && (faults || index < 2) ? "hosted-workspace-files/1" : "hosted-workspace-shell/1",
+                        "toolProfile", profile, "secondarySessionId", secondary,
                         "directory", workspaces.get(index).resolve("child").toString(), "fault", cases.get(index)));
                 if (faults) Files.writeString(workspaces.get(index).resolve("child/proof.txt"), "x");
             }
@@ -292,9 +313,11 @@ class HostedWorkspaceToolTurnIT {
                     ? new HostedProviderControlProbe(jdbc, tenant, sessions, broker, gateServer) : null;
             HostedWorkspaceColdLoadProbe coldLoadProbe = new HostedWorkspaceColdLoadProbe(jdbc, tenant, sessions,
                     "http://127.0.0.1:" + spring.getWebServer().getPort(), gateServer);
+            HostedOperatorRecoveryProbe operatorProbe = operatorRecovery
+                    ? new HostedOperatorRecoveryProbe(spring, sessions, temporary.resolve("runtime"), gateServer) : null;
             gateServer.start();
             List<String> triggers = new ArrayList<>();
-            try (shellProbe; providerProbe; coldLoadProbe) {
+            try (shellProbe; providerProbe; coldLoadProbe; operatorProbe) {
                 if (storeFaults || shellOutput) {
                     for (Map<String, Object> session : sessions) {
                         if (shellOutput ? session.get("fault").equals("receipt-failure")
@@ -326,7 +349,7 @@ class HostedWorkspaceToolTurnIT {
                     assertThat(driver.waitFor(faults || latency ? 130 : 270, TimeUnit.SECONDS)).as("Driver timeout: %s", Files.readString(log)).isTrue();
                     assertThat(driver.exitValue()).as("Driver output: %s", Files.readString(log)).isZero();
                     System.out.println(Files.readString(log));
-                    assertThat(Files.readString(log)).contains(latency ? "HOSTED_LATENCY_OK"
+                    assertThat(Files.readString(log)).contains(operatorRecovery ? "HOSTED_OPERATOR_RECOVERY_OK" : latency ? "HOSTED_LATENCY_OK"
                             : providerControl ? "HOSTED_PROVIDER_FAULTS_OK"
                             : shellOutput ? "HOSTED_SHELL_OUTPUT_FAULTS_OK"
                             : sseGaps ? "HOSTED_SSE_GAP_OK"
@@ -338,6 +361,9 @@ class HostedWorkspaceToolTurnIT {
                     if (faults && cases.contains("status")) assertThat(statusGate.isDone()).isTrue();
                     for (int index = 0; index < workspaces.size(); index++) {
                         Path workspace = workspaces.get(index);
+                        if (operatorRecovery) {
+                            continue;
+                        }
                         if (providerControl) providerProbe.assertReport(sessions.get(index), reports.get(index));
                         else if (latency) {
                             boolean tool = cases.get(index).equals("tool");
@@ -373,7 +399,7 @@ class HostedWorkspaceToolTurnIT {
                     driver.descendants().forEach(process -> process.destroyForcibly());
                     if (driver.isAlive()) driver.destroyForcibly();
                 }
-                if (!faults && !latency) {
+                if (!faults && !latency && !operatorRecovery) {
                     List<ProcessHandle> producers = ProcessHandle.current().descendants().toList();
                     assertThat(producers).as("Runtime producers before Broker shutdown").isNotEmpty();
                     broker.close();