fix(runtime-broker): close drained lost workspace bindings

yiliang114 · Oct 8, 2026 · Sep 30, 2026 · Sep 30, 2026 · Sep 30, 2026 · Sep 30, 2026
commit 8953b8ffb25da9e91b18a2f53b43432fd47580fa
diff --git a/docs/design/workspace-session-reliable-close.md b/docs/design/workspace-session-reliable-close.md
@@ -26,7 +26,7 @@ Persist a permanent tenant/Harness-Session drain fence under the existing tenant
 
 Fence probes use a nonlocking read after the placement guard, avoiding InnoDB gap locks that would block another tenant's fence insertion when no row exists. Execution admission discovers the immutable binding tenant outside the admission transaction; its first consistent read occurs after acquiring the guard, so REPEATABLE READ cannot hide a fence committed while admission was waiting.
 
-Enumerate saved binding generations by tenant, Session isolation class, and isolation key, in bounded pages with byte-exact identities regardless of database collation. Mark them draining without authorizing current Workspace execution. Enumerate Runtime Sessions by exact binding/generation and release using saved records: provider release, activation=false acknowledgement, conditional original-holder release, then durable RELEASED. No acquire, installation, execution replay, or model call is part of close. Unknown execution or startup identity blocks completion. An unusable original worker blocks with an identity failure instead of entering generic lease recovery. A newer holder on shared storage is preserved.
+Enumerate saved binding generations by tenant, Session isolation class, and isolation key, in bounded pages with byte-exact identities regardless of database collation. Mark them draining without authorizing current Workspace execution. Enumerate Runtime Sessions by exact binding/generation and release using saved records: provider release, activation=false acknowledgement, conditional original-holder release, then durable RELEASED. No acquire, installation, execution replay, or model call is part of close. Unknown execution or startup identity blocks completion. A LOST binding with no active Runtime Sessions or executions may retire through the same holder check and durable stop receipt; it remains LOST until the receipt commits RELEASED. A LOST binding with unsettled resources still requires recovery. An unusable original worker with an unreleased Session blocks with an identity failure instead of entering generic lease recovery. A newer holder on shared storage is preserved.
 
 ## Worker stop and completion
 

diff --git a/docs/design/workspace-session-reliable-close.zh-CN.md b/docs/design/workspace-session-reliable-close.zh-CN.md
@@ -26,7 +26,7 @@ Harness prompt 和既有 Runtime 恢复准入路由在本地 close 开始后拒
 
 栅栏查询在取得 placement 锁后进行普通读取，避免缺失行上的 InnoDB 间隙锁阻塞其他租户插入栅栏。Execution 准入在准入事务之外读取不可变的 binding tenant；事务的首次一致性读发生在取得锁之后，避免 REPEATABLE READ 隐藏等待期间已提交的栅栏。
 
-按 tenant、Session isolation class 和 isolation key 分页枚举保存的 binding 代际；身份按字节精确匹配，不依赖数据库排序规则。标记 draining 时不重新授权当前 Workspace 执行。按精确 binding/generation 枚举 Runtime Sessions，使用保存的记录按顺序释放：provider release、activation=false 确认、条件释放原 holder、持久 RELEASED。close 不进行 acquire、安装、执行重放或模型调用。未知执行或启动身份阻止完成。原 worker 不可用时返回身份失败并阻塞，不进入通用租约恢复。共享存储上的新 holder 必须保留。
+按 tenant、Session isolation class 和 isolation key 分页枚举保存的 binding 代际；身份按字节精确匹配，不依赖数据库排序规则。标记 draining 时不重新授权当前 Workspace 执行。按精确 binding/generation 枚举 Runtime Sessions，使用保存的记录按顺序释放：provider release、activation=false 确认、条件释放原 holder、持久 RELEASED。close 不进行 acquire、安装、执行重放或模型调用。未知执行或启动身份阻止完成。没有活跃 Runtime Session 或 execution 的 LOST binding，可以经过同样的 holder 检查与持久停机凭据完成退休；提交凭据并进入 RELEASED 前保持 LOST。有未结算资源的 LOST binding 仍需要恢复。原 worker 不可用且仍有未释放的 Session 时返回身份失败并阻塞，不进入通用租约恢复。共享存储上的新 holder 必须保留。
 
 ## Worker 停机与完成
 

diff --git a/...gent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedPublicWorkspaceIT.java b/...gent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedPublicWorkspaceIT.java
@@ -40,6 +40,8 @@
 import org.junit.jupiter.api.condition.OS;
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.api.io.CleanupMode;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.springframework.boot.builder.SpringApplicationBuilder;
 import org.springframework.boot.web.servlet.FilterRegistrationBean;
 import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
