From 182131171db66fa3864706787cdeca50375eddc3 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 11 Oct 2026 09:26:49 -0500 Subject: [PATCH 1/4] Preserve verified computer waits during stale task lock recovery Keep exact task claims through preparation cleanup and controller expiry so a crashed computer wait can schedule its single successor. Retain the full before-provider proof and leave terminal tasks, Stop receipts, and contradictory evidence eligible for ordinary cleanup. Co-authored-by: Codex Co-Authored-By: Paperclip --- server/src/services/heartbeat/retries.test.ts | 63 ++++++++++++++++++- server/src/services/recovery/service.ts | 10 +++ 2 files changed, 72 insertions(+), 1 deletion(-) diff --git a/server/src/services/heartbeat/retries.test.ts b/server/src/services/heartbeat/retries.test.ts index 71763030173..2ea61f9c7e6 100644 --- a/server/src/services/heartbeat/retries.test.ts +++ b/server/src/services/heartbeat/retries.test.ts @@ -1,8 +1,9 @@ +import { recoveryService } from "../recovery/service.js"; import { canRetryComputerAdmissionWait } from "../cancelled-native-startup.js"; import { ComputerStopPendingError } from "../../modules/computers/index.js"; import { randomUUID } from "node:crypto"; import type { Db } from "@paperclipai/db"; -import { computers, environments, environmentLeases, agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues } from "@paperclipai/db"; +import { activityLog, computers, environments, environmentLeases, agents, agentWakeupRequests, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues } from "@paperclipai/db"; import { and, eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import { HttpError } from "../../errors.js"; @@ -189,6 +190,7 @@ describePostgres("heartbeat retry module database wiring", () => { afterAll(async () => { await database?.cleanup(); }); afterEach(async () => { + await db.delete(activityLog); await db.delete(environmentLeases); await db.delete(computers); await db.delete(environments); @@ -268,6 +270,65 @@ describePostgres("heartbeat retry module database wiring", () => { expect(queued.status).toBe("queued"); }); + it.each([false, true])("preserves a crashed computer wait through stale-lock recovery before scheduling one successor (cleanup receipt %s)", async settled => { + const f = await computerWaitFixture(settled); + const liveDeadline = new Date(Date.now() + 60_000); + const [waiting] = await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: liveDeadline }) + .where(eq(heartbeatRuns.id, f.run.id)).returning(); + // Crash after expected cancellation, before the successor transaction. A + // live old controller deadline intentionally delays preparation recovery. + expect(await canRetryComputerAdmissionWait(db, waiting)).toBe(settled); + const sweep = await recoveryService(db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks(); + expect(sweep.issueIds).not.toContain(f.issue.id); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toMatchObject({ + executionRunId: waiting.id, checkoutRunId: waiting.id, + }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, waiting.id)))[0].controllerLeaseExpiresAt) + .toEqual(liveDeadline); + if (!settled) { + await boundRetries().retries.promoteDueScheduledRetries(new Date()); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, waiting.id))).toHaveLength(0); + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() - 1) }) + .where(eq(heartbeatRuns.id, waiting.id)); + } + await boundRetries().retries.promoteDueScheduledRetries(new Date()); + await boundRetries().retries.promoteDueScheduledRetries(new Date()); + const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, waiting.id)); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "scheduled_retry", scheduledRetryReason: "computer_admission_wait", + contextSnapshot: { preserved: "original context", executionRetryAccounting: { failureRetries: 0 } } }); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0].executionRunId).toBe(successors[0].id); + await boundRetries().retries.promoteDueScheduledRetries(new Date(Date.now() + 31_000)); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, successors[0].id)))[0].status).toBe("queued"); + }); + + it.each(["malformed", "foreign-computer", "foreign-task", "foreign-company", "provider-event", "lease", "acknowledged-stop", "done", "cancelled"])( + "does not preserve stale task ownership from invalid or completed computer wait: %s", async kind => { + const f = await computerWaitFixture(false); + const result = structuredClone(f.run.resultJson!); + if (kind === "malformed") delete result.computerAdmission; + if (kind === "foreign-computer") (result.computerAdmission as Record).computerId = randomUUID(); + if (kind === "acknowledged-stop") { + result.computerAdmissionRetryOutcome = "aborted"; + result.executionCancellation = { state: "acknowledged", acknowledgedAt: new Date().toISOString() }; + } + await db.update(heartbeatRuns).set({ resultJson: result, + ...(kind === "foreign-task" ? { contextSnapshot: { ...f.run.contextSnapshot, issueId: randomUUID() } } : {}), + }).where(eq(heartbeatRuns.id, f.run.id)); + if (kind === "provider-event") await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, + runId: f.run.id, agentId: f.agent.id, seq: 1, eventType: "adapter.invoke", stream: "system" }); + if (kind === "lease") await db.insert(environmentLeases).values({ companyId: f.companyId, + environmentId: f.environment.id, heartbeatRunId: f.run.id, provider: "boat", status: "released", releasedAt: now }); + if (kind === "foreign-company") { + const other = await fixture(); + await db.update(issues).set({ companyId: other.companyId, assigneeAgentId: other.agent.id }).where(eq(issues.id, f.issue.id)); + } + if (kind === "done" || kind === "cancelled") await db.update(issues).set({ status: kind }).where(eq(issues.id, f.issue.id)); + expect((await recoveryService(db, { enqueueWakeup: vi.fn() }).sweepStaleIssueLocks()).issueIds).toContain(f.issue.id); + expect((await db.select().from(issues).where(eq(issues.id, f.issue.id)))[0]).toMatchObject({ executionRunId: null, checkoutRunId: null }); + }, + ); + it("rotates proof-ineligible waits so later recoverable waits are not starved", async () => { const f = await computerWaitFixture(); await db.update(heartbeatRuns).set({ resultJson: { ...f.run.resultJson, computerAdmissionRetryCheckedAt: now.toISOString() } }).where(eq(heartbeatRuns.id, f.run.id)); diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 0a45ab1619b..daaeeee7c4f 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -7,6 +7,7 @@ import { externalConversationStateSql } from "../slack-conversation-state.js"; import { isIssueReviewPathRecoveryRun } from "./review-path-recovery.js"; import { escalateExhaustedIssueReviewPathRecovery } from "./review-path-recovery-escalation.js"; import { executionRetryAccounting } from "../execution-recovery-attempt.js"; +import { isComputerAdmissionWaitBeforeProvider } from "../cancelled-native-startup.js"; import { isExplicitContinuationRetryClaim } from "../explicit-continuation-retry-claim.js"; import { decideLegacyContinuation, legacyDispositionEpisode, legacyDispositionFingerprint, @@ -6217,6 +6218,15 @@ export function recoveryService( // newer terminal status. Do not test claim ownership using the old status. const owner = originalOwner ? { ...originalOwner, status: runStatusById.get(originalOwner.id) ?? originalOwner.status } : undefined; + if (owner && owner.companyId === issue.companyId && owner.contextSnapshot?.issueId === issue.id && + issue.status !== "done" && issue.status !== "cancelled" && + await isComputerAdmissionWaitBeforeProvider(db, owner)) { + // Expected cancellation retains the task claim while preparation is + // unwinding, including a crash before its retry successor is committed. + // Retry admission separately waits for cleanup or old-controller expiry; + // that policy (or Stop) owns release/transfer of this verified claim. + continue; + } if (owner && isExplicitContinuationRetryClaim(issue, owner)) { // A terminal row can still own cleanup and a pending bounded retry. // Re-enter the same policy rather than erasing its exact-owner proof. From 0c4bb6fa57bb7c5c3090291617e5b95d93d77a92 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 11 Oct 2026 09:27:01 -0500 Subject: [PATCH 2/4] Define the computer admission wait retry contract Keep resource waits separate from failure accounting and require recorded task-lock and non-assignee authority. Project the provider-neutral contract from the Boat core prerequisite; keep database-backed admission proof in the integration. Co-Authored-By: Paperclip --- .../run-dispatch/domain/policy.test.ts | 6 +-- .../src/modules/run-dispatch/domain/policy.ts | 5 +- .../run-dispatch/domain/wake-context.test.ts | 9 ++++ .../run-dispatch/domain/wake-context.ts | 3 ++ .../execution-recovery-attempt.test.ts | 46 ++++++++++++++++++- .../services/execution-recovery-attempt.ts | 28 +++++++++-- 6 files changed, 87 insertions(+), 10 deletions(-) diff --git a/server/src/modules/run-dispatch/domain/policy.test.ts b/server/src/modules/run-dispatch/domain/policy.test.ts index 051c676d2c5..38e0dc3b3ee 100644 --- a/server/src/modules/run-dispatch/domain/policy.test.ts +++ b/server/src/modules/run-dispatch/domain/policy.test.ts @@ -510,13 +510,13 @@ describe("native replacement execution authority", () => { }); -describe("AI subscription wait ownership", () => { +describe.each(["ai_connection_wait", "computer_admission_wait"] as const)("%s ownership", (retryReasonKind) => { it.each([null, "another-run"])("preserves assignee lock guards and admits only recorded non-assignee waits: %s", (issueExecutionRunId) => { const gate = { - ...baseGateFacts(), retryReasonKind: "ai_connection_wait" as const, + ...baseGateFacts(), retryReasonKind, enforceIssueExecutionLock: true, issueExecutionRunId, }; - const queued = { ...baseStalenessFacts(), retryReasonKind: "ai_connection_wait" as const, issueExecutionRunId }; + const queued = { ...baseStalenessFacts(), retryReasonKind, issueExecutionRunId }; expect(decideScheduledRetryGate(gate, NOW)).toMatchObject({ allowed: false, errorCode: "issue_execution_lock_changed" }); expect(decideQueuedRunStaleness(queued, NOW)).toMatchObject({ stale: true, errorCode: "issue_execution_lock_changed" }); const nonAssignee = { issueAssigneeAgentId: "another-agent", isNonAssigneeWorkspaceBusyRetry: true }; diff --git a/server/src/modules/run-dispatch/domain/policy.ts b/server/src/modules/run-dispatch/domain/policy.ts index ec4b729839f..a6ac7a5a5fd 100644 --- a/server/src/modules/run-dispatch/domain/policy.ts +++ b/server/src/modules/run-dispatch/domain/policy.ts @@ -10,6 +10,7 @@ export type RetryReasonKind = | "max_turn_continuation" | "disposition_repair" | "native_safe_replacement" + | "computer_admission_wait" | "ai_connection_wait" | "other"; @@ -408,7 +409,7 @@ export function decideScheduledRetryGate( const lockOutcome = decideExecutionLock({ requiresExecutionLock: (requiresInProgress || - (facts.retryReasonKind === "ai_connection_wait" && !facts.isNonAssigneeWorkspaceBusyRetry)) && + ((facts.retryReasonKind === "ai_connection_wait" || facts.retryReasonKind === "computer_admission_wait") && !facts.isNonAssigneeWorkspaceBusyRetry)) && facts.enforceIssueExecutionLock, runId: facts.runId, issueExecutionRunId: facts.issueExecutionRunId, @@ -640,7 +641,7 @@ export function decideQueuedRunStaleness( const lockOutcome = decideExecutionLock({ // A server-recorded non-assignee wake never held the task execution lock. requiresExecutionLock: requiresInProgress || - (facts.retryReasonKind === "ai_connection_wait" && !facts.isNonAssigneeWorkspaceBusyRetry), + ((facts.retryReasonKind === "ai_connection_wait" || facts.retryReasonKind === "computer_admission_wait") && !facts.isNonAssigneeWorkspaceBusyRetry), runId: facts.runId, issueExecutionRunId: facts.issueExecutionRunId, }); diff --git a/server/src/modules/run-dispatch/domain/wake-context.test.ts b/server/src/modules/run-dispatch/domain/wake-context.test.ts index ac6b84eb68e..83ae2dcc79f 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.test.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.test.ts @@ -9,6 +9,15 @@ import { } from "./wake-context.js"; describe("wake context", () => { + it("requires the computer-specific non-assignee receipt for a computer wait", () => { + expect(isNonAssigneeWorkspaceBusyRetry("computer_admission_wait", { + computerAdmissionDeferredWhileAssignee: false, + })).toBe(true); + for (const context of [{}, { computerAdmissionDeferredWhileAssignee: true }, { workspaceBusyDeferredWhileAssignee: false }, { aiConnectionBusyDeferredWhileAssignee: false }]) { + expect(isNonAssigneeWorkspaceBusyRetry("computer_admission_wait", context)).toBe(false); + } + expect(isNonAssigneeWorkspaceBusyRetry("transient_failure", { computerAdmissionDeferredWhileAssignee: false })).toBe(false); + }); it("requires the subscription-specific non-assignee receipt for a subscription wait", () => { expect(isNonAssigneeWorkspaceBusyRetry("ai_connection_busy", { aiConnectionBusyDeferredWhileAssignee: false, diff --git a/server/src/modules/run-dispatch/domain/wake-context.ts b/server/src/modules/run-dispatch/domain/wake-context.ts index 2100b357d9c..2516e7a2eb0 100644 --- a/server/src/modules/run-dispatch/domain/wake-context.ts +++ b/server/src/modules/run-dispatch/domain/wake-context.ts @@ -16,6 +16,7 @@ function readNonEmptyString(value: unknown): string | null { } export const MAX_TURN_CONTINUATION_RETRY_REASON = "max_turns_continuation"; +export const COMPUTER_ADMISSION_WAIT_RETRY_REASON = "computer_admission_wait"; export const WORKSPACE_BUSY_RETRY_REASON = "workspace_busy"; export const AI_CONNECTION_POOL_WAIT_RETRY_REASON = "ai_connection_pool_wait"; export const AI_CONNECTION_BUSY_RETRY_REASON = "ai_connection_busy"; @@ -40,6 +41,8 @@ export function isNonAssigneeWorkspaceBusyRetry( contextSnapshot: Record, ): boolean { return ( + (retryReason === COMPUTER_ADMISSION_WAIT_RETRY_REASON && + contextSnapshot.computerAdmissionDeferredWhileAssignee === false) || (retryReason === WORKSPACE_BUSY_RETRY_REASON && contextSnapshot.workspaceBusyDeferredWhileAssignee === false) || ((retryReason === AI_CONNECTION_BUSY_RETRY_REASON || retryReason === AI_CONNECTION_POOL_WAIT_RETRY_REASON) && diff --git a/server/src/services/execution-recovery-attempt.test.ts b/server/src/services/execution-recovery-attempt.test.ts index c434e7b60a2..b3e2778936b 100644 --- a/server/src/services/execution-recovery-attempt.test.ts +++ b/server/src/services/execution-recovery-attempt.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "vitest"; -import { executionFailureRetryCount } from "./execution-recovery-attempt.js"; +import { executionFailureRetryCount, isComputerAdmissionWait } from "./execution-recovery-attempt.js"; describe("failure attempts across resource waits", () => { it("preserves prior failures through subscription waits without trusting unrelated context", () => { @@ -27,7 +27,7 @@ describe("failure attempts across resource waits", () => { }); describe("persisted independent accounting", () => { - it.each(["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => { + it.each(["computer_admission_wait", "max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"])("%s cannot erase prior infrastructure debits or spend more", scheduledRetryReason => { expect(executionFailureRetryCount({ scheduledRetryReason, scheduledRetryAttempt: 20, contextSnapshot: { executionRetryAccounting: { version: 1, failureRetries: 2, maxTurnContinuations: 1 } }, })).toBe(2); @@ -43,3 +43,45 @@ describe("persisted independent accounting", () => { })).toBe(2); }); }); + + +describe("structural computer admission wait receipt", () => { + function receipt() { + return { + id: "run-1", companyId: "company-1", status: "cancelled", runtimeMode: "legacy", + errorCode: "computer_admission_wait", + resultJson: { + executionRecovery: { kind: "computer_admission_wait", providerWorkStarted: false }, + cancellation: { expected: true, source: "control_plane", initiator: { type: "system" } }, + computerAdmission: { + runId: "run-1", companyId: "company-1", + computerId: "11111111-1111-4111-8111-111111111111", + environmentId: "22222222-2222-4222-8222-222222222222", stopId: "stop-1", + }, + }, + }; + } + + it("recognizes the recorded never-dispatched wait without treating a code alone as proof", () => { + const run = receipt(); + expect(isComputerAdmissionWait(run)).toBe(true); + expect(isComputerAdmissionWait({ ...run, resultJson: {} })).toBe(false); + expect(isComputerAdmissionWait({ ...run, resultJson: { ...run.resultJson, executionRecovery: { kind: "computer_admission_wait", providerWorkStarted: true } } })).toBe(false); + expect(isComputerAdmissionWait({ ...run, resultJson: { ...run.resultJson, cancellation: { ...run.resultJson.cancellation, initiator: { type: "user" } } } })).toBe(false); + }); + + it("rejects foreign run/company receipts, invalid identities and already settled retries", () => { + const run = receipt(); + for (const computerAdmission of [ + { ...run.resultJson.computerAdmission, runId: "another-run" }, + { ...run.resultJson.computerAdmission, companyId: "another-company" }, + { ...run.resultJson.computerAdmission, computerId: "invalid" }, + { ...run.resultJson.computerAdmission, environmentId: "invalid" }, + { ...run.resultJson.computerAdmission, stopId: "" }, + ]) expect(isComputerAdmissionWait({ ...run, resultJson: { ...run.resultJson, computerAdmission } })).toBe(false); + expect(isComputerAdmissionWait({ ...run, resultJson: { ...run.resultJson, computerAdmissionRetryOutcome: "superseded" } })).toBe(false); + expect(isComputerAdmissionWait({ ...run, runnerProfileJson: { nativeExecutionInput: { version: 1 } } })).toBe(false); + expect(isComputerAdmissionWait({ ...run, runtimeMode: "native" })).toBe(false); + expect(isComputerAdmissionWait({ ...run, status: "failed" })).toBe(false); + }); +}); diff --git a/server/src/services/execution-recovery-attempt.ts b/server/src/services/execution-recovery-attempt.ts index dec0d847cf7..85d8ef10636 100644 --- a/server/src/services/execution-recovery-attempt.ts +++ b/server/src/services/execution-recovery-attempt.ts @@ -32,6 +32,10 @@ function historicalFailureCount(run: RetryRun): number { const saved = count(run.contextSnapshot?.failureRetriesBeforeAiConnectionWait); if (saved !== null) return saved; } + if (run.scheduledRetryReason === "computer_admission_wait") { + const saved = count(run.contextSnapshot?.failureRetriesBeforeComputerWait); + if (saved !== null) return saved; + } if (run.scheduledRetryReason === "workspace_busy") { const saved = count(run.contextSnapshot?.failureRetriesBeforeWorkspaceWait); if (saved !== null) return saved; @@ -42,7 +46,7 @@ function historicalFailureCount(run: RetryRun): number { export function executionRetryAccounting(run: RetryRun): ExecutionRetryAccounting { const saved = savedAccounting(run); - const nonFailureLane = ["max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"].includes(run.scheduledRetryReason ?? ""); + const nonFailureLane = ["computer_admission_wait", "max_turns_continuation", "issue_disposition_repair", "workspace_busy", "ai_connection_busy", "ai_connection_pool_wait"].includes(run.scheduledRetryReason ?? ""); return { version: 1, failureRetries: Math.max(saved?.failureRetries ?? 0, saved && nonFailureLane ? 0 : historicalFailureCount(run)), @@ -57,7 +61,7 @@ export function executionFailureRetryCount(run: RetryRun): number { } export function executionRetryAttemptCount(run: RetryRun, reason: string): number { - if (reason === "workspace_busy" || reason === "ai_connection_busy" || reason === "ai_connection_pool_wait") { + if (reason === "computer_admission_wait" || reason === "workspace_busy" || reason === "ai_connection_busy" || reason === "ai_connection_pool_wait") { return run.scheduledRetryReason === reason ? count(run.scheduledRetryAttempt) ?? 0 : 0; } const accounting = executionRetryAccounting(run); @@ -67,6 +71,24 @@ export function executionRetryAttemptCount(run: RetryRun, reason: string): numbe export function accountingForScheduledRetry(run: RetryRun, reason: string, attempt: number): ExecutionRetryAccounting { const accounting = executionRetryAccounting(run); if (reason === "max_turns_continuation") accounting.maxTurnContinuations = attempt; - else if (reason !== "workspace_busy" && reason !== "ai_connection_busy" && reason !== "ai_connection_pool_wait") accounting.failureRetries = attempt; + else if (reason !== "computer_admission_wait" && reason !== "workspace_busy" && reason !== "ai_connection_busy" && reason !== "ai_connection_pool_wait") accounting.failureRetries = attempt; return accounting; } + +/** Only the executor can attest that a pending-stop wait never dispatched and finished cleanup. */ +export function isComputerAdmissionWait(run: { + id: string; companyId: string; status: string; runtimeMode: string; errorCode?: string | null; + resultJson?: Record | null; runnerProfileJson?: Record | null; +}): boolean { + const object = (value: unknown): Record => value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + const result = object(run.resultJson), recovery = object(result.executionRecovery), cancellation = object(result.cancellation), admission = object(result.computerAdmission); + return run.runtimeMode === "legacy" && run.status === "cancelled" && run.errorCode === "computer_admission_wait" && + recovery.kind === "computer_admission_wait" && recovery.providerWorkStarted === false && + result.computerAdmissionRetryOutcome == null && + cancellation.expected === true && cancellation.source === "control_plane" && + object(cancellation.initiator).type === "system" && + admission.runId === run.id && admission.companyId === run.companyId && + [admission.computerId, admission.environmentId].every(value => typeof value === "string" && /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(value)) && + typeof admission.stopId === "string" && admission.stopId.length > 0 && + !object(run.runnerProfileJson).nativeExecutionInput; +} From 01bdb33e5cf182cae472f9798acc76b837384061 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 11 Oct 2026 09:28:04 -0500 Subject: [PATCH 3/4] Preserve empty directories during persistent home seeding Stage directory entries under the existing exact upload token and no-follow directory handles before atomic publication. Preserve adoption of existing homes and bounded file streaming. Co-Authored-By: Paperclip --- .../modules/computers/adapters/boat.test.ts | 19 +++++ .../computers/adapters/remote-program.ts | 11 ++- .../computers/application/service.test.ts | 6 ++ .../modules/computers/application/service.ts | 4 +- .../services/persistent-agent-files.test.ts | 80 +++++++++++++++++++ server/src/services/persistent-agent-files.ts | 8 +- 6 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 server/src/services/persistent-agent-files.test.ts diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index e7f7c088fa1..3146a41e0ea 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -1,5 +1,6 @@ import { existsSync, + unlinkSync, readFileSync, readdirSync, statSync, @@ -511,6 +512,24 @@ describe("confined computer files", () => { error: "invalid", }); }); + it("stages empty seed directories with exact upload ownership and no-follow paths", () => { + const f = fixture(); const seedId = randomUUID(); + f.call({ action: "seed-begin", seedId }); + expect(f.call({ action: "seed-directory", seedId, path: "empty/nested" })).toEqual({}); + expect(existsSync(f.root)).toBe(false); + expect(f.call({ action: "seed-directory", seedId: randomUUID(), path: "foreign" })).toEqual({ error: "conflict" }); + expect(f.call({ action: "seed-directory", seedId, path: "../escape" })).toEqual({ error: "invalid" }); + const staging = join(f.temp, `.paperclip-seed-home-${seedId}`, "tree"); + const outside = join(f.temp, "outside"); mkdirSync(outside); + symlinkSync(outside, join(staging, "alias")); + expect(f.call({ action: "seed-directory", seedId, path: "alias/escape" })).toEqual({ error: "invalid" }); + expect(readdirSync(outside)).toEqual([]); + // A malicious link must not be included in the successful fixture publication. + unlinkSync(join(staging, "alias")); + expect(f.call({ action: "seed-commit", seedId })).toEqual({ seeded: true }); + expect(statSync(join(f.root, "empty/nested")).isDirectory()).toBe(true); + expect(statSync(join(f.root, "empty/nested")).mode & 0o777).toBe(0o700); + }); it("publishes bounded seed chunks atomically and aborts partial homes", () => { const f = fixture(); const seedId = randomUUID(); expect(f.call({ action: "seed-begin", seedId })).toEqual({ started: true }); diff --git a/server/src/modules/computers/adapters/remote-program.ts b/server/src/modules/computers/adapters/remote-program.ts index 638a267a152..9f58ed285da 100644 --- a/server/src/modules/computers/adapters/remote-program.ts +++ b/server/src/modules/computers/adapters/remote-program.ts @@ -124,7 +124,7 @@ def directory(parts,create=False,start=None): except BaseException:os.close(fd);raise # Initial uploads stay in a private sibling until a no-replace rename publishes them. # Each request carries at most one bounded chunk, never the whole home. -seed_actions=('seed-begin','seed-chunk','seed-commit','seed-abort') +seed_actions=('seed-begin','seed-directory','seed-chunk','seed-commit','seed-abort') # The durable receipt names one exact upload. Root listings (including home # admission) reclaim expired uploads even after a controller crash or publish. if act in seed_actions or (act=='list' and not p.get('path') and os.path.lexists(os.path.join(os.path.dirname(root),'.paperclip-seed-'+os.path.basename(root)+'.json'))): @@ -209,7 +209,14 @@ if act in seed_actions or (act=='list' and not p.get('path') and os.path.lexists if receipt is None or receipt['seedId']!=token:fail('conflict') stagefd=os.open(staging,os.O_RDONLY|os.O_DIRECTORY|os.O_NOFOLLOW,dir_fd=parentfd) if read_metadata(stagefd,'owner.json')!=identity(token):fail('invalid') - if act=='seed-chunk': + if act=='seed-directory': + relative=p.get('path','');safe(relative) + parts=[part for part in relative.split('/') if part not in ('','.')] + if not parts:fail('invalid') + fd=directory(['tree']+parts,create=True,start=stagefd);os.close(fd) + receipt['updatedAt']=time.time();save_receipt(receipt) + print('{}') + elif act=='seed-chunk': relative=p.get('path','');safe(relative) parts=[part for part in relative.split('/') if part not in ('','.')] if not parts:fail('invalid') diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index d1526691aed..09b558e7807 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -770,6 +770,12 @@ with open(output,'wb') as out,open(error,'wb') as err: expect(chunks.map(input => input.offset)).toEqual([0, 1024 * 1024, 2 * 1024 * 1024]); expect(chunks.every(input => Buffer.from(input.base64 as string, "base64").length <= 1024 * 1024)).toBe(true); vi.mocked(f.backend.remote).mockClear(); + await expect(files.seedFiles((async function* () {})(), ["empty", "empty/nested"])).resolves.toEqual({ seeded: true }); + const directoryRequests = vi.mocked(f.backend.remote).mock.calls.map(([, input]) => input); + expect(directoryRequests.map(input => input.action)).toEqual(["seed-begin", "seed-directory", "seed-directory", "seed-commit"]); + expect(directoryRequests.slice(1, 3).map(input => input.path)).toEqual(["empty", "empty/nested"]); + expect(new Set(directoryRequests.map(input => input.seedId)).size).toBe(1); + vi.mocked(f.backend.remote).mockClear(); await expect(files.seedFiles((async function* () { yield { path: "first", offset: 0, bytes: Buffer.from("first") }; throw new Error("source disappeared"); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index 8a2d1e85b95..ab471a74096 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -797,11 +797,13 @@ finally: base64: bytes.toString("base64"), expectedSha256, }); - const seedFiles = async (chunks: AsyncIterable<{ path: string; offset: number; bytes: Buffer }>): Promise<{ seeded: boolean }> => { + const seedFiles = async (chunks: AsyncIterable<{ path: string; offset: number; bytes: Buffer }>, directories: readonly string[] = []): Promise<{ seeded: boolean }> => { const seedId = randomUUID(); const started = await remote({ action: "seed-begin", seedId }); if (!started.started) return { seeded: false }; try { + if (directories.length > 100_000) throw new ComputerError("invalid", "Initial home exceeds its entry limit"); + for (const path of directories) await remote({ action: "seed-directory", seedId, path }); for await (const chunk of chunks) { if (chunk.bytes.length > 1024 * 1024) throw new ComputerError("invalid", "Initial home chunk exceeds its size limit"); await remote({ action: "seed-chunk", seedId, path: chunk.path, offset: chunk.offset, base64: chunk.bytes.toString("base64") }); diff --git a/server/src/services/persistent-agent-files.test.ts b/server/src/services/persistent-agent-files.test.ts new file mode 100644 index 00000000000..5d61f444829 --- /dev/null +++ b/server/src/services/persistent-agent-files.test.ts @@ -0,0 +1,80 @@ +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { seedPersistentAgentHome } from "./persistent-agent-files.js"; + +describe("persistent home initialization", () => { + const roots: string[] = []; + afterEach(async () => { for (const root of roots.splice(0)) await fs.rm(root, { recursive: true, force: true }); }); + async function local() { + const root = await fs.mkdtemp(path.join(os.tmpdir(), "boat-agent-seed-")); roots.push(root); + await fs.mkdir(path.join(root, "memory")); + await fs.writeFile(path.join(root, "AGENTS.md"), "instructions"); + await fs.writeFile(path.join(root, "memory", "binary.bin"), Buffer.from([0, 255, 1])); + return root; + } + it.each([false, true])("adopts existing remote storage without reading the controller tree (truncated: %s)", async truncated => { + const seedFiles = vi.fn(); + await seedPersistentAgentHome({ listPage: async () => ({ entries: [], truncated }), seedFiles } as never, "/missing-controller-folder"); + expect(seedFiles).not.toHaveBeenCalled(); + }); + it("seeds absent personal storage once with exact binary bytes", async () => { + const received: Record = {}; + const seedFiles = vi.fn(async (chunks: AsyncIterable<{ path: string; offset: number; bytes: Buffer }>) => { + for await (const chunk of chunks) received[chunk.path] = Buffer.concat([received[chunk.path] ?? Buffer.alloc(0), chunk.bytes]); + return { seeded: true }; + }); + await seedPersistentAgentHome({ listPage: async () => { throw { code: "not_found" }; }, seedFiles } as never, await local()); + expect(received).toEqual({ "AGENTS.md": Buffer.from("instructions"), "memory/binary.bin": Buffer.from([0, 255, 1]) }); + }); + it("streams large files in bounded chunks without importing runtime-owned files", async () => { + const root = await local(); + const expected = Buffer.alloc(2 * 1024 * 1024 + 17, 93); + await fs.writeFile(path.join(root, "large.bin"), expected); + await fs.mkdir(path.join(root, ".paperclip-runtime")); + await fs.writeFile(path.join(root, ".paperclip-runtime", "private"), "runtime material"); + await fs.mkdir(path.join(root, "memory", ".paperclip-runtime")); + await fs.writeFile(path.join(root, "memory", ".paperclip-runtime", "user.txt"), "user content"); + const chunks: Buffer[] = []; + const paths: string[] = []; + let maxChunk = 0; + await seedPersistentAgentHome({ listPage: async () => { throw { code: "not_found" }; }, seedFiles: async (stream: AsyncIterable<{ path: string; offset: number; bytes: Buffer }>) => { + for await (const chunk of stream) { + maxChunk = Math.max(maxChunk, chunk.bytes.length); paths.push(chunk.path); + if (chunk.path === "large.bin") { + expect(chunk.offset).toBe(chunks.reduce((total, b) => total + b.length, 0)); + chunks.push(Buffer.from(chunk.bytes)); + } + } + } } as never, root); + expect(maxChunk).toBe(1024 * 1024); + expect(chunks).toHaveLength(3); + expect(Buffer.concat(chunks)).toEqual(expected); + expect(paths).not.toContain(".paperclip-runtime/private"); + expect(paths).toContain("memory/.paperclip-runtime/user.txt"); + }); + it("preserves empty and nested directories in the first seed", async () => { + const root = await local(); + await fs.mkdir(path.join(root, "empty", "nested"), { recursive: true }); + await fs.mkdir(path.join(root, ".paperclip-runtime")); + const seedFiles = vi.fn(async (stream: AsyncIterable) => { + for await (const _chunk of stream) { /* consume the source */ } + return { seeded: true }; + }); + await seedPersistentAgentHome({ listPage: async () => { throw { code: "not_found" }; }, seedFiles } as never, root); + expect(seedFiles.mock.calls[0]).toEqual([expect.anything(), ["empty", "empty/nested", "memory"]]); + }); + it("does not interpret provider outages as an absent directory", async () => { + const seedFiles = vi.fn(); + await expect(seedPersistentAgentHome({ listPage: async () => { throw new Error("offline"); }, seedFiles } as never, await local())).rejects.toThrow("offline"); + expect(seedFiles).not.toHaveBeenCalled(); + }); + it("rejects symlinked initial files before any upload", async () => { + const root = await local(); + await fs.symlink(path.join(root, "AGENTS.md"), path.join(root, "alias.md")); + const seedFiles = vi.fn(); + await expect(seedPersistentAgentHome({ listPage: async () => { throw { code: "not_found" }; }, seedFiles } as never, root)).rejects.toThrow("links or special files"); + expect(seedFiles).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/services/persistent-agent-files.ts b/server/src/services/persistent-agent-files.ts index e47d095f2cf..c8d2356daef 100644 --- a/server/src/services/persistent-agent-files.ts +++ b/server/src/services/persistent-agent-files.ts @@ -51,6 +51,7 @@ export async function seedPersistentAgentHome(files: NonNullable = []; + const initialDirectories: string[] = []; let total = 0, count = 0; async function walk(relative: string) { const entries = await fs.readdir(path.join(localRoot, relative), { withFileTypes: true }).catch((error: NodeJS.ErrnoException) => { @@ -63,7 +64,10 @@ export async function seedPersistentAgentHome(files: NonNullable 100_000) throw new Error("Initial agent folder exceeds its entry limit"); const stat = await fs.lstat(path.join(localRoot, name)); - if (stat.isDirectory()) await walk(name); + if (stat.isDirectory()) { + initialDirectories.push(name); + await walk(name); + } else if (stat.isFile() && stat.nlink === 1) { total += stat.size; if (stat.size > 256 * 1024 * 1024 || total > 2 * 1024 * 1024 * 1024) throw new Error("Initial agent folder exceeds its storage limit"); @@ -88,7 +92,7 @@ export async function seedPersistentAgentHome(files: NonNullable Date: Sun, 11 Oct 2026 09:28:17 -0500 Subject: [PATCH 4/4] Start viewer lifetime after bounded computer preparation Keep a separate starting reservation for Connect, propagate its preparation deadline, and atomically recheck the exact live owner before publishing credentials and the viewer idle deadline. Co-Authored-By: Paperclip --- .../modules/computers/adapters/boat.test.ts | 22 ++++++ server/src/modules/computers/adapters/boat.ts | 8 +-- .../modules/computers/application/ports.ts | 4 +- .../computers/application/service.test.ts | 39 +++++++++++ .../modules/computers/application/service.ts | 67 ++++++++++++++----- 5 files changed, 116 insertions(+), 24 deletions(-) diff --git a/server/src/modules/computers/adapters/boat.test.ts b/server/src/modules/computers/adapters/boat.test.ts index 3146a41e0ea..0d8a62d3dc7 100644 --- a/server/src/modules/computers/adapters/boat.test.ts +++ b/server/src/modules/computers/adapters/boat.test.ts @@ -960,6 +960,28 @@ os.execve(command[0],command,env) expect(readFileSync(f.calls, "utf8")).toBe(expected.repeat(2)); }); + it("shares the remaining Connect preparation bound with input readiness and provider credentials", async () => { + let clock = 1_000_000; + const time = vi.spyOn(Date, "now").mockImplementation(() => clock); + const scoped = { ...record, id: randomUUID(), providerId: `bx_${randomUUID()}` }; + const execute = vi.fn(async () => { + clock = 1_120_001; + return { exitCode: 0, stdout: "{}", stderr: "", timedOut: false, signal: null, pid: null, startedAt: "" }; + }); + sshFactory.mockReturnValue({ execute }); + const fetcher = vi.fn(async () => { + clock += 90_000; + return json({ hostKey: "ssh-ed25519 AAAA", sshEndpoint: "fixture.invalid:2222" }); + }); + const backend = boatBackend(async () => "fixture", fetcher); + try { + await expect(backend.desktop(scoped, { deadlineMs: 1_120_000 })).rejects.toThrow("timed out"); + expect(execute).toHaveBeenCalledWith(expect.objectContaining({ command: "timeout", timeoutMs: 30_000 })); + expect(fetcher).toHaveBeenCalledOnce(); // No late desktop credential request. + await expect(backend.renew(scoped, { deadlineMs: 1_120_000 })).rejects.toThrow("timed out"); + expect(fetcher).toHaveBeenCalledOnce(); + } finally { time.mockRestore(); sshFactory.mockReset(); } + }); it("requests a fresh stream on every Connect but reuses credentials during presence", async () => { const id = randomUUID(); const scoped = { ...record, id, providerId: `bx_${id}` }; diff --git a/server/src/modules/computers/adapters/boat.ts b/server/src/modules/computers/adapters/boat.ts index eddda3aa20b..22c8ddb2686 100644 --- a/server/src/modules/computers/adapters/boat.ts +++ b/server/src/modules/computers/adapters/boat.ts @@ -633,8 +633,8 @@ print('{}') const value = await api(record, "GET", `/stops/${segment(id)}`); return stopReceipt(value.stop ?? value, id); }, - async renew(record) { - await api(record, "PATCH", "", { ttlSeconds: 300 }); + async renew(record, options) { + await api(record, "PATCH", "", { ttlSeconds: 300 }, options); }, computerTool() { return { @@ -644,7 +644,7 @@ print('{}') }, async desktop(record, options) { if (options?.checkInput !== false) - await execute(record, `${desktopReadinessProgram}\nprint('{}')`, {}); + await execute(record, `${desktopReadinessProgram}\nprint('{}')`, {}, true, options); const cached = desktopCache.get(record.id); // Explicit Connect must obtain a new stream session, including after a // failed connection. Only presence renewals reuse existing credentials. @@ -654,7 +654,7 @@ print('{}') Date.parse(cached.expiresAt) > Date.now() + 60_000 ) return cached; - const value = await api(record, "POST", "/desktop", {}); + const value = await api(record, "POST", "/desktop", {}, options); const url = new URL(value.desktopUrl); if (url.protocol !== "https:" || !url.hostname.endsWith(".on.boat.dev")) throw new ComputerError("provider_error", "Invalid Boat desktop URL"); diff --git a/server/src/modules/computers/application/ports.ts b/server/src/modules/computers/application/ports.ts index 609fc2e73ca..c3ae9409411 100644 --- a/server/src/modules/computers/application/ports.ts +++ b/server/src/modules/computers/application/ports.ts @@ -56,11 +56,11 @@ export interface ComputerBackend { record: ComputerRecord, id: string, ): Promise<{ id: string; status: string }>; - renew(record: ComputerRecord): Promise; + renew(record: ComputerRecord, options?: ComputerOperationOptions): Promise; computerTool(): { command: string; args: string[] }; desktop( record: ComputerRecord, - options?: { checkInput?: boolean }, + options?: ComputerOperationOptions & { checkInput?: boolean }, ): Promise<{ viewerUrl: string; expiresAt: string }>; ingress( record: ComputerRecord, diff --git a/server/src/modules/computers/application/service.test.ts b/server/src/modules/computers/application/service.test.ts index 09b558e7807..ab7d98e1ced 100644 --- a/server/src/modules/computers/application/service.test.ts +++ b/server/src/modules/computers/application/service.test.ts @@ -300,6 +300,45 @@ describe("computer ownership", () => { expect((await f.repository.get(f.scope)).ledger).toMatchObject({ status: "detached", action: null }); expect(f.backend.stop).toHaveBeenCalledOnce(); }); + it("starts viewer lifetime after bounded preparation, not when Connect was requested", async () => { + const f = fixture(); await f.attach(); + vi.mocked(f.backend.ready).mockImplementationOnce(async () => { + f.advance(110_000); + await f.service.reconcile(); + }); + const viewer = await f.service.connect({ ...f.scope, userId: "alice", idleTimeoutMs: 10_000 }); + expect(viewer.expiresAt).toBe("2026-10-10T12:02:00.000Z"); + expect((await f.repository.get(f.scope)).ledger.owners).toEqual([expect.objectContaining({ + id: viewer.owner.ownerId, phase: "active", deadline: viewer.expiresAt, absoluteDeadline: viewer.expiresAt, + })]); + expect(f.backend.stop).not.toHaveBeenCalled(); + expect(f.backend.ready).toHaveBeenCalledWith(expect.anything(), { deadlineMs: Date.parse("2026-10-10T12:02:00Z") }); + }); + it.each(["ready", "desktop"] as const)("does not publish a viewer retired during %s preparation", async (step) => { + const f = fixture(); await f.attach(); + const retireViewer = async () => { + const record = await f.repository.get(f.scope); + const owner = record.ledger.owners.find(owner => owner.kind === "viewer")!; + await f.service.retire({ ...f.scope, owner: { computerId: record.id, ownerId: owner.id, generation: owner.generation } }); + }; + if (step === "ready") vi.mocked(f.backend.ready).mockImplementationOnce(retireViewer); + else vi.mocked(f.backend.desktop).mockImplementationOnce(async () => { + await retireViewer(); + return { viewerUrl: "https://test.on.boat.dev/#credential", expiresAt: "2026-10-10T12:09:00Z" }; + }); + await expect(f.service.connect({ ...f.scope, userId: "alice", idleTimeoutMs: 10_000 })).rejects.toMatchObject({ code: "conflict" }); + if (step === "ready") expect(f.backend.desktop).not.toHaveBeenCalled(); + expect((await f.repository.get(f.scope)).ledger.owners.every(owner => owner.phase === "retired")).toBe(true); + }); + it("does not publish or continue viewer preparation after its absolute preparation bound", async () => { + const f = fixture(); await f.attach(); + vi.mocked(f.backend.claim).mockClear(); + vi.mocked(f.backend.ready).mockImplementationOnce(async () => { f.advance(120_001); }); + await expect(f.service.connect({ ...f.scope, userId: "alice", idleTimeoutMs: 10_000 })).rejects.toMatchObject({ code: "provider_error" }); + expect(f.backend.claim).not.toHaveBeenCalled(); + expect(f.backend.desktop).not.toHaveBeenCalled(); + expect((await f.repository.get(f.scope)).ledger.owners.every(owner => owner.phase === "retired")).toBe(true); + }); it("viewer renewals cannot extend the absolute warm deadline or use another user", async () => { const f = fixture(); await f.attach(); diff --git a/server/src/modules/computers/application/service.ts b/server/src/modules/computers/application/service.ts index ab471a74096..4fa5fccdfa2 100644 --- a/server/src/modules/computers/application/service.ts +++ b/server/src/modules/computers/application/service.ts @@ -27,6 +27,7 @@ class ComputerSessionBusyError extends ComputerError { } } +const VIEWER_PREPARATION_BUDGET_MS = 120_000; const FILE_OPERATION_BUDGET_MS = 120_000; const FILE_OPERATION_SETTLEMENT_MS = 5000; type Scope = { companyId: string; environmentId: string }; @@ -618,36 +619,66 @@ finally: input: Scope & { userId: string; idleTimeoutMs: number }, ) { timeout(input.idleTimeoutMs); - const deadline = new Date( - now().getTime() + input.idleTimeoutMs, - ).toISOString(); const data = await admitRecord(input, (record) => { + const deadlineMs = now().getTime() + VIEWER_PREPARATION_BUDGET_MS; const owner: Owner = { id: randomUUID(), generation: 1, kind: "viewer", - phase: "active", + phase: "starting", userId: input.userId, port: 0, - deadline, - absoluteDeadline: deadline, + deadline: new Date(deadlineMs).toISOString(), + absoluteDeadline: null, process: null, }; record.ledger.owners.push(owner); - return { record: structuredClone(record), owner }; + return { record: structuredClone(record), owner, deadlineMs }; }); - await backend.ready(data.record); - await backend.claim(data.record); - await backend.renew(data.record); - const viewer = await backend.desktop(data.record); - return { - ...viewer, - expiresAt: new Date( - Math.min(Date.parse(viewer.expiresAt), Date.parse(deadline)), - ).toISOString(), - owner: ref(data.record, data.owner), - }; + const ownerRef = ref(data.record, data.owner); + const operation = { deadlineMs: data.deadlineMs }; + function preparing(record: ComputerRecord) { + const owner = exactOwner(record, ownerRef); + assertAdmission(record.ledger); + if (owner.kind !== "viewer" || owner.userId !== input.userId || owner.phase !== "starting") + throw new ComputerError("conflict", "Computer connection is no longer active"); + if (now().getTime() >= data.deadlineMs || expired(owner, now())) + throw new ComputerError("provider_error", "Computer connection preparation timed out"); + return owner; + } + const assertPreparing = async () => preparing(await repository.get(input)); + try { + await assertPreparing(); + await backend.ready(data.record, operation); + await assertPreparing(); + await backend.claim(data.record, operation); + await assertPreparing(); + await backend.renew(data.record, operation); + await assertPreparing(); + const viewer = await backend.desktop(data.record, operation); + const deadline = await repository.update(input, (record) => { + const owner = preparing(record); + const expiresAt = Math.min(Date.parse(viewer.expiresAt), now().getTime() + input.idleTimeoutMs); + if (!Number.isFinite(expiresAt) || expiresAt <= now().getTime()) + throw new ComputerError("provider_error", "Computer viewer credentials expired during preparation"); + owner.phase = "active"; + owner.deadline = new Date(expiresAt).toISOString(); + owner.absoluteDeadline = owner.deadline; + return owner.deadline; + }); + return { ...viewer, expiresAt: deadline, owner: ownerRef }; + } catch (error) { + // Retire only this still-preparing reservation; never revive or change a + // concurrently disconnected/replaced owner when preparation completes late. + await repository.update(input, (record) => { + if (record.id !== ownerRef.computerId) return; + const owner = record.ledger.owners.find(owner => owner.id === ownerRef.ownerId && owner.generation === ownerRef.generation); + if (owner?.phase === "starting") { owner.phase = "retired"; owner.deadline = null; } + }); + throw error; + } } + async function renewViewer( input: Scope & { owner: OwnerRef; userId: string }, ) {