Merge remote-tracking branch 'origin/master' into codex/slack-origin-tools

* origin/master:
  refactor(server): remove retired operator UI snippet injection (#13789)
  fix(runner): keep warm sessions alive with managed GitHub access (#13815)
  fix: continue native agent chats after worker loss (#13813)
  fix: preserve chat message bindings in review recovery (#13818)
  chore(lockfile): refresh pnpm-lock.yaml (#13780)
  feat(ui): enable Grok in Cloud agent setup (#13791)

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-22 13:23:06 -05:00
commit 653c9ad2bf
39 files changed
+1183 -519

No files matched your search

@@ -1,102 +0,0 @@
import { describe, expect, it } from "vitest";
import { injectCloudUiSnippet } from "../cloud-ui-snippet.js";
const html = '<html><body><div id="root"></div></body></html>';
const snippet = '<script src="https://example.com/widget.js"></script>';
const encoded = Buffer.from(snippet, "utf-8").toString("base64");
describe("Cloud UI snippet", () => {
it("leaves self-hosted HTML unchanged even when a snippet is configured", () => {
expect(injectCloudUiSnippet(html, { PAPERCLIP_CLOUD_UI_SNIPPET: snippet })).toBe(html);
expect(injectCloudUiSnippet(html, { PAPERCLIP_CLOUD_UI_SNIPPET_B64: encoded })).toBe(html);
});
it.each([
{ PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN: "test-token" },
{ PAPERCLIP_MANAGED_CONFIG: "{}" },
])("injects only on a configured Cloud instance: %j", (cloud) => {
expect(injectCloudUiSnippet(html, { ...cloud, PAPERCLIP_CLOUD_UI_SNIPPET: snippet }))
.toBe(html.replace("</body>", `${snippet}\n</body>`));
expect(injectCloudUiSnippet(html, cloud)).toBe(html);
expect(injectCloudUiSnippet(html, { ...cloud, PAPERCLIP_CLOUD_UI_SNIPPET: " " })).toBe(html);
});
it("preserves literal replacement tokens in operator JavaScript", () => {
const script = '<script>console.log("$&", "$`", "$\'");</script>';
const result = injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}", PAPERCLIP_CLOUD_UI_SNIPPET: script,
});
expect(result).toContain(script);
expect(result).not.toContain("test-token");
});
it("decodes a base64 snippet on a Cloud instance", () => {
expect(injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}", PAPERCLIP_CLOUD_UI_SNIPPET_B64: encoded,
})).toBe(html.replace("</body>", `${snippet}\n</body>`));
});
it("tolerates whitespace and line wrapping in the base64 value", () => {
const wrapped = ` ${encoded.slice(0, 20)}\n${encoded.slice(20)}\n`;
expect(injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}", PAPERCLIP_CLOUD_UI_SNIPPET_B64: wrapped,
})).toBe(html.replace("</body>", `${snippet}\n</body>`));
});
it("wraps a bare script body that carries no markup in a <script> element", () => {
const body = '(()=>{window.__feedback=true;})();';
const wrapped = `<script>${body}</script>`;
expect(injectCloudUiSnippet(html, { PAPERCLIP_MANAGED_CONFIG: "{}", PAPERCLIP_CLOUD_UI_SNIPPET: body }))
.toBe(html.replace("</body>", `${wrapped}\n</body>`));
expect(injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}",
PAPERCLIP_CLOUD_UI_SNIPPET_B64: Buffer.from(body, "utf-8").toString("base64"),
})).toBe(html.replace("</body>", `${wrapped}\n</body>`));
});
it("preserves literal replacement tokens when wrapping a bare script body", () => {
const body = 'console.log("$&", "$`", "$\'");';
const result = injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}", PAPERCLIP_CLOUD_UI_SNIPPET: body,
});
expect(result).toContain(`<script>${body}</script>`);
});
it("prefers the plain snippet when both variables are set", () => {
const other = Buffer.from("<script>other()</script>", "utf-8").toString("base64");
const result = injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}",
PAPERCLIP_CLOUD_UI_SNIPPET: snippet,
PAPERCLIP_CLOUD_UI_SNIPPET_B64: other,
});
expect(result).toContain(snippet);
expect(result).not.toContain("other()");
});
it("treats a blank plain variable as disabled even when a base64 value is set", () => {
expect(injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}",
PAPERCLIP_CLOUD_UI_SNIPPET: " ",
PAPERCLIP_CLOUD_UI_SNIPPET_B64: encoded,
})).toBe(html);
expect(injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}",
PAPERCLIP_CLOUD_UI_SNIPPET: "",
PAPERCLIP_CLOUD_UI_SNIPPET_B64: encoded,
})).toBe(html);
});
it.each([
{ label: "invalid characters", value: "!!!not-base64!!!" },
{ label: "wrong length", value: "abcde" },
{ label: "unpadded", value: Buffer.from("<b>x</b>", "utf-8").toString("base64").replace(/=+$/, "") },
{ label: "carrying nonzero padding bits", value: "PB==" },
{ label: "not valid UTF-8 once decoded", value: "/w==" },
{ label: "blank once decoded", value: Buffer.from(" \n ", "utf-8").toString("base64") },
{ label: "blank", value: " " },
])("ignores a base64 value that is $label", ({ value }) => {
expect(injectCloudUiSnippet(html, {
PAPERCLIP_MANAGED_CONFIG: "{}", PAPERCLIP_CLOUD_UI_SNIPPET_B64: value,
})).toBe(html);
});
});
@@ -1,3 +1,4 @@
import { createNativeGitHubAccess } from "../services/native-runtime/native-github-access.js";
import express from "express";
import request from "supertest";
import { runtimeConnectionIntentRoutes } from "../routes/connection-intents.js";
@@ -800,6 +801,39 @@ const support = await getEmbeddedPostgresTestSupport();
}
});
it("reuses a session broker across runs while rechecking live-run identity and revocation", async () => {
const input = await seed();
await grant(input, "A");
await grant(input, "B");
const broker = await createNativeGitHubAccess({
scope: input, target: null, cwd: process.cwd(), env: { PATH: process.env.PATH },
resolveCredentials: (binding) => resolveGitHubOperationCredentials(db, binding),
});
const post = () => fetch(`${broker.env.PAPERCLIP_GITHUB_BROKER_URL}/runtime-tools/github/credentials`, {
method: "POST", headers: { authorization: `Bearer ${broker.env.PAPERCLIP_GITHUB_BRIDGE_TOKEN}` },
});
try {
const releaseA = broker.activate(input);
const a = await post();
expect(a.status).toBe(200);
expect((await a.json()).login).toBe("A");
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, input.runId));
// Even a delayed controller release cannot authorize a finished DB run.
expect((await post()).status).toBe(403);
releaseA();
const next = { ...input, runId: randomUUID() };
await db.insert(heartbeatRuns).values({ id: next.runId, companyId: next.companyId, agentId: next.agentId, status: "running", contextSnapshot: { issueId: input.issueId } });
await initializeRunIdentity(db, { companyId: input.companyId, runId: next.runId, responsibleUserId: "B", cause: "instruction" });
broker.activate(next);
const b = await post();
expect(b.status).toBe(200);
expect((await b.json()).login).toBe("B");
await db.update(connectionGrants).set({ status: "revoked" }).where(eq(connectionGrants.companyId, input.companyId));
const revoked = await post();
expect((await revoked.json()).env).toEqual({});
} finally { await broker.stop(); }
});
it("requires a run-scoped runtime capability and never accepts browser authentication or supplied identities", async () => {
const input = await seed();
await grant(input, "A");
@@ -30,14 +30,16 @@ import {
listAuthorizedChatAttachments,
resolveExternalChatResponseWaitAuthorization,
} from "../services/native-runtime/chat-attachment-reuse.js";
import { decideIssueReviewPathRecovery } from "../services/recovery/review-path-recovery.js";
import { resolveCurrentWakeCommentsBinding } from "../services/native-runtime/current-wake-comments.js";
describe("reviewed external-chat execution binding", () => {
describe.each(["slack", "discord"] as const)("reviewed %s execution binding", (provider) => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
const companyId = randomUUID(),
agentId = randomUUID(),
issueId = randomUUID(),
otherIssueId = randomUUID(),
runId = randomUUID();
const endpointId = randomUUID(),
resourceId = randomUUID(),
@@ -49,7 +51,7 @@ describe("reviewed external-chat execution binding", () => {
userId = "reviewed-chat-user";
const context = {
issueId,
source: "chat:discord",
source: `chat:${provider}`,
wakeReason: "External chat message received",
wakeCommentIds: [commentId],
commentId,
@@ -93,6 +95,15 @@ describe("reviewed external-chat execution binding", () => {
workMode: "standard",
assigneeAgentId: agentId,
});
await db.insert(issues).values({
id: otherIssueId,
companyId,
title: "Unrelated conversation",
issueNumber: 2,
identifier: "RCB-2",
status: "in_review",
assigneeAgentId: agentId,
});
await db
.insert(heartbeatRuns)
.values({
@@ -132,7 +143,7 @@ describe("reviewed external-chat execution binding", () => {
.values({
id: applicationId,
companyId,
applicationKey: `chat:discord:${endpointId}`,
applicationKey: `chat:${provider}:${endpointId}`,
name: "Discord",
type: "chat",
status: "active",
@@ -144,7 +155,7 @@ describe("reviewed external-chat execution binding", () => {
companyId,
applicationId,
name: "Discord",
uid: `chat-discord-${endpointId}`,
uid: `chat-${provider}-${endpointId}`,
connectionPurpose: "channel",
transport: "chat_sdk",
status: "active",
@@ -156,7 +167,7 @@ describe("reviewed external-chat execution binding", () => {
id: endpointId,
companyId,
connectionId,
provider: "discord",
provider,
publicId: randomUUID(),
assignedAgentId: agentId,
status: "active",
@@ -193,7 +204,7 @@ describe("reviewed external-chat execution binding", () => {
.values({
id: principalId,
companyId,
provider: "discord",
provider,
providerAccountId: "guild-1",
externalId: "user-1",
kind: "user",
@@ -296,7 +307,7 @@ describe("reviewed external-chat execution binding", () => {
expect(wake).toMatchObject({
checkedOutByHarness: false,
externalChatExecutionBound: true,
externalChatProvider: "discord",
externalChatProvider: provider,
});
const prompt = renderPaperclipWakePrompt(wake);
expect(prompt).toContain("not a checkout, approval");
@@ -304,6 +315,84 @@ describe("reviewed external-chat execution binding", () => {
expect(prompt).not.toContain("checked out the issue for this run");
});
async function withReviewRecovery(
check: (recovery: Record<string, unknown>) => Promise<void>,
) {
const decision = decideIssueReviewPathRecovery({
issueId,
sourceRunId: randomUUID(),
assigneeAgentId: agentId,
contextSnapshot: {
...context,
paperclipHarnessCheckedOut: true,
paperclipExternalChatExecutionBound: true,
},
reviewAttention: { state: "stalled", paths: [], reason: "Review path consumed" },
existingWake: false,
});
expect(decision.kind).toBe("enqueue");
if (decision.kind !== "enqueue") return;
await db.update(heartbeatRuns)
.set({ contextSnapshot: decision.contextSnapshot })
.where(eq(heartbeatRuns.id, runId));
try {
await check(decision.contextSnapshot);
} finally {
await db.update(heartbeatRuns)
.set({ contextSnapshot: context })
.where(eq(heartbeatRuns.id, runId));
}
}
it("reauthorizes a review recovery from its retained message batch without approving the task", async () => {
const [before] = await db.select().from(issues).where(eq(issues.id, issueId));
await withReviewRecovery(async (recovery) => {
expect(recovery).not.toHaveProperty("paperclipHarnessCheckedOut");
expect(recovery).not.toHaveProperty("paperclipExternalChatExecutionBound");
await expect(attest(recovery)).resolves.toBe(true);
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0]).toEqual(before);
expect((await db.select().from(issueThreadInteractions)
.where(eq(issueThreadInteractions.id, interactionId)))[0].status).toBe("pending");
});
});
it("denies recovery after access revocation or a changed company, task, or message binding", async () => {
await withReviewRecovery(async (recovery) => {
await expect(attest(recovery)).resolves.toBe(true);
await expect(attestReviewedExternalChatRun({
db, ...binding, companyId: randomUUID(), contextSnapshot: recovery,
})).resolves.toBe(false);
await expect(attestReviewedExternalChatRun({
db, ...binding, issueId: otherIssueId, contextSnapshot: recovery,
})).resolves.toBe(false);
await expect(attest({ ...recovery, wakeCommentIds: [randomUUID()] })).resolves.toBe(false);
await db.update(companyMemberships).set({ status: "suspended" })
.where(eq(companyMemberships.principalId, userId));
try {
await expect(attest(recovery)).resolves.toBe(false);
} finally {
await db.update(companyMemberships).set({ status: "active" })
.where(eq(companyMemberships.principalId, userId));
}
await db.update(chatEndpointResources).set({ enabled: false })
.where(eq(chatEndpointResources.id, resourceId));
try {
await expect(attest(recovery)).resolves.toBe(false);
} finally {
await db.update(chatEndpointResources).set({ enabled: true })
.where(eq(chatEndpointResources.id, resourceId));
}
await db.update(chatConversations).set({ issueId: otherIssueId })
.where(eq(chatConversations.id, conversationId));
try {
await expect(attest(recovery)).resolves.toBe(false);
} finally {
await db.update(chatConversations).set({ issueId })
.where(eq(chatConversations.id, conversationId));
}
});
});
it("does not trust a supplied marker, owner mismatch, different wake batch, or wrong provider", async () => {
await db
.update(issues)
@@ -338,7 +427,7 @@ describe("reviewed external-chat execution binding", () => {
} finally {
await db
.update(chatEndpoints)
.set({ provider: "discord" })
.set({ provider })
.where(eq(chatEndpoints.id, endpointId));
}
});
@@ -493,7 +582,7 @@ describe("reviewed external-chat execution binding", () => {
await expect(
resolveCurrentWakeCommentsBinding(db, binding),
).resolves.toMatchObject({
provider: "discord",
provider,
commentIds: [commentId],
});
await db
@@ -16,16 +16,19 @@ describe("static SPA fallback HTML", () => {
}
});
it("includes the operator snippet only in Cloud-served static HTML", () => {
it("ignores retired snippet settings in managed and self-hosted static HTML", () => {
const dir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-cloud-html-"));
tempDirs.push(dir);
fs.writeFileSync(path.join(dir, "index.html"), "<html><body>App</body></html>");
vi.stubEnv("PAPERCLIP_CLOUD_UI_SNIPPET", '<script src="https://example.com/chat.js"></script>');
vi.stubEnv("PAPERCLIP_CLOUD_UI_SNIPPET_B64", Buffer.from('<script src="https://example.com/legacy.js"></script>').toString("base64"));
vi.stubEnv("PAPERCLIP_CLOUD_TENANT_SERVER_TOKEN", undefined);
vi.stubEnv("PAPERCLIP_MANAGED_CONFIG", undefined);
expect(readBrandedStaticIndexHtml(dir)).not.toContain("chat.js");
vi.stubEnv("PAPERCLIP_MANAGED_CONFIG", "{}");
expect(readBrandedStaticIndexHtml(dir)).toContain('chat.js"></script>\n</body>');
expect(readBrandedStaticIndexHtml(dir)).not.toContain("chat.js");
vi.stubEnv("PAPERCLIP_CLOUD_UI_SNIPPET", undefined);
expect(readBrandedStaticIndexHtml(dir)).not.toContain("legacy.js");
});
it("serves the current index.html instead of reusing stale asset hashes", async () => {
@@ -2,6 +2,7 @@ import fs from "node:fs";
import os from "node:os";
import path from "node:path";
import { afterEach, describe, expect, it, vi } from "vitest";
import { applyUiBranding } from "../ui-branding.js";
import { createCachedViteHtmlRenderer, type ViteWatcherHost } from "../vite-html-renderer.js";
function createWatcher() {
@@ -27,11 +28,37 @@ describe("createCachedViteHtmlRenderer", () => {
const tempDirs: string[] = [];
afterEach(() => {
vi.unstubAllEnvs();
for (const dir of tempDirs.splice(0)) {
fs.rmSync(dir, { recursive: true, force: true });
}
});
it("ignores retired snippet settings in branded development HTML", async () => {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-vite-html-"));
tempDirs.push(tempDir);
fs.writeFileSync(path.join(tempDir, "index.html"), "<html><body>App</body></html>");
vi.stubEnv("PAPERCLIP_MANAGED_CONFIG", "{}");
vi.stubEnv("PAPERCLIP_CLOUD_UI_SNIPPET", '<script src="https://example.com/legacy-plain.js"></script>');
vi.stubEnv("PAPERCLIP_CLOUD_UI_SNIPPET_B64", Buffer.from('<script src="https://example.com/legacy-encoded.js"></script>').toString("base64"));
const renderer = createCachedViteHtmlRenderer({
vite: { watcher: createWatcher(), transformIndexHtml: async (_url, html) => html },
uiRoot: tempDir,
brandHtml: applyUiBranding,
});
try {
const html = await renderer.render("/");
expect(html).toContain("App");
expect(html).not.toContain("legacy-plain.js");
expect(html).not.toContain("legacy-encoded.js");
vi.stubEnv("PAPERCLIP_CLOUD_UI_SNIPPET", undefined);
expect(await renderer.render("/issues")).not.toContain("legacy-encoded.js");
} finally {
renderer.dispose();
}
});
it("caches the branded template until index.html changes while transforming every request", async () => {
const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "paperclip-vite-html-"));
tempDirs.push(tempDir);
+1 -2
View File
@@ -121,7 +121,6 @@ import { adapterRoutes } from "./routes/adapters.js";
import { managedAgentProfileRoutes } from "./routes/managed-agent-profiles.js";
import { remoteAgentProfileRoutes } from "./routes/remote-agent-profiles.js";
import { pluginUiStaticRoutes } from "./routes/plugin-ui-static.js";
import { injectCloudUiSnippet } from "./cloud-ui-snippet.js";
import { readBrandedStaticIndexHtml } from "./static-index-html.js";
import { staticUiCacheControl } from "./static-ui-cache.js";
import { applyUiBranding } from "./ui-branding.js";
@@ -1094,7 +1093,7 @@ export async function createApp(
viteHtmlRenderer = createCachedViteHtmlRenderer({
vite,
uiRoot,
brandHtml: (html) => injectCloudUiSnippet(applyUiBranding(html)),
brandHtml: applyUiBranding,
});
const renderViteHtml = viteHtmlRenderer;
-59
View File
@@ -1,59 +0,0 @@
import { isCloudManagedInstance, type CloudInstanceEnv } from "./services/cloud-instance.js";
/** Trusted operator HTML only. This content is public and runs in the app origin. */
export function injectCloudUiSnippet(html: string, env: CloudInstanceEnv = process.env): string {
const snippet = resolveCloudUiSnippet(env);
if (!isCloudManagedInstance(env) || !snippet) return html;
return html.replace(/<\/body>/i, () => `${snippet}\n</body>`);
}
/**
* A present plain variable always wins — blank included, so clearing it to
* blank disables injection even when a base64 value is still deployed. The
* base64 variant exists because delivery pipelines that write env vars
* through provider APIs can sit behind web application firewalls that
* reject values containing raw script markup; base64 carries the same
* snippet through them unchanged.
*/
function resolveCloudUiSnippet(env: CloudInstanceEnv): string | null {
const plain = env.PAPERCLIP_CLOUD_UI_SNIPPET;
if (plain !== undefined) return asInjectableMarkup(plain);
const encoded = env.PAPERCLIP_CLOUD_UI_SNIPPET_B64?.replace(/\s+/g, "");
if (!encoded) return null;
const decoded = decodeBase64(encoded);
return decoded !== null ? asInjectableMarkup(decoded) : null;
}
/**
* A configured value that already looks like markup (it starts with `<`) is
* injected verbatim, preserving the original bytes. A value that does not is
* treated as a bare script body and wrapped in a `<script>` element.
*
* The bare-body form exists for the WAF case above: even base64 no longer
* carries raw script markup past every provider firewall, because some now
* base64-decode the value before matching. A body with no `<script` marker
* clears them, and the tenant restores the element here — the one place the
* value is trusted app-origin HTML rather than a provider API payload.
* Blank in either form stays disabled.
*/
function asInjectableMarkup(value: string): string | null {
const trimmed = value.trim();
if (!trimmed) return null;
return trimmed.startsWith("<") ? value : `<script>${value}</script>`;
}
/**
* A value that is not canonical, padded base64 of valid UTF-8 is ignored
* rather than injected as garbage: the round trip rejects stray padding
* bits, and the fatal decoder rejects byte sequences that are not UTF-8.
*/
function decodeBase64(encoded: string): string | null {
if (encoded.length % 4 !== 0 || !/^[A-Za-z0-9+/]*={0,2}$/.test(encoded)) return null;
const bytes = Buffer.from(encoded, "base64");
if (bytes.toString("base64") !== encoded) return null;
try {
return new TextDecoder("utf-8", { fatal: true }).decode(bytes);
} catch {
return null;
}
}
@@ -1,3 +1,4 @@
import * as nativeExecutor from "./native-runtime/native-session-executor.js";
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
@@ -374,6 +375,32 @@ const support = await getEmbeddedPostgresTestSupport();
return result;
});
it.each(["verified", "unproven", "changed", "dry_run", "retry", "duplicate"])(
"requires exact local cleanup for a new turn after worker loss (%s)", async mode => {
const f = await seed();
await db.update(heartbeatRuns).set({ errorCode: "native_session_cleanup_quarantined" }).where(eq(heartbeatRuns.id, f.sourceRunId));
const retire = vi.fn(() => mode !== "changed");
const verify = vi.spyOn(nativeExecutor, "verifyStoppedNativeSessionForContinuation").mockResolvedValue(
mode === "unproven" ? null : { evidence: { runId: f.sourceRunId, schema: "paperclip.stopped_native_conversation.v1" }, retire });
try {
const result = mode === "retry"
? await db.transaction(tx => admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db,
reason: "retry_failed_run", failedRunId: f.sourceRunId }))
: await admit(f, mode === "dry_run");
expect(Boolean(result)).toBe(["verified", "dry_run", "duplicate"].includes(mode));
expect(retire).toHaveBeenCalledTimes(["verified", "changed", "duplicate"].includes(mode) ? 1 : 0);
const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
expect(Boolean(actions[0].evidence.explicitUserContinuation)).toBe(["verified", "duplicate"].includes(mode));
const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.sourceRunId),
eq(heartbeatRunEvents.eventType, "native.stopped_conversation_verified")));
expect(events).toHaveLength(["verified", "duplicate"].includes(mode) ? 1 : 0);
if (mode === "duplicate") {
expect(await admit(f)).toBeNull();
expect(retire).toHaveBeenCalledTimes(1);
}
} finally { verify.mockRestore(); }
});
it.each(["suspended", "ready", "wrong_run", "wrong_thread", "active_provider", "pending_tool", "pending_output", "missing_state", "new_launch"])(
"recovers a historical run without process metadata only from exact suspended state (%s)", async kind => {
const f = await seed();
@@ -1,3 +1,4 @@
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
import { readQueuedInteractionResponse } from "./queued-interaction-response.js";
import { isCancelledNativeStartup } from "./cancelled-native-startup.js";
import { hasNativeLocalProcessStop, hasHistoricalSuspendedNativeSession } from "./native-local-process-stop.js";
@@ -146,6 +147,7 @@ export async function admitExplicitNativeContinuation(input: {
if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start.");
const sources: Run[] = [];
const stoppedSessions: Array<{ evidence: Record<string, unknown>; retire: () => boolean }> = [];
const cancelledStartupIds = new Set<string>();
for (const action of actions) {
const runId = action.evidence.runId ?? action.evidence.sourceRunId;
@@ -235,6 +237,14 @@ export async function admitExplicitNativeContinuation(input: {
if (run.processGroupId && !processStopped(-run.processGroupId)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically.");
}
}
if (!remote && run.runtimeMode === "native" && run.errorCode === "native_session_cleanup_quarantined") {
// This is new user input, never permission to retry the interrupted turn.
if (retry) return blocked("cleanup_quarantined", "Send a new message after the previous provider has stopped.");
const { verifyStoppedNativeSessionForContinuation } = await import("./native-runtime/native-session-executor.js");
const stopped = await verifyStoppedNativeSessionForContinuation(db, run);
if (!stopped) return blocked("local_cleanup", "Waiting for the previous provider and its tools to stop. Your message is saved.");
stoppedSessions.push(stopped);
}
sources.push(run);
}
const nativeSources = sources.filter(run => run.runtimeMode === "native");
@@ -253,6 +263,15 @@ export async function admitExplicitNativeContinuation(input: {
context: { previousRunId: previous.id, wakeCommentId: commentId },
summary: null, exposeLowTrustRaw: false });
if (input.dryRun) return { previousRunId: previous.id, commentId, ...(retry ? { failedRunId: input.failedRunId! } : {}) };
for (const stopped of stoppedSessions) {
if (!stopped.retire()) return blocked("local_cleanup", "The previous provider cleanup changed. Your message is saved.");
await appendHeartbeatRunEvent(db, {
companyId, runId: String(stopped.evidence.runId), agentId,
eventType: "native.stopped_conversation_verified", stream: "system", level: "info",
message: "The old runner and provider stopped. New user input will use a fresh session; prior action outcomes remain recorded.",
payload: stopped.evidence,
});
}
const authorization = { actorId, commentId, ...(response ? { interactionId: response.source.interactionId } : {}), ...(retry ? { failedRunId: input.failedRunId } : {}),
...(queuedInterrupt ? { queuedCommentInterruptId: input.queuedCommentInterruptId } : {}),
...(queuedRequest ? { queuedCommentRequestId: input.queuedCommentRequestId } : {}), runId: input.successorRunId,
@@ -11,6 +11,18 @@ import { prepareHeartbeatGitHubLaunchers } from "./heartbeat-github-launchers.js
const target = { kind: "remote" as const, transport: "sandbox" as const, providerKey: "daytona", remoteCwd: "/workspace" };
describe("heartbeat GitHub launcher lifetime", () => {
it.each([target, null])("defers native managed authorization/staging to the session owner (%j)", async (target) => {
const prepare = vi.fn();
const mint = vi.fn();
const result = await prepareHeartbeatGitHubLaunchers({
native: true, githubConfigured: true, agentId: "agent-a", runId: "run-a", target,
cwd: "/workspace", env: { GH_TOKEN: "ambient" }, brokerUrl: "https://paperclip.test", createBrokerToken: mint,
}, prepare);
expect(prepare).not.toHaveBeenCalled();
expect(mint).not.toHaveBeenCalled();
expect(result.cleanupLocation).toBeNull();
expect(result.env).toMatchObject({ GH_TOKEN: "", PAPERCLIP_GITHUB_BROKER_TOKEN: "" });
});
it("keeps anonymous native sandbox launchers stable without issuing a run capability", async () => {
const createBrokerToken = vi.fn(() => "run-secret");
const prepareLaunchers = vi.fn(async (input) => input.env);
@@ -31,7 +43,7 @@ describe("heartbeat GitHub launcher lifetime", () => {
if (cleanupFails) throw new Error("cleanup unavailable");
});
await expect(prepareHeartbeatGitHubLaunchers({
native: true, githubConfigured: true, agentId: "agent-a", target,
native: false, githubConfigured: true, agentId: "agent-a", target,
runId: "failed-run", cwd: "/workspace", env: {}, brokerUrl: "https://paperclip.test",
createBrokerToken: () => "current-run-secret",
}, prepareLaunchers, cleanupLaunchers)).rejects.toBe(stagingError);
@@ -50,7 +62,7 @@ describe("heartbeat GitHub launcher lifetime", () => {
});
it.each([
{ native: true, githubConfigured: true, target },
{ native: false, githubConfigured: true, target },
{ native: false, githubConfigured: false, target },
{ native: true, githubConfigured: false, target: null },
])("preserves run-scoped managed capabilities outside anonymous native sandboxes: %j", async (mode) => {
@@ -15,10 +15,16 @@ export async function prepareHeartbeatGitHubLaunchers(
prepareLaunchers = prepareGitHubOperationLaunchers,
cleanupLaunchers = cleanupGitHubOperationLaunchers,
) {
// Native configured access is owned by the provider session supervisor.
// Defer staging until it has acquired that session; never mutate a live
// process's authorization from heartbeat preparation.
if (input.native && input.githubConfigured) {
return { env: githubBrokerEnvironment(input.env, { url: "", token: "" }), cleanupLocation: null };
}
// An unconfigured sandbox has no managed GitHub identity to broker. Its
// token-free wrappers still isolate image credentials, but may live as long
// as the workspace so a warm provider never inherits a deleted run path.
// Configured identities keep their run-scoped capability/retirement rules.
// Other adapter paths retain their run-scoped capability/retirement rules.
const anonymous = input.native && !input.githubConfigured &&
input.target?.kind === "remote" && input.target.transport === "sandbox";
const location = {
+121 -165
View File
@@ -24031,173 +24031,129 @@ export function heartbeatService(
nativeDispatchAtMs,
}),
);
// Native Git/gh uses the same authenticated remote callback
// transport as managed adapters. A bridge failure must not make
// GitHub a prerequisite for otherwise unrelated native work.
let nativeGitHubBridge: Awaited<
ReturnType<typeof startAdapterExecutionTargetPaperclipBridge>
> = null;
if (
executionTarget?.kind === "remote" &&
adapterEnv.PAPERCLIP_GITHUB_BROKER_TOKEN
) {
try {
nativeGitHubBridge =
await startAdapterExecutionTargetPaperclipBridge({
runId: run.id,
target: executionTarget,
runtimeRootDir: path.posix.join(
executionTarget.remoteCwd,
".paperclip-runtime",
"github",
run.id,
),
adapterKey: "native-github",
hostApiToken: adapterEnv.PAPERCLIP_GITHUB_BROKER_TOKEN,
hostApiUrl: adapterEnv.PAPERCLIP_GITHUB_BROKER_URL,
const guardedDispatch =
await dispatchResolvedInteractionContinuationWithAtomicGate(
(markDispatchStarted) =>
executePaperclipNativeSession({
db,
execution: nativeExecution,
conversationMode: isConversation(issueContext),
turnTimeoutMs: Math.max(0, asNumber(runtimeConfig.timeoutSec, 0)) * 1_000,
runnerInstanceId: nativeRunnerInstanceId,
leaseOwner: runOptions.nativeLeaseOwner,
restartRecovery: runOptions.nativeRestartRecovery,
backend:
options.nativeSessionBackendFactory?.(nativeExecution),
useRunnerd: agent.adapterType === "paperclip_runner",
adapterType: agent.adapterType,
sessionGoalControl,
resumeSessionGoalHeartbeat:
context.resumeSessionGoalHeartbeat === true ||
completedGoalControl,
onGoalCheckpoint: async (snapshot) => {
if (!taskKey) return;
const params =
attachPaperclipSessionMetadataToSessionParams(
{
...runtimeSessionParamsForAdapter,
sessionId: snapshot.identity.sessionId,
cwd: executionWorkspace.cwd,
},
configuredModel,
sessionConfigMetadata,
)!;
const displayId =
snapshot.providerSessionId ?? snapshot.sessionId;
await upsertTaskSession({
companyId: agent.companyId,
agentId: agent.id,
adapterType: agent.adapterType,
taskKey,
sessionParamsJson: params,
sessionDisplayId: displayId,
lastRunId: run.id,
lastError: null,
});
goalCheckpointSession.current = { params, displayId };
},
onLog,
});
} catch {
await onLog(
"stderr",
"[paperclip] GitHub runtime transport unavailable; continuing without managed GitHub access.\n",
);
}
}
try {
const guardedDispatch =
await dispatchResolvedInteractionContinuationWithAtomicGate(
(markDispatchStarted) =>
executePaperclipNativeSession({
db,
execution: nativeExecution,
conversationMode: isConversation(issueContext),
turnTimeoutMs: Math.max(0, asNumber(runtimeConfig.timeoutSec, 0)) * 1_000,
runnerInstanceId: nativeRunnerInstanceId,
leaseOwner: runOptions.nativeLeaseOwner,
restartRecovery: runOptions.nativeRestartRecovery,
backend:
options.nativeSessionBackendFactory?.(nativeExecution),
useRunnerd: agent.adapterType === "paperclip_runner",
adapterType: agent.adapterType,
sessionGoalControl,
resumeSessionGoalHeartbeat:
context.resumeSessionGoalHeartbeat === true ||
completedGoalControl,
onGoalCheckpoint: async (snapshot) => {
if (!taskKey) return;
const params =
attachPaperclipSessionMetadataToSessionParams(
{
...runtimeSessionParamsForAdapter,
sessionId: snapshot.identity.sessionId,
cwd: executionWorkspace.cwd,
},
configuredModel,
sessionConfigMetadata,
)!;
const displayId =
snapshot.providerSessionId ?? snapshot.sessionId;
await upsertTaskSession({
companyId: agent.companyId,
agentId: agent.id,
adapterType: agent.adapterType,
taskKey,
sessionParamsJson: params,
sessionDisplayId: displayId,
lastRunId: run.id,
lastError: null,
});
goalCheckpointSession.current = { params, displayId };
},
onLog,
onEvent: onAdapterEvent,
preparationSpans: nativeRunnerPreparationSpans,
// Bootstrap with executable/home discovery while keeping
// configured provider values and the server-selected
// workspace boundary authoritative.
managedAiCredentialIdentity: managedAiRuntime?.identity,
managedAiCredentialHome: managedAiRuntime ? String((managedAiRuntime.config.env as Record<string, unknown>).CODEX_HOME) : undefined,
runnerEnvironment: {
...buildNativeProviderEnvironment(
adapterEnv,
process.env,
executionWorkspace.cwd,
),
...(nativeGitHubBridge
? {
PAPERCLIP_GITHUB_BROKER_URL:
nativeGitHubBridge.env.PAPERCLIP_API_URL,
PAPERCLIP_GITHUB_BRIDGE_TOKEN:
nativeGitHubBridge.env.PAPERCLIP_API_KEY,
}
: {}),
...(nativeMcpServer
? {
PAPERCLIP_NATIVE_MCP_NAME: nativeMcpServer.name,
PAPERCLIP_NATIVE_MCP_URL: nativeMcpServer.url,
PAPERCLIP_NATIVE_MCP_TOKEN: nativeMcpServer.token,
}
: {}),
...(providerTraceCapture
? {
PAPERCLIP_PROVIDER_TRACE_PATH:
providerTraceCapture.path,
PAPERCLIP_PROVIDER_TRACE_MAX_BYTES: String(
PROVIDER_TRACE_MAX_BYTES,
),
}
: {}),
},
runnerExecutionTarget: executionTarget,
runnerIngressAuthorized: isRunnerIngressAuthorized(
nativeRuntimeResolution,
onEvent: onAdapterEvent,
preparationSpans: nativeRunnerPreparationSpans,
// Bootstrap with executable/home discovery while keeping
// configured provider values and the server-selected
// workspace boundary authoritative.
managedGitHub: !useHostGitHub && githubSelection.configured,
managedAiCredentialIdentity: managedAiRuntime?.identity,
managedAiCredentialHome: managedAiRuntime ? String((managedAiRuntime.config.env as Record<string, unknown>).CODEX_HOME) : undefined,
runnerEnvironment: {
...buildNativeProviderEnvironment(
adapterEnv,
process.env,
executionWorkspace.cwd,
),
runnerPublicUrl:
runtimeEnv.PAPERCLIP_RUNNER_PUBLIC_URL?.trim() || null,
runnerCaBundlePath:
runtimeEnv.PAPERCLIP_RUNNER_CA_BUNDLE_PATH?.trim() ||
null,
runnerRemoteBinaryPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_BINARY_PATH?.trim() ||
null,
runnerRemoteCodexPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH?.trim() ||
null,
runnerRemoteCodexNpmSpec:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC?.trim() ||
null,
runnerRemoteProviderPackPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_PROVIDER_PACK_PATH?.trim() ||
null,
stopTaskForReassignment: async (target) => {
await settleLiveRunnerGoalBeforeInterrupt(db, target);
if (!target.runId) return;
const prior = await getRun(target.runId);
if (!prior || prior.companyId !== target.companyId || prior.agentId !== target.agentId) {
throw conflict("Reassignment run binding changed");
}
const stopped = await cancelRunInternal(target.runId, "Cancelled for task reassignment", {
errorCode: "issue_reassigned", suppressImmediateRecovery: true,
resultJson: { reassignmentStopConfirmed: true },
});
if (stopped && ["running", "queued", "scheduled_retry"].includes(stopped.status)) {
throw conflict("The previous run did not stop; reassignment was not applied");
}
},
enqueueWakeup,
onSpawn: async (meta) => {
markDispatchStarted();
await persistRunProcessMetadata(run.id, meta);
},
}),
);
if (!guardedDispatch.dispatched) return;
nativeDispatchStarted = true;
adapterResult = await guardedDispatch.resultPromise;
} finally {
await nativeGitHubBridge?.stop();
}
...(nativeMcpServer
? {
PAPERCLIP_NATIVE_MCP_NAME: nativeMcpServer.name,
PAPERCLIP_NATIVE_MCP_URL: nativeMcpServer.url,
PAPERCLIP_NATIVE_MCP_TOKEN: nativeMcpServer.token,
}
: {}),
...(providerTraceCapture
? {
PAPERCLIP_PROVIDER_TRACE_PATH:
providerTraceCapture.path,
PAPERCLIP_PROVIDER_TRACE_MAX_BYTES: String(
PROVIDER_TRACE_MAX_BYTES,
),
}
: {}),
},
runnerExecutionTarget: executionTarget,
runnerIngressAuthorized: isRunnerIngressAuthorized(
nativeRuntimeResolution,
),
runnerPublicUrl:
runtimeEnv.PAPERCLIP_RUNNER_PUBLIC_URL?.trim() || null,
runnerCaBundlePath:
runtimeEnv.PAPERCLIP_RUNNER_CA_BUNDLE_PATH?.trim() ||
null,
runnerRemoteBinaryPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_BINARY_PATH?.trim() ||
null,
runnerRemoteCodexPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_PATH?.trim() ||
null,
runnerRemoteCodexNpmSpec:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_CODEX_NPM_SPEC?.trim() ||
null,
runnerRemoteProviderPackPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_PROVIDER_PACK_PATH?.trim() ||
null,
stopTaskForReassignment: async (target) => {
await settleLiveRunnerGoalBeforeInterrupt(db, target);
if (!target.runId) return;
const prior = await getRun(target.runId);
if (!prior || prior.companyId !== target.companyId || prior.agentId !== target.agentId) {
throw conflict("Reassignment run binding changed");
}
const stopped = await cancelRunInternal(target.runId, "Cancelled for task reassignment", {
errorCode: "issue_reassigned", suppressImmediateRecovery: true,
resultJson: { reassignmentStopConfirmed: true },
});
if (stopped && ["running", "queued", "scheduled_retry"].includes(stopped.status)) {
throw conflict("The previous run did not stop; reassignment was not applied");
}
},
enqueueWakeup,
onSpawn: async (meta) => {
markDispatchStarted();
await persistRunProcessMetadata(run.id, meta);
},
}),
);
if (!guardedDispatch.dispatched) return;
nativeDispatchStarted = true;
adapterResult = await guardedDispatch.resultPromise;
} else {
const interactionId = readNonEmptyString(context.interactionId);
const legacyQuestionResponse =
@@ -29,6 +29,7 @@ import {
import { getStorageService } from "../../storage/index.js";
import type { StorageService } from "../../storage/types.js";
import { issueService } from "../issues.js";
import { boundExternalChatProvider } from "./external-chat-provider.js";
import { resolveExternalChatQuestionResponse } from "./external-chat-question-response.js";
export const LIST_CHAT_ATTACHMENTS_TOOL_NAME = "list_chat_attachments";
@@ -418,17 +419,7 @@ export async function authorizeChatConversationForBoundRun(
context = answer.authorizationContext;
}
const source = typeof context.source === "string" ? context.source : "";
const provider = [
"slack",
"github",
"discord",
"microsoft-teams",
"telegram",
"imessage-photon",
].find(
(candidate) =>
source === `chat:${candidate}` || source === `chat:${candidate}:recovery`,
);
const provider = boundExternalChatProvider(source);
const commentIds = wakeCommentIds(context);
if (
!provider ||
@@ -549,17 +540,7 @@ function externalChatWaitCandidate(
): { provider: string; commentIds: string[] } | null {
const context = record(contextSnapshot);
const source = typeof context.source === "string" ? context.source : "";
const provider = [
"slack",
"github",
"discord",
"microsoft-teams",
"telegram",
"imessage-photon",
].find(
(candidate) =>
source === `chat:${candidate}` || source === `chat:${candidate}:recovery`,
);
const provider = boundExternalChatProvider(source);
const commentIds = wakeCommentIds(context);
const wake = record(context.paperclipWake);
const wakeIssue = record(wake.issue);
@@ -0,0 +1,16 @@
// Providers supported by the run-bound external-chat authorization path.
// AgentMail uses the email inbox path and is not admitted by this boundary.
const BOUND_EXTERNAL_CHAT_PROVIDERS = [
"slack",
"github",
"discord",
"microsoft-teams",
"telegram",
"imessage-photon",
] as const;
export function boundExternalChatProvider(source: unknown) {
return BOUND_EXTERNAL_CHAT_PROVIDERS.find(
(provider) => source === `chat:${provider}` || source === `chat:${provider}:recovery`,
) ?? null;
}
@@ -0,0 +1,128 @@
import { afterEach, expect, it, vi } from "vitest";
import { spawn, execFile } from "node:child_process";
import { mkdtemp, mkdir, readFile, rm, writeFile, access } from "node:fs/promises";
import { promisify } from "node:util";
import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime";
import { tmpdir } from "node:os";
import path from "node:path";
import { createInterface } from "node:readline";
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
const scope = { companyId: "company-a", agentId: "agent-a", issueId: "task-a" };
const run = (runId: string) => ({ ...scope, runId });
const cleanups: (() => Promise<unknown>)[] = [];
afterEach(async () => { for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); });
async function broker(resolveCredentials = vi.fn(async (binding: ReturnType<typeof run>) => ({
status: "available", env: { GH_TOKEN: `fixture-${binding.runId}` },
})), remote = false, bridgeUnavailable = false, onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>) {
const root = await mkdtemp(path.join(tmpdir(), "native-github-reuse-"));
cleanups.push(() => rm(root, { recursive: true, force: true }));
const bin = path.join(root, "real-bin");
await mkdir(bin);
await writeFile(path.join(bin, "gh"), `#!${process.execPath}\nprocess.stdout.write(process.env.GH_TOKEN || 'anonymous');`, { mode: 0o700 });
const execute: CommandManagedRuntimeRunner["execute"] = async (input) => {
const startedAt = new Date().toISOString();
const child = promisify(execFile)(input.command, input.args ?? [], {
cwd: input.cwd ?? root, env: { ...process.env, ...input.env }, timeout: 15_000, maxBuffer: 4 * 1024 * 1024,
});
child.child.stdin?.on("error", () => undefined);
child.child.stdin?.end(input.stdin ?? "");
const result = await child;
return { ...result, exitCode: 0, signal: null, timedOut: false, pid: null, startedAt };
};
const target = remote ? { kind: "remote" as const, transport: "sandbox" as const, providerKey: "test",
remoteCwd: root, runner: { execute }, streamRunLogs: false } : null;
const result = await createNativeGitHubAccess({ scope, target, cwd: root,
env: { PATH: `${bin}:${path.dirname(process.execPath)}:/usr/bin:/bin`, PAPERCLIP_API_KEY: "old-run-api-key" }, resolveCredentials, onLog },
bridgeUnavailable ? async () => { throw new Error("fixture bridge unavailable"); } : undefined);
cleanups.push(result.stop);
return { ...result, root, resolveCredentials };
}
function request(broker: NativeGitHubAccess, extra: RequestInit = {}, endpoint = "/runtime-tools/github/credentials") {
return fetch(broker.env.PAPERCLIP_GITHUB_BROKER_URL + endpoint, {
method: "POST", body: "{}", headers: { authorization: `Bearer ${broker.env.PAPERCLIP_GITHUB_BRIDGE_TOKEN}`, "content-type": "application/json" }, ...extra,
});
}
it.each([false, true])("keeps one live parent and its original launcher environment across two authorized runs (callback bridge: %s)", async (remote) => {
const b = await broker(undefined, remote);
const parent = spawn(process.execPath, ["-e", `
const {execFile}=require('node:child_process');
require('node:readline').createInterface({input:process.stdin}).on('line', () => {
execFile('gh', [], (error, stdout, stderr) => console.log(JSON.stringify({pid:process.pid, value:stdout, error:error?.message, stderr})));
});
`], { env: { ...process.env, ...b.env, PAPERCLIP_API_KEY: "stale-api-key" }, stdio: ["pipe", "pipe", "pipe"] });
cleanups.push(async () => { const exited = new Promise<void>(resolve => parent.once("exit", () => resolve())); parent.kill(); await exited; });
const lines = createInterface({ input: parent.stdout });
const operation = () => new Promise<{ pid: number; value: string; stderr: string }>(resolve => {
lines.once("line", line => resolve(JSON.parse(line))); parent.stdin.write("run\n");
});
expect((await request(b)).status).toBe(403);
const releaseA = b.activate(run("run-a"));
const a = await operation();
expect(a.value).toBe("fixture-run-a");
releaseA();
expect((await operation()).value).toBe("anonymous");
const releaseB = b.activate(run("run-b"));
releaseA(); // A's delayed cleanup cannot revoke B.
const second = await operation();
expect(second.value).toBe("fixture-run-b");
expect(second.pid).toBe(a.pid);
expect(b.resolveCredentials.mock.calls.map(([binding]) => binding.runId)).toEqual(["run-a", "run-b"]);
expect(await readFile(path.join(b.env.PAPERCLIP_GITHUB_LAUNCHER_DIR, "gh"), "utf8")).not.toContain(b.env.PAPERCLIP_GITHUB_BROKER_TOKEN);
releaseB(); await b.stop();
await expect(access(b.env.PAPERCLIP_GITHUB_LAUNCHER_DIR)).rejects.toThrow();
}, 45_000);
it("rejects other scopes, concurrent bindings, browser requests and forged authority", async () => {
const b = await broker();
for (const field of ["companyId", "agentId", "issueId"] as const) {
expect(() => b.activate({ ...run("a"), [field]: "other" })).toThrow("scope mismatch");
}
const release = b.activate(run("a"));
expect(() => b.activate(run("b"))).toThrow("busy");
const rejectedHeaders: Record<string, string>[] = [{ authorization: "Bearer wrong" }, { authorization: "é".repeat(71) },
{ authorization: `Bearer ${b.env.PAPERCLIP_GITHUB_BRIDGE_TOKEN}`, origin: "https://browser.test" }];
for (const headers of rejectedHeaders) {
expect((await request(b, { headers })).status).toBe(403);
}
expect((await request(b, {}, "/api/companies")).status).toBe(404);
expect((await request(b, { method: "GET", body: undefined })).status).toBe(404);
const response = await request(b, { body: JSON.stringify({ runId: "b", responsibleUserId: "other" }) });
expect(response.status).toBe(200);
expect(response.headers.get("cache-control")).toBe("no-store");
expect(b.resolveCredentials).toHaveBeenCalledExactlyOnceWith(run("a"));
release(); await b.stop();
expect(() => b.activate(run("b"))).toThrow("closed");
});
it("does not release an in-flight credential after its run has been replaced", async () => {
let complete!: (value: { status: string; env: { GH_TOKEN: string } }) => void;
const resolver = vi.fn(() => new Promise<{ status: string; env: { GH_TOKEN: string } }>(resolve => { complete = resolve; }));
const b = await broker(resolver);
const release = b.activate(run("a"));
const pending = request(b);
await vi.waitFor(() => expect(resolver).toHaveBeenCalledOnce());
release(); b.activate(run("b"));
complete({ status: "available", env: { GH_TOKEN: "must-not-escape" } });
const response = await pending;
expect(response.status).toBe(403);
expect(await response.text()).not.toContain("must-not-escape");
});
it.each([403, 409, 500])("preserves denial/steering without leaking errors (%s)", async (status) => {
const b = await broker(vi.fn(async () => { throw Object.assign(new Error("secret-value"), { status }); }));
b.activate(run("a"));
const response = await request(b);
expect(response.status).toBe(status === 500 ? 503 : status);
expect(await response.text()).not.toContain("secret-value");
});
it.each([false, true])("keeps anonymous launchers usable when the remote broker cannot start (logging fails: %s)", async (loggingFails) => {
const onLog = vi.fn(async () => { if (loggingFails) throw new Error("fixture log sink unavailable"); });
const b = await broker(undefined, true, true, onLog);
expect(onLog).toHaveBeenCalledWith("stderr", expect.stringContaining("continuing without managed GitHub access"));
b.activate(run("a"));
expect(b.ready).toBe(false);
expect(b.env.PAPERCLIP_GITHUB_BROKER_TOKEN).toBe("");
const result = await promisify(execFile)(path.join(b.env.PAPERCLIP_GITHUB_LAUNCHER_DIR, "gh"), [], {
env: { ...process.env, ...b.env, GH_TOKEN: "ambient-must-not-leak" },
});
expect(result.stdout).toBe("anonymous");
expect(b.resolveCredentials).not.toHaveBeenCalled();
});
@@ -0,0 +1,146 @@
import { randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";
import path from "node:path";
import {
cleanupGitHubOperationLaunchers,
prepareGitHubOperationLaunchers,
startAdapterExecutionTargetPaperclipBridge,
} from "@paperclipai/adapter-utils/execution-target";
import { githubBrokerEnvironment } from "@paperclipai/adapter-utils/github-launcher";
type Binding = { companyId: string; agentId: string; issueId: string; runId: string };
type LauncherInput = Parameters<typeof prepareGitHubOperationLaunchers>[0];
export type NativeGitHubAccess = Awaited<ReturnType<typeof createNativeGitHubAccess>>;
/** A process-lifetime transport, with authority only while its controller owns a run.
* No run token or GitHub credential is stored in the provider's environment/files.
* The resolver still checks the active run and current identity/grants per operation.
*/
export async function createNativeGitHubAccess(input: {
scope: Omit<Binding, "runId">;
target: LauncherInput["target"];
cwd: string;
env: NodeJS.ProcessEnv;
resolveCredentials: (binding: Binding) => Promise<unknown>;
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
}, startBridge = startAdapterExecutionTargetPaperclipBridge) {
const token = randomBytes(32).toString("hex");
const location = { runId: `native-session-${randomUUID()}`, target: input.target };
let active: Binding | null = null;
let stopped = false;
let ready = true;
let stopping: Promise<void> | undefined;
let bridge: Awaited<ReturnType<typeof startAdapterExecutionTargetPaperclipBridge>> = null;
const server = createServer(async (req, res) => {
res.setHeader("Cache-Control", "no-store");
res.setHeader("Content-Type", "application/json");
req.resume();
const reply = (status: number, body: unknown) => {
res.writeHead(status);
res.end(JSON.stringify(body));
};
const bearer = req.headers.authorization ?? "";
const expected = `Bearer ${token}`;
if (req.headers.origin || req.headers.cookie || req.headers["sec-fetch-site"] ||
Buffer.byteLength(bearer) !== Buffer.byteLength(expected) ||
!timingSafeEqual(Buffer.from(bearer), Buffer.from(expected))) {
reply(403, { error: "GitHub session authentication required" });
return;
}
if (req.method !== "POST" || req.url !== "/runtime-tools/github/credentials") {
reply(404, { error: "Unknown GitHub session operation" });
return;
}
const binding = active;
if (!binding || stopped) {
reply(403, { error: "No active GitHub run" });
return;
}
try {
// Bind at receipt; body/headers cannot select a run or responsible user.
const result = await input.resolveCredentials({ ...binding });
if (active !== binding || stopped) {
reply(403, { error: "GitHub run ended during credential acquisition" });
return;
}
reply(200, result);
} catch (error) {
const status = (error as { status?: number })?.status;
// Never leak secret-store/provider errors. Preserve steering's retry signal.
reply(status === 409 ? 409 : status === 403 ? 403 : 503,
{ error: "GitHub credentials unavailable" });
}
});
server.requestTimeout = 15_000;
server.headersTimeout = 10_000;
const stop = () => stopping ??= (async () => {
stopped = true;
active = null;
server.closeAllConnections();
const results = await Promise.allSettled([
bridge?.stop(),
new Promise<void>(resolve => server.close(() => resolve())),
cleanupGitHubOperationLaunchers(location),
]);
if (results.some(result => result.status === "rejected")) {
await input.onLog?.("stderr", "[paperclip] GitHub session cleanup incomplete.\n").catch(() => undefined);
}
})();
try {
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(0, "127.0.0.1", () => { server.off("error", reject); resolve(); });
});
const address = server.address();
if (!address || typeof address === "string") throw new Error("GitHub broker did not listen");
const url = `http://127.0.0.1:${address.port}`;
try {
bridge = await startBridge({
...location,
runtimeRootDir: input.target?.kind === "remote"
? path.posix.join(input.target.remoteCwd, ".paperclip-runtime", "github", location.runId)
: null,
adapterKey: "native-github",
hostApiToken: token,
hostApiUrl: url,
onLog: input.onLog,
});
if (input.target?.kind === "remote" && !bridge) {
throw new Error("GitHub session requires a remote callback bridge");
}
} catch {
// GitHub remains optional. Stage anonymous wrappers for this session;
// the supervisor retries transport setup on the next run.
ready = false;
await new Promise<void>(resolve => server.close(() => resolve()));
await input.onLog?.("stderr", "[paperclip] GitHub runtime transport unavailable; continuing without managed GitHub access.\n").catch(() => undefined);
}
const env = await prepareGitHubOperationLaunchers({
...location, cwd: input.cwd,
env: {
...githubBrokerEnvironment({ PATH: input.env.PATH }, {
url: ready ? bridge?.env.PAPERCLIP_API_URL ?? url : "",
token: ready ? bridge?.env.PAPERCLIP_API_KEY ?? token : "",
}),
// Never retain an old run's bridge authentication override.
PAPERCLIP_GITHUB_BRIDGE_TOKEN: ready ? bridge?.env.PAPERCLIP_API_KEY ?? token : "",
},
});
return {
env,
ready,
activate(binding: Binding) {
if (stopped || active) throw new Error("GitHub session is closed or busy");
if (binding.companyId !== input.scope.companyId || binding.agentId !== input.scope.agentId ||
binding.issueId !== input.scope.issueId) throw new Error("GitHub session scope mismatch");
const owner = { ...binding };
active = owner;
return () => { if (active === owner) active = null; };
},
stop,
};
} catch (error) {
await stop().catch(() => undefined);
throw error;
}
}
@@ -29,6 +29,7 @@ import {
createPrpSemanticToolResultEnvelope,
validatePrpStructuredRunResult,
validatePrpEvent,
parseNativeExecutionInput,
type NativeExecutionInputV1,
type PrpEvent,
} from "@paperclipai/paperclip-runner";
@@ -52,6 +53,15 @@ import { buildNativeHeartbeatPreparationSpans } from "./native-run-trace.js";
import { NativeRunnerOwnershipUnverifiedError } from "./native-runner-ownership.js";
import type { AdapterRuntimeEvent } from "../../adapters/index.js";
const githubAccess = vi.hoisted(() => ({
activate: vi.fn((_binding: { runId: string }) => vi.fn()),
stop: vi.fn(async () => undefined),
create: vi.fn(),
}));
vi.mock("./native-github-access.js", () => ({
createNativeGitHubAccess: githubAccess.create,
}));
type BackendFactoryOptions = {
runnerInstanceId?: string;
acpxRuntimeDirectory?: string;
@@ -253,6 +263,7 @@ import {
assertRemoteRunnerBuildMetadata,
nativeSessionFailureDisposition,
nativeFailedRunRetryStateIsSafe,
verifyStoppedNativeSessionForContinuation,
nativePreProviderRetryAfterCleanupStateIsSafe,
reconcileRetainedNativeSessionCleanup,
retainedNativeCleanupJournalMatches,
@@ -3853,6 +3864,85 @@ describe("retained native cleanup activation", () => {
});
});
describe("stopped native conversation physical cleanup", () => {
it.each(["codex", "acpx"].flatMap(provider => [
"stopped", "provider_alive", "worker_alive", "missing_receipt", "wrong_receipt", "new_launch",
"foreign_run", "foreign_company", "foreign_runner", "remote", "unreleased", "changed_state", "changed_pid", "symlink", "startup_intent", "pending_identity", "wrong_schema", "replacement", "replacement_alive", "agent_alive", "checkpoint_owner_alive", "diagnostic_owner_alive", "normalized_session_receipt",
].map(mode => ({ provider, mode }))))("$provider $mode", async ({ provider, mode }) => {
const base = await mkdtemp(join(tmpdir(), "native-conversation-cleanup-"));
const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR;
process.env.PAPERCLIP_RUNNER_STATE_DIR = base;
const input = parseNativeExecutionInput({ ...execution,
provider: provider === "codex" ? execution.provider : { kind: "acpx", agent: "claude", model: "claude-sonnet-5", permissionPolicy: "interactive",
profile: { driverKind: "acpx_runtime", protocolVersion: 1, acpxVersion: "0.13.1", agent: "claude", agentProfileVersion: 1,
agentServerPackage: "@zed-industries/claude-agent-acp", agentServerVersion: "1", agentRuntimePackage: null,
agentRuntimeVersion: null, commandDigest: "fixture" } },
session: { ...execution.session, driverKind: provider === "codex" ? "codex_app_server" : "acpx_runtime" },
});
const canonical = (value: any): string => value && typeof value === "object" && !Array.isArray(value)
? `{${Object.entries(value).filter(([, v]) => v !== undefined).sort(([a], [b]) => a.localeCompare(b)).map(([k,v]) => `${JSON.stringify(k)}:${canonical(v)}`).join(",")}}`
: JSON.stringify(value);
const hash = (value: unknown) => createHash("sha256").update(canonical(value)).digest("hex");
const providerScope = input.provider.kind === "acpx" ? { kind: "acpx", agent: input.provider.agent, profile: input.provider.profile } : { kind: "codex" };
const root = join(base, hash({ schema: "paperclip.native-session-scope.v2", companyId: input.binding.companyId,
agentId: input.binding.agentId, workspace: { kind: "managed", executionWorkspaceId: input.binding.executionWorkspaceId },
provider: { driverKind: input.session.driverKind, identity: providerScope }, normalizedSessionId: input.session.normalizedSessionId }));
const identity = { runId: input.binding.runId, normalizedSessionId: input.session.normalizedSessionId,
runnerInstanceId: "runner-crashed", environmentLeaseId: "lease", turnId: "turn", itemId: "item" };
const event = { schema: "paperclip.prp.event.v1", schemaVersion: 1, sourceKind: "runner", sourceSeq: 1,
sourceEventId: "provider-identity", sourceInstanceId: identity.runnerInstanceId, runId: identity.runId,
normalizedSessionId: identity.normalizedSessionId, turnId: identity.turnId, itemId: identity.itemId,
priority: 0, emittedAt: new Date().toISOString(), eventType: "session.started",
payload: { providerDescriptor: { agentProcessId: mode === "agent_alive" ? process.pid : 99_999_996 }, providerSessionId: "provider-session", processId: mode === "provider_alive" ? process.pid : 99_999_998 } };
const replacement = { ...event, sourceSeq: 2, sourceEventId: "replacement", eventType: "session.reconciled",
payload: { ...event.payload, processId: mode === "replacement_alive" ? process.pid : 99_999_997, previousProcessId: event.payload.processId } };
const providerEvents = mode.startsWith("replacement") ? [event, replacement] : [event];
if (mode === "diagnostic_owner_alive") providerEvents.push({ ...event, eventType: "harness.diagnostic",
payload: { providerMethod: "acpx/process", role: "acp_agent", pid: process.pid } } as unknown as typeof event);
const providerState = { schema: mode === "wrong_schema" ? "unknown" : `paperclip.runner.${provider}-provider-state.${provider === "codex" ? "v1" : "v3"}`, lifecycle: "turn_active",
...(mode === "startup_intent" ? { startupAttempt: { phase: "intent" } } : {}),
...(mode === "pending_identity" ? { pendingEvents: [{ eventType: "session.started" }] } : {}),
};
const normalizedEvent = mode === "normalized_session_receipt" ? { ...event, turnId: undefined, itemId: undefined } : event;
const receipt = { sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:1`,
sourcePayloadSha256: mode === "wrong_receipt" ? "wrong" : hash(normalizedEvent), payload: { prpEvent: normalizedEvent } };
const stop = { eventType: mode === "new_launch" ? "native.process_start_requested" : "native.local_process_stopped",
payload: { processPid: mode === "worker_alive" ? process.pid : 99_999_999, processGroupId: mode === "worker_alive" ? process.pid : 99_999_999 } };
let queryIndex = 0;
const db = { select() { const rows = [
[{ provider: mode === "remote" ? "daytona" : "local", releasedAt: mode === "unreleased" ? null : new Date() }],
[stop], mode === "missing_receipt" ? [] : [receipt, ...(mode.startsWith("replacement") ? [{ ...receipt, sourceEventId: `${identity.runnerInstanceId}:${identity.runId}:2`, sourcePayloadSha256: hash(replacement), payload: { prpEvent: replacement } }] : [])],
][queryIndex++] ?? [];
const q: any = { from: () => q, where: () => q, orderBy: () => q, limit: () => q,
then: Promise.resolve(rows).then.bind(Promise.resolve(rows)) }; return q;
} } as unknown as Db;
const run = { id: input.binding.runId, companyId: mode === "foreign_company" ? "foreign" : input.binding.companyId,
agentId: input.binding.agentId, nativeIssueId: input.binding.issueId, runtimeMode: "native", status: "failed", finishedAt: new Date(),
nativeSessionId: input.session.normalizedSessionId, runnerInstanceId: mode === "foreign_runner" ? "foreign" : identity.runnerInstanceId,
runnerProfileJson: { sessionCheckpoint: { process: { codexPid: mode === "checkpoint_owner_alive" ? process.pid : null } }, nativeExecutionInput: input, nativeToolContractFingerprint: nativeToolContractFingerprintForTarget("local") } } as unknown as typeof heartbeatRuns.$inferSelect;
try {
await mkdir(join(root, "runner"), { recursive: true });
await mkdir(join(root, "control-plane"), { recursive: true });
const runnerPath = join(root, "runner/runner-state.json");
await writeFile(join(root, "control-plane/control-plane-state.json"), JSON.stringify({ ...durableControlPlaneState(identity), committedEvents: providerEvents.map(payload => ({ envelope: { payload } })) }));
await writeFile(runnerPath, JSON.stringify(durableRunnerState({ ...identity, ...(mode === "foreign_run" ? { runId: "foreign" } : {}) }, "ready")));
await writeFile(join(root, `runner/${provider}-provider-state.json`), JSON.stringify(providerState));
if (mode === "symlink") { await rename(runnerPath, join(base, "external")); await symlink(join(base, "external"), runnerPath); }
const proof = await verifyStoppedNativeSessionForContinuation(db, run);
if (["stopped", "changed_state", "changed_pid", "replacement", "normalized_session_receipt"].includes(mode)) {
expect(proof).not.toBeNull();
if (mode === "changed_state") await writeFile(runnerPath, "{}");
const kill = mode === "changed_pid" ? vi.spyOn(process, "kill").mockReturnValue(true) : null;
try { expect(proof!.retire()).toBe(["stopped", "replacement", "normalized_session_receipt"].includes(mode)); } finally { kill?.mockRestore(); }
expect(await readFile(join(root, `runner/${provider}-provider-state.json`), "utf8")).toBe(JSON.stringify(providerState));
} else expect(proof).toBeNull();
} finally {
if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous;
await rm(base, { recursive: true, force: true });
}
});
});
describe("explicit failed native retry physical evidence", () => {
it.each([
"suspended",
@@ -6069,6 +6159,10 @@ describe("native warm session supervision", () => {
it.each(
[
...[false, true].flatMap((local) => [true, false].map(brokerReady => ({
firstBroker: true, secondBroker: true, projectless: false, local, brokerReady,
firstMode: "managed", secondMode: "managed", managedGitHub: true,
}))),
...[
{ firstBroker: false, secondBroker: false },
{ firstBroker: false, secondBroker: true },
@@ -6132,10 +6226,12 @@ describe("native warm session supervision", () => {
firstNetwork: "disabled",
secondNetwork: "disabled",
checkpointContract: "valid",
managedGitHub: false,
brokerReady: true,
...scenario,
})),
)(
"verifies a live warm owner before refreshing run authority (broker: $firstBroker -> $secondBroker, projectless: $projectless, local: $local, auth: $firstMode -> $secondMode, network: $firstNetwork -> $secondNetwork, checkpoint: $checkpointContract)",
"verifies a live warm owner before refreshing run authority (broker: $firstBroker -> $secondBroker, projectless: $projectless, local: $local, auth: $firstMode -> $secondMode, network: $firstNetwork -> $secondNetwork, checkpoint: $checkpointContract, managed access: $managedGitHub, broker ready: $brokerReady)",
async ({
firstBroker,
secondBroker,
@@ -6146,10 +6242,18 @@ describe("native warm session supervision", () => {
firstNetwork,
secondNetwork,
checkpointContract,
managedGitHub,
brokerReady,
}) => {
githubAccess.create.mockReset().mockResolvedValue({
env: { PAPERCLIP_GITHUB_BROKER_TOKEN: "stable-session-capability" },
ready: brokerReady,
activate: githubAccess.activate.mockReset().mockImplementation(() => vi.fn()),
stop: githubAccess.stop.mockReset().mockResolvedValue(undefined),
});
const replacesProvider =
firstBroker ||
secondBroker ||
!brokerReady ||
(!managedGitHub && (firstBroker || secondBroker)) ||
firstMode !== secondMode ||
firstNetwork !== secondNetwork;
const stateBase = await mkdtemp(
@@ -6256,7 +6360,7 @@ describe("native warm session supervision", () => {
// the undefined value when a live owner is reused without a load.
expect(options.persistedSession).toBeNull();
}
} else {
} else if (!managedGitHub) {
expect(options.existingSession).toBe(firstSession);
expect(options.persistedSession).toBeUndefined();
}
@@ -6276,6 +6380,7 @@ describe("native warm session supervision", () => {
},
runnerInstanceId: "runner-runnerd-warm",
useRunnerd: true,
managedGitHub,
runnerExecutionTarget: remoteTarget,
});
if (projectless && replacesProvider) {
@@ -6370,8 +6475,21 @@ describe("native warm session supervision", () => {
},
runnerInstanceId: "runner-runnerd-warm",
useRunnerd: true,
managedGitHub,
runnerExecutionTarget: remoteTarget,
});
if (managedGitHub) {
expect(state.execute.mock.calls[1]?.[0].existingSession).toBe(brokerReady ? firstSession : undefined);
expect(githubAccess.create).toHaveBeenCalledTimes(brokerReady ? 1 : 2);
expect(githubAccess.activate.mock.calls.map(([binding]) => binding.runId))
.toEqual([first.binding.runId, second.binding.runId]);
// Replacement fixture publishes no new provider handle: both the old
// owner and the unclaimed replacement broker must be cleaned up.
expect(githubAccess.stop).toHaveBeenCalledTimes(brokerReady ? 0 : 2);
for (const activation of githubAccess.activate.mock.results) {
expect(activation.value).toHaveBeenCalledOnce();
}
}
if (replacesProvider) {
expect(firstClose).toHaveBeenCalledOnce();
expect(firstClose).toHaveBeenCalledWith({
@@ -6388,6 +6506,7 @@ describe("native warm session supervision", () => {
timeout: 1_500,
},
);
if (managedGitHub) expect(githubAccess.stop).toHaveBeenCalledOnce();
}
} finally {
if (previousStateDirectory === undefined) {
@@ -1,3 +1,5 @@
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js";
import { bindManagedNativeCredentialTurn, completeManagedNativeCredentialTurn } from "./managed-native-credentials.js";
import { createLocalNativeQuestionBridge } from "./local-native-question-bridge.js";
import { readVerifiedRemoteWorkspaceFile } from "./remote-deliverable-file.js";
@@ -376,6 +378,7 @@ function clearNativeRuntimeRequestResolutions(runId: string): void {
type WarmNativeSession = {
managedAiCredentialIdentity?: string;
credentialRunId?: string;
githubAccess?: NativeGitHubAccess;
githubAuthenticationMode?: string;
networkAccess: boolean;
session: NativeSession;
@@ -389,6 +392,13 @@ type WarmNativeSession = {
lastActivityAt: string;
};
async function closeWarmNativeSession(entry: WarmNativeSession, reason: string) {
// Revoke before awaiting process retirement/checkpoint IO.
const stopping = entry.githubAccess?.stop();
try { await entry.session.close({ reason }); }
finally { await stopping; }
}
const warmNativeSessions = new Map<string, WarmNativeSession>();
// Closing a remote owner saves its checkpoint asynchronously. A new turn must
// not inspect or quarantine that owner's state until the save has finished.
@@ -444,7 +454,7 @@ async function closeIdleWarmNativeSessions(input: {
// adopt a session whose transport is already shutting down.
warmNativeSessions.delete(sessionId);
const closing = Promise.resolve().then(() =>
entry.session.close({ reason: input.reason }),
closeWarmNativeSession(entry, input.reason),
);
closingWarmNativeSessions.set(sessionId, closing);
try {
@@ -1756,7 +1766,7 @@ const CLEANUP_CANONICAL_FILES = [
] as const;
const CLEANUP_ACTIVATION_FILE = "cleanup-activation.json";
function cleanupStateSnapshot(root: string) {
function cleanupStateSnapshot(root: string, providerFile = "codex-provider-state.json") {
if (
![root, resolve(root, "runner"), resolve(root, "control-plane")].every(
isSafeNativeStateDirectory,
@@ -1764,7 +1774,8 @@ function cleanupStateSnapshot(root: string) {
) {
throw new Error("native_cleanup_maintenance_unproven");
}
const bytes = CLEANUP_CANONICAL_FILES.map((file) =>
const files = [...CLEANUP_CANONICAL_FILES.slice(0, 2), `runner/${providerFile}`];
const bytes = files.map((file) =>
readBoundedNativeFile(
resolve(root, file),
NATIVE_RUNNER_STATE_MAX_BYTES,
@@ -2362,6 +2373,117 @@ export function rebaseRetainedNativeCleanupProviderHome(
}
}
/** Physical cleanup for an explicitly authorized NEW conversation turn. Unlike
* automatic replacement, this does not certify or replay the interrupted actions.
* The caller holds the issue/controller/run locks and preserves their history.
* Retain the old durable root permanently; only the exact in-memory cleanup
* owner is retired, and the successor must use a fresh normalized session.
*/
export async function verifyStoppedNativeSessionForContinuation(
db: Db,
run: typeof heartbeatRuns.$inferSelect,
): Promise<{ evidence: Record<string, unknown>; retire: () => boolean } | null> {
try {
if (run.runtimeMode !== "native" || !["failed", "cancelled", "interrupted", "timed_out"].includes(run.status) ||
!run.finishedAt || !run.nativeIssueId || !run.nativeSessionId || !run.runnerInstanceId) return null;
const execution = parseNativeExecutionInput(record(run.runnerProfileJson).nativeExecutionInput);
if (!(["codex", "acpx"] as string[]).includes(execution.provider.kind) ||
execution.binding.runId !== run.id || execution.binding.companyId !== run.companyId ||
execution.binding.agentId !== run.agentId || execution.binding.issueId !== run.nativeIssueId ||
nativeSessionKey(execution) !== run.nativeSessionId ||
record(run.runnerProfileJson).nativeToolContractFingerprint !== nativeToolContractFingerprintForTarget("local")) return null;
const scope = nativeSessionScopeKey(execution);
const idle = () => !activeNativeSessions.has(run.id) && !executingRunnerdSessionScopes.has(scope) &&
!initializingSessionToolAuthorities.has(scope) && !warmNativeSessions.has(scope);
if (!idle()) return null;
const leases = await db.select().from(environmentLeases).where(and(
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id)));
if (leases.some(lease => lease.provider !== "local" || !lease.releasedAt || lease.cleanupStatus === "failed")) return null;
const stopped = await readNativeLocalProcessStop(db, run.companyId, run.id);
if (!stopped) return null;
const root = scopedRunnerdStateRoot(execution);
const providerFile = runnerProviderStateFilename(execution);
const snapshot = cleanupStateSnapshot(root, providerFile);
const identity = record(snapshot.control.identity);
if (!durableIdentityMatchesExecution(identity, execution) || identity.runnerInstanceId !== run.runnerInstanceId ||
!["runnerInstanceId", "environmentLeaseId", "runId", "normalizedSessionId", "turnId", "itemId"].every(key =>
typeof identity[key] === "string" && identity[key] && snapshot.runner[key] === identity[key]) ||
!Array.isArray(snapshot.control.committedEvents)) return null;
// An incomplete provider launch can own a process that never emitted its
// session identity. It cannot be certified from an earlier owner's receipt.
if (snapshot.control.schema !== "paperclip.runner.durable.control-plane-state.v1" ||
snapshot.runner.schema !== RUNNERD_STATE_SCHEMA ||
snapshot.provider.schema !== (execution.provider.kind === "codex" ? "paperclip.runner.codex-provider-state.v1" : ACPX_PROVIDER_STATE_SCHEMA) ||
!["turn_active", "prepared", "suspended"].includes(String(snapshot.provider.lifecycle)) ||
snapshot.provider.startupAttempt != null) return null;
const pending = JSON.stringify([snapshot.runner.outbox, snapshot.provider.pendingEvents, snapshot.provider.queuedEvents]);
if (/session\.(started|resumed|reconciled)/.test(pending)) return null;
const events = snapshot.control.committedEvents.map(entry => record(record(record(entry).envelope).payload));
const providerEvents = events.filter(event => ["session.started", "session.resumed", "session.reconciled"].includes(String(event.eventType)));
if (!providerEvents.length) return null;
const receipts = await db.select().from(heartbeatRunEvents).where(and(
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
eq(heartbeatRunEvents.sourceInstanceId, run.runnerInstanceId),
inArray(heartbeatRunEvents.eventType, ["session.started", "session.resumed", "session.reconciled", "harness.diagnostic"])));
const providerPids = new Set<number>();
for (const event of providerEvents) {
const provider = record(event.payload);
if (!validatePrpEvent(event).ok || event.sourceKind !== "runner" || event.sourceInstanceId !== run.runnerInstanceId ||
event.runId !== run.id || event.normalizedSessionId !== run.nativeSessionId ||
typeof provider.providerSessionId !== "string" || !provider.providerSessionId ||
!cleanupProcessAbsent(provider.processId) || provider.processId === stopped.processPid) return null;
if (event.eventType === "session.reconciled" &&
(!providerPids.has(Number(provider.previousProcessId)) || !cleanupProcessAbsent(provider.previousProcessId))) return null;
const receipt = receipts.find(row => row.sourceEventId === `${run.runnerInstanceId}:${run.id}:${event.sourceSeq}`);
const durable = record(record(receipt?.payload).prpEvent);
const durableProvider = record(durable.payload);
if (!receipt || !validatePrpEvent(durable).ok || durable.sourceInstanceId !== event.sourceInstanceId ||
durable.runId !== run.id || durable.normalizedSessionId !== run.nativeSessionId ||
// Normalized session-open receipts omit turn/item IDs. Those belong
// to the durable runner identity checked above, not the open event.
durable.sourceSeq !== event.sourceSeq || durable.eventType !== event.eventType ||
receipt.sourcePayloadSha256 !== nativeSha256(durable) ||
(durableProvider.processId !== undefined && durableProvider.processId !== provider.processId) ||
(durableProvider.driverSessionId ?? durableProvider.providerSessionId) !== provider.providerSessionId) return null;
providerPids.add(provider.processId);
}
if (receipts.filter(row => record(record(row.payload).prpEvent).eventType !== "harness.diagnostic").length !== providerEvents.length) return null;
// ACPX's sidecar is not its agent process. Include separately recorded
// owners from both the durable journal and the independent DB/checkpoint.
// Extra evidence can only add a stop requirement, never waive one.
const owners: Record<string, unknown>[] = [record(record(run.runnerProfileJson?.sessionCheckpoint).process),
snapshot.provider, record(snapshot.provider.identity), record(snapshot.provider.descriptor)];
for (const event of [...events, ...receipts.map(row => record(record(row.payload).prpEvent))]) {
const payload = record(event.payload);
if (["session.started", "session.resumed", "session.reconciled"].includes(String(event.eventType))) {
owners.push(payload, record(payload.providerDescriptor), record(payload.providerIdentity), record(payload.runtimeIdentity));
} else if (event.eventType === "harness.diagnostic" && payload.providerMethod === "acpx/process") {
owners.push({ agentPid: payload.pid });
}
}
for (const owner of owners) for (const key of [
"processId", "process_id", "processGroupId", "providerPid", "codexPid", "sidecarPid", "agentPid", "agentProcessId",
]) {
const pid = owner[key];
if (pid === null || pid === undefined) continue;
if (!cleanupProcessAbsent(pid)) return null;
providerPids.add(pid);
}
const unchanged = () => idle() && cleanupProcessAbsent(stopped.processPid) &&
[...providerPids].every(cleanupProcessAbsent) && cleanupStateSnapshot(root, providerFile).fingerprint === snapshot.fingerprint;
return {
evidence: { schema: "paperclip.stopped_native_conversation.v1", runId: run.id,
nativeSessionId: run.nativeSessionId, runnerInstanceId: run.runnerInstanceId,
processPid: stopped.processPid, providerPids: [...providerPids], stateFingerprint: snapshot.fingerprint },
retire: () => {
try { return unchanged() && completeTerminatedLocalNativeSessionCleanup({
companyId: run.companyId, runId: run.id, runnerInstanceId: run.runnerInstanceId!,
}); } catch { return false; }
},
};
} catch { return null; }
}
/** Prove that a crashed local Codex turn ended without external effects. No provider
* is launched and no retained state is rewritten. The successor must use a fresh
* normalized session; the old directory remains available for investigation. */
@@ -5713,9 +5835,8 @@ async function releaseWarmNativeSession(
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
if (failed || entry.closeOnReleaseReason !== undefined) {
warmNativeSessions.delete(sessionId);
const closing = entry.session.close({
reason: entry.closeOnReleaseReason ?? "warm native session failed",
});
const closing = closeWarmNativeSession(entry,
entry.closeOnReleaseReason ?? "warm native session failed");
// Restart checkpointing is required to restore this successful session.
// Surface failure instead of reporting a clean release without authority.
if (entry.closeOnReleaseReason !== undefined) await closing;
@@ -5728,8 +5849,7 @@ async function releaseWarmNativeSession(
// is the idle timer's ownership fence across a later warm acquisition.
if (current !== entry || current.busy) return;
warmNativeSessions.delete(sessionId);
void current.session
.close({ reason: "warm native session idle timeout" })
void closeWarmNativeSession(current, "warm native session idle timeout")
.catch(() => undefined);
}, idleTimeoutMs);
entry.idleTimer.unref();
@@ -5909,7 +6029,7 @@ export function nativeSessionFailureSourceCode(
}
const NATIVE_CLEANUP_OPERATOR_RECOVERY_MESSAGE =
"Verify the prior session's retained process ownership and checkpoint before a controlled server restart and explicit task retry. Clearing a task session does not resolve this quarantine. Automatic retries are stopped.";
"Send a new message to continue after Paperclip verifies that the previous provider and its tools have stopped. If cleanup cannot be verified, inspect the run and its environment. Clearing a task session does not resolve this quarantine. Automatic retries are stopped.";
const PROVIDER_DURABLE_EVENT_TYPES = new Set([
"harness.ready",
@@ -6951,6 +7071,8 @@ export async function executePaperclipNativeSession(input: {
sessionGoalControl?: NativeSessionGoalControl | null;
resumeSessionGoalHeartbeat?: boolean;
preparationSpans?: NativeRunHistoricalSpan[];
/** Use a session-owned GitHub broker, rebound only after run ownership is acquired. */
managedGitHub?: boolean;
/** Resolved adapter env; the runner transport applies a provider allowlist before spawn. */
runnerEnvironment?: NodeJS.ProcessEnv;
/** Private grant materialization; never a user-configured host path. */
@@ -7801,14 +7923,16 @@ async function executePaperclipNativeSessionWithinScope(
);
let existingWarmSession: NativeSession | undefined;
let managedCredentialSession: NativeSession | undefined;
let githubAccess: NativeGitHubAccess | undefined;
let releaseGitHubRun: (() => void) | undefined;
let persistedWarmSession: PersistedNativeSession | null | undefined;
if (warmSessionId !== null && warmConfigDigest !== null) {
const entry = warmNativeSessions.get(warmSessionId);
if (entry) {
// Run-scoped broker capabilities must rotate with the process, while the
// settled provider checkpoint retains the conversation across runs.
// Old run-scoped environments still require process replacement. A
// session-owned broker can change run authority without replacing it.
const hasBrokerCapability = Boolean(
input.runnerEnvironment?.PAPERCLIP_GITHUB_BROKER_TOKEN,
!input.managedGitHub && input.runnerEnvironment?.PAPERCLIP_GITHUB_BROKER_TOKEN,
);
const credentialRunChanged =
Boolean(entry.credentialRunId) !== hasBrokerCapability ||
@@ -7818,6 +7942,8 @@ async function executePaperclipNativeSessionWithinScope(
entry.configDigest !== warmConfigDigest ||
entry.managedAiCredentialIdentity !== input.managedAiCredentialIdentity ||
credentialRunChanged ||
Boolean(entry.githubAccess) !== Boolean(input.managedGitHub) ||
entry.githubAccess?.ready === false ||
entry.githubAuthenticationMode !==
input.runnerEnvironment?.PAPERCLIP_GITHUB_AUTH_MODE ||
entry.networkAccess !==
@@ -7827,9 +7953,7 @@ async function executePaperclipNativeSessionWithinScope(
if (entry.busy) throw new Error("native_session_supervisor_busy");
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
warmNativeSessions.delete(warmSessionId);
await entry.session.close({
reason: "warm native session configuration changed",
});
await closeWarmNativeSession(entry, "warm native session configuration changed");
persistedWarmSession = loadWarmNativeCheckpoint(
input.execution,
warmConfigDigest,
@@ -7844,6 +7968,7 @@ async function executePaperclipNativeSessionWithinScope(
if (entry.idleTimer !== null) clearTimeout(entry.idleTimer);
entry.idleTimer = null;
existingWarmSession = entry.session;
githubAccess = entry.githubAccess;
}
} else {
persistedWarmSession = loadWarmNativeCheckpoint(
@@ -7956,6 +8081,18 @@ async function executePaperclipNativeSessionWithinScope(
controller,
});
try {
if (input.managedGitHub) {
githubAccess ??= await createNativeGitHubAccess({
scope: input.execution.binding,
target: input.runnerExecutionTarget,
cwd: input.execution.workspace.cwd,
env: input.runnerEnvironment ?? process.env,
resolveCredentials: (binding) => resolveGitHubOperationCredentials(input.db, binding),
onLog: input.onLog,
});
releaseGitHubRun = githubAccess.activate(input.execution.binding);
input = { ...input, runnerEnvironment: { ...input.runnerEnvironment, ...githubAccess.env } };
}
const expectedCurrentWakeComments = await resolveCurrentWakeCommentsBinding(
input.db,
input.execution.binding,
@@ -8157,7 +8294,8 @@ async function executePaperclipNativeSessionWithinScope(
networkAccess:
input.runnerEnvironment
?.PAPERCLIP_RUNNER_NETWORK_ACCESS === "enabled",
credentialRunId: input.runnerEnvironment
githubAccess,
credentialRunId: !input.managedGitHub && input.runnerEnvironment
?.PAPERCLIP_GITHUB_BROKER_TOKEN
? input.execution.binding.runId
: undefined,
@@ -8711,6 +8849,13 @@ async function executePaperclipNativeSessionWithinScope(
if (ownershipUnverified) throw new NativeRunnerOwnershipUnverifiedError();
if (protocolIntegrityFailure !== null) throw protocolIntegrityFailure;
}
} finally {
releaseGitHubRun?.();
// A startup failure or onSession(null) must not leak a transport. A warm
// owner retains only the inactive broker until its normal retirement.
if (githubAccess && (!warmSessionId || warmNativeSessions.get(warmSessionId)?.githubAccess !== githubAccess)) {
await githubAccess.stop();
}
}
if (
planSynchronizations.length === 0 &&
@@ -57,6 +57,88 @@ describe("review-path recovery", () => {
expect(duplicate).toEqual({ kind: "skip", reason: "review-path recovery wake already exists" });
});
it.each(["chat:slack", "chat:slack:recovery", "chat:discord"])(
"retains the admitted message batch for %s without inheriting authorization",
(source) => {
const decision = decideIssueReviewPathRecovery({
issueId: "issue-1",
sourceRunId: "run-1",
assigneeAgentId: "agent-1",
contextSnapshot: {
issueId: "issue-1",
source,
wakeCommentIds: ["comment-1", "comment-2", "comment-1"],
wakeCommentId: "stale-comment",
commentId: "stale-comment",
paperclipHarnessCheckedOut: true,
paperclipExternalChatExecutionBound: true,
paperclipWake: { checkedOutByHarness: true },
sessionId: "old-session",
instruction: "old instruction",
},
reviewAttention: stalled,
existingWake: false,
});
expect(decision.kind).toBe("enqueue");
if (decision.kind !== "enqueue") return;
expect(decision.contextSnapshot).toMatchObject({
issueId: "issue-1",
source,
wakeReason: ISSUE_REVIEW_PATH_LOST_WAKE_REASON,
wakeCommentIds: ["comment-1", "comment-2"],
wakeCommentId: "comment-2",
commentId: "comment-2",
});
for (const key of [
"paperclipHarnessCheckedOut",
"paperclipExternalChatExecutionBound",
"paperclipWake",
"sessionId",
]) expect(decision.contextSnapshot).not.toHaveProperty(key);
expect(decision.contextSnapshot.instruction).not.toBe("old instruction");
},
);
it.each([undefined, null, "comment-1", [], [null, 1, " "]])(
"does not invent an admitted message batch from invalid input %j",
(wakeCommentIds) => {
const decision = decideIssueReviewPathRecovery({
issueId: "issue-1",
sourceRunId: "run-1",
assigneeAgentId: "agent-1",
contextSnapshot: {
source: "chat:slack",
wakeCommentIds,
wakeCommentId: "unproven-comment",
},
reviewAttention: stalled,
existingWake: false,
});
expect(decision.kind).toBe("enqueue");
if (decision.kind !== "enqueue") return;
expect(decision.contextSnapshot.source).toBe("chat:slack");
expect(decision.contextSnapshot).not.toHaveProperty("wakeCommentIds");
expect(decision.contextSnapshot).not.toHaveProperty("wakeCommentId");
},
);
it.each(["issue.comment", "chat:agentmail", "chat:agentmail:recovery"])(
"does not carry a chat batch for unsupported source %s",
(source) => {
const decision = decideIssueReviewPathRecovery({
issueId: "issue-1",
sourceRunId: "run-1",
assigneeAgentId: "agent-1",
contextSnapshot: { source, wakeCommentIds: ["comment-1"] },
reviewAttention: stalled,
existingWake: false,
});
expect(decision.kind).toBe("enqueue");
if (decision.kind !== "enqueue") return;
expect(decision.contextSnapshot).not.toHaveProperty("wakeCommentIds");
},
);
it("does not requeue when the bounded recovery run also ends pathless", () => {
const decision = decideIssueReviewPathRecovery({
issueId: "issue-1",
@@ -1,5 +1,7 @@
import { createHash } from "node:crypto";
import type { IssueReviewAttention } from "@paperclipai/shared";
import { boundExternalChatProvider } from "../native-runtime/external-chat-provider.js";
import { extractWakeCommentIds } from "../../modules/run-dispatch/index.js";
import { withRecoveryContext } from "./status-only-context.js";
export const ISSUE_REVIEW_PATH_LOST_WAKE_REASON = "issue_review_path_lost";
@@ -100,6 +102,9 @@ export function decideIssueReviewPathRecovery(input: {
});
if (input.existingWake) return { kind: "skip", reason: "review-path recovery wake already exists" };
const source = readNonEmptyString(context.source) ?? "heartbeat.review_path_disposition";
const chatCommentIds = boundExternalChatProvider(source) ? extractWakeCommentIds(context) : [];
const payload = withRecoveryContext({
issueId: input.issueId,
taskId: input.issueId,
@@ -119,7 +124,15 @@ export function decideIssueReviewPathRecovery(input: {
contextSnapshot: withRecoveryContext({
...payload,
wakeReason: ISSUE_REVIEW_PATH_LOST_WAKE_REASON,
source: readNonEmptyString(context.source) ?? "heartbeat.review_path_disposition",
source,
// Keep the admitted message references, not the source run's authority.
// Dispatch must prove the new run owns this task and recheck current
// endpoint, conversation, and principal access for the entire batch.
...(chatCommentIds.length > 0 ? {
wakeCommentIds: chatCommentIds,
wakeCommentId: chatCommentIds.at(-1),
commentId: chatCommentIds.at(-1),
} : {}),
}, "normal_model"),
};
}
+1 -2
View File
@@ -1,8 +1,7 @@
import fs from "node:fs";
import path from "node:path";
import { injectCloudUiSnippet } from "./cloud-ui-snippet.js";
import { applyUiBranding } from "./ui-branding.js";
export function readBrandedStaticIndexHtml(uiDist: string): string {
return injectCloudUiSnippet(applyUiBranding(fs.readFileSync(path.join(uiDist, "index.html"), "utf-8")));
return applyUiBranding(fs.readFileSync(path.join(uiDist, "index.html"), "utf-8"));
}