perf(managed-agent): take queue dispatch and workspace session creation off the tenant lock (#16327)

startNext locked the tenant placement guard before its Session row, a leftover from when Turn admission took the guard; the Session row now comes first, as in Turn admission. Workspace Session creation now locks the selected Workspace row as its first statement instead of the tenant guard, so creations in different Workspaces of a tenant run at once. Selections that may resolve to a configured Resident Workspace keep the guard, because its single-Session range check shares gaps across Workspaces. The default Workspace is resolved through its generated unique key, so a locking resolution takes one row instead of every Workspace row of the tenant. Co-authored-by: Claude Opus 5.5 <[email protected]>
2026-11-07 06:19:43 -10:00 · 2026-11-05 02:24:56 +00:00
1 parent 4b980c9636
commit 83248491cf
@@ -0,0 +1,118 @@
+# Queue dispatch and Workspace Session creation without the tenant guard
+
+[English](2026-11-04-managed-session-admission-tenant-guard.md) | [简体中文](2026-11-04-managed-session-admission-tenant-guard.zh-CN.md)
+
+## Problem and current state
+
+Turn admission, the Session commands and ordinary journal commits no longer
+take the tenant placement guard
+([Turn admission](2026-11-03-managed-turn-admission-tenant-guard.md),
+[Session commands](2026-11-03-managed-session-commands-tenant-guard.md),
+[journal commit](2026-11-03-managed-journal-commit-tenant-guard.md)). Two
+admissions that run for ordinary traffic still begin with it:
+
+- `ManagedUserMessageStore.startNext` starts the next queued user message as a
+  Turn, once per queued message.
+- `AgentSessionAdmissions.insertWorkspaceSessionCommand` creates a Session in
+  a Workspace. Its comment requires the transaction's first consistent read to
+  follow a lock.
+
+Each therefore queues behind every other guard holder of the tenant.
+
+## Queue dispatch
+
+`startNext` takes the guard, then the Session row (`lock`), reads and updates
+the Session's queue rows, and calls the submitter, which runs Turn admission
+in the same transaction. Turn admission takes no guard since #16223; it locks
+the same Session row and then the storage fence rows. The dispatch to the
+Harness runs after commit.
+
+The guard dates from when Turn admission took it, and kept the order guard,
+Session row. Without it the first statement is the Session row lock, which
+orders the dispatch against lifecycle begin, other queue commands and Turn
+admission of the Session. The queue rows are Session-scoped and written only
+under that row. Remove the guard.
+
+## Workspace Session creation
+
+A new Session has no row to queue on yet. The creation path reads the
+idempotency key with a plain read, then resolves the Workspace with locking
+reads of its `ma_workspace` row and the actor's access row, checks the storage
+fence (alias and directory rows), and inserts the Session, its stream, its
+command receipts and its creation-mode claim. Plain Session creation without a
+Workspace already runs this insert path without the guard.
+
+| Role of the guard                                | Replacement                                                                                                                                                                                                                                                                                                            |
+| ------------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| First consistent read after a lock               | The first statement locks the selected `ma_workspace` row with the predicate `resolveForCreation` uses, so its later lock is the same row.                                                                                                                                                                             |
+| Same idempotency key from two requests           | Unchanged: the receipt's unique key decides, the creation-mode claim inserts first and re-reads with a lock, and the service answers a `DuplicateKeyException` with the recorded admission.                                                                                                                            |
+| Migration                                        | The storage fence rows are locked before any insert. After locking the alias, migration checks committed work and the complete Session profile inventory on a fresh connection. Migration does not lock `ma_workspace`, so there is no reversed order.                                                                 |
+| Resident Workspace retirement, Workspace removal | They lock the same `ma_workspace` row, so creations in that Workspace queue behind them.                                                                                                                                                                                                                               |
+| Resident Workspace single-Session check          | `ResidentWorkspaces.requireCreation` range-locks the Workspace's open Sessions and its runtime sessions; two adjacent Workspaces' ranges share a gap that both creations insert into. Selections that may resolve to a configured Resident Workspace keep the guard first, decided from configuration before any read. |
+
+Default Workspace resolution retains `tenant_id = ? AND is_default = TRUE`
+under `FOR UPDATE`, a scan of the tenant prefix of the
+`(tenant_id, workspace_id)` key that can lock other Workspace rows of the
+tenant. The proposed `default_tenant = ?` lookup deadlocks on MySQL 8.4 when
+an administrator first updates Workspace metadata and then switches the
+default: the reader holds the generated unique index and waits for the row,
+while the administrator holds the row and waits for that index. Keep the
+existing lookup order; narrowing default-selection locks needs a separate
+design that preserves concurrent default-switch behavior.
+
+Team lead creation and child Session creation keep the guard. Child creation
+takes it again in `ChildWorkspaceStore.admit` and serializes with the parent's
+child run record; both are low frequency.
+
+The first Workspace locking read is owned by `ManagedWorkspaceRegistry`,
+which keeps its selection predicate and arguments private. Session admission
+retains the Resident Workspace guard decision and its lock-order explanation.
+For this change, an explicitly approved per-class ceiling adjustment from
+889 to 900 lines accommodates that necessary logic and explanation. This is
+a scoped exception to the ratchet-down convention; the global ceiling and
+the other per-class ceilings remain unchanged.
+
+Migration can establish a REPEATABLE READ snapshot before waiting for a
+creation's storage alias lock. Its existing fresh autocommit idle check must
+also validate the complete Session profile inventory, including Sessions
+committed during that wait. Otherwise it can miss an uninitialized or
+unsupported Session. Read the inventory, receipts, genesis and definition
+resources through that same fresh connection. The outer transaction holds the
+alias and publication authority locks; the fresh reader must not lock Session
+or head rows and reverse ordinary traffic's Session-before-alias order.
+
+## Risks
+
+- Creations in one Workspace still queue on its row, and every creation still
+  locks the storage's alias row exclusively, so creations sharing one storage
+  queue there. Default selection retains the tenant-prefix scan and can also
+  delay creations in other Workspaces of the tenant.
+- A missing-key locking read in the creation path that the audit missed would
+  show up as an InnoDB deadlock between two creations of one tenant. The MySQL
+  test below targets it.
+- `mayContain` is conservative: a tenant that configures any Resident
+  Workspace keeps the guard for its default-Workspace creations.
+
+## Verification and acceptance
+
+1. `ManagedUserMessageStoreTest`: with the tenant placement guard held on
+   another connection, `startNext` claims the queued message and admits its
+   Turn.
+2. `ManagedModelSelectionTest`: with the guard held, a Workspace Session
+   creation completes.
+3. `ManagedSessionGuardConvoyIT` (MySQL): four writers create ten Sessions
+   each in one tenant at once, with no deadlock. Cover explicit selections
+   across four Workspaces and selection of the default Workspace, each with
+   separate storage registrations and with shared storage. Default selections
+   must bind every Session to the default Workspace. The existing MySQL
+   default-switch test must resolve the newly selected default without a deadlock.
+4. `WorkspaceMigrationMySqlIT`: creation and legacy CWD settlement wait on an
+   uncommitted `ma_storage` fence, reject it after commit, preserve original
+   request replay and allow the unrelated tenant to progress. A Session
+   committed after migration's REPEATABLE READ snapshot must still be checked:
+   an uninitialized Session rejects migration without leaving an operation or
+   fence.
+5. Existing queue, creation, Workspace, migration and lock contract suites
+   pass, including the MySQL profiles in CI.
+6. The source-size gate accepts the 900-line admission class at its exact
+   approved ceiling and preserves main's 4,253-line broker and global ceilings.
@@ -0,0 +1,89 @@
+# 去掉队列派发与工作区 Session 创建的租户 guard
+
+[English](2026-11-04-managed-session-admission-tenant-guard.md) | [简体中文](2026-11-04-managed-session-admission-tenant-guard.zh-CN.md)
+
+## 问题与现状
+
+Turn 准入、Session 命令和普通 journal commit 都已不再获取租户 placement guard
+（[Turn 准入](2026-11-03-managed-turn-admission-tenant-guard.zh-CN.md)、
+[Session 命令](2026-11-03-managed-session-commands-tenant-guard.zh-CN.md)、
+[journal commit](2026-11-03-managed-journal-commit-tenant-guard.zh-CN.md)）。
+普通流量中还有两个准入以它开头：
+
+- `ManagedUserMessageStore.startNext`：把下一条排队的用户消息作为 Turn 启动，每条排队
+  消息执行一次。
+- `AgentSessionAdmissions.insertWorkspaceSessionCommand`：在工作区中创建 Session。其注释
+  要求事务的第一次一致性读发生在某个锁之后。
+
+因此它们都要排在该租户的其他 guard 持有者之后。
+
+## 队列派发
+
+`startNext` 先获取 guard，再锁 Session 行（`lock`），读写该 Session 的队列行，然后调用
+submitter；submitter 在同一事务中执行 Turn 准入。自 #16223 起 Turn 准入不再获取 guard，
+它锁定同一 Session 行，再锁存储 fence 行。向 Harness 派发在事务提交后执行。
+
+这个 guard 来自 Turn 准入还持有 guard 的时期，用来保持"guard、Session 行"的顺序。去掉后，
+第一条语句就是 Session 行锁，它让派发与该 Session 的生命周期 begin、其他队列命令以及 Turn
+准入排序。队列行按 Session 划分，且只在该行锁下写入。去掉 guard。
+
+## 工作区 Session 创建
+
+新 Session 还没有可以排队的行。创建路径先用普通读查幂等键，再用锁定读解析工作区（其
+`ma_workspace` 行和操作者的 access 行），检查存储 fence（别名行和目录行），然后插入
+Session、事件流、命令回执和创建模式声明。不带工作区的普通 Session 创建，本来就在不持有
+guard 的情况下走这条插入路径。
+
+| guard 的作用                | 替代方式                                                                                                                                                                                                                               |
+| --------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| 第一次一致性读发生在锁之后  | 第一条语句用 `resolveForCreation` 相同的条件锁定所选 `ma_workspace` 行，之后的锁是同一行。                                                                                                                                             |
+| 两个请求使用同一幂等键      | 不变：由回执的唯一键裁决；创建模式声明先插入、再锁定重读；服务层对 `DuplicateKeyException` 返回已记录的准入。                                                                                                                          |
+| 迁移                        | 任何插入之前都会锁定存储 fence 行。迁移锁定别名后，通过新连接检查已提交的工作和完整的 Session 执行配置清单。迁移不锁 `ma_workspace`，不存在反向顺序。                                                                                  |
+| 常驻工作区退役、工作区删除  | 它们锁定同一 `ma_workspace` 行，该工作区内的创建排在其后。                                                                                                                                                                             |
+| 常驻工作区的单 Session 检查 | `ResidentWorkspaces.requireCreation` 对工作区内未关闭的 Session 和运行时会话加范围锁；两个相邻工作区的范围共享一个间隙，两次创建都会向其中插入。可能解析到已配置常驻工作区的选择仍先获取 guard，这一判断在任何读取之前仅依据配置完成。 |
+
+默认工作区继续使用 `tenant_id = ? AND is_default = TRUE` 加 `FOR UPDATE` 解析，这会扫描
+`(tenant_id, workspace_id)` 键的租户前缀，可能锁住该租户的其他工作区行。拟采用的
+`default_tenant = ?` 查询在 MySQL 8.4 上会与先更新工作区元数据、再切换默认工作区的
+管理员操作死锁：读取方持有生成列唯一索引并等待数据行，管理员持有数据行并等待该索引。
+因此保留原有查找顺序；缩小默认选择的锁范围需要另行设计，并保持并发切换默认工作区的行为。
+
+团队负责人创建和子 Session 创建保留 guard。子 Session 创建会在 `ChildWorkspaceStore.admit`
+中再次获取 guard，并与父 Session 的子运行记录串行；两者频率都低。
+
+首次工作区锁定读由 `ManagedWorkspaceRegistry` 负责，其选择条件和参数保持私有。
+Session 准入保留常驻工作区 guard 的判断及其锁顺序说明。本次改动经明确批准，将该类
+的行数上限从 889 调整至 900，以容纳必要的逻辑和说明。这是只降不升约定的一次限定
+例外；全局上限和其他类的上限保持不变。
+
+迁移可能在等待创建方的存储别名锁之前就固定 REPEATABLE READ 快照。因此，现有使用
+独立 autocommit 连接的空闲检查，还必须验证完整的 Session 执行配置清单，包括等待
+期间提交的 Session，否则可能漏掉尚未初始化或不支持迁移的 Session。清单、回执、
+genesis 和定义资源都通过同一新连接读取。外层事务继续持有别名锁和 publication
+authority 锁；新连接不能再锁 Session 或 head 行，以免与普通流量先 Session、后
+别名的顺序相反。
+
+## 风险
+
+- 同一工作区内的创建仍在其行上排队；每次创建仍以排他方式锁定存储别名行，共享同一存储的
+  创建会在那里排队。默认选择保留租户前缀扫描，也可能阻塞该租户其他工作区中的创建。
+- 若审计漏掉创建路径中某处对不存在键的锁定读，会表现为同一租户两次创建之间的 InnoDB
+  死锁。下面的 MySQL 测试专门覆盖这种情况。
+- `mayContain` 是保守的：只要租户配置了任一常驻工作区，其默认工作区的创建就仍获取 guard。
+
+## 验证与验收
+
+1. `ManagedUserMessageStoreTest`：另一连接持有租户 placement guard 时，`startNext` 仍能
+   认领排队消息并准入其 Turn。
+2. `ManagedModelSelectionTest`：持有 guard 时，工作区 Session 创建仍能完成。
+3. `ManagedSessionGuardConvoyIT`（MySQL）：四个 writer 同时在同一租户中各创建十个
+   Session，不出现死锁。覆盖显式选择四个工作区和选择默认工作区，并分别使用独立的
+   存储注册和共享存储；默认选择创建的所有 Session 必须绑定到默认工作区。现有 MySQL
+   默认切换测试必须在不发生死锁的情况下解析到新选定的默认工作区。
+4. `WorkspaceMigrationMySqlIT`：创建和旧版 CWD 完成等待尚未提交的 `ma_storage`
+   fence，在其提交后拒绝操作，同时保留原请求的幂等重放，并允许无关租户继续推进。
+   迁移的 REPEATABLE READ 快照固定后提交的 Session 仍必须接受检查；尚未初始化的
+   Session 应使迁移拒绝，且不遗留操作记录或 fence。
+5. 现有的队列、创建、工作区、迁移与锁契约测试继续通过，包括 CI 中的 MySQL profile。
+6. 源码行数门禁应接受 900 行的准入类，精确符合已批准的上限；Broker 类和全局上限
+   保持 main 的 4,253 行。
@@ -25,6 +25,16 @@ public class ResidentWorkspaceCatalog {
        }
    }

