From afcffe21bc876d8372a12bf9f245f9968d2ec75f Mon Sep 17 00:00:00 2001 From: Qiyuanqiii <2297740147@qq.com> Date: Mon, 5 Oct 2026 11:11:04 +0800 Subject: [PATCH 1/2] refactor: share Desktop and API command runtime assembly Centralize Host, Local and Docker command backend assembly and manager cleanup. Keep entry-specific startup gates and shut down owned managers before closing the Desktop store when later initialization fails. Add a partial-startup rollback regression and update the Windows Auto preference fixture to the current CLI contract. Co-authored-by: Codex --- docs/architecture.md | 2 +- internal/app/api_command.go | 89 ++------- internal/app/local_sandbox_windows_test.go | 6 +- internal/application/command_runtime_set.go | 128 ++++++++++++ .../application/command_runtime_set_test.go | 76 +++++++ internal/desktop/control_plane.go | 186 +++--------------- 6 files changed, 251 insertions(+), 236 deletions(-) create mode 100644 internal/application/command_runtime_set.go create mode 100644 internal/application/command_runtime_set_test.go diff --git a/docs/architecture.md b/docs/architecture.md index 6c086358..bee528d2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -29,7 +29,7 @@ Implementation status for this revision: | Migration fixtures | Simplified | [Historical fixtures](../internal/store/test_database_fixture_test.go) build real migration prefixes. [Seed import](../internal/store/historical_fixture_import_test.go) copies compatible test data; the cumulative reverse-DDL chain is gone. | | CI execution | Deduplicated | [CI](../.github/workflows/ci.yml) runs analyzer vectors within the Go suite and desktop boundary tests through the platform build scripts. Platform, race and standalone release checks remain distinct. | | Model and extension integrations | Existing boundaries | `llm` owns provider protocols; MCP shares its SDK execution path. Skills and Plugins use their supported formats. Compatibility is tracked per integration. | -| Runtime construction and teardown | Next refactoring candidate | Desktop, CLI and HTTP composition can share more setup and cleanup. This is separate from the completed registry and fixture work above. | +| Command runtime construction and teardown | Shared for Desktop and HTTP | [Command runtime set](../internal/application/command_runtime_set.go) assembles Host, Local and Docker adapters and shuts down their managers. Each entry selects its enabled backends; CLI retains its invocation-owned lifetime. | | Ecosystem product validation | Follow-up | Exercise representative integrations through install, configuration, authentication, execution, failure and cancellation. External Agent orchestration remains outside this phase. | The Gateway extension target is concrete: add a tool implementation and one diff --git a/internal/app/api_command.go b/internal/app/api_command.go index 9374f8db..927c696d 100644 --- a/internal/app/api_command.go +++ b/internal/app/api_command.go @@ -364,86 +364,31 @@ func (a *App) apiServeCommand(ctx context.Context, args []string) (resultErr err if err != nil { return err } - if _, err := commandManager.ReconcileStartup(ctx); err != nil { - return apperror.Wrap(apperror.CodeUnavailable, - "command runtime startup reconciliation failed", err) - } - commandManagers := []*runner.CommandRuntimeManager{commandManager} - defer func() { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 7*time.Second) - defer cancel() - for _, manager := range commandManagers { - _ = manager.Shutdown(shutdownCtx) - } - }() - commandAdapters := make([]*application.CommandRuntimeService, 0, 3) - if controlToken != "" && permissionCapabilities.DangerFullAccessEnabled { - hostRuntime, serviceErr := application.NewCommandRuntimeService(a.store, - commandManager, permissionCapabilities) - if serviceErr != nil { - return serviceErr - } - commandAdapters = append(commandAdapters, hostRuntime) + commandOptions := application.CommandRuntimeSetOptions{ + HostEnabled: controlToken != "" && permissionCapabilities.DangerFullAccessEnabled, + Capabilities: permissionCapabilities, Drydocks: commandRuntimeDrydocks, + StartupShutdownTimeout: 7 * time.Second, } if controlToken != "" && localSandboxBackend != nil && localSandboxReadiness != nil && commandRuntimeDrydocks != nil { - localExecutor, executorErr := application.NewLocalSandboxCommandRuntimeExecutor( - a.store, localSandboxBackend, *localSandboxReadiness) - if executorErr != nil { - return executorErr - } - localManager, managerErr := runner.NewSandboxCommandRuntimeManager(a.store, - localExecutor, idgen.New("command-runtime-local-owner")) - if managerErr != nil { - return managerErr - } - commandManagers = append(commandManagers, localManager) - if _, managerErr = localManager.ReconcileStartup(ctx); managerErr != nil { - return apperror.Wrap(apperror.CodeUnavailable, - "Local Command Runtime startup reconciliation failed", managerErr) - } - localRuntime, serviceErr := application.NewSandboxedCommandRuntimeService( - a.store, localManager, localExecutor, permissionCapabilities, - commandRuntimeDrydocks) - if serviceErr != nil { - return serviceErr - } - commandAdapters = append(commandAdapters, localRuntime) + commandOptions.LocalBackend = localSandboxBackend + commandOptions.LocalReadiness = localSandboxReadiness } if controlToken != "" && standardCodeRuntime != nil && commandRuntimeDrydocks != nil { - dockerExecutor, executorErr := - application.NewDockerSandboxCommandRuntimeExecutor(standardCodeRuntime) - if executorErr != nil { - return executorErr - } - dockerManager, managerErr := runner.NewSandboxCommandRuntimeManager(a.store, - dockerExecutor, idgen.New("command-runtime-docker-owner")) - if managerErr != nil { - return managerErr - } - commandManagers = append(commandManagers, dockerManager) - if _, managerErr = dockerManager.ReconcileStartup(ctx); managerErr != nil { - return apperror.Wrap(apperror.CodeUnavailable, - "Docker Command Runtime startup reconciliation failed", managerErr) - } - dockerRuntime, serviceErr := application.NewSandboxedCommandRuntimeService( - a.store, dockerManager, dockerExecutor, permissionCapabilities, - commandRuntimeDrydocks) - if serviceErr != nil { - return serviceErr - } - commandAdapters = append(commandAdapters, dockerRuntime) + commandOptions.StandardCodeDockerRuntime = standardCodeRuntime } - var commandRuntime application.CommandRuntimeRuntime - if len(commandAdapters) == 1 { - commandRuntime = commandAdapters[0] - } else if len(commandAdapters) > 1 { - commandRuntime, err = application.NewCommandRuntimeMultiplexer(commandAdapters...) - if err != nil { - return err - } + commandSet, err := application.OpenCommandRuntimeSet(ctx, a.store, commandManager, + commandOptions) + if err != nil { + return err } + defer func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 7*time.Second) + defer cancel() + _ = commandSet.Shutdown(shutdownCtx) + }() + commandRuntime := commandSet.Runtime if commandRuntime != nil { runtimeDependencies.CommandRuntime = commandRuntime } diff --git a/internal/app/local_sandbox_windows_test.go b/internal/app/local_sandbox_windows_test.go index a3787c94..231ea893 100644 --- a/internal/app/local_sandbox_windows_test.go +++ b/internal/app/local_sandbox_windows_test.go @@ -83,13 +83,11 @@ func TestWindowsLocalSandboxCLIProbeOpensOnlyWorkspaceGate(t *testing.T) { } selected, stderr, code := executeTestCommand(t, "run", "execution-permission", "set", - runID, "auto", "--operation-key", "local-sandbox-select-0001", - "--enable-permission-control", - "--enable-workspace-sandbox") + runID, "auto", "--operation-key", "local-sandbox-select-0001") if code != 0 || stderr != "" || !strings.Contains(selected, "mode: auto") || !strings.Contains(selected, "runtime_gate_available: true") || !strings.Contains(selected, "capability_grant: false") { - t.Fatalf("Workspace Access selection stdout=%s stderr=%s code=%d", + t.Fatalf("Auto preference selection stdout=%s stderr=%s code=%d", selected, stderr, code) } } diff --git a/internal/application/command_runtime_set.go b/internal/application/command_runtime_set.go new file mode 100644 index 00000000..66f32574 --- /dev/null +++ b/internal/application/command_runtime_set.go @@ -0,0 +1,128 @@ +package application + +import ( + "context" + "errors" + "time" + + "cyberagent-workbench/internal/apperror" + "cyberagent-workbench/internal/domain" + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/runner" + "cyberagent-workbench/internal/sandbox" +) + +type CommandRuntimeSetStore interface { + CommandRuntimeStore + LocalSandboxCommandRuntimeStore +} + +// Hosts select their startup gates and pass the same capabilities to every +// service. The set borrows the store, Drydock and sandbox backends. +type CommandRuntimeSetOptions struct { + HostEnabled bool + Capabilities domain.ExecutionPermissionRuntimeCapabilities + Drydocks *RunWorktreeService + LocalBackend sandbox.LocalBackend + LocalReadiness *sandbox.LocalReadiness + StandardCodeDockerRuntime *StandardCodeDockerService + StartupShutdownTimeout time.Duration +} + +type CommandRuntimeSet struct { + Runtime CommandRuntimeRuntime + managers []*runner.CommandRuntimeManager +} + +// OpenCommandRuntimeSet takes ownership of recoveryManager, including when +// startup fails. It starts no workers or jobs and does not activate authority. +func OpenCommandRuntimeSet(ctx context.Context, store CommandRuntimeSetStore, + recoveryManager *runner.CommandRuntimeManager, options CommandRuntimeSetOptions, +) (_ *CommandRuntimeSet, resultErr error) { + set := &CommandRuntimeSet{managers: []*runner.CommandRuntimeManager{recoveryManager}} + defer func() { + if resultErr != nil { + shutdownCtx, cancel := context.WithTimeout(context.Background(), options.StartupShutdownTimeout) + defer cancel() + _ = set.Shutdown(shutdownCtx) + } + }() + if _, err := recoveryManager.ReconcileStartup(ctx); err != nil { + return nil, apperror.Wrap(apperror.CodeUnavailable, + "command runtime startup reconciliation failed", err) + } + adapters := make([]*CommandRuntimeService, 0, 3) + if options.HostEnabled { + service, err := NewCommandRuntimeService(store, recoveryManager, options.Capabilities) + if err != nil { + return nil, err + } + adapters = append(adapters, service) + } + if options.LocalBackend != nil && options.LocalReadiness != nil && options.Drydocks != nil { + executor, err := NewLocalSandboxCommandRuntimeExecutor(store, + options.LocalBackend, *options.LocalReadiness) + if err != nil { + return nil, err + } + manager, err := runner.NewSandboxCommandRuntimeManager(store, executor, + idgen.New("command-runtime-local-owner")) + if err != nil { + return nil, err + } + set.managers = append(set.managers, manager) + if _, err := manager.ReconcileStartup(ctx); err != nil { + return nil, apperror.Wrap(apperror.CodeUnavailable, + "Local Command Runtime startup reconciliation failed", err) + } + service, err := NewSandboxedCommandRuntimeService(store, manager, executor, + options.Capabilities, options.Drydocks) + if err != nil { + return nil, err + } + adapters = append(adapters, service) + } + if options.StandardCodeDockerRuntime != nil && options.Drydocks != nil { + executor, err := NewDockerSandboxCommandRuntimeExecutor(options.StandardCodeDockerRuntime) + if err != nil { + return nil, err + } + manager, err := runner.NewSandboxCommandRuntimeManager(store, executor, + idgen.New("command-runtime-docker-owner")) + if err != nil { + return nil, err + } + set.managers = append(set.managers, manager) + if _, err := manager.ReconcileStartup(ctx); err != nil { + return nil, apperror.Wrap(apperror.CodeUnavailable, + "Docker Command Runtime startup reconciliation failed", err) + } + service, err := NewSandboxedCommandRuntimeService(store, manager, executor, + options.Capabilities, options.Drydocks) + if err != nil { + return nil, err + } + adapters = append(adapters, service) + } + if len(adapters) == 1 { + set.Runtime = adapters[0] + } else if len(adapters) > 1 { + var err error + set.Runtime, err = NewCommandRuntimeMultiplexer(adapters...) + if err != nil { + return nil, err + } + } + return set, nil +} + +func (s *CommandRuntimeSet) Shutdown(ctx context.Context) error { + if s == nil { + return nil + } + var result error + for _, manager := range s.managers { + result = errors.Join(result, manager.Shutdown(ctx)) + } + return result +} diff --git a/internal/application/command_runtime_set_test.go b/internal/application/command_runtime_set_test.go new file mode 100644 index 00000000..cc68afcb --- /dev/null +++ b/internal/application/command_runtime_set_test.go @@ -0,0 +1,76 @@ +package application + +import ( + "context" + "runtime" + "strings" + "testing" + "time" + + "cyberagent-workbench/internal/idgen" + "cyberagent-workbench/internal/runner" + "cyberagent-workbench/internal/sandbox" + "cyberagent-workbench/internal/toolgateway" +) + +type commandRuntimeSetUnreadyLocalBackend struct{ sandbox.LocalBackend } + +func TestCommandRuntimeSetPartialStartupFailureReapsOwnedJobBeforeReturning(t *testing.T) { + ctx := t.Context() + state, run, root, lease, capabilities := newCommandRuntimeTestRuntime(t, ctx) + manager, err := runner.NewPlatformCommandRuntimeManager(state, + idgen.New("partial-startup-owner")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := manager.Shutdown(shutdownCtx); err != nil { + t.Error(err) + } + }) + service, err := NewCommandRuntimeService(state, manager, capabilities) + if err != nil { + t.Fatal(err) + } + profile, script := runner.CommandRuntimeBash, "sleep 10" + if runtime.GOOS == "windows" { + profile, script = runner.CommandRuntimePowerShell, "Start-Sleep -Seconds 10" + } + scope := commandRuntimeTestScope(t, ctx, state, service, run, root, lease, "partial-startup") + fixture := commandFixtureForScope(t, state, service, scope) + started, err := fixture.execute(t, ctx, toolgateway.CommandRuntimeInput{ + Version: toolgateway.CommandRuntimeToolProtocolVersion, + Action: toolgateway.CommandRuntimeActionStart, + Commands: []runner.CommandRuntimeSpec{{ + Version: runner.CommandRuntimeProtocolVersion, Profile: profile, + Script: script, WorkingDirectory: ".", Environment: []runner.CommandRuntimeEnvironment{}, + StdinPolicy: runner.CommandRuntimeStdinClosed, CloseInitialStdin: true, + TimeoutMilliseconds: 30_000, + Output: runner.CommandRuntimeOutputPolicy{InlineBytes: 4096, ArtifactBytes: 4096}, + Network: runner.CommandRuntimeNetworkDisabled, Credentials: runner.CommandRuntimeCredentialsNone, + Purpose: "hold an owned Job until partial startup rollback", + }}, + }, 1) + if err != nil || len(started.Jobs) != 1 || started.Jobs[0].State != runner.CommandRuntimeJobRunning { + t.Fatalf("owned Job did not start: jobs=%+v err=%v", started.Jobs, err) + } + set, err := OpenCommandRuntimeSet(ctx, state, manager, CommandRuntimeSetOptions{ + HostEnabled: true, Capabilities: capabilities, + Drydocks: &RunWorktreeService{}, LocalBackend: commandRuntimeSetUnreadyLocalBackend{}, + LocalReadiness: &sandbox.LocalReadiness{}, StartupShutdownTimeout: 5 * time.Second, + }) + if set != nil || err == nil || !strings.Contains(err.Error(), "Local Sandbox Command Runtime readiness is invalid") { + t.Fatalf("partial Local startup failure was lost: set=%v err=%v", set, err) + } + job, err := state.GetCommandRuntimeJob(ctx, started.Jobs[0].ID) + if err != nil || job.State != runner.CommandRuntimeJobInterrupted || !job.TreeReaped { + t.Fatalf("failed startup returned before the owned Job drained: job=%+v err=%v", job, err) + } + if _, err := NewNoteService(state).Create(ctx, CreateNoteRequest{ + RunID: run.ID, Title: "startup failure retained store", Content: "the caller still owns the writable SQLite connection", + }); err != nil { + t.Fatalf("startup rollback closed or invalidated the borrowed store: %v", err) + } +} diff --git a/internal/desktop/control_plane.go b/internal/desktop/control_plane.go index 4cf3bcbf..454b5e6f 100644 --- a/internal/desktop/control_plane.go +++ b/internal/desktop/control_plane.go @@ -59,7 +59,7 @@ type ControlPlane struct { fullCDPSessions *application.FullCDPProductionService agentBrowser *application.AgentBrowserService commandRuntimeManager *runner.CommandRuntimeManager - commandRuntimeManagers []*runner.CommandRuntimeManager + commandRuntimeSet *application.CommandRuntimeSet commandRuntimeAdapterInstalled bool commandRuntimeAdapterReady bool standardCodePresetEnabled bool @@ -499,131 +499,49 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { if standardCodeDelivery != nil { standardCodeDeliveryController = standardCodeDelivery } + var commandSet *application.CommandRuntimeSet + commandRuntimeSetTransferred := false + defer func() { + if !commandRuntimeSetTransferred { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + _ = commandSet.Shutdown(shutdownCtx) + cancel() + _ = stateStore.Close() + } + }() commandManager, err := runner.NewPlatformCommandRuntimeManager(stateStore, idgen.New("command-runtime-owner")) if err != nil { - _ = stateStore.Close() return nil, err } - if _, err := commandManager.ReconcileStartup(context.Background()); err != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - _ = commandManager.Shutdown(shutdownCtx) - cancel() - _ = stateStore.Close() - return nil, apperror.Wrap(apperror.CodeUnavailable, - "command runtime startup reconciliation failed", err) - } - commandManagers := []*runner.CommandRuntimeManager{commandManager} - commandManagersTransferred := false - defer func() { - if !commandManagersTransferred { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - } - }() - commandAdapters := make([]*application.CommandRuntimeService, 0, 3) - if config.RunExecutionEnabled && config.ExecutionPermissionCapabilities.DangerFullAccessEnabled { - hostRuntime, serviceErr := application.NewCommandRuntimeService(stateStore, - commandManager, config.ExecutionPermissionCapabilities) - if serviceErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, serviceErr - } - commandAdapters = append(commandAdapters, hostRuntime) + commandOptions := application.CommandRuntimeSetOptions{ + HostEnabled: config.RunExecutionEnabled && config.ExecutionPermissionCapabilities.DangerFullAccessEnabled, + Capabilities: config.ExecutionPermissionCapabilities, Drydocks: commandRuntimeDrydocks, + StartupShutdownTimeout: 2 * time.Second, } if config.RunExecutionEnabled && config.LocalSandboxBackend != nil && config.LocalSandboxReadiness != nil && commandRuntimeDrydocks != nil { - localExecutor, executorErr := application.NewLocalSandboxCommandRuntimeExecutor( - stateStore, config.LocalSandboxBackend, *config.LocalSandboxReadiness) - if executorErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, executorErr - } - localManager, managerErr := runner.NewSandboxCommandRuntimeManager(stateStore, - localExecutor, idgen.New("command-runtime-local-owner")) - if managerErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, managerErr - } - commandManagers = append(commandManagers, localManager) - if _, managerErr = localManager.ReconcileStartup(context.Background()); managerErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, apperror.Wrap(apperror.CodeUnavailable, - "Local Command Runtime startup reconciliation failed", managerErr) - } - localRuntime, serviceErr := application.NewSandboxedCommandRuntimeService( - stateStore, localManager, localExecutor, - config.ExecutionPermissionCapabilities, commandRuntimeDrydocks) - if serviceErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, serviceErr - } - commandAdapters = append(commandAdapters, localRuntime) + commandOptions.LocalBackend = config.LocalSandboxBackend + commandOptions.LocalReadiness = config.LocalSandboxReadiness } if config.RunExecutionEnabled && standardCodeRuntime != nil && commandRuntimeDrydocks != nil { - dockerExecutor, executorErr := - application.NewDockerSandboxCommandRuntimeExecutor(standardCodeRuntime) - if executorErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, executorErr - } - dockerManager, managerErr := runner.NewSandboxCommandRuntimeManager(stateStore, - dockerExecutor, idgen.New("command-runtime-docker-owner")) - if managerErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, managerErr - } - commandManagers = append(commandManagers, dockerManager) - if _, managerErr = dockerManager.ReconcileStartup(context.Background()); managerErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, apperror.Wrap(apperror.CodeUnavailable, - "Docker Command Runtime startup reconciliation failed", managerErr) - } - dockerRuntime, serviceErr := application.NewSandboxedCommandRuntimeService( - stateStore, dockerManager, dockerExecutor, - config.ExecutionPermissionCapabilities, commandRuntimeDrydocks) - if serviceErr != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, serviceErr - } - commandAdapters = append(commandAdapters, dockerRuntime) + commandOptions.StandardCodeDockerRuntime = standardCodeRuntime } - var commandRuntime application.CommandRuntimeRuntime - if len(commandAdapters) == 1 { - commandRuntime = commandAdapters[0] - } else if len(commandAdapters) > 1 { - commandRuntime, err = application.NewCommandRuntimeMultiplexer(commandAdapters...) - if err != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() - return nil, err - } + commandSet, err = application.OpenCommandRuntimeSet(context.Background(), stateStore, + commandManager, commandOptions) + if err != nil { + return nil, err } + commandRuntime := commandSet.Runtime if commandRuntime != nil { runtimeDependencies.CommandRuntime = commandRuntime } uiEvidence, err := application.NewUIEvidenceReadService(stateStore) if err != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - _ = commandManager.Shutdown(shutdownCtx) - cancel() - _ = stateStore.Close() return nil, err } if _, err := uiEvidence.Reconcile(context.Background()); err != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - _ = commandManager.Shutdown(shutdownCtx) - cancel() - _ = stateStore.Close() return nil, apperror.Wrap(apperror.CodeUnavailable, "desktop UI evidence startup reconciliation failed", err) } @@ -632,10 +550,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { var controllerErr error browserController, controllerErr = browserruntime.NewPlatformBrowserProcessController() if controllerErr != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - _ = commandManager.Shutdown(shutdownCtx) - cancel() - _ = stateStore.Close() return nil, controllerErr } } @@ -646,19 +560,11 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { browserProvider, providerErr := application.NewSafeWebUIEvidenceBrowserProvider(browserService) if providerErr != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - _ = commandManager.Shutdown(shutdownCtx) - cancel() - _ = stateStore.Close() return nil, providerErr } uiEvidence, err = application.NewUIEvidenceService(stateStore, commandRuntime, browserProvider, filepath.Join(home, "runtime", "ui-evidence-profiles"), config.ExecutionPermissionCapabilities) if err != nil { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) - _ = commandManager.Shutdown(shutdownCtx) - cancel() - _ = stateStore.Close() return nil, err } } @@ -671,8 +577,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { config.BrowserCDPPermissionCapabilities, config.ExecutionPermissionCapabilities, home) if err != nil { - shutdownDesktopCommandRuntimeManagers(commandManagers, 2*time.Second) - _ = stateStore.Close() return nil, err } fullCDPSessionControlEnabled = true @@ -697,7 +601,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { dockerProposalExecutor, err := application.NewDockerSandboxProposalExecutor( dockerSandbox) if err != nil { - _ = stateStore.Close() return nil, err } runtimeDependencies.DockerSandbox = dockerProposalExecutor @@ -711,7 +614,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { mcpExecutor, err := application.NewMCPClientToolExecutor(mcpClient, stateStore, config.ExecutionPermissionCapabilities) if err != nil { - _ = stateStore.Close() return nil, err } approvalGateway.WithMCPExecutor(mcpExecutor) @@ -723,7 +625,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { stateStore, providerSearchResolver) providerDefinitionControl, err := application.NewProviderDefinitionService(stateStore, models) if err != nil { - _ = stateStore.Close() return nil, err } providerDefinitionControl.WithModelDiscoveryCredentials(credentialStore) @@ -742,7 +643,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { application.NewRunWakeCoordinator(stateStore), runWakeExecution, application.RunWakeWorkerConfig{OnError: config.OnWakeWorkerError}) if err != nil { - _ = stateStore.Close() return nil, err } } @@ -761,7 +661,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { OnError: config.OnScheduledJobWorkerError, }) if err != nil { - _ = stateStore.Close() return nil, err } } @@ -774,12 +673,10 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { if config.SkillInstallationEnabled { objects, objectErr := skills.NewLocalPackageObjectStore(home) if objectErr != nil { - _ = stateStore.Close() return nil, objectErr } registry, registryErr := skills.BuiltinRegistry() if registryErr != nil { - _ = stateStore.Close() return nil, registryErr } skillInstaller = application.NewSkillPackageRegistryService(stateStore, @@ -793,21 +690,18 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { terminalBroker := executionauth.NewTerminalInputBroker() terminalManager, err = terminalruntime.NewPlatformManager(terminalBroker) if err != nil { - _ = stateStore.Close() return nil, err } userTerminal, err = newDesktopUserTerminalService(stateStore, terminalManager, config.ExecutionPermissionCapabilities) if err != nil { _ = terminalManager.Shutdown() - _ = stateStore.Close() return nil, err } agentInputBridge, bridgeErr := terminalruntime.NewAgentInputBridge( terminalManager, terminalBroker) if bridgeErr != nil { _ = terminalManager.Shutdown() - _ = stateStore.Close() return nil, bridgeErr } debugAgentInput, err = application.NewDebugTerminalAgentInputService( @@ -815,14 +709,12 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { config.ExecutionPermissionCapabilities, true) if err != nil { _ = terminalManager.Shutdown() - _ = stateStore.Close() return nil, err } boundaryMonitor, err = terminalruntime.NewPlatformHostBoundaryMonitor( terminalManager) if err != nil { _ = terminalManager.Shutdown() - _ = stateStore.Close() return nil, err } } @@ -857,7 +749,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { if terminalManager != nil { _ = terminalManager.Shutdown() } - _ = stateStore.Close() return nil, err } } @@ -880,12 +771,10 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { if config.GitAdvancedControlEnabled || config.GitHubReviewControlEnabled { localGit, gitErr := repository.NewMutationExecutor() if gitErr != nil { - _ = stateStore.Close() return nil, gitErr } remoteGit, gitErr := repository.NewRemoteExecutor(credentialStore) if gitErr != nil { - _ = stateStore.Close() return nil, gitErr } threadGit = application.NewThreadGitService(stateStore, localGit, remoteGit, @@ -985,7 +874,6 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { if terminalManager != nil { _ = terminalManager.Shutdown() } - _ = stateStore.Close() return nil, err } // A review decision and its provider/tool continuation cannot share one @@ -1005,7 +893,7 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { }() } codeIntelTransferred = true - commandManagersTransferred = true + commandRuntimeSetTransferred = true return &ControlPlane{stateStore: stateStore, workspaceManager: workspaceManager, handler: api.Handler(), skillInstaller: skillInstaller, dockerSandbox: dockerSandbox, @@ -1019,7 +907,7 @@ func OpenControlPlane(config ControlPlaneConfig) (*ControlPlane, error) { fullCDPSessions: fullCDPSessions, agentBrowser: agentBrowser, commandRuntimeManager: commandManager, - commandRuntimeManagers: commandManagers, + commandRuntimeSet: commandSet, commandRuntimeAdapterInstalled: len(installedCommandRuntimeAdapters) > 0, commandRuntimeAdapterReady: commandRuntimeAdapterReady, standardCodePresetEnabled: standardCodePreset != nil, @@ -1262,21 +1150,6 @@ func (c *ControlPlane) reconcileCommandRuntime(ctx context.Context) error { return nil } -func shutdownDesktopCommandRuntimeManagers(managers []*runner.CommandRuntimeManager, - timeout time.Duration, -) { - if timeout <= 0 { - timeout = 2 * time.Second - } - ctx, cancel := context.WithTimeout(context.Background(), timeout) - defer cancel() - for _, manager := range managers { - if manager != nil { - _ = manager.Shutdown(ctx) - } - } -} - func (c *ControlPlane) Close() error { if c == nil { return nil @@ -1361,15 +1234,10 @@ func (c *ControlPlane) Close() error { c.closeErr = errors.Join(c.closeErr, <-fullCDPShutdown) fullCDPShutdownCancel() } - if len(c.commandRuntimeManagers) > 0 { + if c.commandRuntimeSet != nil { shutdownContext, shutdownCancel := context.WithTimeout( context.Background(), 7*time.Second) - for _, manager := range c.commandRuntimeManagers { - if manager != nil { - c.closeErr = errors.Join(c.closeErr, - manager.Shutdown(shutdownContext)) - } - } + c.closeErr = errors.Join(c.closeErr, c.commandRuntimeSet.Shutdown(shutdownContext)) shutdownCancel() } else if c.commandRuntimeManager != nil { shutdownContext, shutdownCancel := context.WithTimeout( From 14b5fc02fd4d9268364fc7351955fbffee5b599a Mon Sep 17 00:00:00 2001 From: Qiyuanqiii <2297740147@qq.com> Date: Mon, 5 Oct 2026 11:19:13 +0800 Subject: [PATCH 2/2] fix: join API command reconciliation before teardown Wait for both active-command and recovery-only reconcilers to finish before shutting down command managers and closing the application store. Co-authored-by: Codex --- internal/app/api_command.go | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/internal/app/api_command.go b/internal/app/api_command.go index 927c696d..e8bb4b33 100644 --- a/internal/app/api_command.go +++ b/internal/app/api_command.go @@ -670,7 +670,11 @@ func (a *App) apiServeCommand(ctx context.Context, args []string) (resultErr err return err } reconcileCtx, reconcileCancel := context.WithCancel(ctx) - defer reconcileCancel() + reconcileDone := make(chan struct{}) + defer func() { + reconcileCancel() + <-reconcileDone + }() if webFetchAuthorizationSchedulerEnabled { webFetchReconcileCtx, cancelWebFetchReconcile := context.WithCancel(ctx) webFetchReconcileDone := make(chan struct{}) @@ -693,16 +697,17 @@ func (a *App) apiServeCommand(ctx context.Context, args []string) (resultErr err <-webFetchReconcileDone }() } - if commandRuntime != nil { - go func() { + go func() { + defer close(reconcileDone) + if commandRuntime != nil { if reconcileErr := commandRuntime.RunReconciler(reconcileCtx, 500*time.Millisecond); reconcileErr != nil && reconcileCtx.Err() == nil { fmt.Fprintln(a.errOut, "command-runtime-reconciler:", reconcileErr) } - }() - } else { - go runCommandRuntimeStartupReconciler(reconcileCtx, commandManager, a.errOut) - } + } else { + runCommandRuntimeStartupReconciler(reconcileCtx, commandManager, a.errOut) + } + }() origin := "http://" + listener.Addr().String() baseURL := origin + "/api/v1" fmt.Fprintf(a.out, "api_url: %s\napi_version: %s\napi_token_generated: %t\napi_control_enabled: %t\n",