feat: schedule memory evolution in a daily processing window

[lapaca]

Description

Fixes #2333.

Add an opt-in daily idle window for memory evolution in memos-local-plugin. With algorithm.deepProcessing.mode: window, daytime episode closures persist their pending work while lightweight capture remains available. Reflection, automatic reward, L2, L3, and skill requests wait for the configured window; mode: always preserves immediate processing.

algorithm:
  lightweightMemory:
    enabled: false
  deepProcessing:
    mode: window
    window: "02:00-06:00"
    timezone: Asia/Shanghai
    drainIntervalSec: 600
    maxBatchPerCycle: 10

• Recheck the window after acquiring an LLM permit, releasing that permit while waiting for the next window. Shutdown cancels those waits.
• Track pending evolution independently of reward coverage. Explicit feedback and an older processing chain cannot acknowledge a newer deferred closure. Fully completed chains are acknowledged even if their final work finishes after the window closes.
• Coordinate startup recovery, periodic rescans, and queue draining. Recheck current episode state before replaying a startup selection, preserve abandoned close reasons, and recover lost queue entries or a switch back to always.
• Let session/episode close requests return when the window ends. Clear recovery timers and prevent delayed recovery accounting from accessing SQLite after shutdown. Explicit/manual reward remains immediate.

No new project dependencies, API routes, or database schema changes.

Type of change

• New feature (non-breaking change which adds functionality)
• Bug fix (non-breaking change which fixes an issue)

How Has This Been Tested?

Run from apps/memos-local-plugin on Windows with Node 24:

npm test -- tests/unit/pipeline tests/unit/util tests/unit/capture tests/unit/reward tests/unit/startup-recovery.test.ts --silent
Test Files  40 passed (40)
     Tests  316 passed (316)

npm run lint
> tsc -p tsconfig.json --noEmit

npm run build
> tsc -p tsconfig.build.json && node scripts/copy-runtime-assets.cjs

Regression coverage includes semaphore admission at window boundaries, cancellation, foreground permit availability, feedback before reflection, lost queues, concurrent recovery, completed queue cleanup, nonblocking close requests, and restoring always mode. Six recovery regressions for abandoned closures (queue present/lost), overlapping closures, stale startup selections, window-end completion, and post-shutdown database access were observed failing before their fixes.

• Unit Test: 316 relevant tests passed.
• Test Script Or Test Steps: commands and results above; applicable repository pre-commit checks passed for all PR files and the final repair files. Commit hooks ran normally.
• Pipeline Automated API Test: not applicable; this change is confined to plugin internals.

A broader baseline run including tests/unit/config exposed five existing Windows path assertion failures in config/paths.test.ts and config/load.test.ts. The failing assertions and the path implementation are unchanged from main; that broader suite is not fully green on this machine.

Root make format was unavailable because GNU make is not installed. The plugin's TypeScript checks, build, git diff --check, and applicable repository pre-commit checks passed. No Python source was changed.

Checklist

• I have performed a self-review of my own code.
• I have commented my code in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• Related MemOS-Docs issue/PR: not applicable to this plugin-only change; configuration and behavior are described above and in schema comments.
• I have linked the issue to this PR: [Feature Request] memos-local-plugin 完整模式（自进化）支持「闲时批量执行」与 LLM 并发隔离 #2333.
• Reviewers: @hijzy @whipser030 (existing review requests retained).

Reviewer Checklist

• Closes [Feature Request] memos-local-plugin 完整模式（自进化）支持「闲时批量执行」与 LLM 并发隔离 #2333.
• Local check results reviewed; GitHub CI status is tracked separately in this PR's Checks tab.
• Tests have been provided.

[Memtensor-AI]

🤖 Open Code Review

Target: PR #2344
Task: 9ebf8dd41627e889
Base: main
Head: feat/deep-processing-window

🔍 OpenCodeReview found 21 issue(s) in this PR.

---

1. apps/memos-local-plugin/core/config/index.ts (L155)

