mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 12:29:37 +01:00
feat(server): add domain-owned retained-history sleep proofs
Keep finished delivery and resource history while preserving its downstream work, accounting, cleanup and transport barriers. Share the narrow waiting conversation proof across ordinary and scheduled inspection. Retain local runtime resources through owner reconstruction and process-local inspection. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed
1 parent
0748b90062
commit
7f454eb377
16 files changed
+1205
-21
No files matched your search
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
|
||||
let db: ReturnType<typeof tables.createDb>;
|
||||
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<ReturnType<typeof seed>>) {
|
||||
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<ReturnType<typeof seed>>, 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<ReturnType<typeof seed>>) {
|
||||
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<ReturnType<typeof supervisor.findLocalServiceRegistryRecordByRuntimeServiceId>> = 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<void>((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<void>((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);
|
||||
});
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
@@ -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')`;
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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)))`;
|
||||
}
|
||||
@@ -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'))))`;
|
||||
}
|
||||
@@ -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')))`;
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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<Db, "execute">, 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<Db, "execute">, 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<Db, "execute">, 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<Db, "execute">, 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"
|
||||
|
||||
@@ -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()})`);
|
||||
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
@@ -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;
|
||||
@@ -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<number>();
|
||||
* companion together or not at all.
|
||||
*/
|
||||
const exposurePortPairClaims = new ExposurePortPairClaims();
|
||||
let runtimeReconstructionComplete = false;
|
||||
let runtimeReconstructionEpoch = 0;
|
||||
let runtimeReconciliation: ReturnType<typeof reconstructRuntimeServices> | 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)
|
||||
|
||||
Reference in new issue
Block a user