diff --git a/README.md b/README.md index 4698cd63..78cc827a 100644 --- a/README.md +++ b/README.md @@ -688,7 +688,9 @@ A **plugin** is a self-contained directory under `~/.config/task/plugins/` with - **workflows** (`workflows/*.yaml`) — new `ty pipeline -d ` definitions - **hooks** — scripts that fire on task events. Unlike the one-script-per-event hooks - dir above, any number of plugins can handle the same event and **all of them run** + dir above, any number of plugins can handle the same event and **all of them run**. + One of them, `task.route`, fires *before* a task spawns and lets the plugin pick + which Claude account it runs under — see [Routing](docs/plugins.md#routing-pre-spawn) - **actions** — user-invoked commands (`ty plugins run `) Install one — or a whole collection, since a single git repo can hold many plugins — diff --git a/docs/plugin-ideas.md b/docs/plugin-ideas.md index caa138a3..d8cc259f 100644 --- a/docs/plugin-ideas.md +++ b/docs/plugin-ideas.md @@ -7,6 +7,8 @@ demand with the task's env). A plugin is just a directory with executables; it can be any language and can bundle its own config/binaries. ✅ = shipped as an example in [`examples/plugins/`](../examples/plugins/). +📦 = shipped in the [community collection](https://github.com/taskyou/plugins) +(`ty plugins add https://github.com/taskyou/plugins`). ## Notifications & awareness (hooks) @@ -28,6 +30,20 @@ can be any language and can bundle its own config/binaries. - **status-file** — maintain a tiny JSON of live counts for a tmux statusline or menubar widget. +## Routing (the `task.route` hook) + +Fires before a task spawns and its stdout is read back as a decision — the one +hook that changes how a task runs rather than reporting on it. See +[Routing](plugins.md#routing-pre-spawn). + +- 📦 **claude-profile-router** — send each task to whichever Claude account has + the most rate-limit headroom; hold the task when both are spent. Ships in the + [community collection](https://github.com/taskyou/plugins). +- **quiet-hours** — `HOLD=1` outside working hours, so overnight queueing doesn't + spend your weekly limit while you sleep. +- **cheap-account-first** — route routine task types (docs, chores) to a Pro + account and keep the Max one for the heavy work. + ## Worktree & quality (actions) - ✅ **worktree** — show the task's diff; run its tests. @@ -51,8 +67,8 @@ can be any language and can bundle its own config/binaries. ## Where should plugins live? (in-repo vs. own repo) - **In-repo `examples/plugins/`** — small, canonical, copy-paste starting points - that ship with TaskYou and are covered by the loader's tests. The three above - live here. Best for anything short enough to read in one sitting. + that ship with TaskYou and are covered by the loader's tests. The ✅ entries + above live here. Best for anything short enough to read in one sitting. - **Its own repo** — when a plugin grows an independent release cadence, ships a compiled binary or heavier dependencies, or has a real surface of its own (config, docs, versioning). Install by dropping (or symlinking) its directory diff --git a/docs/plugins.md b/docs/plugins.md index 2660502f..7b9756c4 100644 --- a/docs/plugins.md +++ b/docs/plugins.md @@ -146,10 +146,75 @@ system. The ones dispatched today: | `task.blocked` | Task needs input | | `task.failed` | Agent execution failed | | `task.auth_required` | Executor session needs re-authentication | +| `task.route` | **Before** a task spawns — the one event you *answer*. See [Routing](#routing-pre-spawn) | A plugin may declare any event string; it only runs for events TaskYou actually emits, so unknown events are harmless. +## Routing (pre-spawn) + +Every hook above is a notification: it fires after the fact, runs detached, and +nothing waits for it. `task.route` is the exception. It fires *before* a task is +spawned, TaskYou waits for it, and it reads your script's **stdout back as a +decision** — which is the only way a plugin can influence how a task runs rather +than just react to it having run. + +The decision format is KEY=VALUE, one per line: + +```sh +#!/bin/sh +echo "CLAUDE_CONFIG_DIR=$HOME/.claude-work" # run this task under that Claude profile +echo "REASON=7% of its limits used" # optional note for the task log +``` + +| Key | Effect | +|-----|--------| +| `CLAUDE_CONFIG_DIR` | Run the task under that Claude profile (config dir) | +| `HOLD=1` | Don't start this task yet; leave it queued and reconsider next tick | +| `REASON=…` | Free text, written to the task log alongside the decision | + +Everything else on stdout is ignored, so unknown keys and stray output are +harmless — but keep diagnostics on **stderr** (which goes to the daemon log), +since stdout is the decision channel. + +**Guarantees.** Printing nothing is always safe, and so is failing: + +- **Silence means carry on.** No router installed, a script that errors, exceeds + the 15s timeout, or prints nothing usable — the task spawns exactly as it would + have. Routing is an optimization; failing to optimize never blocks work. +- **An explicit choice wins.** A task that already names a config dir (set by + hand, or by a workflow step's `config_dir:`) is left alone. +- **A routed task stays put.** The decision is written to the task, so a task + resumed later runs under the same profile it started on. This is required, not + merely tidy: a Claude session lives inside one config dir, and resuming under a + different one would find no session and quietly start a fresh conversation. The + cost is that a long-lived task pinned to a profile waits for *that* profile's + limits to reset — it cannot be migrated mid-conversation. +- **First answer stands.** Plugins are consulted in name order and the first + non-empty decision is used, so two installed routers give a deterministic + result. +- **`HOLD` keeps a task queued**, never blocked — it starts by itself once a + later tick gets a different answer. A hold is honored for queued tasks; a task + you started by hand (`ty run`, "start now") runs regardless. +- **Claude only.** `CLAUDE_CONFIG_DIR` means nothing to the codex or gemini + executors, so tasks using them are not routed. + +Extra environment on a routing hook, beyond the standard `TASK_*` and +`TASK_PLUGIN_*` variables: `TASK_EXECUTOR` and `TASK_CLAUDE_CONFIG_DIR` (the +task's current config dir, empty when unset). + +The worked example is **claude-profile-router** in the +[community collection](https://github.com/taskyou/plugins), which routes each task +to whichever of your Claude accounts has the most rate-limit headroom and holds a +task when every account is spent: + +```bash +ty plugins add https://github.com/taskyou/plugins +``` + +It reads each account's remaining limits itself — ty supplies the hook, the +plugin supplies the policy and the data it routes on. + ## Environment Every hook receives the standard task variables: diff --git a/internal/db/tasks.go b/internal/db/tasks.go index 74c23c0e..a312e9a4 100644 --- a/internal/db/tasks.go +++ b/internal/db/tasks.go @@ -937,6 +937,23 @@ func (db *DB) UpdateTaskPermissionMode(taskID int64, mode string) error { return nil } +// UpdateTaskClaudeConfigDir sets the per-task CLAUDE_CONFIG_DIR override, +// which is how a task is pinned to one Claude profile (account). Writing it as +// its own column update — rather than through UpdateTask — matters at spawn +// time: the routing decision is made from a task struct the daemon has been +// holding, and a full-row write would stomp any field another surface (the TUI, +// a hook) changed in the meantime. +func (db *DB) UpdateTaskClaudeConfigDir(taskID int64, configDir string) error { + _, err := db.Exec(` + UPDATE tasks SET claude_config_dir = ?, updated_at = CURRENT_TIMESTAMP + WHERE id = ? + `, configDir, taskID) + if err != nil { + return fmt.Errorf("update task claude config dir: %w", err) + } + return nil +} + // UpdateTaskPinned updates only the pinned flag for a task. func (db *DB) UpdateTaskPinned(taskID int64, pinned bool) error { _, err := db.Exec(` diff --git a/internal/executor/executor.go b/internal/executor/executor.go index cad6d188..484fa5fd 100644 --- a/internal/executor/executor.go +++ b/internal/executor/executor.go @@ -1788,10 +1788,23 @@ func (e *Executor) processNextTask(ctx context.Context) { // A step deferred for branch contention serves its backoff here. Without // this gate the task is re-entered on every 2s tick, and each pass writes // a fresh "Starting task #N" line for a step that cannot start. + // + // This gate goes before routing deliberately: it is a map lookup, while + // routing may shell out to a plugin. A task sitting out a branch backoff + // shouldn't pay for a usage probe on every tick to learn it still can't run. if !e.branchWaitDue(task.ID) { continue } + // Last decision before the spawn: which Claude profile does this run + // under? A routing plugin may pick one (stamping task.ClaudeConfigDir, + // which both command builders already honor) or ask to hold the task + // when every account is out of headroom. With no router installed this + // is a no-op. See routing.go. + if !e.routeTask(ctx, task, true) { + continue + } + // Atomically check-and-set to prevent race where two ticks // both see the task as not-running and spawn duplicate goroutines e.mu.Lock() @@ -1855,6 +1868,11 @@ func (e *Executor) ExecuteNow(ctx context.Context, taskID int64) error { e.runningTasks[taskID] = true e.mu.Unlock() + // Route this run to a Claude profile too, so a manually started task lands + // on the same account the queue would have chosen. A hold is not honored + // here: the user asked for this task to run now. + e.routeTask(ctx, task, false) + e.executeTask(ctx, task) return nil } diff --git a/internal/executor/routing.go b/internal/executor/routing.go new file mode 100644 index 00000000..33947d0e --- /dev/null +++ b/internal/executor/routing.go @@ -0,0 +1,113 @@ +package executor + +import ( + "context" + "fmt" + "strings" + "sync" + + "github.com/bborn/workflow/internal/db" + "github.com/bborn/workflow/internal/hooks" +) + +// Profile routing gives a plugin the last word on which Claude account a task +// runs under, at the only moment where that word is still worth anything: after +// the task is cleared to run, before its command is built. +// +// Everything downstream already supports this — Task.ClaudeConfigDir has always +// been the per-task profile lever, honored identically by the daemon's command +// builder and the TUI's. What was missing was anyone to set it automatically. +// Routing fills that in: it stamps the column and lets the existing machinery +// carry the decision the rest of the way, so there is no second code path for a +// routed task and no chance of the two builders disagreeing about which profile +// is in play. +// +// Two rules keep it from getting in the way: +// +// - An explicit choice always wins. A task that already names a config dir — +// set by hand, by a workflow step, or by an earlier routing pass — is left +// alone. Routing fills a vacuum; it does not overrule a person. +// - Silence means "carry on". No router installed, a script that fails, times +// out, or prints nothing: the task spawns exactly as it would have before +// any of this existed. + +// routeHoldLog remembers the last hold reason logged per task, so a task parked +// behind exhausted profiles writes one log line rather than one per daemon tick. +var routeHoldLog sync.Map // taskID -> last reason written + +// routeTask consults the task.route plugin hook and applies its decision. +// +// It returns false only when a router asked to hold the task — every other +// outcome, including every kind of failure, returns true and lets the spawn +// proceed. allowHold is false on the manual "run this now" path: a person who +// explicitly started a task has already made the call, and silently refusing +// would look like the button was broken. +func (e *Executor) routeTask(ctx context.Context, task *db.Task, allowHold bool) bool { + if task == nil || e.hooks == nil { + return true + } + // CLAUDE_CONFIG_DIR is a Claude concept; a codex or gemini task has no + // profile to route between. + if task.Executor != "" && task.Executor != db.ExecutorClaude { + return true + } + if strings.TrimSpace(task.ClaudeConfigDir) != "" { + return true + } + if !e.hooks.HandlesRoute() { + return true + } + + decision := e.hooks.Route(ctx, task) + if decision.Empty() { + routeHoldLog.Delete(task.ID) + return true + } + + if decision.Hold && allowHold { + e.noteRouteHold(task, decision) + return false + } + routeHoldLog.Delete(task.ID) + + dir := strings.TrimSpace(decision.ClaudeConfigDir) + if dir == "" { + return true + } + resolved := ResolveClaudeConfigDir(dir) + if err := e.db.UpdateTaskClaudeConfigDir(task.ID, resolved); err != nil { + // The write is what makes the decision visible to the TUI and to a + // later resume. If it fails, don't apply the route in memory either — + // a task whose spawned profile disagrees with its recorded one is the + // exact confusion this feature is supposed to remove. + e.logger.Error("Failed to record routed Claude profile", "id", task.ID, "dir", resolved, "error", err) + return true + } + task.ClaudeConfigDir = resolved + + msg := fmt.Sprintf("Routed to Claude profile %s (by plugin %q)", resolved, decision.Plugin) + if decision.Reason != "" { + msg += ": " + decision.Reason + } + e.logger.Info("Routed task to Claude profile", "id", task.ID, "dir", resolved, "plugin", decision.Plugin) + e.logLine(task.ID, "system", msg) + return true +} + +// noteRouteHold records a hold, writing to the task log only when the reason +// changes. The daemon reconsiders a queued task every tick, so an unconditional +// log line would bury the task's real history under thousands of repeats of +// "waiting for headroom". +func (e *Executor) noteRouteHold(task *db.Task, decision hooks.RouteDecision) { + reason := decision.Reason + if reason == "" { + reason = "no Claude profile has headroom right now" + } + e.logger.Info("Holding task: no Claude profile available", "id", task.ID, "plugin", decision.Plugin, "reason", reason) + + if prev, ok := routeHoldLog.Load(task.ID); ok && prev == reason { + return + } + routeHoldLog.Store(task.ID, reason) + e.logLine(task.ID, "system", fmt.Sprintf("Waiting to start — %s (plugin %q). Will retry automatically.", reason, decision.Plugin)) +} diff --git a/internal/executor/routing_test.go b/internal/executor/routing_test.go new file mode 100644 index 00000000..eb0ab290 --- /dev/null +++ b/internal/executor/routing_test.go @@ -0,0 +1,277 @@ +package executor + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/bborn/workflow/internal/config" + "github.com/bborn/workflow/internal/db" +) + +// newRoutingExecutor builds an Executor whose plugins come from a temp dir, so a +// routing test exercises the real hook path (subprocess, stdout parsing, DB +// write) without depending on what is installed on the machine. +func newRoutingExecutor(t *testing.T, routeScript string) (*Executor, *db.DB) { + t.Helper() + + pluginsDir := t.TempDir() + if routeScript != "" { + dir := filepath.Join(pluginsDir, "router") + if err := os.MkdirAll(dir, 0o755); err != nil { + t.Fatal(err) + } + manifest := "name: router\nhooks:\n task.route: route.sh\n" + if err := os.WriteFile(filepath.Join(dir, "plugin.yaml"), []byte(manifest), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "route.sh"), []byte(routeScript), 0o755); err != nil { + t.Fatal(err) + } + } + t.Setenv("TY_PLUGINS_DIR", pluginsDir) + + tmpFile, err := os.CreateTemp("", "test-routing-*.db") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.Remove(tmpFile.Name()) }) + tmpFile.Close() + + database, err := db.Open(tmpFile.Name()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { database.Close() }) + if err := database.CreateProject(&db.Project{Name: "test", Path: "/tmp/test"}); err != nil { + t.Fatal(err) + } + + // The hold-log memo is package state; keep tests from leaking into each other. + t.Cleanup(func() { routeHoldLog.Range(func(k, _ any) bool { routeHoldLog.Delete(k); return true }) }) + + return New(database, &config.Config{}), database +} + +func newRoutingTask(t *testing.T, database *db.DB, executorName string) *db.Task { + t.Helper() + task := &db.Task{Title: "route me", Type: "task", Project: "test", Executor: executorName} + if err := database.CreateTask(task); err != nil { + t.Fatal(err) + } + return task +} + +func TestRouteTask_AppliesAndPersistsConfigDir(t *testing.T) { + e, database := newRoutingExecutor(t, "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/tmp/claude-work\necho REASON=12% used\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false, want the task cleared to run") + } + if task.ClaudeConfigDir != "/tmp/claude-work" { + t.Errorf("in-memory ClaudeConfigDir = %q", task.ClaudeConfigDir) + } + + // The write matters as much as the in-memory value: the TUI and any later + // resume read the column, and a disagreement there is exactly the confusion + // routing is meant to remove. + reloaded, err := database.GetTask(task.ID) + if err != nil { + t.Fatal(err) + } + if reloaded.ClaudeConfigDir != "/tmp/claude-work" { + t.Errorf("persisted ClaudeConfigDir = %q", reloaded.ClaudeConfigDir) + } + + logs, err := database.GetTaskLogs(task.ID, 10) + if err != nil { + t.Fatal(err) + } + found := false + for _, l := range logs { + if strings.Contains(l.Content, "/tmp/claude-work") && strings.Contains(l.Content, "12% used") { + found = true + } + } + if !found { + t.Errorf("routing decision was not written to the task log: %+v", logs) + } +} + +func TestRouteTask_ExplicitConfigDirIsNotOverridden(t *testing.T) { + // A dir chosen by a person or a workflow step is a decision, not a default. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/tmp/router-choice\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + task.ClaudeConfigDir = "/tmp/chosen-by-hand" + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false") + } + if task.ClaudeConfigDir != "/tmp/chosen-by-hand" { + t.Errorf("router overrode an explicit config dir: %q", task.ClaudeConfigDir) + } +} + +func TestRouteTask_ResumedTaskStaysOnItsProfile(t *testing.T) { + // Session affinity, and it is not optional: a Claude session lives inside + // one config dir, so a task resumed under a different profile would find no + // session to resume and silently start a fresh conversation. Once a task has + // been routed, the stamped dir must pin it for the rest of its life — even + // when the router would now prefer somewhere else. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/tmp/now-emptier\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + + // First spawn: the router picks a profile and it is recorded. + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false") + } + first := task.ClaudeConfigDir + if first == "" { + t.Fatal("first spawn was not routed") + } + task.ClaudeSessionID = "sess-abc" + + // Second pass (a resume after the task was blocked, say) must not move it. + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false on resume") + } + if task.ClaudeConfigDir != first { + t.Errorf("resumed task moved profiles: %q -> %q", first, task.ClaudeConfigDir) + } +} + +func TestRouteTask_NonClaudeExecutorIsUntouched(t *testing.T) { + // CLAUDE_CONFIG_DIR means nothing to codex; setting it would be noise at + // best and a misleading task log at worst. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/tmp/nope\n") + task := newRoutingTask(t, database, db.ExecutorCodex) + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false") + } + if task.ClaudeConfigDir != "" { + t.Errorf("codex task was routed: %q", task.ClaudeConfigDir) + } +} + +func TestRouteTask_HoldKeepsTaskQueued(t *testing.T) { + e, database := newRoutingExecutor(t, "#!/bin/sh\necho HOLD=1\necho 'REASON=every profile above 90%'\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + if err := database.UpdateTaskStatus(task.ID, db.StatusQueued); err != nil { + t.Fatal(err) + } + + if ok := e.routeTask(context.Background(), task, true); ok { + t.Fatal("routeTask returned true, want the spawn held") + } + + // A held task must stay queued: parking it as blocked would take a human to + // undo, when the whole point is that it starts by itself once limits reset. + reloaded, err := database.GetTask(task.ID) + if err != nil { + t.Fatal(err) + } + if reloaded.Status != db.StatusQueued { + t.Errorf("status = %q, want it left queued", reloaded.Status) + } +} + +func TestRouteTask_RepeatedHoldLogsOnce(t *testing.T) { + // The daemon reconsiders a queued task every tick. Logging each refusal + // would bury the task's real history under thousands of identical lines. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho HOLD=1\necho 'REASON=every profile above 90%'\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + + for i := 0; i < 3; i++ { + if ok := e.routeTask(context.Background(), task, true); ok { + t.Fatal("routeTask returned true, want held") + } + } + + logs, err := database.GetTaskLogs(task.ID, 50) + if err != nil { + t.Fatal(err) + } + holds := 0 + for _, l := range logs { + if strings.Contains(l.Content, "Waiting to start") { + holds++ + } + } + if holds != 1 { + t.Errorf("wrote %d hold log lines across 3 ticks, want 1", holds) + } +} + +func TestRouteTask_HoldIgnoredOnManualRun(t *testing.T) { + // `ty run` / "start now" is an explicit instruction. Silently refusing it + // would read as a broken button. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho HOLD=1\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + + if ok := e.routeTask(context.Background(), task, false); !ok { + t.Error("a manual run should not be held") + } +} + +func TestRouteTask_NoRouterIsANoOp(t *testing.T) { + e, database := newRoutingExecutor(t, "") + task := newRoutingTask(t, database, db.ExecutorClaude) + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false with no router installed") + } + if task.ClaudeConfigDir != "" { + t.Errorf("ClaudeConfigDir = %q, want untouched", task.ClaudeConfigDir) + } +} + +func TestRouteTask_FailingRouterStillSpawns(t *testing.T) { + // Routing is an optimization. Failing to optimize must never be why a task + // doesn't run. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho boom >&2\nexit 1\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Error("a failing router should not block the spawn") + } + if task.ClaudeConfigDir != "" { + t.Errorf("ClaudeConfigDir = %q, want untouched", task.ClaudeConfigDir) + } +} + +func TestRouteTask_ExpandsTildeInRoutedDir(t *testing.T) { + // A router written in shell may well emit a literal ~; the stored value has + // to be the resolved path, since it is spliced straight into the spawn + // command as CLAUDE_CONFIG_DIR="…". + e, database := newRoutingExecutor(t, "#!/bin/sh\necho 'CLAUDE_CONFIG_DIR=~/.claude-work'\n") + task := newRoutingTask(t, database, db.ExecutorClaude) + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false") + } + home, err := os.UserHomeDir() + if err != nil { + t.Skip("no home dir") + } + want := filepath.Join(home, ".claude-work") + if task.ClaudeConfigDir != want { + t.Errorf("ClaudeConfigDir = %q, want %q", task.ClaudeConfigDir, want) + } +} + +func TestRouteTask_EmptyExecutorIsTreatedAsClaude(t *testing.T) { + // Older tasks carry no executor; claude is the default, so they should route. + e, database := newRoutingExecutor(t, "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/tmp/x\n") + task := newRoutingTask(t, database, "") + + if ok := e.routeTask(context.Background(), task, true); !ok { + t.Fatal("routeTask returned false") + } + if task.ClaudeConfigDir != "/tmp/x" { + t.Errorf("ClaudeConfigDir = %q, want /tmp/x", task.ClaudeConfigDir) + } +} diff --git a/internal/hooks/route.go b/internal/hooks/route.go new file mode 100644 index 00000000..2f40661b --- /dev/null +++ b/internal/hooks/route.go @@ -0,0 +1,184 @@ +package hooks + +import ( + "bufio" + "bytes" + "context" + "fmt" + "os/exec" + "strings" + "time" + + "github.com/bborn/workflow/internal/db" +) + +// EventTaskRoute is fired immediately before a task is spawned, and is the one +// event a plugin can *answer* rather than merely observe. +// +// Every other hook is a notification: it fires after the fact, runs detached, +// and nothing waits for it. A router has to be the opposite — the decision it +// makes (which Claude profile this task runs under) is only useful before the +// command is built, so this hook runs synchronously, in the foreground of the +// spawn, and its stdout is read back. +// +// That inversion is deliberate, and bounded: RouteTimeout caps the wait, a +// failing or silent script yields no decision and the task spawns exactly as it +// would have, and the first plugin to answer wins so a slow one can't be made to +// re-litigate a settled choice. +const EventTaskRoute = "task.route" + +// RouteTimeout bounds a routing hook. A task spawn blocks on this, so it is +// tight: a router that needs longer than this to pick a profile is a router that +// should be caching, and the safe answer while it does is "spawn as configured". +const RouteTimeout = 15 * time.Second + +// RouteDecision is what a routing hook answers with. The zero value means "no +// opinion" — the caller proceeds with the task's existing configuration. +type RouteDecision struct { + // Plugin is the name of the plugin that answered, for logging. + Plugin string + // ClaudeConfigDir routes the task to a particular Claude profile. Empty + // leaves the task's existing (project or per-task) config dir alone. + ClaudeConfigDir string + // Hold asks the caller not to start this task yet — every candidate profile + // is out of headroom, and running now would only burn a session on a 429. + // The task stays queued and is reconsidered on the next tick. + Hold bool + // Reason explains a Hold (or annotates a routing choice) for the task log. + Reason string +} + +// Empty reports whether the decision carries no instruction at all. +func (d RouteDecision) Empty() bool { + return !d.Hold && strings.TrimSpace(d.ClaudeConfigDir) == "" +} + +// HandlesRoute reports whether any loaded plugin declares a task.route hook. +// Spawn checks this first so the overwhelmingly common case — nobody has +// installed a router — costs a slice scan instead of a subprocess. +func (r *Runner) HandlesRoute() bool { + for _, p := range r.plugins { + if _, ok := p.ScriptFor(EventTaskRoute); ok { + return true + } + } + return false +} + +// Route asks every plugin that handles task.route what to do with this task, +// in plugin-name order, and returns the first non-empty decision. +// +// Plugins are consulted in order rather than in parallel and the first answer +// stands, which keeps the outcome deterministic when more than one router is +// installed — the alternative (merging or last-write-wins) makes the effective +// policy depend on which script happened to finish first. +// +// A hook that errors, times out, or prints nothing usable is skipped: routing +// is an optimization, and failing to optimize must never be the reason a task +// doesn't run. +func (r *Runner) Route(ctx context.Context, task *db.Task) RouteDecision { + if task == nil { + return RouteDecision{} + } + for _, p := range r.plugins { + script, ok := p.ScriptFor(EventTaskRoute) + if !ok { + continue + } + env := append(taskEnv(EventTaskRoute, task, ""), + "TASK_PLUGIN_NAME="+p.Name, + "TASK_PLUGIN_DIR="+p.Dir, + "TASK_EXECUTOR="+task.Executor, + "TASK_CLAUDE_CONFIG_DIR="+task.ClaudeConfigDir, + ) + + decision, err := runRouteScript(ctx, script, p.Dir, env) + if err != nil { + r.logger.Warn("route hook failed", "plugin", p.Name, "task", task.ID, "error", err) + continue + } + if decision.Empty() { + continue + } + decision.Plugin = p.Name + return decision + } + return RouteDecision{} +} + +// runRouteScript executes one routing script and parses its verdict. +func runRouteScript(ctx context.Context, script, workDir string, env []string) (RouteDecision, error) { + ctx, cancel := context.WithTimeout(ctx, RouteTimeout) + defer cancel() + + cmd := exec.CommandContext(ctx, script) + cmd.Dir = workDir + cmd.Env = env + + // stdout is the decision channel and stderr is free for the script to log + // on, so they are captured separately — otherwise an `echo "checking..." >&2` + // in a router would be parsed as part of its answer. + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + + if err := cmd.Run(); err != nil { + return RouteDecision{}, fmt.Errorf("%w: %s", err, strings.TrimSpace(stderr.String())) + } + return ParseRouteOutput(stdout.String()), nil +} + +// ParseRouteOutput reads a routing script's stdout. +// +// The format is deliberately the dullest thing that works — KEY=VALUE, one per +// line, unknown keys ignored — because the scripts writing it are shell. A +// router shouldn't need a JSON encoder to say "use this directory". +// +// CLAUDE_CONFIG_DIR=/Users/me/.claude-work +// HOLD=1 +// REASON=both profiles above 90%, next reset 14:00 +// +// Values are taken literally after the first '='; surrounding quotes are +// stripped so `CLAUDE_CONFIG_DIR="$dir"` from a script that quoted its output +// still parses. +func ParseRouteOutput(out string) RouteDecision { + var d RouteDecision + scanner := bufio.NewScanner(strings.NewReader(out)) + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + key, value, found := strings.Cut(line, "=") + if !found { + continue + } + value = unquote(strings.TrimSpace(value)) + switch strings.ToUpper(strings.TrimSpace(key)) { + case "CLAUDE_CONFIG_DIR": + d.ClaudeConfigDir = value + case "HOLD", "DEFER": + d.Hold = isTruthy(value) + case "REASON": + d.Reason = value + } + } + return d +} + +func unquote(s string) string { + if len(s) >= 2 { + if (s[0] == '"' && s[len(s)-1] == '"') || (s[0] == '\'' && s[len(s)-1] == '\'') { + return s[1 : len(s)-1] + } + } + return s +} + +func isTruthy(s string) bool { + switch strings.ToLower(strings.TrimSpace(s)) { + case "1", "true", "yes", "y", "on": + return true + } + return false +} diff --git a/internal/hooks/route_test.go b/internal/hooks/route_test.go new file mode 100644 index 00000000..11e64385 --- /dev/null +++ b/internal/hooks/route_test.go @@ -0,0 +1,205 @@ +package hooks + +import ( + "context" + "os" + "testing" + + "github.com/charmbracelet/log" + + "github.com/bborn/workflow/internal/db" +) + +func routeRunner(t *testing.T, root string) *Runner { + t.Helper() + return newRunner("", root, log.NewWithOptions(os.Stderr, log.Options{Level: log.FatalLevel})) +} + +func TestParseRouteOutput(t *testing.T) { + tests := []struct { + name string + out string + want RouteDecision + }{ + { + name: "config dir", + out: "CLAUDE_CONFIG_DIR=/home/me/.claude-work\n", + want: RouteDecision{ClaudeConfigDir: "/home/me/.claude-work"}, + }, + { + name: "quoted value", + out: "CLAUDE_CONFIG_DIR=\"/home/me/my claude\"\n", + want: RouteDecision{ClaudeConfigDir: "/home/me/my claude"}, + }, + { + name: "hold with reason", + out: "HOLD=1\nREASON=all profiles above 90%\n", + want: RouteDecision{Hold: true, Reason: "all profiles above 90%"}, + }, + { + name: "defer is an alias for hold", + out: "DEFER=true\n", + want: RouteDecision{Hold: true}, + }, + { + name: "hold=0 is not a hold", + out: "HOLD=0\nCLAUDE_CONFIG_DIR=/a\n", + want: RouteDecision{ClaudeConfigDir: "/a"}, + }, + { + // A router's stdout is decision-only, but scripts still leak the odd + // line. Anything unrecognized must be inert rather than fatal. + name: "noise, comments and blank lines are ignored", + out: "\n# picking a profile\nchecking usage...\nCLAUDE_CONFIG_DIR=/a\nUNKNOWN_KEY=x\n", + want: RouteDecision{ClaudeConfigDir: "/a"}, + }, + { + name: "value containing = is kept whole", + out: "REASON=used=93%\nHOLD=yes\n", + want: RouteDecision{Hold: true, Reason: "used=93%"}, + }, + { + name: "empty output is no opinion", + out: "", + want: RouteDecision{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + got := ParseRouteOutput(tc.out) + if got != tc.want { + t.Errorf("ParseRouteOutput(%q) = %+v, want %+v", tc.out, got, tc.want) + } + }) + } +} + +func TestRouteDecisionEmpty(t *testing.T) { + if !(RouteDecision{}).Empty() { + t.Error("zero decision should be empty") + } + if (RouteDecision{ClaudeConfigDir: "/a"}).Empty() { + t.Error("decision with a config dir is not empty") + } + if (RouteDecision{Hold: true}).Empty() { + t.Error("hold decision is not empty") + } + // A reason on its own carries no instruction, so it must not count as an + // answer — otherwise a script that only logged would silently shadow the + // next router in line. + if !(RouteDecision{Reason: "just saying"}).Empty() { + t.Error("reason-only decision should be empty") + } +} + +func TestRoute_AppliesDecisionAndInjectsEnv(t *testing.T) { + root := t.TempDir() + writePlugin(t, root, "router", + "name: router\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho \"CLAUDE_CONFIG_DIR=/dirs/$TASK_PROJECT-$TASK_ID-$TASK_EXECUTOR\"\n"}) + + r := routeRunner(t, root) + if !r.HandlesRoute() { + t.Fatal("HandlesRoute() = false, want true") + } + + task := &db.Task{ID: 7, Title: "t", Project: "proj", Executor: db.ExecutorClaude} + got := r.Route(context.Background(), task) + if got.ClaudeConfigDir != "/dirs/proj-7-claude" { + t.Errorf("ClaudeConfigDir = %q", got.ClaudeConfigDir) + } + if got.Plugin != "router" { + t.Errorf("Plugin = %q, want router", got.Plugin) + } +} + +func TestRoute_FirstNonEmptyDecisionWinsInNameOrder(t *testing.T) { + root := t.TempDir() + // "a-quiet" sorts first but abstains; "b-router" must then be consulted. + writePlugin(t, root, "a-quiet", + "name: a-quiet\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\nexit 0\n"}) + writePlugin(t, root, "b-router", + "name: b-router\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/from-b\n"}) + writePlugin(t, root, "c-router", + "name: c-router\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/from-c\n"}) + + r := routeRunner(t, root) + got := r.Route(context.Background(), &db.Task{ID: 1, Executor: db.ExecutorClaude}) + if got.ClaudeConfigDir != "/from-b" { + t.Errorf("ClaudeConfigDir = %q, want /from-b (first answering plugin by name)", got.ClaudeConfigDir) + } +} + +func TestRoute_FailingScriptIsSkipped(t *testing.T) { + root := t.TempDir() + // A router that prints a decision *and* exits non-zero must not be trusted: + // a half-finished script's last echo is not a decision. + writePlugin(t, root, "a-broken", + "name: a-broken\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/bad\nexit 3\n"}) + writePlugin(t, root, "b-good", + "name: b-good\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/good\n"}) + + r := routeRunner(t, root) + got := r.Route(context.Background(), &db.Task{ID: 1, Executor: db.ExecutorClaude}) + if got.ClaudeConfigDir != "/good" { + t.Errorf("ClaudeConfigDir = %q, want /good", got.ClaudeConfigDir) + } +} + +func TestRoute_StderrIsNotParsedAsDecision(t *testing.T) { + root := t.TempDir() + writePlugin(t, root, "router", + "name: router\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho 'CLAUDE_CONFIG_DIR=/from-stderr' >&2\necho CLAUDE_CONFIG_DIR=/from-stdout\n"}) + + r := routeRunner(t, root) + got := r.Route(context.Background(), &db.Task{ID: 1, Executor: db.ExecutorClaude}) + if got.ClaudeConfigDir != "/from-stdout" { + t.Errorf("ClaudeConfigDir = %q, want /from-stdout", got.ClaudeConfigDir) + } +} + +func TestRoute_NoRoutePluginsIsNoOpinion(t *testing.T) { + root := t.TempDir() + writePlugin(t, root, "notifier", + "name: notifier\nhooks:\n task.done: done.sh\n", + map[string]string{"done.sh": "#!/bin/sh\n"}) + + r := routeRunner(t, root) + if r.HandlesRoute() { + t.Error("HandlesRoute() = true with no task.route hook") + } + if got := r.Route(context.Background(), &db.Task{ID: 1}); !got.Empty() { + t.Errorf("Route = %+v, want empty", got) + } +} + +func TestRoute_NilTaskIsSafe(t *testing.T) { + root := t.TempDir() + writePlugin(t, root, "router", + "name: router\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/x\n"}) + + if got := routeRunner(t, root).Route(context.Background(), nil); !got.Empty() { + t.Errorf("Route(nil) = %+v, want empty", got) + } +} + +func TestRoute_CancelledContextYieldsNoDecision(t *testing.T) { + root := t.TempDir() + writePlugin(t, root, "router", + "name: router\nhooks:\n task.route: route.sh\n", + map[string]string{"route.sh": "#!/bin/sh\necho CLAUDE_CONFIG_DIR=/x\n"}) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if got := routeRunner(t, root).Route(ctx, &db.Task{ID: 1}); !got.Empty() { + t.Errorf("Route with cancelled ctx = %+v, want empty (spawn as configured)", got) + } +}