Address cloud upstream review feedback

packages/db/src/migrations/0087_cloud_upstreams.sql

@@ -5,11 +5,11 @@ CREATE TABLE IF NOT EXISTS "cloud_upstream_connections" (
   "source_instance_id" text NOT NULL,
   "source_instance_fingerprint" text NOT NULL,
   "source_public_key" text NOT NULL,
-  "private_key_pem" text NOT NULL,
+  "private_key_material" jsonb NOT NULL,
   "token_status" text NOT NULL,
   "scopes" text[] DEFAULT '{}' NOT NULL,
   "authorized_global_user_id" text,
-  "access_token" text,
+  "access_token_material" jsonb,
   "token_id" text,
   "token_expires_at" timestamp with time zone,
   "target_stack_id" text NOT NULL,

packages/db/src/schema/cloud_upstreams.ts

@@ -10,11 +10,11 @@ export const cloudUpstreamConnections = pgTable(
     sourceInstanceId: text("source_instance_id").notNull(),
     sourceInstanceFingerprint: text("source_instance_fingerprint").notNull(),
     sourcePublicKey: text("source_public_key").notNull(),
-    privateKeyPem: text("private_key_pem").notNull(),
+    privateKeyMaterial: jsonb("private_key_material").$type<Record<string, unknown>>().notNull(),
     tokenStatus: text("token_status").notNull(),
     scopes: text("scopes").array().notNull().default([]),
     authorizedGlobalUserId: text("authorized_global_user_id"),
-    accessToken: text("access_token"),
+    accessTokenMaterial: jsonb("access_token_material").$type<Record<string, unknown>>(),
     tokenId: text("token_id"),
     tokenExpiresAt: timestamp("token_expires_at", { withTimezone: true }),
 

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

@@ -0,0 +1,144 @@
+import express from "express";
+import request from "supertest";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const mockCloudUpstreamService = vi.hoisted(() => ({
+  list: vi.fn(),
+  readConnection: vi.fn(),
+  startConnect: vi.fn(),
+  finishConnect: vi.fn(),
+  preview: vi.fn(),
+  createRun: vi.fn(),
+  readRun: vi.fn(),
+  cancelRun: vi.fn(),
+  activateRunEntities: vi.fn(),
+}));
+
+vi.mock("../routes/authz.js", async () => {
+  const { forbidden } = await vi.importActual<typeof import("../errors.js")>("../errors.js");
+
+  function assertBoardOrgAccess(req: Express.Request) {
+    if (req.actor.type !== "board") {
+      throw forbidden("Board access required");
+    }
+  }
+
+  function assertCompanyAccess(req: Express.Request, companyId: string) {
+    assertBoardOrgAccess(req);
+    if (!req.actor.companyIds?.includes(companyId)) {
+      throw forbidden("User does not have access to this company");
+    }
+  }
+
+  return {
+    assertBoardOrgAccess,
+    assertCompanyAccess,
+  };
+});
+
+vi.mock("../services/index.js", () => ({
+  cloudUpstreamService: () => mockCloudUpstreamService,
+  instanceSettingsService: () => ({
+    getExperimental: vi.fn(async () => ({ enableCloudSync: true })),
+  }),
+}));
+
+let routeModules:
+  | Promise<[
+    typeof import("../middleware/index.js"),
+    typeof import("../routes/cloud-upstreams.js"),
+  ]>
+  | null = null;
+
+async function loadRouteModules() {
+  routeModules ??= Promise.all([
+    import("../middleware/index.js"),
+    import("../routes/cloud-upstreams.js"),
+  ]);
+  return routeModules;
+}
+
+async function createApp(companyIds: string[]) {
+  const [{ errorHandler }, { cloudUpstreamRoutes }] = await loadRouteModules();
+  const app = express();
+  app.use(express.json());
+  app.use((req, _res, next) => {
+    (req as any).actor = {
+      type: "board",
+      source: "session",
+      companyIds,
+    };
+    next();
+  });
+  app.use("/api", cloudUpstreamRoutes({} as any));
+  app.use(errorHandler);
+  return app;
+}
+
+describe("cloud upstream route authorization", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mockCloudUpstreamService.list.mockResolvedValue({ connections: [], runs: [] });
+    mockCloudUpstreamService.readConnection.mockResolvedValue({
+      id: "connection-1",
+      companyId: "company-a",
+      remoteUrl: "https://cloud.example",
+      target: {
+        stackId: "stack-1",
+        stackSlug: null,
+        stackDisplayName: null,
+        companyId: "cloud-company",
+        primaryHost: "cloud.example",
+        origin: "https://cloud.example",
+        product: "Paperclip Cloud",
+        schemaMajor: 1,
+        maxChunkBytes: 1024,
+      },
+      tokenStatus: "connected",
+      scopes: [],
+      authorizedGlobalUserId: null,
+      expiresAt: null,
+      createdAt: new Date(0).toISOString(),
+      updatedAt: new Date(0).toISOString(),
+      lastRunId: null,
+    });
+    mockCloudUpstreamService.preview.mockResolvedValue({
+      connectionId: "connection-1",
+      sourceCompanyId: "company-a",
+      target: {
+        stackId: "stack-1",
+        stackSlug: null,
+        stackDisplayName: null,
+        companyId: "cloud-company",
+        primaryHost: "cloud.example",
+        origin: "https://cloud.example",
+        product: "Paperclip Cloud",
+        schemaMajor: 1,
+        maxChunkBytes: 1024,
+      },
+      schemaCompatible: true,
+      summary: [],
+      warnings: [],
+      conflicts: [],
+      generatedAt: new Date(0).toISOString(),
+    });
+  });
+
+  it("rejects list requests for companies outside the board actor membership", async () => {
+    const app = await createApp(["company-a"]);
+
+    const res = await request(app).get("/api/cloud-upstreams?companyId=company-b");
+
+    expect(res.status).toBe(403);
+    expect(mockCloudUpstreamService.list).not.toHaveBeenCalled();
+  });
+
+  it("checks connection company access before previewing a connection-scoped endpoint", async () => {
+    const app = await createApp(["company-b"]);
+
+    const res = await request(app).post("/api/cloud-upstreams/connection-1/push-runs/preview").send({});
+
+    expect(res.status).toBe(403);
+    expect(mockCloudUpstreamService.preview).not.toHaveBeenCalled();
+  });
+});

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

