diff --git a/src/agent-runtime/procedural-current-lifecycle.ts b/src/agent-runtime/procedural-current-lifecycle.ts index 3390a577e..9ac8cf522 100644 --- a/src/agent-runtime/procedural-current-lifecycle.ts +++ b/src/agent-runtime/procedural-current-lifecycle.ts @@ -26,6 +26,7 @@ const TERMINAL_WORKFLOW_TASK_STATES: ReadonlySet = new "blocked", ]); const AUTHORITY_ID_PATTERN = /^[\x21-\x7e]{1,128}$/u; +const MAX_CURRENT_WORKFLOW_RESPONSE_BYTES = 1024 * 1024; type CurrentWorkflowTaskState = "pending" | "running" | "succeeded" | "failed" | "cancelled" | "blocked"; @@ -120,6 +121,53 @@ function currentWorkflowEvidence( }); } +/** + * Reads one private Workflow / Task Execution response into fixed retained storage before JSON admission. + * The byte ceiling is deliberately much larger than the bounded workflow snapshot schema but prevents a + * corrupt internal response from making Agent Runtime retain an unbounded body before fail-closed validation. + * Oversize streams are cancelled before any over-limit chunk is copied; cancellation failure cannot replace + * the stable `invalid_workflow_state_response` diagnostic, and the reader lock is always released. + */ +async function boundedCurrentWorkflowResponse(response: Response): Promise { + if (response.body === null) { + return rejectCurrentLifecycle("invalid_workflow_state_response"); + } + + const reader = response.body.getReader(); + const storage = new Uint8Array(MAX_CURRENT_WORKFLOW_RESPONSE_BYTES); + let totalBytes = 0; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) break; + if (!(value instanceof Uint8Array)) { + return rejectCurrentLifecycle("invalid_workflow_state_response"); + } + if (value.byteLength > MAX_CURRENT_WORKFLOW_RESPONSE_BYTES - totalBytes) { + try { + await reader.cancel("Noema current workflow-state response exceeded byte ceiling"); + } finally { + return rejectCurrentLifecycle("invalid_workflow_state_response"); + } + } + storage.set(value, totalBytes); + totalBytes += value.byteLength; + } + } catch (error) { + if (error instanceof ProceduralCurrentLifecycleError) throw error; + return rejectCurrentLifecycle("invalid_workflow_state_response"); + } finally { + reader.releaseLock(); + } + + try { + const text = new TextDecoder("utf-8", { fatal: true, ignoreBOM: false }).decode(storage.subarray(0, totalBytes)); + return JSON.parse(text) as unknown; + } catch { + return rejectCurrentLifecycle("invalid_workflow_state_response"); + } +} + async function readCurrentWorkflowEvidence( env: WorkflowStateDurableObjectEnv, plan: AdmittedWorkflowTaskPlan, @@ -135,12 +183,7 @@ async function readCurrentWorkflowEvidence( if (response.status === 503) return rejectCurrentLifecycle("workflow_state_unavailable"); if (response.status !== 200) return rejectCurrentLifecycle("invalid_workflow_state_response"); - let body: unknown; - try { - body = await response.json(); - } catch { - return rejectCurrentLifecycle("invalid_workflow_state_response"); - } + const body = await boundedCurrentWorkflowResponse(response); if (!isRecord(body) || body.ok !== true) { return rejectCurrentLifecycle("invalid_workflow_state_response"); } @@ -202,4 +245,4 @@ export async function guideProceduralExecutionFromCurrentWorkflowState( state: proceduralGateState(evidence), }); return guideProceduralExecution(lifecycle, session, request); -} \ No newline at end of file +} diff --git a/test/procedural-current-lifecycle-response-bounds.test.ts b/test/procedural-current-lifecycle-response-bounds.test.ts new file mode 100644 index 000000000..6998e3de4 --- /dev/null +++ b/test/procedural-current-lifecycle-response-bounds.test.ts @@ -0,0 +1,126 @@ +import { describe, expect, it } from "vitest"; + +import { createProceduralGraph, startProceduralSession } from "../src/agent-runtime/procedural-graph"; +import { + guideProceduralExecutionFromCurrentWorkflowState, +} from "../src/agent-runtime/procedural-current-lifecycle"; +import type { WorkflowStateDurableObjectEnv } from "../src/workflow-task-execution/workflow-state-durable-object"; +import type { WorkflowTaskPlan } from "../src/workflow-task-execution/task-plan"; + +const executionId = "run-current-lifecycle-response-bound"; + +function plan(): WorkflowTaskPlan { + return { + executionId, + planId: "plan-current-lifecycle-response-bound", + maxConcurrency: 1, + tasks: [{ taskId: "review", dependsOn: [], effect: "pure" }], + }; +} + +async function session() { + const graph = await createProceduralGraph({ + schemaVersion: "noema.procedural-graph/v1", + tenantId: "tenant-a", + taskType: "pr-repair", + graphId: "current-lifecycle-response-bound", + revision: 1, + parentDigest: null, + nodes: ["Start", "review"], + edges: [{ + from: "Start", + relation: "leads_to", + to: "review", + condition: "", + guidance: "Use only bounded current durable evidence", + pitfalls: "Do not buffer unbounded internal responses", + }], + }); + return startProceduralSession(graph, { + tenantId: "tenant-a", + taskType: "pr-repair", + executionId, + graphDigest: graph.digest, + }); +} + +function envFor(response: Response): WorkflowStateDurableObjectEnv { + const namespace = { + idFromName(name: string): DurableObjectId { + return { name, toString: () => name } as unknown as DurableObjectId; + }, + get(_id: DurableObjectId): DurableObjectStub { + return { fetch: async () => response } as unknown as DurableObjectStub; + }, + }; + return { NOEMA_WORKFLOW_STATE: namespace as unknown as DurableObjectNamespace }; +} + +async function expectInvalid(response: Response): Promise { + await expect(guideProceduralExecutionFromCurrentWorkflowState( + envFor(response), + plan(), + await session(), + { lastProcedure: null, hops: 1, maxEdges: 4 }, + )).rejects.toMatchObject({ + name: "ProceduralCurrentLifecycleError", + code: "invalid_workflow_state_response", + }); +} + +describe("procedural current-lifecycle response bounds", () => { + it("cancels an oversized durable-owner stream before consuming later chunks", async () => { + const chunks = [ + new Uint8Array(700 * 1024).fill(0x20), + new Uint8Array(400 * 1024).fill(0x20), + new TextEncoder().encode("{}"), + ]; + let nextChunk = 0; + let cancelled = false; + const stream = new ReadableStream({ + pull(controller) { + if (nextChunk >= chunks.length) { + controller.close(); + return; + } + controller.enqueue(chunks[nextChunk]!); + nextChunk += 1; + }, + cancel() { + cancelled = true; + throw new Error("cleanup transport failed"); + }, + }, { highWaterMark: 0 }); + + await expectInvalid(new Response(stream, { + status: 200, + headers: { "content-type": "application/json; charset=utf-8" }, + })); + + expect(cancelled).toBe(true); + expect(nextChunk).toBe(2); + }); + + it("rejects a successful status with no response body", async () => { + await expectInvalid(new Response(null, { status: 200 })); + }); + + it("fails closed on a malformed non-byte response chunk", async () => { + const stream = new ReadableStream({ + start(controller) { + controller.enqueue("not-bytes" as unknown as Uint8Array); + controller.close(); + }, + }); + await expectInvalid(new Response(stream, { status: 200 })); + }); + + it("normalizes a body-stream read failure to the stable fail-closed diagnostic", async () => { + const stream = new ReadableStream({ + start(controller) { + controller.error(new Error("durable stream failed")); + }, + }); + await expectInvalid(new Response(stream, { status: 200 })); + }); +});