The destructured local variable window shadows the browser global window. While this plugin runs in Node.js where window is undefined, it is still a style/lint concern that could confuse readers or cause unexpected issues if this code is ever re-used in an environment where window exists (e.g., shared config utilities). Consider renaming the destructured binding to avoid the shadowing:

const { window: dailyWindowSpec, timezone } = completed.algorithm.deepProcessing;
const parsedWindow = parseDailyWindow(dailyWindowSpec);

And update the error message interpolation to use dailyWindowSpec accordingly.

💡 Suggested Change

Before:

    const { window, timezone } = completed.algorithm.deepProcessing;

After:

    const { window: dailyWindowSpec, timezone } = completed.algorithm.deepProcessing;
    const parsedWindow = parseDailyWindow(dailyWindowSpec);
    if (!parsedWindow || parsedWindow.startMin === parsedWindow.endMin) {
      throw new MemosError(
        "config_invalid",
        `invalid algorithm.deepProcessing.window: ${dailyWindowSpec} ` +
          `(expected HH:MM-HH:MM with different start and end times, e.g. 02:00-06:00)`,
      );
    }

---

2. apps/memos-local-plugin/core/capture/subscriber.ts (L69-L85)

When onDeferred throws an exception, the code logs the error but then unconditionally returns (line 84), meaning the episode is neither queued nor immediately processed via runner.runReflect. While the comment indicates this is intentional (relying on a dirty-reward rescan), this creates a silent failure mode where episodes can be dropped from immediate processing without clear visibility. Consider either: 1) falling through to normal processing when enqueue fails (remove the outer return and add it inside the try block after successful enqueue), or 2) emitting a high-severity alert/metric when this fallback path is taken, since it represents degraded behavior that delays episode processing until the next rescan cycle.

---

3. apps/memos-local-plugin/core/config/defaults.ts (L111-L113)

The hardcoded default window: "02:00-06:00" embeds a business-logic assumption (2 AM – 6 AM as the "idle" window) directly in the shipped defaults. This is a timezone-naïve assumption: 02:00-06:00 in the host's local timezone is only a reasonable idle window for users whose system clock aligns with a typical sleep schedule. For users in UTC+0 servers (common in cloud deployments) or users in non-Western timezones this window may fall squarely in the middle of active usage hours.

Consider either:

1. Documenting this constraint explicitly (e.g. "assumes system timezone reflects the user's local time"), or
2. Defaulting timezone to "UTC" with a clear note that operators should override it, to avoid silent misconfiguration in server deployments where the OS timezone is UTC.

---

4. apps/memos-local-plugin/core/pipeline/deep-window.ts (L138-L145)

takeBatch slices by index after sorting, but then writes back entries.slice(batch.length) — it discards exactly batch.length entries from the front of the sorted array. Since batch = entries.slice(0, maxBatch) and batch.length === maxBatch (unless the queue has fewer entries), this is correct in the normal case. However, if entries.length < maxBatch, batch.length equals entries.length, so entries.slice(batch.length) is [] and all entries are consumed — which is also correct. The actual bug is more subtle: entries.slice(batch.length) always starts at position batch.length, not at the end of batch. These are the same only when batch = entries.slice(0, batch.length), which holds here. But the intent would be clearer and safer as entries.slice(maxBatch), avoiding any dependency on batch.length matching the slice index. More importantly, the sort is O(n log n) on every drain cycle over up to 1 000 entries. Since enqueue always appends to the array and entries are appended in chronological order, insertion order already approximates queuedAt order. You can skip the sort entirely and use entries.slice(0, maxBatch) directly, or at most use a partial selection if strict ordering is required.

💡 Suggested Change

Before:

    takeBatch(): DeepWindowQueueEntry[] {
      const entries = read();
      if (entries.length === 0) return [];
      entries.sort((a, b) => a.queuedAt - b.queuedAt);
      const batch = entries.slice(0, maxBatch);
      deps.kv.set(DEEP_PROCESSING_QUEUE_KEY, entries.slice(batch.length));
      return batch;
    },

