fix(agent-core-v2): reproject a stale session index at startup

packages/agent-core-v2/src/app/sessionIndex/sessionIndexProjector.ts

@@ -1,4 +1,5 @@
 import { ILogService } from '#/_base/log/log';
+import { SESSION_INDEX_KEY, SESSION_INDEX_SCOPE } from '#/app/workspace/workspaceAlias';
 import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
 import { IQueryStore, type WriteOp } from '#/persistence/interface/queryStore';
 import { IFileSystemStorageService } from '#/persistence/interface/storage';
@@ -18,6 +19,7 @@ import {
   listWorkspaceIds,
   mapBounded,
   readSessionSummary,
+  sessionStateMaxMtime,
   summaryEquals,
 } from './sessionIndexSource';
 
@@ -47,6 +49,7 @@ export interface ReconcileResult {
 export interface AuthoritativeScan {
   readonly summaries: SessionSummary[];
   readonly counts: Map<string, { active: number; archived: number }>;
+  readonly sourceMaxMtimeMs: number;
 }
 
 interface ScanSlot {
@@ -112,7 +115,7 @@ export class SessionIndexProjector {
       field: `custom.${PARENT_SESSION_ID_KEY}`,
     });
 
-    const { summaries, counts } = await scan;
+    const { summaries, counts, sourceMaxMtimeMs } = await scan;
     await this.batchChunks(
       summaries.map((summary) => ({
         kind: 'put' as const,
@@ -123,7 +126,10 @@ export class SessionIndexProjector {
       })),
     );
     await this.writeCounters(counters, counts);
-    await queryStore.setCheckpoint(SESSION_INDEX_MANIFEST, { seq: generation });
+    await queryStore.setCheckpoint(SESSION_INDEX_MANIFEST, {
+      seq: generation,
+      sourceMaxMtimeMs,
+    });
     log.info('session index generation published', {
       generation,
       sessions: summaries.length,
@@ -149,7 +155,7 @@ export class SessionIndexProjector {
     const { queryStore, log } = this.deps;
     const collection = sessionCollection(generation);
     const counters = sessionCountersCollection(generation);
-    const { summaries, counts } = await this.scanAuthoritative();
+    const { summaries, counts, sourceMaxMtimeMs } = await this.scanAuthoritative();
     const authoritativeIds = new Set(summaries.map((s) => s.id));
 
     const storedKeys = await queryStore.listKeys(collection);
@@ -177,6 +183,13 @@ export class SessionIndexProjector {
 
     await this.batchChunks([...upserts, ...removals]);
     await this.writeCounters(counters, counts);
+    const manifest = await queryStore.getCheckpoint(SESSION_INDEX_MANIFEST);
+    if (manifest?.seq === generation) {
+      await queryStore.setCheckpoint(SESSION_INDEX_MANIFEST, {
+        seq: generation,
+        sourceMaxMtimeMs: Math.max(manifest.sourceMaxMtimeMs ?? 0, sourceMaxMtimeMs),
+      });
+    }
     const result = { sessions: summaries.length, upserted: upserts.length, removed: removals.length };
     if (result.upserted > 0 || result.removed > 0) {
       log.info('session index reconciliation repaired drift', { generation, ...result });
@@ -188,11 +201,14 @@ export class SessionIndexProjector {
     const { storage, docs, sessionsScope } = this.deps;
     const summaries: SessionSummary[] = [];
     const counts = new Map<string, { active: number; archived: number }>();
+    let sourceMaxMtimeMs = (await storage.mtime(SESSION_INDEX_SCOPE, SESSION_INDEX_KEY)) ?? 0;
     for (const workspaceId of await listWorkspaceIds(storage, sessionsScope)) {
       const sessionIds = await listSessionIds(storage, sessionsScope, workspaceId);
-      const found = await mapBounded(sessionIds, SCAN_CONCURRENCY, (sessionId) =>
-        readSessionSummary(docs, sessionsScope, workspaceId, sessionId),
-      );
+      const found = await mapBounded(sessionIds, SCAN_CONCURRENCY, async (sessionId) => {
+        const mtime = await sessionStateMaxMtime(storage, sessionsScope, workspaceId, sessionId);
+        if (mtime > sourceMaxMtimeMs) sourceMaxMtimeMs = mtime;
+        return readSessionSummary(docs, sessionsScope, workspaceId, sessionId);
+      });
       const entry = counts.get(workspaceId) ?? { active: 0, archived: 0 };
       for (const summary of found) {
         summaries.push(summary);
@@ -201,7 +217,7 @@ export class SessionIndexProjector {
       }
       counts.set(workspaceId, entry);
     }
-    return { summaries, counts };
+    return { summaries, counts, sourceMaxMtimeMs };
   }
 
   private async writeCounters(

packages/agent-core-v2/src/app/sessionIndex/sessionIndexService.ts

@@ -44,6 +44,7 @@ import {
   listSessionIds,
   listWorkspaceIds,
   readSessionSummary,
+  scanSessionsMaxMtime,
   summaryMatchesChildOf,
 } from './sessionIndexSource';
 
@@ -131,7 +132,7 @@ export class FileSessionIndex extends Disposable implements ISessionIndex {
     this.state = 'preparing';
     try {
       const manifest = await this.queryStore.getCheckpoint(SESSION_INDEX_MANIFEST);
-      if (manifest === undefined) {
+      if (manifest === undefined || !(await this.manifestFresh(manifest))) {
         const projection = this.ensureProjection();
         if (deadlineMs === undefined) {
           await projection;
@@ -158,6 +159,19 @@ export class FileSessionIndex extends Disposable implements ISessionIndex {
     return this.status();
   }
 
+  private async manifestFresh(manifest: Checkpoint): Promise<boolean> {
+    const published = manifest.sourceMaxMtimeMs;
+    if (published === undefined) return false;
+    try {
+      return (await scanSessionsMaxMtime(this.storage, this.sessionsScope)) <= published;
+    } catch (error) {
+      this.log.warn('session index freshness check failed; re-projecting', {
+        error: String(error),
+      });
+      return false;
+    }
+  }
+
   private ensureProjection(): Promise<void> {
     this.projectFlight ??= this.runProjection().finally(() => {
       this.projectFlight = undefined;

packages/agent-core-v2/src/app/sessionIndex/sessionIndexSource.ts

@@ -1,10 +1,12 @@
+import { SESSION_INDEX_KEY, SESSION_INDEX_SCOPE } from '#/app/workspace/workspaceAlias';
 import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
 import { IFileSystemStorageService } from '#/persistence/interface/storage';
 
 import { CHILD_SESSION_KIND, CHILD_SESSION_KIND_KEY, type SessionSummary } from './sessionIndex';
 
 const META_SCOPE = 'session-meta';
 const META_KEY = 'state.json';
+const MTIME_SCAN_CONCURRENCY = 16;
 
 export function parseTime(value: unknown): number {
   if (typeof value === 'number' && Number.isFinite(value)) return value;
@@ -168,3 +170,32 @@ export async function mapBounded<T, R>(
   await Promise.all(workers);
   return out;
 }
+
+export async function sessionStateMaxMtime(
+  storage: IFileSystemStorageService,
+  sessionsScope: string,
+  workspaceId: string,
+  sessionId: string,
+): Promise<number> {
+  const base = `${sessionsScope}/${workspaceId}/${sessionId}`;
+  const direct = await storage.mtime(base, META_KEY);
+  const nested = await storage.mtime(`${base}/${META_SCOPE}`, META_KEY);
+  return Math.max(direct ?? 0, nested ?? 0);
+}
+
+export async function scanSessionsMaxMtime(
+  storage: IFileSystemStorageService,
+  sessionsScope: string,
+): Promise<number> {
+  let max = (await storage.mtime(SESSION_INDEX_SCOPE, SESSION_INDEX_KEY)) ?? 0;
+  for (const workspaceId of await listWorkspaceIds(storage, sessionsScope)) {
+    const sessionIds = await listSessionIds(storage, sessionsScope, workspaceId);
+    const mtimes = await mapBounded(sessionIds, MTIME_SCAN_CONCURRENCY, (sessionId) =>
+      sessionStateMaxMtime(storage, sessionsScope, workspaceId, sessionId),
+    );
+    for (const mtime of mtimes) {
+      if (mtime > max) max = mtime;
+    }
+  }
+  return max;
+}

packages/agent-core-v2/src/persistence/backends/memory/inMemoryStorageService.ts

@@ -23,6 +23,7 @@ export class InMemoryStorageService implements IFileSystemStorageService {
 
   private readonly scopes = new Map<string, Map<string, Uint8Array>>();
   private readonly watchers = new Map<string, WatchEntry>();
+  private readonly mtimes = new Map<string, number>();
 
   async read(scope: string, key: string): Promise<Uint8Array | undefined> {
     return this.scopes.get(scope)?.get(key);
@@ -52,6 +53,7 @@ export class InMemoryStorageService implements IFileSystemStorageService {
   ): Promise<void> {
     options.signal?.throwIfAborted();
     this.bucket(scope).set(key, data);
+    this.mtimes.set(this.watchKey(scope, key), Date.now());
     this.notifyWatchers(scope, key);
   }
 
@@ -76,6 +78,7 @@ export class InMemoryStorageService implements IFileSystemStorageService {
       offset += chunk.byteLength;
     }
     this.bucket(scope).set(key, merged);
+    this.mtimes.set(this.watchKey(scope, key), Date.now());
     this.notifyWatchers(scope, key);
   }
 
@@ -89,13 +92,15 @@ export class InMemoryStorageService implements IFileSystemStorageService {
     const existing = bucket.get(key);
     if (existing === undefined) {
       bucket.set(key, data);
+      this.mtimes.set(this.watchKey(scope, key), Date.now());
       this.notifyWatchers(scope, key);
       return;
     }
     const merged = new Uint8Array(existing.byteLength + data.byteLength);
     merged.set(existing, 0);
     merged.set(data, existing.byteLength);
     bucket.set(key, merged);
+    this.mtimes.set(this.watchKey(scope, key), Date.now());
     this.notifyWatchers(scope, key);
   }
 
@@ -108,13 +113,18 @@ export class InMemoryStorageService implements IFileSystemStorageService {
 
   async delete(scope: string, key: string): Promise<void> {
     this.scopes.get(scope)?.delete(key);
+    this.mtimes.delete(this.watchKey(scope, key));
     this.notifyWatchers(scope, key);
   }
 
   async size(scope: string, key: string): Promise<number | undefined> {
     return this.scopes.get(scope)?.get(key)?.byteLength;
   }
 
+  async mtime(scope: string, key: string): Promise<number | undefined> {
+    return this.mtimes.get(this.watchKey(scope, key));
+  }
+
   pathFor(_scope: string, _key: string): undefined {
     return undefined;
   }

packages/agent-core-v2/src/persistence/backends/node-fs/fileStorageService.ts

@@ -170,6 +170,16 @@ export class FileStorageService implements IFileSystemStorageService {
     }
   }
 
+  async mtime(scope: string, key: string): Promise<number | undefined> {
+    const filePath = this.pathFor(scope, key);
+    try {
+      return (await stat(filePath)).mtimeMs;
+    } catch (error) {
+      if (isEnoent(error)) return undefined;
+      throw toStorageIoError(error, { path: filePath, op: 'stat' });
+    }
+  }
+
   watch(scope: string, key: string): Event<void> {
     const target = this.pathFor(scope, key);
     const dir = dirname(target);

packages/agent-core-v2/src/persistence/interface/queryStore.ts

@@ -66,6 +66,7 @@ export type WriteOp =
 
 export interface Checkpoint {
   readonly seq: number;
+  readonly sourceMaxMtimeMs?: number;
 }
 
 export interface ColumnBounds {

packages/agent-core-v2/src/persistence/interface/storage.ts

@@ -146,6 +146,7 @@ export interface IFileSystemStorageService {
   list(scope: string, prefix?: string): Promise<readonly string[]>;
   delete(scope: string, key: string): Promise<void>;
   size(scope: string, key: string): Promise<number | undefined>;
+  mtime(scope: string, key: string): Promise<number | undefined>;
   pathFor(scope: string, key: string): string | undefined;
   watch?(scope: string, key: string): Event<void>;
   flush(): Promise<void>;

packages/agent-core-v2/test/app/sessionIndex/sessionIndex.test.ts

@@ -22,7 +22,11 @@ import {
   ISessionIndexMirror,
   type SessionSummary,
 } from '#/app/sessionIndex/sessionIndex';
-import { recencyColumn, sessionCollection } from '#/app/sessionIndex/sessionIndexModel';
+import {
+  SESSION_INDEX_MANIFEST,
+  recencyColumn,
+  sessionCollection,
+} from '#/app/sessionIndex/sessionIndexModel';
 import { FileSessionIndex } from '#/app/sessionIndex/sessionIndexService';
 import {
   drainSessionIndexMirror,
@@ -1272,6 +1276,8 @@ describe('FileSessionIndex (read model)', () => {
     const first = build();
     await first.prepare();
     expect(first.status()).toEqual({ state: 'ready', generation: 1, degradedCount: 0 });
+    const published = await queryStore.getCheckpoint(SESSION_INDEX_MANIFEST);
+    expect(published).toMatchObject({ seq: 1, sourceMaxMtimeMs: expect.any(Number) });
     disposeHost?.();
     disposeHost = undefined;
     await drainSessionIndexMirror();
@@ -1309,6 +1315,74 @@ describe('FileSessionIndex (read model)', () => {
     expect(docs.gets).toBe(0);
   });
 
+  it('re-projects on the next startup when the session directories changed externally', async () => {
+    await seedSession('a', { title: 'a', createdAt: 1, updatedAt: 2 });
+    await seedSession('b', { title: 'b', createdAt: 2, updatedAt: 3 });
+
+    const first = build();
+    await first.prepare();
+    expect(first.status()).toEqual({ state: 'ready', generation: 1, degradedCount: 0 });
+    disposeHost?.();
+    disposeHost = undefined;
+    await drainSessionIndexMirror();
+    await drainQueryStoreDisposals();
+
+    await seedSession('c', { title: 'c', createdAt: 3, updatedAt: 4 });
+    const future = new Date(Date.now() + 60_000);
+    await fsp.utimes(
+      join(sessionsDir, workspaceId, 'c', 'session-meta', 'state.json'),
+      future,
+      future,
+    );
+
+    const second = build();
+    const status = await second.prepare();
+    expect(status).toEqual({ state: 'ready', generation: 2, degradedCount: 0 });
+    const page = await second.listRecent({ workspaceIds: [workspaceId] });
+    expect(page.items.map((s) => s.id)).toEqual(['c', 'b', 'a']);
+  });
+
+  it('treats a published checkpoint without sourceMaxMtimeMs as stale and re-projects', async () => {
+    await seedSession('a', { title: 'a', createdAt: 1, updatedAt: 2 });
+
+    const first = build();
+    await first.prepare();
+    await queryStore.setCheckpoint(SESSION_INDEX_MANIFEST, { seq: 1 });
+    disposeHost?.();
+    disposeHost = undefined;
+    await drainSessionIndexMirror();
+    await drainQueryStoreDisposals();
+
+    const second = build();
+    const status = await second.prepare();
+    expect(status).toEqual({ state: 'ready', generation: 2, degradedCount: 0 });
+    const page = await second.listRecent({ workspaceIds: [workspaceId] });
+    expect(page.items.map((s) => s.id)).toEqual(['a']);
+  });
+
+  it('reconciliation refreshes the published source max mtime', async () => {
+    await seedSession('a', { title: 'a', createdAt: 1, updatedAt: 2 });
+
+    const store = build();
+    await store.prepare();
+    const published = await queryStore.getCheckpoint(SESSION_INDEX_MANIFEST);
+    expect(published).toMatchObject({ seq: 1, sourceMaxMtimeMs: expect.any(Number) });
+
+    await seedSession('a', { title: 'a2', createdAt: 1, updatedAt: 5 });
+    const future = new Date((published?.sourceMaxMtimeMs ?? 0) + 60_000);
+    await fsp.utimes(
+      join(sessionsDir, workspaceId, 'a', 'session-meta', 'state.json'),
+      future,
+      future,
+    );
+
+    await store.reconcileNow();
+    const refreshed = await queryStore.getCheckpoint(SESSION_INDEX_MANIFEST);
+    expect(refreshed?.seq).toBe(1);
+    expect(refreshed?.sourceMaxMtimeMs).toBeGreaterThan(published?.sourceMaxMtimeMs ?? 0);
+    expect((await store.get('a'))?.title).toBe('a2');
+  });
+
   it('the resume-startup sequence pays one scan: point lookup, projection, then warm lists', async () => {
     await seedSession('a', { title: 'a', createdAt: 1, updatedAt: 3 });
     await seedSession('b', { title: 'b', createdAt: 2, updatedAt: 2 });

packages/agent-core-v2/test/persistence/backends/memory/inMemoryStorageService.test.ts

@@ -0,0 +1,32 @@
+import { describe, expect, it } from 'vitest';
+
+import { InMemoryStorageService } from '#/persistence/backends/memory/inMemoryStorageService';
+
+const encoder = new TextEncoder();
+
+describe('InMemoryStorageService — mtime', () => {
+  it('returns undefined for a missing key and the write time for an existing one', async () => {
+    const svc = new InMemoryStorageService();
+    expect(await svc.mtime('scope', 'missing.json')).toBeUndefined();
+
+    const before = Date.now();
+    await svc.write('scope', 'k.json', encoder.encode('{}'));
+    const mtime = await svc.mtime('scope', 'k.json');
+    expect(mtime).toBeGreaterThanOrEqual(before);
+  });
+
+  it('tracks writes, appends and deletions', async () => {
+    const svc = new InMemoryStorageService();
+
+    await svc.write('scope', 'k.json', encoder.encode('a'));
+    const written = await svc.mtime('scope', 'k.json');
+    expect(written).toBeDefined();
+
+    await svc.append('scope', 'k.json', encoder.encode('b'));
+    const appended = await svc.mtime('scope', 'k.json');
+    expect(appended).toBeGreaterThanOrEqual(written ?? 0);
+
+    await svc.delete('scope', 'k.json');
+    expect(await svc.mtime('scope', 'k.json')).toBeUndefined();
+  });
+});