fix(dataflow): close lifecycle and worker hazards (Phase 0 W2/W2-bis/W3)

Three related stability fixes that the v10 deprecation work needs to
lean on. Each one closes a real bug that was discovered during plan
validation:

W2 — onunload async cleanup race
The plugin's onunload() fired dataflowOrchestrator.cleanup() with a
floating .catch() and never awaited it, so workers, event refs, and
debounced timers could outlive Obsidian's teardown. Now the async
cleanup is captured into a `plugin.unloadComplete` promise that
tests and any external observer can await. Repository.cleanup()
also unloads the underlying TaskIndexer Component, fixing a real
listener leak: TaskIndexer's vault modify/delete/create handlers
were never deregistered across plugin reloads.

The obsidian mock's Component.unload() previously didn't drain
registered events, hiding leak bugs in jest. Fixed in the mock so
listener-leak tests work for any future cleanup work.

W2-bis — last-resort cleanup on rebuild failure
Orchestrator.rebuild() now wraps its work in try/catch. On any
thrown error every cache namespace is cleared so the next plugin
load can't read a partial cache that looks complete. Snapshot/restore
is deferred to a later phase; the namespace clear is the floor.

W3 — per-task worker timeout with kill + main-thread fallback
TaskWorkerManager now arms a per-task timeout (default 8s,
configurable) when dispatching work. On timeout the hung worker is
terminated, the active promise rejects with WorkerTimeoutError,
a replacement worker is spawned, and the queue keeps draining.
Previously a hung worker would hold its slot until the 30s circuit
breaker tripped after 10 failures.

WorkerOrchestrator counts WorkerTimeoutError separately in
metrics.taskWorkerTimeouts and skips its retry-with-backoff path
for timeouts (a worker that just hung will likely hang on the
same file again — fall back to main thread immediately).

22/22 integration tests pass (Lifecycle, Rebuild, WorkerTimeout, smoke).

src/__mocks__/obsidian.ts

@@ -395,6 +395,18 @@ export class Component {
 	unload(): void {
 		this.loaded = false;
 		this.children.forEach((child) => child.unload());
+		// Drain registered event refs by calling their unload() — this matches
+		// real Obsidian Component behavior so that listener-leak tests work.
+		// Without this, code that correctly uses registerEvent() still appears
+		// to leak in jest because the mock never calls eventRef.unload().
+		for (const ref of this._events) {
+			try {
+				ref.unload();
+			} catch {
+				/* listener cleanup must not throw */
+			}
+		}
+		this._events = [];
 		this.onunload();
 	}
 

src/__tests__/integration/Lifecycle.unload.test.ts

@@ -0,0 +1,98 @@
+/**
+ * Phase 0 W2 — verify the dataflow orchestrator cleans up cleanly.
+ *
+ * Background: prior to W2, src/index.ts:1758 fired `dataflowOrchestrator.cleanup()`
+ * without awaiting it, so workers / event refs / debounced timers could outlive
+ * Obsidian's onunload contract. The fix exposes `plugin.unloadComplete` as a
+ * promise that resolves once async cleanup work has settled.
+ *
+ * This test boots a real orchestrator via the W0 fixture, simulates teardown,
+ * and asserts the event-bus listener counts on the fake app are zero. We don't
+ * test the plugin class directly (that requires booting the entire plugin under
+ * jsdom which the test infrastructure isn't set up for), but the orchestrator
+ * is the only async work in onunload — its cleanup is what unloadComplete
+ * actually awaits.
+ */
+
+import { buildOrchestrator } from "./_fixtures/buildOrchestrator";
+
+describe("Orchestrator lifecycle (W2)", () => {
+	it("registers workspace listeners and removes them on cleanup", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one", "b.md": "- [x] two" },
+		});
+
+		// After construction the orchestrator subscribes to workspace events
+		// for things like ICS updates and write operations. The exact count is
+		// an implementation detail; just assert it's > 0 so the test would
+		// catch a regression where listeners stop attaching at all (which would
+		// silently break the dataflow).
+		const beforeCleanup = fx.app.__totalListenerCount();
+		expect(beforeCleanup).toBeGreaterThan(0);
+
+		await fx.dispose();
+
+		// After dispose, every listener attached via the orchestrator's
+		// eventRefs (which cleanup walks via app.workspace.offref) must be gone.
+		// Vault listeners attached by sub-sources should also be gone.
+		const afterCleanup = fx.app.__totalListenerCount();
+		expect(afterCleanup).toBe(0);
+	});
+
+	it("dispose is idempotent and does not throw on a re-dispose", async () => {
+		const fx = await buildOrchestrator({ files: { "a.md": "- [ ] one" } });
+		await fx.dispose();
+		// Second dispose path: orchestrator was already nulled out, so the
+		// fixture's stored reference still points to the old instance. Calling
+		// cleanup() on it again should not throw.
+		await expect(
+			(fx.orchestrator as any).cleanup(),
+		).resolves.not.toThrow();
+	});
+
+	it("file events fire to listeners before dispose, but not after", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one" },
+		});
+
+		const events: string[] = [];
+		fx.vault.on("modify", (file: any) => {
+			events.push(`modify:${file.path}`);
+		});
+
+		await fx.writeFile("a.md", "- [x] one");
+		expect(events).toContain("modify:a.md");
+
+		await fx.dispose();
+
+		// Note: the W0 fixture's vault still has its own listener bus. The
+		// "modify" listener we attached in this test is on the fake vault,
+		// not on the orchestrator's eventRefs, so it's still active here.
+		// What we care about is that the orchestrator-level listeners are gone
+		// (covered by the first test) and that the orchestrator no longer
+		// processes new vault events that arrive post-cleanup.
+		events.length = 0;
+		await fx.writeFile("a.md", "- [ ] one");
+		// Test-local listener still fires, that's fine.
+		expect(events).toEqual(["modify:a.md"]);
+	});
+
+	it("cleanup awaits Repository.cleanup (the async cost in onunload)", async () => {
+		const fx = await buildOrchestrator();
+		// Spy on the repository cleanup method (accessed via the QueryAPI's
+		// repository) to confirm it's awaited.
+		const repo = (fx.orchestrator as any).repository;
+		expect(repo).toBeDefined();
+		const repoCleanupSpy = jest.spyOn(repo, "cleanup");
+
+		await fx.dispose();
+
+		expect(repoCleanupSpy).toHaveBeenCalledTimes(1);
+		// And the spy must have been called and resolved BEFORE dispose returned.
+		// Jest's awaited spies satisfy that just by virtue of dispose having
+		// returned. If repo.cleanup() were fire-and-forget the spy would still
+		// be in a pending state when dispose returned, but Jest can't observe
+		// that distinction directly — the orchestrator's own `await` enforces it.
+		repoCleanupSpy.mockRestore();
+	});
+});

