Classify images with Beam DiffusionGemma

.github/workflows/benchmark-images.yml

@@ -0,0 +1,32 @@
+name: Beam and RunPod latency
+on:
+  workflow_dispatch:
+  push:
+    branches: [codex/diffusiongemma-images]
+    paths:
+      - eval/diffusiongemma_latency.mjs
+      - .github/workflows/benchmark-images.yml
+permissions:
+  contents: read
+jobs:
+  benchmark:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v7
+      - uses: actions/setup-node@v7
+        with:
+          node-version: 22
+      - name: Compare direct inference APIs
+        env:
+          BEAM_API_KEY: ${{ secrets.BEAM_API_KEY }}
+          DGEMMA_URL: ${{ secrets.DGEMMA_URL }}
+          DGEMMA_TOKEN: ${{ secrets.DGEMMA_TOKEN }}
+          BENCH_SAMPLES: "30"
+        run: node eval/diffusiongemma_latency.mjs
+      - name: Save latency measurements
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: beam-runpod-latency
+          path: captures/beam-runpod-latency.json

.github/workflows/check.yml

@@ -24,6 +24,14 @@ jobs:
         with:
           bun-version: latest
       - run: npm ci
+      - name: Verify Beam image routing and SDK compatibility
+        run: npm run test:e2e:images
+      - name: Save image classification evidence
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: diffusiongemma-e2e
+          path: captures/diffusiongemma-fixture.json
       - run: npm run typecheck
       - run: npm test
         env:

e2e/diffusiongemma.live.py

@@ -0,0 +1,19 @@
+# /// script
+# dependencies = ["typesafe-sdk==0.7.1"]
+# ///
+# First run the JS image E2E to create the fixture, then:
+# uv run e2e/diffusiongemma.live.py (against npm run dev on port 3000)
+import base64, os
+from pathlib import Path
+from typesafe_sdk import TypeSafeClient,Choice,Noul,Score
+image='data:image/png;base64,'+base64.b64encode(Path('captures/diffusiongemma-red.png').read_bytes()).decode()
+with TypeSafeClient(api_key=os.environ.get('CLASSIFIER_API_KEY', 'unused'),base_url=os.environ.get('CLASSIFIER_BASE_URL', 'http://127.0.0.1:3000')) as client:
+ r=client.system_one(model='jev/diffusiongemma',state='Look at the image.',timeout=60,extra_body={'images':[image]},questions={
+  'color':Choice(instructions='What color is the image?',criteria={'red':None,'blue':None}),
+  'red':Noul(instructions='Is the image red?'),
+  'intensity':Score(instructions='How red is the image?',criteria=['Not red','Some red','Entirely red'])})
+ print('Python TypeSafe SDK image E2E passed')
+ Path('captures/diffusiongemma-python.json').write_text(r.model_dump_json(indent=2))
+ assert r.choices['color'].choice=='red'
+ assert r.nouls['red'].noul > 0.9
+ assert r.scores['intensity'].score > 1.8

e2e/diffusiongemma.ts

