fix(k6): keep response content out of load diagnostics

frontend/src/k6Diagnostics.test.ts

@@ -0,0 +1,229 @@
+// @vitest-environment node
+import { readFileSync } from "node:fs";
+import { runInNewContext } from "node:vm";
+import { describe, expect, it } from "vitest";
+
+const privatePayload = "synthetic-private-response-marker";
+
+function harness(name: string, post: () => object = () => ({
+  status: 503,
+  body: privatePayload,
+  timings: { duration: 1 },
+})) {
+  const source = readFileSync(new URL(`../../scripts/${name}`, import.meta.url), "utf8")
+    .replace(/^import .*;$/gm, "")
+    .replace(/export default function/g, "function iteration")
+    .replace(/export function/g, "function");
+  const context = {
+    __ENV: { REQUEST_TIMEOUT: "1s" },
+    http: { post },
+    check: () => true,
+    fail: (message: string) => { throw new Error(message); },
+    Counter: class { add() {} },
+    Trend: class { add() {} },
+  };
+  return runInNewContext(`${source}\n({ setup, ${name.includes("mcp") ? "result, structured" : ""} })`, context);
+}
+
+function rpcResult(result: unknown, id = 2) {
+  return { jsonrpc: "2.0", id, result };
+}
+
+describe("k6 diagnostic confidentiality", () => {
+  for (const name of ["k6_http_e2e.js", "k6_mcp_e2e.js"]) {
+    it(`${name} contains sign-in parser errors before sending another request`, () => {
+      let calls = 0;
+      const test = harness(name, () => {
+        calls++;
+        return { status: 200, json: () => { throw new Error(privatePayload); } };
+      });
+      expect(test.setup).toThrow(/^synthetic OIDC login returned an invalid token response$/);
+      expect(calls).toBe(1);
+    });
+
+    it.each([undefined, null, "", " ", 7, []].map(value => [value]))(`${name} rejects an invalid access token (%j)`, (token) => {
+      let calls = 0;
+      const test = harness(name, () => {
+        calls++;
+        return { status: 200, json: () => token };
+      });
+      expect(test.setup).toThrow(/^synthetic OIDC login returned an invalid token response$/);
+      expect(calls).toBe(1);
+    });
+  }
+
+  it("contains an accepted Ask response parser error", () => {
+    let calls = 0;
+    const test = harness("k6_http_e2e.js", () => ++calls === 1
+      ? { status: 200, json: () => "synthetic-token" }
+      : { status: 202, timings: { duration: 1 }, json: () => { throw new Error(privatePayload); } });
+    expect(test.setup).toThrow(/^synthetic Ask enqueue returned an unreadable response$/);
+    expect(calls).toBe(2);
+  });
+
+  for (const name of ["k6_http_e2e.js", "k6_mcp_e2e.js"]) {
+    it.each([undefined, null, "", " ", 7, [], "synthetic-job"].map(value => [value]))(`${name} validates the accepted job identifier (%j)`, (askJobId) => {
+      let calls = 0;
+      const test = harness(name, () => {
+        calls++;
+        if (calls === 1) return { status: 200, json: () => "synthetic-token" };
+        if (name.includes("mcp") && calls === 2) return {
+          status: 200, timings: { duration: 1 }, headers: { "Mcp-Session-Id": "synthetic-session" },
+          body: `data: ${JSON.stringify(rpcResult({}, 1))}`,
+        };
+        if (name.includes("mcp") && calls === 3) return { status: 202 };
+        return {
+          status: name.includes("mcp") ? 200 : 202, timings: { duration: 1 },
+          json: () => askJobId,
+          body: `data: ${JSON.stringify(rpcResult({ structuredContent: { ask_job_id: askJobId } }))}`,
+        };
+      });
+      if (askJobId === "synthetic-job") {
+        expect(test.setup()).toEqual({ token: "synthetic-token", askJobId });
+      } else {
+        expect(test.setup).toThrow(name.includes("mcp")
+          ? /^MCP Ask submit returned an invalid job identifier$/
+          : /^synthetic Ask enqueue returned an invalid job identifier$/);
+      }
+      expect(calls).toBe(name.includes("mcp") ? 4 : 2);
+    });
+  }
+
+  it("omits a rejected Ask response body", () => {
+    const source = readFileSync(new URL("../../scripts/k6_http_e2e.js", import.meta.url), "utf8")
+      .replace(/^import .*;$/gm, "")
+      .replace(/export default function/g, "function iteration")
+      .replace(/export function/g, "function");
+    let calls = 0;
+    const setup = runInNewContext(`${source}\nsetup`, {
+      __ENV: { REQUEST_TIMEOUT: "1s" },
+      http: { post: () => ++calls === 1
+        ? { status: 200, json: () => "synthetic-token" }
+        : { status: 503, body: privatePayload, timings: { duration: 1 } } },
+      fail: (message: string) => { throw new Error(message); },
+      Counter: class { add() {} },
+      Trend: class { add() {} },
+    });
+    expect(setup).toThrow(/^synthetic Ask enqueue failed with HTTP 503$/);
+    expect(calls).toBe(2);
+  });
+
+  it("omits JSON-RPC error details", () => {
+    expect(() => harness("k6_mcp_e2e.js").result({
+      status: 200,
+      body: `data: ${JSON.stringify({ jsonrpc: "2.0", id: 2, error: { message: privatePayload } })}`,
+    }, 2)).toThrow(/^MCP request failed: HTTP 200$/);
+  });
+
+  it("omits failed tool content", () => {
+    expect(() => harness("k6_mcp_e2e.js").structured({
+      status: 200,
+      body: `data: ${JSON.stringify(rpcResult({ isError: true, content: privatePayload }))}`,
+    }, 2)).toThrow(/^MCP tool failed: HTTP 200$/);
+  });
+
+  it("contains malformed JSON without exposing the parser excerpt", () => {
+    expect(() => harness("k6_mcp_e2e.js").result({
+      status: 200, body: `data: ${privatePayload}`,
+    }, 2)).toThrow(/^MCP response was unreadable: HTTP 200$/);
+  });
+
+  it("preserves successful structured evidence", () => {
+    const payload = { ask_job_id: "synthetic-job", job_status_code: "queued" };
+    expect(harness("k6_mcp_e2e.js").structured({
+      status: 200,
+      body: `data: ${JSON.stringify(rpcResult({ structuredContent: payload }))}`,
+    }, 2)).toEqual(payload);
+  });
+
+  it.each([null, [], "synthetic-private-response-marker", 7].map(value => [value]))("rejects invalid envelopes (%j)", (envelope) => {
+    expect(() => harness("k6_mcp_e2e.js").result({
+      status: 200, body: `data: ${JSON.stringify(envelope)}`,
+    }, 2)).toThrow(/^MCP response envelope was invalid: HTTP 200$/);
+  });
+
+  it("rejects an envelope without result or error", () => {
+    expect(() => harness("k6_mcp_e2e.js").result({
+      status: 200, body: `data: ${JSON.stringify({ jsonrpc: "2.0", id: 2 })}`,
+    }, 2)).toThrow(/^MCP response result\/error shape was invalid: HTTP 200$/);
+  });
+
+  it.each([
+    [{ id: 2, result: {} }, /^MCP response protocol was invalid: HTTP 200$/],
+    [{ jsonrpc: "1.0", id: 2, result: {} }, /^MCP response protocol was invalid: HTTP 200$/],
+    [{ jsonrpc: "2.0", result: {} }, /^MCP response id mismatch: HTTP 200$/],
+    [{ jsonrpc: "2.0", id: 7, result: {} }, /^MCP response id mismatch: HTTP 200$/],
+    [{ jsonrpc: "2.0", id: 2, result: {}, error: null }, /^MCP response result\/error shape was invalid: HTTP 200$/],
+  ])("rejects malformed or unrelated JSON-RPC responses (%j)", (envelope, expectedError) => {
+    expect(() => harness("k6_mcp_e2e.js").result({
+      status: 200, body: `data: ${JSON.stringify(envelope)}`,
+    }, 2)).toThrow(expectedError);
+  });
+
+  it.each([null, [], privatePayload].map(value => [value]))("rejects invalid tool results (%j)", (result) => {
+    expect(() => harness("k6_mcp_e2e.js").structured({
+      status: 200, body: `data: ${JSON.stringify(rpcResult(result))}`,
+    }, 2)).toThrow(/^MCP tool result was invalid: HTTP 200$/);
+  });
+
+  it.each([undefined, null, [], privatePayload].map(value => [value]))("rejects invalid structured content (%j)", (structuredContent) => {
+    expect(() => harness("k6_mcp_e2e.js").structured({
+      status: 200, body: `data: ${JSON.stringify(rpcResult({ structuredContent }))}`,
+    }, 2)).toThrow(/^MCP structured content was invalid: HTTP 200$/);
+  });
+
+  it("contains a missing response body", () => {
+    expect(() => harness("k6_mcp_e2e.js").result({ status: 0, body: null }, 2))
+      .toThrow(/^MCP response omitted a data event: HTTP 0$/);
+  });
+});
+
+
+describe("k6 job-state metric confidentiality", () => {
+  function iteration(name: string, status: unknown, unreadable = false) {
+    const source = readFileSync(new URL(`../../scripts/${name}`, import.meta.url), "utf8")
+      .replace(/^import .*;$/gm, "")
+      .replace(/export default function/g, "function iteration")
+      .replace(/export function/g, "function");
+    const observations: unknown[] = [];
+    const checks: Record<string, boolean> = {};
+    const response = {
+      status: 200, timings: { duration: 1 },
+      json: () => { if (unreadable) throw new Error(privatePayload); return status; },
+      body: `data: ${JSON.stringify(rpcResult({ structuredContent: { job_status_code: status } }, 3))}`,
+    };
+    const run = runInNewContext(`${source}\n${name.includes("mcp") ? 'vuSession = "synthetic-session";' : ""} iteration`, {
+      __ENV: { REQUEST_TIMEOUT: "1s" },
+      http: { batch: () => [response, response, response], post: () => response },
+      check: (value: unknown, predicates: Record<string, (value: unknown) => boolean>) => {
+        for (const [name, predicate] of Object.entries(predicates)) checks[name] = predicate(value);
+        return Object.keys(predicates).every(name => checks[name]);
+      },
+      fail: (message: string) => { throw new Error(message); },
+      Counter: class { add(_value: number, tags: unknown) { observations.push(tags); } },
+      Trend: class { add() {} },
+    });
+    return { run: () => run({ token: "synthetic-token", askJobId: "synthetic-job" }), observations, checks };
+  }
+
+  for (const name of ["k6_http_e2e.js", "k6_mcp_e2e.js"]) {
+    it.each(["queued", "running", "succeeded", "failed"])(`${name} retains declared state %s`, (status) => {
+      const test = iteration(name, status);
+      test.run();
+      expect(test.observations).toEqual([{ job_status: status }]);
+      expect(test.checks[name.includes("mcp") ? "MCP Ask state is declared" : "Ask state is declared"]).toBe(true);
+    });
+    it.each([privatePayload, null, [], { body: privatePayload }, 7].map(value => [value]))(`${name} contains undeclared state %j`, (status) => {
+      const test = iteration(name, status);
+      test.run();
+      expect(test.observations).toEqual([{ job_status: "unknown" }]);
+      expect(test.checks[name.includes("mcp") ? "MCP Ask state is declared" : "Ask state is declared"]).toBe(false);
+    });
+  }
+
+  it("contains HTTP poll parser errors without discarding successful read observations", () => {
+    const test = iteration("k6_http_e2e.js", null, true);
+    expect(test.run).not.toThrow();
+    expect(test.observations).toEqual([{ job_status: "unknown" }]);
+  });
+});

