diff options
| -rw-r--r-- | CLAUDE.md | 68 | ||||
| -rw-r--r-- | docs/api-keys-setup.md | 15 | ||||
| -rw-r--r-- | internal/api/roles.go | 121 | ||||
| -rw-r--r-- | internal/api/roles_test.go | 177 | ||||
| -rw-r--r-- | internal/api/server.go | 3 | ||||
| -rw-r--r-- | internal/cli/cloudrunners.go | 91 | ||||
| -rw-r--r-- | internal/cli/cloudrunners_test.go | 78 | ||||
| -rw-r--r-- | internal/cli/run.go | 40 | ||||
| -rw-r--r-- | internal/cli/serve.go | 63 | ||||
| -rw-r--r-- | internal/config/config.go | 43 | ||||
| -rw-r--r-- | internal/config/config_test.go | 24 | ||||
| -rw-r--r-- | internal/event/event.go | 6 | ||||
| -rw-r--r-- | internal/executor/executor.go | 150 | ||||
| -rw-r--r-- | internal/executor/executor_test.go | 4 | ||||
| -rw-r--r-- | internal/executor/groq_routing_test.go | 98 | ||||
| -rw-r--r-- | internal/executor/role_dispatch_test.go | 206 | ||||
| -rw-r--r-- | internal/role/role.go | 73 | ||||
| -rw-r--r-- | internal/role/role_test.go | 73 | ||||
| -rw-r--r-- | internal/scheduler/scheduler.go | 277 | ||||
| -rw-r--r-- | internal/scheduler/scheduler_test.go | 369 | ||||
| -rw-r--r-- | internal/storage/db.go | 45 | ||||
| -rw-r--r-- | internal/storage/roleconfig.go | 147 | ||||
| -rw-r--r-- | internal/storage/roleconfig_test.go | 151 | ||||
| -rw-r--r-- | internal/task/task.go | 8 |
24 files changed, 2213 insertions, 117 deletions
@@ -74,6 +74,11 @@ Config defaults to `~/.claudomator/config.toml`. Data is stored in `~/.claudomat | `internal/config` | TOML config loading + data-dir layout helpers | | `internal/cli` | Cobra commands: `run`, `serve`, `list`, `status`, `start`, `logs`, `create`, `report`, `init` | | `internal/version` | VCS version detection (`debug.ReadBuildInfo`) | +| `internal/provider` | Provider-neutral chat/tool-use interface (`Provider`); adapters: `anthropic`, `google`, `openaicompat` (wraps `internal/llm`, also backs the `groq`/`openrouter`/`openai` runners) | +| `internal/agentloop` | Provider-neutral tool-use control-flow loop shared by all `NativeRunner`s | +| `internal/sandbox` | `Sandbox` interface (`HostSandbox`/`DockerSandbox`) + pre-tool-use guardrail hooks | +| `internal/role` | `RoleConfig`/`Tier`/`Rung` — per-role system prompt + provider/model escalation ladder | +| `internal/scheduler` | `Scheduler` — polls role-typed FAILED tasks and retries/escalates them per their role's ladder | | `web` | Embedded static UI (`embed.go`) | ### Key Data Flows @@ -174,28 +179,40 @@ Batch files wrap multiple tasks under a `tasks:` key. ### Storage Schema -Two tables. Schema is auto-migrated additively on `storage.Open()` — new columns are `ALTER TABLE ... ADD COLUMN` statements that silently succeed if the column already exists. +Schema is auto-migrated additively on `storage.Open()` — new columns are `ALTER TABLE ... ADD COLUMN` statements that silently succeed if the column already exists. ``` -tasks: id, name, description, config_json, priority, timeout_ns, retry_json, - tags_json, depends_on_json, parent_task_id, state, rejection_comment, - question_json, summary, elaboration_input, interactions_json, - created_at, updated_at +tasks: id, name, description, config_json, priority, timeout_ns, retry_json, + tags_json, depends_on_json, parent_task_id, state, rejection_comment, + question_json, summary, elaboration_input, interactions_json, + created_at, updated_at -executions: id, task_id, start_time, end_time, exit_code, status, stdout_path, - stderr_path, artifact_dir, cost_usd, error_msg, session_id, - sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, - agent +executions: id, task_id, start_time, end_time, exit_code, status, stdout_path, + stderr_path, artifact_dir, cost_usd, error_msg, session_id, + sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, + agent, escalation_rung -events: id, task_id, seq, ts, kind, actor, payload_json +events: id, task_id, seq, ts, kind, actor, payload_json + +role_configs: id, role, version, status, config_json, created_at, activated_at, + retired_at, proposed_by (UNIQUE(role, version)) ``` The `events` table is the per-task observability stream (`internal/event`); see the REST `GET /api/tasks/{id}/events` and the chatbot MCP `get_events` tool. `executions.agent` records which provider ran each execution, for budget -accounting (`SpendByProviderSince`). +accounting (`SpendByProviderSince`). `executions.escalation_rung` records the +0-based index into the active role's `EscalationLadder` a role-typed task's +execution ran at (0 for non-role tasks); see "Role-based dispatch & +escalation" below. + +`role_configs` holds versioned `internal/role.RoleConfig` blobs (`config_json`) +per role, `status` one of `draft`/`active`/`retired`. `ActivateRoleConfigVersion` +enforces "at most one active row per role" transactionally (retire-then-activate +in one transaction), the same way `UpdateTaskState` enforces the task state +machine in a transaction rather than in schema — see `internal/storage/roleconfig.go`. -JSON blobs: `config_json` (AgentConfig), `retry_json`, `tags_json`, `depends_on_json`, `interactions_json`, `changestats_json`, `commits_json`. +JSON blobs: `config_json` (AgentConfig on `tasks`; RoleConfig on `role_configs`), `retry_json`, `tags_json`, `depends_on_json`, `interactions_json`, `changestats_json`, `commits_json`. --- @@ -276,6 +293,18 @@ tokens; GitHub webhooks use HMAC. > **Implementation gap:** Output parsing is brittle — strips `"Loaded cached credentials."` lines and markdown fences by string matching. No fallback if Gemini CLI isn't installed. Classification results are not cached or logged for learning. +### Role-based dispatch & escalation + +A task opts into role-based dispatch by setting `agent.role` (`task.AgentConfig.Role`) to a role name with an **active** `role_configs` version (see the REST endpoints below). Tasks with `agent.role` unset (every existing YAML/chatbot task shape) are completely unaffected — this is purely additive. + +- **Initial dispatch** (`internal/executor.Pool.execute()`): when `Agent.Role != ""` and `Agent.Type == ""`, the pool resolves tier 0 of the active `role_configs` row's `EscalationLadder`, setting `Agent.Type`/`Agent.Model` on a copy of the task (mirroring the `withFailureHistory` copy-before-mutate pattern) and applying `RoleConfig.SystemPrompt` to `Agent.SystemPromptAppend`. Multi-candidate `round_robin` tiers rotate via `Pool.selectRung`, skipping any provider currently in `Pool.rateLimited` (falling back to whichever clears soonest if all are limited). +- **Retry/escalation** (`internal/scheduler.Scheduler`, started by `serve` only — not the one-shot `run` command): polls for role-typed tasks whose latest execution is `FAILED` and, per the ladder tier at that execution's `escalation_rung`: retries at the same rung while under `tier.MaxRetries`, or escalates to the next tier's first candidate if `budget.Accountant.Allow` permits it (recording an `event.KindEscalated` event), or leaves the task `FAILED` for human attention — recording a `final: true` `KindEscalated` event — if the budget denies it or the ladder is exhausted. An in-memory (per-process) "already handled" set keyed by execution ID keeps the poll loop convergent: a task stuck at the end of its ladder gets exactly one `final` event, not one per poll tick. This resets on restart (by design — it's idempotent bookkeeping, not orchestration state, so re-deriving it once is harmless). +- **REST**: `POST /api/roles/{role}/versions` (create draft), `GET /api/roles/{role}/versions` (list), `POST /api/roles/{role}/activate?version=N` (activate, atomically retiring whatever was active). + +> **Stored but not yet enforced:** `RoleConfig.Tools`/`SandboxKind` are round-tripped through storage/API but don't affect tool availability or sandbox selection — those are still applied uniformly by `NativeRunner` regardless of role. `RoleConfig.DefaultBudgetUSD` is read narrowly by the scheduler as the estimated cost passed to `Allow()` when considering an escalation — it is not enforced at initial-dispatch time the way `task.AgentConfig.MaxBudgetUSD` is. See `internal/role/role.go`'s field docs. + +> **Known simplification:** the scheduler always escalates to `nextTier.Candidates[0]` — it does not round-robin across multiple candidates in a tier the way `Pool.execute()`'s initial-dispatch resolution does. A later phase could extend this if multi-candidate escalation tiers turn out to matter in practice. + --- @@ -283,6 +312,18 @@ tokens; GitHub webhooks use HMAC. ## Design Debt +### RoleConfig.Tools/SandboxKind/DefaultBudgetUSD are stored but not fully enforced + +Same pattern as `task.Priority`/`RetryConfig` below: `internal/role.RoleConfig.Tools` and `.SandboxKind` round-trip through `role_configs`/the REST endpoints but have no effect on dispatch — tool availability and sandbox selection are still applied uniformly by `NativeRunner` regardless of role. `.DefaultBudgetUSD` is read only narrowly, as the scheduler's estimated escalation cost, not enforced generally at dispatch time. + +### Scheduler's escalation candidate selection doesn't round-robin + +`internal/scheduler.Scheduler` always escalates to `nextTier.Candidates[0]`, unlike `Pool.execute()`'s initial-dispatch tier-0 resolution, which round-robins across multi-candidate tiers. Same-rung retries intentionally reuse the exact same provider/model rather than re-resolving. + +### Scheduler's double-processing guard is in-memory only + +`Scheduler.handled` (keyed by execution ID) prevents re-emitting a `final: true` `KindEscalated` event on every poll tick once a role-typed task's ladder is exhausted, but it's per-process and resets on restart. A restart can produce one extra "reconsideration" (and, if still exhausted/denied, one more `KindEscalated` event) — not an infinite loop, just not persisted. A future phase could persist this via a `tasks` column if that turns out to matter. + ### Deprecated task columns not yet dropped (Phase 8 follow-up) `tasks.question_json`, `summary`, `interactions_json`, `elaboration_input` are @@ -360,6 +401,9 @@ In `executor.go`, `withFailureHistory` creates a copy of the task struct (`copy | GET | `/api/workspaces` | List directories under `workspace_root` | | GET | `/api/health` | Server health | | POST | `/api/webhooks/github` | GitHub CI webhook | +| POST | `/api/roles/{role}/versions` | Create a new draft `role_configs` version | +| GET | `/api/roles/{role}/versions` | List all versions for a role | +| POST | `/api/roles/{role}/activate?version=N` | Activate a version (atomically retires the prior active one) | --- diff --git a/docs/api-keys-setup.md b/docs/api-keys-setup.md index 1a9daf9..83c98e1 100644 --- a/docs/api-keys-setup.md +++ b/docs/api-keys-setup.md @@ -6,9 +6,9 @@ Providers, in escalation-ladder order (cheapest/free first): | Provider | Status | Cost | Wire format | |---|---|---|---| -| Groq | to add (works once configured — Phase 1's `openaicompat` adapter, no dedicated phase needed) | free tier (real, no card required historically) | OpenAI-compatible | -| OpenRouter | to add (same as Groq) | free tier (rate-limited) + pay-as-you-go | OpenAI-compatible | -| OpenAI | to add (same as Groq) | prepaid credit, no ongoing free tier | OpenAI-compatible (native) | +| Groq | **wired up (Phase 5)** — `agent.type: "groq"` works as soon as a key is in `config.toml` (Phase 1's `openaicompat` adapter, no dedicated adapter package needed) | free tier (real, no card required historically) | OpenAI-compatible | +| OpenRouter | **wired up (Phase 5)** — `agent.type: "openrouter"` works as soon as a key is in `config.toml` (same adapter as Groq) | free tier (rate-limited) + pay-as-you-go | OpenAI-compatible | +| OpenAI | **wired up (Phase 5)** — `agent.type: "openai"` works as soon as a key is in `config.toml` (same adapter as Groq) | prepaid credit, no ongoing free tier | OpenAI-compatible (native) | | Google (Gemini) | **wired up (Phase 4)** — `agent.type: "google"` works as soon as a key is in `config.toml` | paid (already have account) | native (Gemini API) | | Anthropic (Claude) | **wired up (Phase 2)** — `agent.type: "anthropic"` works as soon as a key is in `config.toml` | paid (already have account) | native (Messages API) | @@ -60,6 +60,7 @@ Free tier, no credit card required to start. Good first escalation rung above lo 3. Copy the key immediately — Groq only shows it once (`gsk_...`). 4. Rate limits (free tier, vary per model): roughly 30 requests/minute, 6K–70K tokens/minute, 1K–14.4K requests/day, 100K–500K tokens/day depending on the specific model. Check `console.groq.com/settings/limits` for your account's current numbers per model — this varies enough by model that it's worth checking directly rather than trusting a fixed table here. 5. Put the key in `config.toml` under `[providers.groq]` as shown above. `endpoint` is Groq's OpenAI-compatible base: `https://api.groq.com/openai/v1`. +6. As soon as the key is set, `claudomator serve`/`run` registers `agent.type: "groq"` as a usable runner — no restart-time flag needed beyond having the key present. To turn it off without removing the key, set `[runners] groq = false`. --- @@ -72,6 +73,7 @@ Aggregates many providers/models behind one key, including models with a `:free` 3. Free-tier limits: **without any credit purchase**, 50 free-model requests/day total across all `:free`-suffixed models. **With $10+ in purchased credits**, that jumps to 1000 requests/day (a one-time credit purchase, not a subscription) — worth doing once your usage grows past the bare free tier, since it's a 20x limit increase for a fixed cost, not a recurring charge. 4. Pick a free model when configuring the harness — model IDs ending in `:free` (e.g. `meta-llama/llama-3.3-70b-instruct:free`) don't draw down credits. Browse current free models at `openrouter.ai/models?max_price=0`. 5. `endpoint` is `https://openrouter.ai/api/v1`. +6. As soon as the key is set, `claudomator serve`/`run` registers `agent.type: "openrouter"` as a usable runner. To turn it off without removing the key, set `[runners] openrouter = false`. --- @@ -84,6 +86,7 @@ No meaningful ongoing free tier at this point — expect to prepay. Included in 3. **Dashboard → API keys** (`platform.openai.com/api-keys`) → **Create new secret key**. Name it, copy it (`sk-...`) — shown once. 4. Consider setting a spend limit under **Billing → Limits** so a runaway harness escalation can't produce a surprise bill — this matters more here than for Groq/OpenRouter's free tiers. 5. `endpoint` is `https://api.openai.com/v1`. +6. As soon as the key is set, `claudomator serve`/`run` registers `agent.type: "openai"` as a usable runner. To turn it off without removing the key, set `[runners] openai = false`. --- @@ -119,7 +122,7 @@ Same situation as Gemini: `credentials/claude/` holds the `claude` CLI's OAuth s ## Sanity-checking a key works -Once phase 1/2/4 of the harness redesign land (native `Provider` adapters), the quickest check is submitting a trivial role-typed task pinned to a single rung, e.g. via the CLI: +All five providers are wired up as of Phase 5 (`internal/cli/cloudrunners.go`). The quickest check is submitting a trivial task pinned to a single provider directly (bypassing role-based escalation), e.g. via the CLI: ```bash ./claudomator run - <<'EOF' @@ -131,4 +134,6 @@ agent: EOF ``` -and checking `GET /api/budget` afterward to confirm the provider's spend/usage was recorded. Until then, you can at minimum confirm the key itself is valid with a raw `curl` against the provider's chat-completions endpoint using the docs above. +and checking `GET /api/budget` afterward to confirm the provider's spend/usage was recorded. You can also confirm the key itself is valid directly with a raw `curl` against the provider's chat-completions endpoint using the docs above. + +To exercise a provider through the escalation ladder instead of pinning `agent.type` directly, create a role via `POST /api/roles/{role}/versions` with an `escalation_ladder` naming the provider, activate it with `POST /api/roles/{role}/activate?version=1`, and submit a task with `agent.role` set instead of `agent.type`/`agent.model` — see `CLAUDE.md`'s "Role-based dispatch & escalation" section. diff --git a/internal/api/roles.go b/internal/api/roles.go new file mode 100644 index 0000000..700bec4 --- /dev/null +++ b/internal/api/roles.go @@ -0,0 +1,121 @@ +package api + +import ( + "encoding/json" + "net/http" + "strconv" + "time" + + "github.com/thepeterstone/claudomator/internal/role" + "github.com/thepeterstone/claudomator/internal/storage" +) + +// roleVersionView is the JSON shape returned for a single role_configs +// version — decodes the stored config_json back into a role.RoleConfig so +// clients get structured fields rather than an opaque string. +type roleVersionView struct { + ID string `json:"id"` + Role string `json:"role"` + Version int `json:"version"` + Status string `json:"status"` + Config role.RoleConfig `json:"config"` + CreatedAt string `json:"created_at"` + ActivatedAt string `json:"activated_at,omitempty"` + RetiredAt string `json:"retired_at,omitempty"` + ProposedBy string `json:"proposed_by,omitempty"` +} + +// handleCreateRoleVersion handles POST /api/roles/{role}/versions. The +// request body is a role.RoleConfig (config_json shape, plus an optional +// proposed_by field); a new draft version is created with the next version +// number for that role. Mirrors handleCreateProject's shape (see +// internal/api/projects.go). +func (s *Server) handleCreateRoleVersion(w http.ResponseWriter, r *http.Request) { + roleName := r.PathValue("role") + if roleName == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "role is required"}) + return + } + + var input struct { + role.RoleConfig + ProposedBy string `json:"proposed_by"` + } + if err := json.NewDecoder(r.Body).Decode(&input); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON: " + err.Error()}) + return + } + input.RoleConfig.Role = roleName + + proposedBy := input.ProposedBy + if proposedBy == "" { + proposedBy = "human" + } + + configJSON, err := json.Marshal(input.RoleConfig) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + + created, err := s.store.CreateRoleConfig(roleName, string(configJSON), proposedBy) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusCreated, toRoleVersionView(created)) +} + +// handleListRoleVersions handles GET /api/roles/{role}/versions. +func (s *Server) handleListRoleVersions(w http.ResponseWriter, r *http.Request) { + roleName := r.PathValue("role") + rows, err := s.store.ListRoleConfigVersions(roleName) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": err.Error()}) + return + } + views := make([]*roleVersionView, 0, len(rows)) + for _, row := range rows { + views = append(views, toRoleVersionView(row)) + } + writeJSON(w, http.StatusOK, views) +} + +// handleActivateRoleVersion handles POST /api/roles/{role}/activate?version=N. +func (s *Server) handleActivateRoleVersion(w http.ResponseWriter, r *http.Request) { + roleName := r.PathValue("role") + versionStr := r.URL.Query().Get("version") + version, err := strconv.Atoi(versionStr) + if err != nil || version < 1 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid or missing version query param"}) + return + } + if err := s.store.ActivateRoleConfigVersion(roleName, version); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, map[string]interface{}{ + "message": "role config version activated", + "role": roleName, + "version": version, + }) +} + +func toRoleVersionView(row *storage.RoleConfigRow) *roleVersionView { + v := &roleVersionView{ + ID: row.ID, + Role: row.Role, + Version: row.Version, + Status: row.Status, + CreatedAt: row.CreatedAt.Format(time.RFC3339), + ProposedBy: row.ProposedBy, + } + if row.ActivatedAt != nil { + v.ActivatedAt = row.ActivatedAt.Format(time.RFC3339) + } + if row.RetiredAt != nil { + v.RetiredAt = row.RetiredAt.Format(time.RFC3339) + } + _ = json.Unmarshal([]byte(row.ConfigJSON), &v.Config) + return v +} diff --git a/internal/api/roles_test.go b/internal/api/roles_test.go new file mode 100644 index 0000000..97afdfe --- /dev/null +++ b/internal/api/roles_test.go @@ -0,0 +1,177 @@ +package api + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "strconv" + "testing" + + "github.com/thepeterstone/claudomator/internal/role" +) + +// TestServer_CreateRoleVersion_CreatesDraft mirrors the projects endpoint +// test pattern (see server_test.go's testServer/httptest.NewRequest usage): +// POSTing a role.RoleConfig body creates a new draft version. +func TestServer_CreateRoleVersion_CreatesDraft(t *testing.T) { + srv, _ := testServer(t) + + body, _ := json.Marshal(map[string]any{ + "system_prompt": "You are a careful coder.", + "escalation_ladder": []map[string]any{ + { + "candidates": []map[string]any{{"provider": "local", "model": "llama3.1:8b"}}, + "selection_mode": "single", + "max_retries": 2, + }, + }, + }) + req := httptest.NewRequest("POST", "/api/roles/coder/versions", bytes.NewReader(body)) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + + if w.Code != http.StatusCreated { + t.Fatalf("expected 201, got %d: %s", w.Code, w.Body.String()) + } + var got struct { + Role string `json:"role"` + Version int `json:"version"` + Status string `json:"status"` + Config role.RoleConfig `json:"config"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatalf("decode response: %v", err) + } + if got.Role != "coder" { + t.Errorf("role = %q, want %q", got.Role, "coder") + } + if got.Version != 1 { + t.Errorf("version = %d, want 1", got.Version) + } + if got.Status != "draft" { + t.Errorf("status = %q, want draft", got.Status) + } + if got.Config.SystemPrompt != "You are a careful coder." { + t.Errorf("system_prompt round-trip failed: got %q", got.Config.SystemPrompt) + } + if len(got.Config.EscalationLadder) != 1 { + t.Fatalf("expected 1 tier, got %d", len(got.Config.EscalationLadder)) + } +} + +// TestServer_ListRoleVersions_ReturnsAllVersions posts two versions and +// confirms the list endpoint returns both, ordered by version. +func TestServer_ListRoleVersions_ReturnsAllVersions(t *testing.T) { + srv, _ := testServer(t) + + for i := 0; i < 2; i++ { + body, _ := json.Marshal(map[string]any{"system_prompt": "v"}) + req := httptest.NewRequest("POST", "/api/roles/coder/versions", bytes.NewReader(body)) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("create version %d: expected 201, got %d", i, w.Code) + } + } + + req := httptest.NewRequest("GET", "/api/roles/coder/versions", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", w.Code, w.Body.String()) + } + + var got []struct { + Version int `json:"version"` + Status string `json:"status"` + } + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatalf("decode response: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 versions, got %d", len(got)) + } + if got[0].Version != 1 || got[1].Version != 2 { + t.Errorf("expected versions [1,2], got [%d,%d]", got[0].Version, got[1].Version) + } + for _, v := range got { + if v.Status != "draft" { + t.Errorf("version %d: expected status draft, got %q", v.Version, v.Status) + } + } +} + +// TestServer_ActivateRoleVersion_ActivatesAndRetiresPrior exercises the full +// create -> activate -> create -> activate cycle over HTTP, confirming the +// exactly-one-active-row invariant surfaces correctly through the API. +func TestServer_ActivateRoleVersion_ActivatesAndRetiresPrior(t *testing.T) { + srv, store := testServer(t) + + post := func(body string) { + req := httptest.NewRequest("POST", "/api/roles/coder/versions", bytes.NewReader([]byte(body))) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusCreated { + t.Fatalf("create version: expected 201, got %d: %s", w.Code, w.Body.String()) + } + } + post(`{"system_prompt":"v1"}`) + post(`{"system_prompt":"v2"}`) + + activate := func(version int) int { + req := httptest.NewRequest("POST", "/api/roles/coder/activate?version="+strconv.Itoa(version), nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + return w.Code + } + + if code := activate(1); code != http.StatusOK { + t.Fatalf("activate v1: expected 200, got %d", code) + } + active, err := store.GetActiveRoleConfig("coder") + if err != nil { + t.Fatalf("GetActiveRoleConfig: %v", err) + } + if active.Version != 1 { + t.Fatalf("expected active version 1, got %d", active.Version) + } + + if code := activate(2); code != http.StatusOK { + t.Fatalf("activate v2: expected 200, got %d", code) + } + active, err = store.GetActiveRoleConfig("coder") + if err != nil { + t.Fatalf("GetActiveRoleConfig: %v", err) + } + if active.Version != 2 { + t.Fatalf("expected active version 2 after activating it, got %d", active.Version) + } + + versions, err := store.ListRoleConfigVersions("coder") + if err != nil { + t.Fatalf("ListRoleConfigVersions: %v", err) + } + activeCount := 0 + for _, v := range versions { + if v.Status == "active" { + activeCount++ + } + } + if activeCount != 1 { + t.Fatalf("expected exactly 1 active version, got %d", activeCount) + } +} + +// TestServer_ActivateRoleVersion_UnknownVersion_Returns400 confirms +// activating a version that doesn't exist is a client error, not a panic or +// 500. +func TestServer_ActivateRoleVersion_UnknownVersion_Returns400(t *testing.T) { + srv, _ := testServer(t) + req := httptest.NewRequest("POST", "/api/roles/coder/activate?version=99", nil) + w := httptest.NewRecorder() + srv.Handler().ServeHTTP(w, req) + if w.Code != http.StatusBadRequest { + t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String()) + } +} diff --git a/internal/api/server.go b/internal/api/server.go index 35128d0..0e5b185 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -172,6 +172,9 @@ func (s *Server) routes() { s.mux.HandleFunc("POST /api/projects", s.handleCreateProject) s.mux.HandleFunc("GET /api/projects/{id}", s.handleGetProject) s.mux.HandleFunc("PUT /api/projects/{id}", s.handleUpdateProject) + s.mux.HandleFunc("POST /api/roles/{role}/versions", s.handleCreateRoleVersion) + s.mux.HandleFunc("GET /api/roles/{role}/versions", s.handleListRoleVersions) + s.mux.HandleFunc("POST /api/roles/{role}/activate", s.handleActivateRoleVersion) s.mux.HandleFunc("GET /api/health", s.handleHealth) s.mux.HandleFunc("GET /api/version", s.handleVersion) s.mux.HandleFunc("POST /api/webhooks/github", s.handleGitHubWebhook) diff --git a/internal/cli/cloudrunners.go b/internal/cli/cloudrunners.go new file mode 100644 index 0000000..55b3efb --- /dev/null +++ b/internal/cli/cloudrunners.go @@ -0,0 +1,91 @@ +package cli + +import ( + "log/slog" + "net/http" + "time" + + "github.com/thepeterstone/claudomator/internal/config" + "github.com/thepeterstone/claudomator/internal/executor" + "github.com/thepeterstone/claudomator/internal/llm" + "github.com/thepeterstone/claudomator/internal/provider" + "github.com/thepeterstone/claudomator/internal/provider/anthropic" + "github.com/thepeterstone/claudomator/internal/provider/google" + "github.com/thepeterstone/claudomator/internal/provider/openaicompat" +) + +// cloudProviderSpec describes how to construct a provider.Provider for one +// cloud provider entry in cfg.Providers, and which RunnersConfig field gates +// it. anthropic/google use their native wire-format adapters; groq/ +// openrouter/openai are OpenAI-wire-compatible and reuse the Phase 1 +// openaicompat adapter — no dedicated adapter package needed for those three +// (see docs/api-keys-setup.md). +type cloudProviderSpec struct { + name string + enabled func(config.RunnersConfig) bool + build func(pc config.ProviderConfig, timeout time.Duration) provider.Provider +} + +func openaiCompatBuild(pc config.ProviderConfig, timeout time.Duration) provider.Provider { + return openaicompat.New(&llm.Client{ + Endpoint: pc.Endpoint, + Model: pc.DefaultModel, + APIKey: pc.APIKey, + HTTPClient: &http.Client{Timeout: timeout}, + }) +} + +var cloudProviderSpecs = []cloudProviderSpec{ + { + name: "anthropic", + enabled: config.RunnersConfig.AnthropicEnabled, + build: func(pc config.ProviderConfig, timeout time.Duration) provider.Provider { + return anthropic.New(pc.APIKey, pc.Endpoint, timeout) + }, + }, + { + name: "google", + enabled: config.RunnersConfig.GoogleEnabled, + build: func(pc config.ProviderConfig, timeout time.Duration) provider.Provider { + return google.New(pc.APIKey, pc.Endpoint, timeout) + }, + }, + {name: "groq", enabled: config.RunnersConfig.GroqEnabled, build: openaiCompatBuild}, + {name: "openrouter", enabled: config.RunnersConfig.OpenRouterEnabled, build: openaiCompatBuild}, + {name: "openai", enabled: config.RunnersConfig.OpenAIEnabled, build: openaiCompatBuild}, +} + +// registerCloudRunners constructs and adds a NativeRunner (SandboxKind: +// "docker", matching the original anthropic/google-only wiring — cloud +// providers get real container isolation, not the weaker HostSandbox local +// runners use) into runners for every cloud provider in cloudProviderSpecs +// that has a non-empty cfg.Providers[name].APIKey and is enabled via its +// RunnersConfig.<Name>Enabled() gate. Called identically from both `serve` +// and `run` — the only difference between them (logDir) is a parameter, and +// neither command does anything provider-specific beyond this. +func registerCloudRunners(runners map[string]executor.Runner, cfg *config.Config, logger *slog.Logger, logDir string) { + for _, spec := range cloudProviderSpecs { + pc, ok := cfg.Providers[spec.name] + if !ok || pc.APIKey == "" || !spec.enabled(cfg.Runners) { + continue + } + timeout := time.Duration(0) + if pc.TimeoutSeconds > 0 { + timeout = time.Duration(pc.TimeoutSeconds) * time.Second + } + runners[spec.name] = &executor.NativeRunner{ + Provider: spec.build(pc, timeout), + Logger: logger, + LogDir: logDir, + DefaultModel: pc.DefaultModel, + // Native cloud providers get real container isolation + // (sandbox.DockerSandbox) rather than HostSandbox's host-side + // path-prefix confinement — see NativeRunner.SandboxKind doc. + SandboxKind: "docker", + SandboxImage: cfg.SandboxImage, + } + if logger != nil { + logger.Info(spec.name+" runner registered", "default_model", pc.DefaultModel, "sandbox", "docker") + } + } +} diff --git a/internal/cli/cloudrunners_test.go b/internal/cli/cloudrunners_test.go new file mode 100644 index 0000000..7943347 --- /dev/null +++ b/internal/cli/cloudrunners_test.go @@ -0,0 +1,78 @@ +package cli + +import ( + "log/slog" + "os" + "testing" + + "github.com/thepeterstone/claudomator/internal/config" + "github.com/thepeterstone/claudomator/internal/executor" +) + +// TestRegisterCloudRunners_RegistersAllFiveConfiguredProviders confirms +// registerCloudRunners (used identically by both `serve` and `run`) wires up +// a NativeRunner for every cloud provider that has a non-empty API key — +// including groq/openrouter/openai, which docs/api-keys-setup.md documented +// as "works once configured" before this phase actually configured them. +func TestRegisterCloudRunners_RegistersAllFiveConfiguredProviders(t *testing.T) { + cfg := &config.Config{ + SandboxImage: "test-image:latest", + Providers: map[string]config.ProviderConfig{ + "anthropic": {APIKey: "ak", DefaultModel: "claude-sonnet-5"}, + "google": {APIKey: "gk", DefaultModel: "gemini-2.5-flash"}, + "groq": {APIKey: "grk", Endpoint: "https://api.groq.com/openai/v1", DefaultModel: "llama-3.3-70b-versatile"}, + "openrouter": {APIKey: "ork", Endpoint: "https://openrouter.ai/api/v1", DefaultModel: "meta-llama/llama-3.3-70b-instruct:free"}, + "openai": {APIKey: "oak", Endpoint: "https://api.openai.com/v1", DefaultModel: "gpt-4o-mini"}, + }, + } + runners := map[string]executor.Runner{} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + registerCloudRunners(runners, cfg, logger, t.TempDir()) + + for _, name := range []string{"anthropic", "google", "groq", "openrouter", "openai"} { + r, ok := runners[name] + if !ok { + t.Errorf("expected a %q runner to be registered", name) + continue + } + nr, ok := r.(*executor.NativeRunner) + if !ok { + t.Errorf("%q runner should be a *executor.NativeRunner", name) + continue + } + if nr.Provider == nil { + t.Errorf("%q runner has a nil Provider", name) + } + if nr.SandboxKind != "docker" { + t.Errorf("%q runner should use SandboxKind \"docker\", got %q", name, nr.SandboxKind) + } + } +} + +// TestRegisterCloudRunners_SkipsProvidersWithoutAPIKeyOrDisabled confirms +// providers are skipped when unconfigured (no api_key) or explicitly +// disabled via [runners], matching the anthropic/google-only behavior this +// generalizes. +func TestRegisterCloudRunners_SkipsProvidersWithoutAPIKeyOrDisabled(t *testing.T) { + f := false + cfg := &config.Config{ + Providers: map[string]config.ProviderConfig{ + "groq": {APIKey: "", DefaultModel: "x"}, // no key + "openrouter": {APIKey: "ork", DefaultModel: "x"}, + }, + Runners: config.RunnersConfig{OpenRouter: &f}, // explicitly disabled + } + runners := map[string]executor.Runner{} + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + registerCloudRunners(runners, cfg, logger, t.TempDir()) + + if _, ok := runners["groq"]; ok { + t.Error("groq should not be registered without an api_key") + } + if _, ok := runners["openrouter"]; ok { + t.Error("openrouter should not be registered when explicitly disabled") + } + if _, ok := runners["openai"]; ok { + t.Error("openai should not be registered when absent from cfg.Providers") + } +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 34c3d42..08bfb68 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -6,12 +6,9 @@ import ( "os" "os/signal" "syscall" - "time" "github.com/spf13/cobra" "github.com/thepeterstone/claudomator/internal/executor" - "github.com/thepeterstone/claudomator/internal/provider/anthropic" - "github.com/thepeterstone/claudomator/internal/provider/google" "github.com/thepeterstone/claudomator/internal/provider/openaicompat" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" @@ -119,39 +116,10 @@ func runTasks(file string, parallel int, dryRun bool) error { } } - if pc, ok := cfg.Providers["anthropic"]; ok && pc.APIKey != "" && cfg.Runners.AnthropicEnabled() { - timeout := time.Duration(0) - if pc.TimeoutSeconds > 0 { - timeout = time.Duration(pc.TimeoutSeconds) * time.Second - } - runners["anthropic"] = &executor.NativeRunner{ - Provider: anthropic.New(pc.APIKey, pc.Endpoint, timeout), - Logger: logger, - LogDir: cfg.LogDir, - DefaultModel: pc.DefaultModel, - // See internal/cli/serve.go: native cloud providers get - // sandbox.DockerSandbox rather than the weaker HostSandbox. - SandboxKind: "docker", - SandboxImage: cfg.SandboxImage, - } - } - - if pc, ok := cfg.Providers["google"]; ok && pc.APIKey != "" && cfg.Runners.GoogleEnabled() { - timeout := time.Duration(0) - if pc.TimeoutSeconds > 0 { - timeout = time.Duration(pc.TimeoutSeconds) * time.Second - } - runners["google"] = &executor.NativeRunner{ - Provider: google.New(pc.APIKey, pc.Endpoint, timeout), - Logger: logger, - LogDir: cfg.LogDir, - DefaultModel: pc.DefaultModel, - // See internal/cli/serve.go: native cloud providers get - // sandbox.DockerSandbox rather than the weaker HostSandbox. - SandboxKind: "docker", - SandboxImage: cfg.SandboxImage, - } - } + // Native cloud providers (anthropic/google's own wire formats, plus + // groq/openrouter/openai via the OpenAI-compatible adapter) — see + // internal/cli/cloudrunners.go. + registerCloudRunners(runners, cfg, logger, cfg.LogDir) pool := executor.NewPool(parallel, runners, store, logger) pool.Classifier = &executor.Classifier{ diff --git a/internal/cli/serve.go b/internal/cli/serve.go index 57c2cbd..656ee66 100644 --- a/internal/cli/serve.go +++ b/internal/cli/serve.go @@ -16,9 +16,8 @@ import ( "github.com/thepeterstone/claudomator/internal/config" "github.com/thepeterstone/claudomator/internal/executor" "github.com/thepeterstone/claudomator/internal/notify" - "github.com/thepeterstone/claudomator/internal/provider/anthropic" - "github.com/thepeterstone/claudomator/internal/provider/google" "github.com/thepeterstone/claudomator/internal/provider/openaicompat" + "github.com/thepeterstone/claudomator/internal/scheduler" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/version" ) @@ -156,44 +155,10 @@ func serve(addr, basePath string) error { logger.Info("local runner registered", "endpoint", cfg.LocalModel.Endpoint, "model", cfg.LocalModel.Model) } - if pc, ok := cfg.Providers["anthropic"]; ok && pc.APIKey != "" && cfg.Runners.AnthropicEnabled() { - timeout := time.Duration(0) - if pc.TimeoutSeconds > 0 { - timeout = time.Duration(pc.TimeoutSeconds) * time.Second - } - runners["anthropic"] = &executor.NativeRunner{ - Provider: anthropic.New(pc.APIKey, pc.Endpoint, timeout), - Logger: logger, - LogDir: cfg.LogDir, - DefaultModel: pc.DefaultModel, - // Native cloud providers get real container isolation - // (sandbox.DockerSandbox) rather than HostSandbox's host-side - // path-prefix confinement. - SandboxKind: "docker", - SandboxImage: cfg.SandboxImage, - } - logger.Info("anthropic runner registered", "default_model", pc.DefaultModel, "sandbox", "docker") - } - - if pc, ok := cfg.Providers["google"]; ok && pc.APIKey != "" && cfg.Runners.GoogleEnabled() { - timeout := time.Duration(0) - if pc.TimeoutSeconds > 0 { - timeout = time.Duration(pc.TimeoutSeconds) * time.Second - } - runners["google"] = &executor.NativeRunner{ - Provider: google.New(pc.APIKey, pc.Endpoint, timeout), - Logger: logger, - LogDir: cfg.LogDir, - DefaultModel: pc.DefaultModel, - // Native cloud providers get real container isolation - // (sandbox.DockerSandbox) rather than HostSandbox's host-side - // path-prefix confinement — same reasoning as the "anthropic" - // runner above. - SandboxKind: "docker", - SandboxImage: cfg.SandboxImage, - } - logger.Info("google runner registered", "default_model", pc.DefaultModel, "sandbox", "docker") - } + // Native cloud providers (anthropic/google's own wire formats, plus + // groq/openrouter/openai via the OpenAI-compatible adapter) — see + // internal/cli/cloudrunners.go. + registerCloudRunners(runners, cfg, logger, cfg.LogDir) pool := executor.NewPool(cfg.MaxConcurrent, runners, store, logger) pool.Classifier = &executor.Classifier{ @@ -282,6 +247,24 @@ func serve(addr, basePath string) error { srv.SetContext(ctx) srv.StartHub() + // Retry-then-escalate scheduler for role-typed tasks (internal/role, + // internal/scheduler). Only `serve` runs this background loop — the + // one-shot `run` command has no long-lived process to host it in. + sch := &scheduler.Scheduler{ + Store: store, + Pool: pool, + Logger: logger, + } + if accountant != nil { + // Only assign when non-nil: a nil *budget.Accountant boxed into the + // scheduler.BudgetGate interface would be a non-nil interface value + // whose methods panic on the nil receiver — same reason pool.Budget + // above is only ever assigned inside the `if accountant configured` + // branch, never unconditionally. + sch.Budget = accountant + } + go sch.Run(ctx, cfg.Scheduler.PollInterval()) + httpSrv := &http.Server{ Addr: addr, Handler: srv.Handler(), diff --git a/internal/config/config.go b/internal/config/config.go index b9454d1..9e065b5 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -72,14 +72,26 @@ type RunnersConfig struct { // Anthropic vs. Claude above. Also requires [providers.google].api_key to // be set; Google = true with no configured key has no effect. Google *bool `toml:"google"` + // Groq, OpenRouter, and OpenAI gate NativeRunners registered under agent + // types "groq"/"openrouter"/"openai", all backed by + // internal/provider/openaicompat (they're OpenAI-wire-compatible, so no + // dedicated adapter package is needed — see docs/api-keys-setup.md). + // Each also requires the matching [providers.<name>].api_key to be set; + // true with no configured key has no effect, same as Anthropic/Google. + Groq *bool `toml:"groq"` + OpenRouter *bool `toml:"openrouter"` + OpenAI *bool `toml:"openai"` } -func (r RunnersConfig) ClaudeEnabled() bool { return r.Claude == nil || *r.Claude } -func (r RunnersConfig) GeminiEnabled() bool { return r.Gemini == nil || *r.Gemini } -func (r RunnersConfig) LocalEnabled() bool { return r.Local == nil || *r.Local } -func (r RunnersConfig) ContainerEnabled() bool { return r.Container == nil || *r.Container } -func (r RunnersConfig) AnthropicEnabled() bool { return r.Anthropic == nil || *r.Anthropic } -func (r RunnersConfig) GoogleEnabled() bool { return r.Google == nil || *r.Google } +func (r RunnersConfig) ClaudeEnabled() bool { return r.Claude == nil || *r.Claude } +func (r RunnersConfig) GeminiEnabled() bool { return r.Gemini == nil || *r.Gemini } +func (r RunnersConfig) LocalEnabled() bool { return r.Local == nil || *r.Local } +func (r RunnersConfig) ContainerEnabled() bool { return r.Container == nil || *r.Container } +func (r RunnersConfig) AnthropicEnabled() bool { return r.Anthropic == nil || *r.Anthropic } +func (r RunnersConfig) GoogleEnabled() bool { return r.Google == nil || *r.Google } +func (r RunnersConfig) GroqEnabled() bool { return r.Groq == nil || *r.Groq } +func (r RunnersConfig) OpenRouterEnabled() bool { return r.OpenRouter == nil || *r.OpenRouter } +func (r RunnersConfig) OpenAIEnabled() bool { return r.OpenAI == nil || *r.OpenAI } // ProviderConfig configures a native (or OpenAI-compatible) LLM provider // backend for the multi-provider harness — see docs/api-keys-setup.md. Phase @@ -131,6 +143,10 @@ type Config struct { LocalModel LocalModel `toml:"local_model"` Runners RunnersConfig `toml:"runners"` Budget BudgetConfig `toml:"budget"` + // Scheduler configures internal/scheduler.Scheduler, the retry-then- + // escalate loop for role-typed tasks. Started only by `serve` (the + // one-shot `run` command has no background scheduler). + Scheduler SchedulerConfig `toml:"scheduler"` // Providers configures native/OpenAI-compatible LLM backends by name (see // ProviderConfig). Unused by runner construction in Phase 1. Providers map[string]ProviderConfig `toml:"providers"` @@ -145,6 +161,21 @@ type BudgetConfig struct { Provider5hUSD map[string]float64 `toml:"provider_5h_usd"` } +// SchedulerConfig configures internal/scheduler.Scheduler's poll loop. +type SchedulerConfig struct { + // PollIntervalSeconds is how often the scheduler checks for role-typed + // FAILED tasks to retry or escalate. Defaults to 30 when zero/unset. + PollIntervalSeconds int `toml:"poll_interval_seconds"` +} + +// PollInterval returns the configured poll interval, defaulting to 30s. +func (c SchedulerConfig) PollInterval() time.Duration { + if c.PollIntervalSeconds <= 0 { + return 30 * time.Second + } + return time.Duration(c.PollIntervalSeconds) * time.Second +} + func Default() (*Config, error) { home, err := os.UserHomeDir() if err != nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index 4167ff7..31863f8 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -74,11 +74,20 @@ func TestRunnersConfig_DefaultsAllEnabled(t *testing.T) { if !r.GoogleEnabled() { t.Error("google should be enabled by default (nil)") } + if !r.GroqEnabled() { + t.Error("groq should be enabled by default (nil)") + } + if !r.OpenRouterEnabled() { + t.Error("openrouter should be enabled by default (nil)") + } + if !r.OpenAIEnabled() { + t.Error("openai should be enabled by default (nil)") + } } func TestRunnersConfig_ExplicitFalseDisables(t *testing.T) { f := false - r := RunnersConfig{Claude: &f, Gemini: &f, Local: &f, Container: &f, Anthropic: &f, Google: &f} + r := RunnersConfig{Claude: &f, Gemini: &f, Local: &f, Container: &f, Anthropic: &f, Google: &f, Groq: &f, OpenRouter: &f, OpenAI: &f} if r.ClaudeEnabled() { t.Error("explicit false should disable claude") } @@ -97,12 +106,21 @@ func TestRunnersConfig_ExplicitFalseDisables(t *testing.T) { if r.GoogleEnabled() { t.Error("explicit false should disable google") } + if r.GroqEnabled() { + t.Error("explicit false should disable groq") + } + if r.OpenRouterEnabled() { + t.Error("explicit false should disable openrouter") + } + if r.OpenAIEnabled() { + t.Error("explicit false should disable openai") + } } func TestRunnersConfig_ExplicitTrueEnables(t *testing.T) { tr := true - r := RunnersConfig{Claude: &tr, Gemini: &tr, Local: &tr, Container: &tr, Anthropic: &tr, Google: &tr} - if !r.ClaudeEnabled() || !r.GeminiEnabled() || !r.LocalEnabled() || !r.ContainerEnabled() || !r.AnthropicEnabled() || !r.GoogleEnabled() { + r := RunnersConfig{Claude: &tr, Gemini: &tr, Local: &tr, Container: &tr, Anthropic: &tr, Google: &tr, Groq: &tr, OpenRouter: &tr, OpenAI: &tr} + if !r.ClaudeEnabled() || !r.GeminiEnabled() || !r.LocalEnabled() || !r.ContainerEnabled() || !r.AnthropicEnabled() || !r.GoogleEnabled() || !r.GroqEnabled() || !r.OpenRouterEnabled() || !r.OpenAIEnabled() { t.Error("explicit true should keep all runners enabled") } } diff --git a/internal/event/event.go b/internal/event/event.go index 2573fde..202df63 100644 --- a/internal/event/event.go +++ b/internal/event/event.go @@ -23,6 +23,12 @@ const ( KindCostReport Kind = "cost_report" KindExecutionStarted Kind = "execution_started" KindExecutionEnded Kind = "execution_ended" + // KindEscalated records a scheduler decision on a role-typed task's + // escalation ladder (internal/scheduler): either bumping it to the next + // tier, or declining to (budget denied or ladder exhausted, payload + // includes "final": true — the task stays FAILED for human attention). + // Payload shape: from_rung, to_rung, from_provider, to_provider, final. + KindEscalated Kind = "escalated" ) // Actor identifies the originator of an event. diff --git a/internal/executor/executor.go b/internal/executor/executor.go index 51cf5d9..981a3ad 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -2,6 +2,7 @@ package executor import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -13,6 +14,7 @@ import ( "github.com/thepeterstone/claudomator/internal/event" "github.com/thepeterstone/claudomator/internal/llm" "github.com/thepeterstone/claudomator/internal/retry" + "github.com/thepeterstone/claudomator/internal/role" "github.com/thepeterstone/claudomator/internal/storage" "github.com/thepeterstone/claudomator/internal/task" "github.com/google/uuid" @@ -38,6 +40,11 @@ type Store interface { GetProject(id string) (*task.Project, error) CreateTask(t *task.Task) error CreateEvent(e *event.Event) error + // GetActiveRoleConfig returns the active role_configs row for a role + // (see internal/role.RoleConfig), or an error (typically sql.ErrNoRows) + // if none is active. Used by execute() to resolve tier 0 of a role-typed + // task's escalation ladder. + GetActiveRoleConfig(role string) (*storage.RoleConfigRow, error) } // LogPather is an optional interface runners can implement to provide the log @@ -77,6 +84,10 @@ type Pool struct { rateLimited map[string]time.Time // agentType -> until cancels map[string]context.CancelFunc // taskID → cancel consecutiveFailures map[string]int // agentType -> count + // roleTierIndex tracks a rotating candidate index per "role:tierIndex" key + // so repeated round_robin tier resolutions (see selectRung) actually + // rotate across candidates rather than always picking the first one. + roleTierIndex map[string]int closed bool // set to true when Shutdown has been called resultCh chan *Result startedCh chan string // task IDs that just transitioned to RUNNING @@ -119,6 +130,7 @@ func NewPool(maxConcurrent int, runners map[string]Runner, store Store, logger * rateLimited: make(map[string]time.Time), cancels: make(map[string]context.CancelFunc), consecutiveFailures: make(map[string]int), + roleTierIndex: make(map[string]int), resultCh: make(chan *Result, maxConcurrent*2), startedCh: make(chan string, maxConcurrent*2), workCh: make(chan workItem, maxConcurrent*10+100), @@ -574,6 +586,85 @@ func (p *Pool) AgentStatuses() []AgentStatusInfo { return out } +// decodeRoleConfig unmarshals a storage.RoleConfigRow's raw ConfigJSON into a +// role.RoleConfig. storage intentionally doesn't depend on internal/role, so +// this decode step lives in each consumer package (executor, scheduler, api). +func decodeRoleConfig(row *storage.RoleConfigRow) (role.RoleConfig, error) { + var rc role.RoleConfig + if row == nil { + return rc, fmt.Errorf("nil role config row") + } + if err := json.Unmarshal([]byte(row.ConfigJSON), &rc); err != nil { + return rc, fmt.Errorf("decoding role config: %w", err) + } + return rc, nil +} + +// findTierIndex returns the index of the first tier in ladder whose +// Candidates contains (provider, model), or -1 if none match. Used to +// determine which tier an already-resolved Agent.Type/Model (set by +// internal/scheduler for a retry/escalation resubmit) belongs to, so the new +// execution's EscalationRung can be stamped correctly without re-resolving. +func findTierIndex(ladder []role.Tier, provider, model string) int { + for i, tier := range ladder { + for _, c := range tier.Candidates { + if c.Provider == provider && c.Model == model { + return i + } + } + } + return -1 +} + +// selectRung picks one candidate rung from tier for roleName's tierIdx-th +// tier. "single" selection mode (or a single-candidate tier) always returns +// Candidates[0]. "round_robin" (the default) rotates through candidates +// across successive calls via p.roleTierIndex, skipping any provider +// currently in p.rateLimited; if every candidate is rate-limited it falls +// back to the one whose rate limit clears soonest (same tie-break spirit as +// pickAgent's rate-limited fallback). +func (p *Pool) selectRung(roleName string, tierIdx int, tier role.Tier) role.Rung { + if len(tier.Candidates) == 0 { + return role.Rung{} + } + if tier.EffectiveSelectionMode() == "single" || len(tier.Candidates) == 1 { + return tier.Candidates[0] + } + + key := fmt.Sprintf("%s:%d", roleName, tierIdx) + n := len(tier.Candidates) + now := time.Now() + + p.mu.Lock() + defer p.mu.Unlock() + if p.roleTierIndex == nil { + p.roleTierIndex = make(map[string]int) + } + start := p.roleTierIndex[key] % n + p.roleTierIndex[key] = (start + 1) % n + + for i := 0; i < n; i++ { + idx := (start + i) % n + cand := tier.Candidates[idx] + if deadline, limited := p.rateLimited[cand.Provider]; !limited || now.After(deadline) { + return cand + } + } + + // All candidates are currently rate-limited: fall back to the one + // clearing soonest. + bestIdx := 0 + var bestDeadline time.Time + for i, cand := range tier.Candidates { + d := p.rateLimited[cand.Provider] + if i == 0 || d.Before(bestDeadline) { + bestDeadline = d + bestIdx = i + } + } + return tier.Candidates[bestIdx] +} + // pickAgent selects the best agent from the given SystemStatus using explicit // load balancing: prefer the available (non-rate-limited) agent with the fewest // active tasks. If all agents are rate-limited, fall back to fewest active. @@ -608,6 +699,50 @@ func pickAgent(status SystemStatus) string { func (p *Pool) execute(ctx context.Context, t *task.Task) { defer p.releaseDispatchSlot() + // 0. Role-based dispatch (additive; every existing task shape has + // Agent.Role == "" and takes none of the branches below). For a + // role-typed task that hasn't yet been assigned a concrete Agent.Type + // (the initial dispatch — Type == ""), resolve tier 0 of the active + // role_configs row's EscalationLadder and apply it to a copy of t, the + // same copy-before-mutate pattern withFailureHistory uses. Escalating to + // later tiers on failure is internal/scheduler's job, not execute()'s — + // by the time a role-typed task reaches execute() a second time with + // Agent.Type already set (scheduler-driven retry/escalation resubmit), + // this block only looks up which tier that (Type, Model) pair belongs to + // (read-only) so the new execution's EscalationRung can be stamped + // correctly; it does not re-resolve or mutate the task. + resolvedRung := -1 + if t.Agent.Role != "" { + if row, err := p.store.GetActiveRoleConfig(t.Agent.Role); err != nil { + p.logger.Warn("no active role config; dispatching without role resolution", "role", t.Agent.Role, "taskID", t.ID, "error", err) + } else if rc, err := decodeRoleConfig(row); err != nil { + p.logger.Error("failed to decode role config", "role", t.Agent.Role, "taskID", t.ID, "error", err) + } else if len(rc.EscalationLadder) > 0 { + if t.Agent.Type == "" { + tier0 := rc.EscalationLadder[0] + selected := p.selectRung(t.Agent.Role, 0, tier0) + if selected.Provider != "" { + nt := *t + nt.Agent = t.Agent + nt.Agent.Type = selected.Provider + nt.Agent.Model = selected.Model + if rc.SystemPrompt != "" { + if nt.Agent.SystemPromptAppend != "" { + nt.Agent.SystemPromptAppend = rc.SystemPrompt + "\n\n" + nt.Agent.SystemPromptAppend + } else { + nt.Agent.SystemPromptAppend = rc.SystemPrompt + } + } + t = &nt + resolvedRung = 0 + p.logger.Info("role dispatch resolved tier 0", "role", t.Agent.Role, "taskID", t.ID, "provider", selected.Provider, "model", selected.Model) + } + } else { + resolvedRung = findTierIndex(rc.EscalationLadder, t.Agent.Type, t.Agent.Model) + } + } + } + // 1. Load-balanced agent selection + model classification. p.mu.Lock() activeTasks := make(map[string]int) @@ -766,12 +901,17 @@ func (p *Pool) execute(ctx context.Context, t *task.Task) { } execID := uuid.New().String() + escalationRung := resolvedRung + if escalationRung < 0 { + escalationRung = 0 + } exec := &storage.Execution{ - ID: execID, - TaskID: t.ID, - StartTime: time.Now().UTC(), - Status: "RUNNING", - Agent: agentType, + ID: execID, + TaskID: t.ID, + StartTime: time.Now().UTC(), + Status: "RUNNING", + Agent: agentType, + EscalationRung: escalationRung, } // Pre-populate log paths so they're available in the DB immediately — diff --git a/internal/executor/executor_test.go b/internal/executor/executor_test.go index cb6554c..8eb7e72 100644 --- a/internal/executor/executor_test.go +++ b/internal/executor/executor_test.go @@ -2,6 +2,7 @@ package executor import ( "context" + "database/sql" "fmt" "log/slog" "os" @@ -1162,6 +1163,9 @@ func (m *minimalMockStore) RecordAgentEvent(_ storage.AgentEvent) error { func (m *minimalMockStore) GetProject(_ string) (*task.Project, error) { return nil, nil } func (m *minimalMockStore) CreateTask(_ *task.Task) error { return nil } func (m *minimalMockStore) CreateEvent(_ *event.Event) error { return nil } +func (m *minimalMockStore) GetActiveRoleConfig(_ string) (*storage.RoleConfigRow, error) { + return nil, sql.ErrNoRows +} func (m *minimalMockStore) lastStateUpdate() (string, task.State, bool) { m.mu.Lock() diff --git a/internal/executor/groq_routing_test.go b/internal/executor/groq_routing_test.go new file mode 100644 index 0000000..722bf95 --- /dev/null +++ b/internal/executor/groq_routing_test.go @@ -0,0 +1,98 @@ +package executor + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "sync/atomic" + "testing" + + "github.com/thepeterstone/claudomator/internal/llm" + "github.com/thepeterstone/claudomator/internal/provider/openaicompat" +) + +// TestPool_Submit_GroqAgentType_RoutesThroughOpenAICompatProvider is the +// Part A wiring-level confirmation (verification item 5) that Groq — one of +// the three providers that docs/api-keys-setup.md documented as "works once +// configured" but was never actually registered by internal/cli/serve.go or +// run.go before this phase — is a real, usable runner once wired up. Groq is +// OpenAI-wire-compatible, so this reuses the existing Phase 1 +// internal/provider/openaicompat adapter (internal/llm.Client) exactly the +// way internal/cli/cloudrunners.go's registerCloudRunners constructs it: no +// new adapter package, purely config-driven construction pointed at Groq's +// endpoint. +func TestPool_Submit_GroqAgentType_RoutesThroughOpenAICompatProvider(t *testing.T) { + var calls int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt32(&calls, 1) + if r.URL.Path != "/chat/completions" { + t.Errorf("unexpected path %q", r.URL.Path) + } + if got := r.Header.Get("Authorization"); got != "Bearer test-groq-key" { + t.Errorf("wrong Authorization header: %q", got) + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "choices": []map[string]any{ + { + "message": map[string]any{"role": "assistant", "content": "## Summary\ndone"}, + "finish_reason": "stop", + }, + }, + "model": "llama-3.3-70b-versatile", + "usage": map[string]any{"prompt_tokens": 10, "completion_tokens": 3}, + }) + })) + defer srv.Close() + + // Mirrors registerCloudRunners' construction for groq/openrouter/openai: + // openaicompat.New(&llm.Client{...}) wrapped in a NativeRunner registered + // under the "groq" runner-map key, SandboxKind "docker" for the real + // wiring (irrelevant here since the task has no ProjectDir/sandbox use). + client := &llm.Client{ + Endpoint: srv.URL, + Model: "llama-3.3-70b-versatile", + APIKey: "test-groq-key", + } + runners := map[string]Runner{ + "groq": &NativeRunner{ + Provider: openaicompat.New(client), + Logger: slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})), + LogDir: t.TempDir(), + }, + } + + store := testStore(t) + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, runners, store, logger) + + tk := makeTask("groq-routing-1") + tk.Agent.Type = "groq" + tk.Agent.Model = "llama-3.3-70b-versatile" + tk.Agent.SkipPlanning = true + if err := store.CreateTask(tk); err != nil { + t.Fatalf("CreateTask: %v", err) + } + + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("Submit: %v", err) + } + + result := <-pool.Results() + if result.Err != nil { + t.Fatalf("expected no error, got: %v", result.Err) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("expected exactly 1 call to the fake Groq server, got %d", got) + } + if result.Execution.Agent != "groq" { + t.Errorf("execution.Agent: want %q, got %q", "groq", result.Execution.Agent) + } + if result.Execution.TokensIn != 10 || result.Execution.TokensOut != 3 { + t.Errorf("tokens should come from the openaicompat provider's usage: got in=%d out=%d", + result.Execution.TokensIn, result.Execution.TokensOut) + } +} diff --git a/internal/executor/role_dispatch_test.go b/internal/executor/role_dispatch_test.go new file mode 100644 index 0000000..5628bdb --- /dev/null +++ b/internal/executor/role_dispatch_test.go @@ -0,0 +1,206 @@ +package executor + +import ( + "context" + "encoding/json" + "log/slog" + "os" + "testing" + "time" + + "github.com/thepeterstone/claudomator/internal/role" +) + +// TestPool_Execute_RoleTypedTask_ResolvesTier0 is the Pool.execute()-level +// verification (Phase 5 item 3) that a role-typed task with an empty +// Agent.Type resolves tier 0 of the active role_configs ladder before +// dispatch: it should run through the tier-0 provider/model, and the +// resulting execution should be stamped with EscalationRung 0. +func TestPool_Execute_RoleTypedTask_ResolvesTier0(t *testing.T) { + store := testStore(t) + + rc := role.RoleConfig{ + Role: "coder", + EscalationLadder: []role.Tier{ + { + Candidates: []role.Rung{{Provider: "groq", Model: "llama-3.3-70b-versatile"}}, + SelectionMode: "single", + MaxRetries: 1, + }, + { + Candidates: []role.Rung{{Provider: "anthropic", Model: "claude-sonnet-5"}}, + MaxRetries: 1, + }, + }, + } + configJSON, err := json.Marshal(rc) + if err != nil { + t.Fatalf("marshal role config: %v", err) + } + created, err := store.CreateRoleConfig("coder", string(configJSON), "human") + if err != nil { + t.Fatalf("CreateRoleConfig: %v", err) + } + if err := store.ActivateRoleConfigVersion("coder", created.Version); err != nil { + t.Fatalf("ActivateRoleConfigVersion: %v", err) + } + + runners := map[string]Runner{ + "groq": &mockRunner{}, + } + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, runners, store, logger) + + tk := makeTask("role-task-1") + tk.Agent.Type = "" + tk.Agent.Model = "" + tk.Agent.Role = "coder" + tk.Agent.SkipPlanning = true + if err := store.CreateTask(tk); err != nil { + t.Fatalf("CreateTask: %v", err) + } + + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("Submit: %v", err) + } + + select { + case result := <-pool.Results(): + if result.Err != nil { + t.Fatalf("expected no error, got: %v", result.Err) + } + if result.Execution.Agent != "groq" { + t.Errorf("execution.Agent: want %q, got %q", "groq", result.Execution.Agent) + } + if result.Execution.EscalationRung != 0 { + t.Errorf("execution.EscalationRung: want 0, got %d", result.Execution.EscalationRung) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for result") + } + + persisted, err := store.GetTask(tk.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if persisted.Agent.Type != "groq" { + t.Errorf("persisted Agent.Type: want %q, got %q", "groq", persisted.Agent.Type) + } + if persisted.Agent.Model != "llama-3.3-70b-versatile" { + t.Errorf("persisted Agent.Model: want %q, got %q", "llama-3.3-70b-versatile", persisted.Agent.Model) + } +} + +// TestPool_Execute_NonRoleTask_Unaffected proves that a task with +// Agent.Role == "" (every existing YAML/chatbot task shape) takes none of +// the role-resolution branches in execute() — it dispatches exactly as it +// did before this phase, using its explicitly-set Agent.Type/Model verbatim. +func TestPool_Execute_NonRoleTask_Unaffected(t *testing.T) { + store := testStore(t) + runners := map[string]Runner{ + "claude": &mockRunner{}, + } + logger := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + pool := NewPool(2, runners, store, logger) + + tk := makeTask("non-role-task-1") + tk.Agent.Type = "claude" + tk.Agent.Model = "sonnet" + // Agent.Role intentionally left "" (zero value). + if err := store.CreateTask(tk); err != nil { + t.Fatalf("CreateTask: %v", err) + } + + if err := pool.Submit(context.Background(), tk); err != nil { + t.Fatalf("Submit: %v", err) + } + + select { + case result := <-pool.Results(): + if result.Err != nil { + t.Fatalf("expected no error, got: %v", result.Err) + } + if result.Execution.Agent != "claude" { + t.Errorf("execution.Agent: want %q, got %q", "claude", result.Execution.Agent) + } + if result.Execution.EscalationRung != 0 { + t.Errorf("non-role task execution.EscalationRung should default to 0, got %d", result.Execution.EscalationRung) + } + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for result") + } + + persisted, err := store.GetTask(tk.ID) + if err != nil { + t.Fatalf("GetTask: %v", err) + } + if persisted.Agent.Type != "claude" || persisted.Agent.Model != "sonnet" { + t.Errorf("non-role task's Agent.Type/Model should be unchanged: got %q/%q", persisted.Agent.Type, persisted.Agent.Model) + } +} + +// TestPool_SelectRung_RoundRobin_SkipsRateLimited proves that repeated +// selectRung calls on a round_robin tier rotate across candidates rather +// than always returning candidate 0, and skip a candidate whose provider is +// currently rate-limited. +func TestPool_SelectRung_RoundRobin_SkipsRateLimited(t *testing.T) { + p := &Pool{ + rateLimited: make(map[string]time.Time), + roleTierIndex: make(map[string]int), + } + tier := role.Tier{ + Candidates: []role.Rung{ + {Provider: "local", Model: "m1"}, + {Provider: "groq", Model: "m2"}, + {Provider: "openrouter", Model: "m3"}, + }, + SelectionMode: "round_robin", + } + + // With nothing rate-limited, repeated calls should rotate through all + // three candidates before repeating. + seen := map[string]int{} + for i := 0; i < 6; i++ { + r := p.selectRung("coder", 0, tier) + seen[r.Provider]++ + } + for _, prov := range []string{"local", "groq", "openrouter"} { + if seen[prov] != 2 { + t.Errorf("expected provider %q to be picked exactly twice across 6 calls, got %d (seen=%v)", prov, seen[prov], seen) + } + } + + // Now rate-limit "groq" and confirm it's skipped on subsequent picks. + p.mu.Lock() + p.rateLimited["groq"] = time.Now().Add(1 * time.Hour) + p.mu.Unlock() + + for i := 0; i < 6; i++ { + r := p.selectRung("coder", 0, tier) + if r.Provider == "groq" { + t.Errorf("expected rate-limited provider %q to be skipped, got picked at iteration %d", "groq", i) + } + } +} + +// TestPool_SelectRung_SingleMode_AlwaysFirstCandidate proves "single" +// selection mode never rotates. +func TestPool_SelectRung_SingleMode_AlwaysFirstCandidate(t *testing.T) { + p := &Pool{ + rateLimited: make(map[string]time.Time), + roleTierIndex: make(map[string]int), + } + tier := role.Tier{ + Candidates: []role.Rung{ + {Provider: "local", Model: "m1"}, + {Provider: "groq", Model: "m2"}, + }, + SelectionMode: "single", + } + for i := 0; i < 4; i++ { + r := p.selectRung("coder", 0, tier) + if r.Provider != "local" { + t.Errorf("iteration %d: expected always \"local\" in single mode, got %q", i, r.Provider) + } + } +} diff --git a/internal/role/role.go b/internal/role/role.go new file mode 100644 index 0000000..f65edba --- /dev/null +++ b/internal/role/role.go @@ -0,0 +1,73 @@ +// Package role defines the per-role configuration shape (system prompt, tool +// allowlist, sandbox kind, budget, and — the part actually driven this phase — +// a multi-tier provider/model escalation ladder) used by the token-husbanding +// harness. A RoleConfig is stored as a versioned row in storage's +// role_configs table (see internal/storage/roleconfig.go); internal/executor +// resolves tier 0 of a role's EscalationLadder for the initial dispatch of a +// role-typed task, and internal/scheduler walks the ladder on failure +// (retry-then-escalate). +// +// Not everything on RoleConfig is enforced yet. See the field doc comments +// below and CLAUDE.md's Design Debt section for what's stored-but-not-wired, +// matching how this codebase already documents task.Priority/RetryConfig. +package role + +// RoleConfig is one version of a role's configuration. +type RoleConfig struct { + // Role names the role this config applies to (e.g. "coder", "reviewer"). + Role string `json:"role"` + + // SystemPrompt is appended to Agent.SystemPromptAppend when a task + // dispatches through this role (internal/executor.Pool.execute()). Wired. + SystemPrompt string `json:"system_prompt,omitempty"` + + // Tools, SandboxKind, and DefaultBudgetUSD are stored and round-tripped + // through the API/storage layer but are NOT YET enforced by the executor: + // guardrails/tool availability are still applied uniformly by + // NativeRunner (Phase 3) regardless of role, and SandboxKind here has no + // effect on which sandbox.Sandbox implementation a runner picks (that's + // still purely provider-driven, see NativeRunner.SandboxKind). A later + // phase should thread these through. DefaultBudgetUSD is read by + // internal/scheduler as the estimated cost passed to + // budget.Accountant.Allow() when considering an escalation, which is a + // narrow, specific use — not general per-role budget enforcement at + // dispatch time. + Tools []string `json:"tools,omitempty"` + SandboxKind string `json:"sandbox_kind,omitempty"` + DefaultBudgetUSD float64 `json:"default_budget_usd,omitempty"` + + // EscalationLadder is the ordered list of tiers a role-typed task climbs + // on repeated failure. Tier 0 is used for the initial dispatch. + EscalationLadder []Tier `json:"escalation_ladder,omitempty"` +} + +// Tier is one rung-group in an escalation ladder: a set of candidate +// (provider, model) rungs to choose among, plus how many attempts may be made +// at this tier before the scheduler escalates to the next one. +type Tier struct { + Candidates []Rung `json:"candidates"` + // SelectionMode is "round_robin" (default, when empty) or "single" (always + // use Candidates[0]). + SelectionMode string `json:"selection_mode,omitempty"` + // MaxRetries is the number of attempts allowed at this tier before the + // scheduler escalates to the next tier. 0 means "escalate immediately + // after the first failure at this tier" (no retries held here). + MaxRetries int `json:"max_retries"` +} + +// Rung is a single (provider, model) pair. Provider must match a registered +// executor runner key: "local", "anthropic", "google", "groq", "openrouter", +// "openai". +type Rung struct { + Provider string `json:"provider"` + Model string `json:"model"` +} + +// EffectiveSelectionMode returns t.SelectionMode, defaulting to +// "round_robin" when unset. +func (t Tier) EffectiveSelectionMode() string { + if t.SelectionMode == "" { + return "round_robin" + } + return t.SelectionMode +} diff --git a/internal/role/role_test.go b/internal/role/role_test.go new file mode 100644 index 0000000..6d2294e --- /dev/null +++ b/internal/role/role_test.go @@ -0,0 +1,73 @@ +package role + +import ( + "encoding/json" + "reflect" + "testing" +) + +func TestRoleConfig_JSONRoundTrip(t *testing.T) { + rc := RoleConfig{ + Role: "coder", + SystemPrompt: "You are a careful coding agent.", + Tools: []string{"Bash", "Read", "Edit"}, + SandboxKind: "docker", + DefaultBudgetUSD: 1.5, + EscalationLadder: []Tier{ + { + Candidates: []Rung{ + {Provider: "local", Model: "llama3.1:8b"}, + }, + SelectionMode: "single", + MaxRetries: 2, + }, + { + Candidates: []Rung{ + {Provider: "groq", Model: "llama-3.3-70b-versatile"}, + {Provider: "openrouter", Model: "meta-llama/llama-3.3-70b-instruct:free"}, + }, + SelectionMode: "round_robin", + MaxRetries: 1, + }, + { + Candidates: []Rung{ + {Provider: "anthropic", Model: "claude-sonnet-5"}, + }, + MaxRetries: 0, + }, + }, + } + + b, err := json.Marshal(rc) + if err != nil { + t.Fatalf("Marshal: %v", err) + } + + var got RoleConfig + if err := json.Unmarshal(b, &got); err != nil { + t.Fatalf("Unmarshal: %v", err) + } + + if !reflect.DeepEqual(rc, got) { + t.Errorf("round trip mismatch:\n want: %+v\n got: %+v", rc, got) + } +} + +func TestTier_EffectiveSelectionMode(t *testing.T) { + cases := []struct { + name string + tier Tier + want string + }{ + {"empty defaults to round_robin", Tier{}, "round_robin"}, + {"explicit single", Tier{SelectionMode: "single"}, "single"}, + {"explicit round_robin", Tier{SelectionMode: "round_robin"}, "round_robin"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + if got := c.tier.EffectiveSelectionMode(); got != c.want { + t.Errorf("got %q, want %q", got, c.want) + } + }) + } +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go new file mode 100644 index 0000000..b3756fc --- /dev/null +++ b/internal/scheduler/scheduler.go @@ -0,0 +1,277 @@ +// Package scheduler implements a first-pass retry-then-escalate loop for +// role-typed tasks (task.AgentConfig.Role != ""). On a poll interval it looks +// for tasks whose most recent execution ended FAILED, and either resubmits +// them at the same escalation-ladder tier (if under that tier's MaxRetries) +// or escalates them to the next tier (if the budget allows) — recording an +// event.KindEscalated event either way. If the ladder is exhausted or the +// budget denies the escalation, the task is left FAILED for human attention. +// +// Explicit non-goals for this phase (see the Phase 5 task description): +// no AskUser-timeout escalation, no DAG/cascade-fail logic. Handling for +// TIMED_OUT/CANCELLED/BUDGET_EXCEEDED tasks follows the same shape as FAILED +// but isn't implemented yet — only FAILED is polled. +package scheduler + +import ( + "context" + "encoding/json" + "log/slog" + "sync" + "time" + + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/role" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +// Store is the subset of storage.DB methods the Scheduler needs. +type Store interface { + ListTasks(filter storage.TaskFilter) ([]*task.Task, error) + ListExecutions(taskID string) ([]*storage.Execution, error) + GetActiveRoleConfig(role string) (*storage.RoleConfigRow, error) + UpdateTaskAgent(id string, agent task.AgentConfig) error + UpdateTaskState(id string, newState task.State) error + CreateEvent(e *event.Event) error +} + +// Pool is the subset of *executor.Pool the Scheduler needs. Satisfied by +// *executor.Pool directly (see internal/cli/serve.go); declared as an +// interface here purely so tests can supply a fake without dragging in the +// executor package's runner/sandbox machinery. +type Pool interface { + Submit(ctx context.Context, t *task.Task) error +} + +// BudgetGate reports whether an escalation to provider estimated at estCost +// is allowed. Satisfied by *budget.Accountant. +type BudgetGate interface { + Allow(provider string, estCost float64) (bool, error) +} + +// Scheduler polls for role-typed FAILED tasks and retries or escalates them +// per their active role_configs escalation ladder. +type Scheduler struct { + Store Store + Pool Pool + Budget BudgetGate // nil means "no budget gating" (always allow) + Logger *slog.Logger + + // handled dedupes processing within a single running process: once a + // decision (retry/escalate/decline) has been made for a given + // execution ID, it is never reconsidered again by this Scheduler + // instance. This is what keeps Run's poll loop convergent — a task left + // FAILED after its ladder is exhausted (or an escalation is budget- + // denied) has the same "latest execution" on every subsequent tick, so + // without this it would emit a fresh "final" KindEscalated event, and + // re-run the same decision, every single poll forever. A restart clears + // this map, so a task can be reconsidered once more after a restart — + // intentional: it's an idempotent bookkeeping decision, not orchestration + // state, so re-deriving it once is harmless. + mu sync.Mutex + handled map[string]bool +} + +// DefaultPollInterval is used by Run when pollInterval <= 0. +const DefaultPollInterval = 30 * time.Second + +// Run polls for role-typed FAILED tasks every pollInterval until ctx is +// cancelled. +func (s *Scheduler) Run(ctx context.Context, pollInterval time.Duration) { + if pollInterval <= 0 { + pollInterval = DefaultPollInterval + } + ticker := time.NewTicker(pollInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + s.Tick(ctx) + } + } +} + +// Tick runs a single poll pass. Exported so tests can drive it directly +// without waiting on a ticker. +func (s *Scheduler) Tick(ctx context.Context) { + tasks, err := s.Store.ListTasks(storage.TaskFilter{State: task.StateFailed}) + if err != nil { + s.logf("scheduler: list failed tasks", "error", err) + return + } + for _, t := range tasks { + if t.Agent.Role == "" { + continue + } + s.processTask(ctx, t) + } +} + +func (s *Scheduler) logf(msg string, args ...any) { + if s.Logger != nil { + s.Logger.Warn(msg, args...) + } +} + +func (s *Scheduler) markHandled(execID string) bool { + s.mu.Lock() + defer s.mu.Unlock() + if s.handled == nil { + s.handled = make(map[string]bool) + } + if s.handled[execID] { + return false + } + s.handled[execID] = true + return true +} + +func (s *Scheduler) processTask(ctx context.Context, t *task.Task) { + execs, err := s.Store.ListExecutions(t.ID) + if err != nil || len(execs) == 0 { + return + } + latest := execs[0] // ListExecutions orders DESC by start_time. + if latest.Status != "FAILED" { + return + } + if !s.markHandled(latest.ID) { + return // already decided for this execution — converged, nothing to do. + } + + row, err := s.Store.GetActiveRoleConfig(t.Agent.Role) + if err != nil { + s.logf("scheduler: no active role config for role", "role", t.Agent.Role, "taskID", t.ID, "error", err) + return + } + var rc role.RoleConfig + if err := json.Unmarshal([]byte(row.ConfigJSON), &rc); err != nil { + s.logf("scheduler: decode role config", "role", t.Agent.Role, "taskID", t.ID, "error", err) + return + } + if len(rc.EscalationLadder) == 0 { + return + } + + currentRung := latest.EscalationRung + if currentRung < 0 { + currentRung = 0 + } + if currentRung >= len(rc.EscalationLadder) { + // Ladder already exhausted (e.g. the ladder was shortened after this + // task started climbing it) — nothing more to do. + return + } + tier := rc.EscalationLadder[currentRung] + attempts := attemptsAtRung(execs, currentRung) + + if attempts < tier.MaxRetries { + s.retrySameRung(ctx, t, currentRung) + return + } + + nextRung := currentRung + 1 + if nextRung >= len(rc.EscalationLadder) || len(rc.EscalationLadder[nextRung].Candidates) == 0 { + s.decline(ctx, t, currentRung, "", "escalation ladder exhausted") + return + } + nextTier := rc.EscalationLadder[nextRung] + target := nextTier.Candidates[0] + + estCost := rc.DefaultBudgetUSD + if estCost <= 0 { + estCost = t.Agent.MaxBudgetUSD + } + allowed := true + if s.Budget != nil { + var berr error + allowed, berr = s.Budget.Allow(target.Provider, estCost) + if berr != nil { + s.logf("scheduler: budget check failed; declining escalation", "taskID", t.ID, "error", berr) + allowed = false + } + } + if !allowed { + s.decline(ctx, t, currentRung, target.Provider, "budget denied") + return + } + s.escalate(ctx, t, currentRung, nextRung, target) +} + +// attemptsAtRung counts how many executions, starting from the most recent +// (execs[0]) and moving backward, ran at rung consecutively — i.e. how many +// attempts have already been made at the task's current tier since it last +// moved to (or started at) that tier. +func attemptsAtRung(execs []*storage.Execution, rung int) int { + n := 0 + for _, e := range execs { + if e.EscalationRung != rung { + break + } + n++ + } + return n +} + +func (s *Scheduler) retrySameRung(ctx context.Context, t *task.Task, rung int) { + if err := s.Store.UpdateTaskState(t.ID, task.StateQueued); err != nil { + s.logf("scheduler: retry: update task state", "taskID", t.ID, "error", err) + return + } + resubmit := *t + resubmit.State = task.StateQueued + if err := s.Pool.Submit(ctx, &resubmit); err != nil { + s.logf("scheduler: retry: submit", "taskID", t.ID, "error", err) + } +} + +func (s *Scheduler) escalate(ctx context.Context, t *task.Task, fromRung, toRung int, target role.Rung) { + fromProvider := t.Agent.Type + newAgent := t.Agent + newAgent.Type = target.Provider + newAgent.Model = target.Model + if err := s.Store.UpdateTaskAgent(t.ID, newAgent); err != nil { + s.logf("scheduler: escalate: update task agent", "taskID", t.ID, "error", err) + return + } + if err := s.Store.UpdateTaskState(t.ID, task.StateQueued); err != nil { + s.logf("scheduler: escalate: update task state", "taskID", t.ID, "error", err) + return + } + s.emitEscalated(t.ID, fromRung, toRung, fromProvider, target.Provider, false, "") + + resubmit := *t + resubmit.Agent = newAgent + resubmit.State = task.StateQueued + if err := s.Pool.Submit(ctx, &resubmit); err != nil { + s.logf("scheduler: escalate: submit", "taskID", t.ID, "error", err) + } +} + +// decline records that no further escalation will happen for t right now +// (ladder exhausted or budget denied) and leaves it FAILED for human +// attention. +func (s *Scheduler) decline(ctx context.Context, t *task.Task, atRung int, consideredProvider, reason string) { + s.emitEscalated(t.ID, atRung, atRung, t.Agent.Type, consideredProvider, true, reason) +} + +func (s *Scheduler) emitEscalated(taskID string, fromRung, toRung int, fromProvider, toProvider string, final bool, reason string) { + payload, _ := json.Marshal(struct { + FromRung int `json:"from_rung"` + ToRung int `json:"to_rung"` + FromProvider string `json:"from_provider"` + ToProvider string `json:"to_provider,omitempty"` + Final bool `json:"final"` + Reason string `json:"reason,omitempty"` + }{FromRung: fromRung, ToRung: toRung, FromProvider: fromProvider, ToProvider: toProvider, Final: final, Reason: reason}) + if err := s.Store.CreateEvent(&event.Event{ + TaskID: taskID, + Kind: event.KindEscalated, + Actor: event.ActorSystem, + Payload: payload, + }); err != nil { + s.logf("scheduler: emit escalated event", "taskID", taskID, "error", err) + } +} diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go new file mode 100644 index 0000000..18c9e42 --- /dev/null +++ b/internal/scheduler/scheduler_test.go @@ -0,0 +1,369 @@ +package scheduler + +import ( + "context" + "encoding/json" + "errors" + "sync" + "testing" + "time" + + "github.com/google/uuid" + "github.com/thepeterstone/claudomator/internal/event" + "github.com/thepeterstone/claudomator/internal/role" + "github.com/thepeterstone/claudomator/internal/storage" + "github.com/thepeterstone/claudomator/internal/task" +) + +// fakeStore is a minimal, in-memory implementation of Store for unit-testing +// the Scheduler without a real SQLite database. +type fakeStore struct { + mu sync.Mutex + tasks map[string]*task.Task + execsByTask map[string][]*storage.Execution // index 0 = most recent (matches ListExecutions' DESC order) + roleConfigs map[string]*storage.RoleConfigRow + agentUpdates []task.AgentConfig + stateUpdates []task.State + events []*event.Event +} + +func newFakeStore() *fakeStore { + return &fakeStore{ + tasks: make(map[string]*task.Task), + execsByTask: make(map[string][]*storage.Execution), + roleConfigs: make(map[string]*storage.RoleConfigRow), + } +} + +func (f *fakeStore) ListTasks(filter storage.TaskFilter) ([]*task.Task, error) { + f.mu.Lock() + defer f.mu.Unlock() + var out []*task.Task + for _, t := range f.tasks { + if filter.State != "" && t.State != filter.State { + continue + } + out = append(out, t) + } + return out, nil +} + +func (f *fakeStore) ListExecutions(taskID string) ([]*storage.Execution, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.execsByTask[taskID], nil +} + +func (f *fakeStore) GetActiveRoleConfig(roleName string) (*storage.RoleConfigRow, error) { + f.mu.Lock() + defer f.mu.Unlock() + row, ok := f.roleConfigs[roleName] + if !ok { + return nil, errors.New("not found") + } + return row, nil +} + +func (f *fakeStore) UpdateTaskAgent(id string, agent task.AgentConfig) error { + f.mu.Lock() + defer f.mu.Unlock() + f.agentUpdates = append(f.agentUpdates, agent) + if t, ok := f.tasks[id]; ok { + t.Agent = agent + } + return nil +} + +func (f *fakeStore) UpdateTaskState(id string, newState task.State) error { + f.mu.Lock() + defer f.mu.Unlock() + f.stateUpdates = append(f.stateUpdates, newState) + if t, ok := f.tasks[id]; ok { + t.State = newState + } + return nil +} + +func (f *fakeStore) CreateEvent(e *event.Event) error { + f.mu.Lock() + defer f.mu.Unlock() + e.ID = uuid.NewString() + f.events = append(f.events, e) + return nil +} + +func (f *fakeStore) eventsOfKind(k event.Kind) []*event.Event { + f.mu.Lock() + defer f.mu.Unlock() + var out []*event.Event + for _, e := range f.events { + if e.Kind == k { + out = append(out, e) + } + } + return out +} + +// fakePool records every task submitted to it. +type fakePool struct { + mu sync.Mutex + submitted []*task.Task + err error +} + +func (f *fakePool) Submit(_ context.Context, t *task.Task) error { + f.mu.Lock() + defer f.mu.Unlock() + f.submitted = append(f.submitted, t) + return f.err +} + +func (f *fakePool) submitCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.submitted) +} + +// fakeBudget is a configurable BudgetGate. +type fakeBudget struct { + allow bool + err error +} + +func (f *fakeBudget) Allow(_ string, _ float64) (bool, error) { return f.allow, f.err } + +func twoTierLadder() role.RoleConfig { + return role.RoleConfig{ + Role: "coder", + EscalationLadder: []role.Tier{ + { + Candidates: []role.Rung{{Provider: "local", Model: "m0"}}, + SelectionMode: "single", + MaxRetries: 2, + }, + { + Candidates: []role.Rung{{Provider: "anthropic", Model: "claude-sonnet-5"}}, + MaxRetries: 1, + }, + }, + } +} + +func seedRoleConfig(t *testing.T, f *fakeStore, rc role.RoleConfig) { + t.Helper() + b, err := json.Marshal(rc) + if err != nil { + t.Fatalf("marshal role config: %v", err) + } + f.roleConfigs[rc.Role] = &storage.RoleConfigRow{ + ID: uuid.NewString(), Role: rc.Role, Version: 1, Status: "active", ConfigJSON: string(b), + } +} + +func failedTask(id, roleName string, agentType string) *task.Task { + return &task.Task{ + ID: id, + Name: "test", + Agent: task.AgentConfig{Type: agentType, Role: roleName, MaxBudgetUSD: 0.5}, + State: task.StateFailed, + } +} + +func failedExec(rung int) *storage.Execution { + return &storage.Execution{ + ID: uuid.NewString(), + Status: "FAILED", + EscalationRung: rung, + StartTime: time.Now(), + } +} + +// TestScheduler_RetriesSameRung_WhileUnderMaxRetries proves that a task +// whose current rung has fewer attempts than tier.MaxRetries is resubmitted +// at the same rung, with no escalation event and no Agent.Type/Model change. +func TestScheduler_RetriesSameRung_WhileUnderMaxRetries(t *testing.T) { + store := newFakeStore() + seedRoleConfig(t, store, twoTierLadder()) + + tk := failedTask("t1", "coder", "local") + store.tasks[tk.ID] = tk + // Only 1 attempt so far at rung 0; tier0.MaxRetries == 2, so 1 < 2 → retry. + store.execsByTask[tk.ID] = []*storage.Execution{failedExec(0)} + + pool := &fakePool{} + sch := &Scheduler{Store: store, Pool: pool} + sch.Tick(context.Background()) + + if pool.submitCount() != 1 { + t.Fatalf("expected 1 submission, got %d", pool.submitCount()) + } + resubmitted := pool.submitted[0] + if resubmitted.Agent.Type != "local" { + t.Errorf("retry should keep the same provider: got %q", resubmitted.Agent.Type) + } + if resubmitted.State != task.StateQueued { + t.Errorf("resubmitted task should be QUEUED, got %v", resubmitted.State) + } + if len(store.eventsOfKind(event.KindEscalated)) != 0 { + t.Errorf("a same-rung retry should not emit a KindEscalated event") + } + if len(store.agentUpdates) != 0 { + t.Errorf("a same-rung retry should not change the task's Agent config, got %d UpdateTaskAgent calls", len(store.agentUpdates)) + } +} + +// TestScheduler_EscalatesToNextRung_WhenBudgetAllows proves that once a +// tier's MaxRetries is exhausted, the scheduler escalates to the next tier +// (when the budget allows it), updates the task's Agent to the new rung's +// provider/model, resubmits, and emits a non-final KindEscalated event. +func TestScheduler_EscalatesToNextRung_WhenBudgetAllows(t *testing.T) { + store := newFakeStore() + seedRoleConfig(t, store, twoTierLadder()) + + tk := failedTask("t1", "coder", "local") + store.tasks[tk.ID] = tk + // 2 consecutive failed attempts at rung 0; tier0.MaxRetries == 2, so + // 2 < 2 is false → escalate. + store.execsByTask[tk.ID] = []*storage.Execution{failedExec(0), failedExec(0)} + + pool := &fakePool{} + sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: true}} + sch.Tick(context.Background()) + + if pool.submitCount() != 1 { + t.Fatalf("expected 1 submission, got %d", pool.submitCount()) + } + resubmitted := pool.submitted[0] + if resubmitted.Agent.Type != "anthropic" || resubmitted.Agent.Model != "claude-sonnet-5" { + t.Errorf("escalated task should carry tier 1's rung: got %q/%q", resubmitted.Agent.Type, resubmitted.Agent.Model) + } + if resubmitted.State != task.StateQueued { + t.Errorf("resubmitted task should be QUEUED, got %v", resubmitted.State) + } + if len(store.agentUpdates) != 1 { + t.Fatalf("expected 1 UpdateTaskAgent call, got %d", len(store.agentUpdates)) + } + + evs := store.eventsOfKind(event.KindEscalated) + if len(evs) != 1 { + t.Fatalf("expected 1 KindEscalated event, got %d", len(evs)) + } + var payload struct { + FromRung int `json:"from_rung"` + ToRung int `json:"to_rung"` + FromProvider string `json:"from_provider"` + ToProvider string `json:"to_provider"` + Final bool `json:"final"` + } + if err := json.Unmarshal(evs[0].Payload, &payload); err != nil { + t.Fatalf("unmarshal event payload: %v", err) + } + if payload.Final { + t.Errorf("escalation event should have final=false") + } + if payload.FromRung != 0 || payload.ToRung != 1 { + t.Errorf("expected from_rung=0 to_rung=1, got %d -> %d", payload.FromRung, payload.ToRung) + } + if payload.ToProvider != "anthropic" { + t.Errorf("expected to_provider=anthropic, got %q", payload.ToProvider) + } +} + +// TestScheduler_DeclinesFinal_WhenBudgetDenies proves that when the budget +// denies an otherwise-due escalation, the task is left FAILED (no state +// change, no resubmission) and a final:true KindEscalated event is recorded. +func TestScheduler_DeclinesFinal_WhenBudgetDenies(t *testing.T) { + store := newFakeStore() + seedRoleConfig(t, store, twoTierLadder()) + + tk := failedTask("t1", "coder", "local") + store.tasks[tk.ID] = tk + store.execsByTask[tk.ID] = []*storage.Execution{failedExec(0), failedExec(0)} + + pool := &fakePool{} + sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: false}} + sch.Tick(context.Background()) + + if pool.submitCount() != 0 { + t.Fatalf("budget-denied escalation should not resubmit, got %d submissions", pool.submitCount()) + } + if len(store.stateUpdates) != 0 { + t.Errorf("budget-denied escalation should not change task state, got %d state updates", len(store.stateUpdates)) + } + + evs := store.eventsOfKind(event.KindEscalated) + if len(evs) != 1 { + t.Fatalf("expected 1 KindEscalated event, got %d", len(evs)) + } + var payload struct { + Final bool `json:"final"` + } + if err := json.Unmarshal(evs[0].Payload, &payload); err != nil { + t.Fatalf("unmarshal event payload: %v", err) + } + if !payload.Final { + t.Errorf("budget-denied escalation event should have final=true") + } +} + +// TestScheduler_DeclinesFinal_WhenLadderExhausted proves the same +// final:true behavior when the task is already at the ladder's last tier. +func TestScheduler_DeclinesFinal_WhenLadderExhausted(t *testing.T) { + store := newFakeStore() + seedRoleConfig(t, store, twoTierLadder()) + + tk := failedTask("t1", "coder", "anthropic") + store.tasks[tk.ID] = tk + // At rung 1 (the last tier), tier1.MaxRetries == 1, so 1 attempt already + // made means 1 < 1 is false → would escalate, but there is no rung 2. + store.execsByTask[tk.ID] = []*storage.Execution{failedExec(1)} + + pool := &fakePool{} + sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: true}} + sch.Tick(context.Background()) + + if pool.submitCount() != 0 { + t.Fatalf("exhausted ladder should not resubmit, got %d submissions", pool.submitCount()) + } + evs := store.eventsOfKind(event.KindEscalated) + if len(evs) != 1 { + t.Fatalf("expected 1 KindEscalated event, got %d", len(evs)) + } + var payload struct { + Final bool `json:"final"` + } + if err := json.Unmarshal(evs[0].Payload, &payload); err != nil { + t.Fatalf("unmarshal event payload: %v", err) + } + if !payload.Final { + t.Errorf("exhausted-ladder decline event should have final=true") + } +} + +// TestScheduler_Convergence_DoesNotReprocessSameExecution proves the +// double-processing guard: ticking twice against the same unchanged FAILED +// execution only acts once (no duplicate submissions or events), which is +// what keeps a task stuck at the end of its ladder from generating a fresh +// "final" event (or worse, a fresh escalation) on every single poll forever. +func TestScheduler_Convergence_DoesNotReprocessSameExecution(t *testing.T) { + store := newFakeStore() + seedRoleConfig(t, store, twoTierLadder()) + + tk := failedTask("t1", "coder", "anthropic") + store.tasks[tk.ID] = tk + store.execsByTask[tk.ID] = []*storage.Execution{failedExec(1)} // last tier, exhausted + + pool := &fakePool{} + sch := &Scheduler{Store: store, Pool: pool, Budget: &fakeBudget{allow: true}} + + sch.Tick(context.Background()) + sch.Tick(context.Background()) + sch.Tick(context.Background()) + + if got := len(store.eventsOfKind(event.KindEscalated)); got != 1 { + t.Fatalf("expected exactly 1 KindEscalated event across 3 ticks, got %d", got) + } + if pool.submitCount() != 0 { + t.Fatalf("expected 0 submissions, got %d", pool.submitCount()) + } +} diff --git a/internal/storage/db.go b/internal/storage/db.go index 22a3d7b..a16ad4e 100644 --- a/internal/storage/db.go +++ b/internal/storage/db.go @@ -140,6 +140,20 @@ func (s *DB) migrate() error { `CREATE INDEX IF NOT EXISTS idx_events_task_id_seq ON events(task_id, seq)`, `CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts)`, `CREATE INDEX IF NOT EXISTS idx_events_kind ON events(kind)`, + `ALTER TABLE executions ADD COLUMN escalation_rung INTEGER NOT NULL DEFAULT 0`, + `CREATE TABLE IF NOT EXISTS role_configs ( + id TEXT PRIMARY KEY, + role TEXT NOT NULL, + version INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'draft', + config_json TEXT NOT NULL, + created_at DATETIME NOT NULL, + activated_at DATETIME, + retired_at DATETIME, + proposed_by TEXT, + UNIQUE(role, version) + )`, + `CREATE INDEX IF NOT EXISTS idx_role_configs_role_status ON role_configs(role, status)`, } for _, m := range migrations { if _, err := s.db.Exec(m); err != nil { @@ -522,6 +536,15 @@ type Execution struct { SessionID string // claude --session-id; persisted for resume SandboxDir string // preserved sandbox path when task is BLOCKED; resume must run here Agent string // provider that ran this execution (claude/gemini/local); for budget accounting + // EscalationRung is the 0-based index into the active role's + // EscalationLadder this execution ran at, for role-typed tasks + // (task.AgentConfig.Role != ""). 0 (the default) for non-role tasks and + // for role-typed executions where the resolved Agent.Type/Model didn't + // match any tier in the ladder (e.g. the budget gate rerouted to + // "local"). Set once at execution-creation time by + // internal/executor.Pool.execute(); internal/scheduler reads it back to + // know which tier to retry/escalate from. + EscalationRung int Changestats *task.Changestats // stored as JSON; nil if not yet recorded Commits []task.GitCommit // stored as JSON; empty if no commits @@ -570,10 +593,10 @@ func (s *DB) CreateExecutionAndSetRunning(e *Execution) error { commitsJSON = string(b) } if _, err := tx.Exec(` - INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, agent) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)`, + INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, agent, escalation_rung) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?)`, e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status, - e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON, e.Agent, + e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, commitsJSON, e.Agent, e.EscalationRung, ); err != nil { return err } @@ -633,23 +656,23 @@ func (s *DB) CreateExecution(e *Execution) error { commitsJSON = string(b) } _, err := s.db.Exec(` - INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + INSERT INTO executions (id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent, escalation_rung) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, e.ID, e.TaskID, e.StartTime.UTC(), e.EndTime.UTC(), e.ExitCode, e.Status, - e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut, e.Agent, + e.StdoutPath, e.StderrPath, e.ArtifactDir, e.CostUSD, e.ErrorMsg, e.SessionID, e.SandboxDir, changestatsJSON, commitsJSON, e.TokensIn, e.TokensOut, e.Agent, e.EscalationRung, ) return err } // GetExecution retrieves an execution by ID. func (s *DB) GetExecution(id string) (*Execution, error) { - row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE id = ?`, id) + row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent, escalation_rung FROM executions WHERE id = ?`, id) return scanExecution(row) } // ListExecutions returns executions for a task. func (s *DB) ListExecutions(taskID string) ([]*Execution, error) { - rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID) + rows, err := s.db.Query(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent, escalation_rung FROM executions WHERE task_id = ? ORDER BY start_time DESC`, taskID) if err != nil { return nil, err } @@ -668,7 +691,7 @@ func (s *DB) ListExecutions(taskID string) ([]*Execution, error) { // GetLatestExecution returns the most recent execution for a task. func (s *DB) GetLatestExecution(taskID string) (*Execution, error) { - row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID) + row := s.db.QueryRow(`SELECT id, task_id, start_time, end_time, exit_code, status, stdout_path, stderr_path, artifact_dir, cost_usd, error_msg, session_id, sandbox_dir, changestats_json, commits_json, tokens_in, tokens_out, agent, escalation_rung FROM executions WHERE task_id = ? ORDER BY start_time DESC LIMIT 1`, taskID) return scanExecution(row) } @@ -1123,11 +1146,13 @@ func scanExecution(row scanner) (*Execution, error) { var tokensIn sql.NullInt64 var tokensOut sql.NullInt64 var agent sql.NullString + var escalationRung sql.NullInt64 err := row.Scan(&e.ID, &e.TaskID, &e.StartTime, &e.EndTime, &e.ExitCode, &e.Status, - &e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut, &agent) + &e.StdoutPath, &e.StderrPath, &e.ArtifactDir, &e.CostUSD, &e.ErrorMsg, &sessionID, &sandboxDir, &changestatsJSON, &commitsJSON, &tokensIn, &tokensOut, &agent, &escalationRung) if err != nil { return nil, err } + e.EscalationRung = int(escalationRung.Int64) e.SessionID = sessionID.String e.SandboxDir = sandboxDir.String e.TokensIn = tokensIn.Int64 diff --git a/internal/storage/roleconfig.go b/internal/storage/roleconfig.go new file mode 100644 index 0000000..6ae043b --- /dev/null +++ b/internal/storage/roleconfig.go @@ -0,0 +1,147 @@ +package storage + +import ( + "database/sql" + "fmt" + "time" + + "github.com/google/uuid" +) + +// RoleConfigRow is one versioned row in the role_configs table. ConfigJSON +// holds the raw JSON-encoded role.RoleConfig; storage intentionally does not +// import internal/role (no need to — it never inspects the payload, only +// stores/retrieves it), so callers (internal/executor, internal/scheduler, +// internal/api) decode it themselves via encoding/json. +type RoleConfigRow struct { + ID string + Role string + Version int + Status string // "draft" | "active" | "retired" + ConfigJSON string + CreatedAt time.Time + ActivatedAt *time.Time + RetiredAt *time.Time + ProposedBy string +} + +// CreateRoleConfig inserts a new draft version for role, auto-assigning the +// next version number for that role (1 if none exist yet). +func (s *DB) CreateRoleConfig(roleName, configJSON, proposedBy string) (*RoleConfigRow, error) { + tx, err := s.db.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() //nolint:errcheck + + var maxVersion sql.NullInt64 + if err := tx.QueryRow(`SELECT MAX(version) FROM role_configs WHERE role = ?`, roleName).Scan(&maxVersion); err != nil { + return nil, err + } + version := int(maxVersion.Int64) + 1 + + id := uuid.NewString() + now := time.Now().UTC() + if _, err := tx.Exec(` + INSERT INTO role_configs (id, role, version, status, config_json, created_at, proposed_by) + VALUES (?, ?, ?, 'draft', ?, ?, ?)`, + id, roleName, version, configJSON, now, proposedBy, + ); err != nil { + return nil, err + } + if err := tx.Commit(); err != nil { + return nil, err + } + + return &RoleConfigRow{ + ID: id, + Role: roleName, + Version: version, + Status: "draft", + ConfigJSON: configJSON, + CreatedAt: now, + ProposedBy: proposedBy, + }, nil +} + +// GetActiveRoleConfig returns the currently active role_configs row for +// role. Returns sql.ErrNoRows (unwrapped, matching GetTask/GetProject +// convention in this package) if no version is active. +func (s *DB) GetActiveRoleConfig(roleName string) (*RoleConfigRow, error) { + row := s.db.QueryRow(` + SELECT id, role, version, status, config_json, created_at, activated_at, retired_at, proposed_by + FROM role_configs WHERE role = ? AND status = 'active' LIMIT 1`, roleName) + return scanRoleConfigRow(row) +} + +// ListRoleConfigVersions returns all versions for role, oldest first. +func (s *DB) ListRoleConfigVersions(roleName string) ([]*RoleConfigRow, error) { + rows, err := s.db.Query(` + SELECT id, role, version, status, config_json, created_at, activated_at, retired_at, proposed_by + FROM role_configs WHERE role = ? ORDER BY version ASC`, roleName) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []*RoleConfigRow + for rows.Next() { + r, err := scanRoleConfigRow(rows) + if err != nil { + return nil, err + } + out = append(out, r) + } + return out, rows.Err() +} + +// ActivateRoleConfigVersion promotes the given version of role to active, +// atomically retiring whatever version is currently active for that role (if +// any) in the same transaction. This enforces "at most one active row per +// role" without a DB-level constraint, the same way UpdateTaskState enforces +// the task state machine in a transaction rather than in schema. +func (s *DB) ActivateRoleConfigVersion(roleName string, version int) error { + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck + + var exists int + if err := tx.QueryRow(`SELECT COUNT(*) FROM role_configs WHERE role = ? AND version = ?`, roleName, version).Scan(&exists); err != nil { + return err + } + if exists == 0 { + return fmt.Errorf("role %q version %d not found", roleName, version) + } + + now := time.Now().UTC() + if _, err := tx.Exec(`UPDATE role_configs SET status = 'retired', retired_at = ? WHERE role = ? AND status = 'active'`, now, roleName); err != nil { + return err + } + if _, err := tx.Exec(`UPDATE role_configs SET status = 'active', activated_at = ? WHERE role = ? AND version = ?`, now, roleName, version); err != nil { + return err + } + return tx.Commit() +} + +func scanRoleConfigRow(row scanner) (*RoleConfigRow, error) { + var r RoleConfigRow + var createdAt time.Time + var activatedAt, retiredAt sql.NullTime + var proposedBy sql.NullString + if err := row.Scan(&r.ID, &r.Role, &r.Version, &r.Status, &r.ConfigJSON, &createdAt, &activatedAt, &retiredAt, &proposedBy); err != nil { + return nil, err + } + r.CreatedAt = createdAt + if activatedAt.Valid { + t := activatedAt.Time + r.ActivatedAt = &t + } + if retiredAt.Valid { + t := retiredAt.Time + r.RetiredAt = &t + } + r.ProposedBy = proposedBy.String + return &r, nil +} diff --git a/internal/storage/roleconfig_test.go b/internal/storage/roleconfig_test.go new file mode 100644 index 0000000..b18eaef --- /dev/null +++ b/internal/storage/roleconfig_test.go @@ -0,0 +1,151 @@ +package storage + +import ( + "database/sql" + "errors" + "testing" +) + +func TestCreateRoleConfig_AssignsIncrementingVersions(t *testing.T) { + db := testDB(t) + + v1, err := db.CreateRoleConfig("coder", `{"role":"coder"}`, "human") + if err != nil { + t.Fatalf("CreateRoleConfig v1: %v", err) + } + if v1.Version != 1 { + t.Errorf("expected version 1, got %d", v1.Version) + } + if v1.Status != "draft" { + t.Errorf("expected status draft, got %q", v1.Status) + } + + v2, err := db.CreateRoleConfig("coder", `{"role":"coder","system_prompt":"v2"}`, "human") + if err != nil { + t.Fatalf("CreateRoleConfig v2: %v", err) + } + if v2.Version != 2 { + t.Errorf("expected version 2, got %d", v2.Version) + } + + // A different role starts its own version sequence at 1. + other, err := db.CreateRoleConfig("reviewer", `{"role":"reviewer"}`, "human") + if err != nil { + t.Fatalf("CreateRoleConfig other role: %v", err) + } + if other.Version != 1 { + t.Errorf("expected version 1 for a new role, got %d", other.Version) + } +} + +func TestGetActiveRoleConfig_NoneActive_ReturnsErrNoRows(t *testing.T) { + db := testDB(t) + if _, err := db.CreateRoleConfig("coder", `{"role":"coder"}`, "human"); err != nil { + t.Fatalf("CreateRoleConfig: %v", err) + } + _, err := db.GetActiveRoleConfig("coder") + if !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("expected sql.ErrNoRows, got %v", err) + } +} + +func TestListRoleConfigVersions_OrderedByVersion(t *testing.T) { + db := testDB(t) + for i := 0; i < 3; i++ { + if _, err := db.CreateRoleConfig("coder", `{"role":"coder"}`, "human"); err != nil { + t.Fatalf("CreateRoleConfig: %v", err) + } + } + versions, err := db.ListRoleConfigVersions("coder") + if err != nil { + t.Fatalf("ListRoleConfigVersions: %v", err) + } + if len(versions) != 3 { + t.Fatalf("expected 3 versions, got %d", len(versions)) + } + for i, v := range versions { + if v.Version != i+1 { + t.Errorf("versions[%d].Version = %d, want %d", i, v.Version, i+1) + } + } +} + +// TestActivateRoleConfigVersion_ExactlyOneActive is the invariant test called +// out in the phase spec: activating version 2 while version 1 is active must +// atomically flip version 1 to retired. +func TestActivateRoleConfigVersion_ExactlyOneActive(t *testing.T) { + db := testDB(t) + + v1, err := db.CreateRoleConfig("coder", `{"role":"coder","system_prompt":"v1"}`, "human") + if err != nil { + t.Fatalf("CreateRoleConfig v1: %v", err) + } + v2, err := db.CreateRoleConfig("coder", `{"role":"coder","system_prompt":"v2"}`, "human") + if err != nil { + t.Fatalf("CreateRoleConfig v2: %v", err) + } + + if err := db.ActivateRoleConfigVersion("coder", v1.Version); err != nil { + t.Fatalf("activate v1: %v", err) + } + active, err := db.GetActiveRoleConfig("coder") + if err != nil { + t.Fatalf("GetActiveRoleConfig: %v", err) + } + if active.Version != v1.Version { + t.Fatalf("expected active version %d, got %d", v1.Version, active.Version) + } + if active.ActivatedAt == nil { + t.Errorf("expected ActivatedAt to be set") + } + + // Now activate v2 — v1 must atomically flip to retired, and v2 becomes + // the sole active row. + if err := db.ActivateRoleConfigVersion("coder", v2.Version); err != nil { + t.Fatalf("activate v2: %v", err) + } + + active, err = db.GetActiveRoleConfig("coder") + if err != nil { + t.Fatalf("GetActiveRoleConfig after v2 activation: %v", err) + } + if active.Version != v2.Version { + t.Fatalf("expected active version %d, got %d", v2.Version, active.Version) + } + + all, err := db.ListRoleConfigVersions("coder") + if err != nil { + t.Fatalf("ListRoleConfigVersions: %v", err) + } + activeCount := 0 + for _, row := range all { + if row.Status == "active" { + activeCount++ + } + if row.Version == v1.Version { + if row.Status != "retired" { + t.Errorf("expected v1 to be retired, got %q", row.Status) + } + if row.RetiredAt == nil { + t.Errorf("expected v1 RetiredAt to be set") + } + } + } + if activeCount != 1 { + t.Fatalf("expected exactly 1 active row, got %d", activeCount) + } +} + +func TestActivateRoleConfigVersion_UnknownVersion_Errors(t *testing.T) { + db := testDB(t) + if _, err := db.CreateRoleConfig("coder", `{"role":"coder"}`, "human"); err != nil { + t.Fatalf("CreateRoleConfig: %v", err) + } + if err := db.ActivateRoleConfigVersion("coder", 99); err == nil { + t.Fatalf("expected error activating unknown version") + } + // No row should have been left active. + if _, err := db.GetActiveRoleConfig("coder"); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("expected sql.ErrNoRows, got %v", err) + } +} diff --git a/internal/task/task.go b/internal/task/task.go index 7092ce8..811379e 100644 --- a/internal/task/task.go +++ b/internal/task/task.go @@ -30,6 +30,14 @@ const ( type AgentConfig struct { Type string `yaml:"type" json:"type"` Model string `yaml:"model" json:"model"` + // Role, when non-empty, names an internal/role.RoleConfig to dispatch + // through: internal/executor.Pool.execute() resolves tier 0 of the + // active role_configs row's EscalationLadder for the initial dispatch + // (setting Type/Model from the resolved rung), and internal/scheduler + // walks the ladder on failure (retry-then-escalate). Tasks with + // Role == "" (every existing YAML/chatbot task shape) are completely + // unaffected — this is purely additive. + Role string `yaml:"role,omitempty" json:"role,omitempty"` ContextFiles []string `yaml:"context_files" json:"context_files"` Instructions string `yaml:"instructions" json:"instructions"` ContainerImage string `yaml:"container_image" json:"container_image"` |