H-6364: Temporal wiring + ingest UI redesign

apps/hash-external-services/docker-compose.yml

@@ -199,10 +199,9 @@ services:
         [
           "CMD",
           "temporal",
-          "workflow",
-          "list",
-          "--namespace",
-          "HASH",
+          "operator",
+          "cluster",
+          "health",
           "--address",
           "temporal:7233",
         ]

apps/hash-frontend/next.config.js

@@ -88,15 +88,24 @@ export default withSentryConfig(
           // Ingest pipeline proxy → Mastra API
           {
             source: "/api/ingest",
-            destination: `${mastraApiOrigin}/discovery-runs`,
+            destination: `${mastraApiOrigin}/ingest-runs`,
           },
           {
             source: "/api/ingest/:path*",
-            destination: `${mastraApiOrigin}/discovery-runs/:path*`,
+            destination: `${mastraApiOrigin}/ingest-runs/:path*`,
           },
           {
             source: "/api/ingest-fixtures/:path*",
-            destination: `${mastraApiOrigin}/discovery-fixtures/:path*`,
+            destination: `${mastraApiOrigin}/ingest-fixtures/:path*`,
+          },
+          {
+            source: "/api/ingest-artifacts/:path*",
+            destination: `${mastraApiOrigin}/artifacts/:path*`,
+          },
+          // Page images are referenced as /artifacts/... in discovery view data
+          {
+            source: "/artifacts/:path*",
+            destination: `${mastraApiOrigin}/artifacts/:path*`,
           },
           {
             source: "/pages",

apps/hash-frontend/package.json

@@ -10,14 +10,16 @@
     "build:docker": "docker buildx build --tag hash-frontend --file ../../infra/docker/frontend/prod/Dockerfile ../../ --load --build-arg FRONTEND_URL=\"$FRONTEND_URL\" --build-arg API_ORIGIN=\"$API_ORIGIN\"",
     "codegen": "rimraf './src/**/*.gen.*'; graphql-codegen --config codegen.config.ts",
     "dev": "next dev",
+    "dev:test": "vitest",
     "fix:eslint": "eslint --fix .",
     "fix:format": "biome format --write",
     "lint:eslint": "eslint --report-unused-disable-directives .",
     "lint:tsc": "tsc --noEmit",
     "start": "next start",
     "start:healthcheck": "wait-on --timeout 1200000 http://localhost:3000",
     "start:test": "next start",
-    "start:test:healthcheck": "wait-on --timeout 600000 http://localhost:3000"
+    "start:test:healthcheck": "wait-on --timeout 600000 http://localhost:3000",
+    "test:unit": "vitest --run"
   },
   "dependencies": {
     "@apollo/client": "3.10.5",
@@ -153,12 +155,14 @@
     "@types/react-dom": "19.2.3",
     "@types/react-window": "1.8.8",
     "@types/url-regex-safe": "1.0.2",
+    "@vitest/coverage-istanbul": "3.2.4",
     "@welldone-software/why-did-you-render": "10.0.1",
     "eslint": "9.39.3",
     "graphology-types": "0.24.8",
     "rimraf": "6.1.3",
     "sass": "1.93.2",
     "typescript": "5.9.3",
+    "vitest": "3.2.4",
     "wait-on": "9.0.1",
     "webpack": "5.104.1"
   }

apps/hash-frontend/src/pages/api/ingest/[runId]/events.api.test.ts

@@ -0,0 +1,227 @@
+import { EventEmitter } from "node:events";
+
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import handler from "./events.api";
+
+const encoder = new TextEncoder();
+const decodeChunks = (chunks: Uint8Array[]) =>
+  Buffer.concat(chunks.map((chunk) => Buffer.from(chunk))).toString("utf8");
+
+class MockRequest extends EventEmitter {
+  public method = "GET";
+  public query: Record<string, string | string[] | undefined>;
+  public headers: Record<string, string | string[] | undefined> = {};
+
+  constructor(query: Record<string, string | string[] | undefined>) {
+    super();
+    this.query = query;
+  }
+}
+
+class MockResponse extends EventEmitter {
+  public headersSent = false;
+  public writableEnded = false;
+  public statusCode = 200;
+  public readonly headers: Record<string, string> = {};
+  public readonly writes: Uint8Array[] = [];
+  public jsonBody: unknown;
+
+  setHeader(name: string, value: string) {
+    this.headers[name] = value;
+  }
+
+  status(code: number) {
+    this.statusCode = code;
+    return this;
+  }
+
+  json(body: unknown) {
+    this.jsonBody = body;
+    this.headersSent = true;
+    this.writableEnded = true;
+    return this;
+  }
+
+  writeHead(statusCode: number, headers: Record<string, string>) {
+    this.statusCode = statusCode;
+    Object.assign(this.headers, headers);
+    this.headersSent = true;
+  }
+
+  flushHeaders() {
+    this.headersSent = true;
+  }
+
+  write(chunk: Uint8Array) {
+    this.writes.push(chunk);
+    return true;
+  }
+
+  end(chunk?: Uint8Array) {
+    if (chunk) {
+      this.write(chunk);
+    }
+
+    this.writableEnded = true;
+    this.emit("finish");
+  }
+}
+
+const flushAsync = () =>
+  new Promise<void>((resolve) => {
+    setTimeout(resolve, 0);
+  });
+
+describe("ingest events api route", () => {
+  afterEach(() => {
+    vi.unstubAllGlobals();
+  });
+
+  it("streams live SSE chunks through to the client response", async () => {
+    let enqueueChunk: (chunk: string) => void = (_chunk: string) => {
+      throw new Error("Expected stream controller to be initialized");
+    };
+    let closeStream: () => void = () => {
+      throw new Error("Expected stream controller to be initialized");
+    };
+
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(
+        () =>
+          new Response(
+            new ReadableStream<Uint8Array>({
+              start(streamController) {
+                enqueueChunk = (chunk) => {
+                  streamController.enqueue(encoder.encode(chunk));
+                };
+                closeStream = () => {
+                  streamController.close();
+                };
+              },
+            }),
+            {
+              status: 200,
+              headers: { "content-type": "text/event-stream; charset=utf-8" },
+            },
+          ),
+      ),
+    );
+
+    const req = new MockRequest({ runId: "run-123" });
+    const res = new MockResponse();
+
+    const handlerPromise = handler(req as never, res as never);
+    await flushAsync();
+
+    enqueueChunk('event: phase-start\ndata: {"status":"running"}\n\n');
+    await flushAsync();
+
+    expect(decodeChunks(res.writes)).toContain("event: phase-start");
+    expect(res.writableEnded).toBe(false);
+
+    enqueueChunk('event: run-succeeded\ndata: {"status":"succeeded"}\n\n');
+    closeStream();
+
+    await handlerPromise;
+
+    expect(decodeChunks(res.writes)).toContain("event: run-succeeded");
+    expect(res.statusCode).toBe(200);
+    expect(res.headers["Content-Type"]).toContain("text/event-stream");
+    expect(res.writableEnded).toBe(true);
+  });
+
+  it("keeps the upstream stream alive through request close until upstream completes", async () => {
+    let enqueueChunk: (chunk: string) => void = (_chunk: string) => {
+      throw new Error("Expected stream controller to be initialized");
+    };
+    let closeStream: () => void = () => {
+      throw new Error("Expected stream controller to be initialized");
+    };
+    const cancelSpy = vi.fn();
+
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(
+        () =>
+          new Response(
+            new ReadableStream<Uint8Array>({
+              start(streamController) {
+                enqueueChunk = (chunk) => {
+                  streamController.enqueue(encoder.encode(chunk));
+                };
+                closeStream = () => {
+                  streamController.close();
+                };
+              },
+              cancel(reason) {
+                cancelSpy(reason);
+              },
+            }),
+            {
+              status: 200,
+              headers: { "content-type": "text/event-stream; charset=utf-8" },
+            },
+          ),
+      ),
+    );
+
+    const req = new MockRequest({ runId: "run-123" });
+    const res = new MockResponse();
+
+    const handlerPromise = handler(req as never, res as never);
+    await flushAsync();
+
+    req.emit("close");
+    await flushAsync();
+
+    enqueueChunk('event: run-succeeded\ndata: {"status":"succeeded"}\n\n');
+    closeStream();
+
+    await handlerPromise;
+
+    expect(cancelSpy).not.toHaveBeenCalled();
+    expect(decodeChunks(res.writes)).toContain("event: run-succeeded");
+    expect(res.writableEnded).toBe(true);
+  });
+
+  it("aborts the upstream stream when the client disconnects before completion", async () => {
+    let cancelResolve!: () => void;
+    const cancelPromise = new Promise<void>((resolve) => {
+      cancelResolve = resolve;
+    });
+
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(
+        () =>
+          new Response(
+            new ReadableStream<Uint8Array>({
+              start() {},
+              cancel() {
+                cancelResolve();
+              },
+            }),
+            {
+              status: 200,
+              headers: { "content-type": "text/event-stream; charset=utf-8" },
+            },
+          ),
+      ),
+    );
+
+    const req = new MockRequest({ runId: "run-123" });
+    const res = new MockResponse();
+
+    const handlerPromise = handler(req as never, res as never);
+    await flushAsync();
+
+    res.emit("close");
+
+    await cancelPromise;
+    await handlerPromise;
+
+    expect(res.writableEnded).toBe(true);
+  });
+});