fix(agent-runtime): require JSON media type for current workflow evidence

src/agent-runtime/procedural-current-lifecycle.ts

@@ -70,6 +70,11 @@ function isRecord(value: unknown): value is Record<string, unknown> {
   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<Uint8Array> | 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<Uint8Array>,
   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) {

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",
     );
   });

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;
 

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<void> {
   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<Uint8Array>(), { status: 200 });
+    const response = new Response(new ReadableStream<Uint8Array>(), {
+      status: 200,
+      headers: JSON_HEADERS,
+    });
     const heldReader = response.body!.getReader();
     try {
       await expectInvalid(response);
@@ -164,17 +172,111 @@ 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<Uint8Array>({
+      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<void>((resolve) => {
+      markCancelStarted = resolve;
+    });
+    let releaseCancellation!: () => void;
+    const cancellationBarrier = new Promise<void>((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<Uint8Array>({
+      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<Uint8Array>({
       start(controller) {
         controller.enqueue("not-bytes" as unknown as Uint8Array);
         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 }));
   });
 });