diff --git a/.changeset/fix-pg-store-sqlite-residue.md b/.changeset/fix-pg-store-sqlite-residue.md new file mode 100644 index 0000000000..c91d129843 --- /dev/null +++ b/.changeset/fix-pg-store-sqlite-residue.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Fix data stores that silently failed against PostgreSQL by hitting removed SQLite paths. +category: fix +dev: Residual SQLite-stub sites reachable in backend mode are now routed through the AsyncDataLayer: the executor's authoritative assigned-agent fallback (executor.ts) now inherits the TaskStore asyncLayer (was silently returning null → model drift); `pruneAgentLogFilesAsync` replaces the sync self-healing prune call (was throwing every maintenance sweep); `cleanupOrphanedMaterializedSteps` deletes PG workflow_steps rows on failed create (was leaking); PG hard-delete now runs the async mission feature/task-link unlink (deleteTaskBackendImpl); `getWorkflowSettingsProjectId` no longer touches the SQLite stub for unscoped backend stores; the `fn plugin` unregistered-project fallback bootstraps a CentralCore AsyncDataLayer. Formerly-unported paths are now real backend implementations: `cleanupArchivedTasks` and `deleteWorkflowStep` delete via the async layer; `AgentStore.importLegacyFileRuns` cleanly no-ops in backend mode (no legacy SQLite run-files exist there); the dead zero-caller `applyTaskPatch` SQLite primitive was removed. diff --git a/packages/cli/src/commands/plugin.ts b/packages/cli/src/commands/plugin.ts index 730f509112..ec99632d51 100644 --- a/packages/cli/src/commands/plugin.ts +++ b/packages/cli/src/commands/plugin.ts @@ -13,7 +13,7 @@ import { existsSync } from "node:fs"; import { dirname, extname, join, resolve } from "node:path"; import { readFile, stat } from "node:fs/promises"; import * as readline from "node:readline"; -import { PluginStore, PluginLoader, validatePluginManifest, resolveGlobalDir } from "@fusion/core"; +import { PluginStore, PluginLoader, validatePluginManifest, resolveGlobalDir, CentralCore } from "@fusion/core"; import { resolveProject } from "../project-context.js"; export interface BuiltinPluginCatalogEntry { @@ -110,11 +110,53 @@ export async function createPluginStore( await pluginStore.init(); return pluginStore; } catch { + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + Unregistered-project fallback. PluginStore persists install/state rows in the + central DB and needs an AsyncDataLayer to run in PostgreSQL backend mode; the + old fallback constructed it WITHOUT one, so `init()` hit the removed-SQLite + stub and threw in every PG deployment. Bootstrap a layer-less CentralCore + (which self-bootstraps the embedded-PG AsyncDataLayer in init()) and pass its + layer so the fallback store is backend-mode just like the resolveProject path. + */ const projectPath = await getProjectPath(projectName); + const centralGlobalDir = options?.centralGlobalDir ?? resolveGlobalDir(); + const central = new CentralCore(centralGlobalDir); + await central.init(); + const backendLayer = central.asyncLayer; const pluginStore = new PluginStore(projectPath, { - centralGlobalDir: options?.centralGlobalDir ?? resolveGlobalDir(), + centralGlobalDir, + ...(backendLayer ? { asyncLayer: backendLayer } : {}), }); - await pluginStore.init(); + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-18:10: + `central.init()` bootstraps and OWNS an embedded-Postgres backend (pool + + postmaster). Since we don't return `central`, tie its teardown to the returned + store: closing the PluginStore also stops the CentralCore. `closeWithCentral` is + async so a lifecycle-managing caller can `await store.close()` for a graceful + shutdown (a Promise return is void-compatible with `close(): void`). Even a + fire-and-forget CLI exit does NOT leak the backend: EmbeddedPostgres self-registers + beforeExit/SIGTERM/SIGINT stop hooks (see embedded-lifecycle.ts) that stop the + postmaster if the caller forgets, and `beforeExit` keeps the process alive for the + async stop. + */ + const closePluginStore = pluginStore.close.bind(pluginStore); + const closeWithCentral = async (): Promise => { + closePluginStore(); + await central.close().catch(() => undefined); + }; + pluginStore.close = closeWithCentral; + try { + await pluginStore.init(); + } catch (initErr) { + /* + If init() rejects the store is never returned, so the reassigned close() is + unreachable. Await the teardown here so the owned CentralCore's embedded + Postgres is stopped before we rethrow. + */ + await closeWithCentral(); + throw initErr; + } return pluginStore; } } diff --git a/packages/core/src/__tests__/postgres/store-sqlite-residue-fixes.pg.test.ts b/packages/core/src/__tests__/postgres/store-sqlite-residue-fixes.pg.test.ts new file mode 100644 index 0000000000..2764346b1f --- /dev/null +++ b/packages/core/src/__tests__/postgres/store-sqlite-residue-fixes.pg.test.ts @@ -0,0 +1,148 @@ +/** + * FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + * Regression coverage for the SQLite→PostgreSQL store-migration residue fixes. + * Each of these store paths reached the removed-SQLite stub (`store.db`) while + * running in backend mode; the throw was either surfaced every run or swallowed + * into a silent wrong result. These tests reproduce the original symptom + * (exercised against the real embedded-PG backend) and assert it is gone. + * + * Symptom Verification: + * - #2 pruneAgentLogFilesAsync: the self-healing maintenance sweep called the + * sync `pruneAgentLogFiles`, which threw "SQLite Database is not available" + * every backend-mode run → agent-log pruning never ran. Assert the async + * variant resolves and prunes inactive-task files without throwing. + * - #3 cleanupOrphanedMaterializedSteps: the sync `store.db.prepare(DELETE ...)` + * threw, was swallowed by the best-effort catch, and orphaned workflow_steps + * rows leaked on a failed task-create. Assert the row is actually deleted. + * - #4 hard delete mission unlink: the sync path could only drive the sync + * MissionStore, so PG hard delete left orphaned mission feature→task links. + * Assert the feature is unlinked after a backend hard delete. + */ +import { afterAll, afterEach, beforeAll, beforeEach, expect, it } from "vitest"; +import { eq, inArray } from "drizzle-orm"; +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import * as schema from "../../postgres/schema/index.js"; +import { AsyncMissionStore } from "../../async-mission-store.js"; +import { + createSharedPgTaskStoreTestHarness, + pgDescribe, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; + +pgDescribe("PostgreSQL store-migration residue fixes", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ prefix: "fusion_pg_residue" }); + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + it("#3 cleanupOrphanedMaterializedSteps deletes workflow_steps rows in backend mode (no swallowed throw)", async () => { + const store = h.store(); + expect(store.backendMode).toBe(true); + + const step = await store.createWorkflowStep({ name: "Orphan", description: "leaked step" }); + // Precondition: the row exists in PostgreSQL. + const before = await h.adminDb() + .select({ id: schema.project.workflowSteps.id }) + .from(schema.project.workflowSteps) + .where(inArray(schema.project.workflowSteps.id, [step.id])); + expect(before.map((r) => r.id)).toEqual([step.id]); + + // The old sync path threw "SQLite Database is not available" here and the + // best-effort catch swallowed it — the row would have leaked. + await expect(store.cleanupOrphanedMaterializedSteps([step.id])).resolves.toBeUndefined(); + + const after = await h.adminDb() + .select({ id: schema.project.workflowSteps.id }) + .from(schema.project.workflowSteps) + .where(inArray(schema.project.workflowSteps.id, [step.id])); + expect(after).toEqual([]); + }); + + it("#2 pruneAgentLogFilesAsync resolves in backend mode and prunes inactive-task log files", async () => { + const store = h.store(); + expect(store.backendMode).toBe(true); + + // A soft-deleted task is "inactive" and eligible for JSONL pruning. + const task = await store.createTask({ description: "to be deleted" }); + await store.deleteTask(task.id); + + // Seed an expired agent-log JSONL entry for the inactive task. + const taskDir = join(store.tasksDir, task.id); + await mkdir(taskDir, { recursive: true }); + const oldTs = new Date(Date.now() - 90 * 86_400_000).toISOString(); + await writeFile( + join(taskDir, "agent-log.jsonl"), + `${JSON.stringify({ timestamp: oldTs, type: "info", message: "old", taskId: task.id })}\n`, + ); + + // The old sync `pruneAgentLogFiles` threw the removed-SQLite stub here. + const result = await store.pruneAgentLogFilesAsync(30); + expect(result).toBeDefined(); + expect(typeof result.prunedEntries).toBe("number"); + // The seeded expired entry for the inactive task is pruned. + expect(result.prunedEntries).toBeGreaterThanOrEqual(1); + }); + + it("#4 hard delete unlinks the mission feature from the task in backend mode", async () => { + const store = h.store(); + const missions = store.getMissionStore() as AsyncMissionStore; + expect(missions).toBeInstanceOf(AsyncMissionStore); + + const mission = await missions.createMission({ title: "Residue mission" }); + const milestone = await missions.addMilestone(mission.id, { title: "MS" }); + const slice = await missions.addSlice(milestone.id, { title: "SL" }); + const feature = await missions.addFeature(slice.id, { title: "F" }); + const task = await store.createTask({ description: "linked delivery task" }); + + const linked = await missions.linkFeatureToTask(feature.id, task.id); + expect(linked.taskId).toBe(task.id); + + // Hard delete used to skip the async mission unlink → orphaned link. + await store.deleteTask(task.id); + + const orphan = await missions.getFeatureByTaskId(task.id); + expect(orphan).toBeUndefined(); + const refreshed = await missions.getFeature(feature.id); + expect(refreshed?.taskId).toBeUndefined(); + }); + + it("deleteWorkflowStep removes the row in backend mode and reports not-found for a missing id", async () => { + const store = h.store(); + const step = await store.createWorkflowStep({ name: "Deletable", description: "step" }); + + await store.deleteWorkflowStep(step.id); + const rows = await h.adminDb() + .select({ id: schema.project.workflowSteps.id }) + .from(schema.project.workflowSteps) + .where(inArray(schema.project.workflowSteps.id, [step.id])); + expect(rows).toEqual([]); + + // Contract preserved: deleting a non-existent step throws. + await expect(store.deleteWorkflowStep("WS-does-not-exist")).rejects.toThrow(/not found/); + }); + + it("cleanupArchivedTasks hard-deletes archived project rows while retaining the cold snapshot", async () => { + const store = h.store(); + const task = await store.createTask({ description: "to be purged" }); + await store.archiveTask(task.id); + + const cleaned = await store.cleanupArchivedTasks(); + expect(cleaned).toContain(task.id); + + // Live project row is gone... + const liveRows = await h.adminDb() + .select({ id: schema.project.tasks.id }) + .from(schema.project.tasks) + .where(eq(schema.project.tasks.id, task.id)); + expect(liveRows).toEqual([]); + + // ...but the cold-storage snapshot survives for restore. + const coldRows = await h.adminDb() + .select({ id: schema.archive.archivedTasks.id }) + .from(schema.archive.archivedTasks) + .where(eq(schema.archive.archivedTasks.id, task.id)); + expect(coldRows.map((r) => r.id)).toContain(task.id); + }); +}); diff --git a/packages/core/src/agent-store.ts b/packages/core/src/agent-store.ts index 7d6244774f..2032dc8f6b 100644 --- a/packages/core/src/agent-store.ts +++ b/packages/core/src/agent-store.ts @@ -541,6 +541,15 @@ export class AgentStore extends EventEmitter { * Runtime reads come from SQLite; this only seeds old projects. */ async importLegacyFileRuns(): Promise { + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + One-shot legacy SQLite migration. In backend mode the INSERT below hits the + removed-SQLite stub, and this method's per-file try/catch would swallow the + throw and silently report "0 imported". There are no legacy SQLite file runs + to import against PostgreSQL (the PG baseline already covers this), so no-op + cleanly instead of laundering a stub throw into a false success. + */ + if (this.backendMode) return 0; const entries = await readdir(this.agentsDir, { withFileTypes: true }).catch(() => []); const runDirs = entries.filter((entry) => entry.isDirectory() && entry.name.endsWith("-runs")); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index aa4b321d97..c7c81f915a 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -97,7 +97,7 @@ import { TASK_JSONB_COLUMNS, type TaskRow, type TaskPersistSerializationContext, import { pgRowToTaskRow as pgRowToTaskRowExternal, rowToTask as rowToTaskExternal, rowToBranchGroup as rowToBranchGroupExternal, generateBranchGroupId as generateBranchGroupIdExternal, computeTimedExecutionMs as computeTimedExecutionMsExternal, archiveEntryToTask as archiveEntryToTaskExternal, summarizeAgentLog as summarizeAgentLogExternal, rowToTaskDocument as rowToTaskDocumentExternal, rowToArtifact as rowToArtifactExternal, rowToTaskDocumentRevision as rowToTaskDocumentRevisionExternal, rowToGoalCitation as rowToGoalCitationExternal } from "./task-store/serialization.js"; import { moveTaskImpl, handoffToReviewImpl, moveTaskInternalImpl } from "./task-store/moves.js"; import { recordGoalCitationsImpl, insertTaskWithFtsRecoveryImpl2, assertTaskIdAvailableImpl, atomicWriteTaskJsonImpl2, createTaskWithDistributedReservationImpl, toStoredWorkflowStepImpl, ensureWorkflowStepForTemplateImpl, resolveEnabledWorkflowStepsImpl, setTaskBranchGroupImpl, getTaskColumnsImpl, prepareWorkflowMovePolicyPreflightImpl, updateTaskCustomFieldsImpl, listWorkflowPromptOverridesForProjectImpl, listWorkflowWorkItemsForTaskImpl, listDueWorkflowWorkItemsImpl, rewriteBlockedByResidueDependentsForRemovalImpl, getAllDocumentsImpl, deleteWorkflowStepImpl, toWorkflowDefinitionImpl, materializeDefaultWorkflowStepsImpl, reconcileTaskCustomFieldsForSchemaImpl, getTaskMovedCountsByDayImpl, getGoalStoreImpl, upsertTaskCommitAssociationImpl } from "./task-store/remaining-ops-4.js"; -import { applyLegacyWorkflowStepOverridesImpl, applyTaskPatchImpl, archiveDbImpl, assertNoDependencyCycleImpl, atomicCreateTaskJsonImpl, buildActiveTaskDependencyLookupImpl, buildArchivedAgentLogFieldsImpl, buildTaskIdIntegrityFallbackReportImpl, createBranchGroupImpl, dbImpl, detectAndCacheTaskIdIntegrityReportImpl, findLiveDependentsImpl, findLiveLineageChildrenImpl, getLegacyWorkflowStepSnapshotImpl, getMalformedTaskMetadataReasonImpl, getMergeQueuedTaskIdsAsyncImpl, insertRunAuditEventRowImpl, insertTaskImpl, invokeTaskCreatedHookImpl, isTaskArchivedImpl, isTaskIdPresentInArchivedTasksTableImpl, logTaskCreateConflictImpl, maybeResolveTombstonedTaskIdImpl, mergeTaskIdIntegrityReportsImpl, optionalGroupIdSetImpl, patchTaskRowInTransactionImpl, readConfigFastImpl, readConfigImpl, readPromptForArchiveImpl, readTaskFromDbImpl, reconcileDistributedTaskIdStateOnOpenImpl, recordActivityFromListenerImpl, recordDependencyCycleRejectedAuditImpl, refreshTaskIdIntegrityReportImpl, resolveLocalNodeIdForTaskAllocationImpl, runTaskFtsWriteWithRecoveryImpl, scanAndRecordCitationsImpl, taskIdExistsAnywhereImpl, throwSoftDeletedWriteBlockedImpl, toBuiltInWorkflowStepImpl, trackDeferredTaskCreatedWorkImpl, upsertTaskImpl, withConfigLockImpl, withTaskLockImpl, withWorktreeAllocationLockImpl } from "./task-store/remaining-ops-5.js"; +import { applyLegacyWorkflowStepOverridesImpl, archiveDbImpl, assertNoDependencyCycleImpl, atomicCreateTaskJsonImpl, buildActiveTaskDependencyLookupImpl, buildArchivedAgentLogFieldsImpl, buildTaskIdIntegrityFallbackReportImpl, createBranchGroupImpl, dbImpl, detectAndCacheTaskIdIntegrityReportImpl, findLiveDependentsImpl, findLiveLineageChildrenImpl, getLegacyWorkflowStepSnapshotImpl, getMalformedTaskMetadataReasonImpl, getMergeQueuedTaskIdsAsyncImpl, insertRunAuditEventRowImpl, insertTaskImpl, invokeTaskCreatedHookImpl, isTaskArchivedImpl, isTaskIdPresentInArchivedTasksTableImpl, logTaskCreateConflictImpl, maybeResolveTombstonedTaskIdImpl, mergeTaskIdIntegrityReportsImpl, optionalGroupIdSetImpl, patchTaskRowInTransactionImpl, readConfigFastImpl, readConfigImpl, readPromptForArchiveImpl, readTaskFromDbImpl, reconcileDistributedTaskIdStateOnOpenImpl, recordActivityFromListenerImpl, recordDependencyCycleRejectedAuditImpl, refreshTaskIdIntegrityReportImpl, resolveLocalNodeIdForTaskAllocationImpl, runTaskFtsWriteWithRecoveryImpl, scanAndRecordCitationsImpl, taskIdExistsAnywhereImpl, throwSoftDeletedWriteBlockedImpl, toBuiltInWorkflowStepImpl, trackDeferredTaskCreatedWorkImpl, upsertTaskImpl, withConfigLockImpl, withTaskLockImpl, withWorktreeAllocationLockImpl } from "./task-store/remaining-ops-5.js"; import { claimNextToolFailureRetryImpl, clearNearDuplicateReferencesToFailSoftImpl, clearWorkflowRunStepInstancesAsyncImpl, clearWorkflowRunStepInstancesImpl, computeMovedSettingsTargetWorkflowIdsImpl, ensureBranchGroupForSourceImpl, ensurePrEntityForSourceImpl, findRecentTasksByContentFingerprintImpl, getActiveMergingTaskImpl, getActivePrEntityBySourceImpl, getBranchGroupByBranchNameImpl, getBranchGroupBySourceImpl, getBranchGroupImpl, getBranchProgressByTaskImpl, getMutationsForRunImpl, getPrEntityByNumberImpl, getPrEntityImpl, getPrThreadStateImpl, getTasksByAssignedAgentImpl, getWorkflowPromptOverridesAsyncImpl, getWorkflowSettingValuesAsyncImpl, getWorkflowSettingValuesImpl, getWorkflowSettingsProjectIdImpl, getWorkflowWorkItemImpl, insertCompletionHandoffWorkflowWorkAuditImpl, listActivePrEntitiesImpl, listBranchGroupsImpl, listPrThreadStatesImpl, listTasksByBranchGroupImpl, listWorkflowSettingValuesForProjectImpl, loadWorkflowRunBranchesImpl, loadWorkflowRunStepInstancesAsyncImpl, loadWorkflowRunStepInstancesImpl, markToolFailureRetryExhaustedAuditImpl, mergeCustomFieldPatchImpl, normalizeMergeRequestStateImpl, normalizeWorkflowWorkItemKindImpl, normalizeWorkflowWorkItemStateImpl, parseWorkflowPromptOverrideJsonImpl, recordPrThreadOutcomeImpl, resetAllStepsToPendingImpl, resetPromptCheckboxesImpl, resolveWorkflowMoveActorImpl, resolveWorkflowSettingDeclarationsImpl, saveWorkflowRunStepInstanceAsyncImpl, saveWorkflowRunStepInstanceImpl, transitionMergeRequestStateImpl, transitionWorkflowWorkItemSyncImpl, updateTaskImpl, updateWorkflowPromptOverridesImpl, upsertMergeRequestRecordImpl, workflowStateForMergeRequestStateImpl } from "./task-store/remaining-ops-6.js"; import { addPrInfoImpl, addSteeringCommentImpl, archiveAllDoneImpl, cleanupStaleMergeQueueRowsImpl, clearCompletionHandoffAcceptedMarkerImpl, clearDoneTransientFieldsImpl, clearStaleExecutionStartBranchReferencesImpl, computeWorkflowColumnsGraduationReportImpl, deleteTaskCommentImpl, deleteTaskDocumentImpl, emitUsageEventImpl, enqueueMergeQueueImpl, getAgentLogCountImpl, getAgentLogsImpl, getArtifactImpl, getArtifactsImpl, getAttachmentImpl, getCompletionHandoffAcceptedMarkerImpl, getTaskDocumentImpl, getTaskDocumentRevisionsImpl, getTaskDocumentsImpl, insertArtifactRowImpl, linkGithubIssueImpl, listWorkflowWorkItemsForTaskSyncImpl, moveToDoneImpl, parseDependenciesFromPromptImpl, parseFileScopeFromPromptImpl, parseStepsFromPromptImpl, peekMergeQueueHeadImpl, peekMergeQueueImpl, readPreArchiveColumnFromTaskFileImpl, recordPluginActivationImpl, recordRunAuditEventBackendImpl, removePrInfoByNumberImpl, resolvePrimaryPrInfoImpl, resolveUnarchiveTargetColumnImpl, rewriteLineageChildrenForRemovalImpl, runGitCommandImpl, stopWatchingImpl, syncAgentTaskLinkOnReassignmentImpl, updateArtifactImpl, updateGithubTrackingImpl, updatePrInfoByNumberImpl, updateTaskCommentImpl, upsertPrInfoByNumberImpl, writeArtifactDataImpl } from "./task-store/remaining-ops-7.js"; import { approveCliAutonomyImpl, approveWorkflowCliCommandImpl, cleanupOrphanedMaterializedStepsImpl, consumePluginGateVerdictsImpl, getAgentLogsByTimeRangeImpl, getDatabaseHealthImpl, getDistributedTaskIdAllocatorImpl, getExperimentSessionStoreImpl, getInReviewDurationEventsImpl, getMissionStoreImpl, getPluginStoreImpl, getSecretsStoreImpl, getSettingsSyncImpl, getTaskMergedTaskIdsImpl, getTaskWorkflowSelectionImpl, getImportTranslationImpl, recordImportTranslationImpl, pruneImportTranslationsImpl, type ImportTranslationCacheKey, type ImportTranslationCacheEntry, getVerificationCacheHitImpl, getWorkflowDefinitionImpl, healthCheckImpl, importLegacyAgentLogsOnceImpl, insertWorkflowDefinitionSyncImpl, isCliAutonomyApprovedImpl, isPluginInstalledImpl, isWorkflowCliCommandApprovedImpl, listWorkflowDefinitionsImpl, materializeExplicitWorkflowStepsImpl, materializeWorkflowStepsImpl, migrateActiveArchivedTasksToArchiveDbImpl, migrateLegacyArchiveEntriesToArchiveDbImpl, nextWorkflowDefinitionIdImpl, occupantsByColumnForWorkflowImpl, parseWorkflowLayoutImpl, pruneAgentLogFilesImpl, purgeTaskWorkflowSelectionRowsImpl, readAllWorkflowDefinitionsImpl, readRawProjectSettingsImpl, recordPluginGateVerdictImpl, recordVerificationCachePassImpl, removeMaterializedSelectionImpl, resolvePluginWorkflowStepImpl, resolveTaskWorkflowIrSyncImpl, revokeCliAutonomyImpl, selectTaskWorkflowAndReconcileImpl, writeTaskWorkflowSelectionImpl, getTaskWorkflowSelectionAsyncImpl, } from "./task-store/remaining-ops-8.js"; @@ -109,7 +109,7 @@ import { getOrCreateForProjectImpl, listGoalCitationsImpl, atomicWriteTaskJsonWi import { markLegacyAutoMergeStampsOnceImpl, appendAgentLogImpl, importLegacyAgentLogsImpl, cleanupNoOpTaskMovedActivityRowsOnceImpl, runWorkflowColumnsIntegrityPassImpl, backfillCommitAssociationDiffStatsImpl } from "./task-store/workflow-integrity.js"; import { saveWorkflowRunBranchImpl, clearNearDuplicateReferencesToImpl, selectNextTaskForAgentImpl, pauseTaskImpl, clearLinkedAgentTaskIdsImpl, listArtifactsImpl, rehomeOccupantImpl } from "./task-store/branch-group-ops.js"; import { taskToArchiveEntryImpl, deleteTaskBackendImpl, archiveTaskBackendImpl, unarchiveTaskImpl, restoreFromArchiveImpl, listArchivedTasksImpl } from "./task-store/archive-lifecycle-2.js"; -import { pruneOperationalLogsAsync, type OperationalLogPruneResult } from "./task-store/async-maintenance.js"; +import { pruneOperationalLogsAsync, pruneAgentLogFilesAsync, type OperationalLogPruneResult } from "./task-store/async-maintenance.js"; import { reconcilePhantomCommittedReservationsAsync } from "./task-store/async-phantom-reservations.js"; import { queryRunAuditEvents } from "./task-store/async-audit.js"; import { isValidMergeRequestTransitionImpl, enqueueMergeQueueSyncInternalImpl, releaseMergeQueueLeaseImpl, collectMergeDetailsImpl, applyPrMergedTransitionImpl } from "./task-store/merge-queue-ops-2.js"; @@ -612,8 +612,6 @@ export class TaskStore extends EventEmitter { public patchTaskRowInTransaction( id: string, task: Task, changedColumns: Iterable, existingRow?: TaskRow, ): { deletedAt?: string; current?: Task } { return patchTaskRowInTransactionImpl(this, id, task, changedColumns, existingRow); } - public async applyTaskPatch( dir: string, id: string, task: Task, changedColumns: Iterable, options?: { existingRow?: TaskRow; auditInput?: { agentId?: string; runId?: string; timestamp?: string; operation?: string } }, ): Promise { return applyTaskPatchImpl(this, dir, id, task, changedColumns, options); - } public readTaskFromDb(id: string, options?: { activityLogLimit?: number; includeDeleted?: boolean }): Task | undefined { return readTaskFromDbImpl(this, id, options); } @@ -2394,7 +2392,7 @@ Issue #2149 requires read-only type filtering to occur in the file-store before public purgeTaskWorkflowSelectionRows(taskId: string): void { return purgeTaskWorkflowSelectionRowsImpl(this, taskId); } - public cleanupOrphanedMaterializedSteps(stepIds: string[] | undefined): void { + public cleanupOrphanedMaterializedSteps(stepIds: string[] | undefined): Promise { return cleanupOrphanedMaterializedStepsImpl(this, stepIds); } public async materializeWorkflowSteps( workflowId: string, inputs: import("./types.js").WorkflowStepInput[], ): Promise { @@ -2480,6 +2478,21 @@ Issue #2149 requires read-only type filtering to occur in the file-store before pruneAgentLogFiles(retentionDays: number): { prunedFiles: number; prunedEntries: number; freedBytes: number } { return pruneAgentLogFilesImpl(this, retentionDays); } + /** + * FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + * Backend-mode entry point for agent-log-file pruning. The sync + * `pruneAgentLogFiles` reads inactive task ids via `this.db` and throws in + * backend mode; callers (the self-healing maintenance sweep) MUST use this + * async variant, which routes to `project.tasks` when an AsyncDataLayer is + * present and falls back to the legacy sync path otherwise. Mirrors + * `pruneOperationalLogsAsync`. + */ + async pruneAgentLogFilesAsync(retentionDays: number): Promise<{ prunedFiles: number; prunedEntries: number; freedBytes: number }> { + if (!this.asyncLayer) { + return this.pruneAgentLogFiles(retentionDays); + } + return pruneAgentLogFilesAsync(this.asyncLayer, this.tasksDir, retentionDays); + } getRootDir(): string { return this.rootDir; } diff --git a/packages/core/src/task-store/archive-lifecycle-2.ts b/packages/core/src/task-store/archive-lifecycle-2.ts index 70dfccb49d..132af7af66 100644 --- a/packages/core/src/task-store/archive-lifecycle-2.ts +++ b/packages/core/src/task-store/archive-lifecycle-2.ts @@ -7,6 +7,7 @@ * instance as its first parameter and performs byte-identical work. */ import {TaskStore, storeLog} from "../store.js"; +import {getFeatureByTaskId as getMissionFeatureByTaskId, unlinkFeatureFromTaskId as unlinkMissionFeatureFromTaskId} from "../async-mission-store-queries.js"; import {TaskHasLineageChildrenError, TaskSelfDeleteError} from "./errors.js"; import {mkdir, writeFile} from "node:fs/promises"; import {join} from "node:path"; @@ -128,12 +129,26 @@ export async function deleteTaskBackendImpl(store: TaskStore, id: string, option const deletedAt = new Date().toISOString(); const allowResurrection = options?.allowResurrection === true; - // Soft-delete + lineage clear + audit in one transaction (atomicity). + // Soft-delete + lineage clear + mission unlink + audit in one transaction (atomicity). await layer.transactionImmediate(async (tx) => { // Clear lineage references on live children so the parent can be deleted. if (lineageChildIds.length > 0) { await removeLineageReferences(tx, id, lineageChildIds, deletedAt, layer.projectId); } + /* + FNXC:MissionStore 2026-07-17-17:40: + Clear any mission feature→task link IN THIS TRANSACTION so it commits (or rolls + back) atomically with the soft-delete. The prior post-commit / pre-commit variants + could leave the two out of sync on a partial failure: a committed delete with a + dangling feature pointer, or a committed unlink whose delete then failed and could + not be recovered (getFeatureByTaskId no longer finds it). Running the tx-scoped + taskId=NULL clear alongside the delete removes that window. The feature's status + rollup is non-critical for a deleted task and self-heals on the next mission read. + */ + const linkedFeature = await getMissionFeatureByTaskId(tx, id); + if (linkedFeature) { + await unlinkMissionFeatureFromTaskId(tx, linkedFeature.id); + } // Soft-delete the task row. await softDeleteTaskRowInTransaction(tx, id, deletedAt, allowResurrection, layer.projectId); // Record the audit event. diff --git a/packages/core/src/task-store/async-maintenance.ts b/packages/core/src/task-store/async-maintenance.ts index db12809c7f..c9f5fc6f6b 100644 --- a/packages/core/src/task-store/async-maintenance.ts +++ b/packages/core/src/task-store/async-maintenance.ts @@ -1,11 +1,46 @@ import { sql } from "drizzle-orm"; import type { AsyncDataLayer } from "../postgres/data-layer.js"; +import { pruneAgentLogFiles as pruneAgentLogFileEntries } from "../agent-log-file-store.js"; export interface OperationalLogPruneResult { deletedByTable: Record; deletedTotal: number; } +/** + * Prune per-task agent-log JSONL files for tasks that are no longer active, + * against the PostgreSQL backend. + * + * FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + * The sync `pruneAgentLogFilesImpl` reads inactive task ids via `store.db`, + * which throws the removed-SQLite stub in backend mode. The self-healing + * maintenance sweep (`prune-agent-log-files`) called it unguarded, so every PG + * sweep with retention enabled threw and agent-log pruning never ran. This + * async variant is the PostgreSQL equivalent: it reads the inactive task ids + * (`deleted_at IS NOT NULL OR column = 'archived'`, project-scoped) from + * `project.tasks` and delegates the file pruning to the shared helper. Query + * semantics mirror the SQLite path exactly. + */ +export async function pruneAgentLogFilesAsync( + layer: AsyncDataLayer, + tasksDir: string, + retentionDays: number, +): Promise<{ prunedFiles: number; prunedEntries: number; freedBytes: number }> { + if (!Number.isFinite(retentionDays) || retentionDays <= 0) { + return { prunedFiles: 0, prunedEntries: 0, freedBytes: 0 }; + } + const boundProjectId = layer.projectId?.trim(); + if (!boundProjectId) { + console.warn("[fusion] PostgreSQL agent-log-file pruning is using the legacy unscoped project sentinel because asyncLayer.projectId is missing"); + } + const projectId = boundProjectId || "__legacy_unscoped__"; + const rows = (await layer.db.execute( + sql`SELECT id FROM project.tasks WHERE project_id = ${projectId} AND (deleted_at IS NOT NULL OR "column" = 'archived')`, + )) as unknown as Array<{ id: string }>; + const inactiveTaskIds = new Set(rows.map((row) => row.id)); + return pruneAgentLogFileEntries(tasksDir, retentionDays, inactiveTaskIds); +} + /** * Delete project-scoped operational history older than the retention cutoff. * diff --git a/packages/core/src/task-store/remaining-ops-2.ts b/packages/core/src/task-store/remaining-ops-2.ts index d82724815d..6d10cd1060 100644 --- a/packages/core/src/task-store/remaining-ops-2.ts +++ b/packages/core/src/task-store/remaining-ops-2.ts @@ -26,6 +26,8 @@ import {type TaskRow, TASK_COLUMN_DESCRIPTORS} from "../task-store/persistence.j import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; import {assertSafeGitBranchName} from "../task-store/shell-safety.js"; import {readTaskRow as readTaskRowAsync, readTaskRowInTransaction} from "../task-store/async-persistence.js"; +import {upsertArchivedTaskEntry} from "./async-archive-lineage.js"; +import {purgeTaskWorkflowSelectionRowsAsyncImpl} from "./remaining-ops-8.js"; import * as schema from "../postgres/schema/index.js"; import {and, asc, eq, isNotNull, isNull, sql} from "drizzle-orm"; import {recoverExpiredMergeQueueLeases as recoverExpiredMergeQueueLeasesAsync} from "../task-store/async-merge-coordination.js"; @@ -1241,6 +1243,62 @@ export async function unlinkGithubIssueImpl(store: TaskStore, id: string): Promi } export async function cleanupArchivedTasksImpl(store: TaskStore): Promise { + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-15:10: + Backend-mode port. `cleanupArchivedTasks` is the hard-removal path for tasks + already in the `archived` column (the CLI documents it as such): it snapshots + each to cold storage, hard-deletes the live project row, and removes the task + directory. In PostgreSQL, archived rows are soft-deleted (`deleted_at` set), so + enumeration MUST pass `includeDeleted`. The cold snapshot upsert is idempotent + (archive already holds it from archive time); the project-row DELETE fires the + ON DELETE CASCADE that purges the task's documents/artifacts, matching the + SQLite path's dir removal. Selection rows are purged via the async helper. + */ + if (store.backendMode) { + const layer = store.asyncLayer!; + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-17:40: + Enumerate the archived rows with an EXPLICIT project predicate. `listTasks()` + derives its scope from `taskProjectScope(layer)`, which is a NO-OP when the + layer is unbound (projectId absent) — i.e. it would read archived rows across + every project, and this destructive sweep (snapshot + dir removal + cache + evict) would then touch tasks it must never own. Scoping the read here to the + same `projectId` the DELETE below uses keeps enumerate+delete lockstep: a bound + store sees only its project, an unbound store only the `__legacy_unscoped__` + quarantine partition. + */ + const projectId = layer.projectId?.trim() || "__legacy_unscoped__"; + const archivedRows = await layer.db + .select() + .from(schema.project.tasks) + .where(and(eq(schema.project.tasks.projectId, projectId), eq(schema.project.tasks.column, "archived"))); + const cleanedUpIds: string[] = []; + const { rm } = await import("node:fs/promises"); + + for (const row of archivedRows) { + const task = store.rowToTask(store.pgRowToTaskRow(row)); + const dir = store.taskDir(task.id); + // Guarantee a cold-storage snapshot before the destructive delete. + const entry = await store.taskToArchiveEntry(task, task.deletedAt ?? new Date().toISOString()); + await upsertArchivedTaskEntry(layer.db, entry, layer.projectId); + + await purgeTaskWorkflowSelectionRowsAsyncImpl(store, task.id); + await layer.db + .delete(schema.project.tasks) + .where(and(eq(schema.project.tasks.projectId, projectId), eq(schema.project.tasks.id, task.id))); + + if (existsSync(dir)) { + await rm(dir, { recursive: true, force: true }); + } + if (store.isWatching) { + store.taskCache.delete(task.id); + } + cleanedUpIds.push(task.id); + } + + return cleanedUpIds; + } + const archivedTasks = await store.listTasks({ column: "archived" }); const cleanedUpIds: string[] = []; diff --git a/packages/core/src/task-store/remaining-ops-4.ts b/packages/core/src/task-store/remaining-ops-4.ts index 001e2d098a..517b061ec4 100644 --- a/packages/core/src/task-store/remaining-ops-4.ts +++ b/packages/core/src/task-store/remaining-ops-4.ts @@ -635,17 +635,37 @@ export async function getAllDocumentsImpl(store: TaskStore, options?: { searchQu } export async function deleteWorkflowStepImpl(store: TaskStore, id: string): Promise { - const deleted = store.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(id) as { - changes?: number; - }; + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-15:10: + Backend mode deletes the workflow_steps row via the AsyncDataLayer (mirrors the + async workflow_steps deletes in workflow-ops.ts). The `.returning()` clause lets + us preserve the sync path's "not found" contract. `bumpLastModified` is a + SQLite-only mtime touch with no PostgreSQL analogue. The task-reference cleanup + below is already async and backend-safe, so it runs in both modes. + */ + if (store.backendMode) { + const layer = store.asyncLayer!; + const deletedRows = await layer.db + .delete(schema.project.workflowSteps) + .where(eq(schema.project.workflowSteps.id, id)) + .returning({ id: schema.project.workflowSteps.id }); + if (deletedRows.length === 0) { + throw new Error(`Workflow step '${id}' not found`); + } + store.workflowStepsCache = null; + } else { + const deleted = store.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(id) as { + changes?: number; + }; - if ((deleted.changes || 0) === 0) { - throw new Error(`Workflow step '${id}' not found`); + if ((deleted.changes || 0) === 0) { + throw new Error(`Workflow step '${id}' not found`); + } + + store.db.bumpLastModified(); + store.workflowStepsCache = null; } - store.db.bumpLastModified(); - store.workflowStepsCache = null; - // Clean up references from existing tasks (best-effort, outside config lock) try { const tasks = await store.listTasks({ slim: true }); diff --git a/packages/core/src/task-store/remaining-ops-5.ts b/packages/core/src/task-store/remaining-ops-5.ts index e94ab5e146..1eb64c25b2 100644 --- a/packages/core/src/task-store/remaining-ops-5.ts +++ b/packages/core/src/task-store/remaining-ops-5.ts @@ -300,26 +300,14 @@ export function patchTaskRowInTransactionImpl(store: TaskStore, return { current: store.readTaskFromDb(id) }; } -export async function applyTaskPatchImpl(store: TaskStore, - dir: string, - id: string, - task: Task, - changedColumns: Iterable, - options?: { existingRow?: TaskRow; auditInput?: { agentId?: string; runId?: string; timestamp?: string; operation?: string } }, - ): Promise { - let result: { deletedAt?: string; current?: Task } | undefined; - store.db.transactionImmediate(() => { - result = store.patchTaskRowInTransaction(id, task, changedColumns, options?.existingRow); - }); - if (result?.deletedAt) { - store.throwSoftDeletedWriteBlocked(id, result.deletedAt, options?.auditInput?.operation ?? "applyTaskPatch", { - agentId: options?.auditInput?.agentId, - runId: options?.auditInput?.runId, - timestamp: options?.auditInput?.timestamp, - }); - } - await store.writeTaskJsonFile(dir, result?.current ?? task); -} +/* +FNXC:PostgresOnlyDataAccess 2026-07-17-15:10: +`applyTaskPatchImpl` (the low-level sync SQLite column-patch primitive) was +removed: it had zero callers in either mode, and its `store.db.transactionImmediate` ++ `patchTaskRowInTransaction` body only ran against the deleted SQLite runtime. +Task writes go through the async persistence helpers (upsertTaskRowInTransaction / +updateTaskColumns). The public `TaskStore.applyTaskPatch` facade was removed with it. +*/ export function readTaskFromDbImpl(store: TaskStore, id: string, options?: { activityLogLimit?: number; includeDeleted?: boolean }): Task | undefined { const selectClause = options?.activityLogLimit diff --git a/packages/core/src/task-store/remaining-ops-6.ts b/packages/core/src/task-store/remaining-ops-6.ts index 619fb8e7e2..550ed51a64 100644 --- a/packages/core/src/task-store/remaining-ops-6.ts +++ b/packages/core/src/task-store/remaining-ops-6.ts @@ -937,6 +937,15 @@ export function getWorkflowSettingsProjectIdImpl(store: TaskStore): string { */ const boundProjectId = store.asyncLayer?.projectId; if (boundProjectId) return boundProjectId; + /* + FNXC:PostgresWorkflowSettings 2026-07-17-14:20: + An unscoped backend store (asyncLayer present but projectId empty) must NOT + reach `store.db` below — it throws the removed-SQLite stub, which the catch + then swallows, so the throw was invisible. Return the same rootDir key the + swallow produced, without the spurious stub throw. Only the true legacy + (non-backend) path consults the SQLite identity. + */ + if (store.backendMode) return store.rootDir; try { return store.db.getProjectIdentity()?.id ?? store.rootDir; } catch { diff --git a/packages/core/src/task-store/remaining-ops-8.ts b/packages/core/src/task-store/remaining-ops-8.ts index 515e1bd2b0..9994438779 100644 --- a/packages/core/src/task-store/remaining-ops-8.ts +++ b/packages/core/src/task-store/remaining-ops-8.ts @@ -656,8 +656,26 @@ export async function purgeTaskWorkflowSelectionRowsAsyncImpl(store: TaskStore, store.workflowStepsCache = null; } -export function cleanupOrphanedMaterializedStepsImpl(store: TaskStore, stepIds: string[] | undefined): void { +export async function cleanupOrphanedMaterializedStepsImpl(store: TaskStore, stepIds: string[] | undefined): Promise { if (!stepIds || stepIds.length === 0) return; + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + Backend mode deletes the orphaned workflow_steps rows via the AsyncDataLayer. + The former sync `store.db.prepare(DELETE...)` threw the removed-SQLite stub in + backend mode and was swallowed by the best-effort catch, silently leaking every + materialized workflow_steps row created before a failed task-create (the callers + are the task-creation failure catches). Mirrors removeMaterializedSelectionImpl. + */ + const layer = store.getAsyncLayer(); + if (layer) { + try { + await layer.db.delete(schema.project.workflowSteps).where(inArray(schema.project.workflowSteps.id, stepIds)); + } catch { + // Best-effort cleanup. + } + store.workflowStepsCache = null; + return; + } for (const stepId of stepIds) { try { store.db.prepare("DELETE FROM workflow_steps WHERE id = ?").run(stepId); diff --git a/packages/core/src/task-store/task-creation.ts b/packages/core/src/task-store/task-creation.ts index ebd6f4eac3..674dc5e548 100644 --- a/packages/core/src/task-store/task-creation.ts +++ b/packages/core/src/task-store/task-creation.ts @@ -597,7 +597,7 @@ export async function createTaskImpl(store: TaskStore, input: TaskCreateInput, o } catch (err) { // The task row was never created, so any default-workflow steps we // materialized above would orphan with no task/selection pointing at them. - store.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); + await store.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); throw err; } @@ -800,7 +800,7 @@ export async function createTaskWithReservedIdImpl(store: TaskStore, input: Task } catch (err) { // The task row was never created, so any default-workflow steps we // materialized above would orphan with no task/selection pointing at them. - store.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); + await store.cleanupOrphanedMaterializedSteps(pendingWorkflowSelection?.stepIds); throw err; } diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index fe909f6930..24f0122684 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -4861,7 +4861,26 @@ export class TaskExecutor { Task execution sessions must honor the assigned permanent agent's runtimeConfig like chat sessions do. If the live executor was handed an agents-less worktree AgentStore, fall back to the authoritative project `.fusion` AgentStore instead of letting `resolveExecutorSessionModel` see an empty runtimeConfig and silently drift to the pi built-in model. */ try { - this.authoritativeAssignedAgentStore ??= new AgentStore({ rootDir: join(this.rootDir, ".fusion"), taskStore: this.store }); + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-14:20: + The authoritative-agent fallback AgentStore MUST inherit the TaskStore's AsyncDataLayer so it runs in PostgreSQL backend mode. AgentStore does not derive `asyncLayer` from `taskStore`, so omitting it left this store in legacy-SQLite mode; in a PG deployment `init()`/`getAgent()` then hit the removed SQLite stub, the throw was swallowed by the catch below, and this method silently returned null — reintroducing the exact model-drift to the pi built-in that this fallback exists to prevent. Pass the layer (mirrors the canonical site in agent-tools.ts). + */ + const authoritativeAgentLayer = this.store.getAsyncLayer(); + /* + FNXC:PostgresOnlyDataAccess 2026-07-17-16:10: + Do NOT memoize a layer-less AgentStore. If the very first lookup runs before + the TaskStore's AsyncDataLayer is attached, a plain `??=` would cache a + legacy-SQLite-mode store forever, so every later call keeps failing through the + removed SQLite path even after the layer arrives. Rebuild when a layer is now + available but the cached store is not in backend mode. + */ + if (!this.authoritativeAssignedAgentStore || (authoritativeAgentLayer && !this.authoritativeAssignedAgentStore.backendMode)) { + this.authoritativeAssignedAgentStore = new AgentStore({ + rootDir: join(this.rootDir, ".fusion"), + taskStore: this.store, + ...(authoritativeAgentLayer ? { asyncLayer: authoritativeAgentLayer } : {}), + }); + } await this.authoritativeAssignedAgentStore.init(); return await this.authoritativeAssignedAgentStore.getAgent(normalizedId).catch(() => null); } catch (err: unknown) { diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index c0b3f23afe..4dbd9a9013 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -2584,7 +2584,7 @@ export class SelfHealingManager { log.log("Maintenance batch 1 step \"prune-agent-log-files\" skipped — agentLogFileRetentionDays is not enabled"); return; } - const { prunedFiles, prunedEntries, freedBytes } = this.store.pruneAgentLogFiles(days); + const { prunedFiles, prunedEntries, freedBytes } = await this.store.pruneAgentLogFilesAsync(days); log.log(`Maintenance batch 1 step "prune-agent-log-files" succeeded — files=${prunedFiles} entries=${prunedEntries} bytes=${freedBytes}`); }, },