feat(managed-agent): allow concurrent workspace rewinds (#16369)

Allow neighboring Sessions, uploads and rewinds to coexist while retaining pending rewind recovery on the originating Runtime Session.

Co-authored-by: Qwen-Coder <[email protected]>

docs/design/2026-10-21-managed-workspace-concurrency.md

@@ -2,6 +2,8 @@
 
 [English](2026-10-21-managed-workspace-concurrency.md) | [简体中文](2026-10-21-managed-workspace-concurrency.zh-CN.md)
 
+The exclusive rewind behavior below is superseded by [rewind without directory exclusion](2026-11-05-managed-rewind-concurrency.md). Shared execution membership and physical-maintenance protection remain in place.
+
 ## Status and scope
 
 This change permits ordinary Managed Sessions on LOCAL and SSH_LOCAL storage to execute concurrently in the same directory. It covers both Session-isolated and Workspace-isolated workers. A Session still runs one Turn at a time. Resident and private CSI profiles retain their current admission, reservation, and retirement rules. Existing Agent/Workflow child-admission restrictions are outside this slice.

docs/design/2026-10-21-managed-workspace-concurrency.zh-CN.md

@@ -2,6 +2,8 @@
 
 [English](2026-10-21-managed-workspace-concurrency.md) | [简体中文](2026-10-21-managed-workspace-concurrency.zh-CN.md)
 
+下述 rewind 排他行为已由[取消 rewind 的目录排他](2026-11-05-managed-rewind-concurrency.zh-CN.md)替代。共享执行成员资格和物理维护保护仍保留。
+
 ## 状态与范围
 
 本次允许普通 Managed Session 在 LOCAL、SSH_LOCAL 的同一目录并发执行，覆盖 Session 隔离和 Workspace 隔离的 worker。同一 Session 仍然一次执行一个 Turn。Resident 和私有 CSI profile 保留当前准入、预留和退役规则。现有 Agent/Workflow 子任务准入限制不在本次范围。

docs/design/2026-11-03-managed-shared-workspace-uploads.md

@@ -2,6 +2,8 @@
 
 [English](2026-11-03-managed-shared-workspace-uploads.md) | [简体中文](2026-11-03-managed-shared-workspace-uploads.zh-CN.md)
 
+The rewind exclusion described below is superseded by [rewind without directory exclusion](2026-11-05-managed-rewind-concurrency.md). Shared uploads and physical maintenance protections remain.
+
 ## Problem and scope
 
 Ordinary Sessions already share a Workspace directory. Upload admission still

docs/design/2026-11-03-managed-shared-workspace-uploads.zh-CN.md

@@ -2,6 +2,8 @@
 
 [English](2026-11-03-managed-shared-workspace-uploads.md) | [简体中文](2026-11-03-managed-shared-workspace-uploads.zh-CN.md)
 
+下述 rewind 排他决定已由[取消 rewind 的目录排他](2026-11-05-managed-rewind-concurrency.zh-CN.md)替代。共享上传和物理维护保护仍保留。
+
 ## 问题与范围
 
 普通 Session 已经可以共享 Workspace 目录。上传准入仍要求没有执行成员，并设置

docs/design/2026-11-03-managed-workspace-membership-cas.md

@@ -2,6 +2,8 @@
 
 [English](2026-11-03-managed-workspace-membership-cas.md) | [简体中文](2026-11-03-managed-workspace-membership-cas.zh-CN.md)
 
+The rewind exclusion described below is superseded by [rewind without directory exclusion](2026-11-05-managed-rewind-concurrency.md). Shared membership and physical maintenance protections remain.
+
 ## Status and scope
 
 Status: Implemented. Focused H2 and existing MySQL recovery and inventory cases

docs/design/2026-11-03-managed-workspace-membership-cas.zh-CN.md

@@ -2,6 +2,8 @@
 
 [English](2026-11-03-managed-workspace-membership-cas.md) | [简体中文](2026-11-03-managed-workspace-membership-cas.zh-CN.md)
 
+下述 rewind 排他决定已由[取消 rewind 的目录排他](2026-11-05-managed-rewind-concurrency.zh-CN.md)替代。共享成员资格和物理维护保护仍保留。
+
 ## 状态与范围
 
 状态：Implemented。针对性的 H2 用例和现有 MySQL 恢复、盘点用例已通过。

docs/design/2026-11-05-managed-rewind-concurrency.md

@@ -0,0 +1,41 @@
+# Rewind without directory exclusion
+
+[English](2026-11-05-managed-rewind-concurrency.md) | [简体中文](2026-11-05-managed-rewind-concurrency.zh-CN.md)
+
+## Problem and decision
+
+Ordinary tools and uploads already share a Workspace. Managed rewind still requires its Runtime Session to be the only directory member and records an exclusive owner until the file operation returns. Another Session or upload prevents rewind, and a lost rewind response blocks the whole directory.
+
+The user explicitly chose CLI-style cross-Session file concurrency for rewind too. This supersedes the rewind exclusion in [shared Workspace uploads](2026-11-03-managed-shared-workspace-uploads.md). Native file history restores files without a cross-Session directory lock; the daemon queues its own Session. Managed file-history digest/mode checks remain optimistic, not a filesystem transaction or a guarantee against external writers.
+
+## Implementation
+
+Keep durable pending-rewind state on the exact Runtime Session. `beginRewind` validates the existing owned-root and membership requirements and atomically changes that READY Session's pending flag from false to true. It no longer requires `member_count = 1` or installs a physical-directory exclusive owner. A repeated or unresolved rewind on that Runtime Session is refused.
+
+The transport forwards an admitted rewind to the original worker instead of synthesizing a conflict solely because another directory member exists. A successful worker response clears pending state; an exception or unknown outcome retains it. Worker file-conflict results are returned unchanged. Another Session can execute, upload or rewind while the original remains pending.
+
+Retain Runtime Session membership and its count throughout rewind. Pending rewind still prevents that Runtime Session's ordinary execution and mount handback. Move the handback pending check into the membership-release transaction so it cannot race rewind admission. Confirmed worker release or proven stopped/LOST cleanup can clear pending state and release only their original memberships.
+
+Remove `ma_storage_owner.exclusive_member_key`, its hash helper, admission predicates and single-exclusive-member cleanup. Rename the Runtime Session flag to `is_rewind_pending` and its query to `hasPendingRewind`. `beginRewind` returns void: admission succeeds or throws. Update schema projections, boolean decoding, CSI inventory and affected tests together.
+
+Short transactions retain the existing alias/physical-owner/Runtime Session lock order for membership bookkeeping. Mount state remains a current read, and pending admission uses a conditional update. SQL locks end before file I/O. No tenant guard, new table, lock service or configuration is introduced.
+
+## Boundaries and compatibility
+
+- Same-Session idle/history checks, exact binding/generation ownership and unknown-result recovery remain required. No automatic replay of uncertain rewind is added.
+- Other writers may change the same file during rewind, as with CLI. No merge, atomic multi-file restore or conflict-free result is promised.
+- Worktree maintenance, physical fencing, migration and collection still require an unused directory. Pending Session membership continues to prevent them even when it is the only member.
+- Owned-root authorization and existing unsupported SSH/CSI/Resident control behavior remain unchanged. Only cross-Session directory exclusion is relaxed.
+- Follow the pre-release schema policy: update V1 in place without a version bump or migration. Old development catalogs require recreation; this task does not modify or recreate user catalogs.
+
+## Validation and acceptance
+
+1. Reproduce old refusal before implementation with tests expecting rewind alongside an existing Session/upload and admission alongside pending rewind.
+2. Existing and newly admitted neighboring Sessions/uploads stay usable during successful and uncertain rewind. Two neighboring Sessions can retain independent pending markers.
+3. The originating Runtime Session cannot start another rewind, execute ordinary work or hand back its mount while pending. Successful settlement or proven physical release clears only its own state.
+4. Membership counts stay exact across upload settlement, normal release, stopped cleanup and LOST recovery. Maintenance and fencing still refuse while any member remains.
+5. Run focused H2 and real MySQL membership/upload/publication tests, schema/inventory regressions and Checkstyle within the shared ten-minute Java budget. Full Hosted/fault, MySQL and SpotBugs acceptance remains CI work.
+
+## Open questions
+
+None. The user explicitly expanded this increment to include rewind concurrency.

docs/design/2026-11-05-managed-rewind-concurrency.zh-CN.md

@@ -0,0 +1,41 @@
+# 取消 rewind 的目录排他
+
+[English](2026-11-05-managed-rewind-concurrency.md) | [简体中文](2026-11-05-managed-rewind-concurrency.zh-CN.md)
+
+## 问题与决定
+
+普通工具与上传已经共享 Workspace。Managed rewind 仍要求其 Runtime Session 是目录的唯一成员，并在文件操作返回前记录排他所有者。其他 Session 或上传会阻止 rewind，而丢失 rewind 响应会阻塞整个目录。
+
+用户明确选择让 rewind 也沿用 CLI 的跨 Session 文件并发语义。这替代了[共享 Workspace 上传](2026-11-03-managed-shared-workspace-uploads.zh-CN.md)中的 rewind 排他决定。原生文件历史恢复不获取跨 Session 目录锁，daemon 只对自身 Session 排队。Managed 文件历史的 digest/mode 检查仍是乐观检查，不是文件系统事务，也不能保证隔离外部写入。
+
+## 实现
+
+在精确的 Runtime Session 上保留持久化待确认 rewind 状态。`beginRewind` 校验现有 owned-root 和成员资格要求，并将该 READY Session 的 pending 标志原子地从 false 改为 true。不再要求 `member_count = 1`，也不再安装物理目录的排他所有者。同一 Runtime Session 的重复或结果未明 rewind 仍被拒绝。
+
+传输层将已准入的 rewind 发往原始 worker，不再仅因目录存在其他成员而人工返回 conflict。worker 成功返回后清除 pending；异常或未知结果保留 pending。worker 原有的文件冲突结果保持原样返回。原 Session 仍 pending 时，其他 Session 可以执行工具、上传或 rewind。
+
+rewind 全程保留 Runtime Session 成员资格和计数。pending rewind 仍阻止该 Runtime Session 的普通执行和 mount 归还。将归还时的 pending 检查移入释放成员资格的事务，避免与 rewind 准入竞态。确认的 worker release 或具有停止/LOST 证据的清理可以清除 pending，并仅释放原始成员资格。
+
+删除 `ma_storage_owner.exclusive_member_key`、配套 hash helper、准入谓词和单一排他成员清理。将 Runtime Session 标志重命名为 `is_rewind_pending`，查询重命名为 `hasPendingRewind`。`beginRewind` 返回 void：准入成功或抛错。同步更新 schema 投影、布尔解码、CSI inventory 和受影响测试。
+
+成员记账的短事务保留现有 alias/物理所有者/Runtime Session 锁顺序。mount 状态仍为当前读，pending 准入使用条件更新。SQL 锁在文件 I/O 前结束。不引入租户 guard、新表、锁服务或配置。
+
+## 边界与兼容性
+
+- 仍要求同 Session 的 idle/history 校验、精确 Binding/代次所有权和未知结果恢复。不增加对不确定 rewind 的自动重放。
+- 与 CLI 一样，其他写入者可以在 rewind 期间修改同一文件。不保证自动合并、多文件原子恢复或无冲突结果。
+- worktree 维护、物理 fencing、迁移和回收仍要求目录无人使用。pending Session 的成员资格继续阻止这些操作，即使它是唯一成员。
+- owned-root 授权和现有 SSH/CSI/Resident 不支持的控制行为保持不变。仅放宽跨 Session 目录排他。
+- 按发布前 schema 策略原地更新 V1，不提升版本、不添加迁移。旧开发 catalog 需要重建；本任务不修改或重建用户 catalog。
+
+## 验证与验收
+
+1. 实现前，期望 rewind 与已有 Session/上传并发、以及在 pending rewind 期间准入的测试复现旧拒绝行为。
+2. 成功及结果未明的 rewind 期间，已有和新加入的相邻 Session/上传仍可使用目录。两个相邻 Session 可以保留独立 pending 标记。
+3. 发起方 Runtime Session 在 pending 时不能再次 rewind、执行普通工作或归还 mount。成功结束或已证明的物理释放仅清理自身状态。
+4. 上传结束、正常释放、停止清理和 LOST 恢复之后，成员计数仍精确。仍有成员时，维护和 fencing 继续拒绝。
+5. 在共享的十分钟 Java 预算内，执行聚焦 H2、真实 MySQL 成员/上传/发布测试、schema/inventory 回归和 Checkstyle。完整 Hosted/fault、MySQL 和 SpotBugs 验收仍交给 CI。
+
+## 未决问题
+
+无。用户已明确将本次范围扩大到 rewind 并发。

packages/managed-agent-server/README.md

@@ -1946,7 +1946,7 @@ each arm runs even when an earlier one failed.
 
 ### Hosted recovery scope
 
-Ordinary LOCAL and SSH_LOCAL Managed Workspace Sessions may execute concurrently in the same directory, with one active Turn per Session. An UNKNOWN tool result on a READY binding retains its original Runtime Session and cannot be replayed, but does not exclude another ordinary Session. Existing LOST, DRAINING, OPERATOR_RECOVERY and attested RECOVERY_BLOCKED placement barriers remain in force. Physical registration, fencing, migration and maintenance still require every execution membership and unproven writer to drain. Shared files have no transactional isolation. Past execution overlap does not disable file undo. Undo requires other current memberships to release, then briefly excludes new acquisitions; existing file-level history checks still apply. Resident/private CSI and existing Agent/Workflow child-admission restrictions are unchanged. See the [concurrency design](../../docs/design/2026-10-21-managed-workspace-concurrency.md) ([简体中文](../../docs/design/2026-10-21-managed-workspace-concurrency.zh-CN.md)).
+Ordinary LOCAL and SSH_LOCAL Managed Workspace Sessions may execute concurrently in the same directory, with one active Turn per Session. An UNKNOWN tool result on a READY binding retains its original Runtime Session and cannot be replayed, but does not exclude another ordinary Session. Existing LOST, DRAINING, OPERATOR_RECOVERY and attested RECOVERY_BLOCKED placement barriers remain in force. Physical registration, fencing, migration and maintenance still require every execution membership and unproven writer to drain. Shared files have no transactional isolation. Undo can run alongside other Sessions and uploads, subject to the existing file-level history checks. An active or unconfirmed undo blocks only its originating Runtime Session's execution, repeated undo and mount handback; neighboring Sessions can continue. Resident/private CSI and existing Agent/Workflow child-admission restrictions are unchanged. See the [concurrency design](../../docs/design/2026-10-21-managed-workspace-concurrency.md) ([简体中文](../../docs/design/2026-10-21-managed-workspace-concurrency.zh-CN.md)) and the updated [rewind semantics](../../docs/design/2026-11-05-managed-rewind-concurrency.md) ([简体中文](../../docs/design/2026-11-05-managed-rewind-concurrency.zh-CN.md)).
 
 The baseline stores ordinary execution memberships per Runtime Session. Use matching Server/Broker builds with a fresh catalog. CSI reservation and physical-maintenance ownership retain their existing roles. Development catalogs using singleton holders require their original deployment to finish or recover and release those holders; they cannot be reused by this build.
 

packages/managed-agent-server/contracts/src/main/java/com/alibaba/qwen/code/managedagent/repository/WorkspaceExecutionStore.java

@@ -32,11 +32,11 @@ public interface WorkspaceExecutionStore {
 
     public void claim(ContextBinding binding, RuntimeSessionRecord session);
 
-    public boolean beginRewind(ContextBinding binding, RuntimeSessionRecord session);
+    public void beginRewind(ContextBinding binding, RuntimeSessionRecord session);
 
     public void endRewind(ContextBinding binding, RuntimeSessionRecord session);
 
-    public boolean hasExclusiveClaim(RuntimeSessionRecord session);
+    public boolean hasPendingRewind(RuntimeSessionRecord session);
 
     public void assertHeld(ContextBinding binding, RuntimeSessionRecord session);
 
@@ -44,6 +44,8 @@ public interface WorkspaceExecutionStore {
 
     public void release(ContextBinding binding, RuntimeSessionRecord session);
 
+    public void releaseMount(ContextBinding binding, RuntimeSessionRecord session);
+
     public String holdForMaintenance(ContextBinding binding, String maintenanceId,
             long claimGeneration);
 

packages/managed-agent-server/control-plane/src/main/java/com/alibaba/qwen/code/managedagent/workspace/runtime/WorkspaceRuntimeTransport.java

@@ -447,7 +447,7 @@ public final class WorkspaceRuntimeTransport implements RuntimeTransport {
             if (recovery) {
                 if (!ownership.isHeld(context.binding(), context.session())) {
                     throw new RuntimeBrokerException(409, "workspace_busy",
-                            "Workspace storage is held by another tool turn.", true);
+                            "Workspace is not available for this operation.", true);
                 }
             } else {
                 ownership.assertHeld(context.binding(), context.session());
@@ -457,11 +457,7 @@ public final class WorkspaceRuntimeTransport implements RuntimeTransport {
             ownership.assertHeld(context.binding(), context.session());
             if (!ResidentWorkspaceProfile.selects(session.getScope())
                     && "raw-file-history".equals(operation.get("kind")) && "rewind".equals(operation.get("action"))) {
-                if (!ownership.beginRewind(context.binding(), context.session())) {
-                    return delegate.control(lease, session, Map.of("kind", "raw-file-history", "action", "snapshot"))
-                            .thenApply(state -> Map.of("state", state, "filesChanged", List.of(),
-                                    "filesFailed", List.of(), "conflict", true));
-                }
+                ownership.beginRewind(context.binding(), context.session());
                 return delegate.control(lease, session, operation).thenApply(result -> {
                     ownership.endRewind(context.binding(), context.session());
                     return result;
@@ -540,15 +536,11 @@ public final class WorkspaceRuntimeTransport implements RuntimeTransport {
         // worker), so it owes the same detach; one with no holder never
         // installed anything and keeps the shortcut.
         Context context = context(lease, session, false);
-        if (ownership.hasExclusiveClaim(context.session())) {
-            throw new RuntimeBrokerException(409, "workspace_writer_unproven",
-                    "A Workspace rewind must settle before its mount can be handed back.", false);
-        }
         if (context.session().getState() == RuntimeSessionRecord.State.READY
                 || ownership.isHeld(context.binding(), context.session())) {
             detachOwed.add(detachKey(context));
         }
-        ownership.release(context.binding(), context.session());
+        ownership.releaseMount(context.binding(), context.session());
         return CompletableFuture.completedFuture(true);
     }
 

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

@@ -198,7 +198,7 @@ final class WorkspaceCsiApplicationInventory {
                     + "workspace_generation:VARBINARY,canonical_cwd:VARCHAR,capability_digest:VARBINARY,"
                     + "isolation_class:VARCHAR,harness_session_id:VARBINARY,turn_kind:VARCHAR,binding_id:VARBINARY,"
                     + "runtime_generation:BIGINT,session_state:VARCHAR,record_version:BIGINT,last_active_at:BIGINT,"
-                    + "execution_storage_key:CHAR,is_execution_exclusive:BOOLEAN"),
+                    + "execution_storage_key:CHAR,is_rewind_pending:BOOLEAN"),
             declaration("ma_runtime_slot", "request_key",
                     "request_key:CHAR,tenant_id:VARBINARY,workspace_id:VARBINARY,workspace_generation:VARBINARY,"
                     + "canonical_cwd:VARCHAR,capability_digest:VARBINARY,isolation_class:VARCHAR,isolation_key:VARBINARY,"
@@ -281,7 +281,7 @@ final class WorkspaceCsiApplicationInventory {
                     + "mount_registration_id:CHAR,storage_kind:VARCHAR,csi_phase:VARCHAR,csi_revision:BIGINT,"
                     + "csi_reservation_id:CHAR,csi_registration_key:CHAR,csi_registration_revision:BIGINT,"
                     + "csi_provision_request_id:VARBINARY,maintenance_id:CHAR,maintenance_generation:BIGINT,"
-                    + "physical_key:CHAR,member_count:BIGINT,exclusive_member_key:CHAR"),
+                    + "physical_key:CHAR,member_count:BIGINT"),
             declaration("ma_tenant_guard", "tenant_key,guard_domain",
                     "tenant_key:CHAR,tenant_id:VARBINARY,guard_domain:VARCHAR"),
             declaration("ma_tool_execution", "execution_call_id_hash",
@@ -2950,8 +2950,7 @@ final class WorkspaceCsiApplicationInventory {
                         && Objects.equals(row.number("csi_revision"),
                                 CsiOwnershipRevision.next(intent.reservation().revision()))
                         && row.cell("maintenance_id").isNull()
-                        && Objects.equals(0L, row.number("member_count"))
-                        && row.cell("exclusive_member_key").isNull());
+                        && Objects.equals(0L, row.number("member_count")));
                 case "ma_runtime_slot" -> expect(table, row, request(row)
                         && original.bindingId().equals(row.text("active_binding_id"))
                         && Objects.equals(original.generation(), row.number("last_generation")));
@@ -2965,7 +2964,7 @@ final class WorkspaceCsiApplicationInventory {
                         && Objects.equals(intent.resourceHandleVersion().longValue(), row.number("resource_handle_version"))
                         && Objects.equals(intent.attestationGeneration(), row.number("attestation_generation")));
                 case "ma_runtime_session" -> expect(table, row, scope(row)
-                        && row.cell("execution_storage_key").isNull() && Boolean.FALSE.equals(row.bool("is_execution_exclusive"))
+                        && row.cell("execution_storage_key").isNull() && Boolean.FALSE.equals(row.bool("is_rewind_pending"))
                         && original.bindingId().equals(row.text("binding_id")) && session.equals(row.text("runtime_session_id"))
                         && session.equals(row.text("harness_session_id")) && "bootstrap".equals(row.text("turn_kind"))
                         && Objects.equals(original.generation(), row.number("runtime_generation"))

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

@@ -339,7 +339,7 @@ public final class WorkspaceCsiCheckpointSnapshotStore {
                 "SELECT " + ManagedSchemaProjection.RUNTIME_SESSION.columns() + " FROM ma_runtime_session WHERE scope_key = ? AND runtime_session_id = ?",
                 scopeHash(runtime.getRequest().getScope()), ref.path("sessionId").asText());
         require(runtime.getRequest().getScope().equals(scope(session))
-                && session.get("execution_storage_key") == null && !flag(session, "is_execution_exclusive")
+                && session.get("execution_storage_key") == null && !flag(session, "is_rewind_pending")
                 && runtime.getBindingId().equals(session.get("binding_id"))
                 && runtime.getGeneration() == rawNumber(session, "runtime_generation")
                 && binding.path("sessionKey").path("sessionId").asText().equals(session.get("harness_session_id")));

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

@@ -85,7 +85,7 @@ public class WorkspaceExecutionStore
                 + " JOIN ma_runtime_slot slot ON slot.active_binding_id = b.binding_id AND slot.request_key = b.request_key"
                 + " JOIN ma_storage_owner o ON o.binding_id = b.binding_id AND o.runtime_generation = b.runtime_generation"
                 + " WHERE rs.binding_id = ? AND rs.runtime_generation = ? AND rs.runtime_session_id = ?"
-                + " AND rs.execution_storage_key IS NULL AND rs.is_execution_exclusive = FALSE"
+                + " AND rs.execution_storage_key IS NULL AND rs.is_rewind_pending = FALSE"
                 + " AND rs.harness_session_id = ? AND rs.session_state IN (" + (releasing
                         ? "'ACQUIRING', 'READY', 'RELEASING', 'RELEASED'" : "'ACQUIRING', 'READY'") + ")"
                 + " AND b.request_key = ? AND b.binding_state = 'READY' AND b.is_drain_requested = FALSE"
@@ -437,7 +437,7 @@ public class WorkspaceExecutionStore
             throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
         }
         if (key.equals(held.getFirst())) {
-            if (!directoryAdmits(key, exclusiveMemberKey(session))) {
+            if (!directoryAdmits(key)) {
                 throw busy();
             }
             return;
@@ -447,7 +447,7 @@ public class WorkspaceExecutionStore
         }
         if (jdbc.update("UPDATE ma_storage_owner SET member_count = member_count + 1 WHERE storage_key = ?"
                 + " AND mount_state = 'READY' AND mount_revision = ? AND holder_key IS NULL"
-                + " AND maintenance_id IS NULL AND exclusive_member_key IS NULL",
+                + " AND maintenance_id IS NULL",
                 key, mountRevision) != 1) {
             throw busy();
         }
@@ -467,62 +467,52 @@ public class WorkspaceExecutionStore
         return Boolean.TRUE.equals(h2) ? "FOR UPDATE" : "FOR SHARE";
     }
 
-    private boolean directoryAdmits(String key, String member) {
-        List<Boolean> open = jdbc.query("SELECT holder_key, maintenance_id, mount_state, exclusive_member_key"
-                + " FROM ma_storage_owner WHERE storage_key = ?", (row, index) ->
+    private boolean directoryAdmits(String key) {
+        List<Boolean> open = jdbc.query("SELECT holder_key, maintenance_id, mount_state"
+                + " FROM ma_storage_owner WHERE storage_key = ? FOR UPDATE", (row, index) ->
                         row.getString("holder_key") == null && row.getString("maintenance_id") == null
-                        && "READY".equals(row.getString("mount_state"))
-                        && (row.getString("exclusive_member_key") == null
-                                || member.equals(row.getString("exclusive_member_key"))), key);
+                        && "READY".equals(row.getString("mount_state")), key);
         return open.size() == 1 && Boolean.TRUE.equals(open.getFirst());
     }
 
-    public boolean beginRewind(ContextBinding binding, RuntimeSessionRecord session) {
+    public void beginRewind(ContextBinding binding, RuntimeSessionRecord session) {
         if (storageGuard != null) {
             storageGuard.requireOwnedRoot(binding.getTenantId(), binding.getStorageId());
         }
-        return Boolean.TRUE.equals(transaction.execute(status -> {
+        transaction.executeWithoutResult(status -> {
             String key = storageKey(binding);
-            if (!isHeld(binding, session)) {
+            if (!isHeld(binding, session) || !directoryAdmits(key)) {
                 throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
             }
-            String member = exclusiveMemberKey(session);
-            if (jdbc.update("UPDATE ma_storage_owner SET exclusive_member_key = ? WHERE storage_key = ?"
-                    + " AND member_count = 1 AND mount_state = 'READY' AND exclusive_member_key IS NULL",
-                    member, key) != 1) {
-                if (member.equals(exclusiveMember(key))) {
+            if (jdbc.update("UPDATE ma_runtime_session SET is_rewind_pending = TRUE WHERE binding_id = ?"
+                    + " AND runtime_generation = ? AND runtime_session_id = ? AND harness_session_id = ?"
+                    + " AND execution_storage_key = ? AND session_state = 'READY' AND is_rewind_pending = FALSE",
+                    session.getBindingId(), session.getRuntimeGeneration(), session.getRuntimeSessionId(),
+                    session.getSession().getHarnessSessionId(), key) != 1) {
+                if (hasPendingRewind(session)) {
                     throw new RuntimeBrokerException(409, "workspace_rewind_recovery_required",
                             "An earlier Workspace rewind has an unconfirmed outcome.", false);
                 }
-                return false;
-            }
-            if (jdbc.update("UPDATE ma_runtime_session SET is_execution_exclusive = TRUE WHERE binding_id = ?"
-                    + " AND runtime_generation = ? AND runtime_session_id = ? AND execution_storage_key = ?",
-                    session.getBindingId(), session.getRuntimeGeneration(), session.getRuntimeSessionId(), key) != 1) {
-                throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
-            }
-            return true;
-        }));
-    }
-
-    public void endRewind(ContextBinding binding, RuntimeSessionRecord session) {
-        transaction.executeWithoutResult(status -> {
-            String key = storageKey(binding);
-            int cleared = jdbc.update("UPDATE ma_runtime_session SET is_execution_exclusive = FALSE WHERE binding_id = ?"
-                    + " AND runtime_generation = ? AND runtime_session_id = ? AND execution_storage_key = ?"
-                    + " AND is_execution_exclusive = TRUE",
-                    session.getBindingId(), session.getRuntimeGeneration(), session.getRuntimeSessionId(), key);
-            if (cleared == 1 && jdbc.update("UPDATE ma_storage_owner SET exclusive_member_key = NULL"
-                    + " WHERE storage_key = ? AND exclusive_member_key = ?", key, exclusiveMemberKey(session)) != 1) {
                 throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
             }
         });
     }
 
-    public boolean hasExclusiveClaim(RuntimeSessionRecord session) {
+    public void endRewind(ContextBinding binding, RuntimeSessionRecord session) {
+        transaction.executeWithoutResult(status -> {
+            String key = storageKey(binding);
+            jdbc.update("UPDATE ma_runtime_session SET is_rewind_pending = FALSE WHERE binding_id = ?"
+                    + " AND runtime_generation = ? AND runtime_session_id = ? AND harness_session_id = ? AND execution_storage_key = ?"
+                    + " AND is_rewind_pending = TRUE",
+                    session.getBindingId(), session.getRuntimeGeneration(), session.getRuntimeSessionId(),
+                    session.getSession().getHarnessSessionId(), key);
+        });
+    }
+
+    public boolean hasPendingRewind(RuntimeSessionRecord session) {
         return !jdbc.queryForList("SELECT runtime_session_id FROM ma_runtime_session WHERE binding_id = ?"
                 + " AND runtime_generation = ? AND runtime_session_id = ? AND execution_storage_key IS NOT NULL"
-                + " AND is_execution_exclusive = TRUE", String.class, session.getBindingId(), session.getRuntimeGeneration(),
+                + " AND is_rewind_pending = TRUE", String.class, session.getBindingId(), session.getRuntimeGeneration(),
                 session.getRuntimeSessionId()).isEmpty();
     }
 
@@ -537,7 +527,7 @@ public class WorkspaceExecutionStore
         if (!isHeld(binding, session)) {
             throw busy();
         }
-        if (hasExclusiveClaim(session)) {
+        if (hasPendingRewind(session)) {
             throw new RuntimeBrokerException(409, "workspace_rewind_recovery_required",
                     "Workspace rewind is active or has an unconfirmed outcome.", false);
         }
@@ -562,6 +552,14 @@ public class WorkspaceExecutionStore
     }
 
     public void release(ContextBinding binding, RuntimeSessionRecord session) {
+        release(binding, session, true);
+    }
+
+    public void releaseMount(ContextBinding binding, RuntimeSessionRecord session) {
+        release(binding, session, false);
+    }
+
+    private void release(ContextBinding binding, RuntimeSessionRecord session, boolean stopped) {
         if (residents.find(binding) != null) {
             // Read-only like the resident claim: no tenant guard.
             transaction.executeWithoutResult(status -> {
@@ -586,17 +584,25 @@ public class WorkspaceExecutionStore
             if (live.size() != 1 || !live.getFirst()) {
                 throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
             }
-            int cleared = jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_execution_exclusive = FALSE"
+            if (!stopped && !jdbc.queryForList("SELECT runtime_session_id FROM ma_runtime_session"
+                    + " WHERE binding_id = ? AND runtime_generation = ? AND runtime_session_id = ?"
+                    + " AND execution_storage_key = ? AND is_rewind_pending = TRUE FOR UPDATE",
+                    String.class, session.getBindingId(), session.getRuntimeGeneration(),
+                    session.getRuntimeSessionId(), key).isEmpty()) {
+                throw new RuntimeBrokerException(409, "workspace_writer_unproven",
+                        "A Workspace rewind must settle before its mount can be handed back.", false);
+            }
+            int cleared = jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_rewind_pending = FALSE"
                     + " WHERE execution_storage_key = ? AND binding_id = ? AND runtime_generation = ?"
                     + " AND runtime_session_id = ? AND harness_session_id = ?", key, session.getBindingId(),
                     session.getRuntimeGeneration(), session.getRuntimeSessionId(), session.getSession().getHarnessSessionId());
-            releaseMembers(jdbc, key, cleared, exclusiveMemberKey(session));
+            releaseMembers(jdbc, key, cleared);
         });
     }
 
     /**
      * The child Workspace maintenance holder (#13753 I1): a physical step
-     * of a child Workspace holds the storage exactly as a tool turn does,
+     * of a child Workspace excludes execution members,
      * so a turn meets it as workspace_busy and polls, while W1a fence and
      * registration, which need an idle lease, refuse. The Runtime holder
      * columns stay null and maintenance_id names the child Workspace. The
@@ -757,14 +763,10 @@ public class WorkspaceExecutionStore
                     saved.getGeneration(), saved.getOperationOwner(), saved.getOperationGeneration()) != 1) {
                 throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
             }
-            List<String> exclusiveSessions = jdbc.queryForList("SELECT runtime_session_id FROM ma_runtime_session"
-                    + " WHERE binding_id = ? AND runtime_generation = ? AND execution_storage_key = ?"
-                    + " AND is_execution_exclusive = TRUE", String.class, saved.getBindingId(), saved.getGeneration(), key);
-            int cleared = jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_execution_exclusive = FALSE"
+            int cleared = jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_rewind_pending = FALSE"
                     + " WHERE binding_id = ? AND runtime_generation = ? AND execution_storage_key = ?",
                     saved.getBindingId(), saved.getGeneration(), key);
-            releaseMembers(jdbc, key, cleared, exclusiveSessions.isEmpty()
-                    ? "" : exclusiveMemberKey(saved.getBindingId(), saved.getGeneration(), exclusiveSessions.getFirst()));
+            releaseMembers(jdbc, key, cleared);
         });
     }
 
@@ -790,45 +792,30 @@ public class WorkspaceExecutionStore
 
     static RuntimeBrokerException busy() {
         return new RuntimeBrokerException(409, "workspace_busy",
-                "Workspace storage is held by another tool turn.", true);
+                "Workspace is not available for this operation.", true);
     }
 
-    static String exclusiveMemberKey(RuntimeSessionRecord session) {
-        return exclusiveMemberKey(session.getBindingId(), session.getRuntimeGeneration(), session.getRuntimeSessionId());
-    }
-
-    public static String exclusiveMemberKey(String bindingId, long generation, String runtimeSessionId) {
-        return digest(bindingId + "\u0000" + generation + "\u0000" + runtimeSessionId);
-    }
-
-    static void releaseMembers(JdbcTemplate jdbc, String storageKey, int count, String exclusiveMemberKey) {
-        if (count > 0 && jdbc.update(RELEASE_MEMBERS, count, exclusiveMemberKey, storageKey, count) != 1) {
+    static void releaseMembers(JdbcTemplate jdbc, String storageKey, int count) {
+        if (count > 0 && jdbc.update(RELEASE_MEMBERS, count, storageKey, count) != 1) {
             throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
         }
     }
 
-    public static void releaseMembers(Connection connection, String storageKey, int count, String exclusiveMemberKey)
+    public static void releaseMembers(Connection connection, String storageKey, int count)
             throws SQLException {
         if (count <= 0) {
             return;
         }
         try (var statement = connection.prepareStatement(RELEASE_MEMBERS)) {
             statement.setInt(1, count);
-            statement.setString(2, exclusiveMemberKey);
-            statement.setString(3, storageKey);
-            statement.setInt(4, count);
+            statement.setString(2, storageKey);
+            statement.setInt(3, count);
             if (statement.executeUpdate() != 1) {
                 throw com.alibaba.qwen.code.managedagent.repository.WorkspaceExecutionStore.unavailable();
             }
         }
     }
 
-    private String exclusiveMember(String key) {
-        List<String> keys = jdbc.query("SELECT exclusive_member_key FROM ma_storage_owner WHERE storage_key = ?",
-                (row, index) -> row.getString("exclusive_member_key"), key);
-        return keys.size() == 1 ? keys.getFirst() : null;
-    }
-
     private String storageKey(ContextBinding binding) {
         return storageKey(jdbc, binding);
     }
@@ -843,8 +830,7 @@ public class WorkspaceExecutionStore
         return ssh == null ? LocalStorageOwners.key(jdbc, binding.getTenantId(), binding.getStorageId()) : ssh;
     }
 
-    private static final String RELEASE_MEMBERS = "UPDATE ma_storage_owner SET member_count = member_count - ?,"
-            + " exclusive_member_key = CASE WHEN exclusive_member_key = ? THEN NULL ELSE exclusive_member_key END"
+    private static final String RELEASE_MEMBERS = "UPDATE ma_storage_owner SET member_count = member_count - ?"
             + " WHERE storage_key = ? AND member_count >= ?";
 
     private static String maintenanceHolderKey(String maintenanceId, long claimGeneration) {

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

@@ -262,14 +262,13 @@ public class WorkspaceFileUploadStore {
             WorkspaceMigrationAdmission.requireOpen(jdbc, tenant, scope.binding().getStorageId());
             String physical = jdbc.queryForObject("SELECT physical_key FROM ma_storage WHERE alias_key = ? FOR UPDATE", String.class,
                     JdbcRuntimeBindingRepository.storageAliasKey(tenant, scope.binding().getStorageId()));
-            var owners = jdbc.queryForList("SELECT holder_key, maintenance_id, mount_state, mount_revision, exclusive_member_key FROM ma_storage_owner WHERE storage_key = ? FOR UPDATE", physical);
+            var owners = jdbc.queryForList("SELECT holder_key, maintenance_id, mount_state, mount_revision FROM ma_storage_owner WHERE storage_key = ? FOR UPDATE", physical);
             if (owners.size() != 1 || !"READY".equals(owners.getFirst().get("mount_state"))
                     || !(owners.getFirst().get("mount_revision") instanceof Number mount)
                     || !Long.valueOf(mount.longValue()).equals(scope.mountRevision())) {
                 throw changed();
             }
-            if (owners.getFirst().get("holder_key") != null || owners.getFirst().get("maintenance_id") != null
-                    || owners.getFirst().get("exclusive_member_key") != null) {
+            if (owners.getFirst().get("holder_key") != null || owners.getFirst().get("maintenance_id") != null) {
                 throw busy();
             }
             Map<String, Object> request = null;
@@ -302,8 +301,8 @@ public class WorkspaceFileUploadStore {
                     JdbcRuntimeBindingRepository.storageAliasKey(tenant, scope.binding().getStorageId()), physical,
                     com.alibaba.qwen.code.managedagent.repository.ManagedWorkspaceRegistry.actorKey(tenant, actor), digest, encode(target));
             if (jdbc.update("UPDATE ma_storage_owner SET member_count = member_count + 1 WHERE storage_key = ?"
-                    + " AND mount_state = 'READY' AND mount_revision = ? AND holder_key IS NULL AND maintenance_id IS NULL"
-                    + " AND exclusive_member_key IS NULL", physical, scope.mountRevision()) != 1) {
+                    + " AND mount_state = 'READY' AND mount_revision = ? AND holder_key IS NULL AND maintenance_id IS NULL",
+                    physical, scope.mountRevision()) != 1) {
                 throw busy();
             }
             return new Operation(key, digest, target, null, true);
@@ -369,8 +368,7 @@ public class WorkspaceFileUploadStore {
                 return current.result();
             }
             if (jdbc.update("UPDATE ma_storage_owner SET member_count = member_count - 1 WHERE storage_key = ?"
-                    + " AND member_count > 0 AND holder_key IS NULL AND maintenance_id IS NULL"
-                    + " AND exclusive_member_key IS NULL", target.physical()) != 1) {
+                    + " AND member_count > 0 AND holder_key IS NULL AND maintenance_id IS NULL", target.physical()) != 1) {
                 throw pending();
             }
             if (jdbc.update("UPDATE ma_operation SET state = ?, phase = ?, result_json = ?, completed_at = "

packages/managed-agent-server/persistence/src/main/java/com/alibaba/qwen/code/runtimebroker/managedschema/ManagedSchemaProjection.java

@@ -97,7 +97,7 @@ public enum ManagedSchemaProjection {
             "scope_key, runtime_session_id, tenant_id, workspace_id, workspace_generation, "
                     + "canonical_cwd, capability_digest, isolation_class, harness_session_id, turn_kind, "
                     + "binding_id, runtime_generation, session_state, record_version, last_active_at, "
-                    + "execution_storage_key, is_execution_exclusive"),
+                    + "execution_storage_key, is_rewind_pending"),
     RUNTIME_SLOT(
             "request_key, tenant_id, workspace_id, workspace_generation, canonical_cwd, "
                     + "capability_digest, isolation_class, isolation_key, provisioner_kind, storage_id, "
@@ -162,7 +162,7 @@ public enum ManagedSchemaProjection {
                     + "mount_birth_time, mount_registration_id, storage_kind, csi_phase, csi_revision, "
                     + "csi_reservation_id, csi_registration_key, csi_registration_revision, "
                     + "csi_provision_request_id, maintenance_id, maintenance_generation, physical_key, "
-                    + "member_count, exclusive_member_key"),
+                    + "member_count"),
     TENANT_GUARD(
             "tenant_key, tenant_id, guard_domain"),
     TOOL_EXECUTION(

packages/managed-agent-server/persistence/src/main/java/com/alibaba/qwen/code/runtimebroker/managedschema/ManagedSchemaRows.java

@@ -22,7 +22,7 @@ public final class ManagedSchemaRows {
             "is_default",
             "is_drain_requested",
             "is_enabled",
-            "is_execution_exclusive",
+            "is_rewind_pending",
             "is_paused",
             "is_plan_enabled",
             "is_projection_pending",

packages/managed-agent-server/persistence/src/main/java/com/alibaba/qwen/code/runtimebroker/persistence/JdbcRuntimeBindingRepository.java

@@ -202,16 +202,14 @@ public final class JdbcRuntimeBindingRepository
             }
         }
         try (var clear = connection.prepareStatement("UPDATE ma_runtime_session SET execution_storage_key = NULL,"
-                + " is_execution_exclusive = FALSE WHERE execution_storage_key = ? AND binding_id = ?"
+                + " is_rewind_pending = FALSE WHERE execution_storage_key = ? AND binding_id = ?"
                 + " AND runtime_generation = ? AND runtime_session_id = ?")) {
             clear.setString(1, key);
             JdbcRepositorySupport.setIdentity(clear, 2, session.getBindingId());
             clear.setLong(3, session.getRuntimeGeneration());
             JdbcRepositorySupport.setIdentity(clear, 4, session.getRuntimeSessionId());
             com.alibaba.qwen.code.managedagent.persistence.WorkspaceExecutionStore.releaseMembers(connection, key,
-                    clear.executeUpdate(), com.alibaba.qwen.code.managedagent.persistence.WorkspaceExecutionStore
-                            .exclusiveMemberKey(session.getBindingId(), session.getRuntimeGeneration(),
-                                    session.getRuntimeSessionId()));
+                    clear.executeUpdate());
         }
     }
 

packages/managed-agent-server/persistence/src/main/resources/db/managed-schema/V1__managed_schema.sql

@@ -1003,10 +1003,10 @@ CREATE TABLE ma_runtime_session (
         (binding_id, runtime_generation, session_state),
     INDEX idx_runtime_session_id (runtime_session_id),
     execution_storage_key CHAR(64) COMMENT 'Physical storage coordination key for active execution membership.',
-    is_execution_exclusive TINYINT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'Whether this execution membership requires exclusive storage access.',
+    is_rewind_pending TINYINT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'Whether this Runtime Session has an active or unconfirmed rewind.',
     id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'Database-generated identity of this row.',
     CONSTRAINT pk_ma_runtime_session_id PRIMARY KEY (id),
-    CONSTRAINT ck_ma_runtime_session_boolean_values CHECK ((is_execution_exclusive >= 0 AND is_execution_exclusive <= 1))
+    CONSTRAINT ck_ma_runtime_session_boolean_values CHECK ((is_rewind_pending >= 0 AND is_rewind_pending <= 1))
 ) ENGINE=InnoDB DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin COMMENT='Runtime execution Sessions attached to a binding and originating Harness Session.';
 
 CREATE TABLE ma_tool_execution (
@@ -1082,10 +1082,8 @@ CREATE TABLE ma_storage_owner (
     INDEX idx_execution_lease_maintenance (maintenance_id),
     physical_key CHAR(64) COMMENT 'Canonical physical storage identity shared by logical aliases.',
     member_count BIGINT UNSIGNED NOT NULL DEFAULT 0 COMMENT 'Number of Runtime Session memberships and running uploads on this storage.',
-    exclusive_member_key CHAR(64) COMMENT 'Identity of the execution member holding exclusive rewind access.',
     CONSTRAINT uk_ma_storage_owner_physical UNIQUE (physical_key),
     CONSTRAINT ck_ma_storage_owner_members CHECK (member_count >= 0),
-    CONSTRAINT ck_ma_storage_owner_exclusive CHECK (exclusive_member_key IS NULL OR member_count >= 1),
     CONSTRAINT uk_ma_storage_owner_business_key UNIQUE (storage_key),
     id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT 'Database-generated identity of this row.',
     CONSTRAINT pk_ma_storage_owner_id PRIMARY KEY (id)

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

@@ -673,7 +673,7 @@ class HostedProcessCrashIT {
                         .containsEntry("runtime_session_id", execution.get("runtime_session_id"))
                         .containsEntry("binding_id", execution.get("binding_id"))
                         .containsEntry("runtime_generation", execution.get("runtime_generation"))
-                        .containsEntry("session_state", "READY").containsEntry("is_execution_exclusive", false);
+                        .containsEntry("session_state", "READY").containsEntry("is_rewind_pending", false);
                 assertThat(((Number) member.get("runtime_generation")).longValue()).isEqualTo(1);
                 assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_runtime_binding b"
                         + " JOIN ma_runtime_session rs ON rs.binding_id=b.binding_id AND rs.runtime_generation=b.runtime_generation"

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

@@ -246,7 +246,7 @@ final class HostedProviderMediaProbe implements AutoCloseable {
                 tenant, storageId);
         assertThat(physical).isNotBlank();
         var member = jdbc.queryForMap("SELECT binding_id, runtime_generation, session_state,"
-                + " execution_storage_key, is_execution_exclusive FROM ma_runtime_session"
+                + " execution_storage_key, is_rewind_pending FROM ma_runtime_session"
                 + " WHERE tenant_id = ? AND harness_session_id = ? AND runtime_session_id = ?",
                 tenant, harness, runtimeId);
         boolean released = phase.equals("released") || phase.equals("restarted");
@@ -254,7 +254,7 @@ final class HostedProviderMediaProbe implements AutoCloseable {
         assertThat(member.get("runtime_generation")).as("original runtime generation").isEqualTo(row.get("runtime_generation"));
         assertThat(member.get("execution_storage_key")).as("physical execution membership")
                 .isEqualTo(released ? null : physical);
-        assertThat(member.get("is_execution_exclusive")).as("shared execution membership").isEqualTo(false);
+        assertThat(member.get("is_rewind_pending")).as("shared execution membership").isEqualTo(false);
         assertThat(member.get("session_state")).isEqualTo(released ? "RELEASED" : "READY");
         if (phase.equals("restarted")) {
             assertThat(restarted).isNotNull();

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

@@ -170,10 +170,10 @@ class HostedProviderMediaProbeTest {
         jdbc.update("UPDATE ma_runtime_session SET binding_id='binding', runtime_generation=2");
         assertThatThrownBy(() -> ReflectionTestUtils.invokeMethod(probe, "check", session, "prepared", fields))
                 .isInstanceOf(AssertionError.class).hasMessageContaining("original runtime generation");
-        jdbc.update("UPDATE ma_runtime_session SET runtime_generation=1, is_execution_exclusive=TRUE");
+        jdbc.update("UPDATE ma_runtime_session SET runtime_generation=1, is_rewind_pending=TRUE");
         assertThatThrownBy(() -> ReflectionTestUtils.invokeMethod(probe, "check", session, "prepared", fields))
                 .isInstanceOf(AssertionError.class).hasMessageContaining("shared execution membership");
-        jdbc.update("UPDATE ma_runtime_session SET is_execution_exclusive=FALSE");
+        jdbc.update("UPDATE ma_runtime_session SET is_rewind_pending=FALSE");
         jdbc.update("UPDATE ma_tool_execution SET execution_state='SETTLED', execution_status='success',"
                 + " dispatch_generation=1, result_json='{}'");
         ReflectionTestUtils.setField(probe, "dispatches", Map.of("runtime", 1));

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

@@ -136,7 +136,7 @@ final class HostedSseGapProbe implements AutoCloseable {
         }
         assertThat(execution.get("execution_call_id")).isEqualTo(executionId);
         String storage = "storage-0";
-        var member = jdbc.queryForMap("SELECT m.execution_storage_key, m.is_execution_exclusive, s.physical_key"
+        var member = jdbc.queryForMap("SELECT m.execution_storage_key, m.is_rewind_pending, s.physical_key"
                 + " FROM ma_runtime_session m JOIN ma_storage s ON s.tenant_id = m.tenant_id"
                 + " WHERE s.alias_key = ? AND s.tenant_id = ? AND s.storage_id = ?"
                 + " AND m.harness_session_id = ? AND m.binding_id = ?"
@@ -145,7 +145,7 @@ final class HostedSseGapProbe implements AutoCloseable {
                 tenant, storage, session, execution.get("binding_id"), execution.get("runtime_generation"), promptId);
         assertThat(member.get("physical_key")).isNotNull();
         assertThat(member.get("execution_storage_key")).isEqualTo(finished ? null : member.get("physical_key"));
-        assertThat(member.get("is_execution_exclusive")).isEqualTo(false);
+        assertThat(member.get("is_rewind_pending")).isEqualTo(false);
         assertThat(execution.get("turn_id")).isEqualTo(promptId);
         assertThat(execution.get("runtime_session_id")).isEqualTo(promptId);
         assertThat(execution.get("execution_state")).isEqualTo(finished ? "SETTLED" : "EXECUTING");

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

@@ -684,7 +684,7 @@ class HostedWorkspaceToolTurnIT {
                 tenant, storageId);
         assertThat(physical).isNotBlank();
         Map<String, Object> member = jdbc.queryForMap("SELECT binding_id, runtime_generation, runtime_session_id,"
-                + " session_state, execution_storage_key, is_execution_exclusive FROM ma_runtime_session"
+                + " session_state, execution_storage_key, is_rewind_pending FROM ma_runtime_session"
                 + " WHERE tenant_id = ? AND harness_session_id = ? AND runtime_session_id = ?",
                 tenant, session.get("sessionId"), report.path("promptId").asText());
         boolean released = storeFaults ? fault.equals("turn-reply") : List.of("prepare", "start", "release").contains(fault);
@@ -698,7 +698,7 @@ class HostedWorkspaceToolTurnIT {
         boolean retained = !storeFaults && fault.equals("prepare-twice");
         assertThat(member.get("execution_storage_key")).as(fault + " execution membership")
                 .isEqualTo(retained ? physical : null);
-        assertThat(member.get("is_execution_exclusive")).isEqualTo(false);
+        assertThat(member.get("is_rewind_pending")).isEqualTo(false);
         if (retained) {
             assertThat(member.get("binding_id")).isEqualTo(executions.getFirst().get("binding_id"));
             assertThat(member.get("runtime_generation")).isEqualTo(executions.getFirst().get("runtime_generation"));

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

@@ -126,7 +126,7 @@ class ManagedRuntimeRetentionRaceTest {
         fixture.retire();
         fixture.jdbc.update("UPDATE ma_storage_owner SET holder_key = NULL, binding_id = NULL,"
                 + " runtime_generation = NULL, runtime_session_id = NULL");
-        fixture.jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_execution_exclusive = FALSE");
+        fixture.jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_rewind_pending = FALSE");
         var locked = new CountDownLatch(1);
         var release = new CountDownLatch(1);
         var first = new AtomicBoolean(true);
@@ -281,7 +281,7 @@ class ManagedRuntimeRetentionRaceTest {
             if (reference == Reference.OPERATOR || reference == Reference.CSI_RETIREMENT) {
                 jdbc.update("UPDATE ma_storage_owner SET holder_key = NULL, binding_id = NULL,"
                         + " runtime_generation = NULL, runtime_session_id = NULL");
-                jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_execution_exclusive = FALSE");
+                jdbc.update("UPDATE ma_runtime_session SET execution_storage_key = NULL, is_rewind_pending = FALSE");
             }
         }
 

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

@@ -788,7 +788,7 @@ class WorkspaceCsiApplicationInventoryTest {
     void inventoriesEveryDeclaredTableAndPreservesRealPrivateCreationFacts() throws Exception {
         var snapshot = inspect();
         assertThat(snapshot.tables()).hasSize(com.alibaba.qwen.code.runtimebroker.schema.ManagedSchemaContract.TABLES.size());
-        assertThat(snapshot.tables().stream().map(table -> table.columns().size()).mapToInt(Integer::intValue).sum()).isEqualTo(810);
+        assertThat(snapshot.tables().stream().map(table -> table.columns().size()).mapToInt(Integer::intValue).sum()).isEqualTo(809);
         assertThat(snapshot.blockers()).noneMatch(blocker -> blocker.code().equals("schema_drift")
                 || blocker.code().equals("query_unavailable"));
         for (String table : new String[] {"ma_session", "ma_public_event", "ma_event_stream", "ma_storage", "ma_operation",

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

@@ -182,7 +182,7 @@ class WorkspaceCsiCheckpointSnapshotStoreTest {
         "UPDATE ma_tool_execution SET reference_json = REPLACE(reference_json, 'payloadDigest', 'wrongDigest')",
         "UPDATE ma_tool_execution SET reference_json = '{}'",
         "UPDATE ma_runtime_session SET binding_id = 'different'",
-        "UPDATE ma_runtime_session SET is_execution_exclusive = TRUE",
+        "UPDATE ma_runtime_session SET is_rewind_pending = TRUE",
         "UPDATE ma_output_capture SET is_quarantined = TRUE",
         "UPDATE ma_output_capture SET producer_phase = 'FINISHED'",
         "UPDATE ma_output_capture SET binding_digest = 'bad'",

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

@@ -440,6 +440,7 @@ class WorkspaceFileUploadStoreTest {
         String id = UUID.randomUUID().toString();
         if (runtimeFirst) {
             authority.claim(binding, runtime);
+            authority.beginRewind(binding, runtime);
         }
         try (var process = mockStatic(WorkspaceFileProcess.class)) {
             process.when(() -> WorkspaceFileProcess.upload(anyString(), anyString(), any())).thenAnswer(call -> {
@@ -452,23 +453,21 @@ class WorkspaceFileUploadStoreTest {
             assertSharedMembers(runtimeFirst ? 2 : 1);
             if (!runtimeFirst) {
                 authority.claim(binding, runtime);
+                authority.beginRewind(binding, runtime);
             }
-            authority.assertHeld(binding, runtime);
+            assertThat(authority.isHeld(binding, runtime)).isTrue();
+            assertThat(authority.hasPendingRewind(runtime)).isTrue();
             assertSharedMembers(2);
-            assertThat(authority.beginRewind(binding, runtime)).isFalse();
             if (runtimeFirst) {
                 assertThat(uploads.query("tenant", "actor", id).state()).isEqualTo("completed");
                 assertThat(uploads.query("tenant", "actor", id).state()).isEqualTo("completed");
                 assertSharedMembers(1);
-                authority.assertHeld(binding, runtime);
-                assertThat(authority.beginRewind(binding, runtime)).isTrue();
-                assertThatThrownBy(() -> upload(UUID.randomUUID().toString(), "rewind.bin", revision, new byte[0]))
-                        .isInstanceOfSatisfying(ManagedAgentException.class, error -> assertThat(error.getCode()).isEqualTo("workspace_busy"));
-                assertSharedMembers(1);
+                assertThat(authority.hasPendingRewind(runtime)).isTrue();
                 authority.endRewind(binding, runtime);
                 authority.release(binding, runtime);
             } else {
                 authority.release(binding, runtime);
+                assertThat(authority.hasPendingRewind(runtime)).isFalse();
                 authority.release(binding, runtime);
                 assertSharedMembers(1);
                 assertThatThrownBy(() -> guard.fence("tenant", storage(), 1, UUID.randomUUID().toString()))
@@ -627,8 +626,9 @@ class WorkspaceFileUploadStoreTest {
                 seed.getToken(), seed.getLeaseId(), seed.getEpoch());
         var ready = bindings.compareAndSet(claimed, claimed.withAttestation(lease,
                 new RuntimeResourceHandle("local-process", 1, Map.of("provider", "local-process")), Instant.now(), Instant.now()));
-        return bindings.admitSession(sessions, new RuntimeSessionRecord(new RuntimeSession(id, id, "bootstrap", scope),
+        var acquiring = bindings.admitSession(sessions, new RuntimeSessionRecord(new RuntimeSession(id, id, "bootstrap", scope),
                 ready.getBindingId(), ready.getGeneration(), RuntimeSessionRecord.State.ACQUIRING, 0, Instant.now()));
+        return sessions.compareAndSet(acquiring, acquiring.withState(RuntimeSessionRecord.State.READY, Instant.now()));
     }
 
     private void assertSharedMembers(long expected) {

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

@@ -1342,7 +1342,7 @@ class ChildWorkspaceServiceTest {
                 }
                 if (calls.incrementAndGet() > budget) {
                     throw new RuntimeBrokerException(409, "workspace_busy",
-                            "Workspace storage is held by another tool turn.", true);
+                            "Workspace is not available for this operation.", true);
                 }
                 return storageRoot;
             }

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/workspace/runtime/WorkspaceExecutionMembershipMySqlIT.java

@@ -4,7 +4,9 @@ import com.alibaba.qwen.code.managedagent.MySqlLockProbe;
 import com.alibaba.qwen.code.managedagent.VerificationFixtures;
 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.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.harness.SdkHarnessRules;
 import com.alibaba.qwen.code.managedagent.persistence.ChildWorkspaceStore;
 import com.alibaba.qwen.code.managedagent.persistence.IsolatedManagedSchemaMySqlTestDatabase;
@@ -18,6 +20,8 @@ import com.alibaba.qwen.code.managedagent.workspace.runtime.WorkspaceRecoveryCon
 import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRecord;
 import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
 import com.alibaba.qwen.code.runtimebroker.RuntimeRequestKeys;
+import com.alibaba.qwen.code.runtimebroker.RuntimeSession;
+import com.alibaba.qwen.code.runtimebroker.RuntimeSessionRecord;
 import com.alibaba.qwen.code.runtimebroker.ToolExecutionRecord;
 import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchema;
 import com.alibaba.qwen.code.runtimebroker.persistence.JdbcRuntimeBindingRepository;
@@ -43,7 +47,6 @@ import java.time.Duration;
 import java.time.Instant;
 import java.util.List;
 import java.util.Map;
-import java.util.Objects;
 import java.util.UUID;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
@@ -97,7 +100,7 @@ class WorkspaceExecutionMembershipMySqlIT {
                     checkMembers(fixture);
                     authority.claim(fixture.session.workspace(), runtime.session());
                     checkMembers(fixture);
-                    assertThat(authority.beginRewind(fixture.session.workspace(), runtime.session())).isTrue();
+                    authority.beginRewind(fixture.session.workspace(), runtime.session());
                     checkMembers(fixture);
                     authority.endRewind(fixture.session.workspace(), runtime.session());
                     checkMembers(fixture);
@@ -237,6 +240,8 @@ class WorkspaceExecutionMembershipMySqlIT {
             }
         }
         checkMembers(fixture);
+        authority.beginRewind(fixture.session.workspace(), runtime.session());
+        assertThatThrownBy(physical::call).isInstanceOf(RuntimeBrokerException.class);
         authority.release(fixture.session.workspace(), runtime.session());
         checkMembers(fixture);
         assertThat(physical.call()).isNotBlank();
@@ -244,10 +249,10 @@ class WorkspaceExecutionMembershipMySqlIT {
     }
 
     @Test
-    void exclusiveRewindRefusesACompetingMemberUntilItEnds() throws Exception {
+    void pendingRewindAllowsANeighborAfterItsShortAdmissionTransaction() throws Exception {
         var fixture = new Fixture(source, jdbc, store);
         var owner = fixture.runtime("owner");
-        var neighbor = fixture.runtime("neighbor");
+        var neighbor = fixture.runtime("neighbor", neighbor(fixture));
         authority.claim(fixture.session.workspace(), owner.session());
         checkMembers(fixture);
         var pause = new CommitPause(1);
@@ -259,22 +264,107 @@ class WorkspaceExecutionMembershipMySqlIT {
                 var claim = pool.submit(() -> authority.claim(fixture.session.workspace(), neighbor.session()));
                 MySqlLockProbe.awaitLockWaits(jdbc, schema, 1, "claim waiting for rewind");
                 pause.resume.countDown();
-                assertThat(rewind.get(5, TimeUnit.SECONDS)).isTrue();
-                assertRefused(claim);
+                rewind.get(5, TimeUnit.SECONDS);
+                claim.get(5, TimeUnit.SECONDS);
             } finally {
                 pause.resume.countDown();
             }
         }
         checkMembers(fixture);
-        authority.endRewind(fixture.session.workspace(), owner.session());
+        authority.assertHeld(fixture.session.workspace(), neighbor.session());
+        assertThat(authority.hasPendingRewind(owner.session())).isTrue();
+        authority.beginRewind(fixture.session.workspace(), neighbor.session());
         checkMembers(fixture);
-        authority.claim(fixture.session.workspace(), neighbor.session());
+        authority.endRewind(fixture.session.workspace(), owner.session());
+        assertThat(authority.hasPendingRewind(owner.session())).isFalse();
+        assertThat(authority.hasPendingRewind(neighbor.session())).isTrue();
         checkMembers(fixture);
         authority.release(fixture.session.workspace(), owner.session());
+        assertThat(authority.hasPendingRewind(neighbor.session())).isTrue();
         authority.release(fixture.session.workspace(), neighbor.session());
         checkMembers(fixture);
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    void rewindAdmissionAndMountHandbackHaveOneAtomicWinner(boolean rewindFirst) throws Exception {
+        var fixture = new Fixture(source, jdbc, store);
+        var runtime = fixture.runtime("rewind-handback");
+        authority.claim(fixture.session.workspace(), runtime.session());
+        var pause = new CommitPause(1);
+        var paused = authority(pause);
+        try (var pool = Executors.newFixedThreadPool(2)) {
+            try {
+                var first = pool.submit(() -> {
+                    if (rewindFirst) {
+                        paused.beginRewind(fixture.session.workspace(), runtime.session());
+                    } else {
+                        paused.releaseMount(fixture.session.workspace(), runtime.session());
+                    }
+                });
+                assertThat(pause.reached.await(5, TimeUnit.SECONDS)).isTrue();
+                var second = pool.submit(() -> {
+                    if (rewindFirst) {
+                        authority.releaseMount(fixture.session.workspace(), runtime.session());
+                    } else {
+                        authority.beginRewind(fixture.session.workspace(), runtime.session());
+                    }
+                });
+                MySqlLockProbe.awaitLockWaits(jdbc, schema, 1, "rewind and handback arbitration");
+                pause.resume.countDown();
+                first.get(5, TimeUnit.SECONDS);
+                assertThatThrownBy(() -> second.get(5, TimeUnit.SECONDS))
+                        .hasCauseInstanceOf(RuntimeBrokerException.class)
+                        .satisfies(error -> assertThat(((RuntimeBrokerException) error.getCause()).getCode())
+                                .isEqualTo(rewindFirst ? "workspace_writer_unproven" : "workspace_unavailable"));
+            } finally {
+                pause.resume.countDown();
+            }
+        }
+        assertThat(authority.hasPendingRewind(runtime.session())).isEqualTo(rewindFirst);
+        assertThat(authority.isHeld(fixture.session.workspace(), runtime.session())).isEqualTo(rewindFirst);
+        checkMembers(fixture);
+        authority.release(fixture.session.workspace(), runtime.session());
+        checkMembers(fixture);
+    }
+
+    @Test
+    void lostRecoveryClearsAllItsPendingMembersAndPreservesANeighbor() throws Exception {
+        var fixture = new Fixture(source, jdbc, store);
+        var original = fixture.runtime("original");
+        var acquiring = fixture.bindings.admitSession(fixture.sessions, new RuntimeSessionRecord(
+                new RuntimeSession(fixture.session.sessionId(), "second-original", "bootstrap", original.session().getSession().getScope()),
+                original.binding().getBindingId(), original.binding().getGeneration(), RuntimeSessionRecord.State.ACQUIRING,
+                0, Instant.now()));
+        var second = fixture.sessions.compareAndSet(acquiring, acquiring.withState(RuntimeSessionRecord.State.READY, Instant.now()));
+        var neighbor = fixture.runtime("neighbor", neighbor(fixture));
+        for (var member : List.of(original.session(), second, neighbor.session())) {
+            authority.claim(fixture.session.workspace(), member);
+            authority.beginRewind(fixture.session.workspace(), member);
+        }
+        checkMembers(fixture);
+        var lost = fixture.lose(original.binding());
+        var recovered = fixture.bindings.recoverLost(fixture.sessions, fixture.executions, lost);
+        authority.releaseLost(recovered);
+        for (var member : List.of(original.session(), second)) {
+            assertThat(authority.hasPendingRewind(member)).isFalse();
+            assertThat(authority.isHeld(fixture.session.workspace(), member)).isFalse();
+        }
+        assertThat(authority.hasPendingRewind(neighbor.session())).isTrue();
+        assertThat(authority.isHeld(fixture.session.workspace(), neighbor.session())).isTrue();
+        checkMembers(fixture);
+        authority.release(fixture.session.workspace(), neighbor.session());
+        checkMembers(fixture);
+    }
+
+    private String neighbor(Fixture fixture) {
+        return new org.springframework.transaction.support.TransactionTemplate(new DataSourceTransactionManager(source))
+                .execute(status -> store.insertWorkspaceSessionCommand(SessionCreationCommand.builder()
+                        .tenantId(fixture.tenant).actorId("actor").idempotencyKey(UUID.randomUUID().toString())
+                        .requestDigest("sha256:" + "b".repeat(64)).agentId("qwen-code")
+                        .selection(new WorkspaceSelection("workspace", ".")).build()).sessionId());
+    }
+
     private WorkspaceExecutionStore authority(DataSourceTransactionManager manager) {
         return VerificationFixtures.configure(new WorkspaceExecutionStore(jdbc, manager));
     }
@@ -299,18 +389,11 @@ class WorkspaceExecutionMembershipMySqlIT {
     private void checkMembers(Fixture fixture) {
         String key = jdbc.queryForObject("SELECT physical_key FROM ma_storage WHERE alias_key = ?", String.class,
                 JdbcRuntimeBindingRepository.storageAliasKey(fixture.tenant, fixture.session.workspace().getStorageId()));
-        var owner = jdbc.queryForMap("SELECT member_count, exclusive_member_key FROM ma_storage_owner WHERE storage_key = ?", key);
-        var members = jdbc.query("SELECT binding_id, runtime_generation, runtime_session_id, is_execution_exclusive"
-                + " FROM ma_runtime_session WHERE execution_storage_key = ?", (row, index) -> row.getBoolean("is_execution_exclusive")
-                        ? WorkspaceExecutionStore.exclusiveMemberKey(row.getString("binding_id"), row.getLong("runtime_generation"),
-                                row.getString("runtime_session_id")) : null, key);
-        assertThat(((Number) owner.get("member_count")).longValue()).isEqualTo(members.size());
-        var exclusive = members.stream().filter(Objects::nonNull).toList();
-        assertThat(exclusive).hasSizeLessThanOrEqualTo(1);
-        assertThat(owner.get("exclusive_member_key")).isEqualTo(exclusive.isEmpty() ? null : exclusive.getFirst());
-        if (!exclusive.isEmpty()) {
-            assertThat(members).hasSize(1);
-        }
+        long members = jdbc.queryForObject("SELECT COUNT(*) FROM ma_runtime_session WHERE execution_storage_key = ?", Long.class, key);
+        assertThat(jdbc.queryForObject("SELECT member_count FROM ma_storage_owner WHERE storage_key = ?", Long.class, key))
+                .isEqualTo(members);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_runtime_session WHERE execution_storage_key IS NULL"
+                + " AND is_rewind_pending = TRUE", Long.class)).isZero();
     }
 
     private static void assertRefused(Future<?> result) {

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/workspace/runtime/WorkspaceRecoveryContract.java

@@ -74,7 +74,11 @@ public final class WorkspaceRecoveryContract {
             assertThat(a.get(10, TimeUnit.SECONDS)).isTrue();
             assertThat(b.get(10, TimeUnit.SECONDS)).isTrue();
         }
-        assertThat(authority.beginRewind(fixture.session.workspace(), first.session())).isFalse();
+        authority.beginRewind(fixture.session.workspace(), first.session());
+        other.assertHeld(fixture.session.workspace(), second.session());
+        other.beginRewind(fixture.session.workspace(), second.session());
+        assertThat(authority.hasPendingRewind(first.session())).isTrue();
+        assertThat(authority.hasPendingRewind(second.session())).isTrue();
         var children = new com.alibaba.qwen.code.managedagent.persistence.ChildWorkspaceStore(jdbc,
                 new org.springframework.jdbc.datasource.DataSourceTransactionManager(source));
         var admitted = children.admit(fixture.tenant, fixture.session.sessionId(), UUID.randomUUID().toString(),
@@ -83,6 +87,9 @@ public final class WorkspaceRecoveryContract {
         assertThatThrownBy(() -> authority.holdForMaintenance(fixture.session.workspace(), maintenance.childWorkspaceId(), 1))
                 .isInstanceOf(RuntimeBrokerException.class);
         authority.release(fixture.session.workspace(), first.session());
+        assertThat(authority.hasPendingRewind(first.session())).isFalse();
+        assertThat(authority.hasPendingRewind(second.session())).isTrue();
+        other.endRewind(fixture.session.workspace(), second.session());
         // A locally released lifecycle row does not prove its physical membership ended.
         fixture.sessions.compareAndSet(second.session(), second.session().withState(RuntimeSessionRecord.State.RELEASED, Instant.now()));
         assertThatThrownBy(() -> authority.holdForMaintenance(fixture.session.workspace(), maintenance.childWorkspaceId(), 1))
@@ -93,7 +100,7 @@ public final class WorkspaceRecoveryContract {
         assertThatThrownBy(() -> other.claim(fixture.session.workspace(), first.session())).isInstanceOf(RuntimeBrokerException.class);
         authority.releaseMaintenance(fixture.session.workspace(), maintenance.childWorkspaceId(), 1, step);
         authority.claim(fixture.session.workspace(), first.session());
-        assertThat(authority.beginRewind(fixture.session.workspace(), first.session())).isTrue();
+        authority.beginRewind(fixture.session.workspace(), first.session());
         authority.endRewind(fixture.session.workspace(), first.session());
         authority.assertHeld(fixture.session.workspace(), first.session());
     }

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/managedagent/workspace/runtime/WorkspaceRuntimeTest.java

@@ -609,9 +609,9 @@ class WorkspaceRuntimeTest {
                     binding.getStorageId(), "child", binding.getContextConfigRef(), 1);
             assertUnavailable(() -> otherClient.claim(changedGeneration, second));
             authority.release(binding, first);
-            var undo = holder(secondSession, "undo");
+            var undo = ready(holder(secondSession, "undo"));
             otherClient.claim(binding, undo);
-            assertThat(otherClient.beginRewind(binding, undo)).isTrue();
+            otherClient.beginRewind(binding, undo);
             otherClient.endRewind(binding, undo);
             otherClient.release(binding, undo);
         }
@@ -657,7 +657,7 @@ class WorkspaceRuntimeTest {
 
     @org.junit.jupiter.params.ParameterizedTest
     @org.junit.jupiter.params.provider.ValueSource(booleans = {false, true})
-    void rewindDependsOnCurrentMembersRatherThanPastOverlap(boolean neighborHeld) throws Exception {
+    void rewindAllowsAnotherSessionWithCurrentMembership(boolean neighborHeld) throws Exception {
         var session = createSession("storage", ".");
         var fixture = transport(session);
         var ready = acquireReady(fixture);
@@ -672,41 +672,40 @@ class WorkspaceRuntimeTest {
         var neighbor = holder(sessions.findSessionById(created.sessionId()).orElseThrow(), "neighbor");
         authority.claim(session.workspace(), neighbor);
         var rewind = Map.<String, Object>of("kind", "raw-file-history", "action", "rewind", "promptId", "prompt");
-        if (neighborHeld) {
-            var snapshot = Map.of("ownerSessionId", session.sessionId(), "files", Map.of(), "snapshots", List.of());
-            when(fixture.http().control(any(), any(), eq(Map.of("kind", "raw-file-history", "action", "snapshot"))))
-                    .thenReturn(CompletableFuture.completedFuture(snapshot));
-            assertThat(fixture.transport().control(fixture.lease(), ready.getSession(), rewind).toCompletableFuture().join())
-                    .isEqualTo(Map.of("state", snapshot, "filesChanged", List.of(), "filesFailed", List.of(), "conflict", true));
-            verify(fixture.http(), never()).control(any(), any(), eq(rewind));
-            authority.assertHeld(session.workspace(), neighbor);
-        } else {
+        if (!neighborHeld) {
             authority.release(session.workspace(), neighbor);
-            var result = Map.of("state", Map.of(), "filesChanged", List.of("tracked.txt"), "filesFailed", List.of(), "conflict", false);
-            when(fixture.http().control(any(), any(), eq(rewind))).thenReturn(CompletableFuture.completedFuture(result));
-            assertThat(fixture.transport().control(fixture.lease(), ready.getSession(), rewind).toCompletableFuture().join())
-                    .isEqualTo(result);
-            verify(fixture.http()).control(any(), any(), eq(rewind));
+        }
+        var result = Map.of("state", Map.of(), "filesChanged", List.of("tracked.txt"), "filesFailed", List.of(), "conflict", false);
+        when(fixture.http().control(any(), any(), eq(rewind))).thenReturn(CompletableFuture.completedFuture(result));
+        assertThat(fixture.transport().control(fixture.lease(), ready.getSession(), rewind).toCompletableFuture().join())
+                .isEqualTo(result);
+        verify(fixture.http()).control(any(), any(), eq(rewind));
+        if (neighborHeld) {
+            authority.assertHeld(session.workspace(), neighbor);
         }
         authority.assertHeld(session.workspace(), ready);
     }
 
     @org.junit.jupiter.params.ParameterizedTest
     @org.junit.jupiter.params.provider.ValueSource(booleans = {false, true})
-    void rewindExcludesLateAdmissionAndRetainsExclusivityOnlyOnUncertainty(boolean uncertain) throws Exception {
+    void rewindAllowsLateSessionsAndRetainsOnlyItsOwnPendingStateOnUncertainty(boolean uncertain) throws Exception {
         var session = createSession("storage", ".");
         var fixture = transport(session);
         var ready = acquireReady(fixture);
-        var rival = holder(session, "late");
+        var rival = holder(neighbor(session), "late");
         var operation = Map.<String, Object>of("kind", "raw-file-history", "action", "rewind", "promptId", "prompt");
         var response = new CompletableFuture<Object>();
         when(fixture.http().control(any(), any(), eq(operation))).thenReturn(response);
         var rewinding = fixture.transport().control(fixture.lease(), ready.getSession(), operation).toCompletableFuture();
-        assertBusy(() -> authority.claim(session.workspace(), rival));
+        authority.claim(session.workspace(), rival);
+        authority.assertHeld(session.workspace(), rival);
+        assertThat(authority.hasPendingRewind(ready)).isTrue();
         if (uncertain) {
             response.completeExceptionally(new IllegalStateException("lost rewind result"));
             assertThatThrownBy(rewinding::join).hasRootCauseMessage("lost rewind result");
-            assertBusy(() -> authority.claim(session.workspace(), rival));
+            authority.claim(session.workspace(), rival);
+            authority.assertHeld(session.workspace(), rival);
+            assertThat(authority.hasPendingRewind(ready)).isTrue();
             assertThatThrownBy(() -> fixture.transport().releaseMount(fixture.lease(), ready.getSession()))
                     .isInstanceOfSatisfying(RuntimeBrokerException.class,
                             error -> assertThat(error.getCode()).isEqualTo("workspace_writer_unproven"));
@@ -720,26 +719,68 @@ class WorkspaceRuntimeTest {
             var result = Map.of("state", Map.of(), "filesChanged", List.of(), "filesFailed", List.of(), "conflict", false);
             response.complete(result);
             assertThat(rewinding.join()).isEqualTo(result);
+            assertThat(authority.hasPendingRewind(ready)).isFalse();
             authority.claim(session.workspace(), rival);
             authority.assertHeld(session.workspace(), ready);
             authority.assertHeld(session.workspace(), rival);
         }
     }
 
+    @Test
+    void twoSessionsKeepIndependentPendingRewindsUntilTheirOwnConfirmedRelease() throws Exception {
+        var session = createSession("storage", ".");
+        var first = transport(session);
+        var firstReady = acquireReady(first);
+        var second = transport(neighbor(session));
+        var secondReady = acquireReady(second);
+        var operation = Map.<String, Object>of("kind", "raw-file-history", "action", "rewind", "promptId", "prompt");
+        when(first.http().control(any(), any(), eq(operation)))
+                .thenReturn(CompletableFuture.failedFuture(new IllegalStateException("first outcome unknown")));
+        when(second.http().control(any(), any(), eq(operation)))
+                .thenReturn(CompletableFuture.failedFuture(new IllegalStateException("second outcome unknown")));
+        assertThatThrownBy(() -> first.transport().control(first.lease(), firstReady.getSession(), operation)
+                .toCompletableFuture().join()).hasRootCauseMessage("first outcome unknown");
+        assertThatThrownBy(() -> second.transport().control(second.lease(), secondReady.getSession(), operation)
+                .toCompletableFuture().join()).hasRootCauseMessage("second outcome unknown");
+        assertThat(authority.hasPendingRewind(firstReady)).isTrue();
+        assertThat(authority.hasPendingRewind(secondReady)).isTrue();
+        assertThat(first.transport().release(first.lease(), firstReady.getSession()).toCompletableFuture().join()).isTrue();
+        assertThat(authority.hasPendingRewind(firstReady)).isFalse();
+        assertThat(authority.isHeld(session.workspace(), firstReady)).isFalse();
+        assertThat(authority.hasPendingRewind(secondReady)).isTrue();
+        assertThat(authority.isHeld(session.workspace(), secondReady)).isTrue();
+        assertThat(second.transport().release(second.lease(), secondReady.getSession()).toCompletableFuture().join()).isTrue();
+        assertThat(authority.hasPendingRewind(secondReady)).isFalse();
+        assertThat(authority.isHeld(session.workspace(), secondReady)).isFalse();
+    }
+
+    @Test
+    void rewindRequiresReadyOriginalMembership() throws Exception {
+        var session = createSession("storage", ".");
+        var fixture = transport(session);
+        authority.claim(session.workspace(), fixture.record());
+        assertUnavailable(() -> authority.beginRewind(session.workspace(), fixture.record()));
+        var running = ready(fixture.record());
+        authority.beginRewind(session.workspace(), running);
+        authority.endRewind(session.workspace(), running);
+        authority.release(session.workspace(), running);
+        assertUnavailable(() -> authority.beginRewind(session.workspace(), running));
+    }
+
     @Test
     void releasesAnUnclaimedSessionWithoutActivatingItOrReleasingTheWorkspaceHolder() throws Exception {
         SessionRecord session = createSession("storage", ".");
         var fixture = transport(session);
-        var rival = holder(session, "holder");
+        var rival = ready(holder(session, "holder"));
         authority.claim(session.workspace(), rival);
-        assertThat(authority.beginRewind(session.workspace(), rival)).isTrue();
-        assertBusy(() -> fixture.transport().acquire(fixture.lease(), fixture.record().getSession()));
+        authority.beginRewind(session.workspace(), rival);
         var repository = new JdbcRuntimeSessionRepository(dataSource);
         var releasing = repository.compareAndSet(fixture.record(),
                 fixture.record().withState(RuntimeSessionRecord.State.RELEASING, Instant.now()));
         assertThat(releasing).isNotNull();
         assertThat(fixture.transport().release(fixture.lease(), releasing.getSession()).toCompletableFuture().join()).isTrue();
         assertThat(authority.isHeld(session.workspace(), rival)).isTrue();
+        assertThat(authority.hasPendingRewind(rival)).isTrue();
         assertUnavailable(() -> authority.claim(session.workspace(), releasing));
         verify(fixture.http(), never()).installContext(any(), any(), any(), any());
         verify(fixture.http(), never()).activateWorkspace(any(), any(), any(), any(Boolean.class));
@@ -805,10 +846,9 @@ class WorkspaceRuntimeTest {
     void mountHandbackOfANeverInstalledSessionOwesNoDetach() throws Exception {
         SessionRecord session = createSession("storage", ".");
         var fixture = transport(session);
-        var rival = holder(session, "holder");
+        var rival = ready(holder(session, "holder"));
         authority.claim(session.workspace(), rival);
-        assertThat(authority.beginRewind(session.workspace(), rival)).isTrue();
-        assertBusy(() -> fixture.transport().acquire(fixture.lease(), fixture.record().getSession()));
+        authority.beginRewind(session.workspace(), rival);
         assertThat(fixture.transport().releaseMount(fixture.lease(), fixture.record().getSession())
                 .toCompletableFuture().join()).isTrue();
         var releasing = new JdbcRuntimeSessionRepository(dataSource).compareAndSet(fixture.record(),
@@ -817,6 +857,7 @@ class WorkspaceRuntimeTest {
         assertThat(fixture.transport().release(fixture.lease(), releasing.getSession()).toCompletableFuture().join())
                 .isTrue();
         assertThat(authority.isHeld(session.workspace(), rival)).isTrue();
+        assertThat(authority.hasPendingRewind(rival)).isTrue();
         verify(fixture.http(), never()).release(any(), any());
         verify(fixture.http(), never()).activateWorkspace(any(), any(), any(), any(Boolean.class));
     }
@@ -1400,6 +1441,7 @@ class WorkspaceRuntimeTest {
     void heldTenantGuardDoesNotBlockMembershipChanges() throws Exception {
         SessionRecord session = createSession("storage", ".");
         var fixture = transport(session);
+        var running = ready(fixture.record());
         String storageKey = com.alibaba.qwen.code.managedagent.persistence.WorkspaceTestStorage.key(
                 session.tenantId(), session.workspace().getStorageId());
         try (var connection = dataSource.getConnection()) {
@@ -1411,11 +1453,11 @@ class WorkspaceRuntimeTest {
                     assertThat(rows.next()).isTrue();
                 }
             }
-            authority.claim(session.workspace(), fixture.record());
+            authority.claim(session.workspace(), running);
             assertThat(memberCount(storageKey)).isEqualTo(1L);
-            assertThat(authority.beginRewind(session.workspace(), fixture.record())).isTrue();
-            authority.endRewind(session.workspace(), fixture.record());
-            authority.release(session.workspace(), fixture.record());
+            authority.beginRewind(session.workspace(), running);
+            authority.endRewind(session.workspace(), running);
+            authority.release(session.workspace(), running);
             assertThat(memberCount(storageKey)).isZero();
             connection.rollback();
         }
@@ -1467,6 +1509,19 @@ class WorkspaceRuntimeTest {
                         runtime.getBindingId(), runtime.getGeneration(), RuntimeSessionRecord.State.ACQUIRING, 0, Instant.now()));
     }
 
+    private RuntimeSessionRecord ready(RuntimeSessionRecord session) {
+        return new JdbcRuntimeSessionRepository(dataSource).compareAndSet(session,
+                session.withState(RuntimeSessionRecord.State.READY, Instant.now()));
+    }
+
+    private SessionRecord neighbor(SessionRecord session) {
+        var created = sessions.insertWorkspaceSessionCommand(SessionCreationCommand.builder()
+                .tenantId(session.tenantId()).actorId("actor").idempotencyKey(UUID.randomUUID().toString())
+                .requestDigest("sha256:" + "b".repeat(64)).agentId("qwen-code")
+                .selection(new WorkspaceSelection("workspace", ".")).build());
+        return sessions.requireSession(session.tenantId(), created.sessionId());
+    }
+
     private static String digest(String text) throws Exception {
         return "sha256:" + HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(text.getBytes(StandardCharsets.UTF_8)));
     }

packages/managed-agent-server/verification/src/test/java/com/alibaba/qwen/code/runtimebroker/managedschema/ManagedSchemaRowsTest.java

@@ -58,7 +58,7 @@ class ManagedSchemaRowsTest {
             statement.execute("CREATE TABLE fixture (tenant_id VARBINARY(64), actor_id VARBINARY(64),"
                     + " record_bytes BLOB, revision BIGINT, is_enabled TINYINT, is_workspace_bound BOOLEAN,"
                     + " committed_sequence BIGINT, gmt_create BIGINT UNSIGNED,"
-                    + " is_execution_exclusive TINYINT)");
+                    + " is_rewind_pending TINYINT)");
             byte[] actor = {(byte) 0xff, 0};
             byte[] payload = {0, (byte) 0xfe, 10};
             long timestamp = java.time.Instant.parse("2026-10-14T01:02:03.123Z").toEpochMilli();
@@ -75,7 +75,7 @@ class ManagedSchemaRowsTest {
                 insert.executeUpdate();
             }
             try (var result = statement.executeQuery("SELECT tenant_id, actor_id, record_bytes,"
-                    + " revision, is_enabled, is_workspace_bound, committed_sequence, gmt_create, is_execution_exclusive,"
+                    + " revision, is_enabled, is_workspace_bound, committed_sequence, gmt_create, is_rewind_pending,"
                     + " CASE WHEN is_enabled = 1 THEN 1 ELSE 0 END AS is_quarantined FROM fixture")) {
                 result.next();
                 ResultSet driver = differentObjectTypes(result);
@@ -85,7 +85,7 @@ class ManagedSchemaRowsTest {
                 assertArrayEquals(payload, (byte[]) row.get("record_bytes"));
                 assertEquals(9_007_199_254_740_993L, row.get("revision"));
                 assertEquals(Boolean.TRUE, row.get("is_enabled"));
-                assertEquals(Boolean.TRUE, row.get("is_execution_exclusive"));
+                assertEquals(Boolean.TRUE, row.get("is_rewind_pending"));
                 assertEquals(1, row.get("is_quarantined"));
                 assertNull(row.get("is_workspace_bound"));
                 assertNull(row.get("committed_sequence"));