From e61ad3959e5851885f8eb149fb030fd42abe85b1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=93=E8=89=AF?= <1204183885@qq.com> Date: Fri, 6 Nov 2026 21:15:12 +0800 Subject: [PATCH] fix(managed-agent): settle a title the dispatcher claimed first (#16447) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(managed-agent): settle a title the dispatcher claimed first renameSession admits the RENAME row and then claims it. The background SessionRenameCoordinator scans rows a request has just admitted, so its claim can win without any concurrent caller, and the request answered 503 "The Session title is already being delivered." for its own rename. Wait a bounded 2s for that delivery to settle and answer the recorded command instead of failing the request. Fixes #16431 Co-authored-by: Qwen-Coder * fix(managed-agent): keep the creator gate on the rename settle arm The settle arm answered the recorded rename outcome without re-running the replay actor gate, although it shares the COMPLETED read-back's TOCTOU window: a pre-read that missed the recorder's row lets a non-creator OPERATOR reach the settle through the duplicate-key winner path and take a 200 the COMPLETED arm refuses with 409. Mirror that arm's gates — requireReplayActor on the settled Session, and the last-visible serialization — so both arms answer identically. Test: a non-creator OPERATOR racing the recorder's key through the settle arm meets the same 409 workspace_unavailable as through the COMPLETED arm, while the recorder still replays its own outcome; the dispatcher's completion is asserted so the refusal provably comes from the settle arm. Load-bearing: reverting the gate answers the raced request 200 and fails the test. --------- Co-authored-by: Qwen-Coder --- .../session/ManagedAgentService.java | 53 +++++++ .../ManagedSessionLifecycleTest.java | 33 ++++ .../ManagedWorkspaceAdmissionTest.java | 148 ++++++++++++++++++ 3 files changed, 234 insertions(+) diff --git a/packages/managed-agent-server/control-plane/src/main/java/com/alibaba/qwen/code/managedagent/session/ManagedAgentService.java b/packages/managed-agent-server/control-plane/src/main/java/com/alibaba/qwen/code/managedagent/session/ManagedAgentService.java index 1969ad7cd0f..e77862f4507 100644 --- a/packages/managed-agent-server/control-plane/src/main/java/com/alibaba/qwen/code/managedagent/session/ManagedAgentService.java +++ b/packages/managed-agent-server/control-plane/src/main/java/com/alibaba/qwen/code/managedagent/session/ManagedAgentService.java @@ -103,6 +103,13 @@ public class ManagedAgentService implements com.alibaba.qwen.code.managedagent.c private static final Pattern TURN_CURSOR = Pattern.compile( "^(0|[1-9][0-9]{0,18}):([A-Za-z0-9_-]{1,64})$"); private static final int TURN_ID_MAX_LENGTH = 64; + /** + * How long a rename waits for a delivery the background coordinator + * claimed first. The coordinator scans {@code RENAME} rows a request has + * just admitted, so its claim can beat this request's own without any + * concurrent caller. + */ + private static final java.time.Duration RENAME_CLAIM_SETTLE = java.time.Duration.ofSeconds(2); // Catch-up reads of a stream use pages of this size. static final int STREAM_PAGE = 100; // "text" is the spelling that clients used before the contract. @@ -922,6 +929,23 @@ public class ManagedAgentService implements com.alibaba.qwen.code.managedagent.c idempotencyKey, effectiveTitle, command.renameRevision(), 0); var delivery = store.claimRename(candidate, owner, renameLeaseDuration.get()).orElse(null); if (delivery == null) { + // The dispatcher can claim the row this request just admitted. + // Wait for that delivery to settle so one request never + // answers "already being delivered" for its own rename. + if (awaitRenameSettle(tenantId, idempotencyKey, sessionId, + renameDigest(sessionId, title)) != null) { + // The settle arm answers a recorded outcome through the + // same TOCTOU window as the COMPLETED read-back below: + // the pre-read above could not see a key the recorder + // committed after it, so the creator gate re-runs here, + // and the body serializes the Session as last visible. + SessionRecord settled = store.requireSession(tenantId, + sessionId); + requireReplayActor(tenantId, RENAME, idempotencyKey, + settled, actorId); + return new SessionMutationResult<>(publicSession( + asLastVisible(settled)), true); + } throw dependencyUnavailable("hosted_harness_unavailable", "The Session title is already being delivered."); } @@ -987,6 +1011,35 @@ public class ManagedAgentService implements com.alibaba.qwen.code.managedagent.c asLastVisible(replayed)), true); } + /** + * Waits a bounded time for the delivery that won this rename's claim, + * and answers the recorded command once it completes. Returns null when + * the row is gone, the digest belongs to another request, or the + * delivery is still in flight when the wait expires. + */ + private StoreModels.CommandRecord awaitRenameSettle(String tenantId, String idempotencyKey, + String sessionId, String digest) { + long deadline = System.nanoTime() + RENAME_CLAIM_SETTLE.toNanos(); + while (System.nanoTime() < deadline) { + StoreModels.CommandRecord recorded = store + .findCommand(tenantId, RENAME, idempotencyKey).orElse(null); + if (recorded == null || !recorded.sessionId().equals(sessionId) + || !digest.equals(recorded.requestDigest())) { + return null; + } + if ("COMPLETED".equals(recorded.status())) { + return recorded; + } + try { + Thread.sleep(25); + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); + return null; + } + } + return null; + } + // An archived Session stays closed, so unarchive needs neither the // Harness nor the Runtime. public SessionMutationResult unarchiveSession( diff --git a/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionLifecycleTest.java b/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionLifecycleTest.java index 8c8c5bd1e45..5ae00b06908 100644 --- a/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionLifecycleTest.java +++ b/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedSessionLifecycleTest.java @@ -616,6 +616,39 @@ class ManagedSessionLifecycleTest { } } + @Test + void answersTheTitleTheDispatcherDeliveredFirst() throws Exception { + String tenant = tenant(); + String sessionId = attachedSession(tenant); + boolean[] raced = {false}; + AgentStateStore racing = mock(AgentStateStore.class, delegatesTo(store)); + doAnswer(call -> { + // The dispatcher wins the claim the request just admitted, so the + // request's own claim finds nothing and has to wait for the + // delivery instead of answering "already being delivered". + if (raced[0]) { + return java.util.Optional.empty(); + } + raced[0] = true; + var candidate = (com.alibaba.qwen.code.managedagent.model.StoreModels.RenameDelivery) call + .getArgument(0); + var delivery = store.claimRename(candidate, "dispatcher", Duration.ofMinutes(1)) + .orElseThrow(); + store.markRenameSubmitted(delivery, "dispatcher"); + store.completeSessionRename(delivery, "dispatcher", FixtureHarness.BOOT_ID); + return java.util.Optional.empty(); + }).when(racing).claimRename(any(), anyString(), any()); + ManagedAgentService subject = new ManagedAgentService(racing, + new RequestDigests(), null, harness, null, new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules()); + + var result = subject.renameSession(tenant, null, "dispatcher-race", sessionId, "renamed"); + + assertThat(result.body().metadata().get("title")).isEqualTo("renamed"); + assertThat(result.replayed()).isTrue(); + assertThat(store.findCommand(tenant, "RENAME_SESSION", "dispatcher-race").orElseThrow().status()) + .isEqualTo("COMPLETED"); + } + @Test void expiredTitleOwnerCannotRetireTheReplacementOwnersReceipt() throws Exception { String tenant = tenant(); diff --git a/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedWorkspaceAdmissionTest.java b/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedWorkspaceAdmissionTest.java index f57edf0be4e..a2610a97ed1 100644 --- a/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedWorkspaceAdmissionTest.java +++ b/packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/ManagedWorkspaceAdmissionTest.java @@ -1070,6 +1070,154 @@ class ManagedWorkspaceAdmissionTest { .isEqualTo("session_not_found"))); } + // The settle arm answers a recorded outcome through the same TOCTOU + // window as the COMPLETED read-back below it: a non-creator OPERATOR + // whose pre-read missed the recorder's row must meet the same creator + // gate there, never a 200 the COMPLETED arm refuses with 409. + @Test + void theSettleArmKeepsTheCreatorGateForANonCreatorOperator() { + String tenant = "tenant-" + UUID.randomUUID(); + register(tenant, "ws-a", "storage-a", + WorkspaceExecutionProfile.CONFIG_REF, + WorkspaceExecutionProfile.POLICY_REF); + grant(tenant, "ws-a", "actor-a", true); + grant(tenant, "ws-a", "actor-b", true); + String sessionId = store.insertWorkspaceSessionCommand(SessionCreationCommand.builder() + .tenantId(tenant) + .actorId("actor-a") + .idempotencyKey("create") + .requestDigest("sha256:" + "a".repeat(64)) + .agentId("qwen-code") + .selection(new WorkspaceSelection("ws-a", + ".")) + .build()) + .sessionId(); + ManagedAgentProperties enabled = new ManagedAgentProperties(); + enabled.getHarness().setWorkspaceFilesEnabled(true); + ManagedAgentStore gated = com.alibaba.qwen.code.managedagent.VerificationFixtures.managedAgentStore(jdbc, mapper, + Clock.systemUTC(), ignored -> {}, registry, enabled, new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules()); + UnavailableHarnessConnector harness = + new UnavailableHarnessConnector() { + @Override + public boolean isAvailable() { + return true; + } + + @Override + public boolean isWorkspaceFilesAvailable() { + return true; + } + + @Override + public Attachment createOrLoad(String tenantId, + String sessionId, boolean loadExisting) { + return new Attachment("boot"); + } + + @Override + public boolean supportsFencedTitles() { + return true; + } + + @Override + public void rename(String tenantId, String sessionId, + String title) { + } + }; + // actor-a admits the rename; the row stays PENDING until the + // dispatcher delivers it inside actor-b's raced call below. + String digest = new RequestDigests().digest(java.util.Map.of( + "sessionId", sessionId, "title", "new title")); + var admitted = gated.beginSessionRename(tenant, "rename", digest, + sessionId, "new title", "actor-a"); + com.alibaba.qwen.code.managedagent.contracts.session.AgentStateStore racing = + mock(com.alibaba.qwen.code.managedagent.contracts.session.AgentStateStore.class, + org.mockito.AdditionalAnswers.delegatesTo(gated)); + // actor-b's pre-read misses the row the recorder committed after + // it in the raced timeline; every later read sees the row. + java.util.concurrent.atomic.AtomicBoolean preRead = + new java.util.concurrent.atomic.AtomicBoolean(true); + org.mockito.Mockito.doAnswer(call -> { + if (preRead.compareAndSet(true, false)) { + return java.util.Optional.empty(); + } + return gated.findCommand(call.getArgument(0), + call.getArgument(1), call.getArgument(2)); + }).when(racing).findCommand( + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString()); + // The raced admission answers exactly what the store's + // duplicate-key winner path answers when the recorder's receipt + // commits between the request's snapshot pre-read and its receipt + // insert: the winner's PENDING row, replayed, no actor gate on that + // path. Simulating the post-race answer here keeps the test + // deterministic; the entry gate's own refusal of a visible + // recorder row is covered by the store-level replay tests. + org.mockito.Mockito.doAnswer(call -> + new com.alibaba.qwen.code.managedagent.model.StoreModels.SessionMutationCommand( + sessionId, "PENDING", true, + admitted.renameRevision())) + .when(racing).beginSessionRename( + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.anyString()); + // The dispatcher wins the claim actor-b's request just admitted + // and completes the delivery while the request waits. + org.mockito.Mockito.doAnswer(call -> { + var candidate = (com.alibaba.qwen.code.managedagent.model.StoreModels.RenameDelivery) call + .getArgument(0); + var delivery = gated.claimRename(candidate, "dispatcher", + Duration.ofMinutes(1)).orElseThrow(); + gated.markRenameSubmitted(delivery, "dispatcher"); + gated.completeSessionRename(delivery, "dispatcher", "boot"); + return java.util.Optional.empty(); + }).when(racing).claimRename( + org.mockito.ArgumentMatchers.any(), + org.mockito.ArgumentMatchers.anyString(), + org.mockito.ArgumentMatchers.any()); + ManagedAgentService settleService = new ManagedAgentService(racing, + new RequestDigests(), mock(HarnessCoordinator.class), harness, + registry, new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules()); + TransactionTemplate transaction = new TransactionTemplate( + transactionManager); + // Through the settle arm: the live-Session creator refusal, not a + // 200 carrying the recorder's outcome. + transaction.executeWithoutResult(status -> + assertThatThrownBy(() -> settleService.renameSession(tenant, + "actor-b", "rename", sessionId, "new title")) + .isInstanceOfSatisfying(ManagedAgentException.class, error -> { + assertThat(error.getStatus()).isEqualTo(409); + assertThat(error.getCode()) + .isEqualTo("workspace_unavailable"); + })); + // The refusal must come from the settle arm's creator gate, not an + // earlier admission gate: the dispatcher's delivery completed, so + // the request reached the claim and waited for the settle. + assertThat(store.findCommand(tenant, "RENAME_SESSION", "rename") + .orElseThrow().status()).isEqualTo("COMPLETED"); + // The COMPLETED arm answers the identical request identically. + transaction.executeWithoutResult(status -> + assertThatThrownBy(() -> settleService.renameSession(tenant, + "actor-b", "rename", sessionId, "new title")) + .isInstanceOfSatisfying(ManagedAgentException.class, error -> { + assertThat(error.getStatus()).isEqualTo(409); + assertThat(error.getCode()) + .isEqualTo("workspace_unavailable"); + })); + // The recorder still replays its own completed outcome. + transaction.executeWithoutResult(status -> { + var replay = settleService.renameSession(tenant, "actor-a", + "rename", sessionId, "new title"); + assertThat(replay.replayed()).isTrue(); + assertThat(replay.body().metadata().get("title")) + .isEqualTo("new title"); + }); + } + // The grant-less arm of the rename replay gate: with the key recorded // as a completed rename of the Session, every title shape — over-long, // blank, control-bearing — must answer the same 404 a fresh key