fix(coding-agent): resume suspended input pump on programmatic admission

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 };
 	}
 

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,

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,
 		});

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