perf(managed-agent): take turn admission off the tenant lock (#16223)

Turn submit no longer takes the tenant placement guard. Its Session row
lock already orders it against lifecycle begin, and requireOpen's
locking reads of the storage's alias (fence) and directory rows order
it against migration fence installation and physical fencing.

Migration prepare locks that fence row in migrationSource but could
check idleness on a REPEATABLE READ snapshot taken before the lock. It
now reads committed work on a separate autocommit connection after the
lock, so a Turn admitted before it is seen and a later one is refused
by the fence.

Co-authored-by: Claude Opus 5.5 <[email protected]>

docs/design/2026-10-11-managed-tenant-guard-wedge.md

@@ -4,6 +4,8 @@
 
 - Status: implemented in this change (witness: `ManagedSessionGuardConvoyIT`);
   the lanes listed under Follow-ups keep unbounded acquisitions
+- Update 2026-11-03: Turn submit admission no longer takes the placement
+  guard; see the [turn admission design](2026-11-03-managed-turn-admission-tenant-guard.md)
 - Issue: [#13800](https://github.com/QwenLM/qwen-code/issues/13800) — a
   `recovery_blocked` Session wedges later turns of other Sessions on the same
   daemon. This change does not implement that issue's W1 or W2; see Scope.

docs/design/2026-10-11-managed-tenant-guard-wedge.zh-CN.md

@@ -4,6 +4,8 @@
 
 - 状态：已在本改动中实现（见证：`ManagedSessionGuardConvoyIT`）；「后续」
   一节列出的车道仍保留无界获取
+- 2026-11-03 更新：Turn 提交准入不再获取 placement 守卫；参见
+  [Turn 准入设计](2026-11-03-managed-turn-admission-tenant-guard.zh-CN.md)
 - Issue：[#13800](https://github.com/QwenLM/qwen-code/issues/13800)——一个处于
   `recovery_blocked` 的 Session 楔死同 daemon 上其他 Session 的后续轮次。
   本改动并不实现该 issue 的 W1 或 W2，见「范围」。

docs/design/2026-11-03-managed-turn-admission-tenant-guard.md

@@ -0,0 +1,99 @@
+# Turn admission without the tenant guard
+
+[English](2026-11-03-managed-turn-admission-tenant-guard.md) | [简体中文](2026-11-03-managed-turn-admission-tenant-guard.zh-CN.md)
+
+## Problem and current state
+
+Every Turn submit runs `AgentSessionCommands.insertTurnCommand`, which first takes
+the tenant placement guard through `WorkspaceMigrationAdmission.lockTenant`,
+bounded at five seconds since [#13800](2026-10-11-managed-tenant-guard-wedge.md).
+Every Turn of every Session in a tenant therefore queues on one row, behind
+Runtime placement, storage registration, fencing, migration, maintenance, CSI
+work and retention, and a long holder turns into `tenant_guard_contended`.
+
+The #13800 design records why the guard is there: a storage migration fence must
+exclude new work, and `WorkspaceMigrationStore.prepare` checks that the storage is
+idle before it writes the fence. Turn admission and fence installation share the
+guard, so a Turn cannot be admitted between the idle check and the fence.
+
+## What already orders Turn admission
+
+Turn admission locks its rows in this order, all in one transaction:
+
+1. The `ma_session` row (`FOR UPDATE`). Lifecycle begin (close, archive, delete)
+   locks the same row before it checks for open Turns and operations, so the two
+   already serialize on it.
+2. For a Workspace Session, `WorkspaceMigrationAdmission.requireOpen` takes
+   locking reads of the storage's `ma_storage` alias row, which is also the
+   migration fence row, and of its `ma_storage_owner` directory row, and requires
+   no fence and a `READY` mount. Physical fencing and restore lock the alias and
+   directory rows, and workspace recovery runs only while the directory is
+   `FENCED`.
+3. It inserts the Turn and its command.
+
+Nothing later in the transaction takes the tenant guard: event append, worktree
+readiness and configuration checks use row locks only. Removing the first
+acquisition therefore removes a wait edge and adds none.
+
+## The gap and the change
+
+On the migration side, `prepare` takes the guard and then calls
+`migrationSource`, which locks every alias row of the physical storage, including
+the fence row, through `LocalStorageOwners.lockCovered`. An admission that holds
+the fence row has committed before that lock is granted; a later one waits and
+then sees the fence. But `migrationSource` reads `ma_storage` without a lock
+before taking that lock, so on MySQL the migration transaction's REPEATABLE READ
+snapshot can predate it. Without the guard, a Turn committed between that
+snapshot and the lock would be invisible to the idle check.
+
+The change has two parts:
+
+- `insertTurnCommand` no longer takes the tenant guard. The Session row lock is
+  its first statement, so its own snapshot follows that lock.
+- `prepare` runs its idle check (`requireCommittedIdle`) on a separate autocommit,
+  read-only connection after `migrationSource` has locked the fence row. This is
+  the pattern `LocalStorageOwners.lockCovered` already uses for its authority
+  inventory. Every unsettled Turn admitted before the lock is committed and
+  visible, and every later admission is refused by the fence.
+
+The other idle checks (`retire` and later phases) run after the fence is
+committed, inside transactions whose first reads are locking reads of the fence
+row, or outside any transaction, so their snapshots already follow the fence.
+
+## Scope
+
+In scope: Turn submit admission and the migration prepare idle check.
+
+Out of scope, keeping the guard: the other Session commands that take
+`WorkspaceMigrationAdmission.lockTenant` (rename, model and approval changes,
+MCP and extension management, cwd change, task cancel, context compression,
+Session and child insertion, actions), the user message queue's `startNext`
+(it still takes the guard before submitting through the same path), lifecycle
+operations, Journal writers and publication. Each needs the same audit before its
+guard can go.
+
+## Risks
+
+- Turn admission now waits on the storage's alias and directory rows instead of
+  the tenant row, so Turns for one storage still serialize briefly there, and a
+  migration or physical fence of that storage briefly blocks them.
+- Lock order is Session row, then alias, then directory, matching lifecycle
+  begin. A guard holder that locked an alias row before a Session row could
+  deadlock with it; InnoDB detects that and rolls back one side, which callers
+  surface as a retryable failure. No such holder was found.
+- The idle check uses one extra pooled connection while `prepare` holds its
+  locks. Migration is an administrator operation.
+
+## Verification and acceptance
+
+1. `WorkspaceMigrationStoreTest`: inside a REPEATABLE READ transaction whose
+   snapshot is already pinned, a Turn committed from another connection still
+   makes `prepare` refuse with `migration_work_unsettled` and install no fence.
+2. `ManagedSessionGuardConvoyIT` (MySQL): while another transaction holds the
+   tenant guards, Turn submit no longer fails with `tenant_guard_contended`.
+3. Existing migration, Turn admission, lifecycle and user message queue suites
+   pass, including MySQL in CI.
+
+## Open questions
+
+None for this increment. The remaining Session commands are follow-up work.

docs/design/2026-11-03-managed-turn-admission-tenant-guard.zh-CN.md

@@ -0,0 +1,83 @@
+# 去掉 Turn 准入的租户 guard
+
+[English](2026-11-03-managed-turn-admission-tenant-guard.md) | [简体中文](2026-11-03-managed-turn-admission-tenant-guard.zh-CN.md)
+
+## 问题与现状
+
+每次 Turn 提交都会执行 `AgentSessionCommands.insertTurnCommand`。它首先通过
+`WorkspaceMigrationAdmission.lockTenant` 获取租户 placement guard；自
+[#13800](2026-10-11-managed-tenant-guard-wedge.zh-CN.md) 起，该获取限时五秒。因此，
+一个租户内所有 Session 的每个 Turn 都在同一行上排队，排在 Runtime placement、存储
+注册、fencing、迁移、维护、CSI 工作和 retention 之后；遇到长时间持有者时就会变成
+`tenant_guard_contended`。
+
+#13800 设计记录了保留这把 guard 的原因：存储迁移 fence 必须排除新工作，而
+`WorkspaceMigrationStore.prepare` 先检查存储是否空闲，再写入 fence。Turn 准入与
+fence 安装共用这把 guard，所以不会有 Turn 在空闲检查和 fence 之间被准入。
+
+## 已经在约束 Turn 准入顺序的锁
+
+Turn 准入在同一事务内按以下顺序加锁：
+
+1. `ma_session` 行（`FOR UPDATE`）。生命周期 begin（close、archive、delete）在检查
+   未结束的 Turn 和操作之前也锁定同一行，所以两者已在该行上串行。
+2. 对 Workspace Session，`WorkspaceMigrationAdmission.requireOpen` 以锁定读读取该
+   存储的 `ma_storage` 别名行（同时也是迁移 fence 行）及其 `ma_storage_owner` 目录行，
+   并要求没有 fence、挂载为 `READY`。物理 fencing 与 restore 会锁定别名行和目录行，
+   workspace recovery 只在目录处于 `FENCED` 时执行。
+3. 插入 Turn 及其命令。
+
+事务后续步骤都不再获取租户 guard：事件追加、worktree 就绪检查和配置检查只使用行锁。
+因此去掉开头这次获取只会删除一条等待边，不会新增等待边。
+
+## 缺口与改动
+
+迁移一侧，`prepare` 先获取 guard，再调用 `migrationSource`；后者通过
+`LocalStorageOwners.lockCovered` 锁定该物理存储的所有别名行，其中包括 fence 行。
+持有 fence 行的准入会在该锁授予之前提交；之后的准入会等待，然后看到 fence。但
+`migrationSource` 在加锁之前先以非锁定读读取了 `ma_storage`，因此在 MySQL 上，迁移
+事务的 REPEATABLE READ 快照可能早于加锁时间。去掉 guard 后，在该快照与加锁之间提交
+的 Turn 将对空闲检查不可见。
+
+改动包括两部分：
+
+- `insertTurnCommand` 不再获取租户 guard。锁定 Session 行是它的第一条语句，因此它
+  自己的快照建立在该锁之后。
+- `prepare` 在 `migrationSource` 锁定 fence 行之后，在一个独立的、自动提交的只读
+  连接上执行空闲检查（`requireCommittedIdle`）。这与
+  `LocalStorageOwners.lockCovered` 检查 authority 清单时已采用的方式相同。加锁之前
+  准入的未结束 Turn 都已提交且可见，之后的准入都会被 fence 拒绝。
+
+其他空闲检查（`retire` 及后续阶段）都在 fence 提交之后执行：要么在首次读取为 fence
+行锁定读的事务内，要么不在事务内，因此它们的快照已经在 fence 之后。
+
+## 范围
+
+范围内：Turn 提交准入，以及迁移 prepare 的空闲检查。
+
+范围外、继续持有 guard：其他使用 `WorkspaceMigrationAdmission.lockTenant` 的 Session
+命令（改名、修改模型与审批、MCP 与扩展管理、cwd 变更、任务取消、上下文压缩、Session
+与子会话插入、actions）、用户消息队列的 `startNext`（它在通过同一路径提交前仍获取
+guard）、生命周期操作、Journal 写者以及发布。它们都需要先完成同样的审计，才能去掉
+guard。
+
+## 风险
+
+- Turn 准入现在等待的是存储的别名行和目录行，而不是租户行。因此同一存储的 Turn 仍会
+  在这些行上短暂串行，该存储的迁移或物理 fence 也会短暂阻塞它们。
+- 锁顺序为 Session 行、别名行、目录行，与生命周期 begin 一致。若某个 guard 持有者先锁
+  别名行再锁 Session 行，可能与之死锁；InnoDB 会检测并回滚其中一方，调用方将其表现为
+  可重试的失败。审计中没有发现这样的持有者。
+- `prepare` 持锁期间，空闲检查会额外使用一个连接池连接。迁移是管理员操作。
+
+## 验证与验收
+
+1. `WorkspaceMigrationStoreTest`：在快照已固定的 REPEATABLE READ 事务内，由另一连接
+   提交的 Turn 仍使 `prepare` 以 `migration_work_unsettled` 拒绝，并且不安装 fence。
+2. `ManagedSessionGuardConvoyIT`（MySQL）：另一事务持有租户 guard 时，Turn 提交不再
+   以 `tenant_guard_contended` 失败。
+3. 现有迁移、Turn 准入、生命周期与用户消息队列测试继续通过，包括 CI 中的 MySQL。
+
+## 未决问题
+
+本增量没有。其余 Session 命令留作后续工作。

packages/managed-agent-server/persistence/src/main/java/com/alibaba/qwen/code/managedagent/persistence/AgentSessionCommands.java

@@ -69,11 +69,12 @@ final class AgentSessionCommands {
             String idempotencyKey, String requestDigest, String sessionId,
             List<Map<String, Object>> input, String payloadDigest,
             String actorId) {
-        // A per-turn admission: a long holder of the tenant's placement row
-        // answers tenant_guard_contended within the hot-path bound instead
-        // of parking the submit past the caller's budget (#13800).
-        ManagedSessionStore.lockBounded(() -> WorkspaceMigrationAdmission.lockTenant(jdbc, tenantId,
-                ManagedSessionStore.GUARD_ACQUIRE_TIMEOUT_SECONDS));
+        // A per-turn admission takes no tenant guard. The Session row orders it
+        // against lifecycle begin, requireOpen's locking reads of this storage's
+        // alias (fence) and directory rows order it against migration fence
+        // installation and physical fencing, and nothing later in this
+        // transaction takes the guard. The Session lock comes first, so this
+        // transaction's snapshot is taken after it.
         SessionRecord session = storeRows.requireLegacySessionForUpdate(tenantId, sessionId);
         sessionPolicies.requirePublicChildImmutable(session);
         // A bound Session's later Turn needs the same deployment opt-in as

packages/managed-agent-server/persistence/src/main/java/com/alibaba/qwen/code/managedagent/persistence/WorkspaceMigrationStore.java

@@ -112,7 +112,7 @@ public final class WorkspaceMigrationStore {
         transaction.executeWithoutResult(status -> {
             lockAuthority();
             guard.migrationSource(tenant, storage, revision, fence);
-            requireIdle();
+            requireCommittedIdle();
             requireFilesProfiles();
             var raced = row(true);
             if (raced != null) {
@@ -345,46 +345,66 @@ public final class WorkspaceMigrationStore {
     }
 
     private void requireIdle() {
-        check(count("SELECT COUNT(*) FROM ma_session s JOIN ma_turn t"
+        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() {
+        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)));
+        } catch (java.sql.SQLException error) {
+            throw new IllegalStateException("Migration idle check is unavailable", error);
+        }
+    }
+
+    private void requireIdle(JdbcTemplate reader) {
+        check(countIn(reader, "SELECT COUNT(*) FROM ma_session s JOIN ma_turn t"
                 + " ON t.tenant_id = s.tenant_id AND t.session_id = s.session_id WHERE s.tenant_id = ?"
                 + " AND s.workspace_storage_id = ? AND t.status IN ('ACCEPTED', 'RUNNING', 'CANCELLING')") == 0,
                 "migration_work_unsettled");
         // A parked task cancel counts here although it holds no lifecycle
         // hostage: its acceptance is unknown and its task may still use this
         // storage, so the move waits for the reconciler (D7-2).
-        check(count("SELECT COUNT(*) FROM ma_session s JOIN ma_operation o"
+        check(countIn(reader, "SELECT COUNT(*) FROM ma_session s JOIN ma_operation o"
                 + " ON o.session_scope_key = s.session_scope_key AND o.tenant_id = s.tenant_id AND o.session_id = s.session_id WHERE s.tenant_id = ?"
                 + " AND s.workspace_storage_id = ? AND o.operation_kind IN ('CLOSE', 'ARCHIVE', 'DELETE', 'ACTION_RESPONSE', 'CWD_CHANGE', 'TASK_CANCEL') AND o.state NOT IN ('COMPLETED', 'FAILED')") == 0,
                 "migration_work_unsettled");
-        check(count("SELECT COUNT(*) FROM ma_session s JOIN ma_request_receipt c"
+        check(countIn(reader, "SELECT COUNT(*) FROM ma_session s JOIN ma_request_receipt c"
                 + " ON c.tenant_id = s.tenant_id AND c.session_id = s.session_id WHERE s.tenant_id = ?"
                 + " AND s.workspace_storage_id = ? AND c.namespace = 'SESSION_COMMAND' AND c.state = 'PENDING'") == 0,
                 "migration_work_unsettled");
         // An open or parked title waits; one a later title superseded, or
         // one proven never delivered, has settled.
-        check(count("SELECT COUNT(*) FROM ma_session s JOIN ma_operation r"
+        check(countIn(reader, "SELECT COUNT(*) FROM ma_session s JOIN ma_operation r"
                 + " ON r.session_scope_key = s.session_scope_key AND r.tenant_id = s.tenant_id AND r.session_id = s.session_id WHERE s.tenant_id = ?"
                 + " AND s.workspace_storage_id = ? AND r.operation_kind = 'RENAME' AND r.state IN ('PENDING', 'RUNNING', 'RECOVERY_BLOCKED')") == 0,
                 "migration_work_unsettled");
         // A child Workspace's worktree records absolute paths of this root
         // (#13753 I1): it must be merged or discarded before the storage
         // moves. A conflicted one keeps only refs, which move with it.
-        check(count("SELECT COUNT(*) FROM ma_workspace WHERE workspace_kind = 'CHILD_GIT' AND tenant_id = ?"
+        check(countIn(reader, "SELECT COUNT(*) FROM ma_workspace WHERE workspace_kind = 'CHILD_GIT' AND tenant_id = ?"
                 + " AND storage_id = ? AND child_state NOT IN ('merged', 'discarded', 'failed', 'conflicted')") == 0,
                 "migration_work_unsettled");
-        check(jdbc.queryForObject("SELECT COUNT(*) FROM ma_session_worktree WHERE physical_key IN"
+        check(reader.queryForObject("SELECT COUNT(*) FROM ma_session_worktree WHERE physical_key IN"
                         + " (SELECT physical_key FROM ma_storage WHERE alias_key = ?) AND state <> 'REMOVED'", Integer.class,
                 com.alibaba.qwen.code.runtimebroker.persistence.JdbcRuntimeBindingRepository.storageAliasKey(tenant, storage)) == 0,
                 "migration_work_unsettled");
-        long now = jdbc.queryForObject("SELECT UNIX_TIMESTAMP(), EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(6))",
+        long now = reader.queryForObject("SELECT UNIX_TIMESTAMP(), EXTRACT(MICROSECOND FROM CURRENT_TIMESTAMP(6))",
                 (row, index) -> Math.addExact(Math.multiplyExact(row.getLong(1), 1000), row.getLong(2) / 1000));
         String after = "";
         while (true) {
-            var sessions = jdbc.queryForList("SELECT session_id FROM ma_session WHERE tenant_id = ?"
+            var sessions = reader.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 : sessions) {
-                for (String options : jdbc.queryForList("SELECT options_json FROM ma_session_record"
+                for (String options : reader.queryForList("SELECT options_json FROM ma_session_record"
                         + " WHERE tenant_id = ? AND session_id = ? AND record_kind = 'ACTION' AND state = 'requested'", String.class, tenant, session)) {
                     check(parse(options).path("expiresAt").asLong(Long.MAX_VALUE) <= now, "migration_work_unsettled");
                 }
@@ -394,12 +414,12 @@ public final class WorkspaceMigrationStore {
                 break;
             }
         }
-        check(count("SELECT COUNT(*) FROM ma_session s JOIN ma_session_head h ON h.tenant_key = s.tenant_key AND h.harness_key = s.session_key AND h.tenant_id = s.tenant_id AND h.session_id = s.session_id WHERE s.tenant_id = ? AND s.workspace_storage_id = ? AND h.state = 'ACTIVE' AND (h.writer_lease_until IS NULL OR h.writer_lease_until > " + ManagedSchemaTime.SQL_NOW + ")") == 0,
+        check(countIn(reader, "SELECT COUNT(*) FROM ma_session s JOIN ma_session_head h ON h.tenant_key = s.tenant_key AND h.harness_key = s.session_key AND h.tenant_id = s.tenant_id AND h.session_id = s.session_id WHERE s.tenant_id = ? AND s.workspace_storage_id = ? AND h.state = 'ACTIVE' AND (h.writer_lease_until IS NULL OR h.writer_lease_until > " + ManagedSchemaTime.SQL_NOW + ")") == 0,
                 "migration_writer_active");
     }
 
-    private long count(String sql) {
-        return jdbc.queryForObject(sql, Long.class, tenant, storage);
+    private long countIn(JdbcTemplate reader, String sql) {
+        return reader.queryForObject(sql, Long.class, tenant, storage);
     }
 
     private void assertRetired() {

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/persistence/ManagedSessionGuardConvoyIT.java

@@ -170,6 +170,11 @@ class ManagedSessionGuardConvoyIT {
                 + " recovery_detail_code = 'a2-witness' WHERE tenant_id = ? AND session_id = ?"
                 + " AND head_kind = 'MANAGED_SESSION'",
                 tenant, "blocked");
+        var agents = com.alibaba.qwen.code.managedagent.VerificationFixtures.managedAgentStore(jdbc, JSON, Clock.systemUTC(), ignored -> {},
+                new ManagedWorkspaceRegistry(jdbc), new ManagedAgentProperties(), new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules());
+        String turnSession = transactions.execute(ignored -> agents.insertSessionCommand(tenant,
+                "CREATE_SESSION", "turn-session", "digest-create", "qwen-code", null, null,
+                List.of(), null)).sessionId();
 
         var holdTx = manager.getTransaction(new DefaultTransactionDefinition());
         try {
@@ -261,12 +266,18 @@ class ManagedSessionGuardConvoyIT {
             long lifecycleElapsedMs = (System.nanoTime() - lifecycleBefore) / 1_000_000;
             assertThat(lifecycleElapsedMs).isLessThan(BOUNDED_FAILURE_CEILING_MS);
 
-            // A Turn submit admits through the same placement row on its
-            // per-turn path, so it is bounded and named the same way.
-            var agents = com.alibaba.qwen.code.managedagent.VerificationFixtures.managedAgentStore(jdbc, JSON, Clock.systemUTC(), ignored -> {},
-                    new ManagedWorkspaceRegistry(jdbc), new ManagedAgentProperties(), new com.alibaba.qwen.code.managedagent.harness.SdkHarnessRules());
-            assertFastGuardFailure(() -> transactions.execute(ignored -> agents.insertTurnCommand(tenant,
-                    "turn.submit", "submit-witness", "a".repeat(64), "victim", List.of(), "b".repeat(64))));
+            // A Turn submit no longer admits through the placement row: it
+            // orders on its Session row and the storage's fence row instead,
+            // so a held tenant guard neither parks nor names it.
+            long submitBefore = System.nanoTime();
+            try (var submitPool = Executors.newSingleThreadExecutor()) {
+                var admitted = submitPool.submit(() -> transactions.execute(ignored -> agents.insertTurnCommand(tenant,
+                        "turn.submit", "submit-witness", "a".repeat(64), turnSession, List.of(), "b".repeat(64))))
+                        .get(30, TimeUnit.SECONDS);
+                assertThat(admitted.turnId()).startsWith("turn_");
+                assertThat(admitted.commandEffect()).isTrue();
+            }
+            assertThat((System.nanoTime() - submitBefore) / 1_000_000).isLessThan(BOUNDED_FAILURE_CEILING_MS);
         } finally {
             manager.rollback(holdTx);
         }

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/persistence/WorkspaceMigrationStoreTest.java

@@ -370,6 +370,36 @@ class WorkspaceMigrationStoreTest {
         assertThat(jdbc.queryForObject("SELECT delivery_state FROM ma_operation WHERE operation_kind = 'RENAME'", String.class)).isEqualTo(state);
     }
 
+    // Turn admission orders on the storage's fence row, not the tenant guard, so
+    // the idle check must see a Turn committed after the caller's snapshot was
+    // taken but before the migration locked that row.
+    @Test
+    void theIdleCheckSeesATurnCommittedAfterTheTransactionSnapshot() {
+        jdbc.update("INSERT INTO ma_session (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)"
+                + " VALUES ('tenant', 'session', 'qwen-code', 'ACTIVE', 1, 1, 'workspace', 1, 'storage', '.', ?, 1, ?, ?)",
+                WorkspaceExecutionProfile.CONTEXT_CONFIG_REF, WorkspaceExecutionProfile.CONFIG_REF,
+                WorkspaceExecutionProfile.POLICY_REF);
+        var outer = new org.springframework.transaction.support.TransactionTemplate(manager);
+        outer.setIsolationLevel(org.springframework.transaction.TransactionDefinition.ISOLATION_REPEATABLE_READ);
+        assertThatThrownBy(() -> outer.executeWithoutResult(status -> {
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_turn", Long.class)).isZero();
+            try (var connection = java.util.Objects.requireNonNull(jdbc.getDataSource()).getConnection()) {
+                connection.setAutoCommit(true);
+                new JdbcTemplate(new org.springframework.jdbc.datasource.SingleConnectionDataSource(connection, true))
+                        .update("INSERT INTO ma_turn (tenant_id, session_id, turn_id, prompt_id, input_json, payload_digest,"
+                                + " status, is_submission_attempted, gmt_create, gmt_modified, version)"
+                                + " VALUES ('tenant', 'session', 'turn-1', 'prompt-1', '[]', 'd', 'ACCEPTED', FALSE, 1, 1, 1)");
+            } catch (java.sql.SQLException error) {
+                throw new IllegalStateException(error);
+            }
+            assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_turn", Long.class)).isZero();
+            store(true);
+        })).hasMessageContaining("migration_work_unsettled");
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE fence_operation_id IS NOT NULL", Long.class)).isZero();
+    }
+
     // A title a later title superseded, or one proven never delivered, has
     // settled and holds nothing: superseded rows used to hold the storage for
     // good, since nothing ever completes them.