fix: honor composer prediction deadlines and cancel on send

docs/platformclaw/web-ingress-runtime.md

@@ -24,6 +24,26 @@ in `plugins.entries.admin-http-rpc.config.composerPredictionsEnabled`, then rest
 the development Gateway. Each employee can enable or disable their own preference;
 it persists in the plugin's SQLite keyed store without a schema-version change.
 
+Set `composerPredictionsTimeoutMs` in the same plugin config to control how long a
+prediction may wait. The default is 30,000 ms; supported values are integers from
+1,000 to 300,000 ms. For example, merge these values into the existing
+`plugins.entries.admin-http-rpc.config` object:
+
+```json
+{
+  "composerPredictionsEnabled": true,
+  "composerPredictionsTimeoutMs": 60000
+}
+```
+
+The server gate and each employee's preference still control whether predictions
+run. The timeout does not enable the feature or change saved account preferences.
+The Gateway advertises its configured deadline to the Web UI and BFF; inference
+and its internal HTTP transport use that budget instead of separate short limits.
+Settings lookups retain a separate ten-second control-request deadline.
+Refresh the Web page after updating the Gateway config so it reads the new budget.
+An external reverse proxy must also allow responses to wait for that budget.
+
 The browser waits one second after a completed response while the composer is
 empty. It requests `/platformclaw/api/composer-predictions` using the employee
 cookie and same-origin policy. The BFF reads history through the existing personal
@@ -55,15 +75,17 @@ Normal conversation defaults and fallback settings remain unchanged.
 Inference receives only the last six user/assistant text messages, each capped
 at 300 Unicode characters. This bounded context supports two or three exchanges
 without copying attachments, reasoning, tool output, or an unbounded transcript.
-Output is capped at 64 tokens and 300 characters. The whole browser/BFF request
-has a ten-second deadline, including Gateway history access. There is one active
+Output is capped at 64 tokens and 300 characters. The browser/BFF prediction deadline
+includes Gateway history access. There is one active
 prediction per Gateway process and no prediction queue. Ordinary model calls
 preempt predictions. No vLLM load metrics are assumed; the single-slot policy is
 the conservative admission control. A multi-Gateway deployment needs a shared
 admission owner before enabling this feature across instances.
 
 Tab inserts Ghost Text without sending. Escape, typing, IME composition, session
-changes, and a new response invalidate it. Enter retains its existing behavior.
+changes, and a new response invalidate it. Sending a new browser message cancels
+pending prediction work immediately, before waiting for the send acknowledgment
+or the next model call. Enter retains its existing behavior.
 Cancellation closes the HTTP request to the model; actual GPU cancellation
 depends on the company's serving stack and must be measured there. A server
 that continues inference after disconnect can still consume up to the 64-token

extensions/admin-http-rpc/index.test.ts

@@ -1,5 +1,6 @@
 // Admin Http Rpc tests cover index plugin behavior.
 import { createServer } from "node:http";
+import { validateJsonSchemaValue } from "openclaw/plugin-sdk/json-schema-runtime";
 import { afterEach, describe, expect, it, vi } from "vitest";
 import plugin from "./index.js";
 import manifest from "./openclaw.plugin.json" with { type: "json" };
