Skip to content

fix(task-history): atomic per-task merge and drop shared index file (#1231) #1261

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
edelauna wants to merge 5 commits into
base: main
Choose a base branch
from
Open

fix(task-history): atomic per-task merge and drop shared index file (#1231) #1261

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
aee7d64
fix(task-history): prevent concurrent index clobbering
edelauna Aug 16, 2026
2fe1189
test(task-history): add regression coverage for self-healing index re…
edelauna Aug 18, 2026
4e6fa84
fix(task-history): atomic read-modify-write
edelauna Aug 18, 2026
cb04fd4
fix(task-history): atomic read-modify-write
edelauna Aug 18, 2026
98cd2bc
fix(task-history): drop _index.json, scan task dirs on read
edelauna Aug 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
211 changes: 55 additions & 156 deletions src/core/task-persistence/TaskHistoryStore.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,15 +33,6 @@ export function assertValidTransition(from: HistoryItemStatus | undefined, to: H
}
}

/**
* Index file format for fast startup reads.
*/
interface HistoryIndex {
version: number
updatedAt: number
entries: HistoryItem[]
}

/**
* Durable intent for the one repair that spans an active delegated child and
* its parent. Task files remain authoritative; this file only records the
Expand Down Expand Up @@ -75,12 +66,14 @@ interface DelegationRepairIntent {
*
* Each task's HistoryItem is stored as an individual JSON file in its
* existing task directory (`globalStorage/tasks/<taskId>/history_item.json`).
* A single index file (`globalStorage/tasks/_index.json`) is maintained
* as a cache for fast list reads at startup.
* There is no shared index file. Reads scan the task directories.
*
* Cross-process safety comes from `safeWriteJson`'s `proper-lockfile`
* on per-task file writes. Within a single extension host process,
* an in-process write lock serializes mutations.
* Cross-process safety for per-task files comes from `safeWriteJson`'s
* `proper-lockfile` with a `merge` callback: each write reads the
* current file under the advisory lock and merges incoming fields, so
* a concurrent writer's changes are preserved rather than silently
* dropped. Within a single extension host process, an in-process write
* lock serializes mutations.
*/
/**
* Options for TaskHistoryStore constructor.
Expand All @@ -100,7 +93,6 @@ export class TaskHistoryStore {
private cache: Map<string, HistoryItem> = new Map()
private taskFileMtimes: Map<string, number> = new Map()
private writeLock: Promise<void> = Promise.resolve()
private indexWriteTimer: ReturnType<typeof setTimeout> | null = null
private fsWatcher: fsSync.FSWatcher | null = null
private reconcileTimer: ReturnType<typeof setTimeout> | null = null
private disposed = false
Expand All @@ -112,9 +104,6 @@ export class TaskHistoryStore {
public readonly initialized: Promise<void>
private resolveInitialized!: () => void

/** Debounce window for index writes in milliseconds. */
private static readonly INDEX_WRITE_DEBOUNCE_MS = 2000

/** Periodic reconciliation interval in milliseconds. */
private static readonly RECONCILE_INTERVAL_MS = 5 * 60 * 1000

Expand All @@ -136,30 +125,27 @@ export class TaskHistoryStore {
const tasksDir = await this.getTasksDir()
await fs.mkdir(tasksDir, { recursive: true })

// 1. Load existing index into the cache
await this.loadIndex()

// 2. Reconcile cache against actual task directories on disk
// 1. Scan task directories to populate the cache
await this.reconcile({ forceRefresh: true })
// Capture which active tasks were present in persisted state before replay can
// change any statuses. Reconciliation must not treat a replay-repaired parent
// as an orphaned active child in the same startup pass.
const persistedActiveIds = this.getPersistedActiveIds()

// 3. Complete any two-record repair interrupted after its intent was durable.
// 2. Complete any two-record repair interrupted after its intent was durable.
try {
await this.replayDelegationRepairIntent()
} catch (error) {
console.error("[TaskHistoryStore] Failed to replay delegation repair intent:", error)
}

// 4. Repair delegation inconsistencies left by a previous crash
// 3. Repair delegation inconsistencies left by a previous crash
await this.reconcileDelegationState(persistedActiveIds)

// 5. Start fs.watch for cross-instance reactivity
// 4. Start fs.watch for cross-instance reactivity
this.startWatcher()

// 6. Start periodic reconciliation as a defensive fallback
// 5. Start periodic reconciliation as a defensive fallback
this.startPeriodicReconciliation()
} finally {
// Mark initialization as complete so callers awaiting `initialized` can proceed
Expand All @@ -173,11 +159,6 @@ export class TaskHistoryStore {
dispose(): void {
this.disposed = true

if (this.indexWriteTimer) {
clearTimeout(this.indexWriteTimer)
this.indexWriteTimer = null
}

if (this.reconcileTimer) {
clearTimeout(this.reconcileTimer)
this.reconcileTimer = null
Expand All @@ -187,11 +168,6 @@ export class TaskHistoryStore {
this.fsWatcher.close()
this.fsWatcher = null
}

// Synchronously flush the index (best-effort)
this.flushIndex().catch((err) => {
console.error("[TaskHistoryStore] Error flushing index on dispose:", err)
})
}

// ────────────────────────────── Reads ──────────────────────────────
Expand Down Expand Up @@ -257,13 +233,18 @@ export class TaskHistoryStore {
// Merge: preserve existing metadata unless explicitly overwritten
const merged = existing ? { ...existing, ...item } : item

// Write per-task file (source of truth)
await this.writeTaskFile(merged)
// Compute the actual changed fields relative to the cached state.
// Only these are applied to the disk version, so fields updated by
// another process are preserved rather than reverted from a stale cache.
const delta = existing
? Object.fromEntries(
Object.entries(item).filter(([k, v]) => !deepEqual(v, (existing as Record<string, unknown>)[k])),
)
: undefined
await this.writeTaskFile(merged, delta ? ({ id: item.id, ...delta } as HistoryItem) : undefined)

// Update in-memory cache
this.cache.set(merged.id, merged)
// Schedule debounced index write
this.scheduleIndexWrite()

const all = this.getAll()

Expand Down Expand Up @@ -291,8 +272,6 @@ export class TaskHistoryStore {
// File may already be deleted
}

this.scheduleIndexWrite()

// Call onWrite callback inside the lock for serialized write-through
if (this.onWrite) {
await this.onWrite(this.getAll())
Expand All @@ -317,8 +296,6 @@ export class TaskHistoryStore {
}
}

this.scheduleIndexWrite()

// Call onWrite callback inside the lock for serialized write-through
if (this.onWrite) {
await this.onWrite(this.getAll())
Expand Down Expand Up @@ -346,20 +323,18 @@ export class TaskHistoryStore {
return // tasks dir doesn't exist yet
}

// Filter out the index file and hidden files
// Filter out hidden and reserved names
const taskDirNames = dirEntries.filter((name) => !name.startsWith("_") && !name.startsWith("."))

const onDiskIds = new Set(taskDirNames)
const cacheIds = new Set(this.cache.keys())
let changed = false
const liveIds = new Set<string>()

// Task files are authoritative during startup. Later watcher and periodic
// reconciliations use mtime change detection to avoid rewriting the index when
// nothing changed on disk.
for (const taskId of onDiskIds) {
try {
const taskFilePath = await this.getTaskFilePath(taskId)
const { mtimeMs } = await fs.stat(taskFilePath)
liveIds.add(taskId)
if (
!options.forceRefresh &&
this.cache.has(taskId) &&
Expand All @@ -374,26 +349,20 @@ export class TaskHistoryStore {
this.taskFileMtimes.set(taskId, mtimeMs)
if (!deepEqual(previous, item)) {
this.cache.set(taskId, item)
changed = true
}
}
} catch {
// Corrupted or missing file, skip
// history_item.json missing or corrupt — not live
}
}

// Tasks in cache but not on disk: remove from cache
// Evict tasks whose history_item.json no longer exists
for (const taskId of cacheIds) {
if (!onDiskIds.has(taskId)) {
if (!liveIds.has(taskId)) {
this.cache.delete(taskId)
this.taskFileMtimes.delete(taskId)
changed = true
}
}

if (changed) {
this.scheduleIndexWrite()
}
})
}

Expand Down Expand Up @@ -584,12 +553,6 @@ export class TaskHistoryStore {
await this.onWrite(this.getAll())
}
await this.removeDelegationRepairIntent()
// Task files are authoritative and the intent is the recovery journal.
// Clean up the journal before scheduling the derived index: a crash after
// cleanup but before the index write is safe because startup rebuilds the
// index from task files, while the reverse ordering could make the index
// appear durable before recovery metadata is settled.
this.scheduleIndexWrite()
})
}

Expand Down Expand Up @@ -645,9 +608,6 @@ export class TaskHistoryStore {
await this.onWrite(this.getAll())
}
await this.removeDelegationRepairIntent()
// The index is derived state; keep the intent until authoritative task-file
// writes and write-through have completed, then schedule the index update.
this.scheduleIndexWrite()
}

private matchesDelegationRepairParentPreconditions(intent: DelegationRepairIntent, parent: HistoryItem): boolean {
Expand Down Expand Up @@ -845,96 +805,36 @@ export class TaskHistoryStore {
}
}

// Write the index
await this.writeIndex()

// Repair any delegation inconsistencies introduced by the migrated entries.
// Run the lock-free core because migration already holds the store lock.
await this.reconcileDelegationStateCore(this.getPersistedActiveIds())
})
}

