diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5efb020a4..e484884af 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -172,6 +172,32 @@ jobs: - name: Test run: cd server && go test -race ./... + windows-execenv: + # The environment-preparation deadline owns a process tree, not just a Go + # process. This Windows runtime test verifies Job Object cancellation kills + # a delayed descendant before an immediate retry can reuse the same root. + runs-on: windows-latest + steps: + - name: Checkout + uses: actions/checkout@v6 + + - name: Setup Go + uses: actions/setup-go@v5 + with: + go-version: "1.26.1" + cache-dependency-path: server/go.sum + + - name: Test Windows execution-environment isolation + working-directory: server + # Keep this job scoped to the runtime regression it exists to prove. + # The package's legacy OpenClaw HOME tests are not Windows-safe and are + # outside this PR; the normal backend job still runs the full package. + run: go test ./internal/daemon/execenv -run '^TestPrepareIsolated_WindowsKillsDescendantBeforeRetry$' -count=1 -timeout=5m + + - name: Build Windows CLI helper entrypoint + working-directory: server + run: go build ./cmd/multica + installer: # Stub-driven shell tests for scripts/install.sh. Kept off the heavy # backend job so installer regressions surface independently, and diff --git a/server/cmd/multica/main.go b/server/cmd/multica/main.go index 7491fd18d..81d14b6c0 100644 --- a/server/cmd/multica/main.go +++ b/server/cmd/multica/main.go @@ -2,12 +2,14 @@ package main import ( "fmt" + "log/slog" "os" "runtime" "github.com/spf13/cobra" "github.com/multica-ai/multica/server/internal/cli" + "github.com/multica-ai/multica/server/internal/daemon/execenv" ) var ( @@ -95,6 +97,14 @@ func init() { } func main() { + if len(os.Args) == 2 && os.Args[1] == execenv.PreparationHelperArg { + logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) + if err := execenv.RunPreparationHelper(os.Stdin, os.Stdout, logger); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + return + } cli.CleanupStaleUpdateArtifacts() if err := rootCmd.Execute(); err != nil { if err != errSilent { diff --git a/server/internal/daemon/daemon.go b/server/internal/daemon/daemon.go index 41923cd66..49abfe149 100644 --- a/server/internal/daemon/daemon.go +++ b/server/internal/daemon/daemon.go @@ -48,11 +48,23 @@ var ErrRepoNotConfigured = errors.New("repo is not configured for this workspace // stale-heartbeat sweep. var ErrNoRuntimesToRegister = errors.New("no agent runtimes could be registered") +// errTaskPrepareTimeout distinguishes the daemon's dispatched -> running +// startup deadline from provider execution timeouts. handleTask maps it to the +// platform-side timeout failure reason so the server's existing retry path can +// recover the task on a fresh attempt. +var errTaskPrepareTimeout = errors.New("task preparation timed out") + const ( taskSlotWaitTimeout = 2 * time.Second taskSlotCapacityBackoff = 5 * time.Second repoCheckoutModeEnv = "MULTICA_REPO_CHECKOUT_MODE" repoCheckoutModeIsolated = "isolated" + // defaultTaskPrepareTimeout is a hard liveness bound for everything after + // claim and before StartTask succeeds: runtime resolution, skill bundles, + // execution-environment setup, and the StartTask request itself. It is + // intentionally independent from AgentTimeout, which only governs the + // provider process after the task reaches running. + defaultTaskPrepareTimeout = 5 * time.Minute ) func repoCheckoutModeFor(provider, goos string) string { @@ -122,6 +134,16 @@ type terminalTaskReport struct { failureReason string } +type executionEnvironmentCommand func() ([]string, error) + +func defaultExecutionEnvironmentCommand() ([]string, error) { + executable, err := resolveSelfExecutable() + if err != nil { + return nil, fmt.Errorf("resolve execution-environment helper: %w", err) + } + return []string{executable, execenv.PreparationHelperArg}, nil +} + var ( isBrewInstall = cli.IsBrewInstall getBrewPrefix = cli.GetBrewPrefix @@ -323,6 +345,13 @@ type Daemon struct { runner taskRunner // executes agent tasks; set to d.runTask by New(), overridable in tests cancelPollInterval time.Duration // how often handleTask polls for server-side cancellation; overridable in tests + // executionEnvironmentCommand resolves the killable helper used for + // Prepare/Reuse. New always sets it; nil keeps focused unit tests in-process. + executionEnvironmentCommand executionEnvironmentCommand + // taskPrepareTimeout is the dispatched -> running hard deadline. New sets + // the production default; zero-valued test daemons fall back to the same + // default in effectiveTaskPrepareTimeout. + taskPrepareTimeout time.Duration // runUpdateFn executes the brew-or-download upgrade. Set to d.runUpdate by // New() and overridable in tests so the auto-update poller can be exercised // without touching the real network or the brew CLI. @@ -365,12 +394,14 @@ func New(cfg Config, logger *slog.Logger) *Daemon { reregisterNextAttempt: make(map[string]time.Time), reregisterLastCompletedAt: make(map[string]time.Time), cancelPollInterval: 5 * time.Second, + taskPrepareTimeout: defaultTaskPrepareTimeout, reconcile: newReconcileBroadcaster(), workspaceChanges: newWorkspaceChangeSignal(), wsRPC: newWSRPCClient(wsRPCResponseGrace), } d.activeEnvRootsCond = sync.NewCond(&d.activeEnvRootsMu) d.activeCodexStoresCond = sync.NewCond(&d.activeCodexStoresMu) + d.executionEnvironmentCommand = defaultExecutionEnvironmentCommand d.runner = taskRunnerFunc(d.runTask) d.runUpdateFn = d.runUpdate return d @@ -3239,7 +3270,7 @@ func (d *Daemon) handleTask(ctx context.Context, task Task, slot int) { kind: terminalTaskReportFail, taskID: task.ID, errorMessage: err.Error(), - failureReason: taskfailure.Classify(err.Error()).String(), + failureReason: taskRunFailureReason(err), }); failErr != nil { taskLog.Error("fail task callback failed", "error", failErr) } @@ -3287,6 +3318,13 @@ func (d *Daemon) handleTask(ctx context.Context, task Task, slot int) { } } +func taskRunFailureReason(err error) string { + if errors.Is(err, errTaskPrepareTimeout) { + return taskfailure.ReasonTimeout.String() + } + return taskfailure.Classify(err.Error()).String() +} + // acquireLocalDirectoryLockIfNeeded inspects the task's project resources for // a local_directory pinned to this daemon, validates the path, and takes the // path mutex. Returns a release callback (nil when no local_directory @@ -3875,6 +3913,37 @@ func (d *Daemon) startTaskPrepareLeaseExtender(ctx context.Context, task Task, t } } +func (d *Daemon) prepareExecutionEnvironment(ctx context.Context, params execenv.PrepareParams) (*execenv.Environment, error) { + if d.executionEnvironmentCommand == nil { + // Focused runTask tests construct a zero-valued Daemon and keep setup + // in-process. Production Daemons created by New always use isolation. + return execenv.Prepare(params, d.logger) + } + command, err := d.executionEnvironmentCommand() + if err != nil { + return nil, err + } + return execenv.PrepareIsolated(ctx, command, params, d.logger) +} + +func (d *Daemon) reuseExecutionEnvironment(ctx context.Context, params execenv.ReuseParams) (*execenv.Environment, error) { + if d.executionEnvironmentCommand == nil { + return execenv.Reuse(params, d.logger), nil + } + command, err := d.executionEnvironmentCommand() + if err != nil { + return nil, err + } + return execenv.ReuseIsolated(ctx, command, params, d.logger) +} + +func (d *Daemon) effectiveTaskPrepareTimeout() time.Duration { + if d.taskPrepareTimeout > 0 { + return d.taskPrepareTimeout + } + return defaultTaskPrepareTimeout +} + func skillRefKey(source, id string) string { return source + "\x00" + id } @@ -3906,7 +3975,7 @@ func skillRefFromBundle(bundle SkillData) SkillRefData { } } -func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot int, taskLog *slog.Logger) (TaskResult, error) { +func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot int, taskLog *slog.Logger) (taskResult TaskResult, returnErr error) { // Refuse to spawn an agent without a workspace. An empty workspace_id // here would make MULTICA_WORKSPACE_ID empty in the agent env, and the // CLI would otherwise silently fall back to the user-global config — a @@ -3916,6 +3985,21 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i return TaskResult{}, fmt.Errorf("refusing to spawn agent: task has no workspace_id (task_id=%s)", task.ID) } + prepareTimeout := d.effectiveTaskPrepareTimeout() + prepareCtx, cancelPrepare := context.WithTimeoutCause(ctx, prepareTimeout, errTaskPrepareTimeout) + prepareComplete := false + defer func() { + cancelPrepare() + if prepareComplete || returnErr == nil || !errors.Is(context.Cause(prepareCtx), errTaskPrepareTimeout) { + return + } + // Collapse every deadline shape (context deadline, HTTP cancellation, + // or the explicit waitForExecutionEnvironment cause) into one sentinel + // that handleTask can classify as a retryable platform timeout. + taskResult = TaskResult{} + returnErr = fmt.Errorf("%w after %s", errTaskPrepareTimeout, prepareTimeout) + }() + // task.Repos is the authoritative repo list for this task — when the // claimed task belongs to a project with github_repo resources the server // has already narrowed it to project repos only. Make sure those URLs are @@ -3952,16 +4036,16 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i // upgrade deleted (MUL-4486). Only reached when no custom profile owns // the launch, so a custom runtime's path is never second-guessed and a // custom-only host pays no wasted re-resolution. - entry, resolvedVersion = d.resolveAgentEntry(ctx, provider, entry) + entry, resolvedVersion = d.resolveAgentEntry(prepareCtx, provider, entry) } if !ok { return TaskResult{}, fmt.Errorf("no agent configured for provider %q", provider) } - stopPrepareLease := d.startTaskPrepareLeaseExtender(ctx, task, taskLog) + stopPrepareLease := d.startTaskPrepareLeaseExtender(prepareCtx, task, taskLog) defer stopPrepareLease() - if err := d.ensureTaskSkillBundles(ctx, &task); err != nil { + if err := d.ensureTaskSkillBundles(prepareCtx, &task); err != nil { return TaskResult{}, err } @@ -4132,7 +4216,8 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i } } if shouldReusePriorWorkdir(task, localAssignment, d.cfg.WorkspacesRoot) { - env = execenv.Reuse(execenv.ReuseParams{ + var err error + env, err = d.reuseExecutionEnvironment(prepareCtx, execenv.ReuseParams{ WorkspacesRoot: d.cfg.WorkspacesRoot, Profile: d.cfg.Profile, WorkDir: task.PriorWorkDir, @@ -4147,7 +4232,10 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i HermesSourceMustExist: hermesSourceMustExist, HermesEnv: hermesEnv, Task: taskCtx, - }, d.logger) + }) + if err != nil { + return TaskResult{}, fmt.Errorf("reuse execution environment: %w", err) + } } if env == nil { var err error @@ -4171,7 +4259,7 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i if localAssignment != nil { prepParams.LocalWorkDir = localAssignment.AbsPath } - env, err = execenv.Prepare(prepParams, d.logger) + env, err = d.prepareExecutionEnvironment(prepareCtx, prepParams) if err != nil { return TaskResult{}, fmt.Errorf("prepare execution environment: %w", err) } @@ -4203,11 +4291,13 @@ func (d *Daemon) runTask(ctx context.Context, task Task, provider string, slot i // taskfailure.Classify path records the failure with the same // "start task failed: <…>" string and the same failure_reason // taxonomy as before — see MUL-2946 for the classifier contract. - if err := d.client.StartTask(ctx, task.ID); err != nil { + if err := d.client.StartTask(prepareCtx, task.ID); err != nil { stopPrepareLease() return TaskResult{}, fmt.Errorf("start task failed: %w", err) } stopPrepareLease() + prepareComplete = true + cancelPrepare() _ = d.client.ReportProgress(ctx, task.ID, fmt.Sprintf("Launching %s", provider), 1, 2) reused := gateResumeToReusedWorkdir(&task, &taskCtx, env.WorkDir, taskLog) diff --git a/server/internal/daemon/execenv/isolation.go b/server/internal/daemon/execenv/isolation.go new file mode 100644 index 000000000..9b8e55e13 --- /dev/null +++ b/server/internal/daemon/execenv/isolation.go @@ -0,0 +1,188 @@ +package execenv + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "os/exec" + "strings" + "time" +) + +// PreparationHelperArg selects the private execution-environment helper mode +// in the multica binary. The daemon runs Prepare/Reuse in that subprocess so a +// blocked filesystem syscall can be terminated without leaving an in-process +// goroutine that may resume writing after the task has already been retried. +const PreparationHelperArg = "__multica_execenv_prepare" + +const ( + preparationActionPrepare = "prepare" + preparationActionReuse = "reuse" + preparationWaitDelay = 2 * time.Second +) + +type preparationRequest struct { + Action string `json:"action"` + Prepare *PrepareParams `json:"prepare,omitempty"` + Reuse *ReuseParams `json:"reuse,omitempty"` +} + +type preparationResponse struct { + Environment *Environment `json:"environment,omitempty"` + Error string `json:"error,omitempty"` +} + +// PrepareIsolated executes Prepare in a killable helper process. command must +// name the current multica binary followed by PreparationHelperArg in +// production; accepting a slice also lets tests use the Go test binary as the +// helper without installing a CLI binary. +func PrepareIsolated(ctx context.Context, command []string, params PrepareParams, logger *slog.Logger) (*Environment, error) { + return runPreparationProcess(ctx, command, preparationRequest{ + Action: preparationActionPrepare, + Prepare: ¶ms, + }, logger) +} + +// ReuseIsolated executes Reuse in the same killable helper process contract as +// PrepareIsolated. +func ReuseIsolated(ctx context.Context, command []string, params ReuseParams, logger *slog.Logger) (*Environment, error) { + return runPreparationProcess(ctx, command, preparationRequest{ + Action: preparationActionReuse, + Reuse: ¶ms, + }, logger) +} + +func runPreparationProcess(ctx context.Context, command []string, request preparationRequest, logger *slog.Logger) (*Environment, error) { + if len(command) == 0 || strings.TrimSpace(command[0]) == "" { + return nil, errors.New("execenv: preparation helper command is empty") + } + if ctx.Err() != nil { + return nil, context.Cause(ctx) + } + payload, err := json.Marshal(request) + if err != nil { + return nil, fmt.Errorf("execenv: encode preparation request: %w", err) + } + + cmd := exec.Command(command[0], command[1:]...) + controller, err := newPreparationProcessController(cmd) + if err != nil { + return nil, fmt.Errorf("execenv: create preparation process controller: %w", err) + } + defer controller.close() + cmd.WaitDelay = preparationWaitDelay + stdin, err := cmd.StdinPipe() + if err != nil { + return nil, fmt.Errorf("execenv: create preparation stdin: %w", err) + } + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + + if err := cmd.Start(); err != nil { + stdin.Close() + return nil, fmt.Errorf("execenv: start preparation helper: %w", err) + } + // The helper blocks decoding stdin. Attach it to the platform's process-tree + // boundary before releasing the finite request payload, so it cannot spawn a + // descendant in the gap between Start and ownership setup. + if err := controller.attach(cmd); err != nil { + stdin.Close() + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = controller.finish() + return nil, fmt.Errorf("execenv: attach preparation helper process: %w", err) + } + + writeDone := make(chan error, 1) + go func() { + _, writeErr := stdin.Write(payload) + closeErr := stdin.Close() + writeDone <- errors.Join(writeErr, closeErr) + }() + waitDone := make(chan error, 1) + go func() { waitDone <- cmd.Wait() }() + + var stopErr error + select { + case err = <-waitDone: + case <-ctx.Done(): + stopErr = controller.stop(cmd) + err = <-waitDone + } + writeErr := <-writeDone + finishErr := controller.finish() + if lifecycleErr := errors.Join(stopErr, finishErr); lifecycleErr != nil { + return nil, fmt.Errorf("execenv: stop preparation process tree: %w", lifecycleErr) + } + // The context cause is the daemon's stable failure contract. Prefer it over + // the platform-specific process exit text ("signal: killed", exit 1, ...). + if ctx.Err() != nil { + return nil, context.Cause(ctx) + } + if err != nil { + if detail := strings.TrimSpace(stderr.String()); detail != "" { + return nil, fmt.Errorf("execenv: preparation helper failed: %w: %s", err, detail) + } + return nil, fmt.Errorf("execenv: preparation helper failed: %w", err) + } + if writeErr != nil { + return nil, fmt.Errorf("execenv: write preparation request: %w", writeErr) + } + + var response preparationResponse + if err := json.Unmarshal(stdout.Bytes(), &response); err != nil { + return nil, fmt.Errorf("execenv: decode preparation response: %w", err) + } + if response.Error != "" { + return nil, errors.New(response.Error) + } + if response.Environment != nil { + // logger is intentionally omitted from JSON. Reattach the owning daemon's + // logger so later cleanup retains its normal diagnostics. + response.Environment.logger = logger + } + return response.Environment, nil +} + +// RunPreparationHelper serves the private helper protocol on stdin/stdout. +// Operational errors from Prepare are encoded in the response so the parent +// can preserve them; malformed protocol input/output is returned as a process +// error because the parent cannot safely interpret the result. +func RunPreparationHelper(in io.Reader, out io.Writer, logger *slog.Logger) error { + var request preparationRequest + decoder := json.NewDecoder(in) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&request); err != nil { + return fmt.Errorf("decode preparation request: %w", err) + } + + var response preparationResponse + switch request.Action { + case preparationActionPrepare: + if request.Prepare == nil || request.Reuse != nil { + return errors.New("invalid prepare request") + } + env, err := Prepare(*request.Prepare, logger) + response.Environment = env + if err != nil { + response.Error = err.Error() + } + case preparationActionReuse: + if request.Reuse == nil || request.Prepare != nil { + return errors.New("invalid reuse request") + } + response.Environment = Reuse(*request.Reuse, logger) + default: + return fmt.Errorf("unknown preparation action %q", request.Action) + } + + if err := json.NewEncoder(out).Encode(response); err != nil { + return fmt.Errorf("encode preparation response: %w", err) + } + return nil +} diff --git a/server/internal/daemon/execenv/isolation_test.go b/server/internal/daemon/execenv/isolation_test.go new file mode 100644 index 000000000..fc2d3be71 --- /dev/null +++ b/server/internal/daemon/execenv/isolation_test.go @@ -0,0 +1,64 @@ +package execenv + +import ( + "context" + "io" + "log/slog" + "os" + "testing" + "time" +) + +const preparationHelperTestMode = "execenv-preparation-helper" + +func preparationHelperTestCommand() []string { + return []string{ + os.Args[0], + "-test.run=^TestPreparationHelperProcess$", + "--", + preparationHelperTestMode, + } +} + +// TestPreparationHelperProcess is both a no-op parent-side test and the child +// entry point used by isolation tests. Keeping it in the package test binary +// exercises the same stdin/stdout protocol as the real multica helper. +func TestPreparationHelperProcess(t *testing.T) { + if len(os.Args) == 0 || os.Args[len(os.Args)-1] != preparationHelperTestMode { + return + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + if err := RunPreparationHelper(os.Stdin, os.Stdout, logger); err != nil { + os.Exit(2) + } + os.Exit(0) +} + +func TestPreparationHelperRoundTripsReuse(t *testing.T) { + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + params := PrepareParams{ + WorkspacesRoot: t.TempDir(), + WorkspaceID: "ws-helper-reuse", + TaskID: "99999999-8888-7777-6666-555555555555", + Provider: "claude", + Task: TaskContextForEnv{IssueID: "issue-helper-reuse"}, + } + env, err := PrepareIsolated(ctx, preparationHelperTestCommand(), params, logger) + if err != nil { + t.Fatalf("PrepareIsolated: %v", err) + } + reused, err := ReuseIsolated(ctx, preparationHelperTestCommand(), ReuseParams{ + WorkspacesRoot: params.WorkspacesRoot, + WorkDir: env.WorkDir, + Provider: params.Provider, + Task: TaskContextForEnv{IssueID: "issue-helper-reuse", NewCommentCount: 1}, + }, logger) + if err != nil { + t.Fatalf("ReuseIsolated: %v", err) + } + if reused == nil || reused.RootDir != env.RootDir || reused.WorkDir != env.WorkDir { + t.Fatalf("reused environment = %#v, want root %q workdir %q", reused, env.RootDir, env.WorkDir) + } +} diff --git a/server/internal/daemon/execenv/isolation_unix.go b/server/internal/daemon/execenv/isolation_unix.go new file mode 100644 index 000000000..fa87dc2bc --- /dev/null +++ b/server/internal/daemon/execenv/isolation_unix.go @@ -0,0 +1,39 @@ +//go:build !windows + +package execenv + +import ( + "errors" + "os" + "os/exec" + "syscall" +) + +type preparationProcessController struct{} + +func newPreparationProcessController(cmd *exec.Cmd) (*preparationProcessController, error) { + if cmd.SysProcAttr == nil { + cmd.SysProcAttr = &syscall.SysProcAttr{} + } + cmd.SysProcAttr.Setpgid = true + return &preparationProcessController{}, nil +} + +func (*preparationProcessController) attach(_ *exec.Cmd) error { return nil } + +func (*preparationProcessController) stop(cmd *exec.Cmd) error { + if cmd.Process == nil { + return os.ErrProcessDone + } + // Kill the helper and any CLI it spawned. After SIGKILL is pending, a + // helper blocked in a kernel filesystem call cannot return to Go and + // perform another write when that call eventually unblocks. + err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL) + if errors.Is(err, syscall.ESRCH) { + return nil + } + return err +} + +func (*preparationProcessController) finish() error { return nil } +func (*preparationProcessController) close() {} diff --git a/server/internal/daemon/execenv/isolation_unix_test.go b/server/internal/daemon/execenv/isolation_unix_test.go new file mode 100644 index 000000000..e4fb04af8 --- /dev/null +++ b/server/internal/daemon/execenv/isolation_unix_test.go @@ -0,0 +1,85 @@ +//go:build !windows + +package execenv + +import ( + "context" + "errors" + "io" + "log/slog" + "os" + "path/filepath" + "syscall" + "testing" + "time" +) + +// TestPrepareIsolated_PermanentFIFOBlockThenImmediateRetry is the lifecycle +// regression for MUL-4923. The first real Codex Prepare blocks forever opening +// a shared config FIFO. After the deadline, there must be no reader left on +// that FIFO before an immediate retry mutates the same env root. +func TestPrepareIsolated_PermanentFIFOBlockThenImmediateRetry(t *testing.T) { + sharedCodexHome := t.TempDir() + t.Setenv("CODEX_HOME", sharedCodexHome) + fifo := filepath.Join(sharedCodexHome, "config.json") + if err := syscall.Mkfifo(fifo, 0o600); err != nil { + t.Fatalf("create config FIFO: %v", err) + } + + params := PrepareParams{ + WorkspacesRoot: t.TempDir(), + WorkspaceID: "ws-isolated-prepare", + TaskID: "11111111-2222-3333-4444-555555555555", + AgentName: "isolation-test", + Provider: "codex", + Task: TaskContextForEnv{ + IssueID: "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + AgentID: "ffffffff-1111-2222-3333-444444444444", + }, + } + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + startedAt := time.Now() + _, err := PrepareIsolated(ctx, preparationHelperTestCommand(), params, logger) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("PrepareIsolated error = %v, want deadline exceeded", err) + } + if elapsed := time.Since(startedAt); elapsed > 3*time.Second { + t.Fatalf("PrepareIsolated took %s, want a bounded process termination", elapsed) + } + if _, err := os.Stat(PredictRootDir(params.WorkspacesRoot, params.WorkspaceID, params.TaskID)); err != nil { + t.Fatalf("first helper did not reach environment preparation before timeout: %v", err) + } + + // A non-blocking FIFO writer succeeds only while a reader is still alive. + // ENXIO therefore proves the timed-out helper was reaped before ownership + // returned to the caller; there is no old worker that can resume later. + writer, writerErr := os.OpenFile(fifo, os.O_WRONLY|syscall.O_NONBLOCK, 0) + if writerErr == nil { + writer.Close() + t.Fatal("timed-out preparation helper still has the FIFO open for reading") + } + if !errors.Is(writerErr, syscall.ENXIO) { + t.Fatalf("probe FIFO reader: %v, want ENXIO", writerErr) + } + + // Retry immediately against the same root. It must be the sole writer and + // complete successfully after the pathological source is replaced. + if err := os.Remove(fifo); err != nil { + t.Fatalf("remove config FIFO: %v", err) + } + if err := os.WriteFile(fifo, []byte("{}\n"), 0o600); err != nil { + t.Fatalf("replace config FIFO: %v", err) + } + retryCtx, retryCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer retryCancel() + env, err := PrepareIsolated(retryCtx, preparationHelperTestCommand(), params, logger) + if err != nil { + t.Fatalf("immediate retry PrepareIsolated: %v", err) + } + if env == nil || env.RootDir != PredictRootDir(params.WorkspacesRoot, params.WorkspaceID, params.TaskID) { + t.Fatalf("retry environment = %#v, want the predicted root", env) + } +} diff --git a/server/internal/daemon/execenv/isolation_windows.go b/server/internal/daemon/execenv/isolation_windows.go new file mode 100644 index 000000000..c95b24f50 --- /dev/null +++ b/server/internal/daemon/execenv/isolation_windows.go @@ -0,0 +1,123 @@ +//go:build windows + +package execenv + +import ( + "fmt" + "os/exec" + "time" + "unsafe" + + "golang.org/x/sys/windows" +) + +type preparationProcessController struct { + job windows.Handle +} + +// jobObjectBasicAccountingInformation mirrors JOBOBJECT_BASIC_ACCOUNTING_INFORMATION. +// x/sys exposes QueryInformationJobObject and the information-class constant, +// but not this result struct. +type jobObjectBasicAccountingInformation struct { + TotalUserTime int64 + TotalKernelTime int64 + ThisPeriodTotalUserTime int64 + ThisPeriodTotalKernelTime int64 + TotalPageFaultCount uint32 + TotalProcesses uint32 + ActiveProcesses uint32 + TotalTerminatedProcesses uint32 +} + +func newPreparationProcessController(_ *exec.Cmd) (*preparationProcessController, error) { + job, err := windows.CreateJobObject(nil, nil) + if err != nil { + return nil, err + } + info := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{} + info.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE + if _, err := windows.SetInformationJobObject( + job, + windows.JobObjectExtendedLimitInformation, + uintptr(unsafe.Pointer(&info)), + uint32(unsafe.Sizeof(info)), + ); err != nil { + windows.CloseHandle(job) + return nil, fmt.Errorf("set KILL_ON_JOB_CLOSE: %w", err) + } + return &preparationProcessController{job: job}, nil +} + +func (c *preparationProcessController) attach(cmd *exec.Cmd) error { + process, err := windows.OpenProcess( + windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE, + false, + uint32(cmd.Process.Pid), + ) + if err != nil { + return fmt.Errorf("open helper process: %w", err) + } + defer windows.CloseHandle(process) + if err := windows.AssignProcessToJobObject(c.job, process); err != nil { + return fmt.Errorf("assign helper to job object: %w", err) + } + return nil +} + +func (c *preparationProcessController) stop(_ *exec.Cmd) error { + if err := windows.TerminateJobObject(c.job, 1); err != nil { + active, queryErr := c.activeProcesses() + if queryErr == nil && active == 0 { + return nil + } + return err + } + return nil +} + +// finish is called only after the direct helper has exited. Terminate any +// descendant it left behind, then wait until the Job Object reports no active +// members before the daemon may release directory ownership and retry. +func (c *preparationProcessController) finish() error { + active, err := c.activeProcesses() + if err != nil { + return err + } + if active > 0 { + if err := windows.TerminateJobObject(c.job, 1); err != nil { + return err + } + } + for { + active, err = c.activeProcesses() + if err != nil { + return err + } + if active == 0 { + return nil + } + time.Sleep(10 * time.Millisecond) + } +} + +func (c *preparationProcessController) activeProcesses() (uint32, error) { + var info jobObjectBasicAccountingInformation + if err := windows.QueryInformationJobObject( + c.job, + windows.JobObjectBasicAccountingInformation, + uintptr(unsafe.Pointer(&info)), + uint32(unsafe.Sizeof(info)), + nil, + ); err != nil { + return 0, fmt.Errorf("query job object members: %w", err) + } + return info.ActiveProcesses, nil +} + +func (c *preparationProcessController) close() { + if c.job == 0 { + return + } + _ = windows.CloseHandle(c.job) + c.job = 0 +} diff --git a/server/internal/daemon/execenv/isolation_windows_test.go b/server/internal/daemon/execenv/isolation_windows_test.go new file mode 100644 index 000000000..f3396816a --- /dev/null +++ b/server/internal/daemon/execenv/isolation_windows_test.go @@ -0,0 +1,131 @@ +//go:build windows + +package execenv + +import ( + "context" + "encoding/json" + "errors" + "os" + "os/exec" + "path/filepath" + "testing" + "time" +) + +const ( + windowsPreparationTreeHelperMode = "windows-preparation-tree-helper" + windowsPreparationDelayedMode = "windows-preparation-delayed-writer" +) + +func testArgsAfterDoubleDash() []string { + for i, arg := range os.Args { + if arg == "--" && i+1 < len(os.Args) { + return os.Args[i+1:] + } + } + return nil +} + +// TestWindowsPreparationJobHelperProcess starts a deliberately orphaned child +// and then blocks forever. The parent test cancels this helper and verifies the +// Job Object kills the delayed child before it can write. +func TestWindowsPreparationJobHelperProcess(t *testing.T) { + args := testArgsAfterDoubleDash() + if len(args) != 3 || args[0] != windowsPreparationTreeHelperMode { + return + } + var request preparationRequest + if err := json.NewDecoder(os.Stdin).Decode(&request); err != nil { + os.Exit(2) + } + marker, ready := args[1], args[2] + child := exec.Command( + os.Args[0], + "-test.run=^TestWindowsPreparationDelayedWriter$", + "--", + windowsPreparationDelayedMode, + marker, + ) + if err := child.Start(); err != nil { + os.Exit(3) + } + if err := os.WriteFile(ready, []byte("ready"), 0o600); err != nil { + os.Exit(4) + } + select {} +} + +func TestWindowsPreparationDelayedWriter(t *testing.T) { + args := testArgsAfterDoubleDash() + if len(args) != 2 || args[0] != windowsPreparationDelayedMode { + return + } + time.Sleep(2 * time.Second) + if err := os.WriteFile(args[1], []byte("leaked"), 0o600); err != nil { + os.Exit(2) + } + os.Exit(0) +} + +func TestPrepareIsolated_WindowsKillsDescendantBeforeRetry(t *testing.T) { + dir := t.TempDir() + marker := filepath.Join(dir, "old-attempt-marker") + ready := filepath.Join(dir, "descendant-ready") + command := []string{ + os.Args[0], + "-test.run=^TestWindowsPreparationJobHelperProcess$", + "--", + windowsPreparationTreeHelperMode, + marker, + ready, + } + params := PrepareParams{ + WorkspacesRoot: filepath.Join(dir, "workspaces"), + WorkspaceID: "ws-windows-job", + TaskID: "12345678-1111-2222-3333-444444444444", + Provider: "claude", + Task: TaskContextForEnv{IssueID: "issue-windows-job"}, + } + + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { + _, err := PrepareIsolated(ctx, command, params, nil) + result <- err + }() + deadline := time.Now().Add(10 * time.Second) + for { + if _, err := os.Stat(ready); err == nil { + break + } + if time.Now().After(deadline) { + cancel() + <-result + t.Fatal("helper did not start its delayed descendant") + } + time.Sleep(10 * time.Millisecond) + } + cancel() + if err := <-result; !errors.Is(err, context.Canceled) { + t.Fatalf("PrepareIsolated error = %v, want context canceled", err) + } + + // The child would write after two seconds if only the direct helper were + // killed. Waiting beyond that point makes the absence of the marker a + // runtime proof that the Job Object terminated the whole tree. + time.Sleep(2500 * time.Millisecond) + if _, err := os.Stat(marker); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("timed-out helper descendant survived and wrote marker: %v", err) + } + + retryCtx, retryCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer retryCancel() + env, err := PrepareIsolated(retryCtx, preparationHelperTestCommand(), params, nil) + if err != nil { + t.Fatalf("immediate retry PrepareIsolated: %v", err) + } + if env == nil || env.RootDir != PredictRootDir(params.WorkspacesRoot, params.WorkspaceID, params.TaskID) { + t.Fatalf("retry environment = %#v, want the predicted root", env) + } +} diff --git a/server/internal/daemon/workdir_race_test.go b/server/internal/daemon/workdir_race_test.go index 3e07b56fd..def8fa3fd 100644 --- a/server/internal/daemon/workdir_race_test.go +++ b/server/internal/daemon/workdir_race_test.go @@ -2,6 +2,7 @@ package daemon import ( "context" + "errors" "io" "log/slog" "net/http" @@ -346,6 +347,92 @@ func TestRunTask_ExtendsPrepareLeaseDuringStartTask(t *testing.T) { } } +func TestRunTask_PrepareTimeoutStopsLeaseDuringBlockedStartTask(t *testing.T) { + oldRefresh := taskPrepareLeaseRefresh + oldTimeout := taskPrepareLeaseTimeout + taskPrepareLeaseRefresh = 10 * time.Millisecond + taskPrepareLeaseTimeout = 500 * time.Millisecond + t.Cleanup(func() { + taskPrepareLeaseRefresh = oldRefresh + taskPrepareLeaseTimeout = oldTimeout + }) + + var leaseCalls atomic.Int64 + startEntered := make(chan struct{}) + var closeStartOnce sync.Once + releaseStart := make(chan struct{}) + var releaseStartOnce sync.Once + t.Cleanup(func() { releaseStartOnce.Do(func() { close(releaseStart) }) }) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasSuffix(r.URL.Path, "/prepare-lease"): + leaseCalls.Add(1) + w.WriteHeader(http.StatusOK) + case strings.HasSuffix(r.URL.Path, "/start"): + closeStartOnce.Do(func() { close(startEntered) }) + <-releaseStart + w.WriteHeader(http.StatusOK) + default: + w.WriteHeader(http.StatusOK) + } + })) + t.Cleanup(srv.Close) + + workspacesRoot := t.TempDir() + fakeBin := filepath.Join(t.TempDir(), "claude") + if err := os.WriteFile(fakeBin, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil { + t.Fatalf("write fake agent: %v", err) + } + d := &Daemon{ + client: NewClient(srv.URL), + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + workspaces: make(map[string]*workspaceState), + runtimeIndex: map[string]Runtime{"rt-1": {ID: "rt-1", Provider: "claude"}}, + activeEnvRoots: make(map[string]int), + taskPrepareTimeout: 150 * time.Millisecond, + cfg: Config{ + WorkspacesRoot: workspacesRoot, + Agents: map[string]AgentEntry{ + "claude": {Path: fakeBin}, + }, + }, + } + + task := Task{ + ID: "task-runtask-start-timeout", + WorkspaceID: "ws-runtask-start-timeout", + RuntimeID: "rt-1", + IssueID: "issue-runtask-start-timeout", + Agent: &AgentData{Name: "test-agent"}, + } + taskLog := slog.New(slog.NewTextHandler(io.Discard, nil)) + startedAt := time.Now() + _, err := d.runTask(context.Background(), task, "claude", 0, taskLog) + if !errors.Is(err, errTaskPrepareTimeout) { + t.Fatalf("runTask error = %v, want task prepare timeout", err) + } + if elapsed := time.Since(startedAt); elapsed > time.Second { + t.Fatalf("runTask took %s, want prepare deadline to stop blocked /start", elapsed) + } + select { + case <-startEntered: + default: + t.Fatal("runTask did not reach /start") + } + releaseStartOnce.Do(func() { close(releaseStart) }) + if got := leaseCalls.Load(); got == 0 { + t.Fatal("prepare lease was never extended while /start was blocked") + } + leaseCallsAtReturn := leaseCalls.Load() + time.Sleep(4 * taskPrepareLeaseRefresh) + if got := leaseCalls.Load(); got != leaseCallsAtReturn { + t.Fatalf("prepare lease kept extending after timeout: calls %d -> %d", leaseCallsAtReturn, got) + } + if got := taskRunFailureReason(err); got != "timeout" { + t.Fatalf("taskRunFailureReason = %q, want retryable platform timeout", got) + } +} + // TestHandleTask_KeepsEnvRootActiveAcrossCompletion is the regression guard // for issue #3999 race B. After runner.run returns, the in-process active // guard installed inside runTask (defer unmarkActiveEnvRoot at the diff --git a/server/internal/handler/issue_child_done_test.go b/server/internal/handler/issue_child_done_test.go index d3949964a..36958421b 100644 --- a/server/internal/handler/issue_child_done_test.go +++ b/server/internal/handler/issue_child_done_test.go @@ -554,3 +554,124 @@ func TestChildDoneWakesLeaderWhenChildIsSameSquad(t *testing.T) { t.Errorf("expected 1 pending leader task for same-squad child (MUL-3969), got %d", got) } } + +// TestStageLeaderPrepareTimeoutRetryCanAdvanceNextStage covers the full server +// half of MUL-4923's recovery chain: a stage barrier wakes the squad leader, +// the pre-start attempt fails with the daemon's timeout reason, the atomic +// retry preserves leader/squad/trigger provenance, and that retry can promote +// the parked next stage as the leader actor. +func TestStageLeaderPrepareTimeoutRetryCanAdvanceNextStage(t *testing.T) { + if testHandler == nil || testPool == nil { + t.Skip("database not available") + } + ctx := context.Background() + fx := newChildDoneFixture(t, "in_progress") + sq := newSquadCommentTriggerFixture(t) + setIssueAssigneeDirect(t, fx.parent.ID, "squad", sq.SquadID) + setIssueAssigneeDirect(t, fx.child.ID, "squad", sq.SquadID) + if _, err := testPool.Exec(ctx, `UPDATE issue SET stage = 1 WHERE id = $1`, fx.child.ID); err != nil { + t.Fatalf("set stage 1: %v", err) + } + + // Stage 2 exists but is deliberately parked. The server wakes the leader at + // the Stage 1 barrier; only the leader decides to promote this child. + w := httptest.NewRecorder() + req := newRequest("POST", "/api/issues?workspace_id="+testWorkspaceID, map[string]any{ + "title": "stage 2 after prepare timeout", + "status": "backlog", + "parent_issue_id": fx.parent.ID, + "stage": 2, + "assignee_type": "squad", + "assignee_id": sq.SquadID, + }) + testHandler.CreateIssue(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("create stage 2: expected 201, got %d: %s", w.Code, w.Body.String()) + } + var stage2 IssueResponse + if err := json.NewDecoder(w.Body).Decode(&stage2); err != nil { + t.Fatalf("decode stage 2: %v", err) + } + t.Cleanup(func() { + testPool.Exec(context.Background(), `DELETE FROM agent_task_queue WHERE issue_id IN ($1, $2)`, fx.parent.ID, stage2.ID) + testPool.Exec(context.Background(), `DELETE FROM issue WHERE id = $1`, stage2.ID) + }) + + updateChildStatus(t, fx.child.ID, "done") + content := parentSystemCommentContent(t, fx.parent.ID) + if !strings.Contains(content, "Stage 2 is next") { + t.Fatalf("stage barrier comment does not identify Stage 2: %s", content) + } + + var originalID, originalSquadID, originalTriggerID string + var originalLeader bool + if err := testPool.QueryRow(ctx, ` + SELECT id::text, is_leader_task, squad_id::text, trigger_comment_id::text + FROM agent_task_queue + WHERE issue_id = $1 AND agent_id = $2 AND status = 'queued' + ORDER BY created_at DESC + LIMIT 1 + `, fx.parent.ID, sq.LeaderID).Scan(&originalID, &originalLeader, &originalSquadID, &originalTriggerID); err != nil { + t.Fatalf("load Stage 1 leader wake: %v", err) + } + if !originalLeader || originalSquadID != sq.SquadID || originalTriggerID == "" { + t.Fatalf("leader wake provenance = leader:%v squad:%q trigger:%q", originalLeader, originalSquadID, originalTriggerID) + } + if _, err := testPool.Exec(ctx, ` + UPDATE agent_task_queue + SET status = 'dispatched', dispatched_at = now() + WHERE id = $1 + `, originalID); err != nil { + t.Fatalf("dispatch original leader task: %v", err) + } + + if _, err := testHandler.TaskService.FailTask(ctx, parseUUID(originalID), "task preparation timed out after 5m0s", "", "", "timeout"); err != nil { + t.Fatalf("fail original leader task: %v", err) + } + + var retryID, retryStatus, retrySquadID, retryTriggerID string + var retryLeader bool + var retryAttempt int32 + if err := testPool.QueryRow(ctx, ` + SELECT id::text, status, is_leader_task, squad_id::text, + trigger_comment_id::text, attempt + FROM agent_task_queue + WHERE parent_task_id = $1 + `, originalID).Scan(&retryID, &retryStatus, &retryLeader, &retrySquadID, &retryTriggerID, &retryAttempt); err != nil { + t.Fatalf("load automatic retry: %v", err) + } + if retryStatus != "queued" || retryAttempt != 2 || !retryLeader || retrySquadID != originalSquadID || retryTriggerID != originalTriggerID { + t.Fatalf("retry provenance = status:%q attempt:%d leader:%v squad:%q trigger:%q; want queued attempt 2 with original leader context", + retryStatus, retryAttempt, retryLeader, retrySquadID, retryTriggerID) + } + if _, err := testPool.Exec(ctx, ` + UPDATE agent_task_queue + SET status = 'running', dispatched_at = now(), started_at = now() + WHERE id = $1 + `, retryID); err != nil { + t.Fatalf("start retry leader task: %v", err) + } + + // Act through the normal issue handler with the retry task as the agent + // identity. This is the operation the Stage handoff prompt asks the leader + // to perform, and proves the retry was not demoted to a generic worker. + w = httptest.NewRecorder() + req = newRequest("PUT", "/api/issues/"+stage2.ID, map[string]any{"status": "todo"}) + req.Header.Set("X-Agent-ID", sq.LeaderID) + req.Header.Set("X-Task-ID", retryID) + req = withURLParam(req, "id", stage2.ID) + testHandler.UpdateIssue(w, req) + if w.Code != http.StatusOK { + t.Fatalf("retry leader promote Stage 2: expected 200, got %d: %s", w.Code, w.Body.String()) + } + var stage2Status string + if err := testPool.QueryRow(ctx, `SELECT status FROM issue WHERE id = $1`, stage2.ID).Scan(&stage2Status); err != nil { + t.Fatalf("load promoted Stage 2: %v", err) + } + if stage2Status != "todo" { + t.Fatalf("Stage 2 status = %q, want todo", stage2Status) + } + if got := countPendingTasksForAgent(t, stage2.ID, sq.LeaderID); got != 1 { + t.Fatalf("promoted Stage 2 queued %d leader tasks, want 1", got) + } +}