pikasTech-HWLAB/internal/cloud/server-agent-observer-http.ts

// SPEC: PJ2026-010405 CONSOLE-REQ-007; MDTODO R3.
// Responsibility: expose one authenticated Kafka retention-to-live SSE stream for the AgentRun observer.

import { createWorkbenchKafkaRefreshHandoff } from "./workbench-kafka-refresh-handoff.ts";

export async function handleAgentObserverHttp(request, response, options = {}) {
  if (request.method !== "GET") {
    response.writeHead(405, { allow: "GET", "content-type": "application/json; charset=utf-8" });
    response.end(JSON.stringify({ ok: false, error: { code: "method_not_allowed", message: "GET required." } }));
    return;
  }

  const bridge = options.kafkaEventBridge;
  if (!bridge?.started || typeof bridge.subscribeLiveHwlabEvents !== "function" || typeof bridge.queryHwlabEventRetention !== "function") {
    sendJson(response, 503, observerError("agent_observer_kafka_unavailable", "Agent observer Kafka retention/live runtime is unavailable."));
    return;
  }

  let policy;
  try {
    policy = agentObserverPolicy(options.env ?? process.env);
  } catch (error) {
    sendJson(response, 503, observerError(error?.code ?? "agent_observer_config_invalid", error instanceof Error ? error.message : String(error)));
    return;
  }
  const requestUrl = new URL(request.url ?? "/v1/agent-observer/events", "http://agent-observer.local");
  const resumeAfterId = text(requestUrl.searchParams.get("afterId") ?? request.headers?.["last-event-id"]);
  const liveOnly = requestUrl.searchParams.get("mode") === "live";

  try {
    await (bridge.liveReady ?? bridge.ready);
  } catch (error) {
    sendJson(response, 503, observerError(error?.code ?? "agent_observer_kafka_start_failed", error instanceof Error ? error.message : String(error)));
    return;
  }

  let closed = false;
  const cleanup = [];
  const active = () => !closed && !response.destroyed && !response.writableEnded;
  const close = () => {
    if (closed) return;
    closed = true;
    for (const dispose of cleanup.splice(0)) dispose();
  };
  response.once("close", close);
  request.once?.("aborted", close);
  request.socket?.once?.("close", close);
  cleanup.push(() => request.off?.("aborted", close));
  cleanup.push(() => request.socket?.off?.("close", close));

  response.writeHead(200, {
    "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store, no-transform",
    connection: "keep-alive",
    "x-accel-buffering": "no",
    "x-content-type-options": "nosniff"
  });
  response.flushHeaders?.();

  const writeEvent = async (name, payload) => {
    if (!active()) return false;
    const eventId = text(payload?.eventId ?? payload?.sourceEventId);
    const block = `${eventId ? `id: ${eventId}\n` : ""}event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`;
    if (!response.write(block)) await waitForDrain(response, active);
    return active();
  };
  let writeChain = Promise.resolve(true);
  const enqueue = (name, payload) => {
    writeChain = writeChain.then(() => writeEvent(name, payload));
    return writeChain;
  };
  const fail = async (error) => {
    if (!active()) return;
    await enqueue("agent-observer.error", observerError(error?.code ?? "agent_observer_stream_failed", error instanceof Error ? error.message : String(error)).error);
    await writeChain;
    if (!response.writableEnded) response.end();
  };

  if (liveOnly) {
    const unsubscribe = bridge.subscribeLiveHwlabEvents((envelope) => { void enqueue("hwlab.event.v1", envelope); });
    cleanup.push(() => unsubscribe?.());
    await enqueue("agent-observer.connected", {
      type: "connected",
      status: "live",
      deliverySemantics: "agentrun-snapshot-then-kafka-live",
      authority: "hwlab.event.v1",
      replay: null,
      serverSentAt: new Date().toISOString(),
      valuesPrinted: false
    });
  } else {
    const handoff = createWorkbenchKafkaRefreshHandoff({
    allowUnscoped: true,
    resumeAfterId,
    liveBufferLimit: policy.liveBufferLimit,
    identityWindowLimit: policy.identityWindowLimit,
    subscribeLive: (listener) => bridge.subscribeLiveHwlabEvents(listener),
    queryRetention: ({ signal }) => bridge.queryHwlabEventRetention({
      limit: policy.retentionEventLimit,
      scanLimit: policy.retentionScanLimit,
      timeoutMs: policy.retentionTimeoutMs,
      groupIdPrefix: policy.retentionGroupPrefix,
      fromBeginning: true,
      signal
    }),
    deliverEvent: (envelope) => enqueue("hwlab.event.v1", envelope),
    deliverHandoff: (summary) => enqueue("agent-observer.handoff", {
      type: "handoff",
      status: "handoff",
      deliverySemantics: "kafka-retention-then-live",
      authority: "hwlab.event.v1",
      replay: summary,
      serverSentAt: new Date().toISOString(),
      valuesPrinted: false
    }),
    deliverConnected: (summary) => enqueue("agent-observer.connected", {
      type: "connected",
      status: "live",
      deliverySemantics: "kafka-retention-then-live",
      authority: "hwlab.event.v1",
      replay: summary,
      serverSentAt: new Date().toISOString(),
      valuesPrinted: false
    }),
    onFailure: fail
    });
    cleanup.push(() => handoff.stop("connection-closed"));

    try {
      await handoff.start();
    } catch (error) {
      await fail(error);
      return;
    }
  }
  if (!active()) return;

  const heartbeat = setInterval(() => {
    void enqueue("agent-observer.heartbeat", {
      type: "heartbeat",
      status: "live",
      authority: "hwlab.event.v1",
      serverSentAt: new Date().toISOString(),
      valuesPrinted: false
    });
  }, policy.heartbeatMs);
  heartbeat.unref?.();
  cleanup.push(() => clearInterval(heartbeat));
}

