fix(agent-runtime): bound current workflow-state response

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

@@ -26,6 +26,7 @@ const TERMINAL_WORKFLOW_TASK_STATES: ReadonlySet<CurrentWorkflowTaskState> = 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<unknown> {
+  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);
-}
+}

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<void> {
+  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<Uint8Array>({
+      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<Uint8Array>({
+      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<Uint8Array>({
+      start(controller) {
+        controller.error(new Error("durable stream failed"));
+      },
+    });
+    await expectInvalid(new Response(stream, { status: 200 }));
+  });
+});