Skip to content
This repository was archived by the owner on Aug 10, 2026. It is now read-only.

fix: wake coordinator sessions reliably - #240

Merged
pedronauck merged 3 commits into
compozy:mainfrom
richardfogaca:fix/session-continuation
May 31, 2026
Merged

pedronauck merged 3 commits into
compozy:mainfrom
richardfogaca:fix/session-continuation

Conversation

@richardfogaca

@richardfogaca richardfogaca commented May 30, 2026 •

Copy link
Copy Markdown
Contributor

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

  • Adds synthetic prompt support for waking waiting agent sessions.
  • Wakes coordinator sessions from enqueue/retry/recovery paths.
  • Preserves queued prompt semantics, including interrupt behavior for sessions already waiting on input.
  • Adds regression coverage around coordinator wake and heartbeat behavior.
  • Includes the existing MDX formatting cleanup required by repo verification.

Reviewer focus

  • Confirm the wake path only nudges existing coordinator sessions and does not create duplicate coordinators.
  • Confirm queued prompt/interrupt behavior is still conservative for active sessions.

Validation

  • PATH="$HOME/.bun/bin:$PATH" make verify

Summary by CodeRabbit

Release Notes

  • New Features

    • Added synthetic prompt functionality to enable interrupting agent-waiting sessions.
    • Coordinator runtime now supports wake capabilities with deduplication and improved session management.
  • Bug Fixes

    • Force retry and recovery operations now properly dispatch enqueue hooks.
    • Improved heartbeat wake handling for edge cases.
  • Tests

    • Enhanced test coverage for synthetic prompting and coordinator runtime scenarios.

Copilot AI review requested due to automatic review settings May 30, 2026 01:11
@vercel

vercel Bot commented May 30, 2026

Copy link
Copy Markdown

@richardfogaca is attempting to deploy a commit to the Compozy Team on Vercel.

A member of the Team first needs to authorize it.

@coderabbitai

coderabbitai Bot commented May 30, 2026 •

Copy link
Copy Markdown

Review Change Stack

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info
⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

Run ID: 557e6c47-1c1c-443e-8443-383fe3afe412

📥 Commits

Reviewing files that changed from the base of the PR and between 8bcff10 and 565e724.

📒 Files selected for processing (2)
  • internal/daemon/coordinator_runtime.go
  • internal/daemon/coordinator_runtime_test.go
🚧 Files skipped from review as they are similar to previous changes (2)
  • internal/daemon/coordinator_runtime_test.go
  • internal/daemon/coordinator_runtime.go

Walkthrough

This 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.

Changes

Synthetic Prompt Interruption Feature

Layer / File(s) Summary
SessionManager interface and test stubs
internal/api/core/interfaces.go, internal/api/core/session_manager_stub_test.go, internal/api/testutil/session_stub.go
Adds PromptSynthetic method to SessionManager; implements method in test stubs with configurable hooks for synthetic prompt behavior simulation.
Session manager interrupt implementation and helpers
internal/session/synthetic_prompt.go, internal/session/session.go
Introduces InterruptIfAgentWaiting field to SyntheticPromptOpts; implements interrupt-then-submit logic that cancels waiting prompts and retries submission; adds agent-waiting state detector and error classification helpers.
Session interrupt tests and context-aware cancel
internal/session/manager_test.go
Adds test coverage for interrupt flow with agent-waiting liveness; extends fakeDriver with context-aware cancel hook to validate bounded cancellation behavior.
Coordinator runtime lifecycle and structure
internal/daemon/coordinator_runtime.go (setup portion)
Extends coordinatorSessionManager with PromptSynthetic method; adds lifecycle context/cancel, wake deduplication, and goroutine tracking to coordinatorRuntime; updates constructor to require daemon context; registers shutdown callback in bootCoordinator.
Coordinator bootstrap rework and prompting
internal/daemon/coordinator_runtime.go (bootstrap and prompting)
Refactors bootstrapRun to detect and optionally prompt existing coordinators; updates createCoordinatorSession to return config alongside created flag; implements promptCoordinator with wake message construction and async event draining; adds dedupe and shutdown helpers.
Coordinator runtime tests and test infrastructure
internal/daemon/coordinator_runtime_test.go
Extends bootstrap test to assert PromptSynthetic calls and metadata; adds concurrent enqueue/coalescing test with prompt sync and assertions; introduces TestCoordinatorRuntimePromptsExistingCoordinatorForQueuedRun; substantially expands test double with prompt recording, sync gating, and accessor helpers.
Harness reentry and heartbeat wake integration
internal/daemon/harness_reentry_bridge.go, internal/daemon/heartbeat_wake_runtime_test.go
Sets InterruptIfAgentWaiting on harness reentry synthetic prompts; refines heartbeat wake skip handling for active-prompt/race scenarios; adds integration test validating skip persistence and direct reentry metadata.
Daemon composition and shutdown wiring
internal/daemon/daemon.go, internal/daemon/boot.go, internal/daemon/coordinator_runtime_integration_test.go
Adds coordinator runtime to Daemon struct and shutdown targets; includes population during detach, reset on state clear, and shutdown sequence; copies boot coordinator into daemon on publishBootState; updates test call site to pass context.
Task retry/recover enqueue hooks and tests
internal/task/force_ops.go, internal/task/manager_test.go
Ensures RetryRun and RecoverRun call dispatchTaskRunEnqueued after child run creation; extends tests with TaskRunEnqueued payload capture and validation helper asserting single enqueue event per operation.

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
Loading
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
Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

  • compozy/agh#7: Introduces and refactors SessionManager interface in internal/api/core/interfaces.go, which this PR extends with the PromptSynthetic method.
🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name Status Explanation Resolution
Docstring Coverage ⚠️ Warning Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%. Write docstrings for the functions missing them to satisfy the coverage threshold.
✅ Passed checks (4 passed)
Check name Status Explanation
Description Check ✅ Passed Check skipped - CodeRabbit’s high-level summary is enabled.
Title check ✅ Passed The title 'fix: wake coordinator sessions reliably' directly describes the main objective of the PR, which is to ensure coordinator sessions are woken reliably when task runs are enqueued, retried, or recovered.
Linked Issues check ✅ Passed Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check ✅ Passed Check skipped because no linked issues were found for this pull request.

✏️ Tip: You can configure your own custom pre-merge checks in the settings.

✨ Finishing Touches
🧪 Generate unit tests (beta)
  • Create PR with unit tests

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.

❤️ Share

Comment @coderabbitai help to get the list of available commands and usage tips.

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

📥 Commits

Reviewing files that changed from the base of the PR and between cd4b155 and f04c843.

⛔ Files ignored due to path filters (1)
  • packages/site/content/runtime/core/configuration/config-toml.mdx is excluded by !**/*.mdx
📒 Files selected for processing (12)
  • internal/api/core/interfaces.go
  • internal/api/core/session_manager_stub_test.go
  • internal/api/testutil/session_stub.go
  • internal/daemon/coordinator_runtime.go
  • internal/daemon/coordinator_runtime_test.go
  • internal/daemon/harness_reentry_bridge.go
  • internal/daemon/heartbeat_wake_runtime_test.go
  • internal/session/manager_test.go
  • internal/session/session.go
  • internal/session/synthetic_prompt.go
  • internal/task/force_ops.go
  • internal/task/manager_test.go

Comment thread internal/daemon/coordinator_runtime_test.go
Comment thread internal/daemon/coordinator_runtime.go Outdated
Comment thread internal/daemon/heartbeat_wake_runtime_test.go Outdated
Comment thread internal/session/manager_test.go
Comment thread internal/session/manager_test.go Outdated
@richardfogaca
richardfogaca force-pushed the fix/session-continuation branch from f04c843 to c4e924c Compare May 30, 2026 01:41

@pedronauck pedronauck left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 — bootstrapRun holds coordinatorRuntime.mu across promptCoordinator, whose InterruptIfAgentWaiting path can block up to TimeoutCancelGrace (~30s) inside a synchronous enqueue observer, serializing all coordinator enqueue/retry/recovery bootstrap behind one slow/stuck session. Fix: resolve the target session under r.mu, then release the lock before the wake (or wake on an owned goroutine after Unlock).
  • [codex B-2] internal/session/synthetic_prompt.go:90 — interruptAndSubmitSyntheticPrompt calls CancelPrompt under context.WithoutCancel with no timeout before waitForPromptIdle, 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 covers WakeReasonSessionPromptActive but not WakeReasonSessionPromptRace. Fix: add a sibling SessionPromptRace case.
  • [codex R-3] manager_test.go:4874 — force-op enqueue-hook assertions check only payload.RunID, not the full TaskRunContext. Fix: assert the full TaskRunEnqueuedPayload for 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.

Comment thread internal/daemon/coordinator_runtime.go Outdated

@coderabbitai coderabbitai Bot left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actionable comments posted: 1

🧹 Nitpick comments (1)
internal/daemon/coordinator_runtime.go (1)

316-356: ⚖️ Poor tradeoff

Consider reducing the lock scope around coordinator creation.

The mutex r.mu is held from line 316 through the call to createCoordinatorSession (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 update wakeInFlight:

 	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

📥 Commits

Reviewing files that changed from the base of the PR and between c4e924c and 8bcff10.

📒 Files selected for processing (10)
  • internal/daemon/boot.go
  • internal/daemon/coordinator_runtime.go
  • internal/daemon/coordinator_runtime_integration_test.go
  • internal/daemon/coordinator_runtime_test.go
  • internal/daemon/daemon.go
  • internal/daemon/heartbeat_wake_runtime_test.go
  • internal/session/manager_test.go
  • internal/session/synthetic_prompt.go
  • internal/task/manager_test.go
  • internal/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

Comment thread internal/session/manager_test.go
@pedronauck
pedronauck merged commit 46b5fbb into compozy:main May 31, 2026
7 of 9 checks passed
This was referenced Jun 2, 2026
Sign up for free to subscribe to this conversation on GitHub. Already have an account? Sign in.

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants