FN-9037: add indexed recommendation task lookups
Keep recommendation task creation responsive without changing duplicate-intake semantics. - Add project-scoped proposal claim and source-lineage persistence reads with supporting indexes. - Use bounded lookups for recommendation replay and same-agent duplicate intake. - Resolve terminal lanes from each candidate's workflow and cover the indexed query paths. Files changed: .changeset/fn-9037-recommendation-create-perf.md | 7 ++ docs/storage.md | 4 + .../intake-duplicate-terminal-lanes.test.ts | 3 +- .../postgres/proposal-claim-lookup.pg.test.ts | 106 +++++++++++++++++++++ .../__tests__/same-agent-duplicate-intake.test.ts | 14 ++- .../0059_fn_9037_tasks_source_agent_index.sql | 3 + packages/core/src/postgres/schema-applier.ts | 12 ++- packages/core/src/postgres/schema/project.ts | 3 + packages/core/src/store.ts | 8 +- .../core/src/task-store/async/async-persistence.ts | 40 ++++++++ packages/core/src/task-store/reads.ts | 18 +++- packages/core/src/task-store/task-creation.ts | 10 +- .../__tests__/task-recommendation-routes.test.ts | 38 +++++++- .../src/routes/register-task-workflow-routes.ts | 20 ++-- 14 files changed, 259 insertions(+), 27 deletions(-) Fusion-Task-Id: FN-9037 Fusion-Task-Lineage: ea44d65c-98cc-4a5b-a116-b3c28f14c292 Co-authored-by: Fusion (runfusion.ai) <noreply@runfusion.ai>
This commit is contained in:
gsxdsm
7
.changeset/fn-9037-recommendation-create-perf.md
Normal file
7
.changeset/fn-9037-recommendation-create-perf.md
Normal file
@@ -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.
|
||||
@@ -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: `<rootDir>/.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.
|
||||
|
||||
@@ -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", () => {
|
||||
|
||||
@@ -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<string, unknown> = {},
|
||||
) {
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -39,6 +39,8 @@ function createStore(overrides: Record<string, unknown> = {}) {
|
||||
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",
|
||||
}));
|
||||
|
||||
@@ -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;
|
||||
@@ -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 };
|
||||
});
|
||||
}
|
||||
|
||||
@@ -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`),
|
||||
/*
|
||||
|
||||
@@ -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<TaskStoreEvents> {
|
||||
async getTaskColumns(ids: string[]): Promise<Map<string, Column>> {
|
||||
return getTaskColumnsImpl(this, ids);
|
||||
}
|
||||
async findTaskByProposalClaimId(proposalClaimId: string, options?: { includeDeleted?: boolean }): Promise<Task | null> {
|
||||
return findTaskByProposalClaimIdImpl(this, proposalClaimId, options);
|
||||
}
|
||||
async listTasksBySourceLineage(input: { sourceAgentId?: string | null; sourceParentTaskId?: string | null }): Promise<Task[]> {
|
||||
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<Task[]> {
|
||||
return listTasksImpl(this, options);
|
||||
}
|
||||
|
||||
@@ -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<Record<string, unknown> | 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<Record<string, unknown>[]> {
|
||||
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,
|
||||
|
||||
@@ -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<Task | null> {
|
||||
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<Task[]> {
|
||||
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<TaskDetail> {
|
||||
return store.withTaskLock(id, async () => {
|
||||
// FNXC:RuntimePersistenceAsync 2026-06-24-10:50:
|
||||
|
||||
@@ -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<SameAgentDuplicateCandidate>((candidate) => {
|
||||
|
||||
@@ -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<string>; 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<typeof vi.fn>).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<typeof vi.fn>).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 () => {
|
||||
|
||||
@@ -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<string, NearDuplicateCandidate>();
|
||||
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<string>();
|
||||
|
||||
Reference in New Issue
Block a user