pikasTech-HWLAB/internal/cloud/server-workbench-http.test.ts

import assert from "node:assert/strict";
import { test } from "bun:test";

import { createCloudApiServer } from "./server.ts";
import { createCodeAgentTraceStore } from "./code-agent-trace-store.ts";
import { createCodeAgentChatResultStore } from "./server-code-agent-http.ts";

const ACTOR = { id: "usr_workbench_reader", username: "reader", displayName: "Reader", role: "user", status: "active" };

test("workbench read model exposes session, messages, turn, and trace without write repair", async () => {
  const traceStore = createCodeAgentTraceStore();
  const results = createCodeAgentChatResultStore();
  const writes = { recordAgentSessionOwner: 0, updateWorkspace: 0 };
  const traceId = "trc_workbench_read_model";
  const session = {
    id: "ses_workbench_read_model",
    projectId: "prj_workbench_authority_session",
    agentId: "hwlab-code-agent",
    status: "completed",
    startedAt: "2026-06-17T00:00:00.000Z",
    endedAt: "2026-06-17T00:00:03.000Z",
    ownerUserId: ACTOR.id,
    conversationId: "cnv_workbench_read_model",
    threadId: "thread-workbench-read-model",
    lastTraceId: traceId,
    updatedAt: "2026-06-17T00:00:03.000Z",
    session: {
      sessionStatus: "completed",
      providerProfile: "codex-api",
      lastTraceId: traceId,
      messages: [
        { role: "user", text: "ping", traceId, createdAt: "2026-06-17T00:00:00.000Z" },
        { role: "assistant", text: "pong", traceId, createdAt: "2026-06-17T00:00:03.000Z" }
      ],
      valuesRedacted: true,
      secretMaterialStored: false
    }
  };
  const listInputs = [];
  traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
  traceStore.append(traceId, { type: "result", status: "completed", label: "result:completed", terminal: true });
  results.set(traceId, {
    status: "completed",
    traceId,
    ownerUserId: ACTOR.id,
    conversationId: session.conversationId,
    sessionId: session.id,
    threadId: session.threadId,
    finalResponse: "pong",
    agentRun: { runId: "run_workbench_read_model", commandId: "cmd_workbench_read_model", status: "completed" }
  });

  const accessController = {
    store: {
      async listAgentSessionsForUser(input = {}) {
        listInputs.push(input);
        return [session];
      },
      async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
      async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; },
      async recordAgentSessionOwner() { writes.recordAgentSessionOwner += 1; throw new Error("read model must not write session owner"); },
      async updateWorkspace() { writes.updateWorkspace += 1; throw new Error("read model must not update workspace"); }
    },
    async ensureBootstrap() {},
    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
  };
  const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results });
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));

  try {
    const { port } = server.address();
    const sessions = await getJson(port, "/v1/workbench/sessions");
    assert.equal(sessions.status, 200);
    assert.equal(sessions.body.contractVersion, "workbench-sessions-v1");
    assert.equal(sessions.body.sessions[0].sessionId, session.id);
    assert.equal(sessions.body.sessions[0].turnSummary.traceId, traceId);
    assert.equal(listInputs[0].projectId, undefined);

    const staleProject = await getJson(port, `/v1/workbench/sessions?projectId=prj_stale_filter&includeSessionId=${encodeURIComponent(session.id)}`);
    assert.equal(staleProject.status, 400);
    assert.equal(staleProject.body.error.code, "workbench_authority_removed");
    assert.equal(listInputs.length, 1);

    const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`);
    assert.equal(detail.status, 200);
    assert.equal(detail.body.session.sessionId, session.id);
    assert.equal(detail.body.session.messagePageUrl, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages`);
    assert.equal(detail.body.session.messageCount, 2);

    const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=1`);
    assert.equal(messages.status, 200);
    assert.equal(messages.body.contractVersion, "workbench-message-page-v1");
    assert.equal(messages.body.messages.length, 1);
    assert.equal(messages.body.messages[0].role, "user");
    assert.equal(messages.body.messages[0].text, "ping");
    assert.equal(messages.body.messages[0].parts[0].text, "ping");
    assert.equal(messages.body.hasMore, true);

    const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
    assert.equal(turn.status, 200);
    assert.equal(turn.body.contractVersion, "workbench-turn-snapshot-v1");
    assert.equal(turn.body.turn.traceId, traceId);
    assert.equal(turn.body.turn.userMessageId, messages.body.messages[0].messageId);
    assert.equal(turn.body.turn.terminal, true);

    const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=1`);
    assert.equal(trace.status, 200);
    assert.equal(trace.body.contractVersion, "workbench-trace-events-v1");
    assert.equal(trace.body.events.length, 1);
    assert.equal(trace.body.hasMore, true);

    assert.deepEqual(writes, { recordAgentSessionOwner: 0, updateWorkspace: 0 });
  } finally {
    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  }
});

