Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
fafa0be
autoresearch: compact idle/dead pills to 4-char names in strip
bearmug Mar 17, 2026
a1d528d
autoresearch: attention-first ordering in unified strip
bearmug Mar 17, 2026
f53485e
autoresearch: filter terminal PRs, show compact done count
bearmug Mar 17, 2026
dda4fb2
autoresearch: badge-style pending alert in status bar
bearmug Mar 17, 2026
ba78e50
autoresearch: remove background from passive pills for visual hierarchy
bearmug Mar 17, 2026
40e87ee
autoresearch: mini fleet map line above session zoom
bearmug Mar 17, 2026
df459fb
autoresearch: badge-style failing PR alert in status bar
bearmug Mar 17, 2026
62d410b
autoresearch: merge readiness summary line at top of PR zoom
bearmug Mar 17, 2026
ed04147
autoresearch: show session index/total in fleet map line
bearmug Mar 17, 2026
b6707ba
autoresearch: compact PR pills — title only for critical/selected state
bearmug Mar 17, 2026
390b7ca
autoresearch: group queue tools by safety level (destructive vs safe)
bearmug Mar 17, 2026
efa75f9
autoresearch: show oldest-pending age next to pending badge
bearmug Mar 17, 2026
468157a
autoresearch: done-state treatment for merged/closed PR zoom
bearmug Mar 17, 2026
97a65c5
autoresearch: enrich overflow indicator with active-session state bre…
bearmug Mar 17, 2026
87fc162
autoresearch: PR state breakdown in status bar (3✓ 1✗ 1⏳)
bearmug Mar 17, 2026
ead3430
autoresearch: bold+tinted background for critical unselected PR pills
bearmug Mar 17, 2026
2f2b850
autoresearch: agent-c findings and log (10 cycles)
bearmug Mar 17, 2026
7671cbe
autoresearch: collapse dead sessions to compact count at 8+ session load
bearmug Mar 17, 2026
a05f46d
autoresearch: visual clarity sweep complete (30 cycles, 3 agents)
bearmug Mar 17, 2026
9aaffba
fix: remove duplicate mergeable/method from PR zoom line 2, show repo…
bearmug Mar 17, 2026
90ead62
chore: gitignore .autoresearch/, remove from tracking
bearmug Mar 17, 2026
0eeac9b
Streaming agent output with live timeline updates, cleanup, and tests
bearmug Mar 18, 2026
0fc3174
fix: CI test failure — disable review spawn in TestPoll_MultiplePRs
bearmug Mar 18, 2026
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -20,3 +20,4 @@ node_modules/
# Logs
*.log
.claude/worktrees/
.autoresearch/
190 changes: 183 additions & 7 deletions daemon/internal/pr/agent.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package pr

