From 5e708010cdb5345786f1cbf6bf5a885ccd8c2cf8 Mon Sep 17 00:00:00 2001 From: Max Date: Mon, 30 Mar 2026 08:10:58 +0800 Subject: [PATCH] fix(sandbox/v2): update initSandboxV2 to return a per-request SandboxConfig Modified the initSandboxV2 function to return a per-request copy of the SandboxConfig, ensuring that concurrent requests to the same assistant have their own mutable configuration. Updated related logging and cleanup processes to utilize the new configuration structure. Additionally, introduced a DefaultOneShotMaxAge constant to manage the lifecycle of OneShot containers more effectively. --- agent/assistant/agent.go | 6 +++-- agent/assistant/sandbox_v2.go | 25 +++++++++++-------- agent/sandbox/v2/stream.go | 47 ++++------------------------------- sandbox/v2/types.go | 1 + sandbox/v2/watcher.go | 25 +++++++++++++++++++ 5 files changed, 50 insertions(+), 54 deletions(-) diff --git a/agent/assistant/agent.go b/agent/assistant/agent.go index 4f2fc7df..c1f62e29 100644 --- a/agent/assistant/agent.go +++ b/agent/assistant/agent.go @@ -171,11 +171,12 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa var v2Computer infraV2.Computer var v2LoadingMsgID string + var v2Cfg *sandboxTypes.SandboxConfig if ast.HasSandboxV2() { ctx.Logger.Phase("Sandbox V2") var err error var v2Cleanup func() - v2Runner, v2Computer, v2Cleanup, v2LoadingMsgID, err = ast.initSandboxV2(ctx, opts) + v2Runner, v2Computer, v2Cfg, v2Cleanup, v2LoadingMsgID, err = ast.initSandboxV2(ctx, opts) if err != nil { ast.traceAgentFail(agentNode, err) ast.sendStreamEndOnError(ctx, streamHandler, streamStartTime, err) @@ -189,7 +190,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa if ci.BoxID != "" { ctx.Logger.Trace("Computer: %s", ci.BoxID) } - ctx.Logger.Trace("Workspace: %s", ast.SandboxV2.WorkspaceID) + ctx.Logger.Trace("Workspace: %s", v2Cfg.WorkspaceID) if conn, _, err := ast.GetConnector(ctx, opts); err == nil && conn != nil { ctx.Logger.Trace("Connector: %s", conn.ID()) } @@ -337,6 +338,7 @@ func (ast *Assistant) Stream(ctx *context.Context, inputMessages []context.Messa Handler: streamHandler, Runner: v2Runner, Computer: v2Computer, + Config: v2Cfg, LoadingMsgID: v2LoadingMsgID, Options: opts, }) diff --git a/agent/assistant/sandbox_v2.go b/agent/assistant/sandbox_v2.go index 72fdc871..d01e33ba 100644 --- a/agent/assistant/sandbox_v2.go +++ b/agent/assistant/sandbox_v2.go @@ -26,12 +26,16 @@ func (ast *Assistant) HasSandboxV2() bool { } // initSandboxV2 initializes the V2 sandbox: obtains a Computer, gets a Runner, -// runs Prepare, and returns the runner, computer, cleanup closure, loading -// message ID, and any error. +// runs Prepare, and returns the runner, computer, a per-request copy of the +// SandboxConfig, cleanup closure, loading message ID, and any error. +// +// A shallow copy of ast.SandboxV2 is made so that concurrent requests to the +// same assistant each get their own mutable config (Owner, ID, NodeID, etc.). func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) ( - sandboxTypes.Runner, infraV2.Computer, func(), string, error, + sandboxTypes.Runner, infraV2.Computer, *sandboxTypes.SandboxConfig, func(), string, error, ) { - cfg := ast.SandboxV2 + cfgCopy := *ast.SandboxV2 + cfg := &cfgCopy manager := infraV2.M() loadingMsg := &message.Message{ @@ -48,7 +52,7 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) conn, _, err := ast.GetConnector(ctx, opts) if err != nil && cfg.Runner.Name != "yao" { closeLoadingV2(ctx, loadingMsgID, "sandbox.failed") - return nil, nil, nil, "", fmt.Errorf("get connector: %w", err) + return nil, nil, nil, nil, "", fmt.Errorf("get connector: %w", err) } // 2. Build human-readable DisplayName from real Agent name + Workspace name. @@ -85,7 +89,7 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) computer, identifier, err := sandboxv2.GetComputer(ctx, cfg, manager) if err != nil { closeLoadingV2(ctx, loadingMsgID, "sandbox.failed") - return nil, nil, nil, "", fmt.Errorf("getComputer failed: %w", err) + return nil, nil, nil, nil, "", fmt.Errorf("getComputer failed: %w", err) } _ = identifier @@ -94,7 +98,7 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) if err != nil { sandboxv2.LifecycleAction(stdCtx, cfg, computer, manager) closeLoadingV2(ctx, loadingMsgID, "sandbox.failed") - return nil, nil, nil, "", fmt.Errorf("get runner %q: %w", cfg.Runner.Name, err) + return nil, nil, nil, nil, "", fmt.Errorf("get runner %q: %w", cfg.Runner.Name, err) } // 5. Resolve assistant directory and skills subdirectory. @@ -136,7 +140,7 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) runner.Cleanup(stdCtx, computer) sandboxv2.LifecycleAction(stdCtx, cfg, computer, manager) closeLoadingV2(ctx, loadingMsgID, "sandbox.failed") - return nil, nil, nil, "", fmt.Errorf("runner.Prepare: %w", err) + return nil, nil, nil, nil, "", fmt.Errorf("runner.Prepare: %w", err) } // Inject computer + workspace into context so Create/Next hooks @@ -150,7 +154,7 @@ func (ast *Assistant) initSandboxV2(ctx *context.Context, opts *context.Options) sandboxv2.LifecycleAction(cleanCtx, cfg, computer, manager) } - return runner, computer, cleanup, loadingMsgID, nil + return runner, computer, cfg, cleanup, loadingMsgID, nil } // sandboxV2StreamParams groups arguments for executeSandboxV2Stream. @@ -160,6 +164,7 @@ type sandboxV2StreamParams struct { Handler message.StreamFunc Runner sandboxTypes.Runner Computer infraV2.Computer + Config *sandboxTypes.SandboxConfig LoadingMsgID string Options *context.Options } @@ -171,7 +176,7 @@ func (ast *Assistant) executeSandboxV2Stream( ) (*context.CompletionResponse, error) { _ = p.AgentNode - cfg := ast.SandboxV2 + cfg := p.Config manager := infraV2.M() // Build system prompt (parse $CTX variables the same way as buildSystemPrompts). diff --git a/agent/sandbox/v2/stream.go b/agent/sandbox/v2/stream.go index c6615ceb..e72ea868 100644 --- a/agent/sandbox/v2/stream.go +++ b/agent/sandbox/v2/stream.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "log" "time" agentContext "github.com/yaoapp/yao/agent/context" @@ -24,9 +23,11 @@ type ExecuteRequest struct { LoadingMsgID string } -// ExecuteSandboxStream is the V2 replacement for executeSandboxStream. -// It calls runner.Stream, handles interrupts, and performs cleanup/lifecycle -// in defer. +// ExecuteSandboxStream runs runner.Stream and bridges agentContext interrupts. +// +// Cleanup (runner.Cleanup + LifecycleAction) is NOT performed here; the caller +// (agent.go sandboxCleanup closure) is responsible for all lifecycle management +// so that cleanup happens exactly once regardless of code path. func ExecuteSandboxStream( ctx *agentContext.Context, req *ExecuteRequest, @@ -38,42 +39,6 @@ func ExecuteSandboxStream( } stdCtx := ctx.Context - panicked := true // Assume panic; set false on normal exit. - - // Resolve stop timeout from config (default 2s). - stopTimeout := 2 * time.Second - if req.Config != nil && req.Config.StopTimeout != "" { - if d, err := time.ParseDuration(req.Config.StopTimeout); err == nil { - stopTimeout = d - } - } - - // Panic recovery (registered first, executes last in LIFO order). - defer func() { - if r := recover(); r != nil { - log.Printf("[sandbox/v2] panic in stream: %v", r) - cleanCtx, cancel := context.WithTimeout(context.Background(), stopTimeout) - defer cancel() - req.Runner.Cleanup(cleanCtx, req.Computer) - LifecycleAction(cleanCtx, req.Config, req.Computer, req.Manager) - } - }() - - // Lifecycle action (registered second, executes second-to-last). - defer func() { - if !panicked { - LifecycleAction(stdCtx, req.Config, req.Computer, req.Manager) - } - }() - - // Runner cleanup (registered last, executes first). - defer func() { - if !panicked { - cleanCtx, cancel := context.WithTimeout(context.Background(), stopTimeout) - defer cancel() - req.Runner.Cleanup(cleanCtx, req.Computer) - } - }() // Build a cancellable runnerCtx that bridges agentContext interrupts. runnerCtx, cancelRunner := context.WithCancel(stdCtx) @@ -144,8 +109,6 @@ func ExecuteSandboxStream( closeLoading(ctx, req.LoadingMsgID) } - panicked = false // Normal exit reached. - if err != nil { if errors.Is(err, context.Canceled) { return nil, err diff --git a/sandbox/v2/types.go b/sandbox/v2/types.go index 312aa54b..6334af99 100644 --- a/sandbox/v2/types.go +++ b/sandbox/v2/types.go @@ -75,6 +75,7 @@ const ( DefaultStopTimeout = 2 * time.Second DefaultSessionIdleTimeout = 30 * time.Minute DefaultLongRunningIdleTimeout = 2 * time.Hour + DefaultOneShotMaxAge = 8 * time.Hour ) // --------------------------------------------------------------------------- diff --git a/sandbox/v2/watcher.go b/sandbox/v2/watcher.go index 32c89af3..98034ed7 100644 --- a/sandbox/v2/watcher.go +++ b/sandbox/v2/watcher.go @@ -71,6 +71,31 @@ func (w *sandboxWatcher) Check(ctx context.Context) []monitor.Alert { } } + // OneShot safety net: these containers should have been removed by + // LifecycleAction right after execution. If they still exist after + // DefaultOneShotMaxAge it means cleanup failed (e.g. process crash, + // cfg.ID race before the fix). Remove them based on createdAt so we + // never kill a container that is still actively executing. + if b.policy == OneShot { + age := time.Since(b.createdAt) + if age > DefaultOneShotMaxAge { + alerts = append(alerts, monitor.Alert{ + Level: monitor.Warn, + Target: "box:" + b.id, + Message: fmt.Sprintf("oneshot exceeded max age (%s > %s), removing", age.Round(time.Second), DefaultOneShotMaxAge), + Action: func(ctx context.Context) { mgr.Remove(ctx, b.id) }, + }) + } else { + alerts = append(alerts, monitor.Alert{ + Level: monitor.Trace, + Target: "box:" + b.id, + Message: fmt.Sprintf("oneshot age %s (max=%s)", + age.Round(time.Second), DefaultOneShotMaxAge), + }) + } + return true + } + idle := time.Since(b.idleSince()) timeout := b.idleTimeout()