feat(memhub): memory mode (lite/pro/ultra), evolve and its nightly schedule

- Mode: a saved setting (config.json memory_mode, default pro);
  ONE_MEMORY_GRAPH_STORE still forces ultra for development.
  - API: GET/PUT /api/v1/settings/memory-mode, which reports the switch's
    effects and the graph backfill.
  - CLI: memhub settings mode [lite|pro|ultra] [--yes], which prints the plan
    first.
- The mode drives every SDK client.
  - acquire_client passes it to HyMemoryClient. The read handle rebuilds on a
    change.
  - Write handles are keyed by (profile, revision, mode) and retired with the
    #218 use count, so a batch in flight finishes in the old mode.
  - Jobs read the mode when they run: a job completed in lite is never
    reprocessed; pending jobs take the new mode.
  - Mode never touches the vector collection identity.
- lite needs no generation model: the built-in embedder writes raw, and the
  claim gate admits lite without an active profile. lite never calls
  add_extracted (a note is one raw user turn through add). Recall serves the
  raw layer.
- → ultra: build_memory_graph once per known user, in the background, with a
  status (graph_build).
- Raw retention removed in every mode: MemHub deletes no raw (owner
  decision); the retention loop, its startup step and its test are gone.
- P1-3 / R7: POST /api/v2/evolve answers 202 with a run receipt;
  GET /api/v2/evolve[/{run_id}].
  - Ultra only: 409 memory_mode_not_ultra otherwise.
  - One run per user. Never beside that user's write: it waits for the write
    in flight, and holds that user's jobs (released pre-SDK) until it ends.
  - Loops while should_continue, up to 20 rounds.
  - Per-user last_run_at, last_result and last_error in evolve-state.json
    (counts and codes only).
  - CLI: memhub evolve [--user-id] [--agent-id] [--max-batches] [--wait].
- Nightly scheduler: 03:00 local by default, ultra only, with catch-up when
  the last run is older than the slot or 24h.
  - Users come from Core's records (local, grant holders, write-queue users)
    that the SDK holds memories for.
  - GET/PUT /api/v1/settings/evolve-schedule.
- R10: pipeline {mode, configured_mode, forced_by_env, evolve, last_evolve_at,
  graph_build} in /api/v1/memory-processing, memhub status --json and
  memhub progress.
- Contract for the App: docs/contracts/memory-mode-evolve-v1.md, including
  what SDK ultra additionally enables and zh/en/es copy.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

memhub/MEMORY.md

@@ -13,8 +13,8 @@
    接口先补 SDK。Runtime writers 停止后的 App 启动迁移可直接操作底层存储，
    但不得把迁移专用方法加入公共 Client。
 3. MemHub **不存 memory 实体**；**可以存 `memory_id`** 做组织/回执。原文进入 **SDK raw**。
-4. SDK raw 只保留滚动三天；一条 raw 对应一个完整 turn，只含真人 user 与
-   assistant，tool 不持久化。提取后的高层 memory 不受该窗口影响。
+4. SDK raw 全部保留，任何模式（lite / pro / ultra）都不删除；一条 raw 对应一个
+   完整 turn，只含真人 user 与 assistant，tool 不持久化。
 5. `source_raw_ids` 是 SDK 内部溯源关系，不是产品 Evidence 合同；记忆详情不提供
    “原始依据”或按 raw ID 展开原文的入口。
 

memhub/apps/cli/src/cli-contract-fixes.test.ts

