mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 06:16:06 -10:00
Merge remote-tracking branch 'origin/master' into chat-eval/master-candidate
* origin/master: test(runner): set default inputs in workflow target fixture (#19602) fix(native): re-bootstrap a Daytona runner restarted before session.open (#19583) fix(tasks): advertise a native resume only while the scheduler would claim it (#19523) fix(evals): treat a job-less TestOps plan row as unassigned during registration (#19601) fix(workspaces): let non-git workspace directories pass close readiness (#19553)
This commit is contained in:
commit
e5726b9dbf
16 files changed
+622
-63
No files matched your search
@@ -347,6 +347,14 @@ For a prepared remote workspace interrupted before its first executor claim,
|
||||
that ledger records `remote_execution_never_claimed` with a
|
||||
`bootstrap_incomplete` disposition. The provider attempt remains zero during
|
||||
classification; the executor increments it before starting provider work.
|
||||
A claimed remote bootstrap (the sandbox runner was launched and may have
|
||||
connected) records `remote_bootstrap_proven_incomplete` with the same
|
||||
disposition. Recovery admits it only after the previous controller lease has
|
||||
expired and the Postgres controller checkpoint holds just `run.prepare` and an
|
||||
unfinished `session.open`, with no committed or acknowledged event. The
|
||||
executor restores that checkpoint and checks the same proof before it
|
||||
bootstraps again. After three provider attempts it records
|
||||
`execution_recovery_budget_exhausted` instead.
|
||||
|
||||
For a local Codex, OpenCode, or ACPX turn lost while runnerd was stopped, the canonical `turn.failed`
|
||||
event retains `error.code: "provider_turn_lost_on_restore"` and
|
||||
|
||||
@@ -2419,6 +2419,129 @@ describeEmbeddedPostgres("executionWorkspaceService.getCloseReadiness", () => {
|
||||
]));
|
||||
});
|
||||
|
||||
async function seedSharedLocalWorkspace(cwd: string, options: { metadata?: Record<string, unknown> } = {}) {
|
||||
const companyId = randomUUID();
|
||||
const projectId = randomUUID();
|
||||
const projectWorkspaceId = randomUUID();
|
||||
const executionWorkspaceId = randomUUID();
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: "PAP",
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(projects).values({
|
||||
id: projectId,
|
||||
companyId,
|
||||
name: "Workspaces",
|
||||
status: "in_progress",
|
||||
executionWorkspacePolicy: { enabled: true },
|
||||
});
|
||||
await db.insert(projectWorkspaces).values({
|
||||
id: projectWorkspaceId,
|
||||
companyId,
|
||||
projectId,
|
||||
name: "Primary",
|
||||
sourceType: "local_path",
|
||||
isPrimary: true,
|
||||
cwd,
|
||||
});
|
||||
await db.insert(executionWorkspaces).values({
|
||||
id: executionWorkspaceId,
|
||||
companyId,
|
||||
projectId,
|
||||
projectWorkspaceId,
|
||||
mode: "shared_workspace",
|
||||
strategyType: "project_primary",
|
||||
name: "Shared workspace",
|
||||
status: "active",
|
||||
providerType: "local_fs",
|
||||
cwd,
|
||||
metadata: options.metadata ?? null,
|
||||
});
|
||||
return { executionWorkspaceId };
|
||||
}
|
||||
|
||||
it("treats an existing workspace directory that is not a git repository as ready", async () => {
|
||||
const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-non-git-workspace-"));
|
||||
tempDirs.add(workspaceDir);
|
||||
await fs.writeFile(path.join(workspaceDir, "notes.txt"), "plain files\n", "utf8");
|
||||
const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir);
|
||||
|
||||
const readiness = await svc.getCloseReadiness(executionWorkspaceId);
|
||||
|
||||
expect(readiness).toMatchObject({
|
||||
state: "ready_with_warnings",
|
||||
isDestructiveCloseAllowed: true,
|
||||
git: { repoRoot: null, hasDirtyTrackedFiles: false, hasUntrackedFiles: false },
|
||||
});
|
||||
expect(readiness?.blockingReasons).toEqual([]);
|
||||
expect(readiness?.warnings.some((warning) => warning.startsWith("Could not inspect git status"))).toBe(false);
|
||||
});
|
||||
|
||||
it("plans removal of a runtime-created plain workspace directory", async () => {
|
||||
const parentDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-non-git-runtime-"));
|
||||
tempDirs.add(parentDir);
|
||||
const workspaceDir = path.join(parentDir, "task");
|
||||
await fs.mkdir(workspaceDir);
|
||||
const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir, {
|
||||
metadata: { createdByRuntime: true },
|
||||
});
|
||||
await db.update(projectWorkspaces).set({ cwd: path.join(parentDir, "primary") })
|
||||
.where(eq(projectWorkspaces.cwd, workspaceDir));
|
||||
|
||||
const readiness = await svc.getCloseReadiness(executionWorkspaceId);
|
||||
|
||||
expect(readiness?.blockingReasons).toEqual([]);
|
||||
expect(readiness?.git).toMatchObject({ repoRoot: null, createdByRuntime: true });
|
||||
expect(readiness?.plannedActions.map((action) => action.kind)).toContain("remove_local_directory");
|
||||
});
|
||||
|
||||
it("still blocks close when an ancestor directory holds an unreadable .git marker", async () => {
|
||||
const parentDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-ancestor-git-workspace-"));
|
||||
tempDirs.add(parentDir);
|
||||
await fs.writeFile(path.join(parentDir, ".git"), `gitdir: ${path.join(parentDir, "missing-git-dir")}\n`, "utf8");
|
||||
const workspaceDir = path.join(parentDir, "nested", "task");
|
||||
await fs.mkdir(workspaceDir, { recursive: true });
|
||||
const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir);
|
||||
|
||||
const readiness = await svc.getCloseReadiness(executionWorkspaceId);
|
||||
|
||||
expect(readiness).toMatchObject({ state: "blocked", isDestructiveCloseAllowed: false });
|
||||
expect(readiness?.blockingReasons).toContain(
|
||||
"Paperclip could not verify the workspace git status. Retry before destructive cleanup.",
|
||||
);
|
||||
});
|
||||
|
||||
it("still blocks close when the workspace .git marker is a dangling symlink", async () => {
|
||||
const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-dangling-git-workspace-"));
|
||||
tempDirs.add(workspaceDir);
|
||||
await fs.symlink(path.join(workspaceDir, "unmounted-repo", ".git"), path.join(workspaceDir, ".git"));
|
||||
const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir);
|
||||
|
||||
const readiness = await svc.getCloseReadiness(executionWorkspaceId);
|
||||
|
||||
expect(readiness).toMatchObject({ state: "blocked", isDestructiveCloseAllowed: false });
|
||||
expect(readiness?.blockingReasons).toContain(
|
||||
"Paperclip could not verify the workspace git status. Retry before destructive cleanup.",
|
||||
);
|
||||
});
|
||||
|
||||
it("still blocks close when a workspace git repository cannot be inspected", async () => {
|
||||
const workspaceDir = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-broken-git-workspace-"));
|
||||
tempDirs.add(workspaceDir);
|
||||
// A worktree-style .git pointer whose git dir is missing: git refuses to read it.
|
||||
await fs.writeFile(path.join(workspaceDir, ".git"), `gitdir: ${path.join(workspaceDir, "missing-git-dir")}\n`, "utf8");
|
||||
const { executionWorkspaceId } = await seedSharedLocalWorkspace(workspaceDir);
|
||||
|
||||
const readiness = await svc.getCloseReadiness(executionWorkspaceId);
|
||||
|
||||
expect(readiness).toMatchObject({ state: "blocked", isDestructiveCloseAllowed: false });
|
||||
expect(readiness?.blockingReasons).toContain(
|
||||
"Paperclip could not verify the workspace git status. Retry before destructive cleanup.",
|
||||
);
|
||||
});
|
||||
|
||||
it("clears matching environment selections transactionally without touching other workspaces", async () => {
|
||||
const companyId = randomUUID();
|
||||
const projectId = randomUUID();
|
||||
|
||||
@@ -4,6 +4,7 @@ import { createRunSecretRedactionRegistry } from "../services/run-secret-redacti
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, agentWakeupRequests, companies, createDb, applyPendingMigrations, heartbeatRuns, issues, nativeRunFinalizations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS } from "@paperclipai/db";
|
||||
import { taskInterruptionPresentation, taskInterruptionPresentations } from "../services/task-interruption-presentation.js";
|
||||
import { readNativeSessionResumptionAgent } from "../services/native-runtime/native-session-resumption-admission.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import { captureInterruptionQuery, seedInterruptionFixture } from "./helpers/task-interruption-fixture.js";
|
||||
@@ -39,6 +40,49 @@ describe.skipIf(!support.supported)("task interruption latest run query", () =>
|
||||
await taskInterruptionPresentation(db, issue)]) {
|
||||
expect(presentation).toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() });
|
||||
}
|
||||
// The coordinator also pushes nextAttemptAt while it defers an ineligible
|
||||
// candidate. Those deferrals, or one far past the failure, never hide Retry.
|
||||
const otherAgentId = randomUUID();
|
||||
await db.insert(agents).values({ id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" });
|
||||
const admission = async () => {
|
||||
const [currentIssue] = await db.select().from(issues).where(eq(issues.id, issue.id));
|
||||
const [currentRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId));
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId));
|
||||
return readNativeSessionResumptionAgent(db, { issue: currentIssue!, run: currentRun!, coordinator: coordinator! });
|
||||
};
|
||||
expect(await admission()).not.toBeNull();
|
||||
const otherRunId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: otherRunId, companyId, agentId: otherAgentId, status: "cancelled",
|
||||
finishedAt: new Date(Date.now() - 60_000), createdAt: new Date(Date.now() - 60 * 60_000) });
|
||||
const terminatedManagerId = randomUUID();
|
||||
await db.insert(agents).values({ id: terminatedManagerId, companyId, name: "Former manager",
|
||||
adapterType: "paperclip_runner", status: "terminated" });
|
||||
for (const [table, change, restore] of [
|
||||
["issue", { assigneeAgentId: otherAgentId }, { assigneeAgentId: agentId }],
|
||||
["issue", { status: "blocked" }, { status: "in_progress" }],
|
||||
["issue", { executionRunId: otherRunId }, { executionRunId: null }],
|
||||
["issue", { checkoutRunId: otherRunId }, { checkoutRunId: null }],
|
||||
["run", { contextSnapshot: { nativeQuestionCancellation: { requestId: "q" } } }, { contextSnapshot: {} }],
|
||||
// Invokability is more than status: an idle agent under a terminated manager is refused.
|
||||
["agent", { reportsTo: terminatedManagerId }, { reportsTo: null }],
|
||||
] as const) {
|
||||
if (table === "issue") await db.update(issues).set(change).where(eq(issues.id, issue.id));
|
||||
else if (table === "run") await db.update(heartbeatRuns).set(change).where(eq(heartbeatRuns.id, runId));
|
||||
else await db.update(agents).set(change).where(eq(agents.id, agentId));
|
||||
// The scheduler would defer this candidate, so the task offers Retry.
|
||||
if (table !== "run") expect(await admission()).toBeNull();
|
||||
expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id))
|
||||
.toMatchObject({ state: "failed", canRetry: true });
|
||||
if (table === "issue") await db.update(issues).set(restore).where(eq(issues.id, issue.id));
|
||||
else if (table === "run") await db.update(heartbeatRuns).set(restore).where(eq(heartbeatRuns.id, runId));
|
||||
else await db.update(agents).set(restore).where(eq(agents.id, agentId));
|
||||
}
|
||||
expect(await admission()).not.toBeNull();
|
||||
// A provider cooldown can schedule the resume an hour out; it stays visible.
|
||||
const cooldown = new Date(Date.now() + 60 * 60_000);
|
||||
await db.update(nativeRunFinalizations).set({ nextAttemptAt: cooldown }).where(eq(nativeRunFinalizations.runId, runId));
|
||||
expect((await taskInterruptionPresentations(db, companyId, [issue])).get(issue.id))
|
||||
.toMatchObject({ state: "retry_scheduled", canRetry: false, nextRetryAt: cooldown.toISOString() });
|
||||
// Exhausted recovery is terminal: exactly one Retry returns.
|
||||
await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", nextAttemptAt: null })
|
||||
.where(eq(nativeRunFinalizations.runId, runId));
|
||||
@@ -46,6 +90,29 @@ describe.skipIf(!support.supported)("task interruption latest run query", () =>
|
||||
.toMatchObject({ state: "failed", canRetry: true });
|
||||
});
|
||||
|
||||
it("decides several pending native resumes in one list read", async () => {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID();
|
||||
await db.insert(companies).values({ id: companyId, name: "Resuming list", issuePrefix: `L${companyId.slice(0, 7)}` });
|
||||
await db.insert(agents).values([{ id: agentId, companyId, name: "Resuming agent", adapterType: "paperclip_runner" },
|
||||
{ id: otherAgentId, companyId, name: "Other agent", adapterType: "paperclip_runner" }]);
|
||||
const resumeAt = new Date(Date.now() + 30_000);
|
||||
const list = [];
|
||||
for (let index = 0; index < 6; index++) {
|
||||
const [issue] = await db.insert(issues).values({ companyId, title: `Resume ${index}`, status: "in_progress",
|
||||
assigneeAgentId: index % 2 ? otherAgentId : agentId }).returning();
|
||||
const runId = randomUUID();
|
||||
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issue.id, scopeKind: "issue",
|
||||
status: "failed", runtimeMode: "native", errorCode: "native_session_interrupted", finishedAt: new Date() });
|
||||
await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId, phase: "retryable_failure",
|
||||
attempt: 1, nextAttemptAt: resumeAt });
|
||||
list.push(issue);
|
||||
}
|
||||
const result = await taskInterruptionPresentations(db, companyId, list);
|
||||
list.forEach((issue, index) => expect(result.get(issue.id)).toMatchObject(index % 2
|
||||
? { state: "failed", canRetry: true }
|
||||
: { state: "retry_scheduled", canRetry: false, nextRetryAt: resumeAt.toISOString() }));
|
||||
});
|
||||
|
||||
it("redacts workspace preparation errors on task detail and list reads", async () => {
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "a".repeat(64));
|
||||
try {
|
||||
|
||||
@@ -521,6 +521,31 @@ async function pathExists(value: string | null | undefined) {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns false only when no `.git` entry exists on the path or any ancestor.
|
||||
* Uses lstat so a dangling `.git` symlink still counts, and treats any error
|
||||
* other than ENOENT as "may exist" so unreadable locations fail closed.
|
||||
*/
|
||||
async function hasGitMarkerInPathOrAncestors(start: string) {
|
||||
let current: string;
|
||||
try {
|
||||
current = await fs.realpath(start);
|
||||
} catch {
|
||||
return true;
|
||||
}
|
||||
while (true) {
|
||||
try {
|
||||
await fs.lstat(path.join(current, ".git"));
|
||||
return true;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException | null)?.code !== "ENOENT") return true;
|
||||
}
|
||||
const parent = path.dirname(current);
|
||||
if (parent === current) return false;
|
||||
current = parent;
|
||||
}
|
||||
}
|
||||
|
||||
async function runGit(args: string[], cwd: string) {
|
||||
return await execFileAsync("git", ["-C", cwd, ...args], { cwd });
|
||||
}
|
||||
@@ -963,6 +988,29 @@ async function inspectGitCloseReadiness(workspace: ExecutionWorkspace): Promise<
|
||||
try {
|
||||
repoRoot = (await runGit(["rev-parse", "--show-toplevel"], workspacePath)).stdout.trim() || null;
|
||||
} catch (error) {
|
||||
// A plain directory (e.g. a local_fs shared workspace) has no git state to
|
||||
// lose. Only treat it that way when no `.git` exists on the path, so a
|
||||
// broken or unreadable repository still fails closed.
|
||||
if (workspace.providerType !== "git_worktree" && !(await hasGitMarkerInPathOrAncestors(workspacePath))) {
|
||||
return {
|
||||
git: {
|
||||
repoRoot: null,
|
||||
workspacePath,
|
||||
branchName: workspace.branchName,
|
||||
baseRef: workspace.baseRef,
|
||||
hasDirtyTrackedFiles: false,
|
||||
hasUntrackedFiles: false,
|
||||
dirtyEntryCount: 0,
|
||||
untrackedEntryCount: 0,
|
||||
aheadCount: null,
|
||||
behindCount: null,
|
||||
isMergedIntoBase: null,
|
||||
createdByRuntime,
|
||||
},
|
||||
warnings,
|
||||
statusInspectionSucceeded: true,
|
||||
};
|
||||
}
|
||||
warnings.push(
|
||||
`Could not inspect git status for "${workspacePath}": ${error instanceof Error ? error.message : String(error)}`,
|
||||
);
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
/**
|
||||
* Pure proofs over durable runner state that a bootstrap never reached
|
||||
* provider work. Restart recovery and the executor share them, so the claim
|
||||
* and the executor's own check accept exactly the same shapes.
|
||||
*/
|
||||
|
||||
export const RUNNERD_CONTROL_PLANE_STATE_SCHEMAS: ReadonlySet<string> = new Set([
|
||||
"paperclip.runner.durable.control-plane-state.v1",
|
||||
"paperclip.runner.durable.control-plane-state.v2",
|
||||
]);
|
||||
|
||||
const RUNNERD_BOOTSTRAP_COMMAND_TYPES = new Set(["run.prepare", "session.open"]);
|
||||
|
||||
function record(value: unknown): Record<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? (value as Record<string, unknown>)
|
||||
: {};
|
||||
}
|
||||
|
||||
/** The runner never connected, so every bootstrap command is still undelivered. */
|
||||
export function runnerdStateIsUnconnectedBootstrap(state: Record<string, unknown>): boolean {
|
||||
const commands = Array.isArray(state.commands) ? state.commands.map(record) : [];
|
||||
const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : [];
|
||||
return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) &&
|
||||
state.connectionCount === 0 && committedEvents.length === 0 &&
|
||||
commands.every(command => command.status === "pending" &&
|
||||
RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)));
|
||||
}
|
||||
|
||||
/**
|
||||
* A provider turn can start only after session.open completes and the
|
||||
* controller issues a turn command. A root whose controller journal stops
|
||||
* before that point (run.prepare may be acknowledged, session.open never is,
|
||||
* no event was committed or acknowledged) and whose runner journal emitted no
|
||||
* event either holds no provider work, so restart recovery may quarantine it
|
||||
* and bootstrap again. Any other shape stays fail-closed.
|
||||
*
|
||||
* `runnerJournal` is `undefined` when the runner journal is not on this host
|
||||
* (never written locally, or a remote runner), `null` when it is unreadable.
|
||||
*/
|
||||
export function runnerdStateIsIncompleteBootstrap(
|
||||
state: Record<string, unknown>,
|
||||
runnerJournal?: Record<string, unknown> | null,
|
||||
): boolean {
|
||||
if (runnerdStateIsUnconnectedBootstrap(state)) return true;
|
||||
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema))) return false;
|
||||
if (typeof state.connectionCount !== "number" || !Number.isSafeInteger(state.connectionCount) ||
|
||||
state.connectionCount < 0) return false;
|
||||
if (!Array.isArray(state.committedEvents) || state.committedEvents.length !== 0) return false;
|
||||
if (state.ackedSourceSeq !== 0) return false;
|
||||
if (state.warmTransition != null || state.completedWarmTransition != null ||
|
||||
state.stagedRunAttachment != null) return false;
|
||||
if (record(state.bootstrapHistory).adopted === true) return false;
|
||||
if (!Array.isArray(state.commands)) return false;
|
||||
const commands = state.commands.map(record);
|
||||
if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) &&
|
||||
(command.status === "pending" ||
|
||||
(command.type === "run.prepare" && command.status === "completed")))) return false;
|
||||
// The controller cannot see runner events that were emitted but never
|
||||
// committed. When the runner journal exists it must agree that none were.
|
||||
if (runnerJournal === undefined) return true;
|
||||
if (runnerJournal === null) return false;
|
||||
const processed = Object.values(record(runnerJournal.processedCommands)).map(record);
|
||||
return Array.isArray(runnerJournal.outbox) && runnerJournal.outbox.length === 0 &&
|
||||
runnerJournal.nextSourceSeq === 1 && runnerJournal.ackedSourceSeq === 0 &&
|
||||
processed.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.commandType)) &&
|
||||
(command.commandType === "run.prepare" || command.status !== "completed"));
|
||||
}
|
||||
@@ -1,4 +1,5 @@
|
||||
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
||||
import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js";
|
||||
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
|
||||
import { pendingNativeRunProjectionCondition } from "./native-run-projection.js";
|
||||
import { nativeStatusReassessmentSourceIsSuperseded, reassessNativeStatusFromSource } from "./native-status-reassessment.js";
|
||||
@@ -158,7 +159,7 @@ export async function claimNativeSessionResumptions(input: {
|
||||
if (await retireSupersededNativeRetry(tx as unknown as Db, {
|
||||
run, coordinator, issueExecutionRunId: issue.executionRunId, now,
|
||||
})) return false;
|
||||
if (!["todo", "in_progress", "in_review"].includes(issue.status)) return false;
|
||||
if (!(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status)) return false;
|
||||
const row = { run, coordinator };
|
||||
if (
|
||||
row.run.runtimeMode !== "native" ||
|
||||
|
||||
@@ -11,7 +11,8 @@ import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-pre
|
||||
import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
|
||||
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
|
||||
import { preserveNativeFailedWorkspace } from "./native-failed-workspace-recovery.js";
|
||||
import { hasNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader } from "./native-runner-checkpoint.js";
|
||||
import { hasNativeRunnerCheckpoint, loadNativeRunnerCheckpointHeader,
|
||||
nativeRunnerCheckpointProvesIncompleteBootstrap } from "./native-runner-checkpoint.js";
|
||||
import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js";
|
||||
import { readNativeWorkspaceSyncReference } from "./native-workspace-sync.js";
|
||||
import { activeNativeExecutionRunIds, NATIVE_PREPARATION_FENCING_VERSION } from "./native-active-runs.js";
|
||||
@@ -340,6 +341,8 @@ export function classifyNativeRunnerRecoveryEvidence(input: {
|
||||
runnerGroupAlive: boolean;
|
||||
remoteSandbox?: boolean;
|
||||
unclaimedBootstrap?: boolean;
|
||||
/** The durable controller checkpoint of a claimed remote bootstrap proves no provider work. */
|
||||
remoteBootstrapProvenIncomplete?: boolean;
|
||||
processStartMatches: boolean;
|
||||
knownProviderProcessAlive?: boolean;
|
||||
knownProviderProcessIdentityAmbiguous?: boolean;
|
||||
@@ -368,6 +371,12 @@ export function classifyNativeRunnerRecoveryEvidence(input: {
|
||||
if (input.unclaimedBootstrap === true && !input.hasCheckpoint && !input.hasProviderEvidence) {
|
||||
return { claimKind: "bootstrap_incomplete", reason: "remote_execution_never_claimed" };
|
||||
}
|
||||
if (input.remoteBootstrapProvenIncomplete === true && !input.hasCheckpoint && !input.hasProviderEvidence) {
|
||||
// Each same-run bootstrap is another provider attempt; keep restarts bounded.
|
||||
if ((input.providerAttempt ?? 0) >= 3)
|
||||
return { claimKind: null, reason: "execution_recovery_budget_exhausted" };
|
||||
return { claimKind: "bootstrap_incomplete", reason: "remote_bootstrap_proven_incomplete" };
|
||||
}
|
||||
return input.hasCheckpoint &&
|
||||
input.checkpointIdentityMatches === true &&
|
||||
input.hasProviderEvidence
|
||||
@@ -1017,9 +1026,23 @@ export async function claimNativeRestartRecoveries(input: {
|
||||
unclaimedBootstrap = !executionEvent;
|
||||
}
|
||||
|
||||
// A claimed remote bootstrap (runner launched, perhaps connected) has no
|
||||
// provider evidence when its controller dies before session.open
|
||||
// completes. The owner's lease has expired above, and the controller
|
||||
// writes every command to Postgres before sending it, so that checkpoint
|
||||
// is the complete record of what reached the sandbox runner. When it
|
||||
// proves no provider work, bootstrap again instead of holding the task.
|
||||
const remoteBootstrapProvenIncomplete = !unclaimedBootstrap && remoteWorkspace !== null && !pluginTask
|
||||
&& Boolean(durableAuthority) && stagedAttachment === null && !hasCheckpoint && !hasProviderEvidence
|
||||
&& Boolean(row.run.runnerInstanceId && row.run.nativeSessionId)
|
||||
&& await nativeRunnerCheckpointProvesIncompleteBootstrap(tx, {
|
||||
companyId: row.run.companyId, issueId: row.coordinator.issueId, runId: row.run.id,
|
||||
}, { runnerInstanceId: row.run.runnerInstanceId!, normalizedSessionId: row.run.nativeSessionId! });
|
||||
|
||||
const classification = classifyNativeRunnerRecoveryEvidence({
|
||||
remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority),
|
||||
unclaimedBootstrap,
|
||||
remoteBootstrapProvenIncomplete,
|
||||
runnerPidAlive,
|
||||
runnerGroupAlive,
|
||||
processStartMatches: exactRunnerIdentity,
|
||||
|
||||
@@ -5,6 +5,7 @@ import { inspectDurablePrpCheckpointHeader, inspectUnclaimedBootstrapCheckpoint,
|
||||
import { createLocalEncryptedCipher } from "../../secrets/local-encrypted-provider.js";
|
||||
import { legacyControllerBootId } from "../legacy-controller-lease.js";
|
||||
import { assertNativeSourceCheckpointConsumer, lockSourceRecoveryCoordinators } from "./native-source-recovery-ownership.js";
|
||||
import { runnerdStateIsIncompleteBootstrap } from "./native-bootstrap-proof.js";
|
||||
|
||||
type Binding = { companyId: string; issueId: string; runId: string };
|
||||
export const NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES = 1024 * 1024;
|
||||
@@ -248,6 +249,53 @@ export async function loadNativeUnclaimedBootstrapCheckpoint(
|
||||
} catch { return null; }
|
||||
}
|
||||
|
||||
const BOOTSTRAP_CHECKPOINT_PARTS = new Set(["header", "cursor", "command:1", "command:2"]);
|
||||
|
||||
/**
|
||||
* A remote runner's controller state lives only in this checkpoint, and every
|
||||
* command is written here before the controller sends it. Once the previous
|
||||
* controller's lease has expired, the checkpoint is therefore a complete record
|
||||
* of what reached the sandbox runner. Returns true only when that record holds
|
||||
* the bootstrap commands alone (run.prepare, an unfinished session.open) with
|
||||
* no committed or acknowledged event, so no provider turn can exist. The
|
||||
* sandbox runner journal is not on this host; the proof uses the controller
|
||||
* record alone, as for a local root whose runner never journaled.
|
||||
*/
|
||||
export async function nativeRunnerCheckpointProvesIncompleteBootstrap(
|
||||
db: Pick<Db, "select">, binding: Binding, expected: { runnerInstanceId: string; normalizedSessionId: string },
|
||||
): Promise<boolean> {
|
||||
try {
|
||||
const parts = await db.select({ part: nativeRunnerCheckpoints.part }).from(nativeRunnerCheckpoints).where(and(
|
||||
eq(nativeRunnerCheckpoints.companyId, binding.companyId), eq(nativeRunnerCheckpoints.issueId, binding.issueId),
|
||||
eq(nativeRunnerCheckpoints.runId, binding.runId))).limit(BOOTSTRAP_CHECKPOINT_PARTS.size + 1);
|
||||
if (!parts.length || parts.length > BOOTSTRAP_CHECKPOINT_PARTS.size
|
||||
|| parts.some(row => !BOOTSTRAP_CHECKPOINT_PARTS.has(row.part))) return false;
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
|
||||
if (!checkpoint || checkpoint.predecessor !== null
|
||||
|| checkpoint.recoveryEvidence.runnerInstanceId !== expected.runnerInstanceId
|
||||
|| checkpoint.recoveryEvidence.normalizedSessionId !== expected.normalizedSessionId) return false;
|
||||
// The decoder above validated these canonical v2 records. Rebuild the
|
||||
// controller state in memory; the part check admits no event records.
|
||||
const read = (part: string) => {
|
||||
const value = checkpoint.records.get(part);
|
||||
return value === undefined ? undefined : JSON.parse(value) as Record<string, unknown>;
|
||||
};
|
||||
const header = read("header");
|
||||
const cursor = read("cursor");
|
||||
if (header?.schema !== "paperclip.runner.controller-checkpoint.v2" || !cursor) return false;
|
||||
const state: Record<string, unknown> = {
|
||||
...(header.state as Record<string, unknown>),
|
||||
ackedSourceSeq: cursor.ackedSourceSeq,
|
||||
commands: ["command:1", "command:2"].map(read).filter(command => command !== undefined),
|
||||
committedEvents: [],
|
||||
};
|
||||
return (state.identity as { runId?: unknown } | undefined)?.runId === binding.runId
|
||||
&& runnerdStateIsIncompleteBootstrap(state);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
export async function restoreNativeRunnerCheckpoint(db: Db, binding: Binding, root: string, normalizedSessionId?: string) {
|
||||
const checkpoint = await loadNativeRunnerCheckpoint(db, binding);
|
||||
if (!checkpoint) return null;
|
||||
|
||||
+137
@@ -60,7 +60,10 @@ import {
|
||||
detachNativeSessionsForRestart,
|
||||
executePaperclipNativeSession,
|
||||
NativeControllerDetachedForRestartError,
|
||||
restoreRemoteBootstrapCheckpointForRecovery,
|
||||
runnerdStateProvesIncompleteBootstrap,
|
||||
} from "./native-session-executor.js";
|
||||
import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import {
|
||||
listExpiredNativeRunnerAuthorities,
|
||||
loadNativeRunnerCheckpoint,
|
||||
@@ -2220,6 +2223,140 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
||||
},
|
||||
);
|
||||
|
||||
// RC-29: a restart while a Daytona runner was connecting. The executor had
|
||||
// claimed the run and launched the sandbox runner, which authenticated, but
|
||||
// session.open never completed. The controller's Postgres checkpoint is the
|
||||
// only record of what the runner received.
|
||||
async function seedClaimedRemoteBootstrap(label: string, options: {
|
||||
attempt?: number;
|
||||
mutate?: (core: DurablePrpControlPlane) => void;
|
||||
checkpoint?: boolean;
|
||||
} = {}) {
|
||||
const fixture = await seedRun(`REMOTE-CLAIMED-${label}`);
|
||||
const { runId, issueId, native } = fixture;
|
||||
const remote = remotePreparedWorkspace();
|
||||
await fixture.db.update(heartbeatRuns).set({
|
||||
runnerProfileJson: { ...fixture.preparedProfile, nativeWorkspaceSync: remote },
|
||||
processPid: 443, processStartedAt: new Date(Date.now() - 30_000),
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
await fixture.db.insert(heartbeatRunEvents).values([
|
||||
{ companyId, runId, agentId, seq: 900, stream: "system", level: "info", eventType: "native.process_start_requested" },
|
||||
{ companyId, runId, agentId, seq: 901, stream: "system", level: "info", eventType: "native.process_identity_recorded",
|
||||
payload: { targetKind: "remote", processPid: 443 } },
|
||||
]);
|
||||
const bootId = randomUUID();
|
||||
const leaseOwner = `${bootId}:1:${randomUUID()}`;
|
||||
await fixture.db.update(nativeRunFinalizations).set({
|
||||
attempt: options.attempt ?? 1, controllerGeneration: 1, controllerBootId: bootId, controllerPid: 2_000_000_001,
|
||||
controllerProcessStartedAt: new Date(Date.now() - 60_000), leaseOwner, leaseExpiresAt: new Date(Date.now() + 60_000),
|
||||
}).where(eq(nativeRunFinalizations.runId, runId));
|
||||
if (options.checkpoint !== false) {
|
||||
const identity = { runnerInstanceId: native.runnerInstanceId, environmentLeaseId: runId, runId,
|
||||
normalizedSessionId: native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` };
|
||||
const core = new DurablePrpControlPlane({ stateDirectory: resolve(runtimeRoot, runId, "control-plane"), identity,
|
||||
expectedRunnerVersion: "0.3.0", expectedRunnerDigest: `sha256:${"a".repeat(64)}` });
|
||||
try {
|
||||
const template = { provider: { provider: "acpx", command: "fixture" }, authorizedTools: {} };
|
||||
core.persistRunAttachTemplate(template);
|
||||
core.queueCommand("run.prepare", template);
|
||||
core.queueCommand("session.open", { reuse: "same_session" });
|
||||
core.issueBootstrapTicket();
|
||||
// The runner authenticated once and received run.prepare.
|
||||
core.store.state.connectionCount = 1;
|
||||
core.store.state.freshBootstraps = 1;
|
||||
core.store.state.commandDeliveryCounts = { [core.store.state.commands[0]!.commandId]: 1 };
|
||||
options.mutate?.(core);
|
||||
core.setCheckpointPersistence(nativeRunnerCheckpointWriter(fixture.db, {
|
||||
companyId, issueId, runId, authorityRunId: runId, bootId, leaseOwner, generation: 1,
|
||||
}));
|
||||
await core.flushCheckpoint();
|
||||
} finally {
|
||||
await core.stop();
|
||||
}
|
||||
}
|
||||
// The killed controller can no longer renew its lease.
|
||||
await fixture.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1_000) })
|
||||
.where(eq(nativeRunFinalizations.runId, runId));
|
||||
return fixture;
|
||||
}
|
||||
|
||||
it("restarts a claimed remote bootstrap whose durable checkpoint proves no provider work", async () => {
|
||||
const fixture = await seedClaimedRemoteBootstrap("PROVEN");
|
||||
const dispositions = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId] });
|
||||
expect(dispositions).toEqual([expect.objectContaining({ kind: "bootstrap_incomplete", runId: fixture.runId,
|
||||
providerAttempt: 1, controllerGeneration: 2 })]);
|
||||
const [coordinator] = await fixture.db.select().from(nativeRunFinalizations)
|
||||
.where(eq(nativeRunFinalizations.runId, fixture.runId));
|
||||
expect(coordinator).toMatchObject({ phase: "observed", recoveryState: "bootstrap_incomplete", failureCode: null });
|
||||
expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ disposition: "bootstrap_incomplete",
|
||||
reason: "remote_bootstrap_proven_incomplete" });
|
||||
const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(run).toMatchObject({ status: "running", errorCode: null, processPid: null });
|
||||
expect(await fixture.db.select().from(issueRecoveryActions)
|
||||
.where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("lets the executor prove the remote bootstrap from its Postgres checkpoint, not the local marker", async () => {
|
||||
const fixture = await seedClaimedRemoteBootstrap("EXECUTOR");
|
||||
const execution = buildNativeExecutionInput({
|
||||
companyId, agentId, runId: fixture.runId,
|
||||
issue: { id: fixture.issueId, identifier: null, title: "Restart bootstrap", description: null, workMode: "standard" },
|
||||
taskPrompt: "Restart bootstrap", provider: "codex",
|
||||
workspace: { id: randomUUID(), cwd: runtimeRoot, repoUrl: null, repoRef: null, branchName: null },
|
||||
normalizedSessionId: fixture.native.normalizedSessionId,
|
||||
runtimeContext: nativeRuntimeContextFixture(),
|
||||
completionContract: { id: randomUUID(), sha256: `sha256:${"a".repeat(64)}`, schemaVersion: "paperclip.run-result.v1",
|
||||
contract: { revision: "1", objective: "Restart bootstrap", criteria: [{ id: "result", requirement: "Complete work" }] } },
|
||||
});
|
||||
const root = resolve(runtimeRoot, `marker-${fixture.runId}`);
|
||||
await mkdir(resolve(root, "control-plane"), { recursive: true, mode: 0o700 });
|
||||
const writeMarker = (runId: string) => writeFile(resolve(root, "control-plane", "control-plane-state.json"), JSON.stringify({
|
||||
schema: "paperclip.runner.external-control-plane-state.v1",
|
||||
identity: { runnerInstanceId: fixture.native.runnerInstanceId, environmentLeaseId: fixture.runId, runId,
|
||||
normalizedSessionId: fixture.native.normalizedSessionId, turnId: `turn-${runId}`, itemId: `item-${runId}` },
|
||||
}));
|
||||
const claim = { kind: "bootstrap_incomplete" as const, runId: fixture.runId, leaseOwner: "successor",
|
||||
controllerGeneration: 2, providerAttempt: 1, restartKind: "hard" as const, recoveryRequestId: null };
|
||||
const target = { kind: "remote", transport: "sandbox", remoteCwd: "/workspace" } as AdapterExecutionTarget;
|
||||
try {
|
||||
await writeMarker(fixture.runId);
|
||||
expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false);
|
||||
// Only a bootstrap recovery on a classic sandbox restores the checkpoint.
|
||||
for (const other of [{ restartRecovery: undefined, runnerExecutionTarget: target },
|
||||
{ restartRecovery: claim, runnerExecutionTarget: { ...target, runnerTask: vi.fn() } as AdapterExecutionTarget }]) {
|
||||
await restoreRemoteBootstrapCheckpointForRecovery({ db: fixture.db, execution, root, ...other });
|
||||
expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false);
|
||||
}
|
||||
// A marker for another run never receives this run's checkpoint.
|
||||
await writeMarker(randomUUID());
|
||||
await restoreRemoteBootstrapCheckpointForRecovery({ db: fixture.db, execution, root,
|
||||
restartRecovery: claim, runnerExecutionTarget: target });
|
||||
expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(false);
|
||||
await writeMarker(fixture.runId);
|
||||
await restoreRemoteBootstrapCheckpointForRecovery({ db: fixture.db, execution, root,
|
||||
restartRecovery: claim, runnerExecutionTarget: target });
|
||||
expect(runnerdStateProvesIncompleteBootstrap(root)).toBe(true);
|
||||
} finally {
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each<[string, Parameters<typeof seedClaimedRemoteBootstrap>[1], string]>([
|
||||
["a turn was issued", { mutate: core => { core.queueCommand("turn.start", { prompt: "Prior work must stay held." }); } },
|
||||
"ambiguous_remote_recovery_evidence"],
|
||||
["an event was acknowledged", { mutate: core => { core.store.state.ackedSourceSeq = 1; } },
|
||||
"ambiguous_remote_recovery_evidence"],
|
||||
["the state was adopted", { mutate: core => { core.store.state.bootstrapHistory!.adopted = true; } },
|
||||
"ambiguous_remote_recovery_evidence"],
|
||||
["no checkpoint exists", { checkpoint: false }, "ambiguous_remote_recovery_evidence"],
|
||||
["the bootstrap budget is spent", { attempt: 3 }, "execution_recovery_budget_exhausted"],
|
||||
])("keeps a claimed remote bootstrap held when %s", async (label, options, reason) => {
|
||||
const fixture = await seedClaimedRemoteBootstrap(label.replace(/\W+/g, "-"), options);
|
||||
expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor,
|
||||
restartKind: "hard", runIds: [fixture.runId] })).toEqual([{ kind: "blocked", runId: fixture.runId, reason }]);
|
||||
});
|
||||
|
||||
it.each([
|
||||
{ attempt: 2, driverKind: "codex_app_server" },
|
||||
{ attempt: 3, driverKind: "codex_app_server" },
|
||||
|
||||
@@ -85,6 +85,8 @@ import { createNativeToolTrace, type NativeToolTrace } from "./native-tool-trace
|
||||
import { createNativeProviderFailureObservation } from "./native-provider-failure.js";
|
||||
import { nativeStartupRetryExhaustedAction, nativeStartupRetryWait, planNativeStartupRetry } from "./native-startup-retry.js";
|
||||
import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js";
|
||||
import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS, runnerdStateIsIncompleteBootstrap,
|
||||
runnerdStateIsUnconnectedBootstrap } from "./native-bootstrap-proof.js";
|
||||
import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js";
|
||||
import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js";
|
||||
import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js";
|
||||
@@ -522,10 +524,6 @@ const ACPX_RUNTIME_CHECKPOINT_EXCLUDED_ENTRIES = [
|
||||
"home/.npm",
|
||||
"data/pnpm/store",
|
||||
] as const;
|
||||
const RUNNERD_CONTROL_PLANE_STATE_SCHEMAS = new Set([
|
||||
"paperclip.runner.durable.control-plane-state.v1",
|
||||
"paperclip.runner.durable.control-plane-state.v2",
|
||||
]);
|
||||
const RUNNERD_STATE_SCHEMA = "paperclip.runner.durable.state.v1";
|
||||
const CODEX_PROVIDER_STATE_SCHEMA = "paperclip.runner.codex-provider-state.v1";
|
||||
const ACPX_PROVIDER_STATE_SCHEMA = "paperclip.runner.acpx-provider-state.v3";
|
||||
@@ -5486,6 +5484,7 @@ async function migrateRunnerdStateRootForExecution(input: {
|
||||
if (input.restartRecovery?.kind === "reattach_remote_runner" && !existsSync(scoped)) {
|
||||
throw new Error("runner_state_identity_mismatch: remote_reattach_root_missing");
|
||||
}
|
||||
await restoreRemoteBootstrapCheckpointForRecovery({ ...input, root: scoped });
|
||||
if (existsSync(scoped)) {
|
||||
if (!isSafeNativeStateDirectory(scoped)) {
|
||||
throw new Error("runner_state_directory_unsafe");
|
||||
@@ -5952,55 +5951,6 @@ function localProcessDefinitelyGone(value: unknown, group = false): boolean {
|
||||
}
|
||||
}
|
||||
|
||||
const RUNNERD_BOOTSTRAP_COMMAND_TYPES = new Set(["run.prepare", "session.open"]);
|
||||
|
||||
/** The runner never connected, so every bootstrap command is still undelivered. */
|
||||
function runnerdStateIsUnconnectedBootstrap(state: Record<string, unknown>): boolean {
|
||||
const commands = Array.isArray(state.commands) ? state.commands.map(record) : [];
|
||||
const committedEvents = Array.isArray(state.committedEvents) ? state.committedEvents : [];
|
||||
return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) &&
|
||||
state.connectionCount === 0 && committedEvents.length === 0 &&
|
||||
commands.every(command => command.status === "pending" &&
|
||||
RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)));
|
||||
}
|
||||
|
||||
/**
|
||||
* A provider turn can start only after session.open completes and the
|
||||
* controller issues a turn command. A root whose controller journal stops
|
||||
* before that point (run.prepare may be acknowledged, session.open never is,
|
||||
* no event was committed or acknowledged) and whose runner journal emitted no
|
||||
* event either holds no provider work, so restart recovery may quarantine it
|
||||
* and bootstrap again. Any other shape stays fail-closed.
|
||||
*/
|
||||
function runnerdStateIsIncompleteBootstrap(
|
||||
state: Record<string, unknown>,
|
||||
runnerJournal?: Record<string, unknown> | null,
|
||||
): boolean {
|
||||
if (runnerdStateIsUnconnectedBootstrap(state)) return true;
|
||||
if (!RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema))) return false;
|
||||
if (typeof state.connectionCount !== "number" || !Number.isSafeInteger(state.connectionCount) ||
|
||||
state.connectionCount < 0) return false;
|
||||
if (!Array.isArray(state.committedEvents) || state.committedEvents.length !== 0) return false;
|
||||
if (state.ackedSourceSeq !== 0) return false;
|
||||
if (state.warmTransition != null || state.completedWarmTransition != null ||
|
||||
state.stagedRunAttachment != null) return false;
|
||||
if (record(state.bootstrapHistory).adopted === true) return false;
|
||||
if (!Array.isArray(state.commands)) return false;
|
||||
const commands = state.commands.map(record);
|
||||
if (!commands.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.type)) &&
|
||||
(command.status === "pending" ||
|
||||
(command.type === "run.prepare" && command.status === "completed")))) return false;
|
||||
// The controller cannot see runner events that were emitted but never
|
||||
// committed. When the runner journal exists it must agree that none were.
|
||||
if (runnerJournal === undefined) return true;
|
||||
if (runnerJournal === null) return false;
|
||||
const processed = Object.values(record(runnerJournal.processedCommands)).map(record);
|
||||
return Array.isArray(runnerJournal.outbox) && runnerJournal.outbox.length === 0 &&
|
||||
runnerJournal.nextSourceSeq === 1 && runnerJournal.ackedSourceSeq === 0 &&
|
||||
processed.every(command => RUNNERD_BOOTSTRAP_COMMAND_TYPES.has(String(command.commandType)) &&
|
||||
(command.commandType === "run.prepare" || command.status !== "completed"));
|
||||
}
|
||||
|
||||
/** `undefined` when the runner never journaled, `null` when unreadable. */
|
||||
function readRunnerdBootstrapJournal(root: string): Record<string, unknown> | null | undefined {
|
||||
const runnerRoot = resolve(root, "runner");
|
||||
@@ -6029,6 +5979,26 @@ function readRunnerdControlPlaneState(root: string): Record<string, unknown> {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* A remote runner's controller state lives in Postgres; its local root holds
|
||||
* only a fail-closed marker. Before a `bootstrap_incomplete` recovery checks
|
||||
* that root, restore the exact checkpoint so the same incomplete-bootstrap
|
||||
* proof applies. A missing checkpoint leaves the marker, which stays fail-closed.
|
||||
*/
|
||||
export async function restoreRemoteBootstrapCheckpointForRecovery(input: {
|
||||
db: Db; execution: NativeExecutionInput; root: string;
|
||||
restartRecovery?: NativeRestartRecoveryClaim;
|
||||
runnerExecutionTarget?: AdapterExecutionTarget | null;
|
||||
}): Promise<void> {
|
||||
const target = input.runnerExecutionTarget;
|
||||
if (input.restartRecovery?.kind !== "bootstrap_incomplete" || target?.kind !== "remote"
|
||||
|| target.transport !== "sandbox" || target.runnerTask || !existsSync(input.root)
|
||||
|| readRunnerdDurableIdentity(input.root) !== null) return;
|
||||
const marker = readRunnerdDurableIdentity(input.root, true);
|
||||
if (!marker || marker.runId !== input.execution.binding.runId) return;
|
||||
await restoreNativeRunnerCheckpoint(input.db, input.execution.binding, input.root, nativeSessionKey(input.execution));
|
||||
}
|
||||
|
||||
export function runnerdStateProvesIncompleteBootstrap(root: string): boolean {
|
||||
try {
|
||||
return runnerdStateIsIncompleteBootstrap(readRunnerdControlPlaneState(root), readRunnerdBootstrapJournal(root));
|
||||
|
||||
@@ -8,6 +8,7 @@ import { getNativeReviewAssignment } from "./native-review-participant.js";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
import { notifyStatusCardIssueChange } from "../status-card-notifications.js";
|
||||
import { normalizeAgentNameKey } from "../agent-name-key.js";
|
||||
import { NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES } from "./native-session-resumption-statuses.js";
|
||||
|
||||
export async function restoreNativeSessionExecution(db: Db, input: {
|
||||
issue: Pick<typeof issues.$inferSelect, "id" | "executionRunId">;
|
||||
@@ -48,7 +49,7 @@ export async function readNativeSessionResumptionAgent(db: Db, input: {
|
||||
const { issue, run, coordinator } = input;
|
||||
if (issue.companyId !== run.companyId || issue.id !== run.nativeIssueId ||
|
||||
coordinator.companyId !== run.companyId || coordinator.issueId !== issue.id || coordinator.runId !== run.id ||
|
||||
!["todo", "in_progress", "in_review"].includes(issue.status) ||
|
||||
!(NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES as readonly string[]).includes(issue.status) ||
|
||||
(issue.executionRunId && issue.executionRunId !== run.id) ||
|
||||
(issue.checkoutRunId && issue.checkoutRunId !== run.id)) return null;
|
||||
const [agent] = await db.select().from(agents).where(and(
|
||||
|
||||
@@ -0,0 +1,4 @@
|
||||
/** Task statuses from which a native same-run resume may be claimed. The task
|
||||
* interruption read model shares this set with resumption admission so it
|
||||
* never advertises a resume the scheduler would refuse. */
|
||||
export const NATIVE_SESSION_RESUMPTION_ISSUE_STATUSES = ["todo", "in_progress", "in_review"] as const;
|
||||
@@ -216,10 +216,15 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i
|
||||
or ${heartbeatRuns.errorCode} = 'workspace_validation_failed' then ${heartbeatRuns.error} else null end`,
|
||||
errorCode: heartbeatRuns.errorCode, livenessState: heartbeatRuns.livenessState, scheduledRetryAt: heartbeatRuns.scheduledRetryAt,
|
||||
scheduledRetryReason: heartbeatRuns.scheduledRetryReason,
|
||||
// Candidate only: the coordinator also pushes nextAttemptAt forward while
|
||||
// it defers an ineligible run. The scheduler's own admission check below
|
||||
// decides whether this resume is real.
|
||||
nativeResumeAt: sql<Date | null>`(select ${nativeRunFinalizations.nextAttemptAt} from ${nativeRunFinalizations}
|
||||
where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id}
|
||||
and ${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'failed'
|
||||
and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure')`.mapWith(value => value == null ? null : new Date(value)),
|
||||
and ${nativeRunFinalizations.resultId} is null and ${nativeRunFinalizations.phase} = 'retryable_failure'
|
||||
and ${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null
|
||||
and ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null)`.mapWith(value => value == null ? null : new Date(value)),
|
||||
// The run whose resultJson supplied providerFailure. A scheduled retry
|
||||
// shows its predecessor's reason, so redaction must use that run's registry.
|
||||
providerFailureRunId: sql<string>`case when ${heartbeatRuns.resultJson}->'providerFailure' is null
|
||||
@@ -279,6 +284,38 @@ export async function taskInterruptionPresentations(db: Db, companyId: string, i
|
||||
inArray(approvals.status, ["pending", "revision_requested"]))),
|
||||
db.select({ status: companies.status, pauseReason: companies.pauseReason }).from(companies).where(eq(companies.id, companyId)),
|
||||
]);
|
||||
// Advertise a pending resume only when the scheduler would admit it, using
|
||||
// its own check, so the task never hides Retry behind a deferred candidate
|
||||
// and never offers Retry beside a resume that will run. Rare rows only.
|
||||
// Candidates are failed native runs inside an active coordinator retry, so
|
||||
// there are few. Their rows load in three batched reads; admission itself
|
||||
// runs a few bounded reads per candidate, four at a time.
|
||||
const resumeCandidates = runs.filter(run => run.nativeResumeAt);
|
||||
if (resumeCandidates.length) {
|
||||
const { readNativeSessionResumptionAgent } = await import("./native-runtime/native-session-resumption-admission.js");
|
||||
const candidateIds = resumeCandidates.map(run => run.id);
|
||||
const [fullRuns, coordinators] = await Promise.all([
|
||||
db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.id, candidateIds))),
|
||||
db.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, companyId),
|
||||
inArray(nativeRunFinalizations.runId, candidateIds))),
|
||||
]);
|
||||
const issueIds = [...new Set(fullRuns.flatMap(run => run.nativeIssueId ? [run.nativeIssueId] : []))];
|
||||
const issueRows = issueIds.length ? await db.select().from(issues).where(and(eq(issues.companyId, companyId),
|
||||
inArray(issues.id, issueIds))) : [];
|
||||
const runById = new Map(fullRuns.map(run => [run.id, run]));
|
||||
const coordinatorByRun = new Map(coordinators.map(row => [row.runId, row]));
|
||||
const issueById = new Map(issueRows.map(row => [row.id, row]));
|
||||
for (let start = 0; start < resumeCandidates.length; start += 4) {
|
||||
await Promise.all(resumeCandidates.slice(start, start + 4).map(async candidate => {
|
||||
const run = runById.get(candidate.id);
|
||||
const issue = run?.nativeIssueId ? issueById.get(run.nativeIssueId) : undefined;
|
||||
const coordinator = coordinatorByRun.get(candidate.id);
|
||||
const admitted = run && issue && coordinator
|
||||
&& await readNativeSessionResumptionAgent(db, { issue, run, coordinator }).catch(() => null);
|
||||
if (!admitted) candidate.nativeResumeAt = null;
|
||||
}));
|
||||
}
|
||||
}
|
||||
const runIds = runs.map(run => run.id);
|
||||
const actions = runIds.length ? await db.selectDistinctOn([issueRecoveryActions.sourceIssueId]).from(issueRecoveryActions).where(and(
|
||||
eq(issueRecoveryActions.companyId, companyId), inArray(issueRecoveryActions.sourceIssueId, ids),
|
||||
|
||||
@@ -94,8 +94,7 @@ export async function inspectReliability(api) {
|
||||
if (candidates.length !== 1 || !positive(candidates[0].job.id)) fail('standing_job');
|
||||
const job = await api(`/api/job/${candidates[0].job.id}`);
|
||||
if (job?.id !== candidates[0].job.id || job.type !== 'github' || !workflowUrls.has(job.url)) fail('standing_job');
|
||||
const assignments = plan ? await api(`/api/testplan/${plan.id}/job`) : [];
|
||||
if (!Array.isArray(assignments)) fail('plan_job');
|
||||
const assignments = plan ? jobAssignments(await api(`/api/testplan/${plan.id}/job`)) : [];
|
||||
return { caseIds: Object.fromEntries(found), plan, jobId: job.id,
|
||||
standingParameters: Array.isArray(candidates[0].job.parameters) ? candidates[0].job.parameters : [], assignments };
|
||||
}
|
||||
@@ -134,8 +133,8 @@ export async function registerReliability(api) {
|
||||
plan = await api(`/api/testplan/${plan.id}`);
|
||||
}
|
||||
assertPlan(plan, ids);
|
||||
let assignments = await api(`/api/testplan/${plan.id}/job`);
|
||||
if (!Array.isArray(assignments) || assignments.length > 1 || assignments.some(row => row?.job?.id !== before.jobId)) fail('plan_job_conflict');
|
||||
let assignments = jobAssignments(await api(`/api/testplan/${plan.id}/job`));
|
||||
if (assignments.length > 1 || assignments.some(row => row?.job?.id !== before.jobId)) fail('plan_job_conflict');
|
||||
if (!assignments.length) {
|
||||
await api(`/api/testplan/${plan.id}/assign`, 'POST', { jobId: before.jobId, selection: { inverted: false, manual: false, leafsInclude: ids } });
|
||||
created.assignment = true;
|
||||
@@ -145,13 +144,21 @@ export async function registerReliability(api) {
|
||||
await api(`/api/testplan/${plan.id}/jobparameter`, 'POST', { jobId: before.jobId, parameters: before.standingParameters });
|
||||
created.parameters = true;
|
||||
}
|
||||
assignments = await api(`/api/testplan/${plan.id}/job`);
|
||||
if (!Array.isArray(assignments) || assignments.length !== 1 || assignments[0]?.job?.id !== before.jobId
|
||||
assignments = jobAssignments(await api(`/api/testplan/${plan.id}/job`));
|
||||
if (assignments.length !== 1 || assignments[0]?.job?.id !== before.jobId
|
||||
|| assignments[0].testCasesCount !== ids.length
|
||||
|| (before.standingParameters.length && !same(assignments[0].job.parameters ?? [], before.standingParameters))) fail('plan_job');
|
||||
return { caseIds, plan, jobId: before.jobId, created };
|
||||
}
|
||||
|
||||
// TestOps lists a plan's cases that no job covers as a job-less row, e.g.
|
||||
// `[{ testCasesCount: 7 }]` for a freshly created plan. That row is not a job
|
||||
// assignment; only rows naming a job are.
|
||||
function jobAssignments(rows) {
|
||||
if (!Array.isArray(rows)) fail('plan_job');
|
||||
return rows.filter(row => row?.job != null);
|
||||
}
|
||||
|
||||
export function registrationReceipt(operation, state, env) {
|
||||
const caseIds = Object.fromEntries(reliabilityDefinitions.filter(row => state.caseIds[row.name]).map(row => [row.name, state.caseIds[row.name]]));
|
||||
const complete = Object.keys(caseIds).length === reliabilityDefinitions.length && positive(state.plan?.id)
|
||||
|
||||
@@ -196,6 +196,8 @@ function fakeTestOps({ existing = [], plan = null, assignments = [], standing =
|
||||
if (method === 'GET' && url.pathname === '/api/testplan') return { content: state.plan ? [{ id: state.plan.id, name: state.plan.name }] : [], totalPages: 1 };
|
||||
if (method === 'POST' && url.pathname === '/api/testplan') {
|
||||
state.plan = { id: 77, projectId: 1, name: body.name, treeSelection: body.treeSelection, testCasesCount: body.treeSelection.leafsInclude.length };
|
||||
// Live TestOps lists a new plan's uncovered cases as one job-less row.
|
||||
state.assignments = [{ testCasesCount: body.treeSelection.leafsInclude.length }];
|
||||
return { id: 77 };
|
||||
}
|
||||
if (method === 'GET' && url.pathname === '/api/testplan/77') return state.plan;
|
||||
@@ -293,6 +295,20 @@ test('registration refuses a standing plan without exactly one trusted job or a
|
||||
await assert.rejects(registerReliability(api), /plan_job_conflict/);
|
||||
});
|
||||
|
||||
test('a plan TestOps reports only as a job-less row is unassigned, so registration assigns the standing job', async () => {
|
||||
const { api, state } = fakeTestOps();
|
||||
await registerReliability(api);
|
||||
// Reproduces staging plan 135: created, then the run stopped before assignment.
|
||||
state.assignments = [{ testCasesCount: 7 }];
|
||||
const inspected = await inspectReliability(api);
|
||||
assert.deepEqual(inspected.assignments, []);
|
||||
assert.equal(registrationReceipt('inspect', inspected, env).complete, false);
|
||||
const repaired = await registerReliability(api);
|
||||
assert.equal(repaired.created.assignment, true);
|
||||
assert.deepEqual(state.assignments.map(row => row.job?.id), [50]);
|
||||
assert.equal(registrationReceipt('register', repaired, { GITHUB_RUN_ID: '1', GITHUB_SHA: workflowSha }).complete, true);
|
||||
});
|
||||
|
||||
test('the workflows expose the reliability plan without new provider or reporting authority', async () => {
|
||||
const full = parse(await readFile(new URL('../../.github/workflows/runner-full-stack-e2e.yml', import.meta.url), 'utf8'));
|
||||
assert.deepEqual(full.on.workflow_dispatch.inputs.testops_plan.options, ['product-regression', 'reliability']);
|
||||
|
||||
@@ -16,6 +16,7 @@ async function resolveTarget(event: string, cadence: string, branch = "master")
|
||||
const output = path.join(directory, "output");
|
||||
const result = spawnSync("bash", ["-c", `gh() { printf '%s\\n' "$MOCK_BRANCH_SHA"; }\n${script}`], {
|
||||
encoding: "utf8", env: { ...process.env, EVENT_NAME: event, EVAL_CADENCE: cadence,
|
||||
IMAGE_ONLY: "false", TESTOPS_REQUESTED: "false",
|
||||
TARGET_BRANCH: branch, DEFAULT_BRANCH: "master", GITHUB_SHA: queuedSha,
|
||||
MOCK_BRANCH_SHA: advancedSha, GITHUB_OUTPUT: output, REPOSITORY: "paperclipai/paperclip" },
|
||||
});
|
||||
|
||||
Reference in new issue
Block a user