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)) +}