After:

    takeBatch(): DeepWindowQueueEntry[] {
      const entries = read();
      if (entries.length === 0) return [];
      // Entries are appended in arrival order, so queuedAt is already
      // monotonically non-decreasing — no sort needed.
      const batch = entries.slice(0, maxBatch);
      deps.kv.set(DEEP_PROCESSING_QUEUE_KEY, entries.slice(maxBatch));
      return batch;
    },

---

5. apps/memos-local-plugin/core/pipeline/deep-window.ts (L166-L170)

The isQueueEntry type guard hard-codes the only two accepted closedBy values ("finalized" and "abandoned"). If EpisodeCloseReason gains new variants in the future, persisted queue entries written with those new values will be silently dropped by read() — the filter will discard them as if they were corrupt, causing permanent data loss for those episodes (the dirty-reward rescan would still recover them, but the queue-based ordering and batching is lost). The guard should validate using the actual union type rather than re-enumerating the literals.

Suggested fix: import the set of valid reasons as a runtime constant and use includes — or simply validate that closedBy is a non-empty string and let callers rely on TypeScript's static typing for correctness.

💡 Suggested Change

Before:

  return (
    typeof e.episodeId === "string" &&
    (e.closedBy === "finalized" || e.closedBy === "abandoned") &&
    typeof e.queuedAt === "number"
  );

After:

const VALID_CLOSE_REASONS = new Set<string>(["finalized", "abandoned"] satisfies EpisodeCloseReason[]);

  return (
    typeof e.episodeId === "string" &&
    typeof e.closedBy === "string" &&
    VALID_CLOSE_REASONS.has(e.closedBy) &&
    typeof e.queuedAt === "number"
  );

---

6. apps/memos-local-plugin/core/pipeline/deep-window.ts (L111-L116)

waitUntilOpen polls at a fixed 60-second interval with no maximum iteration count, no overall timeout, and no jitter. In a scenario where the window spec is misconfigured (e.g., parseDailyWindow returns a valid-looking but unreachable window like "03:00-03:00"), isOpen() permanently returns false and this loop runs until the abort signal fires — potentially holding up an evolution chain indefinitely. Additionally, waitForRetry(60_000, signal) is awaited without catching the AbortError it may throw; callers must propagate the throw correctly, which is easy to miss. A short comment or explicit re-throw documentation would help.

Consider adding a max-wait guard or at minimum a log entry on each poll iteration so operators can detect stuck waiters.

💡 Suggested Change

Before:

    async waitUntilOpen(signal?: AbortSignal): Promise<void> {
      signal?.throwIfAborted();
      while (!isOpen()) {
        await waitForRetry(60_000, signal);
      }
    },

After:

    async waitUntilOpen(signal?: AbortSignal): Promise<void> {
      signal?.throwIfAborted();
      let waited = 0;
      while (!isOpen()) {
        deps.log.debug("deep_window.waiting_for_window", { waitedMs: waited });
        await waitForRetry(60_000, signal); // throws AbortError on cancel
        waited += 60_000;
      }
    },

---

7. apps/memos-local-plugin/core/pipeline/deep-window.ts (L123-L136)

The enqueue, takeBatch, and acknowledge methods all follow a non-atomic read-modify-write pattern: read() from kv, mutate the array in memory, then set() it back. If two episodes close concurrently (or if multiple orchestrator instances run in parallel), both could read the same array state, append their entry, and write back—the second write overwrites the first, silently losing the first episode's queue entry. The same race affects takeBatch (two drains could dequeue the same batch) and acknowledge (concurrent acks could lose each other's removals). While the dirty-reward rescan eventually recovers lost episodes, the queue loses its ordering guarantees and episodes may be delayed or processed out of order.

The kv layer needs atomic compare-and-swap, optimistic locking with version numbers, or external serialization (e.g., a mutex) to make these operations safe. If the kv implementation is single-threaded or Node.js event-loop-serialized, document that assumption explicitly.

