diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index 2057e8ee56b..3ac90be6d7d 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -118,6 +118,20 @@ export function nativeRestartInterruptedTurnId(snapshot: Pick): boolean { + const activeTurnId = snapshot.activeTurnId; + if (typeof activeTurnId !== "string" || !activeTurnId) return false; + const last = Array.isArray(snapshot.terminalTurns) ? snapshot.terminalTurns.at(-1) : null; + if (!last || typeof last.fingerprint !== "string" || !last.turnId || last.turnId === activeTurnId) return false; + try { + const fingerprint = JSON.parse(last.fingerprint); + return fingerprint?.terminalState === "failed" && fingerprint.result === null && + isNativeRestartInterruption(fingerprint.error); + } catch { return false; } +} + /** * Cancellation is a synchronous authority transition followed by passive * provider cleanup. Once `cancel` returns, the provider session must no longer diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 73bff682ae6..9d40f756c99 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -22,7 +22,7 @@ import { NativeSessionCloseUnrecoverableError, NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, - nativeRestartInterruptedTurnId, + nativeRestartContinuationInFlight, nativeRestartInterruptedTurnId, } from "./contracts/native-session-backend.js"; import { PROVIDER_ATTACH_REFUSED, PROVIDER_SIDE_EFFECT_PENDING, PROVIDER_STATE_INCOMPATIBLE } from "./contracts/harness-driver.js"; import type { @@ -642,6 +642,53 @@ describe("executeNativeSession recovery", () => { expect(events.filter(event => event.eventType === "turn.failed").map(event => event.turnId)).toEqual(["turn-unrecorded-continuation"]); }); + it("recovers a restart continuation whose checkpoint still carries the lost turn's failed terminal", async () => { + const failure = { code: "provider_turn_lost_on_restore", recoverable: true }; + const checkpoint: PersistedNativeSession = { + backendKind: "runner", driverKind: "codex_app_server", sessionId: "thread-recovery", identity, + providerSessionId: "provider-recovery", cursor: "0", activeTurnId: "turn-continuation", + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "failed", turnTerminalState: "failed", reportedWorkDisposition: "yielded" }, + terminalTurns: [{ turnId: "turn-recovery", fingerprint: JSON.stringify({ terminalState: "failed", error: failure, result: null }) }], + semanticResult: null, pendingRuntimeRequests: [], lineage: [], + dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "turn-continuation", + }; + const startTurn = vi.fn(async () => ({ turnId: "turn-unexpected" })); + const session: NativeSession = { + identity: () => identity, + async capabilities() { return { resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }; }, + async *events() { + yield { ...runnerEvent(1, "turn.completed", { status: "completed" }), turnId: "turn-continuation" }; + }, + startTurn, + async result() { return { result, terminal, turnId: "turn-continuation" }; }, + async snapshot() { return structuredClone(checkpoint); }, + async close() {}, + }; + const backend: NativeSessionBackend = { + async descriptor() { return { kind: "runner", name: "restart", version: "1", capabilities: await session.capabilities() }; }, + async openSession() { throw new Error("must preserve the provider session"); }, + async recoverSession() { return { recovered: true, session }; }, + }; + const port: ControlPlanePort = { + async openRun() {}, + async checkpointSession() {}, + async appendEvent(event: PrpEvent) { return { cursor: event.sourceSeq, highestContiguousSourceSeq: event.sourceSeq, disposition: "committed" }; }, + async replayEvents() { return { events: [], highestContiguousSourceSeq: 0 }; }, + async completeRun() {}, + }; + await expect(executeNativeSession({ + input, backend, controlPlane: port, persistedSession: checkpoint, + runnerInstanceId: "runner-recovery", controlPlaneInstanceId: "control-recovery", + resumeInterruptedTurn: true, + })).resolves.toMatchObject({ result, turnId: "turn-continuation", terminal: { runTerminalState: "succeeded" } }); + expect(startTurn).not.toHaveBeenCalled(); + expect(nativeRestartContinuationInFlight(checkpoint)).toBe(true); + expect(nativeRestartContinuationInFlight({ ...checkpoint, activeTurnId: "turn-recovery" })).toBe(false); + expect(nativeRestartContinuationInFlight({ ...checkpoint, terminalTurns: [{ turnId: "turn-recovery", + fingerprint: JSON.stringify({ terminalState: "failed", error: { code: "model_not_found", recoverable: false }, result: null }) }] })) + .toBe(false); + }); + it.each(["acpx_runtime", "opencode_server"] as const)("continues a restart-lost %s turn once in the same conversation", async (driverKind) => { const qualified = resolveQualifiedAcpxProfile("claude", "claude-sonnet-5"); const acpxInput: NativeExecutionInputV1 = (driverKind === "acpx_runtime" ? { diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 180521404b6..2e8bcb37074 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -33,6 +33,7 @@ import { NativeSessionCleanupQuarantinedError, NativeSessionProtocolIntegrityError, isNativeRestartInterruption, + nativeRestartContinuationInFlight, nativeRestartInterruptedTurnId, } from "./contracts/native-session-backend.js"; import { @@ -2327,7 +2328,7 @@ export async function executeNativeSession( } | null = null; let reconciledRecoveryCheckpoint: PersistedNativeSession | null = null; const checkpointedInterruption = persistedSession && nativeRestartInterruptedTurnId(persistedSession); - if (checkpointedInterruption && persistedSession) { + if (persistedSession && (checkpointedInterruption || nativeRestartContinuationInFlight(persistedSession))) { // The turn failed because its process vanished; the conversation is still // usable. Keep its terminal fingerprint so replay cannot execute it twice. persistedSession = { ...persistedSession, terminal: null }; diff --git a/packages/shared/src/types/skill-source.ts b/packages/shared/src/types/skill-source.ts index f92df553fb8..f34624008f2 100644 --- a/packages/shared/src/types/skill-source.ts +++ b/packages/shared/src/types/skill-source.ts @@ -43,6 +43,30 @@ export interface SkillSourceFilePreview { truncated: boolean; commitSha: string; } +/** One file that differs between the installed and latest tracked source revisions. */ +export interface SkillSourceUpdateDiffFile { + path: string; + /** `not_compared`: the scan left this path out of at least one revision (too large, a symlink, or rejected by the audit). */ + change: 'added' | 'removed' | 'modified' | 'not_compared'; + binary: boolean; + /** The file's executable bit differs between the two revisions. */ + executableChanged: boolean; + /** Unified line diff ("-" installed, "+" latest); "" when only the executable bit changed; null when unavailable. */ + diff: string | null; + /** Why `diff` is null: binary content, a file too large to diff, a path left out of a snapshot, or the response budget ran out. */ + diffUnavailable: 'binary' | 'too_large' | 'omitted' | 'budget' | null; +} +export interface SkillSourceUpdateDiff { + supported: boolean; + reason: string | null; + currentRef: string | null; + latestRef: string | null; + /** False when the installed revision is already the latest one. */ + hasUpdate: boolean; + files: SkillSourceUpdateDiffFile[]; + /** Some files were not listed or carry no diff because the response reached its size or work budget. */ + truncated: boolean; +} export interface SkillSourceEntry { id: string; sourceId: string; diff --git a/server/src/__tests__/agent-permissions-routes.test.ts b/server/src/__tests__/agent-permissions-routes.test.ts index c5ccd4673e3..991a906a93f 100644 --- a/server/src/__tests__/agent-permissions-routes.test.ts +++ b/server/src/__tests__/agent-permissions-routes.test.ts @@ -88,6 +88,7 @@ const mockHeartbeatService = vi.hoisted(() => ({ getRun: vi.fn(), cancelRun: vi.fn(), cancelInvocationsForAgents: vi.fn(), + cancelActiveForAgent: vi.fn(), })); const mockIssueApprovalService = vi.hoisted(() => ({ @@ -1790,6 +1791,125 @@ describe("agent permission routes", () => { expect(mockAgentService.updateAndTransition).toHaveBeenCalledWith(agentId, "terminate", expect.anything(), expect.anything()); }); + it.each([ + ["PATCH status", (base: string) => request(base).patch(`/api/agents/${agentId}`).send({ status: "paused" })], + ["POST pause", (base: string) => request(base).post(`/api/agents/${agentId}/pause`).send({})], + ])("keeps pausing an agent board-only through %s, even for an agent allowed to configure the target", async (_route, send) => { + mockAgentService.getById.mockResolvedValue({ ...baseAgent, status: "running" }); + const app = await createApp({ + type: "agent", + agentId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + companyId, + source: "agent_key", + runId: "run-1", + }); + + const res = await requestApp(app, send); + + expect(res.status, JSON.stringify(res.body)).toBe(403); + expect(res.body.error).toBe("Board access required"); + expect(res.body.code).toBe("board_access_required"); + expect(res.body.remediation).toContain("this denial is final"); + expect(res.body.remediation).toContain(`[Open the agent's profile](//agents/${agentId})`); + expect(mockAgentService.updateAndTransition).not.toHaveBeenCalled(); + expect(mockAgentService.update).not.toHaveBeenCalled(); + expect(mockHeartbeatService.cancelActiveForAgent).not.toHaveBeenCalled(); + expect(mockLogActivity).not.toHaveBeenCalled(); + }); + + it("still lets a board user pause through PATCH status", async () => { + mockAgentService.getById.mockResolvedValue({ ...baseAgent, status: "idle" }); + mockAgentService.updateAndTransition.mockResolvedValue({ ...baseAgent, status: "paused" }); + const app = await createApp({ + type: "board", + userId: "board-user", + source: "local_implicit", + isInstanceAdmin: true, + companyIds: [companyId], + }); + + const res = await requestApp(app, (baseUrl) => request(baseUrl) + .patch(`/api/agents/${agentId}`) + .send({ status: "paused" })); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(mockAgentService.updateAndTransition).toHaveBeenCalledWith(agentId, "pause", expect.anything(), expect.anything()); + }); + + it("applies the resume change-grant decision to PATCH status idle from an agent", async () => { + mockAgentService.getById.mockResolvedValue({ ...baseAgent, status: "paused" }); + mockAccessService.decide.mockImplementation(async (input: { action?: string; scope?: { requiresChangeGrant?: boolean } }) => { + const allowed = !input.scope?.requiresChangeGrant; + return { + allowed, + action: input.action, + reason: allowed ? "allow_explicit_grant" : "deny_missing_consent", + explanation: allowed ? "Allowed by test grant" : "Accepted consent is required for this suggested change.", + }; + }); + const app = await createApp({ + type: "agent", + agentId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + companyId, + source: "agent_key", + runId: "run-1", + }); + + const res = await requestApp(app, (baseUrl) => request(baseUrl) + .patch(`/api/agents/${agentId}`) + .send({ status: "idle" })); + + expect(res.status, JSON.stringify(res.body)).toBe(403); + expect(mockAccessService.decide).toHaveBeenCalledWith(expect.objectContaining({ + action: "agent_config:update", + scope: { requiresChangeGrant: true }, + })); + expect(mockAgentService.updateAndTransition).not.toHaveBeenCalled(); + expect(mockLogActivity).not.toHaveBeenCalled(); + }); + + it("lets a granted agent resume through PATCH status idle", async () => { + mockAgentService.getById.mockResolvedValue({ ...baseAgent, status: "paused" }); + mockAgentService.updateAndTransition.mockResolvedValue({ ...baseAgent, status: "idle" }); + const app = await createApp({ + type: "agent", + agentId: "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + companyId, + source: "agent_key", + runId: "run-1", + }); + + const res = await requestApp(app, (baseUrl) => request(baseUrl) + .patch(`/api/agents/${agentId}`) + .send({ status: "idle" })); + + expect(res.status, JSON.stringify(res.body)).toBe(200); + expect(mockAgentService.updateAndTransition).toHaveBeenCalledWith(agentId, "resume", expect.anything(), expect.anything()); + }); + + it("keeps the invalid-org-chain resume guard on PATCH status idle", async () => { + mockAgentService.getById.mockResolvedValue({ + ...baseAgent, + status: "paused", + orgChainHealth: { status: "invalid_org_chain", reason: "missing_manager", repairGuidance: "Repair the reporting chain first." }, + }); + const app = await createApp({ + type: "board", + userId: "board-user", + source: "local_implicit", + isInstanceAdmin: true, + companyIds: [companyId], + }); + + const res = await requestApp(app, (baseUrl) => request(baseUrl) + .patch(`/api/agents/${agentId}`) + .send({ status: "idle" })); + + expect(res.status).toBe(409); + expect(res.body.error).toBe("Repair the reporting chain first."); + expect(mockAgentService.updateAndTransition).not.toHaveBeenCalled(); + }); + it("terminates directly when no open hire approval is linked", async () => { const idleAgent = { ...baseAgent, status: "idle" }; const terminatedAgent = { ...baseAgent, status: "terminated" }; diff --git a/server/src/__tests__/company-model-service-settings-guidance.test.ts b/server/src/__tests__/company-model-service-settings-guidance.test.ts index 20c7ce7fc03..ae66a624877 100644 --- a/server/src/__tests__/company-model-service-settings-guidance.test.ts +++ b/server/src/__tests__/company-model-service-settings-guidance.test.ts @@ -1,5 +1,10 @@ +import { readFileSync } from "node:fs"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; import { expect, it } from "vitest"; -import { searchRunnerApi } from "../services/native-runtime/runner-api-catalog.js"; +import { runnerApiCatalog, searchRunnerApi } from "../services/native-runtime/runner-api-catalog.js"; + +const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../../.."); // An agent asked to turn on a company model service cannot do it and must link the page that holds the control. // The general /company/settings page has none of these sections, so the catalog names the Connections page. @@ -19,3 +24,17 @@ it("names the experiment that hides the Fast response section", () => { const operation = searchRunnerApi({ query: "fast response", limit: 10 }).results.find(row => row.operationId === "GET /api/companies/{companyId}/fast-response"); expect(operation?.description).toContain("//company/settings/instance/experimental"); }); + +// The catalog text is the only place these page paths live on the server, so a renamed UI route +// would leave agents linking a page that no longer exists. Every page the guidance names must be +// a route the board UI mounts under the company prefix. +it("names only pages that the board UI mounts", () => { + const appRoutes = readFileSync(path.join(repoRoot, "ui/src/App.tsx"), "utf8"); + const pages = new Set( + runnerApiCatalog() + .filter(row => /^\/api\/companies\/\{companyId\}\/(fast-response|decision-model|prompt-injection-prevention)/.test(row.path)) + .flatMap(row => [...(row.description ?? "").matchAll(/\/\/([a-z0-9/-]+)/g)].map(match => match[1]!)), + ); + expect([...pages].sort()).toEqual(["company/settings/connections", "company/settings/instance/experimental"]); + for (const page of pages) expect(appRoutes, page).toContain(` { + const responses = buildOpenApiDocument().paths["/api/instance/database-backups"].post.responses; + expect(Object.keys(responses)).toContain("201"); + expect(Object.keys(responses)).not.toContain("200"); +}); diff --git a/server/src/__tests__/generic-mcp-connection.test.ts b/server/src/__tests__/generic-mcp-connection.test.ts index eb9f3100ce1..70b97b1d1d5 100644 --- a/server/src/__tests__/generic-mcp-connection.test.ts +++ b/server/src/__tests__/generic-mcp-connection.test.ts @@ -1079,6 +1079,80 @@ describeEmbeddedPostgres("generic remote MCP connections", () => { expect(await summary()).toEqual([["create_insight", "allowed"], ["list_insights", "allowed"]]); }); + async function setUpGovernedTwoActionConnection(options: { enabled: "read" | "none"; access?: "all_agents" }) { + installMcpOAuthFixture({ auth: "header", requiredHeader: { name: "Authorization", value: "Bearer fixture-key-123" }, tools: [ + { name: "list_insights", description: "List insights", annotations: { readOnlyHint: true, destructiveHint: false } }, + { name: "create_insight", description: "Create an insight", annotations: { readOnlyHint: false, destructiveHint: true } }, + ] }); + const company = await createCompany(db); + const service = toolAccessService(db); + const actor = { actorType: "user" as const, actorId: "board-user", userId: "board-user" } as never; + const connected = await service.connectGalleryApp(company.id, { link: MCP_URL, name: "Fixture bearer", authMode: "bearer", + credentialValues: { "credentials.authorization": "fixture-key-123" }, grantKind: "organization" } as never, actor); + const readOnly = connected.actions.readOnly.map((row: { catalogEntryId: string }) => row.catalogEntryId); + const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Morgan", role: "engineer", status: "active", adapterType: "process", adapterConfig: {}, runtimeConfig: {} }).returning(); + await service.finishGalleryAppConnection(company.id, connected.connectionId, { + enabledCatalogEntryIds: options.enabled === "read" ? readOnly : [], askFirstCatalogEntryIds: [], + access: options.access ?? { agentIds: [] }, + }, actor); + if (!options.access) await service.putConnectionInstalls(connected.connectionId, { installs: [{ targetType: "agent", targetId: agent!.id }] }, actor); + // Every new standard agent receives this unscoped grant. + await db.insert(principalPermissionGrants).values({ companyId: company.id, principalType: "agent", principalId: agent!.id, permissionKey: "tools:use", scope: null }); + const gateway = createToolGatewayService(db, { toolActionSigningSecret: "fixture-secret" }); + const summary = async (agentId = agent!.id) => (await gateway.summarizeConnectionAccessForAgent({ companyId: company.id, connectionId: connected.connectionId, agentId })) + .tools.map(tool => [tool.toolName, tool.decision]).sort(); + const setGrantScope = (scope: Record | null) => db.update(principalPermissionGrants).set({ scope }) + .where(and(eq(principalPermissionGrants.principalId, agent!.id), eq(principalPermissionGrants.permissionKey, "tools:use"))); + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.id, connected.connectionId)); + return { company, agent: agent!, connection: connection!, summary, setGrantScope }; + } + + it("keeps every action off when the connection turns all of them off, even for the default unscoped tools:use grant", async () => { + const { connection, summary, setGrantScope } = await setUpGovernedTwoActionConnection({ enabled: "none" }); + expect(await summary()).toEqual([["create_insight", "off"], ["list_insights", "off"]]); + // A grant that only narrows when it applies (agent, project) still covers every tool, so it is not a choice about this connection. + await setGrantScope({ actorType: "agent" }); + expect(await summary()).toEqual([["create_insight", "off"], ["list_insights", "off"]]); + // Grants that name the connection or the application are explicit choices and still apply. + await setGrantScope({ connectionId: connection.id }); + expect(await summary()).toEqual([["create_insight", "allowed"], ["list_insights", "allowed"]]); + await setGrantScope({ allow: [`application:${connection.applicationId}`] }); + expect(await summary()).toEqual([["create_insight", "allowed"], ["list_insights", "allowed"]]); + }); + + it("keeps the unscoped tools:use grant for agents the connection's profile does not cover", async () => { + const { company, summary } = await setUpGovernedTwoActionConnection({ enabled: "read" }); + const [other] = await db.insert(agents).values({ companyId: company.id, name: "Riley", role: "engineer", status: "active", adapterType: "process", adapterConfig: {}, runtimeConfig: {} }).returning(); + await db.insert(principalPermissionGrants).values({ companyId: company.id, principalType: "agent", principalId: other!.id, permissionKey: "tools:use", scope: null }); + expect(await summary(other!.id)).toEqual([["create_insight", "allowed"], ["list_insights", "allowed"]]); + }); + + it("does not let the unscoped tools:use grant override a profile entry that excludes the application", async () => { + const { company, summary } = await setUpGovernedTwoActionConnection({ enabled: "read" }); + const [other] = await db.insert(agents).values({ companyId: company.id, name: "Riley", role: "engineer", status: "active", adapterType: "process", adapterConfig: {}, runtimeConfig: {} }).returning(); + await db.insert(principalPermissionGrants).values({ companyId: company.id, principalType: "agent", principalId: other!.id, permissionKey: "tools:use", scope: null }); + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.companyId, company.id)); + const [profile] = await db.insert(toolProfiles).values({ companyId: company.id, profileKey: `no-insights-${randomUUID()}`, name: "No insights", defaultAction: "allow" }).returning(); + // An application selector carries no connection id. + await db.insert(toolProfileEntries).values({ companyId: company.id, profileId: profile!.id, selectorType: "application", effect: "exclude", applicationId: connection!.applicationId }); + await db.insert(toolProfileBindings).values({ companyId: company.id, profileId: profile!.id, targetType: "agent", targetId: other!.id }); + expect(await summary(other!.id)).toEqual([["create_insight", "off"], ["list_insights", "off"]]); + }); + + it("applies the connection profile on the direct policy check and keeps a board user's unscoped grant", async () => { + const { company, agent, connection, summary } = await setUpGovernedTwoActionConnection({ enabled: "read", access: "all_agents" }); + expect(await summary()).toEqual([["create_insight", "off"], ["list_insights", "allowed"]]); + const [create] = await db.select().from(toolCatalogEntries).where(and(eq(toolCatalogEntries.connectionId, connection.id), eq(toolCatalogEntries.toolName, "create_insight"))); + const request = { connectionId: connection.id, catalogEntryId: create!.id, toolName: create!.toolName, arguments: {} }; + const policy = toolAccessPolicyService(db); + const agentDecision = await policy.decide({ companyId: company.id, actor: { actorType: "agent", actorId: agent.id, agentId: agent.id }, request }); + expect([agentDecision.decision, agentDecision.reasonCode]).toEqual(["deny", "deny_default"]); + // The connection's profile is bound to the company, which also matches board users; their operator grant still applies. + await db.insert(principalPermissionGrants).values({ companyId: company.id, principalType: "user", principalId: "board-user", permissionKey: "tools:use", scope: null }); + const userDecision = await policy.decide({ companyId: company.id, actor: { actorType: "user", actorId: "board-user" }, request }); + expect([userDecision.decision, userDecision.reasonCode]).toEqual(["allow", "allow_explicit_grant"]); + }); + it("stores a bearer key as a secret and never reads it back", async () => { installMcpOAuthFixture({ auth: "header", diff --git a/server/src/__tests__/skill-update-diff-routes.test.ts b/server/src/__tests__/skill-update-diff-routes.test.ts new file mode 100644 index 00000000000..76110bc1425 --- /dev/null +++ b/server/src/__tests__/skill-update-diff-routes.test.ts @@ -0,0 +1,126 @@ +import express from "express"; +import request from "supertest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; +import { hoistModuleGraph } from "./helpers/hoist-module-graph.js"; + +const INSTALLED = "a".repeat(40); +const LATEST = "b".repeat(40); +const mockSkills = vi.hoisted(() => ({ getById: vi.fn(), getByRouteRef: vi.fn() })); +const mockSources = vi.hoisted(() => ({ sourceForSkill: vi.fn() })); +const mockRead = vi.hoisted(() => vi.fn()); +const mockDiff = vi.hoisted(() => vi.fn()); + +function registerModuleMocks() { + vi.doMock("../services/skill-sources.js", () => ({ skillSourceService: () => mockSources })); + vi.doMock("../services/skill-source-github-access.js", () => ({ skillSourceGitHubReader: () => mockRead })); + vi.doMock("../services/skill-update-diff.js", () => ({ diffGitHubSkillUpdate: mockDiff })); + vi.doMock("../services/company-skills.js", async () => ({ + ...await vi.importActual("../services/company-skills.js"), + companySkillService: () => mockSkills, + })); + vi.doMock("../services/index.js", async () => ({ + ...await vi.importActual("../services/index.js"), + companySkillService: () => mockSkills, + })); +} + +const board = { type: "board", userId: "user-1", companyIds: ["company-1"], source: "session", isInstanceAdmin: false }; +const trackedSource = { + enabled: true, connectionId: null, fullName: "example/skills", trackingRef: "main", repositoryUrl: "https://github.com/example/skills", + packageMode: "skills", entries: [{ skillId: "skill-1", selection: "selected", path: "design/SKILL.md" }], +}; + +describe("GET /companies/:companyId/skills/:skillId/update-diff", () => { + const routeModules = hoistModuleGraph(registerModuleMocks, async () => { + const [{ companySkillRoutes }, { errorHandler }] = await Promise.all([ + vi.importActual("../routes/company-skills.js"), + vi.importActual("../middleware/index.js"), + ]); + return { companySkillRoutes, errorHandler }; + }); + + function app(actor: Record = board) { + const { companySkillRoutes, errorHandler } = routeModules.value; + const server = express(); + server.use((req, _res, next) => { (req as any).actor = actor; next(); }); + server.use("/api", companySkillRoutes({} as never)); + server.use(errorHandler); + return server; + } + + beforeEach(() => { + vi.clearAllMocks(); + // The real lookup applies the caller's skill visibility; a skill the caller cannot read is not found. + mockSkills.getByRouteRef.mockImplementation(async (_companyId: string, ref: string) => ref === "skill-1" ? { id: "skill-1", sourceRef: INSTALLED } : null); + mockSkills.getById.mockImplementation(async (_companyId: string, id: string) => id === "skill-1" ? { id: "skill-1", sourceRef: INSTALLED } : null); + mockSources.sourceForSkill.mockResolvedValue(trackedSource); + mockRead.mockResolvedValue({ sha: LATEST }); + mockDiff.mockResolvedValue({ supported: true, truncated: false, files: [{ path: "SKILL.md", change: "modified", binary: false, executableChanged: false, diff: "@@ -1,1 +1,1 @@\n-a\n+b", diffUnavailable: null }] }); + }); + + it("returns the per-file diff between the installed and latest revisions", async () => { + const response = await request(app()).get("/api/companies/company-1/skills/skill-1/update-diff"); + expect(response.status).toBe(200); + expect(response.body).toMatchObject({ supported: true, hasUpdate: true, currentRef: INSTALLED, latestRef: LATEST, truncated: false }); + expect(response.body.files).toHaveLength(1); + expect(mockDiff).toHaveBeenCalledWith(expect.objectContaining({ installedCommitSha: INSTALLED, latestCommitSha: LATEST, skillPath: "design/SKILL.md" }), mockRead, expect.any(AbortSignal)); + }); + + it("answers a skill with no pending update without scanning either revision", async () => { + mockRead.mockResolvedValue({ sha: INSTALLED.toUpperCase() }); + const response = await request(app()).get("/api/companies/company-1/skills/skill-1/update-diff"); + expect(response.status).toBe(200); + expect(response.body).toEqual({ supported: true, reason: null, currentRef: INSTALLED, latestRef: INSTALLED.toUpperCase(), hasUpdate: false, files: [], truncated: false }); + expect(mockDiff).not.toHaveBeenCalled(); + }); + + it("rejects a caller from another company before reading anything", async () => { + const response = await request(app({ ...board, companyIds: ["company-2"] })).get("/api/companies/company-1/skills/skill-1/update-diff"); + expect(response.status).toBe(403); + expect(mockSkills.getByRouteRef).not.toHaveBeenCalled(); + expect(mockRead).not.toHaveBeenCalled(); + const agent = await request(app({ type: "agent", agentId: "agent-1", companyId: "company-2", runId: "run-1" })).get("/api/companies/company-1/skills/skill-1/update-diff"); + expect(agent.status).toBe(403); + }); + + it("returns 404 for a skill the caller cannot read", async () => { + const response = await request(app()).get("/api/companies/company-1/skills/private-skill/update-diff"); + expect(response.status).toBe(404); + expect(mockSources.sourceForSkill).not.toHaveBeenCalled(); + expect(mockRead).not.toHaveBeenCalled(); + }); + + it("explains why a skill without a usable tracked source has no diff", async () => { + mockSources.sourceForSkill.mockResolvedValueOnce(null); + expect((await request(app()).get("/api/companies/company-1/skills/skill-1/update-diff")).body) + .toMatchObject({ supported: false, reason: "Only skills installed from a tracked GitHub source have an update diff.", files: [] }); + mockSources.sourceForSkill.mockResolvedValueOnce({ ...trackedSource, enabled: false }); + expect((await request(app()).get("/api/companies/company-1/skills/skill-1/update-diff")).body) + .toMatchObject({ supported: false, reason: "Source is disconnected. Reconnect it in Skills → Sources." }); + mockSkills.getById.mockResolvedValueOnce({ id: "skill-1", sourceRef: "main" }); + expect((await request(app()).get("/api/companies/company-1/skills/skill-1/update-diff")).body) + .toMatchObject({ supported: false, reason: "The installed source revision is unknown.", currentRef: "main" }); + mockDiff.mockResolvedValueOnce({ supported: false, reason: "The repository is larger than the supported package size, so its revisions cannot be compared file by file." }); + expect((await request(app()).get("/api/companies/company-1/skills/skill-1/update-diff")).body) + .toMatchObject({ supported: false, hasUpdate: true, latestRef: LATEST, files: [] }); + expect(mockDiff).toHaveBeenCalledTimes(1); + }); + + it("aborts the scans when the caller disconnects", async () => { + let seen: AbortSignal | undefined; + mockDiff.mockImplementationOnce((_input: unknown, _read: unknown, signal: AbortSignal) => new Promise((_resolve, reject) => { + seen = signal; + signal.addEventListener("abort", () => reject(signal.reason), { once: true }); + })); + const server = app().listen(0); + try { + const { port } = server.address() as { port: number }; + const controller = new AbortController(); + const pending = fetch(`http://127.0.0.1:${port}/api/companies/company-1/skills/skill-1/update-diff`, { signal: controller.signal }).catch(() => null); + await vi.waitFor(() => expect(seen).toBeDefined()); + controller.abort(); + await pending; + await vi.waitFor(() => expect(seen!.aborted).toBe(true)); + } finally { server.close(); } + }); +}); diff --git a/server/src/__tests__/skill-update-diff.test.ts b/server/src/__tests__/skill-update-diff.test.ts new file mode 100644 index 00000000000..21d26edbd54 --- /dev/null +++ b/server/src/__tests__/skill-update-diff.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, it, vi } from "vitest"; + +type ScanFile = { path: string; content: string; encoding?: "utf8" | "base64"; executable?: boolean }; +type Scan = { commitSha: string; packageMode?: "skills" | "repository"; repositorySubset?: boolean; repositoryFiles?: ScanFile[]; + repositoryPackage?: { warnings: string[] }; warnings?: string[]; skills: Array<{ path: string; files: ScanFile[]; warnings?: string[] }> }; +const scans = vi.hoisted(() => new Map()); +const scanSignals = vi.hoisted(() => [] as Array); +vi.mock("../services/github-skill-source.js", () => ({ + scanGitHubSkillsWithPackageFallback: vi.fn(async (input: { commitSha: string }, _read: unknown, options: { signal?: AbortSignal } = {}) => { + scanSignals.push(options.signal); + return { warnings: [], ...scans.get(input.commitSha) }; + }), +})); +import { diffGitHubSkillUpdate, diffSkillPackageFiles, MAX_UPDATE_DIFF_CHARS, MAX_UPDATE_DIFF_FILES, unifiedLineDiff } from "../services/skill-update-diff.js"; + +const installed = ["---", "version: \"0.4\"", "---", "", "Run checks so every build or release runs them.", "", "## Notes", "Keep it short."].join("\n"); +const latest = ["---", "version: \"0.5\"", "---", "", "Run checks so every new build runs them.", "", "## Design", "Breathable layouts.", "", "## Notes", "Keep it short."].join("\n"); + +describe("skill update diff", () => { + it("marks installed-only lines with - and latest-only lines with +", () => { + const diff = unifiedLineDiff(installed, latest)!; + const lines = diff.split("\n"); + expect(lines).toContain("-version: \"0.4\""); + expect(lines).toContain("+version: \"0.5\""); + expect(lines).toContain("-Run checks so every build or release runs them."); + expect(lines).toContain("+Run checks so every new build runs them."); + expect(lines).toContain("+## Design"); + expect(lines).toContain("+Breathable layouts."); + expect(lines).not.toContain("-## Notes"); + expect(lines.filter(line => line.startsWith("@@"))).toEqual(["@@ -1,8 +1,11 @@"]); + expect(unifiedLineDiff(installed, installed)).toBe(""); + }); + + it("keeps distant changes in separate hunks with bounded context", () => { + const before = Array.from({ length: 30 }, (_, i) => `line ${i + 1}`); + const after = [...before]; after[1] = "changed 2"; after[27] = "changed 28"; + const diff = unifiedLineDiff(before.join("\n"), after.join("\n"))!; + expect(diff.split("\n").filter(line => line.startsWith("@@"))).toEqual(["@@ -1,5 +1,5 @@", "@@ -25,6 +25,6 @@"]); + expect(diff).not.toContain(" line 15"); + }); + + it("lists added, removed, modified and binary files and skips unchanged ones", () => { + const { files, truncated } = diffSkillPackageFiles( + [{ path: "SKILL.md", content: installed }, { path: "old.md", content: "gone\n" }, { path: "same.md", content: "same" }, { path: "logo.png", content: "AAA=", encoding: "base64" }], + [{ path: "SKILL.md", content: latest }, { path: "new.md", content: "fresh\n" }, { path: "same.md", content: "same" }, { path: "logo.png", content: "AAE=", encoding: "base64" }], + ); + expect(truncated).toBe(false); + expect(files.map(file => [file.path, file.change, file.binary, file.diffUnavailable])).toEqual([ + ["SKILL.md", "modified", false, null], ["logo.png", "modified", true, "binary"], ["new.md", "added", false, null], ["old.md", "removed", false, null], + ]); + // An absent side has no lines, so an added or removed file shows only its own lines. + expect(files.find(file => file.path === "new.md")!.diff).toBe("@@ -0,0 +1,1 @@\n+fresh"); + expect(files.find(file => file.path === "old.md")!.diff).toBe("@@ -1,1 +0,0 @@\n-gone"); + expect(files.find(file => file.path === "logo.png")!.diff).toBeNull(); + }); + + it("does not count a final newline as a line and shows when only the final newline changes", () => { + expect(unifiedLineDiff("a\nb\n", "a\nc\n")).toBe("@@ -1,2 +1,2 @@\n a\n-b\n+c"); + expect(unifiedLineDiff("a\n", "a")).toBe("@@ -1,1 +1,1 @@\n-a\n+a\n\\ No newline at end of file"); + }); + + it("reports a change to only the executable bit", () => { + const { files } = diffSkillPackageFiles([{ path: "run.sh", content: "echo hi\n" }], [{ path: "run.sh", content: "echo hi\n", executable: true }]); + expect(files).toEqual([{ path: "run.sh", change: "modified", binary: false, executableChanged: true, diff: "", diffUnavailable: null }]); + }); + + it("does not call a file the scan left out of one revision added or removed", () => { + const { files } = diffSkillPackageFiles([{ path: "SKILL.md", content: "a" }], [{ path: "SKILL.md", content: "a" }, { path: "assets/logo.svg", content: "" }], + { installed: new Set(["assets/logo.svg"]), latest: new Set() }); + expect(files).toEqual([{ path: "assets/logo.svg", change: "not_compared", binary: false, executableChanged: false, diff: null, diffUnavailable: "omitted" }]); + }); + + it("bounds the diff text and the file count of one response", () => { + const big = Array.from({ length: 400 }, (_, i) => ({ path: `f${String(i).padStart(3, "0")}.md`, content: `${"x".repeat(5_000)}\n` })); + const { files, truncated } = diffSkillPackageFiles(big.map(file => ({ ...file, content: "" })), big); + expect(truncated).toBe(true); + expect(files.reduce((sum, file) => sum + (file.diff?.length ?? 0), 0)).toBeLessThanOrEqual(MAX_UPDATE_DIFF_CHARS); + expect(files.at(-1)!.diffUnavailable).toBe("budget"); + const many = Array.from({ length: MAX_UPDATE_DIFF_FILES + 5 }, (_, i) => ({ path: `m${i}.md`, content: "x" })); + const listed = diffSkillPackageFiles([], many); + expect([listed.files.length, listed.truncated]).toEqual([MAX_UPDATE_DIFF_FILES, true]); + }); + + it("compares the installed commit snapshot with the latest one for the selected skill", async () => { + scans.set("a".repeat(40), { commitSha: "a".repeat(40), skills: [{ path: "accessibility/SKILL.md", files: [{ path: "SKILL.md", content: installed }] }] }); + scans.set("b".repeat(40), { commitSha: "b".repeat(40), skills: [{ path: "accessibility/SKILL.md", files: [{ path: "SKILL.md", content: latest }] }] }); + const controller = new AbortController(); + scanSignals.length = 0; + const result = await diffGitHubSkillUpdate({ repositoryUrl: "https://github.com/example/skills", skillPath: "accessibility/SKILL.md", installedCommitSha: "a".repeat(40), latestCommitSha: "b".repeat(40) }, {} as never, controller.signal); + if (!result.supported) throw new Error(result.reason); + expect(result.files).toHaveLength(1); + expect(result.files[0]!.diff).toContain("-Run checks so every build or release runs them."); + expect(scanSignals).toEqual([controller.signal, controller.signal]); + scans.set("c".repeat(40), { commitSha: "d".repeat(40), skills: [] }); + await expect(diffGitHubSkillUpdate({ repositoryUrl: "https://github.com/example/skills", skillPath: "accessibility/SKILL.md", installedCommitSha: "a".repeat(40), latestCommitSha: "c".repeat(40) }, {} as never)).rejects.toThrow("did not match"); + }); + + it("marks files the scan left out of a skill folder as not compared", async () => { + scans.set("e".repeat(40), { commitSha: "e".repeat(40), skills: [{ path: "design/SKILL.md", files: [{ path: "SKILL.md", content: "a" }], + warnings: ["design/assets/hero.png — Exceeds the 1 MB file limit; not included."] }] }); + scans.set("f".repeat(40), { commitSha: "f".repeat(40), skills: [{ path: "design/SKILL.md", files: [{ path: "SKILL.md", content: "a" }, { path: "assets/hero.png", content: "AAA=", encoding: "base64" }] }] }); + const result = await diffGitHubSkillUpdate({ repositoryUrl: "https://github.com/example/skills", skillPath: "design/SKILL.md", installedCommitSha: "e".repeat(40), latestCommitSha: "f".repeat(40) }, {} as never); + expect(result).toEqual({ supported: true, truncated: false, files: [{ path: "assets/hero.png", change: "not_compared", binary: false, executableChanged: false, diff: null, diffUnavailable: "omitted" }] }); + }); + + it("refuses to compare a repository package that a scan could only include in part", async () => { + const full = { commitSha: "1".repeat(40), packageMode: "repository" as const, repositoryFiles: [{ path: "README.md", content: "a" }], skills: [] }; + const input = { repositoryUrl: "https://github.com/example/skills", packageMode: "repository" as const, skillPath: "x/SKILL.md", installedCommitSha: "1".repeat(40), latestCommitSha: "2".repeat(40) }; + scans.set("1".repeat(40), full); + for (const partial of [ + { commitSha: "2".repeat(40), packageMode: "repository" as const, repositorySubset: true, repositoryFiles: [], skills: [] }, + { commitSha: "2".repeat(40), packageMode: "skills" as const, skills: [] }, + ]) { + scans.set("2".repeat(40), partial); + const result = await diffGitHubSkillUpdate(input, {} as never); + expect(result.supported).toBe(false); + } + scans.set("2".repeat(40), { ...full, commitSha: "2".repeat(40), repositoryFiles: [{ path: "README.md", content: "b" }] }); + const result = await diffGitHubSkillUpdate(input, {} as never); + expect(result.supported && result.files.map(file => file.path)).toEqual(["README.md"]); + }); +}); diff --git a/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts b/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts index 2f1999565b7..da514753201 100644 --- a/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/driver-scope.test.ts @@ -8,10 +8,13 @@ import { createLifecycleDriver } from "../../../services/agent-lifecycle-driver. describe("lifecycle host stop scope", () => { it("pauses only the requested agent without loading its company peers", async () => { - const db = { select: vi.fn(() => { throw new Error("No peer lookup expected"); }) } as unknown as Db; + // The pause stamps its cause on the agent's own runs (one update), never a peer lookup. + const update = vi.fn(() => ({ set: () => ({ where: async () => [] }) })); + const db = { select: vi.fn(() => { throw new Error("No peer lookup expected"); }), update } as unknown as Db; const driver = createLifecycleDriver(db, {} as never); - expect(await driver.runHost({ id: "target", companyId: "company", lifecycleState: "pausing", - lifecycleVersion: 2, lifecycleRequiredPluginIds: [], lifecycleError: null, lifecycleOperation: null })).toBe("complete"); + expect(await driver.runHost({ id: "target", companyId: "company", lifecycleState: "pausing", lifecycleHolds: ["manual"], + lifecycleVersion: 2, lifecycleRequiredPluginIds: [], lifecycleError: null, lifecycleOperation: null } as never)).toBe("complete"); + expect(update).toHaveBeenCalledOnce(); // A pause keeps the messages it deferred (work-intent outbox). expect(stop).toHaveBeenCalledWith(["target"], "Agent lifecycle stop requested", { preservePauseDeferred: true }); expect(db.select).not.toHaveBeenCalled(); diff --git a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts index 7048ccb0e53..af9c1709c00 100644 --- a/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts +++ b/server/src/modules/agent-lifecycle/__tests__/lifecycle.test.ts @@ -414,6 +414,19 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu expect(await db.select().from(principalPermissionGrants).where(eq(principalPermissionGrants.companyId, companyId))).toHaveLength(0); }); + // H1 bug B: a pause's stop is told apart from termination by the cause the + // pause records on each run it stops. + it("records what a pause stopped each active run for", async () => { + const agent = await hire(); + await db.update(companies).set({ status: "paused", pauseReason: "budget" }).where(eq(companies.id, companyId)); + await db.update(agents).set({ lifecycleState: "pausing", lifecycleHolds: ["company_paused"] }).where(eq(agents.id, agent.id)); + const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId: agent.id, status: "queued" }).returning(); + await createLifecycleDriver(db, {} as never).runHost(await current(agent.id) as never); + const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + expect(stopped.status).toBe("cancelled"); + expect(stopped.resultJson).toMatchObject({ pauseStop: { holds: ["company_paused"], companyPauseReason: "budget" } }); + }); + it("runs the saved harness test and keeps a failed configuration out of ready", async () => { const driver = createLifecycleDriver(db, {} as never); const work = worker(driver.runHost); diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 831d8228d3d..761624744fa 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -3,7 +3,7 @@ import { DELIVERY_QUEUES } from "../../../services/delivery-work-notifications.j import { instanceSettingsService } from "../../../services/instance-settings.js"; import { randomBytes, randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; -import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; import type { Db } from "@paperclipai/db"; import { activityLog, @@ -754,6 +754,94 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0].status).toBe("deferred_issue_execution"); }); + // H1 bug B: a budget pause stopped the agent's run, and the release blocked + // the task behind a reconciliation incident with no control. A pause is an + // intentional stop; resuming continues the task. + it.each(["Agent lifecycle stop requested", "Cancelled due to budget pause", "Cancelled due to agent pause"])( + "keeps a native task's status and opens no incident when a pause stopped its run (%s)", async (reason) => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason, source: "control_plane", expected: true, initiator: { type: "system" }, + recordedAt: new Date().toISOString() }, + ...(reason === "Agent lifecycle stop requested" ? { pauseStop: { holds: ["budget"], companyPauseReason: null } } : {}) } }).where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + }); + + // Review: the full release (createReleaseIssueExecution), not only the + // native incident step, must leave a paused agent's task unblocked. + it.each(["native", "legacy"] as const)("keeps a %s task unblocked through the full release when its agent's pause stopped the run", async (runtimeMode) => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + await db.update(agents).set({ status: "paused", lifecycleState: "pausing", lifecycleHolds: ["manual"] }).where(eq(agents.id, agentId)); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode, ...(runtimeMode === "native" ? { nativeIssueId: issueId } : {}), + errorCode: "cancelled", resultJson: { cancellation: { reason: "Agent lifecycle stop requested", source: "control_plane", + expected: true, initiator: { type: "system" } }, pauseStop: { holds: ["manual"], companyPauseReason: null } } }) + .where(eq(heartbeatRuns.id, runId)); + await db.update(issues).set({ executionRunId: runId, checkoutRunId: runId }).where(eq(issues.id, issueId)); + const escalateStrandedAssignedIssue = vi.fn(async () => undefined); + const release = createReleaseIssueExecution({ issueLock: createPostgresWakeQueueAdapter(db, stubDeps), + recovery: { escalateStrandedAssignedIssue, escalateStrandedRecoveryIssueInPlace: async () => undefined }, + }); + await release({ companyId, runId, now: new Date() }); + expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(0); + expect(escalateStrandedAssignedIssue).not.toHaveBeenCalled(); + }); + + it("still opens a reconciliation incident when termination replaced the pause that stamped the run", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + await db.update(agents).set({ lifecycleState: "terminating" }).where(eq(agents.id, agentId)); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason: "Agent lifecycle stop requested", source: "control_plane", expected: true, + initiator: { type: "system" } }, pauseStop: { holds: ["budget"], companyPauseReason: null } } }) + .where(eq(heartbeatRuns.id, runId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0]) + .toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + }); + + it("still opens a reconciliation incident when agent termination stopped the run", async () => { + // Termination and cleanup use the lifecycle stop reason too, with no pauseStop. + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason: "Agent lifecycle stop requested", source: "control_plane", + expected: true, initiator: { type: "system" } } } }).where(eq(heartbeatRuns.id, runId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0]) + .toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + }); + + it("still opens a reconciliation incident for a native run cancelled by another system stop", async () => { + const companyId = await seedCompany(); + const agentId = await seedAgent({ companyId }); + const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" }); + const runId = await seedRun({ companyId, agentId, status: "cancelled", contextSnapshot: { issueId } }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: issueId, errorCode: "cancelled", + resultJson: { cancellation: { reason: "Cancelled because the company was archived", source: "control_plane", + expected: true, initiator: { type: "system" } } } }).where(eq(heartbeatRuns.id, runId)); + const adapter = createPostgresWakeQueueAdapter(db, stubDeps); + await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => undefined as never); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0]) + .toMatchObject({ cause: "native_continuation_requires_reconciliation" }); + }); + it.each(["in_progress", "blocked"])("preserves recovery ownership and queued messages when a native task fails from %s", async (status) => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); diff --git a/server/src/modules/wake-queue/adapters/postgres.ts b/server/src/modules/wake-queue/adapters/postgres.ts index c3bbf5219a8..d59c9340a0a 100644 --- a/server/src/modules/wake-queue/adapters/postgres.ts +++ b/server/src/modules/wake-queue/adapters/postgres.ts @@ -17,6 +17,7 @@ import { getExecutionBlocker, hasPendingRunCleanup } from "../../../services/exe import { nativeManualAdmissionHoldsExecution } from "../../../services/native-runtime/native-runner-ownership.js"; import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"; import { readReleaseRun, type ReleaseRun } from "./release-run-read.js"; +import { isGovernancePauseStop } from "../../../services/governance-pause-stop.js"; import type { Db } from "@paperclipai/db"; import { deliveredQueuedCommentIds } from "../../../services/queued-comment-delivery.js"; import { extractIssueReferenceIdentifiers, isUuidLike } from "@paperclipai/shared"; @@ -124,6 +125,9 @@ export function isOperatorStoppedRunRow(row: Pick { + const companyId = req.params.companyId as string; + const skillId = req.params.skillId as string; + // The skillId parameter handler has already checked company access and that this actor can read the skill. + assertCompanyAccess(req, companyId); + const unsupported = (reason: string, currentRef: string | null = null, latestRef: string | null = null): SkillSourceUpdateDiff => + ({ supported: false, reason, currentRef, latestRef, hasUpdate: false, files: [], truncated: false }); + const skill = await svc.getById(companyId, skillId); + if (!skill) { res.status(404).json({ error: "Skill not found" }); return; } + const managed = await sourceSvc.sourceForSkill(companyId, skillId); + const entry = managed?.entries.find(row => row.skillId === skillId); + if (!managed || !entry) { res.json(unsupported("Only skills installed from a tracked GitHub source have an update diff.", skill.sourceRef ?? null)); return; } + if (!managed.enabled || entry.selection !== "selected") { res.json(unsupported("Source is disconnected. Reconnect it in Skills → Sources.", skill.sourceRef ?? null)); return; } + const currentRef = skill.sourceRef ?? null; + if (!currentRef || !/^[a-f0-9]{40}$/i.test(currentRef)) { res.json(unsupported("The installed source revision is unknown.", currentRef)); return; } + // Stop the GitHub reads and both scans when the caller goes away; they hold shared scan and download slots. + const controller = new AbortController(); + const stop = () => { if (!res.writableEnded) controller.abort(); }; + res.once("close", stop); + try { + const read = skillSourceGitHubReader(db, companyId, req.actor, managed.connectionId); + const latest = await read(`/repos/${managed.fullName}/commits/${encodeURIComponent(managed.trackingRef)}`, controller.signal) as { sha: string }; + if (latest.sha.toLowerCase() === currentRef.toLowerCase()) { + res.json({ supported: true, reason: null, currentRef, latestRef: latest.sha, hasUpdate: false, files: [], truncated: false } satisfies SkillSourceUpdateDiff); + return; + } + const diff = await diffGitHubSkillUpdate({ + repositoryUrl: managed.repositoryUrl, packageMode: managed.packageMode, skillPath: entry.path, installedCommitSha: currentRef, latestCommitSha: latest.sha, + }, read, controller.signal); + const result: SkillSourceUpdateDiff = diff.supported + ? { supported: true, reason: null, currentRef, latestRef: latest.sha, hasUpdate: true, files: diff.files, truncated: diff.truncated } + : { ...unsupported(diff.reason, currentRef, latest.sha), hasUpdate: true }; + res.json(result); + } finally { res.off("close", stop); } + }); + router.get("/companies/:companyId/skills/:skillId/files", async (req, res) => { const companyId = req.params.companyId as string; const skillId = req.params.skillId as string; diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 068495bba5e..318fd711e08 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -8647,6 +8647,16 @@ registry.registerPath({ responses: { 200: r.ok(), 401: r.unauthorized }, }); +registry.registerPath({ + method: "get", + path: "/api/companies/{companyId}/skills/{skillId}/update-diff", + tags: ["skills"], + summary: "Show what a pending skill source update changes (diff installed vs latest revision)", + description: "For a skill installed from a tracked GitHub source, compares the installed source revision (currentRef) with the latest tracked revision (latestRef) and returns each changed package file with a unified line diff: lines starting with \"-\" exist only in the installed revision and lines starting with \"+\" only in the latest one. hasUpdate is false and files is empty when the installed revision is already the latest. A file with change \"not_compared\" was left out of a revision's scan, and diffUnavailable says why a file has no diff. truncated is true when the response reached its size budget. Read-only; it does not apply the update. Use this to review an update before installing it.", + request: { params: z.object({ companyId: z.string(), skillId: z.string() }) }, + responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden, 404: r.notFound }, +}); + registry.registerPath({ method: "get", path: "/api/companies/{companyId}/skills/{skillId}/files", @@ -9887,7 +9897,7 @@ registry.registerPath({ tags: ["instance"], summary: "Trigger a database backup", description: "Instance admins only: agent credentials receive 403, and cloud-managed instances run backups for you. Automatic gzip-compressed backups run about hourly by default; instance admins set how long they are kept in the Backup retention section of the company-prefixed /company/settings page. The web UI has no control that creates or restores a backup. For a backup now, an instance admin calls this endpoint, or an operator with shell access to the server runs `paperclipai db:backup` (add --json for metadata). The result names a gzip-compressed SQL file (backupFile) on the server host and its size. Paperclip has no restore endpoint or command: to check a backup, an operator restores the file with psql into a separate, empty PostgreSQL database, never over the live one, and then checks the restored data.", - responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden }, + responses: { 201: r.ok(), 401: r.unauthorized, 403: r.forbidden }, }); // ─── LLM text endpoints ─────────────────────────────────────────────────────── diff --git a/server/src/services/agent-lifecycle-driver.ts b/server/src/services/agent-lifecycle-driver.ts index e02c9e9472a..f3147acd956 100644 --- a/server/src/services/agent-lifecycle-driver.ts +++ b/server/src/services/agent-lifecycle-driver.ts @@ -1,6 +1,7 @@ -import { and, eq, inArray } from "drizzle-orm"; +import { and, eq, inArray, sql } from "drizzle-orm"; +import { CANCELLABLE_HEARTBEAT_RUN_STATUSES } from "./heartbeat/recovery.js"; import { promoteDeferredWorkIntents } from "./work-intent-outbox.js"; -import { agents, environments, plugins, pluginCompanySettings, type Db } from "@paperclipai/db"; +import { agents, companies, environments, heartbeatRuns, plugins, pluginCompanySettings, type Db } from "@paperclipai/db"; import { instanceSettingsService } from "./instance-settings.js"; import { pluginHasAgentEnvironments, pluginServesAgentEnvironment } from "./plugin-environment-driver.js"; import { heartbeatService } from "./heartbeat.js"; @@ -62,6 +63,18 @@ export function createLifecycleDriver(db: Db, manager: PluginWorkerManager): Lif await db.select().from(agents).where(eq(agents.companyId, agent.companyId)))]; const reason = agent.lifecycleHolds?.includes("company_archived") ? "Cancelled because the company was archived" : "Agent lifecycle stop requested"; + // A pause records why it stopped each run (pauseStop). It marks the + // stop as a pause, not termination or cleanup, which share this reason. + if (agent.lifecycleState === "pausing") { + const pauseStop = sql`jsonb_build_object('pauseStop', jsonb_build_object( + 'holds', ${JSON.stringify(agent.lifecycleHolds ?? [])}::jsonb, + 'companyPauseReason', (select ${companies.pauseReason} from ${companies} where ${companies.id} = ${agent.companyId})))`; + await db.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${pauseStop}`, + }).where(and(eq(heartbeatRuns.companyId, agent.companyId), eq(heartbeatRuns.agentId, agent.id), + inArray(heartbeatRuns.status, [...CANCELLABLE_HEARTBEAT_RUN_STATUSES]), + sql`not (coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ? 'pauseStop')`)); + } if (!await heartbeat.stopInvocationsForAgents(ids, reason, { preservePauseDeferred: agent.lifecycleState === "pausing" })) return "pending"; } diff --git a/server/src/services/governance-pause-stop.ts b/server/src/services/governance-pause-stop.ts new file mode 100644 index 00000000000..04a1e5a5616 --- /dev/null +++ b/server/src/services/governance-pause-stop.ts @@ -0,0 +1,29 @@ +/** + * A governance pause stop: the control plane cancelled the run because its + * agent was paused (the direct pause route, or the agent lifecycle stop that a + * budget, company or manual pause issues) or because a budget hard stop fenced it. The stop is + * intentional, not a failure: the task keeps its status, and resuming + * continues the work (budget-resume-continuation.ts). It never becomes a + * reconciliation incident that has no control. + */ +export const AGENT_LIFECYCLE_STOP_REASON = "Agent lifecycle stop requested"; +export const BUDGET_PAUSE_STOP_REASON = "Cancelled due to budget pause"; +/** The direct pause route stops the agent's runs with this reason before the lifecycle worker runs. */ +export const AGENT_PAUSE_STOP_REASON = "Cancelled due to agent pause"; + +const GOVERNANCE_PAUSE_STOP_REASONS: ReadonlySet = new Set([ + AGENT_LIFECYCLE_STOP_REASON, BUDGET_PAUSE_STOP_REASON, AGENT_PAUSE_STOP_REASON]); + +const record = (value: unknown): Record => + value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +export function isGovernancePauseStop(run: { status: string; resultJson?: Record | null }): boolean { + if (run.status !== "cancelled") return false; + const cancellation = record(run.resultJson?.cancellation); + if (record(cancellation.initiator).type !== "system" || cancellation.expected !== true + || typeof cancellation.reason !== "string" || !GOVERNANCE_PAUSE_STOP_REASONS.has(cancellation.reason)) return false; + // Termination and cleanup issue the same lifecycle stop reason. Only the + // pausing path records pauseStop, so require it for that reason. + return cancellation.reason !== AGENT_LIFECYCLE_STOP_REASON + || Object.keys(record(run.resultJson?.pauseStop)).length > 0; +} diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index 178faddc960..e1887754d71 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -90,7 +90,9 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered * later committed its result and released the controller lease, but no * suspended Runner receipt or failover backup ever reached the controller. * Only the database controller checkpoint and recovery pointer survive. */ - async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" = "settled", + async function seedRecoveredPriorRun(control: "settled" | "pending-command" | "unanswered-tool" + | "failed-turn-start" | "rejected-turn-start" | "indeterminate-turn-start" | "unknown-outcome-turn-start" + | "unsuperseded-unknown-outcome-turn-start" | "failed-suspend" = "settled", options: { controller?: { bootId: string; pid: number; processStartedAt: Date }; provider?: "codex" | "acpx" | "opencode" } = {}) { const issueId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(); await db.insert(issues).values({ id: issueId, companyId, title: "Continue after recovery", assigneeAgentId: agentId, status: "in_progress" }); @@ -127,10 +129,25 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered try { // Shape the file-backed journal first; the database checkpoint captures // it when persistence moves to the controller's external writer. - const commands = ["run.prepare", "session.open", "turn.start", "runner.suspend"].map(type => controller.queueCommand(type)); + const unknownOutcome = { status: "failed", result: { commandType: "turn.start", status: "failed", + result: { code: "execution_indeterminate", message: "runner recovered after journaling this command; it will not execute twice" } } }; + const answers = { + "failed-turn-start": { status: "failed", result: { status: "failed" } }, + "rejected-turn-start": { status: "rejected", result: { status: "rejected" } }, + "indeterminate-turn-start": { status: "indeterminate", result: { status: "indeterminate" } }, + "unknown-outcome-turn-start": unknownOutcome, + } as const; + const answered = answers as Record; + const commands = ["run.prepare", "session.open", ...(control in answered ? ["turn.start"] : []), "turn.start", "runner.suspend"] + .map(type => controller.queueCommand(type)); for (const command of control === "pending-command" ? commands.slice(0, -1) : commands) { Object.assign(command, { status: "completed", result: { status: "completed" } }); } + if (control in answered) Object.assign(commands[2]!, answers[control as keyof typeof answers]); + if (control === "unsuperseded-unknown-outcome-turn-start") { + Object.assign(commands.find(command => command.type === "turn.start")!, unknownOutcome); + } + if (control === "failed-suspend") Object.assign(commands.at(-1)!, { status: "failed", result: { status: "failed" } }); const event = (sourceSeq: number, sourceEventId: string, eventType: string) => ({ sourceSeq, sourceEventId, eventType, priority: 0 as const, envelope: { ...identity }, deliveryCount: 1, logicalEffectCount: 1 }); controller.store.state.committedEvents.push( @@ -185,8 +202,10 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered const remoteExecute = vi.fn(async (_request: unknown): Promise<{ exitCode: number; timedOut: boolean; stdout: string; stderr: string }> => { throw new Error("admission must not reach the sandbox"); }); - const start = (outcome: "created" | "replacement" | "resumed" = "created") => createRunnerdBackend({ + const start = (outcome: "created" | "replacement" | "resumed" = "created", + restartRecovery?: Parameters[0]["restartRecovery"]) => createRunnerdBackend({ db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); }, + ...(restartRecovery ? { restartRecovery } : {}), runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId, leaseId, remoteCwd: "/workspace", reusableLeaseConfigured: true, runner: { execute: remoteExecute }, @@ -628,6 +647,29 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(await retry.start()).toBe("native_controller_recovery_unsupported:metadata_format"); }, 60_000); + it.each([ + { control: "failed-turn-start", outcome: ADMITTED }, + { control: "rejected-turn-start", outcome: ADMITTED }, + { control: "indeterminate-turn-start", outcome: ADMITTED }, + { control: "unknown-outcome-turn-start", outcome: ADMITTED }, + { control: "unsuperseded-unknown-outcome-turn-start", outcome: stranded("journal_commands_unsettled(turn.start.failed)") }, + { control: "failed-suspend", outcome: stranded("journal_commands_unsettled(runner.suspend.failed)") }, + ] as const)("retires a finished owner whose journal holds an answered $control command", async ({ control, outcome }) => { + const prior = await seedRecoveredPriorRun(control); + const followUp = await admitFollowUp(prior); + expect(await followUp.start()).toBe(outcome); + }, 60_000); + + it("starts a follow-up whose own bootstrap never reached its root from the previous owner's settled root", async () => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + const root = await restoreHeldRoot(prior, followUp); + expect(await followUp.start("created", { kind: "bootstrap_incomplete", runId: followUp.runId, leaseOwner: "successor", + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null } as never)).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + expect(await archivedOwners(root)).toEqual([prior.runId]); + }, 60_000); + it("admits again when a retry rebuilds the same projection from the database", async () => { const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index a5c799ff64a..23f9ba7c2e6 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -47,7 +47,7 @@ import { logger } from "../../middleware/logger.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; -import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { nativeRestartContinuationInFlight, nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js"; @@ -678,7 +678,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: { /** Durable evidence proves a claimed remote bootstrap reached no provider work. */ remoteBootstrapProvenIncomplete?: boolean; remoteBootstrapIncompleteReason?: "remote_runner_never_launched" | "remote_session_opened_without_turn" - | "remote_runner_exited_before_provider_work"; + | "remote_runner_exited_before_provider_work" | "remote_runner_never_received_commands"; processStartMatches: boolean; knownProviderProcessAlive?: boolean; knownProviderProcessIdentityAmbiguous?: boolean; @@ -793,6 +793,16 @@ async function hasNativeRunnerOriginatedEvidence( return event !== undefined; } +async function hasNativeRunEvent( + tx: Pick, run: { companyId: string; id: string }, eventType: string, +): Promise { + const [event] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, eventType))) + .limit(1); + return event !== undefined; +} + export async function hasNativeExecutionStartEvidence( tx: Pick, run: { companyId: string; id: string }, ): Promise { @@ -1408,6 +1418,24 @@ export async function claimNativeRestartRecoveries(input: { const remoteBootstrapProvenIncomplete = remoteBootstrapProof !== null && (!hasProviderEvidence || remoteBootstrapProof === "session_opened_without_turn"); + const admissionCheckpointOnly = !hasCheckpoint || (checkpointIdentityMatches && checkpointRecord.cursor == null + && (!Array.isArray(checkpointRecord.terminalTurns) || checkpointRecord.terminalTurns.length === 0) + && checkpointRecord.activeTurnId == null && checkpointRecord.semanticResult == null && checkpointRecord.terminal == null); + const remoteRunnerNeverReceivedCommands = !unclaimedBootstrap && !remoteRunnerNeverLaunched + && !remoteRunnerExitedBeforeProviderWork && !remoteBootstrapProvenIncomplete + && remoteSandbox && remoteWorkspace !== null && !pluginTask && !durableAuthority && stagedAttachment === null + && admissionCheckpointOnly && providerEvents.length === 0 + && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION + && row.coordinator.phase === "observed" && Boolean(row.coordinator.leaseOwner) + && row.coordinator.attempt >= 1 && row.coordinator.controllerGeneration >= 1 + && Boolean(row.coordinator.controllerBootId && row.coordinator.controllerPid && row.coordinator.controllerProcessStartedAt) + && !row.coordinator.failureDetail?.successorRunId + && !row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt + && !row.run.sessionIdBefore && !row.run.sessionIdAfter + && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId) + && !(await hasNativeRunnerOriginatedEvidence(tx, row.run)) + && !(await hasNativeRunEvent(tx, row.run, PROCESS_IDENTITY_RECORDED)); + // Read before this claim appends its own history entry. const cleanDrainSuspension = suspendedByCleanDrain(row.coordinator); const resumeKeepsAttempt = cleanDrainSuspension || await restartResumeKeepsAttempt(tx, { @@ -1417,9 +1445,10 @@ export async function claimNativeRestartRecoveries(input: { remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority), unclaimedBootstrap, remoteBootstrapProvenIncomplete: remoteBootstrapProvenIncomplete || remoteRunnerNeverLaunched - || remoteRunnerExitedBeforeProviderWork, + || remoteRunnerExitedBeforeProviderWork || remoteRunnerNeverReceivedCommands, remoteBootstrapIncompleteReason: remoteRunnerNeverLaunched ? "remote_runner_never_launched" : remoteRunnerExitedBeforeProviderWork ? "remote_runner_exited_before_provider_work" + : remoteRunnerNeverReceivedCommands ? "remote_runner_never_received_commands" : remoteBootstrapProof === "session_opened_without_turn" ? "remote_session_opened_without_turn" : undefined, runnerPidAlive, runnerGroupAlive, @@ -1427,7 +1456,7 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessAlive: providerProcesses.livePids.length > 0, knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, - hasCheckpoint, + hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands, stagedAttachmentVerified: stagedAttachment !== null, checkpointIdentityMatches: checkpointIdentityMatches && @@ -1436,7 +1465,8 @@ export async function claimNativeRestartRecoveries(input: { checkpointFailed: (checkpointRecord.terminal as Record | undefined) ?.runTerminalState === "failed" && - nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null, + nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null && + !nativeRestartContinuationInFlight(checkpointRecord as unknown as PersistedNativeSession), providerAttempt: row.coordinator.attempt, resumeChargesAttempt: !resumeKeepsAttempt, }); diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 31ef5944755..26b9a82da58 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2690,6 +2690,34 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); return fixture; } + it.each([ + { label: "restarts a remote follow-up", identityRecorded: false, inherited: true }, + { label: "restarts a first remote run", identityRecorded: false, inherited: false }, + { label: "keeps a recorded runner process held instead of restarting", identityRecorded: true, inherited: true }, + ])("$label whose controller never sent its runner a command", async ({ identityRecorded, inherited }) => { + const fixture = await seedUnlaunchedRemoteClaim(`NEVER-COMMANDED-${identityRecorded}-${inherited}`, + { event: { eventType: "native.process_start_requested" } }); + const [seeded] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + if (inherited) await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...seeded.runnerProfileJson, sessionCheckpoint: { + driverKind: "codex_app_server", backendKind: "runner", sessionId: "inherited-thread", cursor: null, + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "inherited-thread", activeTurnId: null, terminal: null, terminalTurns: [], semanticResult: null, + pendingRuntimeRequests: [], lineage: [], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + if (identityRecorded) await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, seq: 901, + stream: "system", level: "info", eventType: "native.process_identity_recorded" }); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + if (identityRecorded) { + expect(claim).not.toMatchObject({ kind: "bootstrap_incomplete" }); + return; + } + expect(claim).toMatchObject({ kind: "bootstrap_incomplete", runId: fixture.runId, controllerGeneration: 2 }); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_runner_never_received_commands" }); + }); + it("restarts a remote bootstrap retry that shutdown cut before any runner was requested", async () => { const fixture = await seedUnlaunchedRemoteClaim("BOOTSTRAP-RETRY", { bootstrapRetry: true }); expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, @@ -2865,8 +2893,6 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); }); it.each<[string, Parameters[1], { kind: string; reason: string }]>([ - ["a runner process start was requested", { event: { eventType: "native.process_start_requested" } }, - { kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }], ["a runner process identity was recorded", { event: { eventType: "native.process_identity_recorded" } }, { kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }], ["the adapter was invoked", { event: { eventType: "adapter.invoke" } }, @@ -2892,8 +2918,10 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); { attempt: 2, driverKind: "opencode_server" }, // A second restart that loses the first restart's continuation turn. { attempt: 2, driverKind: "codex_app_server", lostContinuation: true }, - ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind, lostContinuation }) => { - const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}${lostContinuation ? "-CONTINUATION" : ""}`); + { attempt: 2, driverKind: "codex_app_server", lostContinuation: true, staleTerminal: true }, + { attempt: 2, driverKind: "acpx_runtime", lostContinuation: true, staleTerminal: true, heldResult: true }, + ])("recovers a checkpointed $driverKind restart interruption within provider budget $attempt", async ({ attempt, driverKind, lostContinuation, staleTerminal, heldResult }) => { + const fixture = await seedRun(`INTERRUPTED-${driverKind}-${attempt}${lostContinuation ? "-CONTINUATION" : ""}${staleTerminal ? "-STALE" : ""}${heldResult ? "-RESULT" : ""}`); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { driverKind, sessionId: fixture.native.normalizedSessionId, @@ -2908,7 +2936,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); }) }], // A restart during the first continuation: its turn is still the // checkpointed active turn and the consumed marker names it. - ...(lostContinuation ? { activeTurnId: "continuation-turn", terminal: null, + ...(heldResult ? { semanticResult: { schema: "paperclip.run_result.v1", summary: "Saved before the restart." } } : {}), + ...(lostContinuation ? { activeTurnId: "continuation-turn", ...(staleTerminal ? {} : { terminal: null }), dispositionOnlyRecoveryConsumed: true, dispositionOnlyRecoveryTurnId: "continuation-turn" } : {}), } } }).where(eq(heartbeatRuns.id, fixture.runId)); await fixture.db.update(nativeRunFinalizations).set({ attempt }).where(eq(nativeRunFinalizations.runId, fixture.runId)); @@ -2923,6 +2952,26 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); : { kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]); }); + it("keeps a run whose active turn follows a real provider failure blocked as permanently failed", async () => { + const fixture = await seedRun("ACTIVE-AFTER-PROVIDER-FAILURE"); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { + driverKind: "codex_app_server", + sessionId: fixture.native.normalizedSessionId, + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "failed-provider", + activeTurnId: "later-turn", + semanticResult: null, + terminal: { runTerminalState: "failed", turnTerminalState: "failed" }, + terminalTurns: [{ turnId: "failed-turn", fingerprint: JSON.stringify({ + terminalState: "failed", result: null, error: { code: "model_not_found", recoverable: false }, + }) }], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }, { eventType: "turn.accepted", turnId: "later-turn" }]); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, restartKind: "hot", runIds: [fixture.runId] })) + .toEqual([expect.objectContaining({ kind: "blocked", runId: fixture.runId, reason: "provider_checkpoint_permanently_failed" })]); + }); + // RC-22b: controller crashes (no drain record) that each interrupted a resume // which had already done work exhausted the three-attempt budget. async function seedInterruptedCheckpoint(label: string) { diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index fb0650cd6ef..c1b67f1737f 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -19641,6 +19641,8 @@ describe("runnerd provider runtime wiring", () => { ...["resumed", "replacement"].map(outcome => ({ outcome, scenario: "fresh", homeMode: "preconfigured" })), ...["interrupted_bootstrap", "interrupted_bootstrap_other_run", "interrupted_bootstrap_unproven", "interrupted_bootstrap_symlink_parent"] .map(scenario => ({ outcome: "resumed", scenario, homeMode: "isolated" })), + { outcome: "created", scenario: "interrupted_bootstrap", homeMode: "isolated" }, + { outcome: "created", scenario: "interrupted_bootstrap_symlink_parent", homeMode: "isolated" }, ])( "bootstraps only an untouched provider session in a $outcome workspace lease: $scenario ($homeMode)", async ({ outcome, scenario, homeMode }) => { const remoteCwd = join(isolatedStateDirectory, "remote"); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c74977841dd..c2d683a6cbe 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -267,7 +267,7 @@ import { remoteRecoveryProbeReason, } from "./remote-runner-recovery-diagnostics.js"; import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js"; -import { describeOwnFailedAttachment, inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, summarizePostTerminalEvents, type RemoteWarmSessionRecovery } from "./remote-warm-session-recovery.js"; +import { describeOwnFailedAttachment, inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, summarizePostTerminalEvents, type RemoteWarmSessionRecovery, unsettledControllerCommands } from "./remote-warm-session-recovery.js"; import { assertNativeWarmIngressRetirement, nativeWarmIngressAllocation, nativeWarmIngressIdentity, type NativeWarmIngressClaim, type NativeWarmIngressOwner } from "./native-warm-ingress-ownership.js"; import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js"; import { emitStartupReuse, environmentStartupReuse } from "@paperclipai/adapter-utils/startup-reuse"; @@ -5458,7 +5458,7 @@ function readSettledRunnerdProjection(root: string, identity: Record command.status !== "completed") + const unsettled = unsettledControllerCommands(settledCommands) .map(command => `${retirementReasonName("commandType", command.type)}.${retirementReasonName("commandStatus", command.status)}`); return refuse(unsettled.length > 0 ? `journal_commands_unsettled(${[...new Set(unsettled)].slice(0, 4).join(",")})` : "journal_terminal_unsettled"); @@ -5808,7 +5808,7 @@ async function migrateRunnerdStateRootForExecution(input: { if (result.kind !== "state") throw new Error("Plugin Runner checkpoint unavailable"); savePluginRunnerCheckpoint(scoped, result.state, identity); } - if (input.restartRecovery?.kind === "bootstrap_incomplete") { + if (input.restartRecovery?.kind === "bootstrap_incomplete" && identity.runId === input.execution.binding.runId) { // A local runner's process ids belong to this host, so a fresh provider // session it opened before any turn can be proven gone here. const localProcesses = !input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local"; @@ -5883,7 +5883,8 @@ async function migrateRunnerdStateRootForExecution(input: { // Name the failed retirement proof: the shared error otherwise hides // which of several independent fences held the follow-up. let retirementRefusal = ""; - if (verification === "terminal_state_indeterminate" && !input.restartRecovery && !input.retainedWarmOwnerRunId) { + if (verification === "terminal_state_indeterminate" && !input.retainedWarmOwnerRunId + && (!input.restartRecovery || input.restartRecovery.kind === "bootstrap_incomplete")) { const retirement = await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity }); if (retirement.retired) return; if (retirement.refused !== "lifecycle_not_per_turn") retirementRefusal = `; retirement_refused=${retirement.refused}`; @@ -15673,6 +15674,15 @@ async function createRunnerdBackendWithinSessionClaim( // lease was not provider-confirmed lost. Never silently create a new // provider session from that ambiguous state. throw new Error("runner_harness_state_mismatch: backup_without_reusable_lease"); + } else if (state.incompleteReason === "not_suspended" + && input.restartRecovery?.kind === "bootstrap_incomplete" + && state.notSuspendedRunId === input.execution.binding.runId + && hasRemoteBootstrapRestartProof(input.execution.binding.runId)) { + if (!(await moveInterruptedBootstrapSessionAside() + && takeRemoteBootstrapRestartProof(input.execution.binding.runId) + && await claimUntouchedRemoteSession())) { + throw new Error("runner_harness_state_mismatch: created_not_suspended"); + } } } } else if (remoteTarget && remoteCommandRunner) { diff --git a/server/src/services/native-runtime/remote-warm-session-recovery.ts b/server/src/services/native-runtime/remote-warm-session-recovery.ts index 40324702407..d1759466ed7 100644 --- a/server/src/services/native-runtime/remote-warm-session-recovery.ts +++ b/server/src/services/native-runtime/remote-warm-session-recovery.ts @@ -31,10 +31,32 @@ function postTerminalEvents(control: Record): unknown[] { /** A session is settled when its commands finished and nothing is live or waiting. * Events after the terminal describe the past, so refusing cannot undo them: they * are reported instead. Every event must still authenticate as this session's. */ +const RUNNER_LIFECYCLE_COMMAND_TYPES = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]); + +function outcomeUnknown(command: Record): boolean { + return command.status === "indeterminate" || + (command.status === "failed" && (record(command.result).code === "execution_indeterminate" || + record(record(command.result).result).code === "execution_indeterminate")); +} + +export function unsettledControllerCommands(commands: readonly Record[]): Record[] { + const lastCompletedTurnStart = Math.max(-1, ...commands + .filter(command => command.type === "turn.start" && command.status === "completed") + .map(command => Number(command.controllerSeq))); + return commands.filter(command => { + if (command.status === "completed") return false; + if (outcomeUnknown(command)) { + return !(command.type === "turn.start" && Number(command.controllerSeq) < lastCompletedTurnStart); + } + return !((command.status === "failed" || command.status === "rejected") && + !RUNNER_LIFECYCLE_COMMAND_TYPES.has(String(command.type))); + }); +} + export function isSettledRemoteWarmControlState(control: Record, identity: Record): boolean { const commands = control.commands; const events = control.committedEvents; - if (!Array.isArray(commands) || commands.some(command => record(command).status !== "completed") || + if (!Array.isArray(commands) || unsettledControllerCommands(commands.map(record)).length > 0 || !Array.isArray(events) || Object.entries(identity).some(([key, value]) => record(control.identity)[key] !== value)) return false; const terminalIndex = events.findLastIndex(event => record(event).eventType === "run.terminal"); if (terminalIndex < 0) return false; diff --git a/server/src/services/skill-update-diff.ts b/server/src/services/skill-update-diff.ts new file mode 100644 index 00000000000..10bc2b5a448 --- /dev/null +++ b/server/src/services/skill-update-diff.ts @@ -0,0 +1,170 @@ +import path from 'node:path'; +import type { CompanySkillVersionFileInventoryEntry, SkillSourceUpdateDiffFile, SkillSourcePackageMode } from '@paperclipai/shared'; +import { unprocessable } from '../errors.js'; +import { scanGitHubSkillsWithPackageFallback, type GitHubRead, type ScannedSkillSource } from './github-skill-source.js'; + +// Bounded LCS work for one file, after trimming the shared head and tail of the two files. +const MAX_DIFF_CELLS = 4_000_000; +// One response shares these budgets across all of its files, so a large update +// cannot hold the event loop or produce an unbounded response body. +export const MAX_UPDATE_DIFF_CELLS = 24_000_000; +export const MAX_UPDATE_DIFF_CHARS = 1_000_000; +export const MAX_UPDATE_DIFF_FILES = 1_000; +const CONTEXT_LINES = 3; +const NO_FINAL_NEWLINE = '\n\\ No newline at end of file'; + +type Row = { kind: ' ' | '-' | '+'; text: string; oldLine: number; newLine: number }; +type LineDiff = { diff: string | null; cells: number }; + +/** The lines of a file. An empty file has none; a last line without a final newline carries the unified-diff marker. */ +function fileLines(text: string): string[] { + if (text === '') return []; + const lines = text.split('\n'); + if (lines.at(-1) === '') lines.pop(); + else lines[lines.length - 1] += NO_FINAL_NEWLINE; + return lines; +} + +function lineDiff(before: string, after: string, maxCells: number): LineDiff { + if (before === after) return { diff: '', cells: 0 }; + const a = fileLines(before), b = fileLines(after); + let head = 0; + while (head < a.length && head < b.length && a[head] === b[head]) head++; + let tail = 0; + while (tail < a.length - head && tail < b.length - head && a[a.length - 1 - tail] === b[b.length - 1 - tail]) tail++; + const midA = a.slice(head, a.length - tail), midB = b.slice(head, b.length - tail); + const cells = (midA.length + 1) * (midB.length + 1); + if (cells > maxCells) return { diff: null, cells }; + const lcs = Array.from({ length: midA.length + 1 }, () => new Uint32Array(midB.length + 1)); + for (let i = midA.length - 1; i >= 0; i--) for (let j = midB.length - 1; j >= 0; j--) + lcs[i]![j] = midA[i] === midB[j] ? lcs[i + 1]![j + 1]! + 1 : Math.max(lcs[i + 1]![j]!, lcs[i]![j + 1]!); + const rows: Row[] = []; + for (let k = 0; k < head; k++) rows.push({ kind: ' ', text: a[k]!, oldLine: k + 1, newLine: k + 1 }); + let i = 0, j = 0; + while (i < midA.length || j < midB.length) { + if (i < midA.length && j < midB.length && midA[i] === midB[j]) { rows.push({ kind: ' ', text: midA[i]!, oldLine: head + i + 1, newLine: head + j + 1 }); i++; j++; } + // On a tie the installed line goes first, so a replaced line reads "-old" then "+new". + else if (i < midA.length && (j >= midB.length || lcs[i + 1]![j]! >= lcs[i]![j + 1]!)) { rows.push({ kind: '-', text: midA[i]!, oldLine: head + i + 1, newLine: head + j }); i++; } + else { rows.push({ kind: '+', text: midB[j]!, oldLine: head + i, newLine: head + j + 1 }); j++; } + } + for (let k = 0; k < tail; k++) rows.push({ kind: ' ', text: a[a.length - tail + k]!, oldLine: a.length - tail + k + 1, newLine: b.length - tail + k + 1 }); + const changed = rows.map((row, index) => row.kind === ' ' ? -1 : index).filter(index => index >= 0); + const hunks: Array<[number, number]> = []; + for (const index of changed) { + const start = Math.max(0, index - CONTEXT_LINES), end = Math.min(rows.length - 1, index + CONTEXT_LINES); + const last = hunks.at(-1); + if (last && start <= last[1] + 1) last[1] = Math.max(last[1], end); else hunks.push([start, end]); + } + const diff = hunks.map(([start, end]) => { + const slice = rows.slice(start, end + 1); + const oldCount = slice.filter(row => row.kind !== '+').length, newCount = slice.filter(row => row.kind !== '-').length; + // As in unified diff, a side with no lines starts at the line before the hunk (0 for an empty file). + const oldStart = oldCount ? slice.find(row => row.kind !== '+')!.oldLine : slice[0]!.oldLine; + const newStart = newCount ? slice.find(row => row.kind !== '-')!.newLine : slice[0]!.newLine; + return [`@@ -${oldStart},${oldCount} +${newStart},${newCount} @@`, ...slice.map(row => `${row.kind}${row.text}`)].join('\n'); + }).join('\n'); + return { diff, cells }; +} + +/** Unified line diff from `before` (installed) to `after` (latest); "" when equal, null when too large. */ +export function unifiedLineDiff(before: string, after: string): string | null { + return lineDiff(before, after, MAX_DIFF_CELLS).diff; +} + +type DiffFile = Pick; +/** Paths that a source scan left out of each snapshot (too large, a symlink, or rejected by the audit). */ +export type OmittedPaths = { installed: ReadonlySet; latest: ReadonlySet }; +export type SkillPackageDiff = { files: SkillSourceUpdateDiffFile[]; truncated: boolean }; + +/** Per-file changes between two snapshots of one skill package, sorted by path. */ +export function diffSkillPackageFiles(installed: DiffFile[], latest: DiffFile[], omitted: OmittedPaths = { installed: new Set(), latest: new Set() }): SkillPackageDiff { + const before = new Map(installed.map(file => [file.path, file])), after = new Map(latest.map(file => [file.path, file])); + const text = (file: DiffFile) => file.encoding !== 'base64' ? file.content : null; + const paths = [...new Set([...before.keys(), ...after.keys(), ...omitted.installed, ...omitted.latest])].sort(); + const files: SkillSourceUpdateDiffFile[] = []; + let cells = 0, chars = 0, truncated = false; + for (const filePath of paths) { + const old = before.get(filePath), next = after.get(filePath); + if (omitted.installed.has(filePath) || omitted.latest.has(filePath)) { + // Without both snapshots of this path, any added/removed/modified claim would be a guess. + files.push({ path: filePath, change: 'not_compared', binary: false, executableChanged: false, diff: null, diffUnavailable: 'omitted' }); + } else { + const sameContent = !!old && !!next && old.content === next.content && (old.encoding ?? 'utf8') === (next.encoding ?? 'utf8'); + const executableChanged = !!old && !!next && !!old.executable !== !!next.executable; + if (sameContent && !executableChanged) continue; + const change = !old ? 'added' as const : !next ? 'removed' as const : 'modified' as const; + const oldText = old ? text(old) : '', newText = next ? text(next) : ''; + const binary = oldText === null || newText === null; + let diff: string | null = null; + let diffUnavailable: SkillSourceUpdateDiffFile['diffUnavailable'] = binary ? 'binary' : null; + if (!binary) { + const result = lineDiff(oldText, newText, Math.min(MAX_DIFF_CELLS, MAX_UPDATE_DIFF_CELLS - cells)); + if (result.diff === null) { + diffUnavailable = result.cells > MAX_DIFF_CELLS ? 'too_large' : 'budget'; + truncated ||= diffUnavailable === 'budget'; + } else if (chars + result.diff.length > MAX_UPDATE_DIFF_CHARS) { + cells += result.cells; + diffUnavailable = 'budget'; + truncated = true; + } else { + cells += result.cells; + chars += result.diff.length; + diff = result.diff; + } + } + files.push({ path: filePath, change, binary, executableChanged, diff, diffUnavailable }); + } + if (files.length >= MAX_UPDATE_DIFF_FILES && filePath !== paths.at(-1)) { truncated = true; break; } + } + return { files, truncated }; +} + +const OMITTED_FILE_WARNING = /^(.+?) — .*(?:not included|couldn’t be included)\.?$/; +function omittedPaths(warnings: Iterable, toPackagePath: (repositoryPath: string) => string | null) { + const paths = new Set(); + for (const warning of warnings) { + const match = OMITTED_FILE_WARNING.exec(warning); + const packagePath = match ? toPackagePath(match[1]!) : null; + if (packagePath) paths.add(packagePath); + } + return paths; +} + +type Snapshot = { files: DiffFile[]; omitted: Set } | { partialReason: string }; +function packageSnapshot(scan: ScannedSkillSource, input: { packageMode?: SkillSourcePackageMode; skillPath: string }): Snapshot { + if (input.packageMode === 'repository') { + // A repository that outgrew the package limits is scanned as a subset or as + // skill folders only; files missing from such a scan may still exist upstream. + if (scan.packageMode !== 'repository' || scan.repositorySubset || !scan.repositoryFiles) { + return { partialReason: 'The repository is larger than the supported package size, so its revisions cannot be compared file by file.' }; + } + return { files: scan.repositoryFiles, omitted: omittedPaths([...scan.warnings, ...(scan.repositoryPackage?.warnings ?? [])], repositoryPath => repositoryPath) }; + } + const skill = scan.skills.find(candidate => candidate.path === input.skillPath); + const dir = path.posix.dirname(input.skillPath); + const prefix = dir === '.' ? '' : `${dir}/`; + const toPackagePath = (repositoryPath: string) => repositoryPath === input.skillPath ? 'SKILL.md' + : repositoryPath.startsWith(prefix) ? repositoryPath.slice(prefix.length) : null; + return { files: skill?.files ?? [], omitted: omittedPaths(skill?.warnings ?? [], toPackagePath) }; +} + +export type GitHubSkillUpdateDiff = ({ supported: true } & SkillPackageDiff) | { supported: false; reason: string }; + +export async function diffGitHubSkillUpdate( + input: { repositoryUrl: string; packageMode?: SkillSourcePackageMode; skillPath: string; installedCommitSha: string; latestCommitSha: string }, + read: GitHubRead, + signal?: AbortSignal, +): Promise { + const snapshotAt = async (commitSha: string) => { + signal?.throwIfAborted(); + const scan = await scanGitHubSkillsWithPackageFallback({ repositoryUrl: input.repositoryUrl, packageMode: input.packageMode, commitSha, onlySkillPath: input.skillPath }, read, { signal }); + if (scan.commitSha.toLowerCase() !== commitSha.toLowerCase()) throw unprocessable('The source snapshot did not match the requested revision.'); + return packageSnapshot(scan, input); + }; + const installed = await snapshotAt(input.installedCommitSha); + if ('partialReason' in installed) return { supported: false, reason: installed.partialReason }; + const latest = await snapshotAt(input.latestCommitSha); + if ('partialReason' in latest) return { supported: false, reason: latest.partialReason }; + signal?.throwIfAborted(); + return { supported: true, ...diffSkillPackageFiles(installed.files, latest.files, { installed: installed.omitted, latest: latest.omitted }) }; +} diff --git a/server/src/services/tool-access-policy.ts b/server/src/services/tool-access-policy.ts index 7f0192453ae..839c2a43c54 100644 --- a/server/src/services/tool-access-policy.ts +++ b/server/src/services/tool-access-policy.ts @@ -49,6 +49,7 @@ import { toolPolicyConditionsSchema } from "@paperclipai/shared"; import { badRequest, conflict, notFound, unprocessable } from "../errors.js"; import { effectiveToolProfileBindings, + isConnectionAccessProfile, profileIdsInBindingOrder, } from "./tool-profile-binding-precedence.js"; import { recordToolRuntimeAuditWriteFailure } from "./tool-runtime-metrics.js"; @@ -760,8 +761,21 @@ function rateBucket(rule: ToolRateLimitRule, ctx: ToolAccessContext): string { }).join("|"); } -function isScopedToolGrant(scope: unknown) { - return isRecord(scope) && Object.keys(scope).length > 0; +// Grant scope keys that name which tools the grant covers: an action, a +// connection or an application. Context keys such as agentId or projectId only +// narrow when the grant applies, so a grant with only those still covers every +// tool and does not count as a choice about one connection's actions. +const TOOL_GRANT_TARGET_SCOPE_KEYS = new Set([ + "allow", + "toolName", "toolNames", + "catalogEntryId", "catalogEntryIds", + "connectionId", "connectionIds", + "applicationId", "applicationIds", + "applicationKey", "applicationKeys", +]); + +function isToolTargetedGrant(scope: unknown) { + return isRecord(scope) && Object.keys(scope).some((key) => TOOL_GRANT_TARGET_SCOPE_KEYS.has(key)); } function scopeAllowsTool(scope: Record | null, ctx: ToolAccessContext) { @@ -1250,7 +1264,7 @@ export function toolAccessPolicyService(db: Pick { + async function explicitGrant(ctx: ToolAccessContext, cache?: ToolAccessDecisionCache, options: { toolTargetedOnly?: boolean } = {}): Promise { const principalType = ctx.actorType === "agent" ? "agent" : ctx.actorType === "user" ? "user" : null; const principalId = ctx.actorType === "agent" ? ctx.agentId : ctx.actorId; if (!principalType || !principalId) return false; @@ -1264,7 +1278,7 @@ export function toolAccessPolicyService(db: Pick (!options.scopedOnly || isScopedToolGrant(grant.scope)) && scopeAllowsTool(grant.scope, ctx)); + return grants.some((grant) => (!options.toolTargetedOnly || isToolTargetedGrant(grant.scope)) && scopeAllowsTool(grant.scope, ctx)); } async function enforceRateLimit(policy: typeof toolPolicies.$inferSelect, ctx: ToolAccessContext, consume: boolean) { @@ -1657,11 +1671,16 @@ export function toolAccessPolicyService(db: Pick entry.connectionId === ctx.connectionId); - if (await explicitGrant(ctx, cache, { scopedOnly: governedByConnectionProfile })) { + // profile governing this connection leaves off. The connection's own + // access profile governs it even with no entries (every action turned + // off), and so does any entry that selects this action, connection or + // application. A grant that names the action, connection or application + // still applies. Board users keep their operator grants. + const governedByConnectionProfile = ctx.actorType === "agent" && !!ctx.connectionId && ( + profileState.profiles.some((profile) => isConnectionAccessProfile(profile, ctx.connectionId)) + || profileState.entries.some((entry) => entry.connectionId === ctx.connectionId || profileEntryMatches(entry, ctx)) + ); + if (await explicitGrant(ctx, cache, { toolTargetedOnly: governedByConnectionProfile })) { return decision("allow", "allow_explicit_grant", "Tool access allowed by explicit grant.", effectiveProfileIds, [], { redactionPlan: redaction.redactionPlan }); } diff --git a/server/src/services/tool-profile-binding-precedence.ts b/server/src/services/tool-profile-binding-precedence.ts index 73b94c71314..c43711c80a0 100644 --- a/server/src/services/tool-profile-binding-precedence.ts +++ b/server/src/services/tool-profile-binding-precedence.ts @@ -56,7 +56,12 @@ export function profileIdsInBindingOrder` profile or an agent's `connection-intent` profile. + * With a connection id, only that connection's profiles match. + */ +export function isConnectionAccessProfile(profile: ProfileLike, connectionId?: string | null): boolean { if (!profile.metadata || typeof profile.metadata !== "object" || Array.isArray(profile.metadata)) return false; const metadata = profile.metadata as Record; if (metadata.source === "connection_intent" && typeof metadata.connectionId === "string" && typeof metadata.agentId === "string") { @@ -85,7 +90,7 @@ export function effectiveToolProfileBindings( return narrowestScopeBindings(bindings); } const appProfileIds = new Set( - profiles.filter((profile) => isWizardAppProfile(profile, connectionId)).map((profile) => profile.id), + profiles.filter((profile) => isConnectionAccessProfile(profile, connectionId)).map((profile) => profile.id), ); const accessProfileIds = new Set(profiles.filter(profile => { const metadata = profile.metadata as Record | null; diff --git a/server/src/vendor/paperclip-runner/index.ts b/server/src/vendor/paperclip-runner/index.ts index 38108ec7a08..732c76e741d 100644 --- a/server/src/vendor/paperclip-runner/index.ts +++ b/server/src/vendor/paperclip-runner/index.ts @@ -193,6 +193,7 @@ export const NativeProviderStateIncompatibleError = runner.NativeProviderStateIn export const NativeProviderSideEffectPendingError = runner.NativeProviderSideEffectPendingError; export const NativeRunnerSourceReconciliationRequiredError = runner.NativeRunnerSourceReconciliationRequiredError; export const nativeRestartInterruptedTurnId = runner.nativeRestartInterruptedTurnId; +export const nativeRestartContinuationInFlight = runner.nativeRestartContinuationInFlight; export const completeTerminatedRemoteNativeSessionCleanup = runner.completeTerminatedRemoteNativeSessionCleanup; export const completeTerminatedLocalNativeSessionCleanup = runner.completeTerminatedLocalNativeSessionCleanup; diff --git a/tests/restart-chaos/main.ts b/tests/restart-chaos/main.ts index 98ec0309fbe..2f584a3e888 100644 --- a/tests/restart-chaos/main.ts +++ b/tests/restart-chaos/main.ts @@ -236,7 +236,7 @@ async function main(): Promise { try { await supervisor.start(serverEnvironment); const api = new ChaosApi(supervisor.baseUrl); - const lab = await seedChaosLab({ api, env: plan.options.env, daytonaImage: plan.options.daytonaImage, daytonaCpu: plan.options.daytonaCpu, + const lab = await seedChaosLab({ api, env: plan.options.env, waitForReadyAgents: true, cancelled: () => interrupted, daytonaImage: plan.options.daytonaImage, daytonaCpu: plan.options.daytonaCpu, daytonaReuseLease: plan.options.daytonaReuseLease, adapters: [...new Set([...plan.tasks.map(task => task.adapter), ...(plan.tasks.some(task => task.workload === "M1") ? plan.options.adapters : [])])], diff --git a/tests/restart-chaos/seed.test.ts b/tests/restart-chaos/seed.test.ts index c00295926d0..24891ee091d 100644 --- a/tests/restart-chaos/seed.test.ts +++ b/tests/restart-chaos/seed.test.ts @@ -5,7 +5,7 @@ import os from "node:os"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import type { ChaosApi } from "./api.js"; -import { cloneWorkspaceRepository, createFixtureRepository, seedChaosLab } from "./seed.js"; +import { cloneWorkspaceRepository, createFixtureRepository, seedChaosLab, waitForAgentsReady } from "./seed.js"; import { Timeline } from "./timeline.js"; type Call = { method: string; route: string; body?: unknown }; @@ -18,6 +18,7 @@ function fakeApi(experimentalResponse: (body: Record) => Record get: async (route: string) => { calls.push({ method: "GET", route }); if (route.includes("/environments")) return [{ id: "environment-1", driver: "local" }]; + if (route.startsWith("/api/agents/")) return { id: route.split("/")[3], lifecycleState: "ready" }; throw new Error(`unexpected GET ${route}`); }, post: async (route: string, body?: unknown) => { @@ -126,3 +127,35 @@ describe("cloneWorkspaceRepository", () => { } }); }); + +describe("waitForAgentsReady", () => { + it("waits for each new agent to finish lifecycle verification and records how long it took", async () => { + let reads = 0; + const api = { get: async () => ({ lifecycleState: ++reads >= 3 ? "ready" : "verifying" }) } as unknown as ChaosApi; + const timeline = new Timeline(null); + await waitForAgentsReady(api, { codex: { id: "a1" } }, timeline, 5_000, 1); + expect(reads).toBe(3); + expect(timeline.entries.find(entry => entry.kind === "seed_agent_ready")).toMatchObject({ adapter: "codex", agentId: "a1" }); + }); + + it("measures each agent's verification from its own creation and stops waiting when the harness is interrupted", async () => { + const api = { get: async () => ({ lifecycleState: "ready" }) } as unknown as ChaosApi; + const timeline = new Timeline(null); + const now = Date.now(); + await waitForAgentsReady(api, { codex: { id: "a1", createdAt: now - 9_000 }, claude: { id: "a2", createdAt: now - 1_000 } }, timeline, 5_000, 1); + const ready = Object.fromEntries(timeline.entries.filter(entry => entry.kind === "seed_agent_ready").map(entry => [entry.adapter, entry.readyAfterMs as number])); + expect(ready.codex).toBeGreaterThanOrEqual(9_000); + expect(ready.claude).toBeGreaterThanOrEqual(1_000); + expect(ready.claude).toBeLessThan(9_000); + const stuck = { get: async () => ({ lifecycleState: "verifying" }) } as unknown as ChaosApi; + let polls = 0; + await expect(waitForAgentsReady(stuck, { codex: { id: "a3" } }, new Timeline(null), 60_000, 1, () => ++polls > 2)).rejects.toThrow(/interrupted/); + }); + + it("fails the wave when an agent fails verification or never becomes ready", async () => { + const failed = { get: async () => ({ lifecycleState: "verifying", lifecycleError: "hello probe failed" }) } as unknown as ChaosApi; + await expect(waitForAgentsReady(failed, { claude: { id: "a2" } }, new Timeline(null), 5_000, 1)).rejects.toThrow(/hello probe failed/); + const stuck = { get: async () => ({ lifecycleState: "verifying" }) } as unknown as ChaosApi; + await expect(waitForAgentsReady(stuck, { claude: { id: "a3" } }, new Timeline(null), 20, 5)).rejects.toThrow(/never reached lifecycleState ready/); + }); +}); diff --git a/tests/restart-chaos/seed.ts b/tests/restart-chaos/seed.ts index 0796d7f86db..ee6783e9efd 100644 --- a/tests/restart-chaos/seed.ts +++ b/tests/restart-chaos/seed.ts @@ -22,6 +22,29 @@ export interface ChaosLab { } /** Create the `chaos-sandbox` git repository: python CLI, one failing test, bench.sh. */ +export async function waitForAgentsReady(api: ChaosApi, agents: Partial>, timeline: Timeline, + timeoutMs = 4 * 60_000, pollMs = 1_000, cancelled: () => boolean = () => false): Promise { + const started = Date.now(); + const pending = new Map(Object.entries(agents).flatMap(([adapter, agent]) => agent ? [[adapter, agent] as const] : [])); + while (pending.size) { + if (cancelled()) throw new Error("Harness interrupted while waiting for agents to finish verification"); + for (const [adapter, { id: agentId, createdAt }] of [...pending]) { + const agent = await api.get>(`/api/agents/${agentId}`).catch(() => null); + if (agent?.lifecycleState === "ready") { + timeline.add("seed_agent_ready", { adapter, agentId, readyAfterMs: Date.now() - (createdAt ?? started) }); + pending.delete(adapter); + } else if (agent?.lifecycleError) { + throw new Error(`Agent ${adapter} (${agentId}) failed lifecycle verification: ${String(agent.lifecycleError).slice(0, 300)}`); + } + } + if (!pending.size) return; + if (Date.now() - started > timeoutMs) { + throw new Error(`Agents never reached lifecycleState ready within ${Math.round(timeoutMs / 1000)} s: ${[...pending.keys()].join(", ")}`); + } + await new Promise(resolve => setTimeout(resolve, pollMs)); + } +} + export function cloneWorkspaceRepository(source: string, directory: string, commit?: string): string { const git = (...args: string[]) => execFileSync("git", args, { cwd: directory, encoding: "utf8", stdio: ["ignore", "pipe", "pipe"] }).trim(); execFileSync("git", ["clone", "--quiet", "--depth", "1", "--no-hardlinks", `file://${source}`, directory], { stdio: "pipe" }); @@ -130,6 +153,9 @@ export async function seedChaosLab(input: { budgetCents: number; masks?: readonly string[]; fixtureRoot: string; + waitForReadyAgents?: boolean; + cancelled?: () => boolean; + agentReadyTimeoutMs?: number; workspaceRepo?: string; workspaceCommit?: string; waveId: string; @@ -190,6 +216,7 @@ export async function seedChaosLab(input: { } const agents = {} as ChaosLab["agents"]; + const agentCreatedAt = new Map(); for (const adapter of input.adapters) { const profile = runnerProfiles.find(candidate => candidate.id === PROFILE_FOR_ADAPTER[adapter]); if (!profile) throw new Error(`runner-e2e catalog has no profile ${PROFILE_FOR_ADAPTER[adapter]}`); @@ -209,8 +236,13 @@ export async function seedChaosLab(input: { }; const created = await api.post<{ id: string; name: string }>(`/api/companies/${company.id}/agents`, body); agents[adapter] = { id: created.id, name, body }; + agentCreatedAt.set(adapter, Date.now()); timeline.add("seed_agent", { adapter, agentId: created.id, profile: profile.id, model: profile.model, environmentId }); } + if (input.waitForReadyAgents) { + await waitForAgentsReady(api, Object.fromEntries(Object.entries(agents).map(([adapter, agent]) => + [adapter, { id: agent.id, createdAt: agentCreatedAt.get(adapter) }])), timeline, input.agentReadyTimeoutMs, 1_000, input.cancelled); + } // The project's git_worktree policy only takes effect when the instance has // isolated workspaces enabled (the Experimental setting users turn on, as in