From 2b55077546a7d3a6a54dd6c3fbf9e487b96e82b7 Mon Sep 17 00:00:00 2001 From: gsxdsm Date: Sun, 26 Jul 2026 13:10:20 -0700 Subject: [PATCH] fix: wire incomplete PostgreSQL ports for archive, reconcile, health Replace empty backendMode stubs with real AsyncDataLayer paths: archive ID reservation and isTaskArchivedAsync, orphaned task.json re-import, health snapshots via checkPostgresHealth, settings/agent memory caches for sync readers, async builtin prompt overrides, and self-healing audit/health callers that previously used dead sync SQLite fallbacks. --- .../postgres/agent-wake-getagent.pg.test.ts | 8 +- .../postgres/incomplete-pg-ports.pg.test.ts | 121 +++++++++++++++++ ...sqlite-production-reader-inventory.test.ts | 64 +++++++-- packages/core/src/agent-store.ts | 30 ++++- packages/core/src/store.ts | 43 +++++- packages/core/src/task-store/audit-ops.ts | 3 +- packages/core/src/task-store/lifecycle-ops.ts | 123 ++++++++++++------ packages/core/src/task-store/moves.ts | 9 +- .../core/src/task-store/settings-ops-2.ts | 10 +- .../core/src/task-store/task-id-integrity.ts | 88 ++++++++++--- .../core/src/task-store/task-store-helpers.ts | 57 +++++++- .../src/task-store/workflow-definitions.ts | 45 +++---- packages/engine/src/self-healing.ts | 32 ++++- 13 files changed, 506 insertions(+), 127 deletions(-) create mode 100644 packages/core/src/__tests__/postgres/incomplete-pg-ports.pg.test.ts diff --git a/packages/core/src/__tests__/postgres/agent-wake-getagent.pg.test.ts b/packages/core/src/__tests__/postgres/agent-wake-getagent.pg.test.ts index 3d280ced32..7efd3d74d2 100644 --- a/packages/core/src/__tests__/postgres/agent-wake-getagent.pg.test.ts +++ b/packages/core/src/__tests__/postgres/agent-wake-getagent.pg.test.ts @@ -64,13 +64,13 @@ pgTest("AgentStore.getAgent backs the async wake hook (PostgreSQL)", () => { it("async getAgent returns null for an unknown recipient (hook early-returns)", async () => { expect(await agentStore.getAgent("agent-does-not-exist")).toBeNull(); }); - it("getCachedAgent returns null in PG backend mode (sync SQLite fallback)", async () => { + it("getCachedAgent returns the agent after getAgent warms the PG memory cache", async () => { const created = await agentStore.createAgent({ name: "cached-null-target", role: "executor" }); - // Sync read has no DB handle in PG mode — degrades to null by design. - // Async callers route through getAgent() instead (proven by the test above). + // FNXC:IncompletePgPorts 2026-07-26-20:45: cold cache is null until getAgent loads PG. expect(agentStore.getCachedAgent(created.id)).toBeNull(); - // The async path resolves the same agent the sync path cannot reach. expect((await agentStore.getAgent(created.id))?.id).toBe(created.id); + // Warm cache serves sync heartbeat resolveAgentConfig without SQLite. + expect(agentStore.getCachedAgent(created.id)?.id).toBe(created.id); }); }); diff --git a/packages/core/src/__tests__/postgres/incomplete-pg-ports.pg.test.ts b/packages/core/src/__tests__/postgres/incomplete-pg-ports.pg.test.ts new file mode 100644 index 0000000000..594bb20bf5 --- /dev/null +++ b/packages/core/src/__tests__/postgres/incomplete-pg-ports.pg.test.ts @@ -0,0 +1,121 @@ +/** + * FNXC:IncompletePgPorts 2026-07-26-20:50: + * End-to-end PostgreSQL coverage for incomplete-port fixes: archive ID + * reservation, isTaskArchivedAsync, orphaned task.json reconcile, health + * snapshot refresh, settingsSyncCache, and prompt-override async load. + */ +import { it, expect, beforeAll, beforeEach, afterEach, afterAll } from "vitest"; +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { + pgDescribe, + createSharedPgTaskStoreTestHarness, + type SharedPgTaskStoreHarness, +} from "../../__test-utils__/pg-test-harness.js"; +import * as schema from "../../postgres/schema/index.js"; + +const pgTest = pgDescribe; + +pgTest("incomplete PG ports (archive, reconcile, health, settings)", () => { + const h: SharedPgTaskStoreHarness = createSharedPgTaskStoreTestHarness({ + prefix: "fusion_incomplete_pg_ports", + }); + + beforeAll(h.beforeAll); + beforeEach(h.beforeEach); + afterEach(h.afterEach); + afterAll(h.afterAll); + + it("taskIdExistsAnywhere includes cold archive.archived_tasks IDs", async () => { + const store = h.store(); + const layer = h.layer(); + const projectId = layer.projectId?.trim() || "__legacy_unscoped__"; + const task = await store.createTask({ description: "Cold archive representation" }); + // Remove live row so only cold archive remains. + await layer.db.delete(schema.project.tasks); + await layer.db.insert(schema.archive.archivedTasks).values({ + id: task.id, + projectId, + taskJson: JSON.stringify(task), + archivedAt: new Date().toISOString(), + title: task.title, + description: task.description, + createdAt: task.createdAt, + updatedAt: task.updatedAt, + }); + expect(await store.taskIdExistsAnywhere(task.id)).toBe(true); + expect(await store.isTaskIdPresentInArchivedTasksTableAsync(task.id)).toBe(true); + }); + + it("isTaskArchivedAsync is true after archiveTask", async () => { + const store = h.store(); + const task = await store.createTask({ description: "to-archive" }); + await store.archiveTask(task.id); + expect(await store.isTaskArchivedAsync(task.id)).toBe(true); + expect(await store.isTaskArchivedAsync("FN-MISSING-ARCHIVE")).toBe(false); + }); + + it("reconcileOrphanedTaskDirs re-imports a task.json missing from PostgreSQL", async () => { + const store = h.store(); + const id = "FN-ORPHAN-1"; + const taskDir = join(store.tasksDir, id); + await mkdir(taskDir, { recursive: true }); + const now = new Date().toISOString(); + const task = { + id, + title: "orphan recover", + description: "from disk", + column: "todo", + status: "todo", + createdAt: now, + updatedAt: now, + dependencies: [], + comments: [], + priority: "medium", + }; + await writeFile(join(taskDir, "task.json"), JSON.stringify(task, null, 2), "utf8"); + + expect(await store.taskIdExistsAnywhere(id)).toBe(false); + const result = await store.reconcileOrphanedTaskDirs({ ignoreRecencyWindow: true }); + expect(result.recovered).toContain(id); + expect(await store.taskIdExistsAnywhere(id)).toBe(true); + const live = await store.getTask(id); + expect(live?.title).toBe("orphan recover"); + }); + + it("refreshDatabaseHealthAsync records a healthy postgresHealthSnapshot", async () => { + const store = h.store(); + expect(store.postgresHealthSnapshot).toBeNull(); + const health = await store.refreshDatabaseHealthAsync(); + expect(health.healthy).toBe(true); + expect(health.corruptionDetected).toBe(false); + expect(health.lastCheckedAt).toBeInstanceOf(Date); + expect(store.getDatabaseHealth().healthy).toBe(true); + expect(store.healthCheck()).toBe(true); + }); + + it("getSettings populates settingsSyncCache for getSettingsSync", async () => { + const store = h.store(); + expect(store.getSettingsSync()).toBeTruthy(); + await store.getSettings(); + expect(store.settingsSyncCache).not.toBeNull(); + expect(store.getSettingsSync()).toEqual(store.settingsSyncCache); + }); + + it("applyBuiltInPromptOverridesAsync loads PostgreSQL prompt overrides", async () => { + const store = h.store(); + const projectId = store.getWorkflowSettingsProjectId(); + const workflowId = "builtin:coding"; + await store.updateWorkflowPromptOverrides(workflowId, projectId, { + plan: "CUSTOM_PLAN_PROMPT_OVERRIDE", + }); + // Direct async helper must load the row. + const loaded = await store.getWorkflowPromptOverridesAsync(workflowId, projectId); + expect(loaded.plan).toBe("CUSTOM_PLAN_PROMPT_OVERRIDE"); + const def = await store.getWorkflowDefinition(workflowId); + expect(def).toBeTruthy(); + // Builtin IR after applyBuiltInPromptOverridesAsync should include the custom plan prompt. + const irText = JSON.stringify(def!.ir); + expect(irText).toContain("CUSTOM_PLAN_PROMPT_OVERRIDE"); + }); +}); diff --git a/packages/core/src/__tests__/postgres/sqlite-production-reader-inventory.test.ts b/packages/core/src/__tests__/postgres/sqlite-production-reader-inventory.test.ts index 38ce6d9522..bd89ea3867 100644 --- a/packages/core/src/__tests__/postgres/sqlite-production-reader-inventory.test.ts +++ b/packages/core/src/__tests__/postgres/sqlite-production-reader-inventory.test.ts @@ -161,11 +161,14 @@ describe("incomplete PG sync-reader stubs (shipped helpers)", () => { PostgreSQL is async-only. Drive the real exported impls so a future “fix” that reintroduces store.db.prepare on these paths fails loudly. */ - function backendModeStore(): TaskStore { + function backendModeStore(overrides: Partial = {}): TaskStore { // Minimal TaskStore shape: backendMode true must never touch store.db. return { backendMode: true, asyncLayer: { projectId: "proj_test" }, + taskCache: new Map(), + postgresHealthSnapshot: null, + settingsSyncCache: null, db: { prepare() { throw new Error("SQLite Database must not be consulted in backend mode"); @@ -176,6 +179,14 @@ describe("incomplete PG sync-reader stubs (shipped helpers)", () => { throw new Error("SQLite ArchiveDatabase must not be consulted in backend mode"); }, }, + refreshDatabaseHealthAsync: async () => ({ + healthy: true, + corruptionDetected: false, + corruptionErrors: [], + lastCheckedAt: null, + isRunning: false, + }), + ...overrides, } as unknown as TaskStore; } @@ -183,8 +194,11 @@ describe("incomplete PG sync-reader stubs (shipped helpers)", () => { expect(isTaskIdPresentInArchivedTasksTableImpl(backendModeStore(), "FN-9999")).toBe(false); }); - it("isTaskArchivedImpl returns false under backend mode without opening SQLite", () => { + it("isTaskArchivedImpl uses taskCache under backend mode without opening SQLite", () => { expect(isTaskArchivedImpl(backendModeStore(), "FN-9999")).toBe(false); + const store = backendModeStore(); + store.taskCache.set("FN-ARCH", { id: "FN-ARCH", column: "archived" } as never); + expect(isTaskArchivedImpl(store, "FN-ARCH")).toBe(true); }); it("getMergeRequestRecordImpl returns null under backend mode (sync callers must use Async sibling)", () => { @@ -217,32 +231,54 @@ describe("incomplete PG sync-reader stubs (shipped helpers)", () => { expect(typeof settings).toBe("object"); }); - it("healthCheckImpl returns true under backend mode (real health is AsyncDataLayer.ping)", () => { - expect(healthCheckImpl(backendModeStore())).toBe(true); + it("healthCheckImpl reports postgresHealthSnapshot under backend mode", () => { + const store = backendModeStore({ + postgresHealthSnapshot: { + healthy: false, + corruptionDetected: true, + corruptionErrors: ["PostgreSQL backend unreachable: boom"], + lastCheckedAt: new Date("2026-07-26T00:00:00.000Z"), + isRunning: false, + }, + getDatabaseHealth: undefined as never, + }); + store.getDatabaseHealth = () => getDatabaseHealthImpl(store); + expect(healthCheckImpl(store)).toBe(false); }); - it("getDatabaseHealthImpl returns always-healthy sentinel under backend mode without opening SQLite", () => { - const health = getDatabaseHealthImpl(backendModeStore()); + it("getDatabaseHealthImpl returns cached postgresHealthSnapshot under backend mode", () => { + const checkedAt = new Date("2026-07-26T12:00:00.000Z"); + const health = getDatabaseHealthImpl(backendModeStore({ + postgresHealthSnapshot: { + healthy: false, + corruptionDetected: true, + corruptionErrors: ["unreachable"], + lastCheckedAt: checkedAt, + isRunning: false, + }, + })); expect(health).toEqual({ - healthy: true, - corruptionDetected: false, - corruptionErrors: [], - lastCheckedAt: null, + healthy: false, + corruptionDetected: true, + corruptionErrors: ["unreachable"], + lastCheckedAt: checkedAt, isRunning: false, }); }); - it("refreshDatabaseHealthImpl delegates to healthy sentinel under backend mode (no integrity_check)", () => { + it("refreshDatabaseHealthImpl schedules async refresh and returns current snapshot", () => { const store = backendModeStore(); - // TaskStore.refreshDatabaseHealth / getDatabaseHealth are methods that call the impls. store.getDatabaseHealth = () => getDatabaseHealthImpl(store); const health = refreshDatabaseHealthImpl(store); expect(health.healthy).toBe(true); expect(health.corruptionDetected).toBe(false); }); - it("reconcileOrphanedTaskDirsImpl returns empty recovered/skipped under backend mode (PG self-healing no-op)", async () => { - const result = await reconcileOrphanedTaskDirsImpl(backendModeStore(), {}); + it("reconcileOrphanedTaskDirsImpl no-ops when tasksDir is missing under backend mode", async () => { + const store = backendModeStore({ + tasksDir: "/nonexistent/fusion-tasks-dir-for-inventory", + }); + const result = await reconcileOrphanedTaskDirsImpl(store, {}); expect(result).toEqual({ recovered: [], skipped: [] }); }); }); diff --git a/packages/core/src/agent-store.ts b/packages/core/src/agent-store.ts index 1329761197..8d2e9f0440 100644 --- a/packages/core/src/agent-store.ts +++ b/packages/core/src/agent-store.ts @@ -305,6 +305,13 @@ export class AgentStore extends EventEmitter { */ public readonly asyncLayer: AsyncDataLayer | null = null; + /* + FNXC:IncompletePgPorts 2026-07-26-20:40: + In-memory agent rows filled by getAgent() so getCachedAgent can serve sync + heartbeat config resolution under PostgreSQL (no sync SQLite handle). + */ + private readonly agentMemoryCache = new Map(); + /** True when AsyncDataLayer was injected. Gates all SQLite construction. */ public get backendMode(): boolean { return this.asyncLayer !== null; @@ -827,7 +834,15 @@ export class AgentStore extends EventEmitter { // Backend mode: read via async Drizzle helper instead of sync readAgent. if (this.backendMode) { const agent = await readAgentAsync(this.asyncLayer!.db, agentId); - return agent ? this.parseAgent(agent) : null; + const parsed = agent ? this.parseAgent(agent) : null; + /* + FNXC:IncompletePgPorts 2026-07-26-20:40: + Populate getCachedAgent memory so sync heartbeat resolveAgentConfig can + honor per-agent runtimeConfig without a sync SQLite handle. + */ + if (parsed) this.agentMemoryCache.set(agentId, parsed); + else this.agentMemoryCache.delete(agentId); + return parsed; } return this.readAgent(agentId); } @@ -3187,13 +3202,14 @@ export class AgentStore extends EventEmitter { * @param agentId - The agent ID */ getCachedAgent(agentId: string): Agent | null { - // SQLite sync fast-path. In PG backend mode there is no sync DB handle, so - // this returns null. Both production callers (HeartbeatMonitor's sync - // resolveAgentConfig and the reports-health interval resolver) wrap this in - // try/catch and degrade to monitor defaults; the async getAgentConfig() path - // does its own async getAgent() lookup, so per-agent runtimeConfig is honored. + /* + FNXC:IncompletePgPorts 2026-07-26-20:40: + PostgreSQL path: return agentMemoryCache populated by getAgent(). Still + null until the first async getAgent warms the cache (heartbeat timer path + uses async getAgentConfig as authority). + */ if (this.backendMode) { - return null; + return this.agentMemoryCache.get(agentId) ?? null; } return this.readAgent(agentId); } diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 675e7cf681..95299c1d79 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -98,13 +98,13 @@ 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, moveTaskIfImpl, handoffToReviewImpl, moveTaskInternalImpl, type MoveTaskIfResult } 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/workflow-task-create-ops.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/task-id-integrity.js"; +import { applyLegacyWorkflowStepOverridesImpl, archiveDbImpl, assertNoDependencyCycleImpl, atomicCreateTaskJsonImpl, buildActiveTaskDependencyLookupImpl, buildArchivedAgentLogFieldsImpl, buildTaskIdIntegrityFallbackReportImpl, createBranchGroupImpl, dbImpl, detectAndCacheTaskIdIntegrityReportImpl, findLiveDependentsImpl, findLiveLineageChildrenImpl, getLegacyWorkflowStepSnapshotImpl, getMalformedTaskMetadataReasonImpl, getMergeQueuedTaskIdsAsyncImpl, insertRunAuditEventRowImpl, insertTaskImpl, invokeTaskCreatedHookImpl, isTaskArchivedAsyncImpl, isTaskArchivedImpl, isTaskIdPresentInArchivedTasksTableAsyncImpl, 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/task-id-integrity.js"; import { claimNextToolFailureRetryImpl, createTaskVerificationRequestImpl, claimTaskVerificationRequestImpl, finishTaskVerificationRequestImpl, 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/branch-and-pr-entities.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/task-artifacts-ops.js"; import { approveCliAutonomyImpl, approveWorkflowCliCommandImpl, cleanupOrphanedMaterializedStepsImpl, consumePluginGateVerdictsImpl, getAgentLogsByTimeRangeImpl, getDatabaseHealthImpl, getDistributedTaskIdAllocatorImpl, getExperimentSessionStoreImpl, getInReviewDurationEventsImpl, getMissionStoreImpl, getIdeationStoreImpl, 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/workflow-definitions.js"; import { getTaskCommitAssociationsByLineageIdImpl, replaceLegacyTaskCommitAssociationsImpl } from "./task-store/task-commit-associations.js"; import { findRecentTasksBySourceParentTaskIdImpl } from "./task-store/branch-and-pr-entities.js"; -import { addTaskCommentImpl, applyBuiltInPromptOverridesSyncImpl, areAllDependenciesDoneImpl, artifactStoredNameImpl, assertWorkflowIrTraitsValidImpl, clearActivityLogImpl, clearTaskWorkflowSelectionImpl, deleteTaskByIdImpl, getDefaultWorkflowIdImpl, getInsightStoreImpl, getMergeQueuedTaskIdsImpl, getMergeRequestRecordImpl, getMergeRequestRecordAsyncImpl, getResearchStoreImpl, getTaskIdFromDirImpl, getTodoStoreImpl, getWorkflowWorkItemByIdentityImpl, hasActiveTaskImpl, invalidateConfigCacheAfterMigrationImpl, isTaskIdConflictErrorImpl, listLegacyAutoMergeStampCandidatesImpl, readTaskRowFromDbImpl, recordBranchGroupMemberLandedImpl, refreshDatabaseHealthImpl, resolveEffectiveWorkflowIdSyncImpl, resolveTaskCustomFieldDefsSyncImpl, resolveWorkflowBypassGuardsImpl, serializeConfigForDiskImpl, setPluginWorkflowStepTemplatesImpl, shouldSkipWorkflowMovePoliciesImpl, suppressWatcherImpl, upsertTaskWithFtsRecoveryImpl } from "./task-store/task-store-helpers.js"; +import { addTaskCommentImpl, applyBuiltInPromptOverridesAsyncImpl, applyBuiltInPromptOverridesSyncImpl, areAllDependenciesDoneImpl, artifactStoredNameImpl, assertWorkflowIrTraitsValidImpl, clearActivityLogImpl, clearTaskWorkflowSelectionImpl, deleteTaskByIdImpl, getDefaultWorkflowIdImpl, getInsightStoreImpl, getMergeQueuedTaskIdsImpl, getMergeRequestRecordImpl, getMergeRequestRecordAsyncImpl, getResearchStoreImpl, getTaskIdFromDirImpl, getTodoStoreImpl, getWorkflowWorkItemByIdentityImpl, hasActiveTaskImpl, invalidateConfigCacheAfterMigrationImpl, isTaskIdConflictErrorImpl, listLegacyAutoMergeStampCandidatesImpl, readTaskRowFromDbImpl, recordBranchGroupMemberLandedImpl, refreshDatabaseHealthAsyncImpl, refreshDatabaseHealthImpl, resolveEffectiveWorkflowIdSyncImpl, resolveTaskCustomFieldDefsSyncImpl, resolveWorkflowBypassGuardsImpl, serializeConfigForDiskImpl, setPluginWorkflowStepTemplatesImpl, shouldSkipWorkflowMovePoliciesImpl, suppressWatcherImpl, upsertTaskWithFtsRecoveryImpl } from "./task-store/task-store-helpers.js"; import { getTaskSelectClauseImpl2, createTaskPersistSerializationContextImpl, getTaskPersistValuesImpl, getTaskPatchDescriptorsImpl, normalizeTaskFromDiskImpl, writeTaskJsonFileImpl, rowToPrEntityImpl, generatePrEntityIdImpl, readTaskForMoveImpl, rowToMergeQueueEntryImpl, rowToMergeRequestRecordImpl, rowToCompletionHandoffMarkerImpl, rowToWorkflowWorkItemImpl, rowToRunAuditEventImpl } from "./task-store/task-row-mappers.js"; import { getTaskSelectClauseWithActivityLogLimitImpl, getChangedTaskColumnsImpl, getSoftDeletedWriteConflictImpl, readTaskJsonImpl, writeConfigImpl, _maybeAutoArchiveSameAgentDuplicateBackendImpl, updateBranchGroupImpl, updatePrEntityImpl, listTasksForGithubTrackingReconcileImpl, listTasksForGitlabTrackingReconcileImpl, renewCheckoutLeaseImpl, updateTaskAtomicImpl, getWorkflowPromptOverridesImpl, updateWorkflowSettingValuesImpl, rollbackConfigurationImpl, cancelActiveWorkflowWorkItemsForTaskImpl, setCompletionHandoffAcceptedMarkerImpl, reconcileLegacyAutoMergeStampsImpl, recoverExpiredMergeQueueLeasesImpl, rewriteDependentsForRemovalImpl, cleanupBranchForTaskImpl, addAttachmentImpl, deleteAttachmentImpl, registerArtifactImpl, updatePrInfoImpl, unlinkGithubIssueImpl, cleanupArchivedTasksImpl, generatePromptFromArchiveEntryImpl, listWorkflowOccupantTaskIdsImpl, evacuateCustomColumnsToLegacyImpl, listApprovedCliAutonomyAdaptersImpl, closeImpl, getActivityLogImpl } from "./task-store/task-mutation-ops.js"; import { getOrCreateForProjectImpl, listGoalCitationsImpl, atomicWriteTaskJsonWithAuditImpl, duplicateTaskImpl, listStrandedRefinementsImpl, tryClaimCheckoutImpl, evaluateWorkflowMovePoliciesImpl, recordRunAuditEventImpl, getRunAuditEventsImpl, getWorkflowParitySummaryImpl, dequeueMergeQueueOnColumnExitImpl, updateIssueInfoImpl, listWorkflowStepsImpl, getWorkflowStepImpl, createWorkflowDefinitionImpl, countActiveInCapacitySlotSyncImpl, countActiveInCapacitySlotAsyncImpl, generateSpecifiedPromptImpl, recordActivityImpl, getEvalStoreImpl } from "./task-store/project-store-ops.js"; @@ -363,6 +363,26 @@ export class TaskStore extends EventEmitter { public watcher: FSWatcher | null = null; public taskCache: Map = new Map(); + /* + FNXC:IncompletePgPorts 2026-07-26-20:35: + Sync getDatabaseHealth/healthCheck cannot await PostgreSQL. Cache the last + AsyncDataLayer ping result so self-healing and CLI health probes observe real + connectivity rather than the always-healthy cutover sentinel. + */ + public postgresHealthSnapshot: { + healthy: boolean; + corruptionDetected: boolean; + corruptionErrors: string[]; + lastCheckedAt: Date | null; + isRunning: boolean; + } | null = null; + /* + FNXC:IncompletePgPorts 2026-07-26-20:35: + Sync getSettingsSync (generateSpecifiedPrompt ntfy section) uses the last + merged settings from getSettings/getSettingsFast so PG mode is not stuck on + DEFAULT_SETTINGS after a successful async load. + */ + public settingsSyncCache: Settings | null = null; /** U8 (KTD-2): pre-evaluated plugin gate verdicts, keyed `taskId` → `toColumn` */ public pluginGateVerdicts: Map> = new Map(); public recentlyWritten: Set = new Set(); @@ -640,6 +660,10 @@ export class TaskStore extends EventEmitter { public isTaskIdPresentInArchivedTasksTable(id: string): boolean { return isTaskIdPresentInArchivedTasksTableImpl(this, id); } + /** PostgreSQL-authoritative archive presence check (warm + cold archive tables). */ + public async isTaskIdPresentInArchivedTasksTableAsync(id: string): Promise { + return isTaskIdPresentInArchivedTasksTableAsyncImpl(this, id); + } public async taskIdExistsAnywhere(id: string): Promise { return taskIdExistsAnywhereImpl(this, id); } @@ -652,6 +676,10 @@ export class TaskStore extends EventEmitter { public isTaskArchived(id: string): boolean { return isTaskArchivedImpl(this, id); } + /** PostgreSQL-authoritative archived check (live column + cold archive). */ + public async isTaskArchivedAsync(id: string): Promise { + return isTaskArchivedAsyncImpl(this, id); + } public findLiveDependents(id: string): string[] { return findLiveDependentsImpl(this, id); } @@ -2359,6 +2387,10 @@ Issue #2149 requires read-only type filtering to occur in the file-store before public applyBuiltInPromptOverridesSync(workflowId: string, ir: WorkflowIr): WorkflowIr { return applyBuiltInPromptOverridesSyncImpl(this, workflowId, ir); } + /** PostgreSQL-aware prompt override application (loads overrides via Drizzle). */ + public async applyBuiltInPromptOverridesAsync(workflowId: string, ir: WorkflowIr): Promise { + return applyBuiltInPromptOverridesAsyncImpl(this, workflowId, ir); + } /** Get a single workflow definition by id, or undefined when absent. */ async getWorkflowDefinition( id: string, ): Promise { @@ -2672,6 +2704,13 @@ Issue #2149 requires read-only type filtering to occur in the file-store before refreshDatabaseHealth(): ReturnType { return refreshDatabaseHealthImpl(this); } + /** + * FNXC:IncompletePgPorts 2026-07-26-20:35: + * Refresh PostgreSQL connectivity into postgresHealthSnapshot, then return it. + */ + async refreshDatabaseHealthAsync(): Promise> { + return refreshDatabaseHealthAsyncImpl(this); + } getDistributedTaskIdAllocator(): DistributedTaskIdAllocator { return getDistributedTaskIdAllocatorImpl(this); } diff --git a/packages/core/src/task-store/audit-ops.ts b/packages/core/src/task-store/audit-ops.ts index 52669e43e5..dc5be98253 100644 --- a/packages/core/src/task-store/audit-ops.ts +++ b/packages/core/src/task-store/audit-ops.ts @@ -136,8 +136,7 @@ export async function logEntryImpl(store: TaskStore, id: string, action: string, const state = await getLiveTaskColumn(layer.db, id, layer.projectId); if (state === "archived") throw new Error(`Task ${id} is archived — logging is read-only`); if (state === null) throw new Error(`Task ${id} not found`); - } - if (store.isTaskArchived(id)) { + } else if (store.isTaskArchived(id)) { throw new Error(`Task ${id} is archived — logging is read-only`); } diff --git a/packages/core/src/task-store/lifecycle-ops.ts b/packages/core/src/task-store/lifecycle-ops.ts index 70dfef6fc0..c18c47b0d1 100644 --- a/packages/core/src/task-store/lifecycle-ops.ts +++ b/packages/core/src/task-store/lifecycle-ops.ts @@ -35,6 +35,9 @@ import {getErrorMessage} from "../error-message.js"; import {type TaskRow} from "../task-store/persistence.js"; import {__setTaskActivityLogLimitsForTesting} from "../task-store/comments.js"; import {reconcileTaskIdStateAsync} from "../task-store/async-allocator.js"; +import {ACTIVE_TASK_FILTER, insertTaskRow, isTaskIdConflictError as isPgTaskIdConflictError} from "./async-persistence.js"; +import {recordRunAuditEvent as recordRunAuditEventAsync} from "./async-audit.js"; +import * as schema from "../postgres/schema/index.js"; export async function initImpl(store: TaskStore): Promise { store.closing = false; @@ -586,18 +589,11 @@ export function setupActivityLogListenersImpl(store: TaskStore): void { export async function reconcileOrphanedTaskDirsImpl(store: TaskStore, opts: { ignoreRecencyWindow?: boolean } = {},): Promise<{ recovered: string[]; skipped: Array<{ id: string; reason: string }> }> { /* - FNXC:PostgresCutover 2026-07-04-00:00: - Assessed safe-default: in PG backend mode, the sync filesystem scan + store.db re-insert - path cannot run (Drizzle is async, store.db is removed). The self-healing caller (line 2302) - receives an empty result — orphaned task dirs are NOT reconciled in PG mode. This is low-risk - because PG soft-delete is the norm (task.json dirs persist for active tasks; deleted tasks - keep their dirs but are tombstoned in PG, not lost). A full async reconcile (scan dirs, - check PG for matching rows, re-import missing) is feasible but not P0 given the rarity of - PG-mode orphans. Not claiming a non-existent async fallback. + FNXC:IncompletePgPorts 2026-07-26-20:45: + PostgreSQL path: scan task.json dirs and re-insert missing IDs via insertTaskRow + after taskIdExistsAnywhere (live + archive). Same recency/empty-board gates as + the SQLite path; previously backendMode returned empty and never recovered. */ - if (store.backendMode) { - return { recovered: [], skipped: [] }; - } const result: { recovered: string[]; skipped: Array<{ id: string; reason: string }> } = { recovered: [], skipped: [], @@ -616,10 +612,20 @@ export async function reconcileOrphanedTaskDirsImpl(store: TaskStore, opts: { ig // the same guard added to stop resurrection. Callers may also force the bypass explicitly. let dbHasLiveTasks = true; try { - const row = store.db - .prepare('SELECT EXISTS(SELECT 1 FROM tasks WHERE deletedAt IS NULL LIMIT 1) AS present') - .get() as { present?: number } | undefined; - dbHasLiveTasks = (row?.present ?? 0) === 1; + if (store.backendMode) { + const layer = store.asyncLayer!; + const rows = await layer.db + .select({ id: schema.project.tasks.id }) + .from(schema.project.tasks) + .where(ACTIVE_TASK_FILTER) + .limit(1); + dbHasLiveTasks = rows.length > 0; + } else { + const row = store.db + .prepare('SELECT EXISTS(SELECT 1 FROM tasks WHERE deletedAt IS NULL LIMIT 1) AS present') + .get() as { present?: number } | undefined; + dbHasLiveTasks = (row?.present ?? 0) === 1; + } } catch { // If the count probe fails, keep the gate on (conservative — don't mass-resurrect). dbHasLiveTasks = true; @@ -711,36 +717,68 @@ export async function reconcileOrphanedTaskDirsImpl(store: TaskStore, opts: { ig let recovered = false; let skipReason: string | undefined; try { - store.db.transactionImmediate(() => { - // FNXC:SqliteFinalRemoval 2026-06-26: taskIdExistsAnywhere is now async; - // inline the sync SQLite check here since this runs inside transactionImmediate. - if (store.readTaskFromDb(id, { includeDeleted: true }) || store.isTaskIdPresentInArchivedTasksTable(id) || store.archiveDb.get(id) !== undefined) { + if (store.backendMode) { + if (await store.taskIdExistsAnywhere(id)) { skipReason = "id-exists-anywhere"; - return; + } else { + try { + const context = store.createTaskPersistSerializationContext(task); + await insertTaskRow(store.asyncLayer!, task as unknown as Record, context); + await recordRunAuditEventAsync(store.asyncLayer!, { + taskId: id, + agentId: "system", + runId: "unknown", + domain: "database", + mutationType: "task:reconcile-orphaned-task-dir", + target: id, + metadata: { + id, + column: task.column, + status: task.status ?? null, + taskJsonPath, + }, + }); + recovered = true; + } catch (error) { + if (isPgTaskIdConflictError(error) || /Task ID already exists/i.test(error instanceof Error ? error.message : String(error))) { + skipReason = "id-conflict-during-insert"; + } else { + throw error; + } + } } - try { - store.insertTaskWithFtsRecovery(task, "reconcileOrphanedTaskDirs"); - store.insertRunAuditEventRow({ - taskId: id, - domain: "database", - mutationType: "task:reconcile-orphaned-task-dir", - target: id, - metadata: { - id, - column: task.column, - status: task.status ?? null, - taskJsonPath, - }, - }); - recovered = true; - } catch (error) { - if (store.isTaskIdConflictError(error) || /Task ID already exists/i.test(error instanceof Error ? error.message : String(error))) { - skipReason = "id-conflict-during-insert"; + } else { + store.db.transactionImmediate(() => { + // FNXC:SqliteFinalRemoval 2026-06-26: taskIdExistsAnywhere is now async; + // inline the sync SQLite check here since this runs inside transactionImmediate. + if (store.readTaskFromDb(id, { includeDeleted: true }) || store.isTaskIdPresentInArchivedTasksTable(id) || store.archiveDb.get(id) !== undefined) { + skipReason = "id-exists-anywhere"; return; } - throw error; - } - }); + try { + store.insertTaskWithFtsRecovery(task, "reconcileOrphanedTaskDirs"); + store.insertRunAuditEventRow({ + taskId: id, + domain: "database", + mutationType: "task:reconcile-orphaned-task-dir", + target: id, + metadata: { + id, + column: task.column, + status: task.status ?? null, + taskJsonPath, + }, + }); + recovered = true; + } catch (error) { + if (store.isTaskIdConflictError(error) || /Task ID already exists/i.test(error instanceof Error ? error.message : String(error))) { + skipReason = "id-conflict-during-insert"; + return; + } + throw error; + } + }); + } } catch (error) { const reason = `insert-failed: ${error instanceof Error ? error.message : String(error)}`; result.skipped.push({ id, reason }); @@ -756,12 +794,13 @@ export async function reconcileOrphanedTaskDirsImpl(store: TaskStore, opts: { ig if (recovered) { result.recovered.push(id); if (store.isWatching) store.taskCache.set(id, { ...task }); - storeLog.warn("Recovered orphaned task.json into SQLite task index", { + storeLog.warn("Recovered orphaned task.json into task index", { phase: "reconcileOrphanedTaskDirs:recovered", taskId: id, column: task.column, status: task.status, taskJsonPath, + backend: store.backendMode ? "postgres" : "sqlite", }); store.emitTaskLifecycleEventSafely("task:created", [task]); } else { diff --git a/packages/core/src/task-store/moves.ts b/packages/core/src/task-store/moves.ts index 05914a556d..d796458c2c 100644 --- a/packages/core/src/task-store/moves.ts +++ b/packages/core/src/task-store/moves.ts @@ -53,11 +53,16 @@ async function resolveTaskWorkflowIrForMove(store: TaskStore, id: string): Promi const selection = await store.getTaskWorkflowSelectionAsync(id); const workflowId = selection?.workflowId; /* FNXC:WorkflowBuiltins 2026-07-19-10:24: every no-selection/unresolvable fallback goes through resolveDefaultWorkflowIr() so this resolver and prepareWorkflowMovePolicyPreflightImpl agree on the default IR (see the helper's note on the "preflight is stale" drift). */ - if (!workflowId) return store.applyBuiltInPromptOverridesSync(DEFAULT_WORKFLOW_ID, resolveDefaultWorkflowIr()); + if (!workflowId) { + return store.applyBuiltInPromptOverridesAsync(DEFAULT_WORKFLOW_ID, resolveDefaultWorkflowIr()); + } if (isBuiltinWorkflowId(workflowId)) { const builtin = getBuiltinWorkflow(workflowId); const ir = builtin?.ir; - return store.applyBuiltInPromptOverridesSync(workflowId, ir === undefined ? resolveDefaultWorkflowIr() : typeof ir === "string" ? parseWorkflowIr(ir) : ir); + return store.applyBuiltInPromptOverridesAsync( + workflowId, + ir === undefined ? resolveDefaultWorkflowIr() : typeof ir === "string" ? parseWorkflowIr(ir) : ir, + ); } try { const def = await store.getWorkflowDefinition(workflowId); diff --git a/packages/core/src/task-store/settings-ops-2.ts b/packages/core/src/task-store/settings-ops-2.ts index db55a696e3..30b36ac079 100644 --- a/packages/core/src/task-store/settings-ops-2.ts +++ b/packages/core/src/task-store/settings-ops-2.ts @@ -49,7 +49,10 @@ export async function getSettingsImpl(store: TaskStore): Promise { } catch { merged.secretsSyncPassphraseConfigured = false; } - return canonicalizeSettings(merged); + const canonical = canonicalizeSettings(merged); + // FNXC:IncompletePgPorts 2026-07-26-20:40: feed getSettingsSync cache for sync ntfy/prompt readers. + store.settingsSyncCache = canonical; + return canonical; } const [globalSettings, config] = await Promise.all([ store.globalSettingsStore.getSettings(), @@ -107,7 +110,10 @@ export async function getSettingsFastImpl(store: TaskStore): Promise { } catch { merged.secretsSyncPassphraseConfigured = false; } - return canonicalizeSettings(merged); + const canonical = canonicalizeSettings(merged); + // FNXC:IncompletePgPorts 2026-07-26-20:40: feed getSettingsSync cache (fast path). + store.settingsSyncCache = canonical; + return canonical; } const [globalSettings, row] = await Promise.all([ store.globalSettingsStore.getSettings(), diff --git a/packages/core/src/task-store/task-id-integrity.ts b/packages/core/src/task-store/task-id-integrity.ts index 781092e7c4..f657c913b8 100644 --- a/packages/core/src/task-store/task-id-integrity.ts +++ b/packages/core/src/task-store/task-id-integrity.ts @@ -11,7 +11,7 @@ import { TaskStore } from "../store.js"; import { randomUUID } from "node:crypto"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import { ArchiveDatabase } from "../archive-db.js"; import { validateBranchGroupBranchName } from "../branch-assignment.js"; import { CentralCore } from "../central-core.js"; @@ -23,8 +23,9 @@ import * as schema from "../postgres/schema/index.js"; import { getTaskCreatedHook } from "../task-creation-hooks.js"; import { type TaskIdIntegrityReport, detectTaskIdIntegrityAnomalies } from "../task-id-integrity.js"; import { createBranchGroup as createBranchGroupAsync } from "./async-branch-groups.js"; -import { findLiveLineageChildren as findLiveLineageChildrenAsync } from "./async-lifecycle.js"; +import { findLiveLineageChildren as findLiveLineageChildrenAsync, projectPartition } from "./async-lifecycle.js"; import { recordRunAuditEvent as recordRunAuditEventAsync } from "./async-audit.js"; +import { getLiveTaskColumn } from "./async-comments-attachments.js"; import { insertTaskRowInTransaction, isTaskIdConflictError, readTaskRow, readTaskRowInTransaction } from "./async-persistence.js"; import { TASK_PERSIST_SQL_COLUMNS, TASK_UPSERT_SQL_ASSIGNMENTS, type TaskRow } from "./persistence.js"; import { purgeTaskWorkflowSelectionRowsAsyncImpl } from "./workflow-definitions.js"; @@ -334,11 +335,11 @@ export async function getMergeQueuedTaskIdsAsyncImpl(store: TaskStore): Promise< export function isTaskIdPresentInArchivedTasksTableImpl(store: TaskStore, id: string): boolean { /* - * FNXC:SqliteFinalRemoval 2026-06-26-10:20: - * Backend-mode: archived tasks are not yet wired to async. Return false - * as a safety guard (the archive check is secondary to the live-tasks - * check in taskIdExistsAnywhere). - */ + FNXC:IncompletePgPorts 2026-07-26-20:30: + Sync archive-table probe remains SQLite-only. PostgreSQL callers must use + isTaskIdPresentInArchivedTasksTableAsyncImpl / taskIdExistsAnywhere (async). + Returning false here avoids opening the removed SQLite Database stub. + */ if (store.backendMode) { return false; } @@ -350,16 +351,49 @@ export function isTaskIdPresentInArchivedTasksTableImpl(store: TaskStore, id: st } } +/* +FNXC:IncompletePgPorts 2026-07-26-20:30: +PostgreSQL archive authority is project.archived_tasks (warm) and +archive.archived_tasks (cold). Both must reserve task IDs the same way the +legacy SQLite archivedTasks / archive.db tables did. +*/ +export async function isTaskIdPresentInArchivedTasksTableAsyncImpl(store: TaskStore, id: string): Promise { + if (!store.backendMode) { + return store.isTaskIdPresentInArchivedTasksTable(id); + } + const layer = store.asyncLayer!; + const partition = projectPartition(layer.projectId); + const [projectArchive, coldArchive] = await Promise.all([ + layer.db + .select({ id: schema.project.archivedTasks.id }) + .from(schema.project.archivedTasks) + .where(and( + eq(schema.project.archivedTasks.projectId, partition), + eq(schema.project.archivedTasks.id, id), + )) + .limit(1), + layer.db + .select({ id: schema.archive.archivedTasks.id }) + .from(schema.archive.archivedTasks) + .where(and( + eq(schema.archive.archivedTasks.projectId, partition), + eq(schema.archive.archivedTasks.id, id), + )) + .limit(1), + ]); + return projectArchive.length > 0 || coldArchive.length > 0; +} + export async function taskIdExistsAnywhereImpl(store: TaskStore, id: string): Promise { /* - * FNXC:SqliteFinalRemoval 2026-06-26-10:20: - * Backend-mode: use async readTaskRow (includeDeleted) for the live-tasks - * check. Archive checks are deferred (safety guard returns false above). - */ + FNXC:IncompletePgPorts 2026-07-26-20:30: + Backend-mode: live/soft-deleted rows via readTaskRow, then warm+cold archive + tables so IDs stay permanently reserved (FN-5105 parity with SQLite). + */ if (store.backendMode) { const row = await readTaskRow(store.asyncLayer!, id, { includeDeleted: true }); if (row) return true; - return false; + return isTaskIdPresentInArchivedTasksTableAsyncImpl(store, id); } // FN-5105: include soft-deleted rows so IDs remain permanently reserved. if (store.readTaskFromDb(id, { includeDeleted: true })) { @@ -452,13 +486,14 @@ export async function maybeResolveTombstonedTaskIdImpl(store: TaskStore, export function isTaskArchivedImpl(store: TaskStore, id: string): boolean { /* - * FNXC:SqliteFinalRemoval 2026-06-26: - * In backend mode, store.db is unavailable. Return false — the archive - * check in logEntry is a safety guard, and the task is loaded below - * anyway. For full correctness this should use the async layer. - */ + FNXC:IncompletePgPorts 2026-07-26-20:30: + Sync isTaskArchived cannot query PostgreSQL. Prefer isTaskArchivedAsyncImpl + from async callers. In backend mode use the in-memory task cache when the + row is already hydrated; otherwise false (caller should have used async). + */ if (store.backendMode) { - return false; + const cached = store.taskCache.get(id); + return cached?.column === "archived"; } const row = store.db.prepare(`SELECT "column" FROM tasks WHERE id = ? AND ${TaskStore.ACTIVE_TASKS_WHERE}`).get(id) as { column: Column } | undefined; if (row) { @@ -468,6 +503,23 @@ export function isTaskArchivedImpl(store: TaskStore, id: string): boolean { return store.archiveDb.get(id) !== undefined; } +/* +FNXC:IncompletePgPorts 2026-07-26-20:30: +Authoritative archived check for PostgreSQL: live column gate via +getLiveTaskColumn, plus cold archive.archived_tasks presence. +*/ +export async function isTaskArchivedAsyncImpl(store: TaskStore, id: string): Promise { + if (!store.backendMode) { + return store.isTaskArchived(id); + } + const layer = store.asyncLayer!; + const live = await getLiveTaskColumn(layer.db, id, layer.projectId); + // getLiveTaskColumn returns "archived" for archived OR soft-deleted rows. + if (live === "archived") return true; + if (live !== null) return false; + return isTaskIdPresentInArchivedTasksTableAsyncImpl(store, id); +} + export function findLiveDependentsImpl(store: TaskStore, id: string): string[] { const rows = store.db .prepare(`SELECT id, dependencies FROM tasks WHERE dependencies LIKE ? AND id != ? AND ${TaskStore.ACTIVE_TASKS_WHERE}`) diff --git a/packages/core/src/task-store/task-store-helpers.ts b/packages/core/src/task-store/task-store-helpers.ts index 41dccee9bf..3403229866 100644 --- a/packages/core/src/task-store/task-store-helpers.ts +++ b/packages/core/src/task-store/task-store-helpers.ts @@ -213,6 +213,21 @@ export function applyBuiltInPromptOverridesSyncImpl(store: TaskStore, workflowId return applyPromptOverridesToIr(ir, overrides); } +/* +FNXC:IncompletePgPorts 2026-07-26-20:40: +Async sibling for applyBuiltInPromptOverridesSyncImpl. Backend mode must load +workflow_prompt_overrides via Drizzle; the sync reader returns {} on PG and +silently dropped operator customizations from getWorkflowDefinition / move IR. +*/ +export async function applyBuiltInPromptOverridesAsyncImpl(store: TaskStore, workflowId: string, ir: WorkflowIr): Promise { + if (!isBuiltinWorkflowId(workflowId)) return ir; + const projectId = store.getWorkflowSettingsProjectId(); + const overrides = store.backendMode + ? await store.getWorkflowPromptOverridesAsync(workflowId, projectId) + : store.getWorkflowPromptOverrides(workflowId, projectId); + return applyPromptOverridesToIr(ir, overrides); +} + export async function getDefaultWorkflowIdImpl(store: TaskStore): Promise { const settings = await store.getSettingsFast(); const id = (settings as { defaultWorkflowId?: string }).defaultWorkflowId; @@ -238,18 +253,50 @@ export async function clearTaskWorkflowSelectionImpl(store: TaskStore, taskId: s export function refreshDatabaseHealthImpl(store: TaskStore): ReturnType { /* - * FNXC:SqliteFinalRemoval 2026-06-25-16:30: - * In backend mode, the SQLite integrity_check refresh path is not - * applicable (PostgreSQL manages its own integrity). Delegate to - * getDatabaseHealth() which returns the healthy sentinel. - */ + FNXC:IncompletePgPorts 2026-07-26-20:40: + Sync refresh cannot await PostgreSQL. Kick a background async refresh when + a layer is present, then return the last snapshot (or optimistic healthy + until the first probe completes). Prefer refreshDatabaseHealthAsync from + async callers (self-healing surfaceDbCorruption). + */ if (store.backendMode) { + void refreshDatabaseHealthAsyncImpl(store).catch(() => undefined); return store.getDatabaseHealth(); } store.db.refreshIntegrityCheck(); return store.getDatabaseHealth(); } +/* +FNXC:IncompletePgPorts 2026-07-26-20:40: +Probe AsyncDataLayer.ping + pg_stat_database via checkPostgresHealth and store +the result on TaskStore.postgresHealthSnapshot for sync getDatabaseHealth / +healthCheck readers. +*/ +export async function refreshDatabaseHealthAsyncImpl(store: TaskStore): Promise> { + if (!store.backendMode || !store.asyncLayer) { + return refreshDatabaseHealthImpl(store); + } + store.postgresHealthSnapshot = { + healthy: store.postgresHealthSnapshot?.healthy ?? true, + corruptionDetected: store.postgresHealthSnapshot?.corruptionDetected ?? false, + corruptionErrors: store.postgresHealthSnapshot?.corruptionErrors ?? [], + lastCheckedAt: store.postgresHealthSnapshot?.lastCheckedAt ?? null, + isRunning: true, + }; + const { checkPostgresHealth } = await import("../postgres/postgres-health.js"); + const errors = await checkPostgresHealth(store.asyncLayer); + const now = new Date(); + store.postgresHealthSnapshot = { + healthy: errors.length === 0, + corruptionDetected: errors.length > 0, + corruptionErrors: errors.slice(0, 5), + lastCheckedAt: now, + isRunning: false, + }; + return store.postgresHealthSnapshot; +} + export async function clearActivityLogImpl(store: TaskStore): Promise { /* * FNXC:SqliteFinalRemoval 2026-06-25-16:35: diff --git a/packages/core/src/task-store/workflow-definitions.ts b/packages/core/src/task-store/workflow-definitions.ts index 3db50b4f61..502d3d88ce 100644 --- a/packages/core/src/task-store/workflow-definitions.ts +++ b/packages/core/src/task-store/workflow-definitions.ts @@ -339,7 +339,10 @@ export async function getWorkflowDefinitionImpl(store: TaskStore, const requiredPluginId = getRequiredPluginIdForBuiltinWorkflow(id); if (!requiredPluginId || !(await store.isPluginInstalled(requiredPluginId))) return undefined; } - return { ...builtin, ir: store.applyBuiltInPromptOverridesSync(id, builtin.ir) }; + const ir = store.backendMode + ? await store.applyBuiltInPromptOverridesAsync(id, builtin.ir) + : store.applyBuiltInPromptOverridesSync(id, builtin.ir); + return { ...builtin, ir }; } // FNXC:WorkflowDefinitions 2026-06-27-06:00: PG backend reads the custom row // from project.workflows via the AsyncDataLayer; sync store.db otherwise. @@ -863,14 +866,13 @@ export function getDatabaseHealthImpl(store: TaskStore): { isRunning: boolean; } { /* - * FNXC:SqliteFinalRemoval 2026-06-25-16:30: - * In backend mode, SQLite-specific corruption detection (PRAGMA - * integrity_check) is not applicable. PostgreSQL health is checked via - * the async layer. Return a healthy sentinel so synchronous callers do - * not block; the real health signal comes from /api/health. - */ + FNXC:IncompletePgPorts 2026-07-26-20:40: + Backend mode returns the last refreshDatabaseHealthAsync snapshot. Until the + first probe runs, report healthy with lastCheckedAt null (unknown, not a + lie about a successful integrity check). + */ if (store.backendMode) { - return { + return store.postgresHealthSnapshot ?? { healthy: true, corruptionDetected: false, corruptionErrors: [], @@ -910,14 +912,14 @@ export function getDistributedTaskIdAllocatorImpl(store: TaskStore): Distributed } export function healthCheckImpl(store: TaskStore): boolean { - // FNXC:RuntimePersistenceAsync 2026-06-24-11:08: - // In backend mode, the sync SQLite health check is not applicable. - // PostgreSQL health is checked via the async ping() method on the - // AsyncDataLayer (wired by postgres-health.ts). Return true here so - // synchronous callers do not block; the real health signal comes from - // the /api/health endpoint which uses the async path. + /* + FNXC:IncompletePgPorts 2026-07-26-20:40: + Backend mode: report last postgresHealthSnapshot.healthy and schedule a + background refresh so CLI/daemon probes converge on real connectivity. + */ if (store.backendMode) { - return true; + void store.refreshDatabaseHealthAsync().catch(() => undefined); + return store.getDatabaseHealth().healthy; } try { // Simple query to verify database responsiveness @@ -929,14 +931,13 @@ export function healthCheckImpl(store: TaskStore): boolean { } export function getSettingsSyncImpl(store: TaskStore): Settings { - // FNXC:RuntimePersistenceAsync 2026-06-24-10:30: - // In backend mode, no synchronous DB read is possible (PostgreSQL is async). - // This method is only used by generateSpecifiedPrompt for ntfy settings. - // Return DEFAULT_SETTINGS; the async getSettings() path is the authoritative - // settings read in backend mode. Callers needing live settings must use the - // async path (getSettings/getSettingsFast). + /* + FNXC:IncompletePgPorts 2026-07-26-20:40: + Backend mode returns settingsSyncCache populated by getSettings/getSettingsFast. + Before the first async load, DEFAULT_SETTINGS is the only safe sync value. + */ if (store.backendMode) { - return DEFAULT_SETTINGS; + return store.settingsSyncCache ?? DEFAULT_SETTINGS; } try { const row = store.db.prepare("SELECT settings FROM config WHERE id = 1").get() as { settings: string | null } | undefined; diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 6531af8076..d754805349 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -1299,16 +1299,24 @@ export class SelfHealingManager { } private async getRecentRunAuditActivityAgeMs(task: Task, nowMs: number): Promise { - const getRunAuditEvents = (this.store as unknown as { + /* + FNXC:IncompletePgPorts 2026-07-26-20:45: + Prefer getRunAuditEventsAsync — sync getRunAuditEvents returns [] on + PostgreSQL and incorrectly treated busy tasks as inactive. + */ + const store = this.store as unknown as { + getRunAuditEventsAsync?: (filter: { taskId?: string; startTime?: string; limit?: number }) => Promise>; getRunAuditEvents?: (filter: { taskId?: string; startTime?: string; limit?: number }) => Array<{ timestamp?: string }>; - }).getRunAuditEvents; - if (typeof getRunAuditEvents !== "function") { - return null; - } - + }; try { const since = new Date(nowMs - RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS).toISOString(); - const events = getRunAuditEvents.call(this.store, { taskId: task.id, startTime: since, limit: 1 }); + const filter = { taskId: task.id, startTime: since, limit: 1 }; + const events = typeof store.getRunAuditEventsAsync === "function" + ? await store.getRunAuditEventsAsync(filter) + : typeof store.getRunAuditEvents === "function" + ? store.getRunAuditEvents(filter) + : null; + if (!events) return null; const newest = events.find((event) => typeof event.timestamp === "string"); if (!newest?.timestamp) return null; const timestampMs = Date.parse(newest.timestamp); @@ -5730,6 +5738,16 @@ export class SelfHealingManager { } private async surfaceDbCorruption(): Promise { + /* + FNXC:IncompletePgPorts 2026-07-26-20:45: + Refresh PostgreSQL connectivity before reading the snapshot so corruption + notifications are not permanently suppressed by the always-healthy sentinel. + */ + if (typeof this.store.refreshDatabaseHealthAsync === "function") { + await this.store.refreshDatabaseHealthAsync(); + } else { + this.store.refreshDatabaseHealth(); + } const health = this.store.getDatabaseHealth(); if (!health.corruptionDetected) { this.lastDbCorruptionNotifiedAt = null;