Repository navigation
fix: wake coordinator sessions reliably - #240
Conversation
|
@richardfogaca is attempting to deploy a commit to the Compozy Team on Vercel. A member of the Team first needs to authorize it. |
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (2)
🚧 Files skipped from review as they are similar to previous changes (2)
WalkthroughThis PR introduces synthetic prompt interruption capability, enabling coordinator bootstrap and harness reentry to interrupt agent-waiting session prompts. Changes span session manager interrupt logic, coordinator runtime lifecycle and prompting, harness integration, daemon wiring, and extended test coverage across subsystems. ChangesSynthetic Prompt Interruption Feature
Sequence Diagram(s)sequenceDiagram
participant Client
participant SessionMgr as Session Manager
participant CurrentPrompt as Current Prompt Turn
participant Agent as Agent
participant SessionState as Session State
Client->>SessionMgr: PromptSynthetic(InterruptIfAgentWaiting=true)
SessionMgr->>SessionState: Check if agent is waiting
alt Agent Waiting
SessionMgr->>CurrentPrompt: Cancel current prompt
CurrentPrompt->>Agent: Interrupt with cancel
SessionMgr->>SessionState: Wait for idle state
SessionState->>SessionMgr: Idle
SessionMgr->>SessionMgr: Submit synthetic prompt
else Agent Not Waiting
SessionMgr->>SessionMgr: Queue synthetic prompt (existing path)
end
SessionMgr->>Client: Return event channel
sequenceDiagram
participant Runtime as Coordinator Runtime
participant Sessions as Sessions Manager
participant Coordinator as Coordinator Session
participant Daemon as Daemon
Runtime->>Runtime: Detect queued run for bootstrap
Runtime->>Runtime: Check existing coordinator
alt No Existing Coordinator
Runtime->>Sessions: Create coordinator session
Runtime->>Sessions: PromptSynthetic (wake)
else Existing Coordinator Healthy
Runtime->>Runtime: Begin dedupe wake
Runtime->>Sessions: PromptSynthetic (existing, wake)
end
Sessions->>Coordinator: Synthetic prompt with metadata
Coordinator->>Runtime: Drain agent events
Runtime->>Daemon: Dispatch coordinator spawned/ready hook
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/daemon/coordinator_runtime_test.go`:
- Around line 83-97: The test currently verifies prompt count, id, metadata and
message but misses asserting the InterruptIfAgentWaiting flag; update the
synthetic prompt assertions (where you call sessions.promptCount(),
sessions.promptCall(0) and inspect prompt.opts) to add an explicit check that
prompt.opts.InterruptIfAgentWaiting == true (or use a boolean assertion) so the
InterruptIfAgentWaiting bit cannot be dropped; apply the same addition to the
other similar prompt assertion sites that inspect prompt.opts (the other
promptCall checks in this file).
In `@internal/daemon/coordinator_runtime.go`:
- Around line 388-409: The drain goroutine started by coordinatorRuntime calling
drainCoordinatorPromptEvents can leak because it blocks on range events forever;
change drainCoordinatorPromptEvents to accept a context (e.g., ctx
context.Context) and inside the loop use a select to listen for ctx.Done() in
addition to receiving from the events channel so the goroutine exits when the
runtime shuts down, and update the caller in coordinatorRuntime (where
PromptSynthetic is used) to pass the runtime shutdown context (or a managed
worker context) so each wake's drain is bounded to the runtime lifecycle;
alternatively register the goroutine with the runtime's worker manager or a
WaitGroup so it is tracked and cancelled on shutdown.
In `@internal/daemon/heartbeat_wake_runtime_test.go`:
- Around line 318-389: The test currently only asserts
dispatchHeartbeatWake(...) returned false; to exercise the full fallback, after
the handled == false check call the bridge's dispatchWake(...) (the terminal
path that issues the direct PromptSynthetic fallback) with the same
harnessSyntheticWake payload, then assert sessions.syntheticPromptCount() == 1
and re-query db.ListHeartbeatWakeEvents(...) to verify the wake event reflects
the direct reentry (check Result/Reason or new event indicating the fallback was
sent); this ensures dispatchHeartbeatWake + dispatchWake end-to-end fallback
behavior is validated rather than only the helper return value.
In `@internal/session/manager_test.go`:
- Around line 2411-2493: The test
TestPromptSyntheticInterruptsAgentWaitingTurnWhenRequested must be converted to
use the required t.Run("Should ...") subtest pattern: wrap the existing test
body inside t.Run("Should interrupt agent waiting when synthetic prompt requests
it", func(t *testing.T) { ... }), move t.Parallel() into the subtest, and ensure
all uses of t (Cleanup, Fatalf, etc.) remain using the subtest's t; keep the
existing references like h.driver.cancelHook, h.driver.promptHook,
manager.Prompt, manager.PromptSynthetic, collectEvents and managerPromptCalls
unchanged inside that subtest.
- Around line 2416-2418: The cleanup currently ignores the error returned by
h.manager.Stop(testutil.Context(t), session.ID); change the t.Cleanup closure to
handle that error instead of assigning to `_` — call h.manager.Stop with the
same args and if it returns a non-nil error report it via t.Fatalf or t.Errorf
(e.g., t.Fatalf("failed to stop session %s: %v", session.ID, err)) so the test
doesn't silently swallow Stop errors; keep the call inside the existing
t.Cleanup and preserve use of testutil.Context(t), h.manager.Stop, and
session.ID.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 05468ba9-59f6-45aa-96c0-84dfa6a85814
⛔ Files ignored due to path filters (1)
packages/site/content/runtime/core/configuration/config-toml.mdxis excluded by!**/*.mdx
📒 Files selected for processing (12)
internal/api/core/interfaces.gointernal/api/core/session_manager_stub_test.gointernal/api/testutil/session_stub.gointernal/daemon/coordinator_runtime.gointernal/daemon/coordinator_runtime_test.gointernal/daemon/harness_reentry_bridge.gointernal/daemon/heartbeat_wake_runtime_test.gointernal/session/manager_test.gointernal/session/session.gointernal/session/synthetic_prompt.gointernal/task/force_ops.gointernal/task/manager_test.go
f04c843 to
c4e924c
Compare
pedronauck
left a comment
There was a problem hiding this comment.
Automated review — self + Codex (gpt-5.5, xhigh)
Two independent reviews of PR #240 (fix: wake coordinator sessions reliably, commit c4e924c5). The fix wakes an existing detached/waiting coordinator via a synthetic prompt on the enqueue/retry/recovery paths and makes the heartbeat-wake path fall back to a direct synthetic prompt instead of dropping the run. No schema change, no raw claim_token leakage (claim_token_hash-only), %w + errors.Is/As intact, authoritative-primitive exclusivity preserved (the wake instructs the coordinator to run agh task next; it never calls ClaimNextRun).
- Self verdict: FIX_BEFORE_SHIP (2 risks)
- Codex verdict: FIX_BEFORE_SHIP (2 blockers, 4 risks, 0 nits)
Both reviews agree the change is functionally correct in the common case; the lock-hold and unbounded-interrupt paths should be fixed before relying on this under contention.
Blockers
- [codex B-1]
internal/daemon/coordinator_runtime.go:298,308,351—bootstrapRunholdscoordinatorRuntime.muacrosspromptCoordinator, whoseInterruptIfAgentWaitingpath can block up toTimeoutCancelGrace(~30s) inside a synchronous enqueue observer, serializing all coordinator enqueue/retry/recovery bootstrap behind one slow/stuck session. Fix: resolve the target session underr.mu, then release the lock before the wake (or wake on an owned goroutine afterUnlock). - [codex B-2]
internal/session/synthetic_prompt.go:90—interruptAndSubmitSyntheticPromptcallsCancelPromptundercontext.WithoutCancelwith no timeout beforewaitForPromptIdle, so a stuck cancel/tool-interrupt can block the wake indefinitely. Fix: wrap CancelPrompt + interrupt + wait in a bounded context.
Risks
- [self+codex R-1]
coordinator_runtime.go:388—go drainCoordinatorPromptEvents(...)is an untracked goroutine, not joined by any daemon WaitGroup at shutdown. Fix: track via a runtime-owned WaitGroup / lifecycle context. - [codex R-2]
heartbeat_wake_runtime_test.go:318— fallback regression coversWakeReasonSessionPromptActivebut notWakeReasonSessionPromptRace. Fix: add a siblingSessionPromptRacecase. - [codex R-3]
manager_test.go:4874— force-op enqueue-hook assertions check onlypayload.RunID, not the fullTaskRunContext. Fix: assert the fullTaskRunEnqueuedPayloadfor retry and recover. - [codex R-4] concurrency test asserts singleton creation but not the synthetic-prompt count/metadata. Fix: assert exactly one prompt with expected metadata.
Codex ran once via compozy exec --ide codex --model gpt-5.5 --reasoning-effort xhigh (exit 0, schema-valid findings). No source files modified.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
internal/daemon/coordinator_runtime.go (1)
316-356: ⚖️ Poor tradeoffConsider reducing the lock scope around coordinator creation.
The mutex
r.muis held from line 316 through the call tocreateCoordinatorSession(line 338), which performs hook dispatch and session creation. While there's no obvious deadlock risk (no re-entrant lock acquisition detected), holding the lock during external operations like hook dispatch can cause unnecessary lock contention.Consider releasing the lock before calling
createCoordinatorSession, then re-acquiring it only to updatewakeInFlight:r.mu.Lock() existing, err := r.activeCoordinator(ctx, decision.WorkspaceID) + r.mu.Unlock() if err != nil { - r.mu.Unlock() r.dispatchFailed(ctx, decision, reason, err) return nil, false, err } if existing != nil { + r.mu.Lock() shouldPrompt := r.beginCoordinatorWakeLocked(existing, decision) r.mu.Unlock() ... } info, createdCfg, created, err := r.createCoordinatorSession(ctx, decision, cfg, reason) if err != nil { - r.mu.Unlock() return nil, false, err } if !created { - r.mu.Unlock() return nil, false, nil } + r.mu.Lock() shouldPrompt := r.beginCoordinatorWakeLocked(info, decision) r.mu.Unlock()This limits lock hold time to only the critical sections that touch shared state.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@internal/daemon/coordinator_runtime.go` around lines 316 - 356, The code holds r.mu across the potentially long call to createCoordinatorSession; release the lock before invoking createCoordinatorSession and then re-acquire it to reconcile shared state and update wakeInFlight so we only hold the mutex for the critical section. Concretely: call r.activeCoordinator(...) while holding r.mu and if nil, unlock and call r.createCoordinatorSession(ctx, decision, cfg, reason) outside the lock; after that, re-lock r.mu, re-check r.activeCoordinator(ctx, decision.WorkspaceID) to handle a race (if an existing coordinator appeared, use beginCoordinatorWakeLocked(existing, decision), dispatchDecision/dispatchFailed as before and drop the created session if needed), otherwise update wakeInFlight with the newly created info and proceed to call beginCoordinatorWakeLocked(info, decision), unlock, then run promptCoordinator/finishCoordinatorWake and dispatch as the original flow intended. Ensure error paths still release the lock and dispatchFailed consistently.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@internal/session/manager_test.go`:
- Around line 2498-2577: The test "Should bound cancel when interrupting agent
waiting for synthetic prompt" never asserts that Cancel was invoked; after
calling h.manager.PromptSynthetic and collecting events from userEventsCh and
syntheticEventsCh, add a concrete assertion that the fake driver's cancel path
ran (e.g., assert h.driver.cancelCalls == 1 or >0). Locate the fake driver on
the test harness (h.driver) and the test block around the Prompt/PromptSynthetic
calls to insert a small retry/wait loop (with timeout) that checks
h.driver.cancelCalls to avoid flakiness, then fail the test if the counter is
not incremented.
---
Nitpick comments:
In `@internal/daemon/coordinator_runtime.go`:
- Around line 316-356: The code holds r.mu across the potentially long call to
createCoordinatorSession; release the lock before invoking
createCoordinatorSession and then re-acquire it to reconcile shared state and
update wakeInFlight so we only hold the mutex for the critical section.
Concretely: call r.activeCoordinator(...) while holding r.mu and if nil, unlock
and call r.createCoordinatorSession(ctx, decision, cfg, reason) outside the
lock; after that, re-lock r.mu, re-check r.activeCoordinator(ctx,
decision.WorkspaceID) to handle a race (if an existing coordinator appeared, use
beginCoordinatorWakeLocked(existing, decision), dispatchDecision/dispatchFailed
as before and drop the created session if needed), otherwise update wakeInFlight
with the newly created info and proceed to call beginCoordinatorWakeLocked(info,
decision), unlock, then run promptCoordinator/finishCoordinatorWake and dispatch
as the original flow intended. Ensure error paths still release the lock and
dispatchFailed consistently.
🪄 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: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 7b8fda2e-aabd-40eb-9afd-dae8845b53cb
📒 Files selected for processing (10)
internal/daemon/boot.gointernal/daemon/coordinator_runtime.gointernal/daemon/coordinator_runtime_integration_test.gointernal/daemon/coordinator_runtime_test.gointernal/daemon/daemon.gointernal/daemon/heartbeat_wake_runtime_test.gointernal/session/manager_test.gointernal/session/synthetic_prompt.gointernal/task/manager_test.gointernal/testutil/e2e/runtime_harness_lifecycle_test.go
✅ Files skipped from review due to trivial changes (1)
- internal/testutil/e2e/runtime_harness_lifecycle_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
- internal/session/synthetic_prompt.go
- internal/daemon/coordinator_runtime_test.go
What this changes
This PR makes task orchestration continue without manual intervention after a worker run changes state. When a task run is enqueued, retried, or recovered, AGH now wakes the detached/waiting coordinator session so it can route the next step immediately.
Why
In multi-agent task execution, a worker can finish and enqueue the next transition while the coordinator is parked. Without an explicit wake signal, the coordinator may sit idle until someone manually interrupts or prompts it. That made normal child -> receipt -> coordinator continuation unreliable.
Implementation notes
Reviewer focus
Validation
PATH="$HOME/.bun/bin:$PATH" make verifySummary by CodeRabbit
Release Notes
New Features
Bug Fixes
Tests