Fix materializer hash mismatch by making ExecutionCompleted/Cancelled materializers deterministic (#39)

- Add cellId to ExecutionCompleted and ExecutionCancelled event schemas
- Remove ctx.query() calls from materializers to make them pure functions
- Update all event commits to include cellId in the payload
- This fixes LiveStore.UnexpectedError about materializer hash mismatch

The issue was that materializers were using ctx.query() to look up cellId
during materialization, making them non-deterministic. LiveStore requires
materializers to be pure functions without side effects, with all necessary
data passed via the event payload.

Author: rgbkrk

packages/dev-server-kernel-ls-client/src/kernel-adapter.ts

@@ -565,6 +565,7 @@ async function processExecution(queueEntry: any) {
     const hasErrors = outputs.some(o => o.type === "error");
     store.commit(events.executionCompleted({
       queueId: queueEntry.id,
+      cellId: queueEntry.cellId,
       status: hasErrors ? "error" : "success",
       error: hasErrors ? "Execution completed with errors" : undefined,
     }));
@@ -580,6 +581,7 @@ async function processExecution(queueEntry: any) {
     try {
       store.commit(events.executionCompleted({
         queueId: queueEntry.id,
+        cellId: queueEntry.cellId,
         status: "error",
         error: error instanceof Error ? error.message : String(error),
       }));
@@ -624,6 +626,7 @@ assignedWorkSubscription = store.subscribe(assignedWorkQuery$ as any, {
           try {
             store.commit(events.executionCompleted({
               queueId: queueEntry.id,
+              cellId: queueEntry.cellId,
               status: "error",
               error: error instanceof Error ? error.message : String(error),
             }));

packages/dev-server-kernel-ls-client/test/kernel-adapter.test.ts

@@ -206,6 +206,7 @@ describe('Kernel Adapter', () => {
       // Complete execution
       store.commit(events.executionCompleted({
         queueId,
+        cellId,
         status: 'success',
       }))
 
@@ -384,6 +385,7 @@ describe('Kernel Adapter', () => {
         if (i % 2 === 0) {
           store.commit(events.executionCompleted({
             queueId,
+            cellId: `cell-${i}`,
             status: 'success'
           }))
         }

shared/schema.ts

@@ -288,6 +288,7 @@ export const events = {
     name: "v1.ExecutionCompleted",
     schema: Schema.Struct({
       queueId: Schema.String,
+      cellId: Schema.String,
       status: Schema.Literal("success", "error", "cancelled"),
       error: Schema.optional(Schema.String),
     }),
@@ -297,6 +298,7 @@ export const events = {
     name: "v1.ExecutionCancelled",
     schema: Schema.Struct({
       queueId: Schema.String,
+      cellId: Schema.String,
       cancelledBy: Schema.String,
       reason: Schema.String,
     }),
@@ -491,55 +493,35 @@ const materializers = State.SQLite.materializers(events, {
     ];
   },
 
-  "v1.ExecutionCompleted": ({ queueId, status }, ctx) => {
-    // Get the queue entry to find the cell ID
-    const queueEntries = ctx.query(
-      tables.executionQueue.select().where({ id: queueId }).limit(1),
-    );
-    if (queueEntries.length === 0) return [];
-
-    const queueEntry = queueEntries[0] as any;
-
-    return [
-      // Update execution queue
-      tables.executionQueue
-        .update({
-          status: status === "success" ? "completed" : "failed",
-        })
-        .where({ id: queueId }),
-      // Update cell execution state
-      tables.cells
-        .update({
-          executionState: status === "success" ? "completed" : "error",
-        })
-        .where({ id: queueEntry.cellId }),
-    ];
-  },
-
-  "v1.ExecutionCancelled": ({ queueId }, ctx) => {
-    // Get the queue entry to find the cell ID
-    const queueEntries = ctx.query(
-      tables.executionQueue.select().where({ id: queueId }).limit(1),
-    );
-    if (queueEntries.length === 0) return [];
-
-    const queueEntry = queueEntries[0] as any;
+  "v1.ExecutionCompleted": ({ queueId, cellId, status }) => [
+    // Update execution queue
+    tables.executionQueue
+      .update({
+        status: status === "success" ? "completed" : "failed",
+      })
+      .where({ id: queueId }),
+    // Update cell execution state
+    tables.cells
+      .update({
+        executionState: status === "success" ? "completed" : "error",
+      })
+      .where({ id: cellId }),
+  ],
 
-    return [
-      // Update execution queue
-      tables.executionQueue
-        .update({
-          status: "cancelled",
-        })
-        .where({ id: queueId }),
-      // Update cell execution state
-      tables.cells
-        .update({
-          executionState: "idle",
-        })
-        .where({ id: queueEntry.cellId }),
-    ];
-  },
+  "v1.ExecutionCancelled": ({ queueId, cellId }) => [
+    // Update execution queue
+    tables.executionQueue
+      .update({
+        status: "cancelled",
+      })
+      .where({ id: queueId }),
+    // Update cell execution state
+    tables.cells
+      .update({
+        executionState: "idle",
+      })
+      .where({ id: cellId }),
+  ],
 
   // Output materializers
   "v1.CellOutputAdded": ({

test/fixtures/index.ts

@@ -120,6 +120,7 @@ export const mockEvents = {
     name: "v1.ExecutionCompleted",
     args: {
       queueId: mockExecutionQueueEntry.id,
+      cellId: mockCellData.id,
       status: "success" as const,
       error: null,
     },

test/integration/execution-flow.test.ts

@@ -185,6 +185,7 @@ describe.skip("End-to-End Execution Flow", () => {
       store.commit(
         events.executionCompleted({
           queueId,
+          cellId,
           status: "success",
         }),
       );
@@ -303,6 +304,7 @@ describe.skip("End-to-End Execution Flow", () => {
       store.commit(
         events.executionCompleted({
           queueId,
+          cellId,
           status: "error",
           error: "ValueError: Test error",
         }),
@@ -423,6 +425,7 @@ describe.skip("End-to-End Execution Flow", () => {
         store.commit(
           events.executionCompleted({
             queueId,
+            cellId: cellIds[i],
             status: "success",
           }),
         );

test/integration/reactivity-debugging.test.ts

@@ -168,6 +168,7 @@ describe("Reactivity Debugging", () => {
       store.commit(
         events.executionCompleted({
           queueId: "multi-sub-queue",
+          cellId: "multi-sub-cell",
           status: "success",
         }),
       );