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
2 changes: 1 addition & 1 deletion docs/architecture.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
108 changes: 29 additions & 79 deletions internal/app/api_command.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down Expand Up @@ -725,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{})
Expand All @@ -748,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",
Expand Down
6 changes: 2 additions & 4 deletions internal/app/local_sandbox_windows_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
Expand Down
128 changes: 128 additions & 0 deletions internal/application/command_runtime_set.go
Original file line number Diff line number Diff line change
@@ -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
}
76 changes: 76 additions & 0 deletions internal/application/command_runtime_set_test.go
Original file line number Diff line number Diff line change
@@ -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)
}
}
Loading
Loading