diff --git a/cli/config.go b/cli/config.go index 89e2b7c..2d63a90 100644 --- a/cli/config.go +++ b/cli/config.go @@ -9,10 +9,11 @@ import ( ) type Config struct { - URL string `mapstructure:"url"` - APIKey string `mapstructure:"api_key"` - Realm string `mapstructure:"realm"` - Warnings []string `mapstructure:"-"` + URL string `mapstructure:"url"` + APIKey string `mapstructure:"api_key"` + Realm string `mapstructure:"realm"` + Orchestrate OrchestrateConfig `mapstructure:"orchestrate"` + Warnings []string `mapstructure:"-"` } func LoadConfig(workDir, homeDir string) (*Config, error) { diff --git a/cli/orchestrate.go b/cli/orchestrate.go new file mode 100644 index 0000000..7c579f0 --- /dev/null +++ b/cli/orchestrate.go @@ -0,0 +1,325 @@ +package cli + +import ( + "context" + "encoding/json" + "fmt" + "os" + "os/signal" + "os/user" + "sync" + "syscall" + "time" + + "github.com/spf13/cobra" +) + +// OrchestrateConfig holds the orchestrate section of .bifrost.yaml. +type OrchestrateConfig struct { + Dispatcher string `mapstructure:"dispatcher"` + PollInterval time.Duration `mapstructure:"poll_interval"` + Concurrency int `mapstructure:"concurrency"` + Claimant string `mapstructure:"claimant"` +} + +// OrchestrateCmd is the bf orchestrate command. +type OrchestrateCmd struct { + Command *cobra.Command +} + +func NewOrchestrateCmd(clientFn func() *Client, cfgFn func() *Config) *OrchestrateCmd { + c := &OrchestrateCmd{} + + cmd := &cobra.Command{ + Use: "orchestrate", + Short: "Poll for ready runes and dispatch them to configured agents", + RunE: func(cmd *cobra.Command, args []string) error { + cfg := cfgFn() + + // Resolve effective config: flags override yaml config. + oCfg := cfg.Orchestrate + + if v, _ := cmd.Flags().GetString("dispatcher"); v != "" { + oCfg.Dispatcher = v + } + if v, _ := cmd.Flags().GetDuration("poll-interval"); v != 0 { + oCfg.PollInterval = v + } + if v, _ := cmd.Flags().GetInt("concurrency"); v != 0 { + oCfg.Concurrency = v + } + if v, _ := cmd.Flags().GetString("claimant"); v != "" { + oCfg.Claimant = v + } + + // Apply defaults. + if oCfg.PollInterval == 0 { + oCfg.PollInterval = 10 * time.Second + } + if oCfg.Concurrency == 0 { + oCfg.Concurrency = 1 + } + if oCfg.Claimant == "" { + if u, err := user.Current(); err == nil { + oCfg.Claimant = u.Username + } + } + + // Validate configuration values. + if oCfg.Concurrency <= 0 { + return fmt.Errorf("concurrency must be positive, got %d", oCfg.Concurrency) + } + if oCfg.PollInterval <= 0 { + return fmt.Errorf("poll-interval must be positive, got %s", oCfg.PollInterval) + } + + if oCfg.Dispatcher == "" { + return fmt.Errorf("dispatcher is required: set orchestrate.dispatcher in .bifrost.yaml or use --dispatcher") + } + + // Validate dispatcher is accessible. + if _, err := os.Stat(oCfg.Dispatcher); err != nil { + return fmt.Errorf("dispatcher script not found: %s", oCfg.Dispatcher) + } + + saga, _ := cmd.Flags().GetString("saga") + dryRun, _ := cmd.Flags().GetBool("dry-run") + once, _ := cmd.Flags().GetBool("once") + unclaimOnFailure, _ := cmd.Flags().GetBool("unclaim-on-failure") + + dispatcher := &ScriptDispatcher{ScriptPath: oCfg.Dispatcher} + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + return runOrchestrator(ctx, clientFn(), oCfg, dispatcher, saga, dryRun, once, unclaimOnFailure) + }, + } + + cmd.Flags().String("dispatcher", "", "path to dispatcher script (overrides config)") + cmd.Flags().Duration("poll-interval", 0, "polling interval (default 10s)") + cmd.Flags().Int("concurrency", 0, "number of parallel workers (default 1)") + cmd.Flags().String("claimant", "", "claimant name (default: system username)") + cmd.Flags().Bool("unclaim-on-failure", false, "unclaim rune when dispatched command exits non-zero") + cmd.Flags().String("saga", "", "only orchestrate runes in this saga") + cmd.Flags().Bool("dry-run", false, "resolve dispatch but do not execute or fulfill") + cmd.Flags().Bool("once", false, "process one batch then exit") + + c.Command = cmd + return c +} + +func runOrchestrator( + ctx context.Context, + client *Client, + cfg OrchestrateConfig, + dispatcher Dispatcher, + saga string, + dryRun, once, unclaimOnFailure bool, +) error { + queue := make(chan map[string]any, cfg.Concurrency*2) + var inFlight sync.Map + + var wg sync.WaitGroup + for i := 0; i < cfg.Concurrency; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-ctx.Done(): + return + case rune, ok := <-queue: + if !ok { + return + } + processRune(ctx, client, cfg, dispatcher, rune, dryRun, unclaimOnFailure, &inFlight) + } + } + }() + } + + poll := func() { + runes, err := fetchReadyRunes(client, saga) + if err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: poll error: %v\n", err) + return + } + + for _, r := range runes { + id, _ := r["id"].(string) + if id == "" { + continue + } + // Skip if already claimed by someone else. + if claimant, _ := r["claimant"].(string); claimant != "" { + continue + } + // Skip if already in-flight. + if _, loaded := inFlight.LoadOrStore(id, struct{}{}); loaded { + continue + } + // Blocking send in --once mode to guarantee all items are enqueued. + // Non-blocking send otherwise — if queue is full, release from in-flight and skip. + if once { + queue <- r + } else { + select { + case queue <- r: + default: + inFlight.Delete(id) + } + } + } + } + + // Run first poll immediately. + poll() + + if once { + // Wait for queue to drain, then shut down workers. + // We close the queue after the first poll drains. + // We signal workers by closing the channel once all items are enqueued. + // But since queue is buffered and workers are async, we need to wait + // for in-flight items to complete. We do this by closing queue and + // waiting for wg. + close(queue) + wg.Wait() + return nil + } + + ticker := time.NewTicker(cfg.PollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + close(queue) + wg.Wait() + return nil + case <-ticker.C: + poll() + } + } +} + +func fetchReadyRunes(client *Client, saga string) ([]map[string]any, error) { + params := map[string]string{ + "status": "open", + "blocked": "false", + "is_saga": "false", + } + if saga != "" { + params["saga"] = saga + } + + body, err := client.DoGetWithParams("/runes", params) + if err != nil { + return nil, err + } + + var runes []map[string]any + if err := json.Unmarshal(body, &runes); err != nil { + return nil, fmt.Errorf("parsing runes response: %w", err) + } + return runes, nil +} + +func processRune( + ctx context.Context, + client *Client, + cfg OrchestrateConfig, + dispatcher Dispatcher, + summary map[string]any, + dryRun, unclaimOnFailure bool, + inFlight *sync.Map, +) { + id, _ := summary["id"].(string) + defer inFlight.Delete(id) + + // Fetch full rune detail. + detail, err := fetchRuneDetail(client, id) + if err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] fetch detail error: %v\n", id, err) + return + } + + // Claim the rune. + if err := claimRune(client, id, cfg.Claimant); err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] claim error: %v\n", id, err) + return + } + + // Resolve dispatch. + input := dispatchInputFromRune(detail) + result, err := dispatcher.Dispatch(ctx, input) + if err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] dispatcher error: %v\n", id, err) + unclaimRune(client, id) + return + } + + // Empty command means skip. + if result.Command == "" { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] no handler, unclaiming\n", id) + unclaimRune(client, id) + return + } + + if dryRun { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] dry-run: would invoke: %s %v\n", id, result.Command, result.Args) + unclaimRune(client, id) + return + } + + fmt.Fprintf(os.Stderr, "orchestrate: [%s] invoking: %s %v\n", id, result.Command, result.Args) + + exitCode, err := RunDispatched(ctx, result, os.Stdout, os.Stderr) + if err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] exec error: %v\n", id, err) + if unclaimOnFailure { + unclaimRune(client, id) + } + return + } + + if exitCode != 0 { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] agent exited with code %d\n", id, exitCode) + if unclaimOnFailure { + unclaimRune(client, id) + } + return + } + + if err := fulfillRune(client, id); err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] fulfill error: %v\n", id, err) + } +} + +func fetchRuneDetail(client *Client, id string) (map[string]any, error) { + body, err := client.DoGetWithParams("/rune", map[string]string{"id": id}) + if err != nil { + return nil, err + } + var detail map[string]any + if err := json.Unmarshal(body, &detail); err != nil { + return nil, fmt.Errorf("parsing rune detail: %w", err) + } + return detail, nil +} + +func claimRune(client *Client, id, claimant string) error { + _, err := client.DoPost("/claim-rune", map[string]string{"id": id, "claimant": claimant}) + return err +} + +func unclaimRune(client *Client, id string) { + if _, err := client.DoPost("/unclaim-rune", map[string]string{"id": id}); err != nil { + fmt.Fprintf(os.Stderr, "orchestrate: [%s] unclaim error: %v\n", id, err) + } +} + +func fulfillRune(client *Client, id string) error { + _, err := client.DoPost("/fulfill-rune", map[string]string{"id": id}) + return err +} \ No newline at end of file diff --git a/cli/orchestrate_dispatch.go b/cli/orchestrate_dispatch.go new file mode 100644 index 0000000..9900f35 --- /dev/null +++ b/cli/orchestrate_dispatch.go @@ -0,0 +1,113 @@ +package cli + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "os/exec" +) + +// DispatchInput is the rune data sent to the dispatcher script via stdin. +type DispatchInput struct { + ID string `json:"id"` + Title string `json:"title"` + Description string `json:"description,omitempty"` + Status string `json:"status"` + Priority int `json:"priority"` + Tags []string `json:"tags,omitempty"` + Notes []any `json:"notes,omitempty"` + Dependencies []any `json:"dependencies,omitempty"` +} + +// DispatchResult is the execution plan returned by the dispatcher script via stdout. +// If Command is empty, the rune should be skipped (unclaimed). +type DispatchResult struct { + Command string `json:"command"` + Args []string `json:"args"` + Stdin string `json:"stdin"` + Env map[string]string `json:"env"` +} + +// Dispatcher resolves a rune to an execution plan. +type Dispatcher interface { + Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) +} + +// ScriptDispatcher invokes an external script to resolve a rune. +// The script receives the rune JSON on stdin and writes a DispatchResult JSON to stdout. +type ScriptDispatcher struct { + ScriptPath string +} + +// Dispatch invokes the external script with rune data on stdin and parses the result. +// Returns nil result (no error) when the script signals skip via empty Command. +func (d *ScriptDispatcher) Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) { + inputJSON, err := json.Marshal(rune) + if err != nil { + return nil, fmt.Errorf("marshaling dispatch input: %w", err) + } + + cmd := exec.CommandContext(ctx, d.ScriptPath) //nolint:gosec + cmd.Stdin = bytes.NewReader(inputJSON) + + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + + if err := cmd.Run(); err != nil { + // Check for context cancellation + if ctx.Err() != nil { + return nil, ctx.Err() + } + raw := stdout.String() + if raw == "" { + raw = stderr.String() + } + return nil, fmt.Errorf("dispatcher exited with error: %w\n%s", err, raw) + } + + var result DispatchResult + if err := json.Unmarshal(stdout.Bytes(), &result); err != nil { + return nil, fmt.Errorf("dispatcher output is not valid JSON: %w\nraw output: %s", err, stdout.String()) + } + + return &result, nil +} + +// dispatchInputFromRune converts a rune detail map (from the API) into a DispatchInput. +func dispatchInputFromRune(detail map[string]any) DispatchInput { + input := DispatchInput{ + ID: stringField(detail, "id"), + Title: stringField(detail, "title"), + Description: stringField(detail, "description"), + Status: stringField(detail, "status"), + } + + if p, ok := detail["priority"].(float64); ok { + input.Priority = int(p) + } + + if tags, ok := detail["tags"].([]any); ok { + for _, t := range tags { + if s, ok := t.(string); ok { + input.Tags = append(input.Tags, s) + } + } + } + + if notes, ok := detail["notes"].([]any); ok { + input.Notes = notes + } + + if deps, ok := detail["dependencies"].([]any); ok { + input.Dependencies = deps + } + + return input +} + +func stringField(m map[string]any, key string) string { + s, _ := m[key].(string) + return s +} \ No newline at end of file diff --git a/cli/orchestrate_dispatch_test.go b/cli/orchestrate_dispatch_test.go new file mode 100644 index 0000000..b2479af --- /dev/null +++ b/cli/orchestrate_dispatch_test.go @@ -0,0 +1,150 @@ +package cli + +import ( + "context" + "encoding/json" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestScriptDispatcher(t *testing.T) { + t.Run("sends rune JSON to dispatcher stdin and parses result", func(t *testing.T) { + script := writeScript(t, `#!/bin/sh +input=$(cat) +id=$(echo "$input" | grep -o '"id":"[^"]*"' | head -1 | cut -d'"' -f4) +echo "{\"command\":\"echo\",\"args\":[\"$id\"],\"stdin\":\"\",\"env\":{}}" +`) + d := &ScriptDispatcher{ScriptPath: script} + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc123", Title: "Test"}) + + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, "echo", result.Command) + assert.Equal(t, []string{"bf-abc123"}, result.Args) + }) + + t.Run("returns nil result when command is empty (skip signal)", func(t *testing.T) { + script := writeScript(t, `#!/bin/sh +echo '{"command":""}' +`) + d := &ScriptDispatcher{ScriptPath: script} + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc"}) + + require.NoError(t, err) + require.NotNil(t, result) + assert.Equal(t, "", result.Command) + }) + + t.Run("returns error when dispatcher exits non-zero", func(t *testing.T) { + script := writeScript(t, `#!/bin/sh +echo "something went wrong" >&2 +exit 1 +`) + d := &ScriptDispatcher{ScriptPath: script} + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc"}) + + require.Error(t, err) + assert.Nil(t, result) + assert.Contains(t, err.Error(), "dispatcher exited with error") + }) + + t.Run("returns error when dispatcher outputs invalid JSON", func(t *testing.T) { + script := writeScript(t, `#!/bin/sh +echo "not valid json" +`) + d := &ScriptDispatcher{ScriptPath: script} + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc"}) + + require.Error(t, err) + assert.Nil(t, result) + assert.Contains(t, err.Error(), "not valid JSON") + }) + + t.Run("passes all rune fields to dispatcher", func(t *testing.T) { + var capturedInput DispatchInput + script := writeScript(t, `#!/bin/sh +cat > /tmp/bf_dispatch_test_input.json +echo '{"command":"echo","args":[],"stdin":"","env":{}}' +`) + d := &ScriptDispatcher{ScriptPath: script} + input := DispatchInput{ + ID: "bf-xyz", + Title: "My Task", + Description: "Do something", + Status: "open", + Priority: 2, + Tags: []string{"backend", "urgent"}, + } + _, err := d.Dispatch(context.Background(), input) + require.NoError(t, err) + + // Read the captured input file the script wrote. + data, err := os.ReadFile("/tmp/bf_dispatch_test_input.json") + require.NoError(t, err) + require.NoError(t, json.Unmarshal(data, &capturedInput)) + + assert.Equal(t, "bf-xyz", capturedInput.ID) + assert.Equal(t, "My Task", capturedInput.Title) + assert.Equal(t, "Do something", capturedInput.Description) + assert.Equal(t, 2, capturedInput.Priority) + assert.Equal(t, []string{"backend", "urgent"}, capturedInput.Tags) + }) +} + +func TestDispatchInputFromRune(t *testing.T) { + t.Run("converts rune detail map to DispatchInput", func(t *testing.T) { + detail := map[string]any{ + "id": "bf-abc", + "title": "Do work", + "description": "Some desc", + "status": "open", + "priority": float64(1), + "tags": []any{"alpha", "beta"}, + "notes": []any{map[string]any{"text": "a note"}}, + "dependencies": []any{map[string]any{ + "target_id": "bf-dep", + "relationship": "blocks", + }}, + } + + input := dispatchInputFromRune(detail) + + assert.Equal(t, "bf-abc", input.ID) + assert.Equal(t, "Do work", input.Title) + assert.Equal(t, "Some desc", input.Description) + assert.Equal(t, "open", input.Status) + assert.Equal(t, 1, input.Priority) + assert.Equal(t, []string{"alpha", "beta"}, input.Tags) + assert.Len(t, input.Notes, 1) + assert.Len(t, input.Dependencies, 1) + }) + + t.Run("handles missing optional fields gracefully", func(t *testing.T) { + detail := map[string]any{ + "id": "bf-minimal", + "title": "Minimal", + } + + input := dispatchInputFromRune(detail) + + assert.Equal(t, "bf-minimal", input.ID) + assert.Equal(t, "", input.Description) + assert.Equal(t, 0, input.Priority) + assert.Nil(t, input.Tags) + assert.Nil(t, input.Notes) + assert.Nil(t, input.Dependencies) + }) +} + +// writeScript creates a temporary executable shell script and returns its path. +func writeScript(t *testing.T, content string) string { + t.Helper() + dir := t.TempDir() + path := filepath.Join(dir, "dispatcher.sh") + require.NoError(t, os.WriteFile(path, []byte(content), 0o755)) + return path +} diff --git a/cli/orchestrate_exec.go b/cli/orchestrate_exec.go new file mode 100644 index 0000000..1672998 --- /dev/null +++ b/cli/orchestrate_exec.go @@ -0,0 +1,39 @@ +package cli + +import ( + "context" + "fmt" + "io" + "os" + "os/exec" + "strings" +) + +// RunDispatched executes a DispatchResult as a subprocess, streaming stdout/stderr +// to the provided writers. Returns the exit code and any execution error. +func RunDispatched(ctx context.Context, result *DispatchResult, stdout, stderr io.Writer) (int, error) { + cmd := exec.CommandContext(ctx, result.Command, result.Args...) //nolint:gosec + + if result.Stdin != "" { + cmd.Stdin = strings.NewReader(result.Stdin) + } + + if cmd.Env == nil { + cmd.Env = os.Environ() + } + for k, v := range result.Env { + cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", k, v)) + } + + cmd.Stdout = stdout + cmd.Stderr = stderr + + if err := cmd.Run(); err != nil { + if exitErr, ok := err.(*exec.ExitError); ok { + return exitErr.ExitCode(), nil + } + return -1, err + } + + return 0, nil +} \ No newline at end of file diff --git a/cli/orchestrate_exec_test.go b/cli/orchestrate_exec_test.go new file mode 100644 index 0000000..5ca21fe --- /dev/null +++ b/cli/orchestrate_exec_test.go @@ -0,0 +1,80 @@ +package cli + +import ( + "bytes" + "context" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRunDispatched(t *testing.T) { + t.Run("streams stdout to provided writer", func(t *testing.T) { + var stdout, stderr bytes.Buffer + result := &DispatchResult{ + Command: "echo", + Args: []string{"hello world"}, + } + + code, err := RunDispatched(context.Background(), result, &stdout, &stderr) + + require.NoError(t, err) + assert.Equal(t, 0, code) + assert.Contains(t, stdout.String(), "hello world") + }) + + t.Run("returns non-zero exit code without error", func(t *testing.T) { + var stdout, stderr bytes.Buffer + result := &DispatchResult{ + Command: "sh", + Args: []string{"-c", "exit 42"}, + } + + code, err := RunDispatched(context.Background(), result, &stdout, &stderr) + + require.NoError(t, err) + assert.Equal(t, 42, code) + }) + + t.Run("pipes stdin content to subprocess", func(t *testing.T) { + var stdout, stderr bytes.Buffer + result := &DispatchResult{ + Command: "cat", + Stdin: "hello from stdin", + } + + code, err := RunDispatched(context.Background(), result, &stdout, &stderr) + + require.NoError(t, err) + assert.Equal(t, 0, code) + assert.Contains(t, stdout.String(), "hello from stdin") + }) + + t.Run("returns error when command is not found", func(t *testing.T) { + var stdout, stderr bytes.Buffer + result := &DispatchResult{ + Command: "/nonexistent/command/that/does/not/exist", + } + + code, err := RunDispatched(context.Background(), result, &stdout, &stderr) + + require.Error(t, err) + assert.Equal(t, -1, code) + }) + + t.Run("injects env vars into subprocess", func(t *testing.T) { + var stdout, stderr bytes.Buffer + result := &DispatchResult{ + Command: "sh", + Args: []string{"-c", "echo $MY_TEST_VAR"}, + Env: map[string]string{"MY_TEST_VAR": "injected_value"}, + } + + code, err := RunDispatched(context.Background(), result, &stdout, &stderr) + + require.NoError(t, err) + assert.Equal(t, 0, code) + assert.Contains(t, stdout.String(), "injected_value") + }) +} diff --git a/cli/orchestrate_test.go b/cli/orchestrate_test.go new file mode 100644 index 0000000..45174b5 --- /dev/null +++ b/cli/orchestrate_test.go @@ -0,0 +1,383 @@ +package cli + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRunOrchestrator(t *testing.T) { + t.Run("polls ready runes with correct parameters", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + tc.ready_runes([]map[string]any{}) + dispatcher := &stubDispatcher{} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + tc.assert_request_made("GET", "/api/runes") + tc.assert_query_param("status", "open") + tc.assert_query_param("blocked", "false") + tc.assert_query_param("is_saga", "false") + }) + + t.Run("filters by saga when provided", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + tc.ready_runes([]map[string]any{}) + dispatcher := &stubDispatcher{} + + // When + tc.run_once_with_saga(dispatcher, "bf-saga-1") + + // Then + tc.assert_query_param("saga", "bf-saga-1") + }) + + t.Run("skips already-claimed runes", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + tc.ready_runes([]map[string]any{ + {"id": "bf-1", "title": "Already Claimed", "claimant": "someone"}, + }) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: "echo", Args: []string{"hi"}}} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + assert.Equal(t, 0, dispatcher.callCount, "dispatcher should not be called for claimed runes") + }) + + t.Run("claims rune before dispatching", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: "true"}} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + tc.assert_request_made("POST", "/api/claim-rune") + tc.assert_claim_body("bf-abc", "orchestrator") + }) + + t.Run("unclaims rune when dispatcher returns error", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &failingDispatcher{} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + tc.assert_request_made("POST", "/api/unclaim-rune") + }) + + t.Run("unclaims rune when dispatcher returns empty command", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: ""}} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + tc.assert_request_made("POST", "/api/unclaim-rune") + }) + + t.Run("fulfills rune on successful agent execution", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: "true"}} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + tc.assert_request_made("POST", "/api/fulfill-rune") + }) + + t.Run("leaves rune claimed on failure by default", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: "sh", Args: []string{"-c", "exit 1"}}} + + // When + tc.run_once(dispatcher, "", false, false) + + // Then + tc.assert_no_request("POST", "/api/unclaim-rune") + tc.assert_no_request("POST", "/api/fulfill-rune") + }) + + t.Run("unclaims rune on failure when --unclaim-on-failure set", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: "sh", Args: []string{"-c", "exit 1"}}} + + // When + tc.run_once(dispatcher, "", false, true) + + // Then + tc.assert_request_made("POST", "/api/unclaim-rune") + tc.assert_no_request("POST", "/api/fulfill-rune") + }) + + t.Run("dry-run: resolves dispatch but does not execute or fulfill", func(t *testing.T) { + tc := newOrchestratorTestContext(t) + + // Given + rune := map[string]any{"id": "bf-abc", "title": "Test", "claimant": ""} + tc.ready_runes_then_detail([]map[string]any{rune}, rune) + dispatcher := &stubDispatcher{result: &DispatchResult{Command: "echo", Args: []string{"hi"}}} + + // When + tc.run_once(dispatcher, "", true, false) + + // Then + assert.Equal(t, 1, dispatcher.callCount, "dispatcher should still be called in dry-run") + tc.assert_request_made("POST", "/api/claim-rune") + tc.assert_request_made("POST", "/api/unclaim-rune") + tc.assert_no_request("POST", "/api/fulfill-rune") + }) +} + +// --- Stub helpers --- + +type stubDispatcher struct { + result *DispatchResult + callCount int + mu sync.Mutex +} + +func (s *stubDispatcher) Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) { + s.mu.Lock() + s.callCount++ + s.mu.Unlock() + return s.result, nil +} + +type failingDispatcher struct{} + +func (f *failingDispatcher) Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) { + return nil, assert.AnError +} + +// --- Test context --- + +type orchestratorTestContext struct { + t *testing.T + server *httptest.Server + requests []recordedRequest + mu sync.Mutex + claimBodies []map[string]any +} + +type recordedRequest struct { + method string + path string + query map[string]string + body map[string]any +} + +// recordRequestHandler captures HTTP requests and routes responses based on handlerFn. +// handlerFn should encode the response body and write to w; recordRequestHandler handles +// request capture, query parsing, POST body unmarshaling, and cleanup. +type recordRequestHandler struct { + tc *orchestratorTestContext + handlerFn func(w http.ResponseWriter, r *http.Request, rec recordedRequest) +} + +func (h *recordRequestHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + rec := recordedRequest{ + method: r.Method, + path: r.URL.Path, + query: make(map[string]string), + } + for k, v := range r.URL.Query() { + if len(v) > 0 { + rec.query[k] = v[0] + } + } + if r.Method == "POST" { + body, _ := io.ReadAll(r.Body) + _ = json.Unmarshal(body, &rec.body) + } + + h.tc.mu.Lock() + h.tc.requests = append(h.tc.requests, rec) + if r.URL.Path == "/api/claim-rune" { + h.tc.claimBodies = append(h.tc.claimBodies, rec.body) + } + h.tc.mu.Unlock() + + h.handlerFn(w, r, rec) +} + +func newOrchestratorTestContext(t *testing.T) *orchestratorTestContext { + t.Helper() + tc := &orchestratorTestContext{t: t} + + tc.server = httptest.NewServer(&recordRequestHandler{ + tc: tc, + handlerFn: func(w http.ResponseWriter, r *http.Request, rec recordedRequest) { + // Route responses. + switch { + case r.Method == "GET" && r.URL.Path == "/api/runes": + // Served by handler set in test setup; default empty. + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode([]map[string]any{}) + default: + w.WriteHeader(http.StatusNoContent) + } + }, + }) + t.Cleanup(tc.server.Close) + + return tc +} + +// ready_runes configures the server to return these runes from GET /api/runes. +func (tc *orchestratorTestContext) ready_runes(runes []map[string]any) { + tc.t.Helper() + tc.server.Close() + + tc.server = httptest.NewServer(&recordRequestHandler{ + tc: tc, + handlerFn: func(w http.ResponseWriter, r *http.Request, rec recordedRequest) { + w.Header().Set("Content-Type", "application/json") + if r.Method == "GET" && r.URL.Path == "/api/runes" { + _ = json.NewEncoder(w).Encode(runes) + } else { + w.WriteHeader(http.StatusNoContent) + } + }, + }) + tc.t.Cleanup(tc.server.Close) +} + +// ready_runes_then_detail configures the server to return list runes and one detail rune. +func (tc *orchestratorTestContext) ready_runes_then_detail(runes []map[string]any, detail map[string]any) { + tc.t.Helper() + tc.server.Close() + + tc.server = httptest.NewServer(&recordRequestHandler{ + tc: tc, + handlerFn: func(w http.ResponseWriter, r *http.Request, rec recordedRequest) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == "GET" && r.URL.Path == "/api/runes": + _ = json.NewEncoder(w).Encode(runes) + case r.Method == "GET" && r.URL.Path == "/api/rune": + _ = json.NewEncoder(w).Encode(detail) + default: + w.WriteHeader(http.StatusNoContent) + } + }, + }) + tc.t.Cleanup(tc.server.Close) +} + +func (tc *orchestratorTestContext) client() *Client { + return NewClient(tc.server.URL, "test-key", "test-realm") +} + +func (tc *orchestratorTestContext) run_once(d Dispatcher, saga string, dryRun, unclaimOnFailure bool) { + tc.t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + cfg := OrchestrateConfig{ + Claimant: "orchestrator", + Concurrency: 1, + } + err := runOrchestrator(ctx, tc.client(), cfg, d, saga, dryRun, true, unclaimOnFailure) + require.NoError(tc.t, err) + // runOrchestrator returns synchronously after all workers finish in --once mode, + // so no additional wait is needed — all requests have been made. +} + +func (tc *orchestratorTestContext) run_once_with_saga(d Dispatcher, saga string) { + tc.t.Helper() + tc.run_once(d, saga, false, false) +} + +func (tc *orchestratorTestContext) assert_request_made(method, path string) { + tc.t.Helper() + tc.mu.Lock() + defer tc.mu.Unlock() + for _, r := range tc.requests { + if r.method == method && r.path == path { + return + } + } + tc.t.Errorf("expected request %s %s but it was not made; requests: %v", method, path, tc.requests) +} + +func (tc *orchestratorTestContext) assert_no_request(method, path string) { + tc.t.Helper() + tc.mu.Lock() + defer tc.mu.Unlock() + for _, r := range tc.requests { + if r.method == method && r.path == path { + tc.t.Errorf("expected NO request %s %s but it was made", method, path) + return + } + } +} + +func (tc *orchestratorTestContext) assert_query_param(key, value string) { + tc.t.Helper() + tc.mu.Lock() + defer tc.mu.Unlock() + for _, r := range tc.requests { + if r.method == "GET" && r.path == "/api/runes" { + assert.Equal(tc.t, value, r.query[key], "expected query param %s=%s", key, value) + return + } + } + tc.t.Errorf("no GET /api/runes request found") +} + +func (tc *orchestratorTestContext) assert_claim_body(id, claimant string) { + tc.t.Helper() + tc.mu.Lock() + defer tc.mu.Unlock() + require.NotEmpty(tc.t, tc.claimBodies, "no claim requests made") + body := tc.claimBodies[0] + assert.Equal(tc.t, id, body["id"]) + assert.Equal(tc.t, claimant, body["claimant"]) +} diff --git a/cli/rune_commands.go b/cli/rune_commands.go index b667c39..67c3b9f 100644 --- a/cli/rune_commands.go +++ b/cli/rune_commands.go @@ -7,6 +7,7 @@ import ( func RegisterRuneCommands(root *RootCmd, out *bytes.Buffer) { clientFn := func() *Client { return root.Client } + cfgFn := func() *Config { return root.Cfg } root.Command.AddCommand(NewCreateCmd(clientFn, out).Command) root.Command.AddCommand(NewShowCmd(clientFn, out).Command) @@ -23,4 +24,5 @@ func RegisterRuneCommands(root *RootCmd, out *bytes.Buffer) { root.Command.AddCommand(NewEventsCmd(clientFn, out).Command) root.Command.AddCommand(NewSweepCmd(clientFn, out, os.Stdin).Command) root.Command.AddCommand(NewShatterCmd(clientFn, out, os.Stdin).Command) + root.Command.AddCommand(NewOrchestrateCmd(clientFn, cfgFn).Command) }