@@ -18,6 +18,32 @@ describe("cloud upstream remote failures", () => {
     });
   });
 
+  it("redacts credential-shaped fields from cloud response details", () => {
+    const body = {
+      error: "bad_request",
+      message: "remote rejected request",
+      accessToken: "upt_secret",
+      nested: {
+        authorization: "Bearer upt_secret",
+        privateKeyPem: "-----BEGIN PRIVATE KEY-----",
+      },
+    };
+
+    expect(cloudUpstreamRemoteFailureReport(new HttpError(400, "bad_request", body))).toEqual({
+      error: "bad_request",
+      errorMessage: "remote rejected request",
+      details: {
+        error: "bad_request",
+        message: "remote rejected request",
+        accessToken: "[redacted]",
+        nested: {
+          authorization: "[redacted]",
+          privateKeyPem: "[redacted]",
+        },
+      },
+    });
+  });
+
   it("falls back to the thrown error message for non-remote failures", () => {
     expect(cloudUpstreamRemoteFailureReport(new Error("network failed"))).toEqual({
       error: "network failed",

server/src/routes/cloud-upstreams.ts

@@ -1,7 +1,7 @@
 import { Router } from "express";
 import type { Db } from "@paperclipai/db";
 import { badRequest, notFound } from "../errors.js";
-import { assertBoardOrgAccess } from "./authz.js";
+import { assertBoardOrgAccess, assertCompanyAccess } from "./authz.js";
 import { cloudUpstreamService, instanceSettingsService } from "../services/index.js";
 
 export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {}) {
@@ -16,10 +16,16 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
     }
   }
 
+  async function assertConnectionAccess(req: Parameters<typeof assertCompanyAccess>[0], connectionId: string) {
+    const connection = await service.readConnection(connectionId);
+    assertCompanyAccess(req, connection.companyId);
+  }
+
   router.get("/cloud-upstreams", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
     const companyId = stringQuery(req.query.companyId, "companyId");
+    assertCompanyAccess(req, companyId);
     res.json(await service.list(companyId));
   });
 
@@ -27,6 +33,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
     assertBoardOrgAccess(req);
     await assertEnabled();
     const companyId = stringBody(req.body, "companyId");
+    assertCompanyAccess(req, companyId);
     const remoteUrl = stringBody(req.body, "remoteUrl");
     const redirectUri = stringBody(req.body, "redirectUri");
     res.json(await service.startConnect({ companyId, remoteUrl, redirectUri }));
@@ -35,8 +42,10 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
   router.post("/cloud-upstreams/connect/finish", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
+    const pendingConnectionId = stringBody(req.body, "pendingConnectionId");
+    await assertConnectionAccess(req, pendingConnectionId);
     res.json(await service.finishConnect({
-      pendingConnectionId: stringBody(req.body, "pendingConnectionId"),
+      pendingConnectionId,
       code: stringBody(req.body, "code"),
       state: stringBody(req.body, "state"),
     }));
@@ -45,12 +54,14 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
   router.post("/cloud-upstreams/:connectionId/push-runs/preview", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
+    await assertConnectionAccess(req, req.params.connectionId);
     res.json(await service.preview(req.params.connectionId));
   });
 
   router.post("/cloud-upstreams/:connectionId/push-runs", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
+    await assertConnectionAccess(req, req.params.connectionId);
     res.json(await service.createRun({
       connectionId: req.params.connectionId,
       retryOfRunId: optionalString(req.body?.retryOfRunId),
@@ -60,18 +71,21 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
   router.get("/cloud-upstreams/:connectionId/push-runs/:runId", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
+    await assertConnectionAccess(req, req.params.connectionId);
     res.json(await service.readRun(req.params.connectionId, req.params.runId));
   });
 
   router.post("/cloud-upstreams/:connectionId/push-runs/:runId/cancel", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
+    await assertConnectionAccess(req, req.params.connectionId);
     res.json(await service.cancelRun(req.params.connectionId, req.params.runId));
   });
 
   router.post("/cloud-upstreams/:connectionId/push-runs/:runId/activation", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
+    await assertConnectionAccess(req, req.params.connectionId);
     res.json(await service.activateRunEntities({
       connectionId: req.params.connectionId,
       runId: req.params.runId,

server/src/services/cloud-upstreams.ts

@@ -1,5 +1,5 @@
 import crypto, { sign } from "node:crypto";
-import { and, count, desc, eq } from "drizzle-orm";
+import { and, count, desc, eq, lt } from "drizzle-orm";
 import type {
   CloudUpstreamConnectStartResponse,
   CloudUpstreamActivationDecision,
@@ -29,9 +29,12 @@ import {
   routines,
 } from "@paperclipai/db";
 import { badRequest, HttpError, notFound } from "../errors.js";
+import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
 import { companyPortabilityService } from "./company-portability.js";
 
 const DEFAULT_SCOPES = ["upstream_import:preview", "upstream_import:write", "upstream_import:read"];
+const RUN_HISTORY_LIMIT = 50;
+const RUN_STALE_AFTER_MS = 30 * 60 * 1000;
 const TRANSFER_SCHEMA = {
   family: "paperclip-upstream-transfer",
   version: "1.0.0",
@@ -123,6 +126,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
 
   return {
     list: async (companyId: string): Promise<CloudUpstreamsState> => {
+      await recoverStaleRunningRuns(companyId);
       const [connectionRows, runRows] = await Promise.all([
         db
           .select()
@@ -133,7 +137,8 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
           .select()
           .from(cloudUpstreamRuns)
           .where(eq(cloudUpstreamRuns.companyId, companyId))
-          .orderBy(desc(cloudUpstreamRuns.createdAt)),
+          .orderBy(desc(cloudUpstreamRuns.createdAt))
+          .limit(RUN_HISTORY_LIMIT),
       ]);
       return {
         connections: connectionRows.map(connectionFromRow),
@@ -141,6 +146,10 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
       };
     },
 
+    readConnection: async (connectionId: string): Promise<CloudUpstreamConnection> => {
+      return connectionFromRow(await getConnectionRow(connectionId));
+    },
+
     startConnect: async (input: {
       companyId: string;
       remoteUrl: string;
@@ -158,6 +167,9 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
       const codeChallenge = crypto.createHash("sha256").update(codeVerifier, "utf8").digest("base64url");
       const { publicKey, privateKey } = crypto.generateKeyPairSync("ed25519");
       const sourcePublicKey = publicKey.export({ type: "spki", format: "pem" }).toString();
+      const privateKeyMaterial = await encryptCloudCredential(
+        privateKey.export({ type: "pkcs8", format: "pem" }).toString(),
+      );
       const sourceInstanceFingerprint = `sha256:${crypto
         .createHash("sha256")
         .update(publicKey.export({ type: "spki", format: "der" }))
@@ -170,7 +182,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         sourceInstanceId,
         sourceInstanceFingerprint,
         sourcePublicKey,
-        privateKeyPem: privateKey.export({ type: "pkcs8", format: "pem" }).toString(),
+        privateKeyMaterial,
         tokenStatus: "pending",
         scopes: scopesFromDiscovery(discovery),
         targetStackId: target.stackId,
@@ -224,6 +236,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         codeVerifier: pending.pendingCodeVerifier,
       });
       const accessToken = stringField(tokenResponse, "accessToken");
+      const accessTokenMaterial = await encryptCloudCredential(accessToken);
       const token = objectField(tokenResponse, "token");
       const expiresAt = optionalString(token.expiresAt) ?? optionalString(tokenResponse.expiresAt);
       const [updated] = await db
@@ -231,7 +244,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         .set({
           tokenStatus: "connected",
           authorizedGlobalUserId: optionalString(token.globalUserId),
-          accessToken,
+          accessTokenMaterial,
           tokenId: optionalString(token.id),
           tokenExpiresAt: expiresAt ? new Date(expiresAt) : null,
           pendingState: null,
@@ -276,18 +289,11 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
       if (connection.tokenStatus !== "connected") {
         throw badRequest("Cloud upstream connection is not connected");
       }
-      const preview = await localPreview(connection);
-      if (!preview.schemaCompatible) {
-        throw badRequest("Cloud stack schema is not compatible with this local Paperclip version");
-      }
-
-      const bundle = await buildBundle(connection, "apply");
       const runId = crypto.randomUUID();
       const now = new Date();
       const initialEvents = [
         event(now.toISOString(), "connect", "completed", "Connected to the target Paperclip Cloud stack."),
-        event(now.toISOString(), "scan", "completed", "Scanned the local company inventory."),
-        event(now.toISOString(), "preview", "completed", "Generated the transfer manifest."),
+        event(now.toISOString(), "scan", "updated", "Queued local company inventory scan."),
         ...(input.retryOfRunId
           ? [event(now.toISOString(), "push", "retrying", `Retrying run ${input.retryOfRunId} with the same import ledger idempotency key.`)]
           : []),
@@ -297,113 +303,31 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         connectionId: connection.id,
         companyId: connection.companyId,
         status: "running",
-        activeStep: "push",
-        progressPercent: 45,
+        activeStep: "scan",
+        progressPercent: 5,
         dryRun: false,
         retryOfRunId: input.retryOfRunId ?? null,
-        summary: preview.summary,
-        warnings: preview.warnings,
-        conflicts: preview.conflicts,
+        summary: [],
+        warnings: [],
+        conflicts: [],
         events: initialEvents,
         report: {},
-        idempotencyKey: bundle.manifest.idempotencyKey,
-        manifestHash: bundle.manifest.manifestHash,
+        idempotencyKey: `pending:${runId}`,
+        manifestHash: `pending:${runId}`,
         targetUrl: connection.targetOrigin,
         createdAt: now,
         updatedAt: now,
       }).returning();
       if (!created) throw badRequest("Failed to create cloud upstream run");
 
-      try {
-        const remoteRun = await remotePost(connection, `/api/companies/${encodeURIComponent(connection.targetCompanyId)}/upstream-imports/runs`, {
-          mode: "apply",
-          manifest: bundle.manifest,
-          entities: bundle.entities,
-        });
-        const remoteRunId = remoteRunIdFromResponse(remoteRun);
-        await updateRun(runId, {
-          remoteRunId,
-          activeStep: "push",
-          progressPercent: 60,
-          events: [
-            ...initialEvents,
-            event(new Date().toISOString(), "push", "updated", "Created or resumed the cloud import ledger run."),
-          ],
-        });
-
-        for (const chunk of bundle.chunks) {
-          await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/chunks`, chunk);
-        }
-        await updateRun(runId, {
-          activeStep: "verify",
-          progressPercent: 82,
-          events: [
-            ...initialEvents,
-            event(new Date().toISOString(), "push", "completed", `Uploaded ${bundle.chunks.length} manifest chunk${bundle.chunks.length === 1 ? "" : "s"}.`),
-          ],
-        });
-
-        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);
-        const completedAt = new Date();
-        const finalEvents = [
-          ...initialEvents,
-          event(completedAt.toISOString(), "push", "completed", "Pushed mapped objects without duplicate creation."),
-          event(completedAt.toISOString(), "verify", "completed", "Verified the cloud import ledger and generated a run report."),
-          event(completedAt.toISOString(), "activate", "completed", "Activation checklist is ready for manual unpause decisions."),
-          ...eventsFromRemote(remoteEvents),
-        ];
-        const finalRun = await updateRun(runId, {
-          remoteRunId,
-          status: "succeeded",
-          activeStep: "activate",
-          progressPercent: 100,
-          warnings: mergeWarnings(preview.warnings, warningsFromRemote(applied)),
-          conflicts: conflictsFromRemote(applied),
-          events: finalEvents,
-          report: {
-            runId,
-            remoteRunId,
-            target: targetFromConnectionRow(connection),
-            manifestHash: bundle.manifest.manifestHash,
-            idempotencyKey: bundle.manifest.idempotencyKey,
-            retryOfRunId: input.retryOfRunId ?? null,
-            result: applied,
-            events: remoteEvents,
-          },
-          completedAt,
-        });
-        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, {
-          status: "failed",
-          activeStep: "push",
-          progressPercent: 100,
-          events: [
-            ...initialEvents,
-            event(failedAt.toISOString(), "push", "failed", failure.errorMessage ?? failure.error),
-          ],
-          report: {
-            runId,
-            target: targetFromConnectionRow(connection),
-            manifestHash: bundle.manifest.manifestHash,
-            idempotencyKey: bundle.manifest.idempotencyKey,
-            retryOfRunId: input.retryOfRunId ?? null,
-            ...failure,
-          },
-          completedAt: failedAt,
-        });
-      }
+      void runCloudPush({ connection, runId, retryOfRunId: input.retryOfRunId ?? null, initialEvents }).catch(() => undefined);
+      return runFromRow(created);
     },
 
     readRun: async (connectionId: string, runId: string): Promise<CloudUpstreamRun> => {
       const row = await getRunRow(connectionId, runId);
+      const recovered = await recoverStaleRun(row);
+      if (recovered) return recovered;
       return runFromRow(row);
     },
 
@@ -500,6 +424,179 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
     return runFromRow(updated);
   }
 
+  async function runCloudPush(input: {
+    connection: ConnectionRow;
+    runId: string;
+    retryOfRunId: string | null;
+    initialEvents: CloudUpstreamRunEvent[];
+  }): Promise<void> {
+    let preview: CloudUpstreamPreview | null = null;
+    let bundle: LocalUpstreamExportBundle | null = null;
+    try {
+      preview = await localPreview(input.connection);
+      if (!preview.schemaCompatible) {
+        throw badRequest("Cloud stack schema is not compatible with this local Paperclip version");
+      }
+      await updateRun(input.runId, {
+        activeStep: "preview",
+        progressPercent: 25,
+        summary: preview.summary,
+        warnings: preview.warnings,
+        conflicts: preview.conflicts,
+        events: [
+          ...input.initialEvents,
+          event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."),
+          event(new Date().toISOString(), "preview", "updated", "Building the transfer manifest."),
+        ],
+      });
+
+      bundle = await buildBundle(input.connection, "apply");
+      await updateRun(input.runId, {
+        activeStep: "push",
+        progressPercent: 45,
+        idempotencyKey: bundle.manifest.idempotencyKey,
+        manifestHash: bundle.manifest.manifestHash,
+        events: [
+          ...input.initialEvents,
+          event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."),
+          event(new Date().toISOString(), "preview", "completed", "Generated the transfer manifest."),
+        ],
+      });
+
+      const remoteRun = await remotePost(input.connection, `/api/companies/${encodeURIComponent(input.connection.targetCompanyId)}/upstream-imports/runs`, {
+        mode: "apply",
+        manifest: bundle.manifest,
+        entities: bundle.entities,
+      });
+      const remoteRunId = remoteRunIdFromResponse(remoteRun);
+      await updateRun(input.runId, {
+        remoteRunId,
+        activeStep: "push",
+        progressPercent: 60,
+        events: [
+          ...input.initialEvents,
+          event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."),
+          event(new Date().toISOString(), "preview", "completed", "Generated the transfer manifest."),
+          event(new Date().toISOString(), "push", "updated", "Created or resumed the cloud import ledger run."),
+        ],
+      });
+
+      for (const chunk of bundle.chunks) {
+        await remotePost(input.connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/chunks`, chunk);
+      }
+      await updateRun(input.runId, {
+        activeStep: "verify",
+        progressPercent: 82,
+        events: [
+          ...input.initialEvents,
+          event(new Date().toISOString(), "scan", "completed", "Scanned the local company inventory."),
+          event(new Date().toISOString(), "preview", "completed", "Generated the transfer manifest."),
+          event(new Date().toISOString(), "push", "completed", `Uploaded ${bundle.chunks.length} manifest chunk${bundle.chunks.length === 1 ? "" : "s"}.`),
+        ],
+      });
+
+      const applied = await remotePost(input.connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/apply`, {});
+      const remoteEvents = await remoteGet(input.connection, `/api/upstream-import-runs/${encodeURIComponent(remoteRunId)}/events`).catch(() => null);
+      const completedAt = new Date();
+      const finalEvents = [
+        ...input.initialEvents,
+        event(completedAt.toISOString(), "push", "completed", "Pushed mapped objects without duplicate creation."),
+        event(completedAt.toISOString(), "verify", "completed", "Verified the cloud import ledger and generated a run report."),
+        event(completedAt.toISOString(), "activate", "completed", "Activation checklist is ready for manual unpause decisions."),
+        ...eventsFromRemote(remoteEvents),
+      ];
+      const finalRun = await updateRun(input.runId, {
+        remoteRunId,
+        status: "succeeded",
+        activeStep: "activate",
+        progressPercent: 100,
+        warnings: mergeWarnings(preview.warnings, warningsFromRemote(applied)),
+        conflicts: conflictsFromRemote(applied),
+        events: finalEvents,
+        report: {
+          runId: input.runId,
+          remoteRunId,
+          target: targetFromConnectionRow(input.connection),
+          manifestHash: bundle.manifest.manifestHash,
+          idempotencyKey: bundle.manifest.idempotencyKey,
+          retryOfRunId: input.retryOfRunId,
+          result: applied,
+          events: remoteEvents,
+        },
+        completedAt,
+      });
+      await db
+        .update(cloudUpstreamConnections)
+        .set({ lastRunId: finalRun.id, updatedAt: new Date() })
+        .where(eq(cloudUpstreamConnections.id, input.connection.id));
+    } catch (error) {
+      const failedAt = new Date();
+      const failure = cloudUpstreamRemoteFailureReport(error);
+      await updateRun(input.runId, {
+        status: "failed",
+        activeStep: "push",
+        progressPercent: 100,
+        ...(preview ? {
+          summary: preview.summary,
+          warnings: preview.warnings,
+          conflicts: preview.conflicts,
+        } : {}),
+        ...(bundle ? {
+          idempotencyKey: bundle.manifest.idempotencyKey,
+          manifestHash: bundle.manifest.manifestHash,
+        } : {}),
+        events: [
+          ...input.initialEvents,
+          event(failedAt.toISOString(), "push", "failed", failure.errorMessage ?? failure.error),
+        ],
+        report: {
+          runId: input.runId,
+          target: targetFromConnectionRow(input.connection),
+          manifestHash: bundle?.manifest.manifestHash ?? null,
+          idempotencyKey: bundle?.manifest.idempotencyKey ?? null,
+          retryOfRunId: input.retryOfRunId,
+          ...failure,
+        },
+        completedAt: failedAt,
+      });
+    }
+  }
+
+  async function recoverStaleRunningRuns(companyId: string): Promise<void> {
+    const cutoff = new Date(Date.now() - RUN_STALE_AFTER_MS);
+    const staleRows = await db
+      .select()
+      .from(cloudUpstreamRuns)
+      .where(and(
+        eq(cloudUpstreamRuns.companyId, companyId),
+        eq(cloudUpstreamRuns.status, "running"),
+        lt(cloudUpstreamRuns.updatedAt, cutoff),
+      ));
+    await Promise.all(staleRows.map((row) => recoverStaleRun(row)));
+  }
+
+  async function recoverStaleRun(row: RunRow): Promise<CloudUpstreamRun | null> {
+    if (row.status !== "running") return null;
+    if (Date.now() - row.updatedAt.getTime() <= RUN_STALE_AFTER_MS) return null;
+    return updateRun(row.id, {
+      status: "failed",
+      activeStep: row.activeStep,
+      progressPercent: 100,
+      completedAt: new Date(),
+      events: [
+        ...row.events,
+        event(new Date().toISOString(), "push", "failed", "Cloud upstream push stopped before completion. Retry to resume with the same idempotency safeguards."),
+      ],
+      report: {
+        ...asRecord(row.report),
+        staleRecovery: {
+          recoveredAt: new Date().toISOString(),
+          reason: "server_restart_or_worker_timeout",
+        },
+      },
+    });
+  }
+
   async function localPreview(connection: ConnectionRow): Promise<CloudUpstreamPreview> {
     return {
       connectionId: connection.id,
@@ -873,7 +970,7 @@ function buildLocalChunks(entities: LocalUpstreamExportEntity[], maxEntitiesPerC
 async function remoteGet(connection: ConnectionRow, path: string): Promise<unknown> {
   const response = await fetch(`${connection.targetOrigin}${path}`, {
     method: "GET",
-    headers: proofHeaders(connection, "GET", path),
+    headers: await proofHeaders(connection, "GET", path),
   });
   return parseRemoteResponse(response);
 }
@@ -883,17 +980,21 @@ async function remotePost(connection: ConnectionRow, path: string, body: unknown
     method: "POST",
     headers: {
       "Content-Type": "application/json",
-      ...proofHeaders(connection, "POST", path),
+      ...await proofHeaders(connection, "POST", path),
     },
     body: JSON.stringify(body),
   });
   return parseRemoteResponse(response);
 }
 
-function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Record<string, string> {
-  if (!connection.accessToken || !connection.tokenId) {
+async function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Promise<Record<string, string>> {
+  if (!connection.accessTokenMaterial || !connection.tokenId) {
     throw badRequest("Cloud upstream connection is missing an import token");
   }
+  const [accessToken, privateKeyPem] = await Promise.all([
+    decryptCloudCredential(connection.accessTokenMaterial, "Cloud upstream import token"),
+    decryptCloudCredential(connection.privateKeyMaterial, "Cloud upstream signing key"),
+  ]);
   const timestamp = new Date().toISOString();
   const nonce = crypto.randomUUID();
   const payload = [
@@ -906,14 +1007,14 @@ function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch:
     nonce,
   ].join("\n");
   return {
-    Authorization: `Bearer ${connection.accessToken}`,
+    Authorization: `Bearer ${accessToken}`,
     "X-Paperclip-Upstream-Source-Instance-Id": connection.sourceInstanceId,
     "X-Paperclip-Upstream-Proof-Timestamp": timestamp,
     "X-Paperclip-Upstream-Proof-Nonce": nonce,
     "X-Paperclip-Upstream-Proof-Signature": sign(
       null,
       Buffer.from(payload, "utf8"),
-      connection.privateKeyPem,
+      privateKeyPem,
     ).toString("base64url"),
   };
 }
@@ -943,10 +1044,39 @@ export function cloudUpstreamRemoteFailureReport(error: unknown): {
   return {
     error: remote.error ?? error.message,
     ...(remote.message ? { errorMessage: remote.message } : {}),
-    ...(error.details !== undefined ? { details: error.details } : {}),
+    ...(error.details !== undefined ? { details: redactCloudCredentialDetails(error.details) } : {}),
   };
 }
 
+async function encryptCloudCredential(value: string): Promise<Record<string, unknown>> {
+  const prepared = await localEncryptedProvider.createSecret({ value });
+  return prepared.material;
+}
+
+async function decryptCloudCredential(material: Record<string, unknown>, label: string): Promise<string> {
+  try {
+    return await localEncryptedProvider.resolveVersion({ material, externalRef: null });
+  } catch (error) {
+    throw badRequest(`${label} could not be decrypted`, {
+      error: error instanceof Error ? error.message : String(error),
+    });
+  }
+}
+
+function redactCloudCredentialDetails(value: unknown): unknown {
+  if (Array.isArray(value)) return value.map(redactCloudCredentialDetails);
+  if (!value || typeof value !== "object") return value;
+  const output: Record<string, unknown> = {};
+  for (const [key, entry] of Object.entries(value)) {
+    if (/access.?token|authorization|private.?key|bearer/i.test(key)) {
+      output[key] = "[redacted]";
+    } else {
+      output[key] = redactCloudCredentialDetails(entry);
+    }
+  }
+  return output;
+}
+
 function remoteErrorBody(details: unknown): { error?: string; message?: string } {
   const record = optionalObject(details);
   if (!record) return {};

ui/src/pages/CloudUpstream.tsx

@@ -92,10 +92,14 @@ export function CloudUpstream() {
     queryKey: selectedCompanyId ? queryKeys.cloudUpstreams(selectedCompanyId) : ["cloud-upstreams", "__disabled__"],
     queryFn: () => cloudUpstreamsApi.list(selectedCompanyId!),
     enabled: !!selectedCompanyId && cloudSyncEnabled,
+    refetchInterval: (query) =>
+      query.state.data?.runs.some((run) => run.status === "running") ? 2000 : false,
   });
 
   const connection = upstreamQuery.data?.connections[0] ?? null;
-  const latestRun = activeRun ?? upstreamQuery.data?.runs[0] ?? null;
+  const latestRun = activeRun
+    ? upstreamQuery.data?.runs.find((run) => run.id === activeRun.id) ?? activeRun
+    : upstreamQuery.data?.runs[0] ?? null;
 
   const callbackParams = useMemo(() => new URLSearchParams(location.search), [location.search]);
   const code = callbackParams.get("code");
@@ -165,9 +169,11 @@ export function CloudUpstream() {
       cloudUpstreamsApi.createRun(input.connectionId, { retryOfRunId: input.retryOfRunId ?? null }),
     onSuccess: async (run) => {
       setActiveRun(run);
-      setNotice(run.status === "succeeded"
-        ? "Push run completed. Review activation before unpausing automations."
-        : "Push run failed. Review the run events and retry after correcting the issue.");
+      setNotice(run.status === "running"
+        ? "Push run started. Progress will update while the upload continues."
+        : run.status === "succeeded"
+          ? "Push run completed. Review activation before unpausing automations."
+          : "Push run failed. Review the run events and retry after correcting the issue.");
       setActionError(null);
       await invalidateUpstreams();
     },