@@ -0,0 +1,109 @@
+import assert from "node:assert/strict";
+import { mkdir, writeFile } from "node:fs/promises";
+import { deflateSync } from "node:zlib";
+import { TypeSafeClient } from "@typesafe-ai/sdk";
+import worker, { type Env } from "../src/index";
+import { newMeter } from "../src/cost";
+import { Permit } from "../src/spending/permit";
+import { parseTokenRateCard, priceTokens } from "../src/server/token-pricing";
+import { providerCallBound } from "../src/server/token-reservation";
+import rates from "../src/retail-rates.json";
+
+// bun --env-file=.dev.vars e2e/diffusiongemma.ts --live
+// SDK → local HTTP Worker → Beam (or a local HTTP provider fixture).
+// Failure modes: dropped images, text-only fallback, wrong model/auth, excess
+// images, malformed replies, provider refusal, cancellation, and unmetered spend.
+const live = process.argv.includes("--live");
+if (live && !process.env.BEAM_API_KEY) throw new Error("BEAM_API_KEY is required for --live");
+const originalFetch = globalThis.fetch;
+let scenario = "success", calls = 0;
+const provider = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) {
+  calls++;
+  assert.equal(req.headers.get("authorization"), "Bearer fixture-beam");
+  const body = await req.json() as Record<string, any>;
+  assert.equal(body.model, "jev/diffusiongemma");
+  if (scenario === "busy") return Response.json({ detail: "busy" }, { status: 429 });
+  if (scenario === "invalid") return Response.json({ detail: "invalid image" }, { status: 422 });
+  if (scenario === "malformed") return Response.json({ model: body.model, answers: {} });
+  return Response.json({ model: body.model, answers: {
+    color: { type: "choice", choice: body.images ? "red" : "blue", confidence: 0.99, probabilities: { red: body.images ? 0.99 : 0.01, blue: body.images ? 0.01 : 0.99 } },
+    red: { type: "noul", noul: 0.99 },
+    intensity: { type: "score", score: 1.99, confidence: 0.99, legend: { 0: "Not red", 1: "Some red", 2: "Entirely red" }, probabilities: { 0: 0, 1: 0.01, 2: 0.99 } },
+  }, usage: { input_tokens: 310, output_tokens: 0 } });
+} });
+if (!live) globalThis.fetch = ((input, init) => {
+  if (String(input).startsWith("http://127.0.0.1:")) return originalFetch(input, init);
+  assert.equal(String(input), "https://app.beam.cloud/v1/systemone", "never fall back to a text model or pod");
+  return originalFetch(provider.url, init);
+}) as typeof fetch;
+const env = { DGEMMA_ENABLED: "true", BEAM_API_KEY: live ? process.env.BEAM_API_KEY : "fixture-beam",
+  DGEMMA_URL: "https://unused-pod.example", DGEMMA_TOKEN: "unused",
+  STATS: { get: async () => null, put: async () => {} },
+  LIMITER: { idFromName: (s: string) => s, get: () => ({ fetch: async () => Response.json({ limited: false, remaining: 59 }) }) },
+} as unknown as Env;
+const report: unknown[] = [];
+const server = Bun.serve({ hostname: "127.0.0.1", port: 0, async fetch(req) {
+  const meter = newMeter();
+  meter.permit = new Permit(10_000_000, Date.now() + 90000);
+  const response = await worker.fetch(req, env, { waitUntil: () => {} } as unknown as ExecutionContext, { meter });
+  await meter.permit.drain();
+  report.push({ status: response.status, response: await response.clone().json(), providerUsd: meter.usd,
+    tokens: meter.tokens, spending: { used: meter.permit.used, unknown: meter.permit.unknown } });
+  if (response.ok) {
+    assert.equal(meter.tokens[0].provider, "beam");
+    assert.equal(meter.tokens[0].model, "jev/diffusiongemma");
+    assert.ok(meter.usd > 0);
+    assert.equal(meter.permit.unknown, false);
+    assert.ok(Math.abs(meter.permit.used - meter.tokens[0].inputTokens! * 21) <= 1);
+  }
+  return response;
+} });
+// Deterministic 128×128 red PNG, constructed without external image assets.
+function chunk(type: string, data: Buffer) {
+  const name = Buffer.from(type), crc = Bun.hash.crc32(Buffer.concat([name, data]));
+  const size = Buffer.alloc(4), checksum = Buffer.alloc(4);
+  size.writeUInt32BE(data.length); checksum.writeUInt32BE(crc);
+  return Buffer.concat([size, name, data, checksum]);
+}
+const header = Buffer.alloc(13); header.writeUInt32BE(128); header.writeUInt32BE(128, 4); header[8] = 8; header[9] = 2;
+const row = Buffer.from([0, ...Array.from({ length: 128 }, () => [255, 0, 0]).flat()]);
+const png = Buffer.concat([Buffer.from("89504e470d0a1a0a", "hex"), chunk("IHDR", header), chunk("IDAT", deflateSync(Buffer.concat(Array(128).fill(row)))), chunk("IEND", Buffer.alloc(0))]);
+const image = `data:image/png;base64,${png.toString("base64")}`;
+const questions = {
+  color: { type: "choice" as const, instructions: "What color is the image or described square?", criteria: { red: null, blue: null } },
+  red: { type: "noul" as const, instructions: "Is the image red?" },
+  intensity: { type: "score" as const, instructions: "How red is the image?", criteria: ["Not red", "Some red", "Entirely red"] as const },
+};
+const client = new TypeSafeClient({ apiKey: "unused", baseURL: server.url.origin, retry: { maxRetries: 0 }, timeout: 65000 });
+try {
+  const request = { model: "jev/diffusiongemma", state: "Look at the image.", images: [image], questions };
+  const result = await client.systemOne(request);
+  assert.equal(result.answers.color.choice, "red");
+  assert.ok(result.answers.red.noul > 0.9);
+  assert.ok(result.answers.intensity.score > 1.8);
+  const alias = await client.systemOne({ ...request, model: "dgemma" });
+  assert.equal(alias.model, "jev/diffusiongemma");
+  const text = await client.systemOne({ model: "jev/diffusiongemma", state: "The square is blue.", questions });
+  assert.equal(text.answers.color.choice, "blue");
+  const post = (body: object) => fetch(new URL("/v1/systemone", server.url), { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify(body) });
+  for (const change of [{ images: [image, image] }, { images: [] }, { images: ["https://example.com/a.png"] }, { model: "jev-latest" }]) {
+    const before = calls;
+    assert.equal((await post({ ...request, ...change })).status, 400);
+    assert.equal(calls, before, "invalid requests must not reach the provider");
+  }
+  if (!live) for (const [mode, status] of [["busy", 429], ["invalid", 400], ["malformed", 503]] as const) {
+    scenario = mode;
+    const response = await post(request);
+    assert.equal(response.status, status);
+    if (mode === "busy") assert.ok(response.headers.get("retry-after"));
+  }
+  const card = parseTokenRateCard(JSON.stringify(rates))!;
+  assert.equal(providerCallBound(card, "beam", "jev/diffusiongemma", 0), 0);
+  assert.equal(priceTokens(card, [{ provider: "beam", model: "jev/diffusiongemma", calls: 1, inputTokens: 310, outputTokens: 0, cachedInputTokens: 0 }])?.nanodollars, 0n);
+  console.log(`DiffusionGemma ${live ? "live" : "fixture"} E2E passed (${report.length} HTTP requests)`);
+} finally {
+  await mkdir("captures", { recursive: true });
+  await writeFile(`captures/diffusiongemma-${live ? "live" : "fixture"}.json`, JSON.stringify({ live, sdk: "@typesafe-ai/sdk 0.6.0", results: report }, null, 2));
+  await writeFile("captures/diffusiongemma-red.png", png);
+  server.stop(true); provider.stop(true); globalThis.fetch = originalFetch;
+}

