fix(agent-core-v2): skip stray files during session index scans

Author: elkaix

.changeset/session-index-stray-files.md

@@ -0,0 +1,5 @@
+---
+"@pymodel/pythinker-code": patch
+---
+
+Fix recent sessions missing from the session list when the sessions folder contains stray files.

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

@@ -198,14 +198,14 @@ export class SessionIndexProjector {
   }
 
   private async scanAuthoritative(): Promise<AuthoritativeScan> {
-    const { storage, docs, sessionsScope } = this.deps;
+    const { storage, docs, sessionsScope, log } = 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, async (sessionId) => {
-        const mtime = await sessionStateMaxMtime(storage, sessionsScope, workspaceId, sessionId);
+        const mtime = await sessionStateMaxMtime(storage, sessionsScope, workspaceId, sessionId, log);
         if (mtime > sourceMaxMtimeMs) sourceMaxMtimeMs = mtime;
         return readSessionSummary(docs, sessionsScope, workspaceId, sessionId);
       });

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

@@ -163,7 +163,7 @@ export class FileSessionIndex extends Disposable implements ISessionIndex {
     const published = manifest.sourceMaxMtimeMs;
     if (published === undefined) return false;
     try {
-      return (await scanSessionsMaxMtime(this.storage, this.sessionsScope)) <= published;
+      return (await scanSessionsMaxMtime(this.storage, this.sessionsScope, this.log)) <= published;
     } catch (error) {
       this.log.warn('session index freshness check failed; re-projecting', {
         error: String(error),

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

@@ -1,6 +1,11 @@
+import { ILogService } from '#/_base/log/log';
 import { SESSION_INDEX_KEY, SESSION_INDEX_SCOPE } from '#/app/workspace/workspaceAlias';
 import { IAtomicDocumentStore } from '#/persistence/interface/atomicDocumentStore';
-import { IFileSystemStorageService } from '#/persistence/interface/storage';
+import {
+  IFileSystemStorageService,
+  StorageError,
+  StorageErrors,
+} from '#/persistence/interface/storage';
 
 import { CHILD_SESSION_KIND, CHILD_SESSION_KIND_KEY, type SessionSummary } from './sessionIndex';
 
@@ -171,27 +176,49 @@ export async function mapBounded<T, R>(
   return out;
 }
 
+async function stateFileMtime(
+  storage: IFileSystemStorageService,
+  scope: string,
+  log: ILogService | undefined,
+): Promise<number | undefined> {
+  try {
+    return await storage.mtime(scope, META_KEY);
+  } catch (error) {
+    if (
+      error instanceof StorageError &&
+      error.code === StorageErrors.codes.STORAGE_IO_FAILED &&
+      error.details?.['errno'] === 'ENOTDIR'
+    ) {
+      log?.warn('session index skips a non-directory entry', { path: error.details['path'] });
+      return undefined;
+    }
+    throw error;
+  }
+}
+
 export async function sessionStateMaxMtime(
   storage: IFileSystemStorageService,
   sessionsScope: string,
   workspaceId: string,
   sessionId: string,
+  log?: ILogService,
 ): 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);
+  const direct = await stateFileMtime(storage, base, log);
+  const nested = await stateFileMtime(storage, `${base}/${META_SCOPE}`, log);
   return Math.max(direct ?? 0, nested ?? 0);
 }
 
 export async function scanSessionsMaxMtime(
   storage: IFileSystemStorageService,
   sessionsScope: string,
+  log?: ILogService,
 ): 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),
+      sessionStateMaxMtime(storage, sessionsScope, workspaceId, sessionId, log),
     );
     for (const mtime of mtimes) {
       if (mtime > max) max = mtime;

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

@@ -524,6 +524,22 @@ describe('FileSessionIndex (read model)', () => {
     expect(await store.count({ workspaceIds: [workspaceId], includeArchived: true })).toBe(2);
   });
 
+  it('prepare skips stray files and state-less directories instead of failing the projection', async () => {
+    await seedSession('active', { title: 'hello', createdAt: 1, updatedAt: 2 });
+    await fsp.writeFile(join(sessionsDir, 'workspace.json'), '{}');
+    await fsp.writeFile(join(sessionsDir, workspaceId, 'workspace.json'), '{}');
+    await fsp.writeFile(join(sessionsDir, workspaceId, '.DS_Store'), 'junk');
+    await fsp.mkdir(join(sessionsDir, workspaceId, 'no-state'), { recursive: true });
+
+    const store = build();
+    const status = await store.prepare();
+    expect(status).toEqual({ state: 'ready', generation: 1, degradedCount: 0 });
+
+    const page = await store.listRecent({ workspaceIds: [workspaceId] });
+    expect(page.items.map((s) => s.id)).toEqual(['active']);
+    expect(await store.count({ workspaceIds: [workspaceId] })).toBe(1);
+  });
+
   it('serves warm reads without touching the session directories', async () => {
     await seedSession('a', { title: 'a', createdAt: 1, updatedAt: 2 });
     await seedSession('b', { title: 'b', createdAt: 2, updatedAt: 3 });