fix(managed-agent): fence closed Sessions on the warm path only

The durable lifecycle fence sat in the shared HarnessSessionResolver lambda, so the release and unknown-outcome reconcile routes — which run after a Session closes — hard-refused a CLOSED Session with a non-retryable 409 and leaked the binding the teardown exists to release. Scope the fence to a new admission resolve that only warm() consults; teardown routes resolve the closed Session's scope again. Also defer a poisoned materialization target only when its retry is not due, so a due pass stops writing the progress row twice, and sync the README/design wording with the shipped fence.
2026-11-07 23:01:29 -10:00 · 2026-10-07 03:30:25 +00:00
1 parent b5bf5b42f4
commit 6a671b32cb
@@ -23,7 +23,7 @@ This review targets the current PR #12692 split. The four earlier design pairs w

 1. Only `executions/{id}:resolve` still returns a non-retryable 501; acquire/control/release are live Broker routes. Credentialed with a bearer token, `--no-web` and the capability digest, `qwen serve --profile hosted-harness` boots on loopback and serves the private no-tool Managed Session API. What remains unproven is the durable recovery block signalling named in gate 4, in-flight recovery, complete Workspace lifecycle, and distributed provisioning. The opt-in Workspace file-tool Turn, the HTTP durable-store adapter and the three owner-failover E2E modes run in the hosted-harness-mysql CI job.
 2. Private Store acquisition accepts a caller-selected writer token and tenant. The writer lease fences successive writers but does not authenticate a service. A trusted service identity or enforced private ingress policy must cover these routes independently of browser-facing Agent routes.
-3. Harness-level drain retires an unbound Session through the persisted Session status the Broker resolver reads on every resolve, so the fence survives restarts; an in-process entry is added only for a Session row that has vanished. The fence sits after the workspace-bound branch, so it covers unbound Sessions only. Drain still does not stop the worker or revoke all private Broker access. Archive/delete must not be advertised as Runtime teardown.
+3. Harness-level drain retires an unbound Session through the persisted Session status the Broker resolver checks on the admission resolve (warm), so the fence survives restarts; an in-process entry is added only for a Session row that has vanished. The fence sits after the workspace-bound branch, so it covers unbound Sessions only, and teardown routes — release and the unknown-outcome reconcile — still resolve a closed Session's scope so a binding held across the close settles. Drain still does not stop the worker or revoke all private Broker access. Archive/delete must not be advertised as Runtime teardown.
 4. The HTTP adapter now fails closed with 409 for Broker `UNKNOWN`, but it cannot yet communicate the durable recovery block promised by the design. Recovery needs an explicit compatible unknown-outcome contract.
 5. The new CI wiring must pass remotely before merge. Existing SDK CI, local H2 tests, and a listener's 401 test are not successful Spring/Broker/worker integration evidence.

@@ -23,7 +23,7 @@

 1. 仅 `executions/{id}:resolve` 仍返回不可重试的 501;acquire/control/release 已是真实 Broker 路由。提供 bearer token、`--no-web` 与 capability digest 后，`qwen serve --profile hosted-harness` 可在 loopback 启动并提供私有 no-tool Managed Session API。仍未证明的是门禁 4 所述的持久化恢复阻塞信号、in-flight 恢复、完整 Workspace 生命周期与分布式 provisioning。opt-in 的 Workspace 文件工具 Turn、HTTP durable-store adapter 以及三种 owner-failover E2E 模式均在 hosted-harness-mysql CI 任务中运行。
 2. 私有 Store 的 acquire 接受调用方自行选择的 writer token 和 tenant。writer 租约约束先后写入者，但不认证服务身份。必须为这些路由提供可信服务身份或强制私网入口策略，并与浏览器 Agent 路由区分。
