diff --git a/doc/idle-sleep-safety.md b/doc/idle-sleep-safety.md index 12f8dada5c4..1caf1cbceca 100644 --- a/doc/idle-sleep-safety.md +++ b/doc/idle-sleep-safety.md @@ -81,6 +81,56 @@ Independent SQL writers and additional processes need an explicit wake protocol; process-local notifications do not provide cross-replica delivery. Worker quiet alone is not evidence that the host or database actually suspended. +## Retained completion history + +The retained chat, tool-review, browser, decision, transfer, public-MCP and +external-agent families use their owning domain's completion receipts. Inspection +does not delete history, expire credentials, cancel work or shorten retention. +A terminal label alone is insufficient. Missing, contradictory, malformed or +foreign company/task/run/binding references keep the instance awake. + +Chat input can remain work after processing because its exact wake still owns +the conversation head. Publications retain provider ambiguity, endpoint leases +and file-transfer obligations. Tool reviews retain invocation completion, +interaction projection and continuation delivery. Browser runs require drained +events and settled/released accounting; a stopped browser must belong to the +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. + +Waiting agent conversations can be passive containers under both ordinary and +scheduled inspection. The same narrow proof requires the exact conversation +agent/company/user, waiting state, no checkout, active run, accounting, monitor +or pending delivery, and a finished wake for every accepted user-comment outbox +entry. An idle presentation state does not exempt an ordinary runnable task. + +Workspace runtime inspection also reads the owner's process-local resource +state. Startup reconstruction must succeed before that probe becomes quiet. +Adopted/live services, provisioning/control/replacement work, live allocation +claims and quarantined exposure ports remain barriers alongside durable rows. +Reset, failed reconstruction and unresolved desired-state restart cannot reuse an +earlier quiet proof. The probe performs no SQL, network request or cleanup and +adds no timer. A stopped row cannot hide a surviving registered process. + +MCP completion receipts do not exempt their downstream task, subscription or +upload cleanup. Active grants/subscriptions and Dot/Muse bindings/credentials +retain their transport/deadline blockers. Revoked or expired credential history +does not grant authority on the next request. A fenced external assignment is +not worker termination proof, and Muse transport notification is not input +consumption. Released Muse holds require their actual settlement/attestation +path. Dot stop holds remain unsupported because that owner has no completed +worker-stop receipt. Grok Bot shares Dot tables but has independent routine and +dispatch/recovery obligations; even revoked Grok bindings retain their blocker. + +Action/provider variants without an established receipt proof remain blocked. +This includes Photon notice lineage and unqualified legacy action results, +adapter-managed runtime resources, and active persistent transports. These +limits are not evidence of deployed wake support. Existing hosting admission, +generation, accounting, finalization and resource-cleanup checks still apply. + ## Native reply recovery Committed native replies use the shared delivery coordinator and its persisted diff --git a/server/src/__tests__/idle-completed-history.test.ts b/server/src/__tests__/idle-completed-history.test.ts new file mode 100644 index 00000000000..5c3bc152f52 --- /dev/null +++ b/server/src/__tests__/idle-completed-history.test.ts @@ -0,0 +1,563 @@ +import { randomBytes, randomUUID } from "node:crypto"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import net from "node:net"; +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 { 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", () => { + let database: Awaited> | undefined; + let db: ReturnType; + const now = new Date(); + const ownerId = randomUUID(); + const hold = () => ({ ownerId, draining: true, startedAt: new Date(now.getTime() - 1000), + expiresAt: new Date(Date.now() + 60_000), activeRuns: 0, pendingWakes: 0 }); + const read = () => readIdleSleepSafety(db, hold, Date.now, ownerId, async () => "none"); + const blocked = async (reason: string) => { + const check = completedHistoryChecks.find(check => check.reason === reason); + if (!check) throw new Error(`Missing family ${reason}`); + const [row] = await db.execute<{ blocked: boolean }>(sql`SELECT EXISTS (${sql.raw(check.query)}) AS blocked`); + 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 }); + }, 90_000); + afterEach(async () => { await db.transaction(async tx => { + await tx.execute(sql`SET LOCAL client_min_messages = warning`); + await tx.execute(sql`TRUNCATE companies CASCADE`); + }); }); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + + async function seed() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + await db.insert(tables.companies).values({ id: companyId, name: "Retained history", issuePrefix: randomUUID().slice(0, 8) }); + await db.insert(tables.agents).values({ id: agentId, companyId, name: "Idle agent", status: "idle", adapterType: "process" }); + await db.insert(tables.issues).values({ id: issueId, companyId, title: "Finished work", status: "done", assigneeAgentId: agentId }); + await db.insert(tables.heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", + invocationSource: "on_demand", finishedAt: now, contextSnapshot: { issueId } }); + return { companyId, agentId, issueId, runId }; + } + + async function connection(f: Awaited>) { + const [application] = await db.insert(tables.toolApplications).values({ companyId: f.companyId, + applicationKey: randomUUID(), name: "History fixture", type: "rest_api" }).returning(); + const [connection] = await db.insert(tables.toolConnections).values({ companyId: f.companyId, + applicationId: application.id, name: "Disabled draft", uid: randomUUID(), transport: "rest_api", status: "draft", enabled: false }).returning(); + const [grant] = await db.insert(tables.connectionGrants).values({ companyId: f.companyId, connectionId: connection.id, + kind: "organization", status: "revoked", revokedAt: now }).returning(); + return { connectionId: connection.id, grantId: grant.id }; + } + + async function mcp(f: Awaited>, agent = false) { + const userId = randomUUID(), clientId = randomUUID(); + await db.insert(tables.authUsers).values({ id: userId, name: "History user", email: `${userId}@example.test`, createdAt: now, updatedAt: now }); + await db.insert(tables.mcpOauthClients).values({ id: clientId, name: "History client", redirectUris: ["https://example.test/callback"] }); + const [grant] = await db.insert(tables.mcpOauthGrants).values({ companyId: f.companyId, userId, clientId, + resource: "https://example.test/mcp", purpose: agent ? "agent" : "personal", agentId: agent ? f.agentId : null, + scopes: agent ? ["paperclip:agent", "offline_access"] : ["paperclip:read"], revokedAt: now }).returning(); + return { userId, grantId: grant.id }; + } + + async function decisionFixture() { + const { decisionService } = await import("../services/decisions.js"); + const f = await seed(), p = await mcp(f); + await db.insert(tables.companyMemberships).values({ companyId: f.companyId, principalType: "user", principalId: p.userId, + status: "active", membershipRole: "member" }); + await db.update(tables.heartbeatRuns).set({ status: "running", responsibleUserId: p.userId }).where(eq(tables.heartbeatRuns.id, f.runId)); + await db.update(tables.issues).set({ status: "in_progress", responsibleUserId: p.userId }).where(eq(tables.issues.id, f.issueId)); + const actor = { type: "agent" as const, companyId: f.companyId, agentId: f.agentId, runId: f.runId, source: "agent_jwt" as const, + onBehalfOfUserId: p.userId, onBehalfOfMemberships: [{ companyId: f.companyId, membershipRole: "member", status: "active" }] }; + const userActor = { type: "board" as const, userId: p.userId, companyIds: [f.companyId], source: "session" as const, + memberships: [{ companyId: f.companyId, membershipRole: "member", status: "active" }] }; + const service = decisionService(db, { wakeOriginAgent: async () => null }); + const finish = () => db.update(tables.heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(tables.heartbeatRuns.id, f.runId)); + return { ...f, ...p, actor, userActor, service, finish }; + } + + async function connectedMcp(f: Awaited>) { + const { createPublicMcpOAuth, hashMcpSecret } = await import("../services/public-mcp/oauth.js"); + const { createMcpApiDispatch } = await import("../services/public-mcp/capabilities.js"); + const { issueRoutes } = await import("../routes/issues.js"); + const { createStorageService } = await import("../storage/service.js"); + const { createLocalDiskStorageProvider } = await import("../storage/local-disk-provider.js"); + const { instanceSettingsService } = await import("../services/instance-settings.js"); + const { Router } = await import("express"); + await instanceSettingsService(db).updateExperimental({ enablePublicMcp: true }); + const p = await mcp(f), origin = "https://history.example", resource = `${origin}/mcp/paperclip`; + 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, resource, scopes: ["paperclip:read", "paperclip:write"] }) + .where(eq(tables.mcpOauthGrants.id, p.grantId)); + const token = randomBytes(32).toString("base64url"); + await db.insert(tables.mcpOauthTokens).values({ grantId: p.grantId, kind: "access", tokenHash: hashMcpSecret(token), + expiresAt: new Date(Date.now() + 600_000) }); + const directory = await mkdtemp(path.join(tmpdir(), "history-mcp-storage-")); + 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 }) }; + } + + it("checks all 32 families on the real migrated schema and keeps an empty instance eligible", async () => { + expect(new Set(completedHistoryChecks.map(c => c.reason)).size).toBe(32); + for (const check of completedHistoryChecks) expect(await blocked(check.reason), check.reason).toBe(false); + expect(await read()).toEqual({ version: 1, backgroundWork: "none" }); + }); + + it.each(["completed", "cancelled"] as const)("retains a %s transfer without hiding its actual spool", async status => { + const f = await seed(); + 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); + 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-")); + try { + await writeFile(path.join(directory, result.run.id), "unfinished cleanup"); + const local = () => inspectIdleSpool(directory); + expect(await readIdleSleepSafety(db, hold, Date.now, ownerId, local)).toEqual({ version: 1, backgroundWork: "present" }); + await rm(path.join(directory, result.run.id)); + expect(await readIdleSleepSafety(db, hold, Date.now, ownerId, local)).toEqual({ version: 1, backgroundWork: "none" }); + expect(await companyTransferRunService.getRun(db, result.run.id)).toMatchObject({ status }); + const next = 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); + await companyTransferRunService.fail(db, next.run.id, "retry remains available"); + expect(await blocked("company_transfer_runs")).toBe(true); + } finally { await rm(directory, { recursive: true, force: 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, + operation: "paperclip_create_task", requestId: randomUUID(), argumentsHash: "a", status: "completed", result: { id: f.issueId } }).returning(); + const [upload] = await db.insert(tables.mcpAttachmentUploads).values({ ...p, companyId: f.companyId, taskId: f.issueId, + requestId: randomUUID(), argumentsHash: "a", originalFilename: "history.txt", contentType: "text/plain", byteSize: 1, + sha256: "a".repeat(64), objectKey: "history/abandoned", storageProvider: "local_disk", expiresAt: now, cleanedAt: now }).returning(); + expect(await blocked("mcp_mutation_receipts")).toBe(false); + expect(await blocked("mcp_attachment_uploads")).toBe(false); + expect(await read()).toEqual({ version: 1, backgroundWork: "none" }); + await db.update(tables.mcpMutationReceipts).set({ status: "unknown" }).where(eq(tables.mcpMutationReceipts.id, receipt.id)); + expect(await blocked("mcp_mutation_receipts")).toBe(true); + await db.update(tables.mcpMutationReceipts).set({ status: "completed", result: null }).where(eq(tables.mcpMutationReceipts.id, receipt.id)); + expect(await blocked("mcp_mutation_receipts")).toBe(true); + await db.update(tables.mcpAttachmentUploads).set({ cleanedAt: null }).where(eq(tables.mcpAttachmentUploads.id, upload.id)); + expect(await blocked("mcp_attachment_uploads")).toBe(true); + }); + + it("retains the real MCP mutation receipt and still blocks the actual task it creates", async () => { + const { createPublicMcpExecutor } = await import("../services/public-mcp/capabilities.js"); + const f = await seed(), p = await connectedMcp(f); + try { + // Pausing is the product's ordinary admission control. No provider turn + // is run merely to exercise a mutation receipt or task creation. + await db.update(tables.agents).set({ status: "paused" }).where(eq(tables.agents.id, f.agentId)); + const execute = createPublicMcpExecutor(db, p.oauth, p.api); + const input = { companyId: f.companyId, title: "Actual MCP producer", description: "Retained receipt", + requestId: randomUUID(), assigneeAgentId: f.agentId }; + const result = await execute(p.token, "paperclip_create_task", input); + expect(result).toHaveProperty("task.status", "todo"); + expect(await blocked("mcp_mutation_receipts")).toBe(false); + expect(await execute(p.token, "paperclip_create_task", input)).toEqual(result); + expect(await db.select().from(tables.mcpMutationReceipts)).toHaveLength(1); + expect(await read()).toEqual({ version: 1, backgroundWork: "present" }); + const receipt = (await db.select().from(tables.mcpMutationReceipts))[0]; + await db.update(tables.mcpMutationReceipts).set({ status: "reserved" }).where(eq(tables.mcpMutationReceipts.id, receipt.id)); + expect(await blocked("mcp_mutation_receipts")).toBe(true); + } finally { await p.cleanup(); } + }, 60_000); + + it.each([200, 410])("uses actual MCP subscription and callback delivery receipts for HTTP %s", async status => { + const { createPublicMcpEvents } = await import("../services/public-mcp/events.js"); + const { logActivity } = await import("../services/activity-log.js"); + const f = await seed(), p = await connectedMcp(f); + try { + let received = 0; + const events = createPublicMcpEvents(db, p.oauth, p.api, { fetch: async (_url, init) => { + const body = JSON.parse(String(init.body)); + if (body.type === "verification") return Response.json({ challenge: body.challenge }); + received += 1; + return new Response(null, { status }); + } }); + const principal = await p.oauth.authenticate(p.token); + await events.subscribe(principal, { name: "paperclip.task.status_changed", arguments: { companyId: f.companyId, taskId: f.issueId }, + delivery: { mode: "webhook", url: "https://receiver.example/events", secret: `whsec_${randomBytes(32).toString("base64")}` } }); + await logActivity(db, { companyId: f.companyId, actorType: "system", actorId: "history", + action: "issue.updated", entityType: "issue", entityId: f.issueId, details: { status: "done" } }); + await events.tick(); + expect(received).toBeGreaterThan(0); + expect(await blocked("mcp_event_deliveries")).toBe(false); + expect((await db.select().from(tables.mcpEventDeliveries)).every(row => row.finishedAt !== null)).toBe(true); + expect(await blocked("mcp_event_subscriptions")).toBe(status !== 410); + expect(await blocked("mcp_oauth_grants")).toBe(true); + } finally { await p.cleanup(); } + }, 60_000); + + it.each(["delivered", "filtered", "inactive", "delivery_exhausted", "http_400", "http_503"])("retires finished MCP event outcome %s only with its source", async outcome => { + const f = await seed(), p = await mcp(f); + const subscriptionId = randomUUID(); + await db.insert(tables.mcpEventSubscriptions).values({ id: subscriptionId, companyId: f.companyId, grantId: p.grantId, + name: "paperclip.task.updated", taskId: f.issueId, arguments: { taskId: f.issueId }, deliveryMaterial: {}, verifiedAt: now, + expiresAt: now, stoppedAt: now }); + const [activity] = await db.insert(tables.activityLog).values({ companyId: f.companyId, actorType: "system", actorId: "history", + action: "issue.updated", entityType: "issue", entityId: f.issueId }).returning(); + const [delivery] = await db.insert(tables.mcpEventDeliveries).values({ subscriptionId, activityId: activity.id, + event: {}, finishedAt: now, outcome, attempts: 6 }).returning(); + expect(await blocked("mcp_event_subscriptions")).toBe(false); + expect(await blocked("mcp_event_deliveries")).toBe(false); + await db.update(tables.mcpEventDeliveries).set({ finishedAt: null }).where(eq(tables.mcpEventDeliveries.id, delivery.id)); + expect(await blocked("mcp_event_deliveries")).toBe(true); + }); + + 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(); + const [run] = await db.insert(tables.browserUseRuns).values({ companyId: f.companyId, sessionId: session.id, heartbeatRunId: f.runId, + invocationId, providerRunId: randomUUID(), status: "completed", eventsDrained: 1, + result: { paperclipBudgetHold: { invocationId, state: "settled", accountedCentsExact: "0.0000000" } } }).returning(); + const [browser] = await db.insert(tables.browserUseBrowsers).values({ companyId: f.companyId, sessionId: session.id, + providerBrowserId: randomUUID(), status: "stopped" }).returning(); + for (const reason of ["browser_use_sessions", "browser_use_runs", "browser_use_browsers"]) expect(await blocked(reason)).toBe(false); + await db.update(tables.browserUseRuns).set({ eventsDrained: 0 }).where(eq(tables.browserUseRuns.id, run.id)); + expect(await blocked("browser_use_sessions")).toBe(true); + expect(await blocked("browser_use_runs")).toBe(true); + await db.update(tables.browserUseRuns).set({ eventsDrained: 1 }).where(eq(tables.browserUseRuns.id, run.id)); + await db.update(tables.browserUseBrowsers).set({ status: "active" }).where(eq(tables.browserUseBrowsers.id, browser.id)); + expect(await blocked("browser_use_sessions")).toBe(true); + expect(await blocked("browser_use_browsers")).toBe(true); + }); + + it("uses one conversation proof for ordinary and scheduled reports and detects the comment-to-wake crash window", async () => { + const { issueService } = await import("../services/issues.js"); + const { settleConversationTurn } = await import("../services/agent-conversations.js"); + const { instanceSettingsService } = await import("../services/instance-settings.js"); + await instanceSettingsService(db).updateExperimental({ enableAgentChat: true }); + const f = await seed(); + const service = issueService(db); + const conversation = await service.create(f.companyId, { title: "History chat", assigneeAgentId: f.agentId, + conversationAgentId: f.agentId, conversationUserId: "local-board", conversationState: "waiting", status: "in_review" }); + expect(await read()).toEqual({ version: 1, backgroundWork: "none" }); + expect((await readIdleIssueSchedule(db, ownerId, Date.now())).blocked).toBe(false); + const comment = await service.addComment(conversation.id, "A new question", { userId: "local-board" }, + { clientRequestId: randomUUID() }); + expect(await read()).toEqual({ version: 1, backgroundWork: "present" }); + // Preserve the original comment but reproduce the crash before enqueue. + await db.update(tables.issues).set({ conversationState: "waiting" }).where(eq(tables.issues.id, conversation.id)); + expect((await readIdleIssueSchedule(db, ownerId, Date.now())).blocked).toBe(true); + const runId = randomUUID(); + const [run] = await db.insert(tables.heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, + invocationSource: "on_demand", status: "succeeded", finishedAt: now, + contextSnapshot: { issueId: conversation.id, commentId: comment.id, conversationSessionGeneration: 0 } }).returning(); + await db.insert(tables.agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + idempotencyKey: `conversation-comment:${comment.id}`, source: "on_demand", triggerDetail: "manual", reason: "issue_commented", + status: "completed", finishedAt: now, runId, payload: { issueId: conversation.id } }); + await db.insert(tables.issueComments).values({ companyId: f.companyId, issueId: conversation.id, + authorAgentId: f.agentId, createdByRunId: runId, body: "The question is answered." }); + await db.update(tables.issues).set({ conversationState: "active", status: "in_progress" }).where(eq(tables.issues.id, conversation.id)); + expect(await settleConversationTurn(db, run)).toBe(true); + expect(await read()).toEqual({ version: 1, backgroundWork: "none" }); + expect((await readIdleIssueSchedule(db, ownerId, Date.now())).blocked).toBe(false); + await db.update(tables.heartbeatRuns).set({ costAccountingPending: true }).where(eq(tables.heartbeatRuns.id, runId)); + expect(await read()).toEqual({ version: 1, backgroundWork: "present" }); + }, 60_000); + + 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(); + const [proposal] = await db.insert(tables.companySecretProposals).values({ companyId: f.companyId, kind: "secret", + proposedName: "history/token", proposedKey: "TOKEN", justification: "History fixture", proposedByAgentId: f.agentId, + originRunId: f.runId, originIssueId: f.issueId, expiresAt: now, valueCiphertext: { pending: true } }).returning(); + expect(await blocked("company_secret_proposals")).toBe(true); + const result = await createSecretProposalsService(db).transition(f.companyId, proposal.id, status, { proposerAgentId: f.agentId }); + expect(result).toMatchObject({ status, valueCiphertext: null }); + expect(await blocked("company_secret_proposals")).toBe(false); + await db.update(tables.companySecretProposals).set({ valueCiphertext: { unfinished: true } }).where(eq(tables.companySecretProposals.id, proposal.id)); + expect(await blocked("company_secret_proposals")).toBe(true); + }, 60_000); + + it("retains actual decided effects, and reblocks when one effect lacks its completion receipt", async () => { + const { decisionService } = await import("../services/decisions.js"); + const previous = process.env.PAPERCLIP_DECISION_SIGNING_SECRET; + process.env.PAPERCLIP_DECISION_SIGNING_SECRET = "0123456789abcdef0123456789abcdef"; + try { + const f = await seed(), p = await mcp(f); + await db.insert(tables.companyMemberships).values({ companyId: f.companyId, principalType: "user", principalId: p.userId, + status: "active", membershipRole: "member" }); + await db.update(tables.heartbeatRuns).set({ status: "running", responsibleUserId: p.userId }).where(eq(tables.heartbeatRuns.id, f.runId)); + await db.update(tables.issues).set({ status: "in_progress", responsibleUserId: p.userId }).where(eq(tables.issues.id, f.issueId)); + const actor = { type: "agent" as const, companyId: f.companyId, agentId: f.agentId, runId: f.runId, source: "agent_jwt" as const, + onBehalfOfUserId: p.userId, onBehalfOfMemberships: [{ companyId: f.companyId, membershipRole: "member", status: "active" }] }; + const userActor = { type: "board" as const, userId: p.userId, companyIds: [f.companyId], source: "session" as const, + memberships: [{ companyId: f.companyId, membershipRole: "member", status: "active" }] }; + const service = decisionService(db, { wakeOriginAgent: async () => null }); + const decision = await service.create({ companyId: f.companyId, actor, agentId: f.agentId, runId: f.runId, + title: "Save this comment?", body: "Completion proof", continuationPolicy: "none", + options: [{ id: "yes", label: "Yes", effects: [{ type: "comment_on_issue", targetIssueId: f.issueId, + staleness: "lenient", bodyMarkdown: "Saved by the decision owner." }] }] }); + expect(await blocked("decisions")).toBe(true); + await service.decide({ id: decision.id, optionId: "yes", decidedByUserId: p.userId, userActor }); + await db.update(tables.heartbeatRuns).set({ status: "succeeded", finishedAt: now }).where(eq(tables.heartbeatRuns.id, f.runId)); + expect(await blocked("decisions")).toBe(false); + expect(await blocked("decision_effect_executions")).toBe(false); + await db.update(tables.decisions).set({ chosenOptionId: "another-option" }).where(eq(tables.decisions.id, decision.id)); + expect(await blocked("decisions")).toBe(true); + await db.update(tables.decisions).set({ chosenOptionId: "yes" }).where(eq(tables.decisions.id, decision.id)); + const [effect] = await db.select().from(tables.decisionEffectExecutions).where(eq(tables.decisionEffectExecutions.decisionId, decision.id)); + await db.delete(tables.decisionEffectExecutions).where(eq(tables.decisionEffectExecutions.id, effect.id)); + expect(await blocked("decisions")).toBe(true); + await db.insert(tables.decisionEffectExecutions).values(effect); + await db.update(tables.decisionEffectExecutions).set({ executedAt: null }).where(eq(tables.decisionEffectExecutions.decisionId, decision.id)); + expect(await blocked("decisions")).toBe(true); + expect(await blocked("decision_effect_executions")).toBe(true); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_DECISION_SIGNING_SECRET; + else process.env.PAPERCLIP_DECISION_SIGNING_SECRET = previous; + } + }, 60_000); + + it.each(["zero_effect", "dismissed", "cancelled", "expired", "partial", "failed"] as const)("proves the actual decision owner branch %s", async branch => { + const previous = process.env.PAPERCLIP_DECISION_SIGNING_SECRET; + process.env.PAPERCLIP_DECISION_SIGNING_SECRET = "0123456789abcdef0123456789abcdef"; + try { + const f = await decisionFixture(); + const effects = branch === "zero_effect" ? [] : [ + { type: "comment_on_issue" as const, targetIssueId: f.issueId, staleness: "lenient" as const, bodyMarkdown: "First effect" }, + ...(branch === "partial" ? [{ type: "comment_on_issue" as const, targetIssueId: f.issueId, + staleness: "strict" as const, bodyMarkdown: "Snapshot is now stale" }] : []), + ]; + const decision = await f.service.create({ companyId: f.companyId, actor: f.actor, agentId: f.agentId, runId: f.runId, + title: "Decision branch", body: branch, continuationPolicy: "none", options: [{ id: "yes", label: "Yes", effects }] }); + if (branch === "dismissed") await f.service.dismiss(decision.id, f.userId, f.userActor); + else if (branch === "cancelled") await f.service.cancel(decision.id, { actorType: "agent", actorId: f.agentId, runId: f.runId }); + else if (branch === "expired") { + await db.update(tables.decisions).set({ expiresAt: new Date(0) }).where(eq(tables.decisions.id, decision.id)); + await f.service.sweepTtlExpiry(); + } else await f.service.decide({ id: decision.id, optionId: "yes", decidedByUserId: f.userId, + userActor: branch === "failed" ? { type: "none", source: "none" } : f.userActor }); + await f.finish(); + const saved = await f.service.get(decision.id); + expect(saved?.executionStatus).toBe(branch === "partial" ? "partial" : branch === "failed" ? "failed" + : branch === "cancelled" || branch === "expired" ? null : "succeeded"); + expect(await blocked("decisions")).toBe(false); + expect(await blocked("decision_effect_executions")).toBe(false); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_DECISION_SIGNING_SECRET; else process.env.PAPERCLIP_DECISION_SIGNING_SECRET = previous; + } + }, 60_000); + + 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; + process.env.PAPERCLIP_DECISION_SIGNING_SECRET = "0123456789abcdef0123456789abcdef"; + try { + const f = await decisionFixture(), activityAt = new Date("2026-04-01T00:00:00.000Z"); + await db.insert(tables.decisionRetention).values({ companyId: f.companyId, sourceKind: "review", sourceId: f.issueId, + sourceActivityAt: activityAt, version: 1 }); + const manifest = [{ companyId: f.companyId, sourceKind: "review" as const, sourceId: f.issueId, + expectedVersion: 1, activityAt: activityAt.toISOString(), reason: "Reviewed history" }]; + const decision = await f.service.create({ companyId: f.companyId, actor: f.actor, agentId: f.agentId, runId: f.runId, + title: "Archive reviewed set", body: "Only the selected source", continuationPolicy: "none", + options: [{ id: "archive", label: "Archive", style: "destructive", effects: [] }, { id: "keep", label: "Keep", effects: [] }], + metadata: { kind: "attention_archive_proposal", manifestHash: branch === "manifest_failure" ? "invalid" : hashAttentionArchiveManifest(manifest) }, + additionalTargetSnapshots: { [`attention:review:${f.issueId}`]: { status: "attention", assigneeAgentId: null, + assigneeUserId: null, updatedAt: activityAt.toISOString(), attentionArchive: manifest[0] } } }); + const outcome = await f.service.decide({ id: decision.id, optionId: branch === "keep" ? "keep" : "archive", + decidedByUserId: f.userId, userActor: f.userActor }); + expect(outcome.executionStatus).toBe(branch === "manifest_failure" ? "failed" : "succeeded"); + await f.finish(); + expect(await blocked("decisions")).toBe(false); + if (branch === "archive") { + await db.update(tables.decisions).set({ targetSnapshots: {} }).where(eq(tables.decisions.id, decision.id)); + 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; + } + }, 60_000); + + it("settles retained Muse assignments through the real pair, runner receipt and terminal reconciliation paths", async () => { + const { instanceSettingsService } = await import("../services/instance-settings.js"); + const { museIdentity, museCredentialHash } = await import("../services/muse-identity.js"); + const { museRunnerBroker } = await import("../services/muse-runner-broker.js"); + const { createHeartbeatLifecycle } = await import("../services/heartbeat/run-lifecycle.js"); + const f = await seed(), p = await mcp(f); + await instanceSettingsService(db).updateExperimental({ enableNativeRunner: true, enableMuse: true }); + await db.insert(tables.companyMemberships).values([ + { companyId: f.companyId, principalType: "user", principalId: p.userId, membershipRole: "owner", status: "active" }, + { companyId: f.companyId, principalType: "agent", principalId: f.agentId, membershipRole: "member", status: "active" }, + ]); + const adapterConfig = { provider: "muse", allowUnmeteredProvider: true, lifecycle: "per_turn" }; + await db.update(tables.agents).set({ status: "active", adapterType: "paperclip_runner", adapterConfig }).where(eq(tables.agents.id, f.agentId)); + const ticket = randomBytes(32).toString("base64url"); + const [binding] = await db.insert(tables.museAgentBindings).values({ companyId: f.companyId, agentId: f.agentId, operatorId: p.userId, + ticketHash: museCredentialHash(ticket), ticketExpiresAt: new Date(Date.now() + 600_000) }).returning(); + await db.update(tables.agents).set({ adapterConfig: { ...adapterConfig, museBindingId: binding.id } }).where(eq(tables.agents.id, f.agentId)); + const identity = museIdentity(db), credentials = await identity.pair(ticket, "2"); + const broker = museRunnerBroker(db, { publicOrigin: "https://paperclip.example", + heartbeat: { wakeup: async () => null, cancelRun: async () => null } }); + await db.update(tables.museAgentBindings).set({ status: "ready", receiverContactAt: new Date(), verifiedReplyAt: new Date() }) + .where(eq(tables.museAgentBindings.id, binding.id)); + const subject = await identity.authenticate(credentials.accessToken), sessionId = randomUUID(), turnId = `turn_${randomUUID()}`; + 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: "muse_external", + nativeIssueId: f.issueId, nativeSessionId: sessionId, nativePhase: "turn_running", responsibleUserId: p.userId, finishedAt: null }) + .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 execution = { binding: { companyId: f.companyId, agentId: f.agentId, runId: f.runId }, + provider: { binding: await broker.snapshot(f.companyId, f.agentId, binding.id) } }; + const port = broker.port(execution); + const ref = { ...execution.provider.binding, runId: f.runId, normalizedSessionId: sessionId, turnId, assignmentRevision: 1 }; + const detach = await port.attach(async command => { + // 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" } } }); + }); + try { + await port.dispatch({ sourceEventId: `offer_${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.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() }); + 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"]) { + 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); + } finally { await detach(); } + }, 60_000); + + it("uses the actual Dot revocation writer but preserves Grok routine and recovery blockers", async () => { + const { dotRunnerBroker } = await import("../services/dot-runner-broker.js"); + const f = await seed(), p = await mcp(f, true); + await db.update(tables.mcpOauthGrants).set({ revokedAt: null }).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, status: "ready", providerKind: "openai_dot" }).returning(); + expect(await blocked("dot_agent_bindings")).toBe(true); + await dotRunnerBroker(db).revoke(f.companyId, f.agentId, p.userId); + expect(await blocked("dot_agent_bindings")).toBe(false); + expect(await blocked("mcp_oauth_grants")).toBe(false); + const [secret] = await db.insert(tables.companySecrets).values({ companyId: f.companyId, key: "GROK_ROUTINE", + name: "Grok routine material", provider: "local_encrypted" }).returning(); + const [grok] = await db.insert(tables.dotAgentBindings).values({ companyId: f.companyId, agentId: f.agentId, + operatorId: p.userId, status: "revoked", providerKind: "grok_bot", revokedAt: now }).returning(); + await db.insert(tables.grokBotRoutines).values({ companyId: f.companyId, agentId: f.agentId, bindingId: grok.id, + bindingGeneration: grok.generation, botName: "History routine", secretId: secret.id, secretVersion: 1, + recoveryRequiredAt: new Date(), recoveryReason: "work_session" }); + expect(await blocked("dot_agent_bindings")).toBe(true); + expect(await read()).toEqual({ version: 1, backgroundWork: "present" }); + expect(await db.select().from(tables.grokBotRoutines)).toHaveLength(1); + expect((await db.select().from(tables.dotAgentBindings).where(eq(tables.dotAgentBindings.id, binding.id)))[0].status).toBe("revoked"); + }, 60_000); + + it("retains a stale runtime exclusivity lease only after its exact owner and resources finish", async () => { + const { workspaceRuntimeLeaseService } = await import("../services/workspace-runtime-leases.js"); + const f = await seed(); + const [workspace] = await db.insert(tables.executionWorkspaces).values({ companyId: f.companyId, name: "History workspace", + mode: "shared_workspace", strategyType: "project_primary", providerType: "local_fs" }).returning(); + const lease = await workspaceRuntimeLeaseService(db).claim({ companyId: f.companyId, executionWorkspaceId: workspace.id, + action: "start", owner: { actorType: "agent", agentId: f.agentId, issueId: f.issueId, runId: f.runId } }); + expect(lease.outcome).toBe("created"); + expect(await blocked("execution_workspace_runtime_leases")).toBe(false); + const [runtime] = await db.insert(tables.workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId, + executionWorkspaceId: workspace.id, issueId: f.issueId, serviceName: "Unstarted history", scopeType: "execution_workspace", + scopeId: workspace.id, status: "stopped", lifecycle: "shared", provider: "local_process", stoppedAt: now }).returning(); + expect(await blocked("workspace_runtime_services")).toBe(false); + await db.update(tables.workspaceRuntimeServices).set({ status: "running", stoppedAt: null }).where(eq(tables.workspaceRuntimeServices.id, runtime.id)); + expect(await blocked("workspace_runtime_services")).toBe(true); + expect(await blocked("execution_workspace_runtime_leases")).toBe(true); + await db.update(tables.workspaceRuntimeServices).set({ status: "stopped", stoppedAt: now, exposureHandle: "uncleaned" }) + .where(eq(tables.workspaceRuntimeServices.id, runtime.id)); + expect(await blocked("workspace_runtime_services")).toBe(true); + await db.update(tables.executionWorkspaceRuntimeLeases).set({ ownerKey: `issue:${randomUUID()}` }) + .where(eq(tables.executionWorkspaceRuntimeLeases.id, lease.lease!.id)); + expect(await blocked("execution_workspace_runtime_leases")).toBe(true); + }); + + it("inspects an actual owned runtime process across stopped-row fallback, adoption, termination and reconstruction failure", async () => { + const runtime = await import("../services/workspace-runtime.js"); + const supervisor = await import("../services/local-service-supervisor.js"); + let ownedProcess: Awaited> = null; + const directory = await mkdtemp(path.join(tmpdir(), "history-owned-runtime-")); + const previousHome = process.env.PAPERCLIP_HOME, previousInstance = process.env.PAPERCLIP_INSTANCE_ID; + process.env.PAPERCLIP_HOME = directory; + process.env.PAPERCLIP_INSTANCE_ID = "history-owned-runtime"; + const f = await seed(); + const [workspace] = await db.insert(tables.executionWorkspaces).values({ companyId: f.companyId, name: "Owned local runtime", + mode: "shared_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: directory }).returning(); + try { + await runtime.resetRuntimeServicesForTests({ terminateProcesses: true }); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); + await runtime.reconcilePersistedRuntimeServicesOnStartup(db); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("none"); + const server = net.createServer(); + await new Promise((resolve, reject) => { server.once("error", reject); server.listen(0, "127.0.0.1", resolve); }); + const port = (server.address() as net.AddressInfo).port; + await new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())); + const command = "node -e \"require('node:http').createServer((req,res)=>res.end('history')).listen(Number(process.env.PORT),'127.0.0.1')\""; + const refs = await runtime.startRuntimeServicesForWorkspaceControl({ db, + actor: { id: null, name: "History operator", companyId: f.companyId }, issue: null, + executionWorkspaceId: workspace.id, adapterEnv: {}, + workspace: { baseCwd: directory, source: "project_primary", projectId: null, workspaceId: null, + repoUrl: null, repoRef: null, strategy: "project_primary", cwd: directory, branchName: null, + worktreePath: null, warnings: [], created: false, branchCreatedByRuntime: false }, + config: { desiredState: "running", workspaceRuntime: { services: [{ name: "history-http", command, port, + expose: { urlTemplate: "http://127.0.0.1:{{port}}" }, + readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10, intervalMs: 50 }, + lifecycle: "shared", reuseScope: "execution_workspace", stopPolicy: { type: "manual" } }] } } }); + expect(refs).toHaveLength(1); + ownedProcess = await supervisor.findLocalServiceRegistryRecordByRuntimeServiceId({ runtimeServiceId: refs[0].id, profileKind: "workspace-runtime" }); + expect(ownedProcess?.pid).toBeGreaterThan(0); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("present"); + await db.update(tables.workspaceRuntimeServices).set({ status: "stopped", stoppedAt: new Date(), url: null }) + .where(eq(tables.workspaceRuntimeServices.id, refs[0].id)); + expect(await blocked("workspace_runtime_services")).toBe(false); + expect(await readIdleSleepSafety(db, hold, Date.now, ownerId, async () => runtime.readWorkspaceRuntimeIdleWork())) + .toEqual({ version: 1, backgroundWork: "present" }); + await runtime.resetRuntimeServicesForTests({ simulateSupervisorExit: true }); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); + 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 }); + ownedProcess = null; + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("none"); + expect(await blocked("workspace_runtime_services")).toBe(false); + const unavailable = Object.create(db) as typeof db; + unavailable.select = () => { throw new Error("controlled startup database failure"); }; + await expect(runtime.reconcilePersistedRuntimeServicesOnStartup(unavailable)).rejects.toThrow("controlled startup"); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); + await runtime.reconcilePersistedRuntimeServicesOnStartup(db); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("none"); + await runtime.resetRuntimeServicesForTests(); + expect(runtime.readWorkspaceRuntimeIdleWork()).toBe("unknown"); + } finally { + await runtime.resetRuntimeServicesForTests({ terminateProcesses: true }); + if (ownedProcess) await supervisor.terminateLocalService({ pid: ownedProcess.pid, + processGroupId: ownedProcess.processGroupId ?? ownedProcess.pid, port: ownedProcess.port }); + if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome; + if (previousInstance === undefined) delete process.env.PAPERCLIP_INSTANCE_ID; else process.env.PAPERCLIP_INSTANCE_ID = previousInstance; + await rm(directory, { recursive: true, force: true }); + } + }, 60_000); +}); diff --git a/server/src/modules/external-agents/idle-history-proof.ts b/server/src/modules/external-agents/idle-history-proof.ts new file mode 100644 index 00000000000..cbed05b1f77 --- /dev/null +++ b/server/src/modules/external-agents/idle-history-proof.ts @@ -0,0 +1,89 @@ +import { finishedIdleRun } from "../../services/idle-work-proof.js"; + +function completedAssignment(alias: string, provider: "dot" | "muse"): string { + const muse = provider === "muse"; + const identity = muse ? 'muse' : 'openai_dot'; + return `EXISTS (SELECT 1 FROM ${provider}_agent_bindings binding JOIN heartbeat_runs run + ON run.id = ${alias}.run_id AND run.company_id = binding.company_id AND run.agent_id = binding.agent_id + WHERE binding.id = ${alias}.binding_id AND binding.company_id = ${alias}.company_id + ${muse ? '' : "AND binding.provider_kind = 'openai_dot'"} + AND binding.agent_id = ${alias}.agent_id AND ${alias}.binding_generation > 0 + AND ${alias}.binding_generation <= binding.generation AND ${finishedIdleRun('run')} + AND ((${alias}.status = 'settled' AND ${muse + ? `run.status = 'succeeded' AND ${alias}.accepted_result_at IS NOT NULL AND ${alias}.finalized_at = run.finished_at` + : `${alias}.settled_at IS NOT NULL`}) + OR (${alias}.status = 'fenced' AND (${alias}.${muse ? 'claimed_at' : 'accepted_at'} IS NULL + OR EXISTS (SELECT 1 FROM external_agent_holds h WHERE h.provider = '${identity}' + AND ${muse ? 'true' : 'false'} + AND h.assignment_id = ${alias}.id AND h.company_id = ${alias}.company_id + AND h.agent_id = ${alias}.agent_id AND h.run_id = ${alias}.run_id + AND h.binding_id = ${alias}.binding_id AND h.binding_generation = ${alias}.binding_generation + AND h.released_at IS NOT NULL AND NOT h.worker_unknown AND NOT h.native_effects_unknown)))) + AND NOT EXISTS (SELECT 1 FROM external_agent_holds h WHERE h.provider = '${identity}' + AND h.assignment_id = ${alias}.id AND (h.company_id <> ${alias}.company_id + OR h.run_id <> ${alias}.run_id OR h.binding_id <> ${alias}.binding_id + OR h.binding_generation <> ${alias}.binding_generation OR h.agent_id <> ${alias}.agent_id + OR h.released_at IS NULL OR h.worker_unknown OR h.native_effects_unknown)))`; +} + +function completedOperation(alias: string, provider: "dot" | "muse"): string { + return `${alias}.status IN (${provider === 'dot' ? "'completed', 'rejected'" : "'settled', 'rejected'"}) + AND jsonb_typeof(${alias}.outcome) = 'object' + AND EXISTS (SELECT 1 FROM ${provider}_runner_assignments a WHERE a.id = ${alias}.assignment_id + AND a.company_id = ${alias}.company_id AND (${completedAssignment('a', provider)}))`; +} + +function assignmentCheck(provider: "dot" | "muse") { + return { reason: `${provider}_runner_assignments`, query: `SELECT 1 FROM ${provider}_runner_assignments a WHERE ( + (${completedAssignment('a', provider)}) AND NOT EXISTS (SELECT 1 FROM ${provider}_runner_operations o + WHERE o.assignment_id = a.id AND (o.company_id <> a.company_id OR (${completedOperation('o', provider)}) IS NOT TRUE))) IS NOT TRUE` }; +} + +function mailboxCheck(provider: "dot" | "muse") { + return { reason: `${provider}_mailbox_items`, query: `SELECT 1 FROM ${provider}_mailbox_items m WHERE ( + EXISTS (SELECT 1 FROM ${provider}_agent_bindings b WHERE b.id = m.binding_id AND b.company_id = m.company_id + AND m.binding_generation > 0 AND m.binding_generation <= b.generation) + AND ((m.assignment_id IS NOT NULL AND m.kind IN ('assignment', 'operation_result', 'follow_up', 'authority_revoked'${provider === 'muse' ? ", 'input_available'" : ''}) + AND EXISTS (SELECT 1 FROM ${provider}_runner_assignments a WHERE a.id = m.assignment_id + AND a.company_id = m.company_id AND a.binding_id = m.binding_id AND a.binding_generation = m.binding_generation + AND (${completedAssignment('a', provider)}))) + OR (m.assignment_id IS NULL AND m.kind = 'readiness_challenge' AND EXISTS ( + SELECT 1 FROM ${provider}_agent_bindings b WHERE b.id = m.binding_id AND b.company_id = m.company_id + AND b.status = 'revoked' AND b.revoked_at IS NOT NULL)))) IS NOT TRUE` }; +} + +export const idleExternalAgentHistoryChecks = [ + { reason: "dot_agent_bindings", query: `SELECT 1 FROM dot_agent_bindings b WHERE ( + b.provider_kind = 'openai_dot' AND b.status = 'revoked' AND b.revoked_at IS NOT NULL AND b.pairing_code_hash IS NULL + AND b.challenge_hash IS NULL AND EXISTS (SELECT 1 FROM agents a WHERE a.id = b.agent_id AND a.company_id = b.company_id) + AND (b.grant_id IS NULL OR EXISTS (SELECT 1 FROM mcp_oauth_grants g + WHERE g.id = b.grant_id AND g.company_id = b.company_id AND g.user_id = b.operator_id + AND g.agent_id = b.agent_id AND g.purpose = 'agent' AND g.revoked_at IS NOT NULL))) IS NOT TRUE` }, + { reason: "muse_agent_bindings", query: `SELECT 1 FROM muse_agent_bindings b WHERE ( + b.status = 'revoked' AND b.revoked_at IS NOT NULL AND b.ticket_hash IS NULL AND b.challenge_hash IS NULL + AND ((b.qualification_id IS NULL AND b.detector_removal_requested_at IS NULL) + OR (b.detector_removed_at IS NOT NULL AND b.detector_removed_at >= b.detector_removal_requested_at)) + AND EXISTS (SELECT 1 FROM agents a WHERE a.id = b.agent_id AND a.company_id = b.company_id) + AND (b.grant_id IS NULL OR EXISTS (SELECT 1 FROM mcp_oauth_grants g + WHERE g.id = b.grant_id AND g.company_id = b.company_id AND g.user_id = b.operator_id + AND g.agent_id = b.agent_id AND g.purpose = 'agent' AND g.revoked_at IS NOT NULL))) IS NOT TRUE` }, + { reason: "muse_credentials", query: `SELECT 1 FROM muse_credentials c WHERE ( + c.kind IN ('access', 'refresh', 'signal', 'cleanup', 'detector_cleanup') + AND (c.revoked_at IS NOT NULL OR c.consumed_at IS NOT NULL OR c.expires_at <= now()) + AND EXISTS (SELECT 1 FROM muse_agent_bindings b WHERE b.id = c.binding_id AND b.company_id = c.company_id + AND c.binding_generation > 0 AND c.binding_generation <= b.generation)) IS NOT TRUE` }, + assignmentCheck('dot'), assignmentCheck('muse'), + { reason: "dot_runner_operations", query: `SELECT 1 FROM dot_runner_operations o WHERE (${completedOperation('o', 'dot')}) IS NOT TRUE` }, + { reason: "muse_runner_operations", query: `SELECT 1 FROM muse_runner_operations o WHERE (${completedOperation('o', 'muse')}) IS NOT TRUE` }, + mailboxCheck('dot'), mailboxCheck('muse'), + { reason: "muse_input_deliveries", query: `SELECT 1 FROM muse_input_deliveries i WHERE ( + EXISTS (SELECT 1 FROM muse_runner_assignments a WHERE a.id = i.assignment_id AND a.company_id = i.company_id + AND a.turn_id = i.turn_id AND (${completedAssignment('a', 'muse')}))) IS NOT TRUE` }, + { reason: "external_agent_holds", query: `SELECT 1 FROM external_agent_holds h WHERE ( + h.provider = 'muse' AND h.released_at IS NOT NULL AND NOT h.worker_unknown AND NOT h.native_effects_unknown + AND EXISTS (SELECT 1 FROM muse_runner_assignments a WHERE a.id = h.assignment_id + AND a.company_id = h.company_id AND a.agent_id = h.agent_id AND a.run_id = h.run_id + AND a.binding_id = h.binding_id AND a.binding_generation = h.binding_generation + AND (h.operator_attested_at IS NOT NULL OR (a.status = 'settled' AND a.accepted_result_at IS NOT NULL)) + AND (${completedAssignment('a', 'muse')}))) IS NOT TRUE` }, +] as const; diff --git a/server/src/services/browser-use-idle-proof.ts b/server/src/services/browser-use-idle-proof.ts new file mode 100644 index 00000000000..c61baf1914f --- /dev/null +++ b/server/src/services/browser-use-idle-proof.ts @@ -0,0 +1,49 @@ +import { finishedIdleRun } from "./idle-work-proof.js"; + +/** Mirrors sweep/account: terminal provider status cannot settle usage. */ +export function completedBrowserRun(run: string): string { + const hold = `${run}.result->'paperclipBudgetHold'`; + return `${run}.status IN ('completed', 'failed', 'cancelled') + AND ${run}.events_drained = 1 AND ${run}.recovery_cursor IS NULL + AND jsonb_typeof(${hold}) = 'object' AND ${hold}->>'invocationId' = ${run}.invocation_id::text + AND ((${run}.provider_run_id IS NOT NULL AND ${hold}->>'state' = 'settled' + AND ${hold}->>'accountedCentsExact' ~ '^[0-9]+([.][0-9]+)?$') + OR (${run}.provider_run_id IS NULL AND ${run}.status = 'failed' AND ${hold}->>'state' = 'released')) + -- account() replaces held authority with the authoritative terminal + -- summary and ledger charge. It retains prior diagnostic flags in JSON. + AND (${hold}->>'state' = 'settled' OR COALESCE(${hold}->>'billingUnknown', 'false') = 'false') + AND EXISTS (SELECT 1 FROM browser_use_sessions browser_parent_session JOIN heartbeat_runs browser_parent_run + ON browser_parent_run.id = ${run}.heartbeat_run_id AND browser_parent_run.company_id = browser_parent_session.company_id + AND browser_parent_run.agent_id = browser_parent_session.agent_id + WHERE browser_parent_session.id = ${run}.session_id AND browser_parent_session.company_id = ${run}.company_id + AND (browser_parent_run.native_issue_id = browser_parent_session.issue_id + OR browser_parent_run.context_snapshot->>'issueId' = browser_parent_session.issue_id::text + OR browser_parent_run.context_snapshot->>'taskId' = browser_parent_session.issue_id::text) + AND (browser_parent_run.native_issue_id IS NULL OR browser_parent_run.native_issue_id = browser_parent_session.issue_id) + AND (browser_parent_run.context_snapshot->>'issueId' IS NULL + OR browser_parent_run.context_snapshot->>'issueId' = browser_parent_session.issue_id::text) + AND (browser_parent_run.context_snapshot->>'taskId' IS NULL + OR browser_parent_run.context_snapshot->>'taskId' = browser_parent_session.issue_id::text) + AND ${finishedIdleRun('browser_parent_run')})`; +} + +export const idleBrowserHistoryChecks = [ + { reason: "browser_use_sessions", query: `SELECT 1 FROM browser_use_sessions s WHERE ( + (s.status = 'closed' OR (s.status = 'failed' AND s.provider_session_id IS NULL)) + AND s.lease_token IS NULL AND s.lease_until IS NULL AND s.stop_requested IS NULL + AND EXISTS (SELECT 1 FROM issues i JOIN agents a ON a.id = s.agent_id AND a.company_id = i.company_id + JOIN tool_connections c ON c.id = s.connection_id AND c.company_id = i.company_id + JOIN connection_grants g ON g.id = s.grant_id AND g.company_id = i.company_id AND g.connection_id = c.id + WHERE i.id = s.issue_id AND i.company_id = s.company_id) + AND NOT EXISTS (SELECT 1 FROM browser_use_runs r WHERE r.session_id = s.id + AND (r.company_id <> s.company_id OR (${completedBrowserRun('r')}) IS NOT TRUE)) + AND NOT EXISTS (SELECT 1 FROM browser_use_browsers b WHERE b.session_id = s.id + AND (b.company_id <> s.company_id OR b.status <> 'stopped'))) IS NOT TRUE` }, + { reason: "browser_use_runs", query: `SELECT 1 FROM browser_use_runs r WHERE (${completedBrowserRun('r')}) IS NOT TRUE` }, + // stopped is written only after client.stop followed by client.browser + // confirms stopped. Parent/session/run checks remain independent barriers. + { reason: "browser_use_browsers", query: `SELECT 1 FROM browser_use_browsers b WHERE ( + b.status = 'stopped' AND EXISTS (SELECT 1 FROM browser_use_sessions s + WHERE s.id = b.session_id AND s.company_id = b.company_id AND s.status = 'closed' + AND s.stop_requested IS NULL AND s.lease_token IS NULL AND s.lease_until IS NULL)) IS NOT TRUE` }, +] as const; diff --git a/server/src/services/chat-channels/github/idle-history-proof.ts b/server/src/services/chat-channels/github/idle-history-proof.ts new file mode 100644 index 00000000000..3dade037416 --- /dev/null +++ b/server/src/services/chat-channels/github/idle-history-proof.ts @@ -0,0 +1,9 @@ +export function completedGithubAction(a: string): string { + return `(${a}.kind = 'github_response_comment' AND ${a}.status = 'processed' AND ${a}.result->>'final' = 'true') + OR (${a}.kind = 'github_webhook_ingress' AND ( + (${a}.status = 'processed' AND ${a}.result->>'httpStatus' = '202' AND ${a}.result->>'retryable' = 'false') + OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'github_webhook_ingress_payload_invalid' + AND ${a}.result->>'retryable' = 'false'))) + OR (${a}.kind = 'github_review_check' AND ${a}.status = 'processed' + AND ${a}.result->>'retryable' = 'false')`; +} diff --git a/server/src/services/chat-channels/idle-history-proof.ts b/server/src/services/chat-channels/idle-history-proof.ts new file mode 100644 index 00000000000..778779d2b3d --- /dev/null +++ b/server/src/services/chat-channels/idle-history-proof.ts @@ -0,0 +1,79 @@ +import { finishedIdleRun, finishedIdleWake } from "../idle-work-proof.js"; +import { completedGithubAction } from "./github/idle-history-proof.js"; +import { completedSlackAction } from "./slack/idle-history-proof.js"; +import { completedTelegramAction } from "./telegram/idle-history-proof.js"; + +/** A processed inbound receipt can still own an unconsumed wake. */ +function completedAction(a: string): string { + const wake = `EXISTS (SELECT 1 FROM agent_wakeup_requests w WHERE w.company_id = ${a}.company_id + AND w.id = ${a}.id AND w.agent_id::text = ${a}.payload->>'agentId' + AND w.payload->>'issueId' = ${a}.payload->>'issueId' AND ${finishedIdleWake('w')})`; + return `jsonb_typeof(${a}.payload) = 'object' AND jsonb_typeof(${a}.result) = 'object' + AND EXISTS (SELECT 1 FROM chat_endpoints e WHERE e.id = ${a}.endpoint_id AND e.company_id = ${a}.company_id) + AND (${a}.delivery_id IS NULL OR EXISTS (SELECT 1 FROM chat_deliveries d + WHERE d.id = ${a}.delivery_id AND d.company_id = ${a}.company_id AND d.endpoint_id = ${a}.endpoint_id)) + AND CASE + WHEN ${a}.kind = 'inbound_wakeup' THEN ${a}.status IN ('processed', 'failed') AND ${wake} + WHEN ${a}.kind = 'interaction_wakeup' THEN ${a}.status = 'processed' AND + (${a}.result->>'code' = 'interaction_wakeup_no_longer_applicable' OR EXISTS ( + SELECT 1 FROM agent_wakeup_requests w WHERE w.company_id = ${a}.company_id + AND w.idempotency_key = 'interaction:' || (${a}.payload->>'interactionId') || ':' || (${a}.payload->>'interactionStatus') + AND ${finishedIdleWake('w')})) + WHEN ${a}.kind = 'failed_run_retry' THEN ${a}.status = 'processed' + AND ${a}.result->>'code' = 'failed_chat_retry_durable' AND ${wake} + WHEN ${a}.kind = 'provider_effect' THEN + (${a}.status = 'processed' AND length(${a}.result->>'providerMessageId') > 0 AND length(${a}.result->>'threadId') > 0) + OR (${a}.status = 'cancelled' AND ${a}.result->>'code' = 'provider_effect_no_longer_authorized') + OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'provider_effect_failed' AND ${a}.result->>'retryable' = 'false') + WHEN ${a}.kind = 'receipt_reaction' THEN + (${a}.status = 'processed' AND jsonb_typeof(${a}.result->'attempts') = 'number' AND (${a}.result->>'attempts')::numeric > 0) + OR (${a}.status = 'cancelled' AND ${a}.result->>'code' IN ('receipt_reaction_superseded_by_terminal_publication', 'receipt_reaction_runtime_superseded')) + WHEN ${a}.kind = 'confirmation_response' THEN ${a}.status = 'processed' + AND ${a}.result->>'code' IN ('confirmation_resolution_committed_by_provider', 'confirmation_resolution_reconciled') + AND EXISTS (SELECT 1 FROM issue_thread_interactions interaction WHERE interaction.id::text = ${a}.result->>'interactionId' + AND interaction.company_id = ${a}.company_id AND interaction.status = ${a}.result->>'interactionStatus' + AND interaction.resolved_at IS NOT NULL) + WHEN ${a}.kind = 'task_control_authorization' THEN ${a}.status IN ('processed', 'cancelled') + AND EXISTS (SELECT 1 FROM chat_publications p WHERE p.company_id = ${a}.company_id AND p.endpoint_id = ${a}.endpoint_id + AND p.conversation_id = ${a}.conversation_id AND ${a}.provider_action_id = 'task-control-authorization:' || p.id::text + AND ((p.state = 'published' AND p.published_at IS NOT NULL) OR p.state = 'cancelled')) + ELSE (${[completedGithubAction, completedSlackAction, completedTelegramAction] + .map(proof => `(${proof(a)})`).join(' OR ')}) END`; +} + +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 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)) + AND NOT EXISTS (SELECT 1 FROM chat_actions a WHERE a.delivery_id = d.id + AND (a.company_id <> d.company_id OR a.endpoint_id <> d.endpoint_id OR (${completedAction('a')}) IS NOT TRUE))) IS NOT TRUE` }, + { reason: "chat_publications", query: `SELECT 1 FROM chat_publications p WHERE ( + (p.state = 'published' AND p.published_at IS NOT NULL AND p.redacted_error IS NULL + OR (p.state IN ('failed', 'cancelled') AND p.next_attempt_at IS NULL)) + AND EXISTS (SELECT 1 FROM chat_conversations c JOIN chat_endpoints e + ON e.id = c.endpoint_id AND e.company_id = c.company_id + JOIN issues i ON i.id = p.issue_id AND i.company_id = c.company_id + WHERE c.id = p.conversation_id AND c.company_id = p.company_id AND c.endpoint_id = p.endpoint_id) + AND NOT EXISTS (SELECT 1 FROM chat_endpoint_leases l WHERE l.company_id = p.company_id + AND l.endpoint_id = p.endpoint_id AND l.lease_key LIKE 'publication:' || p.id::text || ':%' AND l.expires_at > now()) + AND NOT EXISTS (SELECT 1 FROM chat_teams_file_transfers f WHERE f.publication_id = p.id + AND (f.company_id <> p.company_id OR f.phase NOT IN ('delivered', 'declined', 'expired', 'cancelled') + OR f.attempt_id IS NOT NULL OR f.attempt_expires_at IS NOT NULL))) IS NOT TRUE` }, + { reason: "chat_actions", query: `SELECT 1 FROM chat_actions a WHERE (${completedAction('a')}) IS NOT TRUE` }, + { reason: "chat_completion_deliveries", query: `SELECT 1 FROM chat_completion_deliveries d WHERE ( + d.status IN ('delivered', 'superseded', 'exhausted') + AND EXISTS (SELECT 1 FROM chat_task_handoffs h JOIN issues task ON task.id = h.task_id AND task.company_id = h.company_id + 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 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 (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 (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/chat-channels/slack/idle-history-proof.ts b/server/src/services/chat-channels/slack/idle-history-proof.ts new file mode 100644 index 00000000000..f57e9950f12 --- /dev/null +++ b/server/src/services/chat-channels/slack/idle-history-proof.ts @@ -0,0 +1,16 @@ +export function completedSlackAction(a: string): string { + return `(${a}.kind = 'slack_session_sync' AND ( + (${a}.status = 'processed' AND ${a}.result->>'outcome' IN ('updated', 'unsupported') + AND (${a}.result->>'outcome' <> 'updated' OR ${a}.result->>'sessionStatus' <> 'processing')) + OR (${a}.status = 'cancelled' AND ${a}.result->>'code' IN ('slack_session_no_published_state', 'slack_session_destination_changed')) + OR (${a}.status = 'failed' AND ${a}.result->>'code' = 'slack_session_sync_rejected'))) + OR (${a}.kind = 'slash_task_start' AND ${a}.status = 'processed' + AND length(${a}.result->>'providerMessageId') > 0 AND length(${a}.result->>'threadId') > 0) + OR (${a}.kind = 'slack_file_upload_receipt' AND ${a}.status = 'processed' + AND ${a}.result->>'code' = 'slack_file_upload_identity_confirmed') + OR (${a}.kind = 'slack_session_stop' AND ${a}.status IN ('processed', 'cancelled') + AND ${a}.result->>'code' IN ('slack_session_stop_cancelled', 'slack_session_stop_target_superseded', 'slack_session_stop_no_current_work') + AND NOT EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id::text = ${a}.result->>'runId' + AND (r.company_id <> ${a}.company_id OR r.status NOT IN ('succeeded', 'failed', 'cancelled', 'timed_out', 'interrupted') + OR r.finished_at IS NULL OR r.cost_accounting_pending)))`; +} diff --git a/server/src/services/chat-channels/telegram/idle-history-proof.ts b/server/src/services/chat-channels/telegram/idle-history-proof.ts new file mode 100644 index 00000000000..bd37617d627 --- /dev/null +++ b/server/src/services/chat-channels/telegram/idle-history-proof.ts @@ -0,0 +1,12 @@ +export function completedTelegramAction(a: string): string { + return `(${a}.kind = 'telegram_maintenance' AND ( + (${a}.status = 'processed' AND (${a}.result->>'providerConfirmed' = 'true' + OR ${a}.result->>'code' = 'telegram_stop_subscription_confirmed')) + OR (${a}.status = 'cancelled' AND ${a}.result->>'code' = 'telegram_maintenance_credentials_superseded') + OR (${a}.status = 'failed' AND ${a}.result->>'retryable' = 'false' + AND ${a}.result->>'code' = 'telegram_maintenance_failed'))) + OR (${a}.kind = 'telegram_draft' AND ( + (${a}.status = 'processed' AND ${a}.result->>'phase' = 'published' AND length(${a}.result->>'providerMessageId') > 0) + OR (${a}.status = 'cancelled' AND (${a}.result->>'phase' = 'stopped' + OR ${a}.result->>'code' = 'telegram_draft_presentation_stopped'))))`; +} diff --git a/server/src/services/conversation-idle-proof.ts b/server/src/services/conversation-idle-proof.ts new file mode 100644 index 00000000000..2ce2cf48d0e --- /dev/null +++ b/server/src/services/conversation-idle-proof.ts @@ -0,0 +1,45 @@ +import { finishedIdleRun, finishedIdleWake } from "./idle-work-proof.js"; + +/** The same positive container proof is used by ordinary and scheduled sleep. + * waiting/in_review is set by settleConversationTurn, but is not sufficient: + * a user comment may commit before the wake is enqueued. */ +export function passiveIdleConversation(i: "i"): string { + return `${i}.conversation_agent_id IS NOT NULL AND length(${i}.conversation_user_id) > 0 + AND ${i}.conversation_state = 'waiting' AND ${i}.status = 'in_review' + AND ${i}.assignee_agent_id = ${i}.conversation_agent_id AND ${i}.assignee_user_id IS NULL + AND (${i}.conversation_user_id = 'local-board' OR EXISTS (SELECT 1 FROM company_memberships membership + WHERE membership.company_id = ${i}.company_id AND membership.principal_type = 'user' + AND membership.principal_id = ${i}.conversation_user_id AND membership.status = 'active')) + AND ${i}.conversation_session_generation >= 0 + AND ${i}.checkout_run_id IS NULL AND ${i}.execution_run_id IS NULL + AND ${i}.monitor_next_check_at IS NULL AND ${i}.monitor_wake_requested_at IS NULL + AND (${i}.execution_policy IS NULL OR (jsonb_typeof(${i}.execution_policy) = 'object' + AND COALESCE(${i}.execution_policy->'monitor', 'null'::jsonb) = 'null'::jsonb)) + AND (${i}.execution_state IS NULL OR (jsonb_typeof(${i}.execution_state) = 'object' + AND ${i}.execution_state->>'status' = 'idle' + AND COALESCE(${i}.execution_state->'monitor', 'null'::jsonb) = 'null'::jsonb + AND COALESCE(${i}.execution_state->'currentParticipant', 'null'::jsonb) = 'null'::jsonb + AND COALESCE(${i}.execution_state->'currentStageId', 'null'::jsonb) = 'null'::jsonb + AND COALESCE(${i}.execution_state->'reviewRequest', 'null'::jsonb) = 'null'::jsonb)) + AND EXISTS (SELECT 1 FROM agents a JOIN companies c ON c.id = a.company_id + WHERE a.id = ${i}.conversation_agent_id AND a.company_id = ${i}.company_id + AND c.status = 'active' AND a.status IN ('idle', 'active', 'paused', 'terminated')) + AND NOT EXISTS (SELECT 1 FROM heartbeat_runs r WHERE + (r.native_issue_id = ${i}.id OR r.context_snapshot->>'issueId' = ${i}.id::text OR r.context_snapshot->>'taskId' = ${i}.id::text) + AND (r.company_id <> ${i}.company_id OR r.agent_id <> ${i}.conversation_agent_id OR (${finishedIdleRun('r')}) IS NOT TRUE)) + AND NOT EXISTS (SELECT 1 FROM agent_wakeup_requests w WHERE + (w.payload->>'issueId' = ${i}.id::text OR w.payload->>'taskId' = ${i}.id::text) + AND (w.company_id <> ${i}.company_id OR w.agent_id <> ${i}.conversation_agent_id + OR (${finishedIdleWake('w')}) IS NOT TRUE)) + AND NOT EXISTS (SELECT 1 FROM issue_comments comment WHERE comment.issue_id = ${i}.id + AND comment.deleted_at IS NULL AND comment.author_user_id IS NOT NULL AND comment.author_agent_id IS NULL + AND comment.origin = 'comment' AND comment.client_request_id IS NOT NULL + AND NOT EXISTS (SELECT 1 FROM agent_wakeup_requests w WHERE w.company_id = ${i}.company_id + AND w.agent_id = ${i}.conversation_agent_id AND w.idempotency_key = 'conversation-comment:' || comment.id::text + AND ${finishedIdleWake('w')})) + AND NOT EXISTS (SELECT 1 FROM issue_thread_interactions interaction WHERE interaction.issue_id = ${i}.id + AND (interaction.company_id <> ${i}.company_id OR interaction.status = 'pending')) + AND NOT EXISTS (SELECT 1 FROM chat_completion_deliveries d JOIN chat_task_handoffs h ON h.task_id = d.task_id + WHERE h.conversation_id = ${i}.id AND (h.company_id <> ${i}.company_id OR d.company_id <> ${i}.company_id + OR d.status NOT IN ('delivered', 'superseded', 'exhausted')))`; +} diff --git a/server/src/services/decision-idle-proof.ts b/server/src/services/decision-idle-proof.ts new file mode 100644 index 00000000000..d95f476bbd1 --- /dev/null +++ b/server/src/services/decision-idle-proof.ts @@ -0,0 +1,134 @@ +import { finishedIdleRun } from "./idle-work-proof.js"; + +/** executeDecisionEffects visits every selected ordinal, including failures. + * Its final aggregate cannot prove a missing row or another option's effect. */ +function completedEffectManifest(d: string): string { + const options = `jsonb_array_elements(CASE WHEN jsonb_typeof(${d}.options) = 'array' THEN ${d}.options ELSE '[]'::jsonb END)`; + return `EXISTS (SELECT 1 FROM ${options} option_row WHERE option_row->>'id' = ${d}.chosen_option_id + AND jsonb_typeof(option_row->'effects') = 'array' + AND EXISTS (SELECT 1 FROM activity_log selection WHERE selection.company_id = ${d}.company_id + AND selection.entity_type = 'decision' AND selection.entity_id = ${d}.id::text + AND selection.action IN ('decision.decided', 'decision.dismissed') + AND selection.details->>'chosenOptionId' = ${d}.chosen_option_id + AND selection.details->>'decidedByUserId' = ${d}.decided_by_user_id) + AND (SELECT count(*) FROM ${options} duplicate_option WHERE duplicate_option->>'id' = ${d}.chosen_option_id) = 1 + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(option_row->'effects') WITH ORDINALITY expected(effect, ordinal) + WHERE NOT EXISTS (SELECT 1 FROM decision_effect_executions execution JOIN issues target + ON target.id = execution.target_issue_id AND target.company_id = ${d}.company_id + JOIN activity_log audit ON audit.id = execution.activity_log_id AND audit.company_id = ${d}.company_id + AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text + AND audit.details->>'chosenOptionId' = ${d}.chosen_option_id + AND audit.details->>'executionId' = execution.id::text + WHERE execution.decision_id = ${d}.id AND execution.effect_index = expected.ordinal - 1 + AND execution.effect_type = expected.effect->>'type' AND execution.target_issue_id::text = expected.effect->>'targetIssueId' + AND execution.status IN ('executed', 'failed', 'skipped') AND execution.executed_at IS NOT NULL + AND audit.action = 'decision.effect_' || execution.status)) + AND NOT EXISTS (SELECT 1 FROM decision_effect_executions execution WHERE execution.decision_id = ${d}.id + AND NOT EXISTS (SELECT 1 FROM jsonb_array_elements(option_row->'effects') WITH ORDINALITY expected(effect, ordinal) + WHERE execution.effect_index = expected.ordinal - 1 AND execution.effect_type = expected.effect->>'type' + AND execution.target_issue_id::text = expected.effect->>'targetIssueId')) + AND CASE WHEN ${d}.execution_status = 'succeeded' THEN NOT EXISTS (SELECT 1 FROM decision_effect_executions e + WHERE e.decision_id = ${d}.id AND e.status <> 'executed') + WHEN ${d}.execution_status = 'partial' THEN EXISTS (SELECT 1 FROM decision_effect_executions e + WHERE e.decision_id = ${d}.id AND e.status = 'executed') AND EXISTS (SELECT 1 FROM decision_effect_executions e + WHERE e.decision_id = ${d}.id AND e.status IN ('failed', 'skipped')) + ELSE NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id AND e.status = 'executed') + AND EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id AND e.status IN ('failed', 'skipped')) END)`; +} + +function completedDecisionOutcome(d: string): string { + return `(${d}.status IN ('expired', 'cancelled') AND ${d}.execution_status IS NULL AND ${d}.chosen_option_id IS NULL + AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id) + AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id + AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text AND audit.action = 'decision.' || ${d}.status)) + OR (${d}.status = 'decided' AND ${d}.decided_at IS NOT NULL AND ${d}.execution_status IN ('succeeded', 'partial', 'failed') + AND ((COALESCE(${d}.metadata->>'kind', '') <> 'attention_archive_proposal' AND (${completedEffectManifest(d)})) + OR (${completedArchiveOutcome(d)}) + OR (${d}.chosen_option_id = 'dismissed' AND ${d}.execution_status = 'succeeded' AND ${d}.metadata->>'dismissed' = 'true' + AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id) + AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id + AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text AND audit.action = 'decision.dismissed' + AND audit.details->>'chosenOptionId' = 'dismissed' AND audit.details->>'decidedByUserId' = ${d}.decided_by_user_id)) + OR (${d}.metadata->>'effectRecoveryError' IN ('missing_decider', 'missing_option') + AND ${d}.execution_status IN ('failed', 'partial') + AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id + AND audit.entity_type = 'decision' AND audit.entity_id = ${d}.id::text AND audit.action = 'decision.recovery_failed' + AND audit.details->>'reason' = ${d}.metadata->>'effectRecoveryError' + AND audit.details->>'executionStatus' = ${d}.execution_status + AND audit.details->>'chosenOptionId' IS NOT DISTINCT FROM ${d}.chosen_option_id))))`; +} + +/** Archive applies the reviewed set in one transaction. Every catch is final + * (execution_status=failed); it neither replays the set nor leaves per-effect + * claims. Continuation and notification queues remain independent checks. */ +function completedArchiveOutcome(d: string): string { + return `${d}.metadata->>'kind' = 'attention_archive_proposal' + AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${d}.id) + AND EXISTS (SELECT 1 FROM jsonb_array_elements(CASE WHEN jsonb_typeof(${d}.options) = 'array' + THEN ${d}.options ELSE '[]'::jsonb END) selected WHERE selected->>'id' = ${d}.chosen_option_id + AND selected->'effects' = '[]'::jsonb + AND (SELECT count(*) FROM jsonb_array_elements(${d}.options) duplicate WHERE duplicate->>'id' = ${d}.chosen_option_id) = 1) + AND ((${d}.execution_status = 'failed' AND ${d}.chosen_option_id = 'archive' + AND jsonb_typeof(${d}.metadata->'archiveProposalError') = 'string' + AND length(trim(${d}.metadata->>'archiveProposalError')) > 0) + OR (${d}.execution_status = 'succeeded' AND ${d}.metadata->>'archiveProposalError' IS NULL + AND (${d}.chosen_option_id <> 'archive' OR ( + jsonb_typeof(${d}.target_snapshots) = 'object' AND ${d}.target_snapshots <> '{}'::jsonb + AND ${d}.metadata->>'manifestHash' ~ '^[a-f0-9]{64}$' + AND NOT EXISTS (SELECT 1 FROM jsonb_each(${d}.target_snapshots) snapshot WHERE ( + snapshot.key = 'attention:' || (snapshot.value->'attentionArchive'->>'sourceKind') || ':' || (snapshot.value->'attentionArchive'->>'sourceId') + AND snapshot.value->'attentionArchive'->>'companyId' = ${d}.company_id::text + AND jsonb_typeof(snapshot.value->'attentionArchive'->'expectedVersion') = 'number' + AND snapshot.value->'attentionArchive'->>'expectedVersion' ~ '^[1-9][0-9]*$' + AND EXISTS (SELECT 1 FROM decision_retention retention WHERE retention.company_id = ${d}.company_id + AND retention.source_kind = snapshot.value->'attentionArchive'->>'sourceKind' + AND retention.source_id = snapshot.value->'attentionArchive'->>'sourceId' + AND retention.version::numeric > (snapshot.value->'attentionArchive'->>'expectedVersion')::numeric) + AND EXISTS (SELECT 1 FROM activity_log audit WHERE audit.company_id = ${d}.company_id + AND audit.actor_id = 'decision-archive-proposal' AND audit.action = 'decision_retention.archived' + AND audit.entity_type = 'attention_source' AND audit.entity_id = snapshot.value->'attentionArchive'->>'sourceId' + AND audit.details->>'sourceKind' = snapshot.value->'attentionArchive'->>'sourceKind' + AND audit.created_at >= ${d}.decided_at AND audit.responsible_user_id = ${d}.decided_by_user_id)) IS NOT TRUE)))))`; +} + +function completedDecision(alias: string): string { + return `(${completedDecisionOutcome(alias)}) + AND ${alias}.continuation_policy IN ('none', 'wake_origin_agent') + AND jsonb_typeof(${alias}.metadata) = 'object' + AND COALESCE(${alias}.metadata->>'continuationPending', 'false') = 'false' + AND (${alias}.continuation_policy = 'none' OR ${alias}.metadata->>'continuationDeliveredAt' IS NOT NULL) + AND EXISTS (SELECT 1 FROM heartbeat_runs r JOIN issues i ON i.id = ${alias}.origin_issue_id + AND i.company_id = r.company_id WHERE r.id = ${alias}.origin_run_id + AND r.company_id = ${alias}.company_id AND r.agent_id = ${alias}.origin_agent_id AND ${finishedIdleRun('r')}) + AND NOT EXISTS (SELECT 1 FROM decision_effect_executions e WHERE e.decision_id = ${alias}.id + AND (e.status NOT IN ('executed', 'failed', 'skipped') OR e.executed_at IS NULL))`; +} + +export const idleDecisionHistoryChecks = [ + { reason: "decisions", query: `SELECT 1 FROM decisions d WHERE (${completedDecision('d')}) IS NOT TRUE` }, + { reason: "decision_effect_executions", query: `SELECT 1 FROM decision_effect_executions e WHERE ( + e.status IN ('executed', 'failed', 'skipped') AND e.executed_at IS NOT NULL + AND EXISTS (SELECT 1 FROM decisions d JOIN issues i ON i.id = e.target_issue_id AND i.company_id = d.company_id + WHERE d.id = e.decision_id AND (${completedDecision('d')}))) IS NOT TRUE` }, + { reason: "decision_archive_notification_outbox", query: `SELECT 1 FROM decision_archive_notification_outbox o WHERE ( + o.status = 'delivered' AND o.delivered_at IS NOT NULL + AND o.archive_version > 0 AND EXISTS (SELECT 1 FROM decision_retention retention + WHERE retention.company_id = o.company_id AND retention.source_kind = o.source_kind + AND retention.source_id = o.source_id AND retention.archive_version >= o.archive_version) + AND EXISTS (SELECT 1 FROM agents a JOIN issues i ON i.company_id = a.company_id + WHERE a.id = o.origin_agent_id AND i.id = o.origin_issue_id AND i.company_id = o.company_id)) IS NOT TRUE` }, + { reason: "company_secret_proposals", query: `SELECT 1 FROM company_secret_proposals p WHERE ( + p.kind IN ('secret', 'binding') AND p.status IN ('approved', 'rejected', 'withdrawn', 'expired') + AND p.resolved_at IS NOT NULL AND p.value_ciphertext IS NULL AND p.ciphertext_scrubbed_at IS NOT NULL + AND EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = p.origin_run_id + AND r.company_id = p.company_id AND r.agent_id = p.proposed_by_agent_id AND ${finishedIdleRun('r')}) + AND (p.origin_issue_id IS NULL OR EXISTS (SELECT 1 FROM issues i + WHERE i.id = p.origin_issue_id AND i.company_id = p.company_id)) + AND (p.status <> 'approved' OR (p.kind = 'secret' AND EXISTS (SELECT 1 FROM company_secrets s + WHERE s.id = p.created_secret_id AND s.company_id = p.company_id)) + OR (p.kind = 'binding' AND p.applied_binding_config_path = p.config_path AND EXISTS ( + SELECT 1 FROM agents a WHERE a.id = p.target_id AND a.company_id = p.company_id))) + AND (p.interaction_id IS NULL OR EXISTS (SELECT 1 FROM issue_thread_interactions i + WHERE i.id = p.interaction_id AND i.company_id = p.company_id AND i.issue_id = p.origin_issue_id + AND i.status IN ('accepted', 'rejected', 'cancelled', 'expired') AND i.resolved_at IS NOT NULL))) IS NOT TRUE` }, +] as const; diff --git a/server/src/services/idle-issue-schedule.ts b/server/src/services/idle-issue-schedule.ts index c234403e873..f2ed5efb01e 100644 --- a/server/src/services/idle-issue-schedule.ts +++ b/server/src/services/idle-issue-schedule.ts @@ -5,6 +5,7 @@ import { AGENT_STATUSES, issueExecutionMonitorPolicySchema, PROVIDER_QUOTA_MONIT import type { ScheduledWake } from "./idle-scheduled-work.js"; import { onboardingFirstTaskAwaitingUserSql } from "./recovery/onboarding-first-task-wait.js"; import { completedAgentUnblockRequest, responseDrivenIssueWait } from "./idle-work-proof.js"; +import { passiveIdleConversation } from "./conversation-idle-proof.js"; const processGeneration = randomUUID(); const limit = 10_000; @@ -30,6 +31,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st untouched_onboarding: boolean; response_driven_wait: boolean; unblock_request_finished: boolean; response_wait_generation: unknown; + passive_conversation: boolean; }>(sql`SELECT i.id, i.company_id, i.status, i.updated_at, i.assignee_agent_id, i.assignee_user_id, i.checkout_run_id, i.execution_run_id, i.execution_policy, i.execution_state, i.unblock_descriptor, i.conversation_agent_id, i.monitor_next_check_at, i.monitor_wake_requested_at, i.monitor_attempt_count, @@ -38,6 +40,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st (i.assignee_agent_id IS NULL OR a.id IS NOT NULL) AS scoped, (${sql.raw(responseDrivenIssueWait('i'))}) AS response_driven_wait, (${sql.raw(completedAgentUnblockRequest('i'))}) AS unblock_request_finished, + (${sql.raw(passiveIdleConversation('i'))}) AS passive_conversation, (SELECT jsonb_agg(jsonb_build_array(w.id, w.updated_at, w.kind, w.status, w.source_run_id, w.effective_resolver_policy, w.continuation_policy, w.addressee_agent_id, w.addressee_user_id) ORDER BY w.id) FROM issue_thread_interactions w WHERE w.issue_id = i.id AND w.status = 'pending') AS response_wait_generation, @@ -61,7 +64,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st if (!statuses.has(row.status) || row.company_status !== "active" || row.scoped !== true || (row.assignee_agent_id && !agentStatuses.has(row.agent_status ?? "")) || row.unsafe_dependencies !== false || row.checkout_run_id || row.execution_run_id - || row.conversation_agent_id || row.monitor_wake_requested_at + || (row.conversation_agent_id && row.passive_conversation !== true) || row.monitor_wake_requested_at || (row.execution_state !== null && state?.status !== "idle") || (row.execution_policy !== null && !policy)) return { blocked: true }; if (row.status === "blocked" && row.dependency_ready) return { blocked: true }; @@ -81,7 +84,7 @@ export async function readIdleIssueSchedule(tx: Pick, ownerId: st } if (policy?.monitor != null) return { blocked: true }; const unblock = record(row.unblock_descriptor); - const passive = row.status === "backlog" + const passive = row.passive_conversation === true || row.status === "backlog" || (row.status === "todo" && !row.assignee_agent_id) || (row.status === "todo" && row.assignee_agent_id && row.agent_status === "idle" && row.untouched_onboarding === true) || (row.status === "blocked" && (row.unblock_descriptor === null || unblock?.owner === "board" diff --git a/server/src/services/idle-sleep-safety.ts b/server/src/services/idle-sleep-safety.ts index c0d58a64bdd..f93b158c7f1 100644 --- a/server/src/services/idle-sleep-safety.ts +++ b/server/src/services/idle-sleep-safety.ts @@ -9,6 +9,23 @@ import { verifiedResourceCleanup } from "./cleanup-verification-proof.js"; import { companyHoldMismatch } from "../modules/agent-lifecycle/index.js"; import { pendingBudgetRunCancellationPredicate } from "./heartbeat/budget-cancellation.js"; import { nativeResponsePresentationPendingQuery } from "./native-runtime/native-response-presentation-query.js"; +import { passiveIdleConversation } from "./conversation-idle-proof.js"; +import { idleChatHistoryChecks } from "./chat-channels/idle-history-proof.js"; +import { idleToolActionHistoryChecks } from "./tool-action-idle-proof.js"; +import { idleBrowserHistoryChecks } from "./browser-use-idle-proof.js"; +import { idleDecisionHistoryChecks } from "./decision-idle-proof.js"; +import { idleWorkspaceHistoryChecks } from "./workspace-history-idle-proof.js"; +import { idleMcpHistoryChecks } from "./public-mcp/idle-history-proof.js"; +import { idleExternalAgentHistoryChecks } from "../modules/external-agents/idle-history-proof.js"; + +export const completedHistoryChecks = [ + ...idleChatHistoryChecks, ...idleToolActionHistoryChecks, ...idleBrowserHistoryChecks, + ...idleDecisionHistoryChecks, ...idleWorkspaceHistoryChecks, ...idleMcpHistoryChecks, + ...idleExternalAgentHistoryChecks, +] as const; + +const ordinaryIssues = `SELECT 1 FROM issues i WHERE i.status NOT IN ('done', 'cancelled') + AND (${passiveIdleConversation('i')}) IS NOT TRUE`; export interface IdleSleepDrainStatus { ownerId?: string; @@ -74,7 +91,7 @@ const WORK_CHECKS = [ SELECT 1 FROM heartbeat_runs r WHERE r.id = w.run_id AND r.company_id = w.company_id AND r.agent_id = w.agent_id AND ${finishedRun("r")})))`, - `SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`, + ordinaryIssues, // A completed issue can still need its first watchdog review, including a // retry after immediate evaluation failed before creating a review/run. `SELECT 1 FROM issue_watchdogs WHERE status <> 'disabled'`, @@ -259,23 +276,7 @@ const WORK_CHECKS = [ OR effect.delivery_state <> 'delivered' OR effect.delivered_at IS NULL OR effect.next_attempt_at IS NOT NULL)) ) IS NOT TRUE`, - // These less common work sources fail closed on any retained state. Their - // terminal-state exceptions can be added with tests for the owning service. - ...[ - "chat_deliveries", "chat_publications", "chat_actions", - "chat_completion_deliveries", "tool_action_requests", "tool_action_deliveries", - "workspace_runtime_services", - "execution_workspace_runtime_leases", - "company_transfer_runs", "decisions", "decision_effect_executions", - "decision_archive_notification_outbox", "browser_use_sessions", "browser_use_runs", - "browser_use_browsers", - "company_secret_proposals", - "mcp_oauth_grants", "mcp_mutation_receipts", "mcp_event_subscriptions", - "mcp_event_deliveries", "mcp_attachment_uploads", "dot_agent_bindings", - "dot_runner_assignments", "dot_runner_operations", "dot_mailbox_items", - "muse_agent_bindings", "muse_credentials", "muse_runner_assignments", - "muse_runner_operations", "muse_mailbox_items", "muse_input_deliveries", "external_agent_holds", - ].map((table) => `SELECT 1 FROM ${table}`), + ...completedHistoryChecks.map(check => check.query), ] as const; function sameQuietHold(before: IdleSleepDrainStatus, after: IdleSleepDrainStatus, now: number): boolean { @@ -320,7 +321,7 @@ export async function readIdleSleepSafety( await tx.execute(sql`SET LOCAL statement_timeout = '5s'`); const checks = WORK_CHECKS.filter(query => (!scheduledRoutines || query !== `SELECT 1 FROM routines WHERE status NOT IN ('paused', 'archived')`) - && (!scheduledIssueMonitors || query !== `SELECT 1 FROM issues WHERE status NOT IN ('done', 'cancelled')`)) + && (!scheduledIssueMonitors || query !== ordinaryIssues)) .map((query) => sql`EXISTS (${sql.raw(query)})`); checks.push(sql`EXISTS (SELECT 1 FROM agents JOIN companies ON companies.id = agents.company_id WHERE ${companyHoldMismatch()})`); diff --git a/server/src/services/public-mcp/idle-history-proof.ts b/server/src/services/public-mcp/idle-history-proof.ts new file mode 100644 index 00000000000..b3c7f78abd1 --- /dev/null +++ b/server/src/services/public-mcp/idle-history-proof.ts @@ -0,0 +1,37 @@ +/** Completion receipts are retained for idempotency. Fixed aliases only. */ +export const idleMcpHistoryChecks = [ + { reason: "mcp_oauth_grants", query: `SELECT 1 FROM mcp_oauth_grants g WHERE + (g.revoked_at IS NOT NULL AND EXISTS (SELECT 1 FROM companies c WHERE c.id = g.company_id)) IS NOT TRUE` }, + { reason: "mcp_mutation_receipts", query: `SELECT 1 FROM mcp_mutation_receipts m WHERE ( + m.status = 'completed' AND jsonb_typeof(m.result) = 'object' + AND EXISTS (SELECT 1 FROM mcp_oauth_grants g WHERE g.id = m.grant_id + AND g.company_id = m.company_id AND g.user_id = m.user_id)) IS NOT TRUE` }, + // Stopped subscriptions no longer scan for events. Expiry alone is not a + // handoff of responsibility for active subscriptions or unfinished delivery. + { reason: "mcp_event_subscriptions", query: `SELECT 1 FROM mcp_event_subscriptions s WHERE ( + s.stopped_at IS NOT NULL AND EXISTS (SELECT 1 FROM mcp_oauth_grants g + WHERE g.id = s.grant_id AND g.company_id = s.company_id)) IS NOT TRUE` }, + { reason: "mcp_event_deliveries", query: `SELECT 1 FROM mcp_event_deliveries d WHERE ( + d.finished_at IS NOT NULL AND (d.outcome IN ('delivered', 'filtered', 'inactive', 'delivery_exhausted') + OR (d.outcome ~ '^http_[1-5][0-9][0-9]$' AND d.outcome !~ '^http_2' + AND (d.attempts >= 6 OR (d.outcome NOT IN ('http_408', 'http_429') AND d.outcome !~ '^http_5')))) + AND EXISTS (SELECT 1 FROM mcp_event_subscriptions s WHERE s.id = d.subscription_id + AND EXISTS (SELECT 1 FROM mcp_oauth_grants g WHERE g.id = s.grant_id AND g.company_id = s.company_id) + AND ((d.activity_id IS NOT NULL AND d.mailbox_item_id IS NULL AND EXISTS ( + SELECT 1 FROM activity_log a WHERE a.id = d.activity_id AND a.company_id = s.company_id)) + OR (d.mailbox_item_id IS NOT NULL AND d.activity_id IS NULL AND EXISTS ( + SELECT 1 FROM dot_mailbox_items m WHERE m.id = d.mailbox_item_id + AND m.company_id = s.company_id AND m.binding_id = s.binding_id))))) IS NOT TRUE` }, + // The upload owner allocates object identity before storage writes and only + // stamps cleaned_at after deletion. A committed attachment is never deleted. + { reason: "mcp_attachment_uploads", query: `SELECT 1 FROM mcp_attachment_uploads u WHERE ( + EXISTS (SELECT 1 FROM issues i WHERE i.id = u.task_id AND i.company_id = u.company_id) + AND (u.grant_id IS NULL OR EXISTS (SELECT 1 FROM mcp_oauth_grants g WHERE g.id = u.grant_id + AND g.company_id = u.company_id AND g.user_id = u.user_id)) + AND ((u.attachment_id IS NULL AND u.cleaned_at IS NOT NULL) + OR (u.cleaned_at IS NULL AND u.attachment_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM issue_attachments a JOIN assets stored ON stored.id = a.asset_id + WHERE a.id = u.attachment_id AND a.company_id = u.company_id AND a.issue_id = u.task_id + AND stored.company_id = u.company_id AND stored.object_key = u.object_key + AND stored.provider = u.storage_provider)))) IS NOT TRUE` }, +] as const; diff --git a/server/src/services/tool-action-idle-proof.ts b/server/src/services/tool-action-idle-proof.ts new file mode 100644 index 00000000000..a000039d235 --- /dev/null +++ b/server/src/services/tool-action-idle-proof.ts @@ -0,0 +1,28 @@ +import { finishedIdleRun } from "./idle-work-proof.js"; + +function completedRequest(a: string): string { + return `${a}.status IN ('executed', 'failed', 'rejected', 'expired', 'cancelled') AND ${a}.resolved_at IS NOT NULL + AND EXISTS (SELECT 1 FROM tool_invocations v WHERE v.id = ${a}.invocation_id AND v.company_id = ${a}.company_id + AND v.issue_id IS NOT DISTINCT FROM ${a}.issue_id AND v.completed_at IS NOT NULL + AND v.status IN ('succeeded', 'failed', 'denied', 'cancelled', 'timed_out', 'rate_limited') + AND (v.run_id IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = v.run_id + AND r.company_id = v.company_id AND ${finishedIdleRun('r')}))) + AND (${a}.interaction_id IS NULL OR EXISTS (SELECT 1 FROM issue_thread_interactions i WHERE i.id = ${a}.interaction_id + AND i.company_id = ${a}.company_id AND i.issue_id = ${a}.issue_id AND i.resolved_at IS NOT NULL + AND i.payload->'toolAction'->>'actionRequestId' = ${a}.id::text + AND i.payload->'toolAction'->>'invocationId' = ${a}.invocation_id::text + AND i.status IN ('accepted', 'rejected', 'expired', 'cancelled') + AND (${a}.status = 'rejected' OR i.result->'toolAction'->>'status' = CASE + WHEN ${a}.status = 'cancelled' THEN 'expired' ELSE ${a}.status END)))`; +} + +export const idleToolActionHistoryChecks = [ + { reason: "tool_action_requests", query: `SELECT 1 FROM tool_action_requests a WHERE (${completedRequest('a')}) IS NOT TRUE` }, + // delivered_at is the domain's acknowledgment. It also retires outcomes + // after issue completion/reassignment. Accepted continuations stay covered by + // the global wake/run checks, including coalesced run accounting. + { reason: "tool_action_deliveries", query: `SELECT 1 FROM tool_action_deliveries d WHERE ( + d.delivered_at IS NOT NULL AND EXISTS (SELECT 1 FROM tool_action_requests a + WHERE a.id = d.action_request_id AND a.company_id = d.company_id + AND a.issue_id = d.issue_id AND a.interaction_id = d.interaction_id AND (${completedRequest('a')}))) 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 new file mode 100644 index 00000000000..534968b4ca3 --- /dev/null +++ b/server/src/services/workspace-history-idle-proof.ts @@ -0,0 +1,37 @@ +import { finishedIdleRun } from "./idle-work-proof.js"; + +export const idleWorkspaceHistoryChecks = [ + // Failed imports are intentionally resumable. Completed/cancelled rows can + // still owe spool deletion, independently inspected by idle-local-work. + { reason: "company_transfer_runs", query: `SELECT 1 FROM company_transfer_runs t WHERE ( + 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.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 ( + s.status = 'stopped' AND s.stopped_at IS NOT NULL AND s.provider = 'local_process' + AND s.exposure_handle IS NULL + AND (s.exposure IS NULL OR (jsonb_typeof(s.exposure) = 'object' + AND s.exposure->>'state' = 'removed' AND s.url IS NULL + AND COALESCE(s.exposure->'publicUrl', 'null'::jsonb) = 'null'::jsonb)) + AND (s.execution_workspace_id IS NULL OR EXISTS (SELECT 1 FROM execution_workspaces w + WHERE w.id = s.execution_workspace_id AND w.company_id = s.company_id)) + AND (s.issue_id IS NULL OR EXISTS (SELECT 1 FROM issues i WHERE i.id = s.issue_id AND i.company_id = s.company_id)) + AND (s.started_by_run_id IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs r + WHERE r.id = s.started_by_run_id AND r.company_id = s.company_id AND ${finishedIdleRun('r')}))) IS NOT TRUE` }, + // This is an exclusivity row, not a provider lease. The domain only + // evaluates staleness on the next claim; no expiry job is owed by history. + { reason: "execution_workspace_runtime_leases", query: `SELECT 1 FROM execution_workspace_runtime_leases l WHERE ( + EXISTS (SELECT 1 FROM execution_workspaces w WHERE w.id = l.execution_workspace_id AND w.company_id = l.company_id) + AND ((l.owner_issue_id IS NOT NULL AND l.owner_key = 'issue:' || l.owner_issue_id::text + AND EXISTS (SELECT 1 FROM issues i WHERE i.id = l.owner_issue_id AND i.company_id = l.company_id + AND i.status IN ('done', 'cancelled'))) + OR (l.owner_issue_id IS NULL AND l.owner_run_id IS NOT NULL AND l.owner_key = 'run:' || l.owner_run_id::text + AND EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = l.owner_run_id AND r.company_id = l.company_id + AND r.agent_id = l.owner_agent_id AND ${finishedIdleRun('r')}))) + AND (l.owner_run_id IS NULL OR EXISTS (SELECT 1 FROM heartbeat_runs r WHERE r.id = l.owner_run_id + AND r.company_id = l.company_id AND r.agent_id = l.owner_agent_id AND ${finishedIdleRun('r')})) + AND NOT EXISTS (SELECT 1 FROM workspace_runtime_services s WHERE s.execution_workspace_id = l.execution_workspace_id + AND (s.company_id <> l.company_id OR s.status <> 'stopped' OR s.stopped_at IS NULL))) IS NOT TRUE` }, +] as const; diff --git a/server/src/services/workspace-runtime.ts b/server/src/services/workspace-runtime.ts index 96f0b39edff..f2b1d46ba0e 100644 --- a/server/src/services/workspace-runtime.ts +++ b/server/src/services/workspace-runtime.ts @@ -46,6 +46,7 @@ import { import { and, desc, eq, gte, inArray, isNull, lte, ne, or, sql } from "drizzle-orm"; import { asNumber, asString, parseObject, renderTemplate } from "../adapters/utils.js"; import { conflict, unprocessable } from "../errors.js"; +import { registerIdleLocalWorkProbe, type IdleLocalWork } from "./idle-local-work.js"; import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifecycle-lock.js"; import { resolveHomeAwarePath } from "../home-paths.js"; import { hasVerifiedWorktreeSeedManifest, isVerifiedWorktreeSeedManifest } from "../worktree-seed-manifest.js"; @@ -283,6 +284,22 @@ const quarantinedRuntimeExposurePorts = new Set(); * companion together or not at all. */ const exposurePortPairClaims = new ExposurePortPairClaims(); +let runtimeReconstructionComplete = false; +let runtimeReconstructionEpoch = 0; +let runtimeReconciliation: ReturnType | null = null; + +/** Read owner memory only. Durable stopped rows do not describe a surviving + * 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 + || runtimeReplacementClaimsByReuseKey.size || quarantinedRuntimeExposurePorts.size + || exposurePortPairClaims.activePorts().size + || [...inFlightAllocatedPorts.values()].some(until => until > Date.now()) ? "present" : "none"; +} +// This registry and its probe share the application process lifetime, like the +// native-session owner. Stopping one resource must not unregister its siblings. +registerIdleLocalWorkProbe(readWorkspaceRuntimeIdleWork); /** * Execution-workspace statuses that still hold an exclusive lease. `archived` * is the only terminal state; everything else — including `idle` — is a lane an @@ -525,6 +542,8 @@ type ProcessOutputAccumulator = { export async function resetRuntimeServicesForTests( opts: { terminateProcesses?: boolean; simulateSupervisorExit?: boolean } = {}, ) { + runtimeReconstructionComplete = false; + runtimeReconstructionEpoch += 1; if (opts.terminateProcesses) { for (const serviceId of [...runtimeServicesById.keys()]) { await stopRuntimeService(serviceId).catch(() => undefined); @@ -8577,6 +8596,19 @@ export async function refreshPersistedRuntimeServiceHealth(input: { } export async function reconcilePersistedRuntimeServicesOnStartup(db: Db) { + if (runtimeReconciliation) return runtimeReconciliation; + runtimeReconstructionComplete = false; + const epoch = ++runtimeReconstructionEpoch; + runtimeReconciliation = reconstructRuntimeServices(db).then(result => { + if (epoch === runtimeReconstructionEpoch) { + runtimeReconstructionComplete = result.restartFailed === 0 && result.exposureReservationDrift.length === 0; + } + return result; + }).finally(() => { runtimeReconciliation = null; }); + return runtimeReconciliation; +} + +async function reconstructRuntimeServices(db: Db) { const rows = await db .select() .from(workspaceRuntimeServices)