fix(sidecar): resume Gmail history past page cap when pages are deduped

test/durableObject/sidecar-state.test.ts

@@ -39,6 +39,7 @@ function makeSqlLike(): SqlLike {
         CREATE TABLE sidecar_state (
             id INTEGER PRIMARY KEY CHECK (id = 1),
             history_cursor TEXT,
+            history_page_token TEXT,
             access_token TEXT,
             token_expires_at INTEGER,
             label_ids TEXT,

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

@@ -64,9 +64,10 @@ function makeEnv(stub: ReturnType<typeof makeStub>) {
 const ctx = { waitUntil: vi.fn() } as never;
 
 /** Cached-token state so no token minting happens in most tests. */
-function freshState(cursor: string | null) {
+function freshState(cursor: string | null, pageToken: string | null = null) {
 	return {
 		history_cursor: cursor,
+		history_page_token: pageToken,
 		access_token: "cached-tok",
 		token_expires_at: Date.now() + 3600_000,
 		label_ids: null,
@@ -666,4 +667,68 @@ describe("pollWorkspaceMailbox", () => {
 		expect(r.processed).toBeGreaterThan(0);
 		expect(stub.putSidecarState.mock.calls.at(-1)![0].history_cursor).toBeUndefined();
 	});
+
+	it("truncated history with all pages deduped saves history_page_token to resume past the cap (#667)", async () => {
+		// Deadlock scenario: pages 1–3 fully deduped, page 4+ never fetched.
+		// Poll 1: truncated listing, all ids deduped → save resume token.
+		// Poll 2: resume from saved token, tail id ingested, cursor advances.
+		const stub = makeStub(freshState("100"));
+		mockedReceive.mockResolvedValue({ messageId: "local-tail", verdict: null });
+		let historyCalls = 0;
+		gmailFetch({
+			"/history": (u) => {
+				historyCalls += 1;
+				const pageToken = u.searchParams.get("pageToken");
+				if (!pageToken) {
+					return new Response(JSON.stringify({
+						historyId: "200",
+						nextPageToken: "p2",
+						history: [{ messagesAdded: [{ message: { id: "g1", labelIds: ["INBOX"] } }] }],
+					}), { status: 200 });
+				}
+				if (pageToken === "p2") {
+					return new Response(JSON.stringify({
+						historyId: "250",
+						nextPageToken: "p3",
+						history: [{ messagesAdded: [{ message: { id: "g2", labelIds: ["INBOX"] } }] }],
+					}), { status: 200 });
+				}
+				if (pageToken === "p3") {
+					return new Response(JSON.stringify({
+						historyId: "280",
+						nextPageToken: "p4",
+						history: [{ messagesAdded: [{ message: { id: "g3", labelIds: ["INBOX"] } }] }],
+					}), { status: 200 });
+				}
+				if (pageToken === "p4") {
+					return new Response(JSON.stringify({
+						historyId: "300",
+						history: [{ messagesAdded: [{ message: { id: "g-tail", labelIds: ["INBOX"] } }] }],
+					}), { status: 200 });
+				}
+				throw new Error(`unexpected pageToken ${pageToken}`);
+			},
+			"/messages/": (u) => {
+				const id = u.pathname.split("/").pop()!;
+				return new Response(JSON.stringify({ id, raw: rawMessage(`${id}@x`, "s") }), { status: 200 });
+			},
+		});
+		stub.findEmailIdByMessageId.mockImplementation(async (msgId: string) =>
+			["g1@x", "g2@x", "g3@x"].includes(msgId) ? `existing-${msgId}` : null,
+		);
+
+		const r1 = await pollWorkspaceMailbox(makeEnv(stub), ctx, "user@tenant.example", CFG);
+		expect(r1).toEqual({ processed: 0, deduped: 3, error: null });
+		const patch1 = stub.putSidecarState.mock.calls.at(-1)![0];
+		expect(patch1.history_cursor).toBeUndefined();
+		expect(patch1.history_page_token).toBe("p4");
+
+		stub.getSidecarState.mockResolvedValue(freshState("100", "p4"));
+		const r2 = await pollWorkspaceMailbox(makeEnv(stub), ctx, "user@tenant.example", CFG);
+		expect(r2).toEqual({ processed: 1, deduped: 0, error: null });
+		expect(historyCalls).toBe(4); // 3 pages on poll 1 + 1 resumed page on poll 2
+		const patch2 = stub.putSidecarState.mock.calls.at(-1)![0];
+		expect(patch2.history_cursor).toBe("300");
+		expect(patch2.history_page_token).toBeNull();
+	});
 });

workers/durableObject/migrations.ts

@@ -656,6 +656,14 @@ export const mailboxMigrations: Migration[] = [
             CREATE INDEX IF NOT EXISTS idx_sidecar_audit_gmail_message_id ON sidecar_audit(gmail_message_id);
         `,
 	},
+	{
+		// History page-cap resume (#667): when Gmail history spans more than
+		// MAX_HISTORY_PAGES and pages 1–3 are fully deduped, the cursor stays
+		// frozen (truncated=true) and the poller never reaches page 4+.
+		// Persist the dangling pageToken so the next tick resumes past the cap.
+		name: "32_history_page_token",
+		sql: `ALTER TABLE sidecar_state ADD COLUMN history_page_token TEXT;`,
+	},
 ];
 
 /**

workers/durableObject/sidecar-state.ts

@@ -21,6 +21,8 @@ import type { SqlLike } from "./catchall-intel";
 
 export interface SidecarStateRow {
 	history_cursor: string | null;
+	/** Gmail history.list pageToken resume point when the page cap truncates a listing. */
+	history_page_token: string | null;
 	access_token: string | null;
 	token_expires_at: number | null;
 	label_ids: string | null;
@@ -37,7 +39,7 @@ export interface SidecarStateRow {
 }
 
 const STATE_COLUMNS = [
-	"history_cursor", "access_token", "token_expires_at",
+	"history_cursor", "history_page_token", "access_token", "token_expires_at",
 	"label_ids", "last_poll_at", "last_error", "consecutive_failures",
 	"poll_lease_until", "label_error", "label_failure_count",
 ] as const;

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,21 @@ 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,
+	resumePageToken?: string | null,
+): Promise<HistoryResult> {
 	const ids: string[] = [];
 	const seen = new Set<string>();
 	let latestHistoryId = startHistoryId;
-	let pageToken: string | undefined;
+	let pageToken: string | undefined = resumePageToken ?? undefined;
 	// 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.
+	// When truncated, `nextPageToken` is the resume point for the next fetch.
 	let truncated = false;
+	let nextPageToken: string | undefined;
 	for (let page = 0; page < MAX_HISTORY_PAGES; page++) {
 		const qs = new URLSearchParams({
 			startHistoryId,
@@ -190,9 +196,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;
+			nextPageToken = data.nextPageToken;
+		}
 	}
-	return { ok: true, messageIds: ids, historyId: latestHistoryId, truncated };
+	return {
+		ok: true,
+		messageIds: ids,
+		historyId: latestHistoryId,
+		truncated,
+		...(nextPageToken ? { nextPageToken } : {}),
+	};
 }
 
 export async function getRawMessage(token: string, id: string): Promise<Uint8Array> {

workers/providers/workspace.ts

@@ -149,7 +149,8 @@ export interface PollResult { processed: number; deduped: number; error: string
 
 interface SidecarStub {
 	getSidecarState(): Promise<{
-		history_cursor: string | null; access_token: string | null; token_expires_at: number | null;
+		history_cursor: string | null; history_page_token: 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;
 	} | null>;
@@ -192,7 +193,8 @@ export async function pollWorkspaceMailbox(
 ): Promise<PollResult> {
 	const stub = env.MAILBOX.get(env.MAILBOX.idFromName(mailboxId)) as unknown as SidecarStub;
 	const state = (await stub.getSidecarState()) ?? {
-		history_cursor: null, access_token: null, token_expires_at: null,
+		history_cursor: null, history_page_token: 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,
 	};
@@ -253,7 +255,11 @@ 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,
+			state.history_page_token,
+		);
 		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).
@@ -272,6 +278,7 @@ export async function pollWorkspaceMailbox(
 				detail: "cursor expired past Gmail history retention; mail arriving in the gap was never scored",
 			});
 			patch.history_cursor = profile.historyId;
+			patch.history_page_token = null;
 			patch.last_error = "history gap: cursor expired; monitoring reinitialized from current historyId";
 			await stub.putSidecarState(patch);
 			return { processed: 0, deduped: 0, error: null };
@@ -378,7 +385,20 @@ 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 `history_page_token` once the current window
+		// is exhausted so the next poll resumes past the page cap instead of
+		// wedging when pages 1–3 are fully deduped (#667).
+		if (!hitCap && !history.truncated) {
+			patch.history_cursor = history.historyId;
+			patch.history_page_token = null;
+		} else if (
+			!hitCap &&
+			history.truncated &&
+			history.nextPageToken &&
+			(processed + deduped > 0 || history.messageIds.length === 0)
+		) {
+			patch.history_page_token = history.nextPageToken;
+		}
 		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