-3. Harness 级 drain 通过 Broker resolver 每次解析时读取的持久化 Session 状态退休未绑定工作区的 Session，栅栏可跨重启存活；仅当 Session 行已消失时才写入进程内条目。栅栏位于 workspace 绑定分支之后，只覆盖未绑定 Session。drain 仍不停止 worker，也不撤销全部私有 Broker 访问。不能把 archive/delete 宣称为 Runtime 回收。
+3. Harness 级 drain 通过 Broker resolver 在准入解析（warm）时检查的持久化 Session 状态退休未绑定工作区的 Session，栅栏可跨重启存活；仅当 Session 行已消失时才写入进程内条目。栅栏位于 workspace 绑定分支之后，只覆盖未绑定 Session；拆除路径——release 与未知结果对账——仍会解析已关闭 Session 的作用域，使跨越关闭仍持有的绑定得以结算。drain 仍不停止 worker，也不撤销全部私有 Broker 访问。不能把 archive/delete 宣称为 Runtime 回收。
 4. HTTP 适配器现在对 Broker `UNKNOWN` 返回 409 并拒绝继续，但尚不能表达设计承诺的持久化恢复阻塞。恢复需要明确且兼容的未知结果契约。
 5. 新增 CI 接线仍需远端运行通过才能合并。既有 SDK CI、本地 H2 测试与监听器的 401 测试不能作为 Spring/Broker/worker 成功集成的证据。

@@ -207,8 +207,10 @@ curl -sS -X DELETE \
 Close and delete reject an active Turn and seal input as soon as they are
 admitted. A background worker then closes the Hosted Harness Session, waits
 until no Harness holds its journal writer under an unexpired lease (the
-holding Harness seals it when closing), drains the Runtime binding (currently
-only an in-process retirement flag) and completes the operation; a failed
+holding Harness seals it when closing), drains the Runtime binding (fenced by
+the persisted Session status the Broker checks before re-admitting an unbound
+Session, so the fence survives restarts; an in-process flag is kept only for a
+Session row that has vanished) and completes the operation; a failed
 attempt is retried with the dispatch backoff until it succeeds, so a `202`
 never means that tools stopped. After the Hosted Harness restarts, its calls fail with a
 generation error until Java restarts too, as Turns do, and the operation waits. A Harness whose journal writes stopped after a failed commit answers every close with `503` until it restarts. A delete of a closed or archived Session
