From 5b3207e944cce35d58de78a373b23a082c6f89cf Mon Sep 17 00:00:00 2001 From: maxBRT Date: Sat, 25 Jul 2026 15:24:16 +0000 Subject: [PATCH 1/2] fix: stream agent phase diagnostics live --- internal/agent/claude.go | 165 ++++++++++++++---------------- internal/agent/codex.go | 135 +++++++++++------------- internal/agent/cursor.go | 121 ++++++++++------------ internal/agent/cursor_test.go | 120 +++++++++++++++++++--- internal/agent/pi.go | 118 ++++++++++----------- internal/agent/process_unix.go | 28 +++++ internal/agent/process_windows.go | 16 +++ internal/agent/supervisor.go | 136 ++++++++++++++++++++++++ internal/run/shipfile.go | 3 + internal/run/shipfile_test.go | 23 +++++ 10 files changed, 556 insertions(+), 309 deletions(-) create mode 100644 internal/agent/process_unix.go create mode 100644 internal/agent/process_windows.go create mode 100644 internal/agent/supervisor.go diff --git a/internal/agent/claude.go b/internal/agent/claude.go index d4f5ec5..4470061 100644 --- a/internal/agent/claude.go +++ b/internal/agent/claude.go @@ -27,12 +27,6 @@ func (c Claude) bin() string { // RunPhase spawns a fresh Claude Code print-mode process for one Phase. func (c Claude) RunPhase(ctx context.Context, req PhaseRequest) error { - if req.Timeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, req.Timeout) - defer cancel() - } - args := []string{ "-p", "--dangerously-skip-permissions", @@ -44,30 +38,10 @@ func (c Claude) RunPhase(ctx context.Context, req PhaseRequest) error { args = append(args, "--model", req.Model) } - cmd := exec.CommandContext(ctx, c.bin(), args...) // #nosec G204 -- Claude Code CLI; args built by Ship + cmd := exec.Command(c.bin(), args...) // #nosec G204 -- Claude Code CLI; args built by Ship cmd.Dir = req.Workspace cmd.Stdin = strings.NewReader(req.Prompt) - - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - err := cmd.Run() - if err != nil { - if ctx.Err() != nil { - return fmt.Errorf("phase: %w", ctx.Err()) - } - msg := strings.TrimSpace(stderr.String()) - if msg == "" { - msg = err.Error() - } - return fmt.Errorf("phase: %s", msg) - } - - if err := consumeClaudeStreamJSON(stdout.Bytes(), req.Events); err != nil { - return err - } - return nil + return runStreamedPhase(ctx, req.Timeout, cmd, newClaudeStream(req.Events)) } type claudeEvent struct { @@ -99,87 +73,100 @@ type claudeUsage struct { func consumeClaudeStreamJSON(stdout []byte, sink observe.Sink) error { lines := bytes.Split(stdout, []byte("\n")) - var last *claudeEvent - toolCount := 0 - pending := map[string]string{} // tool_use id -> name + stream := newClaudeStream(sink) for _, line := range lines { - line = bytes.TrimSpace(line) - if len(line) == 0 { - continue - } - var ev claudeEvent - if err := json.Unmarshal(line, &ev); err != nil { - continue + stream.ProcessLine(line) + } + return stream.Finish() +} + +type claudeStream struct { + sink observe.Sink + last *claudeEvent + toolCount int + pending map[string]string +} + +func newClaudeStream(sink observe.Sink) *claudeStream { + return &claudeStream{sink: sink, pending: map[string]string{}} +} + +func (s *claudeStream) ProcessLine(line []byte) { + line = bytes.TrimSpace(line) + if len(line) == 0 { + return + } + var ev claudeEvent + if err := json.Unmarshal(line, &ev); err != nil { + return + } + switch ev.Type { + case "assistant": + if ev.Message == nil { + return } - switch ev.Type { - case "assistant": - if ev.Message == nil { + for _, block := range ev.Message.Content { + if block.Type != "tool_use" { continue } - for _, block := range ev.Message.Content { - if block.Type != "tool_use" { - continue - } - name := block.Name - if name == "" { - name = "unknown" - } - if block.ID != "" { - pending[block.ID] = name - } + name := block.Name + if name == "" { + name = "unknown" + } + if block.ID != "" { + s.pending[block.ID] = name } - case "user": - if ev.Message == nil { + } + case "user": + if ev.Message == nil { + return + } + for _, block := range ev.Message.Content { + if block.Type != "tool_result" { + continue + } + s.toolCount++ + if s.sink == nil { continue } - for _, block := range ev.Message.Content { - if block.Type != "tool_result" { - continue - } - toolCount++ - if sink == nil { - continue - } - name := pending[block.ToolUseID] - if name == "" { - name = "unknown" - } - status := observe.ToolOK - if block.IsError { - status = observe.ToolError - } - sink.Emit(observe.Event{ - Kind: observe.KindTool, - Name: name, - Status: status, - }) + name := s.pending[block.ToolUseID] + if name == "" { + name = "unknown" } - case "result": - last = &ev + status := observe.ToolOK + if block.IsError { + status = observe.ToolError + } + s.sink.Emit(observe.Event{Kind: observe.KindTool, Name: name, Status: status}) } + case "result": + s.last = &ev } - if last == nil { +} + +func (s *claudeStream) Finish() error { + if s.last == nil { return fmt.Errorf("phase: missing terminal result event in stream-json output") } - if last.Subtype != "success" { - return fmt.Errorf("phase: result subtype %q, want success", last.Subtype) + if s.last.Subtype != "success" { + return fmt.Errorf("phase: result subtype %q, want success", s.last.Subtype) } - if sink != nil { + if s.sink != nil { end := observe.Event{ Kind: observe.KindPhaseEnd, Outcome: observe.OutcomeSuccess, - DurationMS: last.DurationMS, - ToolCount: toolCount, + DurationMS: s.last.DurationMS, + ToolCount: s.toolCount, } - if last.Usage != nil { + if s.last.Usage != nil { end.Tokens = &observe.TokenCounts{ - Input: last.Usage.InputTokens, - Output: last.Usage.OutputTokens, - CacheRead: last.Usage.CacheReadInputTokens, - CacheWrite: last.Usage.CacheCreationInputTokens, + Input: s.last.Usage.InputTokens, + Output: s.last.Usage.OutputTokens, + CacheRead: s.last.Usage.CacheReadInputTokens, + CacheWrite: s.last.Usage.CacheCreationInputTokens, } } - sink.Emit(end) + s.sink.Emit(end) } return nil } diff --git a/internal/agent/codex.go b/internal/agent/codex.go index 3c84ef9..e094a42 100644 --- a/internal/agent/codex.go +++ b/internal/agent/codex.go @@ -27,12 +27,6 @@ func (c Codex) bin() string { // RunPhase spawns a fresh ephemeral Codex exec process for one Phase. func (c Codex) RunPhase(ctx context.Context, req PhaseRequest) error { - if req.Timeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, req.Timeout) - defer cancel() - } - args := []string{ "exec", "--ephemeral", @@ -46,36 +40,16 @@ func (c Codex) RunPhase(ctx context.Context, req PhaseRequest) error { // "-" forces the full Phase prompt from stdin (not prompt+context mode). args = append(args, "-") - cmd := exec.CommandContext(ctx, c.bin(), args...) // #nosec G204 -- Codex agent CLI; args built by Ship + cmd := exec.Command(c.bin(), args...) // #nosec G204 -- Codex agent CLI; args built by Ship cmd.Dir = req.Workspace cmd.Stdin = strings.NewReader(req.Prompt) - - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - err := cmd.Run() - if err != nil { - if ctx.Err() != nil { - return fmt.Errorf("phase: %w", ctx.Err()) - } - msg := strings.TrimSpace(stderr.String()) - if msg == "" { - msg = err.Error() - } - return fmt.Errorf("phase: %s", msg) - } - - if err := consumeCodexJSON(stdout.Bytes(), req.Events); err != nil { - return err - } - return nil + return runStreamedPhase(ctx, req.Timeout, cmd, newCodexStream(req.Events)) } type codexEvent struct { - Type string `json:"type"` - Item *codexItem `json:"item"` - Usage *codexUsage `json:"usage"` + Type string `json:"type"` + Item *codexItem `json:"item"` + Usage *codexUsage `json:"usage"` } type codexItem struct { @@ -94,62 +68,71 @@ type codexUsage struct { func consumeCodexJSON(stdout []byte, sink observe.Sink) error { lines := bytes.Split(stdout, []byte("\n")) - sawCompleted := false - sawFailed := false - toolCount := 0 - var tokens *observe.TokenCounts + stream := newCodexStream(sink) for _, line := range lines { - line = bytes.TrimSpace(line) - if len(line) == 0 { - continue - } - var ev codexEvent - if err := json.Unmarshal(line, &ev); err != nil { - continue + stream.ProcessLine(line) + } + return stream.Finish() +} + +type codexStream struct { + sink observe.Sink + sawCompleted bool + sawFailed bool + toolCount int + tokens *observe.TokenCounts +} + +func newCodexStream(sink observe.Sink) *codexStream { + return &codexStream{sink: sink} +} + +func (s *codexStream) ProcessLine(line []byte) { + line = bytes.TrimSpace(line) + if len(line) == 0 { + return + } + var ev codexEvent + if err := json.Unmarshal(line, &ev); err != nil { + return + } + switch ev.Type { + case "item.completed": + if ev.Item == nil || !codexItemIsTool(ev.Item.Type) { + return } - switch ev.Type { - case "item.completed": - if ev.Item == nil || !codexItemIsTool(ev.Item.Type) { - continue - } - toolCount++ - if sink != nil { - name := codexToolName(ev.Item) - status := observe.ToolOK - if ev.Item.Status == "failed" || ev.Item.Status == "error" { - status = observe.ToolError - } - sink.Emit(observe.Event{ - Kind: observe.KindTool, - Name: name, - Status: status, - }) - } - case "turn.completed": - sawCompleted = true - if ev.Usage != nil { - tokens = &observe.TokenCounts{ - Input: ev.Usage.InputTokens, - Output: ev.Usage.OutputTokens + ev.Usage.ReasoningOutputTokens, - CacheRead: ev.Usage.CachedInputTokens, - } + s.toolCount++ + if s.sink != nil { + name := codexToolName(ev.Item) + status := observe.ToolOK + if ev.Item.Status == "failed" || ev.Item.Status == "error" { + status = observe.ToolError } - case "turn.failed": - sawFailed = true + s.sink.Emit(observe.Event{Kind: observe.KindTool, Name: name, Status: status}) } + case "turn.completed": + s.sawCompleted = true + if ev.Usage != nil { + s.tokens = &observe.TokenCounts{Input: ev.Usage.InputTokens, Output: ev.Usage.OutputTokens + ev.Usage.ReasoningOutputTokens, CacheRead: ev.Usage.CachedInputTokens} + } + case "turn.failed": + s.sawFailed = true } - if sawFailed { +} + +func (s *codexStream) Finish() error { + if s.sawFailed { return fmt.Errorf("phase: turn.failed") } - if !sawCompleted { + if !s.sawCompleted { return fmt.Errorf("phase: missing terminal turn.completed event in json output") } - if sink != nil { - sink.Emit(observe.Event{ + if s.sink != nil { + s.sink.Emit(observe.Event{ Kind: observe.KindPhaseEnd, Outcome: observe.OutcomeSuccess, - ToolCount: toolCount, - Tokens: tokens, + ToolCount: s.toolCount, + Tokens: s.tokens, }) } return nil diff --git a/internal/agent/cursor.go b/internal/agent/cursor.go index b8b1a5c..e665e5f 100644 --- a/internal/agent/cursor.go +++ b/internal/agent/cursor.go @@ -28,12 +28,6 @@ func (c Cursor) bin() string { // RunPhase spawns a fresh headless agent process for one Phase. func (c Cursor) RunPhase(ctx context.Context, req PhaseRequest) error { - if req.Timeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, req.Timeout) - defer cancel() - } - args := []string{ "-p", "--trust", @@ -46,30 +40,10 @@ func (c Cursor) RunPhase(ctx context.Context, req PhaseRequest) error { args = append(args, "--model", req.Model) } - cmd := exec.CommandContext(ctx, c.bin(), args...) // #nosec G204 -- Cursor agent CLI; args built by Ship + cmd := exec.Command(c.bin(), args...) // #nosec G204 -- Cursor agent CLI; args built by Ship cmd.Dir = req.Workspace cmd.Stdin = strings.NewReader(req.Prompt) - - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - err := cmd.Run() - if err != nil { - if ctx.Err() != nil { - return fmt.Errorf("phase: %w", ctx.Err()) - } - msg := strings.TrimSpace(stderr.String()) - if msg == "" { - msg = err.Error() - } - return fmt.Errorf("phase: %s", msg) - } - - if err := consumeStreamJSON(stdout.Bytes(), req.Events); err != nil { - return err - } - return nil + return runStreamedPhase(ctx, req.Timeout, cmd, newCursorStream(req.Events)) } type streamEvent struct { @@ -90,57 +64,74 @@ type streamUsage struct { func consumeStreamJSON(stdout []byte, sink observe.Sink) error { lines := bytes.Split(stdout, []byte("\n")) - var last *streamEvent - toolCount := 0 + stream := newCursorStream(sink) for _, line := range lines { - line = bytes.TrimSpace(line) - if len(line) == 0 { - continue - } - var ev streamEvent - if err := json.Unmarshal(line, &ev); err != nil { - continue - } - switch ev.Type { - case "tool_call": - if ev.Subtype == "completed" { - toolCount++ - if sink != nil { - name, status := parseToolCall(ev.ToolCall) - sink.Emit(observe.Event{ - Kind: observe.KindTool, - Name: name, - DurationMS: ev.DurationMS, - Status: status, - }) - } + stream.ProcessLine(line) + } + return stream.Finish() +} + +type cursorStream struct { + sink observe.Sink + last *streamEvent + toolCount int +} + +func newCursorStream(sink observe.Sink) *cursorStream { + return &cursorStream{sink: sink} +} + +func (s *cursorStream) ProcessLine(line []byte) { + line = bytes.TrimSpace(line) + if len(line) == 0 { + return + } + var ev streamEvent + if err := json.Unmarshal(line, &ev); err != nil { + return + } + switch ev.Type { + case "tool_call": + if ev.Subtype == "completed" { + s.toolCount++ + if s.sink != nil { + name, status := parseToolCall(ev.ToolCall) + s.sink.Emit(observe.Event{ + Kind: observe.KindTool, + Name: name, + DurationMS: ev.DurationMS, + Status: status, + }) } - case "result": - last = &ev } + case "result": + s.last = &ev } - if last == nil { +} + +func (s *cursorStream) Finish() error { + if s.last == nil { return fmt.Errorf("phase: missing terminal result event in stream-json output") } - if last.Subtype != "success" { - return fmt.Errorf("phase: result subtype %q, want success", last.Subtype) + if s.last.Subtype != "success" { + return fmt.Errorf("phase: result subtype %q, want success", s.last.Subtype) } - if sink != nil { + if s.sink != nil { end := observe.Event{ Kind: observe.KindPhaseEnd, Outcome: observe.OutcomeSuccess, - DurationMS: last.DurationMS, - ToolCount: toolCount, + DurationMS: s.last.DurationMS, + ToolCount: s.toolCount, } - if last.Usage != nil { + if s.last.Usage != nil { end.Tokens = &observe.TokenCounts{ - Input: last.Usage.InputTokens, - Output: last.Usage.OutputTokens, - CacheRead: last.Usage.CacheReadTokens, - CacheWrite: last.Usage.CacheWriteTokens, + Input: s.last.Usage.InputTokens, + Output: s.last.Usage.OutputTokens, + CacheRead: s.last.Usage.CacheReadTokens, + CacheWrite: s.last.Usage.CacheWriteTokens, } } - sink.Emit(end) + s.sink.Emit(end) } return nil } diff --git a/internal/agent/cursor_test.go b/internal/agent/cursor_test.go index c26c466..f21ff89 100644 --- a/internal/agent/cursor_test.go +++ b/internal/agent/cursor_test.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "strings" + "sync" "testing" "time" @@ -83,12 +84,51 @@ func TestCursor_RunPhase_surfacesStderrOnNonZeroExit(t *testing.T) { } } -func TestCursor_RunPhase_timeoutKillsHungAgent(t *testing.T) { +func TestCursor_RunPhase_streamsToolEventsBeforeProcessExit(t *testing.T) { bin, _ := writeFakeAgent(t, fakeAgentConfig{ - exitCode: 0, - stdout: `{"type":"result","subtype":"success"}` + "\n", - sleep: 2 * time.Second, + exitCode: 0, + stdoutBeforeSleep: `{"type":"tool_call","subtype":"completed","call_id":"c1","duration_ms":42,"tool_call":{"readToolCall":{"result":{"success":{}}}}}` + "\n", + stdout: `{"type":"result","subtype":"success"}` + "\n", + sleep: 350 * time.Millisecond, + }) + sink := &recordingSink{} + c := agent.Cursor{Bin: bin} + done := make(chan error, 1) + + go func() { + done <- c.RunPhase(context.Background(), agent.PhaseRequest{ + Prompt: "work slowly", + Workspace: t.TempDir(), + Timeout: 2 * time.Second, + Events: sink, + }) + }() + + deadline := time.After(200 * time.Millisecond) + for { + if len(sink.tools()) == 1 { + break + } + select { + case err := <-done: + t.Fatalf("RunPhase returned before live event assertion: %v", err) + case <-deadline: + t.Fatalf("tool events = %d, want live event before process exit", len(sink.tools())) + case <-time.After(10 * time.Millisecond): + } + } + if err := <-done; err != nil { + t.Fatalf("RunPhase: %v", err) + } +} + +func TestCursor_RunPhase_timeoutPreservesLiveEventsAndNamesDeadline(t *testing.T) { + bin, _ := writeFakeAgent(t, fakeAgentConfig{ + exitCode: 0, + stdoutBeforeSleep: `{"type":"tool_call","subtype":"completed","call_id":"c1","duration_ms":9,"tool_call":{"shellToolCall":{"result":{"success":{}}}}}` + "\n", + sleep: 5 * time.Second, }) + sink := &recordingSink{} c := agent.Cursor{Bin: bin} start := time.Now() @@ -96,16 +136,47 @@ func TestCursor_RunPhase_timeoutKillsHungAgent(t *testing.T) { Prompt: "hang", Workspace: t.TempDir(), Timeout: 200 * time.Millisecond, + Events: sink, }) elapsed := time.Since(start) if err == nil { t.Fatal("RunPhase: want timeout error") } - if !strings.Contains(err.Error(), "deadline exceeded") && !strings.Contains(err.Error(), "canceled") { - t.Fatalf("RunPhase error = %q, want context timeout", err) + for _, want := range []string{"timeout", "200ms"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("RunPhase error = %q, want %q", err, want) + } } if elapsed > time.Second { - t.Fatalf("RunPhase took %v, want kill near timeout", elapsed) + t.Fatalf("RunPhase took %v, want return near timeout", elapsed) + } + tools := sink.tools() + if len(tools) != 1 || tools[0].Name != "Shell" || tools[0].DurationMS != 9 { + t.Fatalf("live tool events = %+v, want Shell event preserved after timeout", tools) + } +} + +func TestCursor_RunPhase_timeoutKillsHungAgentProcessGroup(t *testing.T) { + bin, _ := writeFakeAgent(t, fakeAgentConfig{ + exitCode: 0, + stdoutBeforeSleep: `{"type":"tool_call","subtype":"completed","call_id":"c1","duration_ms":1,"tool_call":{"shellToolCall":{"result":{"success":{}}}}}` + "\n", + spawnChildKeepsStdout: true, + sleep: 5 * time.Second, + }) + c := agent.Cursor{Bin: bin} + + start := time.Now() + err := c.RunPhase(context.Background(), agent.PhaseRequest{ + Prompt: "child hangs", + Workspace: t.TempDir(), + Timeout: 200 * time.Millisecond, + }) + elapsed := time.Since(start) + if err == nil { + t.Fatal("RunPhase: want timeout error") + } + if elapsed > time.Second { + t.Fatalf("RunPhase took %v, want process group cleanup near timeout", elapsed) } } @@ -292,10 +363,12 @@ func TestCursor_RunPhase_emitsToolErrorStatusFromFailedToolResult(t *testing.T) } type fakeAgentConfig struct { - exitCode int - stdout string - stderr string - sleep time.Duration + exitCode int + stdout string + stdoutBeforeSleep string + stderr string + sleep time.Duration + spawnChildKeepsStdout bool } type fakeCapture struct { @@ -341,6 +414,9 @@ func writeFakeAgent(t *testing.T, cfg fakeAgentConfig) (string, fakeCapture) { if err := os.WriteFile(filepath.Join(dir, "stdout"), []byte(cfg.stdout), 0o644); err != nil { t.Fatalf("write stdout fixture: %v", err) } + if err := os.WriteFile(filepath.Join(dir, "stdout_before_sleep"), []byte(cfg.stdoutBeforeSleep), 0o644); err != nil { + t.Fatalf("write pre-sleep stdout fixture: %v", err) + } if err := os.WriteFile(filepath.Join(dir, "stderr"), []byte(cfg.stderr), 0o644); err != nil { t.Fatalf("write stderr fixture: %v", err) } @@ -351,9 +427,20 @@ func writeFakeAgent(t *testing.T, cfg fakeAgentConfig) (string, fakeCapture) { b.WriteString("printf '%s\\n' \"$@\" > \"$dir/args\"\n") b.WriteString("pwd > \"$dir/cwd\"\n") b.WriteString("cat > \"$dir/stdin\"\n") + if cfg.stdoutBeforeSleep != "" { + b.WriteString("cat \"$dir/stdout_before_sleep\"\n") + } + if cfg.spawnChildKeepsStdout { + fmt.Fprintf(&b, "sleep %g &\n", cfg.sleep.Seconds()) + b.WriteString("wait\n") + } if cfg.sleep > 0 { - // exec so CommandContext kill targets the sleeper, not a parent shell - fmt.Fprintf(&b, "exec sleep %g\n", cfg.sleep.Seconds()) + if cfg.stdoutBeforeSleep != "" || cfg.spawnChildKeepsStdout { + fmt.Fprintf(&b, "sleep %g\n", cfg.sleep.Seconds()) + } else { + // exec so CommandContext kill targets the sleeper, not a parent shell + fmt.Fprintf(&b, "exec sleep %g\n", cfg.sleep.Seconds()) + } } b.WriteString("cat \"$dir/stderr\" >&2\n") b.WriteString("cat \"$dir/stdout\"\n") @@ -401,14 +488,19 @@ func assertNoFlag(t *testing.T, args []string, flag string) { } type recordingSink struct { + mu sync.Mutex events []observe.Event } func (s *recordingSink) Emit(e observe.Event) { + s.mu.Lock() + defer s.mu.Unlock() s.events = append(s.events, e) } func (s *recordingSink) tools() []observe.Event { + s.mu.Lock() + defer s.mu.Unlock() var out []observe.Event for _, e := range s.events { if e.Kind == observe.KindTool { @@ -419,6 +511,8 @@ func (s *recordingSink) tools() []observe.Event { } func (s *recordingSink) phaseEnd() *observe.Event { + s.mu.Lock() + defer s.mu.Unlock() for i := len(s.events) - 1; i >= 0; i-- { if s.events[i].Kind == observe.KindPhaseEnd { e := s.events[i] diff --git a/internal/agent/pi.go b/internal/agent/pi.go index a3bef7a..662bd87 100644 --- a/internal/agent/pi.go +++ b/internal/agent/pi.go @@ -27,12 +27,6 @@ func (p Pi) bin() string { // RunPhase spawns a fresh headless Pi process for one Phase. func (p Pi) RunPhase(ctx context.Context, req PhaseRequest) error { - if req.Timeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, req.Timeout) - defer cancel() - } - args := []string{ "-p", "--mode", "json", @@ -43,30 +37,10 @@ func (p Pi) RunPhase(ctx context.Context, req PhaseRequest) error { args = append(args, "--model", req.Model) } - cmd := exec.CommandContext(ctx, p.bin(), args...) // #nosec G204 -- Pi agent CLI; args built by Ship + cmd := exec.Command(p.bin(), args...) // #nosec G204 -- Pi agent CLI; args built by Ship cmd.Dir = req.Workspace cmd.Stdin = strings.NewReader(req.Prompt) - - var stdout, stderr bytes.Buffer - cmd.Stdout = &stdout - cmd.Stderr = &stderr - - err := cmd.Run() - if err != nil { - if ctx.Err() != nil { - return fmt.Errorf("phase: %w", ctx.Err()) - } - msg := strings.TrimSpace(stderr.String()) - if msg == "" { - msg = err.Error() - } - return fmt.Errorf("phase: %s", msg) - } - - if err := consumePiJSON(stdout.Bytes(), req.Events); err != nil { - return err - } - return nil + return runStreamedPhase(ctx, req.Timeout, cmd, newPiStream(req.Events)) } type piEvent struct { @@ -90,52 +64,64 @@ type piUsage struct { func consumePiJSON(stdout []byte, sink observe.Sink) error { lines := bytes.Split(stdout, []byte("\n")) - sawEnd := false - toolCount := 0 - var tokens *observe.TokenCounts + stream := newPiStream(sink) for _, line := range lines { - line = bytes.TrimSpace(line) - if len(line) == 0 { - continue - } - var ev piEvent - if err := json.Unmarshal(line, &ev); err != nil { - continue - } - switch ev.Type { - case "tool_execution_end": - toolCount++ - if sink != nil { - status := observe.ToolOK - if ev.IsError { - status = observe.ToolError - } - name := ev.ToolName - if name == "" { - name = "unknown" - } - sink.Emit(observe.Event{ - Kind: observe.KindTool, - Name: name, - Status: status, - }) + stream.ProcessLine(line) + } + return stream.Finish() +} + +type piStream struct { + sink observe.Sink + sawEnd bool + toolCount int + tokens *observe.TokenCounts +} + +func newPiStream(sink observe.Sink) *piStream { + return &piStream{sink: sink} +} + +func (s *piStream) ProcessLine(line []byte) { + line = bytes.TrimSpace(line) + if len(line) == 0 { + return + } + var ev piEvent + if err := json.Unmarshal(line, &ev); err != nil { + return + } + switch ev.Type { + case "tool_execution_end": + s.toolCount++ + if s.sink != nil { + status := observe.ToolOK + if ev.IsError { + status = observe.ToolError + } + name := ev.ToolName + if name == "" { + name = "unknown" } - case "agent_end": - sawEnd = true - tokens = sumPiUsage(ev.Messages) + s.sink.Emit(observe.Event{Kind: observe.KindTool, Name: name, Status: status}) } + case "agent_end": + s.sawEnd = true + s.tokens = sumPiUsage(ev.Messages) } - if !sawEnd { +} + +func (s *piStream) Finish() error { + if !s.sawEnd { return fmt.Errorf("phase: missing terminal agent_end event in json output") } - if sink != nil { - end := observe.Event{ + if s.sink != nil { + s.sink.Emit(observe.Event{ Kind: observe.KindPhaseEnd, Outcome: observe.OutcomeSuccess, - ToolCount: toolCount, - Tokens: tokens, - } - sink.Emit(end) + ToolCount: s.toolCount, + Tokens: s.tokens, + }) } return nil } diff --git a/internal/agent/process_unix.go b/internal/agent/process_unix.go new file mode 100644 index 0000000..3763330 --- /dev/null +++ b/internal/agent/process_unix.go @@ -0,0 +1,28 @@ +//go:build !windows + +package agent + +import ( + "os" + "os/exec" + "syscall" + "time" +) + +func setProcessGroup(cmd *exec.Cmd) { + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} +} + +func terminateProcessGroup(proc *os.Process) { + if proc == nil { + return + } + pgid, err := syscall.Getpgid(proc.Pid) + if err != nil { + _ = proc.Kill() + return + } + _ = syscall.Kill(-pgid, syscall.SIGTERM) + time.Sleep(100 * time.Millisecond) + _ = syscall.Kill(-pgid, syscall.SIGKILL) +} diff --git a/internal/agent/process_windows.go b/internal/agent/process_windows.go new file mode 100644 index 0000000..a1c138c --- /dev/null +++ b/internal/agent/process_windows.go @@ -0,0 +1,16 @@ +//go:build windows + +package agent + +import ( + "os" + "os/exec" +) + +func setProcessGroup(cmd *exec.Cmd) {} + +func terminateProcessGroup(proc *os.Process) { + if proc != nil { + _ = proc.Kill() + } +} diff --git a/internal/agent/supervisor.go b/internal/agent/supervisor.go new file mode 100644 index 0000000..bffa0ef --- /dev/null +++ b/internal/agent/supervisor.go @@ -0,0 +1,136 @@ +package agent + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "io/fs" + "os/exec" + "strings" + "sync" + "time" +) + +type streamProcessor interface { + ProcessLine([]byte) + Finish() error +} + +func runStreamedPhase(ctx context.Context, timeout time.Duration, cmd *exec.Cmd, stream streamProcessor) error { + var cancel context.CancelFunc + if timeout > 0 { + ctx, cancel = context.WithTimeout(ctx, timeout) + } else { + ctx, cancel = context.WithCancel(ctx) + } + defer cancel() + + stdout, err := cmd.StdoutPipe() + if err != nil { + return fmt.Errorf("phase: stdout pipe: %w", err) + } + stderr, err := cmd.StderrPipe() + if err != nil { + return fmt.Errorf("phase: stderr pipe: %w", err) + } + + setProcessGroup(cmd) + if err := cmd.Start(); err != nil { + return fmt.Errorf("phase: start: %w", err) + } + + var wg sync.WaitGroup + var recentStderr recentLines + var stdoutErr, stderrErr error + wg.Add(2) + go func() { + defer wg.Done() + stdoutErr = scanLines(stdout, stream.ProcessLine) + }() + go func() { + defer wg.Done() + stderrErr = scanLines(stderr, func(line []byte) { + recentStderr.Add(string(line)) + }) + }() + + waitCh := make(chan error, 1) + go func() { + waitCh <- cmd.Wait() + }() + + var waitErr error + select { + case waitErr = <-waitCh: + case <-ctx.Done(): + terminateProcessGroup(cmd.Process) + waitErr = <-waitCh + wg.Wait() + return timeoutError(timeout, ctx.Err(), recentStderr.String()) + } + wg.Wait() + + if stdoutErr != nil { + return fmt.Errorf("phase: stdout: %w", stdoutErr) + } + if stderrErr != nil { + return fmt.Errorf("phase: stderr: %w", stderrErr) + } + if waitErr != nil { + msg := strings.TrimSpace(recentStderr.String()) + if msg == "" { + msg = waitErr.Error() + } + return fmt.Errorf("phase: %s", msg) + } + return stream.Finish() +} + +func scanLines(r io.Reader, fn func([]byte)) error { + s := bufio.NewScanner(r) + buf := make([]byte, 0, 64*1024) + s.Buffer(buf, 1024*1024) + for s.Scan() { + line := append([]byte(nil), s.Bytes()...) + fn(line) + } + if err := s.Err(); err != nil && !errors.Is(err, fs.ErrClosed) && !strings.Contains(err.Error(), "file already closed") { + return err + } + return nil +} + +func timeoutError(timeout time.Duration, cause error, stderr string) error { + msg := fmt.Sprintf("phase: timeout after %s: %v", timeout, cause) + stderr = strings.TrimSpace(stderr) + if stderr != "" { + msg += "; recent stderr: " + stderr + } + return fmt.Errorf("%s", msg) +} + +type recentLines struct { + mu sync.Mutex + lines []string +} + +func (r *recentLines) Add(line string) { + line = strings.TrimSpace(line) + if line == "" { + return + } + r.mu.Lock() + defer r.mu.Unlock() + r.lines = append(r.lines, line) + if len(r.lines) > 10 { + r.lines = r.lines[len(r.lines)-10:] + } +} + +func (r *recentLines) String() string { + r.mu.Lock() + defer r.mu.Unlock() + return strings.Join(r.lines, "\n") +} diff --git a/internal/run/shipfile.go b/internal/run/shipfile.go index 9374614..a851dce 100644 --- a/internal/run/shipfile.go +++ b/internal/run/shipfile.go @@ -108,6 +108,9 @@ func LoadConfig(dir string) (Config, error) { if err != nil { return Config{}, fmt.Errorf("%s: timeout: invalid duration %q", rel, *raw.Timeout) } + if timeout <= 0 { + return Config{}, fmt.Errorf("%s: timeout must be positive", rel) + } return Config{ Branch: *raw.Branch, diff --git a/internal/run/shipfile_test.go b/internal/run/shipfile_test.go index fc8c183..161ff70 100644 --- a/internal/run/shipfile_test.go +++ b/internal/run/shipfile_test.go @@ -156,6 +156,29 @@ timeout: 10m } } +func TestLoadConfig_rejectsNonPositiveTimeout(t *testing.T) { + for _, timeout := range []string{"0s", "-1s"} { + t.Run(timeout, func(t *testing.T) { + dir := t.TempDir() + content := `branch: "" +agent: cursor +model: "" +max_iterations: 10 +timeout: ` + timeout + ` +` + writeShipConfig(t, dir, content) + + _, err := run.LoadConfig(dir) + if err == nil { + t.Fatal("LoadConfig: want error for non-positive timeout") + } + if !strings.Contains(err.Error(), "timeout must be positive") { + t.Fatalf("LoadConfig error = %q, want positive timeout message", err) + } + }) + } +} + func TestLoadConfig_ignoresUnknownKeys(t *testing.T) { dir := t.TempDir() content := `branch: "" From d7274c68794697de14aa9f989cc70babf6cbacab Mon Sep 17 00:00:00 2001 From: maxBRT Date: Sat, 25 Jul 2026 15:31:02 +0000 Subject: [PATCH 2/2] fix: drain agent streams before wait --- internal/agent/claude.go | 9 --------- internal/agent/codex.go | 9 --------- internal/agent/cursor.go | 9 --------- internal/agent/pi.go | 9 --------- internal/agent/supervisor.go | 11 ++++++++--- 5 files changed, 8 insertions(+), 39 deletions(-) diff --git a/internal/agent/claude.go b/internal/agent/claude.go index 4470061..8faed74 100644 --- a/internal/agent/claude.go +++ b/internal/agent/claude.go @@ -71,15 +71,6 @@ type claudeUsage struct { CacheCreationInputTokens int64 `json:"cache_creation_input_tokens"` } -func consumeClaudeStreamJSON(stdout []byte, sink observe.Sink) error { - lines := bytes.Split(stdout, []byte("\n")) - stream := newClaudeStream(sink) - for _, line := range lines { - stream.ProcessLine(line) - } - return stream.Finish() -} - type claudeStream struct { sink observe.Sink last *claudeEvent diff --git a/internal/agent/codex.go b/internal/agent/codex.go index e094a42..48b94fb 100644 --- a/internal/agent/codex.go +++ b/internal/agent/codex.go @@ -66,15 +66,6 @@ type codexUsage struct { ReasoningOutputTokens int64 `json:"reasoning_output_tokens"` } -func consumeCodexJSON(stdout []byte, sink observe.Sink) error { - lines := bytes.Split(stdout, []byte("\n")) - stream := newCodexStream(sink) - for _, line := range lines { - stream.ProcessLine(line) - } - return stream.Finish() -} - type codexStream struct { sink observe.Sink sawCompleted bool diff --git a/internal/agent/cursor.go b/internal/agent/cursor.go index e665e5f..c70bac1 100644 --- a/internal/agent/cursor.go +++ b/internal/agent/cursor.go @@ -62,15 +62,6 @@ type streamUsage struct { CacheWriteTokens int64 `json:"cacheWriteTokens"` } -func consumeStreamJSON(stdout []byte, sink observe.Sink) error { - lines := bytes.Split(stdout, []byte("\n")) - stream := newCursorStream(sink) - for _, line := range lines { - stream.ProcessLine(line) - } - return stream.Finish() -} - type cursorStream struct { sink observe.Sink last *streamEvent diff --git a/internal/agent/pi.go b/internal/agent/pi.go index 662bd87..da46f8a 100644 --- a/internal/agent/pi.go +++ b/internal/agent/pi.go @@ -62,15 +62,6 @@ type piUsage struct { CacheWrite int64 `json:"cacheWrite"` } -func consumePiJSON(stdout []byte, sink observe.Sink) error { - lines := bytes.Split(stdout, []byte("\n")) - stream := newPiStream(sink) - for _, line := range lines { - stream.ProcessLine(line) - } - return stream.Finish() -} - type piStream struct { sink observe.Sink sawEnd bool diff --git a/internal/agent/supervisor.go b/internal/agent/supervisor.go index bffa0ef..e1e456d 100644 --- a/internal/agent/supervisor.go +++ b/internal/agent/supervisor.go @@ -56,8 +56,14 @@ func runStreamedPhase(ctx context.Context, timeout time.Duration, cmd *exec.Cmd, }) }() + readCh := make(chan struct{}) + go func() { + wg.Wait() + close(readCh) + }() waitCh := make(chan error, 1) go func() { + <-readCh waitCh <- cmd.Wait() }() @@ -66,11 +72,10 @@ func runStreamedPhase(ctx context.Context, timeout time.Duration, cmd *exec.Cmd, case waitErr = <-waitCh: case <-ctx.Done(): terminateProcessGroup(cmd.Process) - waitErr = <-waitCh - wg.Wait() + <-waitCh return timeoutError(timeout, ctx.Err(), recentStderr.String()) } - wg.Wait() + <-readCh if stdoutErr != nil { return fmt.Errorf("phase: stdout: %w", stdoutErr)