test(dataflow): critical-path integration tests (Phase 0 W5)

Adds three integration tests that exercise the full dataflow pipeline
end-to-end. These are the vital-signs tests that any future Phase 1
refactor needs to keep green.

W5.1 — Orchestrator.roundtrip.test.ts (4 tests)
The single most important test. Drives parse → augment → cache → query
through processFileImmediate (bypassing the 300ms debounce), with
workers forced off so parsing routes through ConfigurableTaskParser
in main thread. Asserts that:
  - tasks added to a file end up in the index
  - file modifications are reflected in the next query
  - file deletions remove tasks from the index
  - dispose releases all event listeners

W5.2 — Orchestrator.settingsChange.test.ts (4 tests)
Verifies cache invalidation on settings change. Phase 1 settings
consolidation will rely on this contract:
  - parser scope clears the raw namespace
  - augment scope clears augmented + project namespaces
  - SETTINGS_CHANGED event fires with the scopes payload
  - typed onSettingsFieldsChanged path also clears the right caches

W5.4 — CacheInvariants.sequence.test.ts (2 tests)
Drives the cache invariants checker through a realistic sequence
of operations (process / modify / delete / settings-change) and
asserts the checker reports ok at every step. The smoke test in
CacheInvariants.smoke.test.ts validates the checker's correctness;
this validates it doesn't false-positive on normal usage. I3
(indexer ↔ augmented namespace agreement) is filtered out as a
known transitional concern documented in invariants.ts.

Plan W5.3 (worker fallback) is already covered by WorkerTimeout.test.ts
which exercises the orchestrator's main-thread fallback path on a
WorkerTimeoutError. Plan W5.5 (migration tombstone) is already covered
by legacy-bundle-0.test.ts which parameterizes over real fixture data.

Also adds vault.adapter.{stat,exists} to FakeVault — needed by
processFileImmediate for mtime cache validation.

10 new tests; 83/83 Phase 0 tests pass; full suite stable at 39
pre-existing failures (1417/1557 pass, +71 from baseline 1346).

src/__tests__/integration/CacheInvariants.sequence.test.ts