// ────────────────────────────── Private: Index management ──────────────────────────────

/**
* Load the `_index.json` file into the in-memory cache.
*/
private async loadIndex(): Promise<void> {
const indexPath = await this.getIndexPath()

try {
const raw = await fs.readFile(indexPath, "utf8")
const index: HistoryIndex = JSON.parse(raw)

if (index.version === 1 && Array.isArray(index.entries)) {
for (const entry of index.entries) {
if (entry.id) {
this.cache.set(entry.id, entry)
}
}
}
} catch {
// Index doesn't exist or is corrupted; cache stays empty.
// Reconciliation will rebuild it from per-task files.
}
}

/**
* Write the full index to disk.
*/
private async writeIndex(): Promise<void> {
const indexPath = await this.getIndexPath()
const index: HistoryIndex = {
version: 1,
updatedAt: Date.now(),
entries: this.getAll(),
}

await safeWriteJson(indexPath, index)
}

/**
* Schedule a debounced index write.
*/
private scheduleIndexWrite(): void {
if (this.disposed) {
return
}

if (this.indexWriteTimer) {
clearTimeout(this.indexWriteTimer)
}

this.indexWriteTimer = setTimeout(async () => {
this.indexWriteTimer = null
try {
await this.writeIndex()
} catch (err) {
console.error("[TaskHistoryStore] Failed to write index:", err)
}
}, TaskHistoryStore.INDEX_WRITE_DEBOUNCE_MS)
}

