diff --git a/DOGFOOD.md b/DOGFOOD.md index 121cb57..6302a56 100644 --- a/DOGFOOD.md +++ b/DOGFOOD.md @@ -1,3 +1,3 @@ -This repository is being used to dogfood `sao` against a real GitHub issue. +# Dogfood -The goal is simple: let `sao` discover an issue in this repo, execute the work locally, and validate the end-to-end loop on a small, real task. +This repository is being used to dogfood `sao` against a real GitHub issue and verify the end-to-end local workflow on a small, real task. diff --git a/README.md b/README.md index 1d494e1..ee090cb 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # Simple Agent Orchestration -`sao` is a CLI that watches GitHub repositories for eligible issues, ranks them, and dispatches work to supported coding agents like Codex and Claude. +`sao` is a CLI that watches GitHub repositories for eligible issues, ranks them, and dispatches work to supported coding agents like Codex and Claude through `acpx`. It keeps machine-level config in `~/.config/sao/config.yaml`, repo-level config in `.simple-agent-orchestration.yaml`, and local task state under `~/.local/state/sao/`. @@ -186,5 +186,5 @@ State: ## Notes - This is currently an MVP. -- Dispatch is single-task oriented even though the machine config includes concurrency fields. -- Direct execution is currently implemented for `claude` and `codex`. +- Agent execution is routed through `acpx`. +- Supported agent runtimes today are `codex` and `claude`. diff --git a/internal/sao/sao.go b/internal/sao/sao.go index c7746fd..808bbb6 100644 --- a/internal/sao/sao.go +++ b/internal/sao/sao.go @@ -10,6 +10,7 @@ import ( "path/filepath" "slices" "strings" + "sync" "time" "github.com/nitayr/simple-agent-orchastration/internal/acpx" @@ -359,78 +360,111 @@ func runCycle(ctx context.Context, cfg config.MachineConfig, stdout, stderr io.W return err } - var picked *planner.Candidate - for _, candidate := range candidates { - record, ok := store.Tasks[candidate.Issue.URL] - if ok && !shouldDispatch(record, candidate) { - continue - } - picked = &candidate - break - } - if picked == nil { - fmt.Fprintln(stdout, "no eligible tasks after state filtering") - return nil - } - - agent, err := chooseAgent(cfg, *picked) + plans, err := selectDispatchPlans(cfg, candidates, store) if err != nil { return err } - - fmt.Fprintf(stdout, "dispatching %s #%d with %s\n", picked.Repo.Slug, picked.Issue.Number, agent.Name) + if len(plans) == 0 { + fmt.Fprintln(stdout, "no eligible tasks after state and concurrency filtering") + return nil + } now := time.Now().UTC() - store.Tasks[picked.Issue.URL] = state.TaskRecord{ - IssueURL: picked.Issue.URL, - RepoPath: picked.ProjectPath, - AgentName: agent.Name, - Status: "running", - IssueUpdatedAt: picked.Issue.UpdatedAt, - StartedAt: now, - UpdatedAt: now, + for _, plan := range plans { + fmt.Fprintf(stdout, "dispatching %s #%d with %s\n", plan.Candidate.Repo.Slug, plan.Candidate.Issue.Number, plan.Agent.Name) + store.Tasks[plan.Candidate.Issue.URL] = state.TaskRecord{ + IssueURL: plan.Candidate.Issue.URL, + RepoPath: plan.Candidate.ProjectPath, + AgentName: plan.Agent.Name, + Status: "running", + IssueUpdatedAt: plan.Candidate.Issue.UpdatedAt, + StartedAt: now, + UpdatedAt: now, + } } if err := state.Save(store); err != nil { return err } - prompt := buildPrompt(*picked) - runner := acpx.NewRunner(agent.Command) - agentName, ok := runtimeAgentName(agent) - if !ok { - return fmt.Errorf("agent %q is not supported for direct execution", agent.Name) - } - response, err := runner.Exec(ctx, picked.ProjectPath, agentName, prompt) - if err != nil { - markTaskFailure(store, picked.Issue.URL, picked.Issue.UpdatedAt, err) - _ = state.Save(store) - return err - } + outcomes := make(chan dispatchOutcome, len(plans)) + var wg sync.WaitGroup + for _, plan := range plans { + wg.Add(1) + go func(plan dispatchPlan) { + defer wg.Done() + prompt := buildPrompt(plan.Candidate) + runner := acpx.NewRunner(plan.Agent.Command) + response, err := runner.Exec(ctx, plan.Candidate.ProjectPath, plan.RuntimeName, prompt) + outcomes <- dispatchOutcome{ + Plan: plan, + Response: response, + Err: err, + CompletedAt: time.Now().UTC(), + } + }(plan) + } + go func() { + wg.Wait() + close(outcomes) + }() + + var errs []error + for outcome := range outcomes { + if outcome.Err != nil { + markTaskFailure(store, outcome.Plan.Candidate.Issue.URL, outcome.Plan.Candidate.Issue.UpdatedAt, outcome.CompletedAt, outcome.Err) + fmt.Fprintf( + stderr, + "failed %s #%d with %s: %v\n", + outcome.Plan.Candidate.Repo.Slug, + outcome.Plan.Candidate.Issue.Number, + outcome.Plan.Agent.Name, + outcome.Err, + ) + errs = append(errs, outcome.Err) + } else { + store.Tasks[outcome.Plan.Candidate.Issue.URL] = state.TaskRecord{ + IssueURL: outcome.Plan.Candidate.Issue.URL, + RepoPath: outcome.Plan.Candidate.ProjectPath, + AgentName: outcome.Plan.Agent.Name, + Status: "completed", + IssueUpdatedAt: outcome.Plan.Candidate.Issue.UpdatedAt, + StartedAt: now, + UpdatedAt: outcome.CompletedAt, + CompletedAt: outcome.CompletedAt, + LastResponse: outcome.Response.AssistantText, + } + if outcome.Response.AssistantText != "" { + fmt.Fprintf(stdout, "completed %s #%d\n", outcome.Plan.Candidate.Repo.Slug, outcome.Plan.Candidate.Issue.Number) + fmt.Fprintf(stdout, "agent summary:\n%s\n", outcome.Response.AssistantText) + } else { + fmt.Fprintf(stdout, "completed %s #%d with no summary returned\n", outcome.Plan.Candidate.Repo.Slug, outcome.Plan.Candidate.Issue.Number) + } + } - store.Tasks[picked.Issue.URL] = state.TaskRecord{ - IssueURL: picked.Issue.URL, - RepoPath: picked.ProjectPath, - AgentName: agent.Name, - Status: "completed", - IssueUpdatedAt: picked.Issue.UpdatedAt, - StartedAt: now, - UpdatedAt: time.Now().UTC(), - CompletedAt: time.Now().UTC(), - LastResponse: response.AssistantText, - } - if err := state.Save(store); err != nil { - return err + if err := state.Save(store); err != nil { + errs = append(errs, err) + } } - if response.AssistantText != "" { - fmt.Fprintf(stdout, "completed %s #%d\n", picked.Repo.Slug, picked.Issue.Number) - fmt.Fprintf(stdout, "agent summary:\n%s\n", response.AssistantText) - } else { - fmt.Fprintf(stdout, "completed %s #%d with no summary returned\n", picked.Repo.Slug, picked.Issue.Number) + if len(errs) > 0 { + return errors.Join(errs...) } return nil } +type dispatchPlan struct { + Candidate planner.Candidate + Agent config.InstalledAgent + RuntimeName string +} + +type dispatchOutcome struct { + Plan dispatchPlan + Response acpx.Result + Err error + CompletedAt time.Time +} + func chooseAgent(cfg config.MachineConfig, candidate planner.Candidate) (config.InstalledAgent, error) { for _, name := range candidate.AgentOrder { idx := slices.IndexFunc(cfg.Agents.Installed, func(agent config.InstalledAgent) bool { @@ -451,6 +485,101 @@ func chooseAgent(cfg config.MachineConfig, candidate planner.Candidate) (config. return config.InstalledAgent{}, fmt.Errorf("no enabled agent available for %s #%d", candidate.Repo.Slug, candidate.Issue.Number) } +func selectDispatchPlans(cfg config.MachineConfig, candidates []planner.Candidate, store state.Store) ([]dispatchPlan, error) { + globalLimit := max(cfg.Runtime.MaxConcurrentTasks, 1) + runningTasks := 0 + runningByAgent := map[string]int{} + activeRepos := map[string]struct{}{} + activeIssues := map[string]struct{}{} + for issueURL, record := range store.Tasks { + if record.Status != "running" { + continue + } + runningTasks++ + runningByAgent[record.AgentName]++ + if record.RepoPath != "" { + activeRepos[record.RepoPath] = struct{}{} + } + activeIssues[issueURL] = struct{}{} + } + + availableGlobal := globalLimit - runningTasks + if availableGlobal <= 0 { + return nil, nil + } + + var plans []dispatchPlan + var unavailableErr error + for _, candidate := range candidates { + record, ok := store.Tasks[candidate.Issue.URL] + if ok && !shouldDispatch(record, candidate) { + continue + } + if _, ok := activeIssues[candidate.Issue.URL]; ok { + continue + } + if _, ok := activeRepos[candidate.ProjectPath]; ok { + continue + } + + plan, err := chooseAvailableAgent(cfg, candidate, runningByAgent) + if err != nil { + if unavailableErr == nil { + unavailableErr = err + } + continue + } + + plans = append(plans, plan) + runningByAgent[plan.Agent.Name]++ + activeRepos[candidate.ProjectPath] = struct{}{} + activeIssues[candidate.Issue.URL] = struct{}{} + if len(plans) >= availableGlobal { + break + } + } + + if len(plans) == 0 { + return nil, unavailableErr + } + return plans, nil +} + +func chooseAvailableAgent(cfg config.MachineConfig, candidate planner.Candidate, runningByAgent map[string]int) (dispatchPlan, error) { + for _, name := range candidate.AgentOrder { + idx := slices.IndexFunc(cfg.Agents.Installed, func(agent config.InstalledAgent) bool { + return agent.Name == name && agent.Enabled + }) + if idx < 0 { + continue + } + agent := cfg.Agents.Installed[idx] + if _, err := exec.LookPath(agent.Command[0]); err != nil { + continue + } + runtimeName, ok := runtimeAgentName(agent) + if !ok { + continue + } + if runningByAgent[agent.Name] >= effectiveMaxParallel(agent) { + continue + } + return dispatchPlan{ + Candidate: candidate, + Agent: agent, + RuntimeName: runtimeName, + }, nil + } + return dispatchPlan{}, fmt.Errorf("no enabled agent slot available for %s #%d", candidate.Repo.Slug, candidate.Issue.Number) +} + +func effectiveMaxParallel(agent config.InstalledAgent) int { + if agent.MaxParallel <= 0 { + return 1 + } + return agent.MaxParallel +} + func runtimeAgentName(agent config.InstalledAgent) (string, bool) { if name, ok := acpx.ResolveAgentName(agent.Type); ok { return name, true @@ -486,11 +615,12 @@ Do not ask for orchestration help. Work directly in the repository. `, candidate.ProjectPath, candidate.Repo.Slug, candidate.Issue.Number, candidate.Issue.URL, candidate.Issue.Title, body)) } -func markTaskFailure(store state.Store, issueURL string, issueUpdatedAt time.Time, err error) { +func markTaskFailure(store state.Store, issueURL string, issueUpdatedAt, completedAt time.Time, err error) { record := store.Tasks[issueURL] record.Status = "failed" record.IssueUpdatedAt = issueUpdatedAt - record.UpdatedAt = time.Now().UTC() + record.UpdatedAt = completedAt + record.CompletedAt = completedAt record.LastError = err.Error() store.Tasks[issueURL] = record } @@ -507,3 +637,10 @@ func shouldDispatch(record state.TaskRecord, candidate planner.Candidate) bool { } return true } + +func max(a, b int) int { + if a > b { + return a + } + return b +} diff --git a/internal/sao/sao_test.go b/internal/sao/sao_test.go new file mode 100644 index 0000000..5fea18b --- /dev/null +++ b/internal/sao/sao_test.go @@ -0,0 +1,155 @@ +package sao + +import ( + "testing" + "time" + + "github.com/nitayr/simple-agent-orchastration/internal/config" + "github.com/nitayr/simple-agent-orchastration/internal/gh" + "github.com/nitayr/simple-agent-orchastration/internal/planner" + "github.com/nitayr/simple-agent-orchastration/internal/state" +) + +func TestSelectDispatchPlansHonorsLimits(t *testing.T) { + t.Parallel() + + cfg := config.MachineConfig{ + Runtime: config.MachineRuntime{ + MaxConcurrentTasks: 2, + }, + Agents: config.MachineAgents{ + DefaultOrder: []string{"codex", "claude"}, + Installed: []config.InstalledAgent{ + { + Name: "codex", + Type: "codex", + Command: []string{"sh"}, + Enabled: true, + MaxParallel: 1, + }, + { + Name: "claude", + Type: "claude", + Command: []string{"sh"}, + Enabled: true, + MaxParallel: 1, + }, + }, + }, + } + + now := time.Now().UTC() + candidates := []planner.Candidate{ + { + ProjectPath: "/tmp/repo-a", + Repo: gh.Repository{Slug: "org/repo-a"}, + Issue: gh.Issue{ + Number: 1, + URL: "https://example.com/a/1", + UpdatedAt: now, + }, + AgentOrder: []string{"codex", "claude"}, + }, + { + ProjectPath: "/tmp/repo-b", + Repo: gh.Repository{Slug: "org/repo-b"}, + Issue: gh.Issue{ + Number: 2, + URL: "https://example.com/b/2", + UpdatedAt: now, + }, + AgentOrder: []string{"codex", "claude"}, + }, + { + ProjectPath: "/tmp/repo-c", + Repo: gh.Repository{Slug: "org/repo-c"}, + Issue: gh.Issue{ + Number: 3, + URL: "https://example.com/c/3", + UpdatedAt: now, + }, + AgentOrder: []string{"codex", "claude"}, + }, + } + + plans, err := selectDispatchPlans(cfg, candidates, state.Store{Tasks: map[string]state.TaskRecord{}}) + if err != nil { + t.Fatalf("selectDispatchPlans() error = %v", err) + } + if len(plans) != 2 { + t.Fatalf("len(plans) = %d, want 2", len(plans)) + } + if plans[0].Agent.Name != "codex" { + t.Fatalf("plans[0].Agent.Name = %q, want %q", plans[0].Agent.Name, "codex") + } + if plans[1].Agent.Name != "claude" { + t.Fatalf("plans[1].Agent.Name = %q, want %q", plans[1].Agent.Name, "claude") + } +} + +func TestSelectDispatchPlansAvoidsSameRepoParallelism(t *testing.T) { + t.Parallel() + + cfg := config.MachineConfig{ + Runtime: config.MachineRuntime{ + MaxConcurrentTasks: 2, + }, + Agents: config.MachineAgents{ + DefaultOrder: []string{"codex"}, + Installed: []config.InstalledAgent{ + { + Name: "codex", + Type: "codex", + Command: []string{"sh"}, + Enabled: true, + MaxParallel: 2, + }, + }, + }, + } + + now := time.Now().UTC() + candidates := []planner.Candidate{ + { + ProjectPath: "/tmp/repo-a", + Repo: gh.Repository{Slug: "org/repo-a"}, + Issue: gh.Issue{ + Number: 1, + URL: "https://example.com/a/1", + UpdatedAt: now, + }, + AgentOrder: []string{"codex"}, + }, + { + ProjectPath: "/tmp/repo-a", + Repo: gh.Repository{Slug: "org/repo-a"}, + Issue: gh.Issue{ + Number: 2, + URL: "https://example.com/a/2", + UpdatedAt: now, + }, + AgentOrder: []string{"codex"}, + }, + { + ProjectPath: "/tmp/repo-b", + Repo: gh.Repository{Slug: "org/repo-b"}, + Issue: gh.Issue{ + Number: 3, + URL: "https://example.com/b/3", + UpdatedAt: now, + }, + AgentOrder: []string{"codex"}, + }, + } + + plans, err := selectDispatchPlans(cfg, candidates, state.Store{Tasks: map[string]state.TaskRecord{}}) + if err != nil { + t.Fatalf("selectDispatchPlans() error = %v", err) + } + if len(plans) != 2 { + t.Fatalf("len(plans) = %d, want 2", len(plans)) + } + if plans[0].Candidate.ProjectPath == plans[1].Candidate.ProjectPath { + t.Fatalf("selected two plans for the same repo path: %q", plans[0].Candidate.ProjectPath) + } +}