+    /**
+     * Whether a selection could resolve to a configured Resident Workspace, decided from
+     * configuration alone before any row is read: true for the tenant's default Workspace
+     * (workspaceId null) whenever the tenant configures one.
+     */
+    public boolean mayContain(String tenantId, String workspaceId) {
+        return configured.stream().anyMatch(value -> value.registration().tenantId().equals(tenantId)
+                && (workspaceId == null || value.workspaceId().equals(workspaceId)));
+    }
+
    public ResidentWorkspace find(ContextBinding binding) {
        if (binding == null) {
            return null;
@@ -160,8 +160,19 @@ final class AgentSessionAdmissions {
        if (!input.isEmpty() && !workspaceFilesEnabled) {
            throw AgentSessionPolicies.workspaceExecutionUnavailable();
        }
-        // InnoDB's first consistent read must follow the migration admission lock.
-        WorkspaceMigrationAdmission.lockTenant(jdbc, tenantId);
+        // InnoDB's first consistent read must follow the creation's first lock.
+        // A selection that may resolve to a Resident Workspace keeps the tenant
+        // guard first: its single-Session range lock can deadlock adjacent
+        // Workspaces' inserts. Other creations lock the selected Workspace
+        // before reading receipts, as resolveForCreation does below. Explicit
+        // selections lock one Workspace; default selections retain the tenant
+        // scan to preserve the lock order with default switches (see design
+        // 2026-11-04-managed-session-admission-tenant-guard).
+        if (residentWorkspaces.mayContain(tenantId,
+                selection == null ? null : selection.workspaceId())) {
+            WorkspaceMigrationAdmission.lockTenant(jdbc, tenantId);
+        }
+        workspaces.lockSelectionForCreation(tenantId, selection);
        List<WorkspaceCommand> existing = findWorkspaceCommand(tenantId,
                actorId, idempotencyKey);
        if (!existing.isEmpty()) {
@@ -192,7 +192,10 @@ public class ManagedUserMessageStore
        if (!agents.supportsUserMessageQueue()) {
            return;
        }
-        ManagedSessionStore.lockBounded(() -> WorkspaceLifecycleStore.lockPlacement(jdbc, tenant, ManagedSessionStore.GUARD_ACQUIRE_TIMEOUT_SECONDS));
+        // The Session row is the first lock, as in Turn admission, which the
+        // submitter below runs in this transaction without the tenant guard:
+        // it orders this dispatch against lifecycle and other admissions of
+        // the Session, and the storage fence is checked by Turn admission.
        lock(tenant, session);
        reconcile(tenant, session);
        Snapshot snapshot = snapshot(tenant, session);
@@ -281,6 +281,23 @@ public class ManagedWorkspaceRegistry
        return resolve(tenantId, actorId, selection, false, false);
    }

+    void lockSelectionForCreation(String tenantId, WorkspaceSelection selection) {
+        jdbc.queryForList("SELECT workspace_id FROM ma_workspace WHERE "
+                + selectionPredicate(selection) + " FOR UPDATE",
+                selectionArguments(tenantId, selection));
+    }
+
+    // Default switches lock the tenant/workspace key before the row. Keep the
+    // same lookup order: default_tenant would lock its secondary index first
+    // and deadlock with a switch that already holds the row.
+    private static String selectionPredicate(WorkspaceSelection selection) {
+        return selection == null ? "tenant_id = ? AND is_default = TRUE" : "tenant_id = ? AND workspace_id = ?";
+    }
+
+    private static Object[] selectionArguments(String tenantId, WorkspaceSelection selection) {
+        return selection == null ? new Object[] {tenantId} : new Object[] {tenantId, selection.workspaceId()};
+    }
+
    private ResolvedBinding resolve(String tenantId, String actorId, WorkspaceSelection selection,
            boolean lock, boolean requireOwner) {
        if (!TransactionSynchronizationManager.isActualTransactionActive()) {
@@ -302,11 +319,11 @@ public class ManagedWorkspaceRegistry
                "SELECT workspace_id, workspace_generation, storage_id, display_name,"
                        + " config_ref, policy_ref, state, workspace_kind FROM"
                        + " ma_workspace WHERE "
-                        + (selection == null ? "tenant_id = ? AND is_default = TRUE" : "tenant_id = ? AND workspace_id = ?")
+                        + selectionPredicate(selection)
                        + (lock ? " FOR UPDATE" : ""),
                (result, row) -> new RegistryRow(workspaceRow(tenantId, result.getString("workspace_id"), result),
                        result.getString("workspace_kind")),
-                selection == null ? new Object[] {tenantId} : new Object[] {tenantId, selection.workspaceId()});
+                selectionArguments(tenantId, selection));
        if (records.isEmpty()) {
            if (selection == null) {
                throw workspaceRequired();
@@ -112,8 +112,7 @@ public final class WorkspaceMigrationStore {
        transaction.executeWithoutResult(status -> {
            lockAuthority();
            guard.migrationSource(tenant, storage, revision, fence);
-            requireCommittedIdle();
-            requireFilesProfiles();
+            requireCommittedAdmissionState();
            var raced = row(true);
            if (raced != null) {
                check(hash(bytes).equals(raced.get("request_digest")), "operation_conflict");
@@ -305,17 +304,17 @@ public final class WorkspaceMigrationStore {
        });
    }

-    private void requireFilesProfiles() {
+    private void requireFilesProfiles(JdbcTemplate current) {
        String after = "";
-        var reader = new WorkspaceRecoveryReader(jdbc, null);
+        var reader = new WorkspaceRecoveryReader(current, null);
        while (true) {
-            var page = jdbc.queryForList("SELECT session_id FROM ma_session WHERE tenant_id = ?"
+            var page = current.queryForList("SELECT session_id FROM ma_session WHERE tenant_id = ?"
                    + " AND workspace_storage_id = ? AND session_id > ? ORDER BY session_id LIMIT 32",
                    String.class, tenant, storage, after);
            for (String session : page) {
-                jdbc.query("SELECT " + ManagedSchemaProjection.SESSION.columns() + " FROM ma_session WHERE tenant_id = ? AND session_id = ?",
+                current.query("SELECT " + ManagedSchemaProjection.SESSION.columns() + " FROM ma_session WHERE tenant_id = ? AND session_id = ?",
                        (row, index) -> ManagedAgentStore.readBinding(row), tenant, session);
-                JsonNode source = WorkspaceRecoveryStore.currentSource(jdbc, tenant, storage, session, true);
+                JsonNode source = WorkspaceRecoveryStore.currentSource(current, tenant, storage, session, false);
                after = session;
                if (source.path("head").isNull() && !source.path("retirement").isNull()) {
                    continue;
@@ -348,19 +347,21 @@ public final class WorkspaceMigrationStore {
        requireIdle(jdbc);
    }

-    // Turn admission does not take the tenant guard; it locks this storage's alias (fence) row
-    // through requireOpen. migrationSource has already locked that row here, so an admission
-    // either committed before this point or waits and then sees the fence. This transaction's
-    // REPEATABLE READ snapshot can predate that lock, so read committed work on a separate
-    // autocommit connection, as LocalStorageOwners.lockCovered does for its inventory.
-    private void requireCommittedIdle() {
+    // Turn and Session creation admission lock this storage's alias before inserting.
+    // migrationSource holds that lock now, but this transaction's RR snapshot may predate it.
+    // Check both work and profiles on a fresh connection. The alias and publication authority
+    // protect genesis and definition resources; locking Session/head rows on this connection
+    // would reverse hot paths' Session-before-alias order.
+    private void requireCommittedAdmissionState() {
        try (var connection = java.util.Objects.requireNonNull(jdbc.getDataSource()).getConnection()) {
            connection.setAutoCommit(true);
            connection.setReadOnly(true);
-            requireIdle(new ManagedSchemaJdbcTemplate(
-                    new org.springframework.jdbc.datasource.SingleConnectionDataSource(connection, true)));
+            var current = new ManagedSchemaJdbcTemplate(
+                    new org.springframework.jdbc.datasource.SingleConnectionDataSource(connection, true));
+            requireIdle(current);
+            requireFilesProfiles(current);
        } catch (java.sql.SQLException error) {
-            throw new IllegalStateException("Migration idle check is unavailable", error);
+            throw new IllegalStateException("Migration admission check is unavailable", error);
        }
    }

@@ -400,6 +400,22 @@ class ManagedModelSelectionTest {
        }
    }

+    @Test
+    void workspaceSessionCreationDoesNotWaitForTheTenantPlacementGuard() throws Exception {
+        try (var holder = java.util.Objects.requireNonNull(jdbc.getDataSource()).getConnection()) {
+            holder.setAutoCommit(false);
+            com.alibaba.qwen.code.runtimebroker.persistence.JdbcRuntimeBindingRepository.lockPlacementDomain(holder, tenant);
+            try {
+                String created = transaction().execute(status -> service.createWorkspaceSession(tenant, "actor",
+                        "create-guarded", "qwen-code", null, null, null, List.of(),
+                        new WorkspaceSelection("ws", "."))).sessionId();
+                assertThat(created).isNotBlank().isNotEqualTo(session);
+            } finally {
+                holder.rollback();
+            }
+        }
+    }
+
    private WebShellModelChangeResult change(String actor, String key, String model, long revision) {
        return transaction().execute(status -> service.changeWebShellSessionModel(tenant, actor, session, key, model, revision));
    }
@@ -126,6 +126,29 @@ class ManagedUserMessageStoreTest {
        assertThat(admission.turnId()).isNotBlank();
    }

+    @Test
+    void startsTheNextQueuedTurnWhileTheTenantPlacementGuardIsHeld() throws Exception {
+        Message first = command("guarded", submit("guarded")).messages().getFirst();
+        try (var holder = java.util.Objects.requireNonNull(jdbc.getDataSource()).getConnection()) {
+            holder.setAutoCommit(false);
+            com.alibaba.qwen.code.runtimebroker.persistence.JdbcRuntimeBindingRepository.lockPlacementDomain(holder, TENANT);
+            try {
+                messages.startNext(TENANT, session, (tenant, actor, key, target, input) -> {
+                    var admission = store.insertTurnCommand(tenant, "queue-test", key, "digest", target,
+                            first.input(), "sha256:" + "0".repeat(64), actor);
+                    return new com.alibaba.qwen.code.managedagent.model.ApiModels.CommandAdmission(
+                            target, admission.turnId(), "ACCEPTED", false);
+                });
+            } finally {
+                holder.rollback();
+            }
+        }
+        assertThat(query().messages()).singleElement().satisfies(message -> {
+            assertThat(message.state()).isEqualTo("claimed");
+            assertThat(message.turnId()).isNotBlank();
+        });
+    }
+
    @Test
    void editsInPlaceAndRejectsStaleRevisionBeforeWithdrawing() {
        Message first = command("first", submit("first")).messages().getFirst();
@@ -28,6 +28,8 @@ import com.fasterxml.jackson.databind.node.ObjectNode;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.springframework.http.HttpStatus;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
@@ -345,6 +347,80 @@ class ManagedSessionGuardConvoyIT {
                Integer.class, tenant, "victim")).isOne();
    }

+    /**
+     * Workspace Session creation orders on the selected Workspace row, not
+     * the tenant guard, so creations in different Workspaces of one tenant
+     * run at once. Each inserts a Session, its stream, its command receipts
+     * and its creation-mode claim into indexes those Workspaces share; a
+     * locking probe of a missing key there would deadlock two of them
+     * (InnoDB 1213). H2 takes no gap locks, so only MySQL shows it.
+     */
+    @ParameterizedTest
+    @CsvSource({"false, false", "false, true", "true, false", "true, true"})
+    void concurrentWorkspaceSessionCreationsOfOneTenantDoNotDeadlock(
+            boolean defaultSelection, boolean sharedStorage) throws Exception {
+        var jdbc = new ManagedSchemaJdbcTemplate(source);
+        var transactions = new TransactionTemplate(new DataSourceTransactionManager(source));
+        String tenant = "tenant-create";
+        int writers = 4;
+        var properties = new ManagedAgentProperties();
+        properties.getHarness().setWorkspaceFilesEnabled(true);
+        var mounts = new ArrayList<ManagedAgentProperties.RuntimeBroker.WorkspaceMount>();
+        for (int index = 0; index < writers; index++) {
+            String storage = sharedStorage ? "storage" : "storage-" + index;
+            jdbc.update("INSERT INTO ma_workspace (tenant_id, workspace_id, workspace_generation, storage_id,"
+                    + " display_name, config_ref, policy_ref, state, is_default)"
+                    + " VALUES (?, ?, 1, ?, 'Workspace', ?, ?, 'ACTIVE', ?)",
+                    tenant, "ws-" + index, storage,
+                    com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile.CONFIG_REF,
+                    com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile.POLICY_REF, index == 0);
+            jdbc.update("INSERT INTO ma_workspace_access (tenant_id, workspace_id, actor_id, role) VALUES (?, ?, ?, 'OWNER')",
+                    tenant, "ws-" + index, "actor".getBytes(StandardCharsets.UTF_8));
+            if (!sharedStorage || index == 0) {
+                mounts.add(new ManagedAgentProperties.RuntimeBroker.WorkspaceMount(tenant, storage, "/tmp/create-" + storage));
+            }
+        }
+        properties.getRuntimeBroker().setWorkspaceMounts(mounts);
+        var registry = new ManagedWorkspaceRegistry(jdbc);
+        var store = com.alibaba.qwen.code.managedagent.VerificationFixtures.managedAgentStore(jdbc, JSON, Clock.systemUTC(),
+                ignored -> {}, registry, properties, new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules());
+        var service = new com.alibaba.qwen.code.managedagent.session.ManagedAgentService(store,
+                new com.alibaba.qwen.code.managedagent.model.RequestDigests(),
+                org.mockito.Mockito.mock(com.alibaba.qwen.code.managedagent.session.HarnessCoordinator.class),
+                org.mockito.Mockito.mock(com.alibaba.qwen.code.managedagent.contracts.harness.HarnessConnector.class),
+                registry, new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules());
+        int perWriter = 10;
+        var start = new CountDownLatch(1);
+        try (var pool = Executors.newFixedThreadPool(writers)) {
+            List<Future<Integer>> runs = new ArrayList<>();
+            for (int writer = 0; writer < writers; writer++) {
+                String workspace = "ws-" + writer;
+                var selection = defaultSelection ? null
+                        : new com.alibaba.qwen.code.managedagent.contracts.workspace.WorkspaceSelection(workspace, ".");
+                runs.add(pool.submit(() -> {
+                    start.await();
+                    for (int index = 0; index < perWriter; index++) {
+                        String key = workspace + "-create-" + index;
+                        transactions.execute(ignored -> service.createWorkspaceSession(tenant, "actor", key,
+                                "qwen-code", null, null, null, List.of(), selection));
+                    }
+                    return perWriter;
+                }));
+            }
+            start.countDown();
+            for (var run : runs) {
+                assertThat(run.get(120, TimeUnit.SECONDS)).isEqualTo(perWriter);
+            }
+        }
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session WHERE tenant_id = ? AND session_kind = 'PUBLIC'",
+                Integer.class, tenant)).isEqualTo(writers * perWriter);
+        if (defaultSelection) {
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session WHERE tenant_id = ?"
+                            + " AND session_kind = 'PUBLIC' AND workspace_id = 'ws-0'",
+                    Integer.class, tenant)).isEqualTo(writers * perWriter);
+        }
+    }
+
    /**
     * Commits of different Sessions in one tenant now run at once. Every
     * revision publishes a new resource; a locking probe of each missing key
@@ -2,14 +2,17 @@ package com.alibaba.qwen.code.managedagent.persistence;

 import com.alibaba.qwen.code.managedagent.MySqlLockProbe;
 import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties;
+import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties.RuntimeBroker.WorkspaceMount;
 import com.alibaba.qwen.code.managedagent.contracts.persistence.ManagedSessionStoreModels;
 import com.alibaba.qwen.code.managedagent.contracts.session.SessionCreationCommand;
+import com.alibaba.qwen.code.managedagent.contracts.workspace.StorageIdentityReader;
 import com.alibaba.qwen.code.managedagent.contracts.workspace.WorkspaceSelection;
 import com.alibaba.qwen.code.managedagent.model.StoreModels;
 import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
 import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
 import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchema;
 import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaProjection;
+import com.alibaba.qwen.code.runtimebroker.persistence.InMemoryRuntimeBindingRepository;
 import com.alibaba.qwen.code.runtimebroker.persistence.JdbcRuntimeBindingRepository;
 import com.alibaba.qwen.code.runtimebroker.schema.ManagedSchemaContract;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -17,20 +20,22 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
-import org.springframework.jdbc.core.ConnectionCallback;
+import org.junit.jupiter.api.io.TempDir;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.datasource.DataSourceTransactionManager;
 import org.springframework.jdbc.datasource.DriverManagerDataSource;
 import org.springframework.transaction.support.TransactionTemplate;

 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.PosixFilePermissions;
 import java.sql.Connection;
 import java.sql.SQLException;
 import java.time.Clock;
 import java.time.Duration;
 import java.util.List;
 import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -40,6 +45,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;

 class WorkspaceMigrationMySqlIT {
+    @TempDir Path temp;
    private JdbcTemplate admin;
    private JdbcTemplate jdbc;
    private DriverManagerDataSource data;
@@ -117,6 +123,53 @@ class WorkspaceMigrationMySqlIT {
        admissionWaitingForMigrationSeesCommittedFenceAndPreservesReplay(true);
    }

+    @Test
+    @Timeout(30)
+    void migrationSeesSessionCommittedAfterItsRepeatableReadSnapshot() throws Exception {
+        ManagedSchema.initialize(data);
+        temp = temp.toRealPath();
+        var source = Files.createDirectory(temp.resolve("source")).toRealPath();
+        var target = Files.createDirectory(temp.resolve("target"));
+        var history = Files.createDirectory(temp.resolve("history"));
+        var runtime = Files.createDirectory(temp.resolve("runtime"),
+                PosixFilePermissions.asFileAttribute(PosixFilePermissions.fromString("rwx------")));
+        var properties = new ManagedAgentProperties();
+        properties.getRuntimeBroker().setVerifiedWorkspaceRecoveryEnabled(true);
+        properties.getRuntimeBroker().setWorkspaceMounts(List.of(new WorkspaceMount("tenant", "storage", source.toString())));
+        var manager = new DataSourceTransactionManager(data);
+        var guard = new WorkspaceStorageGuard(jdbc, manager, properties, path ->
+                new StorageIdentityReader.Identity(path.toRealPath().toString(), "test-host", "device",
+                        path.getFileName().toString(), "2026-10-02T00:00:00Z"));
+        guard.register("tenant", "storage", UUID.randomUUID().toString());
+        var request = new ObjectMapper().createObjectNode().put("version", 1)
+                .put("migrationOperationId", UUID.randomUUID().toString()).put("tenantId", "tenant").put("storageId", "storage")
+                .put("fenceOperationId", UUID.randomUUID().toString()).put("captureOperationId", UUID.randomUUID().toString())
+                .put("mountRevision", 1).put("sourceRoot", source.toString()).put("targetRoot", target.toString())
+                .put("bundleRoot", temp.resolve("bundle").toString()).put("fileHistoryRoot", history.toString())
+                .put("stateDirectory", runtime.toString()).put("nodeExecutable", "/test/node").put("cliEntry", "/test/cli.js");
+        var transaction = new TransactionTemplate(manager);
+        transaction.setIsolationLevel(Connection.TRANSACTION_REPEATABLE_READ);
+        transaction.setTimeout(15);
+        assertThatThrownBy(() -> transaction.executeWithoutResult(status -> {
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session", Long.class)).isZero();
+            try (var creation = data.getConnection()) {
+                creation.setAutoCommit(false);
+                seed(new ManagedSchemaJdbcTemplate(new org.springframework.jdbc.datasource.SingleConnectionDataSource(creation, true)),
+                        "tenant", "late-session", "ACTIVE");
+                creation.commit();
+            } catch (SQLException error) {
+                throw new IllegalStateException(error);
+            }
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session", Long.class)).isZero();
+            new WorkspaceMigrationStore(jdbc, manager, guard, new InMemoryRuntimeBindingRepository(),
+                    request.toString().getBytes(StandardCharsets.UTF_8), true,
+                    new com.alibaba.qwen.code.managedagent.adapters.workspace.LocalRecoveryWorker());
+        })).hasMessageContaining("migration_member_uninitialized");
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session", Long.class)).isOne();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_operation WHERE operation_kind = 'STORAGE_MIGRATION'", Long.class)).isZero();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE fence_operation_id IS NOT NULL", Long.class)).isZero();
+    }
+
    private void admissionWaitingForMigrationSeesCommittedFenceAndPreservesReplay(boolean cwd) throws Exception {
        ManagedSchema.initialize(data);
        for (String tenant : List.of("migrating", "unrelated")) {
@@ -144,49 +197,29 @@ class WorkspaceMigrationMySqlIT {
                "unrelated", other.sessionId(), "actor", "actor-digest", "other", "digest", ".", 1) : null;
        var otherClaim = cwd ? transaction.execute(status -> store.claimOperation("unrelated", other.sessionId(),
                otherCwd.operation().operationId(), "owner", Duration.ofSeconds(60)).orElseThrow()) : null;
-        if (cwd) {
-            // Keep the unrelated probe outside the uncommitted fence insert's index gap.
-            jdbc.update("INSERT INTO ma_storage (alias_key, tenant_key, storage_key, tenant_id, storage_id)"
-                            + " VALUES (?, ?, ?, 'unrelated', 'storage')",
-                    JdbcRuntimeBindingRepository.storageAliasKey("unrelated", "storage"),
-                    JdbcRuntimeBindingRepository.storageFenceKey("unrelated"),
-                    JdbcRuntimeBindingRepository.storageFenceKey("storage"));
-        }
+        // Keep the unrelated probe outside the uncommitted fence insert's index gap.
+        jdbc.update("INSERT INTO ma_storage (alias_key, tenant_key, storage_key, tenant_id, storage_id)"
+                        + " VALUES (?, ?, ?, 'unrelated', 'storage')",
+                JdbcRuntimeBindingRepository.storageAliasKey("unrelated", "storage"),
+                JdbcRuntimeBindingRepository.storageFenceKey("unrelated"),
+                JdbcRuntimeBindingRepository.storageFenceKey("storage"));
        var binding = store.findSession("migrating", original.sessionId()).orElseThrow().workspace();
-        var attemptingLock = new CountDownLatch(1);
-        var waitingJdbc = new ManagedSchemaJdbcTemplate(data) {
-            @Override
-            public <T> T execute(ConnectionCallback<T> action) {
-                return super.execute((ConnectionCallback<T>) connection -> {
-                    attemptingLock.countDown();
-                    return action.doInConnection(connection);
-                });
-            }
-        };
-        var waitingStore = com.alibaba.qwen.code.managedagent.VerificationFixtures.managedAgentStore(waitingJdbc, new ObjectMapper(), Clock.systemUTC(), ignored -> {},
-                new ManagedWorkspaceRegistry(waitingJdbc), properties, new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules());
        try (var migration = data.getConnection(); var pool = Executors.newFixedThreadPool(2)) {
            migration.setAutoCommit(false);
            JdbcRuntimeBindingRepository.lockPlacementDomain(migration, "migrating");
-            if (cwd) {
-                installMigrationFence(migration);
-            }
+            installMigrationFence(migration);
            var creation = pool.submit(() -> cwd
-                    ? transaction.execute(status -> waitingStore.completeCwdChangeOperation("migrating", original.sessionId(),
+                    ? transaction.execute(status -> store.completeCwdChangeOperation("migrating", original.sessionId(),
                            originalClaim.operationId(), "owner", originalClaim.claimGeneration()))
-                    : create(waitingStore, transaction, "migrating", "racing"));
+                    : create(store, transaction, "migrating", "racing"));
            try {
-                if (cwd) {
-                    MySqlLockProbe.awaitLockWaits(jdbc, schema, 1, "CWD settlement behind the storage fence");
-                    assertThat(jdbc.queryForObject(
-                            "SELECT COUNT(*) FROM performance_schema.data_lock_waits waits"
-                                    + " JOIN performance_schema.data_locks requested ON requested.ENGINE = waits.ENGINE"
-                                    + " AND requested.ENGINE_LOCK_ID = waits.REQUESTING_ENGINE_LOCK_ID"
-                                    + " WHERE requested.OBJECT_SCHEMA = ? AND requested.OBJECT_NAME = 'ma_storage'",
-                            Integer.class, schema)).isPositive();
-                } else {
-                    assertThat(attemptingLock.await(5, TimeUnit.SECONDS)).isTrue();
-                }
+                MySqlLockProbe.awaitLockWaits(jdbc, schema, 1, "Workspace admission behind the storage fence");
+                assertThat(jdbc.queryForObject(
+                        "SELECT COUNT(*) FROM performance_schema.data_lock_waits waits"
+                                + " JOIN performance_schema.data_locks requested ON requested.ENGINE = waits.ENGINE"
+                                + " AND requested.ENGINE_LOCK_ID = waits.REQUESTING_ENGINE_LOCK_ID"
+                                + " WHERE requested.OBJECT_SCHEMA = ? AND requested.OBJECT_NAME = 'ma_storage'",
+                        Integer.class, schema)).isPositive();
                assertThatThrownBy(() -> creation.get(100, TimeUnit.MILLISECONDS)).isInstanceOf(TimeoutException.class);
                assertThat(pool.submit(() -> cwd
                        ? transaction.execute(status -> store.completeCwdChangeOperation("unrelated", other.sessionId(),
@@ -194,9 +227,6 @@ class WorkspaceMigrationMySqlIT {
                        : create(store, transaction, "unrelated", "other"))
                        .get(5, TimeUnit.SECONDS)).isNotNull();
                assertThat(creation.isDone()).isFalse();
-                if (!cwd) {
-                    installMigrationFence(migration);
-                }
                migration.commit();
                if (cwd) {
                    var outcome = (com.alibaba.qwen.code.managedagent.contracts.session.AgentStateStore.CwdChangeOutcome)
@@ -271,9 +301,9 @@ class WorkspaceMigrationMySqlIT {
    @Timeout(30)
    void headlessRetirementDoesNotBlockAnotherTenantsWriter() throws Exception {
        ManagedSchema.initialize(data);
-        seed("tenant-a", "headless", "DELETED");
-        seed("tenant-b", "m-successor", "ACTIVE");
-        seed("tenant-b", "a-new", "ACTIVE");
+        seed(jdbc, "tenant-a", "headless", "DELETED");
+        seed(jdbc, "tenant-b", "m-successor", "ACTIVE");
+        seed(jdbc, "tenant-b", "a-new", "ACTIVE");
        jdbc.update(
                "UPDATE ma_session SET retirement_operation_id = ?, retirement_generation = 1,"
                        + " retired_at = 1 WHERE tenant_id = ? AND session_id = ?", "delete", "tenant-a", "headless");
@@ -326,15 +356,15 @@ class WorkspaceMigrationMySqlIT {
                                        + " = ?", Integer.class, schema)).isOne();
    }

-    private void seed(String tenant, String session, String state) {
-        jdbc.update(
+    private void seed(JdbcTemplate writer, String tenant, String session, String state) {
+        writer.update(
                "INSERT INTO ma_session (tenant_key, session_key, session_scope_key, tenant_id, session_id, agent_id, status, gmt_create, gmt_modified, workspace_id, workspace_generation, workspace_storage_id, cwd_relative, context_config_ref, context_revision, workspace_config_ref, workspace_policy_ref, deleted_at) VALUES (?, ?, ?, ?, ?, 'qwen-code', ?, 1, 1, 'workspace', 1, 'storage', '.', ?, 1, ?, ?, ?)",
                com.alibaba.qwen.code.runtimebroker.schema.ManagedSchemaKeys.key(tenant),
                com.alibaba.qwen.code.runtimebroker.schema.ManagedSchemaKeys.key(session),
                com.alibaba.qwen.code.managedagent.repository.ManagedSessionStore.sessionScopeKey(tenant, session), tenant, session, state, WorkspaceExecutionProfile.CONTEXT_CONFIG_REF, WorkspaceExecutionProfile.CONFIG_REF,
                WorkspaceExecutionProfile.POLICY_REF, "DELETED".equals(state) ? 1L : null);
-        ManagedReceipts.insertCreation(jdbc, tenant, new byte[]{1}, session, "digest", session, null, 1);
-        jdbc.update(
+        ManagedReceipts.insertCreation(writer, tenant, new byte[]{1}, session, "digest", session, null, 1);
+        writer.update(
                "INSERT INTO ma_event_stream (tenant_id, session_id, stream_kind, stream_id, gmt_modified) VALUES (?, ?, 'SESSION', ?, 1)",
                tenant, session, new byte[0]);
    }
@@ -18,7 +18,8 @@
 com.alibaba.qwen.code.managedagent.persistence.AgentExecutionOperations=1011
 com.alibaba.qwen.code.managedagent.persistence.AgentLifecycleOperations=1055
 com.alibaba.qwen.code.managedagent.persistence.AgentMessageProjection=959
-com.alibaba.qwen.code.managedagent.persistence.AgentSessionAdmissions=889
+# PR #16327: approved 11-line allowance for Workspace creation lock ordering.
+com.alibaba.qwen.code.managedagent.persistence.AgentSessionAdmissions=900
 com.alibaba.qwen.code.managedagent.persistence.AgentSessionCommands=1078
 com.alibaba.qwen.code.managedagent.persistence.AgentTurnExecution=819
 com.alibaba.qwen.code.managedagent.persistence.AutomationLedgerStore=820