src/__tests__/integration/Rebuild.failure.test.ts

@@ -0,0 +1,93 @@
+/**
+ * Phase 0 W2-bis — verify that a failed rebuild doesn't leave a partial cache.
+ *
+ * The fix in src/dataflow/Orchestrator.ts wraps rebuild() in try/catch. On any
+ * thrown error, every cache namespace is cleared so the next plugin load
+ * triggers a clean rebuild from disk instead of trusting a partial cache.
+ *
+ * The test injects a failure into processBatch (the most realistic failure
+ * point) by spying on the orchestrator method, then asserts:
+ *  - rebuild() rejects with the original error
+ *  - storage.clearNamespace was called for all 4 namespaces
+ */
+
+import { buildOrchestrator } from "./_fixtures/buildOrchestrator";
+
+describe("Orchestrator.rebuild failure handling (W2-bis)", () => {
+	it("clears all cache namespaces on rebuild failure and re-throws", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one", "b.md": "- [x] two" },
+		});
+
+		// Reach into the orchestrator's storage to spy on namespace clears.
+		const storage = (fx.orchestrator as any).storage;
+		expect(storage).toBeDefined();
+		const clearSpy = jest.spyOn(storage, "clearNamespace");
+
+		// Inject a failure: replace processBatch with a thrower. This is the
+		// most realistic failure mode (a worker error, a vault read error,
+		// or a parse error mid-batch).
+		const boom = new Error("simulated worker crash");
+		(fx.orchestrator as any).processBatch = jest.fn(async () => {
+			throw boom;
+		});
+
+		await expect(fx.orchestrator.rebuild()).rejects.toBe(boom);
+
+		// All four namespaces must have been cleared as the last-resort path.
+		const clearedNamespaces = clearSpy.mock.calls.map(
+			(call: any[]) => call[0],
+		);
+		expect(clearedNamespaces).toEqual(
+			expect.arrayContaining(["raw", "augmented", "project", "consolidated"]),
+		);
+
+		clearSpy.mockRestore();
+		await fx.dispose();
+	});
+
+	it("does not clear namespaces on a successful rebuild", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one" },
+		});
+
+		const storage = (fx.orchestrator as any).storage;
+		const clearSpy = jest.spyOn(storage, "clearNamespace");
+
+		// processBatch is the slow path; replace with a no-op so rebuild
+		// completes quickly without exercising worker plumbing.
+		(fx.orchestrator as any).processBatch = jest.fn(async () => {});
+
+		await expect(fx.orchestrator.rebuild()).resolves.toBeUndefined();
+
+		// On success, the catch branch should NOT have run, so clearNamespace
+		// should not have been called from rebuild's catch handler. Note: if
+		// repository.clear() internally calls clearNamespace, this assertion
+		// would fail; our implementation calls repository.clear() (a different
+		// path), not storage.clearNamespace, so this is safe.
+		expect(clearSpy).not.toHaveBeenCalled();
+
+		clearSpy.mockRestore();
+		await fx.dispose();
+	});
+
+	it("re-throws even when last-resort cleanup itself fails", async () => {
+		const fx = await buildOrchestrator({ files: { "a.md": "- [ ] one" } });
+
+		const storage = (fx.orchestrator as any).storage;
+		// Make every clearNamespace call fail. The catch-of-catch path should
+		// still re-throw the original error.
+		jest.spyOn(storage, "clearNamespace").mockRejectedValue(
+			new Error("storage is on fire"),
+		);
+
+		const boom = new Error("primary failure");
+		(fx.orchestrator as any).processBatch = jest.fn(async () => {
+			throw boom;
+		});
+
+		await expect(fx.orchestrator.rebuild()).rejects.toBe(boom);
+
+		await fx.dispose();
+	});
+});