/**
* Force an immediate index write (called on dispose/shutdown).
*/
async flushIndex(): Promise<void> {
if (this.indexWriteTimer) {
clearTimeout(this.indexWriteTimer)
this.indexWriteTimer = null
}

await this.writeIndex()
}

// ────────────────────────────── Private: Per-task file I/O ──────────────────────────────

/**
* Write a HistoryItem to its per-task `history_item.json` file.
*
* When `delta` is provided, the merge callback applies only the
* delta to the current disk state, so fields written by another
* process are preserved. Without a delta the full item is written
* as-is (used by administrative repair paths that are authoritative).
*/
private async writeTaskFile(item: HistoryItem): Promise<void> {
private async writeTaskFile(item: HistoryItem, delta?: Partial<HistoryItem>): Promise<void> {
const filePath = await this.getTaskFilePath(item.id)
await safeWriteJson(filePath, item)
if (delta) {
await safeWriteJson(filePath, item, {
merge: (existing, incoming) => {

@martin-rueegg martin-rueegg Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if it would be good to extract the merge functionality into a local function of it's own to make it testable in isolation

if (!existing || typeof existing !== "object" || !("id" in existing)) {
return incoming
}
return { ...existing, ...delta }
},
})
} else {
await safeWriteJson(filePath, item)
}
}

/**
Expand Down Expand Up @@ -1105,16 +1005,23 @@ export class TaskHistoryStore {
const mergedFirst = { ...first, ...updatedFirst }
const mergedSecond = { ...second, ...updatedSecond }

// Compute actual diffs against cached state, mirroring upsertCore.
const deltaFirst = Object.fromEntries(
Object.entries(updatedFirst).filter(([k, v]) => !deepEqual(v, (first as Record<string, unknown>)[k])),

@martin-rueegg martin-rueegg Aug 18, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

to have a single point of truth, AND make it testable in isolation, would it not be good to extract the three delta calculations into one local private method?
same for line 1013 and 241?

)
const deltaSecond = Object.fromEntries(
Object.entries(updatedSecond).filter(([k, v]) => !deepEqual(v, (second as Record<string, unknown>)[k])),
)

// Write both files before touching the cache so readers never observe a
// half-updated in-memory state between the two await points.
await this.writeTaskFile(mergedFirst)
await this.writeTaskFile(mergedSecond)
await this.writeTaskFile(mergedFirst, { id: firstId, ...deltaFirst } as HistoryItem)
await this.writeTaskFile(mergedSecond, { id: secondId, ...deltaSecond } as HistoryItem)

// Both disk writes succeeded — now update the cache atomically.
this.cache.set(firstId, mergedFirst)
this.cache.set(secondId, mergedSecond)

this.scheduleIndexWrite()
const all = this.getAll()
if (this.onWrite) {
await this.onWrite(all)
Expand Down Expand Up @@ -1155,12 +1062,4 @@ export class TaskHistoryStore {
const tasksDir = await this.getTasksDir()
return path.join(tasksDir, taskId, GlobalFileNames.historyItem)
}

/**
* Get the path to the `_index.json` file.
*/
private async getIndexPath(): Promise<string> {
const tasksDir = await this.getTasksDir()
return path.join(tasksDir, GlobalFileNames.historyIndex)
}
}
Loading
Loading