@@ -83,10 +85,11 @@ void ownerAnswersHostedApprovalsThroughBothSurfaces() throws Exception {
         assertThat(answered).hasSize(8);
     }
 
-    @Test
+    @ParameterizedTest
+    @ValueSource(booleans = {false, true})
     @EnabledOnOs(OS.LINUX)
     @Timeout(150)
-    void durableCloseStopsOriginalWorkersAndRetainsHistoryAndFiles() throws Exception {
+    void durableCloseStopsOriginalWorkersAndRetainsHistoryAndFiles(boolean crash) throws Exception {
         durableClose = true;
         runFiles();
         List<String> sessions = jdbc.queryForList("SELECT session_id FROM managed_agent_session WHERE tenant_id = ?"
@@ -101,6 +104,23 @@ void durableCloseStopsOriginalWorkersAndRetainsHistoryAndFiles() throws Exceptio
             long pid = json.readTree(Files.readString(registration)).path("pid").asLong();
             var worker = ProcessHandle.of(pid).orElseThrow();
             assertThat(worker.isAlive()).isTrue();
+            if (crash) {
+                worker.destroyForcibly();
+                worker.onExit().get(5, TimeUnit.SECONDS);
+                if (index == 1) {
+                    jdbc.update("UPDATE managed_workspace_registry SET config_ref = ? WHERE tenant_id = ?"
+                            + " AND workspace_id = ?", WorkspaceExecutionProfile.CONFIG_REF, tenant, "workspace-" + index);
+                    String failedTurn = request("POST", "/v1/agents/sessions/" + session + "/events",
+                            Map.of("type", "agent.session.input.message", "input",
+                                    List.of(Map.of("type", "input_text", "text", "G0_AGAIN"))),
+                            "after-crash", "actor", 202).path("turn_id").asText();
+                    await().atMost(Duration.ofSeconds(35)).untilAsserted(() -> assertThat(jdbc.queryForObject(
+                            "SELECT status FROM managed_agent_turn WHERE session_id = ? AND turn_id = ?",
+                            String.class, session, failedTurn)).isEqualTo("FAILED"));
+                }
+                assertThat(jdbc.queryForObject("SELECT binding_state FROM qwen_runtime_binding WHERE binding_id = ?",
+                        String.class, binding)).isEqualTo(index == 0 ? "READY" : "LOST");
+            }
             var retained = jdbc.queryForList("SELECT resource_id, sha256 FROM qwen_managed_session_resource"
                     + " WHERE tenant_id = ? AND session_id = ? ORDER BY resource_id", tenant, session);
             assertThat(retained).isNotEmpty();
@@ -132,6 +152,19 @@ void durableCloseStopsOriginalWorkersAndRetainsHistoryAndFiles() throws Exceptio
                     .resolve("child/proof.txt"))).isEqualTo("after");
             assertThat(request("POST", route, body, "close", "actor", 202)
                     .path(webShell ? "operationId" : "id").asText()).isEqualTo(operation);
+            if (crash) {
+                jdbc.update("UPDATE managed_workspace_registry SET config_ref = ? WHERE tenant_id = ?"
+                        + " AND workspace_id = ?", WorkspaceExecutionProfile.CONFIG_REF, tenant, "workspace-" + index);
+                String nextSession = request("POST", "/v1/agents/sessions",
+                        Map.of("agent_id", "qwen-code", "input", List.of(Map.of("type", "input_text", "text", "G0_FILES")),
+                                "workspace", Map.of("workspace_id", "workspace-" + index, "cwd_relative", "child")),
+                        "after-close-" + index, "actor", 202).path("id").asText();
+                await().atMost(Duration.ofSeconds(35)).untilAsserted(() -> assertThat(jdbc.queryForObject(
+                        "SELECT status FROM managed_agent_turn WHERE session_id = ?", String.class, nextSession))
+                        .isEqualTo("COMPLETED"));
+                assertThat(Files.readString(temporary.resolve(index == 0 ? "workspace-a" : "workspace-b")
+                        .resolve("child/proof.txt"))).isEqualTo("after");
+            }
         }
     }
 
