test(e2e): deep chat-harness coverage + streaming mock LLM + rust-e2e Linux lane (#1892)

scripts/mock-api/routes/llm.mjs

@@ -1,6 +1,327 @@
-import { json } from "../http.mjs";
+import { json, setCors } from "../http.mjs";
 import { behavior, parseBehaviorJson, setMockBehavior } from "../state.mjs";
 
+// ── Streaming helpers ─────────────────────────────────────────────
+//
+// When the agent harness calls the OpenAI-compatible endpoint with
+// `stream: true`, openhuman/providers/compatible.rs expects SSE chunks
+// shaped like:
+//
+//   data: {"choices":[{"delta":{"content":"hello"},"finish_reason":null}]}
+//   data: {"choices":[{"delta":{},"finish_reason":"stop"}], "usage":{...}}
+//   data: [DONE]
+//
+// The streaming branch is configured via two mock behavior keys:
+//
+//   llmStreamScript        — JSON array of script entries (see below).
+//                            Overrides everything else when present.
+//   llmStreamChunkDelayMs  — default delay between chunks (ms).
+//
+// Script entry shapes:
+//   { "text": "Hello",      "delayMs": 30 }                       text delta
+//   { "thinking": "...",    "delayMs": 30 }                       reasoning delta
+//   { "toolCall": { "id": "call_x", "name": "foo", "arguments": "{\"a\":1}" } }
+//                                                                 emits a tool_call
+//                                                                 start chunk plus
+//                                                                 incremental args
+//                                                                 chunks (split by
+//                                                                 8-char windows)
+//   { "finish": "stop" | "tool_calls" }                           final empty chunk
+//   { "usage": {"prompt_tokens":1,"completion_tokens":2,"total_tokens":3} }
+//                                                                 attached to the
+//                                                                 last emitted chunk
+//   { "error": "kaboom" }                                         emits an error
+//                                                                 SSE event and
+//                                                                 closes the
+//                                                                 connection (no
+//                                                                 [DONE])
+//
+// If no `llmStreamScript` is set, a keyword rule matching the latest
+// user message is auto-converted into a script. If no rule matches we
+// stream a default greeting in three deltas so basic streaming UI
+// behavior is exercised even with zero configuration.
+
+function writeSseHead(res) {
+  setCors(res);
+  res.writeHead(200, {
+    "Content-Type": "text/event-stream; charset=utf-8",
+    "Cache-Control": "no-cache",
+    Connection: "keep-alive",
+  });
+}
+
+function sseChunkEnvelope({
+  model,
+  contentDelta,
+  thinkingDelta,
+  toolCallDelta,
+  finishReason,
+  usage,
+}) {
+  const choice = {
+    index: 0,
+    delta: {},
+    finish_reason: finishReason ?? null,
+  };
+  if (typeof contentDelta === "string") choice.delta.content = contentDelta;
+  if (typeof thinkingDelta === "string")
+    choice.delta.reasoning_content = thinkingDelta;
+  if (toolCallDelta) choice.delta.tool_calls = [toolCallDelta];
+
+  const envelope = {
+    id: `chatcmpl-mock-${Date.now()}`,
+    object: "chat.completion.chunk",
+    created: Math.floor(Date.now() / 1000),
+    model: model || "e2e-mock-model",
+    choices: [choice],
+  };
+  if (usage) envelope.usage = usage;
+  return envelope;
+}
+
+function writeSseEvent(res, payload) {
+  res.write(`data: ${JSON.stringify(payload)}\n\n`);
+}
+
+function sleep(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+// Split a string into N-character windows so we can stream tool-call
+// argument JSON the same way real providers do — clients accumulate the
+// partial fragments and JSON-parse at the end.
+function chunkString(s, windowSize) {
+  const out = [];
+  if (!s) return out;
+  for (let i = 0; i < s.length; i += windowSize) {
+    out.push(s.slice(i, i + windowSize));
+  }
+  return out;
+}
+
+function defaultStreamScript({ content, toolCalls }) {
+  const script = [];
+  // Real OpenAI streams a text preamble (when present) BEFORE tool-call
+  // deltas; collapsing that to nothing the moment tool_calls show up
+  // would diverge from the non-streaming `{ content, toolCalls }`
+  // contract and silently drop assistant-visible reasoning.
+  const text =
+    typeof content === "string" && content.length > 0 ? content : null;
+
+  if (Array.isArray(toolCalls) && toolCalls.length > 0) {
+    if (text) {
+      for (const piece of chunkString(text, 12)) {
+        script.push({ text: piece });
+      }
+    }
+    for (let i = 0; i < toolCalls.length; i += 1) {
+      const tc = toolCalls[i];
+      script.push({
+        toolCall: {
+          // `index` is what the OpenAI streaming protocol uses to
+          // demux multiple parallel tool calls. Preserve it here so a
+          // single-script entry with N tool calls becomes N distinct
+          // calls on the client side instead of being reassembled
+          // into one.
+          index: i,
+          id: tc.id ?? `call_stream_${i}`,
+          name: String(tc.name ?? ""),
+          arguments:
+            typeof tc.arguments === "string"
+              ? tc.arguments
+              : JSON.stringify(tc.arguments ?? {}),
+        },
+      });
+    }
+    script.push({ finish: "tool_calls" });
+    return script;
+  }
+  const fallbackText = text ?? "Hello from e2e mock agent";
+  // Split into ~12-char windows so a UI-side delta watcher sees several
+  // arrival events even for short responses.
+  for (const piece of chunkString(fallbackText, 12)) {
+    script.push({ text: piece });
+  }
+  script.push({ finish: "stop" });
+  return script;
+}
+
+function handleStreamingCompletion({ res, model, mockBehavior, parsedBody }) {
+  writeSseHead(res);
+
+  // 1. Explicit streaming script overrides everything.
+  let script = parseBehaviorJson("llmStreamScript", null);
+
+  if (!Array.isArray(script)) {
+    // 2. Forced queue: pop the next entry and convert it into a script.
+    const forced = parseBehaviorJson("llmForcedResponses", []);
+    if (Array.isArray(forced) && forced.length > 0) {
+      const next = forced.shift();
+      setMockBehavior("llmForcedResponses", JSON.stringify(forced));
+      script = defaultStreamScript({
+        content: next.content,
+        toolCalls: next.toolCalls,
+      });
+    }
+  }
+
+  if (!Array.isArray(script)) {
+    // 3. Keyword rules — match on latest user/tool message.
+    const rules = parseBehaviorJson("llmKeywordRules", []);
+    const probe = pickProbeText(parsedBody).toLowerCase();
+    if (Array.isArray(rules)) {
+      for (const rule of rules) {
+        if (!rule || typeof rule.keyword !== "string") continue;
+        if (probe.includes(rule.keyword.toLowerCase())) {
+          script = defaultStreamScript({
+            content: rule.content,
+            toolCalls: rule.toolCalls,
+          });
+          break;
+        }
+      }
+    }
+  }
+
+  if (!Array.isArray(script)) {
+    // 4. Default: stream a short greeting in a few chunks.
+    const fallback =
+      typeof mockBehavior.llmFallbackContent === "string" &&
+      mockBehavior.llmFallbackContent.length > 0
+        ? mockBehavior.llmFallbackContent
+        : "Hello from e2e mock agent";
+    script = defaultStreamScript({ content: fallback });
+  }
+
+  const defaultDelayMs = Number.isFinite(
+    parseFloat(mockBehavior.llmStreamChunkDelayMs),
+  )
+    ? Math.max(0, parseFloat(mockBehavior.llmStreamChunkDelayMs))
+    : 25;
+
+  // Fire-and-forget — the dispatcher only cares that the handler
+  // claimed the request. Errors mid-stream are surfaced through SSE.
+  streamScriptToResponse({ res, model, script, defaultDelayMs }).catch(
+    (err) => {
+      try {
+        writeSseEvent(res, {
+          error: { message: `mock stream error: ${err?.message ?? err}` },
+        });
+      } catch {
+        // ignore — connection likely already closed
+      }
+      try {
+        res.end();
+      } catch {
+        // ignore
+      }
+    },
+  );
+
+  return true;
+}
+
+async function streamScriptToResponse({ res, model, script, defaultDelayMs }) {
+  let trailingUsage = null;
+  for (let i = 0; i < script.length; i += 1) {
+    const entry = script[i] ?? {};
+    const delay = Number.isFinite(entry.delayMs) ? entry.delayMs : defaultDelayMs;
+    if (delay > 0) await sleep(delay);
+
+    if (entry.error) {
+      writeSseEvent(res, { error: { message: String(entry.error) } });
+      res.end();
+      return;
+    }
+
+    if (entry.usage && typeof entry.usage === "object") {
+      // Buffer usage until the next chunk that carries finish_reason.
+      trailingUsage = entry.usage;
+      continue;
+    }
+
+    if (typeof entry.text === "string") {
+      writeSseEvent(
+        res,
+        sseChunkEnvelope({ model, contentDelta: entry.text }),
+      );
+      continue;
+    }
+
+    if (typeof entry.thinking === "string") {
+      writeSseEvent(
+        res,
+        sseChunkEnvelope({ model, thinkingDelta: entry.thinking }),
+      );
+      continue;
+    }
+
+    if (entry.toolCall) {
+      const tc = entry.toolCall;
+      // Preserve the caller-supplied index when present. Real OpenAI
+      // streams use `index` to demux multiple parallel tool calls in
+      // the same message — collapsing every delta to `index: 0`
+      // breaks the multi-tool reassembly contract on the client.
+      const index = Number.isInteger(tc.index) ? tc.index : 0;
+      const id = tc.id ?? `call_stream_${i}`;
+      const name = String(tc.name ?? "");
+      const argsRaw =
+        typeof tc.arguments === "string"
+          ? tc.arguments
+          : JSON.stringify(tc.arguments ?? {});
+
+      // Opening chunk: carries id + name + first arg fragment.
+      const argPieces = chunkString(argsRaw, 8);
+      const first = argPieces.shift() ?? "";
+      writeSseEvent(
+        res,
+        sseChunkEnvelope({
+          model,
+          toolCallDelta: {
+            index,
+            id,
+            type: "function",
+            function: { name, arguments: first },
+          },
+        }),
+      );
+      for (const piece of argPieces) {
+        if (delay > 0) await sleep(delay);
+        writeSseEvent(
+          res,
+          sseChunkEnvelope({
+            model,
+            toolCallDelta: {
+              index,
+              function: { arguments: piece },
+            },
+          }),
+        );
+      }
+      continue;
+    }
+
+    if (entry.finish) {
+      writeSseEvent(
+        res,
+        sseChunkEnvelope({
+          model,
+          finishReason: entry.finish,
+          usage: trailingUsage ?? undefined,
+        }),
+      );
+      trailingUsage = null;
+      continue;
+    }
+  }
+
+  // Always close with the [DONE] sentinel — clients use it to detect
+  // graceful end-of-stream and clear in-flight state. Skipping it
+  // wedges the in-flight map until the next request lands.
+  res.write("data: [DONE]\n\n");
+  res.end();
+}
+
 /**
  * Smart mock LLM endpoint.
  *
@@ -106,6 +427,20 @@ export function handleLlmCompletions(ctx) {
   const model =
     typeof parsedBody?.model === "string" ? parsedBody.model : "e2e-mock-model";
 
+  // ── Streaming branch ────────────────────────────────────────────
+  // Drive the OpenAI SSE protocol when the caller requested it. The
+  // agent harness sets `stream: true` whenever it has a delta channel
+  // attached, which is the production code path — non-streaming is
+  // only the OH-backend fallback. See compatible.rs `chat()`.
+  if (parsedBody?.stream === true) {
+    return handleStreamingCompletion({
+      res,
+      model,
+      mockBehavior,
+      parsedBody,
+    });
+  }
+
   // 1. Forced queue — replay exact ChatResponse objects in order.
   const forced = parseBehaviorJson("llmForcedResponses", []);
   if (Array.isArray(forced) && forced.length > 0) {

scripts/test-rust-e2e.sh

@@ -0,0 +1,129 @@
+#!/usr/bin/env bash
+#
+# Rust E2E suite — the cargo-test counterpart to the Tauri E2E specs.
+#
+# Boots the mock backend (same `scripts/mock-api-server.mjs` the Tauri
+# E2E uses) on a fixed port and then runs each `tests/*_e2e.rs`
+# integration test against it. Tests that don't currently consume the
+# mock backend still run here so we keep one place to add new
+# mock-driven integration tests over time.
+#
+# This is invoked from:
+#   - `pnpm test:rust:e2e` (local dev + Docker)
+#   - `.github/workflows/e2e.yml` (the `rust-e2e-linux` job)
+#
+# Usage:
+#   ./scripts/test-rust-e2e.sh                       # all default e2e tests
+#   ./scripts/test-rust-e2e.sh --suite json_rpc_e2e  # one specific suite
+#   ./scripts/test-rust-e2e.sh -- --ignored          # extra cargo-test args
+#
+# Env knobs:
+#   MOCK_API_PORT  — mock backend port (default 18505).
+#   MOCK_LOG       — path for mock server stdout/stderr.
+#
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)"
+
+# The full set of `tests/*_e2e.rs` files. Each gets a `--test <name>` flag
+# in the single `cargo test` invocation so cargo compiles them in one
+# unit and only the test binaries that exist get run. Tests guarded by
+# `#[ignore]` stay skipped unless the caller passes `-- --ignored`.
+ALL_E2E_SUITES=(
+  agent_retrieval_e2e
+  autocomplete_memory_e2e
+  calendar_grounding_e2e
+  json_rpc_e2e
+  linux_cef_deb_runtime_e2e
+  live_routing_e2e
+  memory_graph_sync_e2e
+  memory_roundtrip_e2e
+  screen_intelligence_vision_e2e
+  subconscious_e2e
+)
+
+# Parse args: --suite <name> can be passed multiple times to filter.
+# Everything after `--` is forwarded to cargo test as test-binary args.
+SUITES=()
+EXTRA_ARGS=()
+while [ $# -gt 0 ]; do
+  case "$1" in
+    --suite)
+      # Guard against `set -u` blowing up on `--suite` with no argument
+      # — turning that into a clear usage error is friendlier than
+      # the cryptic "$2: unbound variable" from bash.
+      if [ $# -lt 2 ] || [ -z "${2:-}" ]; then
+        echo "[rust-e2e] ERROR: --suite requires a test name (e.g. --suite json_rpc_e2e)" >&2
+        exit 2
+      fi
+      SUITES+=("$2")
+      shift 2
+      ;;
+    --)
+      shift
+      EXTRA_ARGS+=("$@")
+      break
+      ;;
+    *)
+      EXTRA_ARGS+=("$1")
+      shift
+      ;;
+  esac
+done
+if [ "${#SUITES[@]}" -eq 0 ]; then
+  SUITES=("${ALL_E2E_SUITES[@]}")
+fi
+
+MOCK_API_PORT="${MOCK_API_PORT:-18505}"
+MOCK_API_URL="http://127.0.0.1:${MOCK_API_PORT}"
+MOCK_LOG="${MOCK_LOG:-/tmp/openhuman-rust-e2e-mock.log}"
+MOCK_PID=""
+
+cleanup() {
+  if [ -n "$MOCK_PID" ]; then
+    kill "$MOCK_PID" 2>/dev/null || true
+    wait "$MOCK_PID" 2>/dev/null || true
+  fi
+}
+trap cleanup EXIT
+
+echo "[rust-e2e] Starting mock API server on ${MOCK_API_URL} ..."
+node "$SCRIPT_DIR/mock-api-server.mjs" --port "$MOCK_API_PORT" >"$MOCK_LOG" 2>&1 &
+MOCK_PID=$!
+
+for i in $(seq 1 30); do
+  if curl -sf "${MOCK_API_URL}/__admin/health" >/dev/null 2>&1; then
+    break
+  fi
+  if [ "$i" -eq 30 ]; then
+    echo "[rust-e2e] ERROR: mock API server did not become healthy in time." >&2
+    echo "[rust-e2e] See logs: $MOCK_LOG" >&2
+    exit 1
+  fi
+  sleep 1
+done
+echo "[rust-e2e] Mock backend healthy."
+
+export BACKEND_URL="$MOCK_API_URL"
+export VITE_BACKEND_URL="$MOCK_API_URL"
+
+cd "$REPO_ROOT"
+source "$HOME/.cargo/env" 2>/dev/null || true
+
+# Assemble the `--test <name>` flags so a single `cargo test` invocation
+# compiles + runs every suite. Cargo will fail fast if any --test binary
+# doesn't exist, which is the signal you want when a suite gets renamed.
+CARGO_FLAGS=()
+for suite in "${SUITES[@]}"; do
+  CARGO_FLAGS+=(--test "$suite")
+done
+
+echo "[rust-e2e] Running:"
+if [ "${#EXTRA_ARGS[@]}" -gt 0 ]; then
+  echo "[rust-e2e]   cargo test --manifest-path Cargo.toml ${CARGO_FLAGS[*]} -- ${EXTRA_ARGS[*]}"
+  cargo test --manifest-path Cargo.toml "${CARGO_FLAGS[@]}" -- "${EXTRA_ARGS[@]}"
+else
+  echo "[rust-e2e]   cargo test --manifest-path Cargo.toml ${CARGO_FLAGS[*]}"
+  cargo test --manifest-path Cargo.toml "${CARGO_FLAGS[@]}"
+fi