import (
"bufio"
"bytes"
"context"
"encoding/json"
"fmt"
Expand Down Expand Up @@ -57,6 +59,11 @@ func cloneForAgent(owner, repo, branch string) (string, error) {
return tmpDir, nil
}

// statusInstruction is appended to all agent prompts to get live status updates.
const statusInstruction = "\n\nIMPORTANT: At each major step, print a short status line starting with " +
"\"STATUS: \" (e.g. \"STATUS: reading CI logs\", \"STATUS: found root cause in foo.go\", " +
"\"STATUS: running tests\", \"STATUS: pushing fix\"). These are shown in a live dashboard."

// --- command builders ---

func buildFixCICmd(pr *TrackedPR, workDir string) *exec.Cmd {
Expand All @@ -82,10 +89,11 @@ func buildFixCICmd(pr *TrackedPR, workDir string) *exec.Cmd {
"Do not change test expectations unless the test itself is wrong.",
pr.Number, pr.Owner, pr.Repo, pr.HeadBranch,
strings.Join(failing, "\n"),
)
) + statusInstruction

args := []string{
"-p", prompt,
"--output-format", "stream-json", "--verbose",
"--no-session-persistence",
"--max-budget-usd", "5",
"--model", "sonnet",
Expand Down Expand Up @@ -117,10 +125,11 @@ func buildCodeReviewCmd(pr *TrackedPR, workDir string) *exec.Cmd {
"If the code is clean, output: []\n"+
"Output the JSON array and nothing else.",
pr.HeadBranch, pr.BaseBranch, pr.BaseBranch,
)
) + statusInstruction

args := []string{
"-p", prompt,
"--output-format", "stream-json", "--verbose",
"--no-session-persistence",
"--max-budget-usd", "3",
"--model", "sonnet",
Expand Down Expand Up @@ -151,10 +160,11 @@ func buildFixReviewCmd(pr *TrackedPR, workDir string) *exec.Cmd {
"1. Run tests to verify nothing is broken\n"+
"2. Commit and push to the current branch",
strings.Join(issues, "\n"),
)
) + statusInstruction

args := []string{
"-p", prompt,
"--output-format", "stream-json", "--verbose",
"--no-session-persistence",
"--max-budget-usd", "5",
"--model", "sonnet",
Expand All @@ -175,6 +185,147 @@ func buildFixReviewCmd(pr *TrackedPR, workDir string) *exec.Cmd {
return cmd
}

// --- streaming agent runner ---

// streamEvent is the minimal structure for parsing claude stream-json events.
type streamEvent struct {
Type string `json:"type"`
Message struct {
Content []struct {
Type string `json:"type"`
Text string `json:"text"`
} `json:"content"`
} `json:"message"`
Result string `json:"result"`
CostUSD float64 `json:"total_cost_usd"`
Duration float64 `json:"duration_ms"`
}

// runStreamingAgent runs a claude -p command with stream-json output,
// parsing STATUS: lines and forwarding them to the PR timeline in real-time.
// Returns the final result text and accumulated full output for logging.
func (p *Poller) runStreamingAgent(ctx context.Context, cmd *exec.Cmd, key, agentType string) (result string, allOutput []byte, err error) {
stdout, pipeErr := cmd.StdoutPipe()
if pipeErr != nil {
return "", nil, fmt.Errorf("stdout pipe: %w", pipeErr)
}
// Capture stderr separately for diagnostics.
var stderrBuf bytes.Buffer
cmd.Stderr = &stderrBuf

if startErr := cmd.Start(); startErr != nil {
return "", nil, fmt.Errorf("start: %w", startErr)
}

var fullOutput bytes.Buffer
scanner := bufio.NewScanner(stdout)
// stream-json can have long lines (tool results with file contents).
scanner.Buffer(make([]byte, 0, 256*1024), 1024*1024)

// Live stream log — write every event to a file for debugging.
safe := strings.NewReplacer("/", "-", "#", "-").Replace(key)
streamLogPath := fmt.Sprintf("/tmp/csm-agent-%s-%s-stream.log", safe, agentType)
streamLog, _ := os.Create(streamLogPath)
defer func() {
if streamLog != nil {
streamLog.Close()
}
}()
log.Printf("pr: agent %s stream log: %s", agentType, streamLogPath)

for scanner.Scan() {
line := scanner.Bytes()
fullOutput.Write(line)
fullOutput.WriteByte('\n')
if streamLog != nil {
streamLog.Write(line)
streamLog.WriteString("\n")
streamLog.Sync()
}

var ev streamEvent
if json.Unmarshal(line, &ev) != nil {
continue
}

switch ev.Type {
case "assistant":
// Look for STATUS: lines in assistant text content.
for _, block := range ev.Message.Content {
if block.Type != "text" {
continue
}
for _, textLine := range strings.Split(block.Text, "\n") {
trimmed := strings.TrimSpace(textLine)
if strings.HasPrefix(trimmed, "STATUS:") {
status := strings.TrimSpace(strings.TrimPrefix(trimmed, "STATUS:"))
if status != "" {
p.agentProgress(key, agentType, status)
}
}
}
}
case "result":
result = ev.Result
if ev.CostUSD > 0 {
p.agentCostUpdate(key, ev.CostUSD)
}
}
}

waitErr := cmd.Wait()

// Append stderr to output for logging.
if stderrBuf.Len() > 0 {
fullOutput.WriteString("\n--- stderr ---\n")
fullOutput.Write(stderrBuf.Bytes())
}

return result, fullOutput.Bytes(), waitErr
}

// agentLabel returns a human-friendly label for a timeline prefix.
func agentLabel(agentType string) string {
switch agentType {
case "fix_ci":
return "fix-CI"
case "review":
return "review"
case "fix_review":
return "fix-review"
default:
return agentType
}
}

// agentProgress adds a status update to the PR timeline from a running agent.
func (p *Poller) agentProgress(key, agentType, status string) {
p.mu.Lock()
pr, ok := p.tracked[key]
if ok {
pr.Timeline = append(pr.Timeline, PREvent{
Time: time.Now(), Icon: "⚙",
Message: agentLabel(agentType) + ": " + status,
})
p.save()
}
p.mu.Unlock()

if ok && p.onChange != nil {
p.onChange()
}
}

// agentCostUpdate records the agent cost on the PR.
func (p *Poller) agentCostUpdate(key string, costUSD float64) {
p.mu.Lock()
pr, ok := p.tracked[key]
if ok {
pr.AgentCostUSD += costUSD
}
p.mu.Unlock()
}

// --- spawn functions ---

const agentTimeout = 15 * time.Minute
Expand All @@ -198,7 +349,7 @@ func (p *Poller) spawnFixCI(pr *TrackedPR) {
cmd = exec.CommandContext(ctx, cmd.Path, cmd.Args[1:]...)
cmd.Dir = workDir

output, err := cmd.CombinedOutput()
_, output, err := p.runStreamingAgent(ctx, cmd, key, "fix_ci")
p.agentComplete(key, "fix_ci", err, output)
}()
}
Expand All @@ -222,7 +373,12 @@ func (p *Poller) spawnCodeReview(pr *TrackedPR) {
cmd = exec.CommandContext(ctx, cmd.Path, cmd.Args[1:]...)
cmd.Dir = workDir

output, err := cmd.CombinedOutput()
result, output, err := p.runStreamingAgent(ctx, cmd, key, "review")
// For review, the result field contains the final text output.
// Pass it as output for parseReviewOutput.
if err == nil && result != "" {
output = []byte(result)
}
p.agentComplete(key, "review", err, output)
}()
}
Expand All @@ -246,11 +402,30 @@ func (p *Poller) spawnFixReview(pr *TrackedPR) {
cmd = exec.CommandContext(ctx, cmd.Path, cmd.Args[1:]...)
cmd.Dir = workDir

output, err := cmd.CombinedOutput()
_, output, err := p.runStreamingAgent(ctx, cmd, key, "fix_review")
p.agentComplete(key, "fix_review", err, output)
}()
}

