From 4ef018cfe65e96b93c4310e15e479fd4425fad8f Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:17:14 -0500 Subject: [PATCH 1/6] test: model retention reads in native cleanup transaction fixture --- .../native-runtime/native-session-executor.test.ts | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index 4dabeda6954..6fb2e52c777 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -4302,6 +4302,17 @@ describe("retained native cleanup activation", () => { releaseCommit = release; }); const db = { + execute: async (query: SQL) => { + // Retention hints read the fixture's run/issue relationship inside the + // same cleanup transaction; this fixture has one issue and no ancestors. + const statement = new PgDialect().sqlToQuery(query); + expect(transactionOpen).toBe(true); + expect(statement.params).toContain(run.companyId); + if (statement.sql.includes("from heartbeat_runs")) return [{ id: run.nativeIssueId }]; + if (statement.sql.includes("origin_kind")) return [{ id: run.nativeIssueId, originKind: null, originId: null }]; + if (statement.sql.includes("with recursive affected")) return [{ issueId: run.nativeIssueId }]; + throw new Error(`Unexpected cleanup fixture query: ${statement.sql}`); + }, select: () => ({ from: (table: unknown) => { const rows = From f201195a92a3f6baa6e4411aa52da39f93f07d83 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:20:10 -0500 Subject: [PATCH 2/6] test: await committed recovery and adapter admission outcomes --- .../src/__tests__/chat-channels.integration.test.ts | 12 +++++++++--- .../heartbeat-stale-queue-invalidation.test.ts | 8 ++++++++ 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index fcaf642aa84..8b10fe2eec7 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -16341,9 +16341,15 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await tx.update(toolConnections).set({ credentialSecretRefs: connection.credentialSecretRefs.map(ref => ({ ...ref, versionSelector: ref.versionSelector === "1" ? "latest" : "1" })) }).where(eq(toolConnections.id, connection.id)); }); - await expect.poll(async () => ((await f.checkpoint()).value as Record).credentialFingerprint, - { timeout: 5000 }).not.toBe(oldWindow.credentialFingerprint); - expect((await f.checkpoint()).value).toMatchObject({ generation: oldWindow.generation, outcome: "epoch_initialized" }); + // Credential synchronization and epoch initialization commit separately. + // Require the final epoch receipt, not just the intermediate fingerprint. + await expect.poll(async () => { + const value = (await f.checkpoint()).value as Record; + return { ...value, fingerprintChanged: typeof value.credentialFingerprint === "string" + && value.credentialFingerprint !== oldWindow.credentialFingerprint }; + }, { timeout: 5000 }).toMatchObject({ + fingerprintChanged: true, generation: oldWindow.generation, outcome: "epoch_initialized", + }); expect(f.requests).toEqual([]); } finally { await coordinator.stop(); } }); diff --git a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts index 65191c24d2f..c4f0772ad47 100644 --- a/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts +++ b/server/src/__tests__/heartbeat-stale-queue-invalidation.test.ts @@ -262,6 +262,14 @@ describeEmbeddedPostgres("heartbeat stale queued-run invalidation", () => { await heartbeat.resumeQueuedRuns(); const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)); expect(delivered.runId).toBeTruthy(); + // The receipt can commit before execution enters liveRunExecutions. + // Observe actual adapter admission before awaiting its drain. + await vi.waitFor(async () => { + const current = await heartbeat.getRun(delivered.runId!); + expect(countExecuteCallsForRun(delivered.runId!), JSON.stringify({ + status: current?.status, errorCode: current?.errorCode, + })).toBe(1); + }, { timeout: 5_000 }); await heartbeat.waitForRunExecutionDrain(delivered.runId!); expect(countExecuteCallsForRun(delivered.runId!)).toBe(1); expect((await heartbeat.getRun(delivered.runId!))?.contextSnapshot).toMatchObject({ wakeCommentIds: [comment.id] }); From 71d7b9f09e5a1a0c5c38bbd4084802dfd52f313d Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:35:20 -0500 Subject: [PATCH 3/6] fix(server): bound blocker facts and preserve origin authority Resolve liveness/stranded references through trusted company issue rows, including valid none sentinels. Bound shared blocker nodes, edges and query results; return private deferred-root completeness so retention immediately performs finite exact source checks. Preserve already-observed live work, completed-root classification and true deadlines. Cover historical producer metadata, moved recovery tasks, many-heavy graph limits, quiet rechecks and faithful transaction fixtures. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 +- .../attention-retention-work.test.ts | 126 ++++++++++++++++-- server/src/__tests__/issues-service.test.ts | 2 +- server/src/__tests__/work-products.test.ts | 28 +++- server/src/services/attention-retention.ts | 11 +- .../issue-thread-interactions.test.ts | 19 ++- server/src/services/issues.ts | 120 +++++++++-------- .../services/retention-work-notifications.ts | 19 ++- 8 files changed, 250 insertions(+), 77 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index 24212b4b98a..aa78f1bb746 100644 --- a/doc/plans/2026-10-30-attention-retention-sleep.md +++ b/doc/plans/2026-10-30-attention-retention-sleep.md @@ -25,7 +25,7 @@ The feed and worker share confirmation collapse, newer-run probes, issue visibil ## Existing mutation boundaries reactivate work -Committed hints identify a company, source kind, source ID or changed issue where those values are known. Routine issue writers pass identity to the retention domain helper before their mutation. The helper captures old and new parent paths, reverse blocker ancestors, run-linked issues and liveness-origin references; issue-scoped source queries then select reviews, blockers, interactions, decisions, recovery actions and failed runs by their issue identity. Budget incidents and agent errors remain tied to their own source changes. Ordinary issue writes do not reconstruct the unrelated company feed. Queue keys explicitly pick supported identity fields. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Decisions, recovery source links, runs, eligible agents, issue visibility, newer-run probes and blocker graphs also load in company batches per page. A 100-source decision or recovery page uses at most three statements; 100 failed runs use at most six; 100 blocker roots use at most 30 graph/source statements plus retention-state reads for their emitted entries. These evidence maps exist only for one load. The blocker owner shares fetched node and edge facts while tracking depth, node count, expanded nodes and live-work reachability independently for each root. An unrelated graph reaching its limit cannot change a shallow root’s terminal subject or deadline. All later graph queries, including recovery source IDs, agent IDs and workspace-finalize inputs, use bounded chunks. A SQL-generated 70,000-node graph proves PostgreSQL parameter limits are respected at the driver boundary. Archive checks still lock and freshly re-read sources, terminal subjects and related eligibility. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included. +Committed hints identify a company, source kind, source ID or changed issue where those values are known. Routine issue writers pass identity to the retention domain helper before their mutation. The helper captures old and new parent paths, reverse blocker ancestors, run-linked issues and liveness-origin references; issue-scoped source queries then select reviews, blockers, interactions, decisions, recovery actions and failed runs by their issue identity. Budget incidents and agent errors remain tied to their own source changes. Ordinary issue writes do not reconstruct the unrelated company feed. Opaque liveness and stranded-recovery origin references resolve through company-scoped issue rows before entering UUID queries, so valid none sentinels do not roll back writes and moved recovery tasks still invalidate their original review. Queue keys explicitly pick supported identity fields. Retention's own writes use exact source hints; blocker retention changes invalidate the company/source because multiple blocked roots can project one terminal subject. Reconstruction pages and due batches remain bounded. Review attention and interaction invokability are computed in company batches per candidate page; a full 100-candidate review page uses at most 20 statements and 100 confirmations use at most six. Decisions, recovery source links, runs, eligible agents, issue visibility, newer-run probes and blocker graphs also load in company batches per page. A 100-source decision or recovery page uses at most three statements; 100 failed runs use at most six; 100 blocker roots use at most 30 graph/source statements plus retention-state reads for their emitted entries. These evidence maps exist only for one load. The blocker owner shares fetched node and edge facts while tracking depth, node count, expanded nodes and live-work reachability independently for each root. The shared graph fetch stores at most 2,000 nodes and 4,000 edges, and bounds each frontier query before materializing rows. An internal batch result identifies roots whose evaluation is incomplete; the legacy map API exposes no extra transport fields. Retention schedules only those roots for an immediate fresh single-root check, which disarms or saves the actual source deadline. It never archives a partial intermediary or adds a recurring incomplete-root poll. Fully evaluated roots preserve their exact classification. All later graph queries, including recovery source IDs, agent IDs and workspace-finalize inputs, use bounded chunks. A SQL-generated 70,000-node graph proves PostgreSQL parameter limits are respected at the driver boundary. Archive checks still lock and freshly re-read sources, terminal subjects and related eligibility. Ordinary execution, decision and budget mutations pass their affected company; exact decision sources are included. Producer coverage includes decision creation and changes; interaction creation, resolution, expiry and supersession; issue creation, updates, blocker relations, checkout, adoption and release; agent hire, configuration, lifecycle, error clearing and deletion; budget incidents and company pause/resume; exhausted-run events and retry-run creation; native status, review, finalization and reconstruction; external conversation transitions; and imported blocker relations. Existing service transactions retain outer commit ownership. Savepoint release does not publish work before outer commit. diff --git a/server/src/__tests__/attention-retention-work.test.ts b/server/src/__tests__/attention-retention-work.test.ts index b1e7949a8b7..275b4f64570 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -17,7 +17,7 @@ import { buildIssueGraphLivenessIncidentKey, RECOVERY_ORIGIN_KINDS } from "../se import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { createAgentLifecycle } from "../services/agent-lifecycle.js"; import { BLOCKER_ATTENTION_MAX_DEPTH, BLOCKER_ATTENTION_MAX_NODES, issueService } from "../services/issues.js"; -import { notifyAttentionRetentionWork } from "../services/retention-work-notifications.js"; +import { notifyAttentionRetentionWork, notifyIssueAttentionRetentionWork } from "../services/retention-work-notifications.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -278,12 +278,16 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; expect(expected.terminalBlockerIssueId).toBe(terminal.id); expect(limited.terminalBlockerIssueId).toBe(limited.directBlockerIssueId); for (const input of [[root, heavy], [heavy, root]]) { - const batch = await owner.listBlockerAttention(companyId, input); - expect(batch.get(root.id)).toEqual(expected); expect(batch.get(heavy.id)).toEqual(limited); + const batch = await owner.listBlockerAttentionBatch(companyId, input); + if (!batch.deferredRootIds.has(root.id)) expect(batch.attention.get(root.id)).toEqual(expected); + if (!batch.deferredRootIds.has(heavy.id)) expect(batch.attention.get(heavy.id)).toEqual(limited); + expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES); + if (limit === "depth") expect(batch.deferredRootIds.has(root.id)).toBe(false); } const page = await attentionRetentionService(db).load({ companyId, sourceKind: "blocker_attention" }); const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!; - expect(entry.sourceId).toBe(terminal.id); expect(entry.dueAt).toBeLessThan(Date.now()); + expect(entry.dueAt).toBeLessThanOrEqual(Date.now()); + if (limit === "depth") expect(entry.sourceId).toBe(terminal.id); await attentionRetentionService(db).check(entry); expect((await state("blocker_attention", terminal.id))?.archivedAt).toBeInstanceOf(Date); expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, terminal.id))).toHaveLength(1); @@ -317,20 +321,79 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; await db.execute(sql`insert into issues (id, company_id, title, status, updated_at) select gen_random_uuid(), ${companyId}::uuid, 'parameter-root-' || n, 'blocked', ${old().toISOString()}::timestamptz from generate_series(1, 70) n`); - await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, assignee_agent_id, updated_at) - select gen_random_uuid(), ${companyId}::uuid, root.id, 'parameter-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz - from issues root cross join generate_series(1, 999) n + await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, updated_at) + select gen_random_uuid(), ${companyId}::uuid, root.id, 'parameter-branch-' || n, 'todo', ${old().toISOString()}::timestamptz + from issues root cross join generate_series(1, 10) n where root.company_id = ${companyId}::uuid and root.status = 'blocked'`); + await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, assignee_agent_id, updated_at) + select gen_random_uuid(), ${companyId}::uuid, branch.id, 'parameter-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz + from issues branch cross join generate_series(1, 99) n + where branch.company_id = ${companyId}::uuid and branch.title like 'parameter-branch-%'`); + const completed = await insertIssue("blocked"); + const shallow = await insertIssue("blocked", { updatedAt: new Date() }); + const middle = await insertIssue("todo", { updatedAt: new Date() }); + const overdue = await insertIssue("backlog", { assigneeAgentId: agentId }); + await db.insert(issueRelations).values([ + { companyId, issueId: middle.id, relatedIssueId: shallow.id, type: "blocks" }, + { companyId, issueId: overdue.id, relatedIssueId: middle.id, type: "blocks" }, + ]); const roots = await db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.status, "blocked"))); parameterCounts.length = 0; - const result = await issueService(db).listBlockerAttention(companyId, roots).catch(error => { + const result = await issueService(db).listBlockerAttentionBatch(companyId, roots).catch(error => { throw new Error(`Large graph failed: ${String(error.cause?.message ?? error.message).slice(0, 300)}`); }); - expect(result.size).toBe(70); - expect([...result.values()].every(item => item.state === "needs_attention" && item.terminalBlockerIssueId)).toBe(true); + expect(result.attention.size).toBe(72); + expect(result.deferredRootIds.has(completed.id)).toBe(false); + expect(result.attention.get(completed.id)?.terminalBlockerIssueId).toBe(completed.id); + expect([...result.attention.values()].every(item => item.state === "needs_attention" && item.terminalBlockerIssueId)).toBe(true); expect(Math.max(...parameterCounts)).toBeLessThanOrEqual(1100); + expect(parameterCounts.length).toBeLessThanOrEqual(100); + expect(result.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES); + expect(result.edgeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 2); + expect(result.fetchedRows).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 6); + expect(result.deferredRootIds.has(shallow.id)).toBe(true); + const service = attentionRetentionService(db); + const page = await service.load({ companyId, sourceKind: "blocker_attention" }); + const immediate = page.rows!.find(entry => entry.rootId === shallow.id && entry.variant === "terminal")!; + expect(immediate.dueAt).toBeLessThanOrEqual(Date.now()); + await service.check(immediate); + expect((await state("blocker_attention", overdue.id))?.archivedAt).toBeInstanceOf(Date); }, 120_000); + it("preserves observed live work when a single large root exhausts shared fetch capacity", async () => { + const root = await insertIssue("blocked"); + const live = await insertIssue("in_progress", { assigneeAgentId: agentId }); + await db.insert(issueRelations).values({ companyId, issueId: live.id, relatedIssueId: root.id, type: "blocks" }); + await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, assignee_agent_id, updated_at) + select gen_random_uuid(), ${companyId}::uuid, ${root.id}::uuid, 'bounded-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz + from generate_series(1, 3000) n`); + const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]); + expect(batch.deferredRootIds.has(root.id)).toBe(true); + expect(batch.attention.get(root.id)?.blockingTreeLive).toBe(true); + const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" }); + const entry = page.rows!.find(item => item.rootId === root.id && item.variant === "terminal")!; + expect(await service.check(entry)).toBeNull(); + expect(await state("blocker_attention", root.id)).toBeNull(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + }); + + it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => { + await db.execute(sql`insert into issues (id, company_id, title, status, updated_at) + select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz + from generate_series(1, 4) n`); + await db.execute(sql`insert into issues (id, company_id, parent_id, title, status, assignee_agent_id, updated_at) + select gen_random_uuid(), ${companyId}::uuid, root.id, 'held-leaf-' || n, 'backlog', ${agentId}::uuid, ${old().toISOString()}::timestamptz + from issues root cross join generate_series(1, 600) n + where root.company_id = ${companyId}::uuid and root.status = 'blocked'`); + await db.execute(sql`insert into decision_retention (company_id, source_kind, source_id, source_activity_at, keep) + select company_id, 'blocker_attention', id::text, updated_at, true from issues + where company_id = ${companyId}::uuid and status = 'backlog'`); + const service = attentionRetentionService(db), check = vi.spyOn(service, "check"); + await start(service); await quiet(); + expect(check).toHaveBeenCalledTimes(4); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + }); + it("canonicalizes wide circular notification variables to the supported identity fields", async () => { const sourceId = randomUUID(), issueId = randomUUID(), hints: string[] = []; const wide = { companyId, sourceKind: "review" as const, sourceId, issueId, ignored: {} as unknown }; @@ -418,6 +481,49 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; await archived("blocker_attention", terminal.id); await quiet(); }); + it("resolves opaque run issue references through trusted company issue rows", async () => { + const issue = await insertIssue(); + const [valid, opaque] = await db.insert(heartbeatRuns).values([ + { companyId, agentId, status: "succeeded", contextSnapshot: { issueId: issue.id } }, + { companyId, agentId, status: "succeeded", contextSnapshot: { issueId: "none" } }, + ]).returning(); + const hints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, values) => { + if (event === "settled") hints.push(...values ?? []); + }); + try { + await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, runId: valid.id })); + await db.transaction(tx => notifyIssueAttentionRetentionWork(tx, { companyId, runId: opaque.id })); + } finally { unsubscribe(); } + expect(hints).toEqual([JSON.stringify({ companyId, issueId: issue.id }), JSON.stringify({ companyId })]); + }); + + it("accepts ordinary liveness incident creation and completion with a none leaf sentinel", async () => { + const source = await insertIssue("blocked", { responsibleUserId: "board" }); + const service = issueService(db); + const incident = await service.create(companyId, { title: "Invalid participant", status: "todo", assigneeUserId: "board", + createdByUserId: "board", originKind: RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation, + originId: `harness_liveness:${companyId}:${source.id}:invalid_review_participant:none` }); + await service.update(source.id, { blockedByIssueIds: [incident.id] }); + await service.update(incident.id, { status: "done" }); + expect((await service.getById(incident.id))?.status).toBe("done"); + expect((await service.getRelationSummaries(source.id)).blockedBy).toEqual([]); + }); + + it.each(["hidden", "done"] as const)("reactivates an older review when moved stranded recovery is %s", async transition => { + const review = await insertIssue("in_review", { assigneeAgentId: agentId, responsibleUserId: "board" }); + const parent = await insertIssue("todo", { responsibleUserId: "board" }); + const recovery = await insertIssue("todo", { parentId: review.id, originKind: RECOVERY_ORIGIN_KINDS.strandedIssueRecovery, + originId: review.id, assigneeUserId: "board", responsibleUserId: "board" }); + await issueService(db).update(recovery.id, { parentId: parent.id }); + expect((await issueService(db).listReviewAttention(companyId, [review])).get(review.id)?.state).not.toBe("stalled"); + await start(); await quiet(); + await issueService(db).update(recovery.id, transition === "hidden" ? { hiddenAt: new Date() } : { status: "done" }); + await settle(); + expect((await state("review", review.id))?.archivedAt).toBeInstanceOf(Date); + expect(await db.select().from(decisionArchiveNotificationOutbox).where(eq(decisionArchiveNotificationOutbox.sourceId, review.id))).toHaveLength(1); + }); + it("scopes ordinary execution-control writes to their company", async () => { const otherCompany = randomUUID(); await db.insert(companies).values({ id: otherCompany, name: "Other", issuePrefix: "OTHER" }); diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index fb13ff43e82..0d28858b95c 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -4151,7 +4151,7 @@ const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSU beforeAll(async () => { if (!dependencyTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issues-blockers-"); - db = createDb(dependencyTestDatabaseUrl ?? tempDb!.connectionString); + db = createDb(dependencyTestDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 }); svc = issueService(db); await ensureIssueRelationsTable(db); }, 20_000); diff --git a/server/src/__tests__/work-products.test.ts b/server/src/__tests__/work-products.test.ts index 476e1156184..cef395409fd 100644 --- a/server/src/__tests__/work-products.test.ts +++ b/server/src/__tests__/work-products.test.ts @@ -1,3 +1,5 @@ +import { PgDialect } from "drizzle-orm/pg-core"; +import type { SQL } from "drizzle-orm"; import { describe, expect, it, vi } from "vitest"; import { enrichWorkProductMetadataWithDiff, @@ -33,6 +35,18 @@ function createWorkProductRow(overrides: Partial> = {}) }; } +function retentionReads(insideTransaction: () => boolean) { + return vi.fn(async (query: SQL) => { + const statement = new PgDialect().sqlToQuery(query); + expect(insideTransaction()).toBe(true); + expect(statement.params).toContain("company-1"); + expect(statement.params).toContain("issue-1"); + if (statement.sql.includes("origin_kind")) return [{ id: "issue-1", originKind: null, originId: null }]; + if (statement.sql.includes("with recursive affected")) return [{ issueId: "issue-1" }]; + throw new Error(`Unexpected work-product fixture query: ${statement.sql}`); + }); +} + describe("workProductService", () => { it("extracts runner totals and enriches work-product metadata", () => { const summary = workProductDiffSummaryFromEventPayload({ @@ -158,11 +172,16 @@ describe("workProductService", () => { const insertValues = vi.fn(() => ({ returning: insertReturning })); const txInsert = vi.fn(() => ({ values: insertValues })); + let transactionOpen = false; const tx = { + execute: retentionReads(() => transactionOpen), update: txUpdate, insert: txInsert, }; - const transaction = vi.fn(async (callback: (input: typeof tx) => Promise) => await callback(tx)); + const transaction = vi.fn(async (callback: (input: typeof tx) => Promise) => { + transactionOpen = true; + try { return await callback(tx); } finally { transactionOpen = false; } + }); const svc = workProductService({ transaction } as any); const result = await svc.createForIssue("issue-1", "company-1", { @@ -194,11 +213,16 @@ describe("workProductService", () => { const updateSet = vi.fn(() => ({ where: updateWhere })); const txUpdate = vi.fn(() => ({ set: updateSet })); + let transactionOpen = false; const tx = { + execute: retentionReads(() => transactionOpen), select: txSelect, update: txUpdate, }; - const transaction = vi.fn(async (callback: (input: typeof tx) => Promise) => await callback(tx)); + const transaction = vi.fn(async (callback: (input: typeof tx) => Promise) => { + transactionOpen = true; + try { return await callback(tx); } finally { transactionOpen = false; } + }); const svc = workProductService({ transaction } as any); const result = await svc.update("work-product-1", { diff --git a/server/src/services/attention-retention.ts b/server/src/services/attention-retention.ts index 27aedba8c47..ebe91416f52 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -32,6 +32,7 @@ type PageEvidence = { agents: Set; visibleIssues: Map; newerRuns: Set; + deferredRoots: Set; blockers: Awaited["listBlockerAttention"]>>; }; @@ -184,7 +185,9 @@ export function attentionRetentionService(db: Db) { const visibleRows = await readIssues(issueIds); const visible = new Map(visibleRows.map(row => [row.id, row])); const blockerRoots = visibleRows.filter(row => blockerIds.includes(row.id)); - const blockers: PageEvidence["blockers"] = blockerRoots.length ? await issueService(database).listBlockerAttention(companyId, blockerRoots) : new Map(); + const blockerBatch = blockerRoots.length ? await issueService(database).listBlockerAttentionBatch(companyId, blockerRoots) + : { attention: new Map(), deferredRootIds: new Set() }; + const blockers: PageEvidence["blockers"] = blockerBatch.attention; const terminalIds = [...blockers.values()].filter(terminalBlockerRequiresAttention) .map(attention => attention.terminalBlockerIssueId).filter((id): id is string => id != null && !visible.has(id)); for (const row of await readIssues(terminalIds)) visible.set(row.id, row); @@ -192,13 +195,17 @@ export function attentionRetentionService(db: Db) { decisions: new Map(decisionRows.map(row => [row.id, row])), recoveries: new Map(recoveryRows.map(row => [row.id, row])), runs: new Map(runRows.map(row => [row.id, row])), agents: new Set(runAgents.map(row => row.id)), visibleIssues: visible, newerRuns: await listFailedRunsWithNewerRun(database, companyId, - runRows.map(run => ({ ...run, issueId: readRunIssueId(run.contextSnapshot) }))), blockers, + runRows.map(run => ({ ...run, issueId: readRunIssueId(run.contextSnapshot) }))), blockers, deferredRoots: blockerBatch.deferredRootIds, }); } return result; } async function eligible(database: Db, candidate: Candidate, variant?: "owner" | "terminal", lock = false, reviewAttention?: Map, interactionAttention?: Map, evidence?: PageEvidence): Promise { + if (candidate.sourceKind === "blocker_attention" && variant === "terminal" && evidence?.deferredRoots.has(candidate.sourceId)) { + return { ...candidate, rootId: candidate.sourceId, variant, + id: `${candidate.sourceKind}:${candidate.sourceId}:terminal`, dueAt: Date.now() }; + } let sourceId = candidate.sourceId; let activityAt = candidate.activityAt; let dueAt = candidate.deadlineAt; diff --git a/server/src/services/issue-thread-interactions.test.ts b/server/src/services/issue-thread-interactions.test.ts index 24e1489006f..5ad905a8a10 100644 --- a/server/src/services/issue-thread-interactions.test.ts +++ b/server/src/services/issue-thread-interactions.test.ts @@ -1,5 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; -import { getTableName } from "drizzle-orm"; +import { PgDialect } from "drizzle-orm/pg-core"; +import { getTableName, type SQL } from "drizzle-orm"; const mockCreateChild = vi.fn(); @@ -48,8 +49,18 @@ function createFakeDb(args: { const toolActionRequestUpdates: Array> = []; const inserts: Array<{ table: string; values: Record }> = []; let selectCallCount = 0; + let transactionDepth = 0; const db: any = { + execute: vi.fn(async (query: SQL) => { + const statement = new PgDialect().sqlToQuery(query); + expect(transactionDepth).toBeGreaterThan(0); + expect(statement.params).toContain(interactionRow.companyId); + expect(statement.params).toContain(interactionRow.issueId); + if (statement.sql.includes("origin_kind")) return [{ id: interactionRow.issueId, originKind: null, originId: null }]; + if (statement.sql.includes("with recursive affected")) return [{ issueId: interactionRow.issueId }]; + throw new Error(`Unexpected interaction fixture query: ${statement.sql}`); + }), select: vi.fn(() => { selectCallCount += 1; const rows = selectCallCount === 1 ? [interactionRow] : (args.parentRows ?? []); @@ -84,7 +95,10 @@ function createFakeDb(args: { inserts.push({ table: getTableName(table as never), values }); }, })), - transaction: async (callback: (tx: typeof db) => Promise) => callback(db), + transaction: async (callback: (tx: typeof db) => Promise) => { + transactionDepth++; + try { return await callback(db); } finally { transactionDepth--; } + }, }; return { @@ -177,6 +191,7 @@ describe("issueThreadInteractionService", () => { }; let selectCallCount = 0; + let transactionDepth = 0; const db: any = { select: vi.fn(() => { selectCallCount += 1; diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index c91a325ad39..4d5fd5f85f6 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -3436,6 +3436,7 @@ type IssueBlockerAttentionEdge = { blockerIssueId: string; }; type IssueBlockerAttentionQueryRow = IssueBlockerAttentionNode & { + executionWorkspaceId?: string | null; issueId: string | null; blockerIssueId: string; }; @@ -3824,11 +3825,15 @@ async function terminalExplicitBlockersByRoot( return terminalByRoot; } -async function listIssueBlockerAttentionMap( +async function listIssueBlockerAttentionMap(dbOrTx: any, companyId: string, issueRows: IssueBlockerAttentionInputNode[]) { + return (await readIssueBlockerAttentionBatch(dbOrTx, companyId, issueRows)).attention; +} + +async function readIssueBlockerAttentionBatch( dbOrTx: any, companyId: string, issueRows: IssueBlockerAttentionInputNode[], -): Promise> { +) { const roots = issueRows.filter( (row) => row.companyId === companyId && row.status === "blocked", ); @@ -3838,37 +3843,38 @@ async function listIssueBlockerAttentionMap( attentionMap.set(row.id, createIssueBlockerAttention()); } } - if (roots.length === 0) return attentionMap; + const deferredRootIds = new Set(roots.slice(BLOCKER_ATTENTION_MAX_NODES).map(root => root.id)); + let edgeCount = 0, fetchedRows = 0; + const result = () => ({ attention: attentionMap, deferredRootIds, nodeCount: nodesById.size, edgeCount, fetchedRows }); + if (roots.length === 0) return { attention: attentionMap, deferredRootIds, nodeCount: 0, edgeCount, fetchedRows }; const nodesById = new Map(); const edgesByIssueId = new Map(); - for (const root of roots) nodesById.set(root.id, { ...root }); + const admittedRoots = roots.slice(0, BLOCKER_ATTENTION_MAX_NODES); + for (const root of admittedRoots) nodesById.set(root.id, { ...root }); - type Traversal = { nodes: Set; expanded: Set; frontier: string[]; truncated: boolean }; - const traversals = new Map(roots.map(root => [root.id, { - nodes: new Set([root.id]), expanded: new Set(), frontier: [root.id], truncated: false, + type Traversal = { rootId: string; nodes: Set; expanded: Set; frontier: string[]; truncated: boolean }; + const traversals = new Map(admittedRoots.map(root => [root.id, { + rootId: root.id, nodes: new Set([root.id]), expanded: new Set(), frontier: [root.id], truncated: false, }])); const expanded = new Set(); + const queried = new Set(); const pendingFinalizeBlockerIssueIds = new Set(); for (let depth = 0; depth < BLOCKER_ATTENTION_MAX_DEPTH; depth += 1) { const active = [...traversals.values()].filter(traversal => !traversal.truncated && traversal.frontier.length > 0); if (active.length === 0) break; - const frontier = [...new Set(active.flatMap(traversal => traversal.frontier))].filter(id => !expanded.has(id)); + const unread = new Set(); + for (const traversal of active) for (const id of traversal.frontier) if (!expanded.has(id)) unread.add(id); + const frontier = [...unread]; for (const chunk of chunkList( frontier, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE, )) { - const readinessByIssueId = await listIssueDependencyReadinessMap( - dbOrTx, - companyId, - chunk, - ); - for (const readiness of readinessByIssueId.values()) { - for (const blockerIssueId of readiness.pendingFinalizeBlockerIssueIds) { - pendingFinalizeBlockerIssueIds.add(blockerIssueId); - } - } + const remaining = Math.min(BLOCKER_ATTENTION_MAX_NODES - nodesById.size, + BLOCKER_ATTENTION_MAX_NODES * 2 - edgeCount); + if (remaining <= 0) break; + const rowLimit = remaining + 1; const explicitBlockerRowsPromise: Promise< IssueBlockerAttentionQueryRow[] > = dbOrTx @@ -3882,6 +3888,7 @@ async function listIssueBlockerAttentionMap( title: issues.title, status: issues.status, executionRunId: issues.executionRunId, + executionWorkspaceId: issues.executionWorkspaceId, assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, }) @@ -3894,7 +3901,7 @@ async function listIssueBlockerAttentionMap( inArray(issueRelations.relatedIssueId, chunk), eq(issues.companyId, companyId), ), - ); + ).limit(rowLimit); const childRowsPromise: Promise = dbOrTx .select({ issueId: issues.parentId, @@ -3919,56 +3926,56 @@ async function listIssueBlockerAttentionMap( BLOCKER_ATTENTION_CHILD_TERMINAL_STATUSES, ), ), - ); + ).limit(rowLimit); const [explicitBlockerRows, childRows] = await Promise.all([ explicitBlockerRowsPromise, childRowsPromise, ]); + for (const id of chunk) queried.add(id); + fetchedRows += explicitBlockerRows.length + childRows.length; + const pendingFinalize = await listPendingFinalizeBlockerIssueIds(dbOrTx, companyId, + explicitBlockerRows.flatMap(row => row.status === "done" && row.executionWorkspaceId + ? [{ blockerIssueId: row.blockerIssueId, executionWorkspaceId: row.executionWorkspaceId }] : [])); + for (const id of pendingFinalize) pendingFinalizeBlockerIssueIds.add(id); + let complete = explicitBlockerRows.length < rowLimit && childRows.length < rowLimit; const unresolvedExplicitBlockerRows = explicitBlockerRows.filter( (row) => row.status !== "done" || pendingFinalizeBlockerIssueIds.has(row.blockerIssueId), ); - appendBlockerAttentionEdges(edgesByIssueId, [ - ...unresolvedExplicitBlockerRows - .filter( - (row): row is IssueBlockerAttentionQueryRow & { issueId: string } => - row.issueId !== null, - ) - .map((row) => ({ - issueId: row.issueId, - blockerIssueId: row.blockerIssueId, - })), - ...childRows - .filter( - (row): row is IssueBlockerAttentionQueryRow & { issueId: string } => - row.issueId !== null, - ) - .map((row) => ({ - issueId: row.issueId, - blockerIssueId: row.blockerIssueId, - })), - ]); - for (const row of [...unresolvedExplicitBlockerRows, ...childRows]) { - if (!row.issueId || nodesById.has(row.blockerIssueId)) continue; - nodesById.set(row.blockerIssueId, { - id: row.blockerIssueId, - companyId: row.companyId, - parentId: row.parentId, - identifier: row.identifier, - title: row.title, - status: row.status, - executionRunId: row.executionRunId, - assigneeAgentId: row.assigneeAgentId, - assigneeUserId: row.assigneeUserId, - }); + if (!row.issueId) continue; + if (!nodesById.has(row.blockerIssueId)) { + if (nodesById.size >= BLOCKER_ATTENTION_MAX_NODES) { complete = false; continue; } + nodesById.set(row.blockerIssueId, { + id: row.blockerIssueId, companyId: row.companyId, parentId: row.parentId, identifier: row.identifier, + title: row.title, status: row.status, executionRunId: row.executionRunId, + assigneeAgentId: row.assigneeAgentId, assigneeUserId: row.assigneeUserId, + }); + } + const edges = edgesByIssueId.get(row.issueId) ?? []; + if (edges.some(edge => edge.blockerIssueId === row.blockerIssueId)) continue; + if (edgeCount >= BLOCKER_ATTENTION_MAX_NODES * 2) { complete = false; continue; } + edges.push({ issueId: row.issueId, blockerIssueId: row.blockerIssueId }); + edgesByIssueId.set(row.issueId, edges); + edgeCount++; } - for (const id of chunk) expanded.add(id); + if (complete) for (const id of chunk) expanded.add(id); + } for (const traversal of active) { + if (traversal.frontier.some(id => !expanded.has(id))) { + for (const id of traversal.frontier) if (queried.has(id)) { + traversal.expanded.add(id); + for (const edge of edgesByIssueId.get(id) ?? []) if (nodesById.has(edge.blockerIssueId)) traversal.nodes.add(edge.blockerIssueId); + } + traversal.truncated = true; + traversal.frontier = []; + deferredRootIds.add(traversal.rootId); + continue; + } const nextFrontier = new Set(); for (const id of traversal.frontier) { traversal.expanded.add(id); @@ -4548,7 +4555,7 @@ async function listIssueBlockerAttentionMap( ); } - return attentionMap; + return result(); } type IssueReviewAttentionInput = Pick; @@ -9383,6 +9390,9 @@ export function issueService(db: Db) { return listIssueDependencyReadinessMap(dbOrTx, companyId, issueIds); }, + listBlockerAttentionBatch: async (companyId: string, issueRows: IssueBlockerAttentionInputNode[], dbOrTx: any = db) => + readIssueBlockerAttentionBatch(dbOrTx, companyId, issueRows), + listBlockerAttention: async ( companyId: string, issueRows: IssueBlockerAttentionInputNode[], diff --git a/server/src/services/retention-work-notifications.ts b/server/src/services/retention-work-notifications.ts index cee2486f54c..ba0379afee1 100644 --- a/server/src/services/retention-work-notifications.ts +++ b/server/src/services/retention-work-notifications.ts @@ -20,9 +20,10 @@ export async function notifyIssueAttentionRetentionWork(transaction: object, where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid and issue_id is not null union select native_issue_id::text from heartbeat_runs where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid and native_issue_id is not null - union select coalesce(context_snapshot->>'issueId', context_snapshot->>'taskId') from heartbeat_runs - where company_id = ${scope.companyId}::uuid and id = ${scope.runId}::uuid - and coalesce(context_snapshot->>'issueId', context_snapshot->>'taskId') is not null`); + union select issue.id::text from heartbeat_runs run inner join issues issue + on issue.company_id = run.company_id + and issue.id::text = coalesce(run.context_snapshot->>'issueId', run.context_snapshot->>'taskId') + where run.company_id = ${scope.companyId}::uuid and run.id = ${scope.runId}::uuid`); for (const row of rows) seeds.push(row.id as string); } if (!seeds.length) return notifyAttentionRetentionWork(transaction, { companyId: scope.companyId }); @@ -31,10 +32,20 @@ export async function notifyIssueAttentionRetentionWork(transaction: object, and id in (${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)})`); const current = origins.find(source => source.id === scope.issueId); origins.push({ originKind: scope.originKind ?? current?.originKind, originId: scope.originId ?? current?.originId }); + const references: string[] = []; for (const source of origins) { + if (source.originKind === RECOVERY_ORIGIN_KINDS.strandedIssueRecovery && typeof source.originId === "string") { + references.push(source.originId); + } const incident = source.originKind === RECOVERY_ORIGIN_KINDS.issueGraphLivenessEscalation ? parseIssueGraphLivenessIncidentKey(source.originId as string | null) : null; - if (incident?.companyId === scope.companyId) seeds.push(incident.issueId, incident.leafIssueId); + if (incident?.companyId === scope.companyId) references.push(incident.issueId, incident.leafIssueId); + } + if (references.length) { + const trusted = await database.execute(sql`select id::text as id from issues + where company_id = ${scope.companyId}::uuid + and id::text in (${sql.join([...new Set(references)].map(id => sql`${id}`), sql`, `)})`); + for (const row of trusted) seeds.push(row.id as string); } const affected = await database.execute(sql`with recursive affected(id) as ( select unnest(array[${sql.join([...new Set(seeds)].map(id => sql`${id}::uuid`), sql`, `)}]) From 7bb1af657aa2eda4573c8b77b21afaf0fc226d9b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:25:05 -0500 Subject: [PATCH 4/6] test: await issue recovery dispatch and cancellation settlement Co-Authored-By: Paperclip --- .../issue-recovery-deadlines.test.ts | 35 ++++++++++++++++--- 1 file changed, 31 insertions(+), 4 deletions(-) diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts index b3d65f4dbcc..f5416fec571 100644 --- a/server/src/__tests__/issue-recovery-deadlines.test.ts +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -1,3 +1,4 @@ +import { setImmediate as immediate } from "node:timers/promises"; import { idleWorkSnapshot } from "../services/task-admission.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; @@ -27,6 +28,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea let draining = false; const tasks = new Map number | null; hasPending: () => Promise }>(); const errors: unknown[] = []; + const passes = new Set>(); const extraStops: Array<() => Promise> = []; const extraDeadlines: Array<() => number | null> = []; beforeAll(async () => { @@ -52,7 +54,14 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea onError: error => errors.push(error) }); tasks.clear(); const register = coordinator.register.bind(coordinator); - coordinator.register = (queue, task) => { tasks.set(queue, task); return register(queue, task); }; + coordinator.register = (queue, task) => { + tasks.set(queue, task); + return register(queue, { ...task, run: signal => { + const work = task.run(signal); passes.add(work); + void work.then(() => passes.delete(work), () => passes.delete(work)); + return work; + } }); + }; worker = heartbeat.registerIssueStateRecoveryWork(coordinator, () => !draining); dispatch = heartbeat.registerRunDispatchWork(coordinator, () => !draining); cancellation = heartbeat.registerScheduledRetryCancellationWork(coordinator, () => !draining); @@ -89,6 +98,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea } async function quiet() { await settled(); + expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); const reads = vi.spyOn(db, "select"); const executes = vi.spyOn(db, "execute"); const selected = reads.mock.calls.length, executed = executes.mock.calls.length; if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); @@ -97,15 +107,32 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea } async function finish(issueId: string) { await issueService(db).update(issueId, { status: "done" }); - await heartbeat.drainActiveRunExecutions(); + // Closing one blocker can legitimately start its dependent. Wait for this + // task's dispatch and execution; quiet() verifies the whole fixture later. + await vi.waitFor(async () => { + await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); + expect(await db.select().from(heartbeatRuns).where(and( + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`, + sql`${heartbeatRuns.status} in ('queued', 'running')`, + ))).toHaveLength(0); + }, { timeout: 15_000 }); const saved = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); - expect(await db.select().from(heartbeatRuns).where(sql`${heartbeatRuns.status} in ('queued', 'running')`)).toHaveLength(0); if (!vi.isFakeTimers()) vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); if (saved.length > 0) { const due = Math.max(...saved.map(run => run.scheduledRetryAt!.getTime())); - await vi.advanceTimersByTimeAsync(Math.max(0, due - Date.now()) + 1); + // Keep PostgreSQL I/O and coordinator settlement ahead of virtual time. + // A single leap can otherwise occur before the retry deadline is armed. + while (Date.now() <= due) { + await vi.advanceTimersByTimeAsync(Math.min(500, due - Date.now() + 1)); + await Promise.allSettled([...passes]); + await immediate(); + } await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(500); + await Promise.allSettled([...passes]); + await heartbeat.drainActiveRunExecutions(); const remaining = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.status, "scheduled_retry"), sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issueId}`)); expect(remaining).toHaveLength(0); From e3fff84b6f81448c730dd165253a1a673f8d19b3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:39:21 -0500 Subject: [PATCH 5/6] fix: scope comment-triggered recovery to the affected issue Preserve company reconstruction for graph, run and eligibility changes. A real PostgreSQL producer regression verifies one assignment check among1000 open tasks, with broader eligibility reconstruction still intact. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-issue-recovery-sleep.md | 6 ++- .../issue-recovery-deadlines.test.ts | 47 +++++++++++++++++++ server/src/services/issue-comment-writes.ts | 3 +- .../services/issue-recovery-notifications.ts | 11 +++-- ...ative-status-reassessment-notifications.ts | 2 +- server/src/services/recovery/work.ts | 5 +- .../services/run-dispatch-notifications.ts | 2 +- 7 files changed, 67 insertions(+), 9 deletions(-) diff --git a/doc/plans/2026-10-30-issue-recovery-sleep.md b/doc/plans/2026-10-30-issue-recovery-sleep.md index f191d5dca42..0c6cdf6b4d3 100644 --- a/doc/plans/2026-10-30-issue-recovery-sleep.md +++ b/doc/plans/2026-10-30-issue-recovery-sleep.md @@ -1,6 +1,6 @@ # Issue recovery without periodic scans -The issue recovery owners reconstruct their current obligations once at startup. Committing source writes invalidate only their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. +The issue recovery owners reconstruct their current obligations once at startup. Ordinary comment writes invalidate only the affected issue for assignment and lock recovery. Lifecycle, graph, run, configuration and eligibility changes invalidate their company. The coordinator keeps scopes, per-issue deadlines, and per-entry failures in memory. Its pending check does not query the database. | Owner | Durable truth and producer boundary | Deadline or event | Exit | | --- | --- | --- | --- | @@ -11,6 +11,10 @@ The issue recovery owners reconstruct their current obligations once at startup. | Task watchdog | Active issue watchdogs and their current source subtree; normal watchdog upsert/disable; issue/comment/document/work-product and wait changes | Exact source creation plus the existing 15-second first-run grace. An incomplete scan retains the watchdog with error backoff. | Live path, open/reviewed reusable review, disabled configuration, or ineligible source. | | Stale issue lock | Existing checkout/execution run references; issue and run lifecycle changes | A committing ownership change, or the fresh owning controller's saved lease expiry. | No stale reference, a valid owner, or the existing guarded terminalization and lock clear. | +Local comment hints do not broaden into an assignment check for every open task. A real PostgreSQL regression starts with1,000 open assigned issues, writes one comment through `issueService.addComment`, and observes exactly one assignment check; a subsequent company eligibility notification still checks all1,000. Restoring the previous comment hint makes that same regression check all1,000 after the comment. The test substitutes the expensive assignment decision only to count its invocations; it uses the real transaction, notification, coordinator and paged SQL. Separate producer acceptance cases exercise real recovery decisions and process execution. + +Parent/descendant disposition repair and shared execution references make status and run changes broader than a comment. Their existing notification boundaries deliberately retain company scope. Orphan blockers, active recovery actions, resolved dependencies and watchdogs also keep their company/subtree duties; an issue comment does not drop those independent obligations. The scope serializer selects known primitive fields and never serializes an entire insert or update object. + Reconstruction uses keyset pages of at most100rows. Repeated company hints coalesce reconstruction for30seconds. Already-known entry deadlines preempt that spacing without an early probe on each comment. Scheduling suppression is read once per pass. A failed read keeps scoped backoff debt. The two subsidiary company repairs do not run inside the per-issue loop. Legacy direct aggregate API calls retain both repairs. The owner methods accept company and issue/watchdog identity. A narrowed lock check still reads other same-company references before using terminal-issue authority. A stale terminal alias cannot end an active issue's shared run. The controller lease owner supplies its fresh expiry; the existing write predicate revalidates the observed boot ID and database-clock expiry. diff --git a/server/src/__tests__/issue-recovery-deadlines.test.ts b/server/src/__tests__/issue-recovery-deadlines.test.ts index f5416fec571..e23a672c0dc 100644 --- a/server/src/__tests__/issue-recovery-deadlines.test.ts +++ b/server/src/__tests__/issue-recovery-deadlines.test.ts @@ -9,6 +9,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { heartbeatService } from "../services/heartbeat.js"; import { issueService } from "../services/issues.js"; import { issueRecoveryActionService } from "../services/issue-recovery-actions.js"; +import { registerIssueRecoveryWork } from "../services/recovery/work.js"; import { notifyIssueRecovery } from "../services/issue-recovery-notifications.js"; import { taskWatchdogService } from "../services/task-watchdogs.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -223,6 +224,52 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea expect((await originalSelect().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, foreignAction.id)))[0].status).toBe("active"); expect(await originalSelect().from(heartbeatRuns)).toHaveLength(0); await quiet(); }); + it("limits an ordinary comment to its issue among one thousand open assignments", async () => { + const f = await fixture(); + const rows = await db.insert(issues).values(Array.from({ length: 1000 }, (_, index) => ({ + companyId: f.companyId, title: `Open work ${index}`, status: "todo", assigneeAgentId: f.agentId, + }))).returning({ id: issues.id }); + const checked: string[] = []; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"] }); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const register = coordinator.register.bind(coordinator); + coordinator.register = (queue, task) => register(queue, { ...task, run: signal => { + const pass = task.run(signal); passes.add(pass); + void pass.then(() => passes.delete(pass), () => passes.delete(pass)); + return pass; + } }); + // Observe the actual paged reconstruction boundary without executing one + // thousand adapters. The producer, transaction, coordinator and SQL are real; + // the other acceptance cases exercise the resulting recovery decisions. + worker = registerIssueRecoveryWork(db, coordinator, { + stranded: async scope => { checked.push(scope.issueId); }, + orphanBlockers: async () => {}, recoveryActions: async () => {}, + dependencies: async () => ({ enqueueFailed: 0 }), locks: async () => {}, + watchdogs: async () => ({ incomplete: 0 }), schedulingSuppressed: async () => false, + }, () => true); + await worker.ready; + async function settleWork() { + for (let step = 0; step < 120; step++) { + await vi.advanceTimersByTimeAsync(1000); + await Promise.allSettled([...passes]); await immediate(); + if (coordinator.nextWakeAt() === null && idleWorkSnapshot().active === 0) return; + } + throw new Error("Recovery did not settle its committed work within two virtual minutes"); + } + await settleWork(); + expect(new Set(checked).size).toBe(1000); + checked.length = 0; + await issueService(db).addComment(rows[0].id, "New evidence on this task", { userId: "recovery-user" }); + await settleWork(); + expect(checked).toEqual([rows[0].id]); + // Company eligibility changes still reconstruct all assignments. Narrowing + // comment hints must not hide unrelated work after an operator resumes it. + checked.length = 0; + await db.transaction(tx => notifyIssueRecovery(tx, { companyId: f.companyId })); + await settleWork(); + expect(new Set(checked).size).toBe(1000); + await quiet(); + }); it("dispatches an ordinary new assignment after commit and returns to quiet after completion", async () => { const f = await fixture(); await start(); await quiet(); const issue = await issueService(db).create(f.companyId, { title: "Recover assignment", status: "todo", assigneeAgentId: f.agentId }); diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts index a968ecb1fec..39d8f6f891c 100644 --- a/server/src/services/issue-comment-writes.ts +++ b/server/src/services/issue-comment-writes.ts @@ -7,7 +7,8 @@ type WriteDb = Db | Parameters[0]>[0]; export function insertIssueComment(db: WriteDb, input: PgInsertValue & { companyId: string }) { return db.transaction(async tx => { - await notifyIssueRecovery(tx, { companyId: input.companyId }); + await notifyIssueRecovery(tx, { companyId: input.companyId, + issueId: typeof input.issueId === "string" ? input.issueId : undefined }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`); return tx.insert(issueComments).values(input).returning(); }); diff --git a/server/src/services/issue-recovery-notifications.ts b/server/src/services/issue-recovery-notifications.ts index 3c974228d86..7b709960831 100644 --- a/server/src/services/issue-recovery-notifications.ts +++ b/server/src/services/issue-recovery-notifications.ts @@ -1,11 +1,16 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export type IssueRecoveryScope = { companyId: string }; +/** An issue hint is reserved for local facts such as comments. Lifecycle, + * graph, run and eligibility changes retain company scope because their + * consequences can include ancestors and shared execution references. */ +export type IssueRecoveryScope = { companyId: string; issueId?: string }; export async function notifyIssueRecovery(transaction: object, scope?: IssueRecoveryScope) { - const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + const companyHint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; + const issueHint = scope ? JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId }) : "*"; for (const queue of [DELIVERY_QUEUES.orphanBlockers, DELIVERY_QUEUES.activeRecoveryActions, DELIVERY_QUEUES.strandedAssignments, DELIVERY_QUEUES.resolvedDependencies, DELIVERY_QUEUES.taskWatchdogs, DELIVERY_QUEUES.staleIssueLocks]) { - await notifyDeliveryWork(transaction, queue, hint); + await notifyDeliveryWork(transaction, queue, + queue === DELIVERY_QUEUES.strandedAssignments || queue === DELIVERY_QUEUES.staleIssueLocks ? issueHint : companyHint); } } diff --git a/server/src/services/native-status-reassessment-notifications.ts b/server/src/services/native-status-reassessment-notifications.ts index 4b5eac73b2f..d3b543d9914 100644 --- a/server/src/services/native-status-reassessment-notifications.ts +++ b/server/src/services/native-status-reassessment-notifications.ts @@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) { - await notifyIssueRecovery(tx, scope); + await notifyIssueRecovery(tx, scope ? { companyId: scope.companyId } : undefined); return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); } diff --git a/server/src/services/recovery/work.ts b/server/src/services/recovery/work.ts index 8bf7b75c7f4..407e4f1eb6d 100644 --- a/server/src/services/recovery/work.ts +++ b/server/src/services/recovery/work.ts @@ -32,13 +32,14 @@ export function registerIssueRecoveryWork( reconstructionIntervalMs: 30_000, beforePass: async () => { enabled = !(await owner.schedulingSuppressed()); }, decodeScope: hint => hint === "*" ? undefined : JSON.parse(hint) as IssueRecoveryScope, - matches: (entry, scope) => !scope || entry.companyId === scope.companyId, + matches: (entry, scope) => !scope || (entry.companyId === scope.companyId && (!scope.issueId || entry.id === scope.issueId)), load: async scope => { - const key = scope?.companyId ?? "*"; + const key = scope ? JSON.stringify([scope.companyId, scope.issueId]) : "*"; if (!enabled) { cursors.delete(key); return []; } const cursor = cursors.get(key); const rows = await db.select({ id: issues.id, companyId: issues.companyId }).from(issues).where(and( scope ? eq(issues.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(issues.id, scope.issueId) : undefined, cursor ? gt(issues.id, cursor) : undefined, statuses ? inArray(issues.status, statuses) : or(isNotNull(issues.checkoutRunId), isNotNull(issues.executionRunId)), statuses ? isNull(issues.hiddenAt) : undefined, diff --git a/server/src/services/run-dispatch-notifications.ts b/server/src/services/run-dispatch-notifications.ts index 9e8784b7481..cba58357cf0 100644 --- a/server/src/services/run-dispatch-notifications.ts +++ b/server/src/services/run-dispatch-notifications.ts @@ -4,7 +4,7 @@ import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notificatio export type RunDispatchScope = { companyId: string }; export async function notifyRunDispatch(transaction: object, scope?: RunDispatchScope) { - await notifyIssueRecovery(transaction, scope); + await notifyIssueRecovery(transaction, scope ? { companyId: scope.companyId } : undefined); const hint = scope ? JSON.stringify({ companyId: scope.companyId }) : "*"; await notifyDeliveryWork(transaction, DELIVERY_QUEUES.scheduledRetryDispatch, hint); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.queuedRunDispatch, hint); From 6469daec07fed5afc68d8d441f4a0009dcea9673 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 19:41:19 -0500 Subject: [PATCH 6/6] fix: bound status-card active-hours calculation work Reuse bounded timezone formatters and upcoming openings only within their exact searched closed ranges. Short-circuit empty windows while preserving the existing cold DST search and precise instant eligibility. Verify cost bounds, cache eviction, timezone transitions and twenty-card ordinary source delivery. Co-Authored-By: Paperclip --- doc/plans/2026-10-30-status-cards-sleep.md | 14 ++ .../status-card-evaluation-work.test.ts | 47 +++++++ .../status-card-update-engine.test.ts | 124 +++++++++++++++++- .../src/services/status-card-update-engine.ts | 53 ++++++-- 4 files changed, 225 insertions(+), 13 deletions(-) diff --git a/doc/plans/2026-10-30-status-cards-sleep.md b/doc/plans/2026-10-30-status-cards-sleep.md index 61d451caed5..3f3b8933cbc 100644 --- a/doc/plans/2026-10-30-status-cards-sleep.md +++ b/doc/plans/2026-10-30-status-cards-sleep.md @@ -141,3 +141,17 @@ deadline-reconstruction hint. Every card mutation requires a company ID; only instance-level settings retain wildcard ingress. Deleting an already-missing card emits no hint. The real two-company regression proves deletion does not re-run unchanged reactive searches or wake either summarizer, then stays quiet. + +Active-hours eligibility keeps the existing exact UTC-minute search across DST +changes. A bounded 32-entry timezone formatter cache removes repeated ICU +construction. A bounded 256-entry cache reuses the next opening for the exact +(timezone,start,end) policy only while the candidate instant stays within its +previously searched closed range. The precise instant is checked first, so +historical second-based offsets, the opening itself, and millisecond deadlines +keep their behavior. Clock rollback and policy changes cannot reuse the wrong +range. Empty windows return without formatter or search work. A cold nonempty +policy can still perform at most 2881 conversions including its first check; warm +closed-range evaluations perform one conversion, not another minute walk. +Both caches use bounded LRU retention and hold no company/card data. Tests prove +eviction, skipped/repeated hours, half-hour DST, date-line changes, rollback, +strict boundaries, and actual twenty-card source delivery followed by quiet. diff --git a/server/src/__tests__/status-card-evaluation-work.test.ts b/server/src/__tests__/status-card-evaluation-work.test.ts index e4588b16be1..d012a239efd 100644 --- a/server/src/__tests__/status-card-evaluation-work.test.ts +++ b/server/src/__tests__/status-card-evaluation-work.test.ts @@ -272,6 +272,53 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await quiet(); expect(errors).toEqual([]); }, 30_000); + it("bounds active-hours conversion work across twenty cards and repeated ordinary company writes", async () => { + vi.setSystemTime(new Date("2026-11-02T18:00:00Z")); + const hoursPolicy = { ...policy(), activeHours: { start: "09:00", end: "17:00", timezone: "UTC" } }; + const cards = []; + for (let i = 0; i < 20; i++) cards.push(await card("reactive", hoursPolicy)); + await start(); await quiet(); + const RealFormatter = Intl.DateTimeFormat; + const actualParts = RealFormatter.prototype.formatToParts; + let constructions = 0, conversions = 0; + const constructor = vi.spyOn(Intl, "DateTimeFormat").mockImplementation(function (...args: ConstructorParameters) { + if (args[0] === "en-GB" && args[1]?.hourCycle === "h23") constructions++; + return new RealFormatter(...args); + } as unknown as typeof Intl.DateTimeFormat); + const parts = vi.spyOn(RealFormatter.prototype, "formatToParts").mockImplementation(function (this: Intl.DateTimeFormat, date) { + if (this.resolvedOptions().locale === "en-GB" && this.resolvedOptions().hourCycle === "h23") conversions++; + return actualParts.call(this, date); + }); + const target = await issueService(db).create(companyId, { title: "Closed-hours blocker", status: "blocked", priority: "medium" }); + await settled(); + expect(constructions).toBeLessThanOrEqual(1); + expect(conversions).toBeLessThanOrEqual(1000); + const cold = { constructions, conversions }; + constructions = 0; conversions = 0; + for (let i = 0; i < 10; i++) { + await issueService(db).addComment(target.id, `Closed-hours human update ${i}`, { userId: "board-user" }); + await settled(); + } + expect(constructions).toBe(0); + expect(conversions).toBeGreaterThanOrEqual(200); + expect(conversions).toBeLessThanOrEqual(800); + expect(wakeup).not.toHaveBeenCalled(); + const opening = new Date("2026-11-03T09:00:00Z"); + for (const created of cards) expect((await statusCardService(db).getById(created.id))) + .toMatchObject({ state: "paused_hours", pendingChangeCount: 1, nextEvalAt: opening }); + process.stdout.write(JSON.stringify({ measurement: "status-card company active-hours workload", cards: 20, + sourceWrites: 11, cold, warm: { constructions, conversions } }) + "\n"); + constructor.mockRestore(); parts.mockRestore(); + await vi.advanceTimersByTimeAsync(opening.getTime() - Date.now()); await settled(); + expect(wakeup).toHaveBeenCalledTimes(20); + for (const created of cards) { + const generated = (await statusCardService(db).getById(created.id))!; + expect(generated.generatingIssueId).toBeTruthy(); + await writeback(created.id, generated.generatingIssueId!); + } + await quiet(); expect(errors).toEqual([]); + }, 60_000); + it("cancels archived or disabled future work and replaces an earlier interval deadline", async () => { await start(); let created = await card("interval"); await settled(); const old = coordinator!.nextWakeAt(); diff --git a/server/src/__tests__/status-card-update-engine.test.ts b/server/src/__tests__/status-card-update-engine.test.ts index b8309415770..9fdf18584e9 100644 --- a/server/src/__tests__/status-card-update-engine.test.ts +++ b/server/src/__tests__/status-card-update-engine.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import { statusCardRefreshPolicySchema } from "@paperclipai/shared"; import { chooseStatusCardUpdateKind, @@ -13,6 +13,7 @@ import { } from "../services/status-card-update-engine.js"; describe("status card update engine", () => { + afterEach(() => vi.restoreAllMocks()); const defaultPolicy = statusCardRefreshPolicySchema.parse({ mode: "interval", intervalMinutes: 15 }); it("retains non-terminal and terminal status transitions plus membership changes", () => { @@ -101,6 +102,127 @@ describe("status card update engine", () => { expect(nextStatusCardPolicyEligibility({ policy, now, lastChangeAt: now, completedUpdateTimes: [], tokensToday: 0 })).toBeNull(); }); + it("bounds empty-window formatter work before any timezone conversion", () => { + const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, + activeHours: { start: "09:00", end: "09:00", timezone: "Pacific/Chatham" } }); + const formatter = vi.spyOn(Intl, "DateTimeFormat").mockImplementation(function () { + throw new Error("Empty active hours must not construct a formatter"); + }); + const now = new Date("2026-07-23T14:00:00Z"); + expect(isWithinStatusCardActiveHours(policy, now)).toBe(false); + expect(nextStatusCardPolicyEligibility({ policy, now, lastChangeAt: now, completedUpdateTimes: [], tokensToday: 0 })).toBeNull(); + expect(formatter).not.toHaveBeenCalled(); + }); + + it("bounds formatter construction and reuses the next opening across cards and source writes", () => { + const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, + activeHours: { start: "09:00", end: "17:00", timezone: "America/St_Johns" } }); + const RealFormatter = Intl.DateTimeFormat; + let constructions = 0; + vi.spyOn(Intl, "DateTimeFormat").mockImplementation(function (...args: ConstructorParameters) { + if (++constructions > 1) throw new Error("Card evaluation repeatedly constructed timezone formatters"); + return new RealFormatter(...args); + } as unknown as typeof Intl.DateTimeFormat); + const conversions = vi.spyOn(RealFormatter.prototype, "formatToParts"); + const now = new Date("2026-07-23T22:00:00Z"); + const input = { policy, now, lastChangeAt: new Date(now.getTime() - 60_000), completedUpdateTimes: [], tokensToday: 0 }; + const started = performance.now(); + expect(nextStatusCardPolicyEligibility(input)).toEqual(new Date("2026-07-24T11:30:00Z")); + const coldMs = performance.now() - started; + const coldConversions = conversions.mock.calls.length; + expect(coldConversions).toBeLessThanOrEqual(2881); + conversions.mockClear(); + const warmStarted = performance.now(); + for (let write = 0; write < 100; write++) for (let card = 0; card < 20; card++) { + expect(nextStatusCardPolicyEligibility({ ...input, + policy: { ...policy, activeHours: { ...policy.activeHours! } }, + now: new Date(now.getTime() + write * 1000), + })).toEqual(new Date("2026-07-24T11:30:00Z")); + } + expect(constructions).toBeLessThanOrEqual(1); + expect(conversions.mock.calls.length).toBeLessThanOrEqual(2000); + process.stdout.write(JSON.stringify({ measurement: "status-card active-hours cost", coldMs, coldConversions, constructions, + warmMs: performance.now() - warmStarted, warmConversions: conversions.mock.calls.length, evaluations: 2000 }) + "\n"); + }); + + it.each([ + ["America/New_York", "02:15", "03:30", "2026-03-08T06:50:00Z", "2026-03-08T07:00:00Z"], + ["America/New_York", "02:15", "02:45", "2026-03-08T06:50:00Z", "2026-03-09T06:15:00Z"], + ["America/New_York", "02:15", "02:45", "2026-03-07T07:45:00Z", "2026-03-09T06:15:00Z"], + ["America/New_York", "01:15", "01:45", "2026-11-01T05:50:00Z", "2026-11-01T06:15:00Z"], + ["America/New_York", "23:30", "01:30", "2026-11-01T05:40:00Z", "2026-11-01T06:00:00Z"], + ["Australia/Lord_Howe", "02:15", "02:45", "2026-10-03T15:20:00Z", "2026-10-03T15:30:00Z"], + ["Australia/Lord_Howe", "01:35", "01:45", "2026-04-04T14:50:00Z", "2026-04-04T15:05:00Z"], + ["Asia/Kathmandu", "09:00", "17:00", "2026-07-23T12:00:00Z", "2026-07-24T03:15:00Z"], + ["Pacific/Apia", "00:10", "00:20", "2011-12-30T09:50:00Z", "2011-12-30T10:10:00Z"], + ])("preserves the first real opening in %s %s-%s after %s", (timezone, start, end, time, expected) => { + const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, activeHours: { start, end, timezone } }); + const now = new Date(time); + const input = { policy, now, lastChangeAt: new Date(now.getTime() - 60_000), completedUpdateTimes: [], tokensToday: 0 }; + expect(nextStatusCardPolicyEligibility(input)).toEqual(new Date(expected)); + // A later closed instant reuses the opening, while the exact opening itself + // remains eligible without rounding it into the next minute or day. + expect(nextStatusCardPolicyEligibility({ ...input, now: new Date(now.getTime() + 1000) })).toEqual(new Date(expected)); + const opening = new Date(expected); + expect(nextStatusCardPolicyEligibility({ ...input, now: opening })).toEqual(opening); + expect(nextStatusCardPolicyEligibility({ ...input, now: new Date(opening.getTime() + 1) })).toEqual(new Date(opening.getTime() + 1)); + }); + + it("invalidates an upcoming opening for clock rollback and changed policy values", () => { + const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, + activeHours: { start: "09:00", end: "17:00", timezone: "UTC" } }); + const calculate = (time: string, activeHours = policy.activeHours!) => { + const now = new Date(time); + return nextStatusCardPolicyEligibility({ policy: { ...policy, activeHours }, now, + lastChangeAt: new Date(now.getTime() - 60_000), completedUpdateTimes: [], tokensToday: 0 }); + }; + expect(calculate("2026-07-23T18:00:00Z")).toEqual(new Date("2026-07-24T09:00:00Z")); + expect(calculate("2026-07-23T08:00:00Z")).toEqual(new Date("2026-07-23T09:00:00Z")); + expect(calculate("2026-07-23T17:00:00Z")).toEqual(new Date("2026-07-24T09:00:00Z")); + expect(calculate("2026-07-23T18:00:00Z", { ...policy.activeHours!, start: "10:00" })).toEqual(new Date("2026-07-24T10:00:00Z")); + expect(calculate("2026-07-23T18:00:00Z", { ...policy.activeHours!, end: "19:00" })).toEqual(new Date("2026-07-23T18:00:00Z")); + expect(calculate("2026-07-23T18:00:00Z", { ...policy.activeHours!, timezone: "Asia/Tokyo" })).toEqual(new Date("2026-07-24T00:00:00Z")); + }); + + it("checks the precise instant before a cached minute opening for historical second offsets", () => { + const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, + activeHours: { start: "09:00", end: "10:00", timezone: "Africa/Monrovia" } }); + const now = new Date("1970-01-01T08:00:00Z"); + const input = { policy, now, lastChangeAt: new Date(now.getTime() - 60_000), completedUpdateTimes: [], tokensToday: 0 }; + expect(nextStatusCardPolicyEligibility(input)).toEqual(new Date("1970-01-01T09:45:00Z")); + const inside = new Date("1970-01-01T09:44:45Z"); + expect(isWithinStatusCardActiveHours(policy, inside)).toBe(true); + expect(nextStatusCardPolicyEligibility({ ...input, now: inside })).toEqual(inside); + }); + + it("bounds formatter and upcoming-opening cache retention", () => { + const now = new Date("2026-07-25T00:00:00Z"); + const timezonePolicy = (timezone: string) => statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, + activeHours: { start: "09:00", end: "17:00", timezone } }); + const zones = Intl.supportedValuesOf("timeZone").slice(0, 34); + const first = timezonePolicy(zones[0]!); + isWithinStatusCardActiveHours(first, now); + for (const zone of zones.slice(1)) isWithinStatusCardActiveHours(timezonePolicy(zone), now); + const RealFormatter = Intl.DateTimeFormat; + const constructor = vi.spyOn(Intl, "DateTimeFormat").mockImplementation(function (...args: ConstructorParameters) { + return new RealFormatter(...args); + } as unknown as typeof Intl.DateTimeFormat); + isWithinStatusCardActiveHours(first, now); + expect(constructor).toHaveBeenCalledTimes(1); + constructor.mockRestore(); + const time = (minute: number) => `${String(Math.floor(minute / 60)).padStart(2, "0")}:${String(minute % 60).padStart(2, "0")}`; + const calculate = (minute: number) => { + const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1, + activeHours: { start: time(minute), end: time(minute + 1), timezone: "UTC" } }); + return nextStatusCardPolicyEligibility({ policy, now, lastChangeAt: new Date(now.getTime() - 60_000), completedUpdateTimes: [], tokensToday: 0 }); + }; + calculate(1); + for (let minute = 2; minute <= 258; minute++) calculate(minute); + const conversions = vi.spyOn(Intl.DateTimeFormat.prototype, "formatToParts"); + expect(calculate(1)).toEqual(new Date("2026-07-25T00:01:00Z")); + expect(conversions.mock.calls.length).toBeGreaterThan(1); + }); + it("selects full rebuilds for bounded drift rules and incremental otherwise", () => { const base = { hasDocument: true, changeCount: 2, queryVersion: 3, lastUpdateQueryVersion: 3, incrementalCount: 2, configurationChanged: false }; expect(chooseStatusCardUpdateKind(base)).toBe("incremental"); diff --git a/server/src/services/status-card-update-engine.ts b/server/src/services/status-card-update-engine.ts index 1f734227808..0b1e5325446 100644 --- a/server/src/services/status-card-update-engine.ts +++ b/server/src/services/status-card-update-engine.ts @@ -109,21 +109,36 @@ export function statusCardFingerprintHash(fingerprint: StatusCardFingerprint) { return createHash("sha256").update(JSON.stringify(stable)).digest("hex"); } +// These caches contain only timezone/policy calculations, never card or company +// data. Bound both maps because timezone aliases and hour pairs are user input. +const activeHoursFormatters = new Map(); +const upcomingActiveHours = new Map(); + +function remember(cache: Map, key: K, value: V, limit: number): V { + cache.delete(key); + cache.set(key, value); + if (cache.size > limit) cache.delete(cache.keys().next().value!); + return value; +} + +function activeHoursFormatter(timezone: string) { + const cached = activeHoursFormatters.get(timezone); + return remember(activeHoursFormatters, timezone, cached ?? new Intl.DateTimeFormat("en-GB", { + timeZone: timezone, hour: "2-digit", minute: "2-digit", hourCycle: "h23", + }), 32); +} + export function isWithinStatusCardActiveHours(policy: StatusCardRefreshPolicy, now: Date) { if (!policy.activeHours) return true; - const parts = new Intl.DateTimeFormat("en-GB", { - timeZone: policy.activeHours.timezone, - hour: "2-digit", - minute: "2-digit", - hourCycle: "h23", - }).formatToParts(now); - const hour = Number(parts.find((part) => part.type === "hour")?.value ?? 0); - const minute = Number(parts.find((part) => part.type === "minute")?.value ?? 0); - const current = hour * 60 + minute; const [startHour, startMinute] = policy.activeHours.start.split(":").map(Number); const [endHour, endMinute] = policy.activeHours.end.split(":").map(Number); const start = startHour! * 60 + startMinute!; const end = endHour! * 60 + endMinute!; + if (start === end) return false; + const parts = activeHoursFormatter(policy.activeHours.timezone).formatToParts(now); + const hour = Number(parts.find((part) => part.type === "hour")?.value ?? 0); + const minute = Number(parts.find((part) => part.type === "minute")?.value ?? 0); + const current = hour * 60 + minute; return start <= end ? current >= start && current < end : current >= start || current < end; } @@ -181,6 +196,8 @@ export function nextStatusCardPolicyEligibility(input: { tokensToday: number; }): Date | null { if (input.policy.mode === "manual") return null; + const hours = input.policy.activeHours; + if (hours && hours.start === hours.end) return null; let at = input.now.getTime(); if (input.tokensToday >= (input.policy.dailyTokenCap ?? 100_000)) { const day = new Date(input.now); @@ -194,13 +211,25 @@ export function nextStatusCardPolicyEligibility(input: { .filter(time => time > at - 3_600_000).sort((a, b) => a - b); if (recent.length >= cap) at = Math.max(at, recent[recent.length - cap]! + 3_600_000 + 1); } - if (isWithinStatusCardActiveHours(input.policy, new Date(at))) return new Date(at); + // Check the precise instant before reusing a minute-aligned opening. Some + // historical timezone offsets include seconds, so a window may be open now. + if (!hours || isWithinStatusCardActiveHours(input.policy, new Date(at))) return new Date(at); + const key = JSON.stringify([hours.timezone, hours.start, hours.end]); + const cached = upcomingActiveHours.get(key); + if (cached && at >= cached.closedAt && at < cached.opensAt) { + remember(upcomingActiveHours, key, cached, 256); + return new Date(cached.opensAt); + } + const closedAt = at; // Active-hour precision is one local minute. Walking UTC minutes also honors // skipped/repeated local times at daylight-saving transitions. const end = at + 48 * 3_600_000; for (at = Math.floor(at / 60_000) * 60_000 + 60_000; at <= end; at += 60_000) { - if (isWithinStatusCardActiveHours(input.policy, new Date(at))) return new Date(at); + if (isWithinStatusCardActiveHours(input.policy, new Date(at))) { + remember(upcomingActiveHours, key, { closedAt, opensAt: at }, 256); + return new Date(at); + } } - // Equal start/end describes an empty window; retain changes, awaiting config. + // No opening inside the existing bounded search horizon. Retain changes. return null; }