fix(acp): restore ScheduleWakeup wire string broken by Wakeup→Run rename (#998)
PR #914 globally renamed Wakeup→Run in the ACP transport, including the wire-level tool name check (`name != "ScheduleRun"`). That string is hardcoded by the Claude Code SDK binary as `ScheduleWakeup`, so the bridge's tool-call frames stopped matching and every Claude wakeup silently never fired: kandev never armed the timer, never sent the synthetic session/prompt, and the SDK's buffered wakeup turn only drained when the user typed a new message. Commit 0b2c59a2's lifecycle fix (publish AgentReady on wakeup turns) relied on the wakeup turn actually firing — which it wasn't — so the fix appeared not to work in production. Restore the wire string to `"ScheduleWakeup"` and rename the surrounding Go symbols back to wakeup-naming (`wakeupScheduler`, `pendingWakeup`, `handleWakeupEvent`, `fireWakeup`, `wakeupPromptTimeout`) so the next contributor can't repeat the rename. Add a comment at the match site warning that the string is externally hardcoded, plus a negative regression test asserting `"ScheduleRun"` is NOT recognised as a wakeup. Co-authored-by: Kandev Agent <agent@kandev.dev>
This commit is contained in:
@@ -46,11 +46,11 @@ const (
|
||||
configOptionIDModel = "model"
|
||||
)
|
||||
|
||||
// runPromptTimeout bounds how long a synthetic run prompt can run.
|
||||
// Run turns can perform real work (the model often runs a few tool calls
|
||||
// wakeupPromptTimeout bounds how long a synthetic wakeup prompt can run.
|
||||
// Wakeup turns can perform real work (the model often runs a few tool calls
|
||||
// before stopping) so we mirror what a normal user-initiated prompt would
|
||||
// allow rather than a tight RPC deadline.
|
||||
const runPromptTimeout = 30 * time.Minute
|
||||
const wakeupPromptTimeout = 30 * time.Minute
|
||||
|
||||
// AgentInfo contains information about the connected agent.
|
||||
type AgentInfo struct {
|
||||
@@ -116,15 +116,16 @@ type Adapter struct {
|
||||
// and rebuilt during session/load replay.
|
||||
activeMonitors map[string]map[string]string
|
||||
|
||||
// ScheduleRun tracking. The Claude Agent SDK's ScheduleRun tool fires
|
||||
// its timer inside the SDK's async-iterator, but the upstream
|
||||
// ScheduleWakeup tracking. The Claude Agent SDK's ScheduleWakeup tool
|
||||
// fires its timer inside the SDK's async-iterator, but the upstream
|
||||
// @agentclientprotocol/claude-agent-acp bridge only drains that iterator
|
||||
// inside its prompt() handler — so a run that fires while no prompt is
|
||||
// in flight produces no output. run re-injects the run as a synthetic
|
||||
// session/prompt at fire time. pendingRuns tracks per-tool-call info
|
||||
// (prompt + scheduledFor) since these arrive in separate notifications.
|
||||
run *runScheduler
|
||||
pendingRuns map[string]*pendingRun
|
||||
// inside its prompt() handler — so a wakeup that fires while no prompt
|
||||
// is in flight produces no output. wakeup re-injects the wakeup as a
|
||||
// synthetic session/prompt at fire time. pendingWakeups tracks per-tool-
|
||||
// call info (prompt + scheduledFor) since these arrive in separate
|
||||
// notifications.
|
||||
wakeup *wakeupScheduler
|
||||
pendingWakeups map[string]*pendingWakeup
|
||||
|
||||
// OTel tracing: active prompt span context.
|
||||
// Notification spans become children of the prompt span for visual grouping.
|
||||
@@ -169,7 +170,7 @@ type Adapter struct {
|
||||
closed bool
|
||||
|
||||
// lifetimeCtx is cancelled by Close. Background work that may outlive
|
||||
// the call site (e.g. the synthetic run prompt goroutine) derives its
|
||||
// the call site (e.g. the synthetic wakeup prompt goroutine) derives its
|
||||
// context from this one so it aborts when the adapter shuts down rather
|
||||
// than continuing to drive a dead subprocess.
|
||||
lifetimeCtx context.Context
|
||||
@@ -190,13 +191,13 @@ func NewAdapter(cfg *shared.Config, log *logger.Logger) *Adapter {
|
||||
updatesCh: make(chan AgentEvent, 100),
|
||||
activeToolCalls: make(map[string]*streams.NormalizedPayload),
|
||||
activeMonitors: make(map[string]map[string]string),
|
||||
pendingRuns: make(map[string]*pendingRun),
|
||||
pendingWakeups: make(map[string]*pendingWakeup),
|
||||
usageBySession: make(map[string]*usageTracker),
|
||||
attachMgr: shared.NewAttachmentManager(cfg.WorkDir, l.Zap()),
|
||||
lifetimeCtx: ctx,
|
||||
lifetimeCancel: cancel,
|
||||
}
|
||||
a.run = newRunScheduler(l, a.fireRun)
|
||||
a.wakeup = newWakeupScheduler(l, a.fireWakeup)
|
||||
return a
|
||||
}
|
||||
|
||||
@@ -315,13 +316,13 @@ func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer)
|
||||
return "", fmt.Errorf("adapter not initialized")
|
||||
}
|
||||
|
||||
// A fresh session invalidates any pending run keyed to the prior
|
||||
// session. Reset pendingRuns and cancel the scheduler under one
|
||||
// a.mu critical section so a concurrent handleRunEvent can't slip
|
||||
// A fresh session invalidates any pending wakeup keyed to the prior
|
||||
// session. Reset pendingWakeups and cancel the scheduler under one
|
||||
// a.mu critical section so a concurrent handleWakeupEvent can't slip
|
||||
// a stale entry between the two operations.
|
||||
a.mu.Lock()
|
||||
a.pendingRuns = make(map[string]*pendingRun)
|
||||
a.run.cancel()
|
||||
a.pendingWakeups = make(map[string]*pendingWakeup)
|
||||
a.wakeup.cancel()
|
||||
a.mu.Unlock()
|
||||
|
||||
ctx, span := shared.TraceProtocolRequest(ctx, shared.ProtocolACP, a.agentID, "session.new")
|
||||
@@ -498,13 +499,13 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers
|
||||
return fmt.Errorf("agent does not support session loading (LoadSession capability is false)")
|
||||
}
|
||||
|
||||
// Loading a different session invalidates any pending run keyed to the
|
||||
// Loading a different session invalidates any pending wakeup keyed to the
|
||||
// prior session — same reset block as NewSession to avoid leaving an armed
|
||||
// timer for a session id that's about to change and accumulating stale
|
||||
// pendingRuns entries across reloads.
|
||||
// pendingWakeups entries across reloads.
|
||||
a.mu.Lock()
|
||||
a.pendingRuns = make(map[string]*pendingRun)
|
||||
a.run.cancel()
|
||||
a.pendingWakeups = make(map[string]*pendingWakeup)
|
||||
a.wakeup.cancel()
|
||||
a.mu.Unlock()
|
||||
|
||||
ctx, span := shared.TraceProtocolRequest(ctx, shared.ProtocolACP, a.agentID, "session.load")
|
||||
@@ -761,9 +762,9 @@ func (a *Adapter) Prompt(ctx context.Context, message string, attachments []v1.M
|
||||
return nil
|
||||
}
|
||||
|
||||
// fireRun is invoked by runScheduler when a ScheduleRun timer
|
||||
// fireWakeup is invoked by wakeupScheduler when a ScheduleWakeup timer
|
||||
// elapses. It issues a synthetic session/prompt so the upstream
|
||||
// @agentclientprotocol/claude-agent-acp bridge drains the SDK's queued run
|
||||
// @agentclientprotocol/claude-agent-acp bridge drains the SDK's queued wakeup
|
||||
// turn and emits visible ACP frames. The session must still match (the user
|
||||
// hasn't started a fresh session) and the adapter must not be closed.
|
||||
//
|
||||
@@ -774,80 +775,80 @@ func (a *Adapter) Prompt(ctx context.Context, message string, attachments []v1.M
|
||||
// JSON-RPC pairs each response back to its originating request via id —
|
||||
// but it does mean two prompts can be in flight against the bridge at
|
||||
// once. The bridge serialises them in the order it receives them, which
|
||||
// is exactly what we want for a run that races a user message.
|
||||
func (a *Adapter) fireRun(sessionID, prompt string) {
|
||||
// is exactly what we want for a wakeup that races a user message.
|
||||
func (a *Adapter) fireWakeup(sessionID, prompt string) {
|
||||
a.mu.RLock()
|
||||
closed := a.closed
|
||||
currentSession := a.sessionID
|
||||
a.mu.RUnlock()
|
||||
|
||||
if closed {
|
||||
a.logger.Debug("skipping run fire: adapter closed",
|
||||
a.logger.Debug("skipping wakeup fire: adapter closed",
|
||||
zap.String("session_id", sessionID))
|
||||
return
|
||||
}
|
||||
if currentSession != sessionID {
|
||||
a.logger.Info("skipping run fire: session changed",
|
||||
a.logger.Info("skipping wakeup fire: session changed",
|
||||
zap.String("scheduled_for", sessionID),
|
||||
zap.String("current", currentSession))
|
||||
return
|
||||
}
|
||||
|
||||
a.logger.Info("injecting synthetic run prompt",
|
||||
a.logger.Info("injecting synthetic wakeup prompt",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("prompt_len", len(prompt)))
|
||||
|
||||
go func() {
|
||||
// Derive from lifetimeCtx so a concurrent Close aborts the in-flight
|
||||
// prompt instead of letting it run against a dead subprocess.
|
||||
ctx, cancel := context.WithTimeout(a.lifetimeCtx, runPromptTimeout)
|
||||
ctx, cancel := context.WithTimeout(a.lifetimeCtx, wakeupPromptTimeout)
|
||||
defer cancel()
|
||||
if err := a.Prompt(ctx, prompt, nil); err != nil {
|
||||
a.logger.Error("synthetic run prompt failed",
|
||||
a.logger.Error("synthetic wakeup prompt failed",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// handleRunEvent inspects a tool-call meta + rawInput pair, accumulates
|
||||
// pending state per toolCallID, and schedules a run once both the prompt
|
||||
// handleWakeupEvent inspects a tool-call meta + rawInput pair, accumulates
|
||||
// pending state per toolCallID, and schedules a wakeup once both the prompt
|
||||
// and scheduledFor timestamp are known. terminal=true means the tool call has
|
||||
// reached a terminal state, so any pending entry should be cleaned up.
|
||||
func (a *Adapter) handleRunEvent(sessionID, toolCallID string, meta any, rawInput any, terminal bool) {
|
||||
func (a *Adapter) handleWakeupEvent(sessionID, toolCallID string, meta any, rawInput any, terminal bool) {
|
||||
if toolCallID == "" {
|
||||
return
|
||||
}
|
||||
|
||||
scheduledForMs, isRun := extractScheduleRun(meta)
|
||||
scheduledForMs, isWakeup := extractScheduleWakeup(meta)
|
||||
|
||||
a.mu.Lock()
|
||||
pw, tracked := a.pendingRuns[toolCallID]
|
||||
pw, tracked := a.pendingWakeups[toolCallID]
|
||||
if !tracked {
|
||||
if !isRun {
|
||||
if !isWakeup {
|
||||
a.mu.Unlock()
|
||||
return
|
||||
}
|
||||
pw = &pendingRun{}
|
||||
a.pendingRuns[toolCallID] = pw
|
||||
pw = &pendingWakeup{}
|
||||
a.pendingWakeups[toolCallID] = pw
|
||||
}
|
||||
|
||||
if scheduledForMs > 0 {
|
||||
pw.scheduledForMs = scheduledForMs
|
||||
}
|
||||
if prompt, ok := extractRunPrompt(rawInput); ok {
|
||||
if prompt, ok := extractWakeupPrompt(rawInput); ok {
|
||||
pw.prompt = prompt
|
||||
}
|
||||
|
||||
prompt := pw.prompt
|
||||
stamp := pw.scheduledForMs
|
||||
if (prompt != "" && stamp > 0) || terminal {
|
||||
delete(a.pendingRuns, toolCallID)
|
||||
delete(a.pendingWakeups, toolCallID)
|
||||
}
|
||||
a.mu.Unlock()
|
||||
|
||||
if prompt != "" && stamp > 0 {
|
||||
a.run.schedule(sessionID, prompt, stamp)
|
||||
a.wakeup.schedule(sessionID, prompt, stamp)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -983,10 +984,10 @@ func (a *Adapter) Close() error {
|
||||
|
||||
a.logger.Info("closing ACP adapter")
|
||||
|
||||
// Stop any pending ScheduleRun timer so it doesn't fire after close,
|
||||
// and cancel the lifetime context so any in-flight run prompt aborts.
|
||||
if a.run != nil {
|
||||
a.run.cancel()
|
||||
// Stop any pending ScheduleWakeup timer so it doesn't fire after close,
|
||||
// and cancel the lifetime context so any in-flight wakeup prompt aborts.
|
||||
if a.wakeup != nil {
|
||||
a.wakeup.cancel()
|
||||
}
|
||||
if a.lifetimeCancel != nil {
|
||||
a.lifetimeCancel()
|
||||
@@ -1824,10 +1825,10 @@ func (a *Adapter) convertToolCallUpdate(sessionID string, tc *acp.SessionUpdateT
|
||||
a.activeToolCalls[toolCallID] = normalizedPayload
|
||||
a.mu.Unlock()
|
||||
|
||||
// ScheduleRun tracking: meta carries `_meta.claudeCode.toolName`
|
||||
// ScheduleWakeup tracking: meta carries `_meta.claudeCode.toolName`
|
||||
// on the initial tool_call; rawInput is usually empty here but record
|
||||
// the prompt eagerly when it does arrive in the same notification.
|
||||
a.handleRunEvent(sessionID, toolCallID, tc.Meta, tc.RawInput, false)
|
||||
a.handleWakeupEvent(sessionID, toolCallID, tc.Meta, tc.RawInput, false)
|
||||
|
||||
// Detect tool type for logging
|
||||
toolType := DetectToolOperationType(toolKind, args)
|
||||
@@ -1954,10 +1955,10 @@ func (a *Adapter) convertToolCallResultUpdate(sessionID string, tcu *acp.Session
|
||||
}
|
||||
a.mu.Unlock()
|
||||
|
||||
// ScheduleRun tracking: tool_call_update is where rawInput.prompt and
|
||||
// ScheduleWakeup tracking: tool_call_update is where rawInput.prompt and
|
||||
// `_meta.claudeCode.toolResponse.scheduledFor` typically arrive. Once both
|
||||
// are known, schedule the synthetic prompt; on terminal status, clean up.
|
||||
a.handleRunEvent(sessionID, toolCallID, tcu.Meta, tcu.RawInput, isTerminal)
|
||||
a.handleWakeupEvent(sessionID, toolCallID, tcu.Meta, tcu.RawInput, isTerminal)
|
||||
|
||||
// When a switch_mode tool carries a plan (e.g. ExitPlanMode), emit it
|
||||
// as an agent_plan event so the orchestrator creates a visible plan message.
|
||||
|
||||
+46
-37
@@ -8,22 +8,27 @@ import (
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// runScheduler holds at most one pending ScheduleRun-driven prompt for
|
||||
// the adapter. The Claude Agent SDK exposes `ScheduleRun` as a tool inside
|
||||
// wakeupScheduler holds at most one pending ScheduleWakeup-driven prompt for
|
||||
// the adapter. The Claude Agent SDK exposes `ScheduleWakeup` as a tool inside
|
||||
// its harness; when its timer fires, the SDK queues a new turn on its
|
||||
// async-iterator (`session.query`). The upstream
|
||||
// @agentclientprotocol/claude-agent-acp bridge only iterates that channel
|
||||
// inside its `prompt()` handler, so a run that fires while no
|
||||
// inside its `prompt()` handler, so a wakeup that fires while no
|
||||
// `session/prompt` is in flight produces no ACP output at all — the buffered
|
||||
// turn only drains on the next user message.
|
||||
//
|
||||
// This scheduler closes the gap: when we observe a `ScheduleRun` tool
|
||||
// This scheduler closes the gap: when we observe a `ScheduleWakeup` tool
|
||||
// completion, we record `scheduledFor` and the prompt text, and at fire time
|
||||
// issue a synthetic `session/prompt` to the bridge. That triggers the bridge
|
||||
// to drain the buffered output and produce visible ACP frames for the run
|
||||
// turn — matching what standalone Claude Code does, where the run prompt
|
||||
// to drain the buffered output and produce visible ACP frames for the wakeup
|
||||
// turn — matching what standalone Claude Code does, where the wakeup prompt
|
||||
// is injected as a synthesized user turn.
|
||||
type runScheduler struct {
|
||||
//
|
||||
// The wire-level tool name "ScheduleWakeup" is hardcoded in the Claude Code
|
||||
// SDK binary (see `extractScheduleWakeup`) and MUST NOT be renamed when
|
||||
// refactoring; PR #914 globally renamed Wakeup→Run including this string,
|
||||
// which silently broke every Claude wakeup until it was reverted here.
|
||||
type wakeupScheduler struct {
|
||||
logger *logger.Logger
|
||||
fire func(sessionID, prompt string)
|
||||
|
||||
@@ -37,38 +42,38 @@ type runScheduler struct {
|
||||
// goroutine for the old timer (timer.Stop() returns false) and the new
|
||||
// schedule() then writes new state before the stale fireOnce acquires the
|
||||
// lock. Without gen, the stale fireOnce would consume the new state and
|
||||
// fire the new run immediately while clobbering w.timer.
|
||||
// fire the new wakeup immediately while clobbering w.timer.
|
||||
gen uint64
|
||||
}
|
||||
|
||||
// pendingRun accumulates partial state for an in-flight ScheduleRun tool
|
||||
// call. The bridge emits the run metadata across multiple notifications:
|
||||
// the initial tool_call carries `_meta.claudeCode.toolName="ScheduleRun"`
|
||||
// pendingWakeup accumulates partial state for an in-flight ScheduleWakeup tool
|
||||
// call. The bridge emits the wakeup metadata across multiple notifications:
|
||||
// the initial tool_call carries `_meta.claudeCode.toolName="ScheduleWakeup"`
|
||||
// (no rawInput yet); subsequent tool_call_updates fill in `rawInput.prompt`
|
||||
// and `_meta.claudeCode.toolResponse.scheduledFor`. We only have enough to
|
||||
// schedule once both the prompt and scheduledFor timestamp are present.
|
||||
type pendingRun struct {
|
||||
type pendingWakeup struct {
|
||||
prompt string
|
||||
scheduledForMs int64
|
||||
}
|
||||
|
||||
func newRunScheduler(log *logger.Logger, fire func(sessionID, prompt string)) *runScheduler {
|
||||
return &runScheduler{
|
||||
logger: log.WithFields(zap.String("component", "run-scheduler")),
|
||||
func newWakeupScheduler(log *logger.Logger, fire func(sessionID, prompt string)) *wakeupScheduler {
|
||||
return &wakeupScheduler{
|
||||
logger: log.WithFields(zap.String("component", "wakeup-scheduler")),
|
||||
fire: fire,
|
||||
}
|
||||
}
|
||||
|
||||
// schedule registers a run for the given session, replacing any prior
|
||||
// pending run. unixMs is the Unix-epoch millisecond timestamp at which the
|
||||
// run should fire (matches the `_meta.claudeCode.toolResponse.scheduledFor`
|
||||
// schedule registers a wakeup for the given session, replacing any prior
|
||||
// pending wakeup. unixMs is the Unix-epoch millisecond timestamp at which the
|
||||
// wakeup should fire (matches the `_meta.claudeCode.toolResponse.scheduledFor`
|
||||
// field reported by the bridge). If the timestamp is in the past or zero, the
|
||||
// run is dropped without firing — that matches the ACP semantics of "the
|
||||
// wakeup is dropped without firing — that matches the ACP semantics of "the
|
||||
// turn already happened" and avoids unbounded immediate-fire loops on stale
|
||||
// timestamps.
|
||||
func (w *runScheduler) schedule(sessionID, prompt string, unixMs int64) {
|
||||
func (w *wakeupScheduler) schedule(sessionID, prompt string, unixMs int64) {
|
||||
if sessionID == "" || prompt == "" || unixMs <= 0 {
|
||||
w.logger.Warn("ignoring run with missing fields",
|
||||
w.logger.Warn("ignoring wakeup with missing fields",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("prompt_len", len(prompt)),
|
||||
zap.Int64("unix_ms", unixMs))
|
||||
@@ -77,10 +82,10 @@ func (w *runScheduler) schedule(sessionID, prompt string, unixMs int64) {
|
||||
|
||||
delay := time.Until(time.UnixMilli(unixMs))
|
||||
if delay <= 0 {
|
||||
// Warn rather than Debug: a stale run typically signals system
|
||||
// Warn rather than Debug: a stale wakeup typically signals system
|
||||
// suspension, container pause, or significant clock skew — those are
|
||||
// production-relevant and shouldn't be hidden at default log levels.
|
||||
w.logger.Warn("dropping stale run",
|
||||
w.logger.Warn("dropping stale wakeup",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Duration("past_by", -delay))
|
||||
return
|
||||
@@ -97,7 +102,7 @@ func (w *runScheduler) schedule(sessionID, prompt string, unixMs int64) {
|
||||
w.timer = time.AfterFunc(delay, func() { w.fireOnce(myGen) })
|
||||
w.mu.Unlock()
|
||||
|
||||
w.logger.Info("scheduled run",
|
||||
w.logger.Info("scheduled wakeup",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Time("fires_at", time.UnixMilli(unixMs)),
|
||||
zap.Duration("delay", delay))
|
||||
@@ -107,11 +112,11 @@ func (w *runScheduler) schedule(sessionID, prompt string, unixMs int64) {
|
||||
// myGen identifies the schedule call that armed this timer; if a later
|
||||
// schedule or cancel has bumped w.gen, this fire is stale and is dropped.
|
||||
// It clears the pending state before calling the callback so a re-entry from
|
||||
// inside the callback (e.g. the synthetic prompt schedules another run)
|
||||
// inside the callback (e.g. the synthetic prompt schedules another wakeup)
|
||||
// can overwrite cleanly. The gen check guarantees sessionID is non-empty
|
||||
// here — schedule() rejects empty sessionID before bumping gen, and any
|
||||
// cancel would have bumped gen too — so we don't re-check it.
|
||||
func (w *runScheduler) fireOnce(myGen uint64) {
|
||||
func (w *wakeupScheduler) fireOnce(myGen uint64) {
|
||||
w.mu.Lock()
|
||||
if myGen != w.gen {
|
||||
w.mu.Unlock()
|
||||
@@ -125,18 +130,18 @@ func (w *runScheduler) fireOnce(myGen uint64) {
|
||||
w.gen++
|
||||
w.mu.Unlock()
|
||||
|
||||
w.logger.Info("firing run",
|
||||
w.logger.Info("firing wakeup",
|
||||
zap.String("session_id", sessionID))
|
||||
w.fire(sessionID, prompt)
|
||||
}
|
||||
|
||||
// cancel stops any pending run. Safe to call multiple times. Bumping gen
|
||||
// cancel stops any pending wakeup. Safe to call multiple times. Bumping gen
|
||||
// invalidates any in-flight fireOnce that has already been launched but is
|
||||
// blocked on the lock — without that, a stale fireOnce after Stop()=false
|
||||
// would still observe the cleared state and no-op cleanly, but cancel races
|
||||
// with concurrent schedule could otherwise let a stale fire consume newly
|
||||
// scheduled state.
|
||||
func (w *runScheduler) cancel() {
|
||||
func (w *wakeupScheduler) cancel() {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if w.timer != nil {
|
||||
@@ -148,13 +153,17 @@ func (w *runScheduler) cancel() {
|
||||
w.gen++
|
||||
}
|
||||
|
||||
// extractScheduleRun inspects an ACP tool-call meta map and returns
|
||||
// (scheduledForUnixMs, isScheduleRun). It expects the bridge's
|
||||
// extractScheduleWakeup inspects an ACP tool-call meta map and returns
|
||||
// (scheduledForUnixMs, isScheduleWakeup). It expects the bridge's
|
||||
// `_meta.claudeCode.toolName` and `_meta.claudeCode.toolResponse.scheduledFor`
|
||||
// shape that @agentclientprotocol/claude-agent-acp emits; missing/malformed
|
||||
// entries are reported as not-a-run rather than as errors so unrelated
|
||||
// entries are reported as not-a-wakeup rather than as errors so unrelated
|
||||
// tools pass through unaffected.
|
||||
func extractScheduleRun(meta any) (scheduledForMs int64, isRun bool) {
|
||||
//
|
||||
// The literal "ScheduleWakeup" string is the Claude Code SDK's hardcoded tool
|
||||
// name (verified in the upstream binary). Do not rename: that would make this
|
||||
// match nothing and every wakeup turn silently never fire.
|
||||
func extractScheduleWakeup(meta any) (scheduledForMs int64, isWakeup bool) {
|
||||
m, ok := meta.(map[string]any)
|
||||
if !ok {
|
||||
return 0, false
|
||||
@@ -163,12 +172,12 @@ func extractScheduleRun(meta any) (scheduledForMs int64, isRun bool) {
|
||||
if !ok {
|
||||
return 0, false
|
||||
}
|
||||
if name, _ := cc["toolName"].(string); name != "ScheduleRun" {
|
||||
if name, _ := cc["toolName"].(string); name != "ScheduleWakeup" {
|
||||
return 0, false
|
||||
}
|
||||
resp, ok := cc["toolResponse"].(map[string]any)
|
||||
if !ok {
|
||||
return 0, true // ScheduleRun tool, but response not yet present
|
||||
return 0, true // ScheduleWakeup tool, but response not yet present
|
||||
}
|
||||
switch v := resp["scheduledFor"].(type) {
|
||||
case float64:
|
||||
@@ -181,10 +190,10 @@ func extractScheduleRun(meta any) (scheduledForMs int64, isRun bool) {
|
||||
return 0, true
|
||||
}
|
||||
|
||||
// extractRunPrompt pulls the prompt string out of a ScheduleRun
|
||||
// extractWakeupPrompt pulls the prompt string out of a ScheduleWakeup
|
||||
// tool-call rawInput. Returns ("", false) when the field is absent or the
|
||||
// wrong type.
|
||||
func extractRunPrompt(rawInput any) (string, bool) {
|
||||
func extractWakeupPrompt(rawInput any) (string, bool) {
|
||||
m, ok := rawInput.(map[string]any)
|
||||
if !ok {
|
||||
return "", false
|
||||
+82
-78
@@ -10,7 +10,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/common/logger"
|
||||
)
|
||||
|
||||
func newTestRunLogger(t *testing.T) *logger.Logger {
|
||||
func newTestWakeupLogger(t *testing.T) *logger.Logger {
|
||||
t.Helper()
|
||||
log, err := logger.NewLogger(logger.LoggingConfig{
|
||||
Level: "error",
|
||||
@@ -23,7 +23,7 @@ func newTestRunLogger(t *testing.T) *logger.Logger {
|
||||
return log
|
||||
}
|
||||
|
||||
func TestExtractScheduleRun_NotARun(t *testing.T) {
|
||||
func TestExtractScheduleWakeup_NotAWakeup(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
meta any
|
||||
@@ -36,33 +36,41 @@ func TestExtractScheduleRun_NotARun(t *testing.T) {
|
||||
{"different tool", map[string]any{
|
||||
"claudeCode": map[string]any{"toolName": "Bash"},
|
||||
}},
|
||||
// Regression for PR #914: a global Wakeup→Run rename flipped the wire
|
||||
// string check to "ScheduleRun", which matched nothing the Claude SDK
|
||||
// emits and silently broke every wakeup. Lock in that "ScheduleRun"
|
||||
// is NOT recognised as a wakeup so a future rename can't repeat that
|
||||
// mistake without flipping this assertion too.
|
||||
{"ScheduleRun is not a wakeup", map[string]any{
|
||||
"claudeCode": map[string]any{"toolName": "ScheduleRun"},
|
||||
}},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
ms, ok := extractScheduleRun(c.meta)
|
||||
ms, ok := extractScheduleWakeup(c.meta)
|
||||
if ok {
|
||||
t.Errorf("expected isRun=false for %s, got true (ms=%d)", c.name, ms)
|
||||
t.Errorf("expected isWakeup=false for %s, got true (ms=%d)", c.name, ms)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractScheduleRun_NoResponseYet(t *testing.T) {
|
||||
func TestExtractScheduleWakeup_NoResponseYet(t *testing.T) {
|
||||
meta := map[string]any{
|
||||
"claudeCode": map[string]any{
|
||||
"toolName": "ScheduleRun",
|
||||
"toolName": "ScheduleWakeup",
|
||||
},
|
||||
}
|
||||
ms, ok := extractScheduleRun(meta)
|
||||
ms, ok := extractScheduleWakeup(meta)
|
||||
if !ok {
|
||||
t.Fatal("expected isRun=true")
|
||||
t.Fatal("expected isWakeup=true")
|
||||
}
|
||||
if ms != 0 {
|
||||
t.Errorf("expected ms=0 when toolResponse missing, got %d", ms)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractScheduleRun_WithResponse(t *testing.T) {
|
||||
func TestExtractScheduleWakeup_WithResponse(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
val any
|
||||
@@ -77,15 +85,15 @@ func TestExtractScheduleRun_WithResponse(t *testing.T) {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
meta := map[string]any{
|
||||
"claudeCode": map[string]any{
|
||||
"toolName": "ScheduleRun",
|
||||
"toolName": "ScheduleWakeup",
|
||||
"toolResponse": map[string]any{
|
||||
"scheduledFor": c.val,
|
||||
},
|
||||
},
|
||||
}
|
||||
ms, ok := extractScheduleRun(meta)
|
||||
ms, ok := extractScheduleWakeup(meta)
|
||||
if !ok {
|
||||
t.Fatal("expected isRun=true")
|
||||
t.Fatal("expected isWakeup=true")
|
||||
}
|
||||
if ms != c.want {
|
||||
t.Errorf("ms=%d, want %d", ms, c.want)
|
||||
@@ -94,7 +102,7 @@ func TestExtractScheduleRun_WithResponse(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestExtractRunPrompt(t *testing.T) {
|
||||
func TestExtractWakeupPrompt(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
input any
|
||||
@@ -110,7 +118,7 @@ func TestExtractRunPrompt(t *testing.T) {
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
got, ok := extractRunPrompt(c.input)
|
||||
got, ok := extractWakeupPrompt(c.input)
|
||||
if ok != c.wantOK || got != c.wantText {
|
||||
t.Errorf("got (%q, %v), want (%q, %v)", got, ok, c.wantText, c.wantOK)
|
||||
}
|
||||
@@ -118,18 +126,18 @@ func TestExtractRunPrompt(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
type runCapture struct {
|
||||
type wakeupCapture struct {
|
||||
mu sync.Mutex
|
||||
calls int32
|
||||
lastSess string
|
||||
lastPrmpt string
|
||||
}
|
||||
|
||||
func newRunCapture() *runCapture {
|
||||
return &runCapture{}
|
||||
func newWakeupCapture() *wakeupCapture {
|
||||
return &wakeupCapture{}
|
||||
}
|
||||
|
||||
func (c *runCapture) fire(sessionID, prompt string) {
|
||||
func (c *wakeupCapture) fire(sessionID, prompt string) {
|
||||
c.mu.Lock()
|
||||
c.lastSess = sessionID
|
||||
c.lastPrmpt = prompt
|
||||
@@ -144,11 +152,11 @@ func (c *runCapture) fire(sessionID, prompt string) {
|
||||
// synctest.Wait() afterwards ensures any goroutine the timer spawned has
|
||||
// finished its work before the assertion.
|
||||
|
||||
func TestRunScheduler_FiresAfterDelay(t *testing.T) {
|
||||
func TestWakeupScheduler_FiresAfterDelay(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
log := newTestRunLogger(t)
|
||||
cap := newRunCapture()
|
||||
sched := newRunScheduler(log, cap.fire)
|
||||
log := newTestWakeupLogger(t)
|
||||
cap := newWakeupCapture()
|
||||
sched := newWakeupScheduler(log, cap.fire)
|
||||
|
||||
at := time.Now().Add(40 * time.Millisecond).UnixMilli()
|
||||
sched.schedule("sess-1", "wake-prompt", at)
|
||||
@@ -167,11 +175,11 @@ func TestRunScheduler_FiresAfterDelay(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestRunScheduler_CancelPreventsFire(t *testing.T) {
|
||||
func TestWakeupScheduler_CancelPreventsFire(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
log := newTestRunLogger(t)
|
||||
cap := newRunCapture()
|
||||
sched := newRunScheduler(log, cap.fire)
|
||||
log := newTestWakeupLogger(t)
|
||||
cap := newWakeupCapture()
|
||||
sched := newWakeupScheduler(log, cap.fire)
|
||||
|
||||
at := time.Now().Add(50 * time.Millisecond).UnixMilli()
|
||||
sched.schedule("sess-1", "p", at)
|
||||
@@ -187,11 +195,11 @@ func TestRunScheduler_CancelPreventsFire(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestRunScheduler_StaleTimestampDropped(t *testing.T) {
|
||||
func TestWakeupScheduler_StaleTimestampDropped(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
log := newTestRunLogger(t)
|
||||
cap := newRunCapture()
|
||||
sched := newRunScheduler(log, cap.fire)
|
||||
log := newTestWakeupLogger(t)
|
||||
cap := newWakeupCapture()
|
||||
sched := newWakeupScheduler(log, cap.fire)
|
||||
|
||||
// Past timestamp: should not arm a timer at all.
|
||||
sched.schedule("sess-1", "p", time.Now().Add(-time.Hour).UnixMilli())
|
||||
@@ -205,11 +213,11 @@ func TestRunScheduler_StaleTimestampDropped(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestRunScheduler_RescheduleReplacesPrior(t *testing.T) {
|
||||
func TestWakeupScheduler_RescheduleReplacesPrior(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
log := newTestRunLogger(t)
|
||||
cap := newRunCapture()
|
||||
sched := newRunScheduler(log, cap.fire)
|
||||
log := newTestWakeupLogger(t)
|
||||
cap := newWakeupCapture()
|
||||
sched := newWakeupScheduler(log, cap.fire)
|
||||
|
||||
// First schedule far in the future, then reschedule soon — only the
|
||||
// second should fire, and only once.
|
||||
@@ -230,15 +238,15 @@ func TestRunScheduler_RescheduleReplacesPrior(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
// TestRunScheduler_StaleFireOnceDoesNotConsumeNewState reproduces the race
|
||||
// TestWakeupScheduler_StaleFireOnceDoesNotConsumeNewState reproduces the race
|
||||
// where time.AfterFunc has already launched a fireOnce goroutine for the old
|
||||
// timer before schedule() runs. The stale fireOnce must observe the gen
|
||||
// mismatch and bail out, not consume the newly scheduled run.
|
||||
func TestRunScheduler_StaleFireOnceDoesNotConsumeNewState(t *testing.T) {
|
||||
// mismatch and bail out, not consume the newly scheduled wakeup.
|
||||
func TestWakeupScheduler_StaleFireOnceDoesNotConsumeNewState(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
log := newTestRunLogger(t)
|
||||
cap := newRunCapture()
|
||||
sched := newRunScheduler(log, cap.fire)
|
||||
log := newTestWakeupLogger(t)
|
||||
cap := newWakeupCapture()
|
||||
sched := newWakeupScheduler(log, cap.fire)
|
||||
|
||||
// Manually drive the race: arm a timer, capture its gen, then simulate
|
||||
// a stale fire by calling fireOnce with the captured gen *after* a
|
||||
@@ -278,10 +286,10 @@ func TestRunScheduler_StaleFireOnceDoesNotConsumeNewState(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestRunScheduler_IgnoresMissingFields(t *testing.T) {
|
||||
log := newTestRunLogger(t)
|
||||
cap := newRunCapture()
|
||||
sched := newRunScheduler(log, cap.fire)
|
||||
func TestWakeupScheduler_IgnoresMissingFields(t *testing.T) {
|
||||
log := newTestWakeupLogger(t)
|
||||
cap := newWakeupCapture()
|
||||
sched := newWakeupScheduler(log, cap.fire)
|
||||
|
||||
// Missing each required field in turn — none should arm a timer.
|
||||
sched.schedule("", "p", time.Now().Add(time.Minute).UnixMilli())
|
||||
@@ -293,24 +301,24 @@ func TestRunScheduler_IgnoresMissingFields(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleRunEvent_SchedulesOnceBothFieldsArrive(t *testing.T) {
|
||||
func TestHandleWakeupEvent_SchedulesOnceBothFieldsArrive(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
t.Cleanup(func() { _ = a.Close() })
|
||||
|
||||
cap := newRunCapture()
|
||||
a.run = newRunScheduler(a.logger, cap.fire)
|
||||
cap := newWakeupCapture()
|
||||
a.wakeup = newWakeupScheduler(a.logger, cap.fire)
|
||||
|
||||
at := time.Now().Add(40 * time.Millisecond).UnixMilli()
|
||||
|
||||
// Initial tool_call: only the toolName is known.
|
||||
metaInitial := map[string]any{
|
||||
"claudeCode": map[string]any{"toolName": "ScheduleRun"},
|
||||
"claudeCode": map[string]any{"toolName": "ScheduleWakeup"},
|
||||
}
|
||||
a.handleRunEvent("sess-1", "tc-1", metaInitial, nil, false)
|
||||
a.handleWakeupEvent("sess-1", "tc-1", metaInitial, nil, false)
|
||||
|
||||
// Mid-update: rawInput arrives with the prompt.
|
||||
a.handleRunEvent("sess-1", "tc-1", metaInitial, map[string]any{
|
||||
a.handleWakeupEvent("sess-1", "tc-1", metaInitial, map[string]any{
|
||||
"prompt": "wake-now",
|
||||
"delaySeconds": 60,
|
||||
}, false)
|
||||
@@ -323,11 +331,11 @@ func TestHandleRunEvent_SchedulesOnceBothFieldsArrive(t *testing.T) {
|
||||
// Final update: scheduledFor arrives in meta.
|
||||
metaWithResp := map[string]any{
|
||||
"claudeCode": map[string]any{
|
||||
"toolName": "ScheduleRun",
|
||||
"toolName": "ScheduleWakeup",
|
||||
"toolResponse": map[string]any{"scheduledFor": float64(at)},
|
||||
},
|
||||
}
|
||||
a.handleRunEvent("sess-1", "tc-1", metaWithResp, nil, false)
|
||||
a.handleWakeupEvent("sess-1", "tc-1", metaWithResp, nil, false)
|
||||
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
synctest.Wait()
|
||||
@@ -343,12 +351,12 @@ func TestHandleRunEvent_SchedulesOnceBothFieldsArrive(t *testing.T) {
|
||||
})
|
||||
}
|
||||
|
||||
func TestHandleRunEvent_NonRunIgnored(t *testing.T) {
|
||||
func TestHandleWakeupEvent_NonWakeupIgnored(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
t.Cleanup(func() { _ = a.Close() })
|
||||
|
||||
cap := newRunCapture()
|
||||
a.run = newRunScheduler(a.logger, cap.fire)
|
||||
cap := newWakeupCapture()
|
||||
a.wakeup = newWakeupScheduler(a.logger, cap.fire)
|
||||
|
||||
// Bash tool — should be ignored entirely.
|
||||
meta := map[string]any{
|
||||
@@ -357,11 +365,11 @@ func TestHandleRunEvent_NonRunIgnored(t *testing.T) {
|
||||
"toolResponse": map[string]any{"scheduledFor": float64(time.Now().Add(time.Hour).UnixMilli())},
|
||||
},
|
||||
}
|
||||
a.handleRunEvent("sess-1", "tc-1", meta, map[string]any{"prompt": "x"}, false)
|
||||
a.handleWakeupEvent("sess-1", "tc-1", meta, map[string]any{"prompt": "x"}, false)
|
||||
|
||||
a.mu.Lock()
|
||||
if _, present := a.pendingRuns["tc-1"]; present {
|
||||
t.Error("non-run tool should not be tracked")
|
||||
if _, present := a.pendingWakeups["tc-1"]; present {
|
||||
t.Error("non-wakeup tool should not be tracked")
|
||||
}
|
||||
a.mu.Unlock()
|
||||
|
||||
@@ -370,59 +378,55 @@ func TestHandleRunEvent_NonRunIgnored(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleRunEvent_TerminalCleansUpPending(t *testing.T) {
|
||||
func TestHandleWakeupEvent_TerminalCleansUpPending(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
t.Cleanup(func() { _ = a.Close() })
|
||||
|
||||
cap := newRunCapture()
|
||||
a.run = newRunScheduler(a.logger, cap.fire)
|
||||
cap := newWakeupCapture()
|
||||
a.wakeup = newWakeupScheduler(a.logger, cap.fire)
|
||||
|
||||
meta := map[string]any{
|
||||
"claudeCode": map[string]any{"toolName": "ScheduleRun"},
|
||||
"claudeCode": map[string]any{"toolName": "ScheduleWakeup"},
|
||||
}
|
||||
a.handleRunEvent("sess-1", "tc-1", meta, nil, false)
|
||||
a.handleWakeupEvent("sess-1", "tc-1", meta, nil, false)
|
||||
|
||||
a.mu.Lock()
|
||||
_, present := a.pendingRuns["tc-1"]
|
||||
_, present := a.pendingWakeups["tc-1"]
|
||||
a.mu.Unlock()
|
||||
if !present {
|
||||
t.Fatal("expected pending run before terminal")
|
||||
t.Fatal("expected pending wakeup before terminal")
|
||||
}
|
||||
|
||||
// Terminal arrives without prompt or scheduledFor — should clean up.
|
||||
a.handleRunEvent("sess-1", "tc-1", meta, nil, true)
|
||||
a.handleWakeupEvent("sess-1", "tc-1", meta, nil, true)
|
||||
|
||||
a.mu.Lock()
|
||||
_, present = a.pendingRuns["tc-1"]
|
||||
_, present = a.pendingWakeups["tc-1"]
|
||||
a.mu.Unlock()
|
||||
if present {
|
||||
t.Error("expected pending run cleared on terminal")
|
||||
t.Error("expected pending wakeup cleared on terminal")
|
||||
}
|
||||
if got := atomic.LoadInt32(&cap.calls); got != 0 {
|
||||
t.Errorf("terminal-without-data should not fire, got %d calls", got)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFireRun_SkipsOnSessionChange exercises fireRun's session-guard
|
||||
// TestFireWakeup_SkipsOnSessionChange exercises fireWakeup's session-guard
|
||||
// branch — when the adapter's current session has rotated away from the one
|
||||
// the run was scheduled for, the run must drop instead of trying to
|
||||
// drive an unrelated session.
|
||||
// TestFireRun_SkipsOnSessionChange exercises fireRun's session-guard
|
||||
// branch — when the adapter's current session has rotated away from the one
|
||||
// the run was scheduled for, the run must drop instead of trying to
|
||||
// the wakeup was scheduled for, the wakeup must drop instead of trying to
|
||||
// drive an unrelated session.
|
||||
//
|
||||
// We can't observe Prompt directly (it returns early on nil acpConn without
|
||||
// any side effect on the updates channel) so we use a.pendingContext as a
|
||||
// canary: Prompt's first action under a.mu is to read-and-clear it. If the
|
||||
// guard works, fireRun never spawns the goroutine that calls Prompt and
|
||||
// guard works, fireWakeup never spawns the goroutine that calls Prompt and
|
||||
// the canary survives. If the guard regresses, Prompt runs and clears it.
|
||||
func TestFireRun_SkipsOnSessionChange(t *testing.T) {
|
||||
func TestFireWakeup_SkipsOnSessionChange(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
a := newTestAdapter()
|
||||
t.Cleanup(func() { _ = a.Close() })
|
||||
|
||||
// Adapter currently tracks a different session than the one the run
|
||||
// Adapter currently tracks a different session than the one the wakeup
|
||||
// was scheduled against, and pendingContext holds a canary string.
|
||||
const canary = "guard-canary"
|
||||
a.mu.Lock()
|
||||
@@ -430,7 +434,7 @@ func TestFireRun_SkipsOnSessionChange(t *testing.T) {
|
||||
a.pendingContext = canary
|
||||
a.mu.Unlock()
|
||||
|
||||
a.fireRun("old-sess", "should-not-fire")
|
||||
a.fireWakeup("old-sess", "should-not-fire")
|
||||
|
||||
// Let any spawned goroutine run to completion. Inside synctest, after
|
||||
// Wait() returns the bubble has settled and we can inspect state.
|
||||
Reference in New Issue
Block a user