fix(sidecar): resume truncated Gmail history via page token (#595)

test/durableObject/sidecar-state.test.ts

@@ -47,7 +47,8 @@ function makeSqlLike(): SqlLike {
             consecutive_failures INTEGER NOT NULL DEFAULT 0,
             poll_lease_until INTEGER,
             label_error TEXT,
-            label_failure_count INTEGER NOT NULL DEFAULT 0
+            label_failure_count INTEGER NOT NULL DEFAULT 0,
+            history_page_token TEXT
         );
 
         CREATE TABLE sidecar_audit (

tests/providers/gmail-client.test.ts

@@ -153,7 +153,7 @@ describe("listNewMessageIds", () => {
 		}));
 		const { listNewMessageIds } = await import("../../workers/providers/gmail-client");
 		const r = await listNewMessageIds("tok", "100");
-		expect(r).toEqual({ ok: true, messageIds: ["m1", "m2", "m3"], historyId: "300", truncated: true });
+		expect(r).toEqual({ ok: true, messageIds: ["m1", "m2", "m3"], historyId: "300", truncated: true, nextPageToken: "p3" });
 		expect(pages).toBe(3); // MAX_HISTORY_PAGES — the 4th page is never fetched
 	});
 

tests/providers/workspace-poll.test.ts

@@ -76,6 +76,7 @@ function freshState(cursor: string | null) {
 		poll_lease_until: null,
 		label_error: null,
 		label_failure_count: 0,
+		history_page_token: null,
 	};
 }
 
@@ -666,4 +667,62 @@ describe("pollWorkspaceMailbox", () => {
 		expect(r.processed).toBeGreaterThan(0);
 		expect(stub.putSidecarState.mock.calls.at(-1)![0].history_cursor).toBeUndefined();
 	});
+
+	it("truncated history deadlock (#595): all deduped pages save page token, next poll resumes and advances cursor", async () => {
+		// Poll 1: pages 1–3 are already ingested (deduped); page 4 has fresh mail.
+		// Without a continuation token the poller would re-list pages 1–3 forever.
+		const stub1 = makeStub(freshState("100"));
+		stub1.findEmailIdByMessageId.mockImplementation(async (mid: string) =>
+			(["g1@x", "g2@x", "g3@x"].includes(mid) ? "already" : null),
+		);
+		let pages1 = 0;
+		gmailFetch({
+			"/history": () => {
+				pages1 += 1;
+				return new Response(JSON.stringify({
+					historyId: "999",
+					nextPageToken: `p${pages1}`,
+					history: [{ messagesAdded: [{ message: { id: `g${pages1}`, labelIds: ["INBOX"] } }] }],
+				}), { status: 200 });
+			},
+			"/messages/": (u) => {
+				const id = u.pathname.split("/").pop()!;
+				return new Response(JSON.stringify({ id, raw: rawMessage(`${id}@x`, "s") }), { status: 200 });
+			},
+		});
+		const r1 = await pollWorkspaceMailbox(makeEnv(stub1), ctx, "user@tenant.example", CFG);
+		expect(r1).toMatchObject({ processed: 0, deduped: 3, error: null });
+		const patch1 = stub1.putSidecarState.mock.calls.at(-1)![0];
+		expect(patch1.history_cursor).toBeUndefined();
+		expect(patch1.history_page_token).toBe("p3");
+
+		// Poll 2: resume from p3 — only g4 is fresh; listing completes.
+		const stub2 = makeStub({ ...freshState("100"), history_page_token: "p3" });
+		stub2.findEmailIdByMessageId.mockImplementation(async (mid: string) =>
+			(["g1@x", "g2@x", "g3@x"].includes(mid) ? "already" : null),
+		);
+		mockedReceive.mockResolvedValue({ messageId: "local-g4", verdict: null });
+		let pages2 = 0;
+		gmailFetch({
+			"/history": (u) => {
+				pages2 += 1;
+				const token = u.searchParams.get("pageToken");
+				expect(token).toBe(pages2 === 1 ? "p3" : `tail${pages2}`);
+				if (pages2 === 1) {
+					return new Response(JSON.stringify({
+						historyId: "999",
+						nextPageToken: "tail2",
+						history: [{ messagesAdded: [{ message: { id: "g4", labelIds: ["INBOX"] } }] }],
+					}), { status: 200 });
+				}
+				return new Response(JSON.stringify({ historyId: "1000", history: [] }), { status: 200 });
+			},
+			"/messages/g4": () => new Response(JSON.stringify({ id: "g4", raw: rawMessage("g4@x", "fresh") }), { status: 200 }),
+		});
+		const r2 = await pollWorkspaceMailbox(makeEnv(stub2), ctx, "user@tenant.example", CFG);
+		expect(r2).toMatchObject({ processed: 1, deduped: 0, error: null });
+		const patch2 = stub2.putSidecarState.mock.calls.at(-1)![0];
+		expect(patch2.history_page_token).toBeNull();
+		expect(patch2.history_cursor).toBe("1000");
+	});
 });