💡 Suggested Change

Before:

    enqueue(episodeId, closedBy): void {
      const entries = read();
      if (entries.some((e) => e.episodeId === episodeId)) return;
      if (entries.length >= DEEP_PROCESSING_QUEUE_MAX) {
        deps.log.warn("deep_window.queue_full", {
          episodeId,
          size: entries.length,
          max: DEEP_PROCESSING_QUEUE_MAX,
        });
        return;
      }
      entries.push({ episodeId, closedBy, queuedAt: now() });
      deps.kv.set(DEEP_PROCESSING_QUEUE_KEY, entries);
    },

After:

// Option 1: Document the concurrency assumption
    enqueue(episodeId, closedBy): void {
      // SAFETY: assumes kv operations are serialized by the event loop
      // and that only one orchestrator instance writes to this queue.
      const entries = read();
      if (entries.some((e) => e.episodeId === episodeId)) return;
      if (entries.length >= DEEP_PROCESSING_QUEUE_MAX) {
        deps.log.warn("deep_window.queue_full", {
          episodeId,
          size: entries.length,
          max: DEEP_PROCESSING_QUEUE_MAX,
        });
        return;
      }
      entries.push({ episodeId, closedBy, queuedAt: now() });
      deps.kv.set(DEEP_PROCESSING_QUEUE_KEY, entries);
    },

// Option 2: Add optimistic locking with retry
    enqueue(episodeId, closedBy): void {
      for (let attempt = 0; attempt < 3; attempt++) {
        const entries = read();
        if (entries.some((e) => e.episodeId === episodeId)) return;
        if (entries.length >= DEEP_PROCESSING_QUEUE_MAX) {
          deps.log.warn("deep_window.queue_full", { episodeId, size: entries.length });
          return;
        }
        entries.push({ episodeId, closedBy, queuedAt: now() });
        if (deps.kv.compareAndSet(DEEP_PROCESSING_QUEUE_KEY, entries, version)) return;
      }
      deps.log.error("deep_window.enqueue_contention", { episodeId });
    },

---

8. apps/memos-local-plugin/core/pipeline/deep-window.ts (L95)

The processing Set tracks episodes currently being processed in-memory to prevent duplicate work via isProcessing(), but this state is lost on restart. If the process crashes mid-evolution, the episode ID remains in the persisted queue but the in-memory Set is cleared. On restart, isProcessing(episodeId) will return false even though the episode may still be in the queue. If callers use isProcessing as a gate to skip queued work, restarted episodes could be skipped or double-processed depending on how the caller logic is structured.

Review all call sites of isProcessing to ensure they handle the "lost on restart" case correctly, or persist the processing set to the kv store alongside the queue.

💡 Suggested Change

Before:

  const processing = new Set<EpisodeId>();

After:

  // In-memory tracking of active evolution chains. Lost on restart—
  // callers must not rely on this for correctness, only for duplicate
  // suppression during a single process lifetime.
  const processing = new Set<EpisodeId>();

---

9. apps/memos-local-plugin/core/pipeline/deps.ts (L259-L264)

State inconsistency: orphaned deepWindow queue entry when updateMeta throws.

When deps.repos.episodes.updateMeta throws, deepWindow.enqueue(episodeId, closedBy) is called and then the error is re-thrown. At this point:

• previous.superseded = true has already been set (if there was a prior run), removing the previous run from being able to call finishProcessing/acknowledge.
• The new run object has NOT yet been added to deepProcessingRuns — that set call only happens in the caller (runReflect) after markDeepProcessingPending returns, which never happens because it threw.

Result: deepWindow has a queued entry for this episode, but deepProcessingRuns has no corresponding run entry, so prepareDeepProcessingCompletion's closure will never call deepWindow.finishProcessing(episodeId) for it. The processing set inside deepWindow accumulates a startProcessing call that is never balanced by finishProcessing, and the queue entry stays forever unless the dirty-rescan cleans it up.