function agentObserverPolicy(env) {
  const policy = {
    retentionEventLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_RETENTION_EVENT_LIMIT, "HWLAB_AGENT_OBSERVER_RETENTION_EVENT_LIMIT"),
    retentionScanLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_RETENTION_SCAN_LIMIT, "HWLAB_AGENT_OBSERVER_RETENTION_SCAN_LIMIT"),
    retentionTimeoutMs: positiveInteger(env.HWLAB_AGENT_OBSERVER_RETENTION_TIMEOUT_MS, "HWLAB_AGENT_OBSERVER_RETENTION_TIMEOUT_MS"),
    retentionGroupPrefix: requiredText(env.HWLAB_AGENT_OBSERVER_RETENTION_GROUP_PREFIX, "HWLAB_AGENT_OBSERVER_RETENTION_GROUP_PREFIX"),
    liveBufferLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT, "HWLAB_AGENT_OBSERVER_LIVE_BUFFER_LIMIT"),
    identityWindowLimit: positiveInteger(env.HWLAB_AGENT_OBSERVER_IDENTITY_WINDOW_LIMIT, "HWLAB_AGENT_OBSERVER_IDENTITY_WINDOW_LIMIT"),
    heartbeatMs: positiveInteger(env.HWLAB_AGENT_OBSERVER_SSE_HEARTBEAT_MS, "HWLAB_AGENT_OBSERVER_SSE_HEARTBEAT_MS")
  };
  if (policy.identityWindowLimit < policy.retentionEventLimit + policy.liveBufferLimit) {
    throw Object.assign(new Error("HWLAB_AGENT_OBSERVER_IDENTITY_WINDOW_LIMIT must cover retentionEventLimit + liveBufferLimit."), { code: "agent_observer_config_invalid" });
  }
  return policy;
}

function positiveInteger(value, name) {
  const number = Number(value);
  if (!Number.isInteger(number) || number <= 0) throw Object.assign(new Error(`${name} must be a positive integer.`), { code: "agent_observer_config_invalid" });
  return number;
}

function requiredText(value, name) {
  const result = text(value);
  if (!result) throw Object.assign(new Error(`${name} is required.`), { code: "agent_observer_config_invalid" });
  return result;
}

function text(value) {
  return typeof value === "string" && value.trim() ? value.trim() : null;
}

function observerError(code, message) {
  return { ok: false, error: { code, message, authority: "hwlab.event.v1", valuesRedacted: true } };
}

function sendJson(response, status, payload) {
  response.writeHead(status, { "content-type": "application/json; charset=utf-8", "cache-control": "no-store" });
  response.end(JSON.stringify(payload));
}

function waitForDrain(response, active) {
  return new Promise((resolve) => {
    const finish = () => {
      response.off?.("drain", finish);
      response.off?.("close", finish);
      resolve(active());
    };
    response.once("drain", finish);
    response.once("close", finish);
  });
}