src/__tests__/integration/WorkerTimeout.test.ts

@@ -0,0 +1,178 @@
+/**
+ * Phase 0 W3 — verify TaskWorkerManager per-task timeout fires, terminates the
+ * hung worker, and rejects with WorkerTimeoutError; verify WorkerOrchestrator
+ * counts the timeout in its metrics and falls back to main thread.
+ *
+ * Two test groups:
+ *  1. TaskWorkerManager: monkey-patch the underlying worker.worker.postMessage
+ *     to a no-op so the worker never responds, then drive the per-task timeout
+ *     with a short workerTimeoutMs and assert the rejection shape.
+ *  2. WorkerOrchestrator: stub a TaskWorkerManager whose processFile rejects
+ *     with WorkerTimeoutError, assert metrics.taskWorkerTimeouts increments
+ *     and the main-thread fallback runs.
+ */
+
+import { TaskWorkerManager } from "@/dataflow/workers/TaskWorkerManager";
+import { WorkerOrchestrator } from "@/dataflow/workers/WorkerOrchestrator";
+import { WorkerTimeoutError } from "@/dataflow/workers/errors";
+
+// Build a minimal TFile-shaped object — TaskWorkerManager only reads .path,
+// .extension, and .stat from it during processFile.
+function fakeFile(path: string, content = "- [ ] task") {
+	return {
+		path,
+		basename: path.replace(/\.md$/, ""),
+		extension: "md",
+		name: path.split("/").pop() ?? path,
+		parent: null,
+		stat: { mtime: 1, ctime: 1, size: content.length },
+	} as any;
+}
+
+describe("TaskWorkerManager per-task timeout (W3)", () => {
+	it("rejects active task with WorkerTimeoutError when worker hangs", async () => {
+		// Minimal vault/metadataCache mocks. Only methods TaskWorkerManager
+		// reaches into during processFile/getTaskMetadata are needed.
+		const vault: any = {
+			cachedRead: async () => "- [ ] task",
+		};
+		const metadataCache: any = {
+			getFileCache: () => null,
+		};
+
+		// Construct manager with a 100ms timeout — long enough that the
+		// jest event loop can schedule it, short enough that the test stays fast.
+		const mgr = new TaskWorkerManager(vault, metadataCache, {
+			maxWorkers: 1,
+			cpuUtilization: 1,
+			workerTimeoutMs: 100,
+		});
+
+		// Force-spawn one worker by reaching into the private state, then
+		// monkey-patch its underlying worker.postMessage to do nothing so the
+		// worker never responds. (Workers are constructed eagerly in the
+		// constructor via initializeWorkers().)
+		const workersMap: Map<number, any> = (mgr as any).workers;
+		expect(workersMap.size).toBeGreaterThan(0);
+		for (const w of workersMap.values()) {
+			w.worker.postMessage = jest.fn(); // black hole
+		}
+
+		const file = fakeFile("hang.md");
+		const result = mgr.processFile(file).then(
+			(value) => ({ kind: "resolved", value }),
+			(error) => ({ kind: "rejected", error }),
+		);
+
+		// Wait long enough for the timeout to fire (100ms + slack)
+		await new Promise((resolve) => setTimeout(resolve, 200));
+		const settled = await result;
+
+		expect(settled.kind).toBe("rejected");
+		const err = (settled as any).error;
+		expect(err).toBeInstanceOf(WorkerTimeoutError);
+		expect(err.filePath).toBe("hang.md");
+		expect(err.timeoutMs).toBe(100);
+
+		// Timeout was counted on the manager.
+		expect(mgr.getTimeoutCount()).toBe(1);
+
+		// Cleanup
+		(mgr as any).onunload();
+	});
+
+	it("spawns a replacement worker after a timeout so the pool stays warm", async () => {
+		const vault: any = {
+			cachedRead: async () => "- [ ] task",
+		};
+		const metadataCache: any = {
+			getFileCache: () => null,
+		};
+
+		const mgr = new TaskWorkerManager(vault, metadataCache, {
+			maxWorkers: 1,
+			cpuUtilization: 1,
+			workerTimeoutMs: 80,
+		});
+
+		const workersMap: Map<number, any> = (mgr as any).workers;
+		const initialIds = new Set([...workersMap.keys()]);
+		for (const w of workersMap.values()) {
+			w.worker.postMessage = jest.fn();
+		}
+
+		await mgr
+			.processFile(fakeFile("hang.md"))
+			.catch(() => undefined); // we expect rejection
+
+		// Give the timeout time to fire and replacement to spawn
+		await new Promise((resolve) => setTimeout(resolve, 200));
+
+		// After the timeout: the pool size should be back to maxWorkers (1)
+		// and the worker IDs should NOT all match the initial set (one was
+		// replaced). The timeout counter should be exactly 1.
+		expect(workersMap.size).toBe(1);
+		expect(mgr.getTimeoutCount()).toBe(1);
+		const finalIds = new Set([...workersMap.keys()]);
+		const allReplaced = [...finalIds].every((id) => !initialIds.has(id));
+		expect(allReplaced).toBe(true);
+
+		(mgr as any).onunload();
+	});
+});
+
+describe("WorkerOrchestrator counts WorkerTimeoutError separately (W3)", () => {
+	it("increments metrics.taskWorkerTimeouts and falls back to main thread", async () => {
+		// Stub a TaskWorkerManager whose processFile always rejects with the
+		// timeout error. The orchestrator should catch it, bump the counter,
+		// and fall through to parseFileTasksMainThread.
+		const stubTaskMgr: any = {
+			processFile: jest.fn(async (file: any) => {
+				throw new WorkerTimeoutError(file.path, 100);
+			}),
+			processBatch: jest.fn(),
+			isProcessingBatchTask: () => false,
+			getPendingTaskCount: () => 0,
+			getBatchProgress: () => ({ current: 0, total: 0, percentage: 0 }),
+			getStats: () => ({}),
+		};
+		const stubProjectMgr: any = {
+			getProjectData: jest.fn(),
+			getBatchProjectData: jest.fn(),
+			isWorkersEnabled: () => true,
+			getMemoryStats: () => ({}),
+		};
+
+		const orch = new WorkerOrchestrator(stubTaskMgr, stubProjectMgr, {
+			enableWorkerProcessing: true,
+		});
+
+		// Patch the orchestrator's main-thread fallback so the test doesn't
+		// need a real ConfigurableTaskParser. We just need to confirm it's
+		// called and returns something.
+		const fallbackResult: any[] = [{ id: "fallback", content: "ok" }];
+		(orch as any).parseFileTasksMainThread = jest.fn(async () => {
+			return fallbackResult;
+		});
+
+		const file = {
+			path: "hang.md",
+			extension: "md",
+			stat: { mtime: 1, ctime: 1, size: 1 },
+		} as any;
+		const result = await orch.parseFileTasks(file);
+
+		// Fallback was reached
+		expect(result).toBe(fallbackResult);
+		expect(
+			(orch as any).parseFileTasksMainThread,
+		).toHaveBeenCalledTimes(1);
+
+		// Metric was incremented
+		const metrics = orch.getMetrics();
+		expect(metrics.taskWorkerTimeouts).toBe(1);
+		// And the generic failure count went up too — timeouts are a subset
+		// of failures, not a replacement for them.
+		expect(metrics.taskParsingFailures).toBeGreaterThanOrEqual(1);
+	});
+});

