Handle file upload when runtime active

Author: markmiro

packages/agent-core/src/runtime-agent.ts

@@ -26,6 +26,7 @@ import type {
 import type {
   CellData,
   ExecutionQueueData,
+  FileData,
   RuntimeSessionData,
 } from "@runtimed/schema";
 
@@ -46,7 +47,7 @@ export class RuntimeAgent {
   activeExecutions = new Map<string, AbortController>();
   cancellationHandlers: CancellationHandler[] = [];
   renewalInterval?: ReturnType<typeof setInterval>;
-
+  fileUpload?: (id: string) => void;
   artifactClient: IArtifactClient;
 
   config: RuntimeConfig;
@@ -64,6 +65,10 @@ export class RuntimeAgent {
     this.handlers = handlers;
   }
 
+  onFileUpload(cb: (id: string) => void) {
+    this.fileUpload = cb;
+  }
+
   /**
    * Start the runtime agent - connects to LiveStore and begins processing
    */
@@ -295,6 +300,14 @@ export class RuntimeAgent {
       }
     );
 
+    // Watch for file uploads
+    const fileUploadedQuery$ = queryDb(
+      tables.files.select().where({ notebookId: this.config.notebookId }),
+      {
+        label: "fileUploaded",
+      }
+    );
+
     // Watch for cancelled executions
     const cancelledWorkQuery$ = queryDb(
       tables.executionQueue.select().where({ status: "cancelled" }),
@@ -434,13 +447,25 @@ export class RuntimeAgent {
       },
     });
 
+    const fileUploadedSub = this.store.subscribe(fileUploadedQuery$, {
+      onUpdate: (entries: readonly FileData[]) => {
+        if (this.isShuttingDown) return;
+        if (entries.length === 0) return;
+
+        for (const entry of entries) {
+          this.fileUpload?.(entry.id);
+        }
+      },
+    });
+
     // Store subscriptions for cleanup
     this.subscriptions.push(
       assignedWorkSub,
       pendingWorkSub,
       cancelledWorkSub,
       completedExecutionsSub,
-      failedExecutionsSub
+      failedExecutionsSub,
+      fileUploadedSub
     );
   }
 

packages/pyodide-runtime/src/index.ts

@@ -259,18 +259,27 @@ export class PyodideRuntimeAgent extends LocalRuntimeAgent {
 
     // Send uploaded files to worker
     if (this.agent) {
-      const files = this.agent.store.query(
-        tables.files
-          .select()
-          .where({ notebookId: this.agent.config.notebookId })
+      const agent = this.agent;
+      const files = agent.store.query(
+        tables.files.select().where({ notebookId: agent.config.notebookId })
       );
       if (files.length > 0) {
-        const agent = this.agent;
         const filesWithUrls = files.map((file) => ({
           ...file,
           url: agent.artifactClient.getArtifactUrl(file.id),
         }));
         this.sendWorkerMessage("files", { files: filesWithUrls });
+        this.agent.onFileUpload((id) => {
+          const files = agent.store.query(
+            tables.files
+              .select()
+              .where({ notebookId: agent.config.notebookId, id })
+          );
+          const file = files[0];
+          if (file) {
+            this.sendWorkerMessage("files", { files: [file] });
+          }
+        });
       }
     }
 

packages/schema/src/index.ts

@@ -1331,6 +1331,7 @@ export type OutputData = typeof tables.outputs.Type;
 export type RuntimeSessionData = typeof tables.runtimeSessions.Type;
 export type ExecutionQueueData = typeof tables.executionQueue.Type;
 export type UiStateData = typeof tables.uiState.Type;
+export type FileData = typeof tables.files.Type;
 
 // Type guards for MediaContainer
 export function isInlineContainer<T>(