diff --git a/src/agent-runtime/procedural-current-lifecycle.ts b/src/agent-runtime/procedural-current-lifecycle.ts index 4c98ad18f..e22767e72 100644 --- a/src/agent-runtime/procedural-current-lifecycle.ts +++ b/src/agent-runtime/procedural-current-lifecycle.ts @@ -70,6 +70,11 @@ function isRecord(value: unknown): value is Record { return value !== null && typeof value === "object" && !Array.isArray(value); } +/** Accept only the exact JSON media-type forms emitted by the private workflow-state owner. */ +function isJsonMediaType(value: string | null): boolean { + return /^[ \t]*application\/json[ \t]*(?:;[ \t]*charset[ \t]*=[ \t]*utf-8[ \t]*)?$/iu.test(value ?? ""); +} + function currentTaskState(value: unknown): CurrentWorkflowTaskState { if (typeof value !== "string" || !CURRENT_WORKFLOW_TASK_STATES.has(value as CurrentWorkflowTaskState)) { return rejectCurrentLifecycle("invalid_workflow_state_response"); @@ -123,6 +128,19 @@ function currentWorkflowEvidence( }); } +/** Requests response-body cleanup without allowing cleanup completion to become decision authority. */ +function cancelCurrentWorkflowResponseBodyBestEffort( + body: ReadableStream | null, + reason: string, +): void { + if (body === null) return; + try { + void body.cancel(reason).catch(() => undefined); + } catch { + // A locked or hostile body must not replace the stable fail-closed admission result. + } +} + function cancelCurrentWorkflowReaderBestEffort( reader: ReadableStreamDefaultReader, reason: string, @@ -218,6 +236,13 @@ async function readCurrentWorkflowEvidence( if (response.status === 409) return rejectCurrentLifecycle("workflow_state_conflict"); if (response.status === 503) return rejectCurrentLifecycle("workflow_state_unavailable"); if (response.status !== 200) return rejectCurrentLifecycle("invalid_workflow_state_response"); + if (!isJsonMediaType(response.headers.get("content-type"))) { + cancelCurrentWorkflowResponseBodyBestEffort( + response.body, + "Noema current workflow-state response used an unsupported media type", + ); + return rejectCurrentLifecycle("invalid_workflow_state_response"); + } const body = await boundedCurrentWorkflowResponse(response); if (!isRecord(body) || body.ok !== true) { diff --git a/test/procedural-current-lifecycle-fail-closed.test.ts b/test/procedural-current-lifecycle-fail-closed.test.ts index 219395b66..82491ba35 100644 --- a/test/procedural-current-lifecycle-fail-closed.test.ts +++ b/test/procedural-current-lifecycle-fail-closed.test.ts @@ -13,6 +13,7 @@ type TaskState = "pending" | "running" | "succeeded" | "failed" | "cancelled" | type ResponseStep = Response | Error; const executionId = "run-current-lifecycle-fail-closed"; +const JSON_HEADERS = { "content-type": "application/json; charset=utf-8" }; function plan(): WorkflowTaskPlan { return { @@ -50,7 +51,7 @@ function snapshot(states: readonly TaskState[], transitionSequence = 2): Record< } function ok(data: unknown): Response { - return new Response(JSON.stringify({ ok: true, data }), { status: 200 }); + return new Response(JSON.stringify({ ok: true, data }), { status: 200, headers: JSON_HEADERS }); } class WorkflowNamespace { @@ -136,10 +137,13 @@ describe("procedural current lifecycle fail-closed evidence validation", () => { }); it("rejects malformed success envelopes before they can become lifecycle evidence", async () => { - await expectFailure(new Response("{", { status: 200 }), "invalid_workflow_state_response"); + await expectFailure(new Response("{", { status: 200, headers: JSON_HEADERS }), "invalid_workflow_state_response"); await expectFailure(ok(null), "invalid_workflow_state_response"); await expectFailure( - new Response(JSON.stringify({ ok: false, data: snapshot(["running", "pending"]) }), { status: 200 }), + new Response( + JSON.stringify({ ok: false, data: snapshot(["running", "pending"]) }), + { status: 200, headers: JSON_HEADERS }, + ), "invalid_workflow_state_response", ); }); diff --git a/test/procedural-current-lifecycle-read-deadline.test.ts b/test/procedural-current-lifecycle-read-deadline.test.ts index 68c849603..e56f154d5 100644 --- a/test/procedural-current-lifecycle-read-deadline.test.ts +++ b/test/procedural-current-lifecycle-read-deadline.test.ts @@ -82,6 +82,7 @@ describe("procedural current-lifecycle read deadline", () => { }; const response = { status: 200, + headers: new Headers({ "content-type": "application/json; charset=utf-8" }), body: { getReader: () => reader }, } as unknown as Response; diff --git a/test/procedural-current-lifecycle-response-bounds.test.ts b/test/procedural-current-lifecycle-response-bounds.test.ts index 7dd122c3d..d95f20fa8 100644 --- a/test/procedural-current-lifecycle-response-bounds.test.ts +++ b/test/procedural-current-lifecycle-response-bounds.test.ts @@ -8,7 +8,9 @@ import type { WorkflowStateDurableObjectEnv } from "../src/workflow-task-executi import type { WorkflowTaskPlan } from "../src/workflow-task-execution/task-plan"; const executionId = "run-current-lifecycle-response-bound"; +const JSON_HEADERS = { "content-type": "application/json; charset=utf-8" }; +/** Keeps scheduling trivial so failures stay attributable to current-response admission. */ function plan(): WorkflowTaskPlan { return { executionId, @@ -18,6 +20,7 @@ function plan(): WorkflowTaskPlan { }; } +/** Builds the matching Agent Runtime session required to exercise the real procedural ACL. */ async function session() { const graph = await createProceduralGraph({ schemaVersion: "noema.procedural-graph/v1", @@ -44,6 +47,7 @@ async function session() { }); } +/** Routes the supplied hostile response through the normal execution-scoped Durable Object port. */ function envFor(response: Response): WorkflowStateDurableObjectEnv { const namespace = { idFromName(name: string): DurableObjectId { @@ -56,6 +60,7 @@ function envFor(response: Response): WorkflowStateDurableObjectEnv { return { NOEMA_WORKFLOW_STATE: namespace as unknown as DurableObjectNamespace }; } +/** Requires every hostile transport fixture to collapse to the stable fail-closed diagnostic. */ async function expectInvalid(response: Response): Promise { await expect(guideProceduralExecutionFromCurrentWorkflowState( envFor(response), @@ -94,7 +99,7 @@ describe("procedural current-lifecycle response bounds", () => { await expectInvalid(new Response(stream, { status: 200, - headers: { "content-type": "application/json; charset=utf-8" }, + headers: JSON_HEADERS, })); expect(cancelled).toBe(true); @@ -123,7 +128,7 @@ describe("procedural current-lifecycle response bounds", () => { const outcome = guideProceduralExecutionFromCurrentWorkflowState( envFor(new Response(stream, { status: 200, - headers: { "content-type": "application/json; charset=utf-8" }, + headers: JSON_HEADERS, })), plan(), await session(), @@ -154,7 +159,10 @@ describe("procedural current-lifecycle response bounds", () => { }); it("normalizes a locked durable-owner response body to the stable fail-closed diagnostic", async () => { - const response = new Response(new ReadableStream(), { status: 200 }); + const response = new Response(new ReadableStream(), { + status: 200, + headers: JSON_HEADERS, + }); const heldReader = response.body!.getReader(); try { await expectInvalid(response); @@ -164,9 +172,103 @@ describe("procedural current-lifecycle response bounds", () => { }); it("rejects a successful status with no response body", async () => { + await expectInvalid(new Response(null, { status: 200, headers: JSON_HEADERS })); + }); + + it("rejects a successful response without a declared JSON media type when no body exists", async () => { await expectInvalid(new Response(null, { status: 200 })); }); + it("preserves fail-closed admission when non-JSON body cancellation throws synchronously", async () => { + const response = { + status: 200, + headers: new Headers({ "content-type": "text/plain; charset=utf-8" }), + body: { + cancel() { + throw new Error("cleanup transport failed synchronously"); + }, + }, + } as unknown as Response; + + await expectInvalid(response); + }); + + it("preserves fail-closed admission when non-JSON body cancellation rejects asynchronously", async () => { + let cancelRequested = false; + const stream = new ReadableStream({ + cancel() { + cancelRequested = true; + return Promise.reject(new Error("cleanup transport rejected")); + }, + }); + + await expectInvalid(new Response(stream, { + status: 200, + headers: { "content-type": "text/plain; charset=utf-8" }, + })); + await Promise.resolve(); + + expect(cancelRequested).toBe(true); + }); + + it("cancels parseable JSON identified as non-JSON without waiting for cleanup", async () => { + let markCancelStarted!: () => void; + const cancelStarted = new Promise((resolve) => { + markCancelStarted = resolve; + }); + let releaseCancellation!: () => void; + const cancellationBarrier = new Promise((resolve) => { + releaseCancellation = resolve; + }); + const payload = JSON.stringify({ + ok: true, + data: { + executionId, + planId: "plan-current-lifecycle-response-bound", + transitionSequence: 1, + cancellation: { requested: false, cancellationId: null }, + tasks: [{ taskId: "review", state: "pending" }], + }, + }); + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode(payload)); + }, + cancel() { + markCancelStarted(); + return cancellationBarrier; + }, + }, { highWaterMark: 0 }); + const outcome = guideProceduralExecutionFromCurrentWorkflowState( + envFor(new Response(stream, { + status: 200, + headers: { "content-type": "text/plain; charset=utf-8" }, + })), + plan(), + await session(), + { lastProcedure: null, hops: 1, maxEdges: 4 }, + ).then( + () => new Error("expected the non-JSON response to fail closed"), + (error: unknown) => error, + ); + + await cancelStarted; + let settled = false; + void outcome.then(() => { + settled = true; + }); + for (let index = 0; index < 5; index += 1) await Promise.resolve(); + const settledBeforeCleanup = settled; + + releaseCancellation(); + const error = await outcome; + expect(error).toMatchObject({ + name: "ProceduralCurrentLifecycleError", + code: "invalid_workflow_state_response", + }); + expect(settledBeforeCleanup).toBe(true); + }); + it("fails closed on a malformed non-byte response chunk", async () => { const stream = new ReadableStream({ start(controller) { @@ -174,7 +276,7 @@ describe("procedural current-lifecycle response bounds", () => { controller.close(); }, }); - await expectInvalid(new Response(stream, { status: 200 })); + await expectInvalid(new Response(stream, { status: 200, headers: JSON_HEADERS })); }); it("normalizes a body-stream read failure to the stable fail-closed diagnostic", async () => { @@ -183,6 +285,6 @@ describe("procedural current-lifecycle response bounds", () => { controller.error(new Error("durable stream failed")); }, }); - await expectInvalid(new Response(stream, { status: 200 })); + await expectInvalid(new Response(stream, { status: 200, headers: JSON_HEADERS })); }); });