src/__tests__/integration/_fixtures/buildOrchestrator.ts

@@ -86,7 +86,17 @@ class EventBus {
 	on(name: string, handler: (...args: any[]) => void) {
 		if (!this.handlers.has(name)) this.handlers.set(name, new Set());
 		this.handlers.get(name)!.add(handler);
-		return { name, handler } as any;
+		// Return an EventRef-shaped object. Both `name`/`handler` (used by
+		// our offref()) AND a working `unload()` method (used by the obsidian
+		// mock's Component.unload to drain registered events) must be present.
+		const self = this;
+		return {
+			name,
+			handler,
+			unload() {
+				self.off(name, handler);
+			},
+		} as any;
 	}
 
 	off(name: string, handler: (...args: any[]) => void) {

src/dataflow/Orchestrator.ts

@@ -1699,36 +1699,67 @@ export class DataflowOrchestrator {
 	}
 
 	/**
-	 * Clear all data and rebuild
+	 * Clear all data and rebuild.
+	 *
+	 * If rebuild fails partway through (e.g. a worker crash, vault read error,
+	 * or persistence failure), we don't want to leave the cache in a partial
+	 * state — the next plugin load would think it's loading a complete index
+	 * but actually be missing files. As a last-resort, on any thrown error we
+	 * clear every cache namespace so the next load triggers a clean rebuild
+	 * from disk. The error is re-thrown so the caller can react.
+	 *
+	 * This is W2-bis in the v10 Phase 0 plan. It's intentionally minimal —
+	 * snapshot/restore is deferred to a future phase.
 	 */
 	async rebuild(): Promise<void> {
-		// Clear all data
-		await this.repository.clear();
+		try {
+			// Clear all data
+			await this.repository.clear();
 
-		// Process all markdown and canvas files
-		const files = this.vault.getMarkdownFiles();
-		const canvasFiles = this.vault
-			.getFiles()
-			.filter((f) => f.extension === "canvas");
+			// Process all markdown and canvas files
+			const files = this.vault.getMarkdownFiles();
+			const canvasFiles = this.vault
+				.getFiles()
+				.filter((f) => f.extension === "canvas");
 
-		const allFiles = [...files, ...canvasFiles];
+			const allFiles = [...files, ...canvasFiles];
 
-		// Process in batches for performance
-		const BATCH_SIZE = 50;
-		for (let i = 0; i < allFiles.length; i += BATCH_SIZE) {
-			const batch = allFiles.slice(i, i + BATCH_SIZE);
-			await this.processBatch(batch);
+			// Process in batches for performance
+			const BATCH_SIZE = 50;
+			for (let i = 0; i < allFiles.length; i += BATCH_SIZE) {
+				const batch = allFiles.slice(i, i + BATCH_SIZE);
+				await this.processBatch(batch);
+			}
+
+			// Persist the rebuilt index
+			await this.repository.persist();
+
+			// Emit ready event
+			emit(this.app, Events.CACHE_READY, {
+				initial: false,
+				timestamp: Date.now(),
+				seq: Seq.next(),
+			});
+		} catch (err) {
+			console.error(
+				"[DataflowOrchestrator] Rebuild failed; clearing caches to force a clean rebuild on next load:",
+				err,
+			);
+			// Last-resort cleanup: clear every namespace so the next plugin load
+			// can't read a partial cache that looks complete.
+			try {
+				await this.storage.clearNamespace("raw");
+				await this.storage.clearNamespace("augmented");
+				await this.storage.clearNamespace("project");
+				await this.storage.clearNamespace("consolidated");
+			} catch (clearErr) {
+				console.error(
+					"[DataflowOrchestrator] Last-resort cache clear ALSO failed; cache may be in inconsistent state:",
+					clearErr,
+				);
+			}
+			throw err;
 		}
-
-		// Persist the rebuilt index
-		await this.repository.persist();
-
-		// Emit ready event
-		emit(this.app, Events.CACHE_READY, {
-			initial: false,
-			timestamp: Date.now(),
-			seq: Seq.next(),
-		});
 	}
 
 	/**

src/dataflow/indexer/Repository.ts

@@ -562,11 +562,27 @@ export class Repository {
 	}
 
 	/**
-	 * Cleanup and ensure all pending data is persisted
+	 * Cleanup and ensure all pending data is persisted.
+	 *
+	 * IMPORTANT: also unloads the underlying TaskIndexer (a Component) so its
+	 * vault listeners (modify/delete/create) are deregistered. Prior to this,
+	 * those listeners leaked across plugin reloads — TaskIndexer was constructed
+	 * eagerly in the Repository constructor but never explicitly unloaded.
+	 * See W2 in the v10 Phase 0 plan.
 	 */
 	async cleanup(): Promise<void> {
 		// Execute any pending persist operations
 		await this.executePersist();
+
+		// Tear down the indexer's Component lifecycle so its registered events
+		// (vault.on modify/delete/create in TaskIndexer.setupEventListeners)
+		// are released. The Component base class' unload() walks _events and
+		// calls offref on each.
+		try {
+			(this.indexer as any).unload?.();
+		} catch (error) {
+			console.error("[Repository] Error unloading TaskIndexer:", error);
+		}
 	}
 
 	/**

src/dataflow/workers/TaskWorkerManager.ts

@@ -28,6 +28,7 @@ import {
 // @ts-ignore Ignore type error for worker import
 import TaskWorker from "./TaskIndex.worker";
 import { Deferred, deferred } from "./deferred-promise";
+import { WorkerTimeoutError } from "./errors";
 
 // Using similar queue structure as importer.ts
 import { Queue } from "@datastructures-js/queue";
@@ -42,6 +43,14 @@ export interface WorkerPoolOptions {
 	cpuUtilization: number;
 	/** Whether to enable debug logging */
 	debug?: boolean;
+	/**
+	 * Per-task timeout in milliseconds. If a worker doesn't post a response
+	 * within this window the worker is terminated, the active task's promise
+	 * is rejected with WorkerTimeoutError, and a replacement worker is spawned.
+	 * Default: 8000 (8s) — enough headroom for a worst-case 200KB markdown
+	 * parse on a slow desktop. See W3 in v10 Phase 0 plan.
+	 */
+	workerTimeoutMs?: number;
 	/** Settings for the task indexer */
 	settings?: {
 		preferMetadataFormat: "dataview" | "tasks";
@@ -62,6 +71,14 @@ export interface WorkerPoolOptions {
 	};
 }
 
+/**
+ * Default per-task timeout in ms for the task worker pool. A debug override
+ * can be supplied via `(globalThis as any).__taskGeniusDebug?.workerTimeoutMs`
+ * which the constructor reads at boot — used by integration tests to shrink
+ * the timeout to a couple hundred ms.
+ */
+export const DEFAULT_TASK_WORKER_TIMEOUT_MS = 8000;
+
 /**
  * Default worker pool options
  */
@@ -103,6 +120,12 @@ interface PoolWorker {
 	availableAt: number;
 	/** The active task this worker is processing, if any */
 	active?: [TFile, Deferred<any>, number, TaskPriority];
+	/**
+	 * Per-task timeout handle. Set when a task is dispatched to this worker
+	 * (in schedule()), cleared when finish() runs, fires onTaskTimeout() if
+	 * the worker doesn't respond within workerTimeoutMs. See W3 in plan.
+	 */
+	timeoutHandle?: ReturnType<typeof setTimeout>;
 }
 
 /**
@@ -169,6 +192,10 @@ export class TaskWorkerManager extends Component {
 	private initialized: boolean = false;
 	/** Reference to task indexer for cache checking */
 	private taskIndexer?: any;
+	/** Effective per-task timeout in ms (resolved from options + debug override) */
+	private workerTimeoutMs: number;
+	/** Count of timed-out tasks. Exposed via getStats() and the orchestrator metrics. */
+	private timeoutCount: number = 0;
 	/** Performance statistics */
 	private stats = {
 		filesSkipped: 0,
@@ -189,10 +216,36 @@ export class TaskWorkerManager extends Component {
 		this.vault = vault;
 		this.metadataCache = metadataCache;
 
+		// Resolve effective per-task timeout. Precedence:
+		//   1. options.workerTimeoutMs (explicit)
+		//   2. debug override on globalThis (for tests)
+		//   3. DEFAULT_TASK_WORKER_TIMEOUT_MS
+		const debugOverride =
+			(globalThis as any).__taskGeniusDebug?.workerTimeoutMs;
+		this.workerTimeoutMs =
+			this.options.workerTimeoutMs ??
+			(typeof debugOverride === "number" ? debugOverride : undefined) ??
+			DEFAULT_TASK_WORKER_TIMEOUT_MS;
+
 		// Initialize workers up to max
 		this.initializeWorkers();
 	}
 
+	/**
+	 * Get the count of timed-out tasks since the last reset.
+	 * Used by WorkerOrchestrator metrics.
+	 */
+	public getTimeoutCount(): number {
+		return this.timeoutCount;
+	}
+
+	/**
+	 * Reset the timeout counter (e.g. when metrics are reset).
+	 */
+	public resetTimeoutCount(): void {
+		this.timeoutCount = 0;
+	}
+
 	/**
 	 * Set file parsing configuration
 	 */
@@ -689,6 +742,14 @@ export class TaskWorkerManager extends Component {
 
 		const {file, promise, priority} = queueItem;
 		worker.active = [file, promise, 0, priority]; // 0 表示重试次数
+		// Arm a per-task timeout. If the worker doesn't respond within
+		// workerTimeoutMs, onTaskTimeout will terminate it, reject the promise
+		// with WorkerTimeoutError, spawn a replacement, and continue scheduling.
+		// Cleared in finish() and on direct terminate().
+		worker.timeoutHandle = setTimeout(
+			() => this.onTaskTimeout(worker),
+			this.workerTimeoutMs,
+		);
 
 		try {
 			this.getTaskMetadata(file)
@@ -722,6 +783,10 @@ export class TaskWorkerManager extends Component {
 				})
 				.catch((error) => {
 					console.error(`Error reading file ${file.path}:`, error);
+					if (worker.timeoutHandle) {
+						clearTimeout(worker.timeoutHandle);
+						worker.timeoutHandle = undefined;
+					}
 					promise.reject(error);
 					worker.active = undefined;
 
@@ -733,6 +798,10 @@ export class TaskWorkerManager extends Component {
 				});
 		} catch (error) {
 			console.error(`Error processing file ${file.path}:`, error);
+			if (worker.timeoutHandle) {
+				clearTimeout(worker.timeoutHandle);
+				worker.timeoutHandle = undefined;
+			}
 			promise.reject(error);
 			worker.active = undefined;
 
@@ -751,6 +820,14 @@ export class TaskWorkerManager extends Component {
 		worker: PoolWorker,
 		data: IndexerResult
 	): Promise<void> {
+		// Cancel the timeout before doing any work — the worker did respond,
+		// so we no longer want to consider it hung. Even if the response is
+		// itself an error, the timeout path is no longer correct here.
+		if (worker.timeoutHandle) {
+			clearTimeout(worker.timeoutHandle);
+			worker.timeoutHandle = undefined;
+		}
+
 		if (!worker.active) {
 			console.log("Received a stale worker message. Ignoring.", data);
 			return;
@@ -881,6 +958,13 @@ export class TaskWorkerManager extends Component {
 	 * Terminate a worker
 	 */
 	private terminate(worker: PoolWorker): void {
+		// Always clear the timeout — orphaned timers are exactly what plugin
+		// reload races complain about.
+		if (worker.timeoutHandle) {
+			clearTimeout(worker.timeoutHandle);
+			worker.timeoutHandle = undefined;
+		}
+
 		worker.worker.terminate();
 
 		if (worker.active) {
@@ -891,6 +975,68 @@ export class TaskWorkerManager extends Component {
 		this.log(`Terminated worker #${worker.id}`);
 	}
 
+	/**
+	 * Called when a worker doesn't respond within workerTimeoutMs.
+	 *
+	 * Strategy: kill the worker, reject the active promise with
+	 * WorkerTimeoutError so the WorkerOrchestrator can fall back to main-thread
+	 * processing for this file, then spawn a replacement worker so the pool
+	 * stays warm. Without termination, a hung worker would hold its slot and
+	 * starve the queue until the existing 30s circuit breaker tripped.
+	 */
+	private onTaskTimeout(worker: PoolWorker): void {
+		// Defensive: if the timeout fires after finish() already cleared the
+		// handle, do nothing.
+		if (!worker.active) return;
+
+		const [file, promise] = worker.active;
+		console.warn(
+			`[TaskWorkerManager] Worker #${worker.id} timed out on ${file.path} after ${this.workerTimeoutMs}ms; terminating and replacing.`,
+		);
+
+		this.timeoutCount++;
+
+		// Terminate the hung worker. Note: terminate() will also clear the
+		// handle (no-op here since we just fired) and reject worker.active[1]
+		// with "Terminated", but we want a more specific error, so we reject
+		// here first then null out worker.active before terminate runs.
+		try {
+			worker.worker.terminate();
+		} catch (e) {
+			console.error("[TaskWorkerManager] Error terminating worker:", e);
+		}
+
+		// Remove from pool
+		this.workers.delete(worker.id);
+
+		// Reject with the typed error so the orchestrator can distinguish
+		// timeouts from other failures.
+		promise.reject(new WorkerTimeoutError(file.path, this.workerTimeoutMs));
+		this.outstanding.delete(file.path);
+		worker.active = undefined;
+		worker.timeoutHandle = undefined;
+
+		// Keep the pool warm: if we're below capacity (we just dropped one),
+		// spawn a replacement so subsequent tasks aren't starved.
+		if (this.active && this.workers.size < this.options.maxWorkers) {
+			try {
+				const replacement = this.newWorker();
+				this.workers.set(replacement.id, replacement);
+				this.log(
+					`Spawned replacement worker #${replacement.id} after timeout`,
+				);
+			} catch (e) {
+				console.error(
+					"[TaskWorkerManager] Failed to spawn replacement worker after timeout:",
+					e,
+				);
+			}
+		}
+
+		// Drain the queue with whatever workers remain.
+		this.schedule();
+	}
+
 	/**
 	 * Clean up existing workers without affecting the active state
 	 */

src/dataflow/workers/WorkerOrchestrator.ts

@@ -5,6 +5,7 @@ import { TaskWorkerManager, DEFAULT_WORKER_OPTIONS } from "./TaskWorkerManager";
 import { ProjectDataWorkerManager } from "./ProjectDataWorkerManager";
 import { MetadataParseMode } from "../../types/TaskParserConfig";
 import { ConfigurableTaskParser } from "@/dataflow/core/ConfigurableTaskParser";
+import { WorkerTimeoutError } from "./errors";
 
 /**
  * WorkerOrchestrator - Unified task and project worker management
@@ -24,6 +25,9 @@ export class WorkerOrchestrator {
 	private metrics = {
 		taskParsingSuccess: 0,
 		taskParsingFailures: 0,
+		// Subset of taskParsingFailures: hung-worker timeouts (W3). Tracked
+		// separately so we can tell "worker hung" from "worker errored".
+		taskWorkerTimeouts: 0,
 		projectDataSuccess: 0,
 		projectDataFailures: 0,
 		averageTaskParsingTime: 0,
@@ -87,8 +91,12 @@ export class WorkerOrchestrator {
 				error
 			);
 
-			// Update failure metrics
+			// Update failure metrics. Track timeouts separately so observability
+			// can distinguish "worker hung" from "worker threw" (W3).
 			this.metrics.taskParsingFailures++;
+			if (error instanceof WorkerTimeoutError) {
+				this.metrics.taskWorkerTimeouts++;
+			}
 			this.handleWorkerFailure();
 
 			// Fallback to main thread
@@ -132,8 +140,12 @@ export class WorkerOrchestrator {
 				error
 			);
 
-			// Update failure metrics
+			// Update failure metrics. A timeout in a batch is counted once for
+			// the offending file, not for the whole batch.
 			this.metrics.taskParsingFailures += files.length;
+			if (error instanceof WorkerTimeoutError) {
+				this.metrics.taskWorkerTimeouts++;
+			}
 			this.handleWorkerFailure();
 
 			// Fallback to main thread
@@ -220,7 +232,12 @@ export class WorkerOrchestrator {
 	}
 
 	/**
-	 * Generic retry mechanism with exponential backoff
+	 * Generic retry mechanism with exponential backoff.
+	 *
+	 * Skips retries for WorkerTimeoutError: a worker that just hung is being
+	 * replaced anyway, and the file content is the most likely cause of the
+	 * hang. Retrying would just waste 7s of exponential backoff before falling
+	 * back to main thread, which is the right answer for hung tasks. (W3)
 	 */
 	private async retryOperation<T>(
 		operation: () => Promise<T>,
@@ -248,6 +265,11 @@ export class WorkerOrchestrator {
 					error
 				);
 
+				// Don't retry timeout errors — see method docs.
+				if (error instanceof WorkerTimeoutError) {
+					break;
+				}
+
 				// If this is the last attempt, don't wait
 				if (attempt === maxRetries) {
 					break;
@@ -525,6 +547,7 @@ export class WorkerOrchestrator {
 		this.metrics = {
 			taskParsingSuccess: 0,
 			taskParsingFailures: 0,
+			taskWorkerTimeouts: 0,
 			projectDataSuccess: 0,
 			projectDataFailures: 0,
 			averageTaskParsingTime: 0,

src/dataflow/workers/errors.ts

@@ -0,0 +1,24 @@
+/**
+ * Worker-related error types.
+ *
+ * Phase 0 W3: distinguish "worker hung past its timeout" from other worker
+ * failures so the WorkerOrchestrator can:
+ *  - increment a dedicated metric (`workerTimeouts`) separate from generic
+ *    parsing failures
+ *  - decide whether the fallback to main thread is appropriate
+ *  - allow tests to assert specifically on the timeout path
+ */
+
+export class WorkerTimeoutError extends Error {
+	readonly filePath: string;
+	readonly timeoutMs: number;
+
+	constructor(filePath: string, timeoutMs: number) {
+		super(
+			`Task worker timed out after ${timeoutMs}ms while processing ${filePath}`,
+		);
+		this.name = "WorkerTimeoutError";
+		this.filePath = filePath;
+		this.timeoutMs = timeoutMs;
+	}
+}

src/index.ts

@@ -159,6 +159,13 @@ export default class TaskProgressBarPlugin extends Plugin {
 	// Write API for dataflow architecture
 	writeAPI?: WriteAPI;
 
+	// Resolves once all async cleanup work scheduled by onunload() has settled.
+	// Obsidian's onunload() signature is sync (void), so async cleanup work has
+	// to be fired off without awaiting. Tests and any code that needs to know
+	// when the plugin is fully torn down should `await plugin.unloadComplete`.
+	// Reset on each onload(); see W2 in the v10 Phase 0 plan.
+	public unloadComplete: Promise<void> = Promise.resolve();
+
 	// Notification manager (desktop)
 	notificationManager?: DesktopIntegrationManager;
 
@@ -1746,30 +1753,49 @@ export default class TaskProgressBarPlugin extends Plugin {
 	}
 
 	onunload() {
-		// Clean up global suggest manager
+		// Synchronous cleanup paths run immediately. Asynchronous cleanup
+		// (currently just dataflowOrchestrator.cleanup() which awaits Repository
+		// persistence) is gathered into a single promise exposed as
+		// `unloadComplete` so tests and any external observer can await full
+		// teardown. Obsidian itself never awaits this — its onunload signature
+		// is sync — but at least listeners + workers + persistence get a chance
+		// to settle before the next plugin lifecycle, instead of racing.
+		const asyncTasks: Array<Promise<void>> = [];
+
+		// Clean up global suggest manager (sync)
 		if (this.globalSuggestManager) {
 			this.globalSuggestManager.cleanup();
 		}
 
 		// Bases views are automatically unregistered by Obsidian when plugin unloads
 
-		// Clean up dataflow orchestrator (experimental)
+		// Clean up dataflow orchestrator (async — capture into asyncTasks)
 		if (this.dataflowOrchestrator) {
-			this.dataflowOrchestrator.cleanup().catch((error) => {
-				console.error(
-					"Error cleaning up dataflow orchestrator:",
-					error,
-				);
-			});
-			// Set to undefined to prevent any further access
+			const orch = this.dataflowOrchestrator;
+			// Null out immediately so any other code path that fires during
+			// teardown can't reach into a half-cleaned-up orchestrator.
 			this.dataflowOrchestrator = undefined;
+			asyncTasks.push(
+				orch.cleanup().catch((error) => {
+					console.error(
+						"Error cleaning up dataflow orchestrator:",
+						error,
+					);
+				}),
+			);
 		}
 
-		// Clean up MCP server manager (desktop only)
+		// Clean up MCP server manager (desktop only, sync)
 		if (this.mcpServerManager) {
 			this.mcpServerManager.cleanup();
 		}
+
 		// Task Genius Icon Manager cleanup is handled automatically by Component system
+
+		// Expose a promise so tests / external observers can know when async
+		// cleanup is fully done. Never rejects — individual catches above
+		// already log errors.
+		this.unloadComplete = Promise.all(asyncTasks).then(() => undefined);
 	}
 
 	/**