FN-6783: recover orphaned task dirs into the task index
Reconcile on-disk task.json records that are missing from SQLite without resurrecting tombstoned IDs. - Add TaskStore orphaned task directory scanning with metadata validation, FTS-safe insertion, cache updates, lifecycle events, and run-audit entries. - Run the reconcile during store init and self-healing maintenance for tasks created after startup. - Cover recovery, skip, and maintenance behavior with core and engine regression tests. - Document task index reconciliation and add the published package changeset. Files changed: .changeset/fn-6783-orphaned-task-dir-reconcile.md | 5 + AGENTS.md | 1 + docs/architecture.md | 2 + docs/storage.md | 8 + .../store-orphaned-task-dir-reconcile.test.ts | 179 +++++++++++++++++++++ packages/core/src/store.ts | 170 ++++++++++++++++++- .../self-healing-orphaned-task-dirs.test.ts | 40 +++++ packages/engine/src/self-healing.ts | 14 ++ 8 files changed, 412 insertions(+), 7 deletions(-) Fusion-Task-Id: FN-6783 Fusion-Task-Lineage: 5c8d4690-5278-4c29-8f02-32d9f01d581d
This commit is contained in:
gsxdsm
5
.changeset/fn-6783-orphaned-task-dir-reconcile.md
Normal file
5
.changeset/fn-6783-orphaned-task-dir-reconcile.md
Normal file
@@ -0,0 +1,5 @@
|
||||
---
|
||||
"@runfusion/fusion": patch
|
||||
---
|
||||
|
||||
Repair task-store startup and self-healing consistency by non-destructively re-importing orphaned live `.fusion/tasks/{ID}/task.json` records into the SQLite task index while preserving soft-deleted, archived, and tombstoned IDs.
|
||||
@@ -192,6 +192,7 @@ Scoped exception (FN-5819): shared-branch-group members (`branchContext.assignme
|
||||
- FN-5419: git run-audit now includes `pull:fast-forward` and `stash:pop-conflict`; dashboard git surfaces now include the extended `POST /api/git/pull` integration-worktree path plus companion `POST /api/git/stash-resolve`, `POST /api/git/stash-drop`, and `POST /api/git/stash-apply` routes.
|
||||
- FN-6292: self-healing emits `task:reconcile-dependency-blocking-lease` when it rebounds an in-progress holder whose stale file-scope lease blocks an unmet dependency, and `task:reconcile-dependency-blocking-lease-no-action` when triple-proof blocks that backward move.
|
||||
- FN-6736: self-healing emits `task:reclaim-phantom-executor-binding` when it proves an in-memory executor-active binding is stale, clears the binding, and requeues the in-progress task with worktree/progress preserved.
|
||||
- FN-6783: task-store open and self-healing housekeeping emit `task:reconcile-orphaned-task-dir` when they non-destructively re-import a valid live `.fusion/tasks/{ID}/task.json` directory that has no task row anywhere, preserving soft-deleted/archived/tombstoned IDs.
|
||||
|
||||
|
||||
## Reference docs (deeper detail)
|
||||
|
||||
@@ -672,6 +672,7 @@ Runtime action-gate flow (v1):
|
||||
- `GridlockDetector` (`gridlock-detector.ts`) — detects all-blocked todo pipelines and emits notification events (plus explicit clear signals when gridlock resolves)
|
||||
- `TransientErrorDetector` (`transient-error-detector.ts`) — retriable error classification
|
||||
- `SelfHealingManager` (`self-healing.ts`) — auto-unpause/maintenance recovery actions
|
||||
- Batch 1 maintenance now includes `reconcile-orphaned-task-dirs` (FN-6783), a paused-safe housekeeping step that calls `TaskStore.reconcileOrphanedTaskDirs()` so valid live `.fusion/tasks/{ID}/task.json` records missing from the SQLite index become visible without waiting for process restart. The store-level guard skips any ID already present in active, soft-deleted, archived, or tombstoned storage and emits `task:reconcile-orphaned-task-dir` only for recovered rows.
|
||||
- Batch 1 maintenance now includes one `fts-maintenance` step for both search indexes. The live `tasks_fts` branch still runs `merge` every tick, `optimize` every 4th tick, and `rebuild` above `32 MiB` or `1 MiB × live task count`. The archive `archived_tasks_fts` branch is lighter because archive writes are mostly append-only: `merge` every 8th tick, `optimize` every 24th tick, and `rebuild` above `64 MiB` or `512 KiB × archived row count`. Each branch is independently guarded by `fts5Available` and emits `task:fts-maintenance` run-audit telemetry with distinct `target` values (`tasks_fts` vs `archived_tasks_fts`).
|
||||
- AI merge clean-room worktrees are created under the configured worktrees directory's hidden container, `<worktreesDir>/.ai-merge/`, as `fusion-ai-merge-fn-<id>-<random>` detached worktrees. When that container is repo-local, its relative path is added to the repo's local git exclude when possible (alongside the legacy `.fusion/ai-merge/` entry) so an in-flight clean room does not dirty the integration checkout. After `git worktree add` and before the merge/review loop, `runAiMerge` bootstraps the clean room with the shared merge dependency-sync helper: a configured `worktreeInitCommand` is authoritative and always runs, while unset settings infer `pnpm`/`npm`/`yarn`/`bun` installs from lockfiles and can skip only when the `node_modules/.fusion-install-marker` hash still matches. Failures and aborts hard-stop the AI merge before merge agents or verification run, and `merge:ai-deps-sync` records the command, skip state, and duration. Inline cleanup runs from `runAiMerge`'s clean-room `finally` for successful lands, empty/no-op finalization, concurrent-advance retries, and thrown/aborted merges. Cleanup canonicalizes the path, attempts `git worktree remove --force`, always falls back to filesystem removal, then runs `git worktree prune` so stale or partial registrations (including `git worktree add` failures) do not dangle. Cleanup emits `merge:ai-worktree-cleanup` audit events for git-remove, fs-rm, and prune phases; benign already-absent/de-registered paths are treated as idempotent success, while genuine filesystem-removal failures are logged/audited with `success: false` rather than silently swallowed.
|
||||
- Worktrees-dir sweeps that list direct children of `<worktreesDir>` (pool idle scan, orphan cleanup/reap, self-healing unregistered-orphan reap, and cap enforcement) must exclude the `.ai-merge` container by name; those one-level sweeps never inspect or recycle clean rooms beneath it. Batch 1 sweeps stale AI merge clean-room worktrees under the new `<worktreesDir>/.ai-merge/` root and still scans legacy `.fusion/ai-merge/` plus legacy `tmpdir()` locations for pre-relocation leftovers; candidates are bounded to names starting with `fusion-ai-merge-`. `runAiMerge` registers each live clean-room worktree in `activeSessionRegistry` with kind `ai-merge` as soon as the directory exists and keeps both raw and canonical paths registered for the duration of the merge, so the dedicated periodic sweep and pre-merge prune defer when either path is active (including concurrent same-task merge attempts). The default age gate is 2 hours; task-aware cleanup uses a 10-minute grace period for `done`/`archived` tasks and for genuinely missing/deleted task rows, and every removal path is clamped by the same 10-minute minimum-age floor so a freshly created worktree is never reaped. Transient `getTask` lookup failures (for example SQLite busy/parse errors) are not treated as deletion evidence; they log a warning, emit `lookup-error` only if eventually removed, and retain the conservative 2-hour gate. The sweep canonicalizes paths before checking `activeSessionRegistry`, attempts `git worktree remove --force <path>` before filesystem removal, runs `git worktree prune` after cleanup attempts, and emits `worktree:tempdir-sweep` run-audit telemetry for removal attempts and failures. Fresh directories, active-session paths, and individual removal failures are skipped/logged without aborting the maintenance cycle.
|
||||
@@ -1086,6 +1087,7 @@ The run-audit system records every mutation performed by the engine across four
|
||||
- **Database / `task:soft-delete-column-reconciled`** — emitted by `reconcileSoftDeletedColumnDrift` (FN-5566, re-land FN-5446) when a soft-deleted row (`deletedAt IS NOT NULL`) is found with legacy `column != 'archived'`; rewrites only `column` (no resurrection), with metadata `{ previousColumn }`.
|
||||
- **Database / `session:runtime-resolved`** — emitted once per `createResolvedAgentSession` call with metadata `{ sessionPurpose, runtimeId, wasConfigured, provider, modelId, mockProviderActive, testModeActive, runtimeHint? }` for per-lane runtime/provider attribution.
|
||||
- **Database / `task:reconcile-dependency-blocking-lease`** — emitted by `reconcileDependencyBlockingLeases()` (FN-6292) when self-healing rebounds an `in-progress` holder to `todo` because an unmet dependency is blocked by the holder's stale file-scope lease. Metadata includes the dependency ID, blocked-by marker, and unmet dependency list.
|
||||
- **Database / `task:reconcile-orphaned-task-dir`** — emitted by `TaskStore.reconcileOrphanedTaskDirs()` (FN-6783) when store open or self-healing Batch 1 re-imports a valid live `.fusion/tasks/{ID}/task.json` directory with no SQLite task row anywhere. Metadata includes the recovered ID, column, status, and task JSON path.
|
||||
- **Database / `task:*-no-action` backward-move family (FN-5335)** — backward self-healing sweeps now emit annotation-only events when triple proof fails instead of mutating lifecycle state. New mutation types: `task:reclaim-pr-conflict-no-action`, `task:reclaim-self-owned-branch-conflict-no-action`, `task:auto-rebound-scope-decay-no-action`, `task:finalize-no-op-review-no-action`, `task:stale-incomplete-review-no-action`, `task:ghost-review-no-action`, `task:stuck-merge-deadlock-no-action`, `task:no-progress-no-task-done-no-action`, `task:missing-worktree-review-no-action`, `task:partial-progress-no-task-done-no-action`, `task:reconcile-dependency-blocking-lease-no-action`. See `docs/self-healing-backward-move-audit.md` for per-stage disposition.
|
||||
- **Filesystem** — file:write, prompt:write, attachment:create, etc.
|
||||
- **Sandbox** — backend lifecycle events from `SandboxBackend` wiring in executor/merger/routine-runner (`sandbox:prepare`, `sandbox:run`, `sandbox:failure`, `sandbox:fallback`) introduced after FN-4636.
|
||||
|
||||
@@ -15,6 +15,14 @@
|
||||
- Archived-task flows (`archiveTask`, archived cleanup/migration) still hard-delete from the active `tasks` table after copying to cold storage (`archive.db`).
|
||||
- ID reservation is unchanged: soft-deleted IDs remain reserved. `distributed-task-id` and `task-id-integrity` intentionally scan all task rows (including soft-deleted rows), and must not filter on `deletedAt`.
|
||||
|
||||
### Orphaned task-dir reconciliation (FN-6783)
|
||||
|
||||
- Disk-backed `TaskStore` instances reconcile `.fusion/tasks/{ID}/task.json` directories against the SQLite `tasks` index on store open and during `SelfHealingManager` Batch 1 maintenance (`reconcile-orphaned-task-dirs`). This closes the visibility gap where a heartbeat-created task could exist on disk but be absent from `getTask`/`listTasks` and the dashboard board.
|
||||
- The reconcile is non-destructive: when an ID already exists anywhere the create path would reserve it (active task row, soft-deleted row, archived table/archive DB, or tombstone), the scan skips the directory and never overwrites or resurrects that ID. Only a valid live `task.json` with no DB record anywhere is re-imported.
|
||||
- Recovered rows preserve the on-disk task metadata, including `column`, `status`, dependencies, steps, and log, after the same defensive disk normalization used by task JSON fallback reads. Malformed or unparseable `task.json` files are skipped with a warning instead of failing store open or maintenance.
|
||||
- Recovery is visible: each inserted orphan emits a store warning, a `task:reconcile-orphaned-task-dir` run-audit event, and a `task:created` lifecycle event so live boards can render the recovered card.
|
||||
- On-disk retention matters for scan safety. `deleteTask()` leaves `.fusion/tasks/{ID}/task.json` and `agent-log.jsonl` on disk for forensics while marking the row `deletedAt`; the reconcile must skip those soft-deleted IDs. `archiveTask(id)` with the default cleanup removes the task directory, but `archiveTask(id, false)` and legacy archives can leave a `task.json` behind, so archived IDs are also guarded and skipped.
|
||||
|
||||
### Agent log storage + soft-delete visibility (FN-5143 / FN-5911)
|
||||
|
||||
- Agent logs are no longer stored in SQLite. Each task now appends newline-delimited JSON records to `<rootDir>/.fusion/tasks/{ID}/agent-log.jsonl`.
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
import { mkdtempSync } from "node:fs";
|
||||
import { mkdir, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { afterEach, beforeEach, describe, expect, it } from "vitest";
|
||||
|
||||
import { TaskStore } from "../store.js";
|
||||
import type { Task } from "../types.js";
|
||||
|
||||
async function rewriteTaskJson(rootDir: string, task: Task): Promise<void> {
|
||||
const dir = join(rootDir, ".fusion", "tasks", task.id);
|
||||
await mkdir(dir, { recursive: true });
|
||||
await writeFile(join(dir, "task.json"), JSON.stringify(task), "utf-8");
|
||||
}
|
||||
|
||||
describe("TaskStore orphaned task-dir reconciliation", () => {
|
||||
let rootDir: string;
|
||||
let globalDir: string;
|
||||
let store: TaskStore;
|
||||
|
||||
beforeEach(async () => {
|
||||
rootDir = mkdtempSync(join(tmpdir(), "fusion-orphaned-task-dir-"));
|
||||
globalDir = mkdtempSync(join(tmpdir(), "fusion-orphaned-task-dir-global-"));
|
||||
store = new TaskStore(rootDir, globalDir);
|
||||
await store.init();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
store.close();
|
||||
await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
|
||||
await rm(globalDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 });
|
||||
});
|
||||
|
||||
async function createDiskOnlyTask(id: string, patch: Partial<Task> = {}): Promise<Task> {
|
||||
const task = await store.createTaskWithReservedId(
|
||||
{ description: `Disk-only ${id}`, column: "triage" },
|
||||
{ taskId: id, applyDefaultWorkflowSteps: false, invokeTaskCreatedHook: false },
|
||||
);
|
||||
const diskTask: Task = { ...task, status: "planning", ...patch };
|
||||
await rewriteTaskJson(rootDir, diskTask);
|
||||
(store as any).db.prepare("DELETE FROM tasks WHERE id = ?").run(id);
|
||||
(store as any).db.bumpLastModified();
|
||||
return diskTask;
|
||||
}
|
||||
|
||||
it("re-imports a valid task.json with no DB row so getTask and listTasks agree", async () => {
|
||||
const orphan = await createDiskOnlyTask("FN-9101", {
|
||||
dependencies: ["FN-1"],
|
||||
steps: [{ name: "Preflight", status: "pending" }],
|
||||
});
|
||||
|
||||
expect((await store.listTasks({ includeArchived: false })).some((task) => task.id === orphan.id)).toBe(false);
|
||||
await expect(store.getTask(orphan.id)).rejects.toThrow("Task FN-9101 not found");
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result.recovered).toEqual([orphan.id]);
|
||||
const detail = await store.getTask(orphan.id);
|
||||
expect(detail.id).toBe(orphan.id);
|
||||
expect(detail.column).toBe("triage");
|
||||
expect(detail.status).toBe("planning");
|
||||
expect(detail.dependencies).toEqual(["FN-1"]);
|
||||
expect((await store.listTasks({ includeArchived: false })).map((task) => task.id)).toContain(orphan.id);
|
||||
expect(store.getRunAuditEvents({ taskId: orphan.id, mutationType: "task:reconcile-orphaned-task-dir" })).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("re-imports orphaned task dirs during disk-backed store open", async () => {
|
||||
const orphan = await createDiskOnlyTask("FN-9102");
|
||||
|
||||
store.close();
|
||||
store = new TaskStore(rootDir, globalDir);
|
||||
await store.init();
|
||||
|
||||
expect((await store.getTask(orphan.id)).status).toBe("planning");
|
||||
expect((await store.listTasks({ includeArchived: false })).map((task) => task.id)).toContain(orphan.id);
|
||||
});
|
||||
|
||||
it("does not overwrite an already-present DB row", async () => {
|
||||
const task = await store.createTaskWithReservedId(
|
||||
{ description: "Authoritative DB row", title: "Original title" },
|
||||
{ taskId: "FN-9103", applyDefaultWorkflowSteps: false, invokeTaskCreatedHook: false },
|
||||
);
|
||||
await rewriteTaskJson(rootDir, { ...task, title: "Disk drift title", description: "Disk drift" });
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result.recovered).not.toContain(task.id);
|
||||
const detail = await store.getTask(task.id);
|
||||
expect(detail.title).toBe("Original title");
|
||||
expect(detail.description).toBe("Authoritative DB row");
|
||||
});
|
||||
|
||||
it("skips soft-deleted and tombstoned task IDs without resurrection", async () => {
|
||||
const task = await store.createTaskWithReservedId(
|
||||
{ description: "Delete me" },
|
||||
{ taskId: "FN-9104", applyDefaultWorkflowSteps: false, invokeTaskCreatedHook: false },
|
||||
);
|
||||
await store.deleteTask(task.id);
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result.recovered).not.toContain(task.id);
|
||||
expect((await store.listTasks({ includeArchived: true })).map((candidate) => candidate.id)).not.toContain(task.id);
|
||||
await expect(store.getTask(task.id)).rejects.toThrow("Task FN-9104 not found");
|
||||
expect(await store.getTask(task.id, { includeDeleted: true })).toMatchObject({ id: task.id, deletedAt: expect.any(String) });
|
||||
});
|
||||
|
||||
it("skips archived IDs that still have or regain a task.json", async () => {
|
||||
const task = await store.createTaskWithReservedId(
|
||||
{ description: "Archive me" },
|
||||
{ taskId: "FN-9105", applyDefaultWorkflowSteps: false, invokeTaskCreatedHook: false },
|
||||
);
|
||||
await store.archiveTask(task.id, true);
|
||||
await rewriteTaskJson(rootDir, { ...task, column: "triage", status: "planning" });
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result.recovered).not.toContain(task.id);
|
||||
expect((await store.listTasks({ includeArchived: false })).map((candidate) => candidate.id)).not.toContain(task.id);
|
||||
expect((await store.listTasks({ includeArchived: true })).map((candidate) => candidate.id)).toContain(task.id);
|
||||
expect((await store.getTask(task.id)).column).toBe("archived");
|
||||
});
|
||||
|
||||
it("skips malformed task.json and directories without task.json without throwing", async () => {
|
||||
const malformedDir = join(rootDir, ".fusion", "tasks", "FN-9106");
|
||||
await mkdir(malformedDir, { recursive: true });
|
||||
await writeFile(join(malformedDir, "task.json"), "{ nope", "utf-8");
|
||||
await mkdir(join(rootDir, ".fusion", "tasks", "FN-9107"), { recursive: true });
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result.recovered).toEqual([]);
|
||||
expect(result.skipped).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: "FN-9106", reason: expect.stringContaining("malformed-task-json") }),
|
||||
{ id: "FN-9107", reason: "missing-task-json" },
|
||||
]));
|
||||
});
|
||||
|
||||
it("reports malformed live task metadata without overwriting the DB row", async () => {
|
||||
const task = await store.createTaskWithReservedId(
|
||||
{ description: "Malformed file but valid DB" },
|
||||
{ taskId: "FN-9108", applyDefaultWorkflowSteps: false, invokeTaskCreatedHook: false },
|
||||
);
|
||||
await rewriteTaskJson(rootDir, { ...task, createdAt: "riage-FN-6750-1781908063" });
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result.recovered).not.toContain(task.id);
|
||||
expect(result.skipped).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: task.id, reason: expect.stringContaining("malformed-task-metadata") }),
|
||||
]));
|
||||
expect((await store.getTask(task.id)).createdAt).toBe(task.createdAt);
|
||||
});
|
||||
|
||||
it("is a safe no-op for in-memory stores even if task dirs exist", async () => {
|
||||
store.close();
|
||||
store = new TaskStore(rootDir, globalDir, { inMemoryDb: true });
|
||||
await store.init();
|
||||
const now = new Date().toISOString();
|
||||
await rewriteTaskJson(rootDir, {
|
||||
id: "FN-9109",
|
||||
description: "Ignored in-memory orphan",
|
||||
column: "triage",
|
||||
dependencies: [],
|
||||
steps: [],
|
||||
currentStep: 0,
|
||||
log: [],
|
||||
createdAt: now,
|
||||
updatedAt: now,
|
||||
columnMovedAt: now,
|
||||
status: "planning",
|
||||
});
|
||||
|
||||
const result = await store.reconcileOrphanedTaskDirs();
|
||||
|
||||
expect(result).toEqual({ recovered: [], skipped: [] });
|
||||
expect((await store.listTasks({ includeArchived: false })).map((task) => task.id)).not.toContain("FN-9109");
|
||||
});
|
||||
});
|
||||
@@ -2,7 +2,7 @@ import { EventEmitter } from "node:events";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdir, readdir, readFile, writeFile, rename, unlink } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { existsSync, watch, type FSWatcher } from "node:fs";
|
||||
import { existsSync, watch, type Dirent, type FSWatcher } from "node:fs";
|
||||
import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, ColumnId, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, GithubIssueAction, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, HandoffToReviewOptions, GoalCitation, GoalCitationFilter, GoalCitationInput, GoalCitationSurface, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, TaskBranchAssignmentMode, MergeRequestRecord, MergeRequestState, MergeRequestWorkflowProjectionOptions, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind, WorkflowWorkItemState, WorkflowWorkItemTransitionPatch, WorkflowWorkItemUpsertInput, PrEntity, PrEntityCreateInput, PrEntityUpdate, PrEntityState, PrThreadState, PrThreadOutcome, PrConflictState, PrChecksRollup, PrReviewDecision, PluginActivation, PluginActivationInput } from "./types.js";
|
||||
import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js";
|
||||
import { VALID_TRANSITIONS, COLUMNS, DEFAULT_SETTINGS, isColumn, isGlobalOnlySettingsKey, WORKFLOW_STEP_TEMPLATES, validateDocumentKey } from "./types.js";
|
||||
@@ -1865,6 +1865,14 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
|
||||
await this.importLegacyAgentLogsOnce();
|
||||
this.taskIdStateReconciled = false;
|
||||
this.reconcileDistributedTaskIdStateOnOpen();
|
||||
try {
|
||||
await this.reconcileOrphanedTaskDirs();
|
||||
} catch (err) {
|
||||
storeLog.warn("Orphaned task-dir reconcile failed during init (non-fatal)", {
|
||||
phase: "init:orphaned-task-dir-reconcile",
|
||||
error: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
}
|
||||
|
||||
// Write config.json for backward compatibility if it doesn't exist
|
||||
if (!existsSync(this.configPath)) {
|
||||
@@ -3175,6 +3183,159 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
|
||||
* Read a task from SQLite by ID (extracted from dir path for backward compat).
|
||||
* Falls back to file-based reading only when no DB row exists at all.
|
||||
*/
|
||||
private normalizeTaskFromDisk(task: Task): Task {
|
||||
if (!Array.isArray(task.log)) task.log = [];
|
||||
if (!Array.isArray(task.dependencies)) task.dependencies = [];
|
||||
if (!Array.isArray(task.steps)) task.steps = [];
|
||||
task.priority = normalizeTaskPriority(task.priority);
|
||||
return task;
|
||||
}
|
||||
|
||||
private getMalformedTaskMetadataReason(task: Partial<Task>, expectedId: string): string | undefined {
|
||||
if (task.id !== expectedId) {
|
||||
return `task.json id ${typeof task.id === "string" ? task.id : "<missing>"} does not match directory ${expectedId}`;
|
||||
}
|
||||
if (typeof task.description !== "string") {
|
||||
return "task.json description must be a string";
|
||||
}
|
||||
if (typeof task.column !== "string") {
|
||||
return "task.json column must be a string";
|
||||
}
|
||||
if (typeof task.createdAt !== "string" || Number.isNaN(Date.parse(task.createdAt))) {
|
||||
return "task.json createdAt must be a valid ISO timestamp string";
|
||||
}
|
||||
if (typeof task.updatedAt !== "string" || Number.isNaN(Date.parse(task.updatedAt))) {
|
||||
return "task.json updatedAt must be a valid ISO timestamp string";
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/*
|
||||
* FNXC:TaskStoreConsistency 2026-06-20-00:00:
|
||||
* Heartbeat-created tasks persisted on disk but missing from the SQLite index were invisible to fn_task_list/fn_task_show (FN-6783/FN-6784). Reconcile re-imports orphaned task.json rows non-destructively and uses the same exists-anywhere guard as create-time ID allocation so soft-deleted, archived, and tombstoned IDs are never resurrected.
|
||||
*/
|
||||
async reconcileOrphanedTaskDirs(): Promise<{ recovered: string[]; skipped: Array<{ id: string; reason: string }> }> {
|
||||
const result: { recovered: string[]; skipped: Array<{ id: string; reason: string }> } = {
|
||||
recovered: [],
|
||||
skipped: [],
|
||||
};
|
||||
|
||||
if (this.inMemoryDb || !existsSync(this.tasksDir)) {
|
||||
return result;
|
||||
}
|
||||
|
||||
let entries: Dirent[];
|
||||
try {
|
||||
entries = await readdir(this.tasksDir, { withFileTypes: true });
|
||||
} catch (error) {
|
||||
storeLog.warn("Skipping orphaned task-dir reconcile because tasksDir is unreadable", {
|
||||
phase: "reconcileOrphanedTaskDirs:scan",
|
||||
tasksDir: this.tasksDir,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
return result;
|
||||
}
|
||||
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory()) continue;
|
||||
const id = entry.name;
|
||||
const taskDir = join(this.tasksDir, id);
|
||||
const taskJsonPath = join(taskDir, "task.json");
|
||||
if (!existsSync(taskJsonPath)) {
|
||||
result.skipped.push({ id, reason: "missing-task-json" });
|
||||
continue;
|
||||
}
|
||||
|
||||
let task: Task;
|
||||
try {
|
||||
const raw = await readFile(taskJsonPath, "utf-8");
|
||||
task = this.normalizeTaskFromDisk(JSON.parse(raw) as Task);
|
||||
} catch (error) {
|
||||
const reason = `malformed-task-json: ${error instanceof Error ? error.message : String(error)}`;
|
||||
result.skipped.push({ id, reason });
|
||||
storeLog.warn("Skipping malformed task.json during orphaned task-dir reconcile", {
|
||||
phase: "reconcileOrphanedTaskDirs:parse",
|
||||
taskId: id,
|
||||
taskJsonPath,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
const malformedReason = this.getMalformedTaskMetadataReason(task, id);
|
||||
if (malformedReason) {
|
||||
result.skipped.push({ id, reason: `malformed-task-metadata: ${malformedReason}` });
|
||||
storeLog.warn("Skipping malformed task metadata during orphaned task-dir reconcile", {
|
||||
phase: "reconcileOrphanedTaskDirs:validate",
|
||||
taskId: id,
|
||||
taskJsonPath,
|
||||
reason: malformedReason,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
let recovered = false;
|
||||
let skipReason: string | undefined;
|
||||
try {
|
||||
this.db.transactionImmediate(() => {
|
||||
if (this.taskIdExistsAnywhere(id)) {
|
||||
skipReason = "id-exists-anywhere";
|
||||
return;
|
||||
}
|
||||
try {
|
||||
this.insertTaskWithFtsRecovery(task, "reconcileOrphanedTaskDirs");
|
||||
this.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 (this.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 });
|
||||
storeLog.warn("Skipping orphaned task-dir reconcile insert after non-fatal error", {
|
||||
phase: "reconcileOrphanedTaskDirs:insert",
|
||||
taskId: id,
|
||||
taskJsonPath,
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
if (recovered) {
|
||||
result.recovered.push(id);
|
||||
if (this.isWatching) this.taskCache.set(id, { ...task });
|
||||
storeLog.warn("Recovered orphaned task.json into SQLite task index", {
|
||||
phase: "reconcileOrphanedTaskDirs:recovered",
|
||||
taskId: id,
|
||||
column: task.column,
|
||||
status: task.status,
|
||||
taskJsonPath,
|
||||
});
|
||||
this.emitTaskLifecycleEventSafely("task:created", [task]);
|
||||
} else {
|
||||
result.skipped.push({ id, reason: skipReason ?? "not-recovered" });
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private async readTaskJson(dir: string): Promise<Task> {
|
||||
const id = this.getTaskIdFromDir(dir);
|
||||
|
||||
@@ -3190,12 +3351,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
|
||||
const filePath = join(dir, "task.json");
|
||||
const raw = await readFile(filePath, "utf-8");
|
||||
try {
|
||||
const fileTask = JSON.parse(raw) as Task;
|
||||
if (!Array.isArray(fileTask.log)) fileTask.log = [];
|
||||
if (!Array.isArray(fileTask.dependencies)) fileTask.dependencies = [];
|
||||
if (!Array.isArray(fileTask.steps)) fileTask.steps = [];
|
||||
fileTask.priority = normalizeTaskPriority(fileTask.priority);
|
||||
return fileTask;
|
||||
return this.normalizeTaskFromDisk(JSON.parse(raw) as Task);
|
||||
} catch (err) {
|
||||
throw new Error(
|
||||
`Failed to parse task.json at ${filePath}: ${(err as Error).message}`,
|
||||
|
||||
@@ -0,0 +1,40 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { SelfHealingManager } from "../self-healing.js";
|
||||
|
||||
describe("SelfHealingManager orphaned task-dir maintenance", () => {
|
||||
it("runs orphaned task-dir reconcile in pause-safe housekeeping", async () => {
|
||||
const reconcileOrphanedTaskDirs = vi.fn(async () => ({
|
||||
recovered: ["FN-9201"],
|
||||
skipped: [],
|
||||
}));
|
||||
const store = {
|
||||
getSettings: vi.fn(async () => ({
|
||||
maintenanceIntervalMs: 0,
|
||||
globalPause: true,
|
||||
enginePaused: false,
|
||||
chatAutoCleanupDays: 0,
|
||||
mailAutoCleanupDays: 0,
|
||||
operationalLogRetentionDays: 0,
|
||||
agentLogFileRetentionDays: 0,
|
||||
})),
|
||||
reconcileOrphanedTaskDirs,
|
||||
pruneOperationalLogs: vi.fn(() => ({ deletedTotal: 0, deletedByTable: {} })),
|
||||
pruneAgentLogFiles: vi.fn(() => ({ prunedFiles: 0, prunedEntries: 0, freedBytes: 0 })),
|
||||
} as any;
|
||||
|
||||
const manager = new SelfHealingManager(store, { rootDir: "/tmp/fusion-self-healing-orphaned-task-dirs-test" });
|
||||
vi.spyOn(manager as any, "pruneWorktrees").mockResolvedValue(undefined);
|
||||
vi.spyOn(manager as any, "cleanupOrphans").mockResolvedValue(undefined);
|
||||
vi.spyOn(manager as any, "cleanupStaleTempMergeWorktrees").mockResolvedValue(0);
|
||||
vi.spyOn(manager as any, "cleanupOrphanedBranches").mockResolvedValue(undefined);
|
||||
vi.spyOn(manager as any, "maintainTaskFts").mockResolvedValue(undefined);
|
||||
vi.spyOn(manager as any, "checkpointWal").mockReturnValue(undefined);
|
||||
vi.spyOn(manager as any, "enforceWorktreeCap").mockResolvedValue(undefined);
|
||||
vi.spyOn(manager, "archiveStaleDoneTasks").mockResolvedValue(0);
|
||||
|
||||
await (manager as any).runMaintenance();
|
||||
|
||||
expect(reconcileOrphanedTaskDirs).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
@@ -2004,6 +2004,20 @@ export class SelfHealingManager {
|
||||
},
|
||||
},
|
||||
{ name: "cleanup-orphaned-branches", fn: () => this.cleanupOrphanedBranches() },
|
||||
{
|
||||
name: "reconcile-orphaned-task-dirs",
|
||||
fn: async () => {
|
||||
/*
|
||||
* FNXC:TaskStoreConsistency 2026-06-20-00:00:
|
||||
* Runtime heartbeat-created task dirs can appear after store init, so paused-safe housekeeping must reconcile orphaned task.json rows during maintenance instead of waiting for a restart.
|
||||
*/
|
||||
const result = await this.store.reconcileOrphanedTaskDirs();
|
||||
if (result.recovered.length > 0 || result.skipped.some((entry) => entry.reason.startsWith("malformed"))) {
|
||||
log.warn(`Maintenance batch 1 step "reconcile-orphaned-task-dirs" recovered=${result.recovered.length} skipped=${result.skipped.length}`);
|
||||
}
|
||||
return result;
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "cleanup-old-chats",
|
||||
fn: async () => {
|
||||
|
||||
Reference in New Issue
Block a user