Suggestion: Either add the run to deepProcessingRuns and call deepWindow.startProcessing before calling markDeepProcessingPending, or restructure so that the enqueue in the catch block does not require a corresponding startProcessing/finishProcessing pair (e.g., only enqueue, skip startProcessing in error paths).

---

10. apps/memos-local-plugin/core/pipeline/deps.ts (L386-L390)

Nested ternary expression is prohibited by coding standards.

The rewardRunner assignment uses a nested ternary: the outer windowEnabled ? { run: … } : immediateRewardRunner contains an inner input.trigger === 'implicit_fallback' ? automaticRewardRunner : immediateRewardRunner. Nested ternaries are explicitly prohibited by the project's coding standards and make the branching logic hard to audit.

Suggestion: Replace with a plain if/else block or an extracted helper function:

let rewardRunner: RewardRunner;
if (windowEnabled) {
  rewardRunner = {
    run: (input) => {
      const runner = input.trigger === "implicit_fallback"
        ? automaticRewardRunner
        : immediateRewardRunner;
      return runner.run(input);
    },
  };
} else {
  rewardRunner = immediateRewardRunner;
}

💡 Suggested Change

Before:

  const rewardRunner: RewardRunner = windowEnabled ? {
    run: (input) => (input.trigger === "implicit_fallback"
      ? automaticRewardRunner
      : immediateRewardRunner).run(input),
  } : immediateRewardRunner;

After:

  let rewardRunner: RewardRunner;
  if (windowEnabled) {
    rewardRunner = {
      run: (input) => {
        const runner = input.trigger === "implicit_fallback"
          ? automaticRewardRunner
          : immediateRewardRunner;
        return runner.run(input);
      },
    };
  } else {
    rewardRunner = immediateRewardRunner;
  }

---

11. apps/memos-local-plugin/core/pipeline/orchestrator.ts (L1631-L1636)

In lightweight mode, completeDeepProcessing() is captured but never called because the function returns early. While the runReflect override is guarded by !lightweightMode (line 304 in deps.ts), making deepProcessingRuns empty, it's cleaner to either:

1. Move prepareDeepProcessingCompletion() after the lightweight check, or
2. Call completeDeepProcessing() before the early return.

This prevents future confusion if deep-processing logic ever needs to run in lightweight mode.

💡 Suggested Change

Before:

    await subs.subscriptions.capture.drain();
    const completeDeepProcessing = subs.prepareDeepProcessingCompletion();
    if (lightweightMode) {
      await embeddingRetryWorker.flush();
      return;
    }

After:

    await subs.subscriptions.capture.drain();
    if (lightweightMode) {
      await embeddingRetryWorker.flush();
      return;
    }
    const completeDeepProcessing = subs.prepareDeepProcessingCompletion();

---

12. apps/memos-local-plugin/core/util/rate-limited-llm.ts (L42-L53)

The acquire loop re-enters waitUntilOpen immediately after releasing the semaphore when the gate has closed mid-wait. This is correct for the gate wait, but waitForBackground is also re-awaited on every retry even though foreground pressure is independent of the gate state. If a high-priority foreground burst is occurring at the same moment the gate closes, a background caller will spin through both waits repeatedly — paying the cost of re-queuing in backgroundWaiters on each iteration — until both conditions align. Consider separating the foreground-resource wait (which should be awaited only once at the outer call boundary, not inside the retry loop) from the gate check:

private async acquire(signal?: AbortSignal): Promise<() => void> {
  // Wait for foreground resources once, outside the gate-retry loop.
  await this.resources?.waitForBackground(signal);
  for (;;) {
    await this.gate?.waitUntilOpen(signal);
    const release = await this.semaphore.acquire(signal);
    if (!this.gate || this.gate.isOpen()) return release;
    release();
  }
}

This avoids repeatedly re-entering the background-waiter queue when only the gate has toggled, and makes the intent of each wait stage clearer.

---

13. apps/memos-local-plugin/core/pipeline/memory-core.ts (L901)