@@ -113,41 +113,58 @@ public class EmbeddedRuntimeBroker implements RuntimeWarmer, AutoCloseable {
        RuntimeProvisioner baseProvisioner = provisioner(broker, http);
        RuntimeProvisioner provisioner = workspaces == null ? baseProvisioner
                : new WorkspaceRuntimeProvisioner(baseProvisioner, workspaces, workspaceExecutionStore);
-        HarnessSessionResolver resolver = sessionId -> {
-            SessionRecord session = store.findSessionById(sessionId)
-                    .orElse(null);
-            if (session == null) {
-                CompletableFuture<RuntimeScope> failed =
-                        new CompletableFuture<>();
-                failed.completeExceptionally(new IllegalArgumentException(
-                        "Session is not owned by this service"));
-                return failed;
+        HarnessSessionResolver resolver = new HarnessSessionResolver() {
+            @Override
+            public CompletionStage<RuntimeScope> resolve(String sessionId) {
+                return resolveScope(sessionId, false);
            }
-            if (session.workspace() != null) {
-                if (workspaces != null) {
-                    return CompletableFuture.completedFuture(workspaces.resolve(sessionId).scope());
+
+            @Override
+            public CompletionStage<RuntimeScope> resolveAdmission(
+                    String sessionId) {
+                return resolveScope(sessionId, true);
+            }
+
+            private CompletionStage<RuntimeScope> resolveScope(
+                    String sessionId, boolean admission) {
+                SessionRecord session = store.findSessionById(sessionId)
+                        .orElse(null);
+                if (session == null) {
+                    CompletableFuture<RuntimeScope> failed =
+                            new CompletableFuture<>();
+                    failed.completeExceptionally(new IllegalArgumentException(
+                            "Session is not owned by this service"));
+                    return failed;
                }
-                return CompletableFuture.failedFuture(
-                        new RuntimeBrokerException(409,
-                                "workspace_unavailable",
-                                "Hosted Workspace execution is not available.",
-                                false));
+                if (session.workspace() != null) {
+                    if (workspaces != null) {
+                        return CompletableFuture.completedFuture(workspaces.resolve(sessionId).scope());
+                    }
+                    return CompletableFuture.failedFuture(
+                            new RuntimeBrokerException(409,
+                                    "workspace_unavailable",
+                                    "Hosted Workspace execution is not available.",
+                                    false));
+                }
+                // The durable row is the admission fence for closing/closed,
+                // archived and deleted Sessions: unlike the in-process
+                // retired set it survives restarts and never accumulates in
+                // memory. Only the admission resolve fences: release and the
+                // unknown-outcome reconcile run after a close and must still
+                // resolve the scope to settle the binding.
+                if (admission && LIFECYCLE_FENCED.contains(session.status())) {
+                    return CompletableFuture.failedFuture(
+                            new RuntimeBrokerException(409,
+                                    "runtime_broker_session_closed",
+                                    "Harness Session is closed.", false));
+                }
+                return CompletableFuture.completedFuture(new RuntimeScope(
+                        session.tenantId(), workspaceId,
+                        broker.getWorkspaceGeneration(),
+                        workspaceCwd,
+                        properties.getHarness().getCapabilityDigest(),
+                        broker.getIsolationClass()));
            }
-            // The durable row is the fence for closing/closed, archived and
-            // deleted Sessions: unlike the in-process retired set it survives
-            // restarts and never accumulates in memory.
-            if (LIFECYCLE_FENCED.contains(session.status())) {
-                return CompletableFuture.failedFuture(
-                        new RuntimeBrokerException(409,
-                                "runtime_broker_session_closed",
-                                "Harness Session is closed.", false));
-            }
-            return CompletableFuture.completedFuture(new RuntimeScope(
-                    session.tenantId(), workspaceId,
-                    broker.getWorkspaceGeneration(),
-                    workspaceCwd,
-                    properties.getHarness().getCapabilityDigest(),
-                    broker.getIsolationClass()));
        };
        ObjectMapper mapper = new ObjectMapper();
        RuntimePublicationVerifier verifier = publications == null || publicationData == null
@@ -227,8 +244,8 @@ public class EmbeddedRuntimeBroker implements RuntimeWarmer, AutoCloseable {
    @Override
    public CompletionStage<Void> drain(String sessionId) {
        // drain() runs while the row still reads CLOSING/DELETING; the
-        // resolver fences every lifecycle status durably, so only a vanished
-        // row needs the in-process entry.
+        // admission resolve fences every lifecycle status durably, so only
+        // a vanished row needs the in-process entry.
        if (store.findSessionById(sessionId).isEmpty()) {
            retired.add(sessionId);
        }
@@ -36,11 +36,11 @@ public class MessageMaterializer {
            String key = target.tenantId() + ":" + target.sessionId();
            int streak = failures.getOrDefault(key, 0);
            if (streak > 0) {
-                deferQuietly(target);
                boolean due = streak < MAX_BACKOFF_STREAK
                        ? (streak & (streak - 1)) == 0
                        : streak % MAX_BACKOFF_STREAK == 0;
                if (!due) {
+                    deferQuietly(target);
                    failures.put(key, streak + 1);
                    continue;
                }
@@ -14,10 +14,18 @@ import com.alibaba.qwen.code.runtimebroker.InMemoryRuntimeBindingRepository;
 import com.alibaba.qwen.code.runtimebroker.InMemoryRuntimeSessionRepository;
 import com.alibaba.qwen.code.runtimebroker.InMemoryToolExecutionRepository;
 import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
+import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerService;
+import com.alibaba.qwen.code.runtimebroker.RuntimeScope;
+import com.alibaba.qwen.code.runtimebroker.RuntimeSession;
+import com.alibaba.qwen.code.runtimebroker.RuntimeSessionRecord;
+import com.alibaba.qwen.code.runtimebroker.ToolExecutionRecord;
 import com.alibaba.qwen.code.runtimebroker.managedworkspace.ContextBinding;
 import java.net.HttpURLConnection;
 import java.net.URI;
 import java.nio.file.Path;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Map;
 import java.util.Optional;
 import org.junit.jupiter.api.Test;

@@ -299,6 +307,86 @@ class EmbeddedRuntimeBrokerTest {
        }
    }

+    @Test
+    void releaseStillSettlesAnUnboundClosedSession() throws Exception {
+        ManagedAgentStore store = mock(ManagedAgentStore.class);
+        when(store.findSessionById(SESSION_ID)).thenReturn(
+                Optional.of(new SessionRecord("tenant-a", SESSION_ID,
+                        "qwen-code", null, "CLOSED", null, null, 0, 0, 1,
+                        1, null, 0)));
+        ManagedAgentProperties properties = properties();
+        InMemoryRuntimeSessionRepository sessions =
+                new InMemoryRuntimeSessionRepository();
+        seedReleasedSession(sessions, properties);
+        try (EmbeddedRuntimeBroker broker = new EmbeddedRuntimeBroker(store,
+                properties, new InMemoryRuntimeBindingRepository(), sessions,
+                new InMemoryToolExecutionRepository())) {
+            // The admission fence refuses new work on a closed Session, but
+            // release is a teardown route: it must still resolve the
+            // Session's scope so a binding held across the close settles.
+            HttpURLConnection connection = (HttpURLConnection) broker
+                    .getBaseUri().resolve("/internal/runtime-broker/v1/"
+                            + "tool-sessions/" + RUNTIME_ID + ":release")
+                    .toURL().openConnection();
+            connection.setRequestMethod("POST");
+            connection.setRequestProperty("Authorization",
+                    "Bearer broker-token");
+            connection.setDoOutput(true);
+            connection.getOutputStream().write(("{\"protocolVersion\":1,"
+                    + "\"requestId\":\"req-release\","
+                    + "\"harnessSessionId\":\"" + SESSION_ID + "\"}")
+                    .getBytes(java.nio.charset.StandardCharsets.UTF_8));
+            assertThat(connection.getResponseCode()).isEqualTo(200);
+            assertThat(new String(connection.getInputStream().readAllBytes(),
+                    java.nio.charset.StandardCharsets.UTF_8))
+                    .contains("\"released\":true");
+            connection.disconnect();
+        }
+    }
+
+    @Test
+    void reconcileStillAnswersAnUnknownExecutionOfAClosedSession()
+            throws Exception {
+        ManagedAgentStore store = mock(ManagedAgentStore.class);
+        when(store.findSessionById(SESSION_ID)).thenReturn(
+                Optional.of(new SessionRecord("tenant-a", SESSION_ID,
+                        "qwen-code", null, "CLOSED", null, null, 0, 0, 1,
+                        1, null, 0)));
+        ManagedAgentProperties properties = properties();
+        InMemoryRuntimeSessionRepository sessions =
+                new InMemoryRuntimeSessionRepository();
+        seedReleasedSession(sessions, properties);
+        InMemoryToolExecutionRepository executions =
+                new InMemoryToolExecutionRepository();
+        String digest = "sha256:" + "b".repeat(64);
+        ToolExecutionRecord prepared = ToolExecutionRecord.prepared("call-1",
+                "idem-1", "binding-1", 1, SESSION_ID, RUNTIME_ID, "turn-1",
+                "tool-call-1", digest,
+                Map.of("sessionId", RUNTIME_ID, "promptId", "turn-1",
+                        "callId", "tool-call-1", "argsDigest", digest));
+        executions.findOrCreate(prepared);
+        ToolExecutionRecord claimed = executions.claimDispatch("call-1",
+                "owner-1", Duration.ofMinutes(5));
+        assertThat(executions.compareAndSet(claimed, claimed.withUnknown(),
+                "owner-1", claimed.getDispatchGeneration())).isNotNull();
+        try (EmbeddedRuntimeBroker broker = new EmbeddedRuntimeBroker(store,
+                properties, new InMemoryRuntimeBindingRepository(), sessions,
+                executions)) {
+            // The reconcile of an UNKNOWN outcome is evidence work after the
+            // close: the fence must not answer it, so the path reaches the
+            // binding check and reports the retired generation.
+            RuntimeBrokerService service = serviceOf(broker);
+            assertThatThrownBy(() -> service.reconcileExecution(SESSION_ID,
+                    RUNTIME_ID, "call-1").toCompletableFuture().join())
+                    .hasCauseInstanceOf(RuntimeBrokerException.class)
+                    .satisfies(error -> assertThat(
+                            ((RuntimeBrokerException) error.getCause())
+                                    .getCode())
+                            .isEqualTo(
+                                    "runtime_execution_evidence_unavailable"));
+        }
+    }
+
    @Test
    void namesTheSupportedProvisionersInTheKubernetesRejection()
            throws Exception {
@@ -361,6 +449,37 @@ class EmbeddedRuntimeBrokerTest {
        }
    }

+    private static final String RUNTIME_ID =
+            "550e8400-e29b-41d4-a716-446655440001";
+
+    private static void seedReleasedSession(
+            InMemoryRuntimeSessionRepository sessions,
+            ManagedAgentProperties properties) {
+        ManagedAgentProperties.RuntimeBroker broker =
+                properties.getRuntimeBroker();
+        RuntimeScope scope = new RuntimeScope("tenant-a",
+                broker.getWorkspaceId(), broker.getWorkspaceGeneration(),
+                broker.getWorkspaceCwd(),
+                properties.getHarness().getCapabilityDigest(),
+                broker.getIsolationClass());
+        RuntimeSessionRecord acquiring = new RuntimeSessionRecord(
+                new RuntimeSession(SESSION_ID, RUNTIME_ID, "bootstrap",
+                        scope),
+                "binding-1", 1, RuntimeSessionRecord.State.ACQUIRING, 0,
+                Instant.now());
+        sessions.findOrCreate(acquiring);
+        sessions.compareAndSet(acquiring, acquiring.withState(
+                RuntimeSessionRecord.State.RELEASED, Instant.now()));
+    }
+
+    private static RuntimeBrokerService serviceOf(EmbeddedRuntimeBroker broker)
+            throws Exception {
+        java.lang.reflect.Field field = EmbeddedRuntimeBroker.class
+                .getDeclaredField("service");
+        field.setAccessible(true);
+        return (RuntimeBrokerService) field.get(broker);
+    }
+
    private static ManagedAgentProperties properties() throws Exception {
        ManagedAgentProperties properties = new ManagedAgentProperties();
        properties.getHarness().setCapabilityDigest("sha256:"
@@ -94,10 +94,10 @@ class MessageMaterializerTest {
        logged.start();
        logger.addAppender(logged);
        try {
-            // Pass one poisons the target; pass two exercises the streaked
-            // pre-attempt deferral and the catch-path deferral, both
-            // throwing. Catching the escape keeps the verdict on the
-            // assertions below rather than on which call threw.
+            // Pass one poisons the target; pass two retries it (streak 1 is
+            // due) and exercises the catch-path deferral, both throwing.
+            // Catching the escape keeps the verdict on the assertions below
+            // rather than on which call threw.
            try {
                materializer.materialize();
                materializer.materialize();
@@ -138,10 +138,11 @@ class MessageMaterializerTest {
        materializer.materialize();

        // fail, retry-and-succeed, then a normal pass: full attempts, and
-        // the deferred rotation left the row alone once healthy.
+        // the only rotation is the first failure's catch-path deferral — a
+        // due retry does not write the progress row ahead of the attempt.
        verify(store, times(3)).materializeNextBatch("tenant", "poison",
                200);
-        verify(store, times(2)).deferMaterializationTarget("tenant",
+        verify(store, times(1)).deferMaterializationTarget("tenant",
                "poison");
    }
 }
@@ -11,4 +11,15 @@ import java.util.concurrent.CompletionStage;
 */
 public interface HarnessSessionResolver {
    CompletionStage<RuntimeScope> resolve(String harnessSessionId);
+
+    /**
+     * Resolves the scope for a route that admits new work (a warm). The
+     * default is a plain resolve; an implementation that fences a closed
+     * Session does so here, so teardown routes such as release and
+     * unknown-outcome reconciliation still resolve that Session's scope.
+     */
+    default CompletionStage<RuntimeScope> resolveAdmission(
+            String harnessSessionId) {
+        return resolve(harnessSessionId);
+    }
 }
@@ -370,7 +370,7 @@ public final class RuntimeBrokerService implements AutoCloseable {
        String harnessId = BrokerValues.requirePathSafe(
                BrokerValues.requireId(harnessSessionId, "harnessSessionId"),
                "harnessSessionId");
-        return resolveScope(harnessId)
+        return resolveScope(harnessId, true)
                .thenCompose(scope -> ensureBinding(
                        provisionRequest(scope, harnessId)))
                .thenApply(BindingContext::record);
@@ -4067,8 +4067,15 @@ public final class RuntimeBrokerService implements AutoCloseable {

    private CompletionStage<RuntimeScope> resolveScope(
            String harnessSessionId) {
+        return resolveScope(harnessSessionId, false);
+    }
+
+    private CompletionStage<RuntimeScope> resolveScope(
+            String harnessSessionId, boolean admission) {
        return mapFailure(safeStage(
-                () -> sessionResolver.resolve(harnessSessionId)),
+                () -> admission
+                        ? sessionResolver.resolveAdmission(harnessSessionId)
+                        : sessionResolver.resolve(harnessSessionId)),
                "runtime_scope_resolution_failed",
                "Runtime scope resolution failed").thenApply(scope -> {
                    if (scope == null) {