From 09dce598dfef5f24b35c11222b593c3937842f62 Mon Sep 17 00:00:00 2001 From: milymarkovic Date: Thu, 23 Jul 2026 10:09:50 +0300 Subject: [PATCH] MUL-5155: KAP-1051: diagnose Codex thread/start timeouts fail-closed (#5759) * fix(agent/codex): diagnose thread start timeouts (KAP-1051) Co-authored-by: multica-agent * fix(agent/codex): validate thread ID before success lifecycle Co-authored-by: multica-agent * fix(agent/codex): confirm process tree cleanup Co-authored-by: multica-agent * fix(agent/codex): bound Windows pipe cleanup Co-authored-by: multica-agent * ci: run bounded Codex cleanup test on Windows Co-authored-by: multica-agent * fix(agent): reap initialize timeout process groups * fix(agent): redact initialize timeout stderr * fix(agent): redact initialize context failures --------- Co-authored-by: multica-agent --- .github/workflows/ci.yml | 7 + server/pkg/agent/codex.go | 160 ++++++++++- server/pkg/agent/codex_cleanup_unix_test.go | 218 ++++++++++++++ server/pkg/agent/codex_test.go | 303 +++++++++++++++++++- server/pkg/agent/proc_other.go | 19 ++ server/pkg/agent/proc_windows.go | 3 + server/pkg/agent/proc_windows_test.go | 106 +++++++ 7 files changed, 793 insertions(+), 23 deletions(-) create mode 100644 server/pkg/agent/codex_cleanup_unix_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1a0a74a29..e47d39eb8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -212,6 +212,13 @@ jobs: # that stops matching) is visible in the log instead of passing as "ok". run: go test ./pkg/agent -v -run '^(TestCursorExecutePromptSurvivesPowerShellShim|TestPlatformCursorInvocation|TestPlatformCopilotInvocation|TestPlatformPiInvocation)' -count=1 -timeout=5m + - name: Test bounded Codex cleanup with inherited stdout descendant + working-directory: server + # Windows cannot prove whole-tree termination without a Job Object, + # but a descendant holding inherited stdout must never keep Result + # blocked forever. -v makes RUN/PASS evidence explicit in CI logs. + run: go test ./pkg/agent -v -run '^TestCodexWindowsInheritedStdoutDescendantCleanupIsBounded$' -count=1 -timeout=5m + - name: Build Windows CLI helper entrypoint working-directory: server run: go build ./cmd/multica diff --git a/server/pkg/agent/codex.go b/server/pkg/agent/codex.go index ef7f3eb74..8492aa95e 100644 --- a/server/pkg/agent/codex.go +++ b/server/pkg/agent/codex.go @@ -59,6 +59,7 @@ const ( // instead of burning two full grace windows per cleanup phase. Mirrors // the opencodeTerminateGraceNanos hook. var codexGracefulShutdownTimeoutNanos atomic.Int64 +var codexProcessWaitDelayNanos atomic.Int64 var activeCodexLaunches atomic.Int64 var maxActiveCodexLaunchesObserved atomic.Int64 var codexCleanupConfirmationOverride atomic.Int32 @@ -81,6 +82,39 @@ func codexGracefulShutdown() time.Duration { return codexGracefulShutdownTimeout } +func codexProcessWaitDelay() time.Duration { + if n := codexProcessWaitDelayNanos.Load(); n > 0 { + return time.Duration(n) + } + return 10 * time.Second +} + +type codexStderrClassification struct { + modelRefreshTimeout int + mcpInitTransport int + bareTimeout int +} + +// classifyCodexStartupStderr emits bounded counters only. It deliberately does +// not make retry decisions: stderr is sibling diagnostic evidence, not proof +// that thread/start did or did not create a provider-side thread. +func classifyCodexStartupStderr(stderr string, timedOut bool) codexStderrClassification { + lower := strings.ToLower(sanitizeCodexDiagnostic(stderr)) + classification := codexStderrClassification{ + modelRefreshTimeout: strings.Count(lower, codexModelCatalogRefreshTimeoutSignal), + } + for _, line := range strings.Split(lower, "\n") { + if strings.Contains(line, "mcp") && strings.Contains(line, "transport") && + (strings.Contains(line, "error") || strings.Contains(line, "failed") || strings.Contains(line, "closed")) { + classification.mcpInitTransport++ + } + } + if timedOut && classification.modelRefreshTimeout == 0 && classification.mcpInitTransport == 0 { + classification.bareTimeout = 1 + } + return classification +} + // CodexSemanticInactivityMarker prefixes timeout errors emitted when Codex // stops making semantic progress while the process is still alive. const CodexSemanticInactivityMarker = "codex semantic inactivity timeout" @@ -100,6 +134,7 @@ const CodexHandshakeTimeoutMarker = "codex app-server handshake timeout" // prefix rather than one variant: they are all the same startup-blocking // failure from the daemon's point of view. const codexModelCatalogRefreshFailureSignal = "failed to refresh available models" +const codexModelCatalogRefreshTimeoutSignal = "failed to refresh available models: timeout waiting for child process to exit" var errCodexProcessExited = errors.New("codex process exited") @@ -808,7 +843,7 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec // Bound the wait after the context is cancelled so a stuck child (or an // open pipe held by a grandchild) can't hang cmd.Wait() forever. Matches // the other long-lived backends (claude, copilot, cursor, …). - cmd.WaitDelay = 10 * time.Second + cmd.WaitDelay = codexProcessWaitDelay() b.cfg.Logger.Info("agent command", "exec", execPath, "args", codexArgs) if opts.Cwd != "" { cmd.Dir = opts.Cwd @@ -868,6 +903,9 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec pending: make(map[int]*pendingRPC), processDone: make(chan struct{}), handshakeTimeout: handshakeTimeout, + pid: cmd.Process.Pid, + attempt: attempt, + activeLaunches: activeLaunches, notificationProtocol: "unknown", acceptNotification: turnNotificationGate.accept, onDiscardedNotification: func(string, map[string]any) { @@ -973,6 +1011,16 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec stdin.Close() grace := codexGracefulShutdown() + waitCh := make(chan struct{}) + var startWait sync.Once + startProcessWait := func() { + startWait.Do(func() { + go func() { + cleanupWaitErr = cmd.Wait() + close(waitCh) + }() + }) + } // Phase 1: let the reader finish before invoking cmd.Wait(). select { @@ -990,17 +1038,26 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec "grace", grace.String(), ) cancel() - <-readerDone + // On Windows, Cancel terminates only the direct child. A + // descendant may keep inherited stdout open indefinitely. Start + // Wait now so os/exec's WaitDelay closes the pipe after its + // bounded deadline and lets the reader finish. + startProcessWait() + <-waitCh + select { + case <-readerDone: + case <-time.After(grace): + b.cfg.Logger.Warn("codex stdout reader remained open after bounded process wait", + "pid", cmd.Process.Pid, + "grace", grace.String(), + ) + } } // Phase 2: bound cmd.Wait() in case the process is still alive // (scanner-overflow case: reader exited early on its own while // codex stayed blocked writing into a full stdout pipe). - waitCh := make(chan struct{}) - go func() { - cleanupWaitErr = cmd.Wait() - close(waitCh) - }() + startProcessWait() select { case <-waitCh: waitReturned = true @@ -1021,7 +1078,7 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec // Wait returning with a ProcessState is the os/exec reap boundary. // On Unix, ProcessState.Exited reports false for a process terminated // by SIGKILL even though Wait successfully reaped it. - cleanupConfirmed = waitReturned && cmd.ProcessState != nil + cleanupConfirmed = waitReturned && cmd.ProcessState != nil && waitProcessGroupGone(cmd.Process, grace) if codexCleanupConfirmationOverride.Load() < 0 { cleanupConfirmed = false } @@ -1038,7 +1095,6 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec "wait_error", cleanupWaitErr, "stderr_bytes", stderrBuf.TotalBytes(), "stderr_truncated", stderrBuf.TotalBytes() > codexStderrTailBytes, - "stderr_tail", sanitizeCodexDiagnostic(stderrBuf.Tail()), ) }) } @@ -1073,14 +1129,31 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec }) if err != nil { initializeLatency := time.Since(initializeStarted) + var handshakeErr *codexHandshakeTimeoutError + timedOut := errors.As(err, &handshakeErr) && handshakeErr.Method == "initialize" + if timedOut { + // A timed-out initialize may still complete after the host gives up. + // Kill the whole process group before waiting so a leader that exits + // on stdin EOF cannot leave detached-stdio descendants behind. + signalProcessGroup(cmd.Process, syscall.SIGKILL) + } drainAndWait() // flush os/exec stderr goroutine before sampling Tail finalStatus = "failed" - finalError = withAgentStderr(fmt.Sprintf("codex initialize failed: %v", err), "codex", sanitizeCodexDiagnostic(stderrBuf.Tail())) - var handshakeErr *codexHandshakeTimeoutError - retrySafe := errors.As(err, &handshakeErr) && handshakeErr.Method == "initialize" && !semanticObserved.Load() && cleanupConfirmed && codexInitializeRetrySupported() - if errors.As(err, &handshakeErr) && handshakeErr.Method == "initialize" && !cleanupConfirmed { + finalError = fmt.Sprintf("codex initialize failed: %v", err) + contextEnded := errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) + if !timedOut && !contextEnded { + // Timeout stderr is untrusted provider output and may echo opaque + // Config.Env/auth values that pattern sanitization cannot identify. + // The same applies when the parent task deadline/cancellation wins + // the race against the per-RPC handshake timeout. + // Keep it out of persisted/user-visible Results; cleanup lifecycle + // still records bounded byte/truncation metadata. + finalError = withAgentStderr(finalError, "codex", sanitizeCodexDiagnostic(stderrBuf.Tail())) + } + retrySafe := timedOut && !semanticObserved.Load() && cleanupConfirmed && codexInitializeRetrySupported() + if timedOut && !cleanupConfirmed { finalError += "; retry suppressed: process cleanup/reap not confirmed" - } else if errors.As(err, &handshakeErr) && handshakeErr.Method == "initialize" && cleanupConfirmed && !codexInitializeRetrySupported() { + } else if timedOut && cleanupConfirmed && !codexInitializeRetrySupported() { finalError += "; retry suppressed: process-tree cleanup cannot be confirmed on this platform" } b.cfg.Logger.Warn("codex lifecycle", "phase", "initialize_failure", "task_id", b.cfg.TaskID, "runtime_id", b.cfg.RuntimeID, "pid", cmd.Process.Pid, "attempt", attempt, "latency", initializeLatency.Round(time.Millisecond).String(), "semantic_activity", semanticObserved.Load(), "cleanup_confirmed", cleanupConfirmed, "retry_safe", retrySafe) @@ -1095,9 +1168,39 @@ func (b *codexBackend) executeOnce(ctx context.Context, prompt string, opts Exec // back to a fresh thread so the task still makes progress. threadID, resumed, err := c.startOrResumeThread(runCtx, opts, b.cfg.Logger) if err != nil { + var handshakeErr *codexHandshakeTimeoutError + timedOut := errors.As(err, &handshakeErr) && handshakeErr.Method == "thread/start" + if timedOut { + // A timed-out thread/start has an uncertain provider outcome. Kill + // the whole process group before waiting so a leader that exits on + // EOF cannot leave detached-stdio descendants behind. + signalProcessGroup(cmd.Process, syscall.SIGKILL) + } drainAndWait() // flush os/exec stderr goroutine before sampling Tail finalStatus = "failed" - finalError = withAgentStderr(err.Error(), "codex", sanitizeCodexDiagnostic(stderrBuf.Tail())) + stderrTail := sanitizeCodexDiagnostic(stderrBuf.Tail()) + finalError = err.Error() + if c.threadStartSent { + classification := classifyCodexStartupStderr(stderrTail, timedOut) + b.cfg.Logger.Warn("codex lifecycle", + "phase", "thread_start_failure", + "task_id", b.cfg.TaskID, + "runtime_id", b.cfg.RuntimeID, + "pid", cmd.Process.Pid, + "attempt", attempt, + "active_launches", activeLaunches, + "method", "thread/start", + "latency", time.Since(c.threadStartStarted).Round(time.Millisecond).String(), + "latency_ms", time.Since(c.threadStartStarted).Milliseconds(), + "cleanup_confirmed", cleanupConfirmed, + "reaped", cleanupConfirmed, + "retry_safe", false, + "retry_attempted", false, + "stderr_model_refresh_timeout_count", classification.modelRefreshTimeout, + "stderr_mcp_init_transport_count", classification.mcpInitTransport, + "stderr_bare_timeout_count", classification.bareTimeout, + ) + } resCh <- Result{Status: finalStatus, Error: finalError, DurationMs: time.Since(startTime).Milliseconds()} return } @@ -1425,6 +1528,17 @@ func (c *codexClient) startOrResumeThread(ctx context.Context, opts ExecOptions, "persistExtendedHistory": true, } applyCodexReasoningEffort(startParams, opts.ThinkingLevel) + c.threadStartSent = true + c.threadStartStarted = time.Now() + logger.Info("codex lifecycle", + "phase", "thread_start_sent", + "task_id", c.cfg.TaskID, + "runtime_id", c.cfg.RuntimeID, + "pid", c.pid, + "attempt", c.attempt, + "active_launches", c.activeLaunches, + "method", "thread/start", + ) startResult, err := c.request(ctx, "thread/start", startParams) if err != nil { return "", false, fmt.Errorf("codex thread/start failed: %w", err) @@ -1433,6 +1547,17 @@ func (c *codexClient) startOrResumeThread(ctx context.Context, opts ExecOptions, if threadID == "" { return "", false, fmt.Errorf("codex thread/start returned no thread ID") } + logger.Info("codex lifecycle", + "phase", "thread_start_response", + "task_id", c.cfg.TaskID, + "runtime_id", c.cfg.RuntimeID, + "pid", c.pid, + "attempt", c.attempt, + "active_launches", c.activeLaunches, + "method", "thread/start", + "latency", time.Since(c.threadStartStarted).Round(time.Millisecond).String(), + "latency_ms", time.Since(c.threadStartStarted).Milliseconds(), + ) c.trySetThreadName(ctx, threadID, opts.ThreadName, logger) return threadID, false, nil } @@ -1650,6 +1775,11 @@ type codexClient struct { processDone chan struct{} processErr error handshakeTimeout time.Duration + pid int + attempt int + activeLaunches int64 + threadStartSent bool + threadStartStarted time.Time threadID string turnID string onMessage func(Message) diff --git a/server/pkg/agent/codex_cleanup_unix_test.go b/server/pkg/agent/codex_cleanup_unix_test.go new file mode 100644 index 000000000..941998938 --- /dev/null +++ b/server/pkg/agent/codex_cleanup_unix_test.go @@ -0,0 +1,218 @@ +//go:build unix + +package agent + +import ( + "context" + "fmt" + "log/slog" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + "time" +) + +func TestCodexThreadStartTimeoutReapsDetachedStdioDescendant(t *testing.T) { + tempDir := t.TempDir() + pidFile := filepath.Join(tempDir, "descendant.pid") + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n"+ + `read line`+"\n"+ + `read line`+"\n"+ + `sleep 30 >/dev/null 2>&1 & echo $! > "`+pidFile+`"`+"\n"+ + `sleep 3.2`+"\n"+ + `echo '{"jsonrpc":"2.0","id":2,"result":{"thread":{"id":"thr-late"}}}'`+"\n"+ + `read line`+"\n") + + var logs strings.Builder + backend, err := New("codex", Config{ExecutablePath: fakePath, Logger: slog.New(slog.NewJSONHandler(&logs, nil))}) + if err != nil { + t.Fatal(err) + } + session, err := backend.Execute(context.Background(), "prompt", ExecOptions{Timeout: 8 * time.Second, HandshakeTimeout: 3 * time.Second}) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + if result.Status != "failed" { + t.Fatalf("expected thread/start failure, got %+v", result) + } + rawPID, err := os.ReadFile(pidFile) + if err != nil { + t.Fatal(err) + } + pid, err := strconv.Atoi(strings.TrimSpace(string(rawPID))) + if err != nil { + t.Fatal(err) + } + waitProcessGone(t, pid) + failure := findCodexLifecyclePhase(t, parseJSONLogEntries(t, logs.String()), "thread_start_failure") + if failure["reaped"] != true || failure["cleanup_confirmed"] != true { + t.Fatalf("process-tree cleanup not confirmed: %v", failure) + } +} + +func TestCodexInitializeTimeoutReapsDetachedStdioDescendant(t *testing.T) { + tempDir := t.TempDir() + pidFile := filepath.Join(tempDir, "descendant.pid") + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `sleep 30 >/dev/null 2>&1 & echo $! > "`+pidFile+`"`+"\n"+ + `sleep 3.2`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n"+ + `read line`+"\n") + + var logs strings.Builder + backendRaw, err := New("codex", Config{ExecutablePath: fakePath, Logger: slog.New(slog.NewJSONHandler(&logs, nil))}) + if err != nil { + t.Fatal(err) + } + backend := backendRaw.(*codexBackend) + session, err := backend.executeOnce(context.Background(), "prompt", ExecOptions{Timeout: 8 * time.Second, HandshakeTimeout: 3 * time.Second}, 1) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + if result.Status != "failed" || !strings.Contains(result.Error, "initialize") { + t.Fatalf("expected initialize timeout failure, got %+v", result) + } + rawPID, err := os.ReadFile(pidFile) + if err != nil { + t.Fatal(err) + } + pid, err := strconv.Atoi(strings.TrimSpace(string(rawPID))) + if err != nil { + t.Fatal(err) + } + waitProcessGone(t, pid) + failure := findCodexLifecyclePhase(t, parseJSONLogEntries(t, logs.String()), "initialize_failure") + if failure["cleanup_confirmed"] != true || failure["retry_safe"] != true { + t.Fatalf("process-tree cleanup/retry gate not confirmed: %v", failure) + } +} + +func TestCodexInitializeTimeoutDoesNotPersistOpaqueEnv(t *testing.T) { + const secret = "opaque-init-auth-sentinel-7319" + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `echo "$OPAQUE_AUTH_VALUE" >&2`+"\n"+ + `sleep 3.2`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n") + + var logs strings.Builder + backendRaw, err := New("codex", Config{ + ExecutablePath: fakePath, + Env: map[string]string{"OPAQUE_AUTH_VALUE": secret}, + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + }) + if err != nil { + t.Fatal(err) + } + backend := backendRaw.(*codexBackend) + session, err := backend.executeOnce(context.Background(), "prompt", ExecOptions{Timeout: 8 * time.Second, HandshakeTimeout: 3 * time.Second}, 1) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + if result.Status != "failed" || !strings.Contains(result.Error, CodexHandshakeTimeoutMarker) { + t.Fatalf("expected initialize timeout failure, got %+v", result) + } + if strings.Contains(result.Error, secret) { + t.Fatalf("opaque env persisted in Result.Error: %q", result.Error) + } + if strings.Contains(logs.String(), secret) { + t.Fatalf("opaque env persisted in lifecycle logs: %s", logs.String()) + } + failure := findCodexLifecyclePhase(t, parseJSONLogEntries(t, logs.String()), "initialize_failure") + if failure["cleanup_confirmed"] != true || failure["retry_safe"] != true { + t.Fatalf("cleanup/retry gate changed: %v", failure) + } +} + +func TestCodexInitializeParentContextDoesNotPersistOpaqueEnv(t *testing.T) { + for _, tc := range []struct { + name string + newCtx func() (context.Context, context.CancelFunc) + cancelNow bool + }{ + { + name: "deadline before handshake timeout", + newCtx: func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 2*time.Second) + }, + }, + { + name: "parent cancellation", + newCtx: func() (context.Context, context.CancelFunc) { + return context.WithCancel(context.Background()) + }, + cancelNow: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + const secret = "opaque-init-parent-context-sentinel-8841" + readyFile := filepath.Join(t.TempDir(), "stderr-written") + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `echo "$OPAQUE_AUTH_VALUE" >&2`+"\n"+ + `touch "`+readyFile+`"`+"\n"+ + `sleep 5`+"\n") + + var logs strings.Builder + backend, err := New("codex", Config{ + ExecutablePath: fakePath, + Env: map[string]string{"OPAQUE_AUTH_VALUE": secret}, + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + }) + if err != nil { + t.Fatal(err) + } + ctx, cancel := tc.newCtx() + defer cancel() + session, err := backend.Execute(ctx, "prompt", ExecOptions{Timeout: 10 * time.Second, HandshakeTimeout: 4 * time.Second}) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + if tc.cancelNow { + deadline := time.Now().Add(3 * time.Second) + for { + if _, err := os.Stat(readyFile); err == nil { + break + } + if time.Now().After(deadline) { + t.Fatal("fake app-server did not write stderr before cancellation") + } + time.Sleep(20 * time.Millisecond) + } + cancel() + } + result := <-session.Result + persisted := fmt.Sprintf("%+v\n%s", result, logs.String()) + if strings.Contains(persisted, secret) { + t.Fatalf("opaque env persisted after parent context ended: %s", persisted) + } + if result.Status != "failed" || result.codexInitializeRetrySafe { + t.Fatalf("parent context must fail without initialize retry: %+v", result) + } + }) + } +} diff --git a/server/pkg/agent/codex_test.go b/server/pkg/agent/codex_test.go index ac689520c..ae48f56d8 100644 --- a/server/pkg/agent/codex_test.go +++ b/server/pkg/agent/codex_test.go @@ -2149,6 +2149,255 @@ func TestCodexExecuteStartupRPCsHaveBoundedHandshakeTimeout(t *testing.T) { } } +func TestCodexExecuteThreadStartTimeoutLifecycleIsFailClosed(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell-script fixture is POSIX-only") + } + codexGracefulShutdownTimeoutNanos.Store(int64(100 * time.Millisecond)) + t.Cleanup(func() { codexGracefulShutdownTimeoutNanos.Store(0) }) + + fakePath := writeFakeCodexAppServer(t, ""+ + `DIR="$(dirname "$0")"`+"\n"+ + `echo 1 > "$DIR/attempts"`+"\n"+ + `read line`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n"+ + `read line`+"\n"+ + `read line`+"\n"+ + `echo 'ERROR codex_models_manager::manager: failed to refresh available models: timeout waiting for child process to exit' >&2`+"\n"+ + `echo 'ERROR mcp_manager_init: transport error: channel closed' >&2`+"\n"+ + `sleep 5`+"\n"+ + // This response is deliberately later than the host timeout. Killing + // the process tree must prevent it from reaching turn/start. + `echo '{"jsonrpc":"2.0","id":2,"result":{"thread":{"id":"thr-late"}}}'`+"\n"+ + `read line && echo "$line" > "$DIR/turn-start"`+"\n") + + var logs bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logs, nil)) + backend, err := New("codex", Config{ + ExecutablePath: fakePath, + Logger: logger, + TaskID: "task-thread-start-timeout", + RuntimeID: "runtime-thread-start-timeout", + }) + if err != nil { + t.Fatal(err) + } + session, err := backend.Execute(context.Background(), "secret prompt must not be logged", ExecOptions{ + Timeout: 5 * time.Second, + HandshakeTimeout: 3 * time.Second, + SemanticInactivityTimeout: time.Second, + }) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + if result.Status != "failed" || !strings.Contains(result.Error, "thread/start") { + t.Fatalf("expected thread/start timeout failure, got %+v", result) + } + assertCodexAttemptCount(t, fakePath, "1") + if _, err := os.Stat(filepath.Join(filepath.Dir(fakePath), "turn-start")); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("late response reached turn/start; stat err=%v", err) + } + + entries := parseJSONLogEntries(t, logs.String()) + sent := findCodexLifecyclePhase(t, entries, "thread_start_sent") + failure := findCodexLifecyclePhase(t, entries, "thread_start_failure") + for key, want := range map[string]any{ + "task_id": "task-thread-start-timeout", + "runtime_id": "runtime-thread-start-timeout", + "attempt": float64(1), + "active_launches": float64(1), + "method": "thread/start", + } { + if got := sent[key]; got != want { + t.Fatalf("sent[%s]=%v, want %v; entry=%v", key, got, want, sent) + } + } + if failure["cleanup_confirmed"] != true || failure["reaped"] != true { + t.Fatalf("failure lacks confirmed cleanup/reap: %v", failure) + } + if failure["retry_safe"] != false || failure["retry_attempted"] != false { + t.Fatalf("thread/start timeout must remain fail-closed: %v", failure) + } + if failure["stderr_model_refresh_timeout_count"] != float64(1) || + failure["stderr_mcp_init_transport_count"] != float64(1) || + failure["stderr_bare_timeout_count"] != float64(0) { + t.Fatalf("unexpected stderr classification: %v", failure) + } + if _, ok := failure["latency_ms"]; !ok { + t.Fatalf("failure lacks monotonic latency: %v", failure) + } + if strings.Contains(logs.String(), "secret prompt must not be logged") { + t.Fatalf("prompt leaked into lifecycle logs: %s", logs.String()) + } +} + +func TestCodexExecuteThreadStartResponseRequiresThreadID(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell-script fixture is POSIX-only") + } + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n"+ + `read line`+"\n"+ + `read line`+"\n"+ + `echo '{"jsonrpc":"2.0","id":2,"result":{"thread":{}}}'`+"\n") + + var logs bytes.Buffer + logger := slog.New(slog.NewJSONHandler(&logs, nil)) + backend, err := New("codex", Config{ExecutablePath: fakePath, Logger: logger}) + if err != nil { + t.Fatal(err) + } + session, err := backend.Execute(context.Background(), "prompt", ExecOptions{Timeout: 5 * time.Second}) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + if result.Status != "failed" || !strings.Contains(result.Error, "returned no thread ID") { + t.Fatalf("expected missing thread ID failure, got %+v", result) + } + + phaseCounts := map[string]int{} + for _, entry := range parseJSONLogEntries(t, logs.String()) { + if phase, ok := entry["phase"].(string); ok { + phaseCounts[phase]++ + } + } + if phaseCounts["thread_start_sent"] != 1 || phaseCounts["thread_start_failure"] != 1 { + t.Fatalf("expected one sent and one failure phase, got %v", phaseCounts) + } + if phaseCounts["thread_start_response"] != 0 { + t.Fatalf("invalid thread/start response emitted success phase: %v", phaseCounts) + } +} + +func TestCodexThreadStartTimeoutDoesNotPersistSensitiveInputs(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell-script fixture is POSIX-only") + } + const envSecret = "opaque-env-sentinel-9382" + const systemSecret = "opaque-system-sentinel-4721" + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n"+ + `read line`+"\n"+ + `read line`+"\n"+ + `echo "$OPAQUE_AUTH_VALUE" >&2`+"\n"+ + `echo "$line" >&2`+"\n"+ + `sleep 5`+"\n") + + var logs bytes.Buffer + backend, err := New("codex", Config{ + ExecutablePath: fakePath, + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + Env: map[string]string{"OPAQUE_AUTH_VALUE": envSecret}, + }) + if err != nil { + t.Fatal(err) + } + session, err := backend.Execute(context.Background(), "prompt", ExecOptions{ + Timeout: 5 * time.Second, + HandshakeTimeout: time.Second, + SystemPrompt: systemSecret, + }) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + combined := result.Error + "\n" + logs.String() + for _, secret := range []string{envSecret, systemSecret} { + if strings.Contains(combined, secret) { + t.Fatalf("sensitive input persisted in timeout diagnostics") + } + } +} + +func TestCodexExecuteConcurrentThreadStartTimeoutsRemainUnserialized(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("shell-script fixture is POSIX-only") + } + fakePath := writeFakeCodexAppServer(t, ""+ + `read line`+"\n"+ + `echo '{"jsonrpc":"2.0","id":1,"result":{}}'`+"\n"+ + `read line`+"\n"+ + `read line`+"\n"+ + `sleep 5`+"\n") + + maxActiveCodexLaunchesObserved.Store(0) + results := make(chan Result, 2) + for i := 0; i < 2; i++ { + go func() { + backend, err := New("codex", Config{ExecutablePath: fakePath, Logger: slog.Default()}) + if err != nil { + results <- Result{Status: "failed", Error: err.Error()} + return + } + session, err := backend.Execute(context.Background(), "prompt", ExecOptions{ + Timeout: 5 * time.Second, + HandshakeTimeout: 3 * time.Second, + }) + if err != nil { + results <- Result{Status: "failed", Error: err.Error()} + return + } + go func() { + for range session.Messages { + } + }() + results <- <-session.Result + }() + } + for i := 0; i < 2; i++ { + result := <-results + if result.Status != "failed" || !strings.Contains(result.Error, "thread/start") { + t.Fatalf("concurrent timeout result=%+v", result) + } + } + if got := maxActiveCodexLaunchesObserved.Load(); got < 2 { + t.Fatalf("thread/start timeout handling serialized launches: max active=%d", got) + } +} + +func parseJSONLogEntries(t *testing.T, raw string) []map[string]any { + t.Helper() + var entries []map[string]any + for _, line := range strings.Split(strings.TrimSpace(raw), "\n") { + if line == "" { + continue + } + var entry map[string]any + if err := json.Unmarshal([]byte(line), &entry); err != nil { + t.Fatalf("parse log entry: %v: %q", err, line) + } + entries = append(entries, entry) + } + return entries +} + +func findCodexLifecyclePhase(t *testing.T, entries []map[string]any, phase string) map[string]any { + t.Helper() + for _, entry := range entries { + if entry["msg"] == "codex lifecycle" && entry["phase"] == phase { + return entry + } + } + t.Fatalf("missing codex lifecycle phase %q in %v", phase, entries) + return nil +} + func TestCodexExecuteRetriesInitializeTimeoutOnceAfterCleanup(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("shell-script fixture is POSIX-only") @@ -2169,8 +2418,11 @@ func TestCodexExecuteRetriesInitializeTimeoutOnceAfterCleanup(t *testing.T) { `echo '{"jsonrpc":"2.0","method":"turn/completed","params":{"threadId":"thr-retried","turn":{"id":"turn-1","status":"completed"}}}'`+"\n") result := executeFakeCodex(t, fakePath, ExecOptions{ - Timeout: 10 * time.Second, - HandshakeTimeout: 100 * time.Millisecond, + Timeout: 10 * time.Second, + // Forked shell startup regularly exceeds 100ms on loaded CI runners. + // Keep the first attempt's 5s hang above this bound while giving the + // successful second initialize enough scheduling headroom. + HandshakeTimeout: 2 * time.Second, SemanticInactivityTimeout: time.Second, }) if result.Status != "completed" { @@ -2196,8 +2448,8 @@ func TestCodexExecuteInitializeRetrySafetyGates(t *testing.T) { `count=0; test -f `+countPath+` && count=$(cat `+countPath+`)`+"\n"+ `count=$((count + 1)); echo "$count" > `+countPath+"\n"+ `read line`+"\n"+ - `sleep 1`+"\n") - result := executeFakeCodex(t, fakePath, ExecOptions{Timeout: 5 * time.Second, HandshakeTimeout: 50 * time.Millisecond}) + `sleep 3.2`+"\n") + result := executeFakeCodex(t, fakePath, ExecOptions{Timeout: 8 * time.Second, HandshakeTimeout: 3 * time.Second}) if result.Status != "failed" || !strings.Contains(result.Error, CodexHandshakeTimeoutMarker) { t.Fatalf("expected final initialize timeout, got status=%q error=%q", result.Status, result.Error) } @@ -2213,8 +2465,8 @@ func TestCodexExecuteInitializeRetrySafetyGates(t *testing.T) { `echo x >> `+countPath+"\n"+ `read line`+"\n"+ `echo '{"jsonrpc":"2.0","method":"item/started","params":{"threadId":"unexpected","item":{"type":"commandExecution","id":"cmd-1","command":"true"}}}'`+"\n"+ - `sleep 1`+"\n") - result := executeFakeCodex(t, fakePath, ExecOptions{Timeout: 5 * time.Second, HandshakeTimeout: 50 * time.Millisecond}) + `sleep 3.2`+"\n") + result := executeFakeCodex(t, fakePath, ExecOptions{Timeout: 8 * time.Second, HandshakeTimeout: 3 * time.Second}) if result.Status != "failed" { t.Fatalf("expected failed, got %q", result.Status) } @@ -2231,8 +2483,8 @@ func TestCodexExecuteInitializeRetrySafetyGates(t *testing.T) { fakePath := writeFakeCodexAppServer(t, ""+ `echo x >> `+countPath+"\n"+ `read line`+"\n"+ - `sleep 1`+"\n") - result := executeFakeCodex(t, fakePath, ExecOptions{Timeout: 5 * time.Second, HandshakeTimeout: 50 * time.Millisecond}) + `sleep 3.2`+"\n") + result := executeFakeCodex(t, fakePath, ExecOptions{Timeout: 8 * time.Second, HandshakeTimeout: 3 * time.Second}) if !strings.Contains(result.Error, "retry suppressed: process cleanup/reap not confirmed") { t.Fatalf("expected cleanup reason, got %q", result.Error) } @@ -2296,6 +2548,41 @@ func TestSanitizeCodexDiagnosticRedactsSecrets(t *testing.T) { } } +func TestClassifyCodexStartupStderr(t *testing.T) { + tests := []struct { + name string + stderr string + want codexStderrClassification + }{ + { + name: "model refresh timeout", + stderr: "failed to refresh available models: timeout waiting for child process to exit", + want: codexStderrClassification{modelRefreshTimeout: 1}, + }, + { + name: "mcp init transport", + stderr: "mcp_manager_init transport error: channel closed", + want: codexStderrClassification{mcpInitTransport: 1}, + }, + { + name: "bare timeout", + want: codexStderrClassification{bareTimeout: 1}, + }, + { + name: "non-timeout error is not bare timeout", + want: codexStderrClassification{}, + }, + } + for i, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + timedOut := i != len(tests)-1 + if got := classifyCodexStartupStderr(tc.stderr, timedOut); got != tc.want { + t.Fatalf("classification=%+v, want %+v", got, tc.want) + } + }) + } +} + func TestCodexExecuteRedactsStderrFromResultAndLogs(t *testing.T) { if runtime.GOOS == "windows" { t.Skip("shell-script fixture is POSIX-only") diff --git a/server/pkg/agent/proc_other.go b/server/pkg/agent/proc_other.go index 4c43c521e..d99eb77dd 100644 --- a/server/pkg/agent/proc_other.go +++ b/server/pkg/agent/proc_other.go @@ -3,9 +3,11 @@ package agent import ( + "errors" "os" "os/exec" "syscall" + "time" ) // hideAgentWindow is a no-op on non-Windows platforms. @@ -38,3 +40,20 @@ func signalProcessGroup(p *os.Process, sig syscall.Signal) { _ = p.Signal(sig) } } + +func waitProcessGroupGone(p *os.Process, timeout time.Duration) bool { + if p == nil { + return false + } + deadline := time.Now().Add(timeout) + for { + err := syscall.Kill(-p.Pid, 0) + if errors.Is(err, syscall.ESRCH) { + return true + } + if time.Now().After(deadline) { + return false + } + time.Sleep(10 * time.Millisecond) + } +} diff --git a/server/pkg/agent/proc_windows.go b/server/pkg/agent/proc_windows.go index c862cc606..94c6b5458 100644 --- a/server/pkg/agent/proc_windows.go +++ b/server/pkg/agent/proc_windows.go @@ -6,6 +6,7 @@ import ( "os" "os/exec" "syscall" + "time" ) // createNewConsole allocates a fresh console for the child process. Combined @@ -51,3 +52,5 @@ func signalProcessGroup(p *os.Process, _ syscall.Signal) { } _ = p.Kill() } + +func waitProcessGroupGone(_ *os.Process, _ time.Duration) bool { return false } diff --git a/server/pkg/agent/proc_windows_test.go b/server/pkg/agent/proc_windows_test.go index 7594813e0..1b01147a1 100644 --- a/server/pkg/agent/proc_windows_test.go +++ b/server/pkg/agent/proc_windows_test.go @@ -3,9 +3,17 @@ package agent import ( + "bytes" + "context" + "log/slog" + "os" "os/exec" + "path/filepath" + "strconv" + "strings" "syscall" "testing" + "time" ) // TestHideAgentWindowSetsCreateNewConsole guards against a regression where @@ -69,3 +77,101 @@ func TestCodexInitializeRetrySuppressedWithoutConfirmedTreeCleanup(t *testing.T) t.Fatal("Codex initialize retry must remain disabled until Windows descendant cleanup is positively confirmed") } } + +func TestCodexWindowsInheritedStdoutDescendantCleanupIsBounded(t *testing.T) { + tempDir := t.TempDir() + sourcePath := filepath.Join(tempDir, "fake_codex.go") + exePath := filepath.Join(tempDir, "fake_codex.exe") + pidPath := filepath.Join(tempDir, "descendant.pid") + const source = `package main +import ( + "bufio" + "fmt" + "os" + "os/exec" + "time" +) +func main() { + if len(os.Args) > 1 && os.Args[1] == "--version" { fmt.Println("codex-cli windows-test"); return } + if len(os.Args) > 1 && os.Args[1] == "descendant" { time.Sleep(30*time.Second); return } + s := bufio.NewScanner(os.Stdin) + if !s.Scan() { return } + fmt.Println("{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}") + if !s.Scan() { return } + if !s.Scan() { return } + child := exec.Command(os.Args[0], "descendant") + child.Stdout = os.Stdout + child.Stderr = os.Stderr + if err := child.Start(); err != nil { panic(err) } + _ = os.WriteFile(os.Getenv("DESCENDANT_PID_FILE"), []byte(fmt.Sprint(child.Process.Pid)), 0600) + time.Sleep(time.Hour) +}` + if err := os.WriteFile(sourcePath, []byte(source), 0o600); err != nil { + t.Fatal(err) + } + build := exec.Command("go", "build", "-o", exePath, sourcePath) + if output, err := build.CombinedOutput(); err != nil { + t.Fatalf("build Windows fake app-server: %v: %s", err, output) + } + + codexGracefulShutdownTimeoutNanos.Store(int64(100 * time.Millisecond)) + codexProcessWaitDelayNanos.Store(int64(200 * time.Millisecond)) + t.Cleanup(func() { + codexGracefulShutdownTimeoutNanos.Store(0) + codexProcessWaitDelayNanos.Store(0) + if raw, err := os.ReadFile(pidPath); err == nil { + if pid, err := strconv.Atoi(strings.TrimSpace(string(raw))); err == nil { + if process, err := os.FindProcess(pid); err == nil { + _ = process.Kill() + } + } + } + }) + + var logs bytes.Buffer + backend, err := New("codex", Config{ + ExecutablePath: exePath, + Logger: slog.New(slog.NewJSONHandler(&logs, nil)), + Env: map[string]string{"DESCENDANT_PID_FILE": pidPath}, + }) + if err != nil { + t.Fatal(err) + } + started := time.Now() + session, err := backend.Execute(context.Background(), "prompt", ExecOptions{ + Timeout: 5 * time.Second, + HandshakeTimeout: 500 * time.Millisecond, + }) + if err != nil { + t.Fatal(err) + } + go func() { + for range session.Messages { + } + }() + result := <-session.Result + if elapsed := time.Since(started); elapsed > 3*time.Second { + t.Fatalf("cleanup exceeded bound: %s", elapsed) + } + if result.Status != "failed" || !strings.Contains(result.Error, "thread/start") { + t.Fatalf("expected thread/start failure, got %+v", result) + } + entries := parseJSONLogEntries(t, logs.String()) + failure := findCodexLifecyclePhase(t, entries, "thread_start_failure") + if failure["cleanup_confirmed"] != false || failure["reaped"] != false { + t.Fatalf("Windows tree cleanup must remain unconfirmed: %v", failure) + } + if phaseCount(entries, "thread_start_response") != 0 { + t.Fatalf("unexpected thread_start_response: %v", entries) + } +} + +func phaseCount(entries []map[string]any, phase string) int { + count := 0 + for _, entry := range entries { + if entry["phase"] == phase { + count++ + } + } + return count +}