diff --git a/doc/plans/2026-10-30-status-cards-sleep.md b/doc/plans/2026-10-30-status-cards-sleep.md index 2af51ff65cf..dbf229cdace 100644 --- a/doc/plans/2026-10-30-status-cards-sleep.md +++ b/doc/plans/2026-10-30-status-cards-sleep.md @@ -100,3 +100,26 @@ An expired saved deadline cannot schedule an immediate skipped-attempt loop. Existing startup, status-card, company-search, execution and monitor suites cover the production wiring and unchanged consumers. Exact commands and results belong in the PR verification record; this document does not claim deployed savings. + +### Review corrections: pool capacity and restart deadline ownership + +Scheduled evaluation holds its advisory transaction on `withDedicatedDbConnection`, +not on the ordinary write pool. The helper ends that physical connection in +`finally`, including contention and failed evaluation. Its lock still spans the +actual refresh settlement. Summary writeback runs watched-query reads on its +existing transaction, so a one-connection main pool completes the full flow. + +Migration0419 adds nullable internal `next_eval_reason` to the existing card row. +Membership deadlines may be reconstructed at startup after a committed source +hint was lost. Policy eligibility and the atomically recorded five-minute retry +lease retain their saved deadline. Explicit refresh deadlines remain due work. +The HTTP card response removes this internal provenance; the public schema stays +unchanged. + +For legacy rows, a pending change hash preserves policy eligibility. An untagged +active deadline farther than five minutes away cannot be an outstanding +five-minute claim, so startup rechecks it. An ambiguous untagged deadline within +five minutes keeps its remaining time without extension. This bounds migration +compatibility delay while preserving prior crash/failure recovery. After the +first successful evaluation all newly saved deadlines have explicit ownership. +No recurring compatibility scan is added. diff --git a/packages/db/src/migrations/0419_moaning_vulture.sql b/packages/db/src/migrations/0419_moaning_vulture.sql new file mode 100644 index 00000000000..03ea349aec1 --- /dev/null +++ b/packages/db/src/migrations/0419_moaning_vulture.sql @@ -0,0 +1 @@ +ALTER TABLE "status_cards" ADD COLUMN "next_eval_reason" text; \ No newline at end of file diff --git a/packages/db/src/migrations/meta/0414_snapshot.json b/packages/db/src/migrations/meta/0419_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0414_snapshot.json rename to packages/db/src/migrations/meta/0419_snapshot.json index 2679e7d98bf..d03afbb279d 100644 --- a/packages/db/src/migrations/meta/0414_snapshot.json +++ b/packages/db/src/migrations/meta/0419_snapshot.json @@ -1,6 +1,6 @@ { - "id": "27c9008c-60e3-449b-9630-9e4977fad444", - "prevId": "5690a14a-f9cc-41a6-af51-c27f688f4ef6", + "id": "cce00822-b7a4-4acd-9c59-0f6f5822c0cf", + "prevId": "c777fa84-85a6-418f-bea1-0d60b56cf071", "version": "7", "dialect": "postgresql", "tables": { @@ -19052,6 +19052,98 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.company_skill_mutation_receipts": { + "name": "company_skill_mutation_receipts", + "schema": "", + "columns": { + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "fingerprint": { + "name": "fingerprint", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "skill_id": { + "name": "skill_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "version_id": { + "name": "version_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "response": { + "name": "response", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "company_skill_mutation_receipts_key_idx": { + "name": "company_skill_mutation_receipts_key_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "company_skill_mutation_receipts_company_id_companies_id_fk": { + "name": "company_skill_mutation_receipts_company_id_companies_id_fk", + "tableFrom": "company_skill_mutation_receipts", + "tableTo": "companies", + "columnsFrom": [ + "company_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.company_skill_repository_snapshots": { "name": "company_skill_repository_snapshots", "schema": "", @@ -19079,7 +19171,46 @@ "name": "file_inventory", "type": "jsonb", "primaryKey": false, - "notNull": true + "notNull": true, + "default": "'[]'::jsonb" + }, + "storage": { + "name": "storage", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "upload": { + "name": "upload", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "restore": { + "name": "restore", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "migration_at": { + "name": "migration_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false, + "default": "now()" + }, + "maintenance_at": { + "name": "maintenance_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false, + "default": "now()" + }, + "restore_retry_at": { + "name": "restore_retry_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false }, "created_at": { "name": "created_at", @@ -19110,6 +19241,38 @@ "concurrently": false, "method": "btree", "with": {} + }, + "company_skill_snapshots_migration_idx": { + "name": "company_skill_snapshots_migration_idx", + "columns": [ + { + "expression": "migration_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"company_skill_repository_snapshots\".\"migration_at\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "company_skill_snapshots_maintenance_idx": { + "name": "company_skill_snapshots_maintenance_idx", + "columns": [ + { + "expression": "maintenance_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"company_skill_repository_snapshots\".\"maintenance_at\" is not null", + "concurrently": false, + "method": "btree", + "with": {} } }, "foreignKeys": { @@ -19999,6 +20162,25 @@ "primaryKey": false, "notNull": false }, + "package_snapshot_id": { + "name": "package_snapshot_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "package_file_map": { + "name": "package_file_map", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "migration_at": { + "name": "migration_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false, + "default": "now()" + }, "revision_number": { "name": "revision_number", "type": "integer", @@ -20057,6 +20239,52 @@ } }, "indexes": { + "company_skill_versions_migration_idx": { + "name": "company_skill_versions_migration_idx", + "columns": [ + { + "expression": "migration_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"company_skill_versions\".\"migration_at\" is not null", + "concurrently": false, + "method": "btree", + "with": {} + }, + "company_skill_versions_package_snapshot_idx": { + "name": "company_skill_versions_package_snapshot_idx", + "columns": [ + { + "expression": "package_snapshot_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, + "company_skill_versions_repository_snapshot_idx": { + "name": "company_skill_versions_repository_snapshot_idx", + "columns": [ + { + "expression": "repository_snapshot_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "company_skill_versions_skill_revision_idx": { "name": "company_skill_versions_skill_revision_idx", "columns": [ @@ -20168,6 +20396,19 @@ "onDelete": "restrict", "onUpdate": "no action" }, + "company_skill_versions_package_snapshot_id_company_skill_repository_snapshots_id_fk": { + "name": "company_skill_versions_package_snapshot_id_company_skill_repository_snapshots_id_fk", + "tableFrom": "company_skill_versions", + "tableTo": "company_skill_repository_snapshots", + "columnsFrom": [ + "package_snapshot_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "restrict", + "onUpdate": "no action" + }, "company_skill_versions_author_agent_id_agents_id_fk": { "name": "company_skill_versions_author_agent_id_agents_id_fk", "tableFrom": "company_skill_versions", @@ -20613,6 +20854,78 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.skill_package_deletions": { + "name": "skill_package_deletions", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "archive_key": { + "name": "archive_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "manifest_key": { + "name": "manifest_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "delete_after": { + "name": "delete_after", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "skill_package_deletions_due_idx": { + "name": "skill_package_deletions_due_idx", + "columns": [ + { + "expression": "delete_after", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.company_transfer_runs": { "name": "company_transfer_runs", "schema": "", @@ -33849,7 +34162,7 @@ } ], "isUnique": false, - "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE", + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_invalid_metadata', 'restore_failed')) IS NOT TRUE", "concurrently": false, "method": "btree", "with": {} @@ -51261,6 +51574,195 @@ "checkConstraints": {}, "isRLSEnabled": false }, + "public.sandbox_creation_operations": { + "name": "sandbox_creation_operations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true + }, + "company_id": { + "name": "company_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "environment_id": { + "name": "environment_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "original_environment_id": { + "name": "original_environment_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "run_id": { + "name": "run_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "request_key": { + "name": "request_key", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'run'" + }, + "issue_id": { + "name": "issue_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "execution_workspace_id": { + "name": "execution_workspace_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "driver_key": { + "name": "driver_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "state": { + "name": "state", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "config": { + "name": "config", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "metadata": { + "name": "metadata", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "lease_id": { + "name": "lease_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "next_attempt_at": { + "name": "next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "sandbox_creation_operations_pending_deadline_idx": { + "name": "sandbox_creation_operations_pending_deadline_idx", + "columns": [ + { + "expression": "next_attempt_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"sandbox_creation_operations\".\"state\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "sandbox_creation_operations_owner_idx": { + "name": "sandbox_creation_operations_owner_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "original_environment_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "request_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "sandbox_creation_operations_environment_id_environments_id_fk": { + "name": "sandbox_creation_operations_environment_id_environments_id_fk", + "tableFrom": "sandbox_creation_operations", + "tableTo": "environments", + "columnsFrom": [ + "environment_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "set null", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, "public.secret_access_events": { "name": "secret_access_events", "schema": "", @@ -52268,6 +52770,12 @@ "primaryKey": false, "notNull": false }, + "next_eval_reason": { + "name": "next_eval_reason", + "type": "text", + "primaryKey": false, + "notNull": false + }, "next_eval_at": { "name": "next_eval_at", "type": "timestamp with time zone", diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 47d9d449f1a..3cc42512d8c 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2920,6 +2920,13 @@ "when": 1793395364937, "tag": "0418_curved_lady_mastermind", "breakpoints": true + }, + { + "idx": 419, + "version": "7", + "when": 1793402408013, + "tag": "0419_moaning_vulture", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/status_cards.ts b/packages/db/src/schema/status_cards.ts index 06efaa0368b..fd18d22fb8b 100644 --- a/packages/db/src/schema/status_cards.ts +++ b/packages/db/src/schema/status_cards.ts @@ -56,6 +56,9 @@ export const statusCards = pgTable( lastModel: text("last_model"), generatingIssueId: uuid("generating_issue_id").references(() => issues.id, { onDelete: "set null" }), failureReason: text("failure_reason"), + // Internal deadline ownership; null denotes a row written before provenance + // was recorded. Membership may be rechecked at startup, retry/policy may not. + nextEvalReason: text("next_eval_reason").$type<"membership" | "policy" | "retry" | "refresh">(), nextEvalAt: timestamp("next_eval_at", { withTimezone: true }), archivedAt: timestamp("archived_at", { withTimezone: true }), archivedByUserId: text("archived_by_user_id"), diff --git a/server/src/__tests__/status-card-evaluation-work.test.ts b/server/src/__tests__/status-card-evaluation-work.test.ts index 4f369cd76e9..4e00e326855 100644 --- a/server/src/__tests__/status-card-evaluation-work.test.ts +++ b/server/src/__tests__/status-card-evaluation-work.test.ts @@ -3,7 +3,7 @@ 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, approvals, budgetIncidents, budgetPolicies, costEvents, companies, companyMemberships, createDb, documents, +import { activityLog, agentWakeupRequests, agents, approvals, budgetIncidents, budgetPolicies, costEvents, companies, companyMemberships, createDb, documents, withDedicatedDbConnection, heartbeatRuns, instanceSettings, issueComments, issues, statusCards, statusCardUpdates, documentRevisions } from "@paperclipai/db"; import { companySearchQuerySchema, statusCardRefreshPolicySchema } from "@paperclipai/shared"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; @@ -188,6 +188,86 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu .toHaveLength(1); }); + it("finishes generation and writeback with a one-connection pool and releases dedicated locks", async () => { + vi.useRealTimers(); + const previousDb = db; + db = createDb(external ?? temporary!.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + try { + const created = await card("reactive", { ...policy(), debounceSeconds: 1 }); + const countLocks = async () => Number((await db.execute<{ count: string }>(sql` + select count(*) from pg_stat_activity where datname = current_database() + and application_name = 'paperclip-workspace-finalization-lock'`))[0]!.count); + const baseline = await countLocks(); + await issueService(db).create(companyId, { title: "Pool-one blocker", status: "blocked", priority: "medium" }); + const service = statusCardEvaluationService(db, { wakeup }); + const acquired = Promise.withResolvers(), release = Promise.withResolvers(); + const holder = withDedicatedDbConnection(db, connection => connection.transaction(async tx => { + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${created.id}, 0))`); + acquired.resolve(); await release.promise; + })); + await acquired.promise; + try { + await service.run(new AbortController().signal); + expect(wakeup).not.toHaveBeenCalled(); + expect(await service.next()).toBeGreaterThan(Date.now()); + } finally { release.resolve(); await holder; } + await service.run(new AbortController().signal); + expect((await statusCardService(db).getById(created.id))!.pendingChangeCount).toBeGreaterThan(0); + await vi.waitFor(async () => { + await service.run(new AbortController().signal); + expect(wakeup).toHaveBeenCalledOnce(); + }, { timeout: 5000, interval: 100 }); + await writeback(created.id, (await statusCardService(db).getById(created.id))!.generatingIssueId!); + await service.run(new AbortController().signal); + expect(await service.next()).toBeNull(); + expect(idleWorkSnapshot().active).toBe(0); + await vi.waitFor(async () => expect(await countLocks()).toBe(baseline)); + // Failure exits also release the physical connection and advisory lock. + await db.update(statusCards).set({ queries: [{ scope: "invalid" } as never], nextEvalAt: new Date() }).where(eq(statusCards.id, created.id)); + await service.run(new AbortController().signal); + expect((await statusCardService(db).getById(created.id))!.nextEvalReason).toBe("retry"); + await vi.waitFor(async () => expect(await countLocks()).toBe(baseline)); + await withDedicatedDbConnection(db, connection => connection.transaction(async tx => { + expect((await tx.execute<{ acquired: boolean }>(sql`select pg_try_advisory_xact_lock(hashtextextended(${created.id}, 0)) as acquired`))[0]!.acquired).toBe(true); + })); + } finally { await db.$client.end({ timeout: 1 }); db = previousDb; } + }, 15_000); + + it.each([false, true])("recovers a lost committed hint before a future membership deadline (legacy=%s)", async legacy => { + await issueService(db).create(companyId, { title: "Original blocker", status: "blocked", priority: "medium" }); + compiledQuery = companySearchQuerySchema.parse({ scope: "issues", status: ["blocked"], updatedWithin: "1d" }); + const created = await card(); + expect(created.nextEvalReason).toBe("membership"); + const oldExpiry = created.nextEvalAt!.getTime(); + // No coordinator exists to receive this committed source hint. A fresh + // service has no in-memory hint and must reconstruct from durable truth. + await issueService(db).create(companyId, { title: "Committed before crash", status: "blocked", priority: "medium" }); + if (legacy) await db.update(statusCards).set({ nextEvalReason: null }).where(eq(statusCards.id, created.id)); + const restarted = statusCardEvaluationService(db, { wakeup }); + await restarted.run(new AbortController().signal); + const pending = (await statusCardService(db).getById(created.id))!; + expect(pending.pendingChangeCount).toBe(1); + expect(pending.nextEvalReason).toBe("policy"); + expect(pending.nextEvalAt!.getTime()).toBeLessThan(oldExpiry); + expect(wakeup).not.toHaveBeenCalled(); + vi.setSystemTime(pending.nextEvalAt!); await restarted.run(new AbortController().signal); + expect(wakeup).toHaveBeenCalledOnce(); + await writeback(created.id, (await statusCardService(db).getById(created.id))!.generatingIssueId!); + expect((await statusCardService(db).getById(created.id))!.nextEvalReason).toBe("membership"); + }); + + it("preserves a legacy ambiguous five-minute claim without extending it on restart", async () => { + const created = await card(); + const retry = new Date(Date.now() + 240_000); + await db.update(statusCards).set({ nextEvalAt: retry, nextEvalReason: null }).where(eq(statusCards.id, created.id)); + await issueService(db).create(companyId, { title: "Unconsumed source", status: "blocked", priority: "medium" }); + const restarted = statusCardEvaluationService(db, { wakeup }); + await restarted.run(new AbortController().signal); + expect(await restarted.next()).toBe(retry.getTime()); + vi.setSystemTime(retry); await restarted.run(new AbortController().signal); + expect((await statusCardService(db).getById(created.id))!.pendingChangeCount).toBeGreaterThan(0); + }); + 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" } } : {}), @@ -335,8 +415,12 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu 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(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect((await statusCardService(db).getById(created.id))!.state).toBe("error"); + }); const failed = (await statusCardService(db).getById(created.id))!; - expect(failed).toMatchObject({ state: "error", generatingIssueId: null, nextEvalAt: null }); + expect(failed).toMatchObject({ state: "error", generatingIssueId: null, nextEvalAt: null, nextEvalReason: 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")!; diff --git a/server/src/__tests__/status-cards.test.ts b/server/src/__tests__/status-cards.test.ts index 0c24b05dd70..05e1bba48e5 100644 --- a/server/src/__tests__/status-cards.test.ts +++ b/server/src/__tests__/status-cards.test.ts @@ -192,6 +192,7 @@ describeEmbeddedPostgres("status card routes", () => { .send({ title: "Launch health", titlePinned: true }); expect(patched.status).toBe(200); expect(patched.body).toMatchObject({ title: "Launch health", titlePinned: true }); + expect(patched.body).not.toHaveProperty("nextEvalReason"); const scheduled = await request(app) .patch(`/api/status-cards/${created.body.id}`) @@ -1152,6 +1153,7 @@ describeEmbeddedPostgres("status card routes", () => { }); expect(summaryWrite.status).toBe(200); expect(summaryWrite.body.card).toMatchObject({ state: "active", queryVersion: 1, generatingIssueId: null }); + expect(summaryWrite.body.card).not.toHaveProperty("nextEvalReason"); expect(summaryWrite.body.document.latestBody).toContain("**Decide:**"); expect(await db.select().from(statusCardUpdates).then((rows) => rows.find((row) => row.kind === "full"))).toMatchObject({ status: "ok", finishedAt: expect.any(Date), inputTokens: 5200, outputTokens: 980 }); diff --git a/server/src/routes/status-cards.ts b/server/src/routes/status-cards.ts index da4f91680bb..d21616b7cf2 100644 --- a/server/src/routes/status-cards.ts +++ b/server/src/routes/status-cards.ts @@ -16,6 +16,12 @@ import { accessService, heartbeatService, instanceSettingsService, issueService, import { queueIssueAssignmentWakeup, type IssueAssignmentWakeupDeps } from "../services/issue-assignment-wakeup.js"; import { assertCompanyAccess, getAccessibleResource, getActorInfo, hasCompanyAccess } from "./authz.js"; +// Deadline provenance belongs to scheduling, not the public card contract. +function publicCard(card: T) { + const { nextEvalReason: _reason, ...result } = card; + return result; +} + export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWakeupDeps } = {}) { const router = Router(); const access = accessService(db); @@ -144,9 +150,9 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake assertCompanyAccess(req, companyId); await assertStatusCardsEnabled(); const query = listStatusCardsQuerySchema.parse(req.query); - res.json(await service.list(companyId, query.archived, { + res.json((await service.list(companyId, query.archived, { issueReadCondition: await issueReadSqlCondition(db, req.actor), - })); + })).map(publicCard)); }); router.post("/companies/:companyId/status-cards", validate(createStatusCardSchema), async (req, res) => { @@ -162,7 +168,7 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake try { const compile = await enqueueCompile(req, card.id); await logMutation(req, companyId, "status_card.created", card.id, { state: card.state }); - res.status(201).json(compile.card); + res.status(201).json(publicCard(compile.card)); } catch (error) { await service.remove(card.id); throw error; @@ -173,9 +179,9 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake await assertStatusCardsEnabled(); const card = await getAccessibleResource(req, res, service.getById(req.params.id as string), "Status card not found"); if (!card) return; - res.json(await service.hydrate(card, { + res.json(publicCard(await service.hydrate(card, { issueReadCondition: await issueReadSqlCondition(db, req.actor), - })); + }))); }); router.patch("/status-cards/:id", validate(patchStatusCardSchema), async (req, res) => { @@ -197,7 +203,7 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake fields: Object.keys(req.body), archived: Boolean(updated.archivedAt), }); - res.json(compile?.card ?? restore?.card ?? updated); + res.json(publicCard(compile?.card ?? restore?.card ?? updated)); }); router.delete("/status-cards/:id", async (req, res) => { @@ -234,7 +240,7 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake generatingIssueId: result.generatingIssue.id, alreadyGenerating: result.alreadyGenerating, }); - res.status(result.alreadyGenerating ? 200 : 202).json(result); + res.status(result.alreadyGenerating ? 200 : 202).json({ ...result, card: publicCard(result.card) }); }); router.post("/status-cards/:id/refresh", validate(refreshStatusCardSchema), async (req, res) => { @@ -249,7 +255,7 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake alreadyGenerating: result.alreadyGenerating, enqueued: result.enqueued, }); - res.status(result.enqueued && !result.alreadyGenerating ? 202 : 200).json(result); + res.status(result.enqueued && !result.alreadyGenerating ? 202 : 200).json({ ...result, card: publicCard(result.card) }); }); router.get("/status-cards/:id/dry-run", async (req, res) => { @@ -289,7 +295,7 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake generationIssueId: req.body.generationIssueId, changeSummary: req.body.changeSummary, }); - res.json(updated); + res.json(publicCard(updated)); }); router.put("/status-cards/:id/summary", validate(writeStatusCardSummarySchema), async (req, res) => { @@ -309,7 +315,7 @@ export function statusCardRoutes(db: Db, opts: { heartbeat?: IssueAssignmentWake documentId: result.document.id, changeSummary: req.body.changeSummary, }); - res.json(result); + res.json({ ...result, card: publicCard(result.card) }); }); return router; diff --git a/server/src/services/status-card-finalization.ts b/server/src/services/status-card-finalization.ts index 5d7ee62b523..702509be8b4 100644 --- a/server/src/services/status-card-finalization.ts +++ b/server/src/services/status-card-finalization.ts @@ -47,6 +47,7 @@ export async function finalizeStatusCardsForStalledGeneration( failureReason, generatingIssueId: null, nextEvalAt: null, + nextEvalReason: null, updatedAt: now, }) .where( diff --git a/server/src/services/status-cards.ts b/server/src/services/status-cards.ts index 19e7ac75140..f9e6372ea2e 100644 --- a/server/src/services/status-cards.ts +++ b/server/src/services/status-cards.ts @@ -1,6 +1,6 @@ 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 { and, desc, eq, gt, gte, inArray, isNotNull, isNull, lte, ne, or, sql, type SQL } from "drizzle-orm"; import { agents, companies, @@ -11,6 +11,7 @@ import { issueComments, statusCards, statusCardUpdates, + withDedicatedDbConnection, type Db, } from "@paperclipai/db"; import type { @@ -301,11 +302,12 @@ export function statusCardService( ...(input.refreshPolicy !== undefined ? { refreshPolicy: input.refreshPolicy, + nextEvalReason: "refresh", nextEvalAt: card.archivedAt ? null : input.refreshPolicy.mode === "reactive" && card.queries.length > 0 ? now : nextStatusCardEvaluationAt(input.refreshPolicy, now), } : {}), ...(archiveChanged && input.archived - ? { archivedAt: now, archivedByAgentId: actor.agentId, archivedByUserId: actor.userId, nextEvalAt: null } + ? { archivedAt: now, archivedByAgentId: actor.agentId, archivedByUserId: actor.userId, nextEvalAt: null, nextEvalReason: null } : {}), ...(archiveChanged && !input.archived ? { @@ -315,6 +317,7 @@ export function statusCardService( lastChangeAt: now, lastUpdateRunKind: null, nextEvalAt: card.queries.length > 0 ? now : null, + nextEvalReason: "refresh", } : {}), }; @@ -537,9 +540,10 @@ export function statusCardService( companyId: string, issueIds: string[], options: StatusCardReadOptions = {}, + queryDb: Db = db, ): Promise { if (issueIds.length === 0) return []; - const rows = await db + const rows = await queryDb .select({ id: issues.id, identifier: issues.identifier, @@ -589,11 +593,11 @@ export function statusCardService( return loadIssueSummaries(card.companyId, card.mentionedIssueIds ?? [], options); } - async function executeQueries(card: StatusCardRow, options: StatusCardReadOptions = {}, onNextMembershipChange?: (at: Date | null) => void) { + async function executeQueries(card: StatusCardRow, options: StatusCardReadOptions = {}, onNextMembershipChange?: (at: Date | null) => void, queryDb: Db = db) { const issueMap = new Map(); for (const storedQuery of card.queries) { const query = companySearchQuerySchema.parse(storedQuery); - const response = await searchSvc.search(card.companyId, query, { + const response = await companySearchService(queryDb).search(card.companyId, query, { issueReadCondition: options.issueReadCondition, onNextMembershipChange, }); for (const result of response.results) { @@ -605,12 +609,12 @@ export function statusCardService( const mentioned = await loadIssueSummaries( card.companyId, (card.mentionedIssueIds ?? []).filter((issueId) => !issueMap.has(issueId)), - options, + options, queryDb, ); for (const issue of mentioned) issueMap.set(issue.id, issue); const snapshot = [...issueMap.values()]; if (snapshot.length === 0) return snapshot; - const latestHumanComments = await db + const latestHumanComments = await queryDb .select({ issueId: issueComments.issueId, // The postgres-js driver returns the `max()` aggregate over a timestamp @@ -672,6 +676,7 @@ export function statusCardService( lastChangeAt: null, state: "active", nextEvalAt, + nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null, }).where(eq(statusCards.id, card.id)).returning()); return { card: next!, generatingIssue: null, alreadyGenerating: false, enqueued: false }; } @@ -699,6 +704,7 @@ export function statusCardService( pendingChangeHash, lastChangeAt, state: decision.action === "pause_budget" ? "paused_budget" : decision.action === "pause_hours" ? "paused_hours" : "active", + nextEvalReason: "policy", nextEvalAt: nextStatusCardPolicyEligibility({ policy: card.refreshPolicy, now, lastChangeAt, completedUpdateTimes: recent.filter(row => row.kind !== "compile" && row.finishedAt).map(row => row.startedAt), @@ -755,6 +761,7 @@ export function statusCardService( lastChangeAt, state: "active", nextEvalAt, + nextEvalReason: nextEvalAt ? card.refreshPolicy.mode === "reactive" ? "membership" : "refresh" : null, failureReason: null, }).where(and(eq(statusCards.id, card.id), isNull(statusCards.archivedAt), or(isNull(statusCards.generatingIssueId), priorGenerationPredicate))).returning()); if (!next) { @@ -788,9 +795,16 @@ export function statusCardService( async function tickDueStatusCards(now = new Date(), reactiveCompanies?: ReadonlySet | "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); + // Startup recovers lost source hints for membership deadlines as well as + // disarmed cards. Saved policy gates and retry leases keep their owner. + // Legacy untagged leases are at most five minutes from their claim time: + // preserve that bounded ambiguity, never a days-away membership deadline. + const startupScope = reconstruct ? or( + isNull(statusCards.nextEvalAt), eq(statusCards.nextEvalReason, "membership"), + and(isNull(statusCards.nextEvalReason), isNull(statusCards.pendingChangeHash), + eq(statusCards.state, "active"), gt(statusCards.nextEvalAt, new Date(now.getTime() + 5 * 60_000))), + ) : undefined; + const reactiveScope = or(startupScope, 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), @@ -805,13 +819,13 @@ export function statusCardService( // 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 => { + await withDedicatedDbConnection(db, connection => connection.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) { 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 }) + const [claimed] = await mutate(candidate.companyId, tx => tx.update(statusCards).set({ nextEvalAt: claimUntil, nextEvalReason: "retry" }) .where(and(eq(statusCards.id, candidate.id), isNull(statusCards.archivedAt), isNull(statusCards.generatingIssueId), candidate.nextEvalAt ? eq(statusCards.nextEvalAt, candidate.nextEvalAt) : isNull(statusCards.nextEvalAt))) .returning()); @@ -826,7 +840,7 @@ export function statusCardService( "status card scheduled refresh failed", ); } - }); + })); } return { evaluated, enqueued, deferredCompanyIds }; } @@ -859,7 +873,9 @@ export function statusCardService( // Current watched membership: compiled-query matches plus the issues this // summary mentions. let membershipAt: Date | null = null; - const watchedSnapshot = await executeQueries({ ...current, mentionedIssueIds }, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; }); + // All watched-source reads use this transaction. Borrowing the root + // pool here would deadlock a one-connection installation. + const watchedSnapshot = await executeQueries({ ...current, mentionedIssueIds }, {}, at => { if (at && (!membershipAt || at < membershipAt)) membershipAt = at; }, tx as unknown as Db); const watchedNow = buildStatusCardFingerprint(watchedSnapshot); let snapshot: StatusCardFingerprint; if (payloadFingerprint) { @@ -931,6 +947,9 @@ export function statusCardService( pendingChangeCount: 0, pendingChangeHash: null, lastChangeAt: null, + nextEvalReason: current.refreshPolicy.mode === "reactive" && + filterStatusCardChanges(diffStatusCardFingerprint(snapshot, watchedNow), current.refreshPolicy).length > 0 + ? "refresh" : membershipAt ? "membership" : null, nextEvalAt: current.refreshPolicy.mode === "reactive" && filterStatusCardChanges(diffStatusCardFingerprint(snapshot, watchedNow), current.refreshPolicy).length > 0 ? now : nextEvaluationAt(current, membershipAt, now),