ref(task-execution): centralize wake scheduling with ensureConversationWake()

[dcramer]

Background

PR #701 fixed a runaway queue loop (29h, ~2M requests, Jun 29–30 2026) caused by completeConversationWork sending an extra wake nudge when scheduleAgentContinue had already sent one mid-run. The fix uses lastEnqueuedAtMs !== undefined as a proxy for "a wake is already in flight."

That fix is correct but exposes a structural smell: the system has five distinct code paths that directly call queue.send + markConversationWorkEnqueued, and completeConversationWork infers wake ownership from an opaque timestamp side effect. There is already a second footgun documented in slack-runtime.ts:

// Mark the inbound mailbox messages as consumed even though we are not
// replying. Without this, completeConversationWork sees pendingMessages > 0
// and re-enqueues indefinitely — an infinite loop for every skipped message.

Unrelated code has to defensively work around completeConversationWork's behavior.

Root Cause

Multiple wake owners + completeConversationWork inferring scheduling intent from distributed state.

Current wake senders:

1. scheduleAgentContinue (from deep inside run(), 4+ levels down)
2. appendAndEnqueueInboundMessage (inbound path)
3. processConversationWork after completeConversationWork returns "pending"
4. Active-deferral path (when lease is already active)
5. Yield / error / lost-lease paths

Proposed Design

Business code marks durable work needed. One scheduler ensures exactly one wake. Queue messages are disposable nudges, never source of truth.

Phase 1 — Hotfix ✅ (shipped as #701)

shouldNudge = runnable && !hasQueuedWake in completeConversationWork. Correct and minimal.

Phase 2 — Centralize wake sending

Introduce one idempotent primitive:

async function ensureConversationWake(args: {
  conversationId: string;
  destination: Destination;
  reason: WakeReason;
  delayMs?: number;
  queue: ConversationWorkQueue;
  state?: StateAdapter;
}): Promise<"enqueued" | "already_enqueued" | "no_work">;

Invariant: if the conversation has runnable work and no outstanding wake, enqueue exactly one. Otherwise do nothing. All five wake senders become callers of this function. Direct queue.send / markConversationWorkEnqueued calls are removed from business logic.

Phase 3 — Replace lastEnqueuedAtMs with explicit wake state

interface ConversationExecution {
  // ...
  wake?: {
    id: string;              // stable per-wake identity
    enqueuedAtMs?: number;   // set after queue.send succeeds
    delayUntilMs?: number;
    reason: WakeReason;
  };
}

wake.id provides idempotent queue idempotency keys: conversation-wake:${conversationId}:${wakeId}. A consumed wake clears wake from state. ensureConversationWake checks wake !== undefined rather than reasoning about timestamps.

Phase 4 — Make completion declarative

Change completeConversationWork from returning scheduling instructions to returning durable facts:

// Before
Promise<"completed" | "lost_lease" | "pending">

// After
Promise<
  | { status: "lost_lease" }
  | { status: "completed"; runnable: boolean; needsRun: boolean; pendingCount: number }
>

Caller:

if (completion.runnable) {
  await ensureConversationWake({ conversationId, destination, reason: "completion_runnable" });
}

This also fixes the pendingMessages footgun — code that intentionally skips messages no longer needs to defensively drain them to avoid infinite requeue loops.

Phase 5 — First-class continuation on ConversationWorkerContext

interface ConversationWorkerContext {
  // ...existing fields...
  requestContinuation(args: AgentContinueRequest): Promise<void>;
}

requestContinuation internally marks durable state + calls ensureConversationWake. reply-executor.ts and agent-continue-runner.ts continue to receive this via deps.services.scheduleAgentContinue — same injection shape, centralized implementation.

How This Eliminates the Bug Class

Before:

scheduleAgentContinue()    // sends queue wake (side effect on shared state)
completeConversationWork() // infers: "was a wake already sent?" from timestamps

After:

requestContinuation()      // marks runnable, calls ensureConversationWake()
completeConversationWork() // reports runnable=true
processConversationWork()  // calls ensureConversationWake()
// → idempotent; both calls coalesce

The invariant becomes local and testable: ensureConversationWake is idempotent for one runnable conversation.

Required Invariants (to document + test)

1. Queue messages are nudges only — durable state is source of truth.
2. At most one outstanding wake per conversation (unless delayed/retried).
3. Any code may call ensureConversationWake; duplicate calls are safe.
4. A consumed wake clears outstanding wake state.
5. If a wake is consumed while a lease is active, replacement deferral must either succeed or the current delivery must retry (not silently ack).
6. Business logic never calls queue.send directly.
7. Completion never infers wake ownership from timestamps.

Scope

Each phase ships independently. Phase 2 (centralize wake sending) is the highest-leverage next step after the hotfix. Phases 3–5 are structural cleanup that can follow.

Refs #701

View Session in Sentry

[dcramer]

Closing this as fixed by #708 / bb620e3. The original Phase 2 scope, centralizing wake scheduling behind ensureConversationWake, is now implemented on main.

The broader explicit wake-state cleanup (wake object replacing lastEnqueuedAtMs, and making completion fully declarative) can be tracked separately if we still want that refactor.