QVAC-20580 feat[api]: add subscribeServerLogs to capture all server logs

Signed-off-by: Arun Mani J <j.arunmani@proton.me>

Author: arun-mani-j

packages/sdk/client/api/index.ts

@@ -4,6 +4,7 @@ export { completion } from "./completion-stream";
 export { deleteCache } from "./delete-cache";
 export { unloadModel } from "./unload-model";
 export { loggingStream } from "./logging-stream";
+export { subscribeServerLogs, type ServerLogHandler } from "./subscribe-logs";
 export { heartbeat } from "./heartbeat";
 export { transcribe, transcribeStream } from "./transcribe";
 export { bciTranscribe, bciTranscribeStream } from "./bci-transcribe";

packages/sdk/client/api/subscribe-logs.ts

@@ -0,0 +1,38 @@
+import { getClientLogger, SDK_ALL_LOG_ID } from "@/logging";
+import { loggingStream } from "./logging-stream";
+import type { LoggingStreamResponse } from "@/schemas/logging-stream";
+
+const logger = getClientLogger();
+
+export type ServerLogHandler = (log: LoggingStreamResponse) => void;
+
+/**
+ * Subscribes to all server-side SDK logs (SDK, models, RAG, …) through a single
+ * stream and returns a function that stops the subscription.
+ *
+ * @example
+ * ```typescript
+ * const unsubscribe = subscribeServerLogs((log) => {
+ *   console.log(`[${log.level}] ${log.namespace}: ${log.message}`);
+ * });
+ * // later
+ * unsubscribe();
+ * ```
+ */
+export function subscribeServerLogs(handler: ServerLogHandler) {
+  const streamIterator = loggingStream({ id: SDK_ALL_LOG_ID });
+
+  void (async () => {
+    try {
+      for await (const log of streamIterator) {
+        handler(log);
+      }
+    } catch (error) {
+      logger.error("Server log stream error:", error);
+    }
+  })();
+
+  return () => {
+    void streamIterator.return(undefined);
+  };
+}

packages/sdk/examples/config/logging/logging.config.json

@@ -0,0 +1,3 @@
+{
+  "loggerConsoleOutput": false
+}

packages/sdk/examples/logging-global.ts

@@ -0,0 +1,40 @@
+// loggerConsoleOutput: false in this config disables the default console
+// transport on both the client and the worker, so the only logs printed are
+// the ones our handler below receives.
+const configDir = import.meta.dirname ?? process.cwd();
+process.env["QVAC_CONFIG_PATH"] = `${configDir}/config/logging/logging.config.json`;
+
+const { loadModel, completion, unloadModel, subscribeServerLogs, LLAMA_3_2_1B_INST_Q4_0 } =
+  await import("@qvac/sdk");
+
+try {
+  console.log("🚀 Starting global logging demo...\n");
+
+  // One subscription captures every server-side log (SDK, models, RAG, …)
+  // without having to know any stream IDs ahead of time.
+  const unsubscribe = subscribeServerLogs((log) => {
+    console.log(`[${log.level.toUpperCase()}] [${log.namespace}] ${log.message}`);
+  });
+
+  const modelId = await loadModel({
+    modelSrc: LLAMA_3_2_1B_INST_Q4_0,
+    modelConfig: { ctx_size: 2048 },
+  });
+
+  const result = completion({
+    modelId,
+    history: [{ role: "user", content: "Count from 1 to 5." }],
+    stream: true,
+  });
+
+  console.log("📝 Response:\n");
+  for await (const token of result.tokenStream) {
+    process.stdout.write(token);
+  }
+
+  await unloadModel({ modelId, clearStorage: false });
+  unsubscribe();
+} catch (error) {
+  console.error("❌ Error:", error);
+  process.exit(1);
+}

packages/sdk/index.ts

