From 450fcd81fd3bf802d9a83d6e68e28a115fb36fc0 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 2 Nov 2026 12:13:43 -0600 Subject: [PATCH] test(server): qualify completed history through domain writers Bind retained transfers and chat completion receipts to exact owner evidence, qualify every retained-history family through normal producer paths, and fence runtime stop cleanup in memory. Co-Authored-By: Paperclip --- doc/idle-sleep-safety.md | 10 +- .../__tests__/idle-completed-history.test.ts | 382 +++++++++++++++++- .../chat-channels/idle-history-proof.ts | 15 +- .../services/workspace-history-idle-proof.ts | 15 + server/src/services/workspace-runtime.ts | 29 +- 5 files changed, 431 insertions(+), 20 deletions(-) diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index d9f129d2b91..075ace1d59f 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -114,8 +114,9 @@ same closed session. Decision success requires the selected option's full effect manifest, matching ordinal/type/target receipts and audit records. The owning dismissal, explicit recovery stop and reviewed archive paths have separate proofs. Their continuation and archive-notification outboxes remain independent barriers. -Failed company transfers remain resumable. Completed/cancelled transfers still -block while their registered local spool contains cleanup work. +Failed company transfers remain resumable. Completed transfers require their +recorded manifest, exact part receipts and matching counts. Completed/cancelled +transfers still block while their registered local spool contains cleanup work. Waiting agent conversations can be passive containers under both ordinary and scheduled inspection. The same narrow proof requires the exact conversation @@ -504,8 +505,9 @@ operation by itself. Multiple replicas require shared fencing. The protocol does not make existing in-memory-only debt survive an unplanned process crash. Related public PR #13413 adds runtime-service controller requirements and uses -the same `purpose: "idle"` / `ownerId` vocabulary. This report conservatively -blocks all retained runtime-service state instead of approving such services. +the same `purpose: "idle"` / `ownerId` vocabulary. Retained stopped local-process +rows use the domain and process-reconstruction proof described above; live, +uncertain and adapter-managed runtime resources remain blockers. ## Muse maintenance diff --git a/server/src/__tests__/idle-completed-history.test.ts b/server/src/__tests__/idle-completed-history.test.ts index 5c3bc152f52..11c45896ef9 100644 --- a/server/src/__tests__/idle-completed-history.test.ts +++ b/server/src/__tests__/idle-completed-history.test.ts @@ -1,4 +1,4 @@ -import { randomBytes, randomUUID } from "node:crypto"; +import { createHash, randomBytes, randomUUID } from "node:crypto"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; @@ -7,13 +7,16 @@ import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; import { eq, sql } from "drizzle-orm"; import * as tables from "@paperclipai/db"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; -import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { completedHistoryChecks, readIdleSleepSafety } from "../services/idle-sleep-safety.js"; import { readIdleIssueSchedule } from "../services/idle-issue-schedule.js"; import { companyTransferRunService } from "../services/company-transfer-runs.js"; import { inspectIdleSpool } from "../services/idle-local-work.js"; -describe("retained completed history", () => { +const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); +const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); + +(support.supported ? describe : describe.skip)("retained completed history", () => { let database: Awaited> | undefined; let db: ReturnType; const now = new Date(); @@ -28,9 +31,8 @@ describe("retained completed history", () => { return row.blocked; }; beforeAll(async () => { - const external = resolveExternalTestDatabaseUrl("PAPERCLIP_HISTORY_TEST_DATABASE_URL"); - if (!external) database = await startEmbeddedPostgresTestDatabase("idle-history-"); - db = tables.createDb(external ?? database!.connectionString, { maxConnections: 4 }); + if (!externalDatabaseUrl) database = await startEmbeddedPostgresTestDatabase("idle-history-"); + db = tables.createDb(externalDatabaseUrl ?? database!.connectionString, { maxConnections: 4 }); }, 90_000); afterEach(async () => { await db.transaction(async tx => { await tx.execute(sql`SET LOCAL client_min_messages = warning`); @@ -105,7 +107,7 @@ describe("retained completed history", () => { const router = Router(); router.use(issueRoutes(db, createStorageService(createLocalDiskStorageProvider(directory)))); const oauth = createPublicMcpOAuth(db, { origin, resource }), api = createMcpApiDispatch(router); - return { ...p, token, oauth, api, cleanup: () => rm(directory, { recursive: true, force: true }) }; + return { ...p, token, oauth, api, storage: createStorageService(createLocalDiskStorageProvider(directory)), cleanup: () => rm(directory, { recursive: true, force: true }) }; } it("checks all 32 families on the real migrated schema and keeps an empty instance eligible", async () => { @@ -119,7 +121,13 @@ describe("retained completed history", () => { const result = await companyTransferRunService.resumeOrCreate(db, { companyId: f.companyId, direction: "import", actorKey: "board:history", idempotencyKey: randomUUID(), containerRef: { kind: "upload" } }); expect(await blocked("company_transfer_runs")).toBe(true); - if (status === "completed") await companyTransferRunService.complete(db, result.run.id); + if (status === "completed") { + await companyTransferRunService.recordManifest(db, result.run.id, { chunks: [{ name: "tasks.jsonl" }], blobs: [] }, "a".repeat(64)); + await companyTransferRunService.start(db, result.run.id); + expect(await companyTransferRunService.completePart(db, result.run.id, "tasks.jsonl")).toBe(true); + expect(await companyTransferRunService.claimApply(db, result.run.id)).toBe(true); + await companyTransferRunService.complete(db, result.run.id); + } else await companyTransferRunService.cancel(db, result.run.id); expect(await blocked("company_transfer_runs")).toBe(false); const directory = await mkdtemp(path.join(tmpdir(), "history-spool-")); @@ -138,6 +146,56 @@ describe("retained completed history", () => { } finally { await rm(directory, { recursive: true, force: true }); } }); + it("uses the attachment owner to commit bytes and clean an abandoned upload", async () => { + const { createPublicMcpTransfers } = await import("../services/public-mcp/file-transfers.js"); + const { default: express } = await import("express"); + const { default: request } = await import("supertest"); + const f = await seed(), p = await connectedMcp(f); + try { + const transfers = createPublicMcpTransfers(db, p.oauth, p.api, p.storage); + const principal = await p.oauth.authenticate(p.token), bytes = Buffer.from("verified retained attachment"); + const args = { companyId: f.companyId, taskId: f.issueId, requestId: randomUUID(), filename: "result.txt", contentType: "text/plain", + byteSize: bytes.length, sha256: createHash("sha256").update(bytes).digest("hex") }; + const upload = await transfers.getUploadUrl(principal, args); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + const app = express(); app.use(transfers.router); + const url = new URL(String(upload.url)); + const result = await request(app).put(url.pathname + url.search).set("Content-Type", "text/plain").send(bytes); + expect(result.status).toBe(200); + expect(await blocked("mcp_attachment_uploads")).toBe(false); + const abandoned = await transfers.getUploadUrl(principal, { ...args, requestId: randomUUID() }); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + await db.update(tables.mcpAttachmentUploads).set({ expiresAt: new Date(0) }).where(eq(tables.mcpAttachmentUploads.id, String(abandoned.uploadId))); + await transfers.getDownloadUrl(principal, result.body.attachment.id); + expect(await blocked("mcp_attachment_uploads")).toBe(false); + await db.update(tables.mcpAttachmentUploads).set({ attachmentId: null }).where(eq(tables.mcpAttachmentUploads.id, String(upload.uploadId))); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + } finally { await p.cleanup(); } + }); + + it("retains the actual rejected tool review and only its acknowledged delivery", async () => { + const { commitToolActionReview } = await import("../services/tool-action-review.js"); + const { toolActionDeliveryService } = await import("../services/tool-action-delivery.js"); + const f = await seed(), requestId = randomUUID(), invocationId = randomUUID(), interactionId = randomUUID(); + await db.update(tables.issues).set({ status: "in_progress" }).where(eq(tables.issues.id, f.issueId)); + await db.insert(tables.toolInvocations).values({ id: invocationId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + actorType: "agent", actorId: f.agentId, agentId: f.agentId, toolName: "fixture:write", status: "awaiting_approval" }); + await db.insert(tables.issueThreadInteractions).values({ id: interactionId, companyId: f.companyId, issueId: f.issueId, + kind: "request_confirmation", payload: { version: 1, prompt: "Approve fixture action?", toolAction: { actionRequestId: requestId, invocationId } } }); + await db.insert(tables.toolActionRequests).values({ id: requestId, companyId: f.companyId, issueId: f.issueId, + invocationId, interactionId, requestedByAgentId: f.agentId, canonicalArgumentsHash: "a".repeat(64), canonicalArgumentsSummary: { summary: "fixture" } }); + expect(await blocked("tool_action_requests")).toBe(true); + await commitToolActionReview(db, { companyId: f.companyId, actionRequestId: requestId, decision: "rejected", actor: { userId: "local-board" } }); + expect(await blocked("tool_action_requests")).toBe(false); + expect(await blocked("tool_action_deliveries")).toBe(true); + await db.update(tables.issues).set({ status: "done" }).where(eq(tables.issues.id, f.issueId)); + await toolActionDeliveryService(db, { wakeup: async () => { throw new Error("Closed issue must retire its receipt"); } }).deliver(requestId); + expect(await blocked("tool_action_deliveries")).toBe(false); + await db.update(tables.toolInvocations).set({ completedAt: null }).where(eq(tables.toolInvocations.id, invocationId)); + expect(await blocked("tool_action_requests")).toBe(true); + expect(await blocked("tool_action_deliveries")).toBe(true); + }); + it("accepts MCP completion receipts and cleanup while preserving unknown writes and pending uploads", async () => { const f = await seed(), p = await mcp(f); const [receipt] = await db.insert(tables.mcpMutationReceipts).values({ ...p, companyId: f.companyId, @@ -220,6 +278,130 @@ describe("retained completed history", () => { expect(await blocked("mcp_event_deliveries")).toBe(true); }); + it("uses actual chat ingress, lifecycle application and operator publication receipts", async () => { + const { githubIngressFixture } = await import("./helpers/github-webhook-ingress.js"); + const { chatCredentialMutationLease } = await import("../services/chat-credential-mutation-lease.js"); + const { createChatResourceReconciliation } = await import("../services/chat-channels/resource-reconciliation.js"); + const { chatChannelService } = await import("../services/chat-channels.js"); + const { projectSafeChatPublication } = await import("../services/chat-publication-projection.js"); + const f = await githubIngressFixture(db), action = await f.insertAction(); + expect(await blocked("chat_actions")).toBe(true); + await f.service.processGitHubWebhookIngress(action.id); + expect(f.calls).toHaveLength(1); + expect(await blocked("chat_actions")).toBe(false); + const [issue] = await db.insert(tables.issues).values({ companyId: f.companyId, title: "Chat delivery", status: "done" }).returning(); + const [conversation] = await db.insert(tables.chatConversations).values({ companyId: f.companyId, endpointId: f.endpoint.id, + issueId: issue.id, externalConversationId: "fixture", externalThreadId: "fixture", externalLabel: "History", state: "completed" }).returning(); + const effect = { kind: "resource" as const, provider: "github" as const, providerEventId: randomUUID(), providerResourceId: "repository:fixture", + resourceType: "repository", label: "Fixture", availability: "removed" as const }; + const [delivery] = await db.insert(tables.chatDeliveries).values({ companyId: f.companyId, endpointId: f.endpoint.id, + providerEventId: `lifecycle:${effect.providerEventId}`, deduplicationKey: randomUUID(), eventKind: "uninstallation", state: "processing", attempts: 1, + normalizedEvent: { providerEventId: `lifecycle:${effect.providerEventId}`, kind: "uninstallation", lifecycle: effect } }).returning(); + expect(await blocked("chat_deliveries")).toBe(true); + const lease = chatCredentialMutationLease(db); + const resources = createChatResourceReconciliation(db, { endpointRecord: async () => ({ endpoint: f.endpoint }), + withCredentialMutationLease: lease, runtimeGeneration: () => 1, maxErrorText: 500 }); + await lease(f.endpoint, guard => resources.applyProviderAvailability({ currentEndpoint: f.endpoint, effect, + credentialLease: guard, staleResourceEffect: false, claimed: delivery, candidate: { attempts: 0 } })); + expect(await blocked("chat_deliveries")).toBe(false); + await db.update(tables.chatDeliveries).set({ normalizedEvent: {} }).where(eq(tables.chatDeliveries.id, delivery.id)); + expect(await blocked("chat_deliveries")).toBe(true); + const [publication] = await db.insert(tables.chatPublications).values({ companyId: f.companyId, endpointId: f.endpoint.id, conversationId: conversation.id, + issueId: issue.id, idempotencyKey: randomUUID(), state: "delivery_unknown", + payload: projectSafeChatPublication({ classification: "external", source: "explicit_board_send", text: "Retained result" }) }).returning(); + expect(await blocked("chat_publications")).toBe(true); + const service = chatChannelService(db, { heartbeat: { wakeup: async () => { throw new Error("No turn expected"); } }, + fetch: async () => { throw new Error("No external network expected"); } }); + try { + await service.resolvePublication(f.endpoint.id, publication.id, "mark_delivered", "local-board"); + expect(await blocked("chat_publications")).toBe(false); + await db.update(tables.chatPublications).set({ publishedAt: null }).where(eq(tables.chatPublications.id, publication.id)); + expect(await blocked("chat_publications")).toBe(true); + } finally { await service.shutdown(); } + }); + + it("retains a real acknowledged chat completion across later task versions", async () => { + const { issueService } = await import("../services/issues.js"); + const { prepareChatCompletionTurn } = await import("../services/chat-completion-delivery.js"); + const { instanceSettingsService } = await import("../services/instance-settings.js"); + const f = await seed(), service = issueService(db); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true }); + await db.update(tables.issues).set({ status: "in_progress", conversationAgentId: f.agentId, + conversationUserId: "local-board", conversationState: "waiting" }).where(eq(tables.issues.id, f.issueId)); + await db.update(tables.heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, conversationSessionGeneration: 0 } }).where(eq(tables.heartbeatRuns.id, f.runId)); + const task = await service.create(f.companyId, { title: "Actual handoff", status: "todo", createdByAgentId: f.agentId, actorRunId: f.runId }); + await service.update(task.id, { status: "done" }); + expect(await blocked("chat_completion_deliveries")).toBe(true); + const [turn] = await db.insert(tables.heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running", + contextSnapshot: { issueId: f.issueId, wakeReason: "chat_task_completed", conversationSessionGeneration: 0 } }).returning(); + await prepareChatCompletionTurn(db, turn); + const reply = await service.addComment(f.issueId, "The result is ready.", { agentId: f.agentId, runId: turn.id }, { completionReply: true }); + expect(await blocked("chat_completion_deliveries")).toBe(true); // actual reply run still owns work + await db.update(tables.heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(tables.heartbeatRuns.id, turn.id)); + expect(await blocked("chat_completion_deliveries")).toBe(false); + await service.update(task.id, { status: "todo" }); + expect(await blocked("chat_completion_deliveries")).toBe(false); // retained delivered version remains history + await db.update(tables.issueComments).set({ createdByRunId: f.runId }).where(eq(tables.issueComments.id, reply.id)); + expect(await blocked("chat_completion_deliveries")).toBe(true); + }); + + it("uses the real browser accounting and verified provider stop writers", async () => { + const { toolAccessService } = await import("../services/tool-access.js"); + const { browserUseService } = await import("../services/browser-use.js"); + const f = await seed(), userId = "browser-history-reviewer", providerRun = randomUUID(), providerSession = randomUUID(), providerBrowser = randomUUID(); + const oldKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + process.env.PAPERCLIP_SECRETS_MASTER_KEY = "a".repeat(64); + let providerStatus = "running", stopped = false, stopCalls = 0; + const request = async (url: string, init: RequestInit = {}) => { + const route = new URL(url).pathname.replace("/api/v4", ""); + if (route === "/profiles") return Response.json({ items: [], totalItems: 0 }); + if (route === "/runs" && init.method === "POST") return Response.json({ id: providerRun, sessionId: providerSession, status: providerStatus, model: "fixture" }); + if (route.endsWith("/status")) return Response.json({ status: providerStatus }); + if (route.endsWith("/events")) return Response.json({ events: [], nextAfter: 0, hasMore: false }); + if (route.endsWith("/cancel")) { providerStatus = "cancelled"; return Response.json({}); } + if (route === `/runs/${providerRun}`) return Response.json({ status: providerStatus, totalCostUsd: "0.15", model: "fixture", result: "Finished" }); + if (route === "/browsers") return Response.json({ items: [{ id: providerBrowser, status: stopped ? "stopped" : "active" }], totalItems: 1 }); + if (route === `/browsers/${providerBrowser}`) { + if (init.method === "PATCH") { stopped = true; stopCalls++; } + return Response.json({ id: providerBrowser, agentSessionId: providerSession, status: stopped ? "stopped" : "active" }); + } + throw new Error(`Unexpected browser provider route ${route}`); + }; + try { + await db.insert(tables.companyMemberships).values({ companyId: f.companyId, principalType: "user", principalId: userId, membershipRole: "admin", status: "active" }); + const access = toolAccessService(db, { remoteHttpRequest: request, remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }] }); + const actor = { actorType: "user" as const, actorId: userId }; + const connected = await access.connectGalleryApp(f.companyId, { galleryKey: "browser-use-cloud", connectionMethodKey: "cloud-v4", + credentialValues: { "credentials.apiKey": "bu_fixture_secret" } }, actor); + await access.finishGalleryAppConnection(f.companyId, connected.connectionId, { enabledCatalogEntryIds: connected.catalog.map(entry => entry.id), + askFirstCatalogEntryIds: [], access: { agentIds: [f.agentId] } }, actor); + const [grant] = await db.select().from(tables.connectionGrants).where(eq(tables.connectionGrants.connectionId, connected.connectionId)); + await db.update(tables.heartbeatRuns).set({ status: "running", finishedAt: null, responsibleUserId: userId }).where(eq(tables.heartbeatRuns.id, f.runId)); + await db.update(tables.issues).set({ status: "in_progress" }).where(eq(tables.issues.id, f.issueId)); + const service = browserUseService(db, request); + const started = await service.execute({ companyId: f.companyId, agentId: f.agentId, issueId: f.issueId, runId: f.runId }, grant, + randomUUID(), "browser_start", { task: "Read the retained fixture", maxCostUsd: 1 }) as { sessionId: string }; + for (const reason of ["browser_use_sessions", "browser_use_runs"]) expect(await blocked(reason)).toBe(true); + providerStatus = "completed"; + await db.update(tables.browserUseSessions).set({ nextPollAt: new Date(0) }).where(eq(tables.browserUseSessions.id, started.sessionId)); + await service.sweep(connected.connectionId); + expect(await blocked("browser_use_browsers")).toBe(true); + const [session] = await db.select().from(tables.browserUseSessions).where(eq(tables.browserUseSessions.id, started.sessionId)); + await service.control(session, "end"); + await db.update(tables.browserUseSessions).set({ nextPollAt: new Date(0) }).where(eq(tables.browserUseSessions.id, started.sessionId)); + await service.sweep(connected.connectionId); + expect(stopCalls).toBeGreaterThan(0); + await db.update(tables.heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(tables.heartbeatRuns.id, f.runId)); + for (const reason of ["browser_use_sessions", "browser_use_runs", "browser_use_browsers"]) expect(await blocked(reason), reason).toBe(false); + const [run] = await db.select().from(tables.browserUseRuns).where(eq(tables.browserUseRuns.sessionId, started.sessionId)); + expect(run.result?.paperclipBudgetHold).toMatchObject({ state: "settled", accountedCentsExact: "15.0000000" }); + await db.update(tables.browserUseRuns).set({ eventsDrained: 0 }).where(eq(tables.browserUseRuns.id, run.id)); + expect(await blocked("browser_use_runs")).toBe(true); + } finally { + if (oldKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = oldKey; + } + }); + it("keeps browser accounting, run ownership and provider resource cleanup independent", async () => { const f = await seed(), c = await connection(f), invocationId = randomUUID(); const [session] = await db.insert(tables.browserUseSessions).values({ ...f, ...c, status: "closed", providerSessionId: randomUUID() }).returning(); @@ -272,6 +454,48 @@ describe("retained completed history", () => { expect(await read()).toEqual({ version: 1, backgroundWork: "present" }); }, 60_000); + it("uses actual secret and binding approval receipts, including ciphertext scrubbing", async () => { + const { createSecretProposalsService } = await import("../services/secret-proposals.js"); + const f = await seed(), p = await mcp(f), oldKey = process.env.PAPERCLIP_SECRETS_MASTER_KEY; + process.env.PAPERCLIP_SECRETS_MASTER_KEY = "a".repeat(64); + try { + await db.update(tables.heartbeatRuns).set({ status: "running", finishedAt: null }).where(eq(tables.heartbeatRuns.id, f.runId)); + await db.update(tables.issues).set({ status: "in_progress", executionRunId: f.runId }).where(eq(tables.issues.id, f.issueId)); + const service = createSecretProposalsService(db); + const context = { companyId: f.companyId, heartbeatRunId: f.runId, registerForRedaction: () => {} }; + const secret = await service.createSecret(context, { name: "history/token", key: "HISTORY_TOKEN", value: "synthetic-history-secret", justification: "Retained completed approval" }); + expect(await blocked("company_secret_proposals")).toBe(true); + const approved = await service.approve(f.companyId, secret.id, { resolvedByUserId: p.userId }); + const binding = await service.createBinding(context, { secretId: approved.createdSecretId!, configPath: "env.HISTORY_TOKEN", + bindingTargetPolicy: "self_and_reports", justification: "Use the approved credential" }); + await service.approve(f.companyId, binding.id, { resolvedByUserId: p.userId }); + await db.update(tables.heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(tables.heartbeatRuns.id, f.runId)); + expect(await blocked("company_secret_proposals")).toBe(false); + await db.update(tables.companySecretProposals).set({ appliedBindingConfigPath: null }).where(eq(tables.companySecretProposals.id, binding.id)); + expect(await blocked("company_secret_proposals")).toBe(true); + } finally { if (oldKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; else process.env.PAPERCLIP_SECRETS_MASTER_KEY = oldKey; } + }); + + it("uses actual retention archive notification dispatch and durable acknowledgment", async () => { + const { decisionRetentionService } = await import("../services/decision-retention.js"); + const { decisionArchiveDeliveryService } = await import("../services/decision-archive-delivery.js"); + const f = await seed(), activityAt = new Date("2026-01-01T00:00:00.000Z"); + await db.update(tables.agents).set({ status: "active" }).where(eq(tables.agents.id, f.agentId)); + await db.update(tables.issues).set({ createdByAgentId: f.agentId }).where(eq(tables.issues.id, f.issueId)); + await db.insert(tables.decisionRetention).values({ companyId: f.companyId, sourceKind: "review", sourceId: f.issueId, sourceActivityAt: activityAt, version: 1 }); + expect(await decisionRetentionService(db).autoArchive({ companyId: f.companyId, items: [{ sourceKind: "review", subject: { id: f.issueId }, + activityAt: activityAt.toISOString(), retentionVersion: 1, keep: false, archivedAt: null }], now })).toBe(1); + expect(await blocked("decision_archive_notification_outbox")).toBe(true); + let notifications = 0; + expect(await decisionArchiveDeliveryService(db, { notifyOriginAgent: async batch => { + expect(batch).toMatchObject({ companyId: f.companyId, agentId: f.agentId, items: [{ sourceId: f.issueId }] }); notifications++; + } }).deliver()).toMatchObject({ delivered: 1 }); + expect(notifications).toBe(1); + expect(await blocked("decision_archive_notification_outbox")).toBe(false); + await db.update(tables.decisionArchiveNotificationOutbox).set({ deliveredAt: null }); + expect(await blocked("decision_archive_notification_outbox")).toBe(true); + }); + it.each(["rejected", "withdrawn", "expired"] as const)("accepts actual %s secret-proposal scrubbing and refuses retained ciphertext", async status => { const { createSecretProposalsService } = await import("../services/secret-proposals.js"); const f = await seed(); @@ -356,6 +580,31 @@ describe("retained completed history", () => { } }, 60_000); + it.each(["missing_decider", "missing_option"] as const)("requires the actual %s recovery stop receipt after partial effects", async reason => { + const previous = process.env.PAPERCLIP_DECISION_SIGNING_SECRET; + process.env.PAPERCLIP_DECISION_SIGNING_SECRET = "0123456789abcdef0123456789abcdef"; + try { + const f = await decisionFixture(); + const decision = await f.service.create({ companyId: f.companyId, actor: f.actor, agentId: f.agentId, runId: f.runId, + title: "Recover interrupted effect ownership", body: reason, continuationPolicy: "none", options: [{ id: "yes", label: "Yes", + effects: [{ type: "comment_on_issue", targetIssueId: f.issueId, staleness: "lenient", bodyMarkdown: "Already persisted effect" }] }] }); + await f.service.decide({ id: decision.id, optionId: "yes", decidedByUserId: f.userId, userActor: f.userActor }); + // Model a stranded claim whose durable option/decider was lost after an + // already-written effect. Recovery must record its own final stop. + await db.update(tables.decisions).set({ executionStatus: "running", updatedAt: new Date(Date.now() - 120_000), + ...(reason === "missing_decider" ? { decidedByUserId: null } : { chosenOptionId: "missing" }) }).where(eq(tables.decisions.id, decision.id)); + expect(await blocked("decisions")).toBe(true); + expect(await f.service.recoverEffects()).toEqual({ resumed: 1 }); + await f.finish(); + expect(await f.service.get(decision.id)).toMatchObject({ executionStatus: "partial", metadata: { effectRecoveryError: reason } }); + expect(await blocked("decisions")).toBe(false); + await db.delete(tables.activityLog).where(sql`${tables.activityLog.entityId} = ${decision.id} AND ${tables.activityLog.action} = 'decision.recovery_failed'`); + expect(await blocked("decisions")).toBe(true); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_DECISION_SIGNING_SECRET; else process.env.PAPERCLIP_DECISION_SIGNING_SECRET = previous; + } + }); + it.each(["archive", "keep", "manifest_failure"] as const)("checks the separate reviewed retention manifest for %s", async branch => { const { hashAttentionArchiveManifest } = await import("../services/decision-retention.js"); const previous = process.env.PAPERCLIP_DECISION_SIGNING_SECRET; @@ -424,7 +673,8 @@ describe("retained completed history", () => { // Named external runner boundary: persist the real broker's receipt for // the exact command it emitted, without invoking paid native execution. await port.settle({ sourceEventId: `result_${command.requestId}`, payload: { binding: ref, requestId: command.requestId, - outcome: { status: command.action === "accept" ? "accepted" : "completed" } } }); + outcome: command.action === "consume_input" ? { status: "consumed", ...command.input } + : { status: command.action === "accept" ? "accepted" : "completed" } } }); }); try { await port.dispatch({ sourceEventId: `offer_${randomUUID()}`, payload: { kind: "assignment", binding: ref, tools: [], @@ -432,18 +682,95 @@ describe("retained completed history", () => { const [assignment] = await db.select().from(tables.museRunnerAssignments).where(eq(tables.museRunnerAssignments.runId, f.runId)); expect(await blocked("muse_runner_assignments")).toBe(true); await broker.act(subject, { version: 1, command: "accept", assignmentId: assignment.id, requestId: randomUUID() }); + const { digestPaperclipSemanticContent } = await import("../vendor/paperclip-runner/index.js"); + const nativeRequestId = "history_question", response = { schema: "paperclip.question_response.v1", answers: [] }; + const inputDigest = digestPaperclipSemanticContent({ requestId: nativeRequestId, turnId, response }); + await port.inputAvailable({ sourceEventId: randomUUID(), payload: { binding: ref, requestId: nativeRequestId, turnId, inputDigest, response } }); + expect(await blocked("muse_input_deliveries")).toBe(true); + await broker.act(subject, { version: 1, command: "consume_input", assignmentId: assignment.id, requestId: randomUUID(), + nativeRequestId, inputDigest, continuationReceiptId: randomUUID(), continuationPersisted: true }); await broker.act(subject, { version: 1, command: "finish", assignmentId: assignment.id, requestId: randomUUID(), result: {} }); expect(await blocked("muse_runner_assignments")).toBe(true); const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: () => {} } as unknown as import("../services/heartbeat/run-lifecycle.js").HeartbeatLifecycleDependencies); await lifecycle.setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); await broker.reconcileTerminalAssignments(f.companyId, f.agentId, binding.id); - for (const family of ["muse_runner_assignments", "muse_runner_operations", "muse_mailbox_items", "external_agent_holds"]) { + for (const family of ["muse_runner_assignments", "muse_runner_operations", "muse_mailbox_items", "muse_input_deliveries", "external_agent_holds"]) { expect(await blocked(family), family).toBe(false); } await db.update(tables.museRunnerOperations).set({ status: "unknown" }).where(eq(tables.museRunnerOperations.assignmentId, assignment.id)); expect(await blocked("muse_runner_assignments")).toBe(true); expect(await blocked("muse_runner_operations")).toBe(true); + await db.update(tables.museRunnerOperations).set({ status: "settled" }).where(eq(tables.museRunnerOperations.assignmentId, assignment.id)); + expect(await blocked("muse_agent_bindings")).toBe(true); + expect(await blocked("muse_credentials")).toBe(true); + await broker.revoke(f.companyId, f.agentId, p.userId); + expect(await blocked("muse_agent_bindings")).toBe(false); + expect(await blocked("muse_credentials")).toBe(true); // cleanup authority still lives for its owner TTL + await db.update(tables.museCredentials).set({ expiresAt: new Date(0) }).where(eq(tables.museCredentials.bindingId, binding.id)); + expect(await blocked("muse_credentials")).toBe(false); + } finally { await detach(); } + }, 60_000); + + it("settles actual Dot runner assignments, operation receipts and mailbox history", async () => { + const { dotRunnerBroker } = await import("../services/dot-runner-broker.js"); + const { instanceSettingsService } = await import("../services/instance-settings.js"); + const { createHeartbeatLifecycle } = await import("../services/heartbeat/run-lifecycle.js"); + const f = await seed(), p = await mcp(f, true), sessionId = randomUUID(), turnId = `turn_${randomUUID()}`; + await instanceSettingsService(db).updateExperimental({ enablePublicMcp: true, enableOpenAiDot: true }); + await db.insert(tables.companyMemberships).values({ companyId: f.companyId, principalType: "user", principalId: p.userId, membershipRole: "owner", status: "active" }); + await db.update(tables.mcpOauthGrants).set({ revokedAt: null }).where(eq(tables.mcpOauthGrants.id, p.grantId)); + const [grant] = await db.select().from(tables.mcpOauthGrants).where(eq(tables.mcpOauthGrants.id, p.grantId)); + const [binding] = await db.insert(tables.dotAgentBindings).values({ companyId: f.companyId, agentId: f.agentId, operatorId: p.userId, + grantId: p.grantId, providerKind: "openai_dot", status: "connected" }).returning(); + await db.update(tables.agents).set({ status: "active", adapterType: "paperclip_runner", + adapterConfig: { provider: "openai_dot", dotBindingId: binding.id, allowUnmeteredProvider: true } }).where(eq(tables.agents.id, f.agentId)); + await db.update(tables.issues).set({ status: "in_progress", responsibleUserId: p.userId, checkoutRunId: f.runId, executionRunId: f.runId }).where(eq(tables.issues.id, f.issueId)); + await db.update(tables.heartbeatRuns).set({ status: "running", runtimeMode: "native", driverKind: "openai_dot_mcp", finishedAt: null, + nativeIssueId: f.issueId, nativeSessionId: sessionId, nativePhase: "turn_running", responsibleUserId: p.userId }).where(eq(tables.heartbeatRuns.id, f.runId)); + await db.insert(tables.nativeRunFinalizations).values({ runId: f.runId, companyId: f.companyId, issueId: f.issueId, + phase: "running", controllerGeneration: 1, leaseExpiresAt: new Date(Date.now() + 600_000) }); + const broker = dotRunnerBroker(db, { heartbeat: { wakeup: async () => null, cancelRun: async () => null } }); + const { createPublicMcpOAuth } = await import("../services/public-mcp/oauth.js"); + const { createPublicMcpEvents } = await import("../services/public-mcp/events.js"); + const principal = { grant, actor: { type: "agent" as const, agentId: f.agentId, companyId: f.companyId }, + company: { id: f.companyId, name: "History", issuePrefix: "H", status: "active" } }; + const events = createPublicMcpEvents(db, createPublicMcpOAuth(db, { origin: "https://example.test", resource: "https://example.test/mcp" }, { mode: "openai_dot_agent" }), + async () => { throw new Error("No personal API expected"); }, { enableDotRunner: true, fetch: async (_url, init) => { + const body = JSON.parse(String(init.body)); + return body.type === "verification" ? Response.json({ challenge: body.challenge }) : new Response(null, { status: 204 }); + } }); + await events.subscribe(principal, { name: "paperclip.dot.mailbox_updated", arguments: { companyId: f.companyId, bindingId: binding.id }, + delivery: { mode: "webhook", url: "https://receiver.example/dot", secret: `whsec_${randomBytes(32).toString("base64")}` } }); + await events.tick(); + const challenge = (await broker.mailbox(principal)).items.find(item => item.kind === "readiness_challenge")!; + await broker.confirmChallenge(principal, String(challenge.references.challenge)); + const execution = { binding: { companyId: f.companyId, agentId: f.agentId, runId: f.runId }, + provider: { kind: "openai_dot" as const, binding: await broker.snapshot(f.companyId, f.agentId, binding.id) } }; + const ref = { ...execution.provider.binding, runId: f.runId, normalizedSessionId: sessionId, turnId, assignmentRevision: 1 }; + const port = broker.port(execution); + const detach = await port.attach(async command => { + await port.settle({ sourceEventId: randomUUID(), payload: { binding: ref, requestId: command.requestId, + outcome: { status: command.action === "accept" ? "accepted" : "completed" } } }); + }); + try { + await port.dispatch({ sourceEventId: randomUUID(), payload: { kind: "assignment", binding: ref, tools: [], + acceptByUnixMs: Date.now() + 600_000, expiresAtUnixMs: Date.now() + 3_600_000 } }); + const [assignment] = await db.select().from(tables.dotRunnerAssignments).where(eq(tables.dotRunnerAssignments.runId, f.runId)); + expect(await blocked("dot_runner_assignments")).toBe(true); + expect(await blocked("dot_mailbox_items")).toBe(true); + await broker.operation(principal, assignment.id, randomUUID(), "accept", {}); + await broker.operation(principal, assignment.id, randomUUID(), "finish", { result: {} }); + const lifecycle = createHeartbeatLifecycle(db, { emitTerminalAgentTaskRun: () => {} } as unknown as + import("../services/heartbeat/run-lifecycle.js").HeartbeatLifecycleDependencies); + await lifecycle.setRunStatus(f.runId, "succeeded", { finishedAt: new Date() }); + for (const reason of ["dot_runner_assignments", "dot_runner_operations"]) expect(await blocked(reason), reason).toBe(false); + // The independent readiness mailbox is retained until its binding is revoked. + await broker.revoke(f.companyId, f.agentId, p.userId); + expect(await blocked("dot_mailbox_items")).toBe(false); + await db.update(tables.dotRunnerOperations).set({ status: "dispatched", outcome: null }).where(eq(tables.dotRunnerOperations.assignmentId, assignment.id)); + expect(await blocked("dot_runner_assignments")).toBe(true); + expect(await blocked("dot_runner_operations")).toBe(true); } finally { await detach(); } }, 60_000); @@ -539,8 +866,41 @@ describe("retained completed history", () => { const adopted = await runtime.reconcilePersistedRuntimeServicesOnStartup(db); expect(adopted.adopted).toBe(1); expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("present"); - await runtime.stopRuntimeServicesForExecutionWorkspace({ db, executionWorkspaceId: workspace.id, workspaceCwd: directory }); + let enterCleanup!: () => void, releaseCleanup!: () => void; + const cleanupEntered = new Promise(resolve => { enterCleanup = resolve; }); + const cleanupGate = new Promise(resolve => { releaseCleanup = resolve; }); + const failedControl = Object.create(db) as typeof db; + failedControl.select = (() => ({ from: () => ({ where: async () => { + enterCleanup(); await cleanupGate; throw new Error("controlled stop cleanup failure"); + } }) })) as typeof db.select; + const stopping = runtime.stopRuntimeServicesForExecutionWorkspace({ db: failedControl, + executionWorkspaceId: workspace.id, workspaceCwd: directory }); + await cleanupEntered; + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("present"); + let releaseReconstruction!: () => void; + const reconstructionGate = new Promise(resolve => { releaseReconstruction = resolve; }); + const staleReconstruction = Object.create(db) as typeof db; + let firstSelect = true; + staleReconstruction.select = ((...fields: unknown[]) => { + const query = Reflect.apply(db.select, db, fields); + if (!firstSelect) return query; + firstSelect = false; + return { from: (...from: unknown[]) => { + const selected = Reflect.apply(query.from, query, from); + return { where: (...where: unknown[]) => reconstructionGate.then(() => Reflect.apply(selected.where, selected, where)) }; + } }; + }) as typeof db.select; + const earlierReconstruction = runtime.reconcilePersistedRuntimeServicesOnStartup(staleReconstruction); + releaseCleanup(); + await expect(stopping).rejects.toThrow("controlled stop cleanup failure"); + releaseReconstruction(); + await earlierReconstruction; + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); // older successful flight cannot erase the failed control epoch + await expect(fetch(`http://127.0.0.1:${port}`, { signal: AbortSignal.timeout(1000) })).rejects.toThrow(); ownedProcess = null; + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); + await runtime.reconcilePersistedRuntimeServicesOnStartup(db); + await runtime.stopRuntimeServicesForExecutionWorkspace({ db, executionWorkspaceId: workspace.id, workspaceCwd: directory }); expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("none"); expect(await blocked("workspace_runtime_services")).toBe(false); const unavailable = Object.create(db) as typeof db; diff --git a/server/src/services/chat-channels/idle-history-proof.ts b/server/src/services/chat-channels/idle-history-proof.ts index 778779d2b3d..8f9642b3ac5 100644 --- a/server/src/services/chat-channels/idle-history-proof.ts +++ b/server/src/services/chat-channels/idle-history-proof.ts @@ -44,6 +44,15 @@ function completedAction(a: string): string { export const idleChatHistoryChecks = [ { reason: "chat_deliveries", query: `SELECT 1 FROM chat_deliveries d WHERE ( EXISTS (SELECT 1 FROM chat_endpoints e WHERE e.id = d.endpoint_id AND e.company_id = d.company_id) + AND jsonb_typeof(d.normalized_event) = 'object' + AND d.normalized_event->>'providerEventId' = d.provider_event_id + AND d.normalized_event->>'kind' = d.event_kind + AND d.event_kind IN ('mention', 'message', 'direct_message', 'message_updated', 'message_deleted', 'message_restored', + 'reaction_added', 'reaction_removed', 'action', 'modal_submitted', 'modal_closed', 'slash_command', 'file_shared', 'installation', 'uninstallation', 'unknown') + AND (d.state <> 'processed' OR d.event_kind NOT IN ('mention', 'message', 'direct_message') + OR EXISTS (SELECT 1 FROM chat_actions consumed WHERE consumed.delivery_id = d.id + AND consumed.company_id = d.company_id AND consumed.endpoint_id = d.endpoint_id + AND consumed.kind = 'inbound_wakeup' AND (${completedAction('consumed')}))) AND d.next_attempt_at IS NULL AND (d.state = 'failed' OR (d.state IN ('processed', 'filtered') AND d.processed_at IS NOT NULL)) AND (d.conversation_id IS NULL OR EXISTS (SELECT 1 FROM chat_conversations c WHERE c.id = d.conversation_id AND c.company_id = d.company_id AND c.endpoint_id = d.endpoint_id)) @@ -68,12 +77,14 @@ export const idleChatHistoryChecks = [ JOIN issues source ON source.id = h.conversation_id AND source.company_id = h.company_id WHERE h.task_id = d.task_id AND h.company_id = d.company_id AND source.conversation_agent_id = h.agent_id - AND (d.status <> 'delivered' OR (d.status_version = task.status_version AND task.status = 'done' + AND (d.status <> 'delivered' OR (d.status_version <= task.status_version AND EXISTS (SELECT 1 FROM issue_comments c WHERE c.id = d.response_comment_id - AND c.company_id = d.company_id AND c.issue_id = h.conversation_id AND c.author_agent_id = h.agent_id)))) + AND c.company_id = d.company_id AND c.issue_id = h.conversation_id AND c.author_agent_id = h.agent_id + AND c.created_by_run_id = d.target_run_id)))) AND (d.target_run_id IS NULL AND d.status <> 'delivered' OR EXISTS ( SELECT 1 FROM heartbeat_runs r JOIN chat_task_handoffs h ON h.task_id = d.task_id WHERE r.id = d.target_run_id AND r.company_id = d.company_id AND r.agent_id = h.agent_id + AND h.company_id = d.company_id AND r.context_snapshot->>'conversationSessionGeneration' = h.session_generation::text AND (r.native_issue_id = h.conversation_id OR r.context_snapshot->>'issueId' = h.conversation_id::text) AND ${finishedIdleRun('r')}))) IS NOT TRUE` }, ] as const; diff --git a/server/src/services/workspace-history-idle-proof.ts b/server/src/services/workspace-history-idle-proof.ts index 534968b4ca3..d6d3f3ce428 100644 --- a/server/src/services/workspace-history-idle-proof.ts +++ b/server/src/services/workspace-history-idle-proof.ts @@ -7,6 +7,21 @@ export const idleWorkspaceHistoryChecks = [ t.status IN ('completed', 'cancelled') AND t.finished_at IS NOT NULL AND t.direction IN ('import', 'export') AND jsonb_typeof(t.container_ref) = 'object' AND jsonb_typeof(t.completed_parts) = 'array' + AND (t.status = 'cancelled' OR (t.error IS NULL AND t.manifest_sha256 ~ '^[a-f0-9]{64}$' + AND CASE WHEN jsonb_typeof(t.manifest->'chunks') = 'array' AND jsonb_typeof(t.manifest->'blobs') = 'array' THEN + t.chunk_count = jsonb_array_length(t.manifest->'chunks') AND t.blob_count = jsonb_array_length(t.manifest->'blobs') + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements((t.manifest->'chunks') || (t.manifest->'blobs')) part + WHERE jsonb_typeof(part->'name') IS DISTINCT FROM 'string' OR NOT t.completed_parts ? (part->>'name')) + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(t.completed_parts) done + WHERE NOT EXISTS (SELECT 1 FROM jsonb_array_elements((t.manifest->'chunks') || (t.manifest->'blobs')) part WHERE part->>'name' = done)) + WHEN jsonb_typeof(t.manifest->'parts') = 'array' THEN + t.chunk_count = jsonb_array_length(t.manifest->'parts') AND t.blob_count = 0 + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(t.manifest->'parts') WITH ORDINALITY part(value, ordinal) + WHERE NOT t.completed_parts ? ('part-' || (ordinal - 1)::text)) + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements_text(t.completed_parts) done + WHERE NOT EXISTS (SELECT 1 FROM jsonb_array_elements(t.manifest->'parts') WITH ORDINALITY part(value, ordinal) + WHERE done = 'part-' || (ordinal - 1)::text)) + ELSE false END)) AND (t.company_id IS NOT NULL OR (t.direction = 'import' AND t.status = 'cancelled')) AND (t.company_id IS NULL OR EXISTS (SELECT 1 FROM companies c WHERE c.id = t.company_id))) IS NOT TRUE` }, { reason: "workspace_runtime_services", query: `SELECT 1 FROM workspace_runtime_services s WHERE ( diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index f2b1d46ba0e..8a6211968a6 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -285,6 +285,7 @@ const quarantinedRuntimeExposurePorts = new Set(); */ const exposurePortPairClaims = new ExposurePortPairClaims(); let runtimeReconstructionComplete = false; +let runtimeStopsInFlight = 0; let runtimeReconstructionEpoch = 0; let runtimeReconciliation: ReturnType | null = null; @@ -292,7 +293,7 @@ let runtimeReconciliation: ReturnType | null * registry process, an allocation being published, or unfinished exposure cleanup. */ export function readWorkspaceRuntimeIdleWork(): IdleLocalWork { if (!runtimeReconstructionComplete || runtimeReconciliation !== null) return "unknown"; - return runtimeServicesById.size || runtimeProvisionByWorkspace.size || runtimeControlStartByOwner.size + return runtimeStopsInFlight || runtimeServicesById.size || runtimeProvisionByWorkspace.size || runtimeControlStartByOwner.size || runtimeReplacementClaimsByReuseKey.size || quarantinedRuntimeExposurePorts.size || exposurePortPairClaims.activePorts().size || [...inFlightAllocatedPorts.values()].some(until => until > Date.now()) ? "present" : "none"; @@ -7174,7 +7175,21 @@ async function cleanupRecordExposure( await persistRuntimeServiceRecord(record.db, record).catch(() => undefined); } +async function withRuntimeStopWork(stop: () => Promise): Promise { + runtimeStopsInFlight += 1; + try { return await stop(); } + catch (error) { + runtimeReconstructionComplete = false; + runtimeReconstructionEpoch += 1; + throw error; + } finally { runtimeStopsInFlight -= 1; } +} + async function stopRuntimeService(serviceId: string) { + return withRuntimeStopWork(() => stopRuntimeServiceRecord(serviceId)); +} + +async function stopRuntimeServiceRecord(serviceId: string) { const record = runtimeServicesById.get(serviceId); if (!record) return; clearIdleTimer(record); @@ -8315,7 +8330,11 @@ export async function releaseRuntimeServicesForRun(runId: string) { } } -export async function stopRuntimeServicesForExecutionWorkspace(input: { +export async function stopRuntimeServicesForExecutionWorkspace(input: Parameters[0]) { + return withRuntimeStopWork(() => stopRuntimeServicesForExecutionWorkspaceOwned(input)); +} + +async function stopRuntimeServicesForExecutionWorkspaceOwned(input: { db?: Db; executionWorkspaceId: string; workspaceCwd?: string | null; @@ -8370,7 +8389,11 @@ export async function stopRuntimeServicesForExecutionWorkspace(input: { } } -export async function stopRuntimeServicesForProjectWorkspace(input: { +export async function stopRuntimeServicesForProjectWorkspace(input: Parameters[0]) { + return withRuntimeStopWork(() => stopRuntimeServicesForProjectWorkspaceOwned(input)); +} + +async function stopRuntimeServicesForProjectWorkspaceOwned(input: { db?: Db; projectWorkspaceId: string; runtimeServiceId?: string | null;