From f1ff8cdebefde7e57691069939c57d3cd3e0c20d Mon Sep 17 00:00:00 2001
From: Kit Langton <kit.langton@gmail.com>
Date: Tue, 7 Jul 2026 20:58:22 -0400
Subject: [PATCH] refactor(core): tighten restart recovery scan

---
 packages/core/src/session/execution/local.ts | 42 ++++++++++++--------
 1 file changed, 26 insertions(+), 16 deletions(-)

diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts
index 2ea2991a7f..efd38eb386 100644
--- a/packages/core/src/session/execution/local.ts
+++ b/packages/core/src/session/execution/local.ts
@@ -25,28 +25,38 @@ export function terminal(exit: Exit.Exit<void, SessionRunner.RunError>, reason?:
 
 export const sessionsInterruptedByShutdown = Effect.fn("SessionExecutionLocal.sessionsInterruptedByShutdown")(
   function* (db: Database.Interface["db"]) {
-    const latest = yield* db.all<{ sessionID: string; type: string; reason: string | null }>(sql`
-      SELECT aggregate_id AS sessionID, type, json_extract(data, '$.reason') AS reason
+    const started = EventV2.versionedType(
+      SessionEvent.Execution.Started.type,
+      SessionEvent.Execution.Started.durable.version,
+    )
+    const succeeded = EventV2.versionedType(
+      SessionEvent.Execution.Succeeded.type,
+      SessionEvent.Execution.Succeeded.durable.version,
+    )
+    const failed = EventV2.versionedType(
+      SessionEvent.Execution.Failed.type,
+      SessionEvent.Execution.Failed.durable.version,
+    )
+    const interrupted = EventV2.versionedType(
+      SessionEvent.Execution.Interrupted.type,
+      SessionEvent.Execution.Interrupted.durable.version,
+    )
+    const latest = yield* db.all<{ sessionID: string }>(sql`
+      SELECT aggregate_id AS sessionID
       FROM (
         SELECT aggregate_id, type, data,
           row_number() OVER (PARTITION BY aggregate_id ORDER BY seq DESC) AS rank
         FROM ${EventTable}
         WHERE type IN (
-          ${EventV2.versionedType(SessionEvent.Execution.Started.type, 1)},
-          ${EventV2.versionedType(SessionEvent.Execution.Succeeded.type, 1)},
-          ${EventV2.versionedType(SessionEvent.Execution.Failed.type, 1)},
-          ${EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1)}
+          ${started},
+          ${succeeded},
+          ${failed},
+          ${interrupted}
         )
       )
-      WHERE rank = 1
+      WHERE rank = 1 AND type = ${interrupted} AND json_extract(data, '$.reason') = 'shutdown'
     `)
-    return latest
-      .filter(
-        (event) =>
-          event.type === EventV2.versionedType(SessionEvent.Execution.Interrupted.type, 1) &&
-          event.reason === "shutdown",
-      )
-      .map((event) => SessionSchema.ID.make(event.sessionID))
+    return latest.map((event) => SessionSchema.ID.make(event.sessionID))
   },
 )
 
@@ -128,10 +138,10 @@ const layer = Layer.effect(
             { concurrency: "unbounded", discard: true },
           )
         }),
-        "session-shutdown-recovery",
+        `session-shutdown-recovery:${Database.path()}`,
       )
       .pipe(
-        Effect.tapCause((cause) => Effect.logError("Failed to acquire Session recovery ownership", cause)),
+        Effect.tapCause((cause) => Effect.logError("Failed to recover Sessions after shutdown", cause)),
         Effect.ignore,
         Effect.forkScoped,
       )