From f9e546551e5b91ac7a65722d108bcc99146bb4d8 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 10:54:07 +0000 Subject: [PATCH 1/4] Reuse finished Environment preparation across Sessions Core computes a key from the Project, initial files and setup operations and calls RestorePreparation/SavePreparation on the Provider. On a hit only finalize runs. The microsandbox helper owns storage, a 24 hour TTL and an 8 GiB bound. The daemon rebinds an identical restored capability installation. --- .../internal/localworkspace/capabilities.go | 4 + .../capability_preparation_test.go | 16 +- .../localworkspace/capability_rebind_test.go | 37 +++ contracts/agents-api/environments.md | 4 + .../agents-api/node-generation-protocol.md | 4 +- contracts/agents-api/zh/environments.md | 6 +- .../agents-api/zh/node-generation-protocol.md | 6 +- docs/sandbox-provider.md | 3 + docs/zh/sandbox-provider.md | 5 +- internal/agentcapabilities/install.go | 15 ++ .../server/managed_generation_operations.go | 22 ++ .../provider_operations_fixture_test.go | 38 ++-- .../execution/runtime_initialization.go | 36 ++- .../internal/execution/runtime_manager.go | 11 + .../execution/runtime_preparation_cache.go | 87 +++++++ .../runtime_preparation_cache_test.go | 48 ++++ .../internal/sandbox/docker/operations.go | 38 ++-- .../core/internal/sandbox/e2b/operations.go | 38 ++-- .../internal/sandbox/microsandbox/identity.go | 4 + .../sandbox/microsandbox/operations.go | 32 +-- .../internal/sandbox/microsandbox/provider.go | 18 ++ .../internal/sandbox/microsandbox/types.go | 5 +- .../core/internal/sandbox/node/operations.go | 32 +-- .../node/provider_operations_fixture_test.go | 70 +++--- services/core/internal/sandbox/node/proxy.go | 8 + services/core/internal/sandbox/node/wire.go | 12 + .../core/internal/sandbox/sandbox_provider.go | 13 ++ .../internal/sandbox/sandbox_provider_test.go | 2 +- .../provider_operations_fixture_test.go | 70 +++--- .../tools/microsandbox-provider/backend.go | 5 + .../microsandbox-provider/preparation.go | 214 ++++++++++++++++++ .../microsandbox-provider/preparation_test.go | 35 +++ 32 files changed, 778 insertions(+), 160 deletions(-) create mode 100644 apps/daemon/internal/localworkspace/capability_rebind_test.go create mode 100644 services/core/internal/execution/runtime_preparation_cache.go create mode 100644 services/core/internal/execution/runtime_preparation_cache_test.go create mode 100644 services/core/tools/microsandbox-provider/preparation.go create mode 100644 services/core/tools/microsandbox-provider/preparation_test.go diff --git a/apps/daemon/internal/localworkspace/capabilities.go b/apps/daemon/internal/localworkspace/capabilities.go index c2eb36090..4199aac3f 100644 --- a/apps/daemon/internal/localworkspace/capabilities.go +++ b/apps/daemon/internal/localworkspace/capabilities.go @@ -150,6 +150,10 @@ func (b *Binding) loadCapabilitySnapshot(root *os.Root, input agentcapabilities. return agentcapabilities.Manifest{}, agentcapabilities.ErrInvalid } manifest, err := agentcapabilities.Load(root) + if err == nil && !completed && manifest.Identity != identity { + // A preparation restored from an identical earlier Session carries that Session's identity. + err = agentcapabilities.Rebind(root, &manifest, input, identity) + } if err != nil || agentcapabilities.ValidateSelection(manifest, input, identity) != nil { return agentcapabilities.Manifest{}, agentcapabilities.ErrInvalid } diff --git a/apps/daemon/internal/localworkspace/capability_preparation_test.go b/apps/daemon/internal/localworkspace/capability_preparation_test.go index acead77d4..8bb807e19 100644 --- a/apps/daemon/internal/localworkspace/capability_preparation_test.go +++ b/apps/daemon/internal/localworkspace/capability_preparation_test.go @@ -66,20 +66,20 @@ func TestPreparationFreezesLocalContentsAcrossReconnect(t *testing.T) { if err != nil || string(fresh[len(fresh)-6:]) != "second" { t.Fatal("new Session did not capture new source", err) } - // Reusing a snapshot under another identity or selection cannot start native work. - reconnect.stateKey = uuid.NewString() - changed := configured - changed.AgentStateKey = reconnect.stateKey - if _, err = reconnect.Prepare(t.Context(), changed); err == nil { - t.Fatal("foreign snapshot accepted") - } - changed = configured + // Reusing a snapshot under another selection cannot start native work, even + // for another Session. local := *configured.LocalEnvironment local.CapabilitySources = &agentcapabilities.Input{} + changed := configured changed.LocalEnvironment = &local if _, err = b.Prepare(t.Context(), changed); err == nil { t.Fatal("changed selection accepted") } + reconnect.stateKey = uuid.NewString() + changed.AgentStateKey = reconnect.stateKey + if _, err = reconnect.Prepare(t.Context(), changed); err == nil { + t.Fatal("foreign snapshot accepted") + } } func TestPreparationUsesOperatorSourcesAndLeavesFailuresInert(t *testing.T) { diff --git a/apps/daemon/internal/localworkspace/capability_rebind_test.go b/apps/daemon/internal/localworkspace/capability_rebind_test.go new file mode 100644 index 000000000..e1f4cbd3a --- /dev/null +++ b/apps/daemon/internal/localworkspace/capability_rebind_test.go @@ -0,0 +1,37 @@ +package localworkspace + +import ( + "testing" + + "github.com/MiniMax-AI/OpenAgentCore/internal/agentcapabilities" + "github.com/google/uuid" +) + +func TestFinalizeRebindsIdenticalRestoredInstallation(t *testing.T) { + first, _ := testBinding(t) + source := t.TempDir() + writeSourceSkill(t, source, "restored") + input := agentcapabilities.Input{Directories: []string{source}} + root, unlock, err := first.openCapabilities(t.Context(), false) + if err != nil { + t.Fatal(err) + } + defer unlock() + if _, err = first.loadCapabilitySnapshot(root, input, first.capabilityIdentity(), false); err != nil { + t.Fatal(err) + } + // The next Session's directory holds the files restored from the first. + next, err := NewWithCapabilityDirectory(uuid.NewString(), uuid.NewString(), first.workspace, first.capabilityRoot) + if err != nil { + t.Fatal(err) + } + if _, err = next.loadCapabilitySnapshot(root, agentcapabilities.Input{}, next.capabilityIdentity(), false); err == nil { + t.Fatal("different selection accepted") + } + if _, err = next.loadCapabilitySnapshot(root, input, next.capabilityIdentity(), false); err != nil { + t.Fatal(err) + } + if manifest, err := agentcapabilities.Load(root); err != nil || manifest.Identity != next.capabilityIdentity() { + t.Fatal("identity not rebound", err) + } +} diff --git a/contracts/agents-api/environments.md b/contracts/agents-api/environments.md index 573f2ebec..03dc68551 100644 --- a/contracts/agents-api/environments.md +++ b/contracts/agents-api/environments.md @@ -316,6 +316,10 @@ The Session applies the Template first, then its own fields. Composition happens For comparison, an inline `openai_hosted` Session without a Template treats omitted and null `network` as enabled. +### Preparation reuse + +Core may start a hosted Environment from the result of an earlier preparation in the same Project whose effective configuration, after Template composition, is identical, for up to 24 hours after that preparation. Each Session still gets its own copy. Setup commands and package installation then do not run again for this Session. Only `/environment` is reused (see the workspace filesystem scope); HOME and other compute-local changes are not. Setup must not create per-Session unique values, perform external side effects that must happen per Session, or fetch credentials that expire within 24 hours; configurations that need this must differ, for example in `env` or `files`. A Provider that does not declare the [preparation cache](../../docs/sandbox-provider.md#explicit-operation-contracts) always prepares in full. + ### Restricted network `restricted` requires 1–100 exact ASCII hostnames in `allowed_domains`; subdomains and redirect targets need their own entries. Wildcards, URL or port syntax, IP literals, Unicode and trailing dots are rejected. Reads return the supplied spelling, order and duplicates; comparison uses a separate lowercase, deduplicated copy. diff --git a/contracts/agents-api/node-generation-protocol.md b/contracts/agents-api/node-generation-protocol.md index e7dcc3156..6edd472b8 100644 --- a/contracts/agents-api/node-generation-protocol.md +++ b/contracts/agents-api/node-generation-protocol.md @@ -85,12 +85,14 @@ Each operation carries its own arguments and returns the following result on suc | `suspend` | `Suspend` | `suspend` | `state` | | `resume` | `Resume` | `resume` | `state` | | `delete_snapshot` | `DeleteSnapshot` | `snapshot` | None | +| `restore_preparation` | `RestorePreparation` | `key` | `restored` | +| `save_preparation` | `SavePreparation` | `key` | None | A request whose `connection_id`, `owner_epoch` or `sequence` does not match closes the connection. A malformed request gets an `invalid` response. The node runs the request on the provider of the named `deployment_generation` and answers `unconfirmed` when it cannot. Core sends `create` and a `resume` that is not observe-only only to a generation that is ready on that node, and keeps at most 32 requests pending per connection. The budget is relative: the node anchors `timeout_ms` to its own clock on receipt and consumes it while the request waits in its queue, so the hosts' clocks need not agree. Core still bounds its own wait. A full node queue closes the connection. -The `response` frame carries `id` and `connection_id`. A successful response carries the result named in the operation table, with no result field for `kill`, `kill_compute` or `delete_snapshot`. A failed response carries an `error_code`: +The `response` frame carries `id` and `connection_id`. A successful response carries the result named in the operation table, with no result field for `kill`, `kill_compute`, `delete_snapshot` or `save_preparation`. A failed response carries an `error_code`: | `error_code` | Meaning | | --- | --- | diff --git a/contracts/agents-api/zh/environments.md b/contracts/agents-api/zh/environments.md index 9b03039ab..3c0fd8c29 100644 --- a/contracts/agents-api/zh/environments.md +++ b/contracts/agents-api/zh/environments.md @@ -1,7 +1,7 @@ --- title: "Environment 与模板" source: contracts/agents-api/environments.md -source_hash: 3408e2c811042732d35227150533cf12a85cfcff8a5118d7eb62a5d68549cc8b +source_hash: f50217dd1fc465626ac7135a46b69cea0ee40fc14fd30141e23813018192cb5c --- Environment 是 Session 的执行资源，包括 Harness 运行所在的机器、工作区以及已完成准备的能力。Session 通过其 `environment` 配置创建 Environment；不存在独立的 create 调用。Environment 模板是 Session 创建时解析的可复用准备配置。本契约涵盖这两类资源、两种放置方式、输入接纳、能力准备、Skills、Plugins 和 MCP 连接来源。 @@ -318,6 +318,10 @@ Session 会先应用模板，然后应用自身字段。组合仅在加密 Sessi 作为比较，不带模板的内联 `openai_hosted` Session 会将省略或为 null 的 `network` 视为启用。 +### 准备复用 {#preparation-reuse} + +Core 可以用同一 Project 中较早一次准备的结果启动托管 Environment，前提是两者在 Template 组合之后的有效配置完全相同，且距该次准备不超过 24 小时。每个 Session 仍然得到自己的副本。此时设置命令和包安装不会为该 Session 再次运行。只复用 `/environment`（见工作区文件系统范围），HOME 和其他与计算实例本地相关的改动不会复用。设置过程不得产生每个 Session 唯一的值，不得执行必须按 Session 发生的外部副作用，也不得获取 24 小时内就会过期的凭据；需要这些行为的配置必须有所不同，例如在 `env` 或 `files` 上。未声明[准备缓存](../../../docs/zh/sandbox-provider.md#explicit-operation-contracts)的提供商始终完整准备。 + ### 受限网络 {#restricted-network} `restricted` 要求在 `allowed_domains` 中提供 1–100 个精确 ASCII 主机名；子域名和重定向目标需要各自的条目。通配符、URL 或端口语法、IP 字面量、Unicode 和尾随点都会被拒绝。读取操作会返回提供的拼写、顺序和重复项；比较操作使用单独的小写去重副本。 diff --git a/contracts/agents-api/zh/node-generation-protocol.md b/contracts/agents-api/zh/node-generation-protocol.md index bddaaa69f..5931080f4 100644 --- a/contracts/agents-api/zh/node-generation-protocol.md +++ b/contracts/agents-api/zh/node-generation-protocol.md @@ -1,7 +1,7 @@ --- title: "沙箱节点协议" source: contracts/agents-api/node-generation-protocol.md -source_hash: 21fb07f531d4087de59618e5f34211d5dc9674efe97530501170950e8a76f6f2 +source_hash: a8135ed519bdc897237cef7f69c78dde638ab76877c67f096b1e5aef9f28b80b --- 沙箱节点协议连接 Core 与 Docker 或 microsandbox 主机：主机针对其本地提供商执行提供商操作，并报告就绪状态、主机测量值及其持有的部署代次。 @@ -87,12 +87,14 @@ Core 发送包含以下内容的 `request` 帧： | `suspend` | `Suspend` | `suspend` | `state` | | `resume` | `Resume` | `resume` | `state` | | `delete_snapshot` | `DeleteSnapshot` | `snapshot` | 无 | +| `restore_preparation` | `RestorePreparation` | `key` | `restored` | +| `save_preparation` | `SavePreparation` | `key` | 无 | 只要 `connection_id`、`owner_epoch` 或 `sequence` 中任一值不匹配，请求就会关闭连接。格式错误的请求会得到 `invalid` 响应。节点在所指 `deployment_generation` 的提供商上运行请求，无法运行时回复 `unconfirmed`。Core 仅向节点上已就绪的代次发送 `create` 和非 observe-only 的 `resume`，并且每条连接最多保留 32 个待处理请求。 预算采用相对计时：节点收到请求时以自己的时钟为基准锚定 `timeout_ms`，并在请求排队等待期间持续消耗该预算，因此各主机的时钟无需保持一致。Core 仍会限制自身等待时长。节点队列已满时会关闭连接。 -`response` 帧包含 `id` 和 `connection_id`。成功响应携带操作表中指定的结果；对于 `kill`、`kill_compute` 或 `delete_snapshot`，响应不含结果字段。失败响应携带一个 `error_code`： +`response` 帧包含 `id` 和 `connection_id`。成功响应携带操作表中指定的结果；对于 `kill`、`kill_compute`、`delete_snapshot` 或 `save_preparation`，响应不含结果字段。失败响应携带一个 `error_code`： | `error_code` | 含义 | | --- | --- | diff --git a/docs/sandbox-provider.md b/docs/sandbox-provider.md index 71dd64cb7..5b6afc3ae 100644 --- a/docs/sandbox-provider.md +++ b/docs/sandbox-provider.md @@ -63,9 +63,12 @@ Every provider returns a complete `ProviderOperations()` declaration with one en | `Create`, `GetInfo`, `Renew`, `Kill`, `RunCommand` | Supported | Allocation lifecycle and bounded commands | | `Observe` | Explicit decision | Ownership-checked read-only observation of one allocation | | `Initial`, `NewCompute`, `GetCompute`, `Suspend`, `Resume`, `ResumeCompute`, `KillCompute`, `DeleteSnapshot`, `RunCommandCompute` | The same decision for every checkpoint method; supported only by a `nodes` registration | Exact compute incarnations, capture and restore, retained-source resume and cleanup | +| `RestorePreparation`, `SavePreparation` | The same decision for both | Reuse of a finished Environment preparation | `Initial` and `NewCompute` construct compute references without allocating, `ResumeCompute` thaws only the same resident instance after an aborted pause, and `RunCommandCompute` runs a bounded command in one exact compute incarnation. Core uses `RunCommandCompute` to wake a parked daemon after a restore ([`runtime_compute_wake.go`](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/services/core/internal/execution/runtime_compute_wake.go)). +`SavePreparation(reference, key)` stores the `/environment` filesystem of the allocation's initial compute after a complete preparation, and `RestorePreparation(reference, key)` puts a stored one back and returns `false` when it holds no valid entry. The `key` is opaque: Core derives it from the Project and everything that reaches preparation, never from the Environment, Session, Harness or Agent, and the Provider never interprets it. The Provider owns storage, scoping to its Runtime image, expiry and eviction. An entry is valid for 24 hours after its save, and a restore must not use an expired one. Core calls `RestorePreparation` after the Runtime connects and, on a hit, sends only the `finalize` operation before completing initialization. On a miss it prepares in full and calls `SavePreparation`; a save error is logged and never fails the Session. A Provider that declares neither always prepares in full. The microsandbox adapter keeps one archive per key in a private directory under the Runtime home. + Each declaration entry is `state: supported` with no reason, or `state: unsupported` with an authored reason code. Missing, zero, unknown or unsafe entries fail validation. Adding a method to `SandboxProvider` requires an explicit decision and implementation in every adapter; never supply a base type or generate blanket unsupported implementations. An unsupported method returns `providercontract.UnsupportedError` before any native I/O. The error names the exact operation and a safe code, never a native message, resource identity, endpoint or credential. An empty result, a nil error, `Unavailable` or an unknown mutation outcome never stands in for unsupported, and the five required methods can never return it. diff --git a/docs/zh/sandbox-provider.md b/docs/zh/sandbox-provider.md index 62948b727..c70680410 100644 --- a/docs/zh/sandbox-provider.md +++ b/docs/zh/sandbox-provider.md @@ -1,7 +1,7 @@ --- title: "沙箱提供商协议" source: docs/sandbox-provider.md -source_hash: 19dcc4268e7d7ed547c1de470ba1bf0f771a35e909a4f869daaeefeddffeea7c +source_hash: 0b333cf78e2a40479398e57e9e3d7cac7799fb05a5fa976da7e82c6f6893d65d --- 沙箱提供商协议连接 Core 与为 Core 管理的（`openai_hosted`）Environment 提供计算资源的 adapter，并由该 adapter 运行启动 Runtime daemon 的有界引导流程。本页面向 adapter 作者：说明提供商必须实现什么，以及可以依赖 Core 做什么。Docker、microsandbox 和 E2B 实现了该协议。 @@ -65,9 +65,12 @@ Docker 等没有原生可续期租约的 backend 仍遵守 Core 的 hosted expir | `Create`、`GetInfo`、`Renew`、`Kill`、`RunCommand` | 支持 | Allocation 生命周期与有界命令 | | `Observe` | 明确决定 | 检查所有权、只读地观测一个 allocation | | `Initial`、`NewCompute`、`GetCompute`、`Suspend`、`Resume`、`ResumeCompute`、`KillCompute`、`DeleteSnapshot`、`RunCommandCompute` | 所有 checkpoint 方法决定一致；仅 `nodes` 注册可以支持 | 精确计算实例、捕获与恢复、保留源恢复和清理 | +| `RestorePreparation`、`SavePreparation` | 两者决定一致 | 复用已完成的 Environment 准备结果 | `Initial` 和 `NewCompute` 构造 compute reference，不分配资源；`ResumeCompute` 在暂停中止后仅解冻同一驻留实例；`RunCommandCompute` 在一个精确 compute incarnation 中运行有界命令。Core 使用 `RunCommandCompute` 在恢复后唤醒 parked daemon（[`runtime_compute_wake.go`](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/services/core/internal/execution/runtime_compute_wake.go)）。 +`SavePreparation(reference, key)` 在准备完整完成后保存该分配初始 compute 的 `/environment` 文件系统，`RestorePreparation(reference, key)` 把已保存的内容放回，没有有效条目时返回 `false`。`key` 是不透明的：Core 由 Project 和所有进入准备的内容推导它，从不使用 Environment、Session、Harness 或 Agent，提供商也从不解释它。存储、按 Runtime 镜像划分范围、过期和淘汰由提供商负责。条目在保存后 24 小时内有效，恢复不得使用已过期的条目。Core 在 Runtime 连接后调用 `RestorePreparation`，命中时只发送 `finalize` 操作即完成初始化。未命中时完整准备并调用 `SavePreparation`；保存错误只记录日志，不会使 Session 失败。两者都未声明的提供商始终完整准备。microsandbox 适配器在 Runtime home 下的私有目录中为每个 key 保存一个归档。 + 每个声明项为不带 reason 的 `state: supported`，或带 authored reason code 的 `state: unsupported`。缺失、零值、未知或不安全项都会验证失败。给 `SandboxProvider` 添加方法时，必须在每个 adapter 中明确决定并实现；不提供 base type，也不生成笼统的不支持实现。 不支持的方法在任何原生 I/O 前返回 `providercontract.UnsupportedError`。错误指明精确操作和安全 code，不包含原生消息、资源身份、endpoint 或凭据。空结果、nil error、`Unavailable` 或未知 mutation 结果都不能代替 unsupported，五项必需方法不能返回 unsupported。 diff --git a/internal/agentcapabilities/install.go b/internal/agentcapabilities/install.go index 785de407f..ac9585e44 100644 --- a/internal/agentcapabilities/install.go +++ b/internal/agentcapabilities/install.go @@ -104,6 +104,21 @@ func Finalize(installed *os.Root, input Input, identity Identity, resolve Direct return ErrInvalid } } + return writeManifest(installed, manifest) +} + +// Rebind gives an installation restored from an earlier preparation the identity +// of this Session. The earlier preparation must have had the identical selection. +func Rebind(installed *os.Root, manifest *Manifest, input Input, identity Identity) error { + digest, err := selectionHash(input) + if err != nil || validateIdentity(identity) != nil || manifest.Version != 1 || manifest.SelectionSHA256 != digest { + return ErrInvalid + } + manifest.Identity = identity + return writeManifest(installed, *manifest) +} + +func writeManifest(installed *os.Root, manifest Manifest) error { body, err := json.Marshal(manifest) if err != nil || len(body) > 256<<10 || writeFile(installed, ManifestName+".tmp", body, 0400) != nil || installed.Rename(ManifestName+".tmp", ManifestName) != nil { return ErrInvalid diff --git a/services/core/cmd/server/managed_generation_operations.go b/services/core/cmd/server/managed_generation_operations.go index 2f7c99b65..bacf142d9 100644 --- a/services/core/cmd/server/managed_generation_operations.go +++ b/services/core/cmd/server/managed_generation_operations.go @@ -105,3 +105,25 @@ func (p *generationRouter) ResumeCompute(ctx context.Context, r sandbox.Referenc defer done() return v.ResumeCompute(ctx, r, c) } +func (p *generationRouter) RestorePreparation(ctx context.Context, r sandbox.Reference, key string) (bool, error) { + if err := providercontract.Require(p, "RestorePreparation"); err != nil { + return false, err + } + v, done, err := p.route(ctx, r) + if err != nil { + return false, err + } + defer done() + return v.RestorePreparation(ctx, r, key) +} +func (p *generationRouter) SavePreparation(ctx context.Context, r sandbox.Reference, key string) error { + if err := providercontract.Require(p, "SavePreparation"); err != nil { + return err + } + v, done, err := p.route(ctx, r) + if err != nil { + return err + } + defer done() + return v.SavePreparation(ctx, r, key) +} diff --git a/services/core/internal/execution/provider_operations_fixture_test.go b/services/core/internal/execution/provider_operations_fixture_test.go index 98e951c04..58b17c3a8 100644 --- a/services/core/internal/execution/provider_operations_fixture_test.go +++ b/services/core/internal/execution/provider_operations_fixture_test.go @@ -10,21 +10,23 @@ import ( func (*lifecycleOnlySandbox) ProviderOperations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, } } func (*lifecycleOnlySandbox) Initial(context.Context, sandbox.Reference) (sandbox.Compute, error) { @@ -57,3 +59,9 @@ func (*lifecycleOnlySandbox) ResumeCompute(context.Context, sandbox.Reference, s func (*lifecycleOnlySandbox) Observe(context.Context, runtimeobs.Target) (runtimeobs.Sample, error) { return runtimeobs.Sample{}, &providercontract.UnsupportedError{Operation: "Observe", Reason: "fixture_operation_not_supported"} } +func (*lifecycleOnlySandbox) RestorePreparation(context.Context, sandbox.Reference, string) (bool, error) { + return false, &providercontract.UnsupportedError{Operation: "RestorePreparation", Reason: "fixture_operation_not_supported"} +} +func (*lifecycleOnlySandbox) SavePreparation(context.Context, sandbox.Reference, string) error { + return &providercontract.UnsupportedError{Operation: "SavePreparation", Reason: "fixture_operation_not_supported"} +} diff --git a/services/core/internal/execution/runtime_initialization.go b/services/core/internal/execution/runtime_initialization.go index 1c958d8ea..14cf1e203 100644 --- a/services/core/internal/execution/runtime_initialization.go +++ b/services/core/internal/execution/runtime_initialization.go @@ -140,8 +140,8 @@ func (w *Worker) prepareEnvironment(ctx context.Context, owner sessions.Environm } identity := agentcapabilities.Identity{EnvironmentID: owner.EnvironmentID, SessionID: owner.SessionID} operations := setupOperations(setup) - for index := 0; index < len(cfg.Files)+len(operations); index++ { - err = w.lease.CheckOwnership(ctx) + step := func(index int) error { + err := w.lease.CheckOwnership(ctx) if err == nil { var currentPeer = peer currentPeer, err = w.dispatcher.authorizedPeer(ctx, owner.DeviceID) @@ -162,14 +162,30 @@ func (w *Worker) prepareEnvironment(ctx context.Context, owner sessions.Environm candidate = command.provisioningFailure(0) err = runRuntimeSetup(ctx, peer, identity, command) } - if err != nil { - var confirmed *runtimeStepFailure - if errors.As(err, &confirmed) { - candidate.ExitCode = confirmed.exitCode - *failure = candidate - } - return err + var confirmed *runtimeStepFailure + if errors.As(err, &confirmed) { + candidate.ExitCode = confirmed.exitCode + *failure = candidate } + return err } - return nil + total := len(cfg.Files) + len(operations) + cache, err := w.preparationCache(ctx, owner, len(cfg.Files), operations) + if err != nil { + return err + } + return cache.run(ctx, func() error { + for index := 0; index < total; index++ { + if err := step(index); err != nil { + return err + } + } + return nil + }, func() error { + // Only finalize binds the restored capability installation to this Session. + if len(operations) > 0 && operations[len(operations)-1].Request.Action == "finalize" { + return step(total - 1) + } + return nil + }) } diff --git a/services/core/internal/execution/runtime_manager.go b/services/core/internal/execution/runtime_manager.go index c0609ebc8..339bea38b 100644 --- a/services/core/internal/execution/runtime_manager.go +++ b/services/core/internal/execution/runtime_manager.go @@ -14,6 +14,17 @@ import ( "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/workspaces" ) +// provider returns the configured Provider and its installation, or nil during +// a deployment switch. +func (m *runtimeManager) provider() (sandbox.SandboxProvider, string) { + m.mu.Lock() + defer m.mu.Unlock() + if m.switching { + return nil, "" + } + return m.config.Provider, m.config.InstallationID +} + // runtimeManager owns node membership, not provider operations. Each registered // node has one serial lifecycle; slow provider work cannot occupy another node. var errRuntimeTransition = fmt.Errorf("%w: sandbox configuration is changing", ErrExecutionUnavailable) diff --git a/services/core/internal/execution/runtime_preparation_cache.go b/services/core/internal/execution/runtime_preparation_cache.go new file mode 100644 index 000000000..182c01ad0 --- /dev/null +++ b/services/core/internal/execution/runtime_preparation_cache.go @@ -0,0 +1,87 @@ +package execution + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/internal/obs/log" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/providercontract" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sessions" +) + +// preparationCache reuses the result of an earlier preparation with the same key +// through a Provider that declares the operations. The Provider owns storage and +// lifetime; Core only computes the key. +type preparationCache struct { + provider sandbox.SandboxProvider + reference sandbox.Reference + key string +} + +// preparationCache returns nil when this Environment cannot use the cache: no +// preparation work, no declared operations or an allocation of another installation. +func (w *Worker) preparationCache(ctx context.Context, owner sessions.EnvironmentInitialization, fileCount int, operations []runtimeSetupOperation) (*preparationCache, error) { + provider, installation := w.runtimes.provider() + if fileCount+len(operations) == 0 || provider == nil || providercontract.Require(provider, "RestorePreparation") != nil { + return nil, nil + } + allocation, err := w.dispatcher.DeploymentReader.EnvironmentAllocation(ctx, deployment.AllocationKey{TenantID: owner.TenantID, EnvironmentID: owner.EnvironmentID}) + if errors.Is(err, deployment.ErrNotFound) || err == nil && (allocation.ProviderKey != installation || allocation.State != "running") { + return nil, nil + } + if err != nil { + return nil, err + } + hash := sha256.New() + encoder := json.NewEncoder(hash) + _ = encoder.Encode(owner.TenantID) + for index := range fileCount { + metadata, body, err := w.dispatcher.SessionsReader.ReadInitialEnvironmentFile(ctx, owner.TenantID, owner.SessionID, index) + if err != nil { + return nil, err + } + digest := sha256.Sum256(body) + _ = encoder.Encode([]string{metadata.Path, hex.EncodeToString(digest[:])}) + } + for _, operation := range operations { + digest := sha256.Sum256(operation.Data) + _ = encoder.Encode(operation.Request) + _ = encoder.Encode(hex.EncodeToString(digest[:])) + } + return &preparationCache{provider, runtimeReference(allocation), hex.EncodeToString(hash.Sum(nil))}, nil +} + +// run restores a saved preparation and sends only finalize, which binds the +// restored installation to this Session. Otherwise it prepares and saves the +// result. A failed save never fails the Session. +func (c *preparationCache) run(ctx context.Context, prepare, finalize func() error) error { + if c == nil { + return prepare() + } + started := time.Now() + restored, err := c.provider.RestorePreparation(ctx, c.reference, c.key) + c.log(ctx, "restore_preparation", restored, started, err) + if err != nil { + return err + } + if restored { + return finalize() + } + if err := prepare(); err != nil { + return err + } + started = time.Now() + c.log(ctx, "save_preparation", false, started, c.provider.SavePreparation(ctx, c.reference, c.key)) + return nil +} + +func (c *preparationCache) log(ctx context.Context, step string, hit bool, started time.Time, err error) { + log.Ctx(ctx).Info("Environment preparation cache", "allocation_id", c.reference.AllocationID, "step", step, "hit", hit, + "duration_ms", time.Since(started).Milliseconds(), "succeeded", err == nil) +} diff --git a/services/core/internal/execution/runtime_preparation_cache_test.go b/services/core/internal/execution/runtime_preparation_cache_test.go new file mode 100644 index 000000000..fb2ea9d41 --- /dev/null +++ b/services/core/internal/execution/runtime_preparation_cache_test.go @@ -0,0 +1,48 @@ +package execution + +import ( + "context" + "errors" + "testing" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" +) + +type preparationProvider struct { + sandbox.SandboxProvider + hit bool + saveErr error + saved []string +} + +func (p *preparationProvider) RestorePreparation(context.Context, sandbox.Reference, string) (bool, error) { + return p.hit, nil +} +func (p *preparationProvider) SavePreparation(_ context.Context, _ sandbox.Reference, key string) error { + p.saved = append(p.saved, key) + return p.saveErr +} + +func TestPreparationCacheHitFinalizesMissPreparesAndSaves(t *testing.T) { + for _, test := range []struct { + name string + hit bool + saveErr error + prepared, finalized int + saved int + }{ + {"hit", true, nil, 0, 1, 0}, + {"miss", false, nil, 1, 0, 1}, + {"failed save", false, errors.New("disk full"), 1, 0, 1}, + } { + t.Run(test.name, func(t *testing.T) { + provider := &preparationProvider{hit: test.hit, saveErr: test.saveErr} + var prepared, finalized int + cache := &preparationCache{provider: provider, key: "key"} + err := cache.run(context.Background(), func() error { prepared++; return nil }, func() error { finalized++; return nil }) + if err != nil || prepared != test.prepared || finalized != test.finalized || len(provider.saved) != test.saved { + t.Fatal(err, prepared, finalized, provider.saved) + } + }) + } +} diff --git a/services/core/internal/sandbox/docker/operations.go b/services/core/internal/sandbox/docker/operations.go index 2722da0dd..0e2ac9c0d 100644 --- a/services/core/internal/sandbox/docker/operations.go +++ b/services/core/internal/sandbox/docker/operations.go @@ -9,21 +9,23 @@ import ( // Operations is this adapter's complete authored resource contract. func Operations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "NewCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "GetCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "Suspend": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "Resume": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "KillCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "RunCommandCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "ResumeCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, - "Observe": {State: providercontract.Supported}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "NewCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "GetCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "Suspend": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "Resume": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "KillCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "RunCommandCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "ResumeCompute": {State: providercontract.Unsupported, Reason: "docker_does_not_support_checkpoints"}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "docker_does_not_support_preparation_cache"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "docker_does_not_support_preparation_cache"}, + "Observe": {State: providercontract.Supported}, } } func (*Provider) ProviderOperations() providercontract.Operations { return Operations() } @@ -54,3 +56,9 @@ func (p *Provider) RunCommandCompute(context.Context, sandbox.Reference, sandbox func (p *Provider) ResumeCompute(context.Context, sandbox.Reference, sandbox.Compute) (sandbox.ComputeState, error) { return sandbox.ComputeState{}, &providercontract.UnsupportedError{Operation: "ResumeCompute", Reason: Operations()["ResumeCompute"].Reason} } +func (p *Provider) RestorePreparation(context.Context, sandbox.Reference, string) (bool, error) { + return false, &providercontract.UnsupportedError{Operation: "RestorePreparation", Reason: Operations()["RestorePreparation"].Reason} +} +func (p *Provider) SavePreparation(context.Context, sandbox.Reference, string) error { + return &providercontract.UnsupportedError{Operation: "SavePreparation", Reason: Operations()["SavePreparation"].Reason} +} diff --git a/services/core/internal/sandbox/e2b/operations.go b/services/core/internal/sandbox/e2b/operations.go index a3da4688e..1a6feee22 100644 --- a/services/core/internal/sandbox/e2b/operations.go +++ b/services/core/internal/sandbox/e2b/operations.go @@ -9,21 +9,23 @@ import ( // Operations is this adapter's complete authored resource contract. func Operations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "NewCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "GetCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "Suspend": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "Resume": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "KillCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "RunCommandCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "ResumeCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, - "Observe": {State: providercontract.Supported}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "NewCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "GetCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "Suspend": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "Resume": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "KillCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "RunCommandCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "ResumeCompute": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_checkpoints"}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_preparation_cache"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "e2b_does_not_support_preparation_cache"}, + "Observe": {State: providercontract.Supported}, } } func (*Provider) ProviderOperations() providercontract.Operations { return Operations() } @@ -54,3 +56,9 @@ func (p *Provider) RunCommandCompute(context.Context, sandbox.Reference, sandbox func (p *Provider) ResumeCompute(context.Context, sandbox.Reference, sandbox.Compute) (sandbox.ComputeState, error) { return sandbox.ComputeState{}, &providercontract.UnsupportedError{Operation: "ResumeCompute", Reason: Operations()["ResumeCompute"].Reason} } +func (p *Provider) RestorePreparation(context.Context, sandbox.Reference, string) (bool, error) { + return false, &providercontract.UnsupportedError{Operation: "RestorePreparation", Reason: Operations()["RestorePreparation"].Reason} +} +func (p *Provider) SavePreparation(context.Context, sandbox.Reference, string) error { + return &providercontract.UnsupportedError{Operation: "SavePreparation", Reason: Operations()["SavePreparation"].Reason} +} diff --git a/services/core/internal/sandbox/microsandbox/identity.go b/services/core/internal/sandbox/microsandbox/identity.go index dc19b182f..e81ca3e5d 100644 --- a/services/core/internal/sandbox/microsandbox/identity.go +++ b/services/core/internal/sandbox/microsandbox/identity.go @@ -124,6 +124,10 @@ func ValidateRequest(q Request) error { return sandbox.ErrInvalid } return ValidateCompute(q.Config, q.Reference, q.Compute) + case "restore_preparation", "save_preparation": + if !validHash(q.Key) || q.Compute != (Compute{Name: Name(q.Config, q.Reference, 0)}) { + return sandbox.ErrInvalid + } case "command": if q.Command == nil || ValidateCommand(*q.Command) != nil { return sandbox.ErrInvalid diff --git a/services/core/internal/sandbox/microsandbox/operations.go b/services/core/internal/sandbox/microsandbox/operations.go index 13e81ff96..14f788411 100644 --- a/services/core/internal/sandbox/microsandbox/operations.go +++ b/services/core/internal/sandbox/microsandbox/operations.go @@ -5,21 +5,23 @@ import "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/providercontr // Operations is this adapter's complete authored resource contract. func Operations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Supported}, - "NewCompute": {State: providercontract.Supported}, - "GetCompute": {State: providercontract.Supported}, - "Suspend": {State: providercontract.Supported}, - "Resume": {State: providercontract.Supported}, - "KillCompute": {State: providercontract.Supported}, - "DeleteSnapshot": {State: providercontract.Supported}, - "RunCommandCompute": {State: providercontract.Supported}, - "ResumeCompute": {State: providercontract.Supported}, - "Observe": {State: providercontract.Supported}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Supported}, + "NewCompute": {State: providercontract.Supported}, + "GetCompute": {State: providercontract.Supported}, + "Suspend": {State: providercontract.Supported}, + "Resume": {State: providercontract.Supported}, + "KillCompute": {State: providercontract.Supported}, + "DeleteSnapshot": {State: providercontract.Supported}, + "RunCommandCompute": {State: providercontract.Supported}, + "ResumeCompute": {State: providercontract.Supported}, + "RestorePreparation": {State: providercontract.Supported}, + "SavePreparation": {State: providercontract.Supported}, + "Observe": {State: providercontract.Supported}, } } func (*Provider) ProviderOperations() providercontract.Operations { return Operations() } diff --git a/services/core/internal/sandbox/microsandbox/provider.go b/services/core/internal/sandbox/microsandbox/provider.go index abf5acec7..9d4d368fd 100644 --- a/services/core/internal/sandbox/microsandbox/provider.go +++ b/services/core/internal/sandbox/microsandbox/provider.go @@ -238,6 +238,24 @@ func (p *Provider) RunCommandCompute(ctx context.Context, r sandbox.Reference, c return *out.Command, nil } +// The preparation cache addresses the initial compute, like RunCommand. +func (p *Provider) RestorePreparation(ctx context.Context, r sandbox.Reference, key string) (bool, error) { + c, e := p.Initial(ctx, r) + if e != nil { + return false, e + } + out, e := p.call(ctx, Request{Operation: "restore_preparation", Reference: r, Compute: c, Key: key}) + return out.Restored && e == nil, e +} +func (p *Provider) SavePreparation(ctx context.Context, r sandbox.Reference, key string) error { + c, e := p.Initial(ctx, r) + if e != nil { + return e + } + _, e = p.call(ctx, Request{Operation: "save_preparation", Reference: r, Compute: c, Key: key}) + return e +} + // ResumeCompute thaws the exact resident source after an aborted suspension. // It never starts stopped compute or restores a checkpoint. func (p *Provider) ResumeCompute(ctx context.Context, r sandbox.Reference, c Compute) (State, error) { diff --git a/services/core/internal/sandbox/microsandbox/types.go b/services/core/internal/sandbox/microsandbox/types.go index 1947518f6..a9b53badb 100644 --- a/services/core/internal/sandbox/microsandbox/types.go +++ b/services/core/internal/sandbox/microsandbox/types.go @@ -70,7 +70,9 @@ type Request struct { Suspend *SuspendRequest Resume *ResumeRequest Snapshot *SnapshotIdentity - Deadline time.Time + // Key is the opaque preparation cache key. + Key string `json:",omitempty"` + Deadline time.Time } type Response struct { // CreateSettled binds State to a completed Create rejection or an initial_info @@ -80,6 +82,7 @@ type Response struct { State *State Command *sandbox.CommandResult Metrics *Metrics + Restored bool `json:",omitempty"` ErrorCode string } diff --git a/services/core/internal/sandbox/node/operations.go b/services/core/internal/sandbox/node/operations.go index 8f016664b..55e9ba504 100644 --- a/services/core/internal/sandbox/node/operations.go +++ b/services/core/internal/sandbox/node/operations.go @@ -2,21 +2,23 @@ package node // Transport operation names map to the authored Provider method declarations. var operationMethods = map[string]string{ - "create": "Create", - "info": "GetInfo", - "renew": "Renew", - "kill": "Kill", - "command": "RunCommand", - "initial": "Initial", - "new_compute": "NewCompute", - "compute": "GetCompute", - "suspend": "Suspend", - "resume": "Resume", - "kill_compute": "KillCompute", - "delete_snapshot": "DeleteSnapshot", - "command_compute": "RunCommandCompute", - "resume_compute": "ResumeCompute", - "observe": "Observe", + "create": "Create", + "info": "GetInfo", + "renew": "Renew", + "kill": "Kill", + "command": "RunCommand", + "initial": "Initial", + "new_compute": "NewCompute", + "compute": "GetCompute", + "suspend": "Suspend", + "resume": "Resume", + "kill_compute": "KillCompute", + "delete_snapshot": "DeleteSnapshot", + "command_compute": "RunCommandCompute", + "resume_compute": "ResumeCompute", + "restore_preparation": "RestorePreparation", + "save_preparation": "SavePreparation", + "observe": "Observe", } func operationMethod(wire string) string { return operationMethods[wire] } diff --git a/services/core/internal/sandbox/node/provider_operations_fixture_test.go b/services/core/internal/sandbox/node/provider_operations_fixture_test.go index 3f892e9bb..07bfec1e1 100644 --- a/services/core/internal/sandbox/node/provider_operations_fixture_test.go +++ b/services/core/internal/sandbox/node/provider_operations_fixture_test.go @@ -10,21 +10,23 @@ import ( func (*fakeProvider) ProviderOperations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, } } func (*fakeProvider) Initial(context.Context, sandbox.Reference) (sandbox.Compute, error) { @@ -59,20 +61,28 @@ func (*fakeProvider) Observe(context.Context, runtimeobs.Target) (runtimeobs.Sam } func (*observationProvider) ProviderOperations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Observe": {State: providercontract.Supported}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Observe": {State: providercontract.Supported}, } } +func (*fakeProvider) RestorePreparation(context.Context, sandbox.Reference, string) (bool, error) { + return false, &providercontract.UnsupportedError{Operation: "RestorePreparation", Reason: "fixture_operation_not_supported"} +} +func (*fakeProvider) SavePreparation(context.Context, sandbox.Reference, string) error { + return &providercontract.UnsupportedError{Operation: "SavePreparation", Reason: "fixture_operation_not_supported"} +} diff --git a/services/core/internal/sandbox/node/proxy.go b/services/core/internal/sandbox/node/proxy.go index 4949c3795..4a8b5b34c 100644 --- a/services/core/internal/sandbox/node/proxy.go +++ b/services/core/internal/sandbox/node/proxy.go @@ -154,6 +154,14 @@ func (p *provider) DeleteSnapshot(ctx context.Context, r sandbox.Reference, s sa func (p *provider) RunCommandCompute(ctx context.Context, r sandbox.Reference, c sandbox.Compute, v sandbox.Command) (sandbox.CommandResult, error) { return p.command(ctx, request{Operation: "command_compute", Reference: r, Compute: &c, Command: &v}) } +func (p *provider) RestorePreparation(ctx context.Context, r sandbox.Reference, key string) (bool, error) { + out, e := p.call(ctx, request{Operation: "restore_preparation", Reference: r, Key: key}) + return out.Restored && e == nil, e +} +func (p *provider) SavePreparation(ctx context.Context, r sandbox.Reference, key string) error { + _, e := p.call(ctx, request{Operation: "save_preparation", Reference: r, Key: key}) + return e +} func (p *provider) ResumeCompute(ctx context.Context, r sandbox.Reference, c sandbox.Compute) (sandbox.ComputeState, error) { return p.state(ctx, request{Operation: "resume_compute", Reference: r, Compute: &c}) } diff --git a/services/core/internal/sandbox/node/wire.go b/services/core/internal/sandbox/node/wire.go index 1032d20d2..087466bf4 100644 --- a/services/core/internal/sandbox/node/wire.go +++ b/services/core/internal/sandbox/node/wire.go @@ -98,6 +98,8 @@ type request struct { Resume *sandbox.ResumeRequest `json:"resume,omitempty"` Snapshot *sandbox.SnapshotIdentity `json:"snapshot,omitempty"` Observation *runtimeobs.Target `json:"observation,omitempty"` + // Key is the opaque preparation cache key of the preparation operations. + Key string `json:"key,omitempty"` } type response struct { @@ -110,6 +112,7 @@ type response struct { State *sandbox.ComputeState `json:"state,omitempty"` Command *sandbox.CommandResult `json:"command,omitempty"` Sample *runtimeobs.Sample `json:"sample,omitempty"` + Restored bool `json:"restored,omitempty"` } type frame struct { @@ -263,6 +266,10 @@ func (q request) validate() error { if count == 0 { return nil } + case "restore_preparation", "save_preparation": + if count == 0 && q.Key != "" && len(q.Key) <= 256 { + return nil + } case "new_compute": if count == 0 || count == 1 && q.Snapshot != nil { return nil @@ -332,6 +339,10 @@ func execute(ctx context.Context, p sandbox.SandboxProvider, q request) (respons case "command": command, err = p.RunCommand(ctx, q.Reference, *q.Command) out.Command = &command + case "restore_preparation": + out.Restored, err = p.RestorePreparation(ctx, q.Reference, q.Key) + case "save_preparation": + err = p.SavePreparation(ctx, q.Reference, q.Key) default: var state sandbox.ComputeState var compute sandbox.Compute @@ -380,6 +391,7 @@ func execute(ctx context.Context, p sandbox.SandboxProvider, q request) (respons } out.Command = nil out.Compute = nil + out.Restored = false } return out, err } diff --git a/services/core/internal/sandbox/sandbox_provider.go b/services/core/internal/sandbox/sandbox_provider.go index c53320c48..acc47ccc1 100644 --- a/services/core/internal/sandbox/sandbox_provider.go +++ b/services/core/internal/sandbox/sandbox_provider.go @@ -118,6 +118,13 @@ type SandboxProvider interface { // ResumeCompute thaws only the same resident instance after an aborted pause. ResumeCompute(context.Context, Reference, Compute) (ComputeState, error) + // The preparation cache keeps the Environment filesystem left by one finished + // preparation. Key is opaque and the adapter owns storage and lifetime. Both + // address the allocation's initial compute and share one declaration. + // RestorePreparation returns false when it holds no valid entry. + RestorePreparation(ctx context.Context, r Reference, key string) (bool, error) + SavePreparation(ctx context.Context, r Reference, key string) error + // Observe reads one owned Runtime instance after verifying its ownership. It // never renews, restarts or stops compute. Observe(context.Context, runtimeobs.Target) (runtimeobs.Sample, error) @@ -130,6 +137,9 @@ var requiredOperations = []string{"Create", "GetInfo", "Renew", "Kill", "RunComm // restore support cannot safely own a compute incarnation. var checkpointOperations = []string{"Initial", "NewCompute", "GetCompute", "Suspend", "Resume", "KillCompute", "DeleteSnapshot", "RunCommandCompute", "ResumeCompute"} +// preparationOperations are both supported or both unsupported. +var preparationOperations = []string{"RestorePreparation", "SavePreparation"} + // Compute identifies one incarnation of an allocation. Name is provider-derived. // ID is empty only until the original create or restore result is observed. type Compute struct { @@ -254,6 +264,9 @@ func ValidateOperations(operations providercontract.Operations) error { return fmt.Errorf("%w: incomplete checkpoint lifecycle", providercontract.ErrContract) } } + if operations[preparationOperations[0]].State != operations[preparationOperations[1]].State { + return fmt.Errorf("%w: incomplete preparation cache", providercontract.ErrContract) + } return nil } diff --git a/services/core/internal/sandbox/sandbox_provider_test.go b/services/core/internal/sandbox/sandbox_provider_test.go index 56c23b330..1cbde45ab 100644 --- a/services/core/internal/sandbox/sandbox_provider_test.go +++ b/services/core/internal/sandbox/sandbox_provider_test.go @@ -16,7 +16,7 @@ func TestOperationGroupsPartitionTheInterface(t *testing.T) { methods = append(methods, name) } } - groups := slices.Concat(requiredOperations, checkpointOperations, []string{"Observe"}) + groups := slices.Concat(requiredOperations, checkpointOperations, preparationOperations, []string{"Observe"}) slices.Sort(groups) if !slices.Equal(methods, groups) { t.Fatalf("SandboxProvider operations %v, groups %v", methods, groups) diff --git a/services/core/tests/integration/provider_operations_fixture_test.go b/services/core/tests/integration/provider_operations_fixture_test.go index f528d5ae0..c7d052b56 100644 --- a/services/core/tests/integration/provider_operations_fixture_test.go +++ b/services/core/tests/integration/provider_operations_fixture_test.go @@ -10,21 +10,23 @@ import ( func (*lifecycleProvider) ProviderOperations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, - "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "NewCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "GetCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Suspend": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Resume": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "KillCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "DeleteSnapshot": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RunCommandCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "ResumeCompute": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, } } func (*lifecycleProvider) Initial(context.Context, sandbox.Reference) (sandbox.Compute, error) { @@ -59,20 +61,28 @@ func (*lifecycleProvider) Observe(context.Context, runtimeobs.Target) (runtimeob } func (*fakeCheckpointProvider) ProviderOperations() providercontract.Operations { return providercontract.Operations{ - "Create": {State: providercontract.Supported}, - "GetInfo": {State: providercontract.Supported}, - "Renew": {State: providercontract.Supported}, - "Kill": {State: providercontract.Supported}, - "RunCommand": {State: providercontract.Supported}, - "Initial": {State: providercontract.Supported}, - "NewCompute": {State: providercontract.Supported}, - "GetCompute": {State: providercontract.Supported}, - "Suspend": {State: providercontract.Supported}, - "Resume": {State: providercontract.Supported}, - "KillCompute": {State: providercontract.Supported}, - "DeleteSnapshot": {State: providercontract.Supported}, - "RunCommandCompute": {State: providercontract.Supported}, - "ResumeCompute": {State: providercontract.Supported}, - "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Create": {State: providercontract.Supported}, + "GetInfo": {State: providercontract.Supported}, + "Renew": {State: providercontract.Supported}, + "Kill": {State: providercontract.Supported}, + "RunCommand": {State: providercontract.Supported}, + "Initial": {State: providercontract.Supported}, + "NewCompute": {State: providercontract.Supported}, + "GetCompute": {State: providercontract.Supported}, + "Suspend": {State: providercontract.Supported}, + "Resume": {State: providercontract.Supported}, + "KillCompute": {State: providercontract.Supported}, + "DeleteSnapshot": {State: providercontract.Supported}, + "RunCommandCompute": {State: providercontract.Supported}, + "ResumeCompute": {State: providercontract.Supported}, + "RestorePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "SavePreparation": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, + "Observe": {State: providercontract.Unsupported, Reason: "fixture_operation_not_supported"}, } } +func (*lifecycleProvider) RestorePreparation(context.Context, sandbox.Reference, string) (bool, error) { + return false, &providercontract.UnsupportedError{Operation: "RestorePreparation", Reason: "fixture_operation_not_supported"} +} +func (*lifecycleProvider) SavePreparation(context.Context, sandbox.Reference, string) error { + return &providercontract.UnsupportedError{Operation: "SavePreparation", Reason: "fixture_operation_not_supported"} +} diff --git a/services/core/tools/microsandbox-provider/backend.go b/services/core/tools/microsandbox-provider/backend.go index cc27005c6..2bf23bfeb 100644 --- a/services/core/tools/microsandbox-provider/backend.go +++ b/services/core/tools/microsandbox-provider/backend.go @@ -45,6 +45,11 @@ func (b backend) run(ctx context.Context) (wire.Response, error) { defer cancel() result, e := runCommand(commandCtx, live, *b.q.Command, "1000:1000") return wire.Response{Command: &result}, e + case "restore_preparation": + restored, e := b.restorePreparation(ctx) + return wire.Response{Restored: restored}, e + case "save_preparation": + return wire.Response{}, b.savePreparation(ctx) case "suspend": s, e := b.suspend(ctx, *b.q.Suspend) return wire.Response{State: &s}, e diff --git a/services/core/tools/microsandbox-provider/preparation.go b/services/core/tools/microsandbox-provider/preparation.go new file mode 100644 index 000000000..135f45791 --- /dev/null +++ b/services/core/tools/microsandbox-provider/preparation.go @@ -0,0 +1,214 @@ +//go:build linux + +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "sort" + "strings" + "time" + + "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox" + "github.com/google/uuid" + sdk "github.com/superradcompany/microsandbox/sdk/go" +) + +// The preparation cache keeps one archive of /environment per Core-supplied key, +// image digest and workspace mode in a private directory under the Runtime home. +// The guest never sees it. An entry is valid for preparationTTL after its save. +const preparationTTL = 24 * time.Hour + +// The store holds at most this many bytes. A measured Environment archive is +// tens of MiB, so this keeps over a hundred of them and bounds the node's disk +// use; the TTL alone does not, because a busy node can save many keys in a day. +const maxPreparationBytes = 8 << 30 + +// The guest archive lives on the root disk, outside /environment, so it is +// never part of the archive it carries. +const guestArchive = "/tmp/oac-preparation.tar" + +func (b backend) preparationStore() string { + return filepath.Join(b.q.Config.RuntimeHome, "preparation") +} + +// preparationEntry names the archive for this key, scoped by the Runtime image +// digest and by whether /environment is an owned disk or an external directory. +func (b backend) preparationEntry() string { + _, digest, _ := strings.Cut(b.q.Config.Image, "@") + sum := sha256.Sum256([]byte(fmt.Sprintf("%s|%t|%s", digest, b.q.Config.ExternalWorkspace, b.q.Key))) + return filepath.Join(b.preparationStore(), hex.EncodeToString(sum[:])+".tar") +} + +func validPreparation(path string, now time.Time) bool { + st, err := os.Lstat(path) + return err == nil && st.Mode().IsRegular() && now.Sub(st.ModTime()) < preparationTTL +} + +// preparationSandbox connects to the running compute and returns the host +// directory behind /environment, which is empty when the disk is owned. +func (b backend) preparationSandbox(ctx context.Context) (*sdk.Sandbox, string, error) { + h, _, err := b.inspect(ctx, b.q.Compute) + if err != nil { + return nil, "", err + } + live, err := h.Connect(ctx) + if err != nil { + return nil, "", err + } + var config sdk.SandboxConfig + if b.q.Config.ExternalWorkspace && (json.Unmarshal([]byte(h.ConfigJSON()), &config) != nil || config.Labels[workspacePathLabel] == "") { + live.Detach(context.Background()) + return nil, "", sandbox.ErrInvalid + } + return live, config.Labels[workspacePathLabel], nil +} + +func guestRun(ctx context.Context, live *sdk.Sandbox, script string) error { + result, err := runCommand(ctx, live, sandbox.Command{Args: []string{"/bin/sh", "-c", script}, Directory: "/"}, "0:0") + if err == nil && result.ExitCode != 0 { + err = errors.New("preparation command failed") + } + return err +} + +// hostTar copies the SDK's ownership xattrs of an external directory verbatim, +// so the helper never interprets them. +func hostTar(ctx context.Context, stdin *os.File, args ...string) error { + cmd := exec.CommandContext(ctx, "tar", append([]string{"--xattrs", "--xattrs-include=*"}, args...)...) + if stdin != nil { + cmd.Stdin = stdin + } + return cmd.Run() +} + +func (b backend) restorePreparation(ctx context.Context) (bool, error) { + entry := b.preparationEntry() + if !validPreparation(entry, time.Now()) { + return false, nil + } + ctx, cancel := context.WithDeadline(ctx, b.q.Deadline) + defer cancel() + live, external, err := b.preparationSandbox(ctx) + if err != nil { + return false, err + } + defer live.Detach(context.Background()) + if b.q.Config.ExternalWorkspace { + // The running guest sees the extracted files at once. + f, err := os.Open(entry) + if err != nil { + return false, nil + } + defer f.Close() + err = hostTar(ctx, f, "-C", external, "-xf", "-") + return err == nil, err + } + defer b.removeGuestArchive(live) + // Nothing in /environment has changed until the extraction starts. + if live.FS().CopyFromHost(ctx, entry, guestArchive) != nil { + return false, nil + } + err = guestRun(ctx, live, "tar --xattrs --numeric-owner -C /environment -xf "+guestArchive+" && sync") + return err == nil, err +} + +func (b backend) removeGuestArchive(live *sdk.Sandbox) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = guestRun(ctx, live, "rm -f "+guestArchive) +} + +func (b backend) savePreparation(ctx context.Context) error { + ctx, cancel := context.WithDeadline(ctx, b.q.Deadline) + defer cancel() + live, external, err := b.preparationSandbox(ctx) + if err != nil { + return err + } + defer live.Detach(context.Background()) + store := b.preparationStore() + if err = privateDirectory(store); err != nil { + return err + } + // The temporary name does not exist yet, so a copy from the guest creates it. + temporary := filepath.Join(store, ".tmp-"+uuid.NewString()) + defer os.Remove(temporary) + if b.q.Config.ExternalWorkspace { + if err = guestRun(ctx, live, "sync"); err == nil { + err = hostTar(ctx, nil, "-C", external, "-cf", temporary, ".") + } + } else { + defer b.removeGuestArchive(live) + if err = guestRun(ctx, live, "sync && tar --xattrs --numeric-owner -C /environment -cf "+guestArchive+" ."); err == nil { + err = live.FS().CopyToHost(ctx, guestArchive, temporary) + } + } + if err != nil { + return err + } + f, err := os.OpenFile(temporary, os.O_RDWR, 0) + if err != nil { + return err + } + err = f.Sync() + if closeErr := f.Close(); err == nil { + err = closeErr + } + if err == nil { + err = os.Chmod(temporary, 0600) + } + if err == nil { + err = os.Rename(temporary, b.preparationEntry()) + } + if err != nil { + return err + } + _ = prunePreparations(store, time.Now()) + return syncDirectory(store) +} + +// prunePreparations deletes expired entries, then the oldest while the store is +// over its bound. +func prunePreparations(store string, now time.Time) error { + entries, err := os.ReadDir(store) + if err != nil { + return err + } + type kept struct { + path string + size int64 + modified time.Time + } + var live []kept + var total int64 + for _, entry := range entries { + info, err := entry.Info() + if err != nil { + continue + } + path := filepath.Join(store, entry.Name()) + if now.Sub(info.ModTime()) >= preparationTTL { + _ = os.Remove(path) + continue + } + live = append(live, kept{path, info.Size(), info.ModTime()}) + total += info.Size() + } + sort.Slice(live, func(i, j int) bool { return live[i].modified.Before(live[j].modified) }) + for _, entry := range live { + if total <= maxPreparationBytes { + break + } + _ = os.Remove(entry.path) + total -= entry.size + } + return nil +} diff --git a/services/core/tools/microsandbox-provider/preparation_test.go b/services/core/tools/microsandbox-provider/preparation_test.go new file mode 100644 index 000000000..096ccde3e --- /dev/null +++ b/services/core/tools/microsandbox-provider/preparation_test.go @@ -0,0 +1,35 @@ +//go:build linux + +package main + +import ( + "context" + "os" + "strings" + "testing" + "time" + + wire "github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox/microsandbox" +) + +func TestExpiredPreparationIsNotRestored(t *testing.T) { + b := backend{q: wire.Request{Config: wire.Config{RuntimeHome: t.TempDir(), Image: "image@sha256:" + strings.Repeat("a", 64)}, Key: strings.Repeat("b", 64), Deadline: time.Now().Add(time.Minute)}} + if err := privateDirectory(b.preparationStore()); err != nil { + t.Fatal(err) + } + entry := b.preparationEntry() + if err := os.WriteFile(entry, []byte("archive"), 0600); err != nil { + t.Fatal(err) + } + if !validPreparation(entry, time.Now()) { + t.Fatal("fresh entry rejected") + } + old := time.Now().Add(-preparationTTL - time.Minute) + if err := os.Chtimes(entry, old, old); err != nil { + t.Fatal(err) + } + // An expired entry returns before the helper touches any compute. + if restored, err := b.restorePreparation(context.Background()); restored || err != nil { + t.Fatal(restored, err) + } +} From b336d803eff2d6689819449fd7aa1d43cc72ec16 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 10:59:05 +0000 Subject: [PATCH 2/4] Keep Runtime state out of the preparation archive --- docs/sandbox-provider.md | 2 +- docs/zh/sandbox-provider.md | 4 ++-- .../tools/microsandbox-provider/preparation.go | 15 ++++++++++----- 3 files changed, 13 insertions(+), 8 deletions(-) diff --git a/docs/sandbox-provider.md b/docs/sandbox-provider.md index 5b6afc3ae..439b11a4c 100644 --- a/docs/sandbox-provider.md +++ b/docs/sandbox-provider.md @@ -67,7 +67,7 @@ Every provider returns a complete `ProviderOperations()` declaration with one en `Initial` and `NewCompute` construct compute references without allocating, `ResumeCompute` thaws only the same resident instance after an aborted pause, and `RunCommandCompute` runs a bounded command in one exact compute incarnation. Core uses `RunCommandCompute` to wake a parked daemon after a restore ([`runtime_compute_wake.go`](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/services/core/internal/execution/runtime_compute_wake.go)). -`SavePreparation(reference, key)` stores the `/environment` filesystem of the allocation's initial compute after a complete preparation, and `RestorePreparation(reference, key)` puts a stored one back and returns `false` when it holds no valid entry. The `key` is opaque: Core derives it from the Project and everything that reaches preparation, never from the Environment, Session, Harness or Agent, and the Provider never interprets it. The Provider owns storage, scoping to its Runtime image, expiry and eviction. An entry is valid for 24 hours after its save, and a restore must not use an expired one. Core calls `RestorePreparation` after the Runtime connects and, on a hit, sends only the `finalize` operation before completing initialization. On a miss it prepares in full and calls `SavePreparation`; a save error is logged and never fails the Session. A Provider that declares neither always prepares in full. The microsandbox adapter keeps one archive per key in a private directory under the Runtime home. +`SavePreparation(reference, key)` stores what preparation wrote under `/environment` (workspace, packages and initialization, never Runtime state) in the allocation's initial compute after a complete preparation, and `RestorePreparation(reference, key)` puts a stored one back and returns `false` when it holds no valid entry. The `key` is opaque: Core derives it from the Project and everything that reaches preparation, never from the Environment, Session, Harness or Agent, and the Provider never interprets it. The Provider owns storage, scoping to its Runtime image, expiry and eviction. An entry is valid for 24 hours after its save, and a restore must not use an expired one. Core calls `RestorePreparation` after the Runtime connects and, on a hit, sends only the `finalize` operation before completing initialization. On a miss it prepares in full and calls `SavePreparation`; a save error is logged and never fails the Session. A Provider that declares neither always prepares in full. The microsandbox adapter keeps one archive per key in a private directory under the Runtime home. Each declaration entry is `state: supported` with no reason, or `state: unsupported` with an authored reason code. Missing, zero, unknown or unsafe entries fail validation. Adding a method to `SandboxProvider` requires an explicit decision and implementation in every adapter; never supply a base type or generate blanket unsupported implementations. diff --git a/docs/zh/sandbox-provider.md b/docs/zh/sandbox-provider.md index c70680410..e75c09117 100644 --- a/docs/zh/sandbox-provider.md +++ b/docs/zh/sandbox-provider.md @@ -1,7 +1,7 @@ --- title: "沙箱提供商协议" source: docs/sandbox-provider.md -source_hash: 0b333cf78e2a40479398e57e9e3d7cac7799fb05a5fa976da7e82c6f6893d65d +source_hash: 95b02d2c1b46117e603ebadd5fa948c92624c3eb59d50cc402eb5a3907e69378 --- 沙箱提供商协议连接 Core 与为 Core 管理的（`openai_hosted`）Environment 提供计算资源的 adapter，并由该 adapter 运行启动 Runtime daemon 的有界引导流程。本页面向 adapter 作者：说明提供商必须实现什么，以及可以依赖 Core 做什么。Docker、microsandbox 和 E2B 实现了该协议。 @@ -69,7 +69,7 @@ Docker 等没有原生可续期租约的 backend 仍遵守 Core 的 hosted expir `Initial` 和 `NewCompute` 构造 compute reference，不分配资源；`ResumeCompute` 在暂停中止后仅解冻同一驻留实例；`RunCommandCompute` 在一个精确 compute incarnation 中运行有界命令。Core 使用 `RunCommandCompute` 在恢复后唤醒 parked daemon（[`runtime_compute_wake.go`](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/services/core/internal/execution/runtime_compute_wake.go)）。 -`SavePreparation(reference, key)` 在准备完整完成后保存该分配初始 compute 的 `/environment` 文件系统，`RestorePreparation(reference, key)` 把已保存的内容放回，没有有效条目时返回 `false`。`key` 是不透明的：Core 由 Project 和所有进入准备的内容推导它，从不使用 Environment、Session、Harness 或 Agent，提供商也从不解释它。存储、按 Runtime 镜像划分范围、过期和淘汰由提供商负责。条目在保存后 24 小时内有效，恢复不得使用已过期的条目。Core 在 Runtime 连接后调用 `RestorePreparation`，命中时只发送 `finalize` 操作即完成初始化。未命中时完整准备并调用 `SavePreparation`；保存错误只记录日志，不会使 Session 失败。两者都未声明的提供商始终完整准备。microsandbox 适配器在 Runtime home 下的私有目录中为每个 key 保存一个归档。 +`SavePreparation(reference, key)` 在准备完整完成后保存该分配初始 compute 中准备写入 `/environment` 的内容（workspace、packages 和 initialization，从不包括 Runtime 状态），`RestorePreparation(reference, key)` 把已保存的内容放回，没有有效条目时返回 `false`。`key` 是不透明的：Core 由 Project 和所有进入准备的内容推导它，从不使用 Environment、Session、Harness 或 Agent，提供商也从不解释它。存储、按 Runtime 镜像划分范围、过期和淘汰由提供商负责。条目在保存后 24 小时内有效，恢复不得使用已过期的条目。Core 在 Runtime 连接后调用 `RestorePreparation`，命中时只发送 `finalize` 操作即完成初始化。未命中时完整准备并调用 `SavePreparation`；保存错误只记录日志，不会使 Session 失败。两者都未声明的提供商始终完整准备。microsandbox 适配器在 Runtime home 下的私有目录中为每个 key 保存一个归档。 每个声明项为不带 reason 的 `state: supported`，或带 authored reason code 的 `state: unsupported`。缺失、零值、未知或不安全项都会验证失败。给 `SandboxProvider` 添加方法时，必须在每个 adapter 中明确决定并实现；不提供 base type，也不生成笼统的不支持实现。 diff --git a/services/core/tools/microsandbox-provider/preparation.go b/services/core/tools/microsandbox-provider/preparation.go index 135f45791..8cabe59dc 100644 --- a/services/core/tools/microsandbox-provider/preparation.go +++ b/services/core/tools/microsandbox-provider/preparation.go @@ -21,11 +21,16 @@ import ( sdk "github.com/superradcompany/microsandbox/sdk/go" ) -// The preparation cache keeps one archive of /environment per Core-supplied key, -// image digest and workspace mode in a private directory under the Runtime home. -// The guest never sees it. An entry is valid for preparationTTL after its save. +// The preparation cache keeps one archive of the prepared /environment +// directories per Core-supplied key, image digest and workspace mode in a private +// directory under the Runtime home. The guest never sees it. An entry is valid +// for preparationTTL after its save. const preparationTTL = 24 * time.Hour +// preparedDirectories are what preparation writes, as created by the bootstrap. +// runtime-state belongs to the running daemon and is never copied. +var preparedDirectories = []string{"initialization", "packages", "workspace"} + // The store holds at most this many bytes. A measured Environment archive is // tens of MiB, so this keeps over a hundred of them and bounds the node's disk // use; the TTL alone does not, because a busy node can save many keys in a day. @@ -143,11 +148,11 @@ func (b backend) savePreparation(ctx context.Context) error { defer os.Remove(temporary) if b.q.Config.ExternalWorkspace { if err = guestRun(ctx, live, "sync"); err == nil { - err = hostTar(ctx, nil, "-C", external, "-cf", temporary, ".") + err = hostTar(ctx, nil, append([]string{"-C", external, "-cf", temporary}, preparedDirectories...)...) } } else { defer b.removeGuestArchive(live) - if err = guestRun(ctx, live, "sync && tar --xattrs --numeric-owner -C /environment -cf "+guestArchive+" ."); err == nil { + if err = guestRun(ctx, live, "sync && tar --xattrs --numeric-owner -C /environment -cf "+guestArchive+" "+strings.Join(preparedDirectories, " ")); err == nil { err = live.FS().CopyToHost(ctx, guestArchive, temporary) } } From dc932b25e6b2e54787f4c4e19f696767fb93fbe4 Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 11:07:03 +0000 Subject: [PATCH 3/4] Drop a preparation entry that fails to extract and bound its save --- .../execution/runtime_preparation_cache.go | 9 ++++++++- .../tools/microsandbox-provider/preparation.go | 16 ++++++++++------ 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/services/core/internal/execution/runtime_preparation_cache.go b/services/core/internal/execution/runtime_preparation_cache.go index 182c01ad0..fb388379a 100644 --- a/services/core/internal/execution/runtime_preparation_cache.go +++ b/services/core/internal/execution/runtime_preparation_cache.go @@ -77,7 +77,14 @@ func (c *preparationCache) run(ctx context.Context, prepare, finalize func() err return err } started = time.Now() - c.log(ctx, "save_preparation", false, started, c.provider.SavePreparation(ctx, c.reference, c.key)) + save := ctx + if deadline, bounded := ctx.Deadline(); bounded { + // Half the remaining budget, so a slow save never takes the time completion needs. + var cancel context.CancelFunc + save, cancel = context.WithTimeout(ctx, time.Until(deadline)/2) + defer cancel() + } + c.log(ctx, "save_preparation", false, started, c.provider.SavePreparation(save, c.reference, c.key)) return nil } diff --git a/services/core/tools/microsandbox-provider/preparation.go b/services/core/tools/microsandbox-provider/preparation.go index 8cabe59dc..2582b3075 100644 --- a/services/core/tools/microsandbox-provider/preparation.go +++ b/services/core/tools/microsandbox-provider/preparation.go @@ -114,14 +114,18 @@ func (b backend) restorePreparation(ctx context.Context) (bool, error) { } defer f.Close() err = hostTar(ctx, f, "-C", external, "-xf", "-") - return err == nil, err + } else { + defer b.removeGuestArchive(live) + // Nothing in /environment has changed until the extraction starts. + if live.FS().CopyFromHost(ctx, entry, guestArchive) != nil { + return false, nil + } + err = guestRun(ctx, live, "tar --xattrs --numeric-owner -C /environment -xf "+guestArchive+" && sync") } - defer b.removeGuestArchive(live) - // Nothing in /environment has changed until the extraction starts. - if live.FS().CopyFromHost(ctx, entry, guestArchive) != nil { - return false, nil + if err != nil { + // A failed extraction fails this Session; later Sessions prepare in full. + _ = os.Remove(entry) } - err = guestRun(ctx, live, "tar --xattrs --numeric-owner -C /environment -xf "+guestArchive+" && sync") return err == nil, err } From e01726b2601dd72019f1160b348d571102923f7d Mon Sep 17 00:00:00 2001 From: SaladDay <1203511142@qq.com> Date: Wed, 21 Oct 2026 11:41:29 +0000 Subject: [PATCH 4/4] Accept preparation fields in strict node frames --- .../internal/sandbox/node/generation_json.go | 4 ++-- .../sandbox/node/generation_json_test.go | 18 ++++++++++++++++++ 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/services/core/internal/sandbox/node/generation_json.go b/services/core/internal/sandbox/node/generation_json.go index a9418fd3a..b9b9873b9 100644 --- a/services/core/internal/sandbox/node/generation_json.go +++ b/services/core/internal/sandbox/node/generation_json.go @@ -135,12 +135,12 @@ func validateGenerationJSON(raw []byte, kind string) error { } } if value := values["request"]; value != nil { - if _, err := generationObject(value, "deployment_generation id sequence connection_id owner_epoch operation timeout_ms reference", "bootstrap compute generation command suspend resume snapshot observation", ""); err != nil { + if _, err := generationObject(value, "deployment_generation id sequence connection_id owner_epoch operation timeout_ms reference", "bootstrap compute generation command suspend resume snapshot observation key", ""); err != nil { return err } } if value := values["response"]; value != nil { - if _, err := generationObject(value, "id connection_id", "error_code unsupported info compute state command sample", ""); err != nil { + if _, err := generationObject(value, "id connection_id", "error_code unsupported info compute state command sample restored", ""); err != nil { return err } } diff --git a/services/core/internal/sandbox/node/generation_json_test.go b/services/core/internal/sandbox/node/generation_json_test.go index d8fcf99f5..77dfda1fc 100644 --- a/services/core/internal/sandbox/node/generation_json_test.go +++ b/services/core/internal/sandbox/node/generation_json_test.go @@ -123,3 +123,21 @@ func TestCheckpointGenerationHealthJSONRoundTrip(t *testing.T) { }) } } + +func TestPreparationFramesSurviveStrictFrame(t *testing.T) { + q := request{DeploymentGeneration: 1, ID: uuid.NewString(), Sequence: 1, ConnectionID: uuid.NewString(), OwnerEpoch: 1, Operation: "restore_preparation", TimeoutMillis: 60000, Reference: reference(), Key: strings.Repeat("a", 64)} + out := response{ID: q.ID, ConnectionID: q.ConnectionID, Restored: true} + for _, f := range []frame{{Version: ProtocolVersion, Type: "request", Request: &q}, {Version: ProtocolVersion, Type: "response", Response: &out}} { + raw, err := json.Marshal(f) + if err != nil { + t.Fatal(err) + } + decoded, err := decodeFrame(raw) + if err != nil { + t.Fatal("preparation frame rejected", f.Type, err) + } + if (decoded.Request != nil && decoded.Request.Key != q.Key) || (decoded.Response != nil && !decoded.Response.Restored) { + t.Fatal("preparation field lost", f.Type) + } + } +}