diff --git a/.changeset/fn-9037-recommendation-create-perf.md b/.changeset/fn-9037-recommendation-create-perf.md new file mode 100644 index 0000000000..8db147a9d4 --- /dev/null +++ b/.changeset/fn-9037-recommendation-create-perf.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Creating a task from an Insights recommendation is no longer slow on large boards. +category: performance +dev: Adds indexed findTaskByProposalClaimId and listTasksBySourceLineage reads, removes near-duplicate fullRows hydration, and registers migration 0059. diff --git a/docs/storage.md b/docs/storage.md index 0006ca92d5..5d40d17070 100644 --- a/docs/storage.md +++ b/docs/storage.md @@ -841,3 +841,7 @@ Revision listing defaults to 100 rows and clamps `limit` to 1–500. The API acc Project-scoped structured recall records for durable decisions, preferences, and solutions. The table uses the composite `(project_id, id)` key, row-level security, created-at indexes, and a named `(project_id, kind, content_hash)` exact-hash backstop. `graph_node_ids` stores graph cross-references; Memory Keeper merges new IDs under a per-record advisory transaction lock, so identifiers only grow and an unchanged union does not update the row. - Knowledge-graph artifact: `/.fusion-knowledge/graph/` (`nodes.json`, `edges.json`, and `manifest.json`). This is deliberately outside ignored `.fusion` and may be committed at the operator's discretion. + +### Bounded task-intake lookups + +Recommendation proposal claims use the indexed `findTaskByProposalClaimId` read (`uqTasksProjectProposalClaimId`), and same-agent intake reads only matching source lineage (`idxTasksProjectSourceAgentId` and `idxTasksSourceParentTaskId`). Do not replace either read with a `listTasks()` scan. Workflow terminal flags for intake duplicate checks are derived from workflow definitions, not board rows. Guarded-intake near-duplicate checks must remain bounded to their candidates (the fallback is `limit: 50`) and must not hydrate the full board. diff --git a/packages/core/src/__tests__/intake-duplicate-terminal-lanes.test.ts b/packages/core/src/__tests__/intake-duplicate-terminal-lanes.test.ts index 510cd806ef..6d5fcaded7 100644 --- a/packages/core/src/__tests__/intake-duplicate-terminal-lanes.test.ts +++ b/packages/core/src/__tests__/intake-duplicate-terminal-lanes.test.ts @@ -93,8 +93,7 @@ describe("the intake path actually forwards the resolved flags", () => { const source = readFileSync(new URL("../task-store/task-creation.ts", import.meta.url), "utf8"); expect(source).toContain("columnFlagsByColumnId: await resolveIntakeDuplicateColumnFlags(store, allCandidates)"); - // Scoped to distinct columns, not one IR read per candidate row. - expect(source).toContain("if (seenColumns.has(candidate.column)) continue;"); + expect(source).toContain("resolveTaskLifecycleColumns(store, candidate.id, irCache)"); }); it("mirrors the auto-archive into the row using the resolved archived lane", () => { diff --git a/packages/core/src/__tests__/postgres/proposal-claim-lookup.pg.test.ts b/packages/core/src/__tests__/postgres/proposal-claim-lookup.pg.test.ts new file mode 100644 index 0000000000..a691c5790c --- /dev/null +++ b/packages/core/src/__tests__/postgres/proposal-claim-lookup.pg.test.ts @@ -0,0 +1,106 @@ +/* +FNXC:TaskRecommendations 2026-08-13-22:39: +Recommendation replay must use the indexed project-scoped claim lookup rather than a board scan. +These PostgreSQL assertions cover the actual Drizzle predicate, including forensic tombstones and +project isolation that a route mock cannot prove. +*/ +import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest"; +import { eq } from "drizzle-orm"; +import { + createSharedPgTaskStoreTestHarness, + pgDescribe, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; +import type { AsyncDataLayer } from "../../postgres/data-layer.js"; +import * as schema from "../../postgres/schema/index.js"; +import { TaskStore } from "../../store.js"; +import { insertTaskRow, softDeleteTaskRow } from "../../task-store/async/async-persistence.js"; + +pgDescribe("findTaskByProposalClaimId PostgreSQL persistence", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_proposal_claim_lookup", + }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + async function createClaimedTask( + id: string, + claim: string, + overrides: Record = {}, + ) { + return h.store().createTaskWithReservedId({ + id, + description: `Claim holder ${id}`, + column: "todo", + proposalClaimId: claim, + sourceMetadata: { fileScope: ["packages/core/src/store.ts"] }, + ...overrides, + } as never, { + taskId: id, + createdAt: "2026-08-14T00:00:00.000Z", + updatedAt: "2026-08-14T00:00:00.000Z", + applyDefaultWorkflowSteps: false, + }); + } + + it("returns persisted live rows, rejects blank or unknown claims, and keeps archived lanes live", async () => { + const store = h.store(); + await createClaimedTask("FN-CLAIM", "recommendation:parent:one"); + await h.layer().db.update(schema.project.tasks).set({ + sourceMetadata: { fileScope: ["packages/core/src/store.ts"] }, + }).where(eq(schema.project.tasks.id, "FN-CLAIM")); + await createClaimedTask("FN-ARCHIVED-LANE", "recommendation:parent:archived", { column: "archived" }); + + await expect(store.findTaskByProposalClaimId("recommendation:parent:one")).resolves.toMatchObject({ + id: "FN-CLAIM", + column: "todo", + sourceMetadata: { fileScope: ["packages/core/src/store.ts"] }, + }); + await expect(store.findTaskByProposalClaimId("missing")).resolves.toBeNull(); + await expect(store.findTaskByProposalClaimId(" ")).resolves.toBeNull(); + await expect(store.findTaskByProposalClaimId("recommendation:parent:archived")).resolves.toMatchObject({ + id: "FN-ARCHIVED-LANE", + column: "archived", + }); + }); + + it("hides soft-deleted claim holders by default and returns them only for forensic replay", async () => { + const store = h.store(); + await createClaimedTask("FN-DELETED", "recommendation:parent:deleted"); + await softDeleteTaskRow(h.layer(), "FN-DELETED", "2026-08-14T01:00:00.000Z"); + + await expect(store.findTaskByProposalClaimId("recommendation:parent:deleted")).resolves.toBeNull(); + await expect(store.findTaskByProposalClaimId("recommendation:parent:deleted", { includeDeleted: true })) + .resolves.toMatchObject({ id: "FN-DELETED", deletedAt: "2026-08-14T01:00:00.000Z" }); + }); + + it("is project-scoped and never consults cold archive snapshots", async () => { + const layerFor = (projectId: string): AsyncDataLayer => ({ ...h.layer(), projectId }); + const storeA = new TaskStore(h.rootDir(), undefined, { asyncLayer: layerFor("project-a") }); + const storeB = new TaskStore(h.rootDir(), undefined, { asyncLayer: layerFor("project-b") }); + const claim = "recommendation:shared:claim"; + const row = (id: string) => ({ + id, + description: id, + column: "todo", + currentStep: 0, + createdAt: "2026-08-14T00:00:00.000Z", + updatedAt: "2026-08-14T00:00:00.000Z", + proposalClaimId: claim, + }); + + await insertTaskRow(layerFor("project-a"), row("FN-PROJECT-A"), { lineageId: "lineage-a" }); + await expect(storeB.findTaskByProposalClaimId(claim)).resolves.toBeNull(); + await expect(storeA.findTaskByProposalClaimId(claim)).resolves.toMatchObject({ id: "FN-PROJECT-A" }); + + const archived = await createClaimedTask("FN-COLD", "recommendation:cold:snapshot"); + await h.store().archiveTask(archived.id, { cleanup: false }); + // A cold-only snapshot has no live row for the indexed reader to match. + await h.adminDb().delete(schema.project.tasks).where(eq(schema.project.tasks.id, archived.id)); + await expect(h.store().findTaskByProposalClaimId("recommendation:cold:snapshot", { includeDeleted: true })) + .resolves.toBeNull(); + }); +}); diff --git a/packages/core/src/__tests__/same-agent-duplicate-intake.test.ts b/packages/core/src/__tests__/same-agent-duplicate-intake.test.ts index 4a03d95b93..477da2c72c 100644 --- a/packages/core/src/__tests__/same-agent-duplicate-intake.test.ts +++ b/packages/core/src/__tests__/same-agent-duplicate-intake.test.ts @@ -39,6 +39,8 @@ function createStore(overrides: Record = {}) { taskCache: new Map(), getSettings: vi.fn().mockResolvedValue({ autoArchiveDuplicateTasksEnabled: false, tombstoneStickyWindowDays: 7 }), listTasks: vi.fn().mockResolvedValue([]), + listTasksBySourceLineage: vi.fn().mockResolvedValue([]), + listWorkflowDefinitions: vi.fn().mockResolvedValue([]), logEntry: vi.fn().mockResolvedValue(undefined), recordActivity: vi.fn().mockResolvedValue(undefined), updateTask: vi.fn().mockResolvedValue(undefined), @@ -61,6 +63,7 @@ describe("same-agent duplicate intake policy (FN-8401)", () => { await resolveSameAgentDuplicateIntake(store as any, noProvenance as any, noProvenance as any); expect(store.listTasks).not.toHaveBeenCalled(); + expect(store.listTasksBySourceLineage).not.toHaveBeenCalled(); expect(store.moveTask).not.toHaveBeenCalled(); expect(store.updateTask).not.toHaveBeenCalled(); }); @@ -68,7 +71,7 @@ describe("same-agent duplicate intake policy (FN-8401)", () => { it("flags the new live duplicate in place and never deletes its sibling by default", async () => { const sibling = task("FN-SIBLING", { createdAt: new Date(Date.now() - 60_000).toISOString() }); const created = task("FN-NEW"); - const store = createStore({ listTasks: vi.fn().mockResolvedValue([created, sibling]) }); + const store = createStore({ listTasksBySourceLineage: vi.fn().mockResolvedValue([created, sibling]) }); /* FNXC:SameAgentDuplicateIntake 2026-07-19-16:33: @@ -77,6 +80,9 @@ describe("same-agent duplicate intake policy (FN-8401)", () => { */ await _maybeAutoArchiveSameAgentDuplicateBackendImpl(store as any, created as any, created as any); + // Pre-fix performed a board scan; provenance creates must use the narrow lineage read. + expect(store.listTasks).not.toHaveBeenCalled(); + expect(store.listTasksBySourceLineage).toHaveBeenCalledWith({ sourceAgentId: "agent-intake", sourceParentTaskId: null }); expect(store.updateTask).toHaveBeenCalledWith("FN-NEW", { sourceMetadataPatch: expect.objectContaining({ nearDuplicateOf: "FN-SIBLING" }), }); @@ -94,7 +100,7 @@ describe("same-agent duplicate intake policy (FN-8401)", () => { const created = task("FN-NEW"); const store = createStore({ getSettings: vi.fn().mockResolvedValue({ autoArchiveDuplicateTasksEnabled: true, tombstoneStickyWindowDays: 7 }), - listTasks: vi.fn().mockResolvedValue([created, sibling]), + listTasksBySourceLineage: vi.fn().mockResolvedValue([created, sibling]), }); await resolveSameAgentDuplicateIntake(store as any, created as any, created as any); @@ -109,7 +115,7 @@ describe("same-agent duplicate intake policy (FN-8401)", () => { const deletedAt = new Date(Date.now() - 60_000).toISOString(); const tombstone = task("FN-TOMBSTONE", { deletedAt, allowResurrection: false }); const created = task("FN-NEW"); - const store = createStore({ backendMode: true, listTasks: vi.fn().mockResolvedValue([created, tombstone]) }); + const store = createStore({ backendMode: true, listTasksBySourceLineage: vi.fn().mockResolvedValue([created, tombstone]) }); await expect(resolveSameAgentDuplicateIntake(store as any, created as any, created as any)) .rejects.toBeInstanceOf(TombstonedTaskResurrectionError); @@ -119,7 +125,7 @@ describe("same-agent duplicate intake policy (FN-8401)", () => { Soft deletes move to `archived`; sticky tombstones require both flags so same-agent recreation is rejected on every persistence backend. */ - expect(store.listTasks).toHaveBeenCalledWith({ slim: true, includeArchived: true, includeDeleted: true }); + expect(store.listTasksBySourceLineage).toHaveBeenCalledWith({ sourceAgentId: "agent-intake", sourceParentTaskId: null }); expect(recordRunAuditEventAsync).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ taskId: "FN-NEW", mutationType: "intake:resurrection-blocked", })); diff --git a/packages/core/src/postgres/migrations/0059_fn_9037_tasks_source_agent_index.sql b/packages/core/src/postgres/migrations/0059_fn_9037_tasks_source_agent_index.sql new file mode 100644 index 0000000000..37761bfb0b --- /dev/null +++ b/packages/core/src/postgres/migrations/0059_fn_9037_tasks_source_agent_index.sql @@ -0,0 +1,3 @@ +CREATE INDEX IF NOT EXISTS "idxTasksProjectSourceAgentId" + ON project.tasks USING btree ("project_id", "source_agent_id") + WHERE "source_agent_id" IS NOT NULL; diff --git a/packages/core/src/postgres/schema-applier.ts b/packages/core/src/postgres/schema-applier.ts index abe3244cfc..cb37eaff24 100644 --- a/packages/core/src/postgres/schema-applier.ts +++ b/packages/core/src/postgres/schema-applier.ts @@ -63,7 +63,8 @@ capacity-model table drop that landed while this PR was open. /* FNXC:SpecLockMissionAlignment 2026-08-10-16:17: advance the schema ceiling so SQLite and PostgreSQL feature projections retain reconciled drift alignment. */ /* FNXC:MultiProjectIsolation 2026-08-11-10:25: schema startup must register project-local agent ratings before bound stores scope their mutations. */ /* FNXC:MessageArchive 2026-08-12-22:14: 0058 persists non-destructive mailbox archival on upgrades. */ -export const SCHEMA_BASELINE_VERSION = "0058"; +/* FNXC:TaskRecommendations 2026-08-13-22:23: upgrades must install the source-agent index before duplicate intake queries it. */ +export const SCHEMA_BASELINE_VERSION = "0059"; /** FNXC:SymbolLock 2026-07-20-10:00: upgrades need durable task declarations before admission resolves symbols. */ export const TASK_DECLARED_SYMBOLS_VERSION = "0028"; const INITIAL_SCHEMA_VERSION = "0000"; @@ -222,6 +223,7 @@ export const PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION = "0056"; export const PROJECT_OWNERSHIP_DEFAULT_RECONCILIATION_VERSION = "0057"; /** FNXC:MessageArchive 2026-08-12-22:14: explicit registration prevents the archived-message migration from being skipped. */ export const MESSAGE_ARCHIVE_SCHEMA_VERSION = "0058"; +export const TASK_SOURCE_AGENT_INDEX_VERSION = "0059"; /** SECURITY DEFINER helper that only inserts LEGACY_ADOPTION_DRAINED_MARKER. */ export const LEGACY_ADOPTION_DRAINED_MARKER_FUNCTION = "fusion_mark_legacy_adoption_drained"; @@ -454,6 +456,7 @@ const AGENT_RATINGS_PROJECT_PARTITION_MIGRATION_PATH = join(MIGRATIONS_DIR, "005 const PROJECT_OWNERSHIP_DECLARATION_DRIFT_MIGRATION_PATH = join(MIGRATIONS_DIR, "0056_fn_8997_project_ownership_declaration_drift.sql"); const PROJECT_OWNERSHIP_DEFAULT_RECONCILIATION_MIGRATION_PATH = join(MIGRATIONS_DIR, "0057_fn_9004_project_ownership_default_reconciliation.sql"); const MESSAGE_ARCHIVE_SCHEMA_MIGRATION_PATH = join(MIGRATIONS_DIR, "0058_fn_9014_message_archive.sql"); +const TASK_SOURCE_AGENT_INDEX_MIGRATION_PATH = join(MIGRATIONS_DIR, "0059_fn_9037_tasks_source_agent_index.sql"); /** * Ensure the migration bookkeeping table exists. Lives in the public schema so @@ -582,6 +585,7 @@ export async function applySchemaBaseline( const projectOwnershipDeclarationDriftAlreadyApplied = applied.includes(PROJECT_OWNERSHIP_DECLARATION_DRIFT_VERSION); const projectOwnershipDefaultReconciliationAlreadyApplied = applied.includes(PROJECT_OWNERSHIP_DEFAULT_RECONCILIATION_VERSION); const messageArchiveSchemaAlreadyApplied = applied.includes(MESSAGE_ARCHIVE_SCHEMA_VERSION); + const taskSourceAgentIndexAlreadyApplied = applied.includes(TASK_SOURCE_AGENT_INDEX_VERSION); assertBinaryNotOlderThanDatabase(applied); let schemaChanged = false; @@ -1282,6 +1286,12 @@ export async function applySchemaBaseline( await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${MESSAGE_ARCHIVE_SCHEMA_VERSION}) ON CONFLICT (version) DO NOTHING`); schemaChanged = true; } + if (!taskSourceAgentIndexAlreadyApplied) { + const migrationSql = await readFile(TASK_SOURCE_AGENT_INDEX_MIGRATION_PATH, "utf8"); + await tx.execute(sql.raw(migrationSql)); + await tx.execute(sql`INSERT INTO public.${sql.identifier(MIGRATION_BOOKKEEPING_TABLE)} (version) VALUES (${TASK_SOURCE_AGENT_INDEX_VERSION}) ON CONFLICT (version) DO NOTHING`); + schemaChanged = true; + } return { applied: schemaChanged, pluginHooksRun: pluginHooks.length }; }); } diff --git a/packages/core/src/postgres/schema/project.ts b/packages/core/src/postgres/schema/project.ts index 3c75f696ef..63c9b560fc 100644 --- a/packages/core/src/postgres/schema/project.ts +++ b/packages/core/src/postgres/schema/project.ts @@ -316,6 +316,9 @@ export const tasks = projectSchema.table("tasks", { the gate is a full tasks-table scan. Sparse: most rows have NULL parent. */ index("idxTasksSourceParentTaskId").on(t.sourceParentTaskId), + /* FNXC:TaskRecommendations 2026-08-13-22:23: duplicate intake filters source + * lineage on recommendation/agent creates; this sparse index avoids a task-table scan. */ + index("idxTasksProjectSourceAgentId").on(t.projectId, t.sourceAgentId).where(sql`${t.sourceAgentId} IS NOT NULL`), // FNXC:EphemeralAgentTaskCreation 2026-07-30-12:00: proposal retries share one stable key, so the database—not a read-before-create race—enforces at-most-once materialization. uniqueIndex("uqTasksProjectProposalClaimId").on(t.projectId, t.proposalClaimId).where(sql`${t.proposalClaimId} IS NOT NULL`), /* diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 790bb2d92c..2d94722028 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -150,7 +150,7 @@ import { deleteTaskImpl, archiveTaskImpl, type DeleteTaskIfResult } from "./task import type { TaskDeleteAuditContext } from "./task-delete-attribution.js"; import { updateSettingsImpl, updateGlobalSettingsImpl } from "./task-store/settings-ops.js"; import { createTaskBackendImpl, _createTaskInternalBackendImpl, createTaskImpl, createTaskWithReservedIdImpl, _createTaskInternalImpl, _maybeAutoArchiveSameAgentDuplicateImpl } from "./task-store/task-creation.js"; -import { getTaskImpl, listTasksImpl, searchTasksImpl, listTasksModifiedSinceImpl, getTaskVerificationRequestAsyncImpl, listTaskRecommendationsImpl } from "./task-store/reads.js"; +import { getTaskImpl, listTasksImpl, searchTasksImpl, listTasksModifiedSinceImpl, getTaskVerificationRequestAsyncImpl, listTaskRecommendationsImpl, findTaskByProposalClaimIdImpl, listTasksBySourceLineageImpl } from "./task-store/reads.js"; import { updateTaskUnlockedImpl } from "./task-store/task-update.js"; import { __setTaskActivityLogLimitsForTesting } from "./task-store/comments.js"; import { declaresAnyLifecycleTrait, resolveReviewColumns, resolveTaskLifecycleColumns, type LifecycleColumns } from "./workflows/workflow-lifecycle-traits.js"; @@ -1495,6 +1495,12 @@ export class TaskStore extends EventEmitter { async getTaskColumns(ids: string[]): Promise> { return getTaskColumnsImpl(this, ids); } + async findTaskByProposalClaimId(proposalClaimId: string, options?: { includeDeleted?: boolean }): Promise { + return findTaskByProposalClaimIdImpl(this, proposalClaimId, options); + } + async listTasksBySourceLineage(input: { sourceAgentId?: string | null; sourceParentTaskId?: string | null }): Promise { + return listTasksBySourceLineageImpl(this, input); + } async listTasks(options?: { limit?: number; offset?: number; /** When false, exclude tasks in the `archived` column. Default: true (backward compatible). */ includeArchived?: boolean; /** When true, omit heavy fields (log, comments, steps, workflowStepResults, steeringComments) * from each row to make list responses cheap for board-style consumers. Detail fields default * to empty arrays in the returned Task objects; use `getTask(id)` to load full data. */ slim?: boolean; /** Restrict to a single column (e.g. 'in-review' for the auto-merge sweep). * Widened to {@link ColumnId} (#1403) so custom-column filters are accepted. */ column?: ColumnId; /** Opt-in startup-only memo for repeated slim reads during boot choreography. */ startupMemo?: boolean; /** Forensic read: surface soft-deleted tasks (deletedAt IS NOT NULL). * VAL-DATA-006 — only admin/forensic surfaces should set this. */ includeDeleted?: boolean; }): Promise { return listTasksImpl(this, options); } diff --git a/packages/core/src/task-store/async/async-persistence.ts b/packages/core/src/task-store/async/async-persistence.ts index f6e37d8138..08d63f2f6e 100644 --- a/packages/core/src/task-store/async/async-persistence.ts +++ b/packages/core/src/task-store/async/async-persistence.ts @@ -356,6 +356,46 @@ export async function readTaskRow( * @param id The task id to read. * @param options Optional: includeDeleted surfaces soft-deleted rows. */ +/** + * FNXC:TaskRecommendations 2026-08-13-22:23: + * Claim replay used to scan the whole task table, making recommendation creation hang. + * The partial (project_id, proposal_claim_id) unique index permits at most one match; + * project scope is load-bearing because task identity is composite across projects. + */ +export async function readTaskRowByProposalClaimId( + layer: AsyncDataLayer, + proposalClaimId: string, + options?: { includeDeleted?: boolean }, +): Promise | undefined> { + const conditions = [eq(schema.project.tasks.proposalClaimId, proposalClaimId)]; + if (!options?.includeDeleted) conditions.push(ACTIVE_TASK_FILTER); + const projectScope = taskProjectScope(layer); + if (projectScope) conditions.push(projectScope); + const rows = await layer.db.select().from(schema.project.tasks).where(and(...conditions)); + return rows[0]; +} + +/** Read source-lineage candidates without a board-wide scan; tombstones intentionally participate. */ +export async function readTaskRowsBySourceLineage( + layer: AsyncDataLayer, + input: { sourceAgentId?: string | null; sourceParentTaskId?: string | null }, +): Promise[]> { + const predicates = [ + input.sourceAgentId ? eq(schema.project.tasks.sourceAgentId, input.sourceAgentId) : undefined, + input.sourceParentTaskId ? eq(schema.project.tasks.sourceParentTaskId, input.sourceParentTaskId) : undefined, + ].filter((predicate): predicate is SQL => predicate !== undefined); + if (predicates.length === 0) return []; + /* FNXC:TaskRecommendations 2026-08-13-22:23: SQL pre-narrowing is equivalent to the + * existing JS filter; cold archives have no lineage fields and never matched old scans. */ + const scope = taskProjectScope(layer); + return layer.db.select({ + id: schema.project.tasks.id, title: schema.project.tasks.title, description: schema.project.tasks.description, + column: schema.project.tasks.column, createdAt: schema.project.tasks.createdAt, + sourceAgentId: schema.project.tasks.sourceAgentId, sourceParentTaskId: schema.project.tasks.sourceParentTaskId, + deletedAt: schema.project.tasks.deletedAt, allowResurrection: schema.project.tasks.allowResurrection, + }).from(schema.project.tasks).where(and(scope, sql`(${predicates.reduce((left, right) => sql`${left} OR ${right}`)})`)); +} + export async function readTaskRowInTransaction( tx: DbTransaction, id: string, diff --git a/packages/core/src/task-store/reads.ts b/packages/core/src/task-store/reads.ts index 73bbed1c85..6fdd4b3121 100644 --- a/packages/core/src/task-store/reads.ts +++ b/packages/core/src/task-store/reads.ts @@ -128,7 +128,7 @@ function hasFreshAgentLogActivitySinceTaskUpdate( } import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; -import {readTaskRow, readLiveTaskRows} from "./async/async-persistence.js"; +import {readTaskRow, readLiveTaskRows, readTaskRowByProposalClaimId, readTaskRowsBySourceLineage} from "./async/async-persistence.js"; import {searchTasksTsvector, searchTasksLike} from "./async/async-search.js"; import { getArchivedTask, @@ -201,6 +201,22 @@ async function resolveReviewColumnsForTask( } +/** + * FNXC:TaskRecommendations 2026-08-13-22:23: + * Claim replay is a one-row indexed lookup and intentionally skips cold storage: archive + * snapshots have no proposalClaimId. It also skips board-derived signal hydration because replay needs only persisted data. + */ +export async function findTaskByProposalClaimIdImpl(store: TaskStore, proposalClaimId: string, options?: { includeDeleted?: boolean }): Promise { + if (proposalClaimId.trim().length === 0) return null; + const row = await readTaskRowByProposalClaimId(store.asyncLayer!, proposalClaimId, options); + return row ? store.rowToTask(store.pgRowToTaskRow(row)) : null; +} + +export async function listTasksBySourceLineageImpl(store: TaskStore, input: { sourceAgentId?: string | null; sourceParentTaskId?: string | null }): Promise { + const rows = await readTaskRowsBySourceLineage(store.asyncLayer!, input); + return rows.map((row) => store.rowToTask(store.pgRowToTaskRow(row))); +} + export async function getTaskImpl(store: TaskStore, id: string, options?: { activityLogLimit?: number; includeDeleted?: boolean }): Promise { return store.withTaskLock(id, async () => { // FNXC:RuntimePersistenceAsync 2026-06-24-10:50: diff --git a/packages/core/src/task-store/task-creation.ts b/packages/core/src/task-store/task-creation.ts index 9356744000..ca72d4126b 100644 --- a/packages/core/src/task-store/task-creation.ts +++ b/packages/core/src/task-store/task-creation.ts @@ -159,7 +159,7 @@ export async function createTaskBackendImpl(store: TaskStore, input: TaskCreateI successful prior proposal into a failed retry or emit a second owner signal. */ if (input.proposalClaimId) { - const existing = (await store.listTasks()).find((task) => task.proposalClaimId === input.proposalClaimId); + const existing = await store.findTaskByProposalClaimId(input.proposalClaimId); if (existing) { options?.onProposalClaimConflict?.(existing); return existing; @@ -845,7 +845,7 @@ export async function _createTaskInternalBackendImpl(store: TaskStore, input: Ta workflow materialization. Other unique violations remain task-ID errors. */ if (input.proposalClaimId && isTaskIdConflictError(error)) { - const existing = (await store.listTasks()).find((candidate) => candidate.proposalClaimId === input.proposalClaimId); + const existing = await store.findTaskByProposalClaimId(input.proposalClaimId); if (existing) { options?.onProposalClaimConflict?.(existing); return existing; @@ -916,7 +916,7 @@ export async function createTaskWithReservedIdImpl(store: TaskStore, input: Task to its project-scoped partial index, so this replay read must not touch the removed SQLite backend before the shared insert boundary handles a race. */ - const existing = (await store.listTasks()).find((task) => task.proposalClaimId === input.proposalClaimId); + const existing = await store.findTaskByProposalClaimId(input.proposalClaimId); if (existing) return existing; } @@ -1044,7 +1044,7 @@ export async function createTaskWithReservedIdImpl(store: TaskStore, input: Task // materialized above would orphan with no task/selection pointing at them. await store.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); if (input.proposalClaimId && isTaskIdConflictError(err)) { - const existing = (await store.listTasks()).find((candidate) => candidate.proposalClaimId === input.proposalClaimId); + const existing = await store.findTaskByProposalClaimId(input.proposalClaimId); if (existing) return existing; } throw err; @@ -1339,7 +1339,7 @@ export async function resolveSameAgentDuplicateIntake(store: TaskStore, task: Ta const nowMs = Date.now(); const settings = await store.getSettings(); const stickyWindowDays = Math.max(0, settings.tombstoneStickyWindowDays ?? 7); - const allCandidates = await store.listTasks({ slim: true, includeArchived: true, includeDeleted: true }); + const allCandidates = await store.listTasksBySourceLineage({ sourceAgentId, sourceParentTaskId }); const matches = findSameAgentDuplicates( { title: input.title ?? task.title, description: input.description, sourceParentTaskId }, allCandidates.flatMap((candidate) => { diff --git a/packages/dashboard/src/routes/__tests__/task-recommendation-routes.test.ts b/packages/dashboard/src/routes/__tests__/task-recommendation-routes.test.ts index cf25478ff6..e91992113e 100644 --- a/packages/dashboard/src/routes/__tests__/task-recommendation-routes.test.ts +++ b/packages/dashboard/src/routes/__tests__/task-recommendation-routes.test.ts @@ -35,6 +35,9 @@ function buildApp(seed: Task[], projectId = "project-a") { listTasks: vi.fn(async (options?: { includeDeleted?: boolean }) => tasks.filter((item) => options?.includeDeleted || !item.deletedAt), ), + findTaskByProposalClaimId: vi.fn(async (claimId: string, options?: { includeDeleted?: boolean }) => + tasks.find((item) => item.proposalClaimId === claimId && (options?.includeDeleted || !item.deletedAt)) ?? null, + ), listTaskRecommendations: vi.fn(async (options?: { completeColumns?: ReadonlySet; limit?: number; offset?: number }) => { const rows = tasks.filter((item) => !item.deletedAt && !!item.recommendations?.length && (options?.completeColumns ?? new Set(["done"])).has(item.column)) .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt) || b.id.localeCompare(a.id)); @@ -252,6 +255,39 @@ describe("recommendation task creation route", () => { expect(tasks[0]?.recommendations?.[0]?.createdTaskId).toBe(children[0]?.id); }); + it("keeps recommendation creation free of unbounded listTasks reads when search returns candidates (pre-fix violated this invariant)", async () => { + const { app, store } = buildApp([parent()]); + const response = await performRequest(app, "POST", "/api/tasks/FN-1/recommendations/rec-1/create", undefined); + + expect(response.status).toBe(201); + expect(store.findTaskByProposalClaimId).toHaveBeenCalledWith("recommendation:FN-1:rec-1", { includeDeleted: true }); + expect(store.listTasks).not.toHaveBeenCalled(); + }); + + it("uses only the bounded limit-50 fallback when recommendation search has no candidates (pre-fix violated this invariant)", async () => { + const { app, store } = buildApp([parent({ + recommendations: [{ + id: "rec-1", + title: "Export task records", + description: "Add packages/dashboard/app/api/tasks/export.ts outside this task's scope.", + category: "feature", + }], + })]); + (store.searchTasks as ReturnType).mockResolvedValue([]); + + const response = await performRequest(app, "POST", "/api/tasks/FN-1/recommendations/rec-1/create", undefined); + + expect(response.status).toBe(201); + expect(store.listTasks).toHaveBeenCalledTimes(1); + expect(store.listTasks).toHaveBeenCalledWith(expect.objectContaining({ + limit: 50, + })); + for (const [options] of (store.listTasks as ReturnType).mock.calls) { + expect(typeof options?.limit).toBe("number"); + expect(options.limit).toBeLessThanOrEqual(50); + } + }); + /* FNXC:TaskRecommendations 2026-08-08-08:10: Recommendation parent and row IDs are only project-scoped identities. A slow create in one @@ -371,7 +407,7 @@ describe("recommendation task creation route", () => { expect(response.status).toBe(409); expect(store.createTask).not.toHaveBeenCalled(); expect(tasks[0]?.recommendations?.[0]?.createdTaskId).toBeUndefined(); - expect(store.listTasks).toHaveBeenCalledWith(expect.objectContaining({ includeDeleted: true })); + expect(store.findTaskByProposalClaimId).toHaveBeenCalledWith(expect.any(String), { includeDeleted: true }); }); it("returns the normal duplicate conflict shape when post-create reconciliation loses a race", async () => { diff --git a/packages/dashboard/src/routes/register-task-workflow-routes.ts b/packages/dashboard/src/routes/register-task-workflow-routes.ts index 0180b87b5e..b6b62a4351 100644 --- a/packages/dashboard/src/routes/register-task-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-task-workflow-routes.ts @@ -2013,14 +2013,11 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork if (candidateRows.length === 0) { candidateRows = await scopedStore.listTasks({ slim: true, includeArchived: false, limit: 50 }); } - const fullRows = await scopedStore.listTasks({ slim: false, includeArchived: false }); - const byId = new Map(fullRows.map((row) => [row.id, row])); + /* FNXC:TaskIntakeDedup 2026-08-13-22:23: slim candidates retain every field this + * guard reads; they are live non-archived rows, so the old full-board byId lookup always hit. */ const candidateMap = new Map(); - const classifiedRows = await Promise.all(candidateRows.map(async (row) => { - const full = byId.get(row.id); - return { row, full, blocker: full ? await classifyDuplicateBlocker(full) : true }; - })); - for (const { row, full, blocker } of classifiedRows) { + const classifiedRows = await Promise.all(candidateRows.map(async (row) => ({ row, blocker: await classifyDuplicateBlocker(row) }))); + for (const { row, blocker } of classifiedRows) { if (acknowledgedDuplicateIds.includes(row.id)) { continue; } @@ -2032,9 +2029,9 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork title: row.title ?? "", description: row.description ?? "", column: row.column, - createdAt: full?.createdAt ? Date.parse(full.createdAt) : undefined, - fileScope: Array.isArray(full?.sourceMetadata?.fileScope) - ? full.sourceMetadata.fileScope.filter((entry): entry is string => typeof entry === "string") + createdAt: row.createdAt ? Date.parse(row.createdAt) : undefined, + fileScope: Array.isArray(row.sourceMetadata?.fileScope) + ? row.sourceMetadata.fileScope.filter((entry): entry is string => typeof entry === "string") : undefined, }); } @@ -2439,8 +2436,7 @@ export function registerTaskWorkflowRoutes(ctx: ApiRoutesContext, deps: TaskWork from colliding with the unique claim and manufacturing a second child. Read tombstones here only to return a conflict; they are never relinked or exposed as live recommendation tasks. */ - const existing = (await scopedStore.listTasks({ slim: false, includeArchived: true, includeDeleted: true })) - .find((task) => task.proposalClaimId === proposalClaimId); + const existing = await scopedStore.findTaskByProposalClaimId(proposalClaimId, { includeDeleted: true }); const existingArchiveColumns = existing ? await archivedColumnsForTask(scopedStore, existing.id) : new Set();