diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index e99462eabb..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, @@ -5338,6 +5345,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..f29a023eed --- /dev/null +++ b/packages/coding-agent/test/suite/regressions/resume-suspended-input-pump.test.ts @@ -0,0 +1,69 @@ +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"); + }); + }); + + 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"); + }); +});