mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 18:19:05 +02:00
fix(instructions): persist registered edits after provider shutdown
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
5c84d985c2
commit
5495732552
33 files changed
+2118
-25
No files matched your search
@@ -156,14 +156,50 @@ existing versioned content; their filesystem initialization helper fails closed.
|
||||
|
||||
## Runtime handoff boundary
|
||||
|
||||
This foundation does not register run-local copies, change native or legacy
|
||||
staging, dispatch runner tools, collect stopped runs, or persist pending cleanup
|
||||
candidates. Those integrations must retain the baseline and server identity,
|
||||
collect the registered private file after stop, use this commit path, preserve
|
||||
conflicts/loss evidence, and arrange bounded recovery. They must not write shared
|
||||
instruction caches or call the old filesystem `writeFile` for an entry.
|
||||
The next run should read the canonical snapshot and materialize it when disk is
|
||||
needed. Runtime cleanup parity is not established by these service/editor tests.
|
||||
`agentInstructionWorkingCopyService` provides the registered-copy lifecycle.
|
||||
Migration 0286 stores each run's configured entry, original revision/hash,
|
||||
server-bound responsible user, private local/execution roots, captured candidate,
|
||||
attempt count, diagnostics, and save receipt. Preparation handles managed bundles
|
||||
with an existing entry only; external bundles are never imported. Collection
|
||||
reads only the registered entry, accepts normal atomic editor replacement, rejects
|
||||
symlinks/special files, and bounds valid UTF-8 to 1 MiB. Missing and empty files are
|
||||
distinct. A captured candidate is durable before authorization/CAS; duplicate
|
||||
callbacks and restarts can retry bytes without launching a provider. Failed
|
||||
permission/CAS saves preserve candidates, and public diagnostics omit filesystem
|
||||
paths. Explicit saves advance the baseline only when the registered copy currently
|
||||
matches the committed content; unrelated edits retain the original conflict fence.
|
||||
|
||||
The native runner exposes a stopped-session observer only after its required
|
||||
owned close joins successfully. A terminal-turn change probe can retire the exact
|
||||
warm owner through the existing checkpoint/close path before collecting; unchanged
|
||||
warm sessions stay reusable. The optional working-copy context is separate from
|
||||
the immutable instruction bundle and its prompt digest.
|
||||
|
||||
Heartbeat prepares the copy under the selected workspace's private
|
||||
`.paperclip-runtime/instruction-edits-<run>/instructions` directory, publishes its
|
||||
exact configured entry path to the provider, and collects it before workspace
|
||||
restoration or environment disposal. Git's local exclude file excludes the runtime
|
||||
directory from ordinary staging; workspace snapshots already exclude that directory.
|
||||
The run-to-workspace/environment binding cannot change during a same-run retry.
|
||||
Cloud API providers without access to the selected filesystem use the versioned
|
||||
instruction tools; they are not given an inaccessible editable file.
|
||||
|
||||
Native execution waits for owned session close, or closes only a changed warm
|
||||
owner through the existing checkpoint path. Unchanged warm sessions keep their
|
||||
provider process and sandbox lease. Legacy CLI adapters collect after the final
|
||||
invocation's confirmed process exit; ACP waits for confirmed runtime close. A
|
||||
remote timeout, disconnect, or failed close is not stop evidence. The run log and
|
||||
result include the save receipt or diagnostic; no missing or unverified bytes are
|
||||
reported as saved.
|
||||
|
||||
Recovery retries captured bytes without launching a model. An uncaptured local
|
||||
copy requires a durable stopped-process receipt; a later launch invalidates older
|
||||
stop evidence. Unconfirmed stops remain visible as pending collection. Lost remote
|
||||
environments produce explicit unavailable diagnostics, without restarting the
|
||||
provider to retrieve files. Conflicts, permission failures, and unavailable copies
|
||||
remain inspectable in the instruction editor. Real provider and Daytona proofs
|
||||
are recorded with the Product E2E instruction-persistence workflow; unit/service
|
||||
tests alone do not establish runtime parity.
|
||||
|
||||
## Dedicated native tools
|
||||
|
||||
|
||||
@@ -257,3 +257,8 @@ named existing controls on legacy and native Codex. Discover it with
|
||||
results and follow-up coverage are recorded in that suite's guide.
|
||||
|
||||
Continuation accounting has an explicit-only eight-cell Product E2E [baseline suite](../tests/runner-e2e/CONTINUATION-ACCOUNTING.md), complementing the deterministic lifecycle inventory.
|
||||
|
||||
The explicit Product E2E `instruction-persistence` suite verifies private file
|
||||
edits, stopped-provider cleanup revisions, server restart, and a fresh task's
|
||||
downloaded proof on local native/legacy Codex and native Daytona. See the
|
||||
[Product E2E runbook](../tests/runner-e2e/README.md).
|
||||
@@ -0,0 +1,28 @@
|
||||
CREATE TABLE "agent_instruction_working_copies" (
|
||||
"run_id" uuid PRIMARY KEY NOT NULL,
|
||||
"company_id" uuid NOT NULL,
|
||||
"agent_id" uuid NOT NULL,
|
||||
"responsible_user_id" text NOT NULL,
|
||||
"entry_file" text NOT NULL,
|
||||
"base_revision_id" uuid,
|
||||
"base_hash" text NOT NULL,
|
||||
"local_root" text NOT NULL,
|
||||
"execution_root" text NOT NULL,
|
||||
"location" text NOT NULL,
|
||||
"state" text DEFAULT 'prepared' NOT NULL,
|
||||
"candidate_base64" text,
|
||||
"candidate_hash" text,
|
||||
"error_code" text,
|
||||
"error_message" text,
|
||||
"receipt" jsonb,
|
||||
"process_stopped_at" timestamp with time zone,
|
||||
"attempts" integer DEFAULT 0 NOT NULL,
|
||||
"next_attempt_at" timestamp with time zone,
|
||||
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
|
||||
"updated_at" timestamp with time zone DEFAULT now() NOT NULL
|
||||
);
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "agent_instruction_working_copies" ADD CONSTRAINT "agent_instruction_working_copies_run_id_heartbeat_runs_id_fk" FOREIGN KEY ("run_id") REFERENCES "public"."heartbeat_runs"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
ALTER TABLE "agent_instruction_working_copies" ADD CONSTRAINT "agent_instruction_working_copies_company_id_agent_id_agents_company_id_id_fk" FOREIGN KEY ("company_id","agent_id") REFERENCES "public"."agents"("company_id","id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
|
||||
CREATE INDEX "agent_instruction_copies_pending_idx" ON "agent_instruction_working_copies" USING btree ("state","next_attempt_at");--> statement-breakpoint
|
||||
CREATE INDEX "agent_instruction_copies_agent_idx" ON "agent_instruction_working_copies" USING btree ("company_id","agent_id","created_at");
|
||||
+1008
-2
File diff suppressed because it is too large.
Load diff
+7
@@ -1989,6 +1989,13 @@
|
||||
"when": 1790474441205,
|
||||
"tag": "0285_legal_thunderbolts",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 286,
|
||||
"version": "7",
|
||||
"when": 1790543348340,
|
||||
"tag": "0286_amusing_shiver_man",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -7,8 +7,10 @@ import {
|
||||
index,
|
||||
unique,
|
||||
foreignKey,
|
||||
jsonb,
|
||||
} from "drizzle-orm/pg-core";
|
||||
import { agents } from "./agents.js";
|
||||
import { heartbeatRuns } from "./heartbeat_runs.js";
|
||||
|
||||
/** Append-only content. Base64 preserves every UTF-8 byte, including NUL and BOM. */
|
||||
export const agentInstructionRevisions = pgTable(
|
||||
@@ -82,3 +84,36 @@ export const agentInstructionHeads = pgTable(
|
||||
}).onDelete("cascade"),
|
||||
}),
|
||||
);
|
||||
|
||||
/** A server-owned collection receipt, retained even when canonical CAS rejects an edit. */
|
||||
export const agentInstructionWorkingCopies = pgTable(
|
||||
"agent_instruction_working_copies",
|
||||
{
|
||||
runId: uuid("run_id").primaryKey().references(() => heartbeatRuns.id, { onDelete: "cascade" }),
|
||||
companyId: uuid("company_id").notNull(),
|
||||
agentId: uuid("agent_id").notNull(),
|
||||
responsibleUserId: text("responsible_user_id").notNull(),
|
||||
entryFile: text("entry_file").notNull(),
|
||||
baseRevisionId: uuid("base_revision_id"),
|
||||
baseHash: text("base_hash").notNull(),
|
||||
localRoot: text("local_root").notNull(),
|
||||
executionRoot: text("execution_root").notNull(),
|
||||
location: text("location").notNull(),
|
||||
state: text("state").notNull().default("prepared"),
|
||||
candidateBase64: text("candidate_base64"),
|
||||
candidateHash: text("candidate_hash"),
|
||||
errorCode: text("error_code"),
|
||||
errorMessage: text("error_message"),
|
||||
receipt: jsonb("receipt").$type<Record<string, unknown>>(),
|
||||
processStoppedAt: timestamp("process_stopped_at", { withTimezone: true }),
|
||||
attempts: integer("attempts").notNull().default(0),
|
||||
nextAttemptAt: timestamp("next_attempt_at", { withTimezone: true }),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(t) => ({
|
||||
owner: foreignKey({ columns: [t.companyId, t.agentId], foreignColumns: [agents.companyId, agents.id] }).onDelete("cascade"),
|
||||
pending: index("agent_instruction_copies_pending_idx").on(t.state, t.nextAttemptAt),
|
||||
history: index("agent_instruction_copies_agent_idx").on(t.companyId, t.agentId, t.createdAt),
|
||||
}),
|
||||
);
|
||||
@@ -211,4 +211,4 @@ export { aiProviderDefaults } from "./ai_provider_defaults.js";
|
||||
export * from "./email.js";
|
||||
export { announcementDismissals, announcementPublications } from "./announcement_dismissals.js";
|
||||
|
||||
export { agentInstructionRevisions, agentInstructionHeads } from "./agent_instruction_revisions.js";
|
||||
export { agentInstructionRevisions, agentInstructionHeads, agentInstructionWorkingCopies } from "./agent_instruction_revisions.js";
|
||||
@@ -139,6 +139,7 @@ function createTransportBackedNativeSessionBackend(
|
||||
? (input.provider.approvalPolicy ?? "never")
|
||||
: "never",
|
||||
baseInstructions: nativeSystemInstructions(input),
|
||||
instructionWorkingCopyRoot: "runtimeContext" in input ? input.runtimeContext.instructions.workingCopy?.rootPath : undefined,
|
||||
includeSkillInstructions: isCodex && "runtimeContext" in input,
|
||||
skillInputs: isCodex
|
||||
? nativeTaskSkillInputs(
|
||||
@@ -155,7 +156,7 @@ function createTransportBackedNativeSessionBackend(
|
||||
contractRevision: input.completionContract.contract.revision,
|
||||
criteria: input.completionContract.contract.criteria,
|
||||
constraints: [
|
||||
"Work only inside the supplied working directory.",
|
||||
"Work inside the supplied working directory and any explicitly registered private instruction copy.",
|
||||
...(supportsCollaborativePlanning &&
|
||||
"executionMode" in input &&
|
||||
input.executionMode === "plan"
|
||||
|
||||
@@ -103,6 +103,9 @@ export function nativeTaskConstraints(input: NativeExecutionInput): string[] {
|
||||
}
|
||||
return [
|
||||
"Use only the assigned skills and provider-native tools.",
|
||||
...(input.runtimeContext.instructions.workingCopy ? [
|
||||
`For this turn, the editable agent instruction file is ${input.runtimeContext.instructions.workingCopy.rootPath}/${input.runtimeContext.instructions.workingCopy.entryPath}. This replaces any private working-copy path from a previous turn. Ordinary edits save after the provider stops and only with a durable revision receipt. Use the agent instruction tools for immediate saves. Shared instruction assets and repository instructions are not collected.`,
|
||||
] : []),
|
||||
"Use Paperclip semantic tools for coordination and finalization.",
|
||||
"Save requested plans and Paperclip documents directly with write_document. A saved Paperclip document is already a durable deliverable. Do not create a local file, compute file hashes, or call register_deliverable for it unless the user also requests a downloadable file. Cite the saved document in your completion evidence and final response.",
|
||||
"When the requested result is a file, use register_deliverable before paperclip_finish. Compute its exact byte size and SHA-256, register the workspace-relative file, cite deliverable:<attachmentId> from the receipt as completion evidence, and include /api/attachments/<attachmentId>/content as the download link in your answer. A bare workspace filename is not a delivered result. For repository edits, cite an accessible PR or registered work product. Preserve existing work; do not upload unrelated files. If file publication fails, fix it or report the concrete blocker instead of claiming the file is delivered.",
|
||||
|
||||
@@ -15,7 +15,12 @@ export interface NativeRuntimeAssetReference {
|
||||
|
||||
export interface NativeRuntimeContextSnapshot {
|
||||
prompt: { revision: typeof PAPERCLIP_EXECUTION_PROMPT_REVISION; text: typeof PAPERCLIP_EXECUTION_PROMPT; digest: string };
|
||||
instructions: { entryPath: string; bundle: NativeRuntimeAssetReference };
|
||||
instructions: {
|
||||
entryPath: string;
|
||||
bundle: NativeRuntimeAssetReference;
|
||||
/** Server-registered writable copy; excluded from the pinned prompt digest. */
|
||||
workingCopy?: { rootPath: string; entryPath: string };
|
||||
};
|
||||
skills: Array<{ key: string; runtimeName: string; versionId: string | null; bundle: NativeRuntimeAssetReference }>;
|
||||
mcp: { assignmentSetId: string; digest: string; bindingId: string | null };
|
||||
aggregateDigest: string;
|
||||
@@ -103,7 +108,9 @@ export function parseNativeRuntimeContext(value: unknown): NativeRuntimeContextS
|
||||
throw new NativeRuntimeContextError("input.runtimeContext.prompt.digest does not match prompt text");
|
||||
}
|
||||
const instructions = object(context.instructions, "input.runtimeContext.instructions");
|
||||
exact(instructions, ["entryPath", "bundle"], "input.runtimeContext.instructions");
|
||||
exact(instructions, ["entryPath", "bundle", "workingCopy"], "input.runtimeContext.instructions");
|
||||
const workingCopy = instructions.workingCopy === undefined ? undefined : object(instructions.workingCopy, "input.runtimeContext.instructions.workingCopy");
|
||||
if (workingCopy) exact(workingCopy, ["rootPath", "entryPath"], "input.runtimeContext.instructions.workingCopy");
|
||||
if (!Array.isArray(context.skills)) throw new NativeRuntimeContextError("input.runtimeContext.skills must be an array");
|
||||
const skills = context.skills.map((value, index) => {
|
||||
const skill = object(value, `input.runtimeContext.skills[${index}]`);
|
||||
@@ -122,7 +129,14 @@ export function parseNativeRuntimeContext(value: unknown): NativeRuntimeContextS
|
||||
exact(mcp, ["assignmentSetId", "digest", "bindingId"], "input.runtimeContext.mcp");
|
||||
const parsed = {
|
||||
prompt: { revision: PAPERCLIP_EXECUTION_PROMPT_REVISION, text: PAPERCLIP_EXECUTION_PROMPT, digest: nativeRuntimePromptDigest() },
|
||||
instructions: { entryPath: safeRelativePath(instructions.entryPath, "input.runtimeContext.instructions.entryPath"), bundle: parseAsset(instructions.bundle, "input.runtimeContext.instructions.bundle") },
|
||||
instructions: {
|
||||
entryPath: safeRelativePath(instructions.entryPath, "input.runtimeContext.instructions.entryPath"),
|
||||
bundle: parseAsset(instructions.bundle, "input.runtimeContext.instructions.bundle"),
|
||||
...(workingCopy ? { workingCopy: {
|
||||
rootPath: text(workingCopy.rootPath, "input.runtimeContext.instructions.workingCopy.rootPath"),
|
||||
entryPath: safeRelativePath(workingCopy.entryPath, "input.runtimeContext.instructions.workingCopy.entryPath"),
|
||||
} } : {}),
|
||||
},
|
||||
skills,
|
||||
mcp: {
|
||||
assignmentSetId: text(mcp.assignmentSetId, "input.runtimeContext.mcp.assignmentSetId"),
|
||||
@@ -141,6 +155,10 @@ export function composeNativeSystemInstructions(context: NativeRuntimeContextSna
|
||||
return [
|
||||
context.prompt.text,
|
||||
entryContent.trim(),
|
||||
context.instructions.workingCopy
|
||||
? `Your editable agent instruction file is ${context.instructions.workingCopy.rootPath}/${context.instructions.workingCopy.entryPath}. Edit this registered private copy normally. After this run stops, Paperclip saves changed content as a persistent revision if your responsible user still has permission and the baseline has not changed. Check the run's instruction-save receipt before claiming persistence. Conflicts are preserved for explicit resolution. Repository instruction files, skills, and this run's loaded prompt are separate and are not collected.`
|
||||
: null,
|
||||
// Keep this canonical suffix intact for provider-specific asset remapping.
|
||||
`Read-only instruction sibling root: ${context.instructions.bundle.rootPath}`,
|
||||
].filter(Boolean).join("\n\n");
|
||||
}
|
||||
|
||||
@@ -654,7 +654,7 @@ export class CodexAppServerDriver implements HarnessDriver {
|
||||
this.#options.transportFactory?.(context) ??
|
||||
new ProcessCodexAppServerTransport({
|
||||
workingDirectory,
|
||||
args: createIsolatedCodexAppServerArgs(this.#options.environment, codexExecutableReadOnlyRoots(this.#options.environment ?? process.env)),
|
||||
args: createIsolatedCodexAppServerArgs(this.#options.environment, codexExecutableReadOnlyRoots(this.#options.environment ?? process.env), this.#options.instructionWorkingCopyRoot),
|
||||
environment: createSanitizedCodexEnvironment(this.#options.environment),
|
||||
onDiagnostic: this.#options.onDiagnostic,
|
||||
processGroup: true,
|
||||
|
||||
@@ -20,6 +20,8 @@ export interface CodexAppServerDriverOptions {
|
||||
approvalPolicy?: "never" | "on-request" | "untrusted";
|
||||
baseInstructions?: string;
|
||||
includeSkillInstructions?: boolean;
|
||||
/** Private instruction directory registered by the control plane for this run. */
|
||||
instructionWorkingCopyRoot?: string;
|
||||
/** Explicit selected skills, resolved from this task's assigned runtime assets. */
|
||||
skillInputs?: readonly import("../../contracts/runtime-context.js").NativeSkillInput[];
|
||||
conversationMode?: "task" | "direct";
|
||||
|
||||
@@ -12,6 +12,16 @@ import {
|
||||
} from "./codex-security-config.js";
|
||||
|
||||
describe("Codex security configuration", () => {
|
||||
it("allows only the registered private instruction directory while keeping shared context read-only", () => {
|
||||
const args = createIsolatedCodexAppServerArgs({ HOME: "/host/home" }, ["/runtime/immutable-context"], "/runtime/instruction-edits/run-1").join("\n");
|
||||
expect(args).toContain('"/runtime/instruction-edits/run-1"="write"');
|
||||
expect(args).toContain('"/runtime/immutable-context"="read"');
|
||||
expect(args).not.toContain('"/runtime"="write"');
|
||||
expect(args).toContain('"/host/home"="none"');
|
||||
expect(createIsolatedCodexAppServerArgs({ PAPERCLIP_INSTRUCTION_ROOT: "/host/home" }).join("\n"))
|
||||
.not.toContain('"/host/home"="write"');
|
||||
});
|
||||
|
||||
it("makes the installed npm Codex native sandbox executable readable without exposing its parent workspace", () => {
|
||||
const command = evalProviderTransportOptions("codex").codexCommand!;
|
||||
const manifest = createRequire(command).resolve(`@openai/codex-${process.platform}-${process.arch}/package.json`);
|
||||
|
||||
@@ -169,6 +169,8 @@ function tomlString(value: string): string {
|
||||
export function createIsolatedCodexAppServerArgs(
|
||||
source: NodeJS.ProcessEnv = process.env,
|
||||
readOnlyRoots: string[] = [],
|
||||
/** Server-registered run copy, never an environment/config-supplied root. */
|
||||
instructionWorkingCopyRoot?: string,
|
||||
): string[] {
|
||||
const gitRoots = gitFilesystemRoots(source);
|
||||
readOnlyRoots = [...new Set([...readOnlyRoots, ...codexNetworkReadOnlyRoots(source)])];
|
||||
@@ -199,6 +201,7 @@ export function createIsolatedCodexAppServerArgs(
|
||||
...readOnlyRoots.map((path) => `${tomlString(resolve(path))}="read"`),
|
||||
...gitRoots.read.map((path) => `${tomlString(path)}="read"`),
|
||||
...gitRoots.write.map((path) => `${tomlString(path)}="write"`),
|
||||
...(instructionWorkingCopyRoot ? [`${tomlString(resolve(instructionWorkingCopyRoot))}="write"`] : []),
|
||||
...(source.PAPERCLIP_GITHUB_BROKER_TOKEN && source.GH_CONFIG_DIR
|
||||
? [`${tomlString(resolve(source.GH_CONFIG_DIR))}="write"`] : []),
|
||||
`":workspace_roots"={"."="write"}`,
|
||||
@@ -210,6 +213,7 @@ export function createIsolatedCodexAppServerArgs(
|
||||
...deniedHostRoots.map((path) => `${tomlString(path)}="none"`),
|
||||
...readOnlyRoots.map((path) => `${tomlString(resolve(path))}="read"`),
|
||||
...[...gitRoots.read, ...gitRoots.write].map((path) => `${tomlString(path)}="read"`),
|
||||
...(instructionWorkingCopyRoot ? [`${tomlString(resolve(instructionWorkingCopyRoot))}="read"`] : []),
|
||||
...(source.PAPERCLIP_GITHUB_BROKER_TOKEN && source.GH_CONFIG_DIR
|
||||
? [`${tomlString(resolve(source.GH_CONFIG_DIR))}="write"`] : []),
|
||||
`":workspace_roots"={"."="read"}`,
|
||||
|
||||
@@ -3248,6 +3248,7 @@ export function createRunnerdCodexAppServerArgs(input: {
|
||||
codexHome: string;
|
||||
codexCommand?: string;
|
||||
readOnlyRoots?: string[];
|
||||
instructionWorkingCopyRoot?: string;
|
||||
}): string[] {
|
||||
// The filesystem policy denies HOME and CODEX_HOME to keep credentials and
|
||||
// runner state outside provider reach. Always bind those names to the actual
|
||||
@@ -3260,6 +3261,7 @@ export function createRunnerdCodexAppServerArgs(input: {
|
||||
CODEX_HOME: input.codexHome,
|
||||
},
|
||||
[...(input.readOnlyRoots ?? []), ...codexExecutableReadOnlyRoots(input.environment ?? {}, input.codexCommand)],
|
||||
input.instructionWorkingCopyRoot,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -4589,6 +4591,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
environment: this.options.environment,
|
||||
codexHome,
|
||||
codexCommand: this.options.codexCommand,
|
||||
instructionWorkingCopyRoot: runtimeContext?.instructions.workingCopy?.rootPath,
|
||||
readOnlyRoots: [
|
||||
...trustedRuntimeReadOnlyRoots(
|
||||
this.options.environment,
|
||||
@@ -5077,6 +5080,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
environment: this.options.environment,
|
||||
codexHome,
|
||||
codexCommand: this.options.codexCommand,
|
||||
instructionWorkingCopyRoot: runtimeContext?.instructions.workingCopy?.rootPath,
|
||||
readOnlyRoots: [
|
||||
...trustedRuntimeReadOnlyRoots(this.options.environment),
|
||||
...(runtimeContext
|
||||
|
||||
@@ -3991,6 +3991,45 @@ describe("executeNativeSession recovery", () => {
|
||||
},
|
||||
);
|
||||
|
||||
it("collects a failed run's private instructions only after its owned provider close joins", async () => {
|
||||
const scopedIdentity = { ...identity, companyId: "instruction-close-company", runId: "instruction-close-run" };
|
||||
let finishClose!: () => void;
|
||||
const closed = new Promise<void>((resolve) => { finishClose = resolve; });
|
||||
const order: string[] = [];
|
||||
const capabilities = { resume: true, typedEvents: true, steering: false, interruption: false, structuredResult: true };
|
||||
const session: NativeSession = {
|
||||
identity: () => scopedIdentity,
|
||||
capabilities: async () => capabilities,
|
||||
async *events() { throw new Error("provider failed after editing instructions"); },
|
||||
startTurn: async () => ({ turnId: "instruction-turn" }),
|
||||
result: async () => null,
|
||||
snapshot: async () => ({ backendKind: "local", sessionId: "instruction-session", identity: scopedIdentity,
|
||||
providerSessionId: "instruction-provider", cursor: null, activeTurnId: null, pendingRuntimeRequests: [], lineage: [] }),
|
||||
close: async () => { order.push("closing"); await closed; order.push("stopped"); },
|
||||
};
|
||||
const backend: NativeSessionBackend = {
|
||||
descriptor: async () => ({ kind: "local", name: "instruction-close-test", version: "1", capabilities }),
|
||||
openSession: async () => session,
|
||||
};
|
||||
const controlPlane: ControlPlanePort = {
|
||||
openRun: async () => {}, checkpointSession: async () => {},
|
||||
appendEvent: async () => ({ cursor: 0, highestContiguousSourceSeq: 0, disposition: "committed" }),
|
||||
replayEvents: async () => ({ events: [], highestContiguousSourceSeq: 0 }), completeRun: async () => {},
|
||||
};
|
||||
const options = {
|
||||
input: { ...input, binding: { ...input.binding, companyId: scopedIdentity.companyId, runId: scopedIdentity.runId } },
|
||||
backend, controlPlane, runnerInstanceId: "instruction-runner", controlPlaneInstanceId: "control",
|
||||
requireSessionCloseBeforeReturn: true,
|
||||
onSessionClosed: async () => { order.push("collected"); },
|
||||
};
|
||||
const execution = executeNativeSession(options);
|
||||
const failed = expect(execution).rejects.toThrow("provider failed after editing instructions");
|
||||
await vi.waitFor(() => expect(order).toEqual(["closing"]));
|
||||
finishClose();
|
||||
await failed;
|
||||
expect(order).toEqual(["closing", "stopped", "collected"]);
|
||||
});
|
||||
|
||||
it("retires only the terminated remote resource, including two sandboxes for one run", async () => {
|
||||
const scopedIdentity = { ...identity, companyId: "remote-stop-company", runId: "remote-stop-run" };
|
||||
const binding = { ...scopedIdentity, remoteCleanupScope: "first-sandbox" };
|
||||
@@ -4084,6 +4123,7 @@ describe("executeNativeSession recovery", () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const closeFailure = new Error("required remote checkpoint close failed");
|
||||
const onSessionClosed = vi.fn(async () => {});
|
||||
const close = vi.fn(({ reason }: { reason: string }) =>
|
||||
reason === "native session quarantined cleanup recovery"
|
||||
? Promise.resolve()
|
||||
@@ -4168,11 +4208,13 @@ describe("executeNativeSession recovery", () => {
|
||||
runnerInstanceId: "runner-recovery",
|
||||
controlPlaneInstanceId: "control-recovery",
|
||||
requireSessionCloseBeforeReturn: true,
|
||||
onSessionClosed,
|
||||
}),
|
||||
).rejects.toThrow(closeFailure);
|
||||
await vi.advanceTimersByTimeAsync(3_000);
|
||||
await execution;
|
||||
expect(close).toHaveBeenCalledTimes(5);
|
||||
expect(onSessionClosed).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
|
||||
@@ -220,6 +220,9 @@ export interface ExecuteNativeSessionOptions {
|
||||
* carries required persistence (for example, a remote runner checkpoint).
|
||||
*/
|
||||
requireSessionCloseBeforeReturn?: boolean;
|
||||
/** Trusted cleanup observer, called only after the owned close completes.
|
||||
* Requires requireSessionCloseBeforeReturn; never called for a live warm session. */
|
||||
onSessionClosed?: () => Promise<void>;
|
||||
onCheckpoint?: (
|
||||
snapshot: PersistedNativeSession,
|
||||
options?: CheckpointControlPlaneSessionOptions,
|
||||
@@ -2750,6 +2753,7 @@ export async function executeNativeSession(
|
||||
// generic retryable transport failure at the control-plane boundary.
|
||||
throw protocolIntegrityFailure ?? closeError;
|
||||
}
|
||||
await options.onSessionClosed?.();
|
||||
}
|
||||
} else if (shouldClose && !failedCleanupDeferred) {
|
||||
// A provider that ignores close must not keep execution pending forever.
|
||||
|
||||
@@ -24,7 +24,7 @@ describe("canonical instruction revisions", () => {
|
||||
const entryFile = "policy/INSTRUCTIONS.txt";
|
||||
const initial = "\uFEFF# Original\r\n☃\0\n";
|
||||
beforeAll(async () => {
|
||||
home = await fs.mkdtemp(path.join(os.tmpdir(), "instruction-revisions-"));
|
||||
home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "instruction-revisions-")));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
database = await startEmbeddedPostgresTestDatabase("instruction-revisions-db-");
|
||||
db = createDb(database.connectionString);
|
||||
|
||||
@@ -0,0 +1,222 @@
|
||||
import fs from "node:fs/promises";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import { readGitWorkspaceSnapshot } from "@paperclipai/adapter-utils/git-workspace-sync";
|
||||
import { captureDirectorySnapshot } from "@paperclipai/adapter-utils/workspace-restore-merge";
|
||||
const execFile = promisify(execFileCallback);
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { agents, companies, authUsers, companyMemberships, principalPermissionGrants, heartbeatRuns, agentInstructionWorkingCopies, createDb } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { agentInstructionRevisionService } from "../services/agent-instruction-revisions.js";
|
||||
import { agentInstructionWorkingCopyService } from "../services/agent-instruction-working-copies.js";
|
||||
import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
|
||||
import { resolveManagedInstructionsRoot } from "../services/agent-instructions.js";
|
||||
|
||||
describe("registered run instruction copies", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let copies: ReturnType<typeof agentInstructionWorkingCopyService>;
|
||||
let revisions: ReturnType<typeof agentInstructionRevisionService>;
|
||||
const previousHome = process.env.PAPERCLIP_HOME;
|
||||
let home: string;
|
||||
let companyId: string, agentId: string, userId: string, root: string;
|
||||
const entryFile = "policy/INSTRUCTIONS.txt";
|
||||
const initial = "\uFEFF# Original\r\n☃\n";
|
||||
const target = () => ({ companyId, agentId });
|
||||
const board = () => ({ type: "board" as const, userId, source: "session" as const });
|
||||
async function run() {
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId });
|
||||
return (await copies.prepare({ ...target(), runId, cwd: home }))!;
|
||||
}
|
||||
beforeAll(async () => {
|
||||
home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "instruction-working-copies-")));
|
||||
process.env.PAPERCLIP_HOME = home;
|
||||
database = await startEmbeddedPostgresTestDatabase("instruction-copies-db-");
|
||||
db = createDb(database.connectionString);
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
revisions = agentInstructionRevisionService(db);
|
||||
}, 90_000);
|
||||
afterAll(async () => {
|
||||
if (previousHome === undefined) delete process.env.PAPERCLIP_HOME; else process.env.PAPERCLIP_HOME = previousHome;
|
||||
await database?.cleanup();
|
||||
if (home) await fs.rm(home, { recursive: true, force: true });
|
||||
});
|
||||
beforeEach(async () => {
|
||||
companyId = randomUUID(); agentId = randomUUID(); userId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Instruction tests", issuePrefix: randomUUID().slice(0, 8) });
|
||||
await db.insert(authUsers).values({ id: userId, name: "Editor", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
root = resolveManagedInstructionsRoot({ ...target(), id: agentId, name: "Target", adapterConfig: {} });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Target", adapterConfig: { instructionsBundleMode: "managed", instructionsRootPath: root, instructionsEntryFile: entryFile } });
|
||||
await db.insert(companyMemberships).values([
|
||||
{ companyId, principalType: "user", principalId: userId, membershipRole: "operator" },
|
||||
{ companyId, principalType: "agent", principalId: agentId, membershipRole: "member" },
|
||||
]);
|
||||
await db.insert(principalPermissionGrants).values({ companyId, principalType: "user", principalId: userId, permissionKey: "agents:configure", scope: { agentIds: [agentId] } });
|
||||
await fs.mkdir(path.dirname(path.join(root, entryFile)), { recursive: true });
|
||||
await fs.writeFile(path.join(root, entryFile), initial);
|
||||
});
|
||||
|
||||
it("persists an atomic editor replacement, ignores repository instructions, and gives the next run exact bytes", async () => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(home, "AGENTS.md"), "Repository text must not become agent instructions");
|
||||
const edited = "\uFEFF# Updated\r\nKeep this persistent. ☃\n";
|
||||
const replacement = path.join(copy.localRoot, "edited.tmp");
|
||||
await fs.writeFile(replacement, edited);
|
||||
await fs.rename(replacement, path.join(copy.localRoot, entryFile));
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
const saved = await copies.collectStopped({ companyId, runId: copy.runId });
|
||||
expect(saved).toMatchObject({ state: "saved", candidateHash: expect.any(String), processStoppedAt: expect.any(Date) });
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(edited);
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
expect((await copies.collectStopped({ companyId, runId: copy.runId }))?.receipt).toEqual(saved?.receipt);
|
||||
expect((await revisions.history({ ...target(), entryFile }, board())).revisions).toHaveLength(2);
|
||||
const next = await run();
|
||||
expect(await fs.readFile(path.join(next.localRoot, entryFile), "utf8")).toBe(edited);
|
||||
});
|
||||
|
||||
it.each(["failed", "cancelled", "timed_out"])("retains edits from stopped %s runs without restarting execution", async (status) => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(copy.localRoot, entryFile), status);
|
||||
await db.update(heartbeatRuns).set({ status, finishedAt: new Date() }).where(eq(heartbeatRuns.id, copy.runId));
|
||||
expect((await copies.collectStopped({ companyId, runId: copy.runId }))?.state).toBe("saved");
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe(status);
|
||||
const [unchangedRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, copy.runId));
|
||||
expect(unchangedRun.status).toBe(status);
|
||||
});
|
||||
|
||||
it("preserves a competing stale candidate and requires explicit resolution against the new head", async () => {
|
||||
const first = await run();
|
||||
const second = await run();
|
||||
await fs.writeFile(path.join(first.localRoot, entryFile), "first run");
|
||||
await fs.writeFile(path.join(second.localRoot, entryFile), "second run");
|
||||
await copies.collectStopped({ companyId, runId: first.runId });
|
||||
const conflict = await copies.collectStopped({ companyId, runId: second.runId });
|
||||
expect(conflict).toMatchObject({ state: "conflict", errorCode: "INSTRUCTION_REVISION_CONFLICT" });
|
||||
expect(Buffer.from(conflict!.candidateBase64!, "base64").toString("utf8")).toBe("second run");
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("first run");
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
expect(await copies.list(companyId, agentId, board())).toHaveLength(1);
|
||||
const current = (await revisions.readCurrent(target(), board()))!;
|
||||
await copies.resolve({ ...target(), runId: second.runId, baseRevisionId: current.revision.id, content: "explicitly combined" }, board());
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("explicitly combined");
|
||||
expect((await copies.collectStopped({ companyId, runId: second.runId }))?.state).toBe("resolved");
|
||||
});
|
||||
|
||||
it("rechecks responsible-user permission at collection and preserves denied edits", async () => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(copy.localRoot, entryFile), "candidate after revocation");
|
||||
await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.principalId, userId));
|
||||
const denied = await copies.collectStopped({ companyId, runId: copy.runId });
|
||||
expect(denied).toMatchObject({ state: "conflict", errorCode: "RESPONSIBLE_USER_UNAVAILABLE" });
|
||||
expect(denied?.candidateBase64).not.toBeNull();
|
||||
expect(await fs.readFile(path.join(root, entryFile), "utf8")).toBe(initial);
|
||||
});
|
||||
|
||||
it("does not import a symlink or confuse a removed entry with an empty file", async () => {
|
||||
const linked = await run();
|
||||
await fs.unlink(path.join(linked.localRoot, entryFile));
|
||||
await fs.symlink(path.join(home, "AGENTS.md"), path.join(linked.localRoot, entryFile));
|
||||
expect((await copies.collectStopped({ companyId, runId: linked.runId }))?.state).toBe("pending_collection");
|
||||
const missing = await run();
|
||||
await fs.unlink(path.join(missing.localRoot, entryFile));
|
||||
expect((await copies.collectStopped({ companyId, runId: missing.runId }))?.state).toBe("unavailable");
|
||||
const empty = await run();
|
||||
await fs.writeFile(path.join(empty.localRoot, entryFile), "");
|
||||
expect((await copies.collectStopped({ companyId, runId: empty.runId }))?.state).toBe("saved");
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("");
|
||||
});
|
||||
|
||||
it("keeps unchanged warm copies reusable without a content revision", async () => {
|
||||
const copy = await run();
|
||||
expect(await copies.hasChanges({ companyId, runId: copy.runId })).toBe(false);
|
||||
expect((await copies.get(companyId, copy.runId))?.processStoppedAt).toBeNull();
|
||||
expect((await revisions.history({ ...target(), entryFile }, board())).revisions).toHaveLength(1);
|
||||
await fs.writeFile(path.join(copy.localRoot, entryFile), "changed after turn");
|
||||
expect(await copies.hasChanges({ companyId, runId: copy.runId })).toBe(true);
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe(initial);
|
||||
});
|
||||
|
||||
it("advances an explicit save baseline only when the private copy matches that revision", async () => {
|
||||
const copy = await run();
|
||||
const local = path.join(copy.localRoot, entryFile);
|
||||
await fs.writeFile(local, "explicit self-save");
|
||||
const first = await revisions.commit({ ...target(), entryFile, baseRevisionId: copy.baseRevisionId, content: "explicit self-save", source: "tool" }, board());
|
||||
await copies.acknowledgeExplicitSave({ ...target(), runId: copy.runId, entryFile, revisionId: first.revision.id, contentHash: first.revision.contentHash });
|
||||
await fs.writeFile(local, "local edit after explicit save");
|
||||
expect((await copies.collectStopped({ companyId, runId: copy.runId }))?.state).toBe("saved");
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("local edit after explicit save");
|
||||
|
||||
const competing = await run();
|
||||
await fs.writeFile(path.join(competing.localRoot, entryFile), "older local draft");
|
||||
const separate = await revisions.commit({ ...target(), entryFile, baseRevisionId: competing.baseRevisionId, content: "different explicit content", source: "tool" }, board());
|
||||
await copies.acknowledgeExplicitSave({ ...target(), runId: competing.runId, entryFile, revisionId: separate.revision.id, contentHash: separate.revision.contentHash });
|
||||
expect((await copies.collectStopped({ companyId, runId: competing.runId }))?.state).toBe("conflict");
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("different explicit content");
|
||||
expect((await copies.list(companyId, agentId, board()))[0]).not.toHaveProperty("localRoot");
|
||||
});
|
||||
|
||||
it("recovers captured bytes after restart and deduplicates concurrent cleanup callbacks", async () => {
|
||||
const copy = await run();
|
||||
await db.update(agentInstructionWorkingCopies).set({ state: "pending_commit", candidateBase64: Buffer.from("captured before restart").toString("base64"), processStoppedAt: new Date() })
|
||||
.where(eq(agentInstructionWorkingCopies.runId, copy.runId));
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
await copies.recoverCaptured();
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("captured before restart");
|
||||
const duplicate = await run();
|
||||
await fs.writeFile(path.join(duplicate.localRoot, entryFile), "one final edit");
|
||||
await Promise.all([copies.collectStopped({ companyId, runId: duplicate.runId }), copies.collectStopped({ companyId, runId: duplicate.runId })]);
|
||||
expect((await copies.get(companyId, duplicate.runId))?.state).toBe("saved");
|
||||
expect((await revisions.history({ ...target(), entryFile }, board())).revisions).toHaveLength(3);
|
||||
});
|
||||
it("requires durable stop evidence before recovering an uncaptured copy", async () => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(copy.localRoot, entryFile), "edit before controller restart");
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, copy.runId));
|
||||
copies = agentInstructionWorkingCopyService(db);
|
||||
await copies.recoverStopped();
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe(initial);
|
||||
expect((await copies.list(companyId, agentId, board()))[0]).toMatchObject({ state: "pending_collection", content: null });
|
||||
await appendHeartbeatRunEvent(db, { ...target(), runId: copy.runId, eventType: "native.local_process_stopped", stream: "system" });
|
||||
await copies.recoverStopped();
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe("edit before controller restart");
|
||||
});
|
||||
|
||||
it("never treats a superseded stop receipt or unavailable environment as a save", async () => {
|
||||
const copy = await run();
|
||||
await fs.writeFile(path.join(copy.localRoot, entryFile), "unverified live edit");
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", status: "failed", finishedAt: new Date() }).where(eq(heartbeatRuns.id, copy.runId));
|
||||
await appendHeartbeatRunEvent(db, { ...target(), runId: copy.runId, eventType: "native.local_process_stopped", stream: "system" });
|
||||
await appendHeartbeatRunEvent(db, { ...target(), runId: copy.runId, eventType: "native.process_start_requested", stream: "system" });
|
||||
await copies.recoverStopped();
|
||||
expect((await revisions.readCurrent(target(), board()))?.content).toBe(initial);
|
||||
const lost = await copies.reportUnavailable(companyId, copy.runId);
|
||||
expect(lost).toMatchObject({ state: "unavailable", candidateBase64: null, processStoppedAt: null });
|
||||
expect(lost?.errorMessage).not.toContain(copy.localRoot);
|
||||
});
|
||||
|
||||
it("keeps private run copies out of normal Git staging and workspace snapshots", async () => {
|
||||
const workspace = path.join(home, `workspace-${randomUUID()}`);
|
||||
await fs.mkdir(workspace);
|
||||
await execFile("git", ["init", workspace]);
|
||||
await execFile("git", ["-C", workspace, "-c", "user.name=Fixture", "-c", "user.email=fixture@example.test", "commit", "--allow-empty", "-m", "fixture"]);
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", responsibleUserId: userId });
|
||||
const copy = (await copies.prepare({ ...target(), runId, cwd: workspace }))!;
|
||||
expect(copy.localRoot).toBe(path.join(workspace, ".paperclip-runtime", `instruction-edits-${runId}`, "instructions"));
|
||||
await fs.writeFile(path.join(workspace, "deliverable.txt"), "public work");
|
||||
await execFile("git", ["-C", workspace, "add", "."]);
|
||||
const staged = await execFile("git", ["-C", workspace, "diff", "--cached", "--name-only"]);
|
||||
expect(staged.stdout).toBe("deliverable.txt\n");
|
||||
const snapshot = await readGitWorkspaceSnapshot(workspace);
|
||||
expect(snapshot?.overlayPaths).toEqual(["deliverable.txt"]);
|
||||
const files = await captureDirectorySnapshot(workspace, { exclude: [".git", ".paperclip-runtime"] });
|
||||
expect([...files.entries.keys()]).toEqual(["deliverable.txt"]);
|
||||
await expect(copies.prepare({ ...target(), runId, cwd: home })).rejects.toThrow("different run workspace");
|
||||
});
|
||||
|
||||
});
|
||||
@@ -0,0 +1,314 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { execFile as execFileCallback } from "node:child_process";
|
||||
import { promisify } from "node:util";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { and, desc, eq, inArray, lte, or, isNull } from "drizzle-orm";
|
||||
import { agents, heartbeatRuns, agentInstructionWorkingCopies as copies, type Db } from "@paperclipai/db";
|
||||
import {
|
||||
prepareAdapterExecutionTargetRuntime,
|
||||
runAdapterExecutionTargetShellCommand,
|
||||
type AdapterExecutionTarget,
|
||||
} from "@paperclipai/adapter-utils/execution-target";
|
||||
import { conflict, notFound } from "../errors.js";
|
||||
import { agentInstructionsService, agentInstructionsBundleMode } from "./agent-instructions.js";
|
||||
import { agentInstructionRevisionService } from "./agent-instruction-revisions.js";
|
||||
import { resolveInstructionActor } from "./agent-instruction-authorization.js";
|
||||
import { assertInstructionPathSafe, instructionBytes, instructionPath, materializeInstructionBytes, MAX_INSTRUCTION_BYTES, readInstructionBytes } from "./agent-instruction-files.js";
|
||||
import { hasNativeLocalProcessStop } from "./native-local-process-stop.js";
|
||||
import { remoteExecutionHasStopped } from "./remote-execution-termination.js";
|
||||
import type { AuthorizationActor } from "./authorization.js";
|
||||
|
||||
const execFile = promisify(execFileCallback);
|
||||
type Copy = typeof copies.$inferSelect;
|
||||
const hash = (bytes: Uint8Array) => createHash("sha256").update(bytes).digest("hex");
|
||||
const quote = (value: string) => `'${value.replaceAll("'", `'"'"'`)}'`;
|
||||
const completed = new Set(["saved", "unchanged", "resolved"]);
|
||||
const MAX_COLLECTION_ATTEMPTS = 3;
|
||||
// Only orchestration can register a live target; tools and HTTP callers cannot
|
||||
// supply one. These handles are optional: durable captured bytes survive restart.
|
||||
const liveTargets = new Map<string, AdapterExecutionTarget>();
|
||||
const targetKey = (companyId: string, runId: string) => `${companyId}:${runId}`;
|
||||
|
||||
export function instructionWorkingCopyGuidance(copy: Pick<Copy, "executionRoot" | "entryFile">) {
|
||||
return `Your editable agent instruction file is ${copy.executionRoot}/${copy.entryFile}. Edit this registered private copy normally. After this run stops, Paperclip saves changed content as a persistent revision if your responsible user still has permission and the baseline has not changed. Check the run's instruction-save receipt before claiming persistence. Use read_agent_instructions, update_agent_instructions, get_agent_instruction_history, and restore_agent_instructions for immediate saves and history. Read first and pin the returned revision. Preserve conflicts; never silently retry against a newer head. Repository instructions, skills, and the loaded prompt are separate and are not collected.`;
|
||||
}
|
||||
|
||||
/** Remote reads use the registered entry only; never scan for files called AGENTS.md. */
|
||||
export function instructionCollectionScript(root: string, entry: string) {
|
||||
instructionPath(entry);
|
||||
return `node -e ${quote([
|
||||
"const fs=require('node:fs'),path=require('node:path')",
|
||||
"const root=process.argv[1],entry=process.argv[2],max=Number(process.argv[3])",
|
||||
"const absolute=path.resolve(root,entry),parts=absolute.split('/').filter(Boolean)",
|
||||
"let current='/'",
|
||||
"for(let i=0;i<parts.length;i++){current=path.join(current,parts[i]);let s;try{s=fs.lstatSync(current)}catch(e){if(e.code==='ENOENT'){process.stdout.write(JSON.stringify({missing:true}));process.exit(0)}throw e}if(s.isSymbolicLink()||(i<parts.length-1?!s.isDirectory():!s.isFile()))throw Error('INSTRUCTION_PATH_INVALID')}",
|
||||
"const fd=fs.openSync(absolute,fs.constants.O_RDONLY|fs.constants.O_NOFOLLOW)",
|
||||
"try{const s=fs.fstatSync(fd);if(!s.isFile()||s.size>max)throw Error('INSTRUCTION_CONTENT_INVALID');const bytes=Buffer.alloc(max+1);let n=0;while(n<bytes.length){const count=fs.readSync(fd,bytes,n,bytes.length-n,n);if(!count)break;n+=count}if(n>max)throw Error('INSTRUCTION_CONTENT_INVALID');process.stdout.write(JSON.stringify({contentBase64:bytes.subarray(0,n).toString('base64')}))}finally{fs.closeSync(fd)}",
|
||||
].join(";"))} ${quote(root)} ${quote(entry)} ${MAX_INSTRUCTION_BYTES}`;
|
||||
}
|
||||
|
||||
// Shared local/remote Git setup. The reserved runtime directory is never a
|
||||
// deliverable. Use Git's own exclude file (including worktree metadata) without
|
||||
// modifying tracked repository instructions or .gitignore.
|
||||
const instructionGitExcludeProgram = String.raw`
|
||||
const fs=require('node:fs'),path=require('node:path'),cp=require('node:child_process');
|
||||
const cwd=fs.realpathSync(process.argv[1]);
|
||||
const git=(args)=>cp.spawnSync('git',['-C',cwd,...args],{encoding:'utf8',timeout:10000});
|
||||
const top=git(['rev-parse','--show-toplevel']);
|
||||
if(top.status!==0){if(top.error && top.error.code!=='ENOENT')throw top.error;process.exit(0)}
|
||||
const relative=path.relative(fs.realpathSync(top.stdout.trim()),path.join(cwd,'.paperclip-runtime')).split(path.sep).join('/');
|
||||
if(relative.startsWith('../')||path.isAbsolute(relative))throw Error('Invalid instruction workspace');
|
||||
const excluded=git(['rev-parse','--git-path','info/exclude']);
|
||||
if(excluded.status!==0)throw Error('Cannot resolve Git exclusion');
|
||||
const filename=path.resolve(cwd,excluded.stdout.trim());
|
||||
fs.mkdirSync(path.dirname(filename),{recursive:true});
|
||||
let current=path.parse(filename).root;
|
||||
for(const part of filename.slice(current.length).split(path.sep)){current=path.join(current,part);let s;try{s=fs.lstatSync(current)}catch(e){if(e.code==='ENOENT')continue;throw e}if(s.isSymbolicLink())throw Error('Unsafe Git exclusion path')}
|
||||
const pattern='/'+relative.replace(/[\\*?\[\] #!]/g,'\\$&')+'/';
|
||||
const fd=fs.openSync(filename,fs.constants.O_RDWR|fs.constants.O_CREAT|fs.constants.O_APPEND|fs.constants.O_NOFOLLOW,0o600);
|
||||
try{if(!fs.fstatSync(fd).isFile())throw Error('Unsafe Git exclusion file');const text=fs.readFileSync(fd,'utf8');if(!text.split(/\r?\n/).includes(pattern))fs.writeSync(fd,'\n'+pattern+'\n')}finally{fs.closeSync(fd)}
|
||||
`;
|
||||
|
||||
export function agentInstructionWorkingCopyService(db: Db) {
|
||||
const revisions = agentInstructionRevisionService(db);
|
||||
const scope = (companyId: string, runId: string) => and(eq(copies.companyId, companyId), eq(copies.runId, runId));
|
||||
async function get(companyId: string, runId: string) {
|
||||
const [row] = await db.select().from(copies).where(scope(companyId, runId));
|
||||
return row ?? null;
|
||||
}
|
||||
async function patch(row: Copy, values: Partial<typeof copies.$inferInsert>) {
|
||||
const [updated] = await db.update(copies).set({ ...values, updatedAt: new Date() }).where(and(
|
||||
scope(row.companyId, row.runId), eq(copies.state, row.state), eq(copies.attempts, row.attempts), eq(copies.baseHash, row.baseHash),
|
||||
)).returning();
|
||||
// A late cleanup must not overwrite an explicit resolution or a newer
|
||||
// baseline. Canonical content has its own independent head CAS.
|
||||
return updated ?? (await get(row.companyId, row.runId))!;
|
||||
}
|
||||
function actorFor(row: Copy): AuthorizationActor {
|
||||
return { type: "agent", companyId: row.companyId, agentId: row.agentId, runId: row.runId, onBehalfOfUserId: row.responsibleUserId };
|
||||
}
|
||||
async function prepare(input: { companyId: string; agentId: string; runId: string; target?: AdapterExecutionTarget | null; cwd: string }) {
|
||||
const existing = await get(input.companyId, input.runId);
|
||||
const workspace = await fs.realpath(input.cwd);
|
||||
const expectedLocalRoot = path.join(workspace, ".paperclip-runtime", `instruction-edits-${input.runId}`, "instructions");
|
||||
if (existing && existing.localRoot !== expectedLocalRoot) throw conflict("The registered instruction copy belongs to a different run workspace");
|
||||
const location = input.target?.kind === "remote" ? `remote:${input.target.environmentId ?? ""}` : "local";
|
||||
if (existing && (existing.agentId !== input.agentId || existing.location !== location)) {
|
||||
throw conflict("The registered instruction copy belongs to a different run environment");
|
||||
}
|
||||
// Same-run retries preserve the working bytes and original CAS baseline.
|
||||
if (existing && existing.state !== "preparing") {
|
||||
if (["conflict", "pending_commit", "unavailable"].includes(existing.state)) {
|
||||
throw conflict("Resolve this run's preserved instruction candidate before editing its working copy again");
|
||||
}
|
||||
if (input.target) liveTargets.set(targetKey(input.companyId, input.runId), input.target);
|
||||
if (completed.has(existing.state) || existing.state === "unchanged_turn") {
|
||||
const revision = existing.receipt?.revision as { id?: string; contentHash?: string } | undefined;
|
||||
return patch(existing, { state: "prepared", baseRevisionId: revision?.id ?? existing.baseRevisionId,
|
||||
baseHash: revision?.contentHash ?? existing.baseHash, candidateBase64: null, candidateHash: null,
|
||||
receipt: null, processStoppedAt: null, attempts: 0, nextAttemptAt: null });
|
||||
}
|
||||
return existing;
|
||||
}
|
||||
const [agent] = await db.select().from(agents).where(and(eq(agents.id, input.agentId), eq(agents.companyId, input.companyId)));
|
||||
if (!agent) throw notFound("Agent not found");
|
||||
if (agentInstructionsBundleMode(agent) !== "managed") return null;
|
||||
const bound = await resolveInstructionActor(db, { type: "agent", companyId: input.companyId, agentId: input.agentId, runId: input.runId });
|
||||
const baseline = existing?.baseRevisionId
|
||||
? await revisions.readRevision({ companyId: input.companyId, agentId: input.agentId, entryFile: existing.entryFile, revisionId: existing.baseRevisionId }, bound)
|
||||
: await revisions.readCurrent({ companyId: input.companyId, agentId: input.agentId }, bound);
|
||||
if (!baseline) return null;
|
||||
const exported = await agentInstructionsService().exportFiles(agent, { rejectSymlinks: true });
|
||||
const entryFile = instructionPath(baseline?.revision.entryFile ?? exported.entryFile);
|
||||
const content = instructionBytes(baseline?.content ?? exported.files[entryFile] ?? "");
|
||||
const localRoot = existing?.localRoot ?? expectedLocalRoot;
|
||||
const target = input.target?.kind === "remote" ? input.target : null;
|
||||
const executionRoot = target
|
||||
? path.posix.join(target.remoteCwd, ".paperclip-runtime", `instruction-edits-${input.runId}`, "instructions")
|
||||
: localRoot;
|
||||
if (!existing) {
|
||||
await db.insert(copies).values({
|
||||
runId: input.runId, companyId: input.companyId, agentId: input.agentId,
|
||||
responsibleUserId: bound.onBehalfOfUserId!, entryFile,
|
||||
baseRevisionId: baseline?.revision.id ?? null, baseHash: hash(content),
|
||||
localRoot, executionRoot, location, state: "preparing",
|
||||
});
|
||||
}
|
||||
const row = (await get(input.companyId, input.runId))!;
|
||||
await assertInstructionPathSafe(localRoot, entryFile);
|
||||
await execFile(process.execPath, ["-e", instructionGitExcludeProgram, workspace], { timeout: 15_000 });
|
||||
await fs.mkdir(localRoot, { recursive: true, mode: 0o700 });
|
||||
for (const [name, text] of Object.entries({ ...exported.files, [entryFile]: content.toString("utf8") })) {
|
||||
const relative = instructionPath(name);
|
||||
const filename = await assertInstructionPathSafe(localRoot, relative);
|
||||
await fs.mkdir(path.dirname(filename), { recursive: true, mode: 0o700 });
|
||||
// Atomic replacement also allows interrupted preparation to restage an
|
||||
// already-read-only sibling without writing through its existing inode.
|
||||
await materializeInstructionBytes(localRoot, relative, instructionBytes(text));
|
||||
await fs.chmod(filename, name === entryFile ? 0o600 : 0o400);
|
||||
}
|
||||
if (target) {
|
||||
const staged = await prepareAdapterExecutionTargetRuntime({
|
||||
target, runId: input.runId, adapterKey: `instruction-edits-${input.runId}`,
|
||||
workspaceLocalDir: input.cwd, syncWorkspace: false,
|
||||
assets: [{ key: "instructions", localDir: localRoot, followSymlinks: false }],
|
||||
});
|
||||
if (staged.assetDirs.instructions !== executionRoot) throw new Error("Instruction copy staging path changed");
|
||||
const excluded = await runAdapterExecutionTargetShellCommand(input.runId, target,
|
||||
`node -e ${quote(instructionGitExcludeProgram)} ${quote(target.remoteCwd)}`, { cwd: target.remoteCwd, env: {}, timeoutSec: 15 });
|
||||
if (excluded.exitCode !== 0 || excluded.timedOut) throw new Error("Could not exclude private instructions from Git staging");
|
||||
}
|
||||
const prepared = await patch(row, { state: "prepared" });
|
||||
if (input.target) liveTargets.set(targetKey(input.companyId, input.runId), input.target);
|
||||
return prepared;
|
||||
}
|
||||
|
||||
async function readCandidate(row: Copy, target?: AdapterExecutionTarget | null) {
|
||||
if (row.location === "local") return readInstructionBytes(row.localRoot, row.entryFile);
|
||||
if (target?.kind !== "remote") throw new Error("The original execution environment is needed to retrieve this instruction copy");
|
||||
if (row.location !== `remote:${target.environmentId ?? ""}`) throw new Error("Instruction copy execution environment changed");
|
||||
const expected = path.posix.join(target.remoteCwd, ".paperclip-runtime", `instruction-edits-${row.runId}`, "instructions");
|
||||
if (expected !== row.executionRoot) throw new Error("Instruction copy execution environment changed");
|
||||
const result = await runAdapterExecutionTargetShellCommand(row.runId, target, instructionCollectionScript(row.executionRoot, row.entryFile), { cwd: target.remoteCwd, env: {}, timeoutSec: 30 });
|
||||
if (result.exitCode !== 0 || result.timedOut) throw new Error("Could not safely retrieve the stopped run's instruction file");
|
||||
const payload = JSON.parse(result.stdout) as { missing?: boolean; contentBase64?: string };
|
||||
if (payload.missing) return null;
|
||||
if (typeof payload.contentBase64 !== "string" || payload.contentBase64.length > Math.ceil(MAX_INSTRUCTION_BYTES / 3) * 4) throw new Error("Instruction response exceeds the collection bound");
|
||||
return instructionBytes(Buffer.from(payload.contentBase64, "base64"));
|
||||
}
|
||||
|
||||
/** A terminal-turn probe is not a save: changed/invalid copies require a
|
||||
* stopped provider before capture. Unchanged warm providers remain reusable. */
|
||||
async function hasChanges(input: { companyId: string; runId: string; target?: AdapterExecutionTarget | null }) {
|
||||
const row = await get(input.companyId, input.runId);
|
||||
if (!row || completed.has(row.state)) return false;
|
||||
try {
|
||||
const content = await readCandidate(row, input.target);
|
||||
if (content !== null && hash(content) === row.baseHash) {
|
||||
// This is an unchanged turn receipt, not stopped-process collection.
|
||||
// No candidate bytes or new revision are persisted from a live owner.
|
||||
await patch(row, { state: "unchanged_turn", nextAttemptAt: null });
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
} catch { return true; }
|
||||
}
|
||||
|
||||
async function acknowledgeExplicitSave(input: { companyId: string; agentId: string; runId: string; entryFile: string; revisionId: string; contentHash: string }) {
|
||||
const row = await get(input.companyId, input.runId);
|
||||
if (!row || row.agentId !== input.agentId || row.entryFile !== input.entryFile || row.state !== "prepared") return;
|
||||
const target = liveTargets.get(targetKey(row.companyId, row.runId));
|
||||
const content = await readCandidate(row, target).catch(() => null);
|
||||
// Never advance a stale local edit past an unrelated explicit update.
|
||||
if (content === null || hash(content) !== input.contentHash) return;
|
||||
await db.update(copies).set({ baseRevisionId: input.revisionId, baseHash: input.contentHash, updatedAt: new Date() })
|
||||
.where(and(scope(row.companyId, row.runId), eq(copies.state, "prepared"), eq(copies.baseHash, row.baseHash)));
|
||||
}
|
||||
|
||||
async function commitCandidate(row: Copy) {
|
||||
if (row.candidateBase64 === null || completed.has(row.state) || row.state === "conflict") return row;
|
||||
try {
|
||||
const receipt = await revisions.commit({
|
||||
companyId: row.companyId, agentId: row.agentId, entryFile: row.entryFile,
|
||||
baseRevisionId: row.baseRevisionId, content: Buffer.from(row.candidateBase64, "base64"), source: "cleanup",
|
||||
}, actorFor(row));
|
||||
return patch(row, { state: "saved", receipt: { ...receipt }, errorCode: null, errorMessage: null, nextAttemptAt: null });
|
||||
} catch (error) {
|
||||
const detail = error as { status?: number; details?: { code?: string }; message?: string };
|
||||
const code = detail.details?.code ?? "INSTRUCTION_SAVE_FAILED";
|
||||
const retryable = !detail.status || detail.status >= 500;
|
||||
const message = detail.status === 403
|
||||
? "Current permissions do not allow this instruction save. The candidate was preserved."
|
||||
: detail.status === 409
|
||||
? "Instructions changed after this run started. Review the preserved candidate against the current revision."
|
||||
: "Instruction edits were preserved but not saved. Review the candidate before retrying.";
|
||||
return patch(row, { state: retryable ? "pending_commit" : "conflict", errorCode: code, errorMessage: message, nextAttemptAt: retryable && row.attempts < MAX_COLLECTION_ATTEMPTS ? new Date(Date.now() + 30_000) : null });
|
||||
}
|
||||
}
|
||||
|
||||
/** Caller must have joined the owned provider process before granting this proof. */
|
||||
async function collectStopped(input: { companyId: string; runId: string; target?: AdapterExecutionTarget | null }) {
|
||||
let row = await get(input.companyId, input.runId);
|
||||
if (!row || completed.has(row.state) || row.state === "conflict") return row;
|
||||
row = await patch(row, { processStoppedAt: row.processStoppedAt ?? new Date(), attempts: row.attempts + 1 });
|
||||
if (completed.has(row.state) || row.state === "conflict") return row;
|
||||
if (row.candidateBase64 !== null) return commitCandidate(row);
|
||||
try {
|
||||
const candidate = await readCandidate(row, input.target);
|
||||
if (candidate === null) {
|
||||
return patch(row, { state: "unavailable", errorCode: "INSTRUCTION_FILE_MISSING", errorMessage: "The registered instruction file was removed. Canonical instructions were kept; no edits were saved.", nextAttemptAt: null });
|
||||
}
|
||||
if (hash(candidate) === row.baseHash) return patch(row, { state: "unchanged", nextAttemptAt: null });
|
||||
// Save the candidate durably before attempting authorization/CAS or releasing the environment.
|
||||
row = await patch(row, { state: "pending_commit", candidateBase64: candidate.toString("base64"), candidateHash: hash(candidate), nextAttemptAt: new Date() });
|
||||
return commitCandidate(row);
|
||||
} catch (error) {
|
||||
return patch(row, { state: row.attempts < MAX_COLLECTION_ATTEMPTS ? "pending_collection" : "unavailable", errorCode: "INSTRUCTION_COLLECTION_FAILED", errorMessage: "The registered instruction copy could not be read safely. Canonical instructions were kept; no save receipt exists.", nextAttemptAt: row.attempts < MAX_COLLECTION_ATTEMPTS ? new Date(Date.now() + 30_000) : null });
|
||||
}
|
||||
}
|
||||
|
||||
/** Restarts can retry persisted bytes without touching a live filesystem or launching a model. */
|
||||
async function recoverCaptured() {
|
||||
const pending = await db.select().from(copies).where(and(
|
||||
eq(copies.state, "pending_commit"),
|
||||
or(isNull(copies.nextAttemptAt), lte(copies.nextAttemptAt, new Date())),
|
||||
lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1),
|
||||
)).limit(20);
|
||||
for (const row of pending) await commitCandidate(await patch(row, { attempts: row.attempts + 1 }));
|
||||
return pending.length;
|
||||
}
|
||||
|
||||
/** Recover only the registered copy after independently recorded process-stop
|
||||
* proof. Terminal run status is never a substitute for that proof. Remote
|
||||
* receipts mean the environment is stopped; do not restart it just to read. */
|
||||
async function recoverStopped() {
|
||||
const pending = await db.select({ copy: copies, runtimeMode: heartbeatRuns.runtimeMode }).from(copies)
|
||||
.innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, copies.companyId), eq(heartbeatRuns.id, copies.runId)))
|
||||
.where(and(inArray(copies.state, ["prepared", "pending_collection"]),
|
||||
inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out"]),
|
||||
lte(copies.attempts, MAX_COLLECTION_ATTEMPTS - 1))).limit(20);
|
||||
for (const { copy: row, runtimeMode } of pending) {
|
||||
if (row.location === "local" && (row.processStoppedAt || runtimeMode === "native" &&
|
||||
await hasNativeLocalProcessStop(db, row.companyId, row.runId))) {
|
||||
await collectStopped({ companyId: row.companyId, runId: row.runId });
|
||||
} else if (row.location !== "local" && await remoteExecutionHasStopped(db, row.companyId, row.runId)) {
|
||||
await reportUnavailable(row.companyId, row.runId);
|
||||
} else if (row.state === "prepared") {
|
||||
await patch(row, { state: "pending_collection", errorCode: "INSTRUCTION_STOP_UNCONFIRMED",
|
||||
errorMessage: "The provider's stop has not been confirmed. Instruction collection is pending; no save is claimed.", nextAttemptAt: null });
|
||||
}
|
||||
}
|
||||
return pending.length;
|
||||
}
|
||||
|
||||
async function list(companyId: string, agentId: string, actor: AuthorizationActor) {
|
||||
await revisions.readCurrent({ companyId, agentId }, actor);
|
||||
const rows = await db.select().from(copies).where(and(eq(copies.companyId, companyId), eq(copies.agentId, agentId), inArray(copies.state, ["conflict", "pending_collection", "pending_commit", "unavailable"]))).orderBy(desc(copies.createdAt)).limit(50);
|
||||
return rows.map(row => ({ runId: row.runId, entryFile: row.entryFile, baseRevisionId: row.baseRevisionId,
|
||||
baseHash: row.baseHash, state: row.state as "conflict" | "pending_collection" | "pending_commit" | "unavailable",
|
||||
candidateHash: row.candidateHash, content: row.candidateBase64 === null ? null : Buffer.from(row.candidateBase64, "base64").toString("utf8"),
|
||||
errorCode: row.errorCode, errorMessage: row.errorMessage, createdAt: row.createdAt.toISOString(), updatedAt: row.updatedAt.toISOString() }));
|
||||
}
|
||||
|
||||
async function resolve(input: { companyId: string; agentId: string; runId: string; baseRevisionId: string | null; content: string }, actor: AuthorizationActor) {
|
||||
const row = await get(input.companyId, input.runId);
|
||||
if (!row || row.agentId !== input.agentId) throw notFound("Instruction candidate not found");
|
||||
if (completed.has(row.state)) throw conflict("This instruction candidate has already been resolved");
|
||||
const receipt = await revisions.commit({ companyId: input.companyId, agentId: input.agentId, entryFile: row.entryFile, content: input.content, baseRevisionId: input.baseRevisionId, source: "api" }, actor);
|
||||
await patch(row, { state: "resolved", receipt: { ...receipt }, nextAttemptAt: null });
|
||||
return receipt;
|
||||
}
|
||||
|
||||
/** A controller may report loss before disposal, never fabricate a save or
|
||||
* read a possibly live provider. Already captured candidates remain intact. */
|
||||
async function reportUnavailable(companyId: string, runId: string) {
|
||||
const row = await get(companyId, runId);
|
||||
if (!row || completed.has(row.state) || row.state === "unchanged_turn" || row.candidateBase64 !== null || row.state === "conflict") return row;
|
||||
return patch(row, { state: "unavailable", errorCode: "INSTRUCTION_COLLECTION_UNAVAILABLE",
|
||||
errorMessage: "The registered instruction copy could not be retrieved safely before environment release. No instruction save is claimed.", nextAttemptAt: null });
|
||||
}
|
||||
|
||||
function release(companyId: string, runId: string) { liveTargets.delete(targetKey(companyId, runId)); }
|
||||
return { prepare, get, hasChanges, acknowledgeExplicitSave, collectStopped, recoverCaptured, recoverStopped, list, resolve, reportUnavailable, release };
|
||||
}
|
||||
@@ -58,6 +58,7 @@ import {
|
||||
startAdapterExecutionTargetPaperclipBridge,
|
||||
} from "@paperclipai/adapter-utils/execution-target";
|
||||
import { agentService } from "./agents.js";
|
||||
import { agentInstructionWorkingCopyService, instructionWorkingCopyGuidance } from "./agent-instruction-working-copies.js";
|
||||
import { normalizeLegacyRunnerProvider } from "@paperclipai/adapter-utils";
|
||||
import fs from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
@@ -9401,6 +9402,7 @@ export function heartbeatService(
|
||||
options: HeartbeatServiceOptions = {},
|
||||
) {
|
||||
let shutdownInProgress = false;
|
||||
const instructionCopies = agentInstructionWorkingCopyService(db);
|
||||
const instanceSettings = instanceSettingsService(db);
|
||||
const getCurrentUserRedactionOptions = async () => ({
|
||||
enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs,
|
||||
@@ -18841,6 +18843,14 @@ export function heartbeatService(
|
||||
async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) {
|
||||
const staleThresholdMs = opts?.staleThresholdMs ?? 0;
|
||||
const now = new Date();
|
||||
// Recovery never launches a provider or infers stopped ownership from
|
||||
// terminal status. Uncaptured local copies require durable stop evidence.
|
||||
await instructionCopies.recoverStopped().catch(error => {
|
||||
logger.warn({ err: error }, "failed to recover stopped instruction copies");
|
||||
});
|
||||
await instructionCopies.recoverCaptured().catch(error => {
|
||||
logger.warn({ err: error }, "failed to retry captured instruction revisions");
|
||||
});
|
||||
|
||||
// Complete persisted native results before generic orphan recovery. The
|
||||
// reconciler reads the durable workspace barrier and persisted runtime
|
||||
@@ -22254,6 +22264,27 @@ export function heartbeatService(
|
||||
await bindIssueToPersistedExecutionWorkspace(persistedExecutionWorkspace);
|
||||
const workspaceRealization = realizationResult.workspaceRealization;
|
||||
const executionTarget = realizationResult.executionTarget;
|
||||
let instructionCopy: Awaited<ReturnType<typeof instructionCopies.prepare>> = null;
|
||||
let instructionSave: Record<string, unknown> | null = null;
|
||||
const collectStoppedInstructions = async () => {
|
||||
if (!instructionCopy) return;
|
||||
let saved = await instructionCopies.collectStopped({ companyId: agent.companyId, runId: run.id, target: executionTarget });
|
||||
// Capture before disposal. Exhausted bounded collection leaves a durable
|
||||
// explicit loss report, never a claim that missing bytes were saved.
|
||||
while (saved?.state === "pending_collection" && saved.attempts < 3) {
|
||||
saved = await instructionCopies.collectStopped({ companyId: agent.companyId, runId: run.id, target: executionTarget });
|
||||
}
|
||||
if (!saved) return;
|
||||
const receipt = parseObject(saved.receipt);
|
||||
instructionSave = { state: saved.state, entryFile: saved.entryFile,
|
||||
revisionId: parseObject(receipt.revision).id ?? null, errorCode: saved.errorCode };
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system",
|
||||
level: ["saved", "unchanged", "resolved"].includes(saved.state) ? "info" : "warn",
|
||||
message: saved.state === "saved" ? "Instruction edits saved as a persistent revision."
|
||||
: saved.state === "unchanged" ? "Instruction working copy is unchanged."
|
||||
: "Instruction edits were not saved. Review the preserved candidate in the agent instruction editor.",
|
||||
payload: instructionSave });
|
||||
};
|
||||
if (managedAiRuntime && aiBinding) {
|
||||
try { await assertManagedAiProjectAuth({ ...resolvedConfig, cwd: executionWorkspace.cwd }, aiBinding.provider, executionTarget); }
|
||||
catch { throw new ConfigurationIncompleteFailure("Project authentication conflicts with this agent’s managed AI connection", { configurationIncomplete: { reason: "ai_connection_incompatible", actionUrl: `/agents/${agent.id}/runtime` } }); }
|
||||
@@ -23063,6 +23094,34 @@ export function heartbeatService(
|
||||
target: executionTarget,
|
||||
workspaceId: persistedExecutionWorkspace?.id ?? null,
|
||||
});
|
||||
const hasInstructionFilesystem = nativeRuntimeResolution.kind !== "native" || ![
|
||||
"claude_managed_agents_api", "aws_agentcore_harness_api",
|
||||
].includes(nativeRuntimeResolution.profile.backend);
|
||||
if (hasInstructionFilesystem) {
|
||||
try {
|
||||
instructionCopy = await instructionCopies.prepare({
|
||||
companyId: agent.companyId, agentId: agent.id, runId: run.id,
|
||||
target: executionTarget, cwd: executionWorkspace.cwd,
|
||||
});
|
||||
} catch {
|
||||
// Missing write identity must not break a background run's read-only
|
||||
// prompt. It must also never imply that ordinary file edits will save.
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
message: "Persistent instruction editing is unavailable. Use an authenticated user with instruction edit access and a managed instruction bundle.",
|
||||
payload: { state: "unavailable", code: "INSTRUCTION_COPY_UNAVAILABLE" } });
|
||||
const guidance = "No editable agent instruction working copy is registered for this turn. Use the instruction revision tools for persistent edits; do not edit a private copy named in an earlier turn or claim its changes will persist.";
|
||||
for (const key of ["paperclipTaskMarkdown", "paperclipTaskMarkdownCompact"]) {
|
||||
context[key] = [readNonEmptyString(context[key]), guidance].filter(Boolean).join("\n\n");
|
||||
}
|
||||
}
|
||||
if (instructionCopy) {
|
||||
runtimeConfig = { ...runtimeConfig, instructionsFilePath: path.join(instructionCopy.localRoot, instructionCopy.entryFile) };
|
||||
const guidance = instructionWorkingCopyGuidance(instructionCopy);
|
||||
for (const key of ["paperclipTaskMarkdown", "paperclipTaskMarkdownCompact"]) {
|
||||
context[key] = [readNonEmptyString(context[key]), guidance].filter(Boolean).join("\n\n");
|
||||
}
|
||||
}
|
||||
}
|
||||
let nativeExecution: NativeExecutionInput | null = null;
|
||||
let nativeRunnerInstanceId: string | null = null;
|
||||
if (nativeRuntimeResolution.kind === "native") {
|
||||
@@ -23479,6 +23538,7 @@ export function heartbeatService(
|
||||
runId: run.id,
|
||||
runtimeConfig,
|
||||
runtimeSkillEntries,
|
||||
instructionWorkingCopy: instructionCopy ? { rootPath: instructionCopy.executionRoot, entryPath: instructionCopy.entryFile } : undefined,
|
||||
});
|
||||
const nativeExecutionWithCheckpoint =
|
||||
buildNativeExecutionWithCheckpoint({
|
||||
@@ -24184,6 +24244,10 @@ export function heartbeatService(
|
||||
},
|
||||
onLog,
|
||||
onEvent: onAdapterEvent,
|
||||
instructionWorkingCopy: instructionCopy ? {
|
||||
hasChanges: () => instructionCopies.hasChanges({ companyId: agent.companyId, runId: run.id, target: executionTarget }),
|
||||
collectStopped: collectStoppedInstructions,
|
||||
} : undefined,
|
||||
preparationSpans: nativeRunnerPreparationSpans,
|
||||
// Bootstrap with executable/home discovery while keeping
|
||||
// configured provider values and the server-selected
|
||||
@@ -24377,6 +24441,7 @@ export function heartbeatService(
|
||||
issueId,
|
||||
);
|
||||
},
|
||||
onProviderStopped: collectStoppedInstructions,
|
||||
onDispatch: markDispatchStarted,
|
||||
signal: executionControl.controller.signal,
|
||||
...(executionTarget?.kind === "remote" && executionTarget.transport === "sandbox" ? {
|
||||
@@ -24425,6 +24490,7 @@ export function heartbeatService(
|
||||
if (!guardedDispatch.dispatched) return;
|
||||
adapterResult = await guardedDispatch.resultPromise;
|
||||
}
|
||||
if (instructionSave) adapterResult.resultJson = { ...adapterResult.resultJson, instructionSave };
|
||||
adapterResult = applyWorkspaceRestoreFailure(adapterResult);
|
||||
// A returned result can include a failed restore. Keep the workspace
|
||||
// barrier closed until required files have been restored.
|
||||
@@ -24626,6 +24692,7 @@ export function heartbeatService(
|
||||
"failed to revoke heartbeat-run MCP gateway tokens",
|
||||
);
|
||||
}
|
||||
instructionCopies.release(agent.companyId, run.id);
|
||||
}
|
||||
// Reconcile the referenced-project set against the real remote staging outcome. A referenced
|
||||
// project can pass authorization and clone locally at run prep, then fail to stage into the
|
||||
@@ -26007,6 +26074,15 @@ export function heartbeatService(
|
||||
},
|
||||
);
|
||||
}
|
||||
// A retained or unverified process stays above this release boundary.
|
||||
// If no stopped-copy capture occurred, preserve an explicit loss report.
|
||||
const uncapturedInstructions = await instructionCopies.reportUnavailable(run.companyId, run.id);
|
||||
if (uncapturedInstructions?.state === "unavailable") {
|
||||
await appendRunEvent(run, { eventType: "instruction_save", stream: "system", level: "warn",
|
||||
message: "Instruction edits could not be recovered before environment release. No instruction save is claimed.",
|
||||
payload: { state: "unavailable", code: uncapturedInstructions.errorCode } });
|
||||
}
|
||||
instructionCopies.release(run.companyId, run.id);
|
||||
await releaseEnvironmentLeasesForRun({
|
||||
runId: run.id,
|
||||
companyId: run.companyId,
|
||||
|
||||
@@ -5766,6 +5766,57 @@ describe("native session same-turn steering", () => {
|
||||
});
|
||||
|
||||
describe("native warm session supervision", () => {
|
||||
it.each([
|
||||
{ changed: false, closeFails: false },
|
||||
{ changed: true, closeFails: false },
|
||||
{ changed: true, closeFails: true },
|
||||
])("collects changed instructions only after the owned warm provider stops (changed=$changed, close fails=$closeFails)", async ({ changed, closeFails }) => {
|
||||
const identity = `instruction-close-${changed}-${closeFails}`;
|
||||
let releaseClose!: () => void;
|
||||
const closed = new Promise<void>((resolve) => { releaseClose = resolve; });
|
||||
const close = vi.fn(async () => { await closed; if (closeFails) throw new Error("instruction provider close failed"); });
|
||||
const collectStopped = vi.fn(async () => {});
|
||||
const hasChanges = vi.fn(async () => changed);
|
||||
const warmExecution = { ...execution,
|
||||
binding: { ...execution.binding, runId: identity, executionWorkspaceId: identity },
|
||||
session: { ...execution.session, normalizedSessionId: identity, lifecyclePolicy: { mode: "warm" as const, idleTimeoutMs: 60_000 } },
|
||||
} as NativeExecutionInputV1;
|
||||
state.execute.mockReset().mockImplementationOnce(async (options) => {
|
||||
expect(options.requireSessionCloseBeforeReturn).toBe(true);
|
||||
expect(options.onSessionClosed).toBe(collectStopped);
|
||||
await options.onSession?.({ close });
|
||||
return { result: { summary: "completed" }, terminal: { runTerminalState: "succeeded" },
|
||||
turnId: identity, normalizedSessionId: identity, providerSessionId: identity,
|
||||
driverKind: "test", driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null };
|
||||
});
|
||||
const running = executePaperclipNativeSession({ db: leaseDb(warmExecution), execution: warmExecution,
|
||||
runnerInstanceId: identity, runnerExecutionTarget: { kind: "remote", transport: "sandbox", environmentId: identity, remoteCwd: `/tmp/${identity}` },
|
||||
instructionWorkingCopy: { hasChanges, collectStopped } });
|
||||
const observed = running.then(() => null, error => error);
|
||||
try {
|
||||
if (changed) {
|
||||
await vi.waitFor(() => expect(close).toHaveBeenCalledOnce());
|
||||
expect(collectStopped).not.toHaveBeenCalled();
|
||||
releaseClose();
|
||||
const error = await observed;
|
||||
if (closeFails) {
|
||||
expect(error?.message).toBe("instruction provider close failed");
|
||||
expect(collectStopped).not.toHaveBeenCalled();
|
||||
} else {
|
||||
expect(error).toBeNull();
|
||||
expect(collectStopped).toHaveBeenCalledOnce();
|
||||
}
|
||||
} else {
|
||||
expect(await observed).toBeNull();
|
||||
expect(close).not.toHaveBeenCalled();
|
||||
expect(collectStopped).not.toHaveBeenCalled();
|
||||
}
|
||||
} finally {
|
||||
releaseClose();
|
||||
await closeWarmNativeSessionsForEnvironment({ environmentId: identity, reason: "test cleanup" });
|
||||
}
|
||||
});
|
||||
|
||||
it.each([true, false])(
|
||||
"uses provider turn completion without a semantic-result cutoff: chat=%s",
|
||||
async (conversationMode) => {
|
||||
|
||||
@@ -7111,6 +7111,12 @@ export async function executePaperclipNativeSession(input: {
|
||||
chatAttachmentReadScope?: NativeChatAttachmentReadScope;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
onEvent?: (event: AdapterRuntimeEvent) => Promise<void>;
|
||||
/** Only this run's registered private instruction entry.
|
||||
* Probe at terminal; persist only after owned shutdown. */
|
||||
instructionWorkingCopy?: {
|
||||
hasChanges: () => Promise<boolean>;
|
||||
collectStopped: () => Promise<void>;
|
||||
};
|
||||
/** Persist task-level continuity before a durable goal can outlive this run. */
|
||||
onGoalCheckpoint?: (snapshot: PersistedNativeSession) => Promise<void>;
|
||||
sessionGoalControl?: NativeSessionGoalControl | null;
|
||||
@@ -8232,7 +8238,8 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
resumeSessionGoalHeartbeat: input.resumeSessionGoalHeartbeat,
|
||||
// Every durable runner must finish its bounded suspension before
|
||||
// the next run verifies and rotates the saved authority.
|
||||
requireSessionCloseBeforeReturn: runnerdBackend !== null,
|
||||
requireSessionCloseBeforeReturn: runnerdBackend !== null || input.instructionWorkingCopy !== undefined,
|
||||
onSessionClosed: input.instructionWorkingCopy?.collectStopped,
|
||||
onCheckpoint: async (snapshot) => {
|
||||
if (warmSessionId !== null && warmConfigDigest !== null) {
|
||||
await persistWarmNativeCheckpoint(
|
||||
@@ -8989,12 +8996,22 @@ async function executePaperclipNativeSessionWithinScope(
|
||||
// A following run cannot attach until the prior run's durable finalization
|
||||
// is committed. Provider completion alone is not an authority boundary.
|
||||
if (warmSessionId !== null && lifecyclePolicy.mode === "warm") {
|
||||
const instructionCopy = input.instructionWorkingCopy;
|
||||
const ownedSession = warmNativeSessions.get(warmSessionId);
|
||||
const collectInstructions = Boolean(instructionCopy && ownedSession?.ownerToken === warmSessionOwnerToken &&
|
||||
await instructionCopy.hasChanges());
|
||||
if (collectInstructions && ownedSession) {
|
||||
// Keep the unchanged warm path intact. A changed private instruction copy
|
||||
// requires the existing checkpoint-and-close boundary before collection.
|
||||
ownedSession.closeOnReleaseReason = "registered instruction edits require stopped-provider collection";
|
||||
}
|
||||
await releaseWarmNativeSession(
|
||||
warmSessionId,
|
||||
warmSessionOwnerToken,
|
||||
lifecyclePolicy.idleTimeoutMs,
|
||||
false,
|
||||
);
|
||||
if (collectInstructions) await instructionCopy!.collectStopped();
|
||||
}
|
||||
const adapterResult: AdapterExecutionResult = {
|
||||
exitCode: native.terminal.runTerminalState === "succeeded" ? 0 : 1,
|
||||
|
||||
@@ -90,6 +90,28 @@ afterEach(async () => {
|
||||
});
|
||||
|
||||
describe("buildNativeRuntimeContext", () => {
|
||||
it("keeps the pinned prompt immutable while exposing the registered editable instruction copy", async () => {
|
||||
serviceMocks.exportFiles.mockResolvedValue({
|
||||
entryFile: "instructions/CHARter.md",
|
||||
files: { "instructions/CHARter.md": "Original instructions.\n" },
|
||||
});
|
||||
const workingRoot = path.join(temporaryRoots.at(-1)!, "private-run-copy");
|
||||
await mkdir(path.join(workingRoot, "instructions"), { recursive: true });
|
||||
await writeFile(path.join(workingRoot, "instructions/CHARter.md"), "Original instructions.\n");
|
||||
const context = await buildNativeRuntimeContext({
|
||||
db: {} as Db,
|
||||
agent: { id: "agent-1", companyId: "company-1", name: "Reviewer", adapterType: "paperclip_runner", adapterConfig: {} },
|
||||
runId: "run-1",
|
||||
runtimeConfig: {},
|
||||
runtimeSkillEntries: [],
|
||||
instructionWorkingCopy: { rootPath: workingRoot, entryPath: "instructions/CHARter.md" },
|
||||
});
|
||||
expect(context.instructions.workingCopy).toEqual({ rootPath: workingRoot, entryPath: "instructions/CHARter.md" });
|
||||
await writeFile(path.join(workingRoot, "instructions/CHARter.md"), "Persist this next time.\n");
|
||||
expect(await readFile(path.join(context.instructions.bundle.rootPath, context.instructions.entryPath), "utf8"))
|
||||
.toBe("Original instructions.\n");
|
||||
});
|
||||
|
||||
it.each(["disabled", "degraded"] as const)(
|
||||
"omits an unavailable native MCP connection when it is %s without aborting runtime context creation",
|
||||
async (unavailableState) => {
|
||||
|
||||
@@ -222,12 +222,17 @@ export async function resolveNativeRuntimeMcpSnapshot(input: { db: Db; agent: Pi
|
||||
return { assignmentSetId: `sha256:${assignmentDigest}`, digest: assignmentDigest, bindingId: assignment.connections.length ? `native-mcp:${input.runId}` : null };
|
||||
}
|
||||
|
||||
export async function buildNativeRuntimeContext(input: { db: Db; agent: RuntimeAgent; runId: string; runtimeConfig: Record<string, unknown>; runtimeSkillEntries: PaperclipSkillEntry[] }): Promise<NativeRuntimeContextSnapshot> {
|
||||
export async function buildNativeRuntimeContext(input: { db: Db; agent: RuntimeAgent; runId: string; runtimeConfig: Record<string, unknown>; runtimeSkillEntries: PaperclipSkillEntry[]; instructionWorkingCopy?: { rootPath: string; entryPath: string } }): Promise<NativeRuntimeContextSnapshot> {
|
||||
const [instructions, skills, mcp] = await Promise.all([
|
||||
materializeInstructionBundle(input.agent),
|
||||
materializeSelectedSkills(input.runtimeConfig, input.runtimeSkillEntries, input.agent.adapterType === "paperclip_runner"),
|
||||
resolveNativeRuntimeMcpSnapshot({ db: input.db, agent: input.agent, runId: input.runId }),
|
||||
]);
|
||||
const snapshot = { prompt: { revision: PAPERCLIP_EXECUTION_PROMPT_REVISION, text: PAPERCLIP_EXECUTION_PROMPT, digest: nativeRuntimePromptDigest() }, instructions, skills, mcp };
|
||||
const snapshot = {
|
||||
prompt: { revision: PAPERCLIP_EXECUTION_PROMPT_REVISION, text: PAPERCLIP_EXECUTION_PROMPT, digest: nativeRuntimePromptDigest() },
|
||||
instructions: { ...instructions, ...(input.instructionWorkingCopy ? { workingCopy: input.instructionWorkingCopy } : {}) },
|
||||
skills,
|
||||
mcp,
|
||||
};
|
||||
return parseNativeRuntimeContext({ ...snapshot, aggregateDigest: canonicalNativeRuntimeContextDigest(snapshot) });
|
||||
}
|
||||
@@ -225,3 +225,12 @@ first provider turn intentionally omits task disposition, and their second turn
|
||||
must be an automatic, causally bound repair that records completion. They use
|
||||
public task comments/status APIs and run-detail evidence; no private runtime
|
||||
hooks or database mutations are used by the fixture.
|
||||
|
||||
## Registered instruction persistence
|
||||
|
||||
The `instruction_persistence` flow uses production managed instructions and the
|
||||
public versioned entry endpoint. It preserves the initial content, requires an
|
||||
ordinary private file edit to create a cleanup-source revision, restarts the server,
|
||||
and creates a fresh browser task. Its final oracle downloads and verifies the
|
||||
attachment bytes and SHA-256, rather than accepting a filename or model claim.
|
||||
Both tasks' runs count toward billing and teardown. The suite is explicit-only.
|
||||
@@ -122,6 +122,18 @@ duplicating the final response. The second workflow restarts the isolated
|
||||
Paperclip server while the interaction is waiting, reloads that state, and
|
||||
then resumes it. The suite has no Daytona cells.
|
||||
|
||||
`instruction-persistence` is an explicit-only three-cell workflow: legacy and
|
||||
native Codex locally, plus native Codex on Daytona. Each creates two browser tasks
|
||||
for the same agent. The first edits only its registered private instruction file
|
||||
using ordinary filesystem tools; the oracle checks exact canonical bytes and a
|
||||
new cleanup revision bound to that run. The harness restarts Paperclip and creates
|
||||
a fresh task, which must upload a downloaded text attachment containing the saved
|
||||
nonce. It rejects tool/API saves, stale revisions, missing downloads, and incorrect
|
||||
bytes. The deadline is twenty minutes per cell, with two expected provider runs;
|
||||
normal instance/Daytona cleanup, screenshots, evidence, and billing apply. Run with
|
||||
`pnpm test:e2e:runner -- --suite instruction-persistence`. The unchanged-provider
|
||||
warm lifecycle is independently covered by `daytona-warm-continuity`.
|
||||
|
||||
`daytona-warm-continuity` (**Daytona Warm Continuity**) is exactly two paid
|
||||
cells: legacy Codex and Runner Codex against one reusable warm Daytona
|
||||
configuration. Each cell creates a real project with a primary local-path
|
||||
|
||||
@@ -71,10 +71,10 @@ describe("runner E2E catalog", () => {
|
||||
expect(localIntegrityTasks).toHaveLength(2);
|
||||
expect(openRouterBreadthTasks).toHaveLength(3);
|
||||
expect(runnerSuites.map((suite) => suite.expectedMatrixSize)).toEqual([
|
||||
8, 46, 23, 47, 52, 28, 18, 6, 6, 42, 14, 10, 2,
|
||||
3, 8, 46, 23, 47, 52, 28, 18, 6, 6, 42, 14, 10, 2,
|
||||
]);
|
||||
expect(validateRunnerCatalog()).toHaveLength(302);
|
||||
expect(new Set(runnerMatrix.map((entry) => entry.id)).size).toBe(302);
|
||||
expect(validateRunnerCatalog()).toHaveLength(305);
|
||||
expect(new Set(runnerMatrix.map((entry) => entry.id)).size).toBe(305);
|
||||
expect(
|
||||
runnerMatrix.filter((entry) => entry.suite.id === "core-compatibility"),
|
||||
).toHaveLength(42);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { instructionPersistenceTask } from "./instruction-persistence.js";
|
||||
import { accountingTasks } from "./accounting-cases.js";
|
||||
import { continuationTasks } from "./continuation-cases.js";
|
||||
import { lifecycleLiveTasks, lifecycleLiveDefinitionDigest } from "./lifecycle-live-cases.js";
|
||||
@@ -911,6 +912,15 @@ const everydayProfiles = [
|
||||
].map(productionStoryProfile);
|
||||
|
||||
export const runnerSuites: readonly RunnerSuiteFixture[] = [
|
||||
{
|
||||
id: "instruction-persistence", label: "Instruction Persistence",
|
||||
description: "Ordinary private file edits persist as canonical revisions and survive a server restart and fresh browser task.",
|
||||
groups: [], profiles: codexContinuityProfiles,
|
||||
environments: [localEnvironment, runnerEnvironments.find(environment => environment.id === "daytona")!], tasks: [instructionPersistenceTask],
|
||||
excludedExecutionIds: ["instruction-persistence.legacy-codex.daytona.private-copy-persists"],
|
||||
expectedMatrixSize: 3, manualOnly: true,
|
||||
definitionMetadata: { version: 1, oracle: "canonical-revision-and-downloaded-attachment", providerTurns: 2, restart: "between-tasks", instructions: "production" },
|
||||
},
|
||||
{
|
||||
id: "continuation-accounting", label: "Continuation accounting baseline", manualOnly: true,
|
||||
description: "Structured productive steps, bounded repair, restart and late gates; comments cannot buy more attempts.",
|
||||
@@ -1169,6 +1179,7 @@ export function validateRunnerCatalog(): MatrixExecution[] {
|
||||
...localIntegrityTasks,
|
||||
...openRouterBreadthTasks,
|
||||
daytonaWarmContinuityTask,
|
||||
instructionPersistenceTask,
|
||||
];
|
||||
for (const [label, values] of [
|
||||
["suite", runnerSuites],
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { gradeInstructionPersistence } from "./instruction-persistence.js";
|
||||
import { runnerMatrix } from "./catalog.js";
|
||||
const expectedContent = "original\nInstruction persistence nonce: fixture\n";
|
||||
const valid = () => ({ before: { revision: { id: "before" } },
|
||||
after: { content: expectedContent, revision: { id: "after", source: "cleanup", sourceRunId: "first-run", contentHash: createHash("sha256").update(expectedContent).digest("hex") } },
|
||||
firstRunId: "first-run", expectedContent,
|
||||
proof: { body: "Instruction persistence nonce: fixture\n", contentVerified: true },
|
||||
expectedProof: "Instruction persistence nonce: fixture\n" });
|
||||
describe("instruction persistence independent oracle", () => {
|
||||
it("accepts exact cleanup content and a verified downloaded fresh-task proof", () => {
|
||||
expect(gradeInstructionPersistence(valid()).every(row => row.passed)).toBe(true);
|
||||
});
|
||||
it.each(["tool-save", "wrong-run", "unchanged-revision", "wrong-bytes", "missing-proof", "unverified-proof", "wrong-proof"])("rejects %s evidence", kind => {
|
||||
const value = valid();
|
||||
if (kind === "tool-save") value.after.revision.source = "tool";
|
||||
if (kind === "wrong-run") value.after.revision.sourceRunId = "another-run";
|
||||
if (kind === "unchanged-revision") value.after.revision.id = "before";
|
||||
if (kind === "wrong-bytes") value.after.content = "claimed success";
|
||||
if (kind === "missing-proof") return expect(gradeInstructionPersistence({ ...value, proof: undefined }).some(row => !row.passed)).toBe(true);
|
||||
if (kind === "unverified-proof") value.proof.contentVerified = false;
|
||||
if (kind === "wrong-proof") value.proof.body = "claimed success";
|
||||
expect(gradeInstructionPersistence(value).some(row => !row.passed)).toBe(true);
|
||||
});
|
||||
it("registers exactly three explicit cells without changing scheduled campaigns", () => {
|
||||
const cells = runnerMatrix.filter(row => row.suite.id === "instruction-persistence");
|
||||
expect(cells.map(row => `${row.profile.id}.${row.environment.id}`)).toEqual(["legacy-codex.local", "runner-codex.local", "runner-codex.daytona"]);
|
||||
expect(cells.every(row => row.suite.manualOnly && row.task.expectedRunCount === 2)).toBe(true);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,106 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { expect, type Page } from "@playwright/test";
|
||||
import { pollUntil, type RunnerApi } from "./api.js";
|
||||
import { captureFirstTaskAttachments } from "./first-task-attachments.js";
|
||||
import { collectRunEvents } from "./run-observations.js";
|
||||
import { createTaskThroughUi } from "./user-actions.js";
|
||||
import type { LiveFixtureValues } from "./live-fixtures.js";
|
||||
import type { MatrixExecution, RunnerTaskFixture } from "./types.js";
|
||||
|
||||
type Row = Record<string, any>;
|
||||
export const instructionNonceLine = (nonce: string) => `Instruction persistence nonce: ${nonce}\n`;
|
||||
export const instructionPersistenceTask: RunnerTaskFixture = {
|
||||
id: "private-copy-persists", label: "Private instruction edit survives a fresh task",
|
||||
groups: [], workMode: "standard", flow: "instruction_persistence",
|
||||
expectedRunCount: 2, attemptTimeoutMs: { local: 20 * 60_000, daytona: 20 * 60_000 },
|
||||
expectedTerminalState: { issue: "done", run: "succeeded" },
|
||||
buildTitle: nonce => `Persist private instructions ${nonce}`,
|
||||
buildVisibleMarker: nonce => `INSTRUCTIONS-VERIFIED-${nonce}`,
|
||||
buildPrompt: nonce => [
|
||||
"Edit your own registered writable agent instruction entry with ordinary filesystem tools. The runtime guidance gives its exact private path.",
|
||||
`Preserve its existing bytes and append exactly one newline followed by this line and a final newline: Instruction persistence nonce: ${nonce}`,
|
||||
"Do not use update_agent_instructions, restore_agent_instructions, or an instructions API to save it. Do not edit repository AGENTS.md or the read-only loaded bundle.",
|
||||
"Reply only Instruction copy edited without printing filesystem paths, then complete this task after the file edit. Paperclip will collect it after the provider stops; do not claim it has already persisted. Do not create further tasks.",
|
||||
].join("\n"),
|
||||
buildMatchers: () => [], // Independent public revision and attachment oracle below.
|
||||
};
|
||||
|
||||
export function gradeInstructionPersistence(input: { before: Row; after: Row; firstRunId: string; expectedContent: string; proof: Row | undefined; expectedProof: string }) {
|
||||
return [
|
||||
{ id: "new-cleanup-revision", passed: Boolean(input.before.revision?.id && input.after.revision?.id && input.after.revision.id !== input.before.revision.id && input.after.revision.source === "cleanup" && input.after.revision.sourceRunId === input.firstRunId) },
|
||||
{ id: "exact-canonical-bytes", passed: input.after.content === input.expectedContent && input.after.revision?.contentHash === createHash("sha256").update(input.expectedContent).digest("hex") },
|
||||
{ id: "fresh-task-downloaded-proof", passed: input.proof?.contentVerified === true && input.proof.body === input.expectedProof },
|
||||
].map(check => ({ ...check, detail: check.passed ? `${check.id} verified independently` : `${check.id} missing or incorrect` }));
|
||||
}
|
||||
|
||||
export async function runInstructionPersistenceFlow(input: {
|
||||
page: Page; api: RunnerApi; fixtures: LiveFixtureValues; execution: MatrixExecution; nonce: string;
|
||||
secrets: readonly string[]; deadlineAt: number;
|
||||
restart(): Promise<void>;
|
||||
observe(issue: Row, runs: Row[]): void;
|
||||
capture(id: string, label: string, file: string): Promise<void>;
|
||||
evidence(name: string, data: unknown): Promise<void>;
|
||||
}) {
|
||||
const { page, api, fixtures, execution, nonce } = input;
|
||||
const filePath = `/api/agents/${fixtures.agent.id}/instructions-bundle/file?path=AGENTS.md`;
|
||||
const before = await api.get<Row>(filePath);
|
||||
if (typeof before.content !== "string" || !before.revision?.id) throw new Error("Managed instructions must expose a canonical baseline revision");
|
||||
const expectedContent = `${before.content}\n${instructionNonceLine(nonce)}`;
|
||||
let issue: Row = {};
|
||||
let runs: Row[] = [];
|
||||
async function create(title: string, prompt: string) {
|
||||
await createTaskThroughUi({ page, issuePrefix: fixtures.company.issuePrefix!, agentName: fixtures.agent.name, title, prompt, workMode: "standard", projectName: fixtures.project?.name });
|
||||
const found = await pollUntil({ label: `instruction task ${title}`, deadlineAt: input.deadlineAt,
|
||||
load: async () => (await api.get<Row[]>(`/api/companies/${fixtures.company.id}/issues?limit=100`)).find(row => row.title === title), accept: row => Boolean(row) });
|
||||
if (!found) throw new Error("Browser-created instruction task missing");
|
||||
issue = found;
|
||||
input.observe(issue, runs);
|
||||
await page.goto(`/${fixtures.company.issuePrefix}/issues/${issue.identifier ?? issue.id}`);
|
||||
}
|
||||
async function settle(count: number) {
|
||||
await pollUntil({ label: `instruction run ${count} completed`, deadlineAt: input.deadlineAt,
|
||||
load: async () => {
|
||||
issue = await api.get<Row>(`/api/issues/${issue.id}`);
|
||||
const listed = await api.get<Row[]>(`/api/companies/${fixtures.company.id}/heartbeat-runs?limit=100`);
|
||||
runs = await Promise.all(listed.map(row => api.get<Row>(`/api/heartbeat-runs/${row.id}`)));
|
||||
runs.sort((a, b) => String(a.createdAt).localeCompare(String(b.createdAt)));
|
||||
input.observe(issue, runs);
|
||||
return { issue, runs };
|
||||
},
|
||||
accept: state => state.issue.status === "done" && state.runs.length === count && state.runs.every(row => row.status === "succeeded"),
|
||||
reject: state => state.runs.some(row => ["failed", "cancelled", "timed_out"].includes(row.status)) ? "Instruction task provider run failed" : state.runs.length > count ? "Instruction task dispatched an extra run" : undefined,
|
||||
});
|
||||
expect(runs.every(row => row.runtimeMode === execution.profile.expectedRuntimeMode)).toBe(true);
|
||||
await page.reload();
|
||||
await expect(page.getByTestId("issue-detail-header").getByRole("button", { name: "Change status (current: Done)", exact: true })).toBeVisible();
|
||||
}
|
||||
await create(execution.task.buildTitle(nonce), execution.task.buildPrompt(nonce));
|
||||
await settle(1);
|
||||
const firstRunId = runs[0]!.id;
|
||||
const after = await pollUntil({ label: "stopped instruction cleanup revision", deadlineAt: Math.min(input.deadlineAt, Date.now() + 30_000),
|
||||
load: () => api.get<Row>(filePath), accept: row => row.content === expectedContent && row.revision?.sourceRunId === firstRunId });
|
||||
const events = await collectRunEvents<Row>((afterSeq, limit) => api.get(`/api/heartbeat-runs/${firstRunId}/events?afterSeq=${afterSeq}&limit=${limit}`));
|
||||
expect(events.some(row => row.eventType === "instruction_save" && row.payload?.state === "saved")).toBe(true);
|
||||
await input.evidence("instruction-first-save.json", { before, after, run: runs[0], events });
|
||||
await input.capture("instruction-edited", "Private instructions saved after provider stop", "instruction-edited.png");
|
||||
// A new server and a new issue cannot pass by retaining model conversation.
|
||||
await input.restart();
|
||||
expect((await api.get<Row>(filePath)).content).toBe(expectedContent);
|
||||
await create(`Read persisted instructions ${nonce}`, [
|
||||
"Read your own loaded agent instruction entry (or its current registered private copy) using ordinary filesystem tools.",
|
||||
"Find the line beginning 'Instruction persistence nonce: '. Copy that entire line plus one final newline into instruction-proof.txt. Do not infer the value from this task title or other task history. Do not change your instructions.",
|
||||
"Upload instruction-proof.txt as a text/plain task attachment named instruction-proof.txt using the normal artifact workflow. A local file alone is insufficient.",
|
||||
`Reply with exactly ${execution.task.buildVisibleMarker(nonce)} and complete the task.`,
|
||||
].join("\n"));
|
||||
await settle(2);
|
||||
const attachments = await captureFirstTaskAttachments(api, [{ ...issue, id: String(issue.id) }], input.secrets);
|
||||
const proof = attachments.find(row => row.originalFilename === "instruction-proof.txt" || row.name === "instruction-proof.txt");
|
||||
const final = await api.get<Row>(filePath);
|
||||
expect(final.revision.id).toBe(after.revision.id);
|
||||
const checks = gradeInstructionPersistence({ before, after, firstRunId, expectedContent, proof, expectedProof: instructionNonceLine(nonce) });
|
||||
await input.evidence("instruction-persistence.json", { checks, before, after, final, issues: [runs[0]?.nativeIssueId, issue.id], runs, attachments });
|
||||
await expect(page.getByTestId("task-chat-agent-bubble").filter({ hasText: execution.task.buildVisibleMarker(nonce) }).last()).toBeVisible();
|
||||
await input.capture("final-state", "Fresh task downloaded the persisted instruction nonce", "final-state.png");
|
||||
expect(checks.filter(check => !check.passed), "Independent instruction persistence checks").toEqual([]);
|
||||
return { issue, runs, checks };
|
||||
}
|
||||
@@ -1,3 +1,4 @@
|
||||
import { runInstructionPersistenceFlow } from "./instruction-persistence.js";
|
||||
import { observeBrowserBootstrap } from "./browser-bootstrap-diagnostics.js";
|
||||
import { runAccountingFlow } from "./accounting-flow.js";
|
||||
import type { Issue } from "../../packages/shared/src/types/issue.js";
|
||||
@@ -538,7 +539,7 @@ for (const execution of executions) {
|
||||
const credentials = credentialValues();
|
||||
const secrets = normalizedSecrets(Object.values(credentials));
|
||||
const api = new RunnerApi(request);
|
||||
const companyRunFlow = ["continuation_accounting", "continuation", "agent_chat", "everyday_workflow", "first_task"].includes(execution.task.flow);
|
||||
const companyRunFlow = ["continuation_accounting", "continuation", "agent_chat", "everyday_workflow", "first_task", "instruction_persistence"].includes(execution.task.flow);
|
||||
const consoleDiagnostics: Array<Record<string, unknown>> = [];
|
||||
const networkDiagnostics: Array<Record<string, unknown>> = [];
|
||||
const pageLifecycleDiagnostics: Array<Record<string, unknown>> = [];
|
||||
@@ -841,6 +842,16 @@ for (const execution of executions) {
|
||||
evidence: (name, data) => writeSanitizedJson(snapshotsDir, name, data, secrets),
|
||||
});
|
||||
issue = accounting.issue as IssueRecord; selectedRuns = accounting.runs as RunRecord[];
|
||||
} else if (execution.task.flow === "instruction_persistence") {
|
||||
const story = await runInstructionPersistenceFlow({
|
||||
page, api, fixtures, execution, nonce, secrets, deadlineAt: startedAtMs + deadlineMs,
|
||||
restart: () => restartIsolatedPaperclipServer({ api, requestId: `instructions-${nonce}`, deadlineAt: startedAtMs + deadlineMs }),
|
||||
observe: (currentIssue, currentRuns) => { issue = currentIssue as IssueRecord; selectedRuns = currentRuns as RunRecord[]; },
|
||||
capture: captureScreenshot,
|
||||
evidence: (name, data) => writeSanitizedJson(snapshotsDir, name, data, secrets),
|
||||
});
|
||||
issue = story.issue as IssueRecord; selectedRuns = story.runs as RunRecord[];
|
||||
matcherResults = story.checks.map(check => ({ matcher: { kind: "json_path" as const, path: `instructions.${check.id}`, expected: true }, passed: check.passed, detail: check.detail }));
|
||||
} else if (execution.task.flow === "continuation") {
|
||||
const continuation = await runContinuationFlow({
|
||||
page, api, fixtures, execution, nonce, secrets, workspacePath, deadlineAt: startedAtMs + deadlineMs - 60_000,
|
||||
|
||||
@@ -21,7 +21,8 @@ export type RunnerTaskFlow =
|
||||
| "plan_revision_acceptance"
|
||||
| "question_resume_completion"
|
||||
| "plan_approval_completion"
|
||||
| "warm_three_turn";
|
||||
| "warm_three_turn"
|
||||
| "instruction_persistence";
|
||||
|
||||
export interface SecretReference {
|
||||
type: "secret_ref";
|
||||
|
||||
Reference in new issue
Block a user