Preserve cancelled cloud upstream runs

server/src/__tests__/cloud-upstreams.test.ts

@@ -1,4 +1,4 @@
-import { randomUUID } from "node:crypto";
+import { generateKeyPairSync, randomUUID } from "node:crypto";
 import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
 import { companies, cloudUpstreamConnections, cloudUpstreamRuns, createDb } from "@paperclipai/db";
 
@@ -222,6 +222,47 @@ describeEmbeddedPostgres("cloud upstream persistence", () => {
     });
   });
 
+  it("preserves a cancelled run when an in-flight createRun tries to finish", async () => {
+    const companyId = randomUUID();
+    const connectionId = randomUUID();
+    await seedCompany(companyId);
+    await db.insert(cloudUpstreamConnections).values(cloudConnectionRow({ id: connectionId, companyId }));
+
+    const service = cloudUpstreamService(db);
+    const remoteCalls: string[] = [];
+    globalThis.fetch = vi.fn(async (input) => {
+      const path = new URL(String(input)).pathname;
+      remoteCalls.push(path);
+      if (path.endsWith("/upstream-imports/runs")) {
+        return jsonResponse({ run: { id: "remote-run-1" } });
+      }
+      if (path.endsWith("/chunks")) {
+        const run = await db.select().from(cloudUpstreamRuns).then((rows) => rows[0]);
+        expect(run?.status).toBe("running");
+        await service.cancelRun(connectionId, run.id, companyId);
+        return jsonResponse({ ok: true });
+      }
+      if (path.endsWith("/cancel")) {
+        return jsonResponse({ ok: true });
+      }
+      if (path.endsWith("/apply")) {
+        return jsonResponse({ ok: true });
+      }
+      if (path.endsWith("/events")) {
+        return jsonResponse({ events: [] });
+      }
+      return jsonResponse({ error: "not_found" }, 404);
+    }) as typeof fetch;
+
+    const result = await service.createRun({ connectionId, companyId });
+
+    expect(result.status).toBe("cancelled");
+    expect(remoteCalls.some((path) => path.endsWith("/apply"))).toBe(false);
+    const rows = await db.select().from(cloudUpstreamRuns);
+    expect(rows).toHaveLength(1);
+    expect(rows[0]?.status).toBe("cancelled");
+  });
+
   async function seedCompany(companyId: string) {
     await db.insert(companies).values({
       id: companyId,
@@ -240,6 +281,7 @@ function jsonResponse(body: unknown): Response {
 }
 
 function cloudConnectionRow(input: { id: string; companyId: string }) {
+  const { privateKey } = generateKeyPairSync("ed25519");
   return {
     id: input.id,
     companyId: input.companyId,
@@ -247,7 +289,7 @@ function cloudConnectionRow(input: { id: string; companyId: string }) {
     sourceInstanceId: "source-1",
     sourceInstanceFingerprint: "sha256:test",
     sourcePublicKey: "public-key",
-    privateKeyPem: "legacy-private-key",
+    privateKeyPem: privateKey.export({ type: "pkcs8", format: "pem" }).toString(),
     tokenStatus: "connected",
     scopes: ["upstream_import:write"],
     authorizedGlobalUserId: "user-1",

server/src/services/cloud-upstreams.ts

@@ -332,7 +332,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
           entities: bundle.entities,
         });
         const remoteRunId = remoteRunIdFromResponse(remoteRun);
-        await updateRun(runId, {
+        const pushedRun = await updateRunIfRunning(runId, {
           remoteRunId,
           activeStep: "push",
           progressPercent: 60,
@@ -341,11 +341,12 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
             event(new Date().toISOString(), "push", "updated", "Created or resumed the cloud import ledger run."),
           ],
         });
+        if (pushedRun.status !== "running") return pushedRun;
 
         for (const chunk of bundle.chunks) {
           await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/chunks`, chunk);
         }
-        await updateRun(runId, {
+        const verifiedRun = await updateRunIfRunning(runId, {
           activeStep: "verify",
           progressPercent: 82,
           events: [
@@ -353,6 +354,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
             event(new Date().toISOString(), "push", "completed", `Uploaded ${bundle.chunks.length} manifest chunk${bundle.chunks.length === 1 ? "" : "s"}.`),
           ],
         });
+        if (verifiedRun.status !== "running") return verifiedRun;
 
         const applied = await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/apply`, {});
         const remoteEvents = await remoteGet(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/events`).catch(() => null);
@@ -364,7 +366,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
           event(completedAt.toISOString(), "activate", "completed", "Activation checklist is ready for manual unpause decisions."),
           ...eventsFromRemote(remoteEvents),
         ];
-        const finalRun = await updateRun(runId, {
+        const finalRun = await updateRunIfRunning(runId, {
           remoteRunId,
           status: "succeeded",
           activeStep: "activate",
@@ -384,15 +386,17 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
           },
           completedAt,
         });
-        await db
-          .update(cloudUpstreamConnections)
-          .set({ lastRunId: finalRun.id, updatedAt: new Date() })
-          .where(eq(cloudUpstreamConnections.id, connection.id));
+        if (finalRun.status === "succeeded") {
+          await db
+            .update(cloudUpstreamConnections)
+            .set({ lastRunId: finalRun.id, updatedAt: new Date() })
+            .where(eq(cloudUpstreamConnections.id, connection.id));
+        }
         return finalRun;
       } catch (error) {
         const failedAt = new Date();
         const failure = cloudUpstreamRemoteFailureReport(error);
-        return updateRun(runId, {
+        return updateRunIfRunning(runId, {
           status: "failed",
           activeStep: "push",
           progressPercent: 100,
@@ -537,6 +541,23 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
     return runFromRow(updated);
   }
 
+  async function updateRunIfRunning(runId: string, patch: Partial<typeof cloudUpstreamRuns.$inferInsert>): Promise<CloudUpstreamRun> {
+    const [updated] = await db
+      .update(cloudUpstreamRuns)
+      .set({ ...patch, updatedAt: new Date() })
+      .where(and(eq(cloudUpstreamRuns.id, runId), eq(cloudUpstreamRuns.status, "running")))
+      .returning();
+    if (updated) return runFromRow(updated);
+
+    const [current] = await db
+      .select()
+      .from(cloudUpstreamRuns)
+      .where(eq(cloudUpstreamRuns.id, runId))
+      .limit(1);
+    if (!current) throw notFound("Cloud upstream run was not found");
+    return runFromRow(current);
+  }
+
   async function localPreview(connection: ConnectionRow): Promise<CloudUpstreamPreview> {
     return {
       connectionId: connection.id,