// writeAgentLog writes agent output + error to /tmp/csm-agent-<key>-<type>.log.
// Returns the log path for use in the daemon log line.
func writeAgentLog(key, agentType string, output []byte, runErr error) string {
// Sanitize key for use in filename (replace / and # with -).
safe := strings.NewReplacer("/", "-", "#", "-").Replace(key)
path := fmt.Sprintf("/tmp/csm-agent-%s-%s.log", safe, agentType)
var buf strings.Builder
buf.WriteString(fmt.Sprintf("=== CSM agent log: %s %s ===\n", key, agentType))
buf.WriteString(fmt.Sprintf("error: %v\n", runErr))
buf.WriteString("--- output ---\n")
if len(output) > 0 {
buf.Write(output)
} else {
buf.WriteString("(no output)\n")
}
_ = os.WriteFile(path, []byte(buf.String()), 0o644)
return path
}

// --- completion callback ---

func (p *Poller) agentComplete(key, agentType string, err error, output []byte) {
Expand All @@ -268,7 +443,8 @@ func (p *Poller) agentComplete(key, agentType string, err error, output []byte)
Time: time.Now(), Icon: "✗",
Message: fmt.Sprintf("Agent %s failed: %v", agentType, err),
})
log.Printf("pr: agent %s for %s failed: %v", agentType, key, err)
logFile := writeAgentLog(key, agentType, output, err)
log.Printf("pr: agent %s for %s failed: %v (log: %s)", agentType, key, err, logFile)
} else {
msg := fmt.Sprintf("Agent %s completed", agentType)

Expand Down
101 changes: 101 additions & 0 deletions daemon/internal/pr/agent_test.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package pr

import (
"fmt"
"os"
"strings"
"testing"
Expand Down Expand Up @@ -173,6 +174,106 @@ func TestParseReviewOutput_Empty(t *testing.T) {
}
}

// === stream-json flags ===

func TestBuildFixCICmd_StreamJSON(t *testing.T) {
pr := &TrackedPR{
Owner: "test", Repo: "repo", Number: 1,
HeadBranch: "fix", AutopilotMode: PRAuto,
Checks: []Check{{Name: "ci", Conclusion: "FAILURE"}},
}
args := strings.Join(buildFixCICmd(pr, "/tmp").Args, " ")
if !strings.Contains(args, "--output-format stream-json") {
t.Error("fix_ci should use stream-json output")
}
if !strings.Contains(args, "--verbose") {
t.Error("stream-json requires --verbose")
}
if !strings.Contains(args, "STATUS:") {
t.Error("prompt should contain STATUS instruction")
}
}

func TestBuildCodeReviewCmd_StreamJSON(t *testing.T) {
pr := &TrackedPR{
Owner: "test", Repo: "repo", Number: 1,
HeadBranch: "feat", BaseBranch: "main",
}
args := strings.Join(buildCodeReviewCmd(pr, "/tmp").Args, " ")
if !strings.Contains(args, "--output-format stream-json") {
t.Error("review should use stream-json output")
}
if !strings.Contains(args, "--verbose") {
t.Error("stream-json requires --verbose")
}
}

func TestBuildFixReviewCmd_StreamJSON(t *testing.T) {
pr := &TrackedPR{
Owner: "test", Repo: "repo", Number: 1,
HeadBranch: "fix", AutopilotMode: PRAuto,
ReviewFindings: []ReviewFinding{
{Severity: SeverityCritical, File: "a.go", Message: "bug"},
},
}
args := strings.Join(buildFixReviewCmd(pr, "/tmp").Args, " ")
if !strings.Contains(args, "--output-format stream-json") {
t.Error("fix_review should use stream-json output")
}
if !strings.Contains(args, "--verbose") {
t.Error("stream-json requires --verbose")
}
}

// === agentLabel ===

func TestAgentLabel(t *testing.T) {
cases := []struct{ in, want string }{
{"fix_ci", "fix-CI"},
{"review", "review"},
{"fix_review", "fix-review"},
{"unknown", "unknown"},
}
for _, c := range cases {
if got := agentLabel(c.in); got != c.want {
t.Errorf("agentLabel(%q) = %q, want %q", c.in, got, c.want)
}
}
}

// === writeAgentLog ===

func TestWriteAgentLog_CreatesFile(t *testing.T) {
path := writeAgentLog("test/repo#1", "fix_ci", []byte("some output"), nil)
defer os.Remove(path)

data, err := os.ReadFile(path)
if err != nil {
t.Fatalf("failed to read log: %v", err)
}
s := string(data)
if !strings.Contains(s, "test/repo#1") {
t.Error("log should contain PR key")
}
if !strings.Contains(s, "some output") {
t.Error("log should contain output")
}
}

func TestWriteAgentLog_NoOutput(t *testing.T) {
path := writeAgentLog("test/repo#2", "review", nil, fmt.Errorf("signal: killed"))
defer os.Remove(path)

data, _ := os.ReadFile(path)
s := string(data)
if !strings.Contains(s, "signal: killed") {
t.Error("log should contain error")
}
if !strings.Contains(s, "(no output)") {
t.Error("log should indicate no output")
}
}

// === cloneForAgent (mock test) ===

func TestCloneForAgent_BadRepo(t *testing.T) {
Expand Down
Loading
Loading