Missing null check before updateMeta. The row variable from getById(entry.episodeId) at line 892 could be null if the episode was deleted between queue insertion and drain, but the code unconditionally calls updateMeta at line 901 without checking row. This could write orphaned metadata or throw an error depending on the repository implementation. Add if (!row) continue; before this line, or move the updateMeta call inside the existing null check block.

💡 Suggested Change

Before:

        handle.repos.episodes.updateMeta(entry.episodeId, { deepProcessingPending: true });

After:

        if (!row || row.status !== "closed" || isLightweightEpisode(row) ||
          dirtyClosedInFlight.has(entry.episodeId) || deepWindow.isProcessing(entry.episodeId)) {
          continue;
        }
        // Queue membership is an outstanding evolution obligation. Reward
        // coverage alone cannot acknowledge it: feedback may have scored an
        // episode before its first reflection. Also backfill older entries
        // that predate the independent pending marker.
        handle.repos.episodes.updateMeta(entry.episodeId, { deepProcessingPending: true });
        if (!dirtyEpisodeBackoffElapsed(row, nowMs)) continue;
        rows.push(row);

---

14. apps/memos-local-plugin/core/pipeline/memory-core.ts (L1877-L1895)

Missing cleanup of deepProcessingPending marker causes infinite retry loop. The episodeRewardIsDirty function at line 1963 returns true when meta.deepProcessingPending === true, but this cleanup logic only clears rewardDirty (line 1893), never clearing deepProcessingPending. Episodes that successfully complete evolution remain dirty forever, triggering infinite retries in autoRescoreDirtyClosedEpisodes and drainDeepProcessingQueue. Add deepProcessingPending: undefined to the success path at line 1893.

💡 Suggested Change

Before:

      for (const [episodeId, prevAttempts] of priorFailedAttempts) {
        const after = handle.repos.episodes.getById(episodeId);
        if (!after) continue;
        const stillDirty = episodeRewardIsDirty(after);
        if (stillDirty) {
          handle.repos.episodes.updateMeta(episodeId, {
            rewardDirty: {
              failedAttempts: prevAttempts + 1,
              lastFailureAt: now,
            },
          });
        } else if (
          after.meta &&
          typeof after.meta === "object" &&
          "rewardDirty" in after.meta
        ) {
          handle.repos.episodes.updateMeta(episodeId, { rewardDirty: undefined });
        }
      }

After:

      for (const [episodeId, prevAttempts] of priorFailedAttempts) {
        const after = handle.repos.episodes.getById(episodeId);
        if (!after) continue;
        const stillDirty = episodeRewardIsDirty(after);
        if (stillDirty) {
          handle.repos.episodes.updateMeta(episodeId, {
            rewardDirty: {
              failedAttempts: prevAttempts + 1,
              lastFailureAt: now,
            },
          });
        } else if (
          after.meta &&
          typeof after.meta === "object" &&
          ("rewardDirty" in after.meta || "deepProcessingPending" in after.meta)
        ) {
          handle.repos.episodes.updateMeta(episodeId, {
            rewardDirty: undefined,
            deepProcessingPending: undefined,
          });
        }
      }

---

15. apps/memos-local-plugin/core/pipeline/memory-core.ts (L1838-L1839)

Memory leak in dirtyClosedInFlight when startupRecoveryCancelled breaks early. Episodes are added to dirtyClosedInFlight at line 1832 before the loop, but if startupRecoveryCancelled becomes true at line 1839, the loop breaks without processing remaining episodes. The finally block at line 1897 runs and removes all episodes from the set, but episodes that were never emitted to the bus (after the break) remain claimed without any recovery path, blocking subsequent attempts. Consider checking startupRecoveryCancelled before the claim loop at line 1832, or track which episodes were actually emitted and only delete those in finally.

💡 Suggested Change

