From d93464366adaac042f32fa4b5f36de8a2df5a5e1 Mon Sep 17 00:00:00 2001 From: Eric Siebeneich Date: Fri, 10 Apr 2026 10:49:57 -0500 Subject: [PATCH 1/4] orchestrator --- cli/config.go | 9 +- cli/orchestrate.go | 312 +++++++++++++++++++++++ cli/orchestrate_dispatch.go | 108 ++++++++ cli/orchestrate_dispatch_test.go | 149 +++++++++++ cli/orchestrate_exec.go | 35 +++ cli/orchestrate_exec_test.go | 80 ++++++ cli/orchestrate_test.go | 407 +++++++++++++++++++++++++++++++ cli/rune_commands.go | 2 + 8 files changed, 1098 insertions(+), 4 deletions(-) create mode 100644 cli/orchestrate.go create mode 100644 cli/orchestrate_dispatch.go create mode 100644 cli/orchestrate_dispatch_test.go create mode 100644 cli/orchestrate_exec.go create mode 100644 cli/orchestrate_exec_test.go create mode 100644 cli/orchestrate_test.go 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..6c5e8bd --- /dev/null +++ b/cli/orchestrate.go @@ -0,0 +1,312 @@ +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 + } + } + + 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 + } + // Non-blocking send — if queue is full, release from in-flight and skip. + 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(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 +} diff --git a/cli/orchestrate_dispatch.go b/cli/orchestrate_dispatch.go new file mode 100644 index 0000000..8a1b459 --- /dev/null +++ b/cli/orchestrate_dispatch.go @@ -0,0 +1,108 @@ +package cli + +import ( + "bytes" + "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(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(rune DispatchInput) (*DispatchResult, error) { + inputJSON, err := json.Marshal(rune) + if err != nil { + return nil, fmt.Errorf("marshaling dispatch input: %w", err) + } + + cmd := exec.Command(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 { + 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 +} diff --git a/cli/orchestrate_dispatch_test.go b/cli/orchestrate_dispatch_test.go new file mode 100644 index 0000000..310c1fc --- /dev/null +++ b/cli/orchestrate_dispatch_test.go @@ -0,0 +1,149 @@ +package cli + +import ( + "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(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(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(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(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(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..73bc8ee --- /dev/null +++ b/cli/orchestrate_exec.go @@ -0,0 +1,35 @@ +package cli + +import ( + "context" + "fmt" + "io" + "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) + } + + 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 +} 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..c2e247f --- /dev/null +++ b/cli/orchestrate_test.go @@ -0,0 +1,407 @@ +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(rune DispatchInput) (*DispatchResult, error) { + s.mu.Lock() + s.callCount++ + s.mu.Unlock() + return s.result, nil +} + +type failingDispatcher struct{} + +func (f *failingDispatcher) Dispatch(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 +} + +func newOrchestratorTestContext(t *testing.T) *orchestratorTestContext { + t.Helper() + tc := &orchestratorTestContext{t: t} + + tc.server = httptest.NewServer(http.HandlerFunc(func(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) + } + + tc.mu.Lock() + tc.requests = append(tc.requests, rec) + if r.URL.Path == "/api/claim-rune" { + tc.claimBodies = append(tc.claimBodies, rec.body) + } + tc.mu.Unlock() + + // 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(http.HandlerFunc(func(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) + } + + tc.mu.Lock() + tc.requests = append(tc.requests, rec) + if r.URL.Path == "/api/claim-rune" { + tc.claimBodies = append(tc.claimBodies, rec.body) + } + tc.mu.Unlock() + + 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(http.HandlerFunc(func(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) + } + + tc.mu.Lock() + tc.requests = append(tc.requests, rec) + if r.URL.Path == "/api/claim-rune" { + tc.claimBodies = append(tc.claimBodies, rec.body) + } + tc.mu.Unlock() + + 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) + + // Small wait for goroutines to finish after queue drains. + time.Sleep(50 * time.Millisecond) +} + +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) } From b05254859114f26149fdf815431b8477e7314083 Mon Sep 17 00:00:00 2001 From: "coderabbitai[bot]" <136622811+coderabbitai[bot]@users.noreply.github.com> Date: Fri, 10 Apr 2026 16:06:12 +0000 Subject: [PATCH 2/4] fix: apply CodeRabbit auto-fixes Fixed 3 file(s) based on 4 unresolved review comments. Co-authored-by: CodeRabbit --- cli/orchestrate.go | 27 ++++++++++++++++++++------- cli/orchestrate_dispatch.go | 13 +++++++++---- cli/orchestrate_exec.go | 6 +++++- 3 files changed, 34 insertions(+), 12 deletions(-) diff --git a/cli/orchestrate.go b/cli/orchestrate.go index 6c5e8bd..7c579f0 100644 --- a/cli/orchestrate.go +++ b/cli/orchestrate.go @@ -65,6 +65,14 @@ func NewOrchestrateCmd(clientFn func() *Client, cfgFn func() *Config) *Orchestra } } + // 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") } @@ -151,11 +159,16 @@ func runOrchestrator( if _, loaded := inFlight.LoadOrStore(id, struct{}{}); loaded { continue } - // Non-blocking send — if queue is full, release from in-flight and skip. - select { - case queue <- r: - default: - inFlight.Delete(id) + // 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) + } } } } @@ -239,7 +252,7 @@ func processRune( // Resolve dispatch. input := dispatchInputFromRune(detail) - result, err := dispatcher.Dispatch(input) + result, err := dispatcher.Dispatch(ctx, input) if err != nil { fmt.Fprintf(os.Stderr, "orchestrate: [%s] dispatcher error: %v\n", id, err) unclaimRune(client, id) @@ -309,4 +322,4 @@ func unclaimRune(client *Client, id string) { 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 index 8a1b459..9900f35 100644 --- a/cli/orchestrate_dispatch.go +++ b/cli/orchestrate_dispatch.go @@ -2,6 +2,7 @@ package cli import ( "bytes" + "context" "encoding/json" "fmt" "os/exec" @@ -30,7 +31,7 @@ type DispatchResult struct { // Dispatcher resolves a rune to an execution plan. type Dispatcher interface { - Dispatch(rune DispatchInput) (*DispatchResult, error) + Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) } // ScriptDispatcher invokes an external script to resolve a rune. @@ -41,13 +42,13 @@ type ScriptDispatcher struct { // 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(rune DispatchInput) (*DispatchResult, error) { +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.Command(d.ScriptPath) //nolint:gosec + cmd := exec.CommandContext(ctx, d.ScriptPath) //nolint:gosec cmd.Stdin = bytes.NewReader(inputJSON) var stdout, stderr bytes.Buffer @@ -55,6 +56,10 @@ func (d *ScriptDispatcher) Dispatch(rune DispatchInput) (*DispatchResult, error) 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() @@ -105,4 +110,4 @@ func dispatchInputFromRune(detail map[string]any) DispatchInput { 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_exec.go b/cli/orchestrate_exec.go index 73bc8ee..1672998 100644 --- a/cli/orchestrate_exec.go +++ b/cli/orchestrate_exec.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "os" "os/exec" "strings" ) @@ -17,6 +18,9 @@ func RunDispatched(ctx context.Context, result *DispatchResult, stdout, stderr i 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)) } @@ -32,4 +36,4 @@ func RunDispatched(ctx context.Context, result *DispatchResult, stdout, stderr i } return 0, nil -} +} \ No newline at end of file From d0bc97cf8374618e3b84d80035ff1c634a83f9a6 Mon Sep 17 00:00:00 2001 From: Eric Siebeneich Date: Fri, 10 Apr 2026 13:14:52 -0500 Subject: [PATCH 3/4] fix: update dispatcher interface to accept context The Dispatcher interface now accepts context.Context as the first parameter to support proper cancellation and timeout handling. Update all test stubs and test calls to pass context.Background(). This aligns with the changes made to ScriptDispatcher.Dispatch() and improves the overall cancellation semantics in the orchestrator. Co-Authored-By: Claude Haiku 4.5 --- cli/orchestrate_dispatch_test.go | 11 ++++++----- cli/orchestrate_test.go | 4 ++-- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/cli/orchestrate_dispatch_test.go b/cli/orchestrate_dispatch_test.go index 310c1fc..b2479af 100644 --- a/cli/orchestrate_dispatch_test.go +++ b/cli/orchestrate_dispatch_test.go @@ -1,6 +1,7 @@ package cli import ( + "context" "encoding/json" "os" "path/filepath" @@ -18,7 +19,7 @@ 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(DispatchInput{ID: "bf-abc123", Title: "Test"}) + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc123", Title: "Test"}) require.NoError(t, err) require.NotNil(t, result) @@ -31,7 +32,7 @@ echo "{\"command\":\"echo\",\"args\":[\"$id\"],\"stdin\":\"\",\"env\":{}}" echo '{"command":""}' `) d := &ScriptDispatcher{ScriptPath: script} - result, err := d.Dispatch(DispatchInput{ID: "bf-abc"}) + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc"}) require.NoError(t, err) require.NotNil(t, result) @@ -44,7 +45,7 @@ echo "something went wrong" >&2 exit 1 `) d := &ScriptDispatcher{ScriptPath: script} - result, err := d.Dispatch(DispatchInput{ID: "bf-abc"}) + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc"}) require.Error(t, err) assert.Nil(t, result) @@ -56,7 +57,7 @@ exit 1 echo "not valid json" `) d := &ScriptDispatcher{ScriptPath: script} - result, err := d.Dispatch(DispatchInput{ID: "bf-abc"}) + result, err := d.Dispatch(context.Background(), DispatchInput{ID: "bf-abc"}) require.Error(t, err) assert.Nil(t, result) @@ -78,7 +79,7 @@ echo '{"command":"echo","args":[],"stdin":"","env":{}}' Priority: 2, Tags: []string{"backend", "urgent"}, } - _, err := d.Dispatch(input) + _, err := d.Dispatch(context.Background(), input) require.NoError(t, err) // Read the captured input file the script wrote. diff --git a/cli/orchestrate_test.go b/cli/orchestrate_test.go index c2e247f..e5cc04a 100644 --- a/cli/orchestrate_test.go +++ b/cli/orchestrate_test.go @@ -182,7 +182,7 @@ type stubDispatcher struct { mu sync.Mutex } -func (s *stubDispatcher) Dispatch(rune DispatchInput) (*DispatchResult, error) { +func (s *stubDispatcher) Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) { s.mu.Lock() s.callCount++ s.mu.Unlock() @@ -191,7 +191,7 @@ func (s *stubDispatcher) Dispatch(rune DispatchInput) (*DispatchResult, error) { type failingDispatcher struct{} -func (f *failingDispatcher) Dispatch(rune DispatchInput) (*DispatchResult, error) { +func (f *failingDispatcher) Dispatch(ctx context.Context, rune DispatchInput) (*DispatchResult, error) { return nil, assert.AnError } From 7db8f9470bd9557345527943f2379a705ddd17c9 Mon Sep 17 00:00:00 2001 From: Eric Siebeneich Date: Fri, 10 Apr 2026 14:49:05 -0500 Subject: [PATCH 4/4] refactor: improve orchestrate test reliability and reduce duplication Inline comment fixes: - Replace flaky time.Sleep(50ms) with deterministic sync: runOrchestrator returns synchronously after workers finish in --once mode, no additional wait needed. Nitpick fixes: - Extract duplicated request recording logic into recordRequestHandler helper. Three test helpers (newOrchestratorTestContext, ready_runes, ready_runes_then_detail) had identical code for capturing HTTP requests, parsing query params, unmarshaling POST bodies, and recording to tc.requests and tc.claimBodies. Consolidate into an http.Handler that can be wrapped with different routing functions. This reduces code duplication by ~80 lines and makes future changes to request recording centralized. Co-Authored-By: Claude Haiku 4.5 --- cli/orchestrate_test.go | 178 +++++++++++++++++----------------------- 1 file changed, 77 insertions(+), 101 deletions(-) diff --git a/cli/orchestrate_test.go b/cli/orchestrate_test.go index e5cc04a..45174b5 100644 --- a/cli/orchestrate_test.go +++ b/cli/orchestrate_test.go @@ -198,10 +198,10 @@ func (f *failingDispatcher) Dispatch(ctx context.Context, rune DispatchInput) (* // --- Test context --- type orchestratorTestContext struct { - t *testing.T - server *httptest.Server - requests []recordedRequest - mu sync.Mutex + t *testing.T + server *httptest.Server + requests []recordedRequest + mu sync.Mutex claimBodies []map[string]any } @@ -212,44 +212,59 @@ type recordedRequest struct { 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(http.HandlerFunc(func(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] + 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) } - } - if r.Method == "POST" { - body, _ := io.ReadAll(r.Body) - _ = json.Unmarshal(body, &rec.body) - } - - tc.mu.Lock() - tc.requests = append(tc.requests, rec) - if r.URL.Path == "/api/claim-rune" { - tc.claimBodies = append(tc.claimBodies, rec.body) - } - tc.mu.Unlock() - - // 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 @@ -260,36 +275,17 @@ func (tc *orchestratorTestContext) ready_runes(runes []map[string]any) { tc.t.Helper() tc.server.Close() - tc.server = httptest.NewServer(http.HandlerFunc(func(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] + 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) } - } - if r.Method == "POST" { - body, _ := io.ReadAll(r.Body) - _ = json.Unmarshal(body, &rec.body) - } - - tc.mu.Lock() - tc.requests = append(tc.requests, rec) - if r.URL.Path == "/api/claim-rune" { - tc.claimBodies = append(tc.claimBodies, rec.body) - } - tc.mu.Unlock() - - 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) } @@ -298,39 +294,20 @@ func (tc *orchestratorTestContext) ready_runes_then_detail(runes []map[string]an tc.t.Helper() tc.server.Close() - tc.server = httptest.NewServer(http.HandlerFunc(func(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] + 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) } - } - if r.Method == "POST" { - body, _ := io.ReadAll(r.Body) - _ = json.Unmarshal(body, &rec.body) - } - - tc.mu.Lock() - tc.requests = append(tc.requests, rec) - if r.URL.Path == "/api/claim-rune" { - tc.claimBodies = append(tc.claimBodies, rec.body) - } - tc.mu.Unlock() - - 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) } @@ -349,9 +326,8 @@ func (tc *orchestratorTestContext) run_once(d Dispatcher, saga string, dryRun, u } err := runOrchestrator(ctx, tc.client(), cfg, d, saga, dryRun, true, unclaimOnFailure) require.NoError(tc.t, err) - - // Small wait for goroutines to finish after queue drains. - time.Sleep(50 * time.Millisecond) + // 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) {