eval/diffusiongemma_latency.mjs

@@ -0,0 +1,101 @@
+// node --env-file=.dev.vars eval/diffusiongemma_latency.mjs
+// Direct HTTP, one request in flight, alternating service order, no retries.
+import assert from 'node:assert/strict';
+import { mkdir, writeFile } from 'node:fs/promises';
+import { performance } from 'node:perf_hooks';
+
+const samples = Number(process.env.BENCH_SAMPLES || 20);
+assert.ok(Number.isInteger(samples) && samples > 0);
+assert.ok(process.env.DGEMMA_URL, 'DGEMMA_URL is required');
+const runpodUrl = new URL(process.env.DGEMMA_URL);
+assert.equal(runpodUrl.protocol, 'https:', 'RunPod requires HTTPS');
+const services = [
+  { name: 'RunPod DiffusionGemma', url: runpodUrl.href.replace(/\/+$/, '') + '/v1/systemone', model: 'dgemma', key: process.env.DGEMMA_TOKEN },
+  { name: 'Beam DiffusionGemma', url: 'https://app.beam.cloud/v1/systemone', model: 'jev/diffusiongemma', key: process.env.BEAM_API_KEY },
+];
+for (const service of services) assert.ok(service.key, `${service.name} credential is required`);
+const criteria = { billing: null, technical: null, sales: null, feedback: null };
+const tickets = [
+  ['Please refund the duplicate charge on my invoice.', 'billing'],
+  ['The application crashes whenever I open the settings page.', 'technical'],
+  ['Please send a quote for 500 enterprise seats.', 'sales'],
+  ['I love the new design. Thank you for making it easier to use!', 'feedback'],
+];
+const category = (instructions) => ({ type: 'choice', instructions, criteria });
+// Deterministic 128 × 128 red PNG, shared by both services.
+const image = 'data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAIAAAACACAIAAABMXPacAAABcUlEQVR4nO3UwQkAMAwDsey/dDuGHmfQAIdDe+9u4AJbHy+wA+wAeoK9AL/CvqAuXxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC+J8QZwviPMFcb4gzhfE+YI4XxDnC17bByBow7KSKBBvAAAAAElFTkSuQmCC';
+const workloads = [
+  { name: 'single_text', state: tickets[0][0], questions: { category: category('Which team should handle this ticket?') }, expected: { category: 'billing' } },
+  { name: 'batch_16', state: Array.from({ length: 16 }, (_, i) => ({ id: `i${i}`, text: tickets[i % 4][0] })),
+    questions: Object.fromEntries(Array.from({ length: 16 }, (_, i) => [`i${i}`, category(`Which team should handle item i${i}?`)])),
+    expected: Object.fromEntries(Array.from({ length: 16 }, (_, i) => [`i${i}`, tickets[i % 4][1]])) },
+  { name: 'long_text', state: { background: 'The company offers software subscriptions, product documentation, and customer assistance. '.repeat(100), ticket: tickets[0][0] },
+    questions: { category: category('Which team should handle the ticket? The background is context, not the ticket.') }, expected: { category: 'billing' } },
+  { name: 'image_3_decisions', state: 'Look at the attached image.', images: [image], questions: {
+    color: { type: 'choice', instructions: 'What color is the image?', criteria: { red: null, blue: null, green: null } },
+    red: { type: 'noul', instructions: 'Is the image red?' },
+    intensity: { type: 'score', instructions: 'How red is the image?', criteria: ['Not red', 'Some red', 'Entirely red'] },
+  }, expected: { color: 'red' } },
+];
+const report = { startedAt: new Date().toISOString(), runtime: process.version, runner: process.env.GITHUB_ACTIONS === 'true' ? 'GitHub Actions ubuntu-latest' : 'local',
+  protocol: { samplesPerWorkload: samples, warmupsPerServiceAndWorkload: 2, concurrency: 1, retries: 0,
+    order: 'alternate service order every round; same state/questions for both services',
+    percentiles: 'Median averages the two middle observations; p95 uses nearest rank',
+    latency: 'client end-to-end milliseconds including network, headers and complete JSON response; persistent Node fetch connections',
+    limitations: 'Direct provider APIs from the same client; no classifier.dev proxy. Server regions, hardware and inference settings may differ. First request is not a verified cold start. Repeated synthetic workloads are a correctness sanity check, not an accuracy or calibration benchmark.' },
+  workloads: workloads.map(({ name, state, questions, images }) => ({ name, stateCharacters: JSON.stringify(state).length, decisions: Object.keys(questions).length, images: images?.length ?? 0 })),
+  rows: [], summaries: [] };
+const artifact = 'captures/beam-runpod-latency.json';
+await mkdir('captures', { recursive: true });
+async function save() { await writeFile(artifact, JSON.stringify(report, null, 2)); }
+async function call(service, work, round, warmup) {
+  const body = JSON.stringify({ model: service.model, state: work.state, questions: work.questions, ...(work.images ? { images: work.images } : {}) });
+  const start = performance.now();
+  const row = { service: service.name, workload: work.name, round, warmup, at: new Date().toISOString(), requestBytes: Buffer.byteLength(body) };
+  try {
+    const response = await fetch(service.url, { method: 'POST', headers: { authorization: `Bearer ${service.key}`, 'content-type': 'application/json' }, body, signal: AbortSignal.timeout(65000) });
+    row.headersMs = performance.now() - start;
+    const text = await response.text();
+    row.totalMs = performance.now() - start;
+    row.status = response.status;
+    const payload = JSON.parse(text);
+    row.model = payload.model;
+    row.usage = payload.usage;
+    row.providerTiming = payload.diagnostics?.timing;
+    row.valid = response.ok && payload.model === service.model && Object.keys(work.questions).every(id => payload.answers?.[id]?.type === work.questions[id].type);
+    if (row.valid) {
+      row.correct = Object.entries(work.expected).filter(([id, choice]) => payload.answers[id].choice === choice).length;
+      row.decisionsChecked = Object.keys(work.expected).length;
+      row.answers = payload.answers;
+    } else row.error = payload.detail ?? payload.error ?? 'invalid response';
+  } catch (error) { row.totalMs = performance.now() - start; row.valid = false; row.error = error.name; }
+  report.rows.push(row);
+  await save();
+  console.log(`${warmup ? 'warmup' : 'sample'} ${work.name} ${service.name} ${round + 1}: ${row.status ?? row.error} ${Math.round(row.totalMs)}ms`);
+  return row;
+}
+for (const work of workloads) for (const service of services) for (let i = 0; i < 2; i++) {
+  const row = await call(service, work, i, true);
+  if (!row.valid) throw new Error(`${service.name} cannot run ${work.name}; see ${artifact}`);
+}
+for (let round = 0; round < samples; round++) for (const work of workloads) {
+  const order = [...services];
+  if (round % 2) order.reverse();
+  for (const service of order) await call(service, work, round, false);
+}
+const percentile = (values, q) => q === .5 && values.length % 2 === 0
+  ? (values[values.length / 2 - 1] + values[values.length / 2]) / 2
+  : values[Math.max(0, Math.ceil(values.length * q) - 1)];
+for (const work of workloads) for (const service of services) {
+  const all = report.rows.filter(r => !r.warmup && r.service === service.name && r.workload === work.name);
+  const rows = all.filter(r => r.valid), times = rows.map(r => r.totalMs).sort((a, b) => a - b);
+  report.summaries.push({ service: service.name, workload: work.name, requests: all.length, successful: rows.length,
+    medianMs: percentile(times, .5), p95Ms: percentile(times, .95), minMs: times[0], maxMs: times.at(-1),
+    meanMs: times.reduce((a, b) => a + b, 0) / times.length,
+    medianInputTokens: percentile(rows.map(r => r.usage.input_tokens).sort((a, b) => a - b), .5),
+    correct: rows.reduce((n, r) => n + r.correct, 0), checked: rows.reduce((n, r) => n + r.decisionsChecked, 0),
+  });
+}
+report.finishedAt = new Date().toISOString();
+await save();
+console.table(report.summaries);

package.json

@@ -27,6 +27,7 @@
     "test:e2e:long-context-billing": "bun e2e/long-context-billing.ts",
     "test:e2e:long-context-job": "bun e2e/long-context-job.ts --full",
     "test:e2e:whole-document": "node e2e/whole-document.mjs --full",
+    "test:e2e:images": "bun e2e/diffusiongemma.ts",
     "test:e2e:url": "bun e2e/url-classification.ts"
   },
   "keywords": [],