fix: 优化记忆后台任务调度，避免历史导入阻塞实时任务和向量重试

App/backend/local-api-contracts/src/index.ts

@@ -965,6 +965,7 @@ export const CatalogEndpointInputSchema = z.object({
     endpointId: z.string().trim().min(1),
     apiBase: z.string().url(),
     protocol: ModelEndpointProtocolSchema,
+    inheritProviderDefaults: z.boolean().optional(),
     apiKey: z.string().optional(),
     extraHeaders: z.record(z.string(), z.string()).optional(),
     extraBody: z.record(z.string(), z.unknown()).optional(),
@@ -1037,6 +1038,7 @@ export const ModelConfigTestInputSchema = z.object({
     provider: ModelConfigTestProviderSchema,
     endpointId: z.string().trim().min(1),
     protocol: ModelEndpointProtocolSchema,
+    inheritProviderDefaults: z.boolean().optional(),
     apiBase: z.string().url(),
     modelId: z.string().min(1),
     apiKey: z.string().min(1).optional(),
@@ -1146,6 +1148,7 @@ export const CatalogEndpointViewSchema = z.object({
     endpointId: z.string().min(1),
     apiBase: z.string().url(),
     protocol: ModelEndpointProtocolSchema,
+    inheritProviderDefaults: z.boolean().optional(),
     hasApiKey: z.boolean(),
     apiKeyMasked: z.string(),
     apiKey: z.string().default(""),

App/backend/local-api-contracts/src/model-catalog-resolver.ts

@@ -9,6 +9,8 @@ export interface RuntimeCatalogEndpoint {
     apiBase: string;
     protocol: ModelEndpointProtocol;
     apiKey?: string;
+    /** False keeps credentials and request options local to this endpoint. */
+    inheritProviderDefaults?: boolean;
     extraHeaders?: Record<string, string>;
     extraBody?: Record<string, unknown>;
     region?: string;
@@ -155,10 +157,12 @@ export function resolveAssignedModel(input: ResolveAssignedModelInput): ModelSel
     }
     if (!providerOwnerMatches(preset, provider, input.activeAccountId)) return UNAVAILABLE;
 
+    const inherited = endpoint.inheritProviderDefaults === false ? {} : provider;
+    const apiKey = endpoint.apiKey ?? inherited.apiKey;
     let extraBody: Readonly<Record<string, unknown>>;
     try {
         extraBody = deepFreeze(structuredClone({
-            ...(provider.extraBody ?? {}),
+            ...(inherited.extraBody ?? {}),
             ...(endpoint.extraBody ?? {})
         }));
     } catch {
@@ -183,12 +187,12 @@ export function resolveAssignedModel(input: ResolveAssignedModelInput): ModelSel
         endpointId: preset.endpoint,
         protocol: endpoint.protocol,
         apiBase: endpoint.apiBase,
-        ...(endpoint.apiKey ?? provider.apiKey
-            ? { apiKey: endpoint.apiKey ?? provider.apiKey }
+        ...(apiKey
+            ? { apiKey }
             : {}),
         ...(provider.ownerAccountId ? { ownerAccountId: provider.ownerAccountId } : {}),
         extraHeaders: Object.freeze({
-            ...(provider.extraHeaders ?? {}),
+            ...(inherited.extraHeaders ?? {}),
             ...(endpoint.extraHeaders ?? {})
         }),
         extraBody
@@ -329,6 +333,7 @@ function isRuntimeEndpoint(value: unknown): value is RuntimeCatalogEndpoint {
         && typeof value.protocol === "string"
         && PROTOCOLS.has(value.protocol as ModelEndpointProtocol)
         && (value.apiKey === undefined || typeof value.apiKey === "string")
+        && (value.inheritProviderDefaults === undefined || typeof value.inheritProviderDefaults === "boolean")
         && validStringRecord(value.extraHeaders)
         && validUnknownRecord(value.extraBody)
         && (value.region === undefined || typeof value.region === "string");

App/backend/local-api-contracts/tests/endpoint-provider-inheritance.test.ts

@@ -0,0 +1,61 @@
+import { describe, expect, it } from "vitest";
+import { CatalogEndpointInputSchema, CatalogEndpointViewSchema, ModelConfigTestInputSchema } from "../src/index.js";
+import { resolveAssignedModel, type RuntimeModelCatalog } from "../src/model-catalog-resolver.js";
+import { resolveAssignedModel as resolveMemoryModel } from "../../../../Memory/src/contracts/model-catalog-resolver.js";
+
+function catalog(): RuntimeModelCatalog {
+  return {
+    providers: { openai: {
+      apiKey: "unrelated-cloud-key", extraHeaders: { Authorization: "unrelated-header", "X-Cloud": "cloud" },
+      extraBody: { model: "unrelated-model", temperature: 0.9 },
+      endpoints: { memory: { apiBase: "http://127.0.0.1:1234/v1", protocol: "openai-chat-completions" } }
+    } },
+    modelPresets: { memory: { provider: "openai", endpoint: "memory", model: "own-model", source: "byok", capabilities: ["memory_summary"] } },
+    modelAssignments: { byok: { memorySummary: "memory" } }
+  };
+}
+
+describe.each([["shared", resolveAssignedModel], ["standalone Memory", resolveMemoryModel]] as const)("%s endpoint inheritance", (_label, resolve) => {
+  it.each([undefined, "", "own-key"])("keeps an independent endpoint's own credential (%s) and request options", key => {
+    const c = catalog(), e = c.providers!.openai!.endpoints!.memory!;
+    Object.assign(e, { inheritProviderDefaults: false, extraHeaders: { "X-Own": "own" }, extraBody: { seed: 7 } });
+    if (key !== undefined) e.apiKey = key;
+    const before = structuredClone(c);
+    const r = resolve({ catalog: c, mode: "byok", capability: "memory_summary" });
+    expect(r.ok).toBe(true);
+    if (!r.ok) return;
+    expect(r.provider.apiKey).toBe(key || undefined);
+    expect(r.provider.extraHeaders).toEqual({ "X-Own": "own" });
+    expect(r.provider.extraBody).toEqual({ seed: 7 });
+    expect(c).toEqual(before);
+    c.providers!.openai!.apiKey = "rotated-cloud-key";
+    c.providers!.openai!.extraHeaders = { Authorization: "rotated-header" };
+    expect(resolve({ catalog: c, mode: "byok", capability: "memory_summary" })).toEqual(r);
+  });
+
+  it.each([undefined, true])("preserves inherited defaults and later credential rotation (%s)", flag => {
+    const c = catalog(), e = c.providers!.openai!.endpoints!.memory!;
+    if (flag !== undefined) Object.assign(e, { inheritProviderDefaults: flag });
+    e.extraHeaders = { "X-Own": "own" }; e.extraBody = { seed: 7 };
+    const r = resolve({ catalog: c, mode: "byok", capability: "memory_summary" });
+    expect(r.ok && r.provider.apiKey).toBe("unrelated-cloud-key");
+    expect(r.ok && r.provider.extraHeaders).toEqual({ Authorization: "unrelated-header", "X-Cloud": "cloud", "X-Own": "own" });
+    expect(r.ok && r.provider.extraBody).toEqual({ model: "unrelated-model", temperature: 0.9, seed: 7 });
+    c.providers!.openai!.apiKey = "rotated-key";
+    const rotated = resolve({ catalog: c, mode: "byok", capability: "memory_summary" });
+    expect(rotated.ok && rotated.provider.apiKey).toBe("rotated-key");
+  });
+
+  it.each([null, "false", 0])("rejects malformed inheritance configuration (%s)", flag => {
+    const c = catalog(); Object.assign(c.providers!.openai!.endpoints!.memory!, { inheritProviderDefaults: flag });
+    expect(resolve({ catalog: c, mode: "byok", capability: "memory_summary" })).toEqual({ ok: false, code: "model_selection_unavailable" });
+  });
+});
+
+it("retains the optional false flag in catalog and connection-test API round trips", () => {
+  const endpoint = { endpointId: "memory", apiBase: "http://127.0.0.1:1234/v1", protocol: "openai-chat-completions", inheritProviderDefaults: false };
+  expect(CatalogEndpointInputSchema.parse(endpoint)).toHaveProperty("inheritProviderDefaults", false);
+  expect(CatalogEndpointViewSchema.parse({ ...endpoint, hasApiKey: false, apiKeyMasked: "" })).toHaveProperty("inheritProviderDefaults", false);
+  expect(ModelConfigTestInputSchema.parse({ ...endpoint, provider: "openai_compatible", modelId: "own-model" })).toHaveProperty("inheritProviderDefaults", false);
+  expect(CatalogEndpointInputSchema.safeParse({ ...endpoint, inheritProviderDefaults: "false" }).success).toBe(false);
+});

App/backend/src/adapters/inbound/local-api/routes/composio-mcp.ts

@@ -134,6 +134,14 @@ async function dispatchComposioTool(
  * @param options Route dependencies.
  */
 export function registerComposioMcpRoutes(app: FastifyInstance, options: RegisterComposioMcpRoutesOptions): void {
+  const notificationStreams = new Set<StreamableHTTPServerTransport>();
+  let closing = false;
+  app.addHook("preClose", async () => {
+    closing = true;
+    // GET holds an idle notification stream. Tool POSTs may still be writing
+    // external data, so leave those requests to finish normally.
+    await Promise.all([...notificationStreams].map(transport => transport.close()));
+  });
   const handler = async (request: FastifyRequest, reply: FastifyReply): Promise<void> => {
     const token = singleHeader(request.headers[MCP_TOKEN_HEADER]);
     if (!token || token !== options.mcpToken) {
@@ -147,11 +155,20 @@ export function registerComposioMcpRoutes(app: FastifyInstance, options: Registe
 
     reply.hijack();
     reply.raw.on("close", () => {
+      notificationStreams.delete(transport);
       void transport.close();
       void server.close();
     });
 
     await server.connect(transport);
+    if (request.method === "GET") {
+      if (closing) {
+        await transport.close();
+        reply.raw.end();
+        return;
+      }
+      notificationStreams.add(transport);
+    }
     await transport.handleRequest(request.raw, reply.raw, request.body);
   };
 

App/backend/src/adapters/inbound/local-api/server.ts

@@ -52,6 +52,25 @@ export function createLocalApiServer(options: CreateLocalApiServerOptions): Fast
   const app = Fastify({ logger: false });
   const heartbeatIntervalMs = options.heartbeatIntervalMs ?? DEFAULT_HEARTBEAT_INTERVAL_MS;
   const authenticateRuntimeToken = createRuntimeTokenPreHandler(options.permissionManager);
+  const eventStreams = new Set<() => void>();
+  let closing = false;
+  const registerEventStream = (stop: () => void) => {
+    // Authentication may finish after preClose has already drained existing streams.
+    if (closing) stop();
+    else eventStreams.add(stop);
+    return () => { eventStreams.delete(stop); };
+  };
+  app.addHook("preClose", async () => {
+    closing = true;
+    // The renderer closes after this server; do not wait for its EventSource to
+    // disconnect. End only our notification streams, preserving ordinary writes.
+    for (const stop of eventStreams) stop();
+  });
+  app.addHook("onResponse", async () => {
+    // A request already running at preClose can become idle after Fastify's
+    // initial idle sweep. Preserve its response, then retire the idle socket.
+    if (closing) app.server.closeIdleConnections();
+  });
 
   if (options.knowledge) {
     registerKnowledgeRoutes(app, { ...options.knowledge, authenticate: authenticateRuntimeToken });
@@ -151,7 +170,7 @@ export function createLocalApiServer(options: CreateLocalApiServerOptions): Fast
       return reply.code(401).send({ error: "unauthorized" });
     }
 
-    startSse(reply, heartbeatIntervalMs, getSingleHeaderValue(request.headers.origin), options.services.progressBus);
+    startSse(reply, heartbeatIntervalMs, getSingleHeaderValue(request.headers.origin), options.services.progressBus, registerEventStream);
   });
 
   return app;
@@ -216,7 +235,8 @@ function startSse(
   reply: FastifyReply,
   heartbeatIntervalMs: number,
   origin: string | undefined,
-  progressBus: BackendServices["progressBus"]
+  progressBus: BackendServices["progressBus"],
+  registerEventStream: (stop: () => void) => () => void
 ): void {
   // SSE is a long-lived response, so we take over the raw response to prevent Fastify from automatically ending the request.
   reply.hijack();
@@ -284,13 +304,22 @@ function startSse(
     });
   });
 
+  let cleanedUp = false;
+  let unregister = () => {};
   const cleanup = () => {
+    if (cleanedUp) return;
+    cleanedUp = true;
     // Clean up the heartbeat timer promptly after the client disconnects, avoiding orphaned tasks lingering in the background.
     clearInterval(interval);
     unsubscribeScanProgress();
     unsubscribeScanCompleted();
+    unregister();
   };
 
   reply.raw.once("close", cleanup);
   reply.raw.once("error", cleanup);
+  unregister = registerEventStream(() => {
+    cleanup();
+    reply.raw.end();
+  });
 }

App/backend/src/infrastructure/memmy-config/index.ts

@@ -108,6 +108,14 @@ export interface RuntimeProjectionResult {
   memoryConfigAffected: boolean;
 }
 
+export interface CatalogEndpointRequestOptions {
+  apiBase: string;
+  protocol: string;
+  inheritProviderDefaults?: boolean;
+  apiKey?: string;
+  extraHeaders?: Record<string, string>;
+}
+
 export interface MemmyConfigWriter {
   readModelConfig?(): Promise<ModelConfigView>;
 
@@ -119,6 +127,8 @@ export interface MemmyConfigWriter {
 
   readEndpointApiKey?(provider: string, endpointId: string): Promise<string | null>;
 
+  readEndpointRequestOptions?(provider: string, endpointId: string): Promise<CatalogEndpointRequestOptions | null>;
+
   /** Atomically persist the active account/BYOK namespace without rewriting the model catalog. */
   writeUserMode?(mode: UserMode): Promise<void>;
 
@@ -206,6 +216,10 @@ export function createMemmyConfigWriter(options: CreateMemmyConfigWriterOptions
       return readCatalogEndpointApiKey(configPath, provider, endpointId);
     },
 
+    async readEndpointRequestOptions(provider, endpointId) {
+      return readCatalogEndpointRequestOptions(configPath, provider, endpointId);
+    },
+
     async writeUserMode(mode) {
       await mutateRuntimeConfig(configPath, (config) => {
         const app = asRecord(config.app);
@@ -276,7 +290,24 @@ export async function readCatalogEndpointApiKey(
   const provider = asRecord(asRecord(config.providers)?.[providerId]);
   const endpoint = asRecord(asRecord(provider?.endpoints)?.[endpointId]);
   if (!provider || !endpoint) return null;
-  return existingString(endpoint.apiKey) ?? existingString(provider.apiKey) ?? null;
+  return existingString(endpoint.apiKey) ?? (endpoint.inheritProviderDefaults === false ? null : existingString(provider.apiKey)) ?? null;
+}
+
+export async function readCatalogEndpointRequestOptions(
+  configPath: string, providerId: string, endpointId: string
+): Promise<CatalogEndpointRequestOptions | null> {
+  const config = await readCurrentRuntimeConfig(configPath);
+  const provider = asRecord(asRecord(config.providers)?.[providerId]);
+  const endpoint = asRecord(asRecord(provider?.endpoints)?.[endpointId]);
+  if (!provider || !endpoint || !existingString(endpoint.apiBase) || !existingString(endpoint.protocol)) return null;
+  const inherited = endpoint.inheritProviderDefaults === false ? {} : provider;
+  const apiKey = existingString(endpoint.apiKey) ?? existingString(inherited.apiKey);
+  const extraHeaders = Object.fromEntries(Object.entries({
+    ...asRecord(inherited.extraHeaders), ...asRecord(endpoint.extraHeaders)
+  }).filter((entry): entry is [string, string] => typeof entry[1] === "string"));
+  return { apiBase: existingString(endpoint.apiBase)!, protocol: existingString(endpoint.protocol)!,
+    ...(typeof endpoint.inheritProviderDefaults === "boolean" ? { inheritProviderDefaults: endpoint.inheritProviderDefaults } : {}),
+    ...(apiKey ? { apiKey } : {}), ...(Object.keys(extraHeaders).length ? { extraHeaders } : {}) };
 }
 
 async function readCurrentRuntimeConfig(configPath: string): Promise<Record<string, unknown>> {