From 0b2c59a2ddbc11c48b5d87cd80d967ef79460d7e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Thu, 21 May 2026 01:37:03 +0100 Subject: [PATCH] fix(lifecycle): publish AgentReady on wakeup-driven turns (#875) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(lifecycle): publish AgentReady on wakeup-driven turns Wakeup turns flow Adapter.Prompt → bridge → events, bypassing SessionManager.SendPrompt. The execution status stayed Ready across the wakeup turn, so MarkReady's duplicate-suppression guard dropped the second AgentReady — the orchestrator never closed the turn, never evaluated on_turn_complete, and never dispatched queued messages. Detect this in recordActivity and flip Ready → Running on the first non-terminal event, re-emitting AgentRunning so the rest of the pipeline behaves like any other turn. * fix(lifecycle): gate wakeup flip on turn-content events and handle empty turns Three follow-ups on the wakeup AgentReady fix: 1. Restrict the Ready → Running flip in `recordActivity` to events that unambiguously indicate turn-in-progress activity (message_chunk, reasoning, tool_call/update, plan, agent_plan, permission_request). The original whole-event-type flip caused a regression: claude-agent-acp emits available_commands_update asynchronously ~50ms after session/new, well after dispatchInitialPrompt fires MarkBootReady for a no-prompt task. The Ready → Running flip on that delayed metadata event was putting newly-created tasks back into Running and breaking the chat- input UI (changes-panel-focus E2E tests). 2. Handle the empty-wakeup edge case: a wakeup turn that emits only a `complete` event (empty model response) is still excluded from the flip whitelist, so the Ready guard in MarkReady would suppress AgentReady. `handleCompleteEventMarkState` now publishes AgentReady directly when Status is already Ready on a successful complete — the orchestrator still closes the turn for empty wakeup responses. 3. Fold the `executionStore != nil` check into the early-return so the AgentRunning publish only fires when the status flip actually succeeded. Test-file cleanup from review feedback: dead loop, redundant `var _ = io.MultiReader`, and `teeReader.src any` → `io.Reader`. New regression tests: - TestWakeup_PostBootMetadataEventsDoNotFlipToRunning — locks in the whitelist behavior so future events can't accidentally re-arm a boot-ready session. - TestWakeup_EmptyTurnStillPublishesAgentReady — covers the empty-wakeup-complete fallback. * fix(lifecycle): publish AgentRunning before AgentReady on empty wakeup turns Follow-up on the previous wakeup fixes — addresses three issues raised in review: 1. **Empty-wakeup AgentReady was ignored by the orchestrator.** The fallback in `handleCompleteEventMarkState` published AgentReady when Status was already Ready, but `handleAgentReady` at the orchestrator early-returns when `session.State != Running/Starting`. After the previous turn ended the session is in WaitingForInput, so the AgentReady got silently dropped — same suppression the wakeup fix was meant to prevent. Now the fallback flips Ready → Running + publishes AgentRunning first (mirroring `recordActivity` for the non-empty case), then falls through to MarkReady. 2. **firstActivityOnce could still publish AgentRunning on a late metadata event.** Gate it on `Status != Ready` so a delayed `available_commands_update` arriving after MarkBootReady (the original regression scenario) can't slip through via the once-per-execution path. In practice the adapter always emits `agent_capabilities` during Initialize before MarkBootReady, so firstActivityOnce still fires there with Status==Running — this is defensive hardening. 3. **Tighter test assertions.** The post-boot-metadata test now primes firstActivityOnce in setup and asserts strict `runningCount == 0`. The second-turn test verifies AgentRunning precedes each AgentReady in bus order so the orchestrator's session-state guard isn't violated. Same ordering check added to the empty-wakeup test. * test(lifecycle): rename exec to agentExec to avoid os/exec shadow The test variable `exec` in wakeup_e2e_test.go shadowed the imported `os/exec` package — harmless today because all `exec.Command`/`LookPath` calls happen before the local binding, but a future call added below would silently fail. Rename to `agentExec` to remove the shadow. * ci: retrigger CodeQL after transient Ruby analyze flake The default-setup CodeQL run on PR #875 produced a flaky 'CodeQL could not process any code written in Ruby' failure on the Ruby analyze job (github.com/kdlbs/kandev/actions/runs/26124163769). The repo has no Ruby code and main's CodeQL only analyzes javascript-typescript + actions; the Ruby job was added dynamically and failed on the empty diff. Sibling run 26124163815 on the same SHA passed. Cannot rerun default-setup workflows via the API (403 'cannot be retried'), so push an empty commit to retrigger the analysis. * test(lifecycle): tighten wakeup test assertions - wakeup_simulated_test.go: assert exact readyCount==2 instead of <2 so any over-emission (e.g. duplicate AgentReady on the wakeup turn) is also caught. - wakeup_e2e_test.go: fail with t.Fatalf when the adapter surfaces no WAKEUP text instead of logging a warning. Otherwise the bus-level assertions below pass vacuously if the wakeup never fires. Both fixes address review comments from cubic-dev-ai / claude[bot] on PR #875. Co-Authored-By: Claude Opus 4.7 * test(lifecycle): remove unsynchronised SessionID write in forwarder goroutine The forwarding goroutine wrote to agentExec.SessionID without holding the executionStore mutex, racing with the main goroutine's authoritative assignment after NewSession returns. The forwarder's write is redundant — line 264 sets the canonical value — and benign in practice, but `go test -race` would flag it. Drop the write to keep the race detector clean. Addresses review comment from claude[bot] on PR #875. Co-Authored-By: Claude Opus 4.7 --------- Co-authored-by: Kandev Agent Co-authored-by: Claude Opus 4.7 --- .../agent/runtime/lifecycle/manager_events.go | 88 +++- .../runtime/lifecycle/wakeup_e2e_test.go | 404 ++++++++++++++++++ .../lifecycle/wakeup_simulated_test.go | 239 +++++++++++ .../adapter/transport/acp/wakeup_e2e_test.go | 324 ++++++++++++++ 4 files changed, 1050 insertions(+), 5 deletions(-) create mode 100644 apps/backend/internal/agent/runtime/lifecycle/wakeup_e2e_test.go create mode 100644 apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/acp/wakeup_e2e_test.go diff --git a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go index c6280e96f5..00aa3374cb 100644 --- a/apps/backend/internal/agent/runtime/lifecycle/manager_events.go +++ b/apps/backend/internal/agent/runtime/lifecycle/manager_events.go @@ -105,6 +105,29 @@ func (m *Manager) handleCompleteEventMarkState(execution *AgentExecution, event } return } + // Empty wakeup turn fallback: if the wakeup-driven turn produced no + // turn-content events (no message_chunk/tool_call/etc), recordActivity + // won't have flipped Ready → Running. MarkReady would then early-return + // on the Ready guard and silently drop AgentReady — same suppression + // the wakeup fix is designed to prevent. + // + // Publishing AgentReady alone is not enough: the orchestrator's + // handleAgentReady (`event_handlers_agent.go:205`) ignores AgentReady + // when session.State is not Running/Starting, and after the previous + // turn ended the session is WaitingForInput. We mirror what + // recordActivity does for the non-empty case — flip Ready → Running + // and publish AgentRunning first, so the orchestrator's session state + // catches up — then let MarkReady do its normal Running → Ready + // transition and publish AgentReady. + if execution.Status == v1.AgentStatusReady { + m.logger.Info("flipping Ready→Running for empty wakeup turn before publishing AgentReady", + zap.String("execution_id", execution.ID), + zap.String("session_id", execution.SessionID)) + if m.executionStore != nil { + m.executionStore.UpdateStatus(execution.ID, v1.AgentStatusRunning) + } + m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution) + } if err := m.MarkReady(execution.ID); err != nil { m.logger.Error("failed to mark execution as ready after complete", zap.String("execution_id", execution.ID), @@ -287,16 +310,71 @@ func (m *Manager) handleAvailableCommandsEvent(execution *AgentExecution, event m.eventPublisher.PublishAvailableCommands(execution, event.AvailableCommands) } +// turnContentEventTypes is the set of agent event types that unambiguously +// signal a real turn is in progress — assistant text, reasoning, tool work, +// plan/permission updates. These are the only events that drive the +// Ready → Running flip for wakeup-driven turns; boot/metadata events +// (agent_capabilities, available_commands, session_mode/models/status, +// context_window, etc.) can arrive *after* MarkBootReady has put the +// execution into Ready (e.g. claude-agent-acp emits available_commands +// asynchronously ~50ms after session/new, well after dispatchInitialPrompt +// has fired MarkBootReady for a no-prompt task) and must NOT be treated +// as turn starts. Terminal events (complete/error) are excluded too — +// they own their own status transitions via handleCompleteEvent and a +// dedicated empty-turn fallback in handleCompleteEventMarkState. +var turnContentEventTypes = map[string]struct{}{ + "message_chunk": {}, + "reasoning": {}, + "tool_call": {}, + "tool_update": {}, + "plan": {}, + "agent_plan": {}, + "permission_request": {}, +} + // recordActivity updates the last-activity timestamp and, on the very first // event from an execution, publishes AgentRunning to transition STARTING → RUNNING. -func (m *Manager) recordActivity(execution *AgentExecution) { +// +// Wakeup-driven turns also flip Ready → Running here. The adapter's wakeup +// scheduler fires synthetic prompts directly via Adapter.Prompt, bypassing +// SessionManager.SendPrompt — so nothing on the lifecycle side flips the +// execution back to Running for those turns. Without that flip, MarkReady's +// duplicate-suppression guard (manager_interaction.go:896 — early-returns +// when execution.Status is already Ready) silently drops the wakeup turn's +// AgentReady event, the orchestrator never calls completeTurnForSession, +// and workflow on_turn_complete + queued-message dispatch silently break. +// +// The flip is gated on `turnContentEventTypes` so post-boot metadata events +// (available_commands_update arriving 50ms after MarkBootReady, etc.) don't +// accidentally re-arm a freshly-booted no-prompt session as Running. +func (m *Manager) recordActivity(execution *AgentExecution, event agentctl.AgentEvent) { execution.lastActivityAtMu.Lock() execution.lastActivityAt = time.Now() execution.lastActivityAtMu.Unlock() - execution.firstActivityOnce.Do(func() { - m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution) - }) + // Gate firstActivityOnce on `Status != Ready` so a delayed metadata + // event arriving after MarkBootReady can't accidentally fire + // AgentRunning. In practice the adapter always emits agent_capabilities + // during Initialize (before MarkBootReady), so firstActivityOnce fires + // while Status is still Running — this is defensive hardening. + if execution.Status != v1.AgentStatusReady { + execution.firstActivityOnce.Do(func() { + m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution) + }) + return + } + if m.executionStore == nil { + return + } + if _, ok := turnContentEventTypes[event.Type]; !ok { + return + } + m.executionStore.UpdateStatus(execution.ID, v1.AgentStatusRunning) + m.logger.Info("wakeup-driven turn detected; flipping execution back to Running", + zap.String("execution_id", execution.ID), + zap.String("session_id", execution.SessionID), + zap.String("trigger_event_type", event.Type)) + m.eventPublisher.PublishAgentEvent(context.Background(), events.AgentRunning, execution) } // handleStreamDisconnect handles unexpected updates stream disconnections. @@ -320,7 +398,7 @@ func (m *Manager) handleStreamDisconnect(execution *AgentExecution, err error) { // handleAgentEvent processes incoming agent events from the agent func (m *Manager) handleAgentEvent(execution *AgentExecution, event agentctl.AgentEvent) { - m.recordActivity(execution) + m.recordActivity(execution, event) m.logger.Debug("handleAgentEvent entry", zap.String("execution_id", execution.ID), diff --git a/apps/backend/internal/agent/runtime/lifecycle/wakeup_e2e_test.go b/apps/backend/internal/agent/runtime/lifecycle/wakeup_e2e_test.go new file mode 100644 index 0000000000..ea640eefe7 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/wakeup_e2e_test.go @@ -0,0 +1,404 @@ +//go:build wakeup_e2e + +// Full-pipeline isolation test: bridge → ACP Adapter → Lifecycle Manager → +// event bus. Verifies that ScheduleWakeup-driven wakeup turns reach the same +// event-bus consumers (i.e. the orchestrator) that user-initiated turns do. +// +// This is a layer up from acp/wakeup_e2e_test.go — that test stopped at the +// adapter's Updates() channel. This one wires the adapter's events through +// the real Lifecycle Manager's handleAgentEvent path and captures everything +// published to the event bus. +// +// Build/run: +// +// go test -tags wakeup_e2e -v -timeout 5m \ +// ./internal/agent/runtime/lifecycle/ -run WakeupE2EFullPipeline +// +// Skips automatically when `npx` is not on PATH or +// ~/.claude/.credentials.json is missing. + +package lifecycle + +import ( + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/acp" + "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared" + "github.com/kandev/kandev/internal/agentctl/types/streams" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +const ( + wakeupE2EDelay = 60 // ScheduleWakeup clamps to [60, 3600] + wakeupE2EBuffer = 30 * time.Second // grace after the timer should fire +) + +// trackingBus captures every event published, with a timestamp. +type trackingBus struct { + mu sync.Mutex + events []trackedBusEvent +} + +type trackedBusEvent struct { + At time.Time + Subject string + Event *bus.Event +} + +func (b *trackingBus) Publish(_ context.Context, subject string, event *bus.Event) error { + b.mu.Lock() + defer b.mu.Unlock() + b.events = append(b.events, trackedBusEvent{At: time.Now(), Subject: subject, Event: event}) + return nil +} + +func (b *trackingBus) Subscribe(string, bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} +func (b *trackingBus) QueueSubscribe(string, string, bus.EventHandler) (bus.Subscription, error) { + return nil, nil +} +func (b *trackingBus) Request(context.Context, string, *bus.Event, time.Duration) (*bus.Event, error) { + return nil, nil +} +func (b *trackingBus) Close() {} +func (b *trackingBus) IsConnected() bool { return true } + +func (b *trackingBus) snapshot() []trackedBusEvent { + b.mu.Lock() + defer b.mu.Unlock() + out := make([]trackedBusEvent, len(b.events)) + copy(out, b.events) + return out +} + +// teeReader passes bytes through while echoing complete lines to stderr. +type teeReader struct { + src io.Reader + label string + buf []byte +} + +func (t *teeReader) Read(p []byte) (int, error) { + n, err := t.src.Read(p) + if n > 0 { + t.buf = append(t.buf, p[:n]...) + for { + i := -1 + for j, v := range t.buf { + if v == '\n' { + i = j + break + } + } + if i < 0 { + break + } + line := strings.TrimRight(string(t.buf[:i]), "\r") + if len(line) > 240 { + line = line[:240] + "…" + } + if line != "" { + fmt.Fprintf(os.Stderr, "[%s] %s\n", t.label, line) + } + t.buf = t.buf[i+1:] + } + } + return n, err +} + +// TestWakeupE2EFullPipeline_BridgeToEventBus drives the full stack: +// +// @agentclientprotocol/claude-agent-acp (real bridge process) +// ↑↓ stdio +// acp.Adapter (real) +// ↓ Updates() channel +// [test glue: forwards each event to mgr.handleAgentEvent for the execution] +// ↓ +// lifecycle.Manager (real, with tracking event bus) +// ↓ +// trackingBus (captures every Publish) +// +// It sends a prompt that triggers ScheduleWakeup with delaySeconds=60, waits +// for the wakeup to fire, and asserts that wakeup-driven events reach the +// event bus — i.e. they'd reach the orchestrator. If this assertion fails, +// the bug lives in the lifecycle layer (it does receive the events but +// doesn't publish them). +func TestWakeupE2EFullPipeline_BridgeToEventBus(t *testing.T) { + if _, err := exec.LookPath("npx"); err != nil { + t.Skipf("skipping: npx not on PATH (%v)", err) + } + home, _ := os.UserHomeDir() + credPath := filepath.Join(home, ".claude", ".credentials.json") + if _, err := os.Stat(credPath); err != nil { + t.Skipf("skipping: %s not found — Claude Code OAuth credentials required (%v)", credPath, err) + } + + workDir := t.TempDir() + + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: "debug", + Format: "console", + OutputPath: "stderr", + }) + if err != nil { + t.Fatalf("logger: %v", err) + } + + // ---- Set up adapter and bridge ---- + cfg := &shared.Config{AgentID: "wakeup-fullpipe-e2e", WorkDir: workDir} + adp := acp.NewAdapter(cfg, log) + + cmd := exec.Command("npx", "-y", "@agentclientprotocol/claude-agent-acp@0.31.1") + cmd.Dir = workDir + cmd.Stderr = os.Stderr + stdinPipe, err := cmd.StdinPipe() + if err != nil { + t.Fatalf("StdinPipe: %v", err) + } + stdoutPipe, err := cmd.StdoutPipe() + if err != nil { + t.Fatalf("StdoutPipe: %v", err) + } + if err := cmd.Start(); err != nil { + t.Fatalf("start bridge: %v", err) + } + + if err := adp.Connect(stdinPipe, &teeReader{src: stdoutPipe, label: "bridge→adapter"}); err != nil { + _ = cmd.Process.Kill() + t.Fatalf("Connect: %v", err) + } + + t.Cleanup(func() { + _ = adp.Close() + _ = stdinPipe.Close() + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + }) + + // ---- Set up lifecycle Manager with tracking bus ---- + mgr := newTestManager() + trackBus := &trackingBus{} + // Reach into the manager's event publisher: replace the bus with the + // tracker. eventPublisher is the only thing that calls eventBus.Publish. + // We need the manager's eventBus reset too so handleAgentEvent's callers + // flow through the tracker. + mgr.eventBus = trackBus + mgr.eventPublisher = NewEventPublisher(trackBus, log) + + // ---- Create the execution that ties the adapter session to the manager ---- + executionID := "wakeup-e2e-exec-1" + taskID := "wakeup-e2e-task-1" + agentExec := &AgentExecution{ + ID: executionID, + TaskID: taskID, + Status: v1.AgentStatusRunning, + StartedAt: time.Now(), + promptDoneCh: make(chan PromptCompletionSignal, 1), + historyEnabled: false, // skip history writes — we're only checking event flow + } + mgr.executionStore.Add(agentExec) + + // ---- Forward adapter events into the manager ---- + // This mirrors what StreamManager.connectUpdatesStream does in production + // (callback into mgr.handleAgentEvent), but in-process. + stopForward := make(chan struct{}) + var forwardWG sync.WaitGroup + forwardWG.Add(1) + adapterEvents := make([]streams.AgentEvent, 0, 128) + var adapterEventsMu sync.Mutex + go func() { + defer forwardWG.Done() + for { + select { + case ev, ok := <-adp.Updates(): + if !ok { + return + } + adapterEventsMu.Lock() + adapterEvents = append(adapterEvents, ev) + adapterEventsMu.Unlock() + fmt.Fprintf(os.Stderr, "[adapter→mgr] event type=%s sessionID=%s text=%q\n", + ev.Type, ev.SessionID, truncateE2E(ev.Text, 80)) + mgr.handleAgentEvent(agentExec, ev) + case <-stopForward: + return + } + } + }() + t.Cleanup(func() { + close(stopForward) + forwardWG.Wait() + }) + + // ---- Run the wakeup scenario ---- + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute) + defer cancel() + + t.Logf("Initialize…") + if err := adp.Initialize(ctx); err != nil { + t.Fatalf("Initialize: %v", err) + } + t.Logf("NewSession…") + sid, err := adp.NewSession(ctx, nil) + if err != nil { + t.Fatalf("NewSession: %v", err) + } + agentExec.SessionID = sid + t.Logf("session=%s", sid) + + prompt := fmt.Sprintf( + "Use the ScheduleWakeup tool exactly once with delaySeconds=%d, "+ + "prompt=\"continue probe\", reason=\"e2e full-pipeline test\". "+ + "Then stop. After waking up, just say \"WAKEUP_FIRED\" and stop.", + wakeupE2EDelay, + ) + + t.Logf("sending initial prompt…") + initialStart := time.Now() + if err := adp.Prompt(ctx, prompt, nil); err != nil { + t.Fatalf("initial Prompt: %v", err) + } + initialEnd := time.Now() + t.Logf("initial prompt completed in %s", initialEnd.Sub(initialStart)) + + preWakeupBusCount := len(trackBus.snapshot()) + t.Logf("bus events after initial prompt: %d", preWakeupBusCount) + + // Wait past the wakeup time. + waitFor := time.Duration(wakeupE2EDelay)*time.Second + wakeupE2EBuffer + t.Logf("waiting %s for wakeup to fire…", waitFor) + deadline := time.Now().Add(waitFor) + for time.Now().Before(deadline) { + time.Sleep(5 * time.Second) + busCount := len(trackBus.snapshot()) + newSinceWakeup := busCount - preWakeupBusCount + elapsed := time.Since(initialEnd).Truncate(time.Second) + t.Logf(" t+%s: bus events total=%d (post-prompt-end=%d)", elapsed, busCount, newSinceWakeup) + } + + final := trackBus.snapshot() + postWakeupEvents := final[preWakeupBusCount:] + t.Logf("post-prompt-end bus events: %d", len(postWakeupEvents)) + + // Classify each tracked bus event + type classified struct { + At time.Time + Subject string + Type string + Text string + Payload *AgentStreamEventPayload + } + var classifiedEvents []classified + for _, te := range postWakeupEvents { + c := classified{At: te.At, Subject: te.Subject} + if te.Event != nil { + c.Type = te.Event.Type + if payload, ok := te.Event.Data.(AgentStreamEventPayload); ok { + c.Payload = &payload + if payload.Data != nil { + c.Text = payload.Data.Text + } + } + } + classifiedEvents = append(classifiedEvents, c) + } + for i, c := range classifiedEvents { + t.Logf(" [%d] subject=%s type=%s text=%q", + i, c.Subject, c.Type, truncateE2E(c.Text, 80)) + } + + // Count adapter events for comparison (this is what reached the in-process boundary). + adapterEventsMu.Lock() + totalAdapterEvents := len(adapterEvents) + adapterEventsSnapshot := make([]streams.AgentEvent, len(adapterEvents)) + copy(adapterEventsSnapshot, adapterEvents) + adapterEventsMu.Unlock() + + t.Logf("adapter-side total events: %d", totalAdapterEvents) + + // Sanity: did the adapter actually see wakeup-driven text? It must, from + // the prior test. Find any post-prompt-end message_chunk events with text. + wakeupTexts := []string{} + for _, ev := range adapterEventsSnapshot { + if ev.Type == streams.EventTypeMessageChunk && strings.Contains(ev.Text, "WAKEUP") { + wakeupTexts = append(wakeupTexts, ev.Text) + } + } + if len(wakeupTexts) == 0 { + t.Fatalf("adapter did not surface 'WAKEUP' text — wakeup turn did not fire, bus-publishing assertions would be inconclusive") + } + t.Logf("✓ adapter surfaced wakeup text chunks: %v", wakeupTexts) + + // ---- Detailed accounting per turn ---- + // + // Each completed turn (initial + wakeup) should produce: + // - At least one agent.stream message_streaming event (the assistant text) + // - Exactly one agent.ready event (signals turn-end to the orchestrator) + // + // The orchestrator relies on agent.ready to call completeTurnForSession, + // evaluate on_turn_complete workflow transitions, and dispatch queued + // messages. If the wakeup turn doesn't publish agent.ready, the turn never + // closes — even though messages persisted, workflow state is stuck. + + // Count complete events FROM ALL bus events (not just post-prompt-end, since + // the first complete might have arrived just before our snapshot). + all := trackBus.snapshot() + agentReadyCount := 0 + streamingMsgsByText := []string{} + completeCount := 0 + for _, te := range all { + if te.Event == nil { + continue + } + if te.Event.Type == "agent.ready" { + agentReadyCount++ + } + if payload, ok := te.Event.Data.(AgentStreamEventPayload); ok && payload.Data != nil { + switch payload.Data.Type { + case "message_streaming": + streamingMsgsByText = append(streamingMsgsByText, payload.Data.Text) + case "complete": + completeCount++ + } + } + } + + t.Logf("=== full bus accounting ===") + t.Logf("agent.ready events: %d (expect 2 — one per turn)", agentReadyCount) + t.Logf("AgentStreamEventPayload type=complete events: %d (expect 2 — one per turn)", completeCount) + t.Logf("message_streaming text values: %v", streamingMsgsByText) + + // Adapter saw 2 complete events (initial + wakeup), confirmed by debug log. + // The lifecycle manager should publish 2 agent.ready events. + if completeCount >= 2 && agentReadyCount < 2 { + t.Errorf( + "BUG: adapter delivered %d complete events but only %d agent.ready event(s) reached the bus. "+ + "This means MarkReady's `if execution.Status == AgentStatusReady` guard at "+ + "manager_interaction.go:896 is suppressing the wakeup turn's AgentReady event "+ + "because the wakeup path (fireWakeup → a.Prompt) bypasses SendPrompt and never "+ + "flips the execution back to Running. The orchestrator therefore never sees the "+ + "wakeup turn end → completeTurnForSession is never called → workflow on_turn_complete "+ + "never fires → queued messages aren't dispatched.", + completeCount, agentReadyCount, + ) + } +} + +func truncateE2E(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} diff --git a/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go b/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go new file mode 100644 index 0000000000..fb46a30174 --- /dev/null +++ b/apps/backend/internal/agent/runtime/lifecycle/wakeup_simulated_test.go @@ -0,0 +1,239 @@ +package lifecycle + +import ( + "testing" + + agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl" + "github.com/kandev/kandev/internal/events" + v1 "github.com/kandev/kandev/pkg/api/v1" +) + +// TestWakeup_SecondTurnAgentReadyIsSuppressed reproduces the bug deterministically +// without needing the bridge or auth. +// +// Scenario: an execution receives events for two consecutive turns. The first +// turn is user-initiated (status flips Running → Ready on complete, agent.ready +// fires). The second turn is wakeup-initiated — kandev's wakeup scheduler calls +// adapter.Prompt directly, which does NOT flip the execution back to Running +// (that's done by SessionManager.SendPrompt, which fireWakeup bypasses). +// +// When the second `complete` event arrives, the manager calls MarkReady. But +// MarkReady at manager_interaction.go:896 has: +// +// if execution.Status == v1.AgentStatusReady { +// return nil +// } +// +// Since the execution is still Ready (never went back to Running), this early- +// returns, no agent.ready is published, and the orchestrator never sees the +// wakeup turn end → completeTurnForSession never fires → workflow on_turn_complete +// never evaluates. +// +// The wakeup-turn's message_streaming events DO still reach the bus (those +// don't gate on execution status), so the persisted chat history is correct. +// But workflow state and queued-message dispatch are broken. +func TestWakeup_SecondTurnAgentReadyIsSuppressed(t *testing.T) { + mgr, eventBus := createTestManagerWithTracking() + execution := createTestExecution("exec-1", "task-1", "session-1") + _ = mgr.executionStore.Add(execution) + + // === Turn 1: user-initiated prompt === + // + // In production: SessionManager.SendPrompt() runs and calls + // sm.executionStore.UpdateStatus(execution.ID, v1.AgentStatusRunning) + // We mirror that here. + mgr.executionStore.UpdateStatus(execution.ID, v1.AgentStatusRunning) + + // Assistant text + tool call + complete (the agent calls ScheduleWakeup). + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "message_chunk", Text: "Scheduled.\n", + }) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "complete"}) + + // === Turn 2: wakeup-initiated prompt === + // + // In production: the wakeupScheduler timer fires → adapter.fireWakeup() → + // adapter.Prompt(ctx, prompt, nil). The lifecycle layer is NOT involved + // in initiating this prompt, so executionStore.UpdateStatus(Running) is + // NEVER called. We mirror that by leaving the execution status as-is. + + // Wakeup-turn text + complete arrive via the adapter's update stream. + mgr.handleAgentEvent(execution, agentctl.AgentEvent{ + Type: "message_chunk", Text: "WAKEUP_FIRED\n", + }) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "complete"}) + + // === Inspect what the manager published === + readyCount := 0 + streamingTexts := []string{} + streamCompleteCount := 0 + for _, te := range eventBus.PublishedEvents { + if te.Event == nil { + continue + } + if te.Event.Type == events.AgentReady { + readyCount++ + } + if payload, ok := te.Event.Data.(AgentStreamEventPayload); ok && payload.Data != nil { + switch payload.Data.Type { + case "message_streaming": + streamingTexts = append(streamingTexts, payload.Data.Text) + case "complete": + streamCompleteCount++ + } + } + } + + t.Logf("agent.ready events published: %d (expect 2)", readyCount) + t.Logf("message_streaming events: %v", streamingTexts) + t.Logf("AgentStreamEventPayload type=complete events: %d (expect 2)", streamCompleteCount) + t.Logf("execution.Status at end: %s", execution.Status) + + // The agent-stream-level events (message_streaming, complete) are NOT + // gated on execution status — they should reach the bus for BOTH turns. + if streamCompleteCount != 2 { + t.Errorf("expected 2 stream-level complete events, got %d", streamCompleteCount) + } + if len(streamingTexts) != 2 { + t.Errorf("expected 2 message_streaming events (one per turn), got %d: %v", + len(streamingTexts), streamingTexts) + } + + // The bug: agent.ready is suppressed on the second turn because the + // execution status is already Ready when MarkReady runs. + if readyCount != 2 { + t.Errorf( + "expected exactly 2 agent.ready events, got %d. "+ + "MarkReady at manager_interaction.go:896 suppresses the wakeup turn's "+ + "AgentReady because fireWakeup bypasses SessionManager.SendPrompt and "+ + "never flips the execution back to Running. The orchestrator therefore "+ + "never receives agent.ready for the wakeup turn → "+ + "completeTurnForSession is not called → workflow on_turn_complete is "+ + "not evaluated → queued messages are not dispatched.", + readyCount, + ) + } + + // Ordering check: for each AgentReady, there must be a preceding + // AgentRunning. The orchestrator's handleAgentReady early-returns when + // session.State is not Running/Starting, so an AgentReady on the bus + // is only useful if the matching AgentRunning has flipped session + // state first. Verify by walking the published events in order and + // requiring `running_count >= ready_count` at every AgentReady index. + running, ready := 0, 0 + for _, te := range eventBus.PublishedEvents { + if te.Event == nil { + continue + } + switch te.Event.Type { + case events.AgentRunning: + running++ + case events.AgentReady: + ready++ + if running < ready { + t.Errorf("AgentReady #%d published before matching AgentRunning (running=%d at that point) — orchestrator's handleAgentReady will silently drop this", ready, running) + } + } + } +} + +// TestWakeup_PostBootMetadataEventsDoNotFlipToRunning verifies that boot-time +// metadata events (available_commands_update, agent_capabilities, context_window +// etc.) arriving AFTER MarkBootReady has put the execution into Ready do NOT +// re-arm the execution as Running. claude-agent-acp emits +// available_commands_update asynchronously ~50ms after session/new — well +// after dispatchInitialPrompt fires MarkBootReady for a no-prompt task — +// and an over-eager Ready → Running flip on that event would put the session +// into Running state in the orchestrator and break the chat-input UI for +// freshly-created tasks awaiting their first user message. +func TestWakeup_PostBootMetadataEventsDoNotFlipToRunning(t *testing.T) { + mgr, eventBus := createTestManagerWithTracking() + execution := createTestExecution("exec-1", "task-1", "session-1") + _ = mgr.executionStore.Add(execution) + + // Prime firstActivityOnce so it's already consumed before the post-boot + // metadata events arrive. In production, the adapter always emits + // `agent_capabilities` during Initialize (before MarkBootReady), which + // consumes firstActivityOnce while Status is still Running. The + // `Sync.Once.Do` call here mirrors that. + execution.firstActivityOnce.Do(func() {}) + + // Mirror the no-prompt boot path: dispatchInitialPrompt → MarkBootReady → + // status flips to Ready. + mgr.executionStore.UpdateStatus(execution.ID, v1.AgentStatusReady) + + // Boot metadata events arrive after MarkBootReady. None of these + // indicate a new turn — they're protocol metadata. + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "available_commands"}) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "agent_capabilities"}) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "session_mode"}) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "session_models"}) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "context_window"}) + + runningCount := 0 + for _, te := range eventBus.PublishedEvents { + if te.Event != nil && te.Event.Type == events.AgentRunning { + runningCount++ + } + } + // With firstActivityOnce already consumed, the only way to publish + // AgentRunning would be the post-boot whitelist flip. None of these + // metadata events are on the whitelist, so the count must be exactly 0. + if runningCount != 0 { + t.Errorf("expected 0 agent.running events from post-boot metadata, got %d — metadata-driven re-arming is broken", runningCount) + } + if execution.Status != v1.AgentStatusReady { + t.Errorf("execution.Status = %q, want %q — boot metadata events should not change status", execution.Status, v1.AgentStatusReady) + } +} + +// TestWakeup_EmptyTurnStillPublishesAgentReady covers the narrow edge case +// where a wakeup turn produces *only* a `complete` event (no preceding +// message_chunk/tool_call/etc) — e.g. when the model returns an empty +// response on wake-up. recordActivity skips the flip on `complete` events +// (they're outside the turn-content whitelist), so handleCompleteEventMarkState +// must publish AgentReady directly when it detects Status is already Ready. +func TestWakeup_EmptyTurnStillPublishesAgentReady(t *testing.T) { + mgr, eventBus := createTestManagerWithTracking() + execution := createTestExecution("exec-1", "task-1", "session-1") + _ = mgr.executionStore.Add(execution) + + // Turn 1: normal user turn. + mgr.executionStore.UpdateStatus(execution.ID, v1.AgentStatusRunning) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "message_chunk", Text: "ok\n"}) + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "complete"}) + + // Turn 2: wakeup turn with ONLY a complete event — the agent woke up + // and immediately decided it had nothing to say. + mgr.handleAgentEvent(execution, agentctl.AgentEvent{Type: "complete"}) + + readyCount := 0 + for _, te := range eventBus.PublishedEvents { + if te.Event != nil && te.Event.Type == events.AgentReady { + readyCount++ + } + } + if readyCount != 2 { + t.Errorf("expected 2 agent.ready events (one per turn, including the empty wakeup turn), got %d", readyCount) + } + + // The empty-wakeup fallback in handleCompleteEventMarkState must publish + // AgentRunning before AgentReady so the orchestrator's session-state + // guard (handleAgentReady early-returns on non-Running/Starting state) + // lets AgentReady through. + running, ready := 0, 0 + for _, te := range eventBus.PublishedEvents { + if te.Event == nil { + continue + } + switch te.Event.Type { + case events.AgentRunning: + running++ + case events.AgentReady: + ready++ + if running < ready { + t.Errorf("AgentReady #%d for empty wakeup published before matching AgentRunning — orchestrator will drop it", ready) + } + } + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/wakeup_e2e_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/wakeup_e2e_test.go new file mode 100644 index 0000000000..49e0f3f8f9 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/wakeup_e2e_test.go @@ -0,0 +1,324 @@ +//go:build wakeup_e2e + +// End-to-end isolation test for the ScheduleWakeup synthetic-prompt path. +// Spawns the real `claude-agent-acp` bridge in a child process, wires it to +// kandev's ACP Adapter, sends a prompt that calls ScheduleWakeup with a short +// delay, then waits *past* the wakeup time and asserts that the adapter +// produced events corresponding to a second (wakeup-driven) turn. +// +// Build/run: +// +// go test -tags wakeup_e2e -v -timeout 5m \ +// ./internal/agentctl/server/adapter/transport/acp/ -run WakeupE2E +// +// Skips automatically when `claude-agent-acp` (or `npx`) is not on PATH or +// when ~/.claude/.credentials.json is missing. + +package acp + +import ( + "context" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/agentctl/server/adapter/transport/shared" + "github.com/kandev/kandev/internal/agentctl/types/streams" + "github.com/kandev/kandev/internal/common/logger" +) + +const ( + wakeupDelaySeconds = 60 // ScheduleWakeup clamps to [60, 3600] + wakeupGrace = 30 * time.Second // extra time after the timer should fire +) + +// wakeupE2E spawns a single bridge instance and runs one wakeup probe. +type wakeupE2E struct { + t *testing.T + workDir string + cmd *exec.Cmd + adapter *Adapter + log *logger.Logger + + mu sync.Mutex + events []AgentEvent + closed bool +} + +func setupWakeupE2E(t *testing.T) *wakeupE2E { + t.Helper() + + if _, err := exec.LookPath("npx"); err != nil { + t.Skipf("skipping: npx not on PATH (%v)", err) + } + home, _ := os.UserHomeDir() + credPath := filepath.Join(home, ".claude", ".credentials.json") + if _, err := os.Stat(credPath); err != nil { + t.Skipf("skipping: %s not found — Claude Code OAuth credentials required (%v)", credPath, err) + } + + workDir := t.TempDir() + + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: "debug", + Format: "console", + OutputPath: "stderr", + }) + if err != nil { + t.Fatalf("logger init failed: %v", err) + } + + cfg := &shared.Config{ + AgentID: "wakeup-e2e", + WorkDir: workDir, + } + a := NewAdapter(cfg, log) + + cmd := exec.Command("npx", "-y", "@agentclientprotocol/claude-agent-acp@0.31.1") + cmd.Dir = workDir + cmd.Stderr = os.Stderr + stdinPipe, err := cmd.StdinPipe() + if err != nil { + t.Fatalf("StdinPipe: %v", err) + } + stdoutPipe, err := cmd.StdoutPipe() + if err != nil { + t.Fatalf("StdoutPipe: %v", err) + } + if err := cmd.Start(); err != nil { + t.Fatalf("failed to start bridge: %v", err) + } + + if err := a.Connect(stdinPipe, &teeReader{src: stdoutPipe, label: "bridge->adapter"}); err != nil { + _ = cmd.Process.Kill() + t.Fatalf("Connect failed: %v", err) + } + + w := &wakeupE2E{ + t: t, + workDir: workDir, + cmd: cmd, + adapter: a, + log: log, + } + + go w.collectEvents() + + t.Cleanup(func() { + w.mu.Lock() + w.closed = true + w.mu.Unlock() + _ = a.Close() + _ = stdinPipe.Close() + _ = cmd.Process.Kill() + _, _ = cmd.Process.Wait() + }) + + return w +} + +// teeReader passes bytes through to the consumer (the ACP SDK) while +// accumulating them in a side buffer; on each newline we emit the completed +// line to stderr so we can see exactly what JSON-RPC frames the bridge sent. +type teeReader struct { + src io.Reader + label string + buf []byte +} + +func (t *teeReader) Read(p []byte) (int, error) { + n, err := t.src.Read(p) + if n > 0 { + t.buf = append(t.buf, p[:n]...) + for { + i := indexOfByte(t.buf, '\n') + if i < 0 { + break + } + line := strings.TrimRight(string(t.buf[:i]), "\r") + if len(line) > 240 { + line = line[:240] + "…" + } + if line != "" { + fmt.Fprintf(os.Stderr, "[%s] %s\n", t.label, line) + } + t.buf = t.buf[i+1:] + } + } + return n, err +} + +func indexOfByte(b []byte, c byte) int { + for i, v := range b { + if v == c { + return i + } + } + return -1 +} + +func (w *wakeupE2E) collectEvents() { + for { + select { + case ev, ok := <-w.adapter.Updates(): + if !ok { + return + } + w.mu.Lock() + closed := w.closed + if !closed { + w.events = append(w.events, ev) + } + w.mu.Unlock() + fmt.Fprintf(os.Stderr, "[adapter→test] event type=%s sessionID=%s toolCallID=%s status=%s text=%q\n", + ev.Type, ev.SessionID, ev.ToolCallID, ev.ToolStatus, truncate(ev.Text, 120)) + case <-time.After(2 * time.Minute): + // Heartbeat + } + } +} + +func (w *wakeupE2E) snapshotEvents() []AgentEvent { + w.mu.Lock() + defer w.mu.Unlock() + out := make([]AgentEvent, len(w.events)) + copy(out, w.events) + return out +} + +func truncate(s string, n int) string { + if len(s) <= n { + return s + } + return s[:n] + "…" +} + +// TestWakeupE2E_BridgeQueuedTurnDrainsViaSyntheticPrompt verifies the full +// kandev wakeup pipeline against a real claude-agent-acp bridge: +// - send an initial prompt that triggers ScheduleWakeup with a short delay +// - wait past the wakeup time +// - assert that the adapter received events from a second (wakeup-driven) +// turn — i.e. the wakeupScheduler fired a synthetic prompt and the bridge +// drained its queued turn. +func TestWakeupE2E_BridgeQueuedTurnDrainsViaSyntheticPrompt(t *testing.T) { + w := setupWakeupE2E(t) + + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute) + defer cancel() + + t.Logf("initializing adapter…") + if err := w.adapter.Initialize(ctx); err != nil { + t.Fatalf("Initialize failed: %v", err) + } + t.Logf("creating session…") + sid, err := w.adapter.NewSession(ctx, nil) + if err != nil { + t.Fatalf("NewSession failed: %v", err) + } + t.Logf("session created: %s", sid) + + prompt := fmt.Sprintf( + "Use the ScheduleWakeup tool exactly once with delaySeconds=%d, "+ + "prompt=\"continue probe\", reason=\"e2e isolation test\". "+ + "Then stop. After waking up, just say \"WAKEUP_FIRED\" and stop.", + wakeupDelaySeconds, + ) + + t.Logf("sending initial prompt (will trigger ScheduleWakeup, delay=%ds)…", wakeupDelaySeconds) + initialStart := time.Now() + if err := w.adapter.Prompt(ctx, prompt, nil); err != nil { + t.Fatalf("initial Prompt failed: %v", err) + } + initialEnd := time.Now() + t.Logf("initial prompt completed in %s", initialEnd.Sub(initialStart)) + + initialEvents := w.snapshotEvents() + t.Logf("events after initial prompt: %d", len(initialEvents)) + + // Look for the ScheduleWakeup tool_call in the captured events. + sawWakeupToolCall := false + for _, ev := range initialEvents { + if ev.Type == streams.EventTypeToolCall && strings.Contains(strings.ToLower(ev.ToolName), "wakeup") { + sawWakeupToolCall = true + } + if ev.Type == streams.EventTypeToolUpdate && strings.Contains(strings.ToLower(ev.ToolName), "wakeup") { + sawWakeupToolCall = true + } + } + if !sawWakeupToolCall { + // Some agents emit ScheduleWakeup as a generic tool. Look for the title too. + for _, ev := range initialEvents { + if strings.Contains(strings.ToLower(ev.ToolTitle), "wakeup") || strings.Contains(strings.ToLower(ev.Text), "wakeup scheduled") { + sawWakeupToolCall = true + break + } + } + } + if !sawWakeupToolCall { + t.Logf("WARNING: no ScheduleWakeup tool call observed in initial events. The agent may not have called the tool. Dumping events…") + for i, ev := range initialEvents { + t.Logf(" [%d] type=%s tool=%s title=%s status=%s text=%q", + i, ev.Type, ev.ToolName, ev.ToolTitle, ev.ToolStatus, truncate(ev.Text, 100)) + } + t.Fatalf("agent did not call ScheduleWakeup; cannot proceed with wakeup-fire assertion") + } + t.Logf("✓ ScheduleWakeup tool call observed in initial events") + + // Snapshot count so we can detect new events arriving after the timer fires. + postPromptEventCount := len(initialEvents) + + // Wait for the wakeup to fire. The adapter's internal scheduler should call + // fireWakeup → a.Prompt(...) → bridge drains queued turn → events. + waitDuration := time.Duration(wakeupDelaySeconds)*time.Second + wakeupGrace + t.Logf("waiting %s for wakeup to fire (idle window starts now)…", waitDuration) + + // Sample event count every 5s so we get a clear timeline of when events + // (if any) arrive. + sampleInterval := 5 * time.Second + deadline := time.Now().Add(waitDuration) + for time.Now().Before(deadline) { + time.Sleep(sampleInterval) + current := len(w.snapshotEvents()) + elapsed := time.Since(initialEnd) + newCount := current - postPromptEventCount + t.Logf(" t+%s: total events=%d (new since prompt-end: %d)", elapsed.Truncate(time.Second), current, newCount) + } + + final := w.snapshotEvents() + wakeupEvents := final[postPromptEventCount:] + + t.Logf("post-wakeup events: %d", len(wakeupEvents)) + for i, ev := range wakeupEvents { + t.Logf(" [%d] type=%s tool=%s status=%s text=%q", + i, ev.Type, ev.ToolName, ev.ToolStatus, truncate(ev.Text, 100)) + } + + // Pass criterion: at least one agent-message-chunk-like event after the + // initial prompt completed but before the deadline. + if len(wakeupEvents) == 0 { + t.Fatalf("FAIL: no events received after initial prompt completed; wakeup did NOT drain the queued turn (%d total events captured, all from the initial prompt)", len(final)) + } + + // Optional: verify we saw an agent message after the wakeup fired. If we + // only saw e.g. heartbeats, the test is inconclusive. + sawAgentText := false + for _, ev := range wakeupEvents { + if ev.Type == streams.EventTypeMessageChunk && ev.Text != "" { + sawAgentText = true + break + } + } + if !sawAgentText { + t.Logf("NOTE: post-wakeup events arrived, but none contained agent text. " + + "The wakeup may have fired (synthetic prompt accepted) but the agent " + + "didn't produce a visible response.") + } + + t.Logf("✓ wakeup pipeline produced %d events after initial prompt completed", len(wakeupEvents)) +}