fix(plugins): deliver onStreamComplete to disk-installed plugins (#11825)

changelog.d/fixes/11934-plugin-onstreamcomplete-delivery.md

@@ -0,0 +1 @@
+- **fix(plugins):** deliver the `onStreamComplete` event to disk-installed plugins. The event shipped in v3.8.50 (#9669) was emitted internally but had no plugin-facing wiring, so no plugin could ever subscribe: the manifest schema silently dropped `hooks.onStreamComplete`, and the loader/manager only knew the seven legacy hooks. `onStreamComplete` is now a declarable manifest hook, wired through the loader and registered by the manager like the other hooks, and its payload carries a `requestId` so consumers can correlate the stream-completion event with the originating request ([#11934](https://github.com/diegosouzapw/OmniRoute/pull/11934)) — thanks @amaleta

open-sse/handlers/chatCore.ts

@@ -5751,6 +5751,8 @@ export async function handleChatCore({
     });
 
     // Plugin onStreamComplete hook — fire-and-forget, fail-open (#9571)
+    // Pass traceId as requestId so plugins can correlate the stream-completion event
+    // with the originating request (the same id used for onRequest/onResponse). (#11825)
     runPluginOnStreamCompleteHook({
       status: normalizedStreamStatus,
       usage: streamUsage as Record<string, unknown> | undefined,
@@ -5759,6 +5761,7 @@ export async function handleChatCore({
       provider,
       errorCode: streamErrorCode,
       startTime,
+      requestId: traceId,
     });
   };
 

open-sse/handlers/chatCore/pluginOnResponse.ts

@@ -52,6 +52,9 @@ export async function runPluginOnResponseHook(args: {
  */
 export type PluginOnStreamCompletePayload = {
   status: number;
+  /** Correlates this stream-completion event with the originating request — the same
+   *  id passed to onRequest/onResponse for the request (chatCore's traceId). (#11825) */
+  requestId?: string;
   usage?: {
     prompt_tokens?: number;
     completion_tokens?: number;
@@ -81,11 +84,13 @@ export async function runPluginOnStreamCompleteHook(args: {
   provider: string | null | undefined;
   errorCode?: string | null | undefined;
   startTime: number;
+  requestId?: string;
 }): Promise<void> {
   try {
     const { runOnStreamComplete } = await import("@/lib/plugins/hooks");
     runOnStreamComplete({
       status: args.status,
+      requestId: args.requestId,
       usage: args.usage as PluginOnStreamCompletePayload["usage"],
       timing: {
         latencyMs: Date.now() - args.startTime,

src/lib/plugins/hooks.ts

@@ -264,6 +264,9 @@ export interface Plugin {
 
 export type PluginOnStreamCompletePayload = {
   status: number;
+  /** Correlates this stream-completion event with the originating request — the same
+   *  id passed to onRequest/onResponse for the request (chatCore's traceId). (#11825) */
+  requestId?: string;
   usage?: {
     prompt_tokens?: number;
     completion_tokens?: number;

src/lib/plugins/loader.ts

@@ -330,15 +330,20 @@ export async function loadPlugin(
     };
     registeredHooks.push("onError");
   }
-  // ── Lifecycle hooks (fire-and-forget, errors logged but don't block) ──
+  // ── Lifecycle + notification hooks (fire-and-forget, errors logged but don't block) ──
+  // onStreamComplete is wired here too: like the lifecycle hooks it is a one-way
+  // notification (no return value is consumed), so the same fire-and-forget IPC wrapper
+  // applies. Without this branch the event fires into an empty registry and is dropped
+  // for every disk-installed plugin (#11825).
   const lifecycleHooks: Array<{
-    key: "onInstall" | "onActivate" | "onDeactivate" | "onUninstall";
+    key: "onInstall" | "onActivate" | "onDeactivate" | "onUninstall" | "onStreamComplete";
     manifestFlag: boolean;
   }> = [
     { key: "onInstall", manifestFlag: manifest.hooks.onInstall },
     { key: "onActivate", manifestFlag: manifest.hooks.onActivate },
     { key: "onDeactivate", manifestFlag: manifest.hooks.onDeactivate },
     { key: "onUninstall", manifestFlag: manifest.hooks.onUninstall },
+    { key: "onStreamComplete", manifestFlag: manifest.hooks.onStreamComplete },
   ];
 
   for (const { key, manifestFlag } of lifecycleHooks) {

src/lib/plugins/manager.ts

@@ -37,6 +37,7 @@ type LifecycleHookName = Extract<
   | "onActivate"
   | "onDeactivate"
   | "onUninstall"
+  | "onStreamComplete"
 >;
 
 /**
@@ -222,6 +223,7 @@ class PluginManager {
         manifest.hooks.onActivate && "onActivate",
         manifest.hooks.onDeactivate && "onDeactivate",
         manifest.hooks.onUninstall && "onUninstall",
+        manifest.hooks.onStreamComplete && "onStreamComplete",
       ].filter(Boolean) as string[],
       permissions: manifest.requires.permissions,
       pluginDir: destDir,
@@ -352,6 +354,7 @@ class PluginManager {
         manifest.hooks.onActivate && "onActivate",
         manifest.hooks.onDeactivate && "onDeactivate",
         manifest.hooks.onUninstall && "onUninstall",
+        manifest.hooks.onStreamComplete && "onStreamComplete",
       ].filter(Boolean) as string[],
       permissions: manifest.requires.permissions,
       pluginDir: destDir,
@@ -409,6 +412,7 @@ class PluginManager {
         "onActivate",
         "onDeactivate",
         "onUninstall",
+        "onStreamComplete",
       ];
       for (const hookName of hookNames) {
         const handler = loaded.plugin[hookName];
@@ -529,6 +533,7 @@ class PluginManager {
               discovered.manifest.hooks.onActivate && "onActivate",
               discovered.manifest.hooks.onDeactivate && "onDeactivate",
               discovered.manifest.hooks.onUninstall && "onUninstall",
+              discovered.manifest.hooks.onStreamComplete && "onStreamComplete",
             ].filter(Boolean) as string[],
             permissions: discovered.manifest.requires.permissions,
             pluginDir: discovered.pluginDir,

src/lib/plugins/manifest.ts

@@ -43,6 +43,7 @@ export const HooksSchema = z.object({
   onActivate: z.boolean().optional(),
   onDeactivate: z.boolean().optional(),
   onUninstall: z.boolean().optional(),
+  onStreamComplete: z.boolean().optional(),
 });
 
 // ── Requires ──
@@ -105,6 +106,7 @@ export interface PluginManifestWithDefaults extends PluginManifest {
     onActivate: boolean;
     onDeactivate: boolean;
     onUninstall: boolean;
+    onStreamComplete: boolean;
   };
   skills: ManifestSkill[];
   enabledByDefault: boolean;
@@ -130,6 +132,7 @@ export function applyDefaults(manifest: PluginManifest): PluginManifestWithDefau
       onActivate: manifest.hooks?.onActivate ?? false,
       onDeactivate: manifest.hooks?.onDeactivate ?? false,
       onUninstall: manifest.hooks?.onUninstall ?? false,
+      onStreamComplete: manifest.hooks?.onStreamComplete ?? false,
     },
     skills: manifest.skills ?? [],
     enabledByDefault: manifest.enabledByDefault ?? false,
@@ -159,9 +162,7 @@ export function safeValidateManifest(
 
 // ── Config validation ──
 
-export type ValidatePluginConfigResult =
-  | { valid: true }
-  | { valid: false; errors: string[] };
+export type ValidatePluginConfigResult = { valid: true } | { valid: false; errors: string[] };
 
 /**
  * Validate a config object against a ConfigField schema map.

tests/unit/plugins-onstreamcomplete-delivery-11825.test.ts

@@ -0,0 +1,151 @@
+// Regression test for #11825 — the `onStreamComplete` plugin event is emitted
+// internally (chatCore.ts → runPluginOnStreamCompleteHook → runOnStreamComplete →
+// emitHook) but never DELIVERED to disk-installed plugins, because the plugin-facing
+// wiring is missing at three layers:
+//   1. src/lib/plugins/manifest.ts   — HooksSchema has no `onStreamComplete` field, so
+//      `hooks.onStreamComplete: true` in plugin.json is silently stripped by Zod.
+//   2. src/lib/plugins/loader.ts     — loadPlugin() never builds an IPC wrapper for it.
+//   3. src/lib/plugins/manager.ts    — activate()'s hardcoded `hookNames` list omits it,
+//      so even a wired handler is never registerHook()'d into the registry.
+// The pre-existing green tests (tests/unit/chatcore-plugin-onresponse.test.ts) call
+// `registerHook("onStreamComplete", ...)` directly — a path real plugins cannot use — so
+// they never exercised this delivery gap. This test drives the *real* plugin path:
+// install → activate → emit, and asserts a plugin process actually receives the payload.
+//
+// Secondary gap (#11825): the payload must carry a `requestId` so a consumer can join the
+// stream-completion event to the originating request. We assert the plugin sees it.
+import { test, describe, beforeEach } from "node:test";
+import assert from "node:assert/strict";
+import { mkdtempSync, writeFileSync, mkdirSync, rmSync, existsSync, readFileSync } from "node:fs";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+
+const { validateManifest } = await import("../../src/lib/plugins/manifest.ts");
+const mgr = await import("../../src/lib/plugins/manager.ts");
+const db = await import("../../src/lib/db/plugins.ts");
+const { getDbInstance } = await import("../../src/lib/db/core.ts");
+const { runPluginOnStreamCompleteHook } =
+  await import("../../open-sse/handlers/chatCore/pluginOnResponse.ts");
+
+async function waitFor(pred: () => boolean, timeoutMs = 6000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline && !pred()) {
+    await new Promise((r) => setTimeout(r, 25));
+  }
+}
+
+// ── Gap 1: manifest schema must preserve the flag ──────────────────────────────
+
+test("#11825 manifest: HooksSchema preserves hooks.onStreamComplete", () => {
+  const m = validateManifest({
+    name: "sc-manifest",
+    version: "1.0.0",
+    hooks: { onResponse: true, onStreamComplete: true },
+  });
+  // On the buggy code Zod strips the unknown key and applyDefaults never sets it, so this
+  // is `undefined` instead of `true` — plugin.json can't opt into the event at all.
+  assert.equal(
+    m.hooks.onStreamComplete,
+    true,
+    "manifest.hooks.onStreamComplete must survive validation so plugin.json can subscribe"
+  );
+});
+
+// ── Gaps 2 + 3 + requestId: real end-to-end delivery to a plugin process ────────
+
+describe("#11825 onStreamComplete is delivered to an installed+activated plugin", () => {
+  const NAME = "sc-delivery-11825";
+  const outFile = join(tmpdir(), `omniroute-onstreamcomplete-11825-${process.pid}.json`);
+
+  beforeEach(() => {
+    getDbInstance(); // ensure the plugins table migration has run
+    try {
+      db.deletePlugin(NAME);
+    } catch {}
+    try {
+      rmSync(outFile, { force: true });
+    } catch {}
+  });
+
+  test("plugin's onStreamComplete handler fires with usage/timing and a requestId", async (t) => {
+    const tmp = mkdtempSync(join(tmpdir(), "sc-11825-"));
+    const pluginDir = join(tmp, NAME);
+    mkdirSync(pluginDir, { recursive: true });
+    writeFileSync(
+      join(pluginDir, "plugin.json"),
+      JSON.stringify({
+        name: NAME,
+        version: "1.0.0",
+        main: "index.js",
+        hooks: { onStreamComplete: true },
+      })
+    );
+    // The plugin runs in an isolated child process, so it reports what it received by
+    // writing the payload to a file whose path is baked into its source.
+    writeFileSync(
+      join(pluginDir, "index.js"),
+      `const fs = require("fs");
+const OUT = ${JSON.stringify(outFile)};
+module.exports = {
+  onStreamComplete: async (payload) => { fs.writeFileSync(OUT, JSON.stringify(payload)); },
+};
+`
+    );
+
+    t.after(async () => {
+      await mgr.pluginManager.deactivate(NAME).catch(() => {});
+      try {
+        db.deletePlugin(NAME);
+      } catch {}
+      rmSync(tmp, { recursive: true, force: true });
+      rmSync(outFile, { force: true });
+    });
+
+    await mgr.pluginManager.install(pluginDir);
+    await mgr.pluginManager.activate(NAME);
+
+    // Sanity: the DB should record the declared hook (also broken on the buggy path).
+    const row = db.getPluginByName(NAME);
+    assert.ok(row, "plugin should be installed");
+    assert.ok(
+      (row!.hooks as unknown as string[]).includes("onStreamComplete"),
+      `installed plugin should declare onStreamComplete; got ${JSON.stringify(row!.hooks)}`
+    );
+
+    // Drive the REAL producer entry point used by chatCore.ts.
+    const startTime = Date.now() - 500;
+    await runPluginOnStreamCompleteHook({
+      status: 200,
+      usage: { prompt_tokens: 11, completion_tokens: 22, reasoning_tokens: 3 },
+      ttft: 120,
+      model: "claude-3-opus",
+      provider: "anthropic",
+      errorCode: undefined,
+      startTime,
+      requestId: "trace-11825",
+    });
+
+    await waitFor(() => existsSync(outFile));
+    assert.ok(
+      existsSync(outFile),
+      "the plugin's onStreamComplete handler never ran — the event was dropped into an " +
+        "empty registry (no manifest/loader/manager wiring reaches disk-installed plugins)"
+    );
+
+    const payload = JSON.parse(readFileSync(outFile, "utf-8")) as Record<string, unknown>;
+    assert.equal(payload.status, 200);
+    assert.equal((payload.usage as Record<string, number>).prompt_tokens, 11);
+    assert.equal((payload.usage as Record<string, number>).completion_tokens, 22);
+    assert.ok(payload.timing, "timing should be present");
+    assert.equal((payload.timing as Record<string, number>).ttft, 120);
+    assert.ok((payload.timing as Record<string, number>).latencyMs >= 450);
+    assert.equal(payload.model, "claude-3-opus");
+    assert.equal(payload.provider, "anthropic");
+    // Secondary gap: the event must be correlatable to the originating request.
+    assert.equal(
+      payload.requestId,
+      "trace-11825",
+      "onStreamComplete payload must carry requestId so it can be joined to the request"
+    );
+  });
+});