test("workbench read model recovers trace events from durable projection without duplicating final response", async () => {
  const traceStore = createCodeAgentTraceStore();
  const traceId = "trc_workbench_durable_reload";
  const session = {
    id: "ses_workbench_durable_reload",
    projectId: "prj_hwpod_workbench",
    agentId: "hwlab-code-agent",
    status: "completed",
    startedAt: "2026-06-17T01:00:00.000Z",
    endedAt: "2026-06-17T01:00:04.000Z",
    ownerUserId: ACTOR.id,
    conversationId: "cnv_workbench_durable_reload",
    threadId: "thread-workbench-durable-reload",
    lastTraceId: traceId,
    updatedAt: "2026-06-17T01:00:04.000Z",
    session: {
      sessionStatus: "completed",
      lastTraceId: traceId,
      finalResponse: "durable pong",
      messages: [
        { role: "user", text: "durable ping", traceId, createdAt: "2026-06-17T01:00:00.000Z" },
        { role: "agent", text: "durable pong", traceId, createdAt: "2026-06-17T01:00:04.000Z" }
      ],
      valuesRedacted: true,
      secretMaterialStored: false
    }
  };
  const durableEvents = [
    { traceId, seq: 1, type: "request", status: "accepted", label: "request:accepted", createdAt: "2026-06-17T01:00:00.000Z", valuesPrinted: false },
    { traceId, seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true, createdAt: "2026-06-17T01:00:04.000Z", valuesPrinted: false }
  ];
  const accessController = {
    store: {
      async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
      async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
    },
    async ensureBootstrap() {},
    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
  };
  const runtimeStore = {
    async queryAgentTraceEvents(params = {}) {
      assert.equal(params.traceId, traceId);
      return { events: durableEvents, count: durableEvents.length };
    }
  };
  const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() });
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));

  try {
    const { port } = server.address();
    const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`);
    assert.equal(messages.status, 200);
    assert.equal(messages.body.total, 2);
    assert.deepEqual(messages.body.messages.map((message) => message.role), ["user", "agent"]);

    const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
    assert.equal(turn.status, 200);
    assert.equal(turn.body.turn.assistantMessageId, messages.body.messages[1].messageId);
    assert.equal(turn.body.turn.trace.eventCount, 2);
    assert.equal(turn.body.turn.trace.status, "completed");

    const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`);
    assert.equal(trace.status, 200);
    assert.equal(trace.body.events.length, 2);
    assert.equal(trace.body.traceStatus, "completed");
    assert.equal(trace.body.hasMore, false);
  } finally {
    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  }
});

test("workbench read model does not expose trace-only memory without visible session or result owner", async () => {
  const traceStore = createCodeAgentTraceStore();
  const traceId = "trc_workbench_trace_only";
  traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
  const accessController = {
    store: {
      async getAgentSessionByTraceId() { return null; }
    },
    async ensureBootstrap() {},
    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
  };
  const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: createCodeAgentChatResultStore() });
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));

  try {
    const { port } = server.address();
    const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`);
    assert.equal(turn.status, 404);
    assert.equal(turn.body.error.code, "workbench_turn_not_found");

    const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`);
    assert.equal(trace.status, 404);
    assert.equal(trace.body.error.code, "workbench_trace_not_found");
  } finally {
    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  }
});

test("workbench read model treats active session state as running", async () => {
  const traceId = "trc_workbench_active_state";
  const session = {
    id: "ses_workbench_active_state",
    projectId: "prj_hwpod_workbench",
    agentId: "hwlab-code-agent",
    status: "active",
    ownerUserId: ACTOR.id,
    conversationId: "cnv_workbench_active_state",
    threadId: "thread-workbench-active-state",
    lastTraceId: traceId,
    updatedAt: "2026-06-17T00:01:00.000Z",
    session: { sessionStatus: "active", lastTraceId: traceId, messages: [{ role: "user", text: "still running", traceId }] }
  };
  const accessController = {
    store: {
      async listAgentSessionsForUser() { return [session]; }
    },
    async ensureBootstrap() {},
    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
  };
  const server = createCloudApiServer({ accessController, codeAgentChatResults: createCodeAgentChatResultStore() });
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));

  try {
    const { port } = server.address();
    const sessions = await getJson(port, "/v1/workbench/sessions");
    assert.equal(sessions.status, 200);
    assert.equal(sessions.body.sessions[0].status, "active");
    assert.equal(sessions.body.sessions[0].running, true);
    assert.equal(sessions.body.sessions[0].terminal, false);
  } finally {
    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  }
});

