Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion docs/sandbox-provider.md
Original file line number Diff line number Diff line change
Expand Up @@ -242,7 +242,7 @@ sequenceDiagram

### Suspension

Core suspends an allocation after initialization completes, when no root or Subagent Turn is queued, in progress or waiting, no input or file operation is pending, and real activity has been idle for the policy's idle timeout. A Session need not have run a Turn. Capacity pressure can suspend an already idle node allocation earlier, after the inactivity grace in [Timeouts and limits](#timeouts-and-limits), when compatible waiting demand exists. It never interrupts an active Turn or releases capacity before the ordinary quiesce, capture and source-stop proofs complete. If suspension cannot make capacity usable, the waiting request keeps its original deadline. When retained slots are full, pressure can [release an idle checkpoint](#cold-replacement-after-checkpoint-retention), and it suspends an allocation only if that allocation's checkpoint could then be released.
Core suspends an allocation after initialization completes, when no root or Subagent Turn is queued, in progress or waiting, no input or file operation is pending, and real activity has been idle for the policy's idle timeout. A Session need not have run a Turn. Capacity pressure can suspend an already idle node allocation earlier, after the inactivity grace in [Timeouts and limits](#timeouts-and-limits), when compatible waiting demand exists. A node with no free active slot but a free retained slot also keeps one active slot free: after the free-slot inactivity in [Timeouts and limits](#timeouts-and-limits), Core can suspend an idle allocation without waiting demand, so the next Session does not wait for a suspension. It never interrupts an active Turn or releases capacity before the ordinary quiesce, capture and source-stop proofs complete. If suspension cannot make capacity usable, the waiting request keeps its original deadline. When retained slots are full, pressure can [release an idle checkpoint](#cold-replacement-after-checkpoint-retention), and it suspends an allocation only if that allocation's checkpoint could then be released.

A Provider relies on these rules:

