diff --git a/src/common/schemas/project.ts b/src/common/schemas/project.ts index 646b50d75d3..38479bd8eff 100644 --- a/src/common/schemas/project.ts +++ b/src/common/schemas/project.ts @@ -301,6 +301,10 @@ export const WorkspaceConfigSchema = z.object({ description: "Monotonic workflow claim that retired this attempt for replacement. Never cleared; every later admission of the task refuses while it is set.", }), + taskTerminalFailure: z.object({ attemptId: z.string(), errorType: z.string() }).optional().meta({ + description: + "The attempt a terminal stream failure (e.g. model_refusal) ended, written with its interrupted status. Applies only while taskAttemptId still names that attempt.", + }), taskAttentionPolicy: BackgroundWorkAttentionPolicySchema.optional().meta({ description: "How the owner workspace's stream-end treats this child task while it is active. " + diff --git a/src/node/config.test.ts b/src/node/config.test.ts index 193f8a95851..c3d888fb976 100644 --- a/src/node/config.test.ts +++ b/src/node/config.test.ts @@ -3317,6 +3317,37 @@ describe("Config", () => { } ); + it("keeps a task's terminal-failure marker through reload and a metadata write", async () => { + const projectPath = path.join(tempDir, "project"); + const marker = { attemptId: "att_00000000000000a1", errorType: "model_refusal" }; + await config.editConfig((cfg) => { + cfg.projects.set(projectPath, { + workspaces: [ + { + id: "child", + name: "child", + path: projectPath, + createdAt: "2025-01-01T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "owner", + taskStatus: "interrupted", + taskAttemptId: marker.attemptId, + taskTerminalFailure: marker, + }, + ], + }); + return cfg; + }); + const reloaded = new Config(tempDir); + const row = () => + new Config(tempDir).loadConfigOrDefault().projects.get(projectPath)?.workspaces[0]; + expect(row()?.taskTerminalFailure).toEqual(marker); + // Server-owned like the attempt identity: absent from metadata, kept by a metadata write. + const [metadata] = await reloaded.getAllWorkspaceMetadata(); + await reloaded.addWorkspace(projectPath, { ...metadata, title: "Renamed" }); + expect(row()).toMatchObject({ title: "Renamed", taskTerminalFailure: marker }); + }); + it("defaults sparse persisted heartbeat intervals in workspace metadata", async () => { const projectPath = "/fake/project"; const workspacePath = path.join(config.srcDir, "project", "heartbeat-sparse"); diff --git a/src/node/config/index.ts b/src/node/config/index.ts index 8dbe10d2431..8c70948f479 100644 --- a/src/node/config/index.ts +++ b/src/node/config/index.ts @@ -4209,6 +4209,7 @@ export class Config { taskAttemptId: existing.taskAttemptId, taskAttemptUnproven: existing.taskAttemptUnproven, taskAttemptRetiredBy: existing.taskAttemptRetiredBy, + taskTerminalFailure: existing.taskTerminalFailure, }; } else { // Add new workspace diff --git a/src/node/services/taskService.claimedRetirement.test.ts b/src/node/services/taskService.claimedRetirement.test.ts index 0e50ba53476..1ef9df6ef52 100644 --- a/src/node/services/taskService.claimedRetirement.test.ts +++ b/src/node/services/taskService.claimedRetirement.test.ts @@ -198,6 +198,71 @@ describe("TaskService claimed retirement (G2 PR B)", () => { expect(await read(otherProcess(config), "absent")).toEqual(noRecord); }); + test("the row's terminal-failure marker counts only for the attempt it names", async () => { + const config = await setupChild("marked", { + taskLaunchError: "refused by the model", + taskTerminalFailure: { attemptId: ATTEMPT, errorType: "model_refusal" }, + }); + await writeReceipt(config, midId, "marked"); + // No failure artifact (its write failed): the marker still makes it a failure. + expect(await read(otherProcess(config), "marked")).toEqual({ + kind: "terminal-no-report", + attemptId: ATTEMPT, + failure: { errorMessage: "refused by the model" }, + }); + + // A marker left by an earlier attempt: the current attempt's no-report stays replaceable. + const stale = await setupChild("stalemarker", { + taskLaunchError: "refused by the model", + taskTerminalFailure: { attemptId: "att_00000000000000c0", errorType: "model_refusal" }, + }); + await writeReceipt(stale, midId, "stalemarker"); + expect(await read(otherProcess(stale), "stalemarker")).toEqual({ + kind: "terminal-no-report", + attemptId: ATTEMPT, + }); + }); + + test.each([ + ["malformed", "file"], + ["a directory", "dir"], + ] as const)( + "an unreadable failure artifact (%s) defers to the row's marker for the attempt it names", + async (_label, damage) => { + const marker = (attemptId: string) => ({ + taskLaunchError: "refused by the model", + taskTerminalFailure: { attemptId, errorType: "model_refusal" }, + }); + const cases = [ + ["matching", marker(ATTEMPT)], + ["stale", marker("att_00000000000000c0")], + ["absent", {}], + ] as const; + for (const [name, overrides] of cases) { + const taskId = `unread${name}`; + const config = await setupChild(taskId, overrides); + await writeReceipt(config, midId, taskId); + const failures = getSubagentFailureArtifactsFilePath( + path.join(config.sessionsDir, midId) + ); + await fsPromises.rm(failures, { recursive: true, force: true }); + if (damage === "dir") await fsPromises.mkdir(failures, { recursive: true }); + else await fsPromises.writeFile(failures, "{ not json", "utf-8"); + const outcome = await read(otherProcess(config), taskId); + if (name === "matching") { + expect(outcome).toEqual({ + kind: "terminal-no-report", + attemptId: ATTEMPT, + failure: { errorMessage: "refused by the model" }, + }); + } else { + // Fail closed: nothing proves this attempt failed, or that it merely ended. + expect(outcome.kind).toBe("indeterminate"); + } + } + } + ); + test.each([ ["only an ancestor holds the receipt", {}, rootId, ATTEMPT], ["the receipt names an older attempt", {}, midId, "att_00000000000000c0"], diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 0a093d9686d..ba1ef704664 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -1060,6 +1060,22 @@ function rowSupersedes( return expectedAttemptId != null && row?.taskAttemptId !== expectedAttemptId; } +/** + * The terminal failure the row's marker (#4579) records for `attemptId`, or undefined when the + * marker is absent or names another attempt (a later attempt ignores an earlier failure). + */ +function markedTerminalFailure( + row: WorkspaceConfigEntry | undefined, + attemptId: string | undefined +): { errorMessage: string } | undefined { + const marker = row?.taskTerminalFailure; + if (attemptId == null || marker?.attemptId !== attemptId) return undefined; + return { + errorMessage: + coerceNonEmptyString(row?.taskLaunchError) ?? `Task failed terminally (${marker.errorType})`, + }; +} + /** * Why a workflow claim on `attemptId` must be refused (see TaskService.claimRetiredAttempt), or * undefined when it may be granted (or re-stamped, for the same run/step before any replacement). @@ -3881,6 +3897,23 @@ export class TaskService implements AgentTaskIntegration { taskId ); if (failureRead.kind === "unreadable") { + // The row's marker (#4579) still proves a terminal failure for the attempt it names; + // without one, a damaged artifact proves nothing (fail closed). + let row: WorkspaceConfigEntry | undefined; + try { + row = findWorkspaceEntry( + this.config.loadConfigOrDefault({ throwOnError: true }), + taskId + )?.workspace; + } catch { + row = undefined; + } + const owned = this.ownedAttemptByTaskId.get(taskId); + const attemptId = row?.taskAttemptId; + const marked = markedTerminalFailure(row, attemptId); + if (marked != null && (owned?.attemptId == null || owned.attemptId === attemptId)) { + return { kind: "terminal-no-report", attemptId, failure: marked }; + } return indeterminate( `failure artifact unreadable in ${reportOwnerWorkspaceId}: ${failureRead.error}` ); @@ -3893,6 +3926,12 @@ export class TaskService implements AgentTaskIntegration { entry = findWorkspaceEntry(this.config.loadConfigOrDefault(), taskId); } + // The row's terminal-failure marker (#4579) counts only for the attempt it names, so an + // artifact write that failed still fails the step, and a later attempt's no-report is not. + const failureFields = (attemptId: string | undefined) => { + const marked = failure ?? markedTerminalFailure(entry?.workspace, attemptId); + return marked != null ? { failure: marked } : {}; + }; // Owned cleanup in flight (Layer 2 latch): its release is the guaranteed settlement signal. if (this.isWorkspaceStopInProgress(taskId)) { return { kind: "cleanup-pending" }; @@ -3928,7 +3967,7 @@ export class TaskService implements AgentTaskIntegration { return { kind: "terminal-no-report", attemptId: proof.attemptId, - ...(failure != null ? { failure } : {}), + ...failureFields(proof.attemptId), }; } return indeterminate( @@ -3961,7 +4000,7 @@ export class TaskService implements AgentTaskIntegration { return { kind: "terminal-no-report", ...(owned.attemptId != null ? { attemptId: owned.attemptId } : {}), - ...(failure != null ? { failure } : {}), + ...failureFields(owned.attemptId), // Ended before (or without) a published row, e.g. a canceled reservation. ...(entry == null && this.taskRowPositivelyAbsent(taskId) ? { code: "no-record" as const } @@ -16513,6 +16552,11 @@ export class TaskService implements AgentTaskIntegration { parentWorkspaceId = ws.parentWorkspaceId; ws.taskStatus = "interrupted"; ws.taskLaunchError = failure.errorMessage; + // #4579: binds the failure to the attempt this CAS matched. The artifacts below are + // log-only on I/O errors, and taskLaunchError is not attempt-bound or failure-only. + if (ws.taskAttemptId != null) { + ws.taskTerminalFailure = { attemptId: ws.taskAttemptId, errorType: failure.errorType }; + } if (stopRecord == null) { this.closeAttemptAdmission( workspaceId, diff --git a/src/node/services/workflows/WorkflowRunner.replacementRestart.test.ts b/src/node/services/workflows/WorkflowRunner.replacementRestart.test.ts index 7cb5270ef1a..efb7c1c46b9 100644 --- a/src/node/services/workflows/WorkflowRunner.replacementRestart.test.ts +++ b/src/node/services/workflows/WorkflowRunner.replacementRestart.test.ts @@ -95,6 +95,31 @@ describe("workflow step replacement across a backend restart (G2)", () => { expect(resumed.priorRetiredBy).toBeUndefined(); }, 60_000); + test.each([ + ["was lost", "refused-artifact-lost", false], + ["stays unreadable", "refused-artifact-unreadable", true], + ] as const)( + "a terminal failure whose artifact %s still fails the step, never replaced", + async (_label, variant, artifactPresent) => { + const ended = await runFixture(["end", root.path, variant]); + expect(ended).toMatchObject({ + artifactPresent, + row: { taskStatus: "interrupted", taskTerminalFailure: { errorType: "model_refusal" } }, + }); + + const resumed = await runFixture(["resume", root.path]); + expect(String(resumed.error)).toContain("fixture: the model refused"); + expect(resumed).toMatchObject({ + runStatus: "failed", + children: ["priorchild01"], + journal: "priorchild01", + journalStatus: "failed", + }); + expect(resumed.priorRetiredBy).toBeUndefined(); + }, + 60_000 + ); + test("a child that reported is never replaced: the next process uses its report", async () => { await runFixture(["end", root.path, "reported"]); diff --git a/src/node/services/workflows/replacementRestart.testHarness.ts b/src/node/services/workflows/replacementRestart.testHarness.ts index 842e4bfb9bd..4a8191602a6 100644 --- a/src/node/services/workflows/replacementRestart.testHarness.ts +++ b/src/node/services/workflows/replacementRestart.testHarness.ts @@ -13,6 +13,7 @@ import * as path from "node:path"; import { spyOn } from "bun:test"; import { Config, type Workspace as WorkspaceConfigEntry } from "@/node/config"; import { QuickJSRuntimeFactory } from "@/node/services/ptc/quickjsRuntime"; +import { getSubagentFailureArtifactsFilePath } from "@/node/services/subagentFailureArtifacts"; import { upsertSubagentReportArtifact } from "@/node/services/subagentReportArtifacts"; import type { TaskService } from "@/node/services/taskService"; import { @@ -67,7 +68,14 @@ interface TerminalFailureInternals { async function end( root: string, - outcome: "reported" | "no-report" | "refused" | "failed-checkpoint" | "refused-failed-checkpoint" + outcome: + | "reported" + | "no-report" + | "refused" + | "failed-checkpoint" + | "refused-failed-checkpoint" + | "refused-artifact-lost" + | "refused-artifact-unreadable" ) { const config = new Config(root); await fs.mkdir(config.srcDir, { recursive: true }); @@ -142,16 +150,28 @@ async function end( completedAt: new Date().toISOString(), }); } - } else if (outcome === "refused" || outcome === "refused-failed-checkpoint") { + } else if ( + outcome === "refused" || + outcome === "refused-failed-checkpoint" || + outcome === "refused-artifact-lost" || + outcome === "refused-artifact-unreadable" + ) { // The real terminal-failure path: interrupted row, settlement receipt, failure artifact. const row = findWorkspaceInConfig(config, childId); if (row == null) throw new Error("reserved child row missing"); + // "refused-artifact-lost": the parent's failure-artifact write fails (a directory squats on + // its path; the writer only logs), and the squatter is gone before the next process reads. + const artifactPath = getSubagentFailureArtifactsFilePath(sessionDir(config)); + // "refused-artifact-unreadable": the squatter stays, so the next process cannot read it. + const squat = outcome === "refused-artifact-lost" || outcome === "refused-artifact-unreadable"; + if (squat) await fs.mkdir(artifactPath, { recursive: true }); await (taskService as unknown as TerminalFailureInternals).failAgentTaskTerminally( childId, { projectPath, workspace: row }, FIXTURE_REFUSAL, { expectedAttemptId: row.taskAttemptId ?? null } ); + if (outcome === "refused-artifact-lost") await fs.rmdir(artifactPath); if (outcome === "refused-failed-checkpoint") { // What the child's own runner records: the step failed with the refusal, the run failed. await store.recordStepFailed(FIXTURE_RUN_ID, { @@ -174,7 +194,13 @@ async function end( reportMarkdown: "the prior child's report", }); } - return { childId, row: findWorkspaceInConfig(config, childId) }; + const artifactPresent = await fs + .access(getSubagentFailureArtifactsFilePath(sessionDir(config))) + .then( + () => true, + () => false + ); + return { childId, row: findWorkspaceInConfig(config, childId), artifactPresent }; } async function resume(root: string, retryFromFailedCheckpoint: boolean) { @@ -241,7 +267,9 @@ try { outcome === "reported" || outcome === "refused" || outcome === "failed-checkpoint" || - outcome === "refused-failed-checkpoint" + outcome === "refused-failed-checkpoint" || + outcome === "refused-artifact-lost" || + outcome === "refused-artifact-unreadable" ? outcome : "no-report" )