@@ -0,0 +1,102 @@
+/**
+ * Phase 0 W5.4 — Cache invariants hold at every step of a realistic sequence.
+ *
+ * The smoke test (CacheInvariants.smoke.test.ts) verifies the checker logic
+ * itself. This test verifies that the checker doesn't false-positive on
+ * normal pipeline operation: a fresh vault, processing several files,
+ * modifying some, deleting some, running a settings change. After every step
+ * the invariants should hold (modulo I3 indexer-vs-augmented drift, which is
+ * expected during transitional states and is documented in invariants.ts).
+ *
+ * Phase 1 deprecation work that breaks invariants will be caught here.
+ */
+
+import { buildOrchestrator } from "./_fixtures/buildOrchestrator";
+import { checkCacheInvariants } from "@/dataflow/cache/invariants";
+
+// I3 (indexer ↔ augmented namespace agreement) is informational rather than a
+// hard error during transitional states. Filter it out for sequence assertions.
+function nonTransientViolations(report: { violations: any[] }) {
+	return report.violations.filter(
+		(v: any) => v.id !== "I3-index-augmented-drift",
+	);
+}
+
+describe("CacheInvariants — realistic sequence (W5.4)", () => {
+	it("invariants hold across process / modify / delete / settings-change", async () => {
+		const fx = await buildOrchestrator({
+			files: {
+				"a.md": "- [ ] alpha\n",
+				"b.md": "- [x] beta\n",
+				"c.md": "no tasks\n",
+			},
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			// --- Step 0: fresh orchestrator, nothing in cache yet ---
+			let report = await checkCacheInvariants(fx.orchestrator);
+			expect(nonTransientViolations(report)).toEqual([]);
+
+			// --- Step 1: process a, b, c ---
+			for (const path of ["a.md", "b.md", "c.md"]) {
+				const file = fx.vault.getFileByPath(path)!;
+				await (fx.orchestrator as any).processFileImmediate(file, true);
+			}
+			report = await checkCacheInvariants(fx.orchestrator);
+			expect(nonTransientViolations(report)).toEqual([]);
+			// Both files with tasks should be in raw and augmented
+			expect(report.stats.rawCount).toBeGreaterThanOrEqual(2);
+			expect(report.stats.augmentedCount).toBeGreaterThanOrEqual(2);
+			expect(report.stats.missingAugmented).toBe(0);
+
+			// --- Step 2: modify a ---
+			const fileA = fx.vault.getFileByPath("a.md")!;
+			await fx.vault.modify(fileA, "- [ ] alpha\n- [ ] alpha2\n");
+			await (fx.orchestrator as any).processFileImmediate(fileA, true);
+			report = await checkCacheInvariants(fx.orchestrator);
+			expect(nonTransientViolations(report)).toEqual([]);
+
+			// --- Step 3: delete b ---
+			const repo: any = (fx.orchestrator as any).repository;
+			await repo.removeFile("b.md");
+			report = await checkCacheInvariants(fx.orchestrator);
+			expect(nonTransientViolations(report)).toEqual([]);
+
+			// --- Step 4: settings change (parser scope) — clears raw namespace ---
+			// Block rebuild so we don't reprocess everything in this test.
+			(fx.orchestrator as any).rebuild = jest.fn(async () => {});
+			await fx.orchestrator.onSettingsChange(["parser"]);
+			report = await checkCacheInvariants(fx.orchestrator);
+			// After clearing raw, augmented entries become orphans relative to
+			// raw. Our I1 invariant says "every raw has augmented" — that's
+			// still trivially true (zero raws). The reverse isn't asserted by
+			// I1, so this is fine.
+			expect(nonTransientViolations(report)).toEqual([]);
+			expect(report.stats.rawCount).toBe(0);
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("invariants hold immediately after dispose has been called", async () => {
+		// Belt and braces: we shouldn't be able to even call the checker after
+		// dispose without it crashing. (It might return errors, but no throws.)
+		const fx = await buildOrchestrator({ files: { "a.md": "- [ ] x\n" } });
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		const file = fx.vault.getFileByPath("a.md")!;
+		await (fx.orchestrator as any).processFileImmediate(file, true);
+
+		await fx.dispose();
+
+		// After dispose, storage may still be reachable but the orchestrator's
+		// other dependencies (e.g. workspaceManager event refs) are gone.
+		// The checker should still complete without throwing.
+		const report = await checkCacheInvariants(fx.orchestrator);
+		// We don't assert ok=true here; just that the call returns a structured
+		// report rather than throwing.
+		expect(report).toBeDefined();
+		expect(Array.isArray(report.violations)).toBe(true);
+	});
+});

src/__tests__/integration/Orchestrator.roundtrip.test.ts

@@ -0,0 +1,141 @@
+/**
+ * Phase 0 W5.1 — Orchestrator end-to-end roundtrip.
+ *
+ * The vital-signs test for the dataflow pipeline. Exercises:
+ *   parse → augment → store raw → store augmented → store project → query → modify
+ *   → re-parse → re-query → delete → re-query
+ *
+ * Workers are forced off so parsing routes through main-thread fallback
+ * (ConfigurableTaskParser) — the worker mock doesn't return TaskParseResult
+ * shapes. The test isn't validating worker behavior, it's validating that
+ * a task added to a file ends up in the index and a task removed from a
+ * file disappears from the index.
+ *
+ * Scope notes: this test directly invokes the orchestrator's private
+ * `processFileImmediate` to bypass the 300ms debounce in `processFile`.
+ * That's the canonical pattern for fast roundtrip tests; the debounce is
+ * a UI affordance, not a correctness boundary.
+ */
+
+import { buildOrchestrator } from "./_fixtures/buildOrchestrator";
+
+describe("Orchestrator roundtrip (W5.1)", () => {
+	it("parses, indexes, and queries tasks from a markdown file", async () => {
+		const fx = await buildOrchestrator({
+			files: {
+				"notes/a.md": "- [ ] task one\n- [x] task two\n",
+			},
+		});
+
+		// Force main-thread parsing — we don't want the worker mock involved.
+		const workerOrchestrator: any = (fx.orchestrator as any)
+			.workerOrchestrator;
+		workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			const file = fx.vault.getFileByPath("notes/a.md");
+			expect(file).not.toBeNull();
+
+			// Process the file directly (bypass debounce)
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			// Query through the public QueryAPI
+			const queryAPI = fx.orchestrator.getQueryAPI();
+			const tasks = await queryAPI.getAllTasks();
+
+			// We expect at least the two tasks we wrote. The exact shape comes
+			// from ConfigurableTaskParser; we only assert the count and contents.
+			const inFile = tasks.filter((t: any) => t.filePath === "notes/a.md");
+			expect(inFile.length).toBe(2);
+			const contents = inFile.map((t: any) => t.content).sort();
+			expect(contents).toEqual(["task one", "task two"].sort());
+			// One completed, one not
+			const completedFlags = inFile.map((t: any) => t.completed).sort();
+			expect(completedFlags).toEqual([false, true]);
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("reflects file modifications in the next query", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one\n" },
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			const file = fx.vault.getFileByPath("a.md")!;
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			let tasks = await fx.orchestrator.getQueryAPI().getAllTasks();
+			expect(tasks.filter((t: any) => t.filePath === "a.md").length).toBe(
+				1,
+			);
+
+			// Modify the file: add a second task
+			await fx.vault.modify(file, "- [ ] one\n- [ ] two\n");
+			// Re-process directly (forceInvalidate=true to bypass mtime cache)
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			tasks = await fx.orchestrator.getQueryAPI().getAllTasks();
+			const inFile = tasks.filter((t: any) => t.filePath === "a.md");
+			expect(inFile.length).toBe(2);
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("removes tasks from the index when a file is deleted", async () => {
+		const fx = await buildOrchestrator({
+			files: { "doomed.md": "- [ ] gone\n" },
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			const file = fx.vault.getFileByPath("doomed.md")!;
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			let tasks = await fx.orchestrator.getQueryAPI().getAllTasks();
+			expect(
+				tasks.filter((t: any) => t.filePath === "doomed.md").length,
+			).toBe(1);
+
+			// Delete via repository.removeFile (the public API the FILE_UPDATED
+			// "delete" event handler uses internally).
+			const repo: any = (fx.orchestrator as any).repository;
+			await repo.removeFile("doomed.md");
+
+			tasks = await fx.orchestrator.getQueryAPI().getAllTasks();
+			expect(
+				tasks.filter((t: any) => t.filePath === "doomed.md").length,
+			).toBe(0);
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("dispose releases all listeners after a roundtrip", async () => {
+		const fx = await buildOrchestrator({
+			files: {
+				"a.md": "- [ ] one\n",
+				"b.md": "- [x] two\n",
+				"c.md": "no tasks here\n",
+			},
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		const fileA = fx.vault.getFileByPath("a.md")!;
+		const fileB = fx.vault.getFileByPath("b.md")!;
+		await (fx.orchestrator as any).processFileImmediate(fileA, true);
+		await (fx.orchestrator as any).processFileImmediate(fileB, true);
+
+		const beforeDispose = fx.app.__totalListenerCount();
+		expect(beforeDispose).toBeGreaterThan(0);
+
+		await fx.dispose();
+
+		// All event-bus listeners that the orchestrator/sources/indexer
+		// attached should be gone.
+		expect(fx.app.__totalListenerCount()).toBe(0);
+	});
+});

src/__tests__/integration/Orchestrator.settingsChange.test.ts

@@ -0,0 +1,120 @@
+/**
+ * Phase 0 W5.2 — settings change → cache invalidation → re-query.
+ *
+ * Verifies the contract that subsequent settings consolidation work in
+ * Phase 1 will rely on: when a settings change is announced via
+ * onSettingsChange (or its typed sibling onSettingsFieldsChanged), the
+ * affected cache namespaces are cleared and a subsequent query reflects
+ * the new state.
+ *
+ * This test does NOT exercise rebuild() — that's a separate (slower)
+ * concern. It just verifies the cache layer reacts to settings changes
+ * the way Phase 1's deprecation work needs.
+ */
+
+import { buildOrchestrator } from "./_fixtures/buildOrchestrator";
+
+describe("Orchestrator settings change → cache invalidation (W5.2)", () => {
+	it("parser scope clears the raw namespace", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] task one\n" },
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			// Block rebuild to keep this test fast.
+			(fx.orchestrator as any).rebuild = jest.fn(async () => {});
+
+			// Process a file so the raw namespace has content.
+			const file = fx.vault.getFileByPath("a.md")!;
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			const storage: any = (fx.orchestrator as any).storage;
+			const beforeStats = await storage.getStats();
+			expect(beforeStats.byNamespace.raw).toBeGreaterThan(0);
+
+			// Trigger a parser-scope settings change.
+			await fx.orchestrator.onSettingsChange(["parser"]);
+
+			const afterStats = await storage.getStats();
+			expect(afterStats.byNamespace.raw).toBe(0);
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("augment scope clears augmented + project namespaces", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one\n" },
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			(fx.orchestrator as any).rebuild = jest.fn(async () => {});
+
+			const file = fx.vault.getFileByPath("a.md")!;
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			const storage: any = (fx.orchestrator as any).storage;
+			const before = await storage.getStats();
+			expect(before.byNamespace.augmented).toBeGreaterThan(0);
+
+			await fx.orchestrator.onSettingsChange(["augment"]);
+
+			const after = await storage.getStats();
+			expect(after.byNamespace.augmented).toBe(0);
+			expect(after.byNamespace.project).toBe(0);
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("emits SETTINGS_CHANGED event with the scopes payload", async () => {
+		const fx = await buildOrchestrator();
+		try {
+			(fx.orchestrator as any).rebuild = jest.fn(async () => {});
+
+			const events: any[] = [];
+			fx.app.workspace.on(
+				"task-genius:settings-changed",
+				(payload: any) => {
+					events.push(payload);
+				},
+			);
+
+			await fx.orchestrator.onSettingsChange(["parser", "augment"]);
+
+			expect(events.length).toBe(1);
+			expect(events[0].scopes).toEqual(["parser", "augment"]);
+			expect(typeof events[0].timestamp).toBe("number");
+		} finally {
+			await fx.dispose();
+		}
+	});
+
+	it("typed onSettingsFieldsChanged path also clears caches via the legacy delegate", async () => {
+		const fx = await buildOrchestrator({
+			files: { "a.md": "- [ ] one\n" },
+		});
+		(fx.orchestrator as any).workerOrchestrator.setWorkersEnabled(false);
+
+		try {
+			(fx.orchestrator as any).rebuild = jest.fn(async () => {});
+
+			const file = fx.vault.getFileByPath("a.md")!;
+			await (fx.orchestrator as any).processFileImmediate(file, true);
+
+			const storage: any = (fx.orchestrator as any).storage;
+			const before = await storage.getStats();
+			expect(before.byNamespace.raw).toBeGreaterThan(0);
+
+			// taskStatuses is a parser-scope field per scope-map.ts.
+			await fx.orchestrator.onSettingsFieldsChanged(["taskStatuses"]);
+
+			const after = await storage.getStats();
+			expect(after.byNamespace.raw).toBe(0);
+		} finally {
+			await fx.dispose();
+		}
+	});
+});

src/__tests__/integration/_fixtures/buildOrchestrator.ts

@@ -138,6 +138,25 @@ export class FakeVault {
 	private bus = new EventBus();
 	configDir = ".obsidian";
 
+	/**
+	 * Vault adapter — minimal surface for code that does
+	 * `vault.adapter.stat(path)`. Used by Orchestrator.processFileImmediate
+	 * for mtime checks.
+	 */
+	adapter = {
+		stat: async (path: string) => {
+			const f = this.fileMap.get(path);
+			if (!f) return null;
+			return {
+				mtime: f.stat.mtime,
+				ctime: f.stat.ctime,
+				size: f.stat.size,
+				type: "file" as const,
+			};
+		},
+		exists: async (path: string) => this.fileMap.has(path),
+	};
+
 	constructor(initialFiles?: Record<string, string> | VaultFile[]) {
 		if (Array.isArray(initialFiles)) {
 			for (const f of initialFiles) this.__seed(f.path, f.content, f.mtime);