From 34b49307c0e016e94ebccec47447948037e01ae8 Mon Sep 17 00:00:00 2001 From: Erick Date: Sun, 14 Jun 2026 09:12:27 -0700 Subject: [PATCH 1/4] fix: remove 500-row cap that truncated viewer count displays MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit clampLimit() in _helpers.ts capped all repo list() calls at 500, silently truncating skill/policy/world-model counts everywhere they were derived from array length rather than a COUNT(*) query. - Raise clampLimit cap 500 → 100,000 so metrics() and other internal analytics can read full datasets (fixes Analytics tab) - Rewrite /api/v1/overview to use countSkills/countPolicies/ countWorldModels/countEpisodes instead of list+length, making Overview counts correct regardless of scale (fixes Overview tab) - Align countSkills to use limit:100_000 consistent with other count methods --- .../core/pipeline/memory-core.ts | 2 +- .../core/storage/repos/_helpers.ts | 2 +- .../server/routes/overview.ts | 55 +++++++++++-------- 3 files changed, 33 insertions(+), 26 deletions(-) diff --git a/apps/memos-local-plugin/core/pipeline/memory-core.ts b/apps/memos-local-plugin/core/pipeline/memory-core.ts index b4e331c71..e721b1644 100644 --- a/apps/memos-local-plugin/core/pipeline/memory-core.ts +++ b/apps/memos-local-plugin/core/pipeline/memory-core.ts @@ -3362,7 +3362,7 @@ export function createMemoryCore( includeAllNamespaces?: boolean; }): Promise { ensureLive(); - return handle.repos.skills.list({ status: input?.status, limit: 5_000 }).filter((r) => + return handle.repos.skills.list({ status: input?.status, limit: 100_000 }).filter((r) => (input?.includeAllNamespaces || visibleToCurrent(r)) && matchesNamespaceFilter(r, input) ).length; } diff --git a/apps/memos-local-plugin/core/storage/repos/_helpers.ts b/apps/memos-local-plugin/core/storage/repos/_helpers.ts index 645451f88..c00b379c9 100644 --- a/apps/memos-local-plugin/core/storage/repos/_helpers.ts +++ b/apps/memos-local-plugin/core/storage/repos/_helpers.ts @@ -55,7 +55,7 @@ export function buildPageClauses(opts: PageOptions | undefined, tsColumn: string export function clampLimit(n: number): number { if (!Number.isFinite(n) || n <= 0) return 50; - return Math.min(Math.trunc(n), 10_000); + return Math.min(Math.trunc(n), 100_000); } export function timeRangeWhere( diff --git a/apps/memos-local-plugin/server/routes/overview.ts b/apps/memos-local-plugin/server/routes/overview.ts index d19504e2b..487147092 100644 --- a/apps/memos-local-plugin/server/routes/overview.ts +++ b/apps/memos-local-plugin/server/routes/overview.ts @@ -27,42 +27,49 @@ export function registerOverviewRoutes(routes: Routes, deps: ServerDeps): void { // headless callers. Routing the ping through the viewer's mount // hook keeps the semantics honest (a browser actually opened // the page) and is naturally deduped by browser tab lifetime. - const [health, episodeIds, skills, policies, worldModels, metrics] = - await Promise.all([ - deps.core.health(), - deps.core.listEpisodes({ limit: 5_000 }), - deps.core.listSkills({ limit: 500 }), - // Core only exposes `listPolicies({ status? })`; the viewer wants - // the grand total + per-status so we request the biggest page and - // break it down here. 500 is plenty — fresh installs have dozens. - deps.core.listPolicies({ limit: 500 }), - deps.core.listWorldModels({ limit: 500 }), - // `metrics.total` is the grand total of traces — cheaper than a - // dedicated count RPC and already cached by the core. - deps.core.metrics({ days: 1 }), - ]); + const [ + health, + episodeCount, + skillActive, skillCandidate, skillArchived, + policyActive, policyCandidate, policyArchived, + worldModelCount, + metrics, + ] = await Promise.all([ + deps.core.health(), + deps.core.countEpisodes(), + deps.core.countSkills({ status: "active" }), + deps.core.countSkills({ status: "candidate" }), + deps.core.countSkills({ status: "archived" }), + deps.core.countPolicies({ status: "active" }), + deps.core.countPolicies({ status: "candidate" }), + deps.core.countPolicies({ status: "archived" }), + deps.core.countWorldModels(), + // `metrics.total` is the grand total of traces — cheaper than a + // dedicated count RPC and already cached by the core. + deps.core.metrics({ days: 1 }), + ]); const skillStats = { - total: skills.length, - active: skills.filter((s) => s.status === "active").length, - candidate: skills.filter((s) => s.status === "candidate").length, - archived: skills.filter((s) => s.status === "archived").length, + total: skillActive + skillCandidate + skillArchived, + active: skillActive, + candidate: skillCandidate, + archived: skillArchived, }; const policyStats = { - total: policies.length, - active: policies.filter((p) => p.status === "active").length, - candidate: policies.filter((p) => p.status === "candidate").length, - archived: policies.filter((p) => p.status === "archived").length, + total: policyActive + policyCandidate + policyArchived, + active: policyActive, + candidate: policyCandidate, + archived: policyArchived, }; return { ok: health.ok, version: health.version, - episodes: episodeIds.length, + episodes: episodeCount, traces: metrics.total, skills: skillStats, policies: policyStats, - worldModels: worldModels.length, + worldModels: worldModelCount, llm: health.llm, embedder: health.embedder, skillEvolver: health.skillEvolver, From 21a27ed19dcbbfcf830a3ce1253aed5b2ee66898 Mon Sep 17 00:00:00 2001 From: Erick Wipprecht <53452309+chiefmojo@users.noreply.github.com> Date: Thu, 23 Jul 2026 11:32:13 +0200 Subject: [PATCH 2/4] feat: chunk batch reflection scoring (#1957) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Fix #1540: fix: (#1824) docs(memos-local-plugin): clarify install path and stale dir names (#1540) The README's 'Quick start' section told users to use install.sh instead of npm install, but the warning was buried and users still tried 'npm install -g @memtensor/memos-local-plugin' first. The reporter in #1540 encountered this on a Hermes deployment. This change: - Promotes the 'do not run npm install -g' notice to a prominent IMPORTANT callout explaining why global install is wrong (no agent-home deploy, no config.yaml, no bridge/viewer) and that the tarball intentionally ships built artifacts only. - Adds a Troubleshooting subsection covering the two specific symptoms in the bug report: the 'package not found' misread, and the stale web/ and site/ directory names (web/ is now viewer/, site/ was removed by commit 26e7e3db). - Mentions install.ps1 for Windows alongside install.sh. - CHANGELOG: record the docs fix and reference #1540. Documentation-only change; no code or runtime behavior touched. Co-authored-by: MemOS AutoDev Co-authored-by: Matthew * Fix #1888: [Bug] test_system_parser.py: SystemParser.__init__() got an unexpected keyword a (#1889) fix: remove invalid chunker parameter from SystemParser test instantiation - SystemParser.__init__() signature changed to (embedder, llm=None) - Test was still passing chunker=None causing TypeError - Fixes all 5 failing tests in test_system_parser.py Fixes #1888 Co-authored-by: MemOS AutoDev Co-authored-by: Matthew * Fix #1525: [Bug] clean_json_response crashes with cryptic AttributeError when given None (#1884) * test: add comprehensive tests for clean_json_response (issue #1525) - Add test suite in tests/mem_os/test_format_utils.py - Cover None input ValueError with diagnostic message - Cover markdown removal, whitespace stripping, edge cases - Verify fix for AttributeError when LLM returns None * style: format clean_json_response tests --------- Co-authored-by: MemOS AutoDev Co-authored-by: Matthew * Fix #1901: share_cube_with_user passes swapped args to _validate_cube_access — fails for ev (#1903) fix: validate current user not target in share_cube_with_user (#1901) share_cube_with_user(cube_id, target_user_id) called _validate_cube_access(cube_id, target_user_id), but the validator signature is (user_id, cube_id). The cube_id therefore landed in the user_id slot and _validate_user_exists raised "User '' does not exist or is inactive" for every well-formed call, making the API unusable. The in-code comment "Validate current user has access to this cube" already documented the correct intent: the sharing user (self.user_id) must have access to the cube being shared, not the target. Switch the call to self._validate_cube_access(self.user_id, cube_id). The target user's existence is independently checked on the next line via validate_user(target_user_id), so that path is unchanged. Add regression tests in tests/mem_os/test_memos_core.py that pin down: - validate_user_cube_access is consulted with (self.user_id, cube_id), - add_user_to_cube is called with (target_user_id, cube_id) on success, - a missing target raises "Target user '' does not exist". Closes #1901 Co-authored-by: MemOS AutoDev Bot Co-authored-by: Matthew * feat(memos): chunk batch reflection scoring * Fix #1897: bug(memos-local-plugin): Hermes background model status checks can trigger paid (#1899) * Fix #1897: fix(memos-local-plugin): add LLM circuit breaker for terminal provider errors Issue #1897 reported ~12,900 paid LLM requests in 24 h on Hermes against a DeepSeek key with insufficient balance. The local `system_model_status` row count (12,900) closely tracked the provider-side `request_count` (11,344) for the same billing window. The naming is misleading: `system_model_status` is not a health probe; it is the audit row written once per LLM call (ok / fallback / error) inside `core/llm/client.ts`. With no circuit breaker, every pipeline subscriber (capture / session-relation / reward / L2 / L3 / skill / retrieval LLM filter / world-model) kept firing on every turn / closed episode / induction, generating one paid request each. Add a per-`LlmClient` circuit breaker: - Trips on terminal errors: HTTP 401/402/403 or messages containing `insufficient balance` / `invalid api key` / `unauthorized` / `account suspended` / `billing`. - Open: short-circuits subsequent calls inside the facade without contacting the provider. Throws `MemosError(LLM_UNAVAILABLE)` with `details.circuitOpen=true` so existing catch blocks still work. - Half-open after cool-down (default 5 min, configurable, min 30 s): next call probes the provider; success closes the breaker, terminal failure re-opens it for another cool-down. - Host fallback rescues a call without tripping the breaker — fallback exists precisely to keep going when the primary is down. - Coalesces `system_model_status="circuit_open"` audit rows to at most one per ~25 s while the breaker stays open, so we don't replace paid spam with audit-row spam. - Exposes `circuitOpen` / `circuitOpenUntil` / `circuitOpenedReason` via `LlmClientStats` for the Overview viewer card. - Enabled by default; legacy behaviour available via `circuitBreaker.enabled = false`. Tests: 9 new vitest cases under `tests/unit/llm/client.test.ts` covering trip on 402, trip on "insufficient balance" message, no trip on generic transient, coalescing, half-open close on success, host-fallback rescues without trip, disabled mode, stats fields, and re-open on terminal probe failure. All 59 LLM and 28 pipeline tests pass; `tsc --noEmit` clean. Out of scope (tracked separately): 429 `Retry-After` handling (issue #1620), per-tool rate limits, daily budget caps. * Fix LLM breaker with host fallback --------- Co-authored-by: autodev-bot Co-authored-by: Jiang <33757498+hijzy@users.noreply.github.com> Co-authored-by: GU TIANCHUN <96930846+TianchunGu@users.noreply.github.com> Co-authored-by: Dubberman <48425266+whipser030@users.noreply.github.com> --------- Co-authored-by: Memtensor-AI Co-authored-by: MemOS AutoDev Co-authored-by: Matthew Co-authored-by: MemOS AutoDev Co-authored-by: MemOS AutoDev Bot Co-authored-by: Jiang <33757498+hijzy@users.noreply.github.com> Co-authored-by: GU TIANCHUN <96930846+TianchunGu@users.noreply.github.com> Co-authored-by: Dubberman <48425266+whipser030@users.noreply.github.com> --- .../core/capture/ALGORITHMS.md | 13 +- .../core/capture/batch-scorer.ts | 21 +- .../core/capture/capture.ts | 75 +++++-- .../tests/helpers/fake-llm.ts | 4 +- .../tests/unit/capture/capture-batch.test.ts | 183 +++++++++++++++--- 5 files changed, 238 insertions(+), 58 deletions(-) diff --git a/apps/memos-local-plugin/core/capture/ALGORITHMS.md b/apps/memos-local-plugin/core/capture/ALGORITHMS.md index 15a15ac64..49c98a105 100644 --- a/apps/memos-local-plugin/core/capture/ALGORITHMS.md +++ b/apps/memos-local-plugin/core/capture/ALGORITHMS.md @@ -78,7 +78,8 @@ priority once reward arrives. ## V7 §3.2 batched variant — `batch-scorer.ts` The per-step path (`reflection-synth.ts` + `alpha-scorer.ts`) issues 2N -LLM calls per N-step episode. `batch-scorer.ts` collapses them into ONE: +LLM calls per N-step episode. `batch-scorer.ts` collapses up to +`batchThreshold` steps into one call: ``` inputs = [{idx, state, action, outcome, reflection, synth_allowed}, …] @@ -91,8 +92,8 @@ Dispatch (in `capture.ts`): | `cfg.batchMode` | `cfg.batchThreshold` | behavior | |-------------------|----------------------|----------| | `per_step` | (ignored) | legacy: 2N calls | -| `per_episode` | (ignored) | always batch | -| `auto` (default) | `12` | batch when `N ≤ 12`; else per-step | +| `per_episode` | chunk size | batch when `N ≤ threshold`; else chunk-batch | +| `auto` (default) | `12` | batch when `N ≤ 12`; else chunk-batch | The dispatcher also refuses to batch when no LLM is wired — same fallback path as missing-LLM in per-step mode. @@ -107,15 +108,15 @@ Failure handling: - LLM throws / facade gives up after `malformedRetries=1` → capture catches in `runBatchScoring`, surfaces a `{stage: "batch"}` warning, - and the per-step path runs as a fallback. + and the per-step path runs as a fallback for that chunk. - Validator rejects on length mismatch, missing/non-numeric `alpha`, non-boolean `usable`, non-string `reflection_text`. Same fallback. Bookkeeping (`CaptureResult.llmCalls`): -- `batchedReflection`: 0 or 1 per episode (1 on a successful batch). +- `batchedReflection`: number of successful batch/chunk calls. - `reflectionSynth` / `alphaScoring`: only nonzero when the per-step path - ran (either selected directly, or as fallback after a batch failure). + ran (either selected directly, or as fallback after a chunk failure). Stable prompt fingerprint: diff --git a/apps/memos-local-plugin/core/capture/batch-scorer.ts b/apps/memos-local-plugin/core/capture/batch-scorer.ts index e7b8ab50f..da434c3b2 100644 --- a/apps/memos-local-plugin/core/capture/batch-scorer.ts +++ b/apps/memos-local-plugin/core/capture/batch-scorer.ts @@ -16,11 +16,12 @@ * `transferability` axes benefit directly. * * Trade-offs (encoded in capture.ts dispatch): - * - Prompt grows linearly with N steps. Capped via `batchThreshold`; - * long episodes degrade to the per-step path automatically. - * - One bad output value forces a single batched retry instead of N - * isolated retries — but the facade already does `malformedRetries` - * for us, and on hard failure capture.ts falls back to per-step. + * - Prompt grows linearly with N steps. Each call is capped at + * `batchThreshold`; long episodes run as several bounded chunks. + * - One bad chunk forces a single batched retry for that chunk instead + * of N isolated retries — but the facade already does + * `malformedRetries` for us, and on hard failure capture.ts falls + * back to per-step for that chunk only. * * Wire format ↔ prompt: * Send `{ host_context?, task_context?, steps: [{idx, state, action, outcome, reflection, synth_allowed}] }`. @@ -170,6 +171,7 @@ export async function batchScoreReflections( validate: (v) => validateBatchPayload(v, inputs.length), malformedRetries: 1, temperature: 0, + maxTokens: batchMaxTokens(inputs.length), }, ); @@ -321,6 +323,15 @@ function validateBatchPayload(v: unknown, expected: number): void { } } +function batchMaxTokens(stepCount: number): number { + // Batch output scales with step count; keep a per-step budget but cap below + // the 16k range that triggered avoidable reasoning spend on mimo replay. + const perStepOutputBudget = 512; + const baseBudget = 768; + const ceiling = 8_192; + return Math.min(ceiling, baseBudget + Math.max(1, stepCount) * perStepOutputBudget); +} + function lastToolOutcome(step: NormalizedStep, max: number): string { const last = step.toolCalls[step.toolCalls.length - 1]; if (!last) return "(assistant-only step)"; diff --git a/apps/memos-local-plugin/core/capture/capture.ts b/apps/memos-local-plugin/core/capture/capture.ts index d5b62aa38..bc9c7a812 100644 --- a/apps/memos-local-plugin/core/capture/capture.ts +++ b/apps/memos-local-plugin/core/capture/capture.ts @@ -463,14 +463,14 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { } // Batch reflection + α across every step of the now-closed - // episode. Falls back to per-step scoring when over the threshold - // or when batching fails / no LLM is wired. The reflect pass uses + // episode. Long episodes are chunk-batched at `batchThreshold`; + // failed chunks fall back to per-step scoring. The reflect pass uses // `reflectLlm` (skill-evolver model when configured) for higher // quality reflections; per-turn lite capture still uses `llm`. const reflectStart = now(); const rLlm = deps.reflectLlm ?? deps.llm; - const useBatch = shouldBatch(deps.cfg, normalized.length, rLlm !== null); - const contextEnabled = contextModeFor(deps.cfg, useBatch, normalized.length); + const scoringPlan = planScoring(deps.cfg, normalized.length, rLlm !== null); + const contextEnabled = contextModeFor(deps.cfg, scoringPlan, normalized.length); const taskSummary = contextEnabled.includeTask ? buildTaskReflectionSummary(input.episode, normalized, deps.cfg.taskContextMaxChars) : null; @@ -481,7 +481,10 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { episodeId: input.episode.id, sessionId: input.episode.sessionId, steps: normalized.length, - mode: useBatch ? "batch" : contextEnabled.includeDownstream ? "per_step_downstream" : "per_step", + mode: scoringPlan === "per_step" && contextEnabled.includeDownstream ? "per_step_downstream" : scoringPlan, + chunks: scoringPlan === "chunk_batch" + ? Math.ceil(normalized.length / Math.max(1, deps.cfg.batchThreshold)) + : undefined, reflectionContextMode: deps.cfg.reflectionContextMode, downstreamPreview: contextEnabled.includeDownstream, provider: rLlm?.provider ?? "none", @@ -489,10 +492,13 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { taskSummary: taskSummary ? taskSummary.slice(0, 240) : null, }); let scored: ScoredStep[] = []; - if (useBatch) { + if (scoringPlan === "batch") { scored = await runBatchScoring(normalized, rLlm!, deps, warnings, llmCalls, input.episode.id, taskSummary); } - if (!useBatch || scored.length === 0) { + if (scoringPlan === "chunk_batch") { + scored = await runChunkedBatchScoring(normalized, rLlm!, deps, warnings, llmCalls, input.episode.id, taskSummary); + } + if (scoringPlan === "per_step" || scored.length === 0) { scored = await runPerStepScoring( normalized, rLlm, @@ -1062,30 +1068,30 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { // ─── helpers ──────────────────────────────────────────────────────────────── /** - * Decide whether to use the batched reflection+α path. + * Decide which reflection+α path to use. * * `per_step` → never (legacy path). - * `per_episode` → always, when an LLM is available. - * `auto` → batch when step count fits inside `batchThreshold`. + * `per_episode` → batch up to threshold, then chunk-batch. + * `auto` → batch up to threshold, then chunk-batch. */ -function shouldBatch(cfg: CaptureConfig, stepCount: number, hasLlm: boolean): boolean { - if (!hasLlm) return false; - if (stepCount === 0) return false; - if (cfg.batchMode === "per_step") return false; - if (cfg.batchMode === "per_episode") return true; - // "auto" - return stepCount <= cfg.batchThreshold; +type ScoringPlan = "per_step" | "batch" | "chunk_batch"; + +function planScoring(cfg: CaptureConfig, stepCount: number, hasLlm: boolean): ScoringPlan { + if (!hasLlm) return "per_step"; + if (stepCount === 0) return "per_step"; + if (cfg.batchMode === "per_step") return "per_step"; + return stepCount <= Math.max(1, cfg.batchThreshold) ? "batch" : "chunk_batch"; } function contextModeFor( cfg: CaptureConfig, - useBatch: boolean, + scoringPlan: ScoringPlan, stepCount: number, ): { includeTask: boolean; includeDownstream: boolean } { const mode = cfg.reflectionContextMode; const includeTask = mode === "task" || mode === "task_downstream"; const wantsDownstream = mode === "downstream" || mode === "task_downstream"; - const longPerStep = !useBatch && stepCount > cfg.batchThreshold; + const longPerStep = scoringPlan === "per_step" && stepCount > cfg.batchThreshold; const includeDownstream = wantsDownstream && cfg.longEpisodeReflectMode === "per_step_downstream" && @@ -1145,6 +1151,37 @@ async function runBatchScoring( } } +async function runChunkedBatchScoring( + normalized: NormalizedStep[], + llm: LlmClient, + deps: CaptureDeps, + warnings: CaptureResult["warnings"], + llmCalls: { reflectionSynth: number; alphaScoring: number; batchedReflection: number }, + episodeId: string, + taskSummary: string | null, +): Promise { + const chunkSize = Math.max(1, deps.cfg.batchThreshold); + const chunks: NormalizedStep[][] = []; + for (let start = 0; start < normalized.length; start += chunkSize) { + chunks.push(normalized.slice(start, start + chunkSize)); + } + const concurrency = Math.max(1, deps.cfg.llmConcurrency); + const scoredChunks = await runConcurrently(chunks, concurrency, async (chunk): Promise => { + const scored = await runBatchScoring(chunk, llm, deps, warnings, llmCalls, episodeId, taskSummary); + if (scored.length > 0) return scored; + return runPerStepScoring( + chunk, + llm, + deps, + warnings, + llmCalls, + episodeId, + buildReflectionContexts(chunk, taskSummary, chunk.map(() => [])), + ); + }); + return scoredChunks.flat(); +} + async function runPerStepScoring( normalized: NormalizedStep[], llm: LlmClient | null, diff --git a/apps/memos-local-plugin/tests/helpers/fake-llm.ts b/apps/memos-local-plugin/tests/helpers/fake-llm.ts index 22d9fc1a3..ec2c00261 100644 --- a/apps/memos-local-plugin/tests/helpers/fake-llm.ts +++ b/apps/memos-local-plugin/tests/helpers/fake-llm.ts @@ -20,7 +20,7 @@ export interface FakeLlmScript { complete?: Record string | Promise)>; completeJson?: Record< string, - unknown | ((input: unknown) => unknown | Promise) + unknown | ((input: unknown, opts?: unknown) => unknown | Promise) >; /** Override the served-by identifier. */ servedBy?: LlmProviderName | "host_fallback"; @@ -64,7 +64,7 @@ export function fakeLlm(script: FakeLlmScript = {}): LlmClient { throw new Error(`fakeLlm: no completeJson mock for op="${op}"`); } const value = (typeof entry === "function" - ? await (entry as (x: unknown) => unknown)(input) + ? await (entry as (x: unknown, o?: unknown) => unknown)(input, opts) : entry) as T; if (o?.validate) o.validate(value); return { diff --git a/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts b/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts index d86290517..bc4b76f28 100644 --- a/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts +++ b/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts @@ -7,14 +7,15 @@ * 2. existing reflections are preserved verbatim; * 3. synth-disabled steps stay at α=0 even when the LLM tries to write * one for them; - * 4. `auto` mode falls back to per-step when stepCount > batchThreshold; - * 5. a malformed batched response degrades into the per-step path - * instead of crashing capture. + * 4. `auto` mode chunk-batches when stepCount > batchThreshold; + * 5. a malformed chunk degrades only that chunk into the per-step path + * instead of dropping the whole episode to per-step. */ import { afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; import { createCaptureRunner, type CaptureRunner } from "../../../core/capture/capture.js"; +import { batchScoreReflections } from "../../../core/capture/batch-scorer.js"; import { createCaptureEventBus } from "../../../core/capture/events.js"; import { BATCH_REFLECTION_PROMPT, @@ -312,20 +313,31 @@ describe("capture/pipeline (batched ρ+α path)", () => { expect(t.alpha).toBe(0); // V7 disabledScore semantics }); - it("auto mode falls back to per-step when stepCount > batchThreshold", async () => { + it("auto mode chunk-batches when stepCount > batchThreshold", async () => { + const batchStates: string[][] = []; const llm = fakeLlm({ completeJson: { - // ONLY per-step alpha mock; if batched gets called, the test fails - // with "no completeJson mock for op=...batch...". - [alphaOp]: { alpha: 0.5, usable: true, reason: "ok" }, - }, - complete: { - "capture.reflection.synth": "I made this decision deliberately.", + [batchOp]: (input) => { + const messages = input as Array<{ role: string; content: string }>; + const payload = JSON.parse(messages[messages.length - 1]!.content) as { + steps: Array<{ idx: number; state: string }>; + }; + batchStates.push(payload.steps.map((s) => s.state)); + return { + scores: payload.steps.map((step) => ({ + idx: step.idx, + reflection_text: `reflection ${step.state}`, + alpha: step.idx === 0 ? 0.2 : 0.4, + usable: true, + reason: "ok", + })), + }; + }, }, }); const runner = buildRunner({ batchMode: "auto", batchThreshold: 2 }, llm); - // 3 steps → above threshold → per-step path. + // 3 steps → above threshold → two bounded batch chunks. const ep = episodeSnapshot({ id: "ep_1", sessionId: "se_1", @@ -341,10 +353,17 @@ describe("capture/pipeline (batched ρ+α path)", () => { const result = await runCapture(runner, ep); expect(result.traceIds).toHaveLength(3); - expect(result.llmCalls.batchedReflection).toBe(0); - // 3 synth + 3 alpha calls in per-step mode. - expect(result.llmCalls.reflectionSynth).toBe(3); - expect(result.llmCalls.alphaScoring).toBe(3); + expect(batchStates).toEqual([["a", "b"], ["c"]]); + expect(result.llmCalls.batchedReflection).toBe(2); + expect(result.llmCalls.reflectionSynth).toBe(0); + expect(result.llmCalls.alphaScoring).toBe(0); + + const rows = result.traceIds.map((id) => tmp.repos.traces.getById(id)!); + expect(rows.map((row) => row.reflection)).toEqual([ + "reflection a", + "reflection b", + "reflection c", + ]); }); it("long per-step downstream mode injects up to three following steps", async () => { @@ -368,7 +387,7 @@ describe("capture/pipeline (batched ρ+α path)", () => { }); const runner = buildRunner( { - batchMode: "auto", + batchMode: "per_step", batchThreshold: 2, reflectionContextMode: "task_downstream", longEpisodeReflectMode: "per_step_downstream", @@ -415,16 +434,27 @@ describe("capture/pipeline (batched ρ+α path)", () => { expect(step3Prompt).not.toContain("[step+3]"); }); - it("per_episode mode batches even when step count is large", async () => { - const scores = Array.from({ length: 5 }, (_, i) => ({ - idx: i, - reflection_text: `reflection #${i}`, - alpha: 0.4, - usable: true, - reason: "ok", - })); + it("per_episode mode chunk-batches when step count is large", async () => { + const chunkSizes: number[] = []; const llm = fakeLlm({ - completeJson: { [batchOp]: { scores } }, + completeJson: { + [batchOp]: (input) => { + const messages = input as Array<{ role: string; content: string }>; + const payload = JSON.parse(messages[messages.length - 1]!.content) as { + steps: Array<{ idx: number; state: string }>; + }; + chunkSizes.push(payload.steps.length); + return { + scores: payload.steps.map((step) => ({ + idx: step.idx, + reflection_text: `reflection ${step.state}`, + alpha: 0.4, + usable: true, + reason: "ok", + })), + }; + }, + }, }); const runner = buildRunner({ batchMode: "per_episode", batchThreshold: 2 }, llm); @@ -436,10 +466,111 @@ describe("capture/pipeline (batched ρ+α path)", () => { const ep = episodeSnapshot({ id: "ep_1", sessionId: "se_1", turns }); const result = await runCapture(runner, ep); expect(result.traceIds).toHaveLength(5); - expect(result.llmCalls.batchedReflection).toBe(1); + expect(chunkSizes).toEqual([2, 2, 1]); + expect(result.llmCalls.batchedReflection).toBe(3); expect(result.llmCalls.alphaScoring).toBe(0); }); + it("chunk-batch falls back to per-step only for the failed chunk", async () => { + const llm = fakeLlm({ + completeJson: { + [batchOp]: (input) => { + const messages = input as Array<{ role: string; content: string }>; + const payload = JSON.parse(messages[messages.length - 1]!.content) as { + steps: Array<{ idx: number; state: string }>; + }; + if (payload.steps[0]?.state === "q2") { + throw new Error("chunk failed"); + } + return { + scores: payload.steps.map((step) => ({ + idx: step.idx, + reflection_text: `batch ${step.state}`, + alpha: step.state === "q4" ? 0.5 : 0.2, + usable: true, + reason: "ok", + })), + }; + }, + [alphaOp]: { alpha: 0.9, usable: true, reason: "fallback" }, + }, + complete: { + "capture.reflection.synth": "per-step fallback reflection", + }, + }); + const runner = buildRunner({ batchMode: "auto", batchThreshold: 2 }, llm); + + const turns: EpisodeTurn[] = []; + for (let i = 0; i < 5; i++) { + turns.push(turn("user", `q${i}`, 1_000 + i * 100)); + turns.push(turn("assistant", `a${i}`, 1_050 + i * 100)); + } + const ep = episodeSnapshot({ id: "ep_1", sessionId: "se_1", turns }); + + const result = await runCapture(runner, ep); + expect(result.traceIds).toHaveLength(5); + expect(result.llmCalls.batchedReflection).toBe(2); + expect(result.llmCalls.reflectionSynth).toBe(2); + expect(result.llmCalls.alphaScoring).toBe(2); + expect(result.warnings.filter((w) => w.stage === "batch")).toHaveLength(1); + + const rows = result.traceIds.map((id) => tmp.repos.traces.getById(id)!); + expect(rows.map((row) => row.reflection)).toEqual([ + "batch q0", + "batch q1", + "per-step fallback reflection", + "per-step fallback reflection", + "batch q4", + ]); + expect(rows.map((row) => row.alpha)).toEqual([0.2, 0.2, 0.9, 0.9, 0.5]); + }); + + it("batch scorer passes an explicit maxTokens budget", async () => { + let seenMaxTokens: number | undefined; + const llm = fakeLlm({ + completeJson: { + [batchOp]: (_input, opts) => { + seenMaxTokens = (opts as { maxTokens?: number }).maxTokens; + return { + scores: [ + { + idx: 0, + reflection_text: "I made a useful choice.", + alpha: 0.5, + usable: true, + reason: "ok", + }, + ], + }; + }, + }, + }); + + await batchScoreReflections( + llm, + [ + { + step: { + key: "s1", + ts: 1_000 as EpochMs, + type: "text", + userText: "q", + agentText: "a", + agentThinking: null, + toolCalls: [], + rawReflection: null, + meta: {}, + }, + existingReflection: null, + }, + ], + { synthReflections: true }, + ); + + expect(seenMaxTokens).toBeGreaterThan(0); + expect(seenMaxTokens).toBeLessThan(16_384); + }); + it("malformed batched response → falls back to per-step + emits warning", async () => { const llm = fakeLlm({ completeJson: { From 6734a0ea8d01293b4dae3611fae0ab3834366e1b Mon Sep 17 00:00:00 2001 From: zhaxi Date: Thu, 23 Jul 2026 17:50:48 +0800 Subject: [PATCH 3/4] Revert "feat: chunk batch reflection scoring (#1957)" (#2145) This reverts commit 21a27ed19dcbbfcf830a3ce1253aed5b2ee66898. Co-authored-by: jiachengzhen --- .../core/capture/ALGORITHMS.md | 13 +- .../core/capture/batch-scorer.ts | 21 +- .../core/capture/capture.ts | 75 ++----- .../tests/helpers/fake-llm.ts | 4 +- .../tests/unit/capture/capture-batch.test.ts | 183 +++--------------- 5 files changed, 58 insertions(+), 238 deletions(-) diff --git a/apps/memos-local-plugin/core/capture/ALGORITHMS.md b/apps/memos-local-plugin/core/capture/ALGORITHMS.md index 49c98a105..15a15ac64 100644 --- a/apps/memos-local-plugin/core/capture/ALGORITHMS.md +++ b/apps/memos-local-plugin/core/capture/ALGORITHMS.md @@ -78,8 +78,7 @@ priority once reward arrives. ## V7 §3.2 batched variant — `batch-scorer.ts` The per-step path (`reflection-synth.ts` + `alpha-scorer.ts`) issues 2N -LLM calls per N-step episode. `batch-scorer.ts` collapses up to -`batchThreshold` steps into one call: +LLM calls per N-step episode. `batch-scorer.ts` collapses them into ONE: ``` inputs = [{idx, state, action, outcome, reflection, synth_allowed}, …] @@ -92,8 +91,8 @@ Dispatch (in `capture.ts`): | `cfg.batchMode` | `cfg.batchThreshold` | behavior | |-------------------|----------------------|----------| | `per_step` | (ignored) | legacy: 2N calls | -| `per_episode` | chunk size | batch when `N ≤ threshold`; else chunk-batch | -| `auto` (default) | `12` | batch when `N ≤ 12`; else chunk-batch | +| `per_episode` | (ignored) | always batch | +| `auto` (default) | `12` | batch when `N ≤ 12`; else per-step | The dispatcher also refuses to batch when no LLM is wired — same fallback path as missing-LLM in per-step mode. @@ -108,15 +107,15 @@ Failure handling: - LLM throws / facade gives up after `malformedRetries=1` → capture catches in `runBatchScoring`, surfaces a `{stage: "batch"}` warning, - and the per-step path runs as a fallback for that chunk. + and the per-step path runs as a fallback. - Validator rejects on length mismatch, missing/non-numeric `alpha`, non-boolean `usable`, non-string `reflection_text`. Same fallback. Bookkeeping (`CaptureResult.llmCalls`): -- `batchedReflection`: number of successful batch/chunk calls. +- `batchedReflection`: 0 or 1 per episode (1 on a successful batch). - `reflectionSynth` / `alphaScoring`: only nonzero when the per-step path - ran (either selected directly, or as fallback after a chunk failure). + ran (either selected directly, or as fallback after a batch failure). Stable prompt fingerprint: diff --git a/apps/memos-local-plugin/core/capture/batch-scorer.ts b/apps/memos-local-plugin/core/capture/batch-scorer.ts index da434c3b2..e7b8ab50f 100644 --- a/apps/memos-local-plugin/core/capture/batch-scorer.ts +++ b/apps/memos-local-plugin/core/capture/batch-scorer.ts @@ -16,12 +16,11 @@ * `transferability` axes benefit directly. * * Trade-offs (encoded in capture.ts dispatch): - * - Prompt grows linearly with N steps. Each call is capped at - * `batchThreshold`; long episodes run as several bounded chunks. - * - One bad chunk forces a single batched retry for that chunk instead - * of N isolated retries — but the facade already does - * `malformedRetries` for us, and on hard failure capture.ts falls - * back to per-step for that chunk only. + * - Prompt grows linearly with N steps. Capped via `batchThreshold`; + * long episodes degrade to the per-step path automatically. + * - One bad output value forces a single batched retry instead of N + * isolated retries — but the facade already does `malformedRetries` + * for us, and on hard failure capture.ts falls back to per-step. * * Wire format ↔ prompt: * Send `{ host_context?, task_context?, steps: [{idx, state, action, outcome, reflection, synth_allowed}] }`. @@ -171,7 +170,6 @@ export async function batchScoreReflections( validate: (v) => validateBatchPayload(v, inputs.length), malformedRetries: 1, temperature: 0, - maxTokens: batchMaxTokens(inputs.length), }, ); @@ -323,15 +321,6 @@ function validateBatchPayload(v: unknown, expected: number): void { } } -function batchMaxTokens(stepCount: number): number { - // Batch output scales with step count; keep a per-step budget but cap below - // the 16k range that triggered avoidable reasoning spend on mimo replay. - const perStepOutputBudget = 512; - const baseBudget = 768; - const ceiling = 8_192; - return Math.min(ceiling, baseBudget + Math.max(1, stepCount) * perStepOutputBudget); -} - function lastToolOutcome(step: NormalizedStep, max: number): string { const last = step.toolCalls[step.toolCalls.length - 1]; if (!last) return "(assistant-only step)"; diff --git a/apps/memos-local-plugin/core/capture/capture.ts b/apps/memos-local-plugin/core/capture/capture.ts index bc9c7a812..d5b62aa38 100644 --- a/apps/memos-local-plugin/core/capture/capture.ts +++ b/apps/memos-local-plugin/core/capture/capture.ts @@ -463,14 +463,14 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { } // Batch reflection + α across every step of the now-closed - // episode. Long episodes are chunk-batched at `batchThreshold`; - // failed chunks fall back to per-step scoring. The reflect pass uses + // episode. Falls back to per-step scoring when over the threshold + // or when batching fails / no LLM is wired. The reflect pass uses // `reflectLlm` (skill-evolver model when configured) for higher // quality reflections; per-turn lite capture still uses `llm`. const reflectStart = now(); const rLlm = deps.reflectLlm ?? deps.llm; - const scoringPlan = planScoring(deps.cfg, normalized.length, rLlm !== null); - const contextEnabled = contextModeFor(deps.cfg, scoringPlan, normalized.length); + const useBatch = shouldBatch(deps.cfg, normalized.length, rLlm !== null); + const contextEnabled = contextModeFor(deps.cfg, useBatch, normalized.length); const taskSummary = contextEnabled.includeTask ? buildTaskReflectionSummary(input.episode, normalized, deps.cfg.taskContextMaxChars) : null; @@ -481,10 +481,7 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { episodeId: input.episode.id, sessionId: input.episode.sessionId, steps: normalized.length, - mode: scoringPlan === "per_step" && contextEnabled.includeDownstream ? "per_step_downstream" : scoringPlan, - chunks: scoringPlan === "chunk_batch" - ? Math.ceil(normalized.length / Math.max(1, deps.cfg.batchThreshold)) - : undefined, + mode: useBatch ? "batch" : contextEnabled.includeDownstream ? "per_step_downstream" : "per_step", reflectionContextMode: deps.cfg.reflectionContextMode, downstreamPreview: contextEnabled.includeDownstream, provider: rLlm?.provider ?? "none", @@ -492,13 +489,10 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { taskSummary: taskSummary ? taskSummary.slice(0, 240) : null, }); let scored: ScoredStep[] = []; - if (scoringPlan === "batch") { + if (useBatch) { scored = await runBatchScoring(normalized, rLlm!, deps, warnings, llmCalls, input.episode.id, taskSummary); } - if (scoringPlan === "chunk_batch") { - scored = await runChunkedBatchScoring(normalized, rLlm!, deps, warnings, llmCalls, input.episode.id, taskSummary); - } - if (scoringPlan === "per_step" || scored.length === 0) { + if (!useBatch || scored.length === 0) { scored = await runPerStepScoring( normalized, rLlm, @@ -1068,30 +1062,30 @@ export function createCaptureRunner(deps: CaptureDeps): CaptureRunner { // ─── helpers ──────────────────────────────────────────────────────────────── /** - * Decide which reflection+α path to use. + * Decide whether to use the batched reflection+α path. * * `per_step` → never (legacy path). - * `per_episode` → batch up to threshold, then chunk-batch. - * `auto` → batch up to threshold, then chunk-batch. + * `per_episode` → always, when an LLM is available. + * `auto` → batch when step count fits inside `batchThreshold`. */ -type ScoringPlan = "per_step" | "batch" | "chunk_batch"; - -function planScoring(cfg: CaptureConfig, stepCount: number, hasLlm: boolean): ScoringPlan { - if (!hasLlm) return "per_step"; - if (stepCount === 0) return "per_step"; - if (cfg.batchMode === "per_step") return "per_step"; - return stepCount <= Math.max(1, cfg.batchThreshold) ? "batch" : "chunk_batch"; +function shouldBatch(cfg: CaptureConfig, stepCount: number, hasLlm: boolean): boolean { + if (!hasLlm) return false; + if (stepCount === 0) return false; + if (cfg.batchMode === "per_step") return false; + if (cfg.batchMode === "per_episode") return true; + // "auto" + return stepCount <= cfg.batchThreshold; } function contextModeFor( cfg: CaptureConfig, - scoringPlan: ScoringPlan, + useBatch: boolean, stepCount: number, ): { includeTask: boolean; includeDownstream: boolean } { const mode = cfg.reflectionContextMode; const includeTask = mode === "task" || mode === "task_downstream"; const wantsDownstream = mode === "downstream" || mode === "task_downstream"; - const longPerStep = scoringPlan === "per_step" && stepCount > cfg.batchThreshold; + const longPerStep = !useBatch && stepCount > cfg.batchThreshold; const includeDownstream = wantsDownstream && cfg.longEpisodeReflectMode === "per_step_downstream" && @@ -1151,37 +1145,6 @@ async function runBatchScoring( } } -async function runChunkedBatchScoring( - normalized: NormalizedStep[], - llm: LlmClient, - deps: CaptureDeps, - warnings: CaptureResult["warnings"], - llmCalls: { reflectionSynth: number; alphaScoring: number; batchedReflection: number }, - episodeId: string, - taskSummary: string | null, -): Promise { - const chunkSize = Math.max(1, deps.cfg.batchThreshold); - const chunks: NormalizedStep[][] = []; - for (let start = 0; start < normalized.length; start += chunkSize) { - chunks.push(normalized.slice(start, start + chunkSize)); - } - const concurrency = Math.max(1, deps.cfg.llmConcurrency); - const scoredChunks = await runConcurrently(chunks, concurrency, async (chunk): Promise => { - const scored = await runBatchScoring(chunk, llm, deps, warnings, llmCalls, episodeId, taskSummary); - if (scored.length > 0) return scored; - return runPerStepScoring( - chunk, - llm, - deps, - warnings, - llmCalls, - episodeId, - buildReflectionContexts(chunk, taskSummary, chunk.map(() => [])), - ); - }); - return scoredChunks.flat(); -} - async function runPerStepScoring( normalized: NormalizedStep[], llm: LlmClient | null, diff --git a/apps/memos-local-plugin/tests/helpers/fake-llm.ts b/apps/memos-local-plugin/tests/helpers/fake-llm.ts index ec2c00261..22d9fc1a3 100644 --- a/apps/memos-local-plugin/tests/helpers/fake-llm.ts +++ b/apps/memos-local-plugin/tests/helpers/fake-llm.ts @@ -20,7 +20,7 @@ export interface FakeLlmScript { complete?: Record string | Promise)>; completeJson?: Record< string, - unknown | ((input: unknown, opts?: unknown) => unknown | Promise) + unknown | ((input: unknown) => unknown | Promise) >; /** Override the served-by identifier. */ servedBy?: LlmProviderName | "host_fallback"; @@ -64,7 +64,7 @@ export function fakeLlm(script: FakeLlmScript = {}): LlmClient { throw new Error(`fakeLlm: no completeJson mock for op="${op}"`); } const value = (typeof entry === "function" - ? await (entry as (x: unknown, o?: unknown) => unknown)(input, opts) + ? await (entry as (x: unknown) => unknown)(input) : entry) as T; if (o?.validate) o.validate(value); return { diff --git a/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts b/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts index bc4b76f28..d86290517 100644 --- a/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts +++ b/apps/memos-local-plugin/tests/unit/capture/capture-batch.test.ts @@ -7,15 +7,14 @@ * 2. existing reflections are preserved verbatim; * 3. synth-disabled steps stay at α=0 even when the LLM tries to write * one for them; - * 4. `auto` mode chunk-batches when stepCount > batchThreshold; - * 5. a malformed chunk degrades only that chunk into the per-step path - * instead of dropping the whole episode to per-step. + * 4. `auto` mode falls back to per-step when stepCount > batchThreshold; + * 5. a malformed batched response degrades into the per-step path + * instead of crashing capture. */ import { afterEach, beforeAll, beforeEach, describe, expect, it } from "vitest"; import { createCaptureRunner, type CaptureRunner } from "../../../core/capture/capture.js"; -import { batchScoreReflections } from "../../../core/capture/batch-scorer.js"; import { createCaptureEventBus } from "../../../core/capture/events.js"; import { BATCH_REFLECTION_PROMPT, @@ -313,31 +312,20 @@ describe("capture/pipeline (batched ρ+α path)", () => { expect(t.alpha).toBe(0); // V7 disabledScore semantics }); - it("auto mode chunk-batches when stepCount > batchThreshold", async () => { - const batchStates: string[][] = []; + it("auto mode falls back to per-step when stepCount > batchThreshold", async () => { const llm = fakeLlm({ completeJson: { - [batchOp]: (input) => { - const messages = input as Array<{ role: string; content: string }>; - const payload = JSON.parse(messages[messages.length - 1]!.content) as { - steps: Array<{ idx: number; state: string }>; - }; - batchStates.push(payload.steps.map((s) => s.state)); - return { - scores: payload.steps.map((step) => ({ - idx: step.idx, - reflection_text: `reflection ${step.state}`, - alpha: step.idx === 0 ? 0.2 : 0.4, - usable: true, - reason: "ok", - })), - }; - }, + // ONLY per-step alpha mock; if batched gets called, the test fails + // with "no completeJson mock for op=...batch...". + [alphaOp]: { alpha: 0.5, usable: true, reason: "ok" }, + }, + complete: { + "capture.reflection.synth": "I made this decision deliberately.", }, }); const runner = buildRunner({ batchMode: "auto", batchThreshold: 2 }, llm); - // 3 steps → above threshold → two bounded batch chunks. + // 3 steps → above threshold → per-step path. const ep = episodeSnapshot({ id: "ep_1", sessionId: "se_1", @@ -353,17 +341,10 @@ describe("capture/pipeline (batched ρ+α path)", () => { const result = await runCapture(runner, ep); expect(result.traceIds).toHaveLength(3); - expect(batchStates).toEqual([["a", "b"], ["c"]]); - expect(result.llmCalls.batchedReflection).toBe(2); - expect(result.llmCalls.reflectionSynth).toBe(0); - expect(result.llmCalls.alphaScoring).toBe(0); - - const rows = result.traceIds.map((id) => tmp.repos.traces.getById(id)!); - expect(rows.map((row) => row.reflection)).toEqual([ - "reflection a", - "reflection b", - "reflection c", - ]); + expect(result.llmCalls.batchedReflection).toBe(0); + // 3 synth + 3 alpha calls in per-step mode. + expect(result.llmCalls.reflectionSynth).toBe(3); + expect(result.llmCalls.alphaScoring).toBe(3); }); it("long per-step downstream mode injects up to three following steps", async () => { @@ -387,7 +368,7 @@ describe("capture/pipeline (batched ρ+α path)", () => { }); const runner = buildRunner( { - batchMode: "per_step", + batchMode: "auto", batchThreshold: 2, reflectionContextMode: "task_downstream", longEpisodeReflectMode: "per_step_downstream", @@ -434,27 +415,16 @@ describe("capture/pipeline (batched ρ+α path)", () => { expect(step3Prompt).not.toContain("[step+3]"); }); - it("per_episode mode chunk-batches when step count is large", async () => { - const chunkSizes: number[] = []; + it("per_episode mode batches even when step count is large", async () => { + const scores = Array.from({ length: 5 }, (_, i) => ({ + idx: i, + reflection_text: `reflection #${i}`, + alpha: 0.4, + usable: true, + reason: "ok", + })); const llm = fakeLlm({ - completeJson: { - [batchOp]: (input) => { - const messages = input as Array<{ role: string; content: string }>; - const payload = JSON.parse(messages[messages.length - 1]!.content) as { - steps: Array<{ idx: number; state: string }>; - }; - chunkSizes.push(payload.steps.length); - return { - scores: payload.steps.map((step) => ({ - idx: step.idx, - reflection_text: `reflection ${step.state}`, - alpha: 0.4, - usable: true, - reason: "ok", - })), - }; - }, - }, + completeJson: { [batchOp]: { scores } }, }); const runner = buildRunner({ batchMode: "per_episode", batchThreshold: 2 }, llm); @@ -466,111 +436,10 @@ describe("capture/pipeline (batched ρ+α path)", () => { const ep = episodeSnapshot({ id: "ep_1", sessionId: "se_1", turns }); const result = await runCapture(runner, ep); expect(result.traceIds).toHaveLength(5); - expect(chunkSizes).toEqual([2, 2, 1]); - expect(result.llmCalls.batchedReflection).toBe(3); + expect(result.llmCalls.batchedReflection).toBe(1); expect(result.llmCalls.alphaScoring).toBe(0); }); - it("chunk-batch falls back to per-step only for the failed chunk", async () => { - const llm = fakeLlm({ - completeJson: { - [batchOp]: (input) => { - const messages = input as Array<{ role: string; content: string }>; - const payload = JSON.parse(messages[messages.length - 1]!.content) as { - steps: Array<{ idx: number; state: string }>; - }; - if (payload.steps[0]?.state === "q2") { - throw new Error("chunk failed"); - } - return { - scores: payload.steps.map((step) => ({ - idx: step.idx, - reflection_text: `batch ${step.state}`, - alpha: step.state === "q4" ? 0.5 : 0.2, - usable: true, - reason: "ok", - })), - }; - }, - [alphaOp]: { alpha: 0.9, usable: true, reason: "fallback" }, - }, - complete: { - "capture.reflection.synth": "per-step fallback reflection", - }, - }); - const runner = buildRunner({ batchMode: "auto", batchThreshold: 2 }, llm); - - const turns: EpisodeTurn[] = []; - for (let i = 0; i < 5; i++) { - turns.push(turn("user", `q${i}`, 1_000 + i * 100)); - turns.push(turn("assistant", `a${i}`, 1_050 + i * 100)); - } - const ep = episodeSnapshot({ id: "ep_1", sessionId: "se_1", turns }); - - const result = await runCapture(runner, ep); - expect(result.traceIds).toHaveLength(5); - expect(result.llmCalls.batchedReflection).toBe(2); - expect(result.llmCalls.reflectionSynth).toBe(2); - expect(result.llmCalls.alphaScoring).toBe(2); - expect(result.warnings.filter((w) => w.stage === "batch")).toHaveLength(1); - - const rows = result.traceIds.map((id) => tmp.repos.traces.getById(id)!); - expect(rows.map((row) => row.reflection)).toEqual([ - "batch q0", - "batch q1", - "per-step fallback reflection", - "per-step fallback reflection", - "batch q4", - ]); - expect(rows.map((row) => row.alpha)).toEqual([0.2, 0.2, 0.9, 0.9, 0.5]); - }); - - it("batch scorer passes an explicit maxTokens budget", async () => { - let seenMaxTokens: number | undefined; - const llm = fakeLlm({ - completeJson: { - [batchOp]: (_input, opts) => { - seenMaxTokens = (opts as { maxTokens?: number }).maxTokens; - return { - scores: [ - { - idx: 0, - reflection_text: "I made a useful choice.", - alpha: 0.5, - usable: true, - reason: "ok", - }, - ], - }; - }, - }, - }); - - await batchScoreReflections( - llm, - [ - { - step: { - key: "s1", - ts: 1_000 as EpochMs, - type: "text", - userText: "q", - agentText: "a", - agentThinking: null, - toolCalls: [], - rawReflection: null, - meta: {}, - }, - existingReflection: null, - }, - ], - { synthReflections: true }, - ); - - expect(seenMaxTokens).toBeGreaterThan(0); - expect(seenMaxTokens).toBeLessThan(16_384); - }); - it("malformed batched response → falls back to per-step + emits warning", async () => { const llm = fakeLlm({ completeJson: { From 27c916ede9a1641d5ffcae6652dc5c9a67728485 Mon Sep 17 00:00:00 2001 From: jiachengzhen Date: Thu, 23 Jul 2026 18:25:24 +0800 Subject: [PATCH 4/4] test: stub overview count methods --- apps/memos-local-plugin/tests/unit/server/http.test.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/apps/memos-local-plugin/tests/unit/server/http.test.ts b/apps/memos-local-plugin/tests/unit/server/http.test.ts index 9e4c2a1f8..a81889637 100644 --- a/apps/memos-local-plugin/tests/unit/server/http.test.ts +++ b/apps/memos-local-plugin/tests/unit/server/http.test.ts @@ -65,6 +65,7 @@ function stubCore(): MemoryCore { shareTrace: vi.fn(async (id) => ({ id, share: { scope: "public" } } as any)), getPolicy: vi.fn(async (id) => ({ id, title: "p", status: "active" } as any)), listPolicies: vi.fn(async () => []), + countPolicies: vi.fn(async () => 0), setPolicyStatus: vi.fn(async (id, status) => ({ id, status } as any)), deletePolicy: vi.fn(async () => ({ deleted: false })), sharePolicy: vi.fn(async (id, share) => ({ id, share } as any)), @@ -72,6 +73,7 @@ function stubCore(): MemoryCore { editPolicyGuidance: vi.fn(async (id) => ({ id } as any)), getWorldModel: vi.fn(async () => null), listWorldModels: vi.fn(async () => []), + countWorldModels: vi.fn(async () => 0), deleteWorldModel: vi.fn(async () => ({ deleted: false })), shareWorldModel: vi.fn(async (id, share) => ({ id, status: "active", share } as any)), updateWorldModel: vi.fn(async (id, patch) => ({ id, status: "active", ...patch } as any)),