MemTensor · ZongYue99 · Sep 1, 2026 · Aug 31, 2026
@@ -224,7 +224,8 @@ export function createHttpMemoryClient(
           priorityCohortOnly: input.priorityCohortOnly
         },
         signal: input.signal,
-        timeoutMs: input.timeoutMs
+        timeoutMs: input.timeoutMs,
+        maxRetries: 0
       });
     },
 

@@ -228,6 +228,19 @@ describe("HttpMemoryClient", () => {
     });
   });
 
+  it("does not replay a worker request after a server failure", async () => {
+    let calls = 0;
+    const baseUrl = await startServer(async (_request, response) => {
+      calls += 1;
+      response.writeHead(500, { "content-type": "application/json" });
+      response.end("{}");
+    });
+    const client = createHttpMemoryClient({ baseUrl, token: "", timeoutMs: 500, maxRetries: 3 });
+
+    await expect(client.runWorker({ limit: 20 })).rejects.toThrow("memory layer 5xx");
+    expect(calls).toBe(1);
+  });
+
   it("retries 5xx responses and succeeds before max retries is exhausted", async () => {
     let calls = 0;
     const baseUrl = await startServer(async (_request, response) => {

@@ -47,9 +47,12 @@ import {
 export type { ScanProgress } from "../adapters/outbound/agent-source/types.js";
 
 const SCAN_MESSAGE_YIELD_INTERVAL = 100;
-const IMPORT_WORKER_BATCH_SIZE = 20;
+const IMPORT_WORKER_BATCH_SIZE = 1;
 const IMPORT_PROCESSING_COHORT_SIZE = 100;
-const IMPORT_WORKER_TIMEOUT_MS = 600_000;
+// A targeted run leases one job so the timeout never depends on queue ordering.
+// One summary can make three content attempts, each with four 180s HTTP attempts
+// and backoff; keep a safety margin without replaying the worker request.
+const IMPORT_WORKER_TIMEOUT_MS = 2_400_000;
 const IMPORT_PROGRESS_POLL_INTERVAL_MS = 250;
 const INITIAL_GLOBAL_MEMORY_LIMIT = 1_000;
 const INITIAL_ABSENT_SOURCE_MEMORY_LIMIT = 200;
@@ -1077,44 +1080,36 @@ async function processPendingImportSummaries(
     while (pendingMemoryIds.size > 0) {
       scanOptions.signal?.throwIfAborted();
       const targets = [...pendingMemoryIds];
-      const result = await options.memoryClient.runWorker({
+      const workerOutcome = options.memoryClient.runWorker({
         limit: IMPORT_WORKER_BATCH_SIZE,
         targetMemoryIds: targets,
         priorityCohortOnly: true,
         signal: scanOptions.signal,
         timeoutMs: IMPORT_WORKER_TIMEOUT_MS
-      });
-
-      const refreshed = await options.memoryClient.getMemoryProcessingStatus(targets);
-      const processingByMemoryId = new Map(refreshed.items.map((item) => [item.memoryId, item]));
-      const activeMemoryIds = new Set(refreshed.items
-        .filter((item) => item.state === "summary_pending" || item.state === "summarizing" ||
-          item.state === "embedding_pending" || item.state === "embedding")
-        .map((item) => item.memoryId));
-      const previousPending = pendingMemoryIds.size;
-      for (const memoryId of pendingMemoryIds) {
-        if (activeMemoryIds.has(memoryId)) continue;
-        const processing = processingByMemoryId.get(memoryId);
-        if (!processing) {
-          failures.push({ memoryId, reason: "Memory processing state is missing" });
-        } else if (processing.state === "failed") {
-          failures.push({
-            memoryId,
-            reason: processing.errorMessage || "Memory processing failed"
-          });
+      }).then((result) => ({ kind: "worker" as const, result }));
+      let result: Awaited<ReturnType<MemoryClient["runWorker"]>> | undefined;
+
+      while (!result) {
+        const outcome = await Promise.race([
+          workerOutcome,
+          waitForWorkerProgress(IMPORT_PROGRESS_POLL_INTERVAL_MS, undefined, { signal: scanOptions.signal })
+            .then(() => ({ kind: "poll" as const }))
+        ]);
+        if (outcome.kind === "worker") result = outcome.result;
+        const previousPending = pendingMemoryIds.size;
+        await reconcileImportProcessing(options.memoryClient, pendingMemoryIds, failures);
+        if (pendingMemoryIds.size < previousPending) lastProgressAt = Date.now();
+        emitProgress(scanOptions, {
+          sourceId: progressSourceId,
+          phase: "summarize",
+          current: completedMemoryCount + cohort.length - pendingMemoryIds.size,
+          total: ownedMemoryIds.length,
+          message: "Summarizing and indexing latest memories"
+        });
+        if (pendingMemoryIds.size === 0 && !result) {
+          result = (await workerOutcome).result;
         }
-        pendingMemoryIds.delete(memoryId);
-      }
-      if (pendingMemoryIds.size < previousPending) {
-        lastProgressAt = Date.now();
       }
-      emitProgress(scanOptions, {
-        sourceId: progressSourceId,
-        phase: "summarize",
-        current: completedMemoryCount + cohort.length - pendingMemoryIds.size,
-        total: ownedMemoryIds.length,
-        message: "Summarizing and indexing latest memories"
-      });
 
       if (pendingMemoryIds.size === 0) break;
       if (Date.now() - lastProgressAt >= IMPORT_WORKER_TIMEOUT_MS) {
@@ -1130,6 +1125,29 @@ async function processPendingImportSummaries(
   return failures;
 }
 
+async function reconcileImportProcessing(
+  memoryClient: Pick<MemoryClient, "getMemoryProcessingStatus">,
+  pendingMemoryIds: Set<string>,
+  failures: ProcessingFailure[]
+): Promise<void> {
+  const refreshed = await memoryClient.getMemoryProcessingStatus([...pendingMemoryIds]);
+  const processingByMemoryId = new Map(refreshed.items.map((item) => [item.memoryId, item]));
+  const activeMemoryIds = new Set(refreshed.items
+    .filter((item) => item.state === "summary_pending" || item.state === "summarizing" ||
+      item.state === "embedding_pending" || item.state === "embedding")
+    .map((item) => item.memoryId));
+  for (const memoryId of pendingMemoryIds) {
+    if (activeMemoryIds.has(memoryId)) continue;
+    const processing = processingByMemoryId.get(memoryId);
+    if (!processing) {
+      failures.push({ memoryId, reason: "Memory processing state is missing" });
+    } else if (processing.state === "failed") {
+      failures.push({ memoryId, reason: processing.errorMessage || "Memory processing failed" });
+    }
+    pendingMemoryIds.delete(memoryId);
+  }
+}
+
 function appendProcessingFailures(result: ScanResult, failures: readonly ProcessingFailure[]): void {
   result.errors.push(...failures.map((failure) => ({
     conversationId: failure.memoryId,

@@ -134,7 +134,13 @@ async function findSkillFiles(skillsRoot: string): Promise<string[]> {
       if (entry.isFile() && entry.name.toLowerCase() === "skill.md") {
         files.push(entryPath);
       } else if (depth < 2 && (entry.isDirectory() || entry.isSymbolicLink())) {
-        const entryStat = await stat(entryPath);
+        let entryStat;
+        try {
+          entryStat = await stat(entryPath);
+        } catch (error) {
+          if (isNodeError(error) && error.code === "ENOENT") continue;
+          throw error;
+        }
         if (entryStat.isDirectory()) await visit(entryPath, depth + 1);
       }
     }

@@ -4,7 +4,7 @@ import { MANAGED_AGENT_DISCOVERY_PENDING_DATA_PATH } from "@memmy/local-api-cont
 import { mkdtempSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
-import { afterEach, describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import { createSourceRegistry } from "../../adapters/outbound/agent-source/source-registry.js";
 import type {
   ConversationMessage,
@@ -610,9 +610,10 @@ describe("agent source service", () => {
     expect(enqueueCalls).toEqual([["memory-cursor", "memory-custom"]]);
     expect(workerCalls).toEqual([
       expect.objectContaining({
-        limit: 20,
+        limit: 1,
         targetMemoryIds: ["memory-cursor", "memory-custom"],
-        priorityCohortOnly: true
+        priorityCohortOnly: true,
+        timeoutMs: 2_400_000
       })
     ]);
   });
@@ -671,14 +672,68 @@ describe("agent source service", () => {
     })).resolves.toEqual([]);
 
     expect(workerTargets).toEqual([["memory-a", "memory-b"]]);
-    expect(workerLimits).toEqual([20]);
+    expect(workerLimits).toEqual([1]);
     expect(workerPriorityCohorts).toEqual([true]);
     expect(progress).toEqual([
       { current: 0, total: 2 },
       { current: 2, total: 2 }
     ]);
   });
 
+  it("emits persisted summary progress while the worker call remains pending", async () => {
+    const baseMemoryClient = createMockMemoryClient();
+    let releaseWorker = () => undefined;
+    let workerSettled = false;
+    const workerGate = new Promise<void>((resolve) => {
+      releaseWorker = resolve;
+    });
+    const service = createService({
+      memoryClient: {
+        ...baseMemoryClient,
+        async enqueueImportSummaries(memoryIds) {
+          return { enqueued: memoryIds?.length ?? 0, memoryIds: memoryIds ?? [], serverTime: "2026-05-28T10:00:00.000Z" };
+        },
+        async runWorker(input) {
+          await workerGate;
+          workerSettled = true;
+          return baseMemoryClient.runWorker(input);
+        },
+        async getMemoryProcessingStatus(memoryIds) {
+          return {
+            items: memoryIds.map((memoryId) => ({
+              memoryId,
+              state: "ready" as const,
+              stage: null,
+              activeJobId: null,
+              attemptCount: 1,
+              manualRetryCount: 0,
+              retryAction: "retry" as const,
+              errorCode: null,
+              errorMessage: null,
+              failedAt: null,
+              updatedAt: "2026-05-28T10:00:00.000Z"
+            })),
+            serverTime: "2026-05-28T10:00:00.000Z"
+          };
+        }
+      }
+    });
+    const progress: number[] = [];
+    const processing = service.processImportSummaries(["memory-a"], {
+      onProgress(event) {
+        if (event.phase === "summarize") progress.push(event.current);
+      }
+    });
+
+    try {
+      await vi.waitFor(() => expect(progress).toContain(1), { timeout: 1_000, interval: 25 });
+      expect(workerSettled).toBe(false);
+    } finally {
+      releaseWorker();
+      await processing;
+    }
+  });
+
   it("finishes an empty owned-memory batch without starting the worker", async () => {
     const baseMemoryClient = createMockMemoryClient();
     const enqueued: string[][] = [];

@@ -1,5 +1,5 @@
 /** Skill distribution service tests. */
-import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { mkdirSync, mkdtempSync, rmSync, symlinkSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { describe, expect, it } from "vitest";
@@ -40,6 +40,34 @@ describe("skill distribution service", () => {
     }
   });
 
+  it("skips broken directory links while scanning valid sibling skills", async () => {
+    const rootDirectory = mkdtempSync(join(tmpdir(), "memmy-agent-broken-skill-link-"));
+    try {
+      mkdirSync(join(rootDirectory, "skills", "valid-skill"), { recursive: true });
+      writeFileSync(
+        join(rootDirectory, "skills", "valid-skill", "SKILL.md"),
+        "---\nname: valid-skill\n---\nStill discoverable.\n",
+        "utf8"
+      );
+      symlinkSync(
+        join(rootDirectory, "missing-skill-target"),
+        join(rootDirectory, "skills", "broken-skill"),
+        "junction"
+      );
+      const service = createSkillDistributionService({
+        targetRegistry: createSkillTargetRegistry([
+          createFakeTarget({ resolveRootDirectory: () => rootDirectory })
+        ])
+      });
+
+      await expect(service.listSkills?.("cursor")).resolves.toEqual([
+        expect.objectContaining({ sourceSkillId: "valid-skill" })
+      ]);
+    } finally {
+      rmSync(rootDirectory, { recursive: true, force: true });
+    }
+  });
+
   it("renders and installs the fixed Memmy skill manifest", async () => {
     let installed: SkillManifest | undefined;
     const service = createSkillDistributionService({

@@ -135,6 +135,10 @@ export class AnthropicProvider extends LLMProvider {
     return model.startsWith("anthropic/") ? model.slice("anthropic/".length) : model;
   }
 
+  static omitsTemperature(model: string): boolean {
+    return /(?:^|[./])claude-(?:sonnet-5|opus-(?:4-7|5))(?:[-.:@]|$)/i.test(model);
+  }
+
   static toolResultBlock(msg: Record<string, any>): Record<string, any> {
     const content = msg.content;
     const block: Record<string, any> = {
@@ -364,7 +368,7 @@ export class AnthropicProvider extends LLMProvider {
     }
 
     const thinkingEnabled = Boolean(reasoningEffort) && String(reasoningEffort).toLowerCase() !== "none";
-    const omitTemperature = modelName.includes("opus-4-7");
+    const omitTemperature = AnthropicProvider.omitsTemperature(modelName);
     const kwargs: Record<string, any> = {
       model: modelName,
       messages,
@@ -393,6 +397,7 @@ export class AnthropicProvider extends LLMProvider {
 
     if (this.extraHeaders) kwargs.extra_headers = this.extraHeaders;
     if (this.extraBody) Object.assign(kwargs, this.extraBody);
+    if (omitTemperature) delete kwargs.temperature;
     return kwargs;
   }
 

@@ -8,7 +8,11 @@ afterEach(() => {
 
 describe("Anthropic thinking", () => {
   function build(reasoningEffort: string | null, overrides: Record<string, any> = {}): Record<string, any> {
-    const provider = new AnthropicProvider({ apiKey: "key", defaultModel: overrides.defaultModel ?? "claude-sonnet-4-6" });
+    const provider = new AnthropicProvider({
+      apiKey: "key",
+      defaultModel: overrides.defaultModel ?? "claude-sonnet-4-6",
+      extraBody: overrides.extraBody ?? null,
+    });
     return provider.buildKwargs({
       messages: [{ role: "user", content: "hello" }],
       tools: null,
@@ -21,6 +25,16 @@ describe("Anthropic thinking", () => {
     });
   }
 
+  it("does not allow extra body options to reintroduce temperature for fixed-temperature models", () => {
+    const kwargs = build(null, {
+      defaultModel: "publishers/anthropic/models/claude-sonnet-5@20260801",
+      extraBody: { temperature: 0.2, metadata: { user_id: "u1" } },
+    });
+
+    expect(kwargs).not.toHaveProperty("temperature");
+    expect(kwargs.metadata).toEqual({ user_id: "u1" });
+  });
+
   it("keeps dedicated thinking blocks separate from visible content", () => {
     expect(extractReasoning(null, [{ type: "thinking", thinking: "step 1" }], "hello")).toEqual(["step 1", "hello"]);
   });
@@ -195,4 +209,18 @@ describe("Anthropic thinking", () => {
     expect(none).not.toHaveProperty("temperature");
     expect(none).not.toHaveProperty("thinking");
   });
+
+  it.each([
+    "claude-sonnet-5",
+    "anthropic/claude-sonnet-5-20260801",
+    "claude-opus-5",
+    "us.anthropic.claude-opus-5-20260801-v1:0"
+  ])("omits temperature for fixed-temperature Claude model %s", (model) => {
+    expect(build(null, { defaultModel: model })).not.toHaveProperty("temperature");
+    expect(build("adaptive", { defaultModel: model })).not.toHaveProperty("temperature");
+  });
+
+  it("keeps temperature for Claude models that accept it", () => {
+    expect(build(null, { defaultModel: "claude-sonnet-4-6" }).temperature).toBe(0.7);
+  });
 });
@@ -44,6 +44,7 @@
     "jsonc-parser": "^3.3.1",
     "sqlite-vec": "0.1.9",
     "smol-toml": "1.7.0",
+    "tiktoken": "^1.0.22",
     "typescript": "^6.0.3",
     "yaml": "^2.9.0",
     "zod": "^4.3.6"

@@ -301,7 +301,7 @@ export const DEFAULT_MEMMY_CONFIG: MemmyConfig = {
     enableThinking: false,
     temperature: 0,
     maxTokens: MEMORY_SUMMARY_MAX_TOKENS,
-    timeoutMs: 45_000,
+    timeoutMs: 180_000,
     maxRetries: 3,
     malformedRetries: 1
   },