fix: bound print-mode shutdown and truncated-fork cutoffs

Abort print-mode agent quiescence when cleanup times out, keep
dispatcher flush and app dispose running if telemetry shutdown
fails, and compute truncated-fork cutoff from the retained turn
window before dropping file-history records.

Author: elkaix

.changeset/print-flush-and-fork-cutoff.md

@@ -0,0 +1,5 @@
+---
+"@pymodel/pythinker-code": patch
+---
+
+Keep print-mode shutdown from skipping journal flush, and keep truncated forks from dropping later work in the retained turn.

apps/pythinker-code/src/cli/v2/run-v2-print.ts

@@ -201,7 +201,7 @@ export async function runV2Print(
   }
 
   let restorePermission = async (): Promise<void> => {};
-  let quiesceAgents = async (): Promise<void> => {};
+  let quiesceAgents = async (_signal?: AbortSignal): Promise<void> => {};
   let releaseQuiescence: (() => void) | undefined;
   let flushWires = async (): Promise<void> => {};
   let removeTerminationCleanup: (() => void) | undefined;
@@ -211,15 +211,21 @@ export async function runV2Print(
     const pending = (cleanupPromise ??= (async () => {
       removeTerminationCleanup?.();
       setCrashPhase('shutdown');
+      const quiesceAbort = new AbortController();
       try {
         await restorePermission();
-        await raceWithTimeout(quiesceAgents(), CLI_SHUTDOWN_TIMEOUT_MS).catch(() => {});
+        await raceWithTimeout(quiesceAgents(quiesceAbort.signal), CLI_SHUTDOWN_TIMEOUT_MS).catch(
+          () => {},
+        );
       } finally {
+        quiesceAbort.abort();
         try {
           await Promise.all([
             raceWithTimeout(flushWires(), CLI_SHUTDOWN_TIMEOUT_MS).catch(() => {}),
             telemetryService !== undefined
-              ? raceWithTimeout(telemetryService.shutdown(), CLI_SHUTDOWN_TIMEOUT_MS)
+              ? raceWithTimeout(telemetryService.shutdown(), CLI_SHUTDOWN_TIMEOUT_MS).catch(
+                  () => {},
+                )
               : Promise.resolve(),
             shutdownTelemetry({ timeoutMs: CLI_SHUTDOWN_TIMEOUT_MS }).catch(() => {}),
           ]);
@@ -274,8 +280,8 @@ export async function runV2Print(
 
     const resolved = await resolveNativeSession(app, opts, workDir, defaultModel, stderr);
     restorePermission = resolved.restorePermission;
-    quiesceAgents = async () => {
-      releaseQuiescence = await quiesceSessionAgents(resolved.session, resolved.agent);
+    quiesceAgents = async (signal) => {
+      releaseQuiescence = await quiesceSessionAgents(resolved.session, resolved.agent, signal);
     };
     flushWires = () => flushSessionWires(resolved.session, resolved.agent);
 
@@ -967,6 +973,7 @@ function collectSessionAgentHandles(
 async function quiesceSessionAgents(
   session: ISessionScopeHandle,
   mainAgent: IAgentScopeHandle,
+  signal?: AbortSignal,
 ): Promise<(() => void) | undefined> {
   const handles = collectSessionAgentHandles(session, mainAgent);
   const promptServices = handles.flatMap((handle) => {
@@ -993,6 +1000,7 @@ async function quiesceSessionAgents(
     }),
   );
   for (;;) {
+    if (signal?.aborted) return undefined;
     await Promise.allSettled(promptServices.map((service) => service.drain()));
     for (const loop of loops) {
       for (const turnId of loop.status().pendingTurnIds) loop.cancel(turnId);
@@ -1022,6 +1030,10 @@ async function quiesceSessionAgents(
         return false;
       }
     });
+    if (signal?.aborted) {
+      for (const guard of guards) guard.dispose();
+      return undefined;
+    }
     if (frozen && !busy) {
       return () => {
         for (const guard of guards) guard.dispose();
@@ -1039,9 +1051,13 @@ async function flushSessionWires(
   mainAgent: IAgentScopeHandle,
 ): Promise<void> {
   await Promise.allSettled(
-    collectSessionAgentHandles(session, mainAgent).map((handle) =>
-      handle.accessor.get(IEventDispatcher).flush(),
-    ),
+    collectSessionAgentHandles(session, mainAgent).flatMap((handle) => {
+      try {
+        return [handle.accessor.get(IEventDispatcher).flush()];
+      } catch {
+        return [];
+      }
+    }),
   );
 }
 

packages/agent-core-v2/src/app/sessionExport/sessionExportService.ts

@@ -1,4 +1,4 @@
-import { join, resolve } from 'pathe';
+import { join, relative, resolve } from 'pathe';
 import { LifecycleScope } from '#/app/scopes';
 import { ScopeActivation, registerScopedService } from '#/_base/di/scope';
 import type { ISessionScopeHandle } from '#/_base/di/scope';
@@ -205,8 +205,8 @@ export async function exportSessionDirectory(input: {
     const stableSessionLog = sessionLogSource;
     const selectedSessionFiles: SessionZipEntry[] = sessionFiles.filter((file) => {
       if (file === sessionLogPath) return false;
-      const parts = file.split(/[\\/]/);
-      return !parts.includes(FILE_HISTORY_BLOB_PREFIX) && !parts.includes('notify');
+      const parts = relative(sessionDir, file).split(/[\\/]/);
+      return parts[0] !== FILE_HISTORY_BLOB_PREFIX && parts[0] !== 'notify';
     });
     if (stableSessionLog !== undefined) {
       selectedSessionFiles.push({ path: sessionLogPath, source: stableSessionLog });

packages/agent-core-v2/src/features/fileHistory/fileHistoryRetention.ts

@@ -108,8 +108,9 @@ export async function touchForkedFileHistory(input: FileHistoryRetentionInput):
   try {
     const entries = await input.hostFs.readdir(agentsDir);
     agentNames = entries.filter((entry) => entry.isDirectory).map((entry) => entry.name);
-  } catch {
-    return;
+  } catch (error) {
+    if (isMissingPathError(error)) return;
+    throw error;
   }
   for (const name of agentNames) {
     try {
@@ -118,8 +119,9 @@ export async function touchForkedFileHistory(input: FileHistoryRetentionInput):
         await touchFileHistorySession(input);
         return;
       }
-    } catch {
-      continue;
+    } catch (error) {
+      if (isMissingPathError(error)) continue;
+      throw error;
     }
   }
 }

packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts

@@ -480,12 +480,29 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle
         }
         const deadline = Date.now() + REMOVE_PROMPT_QUIESCE_TIMEOUT_MS;
         for (;;) {
-          await prompt.drain(reason);
-          for (const turnId of loop.status().pendingTurnIds) {
-            loop.cancel(turnId, reason);
-          }
-          loop.cancel(undefined, reason);
-          await Promise.all([loop.settled(), compactionSettled]);
+          const remaining = deadline - Date.now();
+          if (remaining <= 0) break;
+          let drainTimedOut = false;
+          let drainTimer: ReturnType<typeof setTimeout> | undefined;
+          await Promise.race([
+            (async () => {
+              await prompt.drain(reason);
+              for (const turnId of loop.status().pendingTurnIds) {
+                loop.cancel(turnId, reason);
+              }
+              loop.cancel(undefined, reason);
+              await Promise.all([loop.settled(), compactionSettled]);
+            })().finally(() => {
+              if (drainTimer !== undefined) clearTimeout(drainTimer);
+            }),
+            new Promise<void>((resolve) => {
+              drainTimer = setTimeout(() => {
+                drainTimedOut = true;
+                resolve();
+              }, remaining);
+            }),
+          ]);
+          if (drainTimedOut) break;
           let idle = true;
           try {
             const snapshot = prompt.list();

packages/agent-core-v2/src/workspace/sessionLifecycle/internal/forkTurnSlice.ts

@@ -43,20 +43,19 @@ export function sliceMainRecordsAtTurn(
 
   const end = turnStarts[turnIndex + 1] ?? records.length;
   const retainedTurnInputs = turnInputIndicesThrough(records, turnIndex);
-  const retained = records
+  const window = records
     .slice(0, end)
     .filter(
       (record, index) =>
-        !record.type.startsWith(FILE_HISTORY_RECORD_PREFIX) &&
-        (!isUserVisibleTurnInputRecord(record) || retainedTurnInputs.has(index)),
+        !isUserVisibleTurnInputRecord(record) || retainedTurnInputs.has(index),
     );
-  const cutoffTimes = retained
-    .map(recordTime)
-    .filter((time): time is number => time !== undefined);
+  const retained = window.filter(
+    (record) => !record.type.startsWith(FILE_HISTORY_RECORD_PREFIX),
+  );
   const lastPrompt = promptMetadataFromTurnRecord(records[start]!);
   return {
     records: retained,
-    cutoffTime: cutoffTimes.length === 0 ? undefined : Math.max(...cutoffTimes),
+    cutoffTime: cutoffFrom(window),
     lastPrompt,
   };
 }
@@ -85,19 +84,15 @@ export function sliceMainRecordsBeforePrompt(
     return sliceMainRecordsAtTurn(records, sourceSessionId, activeTurnIndex - 1);
   }
   const activeStart = turnStarts[0]!;
-  const retained = records
+  const window = records
     .slice(0, activeStart)
-    .filter(
-      (record) =>
-        !record.type.startsWith(FILE_HISTORY_RECORD_PREFIX) &&
-        !isUserVisibleTurnInputRecord(record),
-    );
-  const cutoffTimes = retained
-    .map(recordTime)
-    .filter((time): time is number => time !== undefined);
+    .filter((record) => !isUserVisibleTurnInputRecord(record));
+  const retained = window.filter(
+    (record) => !record.type.startsWith(FILE_HISTORY_RECORD_PREFIX),
+  );
   return {
     records: retained,
-    cutoffTime: cutoffTimes.length === 0 ? undefined : Math.max(...cutoffTimes),
+    cutoffTime: cutoffFrom(window),
   };
 }
 
@@ -117,6 +112,13 @@ export function sliceSubagentRecordsAtTime(
   return records.slice(0, end);
 }
 
+function cutoffFrom(records: readonly WireRecord[]): number | undefined {
+  const times = records
+    .map(recordTime)
+    .filter((time): time is number => time !== undefined);
+  return times.length === 0 ? undefined : Math.max(...times);
+}
+
 function isUserVisibleTurnRecord(record: WireRecord): boolean {
   if (record.type !== 'context.append_message') return false;
   const message = asRecord(record['message']);

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

@@ -159,6 +159,20 @@ describe('sessionExport', () => {
     expect(result.entries.some((entry) => entry.includes('notify'))).toBe(false);
   });
 
+  it('does not treat an ancestor directory named notify as session notify state', async () => {
+    const tmp = await mkdtemp(join(tmpdir(), 'notify-ancestor-'));
+    const sessionDir = join(tmp, 'notify', 'sessions', 'ws_demo', 'ses_nested');
+    await mkdir(sessionDir, { recursive: true });
+    await writeFile(join(sessionDir, 'state.json'), '{}\n', 'utf-8');
+
+    const result = await exportSessionDirectory({
+      request: { sessionId: 'ses_nested', outputPath: join(tmp, 'export.zip'), version: '1.0.0-test' },
+      summary: { id: 'ses_nested', sessionDir },
+    });
+
+    expect(result.entries).toEqual(['manifest.json', 'state.json']);
+  });
+
   it('uses a timestamped default output path when outputPath is omitted', async () => {
     const tmp = await mkdtemp(join(tmpdir(), 'session-export-test-'));
     const sessionDir = join(tmp, 'sessions', 'ws_demo', 'ses_default_output');

packages/agent-core-v2/test/workspace/sessionLifecycle/forkTurnSlice.test.ts

@@ -73,6 +73,7 @@ describe('sliceMainRecordsAtTurn', () => {
       'metadata',
       'context.append_message',
     ]);
+    expect(slice.cutoffTime).toBe(4);
   });
 });
 
@@ -129,6 +130,7 @@ describe('sliceMainRecordsBeforePrompt', () => {
 
     const slice = sliceMainRecordsBeforePrompt(records, 'ses_source', 'prompt-1');
     expect(slice.records).toEqual([{ type: 'metadata', protocol_version: '1.5', created_at: 1 }]);
+    expect(slice.cutoffTime).toBe(2);
   });
 
   it('rejects an unknown active prompt', () => {