Jannik/PaperClipAI
1
0 Fork 0
mirror of https://github.com/paperclipai/paperclip.git synced 2026-11-04 11:01:16 +01:00
Code Issues Packages Projects Releases Wiki Activity

fix: preserve scoped status-card work across mutations and recovery

Browse Source 
Complete transactional source coverage, cached feature gating, startup scope reconstruction, off-page relative query expiry and busy-lock retry. Verify producer, persistence, failure, budget resume, compatibility and startup paths with isolated PostgreSQL and focused tests.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-30 17:50:38 -05:00
1 parent 938e22aa21
commit a12f68ef13
24 files changed
+524 -125
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/__tests__/company-search-service.test.ts
+10 -4
View File
@@ -1,3 +1,4 @@
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
import { randomUUID } from "node:crypto";
import { sql } from "drizzle-orm";
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
@@ -28,12 +29,13 @@ import {
import { issueReadSqlCondition, projectReadSqlCondition } from "../services/authorization.js";
import { issueService } from "../services/issues.js";
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_COMPANY_SEARCH_TEST_DATABASE_URL");
const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
if (!embeddedPostgresSupport.supported) {
console.warn(
`Skipping embedded Postgres company search tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
`Skipping embedded Postgres company search tests on this host: ${("reason" in embeddedPostgresSupport ? embeddedPostgresSupport.reason : "unsupported environment")}`,
);
}
@@ -80,8 +82,11 @@ describeEmbeddedPostgres("companySearchService", () => {
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
beforeAll(async () => {
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-company-search-");
db = createDb(tempDb.connectionString);
if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-company-search-");
db = createDb(externalDatabaseUrl ?? tempDb!.connectionString, { maxConnections: 6 });
// External suites can reuse this dedicated database sequentially. Match
// the fresh embedded fixture before this suite creates its own rows.
if (externalDatabaseUrl) await db.execute(sql`TRUNCATE TABLE companies CASCADE`);
svc = companySearchService(db);
await db.execute(sql.raw("CREATE EXTENSION IF NOT EXISTS pg_trgm"));
}, 20_000);
@@ -100,6 +105,7 @@ describeEmbeddedPostgres("companySearchService", () => {
});
afterAll(async () => {
await db?.$client.end({ timeout: 1 });
await tempDb?.cleanup();
});
server/src/__tests__/server-startup-feedback-export.test.ts
+30
View File
@@ -1,3 +1,4 @@
import { statusCardEvaluationService, registerStatusCardEvaluationWork } from "../services/status-card-evaluation.js";
import { nativeHarnessBackupExpirationService } from "../services/native-runtime/native-harness-backup-expiration.js";
import { skillVersionRetentionService } from "../services/skill-version-retention.js";
import { approvalQueueSeedingService } from "../services/approval-queue-seeding-work.js";
@@ -527,6 +528,13 @@ vi.mock("../services/native-runtime/native-question-bridge.js", () => ({
validateNativeQuestionResponseInput: vi.fn(),
}));
vi.mock("../services/status-card-evaluation.js", () => ({
statusCardEvaluationService: vi.fn(() => ({ run: vi.fn(async () => {}), next: vi.fn(async () => null), changed: vi.fn() })),
registerStatusCardEvaluationWork: vi.fn((_coordinator, service, canRun) => ({
ready: canRun() ? service.run(new AbortController().signal) : Promise.resolve(), wake: vi.fn(), stop: vi.fn(async () => {}),
})),
}));
vi.mock("../services/secret-proposals.js", () => ({
createSecretProposalsService: vi.fn(() => ({
expiryDeadlines: vi.fn(async () => []),
@@ -790,6 +798,28 @@ describe("startServer feedback export wiring", () => {
expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.skillVersionRetention)).toEqual([]);
});
it.each([false, true])("registers status-card evaluation only with maintenance and never ticks it from the heartbeat loop (%s)", async enabled => {
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled, heartbeatSchedulerIntervalMs: 1000 }));
let tick: (() => void) | undefined;
const timer = vi.spyOn(globalThis, "setInterval").mockImplementation(((callback: () => void, interval: number) => {
if (interval === 1000) tick = callback;
return 1 as unknown as ReturnType<typeof setInterval>;
}) as typeof setInterval);
try {
await startServer(); await settleStartupHousekeeping();
expect(registerStatusCardEvaluationWork).toHaveBeenCalledTimes(enabled ? 1 : 0);
expect(statusCardEvaluationService).toHaveBeenCalledTimes(enabled ? 1 : 0);
if (enabled) {
const service = vi.mocked(statusCardEvaluationService).mock.results.at(-1)!.value;
expect(service.run).toHaveBeenCalledOnce();
expect(tick).toBeDefined();
for (let i = 0; i < 3; i++) { tick!(); await settleStartupHousekeeping(); }
expect(service.run).toHaveBeenCalledOnce();
}
expect(deliveryWorkErrors.filter(error => error.queue === DELIVERY_QUEUES.statusCardEvaluation)).toEqual([]);
} finally { timer.mockRestore(); }
});
it.each([false, true])("registers response recovery and its shutdown subscription only when maintenance is enabled (%s)", async enabled => {
loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled }));
await startServer(); await settleStartupHousekeeping();
server/src/__tests__/status-card-evaluation-work.test.ts
+203 -6
View File
@@ -1,7 +1,9 @@
import { budgetService } from "../services/budgets.js";
import { documentService } from "../services/documents.js";
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,
import { activityLog, agentWakeupRequests, agents, approvals, budgetIncidents, budgetPolicies, costEvents, 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";
@@ -22,6 +24,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
let draining = false;
let companyId: string, agentId: string;
const errors: unknown[] = [];
let compiledQuery = companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"] });
const actor = { agentId: null, userId: "board-user" };
const policy = (mode: "manual" | "reactive" | "interval" = "reactive") => statusCardRefreshPolicySchema.parse({
mode, debounceSeconds: 60, intervalMinutes: 15,
@@ -29,7 +32,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
// 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",
const [agent] = await db.select({ companyId: agents.companyId }).from(agents).where(eq(agents.id, id));
return db.insert(agentWakeupRequests).values({ companyId: agent!.companyId, agentId: id, source: "assignment",
reason: options.reason, payload: options.payload, status: "queued" }).returning();
});
@@ -38,7 +42,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
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();
companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0; wakeup.mockReset();
compiledQuery = companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"] });
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 });
@@ -48,6 +53,7 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
afterEach(async () => {
await coordinator?.stop(); coordinator = undefined;
vi.useRealTimers(); vi.restoreAllMocks();
await db.delete(budgetIncidents); await db.delete(budgetPolicies); await db.delete(approvals); await db.delete(costEvents);
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);
@@ -80,14 +86,14 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
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"] })],
if (compile) await cards.writeQuery(cardId, { queries: [compiledQuery],
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") {
async function card(mode: "manual" | "reactive" | "interval" = "reactive", refreshPolicy = policy(mode)) {
const cards = statusCardService(db);
const created = await cards.create(companyId, { interestPrompt: "Blockers", titlePinned: false, agentId, refreshPolicy: policy(mode) }, actor);
const created = await cards.create(companyId, { interestPrompt: "Blockers", titlePinned: false, agentId, refreshPolicy }, 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!);
@@ -131,6 +137,19 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
expect(wakeup).not.toHaveBeenCalled(); expect(errors).toEqual([]);
});
it("restores overdue debt after the feature is disabled across its saved deadline", async () => {
await start(); const created = await card(); await quiet();
await issueService(db).create(companyId, { title: "Disabled blocker", status: "blocked", priority: "medium" });
await vi.waitFor(async () => expect((await statusCardService(db).getById(created.id))?.pendingChangeCount).toBeGreaterThan(0));
const pending = (await statusCardService(db).getById(created.id))!;
await instanceSettingsService(db).updateExperimental({ enableStatusCards: false }); await quiet();
expect(wakeup).not.toHaveBeenCalled();
expect((await statusCardService(db).getById(created.id))!.nextEvalAt).toEqual(pending.nextEvalAt);
await instanceSettingsService(db).updateExperimental({ enableStatusCards: true });
await vi.waitFor(() => expect(wakeup).toHaveBeenCalledOnce());
await writeback(created.id, (await statusCardService(db).getById(created.id))!.generatingIssueId!); await quiet();
});
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 => {
@@ -169,6 +188,184 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
.toHaveLength(1);
});
it.each(["debounce", "hours", "rate", "budget"] as const)("restores pending %s eligibility after restart without advancing it or acting early", async gate => {
vi.setSystemTime(new Date("2026-11-02T18:00:00Z"));
const refreshPolicy = { ...policy(), ...(gate === "hours" ? { activeHours: { start: "09:00", end: "17:00", timezone: "UTC" } } : {}),
...(gate === "rate" ? { maxUpdatesPerHour: 1 } : {}), ...(gate === "budget" ? { dailyTokenCap: 100 } : {}) };
const created = await card("reactive", refreshPolicy);
if (gate === "rate" || gate === "budget") await db.insert(statusCardUpdates).values({ cardId: created.id, kind: "full", trigger: "reactive",
status: "ok", startedAt: new Date(Date.now() - (gate === "rate" ? 30 * 60_000 : 60_000)), finishedAt: new Date(), inputTokens: gate === "budget" ? 100 : 0 });
await issueService(db).create(companyId, { title: "Saved blocker", status: "blocked", priority: "medium", createdByUserId: "board-user" });
const service = statusCardEvaluationService(db, { wakeup });
await service.run(new AbortController().signal);
const pending = (await statusCardService(db).getById(created.id))!;
expect(pending.pendingChangeCount).toBeGreaterThan(0); expect(pending.nextEvalAt!.getTime()).toBeGreaterThan(Date.now());
const restarted = statusCardEvaluationService(db, { wakeup });
await restarted.run(new AbortController().signal);
expect(await restarted.next()).toBe(pending.nextEvalAt!.getTime());
expect((await statusCardService(db).getById(created.id))!.nextEvalAt).toEqual(pending.nextEvalAt);
vi.setSystemTime(pending.nextEvalAt!.getTime() - 1); await restarted.run(new AbortController().signal);
expect(wakeup).not.toHaveBeenCalled();
vi.setSystemTime(pending.nextEvalAt!); await restarted.run(new AbortController().signal);
expect(wakeup).toHaveBeenCalledOnce();
});
it("unions startup reconstruction with a scoped source change in another company", async () => {
const firstCompany = companyId;
const first = await card();
await issueService(db).create(companyId, { title: "First blocker", status: "blocked", priority: "medium" });
companyId = randomUUID(); agentId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Other", issuePrefix: `X${companyId.slice(0, 6)}` });
await db.insert(agents).values({ id: agentId, companyId, name: "Other summarizer", role: "general", status: "idle", adapterType: "process", adapterConfig: {} });
const second = await card();
await issueService(db).create(companyId, { title: "Second blocker", status: "blocked", priority: "medium" });
const service = statusCardEvaluationService(db, { wakeup });
service.changed(new Set([`changes:${companyId}`]));
await service.run(new AbortController().signal);
const cards = statusCardService(db);
expect((await cards.getById(first.id))?.companyId).toBe(firstCompany);
expect((await cards.getById(first.id))?.pendingChangeCount).toBeGreaterThan(0);
expect((await cards.getById(second.id))?.pendingChangeCount).toBeGreaterThan(0);
await vi.advanceTimersByTimeAsync(60_000); await service.run(new AbortController().signal);
expect(wakeup).toHaveBeenCalledTimes(2);
expect(new Set(await db.select().from(agentWakeupRequests).then(rows => rows.map(row => row.companyId))))
.toEqual(new Set([firstCompany, companyId]));
});
it("keeps ignored issue activity quiet and reacts to human comments and document search membership", async () => {
await start(); await card(); await quiet();
await issueService(db).create(companyId, { title: "Ignored todo", status: "todo", priority: "medium" });
await quiet(); expect(wakeup).not.toHaveBeenCalled();
compiledQuery = companySearchQuerySchema.parse({ scope: "all", q: "launch" });
const searchCard = await card(); await quiet();
const target = await issueService(db).create(companyId, { title: "Ordinary task", status: "todo", priority: "medium" });
await quiet();
await documentService(db).upsertIssueDocument({ issueId: target.id, key: "plan", format: "markdown", body: "launch evidence", createdByUserId: "board-user" });
await vi.waitFor(async () => expect((await statusCardService(db).getById(searchCard.id))?.pendingChangeCount).toBeGreaterThan(0));
await vi.advanceTimersByTimeAsync(60_000); await settled(); expect(wakeup).toHaveBeenCalledOnce();
await writeback(searchCard.id, (await statusCardService(db).getById(searchCard.id))!.generatingIssueId!); await quiet();
await issueService(db).addComment(target.id, "A human launch update", { userId: "board-user" });
await vi.waitFor(async () => expect((await statusCardService(db).getById(searchCard.id))?.pendingChangeCount).toBeGreaterThan(0));
});
it("retains relative query membership expiry across restart and emits the eventual removal", async () => {
const now = new Date();
const target = await issueService(db).create(companyId, { title: "Expiring blocker", status: "blocked", priority: "medium" });
await db.update(issues).set({ updatedAt: now }).where(eq(issues.id, target.id));
compiledQuery = companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"], updatedWithin: "1h" });
const created = await card();
const expires = now.getTime() + 3_600_001;
expect(created.nextEvalAt?.getTime()).toBe(expires);
const restarted = statusCardEvaluationService(db, { wakeup });
await restarted.run(new AbortController().signal);
expect(await restarted.next()).toBe(expires);
vi.setSystemTime(expires - 1); await restarted.run(new AbortController().signal); expect(wakeup).not.toHaveBeenCalled();
vi.setSystemTime(expires); await restarted.run(new AbortController().signal);
const pending = (await statusCardService(db).getById(created.id))!;
expect(pending.pendingChangeCount).toBe(1); expect(pending.nextEvalAt!.getTime()).toBe(expires + 60_000);
vi.setSystemTime(pending.nextEvalAt!); await restarted.run(new AbortController().signal);
expect(wakeup).toHaveBeenCalledOnce();
});
it("uses expiry of an off-page candidate to refresh offset-ranked query membership", async () => {
const now = new Date();
const tasks = await Promise.all((["critical", "high", "medium"] as const).map(priority => issueService(db).create(companyId, {
title: `${priority} blocker`, priority, status: "blocked",
})));
await db.update(issues).set({ updatedAt: new Date(now.getTime() - 59 * 60_000) }).where(eq(issues.id, tasks[0]!.id));
for (const task of tasks.slice(1)) await db.update(issues).set({ updatedAt: now }).where(eq(issues.id, task.id));
compiledQuery = companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"], updatedWithin: "1h", sort: "priority", limit: 1, offset: 1 });
const created = await card();
expect(Object.keys(created.fingerprint!)).toEqual([tasks[1]!.id]);
const expires = now.getTime() + 60_001;
expect(created.nextEvalAt?.getTime()).toBe(expires);
const service = statusCardEvaluationService(db, { wakeup });
vi.setSystemTime(expires); await service.run(new AbortController().signal);
const pending = (await statusCardService(db).getById(created.id))!;
expect(pending.pendingChangeCount).toBe(2); expect(wakeup).not.toHaveBeenCalled();
vi.setSystemTime(pending.nextEvalAt!); await service.run(new AbortController().signal);
expect(wakeup).toHaveBeenCalledOnce();
const [update] = await db.select().from(statusCardUpdates).where(eq(statusCardUpdates.generationIssueId, (await statusCardService(db).getById(created.id))!.generatingIssueId!));
expect(new Set(update!.changes.map(change => change.issueId))).toEqual(new Set([tasks[1]!.id, tasks[2]!.id]));
});
it("retains a scoped retry while another evaluator owns the card lock", async () => {
const created = await card();
await issueService(db).create(companyId, { title: "Locked blocker", status: "blocked", priority: "medium" });
const acquired = Promise.withResolvers<void>(), release = Promise.withResolvers<void>();
const holder = db.transaction(async tx => {
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${created.id}, 0))`);
acquired.resolve(); await release.promise;
});
await acquired.promise;
const service = statusCardEvaluationService(db, { wakeup });
service.changed(new Set([`changes:${companyId}`]));
try {
await service.run(new AbortController().signal);
expect(await service.next()).toBeGreaterThan(Date.now());
expect((await statusCardService(db).getById(created.id))!.nextEvalAt).toBeNull();
expect(wakeup).not.toHaveBeenCalled();
} finally { release.resolve(); await holder; }
await vi.advanceTimersByTimeAsync(30_000); await service.run(new AbortController().signal);
const pending = (await statusCardService(db).getById(created.id))!;
expect(pending.pendingChangeCount).toBeGreaterThan(0);
vi.setSystemTime(pending.nextEvalAt!); await service.run(new AbortController().signal);
expect(wakeup).toHaveBeenCalledOnce();
});
it("retains the durable five-minute retry after evaluation failure across restart and drain", async () => {
const created = await card();
await db.update(statusCards).set({ queries: [{ scope: "invalid" } as never], nextEvalAt: new Date() }).where(eq(statusCards.id, created.id));
const service = statusCardEvaluationService(db, { wakeup });
await service.run(new AbortController().signal);
const failed = (await statusCardService(db).getById(created.id))!;
const retryAt = failed.nextEvalAt!.getTime();
expect(retryAt).toBeGreaterThan(Date.now() + 299_000); expect(wakeup).not.toHaveBeenCalled();
await start(); expect(coordinator!.nextWakeAt()).toBe(retryAt);
draining = true; await vi.advanceTimersByTimeAsync(5 * 60_000); await settled(); expect(wakeup).not.toHaveBeenCalled();
// Repair the controlled malformed stored-query fault; no tested result is precreated.
await db.update(statusCards).set({ queries: [compiledQuery] }).where(eq(statusCards.id, created.id));
draining = false; await vi.advanceTimersByTimeAsync(30_000); await quiet();
expect((await statusCardService(db).getById(created.id))!.nextEvalAt).toBeNull();
});
it("cancels failed assignments without rerunning a completed generation and settles after retry", async () => {
await start(); const created = await card(); await quiet();
wakeup.mockRejectedValueOnce(new Error("assignment unavailable"));
await issueService(db).create(companyId, { title: "Assignment blocker", status: "blocked", priority: "medium" });
await vi.waitFor(async () => expect((await statusCardService(db).getById(created.id))?.pendingChangeCount).toBeGreaterThan(0));
await vi.advanceTimersByTimeAsync(60_000); await settled();
const failed = (await statusCardService(db).getById(created.id))!;
expect(failed).toMatchObject({ state: "error", generatingIssueId: null, nextEvalAt: null });
expect(errors.length).toBeGreaterThan(0);
const history = await db.select().from(statusCardUpdates).where(eq(statusCardUpdates.cardId, created.id));
const ledger = history.find(row => row.status === "failed")!;
expect(ledger.finishedAt).toBeInstanceOf(Date);
expect((await db.select().from(issues).where(eq(issues.id, ledger.generationIssueId!)))[0]?.status).toBe("cancelled");
const count = wakeup.mock.calls.length; await quiet(); expect(wakeup).toHaveBeenCalledTimes(count);
// An explicit ordinary refresh is still the user's supported retry action.
const retry = await statusCardService(db).requestRefresh(created.id, { trigger: "manual", actor });
expect(retry.generatingIssue).toBeTruthy(); await writeback(created.id, retry.generatingIssue!.id); await quiet();
expect((await statusCardService(db).getById(created.id))!.state).toBe("active");
});
it("budget-owned company resume reactivates reactive work committed while paused", async () => {
await start(); const created = await card(); await quiet();
const budgets = budgetService(db);
await budgets.upsertPolicy(companyId, { scopeType: "company", scopeId: companyId, amount: 100, hardStopEnabled: true, notifyEnabled: false }, "board-user");
const [cost] = await db.insert(costEvents).values({ companyId, agentId, provider: "openai", biller: "openai", billingType: "metered_api", model: "test",
costCents: 200, occurredAt: new Date() }).returning();
await budgets.evaluateCostEvent(cost!);
expect((await db.select().from(companies).where(eq(companies.id, companyId)))[0]?.status).toBe("paused");
await quiet();
await issueService(db).create(companyId, { title: "Paused blocker", status: "blocked", priority: "medium" }); await quiet();
expect(wakeup).not.toHaveBeenCalled();
await budgets.upsertPolicy(companyId, { scopeType: "company", scopeId: companyId, amount: 1000 }, "board-user");
expect((await db.select().from(companies).where(eq(companies.id, companyId)))[0]?.status).toBe("active");
await vi.waitFor(async () => expect((await statusCardService(db).getById(created.id))?.pendingChangeCount).toBeGreaterThan(0));
await vi.advanceTimersByTimeAsync(60_000); await settled(); expect(wakeup).toHaveBeenCalledOnce();
});
it("preserves the first write during drain until resume", async () => {
await start(); const created = await card("manual"); await quiet();
draining = true;
server/src/__tests__/status-card-notifications.test.ts
+31
View File
@@ -0,0 +1,31 @@
import { expect, it, vi } from "vitest";
import { installDatabaseWorkSignals, subscribeDatabaseWork } from "../../../packages/db/src/work-signals.js";
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
import { notifyExecutionRunWork, writeExecutionControl } from "../services/execution-control-notifications.js";
it.each([false, true])("keeps execution queue consumers intact and emits only a known status-card company scope (%s)", async scoped => {
const owner = installDatabaseWorkSignals({ execute: vi.fn(async () => []), transaction: async (fn: (tx: object) => Promise<unknown>) =>
fn({ execute: async () => [{ xid: "1" }] }) });
const queues = [DELIVERY_QUEUES.executionStatus, DELIVERY_QUEUES.executionFinalization, DELIVERY_QUEUES.executionReplacement,
DELIVERY_QUEUES.executionDisposition, DELIVERY_QUEUES.issueMonitors];
const settled: Array<{ queue: string; hints?: string[] }> = [];
const unsubscribe = [...queues, DELIVERY_QUEUES.statusCardEvaluation, DELIVERY_QUEUES.workspaceRecovery].map(queue => subscribeDatabaseWork(owner, queue, (event, hints) => {
if (event === "settled") settled.push({ queue, hints: hints ? [...hints] : undefined });
}));
try {
const scope = scoped ? { companyId: "changed-company", runId: "changed-run" } : undefined;
await owner.transaction(async tx => {
await notifyExecutionRunWork(tx, scope);
expect(settled).toEqual([]);
});
expect(new Set(settled.map(row => row.queue))).toEqual(new Set([...queues, ...(scoped ? [DELIVERY_QUEUES.statusCardEvaluation] : [])]));
expect(settled.find(row => row.queue === DELIVERY_QUEUES.statusCardEvaluation)?.hints)
.toEqual(scoped ? ["changes:changed-company"] : undefined);
settled.length = 0;
const write = vi.fn(async () => "written");
expect(await writeExecutionControl(owner as never, DELIVERY_QUEUES.workspaceRecovery, write, scope)).toBe("written");
expect(write).toHaveBeenCalledOnce();
expect(new Set(settled.map(row => row.queue))).toEqual(new Set([DELIVERY_QUEUES.issueMonitors, DELIVERY_QUEUES.workspaceRecovery,
...(scoped ? [DELIVERY_QUEUES.statusCardEvaluation] : [])]));
} finally { for (const stop of unsubscribe) stop(); }
});
server/src/modules/wake-queue/adapters/postgres.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
import { notifyExternalConversationSettlement } from "../../../services/external-conversation-settlement-notifications.js";
import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
@@ -610,6 +611,7 @@ function buildTransaction(tx: Db, deps: WakeQueuePostgresAdapterDeps, db: Db, ru
},
async queueReviewParticipantRecoveryRun({ companyId, issue, finishingRun, recoveryAgent, sessionBefore, now }) {
await notifyStatusCardIssueChange(tx, companyId);
const executionState = parseIssueExecutionState(issue.executionState);
const wakeupRequest = await tx
.insert(agentWakeupRequests)
server/src/modules/wake-queue/adapters/queued-comment-postgres.ts
+3
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../../../services/status-card-notifications.js";
import { notifyExternalAgentSettlement } from "../../external-agents/index.js";
import { notifyNativeStatusReassessment } from "../../../services/native-status-reassessment-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../../../services/unsafe-workspace-export-notifications.js";
@@ -55,6 +56,7 @@ export type QueuedCommentQueuePostgresAdapterDeps = {
function buildTransaction(tx: Db, companyId: string, issueId: string, deps: QueuedCommentQueuePostgresAdapterDeps): QueuedCommentQueueTransaction {
return {
async updateCommentBody({ issueId, commentId, body, updatedAt }) {
await notifyStatusCardIssueChange(tx, companyId);
const updated = await tx
.update(issueComments)
.set({ body, updatedAt })
@@ -65,6 +67,7 @@ function buildTransaction(tx: Db, companyId: string, issueId: string, deps: Queu
},
async touchIssueUpdatedAt({ issueId, updatedAt }) {
await notifyStatusCardIssueChange(tx, companyId);
await tx.update(issues).set({ updatedAt }).where(and(eq(issues.id, issueId), eq(issues.companyId, companyId)));
},
server/src/routes/execution-workspaces.ts
+5 -1
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../services/status-card-notifications.js";
import { removePluginWorkspace } from "../services/plugin-workspaces.js";
import { canActorReadWorkspaceOperation } from "../services/heartbeat-run-privacy.js";
import { canActorReadExecutionWorkspace, executionWorkspaceReadSqlCondition } from "../services/authorization.js";
@@ -1245,7 +1246,9 @@ export function executionWorkspaceRoutes(db: Db, opts: { pluginWorkerManager?: P
await runtimeLeases.release({ executionWorkspaceId: existing.id, force: true });
if (existing.mode === "shared_workspace") {
await db
await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, existing.companyId);
await tx
.update(issues)
.set({
executionWorkspaceId: null,
@@ -1257,6 +1260,7 @@ export function executionWorkspaceRoutes(db: Db, opts: { pluginWorkerManager?: P
eq(issues.executionWorkspaceId, existing.id),
),
);
});
}
try {
server/src/services/agent-conversations.ts
+3
View File
@@ -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 { FAST_RESPONSE_AGENT_GUIDANCE, fastResponseHistoryBody } from "@paperclipai/shared";
@@ -269,6 +270,7 @@ export async function prepareConversationTurn(
context.conversationReplayThroughCommentId = replayThrough?.id ?? commentId;
}
}
await notifyStatusCardIssueChange(tx, issue.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
await tx
@@ -389,6 +391,7 @@ export async function settleConversationTurn(
issue.conversationState === conversationState
)
return true;
await notifyStatusCardIssueChange(tx, issue.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
if (status === "in_review") await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: issue.companyId, issueId: issue.id });
server/src/services/budgets.ts
+3
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifyApprovalWork } from "./approval-work-notifications.js";
import { notifyCompanyRetentionWork } from "./retention-work-notifications.js";
import { addCents, compareCents, normalizeCents } from "@paperclipai/shared";
@@ -301,6 +302,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
const scope = await resolveScopeRecord(db, "company", policy.scopeId);
if (scope.paused) return;
await notifyCompanyRetentionWork(db, policy.companyId);
await notifyStatusCardIssueChange(db, policy.companyId);
await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors);
@@ -361,6 +363,7 @@ export function budgetServiceInTransaction(db: Db, publications: ActivityPublica
const scope = await resolveScopeRecord(db, "company", policy.scopeId);
if (scope.pauseReason !== "budget") return;
await notifyCompanyRetentionWork(db, policy.companyId);
await notifyStatusCardIssueChange(db, policy.companyId);
await notifyDeliveryWork(db, DELIVERY_QUEUES.agentTimers);
await notifyDeliveryWork(db, DELIVERY_QUEUES.routineSchedules);
await notifyDeliveryWork(db, DELIVERY_QUEUES.issueMonitors);
server/src/services/chat-channels/telegram/attachment-failure.ts
+2
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../../status-card-notifications.js";
import { createProviderEffects, type ProviderEffectTarget } from "../provider-effects.js";
import type { EndpointRow, RuntimeContext, InboundRuntimeContext } from "../runtime-types.js";
@@ -82,6 +83,7 @@ export function createTelegramAttachmentFailure(db: Db, options: TelegramAttachm
// Paperclip-visible failure. The message link stays authoritative for
// exact provider redelivery, while no agent wakeup is enqueued for an
// empty turn that contains no usable bytes.
await notifyStatusCardIssueChange(tx, input.endpoint.companyId);
await tx
.update(issueComments)
.set({ body: visibleFailure, updatedAt: new Date() })
server/src/services/company-search.ts
+14 -1
View File
@@ -548,7 +548,7 @@ export function companySearchService(db: Db) {
search: async (
companyId: string,
query: CompanySearchQuery,
options?: { issueReadCondition?: SQL<boolean>; projectReadCondition?: SQL<boolean> },
options?: { issueReadCondition?: SQL<boolean>; projectReadCondition?: SQL<boolean>; onNextMembershipChange?: (at: Date | null) => void },
): Promise<CompanySearchResponse> => {
const taskSearch = parseTaskSearch(query.q);
const normalizedQuery = taskSearch.normalizedQuery;
@@ -559,6 +559,7 @@ export function companySearchService(db: Db) {
const limit = query.limit;
const offset = query.offset;
if (!hasSearchText && !issueOnlyFiltersActive(query)) {
options?.onNextMembershipChange?.(null);
return {
query: query.q,
normalizedQuery,
@@ -727,6 +728,13 @@ export function companySearchService(db: Db) {
}
}
const membershipWindow = companySearchUpdatedWithinMs(query.updatedWithin);
if (options?.onNextMembershipChange && membershipWindow !== null && scopeIncludesIssues(scope)) {
// Includes rows outside the visible page: their expiry can move an
// offset/priority-ranked result into the watched set without a write.
branches.push(sql`SELECT 'membership-expiry' AS kind, min(m.updated_at)::text AS value,
0::int AS count ${countTail} FROM matched m ${branchWhere([scopeCond, ...facetsAll])}`);
}
const resultRows = await db.execute(sql`
${taskSearchCtes(companyId, taskSearch, scope !== "issues", and(...issueFilters), options?.issueReadCondition)}
${sql.join(branches, sql` UNION ALL `)}
@@ -738,6 +746,7 @@ export function companySearchService(db: Db) {
totals: { current: 0, unfiltered: 0, omit: {} },
};
const issueRowsRaw: IssueSearchRow[] = [];
let nextMembershipChange: Date | null = null;
for (const row of resultRows) {
if (row.kind === "result") {
issueRowsRaw.push({
@@ -762,6 +771,9 @@ export function companySearchService(db: Db) {
});
continue;
}
if (row.kind === "membership-expiry" && row.value && membershipWindow !== null) {
nextMembershipChange = new Date(new Date(row.value).getTime() + membershipWindow + 1);
}
const count = Number(row.count ?? 0);
if (row.kind === "type:issue") aggregates.typeCounts.issue = count;
else if (row.kind === "type:comment") aggregates.typeCounts.comment = count;
@@ -782,6 +794,7 @@ export function companySearchService(db: Db) {
aggregates.totals.omit[row.kind.slice("total:omit:".length) as CompanySearchIssueFilterKey] = count;
}
}
options?.onNextMembershipChange?.(nextMembershipChange);
return { rows: await enrichIssueSnippets(issueRowsRaw), aggregates };
}
server/src/services/documents.ts
+5
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { randomUUID } from "node:crypto";
import { and, asc, desc, eq, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@@ -238,6 +239,7 @@ export function documentService(db: Db) {
for (let attempt = 0; attempt < maxAttempts; attempt += 1) {
try {
return await db.transaction(async (tx) => {
await notifyStatusCardIssueChange(tx, issue.companyId);
const now = new Date();
const existing = await tx
.select({
@@ -592,6 +594,7 @@ export function documentService(db: Db) {
});
}
await db.transaction(async (tx) => {
for (const companyId of new Set(rows.map(row => row.companyId))) await notifyStatusCardIssueChange(tx, companyId);
await insertRowsInChunks(tx, documents, documentRows);
await insertRowsInChunks(tx, documentRevisions, revisionRows);
await insertRowsInChunks(tx, issueDocuments, linkRows);
@@ -662,6 +665,7 @@ export function documentService(db: Db) {
})
.returning();
await notifyStatusCardIssueChange(tx, existing.companyId);
await tx
.update(documents)
.set({
@@ -818,6 +822,7 @@ export function documentService(db: Db) {
}
await tx.delete(issueDocuments).where(eq(issueDocuments.documentId, existing.id));
await notifyStatusCardIssueChange(tx, existing.companyId);
await tx.delete(documents).where(eq(documents.id, existing.id));
return {
server/src/services/execution-workspaces.ts
+4 -1
View File
@@ -2530,13 +2530,16 @@ export function executionWorkspaceService(db: Db, opts: ExecutionWorkspaceServic
forceWorktreeRemoval: false,
});
if (cleanup.cleaned && workspace.mode === "shared_workspace") {
await db
await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, workspace.companyId);
await tx
.update(issues)
.set({ executionWorkspaceId: null, updatedAt: now() })
.where(and(
eq(issues.companyId, workspace.companyId),
eq(issues.executionWorkspaceId, workspace.id),
));
});
}
const cleanupReason = [ISSUE_TERMINAL_WORKSPACE_CLEANUP_REASON, ...cleanup.warnings].join(" | ");
if (!cleanup.cleaned || cleanup.warnings.length > 0) {
server/src/services/heartbeat/queue.ts
+3
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
@@ -1267,6 +1268,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
}
async function bindClaimedIssueExecution(tx: Db, ownsIssue: boolean, claimedRun: typeof heartbeatRuns.$inferSelect | null | undefined) {
if (!claimedRun || !issueId || !ownsIssue) return;
await notifyStatusCardIssueChange(tx, run.companyId);
await tx.update(issues).set({
executionRunId: claimedRun.id,
executionAgentNameKey: normalizeAgentNameKey(agent.name),
@@ -3758,6 +3760,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
.from(agents)
.where(eq(agents.id, legacyRun.agentId))
.then((rows) => rows[0] ?? null);
await notifyStatusCardIssueChange(tx, issue.companyId);
await tx
.update(issues)
.set({
server/src/services/heartbeat/scheduling.ts
+6
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import { notifyIssueReviewState } from "../issue-review-state-notifications.js";
import { insertIssueComment } from "../issue-comment-writes.js";
@@ -595,6 +596,7 @@ export function createHeartbeatScheduling(
try {
await db.transaction(async (tx) => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyStatusCardIssueChange(tx, claimed.companyId);
const released = await tx.update(issues)
.set({ monitorWakeRequestedAt: null, updatedAt: input.now })
.where(issueMonitorClaimCondition(claimed)).returning({ id: issues.id });
@@ -628,6 +630,7 @@ export function createHeartbeatScheduling(
// failed, and never clobber a monitor that was replaced in the meantime.
await db.transaction(async tx => {
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyStatusCardIssueChange(tx, claimed.companyId);
await tx.update(issues).set({ monitorWakeRequestedAt: null, updatedAt: input.now })
.where(issueMonitorClaimCondition(claimed));
});
@@ -827,6 +830,7 @@ export function createHeartbeatScheduling(
}
const consumed = await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, claimed.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
return tx.update(issues)
.set({
@@ -977,6 +981,7 @@ export function createHeartbeatScheduling(
const staleClaimThreshold = new Date(now.getTime() - 5 * 60 * 1000);
const claimed = await db.transaction(async (tx) => {
await notifyStatusCardIssueChange(tx, issue.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const [updated] = await tx
.update(issues)
@@ -1050,6 +1055,7 @@ export function createHeartbeatScheduling(
for (const due of dueMonitors) {
if (!canContinue()) break;
const claimed = await db.transaction(async (tx) => {
await notifyStatusCardIssueChange(tx, due.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
const [updated] = await tx
.update(issues)
server/src/services/issue-thread-interactions.ts
+30 -33
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
import { notifyIssueReviewState } from "./issue-review-state-notifications.js";
import { notifyChatConfirmationResolution } from "./chat-confirmation-notifications.js";
@@ -376,7 +377,7 @@ type ResolvedInteractionResult = {
};
type IssueThreadInteractionRow = typeof issueThreadInteractions.$inferSelect;
type IssueTouchDb = Pick<Db, "update">;
type IssueTouchDb = Pick<Db, "update" | "transaction">;
function isNativeCompletionReview(
row: Pick<IssueThreadInteractionRow, "kind" | "payload">,
@@ -801,11 +802,11 @@ function hydrateInteraction(
}
}
async function touchIssue(db: IssueTouchDb, issueId: string) {
await db
.update(issues)
.set({ updatedAt: new Date() })
.where(eq(issues.id, issueId));
async function touchIssue(db: IssueTouchDb, issueId: string, companyId: string) {
await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, companyId);
await tx.update(issues).set({ updatedAt: new Date() }).where(eq(issues.id, issueId));
});
}
function isTerminalIssueStatus(status: string) {
@@ -2013,7 +2014,7 @@ async function expireStaleRequestConfirmationTarget(
"Interaction has already been resolved",
);
}
await touchIssue(tx, args.row.issueId);
await touchIssue(tx, args.row.issueId, args.row.companyId);
const expired = hydrateInteraction(updated);
await enqueueTerminalIssueInteractionChatPublications(tx, expired);
return expired;
@@ -2468,7 +2469,7 @@ export function issueThreadInteractionService(
};
}
} else {
await touchIssue(tx, args.issue.id);
await touchIssue(tx, args.issue.id, args.issue.companyId);
}
// Snapshot the continuation target only after every lifecycle mutation
@@ -2682,7 +2683,7 @@ export function issueThreadInteractionService(
tx,
);
} else {
await touchIssue(tx, args.issue.id);
await touchIssue(tx, args.issue.id, args.issue.companyId);
}
await enqueueTerminalIssueInteractionChatPublications(
tx as unknown as Db,
@@ -2864,7 +2865,7 @@ export function issueThreadInteractionService(
await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, hydrateInteraction(previous));
}
}
await touchIssue(tx, issue.id);
await touchIssue(tx, issue.id, issue.companyId);
inserted = true;
await enqueueIssueInteractionChatPublications(tx as unknown as Db, hydrateInteraction(row));
await notifyDeliveryWork(tx, DELIVERY_QUEUES.connection);
@@ -2908,7 +2909,7 @@ export function issueThreadInteractionService(
)
.returning();
if (!updated) throw interactionAlreadyResolvedError();
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
return hydrateInteraction(updated) as ConnectionIntentInteraction;
},
resolveConnectionIntent: async (
@@ -2972,7 +2973,7 @@ export function issueThreadInteractionService(
return row;
});
if (!updated) throw interactionAlreadyResolvedError();
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
const interaction = hydrateInteraction(
updated,
) as ConnectionIntentInteraction;
@@ -3351,7 +3352,7 @@ export function issueThreadInteractionService(
})
.where(eq(issueThreadInteractions.id, current.id))
.returning();
await touchIssue(tx, issue.id);
await touchIssue(tx, issue.id, issue.companyId);
return row;
});
return hydrateInteraction(updated);
@@ -3414,10 +3415,8 @@ export function issueThreadInteractionService(
if (updated) cancelled.push(updated);
}
for (const issueId of new Set(
cancelled.map((interaction) => interaction.issueId),
)) {
await touchIssue(db, issueId);
for (const [issueId, companyId] of new Map(cancelled.map(interaction => [interaction.issueId, interaction.companyId]))) {
await touchIssue(db, issueId, companyId);
}
await emitResolvedInteractionsTelemetry(db, cancelled);
return cancelled;
@@ -3506,10 +3505,8 @@ export function issueThreadInteractionService(
}
if (expired.length > 0) {
for (const issueId of new Set(
expired.map((interaction) => interaction.issueId),
)) {
await touchIssue(db, issueId);
for (const [issueId, companyId] of new Map(expired.map(interaction => [interaction.issueId, interaction.companyId]))) {
await touchIssue(db, issueId, companyId);
}
await emitResolvedInteractionsTelemetry(db, expired);
}
@@ -4052,7 +4049,7 @@ export function issueThreadInteractionService(
return interaction;
}
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
if (superseded.length > 0) {
await emitResolvedInteractionsTelemetry(
db,
@@ -4357,7 +4354,7 @@ export function issueThreadInteractionService(
.where(eq(issueThreadInteractions.id, interactionId))
.returning();
await touchIssue(tx, issue.id);
await touchIssue(tx, issue.id, issue.companyId);
await mutationOptions.afterResolveInTransaction?.(tx, hydrateInteraction(updated));
await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, hydrateInteraction(updated));
current.status = updated.status;
@@ -4549,7 +4546,7 @@ export function issueThreadInteractionService(
throw interactionAlreadyResolvedError();
}
await touchIssue(tx, issue.id);
await touchIssue(tx, issue.id, issue.companyId);
await mutationOptions.afterResolveInTransaction?.(tx, hydrateInteraction(updated));
if (complete) await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, hydrateInteraction(updated));
return {
@@ -4622,7 +4619,7 @@ export function issueThreadInteractionService(
throw interactionAlreadyResolvedError();
}
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
const rejected = hydrateInteraction(updated);
await emitInteractionResolvedTelemetry(db, rejected);
return rejected;
@@ -4720,7 +4717,7 @@ export function issueThreadInteractionService(
}
if (expired.length > 0) {
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
await emitResolvedInteractionsTelemetry(db, expired);
}
return expired;
@@ -4952,7 +4949,7 @@ export function issueThreadInteractionService(
}
if (expired.length > 0) {
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
await emitResolvedInteractionsTelemetry(db, expired);
}
return expired;
@@ -5049,7 +5046,7 @@ export function issueThreadInteractionService(
}
if (expired.length > 0) {
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
await emitResolvedInteractionsTelemetry(db, expired);
}
return expired;
@@ -5189,7 +5186,7 @@ export function issueThreadInteractionService(
if (updated) expired.push(hydrateInteraction(updated));
}
if (expired.length > 0) {
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
await emitResolvedInteractionsTelemetry(db, expired);
}
return expired;
@@ -5294,7 +5291,7 @@ export function issueThreadInteractionService(
return row;
});
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
const withdrawn = hydrateInteraction(updated);
await emitInteractionResolvedTelemetry(db, withdrawn);
return withdrawn;
@@ -5417,7 +5414,7 @@ export function issueThreadInteractionService(
return row;
});
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
const answered = hydrateInteraction(updated);
await emitInteractionResolvedTelemetry(db, answered);
return answered;
@@ -5553,7 +5550,7 @@ export function issueThreadInteractionService(
return row;
});
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
const skipped = hydrateInteraction(updated);
await emitInteractionResolvedTelemetry(db, skipped);
return skipped;
@@ -5630,7 +5627,7 @@ export function issueThreadInteractionService(
return row;
});
await touchIssue(db, issue.id);
await touchIssue(db, issue.id, issue.companyId);
const cancelled = hydrateInteraction(updated);
await emitInteractionResolvedTelemetry(db, cancelled);
return cancelled;
server/src/services/issues.ts
+24 -10
View File
@@ -7730,6 +7730,7 @@ export function issueService(db: Db) {
}
const now = new Date();
await notifyStatusCardIssueChange(tx, lockedIssue.companyId);
const adopted = await tx
.update(issues)
.set({
@@ -7783,13 +7784,14 @@ export function issueService(db: Db) {
sql`select ${heartbeatRuns.id} from ${heartbeatRuns} where ${heartbeatRuns.id} = ${input.actorRunId} for update`,
);
const actorRun = await tx
.select({ status: heartbeatRuns.status })
.select({ status: heartbeatRuns.status, companyId: heartbeatRuns.companyId })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.actorRunId))
.then((rows) => rows[0] ?? null);
if (!actorRun || TERMINAL_HEARTBEAT_RUN_STATUSES.has(actorRun.status))
return null;
await notifyStatusCardIssueChange(tx, actorRun.companyId);
const now = new Date();
const adopted = await tx
.update(issues)
@@ -10892,6 +10894,7 @@ export function issueService(db: Db) {
}
}
await notifyStatusCardIssueChange(tx, companyId);
await insertRowsInChunks(tx, issues, issueRows);
await insertRowsInChunks(tx, issueLabels, labelRows);
});
@@ -10908,6 +10911,7 @@ export function issueService(db: Db) {
const censorUsernameInLogs = (await instanceSettings.getGeneral())
.censorUsernameInLogs;
await db.transaction(async (tx) => {
for (const companyId of new Set(rows.map(row => row.companyId))) await notifyStatusCardIssueChange(tx, companyId);
const commentRows = rows.map((row) => {
const createdAt = row.createdAt ? new Date(row.createdAt) : null;
return {
@@ -11922,7 +11926,9 @@ export function issueService(db: Db) {
);
if (settings?.environmentId !== environmentId) continue;
await db
await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, companyId);
await tx
.update(issues)
.set({
executionWorkspaceSettings: {
@@ -11932,6 +11938,7 @@ export function issueService(db: Db) {
updatedAt: new Date(),
})
.where(eq(issues.id, row.id));
});
cleared += 1;
}
@@ -12088,6 +12095,7 @@ export function issueService(db: Db) {
)
: isNull(issues.executionRunId);
const updated = await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, issueCompany.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
return tx
@@ -12140,7 +12148,9 @@ export function issueService(db: Db) {
current.executionRunId === checkoutRunId) &&
checkoutRunId
) {
const adopted = await db
const adopted = await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, issueCompany.companyId);
return tx
.update(issues)
.set({
checkoutRunId,
@@ -12161,6 +12171,7 @@ export function issueService(db: Db) {
)
.returning()
.then((rows) => rows[0] ?? null);
});
if (adopted) return adopted;
}
@@ -12217,6 +12228,7 @@ export function issueService(db: Db) {
}
const staleExecutionRunId = current.executionRunId;
const adopted = await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, issueCompany.companyId);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors);
await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers);
return tx
@@ -12524,6 +12536,7 @@ export function issueService(db: Db) {
patch.assigneeAgentId = null;
}
await notifyStatusCardIssueChange(tx, existing.companyId);
await notifyTerminalWorkspaceWork(tx, `issue:${id}`);
if (existing.checkoutRunId || existing.executionRunId || options.clearAssignee) {
await notifyExternalConversationSettlement(tx, { companyId: existing.companyId, issueId: id });
@@ -12581,12 +12594,12 @@ export function issueService(db: Db) {
return created;
},
deleteLabel: async (id: string) =>
db
.delete(labels)
.where(eq(labels.id, id))
.returning()
.then((rows) => rows[0] ?? null),
deleteLabel: async (id: string) => db.transaction(async tx => {
const [label] = await tx.select({ companyId: labels.companyId }).from(labels).where(eq(labels.id, id));
if (!label) return null;
await notifyStatusCardIssueChange(tx, label.companyId);
return tx.delete(labels).where(eq(labels.id, id)).returning().then(rows => rows[0] ?? null);
}),
listComments: async (
issueId: string,
@@ -12838,7 +12851,7 @@ export function issueService(db: Db) {
},
dbOrTx: any = db,
): Promise<IssueComment> {
if (dbOrTx === db && (actor.runId || actor.userId)) {
if (dbOrTx === db) {
const append = () =>
measureRunTransaction(db, "response", async (tx) => {
// Serialize run-authored comments on the issue so a provider retry
@@ -13081,6 +13094,7 @@ export function issueService(db: Db) {
currentUserRedactionOptions.enabled,
);
}
await notifyStatusCardIssueChange(dbOrTx, issue.companyId);
comment =
shouldUpgradeExternalAuthorization ||
shouldUpgradeAttachmentAuthorization
server/src/services/native-status-reassessment-notifications.ts
+3 -1
View File
@@ -1,8 +1,10 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js";
export type NativeStatusReassessmentScope = { companyId: string; issueId?: string; runId?: string; agentId?: string };
export function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
export async function notifyNativeStatusReassessment(tx: object, scope: NativeStatusReassessmentScope) {
await notifyStatusCardIssueChange(tx, scope.companyId);
return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeStatusReassessment,
JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId }));
}
server/src/services/projects.ts
+3
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js";
import { listProjectWorkspaceProviders, resolveProjectWorkspaceProvider } from "./project-workspace-provider.js";
import { retireRemovedPluginWorkspaceDependents } from "./plugin-workspaces.js";
@@ -1013,6 +1014,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool
const row = await db.transaction(async (tx) => {
if (updates.visibility !== undefined) {
await notifyStatusCardIssueChange(tx, existingProject.companyId);
await notifyChatInboundWakeNotices(tx, { companyId: existingProject.companyId });
await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${`issue-privacy-tree:${existingProject.companyId}`}, 0))`);
}
@@ -1086,6 +1088,7 @@ export function projectService(db: Db, options: { captureWorkspaceUpdates?: bool
await tx.select({ id: projectWorkspaces.id }).from(projectWorkspaces).where(eq(projectWorkspaces.projectId, id)).for("update");
await recordProjectLifecycleEvent(tx as unknown as Db, project.companyId, id, "delete");
await notifyDeliveryWork(tx, DELIVERY_QUEUES.skillSourceSync);
await notifyStatusCardIssueChange(tx, project.companyId);
const [row] = await tx.delete(projects).where(eq(projects.id, id)).returning();
return row ? { ...row, urlKey: deriveProjectUrlKey(row.name, row.id) } : null;
}),
server/src/services/status-card-evaluation.ts
+9 -5
View File
@@ -18,17 +18,18 @@ export function statusCardEvaluationService(db: Db, heartbeat: IssueAssignmentWa
let reconstructReactive = true;
let allChanges = false;
const changes = new Set<string>();
let enabled = false;
let enabled: boolean | null = null;
let retryAt: number | null = null;
return {
changed(hints?: ReadonlySet<string>) {
for (const hint of hints ?? ["*"]) {
if (hint === "*" || hint === "changes:*") allChanges = true;
if (hint === "*" || hint === "changes:*") { allChanges = true; if (hint === "*") enabled = null; }
else if (hint.startsWith("changes:")) changes.add(hint.slice("changes:".length));
}
},
async run(signal: AbortSignal) {
enabled = (await settings.getExperimental()).enableStatusCards === true;
if (enabled === null) enabled = (await settings.getExperimental()).enableStatusCards === true;
if (!enabled || signal.aborted) return;
const reconstruct = reconstructReactive;
const evaluateAllChanges = allChanges;
@@ -37,7 +38,9 @@ export function statusCardEvaluationService(db: Db, heartbeat: IssueAssignmentWa
reconstructReactive = false;
allChanges = false;
try {
const result = await cards.tickDueStatusCards(new Date(), evaluateAllChanges ? "changed" : reconstruct && companiesChanged.size === 0 ? "all" : companiesChanged);
const result = await cards.tickDueStatusCards(new Date(), evaluateAllChanges ? "changed" : companiesChanged, reconstruct);
retryAt = result.deferredCompanyIds.size ? Date.now() + 30_000 : null;
for (const companyId of result.deferredCompanyIds) changes.add(companyId);
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.
@@ -74,7 +77,8 @@ export function statusCardEvaluationService(db: Db, heartbeat: IssueAssignmentWa
.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;
const deadline = next?.at?.getTime() ?? null;
return retryAt === null ? deadline : Math.min(deadline ?? Infinity, retryAt);
},
};
}
server/src/services/status-cards.ts
+23 -30
View File
@@ -25,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, companySearchUpdatedWithinMs } from "./company-search.js";
import { companySearchService } from "./company-search.js";
import { issueService } from "./issues.js";
import { SUMMARIZER_BUILT_IN_KEY } from "./summary-slots.js";
import {
@@ -47,7 +47,6 @@ 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"]);
@@ -590,22 +589,15 @@ export function statusCardService(
return loadIssueSummaries(card.companyId, card.mentionedIssueIds ?? [], options);
}
async function executeQueries(card: StatusCardRow, options: StatusCardReadOptions = {}) {
const issueMap = new Map<string, WatchedIssue>();
async function executeQueries(card: StatusCardRow, options: StatusCardReadOptions = {}, onNextMembershipChange?: (at: Date | null) => void) {
const issueMap = new Map<string, CompanySearchIssueSummary>();
for (const storedQuery of card.queries) {
const query = companySearchQuerySchema.parse(storedQuery);
const response = await searchSvc.search(card.companyId, query, {
issueReadCondition: options.issueReadCondition,
issueReadCondition: options.issueReadCondition, onNextMembershipChange,
});
for (const result of response.results) {
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 });
if (result.type === "issue" && result.issue) issueMap.set(result.issue.id, result.issue);
}
}
// Issues mentioned in the latest summary join the watched set alongside the
@@ -615,11 +607,7 @@ export function statusCardService(
(card.mentionedIssueIds ?? []).filter((issueId) => !issueMap.has(issueId)),
options,
);
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;
}
for (const issue of mentioned) issueMap.set(issue.id, issue);
const snapshot = [...issueMap.values()];
if (snapshot.length === 0) return snapshot;
const latestHumanComments = await db
@@ -643,11 +631,10 @@ export function statusCardService(
return snapshot.map((issue) => ({ ...issue, latestHumanCommentAt: commentByIssueId.get(issue.id) ?? null }));
}
function nextEvaluationAt(card: StatusCardRow, snapshot: WatchedIssue[], now: Date) {
function nextEvaluationAt(card: StatusCardRow, membershipAt: Date | null, 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;
return card.refreshPolicy.mode === "reactive" && (card.refreshPolicy.triggers.membershipChanges || card.refreshPolicy.triggers.anyUpdate)
? membershipAt : interval;
}
async function requestRefresh(cardId: string, input: {
@@ -670,13 +657,14 @@ export function statusCardService(
}
const now = input.now ?? new Date();
const snapshot = await executeQueries(card);
let membershipAt: Date | null = null;
const snapshot = await executeQueries(card, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; });
const fingerprint = buildStatusCardFingerprint(snapshot);
const allChanges = diffStatusCardFingerprint(card.fingerprint as StatusCardFingerprint | null, fingerprint);
const changes = filterStatusCardChanges(allChanges, card.refreshPolicy);
const trigger = input.trigger ?? "manual";
const forceRun = trigger === "manual" || trigger === "restore";
const nextEvalAt = nextEvaluationAt(card, snapshot, now);
const nextEvalAt = nextEvaluationAt(card, membershipAt, now);
if (!forceRun && changes.length === 0) {
const [next] = await mutate(card.companyId, tx => tx.update(statusCards).set({
pendingChangeCount: 0,
@@ -797,9 +785,12 @@ export function statusCardService(
return { card: next, generatingIssue: generationIssue!, alreadyGenerating: deduplicated, enqueued: true, kind, changes };
}
async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet<string> | "all" | "changed") {
const reactiveScope = reactiveCompanies === "all" ? isNull(statusCards.nextEvalAt) : reactiveCompanies === "changed" ? sql`true` : reactiveCompanies?.size
async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet<string> | "changed", reconstruct = false) {
const changedScope = reactiveCompanies === "changed" ? sql`true` : reactiveCompanies?.size
? inArray(statusCards.companyId, [...reactiveCompanies]) : sql`false`;
// Startup restores every disarmed reactive scope. A concurrent committed
// source hint can additionally preempt its company's saved future deadline.
const reactiveScope = or(reconstruct ? isNull(statusCards.nextEvalAt) : undefined, changedScope);
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),
@@ -808,6 +799,7 @@ export function statusCardService(
.then(rows => rows.map(row => row.card));
const enqueued: Array<{ cardId: string; generatingIssue: typeof issues.$inferSelect }> = [];
let evaluated = 0;
const deferredCompanyIds = new Set<string>();
for (const candidate of due) {
// nextEvalAt is both a retry lease and an evaluation deadline. A source
// event may preempt a genuine future deadline, so serialize scheduler
@@ -817,7 +809,7 @@ export function statusCardService(
const [lock] = await lockTx.execute<{ acquired: boolean }>(
sql`select pg_try_advisory_xact_lock(hashtextextended(${candidate.id}, 0)) as acquired`,
);
if (!lock?.acquired) return;
if (!lock?.acquired) { deferredCompanyIds.add(candidate.companyId); 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),
@@ -836,7 +828,7 @@ export function statusCardService(
}
});
}
return { evaluated, enqueued };
return { evaluated, enqueued, deferredCompanyIds };
}
async function writeSummary(cardId: string, input: WriteStatusCardSummary, actor: StatusCardWriter) {
@@ -866,7 +858,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 watchedSnapshot = await executeQueries({ ...current, mentionedIssueIds });
let membershipAt: Date | null = null;
const watchedSnapshot = await executeQueries({ ...current, mentionedIssueIds }, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; });
const watchedNow = buildStatusCardFingerprint(watchedSnapshot);
let snapshot: StatusCardFingerprint;
if (payloadFingerprint) {
@@ -940,7 +933,7 @@ export function statusCardService(
lastChangeAt: null,
nextEvalAt: current.refreshPolicy.mode === "reactive" &&
filterStatusCardChanges(diffStatusCardFingerprint(snapshot, watchedNow), current.refreshPolicy).length > 0
? now : nextEvaluationAt(current, watchedSnapshot, now),
? now : nextEvaluationAt(current, membershipAt, 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");
server/src/services/task-watchdogs.ts
+5 -1
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { createHash } from "node:crypto";
import { and, asc, eq, gte, inArray, isNull, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@@ -1398,10 +1399,13 @@ export function taskWatchdogService(db: Db, deps: TaskWatchdogServiceDeps = {})
}) ?? fallback
: fallback;
if (!shouldReopen && watchdogIssue.originFingerprint !== input.classification.stopFingerprint) {
await db
await db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, input.watchdog.companyId);
await tx
.update(issues)
.set({ originFingerprint: input.classification.stopFingerprint, updatedAt: new Date() })
.where(and(eq(issues.companyId, input.watchdog.companyId), eq(issues.id, watchdogIssue.id)));
});
watchdogIssue.originFingerprint = input.classification.stopFingerprint;
}
await issuesSvc.addComment(
server/src/services/workspace-runtime.ts
+12 -10
View File
@@ -1,3 +1,4 @@
import { notifyStatusCardIssueChange } from "./status-card-notifications.js";
import { insertIssueComment } from "./issue-comment-writes.js";
import { restoreGitWorkspaceContents } from "./git-workspace-contents.js";
import { savedGitWorkspaceRestoration, ensureSavedGitCommit, GitCacheRestorationFailure } from "./git-workspace-restoration.js";
@@ -1636,10 +1637,11 @@ async function writeDirtyQuarantineAuditComments(input: {
let sourceAuditCommentId: string | null = null;
let claimantAuditCommentId: string | null = null;
const now = new Date();
if (input.evidence.sourceIssueId) {
const sourceIssueId = input.evidence.sourceIssueId;
if (sourceIssueId) {
const [sourceComment] = await insertIssueComment(input.db, {
companyId: input.companyId,
issueId: input.evidence.sourceIssueId,
issueId: sourceIssueId,
authorAgentId: null,
authorUserId: null,
authorType: "system",
@@ -1647,10 +1649,10 @@ async function writeDirtyQuarantineAuditComments(input: {
body,
});
sourceAuditCommentId = sourceComment?.id ?? null;
await input.db
.update(issues)
.set({ updatedAt: now })
.where(eq(issues.id, input.evidence.sourceIssueId));
await input.db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, input.companyId);
await tx.update(issues).set({ updatedAt: now }).where(eq(issues.id, sourceIssueId));
});
}
const claimantIssueId = input.evidence.contention?.claimedByIssueId ?? null;
@@ -1665,10 +1667,10 @@ async function writeDirtyQuarantineAuditComments(input: {
body,
});
claimantAuditCommentId = claimantComment?.id ?? null;
await input.db
.update(issues)
.set({ updatedAt: now })
.where(eq(issues.id, claimantIssueId));
await input.db.transaction(async tx => {
await notifyStatusCardIssueChange(tx, input.companyId);
await tx.update(issues).set({ updatedAt: now }).where(eq(issues.id, claimantIssueId));
});
}
return { sourceAuditCommentId, claimantAuditCommentId };