scripts/k6_http_e2e.js

@@ -18,6 +18,8 @@ const username = __ENV.K6_USERNAME || "demo.analyst";
 const password = __ENV.K6_PASSWORD || "lineageweave-demo-only";
 const requestTimeout = __ENV.REQUEST_TIMEOUT;
 const unitlessDuration = /^\d+(?:\.\d+)?$/;
+// Persisted Global Ask states: migrations/0165_global_ask_job.sql.
+const jobStates = new Set(["queued", "running", "succeeded", "failed"]);
 
 const askEnqueueDuration = new Trend("lineageweave_ask_enqueue_duration", true);
 const readDuration = new Trend("lineageweave_read_duration", true);
@@ -40,7 +42,16 @@ function authenticate() {
   if (response.status !== 200) {
     fail(`synthetic OIDC login failed with HTTP ${response.status}`);
   }
-  return response.json("access_token");
+  let token;
+  try {
+    token = response.json("access_token");
+  } catch {
+    // Parser errors can include response content; validation below stays bounded.
+  }
+  if (typeof token !== "string" || !token.trim()) {
+    fail("synthetic OIDC login returned an invalid token response");
+  }
+  return token;
 }
 
 function readBatch(token, askJobId) {
@@ -83,9 +94,18 @@ export function setup() {
   );
   askEnqueueDuration.add(submitted.timings.duration);
   if (submitted.status !== 202) {
-    fail(`synthetic Ask enqueue failed with HTTP ${submitted.status}: ${submitted.body}`);
+    fail(`synthetic Ask enqueue failed with HTTP ${submitted.status}`);
+  }
+  let askJobId;
+  try {
+    askJobId = submitted.json("ask_job_id");
+  } catch {
+    fail("synthetic Ask enqueue returned an unreadable response");
+  }
+  if (typeof askJobId !== "string" || !askJobId.trim()) {
+    fail("synthetic Ask enqueue returned an invalid job identifier");
   }
-  return { token, askJobId: submitted.json("ask_job_id") };
+  return { token, askJobId };
 }
 
 export default function (data) {
@@ -100,8 +120,15 @@ export default function (data) {
   readDuration.add(responses[1].timings.duration, { endpoint: "lineage" });
   askPollDuration.add(responses[2].timings.duration);
   if (responses[2].status === 200) {
+    let status;
+    try {
+      status = responses[2].json("job_status_code");
+    } catch {
+      // Parser exceptions may include response content; keep the observation bounded.
+    }
+    check(status, { "Ask state is declared": (value) => jobStates.has(value) });
     askStateObservations.add(1, {
-      job_status: String(responses[2].json("job_status_code") || "unknown"),
+      job_status: jobStates.has(status) ? status : "unknown",
     });
   }
   check(responses[0], { "posts read succeeds": (response) => response.status === 200 });