@@ -315,6 +315,8 @@ describe("catalog exit codes", () => {
       "models",
       "memories",
       "profile",
+      "settings",
+      "evolve",
       "graph-width",
       "host",
       "recall",

memhub/apps/cli/src/cli.ts

@@ -188,6 +188,18 @@ import {
   removeManagedCliLink,
   rollbackManagedCliLink,
 } from "./managed-cli-link.js";
+import {
+  type EvolveDocument,
+  evolveExitCode,
+  type ModeDocument,
+  modeExitCode,
+  parseEvolveCommand,
+  parseSettingsCommand,
+  renderEvolveText,
+  renderModeText,
+  runEvolveCommand,
+  runModeCommand,
+} from "./memory-mode-flow.js";
 import {
   addMemory,
   DEFAULT_USER_ID,
@@ -325,6 +337,8 @@ export interface RunControls {
   watchSignal?: AbortSignal;
   watchWait?: (milliseconds: number, signal: AbortSignal) => Promise<void>;
   watchNow?: () => string;
+  /** `memhub evolve --wait` polls on this; tests inject an instant one. */
+  sleep?: (milliseconds: number) => Promise<void>;
   readStdin?: () => Promise<Buffer | null>;
   stdinIsTTY?: boolean;
   /** Injectable onboarding prompter so the wizard can be driven in tests
@@ -2018,12 +2032,17 @@ async function runCommand(
     return sourceProgressExitCode(document);
   }
   if (named === "profile") {
+    const profileFlags = arguments_.slice(1);
     if (
-      arguments_.filter((item) => item === "--json").length > 1 ||
-      arguments_.slice(1).some((value) => value !== "--json")
+      profileFlags.filter((item) => item === "--json").length > 1 ||
+      profileFlags.filter((item) => item === "--versions").length > 1 ||
+      profileFlags.some((value) => value !== "--json" && value !== "--versions")
     ) {
       return usage();
     }
+    if (profileFlags.includes("--versions")) {
+      return await profileVersionsCommand(environment, { asJson, userId });
+    }
     const connected = await connectRuntime(environment, {
       readerTimeoutMs: 15_000,
     });
@@ -2223,6 +2242,55 @@ async function runCommand(
     );
     return outcome.state === "committed" ? EXIT_OK : EXIT_FAILED;
   }
+  if (named === "settings") {
+    const parsed = parseSettingsCommand(arguments_.slice(1));
+    if (!parsed) return usage();
+    const connected = await connectRuntime(environment);
+    if (!connected.ok) {
+      const document: ModeDocument = {
+        schema_version: 1,
+        product: PRODUCT,
+        command: "settings mode",
+        state: "unavailable",
+        code: connected.code,
+      };
+      emit(asJson ? `${JSON.stringify(document)}\n` : renderModeText(document));
+      return connected.exitCode;
+    }
+    const document = await runModeCommand(
+      parsed,
+      connected.reader,
+      connected.operations,
+    );
+    emit(asJson ? `${JSON.stringify(document)}\n` : renderModeText(document));
+    return modeExitCode(document);
+  }
+  if (named === "evolve") {
+    const parsed = parseEvolveCommand(arguments_.slice(1));
+    if (!parsed) return usage();
+    const connected = await connectRuntime(environment);
+    if (!connected.ok) {
+      const document: EvolveDocument = {
+        schema_version: 1,
+        product: PRODUCT,
+        command: "evolve",
+        state: "unavailable",
+        code: connected.code,
+      };
+      emit(
+        asJson ? `${JSON.stringify(document)}\n` : renderEvolveText(document),
+      );
+      return connected.exitCode;
+    }
+    const document = await runEvolveCommand(parsed, {
+      userId: userIdParameters(userId).user_id,
+      reader: connected.reader,
+      operations: connected.operations,
+      sleep: controls.sleep,
+    });
+    emit(asJson ? `${JSON.stringify(document)}\n` : renderEvolveText(document));
+    return evolveExitCode(document);
+  }
   if (named === "graph-width") {
     const parsed = parseGraphWidthCommand(arguments_.slice(1));
     if (!parsed) return usage();
@@ -3888,6 +3956,108 @@ async function emitModelReceipt(
 
 /** Attach to a verified Runtime, telling a stopped one (exit 3) from one
  * that is up but unusable (exit 1). */
+/**
+ * `memhub profile --versions`: the long-term profile's L0 version chains,
+ * latest first, so a value that replaced a better one can be found by id.
+ */
+async function profileVersionsCommand(
+  environment: NodeJS.ProcessEnv,
+  options: { asJson: boolean; userId?: string },
+): Promise<number> {
+  const fail = (state: string, code: string, exitCode: number): number => {
+    emit(
+      options.asJson
+        ? `${JSON.stringify({ schema_version: 2, product: PRODUCT, state, code })}\n`
+        : `画像版本读取失败：${code}\n`,
+    );
+    return exitCode;
+  };
+  const connected = await connectRuntime(environment, {
+    readerTimeoutMs: 15_000,
+  });
+  if (!connected.ok) {
+    return fail("unavailable", connected.code, connected.exitCode);
+  }
+  const outcome = await connected.reader<unknown>(
+    "/api/v2/user-profile/versions",
+    userIdParameters(options.userId),
+  );
+  const chains = outcome.state === "ok" ? profileChains(outcome.data) : null;
+  if (outcome.state !== "ok" || chains === null) {
+    return fail(
+      outcome.state === "ok" ? "invalid" : outcome.state,
+      outcome.state === "ok" ? "response_invalid" : outcome.code,
+      EXIT_FAILED,
+    );
+  }
+  if (options.asJson) {
+    emit(
+      `${JSON.stringify({
+        schema_version: 2,
+        product: PRODUCT,
+        state: "ok",
+        ...(outcome.data as object),
+      })}\n`,
+    );
+    return EXIT_OK;
+  }
+  if (chains.length === 0) {
+    emit("尚未形成用户画像版本。\n");
+    return EXIT_OK;
+  }
+  const lines: string[] = [];
+  for (const chain of chains) {
+    lines.push(`画像 ${chain.memory_id}（${chain.status}）`);
+    for (const version of chain.versions) {
+      const when =
+        typeof version.gmt_created === "number"
+          ? new Date(version.gmt_created * 1000).toISOString()
+          : "时间未知";
+      lines.push(
+        `  ${version.current ? "*" : " "} ${version.memory_id}  ${when}  ${version.content}`,
+      );
+    }
+  }
+  lines.push(
+    "* 表示当前生效的版本；删除错误版本：memhub memories delete <memory-id> --yes",
+  );
+  emit(`${lines.join("\n")}\n`);
+  return EXIT_OK;
+}
+
+type ProfileChainView = {
+  memory_id: string;
+  status: string;
+  versions: {
+    memory_id: string;
+    content: string;
+    gmt_created?: number | null;
+    current: boolean;
+  }[];
+};
+
+function profileChains(value: unknown): ProfileChainView[] | null {
+  if (!value || typeof value !== "object") return null;
+  const chains = (value as { chains?: unknown }).chains;
+  if (!Array.isArray(chains)) return null;
+  const valid = chains.every(
+    (chain) =>
+      chain &&
+      typeof chain === "object" &&
+      typeof (chain as ProfileChainView).memory_id === "string" &&
+      typeof (chain as ProfileChainView).status === "string" &&
+      Array.isArray((chain as ProfileChainView).versions) &&
+      (chain as ProfileChainView).versions.every(
+        (version) =>
+          version &&
+          typeof version.memory_id === "string" &&
+          typeof version.content === "string" &&
+          typeof version.current === "boolean",
+      ),
+  );
+  return valid ? (chains as ProfileChainView[]) : null;
+}
+
 async function connectRuntime(
   environment: NodeJS.ProcessEnv,
   options?: { readerTimeoutMs?: number },
@@ -4155,6 +4325,8 @@ const SDK_MIGRATION_FAILURES: Record<string, string> = {
     "核对发现新版本还读不到你的全部记忆，没有切换数据源。",
   sdk_migration_model_profile_required:
     "迁移需要先有可用的模型配置，用来核对新版本能否读回记忆。请先用 memhub models setup 完成配置，再运行 memhub restart 重试。",
+  sdk_migration_vector_profile_missing:
+    "记忆的向量来自一个已删除的模型配置，当前配置读不回它们，所以没有迁移。请重新添加并启用那个模型配置，再运行 memhub restart 重试。",
 };
 
 const MIGRATION_ISSUES_URL =
@@ -6705,7 +6877,13 @@ const OWN_DATA_DIR_COMMANDS = new Set([
 /** Remove one `--data-dir <absolute path>` from argv; null when it is
  * repeated, missing its value, or not an absolute path. */
 /** Memory commands that honour the global `--user-id`. */
-const USER_ID_COMMANDS = new Set(["memories", "recall", "profile", "write"]);
+const USER_ID_COMMANDS = new Set([
+  "memories",
+  "recall",
+  "profile",
+  "write",
+  "evolve",
+]);
 
 /**
  * Take `--user-id <id>` out of the arguments. The default user (`local`) is

memhub/apps/cli/src/command-catalog.ts

@@ -434,9 +434,55 @@ export const COMMANDS: readonly CommandDescriptor[] = [
   },
   {
     name: "profile",
-    summary: "读取 SDK 组装后的用户长期画像（L0 + 已启用的长期画像补充）。",
-    usage: "memhub profile [--user-id <id>] [--json]",
-    options: [USER_ID_OPTION, JSON_OPTION],
+    summary:
+      "读取 SDK 组装后的用户长期画像（L0 + 已启用的长期画像补充）；--versions 列出 L0 画像的版本链（最新在前，* 为当前版本）。",
+    usage: "memhub profile [--versions] [--user-id <id>] [--json]",
+    options: [
+      {
+        flag: "--versions",
+        summary:
+          "列出 L0 基本信息的版本链（get_user_basic_info），用于找出覆盖了正确值的错误版本。",
+      },
+      USER_ID_OPTION,
+      JSON_OPTION,
+    ],
+    exit_codes: RUNTIME_EXITS,
+  },
+  {
+    name: "settings",
+    summary:
+      "查看或切换记忆模式（lite / pro / ultra）。lite 不调用模型，只在原始对话里检索；pro 由模型提取记忆（System 1）；ultra 在 pro 之上每晚自动进化（System 2，使用知识图谱）。任何模式都保留全部原始对话。切换不迁移数据：已处理的不会重做，排队中的和新的按新模式处理；需 --yes，不加时只显示计划。环境变量 ONE_MEMORY_GRAPH_STORE 会强制 ultra。",
+    usage: "memhub settings mode [lite|pro|ultra] [--yes] [--json]",
+    options: [
+      {
+        flag: "mode [lite|pro|ultra]",
+        summary: "不带模式只显示当前模式；带模式时先显示切换计划。",
+      },
+      CONFIRM_OPTION,
+      JSON_OPTION,
+    ],
+    exit_codes: RUNTIME_EXITS,
+  },
+  {
+    name: "evolve",
+    summary:
+      "立即为一个用户运行记忆进化（System 2，仅 ultra 模式）。在后台运行并返回运行 ID；--wait 等待结果。同一用户同时只运行一次；每晚的自动进化由 Runtime 按设置安排。",
+    usage:
+      "memhub evolve [--user-id <id>] [--agent-id <id>] [--max-batches <n>] [--wait] [--json]",
+    options: [
+      USER_ID_OPTION,
+      {
+        flag: "--agent-id <id>",
+        summary:
+          "进化哪个 agent 的记忆，默认 all-apps（MemHub 写入记忆用的 agent）。",
+      },
+      {
+        flag: "--max-batches <n>",
+        summary: "每轮最多处理的批次数（1–1000），默认由 SDK 决定。",
+      },
+      { flag: "--wait", summary: "等待这次进化结束再返回。" },
+      JSON_OPTION,
+    ],
     exit_codes: RUNTIME_EXITS,
   },
   {

memhub/apps/cli/src/command-contract.test.ts

@@ -76,6 +76,8 @@ const COMMANDS = [
   "models",
   "memories",
   "profile",
+  "settings",
+  "evolve",
   "graph-width",
   "host",
   "recall",

memhub/apps/cli/src/core-client.ts

@@ -17,7 +17,13 @@ const READ_ONLY_PATHS = [
   "/api/v2/home-summary",
   "/api/v2/memory-library",
   "/api/v2/user-profile",
+  "/api/v2/user-profile/versions",
+  "/api/v1/settings/memory-mode",
+  "/api/v2/evolve",
 ] as const;
+/** One evolve run's receipt (`memhub evolve --wait`). */
+const EVOLVE_RUN =
+  /^\/api\/v2\/evolve\/[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
 const SDK_MEMORY_DETAIL = /^\/api\/v2\/memories\/[A-Za-z0-9][\w.-]{0,127}$/;
 /** Status of one `POST /api/v2/memories` addition, by its request id. */
 const MEMORY_ADDITION =
@@ -220,6 +226,7 @@ export function isReadOnlyPath(path: string): boolean {
   if (IMPORT_JOB.test(path)) return true;
   if (OPERATION_RECEIPT.test(path)) return true;
   if (MEMORY_ADDITION.test(path)) return true;
+  if (EVOLVE_RUN.test(path)) return true;
   if (!SDK_MEMORY_DETAIL.test(path)) return false;
   const segment = path.slice("/api/v2/memories/".length);
   return !(RESERVED_MEMORY_SEGMENTS as readonly string[]).includes(segment);

memhub/apps/cli/src/core-operations.ts

@@ -26,6 +26,9 @@ const SOURCE_IMPORT_JOB_PATH = "/api/v1/import-jobs";
 const IMPORT_JOB_ACTION_PATH =
   /^\/api\/v1\/import-jobs\/[A-Za-z0-9][\w.-]{0,127}\/(?:pause|resume|cancel)$/;
 const GRAPH_WIDTH_PATH = "/api/v1/graph-width";
+/** `memhub evolve` starts a run; `memhub settings mode <m> --yes` sets the mode. */
+const EVOLVE_PATH = "/api/v2/evolve";
+const MEMORY_MODE_PATH = "/api/v1/settings/memory-mode";
 const ONBOARDING_COMPLETE_PATH = "/api/v1/onboarding/complete";
 const MAX_BYTES = 4 * 1024 * 1024;
 
@@ -97,6 +100,7 @@ export function createCoreOperations(attachment: {
         !SOURCE_AUTHORIZE_PATH.test(path) &&
         !SOURCE_REVOKE_PATH.test(path) &&
         path !== GRAPH_WIDTH_PATH &&
+        path !== EVOLVE_PATH &&
         path !== ONBOARDING_COMPLETE_PATH &&
         !MODEL_PROFILE_ACTIVATE.test(path)
       ) {
@@ -132,7 +136,10 @@ export function createCoreOperations(attachment: {
       body: Record<string, unknown>,
       signal?: AbortSignal,
     ) {
-      if (!SDK_MEMORY_PATH.test(path) || path.includes("?")) {
+      if (
+        path !== MEMORY_MODE_PATH &&
+        (!SDK_MEMORY_PATH.test(path) || path.includes("?"))
+      ) {
         throw new Error("core_operation_path_not_allowed");
       }
       let response: { status: number; data: unknown };

memhub/apps/cli/src/memory-mode-cli.test.ts

@@ -0,0 +1,309 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { run } from "./cli.js";
+import { modeEffects, parseSettingsCommand } from "./memory-mode-flow.js";
+import { composeStatusReport, renderStatusText } from "./status-report.js";
+
+const transport = vi.hoisted(() => ({
+  get: vi.fn(),
+  post: vi.fn(),
+  put: vi.fn(),
+  patch: vi.fn(),
+  delete: vi.fn(),
+}));
+vi.mock("axios", () => ({
+  default: {
+    create: vi.fn(() => transport),
+    get: vi.fn(),
+    post: vi.fn(),
+    isAxiosError: () => false,
+  },
+}));
+
+const runtime = vi.hoisted(() => ({
+  inspectRuntimeStatus: vi.fn(),
+  discoverRuntimeAttachment: vi.fn(),
+}));
+vi.mock("@one-memory/runtime-control", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@one-memory/runtime-control")>()),
+  ...runtime,
+}));
+
+const RUN_ID = "11111111-2222-4333-8444-555555555555";
+
+function modeState(mode: string, configured = mode) {
+  return {
+    mode,
+    configured_mode: configured,
+    forced_by_env: false,
+    modes: ["lite", "pro", "ultra"],
+    graph_build: { state: "idle", users: {} },
+  };
+}
+
+function receipt(state: string, extra: Record<string, unknown> = {}) {
+  return {
+    run_id: RUN_ID,
+    user_id: "wife",
+    agent_id: "all-apps",
+    state,
+    trigger: "manual",
+    started_at: "2026-11-01T03:00:00Z",
+    rounds: state === "running" ? 0 : 2,
+    tasks_processed: state === "running" ? 0 : 5,
+    batch_failures: 0,
+    capped: false,
+    elapsed_ms: 10,
+    results: { schemas: ["never printed"] },
+    ...extra,
+  };
+}
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  for (const method of Object.values(transport)) method.mockReset();
+  runtime.inspectRuntimeStatus.mockResolvedValue({
+    state: "running",
+    code: "runtime_verified",
+    instanceId: "instance-1",
+    host: "127.0.0.1",
+    port: 8765,
+    version: "1.2.3",
+  });
+  runtime.discoverRuntimeAttachment.mockResolvedValue({
+    instanceId: "instance-1",
+    apiToken: "synthetic-runtime-token",
+    host: "127.0.0.1",
+    port: 8765,
+    servicePid: 4242,
+    health: {
+      status: "ok",
+      service: "one-memory-core",
+      version: "1.2.3",
+      instance_id: "instance-1",
+    },
+  });
+  transport.get.mockImplementation(async (url: string) =>
+    url === "/api/v1/settings/memory-mode"
+      ? { status: 200, data: modeState("pro") }
+      : { status: 404, data: {} },
+  );
+});
+
+describe("memhub settings mode", () => {
+  it("shows the current mode without changing it", async () => {
+    const result = await capturedRun(["settings", "mode", "--json"]);
+    expect(result.code).toBe(0);
+    expect(JSON.parse(result.stdout).mode.mode).toBe("pro");
+    expect(transport.put).not.toHaveBeenCalled();
+  });
+
+  it("prints the plan and changes nothing without --yes", async () => {
+    const result = await capturedRun(["settings", "mode", "lite"]);
+    expect(result.code).toBe(0);
+    expect(transport.put).not.toHaveBeenCalled();
+    expect(result.stdout).toContain("更高模式生成的记忆仍保存着");
+    expect(result.stdout).toContain("memhub settings mode lite --yes");
+    const json = JSON.parse(
+      (await capturedRun(["settings", "mode", "ultra", "--json"])).stdout,
+    );
+    expect(json.state).toBe("preview");
+    expect(json.applied).toBe(false);
+    expect(json.effects).toEqual(["existing_memories_added_to_graph"]);
+  });
+
+  it("changes the mode with --yes and reports what changed", async () => {
+    transport.put.mockResolvedValue({
+      status: 200,
+      data: {
+        ...modeState("lite"),
+        previous_mode: "pro",
+        effects: ["higher_mode_memories_hidden"],
+      },
+    });
+    const result = await capturedRun(["settings", "mode", "lite", "--yes"]);
+    expect(result.code).toBe(0);
+    expect(transport.put).toHaveBeenCalledWith(
+      "/api/v1/settings/memory-mode",
+      { mode: "lite" },
+      undefined,
+    );
+    expect(result.stdout).toContain("已切换");
+    expect(result.stdout).toContain("原始对话：全部保留");
+  });
+
+  it("does nothing for the mode already set, and refuses bad input", async () => {
+    const same = await capturedRun(["settings", "mode", "pro", "--yes"]);
+    expect(same.code).toBe(0);
+    expect(same.stdout).toContain("已经是这个模式");
+    expect(transport.put).not.toHaveBeenCalled();
+    for (const bad of [
+      ["settings"],
+      ["settings", "mode", "max"],
+      ["settings", "mode", "--yes"],
+      ["settings", "mode", "lite", "pro"],
+      ["settings", "language", "en"],
+    ]) {
+      expect((await capturedRun(bad)).code, bad.join(" ")).toBe(2);
+    }
+  });
+
+  it("plans the same effects Core reports", () => {
+    expect(modeEffects("lite", "pro")).toEqual([
+      "earlier_raw_not_extracted",
+      "lite_search_stops",
+    ]);
+    expect(modeEffects("ultra", "lite")).toEqual([
+      "higher_mode_memories_hidden",
+      "evolve_stopped",
+    ]);
+    expect(modeEffects("pro", "pro")).toEqual([]);
+    expect(parseSettingsCommand(["mode", "ultra", "--yes"])).toEqual({
+      kind: "set",
+      mode: "ultra",
+      confirmed: true,
+    });
+  });
+});
+
+describe("memhub evolve", () => {
+  it("starts a run for the named user and returns at once", async () => {
+    transport.post.mockResolvedValue({ status: 202, data: receipt("running") });
+    const result = await capturedRun([
+      "evolve",
+      "--user-id",
+      "wife",
+      "--agent-id",
+      "all-apps",
+      "--max-batches",
+      "3",
+      "--json",
+    ]);
+    expect(result.code).toBe(0);
+    expect(transport.post).toHaveBeenCalledWith(
+      "/api/v2/evolve",
+      { user_id: "wife", agent_id: "all-apps", max_batches: 3 },
+      undefined,
+    );
+    const document = JSON.parse(result.stdout);
+    expect(document.state).toBe("running");
+    expect(document.run.results).toBeUndefined();
+  });
+
+  it("follows the run with --wait until it ends", async () => {
+    transport.post.mockResolvedValue({ status: 202, data: receipt("running") });
+    const reads = [receipt("running"), receipt("succeeded")];
+    transport.get.mockImplementation(async (url: string) =>
+      url === `/api/v2/evolve/${RUN_ID}`
+        ? { status: 200, data: reads.shift() }
+        : { status: 404, data: {} },
+    );
+    const result = await capturedRun(["evolve", "--wait"], {
+      sleep: async () => undefined,
+    });
+    expect(result.code).toBe(0);
+    expect(transport.post).toHaveBeenCalledWith(
+      "/api/v2/evolve",
+      {},
+      undefined,
+    );
+    expect(result.stdout).toContain("进化完成");
+    expect(result.stdout).toContain("2 轮");
+  });
+
+  it("names the mode when Core refuses outside ultra", async () => {
+    transport.post.mockResolvedValue({
+      status: 409,
+      data: { detail: "memory_mode_not_ultra" },
+    });
+    const result = await capturedRun(["evolve"]);
+    expect(result.code).toBe(1);
+    expect(result.stdout).toContain("memhub settings mode ultra --yes");
+    expect((await capturedRun(["evolve", "--max-batches", "0"])).code).toBe(2);
+    expect((await capturedRun(["evolve", "--now"])).code).toBe(2);
+  });
+});
+
+describe("status pipeline", () => {
+  it("reports pipeline.mode and the last evolve from memory-processing", async () => {
+    const processing = {
+      pending: 0,
+      fresh_pending: 0,
+      backlog_pending: 0,
+      outside_window_pending: 0,
+      running: 0,
+      waiting_for_model: 0,
+      completed: 3,
+      failed: 0,
+      blocked_stages: [],
+      pipeline: {
+        mode: "ultra",
+        configured_mode: "ultra",
+        forced_by_env: false,
+        last_evolve_at: "2026-11-01T03:00:05+08:00",
+        graph_build: { state: "succeeded", users: {} },
+        evolve: {
+          mode: "ultra",
+          schedule: { enabled: true, time: "03:00" },
+          running: [],
+          users: {
+            local: {
+              last_run_at: "2026-11-01T03:00:05+08:00",
+              last_result: { state: "succeeded", tasks_processed: 4 },
+              last_error: null,
+            },
+          },
+        },
+      },
+    };
+    const report = await composeStatusReport({
+      dataDir: "/tmp/isolated",
+      runtime: { state: "running", code: "ok", host: "127.0.0.1", port: 1 },
+      background: null,
+      reader: (async (path: string) =>
+        path === "/api/v1/memory-processing"
+          ? { state: "ok", data: processing }
+          : { state: "unavailable", code: "x" }) as never,
+    });
+    expect(report.pipeline?.mode).toBe("ultra");
+    expect(report.pipeline?.last_evolve_at).toBe("2026-11-01T03:00:05+08:00");
+    expect(report.pipeline?.evolve_users.local?.state).toBe("succeeded");
+    const text = renderStatusText(report, "MemHub");
+    expect(text).toContain("记忆模式：ultra");
+    expect(text).toContain("每日 03:00 自动进化");
+  });
+});
+
+async function capturedRun(
+  arguments_: string[],
+  controls: Parameters<typeof run>[2] = {},
+) {
+  let stdout = "";
+  let stderr = "";
+  const stdoutSpy = vi
+    .spyOn(process.stdout, "write")
+    .mockImplementation((chunk) => {
+      stdout += String(chunk);
+      return true;
+    });
+  const stderrSpy = vi
+    .spyOn(process.stderr, "write")
+    .mockImplementation((chunk) => {
+      stderr += String(chunk);
+      return true;
+    });
+  try {
+    return {
+      code: await run(
+        arguments_,
+        { ONE_MEMORY_DATA_DIR: "/synthetic/data" },
+        controls,
+      ),
+      stdout,
+      stderr,
+    };
+  } finally {
+    stdoutSpy.mockRestore();
+    stderrSpy.mockRestore();
+  }
+}

memhub/apps/cli/src/memory-mode-flow.ts

@@ -0,0 +1,372 @@
+/**
+ * `memhub settings mode` and `memhub evolve`.
+ *
+ * The mode (lite / pro / ultra) decides what the SDK does with every
+ * conversation, and switching it migrates nothing: what one mode made is not
+ * shown by a lower one, and raw kept by lite is not extracted until it is
+ * imported again. So a change prints its plan first and needs `--yes`, like
+ * every other command whose effect a person cannot simply take back.
+ *
+ * `memhub evolve` starts System 2 (ultra only) and, with `--wait`, follows
+ * the run until it ends.
+ */
+
+import type { CoreReader } from "./core-client.js";
+import type { CoreOperations, CorePutOperations } from "./core-operations.js";
+import { PRODUCT } from "./version.js";
+
+export const MEMORY_MODES = ["lite", "pro", "ultra"] as const;
+export type MemoryMode = (typeof MEMORY_MODES)[number];
+
+export type ModeEffect =
+  | "earlier_raw_not_extracted"
+  | "higher_mode_memories_hidden"
+  | "existing_memories_added_to_graph"
+  | "evolve_stopped"
+  | "lite_search_stops"
+  | "graph_backfill_already_running";
+
+const MODE_SUMMARY: Record<MemoryMode, string> = {
+  lite: "lite：不调用模型，保留全部原始对话，只在原始对话里检索",
+  pro: "pro：模型从对话中提取记忆（System 1）",
+  ultra: "ultra：pro + 每晚自动进化（System 2，使用知识图谱）",
+};
+
+const EFFECT_TEXT: Record<ModeEffect, string> = {
+  earlier_raw_not_extracted:
+    "lite 下已处理完的对话不会再被提取；只有尚未处理的和之后的新对话会由模型整理成记忆，检索立即改为记忆检索",
+  higher_mode_memories_hidden:
+    "更高模式生成的记忆仍保存着，但在这个模式下不会显示",
+  existing_memories_added_to_graph:
+    "已有记忆会在后台加入知识图谱（每个用户一次）",
+  evolve_stopped: "每晚的自动进化（System 2）停止",
+  lite_search_stops: "lite 只在原始对话里的检索立即停止，改为检索记忆",
+  graph_backfill_already_running: "知识图谱的补建已在进行中",
+};
+
+/** The same rule Core applies (`memory_mode_effects`), for the plan. */
+export function modeEffects(
+  previous: MemoryMode,
+  next: MemoryMode,
+): ModeEffect[] {
+  if (previous === next) return [];
+  const rank: Record<MemoryMode, number> = { lite: 0, pro: 1, ultra: 2 };
+  const effects: ModeEffect[] = [];
+  if (previous === "lite") {
+    effects.push("earlier_raw_not_extracted", "lite_search_stops");
+  }
+  if (rank[next] < rank[previous]) effects.push("higher_mode_memories_hidden");
+  if (previous === "ultra") effects.push("evolve_stopped");
+  if (next === "ultra") effects.push("existing_memories_added_to_graph");
+  return effects;
+}
+
+export type ModeCommand =
+  | { kind: "show" }
+  | { kind: "set"; mode: MemoryMode; confirmed: boolean };
+
+/** `settings mode [lite|pro|ultra] [--yes]` after `settings`; null = usage. */
+export function parseSettingsCommand(
+  values: readonly string[],
+): ModeCommand | null {
+  const rest = values.filter((value) => value !== "--json");
+  if (rest[0] !== "mode") return null;
+  const tail = rest.slice(1);
+  const confirmed = tail.includes("--yes");
+  const positional = tail.filter((value) => value !== "--yes");
+  if (tail.filter((value) => value === "--yes").length > 1) return null;
+  if (positional.length === 0) return confirmed ? null : { kind: "show" };
+  if (positional.length !== 1) return null;
+  const mode = positional[0] as MemoryMode;
+  if (!(MEMORY_MODES as readonly string[]).includes(mode)) return null;
+  return { kind: "set", mode, confirmed };
+}
+
+export interface ModeState {
+  mode: MemoryMode;
+  configured_mode: MemoryMode;
+  forced_by_env: boolean;
+  graph_build?: { state?: string } | null;
+  previous_mode?: MemoryMode;
+  effects?: ModeEffect[];
+}
+
+function modeState(value: unknown): ModeState | null {
+  if (!value || typeof value !== "object") return null;
+  const item = value as Record<string, unknown>;
+  if (
+    !(MEMORY_MODES as readonly unknown[]).includes(item.mode) ||
+    !(MEMORY_MODES as readonly unknown[]).includes(item.configured_mode) ||
+    typeof item.forced_by_env !== "boolean"
+  ) {
+    return null;
+  }
+  return item as unknown as ModeState;
+}
+
+export type ModeDocument = {
+  schema_version: 1;
+  product: string;
+  command: "settings mode";
+} & (
+  | { state: "ok"; mode: ModeState }
+  | {
+      state: "preview";
+      mode: ModeState;
+      target: MemoryMode;
+      effects: ModeEffect[];
+      applied: false;
+      next_step: string;
+    }
+  | { state: "unchanged"; mode: ModeState }
+  | { state: "committed"; mode: ModeState; effects: ModeEffect[] }
+  | { state: "rejected" | "unknown" | "unavailable"; code: string }
+);
+
+const BASE = {
+  schema_version: 1 as const,
+  product: PRODUCT,
+  command: "settings mode" as const,
+};
+
+export async function runModeCommand(
+  command: ModeCommand,
+  reader: CoreReader,
+  operations: CorePutOperations,
+): Promise<ModeDocument> {
+  const read = await reader<unknown>("/api/v1/settings/memory-mode");
+  if (read.state !== "ok") {
+    return { ...BASE, state: "unavailable", code: read.code };
+  }
+  const current = modeState(read.data);
+  if (current === null) {
+    return { ...BASE, state: "unavailable", code: "response_invalid" };
+  }
+  if (command.kind === "show") return { ...BASE, state: "ok", mode: current };
+  if (current.configured_mode === command.mode) {
+    return { ...BASE, state: "unchanged", mode: current };
+  }
+  if (!command.confirmed) {
+    return {
+      ...BASE,
+      state: "preview",
+      mode: current,
+      target: command.mode,
+      effects: modeEffects(current.mode, command.mode),
+      applied: false,
+      next_step: `${PRODUCT} settings mode ${command.mode} --yes`,
+    };
+  }
+  const result = await operations.put<unknown>("/api/v1/settings/memory-mode", {
+    mode: command.mode,
+  });
+  if (result.state !== "committed") {
+    return { ...BASE, state: result.state, code: result.code };
+  }
+  const changed = modeState(result.data);
+  if (changed === null) {
+    return { ...BASE, state: "unknown", code: "response_invalid" };
+  }
+  return {
+    ...BASE,
+    state: "committed",
+    mode: changed,
+    effects: changed.effects ?? [],
+  };
+}
+
+export function modeExitCode(document: ModeDocument): number {
+  return ["ok", "preview", "unchanged", "committed"].includes(document.state)
+    ? 0
+    : 1;
+}
+
+export function renderModeText(document: ModeDocument): string {
+  if ("code" in document) {
+    return `记忆模式未能${document.state === "unavailable" ? "读取" : "修改"}：${document.code}\n`;
+  }
+  const mode = document.mode;
+  const lines = [`当前模式：${MODE_SUMMARY[mode.mode]}`];
+  if (mode.forced_by_env) {
+    lines.push(
+      `设置里是 ${mode.configured_mode}；环境变量 ONE_MEMORY_GRAPH_STORE 强制使用 ultra。`,
+    );
+  }
+  lines.push("原始对话：全部保留（任何模式都不删除）");
+  if (document.state === "unchanged") {
+    lines.push("已经是这个模式，未做改动。");
+  }
+  if (document.state === "preview") {
+    lines.push(`将切换到：${MODE_SUMMARY[document.target]}`);
+    for (const effect of document.effects)
+      lines.push(`- ${EFFECT_TEXT[effect]}`);
+    lines.push(`未执行任何改动。确认后运行：${document.next_step}`);
+  }
+  if (document.state === "committed") {
+    lines.push("已切换。");
+    for (const effect of document.effects)
+      lines.push(`- ${EFFECT_TEXT[effect]}`);
+  }
+  return `${lines.join("\n")}\n`;
+}
+
+export type EvolveCommand = {
+  agentId?: string;
+  maxBatches?: number;
+  wait: boolean;
+};
+
+/** `evolve [--agent-id <id>] [--max-batches <n>] [--wait]`; null = usage. */
+export function parseEvolveCommand(
+  values: readonly string[],
+): EvolveCommand | null {
+  const parsed: EvolveCommand = { wait: false };
+  const rest = values.filter((value) => value !== "--json");
+  for (let index = 0; index < rest.length; index += 1) {
+    const flag = rest[index];
+    if (flag === "--wait" && !parsed.wait) {
+      parsed.wait = true;
+    } else if (flag === "--agent-id" && parsed.agentId === undefined) {
+      const value = rest[index + 1];
+      if (!value || !/^[A-Za-z0-9._:-]{1,128}$/.test(value)) return null;
+      parsed.agentId = value;
+      index += 1;
+    } else if (flag === "--max-batches" && parsed.maxBatches === undefined) {
+      const value = Number(rest[index + 1]);
+      if (!Number.isInteger(value) || value < 1 || value > 1000) return null;
+      parsed.maxBatches = value;
+      index += 1;
+    } else {
+      return null;
+    }
+  }
+  return parsed;
+}
+
+export type EvolveDocument = {
+  schema_version: 1;
+  product: string;
+  command: "evolve";
+} & (
+  | { state: "running" | "succeeded" | "failed"; run: Record<string, unknown> }
+  | { state: "rejected" | "unknown" | "unavailable"; code: string }
+);
+
+const RUN_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/;
+
+function evolveRun(value: unknown): Record<string, unknown> | null {
+  if (!value || typeof value !== "object") return null;
+  const run = value as Record<string, unknown>;
+  if (
+    typeof run.run_id !== "string" ||
+    !RUN_ID.test(run.run_id) ||
+    !["running", "succeeded", "failed"].includes(String(run.state))
+  ) {
+    return null;
+  }
+  return run;
+}
+
+export async function runEvolveCommand(
+  command: EvolveCommand,
+  options: {
+    userId?: string;
+    reader: CoreReader;
+    operations: CoreOperations;
+    sleep?: (ms: number) => Promise<void>;
+    pollMs?: number;
+  },
+): Promise<EvolveDocument> {
+  const base = {
+    schema_version: 1 as const,
+    product: PRODUCT,
+    command: "evolve" as const,
+  };
+  const body: Record<string, unknown> = {};
+  if (options.userId !== undefined) body.user_id = options.userId;
+  if (command.agentId !== undefined) body.agent_id = command.agentId;
+  if (command.maxBatches !== undefined) body.max_batches = command.maxBatches;
+  const started = await options.operations.post<unknown>(
+    "/api/v2/evolve",
+    body,
+  );
+  if (started.state !== "committed") {
+    return { ...base, state: started.state, code: started.code };
+  }
+  let run = evolveRun(started.data);
+  if (run === null)
+    return { ...base, state: "unknown", code: "response_invalid" };
+  const sleep =
+    options.sleep ??
+    ((ms: number) => new Promise((done) => setTimeout(done, ms)));
+  while (command.wait && run.state === "running") {
+    await sleep(options.pollMs ?? 2_000);
+    const read = await options.reader<unknown>(
+      `/api/v2/evolve/${run.run_id}`,
+      options.userId === undefined ? {} : { user_id: options.userId },
+    );
+    if (read.state !== "ok") {
+      return { ...base, state: "unavailable", code: read.code };
+    }
+    const next = evolveRun(read.data);
+    if (next === null)
+      return { ...base, state: "unknown", code: "response_invalid" };
+    run = next;
+  }
+  return {
+    ...base,
+    state: run.state as "running" | "succeeded" | "failed",
+    run: projectRun(run),
+  };
+}
+
+/** Counts and codes; SDK result bodies stay in the Core receipt. */
+function projectRun(run: Record<string, unknown>): Record<string, unknown> {
+  const keep = [
+    "run_id",
+    "user_id",
+    "agent_id",
+    "state",
+    "trigger",
+    "started_at",
+    "finished_at",
+    "rounds",
+    "tasks_processed",
+    "batch_failures",
+    "cycle_status",
+    "should_continue",
+    "continuation_reason",
+    "remaining_clusters",
+    "remaining_actionable_clusters",
+    "remaining_fresh_facts",
+    "capped",
+    "elapsed_ms",
+    "error_code",
+    "failure_kind",
+  ];
+  return Object.fromEntries(
+    keep.filter((key) => key in run).map((key) => [key, run[key]]),
+  );
+}
+
+export function evolveExitCode(document: EvolveDocument): number {
+  return document.state === "running" || document.state === "succeeded" ? 0 : 1;
+}
+
+export function renderEvolveText(document: EvolveDocument): string {
+  if ("code" in document) {
+    return document.code === "memory_mode_not_ultra"
+      ? "进化只在 ultra 模式下可用。切换：memhub settings mode ultra --yes\n"
+      : `进化未能开始：${document.code}\n`;
+  }
+  const run = document.run;
+  if (document.state === "running") {
+    return `进化已开始（${run.run_id}），在后台运行；加 --wait 等待结果。\n`;
+  }
+  if (document.state === "failed") {
+    return `进化失败（${run.run_id}）：${String(run.error_code ?? "unknown")}\n`;
+  }
+  return `进化完成（${run.run_id}）：${run.rounds} 轮，处理 ${run.tasks_processed} 项${
+    run.cycle_status ? `，周期 ${run.cycle_status}` : ""
+  }${run.capped ? "，已到轮数上限，可再次运行继续" : ""}\n`;
+}

memhub/apps/cli/src/profile-versions-cli.test.ts

@@ -0,0 +1,166 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { run } from "./cli.js";
+
+const transport = vi.hoisted(() => ({
+  get: vi.fn(),
+  post: vi.fn(),
+  put: vi.fn(),
+  patch: vi.fn(),
+  delete: vi.fn(),
+}));
+vi.mock("axios", () => ({
+  default: {
+    create: vi.fn(() => transport),
+    get: vi.fn(),
+    post: vi.fn(),
+    isAxiosError: () => false,
+  },
+}));
+
+const runtime = vi.hoisted(() => ({
+  inspectRuntimeStatus: vi.fn(),
+  discoverRuntimeAttachment: vi.fn(),
+}));
+vi.mock("@one-memory/runtime-control", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@one-memory/runtime-control")>()),
+  ...runtime,
+}));
+
+const VERSIONS = {
+  user_id: "wife",
+  request_id: "r-versions",
+  elapsed_ms: 2,
+  chains: [
+    {
+      memory_id: "l0-head",
+      content: "The user's name is Fei.",
+      status: "active",
+      memory_at: null,
+      gmt_created: 1_790_000_100,
+      basic_info: { name: "Fei" },
+      versions: [
+        {
+          memory_id: "l0-head",
+          content: "The user's name is Fei.",
+          memory_at: null,
+          gmt_created: 1_790_000_100,
+          current: true,
+        },
+        {
+          memory_id: "l0-old",
+          content: "The user's name is 费天翔.",
+          memory_at: null,
+          gmt_created: 1_790_000_000,
+          current: false,
+        },
+      ],
+    },
+  ],
+};
+
+beforeEach(() => {
+  vi.clearAllMocks();
+  for (const method of Object.values(transport)) method.mockReset();
+  runtime.inspectRuntimeStatus.mockResolvedValue({
+    state: "running",
+    code: "runtime_verified",
+    instanceId: "instance-1",
+    host: "127.0.0.1",
+    port: 8765,
+    version: "1.2.3",
+  });
+  runtime.discoverRuntimeAttachment.mockResolvedValue({
+    instanceId: "instance-1",
+    apiToken: "synthetic-runtime-token",
+    host: "127.0.0.1",
+    port: 8765,
+    servicePid: 4242,
+    health: {
+      status: "ok",
+      service: "one-memory-core",
+      version: "1.2.3",
+      instance_id: "instance-1",
+    },
+  });
+  transport.get.mockImplementation(async (url: string) =>
+    url === "/api/v2/user-profile/versions"
+      ? { status: 200, data: VERSIONS }
+      : { status: 404, data: {} },
+  );
+});
+
+describe("memhub profile --versions", () => {
+  it("reads the L0 version chain for the named user", async () => {
+    const result = await capturedRun([
+      "profile",
+      "--versions",
+      "--user-id",
+      "wife",
+      "--json",
+    ]);
+    expect(result.code).toBe(0);
+    expect(transport.get).toHaveBeenCalledWith(
+      "/api/v2/user-profile/versions",
+      { params: { user_id: "wife" } },
+    );
+    const document = JSON.parse(result.stdout);
+    expect(document.state).toBe("ok");
+    expect(document.chains[0].versions).toHaveLength(2);
+  });
+
+  it("marks the current version in text and names no user for local", async () => {
+    const result = await capturedRun(["profile", "--versions"]);
+    expect(result.code).toBe(0);
+    const [url, config] = transport.get.mock.calls[0] ?? [];
+    expect(url).toBe("/api/v2/user-profile/versions");
+    expect(config?.params?.user_id).toBeUndefined();
+    expect(result.stdout).toContain("* l0-head");
+    expect(result.stdout).toContain("  l0-old");
+    expect(result.stdout).toContain("费天翔");
+  });
+
+  it("refuses an unknown flag and a repeated --versions", async () => {
+    expect((await capturedRun(["profile", "--version-chain"])).code).toBe(2);
+    expect(
+      (await capturedRun(["profile", "--versions", "--versions"])).code,
+    ).toBe(2);
+    expect(transport.get).not.toHaveBeenCalled();
+  });
+
+  it("reports a malformed answer as invalid", async () => {
+    transport.get.mockResolvedValue({ status: 200, data: { chains: "x" } });
+    const result = await capturedRun(["profile", "--versions", "--json"]);
+    expect(result.code).toBe(1);
+    expect(JSON.parse(result.stdout).code).toBe("response_invalid");
+  });
+});
+
+async function capturedRun(arguments_: string[]) {
+  let stdout = "";
+  let stderr = "";
+  const stdoutSpy = vi
+    .spyOn(process.stdout, "write")
+    .mockImplementation((chunk) => {
+      stdout += String(chunk);
+      return true;
+    });
+  const stderrSpy = vi
+    .spyOn(process.stderr, "write")
+    .mockImplementation((chunk) => {
+      stderr += String(chunk);
+      return true;
+    });
+  try {
+    return {
+      code: await run(arguments_, {
+        ONE_MEMORY_DATA_DIR: "/synthetic/data",
+      }),
+      stdout,
+      stderr,
+    };
+  } finally {
+    stdoutSpy.mockRestore();
+    stderrSpy.mockRestore();
+  }
+}

memhub/apps/cli/src/source-progress-report.ts

@@ -14,6 +14,11 @@ import {
   validateProcessing,
   validateSources,
 } from "./core-responses.js";
+import {
+  describePipeline,
+  type PipelineSummary,
+  pipelineLine,
+} from "./status-report.js";
 import { PRODUCT } from "./version.js";
 
 type ReadState =
@@ -84,6 +89,8 @@ export interface ProgressDocument {
     summary_state: ReadState;
     summary: Record<string, unknown> | null;
   };
+  /** The SDK mode and last evolve per user; absent on an older Core. */
+  pipeline?: PipelineSummary | null;
 }
 
 export async function readSources(
@@ -283,6 +290,10 @@ export async function readProgress(
       summary:
         processing.state === "ok" ? projectProcessing(processing.data) : null,
     },
+    pipeline:
+      processingRead.state === "ok"
+        ? describePipeline(processingRead.data)
+        : null,
   };
 }
 
@@ -353,6 +364,18 @@ export function renderProgressText(document: ProgressDocument): string {
       ? `记忆整理：待处理 ${document.generation.summary.pending} · 进行中 ${document.generation.summary.running} · 等待模型 ${document.generation.summary.waiting_for_model} · 失败 ${document.generation.summary.failed}`
       : "记忆整理：暂时无法读取",
   );
+  if (document.pipeline) {
+    lines.push(`记忆模式：${pipelineLine(document.pipeline)}`);
+    for (const [userId, user] of Object.entries(
+      document.pipeline.evolve_users,
+    )) {
+      lines.push(
+        `  进化 ${userId}：${user.last_run_at ?? "尚未进化"}${
+          user.last_error ? `（失败：${user.last_error}）` : ""
+        }`,
+      );
+    }
+  }
   if (document.imports.output_truncated) {
     lines.push("这里只显示汇总；完整记录请使用 --json。");
   }

memhub/apps/cli/src/status-report.ts

@@ -57,6 +57,9 @@ export interface StatusDocument {
     broker?: BackgroundHealth["broker"];
   };
   sections: Record<string, Section>;
+  /** R10: the SDK mode and System 2 state; null when the Runtime is not read
+   * or is older than the setting. */
+  pipeline?: PipelineSummary | null;
   summary: {
     state: "ok" | "partial";
     degraded: string[];
@@ -64,6 +67,88 @@ export interface StatusDocument {
   };
 }
 
+export interface PipelineSummary {
+  mode: "lite" | "pro" | "ultra";
+  configured_mode: "lite" | "pro" | "ultra";
+  forced_by_env: boolean;
+  last_evolve_at: string | null;
+  evolve_schedule: { enabled: boolean; time: string } | null;
+  evolve_running: number;
+  evolve_users: Record<
+    string,
+    {
+      last_run_at: string | null;
+      last_error: string | null;
+      state: string | null;
+    }
+  >;
+  graph_build: string | null;
+}
+
+const MODES = new Set(["lite", "pro", "ultra"]);
+
+/** Projection of `pipeline` in `/api/v1/memory-processing`; null if absent or malformed. */
+export function describePipeline(value: unknown): PipelineSummary | null {
+  if (!value || typeof value !== "object") return null;
+  const pipeline = (value as { pipeline?: unknown }).pipeline;
+  if (!pipeline || typeof pipeline !== "object") return null;
+  const item = pipeline as Record<string, unknown>;
+  if (
+    !MODES.has(String(item.mode)) ||
+    !MODES.has(String(item.configured_mode))
+  ) {
+    return null;
+  }
+  const evolve =
+    item.evolve && typeof item.evolve === "object"
+      ? (item.evolve as Record<string, unknown>)
+      : {};
+  const schedule =
+    evolve.schedule && typeof evolve.schedule === "object"
+      ? (evolve.schedule as Record<string, unknown>)
+      : null;
+  const users: PipelineSummary["evolve_users"] = {};
+  if (evolve.users && typeof evolve.users === "object") {
+    for (const [userId, raw] of Object.entries(
+      evolve.users as Record<string, unknown>,
+    )) {
+      const user =
+        raw && typeof raw === "object" ? (raw as Record<string, unknown>) : {};
+      const result =
+        user.last_result && typeof user.last_result === "object"
+          ? (user.last_result as Record<string, unknown>)
+          : {};
+      users[userId] = {
+        last_run_at:
+          typeof user.last_run_at === "string" ? user.last_run_at : null,
+        last_error:
+          typeof user.last_error === "string" ? user.last_error : null,
+        state: typeof result.state === "string" ? result.state : null,
+      };
+    }
+  }
+  const graph =
+    item.graph_build && typeof item.graph_build === "object"
+      ? (item.graph_build as Record<string, unknown>)
+      : {};
+  return {
+    mode: item.mode as PipelineSummary["mode"],
+    configured_mode: item.configured_mode as PipelineSummary["mode"],
+    forced_by_env: item.forced_by_env === true,
+    last_evolve_at:
+      typeof item.last_evolve_at === "string" ? item.last_evolve_at : null,
+    evolve_schedule:
+      schedule &&
+      typeof schedule.enabled === "boolean" &&
+      typeof schedule.time === "string"
+        ? { enabled: schedule.enabled, time: schedule.time }
+        : null,
+    evolve_running: Array.isArray(evolve.running) ? evolve.running.length : 0,
+    evolve_users: users,
+    graph_build: typeof graph.state === "string" ? graph.state : null,
+  };
+}
+
 const SECTION_NAMES = [
   "sources",
   "sync",
@@ -97,6 +182,7 @@ export async function composeStatusReport(options: {
   attachFailure?: string;
 }): Promise<StatusDocument> {
   const sections: Record<string, Section> = {};
+  let pipeline: PipelineSummary | null = null;
   if (options.reader === null) {
     const state: SectionState =
       options.attachFailure === undefined ? "not_checked" : "unavailable";
@@ -118,6 +204,8 @@ export async function composeStatusReport(options: {
       validateProcessing,
       describeProcessing,
     );
+    pipeline =
+      processing.state === "ok" ? describePipeline(processing.data) : null;
     sections.model = section(model, validateModelProfiles, describeModel);
     sections.memories = section(
       memories,
@@ -169,6 +257,7 @@ export async function composeStatusReport(options: {
     },
     bridge,
     sections,
+    pipeline,
     summary: {
       state:
         degraded.length === 0 && notChecked.length === 0 ? "ok" : "partial",
@@ -205,6 +294,7 @@ export function renderStatusText(
       `${SECTION_LABELS[name]}：${sectionLine(report.sections[name])}`,
     );
   }
+  if (report.pipeline) lines.push(`记忆模式：${pipelineLine(report.pipeline)}`);
   if (report.summary.state === "partial") {
     const notes = [
       report.summary.degraded.length > 0
@@ -220,6 +310,28 @@ export function renderStatusText(
   return lines.join("\n");
 }
 
+export function pipelineLine(pipeline: PipelineSummary): string {
+  const parts: string[] = [pipeline.mode];
+  if (pipeline.forced_by_env)
+    parts.push("（ONE_MEMORY_GRAPH_STORE 强制 ultra）");
+  if (pipeline.mode === "ultra") {
+    const schedule = pipeline.evolve_schedule;
+    parts.push(
+      schedule?.enabled
+        ? `，每日 ${schedule.time} 自动进化`
+        : "，自动进化已关闭",
+    );
+    parts.push(
+      pipeline.last_evolve_at
+        ? `，上次进化 ${pipeline.last_evolve_at}`
+        : "，尚未进化",
+    );
+    if (pipeline.evolve_running > 0)
+      parts.push(`，进行中 ${pipeline.evolve_running}`);
+  }
+  return parts.join("");
+}
+
 function runtimeLine(report: StatusDocument): string {
   if (report.runtime.state !== "running") {
     return report.runtime.state === "stopped" ? "未运行" : "暂时无法确认";

memhub/apps/core/src/one_memory/cli.py

@@ -584,6 +584,7 @@ def _sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
         }
     else:
         from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
+        from one_memory.hy_sdk_runtime import builtin_retrieval_snapshot
         from one_memory.sdk_data_plane_migration import (
             MemoryAuthority,
             SdkDataPlaneMigrationStore,
@@ -595,14 +596,14 @@ def _sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
 
         authority = SdkDataPlaneMigrationStore(database_path).authority()
         if authority is MemoryAuthority.SDK:
-            profile = read_active_profile(database_path)
+            profile = read_active_profile(database_path) or builtin_retrieval_snapshot()
             leftovers = bool(
                 legacy_tables_present(database_path)
                 or abandoned_staging_present(
                     data_dir=settings.data_dir,
                     namespace=HY_MEMORY_NAMESPACE,
                 )
-                or (profile is not None and legacy_vector_collections(settings.data_dir, profile))
+                or legacy_vector_collections(settings.data_dir, profile)
             )
             state = "required" if leftovers else "not_required"
         else:

memhub/apps/core/src/one_memory/context_models.py

@@ -5,6 +5,7 @@ from typing import Annotated, Literal
 
 from pydantic import BaseModel, ConfigDict, Field, model_validator
 
+from one_memory.sdk_search_options import SdkSearchOptions
 from one_memory.user_scope import DEFAULT_USER_ID, UserId
 
 ContextKind = Literal["raw", "memory"]
@@ -49,12 +50,17 @@ class ContextSearchRequest(BaseModel):
         return self
 
 
-class SdkContextSearchRequest(ContextSearchRequest):
+class SdkContextSearchRequest(ContextSearchRequest, SdkSearchOptions):
     """`POST /api/v2/context/search`: the three SDK recall lanes are set apart.
 
     `limit` is the normal lane's top-k and the cap on returned items;
     `schema_limit` and `intention_limit` are the SDK's schema (L6) and
     proactive intention (L7) lanes. A caller's values are passed through.
+
+    The `SdkSearchOptions` fields reach the memory lane's `client.search`;
+    the raw lane takes the floor and the source, session, agent and time
+    filters (never wider than the caller's raw authority). Absent, both
+    lanes make the call they always made.
     """
 
     limit: int = Field(default=10, ge=1, le=20, strict=True)

memhub/apps/core/src/one_memory/context_routes.py

@@ -24,14 +24,57 @@ from one_memory.hy_runtime_config import HY_MEMORY_READER
 from one_memory.hy_sdk_runtime import sdk_relevance_floor
 from one_memory.hy_sdk_types import (
     HySdkMemoryRecord,
+    parse_sdk_lane_search,
     parse_sdk_raw_search,
     parse_sdk_search,
 )
 from one_memory.imports.storage import EvidenceStore
+from one_memory.sdk_search_options import SdkSearchFilters, SearchLane
+from one_memory.semantic_deadline import SemanticMemoryHit
 from one_memory.user_scope import owns
 
 CALL_EVENT_RESPONSE_BUDGET_MS = 250
 
+#: Pass-through filters the raw lane honours. Scene, layers, reader and the
+#: schema/working floors describe memory lanes; the raw lane is always the
+#: chat store's L1 raw layer.
+_RAW_LANE_FILTERS = ("created_after", "session_ids", "agent_ids", "exclude_sources", "rerank")
+
+
+def raw_lane_sources(
+    allowed: Sequence[str] | frozenset[str] | None,
+    requested: Sequence[str] | None,
+) -> list[str] | None:
+    """The raw lane's source filter: the caller's, never wider than its authority.
+
+    `allowed` empty or None means every source (the call Core always made).
+    None back means nothing is left to search.
+    """
+    allowed_list = list(allowed or ())
+    if requested is None:
+        return allowed_list
+    chosen = [
+        source for source in dict.fromkeys(requested) if not allowed_list or source in allowed_list
+    ]
+    return chosen or None
+
+
+def raw_lane_overrides(value: SdkSearchFilters) -> dict[str, object]:
+    """The caller's raw-lane overrides: its floor and its row filters."""
+    overrides = value.filter_kwargs()
+    found = {name: overrides[name] for name in _RAW_LANE_FILTERS if name in overrides}
+    floor = value.lane_floor("raw")
+    if floor is not None:
+        found["min_score"] = floor
+    return found
+
+
+def hit_namespace(hit: SemanticMemoryHit) -> dict[str, str]:
+    """The `get` namespace of a hit: nothing for chat, kind and variant otherwise."""
+    if hit.kind in {"experience", "fact"} and hit.variant is not None:
+        return {"kind": hit.kind, "variant": hit.variant}
+    return {}
+
 
 def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
     """SDK-native context: no Evidence/Governance ids or fabricated revisions."""
@@ -67,26 +110,36 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
             client = await client_handle.client()
         except Exception:
             raise HTTPException(503, "sdk_context_unavailable") from None
+        memory_mode = getattr(request.app.state, "memory_mode", None)
+        lite = callable(memory_mode) and memory_mode() == "lite"
 
         async def raw_lane() -> tuple[SdkRawContextItem, ...]:
             if "raw" not in value.kinds or not authority.raw_allowed:
                 return ()
+            sources = raw_lane_sources(authority.raw_sources, value.include_sources)
+            if sources is None:
+                return ()
             raw = await asyncio.to_thread(
                 client.search,
                 value.query,
-                user_ids=[value.user_id],
-                agent_ids=[],
-                include_sources=list(authority.raw_sources or ()),
-                limit=value.limit,
-                min_score=0.0,
-                schema_min_score=0.0,
-                schema_limit=0,
-                intention_limit=0,
-                scene="chat",
-                reader=HY_MEMORY_READER,
-                layers=["l1_raw"],
+                **{
+                    "user_ids": [value.user_id],
+                    "agent_ids": [],
+                    "include_sources": sources,
+                    "limit": value.limit,
+                    "min_score": 0.0,
+                    "schema_min_score": 0.0,
+                    "schema_limit": 0,
+                    "intention_limit": 0,
+                    "scene": "chat",
+                    "reader": HY_MEMORY_READER,
+                    "layers": ["l1_raw"],
+                    **raw_lane_overrides(value),
+                },
+            )
+            hits = tuple(
+                hit for hit in parse_sdk_raw_search(raw) if value.passes_floor("raw", hit.score)
             )
-            hits = parse_sdk_raw_search(raw)
             return tuple(
                 SdkRawContextItem(
                     raw_id=hit.raw_id,
@@ -104,25 +157,41 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter:
             )
 
         async def memory_lane() -> tuple[SdkMemoryContextItem, ...]:
-            if "memory" not in value.kinds or not authority.memory_allowed:
+            # lite extracts nothing: its search answers raw turns, which the
+            # raw lane already reads under the raw authority.
+            if "memory" not in value.kinds or not authority.memory_allowed or lite:
                 return ()
+            overrides = value.search_overrides()
             raw = await asyncio.to_thread(
                 client.search,
                 value.query,
-                user_ids=[value.user_id],
-                agent_ids=[],
-                limit=value.limit,
-                min_score=sdk_relevance_floor(),
-                schema_min_score=sdk_relevance_floor(),
-                schema_limit=value.schema_limit,
-                intention_limit=value.intention_limit,
-                scene="normal",
-                reader=HY_MEMORY_READER,
+                **{
+                    "user_ids": [value.user_id],
+                    "agent_ids": [],
+                    "limit": value.limit,
+                    "min_score": sdk_relevance_floor(),
+                    "schema_min_score": sdk_relevance_floor(),
+                    "schema_limit": value.schema_limit,
+                    "intention_limit": value.intention_limit,
+                    "scene": "normal",
+                    "reader": HY_MEMORY_READER,
+                    **overrides,
+                },
+            )
+            hits = (
+                tuple(
+                    hit
+                    for hit in parse_sdk_lane_search(
+                        raw, error_code="hy_memory_search_shape_unknown"
+                    )
+                    if value.passes_floor(cast(SearchLane, hit.lane), hit.score)
+                )
+                if overrides
+                else parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown")
             )
-            hits = parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown")
             items: list[SdkMemoryContextItem] = []
             for hit in hits:
-                record = await asyncio.to_thread(client.get, hit.memory_id)
+                record = await asyncio.to_thread(client.get, hit.memory_id, **hit_namespace(hit))
                 memory = None if record is None else HySdkMemoryRecord.from_sdk(record)
                 # `get` takes no user: a hit is kept only if it is the caller's.
                 if (

memhub/apps/core/src/one_memory/hy_graph_admin.py

@@ -71,6 +71,16 @@ def graph_store_enabled() -> bool:
     return raw != "none"
 
 
+def effective_memory_mode(configured: str = "pro") -> str:
+    """The SDK mode clients are built with: the setting, unless dev forces ultra.
+
+    `ONE_MEMORY_GRAPH_STORE` (any value but empty or `none`) predates the
+    setting and keeps working for development: it forces ultra whatever the
+    configured mode. Without it the configured mode (default pro) applies.
+    """
+    return "ultra" if graph_store_enabled() else configured
+
+
 def inspect_graph_width_mismatch(config: Any) -> GraphWidthMismatch | None:
     """Return a mismatch without renaming or deleting anything."""
     db_path = Path(config.graph_store.db_path)
@@ -83,11 +93,12 @@ def inspect_profile_graph_width(
     *,
     api_key: str | None,
     data_dir: Path,
+    mode: str | None = None,
 ) -> GraphWidthMismatch | None:
     """Inspect the graph path and width selected by the active profile."""
     del api_key
     return _inspect_graph_width(
-        _profile_graph_path(data_dir),
+        _profile_graph_path(data_dir, mode),
         _profile_embedding_width(profile),
     )
 
@@ -98,11 +109,12 @@ def decide_profile_graph_width(
     api_key: str | None,
     data_dir: Path,
     decision: GraphWidthDecision,
+    mode: str | None = None,
 ) -> GraphWidthOutcome:
     """Apply the user's decision to the active profile's graph."""
     del api_key
     config = _GraphAdminConfig(
-        graph_store=_GraphStoreConfig(str(_profile_graph_path(data_dir))),
+        graph_store=_GraphStoreConfig(str(_profile_graph_path(data_dir, mode))),
         embedder=_EmbeddingWidthConfig(_profile_embedding_width(profile)),
     )
     return apply_graph_width_decision(
@@ -240,8 +252,9 @@ def _inspect_graph_width(
     )
 
 
-def _profile_graph_path(data_dir: Path) -> Path:
-    name = "graph" if graph_store_enabled() else "graph-disabled"
+def _profile_graph_path(data_dir: Path, mode: str | None = None) -> Path:
+    ultra = mode == "ultra" if mode is not None else graph_store_enabled()
+    name = "graph" if ultra else "graph-disabled"
     return data_dir / "hy-memory" / HY_MEMORY_NAMESPACE / name
 
 

memhub/apps/core/src/one_memory/hy_sdk_runtime.py

@@ -31,7 +31,7 @@ from one_memory.embedding_assets import (
 )
 from one_memory.hy_graph_admin import (
     GraphWidthMismatchError,
-    graph_store_enabled,
+    effective_memory_mode,
     inspect_graph_width_mismatch,
 )
 from one_memory.hy_sdk_types import HySdkRuntimeError
@@ -170,11 +170,14 @@ class SdkClientHandle:
         api_key: str | None,
         data_dir: Path,
         http_transport_factory: Callable[[], Any] | None = None,
+        mode: str | None = None,
     ) -> None:
         self._profile = profile
         self._api_key = api_key
         self._data_dir = data_dir
         self._http_transport_factory = http_transport_factory
+        #: The SDK mode this handle's client is built with, fixed for its life.
+        self.mode = mode or effective_memory_mode()
         self._lease: ClientLease | None = None
         self._lock = asyncio.Lock()
 
@@ -189,6 +192,7 @@ class SdkClientHandle:
                     api_key=self._api_key,
                     data_dir=self._data_dir,
                     http_transport_factory=self._http_transport_factory,
+                    mode=self.mode,
                 )
             return self._lease.client
 
@@ -217,6 +221,33 @@ def builtin_retrieval_snapshot() -> ModelProfileSnapshot:
     )
 
 
+def retrieval_view(profile: ModelProfileSnapshot) -> ModelProfileSnapshot:
+    """The read side of `profile`: its embedding space, and no generation.
+
+    Reads must open the collection writes fill (#242), and the collection is
+    named by the embedding only. Generation stays the built-in placeholder, so
+    a read never needs the generation model. A remote embedding still needs
+    the profile's API key, which `build_sdk_config` uses as the embedding key.
+    """
+    provider = profile.embedding_provider_kind or profile.provider_kind
+    endpoint: str | None = None
+    if provider is not ProviderKind.LOCAL_ONNX:
+        endpoint = profile.embedding_base_url
+        if not endpoint:
+            raise HySdkRuntimeError("hy_sdk_model_endpoint_missing")
+    return builtin_retrieval_snapshot().model_copy(
+        update={
+            "profile_id": profile.profile_id,
+            "revision": profile.revision,
+            "embedding_provider_kind": provider,
+            "embedding_model": profile.embedding_model,
+            "embedding_base_url": endpoint,
+            "embedding_dimensions": profile.embedding_dimensions,
+            "has_api_key": profile.has_api_key,
+        }
+    )
+
+
 def sdk_relevance_floor() -> float:
     raw = os.environ.get("ONE_MEMORY_HY_RELEVANCE_FLOOR", "").strip()
     if not raw:
@@ -249,10 +280,8 @@ def build_sdk_config(
         embedding_key = None
         dimensions = profile.embedding_dimensions or BUILTIN_EMBEDDING_DIMENSIONS
     else:
-        embedding_endpoint = _openai_endpoint(
-            embedding_provider,
-            profile.embedding_base_url or profile.base_url,
-        )
+        # Never the generation endpoint: the collection name hashes this one.
+        embedding_endpoint = _openai_endpoint(embedding_provider, profile.embedding_base_url)
         embedding_key = "ollama" if embedding_provider is ProviderKind.OLLAMA else api_key
         if embedding_provider is ProviderKind.OPENAI_COMPATIBLE and not embedding_key:
             raise HySdkRuntimeError("hy_sdk_embedding_api_key_missing")
@@ -327,9 +356,12 @@ def acquire_client(
     api_key: str | None,
     data_dir: Path,
     http_transport_factory: Callable[[], Any] | None = None,
+    mode: str | None = None,
 ) -> ClientLease:
     """Construct a public SDK client and return its explicit runtime lease.
 
+    `mode` is the SDK pipeline mode (lite / pro / ultra); None keeps the
+    pre-setting default (ultra when `ONE_MEMORY_GRAPH_STORE` is set, else pro).
     Waits while `hold_stores_offline` holds this home's stores.
     """
     key = data_dir.resolve()
@@ -343,6 +375,7 @@ def acquire_client(
             api_key=api_key,
             data_dir=data_dir,
             http_transport_factory=http_transport_factory,
+            mode=mode or effective_memory_mode(),
         )
     finally:
         with _STORE_GATE:
@@ -381,6 +414,7 @@ def _acquire_client(
     api_key: str | None,
     data_dir: Path,
     http_transport_factory: Callable[[], Any] | None,
+    mode: str = "pro",
 ) -> ClientLease:
     config = build_sdk_config(profile, api_key=api_key, data_dir=data_dir)
     config.http_transport_factory = http_transport_factory
@@ -398,7 +432,7 @@ def _acquire_client(
         with constructing:
             client = HyMemoryClient(
                 config=config,
-                mode="ultra" if graph_store_enabled() else "pro",
+                mode=mode,
                 runtime=runtime,
                 embedding_service=embedding_service,
             )
@@ -467,7 +501,7 @@ def vector_collection_bases(profile: ModelProfileSnapshot) -> tuple[str, tuple[s
         endpoint, dimensions = None, profile.embedding_dimensions or BUILTIN_EMBEDDING_DIMENSIONS
         legacy_endpoint = "local-onnx"
     else:
-        endpoint = _openai_endpoint(provider, profile.embedding_base_url or profile.base_url)
+        endpoint = _openai_endpoint(provider, profile.embedding_base_url)
         dimensions = profile.embedding_dimensions
         legacy_endpoint = endpoint
     legacy_model = profile.embedding_model or "bge-small-zh-v1.5"

memhub/apps/core/src/one_memory/hy_sdk_types.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 from collections.abc import Mapping
 from datetime import UTC, datetime
 from typing import Literal, NamedTuple, cast
@@ -391,6 +392,8 @@ class HySdkRawRecord(BaseModel):
     memory_at: int | None = None
     gmt_created: int | None = None
     custom: dict[str, object] = Field(default_factory=dict)
+    #: active, or shadow once pro extracted it; None when the SDK gave none.
+    status: str | None = None
 
     @classmethod
     def from_sdk(cls, raw: object) -> HySdkRawRecord:
@@ -417,11 +420,34 @@ class HySdkRawRecord(BaseModel):
                     raw.get("gmt_created") if isinstance(raw.get("gmt_created"), int) else None
                 ),
                 custom=dict(custom),
+                status=str(raw["status"]) if isinstance(raw.get("status"), str) else None,
             )
         except (KeyError, TypeError, ValueError) as error:
             raise HySdkReadError("hy_memory_raw_shape_unknown") from error
 
 
+_RAW_TURN_LINE = re.compile(r"^\[(user|assistant|tool|system)\]:\s?(.*)$")
+
+
+def raw_turn_messages(content: str) -> tuple[tuple[str, str], ...]:
+    """The `(role, text)` messages of one SDK raw turn.
+
+    The SDK stores a turn as `[user]: …\\n[assistant]: …`; a line without a
+    role prefix continues the message above it. Content that names no role
+    at all is one message of unknown role (`""`).
+    """
+    messages: list[list[str]] = []
+    for line in content.splitlines():
+        match = _RAW_TURN_LINE.match(line)
+        if match is not None:
+            messages.append([match.group(1), match.group(2)])
+        elif messages:
+            messages[-1][1] += "\n" + line
+        else:
+            messages.append(["", line])
+    return tuple((role, text) for role, text in messages)
+
+
 class HySdkRawHit(HySdkRawRecord):
     score: float = Field(ge=0, le=1)
 
@@ -516,6 +542,96 @@ class HySdkUserProfile(BaseModel):
         )
 
 
+class HySdkProfileVersion(BaseModel):
+    """One version on an L0 basic-info chain."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    memory_id: str
+    content: str
+    #: Conversation time as the SDK renders it on a chain (epoch or text).
+    memory_at: int | str | None = None
+    gmt_created: int | None = None
+    #: The chain head: the version the profile shows now.
+    current: bool = False
+
+
+class HySdkProfileChain(BaseModel):
+    """One L0 basic-info head and its versions, latest first."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    memory_id: str
+    content: str
+    status: str
+    memory_at: int | None = None
+    gmt_created: int | None = None
+    #: The structured fields this head holds (`basic_info_kv`).
+    basic_info: dict[str, object] = Field(default_factory=dict)
+    versions: tuple[HySdkProfileVersion, ...]
+
+
+class HySdkProfileVersions(BaseModel):
+    """`get_user_basic_info`: the long-term profile's L0 version chains."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    request_id: str
+    chains: tuple[HySdkProfileChain, ...]
+    elapsed_ms: float = Field(ge=0)
+
+    @classmethod
+    def from_sdk(cls, raw: object) -> HySdkProfileVersions:
+        if not isinstance(raw, Mapping):
+            raise HySdkReadError("hy_memory_profile_versions_shape_unknown")
+        request_id = raw.get("request_id")
+        rows = raw.get("memories")
+        elapsed_ms = raw.get("elapsed_ms")
+        if (
+            not isinstance(request_id, str)
+            or not isinstance(rows, list)
+            or not isinstance(elapsed_ms, int | float)
+            or isinstance(elapsed_ms, bool)
+        ):
+            raise HySdkReadError("hy_memory_profile_versions_shape_unknown")
+        chains: list[HySdkProfileChain] = []
+        for row in rows:
+            if not isinstance(row, Mapping) or not isinstance(row.get("memory_id"), str):
+                raise HySdkReadError("hy_memory_profile_versions_shape_unknown")
+            head_id = str(row["memory_id"])
+            chain = row.get("evolution_chain")
+            entries = chain if isinstance(chain, list) and chain else [{**row, "node_id": head_id}]
+            versions = tuple(
+                HySdkProfileVersion(
+                    memory_id=str(entry.get("node_id") or entry.get("memory_id")),
+                    content=str(entry.get("content") or ""),
+                    memory_at=(
+                        entry.get("memory_at")
+                        if isinstance(entry.get("memory_at"), int | str)
+                        and not isinstance(entry.get("memory_at"), bool)
+                        else None
+                    ),
+                    gmt_created=_epoch(entry.get("gmt_created")),
+                    current=str(entry.get("node_id") or entry.get("memory_id")) == head_id,
+                )
+                for entry in entries
+                if isinstance(entry, Mapping) and (entry.get("node_id") or entry.get("memory_id"))
+            )
+            basic_info = row.get("basic_info_kv")
+            chains.append(
+                HySdkProfileChain(
+                    memory_id=head_id,
+                    content=str(row.get("content") or ""),
+                    status=str(row.get("status") or "active"),
+                    memory_at=_epoch(row.get("memory_at")),
+                    gmt_created=_epoch(row.get("gmt_created")),
+                    basic_info=dict(basic_info) if isinstance(basic_info, Mapping) else {},
+                    versions=versions,
+                )
+            )
+        return cls(request_id=request_id, chains=tuple(chains), elapsed_ms=float(elapsed_ms))
+
+
 def parse_sdk_search(raw: object, *, error_code: str) -> tuple[SemanticMemoryHit, ...]:
     rows = _reply_rows(raw, error_code)
     found: list[SemanticMemoryHit] = []
@@ -542,6 +658,120 @@ def parse_sdk_search(raw: object, *, error_code: str) -> tuple[SemanticMemoryHit
     return tuple(found)
 
 
+def parse_sdk_lane_search(raw: object, *, error_code: str) -> tuple[SemanticMemoryHit, ...]:
+    """Every scene's search reply as hits that name their lane and namespace.
+
+    `parse_sdk_search` reads the chat reply only and forgets which lane a hit
+    came from. A caller may now pick the scene and a floor per lane, so this
+    reads all three reply shapes:
+
+    - chat: `memories` as `{schema, proactive, normal}` (or one list, normal);
+    - working/coding: `facts` and `experiences` of that variant;
+    - all: one fused `memories` list whose rows carry `scene`, `kind` and
+      their own lane's score (`lane_score`). The fused `score` is a rank
+      fusion (1/(60+rank)), not a relevance, so the lane score is reported.
+
+    Highest score first, first occurrence of an id kept.
+    """
+    if not isinstance(raw, Mapping):
+        raise HySdkReadError(error_code)
+    scene = raw.get("scene")
+    found: list[SemanticMemoryHit] = []
+    if scene == "all":
+        rows = raw.get("memories")
+        if not isinstance(rows, list) or any(not isinstance(row, Mapping) for row in rows):
+            raise HySdkReadError(error_code)
+        for row in rows:
+            hit = _lane_hit(row, scene=str(row.get("scene") or "chat"), score_key="lane_score")
+            if hit is not None:
+                found.append(hit)
+    elif scene in ("productivity", "working", "coding"):
+        variant = raw.get("variant") if raw.get("variant") in {"working", "coding"} else None
+        lane = variant or (scene if scene in {"working", "coding"} else "working")
+        for key, kind in (("facts", "fact"), ("experiences", "experience")):
+            rows = raw.get(key) or []
+            if not isinstance(rows, list) or any(not isinstance(row, Mapping) for row in rows):
+                raise HySdkReadError(error_code)
+            for row in rows:
+                hit = _lane_hit({**row, "kind": kind}, scene=lane, score_key="score")
+                if hit is not None:
+                    found.append(hit)
+    else:
+        memories = raw.get("memories")
+        if isinstance(memories, Mapping):
+            if set(memories) != {"schema", "proactive", "normal"}:
+                raise HySdkReadError(error_code)
+            channels = tuple(memories.items())
+        elif isinstance(memories, list):
+            channels = (("normal", memories),)
+        else:
+            raise HySdkReadError(error_code)
+        for lane, rows in channels:
+            if not isinstance(rows, list) or any(not isinstance(row, Mapping) for row in rows):
+                raise HySdkReadError(error_code)
+            for row in rows:
+                hit = _lane_hit(row, scene="chat", score_key="score", lane=str(lane))
+                if hit is not None:
+                    found.append(hit)
+    ordered: list[SemanticMemoryHit] = []
+    seen: set[tuple[str, str, str | None]] = set()
+    for hit in sorted(found, key=lambda value: value.score, reverse=True):
+        identity = (hit.memory_id, hit.kind, hit.variant)
+        if identity in seen:
+            continue
+        seen.add(identity)
+        ordered.append(hit)
+    return tuple(ordered)
+
+
+def _lane_hit(
+    row: Mapping[str, object],
+    *,
+    scene: str,
+    score_key: str,
+    lane: str | None = None,
+) -> SemanticMemoryHit | None:
+    kind = str(row.get("kind")) if row.get("kind") in {"experience", "fact"} else "chat"
+    if kind == "chat":
+        memory_id = row.get("memory_id") or row.get("id")
+        variant = None
+        hit_lane = lane or "normal"
+    else:
+        id_key = "experience_id" if kind == "experience" else "fact_id"
+        memory_id = row.get("memory_id") or row.get(id_key)
+        variant = scene if scene in {"working", "coding"} else None
+        if variant is None:
+            return None
+        hit_lane = variant
+    if not isinstance(memory_id, str) or not memory_id:
+        return None
+    value = row.get(score_key)
+    if not isinstance(value, int | float) or isinstance(value, bool):
+        value = row.get("score")
+    score = (
+        max(0.0, min(1.0, float(value)))
+        if isinstance(value, int | float) and not isinstance(value, bool)
+        else 0.0
+    )
+    if kind != "chat":
+        return SemanticMemoryHit(
+            memory_id=memory_id, score=score, lane=hit_lane, kind=kind, variant=variant
+        )
+    content = row.get("content")
+    member_ids, formatting_complete = _formatted_member_ids(row)
+    return SemanticMemoryHit(
+        memory_id=memory_id,
+        score=score,
+        formatted_content=(
+            content.strip() if isinstance(content, str) and content.strip() else None
+        ),
+        member_ids=member_ids,
+        sdk_formatted=True,
+        formatting_complete=formatting_complete,
+        lane=hit_lane,
+    )
+
+
 class LibraryHit(NamedTuple):
     """One keyword match in a library search window, with its namespace."""
 

memhub/apps/core/src/one_memory/library_search_window.py

@@ -28,7 +28,10 @@ from one_memory.hy_sdk_types import LibraryHit
 DEFAULT_WINDOW_TTL_SECONDS = 300.0
 DEFAULT_MAX_WINDOWS = 32
 
-WindowKey = tuple[str, str | None, tuple[str, ...], str, str]
+#: (query, source, statuses, scene, user, other filters as sorted name/values pairs)
+WindowKey = tuple[
+    str, str | None, tuple[str, ...], str, str, tuple[tuple[str, tuple[str, ...]], ...]
+]
 
 
 def window_key(
@@ -37,8 +40,12 @@ def window_key(
     statuses: list[str],
     scene: str = "all",
     user_id: str = "local",
+    filters: dict[str, list[str]] | None = None,
 ) -> WindowKey:
-    return (query, source, tuple(sorted(set(statuses))), scene, user_id)
+    extra = tuple(
+        sorted((name, tuple(sorted(set(values)))) for name, values in (filters or {}).items())
+    )
+    return (query, source, tuple(sorted(set(statuses))), scene, user_id, extra)
 
 
 @dataclass(frozen=True)

memhub/apps/core/src/one_memory/memory_evolution.py

@@ -0,0 +1,618 @@
+"""System 2 for ultra: evolve runs, their receipts, the nightly schedule, graph backfill.
+
+hy-memory 2.0.9's `client.evolve(user_id, agent_id, max_batches)` is ultra
+only, holds no lock, and has no scheduler: MemHub owns all three.
+
+- A run is asynchronous: `start` answers a receipt at once and the run goes
+  on in the background (evolve makes model calls across the user's whole
+  library and can take minutes, longer than any HTTP timeout). One run per
+  user at a time: asking again while one runs returns that run. A run holds
+  the write coordinator's per-user exclusivity, so it never overlaps that
+  user's write batch, and loops while the SDK answers `should_continue`, up
+  to `EVOLVE_MAX_ROUNDS`.
+- The schedule runs every user that has memories once a day at the
+  configured local time, and catches up after the machine slept through it.
+- Switching to ultra backfills each user's existing facts into the graph
+  once (`build_memory_graph`).
+
+What is kept on disk is counts and stable codes only, never memory text:
+`<data_dir>/evolve-state.json`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import re
+import threading
+from collections.abc import Awaitable, Callable, Iterable, Mapping
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from typing import Any, Literal, Protocol
+from uuid import uuid4
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from one_memory.diagnostic_log import diagnostic_event
+from one_memory.memory_server.config_state import EvolveSchedule
+from one_memory.memory_server.durable_queue import safe_error_code
+from one_memory.sdk_write_processor import PRODUCT_AGENT_ID
+
+logger = logging.getLogger(__name__)
+
+#: Rounds one run makes while the SDK says there is more to do.
+EVOLVE_MAX_ROUNDS = 20
+#: How often the schedule looks at the clock; also bounds catch-up after wake.
+SCHEDULE_CHECK_SECONDS = 300.0
+#: Receipts kept in memory for `GET /api/v2/evolve/{run_id}`.
+_KEPT_RUNS = 64
+_STABLE_CODE = re.compile(r"[a-z][a-z0-9_]{0,95}")
+_COUNTS = (
+    "remaining_clusters",
+    "remaining_actionable_clusters",
+    "remaining_fresh_facts",
+)
+
+RunState = Literal["running", "succeeded", "failed"]
+Trigger = Literal["manual", "scheduled"]
+
+
+class EvolveModeError(RuntimeError):
+    """Evolve asked for outside ultra."""
+
+    def __init__(self) -> None:
+        super().__init__("memory_mode_not_ultra")
+
+
+class EvolveRunReceipt(BaseModel):
+    """One evolve run. Counts and codes; `results` only while in memory."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    run_id: str
+    user_id: str
+    agent_id: str
+    max_batches: int | None = None
+    trigger: Trigger = "manual"
+    state: RunState = "running"
+    started_at: datetime
+    finished_at: datetime | None = None
+    rounds: int = 0
+    tasks_processed: int = 0
+    batch_failures: int = 0
+    cycle_status: str | None = None
+    should_continue: bool | None = None
+    continuation_reason: str | None = None
+    remaining_clusters: int | None = None
+    remaining_actionable_clusters: int | None = None
+    remaining_fresh_facts: int | None = None
+    #: True when the round cap stopped a run the SDK would have continued.
+    capped: bool = False
+    elapsed_ms: float = 0.0
+    error_code: str | None = None
+    failure_kind: str | None = None
+    #: The last round's SDK `request_id`, for matching SDK logs.
+    request_id: str | None = None
+    #: The last round's SDK `results` and `cross_domain_sweeper`. Kept only
+    #: in this receipt, never written to disk or diagnostics.
+    results: Any = None
+    cross_domain_sweeper: Any = None
+
+    def summary(self) -> dict[str, object]:
+        """What is persisted and shown in status: no SDK result bodies."""
+        return self.model_dump(mode="json", exclude={"results", "cross_domain_sweeper"})
+
+
+class UserEvolveState(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    last_run_at: datetime | None = None
+    last_result: dict[str, object] | None = None
+    last_error: str | None = None
+
+
+class GraphBuildUser(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    state: Literal["pending", "running", "succeeded", "failed"] = "pending"
+    memories_scanned: int | None = None
+    nodes: int | None = None
+    error_code: str | None = None
+
+
+class GraphBuildState(BaseModel):
+    """The one-time backfill of existing facts into the graph after → ultra."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    state: Literal["idle", "running", "succeeded", "failed"] = "idle"
+    started_at: datetime | None = None
+    finished_at: datetime | None = None
+    users: dict[str, GraphBuildUser] = Field(default_factory=dict)
+
+
+class EvolveStateDocument(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: int = Field(default=1, ge=1, le=1)
+    users: dict[str, UserEvolveState] = Field(default_factory=dict)
+    #: When the schedule first ran in ultra: the "last run" of a user that
+    #: never ran, so enabling ultra at noon does not evolve everyone at once.
+    schedule_anchor: datetime | None = None
+    graph_build: GraphBuildState = Field(default_factory=GraphBuildState)
+
+
+class EvolveStateStore:
+    """The only writer of `<data_dir>/evolve-state.json`."""
+
+    def __init__(self, data_dir: Path) -> None:
+        self.path = data_dir / "evolve-state.json"
+        self._lock = threading.RLock()
+        self._document = self._read()
+
+    def document(self) -> EvolveStateDocument:
+        with self._lock:
+            return self._document.model_copy(deep=True)
+
+    def user(self, user_id: str) -> UserEvolveState:
+        with self._lock:
+            return self._document.users.get(user_id, UserEvolveState()).model_copy()
+
+    def record_run(self, receipt: EvolveRunReceipt) -> None:
+        with self._lock:
+            users = dict(self._document.users)
+            users[receipt.user_id] = UserEvolveState(
+                last_run_at=receipt.started_at,
+                last_result=receipt.summary(),
+                last_error=receipt.error_code,
+            )
+            self._save(self._document.model_copy(update={"users": users}))
+
+    def anchor(self, now: datetime) -> datetime:
+        """The schedule anchor, set to `now` the first time it is asked."""
+        with self._lock:
+            if self._document.schedule_anchor is None:
+                self._save(self._document.model_copy(update={"schedule_anchor": now}))
+            assert self._document.schedule_anchor is not None
+            return self._document.schedule_anchor
+
+    def graph_build(self) -> GraphBuildState:
+        with self._lock:
+            return self._document.graph_build.model_copy(deep=True)
+
+    def set_graph_build(self, state: GraphBuildState) -> None:
+        with self._lock:
+            self._save(self._document.model_copy(update={"graph_build": state}))
+
+    def _save(self, document: EvolveStateDocument) -> None:
+        payload = json.dumps(document.model_dump(mode="json"), indent=2, sort_keys=True) + "\n"
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        temporary: Path | None = None
+        try:
+            with NamedTemporaryFile(
+                mode="w",
+                encoding="utf-8",
+                dir=self.path.parent,
+                prefix=".evolve-state.",
+                suffix=".tmp",
+                delete=False,
+            ) as stream:
+                temporary = Path(stream.name)
+                stream.write(payload)
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.chmod(temporary, 0o600)
+            os.replace(temporary, self.path)
+        except OSError:
+            if temporary is not None:
+                temporary.unlink(missing_ok=True)
+            # Status is best-effort: a run must not fail because its record did.
+            logger.warning("evolve_state_write_failed")
+        self._document = document
+
+    def _read(self) -> EvolveStateDocument:
+        try:
+            return EvolveStateDocument.model_validate_json(self.path.read_text(encoding="utf-8"))
+        except FileNotFoundError:
+            return EvolveStateDocument()
+        except (OSError, ValueError):
+            # A damaged status file only loses history; start a fresh one.
+            logger.warning("evolve_state_unreadable")
+            return EvolveStateDocument()
+
+
+class EvolveCoordinator(Protocol):
+    """The write coordinator surface evolve and the graph backfill use."""
+
+    def exclusive_for_evolve(self, user_id: str) -> Any: ...
+
+    def use_client(self) -> Any: ...
+
+
+class EvolveService:
+    """Start, serialize and record evolve runs."""
+
+    def __init__(
+        self,
+        *,
+        coordinator: EvolveCoordinator,
+        state: EvolveStateStore,
+        memory_mode: Callable[[], str],
+        max_rounds: int = EVOLVE_MAX_ROUNDS,
+        now: Callable[[], datetime] = lambda: datetime.now(UTC),
+    ) -> None:
+        self.coordinator = coordinator
+        self.state = state
+        self.memory_mode = memory_mode
+        self.max_rounds = max_rounds
+        self._now = now
+        self._runs: dict[str, EvolveRunReceipt] = {}
+        self._tasks: dict[str, asyncio.Task[None]] = {}
+        self._active: dict[str, str] = {}
+
+    def start(
+        self,
+        user_id: str,
+        *,
+        agent_id: str | None = None,
+        max_batches: int | None = None,
+        trigger: Trigger = "manual",
+    ) -> EvolveRunReceipt:
+        """Start a run, or return the one already running for this user."""
+        if self.memory_mode() != "ultra":
+            raise EvolveModeError()
+        running = self._active.get(user_id)
+        if running is not None:
+            return self._runs[running]
+        receipt = EvolveRunReceipt(
+            run_id=str(uuid4()),
+            user_id=user_id,
+            agent_id=agent_id or PRODUCT_AGENT_ID,
+            max_batches=max_batches,
+            trigger=trigger,
+            started_at=self._now(),
+        )
+        self._remember(receipt)
+        self._active[user_id] = receipt.run_id
+        self._tasks[receipt.run_id] = asyncio.create_task(
+            self._run(receipt.run_id), name="one-memory-evolve"
+        )
+        return receipt
+
+    def get(self, run_id: str) -> EvolveRunReceipt | None:
+        return self._runs.get(run_id)
+
+    def running(self) -> tuple[EvolveRunReceipt, ...]:
+        return tuple(self._runs[run_id] for run_id in self._active.values())
+
+    async def wait(self, run_id: str) -> EvolveRunReceipt | None:
+        task = self._tasks.get(run_id)
+        if task is not None:
+            await asyncio.shield(task)
+        return self._runs.get(run_id)
+
+    async def shutdown(self) -> None:
+        tasks = tuple(self._tasks.values())
+        for task in tasks:
+            task.cancel()
+        await asyncio.gather(*tasks, return_exceptions=True)
+        self._tasks.clear()
+
+    def _remember(self, receipt: EvolveRunReceipt) -> None:
+        self._runs[receipt.run_id] = receipt
+        while len(self._runs) > _KEPT_RUNS:
+            oldest = next(iter(self._runs))
+            if oldest in self._active.values():
+                break
+            del self._runs[oldest]
+
+    async def _run(self, run_id: str) -> None:
+        receipt = self._runs[run_id]
+        diagnostic_event(
+            level="info",
+            component="memory.evolve",
+            event="memory_evolve.started",
+            outcome="started",
+            details={"trigger": receipt.trigger, "max_batches": receipt.max_batches},
+        )
+        try:
+            async with (
+                self.coordinator.exclusive_for_evolve(receipt.user_id),
+                self.coordinator.use_client() as client,
+            ):
+                while receipt.rounds < self.max_rounds:
+                    result = await asyncio.to_thread(
+                        client.evolve,
+                        receipt.user_id,
+                        agent_id=receipt.agent_id,
+                        max_batches=receipt.max_batches,
+                    )
+                    receipt = _folded(receipt, result)
+                    self._runs[run_id] = receipt
+                    if receipt.state == "failed" or not receipt.should_continue:
+                        break
+                else:
+                    receipt = receipt.model_copy(update={"capped": True})
+            if receipt.state == "running":
+                receipt = receipt.model_copy(update={"state": "succeeded"})
+        except asyncio.CancelledError:
+            receipt = receipt.model_copy(
+                update={"state": "failed", "error_code": "memory_evolve_cancelled"}
+            )
+            raise
+        except Exception as error:  # noqa: BLE001 - recorded as a stable code
+            code = safe_error_code(error)
+            receipt = receipt.model_copy(
+                update={
+                    "state": "failed",
+                    "error_code": _stable(code, "hy_memory_evolve_failed"),
+                    "failure_kind": type(error).__name__,
+                }
+            )
+        finally:
+            receipt = receipt.model_copy(update={"finished_at": self._now()})
+            self._runs[run_id] = receipt
+            self._active.pop(receipt.user_id, None)
+            self._tasks.pop(run_id, None)
+            await asyncio.to_thread(self.state.record_run, receipt)
+            diagnostic_event(
+                level="info" if receipt.state == "succeeded" else "warning",
+                component="memory.evolve",
+                event="memory_evolve.finished",
+                outcome="succeeded" if receipt.state == "succeeded" else "failed",
+                details={
+                    "trigger": receipt.trigger,
+                    "rounds": receipt.rounds,
+                    "tasks_processed": receipt.tasks_processed,
+                    "batch_failures": receipt.batch_failures,
+                    "capped": receipt.capped,
+                    "elapsed_ms": round(receipt.elapsed_ms, 1),
+                    "error_code": receipt.error_code,
+                },
+            )
+
+
+def _folded(receipt: EvolveRunReceipt, result: object) -> EvolveRunReceipt:
+    """One more evolve round's result added to the run."""
+    if not isinstance(result, Mapping):
+        return receipt.model_copy(
+            update={"state": "failed", "error_code": "hy_memory_evolve_shape_unknown"}
+        )
+    failures = result.get("batch_failures")
+    update: dict[str, object] = {
+        "rounds": receipt.rounds + 1,
+        "tasks_processed": receipt.tasks_processed + _int(result.get("tasks_processed")),
+        "batch_failures": receipt.batch_failures
+        + (len(failures) if isinstance(failures, list | tuple) else _int(failures)),
+        "elapsed_ms": receipt.elapsed_ms + _float(result.get("elapsed_ms")),
+        "should_continue": result.get("should_continue") is True,
+        "cycle_status": _text(result.get("cycle_status")),
+        "continuation_reason": _text(result.get("continuation_reason")),
+        "request_id": _text(result.get("request_id")),
+        "results": result.get("results"),
+        "cross_domain_sweeper": result.get("cross_domain_sweeper"),
+    }
+    for name in _COUNTS:
+        value = result.get(name)
+        update[name] = value if isinstance(value, int) and not isinstance(value, bool) else None
+    if result.get("success") is not True:
+        code = result.get("error_code")
+        kind = result.get("failure_kind")
+        update.update(
+            state="failed",
+            should_continue=False,
+            error_code=(
+                code
+                if isinstance(code, str) and _STABLE_CODE.fullmatch(code)
+                else "hy_memory_evolve_rejected"
+            ),
+            failure_kind=kind if isinstance(kind, str) and _STABLE_CODE.fullmatch(kind) else None,
+        )
+    return receipt.model_copy(update=update)
+
+
+def _stable(code: str, fallback: str) -> str:
+    return code if _STABLE_CODE.fullmatch(code) else fallback
+
+
+def _int(value: object) -> int:
+    return value if isinstance(value, int) and not isinstance(value, bool) else 0
+
+
+def _float(value: object) -> float:
+    if isinstance(value, int | float) and not isinstance(value, bool):
+        return float(value)
+    return 0.0
+
+
+def _text(value: object) -> str | None:
+    return value[:128] if isinstance(value, str) else None
+
+
+def evolve_due(
+    now: datetime,
+    schedule: EvolveSchedule,
+    last_run_at: datetime | None,
+    anchor: datetime,
+) -> bool:
+    """Whether a user's nightly evolve is due at `now` (aware, local time).
+
+    Due once the day's slot (`schedule.time`, local) has passed and the last
+    run (or, for a user who never ran, the anchor) is older than that slot.
+    A last run more than 24 hours old is due at once as well: a machine asleep
+    at the slot catches up when it wakes, even before the next day's slot.
+    """
+    if not schedule.enabled:
+        return False
+    local = now.astimezone()
+    hour, minute = (int(part) for part in schedule.time.split(":"))
+    slot = local.replace(hour=hour, minute=minute, second=0, microsecond=0)
+    reference = last_run_at or anchor
+    if local >= slot and reference < slot:
+        return True
+    return last_run_at is not None and local - last_run_at > timedelta(hours=24)
+
+
+class EvolveScheduler:
+    """Run evolve nightly for every user that has memories (ultra only)."""
+
+    def __init__(
+        self,
+        *,
+        service: EvolveService,
+        schedule: Callable[[], EvolveSchedule],
+        users_with_memories: Callable[[], Awaitable[Iterable[str]]],
+        now: Callable[[], datetime] = lambda: datetime.now().astimezone(),
+        check_seconds: float = SCHEDULE_CHECK_SECONDS,
+    ) -> None:
+        self.service = service
+        self.schedule = schedule
+        self.users_with_memories = users_with_memories
+        self._now = now
+        self.check_seconds = check_seconds
+        self._task: asyncio.Task[None] | None = None
+
+    def start(self) -> None:
+        if self._task is None or self._task.done():
+            self._task = asyncio.create_task(self._loop(), name="one-memory-evolve-schedule")
+
+    async def tick(self) -> tuple[str, ...]:
+        """Run every due user once, one after another; the users run."""
+        schedule = self.schedule()
+        if not schedule.enabled or self.service.memory_mode() != "ultra":
+            return ()
+        now = self._now()
+        anchor = await asyncio.to_thread(self.service.state.anchor, now)
+        ran: list[str] = []
+        for user_id in await self.users_with_memories():
+            last = self.service.state.user(user_id).last_run_at
+            if not evolve_due(now, schedule, last, anchor):
+                continue
+            try:
+                receipt = self.service.start(user_id, trigger="scheduled")
+            except EvolveModeError:
+                break
+            await self.service.wait(receipt.run_id)
+            ran.append(user_id)
+        if ran:
+            diagnostic_event(
+                level="info",
+                component="memory.evolve",
+                event="memory_evolve.schedule_ran",
+                outcome="succeeded",
+                details={"users": len(ran)},
+            )
+        return tuple(ran)
+
+    async def _loop(self) -> None:
+        while True:
+            try:
+                await self.tick()
+            except asyncio.CancelledError:
+                raise
+            except Exception as error:  # noqa: BLE001 - the loop must survive
+                logger.warning("evolve_schedule_tick_failed error_type=%s", type(error).__name__)
+            await asyncio.sleep(self.check_seconds)
+
+    async def shutdown(self) -> None:
+        if self._task is None:
+            return
+        self._task.cancel()
+        await asyncio.gather(self._task, return_exceptions=True)
+        self._task = None
+
+
+class MemoryGraphBackfill:
+    """After a switch to ultra, project each user's existing facts into the graph once."""
+
+    def __init__(
+        self,
+        *,
+        coordinator: EvolveCoordinator,
+        state: EvolveStateStore,
+        now: Callable[[], datetime] = lambda: datetime.now(UTC),
+    ) -> None:
+        self.coordinator = coordinator
+        self.state = state
+        self._now = now
+        self._task: asyncio.Task[None] | None = None
+
+    @property
+    def running(self) -> bool:
+        return self._task is not None and not self._task.done()
+
+    def start(self, user_ids: Iterable[str]) -> GraphBuildState:
+        if self.running:
+            return self.state.graph_build()
+        users = tuple(dict.fromkeys(user_ids))
+        state = GraphBuildState(
+            state="running",
+            started_at=self._now(),
+            users={user_id: GraphBuildUser() for user_id in users},
+        )
+        self.state.set_graph_build(state)
+        self._task = asyncio.create_task(self._run(users), name="one-memory-graph-backfill")
+        return state
+
+    async def wait(self) -> None:
+        if self._task is not None:
+            await asyncio.shield(self._task)
+
+    async def _run(self, users: tuple[str, ...]) -> None:
+        state = self.state.graph_build()
+        for user_id in users:
+            state.users[user_id] = GraphBuildUser(state="running")
+            await asyncio.to_thread(self.state.set_graph_build, state.model_copy(deep=True))
+            try:
+                async with (
+                    self.coordinator.exclusive_for_evolve(user_id),
+                    self.coordinator.use_client() as client,
+                ):
+                    result = await asyncio.to_thread(
+                        client.build_memory_graph,
+                        user_id,
+                        agent_id=None,
+                        limit=10000,
+                        clear_first=False,
+                    )
+                ok = isinstance(result, Mapping) and result.get("success") is True
+                state.users[user_id] = GraphBuildUser(
+                    state="succeeded" if ok else "failed",
+                    memories_scanned=_int(result.get("memories_scanned"))
+                    if isinstance(result, Mapping)
+                    else None,
+                    nodes=_int(result.get("nodes")) if isinstance(result, Mapping) else None,
+                    error_code=None if ok else "hy_memory_graph_build_rejected",
+                )
+            except asyncio.CancelledError:
+                raise
+            except Exception as error:  # noqa: BLE001 - recorded as a stable code
+                code = safe_error_code(error)
+                state.users[user_id] = GraphBuildUser(
+                    state="failed",
+                    error_code=_stable(code, "hy_memory_graph_build_failed"),
+                )
+        failed = any(user.state == "failed" for user in state.users.values())
+        state.state = "failed" if failed else "succeeded"
+        state.finished_at = self._now()
+        await asyncio.to_thread(self.state.set_graph_build, state)
+        diagnostic_event(
+            level="warning" if failed else "info",
+            component="memory.evolve",
+            event="memory_graph_backfill.finished",
+            outcome="failed" if failed else "succeeded",
+            details={
+                "users": len(state.users),
+                "failed_users": sum(user.state == "failed" for user in state.users.values()),
+            },
+        )
+
+    async def shutdown(self) -> None:
+        if self._task is None:
+            return
+        self._task.cancel()
+        await asyncio.gather(self._task, return_exceptions=True)
+        self._task = None

memhub/apps/core/src/one_memory/memory_server/config_state.py

@@ -13,6 +13,7 @@ import threading
 from collections.abc import Mapping
 from pathlib import Path
 from tempfile import NamedTemporaryFile
+from typing import Literal
 
 from pydantic import BaseModel, ConfigDict, Field, StrictBool, field_validator
 
@@ -27,6 +28,31 @@ class SourceConfig(BaseModel):
     sync_enabled: StrictBool = False
 
 
+#: The SDK pipeline mode (hy-memory `HyMemoryClient(mode=...)`):
+#: lite keeps raw conversations only, with no model call; pro extracts
+#: memories (System 1); ultra adds the knowledge graph and evolve (System 2).
+MemoryMode = Literal["lite", "pro", "ultra"]
+MEMORY_MODES: tuple[MemoryMode, ...] = ("lite", "pro", "ultra")
+_SCHEDULE_TIME = re.compile(r"^(?:[01][0-9]|2[0-3]):[0-5][0-9]$")
+
+
+class EvolveSchedule(BaseModel):
+    """When ultra runs evolve for every user: once a day, local time."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
+
+    enabled: StrictBool = True
+    #: `HH:MM`, 24-hour, the machine's local time.
+    time: str = "03:00"
+
+    @field_validator("time")
+    @classmethod
+    def valid_time(cls, value: str) -> str:
+        if _SCHEDULE_TIME.fullmatch(value) is None:
+            raise ValueError("invalid_evolve_schedule_time")
+        return value
+
+
 class ConfigDocument(BaseModel):
     model_config = ConfigDict(frozen=True, extra="forbid", strict=True)
 
@@ -39,6 +65,10 @@ class ConfigDocument(BaseModel):
     #: needs no re-entry of a key.
     sedimentation_enabled: StrictBool = True
     extraction_window_preset: ExtractionWindowPreset = ExtractionWindowPreset.THREE_MONTHS
+    #: The SDK mode every client is built with. `ONE_MEMORY_GRAPH_STORE` (dev)
+    #: still forces ultra over it; see `hy_graph_admin.effective_memory_mode`.
+    memory_mode: MemoryMode = "pro"
+    evolve_schedule: EvolveSchedule = Field(default_factory=EvolveSchedule)
 
     @field_validator("sources")
     @classmethod
@@ -61,6 +91,9 @@ class ConfigFileError(RuntimeError):
     pass
 
 
+_WRITTEN_WHEN_SET = frozenset({"memory_mode", "evolve_schedule"})
+
+
 class ConfigStore:
     """The only writer of ``<data_dir>/config.json`` inside Core."""
 
@@ -107,6 +140,32 @@ class ConfigStore:
             self._document = updated
             return updated
 
+    def memory_mode(self) -> MemoryMode:
+        with self._lock:
+            return self._document.memory_mode
+
+    def set_memory_mode(self, mode: MemoryMode) -> ConfigDocument:
+        if mode not in MEMORY_MODES:
+            raise ValueError("invalid_memory_mode")
+        with self._lock:
+            updated = self._document.model_copy(update={"memory_mode": mode})
+            self._write(updated)
+            self._document = updated
+            return updated
+
+    def evolve_schedule(self) -> EvolveSchedule:
+        with self._lock:
+            return self._document.evolve_schedule
+
+    def set_evolve_schedule(self, schedule: EvolveSchedule) -> ConfigDocument:
+        if not isinstance(schedule, EvolveSchedule):
+            raise ValueError("invalid_evolve_schedule")
+        with self._lock:
+            updated = self._document.model_copy(update={"evolve_schedule": schedule})
+            self._write(updated)
+            self._document = updated
+            return updated
+
     def migrate_extraction_window_preset(self, preset: ExtractionWindowPreset) -> ConfigDocument:
         """Adopt the legacy SQL value only when config.json has no value."""
         if not isinstance(preset, ExtractionWindowPreset):
@@ -152,9 +211,12 @@ class ConfigStore:
             raise ConfigFileError("config_invalid") from error
 
     def _write(self, document: ConfigDocument) -> None:
+        # Settings added after 1.0.5 are written only once a person chose
+        # them, so an untouched install's file stays what older builds read.
+        unset = {name for name in _WRITTEN_WHEN_SET if name not in document.model_fields_set}
         payload = (
             json.dumps(
-                document.model_dump(mode="json"),
+                document.model_dump(mode="json", exclude=unset),
                 ensure_ascii=False,
                 indent=2,
                 sort_keys=True,

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -2563,6 +2563,8 @@ async def execute_job(
             in {
                 "model_content_source_disabled",
                 "model_sedimentation_disabled",
+                # Evolve holds this user; the job runs once it is done.
+                "memory_evolve_in_progress",
             }
             and not hy_invoked
         ):

memhub/apps/core/src/one_memory/memory_visibility.py

@@ -160,7 +160,7 @@ def pending(db: sqlite3.Connection, *, limit: int = 100) -> list[str]:
 
 
 def _trigger_sql() -> list[tuple[str, str]]:
-    """(table it depends on, statement) for each trigger."""
+    """(table it is on, statement) for each trigger."""
     return [
         (
             "evidence",
@@ -303,9 +303,61 @@ def install_triggers(db: sqlite3.Connection) -> None:
     for table, statement in _trigger_sql():
         if table not in present:
             continue
+        if not _body_tables(statement) <= present:
+            # SQLite resolves a trigger body only when it fires, so a body
+            # naming a dropped table fails the write it rides on (#233: the
+            # SDK cutover drops `governed_memory_evidence`, and every grant
+            # write then failed). Not installed, and an old copy is removed.
+            cursor.execute(f"DROP TRIGGER IF EXISTS {_trigger_name(statement)}")
+            continue
         cursor.execute(statement)
 
 
+def _trigger_name(statement: str) -> str:
+    return statement.split("IF NOT EXISTS", 1)[1].split()[0]
+
+
+def _body_tables(statement: str) -> set[str]:
+    """Tables the trigger body reads or writes besides the one it is on."""
+    _head, _begin, body = statement.partition("BEGIN")
+    words = body.replace("(", " ").replace(")", " ").split()
+    return {
+        name
+        for keyword, name in zip(words, words[1:], strict=False)
+        if keyword.upper() in {"FROM", "JOIN", "INTO"}
+    }
+
+
+def drop_stale_triggers(database_path: Path) -> tuple[str, ...]:
+    """Drop this module's triggers whose body names a table that is gone.
+
+    Startup on SDK authority does not install the cache (it is read only on
+    legacy authority), but a home migrated by an earlier build still carries
+    the triggers it installed. One whose body reads a dropped table fails
+    every write to the table it is on, so it is dropped here. Idempotent.
+    """
+    with main_database(database_path).write() as db, db:
+        cursor = db.cursor()
+        cursor.row_factory = sqlite3.Row
+        present = {
+            str(row["name"])
+            for row in cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
+        }
+        installed = {
+            str(row["name"])
+            for row in cursor.execute("SELECT name FROM sqlite_master WHERE type='trigger'")
+        }
+        dropped = tuple(
+            name
+            for _table, statement in _trigger_sql()
+            if (name := _trigger_name(statement)) in installed
+            and not _body_tables(statement) <= present
+        )
+        for name in dropped:
+            cursor.execute(f"DROP TRIGGER IF EXISTS {name}")
+    return dropped
+
+
 def startup(database_path: Path) -> None:
     """Idempotent: create the derived tables and triggers. No version change.
 

memhub/apps/core/src/one_memory/model_probe.py

@@ -564,7 +564,7 @@ def _embedding_request(
     secret: str | None,
 ) -> tuple[str, str, dict[str, str], dict[str, object]]:
     provider = profile.embedding_provider_kind or profile.provider_kind
-    base_url = (profile.embedding_base_url or profile.base_url or "").rstrip("/")
+    base_url = (profile.embedding_base_url or "").rstrip("/")
     if provider is ProviderKind.OLLAMA:
         return (
             "POST",

memhub/apps/core/src/one_memory/model_profiles.py

@@ -97,7 +97,6 @@ class ModelProfileInput(BaseModel):
         )
         embedding_provider, resolved_embedding_base_url = resolve_embedding_endpoint(
             provider_kind=self.provider_kind,
-            base_url=base_url,
             embedding_model=embedding_model,
             embedding_provider_kind=embedding_provider_kind,
             embedding_base_url=embedding_base_url,
@@ -107,6 +106,12 @@ class ModelProfileInput(BaseModel):
             ProviderKind.ANTHROPIC,
         }:
             raise ValueError("embedding provider must be OpenAI-compatible, Ollama, or local ONNX")
+        if (
+            embedding_provider is not ProviderKind.LOCAL_ONNX
+            and resolved_embedding_base_url is None
+        ):
+            # `ModelProfileStore.create` refuses it with its own code.
+            return self
         if (
             embedding_provider is not ProviderKind.LOCAL_ONNX
             and resolved_embedding_base_url is not None
@@ -308,11 +313,12 @@ class ModelProfileStore:
                     f"""
                     UPDATE {table}
                     SET embedding_provider_kind = provider_kind,
-                        embedding_base_url = base_url
+                        embedding_base_url = COALESCE(embedding_base_url, base_url)
                     WHERE embedding_model IS NOT NULL
                       AND embedding_provider_kind IS NULL
                     """
                 )
+                _pin_embedding_endpoints(connection, table)
             connection.execute(
                 """
                 INSERT OR IGNORE INTO model_profile_versions (
@@ -355,11 +361,11 @@ class ModelProfileStore:
         )
         embedding_provider_kind, embedding_base_url = resolve_embedding_endpoint(
             provider_kind=value.provider_kind,
-            base_url=base_url,
             embedding_model=defaulted_model,
             embedding_provider_kind=defaulted_provider,
             embedding_base_url=defaulted_base_url,
         )
+        require_embedding_base_url(embedding_provider_kind, embedding_base_url)
         try:
             with self._connect(writing=True) as connection:
                 connection.execute(
@@ -612,7 +618,6 @@ class ModelProfileStore:
         )
         embedding_provider_kind, embedding_base_url = resolve_embedding_endpoint(
             provider_kind=current.provider_kind,
-            base_url=base_url,
             embedding_model=embedding_model,
             embedding_provider_kind=(
                 patch.embedding_provider_kind
@@ -627,6 +632,7 @@ class ModelProfileStore:
         )
         if embedding_provider_kind in {ProviderKind.HY_DEFAULT, ProviderKind.ANTHROPIC}:
             raise ValueError("embedding provider must be OpenAI-compatible, Ollama, or local ONNX")
+        require_embedding_base_url(embedding_provider_kind, embedding_base_url)
         if (
             embedding_provider_kind is not ProviderKind.LOCAL_ONNX
             and embedding_base_url is not None
@@ -1167,21 +1173,77 @@ def attach_disable_thinking(
 def resolve_embedding_endpoint(
     *,
     provider_kind: ProviderKind,
-    base_url: str | None,
     embedding_model: str | None,
     embedding_provider_kind: ProviderKind | None,
     embedding_base_url: str | None,
 ) -> tuple[ProviderKind | None, str | None]:
+    """The embedding provider and the endpoint exactly as given.
+
+    The endpoint never follows the generation `base_url`: the collection name
+    hashes it, so a proxy or region edit of the generation endpoint moved
+    writes to a new, empty collection, and providers without embeddings were
+    sent embedding requests. `require_embedding_base_url` refuses a remote
+    embedding without one.
+    """
     if embedding_model is None:
         return None, None
     if embedding_provider_kind is ProviderKind.LOCAL_ONNX:
         return ProviderKind.LOCAL_ONNX, None
-    if embedding_provider_kind is None:
-        return provider_kind, embedding_base_url or base_url
-    return (
-        embedding_provider_kind,
-        embedding_base_url or DEFAULT_BASE_URLS.get(embedding_provider_kind),
-    )
+    return embedding_provider_kind or provider_kind, embedding_base_url
+
+
+class EmbeddingBaseUrlRequiredError(ValueError):
+    def __init__(self) -> None:
+        super().__init__("embedding_base_url_required")
+
+
+def require_embedding_base_url(provider: ProviderKind | None, base_url: str | None) -> None:
+    """A remote embedding must name its endpoint; the UI may offer the generation one."""
+    if provider not in {None, ProviderKind.LOCAL_ONNX} and not base_url:
+        raise EmbeddingBaseUrlRequiredError()
+
+
+def pinned_embedding_base_url(
+    *,
+    provider_kind: ProviderKind | str,
+    base_url: str | None,
+    embedding_model: str | None,
+    embedding_provider_kind: ProviderKind | str | None,
+    embedding_base_url: str | None,
+) -> str | None:
+    """The endpoint a stored profile's embedding resolved to before it was pinned.
+
+    The Runtime used `embedding_base_url or base_url`, and the provider default
+    only when both were empty (no collection could have existed then). Writing
+    that value keeps the endpoint, and so the collection name, byte-identical.
+    """
+    provider = ProviderKind(embedding_provider_kind or provider_kind)
+    if embedding_model is None or embedding_base_url or provider is ProviderKind.LOCAL_ONNX:
+        return embedding_base_url
+    return base_url or DEFAULT_BASE_URLS.get(provider)
+
+
+def _pin_embedding_endpoints(connection: sqlite3.Connection, table: str) -> None:
+    """Write the resolved endpoint into every remote embedding left without one."""
+    rows = connection.execute(
+        f"SELECT rowid, provider_kind, base_url, embedding_model, embedding_provider_kind "
+        f"FROM {table} WHERE embedding_model IS NOT NULL AND embedding_base_url IS NULL "
+        "AND COALESCE(embedding_provider_kind, provider_kind) != 'local_onnx'"
+    ).fetchall()
+    for row in rows:
+        connection.execute(
+            f"UPDATE {table} SET embedding_base_url = ? WHERE rowid = ?",
+            (
+                pinned_embedding_base_url(
+                    provider_kind=row["provider_kind"],
+                    base_url=row["base_url"],
+                    embedding_model=row["embedding_model"],
+                    embedding_provider_kind=row["embedding_provider_kind"],
+                    embedding_base_url=None,
+                ),
+                row["rowid"],
+            ),
+        )
 
 
 def validate_model_base_url(value: str) -> None:

memhub/apps/core/src/one_memory/recall.py

@@ -14,6 +14,7 @@ from pydantic import BaseModel, ConfigDict, Field
 
 from one_memory.hy_sdk_types import HySdkMemoryRecord
 from one_memory.memory_governance import GovernedMemory, MemoryStatus
+from one_memory.sdk_search_options import SdkSearchOptions
 from one_memory.semantic_deadline import (
     SemanticMemoryHit,
     SemanticSearch,
@@ -50,7 +51,15 @@ class TruncationBudget(StrEnum):
     TIMEOUT = "timeout_budget"
 
 
-class RecallRequest(BaseModel):
+class RecallRequest(SdkSearchOptions):
+    """`POST /api/v1/recall`.
+
+    The `SdkSearchOptions` fields, `schema_limit` and `intention_limit` pass
+    through to the SDK memory search (and the floor and row filters to the
+    raw fallback). Absent, recall makes the search it always made: the
+    relevance floor, schema and intention lanes sized to `limit`.
+    """
+
     model_config = ConfigDict(frozen=True, extra="forbid")
 
     query: str = Field(min_length=1, max_length=2_000)
@@ -70,6 +79,17 @@ class RecallRequest(BaseModel):
     timeout_ms: int = Field(default=2_000, ge=50, le=30_000)
     #: Whose memories to recall. Omitted, it is `local`, the App's library.
     user_id: UserId = DEFAULT_USER_ID
+    #: The SDK's schema (L6) and proactive intention (L7) lanes.
+    schema_limit: int | None = Field(default=None, ge=0, le=20)
+    intention_limit: int | None = Field(default=None, ge=0, le=20)
+
+    def search_overrides(self) -> Mapping[str, object]:
+        values = dict(super().search_overrides())
+        if self.schema_limit is not None:
+            values["schema_limit"] = self.schema_limit
+        if self.intention_limit is not None:
+            values["intention_limit"] = self.intention_limit
+        return values
 
 
 class RecallHit(BaseModel):

memhub/apps/core/src/one_memory/sdk_memory_qa.py

@@ -6,7 +6,7 @@ import asyncio
 import json
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import aclosing
-from typing import Any, Literal, Protocol
+from typing import Any, Literal, Protocol, cast
 
 import httpx
 from pydantic import BaseModel, ConfigDict
@@ -14,7 +14,7 @@ from pydantic import BaseModel, ConfigDict
 from one_memory.http_client import create_async_client
 from one_memory.hy_runtime_config import HY_MEMORY_READER
 from one_memory.hy_sdk_runtime import sdk_relevance_floor
-from one_memory.hy_sdk_types import HySdkMemoryRecord, parse_sdk_search
+from one_memory.hy_sdk_types import HySdkMemoryRecord, parse_sdk_lane_search, parse_sdk_search
 from one_memory.memory_qa import (
     MemoryAnswerError,
     _DeferredSecretStore,
@@ -31,6 +31,7 @@ from one_memory.model_profiles import (
     ProviderKind,
     attach_disable_thinking,
 )
+from one_memory.sdk_search_options import SdkSearchFilters, SearchLane
 from one_memory.secrets import SecretStore, SecretStoreUnavailableError
 from one_memory.user_scope import owns
 
@@ -96,22 +97,39 @@ class SdkMemoryAnswerer:
         question: str,
         *,
         user_id: str = "local",
+        filters: SdkSearchFilters | None = None,
     ) -> tuple[SdkAskMemory, ...]:
+        """The user's active memories nearest the question.
+
+        `filters` are a caller's pass-through search options; absent, the
+        search is the one ask always made.
+        """
         client = await self.client_handle.client()
+        overrides = filters.search_overrides() if filters is not None else {}
         raw = await asyncio.to_thread(
             client.search,
             question,
-            user_ids=[user_id],
-            agent_ids=[],
-            limit=12,
-            min_score=sdk_relevance_floor(),
-            schema_min_score=sdk_relevance_floor(),
-            schema_limit=12,
-            intention_limit=12,
-            scene="normal",
-            reader=HY_MEMORY_READER,
+            **{
+                "user_ids": [user_id],
+                "agent_ids": [],
+                "limit": 12,
+                "min_score": sdk_relevance_floor(),
+                "schema_min_score": sdk_relevance_floor(),
+                "schema_limit": 12,
+                "intention_limit": 12,
+                "scene": "normal",
+                "reader": HY_MEMORY_READER,
+                **overrides,
+            },
         )
-        hits = parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown")
+        if filters is not None and overrides:
+            hits = tuple(
+                hit
+                for hit in parse_sdk_lane_search(raw, error_code="hy_memory_search_shape_unknown")
+                if filters.passes_floor(cast(SearchLane, hit.lane), hit.score)
+            )
+        else:
+            hits = parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown")
         found: list[SdkAskMemory] = []
         seen: set[str] = set()
         for hit in hits:
@@ -134,8 +152,9 @@ class SdkMemoryAnswerer:
         question: str,
         *,
         user_id: str = "local",
+        filters: SdkSearchFilters | None = None,
     ) -> SdkMemoryAnswer:
-        candidates = await self.retrieve(question, user_id=user_id)
+        candidates = await self.retrieve(question, user_id=user_id, filters=filters)
         if not candidates:
             return SdkMemoryAnswer(answer=_NO_GROUNDING, memories=(), mode="empty")
         profile = await self._active_profile()
@@ -166,8 +185,9 @@ class SdkMemoryAnswerer:
         question: str,
         *,
         user_id: str = "local",
+        filters: SdkSearchFilters | None = None,
     ) -> AsyncIterator[dict[str, object]]:
-        candidates = await self.retrieve(question, user_id=user_id)
+        candidates = await self.retrieve(question, user_id=user_id, filters=filters)
         await self._require_current(candidates)
         yield {"type": "recall", "memories": candidates}
         if not candidates:

memhub/apps/core/src/one_memory/sdk_memory_runtime.py

@@ -5,8 +5,8 @@ from __future__ import annotations
 import asyncio
 import logging
 import re
-from collections.abc import AsyncIterator, Callable, Iterable
-from contextlib import AsyncExitStack, asynccontextmanager, suppress
+from collections.abc import AsyncIterator, Callable
+from contextlib import AsyncExitStack, asynccontextmanager
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -33,16 +33,6 @@ from one_memory.semantic_deadline import SemanticMemoryHit
 _STABLE_CODE = re.compile(r"[a-z][a-z0-9_]{0,95}")
 
 logger = logging.getLogger(__name__)
-RAW_RETENTION_DAYS = 3
-RAW_RETENTION_CHECK_SECONDS = 6 * 60 * 60
-_RAW_STATUSES = (
-    "active",
-    "superseded",
-    "shadow",
-    "negated",
-    "conflicted",
-    "archived",
-)
 
 
 class SdkMemoryRuntime:
@@ -56,82 +46,17 @@ class SdkMemoryRuntime:
         write_queue: DurableWriteQueue,
         extraction_settings: ExtractionSettingsStore,
         data_dir: Path,
-        user_ids: Callable[[], Iterable[str]] = lambda: ("local",),
+        memory_mode: Callable[[], str] = lambda: "pro",
     ) -> None:
         self.coordinator = coordinator
-        #: Every user whose raw layer is kept to the retention window.
-        self.user_ids = user_ids
+        #: The SDK mode, for the graph store's path (`hy_graph_admin`).
+        self.memory_mode = memory_mode
         self.read_handle = read_handle
         self.write_queue = write_queue
         self.extraction_settings = extraction_settings
         self.data_dir = data_dir
-        self._raw_retention_task: asyncio.Task[None] | None = None
-
-    def start_raw_retention(self) -> None:
-        if self._raw_retention_task is None or self._raw_retention_task.done():
-            self._raw_retention_task = asyncio.create_task(
-                self._raw_retention_loop(),
-                name="one-memory-sdk-raw-retention",
-            )
-
-    async def prune_expired_raw(self, *, now: datetime | None = None) -> int:
-        deleted = 0
-        for user_id in dict.fromkeys(await asyncio.to_thread(lambda: tuple(self.user_ids()))):
-            deleted += await self._prune_expired_raw_for(user_id, now=now)
-        return deleted
-
-    async def _prune_expired_raw_for(self, user_id: str, *, now: datetime | None) -> int:
-        cutoff = (now or datetime.now(UTC)) - timedelta(days=RAW_RETENTION_DAYS)
-        cutoff_timestamp = int(cutoff.timestamp())
-        client = await self.read_handle.client()
-        deleted = 0
-        while True:
-            page = await asyncio.to_thread(
-                client.list_memories,
-                user_id=user_id,
-                layers=["l1_raw"],
-                statuses=list(_RAW_STATUSES),
-                order="asc",
-                limit=500,
-                offset=0,
-            )
-            bucket = page.get("vdb") if isinstance(page, dict) else None
-            rows = bucket.get("memories") if isinstance(bucket, dict) else None
-            if not isinstance(rows, list) or not rows:
-                return deleted
-            expired = [
-                row
-                for row in rows
-                if isinstance(row, dict)
-                and isinstance(row.get("memory_id"), str)
-                and int(row.get("memory_at") or row.get("gmt_created") or 0) < cutoff_timestamp
-            ]
-            if not expired:
-                return deleted
-            removed_page = 0
-            for row in expired:
-                result = await asyncio.to_thread(client.delete, row["memory_id"])
-                if isinstance(result, dict) and result.get("success") is True:
-                    count = int(result.get("deleted_count") or 0)
-                    deleted += count
-                    removed_page += count
-            if removed_page == 0:
-                return deleted
-
-    async def _raw_retention_loop(self) -> None:
-        while True:
-            await asyncio.sleep(RAW_RETENTION_CHECK_SECONDS)
-            try:
-                removed = await self.prune_expired_raw()
-                if removed:
-                    logger.info("sdk_raw_retention_deleted count=%d", removed)
-            except asyncio.CancelledError:
-                raise
-            except Exception as error:
-                logger.warning(
-                    "sdk_raw_retention_failed error_type=%s",
-                    type(error).__name__,
-                )
+        # MemHub keeps every raw the SDK stores, in every mode: there is no
+        # retention window (owner decision; the SDK itself never deletes raw).
 
     async def semantic_recall(
         self,
@@ -231,6 +156,7 @@ class SdkMemoryRuntime:
             profile,
             api_key=secret,
             data_dir=self.data_dir,
+            mode=self.memory_mode(),
         )
 
     async def decide_graph_width(
@@ -247,6 +173,7 @@ class SdkMemoryRuntime:
             api_key=secret,
             data_dir=self.data_dir,
             decision=decision,
+            mode=self.memory_mode(),
         )
 
     @asynccontextmanager
@@ -385,12 +312,7 @@ class SdkMemoryRuntime:
         return None
 
     async def shutdown(self) -> None:
-        if self._raw_retention_task is None:
-            return
-        self._raw_retention_task.cancel()
-        with suppress(asyncio.CancelledError):
-            await self._raw_retention_task
-        self._raw_retention_task = None
+        return None
 
 
 def _agent_ids(raw: object) -> tuple[str, ...]:

memhub/apps/core/src/one_memory/sdk_migration_runner.py

@@ -23,6 +23,7 @@ from pathlib import Path
 from typing import Any
 
 from one_memory.database_access import main_database
+from one_memory.hy_sdk_runtime import builtin_retrieval_snapshot
 from one_memory.imports.models import EvidenceRecord
 from one_memory.imports.storage import EvidenceStore
 from one_memory.memory_server.contracts import (
@@ -51,6 +52,7 @@ from one_memory.sdk_migration_bundle import inspect_sdk_migration_bundle
 from one_memory.sdk_migration_verify import (
     MemoryReadbackFailed,
     ReadbackReport,
+    expected_memory_ids,
     legacy_memory_ids_from,
     legacy_memory_ids_from_bundle,
     read_active_profile,
@@ -61,6 +63,8 @@ from one_memory.sdk_vector_reattach import (
     ReattachedCollection,
     legacy_vector_collections,
     reattach_legacy_vectors,
+    stored_memory_ids,
+    vectors_belong_to_another_space,
 )
 from one_memory.target_schema import drop_legacy_tables, legacy_tables_present
 
@@ -457,13 +461,17 @@ def resume_legacy_cleanup(*, data_dir: Path, database_path: Path) -> tuple[str,
             bundle_id=marker.backup_bundle_id,
         )
     from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
+    from one_memory.memory_visibility import drop_stale_triggers
 
     recover_abandoned_staging(
         data_dir=data_dir,
         database_path=database_path,
         namespace=HY_MEMORY_NAMESPACE,
     )
-    return drop_legacy_tables(database_path)
+    dropped = drop_legacy_tables(database_path)
+    # The visibility cache's grant triggers read a table just dropped (#233).
+    drop_stale_triggers(database_path)
+    return dropped
 
 
 MIGRATION_PREFIX = "sdk-authority-v2"
@@ -700,13 +708,14 @@ def run_startup_migration(
         step["authority"] = authority.value
     log.migration_id = authority_migration
     with log.step("preflight.profile_read") as step:
-        profile = read_active_profile(database_path)
-        step["profile_present"] = profile is not None
+        active = read_active_profile(database_path)
+        step["profile_present"] = active is not None
+    # No active profile means no generation model, not no embedding: the
+    # vectors are the built-in model's, which is also what the Runtime reads.
+    profile = active if active is not None else builtin_retrieval_snapshot()
     if authority is MemoryAuthority.SDK:
         with log.step("preflight.leftovers_scan") as step:
-            vectors_waiting = profile is not None and bool(
-                legacy_vector_collections(data_dir, profile)
-            )
+            vectors_waiting = bool(legacy_vector_collections(data_dir, profile))
             leftovers = bool(legacy_tables_present(database_path)) or abandoned_staging_present(
                 data_dir=data_dir, namespace=namespace
             )
@@ -717,7 +726,6 @@ def run_startup_migration(
         reattached: tuple[ReattachedCollection, ...] = ()
         readback: ReadbackReport | None = None
         if vectors_waiting:
-            assert profile is not None
             progress("vectors")
             with log.step("vectors.reattach") as step:
                 reattached = reattach_legacy_vectors(data_dir, profile)
@@ -769,12 +777,12 @@ def run_startup_migration(
         step["chat"] = len(legacy.chat)
         step["productivity"] = len(legacy.productivity)
         step["users"] = len(legacy.users)
-    if profile is None and not legacy.empty:
-        # Without the embedding profile the Runtime's collections cannot be
-        # named, so neither reattached nor read back. Legacy stays intact.
-        log.check("preflight.model_profile_check", "sdk_migration_model_profile_required")
-        raise DataPlaneMigrationError("sdk_migration_model_profile_required")
-    log.check("preflight.model_profile_check", None)
+    if active is None and not legacy.empty and vectors_belong_to_another_space(data_dir, profile):
+        # The memories live in the space of a profile that is no longer active,
+        # so the built-in model cannot read them. Legacy stays intact.
+        log.check("preflight.vector_space_check", "sdk_migration_vector_profile_missing")
+        raise DataPlaneMigrationError("sdk_migration_vector_profile_missing")
+    log.check("preflight.vector_space_check", None)
     with capture_fence(data_dir):
         # Builds through 19 left a staged SDK copy beside the original store.
         # Settling it before the bundle means the bundle holds the original
@@ -830,17 +838,23 @@ def run_startup_migration(
                 step.update(preparation.as_document())
             reattached = ()
             readback = None
-            if profile is not None:
-                progress("vectors")
-                with log.step("vectors.reattach") as step:
-                    reattached = reattach_legacy_vectors(data_dir, profile)
-                    step["collections"] = len(reattached)
-            if not legacy.empty:
-                assert profile is not None
+            # Before reattach touches the store, and after the staging
+            # recovery above has settled which store is the original.
+            with log.step("verify.stored_ids") as step:
+                expected, missing = expected_memory_ids(legacy, stored_memory_ids(data_dir))
+                step["chat_expected"] = len(expected.chat)
+                step["chat_missing_before_migration"] = missing
+            progress("vectors")
+            with log.step("vectors.reattach") as step:
+                reattached = reattach_legacy_vectors(data_dir, profile)
+                step["collections"] = len(reattached)
+            if not expected.empty:
                 progress("verify")
                 with log.step("verify.readback") as step:
                     with client_for(profile) as client:
-                        readback = verify_readback(client, legacy)
+                        readback = verify_readback(
+                            client, expected, chat_missing_before_migration=missing
+                        )
                     step.update(readback.as_document())
                     if not readback.passed:
                         store.set_phase(migration_id, phase="readback_failed", state="blocked")

memhub/apps/core/src/one_memory/sdk_migration_verify.py

@@ -14,12 +14,12 @@ import json
 import sqlite3
 from collections.abc import Iterator
 from contextlib import closing, contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any
 
 from one_memory.database_access import main_database
-from one_memory.model_profiles import ModelProfileSnapshot
+from one_memory.model_profiles import ModelProfileSnapshot, pinned_embedding_base_url
 from one_memory.sdk_data_plane_migration import DataPlaneMigrationError
 
 _PRODUCTIVITY_KINDS = frozenset({"coding", "working"})
@@ -47,6 +47,8 @@ class ReadbackReport:
     chat_readable: int
     productivity_expected: int
     productivity_readable: int
+    #: Confirmed in the mirror, gone from the SDK before this migration began.
+    chat_missing_before_migration: int = 0
 
     @property
     def passed(self) -> bool:
@@ -63,6 +65,7 @@ class ReadbackReport:
             "chat_readable": self.chat_readable,
             "productivity_expected": self.productivity_expected,
             "productivity_readable": self.productivity_readable,
+            "chat_missing_before_migration": self.chat_missing_before_migration,
             "passed": self.passed,
         }
 
@@ -115,6 +118,27 @@ def read_legacy_memory_ids(connection: sqlite3.Connection) -> LegacyMemoryIds:
     )
 
 
+def expected_memory_ids(
+    legacy: LegacyMemoryIds, stored: frozenset[str]
+) -> tuple[LegacyMemoryIds, int]:
+    """What the readback must find, and how many confirmed chats it need not.
+
+    A 1.0.4 library can hold confirmed rows whose SDK node the SDK had already
+    merged or deleted: no migration could read those back (658 confirmed, 657
+    in the store, and the whole upgrade failed). So chat is held to what the
+    store held before this run, and anything in there must still read back.
+
+    If none of the confirmed chats is in the store, that is not history: the
+    store is empty or is the wrong one. Nothing is excused then, so the
+    readback still fails on every id instead of checking nothing just before
+    the mirror is dropped.
+    """
+    chat = tuple(memory_id for memory_id in legacy.chat if memory_id in stored)
+    if not chat:
+        return legacy, 0
+    return replace(legacy, chat=chat), len(legacy.chat) - len(chat)
+
+
 def legacy_memory_ids_from(database_path: Path) -> LegacyMemoryIds:
     with main_database(database_path).read() as connection:
         return read_legacy_memory_ids(connection)
@@ -160,7 +184,14 @@ def read_active_profile(database_path: Path) -> ModelProfileSnapshot | None:
         model=row["model"],
         embedding_model=row["embedding_model"],
         embedding_provider_kind=row["embedding_provider_kind"],
-        embedding_base_url=row["embedding_base_url"],
+        # Read-only, beside a Core that may not have pinned it yet.
+        embedding_base_url=pinned_embedding_base_url(
+            provider_kind=row["provider_kind"],
+            base_url=row["base_url"],
+            embedding_model=row["embedding_model"],
+            embedding_provider_kind=row["embedding_provider_kind"],
+            embedding_base_url=row["embedding_base_url"],
+        ),
         embedding_dimensions=row["embedding_dimensions"],
         timeout_seconds=row["timeout_seconds"],
         max_concurrency=row["max_concurrency"],
@@ -179,14 +210,18 @@ def runtime_sdk_client(
     from one_memory.model_profiles import ModelProfileStore
     from one_memory.secrets import KeyringSecretStore, SecretStoreUnavailableError
 
-    try:
-        secret = ModelProfileStore(database_path).resolve_snapshot_secret(
-            profile.profile_id, profile.revision, KeyringSecretStore()
-        )
-    except SecretStoreUnavailableError:
-        raise DataPlaneMigrationError("sdk_migration_secure_storage_unavailable") from None
-    if profile.has_api_key and secret is None:
-        raise DataPlaneMigrationError("sdk_migration_model_secret_unavailable")
+    secret: str | None = None
+    # Without a key there is nothing to resolve, and the built-in retrieval
+    # snapshot has no stored version to look one up by.
+    if profile.has_api_key:
+        try:
+            secret = ModelProfileStore(database_path).resolve_snapshot_secret(
+                profile.profile_id, profile.revision, KeyringSecretStore()
+            )
+        except SecretStoreUnavailableError:
+            raise DataPlaneMigrationError("sdk_migration_secure_storage_unavailable") from None
+        if secret is None:
+            raise DataPlaneMigrationError("sdk_migration_model_secret_unavailable")
     lease = acquire_client(profile, api_key=secret, data_dir=data_dir)
     try:
         yield lease.client
@@ -194,7 +229,9 @@ def runtime_sdk_client(
         lease.close()
 
 
-def verify_readback(client: Any, legacy: LegacyMemoryIds) -> ReadbackReport:
+def verify_readback(
+    client: Any, legacy: LegacyMemoryIds, *, chat_missing_before_migration: int = 0
+) -> ReadbackReport:
     listed: set[str] = set()
     for user_id in legacy.users or ("local",):
         offset = 0
@@ -228,5 +265,6 @@ def verify_readback(client: Any, legacy: LegacyMemoryIds) -> ReadbackReport:
         chat_readable=sum(1 for memory_id in legacy.chat if readable(memory_id)),
         productivity_expected=len(legacy.productivity),
         productivity_readable=sum(1 for memory_id in legacy.productivity if readable(memory_id)),
+        chat_missing_before_migration=chat_missing_before_migration,
     )
 

memhub/apps/core/src/one_memory/sdk_read_handle.py

@@ -0,0 +1,125 @@
+"""The read client, kept in the embedding space the active profile writes to.
+
+Every read used to open the built-in 512-dim collection while writes followed
+the active profile, so a custom embedding read an empty space (#242). This
+handle opens the active profile's `retrieval_view` instead, and the built-in
+space only when no profile is active.
+
+A profile change builds a new client and retires the old one; a retired client
+is closed once no task still uses it, as `SdkWriteCoordinator` does for writes
+(#218). A use is the task that asked for the client: callers keep their
+`client = await handle.client()` shape, and the use ends with that task (one
+HTTP request, one retention pass).
+
+The memory mode (lite / pro / ultra) is fixed per SDK client, so a mode change
+rebuilds and retires the same way: handles are keyed by (profile, mode). The
+mode never names a collection; it is a separate client parameter.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from pathlib import Path
+from typing import Any
+
+from one_memory.hy_graph_admin import effective_memory_mode
+from one_memory.hy_sdk_runtime import (
+    BUILTIN_RETRIEVAL_PROFILE_ID,
+    SdkClientHandle,
+    builtin_retrieval_snapshot,
+    retrieval_view,
+)
+from one_memory.hy_sdk_types import HySdkRuntimeError
+from one_memory.model_profiles import ModelProfileStore, ProviderKind
+from one_memory.secrets import SecretStore, SecretStoreUnavailableError
+
+_BUILTIN = (BUILTIN_RETRIEVAL_PROFILE_ID, 0)
+
+
+class ActiveProfileReadHandle(SdkClientHandle):
+    """An `SdkClientHandle` whose profile follows the active one."""
+
+    def __init__(
+        self,
+        *,
+        profiles: ModelProfileStore,
+        secrets: SecretStore,
+        data_dir: Path,
+        memory_mode: Callable[[], str] = effective_memory_mode,
+    ) -> None:
+        super().__init__(builtin_retrieval_snapshot(), api_key=None, data_dir=data_dir)
+        self.profiles = profiles
+        self.secrets = secrets
+        self.data_dir = data_dir
+        self.memory_mode = memory_mode
+        self._handles: dict[tuple[str, int, str], SdkClientHandle] = {}
+        #: Tasks in flight per handle; a retired handle closes at zero.
+        self._uses: dict[SdkClientHandle, int] = {}
+        self._retired: set[SdkClientHandle] = set()
+        self._closing: set[asyncio.Task[None]] = set()
+        self._handle_lock = asyncio.Lock()
+
+    async def client(self) -> Any:
+        profile = await asyncio.to_thread(self.profiles.active_identity) or _BUILTIN
+        identity = (*profile, self.memory_mode())
+        async with self._handle_lock:
+            handle = self._handles.get(identity)
+            if handle is None:
+                # Only on a change of profile: resolving a key can reach the
+                # system keychain, so it is not done on every read.
+                handle = await asyncio.to_thread(self._build, identity)
+                self._handles[identity] = handle
+            self._retired.update(value for key, value in self._handles.items() if key != identity)
+            self._handles = {identity: handle}
+            unused = {value for value in self._retired if not self._uses.get(value)}
+            self._retired -= unused
+            self._hold(handle)
+        await asyncio.gather(*(value.close() for value in unused))
+        return await handle.client()
+
+    async def close(self) -> None:
+        async with self._handle_lock:
+            handles = (*self._handles.values(), *self._retired)
+            self._handles, self._retired, self._uses = {}, set(), {}
+        await asyncio.gather(*(handle.close() for handle in handles), *self._closing)
+
+    def _hold(self, handle: SdkClientHandle) -> None:
+        """Count a use of `handle` until the asking task finishes."""
+        task = asyncio.current_task()
+        if task is None:  # pragma: no cover - always called from a coroutine
+            return
+        self._uses[handle] = self._uses.get(handle, 0) + 1
+        task.add_done_callback(lambda _task: self._release(handle))
+
+    def _release(self, handle: SdkClientHandle) -> None:
+        remaining = self._uses.get(handle, 0) - 1
+        if remaining > 0:
+            self._uses[handle] = remaining
+            return
+        self._uses.pop(handle, None)
+        if handle in self._retired:
+            self._retired.discard(handle)
+            closing = asyncio.get_running_loop().create_task(handle.close())
+            self._closing.add(closing)
+            closing.add_done_callback(self._closing.discard)
+
+    def _build(self, key: tuple[str, int, str]) -> SdkClientHandle:
+        profile_id, revision, mode = key
+        identity = (profile_id, revision)
+        if identity == _BUILTIN:
+            return SdkClientHandle(
+                builtin_retrieval_snapshot(), api_key=None, data_dir=self.data_dir, mode=mode
+            )
+        profile = self.profiles.snapshot(*identity)
+        view = retrieval_view(profile)
+        secret: str | None = None
+        # Only a remote OpenAI-compatible embedding sends the shared key.
+        if view.embedding_provider_kind is ProviderKind.OPENAI_COMPATIBLE:
+            try:
+                secret = self.profiles.resolve_snapshot_secret(*identity, self.secrets)
+            except SecretStoreUnavailableError:
+                raise HySdkRuntimeError("secure_storage_unavailable") from None
+            if secret is None:
+                raise HySdkRuntimeError("model_secret_unavailable")
+        return SdkClientHandle(view, api_key=secret, data_dir=self.data_dir, mode=mode)

memhub/apps/core/src/one_memory/sdk_search_options.py

@@ -0,0 +1,179 @@
+"""SDK search parameters a host may pass through to `client.search`.
+
+`POST /api/v2/context/search`, `POST /api/v1/recall` and the v2 ask routes
+used to fix every SDK search knob, so a host that wanted a stricter relevance
+floor filtered the answer itself. These fields are the public `search`
+parameters of hy-memory 2.0.9 that are safe to hand a caller. Every field is
+optional: absent, the route keeps exactly the call it made before.
+
+Never exposed: `user_ids` (a route searches its one scoped `user_id`),
+`embed_override` and `query_embedding` (they bypass the configured embedder),
+and `request_id` (Core's own correlation id).
+
+A caller's floor is also applied to what Core returns, not only handed to the
+SDK: a hit below the floor that names its lane is dropped (`passes_floor`).
+With `min_score` alone that floor covers every lane, so `min_score=0.9`
+answers only results scoring at least 0.9.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, StrictBool, model_validator
+
+#: The SDK's `MemoryLayer` values (hy-memory 2.0.9).
+MemoryLayerName = Literal[
+    "l0_basic_info",
+    "l1_raw",
+    "l2_fact",
+    "l3_summary",
+    "l4_identity",
+    "l5_knowledge",
+    "l6_schema",
+    "l7_intention",
+]
+MEMORY_LAYERS: tuple[str, ...] = (
+    "l0_basic_info",
+    "l1_raw",
+    "l2_fact",
+    "l3_summary",
+    "l4_identity",
+    "l5_knowledge",
+    "l6_schema",
+    "l7_intention",
+)
+#: The SDK's `MemoryStatus` values.
+MEMORY_STATUSES: tuple[str, ...] = (
+    "active",
+    "superseded",
+    "negated",
+    "conflicted",
+    "shadow",
+    "archived",
+)
+SearchScene = Literal["chat", "working", "coding", "all"]
+#: Which recall lane produced a hit, for the floor that applies to it.
+SearchLane = Literal["normal", "schema", "proactive", "working", "coding", "raw"]
+
+Score = Annotated[float, Field(ge=0.0, le=1.0)]
+SourceName = Annotated[str, Field(min_length=1, max_length=128)]
+ScopeId = Annotated[str, Field(min_length=1, max_length=512)]
+
+
+class SdkSearchFilters(BaseModel):
+    """Filters and floors every search route accepts (recall, context, ask)."""
+
+    model_config = ConfigDict(frozen=True, extra="forbid")
+
+    #: Chat lane floor; with nothing more specific, every lane's.
+    min_score: Score | None = None
+    #: Schema (L6) lane floor; defaults to `min_score` when only that is set.
+    schema_min_score: Score | None = None
+    #: False skips the SDK's reranking stages for this call.
+    rerank: StrictBool | None = None
+    include_sources: tuple[SourceName, ...] | None = Field(default=None, max_length=64)
+    exclude_sources: tuple[SourceName, ...] | None = Field(default=None, max_length=64)
+    #: Unix seconds: only memories created at or after it.
+    created_after: float | None = Field(default=None, ge=0)
+    session_ids: tuple[ScopeId, ...] | None = Field(default=None, max_length=64)
+    agent_ids: tuple[ScopeId, ...] | None = Field(default=None, max_length=64)
+    #: Only these layers (pro/ultra; lite always searches raw).
+    layers: tuple[MemoryLayerName, ...] | None = Field(default=None, min_length=1, max_length=8)
+
+    def filter_kwargs(self) -> dict[str, object]:
+        """The SDK keyword arguments this request sets, and only those."""
+        values: dict[str, object] = {}
+        for name in (
+            "min_score",
+            "schema_min_score",
+            "rerank",
+            "created_after",
+        ):
+            value = getattr(self, name)
+            if value is not None:
+                values[name] = value
+        for name in ("include_sources", "exclude_sources", "session_ids", "agent_ids", "layers"):
+            value = getattr(self, name)
+            if value is not None:
+                values[name] = list(dict.fromkeys(value))
+        if self.min_score is not None and self.schema_min_score is None:
+            values["schema_min_score"] = self.min_score
+        return values
+
+    def lane_floor(self, lane: SearchLane) -> float | None:
+        """The caller's floor for one lane, or None when the caller set none."""
+        if lane == "schema":
+            return self.schema_min_score if self.schema_min_score is not None else self.min_score
+        return self.min_score
+
+    def passes_floor(self, lane: SearchLane, score: float) -> bool:
+        floor = self.lane_floor(lane)
+        return floor is None or score >= floor
+
+    def search_overrides(self) -> Mapping[str, object]:
+        return self.filter_kwargs()
+
+
+class SdkSearchOptions(SdkSearchFilters):
+    """Every SDK search knob a recall or context caller may set."""
+
+    #: working/coding fact-layer floor (SDK default 0.30).
+    working_fact_min_score: Score | None = None
+    #: working/coding experience-layer floor (SDK default fact floor + 0.10).
+    working_experience_min_score: Score | None = None
+    reader: Literal["legacy", "hybrid"] | None = None
+    #: chat (the default), working, coding, or all three fused.
+    scene: SearchScene | None = None
+    #: agentic spends one model call to classify and rewrite the query.
+    search_mode: Literal["default", "agentic"] | None = None
+    #: working/coding lanes only; `branch` needs `workspace_id`.
+    workspace_id: ScopeId | None = None
+    branch: ScopeId | None = None
+
+    @model_validator(mode="after")
+    def branch_needs_workspace(self) -> SdkSearchOptions:
+        if self.branch is not None and self.workspace_id is None:
+            raise ValueError("branch requires workspace_id")
+        return self
+
+    def search_overrides(self) -> Mapping[str, object]:
+        values = self.filter_kwargs()
+        for name in (
+            "working_fact_min_score",
+            "working_experience_min_score",
+            "reader",
+            "scene",
+            "search_mode",
+            "workspace_id",
+            "branch",
+        ):
+            value = getattr(self, name)
+            if value is not None:
+                values[name] = value
+        if self.min_score is not None and self.working_fact_min_score is None:
+            values["working_fact_min_score"] = self.min_score
+        if self.min_score is not None and self.working_experience_min_score is None:
+            values["working_experience_min_score"] = self.min_score
+        return values
+
+    def lane_floor(self, lane: SearchLane) -> float | None:
+        if lane in {"working", "coding"}:
+            explicit = [
+                value
+                for value in (self.working_fact_min_score, self.working_experience_min_score)
+                if value is not None
+            ]
+            # Facts and experiences share one score field on a fused hit, so
+            # the looser of the two floors is the one Core can enforce.
+            return min(explicit) if explicit else self.min_score
+        return super().lane_floor(lane)
+
+
+def split_csv(values: list[str] | None) -> list[str]:
+    """Query values given repeated (`a&a`) or comma separated (`a,b`)."""
+    found: list[str] = []
+    for value in values or ():
+        found.extend(part.strip() for part in value.split(",") if part.strip())
+    return list(dict.fromkeys(found))

memhub/apps/core/src/one_memory/sdk_vector_reattach.py

@@ -9,6 +9,7 @@ one. Runtime code only reaches vectors through the SDK. This module runs from
 
 from __future__ import annotations
 
+import re
 import sqlite3
 from collections.abc import Iterator
 from contextlib import closing
@@ -21,6 +22,8 @@ from one_memory.model_profiles import ModelProfileSnapshot
 from one_memory.sdk_data_plane_migration import DataPlaneMigrationError
 
 _BATCH = 500
+#: Every name a MemHub build gave an embedding space's base collection.
+_SPACE_NAME = re.compile(r"^(?:memhub_[0-9a-f]{16}|one_memory_[0-9a-f]{12})(?:_|$)")
 
 
 def open_offline_chroma(directory: Path) -> Any:
@@ -51,13 +54,20 @@ def _vector_directory(data_dir: Path) -> Path:
     return data_dir / "hy-memory" / "governed-v2" / "vector"
 
 
-def _stored_names(data_dir: Path) -> tuple[str, ...]:
+def _stored_names(data_dir: Path, *, filled: bool = False) -> tuple[str, ...]:
+    """Collection names; with `filled`, only those holding at least one row."""
     database = _vector_directory(data_dir) / "chroma.sqlite3"
     if not database.is_file():
         return ()
     with closing(sqlite3.connect(f"file:{database}?mode=ro", uri=True)) as connection:
         try:
-            rows = connection.execute("SELECT name FROM collections").fetchall()
+            rows = connection.execute(
+                "SELECT DISTINCT collections.name FROM collections "
+                "JOIN segments ON segments.collection = collections.id "
+                "JOIN embeddings ON embeddings.segment_id = segments.id"
+                if filled
+                else "SELECT name FROM collections"
+            ).fetchall()
         except sqlite3.Error:
             return ()
     return tuple(sorted(str(row[0]) for row in rows))
@@ -80,6 +90,59 @@ def legacy_vector_collections(
     return tuple(pairs)
 
 
+def vectors_belong_to_another_space(data_dir: Path, profile: ModelProfileSnapshot) -> bool:
+    """Vectors exist, but only in embedding spaces `profile` does not name.
+
+    Counted rows, not names: every Runtime opens its read collection with
+    `get_or_create`, so an empty collection under the expected base proves
+    nothing. Read-only, so a preflight can ask before anything changes.
+    """
+    current, legacy_bases = vector_collection_bases(profile)
+    known = (current, *legacy_bases)
+    filled = _stored_names(data_dir, filled=True)
+    return not any(
+        name == base or name.startswith(f"{base}_") for name in filled for base in known
+    ) and any(_SPACE_NAME.match(name) for name in filled)
+
+
+def stored_memory_ids(data_dir: Path) -> frozenset[str]:
+    """Every id in the vector store as found, across all embedding spaces.
+
+    All of them on purpose. A memory stranded under a space the Runtime will
+    not read must count as present, so the readback then fails on it instead
+    of excusing it. Both the point id and the SDK's `node_id` are kept: the
+    SDK derives a point id from a node id that is not a UUID.
+
+    Callers must have stopped every SDK writer. A store that cannot be read
+    raises: "unknown" must never be taken for "absent".
+    """
+    if not (_vector_directory(data_dir) / "chroma.sqlite3").is_file():
+        return frozenset()
+    found: set[str] = set()
+    try:
+        client = open_offline_chroma(_vector_directory(data_dir))
+        try:
+            for collection in client.list_collections():
+                offset = 0
+                while True:
+                    page = collection.get(include=["metadatas"], limit=_BATCH, offset=offset)
+                    ids = list(page["ids"])
+                    if not ids:
+                        break
+                    found.update(ids)
+                    found.update(
+                        str(metadata["node_id"])
+                        for metadata in page["metadatas"] or ()
+                        if metadata and metadata.get("node_id")
+                    )
+                    offset += len(ids)
+        finally:
+            client.close()
+    except Exception as error:
+        raise DataPlaneMigrationError("sdk_migration_vector_store_unreadable") from error
+    return frozenset(found)
+
+
 def _records(collection: Any) -> Iterator[dict[str, Any]]:
     offset = 0
     while True:

memhub/apps/core/src/one_memory/sdk_write_processor.py

@@ -9,7 +9,8 @@ from dataclasses import dataclass
 from datetime import datetime
 from typing import Any
 
-from one_memory.hy_sdk_runtime import SdkClientHandle
+from one_memory.hy_graph_admin import effective_memory_mode
+from one_memory.hy_sdk_runtime import SdkClientHandle, builtin_retrieval_snapshot
 from one_memory.imports.models import EvidenceContentKind, EvidenceOrigin, EvidenceRecord
 from one_memory.input_sanitization import sanitize_model_input
 from one_memory.model_profiles import ModelProfileSnapshot, ModelProfileStore
@@ -31,8 +32,11 @@ class SdkWriteReceipt:
 class SdkWriteProcessor:
     """Turn one durable job into exactly one public SDK write call."""
 
-    def __init__(self, client_handle: SdkClientHandle) -> None:
+    def __init__(self, client_handle: SdkClientHandle, *, lite: bool = False) -> None:
         self._client_handle = client_handle
+        #: lite makes no model call: `add_extracted` would still call one, so a
+        #: direct memory note is stored as one raw user turn through `add`.
+        self._lite = lite
 
     async def process_sdk_job(
         self,
@@ -64,7 +68,7 @@ class SdkWriteProcessor:
             # waits and retries instead of being left `unknown`.
             raise RuntimeError("hy_memory_client_unavailable") from error
 
-        if extracted:
+        if extracted and not self._lite:
             contents = [sanitize_model_input(record.verbatim) for record in records]
             await before_hy_invocation()
             result = await asyncio.to_thread(
@@ -81,12 +85,15 @@ class SdkWriteProcessor:
         else:
             messages = [
                 {
-                    "role": record.role,
+                    "role": "user" if extracted else record.role,
                     "content": sanitize_model_input(record.verbatim),
                 }
                 for record in records
-                if record.content_kind is EvidenceContentKind.RAW_CONVERSATION
-                and record.role in _MESSAGE_ROLES
+                if extracted
+                or (
+                    record.content_kind is EvidenceContentKind.RAW_CONVERSATION
+                    and record.role in _MESSAGE_ROLES
+                )
             ]
             if not messages:
                 raise RuntimeError("hy_memory_task_without_inputs")
@@ -138,6 +145,15 @@ class SdkWriteCoordinator:
     from under a write still running on it: the SDK cleared its stores and
     the SharedRuntime closed mid-`add`, the job failed with AttributeError
     after HY was invoked, and it was left `unknown`.
+
+    Handles are keyed by profile and memory mode, so a mode change retires
+    handles the same way. In lite no generation model is needed: without an
+    active profile the built-in local embedder writes raw turns.
+
+    `exclusive_for_evolve(user_id)` keeps evolve and that user's writes
+    apart: it waits for the user's write in flight, and while it runs that
+    user's jobs are refused before the SDK is called
+    (`memory_evolve_in_progress`), so the drain puts them back.
     """
 
     def __init__(
@@ -148,13 +164,22 @@ class SdkWriteCoordinator:
         data_dir: Any,
         transport_factory: Callable[[ModelProfileSnapshot], Callable[[], Any] | None],
         sedimentation_enabled: Callable[[], bool],
+        memory_mode: Callable[[], str] = effective_memory_mode,
     ) -> None:
         self.profiles = profiles
         self.secrets = secrets
         self.data_dir = data_dir
         self.transport_factory = transport_factory
         self.sedimentation_enabled = sedimentation_enabled
-        self._handles: dict[tuple[str, int], SdkClientHandle] = {}
+        self.memory_mode = memory_mode
+        self._handles: dict[tuple[str, int, str], SdkClientHandle] = {}
+        #: Users evolve (or the graph backfill) is running for, the writes in
+        #: flight per user, and one exclusive run per user at a time. Writes
+        #: of one user still run side by side; only evolve waits for them.
+        self._evolving: set[str] = set()
+        self._writing: dict[str, int] = {}
+        self._writes_idle = asyncio.Condition()
+        self._exclusive: dict[str, asyncio.Lock] = {}
         #: Uses in flight per handle; a retired handle closes at zero.
         self._uses: dict[SdkClientHandle, int] = {}
         self._retired: set[SdkClientHandle] = set()
@@ -169,16 +194,66 @@ class SdkWriteCoordinator:
     ) -> SdkWriteReceipt:
         if not self.sedimentation_enabled():
             raise RuntimeError("model_sedimentation_disabled")
-        async with self._active_handle() as handle:
-            return await SdkWriteProcessor(handle).process_sdk_job(
-                records,
-                request_id=request_id,
-                before_hy_invocation=before_hy_invocation,
-            )
+        user_id = records[0].user_id if records else "local"
+        if user_id in self._evolving:
+            # Before any SDK call: the drain returns the job to pending.
+            raise RuntimeError("memory_evolve_in_progress")
+        self._writing[user_id] = self._writing.get(user_id, 0) + 1
+        try:
+            async with self._active_handle() as handle:
+                return await SdkWriteProcessor(
+                    handle, lite=getattr(handle, "mode", None) == "lite"
+                ).process_sdk_job(
+                    records,
+                    request_id=request_id,
+                    before_hy_invocation=before_hy_invocation,
+                )
+        finally:
+            remaining = self._writing.get(user_id, 1) - 1
+            if remaining > 0:
+                self._writing[user_id] = remaining
+            else:
+                self._writing.pop(user_id, None)
+            async with self._writes_idle:
+                self._writes_idle.notify_all()
 
-    async def inquire_write_outcome(
-        self, request_id: str
-    ) -> SdkWriteReceipt | None:
+    async def retire_stale_modes(self) -> None:
+        """After a mode change: retire handles of other modes, closing the idle ones.
+
+        A handle still writing is closed when its last use ends (#218), so a
+        write that started in the old mode finishes in it.
+        """
+        mode = self.memory_mode()
+        async with self._lock:
+            stale = {key: handle for key, handle in self._handles.items() if key[2] != mode}
+            for key in stale:
+                del self._handles[key]
+            self._retired.update(stale.values())
+            unused = {value for value in self._retired if not self._uses.get(value)}
+            self._retired -= unused
+        await asyncio.gather(*(value.close() for value in unused))
+
+    def evolving(self, user_id: str) -> bool:
+        """Whether evolve holds this user, so their queued writes should wait."""
+        return user_id in self._evolving
+
+    @asynccontextmanager
+    async def exclusive_for_evolve(self, user_id: str) -> AsyncIterator[None]:
+        """Hold one user's writes off while evolve runs for them.
+
+        New writes of the user are refused from here on; the ones already in
+        flight are waited for before the body runs.
+        """
+        async with self._exclusive.setdefault(user_id, asyncio.Lock()):
+            self._evolving.add(user_id)
+            try:
+                async with self._writes_idle:
+                    await self._writes_idle.wait_for(lambda: not self._writing.get(user_id))
+                yield
+            finally:
+                self._evolving.discard(user_id)
+
+    async def inquire_write_outcome(self, request_id: str) -> SdkWriteReceipt | None:
         async with self._active_handle() as handle:
             return await SdkWriteProcessor(handle).inquire_write_outcome(request_id)
 
@@ -204,9 +279,11 @@ class SdkWriteCoordinator:
     @asynccontextmanager
     async def _active_handle(self) -> AsyncIterator[SdkClientHandle]:
         """Use the active profile's handle; retire the others, closing the unused."""
-        identity, profile, secret = await asyncio.to_thread(
-            self._read_active_credentials
+        mode = self.memory_mode()
+        profile_identity, profile, secret = await asyncio.to_thread(
+            self._read_active_credentials, mode
         )
+        identity = (*profile_identity, mode)
         async with self._lock:
             handle = self._handles.get(identity)
             if handle is None:
@@ -215,6 +292,7 @@ class SdkWriteCoordinator:
                     api_key=secret,
                     data_dir=self.data_dir,
                     http_transport_factory=self.transport_factory(profile),
+                    mode=mode,
                 )
                 self._handles[identity] = handle
             self._retired.update(value for key, value in self._handles.items() if key != identity)
@@ -239,9 +317,14 @@ class SdkWriteCoordinator:
 
     def _read_active_credentials(
         self,
+        mode: str = "pro",
     ) -> tuple[tuple[str, int], ModelProfileSnapshot, str | None]:
         identity = self.profiles.active_identity()
         if identity is None:
+            if mode == "lite":
+                # lite stores raw turns with one embedding and calls no model.
+                builtin = builtin_retrieval_snapshot()
+                return (builtin.profile_id, builtin.revision), builtin, None
             raise RuntimeError("model_profile_not_active")
         profile = self.profiles.snapshot(*identity)
         try:

memhub/apps/core/src/one_memory/semantic_deadline.py

@@ -27,6 +27,11 @@ class SemanticMemoryHit:
     member_ids: tuple[str, ...] = ()
     sdk_formatted: bool = False
     formatting_complete: bool = False
+    #: The recall lane that produced the hit (`sdk_search_options.SearchLane`)
+    #: and the namespace to read it back from. Defaults are a chat memory.
+    lane: str = "normal"
+    kind: str = "chat"
+    variant: str | None = None
 
     def as_dict(self) -> dict[str, object]:
         return {

memhub/apps/core/tests/TEST_INDEX.md

@@ -161,6 +161,7 @@
 | `test_memory_evidence_freshness.py` | `contract` | 新鲜度拆除后：读路径恒为 current；证据修订 / 畸形 proof / segment ref 不再藏记忆或 503。 |
 | `test_memory_freshness_integration_review.py` | `integration` | Independent synthetic read-path review; no lifespan, model or user database. |
 | `test_memory_library_all_scenes.py` | `integration` | The v2 memory library across chat, working and coding memories. |
+| `test_memory_mode_and_evolve.py` | `contract` | The memory mode (lite / pro / ultra), evolve and its nightly schedule. |
 | `test_memory_qa.py` | `contract` | 覆盖 search terms keep meaningful single chinese character、streamed answer emits recall then incremental grounded text、governance change during generation stops the old answer 等 14 个场景。 |
 | `test_memory_qa_api.py` | `integration` | 覆盖 ask stream endpoint emits recall deltas and final、ask stream endpoint requires the local token 场景。 |
 | `test_memory_qa_permissions.py` | `contract` | 覆盖 initial refusal precedes all secret access and network、invalid authorization result fails before secret access、authorization receives exact question and only outbound grounding 等 16 个场景。 |
@@ -248,6 +249,7 @@
 | `test_schema_gate.py` | `contract` | 覆盖 fresh database records the build schema、future schema refuses to open、legacy database without a ledger is left untouched 场景。 |
 | `test_schema_migrations.py` | `contract` | 覆盖 evidence store adds source ordinal to previous schema、import store migrates previous job and error schema、old active job index lets a sync sit beside paused history 等 4 个场景。 |
 | `test_schema_ownership_gate.py` | `contract` | A1-03/A1-04: schema ownership, and the hard part of telling DDL from reads. |
+| `test_sdk_api_surface.py` | `contract` | The SDK surface MemHub exposes: search pass-through, layers, raw, v2 aliases. |
 | `test_sdk_client_construction.py` | `contract` | SDK clients of one home are constructed one at a time. |
 | `test_sdk_data_plane_migration.py` | `contract` | SDK-authority migration ledger, aliases and atomic cutover marker. |
 | `test_sdk_memory_qa.py` | `contract` | 覆盖 sdk ask propagates sdk search failure without legacy fallback、sdk ask revalidates memory before returning extract、sdk ask uses native sdk memory without governance fields 场景。 |
@@ -311,4 +313,4 @@
 | `test_workspace_recall.py` | `contract` | Workspace memories were unreachable from every host; these pin the way in. |
 | `test_write_queue_lease_renewal.py` | `contract` | Rule 8: a lease must outlive the work it covers, and be renewed while it runs. |
 
-共 306 个测试文件。
+共 308 个测试文件。

memhub/apps/core/tests/core_test_classification.py

@@ -14,8 +14,8 @@ from pathlib import Path
 import pytest
 
 PRIMARY_MARKERS = ("unit", "contract", "integration", "performance", "live")
-EXPECTED_TEST_FILE_COUNT = 306
-EXPECTED_TEST_FILES_SHA256 = "e616dcf9fea288bf15a4445045f7b4278517f3304091c04fffd55be80308d481"
+EXPECTED_TEST_FILE_COUNT = 308
+EXPECTED_TEST_FILES_SHA256 = "3aadf5953e41c01d4b60469f47242c00d2acef3c42550aa017255e31d216b589"
 
 # Pure transformations, validation models, parity checks and state machines.
 # This list is intentionally small: uncertain files default to contract.

memhub/apps/core/tests/graph_fixtures.py

@@ -24,7 +24,8 @@ def model_snapshot(
         model="memory-model",
         embedding_model="embedding-model",
         embedding_provider_kind=embedding_provider,
-        embedding_base_url=embedding_base_url,
+        # Stored profiles always name it; see `pinned_embedding_base_url`.
+        embedding_base_url=embedding_base_url or base_url,
         embedding_dimensions=768,
         timeout_seconds=30,
         max_concurrency=1,

memhub/apps/core/tests/test_cli_restore.py

@@ -175,7 +175,13 @@ def test_sdk_migrate_reports_each_step_then_the_result(tmp_path, monkeypatch) ->
 
     assert code == 0, output.getvalue()
     lines = [json.loads(line) for line in output.getvalue().splitlines()]
-    assert [line["step"] for line in lines[:-1]] == ["bundle", "prepare", "cutover", "cleanup"]
+    assert [line["step"] for line in lines[:-1]] == [
+        "bundle",
+        "prepare",
+        "vectors",
+        "cutover",
+        "cleanup",
+    ]
     assert lines[-1]["state"] == "cutover"
     assert lines[-1]["bundle_id"] == f"{lines[-1]['migration_id']}-pre-cutover"
     assert not (tmp_path / ".sdk-migration-capture-fence.json").exists()

memhub/apps/core/tests/test_core_ready_before_recovery.py

@@ -164,7 +164,7 @@ def test_metadata_repair_never_overlaps_an_sdk_client(tmp_path, monkeypatch):
         events.append("repair_finished")
         return MetadataRepairResult(None, 0, 0, 0)
 
-    def open_client(profile, *, api_key, data_dir, http_transport_factory):
+    def open_client(profile, *, api_key, data_dir, http_transport_factory, mode="pro"):
         events.append("client_opened")
         return ClientLease(object(), data_dir.resolve())
 

memhub/apps/core/tests/test_memory_batch_extraction.py

@@ -26,6 +26,7 @@ def snapshot() -> ModelProfileSnapshot:
         base_url="http://127.0.0.1:11434",
         model="memory-model",
         embedding_model="embedding-model",
+        embedding_base_url="http://127.0.0.1:11434",
         embedding_dimensions=768,
         timeout_seconds=30,
         max_concurrency=1,

memhub/apps/core/tests/test_memory_mode_and_evolve.py

@@ -0,0 +1,861 @@
+"""The memory mode (lite / pro / ultra), evolve and its nightly schedule.
+
+Item 5: the mode is a saved setting (default pro, `ONE_MEMORY_GRAPH_STORE`
+still forces ultra), it decides the mode every SDK client is built with, and
+a change rebuilds the handles. lite needs no generation model, never calls
+`add_extracted`, and keeps every raw. Item 6 (P1-3 / R7): `POST
+/api/v2/evolve` is ultra only, one run per user at a time, never beside that
+user's write, and loops while the SDK says `should_continue`; the schedule
+runs nightly and catches up after sleep. R10: the status carries
+`pipeline.mode` and the last evolve.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from contextlib import asynccontextmanager
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from pydantic import SecretStr
+
+from one_memory import hy_sdk_runtime, sdk_write_processor
+from one_memory.api import _sdk_client_handle, create_app, memory_mode_effects
+from one_memory.config import Settings
+from one_memory.hy_graph_admin import effective_memory_mode
+from one_memory.imports.models import (
+    EvidenceContentKind,
+    EvidenceInput,
+    EvidenceOrigin,
+    OriginMetadata,
+    WrittenBy,
+    materialize_write_input,
+)
+from one_memory.memory_evolution import (
+    EvolveModeError,
+    EvolveScheduler,
+    EvolveService,
+    EvolveStateStore,
+    MemoryGraphBackfill,
+    evolve_due,
+)
+from one_memory.memory_server.config_state import ConfigStore, EvolveSchedule
+from one_memory.sdk_memory_runtime import SdkMemoryRuntime
+from one_memory.secrets import InMemorySecretStore
+
+TOKEN = "memory-mode-test"
+
+
+# --- the setting and the client mode -----------------------------------------
+
+
+def test_mode_defaults_to_pro_and_the_env_still_forces_ultra(tmp_path, monkeypatch) -> None:
+    monkeypatch.delenv("ONE_MEMORY_GRAPH_STORE", raising=False)
+    store = ConfigStore(tmp_path)
+    assert store.memory_mode() == "pro"
+    assert effective_memory_mode(store.memory_mode()) == "pro"
+    store.set_memory_mode("lite")
+    assert ConfigStore(tmp_path).memory_mode() == "lite"
+    assert json.loads((tmp_path / "config.json").read_text())["memory_mode"] == "lite"
+    monkeypatch.setenv("ONE_MEMORY_GRAPH_STORE", "kuzu")
+    assert effective_memory_mode("lite") == "ultra"
+    monkeypatch.setenv("ONE_MEMORY_GRAPH_STORE", "none")
+    assert effective_memory_mode("lite") == "lite"
+    with pytest.raises(ValueError):
+        store.set_memory_mode("max")  # type: ignore[arg-type]
+
+
+def test_an_untouched_install_writes_no_new_settings(tmp_path) -> None:
+    store = ConfigStore(tmp_path)
+    store.set_sync("cursor", True)
+    saved = json.loads((tmp_path / "config.json").read_text())
+    assert "memory_mode" not in saved and "evolve_schedule" not in saved
+
+
+def test_acquire_client_builds_the_sdk_client_in_the_given_mode(tmp_path, monkeypatch) -> None:
+    built: list[str] = []
+
+    class Client:
+        def __init__(self, *, config, mode, runtime, embedding_service):
+            built.append(mode)
+
+        def close(self):
+            return None
+
+    monkeypatch.setattr(hy_sdk_runtime, "HyMemoryClient", Client)
+    monkeypatch.setattr(hy_sdk_runtime, "_acquire_runtime", lambda config, key: object())
+    monkeypatch.setattr(hy_sdk_runtime, "_release_runtime", lambda key: None)
+    monkeypatch.setattr(hy_sdk_runtime, "inspect_graph_width_mismatch", lambda config: None)
+    monkeypatch.delenv("ONE_MEMORY_GRAPH_STORE", raising=False)
+    profile = hy_sdk_runtime.builtin_retrieval_snapshot()
+    for mode in ("lite", "pro", "ultra"):
+        hy_sdk_runtime.acquire_client(profile, api_key=None, data_dir=tmp_path, mode=mode).close()
+    hy_sdk_runtime.acquire_client(profile, api_key=None, data_dir=tmp_path).close()
+    monkeypatch.setenv("ONE_MEMORY_GRAPH_STORE", "kuzu")
+    hy_sdk_runtime.acquire_client(profile, api_key=None, data_dir=tmp_path).close()
+    assert built == ["lite", "pro", "ultra", "pro", "ultra"]
+
+
+def test_the_read_handle_rebuilds_on_a_mode_or_profile_change(monkeypatch, tmp_path) -> None:
+    """Reads follow the active profile's embedding (#255) and the memory mode."""
+    from one_memory import sdk_read_handle
+
+    modes = ["pro"]
+    made: list[_FakeHandle] = []
+
+    def handle_for(profile, *, api_key, data_dir, mode=None):
+        handle = _FakeHandle(mode or "pro", profile.profile_id)
+        made.append(handle)
+        return handle
+
+    monkeypatch.setattr(sdk_read_handle, "SdkClientHandle", handle_for)
+    profiles = _Profiles(identity=None)
+    reads = sdk_read_handle.ActiveProfileReadHandle(
+        profiles=profiles,  # type: ignore[arg-type]
+        secrets=None,  # type: ignore[arg-type]
+        data_dir=tmp_path,
+        memory_mode=lambda: modes[0],
+    )
+
+    async def read():
+        return await reads.client()
+
+    async def scenario():
+        first = await asyncio.create_task(read())
+        assert (first.mode, first.profile_id) == ("pro", "builtin-retrieval")
+        modes[0] = "lite"
+        # A read still in flight on the old client keeps it open.
+        hold = asyncio.Event()
+        holding_client = asyncio.Event()
+
+        async def holding():
+            await reads.client()
+            holding_client.set()
+            await hold.wait()
+
+        modes[0] = "pro"
+        holder = asyncio.create_task(holding())
+        await asyncio.wait_for(holding_client.wait(), 5)
+        modes[0] = "lite"
+        try:
+            assert (await asyncio.create_task(read())).mode == "lite"
+            assert not made[0].closed
+        finally:
+            hold.set()
+            await holder
+        for _ in range(10):
+            await asyncio.sleep(0)
+        assert made[0].closed
+        assert [item.mode for item in made] == ["pro", "lite"]
+        await reads.close()
+        assert made[1].closed
+
+    asyncio.run(scenario())
+
+
+class _FakeHandle:
+    def __init__(self, mode: str, profile_id: str = "p") -> None:
+        self.mode = mode
+        self.profile_id = profile_id
+        self.closed = False
+
+    async def client(self):
+        return self
+
+    async def close(self):
+        self.closed = True
+
+
+class _Profiles:
+    def __init__(self, identity=("profile-a", 1)) -> None:
+        self.identity = identity
+
+    def active_identity(self):
+        return self.identity
+
+    def snapshot(self, profile_id, revision):
+        return SimpleNamespace(profile_id=profile_id, has_api_key=False)
+
+    def resolve_snapshot_secret(self, profile_id, revision, secrets):
+        return None
+
+
+def _coordinator(monkeypatch, modes, profiles=None):
+    opened: list[_FakeHandle] = []
+
+    def handle(profile, **kwargs):
+        value = _FakeHandle(kwargs.get("mode") or "pro", profile.profile_id)
+        opened.append(value)
+        return value
+
+    monkeypatch.setattr(sdk_write_processor, "SdkClientHandle", handle)
+    coordinator = sdk_write_processor.SdkWriteCoordinator(
+        profiles=profiles or _Profiles(),  # type: ignore[arg-type]
+        secrets=None,  # type: ignore[arg-type]
+        data_dir=None,
+        transport_factory=lambda _profile: None,
+        sedimentation_enabled=lambda: True,
+        memory_mode=lambda: modes[0],
+    )
+    return coordinator, opened
+
+
+def test_a_mode_change_retires_idle_write_handles_and_keeps_a_busy_one(monkeypatch) -> None:
+    modes = ["pro"]
+    coordinator, opened = _coordinator(monkeypatch, modes)
+
+    async def scenario():
+        async with coordinator.use_client() as client:
+            assert client.mode == "pro"
+            modes[0] = "ultra"
+            await coordinator.retire_stale_modes()
+            # Still in use: closed only when this use ends (#218).
+            assert not opened[0].closed
+        assert opened[0].closed
+        async with coordinator.use_client() as client:
+            assert client.mode == "ultra"
+        modes[0] = "lite"
+        await coordinator.retire_stale_modes()
+        assert opened[1].closed
+        await coordinator.close()
+
+    asyncio.run(scenario())
+
+
+def test_lite_needs_no_generation_model(monkeypatch) -> None:
+    modes = ["lite"]
+    coordinator, opened = _coordinator(monkeypatch, modes, _Profiles(identity=None))
+
+    async def scenario():
+        async with coordinator.use_client() as client:
+            assert client.mode == "lite"
+            assert client.profile_id == hy_sdk_runtime.BUILTIN_RETRIEVAL_PROFILE_ID
+        modes[0] = "pro"
+        with pytest.raises(RuntimeError, match="model_profile_not_active"):
+            async with coordinator.use_client():
+                pass
+        await coordinator.close()
+
+    asyncio.run(scenario())
+
+
+def _record(text: str, *, extracted: bool, user_id: str = "local", role: str = "user"):
+    record = materialize_write_input(
+        EvidenceInput(
+            user_id=user_id,
+            source_id="one-memory-manual" if extracted else "hermes",
+            conversation_id="session-1",
+            message_id=str(uuid4()),
+            role=role,
+            written_by=WrittenBy.PERSON,
+            created_at=datetime(2026, 10, 1, tzinfo=UTC),
+            verbatim=text,
+            origin_meta={EvidenceOrigin.FILE: OriginMetadata()},
+        )
+    )
+    if extracted:
+        record = record.model_copy(update={"content_kind": EvidenceContentKind.MEMORY_CONTENT})
+    return record
+
+
+class _WritingSdk:
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, object]] = []
+
+    def add(self, messages, **kwargs):
+        self.calls.append(("add", messages))
+        return {"success": True, "raw_ids": ["raw-1"]}
+
+    def add_extracted(self, contents, **kwargs):
+        self.calls.append(("add_extracted", contents))
+        return {"success": True, "memory_ids": ["m-1"]}
+
+
+class _SdkHandle:
+    def __init__(self, sdk, mode="pro") -> None:
+        self.sdk = sdk
+        self.mode = mode
+
+    async def client(self):
+        return self.sdk
+
+    async def close(self):
+        return None
+
+
+async def _noop() -> None:
+    return None
+
+
+def test_a_note_in_lite_is_stored_as_raw_and_never_calls_add_extracted() -> None:
+    sdk = _WritingSdk()
+    note = (_record("I prefer green tea", extracted=True),)
+
+    async def scenario():
+        await sdk_write_processor.SdkWriteProcessor(_SdkHandle(sdk), lite=True).process_sdk_job(
+            note, request_id="r-lite", before_hy_invocation=_noop
+        )
+        await sdk_write_processor.SdkWriteProcessor(_SdkHandle(sdk)).process_sdk_job(
+            note, request_id="r-pro", before_hy_invocation=_noop
+        )
+
+    asyncio.run(scenario())
+    assert sdk.calls == [
+        ("add", [{"role": "user", "content": "I prefer green tea"}]),
+        ("add_extracted", ["I prefer green tea"]),
+    ]
+
+
+def test_a_lite_write_runs_through_the_coordinator_without_a_model(monkeypatch) -> None:
+    sdk = _WritingSdk()
+    coordinator = sdk_write_processor.SdkWriteCoordinator(
+        profiles=_Profiles(identity=None),  # type: ignore[arg-type]
+        secrets=None,  # type: ignore[arg-type]
+        data_dir=None,
+        transport_factory=lambda _profile: None,
+        sedimentation_enabled=lambda: True,
+        memory_mode=lambda: "lite",
+    )
+    monkeypatch.setattr(
+        sdk_write_processor, "SdkClientHandle", lambda profile, **kwargs: _SdkHandle(sdk, "lite")
+    )
+    turn = (
+        _record("hello", extracted=False),
+        _record("hi there", extracted=False, role="assistant"),
+    )
+    receipt = asyncio.run(
+        coordinator.process_sdk_job(turn, request_id="r-1", before_hy_invocation=_noop)
+    )
+    assert receipt.raw_ids == ("raw-1",)
+    assert [name for name, _ in sdk.calls] == ["add"]
+
+
+def test_no_raw_is_ever_deleted_in_any_mode() -> None:
+    """Owner decision: MemHub keeps every raw; there is no retention window."""
+    from one_memory import sdk_memory_runtime
+
+    for name in ("prune_expired_raw", "start_raw_retention", "_raw_retention_loop"):
+        assert not hasattr(SdkMemoryRuntime, name), name
+    assert not hasattr(sdk_memory_runtime, "RAW_RETENTION_DAYS")
+
+
+def test_mode_switch_effects() -> None:
+    assert memory_mode_effects("lite", "pro") == (
+        "earlier_raw_not_extracted",
+        "lite_search_stops",
+    )
+    assert memory_mode_effects("ultra", "pro") == ("higher_mode_memories_hidden", "evolve_stopped")
+    assert memory_mode_effects("pro", "ultra", graph_backfill=True) == (
+        "existing_memories_added_to_graph",
+    )
+    assert memory_mode_effects("pro", "ultra", graph_backfill=False) == (
+        "graph_backfill_already_running",
+    )
+    assert memory_mode_effects("pro", "lite") == ("higher_mode_memories_hidden",)
+    assert memory_mode_effects("pro", "pro") == ()
+
+
+# --- evolve ---------------------------------------------------------------------
+
+
+class _EvolveSdk:
+    """`evolve` answers `should_continue` for the first rounds, then stops."""
+
+    def __init__(self, continuing: int = 2, *, gate: asyncio.Event | None = None) -> None:
+        self.calls: list[dict] = []
+        self.continuing = continuing
+        self.gate = gate
+        self.graph_builds: list[str] = []
+        self.active = 0
+        self.peak = 0
+
+    def evolve(self, user_id, agent_id="default_agent", max_batches=None):
+        self.active += 1
+        self.peak = max(self.peak, self.active)
+        try:
+            self.calls.append(
+                {"user_id": user_id, "agent_id": agent_id, "max_batches": max_batches}
+            )
+            more = len([call for call in self.calls if call["user_id"] == user_id]) <= (
+                self.continuing
+            )
+            return {
+                "success": True,
+                "request_id": f"evolve-{len(self.calls)}",
+                "tasks_processed": 2,
+                "results": {"schemas": ["private text"]},
+                "cross_domain_sweeper": {"merged": 1},
+                "should_continue": more,
+                "continuation_reason": "remaining_clusters" if more else None,
+                "remaining_clusters": 3 if more else 0,
+                "remaining_actionable_clusters": 1 if more else 0,
+                "remaining_fresh_facts": 0,
+                "cycle_status": "partial" if more else "complete",
+                "batch_failures": [],
+                "elapsed_ms": 5.0,
+            }
+        finally:
+            self.active -= 1
+
+    def build_memory_graph(self, user_id, *, agent_id=None, limit=10000, clear_first=False):
+        self.graph_builds.append(user_id)
+        return {"success": True, "memories_scanned": 4, "nodes": 4}
+
+
+class _Coordinator:
+    """The real per-user exclusivity over a stand-in client."""
+
+    def __init__(self, sdk, monkeypatch) -> None:
+        self.inner, _ = _coordinator(monkeypatch, ["ultra"])
+        self.sdk = sdk
+
+    def exclusive_for_evolve(self, user_id):
+        return self.inner.exclusive_for_evolve(user_id)
+
+    @asynccontextmanager
+    async def use_client(self):
+        yield self.sdk
+
+
+def test_evolve_refuses_outside_ultra(tmp_path, monkeypatch) -> None:
+    service = EvolveService(
+        coordinator=_Coordinator(_EvolveSdk(), monkeypatch),
+        state=EvolveStateStore(tmp_path),
+        memory_mode=lambda: "pro",
+    )
+
+    async def scenario():
+        with pytest.raises(EvolveModeError):
+            service.start("local")
+
+    asyncio.run(scenario())
+
+
+def test_evolve_loops_while_should_continue_and_records_counts_only(tmp_path, monkeypatch) -> None:
+    sdk = _EvolveSdk(continuing=2)
+    state = EvolveStateStore(tmp_path)
+    service = EvolveService(
+        coordinator=_Coordinator(sdk, monkeypatch), state=state, memory_mode=lambda: "ultra"
+    )
+
+    async def scenario():
+        started = service.start("wife", max_batches=2)
+        assert started.state == "running"
+        done = await service.wait(started.run_id)
+        assert done is not None
+        return done
+
+    done = asyncio.run(scenario())
+    assert done.state == "succeeded" and done.rounds == 3 and done.tasks_processed == 6
+    assert done.cycle_status == "complete" and done.should_continue is False
+    assert {call["agent_id"] for call in sdk.calls} == {"all-apps"}
+    assert {call["max_batches"] for call in sdk.calls} == {2}
+    saved = (tmp_path / "evolve-state.json").read_text()
+    assert "private text" not in saved
+    assert json.loads(saved)["users"]["wife"]["last_result"]["rounds"] == 3
+
+
+def test_evolve_stops_at_the_round_cap(tmp_path, monkeypatch) -> None:
+    sdk = _EvolveSdk(continuing=100)
+    service = EvolveService(
+        coordinator=_Coordinator(sdk, monkeypatch),
+        state=EvolveStateStore(tmp_path),
+        memory_mode=lambda: "ultra",
+        max_rounds=4,
+    )
+
+    async def scenario():
+        return await service.wait(service.start("local").run_id)
+
+    done = asyncio.run(scenario())
+    assert done is not None and done.rounds == 4 and done.capped is True
+    assert done.state == "succeeded"
+
+
+def test_one_run_per_user_and_never_beside_that_users_write(tmp_path, monkeypatch) -> None:
+    sdk = _EvolveSdk(continuing=0)
+    coordinator = _Coordinator(sdk, monkeypatch)
+    service = EvolveService(
+        coordinator=coordinator, state=EvolveStateStore(tmp_path), memory_mode=lambda: "ultra"
+    )
+    writing = asyncio.Event()
+    release = asyncio.Event()
+
+    async def fake_write(self, records, *, request_id, before_hy_invocation):
+        writing.set()
+        await release.wait()
+        return sdk_write_processor.SdkWriteReceipt()
+
+    monkeypatch.setattr(sdk_write_processor.SdkWriteProcessor, "process_sdk_job", fake_write)
+    record = (_record("hello", extracted=False, user_id="wife"),)
+
+    async def scenario():
+        inner = coordinator.inner
+        write = asyncio.create_task(
+            inner.process_sdk_job(record, request_id="w", before_hy_invocation=_noop)
+        )
+        await writing.wait()
+        first = service.start("wife")
+        again = service.start("wife")
+        assert again.run_id == first.run_id
+        other = service.start("local")
+        await service.wait(other.run_id)
+        # The other user is not held; wife's run waits for wife's write.
+        assert sdk.calls and {call["user_id"] for call in sdk.calls} == {"local"}
+        assert inner.evolving("wife")
+        with pytest.raises(RuntimeError, match="memory_evolve_in_progress"):
+            await inner.process_sdk_job(record, request_id="w2", before_hy_invocation=_noop)
+        release.set()
+        await write
+        done = await service.wait(first.run_id)
+        assert done is not None and done.state == "succeeded"
+        assert not inner.evolving("wife")
+        assert sdk.peak == 1
+
+    asyncio.run(scenario())
+
+
+def test_a_failed_round_records_its_code(tmp_path, monkeypatch) -> None:
+    class Failing(_EvolveSdk):
+        def evolve(self, user_id, agent_id="default_agent", max_batches=None):
+            return {
+                "success": False,
+                "error_code": "schema_llm_failed",
+                "failure_kind": "llm_error",
+                "error_message": "the provider said something private",
+            }
+
+    state = EvolveStateStore(tmp_path)
+    service = EvolveService(
+        coordinator=_Coordinator(Failing(), monkeypatch), state=state, memory_mode=lambda: "ultra"
+    )
+    done = asyncio.run(service.wait(_start(service, "local")))
+    assert done is not None and done.state == "failed"
+    assert done.error_code == "schema_llm_failed" and done.failure_kind == "llm_error"
+    assert state.user("local").last_error == "schema_llm_failed"
+    assert "private" not in (tmp_path / "evolve-state.json").read_text()
+
+
+def _start(service: EvolveService, user_id: str) -> str:
+    async def begin() -> str:
+        return service.start(user_id).run_id
+
+    loop = asyncio.new_event_loop()
+    try:
+        run_id = loop.run_until_complete(begin())
+        loop.run_until_complete(service.wait(run_id))
+    finally:
+        loop.close()
+    return run_id
+
+
+# --- the nightly schedule -------------------------------------------------------
+
+LOCAL = datetime.now().astimezone().tzinfo
+
+
+def _at(day: int, hour: int, minute: int = 0) -> datetime:
+    return datetime(2026, 11, day, hour, minute, tzinfo=LOCAL)
+
+
+def test_evolve_is_due_after_the_slot_once_a_day_and_catches_up_after_sleep() -> None:
+    schedule = EvolveSchedule()
+    anchor = _at(1, 12)
+    # Never ran: not at noon on the day ultra was enabled, then at 03:00 next day.
+    assert not evolve_due(_at(1, 13), schedule, None, anchor)
+    assert not evolve_due(_at(2, 2, 59), schedule, None, anchor)
+    assert evolve_due(_at(2, 3, 0), schedule, None, anchor)
+    # Ran at 03:00: not again that day, due again the next night.
+    assert not evolve_due(_at(2, 22), schedule, _at(2, 3, 1), anchor)
+    assert evolve_due(_at(3, 3, 5), schedule, _at(2, 3, 1), anchor)
+    # Asleep at 03:00: a run older than the slot is due on wake, the same day.
+    assert evolve_due(_at(3, 9), schedule, _at(2, 3, 1), anchor)
+    # Last run more than 24h ago is due even before today's slot.
+    assert evolve_due(_at(4, 1), schedule, _at(2, 3, 1), anchor)
+    assert not evolve_due(_at(3, 1), schedule, _at(2, 3, 1), anchor)
+    # Moved and turned off.
+    assert evolve_due(_at(2, 23, 30), EvolveSchedule(time="23:30"), _at(2, 3), anchor)
+    assert not evolve_due(_at(3, 9), EvolveSchedule(enabled=False), _at(1, 3), anchor)
+
+
+def test_the_scheduler_runs_each_due_user_once_in_ultra_only(tmp_path, monkeypatch) -> None:
+    sdk = _EvolveSdk(continuing=0)
+    state = EvolveStateStore(tmp_path)
+    modes = ["pro"]
+    clock = [_at(1, 12)]
+    service = EvolveService(
+        coordinator=_Coordinator(sdk, monkeypatch),
+        state=state,
+        memory_mode=lambda: modes[0],
+        now=lambda: clock[0],
+    )
+    schedule = [EvolveSchedule()]
+
+    async def users():
+        return ("local", "wife")
+
+    scheduler = EvolveScheduler(
+        service=service,
+        schedule=lambda: schedule[0],
+        users_with_memories=users,
+        now=lambda: clock[0],
+    )
+
+    async def scenario():
+        assert await scheduler.tick() == ()  # pro: nothing
+        modes[0] = "ultra"
+        assert await scheduler.tick() == ()  # anchored at noon
+        clock[0] = _at(2, 3, 2)
+        assert await scheduler.tick() == ("local", "wife")
+        assert await scheduler.tick() == ()  # once a night
+        clock[0] = _at(3, 8)  # slept through 03:00
+        assert await scheduler.tick() == ("local", "wife")
+        schedule[0] = EvolveSchedule(enabled=False)
+        clock[0] = _at(5, 8)
+        assert await scheduler.tick() == ()
+
+    asyncio.run(scenario())
+    assert [call["user_id"] for call in sdk.calls] == ["local", "wife", "local", "wife"]
+    assert state.user("wife").last_result["trigger"] == "scheduled"
+
+
+def test_the_graph_backfill_runs_once_per_user(tmp_path, monkeypatch) -> None:
+    sdk = _EvolveSdk()
+    state = EvolveStateStore(tmp_path)
+    backfill = MemoryGraphBackfill(coordinator=_Coordinator(sdk, monkeypatch), state=state)
+
+    async def scenario():
+        backfill.start(["local", "wife", "local"])
+        await backfill.wait()
+
+    asyncio.run(scenario())
+    assert sdk.graph_builds == ["local", "wife"]
+    built = state.graph_build()
+    assert built.state == "succeeded"
+    assert {user: value.state for user, value in built.users.items()} == {
+        "local": "succeeded",
+        "wife": "succeeded",
+    }
+
+
+# --- the HTTP surface -----------------------------------------------------------
+
+
+def application(tmp_path: Path):
+    settings = Settings(
+        data_dir=tmp_path / "data",
+        source_home=tmp_path,
+        api_token=SecretStr(TOKEN),
+        imports_enabled=False,
+        automatic_memory_processing=False,
+        live_sync_enabled=False,
+    )
+    return create_app(settings, secret_store=InMemorySecretStore())
+
+
+def client_for(app) -> AsyncClient:
+    return AsyncClient(
+        transport=ASGITransport(app=app),
+        base_url="http://test",
+        headers={"Authorization": f"Bearer {TOKEN}"},
+    )
+
+
+class _ReadSdk(_EvolveSdk):
+    def memory_stats(self, *, user_id, statuses=None):
+        return {"total": 1, "scenes": {}}
+
+
+def run_app(tmp_path, monkeypatch, scenario) -> None:
+    monkeypatch.delenv("ONE_MEMORY_GRAPH_STORE", raising=False)
+
+    async def main() -> None:
+        app = application(tmp_path)
+        sdk = _ReadSdk(continuing=1)
+
+        @asynccontextmanager
+        async def write_client():
+            yield sdk
+
+        async with app.router.lifespan_context(app):
+            monkeypatch.setattr(app.state.sdk_write_coordinator, "use_client", write_client)
+            async with client_for(app) as client:
+                await scenario(app, sdk, client)
+
+    asyncio.run(main())
+
+
+@pytest.mark.fresh_sdk
+def test_mode_api_get_put_and_its_effects(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        current = (await client.get("/api/v1/settings/memory-mode")).json()
+        assert current["mode"] == current["configured_mode"] == "pro"
+        assert current["forced_by_env"] is False
+        assert app.state.sdk_write_coordinator.memory_mode() == "pro"
+
+        lite = await client.put("/api/v1/settings/memory-mode", json={"mode": "lite"})
+        assert lite.status_code == 200, lite.text
+        body = lite.json()
+        assert body["mode"] == "lite" and body["previous_mode"] == "pro"
+        assert body["effects"] == ["higher_mode_memories_hidden"]
+        assert app.state.sdk_write_coordinator.memory_mode() == "lite"
+        assert _sdk_client_handle(app).memory_mode() == "lite"
+        # lite claims queued work without a generation model.
+        from one_memory.api import _sedimentation_claim_enabled
+
+        assert app.state.model_profile_store.active_identity() is None
+        app.state.history_import_gate_released = True
+        assert _sedimentation_claim_enabled(app, "one-memory-manual")
+
+        ultra = await client.put("/api/v1/settings/memory-mode", json={"mode": "ultra"})
+        assert ultra.status_code == 200, ultra.text
+        assert "existing_memories_added_to_graph" in ultra.json()["effects"]
+        await app.state.memory_graph_backfill.wait()
+        assert sdk.graph_builds == ["local"]
+        reread = (await client.get("/api/v1/settings/memory-mode")).json()
+        assert reread["graph_build"]["state"] == "succeeded"
+
+        for bad in ({"mode": "max"}, {"mode": "pro", "x": 1}):
+            assert (await client.put("/api/v1/settings/memory-mode", json=bad)).status_code == 422
+        assert (
+            await client.get("/api/v1/settings/memory-mode", params={"user_id": "x"})
+        ).status_code == 422
+
+        monkeypatch.setenv("ONE_MEMORY_GRAPH_STORE", "kuzu")
+        await client.put("/api/v1/settings/memory-mode", json={"mode": "pro"})
+        forced = (await client.get("/api/v1/settings/memory-mode")).json()
+        assert forced["mode"] == "ultra" and forced["configured_mode"] == "pro"
+        assert forced["forced_by_env"] is True
+
+    run_app(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_evolve_api_is_ultra_only_async_and_scoped_per_user(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        refused = await client.post("/api/v2/evolve", json={})
+        assert refused.status_code == 409
+        assert refused.json()["detail"] == "memory_mode_not_ultra"
+
+        await client.put("/api/v1/settings/memory-mode", json={"mode": "ultra"})
+        await app.state.memory_graph_backfill.wait()
+        started = await client.post(
+            "/api/v2/evolve", json={"user_id": "wife", "agent_id": "all-apps", "max_batches": 2}
+        )
+        assert started.status_code == 202, started.text
+        run = started.json()
+        assert run["state"] == "running" and run["user_id"] == "wife"
+        await app.state.evolve_service.wait(run["run_id"])
+        path = f"/api/v2/evolve/{run['run_id']}"
+        assert (await client.get(path)).status_code == 404
+        done = (await client.get(path, params={"user_id": "wife"})).json()
+        assert done["state"] == "succeeded" and done["rounds"] == 2
+        assert sdk.calls[0] == {"user_id": "wife", "agent_id": "all-apps", "max_batches": 2}
+
+        overview = (await client.get("/api/v2/evolve", params={"user_id": "wife"})).json()
+        assert set(overview["users"]) == {"wife"}
+        assert overview["schedule"] == {"enabled": True, "time": "03:00"}
+
+        processing = (await client.get("/api/v1/memory-processing")).json()
+        pipeline = processing["pipeline"]
+        assert pipeline["mode"] == "ultra"
+        assert pipeline["last_evolve_at"] is not None
+        assert pipeline["evolve"]["users"]["wife"]["last_result"]["state"] == "succeeded"
+
+        for bad in (
+            {"user_id": "a:b"},
+            {"max_batches": 0},
+            {"from_beginning": True},
+        ):
+            assert (await client.post("/api/v2/evolve", json=bad)).status_code == 422
+
+        moved = await client.put(
+            "/api/v1/settings/evolve-schedule", json={"time": "04:30", "enabled": False}
+        )
+        assert moved.json() == {"enabled": False, "time": "04:30"}
+        assert (
+            await client.put("/api/v1/settings/evolve-schedule", json={"time": "25:00"})
+        ).status_code == 422
+
+    run_app(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_a_job_written_in_lite_stays_done_and_a_pending_job_takes_the_new_mode(
+    tmp_path, monkeypatch
+) -> None:
+    """lite to pro reprocesses nothing: the mode is read when a job runs."""
+    from one_memory.memory_server.durable_queue import DurableWriteDrain, execute_job
+
+    modes = ["lite"]
+    seen: list[str] = []
+
+    class Sdk:
+        def __init__(self, mode):
+            self.mode = mode
+
+        def add(self, messages, **kwargs):
+            seen.append(self.mode)
+            return {"success": True, "raw_ids": [f"raw-{len(seen)}"]}
+
+    monkeypatch.setattr(
+        sdk_write_processor,
+        "SdkClientHandle",
+        lambda profile, **kwargs: _SdkHandle(Sdk(kwargs["mode"]), kwargs["mode"]),
+    )
+    coordinator = sdk_write_processor.SdkWriteCoordinator(
+        profiles=_Profiles(),  # type: ignore[arg-type]
+        secrets=None,  # type: ignore[arg-type]
+        data_dir=None,
+        transport_factory=lambda _profile: None,
+        sedimentation_enabled=lambda: True,
+        memory_mode=lambda: modes[0],
+    )
+
+    async def never(_job_id):
+        return None
+
+    async def drain_once(queue) -> int:
+        drain = DurableWriteDrain(queue, execute=never, lease=timedelta(seconds=30))
+        claimed = drain.claim(limit=8)
+        for job_id in claimed:
+            await execute_job(
+                job_id,
+                queue=queue,
+                supervisor=coordinator,
+                evidence=None,  # type: ignore[arg-type]
+            )
+        return len(claimed)
+
+    async def scenario(app, _sdk, client):
+        queue = app.state.memory_write_queue
+        for request_id in ("turn-lite", "turn-pro"):
+            written = await client.post(
+                "/api/v2/memory-writes",
+                json={
+                    "request_id": request_id,
+                    "session_id": f"s-{request_id}",
+                    "source_id": "one-memory-manual",
+                    "messages": [
+                        {"role": "user", "content": request_id},
+                        {"role": "assistant", "content": "ok"},
+                    ],
+                },
+            )
+            assert written.status_code == 202, written.text
+            if request_id == "turn-lite":
+                assert await drain_once(queue) == 1
+                modes[0] = "pro"
+                await coordinator.retire_stale_modes()
+        assert await drain_once(queue) == 1
+        assert await drain_once(queue) == 0
+        assert queue.get("turn-lite").state == queue.get("turn-pro").state == "completed"
+        assert seen == ["lite", "pro"]
+        await coordinator.close()
+
+    run_app(tmp_path, monkeypatch, scenario)

memhub/apps/core/tests/test_memory_qa.py

@@ -66,6 +66,7 @@ def _grounded_setup(tmp_path, *, memory_id: str = "memory-real-data"):
             base_url="http://127.0.0.1:11434",
             model="test-model",
             embedding_model="test-embedding",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -263,6 +264,7 @@ def test_answer_uses_only_grounded_confirmed_memory_citations(tmp_path) -> None:
             base_url="http://127.0.0.1:11434",
             model="test-model",
             embedding_model="test-embedding",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -468,6 +470,7 @@ def test_answer_rejects_model_citation_outside_retrieved_scope(tmp_path) -> None
             base_url="http://127.0.0.1:11434",
             model="test-model",
             embedding_model="test-embedding",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -530,6 +533,7 @@ def _compatible_gateway_setup(tmp_path):
             base_url="https://api.example.invalid/standard/v1",
             model="deepseek-v4-pro",
             embedding_model="text-embedding",
+            embedding_base_url="https://api.example.invalid/standard/v1",
             api_key="probe-key",
         ),
         secrets,

memhub/apps/core/tests/test_memory_qa_permissions.py

@@ -186,6 +186,7 @@ def _setup(tmp_path: Path, *, model: str = "available") -> _Setup:
             base_url="https://memory-qa.example.invalid/v1",
             model="synthetic-generation",
             embedding_model="synthetic-embedding",
+            embedding_base_url="https://memory-qa.example.invalid/v1",
             api_key=SecretStr(_SECRET),
         ),
         secrets,

memhub/apps/core/tests/test_memory_task_unknown_retry.py

@@ -25,6 +25,7 @@ def _stores(tmp_path):
             provider_kind=ProviderKind.OLLAMA,
             model="hy-bot-phi4",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )

memhub/apps/core/tests/test_memory_tasks.py

@@ -83,6 +83,7 @@ def test_task_freezes_active_profile_revision_and_is_idempotent(tmp_path) -> Non
             provider_kind=ProviderKind.OLLAMA,
             model="qwen",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -240,6 +241,7 @@ def test_task_extraction_checkpoint_is_idempotent_and_recoverable(tmp_path) -> N
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -293,6 +295,7 @@ def test_task_waits_for_model_and_retries_with_same_snapshot(tmp_path) -> None:
             provider_kind=ProviderKind.OLLAMA,
             model="qwen",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -337,6 +340,7 @@ def test_blocked_stages_name_the_stuck_stage_and_reason(tmp_path) -> None:
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -433,6 +437,7 @@ def test_task_duration_estimate_ignores_time_the_app_was_closed(tmp_path) -> Non
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -537,6 +542,7 @@ def test_running_task_is_recovered_after_worker_restart(tmp_path) -> None:
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -572,6 +578,7 @@ def test_superseded_profile_cancels_only_claimable_tasks(tmp_path) -> None:
                 provider_kind=ProviderKind.OLLAMA,
                 model="phi",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -601,6 +608,7 @@ def test_superseded_profile_cancels_only_claimable_tasks(tmp_path) -> None:
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -630,6 +638,7 @@ def test_superseded_ingest_policy_cancels_only_old_conversation_tasks(tmp_path)
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -676,6 +685,7 @@ def test_task_rejects_inactive_profile_and_request_id_reuse(tmp_path) -> None:
             provider_kind=ProviderKind.OLLAMA,
             model="qwen",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -1700,6 +1710,7 @@ def _profile_and_tasks(tmp_path) -> tuple[MemoryTaskStore, ModelProfile]:
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),
@@ -2058,6 +2069,7 @@ def test_a_failed_task_below_the_ceiling_is_still_claimable(tmp_path) -> None:
                 provider_kind=ProviderKind.OLLAMA,
                 model="qwen",
                 embedding_model="nomic-embed-text",
+                embedding_base_url="http://127.0.0.1:11434",
             ),
             secrets,
         ),

memhub/apps/core/tests/test_memory_visibility.py

@@ -55,6 +55,7 @@ def library(tmp_path):
             provider_kind=ProviderKind.OLLAMA,
             model="hy-bot-phi4",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -611,3 +612,58 @@ def test_granting_a_source_repeatedly_does_not_raise(library):
 
     with main_database_read(database) as db:
         assert memory.memory_id in pending(db)
+
+
+def test_grant_writes_still_work_after_the_cutover_drops_the_link_table(library):
+    """#233: the grant triggers read `governed_memory_evidence`.
+
+    The SDK cutover drops it, and SQLite resolves a trigger body only when it
+    fires, so every grant, revoke and connect then failed with "no such
+    table". A trigger whose body names a missing table is not installed, and
+    the copy an earlier startup installed is dropped.
+    """
+    database, *_ = library
+    from one_memory.imports.candidates import RootCandidateKind
+    from one_memory.imports.grants import SourceGrantStore
+    from one_memory.memory_visibility import drop_stale_triggers
+
+    startup(database)
+    with sqlite3.connect(database) as db:
+        db.execute("DROP TABLE governed_memory_evidence")
+
+    def grant_triggers() -> set[str]:
+        with sqlite3.connect(database) as db:
+            return {
+                str(row[0])
+                for row in db.execute(
+                    "SELECT name FROM sqlite_master "
+                    "WHERE type='trigger' AND tbl_name='source_grants'"
+                )
+            }
+
+    assert grant_triggers() == {
+        "memory_visibility_dirty_on_grants",
+        "memory_visibility_dirty_on_grants_update",
+    }
+    # The `evidence` triggers read the dropped table too; the cutover drops
+    # `evidence` as well, which takes them with it.
+    assert {
+        "memory_visibility_dirty_on_grants",
+        "memory_visibility_dirty_on_grants_update",
+    } <= set(drop_stale_triggers(database))
+    assert grant_triggers() == set()
+    assert drop_stale_triggers(database) == ()
+    # Installing again on this shape skips them instead of bringing them back.
+    startup(database)
+    assert grant_triggers() == set()
+
+    folder = database.parent / "cursor"
+    folder.mkdir(exist_ok=True)
+    store = SourceGrantStore(database, home=database.parent)
+    store.grant(
+        user_id="local",
+        source_id="cursor",
+        root=folder,
+        candidate_kind=RootCandidateKind.DEFAULT,
+    )
+    assert store.revoke(user_id="local", source_id="cursor").revoked_at is not None

memhub/apps/core/tests/test_memory_visibility_equivalence.py

@@ -51,6 +51,7 @@ def library(tmp_path):
             provider_kind=ProviderKind.OLLAMA,
             model="hy-bot-phi4",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )

memhub/apps/core/tests/test_model_probe.py

@@ -28,6 +28,7 @@ def test_probe_still_measures_embedding_when_structured_output_fails(tmp_path) -
             base_url="http://127.0.0.1:11434",
             model="plain-model",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -63,6 +64,7 @@ def test_probe_records_the_real_embedding_width_on_the_profile(tmp_path) -> None
             base_url="http://127.0.0.1:11434",
             model="qwen3:8b",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -119,6 +121,7 @@ def test_openai_compatible_probe_checks_model_and_structured_output(tmp_path) ->
             base_url="https://models.example.com/v1",
             model="memory-model",
             embedding_model="embedding-model",
+            embedding_base_url="https://models.example.com/v1",
             api_key=SecretStr(API_KEY),
         ),
         secrets,
@@ -260,6 +263,7 @@ def test_probe_reports_missing_model_without_sending_generation(tmp_path) -> Non
             provider_kind=ProviderKind.OLLAMA,
             model="memory-model",
             embedding_model="embedding-model",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )

memhub/apps/core/tests/test_model_profile_api.py

@@ -50,6 +50,7 @@ def test_model_profile_api_never_returns_key_and_enforces_revision(tmp_path) ->
                     "base_url": "https://models.example.com/v1",
                     "model": "memory-model",
                     "embedding_model": "embedding-model",
+                    "embedding_base_url": "https://models.example.com/v1",
                     "api_key": API_KEY,
                 },
             )
@@ -134,6 +135,7 @@ def test_model_profile_probe_route_returns_only_machine_results(tmp_path) -> Non
                     "base_url": "https://models.example.com/v1",
                     "model": "memory-model",
                     "embedding_model": "embedding-model",
+                    "embedding_base_url": "https://models.example.com/v1",
                     "api_key": API_KEY,
                 },
             )
@@ -244,6 +246,7 @@ def test_activation_survives_a_corpus_already_imported(tmp_path) -> None:
                     "base_url": "https://models.example.com/v1",
                     "model": "memory-model",
                     "embedding_model": "embedding-model",
+                    "embedding_base_url": "https://models.example.com/v1",
                     "api_key": API_KEY,
                 },
             )

memhub/apps/core/tests/test_model_profiles.py

@@ -198,6 +198,7 @@ def test_active_profile_can_be_deleted_without_changing_other_profiles(tmp_path)
             provider_kind=ProviderKind.OLLAMA,
             model="other",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -208,6 +209,7 @@ def test_active_profile_can_be_deleted_without_changing_other_profiles(tmp_path)
             base_url="https://models.example.com/v1",
             model="qwen",
             embedding_model="nomic-embed-text",
+            embedding_base_url="https://models.example.com/v1",
             api_key=SecretStr(API_KEY),
         ),
         secrets,
@@ -349,6 +351,7 @@ def _probed(store, secrets, *, structured: bool, generates: bool = True):
             provider_kind=ProviderKind.OLLAMA,
             model="qwen",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )
@@ -394,3 +397,102 @@ def test_a_model_that_cannot_generate_still_cannot_be_turned_on(tmp_path) -> Non
 
     with pytest.raises(ModelProfileUnavailableError):
         store.activate(profile.profile_id, secrets)
+
+
+def _space(store: ModelProfileStore, profile_id: str) -> str:
+    """The embedding space's collection base, as the Runtime names it."""
+    from one_memory.hy_sdk_runtime import vector_collection_bases
+
+    identity = (profile_id, store.require(profile_id, InMemorySecretStore()).revision)
+    return vector_collection_bases(store.snapshot(*identity))[0]
+
+
+def test_a_remote_embedding_must_name_its_endpoint(tmp_path) -> None:
+    """The endpoint no longer follows the generation `base_url`."""
+    from one_memory.model_profiles import EmbeddingBaseUrlRequiredError
+
+    store = ModelProfileStore(tmp_path / "one-memory.sqlite3")
+    secrets = InMemorySecretStore()
+    unnamed = ModelProfileInput(
+        name="Remote",
+        provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+        base_url="https://models.example.com/v1",
+        model="qwen",
+        embedding_model="text-embedding-3-small",
+        api_key=SecretStr(API_KEY),
+    )
+    with pytest.raises(EmbeddingBaseUrlRequiredError):
+        store.create(unnamed, secrets)
+    assert store.list(secrets) == []
+
+    named = store.create(
+        unnamed.model_copy(update={"embedding_base_url": "https://models.example.com/v1"}),
+        secrets,
+    )
+    with pytest.raises(EmbeddingBaseUrlRequiredError):
+        store.update(
+            named.profile_id,
+            ModelProfilePatch(
+                expected_revision=named.revision,
+                embedding_provider_kind=ProviderKind.OLLAMA,
+                embedding_base_url="",
+            ),
+            secrets,
+        )
+
+
+def test_stored_profiles_get_the_endpoint_they_resolved_to_and_keep_their_space(
+    tmp_path,
+) -> None:
+    """An older build stored remote embeddings with a NULL endpoint and resolved
+    it to the generation `base_url` at runtime. Opening the store writes that
+    endpoint in once, so the collection, and the memories in it, stay put; a
+    later edit of the generation endpoint no longer moves them.
+    """
+    from one_memory.hy_sdk_runtime import vector_collection_bases
+
+    database = tmp_path / "one-memory.sqlite3"
+    store = ModelProfileStore(database)
+    secrets = InMemorySecretStore()
+    profile = store.create(
+        ModelProfileInput(
+            name="Remote",
+            provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            base_url="https://gateway.example.com/v1",
+            model="qwen",
+            embedding_model="text-embedding-3-small",
+            embedding_base_url="https://gateway.example.com/v1",
+            embedding_dimensions=1536,
+            api_key=SecretStr(API_KEY),
+        ),
+        secrets,
+    )
+    identity = (profile.profile_id, profile.revision)
+    # The space the older build's Runtime resolved: `embedding_base_url or base_url`.
+    before = vector_collection_bases(store.snapshot(*identity))[0]
+    with sqlite3.connect(database) as db:
+        # The shape an older build stored.
+        db.execute("UPDATE model_profiles SET embedding_base_url=NULL")
+        db.execute("UPDATE model_profile_versions SET embedding_base_url=NULL")
+
+    migrated = ModelProfileStore(database)
+    assert migrated.snapshot(*identity).embedding_base_url == "https://gateway.example.com/v1"
+    assert _space(migrated, profile.profile_id) == before
+    # Idempotent: opening again changes nothing.
+    assert _space(ModelProfileStore(database), profile.profile_id) == before
+    with sqlite3.connect(database) as db:
+        assert db.execute(
+            "SELECT COUNT(*) FROM model_profile_versions WHERE embedding_base_url IS NULL"
+        ).fetchone() == (0,)
+
+    edited = migrated.update(
+        profile.profile_id,
+        ModelProfilePatch(
+            expected_revision=profile.revision,
+            base_url="https://gateway.example.com/eu/v1",
+        ),
+        secrets,
+    )
+    assert edited.base_url == "https://gateway.example.com/eu/v1"
+    assert edited.embedding_base_url == "https://gateway.example.com/v1"
+    assert _space(migrated, profile.profile_id) == before

memhub/apps/core/tests/test_multi_user_runtime.py

@@ -650,6 +650,75 @@ def test_two_hermes_grants_each_read_their_own_root_into_separate_batches(tmp_pa
     asyncio.run(scenario())
 
 
+@pytest.mark.fresh_sdk
+def test_a_migrated_home_with_the_old_grant_triggers_can_connect_and_revoke(tmp_path) -> None:
+    """#233: a home an earlier build migrated kept two `source_grants` triggers
+    whose body reads the dropped `governed_memory_evidence`. Every grant write
+    then failed, so connecting and revoking answered 503. Startup drops them.
+    """
+    stale = """
+        CREATE TRIGGER memory_visibility_dirty_on_grants{suffix}
+        AFTER {event} ON source_grants
+        BEGIN
+            INSERT INTO memory_visibility_dirty (memory_id)
+            SELECT DISTINCT link.memory_id
+            FROM governed_memory_evidence link
+            JOIN evidence e ON e.evidence_id = link.evidence_id
+            WHERE e.source_id = NEW.source_id
+            ON CONFLICT(memory_id) DO NOTHING;
+        END;"""
+
+    async def scenario():
+        home = tmp_path / "home"
+        _hermes, wife = _hermes_homes(home)
+        app = application(tmp_path, home=home)
+        async with app.router.lifespan_context(app):
+            database = app.state.settings.data_dir / "one-memory.sqlite3"
+        with sqlite3.connect(database) as db:
+            db.execute("CREATE TABLE IF NOT EXISTS memory_visibility_dirty (memory_id TEXT)")
+            db.execute(stale.format(suffix="", event="INSERT"))
+            db.execute(stale.format(suffix="_update", event="UPDATE"))
+
+        app = application(tmp_path, home=home)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            location = (await client.post("/api/v1/sources/hermes/location")).json()
+            local = await client.post(
+                "/api/v1/sources/connect",
+                json={
+                    "source_ids": ["hermes"],
+                    "request_id": str(uuid4()),
+                    "source_roots": {"hermes": location["root_id"]},
+                },
+            )
+            assert local.status_code == 200, local.text
+            profile = await client.post(
+                "/api/v1/sources/connect",
+                json={
+                    "source_ids": ["hermes"],
+                    "request_id": str(uuid4()),
+                    "user_id": "wife",
+                    "root": str(wife),
+                },
+            )
+            assert profile.status_code == 200, profile.text
+            for user_id in ("wife", "local"):
+                revoked = await client.post(
+                    "/api/v1/sources/hermes/revoke", params={"user_id": user_id}
+                )
+                assert revoked.status_code == 200, revoked.text
+                assert revoked.json()["revoked_at"] is not None
+        with sqlite3.connect(database) as db:
+            assert (
+                db.execute(
+                    "SELECT COUNT(*) FROM sqlite_master "
+                    "WHERE type='trigger' AND tbl_name='source_grants'"
+                ).fetchone()[0]
+                == 0
+            )
+
+    asyncio.run(scenario())
+
+
 @pytest.mark.fresh_sdk
 def test_only_hermes_takes_another_user_or_a_custom_root(tmp_path) -> None:
     async def scenario():

memhub/apps/core/tests/test_read_path_freshness_batch.py

@@ -40,6 +40,7 @@ def library(tmp_path):
             provider_kind=ProviderKind.OLLAMA,
             model="hy-bot-phi4",
             embedding_model="nomic-embed-text",
+            embedding_base_url="http://127.0.0.1:11434",
         ),
         secrets,
     )

memhub/apps/core/tests/test_sdk_api_surface.py

@@ -0,0 +1,744 @@
+"""The SDK surface MemHub exposes: search pass-through, layers, raw, v2 aliases.
+
+P2-1 (R8): `context/search`, `recall` and the v2 ask routes pass the public
+`client.search` knobs through, validated, and keep a caller's floor on what
+they return. P2-2 (R9): the library honours `layers`, `statuses`, `scene` and
+the source/session/agent filters, and counts memories per layer. P2-4: raw
+lists the SDK's L1 raw turns, active and shadow. P2-3: v2 aliases of the
+source and import routes, per user. Profile versions read the L0 chain.
+
+Every route stays scoped to its one `user_id`; the stand-in client records
+each call so a test can see exactly what reached the SDK.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import sqlite3
+from contextlib import asynccontextmanager
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from pydantic import SecretStr
+
+from one_memory.api import create_app
+from one_memory.config import Settings
+from one_memory.imports.sources.registry import build_source_catalog
+from one_memory.secrets import InMemorySecretStore
+
+TOKEN = "sdk-surface-test"
+
+CHAT_ROWS = {
+    "local": [
+        ("m-high", "local likes green tea", 0.95, "l2_fact", "hermes"),
+        ("m-mid", "local walks at dawn", 0.6, "l2_fact", "codex"),
+    ],
+    "wife": [("w-high", "wife likes coffee", 0.92, "l2_fact", "hermes")],
+}
+RAW_ROWS = {
+    "local": [
+        {
+            "memory_id": "raw-1",
+            "content": "[user]: 我叫费天翔\n[assistant]: 好的，记住了。",
+            "status": "shadow",
+            "session_id": "s-1",
+            "source": "hermes",
+            "memory_at": 1_790_000_000,
+            "gmt_created": 1_790_000_100,
+        },
+        {
+            "memory_id": "raw-2",
+            "content": "[user]: 你好",
+            "status": "active",
+            "session_id": "s-2",
+            "source": "codex",
+            "memory_at": 1_790_100_000,
+            "gmt_created": 1_790_100_100,
+        },
+    ],
+    "wife": [
+        {
+            "memory_id": "raw-w",
+            "content": "[user]: I am the wife",
+            "status": "active",
+            "session_id": "s-w",
+            "source": "hermes",
+            "memory_at": 1_790_000_000,
+            "gmt_created": 1_790_000_000,
+        }
+    ],
+}
+
+
+def _chat(user_id: str, memory_id: str, content: str, layer: str, source: str) -> dict:
+    return {
+        "memory_id": memory_id,
+        "content": content,
+        "layer": layer,
+        "status": "active",
+        "user_id": user_id,
+        "agent_id": "all-apps",
+        "session_id": "s-1",
+        "source": source,
+        "tags": [],
+        "gmt_created": 1_790_000_000,
+    }
+
+
+class RecordingSdk:
+    """The public client surface, recording every call's keyword arguments."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, dict]] = []
+
+    def _record(self, name: str, kwargs: dict) -> None:
+        self.calls.append((name, dict(kwargs)))
+
+    def last(self, name: str) -> dict:
+        return [kwargs for called, kwargs in self.calls if called == name][-1]
+
+    def all(self, name: str) -> list[dict]:
+        return [kwargs for called, kwargs in self.calls if called == name]
+
+    # The SDK answers rows below a caller's floor in some lanes (a schema
+    # floor, a fused rank score), so Core must filter what it returns.
+    def search(self, query, **kwargs):
+        self._record("search", {"query": query, **kwargs})
+        [user_id] = kwargs["user_ids"]
+        if kwargs.get("layers") == ["l1_raw"]:
+            hits = [
+                {**row, "user_id": user_id, "agent_id": "all-apps", "score": score}
+                for row, score in zip(RAW_ROWS.get(user_id, []), (0.97, 0.3), strict=False)
+            ]
+            return {"request_id": "s", "memories": hits}
+        rows = [
+            {**_chat(user_id, mid, text, layer, source), "score": score}
+            for mid, text, score, layer, source in CHAT_ROWS.get(user_id, [])
+        ]
+        scene = kwargs.get("scene")
+        if scene == "all":
+            fused = [
+                {**row, "scene": "chat", "kind": "chat", "lane_score": row["score"], "score": 0.016}
+                for row in rows
+            ]
+            return {"request_id": "s", "scene": "all", "memories": fused, "lanes": {}, "errors": {}}
+        if scene in {"working", "coding"}:
+            return {
+                "request_id": "s",
+                "scene": "productivity",
+                "variant": scene,
+                "facts": [
+                    {"fact_id": "f-1", "statement": "use uv", "score": 0.91, "user_id": user_id}
+                ],
+                "experiences": [],
+            }
+        return {
+            "request_id": "s",
+            "scene": "normal",
+            "memories": {"normal": rows, "schema": [], "proactive": []},
+        }
+
+    def get(self, memory_id, kind=None, variant=None):
+        self._record("get", {"memory_id": memory_id, "kind": kind, "variant": variant})
+        if kind == "fact":
+            return {
+                "kind": "fact",
+                "variant": variant,
+                "fact_id": memory_id,
+                "statement": "use uv",
+                "status": "active",
+                "user_id": "local",
+            }
+        for user_id, rows in CHAT_ROWS.items():
+            for mid, text, _score, layer, source in rows:
+                if mid == memory_id:
+                    return {**_chat(user_id, mid, text, layer, source), "kind": "chat"}
+        return None
+
+    def list_memories(self, **kwargs):
+        self._record("list_memories", kwargs)
+        user_id = kwargs["user_id"]
+        layers = kwargs.get("layers")
+        if layers == ["l1_raw"]:
+            statuses = set(kwargs.get("statuses") or ["active"])
+            rows = [
+                {**row, "layer": "l1_raw", "user_id": user_id, "agent_id": "all-apps"}
+                for row in RAW_ROWS.get(user_id, [])
+                if row["status"] in statuses
+                and (
+                    not kwargs.get("include_sources") or row["source"] in kwargs["include_sources"]
+                )
+                and (not kwargs.get("session_ids") or row["session_id"] in kwargs["session_ids"])
+            ]
+            return {"vdb": {"memories": rows, "total": len(rows)}}
+        rows = [_chat(user_id, *row[:2], *row[3:]) for row in CHAT_ROWS.get(user_id, [])]
+        if layers:
+            rows = [row for row in rows if row["layer"] in layers]
+            if "l0_basic_info" in layers:
+                rows.append(_chat(user_id, "l0-1", "name is Fei", "l0_basic_info", "hermes"))
+        if kwargs.get("scene") == "all":
+            return {
+                "scene": "all",
+                "memories": [{**row, "scene": "chat", "kind": "chat"} for row in rows],
+                "total": len(rows),
+                "totals": {"chat": len(rows), "working": 0, "coding": 0},
+                "errors": {},
+            }
+        return {"vdb": {"memories": rows, "total": len(rows)}}
+
+    def memory_stats(self, *, user_id, statuses=None):
+        return {"total": 0, "request_id": "stats", "elapsed_ms": 1.0}
+
+    def get_user_basic_info(self, *, user_id):
+        self._record("get_user_basic_info", {"user_id": user_id})
+        return {
+            "request_id": f"basic-{user_id}",
+            "elapsed_ms": 1.5,
+            "memories": [
+                {
+                    "memory_id": f"{user_id}-l0-head",
+                    "content": "The user's name is Fei.",
+                    "layer": "l0_basic_info",
+                    "status": "active",
+                    "memory_at": 1_790_000_100,
+                    "gmt_created": 1_790_000_100,
+                    "tags": [],
+                    "basic_info_kv": {"name": "Fei"},
+                    "evolution_chain": [
+                        {
+                            "node_id": f"{user_id}-l0-head",
+                            "content": "The user's name is Fei.",
+                            "memory_at": "2026-10-31 10:00",
+                            "gmt_created": 1_790_000_100,
+                            "layer": "l0_basic_info",
+                        },
+                        {
+                            "node_id": f"{user_id}-l0-old",
+                            "content": "The user's name is 费天翔.",
+                            "memory_at": "2026-10-30 10:00",
+                            "gmt_created": 1_790_000_000,
+                            "layer": "l0_basic_info",
+                        },
+                    ],
+                }
+            ],
+        }
+
+
+class Handle:
+    def __init__(self, sdk: RecordingSdk) -> None:
+        self.sdk = sdk
+
+    async def client(self) -> RecordingSdk:
+        return self.sdk
+
+    async def close(self) -> None:
+        return None
+
+
+def application(tmp_path: Path, *, home: Path | None = None):
+    settings = Settings(
+        data_dir=tmp_path / "data",
+        source_home=home or tmp_path,
+        api_token=SecretStr(TOKEN),
+        imports_enabled=home is not None,
+        automatic_memory_processing=False,
+        live_sync_enabled=False,
+    )
+    return create_app(
+        settings,
+        secret_store=InMemorySecretStore(),
+        source_catalog=(
+            build_source_catalog(home=home, environment={}, platform="darwin")
+            if home is not None
+            else None
+        ),
+    )
+
+
+def client_for(app) -> AsyncClient:
+    return AsyncClient(
+        transport=ASGITransport(app=app),
+        base_url="http://test",
+        headers={"Authorization": f"Bearer {TOKEN}"},
+    )
+
+
+async def _model_ready(_app) -> bool:
+    return True
+
+
+def run_with_sdk(tmp_path, monkeypatch, scenario) -> None:
+    async def main() -> None:
+        app = application(tmp_path)
+        sdk = RecordingSdk()
+
+        @asynccontextmanager
+        async def write_client():
+            yield sdk
+
+        async with app.router.lifespan_context(app):
+            app.state.sdk_client_handle = Handle(sdk)
+            monkeypatch.setattr(app.state.sdk_write_coordinator, "use_client", write_client)
+            monkeypatch.setattr("one_memory.api._recall_model_ready", _model_ready)
+            async with client_for(app) as client:
+                await scenario(app, sdk, client)
+
+    asyncio.run(main())
+
+
+PASS_THROUGH = {
+    "min_score": 0.5,
+    "schema_min_score": 0.6,
+    "working_fact_min_score": 0.3,
+    "working_experience_min_score": 0.45,
+    "schema_limit": 4,
+    "intention_limit": 2,
+    "rerank": False,
+    "reader": "legacy",
+    "scene": "chat",
+    "search_mode": "agentic",
+    "include_sources": ["hermes", "codex"],
+    "exclude_sources": ["cursor"],
+    "created_after": 1_780_000_000,
+    "session_ids": ["s-1"],
+    "agent_ids": ["all-apps"],
+    "workspace_id": "ws-1",
+    "branch": "main",
+    "layers": ["l2_fact", "l0_basic_info"],
+}
+#: How each field reaches `client.search` (lists stay lists).
+EXPECTED_SEARCH = {**PASS_THROUGH}
+
+
+def _memory_search(sdk: RecordingSdk) -> dict:
+    return [call for call in sdk.all("search") if call.get("layers") != ["l1_raw"]][-1]
+
+
+@pytest.mark.fresh_sdk
+def test_every_context_search_option_reaches_the_sdk_and_absent_ones_keep_the_old_call(
+    tmp_path, monkeypatch
+) -> None:
+    async def scenario(app, sdk, client):
+        baseline = await client.post(
+            "/api/v2/context/search", json={"query": "tea", "kinds": ["memory"]}
+        )
+        assert baseline.status_code == 200, baseline.text
+        before = _memory_search(sdk)
+        assert before["min_score"] == before["schema_min_score"] == 0.35
+        assert before["scene"] == "normal" and before["reader"] == "hybrid"
+        assert before["schema_limit"] == 3 and before["intention_limit"] == 1
+        for absent in ("rerank", "search_mode", "layers", "session_ids", "created_after"):
+            assert absent not in before
+
+        response = await client.post(
+            "/api/v2/context/search",
+            json={"query": "tea", "kinds": ["memory"], **PASS_THROUGH},
+        )
+        assert response.status_code == 200, response.text
+        sent = _memory_search(sdk)
+        for name, value in EXPECTED_SEARCH.items():
+            assert sent[name] == value, name
+        assert sent["user_ids"] == ["local"]
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_every_recall_and_ask_option_reaches_the_sdk(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        recalled = await client.post(
+            "/api/v1/recall", json={"query": "tea", "user_id": "wife", **PASS_THROUGH}
+        )
+        assert recalled.status_code == 200, recalled.text
+        sent = _memory_search(sdk)
+        for name, value in EXPECTED_SEARCH.items():
+            assert sent[name] == value, name
+        assert sent["user_ids"] == ["wife"]
+
+        ask_options = {
+            name: value
+            for name, value in PASS_THROUGH.items()
+            if name
+            not in {
+                "working_fact_min_score",
+                "working_experience_min_score",
+                "schema_limit",
+                "intention_limit",
+                "reader",
+                "scene",
+                "search_mode",
+                "workspace_id",
+                "branch",
+            }
+        }
+        asked = await client.post(
+            "/api/v2/memories/ask", json={"question": "tea?", "user_id": "wife", **ask_options}
+        )
+        assert asked.status_code == 200, asked.text
+        sent = _memory_search(sdk)
+        for name, value in ask_options.items():
+            assert sent[name] == value, name
+        assert sent["user_ids"] == ["wife"] and sent["scene"] == "normal"
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_min_score_is_enforced_on_what_core_returns(tmp_path, monkeypatch) -> None:
+    """Acceptance item 9: `min_score=0.9` answers only results scoring ≥ 0.9."""
+
+    async def scenario(app, sdk, client):
+        context = await client.post(
+            "/api/v2/context/search", json={"query": "tea", "min_score": 0.9}
+        )
+        assert context.status_code == 200, context.text
+        items = context.json()["items"]
+        assert items and all(item["score"] >= 0.9 for item in items)
+        assert {item.get("memory_id") or item.get("raw_id") for item in items} == {
+            "m-high",
+            "raw-1",
+        }
+        raw_call = sdk.last("search") if sdk.last("search").get("layers") else None
+        assert raw_call is None or raw_call["min_score"] == 0.9
+
+        # The fused scene ranks by 1/(60+rank); the lane's own score is the floor's.
+        fused = await client.post(
+            "/api/v2/context/search",
+            json={"query": "tea", "kinds": ["memory"], "scene": "all", "min_score": 0.9},
+        )
+        assert [item["memory_id"] for item in fused.json()["items"]] == ["m-high"]
+        assert fused.json()["items"][0]["score"] == 0.95
+
+        recalled = await client.post("/api/v1/recall", json={"query": "tea", "min_score": 0.9})
+        assert recalled.status_code == 200, recalled.text
+        assert [hit["memory_id"] for hit in recalled.json()["hits"]] == ["m-high"]
+
+        # A working-scene fact is read back from its own namespace.
+        working = await client.post(
+            "/api/v2/context/search",
+            json={"query": "uv", "kinds": ["memory"], "scene": "working", "min_score": 0.9},
+        )
+        assert [item["memory_id"] for item in working.json()["items"]] == ["f-1"]
+        assert sdk.last("get") == {"memory_id": "f-1", "kind": "fact", "variant": "working"}
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+@pytest.mark.parametrize(
+    "extra",
+    [
+        {"min_score": 1.5},
+        {"min_score": -0.1},
+        {"schema_min_score": "high"},
+        {"scene": "normal"},
+        {"search_mode": "fast"},
+        {"reader": "vector"},
+        {"layers": ["l9_nothing"]},
+        {"layers": []},
+        {"branch": "main"},
+        {"rerank": "no"},
+        {"created_after": -1},
+        {"user_ids": ["wife"]},
+        {"embed_override": {"model": "x"}},
+        {"query_embedding": [0.1]},
+        {"request_id": "r-1"},
+    ],
+)
+def test_out_of_range_and_unexposed_search_fields_answer_422(tmp_path, extra) -> None:
+    async def scenario():
+        app = application(tmp_path)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            for path, body in (
+                ("/api/v2/context/search", {"query": "q"}),
+                ("/api/v1/recall", {"query": "q"}),
+                ("/api/v2/memories/ask", {"question": "q"}),
+            ):
+                response = await client.post(path, json={**body, **extra})
+                assert response.status_code == 422, (path, response.text)
+
+    asyncio.run(scenario())
+
+
+@pytest.mark.fresh_sdk
+def test_raw_lane_takes_the_callers_sources_only_inside_its_authority(
+    tmp_path, monkeypatch
+) -> None:
+    async def scenario(app, sdk, client):
+        response = await client.post(
+            "/api/v2/context/search",
+            json={
+                "query": "tea",
+                "kinds": ["raw"],
+                "include_sources": ["hermes"],
+                "session_ids": ["s-1"],
+                "created_after": 1_700_000_000,
+            },
+        )
+        assert response.status_code == 200, response.text
+        raw_call = sdk.last("search")
+        assert raw_call["layers"] == ["l1_raw"]
+        assert raw_call["include_sources"] == ["hermes"]
+        assert raw_call["session_ids"] == ["s-1"]
+        assert raw_call["created_after"] == 1_700_000_000
+        # Memory-lane knobs never reach the raw search.
+        assert raw_call["scene"] == "chat" and raw_call["schema_limit"] == 0
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_library_honours_layers_statuses_and_filters(tmp_path, monkeypatch) -> None:
+    """Acceptance item 10: `layer=l0_basic_info` returns only L0 records."""
+
+    async def scenario(app, sdk, client):
+        page = await client.get("/api/v2/memory-library", params={"layer": "l0_basic_info"})
+        assert page.status_code == 200, page.text
+        assert {memory["layer"] for memory in page.json()["memories"]} == {"l0_basic_info"}
+        sent = sdk.last("list_memories")
+        assert sent["layers"] == ["l0_basic_info"] and sent["scene"] is None
+
+        await client.get(
+            "/api/v2/memory-library",
+            params=[
+                ("layers", "l2_fact,l6_schema"),
+                ("layers", "l7_intention"),
+                ("statuses", "active,superseded"),
+                ("session_ids", "s-1"),
+                ("agent_ids", "all-apps"),
+                ("exclude_sources", "cursor"),
+                ("user_id", "wife"),
+            ],
+        )
+        sent = sdk.last("list_memories")
+        assert sent["layers"] == ["l2_fact", "l6_schema", "l7_intention"]
+        assert sent["statuses"] == ["active", "superseded"]
+        assert sent["session_ids"] == ["s-1"] and sent["agent_ids"] == ["all-apps"]
+        assert sent["exclude_sources"] == ["cursor"] and sent["user_id"] == "wife"
+
+        # Without a layer the page is what it always was.
+        await client.get("/api/v2/memory-library")
+        sent = sdk.last("list_memories")
+        assert sent["scene"] == "all" and "layers" not in sent
+
+        searched = await client.get(
+            "/api/v2/memory-library", params={"query": "tea", "layers": "l2_fact"}
+        )
+        assert searched.status_code == 200, searched.text
+        assert sdk.last("search")["layers"] == ["l2_fact"]
+
+        for params in (
+            {"layers": "l9_nothing"},
+            {"statuses": "gone"},
+            {"layers": "l2_fact", "scene": "working"},
+        ):
+            refused = await client.get("/api/v2/memory-library", params=params)
+            assert refused.status_code == 422, (params, refused.text)
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_layer_counts_read_each_layer_for_one_user(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        counts = await client.get("/api/v2/memory-library/layers", params={"user_id": "wife"})
+        assert counts.status_code == 200, counts.text
+        body = counts.json()
+        assert body["layers"]["l1_raw"] == 1
+        assert body["layers"]["l2_fact"] == 1
+        assert body["layers"]["l0_basic_info"] == 1
+        assert body["total"] == sum(body["layers"].values())
+        calls = sdk.all("list_memories")
+        assert {call["user_id"] for call in calls} == {"wife"}
+        [raw_call] = [call for call in calls if call["layers"] == ["l1_raw"]]
+        assert raw_call["statuses"] == ["active", "shadow"]
+        assert (
+            await client.get("/api/v2/memory-library/layers", params={"layer": "x"})
+        ).status_code == 422
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_raw_lists_active_and_shadow_turns_with_roles_and_hashes(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        page = await client.get("/api/v2/raw")
+        assert page.status_code == 200, page.text
+        body = page.json()
+        assert body["total"] == 2 and len(body["items"]) == 2
+        sent = sdk.last("list_memories")
+        assert sent["layers"] == ["l1_raw"] and sent["statuses"] == ["active", "shadow"]
+        first = next(item for item in body["items"] if item["raw_id"] == "raw-1")
+        assert first["status"] == "shadow"
+        assert [message["role"] for message in first["messages"]] == ["user", "assistant"]
+        assert first["messages"][0]["content"] == "我叫费天翔"
+        assert first["role"] is None
+        expected = hashlib.sha256(first["content"].encode()).hexdigest()
+        assert first["content_hash"] == f"sha256:{expected}"
+        assert first["session_id"] == "s-1" and first["source"] == "hermes"
+        assert first["created"].startswith("2026-")
+        second = next(item for item in body["items"] if item["raw_id"] == "raw-2")
+        assert second["role"] == "user"
+
+        await client.get(
+            "/api/v2/raw",
+            params=[("source", "hermes,codex"), ("session_id", "s-1"), ("order", "asc")],
+        )
+        sent = sdk.last("list_memories")
+        assert sent["include_sources"] == ["hermes", "codex"]
+        assert sent["session_ids"] == ["s-1"] and sent["order"] == "asc"
+
+        windowed = (await client.get("/api/v2/raw", params={"after": 1_790_050_000})).json()
+        assert windowed["time_filtered"] is True
+        assert [item["raw_id"] for item in windowed["items"]] == ["raw-2"]
+
+        wife = (await client.get("/api/v2/raw", params={"user_id": "wife"})).json()
+        assert [item["raw_id"] for item in wife["items"]] == ["raw-w"]
+        assert sdk.last("list_memories")["user_id"] == "wife"
+        assert (await client.get("/api/v2/raw", params={"statuses": "bogus"})).status_code == 422
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_profile_versions_read_the_l0_chain_for_one_user(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        response = await client.get("/api/v2/user-profile/versions", params={"user_id": "wife"})
+        assert response.status_code == 200, response.text
+        body = response.json()
+        assert body["user_id"] == "wife"
+        assert sdk.last("get_user_basic_info") == {"user_id": "wife"}
+        [chain] = body["chains"]
+        assert chain["memory_id"] == "wife-l0-head"
+        assert chain["basic_info"] == {"name": "Fei"}
+        assert [(v["memory_id"], v["current"]) for v in chain["versions"]] == [
+            ("wife-l0-head", True),
+            ("wife-l0-old", False),
+        ]
+        local = (await client.get("/api/v2/user-profile/versions")).json()
+        assert local["user_id"] == "local"
+        assert sdk.last("get_user_basic_info") == {"user_id": "local"}
+        refused = await client.get("/api/v2/user-profile/versions", params={"owner": "wife"})
+        assert refused.status_code == 422
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+@pytest.mark.fresh_sdk
+def test_graph_takes_layers_and_session_filters(tmp_path, monkeypatch) -> None:
+    async def scenario(app, sdk, client):
+        response = await client.get(
+            "/api/v2/memories/graph",
+            params={"layers": "l2_fact", "session_ids": "s-1", "user_id": "wife"},
+        )
+        assert response.status_code == 200, response.text
+        sent = sdk.last("list_memories")
+        assert sent["layers"] == ["l2_fact"] and sent["scene"] is None
+        assert sent["session_ids"] == ["s-1"] and sent["user_id"] == "wife"
+
+    run_with_sdk(tmp_path, monkeypatch, scenario)
+
+
+def _hermes_state_db(root: Path, text: str) -> None:
+    root.mkdir(parents=True, exist_ok=True)
+    connection = sqlite3.connect(root / "state.db")
+    connection.execute(
+        "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT,"
+        " content TEXT, timestamp REAL, tool_name TEXT, active INTEGER)"
+    )
+    connection.executemany(
+        "INSERT INTO messages VALUES (?, ?, ?, ?, 1790000000, NULL, 1)",
+        [(1, "session-1", "user", text), (2, "session-1", "assistant", "ok")],
+    )
+    connection.commit()
+    connection.close()
+
+
+@pytest.mark.fresh_sdk
+def test_v2_source_and_import_aliases_behave_like_v1_per_user(tmp_path) -> None:
+    async def scenario():
+        home = tmp_path / "home"
+        hermes = home / ".hermes"
+        wife = hermes / "profiles" / "wife"
+        _hermes_state_db(hermes, "I am the husband")
+        _hermes_state_db(wife, "I am the wife")
+        app = application(tmp_path, home=home)
+        async with app.router.lifespan_context(app), client_for(app) as client:
+            location = (await client.post("/api/v1/sources/hermes/location")).json()
+            local = await client.post(
+                "/api/v1/sources/connect",
+                json={
+                    "source_ids": ["hermes"],
+                    "request_id": str(uuid4()),
+                    "source_roots": {"hermes": location["root_id"]},
+                    "enable_sync": True,
+                },
+            )
+            assert local.status_code == 200, local.text
+            connected = await client.post(
+                "/api/v1/sources/connect",
+                json={
+                    "source_ids": ["hermes"],
+                    "request_id": str(uuid4()),
+                    "user_id": "wife",
+                    "root": str(wife),
+                },
+            )
+            assert connected.status_code == 200, connected.text
+            for job in (*local.json()["jobs"], *connected.json()["jobs"]):
+                await app.state.import_supervisor.wait(job["job_id"])
+
+            v1 = (await client.get("/api/v1/sources")).json()
+            v2 = (await client.get("/api/v2/sources")).json()
+            assert v1 == v2
+            wife_sources = (await client.get("/api/v2/sources", params={"user_id": "wife"})).json()
+            assert [source["source_id"] for source in wife_sources] == ["hermes"]
+            assert wife_sources[0]["authorized"] is True
+            assert (await client.get("/api/v2/sources", params={"u": "x"})).status_code == 422
+
+            created = await client.post(
+                "/api/v2/import-jobs",
+                json={"source_id": "hermes", "operation": "sync", "user_id": "wife"},
+            )
+            assert created.status_code == 200, created.text
+            wife_job = created.json()
+            assert wife_job["user_id"] == "wife" and wife_job["source_id"] == "hermes"
+            await app.state.import_supervisor.wait(wife_job["job_id"])
+            refused = await client.post(
+                "/api/v2/import-jobs",
+                json={"source_id": "claude-code", "user_id": "wife"},
+            )
+            assert refused.status_code in {404, 422}
+            unknown = await client.post(
+                "/api/v2/import-jobs",
+                json={"source_id": "hermes", "from_beginning": True},
+            )
+            assert unknown.status_code == 422
+
+            listed = (await client.get("/api/v2/import-jobs", params={"user_id": "wife"})).json()
+            assert {job["user_id"] for job in listed} == {"wife"}
+            assert (await client.get("/api/v2/import-jobs")).json() == (
+                await client.get("/api/v1/import-jobs")
+            ).json()
+            path = f"/api/v2/import-jobs/{wife_job['job_id']}"
+            assert (await client.get(path)).status_code == 404
+            own = await client.get(path, params={"user_id": "wife"})
+            assert own.status_code == 200 and own.json()["job_id"] == wife_job["job_id"]
+
+            synced = await client.post("/api/v2/sources/hermes/sync", json={"user_id": "wife"})
+            v1_synced = await client.post("/api/v1/sources/hermes/sync")
+            assert synced.status_code == v1_synced.status_code, (synced.text, v1_synced.text)
+            assert synced.status_code == 200
+            assert synced.json()["user_id"] == "wife"
+            assert v1_synced.json()["user_id"] == "local"
+            other = await client.post("/api/v2/sources/claude-code/sync", json={"user_id": "wife"})
+            assert other.status_code in {404, 422}
+            assert (
+                await client.post("/api/v2/sources/hermes/sync", json={"owner": "wife"})
+            ).status_code == 422
+
+    asyncio.run(scenario())

memhub/apps/core/tests/test_sdk_data_plane_migration.py

@@ -6,7 +6,7 @@ import io
 import json
 import sqlite3
 from contextlib import closing
-from datetime import UTC, datetime, timedelta
+from datetime import UTC, datetime
 from pathlib import Path
 
 import pytest
@@ -30,7 +30,6 @@ from one_memory.sdk_data_plane_migration import (
     MemoryAuthority,
     SdkDataPlaneMigrationStore,
 )
-from one_memory.sdk_memory_runtime import SdkMemoryRuntime
 from one_memory.sdk_migration_bundle import create_sdk_migration_bundle
 from one_memory.sdk_migration_runner import (
     PREPARED_PHASE,
@@ -708,6 +707,7 @@ def test_cutover_waits_until_the_new_runtime_reads_every_chat_memory(store, acti
         "chat_readable": 1,
         "productivity_expected": 0,
         "productivity_readable": 0,
+        "chat_missing_before_migration": 0,
         "passed": False,
     }
     assert store.authority() is MemoryAuthority.LEGACY
@@ -774,18 +774,186 @@ def test_retired_coding_experiences_do_not_block_but_an_empty_store_does(store,
     assert (result.readback.productivity_expected, result.readback.productivity_readable) == (2, 1)
 
 
-def test_memories_without_a_model_profile_keep_legacy_authority(store, monkeypatch):
+def test_a_home_without_a_model_profile_migrates_on_the_builtin_embedding(store, monkeypatch):
+    """#234: no generation model was ever configured, and the upgrade stopped.
+
+    No profile means no LLM, not no embedding: the vectors are the built-in
+    model's, the same space the Runtime reads with. They are reattached and
+    read back through it, and the client is opened without resolving a key.
+    """
+    from one_memory.hy_sdk_runtime import builtin_retrieval_snapshot, vector_collection_bases
+
     monkeypatch.setattr("one_memory.sdk_migration_runner.read_active_profile", lambda _db: None)
+    home = store.database_path.parent
+    builtin = builtin_retrieval_snapshot()
+    _current, legacy_bases = vector_collection_bases(builtin)
+    client = _chroma(home)
+    _fill(client, f"{legacy_bases[0]}_512", ["m1", "m2"])
+    del client
+    _confirmed(store, "m1", "m2")
+    opened: list[str] = []
+
+    def open_client(profile):
+        opened.append(profile.profile_id)
+        return _reading({"m1", "m2"})(profile)
+
+    result = run_startup_migration(
+        data_dir=home,
+        runtime_build="21",
+        sdk_version="2.0.5",
+        namespace="governed-v2",
+        open_client=open_client,
+    )
+
+    assert result.state == "cutover"
+    assert opened == [builtin.profile_id]
+    assert [item.legacy for item in result.reattached] == [f"{legacy_bases[0]}_512"]
+    assert result.readback is not None and result.readback.passed
+    assert store.authority() is MemoryAuthority.SDK
+
+
+def test_the_builtin_readback_client_resolves_no_secret(tmp_path, monkeypatch):
+    """The built-in snapshot has no stored version and no key to look up."""
+    from one_memory import sdk_migration_verify
+    from one_memory.hy_sdk_runtime import builtin_retrieval_snapshot
+
+    leased: list[tuple[str, object]] = []
+
+    class Lease:
+        client = object()
+
+        def close(self) -> None:
+            leased.append(("closed", None))
+
+    def acquire(profile, *, api_key, data_dir):
+        leased.append((profile.profile_id, api_key))
+        return Lease()
+
+    def no_lookup(*_args, **_kwargs):
+        raise AssertionError("a key-less profile must not touch the secret store")
+
+    monkeypatch.setattr("one_memory.hy_sdk_runtime.acquire_client", acquire)
+    monkeypatch.setattr(
+        "one_memory.model_profiles.ModelProfileStore.resolve_snapshot_secret", no_lookup
+    )
+    with sdk_migration_verify.runtime_sdk_client(
+        tmp_path / "missing.sqlite3", tmp_path, builtin_retrieval_snapshot()
+    ) as client:
+        assert client is Lease.client
+    assert leased == [("builtin-retrieval", None), ("closed", None)]
+
+
+def test_vectors_left_by_a_deleted_profile_stop_the_upgrade_by_name(store, monkeypatch):
+    """No active profile, and the vectors sit in another profile's space.
+
+    The built-in model cannot read them, so the old abort was right for this
+    one shape; it now says what is actually wrong, before anything changes.
+    """
+    from one_memory.hy_sdk_runtime import vector_collection_bases
+    from one_memory.model_profiles import ModelProfileSnapshot, ProviderKind
+
+    monkeypatch.setattr("one_memory.sdk_migration_runner.read_active_profile", lambda _db: None)
+    deleted = ModelProfileSnapshot(
+        profile_id="deleted",
+        revision=3,
+        provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+        base_url="https://models.example/v1",
+        model="chat",
+        embedding_provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+        embedding_base_url="https://models.example/v1",
+        embedding_model="text-embedding-3-large",
+        embedding_dimensions=1024,
+        timeout_seconds=60,
+        max_concurrency=1,
+        context_window=None,
+        has_api_key=True,
+    )
+    current, _legacy = vector_collection_bases(deleted)
+    home = store.database_path.parent
+    client = _chroma(home)
+    _fill(client, f"{current}_1024", ["m1"])
+    del client
     _confirmed(store, "m1")
 
-    with pytest.raises(DataPlaneMigrationError, match="sdk_migration_model_profile_required"):
+    with pytest.raises(DataPlaneMigrationError, match="sdk_migration_vector_profile_missing"):
         run_startup_migration(
-            data_dir=store.database_path.parent,
+            data_dir=home,
             runtime_build="21",
             sdk_version="2.0.5",
             namespace="governed-v2",
+            open_client=_reading({"m1"}),
         )
     assert store.authority() is MemoryAuthority.LEGACY
+    assert not (home / "backups").exists()
+
+
+def test_confirmed_chats_the_sdk_had_already_merged_do_not_fail_the_upgrade(store, active_profile):
+    """#229/#240: 658 confirmed chats, 657 in the store, and the upgrade failed.
+
+    A 1.0.4 library can hold confirmed rows whose SDK node had been merged or
+    deleted before the upgrade; no migration can read those back. The check
+    holds chat to what the store held before this run and counts the rest.
+    """
+    from one_memory.hy_sdk_runtime import vector_collection_bases
+
+    home = store.database_path.parent
+    stored = [f"chat-{index:03d}" for index in range(657)]
+    current, _legacy = vector_collection_bases(active_profile)
+    client = _chroma(home)
+    _fill(client, f"{current}_512", stored)
+    del client
+    _confirmed(store, *stored, "chat-merged-before-upgrade")
+
+    result = run_startup_migration(
+        data_dir=home,
+        runtime_build="21",
+        sdk_version="2.0.5",
+        namespace="governed-v2",
+        open_client=_reading(set(stored)),
+    )
+
+    assert result.state == "cutover"
+    assert result.readback is not None
+    assert result.readback.as_document() == {
+        "chat_expected": 657,
+        "chat_readable": 657,
+        "productivity_expected": 0,
+        "productivity_readable": 0,
+        "chat_missing_before_migration": 1,
+        "passed": True,
+    }
+    assert store.authority() is MemoryAuthority.SDK
+
+
+def test_a_chat_in_the_store_before_the_upgrade_must_still_read_back(store, active_profile):
+    """Only what was already gone is excused; losing a stored memory still fails."""
+    from one_memory.hy_sdk_runtime import vector_collection_bases
+    from one_memory.sdk_migration_verify import MemoryReadbackFailed
+
+    home = store.database_path.parent
+    current, _legacy = vector_collection_bases(active_profile)
+    client = _chroma(home)
+    _fill(client, f"{current}_512", ["m1", "m2"])
+    del client
+    _confirmed(store, "m1", "m2", "merged")
+
+    with pytest.raises(MemoryReadbackFailed) as caught:
+        run_startup_migration(
+            data_dir=home,
+            runtime_build="21",
+            sdk_version="2.0.5",
+            namespace="governed-v2",
+            open_client=_reading({"m1"}),
+        )
+    assert caught.value.report.as_document() == {
+        "chat_expected": 2,
+        "chat_readable": 1,
+        "productivity_expected": 0,
+        "productivity_readable": 0,
+        "chat_missing_before_migration": 1,
+        "passed": False,
+    }
+    assert store.authority() is MemoryAuthority.LEGACY
 
 
 def test_a_build_20_home_gets_its_vectors_back_and_is_checked_against_its_backup(
@@ -1072,7 +1240,13 @@ def test_host_capture_is_fenced_for_the_whole_migration(store):
     )
 
     assert result.state == "cutover"
-    assert fenced == {"bundle": True, "prepare": True, "cutover": True, "cleanup": True}
+    assert fenced == {
+        "bundle": True,
+        "prepare": True,
+        "vectors": True,
+        "cutover": True,
+        "cleanup": True,
+    }
     assert not sdk_migration_capture_fenced(home)
 
 
@@ -1175,57 +1349,6 @@ def test_raw_http_dto_uses_the_unified_memory_list_shape():
     assert page.raw[0].custom == {"role": "user"}
 
 
-def test_sdk_raw_retention_keeps_only_the_latest_three_days():
-    now = datetime(2026, 10, 23, tzinfo=UTC)
-
-    class Client:
-        def __init__(self):
-            self.rows = [
-                {
-                    "memory_id": "old",
-                    "memory_at": int((now - timedelta(days=4)).timestamp()),
-                },
-                {
-                    "memory_id": "recent",
-                    "memory_at": int((now - timedelta(days=2)).timestamp()),
-                },
-            ]
-
-        def list_memories(self, **_kwargs):
-            return {
-                "vdb": {
-                    "memories": list(self.rows),
-                    "total": len(self.rows),
-                    "limit": 500,
-                    "offset": 0,
-                }
-            }
-
-        def delete(self, memory_id):
-            before = len(self.rows)
-            self.rows = [row for row in self.rows if row["memory_id"] != memory_id]
-            return {
-                "success": True,
-                "deleted_count": before - len(self.rows),
-            }
-
-    client = Client()
-
-    class Handle:
-        async def client(self):
-            return client
-
-    runtime = SdkMemoryRuntime(
-        coordinator=None,
-        read_handle=Handle(),
-        write_queue=None,
-        extraction_settings=None,
-        data_dir=Path("."),
-    )
-    assert asyncio.run(runtime.prune_expired_raw(now=now)) == 1
-    assert [row["memory_id"] for row in client.rows] == ["recent"]
-
-
 # The UI asks which memory data plane is authoritative and must get the
 # answer the API gates use.
 

memhub/apps/core/tests/test_sdk_migration_log.py

@@ -306,7 +306,7 @@ def test_the_runner_names_the_step_and_sub_step_that_failed(tmp_path, monkeypatc
         ("step.finished", "preflight.profile_read"),
         ("step.finished", "preflight.resumable_identity"),
         ("step.finished", "preflight.legacy_ids_scan"),
-        ("step.finished", "preflight.model_profile_check"),
+        ("step.finished", "preflight.vector_space_check"),
         ("step.finished", "prepare.recover_abandoned_staging"),
         ("step.finished", "bundle.ensure"),
         ("step.finished", "bundle.restore_point"),

memhub/apps/core/tests/test_sdk_migration_restore.py

@@ -463,11 +463,17 @@ def test_a_run_killed_after_its_restore_point_is_rolled_back_on_the_next_start(h
 
 
 def test_a_preflight_failure_changes_nothing_and_needs_no_restore(home, monkeypatch):
+    def unreadable(_data_dir, _profile):
+        raise DataPlaneMigrationError("sdk_migration_vector_profile_missing")
+
     monkeypatch.setattr("one_memory.sdk_migration_runner.read_active_profile", lambda _db: None)
+    monkeypatch.setattr(
+        "one_memory.sdk_migration_runner.vectors_belong_to_another_space", unreadable
+    )
     before = _state(home)
     report: dict[str, object] = {}
 
-    with pytest.raises(DataPlaneMigrationError, match="sdk_migration_model_profile_required"):
+    with pytest.raises(DataPlaneMigrationError, match="sdk_migration_vector_profile_missing"):
         _run(home, report=report)
 
     assert report == {}

memhub/apps/core/tests/test_search_without_profile.py

@@ -1,5 +1,6 @@
 """Model-free SDK search remains available without an active write profile."""
 
+import asyncio
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -95,3 +96,158 @@ async def test_search_without_profile_uses_public_sdk_search() -> None:
     assert options["user_ids"] == ["local"]
     assert options["agent_ids"] == []
     assert options["scene"] == "normal"
+
+
+def _activate(profiles, secrets, *, dimensions: int, model: str = "chat"):
+    from pydantic import SecretStr
+
+    from one_memory.model_profiles import ModelCapabilities, ModelProfileInput
+
+    profile = profiles.create(
+        ModelProfileInput(
+            name=f"embedding-{dimensions}",
+            provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            base_url="https://models.example/v1",
+            model=model,
+            embedding_provider_kind=ProviderKind.OPENAI_COMPATIBLE,
+            embedding_base_url="https://models.example/v1",
+            embedding_model="qwen3-embedding-4b",
+            embedding_dimensions=dimensions,
+            api_key=SecretStr("sk-test"),
+        ),
+        secrets,
+    )
+    profiles.record_capabilities(
+        profile.profile_id,
+        expected_revision=profile.revision,
+        capabilities=ModelCapabilities(
+            endpoint_reachable=True,
+            model_available=True,
+            structured_output=True,
+            embedding_available=True,
+        ),
+        secrets=secrets,
+    )
+    profiles.activate(profile.profile_id, secrets)
+    return profiles.snapshot(profile.profile_id, profiles.active_identity()[1])
+
+
+class _Recorder:
+    """Stands in for `acquire_client`: records the collection each client opens."""
+
+    def __init__(self, data_dir: Path) -> None:
+        self.data_dir = data_dir
+        self.opened: list[tuple[str, int | None, str | None]] = []
+        self.closed: list[str] = []
+
+    def __call__(self, profile, *, api_key, data_dir, http_transport_factory=None, mode="pro"):
+        from one_memory.hy_sdk_runtime import build_sdk_config
+
+        config = build_sdk_config(profile, api_key=api_key, data_dir=data_dir)
+        name = config.vector_store.collection_name
+        self.opened.append((name, config.vector_store.embedding_dims, api_key))
+        return SimpleNamespace(
+            client=SimpleNamespace(collection=name),
+            close=lambda: self.closed.append(name),
+        )
+
+
+@pytest.mark.anyio
+async def test_reads_open_the_collection_the_active_profile_writes(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """#242: a 2560-dim profile wrote `memhub_<sha>_2560`, reads opened `_512`."""
+    from one_memory.hy_sdk_runtime import build_sdk_config
+    from one_memory.model_profiles import ModelProfileStore
+    from one_memory.sdk_read_handle import ActiveProfileReadHandle
+    from one_memory.secrets import InMemorySecretStore
+
+    recorder = _Recorder(tmp_path)
+    monkeypatch.setattr("one_memory.hy_sdk_runtime.acquire_client", recorder)
+    profiles = ModelProfileStore(tmp_path / "one-memory.sqlite3")
+    secrets = InMemorySecretStore()
+    handle = ActiveProfileReadHandle(profiles=profiles, secrets=secrets, data_dir=tmp_path)
+
+    builtin = build_sdk_config(builtin_retrieval_snapshot(), api_key=None, data_dir=tmp_path)
+    assert (await handle.client()).collection == builtin.vector_store.collection_name
+
+    written = _activate(profiles, secrets, dimensions=2560)
+    writes = build_sdk_config(written, api_key="sk-test", data_dir=tmp_path)
+
+    async def read() -> str:
+        return str((await handle.client()).collection)
+
+    assert await asyncio.create_task(read()) == writes.vector_store.collection_name
+    assert recorder.opened[-1] == (writes.vector_store.collection_name, 2560, "sk-test")
+    assert writes.vector_store.collection_name != builtin.vector_store.collection_name
+    await handle.close()
+
+
+@pytest.mark.anyio
+async def test_a_profile_change_rebuilds_reads_without_closing_one_in_flight(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    from one_memory.model_profiles import ModelProfileStore
+    from one_memory.sdk_read_handle import ActiveProfileReadHandle
+    from one_memory.secrets import InMemorySecretStore
+
+    recorder = _Recorder(tmp_path)
+    monkeypatch.setattr("one_memory.hy_sdk_runtime.acquire_client", recorder)
+    profiles = ModelProfileStore(tmp_path / "one-memory.sqlite3")
+    secrets = InMemorySecretStore()
+    handle = ActiveProfileReadHandle(profiles=profiles, secrets=secrets, data_dir=tmp_path)
+    _activate(profiles, secrets, dimensions=1024)
+
+    holding = asyncio.Event()
+    finish = asyncio.Event()
+
+    async def slow_read() -> str:
+        client = await handle.client()
+        holding.set()
+        await finish.wait()
+        return str(client.collection)
+
+    in_flight = asyncio.create_task(slow_read())
+    await holding.wait()
+    [(old, _, _)] = recorder.opened
+
+    _activate(profiles, secrets, dimensions=2560, model="chat-2")
+
+    async def read() -> str:
+        return str((await handle.client()).collection)
+
+    new = await asyncio.create_task(read())
+    assert new != old
+    # The read that started on the old space still has its client.
+    assert recorder.closed == []
+
+    finish.set()
+    assert await in_flight == old
+    for _ in range(5):
+        await asyncio.sleep(0)
+    assert recorder.closed == [old]
+    await handle.close()
+    assert recorder.closed == [old, new]
+
+
+def test_every_api_read_shares_the_active_profile_handle(tmp_path: Path) -> None:
+    """Library, search, recall, graph, profile, raw, ask and context all read
+    through `_sdk_client_handle`, so it must be the active-profile handle."""
+    from one_memory.api import _sdk_client_handle
+    from one_memory.config import Settings
+    from one_memory.model_profiles import ModelProfileStore
+    from one_memory.sdk_read_handle import ActiveProfileReadHandle
+    from one_memory.secrets import InMemorySecretStore
+
+    state = SimpleNamespace(
+        sdk_client_handle=None,
+        model_profile_store=ModelProfileStore(tmp_path / "one-memory.sqlite3"),
+        secret_store=InMemorySecretStore(),
+        settings=Settings(_env_file=None, data_dir=tmp_path),
+    )
+    app = SimpleNamespace(state=state)
+
+    handle = _sdk_client_handle(app)  # type: ignore[arg-type]
+
+    assert isinstance(handle, ActiveProfileReadHandle)
+    assert _sdk_client_handle(app) is handle is state.sdk_client_handle  # type: ignore[arg-type]

memhub/apps/desktop/src/stable-runtime.ts

@@ -133,6 +133,7 @@ const SAFE_STABLE_RUNTIME_ERROR_CODES = new Set([
   "sdk_migration_preflight_failed",
   "sdk_migration_memory_unreadable",
   "sdk_migration_model_profile_required",
+  "sdk_migration_vector_profile_missing",
   "sdk_migration_model_secret_unavailable",
   "sdk_migration_secure_storage_unavailable",
   "sdk_migration_disk_space_insufficient",
@@ -219,6 +220,8 @@ export const STABLE_RUNTIME_USER_MESSAGES: Record<string, string> = {
     "升级前核对发现，新版本还读不到你的全部记忆，所以没有切换数据源，旧数据保持原样可用。请把这个提示反馈给我们；原有记忆数据不会丢失。",
   sdk_migration_model_profile_required:
     "升级记忆数据格式需要先有可用的模型配置，用来核对新版本能否读回你的记忆。请在旧版本中完成模型配置后重启；你的记忆数据不会受到影响。",
+  sdk_migration_vector_profile_missing:
+    "你的记忆是用一个已删除的模型配置生成向量的，当前没有可用的模型配置能读回它们，所以没有升级。请在旧版本中重新添加那个模型配置并启用后重启；你的记忆数据不会受到影响。",
   sdk_migration_model_secret_unavailable:
     "模型密钥读不到（可能被系统钥匙串拒绝），无法核对新版本能否读回你的记忆。请允许 MemHub 访问钥匙串后重新启动；你的记忆数据不会受到影响。",
   sdk_migration_secure_storage_unavailable:

memhub/apps/web/src/ModelSettings.test.tsx

@@ -480,6 +480,30 @@ describe("ModelSettings", () => {
     });
   });
 
+  it("writes the prefilled embedding endpoint instead of leaving it to follow generation", async () => {
+    profiles = [
+      {
+        ...profile,
+        embedding_provider_kind: "openai_compatible",
+        embedding_model: "vector-old",
+        embedding_base_url: null,
+      },
+    ];
+    render(<ModelSettings />);
+    await userEvent.click(screen.getByRole("button", { name: "编辑配置" }));
+    await userEvent.clear(screen.getByLabelText("向量模型名"));
+    await userEvent.type(screen.getByLabelText("向量模型名"), "vector-new");
+    await userEvent.click(
+      screen.getByRole("button", { name: "高级：自行配置向量模型" }),
+    );
+    await userEvent.click(screen.getByRole("button", { name: "保存修改" }));
+    expect(update).toHaveBeenCalledWith(profile.profile_id, {
+      expected_revision: 1,
+      embedding_model: "vector-new",
+      embedding_base_url: profile.base_url,
+    });
+  });
+
   it("explicitly switches an edited remote embedding to builtin without retaining its dimensions", async () => {
     profiles = [
       {

memhub/apps/web/src/ModelSettings.tsx

@@ -631,10 +631,9 @@ export function ModelSettings({
             patch.embedding_model = embeddingModel;
           if (embeddingProvider !== editing.embedding_provider_kind)
             patch.embedding_provider_kind = embeddingProvider;
-          if (
-            embeddingBaseUrl !==
-            (editing.embedding_base_url ?? editing.base_url ?? "")
-          )
+          // Against the stored value only: the field is prefilled with the
+          // generation endpoint, and Core no longer falls back to it.
+          if (embeddingBaseUrl !== (editing.embedding_base_url ?? ""))
             patch.embedding_base_url = embeddingBaseUrl;
         } else if (
           embeddingModeChanged &&

memhub/apps/web/src/OnboardingModelController.tsx

@@ -173,10 +173,15 @@ export function OnboardingModelController({
         draft.embedding_mode === "local"
           ? "bge-small-zh-v1.5"
           : draft.embedding_model.trim(),
+      // Core no longer lets the embedding endpoint follow the generation
+      // one, so an empty field is written as the endpoint it means.
       embedding_base_url:
         draft.embedding_mode === "local"
           ? null
-          : draft.embedding_base_url.trim() || null,
+          : draft.embedding_base_url.trim() ||
+            (draft.embedding_provider_kind === "ollama"
+              ? "http://127.0.0.1:11434"
+              : draft.base_url.trim()),
       embedding_dimensions: draft.embedding_mode === "local" ? 512 : null,
     };
   }

memhub/docs/ARCHITECTURE.md

@@ -130,7 +130,7 @@ This exception does not extend to VDB, Coding, prompt, reconcile, or lifecycle.
 
 - Storage backends：仅 SDK config/env（Chroma/FAISS/Qdrant、Kuzu/Neo4j、cache/history sqlite|mysql、coding sqlite 等）。
 - Queue：pending `memory_write_jobs` 只表示尚未调用 SDK 的待处理请求，因此没有 SDK raw/memory ID。worker 单次调用 SDK `add()`；SDK 内部完成 raw 写入与 extraction/reconciliation，成功后 MemHub 仅将返回的 raw ID / memory IDs 记作 job outcome。
-- Raw retention：SDK L1 raw 只保留最近三天，由 Runtime 的 `prune_expired_raw` 执行；每个完整 turn 只保存真人 user 与 assistant，tool 不进入 raw。高层 Chat/Coding/Working memory 不受该窗口影响。
+- Raw retention：MemHub 不删除任何 SDK L1 raw，任何模式都全部保留（SDK 自身也不删 raw）；每个完整 turn 只保存真人 user 与 assistant，tool 不进入 raw。记忆模式（lite / pro / ultra）与 evolve 见 `docs/contracts/memory-mode-evolve-v1.md`。
 - Recall：Host Bridge / Core 先做宿主安装凭据与 tool grant，再调 **SDK search/get**。当前产品没有逐 memory 的 A-harness→B-harness 可见性开关，默认允许跨 harness 搜索；未来若增加来源限制，直接透传 SDK `source` / `include_sources` / `exclude_sources`，不在 MemHub 建 memory ACL。
 - Context wire：cutover marker 提交前只使用 legacy Context v1；提交后 Host Bridge 只使用 SDK-native Context v2。两种 schema 原样校验、原样返回，不把 SDK item 补造成 v1 的 `revision`、`evidence_ids`、`paths` 或 `adoption`，已选择的 v2 路径失败时也不回退 legacy 权威。
 - Ask wire：marker 前保留 Governance Ask v1；marker 后 Web / Host Bridge 只调用 SDK Ask v2。v2 的引用只含 SDK 原生 `memory_id/content/layer/status/source/gmt_created`，生成过程中重复 `get` 校验同一 SDK 快照，不补造 revision、Evidence 或 scopes，SDK 失败不回退 v1。

memhub/docs/cli/EXPERIENCE.md

@@ -142,6 +142,30 @@ node apps/cli/dist/one-memory.cjs write status <request_id> --user-id wife
 - 不带 `--user-id`（或写 `--user-id local`）时请求里不出现 `user_id`，和以前完全一样。
 - 用户之间互相隔离：读、改、删另一个用户的记忆 ID 返回「未找到」；write 的 request id 也只属于提交它的用户。
 
+### 画像版本（`profile --versions`）
+
+```bash
+node apps/cli/dist/one-memory.cjs profile --versions            # L0 画像版本链，最新在前，* 为当前版本
+node apps/cli/dist/one-memory.cjs profile --versions --user-id wife --json
+```
+
+用于找出覆盖了正确值的错误版本（例如 `用户` / `The user`），再用
+`memories delete <memory-id> --yes` 删除；对应 `GET /api/v2/user-profile/versions`。
+
+### 记忆模式与进化（`settings mode` / `evolve`）
+
+```bash
+node apps/cli/dist/one-memory.cjs settings mode                 # 当前模式
+node apps/cli/dist/one-memory.cjs settings mode ultra           # 只显示切换计划，不改动
+node apps/cli/dist/one-memory.cjs settings mode ultra --yes     # 切换
+node apps/cli/dist/one-memory.cjs evolve --user-id wife --wait  # 立即进化（仅 ultra）
+```
+
+- 切换不迁移数据：已处理的对话不会重做，排队中的和新的按新模式处理；正在写的一批按旧模式写完。
+- 任何模式都保留全部原始对话。
+- `status --json` 与 `progress --json` 输出 `pipeline.mode`、每晚进化时间和每个用户的上次进化。
+- 契约见 `docs/contracts/memory-mode-evolve-v1.md`。
+
 ## 8. 召回（recall）
 
 `recall` 两条路径（raw / memory）此前已真机验证。文本输出会标明每条是「原始对话摘录」

memhub/docs/contracts/memory-mode-evolve-v1.md

@@ -0,0 +1,346 @@
+# Memory mode, evolve and the nightly schedule v1
+
+Status: Core + CLI implemented (PR B of the SDK-surface batch). The App
+controls are built separately against this contract.
+
+All routes need the Runtime bearer token. Unknown query parameters and body
+fields answer `422 {"detail": "request_validation_failed"}`, like every
+memory route. `user_id` is optional (default `local`) and must match
+`^[A-Za-z0-9._-]{1,64}$`. The App never sends it.
+
+## 1. What the modes are
+
+| Mode | SDK client | What happens to a conversation | What search reads |
+|---|---|---|---|
+| `lite` | `mode="lite"` | Stored as one raw turn with one embedding. No model call; no generation model needs to be configured. | Raw turns only (the SDK forces `L1_RAW`). |
+| `pro` (default) | `mode="pro"` | Stored as raw, then a model extracts memories (System 1). The raw becomes SHADOW once extraction succeeds. | Extracted memories (facts, profile, intentions). |
+| `ultra` | `mode="ultra"` | Same as pro, and new facts are also projected into the knowledge graph. Every night, evolve runs System 2 for every user. | Same as pro, plus the graph-backed lanes below. |
+
+The product difference between pro and ultra is the nightly evolve. MemHub ultra
+uses SDK `mode="ultra"` because `client.evolve` refuses any other mode. Compared
+with pro, SDK ultra (hy-memory 2.0.9) also turns on the following, so the owner
+can judge their cost:
+
+- **The graph store.** Kuzu is always opened in ultra. Client init fails if the
+  graph cannot open, so `PUT ultra` answers `409 graph_store_unavailable` when
+  `kuzu` is missing. In pro it opens only when `MEMORY_ENABLE_GRAPH` is set, and
+  MemHub does not set it. Writes project new L2 facts into the graph.
+- **The schema lane (L6_SCHEMA).** The search's schema channel runs only when
+  a graph store exists, so in MemHub only in ultra. It returns the schemas that
+  evolve wrote (`schema_limit`, `schema_min_score`).
+- **Graph-aware ranking on the normal lane.** With a graph store, search adds
+  graph-entity (MENTIONS) boosts when the query names known entities and no
+  source filter is set. The heavier graph stages need the graph *and* their
+  own env flag, and every flag defaults off, so none of them run in MemHub:
+  graph rerank and episode expansion (`MEMORY_GRAPH_RERANK_ENABLED`), topic
+  graph entry, same-series bundle, hierarchy bundle.
+- **The profile.** `get_user_profile` appends the balanced catalog and the
+  specific supplement from the graph, when the SDK's balanced-catalog flag is
+  on.
+
+`ONE_MEMORY_GRAPH_STORE` (development only) still forces ultra whatever the
+setting says. Any value other than empty or `none` does this. The API reports
+it as `forced_by_env: true`, with `configured_mode` showing the saved choice.
+
+**Raw is never deleted.** In every mode MemHub keeps every raw turn the SDK
+stored, and the SDK never deletes raw itself. There is no retention window.
+
+## 2. Switching mode: what changes and what does not
+
+The SDK migrates nothing when the mode changes. MemHub applies these rules,
+and they are the owner's:
+
+- The mode is read **when a queued job runs**, not when it was queued.
+  - A job that already completed, in any mode, stays completed. It is never
+    reclaimed or reprocessed.
+  - Jobs still waiting (pending, assembling, waiting) and new input run in
+    the new mode.
+- **A batch in flight finishes in the mode it started in.** Its SDK client is
+  retired only when its last use ends (the #218 use count). The next claim
+  uses the new mode.
+- **Search follows the new mode at once.** The read client is rebuilt on the
+  next request (the active-profile read handle is keyed by profile and mode); the client it replaces closes once no request still uses it.
+- **lite → pro/ultra:**
+  - what lite stored stays raw and is not extracted;
+  - lite's raw-only search stops at once;
+  - nothing is re-ingested or backfilled.
+- **pro/ultra → lite:** memories made by the higher mode stay on disk, but lite
+  search does not serve them. Nothing is deleted.
+- **ultra → pro/lite:** the nightly evolve stops.
+- **→ ultra:**
+  - every known user's existing facts are added to the graph once
+    (`build_memory_graph(user_id, limit=10000, clear_first=False)`);
+  - this runs in the background, one user at a time, never beside that
+    user's write;
+  - progress is reported in `graph_build`.
+- Vector collection names never depend on the mode. The mode is a separate
+  client parameter, not part of the embedding-space identity.
+
+## 3. `GET /api/v1/settings/memory-mode`
+
+No parameters. `200` response:
+
+```json
+{
+  "mode": "pro",
+  "configured_mode": "pro",
+  "forced_by_env": false,
+  "modes": ["lite", "pro", "ultra"],
+  "graph_build": {
+    "state": "idle",
+    "started_at": null,
+    "finished_at": null,
+    "users": {}
+  }
+}
+```
+
+- `mode` is what SDK clients are built with now; `configured_mode` is the
+  saved setting.
+- `graph_build.state` is `idle | running | succeeded | failed`. Each entry in
+  `graph_build.users[<user_id>]` is
+  `{state: pending|running|succeeded|failed, memories_scanned, nodes, error_code}`.
+
+## 4. `PUT /api/v1/settings/memory-mode`
+
+Body: `{"mode": "lite" | "pro" | "ultra"}`. `200` response: the GET shape,
+plus these fields:
+
+```json
+{
+  "previous_mode": "pro",
+  "effects": ["existing_memories_added_to_graph"]
+}
+```
+
+`effects` is empty when the effective mode did not change. That happens when
+the mode was already set, or when `ONE_MEMORY_GRAPH_STORE` forces ultra; the
+saved setting still changes in that case. Otherwise `effects` lists these, in
+this order:
+
+| Effect | When | Meaning |
+|---|---|---|
+| `earlier_raw_not_extracted` | from lite | What lite stored stays raw; it is not extracted. |
+| `lite_search_stops` | from lite | lite's raw-only search ends now; search reads memories. |
+| `higher_mode_memories_hidden` | to a lower mode | Memories of the higher mode stay stored but are not shown. |
+| `evolve_stopped` | from ultra | The nightly evolve stops. |
+| `existing_memories_added_to_graph` | to ultra | The one-time graph backfill started (see `graph_build`). |
+| `graph_backfill_already_running` | to ultra | A backfill from an earlier switch is still running; no second one started. |
+
+Errors:
+
+- `409 {"detail": "graph_store_unavailable"}`: `ultra` was asked for, but the
+  graph store package is not installed. Nothing changed.
+- `422 request_validation_failed`: an unknown mode or an extra field.
+
+Show the planned effects before calling PUT. Clients can compute them with the
+table above: the CLI (`memhub settings mode <m>` without `--yes`) and Core
+(`memory_mode_effects`) use the same rule.
+
+## 5. Evolve
+
+### `POST /api/v2/evolve` → `202`
+
+Body (every field optional):
+
+```json
+{"user_id": "local", "agent_id": "all-apps", "max_batches": 3}
+```
+
+- `agent_id` is 1..128 characters. The default is `all-apps`, the agent
+  MemHub writes every memory under.
+- `max_batches` is 1..1000. The default is the SDK's own.
+
+The response is the run receipt below, with `state: "running"`. The run goes
+on in the background.
+
+- **One run per user.** Asking while that user's run is going answers the same
+  receipt, again `202`.
+- **Never beside that user's writes.** The run waits for the user's write in
+  flight. While it runs, that user's queued jobs wait: they are returned to
+  pending before any SDK call. Other users are not held.
+- **Loops while the SDK answers `should_continue: true`**, up to 20 rounds. A
+  run stopped by that cap has `capped: true`; start it again to continue.
+
+Errors:
+
+- `409 {"detail": "memory_mode_not_ultra"}`: the effective mode is not ultra.
+- `422`: a malformed `user_id`, `max_batches` outside its range, or an extra
+  field.
+
+### Run receipt
+
+```json
+{
+  "run_id": "uuid",
+  "user_id": "local",
+  "agent_id": "all-apps",
+  "max_batches": null,
+  "trigger": "manual",
+  "state": "succeeded",
+  "started_at": "2026-11-01T03:00:00Z",
+  "finished_at": "2026-11-01T03:00:42Z",
+  "rounds": 2,
+  "tasks_processed": 5,
+  "batch_failures": 0,
+  "cycle_status": "complete",
+  "should_continue": false,
+  "continuation_reason": null,
+  "remaining_clusters": 0,
+  "remaining_actionable_clusters": 0,
+  "remaining_fresh_facts": 0,
+  "capped": false,
+  "elapsed_ms": 41873.2,
+  "error_code": null,
+  "failure_kind": null,
+  "request_id": "sdk-request-id",
+  "results": {},
+  "cross_domain_sweeper": {}
+}
+```
+
+- `trigger` is `manual` or `scheduled`.
+- `state` is `running`, `succeeded` or `failed`.
+- `cycle_status` is the SDK's `partial` or `complete`.
+- `error_code` and `failure_kind` are stable codes on failure. Examples:
+  `memory_evolve_cancelled` (the Runtime stopped mid-run),
+  `hy_memory_evolve_rejected` (the SDK answered `success: false` without a
+  code), or the SDK's own `error_code`. Provider messages never appear.
+- `results` and `cross_domain_sweeper` are the last round's SDK bodies. They
+  live in memory only and are never written to disk or diagnostics.
+
+### `GET /api/v2/evolve/{run_id}?user_id=` → receipt
+
+Returns `404 {"detail": "evolve_run_not_found"}` in two cases:
+
+- the run belongs to another user (answered exactly like a missing run);
+- the Runtime restarted since the run. Receipts live in memory, and the last
+  64 are kept.
+
+### `GET /api/v2/evolve?user_id=` → overview
+
+```json
+{
+  "mode": "ultra",
+  "schedule": {"enabled": true, "time": "03:00"},
+  "users": {
+    "local": {
+      "last_run_at": "2026-11-01T03:00:00Z",
+      "last_result": {"state": "succeeded", "rounds": 2, "...": "receipt without results"},
+      "last_error": null
+    }
+  },
+  "running": []
+}
+```
+
+- Without `user_id` the overview covers every user; with it, that one user.
+- `users` survives restarts. It is kept in `<data_dir>/evolve-state.json`,
+  counts and codes only.
+
+## 6. Nightly schedule
+
+### `GET /api/v1/settings/evolve-schedule`
+
+Returns `{"enabled": true, "time": "03:00"}`. `time` is `HH:MM`, 24-hour, in
+the machine's local time.
+
+### `PUT /api/v1/settings/evolve-schedule`
+
+Body: `{"enabled"?: bool, "time"?: "HH:MM"}`. An absent field keeps its
+value. The response is the saved schedule. A bad time answers `422`.
+
+How the schedule runs:
+
+- It runs only in ultra.
+- It checks every 5 minutes. A user's run is due when either holds:
+  - today's slot has passed, and the user's last run is older than that slot;
+  - the user's last run is more than 24 hours old.
+
+  The second rule is how a machine that slept through 03:00 catches up when
+  it wakes.
+- A user who never ran is measured from the first time the schedule ran in
+  ultra, not from the epoch. So switching to ultra at noon does not evolve
+  everyone at once; their first run is at the next slot.
+- Users run one after another.
+
+The schedule runs every user that has memories. The SDK cannot list users, so
+Core takes them from its own records:
+
+- `local`;
+- every user holding a source grant (a Hermes profile);
+- every user whose turns or notes passed the write queue.
+
+A user is kept when `memory_stats` reports any memory or fact for them.
+
+## 7. Status fields (R10)
+
+`GET /api/v1/memory-processing` has a new optional field, `pipeline`:
+
+```json
+{
+  "pipeline": {
+    "mode": "ultra",
+    "configured_mode": "ultra",
+    "forced_by_env": false,
+    "evolve": {"...": "the overview of section 5"},
+    "last_evolve_at": "2026-11-01T03:00:00Z",
+    "graph_build": {"...": "as in section 3"}
+  }
+}
+```
+
+`memhub status --json` adds a top-level projection of it:
+
+```json
+{
+  "pipeline": {
+    "mode": "ultra",
+    "configured_mode": "ultra",
+    "forced_by_env": false,
+    "last_evolve_at": "...",
+    "evolve_schedule": {"enabled": true, "time": "03:00"},
+    "evolve_running": 0,
+    "evolve_users": {
+      "local": {"last_run_at": "...", "last_error": null, "state": "succeeded"}
+    },
+    "graph_build": "succeeded"
+  }
+}
+```
+
+`memhub progress --json` carries the same `pipeline`. Both are `null` against
+an older Core.
+
+## 8. CLI
+
+```bash
+memhub settings mode                    # show the current mode
+memhub settings mode lite               # print the plan, change nothing
+memhub settings mode lite --yes         # switch
+memhub evolve [--user-id <id>] [--agent-id <id>] [--max-batches <n>] [--wait] [--json]
+```
+
+## 9. Copy the UI can reuse
+
+| Key | zh-CN | en | es |
+|---|---|---|---|
+| title | 记忆模式 | Memory mode | Modo de memoria |
+| lite | lite：不调用模型，保留全部原始对话，只在原始对话里检索。 | lite: no model calls. Every original conversation is kept, and search reads only those conversations. | lite: sin llamadas al modelo. Se conservan todas las conversaciones originales y la búsqueda solo las consulta a ellas. |
+| pro | pro：由模型从对话中提取记忆（System 1）。 | pro: a model extracts memories from your conversations (System 1). | pro: un modelo extrae memorias de tus conversaciones (System 1). |
+| ultra | ultra：在 pro 之上每晚自动整合记忆（System 2，evolve），并使用知识图谱。 | ultra: pro plus nightly memory consolidation (System 2, evolve), using a knowledge graph. | ultra: pro más una consolidación nocturna de las memorias (System 2, evolve), con un grafo de conocimiento. |
+| confirm | 切换到 {mode} 前请确认： | Before switching to {mode}, please confirm: | Antes de cambiar a {mode}, confirma lo siguiente: |
+| earlier_raw_not_extracted | lite 下已处理完的对话不会再被提取；只有尚未处理的和之后的新对话会由模型整理成记忆。 | Conversations already handled in lite are not extracted again; only conversations still waiting, and new ones, become memories. | Las conversaciones ya procesadas en lite no se vuelven a extraer; solo las pendientes y las nuevas se convierten en memorias. |
+| lite_search_stops | 只在原始对话里的检索会立即停止，改为检索记忆。 | Search over original conversations stops now; search reads memories instead. | La búsqueda en las conversaciones originales se detiene ahora; se buscará en las memorias. |
+| higher_mode_memories_hidden | 更高模式生成的记忆仍然保存，但在这个模式下不会显示。 | Memories created by a higher mode stay stored but are not shown in this mode. | Las memorias creadas por un modo superior se conservan, pero no se muestran en este modo. |
+| evolve_stopped | 每晚的自动整合（System 2）会停止。 | Nightly consolidation (System 2) stops. | La consolidación nocturna (System 2) se detiene. |
+| existing_memories_added_to_graph | 已有记忆会在后台加入知识图谱。 | Your existing memories are added to the knowledge graph in the background. | Tus memorias actuales se añaden al grafo de conocimiento en segundo plano. |
+| graph_backfill_already_running | 知识图谱正在补建中。 | The knowledge graph is still being built. | El grafo de conocimiento todavía se está construyendo. |
+| raw_kept | 原始对话全部保留，任何模式都不删除。 | Every original conversation is kept, in every mode. | Se conservan todas las conversaciones originales, en cualquier modo. |
+| graph_build_running | 正在把已有记忆加入知识图谱… | Adding existing memories to the knowledge graph… | Añadiendo las memorias actuales al grafo de conocimiento… |
+| forced_by_env | 开发环境变量 ONE_MEMORY_GRAPH_STORE 正在强制使用 ultra。 | The development variable ONE_MEMORY_GRAPH_STORE is forcing ultra. | La variable de desarrollo ONE_MEMORY_GRAPH_STORE está forzando ultra. |
+| graph_store_unavailable | 这个安装缺少知识图谱组件，无法切换到 ultra。 | This installation lacks the knowledge-graph component, so it cannot switch to ultra. | A esta instalación le falta el componente del grafo de conocimiento; no puede cambiar a ultra. |
+| schedule | 每天 {time} 自动整合 | Consolidates daily at {time} | Consolida cada día a las {time} |
+| last_evolve | 上次整合：{time} | Last consolidation: {time} | Última consolidación: {time} |
+| never_evolved | 尚未整合 | Not consolidated yet | Aún no consolidado |
+| evolve_not_ultra | 记忆整合只在 ultra 模式下可用。 | Consolidation is available in ultra mode only. | La consolidación solo está disponible en el modo ultra. |

memhub/docs/contracts/sdk-data-plane-migration-v1.md

@@ -26,8 +26,7 @@ and no raw turn, and does not compare the mirror with the SDK:
   cannot be trusted to decide what the SDK should contain;
 - the product has no user editing of memories in practice, so the mirror holds
   no user intent the SDK lacks;
-- SDK L1 raw is retained for three days by the Runtime's own
-  `prune_expired_raw`, not by the migration.
+- SDK L1 raw is kept: neither the migration nor the Runtime deletes raw.
 
 "Already complete" holds only if the new Runtime opens the same stores. Builds
 before #145 named a profile's vector collections `one_memory_<12 hex>…`; #145
@@ -77,25 +76,33 @@ then:
 2. **prepare**: applies the management schema migrations, discards SDK staging
    left by builds through 19 (§5) and makes legacy queue work self-contained
    (§4);
-3. **vectors**: for the active embedding profile, renames every
+3. **vectors**: for the active embedding profile (the built-in model when no
+   profile is active), renames every
    `one_memory_<old hash>…` collection to the current `memhub_<hash>…` name
    through Chroma's public client. A current collection that already holds
    writes is merged in first (newer record wins on a shared id); counts are
    checked after each step;
 4. **verify**: opens the public SDK client exactly as the Runtime will and reads
    the legacy confirmed memory ids back (`list_memories`, then `get`). Every
-   chat memory must read back. Coding/Working may be fewer, because the SDK's
-   Coding reconciler deletes experiences it retires, but not zero when the
-   legacy mirror has some. Otherwise the run stops with
-   `sdk_migration_memory_unreadable` and the counts, before anything
-   irreversible; legacy authority and tables stay intact;
+   confirmed chat memory that was in the vector store before this run (any
+   collection, read before step 3) must read back. A confirmed chat the SDK had
+   already merged or deleted before the upgrade cannot, and is only counted
+   (`chat_missing_before_migration`); if none of them is in the store, none is
+   excused. Coding/Working may be fewer, because the SDK's Coding reconciler
+   deletes experiences it retires, but not zero when the legacy mirror has
+   some. Otherwise the run stops with `sdk_migration_memory_unreadable` and
+   the counts, before anything irreversible; legacy authority and tables stay
+   intact;
 5. **cutover**: re-verifies the bundle, repeats §4, commits SDK authority;
 6. **cleanup**: drops the legacy tables (including the `evidence_search`
    index) and runs `VACUUM` (skipped if the disk is too full).
 
-A home with legacy memories but no active model profile is not migrated
-(`sdk_migration_model_profile_required`): its collections can be neither named
-nor read back.
+A home with no active model profile has no generation model, but its vectors
+are the built-in embedding model's, the same space the Runtime reads with; it
+migrates through that space and no key is resolved. Only when its legacy
+memories have vectors solely under another embedding space (a profile since
+deleted) is it not migrated (`sdk_migration_vector_profile_missing`, in the
+preflight, before anything changes).
 
 A home already on SDK authority still reports `required` while legacy-named
 vector collections exist for its profile (build 20 homes). `sdk-migrate` then

memhub/packages/contracts/src/index.ts

@@ -636,6 +636,16 @@ export interface SdkRawRecord {
   memory_at: number | null;
   gmt_created: number | null;
   custom: Record<string, unknown>;
+  /** active, or shadow once pro extracted it. */
+  status?: string | null;
+  /** The turn's messages, parsed from the SDK's `[role]: text` lines. */
+  messages?: { role: string; content: string }[];
+  /** The one role of a single-message turn; null for a mixed turn. */
+  role?: string | null;
+  /** `sha256:<hex>` of the stored content, computed by Core. */
+  content_hash?: string;
+  /** ISO time the SDK stored the raw. */
+  created?: string | null;
 }
 
 export interface SdkRawPage {
@@ -643,6 +653,37 @@ export interface SdkRawPage {
   total: number;
   limit: number;
   offset: number;
+  /** A time range was applied to this page in Core (the SDK list has none). */
+  time_filtered?: boolean;
+}
+
+/** `GET /api/v2/memory-library/layers`: memories per SDK layer. */
+export interface SdkLayerCounts {
+  layers: Record<string, number>;
+  total: number;
+  statuses: string[] | null;
+}
+
+/** `GET /api/v2/user-profile/versions`: the L0 basic-info version chains. */
+export interface SdkProfileVersions {
+  user_id: string;
+  request_id: string;
+  elapsed_ms: number;
+  chains: {
+    memory_id: string;
+    content: string;
+    status: string;
+    memory_at: number | null;
+    gmt_created: number | null;
+    basic_info: Record<string, unknown>;
+    versions: {
+      memory_id: string;
+      content: string;
+      memory_at: number | string | null;
+      gmt_created: number | null;
+      current: boolean;
+    }[];
+  }[];
 }
 
 export interface FileImportReceipt {