@@ -22,6 +23,36 @@ afterEach(() => {
 });
 
 describe("admin-http-rpc plugin entry", () => {
+  const validateConfig = (value: unknown) =>
+    validateJsonSchemaValue({
+      schema: manifest.configSchema,
+      cacheKey: "admin-http-rpc.timeout-contract",
+      value,
+      applyDefaults: true,
+    });
+
+  it("defaults the inference timeout while retaining explicit server opt-in", () => {
+    expect(validateConfig({})).toEqual({
+      ok: true,
+      value: { composerPredictionsEnabled: false, composerPredictionsTimeoutMs: 30_000 },
+    });
+    expect(validateConfig({ composerPredictionsEnabled: true })).toEqual({
+      ok: true,
+      value: { composerPredictionsEnabled: true, composerPredictionsTimeoutMs: 30_000 },
+    });
+  });
+
+  it.each([1_000, 300_000])("accepts the configured timeout boundary %i", (timeoutMs) => {
+    expect(validateConfig({ composerPredictionsTimeoutMs: timeoutMs })).toEqual({
+      ok: true,
+      value: { composerPredictionsEnabled: false, composerPredictionsTimeoutMs: timeoutMs },
+    });
+  });
+
+  it.each([999, 300_001, 1_000.5, "30000"])("rejects an invalid timeout %j", (timeoutMs) => {
+    expect(validateConfig({ composerPredictionsTimeoutMs: timeoutMs }).ok).toBe(false);
+  });
+
   it("stays startup-off until the plugin entry is explicitly enabled", () => {
     expect(manifest.activation).toEqual({
       onStartup: false,
@@ -156,7 +187,14 @@ describe("admin-http-rpc plugin entry", () => {
         config: {
           current: () => ({
             plugins: {
-              entries: { "admin-http-rpc": { config: { composerPredictionsEnabled: true } } },
+              entries: {
+                "admin-http-rpc": {
+                  config: {
+                    composerPredictionsEnabled: true,
+                    composerPredictionsTimeoutMs: 45_000,
+                  },
+                },
+              },
             },
             models: {
               providers: {
@@ -210,7 +248,11 @@ describe("admin-http-rpc plugin entry", () => {
         expect(response.ok).toBe(true);
         return response.json();
       };
-      await post({ action: "preferences", userId: "user-1", enabled: true });
+      expect(await post({ action: "preferences", userId: "user-1", enabled: true })).toEqual({
+        id: expect.any(String),
+        ok: true,
+        payload: { available: true, enabled: true, timeoutMs: 45_000 },
+      });
       expect(stores.get("platformclaw.composer-preferences")?.get("user-1")).toBe(true);
       expect(stores.get("platformclaw.employee-profiles")?.has("user-1")).toBe(false);
       mocks.lookup.mockResolvedValue([{ address: "10.1.2.3", family: 4 }]);

extensions/admin-http-rpc/openclaw.plugin.json

@@ -17,6 +17,13 @@
         "type": "boolean",
         "default": false,
         "description": "Enable Web-only, account-opt-in predictions using the model selected by PLATFORMCLAW_COMPOSER_PREDICTIONS_MODEL."
+      },
+      "composerPredictionsTimeoutMs": {
+        "type": "integer",
+        "default": 30000,
+        "minimum": 1000,
+        "maximum": 300000,
+        "description": "Maximum duration in milliseconds for composer prediction inference. Normal chat retains priority."
       }
     }
   }

extensions/admin-http-rpc/src/composer-predictions.test.ts

@@ -11,8 +11,12 @@ vi.mock("openclaw/plugin-sdk/ssrf-runtime", () => ({
 
 function setup(providerId = "codemate", modelId = "codellm-fast", apiType = "openai-completions") {
   const values = new Map<string, boolean>();
+  const pluginConfig: {
+    composerPredictionsEnabled?: boolean;
+    composerPredictionsTimeoutMs?: number;
+  } = { composerPredictionsEnabled: true };
   const config = {
-    plugins: { entries: { "admin-http-rpc": { config: { composerPredictionsEnabled: true } } } },
+    plugins: { entries: { "admin-http-rpc": { config: pluginConfig } } },
     models: {
       providers: {
         [providerId]: {
@@ -52,7 +56,20 @@ function setup(providerId = "codemate", modelId = "codellm-fast", apiType = "ope
     ],
   };
   const predict = (signal = new AbortController().signal) => call(params, signal);
-  return { call, predict, composer, values, params };
+  return { call, predict, composer, values, params, pluginConfig };
+}
+
+function useDeadlineClock() {
+  vi.useFakeTimers();
+  // Node's native AbortSignal.timeout bypasses the global timers replaced by Vitest.
+  return vi.spyOn(AbortSignal, "timeout").mockImplementation((milliseconds) => {
+    const deadline = new AbortController();
+    setTimeout(
+      () => deadline.abort(new DOMException("Prediction timed out", "TimeoutError")),
+      milliseconds,
+    );
+    return deadline.signal;
+  });
 }
 beforeEach(() => {
   vi.stubEnv("PLATFORMCLAW_COMPOSER_PREDICTIONS_MODEL", undefined);
@@ -61,6 +78,7 @@ beforeEach(() => {
 afterEach(() => {
   vi.unstubAllEnvs();
   vi.clearAllMocks();
+  vi.restoreAllMocks();
   vi.useRealTimers();
 });
 
@@ -69,7 +87,16 @@ describe("Explicit isolated prediction inference", () => {
     const { call, predict, params } = setup();
     expect(await predict()).toEqual({ text: null });
     const signal = new AbortController().signal;
-    await call({ action: "preferences", userId: "user-1", enabled: true }, signal);
+    expect(await call({ action: "preferences", userId: "user-1" }, signal)).toEqual({
+      available: true,
+      enabled: false,
+      timeoutMs: 30_000,
+    });
+    expect(await call({ action: "preferences", userId: "user-1", enabled: true }, signal)).toEqual({
+      available: true,
+      enabled: true,
+      timeoutMs: 30_000,
+    });
     expect(await call({ ...params, userId: "user-2" }, signal)).toEqual({ text: null });
     expect(await predict()).toEqual({ text: "다음 단계는 무엇인가요?" });
     const request = mocks.fetch.mock.calls[0]?.[0];
@@ -79,6 +106,129 @@ describe("Explicit isolated prediction inference", () => {
     expect(request.capture).toBe(false);
     expect(mocks.fetch).toHaveBeenCalledTimes(1);
   });
+  it.each([
+    {
+      name: "omitted server gate",
+      server: undefined,
+      account: true,
+      available: false,
+      enabled: true,
+    },
+    { name: "disabled server gate", server: false, account: true, available: false, enabled: true },
+    {
+      name: "missing account preference",
+      server: true,
+      account: undefined,
+      available: true,
+      enabled: false,
+    },
+    {
+      name: "disabled account preference",
+      server: true,
+      account: false,
+      available: true,
+      enabled: false,
+    },
+  ])(
+    "keeps $name and returns the configured timeout",
+    async ({ server, account, available, enabled }) => {
+      const { call, predict, values, pluginConfig } = setup();
+      pluginConfig.composerPredictionsEnabled = server;
+      pluginConfig.composerPredictionsTimeoutMs = 45_000;
+      if (account !== undefined) {
+        values.set("user-1", account);
+      }
+      expect(
+        await call({ action: "preferences", userId: "user-1" }, new AbortController().signal),
+      ).toEqual({
+        available,
+        enabled,
+        timeoutMs: 45_000,
+      });
+      expect(await predict()).toEqual({ text: null });
+      expect(mocks.fetch).not.toHaveBeenCalled();
+    },
+  );
+  it("allows a configured inference to finish beyond the former ten-second deadline", async () => {
+    const timeout = useDeadlineClock();
+    const { predict, values, pluginConfig } = setup();
+    pluginConfig.composerPredictionsTimeoutMs = 45_000;
+    values.set("user-1", true);
+    let upstreamSignal: AbortSignal | undefined;
+    mocks.fetch.mockImplementationOnce(
+      ({ signal }: { signal: AbortSignal }) =>
+        new Promise<{ response: Response; release: () => Promise<void> }>((resolve, reject) => {
+          upstreamSignal = signal;
+          const timer = setTimeout(() => {
+            resolve({
+              response: Response.json({
+                choices: [{ message: { content: "다음 단계는 무엇인가요?" } }],
+              }),
+              release: async () => {},
+            });
+          }, 15_000);
+          signal.addEventListener(
+            "abort",
+            () => {
+              clearTimeout(timer);
+              reject(new Error("aborted"));
+            },
+            { once: true },
+          );
+        }),
+    );
+    const pending = predict();
+    await vi.advanceTimersByTimeAsync(15_000);
+    expect(await pending).toEqual({ text: "다음 단계는 무엇인가요?" });
+    expect(upstreamSignal?.aborted).toBe(false);
+    expect(timeout).toHaveBeenCalledWith(45_000);
+  });
+  it("aborts at the configured deadline and releases prediction admission", async () => {
+    useDeadlineClock();
+    const { predict, values, pluginConfig } = setup();
+    pluginConfig.composerPredictionsTimeoutMs = 12_000;
+    values.set("user-1", true);
+    let upstreamSignal: AbortSignal | undefined;
+    mocks.fetch.mockImplementationOnce(
+      ({ signal }: { signal: AbortSignal }) =>
+        new Promise((_resolve, reject) => {
+          upstreamSignal = signal;
+          signal.addEventListener("abort", () => reject(new Error("aborted")), { once: true });
+        }),
+    );
+    const pending = predict();
+    await vi.advanceTimersByTimeAsync(11_999);
+    expect(upstreamSignal?.aborted).toBe(false);
+    await vi.advanceTimersByTimeAsync(1);
+    expect(upstreamSignal?.aborted).toBe(true);
+    expect(await pending).toEqual({ text: null });
+    expect(await predict()).toEqual({ text: "다음 단계는 무엇인가요?" });
+    expect(mocks.fetch).toHaveBeenCalledTimes(2);
+  });
+  it("releases an interrupted response stream and admits the next prediction", async () => {
+    const { predict, values } = setup();
+    values.set("user-1", true);
+    const release = vi.fn(async () => {});
+    let upstreamResponse: Response | undefined;
+    mocks.fetch.mockImplementationOnce(async ({ signal }: { signal: AbortSignal }) => {
+      upstreamResponse = new Response(
+        new ReadableStream<Uint8Array>({
+          start(controller) {
+            signal.addEventListener("abort", () => controller.error(signal.reason), { once: true });
+          },
+        }),
+      );
+      return { response: upstreamResponse, release };
+    });
+    const cancellation = new AbortController();
+    const pending = predict(cancellation.signal);
+    await vi.waitFor(() => expect(upstreamResponse?.body?.locked).toBe(true));
+    cancellation.abort();
+    expect(await pending).toEqual({ text: null });
+    expect(release).toHaveBeenCalledTimes(1);
+    expect(await predict()).toEqual({ text: "다음 단계는 무엇인가요?" });
+    expect(mocks.fetch).toHaveBeenCalledTimes(2);
+  });
   it("uses the configured provider/model and pins it until plugin recreation", async () => {
     vi.stubEnv("PLATFORMCLAW_COMPOSER_PREDICTIONS_MODEL", "  company-fast/org/followup-v2  ");
     const { predict, values } = setup("company-fast", "org/followup-v2");

extensions/admin-http-rpc/src/composer-predictions.ts

@@ -6,6 +6,7 @@ import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime";
 
 export const COMPOSER_RPC = "platformclaw.composer";
 const DEFAULT_MODEL_REF = "codemate/codellm-fast";
+const DEFAULT_TIMEOUT_MS = 30_000;
 const SYSTEM_PROMPT = [
   "Predict one short, useful next message the USER might type in this conversation.",
   "Write in the user's voice and language (Korean when appropriate). Return only one concise question.",
@@ -38,6 +39,12 @@ export function createComposerPredictions(api: OpenClawPluginApi) {
   const enabled = () =>
     api.runtime.config.current().plugins?.entries?.["admin-http-rpc"]?.config
       ?.composerPredictionsEnabled === true;
+  const timeoutMs = () => {
+    // The manifest validates configured values; omitted keys keep the deployment default.
+    const configured = api.runtime.config.current().plugins?.entries?.["admin-http-rpc"]?.config
+      ?.composerPredictionsTimeoutMs as number | undefined;
+    return configured ?? DEFAULT_TIMEOUT_MS;
+  };
   const handleRequest = async (params: unknown, signal: AbortSignal): Promise<unknown> => {
     if (!isRecord(params) || typeof params.userId !== "string" || !params.userId) {
       throw new Error("Invalid composer request");
@@ -47,7 +54,11 @@ export function createComposerPredictions(api: OpenClawPluginApi) {
       if (typeof params.enabled === "boolean") {
         await preferences.register(userId, params.enabled);
       }
-      return { available: enabled(), enabled: (await preferences.lookup(userId)) === true };
+      return {
+        available: enabled(),
+        enabled: (await preferences.lookup(userId)) === true,
+        timeoutMs: timeoutMs(),
+      };
     }
     if (action !== "predict" || !enabled() || (await preferences.lookup(userId)) !== true) {
       return { text: null };
@@ -73,7 +84,7 @@ export function createComposerPredictions(api: OpenClawPluginApi) {
     }
     const controller = new AbortController();
     active = controller;
-    const combined = AbortSignal.any([signal, controller.signal, AbortSignal.timeout(10_000)]);
+    const combined = AbortSignal.any([signal, controller.signal, AbortSignal.timeout(timeoutMs())]);
     try {
       if (!modelRef) {
         return { text: null };