From 4af412eaea9e780e2d867bd930e77865a1c4672b Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 17:25:18 -0500 Subject: [PATCH] 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 --- doc/plans/2026-10-30-status-cards-sleep.md | 33 ++++ .../status-card-evaluation-work.test.ts | 180 ++++++++++++++++++ .../status-card-update-engine.test.ts | 23 +++ server/src/__tests__/status-cards.test.ts | 9 +- server/src/index.ts | 35 +--- .../active-run-watchdog/adapters/postgres.ts | 2 + server/src/services/access.ts | 2 + server/src/services/agent-deletion.ts | 2 + server/src/services/companies.ts | 3 + server/src/services/company-deletion.ts | 2 + server/src/services/company-search.ts | 10 +- server/src/services/company-skills.ts | 2 + .../services/delivery-work-notifications.ts | 1 + .../execution-control-notifications.ts | 3 + server/src/services/execution-workspaces.ts | 4 + server/src/services/heartbeat/retries.ts | 2 + server/src/services/instance-settings.ts | 2 + server/src/services/issue-comment-writes.ts | 2 + server/src/services/issue-prefix.ts | 2 + .../issue-review-state-notifications.ts | 2 + server/src/services/issue-title.ts | 2 + server/src/services/issue-tree-control.ts | 2 + server/src/services/issues.ts | 6 + .../native-finalization-reconciler.ts | 2 + .../native-review-participant.ts | 2 + .../native-workspace-export-recovery.ts | 2 + .../native-workspace-export-retry.ts | 2 + .../native-workspace-reconstruction.ts | 2 + .../paperclip-runner-tool-authority.ts | 2 + server/src/services/plugin-workspaces.ts | 2 + .../src/services/provider-fresh-workspace.ts | 2 + server/src/services/routines.ts | 2 + .../services/slack-conversation-lifecycle.ts | 2 + .../src/services/slack-conversation-state.ts | 2 + server/src/services/status-card-evaluation.ts | 90 +++++++++ .../src/services/status-card-finalization.ts | 2 + .../src/services/status-card-notifications.ts | 10 + .../src/services/status-card-update-engine.ts | 40 +++- server/src/services/status-cards.ts | 159 ++++++++++------ 39 files changed, 561 insertions(+), 93 deletions(-) create mode 100644 doc/plans/2026-10-30-status-cards-sleep.md create mode 100644 server/src/__tests__/status-card-evaluation-work.test.ts create mode 100644 server/src/services/status-card-evaluation.ts create mode 100644 server/src/services/status-card-notifications.ts diff --git a/doc/plans/2026-10-30-status-cards-sleep.md b/doc/plans/2026-10-30-status-cards-sleep.md new file mode 100644 index 00000000000..dcd6c5ca226 --- /dev/null +++ b/doc/plans/2026-10-30-status-cards-sleep.md @@ -0,0 +1,33 @@ +# Status-card evaluation deadlines + +Baseline: `76fde891ccc29283b3f5c8dcdce089c5b3d937d1`. + +Status-card ownership covers evaluation policy, card claims, generation, ordinary assignment wakeups, and failed-assignment cancellation. `index.ts` registers a typed capability. Shared delivery/deadline infrastructure owns transaction settlement, coalescing, admission/drain, retries and shutdown. Existing cards, updates and issues remain durable truth; no timer table or sleep-safety exemption is introduced. + +Manual cards have no automatic obligation. Interval cards retain their requested evaluation cadence. Reactive cards evaluate committed source changes, retain the actual debounce/rate/budget/active-hours deadline while relevant changes remain, and disarm once unchanged. Saved relative-search windows retain their actual membership expiry; the duration interpretation stays with company search. Startup reconstructs existing work without advancing business timestamps. + +Scheduled reactive source events can preempt a future evaluation. A feature-owned transaction advisory lock serializes scheduler evaluation across connections while existing card/issue services perform their own committed writes. The lock holder has no card row lock, avoiding self-deadlock. Existing next-evaluation CAS, generating-issue ownership and issue idempotency keys remain required, including for manual refreshes and older server instances that do not use the advisory lock. The persisted five-minute claim remains the failure retry after a scheduled refresh cannot finish. Internal timers do not establish a Cloud wake alarm. + +## Producer audit (in progress) + +| Source | Owning committed boundary | State | +| --- | --- | --- | +| Card create/config/archive/delete, compile, query/summary writeback, evaluation | status-card mutation transactions | wired; ordinary lifecycle exercised | +| Experimental feature flag | instance settings update transaction | wired; disable exercised | +| Company pause/archive/reactivate/delete | company owning transactions | wired; lifecycle acceptance remains | +| Ordinary issue create/update/remove | issue service transactions | wired; create/change exercised | +| Comments create/delete/tombstone | common insert and issue mutation transactions | wired; comment acceptance remains | +| Review state / chat-run milestones | issue-review-state notification boundary | wired; exact caller coverage audit remains | +| Execution-control writes/run settlement | execution-control transaction helpers | wired; exact caller coverage audit remains | +| Agent deletion/membership reassignment, issue tree transitions, title/prefix | owner transaction | wired; representative acceptance remains | +| Workspace selection/binding, plugin/provider workspace, native recovery/reclaim, active watchdog, retry | owner transactions where uncovered by shared notifications | partially wired; remaining per-write audit required | +| Raw workspace cleanup/routes, conversation activation/settlement, queue paths | direct writes and adjacent owner notifications | exact coverage incomplete; not acceptance-ready | +| Imports/independent writers/replicas | supported service writes and startup reconstruction | full ingress contract audit remains | + +Process-local notifications cover writes using the installed database owner. They do not notify an unrelated SQL writer or another server process; those supported paths need explicit ingress or reconstruction coverage. No claim of cross-replica delivery, host suspension, or fleet savings is made by the current tests. + +## Acceptance evidence and remaining gates + +The original status-card PostgreSQL suite passed25/25. New real PostgreSQL tests exercise ordinary card compilation/writeback, source-change debounce, persisted assignment at a controlled provider boundary, result persistence, quiet SQL after completion, earlier/cancelled deadlines, nested outer commit/rollback and drain/resume. The provider boundary is simulated; no model runs. Additional race, restart, failure, active-hour/rate/budget persistence, multi-company and producer-class coverage remain before PR delivery. + +Shared deadline regressions reproduced and corrected a settled-write admission hold during drain. The correction releases only settled holds, retains an in-memory resume wake, and leaves ambiguous transaction intent fenced by the database coordinator. Unit coverage includes first write during drain, failure entering drain, preserved future deadlines, warm standby and ambiguous commit settlement. diff --git a/server/src/__tests__/status-card-evaluation-work.test.ts b/server/src/__tests__/status-card-evaluation-work.test.ts new file mode 100644 index 00000000000..427ad3385eb --- /dev/null +++ b/server/src/__tests__/status-card-evaluation-work.test.ts @@ -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> | undefined; + let db: ReturnType; + let coordinator: ReturnType | 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 | 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([]); + }); +}); diff --git a/server/src/__tests__/status-card-update-engine.test.ts b/server/src/__tests__/status-card-update-engine.test.ts index 09a660e8741..b8309415770 100644 --- a/server/src/__tests__/status-card-update-engine.test.ts +++ b/server/src/__tests__/status-card-update-engine.test.ts @@ -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"); diff --git a/server/src/__tests__/status-cards.test.ts b/server/src/__tests__/status-cards.test.ts index e1436666316..0c24b05dd70 100644 --- a/server/src/__tests__/status-cards.test.ts +++ b/server/src/__tests__/status-cards.test.ts @@ -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; @@ -79,8 +81,8 @@ describeEmbeddedPostgres("status card routes", () => { let tempDb: Awaited> | 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(); }); diff --git a/server/src/index.ts b/server/src/index.ts index ae229f03518..6fb78f1612d 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -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) { diff --git a/server/src/modules/active-run-watchdog/adapters/postgres.ts b/server/src/modules/active-run-watchdog/adapters/postgres.ts index 9531c449260..002415c03a6 100644 --- a/server/src/modules/active-run-watchdog/adapters/postgres.ts +++ b/server/src/modules/active-run-watchdog/adapters/postgres.ts @@ -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({ diff --git a/server/src/services/access.ts b/server/src/services/access.ts index aa0f9dcc828..477c0434c50 100644 --- a/server/src/services/access.ts +++ b/server/src/services/access.ts @@ -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({ diff --git a/server/src/services/agent-deletion.ts b/server/src/services/agent-deletion.ts index 3650c40f1a6..5907d222e1d 100644 --- a/server/src/services/agent-deletion.ts +++ b/server/src/services/agent-deletion.ts @@ -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) diff --git a/server/src/services/companies.ts b/server/src/services/companies.ts index 0f2ed547a79..8efd6a269fb 100644 --- a/server/src/services/companies.ts +++ b/server/src/services/companies.ts @@ -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); diff --git a/server/src/services/company-deletion.ts b/server/src/services/company-deletion.ts index 7ab88aadfa9..0379638968b 100644 --- a/server/src/services/company-deletion.ts +++ b/server/src/services/company-deletion.ts @@ -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); diff --git a/server/src/services/company-search.ts b/server/src/services/company-search.ts index 220967e362d..bcf2a550d3f 100644 --- a/server/src/services/company-search.ts +++ b/server/src/services/company-search.ts @@ -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) { diff --git a/server/src/services/company-skills.ts b/server/src/services/company-skills.ts index 7ba950039e9..faf54c410a9 100644 --- a/server/src/services/company-skills.ts +++ b/server/src/services/company-skills.ts @@ -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 }) diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9195f1fcb12..fdc4aaadc20 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -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", diff --git a/server/src/services/execution-control-notifications.ts b/server/src/services/execution-control-notifications.ts index 22c8913bbcd..c7b26a8bd72 100644 --- a/server/src/services/execution-control-notifications.ts +++ b/server/src/services/execution-control-notifications.ts @@ -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( write: (tx: Db) => PromiseLike, ): Promise { 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); diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index 4b716916638..caaae186ad8 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -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(); diff --git a/server/src/services/heartbeat/retries.ts b/server/src/services/heartbeat/retries.ts index 946f5e309f8..de03aebeca0 100644 --- a/server/src/services/heartbeat/retries.ts +++ b/server/src/services/heartbeat/retries.ts @@ -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({ diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 0208653a86f..35eddc10660 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -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); diff --git a/server/src/services/issue-comment-writes.ts b/server/src/services/issue-comment-writes.ts index e24e78e9855..4d4cf374e5b 100644 --- a/server/src/services/issue-comment-writes.ts +++ b/server/src/services/issue-comment-writes.ts @@ -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[0]>[0]; export function insertIssueComment(db: WriteDb, input: PgInsertValue & { 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(); }); diff --git a/server/src/services/issue-prefix.ts b/server/src/services/issue-prefix.ts index 5e31f9bd7ff..649a4aaae9b 100644 --- a/server/src/services/issue-prefix.ts +++ b/server/src/services/issue-prefix.ts @@ -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})` }) diff --git a/server/src/services/issue-review-state-notifications.ts b/server/src/services/issue-review-state-notifications.ts index 679b7e37475..d66ca96eb0e 100644 --- a/server/src/services/issue-review-state-notifications.ts +++ b/server/src/services/issue-review-state-notifications.ts @@ -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); diff --git a/server/src/services/issue-title.ts b/server/src/services/issue-title.ts index af6716e3aa4..51e360269d7 100644 --- a/server/src/services/issue-title.ts +++ b/server/src/services/issue-title.ts @@ -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, { diff --git a/server/src/services/issue-tree-control.ts b/server/src/services/issue-tree-control.ts index 0d7633f846a..a8009559827 100644 --- a/server/src/services/issue-tree-control.ts +++ b/server/src/services/issue-tree-control.ts @@ -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({ diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 42c65bb5d02..8787ac867a8 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -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 diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index af828edfb98..c5cf832982e 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -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, diff --git a/server/src/services/native-runtime/native-review-participant.ts b/server/src/services/native-runtime/native-review-participant.ts index 573574dd19d..6391d506d5d 100644 --- a/server/src/services/native-runtime/native-review-participant.ts +++ b/server/src/services/native-runtime/native-review-participant.ts @@ -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, diff --git a/server/src/services/native-runtime/native-workspace-export-recovery.ts b/server/src/services/native-runtime/native-workspace-export-recovery.ts index 41e51f2ddc7..0c3aac34799 100644 --- a/server/src/services/native-runtime/native-workspace-export-recovery.ts +++ b/server/src/services/native-runtime/native-workspace-export-recovery.ts @@ -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", diff --git a/server/src/services/native-runtime/native-workspace-export-retry.ts b/server/src/services/native-runtime/native-workspace-export-retry.ts index 47de8c73cf4..5a79c9e782e 100644 --- a/server/src/services/native-runtime/native-workspace-export-retry.ts +++ b/server/src/services/native-runtime/native-workspace-export-retry.ts @@ -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. diff --git a/server/src/services/native-runtime/native-workspace-reconstruction.ts b/server/src/services/native-runtime/native-workspace-reconstruction.ts index 5273b30216a..33e37704389 100644 --- a/server/src/services/native-runtime/native-workspace-reconstruction.ts +++ b/server/src/services/native-runtime/native-workspace-reconstruction.ts @@ -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, diff --git a/server/src/services/native-runtime/paperclip-runner-tool-authority.ts b/server/src/services/native-runtime/paperclip-runner-tool-authority.ts index aa8970db19f..cf07e70eb7a 100644 --- a/server/src/services/native-runtime/paperclip-runner-tool-authority.ts +++ b/server/src/services/native-runtime/paperclip-runner-tool-authority.ts @@ -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"); diff --git a/server/src/services/plugin-workspaces.ts b/server/src/services/plugin-workspaces.ts index e2383dcfc74..ca04b081e48 100644 --- a/server/src/services/plugin-workspaces.ts +++ b/server/src/services/plugin-workspaces.ts @@ -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({ diff --git a/server/src/services/provider-fresh-workspace.ts b/server/src/services/provider-fresh-workspace.ts index fcc184e15f4..98a478a375d 100644 --- a/server/src/services/provider-fresh-workspace.ts +++ b/server/src/services/provider-fresh-workspace.ts @@ -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 diff --git a/server/src/services/routines.ts b/server/src/services/routines.ts index 23787b45249..a831a3e489a 100644 --- a/server/src/services/routines.ts +++ b/server/src/services/routines.ts @@ -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); diff --git a/server/src/services/slack-conversation-lifecycle.ts b/server/src/services/slack-conversation-lifecycle.ts index 0fd213dc084..9e784ff005d 100644 --- a/server/src/services/slack-conversation-lifecycle.ts +++ b/server/src/services/slack-conversation-lifecycle.ts @@ -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, { diff --git a/server/src/services/slack-conversation-state.ts b/server/src/services/slack-conversation-state.ts index dedaa8d6676..08f6c9c038b 100644 --- a/server/src/services/slack-conversation-state.ts +++ b/server/src/services/slack-conversation-state.ts @@ -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'`)); diff --git a/server/src/services/status-card-evaluation.ts b/server/src/services/status-card-evaluation.ts new file mode 100644 index 00000000000..143df32344a --- /dev/null +++ b/server/src/services/status-card-evaluation.ts @@ -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(); + let enabled = false; + + return { + changed(hints?: ReadonlySet) { + 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, + service: ReturnType, + canRun: () => boolean, +) { + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.statusCardEvaluation, { + run: service.run, next: service.next, onSettled: service.changed, canRun, retryMs: 30_000, + }); +} diff --git a/server/src/services/status-card-finalization.ts b/server/src/services/status-card-finalization.ts index 7c7d260992e..5d7ee62b523 100644 --- a/server/src/services/status-card-finalization.ts +++ b/server/src/services/status-card-finalization.ts @@ -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 diff --git a/server/src/services/status-card-notifications.ts b/server/src/services/status-card-notifications.ts new file mode 100644 index 00000000000..6fc12985847 --- /dev/null +++ b/server/src/services/status-card-notifications.ts @@ -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:*"); +} diff --git a/server/src/services/status-card-update-engine.ts b/server/src/services/status-card-update-engine.ts index ff14d35c640..1f734227808 100644 --- a/server/src/services/status-card-update-engine.ts +++ b/server/src/services/status-card-update-engine.ts @@ -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; +} diff --git a/server/src/services/status-cards.ts b/server/src/services/status-cards.ts index 4a7439b6aa5..84bbf7b2cd4 100644 --- a/server/src/services/status-cards.ts +++ b/server/src/services/status-cards.ts @@ -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 }; const TERMINAL_ISSUE_STATUSES = new Set(["done", "cancelled"]); @@ -151,9 +155,15 @@ export function statusCardService( db: Db, deps: { issuesSvc?: ReturnType } = {}, ) { - const builtIns = builtInAgentService(db); - const issuesSvc = deps.issuesSvc ?? issueService(db); + const issuesSvc = () => deps.issuesSvc ?? issueService(db); const searchSvc = companySearchService(db); + type Transaction = Parameters[0]>[0]; + function mutate(companyId: string | undefined, write: (tx: Transaction) => Promise) { + 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(); + const issueMap = new Map(); 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 | "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");