From 62c30b8f8c89489c7a0ba161fac5d01f55a502d8 Mon Sep 17 00:00:00 2001 From: piekstra Date: Thu, 16 Jul 2026 14:13:01 -0400 Subject: [PATCH 1/2] feat(llm): foreground print-mode transport for Claude tasks (CR_CLAUDE_FOREGROUND) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude tasks run through the CLI's background job service (--bg + job state polling). That layer has proven fragile in production: sessions intermittently die mid-turn without ever writing their state file, so the task starves until its deadline. Observed across CLI versions (2.1.210 and 2.1.211), at reviewer concurrency 1 and 3, on a task that failed 15 consecutive attempts while isolated bg probes succeeded. Add an opt-in foreground transport (CR_CLAUDE_FOREGROUND=1): claude -p --output-format json as a plain bounded child process. Same prompt-file / result-file contract, same flags, same scratch layout; session id parsed from the print-mode JSON. No job service, no state files, no detached workers — lifetime and teardown are fully owned by cr (task deadline + process-group kill + pipe-close teardown). Verification: the 15-consecutive-failure reproducer completed end-to-end in foreground mode (all 6 tasks, rust reviewer 75s). Unit tests cover arg shape, result-file contract, missing-result error, transient stderr classification, and opt-in gating. Background mode remains the default. --- internal/llmadapters/subprocess.go | 205 +++++++++++++++++++++++- internal/llmadapters/subprocess_test.go | 100 ++++++++++++ 2 files changed, 303 insertions(+), 2 deletions(-) diff --git a/internal/llmadapters/subprocess.go b/internal/llmadapters/subprocess.go index 929056e8..816f72f7 100644 --- a/internal/llmadapters/subprocess.go +++ b/internal/llmadapters/subprocess.go @@ -169,6 +169,9 @@ func (a *SubprocessAdapter) Start(ctx context.Context, req Request) (Stream, err return nil, err } if a.kind == subprocessClaude { + if claudeForegroundEnabled(a.env) { + return a.startClaudeForeground(ctx, req, "") + } return a.startClaudeBG(ctx, req, "") } if a.kind == subprocessCodex && !a.allowBestEffortNoTools { @@ -279,6 +282,196 @@ func (a *SubprocessAdapter) startClaudeBG(ctx context.Context, req Request, resu return stream, nil } +// claudeForegroundEnabled reports whether Claude tasks should run in foreground +// print mode instead of --bg. Background mode depends on the Claude CLI's +// job-service layer (detached daemon, job state files, result polling), which +// has proven fragile in production: sessions intermittently die mid-turn +// without ever writing their state file, starving the review until its task +// deadline. Foreground mode is a plain child process — its lifetime, output, +// and teardown are fully owned by cr (bounded by the task timeout, cleaned up +// by the process-group kill and pipe-close teardown). +func claudeForegroundEnabled(env []string) bool { + for i := len(env) - 1; i >= 0; i-- { + if strings.HasPrefix(env[i], "CR_CLAUDE_FOREGROUND=") { + return isTruthyEnvValue(strings.TrimPrefix(env[i], "CR_CLAUDE_FOREGROUND=")) + } + } + return isTruthyEnvValue(os.Getenv("CR_CLAUDE_FOREGROUND")) +} + +func isTruthyEnvValue(v string) bool { + switch strings.ToLower(strings.TrimSpace(v)) { + case "1", "true", "yes", "on": + return true + } + return false +} + +// buildClaudeForegroundArgs mirrors the Claude background args with the job +// service swapped for headless print mode. The prompt-file/result-file +// contract is unchanged: the model writes its structured result to +// cr-result.json in the scratch dir. +func buildClaudeForegroundArgs(req Request, scratch string, resumeSessionID string) []string { + tools := "Read,Write" + if req.ReviewerWorkspace != nil { + tools = "Read,Write,Bash" + } + args := []string{ + "-p", + "--output-format", "json", + "--tools", tools, + "--permission-mode", "acceptEdits", + "--add-dir", scratch, + } + if workspace := req.ReviewerWorkspace; workspace != nil { + args = append(args, "--add-dir", workspace.RepoDir) + } + if resumeSessionID != "" { + args = append(args, "--resume", resumeSessionID) + } + if req.Model != "" { + args = append(args, "--model", req.Model) + } + if req.Effort != "" { + args = append(args, "--effort", req.Effort) + } + if req.Fast { + args = append(args, "--settings", `{"fastMode":true}`) + } + return append(args, "--", claudeBGPositionalPrompt(scratch)) +} + +// startClaudeForeground runs a Claude task as a plain foreground child in +// headless print mode. Same scratch, prompt-file, env, and result-file +// contract as background mode — only the transport differs. +func (a *SubprocessAdapter) startClaudeForeground(ctx context.Context, req Request, resumeSessionID string) (Stream, error) { + scratch, cleanup, err := a.invocationScratchDir(req) + if err != nil { + return nil, err + } + if cleanup == nil { + cleanup = func() error { return nil } + } + scratch, err = validateScratchDir(scratch) + if err != nil { + _ = cleanup() + return nil, err + } + if err := writeClaudeBGPromptFile(req.Prompt, scratch); err != nil { + _ = cleanup() + return nil, err + } + args := buildClaudeForegroundArgs(req, scratch, resumeSessionID) + if err := a.validateArgs(args, scratch, req); err != nil { + _ = cleanup() + return nil, err + } + workDir, err := claudeBGWorkingDir(a.env) + if err != nil { + _ = cleanup() + return nil, err + } + execArgs := append(append([]string(nil), a.commandArgsPrefix...), args...) + env, err := a.processEnv(req, scratch) + if err != nil { + _ = cleanup() + return nil, err + } + process, err := launchProcess(ctx, a.command, execArgs, workDir, env, a.timeout, req.LogPath, cleanup, false) + if err != nil { + return nil, err + } + stream := &subprocessStream{ + baseStream: llm.NewProcessStream(process, cleanup), + logBytesLeft: reviewerWorkspaceLogBytes(req), + } + go stream.runClaudeForeground(process.Context(), process.Command(), process.Stdout(), process.Stderr(), scratch) + return stream, nil +} + +// claudeForegroundOutput is the trailing JSON document `claude -p +// --output-format json` prints on exit. +type claudeForegroundOutput struct { + SessionID string `json:"session_id"` + IsError bool `json:"is_error"` + Result string `json:"result"` +} + +func (s *subprocessStream) runClaudeForeground(ctx context.Context, cmd *exec.Cmd, stdout io.Reader, stderr io.Reader, scratch string) { + stderrDone := make(chan bytes.Buffer, 1) + go func() { + var stderrBuf bytes.Buffer + if s.HasLog() { + _, _ = io.Copy(io.MultiWriter(subprocessLogWriter{stream: s}, &stderrBuf), stderr) + } else { + _, _ = io.Copy(&stderrBuf, stderr) + } + stderrDone <- stderrBuf + }() + + outBytes, readErr := io.ReadAll(stdout) + if len(outBytes) > 0 { + s.WriteLog(outBytes) + } + waitErr := cmd.Wait() + stderrBuf := <-stderrDone + + result := subprocessResult{} + switch { + case readErr != nil: + result.err = readErr + case ctx.Err() != nil: + result.err = fmt.Errorf("llm subprocess: Claude foreground task timed out: %w", ctx.Err()) + case waitErr != nil: + detail := strings.TrimSpace(stderrBuf.String()) + if detail == "" { + detail = strings.TrimSpace(string(outBytes)) + } + if detail != "" { + runErr := fmt.Errorf("llm subprocess: Claude foreground task failed: %w: %s", waitErr, detail) + if isTransientCLIDetail(detail) { + result.err = fmt.Errorf("%w: %w", ErrTransient, runErr) + } else { + result.err = runErr + } + } else { + result.err = waitErr + } + } + + s.CloseProcessGroup() + + if result.err == nil { + var parsed claudeForegroundOutput + if idx := bytes.IndexByte(outBytes, '{'); idx >= 0 { + _ = json.Unmarshal(outBytes[idx:], &parsed) + } + if parsed.SessionID != "" { + s.SetSessionID(parsed.SessionID) + } + output, err := readFirstNonEmptyFile([]string{filepath.Join(scratch, claudeBGResultFilename)}) + switch { + case err == nil: + result.response = Response{StructuredOutput: output} + case parsed.IsError: + detail := strings.TrimSpace(parsed.Result) + runErr := fmt.Errorf("llm subprocess: Claude foreground task errored: %s", detail) + if isTransientCLIDetail(detail) { + result.err = fmt.Errorf("%w: %w", ErrTransient, runErr) + } else { + result.err = runErr + } + default: + result.err = fmt.Errorf("llm subprocess: Claude foreground task completed without a result file: %w", err) + } + } + + s.Cancel() + s.CloseLog() + s.runCleanup() + s.Finish(result.response, result.err) +} + func (a *SubprocessAdapter) processEnv(req Request, scratch string) ([]string, error) { env := append(os.Environ(), a.env...) if req.ReviewerWorkspace == nil { @@ -348,6 +541,9 @@ func (a *SubprocessAdapter) Resume(ctx context.Context, sessionID string, req Re // through optional resume state without special-casing the first run. return a.Start(ctx, req) } + if claudeForegroundEnabled(a.env) { + return a.startClaudeForeground(ctx, req, sessionID) + } return a.startClaudeBG(ctx, req, sessionID) case subprocessCodex: if !a.allowBestEffortNoTools { @@ -469,6 +665,8 @@ func (a *SubprocessAdapter) validateArgs(args []string, scratch string, req Requ } if err := validateAllowedFlags("claude_cli", checkedArgs, map[string]bool{ "--bg": false, + "-p": false, + "--output-format": true, "--tools": true, "--permission-mode": true, "--add-dir": true, @@ -479,8 +677,11 @@ func (a *SubprocessAdapter) validateArgs(args []string, scratch string, req Requ }); err != nil { return err } - if !containsFlag(checkedArgs, "--bg") { - return fmt.Errorf("%w: claude_cli must use background jobs", ErrUnsafeSubprocessConfig) + // Exactly one transport: the detached job service (--bg) or headless + // foreground print mode (-p, see claudeForegroundEnabled). + if containsFlag(checkedArgs, "--bg") == containsFlag(checkedArgs, "-p") { + return fmt.Errorf( + "%w: claude_cli must use exactly one of --bg or -p", ErrUnsafeSubprocessConfig) } requiredTools := "Read,Write" if req.ReviewerWorkspace != nil { diff --git a/internal/llmadapters/subprocess_test.go b/internal/llmadapters/subprocess_test.go index 615c3872..d2bff82b 100644 --- a/internal/llmadapters/subprocess_test.go +++ b/internal/llmadapters/subprocess_test.go @@ -1366,6 +1366,10 @@ func TestSubprocessHelperProcess(_ *testing.T) { runClaudeBGHelper(os.Getenv("LLM_HELPER_MODE"), args) os.Exit(0) } + if containsFlag(args, "-p") && flagValue(args, "--output-format") == "json" { + runClaudeForegroundHelper(os.Getenv("LLM_HELPER_MODE"), args) + os.Exit(0) + } switch os.Getenv("LLM_HELPER_MODE") { case "success": fmt.Println(`{"type":"thread.started","thread_id":"session-1"}`) @@ -1841,3 +1845,99 @@ func TestSubprocessAdapterDefaultsTaskTimeout(t *testing.T) { t.Fatalf("pi timeout = %v, want default %v", pi.timeout, defaultLLMTaskTimeout) } } + +func runClaudeForegroundHelper(mode string, args []string) { + scratch := flagValue(args, "--add-dir") + switch mode { + case "foreground-success": + if scratch != "" { + _ = os.WriteFile(filepath.Join(scratch, claudeBGResultFilename), []byte(`{"ok":true}`), 0o600) + } + fmt.Println(`{"type":"result","subtype":"success","is_error":false,"result":"wrote the result file","session_id":"fg-session-1"}`) + case "foreground-no-result": + fmt.Println(`{"type":"result","subtype":"success","is_error":false,"result":"forgot the file","session_id":"fg-session-2"}`) + case "foreground-fail": + fmt.Fprintln(os.Stderr, "model is overloaded_error") + os.Exit(1) + } +} + +func TestSubprocessClaudeForegroundMode(t *testing.T) { + tempDir := t.TempDir() + recordPath := filepath.Join(tempDir, "records.jsonl") + configDir := filepath.Join(tempDir, "claude") + adapter := newClaudeHelperAdapterWithEnv( + "foreground-success", recordPath, configDir, 5*time.Second, "CR_CLAUDE_FOREGROUND=1") + + stream, err := adapter.Start(context.Background(), Request{ + Model: "claude-sonnet-5", + Effort: "high", + Prompt: "prompt", + LogPath: filepath.Join(tempDir, "events.log"), + }) + if err != nil { + t.Fatalf("Start: %v", err) + } + response, err := stream.Wait(context.Background()) + if err != nil { + t.Fatalf("Wait: %v", err) + } + if string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("StructuredOutput = %s", response.StructuredOutput) + } + if stream.SessionID() != "fg-session-1" { + t.Fatalf("SessionID = %q, want fg-session-1", stream.SessionID()) + } + + records := readHelperRecords(t, recordPath) + record := records[0] + if containsFlag(record.AdapterArgs, "--bg") { + t.Fatalf("args = %#v, foreground mode must not pass --bg", record.AdapterArgs) + } + if !containsFlag(record.AdapterArgs, "-p") || flagValue(record.AdapterArgs, "--output-format") != "json" { + t.Fatalf("args = %#v, want -p --output-format json", record.AdapterArgs) + } + assertFlagValue(t, record.AdapterArgs, "--model", "claude-sonnet-5") + assertFlagValue(t, record.AdapterArgs, "--permission-mode", "acceptEdits") +} + +func TestSubprocessClaudeForegroundNoResultFileErrors(t *testing.T) { + tempDir := t.TempDir() + adapter := newClaudeHelperAdapterWithEnv( + "foreground-no-result", filepath.Join(tempDir, "records.jsonl"), + filepath.Join(tempDir, "claude"), 5*time.Second, "CR_CLAUDE_FOREGROUND=1") + stream, err := adapter.Start(context.Background(), Request{Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + if _, err := stream.Wait(context.Background()); err == nil { + t.Fatal("Wait should fail when no result file was written") + } +} + +func TestSubprocessClaudeForegroundTransientFailure(t *testing.T) { + tempDir := t.TempDir() + adapter := newClaudeHelperAdapterWithEnv( + "foreground-fail", filepath.Join(tempDir, "records.jsonl"), + filepath.Join(tempDir, "claude"), 5*time.Second, "CR_CLAUDE_FOREGROUND=1") + stream, err := adapter.Start(context.Background(), Request{Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + _, err = stream.Wait(context.Background()) + if err == nil || !errors.Is(err, ErrTransient) { + t.Fatalf("Wait err = %v, want ErrTransient (overloaded stderr detail)", err) + } +} + +func TestClaudeForegroundDisabledByDefault(t *testing.T) { + if claudeForegroundEnabled(nil) && os.Getenv("CR_CLAUDE_FOREGROUND") == "" { + t.Fatal("foreground must be opt-in") + } + if !claudeForegroundEnabled([]string{"CR_CLAUDE_FOREGROUND=1"}) { + t.Fatal("env slice should enable foreground") + } + if claudeForegroundEnabled([]string{"CR_CLAUDE_FOREGROUND=0"}) { + t.Fatal("explicit 0 should disable") + } +} From 90e0ed5bffc36d9a2ebccb018df2892ffe5a24b4 Mon Sep 17 00:00:00 2001 From: piekstra Date: Thu, 16 Jul 2026 14:44:18 -0400 Subject: [PATCH 2/2] test(llm): cover foreground resume, transport XOR validation, preamble-tolerant output parsing Addresses review findings on #511: adds the foreground Resume path test (-p + --resume recorded, no --bg fallback), a table test for the exactly-one-transport validateArgs guard, and hardens print-mode output parsing to take the last parseable JSON line (banners/warnings before the payload no longer break session-id extraction; parse failures are logged). --- internal/llmadapters/subprocess.go | 27 +++++++- internal/llmadapters/subprocess_test.go | 87 +++++++++++++++++++++++++ 2 files changed, 111 insertions(+), 3 deletions(-) diff --git a/internal/llmadapters/subprocess.go b/internal/llmadapters/subprocess.go index 816f72f7..bed68c27 100644 --- a/internal/llmadapters/subprocess.go +++ b/internal/llmadapters/subprocess.go @@ -397,6 +397,25 @@ type claudeForegroundOutput struct { Result string `json:"result"` } +// parseClaudeForegroundOutput extracts the print-mode result document from +// captured stdout. The CLI may print banners or warnings before the trailing +// JSON, so scan lines from the end and take the last one that parses as a JSON +// object — never a leading '{' from unrelated preamble. +func parseClaudeForegroundOutput(out []byte) (claudeForegroundOutput, bool) { + lines := bytes.Split(out, []byte("\n")) + for i := len(lines) - 1; i >= 0; i-- { + line := bytes.TrimSpace(lines[i]) + if len(line) == 0 || line[0] != '{' { + continue + } + var parsed claudeForegroundOutput + if err := json.Unmarshal(line, &parsed); err == nil { + return parsed, true + } + } + return claudeForegroundOutput{}, false +} + func (s *subprocessStream) runClaudeForeground(ctx context.Context, cmd *exec.Cmd, stdout io.Reader, stderr io.Reader, scratch string) { stderrDone := make(chan bytes.Buffer, 1) go func() { @@ -442,9 +461,11 @@ func (s *subprocessStream) runClaudeForeground(ctx context.Context, cmd *exec.Cm s.CloseProcessGroup() if result.err == nil { - var parsed claudeForegroundOutput - if idx := bytes.IndexByte(outBytes, '{'); idx >= 0 { - _ = json.Unmarshal(outBytes[idx:], &parsed) + parsed, ok := parseClaudeForegroundOutput(outBytes) + if !ok { + // The session id is only needed for resume, so a parse failure must + // not fail the task — but it must be diagnosable in the task log. + s.WriteLog([]byte("llm subprocess: could not parse Claude foreground output JSON\n")) } if parsed.SessionID != "" { s.SetSessionID(parsed.SessionID) diff --git a/internal/llmadapters/subprocess_test.go b/internal/llmadapters/subprocess_test.go index d2bff82b..c6f866e5 100644 --- a/internal/llmadapters/subprocess_test.go +++ b/internal/llmadapters/subprocess_test.go @@ -1859,6 +1859,13 @@ func runClaudeForegroundHelper(mode string, args []string) { case "foreground-fail": fmt.Fprintln(os.Stderr, "model is overloaded_error") os.Exit(1) + case "foreground-preamble": + fmt.Println("Warning: some banner text before the payload") + fmt.Println(`{"unrelated": true} trailing junk that must not parse`) + if scratch != "" { + _ = os.WriteFile(filepath.Join(scratch, claudeBGResultFilename), []byte(`{"ok":true}`), 0o600) + } + fmt.Println(`{"type":"result","subtype":"success","is_error":false,"result":"done","session_id":"fg-session-3"}`) } } @@ -1941,3 +1948,83 @@ func TestClaudeForegroundDisabledByDefault(t *testing.T) { t.Fatal("explicit 0 should disable") } } + +func TestSubprocessClaudeForegroundResume(t *testing.T) { + tempDir := t.TempDir() + recordPath := filepath.Join(tempDir, "records.jsonl") + adapter := newClaudeHelperAdapterWithEnv( + "foreground-success", recordPath, filepath.Join(tempDir, "claude"), + 5*time.Second, "CR_CLAUDE_FOREGROUND=1") + + stream, err := adapter.Resume(context.Background(), "prior-session", Request{ + Model: "claude-sonnet-5", + Prompt: "resume prompt", + }) + if err != nil { + t.Fatalf("Resume: %v", err) + } + if _, err := stream.Wait(context.Background()); err != nil { + t.Fatalf("Wait: %v", err) + } + + record := readHelperRecords(t, recordPath)[0] + if containsFlag(record.AdapterArgs, "--bg") { + t.Fatalf("args = %#v, foreground resume must not fall back to --bg", record.AdapterArgs) + } + if !containsFlag(record.AdapterArgs, "-p") { + t.Fatalf("args = %#v, want -p for foreground resume", record.AdapterArgs) + } + assertFlagValue(t, record.AdapterArgs, "--resume", "prior-session") +} + +func TestSubprocessClaudeForegroundPreambleParsing(t *testing.T) { + tempDir := t.TempDir() + adapter := newClaudeHelperAdapterWithEnv( + "foreground-preamble", filepath.Join(tempDir, "records.jsonl"), + filepath.Join(tempDir, "claude"), 5*time.Second, "CR_CLAUDE_FOREGROUND=1") + stream, err := adapter.Start(context.Background(), Request{Prompt: "prompt"}) + if err != nil { + t.Fatalf("Start: %v", err) + } + response, err := stream.Wait(context.Background()) + if err != nil { + t.Fatalf("Wait: %v", err) + } + if string(response.StructuredOutput) != `{"ok":true}` { + t.Fatalf("StructuredOutput = %s", response.StructuredOutput) + } + if stream.SessionID() != "fg-session-3" { + t.Fatalf("SessionID = %q, want fg-session-3 (parsed from trailing JSON, not preamble)", stream.SessionID()) + } +} + +func TestClaudeTransportXORValidation(t *testing.T) { + scratch := t.TempDir() + adapter := NewClaudeCLIAdapter(SubprocessOptions{}) + base := []string{ + "--tools", "Read,Write", + "--permission-mode", "acceptEdits", + "--add-dir", scratch, + "--", "positional prompt", + } + cases := []struct { + name string + prefix []string + wantErr bool + }{ + {"bg only", []string{"--bg"}, false}, + {"foreground only", []string{"-p", "--output-format", "json"}, false}, + {"both transports", []string{"--bg", "-p", "--output-format", "json"}, true}, + {"neither transport", nil, true}, + } + for _, tc := range cases { + args := append(append([]string(nil), tc.prefix...), base...) + err := adapter.validateArgs(args, scratch, Request{}) + if tc.wantErr && !errors.Is(err, ErrUnsafeSubprocessConfig) { + t.Fatalf("%s: err = %v, want ErrUnsafeSubprocessConfig", tc.name, err) + } + if !tc.wantErr && err != nil { + t.Fatalf("%s: err = %v, want nil", tc.name, err) + } + } +}