From 2358bc07f3a4dee6d45c179c06a836a592878c26 Mon Sep 17 00:00:00 2001 From: twaldin Date: Fri, 7 Aug 2026 18:59:49 -0700 Subject: [PATCH 1/2] fix(coding-agent): resume suspended input pump on programmatic admission Any abort (requestAbort, abortForUpdateRestart) suspends the session input pump. A user-typed prompt clears the suspension, but programmatic admission via queueAgentMessagePrompt/_queuePreparedPrompt never does, so queued agent messages, heartbeats, and wake prompts starve forever while the session sits idle. Clear the suspension on programmatic turn admission when not streaming, mirroring _prompt(). A starved queue also blocked every heartbeat fire, and a skipped heartbeat lost the fire until the next full interval because nextRunAt was recomputed from the skip time. A skipped heartbeat now retries at the next scheduler tick. Regression tests: queued agent mail post-abort delivers at idle, and a pre-abort backlog drains once a later programmatic admission arrives. Both fail on the previous behavior. --- .../coding-agent/src/core/agent-session.ts | 13 +++++ packages/coding-agent/src/core/cron-jobs.ts | 12 ++++- packages/coding-agent/test/cron-jobs.test.ts | 4 +- .../resume-suspended-input-pump.test.ts | 50 +++++++++++++++++++ 4 files changed, 77 insertions(+), 2 deletions(-) create mode 100644 packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index e99462eabb..ff6c3c679b 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -5338,6 +5338,19 @@ export class AgentSession { if (action.payload.kind === "turn" && action.wake === "immediate") this._sessionInputPumpSuspended = false; this._scheduleSessionInputPump(); } + if ( + !options.restore && + options.wake !== false && + action.payload.kind === "turn" && + !this.isStreaming && + this._sessionInputPumpSuspended + ) { + // Programmatic admissions resume a pump suspended by an abort, matching _prompt(): + // otherwise queued agent messages, heartbeats, and wake prompts starve at idle. + this._sessionInputPumpSuspended = false; + this._notifySessionInputCheckpointChange(); + this._scheduleSessionInputPump(); + } return { accepted: true, disposition, ticket: controller.ticket }; } diff --git a/packages/coding-agent/src/core/cron-jobs.ts b/packages/coding-agent/src/core/cron-jobs.ts index 999e68659d..f292fc80c3 100644 --- a/packages/coding-agent/src/core/cron-jobs.ts +++ b/packages/coding-agent/src/core/cron-jobs.ts @@ -729,7 +729,17 @@ export class AgentCronJobStore { return job; } if (result.outcome === "skipped" && result.error === undefined) { - const nextRunAt = nextRunAtForSchedule(job.schedule, now); + const scheduledNextRunAt = nextRunAtForSchedule(job.schedule, now); + // A deferred heartbeat fire retries at the next scheduler tick instead of + // losing the fire until the following full interval. + const heartbeatRetryAt = + isHeartbeatCronJob(job) && job.schedule.kind !== "once" + ? new Date(now.getTime() + ONE_MINUTE_MS) + : undefined; + const nextRunAt = + heartbeatRetryAt && (!scheduledNextRunAt || heartbeatRetryAt < scheduledNextRunAt) + ? heartbeatRetryAt + : scheduledNextRunAt; updated = { ...job, status: job.schedule.kind === "once" ? "completed" : job.status, diff --git a/packages/coding-agent/test/cron-jobs.test.ts b/packages/coding-agent/test/cron-jobs.test.ts index 838dd688e4..440ccc262c 100644 --- a/packages/coding-agent/test/cron-jobs.test.ts +++ b/packages/coding-agent/test/cron-jobs.test.ts @@ -1128,10 +1128,12 @@ describe("AgentCronScheduler", () => { const handled = await scheduler.runDue(new Date("2026-01-01T12:39:00.000Z")); expect(handled).toBe(0); + // A skipped heartbeat fire retries at the next scheduler tick instead of + // waiting out the whole interval. expect(store.getHeartbeat("active-1")).toMatchObject({ id: job.id, status: "active", - nextRunAt: "2026-01-01T12:45:00.000Z", + nextRunAt: "2026-01-01T12:41:00.000Z", lastSkippedAt: "2026-01-01T12:40:00.000Z", runCount: 0, }); diff --git a/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts b/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts new file mode 100644 index 0000000000..abd5a092ec --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts @@ -0,0 +1,50 @@ +import { fauxAssistantMessage } from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createHarness, getAssistantTexts, getUserTexts, type Harness } from "../harness.js"; + +describe("suspended input pump resumes on programmatic admission", () => { + const harnesses: Harness[] = []; + + afterEach(() => { + while (harnesses.length > 0) { + harnesses.pop()?.cleanup(); + } + }); + + it("delivers a queued agent message at idle after an abort suspended the pump", async () => { + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("agent mail handled")]); + + // An abort at idle leaves the input pump suspended. + harness.session.requestAbort(); + await harness.session.agent.waitForIdle(); + + await harness.session.queueAgentMessagePrompt("queued agent mail", "followUp"); + + await vi.waitFor(() => expect(getAssistantTexts(harness)).toContain("agent mail handled")); + expect(getUserTexts(harness)).toContain("queued agent mail"); + }); + + it("delivers items queued before an abort once a later programmatic admission arrives", async () => { + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("backlog handled"), fauxAssistantMessage("second handled")]); + + const pause = harness.session.acquireQueuedWorkPause(); + await harness.session.queueAgentMessagePrompt("queued before abort", "followUp"); + harness.session.requestAbort(); + pause.release(); + await harness.session.agent.waitForIdle(); + + // Without the fix, the backlog starves forever: only a user-typed prompt or + // resumeQueuedWork() revives delivery. + await harness.session.queueAgentMessagePrompt("queued after abort", "followUp"); + + await vi.waitFor(() => { + const users = getUserTexts(harness); + expect(users).toContain("queued before abort"); + expect(users).toContain("queued after abort"); + }); + }); +}); From 9f77b4d0a8b8def2014539f22ea304994b503e59 Mon Sep 17 00:00:00 2001 From: twaldin Date: Fri, 7 Aug 2026 23:28:58 -0700 Subject: [PATCH 2/2] fix(coding-agent): resume suspended pump on coalesced follow-up re-fire A follow-up re-fire that coalesces into an already-queued owner (same queueKey, e.g. rolling wake prompts or heartbeats) was dropped before the pump-resume path ran, so the queued owner still starved after an abort suspended the pump at idle. Resume the pump on the coalesced path too. --- .../coding-agent/src/core/agent-session.ts | 7 +++++++ .../resume-suspended-input-pump.test.ts | 19 +++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index ff6c3c679b..0c215df924 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -5304,6 +5304,13 @@ export class AgentSession { } const coalescedOwner = options.restore ? undefined : this._coalescedFollowUpOwner(action); if (coalescedOwner) { + // A coalesced re-fire is still evidence a producer wants delivery: resume a + // pump suspended by an abort so the owning queued action can drain at idle. + if (options.wake !== false && !this.isStreaming && this._sessionInputPumpSuspended) { + this._sessionInputPumpSuspended = false; + this._notifySessionInputCheckpointChange(); + this._scheduleSessionInputPump(); + } if (action.agentMessageId !== coalescedOwner.agentMessageId) { this._rejectAgentMessage( action.agentMessageId, diff --git a/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts b/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts index abd5a092ec..f29a023eed 100644 --- a/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts +++ b/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts @@ -47,4 +47,23 @@ describe("suspended input pump resumes on programmatic admission", () => { expect(users).toContain("queued after abort"); }); }); + + it("resumes the pump when a coalesced follow-up re-fires at idle", async () => { + const harness = await createHarness(); + harnesses.push(harness); + harness.setResponses([fauxAssistantMessage("wake handled")]); + + const pause = harness.session.acquireQueuedWorkPause(); + await harness.session.followUp("wake fire", undefined, { queueKey: "wake:test" }); + harness.session.requestAbort(); + pause.release(); + await harness.session.agent.waitForIdle(); + + // A re-fire with the same queueKey coalesces into the queued owner. Without the + // fix it is dropped without resuming the pump, so the owner starves forever. + await harness.session.followUp("wake fire", undefined, { queueKey: "wake:test" }); + + await vi.waitFor(() => expect(getUserTexts(harness)).toContain("wake fire")); + expect(getAssistantTexts(harness)).toContain("wake handled"); + }); });