Before:

      for (const ep of episodes) {
        if (startupRecoveryCancelled) break;

After:

    if (startupRecoveryCancelled || shutDown) return;
    // Claim before emitting any events. Startup recovery, the queue drain
    // and periodic scans can overlap while the LLM/flush is awaiting I/O.
    for (const ep of episodes) dirtyClosedInFlight.add(ep.id as EpisodeId);
    const emittedEpisodes = new Set<EpisodeId>();
    try {
      log.info("init.dirty_closed_episodes.rescore", { count: episodes.length });
      // Snapshot the prior failure counters so we can increment them later
      // (after the bus chain settles) without an extra DB read.
      const priorFailedAttempts = new Map<EpisodeId, number>();
      for (const ep of episodes) {
        if (startupRecoveryCancelled) break;
        if (isLightweightEpisode(ep)) continue;
        const episodeId = ep.id as EpisodeId;
        emittedEpisodes.add(episodeId);
        // ... rest of the loop
      }
      // ...
    } finally {
      for (const ep of episodes) {
        if (emittedEpisodes.has(ep.id as EpisodeId) || startupRecoveryCancelled) {
          dirtyClosedInFlight.delete(ep.id as EpisodeId);
        }
      }
    }

---

16. apps/memos-local-plugin/core/pipeline/memory-core.ts (L2157)

Unsafe timer cleanup during shutdown. Mutating recoveryTimers with splice(0) while iterating can skip timers if the array is being modified concurrently. More critically, timer callbacks (drainDeepProcessingQueue, autoRescoreDirtyClosedEpisodes) are async and may still be in-flight after clearInterval - they can write to the database after it's closed (line 2198 calls handle.shutdown). The bounded flush at line 2173-2177 only waits for startupRecoveryPromise, not for in-flight timer callbacks. Consider storing timer references to clear, then awaiting a short grace period for any in-flight callbacks to observe shutDown === true and exit.

💡 Suggested Change

Before:

    for (const timer of recoveryTimers.splice(0)) clearInterval(timer);

After:

    const timersToStop = recoveryTimers.slice();
    recoveryTimers.length = 0;
    for (const timer of timersToStop) clearInterval(timer);
    // Grace period for in-flight timer callbacks to observe shutDown flag
    await new Promise(resolve => setTimeout(resolve, 100));

---

17. apps/memos-local-plugin/core/pipeline/memory-core.ts (L2307-L2309)

Race condition in flushBeforeClose. The loop condition while (!shutDown && deepWindow.isOpen()) at line 2307 allows one more iteration after shutDown is set but before the next check. If shutdown completes and closes the database while waitForRetry is awaiting, the subsequent handle.flush() at line 2315 could attempt to write to a closed database. Additionally, the abort controller is only passed to waitForRetry, not to handle.flush(), so a stuck flush cannot be interrupted. Consider reversing the race logic to while (!shutDown && !deepWindow.isOpen()) or checking shutDown immediately after each await.

💡 Suggested Change

Before:

      while (!shutDown && deepWindow.isOpen()) {
        await waitForRetry(1_000, controller.signal);
      }

After:

    const windowClosed = async (): Promise<void> => {
      // A provider can outlive shutdown's bounded flush if it ignores abort.
      // The adapter's close request must still be able to finish.
      while (!shutDown && !deepWindow.isOpen()) {
        await waitForRetry(1_000, controller.signal);
        if (shutDown) throw new Error('shutdown_during_window_wait');
      }
    };

---

18. apps/memos-local-plugin/core/pipeline/memory-core.ts (L832)

Redundant flush before dirty episode scan. At line 831, the code calls handle.flush() to "finish and acknowledge live chains before selecting retries", but autoRescoreDirtyClosedEpisodes is only called from periodic timer (every 10 minutes) where the system is presumably idle. This adds latency to every periodic scan even when no in-flight work exists. Consider checking handle.buses.hasPendingWork() or similar state before flushing unconditionally.

💡 Suggested Change

Before:

      if (handle.algorithm.deepProcessing.mode === "window") await handle.flush();

After:

      // Finish and acknowledge live chains before selecting retries. They
      // may have completed via their own subscribers without a facade flush.
      if (handle.algorithm.deepProcessing.mode === "window") {
        // Only flush if there's actually pending work to avoid latency on idle scans
        const hasPending = handle.repos.episodes.list({ status: "open", limit: 1 }).length > 0;
        if (hasPending) await handle.flush();
      }

---

19. apps/memos-local-plugin/core/pipeline/memory-core.ts (L733-L736)

Magic number for tick interval lacks clear rationale. The 60-second tick at line 728-731 is described as "kept short so the window edge is noticed promptly", but if drainIntervalSec is 300 (5 minutes) and the window closes 10 seconds before a tick, episodes wait unnecessarily for up to 60 seconds. Consider making this configurable or calculating it dynamically (e.g., Math.min(60_000, DEEP_WINDOW_DRAIN_INTERVAL_MS / 10)) to be more responsive to actual window boundaries.

---

20. apps/memos-local-plugin/core/pipeline/memory-core.ts (L914-L916)

Inconsistent log levels for deep window operations. Deferral is logged at info level (line 1306 in the diff context), but drain errors are logged at debug level (line 914). If the deep-processing queue grows due to repeated failures, the user has no visibility since debug logs are typically disabled in production. Drain errors represent potential issues and should be at least info or warn level. The queue size should also be logged on error to aid diagnosis.

---

21. apps/memos-local-plugin/core/util/time-window.ts (L86-L92)

The function silently defaults hour and minute to 0 if formatToParts does not return matching parts. While unlikely with the specified formatter options (hourCycle: 'h23', hour: '2-digit', minute: '2-digit'), if an exotic ICU build or locale returns an unexpected format, this would silently produce 0 (midnight) instead of an error, causing the time window check to misbehave silently. Consider adding a validation check after the loop to ensure both hour and minute were found, or document that defaulting to midnight is intentional.

💡 Suggested Change

Before:

  const parts = formatterFor(tz).formatToParts(new Date(nowMs));
  let hour = 0;
  let minute = 0;
  for (const p of parts) {
    if (p.type === "hour") hour = Number(p.value);
    else if (p.type === "minute") minute = Number(p.value);
  }

After:

  const parts = formatterFor(tz).formatToParts(new Date(nowMs));
  let hour = 0;
  let minute = 0;
  let foundHour = false;
  let foundMinute = false;
  for (const p of parts) {
    if (p.type === "hour") {
      hour = Number(p.value);
      foundHour = true;
    } else if (p.type === "minute") {
      minute = Number(p.value);
      foundMinute = true;
    }
  }
  if (!foundHour || !foundMinute) {
    throw new Error(`formatToParts did not return hour/minute for timezone ${tz}`);
  }

Generated by cloud-assistant via Open Code Review.

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (149/149 executed). memos_local_plugin/unit: 149/149. Duration: 14s [advisory, non-gating] AI-generated tests on branch test/auto-gen-ae60b1d30a4886e4-20260905175620: 115/120 passed, 5 failed — these do NOT affect the PR verdict; review the branch manually.

Branch: feat/deep-processing-window

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (161/161 executed). memos_local_plugin/unit: 161/161. Duration: 15s [advisory, non-gating] AI-generated tests on branch test/auto-gen-4c30c812ab9019f9-20260905191659: 0/62 passed, 62 failed — these do NOT affect the PR verdict; review the branch manually.

Branch: feat/deep-processing-window

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (167/167 executed). memos_local_plugin/unit: 167/167. Duration: 16s

Branch: feat/deep-processing-window

[Memtensor-AI]

✅ Automated Test Results: PASSED

All tests passed (175/175 executed). memos_local_plugin/unit: 175/175. Duration: 15s [advisory, non-gating] AI-generated tests on branch test/auto-gen-9ebf8dd41627e889-20260906142232: 63/70 passed, 6 failed — these do NOT affect the PR verdict; review the branch manually.

Branch: feat/deep-processing-window