Fix cloud upstream review findings

packages/db/src/schema/cloud_upstreams.ts

@@ -10,10 +10,12 @@ export const cloudUpstreamConnections = pgTable(
     sourceInstanceId: text("source_instance_id").notNull(),
     sourceInstanceFingerprint: text("source_instance_fingerprint").notNull(),
     sourcePublicKey: text("source_public_key").notNull(),
+    // Stored through the Cloud Upstream service as an encrypted credential envelope.
     privateKeyPem: text("private_key_pem").notNull(),
     tokenStatus: text("token_status").notNull(),
     scopes: text("scopes").array().notNull().default([]),
     authorizedGlobalUserId: text("authorized_global_user_id"),
+    // Stored through the Cloud Upstream service as an encrypted credential envelope.
     accessToken: text("access_token"),
     tokenId: text("token_id"),
     tokenExpiresAt: timestamp("token_expires_at", { withTimezone: true }),

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

@@ -1,7 +1,28 @@
-import { describe, expect, it } from "vitest";
+import { randomUUID } from "node:crypto";
+import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
+import { companies, cloudUpstreamConnections, cloudUpstreamRuns, createDb } from "@paperclipai/db";
 
 import { HttpError } from "../errors.js";
-import { cloudUpstreamRemoteFailureReport } from "../services/cloud-upstreams.js";
+import {
+  cloudUpstreamRemoteFailureReport,
+  cloudUpstreamService,
+  reconcileCloudUpstreamRunsOnStartup,
+  sealCloudUpstreamCredential,
+  unsealCloudUpstreamCredential,
+} from "../services/cloud-upstreams.js";
+import {
+  getEmbeddedPostgresTestSupport,
+  startEmbeddedPostgresTestDatabase,
+} from "./helpers/embedded-postgres.js";
+
+const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
+
+if (!embeddedPostgresSupport.supported) {
+  console.warn(
+    `Skipping embedded Postgres cloud upstream tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`,
+  );
+}
 
 describe("cloud upstream remote failures", () => {
   it("preserves the cloud response body and message on run reports", () => {
@@ -24,3 +45,207 @@ describe("cloud upstream remote failures", () => {
     });
   });
 });
+
+describe("cloud upstream credential storage", () => {
+  const previousMasterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY;
+
+  afterEach(() => {
+    if (previousMasterKey === undefined) {
+      delete process.env.PAPERCLIP_SECRETS_MASTER_KEY;
+    } else {
+      process.env.PAPERCLIP_SECRETS_MASTER_KEY = previousMasterKey;
+    }
+  });
+
+  it("stores new credentials as encrypted envelopes and preserves legacy plaintext reads", async () => {
+    process.env.PAPERCLIP_SECRETS_MASTER_KEY = "12345678901234567890123456789012";
+    const sealed = await sealCloudUpstreamCredential("cloud-access-token");
+
+    expect(sealed).toMatch(/^paperclip-cloud-credential:/);
+    expect(sealed).not.toContain("cloud-access-token");
+    await expect(unsealCloudUpstreamCredential(sealed)).resolves.toBe("cloud-access-token");
+    await expect(unsealCloudUpstreamCredential("legacy-plaintext-token")).resolves.toBe("legacy-plaintext-token");
+  });
+});
+
+describeEmbeddedPostgres("cloud upstream persistence", () => {
+  let db!: ReturnType<typeof createDb>;
+  let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
+  const previousMasterKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY;
+
+  beforeAll(async () => {
+    process.env.PAPERCLIP_SECRETS_MASTER_KEY = "12345678901234567890123456789012";
+    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-cloud-upstreams-");
+    db = createDb(tempDb.connectionString);
+  }, 20_000);
+
+  afterEach(async () => {
+    vi.restoreAllMocks();
+    await db.delete(cloudUpstreamRuns);
+    await db.delete(cloudUpstreamConnections);
+    await db.delete(companies);
+  });
+
+  afterAll(async () => {
+    if (previousMasterKey === undefined) {
+      delete process.env.PAPERCLIP_SECRETS_MASTER_KEY;
+    } else {
+      process.env.PAPERCLIP_SECRETS_MASTER_KEY = previousMasterKey;
+    }
+    await tempDb?.cleanup();
+  });
+
+  it("encrypts stored upstream credentials while keeping connection flows usable", async () => {
+    const companyId = randomUUID();
+    await seedCompany(companyId);
+    const tokenUrl = "https://cloud.example.test/oauth/token";
+    vi.spyOn(globalThis, "fetch").mockImplementation(async (input, init) => {
+      const url = String(input);
+      if (url.startsWith("https://cloud.example.test/.well-known/paperclip-upstream")) {
+        return jsonResponse({
+          product: "Paperclip Cloud",
+          stack: {
+            id: "stack-1",
+            companyId: "cloud-company-1",
+            origin: "https://cloud.example.test",
+            primaryHost: "cloud.example.test",
+          },
+          transfer: {
+            supportedSchemaMajor: 1,
+            maxChunkBytes: 8192,
+          },
+          auth: {
+            scopes: ["upstream_import:write"],
+            pkce: {
+              authorizeUrl: "https://cloud.example.test/oauth/authorize",
+              tokenUrl,
+            },
+          },
+        });
+      }
+      if (url === tokenUrl && init?.method === "POST") {
+        const payload = JSON.parse(String(init.body));
+        expect(payload.codeVerifier).toEqual(expect.any(String));
+        expect(payload.codeVerifier).not.toContain("paperclip-cloud-credential:");
+        return jsonResponse({
+          accessToken: "cloud-access-token",
+          token: {
+            id: "token-1",
+            expiresAt: "2026-05-22T13:00:00.000Z",
+            globalUserId: "user-1",
+          },
+        });
+      }
+      throw new Error(`Unexpected fetch: ${url}`);
+    });
+
+    const service = cloudUpstreamService(db, { instanceId: "test" });
+    const started = await service.startConnect({
+      companyId,
+      remoteUrl: "https://cloud.example.test",
+      redirectUri: "http://localhost:3100/callback",
+    });
+    await service.finishConnect({
+      pendingConnectionId: started.pendingConnectionId,
+      code: "auth-code",
+      state: new URL(started.authorizationUrl).searchParams.get("state") ?? "",
+    });
+
+    const [row] = await db.select().from(cloudUpstreamConnections);
+    expect(row.privateKeyPem).toMatch(/^paperclip-cloud-credential:/);
+    expect(row.privateKeyPem).not.toContain("BEGIN PRIVATE KEY");
+    expect(row.accessToken).toMatch(/^paperclip-cloud-credential:/);
+    expect(row.accessToken).not.toContain("cloud-access-token");
+  });
+
+  it("marks orphaned running runs failed during startup reconciliation", async () => {
+    const companyId = randomUUID();
+    const connectionId = randomUUID();
+    const runningRunId = randomUUID();
+    const succeededRunId = randomUUID();
+    const reconciledAt = new Date("2026-05-22T13:00:00.000Z");
+    await seedCompany(companyId);
+    await db.insert(cloudUpstreamConnections).values({
+      id: connectionId,
+      companyId,
+      remoteUrl: "https://cloud.example.test",
+      sourceInstanceId: "source-1",
+      sourceInstanceFingerprint: "sha256:test",
+      sourcePublicKey: "public-key",
+      privateKeyPem: "legacy-private-key",
+      tokenStatus: "connected",
+      scopes: ["upstream_import:write"],
+      authorizedGlobalUserId: "user-1",
+      accessToken: "legacy-token",
+      tokenId: "token-1",
+      targetStackId: "stack-1",
+      targetCompanyId: "cloud-company-1",
+      targetOrigin: "https://cloud.example.test",
+      targetPrimaryHost: "cloud.example.test",
+      targetProduct: "Paperclip Cloud",
+      targetSchemaMajor: 1,
+      targetMaxChunkBytes: 8192,
+    });
+    await db.insert(cloudUpstreamRuns).values([
+      cloudRunRow({ id: runningRunId, connectionId, companyId, status: "running" }),
+      cloudRunRow({ id: succeededRunId, connectionId, companyId, status: "succeeded", completedAt: reconciledAt }),
+    ]);
+
+    await expect(reconcileCloudUpstreamRunsOnStartup(db, reconciledAt)).resolves.toEqual({ reconciled: 1 });
+
+    const rows = await db.select().from(cloudUpstreamRuns);
+    const running = rows.find((row) => row.id === runningRunId);
+    const succeeded = rows.find((row) => row.id === succeededRunId);
+    expect(running?.status).toBe("failed");
+    expect(running?.completedAt?.toISOString()).toBe(reconciledAt.toISOString());
+    expect(running?.events.at(-1)?.message).toContain("server startup");
+    expect(running?.report).toMatchObject({
+      error: "orphaned_running_run",
+      reconciledAt: reconciledAt.toISOString(),
+    });
+    expect(succeeded?.status).toBe("succeeded");
+  });
+
+  async function seedCompany(companyId: string) {
+    await db.insert(companies).values({
+      id: companyId,
+      name: "Paperclip",
+      issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
+      requireBoardApprovalForNewAgents: false,
+    });
+  }
+});
+
+function jsonResponse(body: unknown): Response {
+  return new Response(JSON.stringify(body), {
+    status: 200,
+    headers: { "Content-Type": "application/json" },
+  });
+}
+
+function cloudRunRow(input: {
+  id: string;
+  connectionId: string;
+  companyId: string;
+  status: string;
+  completedAt?: Date;
+}) {
+  return {
+    id: input.id,
+    connectionId: input.connectionId,
+    companyId: input.companyId,
+    status: input.status,
+    activeStep: "push",
+    progressPercent: input.status === "running" ? 45 : 100,
+    dryRun: false,
+    summary: [],
+    warnings: [],
+    conflicts: [],
+    events: [],
+    report: {},
+    idempotencyKey: `key-${input.id}`,
+    manifestHash: `sha256:${input.id.replace(/-/g, "")}`,
+    targetUrl: "https://cloud.example.test",
+    completedAt: input.completedAt,
+  };
+}

server/src/__tests__/server-startup-feedback-export.test.ts

@@ -166,6 +166,7 @@ vi.mock("../services/index.js", () => ({
       },
     })),
   })),
