diff --git a/.changeset/assigned-heartbeats-reliability.md b/.changeset/assigned-heartbeats-reliability.md new file mode 100644 index 0000000000..293c0f02b1 --- /dev/null +++ b/.changeset/assigned-heartbeats-reliability.md @@ -0,0 +1,7 @@ +--- +"@runfusion/fusion": patch +--- + +summary: Restore agent models, workflow lanes, Skills, goals, and Reliability after PostgreSQL migration. +category: fix +dev: Moves backend workflow, plugin, goal, and audit reads to PostgreSQL and recovers false heartbeat model parks. diff --git a/packages/cli/src/commands/daemon.ts b/packages/cli/src/commands/daemon.ts index 120b208f8d..4178a391e5 100644 --- a/packages/cli/src/commands/daemon.ts +++ b/packages/cli/src/commands/daemon.ts @@ -14,7 +14,6 @@ import { CentralCore, TaskStore, PluginLoader, - PluginStore, getTaskMergeBlocker, INSIGHT_EXTRACTION_SCHEDULE_NAME, processAndAuditInsightExtraction, @@ -746,11 +745,12 @@ export async function runDaemon(opts: DaemonOptions = {}) { string, { enabledKey: string; skills: ReturnType } >(); - const getProjectScopedPluginSkills = async (rootDir: string): Promise> => { + const getProjectScopedPluginSkills = async (rootDir: string, resolvedProjectStore?: TaskStore): Promise> => { const normalizedRootDir = pathResolve(rootDir); - const stateStore = new PluginStore(normalizedRootDir, { centralGlobalDir: resolveGlobalDir() }); + const targetStore = resolvedProjectStore ?? (normalizedRootDir === pathResolve(store.getRootDir()) ? store : undefined); + if (!targetStore) return []; + const stateStore = targetStore.getPluginStore(); await stateStore.init(); - try { const enabledPlugins = await stateStore.listPlugins({ enabled: true }); const enabledKey = enabledPlugins .map((plugin) => `${plugin.id}:${plugin.updatedAt}`) @@ -775,12 +775,10 @@ export async function runDaemon(opts: DaemonOptions = {}) { return skills; } - const scopedPluginStore = new PluginStore(normalizedRootDir, { centralGlobalDir: resolveGlobalDir() }); - const scopedTaskStore = new TaskStore(normalizedRootDir); - const scopedPluginLoader = new PluginLoader({ pluginStore: scopedPluginStore, taskStore: scopedTaskStore }); + const scopedPluginStore = targetStore.getPluginStore(); + const scopedPluginLoader = new PluginLoader({ pluginStore: scopedPluginStore, taskStore: targetStore }); try { await scopedPluginStore.init(); - await scopedTaskStore.init(); const { errors } = await scopedPluginLoader.loadAllPlugins(); if (errors > 0) { console.warn(`[plugins] Project-scoped plugin skill loading for ${normalizedRootDir} had ${errors} error(s)`); @@ -790,12 +788,7 @@ export async function runDaemon(opts: DaemonOptions = {}) { return skills; } finally { await scopedPluginLoader.stopAllPlugins(); - scopedPluginStore.close(); - scopedTaskStore.close(); } - } finally { - stateStore.close(); - } }; const skillsAdapter = packageManager diff --git a/packages/cli/src/commands/dashboard.ts b/packages/cli/src/commands/dashboard.ts index a20becdb86..a26d3c82e2 100644 --- a/packages/cli/src/commands/dashboard.ts +++ b/packages/cli/src/commands/dashboard.ts @@ -11,7 +11,6 @@ import { CentralCore, AgentStore, PluginLoader, - PluginStore, getTaskMergeBlocker, getEnabledPiExtensionPaths, isEphemeralAgent, @@ -1826,11 +1825,12 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: string, { enabledKey: string; skills: ReturnType } >(); - const getProjectScopedPluginSkills = async (rootDir: string): Promise> => { + const getProjectScopedPluginSkills = async (rootDir: string, resolvedProjectStore?: TaskStore): Promise> => { const normalizedRootDir = pathResolve(rootDir); - const stateStore = new PluginStore(normalizedRootDir, { centralGlobalDir: resolveGlobalDir() }); + const targetStore = resolvedProjectStore ?? (normalizedRootDir === pathResolve(store.getRootDir()) ? store : undefined); + if (!targetStore) return []; + const stateStore = targetStore.getPluginStore(); await stateStore.init(); - try { const enabledPlugins = await stateStore.listPlugins({ enabled: true }); const enabledKey = enabledPlugins .map((plugin) => `${plugin.id}:${plugin.updatedAt}`) @@ -1860,12 +1860,10 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: return skills; } - const scopedPluginStore = new PluginStore(normalizedRootDir, { centralGlobalDir: resolveGlobalDir() }); - const scopedTaskStore = new TaskStore(normalizedRootDir); - const scopedPluginLoader = new PluginLoader({ pluginStore: scopedPluginStore, taskStore: scopedTaskStore }); + const scopedPluginStore = targetStore.getPluginStore(); + const scopedPluginLoader = new PluginLoader({ pluginStore: scopedPluginStore, taskStore: targetStore }); try { await scopedPluginStore.init(); - await scopedTaskStore.init(); const { errors } = await scopedPluginLoader.loadAllPlugins(); if (errors > 0) { logSink.warn(`Project-scoped plugin skill loading for ${normalizedRootDir} had ${errors} error(s)`, "plugins"); @@ -1875,12 +1873,7 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: return skills; } finally { await scopedPluginLoader.stopAllPlugins(); - scopedPluginStore.close(); - scopedTaskStore.close(); } - } finally { - stateStore.close(); - } }; const skillsAdapter = packageManager diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index aaa4fdd8a6..d7e9f7d522 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -15,7 +15,6 @@ import { CentralCore, TaskStore, PluginLoader, - PluginStore, getTaskMergeBlocker, INSIGHT_EXTRACTION_SCHEDULE_NAME, processAndAuditInsightExtraction, @@ -856,11 +855,12 @@ export async function runServe( string, { enabledKey: string; skills: ReturnType } >(); - const getProjectScopedPluginSkills = async (rootDir: string): Promise> => { + const getProjectScopedPluginSkills = async (rootDir: string, resolvedProjectStore?: TaskStore): Promise> => { const normalizedRootDir = pathResolve(rootDir); - const stateStore = new PluginStore(normalizedRootDir, { centralGlobalDir: resolveGlobalDir() }); + const targetStore = resolvedProjectStore ?? (normalizedRootDir === pathResolve(store.getRootDir()) ? store : undefined); + if (!targetStore) return []; + const stateStore = targetStore.getPluginStore(); await stateStore.init(); - try { const enabledPlugins = await stateStore.listPlugins({ enabled: true }); const enabledKey = enabledPlugins .map((plugin) => `${plugin.id}:${plugin.updatedAt}`) @@ -885,12 +885,10 @@ export async function runServe( return skills; } - const scopedPluginStore = new PluginStore(normalizedRootDir, { centralGlobalDir: resolveGlobalDir() }); - const scopedTaskStore = new TaskStore(normalizedRootDir); - const scopedPluginLoader = new PluginLoader({ pluginStore: scopedPluginStore, taskStore: scopedTaskStore }); + const scopedPluginStore = targetStore.getPluginStore(); + const scopedPluginLoader = new PluginLoader({ pluginStore: scopedPluginStore, taskStore: targetStore }); try { await scopedPluginStore.init(); - await scopedTaskStore.init(); const { errors } = await scopedPluginLoader.loadAllPlugins(); if (errors > 0) { console.warn(`[plugins] Project-scoped plugin skill loading for ${normalizedRootDir} had ${errors} error(s)`); @@ -900,12 +898,7 @@ export async function runServe( return skills; } finally { await scopedPluginLoader.stopAllPlugins(); - scopedPluginStore.close(); - scopedTaskStore.close(); } - } finally { - stateStore.close(); - } }; const skillsAdapter = packageManager diff --git a/packages/core/src/__tests__/postgres/activity-log-parity.pg.test.ts b/packages/core/src/__tests__/postgres/activity-log-parity.pg.test.ts index 1345a9a565..b3260bf86f 100644 --- a/packages/core/src/__tests__/postgres/activity-log-parity.pg.test.ts +++ b/packages/core/src/__tests__/postgres/activity-log-parity.pg.test.ts @@ -83,6 +83,14 @@ pgDescribe("activity log parity (PostgreSQL)", () => { await h.adminDb().update(schema.project.activityLog).set({ timestamp: "2026-07-13T20:01:00.000Z" }).where(eq(schema.project.activityLog.id, moved.id)); await h.adminDb().update(schema.project.activityLog).set({ timestamp: "2026-07-13T20:02:00.000Z" }).where(eq(schema.project.activityLog.id, latest.id)); + expect((await h.adminDb().select({ projectId: schema.project.activityLog.projectId, id: schema.project.activityLog.id }).from(schema.project.activityLog)) + .filter((row) => row.id !== "other-project-event")) + .toEqual(expect.arrayContaining([ + { projectId: h.layer().projectId ?? "__legacy_unscoped__", id: first.id }, + { projectId: h.layer().projectId ?? "__legacy_unscoped__", id: moved.id }, + { projectId: h.layer().projectId ?? "__legacy_unscoped__", id: latest.id }, + ])); + expect((await store.getActivityLog({ limit: 2 })).map((event) => event.taskId)).toEqual([ "FN-002", "FN-001", @@ -100,4 +108,50 @@ pgDescribe("activity log parity (PostgreSQL)", () => { .where(eq(schema.project.activityLog.projectId, "other-project")); expect(otherProject).toEqual([{ id: "other-project-event" }]); }); + + it("serves reliability duration and merged-task metrics without accessing SQLite", async () => { + const store = h.store(); + const projectId = h.layer().projectId ?? "__legacy_unscoped__"; + await h.adminDb().insert(schema.project.activityLog).values([ + { + projectId, + id: "reliability-entered", + timestamp: "2026-07-14T20:01:00.000Z", + type: "task:moved", + taskId: "FN-REL-1", + details: "Entered review", + metadata: { from: "in-progress", to: "in-review" }, + }, + { + projectId, + id: "reliability-done", + timestamp: "2026-07-14T20:02:00.000Z", + type: "task:moved", + taskId: "FN-REL-1", + details: "Completed review", + metadata: { from: "in-review", to: "done" }, + }, + { + projectId, + id: "reliability-merged", + timestamp: "2026-07-14T20:03:00.000Z", + type: "task:merged", + taskId: "FN-REL-1", + details: "Merged", + }, + { + projectId: "other-project", + id: "reliability-other-project", + timestamp: "2026-07-14T20:04:00.000Z", + type: "task:merged", + taskId: "FN-REL-OTHER", + details: "Must stay isolated", + }, + ]); + + const window = { since: "2026-07-14T20:00:00.000Z", until: "2026-07-14T20:05:00.000Z" }; + const durationEvents = await store.getInReviewDurationEvents(window); + expect(durationEvents.map((event) => event.id)).toEqual(["reliability-entered", "reliability-done"]); + expect(await store.getTaskMergedTaskIds(window)).toEqual(new Set(["FN-REL-1"])); + }); }); diff --git a/packages/core/src/__tests__/postgres/workflow-settings-project-identity.pg.test.ts b/packages/core/src/__tests__/postgres/workflow-settings-project-identity.pg.test.ts index 2ee6bd5618..6a3f79e96d 100644 --- a/packages/core/src/__tests__/postgres/workflow-settings-project-identity.pg.test.ts +++ b/packages/core/src/__tests__/postgres/workflow-settings-project-identity.pg.test.ts @@ -24,7 +24,9 @@ import { } from "../../__test-utils__/pg-test-harness.js"; import type { AsyncDataLayer } from "../../postgres/data-layer.js"; import { getWorkflowSettingsProjectIdImpl } from "../../task-store/remaining-ops-6.js"; +import { resolveEffectiveSettingsById } from "../../workflow-settings-resolver.js"; import type { TaskStore } from "../../store.js"; +import * as schema from "../../postgres/schema/index.js"; const pgTest = pgDescribe; @@ -103,6 +105,72 @@ pgTest("workflow-settings project identity keys by the central-registry id (Post expect(rows[0].project_id).not.toBe(h.rootDir()); }); + it("preserves and resolves every workflow model lane across independent PostgreSQL patches", async () => { + const store = await boundStore(); + const workflowId = "builtin:coding"; + + /* + * FNXC:WorkflowModelLanes 2026-07-14-16:26: + * Migrated execution, planning, and validator model lanes (including their fallback lanes) must coexist in one workflow JSONB row. Saving a later lane must not erase an earlier lane, and runtime resolution must consume the PostgreSQL row rather than its synchronous empty fallback. + */ + await store.updateWorkflowSettingValues(workflowId, BOUND_PROJECT_ID, { + executionProvider: "openai-codex", + executionModelId: "gpt-5.5", + }); + await store.updateWorkflowSettingValues(workflowId, BOUND_PROJECT_ID, { + planningProvider: "anthropic", + planningModelId: "claude-sonnet-5", + planningFallbackProvider: "openai-codex", + planningFallbackModelId: "gpt-5.5", + }); + await store.updateWorkflowSettingValues(workflowId, BOUND_PROJECT_ID, { + validatorProvider: "xai", + validatorModelId: "grok-code-fast-1", + validatorFallbackProvider: "anthropic", + validatorFallbackModelId: "claude-sonnet-5", + }); + + const expected = { + executionProvider: "openai-codex", + executionModelId: "gpt-5.5", + planningProvider: "anthropic", + planningModelId: "claude-sonnet-5", + planningFallbackProvider: "openai-codex", + planningFallbackModelId: "gpt-5.5", + validatorProvider: "xai", + validatorModelId: "grok-code-fast-1", + validatorFallbackProvider: "anthropic", + validatorFallbackModelId: "claude-sonnet-5", + }; + expect(await store.getWorkflowSettingValuesAsync(workflowId, BOUND_PROJECT_ID)).toMatchObject(expected); + expect(await resolveEffectiveSettingsById(store, workflowId, BOUND_PROJECT_ID)).toMatchObject(expected); + }); + + it("reads task workflow selections only from the bound project", async () => { + const store = await boundStore(); + await h.adminDb().insert(schema.project.taskWorkflowSelection).values([ + { + projectId: BOUND_PROJECT_ID, + taskId: "FN-SHARED", + workflowId: "builtin:brainstorming", + stepIds: [], + updatedAt: "2026-07-14T23:34:00.000Z", + }, + { + projectId: "proj_other", + taskId: "FN-SHARED", + workflowId: "builtin:coding", + stepIds: [], + updatedAt: "2026-07-14T23:34:00.000Z", + }, + ]); + + expect(await store.getTaskWorkflowSelectionAsync("FN-SHARED")).toEqual({ + workflowId: "builtin:brainstorming", + stepIds: [], + }); + }); + it("an UNBOUND backend layer falls back to rootDir (legacy key), proving the bound path is what changed", () => { // The shared harness store uses an unbound layer (projectId undefined). The // SQLite stub throws in getProjectIdentity, so resolution falls to rootDir. diff --git a/packages/core/src/__tests__/workflow-settings-resolver.test.ts b/packages/core/src/__tests__/workflow-settings-resolver.test.ts index d5f451af85..a33247de24 100644 --- a/packages/core/src/__tests__/workflow-settings-resolver.test.ts +++ b/packages/core/src/__tests__/workflow-settings-resolver.test.ts @@ -35,14 +35,16 @@ const CUSTOM_WITH_SETTING: WorkflowIr = { function makeStore(opts: { selection?: Record; + asyncSelection?: Record; selectionThrows?: boolean; defs?: Record; values?: Record>; // key: `${workflowId}::${projectId}` + asyncValues?: Record>; valuesThrows?: boolean; projectId?: string; projectIdThrows?: boolean; }): WorkflowSettingsResolverStore { - return { + const store: WorkflowSettingsResolverStore = { getTaskWorkflowSelection: vi.fn((taskId: string) => { if (opts.selectionThrows) throw new Error("boom"); return opts.selection?.[taskId]; @@ -57,6 +59,14 @@ function makeStore(opts: { return opts.projectId ?? PROJECT; }), }; + if (opts.asyncSelection) { + store.getTaskWorkflowSelectionAsync = vi.fn(async (taskId: string) => opts.asyncSelection?.[taskId]); + } + if (opts.asyncValues) { + store.getWorkflowSettingValuesAsync = vi.fn(async (workflowId: string, projectId: string) => + opts.asyncValues?.[`${workflowId}::${projectId}`] ?? {}); + } + return store; } describe("resolveOptionalReviewRevisionBudget", () => { @@ -185,6 +195,22 @@ describe("resolveEffectiveSettings (per-task)", () => { expect(Object.prototype.hasOwnProperty.call(b, "requirePrApproval")).toBe(false); }); + it("uses authoritative async workflow selection and model-lane values when available", async () => { + const store = makeStore({ + // The sync compatibility surface represents PostgreSQL's intentional + // no-selection/empty-values fallback. Async readers are authoritative. + selection: {}, + values: {}, + asyncSelection: { t1: { workflowId: "wf-custom", stepIds: [] } }, + defs: { "wf-custom": { ir: CUSTOM_WITH_SETTING } }, + asyncValues: { "wf-custom::proj-1": { workflowStepTimeoutMs: 12_345 } }, + }); + const effective = await resolveEffectiveSettings(store, { id: "t1" }); + expect(effective.workflowStepTimeoutMs).toBe(12_345); + expect(store.getTaskWorkflowSelection).not.toHaveBeenCalled(); + expect(store.getWorkflowSettingValues).not.toHaveBeenCalled(); + }); + it("custom workflow with empty settings → declaration-absent map (read-site fallback applies)", async () => { const store = makeStore({ selection: { t1: { workflowId: "wf-empty", stepIds: [] } }, diff --git a/packages/core/src/settings-export.ts b/packages/core/src/settings-export.ts index 6e2b180ac1..bf9ae0dc46 100644 --- a/packages/core/src/settings-export.ts +++ b/packages/core/src/settings-export.ts @@ -261,7 +261,7 @@ async function applyWorkflowSettingsSection( if (!merge) { // Replace mode: null out keys present in the current row but absent here so // the row ends up matching the imported workflow exactly. - const current = store.getWorkflowSettingValues(workflowId, projectId); + const current = await store.getWorkflowSettingValuesAsync(workflowId, projectId); for (const key of Object.keys(current)) { if (!(key in patch)) { patch[key] = null; // null-as-delete diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 60307812a3..8ccb514ec8 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -95,7 +95,7 @@ import { pgRowToTaskRow as pgRowToTaskRowExternal, rowToTask as rowToTaskExterna 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 { clearNearDuplicateReferencesToFailSoftImpl, clearWorkflowRunStepInstancesImpl, computeMovedSettingsTargetWorkflowIdsImpl, ensureBranchGroupForSourceImpl, ensurePrEntityForSourceImpl, findRecentTasksByContentFingerprintImpl, getActiveMergingTaskImpl, getActivePrEntityBySourceImpl, getBranchGroupByBranchNameImpl, getBranchGroupBySourceImpl, getBranchGroupImpl, getBranchProgressByTaskImpl, getMutationsForRunImpl, getPrEntityByNumberImpl, getPrEntityImpl, getPrThreadStateImpl, getTasksByAssignedAgentImpl, getWorkflowSettingValuesImpl, getWorkflowSettingsProjectIdImpl, getWorkflowWorkItemImpl, insertCompletionHandoffWorkflowWorkAuditImpl, listActivePrEntitiesImpl, listBranchGroupsImpl, listPrThreadStatesImpl, listTasksByBranchGroupImpl, listWorkflowSettingValuesForProjectImpl, loadWorkflowRunBranchesImpl, loadWorkflowRunStepInstancesImpl, mergeCustomFieldPatchImpl, normalizeMergeRequestStateImpl, normalizeWorkflowWorkItemKindImpl, normalizeWorkflowWorkItemStateImpl, parseWorkflowPromptOverrideJsonImpl, recordPrThreadOutcomeImpl, resetAllStepsToPendingImpl, resetPromptCheckboxesImpl, resolveWorkflowMoveActorImpl, resolveWorkflowSettingDeclarationsImpl, saveWorkflowRunStepInstanceImpl, transitionMergeRequestStateImpl, transitionWorkflowWorkItemSyncImpl, updateTaskImpl, updateWorkflowPromptOverridesImpl, upsertMergeRequestRecordImpl, workflowStateForMergeRequestStateImpl } from "./task-store/remaining-ops-6.js"; +import { clearNearDuplicateReferencesToFailSoftImpl, 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, loadWorkflowRunStepInstancesImpl, mergeCustomFieldPatchImpl, normalizeMergeRequestStateImpl, normalizeWorkflowWorkItemKindImpl, normalizeWorkflowWorkItemStateImpl, parseWorkflowPromptOverrideJsonImpl, recordPrThreadOutcomeImpl, resetAllStepsToPendingImpl, resetPromptCheckboxesImpl, resolveWorkflowMoveActorImpl, resolveWorkflowSettingDeclarationsImpl, 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, 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"; import { getTaskCommitAssociationsByLineageIdImpl, replaceLegacyTaskCommitAssociationsImpl } from "./task-store/remaining-ops-9.js"; @@ -1176,6 +1176,9 @@ export class TaskStore extends EventEmitter { getWorkflowSettingValues(workflowId: string, projectId: string): Record { return getWorkflowSettingValuesImpl(this, workflowId, projectId); } + async getWorkflowSettingValuesAsync(workflowId: string, projectId: string): Promise> { + return getWorkflowSettingValuesAsyncImpl(this, workflowId, projectId); + } // ── Built-in workflow prompt overrides (FN-6893) ─────────────────────────── // FNXC:CustomWorkflows 2026-06-21-19:07: @@ -1192,6 +1195,9 @@ export class TaskStore extends EventEmitter { getWorkflowPromptOverrides(workflowId: string, projectId: string): Record { return getWorkflowPromptOverridesImpl(this, workflowId, projectId); } + async getWorkflowPromptOverridesAsync(workflowId: string, projectId: string): Promise> { + return getWorkflowPromptOverridesAsyncImpl(this, workflowId, projectId); + } /** non-string, empty, or whitespace value deletes that nodeId override, which */ async updateWorkflowPromptOverrides( workflowId: string, projectId: string, patch: Record, ): Promise> { diff --git a/packages/core/src/task-store/async-audit.ts b/packages/core/src/task-store/async-audit.ts index c12949929a..206e460c94 100644 --- a/packages/core/src/task-store/async-audit.ts +++ b/packages/core/src/task-store/async-audit.ts @@ -23,7 +23,7 @@ * These helpers are the async target the migrating store and the PostgreSQL * integration tests consume. */ -import { and, count, desc, eq, gte, lte, sql } from "drizzle-orm"; +import { and, asc, count, desc, eq, gt, gte, isNotNull, lte, or, sql } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; import type { AsyncDataLayer, DbTransaction } from "../postgres/data-layer.js"; import { @@ -169,6 +169,14 @@ export async function countRunAuditEvents( // ── Activity log ───────────────────────────────────────────────────── +/** + * FNXC:ReliabilityHealth 2026-07-14-16:29: + * PostgreSQL rewrites an empty activity project id to the explicit legacy quarantine. Normalize both writes and reads identically so unbound compatibility stores can still observe their own telemetry; bound runtime stores continue using their central project id. + */ +export function activityProjectPartition(projectId: string): string { + return projectId.trim() || "__legacy_unscoped__"; +} + /** * Convert a raw `activity_log` row into the public `ActivityLogEntry` shape. * The `metadata` column is jsonb in PostgreSQL (already-parsed). @@ -215,7 +223,7 @@ export async function recordActivityLogEntry( try { await db.insert(schema.project.activityLog).values({ - projectId, + projectId: activityProjectPartition(projectId), id: fullEntry.id, timestamp: fullEntry.timestamp, type: fullEntry.type, @@ -248,7 +256,7 @@ export async function getActivityLog( projectId: string, options?: { limit?: number; since?: string; type?: ActivityEventType }, ): Promise { - const conditions = [eq(schema.project.activityLog.projectId, projectId)]; + const conditions = [eq(schema.project.activityLog.projectId, activityProjectPartition(projectId))]; if (options?.since) { conditions.push(gte(schema.project.activityLog.timestamp, options.since)); } @@ -290,7 +298,7 @@ export async function getTaskMovedCountsByDay( options: { since: string; until: string; fromColumn?: string; toColumn?: string }, ): Promise> { const conditions = [ - eq(schema.project.activityLog.projectId, projectId), + eq(schema.project.activityLog.projectId, activityProjectPartition(projectId)), eq(schema.project.activityLog.type, "task:moved"), gte(schema.project.activityLog.timestamp, options.since), lte(schema.project.activityLog.timestamp, options.until), @@ -321,3 +329,51 @@ export async function getTaskMovedCountsByDay( } return countsByDay; } + +/* +FNXC:ReliabilityHealth 2026-07-14-16:13: +Reliability metrics must query PostgreSQL activity rows through the async data layer. Keep the bounded duration-event shape and project scope used by the dashboard without falling through to the unavailable SQLite TaskStore database. +*/ +export async function getInReviewDurationEvents( + db: AsyncDataLayer["db"] | DbTransaction, + projectId: string, + options: { since: string; until: string }, +): Promise { + const rows = await db + .select() + .from(schema.project.activityLog) + .where(and( + eq(schema.project.activityLog.projectId, activityProjectPartition(projectId)), + eq(schema.project.activityLog.type, "task:moved"), + gt(schema.project.activityLog.timestamp, options.since), + lte(schema.project.activityLog.timestamp, options.until), + or( + sql`${schema.project.activityLog.metadata}->>'to' = 'in-review'`, + and( + sql`${schema.project.activityLog.metadata}->>'from' = 'in-review'`, + sql`${schema.project.activityLog.metadata}->>'to' = 'done'`, + ), + ), + )) + .orderBy(asc(schema.project.activityLog.timestamp)) + .limit(200_000); + return (rows as ActivityLogRow[]).map((row) => rowToActivityLogEntry(row)); +} + +export async function getTaskMergedTaskIds( + db: AsyncDataLayer["db"] | DbTransaction, + projectId: string, + options: { since: string; until: string }, +): Promise> { + const rows = await db + .selectDistinct({ taskId: schema.project.activityLog.taskId }) + .from(schema.project.activityLog) + .where(and( + eq(schema.project.activityLog.projectId, activityProjectPartition(projectId)), + eq(schema.project.activityLog.type, "task:merged"), + gt(schema.project.activityLog.timestamp, options.since), + lte(schema.project.activityLog.timestamp, options.until), + isNotNull(schema.project.activityLog.taskId), + )); + return new Set(rows.flatMap((row) => row.taskId ? [row.taskId] : [])); +} diff --git a/packages/core/src/task-store/moves.ts b/packages/core/src/task-store/moves.ts index 284908f316..a55e36e507 100644 --- a/packages/core/src/task-store/moves.ts +++ b/packages/core/src/task-store/moves.ts @@ -172,6 +172,9 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum // capacity check is not a guard (U6 fills the enforcement; U4 leaves a // pass-through slot). An explicit option value wins; otherwise derive it. const bypassGuards = store.resolveWorkflowBypassGuards(moveSource, options); + const effectiveWorkflowIdForMove = useWorkflow + ? (await store.getTaskWorkflowSelectionAsync(id))?.workflowId ?? "builtin:coding" + : "builtin:coding"; const workflowIr: WorkflowIr | undefined = useWorkflow ? await resolveTaskWorkflowIrForMove(store, id) : undefined; @@ -719,11 +722,10 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum if (useWorkflow && workflowIr && fromColumn !== toColumn) { const capacity = resolveColumnCapacity(workflowIr, toColumn, mergedSettingsForMove); if (capacity.hasCapacity && Number.isFinite(capacity.limit)) { - const workflowId = store.resolveEffectiveWorkflowIdSync(id); const occupants = await store.countActiveInCapacitySlotAsync({ tx, targetColumn: toColumn, - workflowId, + workflowId: effectiveWorkflowIdForMove, countPending: capacity.countPending, excludeTaskId: id, }); @@ -851,10 +853,9 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum if (useWorkflow && workflowIr && fromColumn !== toColumn) { const capacity = resolveColumnCapacity(workflowIr, toColumn, mergedSettingsForMove); if (capacity.hasCapacity && Number.isFinite(capacity.limit)) { - const workflowId = store.resolveEffectiveWorkflowIdSync(id); const occupants = store.countActiveInCapacitySlotSync({ targetColumn: toColumn, - workflowId, + workflowId: effectiveWorkflowIdForMove, countPending: capacity.countPending, excludeTaskId: id, }); @@ -1045,4 +1046,3 @@ export async function moveTaskInternalImpl(store: TaskStore, id: string, toColum } return task; } - diff --git a/packages/core/src/task-store/remaining-ops-10.ts b/packages/core/src/task-store/remaining-ops-10.ts index 51020a87e4..966fb9c82d 100644 --- a/packages/core/src/task-store/remaining-ops-10.ts +++ b/packages/core/src/task-store/remaining-ops-10.ts @@ -24,6 +24,7 @@ import { BoardConfig, BranchGroup, MergeRequestRecord, Task, WorkflowStepTemplat import { WorkflowFieldDefinition, WorkflowIr, WorkflowIrColumn } from "../workflow-ir-types.js"; import { applyPromptOverridesToIr } from "../workflow-prompt-overrides.js"; import { MoveTaskOptions } from "../store.js"; +import { activityProjectPartition } from "./async-audit.js"; export function readTaskRowFromDbImpl(store: TaskStore, id: string, options?: { includeDeleted?: boolean }): TaskRow | undefined { const whereClause = options?.includeDeleted ? "id = ?" : `id = ? AND ${TaskStore.ACTIVE_TASKS_WHERE}`; @@ -256,7 +257,7 @@ export async function clearActivityLogImpl(store: TaskStore): Promise { const layer = store.asyncLayer!; await layer.db .delete(schema.project.activityLog) - .where(eq(schema.project.activityLog.projectId, layer.projectId ?? "")); + .where(eq(schema.project.activityLog.projectId, activityProjectPartition(layer.projectId ?? ""))); return; } store.db.prepare("DELETE FROM activityLog").run(); @@ -327,4 +328,3 @@ export function getTodoStoreImpl(store: TaskStore): TodoStore | AsyncTodoStore { return store.todoStore; } - diff --git a/packages/core/src/task-store/remaining-ops-2.ts b/packages/core/src/task-store/remaining-ops-2.ts index a681580eaa..0159fb9a90 100644 --- a/packages/core/src/task-store/remaining-ops-2.ts +++ b/packages/core/src/task-store/remaining-ops-2.ts @@ -620,15 +620,24 @@ export async function updateWorkflowSettingValuesImpl(store: TaskStore, workflow // write transaction. Validation/declaration resolution above stays outside // since it's async and doesn't read the row being mutated. /* - * FNXC:SqliteFinalRemoval 2026-06-26: - * P1 fix: no backendMode branch existed, so this threw in PG mode. In - * backend mode, read-merge-upsert via Drizzle inside a transactionImmediate - * (values is jsonb). The lost-update guard is preserved by the transaction. + * FNXC:WorkflowModelLanes 2026-07-14-16:26: + * PostgreSQL workflow setting patches must read and write the existing JSONB row through the same transaction handle. The synchronous backend getter intentionally returns an empty default; using it here erased every previously saved model lane whenever another lane was patched. */ if (store.backendMode) { const layer = store.asyncLayer!; - return layer.transactionImmediate(async () => { - const current = await store.getWorkflowSettingValues(workflowId, projectId); + return layer.transactionImmediate(async (tx) => { + const rows = await tx + .select({ values: schema.project.workflowSettings.values }) + .from(schema.project.workflowSettings) + .where(and( + eq(schema.project.workflowSettings.workflowId, workflowId), + eq(schema.project.workflowSettings.projectId, projectId), + )) + .limit(1); + const rawCurrent = rows[0]?.values; + const current = rawCurrent && typeof rawCurrent === "object" && !Array.isArray(rawCurrent) + ? rawCurrent as Record + : {}; const next: Record = { ...current }; for (const [key, value] of Object.entries(result.accepted)) { if (value === null) { @@ -639,7 +648,7 @@ export async function updateWorkflowSettingValuesImpl(store: TaskStore, workflow } const now = new Date().toISOString(); - await layer.db + await tx .insert(schema.project.workflowSettings) .values({ workflowId, diff --git a/packages/core/src/task-store/remaining-ops-4.ts b/packages/core/src/task-store/remaining-ops-4.ts index c773d83922..ca9226c63b 100644 --- a/packages/core/src/task-store/remaining-ops-4.ts +++ b/packages/core/src/task-store/remaining-ops-4.ts @@ -8,6 +8,7 @@ */ import {TaskStore, isWorkflowColumnsCompatibilityFlagEnabled} from "../store.js"; import {resolveEntryColumnId} from "../workflow-reconciliation.js"; +import {resolveWorkflowIrForTask} from "../workflow-ir-resolver.js"; import * as schema from "../postgres/schema/index.js"; import type {MoveTaskOptions, MoveTaskInternalOptions} from "../store.js"; import {TASK_BRANCH_CONTEXT_METADATA_KEY} from "../store.js"; @@ -372,7 +373,10 @@ export async function prepareWorkflowMovePolicyPreflightImpl(store: TaskStore, i if (!isWorkflowColumnsCompatibilityFlagEnabled(mergedSettingsForMove)) return undefined; if (task.column === toColumn) return undefined; - const workflowIr = store.resolveTaskWorkflowIrSync(id); + /* FNXC:WorkflowModelLanes 2026-07-14-16:31: PostgreSQL move preflight must validate against the task's migrated workflow selection, not the synchronous builtin:coding fallback. */ + const workflowIr = store.backendMode + ? await resolveWorkflowIrForTask(store, id) + : store.resolveTaskWorkflowIrSync(id); const workflowSignature = serializeWorkflowIr(workflowIr); const bypassGuards = store.resolveWorkflowBypassGuards(moveSource, options); const fromColumn = task.column; diff --git a/packages/core/src/task-store/remaining-ops-6.ts b/packages/core/src/task-store/remaining-ops-6.ts index 42451c14d4..7b2c685963 100644 --- a/packages/core/src/task-store/remaining-ops-6.ts +++ b/packages/core/src/task-store/remaining-ops-6.ts @@ -844,6 +844,30 @@ export function getWorkflowSettingValuesImpl(store: TaskStore, workflowId: strin } } +/** + * FNXC:WorkflowModelLanes 2026-07-14-16:26: + * PostgreSQL-backed workflow execution must read the migrated per-project workflow values instead of the synchronous backend fallback. Model lanes, fallback lanes, thinking levels, and other workflow policy are authoritative in this row after the settings hard-move. + */ +export async function getWorkflowSettingValuesAsyncImpl( + store: TaskStore, + workflowId: string, + projectId: string, + ): Promise> { + if (!store.backendMode) return store.getWorkflowSettingValues(workflowId, projectId); + const rows = await store.asyncLayer!.db + .select({ values: schema.project.workflowSettings.values }) + .from(schema.project.workflowSettings) + .where(and( + eq(schema.project.workflowSettings.workflowId, workflowId), + eq(schema.project.workflowSettings.projectId, projectId), + )) + .limit(1); + const values = rows[0]?.values; + return values && typeof values === "object" && !Array.isArray(values) + ? values as Record + : {}; +} + export function parseWorkflowPromptOverrideJsonImpl(store: TaskStore, raw: string | null | undefined): Record { if (!raw) return {}; try { @@ -862,22 +886,54 @@ export function parseWorkflowPromptOverrideJsonImpl(store: TaskStore, raw: strin } } +/** FNXC:WorkflowModelLanes 2026-07-14-16:26: Async workflow resolution must retain migrated project-scoped prompt overrides in PostgreSQL backend mode. */ +export async function getWorkflowPromptOverridesAsyncImpl( + store: TaskStore, + workflowId: string, + projectId: string, + ): Promise> { + if (!store.backendMode) return store.getWorkflowPromptOverrides(workflowId, projectId); + const rows = await store.asyncLayer!.db + .select({ overrides: schema.project.workflowPromptOverrides.overrides }) + .from(schema.project.workflowPromptOverrides) + .where(and( + eq(schema.project.workflowPromptOverrides.workflowId, workflowId), + eq(schema.project.workflowPromptOverrides.projectId, projectId), + )) + .limit(1); + const overrides = rows[0]?.overrides; + if (!overrides || typeof overrides !== "object" || Array.isArray(overrides)) return {}; + const out: Record = {}; + for (const [nodeId, value] of Object.entries(overrides as Record)) { + if (typeof value === "string" && value.trim()) out[nodeId] = value; + } + return out; +} + export async function updateWorkflowPromptOverridesImpl(store: TaskStore, workflowId: string, projectId: string, patch: Record, ): Promise> { /* - * FNXC:SqliteFinalRemoval 2026-06-26: - * P1 fix: no backendMode branch existed, so this threw in PG mode. In - * backend mode, read-merge-upsert the workflow_prompt_overrides row via - * Drizzle inside a transactionImmediate (preserving the lost-update guard - * the sync path's transactionImmediate provides). overrides is jsonb. + * FNXC:WorkflowModelLanes 2026-07-14-16:26: + * Keep PostgreSQL prompt override patches on the same authoritative transaction path as workflow settings; a backend sync-default read must never erase sibling overrides. */ if (store.backendMode) { const layer = store.asyncLayer!; - return layer.transactionImmediate(async () => { - const current = await store.getWorkflowPromptOverrides(workflowId, projectId); + return layer.transactionImmediate(async (tx) => { + const rows = await tx + .select({ overrides: schema.project.workflowPromptOverrides.overrides }) + .from(schema.project.workflowPromptOverrides) + .where(and( + eq(schema.project.workflowPromptOverrides.workflowId, workflowId), + eq(schema.project.workflowPromptOverrides.projectId, projectId), + )) + .limit(1); + const rawCurrent = rows[0]?.overrides; + const current = rawCurrent && typeof rawCurrent === "object" && !Array.isArray(rawCurrent) + ? rawCurrent as Record + : {}; const next: Record = { ...current }; for (const [nodeId, value] of Object.entries(patch)) { if (typeof value !== "string" || value.trim().length === 0) { @@ -888,7 +944,7 @@ export async function updateWorkflowPromptOverridesImpl(store: TaskStore, } const now = new Date().toISOString(); - await layer.db + await tx .insert(schema.project.workflowPromptOverrides) .values({ workflowId, diff --git a/packages/core/src/task-store/remaining-ops-8.ts b/packages/core/src/task-store/remaining-ops-8.ts index 56ee0484f6..68e3d6bf16 100644 --- a/packages/core/src/task-store/remaining-ops-8.ts +++ b/packages/core/src/task-store/remaining-ops-8.ts @@ -24,12 +24,13 @@ import { PluginStore } from "../plugin-store.js"; import { SecretsStore } from "../secrets-store.js"; import { createAsyncDistributedTaskIdAllocator } from "./async-allocator.js"; import { getWorkflowRow, listWorkflowRows } from "../async-workflow-store.js"; +import { getInReviewDurationEvents as getInReviewDurationEventsAsync, getTaskMergedTaskIds as getTaskMergedTaskIdsAsync } from "./async-audit.js"; import { readProjectConfig, writeProjectConfig } from "./async-settings.js"; import { compactTaskActivityLog } from "./comments.js"; import { type TaskRow } from "./persistence.js"; import { ActivityLogRow } from "./row-types.js"; import { ActivityEventType, ActivityLogEntry, AgentLogEntry, ArchivedTaskEntry, DEFAULT_SETTINGS, Settings } from "../types.js"; -import { eq } from "drizzle-orm"; +import { and, eq } from "drizzle-orm"; import * as schema from "../postgres/schema/index.js"; import { normalizeWorkflowIcon, type StoredWorkflowRow, type WorkflowDefinition, type WorkflowDefinitionInput, type WorkflowNodeLayout } from "../workflow-definition-types.js"; import { WorkflowIr } from "../workflow-ir-types.js"; @@ -37,6 +38,7 @@ import { downgradeIrToV1IfPure, parseWorkflowIr, serializeWorkflowIr } from "../ import { resolveDefaultOnOptionalGroupIds } from "../workflow-optional-steps.js"; import { resolveSwitchReconciliation } from "../workflow-reconciliation.js"; import { WORKFLOW_COMPILED_STEP_TEMPLATE_PREFIX } from "../store.js"; +import { resolveWorkflowIrForTask } from "../workflow-ir-resolver.js"; export async function getAgentLogsByTimeRangeImpl(store: TaskStore, taskId: string, @@ -480,10 +482,18 @@ Async backend-mode read of a task's workflow selection (PostgreSQL). stepIds is export async function getTaskWorkflowSelectionAsyncImpl(store: TaskStore, taskId: string): Promise<{ workflowId: string; stepIds: string[] } | undefined> { if (!store.backendMode) return store.getTaskWorkflowSelection(taskId); const layer = store.asyncLayer!; + /* + FNXC:WorkflowModelLanes 2026-07-14-16:34: + A task workflow selection is project-owned. Shared PostgreSQL deployments may reuse task ids across projects, so every authoritative selection read must include the bound central project id instead of relying on taskId or connection state alone. + */ + const projectId = layer.projectId?.trim() || "__legacy_unscoped__"; const rows = await layer.db .select({ workflowId: schema.project.taskWorkflowSelection.workflowId, stepIds: schema.project.taskWorkflowSelection.stepIds }) .from(schema.project.taskWorkflowSelection) - .where(eq(schema.project.taskWorkflowSelection.taskId, taskId)) + .where(and( + eq(schema.project.taskWorkflowSelection.projectId, projectId), + eq(schema.project.taskWorkflowSelection.taskId, taskId), + )) .limit(1); if (rows.length === 0) return undefined; const row = rows[0]!; @@ -678,7 +688,9 @@ export async function selectTaskWorkflowAndReconcileImpl(store: TaskStore, if (!(await store.workflowColumnsFlagOn())) { return { enabledWorkflowSteps }; } - const newIr = store.resolveTaskWorkflowIrSync(taskId); + const newIr = store.backendMode + ? await resolveWorkflowIrForTask(store, taskId) + : store.resolveTaskWorkflowIrSync(taskId); const current = store.readTaskFromDb(taskId, { includeDeleted: false }); if (!current) return { enabledWorkflowSteps }; const fromColumn = current.column; @@ -839,6 +851,10 @@ export function getSettingsSyncImpl(store: TaskStore): Settings { } export async function getInReviewDurationEventsImpl(store: TaskStore, options: { since: string; until: string }): Promise { + if (store.backendMode) { + const layer = store.asyncLayer!; + return getInReviewDurationEventsAsync(layer.db, layer.projectId ?? "", options); + } const rows = store.db .prepare( `SELECT * FROM activityLog @@ -869,6 +885,10 @@ export async function getInReviewDurationEventsImpl(store: TaskStore, options: { } export async function getTaskMergedTaskIdsImpl(store: TaskStore, options: { since: string; until: string }): Promise> { + if (store.backendMode) { + const layer = store.asyncLayer!; + return getTaskMergedTaskIdsAsync(layer.db, layer.projectId ?? "", options); + } const rows = store.db .prepare( `SELECT DISTINCT taskId FROM activityLog diff --git a/packages/core/src/workflow-ir-resolver.ts b/packages/core/src/workflow-ir-resolver.ts index 620a2b20c0..2f79d9d2e2 100644 --- a/packages/core/src/workflow-ir-resolver.ts +++ b/packages/core/src/workflow-ir-resolver.ts @@ -33,9 +33,11 @@ function defaultCodingWorkflowIr(): WorkflowIr { /** Minimal store surface the resolver needs (public APIs only). */ export interface WorkflowIrResolverStore { getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined; + getTaskWorkflowSelectionAsync?(taskId: string): Promise<{ workflowId: string; stepIds: string[] } | undefined>; getWorkflowDefinition(id: string): Promise<{ ir: string | WorkflowIr } | undefined>; getWorkflowSettingsProjectId?(): string; getWorkflowPromptOverrides?(workflowId: string, projectId: string): Record; + getWorkflowPromptOverridesAsync?(workflowId: string, projectId: string): Promise>; } /** @@ -93,7 +95,7 @@ export async function resolveTaskPlanningPrompt( * sweep. Hits short-circuit before any builtin/db lookup. */ export async function resolveWorkflowIrById( - store: Pick & Partial>, + store: Pick & Partial>, workflowId: string, irCache?: Map, ): Promise { @@ -116,7 +118,10 @@ export async function resolveWorkflowIrById( const builtin = getBuiltinWorkflow(workflowId); const ir = builtin?.ir ?? defaultCodingWorkflowIr(); const resolved = typeof ir === "string" ? parseWorkflowIr(ir) : ir; - const overrides = projectId ? store.getWorkflowPromptOverrides?.(workflowId, projectId) : undefined; + const overrides = projectId + ? await (store.getWorkflowPromptOverridesAsync?.(workflowId, projectId) + ?? store.getWorkflowPromptOverrides?.(workflowId, projectId)) + : undefined; // FNXC:CustomWorkflows 2026-06-21-19:12: // Public IR resolution must see the same project-scoped built-in prompt overrides as task execution, while callers without the new store methods keep the canonical built-in IR. const effective = applyPromptOverridesToIr(resolved, overrides); @@ -146,7 +151,14 @@ export async function resolveWorkflowIrForTask( ): Promise { let workflowId: string | undefined; try { - workflowId = store.getTaskWorkflowSelection(taskId)?.workflowId; + /* + * FNXC:WorkflowModelLanes 2026-07-14-16:26: + * Backend-mode task workflow selection is asynchronous. Execution must resolve the migrated task selection before loading its workflow graph; the synchronous PostgreSQL fallback intentionally reports no selection and previously forced every task onto builtin:coding. + */ + const selection = store.getTaskWorkflowSelectionAsync + ? await store.getTaskWorkflowSelectionAsync(taskId) + : store.getTaskWorkflowSelection(taskId); + workflowId = selection?.workflowId; } catch { return defaultCodingWorkflowIr(); } diff --git a/packages/core/src/workflow-settings-resolver.ts b/packages/core/src/workflow-settings-resolver.ts index 37ce418f43..bc3392d17c 100644 --- a/packages/core/src/workflow-settings-resolver.ts +++ b/packages/core/src/workflow-settings-resolver.ts @@ -107,6 +107,7 @@ export interface EffectiveSettingsResult { export interface WorkflowSettingsResolverStore extends WorkflowIrResolverStore { /** Raw stored `(workflowId, projectId)` value map; `{}` when no row exists. */ getWorkflowSettingValues(workflowId: string, projectId: string): Record; + getWorkflowSettingValuesAsync?(workflowId: string, projectId: string): Promise>; /** The stable project id this store scopes `workflow_settings` rows by. A store * instance is bound to one project, so the resolver derives the project key from * the store rather than from the task (Task carries no projectId field). */ @@ -130,17 +131,18 @@ function declarationsFromIr( /** Compose declarations + raw stored values → effective flat map + the set of keys * whose value came from an explicit stored workflow value (never throws). */ -function effectiveFrom( +async function effectiveFrom( store: WorkflowSettingsResolverStore, ir: WorkflowIr, workflowId: string | undefined, projectId: string, -): EffectiveSettingsResult { +): Promise { const declarations = declarationsFromIr(ir, workflowId); let stored: Record = {}; if (workflowId) { try { - stored = store.getWorkflowSettingValues(workflowId, projectId) ?? {}; + stored = await (store.getWorkflowSettingValuesAsync?.(workflowId, projectId) + ?? store.getWorkflowSettingValues(workflowId, projectId)) ?? {}; } catch { stored = {}; } @@ -172,7 +174,7 @@ export async function resolveEffectiveSettingsById( irCache?: Map, ): Promise> { const ir = await resolveWorkflowIrById(store, workflowId, irCache); - return effectiveFrom(store, ir, workflowId, projectId).effective; + return (await effectiveFrom(store, ir, workflowId, projectId)).effective; } /** The minimal task identity the per-task resolver reads. Task carries no @@ -211,7 +213,10 @@ export async function resolveEffectiveSettingsDetailed( ): Promise { let workflowId: string | undefined; try { - workflowId = store.getTaskWorkflowSelection(task.id)?.workflowId; + const selection = store.getTaskWorkflowSelectionAsync + ? await store.getTaskWorkflowSelectionAsync(task.id) + : store.getTaskWorkflowSelection(task.id); + workflowId = selection?.workflowId; } catch { workflowId = undefined; } diff --git a/packages/dashboard/src/routes/__tests__/register-agent-skills-routes.test.ts b/packages/dashboard/src/routes/__tests__/register-agent-skills-routes.test.ts index cc92128c07..b18c794306 100644 --- a/packages/dashboard/src/routes/__tests__/register-agent-skills-routes.test.ts +++ b/packages/dashboard/src/routes/__tests__/register-agent-skills-routes.test.ts @@ -168,6 +168,7 @@ describe("register-agent-skills-routes", () => { "/tmp/file-root", "npm::skills/test-skill", "reference.md", + expect.objectContaining({ getRootDir: expect.any(Function) }), ); }); diff --git a/packages/dashboard/src/routes/board-workflows.ts b/packages/dashboard/src/routes/board-workflows.ts index 6e68226431..7808e4eecd 100644 --- a/packages/dashboard/src/routes/board-workflows.ts +++ b/packages/dashboard/src/routes/board-workflows.ts @@ -156,7 +156,8 @@ async function describeWorkflow( * can return early and the client renders the legacy board. */ export async function buildBoardWorkflowsPayload( - store: Pick, + store: Pick & + Partial>, taskIds: string[], settingsOverride?: Pick, ): Promise { @@ -177,7 +178,9 @@ export async function buildBoardWorkflowsPayload( for (const taskId of taskIds) { let workflowId = DEFAULT_WORKFLOW_LANE_ID; try { - const selection = store.getTaskWorkflowSelection(taskId); + const selection = store.getTaskWorkflowSelectionAsync + ? await store.getTaskWorkflowSelectionAsync(taskId) + : store.getTaskWorkflowSelection(taskId); if (selection?.workflowId) workflowId = selection.workflowId; } catch { workflowId = DEFAULT_WORKFLOW_LANE_ID; diff --git a/packages/dashboard/src/routes/register-agent-skills-routes.ts b/packages/dashboard/src/routes/register-agent-skills-routes.ts index 58a6881104..6e61f98806 100644 --- a/packages/dashboard/src/routes/register-agent-skills-routes.ts +++ b/packages/dashboard/src/routes/register-agent-skills-routes.ts @@ -21,7 +21,7 @@ export function registerAgentSkillsRoutes(ctx: ApiRoutesContext): void { } const rootDir = scopedStore.getRootDir(); - const skills = await skillsAdapter.discoverSkills(rootDir); + const skills = await skillsAdapter.discoverSkills(rootDir, scopedStore); res.json({ skills }); } catch (err: unknown) { @@ -57,7 +57,7 @@ export function registerAgentSkillsRoutes(ctx: ApiRoutesContext): void { const skillId = req.params.id as string; const rootDir = scopedStore.getRootDir(); - const content = await skillsAdapter.readSkillContent(rootDir, skillId); + const content = await skillsAdapter.readSkillContent(rootDir, skillId, scopedStore); res.json({ content }); } catch (err: unknown) { @@ -107,7 +107,7 @@ export function registerAgentSkillsRoutes(ctx: ApiRoutesContext): void { } const rootDir = scopedStore.getRootDir(); - const file = await skillsAdapter.readSkillFileContent(rootDir, skillId, rawPath); + const file = await skillsAdapter.readSkillFileContent(rootDir, skillId, rawPath, scopedStore); res.json({ file }); } catch (err: unknown) { @@ -160,7 +160,7 @@ export function registerAgentSkillsRoutes(ctx: ApiRoutesContext): void { } const rootDir = scopedStore.getRootDir(); - const persistence = await skillsAdapter.toggleExecutionSkill(rootDir, { skillId, enabled }); + const persistence = await skillsAdapter.toggleExecutionSkill(rootDir, { skillId, enabled }, scopedStore); res.json({ success: true, diff --git a/packages/dashboard/src/routes/register-workflow-routes.ts b/packages/dashboard/src/routes/register-workflow-routes.ts index 1d1a2d5c06..4f278a8813 100644 --- a/packages/dashboard/src/routes/register-workflow-routes.ts +++ b/packages/dashboard/src/routes/register-workflow-routes.ts @@ -475,7 +475,7 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { await assertWorkflowExists(store, workflowId); const projectId = store.getWorkflowSettingsProjectId(); const declarations = await resolveSettingDeclarations(store, workflowId); - const stored = store.getWorkflowSettingValues(workflowId, projectId); + const stored = await store.getWorkflowSettingValuesAsync(workflowId, projectId); res.json({ stored, effective: resolveEffectiveSettingValues(declarations, stored), @@ -543,7 +543,7 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { await assertWorkflowExists(store, workflowId); const projectId = store.getWorkflowSettingsProjectId(); const defaults = await resolvePromptOverrideDefaults(store, workflowId); - const stored = store.getWorkflowPromptOverrides(workflowId, projectId); + const stored = await store.getWorkflowPromptOverridesAsync(workflowId, projectId); res.json({ stored, effective: resolveEffectivePromptOverrides(defaults, stored), @@ -600,7 +600,7 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { router.get("/tasks/:taskId/workflow", async (req, res) => { try { const { store } = await getProjectContext(req); - const selection = store.getTaskWorkflowSelection(req.params.taskId); + const selection = await store.getTaskWorkflowSelectionAsync(req.params.taskId); res.json({ workflowId: selection?.workflowId ?? null, enabledWorkflowSteps: selection ? selection.stepIds : null, @@ -774,8 +774,8 @@ export function registerWorkflowRoutes(ctx: ApiRoutesContext): void { ir: def.ir, layout: def.layout, ...(def.icon ? { icon: def.icon } : {}), - settingValues: store.getWorkflowSettingValues(def.id, projectId), - promptOverrides: store.getWorkflowPromptOverrides(def.id, projectId), + settingValues: await store.getWorkflowSettingValuesAsync(def.id, projectId), + promptOverrides: await store.getWorkflowPromptOverridesAsync(def.id, projectId), }); } catch (err: unknown) { if (err instanceof ApiError) throw err; diff --git a/packages/dashboard/src/server.ts b/packages/dashboard/src/server.ts index c343d40847..25edeb01a5 100644 --- a/packages/dashboard/src/server.ts +++ b/packages/dashboard/src/server.ts @@ -15,8 +15,9 @@ import type { MessageStore, AgentLogEntry, TaskIdIntegrityReport, + RunAuditEvent, } from "@fusion/core"; -import { AgentStore, ChatStore, setRunningAgentCountSource } from "@fusion/core"; +import { AgentStore, ChatStore, queryRunAuditEvents, setRunningAgentCountSource } from "@fusion/core"; import type { AuthStorageLike, ModelRegistryLike } from "./routes.js"; import { createApiRoutes } from "./routes.js"; import { createSSE, disconnectSSEClient, markSSEClientAlive } from "./sse.js"; @@ -1826,8 +1827,23 @@ export function createServer(store: TaskStore, options?: ServerOptions): ReturnT const startIso = new Date(effectiveStartMs).toISOString(); const endIso = new Date(nowMs).toISOString(); + const auditFilter = { startTime: startIso, endTime: endIso, limit: 50_000 }; + const asyncLayer = scopedStore.getAsyncLayer(); + /* + FNXC:ReliabilityHealth 2026-07-14-16:13: + Backend-mode Reliability must use the authoritative async run-audit reader. The synchronous reader is a SQLite/test compatibility surface and intentionally degrades to an empty result under PostgreSQL. + */ + const runAuditEventsPromise: Promise = asyncLayer + ? queryRunAuditEvents(asyncLayer.db, auditFilter).then((events) => events.map((event) => ({ + ...event, + domain: event.domain as RunAuditEvent["domain"], + mutationType: event.mutationType as RunAuditEvent["mutationType"], + taskId: event.taskId ?? undefined, + metadata: event.metadata ?? undefined, + }))) + : Promise.resolve(scopedStore.getRunAuditEvents(auditFilter)); const [runAuditEvents, enteredByDay, bouncedByDay, durationEvents, mergedTaskIds] = await Promise.all([ - Promise.resolve(scopedStore.getRunAuditEvents({ startTime: startIso, endTime: endIso, limit: 50_000 })), + runAuditEventsPromise, scopedStore.getTaskMovedCountsByDay({ since: startIso, until: endIso, toColumn: "in-review" }), scopedStore.getTaskMovedCountsByDay({ since: startIso, until: endIso, fromColumn: "in-review", toColumn: "in-progress" }), scopedStore.getInReviewDurationEvents({ since: startIso, until: endIso }), diff --git a/packages/dashboard/src/skills-adapter.ts b/packages/dashboard/src/skills-adapter.ts index eaf7c74dc8..d963065b02 100644 --- a/packages/dashboard/src/skills-adapter.ts +++ b/packages/dashboard/src/skills-adapter.ts @@ -17,6 +17,7 @@ import { superviseSpawn, } from "@fusion/core"; export { computeSkillId, getSkillSettingState, parseSkillId } from "@fusion/core"; +import type { TaskStore } from "@fusion/core"; import type { ChildProcess } from "node:child_process"; /** @@ -162,7 +163,7 @@ export interface SkillsAdapter { * Discover all skills available in the project. * Combines top-level skills and package-scoped skills. */ - discoverSkills(rootDir: string): Promise; + discoverSkills(rootDir: string, projectStore?: TaskStore): Promise; /** * Toggle a skill's enabled/disabled state. @@ -171,6 +172,7 @@ export interface SkillsAdapter { toggleExecutionSkill( rootDir: string, input: { skillId: string; enabled: boolean }, + projectStore?: TaskStore, ): Promise; /** @@ -186,13 +188,13 @@ export interface SkillsAdapter { /** * Read the contents of a skill's SKILL.md file and list supplementary files. */ - readSkillContent(rootDir: string, skillId: string): Promise; + readSkillContent(rootDir: string, skillId: string, projectStore?: TaskStore): Promise; /* FNXC:Skills 2026-06-23-04:15: Read a single supplementary file's text for the detail-pane file viewer. The SkillsView detail pane lists referenced files; clicking one must show its content. The `files` array carried only name/path/type, so a per-file content endpoint is required. `relativePath` is the skill-dir-relative path returned by readSkillContent; it is resolved + path-traversal-guarded against the skill directory so a request can never escape the skill root. */ - readSkillFileContent(rootDir: string, skillId: string, relativePath: string): Promise; + readSkillFileContent(rootDir: string, skillId: string, relativePath: string, projectStore?: TaskStore): Promise; } /* @@ -282,7 +284,7 @@ export function createSkillsAdapter(options: { * skill catalog omits them. Lazy thunk: plugins may load after the adapter is * created, so it is invoked per discovery rather than captured eagerly. */ - getPluginSkills?: (rootDir: string) => + getPluginSkills?: (rootDir: string, projectStore?: TaskStore) => | Array<{ pluginId: string; pluginRoot?: string; @@ -297,7 +299,7 @@ export function createSkillsAdapter(options: { superviseSpawn?: typeof superviseSpawn; }): SkillsAdapter { return { - async discoverSkills(rootDir: string): Promise { + async discoverSkills(rootDir: string, projectStore?: TaskStore): Promise { // Resolve all resources including skills const resolved = await options.packageManager.resolve(); const skillResources = resolved.skills ?? []; @@ -354,7 +356,13 @@ export function createSkillsAdapter(options: { * FNXC:PluginSkills 2026-07-12-00:00: * Plugin skill body paths now come from skillFiles (GitHub #2018) through @fusion/core's traversal-guarded resolver when pluginRoot is available. Discovered plugin skill path is the absolute on-disk SKILL.md location for FN-7857 consumers, while missing pluginRoot keeps the old name-derived relative path for compatibility. */ - const pluginSkills = await (options.getPluginSkills?.(rootDir) ?? []); + /* + * FNXC:Skills 2026-07-14-16:13: + * Plugin skill discovery receives the already-resolved project TaskStore so PostgreSQL callers reuse its AsyncDataLayer-backed PluginStore. Creating a root-only PluginStore here re-entered the removed SQLite Database path and made the Skills page fail after migration. + */ + const pluginSkills = await (projectStore + ? options.getPluginSkills?.(rootDir, projectStore) + : options.getPluginSkills?.(rootDir)) ?? []; if (pluginSkills.length > 0) { const seenBareNames = new Set(discoveredSkills.map((s) => bareSkillName(s.name))); for (const { pluginId, pluginRoot, skill } of pluginSkills) { @@ -396,6 +404,7 @@ export function createSkillsAdapter(options: { async toggleExecutionSkill( rootDir: string, input: { skillId: string; enabled: boolean }, + projectStore?: TaskStore, ): Promise { const { skillId, enabled } = input; const parsed = parseSkillId(skillId); @@ -406,7 +415,7 @@ export function createSkillsAdapter(options: { const { source, relativePath } = parsed; // Validate that the skill exists in discovered skills - const discovered = await this.discoverSkills(rootDir); + const discovered = await this.discoverSkills(rootDir, projectStore); const skillExists = discovered.some((s) => s.id === skillId); if (!skillExists) { throw new Error(`Skill not found: ${skillId}`); @@ -656,13 +665,13 @@ export function createSkillsAdapter(options: { } }, - async readSkillContent(rootDir: string, skillId: string): Promise { + async readSkillContent(rootDir: string, skillId: string, projectStore?: TaskStore): Promise { const parsed = parseSkillId(skillId); if (!parsed) { throw new Error(`Invalid skill ID format: ${skillId}`); } - const discovered = await this.discoverSkills(rootDir); + const discovered = await this.discoverSkills(rootDir, projectStore); const skill = discovered.find((entry) => entry.id === skillId); if (!skill) { throw new Error(`Skill not found: ${skillId}`); @@ -711,13 +720,13 @@ export function createSkillsAdapter(options: { FNXC:Skills 2026-06-23-04:15: Per-file content read for the detail-pane viewer. Resolves the skill directory the same way readSkillContent does, then joins the requested relativePath. Guards against path traversal (resolved target must stay inside the skill dir) and refuses to read SKILL.md through this path (the SKILL.md view has its own endpoint). Binary/oversized files return isText:false with empty content so the UI shows a non-previewable notice rather than garbled output. */ - async readSkillFileContent(rootDir: string, skillId: string, relativePath: string): Promise { + async readSkillFileContent(rootDir: string, skillId: string, relativePath: string, projectStore?: TaskStore): Promise { const parsed = parseSkillId(skillId); if (!parsed) { throw new Error(`Invalid skill ID format: ${skillId}`); } - const discovered = await this.discoverSkills(rootDir); + const discovered = await this.discoverSkills(rootDir, projectStore); const skill = discovered.find((entry) => entry.id === skillId); if (!skill) { throw new Error(`Skill not found: ${skillId}`); diff --git a/packages/engine/src/__tests__/agent-session-helpers.test.ts b/packages/engine/src/__tests__/agent-session-helpers.test.ts index 555dcaaf77..6811834141 100644 --- a/packages/engine/src/__tests__/agent-session-helpers.test.ts +++ b/packages/engine/src/__tests__/agent-session-helpers.test.ts @@ -229,7 +229,7 @@ describe("resolve session model parity", () => { }); }); - it("does not let a stale complete runtime model mask newer task or settings models", () => { + it("does not let a stale complete runtime model mask newer task or settings models outside heartbeat", () => { const staleRuntimeConfig = { model: "openai-codex/gpt-5.3-codex" }; expect(resolveExecutorSessionModel("task-provider", "task-model", settings, staleRuntimeConfig)).toEqual({ @@ -248,12 +248,6 @@ describe("resolve session model parity", () => { provider: "anthropic", modelId: "claude-sonnet-4-5", }); - expect(resolveHeartbeatSessionModels(settings, staleRuntimeConfig)).toEqual({ - defaultProvider: "openai", - defaultModelId: "gpt-4.1", - fallbackProvider: undefined, - fallbackModelId: undefined, - }); expect(resolveValidatorSessionModel("validator-task-provider", "validator-task-model", settings, staleRuntimeConfig)).toEqual({ provider: "validator-task-provider", modelId: "validator-task-model", @@ -272,6 +266,23 @@ describe("resolve session model parity", () => { }); }); + it("uses the durable agent's complete assigned model for heartbeat instead of shared execution settings", () => { + expect(resolveHeartbeatSessionModels( + { + defaultProviderOverride: "anthropic", + defaultModelIdOverride: "claude-sonnet-5", + defaultProvider: "openai-codex", + defaultModelId: "gpt-5.5", + }, + { modelProvider: "grok-cli", modelId: "grok-4.5", model: "grok-cli/grok-4.5" }, + )).toEqual({ + defaultProvider: "grok-cli", + defaultModelId: "grok-4.5", + fallbackProvider: undefined, + fallbackModelId: undefined, + }); + }); + it("does not leak malformed gpt 5.3-style runtimeConfig into any automatic lane", () => { const malformedRuntimeConfig = { modelId: "gpt 5.3" }; @@ -391,7 +402,7 @@ describe("resolve session model parity", () => { }); }); -describe("project model override precedence invariant", () => { +describe("non-heartbeat project model override precedence invariant", () => { const staleRuntimeConfig = { model: "stale-provider/stale-model" }; const partialRuntimeConfigs: Array> = [ { modelProvider: "stale-provider" }, @@ -430,18 +441,6 @@ describe("project model override precedence invariant", () => { }, runtimeConfig), expected: { provider: "project-validator-provider", modelId: "project-validator-model" }, }, - { - label: "heartbeat execution lane", - settings: { executionProvider: "project-heartbeat-provider", executionModelId: "project-heartbeat-model" }, - resolve: (runtimeConfig?: Record) => { - const resolved = resolveHeartbeatSessionModels({ - executionProvider: "project-heartbeat-provider", - executionModelId: "project-heartbeat-model", - }, runtimeConfig); - return { provider: resolved.defaultProvider, modelId: resolved.defaultModelId }; - }, - expected: { provider: "project-heartbeat-provider", modelId: "project-heartbeat-model" }, - }, { label: "merger default lane", settings: { defaultProviderOverride: "project-default-provider", defaultModelIdOverride: "project-default-model" }, diff --git a/packages/engine/src/__tests__/self-healing.test.ts b/packages/engine/src/__tests__/self-healing.test.ts index 3f30d29f0e..97fdfb58a5 100644 --- a/packages/engine/src/__tests__/self-healing.test.ts +++ b/packages/engine/src/__tests__/self-healing.test.ts @@ -981,6 +981,22 @@ describe("SelfHealingManager", () => { { id: "operator-actionable", state: "error", lastError: "OAuth token does not meet scope requirements", updatedAt: new Date(now).toISOString(), metadata: { untouched: true } } as unknown as Agent, { id: "stale-module", state: "error", lastError: staleModuleError, updatedAt: new Date(now).toISOString(), metadata: { untouched: true } } as unknown as Agent, { id: "error-unrecoverable", state: "paused", pauseReason: HEARTBEAT_ERROR_UNRECOVERABLE_PAUSE_REASON, lastError: "socket hang up", updatedAt: new Date(now).toISOString() } as unknown as Agent, + { + id: "misattributed-heartbeat-model", + state: "paused", + pauseReason: "heartbeat-model-unavailable", + lastError: 'No API key for provider: anthropic. Configure credentials for provider "anthropic" in settings, then resume the agent.', + runtimeConfig: { enabled: true, modelProvider: "grok-cli", modelId: "grok-4.5", model: "grok-cli/grok-4.5" }, + updatedAt: new Date(now).toISOString(), + } as unknown as Agent, + { + id: "genuine-heartbeat-model", + state: "paused", + pauseReason: "heartbeat-model-unavailable", + lastError: 'No API key for provider: anthropic. Configure credentials for provider "anthropic" in settings, then resume the agent.', + runtimeConfig: { enabled: true, modelProvider: "anthropic", modelId: "claude-opus-4-8", model: "anthropic/claude-opus-4-8" }, + updatedAt: new Date(now).toISOString(), + } as unknown as Agent, { id: "user-paused", state: "paused", pauseReason: "manual", lastError: "socket hang up", updatedAt: new Date(now).toISOString() } as unknown as Agent, { id: "ephemeral", state: "error", lastError: "socket hang up", metadata: { agentKind: "task-worker" }, updatedAt: new Date(now).toISOString() } as unknown as Agent, { id: "disabled", state: "error", lastError: "socket hang up", runtimeConfig: { enabled: false }, updatedAt: new Date(now).toISOString() } as unknown as Agent, @@ -1004,7 +1020,7 @@ describe("SelfHealingManager", () => { await managerWithAgents.runStartupRecovery(); - for (const agentId of ["fresh-error", "exhausted-parked"]) { + for (const agentId of ["fresh-error", "exhausted-parked", "misattributed-heartbeat-model"]) { const agent = agentStore.getAgent(agentId)!; expect(agent.state).toBe("active"); expect(agent.lastError).toBeUndefined(); @@ -1017,22 +1033,24 @@ describe("SelfHealingManager", () => { } expect(agentStore.getAgent("fresh-error")?.metadata?.unrelated).toBe("keep"); expect(agentStore.getAgent("exhausted-parked")?.metadata?.unrelated).toBe("keep-too"); - expect(restartDurableAgentHeartbeat).toHaveBeenCalledTimes(2); + expect(restartDurableAgentHeartbeat).toHaveBeenCalledTimes(3); expect(restartDurableAgentHeartbeat).toHaveBeenCalledWith("fresh-error", { reason: "startup-error-reset", attempt: 1 }); expect(restartDurableAgentHeartbeat).toHaveBeenCalledWith("exhausted-parked", { reason: "startup-error-reset", attempt: 1 }); + expect(restartDurableAgentHeartbeat).toHaveBeenCalledWith("misattributed-heartbeat-model", { reason: "startup-error-reset", attempt: 1 }); const resetAudits = recordRunAuditEvent.mock.calls .map(([event]) => event) .filter((event) => event.mutationType === "agent:reset-error-state-on-startup"); - expect(resetAudits).toHaveLength(2); + expect(resetAudits).toHaveLength(3); expect(resetAudits).toEqual(expect.arrayContaining([ expect.objectContaining({ target: "fresh-error", metadata: expect.objectContaining({ agentId: "fresh-error", priorState: "error", source: "self-healing" }) }), expect.objectContaining({ target: "exhausted-parked", metadata: expect.objectContaining({ agentId: "exhausted-parked", priorState: "paused", priorPauseReason: HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON, source: "self-healing" }) }), + expect.objectContaining({ target: "misattributed-heartbeat-model", metadata: expect.objectContaining({ agentId: "misattributed-heartbeat-model", priorState: "paused", priorPauseReason: "heartbeat-model-unavailable", source: "self-healing" }) }), ])); expect(recordRunAuditEvent.mock.calls.map(([event]) => event.mutationType).filter((type) => type === "agent:auto-recover-error-state")).toHaveLength(0); - expect(agentStore.updateAgentState).toHaveBeenCalledTimes(2); + expect(agentStore.updateAgentState).toHaveBeenCalledTimes(3); - for (const untouchedId of ["operator-actionable", "stale-module", "error-unrecoverable", "user-paused", "ephemeral", "disabled", "live-agent", "healthy-active", "healthy-idle"]) { + for (const untouchedId of ["operator-actionable", "stale-module", "error-unrecoverable", "genuine-heartbeat-model", "user-paused", "ephemeral", "disabled", "live-agent", "healthy-active", "healthy-idle"]) { expect(agentStore.updateAgentState).not.toHaveBeenCalledWith(untouchedId, expect.anything()); expect(agentStore.updateAgent).not.toHaveBeenCalledWith(untouchedId, expect.anything()); } diff --git a/packages/engine/src/agent-session-helpers.ts b/packages/engine/src/agent-session-helpers.ts index 241fabeb76..b7bb334827 100644 --- a/packages/engine/src/agent-session-helpers.ts +++ b/packages/engine/src/agent-session-helpers.ts @@ -559,7 +559,14 @@ export function resolveHeartbeatSessionModels( } const executionSettingsModel = resolveExecutionSettingsModel(settings); - const resolvedModel = pickSettingsThenRuntimeModel(executionSettingsModel, assignedAgentRuntimeConfig); + const assignedRuntimeModel = extractRuntimeModel(assignedAgentRuntimeConfig); + /* + FNXC:AgentHeartbeat 2026-07-14-16:13: + Durable-agent heartbeats must use the complete model assigned to that agent. Shared project execution defaults are only a fallback for an absent or incomplete assignment; otherwise one broken project override can park every heterogeneous agent under the same unrelated provider. + */ + const resolvedModel = hasCompleteRuntimeModel(assignedRuntimeModel) + ? assignedRuntimeModel + : pickSettingsThenRuntimeModel(executionSettingsModel, assignedAgentRuntimeConfig); return { defaultProvider: resolvedModel.provider, diff --git a/packages/engine/src/agent-tools.ts b/packages/engine/src/agent-tools.ts index bf1fe9cedc..b4eba237c5 100644 --- a/packages/engine/src/agent-tools.ts +++ b/packages/engine/src/agent-tools.ts @@ -2744,7 +2744,7 @@ export function createWorkflowSettingsTool(store: TaskStore): ToolDefinition { if (params.action === "get") { try { - const stored = store.getWorkflowSettingValues(workflowId, projectId); + const stored = await store.getWorkflowSettingValuesAsync(workflowId, projectId); const effective = await resolveEffectiveSettingsById(store, workflowId, projectId); const declarations = await resolveWorkflowSettingDeclarationsForTool(store, workflowId); const orphaned = findOrphanedSettingValues(declarations, stored); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index ccdd6f1337..f639819d5d 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -5168,7 +5168,9 @@ export class TaskExecutor { return false; } try { - selection = this.store.getTaskWorkflowSelection(task.id); + selection = typeof this.store.getTaskWorkflowSelectionAsync === "function" + ? await this.store.getTaskWorkflowSelectionAsync(task.id) + : this.store.getTaskWorkflowSelection(task.id); } catch (err) { await this.handleGraphFailure(task, { disposition: "failed", @@ -5940,7 +5942,9 @@ export class TaskExecutor { if (!isExperimentalFeatureEnabled(settings, WORKFLOW_INTERPRETER_DUAL_OBSERVE_FLAG)) return; if (typeof this.store.getTaskWorkflowSelection !== "function") return; try { - const selection = this.store.getTaskWorkflowSelection(taskId); + const selection = typeof this.store.getTaskWorkflowSelectionAsync === "function" + ? await this.store.getTaskWorkflowSelectionAsync(taskId) + : this.store.getTaskWorkflowSelection(taskId); if (!selection) return; const def = await this.store.getWorkflowDefinition?.(selection.workflowId); if (!def) return; diff --git a/packages/engine/src/goal-injection-diagnostics.ts b/packages/engine/src/goal-injection-diagnostics.ts index 6452eea7a4..15bd7bb72b 100644 --- a/packages/engine/src/goal-injection-diagnostics.ts +++ b/packages/engine/src/goal-injection-diagnostics.ts @@ -1,5 +1,4 @@ import type { Goal } from "@fusion/core"; -import { MissionStore, GoalStore } from "@fusion/core"; import { buildGoalContextSection, type GoalInjectionResult } from "./goal-context-injector.js"; import type { TaskStore } from "@fusion/core"; import type { GoalAnchoringLane } from "./goal-anchoring-audit.js"; @@ -136,33 +135,36 @@ export function resolveGoalContextForDiagnostics(input: ResolveGoalContextInput) } export async function resolveAndEmitGoalContext(input: ResolveAndEmitGoalContextInput): Promise { - // FNXC:GoalStore 2026-06-27-18:25: - // resolveGoalContextForDiagnostics consumes a SYNC listActiveGoals (() => Goal[]) - // on the synchronous agent-execution path. The PG-backed AsyncGoalStore returns - // Promises, so only the sync SQLite GoalStore can supply this. Guard with - // instanceof GoalStore (mirrors the AsyncMissionStore guard below) and leave it - // undefined in PG backend mode — goal context degrades to store-unavailable - // rather than injecting an unresolved Promise. Converting the injection pipeline - // to async is out of scope for the GoalStore port. - const resolvedGoalStore = - typeof input.store.getGoalStore === "function" ? input.store.getGoalStore() : undefined; - const syncGoalStore = resolvedGoalStore instanceof GoalStore ? resolvedGoalStore : undefined; - const resolution = resolveGoalContextForDiagnostics({ - listActiveGoals: syncGoalStore - ? () => syncGoalStore.listGoals({ status: "active" }) - : undefined, - }); + /* + FNXC:GoalStore 2026-07-14-16:13: + Goal injection is an async prompt-construction lane and must await either the SQLite GoalStore or PostgreSQL AsyncGoalStore. Treating the async store as unavailable silently removed active goals from every migrated triage, executor, and heartbeat prompt. + */ + let resolution: GoalContextResolution; + if (typeof input.store.getGoalStore !== "function") { + resolution = { goalContext: "", classification: classifyGoalInjectionFailure("store-unavailable") }; + } else { + try { + const goalStore = input.store.getGoalStore(); + const activeGoals = await goalStore.listGoals({ status: "active" }); + const injectionResult = buildGoalContextSection({ activeGoals }); + resolution = { + goalContext: injectionResult.text, + classification: classifyGoalInjectionResult(injectionResult), + }; + } catch (listError) { + resolution = { + goalContext: "", + classification: classifyGoalInjectionFailure("list-failed", listError), + }; + } + } let provenanceGoalIds: string[] = []; if (input.taskId && typeof input.store.getMissionStore === "function") { try { - // FNXC:MissionStore 2026-06-27-15:40: - // listGoalIdsForTask is a sync-only MissionStore method (not ported to the - // AsyncMissionStore). In PG backend mode getMissionStore() returns the async - // store; guard with instanceof and leave provenance empty (graceful fallback). const resolvedMissionStore = input.store.getMissionStore(); - if (resolvedMissionStore instanceof MissionStore) { - provenanceGoalIds = resolvedMissionStore.listGoalIdsForTask(input.taskId); + if ("listGoalIdsForTask" in resolvedMissionStore && typeof resolvedMissionStore.listGoalIdsForTask === "function") { + provenanceGoalIds = await Promise.resolve(resolvedMissionStore.listGoalIdsForTask(input.taskId)); } } catch (error) { diagnosticsLog.warn( diff --git a/packages/engine/src/hold-release.ts b/packages/engine/src/hold-release.ts index c79e5fd8aa..9f3240f50b 100644 --- a/packages/engine/src/hold-release.ts +++ b/packages/engine/src/hold-release.ts @@ -96,9 +96,9 @@ export interface HoldReleaseResult { // resolveWorkflowIrForTask (GitHub #1402); the optional per-sweep irCache Map is // threaded straight through. -function effectiveWorkflowId(store: TaskStore, taskId: string): string { +async function effectiveWorkflowId(store: TaskStore, taskId: string): Promise { try { - return store.getTaskWorkflowSelection(taskId)?.workflowId ?? DEFAULT_WORKFLOW_POOL_ID; + return (await store.getTaskWorkflowSelectionAsync(taskId))?.workflowId ?? DEFAULT_WORKFLOW_POOL_ID; } catch { return DEFAULT_WORKFLOW_POOL_ID; } @@ -370,7 +370,7 @@ export async function runHoldReleaseSweep( const irCache = new Map(); const effectiveWorkflowIdByTask = new Map(); for (const t of allTasks) { - effectiveWorkflowIdByTask.set(t.id, effectiveWorkflowId(store, t.id)); + effectiveWorkflowIdByTask.set(t.id, await effectiveWorkflowId(store, t.id)); } for (const task of allTasks) { diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index af79d2b82c..0dfa1cc888 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -91,9 +91,32 @@ import type { GhostBugDecision } from "./triage-preflight.js"; import { DependencyBlockedTodoReporter } from "./dependency-blocked-todo-reporter.js"; import { filterPathsByIgnoreList, getUnmetSchedulingDependencies, isCoordinationOnlyTask, pathsOverlap, shouldHoldActiveFileScopeLease } from "./scheduler.js"; import { evaluateParkedAgentTaskLink, PARKED_AGENT_LINK_FRESH_RUN_MS } from "./task-agent-sync.js"; +import { extractRuntimeModel } from "./agent-session-helpers.js"; const log = createLogger("self-healing"); const OPTIONAL_STEP_REVISION_KEY_MARKER = "Workflow revision key:"; +const HEARTBEAT_MODEL_UNAVAILABLE_PAUSE_REASON = "heartbeat-model-unavailable"; + +function extractHeartbeatUnavailableProvider(error: string | undefined): string | undefined { + if (!error) return undefined; + const rawProvider = /no api key for provider:\s*([^\s)]+)/i.exec(error)?.[1] + ?? /configured primary model\s+([^/\s]+)\//i.exec(error)?.[1]; + return rawProvider?.replace(/["'.,:;]+$/g, "").trim() || undefined; +} + +function isMisattributedHeartbeatModelPark(agent: Agent): boolean { + if (agent.state !== "paused" || agent.pauseReason !== HEARTBEAT_MODEL_UNAVAILABLE_PAUSE_REASON) { + return false; + } + const assignedModel = extractRuntimeModel((agent.runtimeConfig ?? {}) as Record); + const failedProvider = extractHeartbeatUnavailableProvider(agent.lastError); + return Boolean( + assignedModel.provider + && assignedModel.modelId + && failedProvider + && assignedModel.provider.toLowerCase() !== failedProvider.toLowerCase(), + ); +} function normalizeOptionalStepRevisionKey(value: string | undefined): string { return (value ?? "").trim().toLowerCase(); @@ -10571,6 +10594,9 @@ export class SelfHealingManager { /* FNXC:AgentHeartbeat 2026-07-12-17:26: FN-7884: Engine restart is an explicit operator retry boundary for durable heartbeat agents. Startup recovery must immediately clear recoverable `error` and `error-retry-exhausted` parks, reset shared heartbeatErrorRecovery/durableErrorRecovery budget state, and re-arm heartbeats without steady-state staleness/cooldown/exhaustion gates; operator-actionable, stale-module, user-paused, error-unrecoverable, disabled, ephemeral, and actively executing agents remain suppressed. + + FNXC:AgentHeartbeat 2026-07-14-16:13: + Startup must also recover a `heartbeat-model-unavailable` park when its recorded failing provider differs from the agent's complete assigned runtime model. This repairs agents falsely parked by the former shared-project-model precedence while preserving genuine assigned-provider authentication failures for operator action. */ async resetDurableAgentErrorStateOnStartup(): Promise { const agentStore = this.options.agentStore; @@ -10584,7 +10610,8 @@ export class SelfHealingManager { for (const agent of allAgents) { const isErrorRetryExhaustedPark = agent.state === "paused" && agent.pauseReason === HEARTBEAT_ERROR_RETRY_EXHAUSTED_PAUSE_REASON; - if (agent.state !== "error" && !isErrorRetryExhaustedPark) { + const isMisattributedModelPark = isMisattributedHeartbeatModelPark(agent); + if (agent.state !== "error" && !isErrorRetryExhaustedPark && !isMisattributedModelPark) { continue; } if (isEphemeralAgent(agent)) { @@ -10597,7 +10624,7 @@ export class SelfHealingManager { if (this.options.hasActiveAgentExecution?.(agent.id) === true) { continue; } - if (!isHeartbeatErrorRecoverable(agent) || isStaleWorktreeModuleResolutionError(agent.lastError ?? "")) { + if ((!isMisattributedModelPark && !isHeartbeatErrorRecoverable(agent)) || isStaleWorktreeModuleResolutionError(agent.lastError ?? "")) { log.warn(`Startup durable-agent error reset suppressed for ${agent.id}: unrecoverable or stale-module error requires existing recovery path`); continue; } diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index a02898dbc0..651368ce80 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -2030,9 +2030,11 @@ export class TriageProcessor { * FNXC:PlanValidation 2026-06-30-09:20: * Triage may run Plan Review before the graph reaches `plan-review`; the graph later skips an already-passed Plan Review result. Read the selected workflow's Plan Review template flag here so Coding (per-step review) enforces external-integration evidence in the same Plan Review gate, while default Coding and other workflows stay unblocked. */ - const selection = typeof this.store.getTaskWorkflowSelection === "function" - ? this.store.getTaskWorkflowSelection(task.id) - : undefined; + const selection = typeof this.store.getTaskWorkflowSelectionAsync === "function" + ? await this.store.getTaskWorkflowSelectionAsync(task.id) + : typeof this.store.getTaskWorkflowSelection === "function" + ? this.store.getTaskWorkflowSelection(task.id) + : undefined; const workflowId = selection?.workflowId; if (!workflowId || typeof this.store.getWorkflowDefinition !== "function") return false; const definition = await this.store.getWorkflowDefinition(workflowId).catch((error: unknown) => { diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index 30bb8dabf3..3080d53eb0 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -61,6 +61,7 @@ export interface WorkflowGraphTaskRunResult { /** The minimal store surface the runner needs — keeps tests fake-friendly. */ export interface WorkflowGraphRunnerStore { getTaskWorkflowSelection(taskId: string): { workflowId: string; stepIds: string[] } | undefined; + getTaskWorkflowSelectionAsync?(taskId: string): Promise<{ workflowId: string; stepIds: string[] } | undefined>; getWorkflowDefinition(id: string): Promise; getTask?(taskId: string): Promise; } @@ -180,7 +181,9 @@ export class WorkflowGraphTaskRunner { ): Promise { let selection: { workflowId: string; stepIds: string[] } | undefined; try { - selection = this.deps.store.getTaskWorkflowSelection(task.id); + selection = this.deps.store.getTaskWorkflowSelectionAsync + ? await this.deps.store.getTaskWorkflowSelectionAsync(task.id) + : this.deps.store.getTaskWorkflowSelection(task.id); } catch (err) { return this.fallBack(task.id, `selection-error: ${err instanceof Error ? err.message : String(err)}`); } diff --git a/packages/engine/src/workflow-task-runtime.ts b/packages/engine/src/workflow-task-runtime.ts index 26cdb74925..1246cd2aae 100644 --- a/packages/engine/src/workflow-task-runtime.ts +++ b/packages/engine/src/workflow-task-runtime.ts @@ -290,7 +290,10 @@ export class WorkflowTaskRuntime { private async resolveRuntimeTarget(taskId: string): Promise { let workflowId: string | undefined; try { - workflowId = this.deps.store.getTaskWorkflowSelection(taskId)?.workflowId; + const selection = this.deps.store.getTaskWorkflowSelectionAsync + ? await this.deps.store.getTaskWorkflowSelectionAsync(taskId) + : this.deps.store.getTaskWorkflowSelection(taskId); + workflowId = selection?.workflowId; } catch (err) { throw new Error(`workflow-selection-failed: ${err instanceof Error ? err.message : String(err)}`); }