fix(managed-agent): settle a title the dispatcher claimed first (#16447)

* 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 <[email protected]>

* 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 <[email protected]>

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<PublicSession> unarchiveSession(

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();

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