test("workbench read model lets running trace override idle session summary", async () => {
  const traceStore = createCodeAgentTraceStore();
  const traceId = "trc_workbench_idle_with_running_trace";
  const session = {
    id: "ses_workbench_idle_with_running_trace",
    projectId: "prj_hwpod_workbench",
    agentId: "hwlab-code-agent",
    status: "idle",
    ownerUserId: ACTOR.id,
    conversationId: "cnv_workbench_idle_with_running_trace",
    threadId: "thread-workbench-idle-with-running-trace",
    lastTraceId: traceId,
    updatedAt: "2026-06-17T00:02:00.000Z",
    session: { sessionStatus: "idle", lastTraceId: traceId, messages: [{ role: "user", text: "still running despite idle row", traceId, status: "sent" }] }
  };
  traceStore.append(traceId, { type: "backend", status: "running", label: "runner:created", terminal: false });
  const accessController = {
    store: {
      async listAgentSessionsForUser() { return [session]; }
    },
    async ensureBootstrap() {},
    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
  };
  const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: createCodeAgentChatResultStore() });
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));

  try {
    const { port } = server.address();
    const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`);
    assert.equal(sessions.status, 200);
    assert.equal(sessions.body.sessions[0].status, "running");
    assert.equal(sessions.body.sessions[0].running, true);
    assert.equal(sessions.body.sessions[0].terminal, false);
    assert.equal(sessions.body.sessions[0].turnSummary.status, "running");
  } finally {
    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  }
});

test("workbench realtime stream accepts session authority without project or workspace", async () => {
  const traceStore = createCodeAgentTraceStore();
  const results = createCodeAgentChatResultStore();
  const traceId = "trc_workbench_realtime";
  const session = {
    id: "ses_workbench_realtime",
    projectId: "prj_hwpod_workbench",
    agentId: "hwlab-code-agent",
    status: "running",
    ownerUserId: ACTOR.id,
    conversationId: "cnv_workbench_realtime",
    threadId: "thread-workbench-realtime",
    lastTraceId: traceId,
    updatedAt: "2026-06-17T02:00:00.000Z",
    session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "stream", traceId }] }
  };
  traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
  results.set(traceId, { status: "running", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId });
  const accessController = {
    store: {
      async getAgentSession(sessionId) { return sessionId === session.id ? session : null; },
      async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
    },
    async ensureBootstrap() {},
    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; }
  };
  const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));

  try {
    const { port } = server.address();
    const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}`, 4);
    setTimeout(() => traceStore.append(traceId, { type: "backend", status: "running", label: "backend:live" }), 50);
    const events = await eventsPromise;
    assert.deepEqual(events.slice(0, 3).map((event) => event.event), [
      "workbench.connected",
      "workbench.trace.snapshot",
      "workbench.turn.snapshot"
    ]);
    assert.equal(events[0].data.filters.sessionId, session.id);
    assert.equal(events[1].data.traceId, traceId);
    assert.equal(events[2].data.turn.traceId, traceId);
    assert.equal(events[3].event, "workbench.trace.event");
    assert.equal(events[3].data.event.label, "backend:live");
  } finally {
    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
  }
});

async function getJson(port, path) {
  const response = await fetch(`http://127.0.0.1:${port}${path}`);
  return {
    status: response.status,
    body: await response.json()
  };
}

async function getSseEvents(port, path, count) {
  const controller = new AbortController();
  const response = await fetch(`http://127.0.0.1:${port}${path}`, { signal: controller.signal });
  assert.equal(response.status, 200);
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  const events = [];
  try {
    while (events.length < count) {
      const { value, done } = await reader.read();
      if (done) break;
      buffer += decoder.decode(value, { stream: true });
      for (;;) {
        const index = buffer.indexOf("\n\n");
        if (index < 0) break;
        const block = buffer.slice(0, index);
        buffer = buffer.slice(index + 2);
        events.push(parseSseBlock(block));
        if (events.length >= count) break;
      }
    }
  } finally {
    controller.abort();
  }
  return events;
}

function parseSseBlock(block) {
  const lines = block.split("\n");
  const event = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() ?? "message";
  const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
  return { event, data: JSON.parse(data) };
}