+  reconcileCloudUpstreamRunsOnStartup: vi.fn(async () => ({ reconciled: 0 })),
   reconcilePersistedRuntimeServicesOnStartup: vi.fn(async () => ({ reconciled: 0 })),
   routineService: vi.fn(() => ({
     tickScheduledTriggers: vi.fn(async () => ({ triggered: 0 })),

server/src/index.ts

@@ -33,6 +33,7 @@ import {
   backfillPrincipalAccessCompatibility,
   heartbeatService,
   instanceSettingsService,
+  reconcileCloudUpstreamRunsOnStartup,
   reconcilePersistedRuntimeServicesOnStartup,
   routineService,
 } from "./services/index.js";
@@ -699,6 +700,19 @@ export async function startServer(): Promise<StartedServer> {
     .catch((err) => {
       logger.error({ err }, "startup reconciliation of persisted runtime services failed");
     });
+
+  void reconcileCloudUpstreamRunsOnStartup(db as any)
+    .then((result) => {
+      if (result.reconciled > 0) {
+        logger.warn(
+          { reconciled: result.reconciled },
+          "reconciled cloud upstream runs from a previous server process",
+        );
+      }
+    })
+    .catch((err) => {
+      logger.error({ err }, "startup reconciliation of cloud upstream runs failed");
+    });
   
   if (config.heartbeatSchedulerEnabled) {
     const heartbeat = heartbeatService(db as any, { pluginWorkerManager });

server/src/routes/cloud-upstreams.ts

@@ -45,7 +45,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
   router.post("/cloud-upstreams/:connectionId/push-runs/preview", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
-    res.json(await service.preview(req.params.connectionId));
+    res.json(await service.preview(req.params.connectionId, stringBody(req.body, "companyId")));
   });
 
   router.post("/cloud-upstreams/:connectionId/push-runs", async (req, res) => {
@@ -53,6 +53,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
     await assertEnabled();
     res.json(await service.createRun({
       connectionId: req.params.connectionId,
+      companyId: stringBody(req.body, "companyId"),
       retryOfRunId: optionalString(req.body?.retryOfRunId),
     }));
   });
@@ -60,13 +61,13 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
   router.get("/cloud-upstreams/:connectionId/push-runs/:runId", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
-    res.json(await service.readRun(req.params.connectionId, req.params.runId));
+    res.json(await service.readRun(req.params.connectionId, req.params.runId, stringQuery(req.query.companyId, "companyId")));
   });
 
   router.post("/cloud-upstreams/:connectionId/push-runs/:runId/cancel", async (req, res) => {
     assertBoardOrgAccess(req);
     await assertEnabled();
-    res.json(await service.cancelRun(req.params.connectionId, req.params.runId));
+    res.json(await service.cancelRun(req.params.connectionId, req.params.runId, stringBody(req.body, "companyId")));
   });
 
   router.post("/cloud-upstreams/:connectionId/push-runs/:runId/activation", async (req, res) => {
@@ -75,6 +76,7 @@ export function cloudUpstreamRoutes(db: Db, options: { instanceId?: string } = {
     res.json(await service.activateRunEntities({
       connectionId: req.params.connectionId,
       runId: req.params.runId,
+      companyId: stringBody(req.body, "companyId"),
       entityType: activationEntityTypeBody(req.body),
     }));
   });

server/src/services/cloud-upstreams.ts

@@ -30,6 +30,7 @@ import {
 } from "@paperclipai/db";
 import { badRequest, HttpError, notFound } from "../errors.js";
 import { companyPortabilityService } from "./company-portability.js";
+import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
 
 const DEFAULT_SCOPES = ["upstream_import:preview", "upstream_import:write", "upstream_import:read"];
 const TRANSFER_SCHEMA = {
@@ -39,6 +40,9 @@ const TRANSFER_SCHEMA = {
   minor: 0,
 } as const;
 const DEFAULT_MAX_ENTITIES_PER_CHUNK = 100;
+const DISCOVERY_FETCH_TIMEOUT_MS = 30_000;
+const REMOTE_FETCH_TIMEOUT_MS = 120_000;
+const CLOUD_CREDENTIAL_PREFIX = "paperclip-cloud-credential:";
 
 type NormalizedSha256 = `sha256:${string}`;
 
@@ -75,7 +79,6 @@ type LocalUpstreamExportChunk = {
   totalChunks: number;
   byteLength: number;
   sha256: NormalizedSha256;
-  manifestHash: NormalizedSha256;
   payload: {
     entityKeys: SourceEntityKey[];
   };
@@ -100,9 +103,8 @@ type UpstreamTransferManifest = {
   idempotencyKey: string;
   generatedAt: string;
   entityCount: number;
-  perEntityTypeCounts: Record<string, number>;
   entities: UpstreamTransferEntityRecord[];
-  chunks: Array<Omit<LocalUpstreamExportChunk, "payload">>;
+  chunks: Array<Omit<LocalUpstreamExportChunk, "payload"> & { manifestHash: NormalizedSha256 }>;
   warnings: UpstreamTransferWarning[];
   featureFlags: string[];
   manifestHash: NormalizedSha256;
@@ -133,7 +135,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(50),
       ]);
       return {
         connections: connectionRows.map(connectionFromRow),
@@ -170,7 +173,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         sourceInstanceId,
         sourceInstanceFingerprint,
         sourcePublicKey,
-        privateKeyPem: privateKey.export({ type: "pkcs8", format: "pem" }).toString(),
+        privateKeyPem: await sealCloudUpstreamCredential(privateKey.export({ type: "pkcs8", format: "pem" }).toString()),
         tokenStatus: "pending",
         scopes: scopesFromDiscovery(discovery),
         targetStackId: target.stackId,
@@ -183,7 +186,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         targetSchemaMajor: target.schemaMajor,
         targetMaxChunkBytes: target.maxChunkBytes,
         pendingState: state,
-        pendingCodeVerifier: codeVerifier,
+        pendingCodeVerifier: await sealCloudUpstreamCredential(codeVerifier),
         pendingRedirectUri: input.redirectUri,
         pendingTokenUrl: tokenUrlFromDiscovery(discovery),
       }).returning();
@@ -221,7 +224,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         grantType: "authorization_code",
         code: input.code,
         redirectUri: pending.pendingRedirectUri,
-        codeVerifier: pending.pendingCodeVerifier,
+        codeVerifier: await unsealCloudUpstreamCredential(pending.pendingCodeVerifier),
       });
       const accessToken = stringField(tokenResponse, "accessToken");
       const token = objectField(tokenResponse, "token");
@@ -231,7 +234,7 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
         .set({
           tokenStatus: "connected",
           authorizedGlobalUserId: optionalString(token.globalUserId),
-          accessToken,
+          accessToken: await sealCloudUpstreamCredential(accessToken),
           tokenId: optionalString(token.id),
           tokenExpiresAt: expiresAt ? new Date(expiresAt) : null,
           pendingState: null,
@@ -246,8 +249,8 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
       return connectionFromRow(updated);
     },
 
-    preview: async (connectionId: string): Promise<CloudUpstreamPreview> => {
-      const connection = await getConnectionRow(connectionId);
+    preview: async (connectionId: string, companyId: string): Promise<CloudUpstreamPreview> => {
+      const connection = await getConnectionRow(connectionId, companyId);
       const basePreview = await localPreview(connection);
       if (!basePreview.schemaCompatible || connection.tokenStatus !== "connected") {
         return basePreview;
@@ -271,8 +274,8 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
       };
     },
 
-    createRun: async (input: { connectionId: string; retryOfRunId?: string | null }): Promise<CloudUpstreamRun> => {
-      const connection = await getConnectionRow(input.connectionId);
+    createRun: async (input: { connectionId: string; companyId: string; retryOfRunId?: string | null }): Promise<CloudUpstreamRun> => {
+      const connection = await getConnectionRow(input.connectionId, input.companyId);
       if (connection.tokenStatus !== "connected") {
         throw badRequest("Cloud upstream connection is not connected");
       }
@@ -402,15 +405,15 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
       }
     },
 
-    readRun: async (connectionId: string, runId: string): Promise<CloudUpstreamRun> => {
-      const row = await getRunRow(connectionId, runId);
+    readRun: async (connectionId: string, runId: string, companyId: string): Promise<CloudUpstreamRun> => {
+      const row = await getRunRow(connectionId, runId, companyId);
       return runFromRow(row);
     },
 
-    cancelRun: async (connectionId: string, runId: string): Promise<CloudUpstreamRun> => {
-      const row = await getRunRow(connectionId, runId);
+    cancelRun: async (connectionId: string, runId: string, companyId: string): Promise<CloudUpstreamRun> => {
+      const row = await getRunRow(connectionId, runId, companyId);
       if (row.status !== "running") return runFromRow(row);
-      const connection = await getConnectionRow(connectionId);
+      const connection = await getConnectionRow(connectionId, companyId);
       if (row.remoteRunId) {
         await remotePost(connection, `/api/upstream-import-runs/${encodeURIComponent(row.remoteRunId)}/cancel`, {}).catch(() => null);
       }
@@ -429,9 +432,10 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
     activateRunEntities: async (input: {
       connectionId: string;
       runId: string;
+      companyId: string;
       entityType: CloudUpstreamActivationEntityType;
     }): Promise<CloudUpstreamRun> => {
-      const row = await getRunRow(input.connectionId, input.runId);
+      const row = await getRunRow(input.connectionId, input.runId, input.companyId);
       assertActivationEntityType(input.entityType);
       if (row.status !== "succeeded") {
         throw badRequest("Only succeeded cloud upstream runs can activate imported entities");
@@ -470,21 +474,27 @@ export function cloudUpstreamService(db: Db, options: { instanceId?: string } =
     if (!row) throw notFound("Company was not found");
   }
 
-  async function getConnectionRow(connectionId: string): Promise<ConnectionRow> {
+  async function getConnectionRow(connectionId: string, companyId?: string): Promise<ConnectionRow> {
     const row = await db
       .select()
       .from(cloudUpstreamConnections)
-      .where(eq(cloudUpstreamConnections.id, connectionId))
+      .where(companyId
+        ? and(eq(cloudUpstreamConnections.id, connectionId), eq(cloudUpstreamConnections.companyId, companyId))
+        : eq(cloudUpstreamConnections.id, connectionId))
       .then((rows) => rows[0]);
     if (!row) throw notFound("Cloud upstream connection was not found");
     return row;
   }
 
-  async function getRunRow(connectionId: string, runId: string): Promise<RunRow> {
+  async function getRunRow(connectionId: string, runId: string, companyId: string): Promise<RunRow> {
     const row = await db
       .select()
       .from(cloudUpstreamRuns)
-      .where(and(eq(cloudUpstreamRuns.id, runId), eq(cloudUpstreamRuns.connectionId, connectionId)))
+      .where(and(
+        eq(cloudUpstreamRuns.id, runId),
+        eq(cloudUpstreamRuns.connectionId, connectionId),
+        eq(cloudUpstreamRuns.companyId, companyId),
+      ))
       .then((rows) => rows[0]);
     if (!row) throw notFound("Cloud upstream run was not found");
     return row;
@@ -596,13 +606,53 @@ async function fetchDiscovery(remoteUrl: string): Promise<Record<string, unknown
   if (stackId) {
     discoveryUrl.searchParams.set("stackId", stackId);
   }
-  const response = await fetch(discoveryUrl);
+  const response = await fetchWithTimeout(discoveryUrl, undefined, DISCOVERY_FETCH_TIMEOUT_MS);
   if (!response.ok) {
     throw badRequest(`Cloud upstream discovery failed: ${response.status}`);
   }
   return await response.json() as Record<string, unknown>;
 }
 
+export async function reconcileCloudUpstreamRunsOnStartup(db: Db, now = new Date()): Promise<{ reconciled: number }> {
+  const runningRows = await db
+    .select()
+    .from(cloudUpstreamRuns)
+    .where(eq(cloudUpstreamRuns.status, "running"));
+  if (runningRows.length === 0) return { reconciled: 0 };
+
+  for (const row of runningRows) {
+    const report = asRecord(row.report);
+    await db
+      .update(cloudUpstreamRuns)
+      .set({
+        status: "failed",
+        activeStep: row.activeStep,
+        progressPercent: 100,
+        completedAt: now,
+        updatedAt: now,
+        events: [
+          ...safeRunEvents(row.events),
+          event(
+            now.toISOString(),
+            cloudUpstreamStep(row.activeStep),
+            "failed",
+            "Marked failed on server startup because the previous process stopped while the cloud upstream run was in progress.",
+          ),
+        ],
+        report: {
+          ...report,
+          error: optionalString(report.error) ?? "orphaned_running_run",
+          errorMessage: optionalString(report.errorMessage)
+            ?? "The server restarted while this cloud upstream run was running, so Paperclip marked it failed instead of leaving it stuck.",
+          reconciledAt: now.toISOString(),
+        },
+      })
+      .where(eq(cloudUpstreamRuns.id, row.id));
+  }
+
+  return { reconciled: runningRows.length };
+}
+
 function firstPathSegment(pathname: string): string | null {
   const segment = pathname.split("/").find(Boolean);
   return segment && segment.toLowerCase() !== "dashboard" ? segment : null;
@@ -816,34 +866,23 @@ function buildLocalUpstreamExportBundle(input: {
     idempotencyKey: input.idempotencyKey,
     generatedAt: new Date(0).toISOString(),
     entityCount: entities.length,
-    perEntityTypeCounts: countEntityTypesForManifest(entities),
     entities: entities.map((entity) => entity.record),
-    chunks: chunksWithoutManifestHash.map(({ payload: _payload, manifestHash: _manifestHash, ...chunk }) => chunk),
+    chunks: chunksWithoutManifestHash.map(({ payload: _payload, ...chunk }) => chunk),
     warnings: input.warnings ?? [],
     featureFlags: (input.featureFlags ?? ["cloud_sync"]).slice().sort(),
   };
   const manifestHash = normalizedContentHash(manifestWithoutHash);
-  const chunks = chunksWithoutManifestHash.map((chunk) => ({ ...chunk, manifestHash }));
   return {
     manifest: {
       ...manifestWithoutHash,
-      chunks: chunks.map(({ payload: _payload, ...chunk }) => chunk),
+      chunks: chunksWithoutManifestHash.map(({ payload: _payload, ...chunk }) => ({ ...chunk, manifestHash })),
       manifestHash,
     },
     entities,
-    chunks,
+    chunks: chunksWithoutManifestHash,
   };
 }
 
-function countEntityTypesForManifest(entities: LocalUpstreamExportEntity[]): Record<string, number> {
-  const counts: Record<string, number> = {};
-  for (const entity of entities) {
-    const entityType = entity.record.key.sourceEntityType;
-    counts[entityType] = (counts[entityType] ?? 0) + 1;
-  }
-  return counts;
-}
-
 function buildLocalChunks(entities: LocalUpstreamExportEntity[], maxEntitiesPerChunk: number): LocalUpstreamExportChunk[] {
   if (!Number.isInteger(maxEntitiesPerChunk) || maxEntitiesPerChunk < 1) {
     throw new Error("maxEntitiesPerChunk must be a positive integer");
@@ -864,36 +903,37 @@ function buildLocalChunks(entities: LocalUpstreamExportEntity[], maxEntitiesPerC
       totalChunks: groups.length,
       byteLength: Buffer.byteLength(canonicalJson(payload)),
       sha256: normalizedContentHash(payload),
-      manifestHash: "sha256:" as NormalizedSha256,
       payload,
     };
   });
 }
 
 async function remoteGet(connection: ConnectionRow, path: string): Promise<unknown> {
-  const response = await fetch(`${connection.targetOrigin}${path}`, {
+  const response = await fetchWithTimeout(`${connection.targetOrigin}${path}`, {
     method: "GET",
-    headers: proofHeaders(connection, "GET", path),
-  });
+    headers: await proofHeaders(connection, "GET", path),
+  }, REMOTE_FETCH_TIMEOUT_MS);
   return parseRemoteResponse(response);
 }
 
 async function remotePost(connection: ConnectionRow, path: string, body: unknown): Promise<unknown> {
-  const response = await fetch(`${connection.targetOrigin}${path}`, {
+  const response = await fetchWithTimeout(`${connection.targetOrigin}${path}`, {
     method: "POST",
     headers: {
       "Content-Type": "application/json",
-      ...proofHeaders(connection, "POST", path),
+      ...await proofHeaders(connection, "POST", path),
     },
     body: JSON.stringify(body),
-  });
+  }, REMOTE_FETCH_TIMEOUT_MS);
   return parseRemoteResponse(response);
 }
 
-function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Record<string, string> {
+async function proofHeaders(connection: ConnectionRow, method: string, pathAndSearch: string): Promise<Record<string, string>> {
   if (!connection.accessToken || !connection.tokenId) {
     throw badRequest("Cloud upstream connection is missing an import token");
   }
+  const accessToken = await unsealCloudUpstreamCredential(connection.accessToken);
+  const privateKeyPem = await unsealCloudUpstreamCredential(connection.privateKeyPem);
   const timestamp = new Date().toISOString();
   const nonce = crypto.randomUUID();
   const payload = [
@@ -906,14 +946,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"),
   };
 }
@@ -930,6 +970,32 @@ async function parseRemoteResponse(response: Response): Promise<unknown> {
   return parsed;
 }
 
+async function fetchWithTimeout(input: RequestInfo | URL, init: RequestInit | undefined, timeoutMs: number): Promise<Response> {
+  return fetch(input, {
+    ...init,
+    signal: AbortSignal.timeout(timeoutMs),
+  });
+}
+
+export async function sealCloudUpstreamCredential(value: string): Promise<string> {
+  const prepared = await localEncryptedProvider.createSecret({ value });
+  return `${CLOUD_CREDENTIAL_PREFIX}${JSON.stringify(prepared.material)}`;
+}
+
+export async function unsealCloudUpstreamCredential(value: string): Promise<string> {
+  if (!value.startsWith(CLOUD_CREDENTIAL_PREFIX)) return value;
+  const encoded = value.slice(CLOUD_CREDENTIAL_PREFIX.length);
+  const parsed = safeParseJson(encoded);
+  const material = optionalObject(parsed);
+  if (!material) {
+    throw badRequest("Invalid encrypted cloud upstream credential material");
+  }
+  return localEncryptedProvider.resolveVersion({
+    material,
+    externalRef: null,
+  });
+}
+
 export function cloudUpstreamRemoteFailureReport(error: unknown): {
   error: string;
   errorMessage?: string;
@@ -957,11 +1023,11 @@ function remoteErrorBody(details: unknown): { error?: string; message?: string }
 }
 
 async function postJson<T>(url: string, body: unknown): Promise<T> {
-  const response = await fetch(url, {
+  const response = await fetchWithTimeout(url, {
     method: "POST",
     headers: { "Content-Type": "application/json" },
     body: JSON.stringify(body),
-  });
+  }, DISCOVERY_FETCH_TIMEOUT_MS);
   const payload = await response.json().catch(() => null);
   if (!response.ok) {
     throw badRequest((payload as { error?: string } | null)?.error ?? `Cloud upstream request failed: ${response.status}`);
@@ -1026,6 +1092,10 @@ function eventsFromRemote(value: unknown): CloudUpstreamRunEvent[] {
   });
 }
 
+function safeRunEvents(value: unknown): CloudUpstreamRunEvent[] {
+  return Array.isArray(value) ? value as CloudUpstreamRunEvent[] : [];
+}
+
 function assertActivationEntityType(value: string): asserts value is CloudUpstreamActivationEntityType {
   if (value !== "agents" && value !== "routines" && value !== "monitors") {
     throw badRequest("entityType must be agents, routines, or monitors");

server/src/services/index.ts

@@ -59,7 +59,7 @@ export type {
 } from "./authorization.js";
 export { boardAuthService } from "./board-auth.js";
 export { instanceSettingsService } from "./instance-settings.js";
-export { cloudUpstreamService } from "./cloud-upstreams.js";
+export { cloudUpstreamService, reconcileCloudUpstreamRunsOnStartup } from "./cloud-upstreams.js";
 export { companyPortabilityService } from "./company-portability.js";
 export { environmentService } from "./environments.js";
 export { executionWorkspaceService } from "./execution-workspaces.js";

ui/src/api/cloudUpstreams.ts

@@ -15,23 +15,23 @@ export const cloudUpstreamsApi = {
     api.post<CloudUpstreamConnectStartResponse>("/cloud-upstreams/connect/start", input),
   finishConnect: (input: { pendingConnectionId: string; code: string; state: string }) =>
     api.post<CloudUpstreamConnection>("/cloud-upstreams/connect/finish", input),
-  preview: (connectionId: string) =>
-    api.post<CloudUpstreamPreview>(`/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/preview`, {}),
-  createRun: (connectionId: string, input?: { retryOfRunId?: string | null }) =>
+  preview: (connectionId: string, input: { companyId: string }) =>
+    api.post<CloudUpstreamPreview>(`/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/preview`, input),
+  createRun: (connectionId: string, input: { companyId: string; retryOfRunId?: string | null }) =>
     api.post<CloudUpstreamRun>(`/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs`, input ?? {}),
-  getRun: (connectionId: string, runId: string) =>
+  getRun: (connectionId: string, runId: string, companyId: string) =>
     api.get<CloudUpstreamRun>(
-      `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}`,
+      `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}?companyId=${encodeURIComponent(companyId)}`,
     ),
-  cancelRun: (connectionId: string, runId: string) =>
+  cancelRun: (connectionId: string, runId: string, input: { companyId: string }) =>
     api.post<CloudUpstreamRun>(
       `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}/cancel`,
-      {},
+      input,
     ),
   activateEntities: (
     connectionId: string,
     runId: string,
-    input: { entityType: CloudUpstreamActivationEntityType },
+    input: { companyId: string; entityType: CloudUpstreamActivationEntityType },
   ) =>
     api.post<CloudUpstreamRun>(
       `/cloud-upstreams/${encodeURIComponent(connectionId)}/push-runs/${encodeURIComponent(runId)}/activation`,

ui/src/pages/CloudUpstream.test.tsx

@@ -142,7 +142,7 @@ describe("CloudUpstream", () => {
     expect(mockCloudUpstreamsApi.activateEntities).toHaveBeenCalledWith(
       "connection-1",
       "run-1",
-      { entityType: "agents" },
+      { companyId: "company-1", entityType: "agents" },
     );
 
     await act(async () => {
@@ -260,6 +260,39 @@ describe("CloudUpstream", () => {
     }
   });
 
+  it("does not retry the OAuth callback finish mutation after an error", async () => {
+    mockLocationState.pathname = "/PAP/company/settings/cloud-upstream";
+    mockLocationState.search = "?code=cb-code&state=cb-state";
+    mockCloudUpstreamsApi.list.mockResolvedValue({ connections: [], runs: [] });
+    mockCloudUpstreamsApi.finishConnect.mockRejectedValue(new Error("state expired"));
+    window.localStorage.setItem("paperclip-cloud-upstream-pending-connection", "pending-1");
+
+    try {
+      const root = createRoot(container);
+      const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } });
+
+      await act(async () => {
+        root.render(
+          <QueryClientProvider client={queryClient}>
+            <CloudUpstream />
+          </QueryClientProvider>,
+        );
+      });
+      await flushReact();
+      await flushReact();
+      await flushReact();
+
+      expect(mockCloudUpstreamsApi.finishConnect).toHaveBeenCalledTimes(1);
+      expect(container.textContent).toContain("state expired");
+
+      await act(async () => {
+        root.unmount();
+      });
+    } finally {
+      window.localStorage.removeItem("paperclip-cloud-upstream-pending-connection");
+    }
+  });
+
   it("keeps retry only for failed or cancelled runs", async () => {
     mockCloudUpstreamsApi.list.mockResolvedValue(stateWithRun(buildRun({ status: "failed" })));
     const root = createRoot(container);

ui/src/pages/CloudUpstream.tsx

@@ -119,16 +119,22 @@ export function CloudUpstream() {
     },
     onError: (error) => setActionError(error instanceof Error ? error.message : "Failed to finish connection."),
   });
+  const {
+    mutate: finishConnect,
+    isError: finishConnectFailed,
+    isPending: finishConnectPending,
+    isSuccess: finishConnectSucceeded,
+  } = finishMutation;
 
   useEffect(() => {
-    if (!cloudSyncEnabled || !code || !state || finishMutation.isPending || finishMutation.isSuccess) return;
+    if (!cloudSyncEnabled || !code || !state || finishConnectPending || finishConnectSucceeded || finishConnectFailed) return;
     const pendingConnectionId = localStorage.getItem(PENDING_CONNECTION_KEY);
     if (!pendingConnectionId) {
       setActionError("No pending cloud upstream connection was found. Start the connection again.");
       return;
     }
-    finishMutation.mutate({ pendingConnectionId, code, state });
-  }, [cloudSyncEnabled, code, state, finishMutation]);
+    finishConnect({ pendingConnectionId, code, state });
+  }, [cloudSyncEnabled, code, finishConnect, finishConnectFailed, finishConnectPending, finishConnectSucceeded, state]);
 
   useEffect(() => {
     if (callbackError) {
@@ -152,7 +158,8 @@ export function CloudUpstream() {
   });
 
   const previewMutation = useMutation({
-    mutationFn: (connectionId: string) => cloudUpstreamsApi.preview(connectionId),
+    mutationFn: (input: { connectionId: string; companyId: string }) =>
+      cloudUpstreamsApi.preview(input.connectionId, { companyId: input.companyId }),
     onSuccess: (nextPreview) => {
       setPreview(nextPreview);
       setActionError(null);
@@ -161,8 +168,11 @@ export function CloudUpstream() {
   });
 
   const runMutation = useMutation({
-    mutationFn: (input: { connectionId: string; retryOfRunId?: string | null }) =>
-      cloudUpstreamsApi.createRun(input.connectionId, { retryOfRunId: input.retryOfRunId ?? null }),
+    mutationFn: (input: { connectionId: string; companyId: string; retryOfRunId?: string | null }) =>
+      cloudUpstreamsApi.createRun(input.connectionId, {
+        companyId: input.companyId,
+        retryOfRunId: input.retryOfRunId ?? null,
+      }),
     onSuccess: async (run) => {
       setActiveRun(run);
       setNotice(run.status === "succeeded"
@@ -175,7 +185,10 @@ export function CloudUpstream() {
   });
   const activationMutation = useMutation({
     mutationFn: (input: { run: CloudUpstreamRun; entityType: CloudUpstreamActivationEntityType }) =>
-      cloudUpstreamsApi.activateEntities(input.run.connectionId, input.run.id, { entityType: input.entityType }),
+      cloudUpstreamsApi.activateEntities(input.run.connectionId, input.run.id, {
+        companyId: input.run.companyId,
+        entityType: input.entityType,
+      }),
     onSuccess: async (run) => {
       setActiveRun(run);
       setNotice("Activation checklist updated.");
@@ -271,7 +284,7 @@ export function CloudUpstream() {
                 <Button
                   variant="outline"
                   size="sm"
-                  onClick={() => previewMutation.mutate(connection.id)}
+                  onClick={() => previewMutation.mutate({ connectionId: connection.id, companyId: connection.companyId })}
                   disabled={previewMutation.isPending || connection.tokenStatus !== "connected"}
                 >
                   {previewMutation.isPending ? <Loader2 className="h-4 w-4 animate-spin" /> : <RefreshCcw className="h-4 w-4" />}
@@ -302,7 +315,7 @@ export function CloudUpstream() {
           <div className="flex items-center justify-between gap-3">
             <div className="text-xs font-medium uppercase tracking-wide text-muted-foreground">Preview</div>
             <Button
-              onClick={() => runMutation.mutate({ connectionId: preview.connectionId })}
+              onClick={() => runMutation.mutate({ connectionId: preview.connectionId, companyId: preview.sourceCompanyId })}
               disabled={runMutation.isPending || !preview.schemaCompatible}
             >
               {runMutation.isPending ? <Loader2 className="h-4 w-4 animate-spin" /> : <CloudUpload className="h-4 w-4" />}
@@ -328,7 +341,11 @@ export function CloudUpstream() {
                 <Button
                   variant="outline"
                   size="sm"
-                  onClick={() => runMutation.mutate({ connectionId: latestRun.connectionId, retryOfRunId: latestRun.id })}
+                  onClick={() => runMutation.mutate({
+                    connectionId: latestRun.connectionId,
+                    companyId: latestRun.companyId,
+                    retryOfRunId: latestRun.id,
+                  })}
                   disabled={runMutation.isPending}
                 >
                   <RefreshCcw className="h-4 w-4" />
@@ -338,7 +355,7 @@ export function CloudUpstream() {
                 <Button
                   variant="outline"
                   size="sm"
-                  onClick={() => runMutation.mutate({ connectionId: latestRun.connectionId })}
+                  onClick={() => runMutation.mutate({ connectionId: latestRun.connectionId, companyId: latestRun.companyId })}
                   disabled={runMutation.isPending}
                 >
                   <RefreshCcw className="h-4 w-4" />