@@ -32,6 +32,8 @@ export {
   getModelInfo,
   getLoadedModelInfo,
   loggingStream,
+  subscribeServerLogs,
+  type ServerLogHandler,
   ocr,
   invokePlugin,
   invokePluginStream,
@@ -224,7 +226,7 @@ export {
 } from "./utils/errors-client";
 
 // Logging exports
-export { getLogger, SDK_LOG_ID } from "./logging";
+export { getLogger, SDK_LOG_ID, SDK_ALL_LOG_ID } from "./logging";
 export type { Logger, LogTransport, LoggerOptions } from "./logging";
 
 // Profiler exports

packages/sdk/logging/index.ts

@@ -12,6 +12,7 @@ export type { Logger, LoggerOptions, LogTransport } from "./types";
 export {
   RAG_NAMESPACE,
   SDK_LOG_ID,
+  SDK_ALL_LOG_ID,
   SDK_SERVER_NAMESPACE,
   type AddonNamespace,
 } from "./namespaces";

packages/sdk/logging/namespaces.ts

@@ -7,5 +7,8 @@ export type AddonNamespace = CanonicalModelType | typeof RAG_NAMESPACE;
 // Reserved ID for SDK server logs
 export const SDK_LOG_ID = "__sdk__";
 
+// Reserved ID for the stream that receives all server-side logs
+export const SDK_ALL_LOG_ID = "__all__";
+
 // Namespace for all SDK server logs
 export const SDK_SERVER_NAMESPACE = "sdk:server";

packages/sdk/server/bare/registry/logging-stream-registry.ts

@@ -12,6 +12,7 @@
  */
 
 import type { LogLevel } from "@qvac/logging";
+import { SDK_ALL_LOG_ID } from "@/logging/namespaces";
 
 const loggingStreams = new Map<
   string,
@@ -111,6 +112,18 @@ export function sendLogToStreams(
   level: LogLevel,
   namespace: string,
   message: string,
+) {
+  deliverToStream(id, level, namespace, message);
+  if (id !== SDK_ALL_LOG_ID) {
+    deliverToStream(SDK_ALL_LOG_ID, level, namespace, message);
+  }
+}
+
+function deliverToStream(
+  id: string,
+  level: LogLevel,
+  namespace: string,
+  message: string,
 ) {
   const streams = loggingStreams.get(id);
   const isBuffering = modelsWithBuffering.has(id);

packages/sdk/server/worker-core.ts

@@ -13,7 +13,12 @@ import {
   clearAllLoggingStreams,
   startLogBuffering,
 } from "@/server/bare/registry/logging-stream-registry";
-import { clearAllAddonLoggers, getServerLogger, SDK_LOG_ID } from "@/logging";
+import {
+  clearAllAddonLoggers,
+  getServerLogger,
+  SDK_LOG_ID,
+  SDK_ALL_LOG_ID,
+} from "@/logging";
 import { clearPlugins } from "@/server/plugins";
 import {
   acquireWorkerLock,
@@ -68,6 +73,7 @@ export function initializeWorkerCore(): { hasRPCConfig: boolean } {
   }
 
   startLogBuffering(SDK_LOG_ID);
+  startLogBuffering(SDK_ALL_LOG_ID);
 
   const { hasRPCConfig } = initEnv();
 

packages/sdk/test/unit/logging-stream-global.test.ts

@@ -0,0 +1,63 @@
+import test from "brittle";
+import { SDK_ALL_LOG_ID, SDK_LOG_ID } from "@/logging";
+import {
+  registerLoggingStream,
+  unregisterLoggingStream,
+  sendLogToStreams,
+  startLogBuffering,
+  clearAllLoggingStreams,
+} from "@/server/bare/registry/logging-stream-registry";
+
+test("global stream receives logs from every source id", (t) => {
+  clearAllLoggingStreams();
+
+  const received: string[] = [];
+  const handler = (_level: string, _ns: string, message: string) =>
+    received.push(message);
+  registerLoggingStream(SDK_ALL_LOG_ID, handler);
+
+  sendLogToStreams(SDK_LOG_ID, "info", "sdk:server", "from sdk");
+  sendLogToStreams("model-123", "debug", "llamacpp-completion", "from model");
+
+  t.alike(received, ["from sdk", "from model"], "captures logs across ids");
+
+  unregisterLoggingStream(SDK_ALL_LOG_ID, handler);
+  sendLogToStreams(SDK_LOG_ID, "info", "sdk:server", "after unsubscribe");
+  t.alike(received, ["from sdk", "from model"], "stops after unsubscribe");
+
+  clearAllLoggingStreams();
+});
+
+test("per-id stream still receives only its own logs", (t) => {
+  clearAllLoggingStreams();
+
+  const global: string[] = [];
+  const model: string[] = [];
+  const globalHandler = (_l: string, _n: string, m: string) => global.push(m);
+  const modelHandler = (_l: string, _n: string, m: string) => model.push(m);
+
+  registerLoggingStream(SDK_ALL_LOG_ID, globalHandler);
+  registerLoggingStream("model-123", modelHandler);
+
+  sendLogToStreams("model-123", "info", "llamacpp-completion", "a");
+  sendLogToStreams("model-456", "info", "llamacpp-completion", "b");
+
+  t.alike(model, ["a"], "model stream only sees its own id");
+  t.alike(global, ["a", "b"], "global stream sees both");
+
+  clearAllLoggingStreams();
+});
+
+test("global stream flushes startup logs buffered before subscribe", (t) => {
+  clearAllLoggingStreams();
+  startLogBuffering(SDK_ALL_LOG_ID);
+
+  sendLogToStreams(SDK_LOG_ID, "info", "sdk:server", "early");
+
+  const received: string[] = [];
+  registerLoggingStream(SDK_ALL_LOG_ID, (_l, _n, m) => received.push(m));
+
+  t.alike(received, ["early"], "buffered log delivered on subscribe");
+
+  clearAllLoggingStreams();
+});