feat: add bf orchestrate command for agent dispatch#15
Merged
Conversation
Contributor
📝 WalkthroughSummary by CodeRabbit
WalkthroughAdds an orchestrator CLI command and supporting code to poll backend runes, claim and dispatch them to an external dispatcher script, execute dispatched commands with worker concurrency, and conditionally fulfill or unclaim runes. Also adds Changes
Sequence Diagram(s)sequenceDiagram
participant Orchestrator
participant BackendAPI as Backend API
participant Dispatcher as Dispatcher Script
participant Worker as Worker
participant Subproc as OS Subprocess
Orchestrator->>BackendAPI: GET /runes?status=open&blocked=false&is_saga=false
BackendAPI-->>Orchestrator: Rune summaries
Orchestrator->>Worker: Enqueue rune ID
Worker->>BackendAPI: GET /rune?id=...
BackendAPI-->>Worker: Rune details
Worker->>BackendAPI: POST /claim-rune
BackendAPI-->>Worker: Claim acknowledged
Worker->>Dispatcher: JSON DispatchInput (stdin)
Dispatcher-->>Worker: JSON DispatchResult (stdout)
alt DispatchResult.Command != ""
Worker->>Subproc: exec command args, stdin, env
Subproc-->>Worker: exit code
alt exit code == 0
Worker->>BackendAPI: POST /fulfill-rune
BackendAPI-->>Worker: Fulfilled
else exit code != 0
Worker->>BackendAPI: POST /unclaim-rune (if configured)
BackendAPI-->>Worker: Unclaimed
end
else empty command or dispatch failure
Worker->>BackendAPI: POST /unclaim-rune
BackendAPI-->>Worker: Unclaimed
end
Orchestrator->>Orchestrator: Shutdown on SIGINT/SIGTERM or --once completion
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. Warning There were issues while running some tools. Please review the errors and either fix the tool's configuration or disable the tool if it's a critical failure. 🔧 golangci-lint (2.11.4)level=error msg="[linters_context] typechecking error: pattern ./...: directory prefix . does not contain modules listed in go.work or their selected dependencies" Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@cli/orchestrate_dispatch.go`:
- Around line 31-57: The Dispatcher interface and ScriptDispatcher.Dispatch must
accept a context so orchestrator cancellation propagates; change the
Dispatcher.Dispatch signature to Dispatch(ctx context.Context, rune
DispatchInput) (*DispatchResult, error), update ScriptDispatcher.Dispatch to the
new signature, and inside ScriptDispatcher.Dispatch use exec.CommandContext(ctx,
d.ScriptPath) instead of exec.Command(d.ScriptPath), wiring ctx into
stdin/stderr handling and returning on context cancellation; also update all
callers of Dispatcher.Dispatch to pass the orchestrator's context.
In `@cli/orchestrate_exec.go`:
- Around line 20-22: The loop that appends result.Env to cmd.Env breaks
inheritance because cmd.Env is nil; initialize cmd.Env with the current process
environment (os.Environ()) before appending result.Env entries so child
processes keep PATH, HOME, etc.; update the code around the loop that uses
cmd.Env and result.Env (the command construction logic in orchestrate_exec.go)
to set cmd.Env = os.Environ() when cmd.Env is nil, then append
fmt.Sprintf("%s=%s", k, v) for each k,v in result.Env.
In `@cli/orchestrate.go`:
- Around line 55-80: After applying defaults, validate that oCfg.Concurrency and
oCfg.PollInterval are strictly positive and return a clear error if not (e.g.,
if oCfg.Concurrency <= 0 or oCfg.PollInterval <= 0) so invalid negative values
from flags or .bifrost.yaml are rejected before calling runOrchestrator; this
prevents panics in channel allocation (cfg.Concurrency) and time.NewTicker
(cfg.PollInterval). Locate the checks around oCfg.PollInterval and
oCfg.Concurrency in orchestrate.go and add the validation immediately after the
defaulting block, returning fmt.Errorf with a concise message identifying the
bad field.
- Around line 154-175: The current non-blocking send to queue silently drops
runes when the buffer is full; for --once mode you must guarantee the entire
fetched batch is enqueued. Change the enqueue logic in poll()/where you
currently do the select { case queue <- r: default: inFlight.Delete(id) } to do
a blocking send when once is true (e.g. if once { queue <- r } else { existing
non-blocking select ... }), so items aren’t skipped for the final batch; keep
the inFlight/Delete behavior for the non-blocking branch and leave the
close(queue) and wg.Wait() shutdown logic unchanged.
🪄 Autofix (Beta)
❌ Autofix failed (check again to retry)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: a0bbf148-ed53-46a7-a713-de6060e914cc
📒 Files selected for processing (8)
cli/config.gocli/orchestrate.gocli/orchestrate_dispatch.gocli/orchestrate_dispatch_test.gocli/orchestrate_exec.gocli/orchestrate_exec_test.gocli/orchestrate_test.gocli/rune_commands.go
Contributor
|
Note Autofix is a beta feature. Expect some limitations and changes as we gather feedback and continue to improve it. The branch was updated while autofix was in progress. Please try again. |
Contributor
|
Note Autofix is a beta feature. Expect some limitations and changes as we gather feedback and continue to improve it. Fixes Applied SuccessfullyFixed 3 file(s) based on 4 unresolved review comments. Files modified:
Commit: The changes have been pushed to the Time taken: |
Fixed 3 file(s) based on 4 unresolved review comments. Co-authored-by: CodeRabbit <noreply@coderabbit.ai>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@cli/orchestrate.go`:
- Around line 162-166: The blocking enqueue in runOrchestrator (the branch using
`once` with `queue <- r`) should be made cancellation-aware so the main
goroutine can observe ctx.Done() when the queue is full; replace the direct send
with a select that waits on either `queue <- r` or `ctx.Done()` and, on ctx
cancellation, stop enqueuing (return or break out to allow shutdown) instead of
blocking indefinitely inside poll() while a worker (e.g., processRune) is stuck.
Ensure the change references the same variables (`once`, `queue`, `r`,
`ctx.Done()`) and that any in-flight bookkeeping remains consistent when you
bail out.
- Around line 206-218: The API calls block on non-cancellable HTTP calls; update
Client.DoGetWithParams and Client.DoPost to accept a context.Context and use it
when creating/performing requests, then propagate that context through all
callers: change fetchReadyRunes(ctx, client, saga), fetchRuneDetail(ctx, ...),
claimRune(ctx, ...), unclaimRune(ctx, ...), and fulfillRune(ctx, ...) to accept
ctx and call the new context-aware Client methods, and ensure runOrchestrator
continues to pass its ctx into dispatcher.Dispatch(ctx, ...) and
RunDispatched(ctx, ...) as well as into these API wrappers so any ctx.Done()
cancels in-flight requests promptly.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 4e449182-a357-4e49-84d7-f5d36d18f240
📒 Files selected for processing (3)
cli/orchestrate.gocli/orchestrate_dispatch.gocli/orchestrate_exec.go
Comment on lines
+162
to
+166
| // 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 { |
Contributor
There was a problem hiding this comment.
Make --once enqueue cancellation-aware.
The blocking queue <- r fixes batch loss, but it also means the main goroutine cannot observe ctx.Done() while the queue is full. If a worker is stuck in processRune, SIGINT/SIGTERM can leave runOrchestrator hung inside poll() instead of shutting down.
Suggested fix
// 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
+ select {
+ case queue <- r:
+ case <-ctx.Done():
+ inFlight.Delete(id)
+ return
+ }
} else {
select {
case queue <- r:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| // 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 { | |
| // 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 { | |
| select { | |
| case queue <- r: | |
| case <-ctx.Done(): | |
| inFlight.Delete(id) | |
| return | |
| } | |
| } else { |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@cli/orchestrate.go` around lines 162 - 166, The blocking enqueue in
runOrchestrator (the branch using `once` with `queue <- r`) should be made
cancellation-aware so the main goroutine can observe ctx.Done() when the queue
is full; replace the direct send with a select that waits on either `queue <- r`
or `ctx.Done()` and, on ctx cancellation, stop enqueuing (return or break out to
allow shutdown) instead of blocking indefinitely inside poll() while a worker
(e.g., processRune) is stuck. Ensure the change references the same variables
(`once`, `queue`, `r`, `ctx.Done()`) and that any in-flight bookkeeping remains
consistent when you bail out.
Comment on lines
+206
to
+218
| 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 |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Find and examine cli/orchestrate.go
echo "=== Checking cli/orchestrate.go structure ==="
wc -l cli/orchestrate.go
echo -e "\n=== Lines 206-218 (fetchReadyRunes) ==="
sed -n '206,218p' cli/orchestrate.go
echo -e "\n=== Lines 299-323 section ==="
sed -n '299,323p' cli/orchestrate.go
echo -e "\n=== Looking for runOrchestrator function ==="
grep -n "func runOrchestrator" cli/orchestrate.go
echo -e "\n=== Checking all API call functions mentioned ==="
grep -n "func fetchReadyRunes\|func fetchRuneDetail\|func claimRune\|func unclaimRune\|func fulfillRune" cli/orchestrate.goRepository: devzeebo/bifrost
Length of output: 1820
🏁 Script executed:
#!/bin/bash
# Examine cli/client.go for DoGetWithParams and DoPost signatures
echo "=== Checking DoGetWithParams signature ==="
grep -A 5 "func.*DoGetWithParams" cli/client.go
echo -e "\n=== Checking DoPost signature ==="
grep -A 5 "func.*DoPost" cli/client.go
echo -e "\n=== Checking Client struct ==="
grep -A 15 "type Client struct" cli/client.go
echo -e "\n=== Looking for context in Client methods ==="
rg "ctx.*context\.|context\.Context" cli/client.go -A 3Repository: devzeebo/bifrost
Length of output: 882
🏁 Script executed:
#!/bin/bash
echo "=== runOrchestrator function (starting at line 112) ==="
sed -n '112,170p' cli/orchestrate.go
echo -e "\n=== Full runOrchestrator signature ==="
grep -A 20 "func runOrchestrator" cli/orchestrate.go | head -25Repository: devzeebo/bifrost
Length of output: 1958
🏁 Script executed:
#!/bin/bash
echo "=== Checking for context.Context in runOrchestrator ==="
sed -n '112,200p' cli/orchestrate.go | grep -i "context\|ctx"
echo -e "\n=== Searching for cmd.CommandContext usage or context setup ==="
rg "CommandContext|context\.|ctx" cli/orchestrate.go -B 2 -A 2Repository: devzeebo/bifrost
Length of output: 1544
🏁 Script executed:
#!/bin/bash
echo "=== processRune function ==="
sed -n '224,298p' cli/orchestrate.go
echo -e "\n=== Where is poll() called ==="
sed -n '150,200p' cli/orchestrate.go | grep -A 5 -B 5 "poll()"Repository: devzeebo/bifrost
Length of output: 2193
🏁 Script executed:
#!/bin/bash
echo "=== Full poll() and main loop context ==="
sed -n '138,203p' cli/orchestrate.goRepository: devzeebo/bifrost
Length of output: 1528
Propagate ctx into the Bifrost API calls to enable graceful shutdown.
runOrchestrator properly passes context to the dispatcher and subprocess calls (dispatcher.Dispatch(ctx, ...) and RunDispatched(ctx, ...)), but the API operations—fetchReadyRunes, fetchRuneDetail, claimRune, unclaimRune, and fulfillRune—all invoke Client.DoGetWithParams and Client.DoPost without context support. A slow or hung API request in any of these functions will block the worker goroutines indefinitely, preventing wg.Wait() from completing even after SIGINT/SIGTERM triggers ctx.Done(). Update the Client methods and their callers to accept and respect context cancellation.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@cli/orchestrate.go` around lines 206 - 218, The API calls block on
non-cancellable HTTP calls; update Client.DoGetWithParams and Client.DoPost to
accept a context.Context and use it when creating/performing requests, then
propagate that context through all callers: change fetchReadyRunes(ctx, client,
saga), fetchRuneDetail(ctx, ...), claimRune(ctx, ...), unclaimRune(ctx, ...),
and fulfillRune(ctx, ...) to accept ctx and call the new context-aware Client
methods, and ensure runOrchestrator continues to pass its ctx into
dispatcher.Dispatch(ctx, ...) and RunDispatched(ctx, ...) as well as into these
API wrappers so any ctx.Done() cancels in-flight requests promptly.
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 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
cli/orchestrate_test.go (1)
219-252: Deduplicate server request-recording/handler setupThe request capture and POST-body decode logic is repeated across three server constructors. Extracting a shared recorder/router helper would reduce drift risk and keep future endpoint additions consistent.
Also applies to: 263-293, 301-335
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cli/orchestrate_test.go` around lines 219 - 252, The test HTTP server setup in orchestrate_test.go duplicates request capture and POST-body decoding across multiple constructors (see tc.server creation and recordedRequest usage); extract this logic into a shared helper function (e.g., newTestRecorderServer or attachRequestRecorder) that installs an http.Handler which records method/path/query/body into recordedRequest and appends to tc.requests and tc.claimBodies where applicable, and re-use that helper from each test instead of repeating the inline http.HandlerFunc (touch symbols: tc.server, recordedRequest, tc.requests, tc.claimBodies, and the existing switch routing for "/api/runes").
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@cli/orchestrate_test.go`:
- Around line 353-354: Replace the flaky fixed wait (time.Sleep(50 *
time.Millisecond)) with deterministic synchronization: remove the Sleep and
instead use require.Eventually (or wait on a channel/WaitGroup) to assert the
async work completed by polling a concrete observable condition such as the
length of observedRequests, a finished flag, or a worker completion channel used
in the test; update the test to wait until that condition is true within a
timeout (e.g., require.Eventually(t, func() bool { return len(observedRequests)
== expected }, timeout, interval)) so the test no longer depends on timing.
---
Nitpick comments:
In `@cli/orchestrate_test.go`:
- Around line 219-252: The test HTTP server setup in orchestrate_test.go
duplicates request capture and POST-body decoding across multiple constructors
(see tc.server creation and recordedRequest usage); extract this logic into a
shared helper function (e.g., newTestRecorderServer or attachRequestRecorder)
that installs an http.Handler which records method/path/query/body into
recordedRequest and appends to tc.requests and tc.claimBodies where applicable,
and re-use that helper from each test instead of repeating the inline
http.HandlerFunc (touch symbols: tc.server, recordedRequest, tc.requests,
tc.claimBodies, and the existing switch routing for "/api/runes").
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 4c1239f6-6bdf-405e-93e7-9e271c05b2e9
📒 Files selected for processing (2)
cli/orchestrate_dispatch_test.gocli/orchestrate_test.go
✅ Files skipped from review due to trivial changes (1)
- cli/orchestrate_dispatch_test.go
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 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
🧹 Nitpick comments (1)
cli/orchestrate_test.go (1)
234-237: Fail fast on request-body capture/parsing errors in the recorder.Ignoring read/unmarshal errors can mask malformed POST payloads and make assertions less trustworthy.
Proposed change
if r.Method == "POST" { - body, _ := io.ReadAll(r.Body) - _ = json.Unmarshal(body, &rec.body) + body, err := io.ReadAll(r.Body) + if err != nil { + h.tc.t.Errorf("failed to read request body: %v", err) + return + } + if len(body) > 0 { + if err := json.Unmarshal(body, &rec.body); err != nil { + h.tc.t.Errorf("failed to parse request body JSON: %v; body=%q", err, string(body)) + } + } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@cli/orchestrate_test.go` around lines 234 - 237, The recorder currently ignores errors from io.ReadAll and json.Unmarshal which hides malformed POST payloads; update the handler in orchestrate_test.go to check both errors (e.g., body, err := io.ReadAll(r.Body); if err != nil { t.Fatalf("failed reading request body: %v", err) } and if err := json.Unmarshal(body, &rec.body); err != nil { t.Fatalf("failed unmarshalling request body into rec.body: %v", err) }) so the test fails fast and includes the error context; reference the handler where r.Method == "POST" and the rec.body variable to locate the code to change.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@cli/orchestrate_test.go`:
- Around line 234-237: The recorder currently ignores errors from io.ReadAll and
json.Unmarshal which hides malformed POST payloads; update the handler in
orchestrate_test.go to check both errors (e.g., body, err := io.ReadAll(r.Body);
if err != nil { t.Fatalf("failed reading request body: %v", err) } and if err :=
json.Unmarshal(body, &rec.body); err != nil { t.Fatalf("failed unmarshalling
request body into rec.body: %v", err) }) so the test fails fast and includes the
error context; reference the handler where r.Method == "POST" and the rec.body
variable to locate the code to change.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: d4253927-f1cc-4fe7-a24b-4006675ff54f
📒 Files selected for processing (1)
cli/orchestrate_test.go
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Add a lightweight orchestrator that polls for ready runes and dispatches them to configurable CLI agents.
.bifrost.yamlor CLI flagsKey Design
Dispatcher interface: Any executable script that reads rune JSON and outputs:
{ "command": "claude", "args": ["-p", "backend-agent"], "stdin": "<rune context>", "env": {} }Empty
commandsignals skip (unclaim rune). Non-zero dispatcher exit or invalid JSON → unclaim and log.Concurrency model: N worker goroutines drain a buffered queue. In-flight set prevents double-queueing same rune across poll cycles. Default 1 (serial).
Claim failure behavior:
--unclaim-on-failureCommand
```bash
bf orchestrate --dispatcher ./scripts/bf-dispatch --saga bf-saga-1 --dry-run
```
Flags:
--dispatcher— path to dispatcher script--poll-interval— time between polls (default 10s)--concurrency— parallel workers (default 1)--claimant— who claims runes (default: system username)--unclaim-on-failure— unclaim when agent exits non-zero--saga— only orchestrate runes in this saga--dry-run— resolve dispatch but don't execute--once— process one batch then exitFiles
cli/orchestrate.go— polling loop, worker pool, command registrationcli/orchestrate_dispatch.go— dispatcher interface, script invocationcli/orchestrate_exec.go— subprocess execution with pipingcli/orchestrate_test.go,orchestrate_dispatch_test.go,orchestrate_exec_test.go— comprehensive testscli/config.go— addedOrchestratefield to configcli/rune_commands.go— registered commandTesting
All tests pass:
`make test MODULES=cli` — all green
`make lint MODULES=cli` — 0 issues
`make build` — builds cleanly
🤖 Generated with Claude Code