diff --git a/docs/sandbox-provider.md b/docs/sandbox-provider.md index 439b11a4c..8db0f4b89 100644 --- a/docs/sandbox-provider.md +++ b/docs/sandbox-provider.md @@ -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: @@ -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). diff --git a/docs/zh/sandbox-provider.md b/docs/zh/sandbox-provider.md index e75c09117..56f02f7ca 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: 95b02d2c1b46117e603ebadd5fa948c92624c3eb59d50cc402eb5a3907e69378 +source_hash: 6a767cd4c884d93698405a7ebbdd78489e204ed0bdacf0c36e5b29bd7346d1dc --- 沙箱提供商协议连接 Core 与为 Core 管理的（`openai_hosted`）Environment 提供计算资源的 adapter，并由该 adapter 运行启动 Runtime daemon 的有界引导流程。本页面向 adapter 作者：说明提供商必须实现什么，以及可以依赖 Core 做什么。Docker、microsandbox 和 E2B 实现了该协议。 @@ -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。 提供商依赖以下规则： @@ -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)。 diff --git a/services/core/internal/deployment/allocations.go b/services/core/internal/deployment/allocations.go index a1c182b67..69f8dd0dc 100644 --- a/services/core/internal/deployment/allocations.go +++ b/services/core/internal/deployment/allocations.go @@ -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 } diff --git a/services/core/internal/deployment/suspension_pressure.go b/services/core/internal/deployment/suspension_pressure.go index 789139b08..68911b0e6 100644 --- a/services/core/internal/deployment/suspension_pressure.go +++ b/services/core/internal/deployment/suspension_pressure.go @@ -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 { @@ -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 { diff --git a/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go b/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go index e81422e41..4a289b06b 100644 --- a/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go +++ b/services/core/internal/persistence/postgres/deploymentpg/suspension_pressure_test.go @@ -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 { @@ -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) @@ -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) }