Expand Down Expand Up @@ -285,6 +285,7 @@ Every other observation keeps ownership and its diagnostic: a `paused`, `running
| Bounded context | Every Provider call | Ends the caller's wait and proves no rollback, stop, cleanup or absence. The adapter never detaches untracked mutations or replays a timed-out command |
| Idle timeout and snapshot retention | Checkpoint-capable allocations, set by the [suspension policy](../contracts/agents-api/sandbox-deployment.md#suspension-policy) | At the idle timeout Core begins [suspension](#suspension). When retention ends Core cleans up the snapshot and, for a qualified Session, replaces the compute through [cold replacement](#cold-replacement-after-checkpoint-retention) |
| 15 s | Inactivity of an idle node allocation before capacity-pressure suspension | Core can suspend it ahead of the idle timeout when compatible waiting demand exists |
| 2 min | Inactivity of an idle allocation on a node with no free active slot and a free retained slot | Core can suspend it ahead of the idle timeout without waiting demand, so the next Session finds a free slot |
| 50 MiB plus 32 bytes (`MaxCommandInputBytes`) | `Command.Stdin` | The command is invalid |

The online window and the other node timers are in the [sandbox node protocol](../contracts/agents-api/node-generation-protocol.md#timeouts-and-limits).
Expand Down
5 changes: 3 additions & 2 deletions docs/zh/sandbox-provider.md
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
---
title: "沙箱提供商协议"
source: docs/sandbox-provider.md
source_hash: 95b02d2c1b46117e603ebadd5fa948c92624c3eb59d50cc402eb5a3907e69378
source_hash: 6a767cd4c884d93698405a7ebbdd78489e204ed0bdacf0c36e5b29bd7346d1dc
---

沙箱提供商协议连接 Core 与为 Core 管理的(`openai_hosted`)Environment 提供计算资源的 adapter,并由该 adapter 运行启动 Runtime daemon 的有界引导流程。本页面向 adapter 作者:说明提供商必须实现什么,以及可以依赖 Core 做什么。Docker、microsandbox 和 E2B 实现了该协议。
Expand Down Expand Up @@ -243,7 +243,7 @@ sequenceDiagram

### 暂停 {#suspension}

Core 在初始化完成、没有 root 或 Subagent Turn 处于排队、进行中或等待状态、没有待处理的 input 或文件操作,并且真实活动已空闲达到策略的空闲超时之后暂停 allocation。Session 不必运行过 Turn。存在兼容的等待需求时,容量压力可以在[超时与限制](#timeouts-and-limits)所列的无活动宽限期之后,提前暂停已经空闲的 node allocation。它从不打断活动 Turn,也不会在常规 quiesce、capture 和 source-stop 证明完成之前释放容量。如果暂停无法让容量可用,等待中的请求保留原有期限。retained slot 已满时,容量压力可以[释放空闲检查点](#cold-replacement-after-checkpoint-retention),并且只暂停其检查点随后可被释放的 allocation。
Core 在初始化完成、没有 root 或 Subagent Turn 处于排队、进行中或等待状态、没有待处理的 input 或文件操作,并且真实活动已空闲达到策略的空闲超时之后暂停 allocation。Session 不必运行过 Turn。存在兼容的等待需求时,容量压力可以在[超时与限制](#timeouts-and-limits)所列的无活动宽限期之后,提前暂停已经空闲的 node allocation。没有空闲 active slot、但有空闲 retained slot 的 node 还会保留一个空闲 active slot:达到[超时与限制](#timeouts-and-limits)所列的空闲 slot 无活动时间后,即使没有等待需求,Core 也可以暂停空闲的 allocation,让下一个 Session 不必等待暂停。它从不打断活动 Turn,也不会在常规 quiesce、capture 和 source-stop 证明完成之前释放容量。如果暂停无法让容量可用,等待中的请求保留原有期限。retained slot 已满时,容量压力可以[释放空闲检查点](#cold-replacement-after-checkpoint-retention),并且只暂停其检查点随后可被释放的 allocation。

提供商依赖以下规则:

Expand Down Expand Up @@ -287,6 +287,7 @@ Core 在释放 allocation 和 placement 归属前,确认原始创建已结算
| 有界 context | 每个提供商调用 | 结束调用方的等待,不证明已回滚、停止、清理或不存在。adapter 从不分离未跟踪的 mutation,也不重放已超时的命令 |
| 空闲超时与快照保留期 | 支持检查点的 allocation,由[暂停策略](../../contracts/agents-api/zh/sandbox-deployment.md#suspension-policy)设定 | 空闲超时到达时 Core 开始[暂停](#suspension)。保留期结束时 Core 清理快照,并对符合条件的 Session 通过[冷替换](#cold-replacement-after-checkpoint-retention)替换计算资源 |
| 15 s | 空闲 node allocation 在容量压力暂停之前的无活动时间 | 存在兼容的等待需求时,Core 可以早于空闲超时暂停它 |
| 2 min | 所在 node 没有空闲 active slot 但有空闲 retained slot 时,空闲 allocation 的无活动时间 | Core 可以早于空闲超时、在没有等待需求时暂停它,让下一个 Session 找到空闲 slot |
| 50 MiB 加 32 字节(`MaxCommandInputBytes`) | `Command.Stdin` | 命令无效 |

在线窗口和其他节点计时器见[沙箱节点协议](../../contracts/agents-api/zh/node-generation-protocol.md#timeouts-and-limits)。
Expand Down
9 changes: 8 additions & 1 deletion services/core/internal/deployment/allocations.go
Original file line number Diff line number Diff line change
Expand Up @@ -287,7 +287,14 @@ func (e *ExecutionOperations) SetCompute(ctx context.Context, owner Allocation,
if !activity.ReadyToSuspend(15*time.Second) || current.NodeID == "" {
return Allocation{}, ErrNotIdle
}
allowed, err := e.canSuspendForDemand(tx, current)
pressure, err := tx.LoadSuspensionDemand(current)
if err != nil {
return Allocation{}, err
}
allowed, err := e.canSuspendForDemand(tx, pressure, current)
if err == nil && !allowed && activity.ReadyToSuspend(freeSlotIdle) {
allowed = canKeepFreeSlot(pressure, current)
}
if err != nil {
return Allocation{}, err
}
Expand Down
39 changes: 28 additions & 11 deletions services/core/internal/deployment/suspension_pressure.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,16 @@ import (
"context"
"errors"
"slices"
"time"

"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment/placement"
"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox"
)

// freeSlotIdle is how long an allocation on a full node stays idle before
// Core may suspend it without waiting demand.
const freeSlotIdle = 2 * time.Minute

// SuspensionDemand is a transaction-bound capacity observation. It is not a
// reservation: only a confirmed suspension frees an active slot.
type SuspensionDemand struct {
Expand All @@ -25,22 +30,34 @@ type CheckpointRestoreDemand struct {
Nodes []placement.CheckpointNode
}

func (e *ExecutionOperations) canSuspendForDemand(tx AllocationTx, owner Allocation) (bool, error) {
pressure, err := tx.LoadSuspensionDemand(owner)
if err != nil {
return false, err
}
// fullNode returns the owner's node when it is online and has no free active
// slot, in a nodes deployment that is not resetting and belongs to the owner's
// installation.
func fullNode(pressure SuspensionDemand, owner Allocation) *placement.Node {
if pressure.Deployment.Resetting || pressure.Deployment.Mode != string(sandbox.DeploymentNodes) || pressure.Deployment.InstallationID != owner.ProviderKey {
return false, nil
return nil
}
var source *placement.Node
for i := range pressure.Nodes {
if pressure.Nodes[i].ID == owner.NodeID {
source = &pressure.Nodes[i]
break
if n := &pressure.Nodes[i]; n.ID == owner.NodeID && n.Online && n.Active >= int64(n.MaxActive) {
return n
}
}
if source == nil || !source.Online || source.Active < int64(source.MaxActive) {
return nil
}

// canKeepFreeSlot allows suspending an allocation that stayed idle for
// freeSlotIdle on a full node with no reclamation in flight. The node must
// also have a free retained slot, or the freed active slot admits no Session.
func canKeepFreeSlot(pressure SuspensionDemand, owner Allocation) bool {
n := fullNode(pressure, owner)
return n != nil && n.Retained < int64(n.MaxRetained) && !slices.Contains(pressure.InFlightNodes, owner.NodeID)
}

// canSuspendForDemand reports whether compatible waiting demand would place on
// the owner's full node once the owner is suspended.
func (e *ExecutionOperations) canSuspendForDemand(tx AllocationTx, pressure SuspensionDemand, owner Allocation) (bool, error) {
source := fullNode(pressure, owner)
if source == nil {
return false, nil
}
for _, demand := range pressure.CheckpointRestores {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,11 @@ import (
"github.com/google/uuid"
)

func runningPressureOwner(t *testing.T, f fixture, changes *deployment.ExecutionOperations, installation, node string, generation uint64) deployment.Allocation {
func runningPressureOwner(t *testing.T, f fixture, changes *deployment.ExecutionOperations, installation, node string, generation uint64, idle ...string) deployment.Allocation {
idleFor := "1 minute"
if len(idle) > 0 {
idleFor = idle[0]
}
t.Helper()
key := hostedEnvironment(t, f.pool)
if _, err := f.pool.Exec(t.Context(), `INSERT INTO runtime_placements(environment_id,node_id,deployment_generation) VALUES($1,$2,$3)`, key.EnvironmentID, node, generation); err != nil {
Expand All @@ -25,7 +29,7 @@ func runningPressureOwner(t *testing.T, f fixture, changes *deployment.Execution
if _, err = f.pool.Exec(t.Context(), `UPDATE environments SET initialization='complete' WHERE id=$1`, key.EnvironmentID); err != nil {
t.Fatal(err)
}
if _, err = f.pool.Exec(t.Context(), `UPDATE runtime_allocations SET state='running',create_settled=true,compute_phase='running',compute_activity_at=clock_timestamp()-interval '1 minute',compute_phase_changed_at=clock_timestamp()-interval '1 minute' WHERE id=$1`, owner.ID); err != nil {
if _, err = f.pool.Exec(t.Context(), `UPDATE runtime_allocations SET state='running',create_settled=true,compute_phase='running',compute_activity_at=clock_timestamp()-$2::interval,compute_phase_changed_at=clock_timestamp()-$2::interval WHERE id=$1`, owner.ID, idleFor); err != nil {
t.Fatal(err)
}
owner, err = f.adapter.EnvironmentAllocation(t.Context(), key)
Expand All @@ -39,21 +43,26 @@ func TestPressureSuspensionRequiresUsefulCapacity(t *testing.T) {
for _, test := range []struct {
name string
retained int
idle string
maxActive int
demand, freeNode bool
want error
}{
{"no demand", 3, false, false, deployment.ErrNotIdle},
{"retained full", 1, true, false, deployment.ErrNotIdle},
{"other node free", 3, true, true, deployment.ErrNotIdle},
{"waiting first session", 3, true, false, nil},
{"no demand", 3, "1 minute", 1, false, false, deployment.ErrNotIdle},
{"retained full", 1, "1 minute", 1, true, false, deployment.ErrNotIdle},
{"other node free", 3, "1 minute", 1, true, true, deployment.ErrNotIdle},
{"waiting first session", 3, "1 minute", 1, true, false, nil},
{"full node idle past two minutes", 3, "3 minutes", 1, false, false, nil},
{"node with a free slot", 3, "3 minutes", 2, false, false, deployment.ErrNotIdle},
{"retained full idle past two minutes", 1, "3 minutes", 1, false, false, deployment.ErrNotIdle},
} {
t.Run(test.name, func(t *testing.T) {
f := newFixture(t)
changes, _ := f.execution(t)
installation, view := f.initialize(t, changes, sandbox.Selection{Provider: "microsandbox", DeploymentSpec: testSpecification("microsandbox")})
node := f.enroll(t, view, deployment.Capacity{MaxActive: 1, MaxRetained: test.retained})
node := f.enroll(t, view, deployment.Capacity{MaxActive: test.maxActive, MaxRetained: test.retained})
f.connect(t, node.NodeID)
owner := runningPressureOwner(t, f, changes, installation, node.NodeID, view.Generation)
owner := runningPressureOwner(t, f, changes, installation, node.NodeID, view.Generation, test.idle)
if test.demand {
pendingPressureDemand(t, f)
}
Expand Down
Loading