@@ -536,7 +569,8 @@ private void modelReply(HttpExchange exchange) throws IOException {
                 String role = message.path("role").asText();
                 if ("user".equals(role) && message.path("content").toString().contains("G0_")) {
                     results.clear();
-                    prompt.set(message.path("content").toString());
+                    String content = message.path("content").toString();
+                    prompt.set(content.substring(content.lastIndexOf("G0_")));
                 } else if ("tool".equals(role)) {
                     results.add(message);
                 }

diff --git a/...erver/src/test/java/com/alibaba/qwen/code/managedagent/ManagedWorkspaceAdmissionTest.java b/...erver/src/test/java/com/alibaba/qwen/code/managedagent/ManagedWorkspaceAdmissionTest.java
@@ -665,6 +665,10 @@ public boolean isWorkspaceFilesAvailable() {
         // stays admitted.
         assertThat(enabled.getWebShellSession(tenant, "actor-a", controlId)
                 .capabilities().workspaceTurns()).isTrue();
+        ManagedAgentService disabled = new ManagedAgentService(store,
+                new RequestDigests(), null, new UnavailableHarnessConnector(), registry);
+        assertThat(disabled.getWebShellSession(tenant, "actor-a", controlId)
+                .capabilities().workspaceTurns()).isFalse();
     }
 
     @Test
@@ -735,6 +739,21 @@ public void rename(String tenantId, String sessionId,
                         + " AND session_id = ? AND command_status = 'PENDING'",
                 Integer.class, tenant, sessionId)).isZero();
 
+        transaction.executeWithoutResult(status -> gated.beginSessionMutation(
+                tenant, "RENAME_SESSION", "rename-blocker", digest, sessionId,
+                SessionMutationKind.RENAME));
+        transaction.executeWithoutResult(status ->
+                assertThatThrownBy(() -> service.renameSession(tenant,
+                        "actor-a", "rename-1", sessionId, "first"))
+                        .isInstanceOfSatisfying(ApiException.class, error ->
+                                assertThat(error.getCode()).isEqualTo("session_operation_active")));
+        assertThat(jdbc.queryForObject("SELECT command_status FROM managed_agent_command"
+                        + " WHERE tenant_id = ? AND operation = 'RENAME_SESSION' AND idempotency_key = 'rename-1'",
+                String.class, tenant)).isEqualTo("FAILED");
+        transaction.executeWithoutResult(status -> gated.completeSessionMutation(
+                tenant, "RENAME_SESSION", "rename-blocker", sessionId,
+                SessionMutationKind.RENAME, "intermediate", "boot"));
+
         // The freed key stays re-usable: a same-key retry re-attempts the
         // mutation instead of colliding on the requested event the refused
         // attempt already published.

diff --git a/...untime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBindingRecord.java b/...untime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBindingRecord.java
@@ -390,7 +390,8 @@ void requireSafeReplacement(RuntimeBindingRecord replacement) {
                         && stopEvidence != null) {
             throw new IllegalArgumentException("Recovery evidence cannot be overwritten");
         }
