feat(managed-agent): register external Workspace storage ownership (#15516)

* feat(managed-agent): register external Workspace storage ownership

* fix(managed-agent): bound external storage registration guard locks

The registration store acquired its tenant and global guard locks through
the unbounded lockPlacementDomain overload (timeoutSeconds=0), so a lock
wait could block until innodb_lock_wait_timeout (~50s), far past the
store's own 10s transaction boundary. Add a bounded WorkspaceStorageKindGuard
.lockDomain overload and pass 10s at both call sites, matching the bounded
precedents elsewhere in the module. Also note on the global lock row that
it is a deliberate cross-tenant serialization row, not a tenant.

* fix(managed-agent): return contract 409 on duplicate-key registration races

The unique constraint guarding owner registration lives on physical_key,
while the pre-check selected on storage_key, and the inserts had no
DuplicateKeyException translation: a residual race past the global lock
would surface as a 500 instead of the contract 409 alias_conflict. Align
the pre-check with physical_key and wrap both INSERT statements in the
repository-standard duplicate-key-to-conflict translation.

* docs(managed-agent): note store ObjectMapper strictness boundary

Document why the store keeps a plain ObjectMapper: strict parsing is
enforced at the controller boundary, and this mapper only round-trips
payloads the store itself produced, matching the existing persistence
precedents.

docs/design/2026-10-25-external-storage-registration.md

@@ -0,0 +1,74 @@
+# External Workspace storage registration
+
+[English](2026-10-25-external-storage-registration.md) | [简体中文](2026-10-25-external-storage-registration.zh-CN.md)
+
+Status: implemented compatibility increment, acceptance pending CI. Tracks PR-A of
+[issue #15512](https://github.com/QwenLM/qwen-code/issues/15512).
+
+## Problem and scope
+
+The control plane needs to reserve provider-owned storage before it can safely
+bind a future external Workspace. Local directory identity cannot represent
+persistent provider storage: the server cannot inspect the remote filesystem,
+and a sandbox ID does not survive sandbox replacement.
+
+This increment registers immutable metadata. It does not create a Workspace,
+launch a worker, make remote calls, select a provisioner, or grant execution.
+CLI admission, adapters, Kubernetes, retirement and re-registration remain later
+increments. Existing LOCAL/UNREGISTERED behavior stays unchanged.
+
+## Contract and decisions
+
+The operator submits `ExternalStorageIdentity` (`providerId`, `physicalKey`,
+positive `registrationRevision`) and `StorageAlias` (`tenantWorkspaceId`,
+`aliasPath`, `mountRoot`). IDs use `[A-Za-z0-9._-]{1,128}`. Paths are absolute,
+non-root, at most 2048 characters, with nonempty `[A-Za-z0-9._-]+` components
+excluding `.` and `..`. This is a portable lexical contract, not filesystem
+attestation; remote permissions and existence remain unverified.
+
+`tenantWorkspaceId` reserves a tenant storage ID. Existing aliases, Workspaces
+and Runtime bindings cannot be adopted. The physical owner key hashes the
+provider and physicalKey, excluding revision, so a new revision cannot evade
+single-tenant ownership. A physical owner has one immutable alias. External alias
+paths are globally exclusive. Conflicts return sanitized HTTP 409 responses.
+
+The registration endpoint is persisted-workspace-scoped operator control:
+`POST /internal/managed-workspaces/v1/storage:register-external`. It requires an
+explicit loopback listener, loopback local and peer addresses, the existing
+Broker bearer credential and a tenant header. The credential is operator-wide;
+Workspace grants and forwarded headers grant no access. Input is bounded to
+16 KiB and rejects duplicate/unknown JSON fields. Responses are no-store.
+
+The tenant-scoped `idempotencyKey` uses `[A-Za-z0-9._:-]{1,128}`. Its digest covers
+only the registration. Optional visible-ASCII `requestId` (1–128 characters)
+sets HTTP correlation, not idempotency. Identical retries return the original
+registration ID and epoch-millisecond timestamp after restart; changed data with
+the same key returns `idempotency_conflict`. New requests cannot change an owner.
+
+## Persistence and compatibility
+
+Use existing `ma_storage`, `ma_storage_owner` and `ma_request_receipt` rows in one
+transaction. The only schema extension is the `EXTERNAL_STORAGE` receipt
+namespace in the fresh V1 baseline and its shared Java contract. Existing
+development catalogs must be recreated; there is no upgrade or repair path.
+
+Acquire the existing tenant placement lock before a shared external-registration
+lock. The latter serializes cross-tenant alias checks; the linear metadata scan
+is a deliberate tradeoff for this infrequent operator operation. Owner rows keep
+`mount_state=UNVERIFIED` and no holder. Local execution/recovery and CSI
+registration reject EXTERNAL aliases. No local Files call interprets their paths.
+
+## Validation and follow-up
+
+The focused registration tests exercise SQL-backed success/replay, restart,
+identity drift, tenant/alias collision, concurrent registration, legacy alias
+compatibility, execution refusal, loopback/bearer admission and sanitized errors.
+The same witnesses run against MySQL through the existing integration profile.
+Route admission and capability inventories include the operator-only endpoint;
+the CLI parity inventory explicitly excludes it from client capabilities.
+
+Local verification is limited to the new tests, compilation and Checkstyle under
+the shared ten-minute Java budget. Real MySQL, route contract suites and complete
+CI gates remain required before acceptance. Future activation must attest remote
+storage and install writer/generation fences before enabling tools; registration
+alone is never sufficient evidence.

docs/design/2026-10-25-external-storage-registration.zh-CN.md

@@ -0,0 +1,62 @@
+# External Workspace 存储注册
+
+[English](2026-10-25-external-storage-registration.md) | [简体中文](2026-10-25-external-storage-registration.zh-CN.md)
+
+状态：已实现兼容性增量，验收等待 CI。对应
+[issue #15512](https://github.com/QwenLM/qwen-code/issues/15512) 的 PR-A。
+
+## 问题与范围
+
+控制面需要先登记 provider 所有的存储，才能安全绑定未来的 external Workspace。
+本地目录身份无法表达 provider 的持久存储：服务器无法检查远端文件系统，sandbox ID
+也无法在 sandbox 更换后保持不变。
+
+本增量只注册不可变元数据，不创建 Workspace、不启动 worker、不发远程请求、不选择
+provisioner，也不授予执行权限。CLI admission、adapter、Kubernetes、退役与重新注册
+均留给后续增量。现有 LOCAL/UNREGISTERED 行为保持不变。
+
+## 契约与决策
+
+Operator 提交 `ExternalStorageIdentity`（`providerId`、`physicalKey`、正数
+`registrationRevision`）与 `StorageAlias`（`tenantWorkspaceId`、`aliasPath`、
+`mountRoot`）。ID 使用 `[A-Za-z0-9._-]{1,128}`。路径必须绝对、非根目录、最多 2048
+字符，由非空 `[A-Za-z0-9._-]+` 分量组成，并排除 `.` 与 `..`。这是可移植的词法
+契约，不是文件系统证明；远端权限与目录存在性仍未验证。
+
+`tenantWorkspaceId` 保留一个租户内 storage ID，不接管已有 alias、Workspace 或
+Runtime binding。物理 owner key 对 provider 与 physicalKey 做 hash，不包含 revision，
+使新 revision 无法绕过单租户 ownership。每个物理 owner 只对应一个不可变 alias。
+External alias path 全局排他。冲突返回脱敏的 HTTP 409。
+
+注册接口属于持久 Workspace 范围的 operator 控制面：
+`POST /internal/managed-workspaces/v1/storage:register-external`。它要求显式 loopback
+监听、本地与对端地址均为 loopback、现有 Broker bearer 凭据及 tenant header。
+该凭据是 operator 全局凭据；Workspace grant 与 forwarded header 不授予访问权限。
+输入上限为 16 KiB，拒绝重复或未知 JSON 字段，响应使用 no-store。
+
+租户范围内的 `idempotencyKey` 使用 `[A-Za-z0-9._:-]{1,128}`，请求 digest 仅包含
+registration。可选的可见 ASCII `requestId`（1–128 字符）只设置 HTTP 关联，不参与
+幂等身份。同内容重试在重启后仍返回原 registration ID 与毫秒时间戳；同 key 改内容
+返回 `idempotency_conflict`。新请求也不能修改已有 owner。
+
+## 持久化与兼容性
+
+在一个事务中写入现有 `ma_storage`、`ma_storage_owner` 与 `ma_request_receipt`。
+唯一 schema 扩展是在 fresh V1 baseline 及共享 Java 契约中增加 `EXTERNAL_STORAGE`
+receipt namespace。已有开发数据库需要重建；不提供升级或 repair 路径。
+
+先获取现有 tenant placement lock，再获取共享 external-registration lock。后者使跨
+租户 alias 检查串行化；元数据线性扫描是针对低频 operator 操作的明确取舍。Owner
+保持 `mount_state=UNVERIFIED` 且无 holder。本地执行、恢复及 CSI 注册拒绝 EXTERNAL
+alias；本地 Files 调用不会解释这些路径。
+
+## 验证与后续工作
+
+聚焦注册测试覆盖基于 SQL 的成功与重试、重启、身份漂移、租户及 alias 冲突、并发
+注册、旧 alias 兼容性、执行拒绝、loopback/bearer admission 和脱敏错误。同一组用例
+通过现有集成 profile 在 MySQL 上执行。路由 admission 与 capability 清单包含这个
+operator 专用接口；CLI parity 清单明确将其排除在客户端能力之外。
+
+本地验证仅运行新增测试、编译和 Checkstyle，共享 Java 预算为十分钟。真实 MySQL、
+路由契约测试与完整 CI gate 仍是验收前置条件。未来激活必须先证明远端存储身份，
+并安装 writer/generation fence，再开放工具；单纯注册永远不构成充分证明。

packages/cli/src/serve/capability-parity.ts

@@ -578,6 +578,8 @@ export const MANAGED_ONLY_NOT_A_GAP = new Set([
   'AUTOMATION_UPDATE',
   'CHANNEL_DELIVERY_LIST',
   'RUNTIME_NODE_LIST',
+  // Loopback operator registration has no client capability.
+  'EXTERNAL_STORAGE_REGISTER',
   // Internal writer and publication routes: between the control plane and its
   // own stores, behind writer credentials, never a client capability.
   'PUB_ADMISSION_PREPARE',

packages/managed-agent-server/docs/runtime-broker.md

@@ -204,6 +204,67 @@ retryable `runtime_external_capacity`. Plan a Broker restart or a capacity chang
 before that many distinct Sessions, and budget provider-side resource retention
 separately.
 
+## External Workspace storage registration
+
+Operators can reserve an immutable `EXTERNAL` storage ownership record through
+`POST /internal/managed-workspaces/v1/storage:register-external`. This is metadata
+registration only: it creates no Workspace, Session, worker, remote call or local
+filesystem access. It needs no external provisioner profile. Existing execution
+and local recovery paths refuse these aliases. See the
+[registration design](../../../docs/design/2026-10-25-external-storage-registration.md).
+
+The listener must be explicitly bound to loopback (`server.address` for a shared
+listener, or `qwen.managed-agent.internal-server.address` for a dedicated internal
+listener). Use literal `127.0.0.1` or `::1`. Both local and peer addresses must be
+loopback; forwarded headers do not grant access. A dedicated internal listener
+also requires its configured port. Send `Authorization: Bearer <runtime-broker.token>`
+and `X-Qwen-Tenant-Id`. An empty Broker token or a non-loopback listener is refused,
+even if ordinary API authentication is open. The operator credential authorizes
+the named tenant; ordinary Workspace grants do not authorize registration.
+
+```json
+{
+  "identity": {
+    "providerId": "provider-a",
+    "physicalKey": "persistent-volume-42",
+    "registrationRevision": 1
+  },
+  "alias": {
+    "tenantWorkspaceId": "workspace-42",
+    "aliasPath": "/external/workspace-42",
+    "mountRoot": "/workspace"
+  },
+  "idempotencyKey": "register-workspace-42",
+  "requestId": "operator-request-42"
+}
+```
+
+`providerId`, `physicalKey` and `tenantWorkspaceId` use `[A-Za-z0-9._-]{1,128}`;
+`registrationRevision` is positive. The provider must supply durable storage
+identity, never a sandbox ID. `tenantWorkspaceId` reserves the tenant's storage ID
+for a future Workspace; an existing alias, Workspace or Runtime binding is never
+adopted. Paths are absolute, non-root, at most 2048 characters and use nonempty
+`[A-Za-z0-9._-]+` components other than `.` and `..`. Backslashes, duplicate or
+trailing separators are rejected. Registration does not attest permissions or
+existence on the worker; its physical owner remains `UNVERIFIED`.
+
+A provider/physicalKey pair has one immutable tenant and alias, independent of
+revision. Alias paths are exclusive across external registrations, including
+across tenants. Attempts to change ownership return `409 alias_conflict` with a
+sanitized message. The tenant-scoped `idempotencyKey` uses
+`[A-Za-z0-9._:-]{1,128}`; the same key with different registration data returns
+`409 idempotency_conflict`. A committed retry returns the original `registrationId`,
+`storageKind`, `registration` and `registeredAt` (epoch milliseconds), including
+after restart. Optional `requestId` is visible ASCII of 1–128 characters and only
+sets HTTP correlation; changing it does not change the durable request digest.
+All responses are `Cache-Control: no-store`. Success is HTTP 200; invalid input is
+400, operator admission failure 403, and payloads over 16 KiB 413.
+
+Registration, owner and idempotency receipt commit atomically. The fresh V1 schema
+adds the `EXTERNAL_STORAGE` receipt namespace; existing development catalogs must
+be recreated under the existing fresh-schema policy. There is no update,
+re-registration, retirement, mount or execution API in this increment.
+
 ## Experimental Kubernetes scratch Runtime
 
 `KubernetesRuntimeProvisioner` implements the K1 slice of the

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/api/ExternalStorageRegistrationController.java

@@ -0,0 +1,86 @@
+package com.alibaba.qwen.code.managedagent.api;
+
+import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties;
+import com.alibaba.qwen.code.managedagent.persistence.ExternalStorageRegistration;
+import com.alibaba.qwen.code.managedagent.persistence.ExternalStorageRegistrationStore;
+import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import org.springframework.boot.autoconfigure.web.ServerProperties;
+import org.springframework.http.CacheControl;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+
+@RestController
+public final class ExternalStorageRegistrationController {
+    private final ExternalStorageRegistrationStore store;
+    private final ManagedAgentProperties properties;
+    private final ServerProperties server;
+    private final ObjectMapper json = new ObjectMapper()
+            .enable(JsonParser.Feature.STRICT_DUPLICATE_DETECTION)
+            .enable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES)
+            .enable(DeserializationFeature.FAIL_ON_TRAILING_TOKENS)
+            .disable(DeserializationFeature.ACCEPT_FLOAT_AS_INT);
+
+    public ExternalStorageRegistrationController(ExternalStorageRegistrationStore store,
+            ManagedAgentProperties properties, ServerProperties server) {
+        this.store = store;
+        this.properties = properties;
+        this.server = server;
+    }
+
+    public record Request(ExternalStorageRegistration.ExternalStorageIdentity identity,
+            ExternalStorageRegistration.StorageAlias alias, String idempotencyKey, String requestId) {
+    }
+
+    @PostMapping(value = "/internal/managed-workspaces/v1/storage:register-external", consumes = "application/json")
+    public ResponseEntity<ExternalStorageRegistrationStore.Receipt> register(
+            HttpServletRequest request, HttpServletResponse response) throws IOException {
+        response.setHeader("Cache-Control", "no-store");
+        var internal = properties.getInternalServer();
+        boolean loopbackListener = internal.getPort() > 0
+                ? request.getLocalPort() == internal.getPort() && loopback(internal.getAddress())
+                : server.getAddress() != null && server.getAddress().isLoopbackAddress();
+        String token = properties.getRuntimeBroker().getToken();
+        String authorization = request.getHeader("Authorization");
+        if (!loopbackListener || !loopback(request.getLocalAddr()) || !loopback(request.getRemoteAddr())
+                || token == null || token.isBlank() || authorization == null || authorization.length() > 4096
+                || !MessageDigest.isEqual(("Bearer " + token).getBytes(StandardCharsets.UTF_8),
+                        authorization.getBytes(StandardCharsets.UTF_8))) {
+            throw new RuntimeBrokerException(403, "external_storage_registration_denied",
+                    "External storage registration requires a loopback operator credential.", false);
+        }
+        String tenant = request.getHeader(TenantContextFilter.HEADER);
+        byte[] bytes = request.getInputStream().readNBytes(16385);
+        if (bytes.length > 16384) {
+            throw new RuntimeBrokerException(413, "payload_too_large", "Registration payload exceeds the limit.", false);
+        }
+        Request input;
+        ExternalStorageRegistration registration;
+        try {
+            input = json.readValue(bytes, Request.class);
+            if (input == null || input.requestId() != null && !input.requestId().matches("[\\x21-\\x7e]{1,128}")) {
+                throw new IllegalArgumentException();
+            }
+            registration = new ExternalStorageRegistration(input.identity(), input.alias());
+        } catch (IOException | IllegalArgumentException error) {
+            throw new RuntimeBrokerException(400, "invalid_external_storage_registration",
+                    "Invalid external storage registration.", false);
+        }
+        RequestIdFilter.useClientId(request, response, input.requestId());
+        return ResponseEntity.ok().cacheControl(CacheControl.noStore())
+                .body(store.registerExternalStorage(tenant, input.idempotencyKey(), registration));
+    }
+
+    private static boolean loopback(String address) {
+        return "127.0.0.1".equals(address) || "::1".equals(address) || "0:0:0:0:0:0:0:1".equals(address);
+    }
+}

packages/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/api/TenantContextFilter.java

@@ -27,6 +27,8 @@ public class TenantContextFilter extends OncePerRequestFilter {
             "/internal/managed-tool-publications/v1/";
     private static final String MANAGED_CHANNELS_PREFIX =
             "/internal/managed-channels/v1/";
+    private static final String MANAGED_WORKSPACES_PREFIX =
+            "/internal/managed-workspaces/v1/";
     private static final Pattern TENANT_PATTERN = Pattern.compile(
             "^[A-Za-z0-9._:-]{1,128}$");
     private final ObjectMapper objectMapper;
@@ -41,7 +43,8 @@ public class TenantContextFilter extends OncePerRequestFilter {
         return !PublicSurface.covers(path)
                 && !path.startsWith(MANAGED_SESSION_STORE_PREFIX)
                 && !path.startsWith(TOOL_PUBLICATION_PREFIX)
-                && !path.startsWith(MANAGED_CHANNELS_PREFIX);
+                && !path.startsWith(MANAGED_CHANNELS_PREFIX)
+                && !path.startsWith(MANAGED_WORKSPACES_PREFIX);
     }
 
     @Override
@@ -52,6 +55,7 @@ public class TenantContextFilter extends OncePerRequestFilter {
         if (path.startsWith(MANAGED_SESSION_STORE_PREFIX)
                 || path.startsWith(TOOL_PUBLICATION_PREFIX)
                 || path.startsWith(MANAGED_CHANNELS_PREFIX)
+                || path.startsWith(MANAGED_WORKSPACES_PREFIX)
                 || path.startsWith("/v1/agents/workspaces")
                 || path.startsWith("/api/agent/web-shell/v1/workspaces/")) {
             response.setHeader(HttpHeaders.CACHE_CONTROL, "no-store");

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

@@ -0,0 +1,66 @@
+package com.alibaba.qwen.code.managedagent.persistence;
+
+import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaKeys;
+
+/** Operator metadata only; registration does not attest or mount storage. */
+public record ExternalStorageRegistration(ExternalStorageIdentity identity, StorageAlias alias) {
+    public ExternalStorageRegistration {
+        if (identity == null || alias == null) {
+            throw invalid();
+        }
+    }
+
+    public record ExternalStorageIdentity(String providerId, String physicalKey, long registrationRevision) {
+        public ExternalStorageIdentity {
+            identifier(providerId);
+            identifier(physicalKey);
+            if (registrationRevision <= 0) {
+                throw invalid();
+            }
+        }
+
+        String ownerKey() {
+            // A revision must never create a second owner for the same provider storage.
+            return ManagedSchemaKeys.key("external-storage/1", providerId, physicalKey);
+        }
+
+        @Override
+        public String toString() {
+            return "ExternalStorageIdentity[redacted]";
+        }
+    }
+
+    public record StorageAlias(String tenantWorkspaceId, String aliasPath, String mountRoot) {
+        public StorageAlias {
+            identifier(tenantWorkspaceId);
+            path(aliasPath);
+            path(mountRoot);
+        }
+
+        @Override
+        public String toString() {
+            return "StorageAlias[redacted]";
+        }
+    }
+
+    private static void identifier(String value) {
+        if (value == null || !value.matches("[A-Za-z0-9._-]{1,128}")) {
+            throw invalid();
+        }
+    }
+
+    private static void path(String value) {
+        if (value == null || value.length() > 2048 || !value.startsWith("/") || value.equals("/")) {
+            throw invalid();
+        }
+        for (String part : value.substring(1).split("/", -1)) {
+            if (!part.matches("[A-Za-z0-9._-]+") || part.equals(".") || part.equals("..")) {
+                throw invalid();
+            }
+        }
+    }
+
+    private static IllegalArgumentException invalid() {
+        return new IllegalArgumentException("Invalid external storage registration.");
+    }
+}

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

@@ -0,0 +1,121 @@
+package com.alibaba.qwen.code.managedagent.persistence;
+
+import com.alibaba.qwen.code.runtimebroker.JdbcRuntimeBindingRepository;
+import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
+import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaContract.ReceiptNamespace;
+import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaContract.ReceiptScope;
+import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaKeys;
+import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchemaTime;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+@Component
+public final class ExternalStorageRegistrationStore {
+    private static final ReceiptNamespace NAMESPACE = ReceiptNamespace.EXTERNAL_STORAGE;
+    private final JdbcTemplate jdbc;
+    private final TransactionTemplate transaction;
+    // Strict parsing (duplicate/unknown/trailing rejection) lives at the controller
+    // boundary; this mapper only encodes/decodes payloads this store produced itself.
+    private final ObjectMapper json = new ObjectMapper();
+
+    public ExternalStorageRegistrationStore(JdbcTemplate jdbc, PlatformTransactionManager manager) {
+        this.jdbc = jdbc;
+        this.transaction = new TransactionTemplate(manager);
+        this.transaction.setTimeout(10);
+    }
+
+    public record Receipt(String registrationId, String storageKind,
+            ExternalStorageRegistration registration, long registeredAt) {
+    }
+
+    public Receipt registerExternalStorage(String tenant, String idempotencyKey,
+            ExternalStorageRegistration registration) {
+        if (tenant == null || !tenant.matches("[A-Za-z0-9._:-]{1,128}")
+                || idempotencyKey == null || !idempotencyKey.matches("[A-Za-z0-9._:-]{1,128}")
+                || registration == null) {
+            throw new RuntimeBrokerException(400, "invalid_external_storage_registration",
+                    "Invalid external storage registration.", false);
+        }
+        WorkspaceStorageKindGuard.requireFreshTransaction();
+        String encoded = encode(registration);
+        String digest = ManagedSchemaKeys.key(encoded);
+        String workspace = registration.alias().tenantWorkspaceId();
+        String aliasKey = JdbcRuntimeBindingRepository.storageAliasKey(tenant, workspace);
+        String physical = registration.identity().ownerKey();
+        return transaction.execute(status -> {
+            WorkspaceStorageKindGuard.lockDomain(jdbc, tenant, 10);
+            // Operator registration is infrequent. Serialize its cross-tenant alias check with a
+            // deliberate GLOBAL lock row (not a tenant) in tenant-keyed ma_tenant_guard.
+            WorkspaceStorageKindGuard.lockDomain(jdbc, "external-storage-registration/1", 10);
+            var prior = ManagedReceipts.find(jdbc, NAMESPACE, tenant, ReceiptScope.TENANT,
+                    tenant, idempotencyKey, true);
+            if (prior != null) {
+                if (!digest.equals(prior.requestDigest())) {
+                    throw conflict("idempotency_conflict");
+                }
+                return decode(prior.resultJson(), Receipt.class);
+            }
+            if (!jdbc.queryForList("SELECT alias_key FROM ma_storage WHERE alias_key = ?", aliasKey).isEmpty()
+                    || !jdbc.queryForList("SELECT workspace_id FROM ma_workspace WHERE tenant_id = ? AND (storage_id = ? OR workspace_id = ?)", tenant, workspace, workspace).isEmpty()
+                    || !jdbc.queryForList("SELECT binding_id FROM ma_runtime_binding WHERE storage_alias_key = ?", aliasKey).isEmpty()
+                    || !jdbc.queryForList("SELECT physical_key FROM ma_storage_owner WHERE physical_key = ?", physical).isEmpty()) {
+                throw conflict("alias_conflict");
+            }
+            for (String saved : jdbc.queryForList("SELECT registration_json FROM ma_storage WHERE storage_kind = 'EXTERNAL'", String.class)) {
+                var other = decode(saved, ExternalStorageRegistration.class);
+                if (registration.alias().aliasPath().equals(other.alias().aliasPath())) {
+                    throw conflict("alias_conflict");
+                }
+            }
+            try {
+                jdbc.update("INSERT INTO ma_storage_owner (storage_key, physical_key, tenant_id, storage_id,"
+                        + " storage_kind, mount_root, mount_revision) VALUES (?, ?, ?, ?, 'EXTERNAL', ?, ?)",
+                        physical, physical, tenant, workspace, registration.alias().mountRoot(),
+                        registration.identity().registrationRevision());
+                jdbc.update("INSERT INTO ma_storage (alias_key, tenant_key, storage_key, tenant_id, storage_id, physical_key,"
+                        + " storage_kind, registration_state, registration_revision, registration_json, gmt_create, gmt_modified)"
+                        + " VALUES (?, ?, ?, ?, ?, ?, 'EXTERNAL', 'REGISTERED', ?, ?, " + ManagedSchemaTime.SQL_NOW
+                        + ", " + ManagedSchemaTime.SQL_NOW + ")", aliasKey,
+                        JdbcRuntimeBindingRepository.storageFenceKey(tenant), JdbcRuntimeBindingRepository.storageFenceKey(workspace),
+                        tenant, workspace, physical, registration.identity().registrationRevision(), encoded);
+            } catch (DuplicateKeyException duplicate) {
+                // The global lock serializes registration, so the pre-checks above
+                // win first; translate any residual duplicate-key race (or a lock
+                // bypass) into the contract 409 instead of leaking a 500.
+                throw conflict("alias_conflict");
+            }
+            long now = jdbc.queryForObject("SELECT " + ManagedSchemaTime.SQL_NOW, Long.class);
+            String receiptKey = ManagedSchemaKeys.receiptKey(NAMESPACE, tenant, ReceiptScope.TENANT, tenant, idempotencyKey);
+            Receipt receipt = new Receipt(receiptKey, "EXTERNAL", registration, now);
+            ManagedReceipts.insert(jdbc, NAMESPACE, tenant, ReceiptScope.TENANT, tenant, idempotencyKey,
+                    digest, "REGISTER_EXTERNAL_STORAGE", null, null, null, "COMPLETED", null, null, now);
+            jdbc.update("UPDATE ma_request_receipt SET result_json = ? WHERE receipt_key = ?", encode(receipt), receiptKey);
+            return receipt;
+        });
+    }
+
+    private String encode(Object value) {
+        try {
+            return json.writeValueAsString(value);
+        } catch (JsonProcessingException error) {
+            throw WorkspaceExecutionStore.unavailable();
+        }
+    }
+
+    private <T> T decode(String value, Class<T> type) {
+        try {
+            return json.readValue(value, type);
+        } catch (JsonProcessingException | IllegalArgumentException error) {
+            throw WorkspaceExecutionStore.unavailable();
+        }
+    }
+
+    private static RuntimeBrokerException conflict(String code) {
+        return new RuntimeBrokerException(409, code, "External storage registration conflicts with existing ownership.", false);
+    }
+}

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

@@ -67,7 +67,7 @@ public final class WorkspaceCsiReservationStore {
         }
         transaction.executeWithoutResult(status -> {
             WorkspaceStorageKindGuard.lockDomain(jdbc, registration.tenantId());
-            if (jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE alias_key = ? AND storage_kind = 'LOCAL'",
+            if (jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE alias_key = ? AND storage_kind IN ('LOCAL', 'EXTERNAL')",
                     Long.class, registration.aliasKey()) != 0) {
                 throw unavailable();
             }

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

@@ -16,21 +16,27 @@ final class WorkspaceStorageKindGuard {
     }
 
     static void lockDomain(JdbcTemplate jdbc, String tenantId) {
+        lockDomain(jdbc, tenantId, 0);
+    }
+
+    // Bounded variant: caps each lock acquisition statement at timeoutSeconds
+    // so a caller cannot be frozen behind a long holder of the guard row.
+    static void lockDomain(JdbcTemplate jdbc, String tenantId, int timeoutSeconds) {
         jdbc.execute((ConnectionCallback<Void>) connection -> {
-            JdbcRuntimeBindingRepository.lockPlacementDomain(connection, tenantId);
+            JdbcRuntimeBindingRepository.lockPlacementDomain(connection, tenantId, timeoutSeconds);
             return null;
         });
     }
 
     static void requireLocalAlias(JdbcTemplate jdbc, String tenantId, String storageId) {
-        if (jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE alias_key = ? AND storage_kind IN ('CSI', 'SSH_LOCAL')",
+        if (jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE alias_key = ? AND storage_kind IN ('CSI', 'SSH_LOCAL', 'EXTERNAL')",
                 Long.class, WorkspaceCsiRegistration.aliasKey(tenantId, storageId)) != 0) {
             throw WorkspaceExecutionStore.unavailable();
         }
     }
 
     static void requireExecutionAlias(JdbcTemplate jdbc, String tenantId, String storageId) {
-        if (jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE alias_key = ? AND storage_kind = 'CSI'",
+        if (jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage WHERE alias_key = ? AND storage_kind IN ('CSI', 'EXTERNAL')",
                 Long.class, WorkspaceCsiRegistration.aliasKey(tenantId, storageId)) != 0) {
             throw WorkspaceExecutionStore.unavailable();
         }

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

@@ -82,7 +82,8 @@ public final class ManagedSchemaContract {
         AUTOMATION,
         CHANNEL_INGRESS,
         ACTION_DECISION,
-        WORKSPACE_CREATE
+        WORKSPACE_CREATE,
+        EXTERNAL_STORAGE
     }
 
     public enum ReceiptScope {

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

@@ -607,7 +607,7 @@ CREATE TABLE ma_request_receipt (
     PRIMARY KEY (receipt_key),
     CONSTRAINT uq_ma_decision_slot UNIQUE (tenant_key, decision_scope_key),
     CONSTRAINT uq_ma_request UNIQUE (namespace, tenant_key, scope_key, idempotency_digest),
-    CONSTRAINT ck_ma_request_namespace CHECK ((CASE namespace WHEN 'SESSION_COMMAND' THEN 15 WHEN 'SESSION_CREATE_MODE' THEN 19 WHEN 'SESSION_CREATE_ACTOR' THEN 20 WHEN 'AGENT_DEFINITION' THEN 16 WHEN 'AUTOMATION' THEN 10 WHEN 'CHANNEL_INGRESS' THEN 15 WHEN 'ACTION_DECISION' THEN 15 WHEN 'WORKSPACE_CREATE' THEN 16 ELSE -1 END = CHAR_LENGTH(namespace))),
+    CONSTRAINT ck_ma_request_namespace CHECK ((CASE namespace WHEN 'SESSION_COMMAND' THEN 15 WHEN 'SESSION_CREATE_MODE' THEN 19 WHEN 'SESSION_CREATE_ACTOR' THEN 20 WHEN 'AGENT_DEFINITION' THEN 16 WHEN 'AUTOMATION' THEN 10 WHEN 'CHANNEL_INGRESS' THEN 15 WHEN 'ACTION_DECISION' THEN 15 WHEN 'WORKSPACE_CREATE' THEN 16 WHEN 'EXTERNAL_STORAGE' THEN 16 ELSE -1 END = CHAR_LENGTH(namespace))),
     CONSTRAINT ck_ma_request_scope CHECK ((CASE scope_kind WHEN 'TENANT' THEN 6 WHEN 'ACTOR' THEN 5 WHEN 'SESSION' THEN 7 WHEN 'CHANNEL' THEN 7 ELSE -1 END = CHAR_LENGTH(scope_kind)))
 ) ENGINE=InnoDB DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_bin;
 

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

@@ -542,6 +542,7 @@ class SurfaceAdmissionAcceptanceTest {
             }
             case SESSION_OPTIONS ->
                 expect(entry, null, 401, "actor_required");
+            case INTERNAL_OPERATOR -> expect(entry, null, 403, "external_storage_registration_denied");
             case INTERNAL_WRITER ->
                 expectInternal(entry);
             default -> throw new AssertionError(

packages/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/api/SurfaceRegistry.java

@@ -392,6 +392,10 @@ public enum SurfaceRegistry {
             Surface.WEBSHELL, RuleClass.OWNER,
             EnumSet.of(Capability.SESSION_APPROVAL_CHANGE)),
     // ManagedSessionStoreController: internal writer routes.
+    INTERNAL_EXTERNAL_STORAGE_REGISTER(Method.POST,
+            "/internal/managed-workspaces/v1/storage:register-external",
+            Surface.INTERNAL, RuleClass.INTERNAL_OPERATOR,
+            EnumSet.of(Capability.EXTERNAL_STORAGE_REGISTER)),
     INTERNAL_APPROVAL_QUERY(Method.POST,
             "/internal/managed-session-store/v1/sessions/{sessionId}/approval:query",
             Surface.INTERNAL, RuleClass.INTERNAL_WRITER,
@@ -611,6 +615,7 @@ public enum SurfaceRegistry {
      * carries both {@link #TURN_SUBMIT} and {@link #TURN_CANCEL}.
      */
     public enum Capability {
+        EXTERNAL_STORAGE_REGISTER,
         USER_MESSAGE_QUERY, USER_MESSAGE_COMMAND, STORE_USER_MESSAGE_CLAIM,
         SESSION_CREATE,
         SESSION_LIST,
@@ -856,6 +861,7 @@ public enum SurfaceRegistry {
          * read). On a dedicated internal listener the routing filter
          * answers a public-connector caller {@code 404}.
          */
-        INTERNAL_WRITER
+        INTERNAL_WRITER,
+        INTERNAL_OPERATOR
     }
 }

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

@@ -0,0 +1,22 @@
+package com.alibaba.qwen.code.managedagent.persistence;
+
+import org.junit.jupiter.api.AfterEach;
+
+import javax.sql.DataSource;
+
+class ExternalStorageRegistrationMySqlIT extends ExternalStorageRegistrationTest {
+    private IsolatedManagedSchemaMySqlTestDatabase mysql;
+
+    @Override
+    DataSource source() {
+        mysql = new IsolatedManagedSchemaMySqlTestDatabase();
+        return mysql.source();
+    }
+
+    @AfterEach
+    void closeDatabase() {
+        if (mysql != null) {
+            mysql.close();
+        }
+    }
+}

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

@@ -0,0 +1,234 @@
+package com.alibaba.qwen.code.managedagent.persistence;
+
+import com.alibaba.qwen.code.managedagent.api.ApiExceptionHandler;
+import com.alibaba.qwen.code.managedagent.api.ExternalStorageRegistrationController;
+import com.alibaba.qwen.code.managedagent.api.RequestIdFilter;
+import com.alibaba.qwen.code.managedagent.api.TenantContextFilter;
+import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties;
+import com.alibaba.qwen.code.managedagent.persistence.ExternalStorageRegistration.ExternalStorageIdentity;
+import com.alibaba.qwen.code.managedagent.persistence.ExternalStorageRegistration.StorageAlias;
+import com.alibaba.qwen.code.runtimebroker.JdbcRuntimeBindingRepository;
+import com.alibaba.qwen.code.runtimebroker.RuntimeBrokerException;
+import com.alibaba.qwen.code.runtimebroker.managedschema.ManagedSchema;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.web.ServerProperties;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+import org.springframework.jdbc.datasource.DriverManagerDataSource;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+
+import javax.sql.DataSource;
+
+import java.net.InetAddress;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+class ExternalStorageRegistrationTest {
+    private static final String ROUTE = "/internal/managed-workspaces/v1/storage:register-external";
+    private final ObjectMapper json = new ObjectMapper();
+    private ManagedSchemaJdbcTemplate jdbc;
+    private DataSourceTransactionManager manager;
+    private ExternalStorageRegistrationStore store;
+    private ManagedAgentProperties properties;
+    private ServerProperties server;
+    private MockMvc mvc;
+
+    DataSource source() {
+        return new DriverManagerDataSource("jdbc:h2:mem:external-storage-" + UUID.randomUUID()
+                + ";MODE=MySQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE", "sa", "");
+    }
+
+    @BeforeEach
+    void setUp() throws Exception {
+        var source = source();
+        ManagedSchema.initialize(source);
+        jdbc = new ManagedSchemaJdbcTemplate(source);
+        manager = new DataSourceTransactionManager(source);
+        store = new ExternalStorageRegistrationStore(jdbc, manager);
+        properties = new ManagedAgentProperties();
+        properties.getRuntimeBroker().setToken("operator-secret");
+        server = new ServerProperties();
+        server.setAddress(InetAddress.getByName("127.0.0.1"));
+        mvc = MockMvcBuilders.standaloneSetup(new ExternalStorageRegistrationController(store, properties, server))
+                .setControllerAdvice(new ApiExceptionHandler())
+                .addFilters(new RequestIdFilter(), new TenantContextFilter(json)).build();
+    }
+
+    @Test
+    void registrationSurvivesRestartAndCorrelationChangesWithoutMutatingOwnership() throws Exception {
+        var registration = registration("provider", "volume", "workspace", "/aliases/one");
+        String body = body(registration, "key", "request-one");
+        var first = mvc.perform(operator().content(body)).andExpect(status().isOk())
+                .andExpect(header().string("X-Request-Id", "request-one"))
+                .andExpect(header().string("Cache-Control", "no-store"))
+                .andExpect(jsonPath("$.storageKind").value("EXTERNAL"))
+                .andReturn().getResponse().getContentAsString();
+        var owner = jdbc.queryForMap("SELECT * FROM ma_storage_owner");
+        assertThat(owner.get("mount_state")).isEqualTo("UNVERIFIED");
+        assertThat(owner.get("holder_key")).isNull();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_runtime_binding", Integer.class)).isZero();
+        ManagedSchema.initialize(jdbc.getDataSource());
+        var replay = new ExternalStorageRegistrationStore(jdbc, manager)
+                .registerExternalStorage("tenant", "key", registration);
+        assertThat(replay.registrationId()).isEqualTo(json.readTree(first).path("registrationId").asText());
+        assertThat(replay.registeredAt()).isEqualTo(json.readTree(first).path("registeredAt").asLong()).isPositive();
+        mvc.perform(operator().content(body(registration, "key", "request-two")))
+                .andExpect(status().isOk()).andExpect(header().string("X-Request-Id", "request-two"))
+                .andExpect(jsonPath("$.registrationId").value(replay.registrationId()));
+        assertThat(jdbc.queryForMap("SELECT * FROM ma_storage_owner")).isEqualTo(owner);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt", Integer.class)).isEqualTo(1);
+        assertThat(jdbc.queryForObject("SELECT result_json FROM ma_request_receipt", String.class))
+                .doesNotContain("request-one", "request-two", "operator-secret");
+    }
+
+    @Test
+    void conflictsNeverRebindAliasesPhysicalOwnersOrIdempotencyKeys() {
+        var original = registration("provider", "volume", "workspace", "/aliases/one");
+        store.registerExternalStorage("tenant", "key", original);
+        var owner = jdbc.queryForMap("SELECT * FROM ma_storage_owner");
+        var alias = jdbc.queryForMap("SELECT * FROM ma_storage");
+        for (var changed : List.of(
+                registration("provider", "other-volume", "workspace", "/aliases/one"),
+                registration("other-provider", "volume", "workspace", "/aliases/one"),
+                new ExternalStorageRegistration(new ExternalStorageIdentity("provider", "volume", 2), original.alias()),
+                registration("provider", "volume", "other-workspace", "/aliases/two"))) {
+            assertConflict(() -> store.registerExternalStorage("tenant", "key", changed), "idempotency_conflict");
+            assertConflict(() -> store.registerExternalStorage("tenant", "other-key", changed), "alias_conflict");
+        }
+        assertConflict(() -> store.registerExternalStorage("other-tenant", "key",
+                registration("provider", "volume", "other-workspace", "/aliases/two")), "alias_conflict");
+        assertConflict(() -> store.registerExternalStorage("other-tenant", "key",
+                registration("other-provider", "other-volume", "other-workspace", "/aliases/one")), "alias_conflict");
+        assertThat(jdbc.queryForMap("SELECT * FROM ma_storage_owner")).isEqualTo(owner);
+        assertThat(jdbc.queryForMap("SELECT * FROM ma_storage")).isEqualTo(alias);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt", Integer.class)).isEqualTo(1);
+        store.registerExternalStorage("other-tenant", "key",
+                registration("other-provider", "volume", "other-workspace", "/aliases/two"));
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage_owner", Integer.class)).isEqualTo(2);
+    }
+
+    @Test
+    void concurrentCrossTenantAliasRegistrationHasExactlyOneWinner() throws Exception {
+        var start = new CountDownLatch(1);
+        try (var pool = Executors.newFixedThreadPool(2)) {
+            var attempts = List.of("one", "two").stream().map(tenant -> pool.submit(() -> {
+                start.await(5, TimeUnit.SECONDS);
+                try {
+                    store.registerExternalStorage(tenant, "key", registration("provider", tenant, "workspace", "/shared"));
+                    return "registered";
+                } catch (RuntimeBrokerException error) {
+                    return error.getCode();
+                }
+            })).toList();
+            start.countDown();
+            assertThat(List.of(attempts.get(0).get(10, TimeUnit.SECONDS), attempts.get(1).get(10, TimeUnit.SECONDS)))
+                    .containsExactlyInAnyOrder("registered", "alias_conflict");
+        }
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage", Integer.class)).isEqualTo(1);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage_owner", Integer.class)).isEqualTo(1);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt", Integer.class)).isEqualTo(1);
+    }
+
+    @Test
+    void externalRegistrationCannotEnterLocalExecutionAndCannotAdoptLegacyAliases() {
+        for (String kind : List.of("LOCAL", "UNREGISTERED")) {
+            String alias = JdbcRuntimeBindingRepository.storageAliasKey("tenant", kind);
+            jdbc.update("INSERT INTO ma_storage (alias_key, tenant_key, storage_key, tenant_id, storage_id, storage_kind) VALUES (?, ?, ?, ?, ?, ?)",
+                    alias, JdbcRuntimeBindingRepository.storageFenceKey("tenant"), JdbcRuntimeBindingRepository.storageFenceKey(kind), "tenant", kind, kind);
+            assertThat(LocalStorageOwners.alias(jdbc, "tenant", kind, false).get("storage_kind")).isEqualTo(kind);
+            WorkspaceStorageKindGuard.requireLocalAlias(jdbc, "tenant", kind);
+            WorkspaceStorageKindGuard.requireExecutionAlias(jdbc, "tenant", kind);
+            assertConflict(() -> store.registerExternalStorage("tenant", kind,
+                    registration("provider", kind, kind, "/aliases/" + kind)), "alias_conflict");
+        }
+        store.registerExternalStorage("tenant", "key", registration("provider", "volume", "workspace", "/aliases/one"));
+        assertThatThrownBy(() -> WorkspaceStorageKindGuard.requireLocalAlias(jdbc, "tenant", "workspace"))
+                .isInstanceOf(RuntimeBrokerException.class);
+        assertThatThrownBy(() -> WorkspaceStorageKindGuard.requireExecutionAlias(jdbc, "tenant", "workspace"))
+                .isInstanceOf(RuntimeBrokerException.class);
+        assertThatThrownBy(() -> LocalStorageOwners.key(jdbc, "tenant", "workspace"))
+                .isInstanceOf(RuntimeBrokerException.class);
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage_owner", Integer.class)).isEqualTo(1);
+    }
+
+    @Test
+    void onlyLoopbackOperatorCanRegisterAndForwardedHeadersGrantNoAccess() throws Exception {
+        String body = body(registration("provider", "volume", "workspace", "/aliases/one"), "key", "request");
+        mvc.perform(post(ROUTE).header("X-Qwen-Tenant-Id", "tenant").contentType("application/json").content(body))
+                .andExpect(status().isForbidden()).andExpect(header().string("Cache-Control", "no-store"));
+        mvc.perform(operator().content(body).with(request -> {
+            request.setRemoteAddr("192.0.2.1");
+            return request;
+        }).header("X-Forwarded-For", "127.0.0.1")).andExpect(status().isForbidden());
+        properties.getInternalServer().setPort(9091);
+        properties.getInternalServer().setAddress("127.0.0.1");
+        mvc.perform(operator().content(body)).andExpect(status().isForbidden());
+        properties.getInternalServer().setPort(0);
+        server.setAddress(InetAddress.getByName("0.0.0.0"));
+        mvc.perform(operator().content(body)).andExpect(status().isForbidden());
+        server.setAddress(InetAddress.getByName("127.0.0.1"));
+        properties.getRuntimeBroker().setToken("");
+        mvc.perform(operator().content(body)).andExpect(status().isForbidden());
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage", Integer.class)).isZero();
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_request_receipt", Integer.class)).isZero();
+    }
+
+    @Test
+    void unsafePathsAndNonStorageIdentitiesAreRejectedWithoutLeakingInput() throws Exception {
+        String valid = body(registration("provider", "volume", "workspace", "/aliases/one"), "key", "request");
+        for (String invalid : List.of("/", "/a/../b", "/a//b", "/a/", "relative", "C:\\private")) {
+            var document = json.readTree(valid);
+            ((com.fasterxml.jackson.databind.node.ObjectNode) document.path("alias")).put("mountRoot", invalid);
+            mvc.perform(operator().content(document.toString())).andExpect(status().isBadRequest())
+                    .andExpect(jsonPath("$.error.code").value("invalid_external_storage_registration"))
+                    .andExpect(jsonPath("$.error.message").value("Invalid external storage registration."));
+        }
+        mvc.perform(operator().content(valid.replace("\"registrationRevision\":1", "\"registrationRevision\":1.5")))
+                .andExpect(status().isBadRequest());
+        mvc.perform(operator().content(valid.replace("\"volume\"", "\"sandbox/id\"")))
+                .andExpect(status().isBadRequest());
+        mvc.perform(operator().content(valid.replace("\"identity\":", "\"identity\":{},\"identity\":")))
+                .andExpect(status().isBadRequest());
+        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM ma_storage", Integer.class)).isZero();
+    }
+
+    private MockHttpServletRequestBuilder operator() {
+        return post(ROUTE).contentType("application/json").header("X-Qwen-Tenant-Id", "tenant")
+                .header("Authorization", "Bearer operator-secret").with(request -> {
+                    request.setLocalAddr("127.0.0.1");
+                    request.setRemoteAddr("127.0.0.1");
+                    return request;
+                });
+    }
+
+    private String body(ExternalStorageRegistration registration, String key, String requestId) throws Exception {
+        return json.writeValueAsString(new ExternalStorageRegistrationController.Request(
+                registration.identity(), registration.alias(), key, requestId));
+    }
+
+    private static ExternalStorageRegistration registration(String provider, String physical, String workspace, String path) {
+        return new ExternalStorageRegistration(new ExternalStorageIdentity(provider, physical, 1),
+                new StorageAlias(workspace, path, "/worker/workspace"));
+    }
+
+    private static void assertConflict(Runnable action, String code) {
+        assertThatThrownBy(action::run).isInstanceOfSatisfying(RuntimeBrokerException.class, error -> {
+            assertThat(error.getStatusCode()).isEqualTo(409);
+            assertThat(error.getCode()).isEqualTo(code);
+            assertThat(error.getMessage()).isEqualTo("External storage registration conflicts with existing ownership.");
+        });
+    }
+}

packages/managed-agent-server/src/test/resources/capability-parity/managed-surface.txt

@@ -1,5 +1,5 @@
 # Generated by ManagedSurfaceCaptureTest (capture).
-# sha256 of the route lines below: 32a38f52e11df5f235dd16614a8653fa0329e9c10ac9716b8ddd97fb007aa73b
+# sha256 of the route lines below: fa9544118b901d736dafe742efe0eacfd01940a6df607582055b92187842f3b2
 AUTOMATION_RETIRE DELETE /v1/agent-automations/{automationId} public
 SESSION_DELETE DELETE /v1/agents/sessions/{sessionId} public
 STORE_RESOURCE_GET GET /internal/managed-session-store/v1/sessions/{sessionId}/resources/{resourceId} internal
@@ -114,6 +114,7 @@ PUB_SEGMENT POST /internal/managed-tool-publications/v1/sessions/{sessionId}/pub
 PUB_STREAM_PREFIX POST /internal/managed-tool-publications/v1/sessions/{sessionId}/publications/{publicationId}/streams/{streamId}/prefix internal
 PUB_STREAM_SEAL POST /internal/managed-tool-publications/v1/sessions/{sessionId}/publications/{publicationId}/streams/{streamId}/seal internal
 PUB_RECEIPT_VERIFY POST /internal/managed-tool-publications/v1/sessions/{sessionId}/receipts/verify internal
+EXTERNAL_STORAGE_REGISTER POST /internal/managed-workspaces/v1/storage:register-external internal
 AUTOMATION_CREATE POST /v1/agent-automations public
 AUTOMATION_UPDATE POST /v1/agent-automations/{automationId} public
 AUTOMATION_RUN POST /v1/agent-automations/{automationId}/runs public