Merge commit 'e4b1ed83e4c4b029a4f6f82e3537e1129076814e' into codex/sleep-09-aggregate

* commit 'e4b1ed83e4c4b029a4f6f82e3537e1129076814e':
  fix: bound status-card active-hours calculation work
  fix: scope comment-triggered recovery to the affected issue
  test: await issue recovery dispatch and cancellation settlement
  fix(server): bound blocker facts and preserve origin authority
  test: await committed recovery and adapter admission outcomes
  test: model retention reads in native cleanup transaction fixture
This commit is contained in:
Dotta committed 2026-10-30 19:45:34 -05:00
commit fa54edbcd3
22 files changed
+601 -106

No files matched your search

@@ -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" });
@@ -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<string, unknown>).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<string, unknown>;
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(); }
});
@@ -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] });
@@ -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";
@@ -8,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";
@@ -27,6 +29,7 @@ describe.skipIf(!support.supported)("issue recovery production producers and dea
let draining = false;
const tasks = new Map<string, { nextRunAt?: () => number | null; hasPending: () => Promise<boolean> }>();
const errors: unknown[] = [];
const passes = new Set<Promise<unknown>>();
const extraStops: Array<() => Promise<void>> = [];
const extraDeadlines: Array<() => number | null> = [];
beforeAll(async () => {
@@ -52,7 +55,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 +99,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 +108,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);
@@ -196,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 });
+1 -1
View File
@@ -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);
@@ -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<typeof Intl.DateTimeFormat>) {
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();
@@ -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<typeof Intl.DateTimeFormat>) {
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<typeof Intl.DateTimeFormat>) {
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");
+26 -2
View File
@@ -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<Record<string, unknown>> = {})
};
}
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<unknown>) => await callback(tx));
const transaction = vi.fn(async (callback: (input: typeof tx) => Promise<unknown>) => {
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<unknown>) => await callback(tx));
const transaction = vi.fn(async (callback: (input: typeof tx) => Promise<unknown>) => {
transactionOpen = true;
try { return await callback(tx); } finally { transactionOpen = false; }
});
const svc = workProductService({ transaction } as any);
const result = await svc.update("work-product-1", {
+9 -2
View File
@@ -32,6 +32,7 @@ type PageEvidence = {
agents: Set<string>;
visibleIssues: Map<string, PageIssue>;
newerRuns: Set<string>;
deferredRoots: Set<string>;
blockers: Awaited<ReturnType<ReturnType<typeof issueService>["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<string>() };
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<string, boolean>, interactionAttention?: Map<string, boolean>, evidence?: PageEvidence): Promise<Entry | null> {
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;
+2 -1
View File
@@ -9,7 +9,8 @@ type WriteDb = Db | Parameters<Parameters<Db["transaction"]>[0]>[0];
export function insertIssueComment(db: WriteDb, input: PgInsertValue<typeof issueComments> & { companyId: string }) {
return db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, input.companyId);
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();
});
@@ -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);
}
}
@@ -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<Record<string, unknown>> = [];
const inserts: Array<{ table: string; values: Record<string, unknown> }> = [];
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<void>) => callback(db),
transaction: async (callback: (tx: typeof db) => Promise<void>) => {
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;
+65 -55
View File
@@ -3443,6 +3443,7 @@ type IssueBlockerAttentionEdge = {
blockerIssueId: string;
};
type IssueBlockerAttentionQueryRow = IssueBlockerAttentionNode & {
executionWorkspaceId?: string | null;
issueId: string | null;
blockerIssueId: string;
};
@@ -3831,11 +3832,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<Map<string, IssueBlockerAttention>> {
) {
const roots = issueRows.filter(
(row) => row.companyId === companyId && row.status === "blocked",
);
@@ -3845,37 +3850,38 @@ async function listIssueBlockerAttentionMap(
attentionMap.set(row.id, createIssueBlockerAttention());
}
}
if (roots.length === 0) return attentionMap;
const deferredRootIds = new Set<string>(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<string, IssueBlockerAttentionNode>();
const edgesByIssueId = new Map<string, IssueBlockerAttentionEdge[]>();
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<string>; expanded: Set<string>; frontier: string[]; truncated: boolean };
const traversals = new Map<string, Traversal>(roots.map(root => [root.id, {
nodes: new Set([root.id]), expanded: new Set<string>(), frontier: [root.id], truncated: false,
type Traversal = { rootId: string; nodes: Set<string>; expanded: Set<string>; frontier: string[]; truncated: boolean };
const traversals = new Map<string, Traversal>(admittedRoots.map(root => [root.id, {
rootId: root.id, nodes: new Set([root.id]), expanded: new Set<string>(), frontier: [root.id], truncated: false,
}]));
const expanded = new Set<string>();
const queried = new Set<string>();
const pendingFinalizeBlockerIssueIds = new Set<string>();
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<string>();
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
@@ -3889,6 +3895,7 @@ async function listIssueBlockerAttentionMap(
title: issues.title,
status: issues.status,
executionRunId: issues.executionRunId,
executionWorkspaceId: issues.executionWorkspaceId,
assigneeAgentId: issues.assigneeAgentId,
assigneeUserId: issues.assigneeUserId,
})
@@ -3901,7 +3908,7 @@ async function listIssueBlockerAttentionMap(
inArray(issueRelations.relatedIssueId, chunk),
eq(issues.companyId, companyId),
),
);
).limit(rowLimit);
const childRowsPromise: Promise<IssueBlockerAttentionQueryRow[]> = dbOrTx
.select({
issueId: issues.parentId,
@@ -3926,56 +3933,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<string>();
for (const id of traversal.frontier) {
traversal.expanded.add(id);
@@ -4555,7 +4562,7 @@ async function listIssueBlockerAttentionMap(
);
}
return attentionMap;
return result();
}
type IssueReviewAttentionInput = Pick<IssueRow, "id" | "companyId" | "status">;
@@ -9393,6 +9400,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[],
@@ -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 =
@@ -8,7 +8,7 @@ export type NativeStatusReassessmentScope = { companyId: string; issueId?: strin
export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
await notifyStatusCardIssueChange(tx, scope.companyId);
await notifyIssueAttentionRetentionWork(tx, { companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId });
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 }));
}
+3 -2
View File
@@ -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,
@@ -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`, `)}])
@@ -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);
@@ -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<string, Intl.DateTimeFormat>();
const upcomingActiveHours = new Map<string, { closedAt: number; opensAt: number }>();
function remember<K, V>(cache: Map<K, V>, 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;
}