workers/durableObject/migrations.ts

@@ -656,6 +656,16 @@ export const mailboxMigrations: Migration[] = [
             CREATE INDEX IF NOT EXISTS idx_sidecar_audit_gmail_message_id ON sidecar_audit(gmail_message_id);
         `,
 	},
+	{
+		// Gmail history page continuation (issue #595). `listNewMessageIds` caps
+		// at MAX_HISTORY_PAGES; when the listing is truncated the cursor must NOT
+		// advance, but if pages 1–3 are already deduped the poller would otherwise
+		// re-fetch them forever and never reach page 4+. `history_page_token`
+		// stores the dangling `nextPageToken` once the current slice is fully
+		// worked; the next poll resumes from that token with the same cursor.
+		name: "32_history_page_token",
+		sql: `ALTER TABLE sidecar_state ADD COLUMN history_page_token TEXT;`,
+	},
 ];
 
 /**

workers/durableObject/sidecar-state.ts

@@ -34,12 +34,15 @@ export interface SidecarStateRow {
 	label_error: string | null;
 	/** Failed label writes accumulated since the last successful one (#590). */
 	label_failure_count: number;
+	/** Gmail history `nextPageToken` when a listing hit the page cap (#595). */
+	history_page_token: string | null;
 }
 
 const STATE_COLUMNS = [
 	"history_cursor", "access_token", "token_expires_at",
 	"label_ids", "last_poll_at", "last_error", "consecutive_failures",
 	"poll_lease_until", "label_error", "label_failure_count",
+	"history_page_token",
 ] as const;
 
 export function _getSidecarStateImpl(sql: SqlLike): SidecarStateRow | null {

workers/providers/gmail-client.ts

@@ -140,7 +140,7 @@ export async function getProfile(token: string): Promise<{ emailAddress: string;
 }
 
 export type HistoryResult =
-	| { ok: true; messageIds: string[]; historyId: string; truncated: boolean }
+	| { ok: true; messageIds: string[]; historyId: string; truncated: boolean; nextPageToken?: string }
 	| { ok: false; expired: true };
 
 /** Gmail-internal labels that mark non-inbound messages we must never score. */
@@ -152,15 +152,20 @@ const MAX_HISTORY_PAGES = 3;
  * cursor is older than Gmail's history retention — the caller must
  * re-initialize from getProfile() and accept the gap.
  */
-export async function listNewMessageIds(token: string, startHistoryId: string): Promise<HistoryResult> {
+export async function listNewMessageIds(
+	token: string,
+	startHistoryId: string,
+	opts?: { startPageToken?: string },
+): Promise<HistoryResult> {
 	const ids: string[] = [];
 	const seen = new Set<string>();
 	let latestHistoryId = startHistoryId;
-	let pageToken: string | undefined;
+	let pageToken: string | undefined = opts?.startPageToken;
 	// truncated = we stopped on the page cap with more pages still pending, so
 	// the listing is INCOMPLETE. The caller must NOT advance the cursor on a
 	// truncated result — otherwise the un-fetched tail is skipped forever.
 	let truncated = false;
+	let danglingPageToken: string | undefined;
 	for (let page = 0; page < MAX_HISTORY_PAGES; page++) {
 		const qs = new URLSearchParams({
 			startHistoryId,
@@ -190,9 +195,18 @@ export async function listNewMessageIds(token: string, startHistoryId: string):
 		pageToken = data.nextPageToken;
 		// If this was the last iteration the loop allows but a page still
 		// dangles, the listing is truncated by the page cap.
-		if (page === MAX_HISTORY_PAGES - 1) truncated = true;
+		if (page === MAX_HISTORY_PAGES - 1) {
+			truncated = true;
+			danglingPageToken = data.nextPageToken;
+		}
 	}
-	return { ok: true, messageIds: ids, historyId: latestHistoryId, truncated };
+	return {
+		ok: true,
+		messageIds: ids,
+		historyId: latestHistoryId,
+		truncated,
+		...(truncated && danglingPageToken ? { nextPageToken: danglingPageToken } : {}),
+	};
 }
 
 export async function getRawMessage(token: string, id: string): Promise<Uint8Array> {

workers/providers/workspace.ts

@@ -152,6 +152,7 @@ interface SidecarStub {
 		history_cursor: string | null; access_token: string | null; token_expires_at: number | null;
 		label_ids: string | null; last_poll_at: number | null; last_error: string | null; consecutive_failures: number;
 		poll_lease_until: number | null; label_error: string | null; label_failure_count: number;
+		history_page_token: string | null;
 	} | null>;
 	acquirePollLease(nowMs: number, ttlMs: number): Promise<boolean>;
 	putSidecarState(patch: Record<string, unknown>): Promise<void>;
@@ -195,6 +196,7 @@ export async function pollWorkspaceMailbox(
 		history_cursor: null, access_token: null, token_expires_at: null,
 		label_ids: null, last_poll_at: null, last_error: null, consecutive_failures: 0,
 		poll_lease_until: null, label_error: null, label_failure_count: 0,
+		history_page_token: null,
 	};
 
 	// Backoff gate (rule 2). Sits BEFORE the lease so a backed-off mailbox
@@ -253,7 +255,9 @@ export async function pollWorkspaceMailbox(
 			return { processed: 0, deduped: 0, error: null };
 		}
 
-		const history = await listNewMessageIds(token, state.history_cursor);
+		const history = await listNewMessageIds(token, state.history_cursor, {
+			startPageToken: state.history_page_token ?? undefined,
+		});
 		if (!history.ok) {
 			// Rule 5: cursor older than Gmail's history retention. Re-anchor and
 			// record the gap — informational, NOT a failure (failures gate backoff).
@@ -378,7 +382,18 @@ export async function pollWorkspaceMailbox(
 
 		// Rule 6: advance the cursor ONLY when the listing was complete (not
 		// page-cap truncated) AND we worked through all of it (not batch-capped).
-		if (!hitCap && !history.truncated) patch.history_cursor = history.historyId;
+		// When truncated, persist the dangling page token once the current slice
+		// is fully worked so the next poll can resume past the page cap (#595).
+		// A batch cap mid-slice clears the token so the next poll restarts from
+		// page 1 of the same cursor and can finish unprocessed ids.
+		if (!history.truncated) {
+			patch.history_page_token = null;
+			if (!hitCap) patch.history_cursor = history.historyId;
+		} else if (!hitCap && history.nextPageToken) {
+			patch.history_page_token = history.nextPageToken;
+		} else if (hitCap) {
+			patch.history_page_token = null;
+		}
 		if (labelIds) patch.label_ids = JSON.stringify(labelIds);
 		// Label failures don't freeze the cursor (ingest succeeded — the cursor
 		// rules above stand), but they DO surface in health so a wrong-scope