refactor: drive status-card evaluation from committed work

Move generation sequencing into a feature-owned deadline worker and retain persisted evaluation, eligibility and relative-search obligations. Add transactional producer hints and initial PostgreSQL acceptance coverage; follow-up completes the producer audit and remaining lifecycle gates.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-30 17:25:18 -05:00
1 parent dbd400ce17
commit 4af412eaea
39 files changed
+561 -93

No files matched your search

@@ -0,0 +1,180 @@
import { randomUUID } from "node:crypto";
import { eq, sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { activityLog, agentWakeupRequests, agents, companies, companyMemberships, createDb, documents,
heartbeatRuns, instanceSettings, issueComments, issues, statusCards, statusCardUpdates, documentRevisions } from "@paperclipai/db";
import { companySearchQuerySchema, statusCardRefreshPolicySchema } from "@paperclipai/shared";
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { registerStatusCardEvaluationWork, statusCardEvaluationService } from "../services/status-card-evaluation.js";
import { statusCardService } from "../services/status-cards.js";
import { issueService } from "../services/issues.js";
import { instanceSettingsService } from "../services/instance-settings.js";
import { idleWorkSnapshot } from "../services/task-admission.js";
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STATUS_CARD_EVALUATION_TEST_DATABASE_URL");
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("status-card evaluation work", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
let db: ReturnType<typeof createDb>;
let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
let draining = false;
let companyId: string, agentId: string;
const errors: unknown[] = [];
const actor = { agentId: null, userId: "board-user" };
const policy = (mode: "manual" | "reactive" | "interval" = "reactive") => statusCardRefreshPolicySchema.parse({
mode, debounceSeconds: 60, intervalMinutes: 15,
});
// Controlled provider boundary: assignment persists a queued wake, and tests
// explicitly run the authorized summarizer writeback. No model/provider runs.
const wakeup = vi.fn(async (id: string, options: { payload?: Record<string, unknown> | null; reason?: string | null }) => {
return db.insert(agentWakeupRequests).values({ companyId, agentId: id, source: "assignment",
reason: options.reason, payload: options.payload, status: "queued" }).returning();
});
beforeAll(async () => {
if (!external) temporary = await startEmbeddedPostgresTestDatabase("status-card-evaluation-");
db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
}, 30_000);
beforeEach(async () => {
companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0; wakeup.mockClear();
await db.insert(companies).values({ id: companyId, name: "Status", issuePrefix: `S${companyId.slice(0, 6)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Summarizer", role: "general", status: "idle", adapterType: "process", adapterConfig: {} });
await instanceSettingsService(db).updateExperimental({ enableStatusCards: true });
await Promise.all(Array.from({ length: 6 }, () => db.execute(sql`select pg_sleep(0.01)`)));
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
});
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
vi.useRealTimers(); vi.restoreAllMocks();
await db.delete(agentWakeupRequests); await db.delete(statusCardUpdates); await db.delete(statusCards);
await db.delete(documentRevisions); await db.delete(documents); await db.delete(activityLog);
await db.delete(issueComments); await db.delete(issues); await db.delete(heartbeatRuns);
await db.delete(instanceSettings); await db.delete(agents); await db.delete(companyMemberships); await db.delete(companies);
});
afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
async function start() {
coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, canReconcile: () => true,
onError: error => errors.push(error) });
await registerStatusCardEvaluationWork(coordinator, statusCardEvaluationService(db, { wakeup }), () => !draining).ready;
}
async function settled() {
await vi.waitFor(async () => {
await vi.advanceTimersByTimeAsync(30_000);
expect(idleWorkSnapshot().active).toBe(0);
}, { timeout: 5000 });
}
async function quiet() {
await settled();
expect(coordinator!.nextWakeAt()).toBeNull();
const reads = vi.spyOn(db, "select"), writes = vi.spyOn(db, "update"), executes = vi.spyOn(db, "execute");
const calls = wakeup.mock.calls.length;
await vi.advanceTimersByTimeAsync(10 * 60_000);
expect(reads).not.toHaveBeenCalled(); expect(writes).not.toHaveBeenCalled(); expect(executes).not.toHaveBeenCalled();
expect(wakeup).toHaveBeenCalledTimes(calls);
reads.mockRestore(); writes.mockRestore(); executes.mockRestore();
}
async function writeback(cardId: string, generationIssueId: string, compile = false) {
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning();
await db.update(issues).set({ checkoutRunId: run!.id }).where(eq(issues.id, generationIssueId));
const writer = { agentId, runId: run!.id };
const cards = statusCardService(db);
if (compile) await cards.writeQuery(cardId, { queries: [companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"] })],
title: "Blockers", changeSummary: "Compiled", generationIssueId }, writer);
await cards.writeSummary(cardId, { markdown: "Current blockers", changeSummary: "Updated", generationIssueId }, writer);
await issueService(db).update(generationIssueId, { status: "done" });
}
async function card(mode: "manual" | "reactive" | "interval" = "reactive") {
const cards = statusCardService(db);
const created = await cards.create(companyId, { interestPrompt: "Blockers", titlePinned: false, agentId, refreshPolicy: policy(mode) }, actor);
const compile = await cards.requestCompile(created.id, actor);
await writeback(created.id, compile.generatingIssue.id, true);
return cards.getById(created.id).then(row => row!);
}
it("settles empty, disabled, configured manual and unchanged reactive history without recurring SQL", async () => {
await start(); await quiet();
await card("manual"); await card("reactive"); await quiet();
await instanceSettingsService(db).updateExperimental({ enableStatusCards: false }); await quiet();
expect(errors).toEqual([]);
});
it("ordinary issue change schedules debounce, assigns one durable wake, writes a result and becomes quiet", async () => {
await start(); const created = await card(); await quiet();
const changed = await issueService(db).create(companyId, { title: "New blocker", status: "blocked", priority: "medium", createdByUserId: "board-user" });
await vi.waitFor(async () => {
expect((await statusCardService(db).getById(created.id))?.pendingChangeCount).toBeGreaterThan(0);
}, { timeout: 5000 });
const pending = await statusCardService(db).getById(created.id);
expect(pending?.pendingChangeCount).toBeGreaterThan(0);
expect(pending?.nextEvalAt).toBeInstanceOf(Date);
expect(wakeup).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(60_000); await settled();
const generating = await statusCardService(db).getById(created.id);
expect(generating?.generatingIssueId).toBeTruthy();
const durable = await db.select().from(agentWakeupRequests);
expect(durable).toHaveLength(1); expect(durable[0]?.payload?.issueId).toBe(generating?.generatingIssueId);
expect(changed.companyId).toBe(companyId);
await writeback(created.id, generating!.generatingIssueId!); await quiet();
expect(await db.select().from(statusCardUpdates).then(rows => rows.filter(row => row.generationIssueId === generating?.generatingIssueId)))
.toMatchObject([{ status: "ok", finishedAt: expect.any(Date) }]);
expect(errors).toEqual([]);
});
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();
created = await statusCardService(db).update(created, { refreshPolicy: { ...policy("interval"), intervalMinutes: 1 } }, actor);
await settled(); expect(coordinator!.nextWakeAt()).toBeLessThan(old!);
await statusCardService(db).update(created, { archived: true }, actor); await quiet();
expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]);
});
it("does not react to rolled-back or nested uncommitted card configuration", async () => {
await start(); const created = await card("manual"); await quiet();
await expect(db.transaction(async tx => {
await statusCardService(tx as unknown as typeof db).update(created, { refreshPolicy: policy("interval") }, actor);
await vi.advanceTimersByTimeAsync(30_000); expect(wakeup).not.toHaveBeenCalled();
throw new Error("rollback");
})).rejects.toThrow("rollback");
await quiet(); expect((await statusCardService(db).getById(created.id))?.nextEvalAt).toBeNull();
await db.transaction(async tx => {
await tx.transaction(async nested => {
await statusCardService(nested as unknown as typeof db).update(created, { refreshPolicy: policy("interval") }, actor);
});
expect(coordinator!.nextWakeAt()).not.toBeNull();
expect((await statusCardService(db).getById(created.id))?.nextEvalAt).toBeNull();
});
await settled(); expect(coordinator!.nextWakeAt()).not.toBeNull(); expect(errors).toEqual([]);
});
it("serializes competing connections when one source event preempts a future evaluation", async () => {
const created = await card();
await db.update(statusCards).set({ nextEvalAt: new Date(Date.now() + 3_600_000) }).where(eq(statusCards.id, created.id));
await issueService(db).create(companyId, { title: "Concurrent blocker", status: "blocked", priority: "medium", createdByUserId: "board-user" });
const first = statusCardEvaluationService(db, { wakeup });
const second = statusCardEvaluationService(db, { wakeup });
first.changed(new Set([`changes:${companyId}`])); second.changed(new Set([`changes:${companyId}`]));
await Promise.all([first.run(new AbortController().signal), second.run(new AbortController().signal)]);
const pending = await statusCardService(db).getById(created.id);
expect(pending?.pendingChangeCount).toBeGreaterThan(0);
expect(pending?.nextEvalAt?.getTime()).toBeLessThan(Date.now() + 3_600_000);
expect(wakeup).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(60_000);
await Promise.all([first.run(new AbortController().signal), second.run(new AbortController().signal)]);
const updated = await statusCardService(db).getById(created.id);
expect(updated?.generatingIssueId).toBeTruthy(); expect(wakeup).toHaveBeenCalledOnce();
expect(await db.select().from(statusCardUpdates).then(rows => rows.filter(row => row.generationIssueId === updated?.generatingIssueId)))
.toHaveLength(1);
});
it("preserves the first write during drain until resume", async () => {
await start(); const created = await card("manual"); await quiet();
draining = true;
await statusCardService(db).update(created, { refreshPolicy: policy("interval") }, actor);
await settled(); expect(wakeup).not.toHaveBeenCalled();
draining = false; await vi.advanceTimersByTimeAsync(30_000); await settled();
expect(coordinator!.nextWakeAt()).not.toBeNull(); expect(errors).toEqual([]);
});
});
@@ -8,6 +8,7 @@ import {
filterStatusCardChanges,
isWithinStatusCardActiveHours,
nextStatusCardEvaluationAt,
nextStatusCardPolicyEligibility,
statusCardChangesHash,
} from "../services/status-card-update-engine.js";
@@ -78,6 +79,28 @@ describe("status card update engine", () => {
expect(evaluateStatusCardPolicy({ policy: hours, now: new Date("2026-07-23T18:00:00.000Z"), lastChangeAt: null, updatesLastHour: 0, tokensToday: 0, manual: false }).action).toBe("pause_hours");
});
it("disarms unchanged reactive evaluation while retaining exact pending eligibility", () => {
const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 90, maxUpdatesPerHour: 2,
activeHours: { start: "09:00", end: "17:00", timezone: "UTC" } });
const now = new Date("2026-07-23T16:59:30Z");
expect(nextStatusCardEvaluationAt(policy, now)).toBeNull();
expect(nextStatusCardPolicyEligibility({ policy, now, lastChangeAt: now, completedUpdateTimes: [], tokensToday: 0 }))
.toEqual(new Date("2026-07-24T09:00:00Z"));
const open = new Date("2026-07-23T14:00:00Z");
expect(nextStatusCardPolicyEligibility({ policy, now: open, lastChangeAt: new Date("2026-07-23T13:00:00Z"),
completedUpdateTimes: [new Date("2026-07-23T13:30:00Z"), new Date("2026-07-23T13:45:00Z")], tokensToday: 0 }))
.toEqual(new Date("2026-07-23T14:30:00.001Z"));
expect(nextStatusCardPolicyEligibility({ policy, now: open, lastChangeAt: open, completedUpdateTimes: [], tokensToday: 100_000 }))
.toEqual(new Date("2026-07-24T09:00:00Z"));
});
it("retains empty-window pending changes without a recurring eligibility scan", () => {
const policy = statusCardRefreshPolicySchema.parse({ mode: "reactive", debounceSeconds: 1,
activeHours: { start: "09:00", end: "09:00", timezone: "UTC" } });
const now = new Date("2026-07-23T14:00:00Z");
expect(nextStatusCardPolicyEligibility({ policy, now, lastChangeAt: now, completedUpdateTimes: [], tokensToday: 0 })).toBeNull();
});
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");
+6 -3
View File
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { randomUUID } from "node:crypto";
import express from "express";
import request from "supertest";
@@ -37,7 +38,8 @@ import {
startEmbeddedPostgresTestDatabase,
} from "./helpers/embedded-postgres.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_STATUS_CARDS_TEST_DATABASE_URL");
const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
type Db = ReturnType<typeof createDb>;
@@ -79,8 +81,8 @@ describeEmbeddedPostgres("status card routes", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-status-cards-");
db = createDb(tempDb.connectionString);
if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-status-cards-");
db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
}, 20_000);
afterEach(async () => {
@@ -100,6 +102,7 @@ describeEmbeddedPostgres("status card routes", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
+5 -30
View File
@@ -145,13 +145,12 @@ import {
reconcileCodexLocalManagedHomesOnStartup,
reconcilePersistedRuntimeServicesOnStartup,
routineService,
statusCardService,
toolAccessService,
workspaceOperationService,
} from "./services/index.js";
import { questionResponseDeliveryService } from "./services/question-response-delivery.js";
import { deliverNativeQuestionResponse } from "./services/native-runtime/native-question-bridge.js";
import { queueIssueAssignmentWakeup } from "./services/issue-assignment-wakeup.js";
import { registerStatusCardEvaluationWork, statusCardEvaluationService } from "./services/status-card-evaluation.js";
import { createSecretProposalsService } from "./services/secret-proposals.js";
import { environmentRuntimeService } from "./services/environment-runtime.js";
import { historicalCleanupVerificationService } from "./services/historical-cleanup-verification.js";
@@ -1569,7 +1568,7 @@ async function startServerWithDatabaseTeardown(
if (result.settled > 0) logger.info(result, "reconciled routine execution issue status");
}).catch(() => logger.warn("routine execution status reconciliation deferred")));
};
const statusCards = statusCardService(db as any);
const statusCardEvaluation = statusCardEvaluationService(db as any, heartbeat);
const issues = issueService(db as any);
const mergedPullRequestConfirmations = issueThreadInteractionService(db as any, {
wakeup: heartbeat.wakeup,
@@ -1980,6 +1979,8 @@ async function startServerWithDatabaseTeardown(
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
await step("join_request_retention", () => registerJoinRequestRetentionWork(deliveryWork, joinRequestRetentionService(db as any),
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
await step("status_card_evaluation", () => registerStatusCardEvaluationWork(deliveryWork, statusCardEvaluation,
() => !isIdleTaskDrainActive() && !isWarmStandby() && !heartbeatSchedulerStopped).ready);
await step("retention_sweep", runRetentionSweep);
};
@@ -2010,33 +2011,7 @@ async function startServerWithDatabaseTeardown(
scheduleGitHubConnectionContinuitySweep();
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
trackHeartbeatSchedulerWork((async () => {
const experimental = await instanceSettingsService(db).getExperimental();
if (experimental.enableStatusCards !== true) return;
const result = await statusCards.tickDueStatusCards(new Date());
await Promise.all(result.enqueued.map(async ({ cardId, generatingIssue }) => {
try {
await queueIssueAssignmentWakeup({
heartbeat,
issue: generatingIssue,
reason: "status_card_update_assigned",
mutation: "status_card.scheduler_update_requested",
contextSource: "status_card_scheduler",
requestedByActorType: "system",
taskKey: `status-card:${cardId}`,
rethrowOnError: true,
});
} catch (err) {
await issues.update(generatingIssue.id, { status: "cancelled" });
throw err;
}
}));
if (result.evaluated > 0 || result.enqueued.length > 0) {
logger.info({ evaluated: result.evaluated, enqueued: result.enqueued.length }, "status-card scheduler tick complete");
}
})().catch((err) => {
logger.error({ err }, "status-card scheduler tick failed");
}));
if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return;
if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) {
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../../../services/delivery-work-notifications.js";
import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
@@ -385,6 +386,7 @@ export function createPostgresWatchdogAdapter(db: Db): WatchdogRunReader & Watch
));
}
await notifyStatusCardIssueChange(tx, companyId);
await tx
.update(issues)
.set({
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -745,6 +746,7 @@ export function accessService(db: Db) {
await notifyNativeStatusReassessment(tx, { companyId });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, companyId);
const resetInProgress = await tx
.update(issues)
.set({
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -38,6 +39,7 @@ export async function deleteAgentDependencies(tx: Db, companyId: string, id: str
await issueThreadInteractionService(tx).cancelPendingForDeletedAddressee(companyId, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, companyId);
await tx.update(agents).set({ reportsTo: null }).where(eq(agents.reportsTo, id));
await tx
.update(issues)
+3
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -336,6 +337,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyStatusCardIssueChange(tx, id);
await notifyCompanyRetentionWork(tx, id);
await notifyChatInboundWakeNotices(tx, { companyId: id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
@@ -492,6 +494,7 @@ export function companyService(db: Db, _budgetHooks: BudgetServiceHooks = {}) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyStatusCardIssueChange(tx, id);
await notifyCompanyRetentionWork(tx, id);
await notifyChatInboundWakeNotices(tx, { companyId: id });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentLifecycle, id);
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { notifyCancelledWorkspacePreservation } from "./cancelled-workspace-preservation-notifications.js";
import { notifyInstructionCollection } from "./instruction-collection-notifications.js";
@@ -77,6 +78,7 @@ export async function deleteCompany(db: Db, id: string) {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillPackages);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyStatusCardEvaluation(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.secretProposalExpiry, id);
await notifyToolReviewChange(tx, id);
await agentObservabilityService(tx as unknown as Db).removeForCompanyDeletion(id);
+8 -2
View File
@@ -207,14 +207,20 @@ function priorityRank(priority: string | null | undefined) {
return index >= 0 ? index : ISSUE_PRIORITIES.length;
}
function updatedWithinStart(value: string | undefined, now = new Date()): Date | null {
/** Duration used by saved search membership and its deadline consumers. */
export function companySearchUpdatedWithinMs(value: string | undefined): number | null {
if (!value) return null;
const match = /^(\d+)(h|d|w|m)$/.exec(value);
if (!match) return null;
const amount = Number.parseInt(match[1]!, 10);
const unit = match[2];
const hours = unit === "h" ? amount : unit === "d" ? amount * 24 : unit === "w" ? amount * 24 * 7 : amount * 24 * 30;
return new Date(now.getTime() - hours * 60 * 60 * 1000);
return hours * 60 * 60 * 1000;
}
function updatedWithinStart(value: string | undefined, now = new Date()): Date | null {
const duration = companySearchUpdatedWithinMs(value);
return duration === null ? null : new Date(now.getTime() - duration);
}
function issueOnlyFiltersActive(query: CompanySearchQuery) {
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifySkillVersionRetention } from "./skill-version-retention-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { verifyFileManifest } from "./file-manifest-cache.js";
@@ -7450,6 +7451,7 @@ export function companySkillService(db: Db, options: { actor?: AuthorizationActo
for (const row of rows) {
await db.transaction(async (tx) => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, companyId);
await tx
.update(issues)
.set({ hiddenAt: now, updatedAt: now })
@@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db";
/** Topics for existing durable queues; register intent before their writes. */
export const DELIVERY_QUEUES = {
statusCardEvaluation: "status-card-evaluation",
runEvidenceMigration: "run-evidence-migration",
providerTraceRetention: "provider-trace-retention",
nativeRunnerEvidenceRetention: "native-runner-evidence-retention",
@@ -1,7 +1,9 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import type { Db } from "@paperclipai/db";
import { DELIVERY_QUEUES, notifyDeliveryWork, type DeliveryQueue } from "./delivery-work-notifications.js";
export async function notifyExecutionRunWork(transaction: object) {
await notifyStatusCardIssueChange(transaction);
for (const queue of [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization,
DELIVERY_QUEUES.executionReplacement, DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors]) {
await notifyDeliveryWork(transaction, queue);
@@ -14,6 +16,7 @@ export function writeExecutionControl<T>(
write: (tx: Db) => PromiseLike<T>,
): Promise<T> {
return db.transaction(async transaction => {
await notifyStatusCardIssueChange(transaction);
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.issueMonitors);
for (const topic of typeof queue === "string" ? [queue] : queue) await notifyDeliveryWork(transaction, topic);
return write(transaction as unknown as Db);
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { effectiveProjectWorkspacePolicy } from "./project-workspace-provider.js";
import { inspectPluginWorkspace, removePluginWorkspace, restorePluginWorkspace, type WorkspaceWorkers } from "./plugin-workspaces.js";
import { acquireExecutionWorkspaceLifecycleLock, withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js";
@@ -1590,6 +1591,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
await validateSelection(input, tx);
const intent: TaskWorkspaceIntent = { version: 1, request: { key: input.requestKey, expectedBindingRevision: input.expectedBindingRevision }, selection: input.selection, source: input.source ?? "explicit" };
const nextRevision = task.workspaceBindingRevision + 1;
await notifyStatusCardIssueChange(tx, task.companyId);
await tx.update(issues).set({ workspaceBindingRevision: nextRevision,
workspacePendingSelection: { version: 1, requestKey: input.requestKey,
expectedBindingRevision: nextRevision, intent }, updatedAt: new Date() }).where(eq(issues.id, task.id));
@@ -1633,6 +1635,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
}
}
}
await notifyStatusCardIssueChange(tx, task.companyId);
await tx.update(issues).set({ ...patch, assigneeAdapterOverrides, workspaceSelection: pending.intent, workspacePendingSelection: null,
workspaceBindingRevision: task.workspaceBindingRevision + 1, updatedAt: new Date() }).where(eq(issues.id, task.id));
return true;
@@ -1651,6 +1654,7 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
const pendingAfterFirstBinding = !task.executionWorkspaceId && pending?.expectedBindingRevision === task.workspaceBindingRevision
? { ...pending, expectedBindingRevision: nextRevision }
: pending;
await notifyStatusCardIssueChange(tx, companyId);
const [bound] = await tx.update(issues).set({ ...patch, executionWorkspaceId: workspaceId,
workspaceBindingRevision: nextRevision, workspacePendingSelection: pendingAfterFirstBinding, updatedAt: new Date() })
.where(eq(issues.id, issueId)).returning();
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { deriveTaskKeyWithHeartbeatFallback } from "./run-state.js";
import { infrastructureRetryDelayMs } from "../system-recovery-state.js";
import {
@@ -1178,6 +1179,7 @@ export function createHeartbeatRetries(db: Db, dependencies: HeartbeatRetryDepen
}
if (issueId) {
await notifyStatusCardIssueChange(tx, run.companyId);
await tx
.update(issues)
.set({
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
import { notifySkillVersionRetentionReconstruction } from "./skill-version-retention-notifications.js";
import { notifyNativeEnvironmentSettlement } from "./native-environment-settlement-notifications.js";
import { notifyExternalAgentSettlement } from "../modules/external-agents/index.js";
@@ -625,6 +626,7 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService
}
const now = new Date();
const [updated] = await db.transaction(async tx => {
if (patch.enableStatusCards !== undefined) await notifyStatusCardEvaluation(tx);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.mcpEvents, "*");
await notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, "*");
await notifyDeliveryWork(tx, DELIVERY_QUEUES.grokDispatch);
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import type { PgInsertValue } from "drizzle-orm/pg-core";
import { issueComments, type Db } from "@paperclipai/db";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -6,6 +7,7 @@ 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 notifyDeliveryWork(tx, DELIVERY_QUEUES.dotEvents, `company:${input.companyId}`);
return tx.insert(issueComments).values(input).returning();
});
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { and, eq, sql } from "drizzle-orm";
import type { PgColumn } from "drizzle-orm/pg-core";
import type { Db } from "@paperclipai/db";
@@ -127,6 +128,7 @@ export async function rekeyCompanyIssueIdentifiers(
// overload, which returns NULL for a non-pattern argument.
const tailStart = sql`${fromPrefix.length + 1}::int`;
await notifyStatusCardIssueChange(tx, companyId);
const rekeyedIssues = await tx
.update(issues)
.set({ identifier: sql`${toPrefix} || substring(${issues.identifier} from ${tailStart})` })
@@ -1,9 +1,11 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { notifyNativeReviewResponse } from "./native-review-response-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
export async function notifyIssueReviewState(tx: object, scope: { companyId: string; issueId?: string; runId?: string; agentId?: string }) {
await notifyStatusCardIssueChange(tx, scope.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyExternalConversationSettlement(tx, scope);
await notifyNativeReviewResponse(tx, scope);
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { and, eq } from "drizzle-orm";
import { heartbeatRuns, issues, type Db } from "@paperclipai/db";
import { setIssueTitleSchema, type SetIssueTitle } from "@paperclipai/shared";
@@ -76,6 +77,7 @@ export async function setIssueTitle(
|| (issue.title === title && !issue.titleNeedsGeneration)) {
return { result: await saveReceipt({ id: issue.id, title: issue.title, titleNeedsGeneration: issue.titleNeedsGeneration, changed: false }), publication: null };
}
await notifyStatusCardIssueChange(tx, companyId);
await tx.update(issues).set({ title, titleNeedsGeneration: false, updatedAt: new Date() }).where(predicate);
await issueReferenceService(tx).syncIssue(issueId, tx);
const { publication } = await persistActivity(tx, {
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
@@ -1011,6 +1012,7 @@ export function issueTreeControlService(db: Db) {
await notifyChatResponseAvailability(tx, { companyId, issueId });
if (status === "blocked" || status === "in_review") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId });
}
await notifyStatusCardIssueChange(tx, companyId);
const rows = await tx
.update(issues)
.set({
+6
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { notifyNativeRecoveryNotices } from "./native-recovery-notice-notifications.js";
import { notifyNativeRunProjection } from "./native-run-projection-notifications.js";
@@ -10606,6 +10607,7 @@ export function issueService(db: Db) {
}
if (values.monitorNextCheckAt) await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, companyId);
const [issue] = await tx.insert(issues).values(values).returning();
await attachOwnedDraftImages(tx, issue, issue.description, { userId: issueData.createdByUserId, agentId: issueData.createdByAgentId });
if (issue.visibility === "private" && !inheritedPrivateSubtree && issueData.createdByAgentId) {
@@ -11486,6 +11488,7 @@ export function issueService(db: Db) {
(blockedByIssueIds !== undefined || issueData.unblockDescriptor !== undefined)))) {
patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number;
}
await notifyStatusCardIssueChange(tx, receiptExisting.companyId);
if ((patch.status !== undefined && patch.status !== receiptExisting.status)
|| patch.completedAt !== undefined || patch.cancelledAt !== undefined
|| issueData.parentId !== undefined || issueData.executionWorkspaceId !== undefined) {
@@ -11954,6 +11957,7 @@ export function issueService(db: Db) {
if (owner) {
await notifyExternalConversationSettlement(tx, { companyId: owner.companyId });
await notifyApprovalWork(tx, owner.companyId);
await notifyStatusCardIssueChange(tx, owner.companyId);
}
await notifyDecisionTargetIssueRemoval(tx, id);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
@@ -12726,6 +12730,7 @@ export function issueService(db: Db) {
.from(issueComments).where(eq(issueComments.id, commentId));
if (!source) return null;
await notifyExternalConversationSettlement(tx, source);
await notifyStatusCardIssueChange(tx, source.companyId);
const [comment] = await tx
.delete(issueComments)
.where(eq(issueComments.id, commentId))
@@ -12765,6 +12770,7 @@ export function issueService(db: Db) {
.for("update");
if (!source) return null;
await notifyExternalConversationSettlement(tx, source);
await notifyStatusCardIssueChange(tx, source.companyId);
await notifyChatInboundWakeNotices(tx, source);
const now = new Date();
const [comment] = await tx
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js";
@@ -334,6 +335,7 @@ export async function claimNativeSessionResumptions(input: {
nativePhaseUpdatedAt: now,
updatedAt: now,
}).where(eq(heartbeatRuns.id, row.run.id));
await notifyStatusCardIssueChange(tx, run.companyId);
await tx.update(issues).set({
executionRunId: run.id,
executionAgentNameKey: agent.name.trim().toLowerCase() || null,
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import {
agents,
heartbeatRuns,
@@ -240,6 +241,7 @@ export async function claimNativeReviewExecutionLock(db: Db, input: {
if (!run || !["queued", "running"].includes(run.status) || !admitted || !expected
|| admitted.nativeReviewInteractionId !== expected.nativeReviewInteractionId
|| admitted.nativeReviewDecisionId !== expected.nativeReviewDecisionId) return false;
await notifyStatusCardIssueChange(tx, input.companyId);
const claimed = await tx.update(issues).set({
executionRunId: input.runId,
executionAgentNameKey: input.agentNameKey,
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
import { alias } from "drizzle-orm/pg-core";
import { notifyUnsafeWorkspaceExportRecovery, type UnsafeWorkspaceExportScope } from "../unsafe-workspace-export-notifications.js";
@@ -107,6 +108,7 @@ async function recoverCandidate(db: Db, candidate: { runId: string; companyId: s
nativePhase: "result_accepted", nativePhaseUpdatedAt: now,
resultJson: { ...run.resultJson, finalizationPhase: "result_accepted", failureCode: null, originalFailureCode: null, nextAttemptAt: null }, updatedAt: now,
}).where(eq(heartbeatRuns.id, run.id)).returning();
await notifyStatusCardIssueChange(tx, run.companyId);
await tx.update(issues).set({ executionRunId: run.id, updatedAt: now }).where(eq(issues.id, issue.id));
await appendHeartbeatRunEvent(tx as unknown as Db, { companyId: run.companyId, agentId: run.agentId, runId: run.id,
eventType: "workspace_export_omitted", stream: "system", level: "info",
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js";
import { randomUUID } from "node:crypto";
@@ -141,6 +142,7 @@ export async function retryNativeWorkspaceExport(input: {
await tx.update(heartbeatRuns).set({ status: "running", finishedAt: null, error: null, errorCode: null,
nativePhase: "result_accepted", nativePhaseUpdatedAt: now,
resultJson: { ...current.run.resultJson, workspaceExportRetry: retry, finalizationPhase: "result_accepted", failureCode: null, originalFailureCode: null, nextAttemptAt: null }, updatedAt: now }).where(eq(heartbeatRuns.id, current.run.id));
await notifyStatusCardIssueChange(tx, current.run.companyId);
await tx.update(issues).set({ executionRunId: current.run.id, updatedAt: now }).where(eq(issues.id, current.issue.id));
// Reclaim the same lease only for control-plane copyback. The normal
// finalization settlement will stop/retain it again on success or failure.
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
import { and, eq, gt, inArray, ne, notInArray, or, sql } from "drizzle-orm";
@@ -187,6 +188,7 @@ export async function reconstructNativeWorkspace(input: {
}).where(eq(issueRecoveryActions.id, source.action.id));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, input.companyId);
await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null,
blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id));
const publication = await persistActivity(db, { companyId: input.companyId, actorType: "user", actorId: input.actorId,
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { requiredQualityReviewSchema } from "@paperclipai/shared";
import { priorRecoveryApiReceipt } from "../recovery-effect-receipts.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
@@ -2059,6 +2060,7 @@ export class PaperclipRunnerToolAuthority {
}
const patch = prepareIssueMonitorUpdate(target, input.monitor, actor);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyStatusCardIssueChange(tx, this.binding.companyId);
const [updated] = await tx.update(issues).set({ ...patch, updatedAt: new Date() })
.where(and(eq(issues.id, target.id), eq(issues.companyId, this.binding.companyId))).returning();
if (!updated) throw notFound("Task not found");
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { effectiveProjectWorkspacePolicy, isWorkspaceProvider } from "./project-workspace-provider.js";
import { releaseProviderRecoveryTransfers, stageProviderFreshWorkspace } from "./provider-fresh-workspace.js";
import { randomUUID, createHash } from "node:crypto";
@@ -418,6 +419,7 @@ export async function preparePluginWorkspace(
},
})
.returning();
await notifyStatusCardIssueChange(tx, task.companyId);
await tx
.update(issues)
.set({
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { assertTaskWorkspaceExecutionAccess } from "./task-workspace-access.js";
import { isDeepStrictEqual } from "node:util";
import { createHash, randomUUID } from "node:crypto";
@@ -152,6 +153,7 @@ export async function reserveFreshPluginTaskWorkspace(input: {
providerType: "adapter_managed", cwd: `/workspaces/${input.fresh.attemptId}`,
metadata: { runnerTaskFreshWorkspace: input.fresh, runnerTaskWorkspaceOwner: input.owner },
});
await notifyStatusCardIssueChange(tx, input.companyId);
await tx.update(issues).set({ executionWorkspaceId: input.fresh.attemptId, executionWorkspacePreference: "reuse_existing",
workspaceBindingRevision: task.workspaceBindingRevision + 1,
workspacePendingSelection: task.workspacePendingSelection?.expectedBindingRevision === task.workspaceBindingRevision
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifyApprovalWork } from "./approval-work-notifications.js";
import { notifyDecisionTargetIssueRemoval } from "./decision-targets.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
@@ -1949,6 +1950,7 @@ export function routineService(
if (createdIssue) {
await notifyApprovalWork(txDb, createdIssue.companyId);
await notifyDecisionTargetIssueRemoval(txDb, createdIssue.id);
await notifyStatusCardIssueChange(txDb, createdIssue.companyId);
await txDb.delete(issues).where(eq(issues.id, createdIssue.id));
}
const failureReason = error instanceof Error ? error.message : String(error);
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "./unsafe-workspace-export-notifications.js";
import { and, asc, desc, eq, gt, inArray, isNull, sql } from "drizzle-orm";
@@ -133,6 +134,7 @@ export async function settleExternalConversation(db: Db, companyId: string, issu
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId, issueId });
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await notifyStatusCardIssueChange(tx, companyId);
await tx.update(issues).set({ status: "in_review", updatedAt: new Date() })
.where(and(eq(issues.id, issueId), eq(issues.companyId, companyId)));
activity = (await persistActivity(tx as unknown as Db, {
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
import { notifyExternalConversationSettlement } from "./external-conversation-settlement-notifications.js";
import { chatConversations, issues, type Db } from "@paperclipai/db";
@@ -49,6 +50,7 @@ export async function resumeExternalConversation(tx: Db, companyId: string, issu
await notifyExternalConversationSettlement(tx, { companyId, issueId });
// Only the verified idle convention may leave In Review automatically.
// A real decision or delivery problem keeps the existing review semantics.
await notifyStatusCardIssueChange(tx, companyId);
await tx.update(issues).set({ status: "todo", updatedAt: new Date() })
.where(and(eq(issues.companyId, companyId), eq(issues.id, issueId),
sql`${externalConversationStateSql()} = 'waiting'`));
@@ -0,0 +1,90 @@
import { and, asc, eq, isNotNull, isNull } from "drizzle-orm";
import { companies, statusCards, type Db } from "@paperclipai/db";
import { logger } from "../middleware/logger.js";
import { registerDeadlineWork } from "./deadline-work.js";
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
import { instanceSettingsService } from "./instance-settings.js";
import { issueService } from "./issues.js";
import { queueIssueAssignmentWakeup, type IssueAssignmentWakeupDeps } from "./issue-assignment-wakeup.js";
import { statusCardService } from "./status-cards.js";
/** Feature-owned generation and assignment. Existing issue services retain
* execution authority, durable wake dedupe and stalled-generation cleanup. */
export function statusCardEvaluationService(db: Db, heartbeat: IssueAssignmentWakeupDeps) {
const cards = statusCardService(db);
const issues = issueService(db);
const settings = instanceSettingsService(db);
let reconstructReactive = true;
let allChanges = false;
const changes = new Set<string>();
let enabled = false;
return {
changed(hints?: ReadonlySet<string>) {
for (const hint of hints ?? ["*"]) {
if (hint === "*" || hint === "changes:*") allChanges = true;
else if (hint.startsWith("changes:")) changes.add(hint.slice("changes:".length));
}
},
async run(signal: AbortSignal) {
enabled = (await settings.getExperimental()).enableStatusCards === true;
if (!enabled || signal.aborted) return;
const reconstruct = reconstructReactive;
const evaluateAllChanges = allChanges;
const companiesChanged = new Set(changes);
changes.clear();
reconstructReactive = false;
allChanges = false;
try {
const result = await cards.tickDueStatusCards(new Date(), evaluateAllChanges ? "changed" : reconstruct && companiesChanged.size === 0 ? "all" : companiesChanged);
const assignments = await Promise.allSettled(result.enqueued.map(async ({ cardId, generatingIssue }) => {
// Once generated, dispatch or cancel even if shutdown begins: leaving
// an unassigned claimed card is not a completed sweep.
try {
await queueIssueAssignmentWakeup({
heartbeat, issue: generatingIssue,
reason: "status_card_update_assigned",
mutation: "status_card.scheduler_update_requested",
contextSource: "status_card_scheduler",
requestedByActorType: "system", taskKey: `status-card:${cardId}`,
rethrowOnError: true,
});
} catch (error) {
await issues.update(generatingIssue.id, { status: "cancelled" });
throw error;
}
}));
const failures = assignments.filter(result => result.status === "rejected");
if (failures.length) throw new AggregateError(failures.map(result => result.reason), "Status-card assignment failed");
if (result.evaluated || result.enqueued.length) {
logger.info({ evaluated: result.evaluated, enqueued: result.enqueued.length }, "status-card evaluation complete");
}
} catch (error) {
reconstructReactive ||= reconstruct;
allChanges ||= evaluateAllChanges;
for (const companyId of companiesChanged) changes.add(companyId);
throw error;
}
},
async next() {
if (!enabled) return null;
const [next] = await db.select({ at: statusCards.nextEvalAt }).from(statusCards)
.innerJoin(companies, eq(companies.id, statusCards.companyId))
.where(and(eq(companies.status, "active"), isNull(statusCards.archivedAt),
isNull(statusCards.generatingIssueId), isNotNull(statusCards.nextEvalAt)))
.orderBy(asc(statusCards.nextEvalAt)).limit(1);
return next?.at?.getTime() ?? null;
},
};
}
export function registerStatusCardEvaluationWork(
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
service: ReturnType<typeof statusCardEvaluationService>,
canRun: () => boolean,
) {
return registerDeadlineWork(coordinator, DELIVERY_QUEUES.statusCardEvaluation, {
run: service.run, next: service.next, onSettled: service.changed, canRun, retryMs: 30_000,
});
}
@@ -1,3 +1,4 @@
import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
import { and, eq, isNull } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { statusCards, statusCardUpdates } from "@paperclipai/db";
@@ -36,6 +37,7 @@ export async function finalizeStatusCardsForStalledGeneration(
) {
if (!STALLED_GENERATION_STATUSES.has(issue.status)) return [];
await notifyStatusCardEvaluation(dbOrTx, issue.companyId);
const now = new Date();
const failureReason = failureReasonForIssue(issue);
const cards = await dbOrTx
@@ -0,0 +1,10 @@
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
/** Card/settings changes reconstruct deadlines; issue changes also evaluate reactive scope. */
export function notifyStatusCardEvaluation(transaction: object, companyId?: string) {
return notifyDeliveryWork(transaction, DELIVERY_QUEUES.statusCardEvaluation, companyId ? `cards:${companyId}` : "*");
}
export function notifyStatusCardIssueChange(transaction: object, companyId?: string) {
return notifyDeliveryWork(transaction, DELIVERY_QUEUES.statusCardEvaluation, companyId ? `changes:${companyId}` : "changes:*");
}
@@ -128,10 +128,8 @@ export function isWithinStatusCardActiveHours(policy: StatusCardRefreshPolicy, n
}
export function nextStatusCardEvaluationAt(policy: StatusCardRefreshPolicy, now: Date) {
if (policy.mode === "manual") return null;
const seconds = policy.mode === "interval"
? (policy.intervalMinutes ?? 15) * 60
: Math.min(policy.debounceSeconds ?? 60, 60);
if (policy.mode !== "interval") return null;
const seconds = (policy.intervalMinutes ?? 15) * 60;
return new Date(now.getTime() + seconds * 1000);
}
@@ -172,3 +170,37 @@ export function evaluateStatusCardPolicy(input: {
}
return { action: "run" as const };
}
/** A retry exists only while filtered changes remain. Use the owning policy's
* actual eligibility boundary instead of repeatedly rediscovering a closed gate. */
export function nextStatusCardPolicyEligibility(input: {
policy: StatusCardRefreshPolicy;
now: Date;
lastChangeAt: Date;
completedUpdateTimes: Date[];
tokensToday: number;
}): Date | null {
if (input.policy.mode === "manual") return null;
let at = input.now.getTime();
if (input.tokensToday >= (input.policy.dailyTokenCap ?? 100_000)) {
const day = new Date(input.now);
day.setUTCHours(24, 0, 0, 0);
at = Math.max(at, day.getTime());
}
if (input.policy.mode === "reactive") {
at = Math.max(at, input.lastChangeAt.getTime() + (input.policy.debounceSeconds ?? 60) * 1000);
const cap = input.policy.maxUpdatesPerHour ?? 6;
const recent = input.completedUpdateTimes.map(time => time.getTime())
.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);
// 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);
}
// Equal start/end describes an empty window; retain changes, awaiting config.
return null;
}
+105 -54
View File
@@ -1,7 +1,9 @@
import { notifyStatusCardEvaluation } from "./status-card-notifications.js";
import { createHash } from "node:crypto";
import { and, desc, eq, gte, inArray, isNotNull, isNull, lte, ne, or, sql, type SQL } from "drizzle-orm";
import {
agents,
companies,
costEvents,
documentRevisions,
documents,
@@ -23,7 +25,7 @@ import { conflict, forbidden, notFound, unprocessable } from "../errors.js";
import { logger } from "../middleware/logger.js";
import { readBuiltInAgentMarker } from "../lib/built-in-agent-metadata.js";
import { builtInAgentService } from "./built-in-agents.js";
import { companySearchService } from "./company-search.js";
import { companySearchService, companySearchUpdatedWithinMs } from "./company-search.js";
import { issueService } from "./issues.js";
import { SUMMARIZER_BUILT_IN_KEY } from "./summary-slots.js";
import {
@@ -34,6 +36,7 @@ import {
extractIssueMentions,
filterStatusCardChanges,
nextStatusCardEvaluationAt,
nextStatusCardPolicyEligibility,
STATUS_CARD_MAX_MENTIONED_ISSUES,
statusCardChangesHash,
statusCardFingerprintHash,
@@ -44,6 +47,7 @@ import {
type StatusCardActor = { agentId: string | null; userId: string | null };
type StatusCardWriter = { agentId: string | null; runId: string | null };
type StatusCardRow = typeof statusCards.$inferSelect;
type WatchedIssue = CompanySearchIssueSummary & { scopeExpiresAt?: Date | null };
type StatusCardReadOptions = { issueReadCondition?: SQL<boolean> };
const TERMINAL_ISSUE_STATUSES = new Set(["done", "cancelled"]);
@@ -151,9 +155,15 @@ export function statusCardService(
db: Db,
deps: { issuesSvc?: ReturnType<typeof issueService> } = {},
) {
const builtIns = builtInAgentService(db);
const issuesSvc = deps.issuesSvc ?? issueService(db);
const issuesSvc = () => deps.issuesSvc ?? issueService(db);
const searchSvc = companySearchService(db);
type Transaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
function mutate<T>(companyId: string | undefined, write: (tx: Transaction) => Promise<T>) {
return db.transaction(async tx => {
await notifyStatusCardEvaluation(tx, companyId);
return write(tx);
});
}
async function readWatchedIssueVisibility(card: StatusCardRow, options: StatusCardReadOptions = {}) {
if (card.queries.length === 0 && (card.mentionedIssueIds?.length ?? 0) === 0) {
@@ -240,10 +250,10 @@ export function statusCardService(
};
const agentId = actor.agentId;
if (!agentId) {
return db.insert(statusCards).values(values).returning().then((rows) => rows[0]!);
return mutate(companyId, tx => tx.insert(statusCards).values(values).returning().then((rows) => rows[0]!));
}
return db.transaction(async (tx) => {
return mutate(companyId, async (tx) => {
const author = await tx
.select({ id: agents.id })
.from(agents)
@@ -292,7 +302,7 @@ export function statusCardService(
...(input.refreshPolicy !== undefined
? {
refreshPolicy: input.refreshPolicy,
nextEvalAt: card.archivedAt ? null : nextStatusCardEvaluationAt(input.refreshPolicy, now),
nextEvalAt: card.archivedAt ? null : input.refreshPolicy.mode === "reactive" && card.queries.length > 0 ? now : nextStatusCardEvaluationAt(input.refreshPolicy, now),
}
: {}),
...(archiveChanged && input.archived
@@ -309,21 +319,21 @@ export function statusCardService(
}
: {}),
};
const next = await db.update(statusCards).set({
const next = await mutate(card.companyId, tx => tx.update(statusCards).set({
...values,
...(archiveChanged && input.archived ? { generatingIssueId: null, pendingChangeHash: null } : {}),
}).where(eq(statusCards.id, card.id)).returning().then((rows) => rows[0]!);
}).where(eq(statusCards.id, card.id)).returning().then((rows) => rows[0]!));
if (archiveChanged && input.archived && card.generatingIssueId) {
const generationIssue = await db.select().from(issues).where(eq(issues.id, card.generatingIssueId)).then((rows) => rows[0] ?? null);
if (generationIssue && !TERMINAL_ISSUE_STATUSES.has(generationIssue.status)) {
await issuesSvc.update(generationIssue.id, { status: "cancelled" });
await issuesSvc().update(generationIssue.id, { status: "cancelled" });
}
}
return next;
}
async function remove(id: string) {
return db.delete(statusCards).where(eq(statusCards.id, id)).returning().then((rows) => rows[0] ?? null);
return mutate(undefined, tx => tx.delete(statusCards).where(eq(statusCards.id, id)).returning().then((rows) => rows[0] ?? null));
}
async function listUpdates(cardId: string) {
@@ -360,7 +370,7 @@ export function statusCardService(
.then((rows) => rows[0] ?? null);
if (override) return override.id;
}
const builtIn = await builtIns.get(card.companyId, SUMMARIZER_BUILT_IN_KEY);
const builtIn = await builtInAgentService(db).get(card.companyId, SUMMARIZER_BUILT_IN_KEY);
if (builtIn.status !== "ready" || !builtIn.agentId) {
throw unprocessable("Summarizer built-in agent is not configured", {
code: "summarizer_not_configured",
@@ -391,7 +401,7 @@ export function statusCardService(
let deduplicated = false;
const createdAt = new Date();
const created = await issuesSvc.create(card.companyId, {
const created = await issuesSvc().create(card.companyId, {
title: `Compile status card: ${card.title ?? card.interestPrompt.slice(0, 80)}`,
description: compileDescription(card, null, hash),
status: "todo",
@@ -409,16 +419,16 @@ export function statusCardService(
// covers idempotency-key hits that resolve to a terminal task (done/cancelled)
// as well as a `blocked` one that a manual re-kick is reviving.
const reopened = deduplicated && (TERMINAL_ISSUE_STATUSES.has(created.status) || created.status === "blocked")
? await issuesSvc.update(created.id, { status: "todo", assigneeAgentId: summarizerAgentId })
? await issuesSvc().update(created.id, { status: "todo", assigneeAgentId: summarizerAgentId })
: created;
const generationIssue = await issuesSvc.update(reopened!.id, {
const generationIssue = await issuesSvc().update(reopened!.id, {
description: compileDescription(card, reopened!.id, hash),
});
const [nextCard] = await db
const [nextCard] = await mutate(card.companyId, tx => tx
.update(statusCards)
.set({ generatingIssueId: generationIssue!.id, state: "compiling", failureReason: null, updatedAt: createdAt })
.where(eq(statusCards.id, card.id))
.returning();
.returning());
return {
card: nextCard!,
generatingIssue: generationIssue!,
@@ -464,7 +474,7 @@ export function statusCardService(
if (card.archivedAt) throw unprocessable("Archived status cards cannot accept generation writes");
await assertSummarizerWriter(card, input.generationIssueId, actor);
const now = new Date();
return db.transaction(async (tx) => {
return mutate(card.companyId, async (tx) => {
const current = await tx.select().from(statusCards).where(eq(statusCards.id, card.id)).then((rows) => rows[0] ?? null);
if (!current || current.archivedAt || current.generatingIssueId !== input.generationIssueId) {
throw conflict("Status-card compilation was superseded by a newer task");
@@ -581,14 +591,21 @@ export function statusCardService(
}
async function executeQueries(card: StatusCardRow, options: StatusCardReadOptions = {}) {
const issueMap = new Map<string, CompanySearchIssueSummary>();
const issueMap = new Map<string, WatchedIssue>();
for (const storedQuery of card.queries) {
const query = companySearchQuerySchema.parse(storedQuery);
const response = await searchSvc.search(card.companyId, query, {
issueReadCondition: options.issueReadCondition,
});
for (const result of response.results) {
if (result.type === "issue" && result.issue) issueMap.set(result.issue.id, result.issue);
if (result.type !== "issue" || !result.issue) continue;
const duration = companySearchUpdatedWithinMs(query.updatedWithin);
const expiry = duration === null ? null : new Date(new Date(result.issue.updatedAt).getTime() + duration + 1);
const previous = issueMap.get(result.issue.id);
issueMap.set(result.issue.id, { ...result.issue, scopeExpiresAt: previous
? previous.scopeExpiresAt == null || expiry === null ? null
: new Date(Math.max(previous.scopeExpiresAt.getTime(), expiry.getTime()))
: expiry });
}
}
// Issues mentioned in the latest summary join the watched set alongside the
@@ -598,7 +615,11 @@ export function statusCardService(
(card.mentionedIssueIds ?? []).filter((issueId) => !issueMap.has(issueId)),
options,
);
for (const issue of mentioned) issueMap.set(issue.id, issue);
for (const issue of mentioned) issueMap.set(issue.id, { ...issue, scopeExpiresAt: null });
for (const id of card.mentionedIssueIds ?? []) {
const issue = issueMap.get(id);
if (issue) issue.scopeExpiresAt = null;
}
const snapshot = [...issueMap.values()];
if (snapshot.length === 0) return snapshot;
const latestHumanComments = await db
@@ -622,6 +643,13 @@ export function statusCardService(
return snapshot.map((issue) => ({ ...issue, latestHumanCommentAt: commentByIssueId.get(issue.id) ?? null }));
}
function nextEvaluationAt(card: StatusCardRow, snapshot: WatchedIssue[], now: Date) {
const interval = nextStatusCardEvaluationAt(card.refreshPolicy, now);
if (card.refreshPolicy.mode !== "reactive" || !card.refreshPolicy.triggers.membershipChanges) return interval;
const expiries = snapshot.flatMap(issue => issue.scopeExpiresAt && issue.scopeExpiresAt > now ? [issue.scopeExpiresAt.getTime()] : []);
return expiries.length ? new Date(Math.min(...expiries)) : null;
}
async function requestRefresh(cardId: string, input: {
full?: boolean;
trigger?: "manual" | "interval" | "reactive" | "restore";
@@ -648,15 +676,15 @@ export function statusCardService(
const changes = filterStatusCardChanges(allChanges, card.refreshPolicy);
const trigger = input.trigger ?? "manual";
const forceRun = trigger === "manual" || trigger === "restore";
const nextEvalAt = nextStatusCardEvaluationAt(card.refreshPolicy, now);
const nextEvalAt = nextEvaluationAt(card, snapshot, now);
if (!forceRun && changes.length === 0) {
const [next] = await db.update(statusCards).set({
const [next] = await mutate(card.companyId, tx => tx.update(statusCards).set({
pendingChangeCount: 0,
pendingChangeHash: null,
lastChangeAt: null,
state: "active",
nextEvalAt,
}).where(eq(statusCards.id, card.id)).returning();
}).where(eq(statusCards.id, card.id)).returning());
return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false };
}
@@ -678,13 +706,17 @@ export function statusCardService(
manual: forceRun,
});
if (decision.action !== "run") {
const [next] = await db.update(statusCards).set({
const [next] = await mutate(card.companyId, tx => tx.update(statusCards).set({
pendingChangeCount: changes.length,
pendingChangeHash,
lastChangeAt,
state: decision.action === "pause_budget" ? "paused_budget" : decision.action === "pause_hours" ? "paused_hours" : "active",
nextEvalAt: decision.action === "wait" && "dueAt" in decision ? decision.dueAt : nextEvalAt,
}).where(eq(statusCards.id, card.id)).returning();
nextEvalAt: nextStatusCardPolicyEligibility({
policy: card.refreshPolicy, now, lastChangeAt,
completedUpdateTimes: recent.filter(row => row.kind !== "compile" && row.finishedAt).map(row => row.startedAt),
tokensToday: Number(daily[0]?.tokens ?? 0),
}),
}).where(eq(statusCards.id, card.id)).returning());
return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false };
}
@@ -707,7 +739,7 @@ export function statusCardService(
: null;
const fingerprintHash = statusCardFingerprintHash(fingerprint);
let deduplicated = false;
const created = await issuesSvc.create(card.companyId, {
const created = await issuesSvc().create(card.companyId, {
title: `${kind === "full" ? "Rebuild" : "Update"} status card: ${card.title ?? card.interestPrompt.slice(0, 80)}`,
description: updateDescription({ card, generationIssueId: null, fingerprint, changes, kind, trigger, previousSummary, snapshot }),
status: "todo",
@@ -720,15 +752,15 @@ export function statusCardService(
onDeduplicated: (reason) => { deduplicated = reason === "idempotency_key"; },
});
const reopened = deduplicated && TERMINAL_ISSUE_STATUSES.has(created.status)
? await issuesSvc.update(created.id, { status: "todo", assigneeAgentId: summarizerAgentId })
? await issuesSvc().update(created.id, { status: "todo", assigneeAgentId: summarizerAgentId })
: created;
const generationIssue = await issuesSvc.update(reopened!.id, {
const generationIssue = await issuesSvc().update(reopened!.id, {
description: updateDescription({ card, generationIssueId: reopened!.id, fingerprint, changes, kind, trigger, previousSummary, snapshot }),
});
const priorGenerationPredicate = card.generatingIssueId
? eq(statusCards.generatingIssueId, card.generatingIssueId)
: isNull(statusCards.generatingIssueId);
const [next] = await db.update(statusCards).set({
const [next] = await mutate(card.companyId, tx => tx.update(statusCards).set({
generatingIssueId: generationIssue!.id,
pendingChangeCount: changes.length,
pendingChangeHash,
@@ -736,17 +768,17 @@ export function statusCardService(
state: "active",
nextEvalAt,
failureReason: null,
}).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning();
}).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning());
if (!next) {
const winner = await getById(card.id);
if (!winner?.generatingIssueId) {
if (!TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) {
await issuesSvc.update(generationIssue!.id, { status: "cancelled" });
await issuesSvc().update(generationIssue!.id, { status: "cancelled" });
}
throw conflict("Status-card refresh claim was lost");
}
if (generationIssue!.id !== winner.generatingIssueId && !TERMINAL_ISSUE_STATUSES.has(generationIssue!.status)) {
await issuesSvc.update(generationIssue!.id, { status: "cancelled" });
await issuesSvc().update(generationIssue!.id, { status: "cancelled" });
}
const winnerIssue = await db.select().from(issues).where(eq(issues.id, winner.generatingIssueId)).then((rows) => rows[0] ?? null);
return { card: winner, generatingIssue: winnerIssue, alreadyGenerating: true, enqueued: false, kind, changes };
@@ -765,26 +797,44 @@ export function statusCardService(
return { card: next, generatingIssue: generationIssue!, alreadyGenerating: deduplicated, enqueued: true, kind, changes };
}
async function tickDueStatusCards(now = new Date()) {
const due = await db.select().from(statusCards).where(and(isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId), isNotNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)));
async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet<string> | "all" | "changed") {
const reactiveScope = reactiveCompanies === "all" ? isNull(statusCards.nextEvalAt) : reactiveCompanies === "changed" ? sql`true` : reactiveCompanies?.size
? inArray(statusCards.companyId, [...reactiveCompanies]) : sql`false`;
const due = await db.select({ card: statusCards }).from(statusCards)
.innerJoin(companies, eq(companies.id, statusCards.companyId))
.where(and(eq(companies.status, "active"), isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId),
or(and(isNotNull(statusCards.nextEvalAt), lte(statusCards.nextEvalAt, now)),
and(reactiveScope, ne(statusCards.state, "error"), ne(statusCards.state, "compiling"), sql`${statusCards.refreshPolicy}->>'mode' = 'reactive'`, sql`jsonb_array_length(${statusCards.queries}) > 0`))))
.then(rows => rows.map(row => row.card));
const enqueued: Array<{ cardId: string; generatingIssue: typeof issues.$inferSelect }> = [];
let evaluated = 0;
for (const candidate of due) {
const claimUntil = new Date(now.getTime() + 5 * 60 * 1000);
const [claimed] = await db.update(statusCards).set({ nextEvalAt: claimUntil })
.where(and(eq(statusCards.id, candidate.id), isNull(statusCards.generatingIssueId), lte(statusCards.nextEvalAt, now)))
.returning();
if (!claimed) continue;
evaluated += 1;
try {
const result = await requestRefresh(claimed.id, { trigger: claimed.refreshPolicy.mode === "reactive" ? "reactive" : "interval", now });
if (result.enqueued && result.generatingIssue) enqueued.push({ cardId: claimed.id, generatingIssue: result.generatingIssue });
} catch (err) {
logger.warn(
{ err, cardId: claimed.id, companyId: claimed.companyId },
"status card scheduled refresh failed",
// nextEvalAt is both a retry lease and an evaluation deadline. A source
// event may preempt a genuine future deadline, so serialize scheduler
// evaluation across connections without holding a row lock while the
// existing issue/card services commit through their own transactions.
await db.transaction(async lockTx => {
const [lock] = await lockTx.execute<{ acquired: boolean }>(
sql`select pg_try_advisory_xact_lock(hashtextextended(${candidate.id}, 0)) as acquired`,
);
}
if (!lock?.acquired) return;
const claimUntil = new Date(now.getTime() + 5 * 60 * 1000);
const [claimed] = await mutate(candidate.companyId, tx => tx.update(statusCards).set({ nextEvalAt: claimUntil })
.where(and(eq(statusCards.id, candidate.id), isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId),
candidate.nextEvalAt ? eq(statusCards.nextEvalAt, candidate.nextEvalAt) : isNull(statusCards.nextEvalAt)))
.returning());
if (!claimed) return;
evaluated += 1;
try {
const result = await requestRefresh(claimed.id, { trigger: claimed.refreshPolicy.mode === "reactive" ? "reactive" : "interval", now });
if (result.enqueued && result.generatingIssue) enqueued.push({ cardId: claimed.id, generatingIssue: result.generatingIssue });
} catch (err) {
logger.warn(
{ err, cardId: claimed.id, companyId: claimed.companyId },
"status card scheduled refresh failed",
);
}
});
}
return { evaluated, enqueued };
}
@@ -796,7 +846,7 @@ export function statusCardService(
await assertSummarizerWriter(card, input.generationIssueId, actor);
if (card.queries.length === 0) throw conflict("Compile the status-card query before writing its summary");
const now = new Date();
return db.transaction(async (tx) => {
return mutate(card.companyId, async (tx) => {
const current = await tx.select().from(statusCards).where(eq(statusCards.id, card.id)).then((rows) => rows[0] ?? null);
if (!current || current.archivedAt || current.generatingIssueId !== input.generationIssueId) {
throw conflict("Status-card generation was superseded by a newer task");
@@ -816,9 +866,8 @@ export function statusCardService(
const mentionedIssueIds = await resolveMentionedIssueIds(current.companyId, input.markdown);
// Current watched membership: compiled-query matches plus the issues this
// summary mentions.
const watchedNow = buildStatusCardFingerprint(
await executeQueries({ ...current, mentionedIssueIds }),
);
const watchedSnapshot = await executeQueries({ ...current, mentionedIssueIds });
const watchedNow = buildStatusCardFingerprint(watchedSnapshot);
let snapshot: StatusCardFingerprint;
if (payloadFingerprint) {
// Keep the generation-time fingerprint as the change baseline: issues
@@ -889,7 +938,9 @@ export function statusCardService(
pendingChangeCount: 0,
pendingChangeHash: null,
lastChangeAt: null,
nextEvalAt: nextStatusCardEvaluationAt(current.refreshPolicy, now),
nextEvalAt: current.refreshPolicy.mode === "reactive" &&
filterStatusCardChanges(diffStatusCardFingerprint(snapshot, watchedNow), current.refreshPolicy).length > 0
? now : nextEvaluationAt(current, watchedSnapshot, now),
updatedAt: now,
}).where(and(eq(statusCards.id, current.id), eq(statusCards.generatingIssueId, input.generationIssueId))).returning();
if (!next) throw conflict("Status-card generation was superseded by a newer task");