From 64f4962f90c2cb39104a7579d22c8b062e3a1248 Mon Sep 17 00:00:00 2001 From: whitmo Date: Mon, 2 Mar 2026 17:11:49 -0800 Subject: [PATCH] feat: Track PR outcomes end-to-end in task history Workers can now report their PR URL via `agent complete --pr-url `, which gets persisted in task history with parsed PR number. A new daemon loop (every 5 min) polls GitHub for open/unknown task history entries and updates their status (merged/closed/open) via the existing but previously unused UpdateTaskHistoryStatus method. The CLI also uses stored terminal statuses (merged/closed) to avoid redundant GitHub API calls at display time. Co-Authored-By: Claude Opus 4.6 --- docs/extending/SOCKET_API.md | 6 +- docs/extending/STATE_FILE_INTEGRATION.md | 3 +- internal/cli/cli.go | 44 +++++- internal/daemon/daemon.go | 180 ++++++++++++++++++++++- internal/daemon/utils_test.go | 146 ++++++++++++++++++ internal/state/state.go | 21 +++ internal/state/state_test.go | 67 +++++++++ 7 files changed, 454 insertions(+), 13 deletions(-) diff --git a/docs/extending/SOCKET_API.md b/docs/extending/SOCKET_API.md index 5d663a7b..b7bca653 100644 --- a/docs/extending/SOCKET_API.md +++ b/docs/extending/SOCKET_API.md @@ -46,7 +46,7 @@ Each command below matches a `case` in `handleRequest`. | `add_agent` | Register an agent in state | `repo`, `name`, `type`, `worktree_path`, `tmux_window`, `session_id`, `pid` | | `remove_agent` | Remove agent from state | `repo`, `name` | | `list_agents` | List agents for a repo | `repo` | -| `complete_agent` | Mark agent ready for cleanup | `repo`, `name`, `summary`, `failure_reason` | +| `complete_agent` | Mark agent ready for cleanup | `repo`, `name`, `summary`, `failure_reason`, `pr_url` | | `restart_agent` | Restart a persistent agent | `repo`, `name` | | `trigger_cleanup` | Force cleanup cycle | none | | `repair_state` | Run state repair routine | none | @@ -540,7 +540,8 @@ class MulticlaudeClient { "repo": "my-app", "name": "clever-fox", "summary": "Added JWT authentication with refresh tokens", - "failure_reason": "" + "failure_reason": "", + "pr_url": "https://github.com/owner/my-app/pull/42" } } ``` @@ -550,6 +551,7 @@ class MulticlaudeClient { - `name` (string, required): Agent name - `summary` (string, optional): Completion summary - `failure_reason` (string, optional): Failure reason (if task failed) +- `pr_url` (string, optional): URL of the pull request created by the worker **Response:** ```json diff --git a/docs/extending/STATE_FILE_INTEGRATION.md b/docs/extending/STATE_FILE_INTEGRATION.md index a6026d4b..91caade2 100644 --- a/docs/extending/STATE_FILE_INTEGRATION.md +++ b/docs/extending/STATE_FILE_INTEGRATION.md @@ -2,7 +2,7 @@ - + @@ -50,6 +50,7 @@ The daemon persists state to `~/.multiclaude/state.json` and writes it atomicall "task": "Implement feature X", // Only for workers "summary": "Added auth module", // Only for workers (completion summary) "failure_reason": "Tests failed", // Only for workers (if task failed) + "pr_url": "https://github.com/user/repo/pull/42", // Only for workers (PR URL if created) "created_at": "2024-01-15T10:30:00Z", "last_nudge": "2024-01-15T10:35:00Z", "ready_for_cleanup": false // Only for workers (signals completion) diff --git a/internal/cli/cli.go b/internal/cli/cli.go index 3e06628b..5d1b8329 100644 --- a/internal/cli/cli.go +++ b/internal/cli/cli.go @@ -565,7 +565,7 @@ func (c *CLI) registerCommands() { agentCmd.Subcommands["complete"] = &Command{ Name: "complete", Description: "Signal worker completion", - Usage: "multiclaude agent complete [--summary ] [--failure ]", + Usage: "multiclaude agent complete [--summary ] [--failure ] [--pr-url ]", Run: c.completeWorker, } @@ -2702,9 +2702,11 @@ func (c *CLI) showHistory(args []string) error { // Try to get PR status from GitHub if we have a branch prStatus, prLink := c.getPRStatusForBranch(repoPath, branch, prURL) - // Use stored status if it indicates failure + // Use stored status if it indicates failure or a terminal state if storedStatus == "failed" { prStatus = "failed" + } else if storedStatus == "merged" || storedStatus == "closed" { + prStatus = storedStatus } // Apply status filter @@ -2814,12 +2816,34 @@ func (c *CLI) showHistory(args []string) error { // getPRStatusForBranch queries GitHub for the PR status of a branch func (c *CLI) getPRStatusForBranch(repoPath, branch, existingPRURL string) (status, prLink string) { - // If we already have a PR URL, just return it formatted + // If we already have a PR URL, query its status directly if existingPRURL != "" { - // Extract PR number from URL for shorter display - parts := strings.Split(existingPRURL, "/") - if len(parts) > 0 { - prNum := parts[len(parts)-1] + // Extract PR number from URL + parts := strings.Split(strings.TrimRight(existingPRURL, "/"), "/") + prNum := "" + if len(parts) >= 2 && parts[len(parts)-2] == "pull" { + prNum = parts[len(parts)-1] + } + + // Try to get live status from GitHub + if prNum != "" { + cmd := exec.Command("gh", "pr", "view", prNum, "--json", "state") + cmd.Dir = repoPath + if output, err := cmd.Output(); err == nil { + var pr struct { + State string `json:"state"` + } + if json.Unmarshal(output, &pr) == nil { + switch strings.ToUpper(pr.State) { + case "MERGED": + return "merged", "#" + prNum + case "OPEN": + return "open", "#" + prNum + case "CLOSED": + return "closed", "#" + prNum + } + } + } return "unknown", "#" + prNum } return "unknown", existingPRURL @@ -4228,6 +4252,12 @@ func (c *CLI) completeWorker(args []string) error { fmt.Printf("Failure reason: %s\n", failureReason) } + // Add optional PR URL + if prURL, ok := flags["pr-url"]; ok && prURL != "" { + reqArgs["pr_url"] = prURL + fmt.Printf("PR URL: %s\n", prURL) + } + client := socket.NewClient(c.paths.DaemonSock) resp, err := client.Send(socket.Request{ Command: "complete_agent", diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go index c7554129..e722f194 100644 --- a/internal/daemon/daemon.go +++ b/internal/daemon/daemon.go @@ -2,11 +2,13 @@ package daemon import ( "context" + "encoding/json" "fmt" "log" "os" "os/exec" "path/filepath" + "strconv" "strings" "sync" "syscall" @@ -106,12 +108,13 @@ func (d *Daemon) Start() error { d.restoreTrackedRepos() // Start core loops after restore completes - d.wg.Add(5) + d.wg.Add(6) go d.healthCheckLoop() go d.messageRouterLoop() go d.wakeLoop() go d.serverLoop() go d.worktreeRefreshLoop() + go d.prOutcomeTrackingLoop() return nil } @@ -611,6 +614,149 @@ func (d *Daemon) TriggerWorktreeRefresh() { d.refreshWorktrees() } +// prOutcomeTrackingLoop periodically checks pending task history entries for PR outcome changes +func (d *Daemon) prOutcomeTrackingLoop() { + defer d.wg.Done() + d.logger.Info("Starting PR outcome tracking loop") + + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + + // Run once after a short delay on startup + select { + case <-time.After(1 * time.Minute): + d.updatePROutcomes() + case <-d.ctx.Done(): + d.logger.Info("PR outcome tracking loop stopped") + return + } + + for { + select { + case <-ticker.C: + d.updatePROutcomes() + case <-d.ctx.Done(): + d.logger.Info("PR outcome tracking loop stopped") + return + } + } +} + +// TriggerPROutcomeTracking triggers an immediate PR outcome check (for testing) +func (d *Daemon) TriggerPROutcomeTracking() { + d.updatePROutcomes() +} + +// updatePROutcomes checks all pending task history entries and updates their PR status +func (d *Daemon) updatePROutcomes() { + d.logger.Debug("Checking PR outcomes for pending tasks") + + repos := d.state.ListRepos() + for _, repoName := range repos { + pending, err := d.state.GetPendingTaskHistory(repoName) + if err != nil { + d.logger.Debug("Could not get pending task history for %s: %v", repoName, err) + continue + } + + if len(pending) == 0 { + continue + } + + repoPath := d.paths.RepoDir(repoName) + if _, err := os.Stat(repoPath); os.IsNotExist(err) { + continue + } + + for _, entry := range pending { + status, prURL, prNumber := d.checkPRStatus(repoPath, entry) + if status == "" || status == state.TaskStatusUnknown { + continue + } + + // Update the entry if status changed + if status != entry.Status || (prURL != "" && entry.PRURL == "") { + if err := d.state.UpdateTaskHistoryStatus(repoName, entry.Name, status, prURL, prNumber); err != nil { + d.logger.Debug("Could not update task history for %s/%s: %v", repoName, entry.Name, err) + } else { + d.logger.Info("Updated PR outcome for %s/%s: %s (PR #%d)", repoName, entry.Name, status, prNumber) + } + } + } + } +} + +// checkPRStatus queries GitHub for the PR status of a task history entry +func (d *Daemon) checkPRStatus(repoPath string, entry state.TaskHistoryEntry) (state.TaskStatus, string, int) { + // If we already have a PR URL with a number, check by number (faster) + if entry.PRNumber > 0 { + return d.checkPRStatusByNumber(repoPath, entry.PRNumber, entry.PRURL) + } + + // Otherwise query by branch + if entry.Branch == "" { + return "", "", 0 + } + + cmd := exec.Command("gh", "pr", "list", "--head", entry.Branch, "--state", "all", + "--json", "number,state,url", "--limit", "1") + cmd.Dir = repoPath + output, err := cmd.Output() + if err != nil { + return "", "", 0 + } + + var prs []struct { + Number int `json:"number"` + State string `json:"state"` + URL string `json:"url"` + } + if err := json.Unmarshal(output, &prs); err != nil || len(prs) == 0 { + return "", "", 0 + } + + pr := prs[0] + return ghStateToTaskStatus(pr.State), pr.URL, pr.Number +} + +// checkPRStatusByNumber queries GitHub for a specific PR number +func (d *Daemon) checkPRStatusByNumber(repoPath string, prNumber int, prURL string) (state.TaskStatus, string, int) { + cmd := exec.Command("gh", "pr", "view", strconv.Itoa(prNumber), "--json", "state,url") + cmd.Dir = repoPath + output, err := cmd.Output() + if err != nil { + return "", "", 0 + } + + var pr struct { + State string `json:"state"` + URL string `json:"url"` + } + if err := json.Unmarshal(output, &pr); err != nil { + return "", "", 0 + } + + url := prURL + if pr.URL != "" { + url = pr.URL + } + return ghStateToTaskStatus(pr.State), url, prNumber +} + +// ghStateToTaskStatus converts GitHub PR state to TaskStatus +func ghStateToTaskStatus(ghState string) state.TaskStatus { + switch strings.ToUpper(ghState) { + case "MERGED": + return state.TaskStatusMerged + case "OPEN": + return state.TaskStatusOpen + case "CLOSED": + return state.TaskStatusClosed + default: + return state.TaskStatusUnknown + } +} + // handleRequest handles incoming socket requests func (d *Daemon) handleRequest(req socket.Request) socket.Response { d.logger.Debug("Handling request: %s", req.Command) @@ -1047,6 +1193,9 @@ func (d *Daemon) handleCompleteAgent(req socket.Request) socket.Response { if failureReason := getOptionalStringArg(req.Args, "failure_reason", ""); failureReason != "" { agent.FailureReason = failureReason } + if prURL := getOptionalStringArg(req.Args, "pr_url", ""); prURL != "" { + agent.PRURL = prURL + } if err := d.state.UpdateAgent(repoName, agentName, agent); err != nil { return socket.ErrorResponse("%s", err.Error()) @@ -1468,11 +1617,24 @@ func (d *Daemon) recordTaskHistory(repoName, agentName string, agent state.Agent status = state.TaskStatusFailed } + // Extract PR URL and number from agent + prURL := agent.PRURL + prNumber := 0 + if prURL != "" { + prNumber = parsePRNumber(prURL) + // If we have a PR URL and no failure, mark as open + if status != state.TaskStatusFailed { + status = state.TaskStatusOpen + } + } + entry := state.TaskHistoryEntry{ Name: agentName, Task: agent.Task, Branch: branch, - Status: status, // Will be updated when displaying if a PR exists + PRURL: prURL, + PRNumber: prNumber, + Status: status, Summary: agent.Summary, FailureReason: agent.FailureReason, CreatedAt: agent.CreatedAt, @@ -1482,8 +1644,20 @@ func (d *Daemon) recordTaskHistory(repoName, agentName string, agent state.Agent if err := d.state.AddTaskHistory(repoName, entry); err != nil { d.logger.Warn("Failed to record task history for %s: %v", agentName, err) } else { - d.logger.Info("Recorded task history for %s (branch: %s, summary: %q)", agentName, branch, agent.Summary) + d.logger.Info("Recorded task history for %s (branch: %s, pr: %s, summary: %q)", agentName, branch, prURL, agent.Summary) + } +} + +// parsePRNumber extracts the PR number from a GitHub PR URL +// e.g., "https://github.com/owner/repo/pull/123" -> 123 +func parsePRNumber(prURL string) int { + parts := strings.Split(strings.TrimRight(prURL, "/"), "/") + if len(parts) >= 2 && parts[len(parts)-2] == "pull" { + if n, err := strconv.Atoi(parts[len(parts)-1]); err == nil { + return n + } } + return 0 } // handleTaskHistory returns the task history for a repository diff --git a/internal/daemon/utils_test.go b/internal/daemon/utils_test.go index 679db434..39e782f3 100644 --- a/internal/daemon/utils_test.go +++ b/internal/daemon/utils_test.go @@ -370,3 +370,149 @@ func TestIsLogFileEdgeCases(t *testing.T) { }) } } + +func TestRecordTaskHistoryWithPRURL(t *testing.T) { + d, cleanup := setupTestDaemon(t) + defer cleanup() + + // Add a test repo + repo := &state.Repository{ + GithubURL: "https://github.com/test/repo", + TmuxSession: "test-session", + Agents: make(map[string]state.Agent), + } + if err := d.state.AddRepo("test-repo", repo); err != nil { + t.Fatalf("Failed to add repo: %v", err) + } + + // Test recording task with PR URL + agent := state.Agent{ + Type: state.AgentTypeWorker, + WorktreePath: "/tmp/test-worker", + TmuxWindow: "test-window", + Task: "implement feature Y", + Summary: "created PR", + PRURL: "https://github.com/test/repo/pull/42", + CreatedAt: time.Now().Add(-1 * time.Hour), + } + + d.recordTaskHistory("test-repo", "pr-worker", agent) + + // Verify task was recorded with PR info + history, err := d.state.GetTaskHistory("test-repo", 10) + if err != nil { + t.Fatalf("GetTaskHistory() failed: %v", err) + } + + if len(history) == 0 { + t.Fatal("Expected task history entry") + } + + entry := history[0] + if entry.PRURL != "https://github.com/test/repo/pull/42" { + t.Errorf("Entry PRURL = %q, want %q", entry.PRURL, "https://github.com/test/repo/pull/42") + } + if entry.PRNumber != 42 { + t.Errorf("Entry PRNumber = %d, want %d", entry.PRNumber, 42) + } + // With a PR URL and no failure, status should be open + if entry.Status != state.TaskStatusOpen { + t.Errorf("Entry status = %q, want %q", entry.Status, state.TaskStatusOpen) + } +} + +func TestRecordTaskHistoryWithPRURLAndFailure(t *testing.T) { + d, cleanup := setupTestDaemon(t) + defer cleanup() + + // Add a test repo + repo := &state.Repository{ + GithubURL: "https://github.com/test/repo", + TmuxSession: "test-session", + Agents: make(map[string]state.Agent), + } + if err := d.state.AddRepo("test-repo", repo); err != nil { + t.Fatalf("Failed to add repo: %v", err) + } + + // A worker with both PR URL and failure reason - failure takes precedence + agent := state.Agent{ + Type: state.AgentTypeWorker, + TmuxWindow: "test-window", + Task: "broken feature", + PRURL: "https://github.com/test/repo/pull/99", + FailureReason: "CI failed", + CreatedAt: time.Now(), + } + + d.recordTaskHistory("test-repo", "failed-pr-worker", agent) + + history, err := d.state.GetTaskHistory("test-repo", 10) + if err != nil { + t.Fatalf("GetTaskHistory() failed: %v", err) + } + + entry := history[0] + // Failure should take precedence over PR URL + if entry.Status != state.TaskStatusFailed { + t.Errorf("Entry status = %q, want %q", entry.Status, state.TaskStatusFailed) + } + // But PR info should still be recorded + if entry.PRURL != "https://github.com/test/repo/pull/99" { + t.Errorf("Entry PRURL = %q, want %q", entry.PRURL, "https://github.com/test/repo/pull/99") + } + if entry.PRNumber != 99 { + t.Errorf("Entry PRNumber = %d, want %d", entry.PRNumber, 99) + } +} + +func TestParsePRNumber(t *testing.T) { + tests := []struct { + name string + url string + want int + }{ + {"standard GitHub PR URL", "https://github.com/owner/repo/pull/123", 123}, + {"URL with trailing slash", "https://github.com/owner/repo/pull/456/", 456}, + {"PR number 1", "https://github.com/test/repo/pull/1", 1}, + {"large PR number", "https://github.com/test/repo/pull/99999", 99999}, + {"not a PR URL", "https://github.com/owner/repo/issues/123", 0}, + {"empty URL", "", 0}, + {"just a number", "123", 0}, + {"short URL", "pull/42", 42}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := parsePRNumber(tt.url) + if got != tt.want { + t.Errorf("parsePRNumber(%q) = %d, want %d", tt.url, got, tt.want) + } + }) + } +} + +func TestGhStateToTaskStatus(t *testing.T) { + tests := []struct { + ghState string + want state.TaskStatus + }{ + {"MERGED", state.TaskStatusMerged}, + {"OPEN", state.TaskStatusOpen}, + {"CLOSED", state.TaskStatusClosed}, + {"merged", state.TaskStatusMerged}, + {"open", state.TaskStatusOpen}, + {"closed", state.TaskStatusClosed}, + {"unknown", state.TaskStatusUnknown}, + {"", state.TaskStatusUnknown}, + } + + for _, tt := range tests { + t.Run(tt.ghState, func(t *testing.T) { + got := ghStateToTaskStatus(tt.ghState) + if got != tt.want { + t.Errorf("ghStateToTaskStatus(%q) = %q, want %q", tt.ghState, got, tt.want) + } + }) + } +} diff --git a/internal/state/state.go b/internal/state/state.go index 960a1a41..ec220a6b 100644 --- a/internal/state/state.go +++ b/internal/state/state.go @@ -150,6 +150,7 @@ type Agent struct { Task string `json:"task,omitempty"` // Only for workers Summary string `json:"summary,omitempty"` // Brief summary of work done (workers only) FailureReason string `json:"failure_reason,omitempty"` // Why the task failed (workers only) + PRURL string `json:"pr_url,omitempty"` // PR URL created by worker (workers only) CreatedAt time.Time `json:"created_at"` LastNudge time.Time `json:"last_nudge,omitempty"` ReadyForCleanup bool `json:"ready_for_cleanup,omitempty"` // Only for workers @@ -685,6 +686,26 @@ func (s *State) UpdateTaskHistorySummary(repoName, taskName, summary, failureRea return fmt.Errorf("task %q not found in history", taskName) } +// GetPendingTaskHistory returns task history entries with open or unknown status +// that have a branch (and thus might have a PR to check). Used by the PR outcome tracking loop. +func (s *State) GetPendingTaskHistory(repoName string) ([]TaskHistoryEntry, error) { + s.mu.RLock() + defer s.mu.RUnlock() + + repo, exists := s.Repos[repoName] + if !exists { + return nil, fmt.Errorf("repository %q not found", repoName) + } + + var pending []TaskHistoryEntry + for _, entry := range repo.TaskHistory { + if entry.Branch != "" && (entry.Status == TaskStatusOpen || entry.Status == TaskStatusUnknown) { + pending = append(pending, entry) + } + } + return pending, nil +} + // saveUnlocked saves state without acquiring lock (caller must hold lock) func (s *State) saveUnlocked() error { data, err := json.MarshalIndent(s, "", " ") diff --git a/internal/state/state_test.go b/internal/state/state_test.go index 9b68c10f..47fb6ef2 100644 --- a/internal/state/state_test.go +++ b/internal/state/state_test.go @@ -2206,3 +2206,70 @@ func TestGetTaskHistoryNoLimit(t *testing.T) { t.Errorf("GetTaskHistory() with limit=0 returned %d entries, want 5", len(history)) } } + +func TestGetPendingTaskHistory(t *testing.T) { + tmpDir := t.TempDir() + statePath := filepath.Join(tmpDir, "state.json") + + s := New(statePath) + + // Create a repo + repo := &Repository{ + GithubURL: "https://github.com/test/repo", + TmuxSession: "mc-test-repo", + Agents: make(map[string]Agent), + } + if err := s.AddRepo("test-repo", repo); err != nil { + t.Fatalf("AddRepo() failed: %v", err) + } + + // Add entries with different statuses + entries := []TaskHistoryEntry{ + {Name: "worker-1", Task: "Task 1", Branch: "work/worker-1", Status: TaskStatusOpen, CreatedAt: time.Now()}, + {Name: "worker-2", Task: "Task 2", Branch: "work/worker-2", Status: TaskStatusMerged, CreatedAt: time.Now()}, + {Name: "worker-3", Task: "Task 3", Branch: "work/worker-3", Status: TaskStatusUnknown, CreatedAt: time.Now()}, + {Name: "worker-4", Task: "Task 4", Branch: "", Status: TaskStatusUnknown, CreatedAt: time.Now()}, // No branch + {Name: "worker-5", Task: "Task 5", Branch: "work/worker-5", Status: TaskStatusFailed, CreatedAt: time.Now()}, + {Name: "worker-6", Task: "Task 6", Branch: "work/worker-6", Status: TaskStatusClosed, CreatedAt: time.Now()}, + } + + for _, entry := range entries { + if err := s.AddTaskHistory("test-repo", entry); err != nil { + t.Fatalf("AddTaskHistory() failed: %v", err) + } + } + + // Get pending entries - should only return open and unknown with branches + pending, err := s.GetPendingTaskHistory("test-repo") + if err != nil { + t.Fatalf("GetPendingTaskHistory() failed: %v", err) + } + + if len(pending) != 2 { + t.Fatalf("GetPendingTaskHistory() returned %d entries, want 2", len(pending)) + } + + // Should be worker-1 (open) and worker-3 (unknown) + names := map[string]bool{} + for _, entry := range pending { + names[entry.Name] = true + } + if !names["worker-1"] { + t.Error("Expected worker-1 (open) in pending results") + } + if !names["worker-3"] { + t.Error("Expected worker-3 (unknown) in pending results") + } +} + +func TestGetPendingTaskHistoryNonExistentRepo(t *testing.T) { + tmpDir := t.TempDir() + statePath := filepath.Join(tmpDir, "state.json") + + s := New(statePath) + + _, err := s.GetPendingTaskHistory("nonexistent") + if err == nil { + t.Error("GetPendingTaskHistory() should fail for nonexistent repo") + } +}