-        if ((state == State.LOST && replacement.state != State.LOST)
+        if ((state == State.LOST && replacement.state != State.LOST
+                && !(replacement.state == State.RELEASED && replacement.drainReceipt != null))
                 || state == State.OPERATOR_RECOVERY
                         && replacement.state != State.OPERATOR_RECOVERY
                         && replacement.state != State.LOST) {

diff --git a/...untime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBrokerService.java b/...untime-broker/src/main/java/com/alibaba/qwen/code/runtimebroker/RuntimeBrokerService.java
@@ -194,7 +194,6 @@ private CompletionStage<Void> drainBinding(RuntimeBindingRecord saved) {
             return CompletableFuture.completedFuture(null);
         }
         if (!saved.getRequest().isManagedContext() || !provisioner.supportsDrainedStop()
-                || saved.getState() == RuntimeBindingRecord.State.LOST
                 || saved.getState() == RuntimeBindingRecord.State.OPERATOR_RECOVERY
                 || saved.getState() == RuntimeBindingRecord.State.FAILED) {
             return failed(conflict("workspace_close_identity_unverified", "Original worker needs recovery"));
@@ -214,8 +213,14 @@ private CompletionStage<Void> drainClaimedBinding(RuntimeBindingRecord saved) {
         if (claimed == null) {
             return failed(unavailable("runtime_close_claim_pending", "Original binding is still claimed"));
         }
+        boolean lost = claimed.getState() == RuntimeBindingRecord.State.LOST;
+        if (lost && (sessionRepository.countActiveByBinding(claimed.getBindingId(), claimed.getGeneration()) != 0
+                || executionRepository.hasActiveByBinding(claimed.getBindingId(), claimed.getGeneration()))) {
+            releaseOperationQuietly(claimed.getBindingId(), claimed.getOperationGeneration());
+            return failed(conflict("workspace_close_execution_unsettled", "Lost Runtime resources require recovery"));
+        }
         var draining = bindingRepository.compareAndSet(claimed, claimed.withDrainRequested(true, clock.instant())
-                .withState(RuntimeBindingRecord.State.DRAINING,
+                .withState(lost ? RuntimeBindingRecord.State.LOST : RuntimeBindingRecord.State.DRAINING,
                         claimed.getLease(), clock.instant()));
         if (draining == null) {
             releaseOperationQuietly(claimed.getBindingId(), claimed.getOperationGeneration());

diff --git a/...ime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/RuntimeHarnessDrainTest.java b/...ime-broker/src/test/java/com/alibaba/qwen/code/runtimebroker/RuntimeHarnessDrainTest.java
@@ -183,12 +183,12 @@ void stalledDrainReleasesItsClaimAndFencesLateCompletion(String step) throws Exc
     }
 
     @ParameterizedTest
-    @ValueSource(booleans = {false, true})
-    void drainingBlocksRivalStoragePlacementUntilStopIsProven(boolean jdbc) throws Exception {
+    @CsvSource({"false,RECOVERY_BLOCKED", "true,RECOVERY_BLOCKED", "false,LOST", "true,LOST"})
+    void drainingBlocksRivalStoragePlacementUntilStopIsProven(boolean jdbc, String state) throws Exception {
         RuntimeBindingRepository registry = jdbc ? bindings : new InMemoryRuntimeBindingRepository();
         var ready = ready(registry);
         var claimed = registry.claimOperation(ready.getBindingId(), "block", Duration.ofSeconds(10));
-        var blocked = registry.compareAndSet(claimed, claimed.withState(RuntimeBindingRecord.State.RECOVERY_BLOCKED,
+        var blocked = registry.compareAndSet(claimed, claimed.withState(RuntimeBindingRecord.State.valueOf(state),
                 claimed.getLease(), Instant.now()));
         assertNotNull(blocked);
         var binding = registry.releaseOperation(blocked.getBindingId(), "block", blocked.getOperationGeneration());
@@ -203,7 +203,8 @@ void drainingBlocksRivalStoragePlacementUntilStopIsProven(boolean jdbc) throws E
                 provisioner, transport, registry, sessions, executions, "drainer", Duration.ofSeconds(2), Duration.ofSeconds(2))) {
             service.requestHarnessDrain("tenant", "harness");
             var close = service.drainHarnessSession("tenant", "harness").toCompletableFuture();
-            assertEquals(RuntimeBindingRecord.State.DRAINING, registry.findById(binding.getBindingId()).getState());
+            assertEquals("LOST".equals(state) ? RuntimeBindingRecord.State.LOST : RuntimeBindingRecord.State.DRAINING,
+                    registry.findById(binding.getBindingId()).getState());
             assertEquals("runtime_placement_recovery_required",
                     assertThrows(RuntimeBrokerException.class, () -> registry.findOrCreate(rival)).getCode());
             registry.findOrCreate(new RuntimeProvisionRequest(rivalScope, "unrelated", "local-process", "other-storage"));
@@ -266,6 +267,28 @@ void oldProvisioningReplyCannotLaunchAfterAnotherBrokerRetiresItsIntent() throws
         }
     }
 
+    @Test
+    void lostBindingWithAnUnreleasedSessionStillRequiresRecovery() throws Exception {
+        var binding = ready();
+        var session = bindings.admitSession(sessions, candidate(binding, "one"));
+        var claimed = bindings.claimOperation(binding.getBindingId(), "setup", Duration.ofSeconds(10));
+        var lost = bindings.compareAndSet(claimed, claimed.withState(RuntimeBindingRecord.State.LOST,
+                claimed.getLease(), Instant.now()));
+        bindings.releaseOperation(lost.getBindingId(), "setup", lost.getOperationGeneration());
+        try (var service = service()) {
+            service.requestHarnessDrain("tenant", "harness");
+            var failure = assertThrows(ExecutionException.class,
+                    () -> service.drainHarnessSession("tenant", "harness").toCompletableFuture().get());
+            assertEquals("workspace_close_execution_unsettled",
+                    assertInstanceOf(RuntimeBrokerException.class, failure.getCause()).getCode());
+            assertTrue(sessions.findById(scope, session.getRuntimeSessionId()).isActive());
+            assertEquals(RuntimeBindingRecord.State.LOST, bindings.findById(binding.getBindingId()).getState());
+            assertNull(bindings.findById(binding.getBindingId()).getDrainReceipt());
+            assertEquals(0, transport.releases);
+            assertEquals(0, provisioner.stops);
+        }
+    }
+
     @Test
     void unknownExecutionBlocksReleaseAndStop() throws Exception {
         var binding = ready();