From bf9eebe5bd3d53049d1bfefee0407ce73612c133 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 00:29:28 -0500 Subject: [PATCH 1/6] Exercise saved legacy input through its eligibility producer Co-Authored-By: Paperclip --- .../2026-10-30-deferred-execution-sleep.md | 20 ++++++++++++ tests/e2e/legacy-failure-continuation.spec.ts | 31 ++++++++++++++----- 2 files changed, 43 insertions(+), 8 deletions(-) diff --git a/doc/plans/2026-10-30-deferred-execution-sleep.md b/doc/plans/2026-10-30-deferred-execution-sleep.md index 3d2cb88ce40..9c90025abe6 100644 --- a/doc/plans/2026-10-30-deferred-execution-sleep.md +++ b/doc/plans/2026-10-30-deferred-execution-sleep.md @@ -32,3 +32,23 @@ Project budget admission locks and re-reads the company-scoped project before pu Native receipt checks share an opaque dispatcher created by the existing queue owner for exactly one bounded worker pass. Its canonical company/agent/issue/review scope map is the same map used by the manual dispatcher; the native worker explicitly admits at most100due entries per pass, matching the existing default batch while retaining company fairness. The next pass creates a fresh map. This restores same-pass grouping without introducing global or cross-invocation deduplication, changing claim authority, or consuming later notices. Each receipt still re-reads its own moved claim/error deadline before dispatch. Actual PostgreSQL controls reconstruct two saved native intents into one dispatch and shared run with real process stdout, cover the100-intent boundary, and keep another issue and company separate. A real issue-row lock holds the first admission while a separate transaction commits a new notice whose requestedAt is explicitly older than admission. The fixed current batch cannot consume that new identity; a fresh pass creates a distinct run, preserving the late notice even though its ordering metadata is old. Once all executions and stored deadlines settle,25virtual minutes produce zero driver SQL. The canonical dispatch body, including parsing, claims, deterministic receipt recovery and error handling, is unchanged apart from moving its map into the owner-created batch closure. + + +### Browser fixture producer boundary + +The legacy recovery browser fixture used to insert a deferred request through a +separate SQL connection after server startup, then wait for the removed poll. +That is neither saved startup history nor a supported in-process producer. The +automatic browser case now seeds while the company is paused, verifies the saved +request and sole failed source run, and reactivates through the ordinary company +API. No Retry or duplicate message is sent. It requires exactly one new run, +one persisted agent response, the original queued comment in the process prompt, +and the response after page reload. Startup reconstruction remains covered by +the PostgreSQL case that persists history before worker registration. + +The corrected automatic path passes three consecutive built-UI/browser runs +(2.1 minutes total). All six legacy continuation browser paths pass together +(1.6 minutes), including task/inbox Retry, Retry after reload, ordinary message, +queued Retry and automatic reactivation. The fixture also reads the configured +external PostgreSQL connection when present, so local proof can reuse a dedicated +test database instead of starting another database process. diff --git a/tests/e2e/legacy-failure-continuation.spec.ts b/tests/e2e/legacy-failure-continuation.spec.ts index ed998017eae..46ce639e5b5 100644 --- a/tests/e2e/legacy-failure-continuation.spec.ts +++ b/tests/e2e/legacy-failure-continuation.spec.ts @@ -11,14 +11,15 @@ async function json(response: APIResponse) { return response.json(); } -for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "message", "queued_retry", "automatic_message"] as const) { +for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "message", "queued_retry", "automatic_message_on_reactivation"] as const) { test(`legacy startup hold: ${action} reaches a new agent response`, async ({ page, request }) => { test.setTimeout(120_000); const root = await mkdtemp(path.join(os.tmpdir(), "legacy-recovery-browser-")); const config = JSON.parse(await readFile(process.env.PAPERCLIP_E2E_SERVER_CONFIG!, "utf8")); // Use the running test server's actual port, including fallback allocation. - const pid = await readFile(path.join(config.database.embeddedPostgresDataDir, "postmaster.pid"), "utf8"); - const url = `postgres://paperclip:paperclip@127.0.0.1:${pid.split("\n")[3]}/paperclip`; + const url = config.database.mode === "postgres" ? config.database.connectionString + : `postgres://paperclip:paperclip@127.0.0.1:${(await readFile( + path.join(config.database.embeddedPostgresDataDir, "postmaster.pid"), "utf8")).split("\n")[3]}/paperclip`; const db = createDb(url); const company = await json(await request.post("/api/companies", { data: { name: `Legacy recovery ${action} ${Date.now()}` } })); try { @@ -37,8 +38,11 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag title: "Continue after startup failure", description: "Answer the pending follow-up once.", status: "backlog", assigneeAgentId: agent.id, } })); + if (action === "automatic_message_on_reactivation") { + await json(await request.patch(`/api/companies/${company.id}`, { data: { status: "paused" } })); + } const sourceRunId = randomUUID(); - // Seed the historical incident, then exercise all recovery through the UI. + // Seed the historical incident, then exercise ordinary recovery entry points. // No adapter.invoke or new dispatch identity exists on this pre-upgrade run. await db.insert(heartbeatRuns).values({ id: sourceRunId, companyId: company.id, agentId: agent.id, status: "failed", runtimeMode: "legacy", processPid: action === "queued_retry" ? process.pid : 999999999, @@ -58,7 +62,7 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag await db.insert(companyMemberships).values({ companyId: company.id, principalType: "user", principalId: "original-board", membershipRole: "operator", status: "active" }); } - if (action === "queued_retry" || action === "automatic_message") { + if (action === "queued_retry" || action === "automatic_message_on_reactivation") { const commentId = randomUUID(); // Reproduce a real user comment saved while the failed run was active, // including queues originally created by a system wake. @@ -104,8 +108,19 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag retry.click(), ]); expect(response.ok(), `${response.status()} ${await response.text()}`).toBe(true); - } else if (action === "automatic_message") { - // No Retry, duplicate message, or status change: the saved input runs. + } else if (action === "automatic_message_on_reactivation") { + // This fixture writes history through a separate SQL connection after boot. + // Reactivation is the supported producer that makes saved work eligible; + // startup reconstruction itself is covered before registration in the PG suite. + const pending = await db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, company.id), eq(agentWakeupRequests.agentId, agent.id), + eq(agentWakeupRequests.status, "deferred_issue_execution"), + )); + expect(pending).toHaveLength(1); + const before = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, company.id)); + expect(before).toHaveLength(1); + await json(await request.patch(`/api/companies/${company.id}`, { data: { status: "active" } })); + // No Retry or duplicate message: the existing saved input runs once. } else if (action === "message") { await page.getByRole("textbox", { name: "editable markdown" }).fill("Please continue the pending follow-up."); await page.getByRole("button", { name: "Send", exact: true }).click(); @@ -133,7 +148,7 @@ for (const action of ["task_retry", "retry_after_reload", "inbox_retry", "messag expect(runs.filter(run => run.id !== sourceRunId)).toHaveLength(1); expect(runs.find(run => run.id === sourceRunId)).toMatchObject({ status: "failed", resultJson: null }); const prompts = await readFile(path.join(root, "prompts"), "utf8"); - if (action === "queued_retry" || action === "automatic_message") { + if (action === "queued_retry" || action === "automatic_message_on_reactivation") { expect(prompts).toContain("Approved"); await expect(page.getByRole("button", { name: "Interrupt", exact: true })).toHaveCount(0); } From 26275cae7f301e8206ce9b3b226f64f7b985123c Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 00:32:36 -0500 Subject: [PATCH 2/6] test: hold lifecycle verification before retry identity assertions Co-Authored-By: Paperclip --- .../__tests__/verification-retry.test.ts | 26 +++++++++++++------ 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts b/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts index cc76c705d32..93cc524613d 100644 --- a/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/verification-retry.test.ts @@ -19,8 +19,8 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu let companyId: string; const workers: Array<{ stop(): Promise }> = []; beforeAll(async () => { - if (external) { await applyPendingMigrations(external); db = createDb(external); } - else { database = await startEmbeddedPostgresTestDatabase("paperclip-verification-retry-"); db = createDb(database.connectionString); } + if (external) { await applyPendingMigrations(external); db = createDb(external, { maxConnections: 6 }); } + else { database = await startEmbeddedPostgresTestDatabase("paperclip-verification-retry-"); db = createDb(database.connectionString, { maxConnections: 6 }); } }, 90_000); afterAll(async () => { if (external) await closeRegisteredClients(external); else await database?.cleanup(); }); beforeEach(async () => { @@ -76,12 +76,22 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu it("permits an explicit retry of the same operation and completes verification before readiness", async () => { const { agent, work, runHost } = await failed(); - runHost.mockResolvedValue("complete"); - await createAgentLifecycle(db).retry(agent.id); - expect((await current(agent.id)).lifecycleOperation).not.toHaveProperty("retryRequired"); - expect((await current(agent.id)).lifecycleOperation).not.toHaveProperty("verificationFailures"); - expect((await current(agent.id)).lifecycleOperation!.id).toBe(agent.lifecycleOperation!.id); - await work.process(agent.id); + let release!: () => void; + const verification = new Promise(resolve => { release = resolve; }); + runHost.mockImplementation(async () => { await verification; return "complete"; }); + let processing: Promise | undefined; + try { + await createAgentLifecycle(db).retry(agent.id); + processing = work.process(agent.id); + const retried = await current(agent.id); + expect(retried).toMatchObject({ lifecycleState: "verifying", status: "paused" }); + expect(retried.lifecycleOperation).not.toHaveProperty("retryRequired"); + expect(retried.lifecycleOperation).not.toHaveProperty("verificationFailures"); + expect(retried.lifecycleOperation!.id).toBe(agent.lifecycleOperation!.id); + } finally { + release(); + await processing; + } expect(await current(agent.id)).toMatchObject({ lifecycleState: "ready", status: "idle", lifecycleError: null }); }); From a53f22861f8dd65a74831c7d9026d3cb6d683b41 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 00:40:30 -0500 Subject: [PATCH 3/6] fix(retention): exclude finalized blockers before graph bounds Share the dependency owner finalization predicate with bounded blocker selection, resolve relation identities before correlated checks, and refuse automatic archives from incomplete fresh graphs. Cover completed history, live evidence and primary-key lookup work. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 6 + .../attention-retention-work.test.ts | 88 ++++++++- server/src/services/attention-retention.ts | 6 +- server/src/services/issues.ts | 173 ++++++------------ 4 files changed, 158 insertions(+), 115 deletions(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index b680241adf7..5a43cb71fef 100644 --- a/doc/plans/2026-10-30-attention-retention-sleep.md +++ b/doc/plans/2026-10-30-attention-retention-sleep.md @@ -50,3 +50,9 @@ Review the source parity fixtures and producer hints together. The intended outc Issue producer invalidation has its own finite discovery budget. Reverse parent/blocker traversal returns at most 101 distinct UUIDs in one recursive query, with each relation adjacency capped at 101 unique edges before materialization. At most 100 affected issues produce one sorted canonical batch hint; an overflow sentinel produces one company hint and the existing complete paged reconstruction. UNION deduplication terminates cycles. A PostgreSQL EXPLAIN ANALYZE on 2,100 reverse edges proves the outer limit stops at 101 CTE rows while the indexed relation branch returns only 100 rows. Repeated identical writes coalesce by canonical key. The shared scheduler is unchanged: a broader company hint does not cancel already pending narrower scopes; those remain finite batches of at most 100 IDs, processed at the existing 25-scope pass limit. This permits redundant bounded reconstruction instead of discarding invalidations or introducing cursor/subsumption policy across other owners. A 151-node cyclic chain proves overflow still registers the far ancestor's exact archive deadline and settles quietly afterward. The ordinary feed retains conservative `needs_attention` entries for roots outside the shared graph budget. The batch finalization visits every input root; a root without explored edges points attention at itself. A real PostgreSQL regression with 2,001 blocked roots verifies the nonempty deferred set, bounded graph size, ordinary issue-list metadata, and all 2,001 attention-feed subject IDs. This is a consumer parity proof; it does not increase graph traversal or change archive eligibility. + +## Completed blocker history and incomplete graph checks + +Explicit blocker queries apply the dependency owner's pending-finalization predicate before the graph row limit. A done blocker with no operation debt is excluded; attributed operations take precedence over legacy unattributed workspace operations, and a later successful export clears the barrier. The readiness helper and bounded attention query share this predicate, including the existing optional workspace/operation locks. Each relation resolves its globally unique issue identity before tenant and eligibility checks, avoiding repeated company-index scans. + +A fresh archive check consumes the private graph completeness result. An unexpanded frontier cannot establish absence of live work, so that check disarms without archiving. Supported issue, relation, run and workspace changes reactivate the root; restart reconstructs it. Completed-only history remains fully classified and can archive normally. No incomplete-root polling is added. diff --git a/server/src/__tests__/attention-retention-work.test.ts b/server/src/__tests__/attention-retention-work.test.ts index ce24ac1081b..c65e4f13487 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -35,15 +35,22 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; const errors: unknown[] = []; const statements: string[] = []; const parameterCounts: number[] = []; + let captureBlockerPlan = false; + let blockerPlanQuery: { query: string; parameters: unknown[] } | undefined; beforeAll(async () => { if (externalUrl) await applyPendingMigrations(externalUrl); else temporary = await startEmbeddedPostgresTestDatabase("attention-retention-"); db = createDb(externalUrl ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); db.$client.options.onnotice = () => {}; - db.$client.options.debug = (_connection, query, parameters) => { statements.push(query); parameterCounts.push(parameters.length); }; + db.$client.options.debug = (_connection, query, parameters) => { + statements.push(query); parameterCounts.push(parameters.length); + if (captureBlockerPlan && !blockerPlanQuery && query.includes("latest_blocker_operation") && query.includes("issue_relations")) + blockerPlanQuery = { query, parameters: [...parameters] }; + }; }, 30_000); beforeEach(async () => { companyId = randomUUID(); agentId = randomUUID(); draining = false; errors.length = 0; + captureBlockerPlan = false; blockerPlanQuery = undefined; await db.insert(companies).values({ id: companyId, name: "Retention", issuePrefix: `A${companyId.slice(0, 6)}` }); await db.insert(authUsers).values({ id: "board", name: "Board", email: "board@retention.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "board", membershipRole: "owner", status: "active" }); @@ -428,6 +435,85 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); }); + it.each(["running", "run", "confirmation", "completed"] as const)("filters completed blocker history before bounded %s attention and archive checks", async kind => { + const root = await insertIssue("blocked"), workspaceId = randomUUID(); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalized shared history', 'shared', 'project_primary')`); + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at) + values (${companyId}::uuid, ${workspaceId}::uuid, 'workspace_finalize', 'succeeded', ${old().toISOString()}::timestamptz)`); + await db.execute(sql`with history as ( + insert into issues (id, company_id, title, status, updated_at, execution_workspace_id) + select ('00000000-0000-4000-8000-' || lpad(to_hex(n), 12, '0'))::uuid, + ${companyId}::uuid, 'completed-blocker-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid + from generate_series(1, 2100) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`); + if (kind !== "completed") { + const live = await insertIssue(kind === "running" ? "in_progress" : "todo", { id: "ffffffff-ffff-4fff-8fff-ffffffffffff", assigneeAgentId: agentId }); + await db.insert(issueRelations).values({ companyId, issueId: live.id, relatedIssueId: root.id, type: "blocks" }); + if (kind === "run") { + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, status: "running" }).returning(); + await db.update(issues).set({ executionRunId: run.id }).where(eq(issues.id, live.id)); + } else if (kind === "confirmation") await db.insert(issueThreadInteractions).values({ companyId, issueId: live.id, + kind: "request_confirmation", status: "pending", payload: { prompt: "Still waiting" } }); + } + statements.length = 0; parameterCounts.length = 0; + captureBlockerPlan = kind === "completed"; + const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]); + captureBlockerPlan = false; + if (kind === "completed") { + expect(blockerPlanQuery).toBeTruthy(); + const explain = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + blockerPlanQuery!.query, + blockerPlanQuery!.parameters as never[]); + type Plan = { "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number; "Rows Removed by Filter"?: number; Plans?: Plan[] }; + const rootPlan = (explain[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!.Plan; + const visit = (node: Plan): void => { + if (node["Relation Name"] === "issues" && node["Actual Loops"] > 1) { + expect(node["Actual Rows"]).toBeLessThanOrEqual(1); + expect(node["Rows Removed by Filter"] ?? 0).toBe(0); + } + node.Plans?.forEach(visit); + }; + visit(rootPlan); + process.stdout.write(JSON.stringify({ completedBlockerHistoryPlan: explain }) + "\n"); + } + expect(batch.nodeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES); + expect(batch.edgeCount).toBeLessThanOrEqual(BLOCKER_ATTENTION_MAX_NODES * 2); + const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" }); + const entry = page.rows!.find(row => row.rootId === root.id && row.variant === "terminal"); + if (kind !== "completed") { + // Exercise the authoritative fresh check even if discovery correctly omits the live root. + await service.check(entry ?? { id: `blocker_attention:${root.id}:terminal`, companyId, sourceKind: "blocker_attention", + sourceId: root.id, rootId: root.id, issueId: root.id, variant: "terminal", activityAt: old().toISOString(), deadlineAt: Date.now(), dueAt: Date.now() }); + expect(await state("blocker_attention", root.id)).toBeNull(); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + } else { + expect(entry).toBeTruthy(); await service.check(entry!); + expect((await state("blocker_attention", root.id))?.archivedAt).toBeInstanceOf(Date); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(1); + } + expect(batch.fetchedRows).toBeLessThanOrEqual(4); + expect(batch.deferredRootIds.size).toBe(0); + expect(Boolean(batch.attention.get(root.id)?.blockingTreeLive)).toBe(kind !== "completed"); + expect(parameterCounts.length).toBeLessThanOrEqual(100); + }); + + it("does not archive an unknown truncated blocker tree during its fresh check", async () => { + const root = await insertIssue("blocked"); + await db.execute(sql`with blockers as ( + insert into issues (company_id, title, status, updated_at) + select ${companyId}::uuid, 'unresolved-' || n, 'backlog', ${old().toISOString()}::timestamptz + from generate_series(1, 2100) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from blockers`); + const service = attentionRetentionService(db), page = await service.load({ companyId, sourceKind: "blocker_attention" }); + const entry = page.rows!.find(row => row.rootId === root.id && row.variant === "terminal")!; + expect(entry).toBeTruthy(); expect(await service.check(entry)).toBeNull(); + expect(await db.select().from(decisionRetention)).toHaveLength(0); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(0); + await start(); await quiet(); + }); + it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => { await db.execute(sql`insert into issues (id, company_id, title, status, updated_at) select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz diff --git a/server/src/services/attention-retention.ts b/server/src/services/attention-retention.ts index a0bca4281f5..ddb74d682e5 100644 --- a/server/src/services/attention-retention.ts +++ b/server/src/services/attention-retention.ts @@ -235,7 +235,11 @@ export function attentionRetentionService(db: Db) { where company_id = ${candidate.companyId}::uuid and id = ${row.id}::uuid`) as unknown as Array<{ at: string; due: string }>)[0]; activityAt = activity.at; dueAt = Number(activity.due); } else { - const attention = evidence ? evidence.blockers.get(row.id) : (await issueService(database).listBlockerAttention(candidate.companyId, [row])).get(row.id); + const fresh = evidence ? undefined : await issueService(database).listBlockerAttentionBatch(candidate.companyId, [row]); + // A bounded graph cannot establish absence of live work beyond its frontier. + // Keep unknown roots visible; a later source change can revalidate them. + if (fresh?.deferredRootIds.has(row.id)) return null; + const attention = evidence ? evidence.blockers.get(row.id) : fresh!.attention.get(row.id); if (!terminalBlockerRequiresAttention(attention)) return null; sourceId = attention.terminalBlockerIssueId ?? row.id; if (lock && sourceId !== candidate.sourceId) await database.execute(sql`select id from issues diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 3ded30015d8..a1c28f296c2 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -2518,124 +2518,53 @@ export async function listUnfinalizedExecutionWorkspaceIds( return unfinalized; } +/** The blocker keeps its own attributed operation barrier; legacy unattributed + * work is a fallback, and a later successful workspace export clears either. */ +function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, workspaceId: SQL) { + return sql`${workspaceId} is not null and exists ( + select 1 from lateral ( + select phase, status, started_at from workspace_operations blocker_operation + where blocker_operation.company_id = ${companyId}::uuid + and blocker_operation.execution_workspace_id = ${workspaceId} + and (blocker_operation.issue_id = ${issueId} or blocker_operation.issue_id is null) + order by (blocker_operation.issue_id is not null) desc, blocker_operation.started_at desc + limit 1 + ) latest_blocker_operation + where not (latest_blocker_operation.phase = 'workspace_finalize' and latest_blocker_operation.status = 'succeeded') + and not exists (select 1 from workspace_operations later_finalize + where later_finalize.company_id = ${companyId}::uuid + and later_finalize.execution_workspace_id = ${workspaceId} + and later_finalize.phase = 'workspace_finalize' and later_finalize.status = 'succeeded' + and later_finalize.started_at > latest_blocker_operation.started_at) + )`; +} + async function listPendingFinalizeBlockerIssueIds( dbOrTx: Pick, companyId: string, - blockerWorkspacePairs: Array<{ - blockerIssueId: string; - executionWorkspaceId: string; - }>, + blockerWorkspacePairs: Array<{ blockerIssueId: string; executionWorkspaceId: string }>, lockSources = false, ): Promise> { const pending = new Set(); - const blockerIssueIds = [ - ...new Set(blockerWorkspacePairs.map((pair) => pair.blockerIssueId)), - ]; - const executionWorkspaceIds = [ - ...new Set(blockerWorkspacePairs.map((pair) => pair.executionWorkspaceId)), - ]; - if (blockerIssueIds.length === 0 || executionWorkspaceIds.length === 0) - return pending; - const blockerWorkspaceKeys = new Set( - blockerWorkspacePairs.map( - (pair) => `${pair.blockerIssueId}:${pair.executionWorkspaceId}`, - ), - ); - - const rows: Array<{ issueId: string | null; executionWorkspaceId: string | null; - phase: string; status: string; startedAt: Date }> = []; - for (const chunk of chunkList(executionWorkspaceIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { - if (lockSources) { - - + const pairs = [...new Map(blockerWorkspacePairs.map(pair => [`${pair.blockerIssueId}:${pair.executionWorkspaceId}`, pair])).values()]; + if (lockSources) { + const workspaceIds = [...new Set(pairs.map(pair => pair.executionWorkspaceId))]; + for (const chunk of chunkList(workspaceIds, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces) .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, chunk))) .for("update", { noWait: true }); - } - const query = dbOrTx - .select({ - issueId: workspaceOperations.issueId, - executionWorkspaceId: workspaceOperations.executionWorkspaceId, - phase: workspaceOperations.phase, - status: workspaceOperations.status, - startedAt: workspaceOperations.startedAt, - }) - .from(workspaceOperations) - .where( - and( - eq(workspaceOperations.companyId, companyId), - inArray( - workspaceOperations.executionWorkspaceId, - chunk, - ), - ), - ); - - rows.push(...await (lockSources ? query.for("share", { noWait: true }) : query)); - } - - const latestAttributedByBlockerWorkspace = new Map< - string, - { phase: string; status: string; startedAt: Date } - >(); - const latestUnattributedByWorkspace = new Map< - string, - { phase: string; status: string; startedAt: Date } - >(); - const latestSuccessfulFinalizeByWorkspace = new Map(); - for (const row of rows) { - if (!row.executionWorkspaceId) continue; - if (row.phase === "workspace_finalize" && row.status === "succeeded") { - const current = latestSuccessfulFinalizeByWorkspace.get( - row.executionWorkspaceId, - ); - if (!current || row.startedAt > current) { - latestSuccessfulFinalizeByWorkspace.set( - row.executionWorkspaceId, - row.startedAt, - ); - } - } - if (row.issueId) { - const key = `${row.issueId}:${row.executionWorkspaceId}`; - if (!blockerWorkspaceKeys.has(key)) continue; - const current = latestAttributedByBlockerWorkspace.get(key); - if (!current || row.startedAt > current.startedAt) { - latestAttributedByBlockerWorkspace.set(key, { - phase: row.phase, - status: row.status, - startedAt: row.startedAt, - }); - } - continue; - } - - const current = latestUnattributedByWorkspace.get(row.executionWorkspaceId); - if (!current || row.startedAt > current.startedAt) { - latestUnattributedByWorkspace.set(row.executionWorkspaceId, { - phase: row.phase, - status: row.status, - startedAt: row.startedAt, - }); + await dbOrTx.select({ id: workspaceOperations.id }).from(workspaceOperations) + .where(and(eq(workspaceOperations.companyId, companyId), inArray(workspaceOperations.executionWorkspaceId, chunk))) + .for("share", { noWait: true }); } } - - for (const pair of blockerWorkspacePairs) { - const latest = - latestAttributedByBlockerWorkspace.get( - `${pair.blockerIssueId}:${pair.executionWorkspaceId}`, - ) ?? latestUnattributedByWorkspace.get(pair.executionWorkspaceId); - if (!latest) continue; // no ops recorded -> nothing to finalize for this blocker - if (latest.phase === "workspace_finalize" && latest.status === "succeeded") - continue; - const laterSuccessfulFinalize = latestSuccessfulFinalizeByWorkspace.get( - pair.executionWorkspaceId, - ); - if (laterSuccessfulFinalize && laterSuccessfulFinalize > latest.startedAt) - continue; - pending.add(pair.blockerIssueId); + for (const chunk of chunkList(pairs, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + const values = sql.join(chunk.map(pair => sql`(${pair.blockerIssueId}::uuid, ${pair.executionWorkspaceId}::uuid)`), sql`, `); + const rows = await dbOrTx.select({ id: sql`blocker_source.issue_id` }) + .from(sql`(values ${values}) as blocker_source(issue_id, workspace_id)`) + .where(pendingFinalizeBlockerCondition(companyId, sql`blocker_source.issue_id`, sql`blocker_source.workspace_id`)); + for (const row of rows) pending.add(row.id); } - return pending; } @@ -3883,12 +3812,8 @@ async function readIssueBlockerAttentionBatch( BLOCKER_ATTENTION_MAX_NODES * 2 - edgeCount); if (remaining <= 0) break; const rowLimit = remaining + 1; - const explicitBlockerRowsPromise: Promise< - IssueBlockerAttentionQueryRow[] - > = dbOrTx + const explicitBlocker = dbOrTx .select({ - issueId: issueRelations.relatedIssueId, - blockerIssueId: issues.id, id: issues.id, companyId: issues.companyId, parentId: issues.parentId, @@ -3900,14 +3825,36 @@ async function readIssueBlockerAttentionBatch( assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId, }) + // Resolve the globally unique relation target first. The company guard + // below remains authoritative, without a repeated company-index scan. + .from(issues).where(eq(issues.id, issueRelations.issueId)) + .limit(1).as("explicit_blocker"); + const explicitBlockerRowsPromise: Promise< + IssueBlockerAttentionQueryRow[] + > = dbOrTx + .select({ + issueId: issueRelations.relatedIssueId, + blockerIssueId: explicitBlocker.id, + id: explicitBlocker.id, + companyId: explicitBlocker.companyId, + parentId: explicitBlocker.parentId, + identifier: explicitBlocker.identifier, + title: explicitBlocker.title, + status: explicitBlocker.status, + executionRunId: explicitBlocker.executionRunId, + executionWorkspaceId: explicitBlocker.executionWorkspaceId, + assigneeAgentId: explicitBlocker.assigneeAgentId, + assigneeUserId: explicitBlocker.assigneeUserId, + }) .from(issueRelations) - .innerJoin(issues, eq(issueRelations.issueId, issues.id)) + .innerJoinLateral(explicitBlocker, sql`true`) .where( and( eq(issueRelations.companyId, companyId), eq(issueRelations.type, "blocks"), inArray(issueRelations.relatedIssueId, chunk), - eq(issues.companyId, companyId), + eq(explicitBlocker.companyId, companyId), + or(ne(explicitBlocker.status, "done"), pendingFinalizeBlockerCondition(companyId, sql`${explicitBlocker.id}`, sql`${explicitBlocker.executionWorkspaceId}`)), ), ).limit(rowLimit); const childRowsPromise: Promise = dbOrTx From b448320a5e5683956e83d5400cfc3a9c933fcc18 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 01:56:02 -0500 Subject: [PATCH 4/6] Validate eligibility before first attention retention materialization Preserve existing retention controls and eligible unseen review Keep while rejecting speculative issue and agent attention sources. Qualify canonical terminal-leaf feed materialization and company isolation with PostgreSQL controls. Co-Authored-By: Paperclip --- ...retention-materialization-security.test.ts | 134 ++++++++++++++++++ server/src/services/decision-retention.ts | 49 ++++++- 2 files changed, 180 insertions(+), 3 deletions(-) create mode 100644 server/src/__tests__/decision-retention-materialization-security.test.ts diff --git a/server/src/__tests__/decision-retention-materialization-security.test.ts b/server/src/__tests__/decision-retention-materialization-security.test.ts new file mode 100644 index 00000000000..b9b8e442f42 --- /dev/null +++ b/server/src/__tests__/decision-retention-materialization-security.test.ts @@ -0,0 +1,134 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { agents, authUsers, companies, companyMemberships, createDb, decisionRetention, issueRelations, issues } from "@paperclipai/db"; +import type { AttentionSourceKind } from "@paperclipai/shared"; +import { attentionService } from "../services/attention.js"; +import { decisionRetentionService } from "../services/decision-retention.js"; +import { issueService } from "../services/issues.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RETENTION_MATERIALIZATION_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("retention first-materialization security", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let companyId: string, agentId: string, userId: string; + const users: string[] = []; + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("retention-security-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.onnotice = () => {}; + }, 30_000); + beforeEach(async () => { + companyId = randomUUID(); agentId = randomUUID(); userId = randomUUID(); users.push(userId); + await db.insert(authUsers).values({ id: userId, name: "Security operator", email: `${userId}@retention.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companies).values({ id: companyId, name: "Materialization", issuePrefix: `S${companyId.slice(0, 6)}`, defaultResponsibleUserId: userId }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "owner", status: "active" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Source agent", status: "idle", adapterType: "process" }); + }); + afterEach(async () => { await db.execute(sql`truncate companies cascade`); }); + afterAll(async () => { if (db) { await db.delete(authUsers).where(inArray(authUsers.id, users)); await db.$client.end(); } await temporary?.cleanup(); }); + const mutation = (sourceKind: AttentionSourceKind, sourceId: string) => ({ companyId, sourceKind, sourceId, + authActor: { type: "board" as const, source: "local_implicit" as const, userId, companyIds: [companyId], isInstanceAdmin: true }, + actor: { actorType: "user" as const, actorId: userId, userId, agentId: null, runId: null, agentApiKeyId: null, responsibleUserId: userId } }); + async function issue(values: Partial = {}) { + return (await db.insert(issues).values({ companyId, title: "Security source", status: "todo", createdByAgentId: agentId, ...values }).returning())[0]!; + } + async function visible(sourceKind: AttentionSourceKind, sourceId: string) { + const feed = await attentionService(db).list(companyId, { all: true, allowUnscopedAll: true, userId, excludeSources: ["approval", "join_request"] }); + return feed.items.some(item => item.sourceKind === sourceKind && item.subject.id === sourceId); + } + + it.each(["review", "blocker_attention"] as const)("cannot pre-archive a future %s before an ordinary transition makes it visible", async kind => { + const row = await issue({ assigneeUserId: userId }); + let rejected: unknown; + try { await decisionRetentionService(db).archive(mutation(kind, row.id)); } catch (error) { rejected = error; } + await issueService(db).update(row.id, kind === "review" + ? { status: "in_review", actorUserId: userId } + : { status: "blocked", unblockDescriptor: { owner: "board", action: "Choose the next owner" }, actorUserId: userId }); + expect(await visible(kind, row.id)).toBe(true); + expect(rejected).toMatchObject({ status: 404 }); + expect((await decisionRetentionService(db).getState(companyId, kind, row.id))?.archivedAt).toBeNull(); + }); + + it.each(["review", "blocker_attention", "productivity_review"] as const)("does not materialize Keep for an ineligible %s", async kind => { + const row = await issue(); + await expect(decisionRetentionService(db).setKeep({ ...mutation(kind, row.id), keep: true })).rejects.toMatchObject({ status: 404 }); + expect(await decisionRetentionService(db).getState(companyId, kind, row.id)).toBeNull(); + }); + + it("allows Keep on an unseen eligible review and preserves existing manual actions after it leaves review", async () => { + const row = await issue({ status: "in_review", assigneeUserId: userId }); + const service = decisionRetentionService(db), input = mutation("review", row.id); + expect(await service.getState(companyId, "review", row.id)).toBeNull(); + expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true); + await issueService(db).update(row.id, { status: "todo", actorUserId: userId }); + expect((await service.archive(input)).archivedReason).toBe("manual"); + expect((await service.setKeep({ ...input, keep: false })).keep).toBe(false); + expect((await service.revive(input)).archivedAt).toBeNull(); + }); + + it("allows unseen current board-owned blocker attention", async () => { + const row = await issue({ status: "blocked", unblockDescriptor: { owner: "board", action: "Choose next owner" }, blockedTransitionAt: new Date() }); + expect((await decisionRetentionService(db).setKeep({ ...mutation("blocker_attention", row.id), keep: true })).keep).toBe(true); + }); + + it("requires current agent error status for a new alert but preserves existing row actions", async () => { + const service = decisionRetentionService(db), input = mutation("agent_error_alert", agentId); + await expect(service.archive(input)).rejects.toMatchObject({ status: 404 }); + expect(await service.getState(companyId, "agent_error_alert", agentId)).toBeNull(); + await db.update(agents).set({ status: "error" }).where(eq(agents.id, agentId)); + expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true); + await db.update(agents).set({ status: "idle" }).where(eq(agents.id, agentId)); + expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date); + expect((await service.revive(input)).archivedAt).toBeNull(); + }); + + it("preserves historical productivity and terminal-leaf rows without inventing a new source", async () => { + const row = await issue(); const service = decisionRetentionService(db); + await service.syncItems(companyId, ["productivity_review", "blocker_attention"].map(sourceKind => ({ sourceKind: sourceKind as AttentionSourceKind, subject: { id: row.id }, activityAt: row.updatedAt.toISOString() }))); + for (const kind of ["productivity_review", "blocker_attention"] as const) { + const input = mutation(kind, row.id); + expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date); + expect((await service.revive(input)).archivedAt).toBeNull(); + expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true); + } + }); + + it("requires ordinary feed materialization for an unseen TODO terminal leaf, then preserves its manual controls", async () => { + const root = await issue({ status: "blocked" }); + const leaf = await issue({ status: "todo" }); + await db.insert(issueRelations).values({ companyId, issueId: leaf.id, relatedIssueId: root.id, type: "blocks" }); + const service = decisionRetentionService(db), input = mutation("blocker_attention", leaf.id); + await expect(service.setKeep({ ...input, keep: true })).rejects.toMatchObject({ status: 404 }); + expect(await service.getState(companyId, "blocker_attention", leaf.id)).toBeNull(); + const attention = await issueService(db).listBlockerAttention(companyId, [root]); + expect(attention.get(root.id)).toMatchObject({ terminalBlockerIssueId: leaf.id, state: "needs_attention", blockingTreeLive: false }); + expect(await visible("blocker_attention", leaf.id)).toBe(true); + expect((await service.archive(input)).archivedAt).toBeInstanceOf(Date); + expect((await service.revive(input)).archivedAt).toBeNull(); + expect((await service.setKeep({ ...input, keep: true })).keep).toBe(true); + }); + + it.each(["review", "blocker_attention"] as const)("does not pre-materialize hidden %s", async kind => { + const row = await issue({ hiddenAt: new Date(), status: kind === "review" ? "in_review" : "blocked", assigneeUserId: userId, + unblockDescriptor: { owner: "board", action: "Choose next owner" }, blockedTransitionAt: new Date() }); + await expect(decisionRetentionService(db).archive(mutation(kind, row.id))).rejects.toMatchObject({ status: 404 }); + expect(await decisionRetentionService(db).getState(companyId, kind, row.id)).toBeNull(); + }); + + it.each(["review", "agent_error_alert"] as const)("keeps company and agent authorization isolated for %s", async kind => { + const foreign = randomUUID(); + await db.insert(companies).values({ id: foreign, name: "Foreign", issuePrefix: "FOREIGN" }); + const sourceId = kind === "review" ? (await issue({ companyId: foreign, status: "in_review", assigneeUserId: userId })).id + : (await db.insert(agents).values({ companyId: foreign, name: "Foreign error", status: "error" }).returning())[0]!.id; + const service = decisionRetentionService(db); + await expect(service.archive(mutation(kind, sourceId))).rejects.toMatchObject({ status: 404 }); + const input = { ...mutation(kind, sourceId), companyId: foreign, authActor: { type: "agent" as const, agentId, companyId }, + actor: { actorType: "agent" as const, actorId: agentId, userId: null, agentId, runId: null, agentApiKeyId: null, responsibleUserId: null } }; + await expect(service.archive(input)).rejects.toMatchObject({ status: 404 }); + expect(await db.select().from(decisionRetention).where(and(eq(decisionRetention.companyId, foreign), eq(decisionRetention.sourceId, sourceId)))).toHaveLength(0); + }); +}); diff --git a/server/src/services/decision-retention.ts b/server/src/services/decision-retention.ts index b5ca6c9f1fb..6e23398c7da 100644 --- a/server/src/services/decision-retention.ts +++ b/server/src/services/decision-retention.ts @@ -1,4 +1,8 @@ import { isProspectiveBlockedTransition } from "./routable-blocked.js"; +import { requiresReviewAttention, terminalBlockerRequiresAttention, ATTENTION_AGENT_ERROR_STATUS } from "./attention-source-eligibility.js"; +import { attentionIssueCondition } from "./issue-visibility.js"; +import { issueService } from "./issues.js"; +import { parseIssueExecutionState } from "./issue-execution-policy.js"; import { notifyRetentionWork } from "./retention-work-notifications.js"; import { decisionArchiveDeliveryService, type ArchiveNotificationBatch } from "./decision-archive-delivery.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; @@ -184,17 +188,56 @@ export function decisionRetentionService( )).then((rows) => rows[0] ?? null); } - async function materializeSourceState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string }) { + async function materializeSourceState(database: Db, input: { companyId: string; sourceKind: AttentionSourceKind; sourceId: string; actor: Pick }) { const tables: Record = { approval: "approvals", join_request: "join_requests", decision: "decisions", issue_thread_interaction: "issue_thread_interactions", recovery_action: "issue_recovery_actions", failed_run: "heartbeat_runs", productivity_review: "issues", blocker_attention: "issues", review: "issues", budget_alert: "budget_incidents", agent_error_alert: "agents", }; - if (input.sourceKind === "blocker_attention") { + const existingState = () => database.select({ id: decisionRetention.id }).from(decisionRetention).where(and( + eq(decisionRetention.companyId, input.companyId), eq(decisionRetention.sourceKind, input.sourceKind), + eq(decisionRetention.sourceId, input.sourceId), + )).then(rows => rows[0]); + if (input.sourceKind === "agent_error_alert") { + const [source] = await database.select().from(agents).where(and(eq(agents.companyId, input.companyId), eq(agents.id, input.sourceId))).for("update"); + if (!source || (!await existingState() && source.status !== ATTENTION_AGENT_ERROR_STATUS)) throw notFound("Attention source not found"); + await decisionRetentionService(database).syncItems(input.companyId, [{ + sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: source.updatedAt.toISOString(), + }]); + return; + } + if (["review", "blocker_attention", "productivity_review"].includes(input.sourceKind)) { const [source] = await database.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.sourceId))).for("update"); if (!source) throw notFound("Attention source not found"); - const activityAt = source.unblockDescriptor?.owner === "board" && isProspectiveBlockedTransition(source) + if (!await existingState()) { + const [visible] = await database.select({ id: issues.id }).from(issues).where(and( + eq(issues.companyId, input.companyId), eq(issues.id, source.id), attentionIssueCondition(false), + )); + let eligible = false; + if (visible && input.sourceKind === "review" && source.status === "in_review") { + const [pendingApproval] = await database.select({ id: approvals.id }).from(issueApprovals) + .innerJoin(approvals, eq(approvals.id, issueApprovals.approvalId)).where(and( + eq(issueApprovals.companyId, input.companyId), eq(approvals.companyId, input.companyId), + eq(issueApprovals.issueId, source.id), eq(approvals.status, "pending"), + )).limit(1); + const state = parseIssueExecutionState(source.executionState); + const attention = await issueService(database).listReviewAttention(input.companyId, [source]); + eligible = requiresReviewAttention({ + hasHumanParticipant: state?.status === "pending" && state.currentParticipant?.type === "user", + hasUserAssignee: Boolean(source.assigneeUserId), hasPendingApproval: Boolean(pendingApproval), + stalled: attention.get(source.id)?.state === "stalled", + }); + } else if (visible && input.sourceKind === "blocker_attention" && source.status === "blocked") { + const owner = source.unblockDescriptor?.owner; + const ownsDescriptor = owner === "board" || (owner && "userId" in owner && owner.userId === input.actor.userId); + const attention = (await issueService(database).listBlockerAttention(input.companyId, [source])).get(source.id); + eligible = Boolean(ownsDescriptor && isProspectiveBlockedTransition(source)) + || (terminalBlockerRequiresAttention(attention) && (attention.terminalBlockerIssueId ?? source.id) === source.id); + } + if (!eligible) throw notFound("Attention source not found"); + } + const activityAt = input.sourceKind === "blocker_attention" && source.unblockDescriptor?.owner === "board" && isProspectiveBlockedTransition(source) ? source.blockedTransitionAt : source.updatedAt; await decisionRetentionService(database).syncItems(input.companyId, [{ sourceKind: input.sourceKind, subject: { id: input.sourceId }, activityAt: activityAt.toISOString(), From 57f914747c0d2aefaca9ceccdeeddb55489dad67 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 02:00:29 -0500 Subject: [PATCH 5/6] Bound blocker workspace history reads by candidate workspace Share finalized and legacy operation facts across blocker pairs, retaining attributed operation precedence and strict later-finalize semantics. Qualify retained and unrelated history using actual query plans and structural work limits. Co-Authored-By: Paperclip --- ...cker-workspace-history-performance.test.ts | 103 ++++++++++++++++++ server/src/services/issues.ts | 75 ++++++++++--- 2 files changed, 164 insertions(+), 14 deletions(-) create mode 100644 server/src/__tests__/blocker-workspace-history-performance.test.ts diff --git a/server/src/__tests__/blocker-workspace-history-performance.test.ts b/server/src/__tests__/blocker-workspace-history-performance.test.ts new file mode 100644 index 00000000000..c40a4b46f1e --- /dev/null +++ b/server/src/__tests__/blocker-workspace-history-performance.test.ts @@ -0,0 +1,103 @@ +import { randomUUID } from "node:crypto"; +import { sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { applyPendingMigrations, companies, createDb, issues } from "@paperclipai/db"; +import { issueService } from "../services/issues.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_BLOCKER_HISTORY_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +type Plan = { "Node Type": string; "Relation Name"?: string; "Actual Rows": number; "Actual Loops": number; + "Rows Removed by Filter"?: number; "Rows Removed by Index Recheck"?: number; Plans?: Plan[] }; + +describe.skipIf(!support.supported)("blocker workspace retained history cost", () => { + let temporary: Awaited> | undefined; + let db: ReturnType; + let capture = false; + const queries: Array<{ query: string; parameters: unknown[] }> = []; + beforeAll(async () => { + if (external) await applyPendingMigrations(external); + else temporary = await startEmbeddedPostgresTestDatabase("blocker-history-"); + db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + db.$client.options.onnotice = () => {}; + db.$client.options.debug = (_connection, query, parameters) => { + if (capture && query.includes("workspace_operations")) queries.push({ query, parameters: [...parameters] }); + }; + }, 30_000); + afterEach(async () => { capture = false; queries.length = 0; await db.execute(sql`truncate companies cascade`); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); + + it.each(["finalized", "pending"] as const)("bounds operation-history work for 2100 blockers sharing 5000 %s operations", async kind => { + const companyId = randomUUID(), workspaceId = randomUUID(), unrelatedWorkspaceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Shared history", issuePrefix: `SH${companyId.slice(0, 6)}` }); + const [root] = await db.insert(issues).values({ companyId, title: "History dependent", status: "blocked" }).returning(); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Shared operation history', 'shared', 'project_primary'), + (${unrelatedWorkspaceId}::uuid, ${companyId}::uuid, 'Unrelated history', 'shared', 'project_primary')`); + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at) + select ${companyId}::uuid, ${unrelatedWorkspaceId}::uuid, 'worktree_prepare', 'succeeded', + '2025-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 50000) n`); + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, phase, status, started_at) + select ${companyId}::uuid, ${workspaceId}::uuid, + case when n = 5000 and ${kind} = 'finalized' then 'workspace_finalize' else 'worktree_prepare' end, + 'succeeded', '2026-01-01T00:00:00Z'::timestamptz + n * interval '1 second' from generate_series(1, 5000) n`); + await db.execute(sql`with history as ( + insert into issues (company_id, title, status, execution_workspace_id) + select ${companyId}::uuid, 'done blocker ' || n, 'done', ${workspaceId}::uuid + from generate_series(1, 2100) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, id, ${root.id}::uuid, 'blocks' from history`); + await db.execute(sql`analyze workspace_operations`); + await db.execute(sql`analyze issues`); + await db.execute(sql`analyze issue_relations`); + capture = true; + const batch = await issueService(db).listBlockerAttentionBatch(companyId, [root]); + capture = false; + expect(queries.length).toBeGreaterThan(0); + expect(Math.max(...queries.map(statement => statement.parameters.length))).toBeLessThanOrEqual(1100); + let operationRowsExamined = 0; + const measurements: unknown[] = []; + for (const statement of queries) { + const result = await db.$client.unsafe("EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) " + statement.query, + statement.parameters as never[]); + const explain = (result[0]!["QUERY PLAN"] as Array<{ Plan: Plan }>)[0]!; + const visit = (node: Plan) => { + if (node["Relation Name"] === "workspace_operations") operationRowsExamined += node["Actual Loops"] + * (node["Actual Rows"] + (node["Rows Removed by Filter"] ?? 0) + (node["Rows Removed by Index Recheck"] ?? 0)); + node.Plans?.forEach(visit); + }; + visit(explain.Plan); + measurements.push({ ...statement, explain }); + } + process.stdout.write(JSON.stringify({ sharedWorkspaceHistory: { kind, blockers: 2100, retainedOperations: 5000, unrelatedOperations: 50000, + operationRowsExamined, queryCount: queries.length, measurements } }) + "\n"); + // The finite graph/pair pages may each read the 5,000 candidate rows once. + // Neither 2,100 blockers nor 50,000 unrelated operations may multiply that work. + expect(operationRowsExamined).toBeLessThanOrEqual(55_000); + expect(batch.deferredRootIds.has(root.id)).toBe(kind === "pending"); + if (kind === "finalized") expect(batch.nodeCount).toBe(1); + }, 120_000); + it("preserves attributed precedence and the strict later-finalize boundary", async () => { + const companyId = randomUUID(), workspaceId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Finalize policy", issuePrefix: `FP${companyId.slice(0, 6)}` }); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Policy workspace', 'shared', 'project_primary')`); + const roots = await db.insert(issues).values(["older", "equal", "fallback"].map(title => ({ companyId, title, status: "blocked" }))).returning(); + const blockers = await db.insert(issues).values(roots.map(root => ({ companyId, title: root.title + " blocker", status: "done", executionWorkspaceId: workspaceId }))).returning(); + for (let index = 0; index < roots.length; index += 1) { + await db.execute(sql`insert into issue_relations (company_id, issue_id, related_issue_id, type) + values (${companyId}::uuid, ${blockers[index]!.id}::uuid, ${roots[index]!.id}::uuid, 'blocks')`); + } + await db.execute(sql`insert into workspace_operations (company_id, execution_workspace_id, issue_id, phase, status, started_at) values + (${companyId}::uuid, ${workspaceId}::uuid, ${blockers[0]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:01Z'), + (${companyId}::uuid, ${workspaceId}::uuid, ${blockers[1]!.id}::uuid, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:02Z'), + (${companyId}::uuid, ${workspaceId}::uuid, null, 'workspace_finalize', 'succeeded', '2026-01-01T00:00:02Z'), + (${companyId}::uuid, ${workspaceId}::uuid, null, 'worktree_prepare', 'succeeded', '2026-01-01T00:00:03Z')`); + const service = issueService(db); + expect((await service.listBlockerAttentionBatch(companyId, [roots[0]!])).nodeCount).toBe(1); + expect((await service.listBlockerAttentionBatch(companyId, [roots[1]!])).nodeCount).toBe(2); + expect((await service.listBlockerAttentionBatch(companyId, [roots[2]!])).nodeCount).toBe(2); + }); + +}); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index a1c28f296c2..c7b9f8aebb2 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -2518,24 +2518,55 @@ export async function listUnfinalizedExecutionWorkspaceIds( return unfinalized; } -/** The blocker keeps its own attributed operation barrier; legacy unattributed - * work is a fallback, and a later successful workspace export clears either. */ +/** Workspace-wide facts are shared by blockers, rather than rereading retained + * operation history for every issue pointing at the same workspace. The ordered + * OFFSET boundary keeps the rare finalize filter from turning into a company + * history scan; the existing workspace/started-at index can stream candidates. */ +function blockerWorkspaceFacts(companyId: string, workspaceScope: SQL) { + return sql`( + with candidate_workspaces as materialized ( + select distinct workspace_id from (${workspaceScope}) scoped_workspaces(workspace_id) + ), workspace_facts as materialized ( + select candidate.workspace_id, finalized.started_at as finalized_at, + legacy.phase, legacy.status, legacy.started_at + from candidate_workspaces candidate + left join lateral ( + select started_at from ( + select phase, status, started_at from workspace_operations + where company_id = ${companyId}::uuid + and execution_workspace_id = candidate.workspace_id + order by started_at desc offset 0 + ) workspace_history + where phase = 'workspace_finalize' and status = 'succeeded' + order by started_at desc limit 1 + ) finalized on true + left join lateral ( + select phase, status, started_at from workspace_operations + where company_id = ${companyId}::uuid + and execution_workspace_id = candidate.workspace_id and issue_id is null + order by started_at desc limit 1 + ) legacy on true + ) select * from workspace_facts + ) as blocker_workspace_facts`; +} + +/** An issue's own operation takes precedence over the legacy NULL fallback. + * Only a strictly later successful finalize clears either barrier. */ function pendingFinalizeBlockerCondition(companyId: string, issueId: SQL, workspaceId: SQL) { return sql`${workspaceId} is not null and exists ( - select 1 from lateral ( + select 1 from (values (1)) seed(value) + left join lateral ( select phase, status, started_at from workspace_operations blocker_operation where blocker_operation.company_id = ${companyId}::uuid and blocker_operation.execution_workspace_id = ${workspaceId} - and (blocker_operation.issue_id = ${issueId} or blocker_operation.issue_id is null) - order by (blocker_operation.issue_id is not null) desc, blocker_operation.started_at desc - limit 1 - ) latest_blocker_operation - where not (latest_blocker_operation.phase = 'workspace_finalize' and latest_blocker_operation.status = 'succeeded') - and not exists (select 1 from workspace_operations later_finalize - where later_finalize.company_id = ${companyId}::uuid - and later_finalize.execution_workspace_id = ${workspaceId} - and later_finalize.phase = 'workspace_finalize' and later_finalize.status = 'succeeded' - and later_finalize.started_at > latest_blocker_operation.started_at) + and blocker_operation.issue_id = ${issueId} + order by blocker_operation.started_at desc limit 1 + ) latest_blocker_operation on true + where coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at) is not null + and not (coalesce(latest_blocker_operation.phase, blocker_workspace_facts.phase) = 'workspace_finalize' + and coalesce(latest_blocker_operation.status, blocker_workspace_facts.status) = 'succeeded') + and (blocker_workspace_facts.finalized_at is null + or blocker_workspace_facts.finalized_at <= coalesce(latest_blocker_operation.started_at, blocker_workspace_facts.started_at)) )`; } @@ -2558,10 +2589,13 @@ async function listPendingFinalizeBlockerIssueIds( .for("share", { noWait: true }); } } - for (const chunk of chunkList(pairs, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE)) { + // Each pair occurs in both the source and the workspace scope. + for (const chunk of chunkList(pairs, ISSUE_LIST_RELATED_QUERY_CHUNK_SIZE / 2)) { const values = sql.join(chunk.map(pair => sql`(${pair.blockerIssueId}::uuid, ${pair.executionWorkspaceId}::uuid)`), sql`, `); const rows = await dbOrTx.select({ id: sql`blocker_source.issue_id` }) .from(sql`(values ${values}) as blocker_source(issue_id, workspace_id)`) + .leftJoin(blockerWorkspaceFacts(companyId, sql`select workspace_id from (values ${values}) workspace_scope(issue_id, workspace_id)`), + sql`blocker_workspace_facts.workspace_id = blocker_source.workspace_id`) .where(pendingFinalizeBlockerCondition(companyId, sql`blocker_source.issue_id`, sql`blocker_source.workspace_id`)); for (const row of rows) pending.add(row.id); } @@ -3829,6 +3863,17 @@ async function readIssueBlockerAttentionBatch( // below remains authoritative, without a repeated company-index scan. .from(issues).where(eq(issues.id, issueRelations.issueId)) .limit(1).as("explicit_blocker"); + const workspaceScope = sql`select distinct candidate.execution_workspace_id + from issue_relations workspace_relation + inner join lateral ( + select company_id, status, execution_workspace_id from issues + where id = workspace_relation.issue_id limit 1 + ) candidate on true + where workspace_relation.company_id = ${companyId}::uuid + and workspace_relation.type = 'blocks' + and workspace_relation.related_issue_id in (${sql.join(chunk.map(id => sql`${id}::uuid`), sql`, `)}) + and candidate.company_id = ${companyId}::uuid + and candidate.status = 'done' and candidate.execution_workspace_id is not null`; const explicitBlockerRowsPromise: Promise< IssueBlockerAttentionQueryRow[] > = dbOrTx @@ -3848,6 +3893,8 @@ async function readIssueBlockerAttentionBatch( }) .from(issueRelations) .innerJoinLateral(explicitBlocker, sql`true`) + .leftJoin(blockerWorkspaceFacts(companyId, workspaceScope), + sql`blocker_workspace_facts.workspace_id = ${explicitBlocker.executionWorkspaceId}`) .where( and( eq(issueRelations.companyId, companyId), From a0e2a3515d420270b30cd5e89fbb966ef641b8a9 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sat, 31 Oct 2026 02:00:39 -0500 Subject: [PATCH 6/6] fix(retention): wake blocker retention after no-run workspace export Register finalization intent before state writes, preserving caller transactions. Shared workspace export invalidates only the company blocker source; known issue-only export stays scoped and unattached operations do not activate retention. Add actual recorder controls and the shared-root integration journey. Co-Authored-By: Paperclip --- .../2026-10-30-attention-retention-sleep.md | 2 + .../attention-retention-work.test.ts | 86 +++++++++++++++++++ server/src/services/workspace-operations.ts | 9 +- 3 files changed, 96 insertions(+), 1 deletion(-) diff --git a/doc/plans/2026-10-30-attention-retention-sleep.md b/doc/plans/2026-10-30-attention-retention-sleep.md index 5a43cb71fef..6ec95d22125 100644 --- a/doc/plans/2026-10-30-attention-retention-sleep.md +++ b/doc/plans/2026-10-30-attention-retention-sleep.md @@ -56,3 +56,5 @@ The ordinary feed retains conservative `needs_attention` entries for roots outsi Explicit blocker queries apply the dependency owner's pending-finalization predicate before the graph row limit. A done blocker with no operation debt is excluded; attributed operations take precedence over legacy unattributed workspace operations, and a later successful export clears the barrier. The readiness helper and bounded attention query share this predicate, including the existing optional workspace/operation locks. Each relation resolves its globally unique issue identity before tenant and eligibility checks, avoiding repeated company-index scans. A fresh archive check consumes the private graph completeness result. An unexpanded frontier cannot establish absence of live work, so that check disarms without archiving. Supported issue, relation, run and workspace changes reactivate the root; restart reconstructs it. Completed-only history remains fully classified and can archive normally. No incomplete-root polling is added. + +The ordinary workspace-operation recorder also signals no-run finalization in its owning transaction. Export with a workspace identity invalidates that company's blocker source, including every root sharing the exported workspace; only a known issue without a workspace uses the existing reverse-issue hint. This rare source change uses bounded company/source paging and does not rebuild unrelated retention kinds. Run-backed finalization retains its existing native-status fanout. This lets a parked incomplete tree reactivate when export clears its done blockers, without an incomplete-root retry timer. Recorder rollback leaves the operation barrier and attention state unchanged. diff --git a/server/src/__tests__/attention-retention-work.test.ts b/server/src/__tests__/attention-retention-work.test.ts index c65e4f13487..b722c75f395 100644 --- a/server/src/__tests__/attention-retention-work.test.ts +++ b/server/src/__tests__/attention-retention-work.test.ts @@ -6,6 +6,8 @@ import { subscribeDatabaseWork, authUsers, companyMemberships, agents, approvals decisionRetention, decisions, costEvents, heartbeatRunEvents, heartbeatRuns, issueApprovals, issueRecoveryActions, issueRelations, issueThreadInteractions, issues, type Db } from "@paperclipai/db"; import { attentionRetentionService, registerAttentionRetentionWork } from "../services/attention-retention.js"; +import { workspaceOperationService } from "../services/workspace-operations.js"; +import * as workspaceOperationLogs from "../services/workspace-operation-log-store.js"; import { budgetService } from "../services/budgets.js"; import { writeExecutionControl } from "../services/execution-control-notifications.js"; import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js"; @@ -514,6 +516,90 @@ const delay = DEFAULT_DECISION_ARCHIVE_DAYS * 86_400_000; await start(); await quiet(); }); + function simulateRecorderLogStorage() { + // The recorder lifecycle and SQL are real; only its log-storage boundary is simulated. + vi.spyOn(workspaceOperationLogs, "getWorkspaceOperationLogStore").mockReturnValue({ + begin: async () => ({ store: "local_file", logRef: "retention-finalize-proof" }), + append: async () => {}, finalize: async () => ({ bytes: 0, compressed: false }), read: async () => ({ content: "" }), + }); + } + + it.each(["issue", "workspace"] as const)("publishes retention intent from ordinary no-run %s finalization", async scope => { + simulateRecorderLogStorage(); + const source = await insertIssue(), workspaceId = randomUUID(); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Finalize producer', 'shared', 'project_primary')`); + await start(); await quiet(); + const committedHints: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.attentionRetention, (event, hints) => { + if (event === "settled") committedHints.push(...hints ?? []); + }); + try { + const result = await workspaceOperationService(db).createRecorder({ companyId, + executionWorkspaceId: scope === "workspace" ? workspaceId : undefined, + issueId: scope === "issue" ? source.id : undefined, + }).recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) }); + expect(result.status).toBe("succeeded"); + expect(committedHints.length).toBeGreaterThan(0); + expect(committedHints.map(hint => JSON.parse(hint).companyId).every(id => id === companyId)).toBe(true); + await quiet(); + if (scope === "workspace") { + committedHints.length = 0; + await workspaceOperationService(db).createRecorder({ companyId }) + .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) }); + expect(committedHints).toEqual([]); await quiet(); + } + } finally { unsubscribe(); } + }); + + describe("no-run finalization with an incomplete blocker graph", () => { + let rootIds: string[], workspaceId: string, subjectId: string; + let loadedScopes: Array["load"]>[0]>; + beforeEach(async () => { + const began = performance.now(); + simulateRecorderLogStorage(); + const roots = await Promise.all([insertIssue("blocked"), insertIssue("blocked")]); + rootIds = roots.map(root => root.id); workspaceId = randomUUID(); + await db.execute(sql`insert into execution_workspaces (id, company_id, name, mode, strategy_type) + values (${workspaceId}::uuid, ${companyId}::uuid, 'Pending shared export', 'shared', 'project_primary')`); + await workspaceOperationService(db).createRecorder({ companyId, executionWorkspaceId: workspaceId }) + .recordOperation({ phase: "workspace_provision", run: async () => ({ status: "succeeded" }) }); + await db.execute(sql`with history as ( + insert into issues (company_id, title, status, updated_at, execution_workspace_id) + select ${companyId}::uuid, 'waiting-for-export-' || n, 'done', ${old().toISOString()}::timestamptz, ${workspaceId}::uuid + from generate_series(1, 2100) n returning id) + insert into issue_relations (company_id, issue_id, related_issue_id, type) + select ${companyId}::uuid, history.id, root.id, 'blocks' from history + cross join (values (${rootIds[0]}::uuid), (${rootIds[1]}::uuid)) root(id)`); + const [subject] = await db.select({ id: issues.id }).from(issues) + .where(and(eq(issues.companyId, companyId), eq(issues.status, "done"))).limit(1); + subjectId = subject!.id; + const review = await insertIssue("in_review", { assigneeUserId: "board" }); + await decisionRetentionService(db).setKeep({ ...mutation("review", review.id), keep: true }); + const service = attentionRetentionService(db), load = service.load.bind(service); loadedScopes = []; + vi.spyOn(service, "load").mockImplementation((scope, signal) => { loadedScopes.push(scope); return load(scope, signal); }); + process.stdout.write(JSON.stringify({ incompleteFinalizePhase: "seeded", setupMs: performance.now() - began }) + "\n"); + await start(service); await quiet(); + for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull(); + process.stdout.write(JSON.stringify({ incompleteFinalizeFixtureRows: 2100, incompleteRoots: 2, setupMs: performance.now() - began }) + "\n"); + }, 30_000); + it("archives only after the ordinary workspace-only finalize commits, then settles quiet", async () => { + const finish = (database: Db) => workspaceOperationService(database).createRecorder({ companyId, executionWorkspaceId: workspaceId, issueId: subjectId }) + .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) }); + await expect(db.transaction(async tx => { + await finish(tx as unknown as Db); throw new Error("rollback finalization"); + })).rejects.toThrow("rollback finalization"); + await settle(); for (const rootId of rootIds) expect(await state("blocker_attention", rootId)).toBeNull(); + loadedScopes.length = 0; + expect((await finish(db)).status).toBe("succeeded"); + for (const rootId of rootIds) await archived("blocker_attention", rootId); + await quiet(); + expect(loadedScopes.length).toBeGreaterThan(0); + expect(loadedScopes.every(scope => scope?.companyId === companyId && scope.sourceKind === "blocker_attention")).toBe(true); + expect(await db.select().from(decisionArchiveNotificationOutbox)).toHaveLength(2); + }); + }); + it("disarms finite deferred-root checks when large retained graphs have no archive obligation", async () => { await db.execute(sql`insert into issues (id, company_id, title, status, updated_at) select gen_random_uuid(), ${companyId}::uuid, 'held-root-' || n, 'blocked', ${old().toISOString()}::timestamptz diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index 7bd7e2ab658..67ff166b8d0 100644 --- a/server/src/services/workspace-operations.ts +++ b/server/src/services/workspace-operations.ts @@ -1,3 +1,4 @@ +import { notifyAttentionRetentionWork, notifyIssueAttentionRetentionWork } from "./retention-work-notifications.js"; import { notifyNativeResultCommit } from "./native-result-commit-notifications.js"; import { notifyNativeWorkspaceFinalization } from "./native-workspace-finalization-notifications.js"; import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; @@ -480,8 +481,14 @@ export function workspaceOperationService(db: Db) { async recordOperation(recordInput) { async function writeState(write: (database: Db) => Promise): Promise { - if (recordInput.phase !== "workspace_finalize" || !input.heartbeatRunId) return write(db); + if (recordInput.phase !== "workspace_finalize" || (!input.heartbeatRunId && !executionWorkspaceId && !input.issueId)) return write(db); return db.transaction(async tx => { + if (!input.heartbeatRunId) { + // An export clears barriers for every issue sharing this workspace. + if (executionWorkspaceId) await notifyAttentionRetentionWork(tx, { companyId: input.companyId, sourceKind: "blocker_attention" }); + else await notifyIssueAttentionRetentionWork(tx, { companyId: input.companyId, issueId: input.issueId ?? undefined }); + return write(tx as unknown as Db); + } await notifyNativeResultCommit(tx, { companyId: input.companyId, runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined }); await notifyNativeWorkspaceFinalization(tx, { companyId: input.companyId,