Compare commits

..
Author SHA1 Message Date
alvinfeiandCursor c4da617d59 fix(migration): stage SDK data before cutover
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-22 20:39:09 +08:00
alvinfeiandCursor 729e699eab refactor(sdk): keep legacy migration out of client APIs
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-22 20:31:22 +08:00
79 changed files with 4764 additions and 4051 deletions

No files matched your search

+61 -84
View File
@@ -2,73 +2,6 @@
Hy Memory and MemHub are versioned and released independently.
## Unreleased
- Fixed legacy-to-SDK startup migration to adopt native Coding Experience and
Working Fact IDs instead of importing them as Chat layers. Raw migration now
rebuilds bounded user/assistant turns from the latest three days, excludes
tools, and enforces the same rolling retention after cutover.
- Paused host capture writes with a live Desktop migration fence; Cursor
History catch-up resumes from its existing checkpoint after cutover.
- Removed MemHub's private `hy_memory_bridge` implementation. MemHub now calls
public `HyMemoryClient` methods at runtime; prompt/reconcile/response
monkey-patches and private VDB/Coding runtime access were deleted. Kuzu Graph
projection/admin and stopped-writer App migration are the documented product
exceptions.
- Added a MemHub-managed native Hermes MemoryProvider with automatic prefetch,
persistent per-session memory-ID deduplication, authorized Broker reads, and
recall-context stripping to prevent memory recursion.
- Added an extensible MemHub interface-localization layer with Simplified
Chinese, English, and Spanish catalogs shared by Web and Electron-native
surfaces. Interface language remains independent from memory content and
model processing language.
- Fixed governed-memory revisions and repeated evidence links failing when the
visibility dirty-marker trigger encounters the same memory twice. Startup
now replaces the affected legacy trigger without changing schema versions
or governed and visibility data.
- Kept post-write authority verification failures available for read-only
reconciliation while preventing automatic or generic confirmed retries from
invoking the model again. Automatic reconciliation skips those sticky rows
before applying its batch limit, and linked durable jobs exclusively own
retry scheduling for their memory tasks.
- Removed MemHub's task-scoped HyPermission store wrappers and global
model-content consent switch. The trusted embedded Hy Memory SDK now uses
its native local stores directly.
- Existing installations ignore the retired config flag, readmit previously
parked pre-invocation work, and drop only the obsolete
`hy_node_permissions` proof table. Evidence, governed memories, model
profiles, SDK stores, and source/host grants are preserved.
## Hy Memory 2.0.5 — 2026-10-22
- Published
[`hy-memory==2.0.5`](https://pypi.org/project/hy-memory/2.0.5/).
- Extended `get()` across Chat, Coding Experience, Working Fact, and raw IDs;
the returned `layer`, `kind`, and `variant` identify the actual record.
- Added explicit raw retrieval through the existing `search()` and
`list_memories()` contracts, including user, agent, session, and source
filters.
- Added public embedding-service injection and corrected extraction language
detection for structured OpenAI-style messages.
- Added lifecycle status filters, durable idempotent write outcomes, and
quiesced local data-plane snapshot/restore.
- Kept legacy migration mechanics out of the public Client API; separate
migration-only raw/import methods from development previews were removed.
- Physical delete now reports partial VDB/Graph cleanup as failure.
- Coding search no longer injects the version-evolution diff block by default;
opt in with `MEMORY_CODING_CHAIN_BLOCK_ENABLED=true`.
- Serialized SDK SQLite cache/history connection lifecycle and tightened
Coding reconciliation to its agent/session/workspace authority.
- Clarified persistent caller context through `meta`, while `metadata` remains
legacy processing configuration.
## Hy Memory 2.0.5.dev4 — 2026-10-21
- Development preview used for SDK-authority migration validation; not the
final `2.0.5` release.
- Published as
[`hy-memory==2.0.5.dev4`](https://pypi.org/project/hy-memory/2.0.5.dev4/).
## Hy Memory 2.0.5.dev3 — 2026-10-21
- Development preview; not the final `2.0.5` release.
@@ -95,12 +28,6 @@ Hy Memory and MemHub are versioned and released independently.
APIs, offline local data-plane snapshot/restore, and fail-closed physical
delete receipts.
## Hy Memory 2.0.4 — 2026-10-17
- Published
[`hy-memory==2.0.4`](https://pypi.org/project/hy-memory/2.0.4/) as a
maintenance release.
## Hy Memory 2.0.3 — 2026-10-17
- Added an optional runtime HTTP transport factory so embedded products can
@@ -111,6 +38,66 @@ Hy Memory and MemHub are versioned and released independently.
unchanged memory extraction, reconciliation, storage, and Coding/Working
behavior.
## Unreleased
- Removed MemHub's private `hy_memory_bridge` implementation. MemHub now calls
public `HyMemoryClient` methods at runtime; prompt/reconcile/response
monkey-patches and private VDB/Coding runtime access were deleted. Kuzu Graph
projection/admin and stopped-writer App migration are the documented product
exceptions.
- Extended public SDK `get()` to resolve Chat, Coding Experience, and Working
Fact IDs, including raw records, owner/session fields, and explicit ambiguity
handling; added constructor-time public embedding-service injection.
- Added a MemHub-managed native Hermes MemoryProvider with automatic prefetch,
persistent per-session memory-ID deduplication, authorized Broker reads, and
recall-context stripping to prevent memory recursion.
- Added an extensible MemHub interface-localization layer with Simplified
Chinese, English, and Spanish catalogs shared by Web and Electron-native
surfaces. Interface language remains independent from memory content and
model processing language.
- `search(..., layers=["l1_raw"])` now explicitly searches SDK raw records in
pro/ultra mode and keeps existing user/agent/session/source filters; default
search continues to hide raw.
- Added SDK-authoritative lifecycle listing with explicit status filters,
idempotent durable write outcomes, and quiesced local data-plane
snapshot/restore contracts for the MemHub authority migration. Exact legacy
backfill is now an App-owned privileged offline operation rather than a
public `HyMemoryClient` API. Backfill runs against a complete staging copy;
cutover atomically swaps SDK roots and restores the previous root on failure.
- Physical delete now reports incomplete VDB/Graph cleanup as a partial failure
instead of claiming success after only one store was changed.
- Coding search no longer injects the version-evolution diff block
(`chain_block`) by default. Enable with
`MEMORY_CODING_CHAIN_BLOCK_ENABLED=true` or
`CodingConfig.chain_block_enabled=True` (write-side observe unchanged).
- Fixed governed-memory revisions and repeated evidence links failing when the
visibility dirty-marker trigger encounters the same memory twice. Startup
now replaces the affected legacy trigger without changing schema versions
or governed and visibility data.
- Serialized each SDK SQLite store's shared connection lifecycle so concurrent
history/cache operations cannot race with `close()` and crash the process.
- Scoped Coding reconciliation targets to the current HY authority,
conversation, and workspace, preventing near-duplicate sessions from
mutating an experience or working fact owned by another evidence context.
- Kept post-write authority verification failures available for read-only
reconciliation while preventing automatic or generic confirmed retries from
invoking the model again. Automatic reconciliation skips those sticky rows
before applying its batch limit, and linked durable jobs exclusively own
retry scheduling for their memory tasks.
- Clarified that `add(metadata=...)` carries legacy SDK processing options,
while persistent caller business context belongs in `meta`.
- Unified exact reads and lists across memory layers: `get(memory_id)` returns
the matched layer, while `list_memories(layers=["l1_raw"])` explicitly lists
raw records. Separate raw get/list methods and the `include_raw` switch were
removed.
- Removed MemHub's task-scoped HyPermission store wrappers and global
model-content consent switch. The trusted embedded Hy Memory SDK now uses
its native local stores directly.
- Existing installations ignore the retired config flag, readmit previously
parked pre-invocation work, and drop only the obsolete
`hy_node_permissions` proof table. Evidence, governed memories, model
profiles, SDK stores, and source/host grants are preserved.
## Hy Memory 2.0.2 — 2026-10-17
- Published
@@ -125,17 +112,7 @@ Hy Memory and MemHub are versioned and released independently.
Because PyPI release metadata is immutable, the next SDK patch release must
carry the updated license metadata; `2.0.1` retains its published metadata.
## MemHub 1.0.5 — Unreleased
- Updated the source version to `1.0.5` (build 14).
- Pinned the public SDK to the stable `hy-memory==2.0.5` release.
## MemHub 1.0.4 — 2026-10-18
- Updated the source version to `1.0.4` (build 13).
- Published the `v1.0.4` source and release metadata.
## MemHub 1.0.3 — 2026-10-17
## MemHub 1.0.3 — Unreleased
- Updated the source version to `1.0.3` (build 7).
- Pinned the public SDK to `hy-memory==2.0.3`.
-2
View File
@@ -13,8 +13,6 @@
接口先补 SDK。Runtime writers 停止后的 App 启动迁移可直接操作底层存储,
但不得把迁移专用方法加入公共 Client。
3. MemHub **不存 memory 实体**;**可以存 `memory_id`** 做组织/回执。原文进入 **SDK raw**。
4. SDK raw 只保留滚动三天;一条 raw 对应一个完整 turn,只含真人 user 与
assistant,tool 不持久化。提取后的高层 memory 不受该窗口影响。
生产入口通过共享 Runtime 租约取得公开 `HyMemoryClient`;MemHub 只在调用前执行
模型配置、密钥和 harness/model 门禁,不维护第二套 memory 正文权威。
+5 -5
View File
@@ -18,13 +18,13 @@ If you run into a problem, please open an Issue with the version, platform,repro
## 发行与平台合同
源码当前固定使用公开 PyPI 正式版 `hy-memory==2.0.5`。MemHub 源码版本为 `1.0.5`
(build 14),GitHub 仓库为
源码当前固定使用公开 PyPI 开发预览 `hy-memory==2.0.5.dev3`。MemHub 源码版本为 `1.0.4`
(build 13),GitHub 仓库为
[`Tencent-Hunyuan/Hy-Memory`](https://github.com/Tencent-Hunyuan/Hy-Memory),npm 包名为
`memhub-cli`。当前代码架构见
[`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md)。
`v1.0.5` 的发行目标是 Linux x64/arm64 Runtime 与 macOS arm64/x64 App 和 Runtime。
`v1.0.4` 的发行目标是 Linux x64/arm64 Runtime 与 macOS arm64/x64 App 和 Runtime。
Linux 制品由 GitHub Actions 在原生架构 runner 上构建并完成生命周期验收;macOS
制品由发行流水线完成 Developer ID 签名、公证和验收后手动上传。Windows 不支持。
只有同时出现在 GitHub Release 与 `release-manifest.json` 中、且通过公开校验的制品才属于
@@ -45,7 +45,7 @@ Linux 制品由 GitHub Actions 在原生架构 runner 上构建并完成生命
## 仓库结构
- `apps/core`:Python / FastAPI Kernel,负责授权、模型、队列、SDK Runtime 边界与 API。
- `apps/core`:Python / FastAPI Kernel,负责授权、模型、队列、SDK bridge 与 API。
- `apps/cli`:安装、控制和诊断共享 Runtime 的 `one-memory` CLI。
- `apps/web`:React / TypeScript 管理界面。
- `apps/desktop`:Electron 桌面壳与图形安装入口。
@@ -117,7 +117,7 @@ MemHub 的 SQLite 数据默认位于 `~/.one-memory`,不在 npm 包或 Runtime
## Hy Memory
SDK 依赖固定为公开 PyPI 正式版 `hy-memory==2.0.5`。SDK 配置、权限边界及 search
SDK 依赖固定为公开 PyPI 开发预览 `hy-memory==2.0.5.dev3`。SDK 配置、权限边界及 search
返回格式见 [`MEMORY.md`](MEMORY.md)。HY search 返回的 `content` 已包含 SDK 统一生成的
逻辑时间和演化上下文;调用方不应再次转换时间戳或重复拼接 `When`。
+3 -3
View File
@@ -3,9 +3,9 @@
"dataLicense": "CC0-1.0",
"SPDXID": "SPDXRef-DOCUMENT",
"name": "one-memory",
"documentNamespace": "https://one-memory.local/spdx/2026-10-22T13:18:16Z",
"documentNamespace": "https://one-memory.local/spdx/2026-10-22T08:09:31Z",
"creationInfo": {
"created": "2026-10-22T13:18:16Z",
"created": "2026-10-22T08:09:31Z",
"creators": [
"Tool: one-memory-generate-sbom"
]
@@ -2352,7 +2352,7 @@
{
"SPDXID": "SPDXRef-Package-hy-memory",
"name": "hy-memory",
"versionInfo": "2.0.5",
"versionInfo": "2.0.5.dev3",
"downloadLocation": "NOASSERTION",
"filesAnalyzed": false,
"licenseConcluded": "Apache-2.0",
+2 -2
View File
@@ -2,14 +2,14 @@
MemHub itself is licensed under Apache-2.0 (see `LICENSE`).
It bundles the runtime dependencies listed below.
Generated from the installed dependency tree on 2026-10-22T13:18:16Z
Generated from the installed dependency tree on 2026-10-22T08:09:31Z
by `python3 scripts/generate_sbom.py`. Do not edit by hand.
Per-package license texts are not reproduced here yet; the license
column is filled in as part of release review, and the SBOM records
`NOASSERTION` until then rather than guessing.
## hy-memory 2.0.5
## hy-memory 2.0.5.dev3
License: Apache-2.0
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/cli",
"version": "1.0.5",
"version": "1.0.4",
"description": "MemHub bootstrap CLI for installing and managing the local-first shared memory runtime",
"license": "Apache-2.0",
"repository": {
+1 -42
View File
@@ -1,7 +1,6 @@
import { mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises";
import { mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { inflateRawSync } from "node:zlib";
import { DiagnosticLogger } from "@one-memory/runtime-control";
import { afterEach, expect, it, vi } from "vitest";
@@ -30,27 +29,6 @@ async function fixture() {
return dataDir;
}
function readZipEntry(archive: Buffer, expected: string): string {
let offset = 0;
while (archive.readUInt32LE(offset) === 0x04034b50) {
const method = archive.readUInt16LE(offset + 8);
const compressedSize = archive.readUInt32LE(offset + 18);
const nameLength = archive.readUInt16LE(offset + 26);
const extraLength = archive.readUInt16LE(offset + 28);
const nameStart = offset + 30;
const dataStart = nameStart + nameLength + extraLength;
const name = archive.subarray(nameStart, nameStart + nameLength).toString();
const compressed = archive.subarray(dataStart, dataStart + compressedSize);
if (name === expected) {
if (method === 0) return compressed.toString();
if (method === 8) return inflateRawSync(compressed).toString();
throw new Error("unsupported zip compression");
}
offset = dataStart + compressedSize;
}
throw new Error(`missing ${expected}`);
}
it("prints matching diagnostic events as NDJSON without starting Runtime", async () => {
const dataDir = await fixture();
let stdout = "";
@@ -96,25 +74,6 @@ it("exports one support zip to the requested downloads directory", async () => {
});
expect(receipt.path).toMatch(/MemHub-日志-3800-\d{8}-\d{4}\.zip$/);
expect(await readdir(downloads)).toEqual([path.basename(receipt.path)]);
const archive = await readFile(receipt.path);
expect(JSON.parse(readZipEntry(archive, "summary.json")).commit).toBeNull();
});
it("fails closed for an explicitly supplied invalid commit", async () => {
const dataDir = await fixture();
const downloads = await mkdtemp(path.join(os.tmpdir(), "om-cli-downloads-"));
roots.push(downloads);
vi.spyOn(process.stdout, "write").mockImplementation(() => true);
await expect(
run(["logs", "export", "--json"], {
ONE_MEMORY_DATA_DIR: dataDir,
ONE_MEMORY_EXPORT_DIR: downloads,
ONE_MEMORY_BUILD_NUMBER: "3800",
ONE_MEMORY_COMMIT: "unknown",
}),
).rejects.toThrow("diagnostic_commit_invalid");
expect(await readdir(downloads)).toEqual([]);
});
it("never emits a secret from a malformed manually written event", async () => {
-108
View File
@@ -57,11 +57,6 @@ const runtimeMocks = vi.hoisted(() => ({
})),
uninstallUserService: vi.fn(async (..._args: unknown[]) => "removed"),
restartUserService: vi.fn(async () => undefined),
runSdkMigration: vi.fn(
async (): Promise<{ state: string; code?: string }> => ({
state: "not_required",
}),
),
rollbackLocalInstall: vi.fn(async () => undefined),
planCliLink: vi.fn(async ({ home, installRoot }) => ({
schema_version: 1,
@@ -131,7 +126,6 @@ vi.mock("@one-memory/runtime-control", async (importOriginal) => {
installUserService: runtimeMocks.installUserService,
uninstallUserService: runtimeMocks.uninstallUserService,
restartUserService: runtimeMocks.restartUserService,
runSdkMigration: runtimeMocks.runSdkMigration,
rollbackLocalInstall: runtimeMocks.rollbackLocalInstall,
planCliLink: runtimeMocks.planCliLink,
installCliLink: runtimeMocks.installCliLink,
@@ -895,108 +889,6 @@ describe("public One Memory CLI", () => {
expect(runtimeMocks.restartUserService).toHaveBeenCalledTimes(2);
});
const upgradeArguments = [
"upgrade",
"--source",
"/tmp/dist",
"--home",
"/tmp/home",
"--install-root",
"/tmp/install",
"--data-dir",
"/tmp/custom data",
"--yes",
];
const presentCliLink = {
schema_version: 1 as const,
platform: "linux" as const,
home: "/tmp/home",
install_root: "/tmp/install",
bin_directory: "/tmp/bin",
status: "present" as const,
blocked_reason: null,
link_path: "/tmp/bin/one-memory",
target_path: "/tmp/install/current/bin/one-memory",
};
it("migrates the home with the new Core before restarting the Runtime", async () => {
vi.clearAllMocks();
runtimeMocks.planCliLink.mockResolvedValueOnce(presentCliLink);
const order: string[] = [];
runtimeMocks.runSdkMigration.mockImplementationOnce(async () => {
order.push("migrate");
return { state: "cutover" };
});
runtimeMocks.restartUserService.mockImplementationOnce(async () => {
order.push("restart");
});
const output = captureStdout();
let code: number;
try {
code = await run(
upgradeArguments,
{},
{ waitForRuntimeReady: async () => true },
);
} finally {
output.restore();
}
expect(code!).toBe(0);
expect(order).toEqual(["migrate", "restart"]);
expect(runtimeMocks.runSdkMigration).toHaveBeenCalledWith(
expect.objectContaining({ dataDir: "/tmp/custom data" }),
);
});
it("rolls back to the previous version when the migration does not finish", async () => {
vi.clearAllMocks();
runtimeMocks.planCliLink.mockResolvedValueOnce(presentCliLink);
runtimeMocks.runSdkMigration.mockResolvedValueOnce({
state: "deferred",
code: "sdk_migration_disk_space_insufficient",
});
const streams = captureStreams();
let code: number;
try {
code = await run(
upgradeArguments,
{},
{ waitForRuntimeReady: async () => true },
);
} finally {
streams.restore();
}
expect(code!).toBe(1);
expect(JSON.parse(streams.stderr())).toMatchObject({
command: "upgrade",
code: "sdk_migration_disk_space_insufficient",
});
expect(runtimeMocks.rollbackLocalInstall).toHaveBeenCalledOnce();
});
it("keeps the new version when a migrated home is slow to come up", async () => {
vi.clearAllMocks();
runtimeMocks.planCliLink.mockResolvedValueOnce(presentCliLink);
runtimeMocks.runSdkMigration.mockResolvedValueOnce({ state: "cutover" });
const streams = captureStreams();
let code: number;
try {
code = await run(
upgradeArguments,
{},
{ waitForRuntimeReady: async () => false },
);
} finally {
streams.restore();
}
expect(code!).toBe(1);
expect(JSON.parse(streams.stderr())).toMatchObject({
command: "upgrade",
code: "runtime_not_ready_after_sdk_migration",
});
expect(runtimeMocks.rollbackLocalInstall).not.toHaveBeenCalled();
});
it("reports success once the upgraded Runtime is healthy", async () => {
vi.clearAllMocks();
runtimeMocks.planCliLink.mockResolvedValueOnce({
+8 -139
View File
@@ -39,9 +39,7 @@ import {
rollbackCliLink,
rollbackLocalInstall,
runRuntimeCli,
runSdkMigration,
runtimeInstallLayout,
type SdkMigrationResult,
stopRuntimeProcess,
} from "@one-memory/runtime-control";
import axios from "axios";
@@ -356,9 +354,6 @@ export interface RunControls {
/** Wait for a restarted Runtime to become healthy after `upgrade`.
* Injectable so the upgrade tests do not spend real seconds polling. */
waitForRuntimeReady?: (dataDir: string) => Promise<boolean>;
/** Run the SDK data migration before install/upgrade starts the new
* Runtime. Injectable so the tests do not spawn a real Core. */
migrateSdkData?: typeof migrateSdkDataBeforeStart;
restartManagedRuntime?: typeof restartManagedService;
stopRuntime?: typeof stopRuntimeProcess;
ensureRuntime?: typeof ensureRuntimeProcess;
@@ -1053,21 +1048,15 @@ async function runCommand(
entry: "cli",
}));
const target = path.join(downloads, diagnosticZipFilename(build));
const environmentCommit = environment.ONE_MEMORY_COMMIT?.trim();
const embeddedCommit =
typeof __ONE_MEMORY_COMMIT__ === "string"
? __ONE_MEMORY_COMMIT__.trim()
: "";
const commit =
environmentCommit ||
(embeddedCommit && embeddedCommit !== "unknown"
? embeddedCommit
: undefined);
const receipt = await exportDiagnosticZip({
dataDir: config.dataDir,
target,
productVersion: build,
...(commit === undefined ? {} : { commit }),
commit:
environment.ONE_MEMORY_COMMIT?.trim() ||
(typeof __ONE_MEMORY_COMMIT__ === "string"
? __ONE_MEMORY_COMMIT__
: "unknown"),
coreSha256: await sha256File(config.coreCommand).catch(() => undefined),
home: os.homedir(),
crashReportsDir: path.join(
@@ -3154,7 +3143,6 @@ async function runCommand(
null;
let installReceiptWritten = false;
let serviceApplied = false;
let migrated = false;
let previousOwner:
| Awaited<ReturnType<typeof claimCliServiceOwnership>>
| undefined;
@@ -3165,22 +3153,6 @@ async function runCommand(
}
shellPathReceipt = await applyShellPath(shellPath);
previousOwner = await claimCliServiceOwnership(layout.dataDir);
// `install` over an existing home (reinstall, App-to-CLI takeover) must
// migrate it too, before the new Runtime starts serving it.
const migration = await (
controls.migrateSdkData ?? migrateSdkDataBeforeStart
)({
coreCommand: layout.coreCommand,
dataDir: layout.dataDir,
runtimeBuild: String(plan.build_number ?? plan.version),
asJson: true,
});
if (migration.state === "deferred") {
throw new SdkMigrationDeferred(
migration.code ?? "sdk_migration_incomplete",
);
}
migrated = migration.state === "cutover";
await applyPreparedServiceInstall(preparedService);
serviceApplied = true;
// The receipt exists so a later command can find the data dir without
@@ -3232,13 +3204,6 @@ async function runCommand(
);
}
};
if (migrated) {
// An older Runtime refuses a migrated home: keep this install.
process.stderr.write(
`${JSON.stringify(notReadyAfterMigration("install"), null, 2)}\n`,
);
return EXIT_FAILED;
}
if (serviceApplied) {
await rollbackPreparedServiceInstall(
preparedService,
@@ -3248,13 +3213,7 @@ async function runCommand(
await rollbackFilesAndOwner();
}
process.stderr.write(
`${JSON.stringify(
error instanceof SdkMigrationDeferred
? sdkMigrationFailure("install", error.code)
: serviceErrorDocument("install", error),
null,
2,
)}\n`,
`${JSON.stringify(serviceErrorDocument("install", error), null, 2)}\n`,
);
return EXIT_FAILED;
}
@@ -3616,7 +3575,6 @@ async function runCommand(
let previousOwner:
| Awaited<ReturnType<typeof claimCliServiceOwnership>>
| undefined;
let migrated = false;
const rollbackUpgrade = async (
dataDir: string | undefined,
): Promise<void> => {
@@ -3648,24 +3606,6 @@ async function runCommand(
memhubReceipts.push(await installManagedCliLink(link));
}
previousOwner = await claimCliServiceOwnership(layout.dataDir);
const migration = await (
controls.migrateSdkData ?? migrateSdkDataBeforeStart
)({
coreCommand: layout.coreCommand,
dataDir: parsed.dataDir ?? layout.dataDir,
runtimeBuild: String(plan.build_number ?? plan.version),
asJson,
});
if (migration.state === "deferred") {
await rollbackUpgrade(parsed.dataDir);
process.stderr.write(
`${JSON.stringify(sdkMigrationFailure("upgrade", migration.code ?? "sdk_migration_incomplete"), null, 2)}\n`,
);
return EXIT_FAILED;
}
// An older Runtime refuses a migrated home, so from here a failure
// keeps the new version instead of rolling back to one that cannot start.
migrated = migration.state === "cutover";
await restartUserService(layout, serviceOptions);
// A restart that returns without error only means launchd/systemd
// accepted the command. O-22: the first `upgrade` implementation
@@ -3677,12 +3617,6 @@ async function runCommand(
const ready = await (controls.waitForRuntimeReady ?? waitForRuntimeReady)(
dataDir,
);
if (!ready && migrated) {
process.stderr.write(
`${JSON.stringify(notReadyAfterMigration("upgrade"), null, 2)}\n`,
);
return EXIT_FAILED;
}
if (!ready) {
await rollbackUpgrade(parsed.dataDir);
process.stderr.write(
@@ -3710,9 +3644,9 @@ async function runCommand(
});
}
} catch (error) {
if (!migrated) await rollbackUpgrade(parsed.dataDir);
await rollbackUpgrade(parsed.dataDir);
process.stderr.write(
`${JSON.stringify(migrated ? notReadyAfterMigration("upgrade") : serviceErrorDocument("upgrade", error), null, 2)}\n`,
`${JSON.stringify(serviceErrorDocument("upgrade", error), null, 2)}\n`,
);
return EXIT_FAILED;
}
@@ -4080,71 +4014,6 @@ export function runtimeIsReady(
);
}
const SDK_MIGRATION_STEPS: Record<string, string> = {
bundle: "正在备份记忆数据,数据较多时需要几分钟……",
prepare: "正在整理待处理的写入任务……",
cutover: "正在切换记忆数据源……",
cleanup: "正在清理旧数据表……",
};
/**
* Migrate a home written by an older release before the new Runtime serves it.
*
* Runs the new Runtime's own Core with the Runtime stopped. Progress goes to
* stderr so stdout stays one result document.
*/
async function migrateSdkDataBeforeStart(options: {
coreCommand: string;
dataDir: string;
runtimeBuild: string;
asJson: boolean;
}): Promise<SdkMigrationResult> {
return runSdkMigration({
dataDir: options.dataDir,
core: { command: options.coreCommand },
runtimeBuild: options.runtimeBuild,
onProgress: (step) => {
process.stderr.write(
options.asJson
? `${JSON.stringify({ schema_version: 1, product: PRODUCT, event: "sdk_migration.progress", step })}\n`
: `${SDK_MIGRATION_STEPS[step] ?? step}\n`,
);
},
});
}
class SdkMigrationDeferred extends Error {
constructor(readonly code: string) {
super(code);
}
}
function notReadyAfterMigration(command: "install" | "upgrade") {
return {
schema_version: 1,
product: PRODUCT,
command,
status: "error",
code: "runtime_not_ready_after_sdk_migration",
message:
"记忆数据已迁移到新版本,但新版本 Runtime 没有在限定时间内就绪。旧版本无法打开已迁移的数据,所以没有回滚。请运行 memhub restart,或 memhub doctor 查看原因。",
};
}
function sdkMigrationFailure(command: "install" | "upgrade", code: string) {
return {
schema_version: 1,
product: PRODUCT,
command,
status: "error",
code,
message:
code === "sdk_migration_disk_space_insufficient"
? "磁盘空间不足,记忆数据迁移没有完成,已回滚到上一版本。请腾出与记忆数据同等大小的空间后重试;原有记忆数据不会丢失。"
: "记忆数据迁移没有完成,已回滚到上一版本。请重试;如果仍然失败,请把这个错误码反馈给我们。原有记忆数据不会丢失。",
};
}
async function waitForRuntimeReady(
dataDir: string,
timeoutMs = 45_000,
+1 -31
View File
@@ -1,4 +1,4 @@
import { mkdtemp, rm, symlink } from "node:fs/promises";
import { mkdtemp, rm } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { DiagnosticLogger } from "@one-memory/runtime-control";
@@ -41,15 +41,11 @@ it("redacts unsafe details and absolute stack frames from the read surface", ()
session_id: "11111111-1111-4111-8111-111111111111",
details: {
count: 1,
reason: "/Users/me/private-query.txt",
status: "sk-live-secret-token",
source_id: "credential.private",
prompt: "secret-token-123",
authorization: "Bearer secret-token-123",
} as unknown as Record<string, string | number | boolean | null>,
safe_stack: [
"packages/runtime-control/src/example.ts:42:7",
"packages/runtime/sk-live-secret-token.ts:1:1",
"/Users/me/private.py:1",
"C:\\Users\\me\\private.py:1",
],
@@ -60,8 +56,6 @@ it("redacts unsafe details and absolute stack frames from the read surface", ()
"packages/runtime-control/src/example.ts:42:7",
]);
expect(JSON.stringify(redacted)).not.toContain("secret-token-123");
expect(JSON.stringify(redacted)).not.toContain("sk-live-secret-token");
expect(JSON.stringify(redacted)).not.toContain("credential.private");
expect(JSON.stringify(redacted)).not.toContain("/Users/me");
expect(JSON.stringify(redacted)).not.toContain("C:\\\\Users");
});
@@ -138,27 +132,3 @@ it("follows a newly appended event without stopping Runtime", async () => {
});
expect(seen).toEqual(["service.ready", "core.failed"]);
});
it("fails follow mode when the diagnostic directory is unsafe", async () => {
const dataDir = await mkdtemp(
path.join(os.tmpdir(), "om-cli-follow-unsafe-"),
);
const external = await mkdtemp(
path.join(os.tmpdir(), "om-cli-follow-external-"),
);
roots.push(dataDir, external);
await symlink(external, path.join(dataDir, "logs"));
const query = parseLogQuery(["--follow"]);
expect(query).not.toBeNull();
if (!query) throw new Error("query_missing");
await expect(
followLogs({
dataDir,
query,
signal: new AbortController().signal,
maxEvents: 1,
emit: () => undefined,
wait: async () => undefined,
}),
).rejects.toThrow("log_directory_unreadable");
});
+102 -6
View File
@@ -3,7 +3,6 @@ import path from "node:path";
import type { DiagnosticEventV1 } from "@one-memory/runtime-control";
import {
diagnosticLogFiles,
projectDiagnosticEventForSafeOutput,
readDiagnosticEvents,
} from "@one-memory/runtime-control";
@@ -25,6 +24,54 @@ export interface LogQuery {
const ID = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,255}$/;
const NAME = /^[a-z][a-z0-9]*(?:[._-][a-z0-9]+){0,15}$/;
const SAFE_DETAIL_KEYS = new Set([
"action",
"attempt",
"count",
"method",
"mode",
"phase",
"platform",
"reason",
"result_count",
"source_id",
"spawned",
"status",
"version",
"parent_session_id",
"invalid_lines",
"truncated",
"runtime_state",
"core_state",
"broker_state",
"instance_id",
"open_file_count",
"memory_bytes",
"thread_count",
"database_error_count",
"file_count",
"sqlite_file_count",
"byte_file_count",
"total_bytes",
"copied_bytes",
"bytes_per_second",
"stored_count",
"revised_count",
"skipped_count",
"error_count",
"pending_count",
"prepared_count",
"queue_depth",
"in_flight_count",
"call_count",
"created_count",
"updated_count",
"superseded_count",
"deleted_count",
"operation_kind",
"file_kind",
]);
export function parseLogQuery(arguments_: readonly string[]): LogQuery | null {
if (arguments_.filter((item) => item === "--json").length > 1) return null;
const values: Partial<LogQuery> = { errorsOnly: false, follow: false };
@@ -190,13 +237,63 @@ export function renderLogText(
return `${lines.join("\n")}\n`;
}
/** Defence in depth for the CLI read surface. Local writers accept arbitrary
* bounded scalar keys, so every outward read surface must reuse the same
* field-and-value projection as the support-bundle exporter. */
/** Defence in depth for the CLI read surface. `DiagnosticLogger` already
* allow-lists keys, but a future writer, a manually edited log, or a malformed
* line must not turn `memhub logs` into an exfiltration path. */
export function redactDiagnosticEvent(
event: DiagnosticEventV1,
): DiagnosticEventV1 {
return projectDiagnosticEventForSafeOutput(event);
const redacted: DiagnosticEventV1 = {
schema_version: 1,
timestamp: event.timestamp,
level: event.level,
service: event.service,
component: event.component,
event: event.event,
outcome: event.outcome,
session_id: event.session_id,
...(event.trace_id === undefined ? {} : { trace_id: event.trace_id }),
...(event.incident_id === undefined
? {}
: { incident_id: event.incident_id }),
...(event.instance_id === undefined
? {}
: { instance_id: event.instance_id }),
...(event.operation_id === undefined
? {}
: { operation_id: event.operation_id }),
...(event.request_id === undefined ? {} : { request_id: event.request_id }),
...(event.job_id === undefined ? {} : { job_id: event.job_id }),
...(event.task_id === undefined ? {} : { task_id: event.task_id }),
...(event.duration_ms === undefined
? {}
: { duration_ms: event.duration_ms }),
...(event.state_before === undefined
? {}
: { state_before: event.state_before }),
...(event.state_after === undefined
? {}
: { state_after: event.state_after }),
...(event.error_code === undefined ? {} : { error_code: event.error_code }),
...(event.error_type === undefined ? {} : { error_type: event.error_type }),
...(event.safe_stack === undefined
? {}
: {
safe_stack: event.safe_stack.filter(
(frame) =>
!frame.startsWith("/") &&
!/^[A-Za-z]:\\/.test(frame) &&
!frame.includes(".."),
),
}),
};
const details = Object.fromEntries(
Object.entries(event.details ?? {}).filter(([key]) =>
SAFE_DETAIL_KEYS.has(key),
),
);
if (Object.keys(details).length > 0) redacted.details = details;
return redacted;
}
export async function followLogs(options: {
@@ -218,7 +315,6 @@ export async function followLogs(options: {
...options.query,
tail: 5_000,
});
if (report.failure) throw new Error(report.failure);
for (const event of report.events) {
const identity = JSON.stringify(event);
if (seen.has(identity)) continue;
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "uv_build"
[project]
name = "one-memory"
version = "1.0.5"
version = "1.0.4"
description = "MemHub local-first memory core"
readme = "README.md"
requires-python = ">=3.12"
@@ -12,7 +12,7 @@ dependencies = [
"fastapi>=0.141.1",
"filelock>=3.32.4",
# Public SDK version validated against MemHub's permission and recall contracts.
"hy-memory==2.0.5",
"hy-memory==2.0.5.dev3",
# Retrieval's derived vector index directly consumes these locked versions.
"chromadb==1.5.9",
"numpy==2.5.2",
+39 -83
View File
@@ -998,7 +998,9 @@ def create_app(
app.state.recall_target = GovernedRecallTarget(
app.state.memory_governance_store,
filter_memories=app.state.source_read_access.memories,
semantic_search=lambda query: _sdk_memory_runtime(app).semantic_recall(query),
semantic_search=lambda query: _sdk_memory_runtime(
app
).semantic_recall(query),
# Read per request rather than cached: an import that lands
# while the app is open should make its workspace reachable
# without a restart. Affordable only because the read is served
@@ -1100,7 +1102,9 @@ def create_app(
authorize_generation=lambda profile, question, grounding: (
_authorize_answer_generation(app, profile, question, grounding)
),
semantic_search=lambda query: _sdk_memory_runtime(app).semantic_recall(query),
semantic_search=lambda query: _sdk_memory_runtime(app).semantic_recall(
query
),
resolve_workspace=lambda workspace_path: resolve_workspace_scopes(
workspace_path,
evidence_store.known_workspaces(user_id="local"),
@@ -1124,21 +1128,21 @@ def create_app(
profiles=app.state.model_profile_store,
secrets=app.state.secret_store,
data_dir=runtime_settings.data_dir,
transport_factory=lambda profile: (
lambda: AuditedModelTransport(
app.state.model_request_audit_store,
profile_id=profile.profile_id,
profile_revision=profile.revision,
provider_kind=str(profile.provider_kind),
model=profile.model,
inner=(
BudgetedModelTransport(model_request_budget)
if model_request_budget is not None
else None
),
)
transport_factory=lambda profile: lambda: AuditedModelTransport(
app.state.model_request_audit_store,
profile_id=profile.profile_id,
profile_revision=profile.revision,
provider_kind=str(profile.provider_kind),
model=profile.model,
inner=(
BudgetedModelTransport(model_request_budget)
if model_request_budget is not None
else None
),
),
sedimentation_enabled=lambda: _config_store(app).sedimentation_enabled(),
sedimentation_enabled=lambda: _config_store(
app
).sedimentation_enabled(),
)
app.state.sdk_memory_runtime = SdkMemoryRuntime(
coordinator=app.state.sdk_write_coordinator,
@@ -1147,8 +1151,6 @@ def create_app(
extraction_settings=app.state.extraction_settings_store,
data_dir=runtime_settings.data_dir,
)
if app.state.memory_authority is MemoryAuthority.SDK:
app.state.sdk_memory_runtime.start_raw_retention()
# C2 owns the single production budget (handle 5s + reserve 250ms,
# total 5.25s, lease 10s, renew 1s, claim 250ms). The composition
# root only consumes it: no local formula, and the 250ms claim
@@ -1163,8 +1165,10 @@ def create_app(
exclude_error_codes=NON_RETRYABLE_POST_WRITE_CODES,
)
for job in unknown:
receipt = await app.state.sdk_write_coordinator.inquire_write_outcome(
job.request_id
receipt = (
await app.state.sdk_write_coordinator.inquire_write_outcome(
job.request_id
)
)
if receipt is None:
continue
@@ -1430,14 +1434,6 @@ def create_app(
app.state.memory_backfill = None
app.state.memory_visibility_task = None
app.state.deletion_sync = None
if app.state.memory_authority is MemoryAuthority.LEGACY:
_start_backfill(app)
_start_visibility_recompute(app)
app.state.deletion_sync = watched_task(
asyncio.get_running_loop(),
lambda: _keep_legacy_deletions_in_sync(app),
name="one-memory-deletion-sync",
)
# Both times the loop was blocked here it was found from outside,
# by hand, days later. This says so at the time, and says what was
# holding it.
@@ -1612,7 +1608,9 @@ def create_app(
)
if getattr(
app.state, "memory_authority", MemoryAuthority.LEGACY
) is MemoryAuthority.SDK and request.url.path.startswith(legacy_paths):
) is MemoryAuthority.SDK and request.url.path.startswith(
legacy_paths
):
return JSONResponse(
status_code=409,
content={"detail": "sdk_memory_authority_required"},
@@ -1734,7 +1732,7 @@ def create_app(
trace_id=trace_id,
instance_id=runtime_settings.instance_id,
duration_ms=elapsed_ms(started),
details={**diagnostic_details, "status_code": response.status_code},
details={**diagnostic_details, "status": str(response.status_code)},
)
return response
except Exception as error:
@@ -3783,7 +3781,9 @@ def create_app(
all_memories, projected, nodes, edges = await asyncio.to_thread(read_governed_part)
try:
hy_graph = (
None if revoked else await _sdk_memory_runtime(app).graph_snapshot(limit=limit)
None
if revoked
else await _sdk_memory_runtime(app).graph_snapshot(limit=limit)
)
except HySdkError:
hy_graph = None
@@ -3892,7 +3892,9 @@ def create_app(
active_queue, in_flight_count = supervisor.active_depths()
planned, current = supervisor.manual_batch_progress()
sdk_counts = (
supervisor.status_counts() if app.state.memory_authority is MemoryAuthority.SDK else {}
supervisor.status_counts()
if app.state.memory_authority is MemoryAuthority.SDK
else {}
)
# `running` counts rows the store still marks running, which includes
# work a worker has already given up on. Reporting that as "what is
@@ -3913,7 +3915,9 @@ def create_app(
),
recent_completions=depths.recent_completions,
running=sdk_counts.get("running", running_rows),
stranded_running=(0 if sdk_counts else max(0, running_rows - in_flight_count)),
stranded_running=(
0 if sdk_counts else max(0, running_rows - in_flight_count)
),
unexecuted_policy_failures=store.unexecuted_policy_failures(),
waiting_for_model=sdk_counts.get(
"waiting_for_model",
@@ -4234,7 +4238,7 @@ def create_app(
if action == "delete":
deletion = _memory_store(app).deletion(memory_id, user_id="local")
if deletion is None or deletion.status is not DeletionStatus.COMPLETED:
await _finish_deletion(app, memory_id)
await _sdk_memory_runtime(app).finish_deletion(memory_id)
return memory
except KeyError:
raise HTTPException(status_code=404, detail="memory_not_found") from None
@@ -4932,54 +4936,6 @@ def _secret_store_detail(error: SecretStoreUnavailableError) -> str:
return str(getattr(error, "code", None) or SecretStoreErrorCode.STORE_UNAVAILABLE)
async def _finish_deletion(app: FastAPI, memory_id: str) -> None:
"""Delete from the SDK and, on legacy authority, settle the deletion row.
Without the settle the legacy row stays `pending` for good, and the person
is shown a deletion that never finishes.
"""
if app.state.memory_authority is MemoryAuthority.SDK:
await _sdk_memory_runtime(app).finish_deletion(memory_id)
return
await _work_legacy_deletions(app, memory_id=memory_id)
async def _work_legacy_deletions(app: FastAPI, *, memory_id: str | None = None) -> None:
store = _memory_store(app)
runtime = _sdk_memory_runtime(app)
claims = await asyncio.to_thread(store.claim_deletions, user_id="local", memory_id=memory_id)
for claim in claims:
try:
await runtime.finish_deletion(claim.memory_id)
except HySdkError as error:
code: str | None = error.code
except Exception as error: # noqa: BLE001 - recorded so it waits its turn
logger.warning("memory_deletion_unexpected_failure: %s", type(error).__name__)
code = "hy_memory_delete_unexpected"
else:
code = None
await asyncio.to_thread(store.settle_deletion, claim, user_id="local", error_code=code)
_LEGACY_DELETION_SYNC_IDLE_SECONDS = 60.0
async def _keep_legacy_deletions_in_sync(app: FastAPI) -> None:
"""Retry legacy deletions the SDK did not take; nothing else asks again."""
store = _memory_store(app)
while True:
wait = _LEGACY_DELETION_SYNC_IDLE_SECONDS
try:
await _work_legacy_deletions(app)
due = await asyncio.to_thread(store.next_deletion_due, user_id="local")
except Exception as error: # noqa: BLE001 - reported, then retried
report_exception("memory.deletion_sync", "memory_deletion_sync.failed", error)
else:
if due is not None:
wait = min(wait, max(0.0, (due - store.clock()).total_seconds()))
await asyncio.sleep(wait)
def _memory_store(app: FastAPI) -> MemoryGovernanceStore:
return cast(MemoryGovernanceStore, app.state.memory_governance_store)
@@ -5328,7 +5284,7 @@ def _report_visibility_failure(phase: str, error: BaseException) -> None:
event="memory_visibility.background_failed",
outcome="failed",
error_type=type(error).__name__,
details={"phase": phase},
details={"phase": phase, "reason": str(error)[:200]},
)
+502 -96
View File
@@ -1,3 +1,5 @@
import asyncio
import hashlib
import os
import signal
import socket
@@ -29,51 +31,13 @@ def _end_by(seconds: str | None) -> None:
if __name__ == "__main__" and sys.argv[1:2] == ["capture"]:
_end_by(os.environ.get("ONE_MEMORY_CAPTURE_DEADLINE_S"))
import errno # noqa: E402
import json # noqa: E402
import re # noqa: E402
from collections.abc import Iterator, Sequence # noqa: E402
from collections.abc import Sequence # noqa: E402
from typing import TextIO # noqa: E402
from one_memory.config import load_settings # noqa: E402
from one_memory.runtime_client import RuntimeStatus, inspect_runtime # noqa: E402
# The startup dialog can only act on the name it is given: running out of disk
# and a damaged backup are different next steps. These are the names that reach
# it.
_SAFE_MIGRATION_CODE = re.compile(r"^[a-z][a-z0-9_]{0,95}$")
_OUT_OF_SPACE = frozenset({errno.ENOSPC, getattr(errno, "EDQUOT", errno.ENOSPC)})
def _exception_chain(error: BaseException) -> Iterator[BaseException]:
seen: set[int] = set()
current: BaseException | None = error
while current is not None and id(current) not in seen:
seen.add(id(current))
yield current
current = current.__cause__ or current.__context__
def migration_failure_code(
error: BaseException, *, fallback: str = "sdk_migration_preflight_failed"
) -> str:
"""Name a migration failure; never guess a code from free text.
Errs toward the caller's generic code: an exception this does not recognize
keeps its message out of the document Desktop reads.
"""
chain = list(_exception_chain(error))
for item in chain:
if isinstance(item, OSError) and item.errno in _OUT_OF_SPACE:
return "sdk_migration_disk_space_insufficient"
for item in chain:
code = getattr(item, "code", "")
if not isinstance(code, str) or not _SAFE_MIGRATION_CODE.match(code):
continue
if code.startswith(("sdk_data_plane_", "sdk_migration_")):
return code
return fallback
def main() -> None:
if sys.argv[1:] == ["--keychain-read"]:
@@ -113,10 +77,16 @@ def run(arguments: Sequence[str], *, output: TextIO | None = None) -> int:
command, *options = arguments
if command == "restore":
return restore(options, stream)
if command == "migration-bundle":
return migration_bundle(options, stream)
if command == "sdk-migrate":
return sdk_migrate(options, stream)
if command == "sdk-cutover":
return sdk_cutover(options, stream)
if command == "sdk-migration-status":
return sdk_migration_status(options, stream)
if command == "sdk-cleanup":
return sdk_cleanup(options, stream)
if command == "boundary" and not options:
from one_memory.workbuddy_boundary import write_hook
@@ -137,8 +107,11 @@ def run(arguments: Sequence[str], *, output: TextIO | None = None) -> int:
stream.write(
"Usage: one-memory [serve [--maintenance|--reader]|status [--json]|doctor [--json]"
"|restore --source <managed-backup> --yes [--json]"
"|sdk-migrate --runtime-build <build> --yes [--json]"
"|migration-bundle --bundle-id <id> --runtime-build <build> --yes [--json]"
"|sdk-migrate --migration-id <id> --bundle-id <id> --yes [--json]"
"|sdk-cutover --migration-id <id> --bundle-id <id> --yes [--json]"
"|sdk-migration-status [--json]"
"|sdk-cleanup --yes [--json]"
"|boundary [--peek|--ack TOKEN]|capture SOURCE]\n"
)
return 2
@@ -280,20 +253,19 @@ def _write_restore(
stream.write(f"{text}\n")
def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
"""Bring a stopped home to SDK authority in one resumable command.
Backs up the home, makes queue work independent of the legacy tables,
commits SDK authority and drops those tables. The SDK already holds every
memory and raw turn, so nothing is copied. With `--json`, each step is one
progress line and the last line is the result.
"""
def migration_bundle(options: Sequence[str], stream: TextIO) -> int:
"""Create a complete management + SDK bundle while Runtime is stopped."""
bundle_id: str | None = None
runtime_build: str | None = None
confirmed = False
as_json = False
index = 0
while index < len(options):
option = options[index]
if option == "--bundle-id" and index + 1 < len(options):
bundle_id = options[index + 1]
index += 2
continue
if option == "--runtime-build" and index + 1 < len(options):
runtime_build = options[index + 1]
index += 2
@@ -308,8 +280,116 @@ def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
continue
stream.write(f"one-memory: 参数无效:{option}\n")
return 2
if not runtime_build:
stream.write("one-memory: sdk-migrate 需要 --runtime-build <build>\n")
if not bundle_id or not runtime_build:
stream.write("one-memory: migration-bundle 需要 --bundle-id <id> --runtime-build <build>\n")
return 2
settings = load_settings()
status = inspect_runtime(settings.data_dir)
if status.state != "stopped":
document = {
"schema_version": 1,
"state": "blocked",
"code": "sdk_migration_bundle_runtime_running",
"runtime_state": status.state,
}
_write_restore(stream, as_json, document, "Runtime 仍在运行;不能创建迁移备份。")
return 4
if not confirmed:
document = {
"schema_version": 1,
"state": "preview",
"bundle_id": bundle_id,
"requires_confirmation": True,
}
_write_restore(stream, as_json, document, "预览通过;加 --yes 后创建完整迁移备份。")
return 4
from one_memory.sdk_migration_bundle import (
MigrationBundleRejected,
create_sdk_migration_bundle,
inspect_sdk_migration_bundle,
)
try:
result = create_sdk_migration_bundle(
data_dir=settings.data_dir,
bundle_id=bundle_id,
runtime_build=runtime_build,
runtime_stopped=True,
)
except MigrationBundleRejected as error:
if error.code == "sdk_migration_bundle_exists":
try:
manifest = inspect_sdk_migration_bundle(
data_dir=settings.data_dir,
bundle_id=bundle_id,
)
except MigrationBundleRejected:
pass
else:
document = {
"schema_version": 1,
"state": "reused",
"bundle_id": bundle_id,
"file_count": manifest.get("file_count"),
"total_bytes": manifest.get("total_bytes"),
}
_write_restore(
stream,
as_json,
document,
f"已验证并复用迁移备份:{bundle_id}",
)
return 0
document = {
"schema_version": 1,
"state": "failed",
"code": error.code,
}
_write_restore(stream, as_json, document, f"迁移备份失败:{error.code}")
return 1
document = {
"schema_version": 1,
"state": "created",
"bundle_id": result.bundle_id,
"bundle_path": str(result.bundle_path),
"file_count": result.file_count,
"total_bytes": result.total_bytes,
}
_write_restore(stream, as_json, document, f"完整迁移备份已创建:{result.bundle_path}")
return 0
def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
"""Backfill legacy memory/raw into SDK while keeping legacy authority."""
migration_id: str | None = None
bundle_id: str | None = None
confirmed = False
as_json = False
index = 0
while index < len(options):
option = options[index]
if option == "--migration-id" and index + 1 < len(options):
migration_id = options[index + 1]
index += 2
continue
if option == "--bundle-id" and index + 1 < len(options):
bundle_id = options[index + 1]
index += 2
continue
if option == "--yes":
confirmed = True
index += 1
continue
if option == "--json":
as_json = True
index += 1
continue
stream.write(f"one-memory: 参数无效:{option}\n")
return 2
if not migration_id or not bundle_id:
stream.write("one-memory: sdk-migrate 需要 --migration-id <id> --bundle-id <id>\n")
return 2
settings = load_settings()
@@ -324,53 +404,271 @@ def sdk_migrate(options: Sequence[str], stream: TextIO) -> int:
_write_restore(stream, as_json, document, "Runtime 仍在运行;不能迁移。")
return 4
if not confirmed:
document = {"schema_version": 1, "state": "preview", "requires_confirmation": True}
_write_restore(stream, as_json, document, "预览通过;加 --yes 后执行迁移。")
document = {
"schema_version": 1,
"state": "preview",
"migration_id": migration_id,
"bundle_id": bundle_id,
"requires_confirmation": True,
"commits_cutover": False,
}
_write_restore(
stream,
as_json,
document,
"预览通过;加 --yes 后执行可续跑 backfill,但不会切换 authority。",
)
return 4
from one_memory.database_access import close_main_databases
from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE, HY_MEMORY_VERSION
from one_memory.hy_sdk_migration import HySdkMigration
from one_memory.hy_sdk_runtime import acquire_client
from one_memory.memory_server.migration_runner import SqliteMigrationStore, migrate
from one_memory.memory_server.migrations import MIGRATIONS
from one_memory.model_profiles import ModelProfileStore
from one_memory.sdk_migration_bundle import (
MigrationBundleRejected,
inspect_sdk_migration_bundle,
)
from one_memory.sdk_migration_runner import (
BackfillResult,
prepare_sdk_migration_staging,
run_sdk_data_plane_backfill,
sdk_migration_staging,
)
from one_memory.secrets import KeyringSecretStore, SecretStoreUnavailableError
database = settings.data_dir / "one-memory.sqlite3"
try:
manifest = inspect_sdk_migration_bundle(
data_dir=settings.data_dir,
bundle_id=bundle_id,
)
asyncio.run(migrate(SqliteMigrationStore(database), MIGRATIONS))
close_main_databases()
profiles = ModelProfileStore(database)
identity = profiles.active_identity()
if identity is None:
raise RuntimeError("model_profile_not_active")
profile = profiles.snapshot(*identity)
secrets = KeyringSecretStore()
secret = profiles.resolve_snapshot_secret(*identity, secrets)
if profile.has_api_key and secret is None:
raise RuntimeError("model_secret_unavailable")
except MigrationBundleRejected as error:
code = error.code
except SecretStoreUnavailableError:
code = "secure_storage_unavailable"
except Exception as error:
code = (
str(error)
if str(error)
in {
"model_profile_not_active",
"model_secret_unavailable",
}
else "sdk_migration_preflight_failed"
)
else:
namespace_fingerprint = hashlib.sha256(HY_MEMORY_NAMESPACE.encode()).hexdigest()
backend_material = json.dumps(
{
"namespace": HY_MEMORY_NAMESPACE,
"embedding_provider": str(profile.embedding_provider_kind or ""),
"embedding_model": profile.embedding_model,
"embedding_dimensions": profile.embedding_dimensions,
"vector": "chroma",
"cache": "sqlite",
"history": "sqlite",
"coding": "sqlite",
},
sort_keys=True,
separators=(",", ":"),
)
backend_fingerprint = hashlib.sha256(backend_material.encode()).hexdigest()
staging = sdk_migration_staging(
data_dir=settings.data_dir,
migration_id=migration_id,
namespace=HY_MEMORY_NAMESPACE,
)
try:
staging_data_dir = prepare_sdk_migration_staging(
paths=staging,
database_path=database,
)
except Exception:
code = "sdk_migration_staging_failed"
staging_data_dir = None
async def run_backfill() -> BackfillResult:
assert staging_data_dir is not None
lease = await asyncio.to_thread(
acquire_client,
profile,
api_key=secret,
data_dir=staging_data_dir,
)
try:
return await run_sdk_data_plane_backfill(
database_path=database,
client=HySdkMigration(lease.client),
migration_id=migration_id,
bundle_id=bundle_id,
source_runtime_build=str(manifest.get("runtime_build") or ""),
target_sdk_version=HY_MEMORY_VERSION,
source_schema_version=MIGRATIONS[-1].version,
namespace_fingerprint=namespace_fingerprint,
backend_fingerprint=backend_fingerprint,
)
finally:
await asyncio.to_thread(lease.close)
try:
result = (
asyncio.run(run_backfill())
if staging_data_dir is not None
else None
)
except Exception:
code = "sdk_migration_backfill_failed"
else:
if result is None:
document = {"schema_version": 1, "state": "failed", "code": code}
_write_restore(stream, as_json, document, f"SDK backfill 失败:{code}")
return 1
document = {
"schema_version": 1,
"state": result.state,
"phase": result.phase,
"migration_id": result.migration_id,
"counts": result.counts,
"memory_items": result.memory_items,
"raw_items": result.raw_items,
"commits_cutover": False,
}
_write_restore(
stream,
as_json,
document,
"SDK backfill 已完成;authority 尚未切换。",
)
return 0 if result.state == "ready" else 4
document = {"schema_version": 1, "state": "failed", "code": code}
_write_restore(stream, as_json, document, f"SDK backfill 失败:{code}")
return 1
def sdk_cutover(options: Sequence[str], stream: TextIO) -> int:
"""Commit SDK authority only after offline reconciliation and work drain."""
migration_id: str | None = None
bundle_id: str | None = None
confirmed = False
as_json = False
index = 0
while index < len(options):
option = options[index]
if option == "--migration-id" and index + 1 < len(options):
migration_id = options[index + 1]
index += 2
continue
if option == "--bundle-id" and index + 1 < len(options):
bundle_id = options[index + 1]
index += 2
continue
if option == "--yes":
confirmed = True
index += 1
continue
if option == "--json":
as_json = True
index += 1
continue
stream.write(f"one-memory: 参数无效:{option}\n")
return 2
if not migration_id or not bundle_id:
stream.write("one-memory: sdk-cutover 需要 --migration-id <id> --bundle-id <id>\n")
return 2
settings = load_settings()
status = inspect_runtime(settings.data_dir)
if status.state != "stopped":
document = {
"schema_version": 1,
"state": "blocked",
"code": "sdk_cutover_runtime_running",
"runtime_state": status.state,
}
_write_restore(stream, as_json, document, "Runtime 仍在运行;不能切换 authority。")
return 4
if not confirmed:
document = {
"schema_version": 1,
"state": "preview",
"migration_id": migration_id,
"bundle_id": bundle_id,
"requires_confirmation": True,
}
_write_restore(stream, as_json, document, "预览通过;加 --yes 后提交 SDK authority。")
return 4
from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE, HY_MEMORY_VERSION
from one_memory.sdk_migration_runner import run_startup_migration
step_names = {
"bundle": "正在备份记忆数据……",
"prepare": "正在整理待处理的写入任务……",
"cutover": "正在切换数据源……",
"cleanup": "正在删除旧表……",
}
def progress(step: str) -> None:
if as_json:
stream.write(
json.dumps({"schema_version": 1, "state": "running", "step": step}) + "\n"
)
else:
stream.write(f"{step_names.get(step, step)}\n")
stream.flush()
from one_memory.sdk_data_plane_migration import DataPlaneMigrationError
from one_memory.sdk_migration_bundle import MigrationBundleRejected
from one_memory.sdk_migration_runner import (
activate_sdk_migration_staging,
commit_reconciled_cutover,
resume_legacy_cleanup,
sdk_migration_staging,
)
try:
result = run_startup_migration(
database_path = settings.data_dir / "one-memory.sqlite3"
staging = sdk_migration_staging(
data_dir=settings.data_dir,
runtime_build=runtime_build,
sdk_version=HY_MEMORY_VERSION,
migration_id=migration_id,
namespace=HY_MEMORY_NAMESPACE,
progress=progress,
)
except Exception as error:
code = migration_failure_code(error)
document = {"schema_version": 1, "state": "failed", "code": code}
_write_restore(stream, as_json, document, f"迁移未完成:{code}")
return 1
messages = {
"not_required": "不需要迁移。",
"complete": "迁移已完成,旧表已清理。",
"cutover": "迁移完成:记忆数据已以 SDK 为准。",
marker = activate_sdk_migration_staging(
paths=staging,
database_path=database_path,
commit=lambda: commit_reconciled_cutover(
data_dir=settings.data_dir,
database_path=database_path,
migration_id=migration_id,
bundle_id=bundle_id,
sdk_version=HY_MEMORY_VERSION,
),
)
resume_legacy_cleanup(
data_dir=settings.data_dir,
database_path=database_path,
)
except (DataPlaneMigrationError, MigrationBundleRejected, OSError) as error:
code = getattr(error, "code", str(error))
document = {
"schema_version": 1,
"state": "blocked",
"code": code,
"migration_id": migration_id,
"bundle_id": bundle_id,
}
_write_restore(stream, as_json, document, f"SDK authority 未切换:{code}")
return 4
document = {
"schema_version": 1,
"state": "cutover",
"migration_id": marker.migration_id,
"bundle_id": marker.backup_bundle_id,
"sdk_version": marker.sdk_version,
"completed_at": marker.completed_at,
}
_write_restore(stream, as_json, result.as_document(), messages[result.state])
_write_restore(stream, as_json, document, "SDK authority 已切换。")
return 0
def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
"""Read-only probe: does this home need `sdk-migrate`?"""
"""Read-only startup probe used before taking the Runtime maintenance fence."""
if list(options) not in ([], ["--json"]):
stream.write("one-memory: sdk-migration-status 只接受 --json\n")
return 2
@@ -378,33 +676,82 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
settings = load_settings()
database_path = settings.data_dir / "one-memory.sqlite3"
if not database_path.is_file():
document: dict[str, object] = {
document = {
"schema_version": 1,
"authority": "sdk",
"state": "not_required",
"authority": "legacy",
"state": "empty",
"migration_id": None,
}
else:
import sqlite3
from one_memory.database_access import main_database
from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
from one_memory.sdk_data_plane_migration import (
MemoryAuthority,
SdkDataPlaneMigrationStore,
)
from one_memory.sdk_migration_runner import abandoned_staging_present
from one_memory.sdk_migration_runner import sdk_migration_staging
from one_memory.target_schema import legacy_tables_present
authority = SdkDataPlaneMigrationStore(database_path).authority()
migration_store = SdkDataPlaneMigrationStore(database_path)
marker = migration_store.cutover()
authority = migration_store.authority()
stat = database_path.stat()
generation = hashlib.sha256(f"{stat.st_size}:{stat.st_mtime_ns}".encode()).hexdigest()[:16]
if authority is MemoryAuthority.SDK:
leftovers = bool(
legacy_tables_present(database_path)
or abandoned_staging_present(
legacy_tables = legacy_tables_present(database_path)
staging_cleanup = False
if marker is not None:
staging = sdk_migration_staging(
data_dir=settings.data_dir,
migration_id=marker.migration_id,
namespace=HY_MEMORY_NAMESPACE,
)
)
state = "required" if leftovers else "not_required"
staging_cleanup = (
staging.protective_root.exists()
or staging.staging_home.exists()
)
document = {
"schema_version": 1,
"authority": "sdk",
"state": (
"cleanup_required"
if legacy_tables or staging_cleanup
else "complete"
),
"migration_id": marker.migration_id if marker else None,
"generation": generation,
"legacy_tables": list(legacy_tables),
}
else:
state = "required"
document = {"schema_version": 1, "authority": authority.value, "state": state}
with main_database(database_path).read(row_factory=sqlite3.Row) as database:
tables = {
str(row["name"])
for row in database.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
counts = {
"memory": (
int(
database.execute("SELECT COUNT(*) FROM governed_memories").fetchone()[0]
)
if "governed_memories" in tables
else 0
),
"raw": (
int(database.execute("SELECT COUNT(*) FROM evidence").fetchone()[0])
if "evidence" in tables
else 0
),
}
document = {
"schema_version": 1,
"authority": "legacy",
"state": "required" if any(counts.values()) else "empty",
"migration_id": None,
"counts": counts,
"generation": generation,
}
if as_json:
stream.write(f"{json.dumps(document, ensure_ascii=False)}\n")
else:
@@ -412,6 +759,65 @@ def sdk_migration_status(options: Sequence[str], stream: TextIO) -> int:
return 0
def sdk_cleanup(options: Sequence[str], stream: TextIO) -> int:
if set(options) - {"--yes", "--json"} or options.count("--yes") > 1:
stream.write("one-memory: sdk-cleanup 只接受 --yes [--json]\n")
return 2
confirmed = "--yes" in options
as_json = "--json" in options
settings = load_settings()
status = inspect_runtime(settings.data_dir)
if status.state != "stopped":
document = {
"schema_version": 1,
"state": "blocked",
"code": "sdk_cleanup_runtime_running",
}
_write_restore(stream, as_json, document, "Runtime 仍在运行;不能清理旧表。")
return 4
if not confirmed:
_write_restore(
stream,
as_json,
{
"schema_version": 1,
"state": "preview",
"requires_confirmation": True,
},
"加 --yes 后删除已迁移旧表。",
)
return 4
from one_memory.sdk_data_plane_migration import DataPlaneMigrationError
from one_memory.sdk_migration_bundle import MigrationBundleRejected
from one_memory.sdk_migration_runner import resume_legacy_cleanup
try:
dropped = resume_legacy_cleanup(
data_dir=settings.data_dir,
database_path=settings.data_dir / "one-memory.sqlite3",
)
except (DataPlaneMigrationError, MigrationBundleRejected) as error:
code = getattr(error, "code", str(error))
_write_restore(
stream,
as_json,
{"schema_version": 1, "state": "blocked", "code": code},
f"旧表清理未完成:{code}",
)
return 4
_write_restore(
stream,
as_json,
{
"schema_version": 1,
"state": "clean",
"dropped_tables": list(dropped),
},
"旧表已删除。",
)
return 0
def _status_text(command: str, status: RuntimeStatus) -> str:
if status.state == "running":
location = f"{status.host}:{status.port}"
+122 -217
View File
@@ -20,10 +20,8 @@ from __future__ import annotations
import atexit
import json
import logging
import math
import os
import re
import stat
import sys
import threading
import time
@@ -74,30 +72,80 @@ class _Refused(OSError):
"""The path is not one we are willing to write to, and never will be."""
DiagnosticLevel = Literal["debug", "info", "warn", "error"]
DiagnosticLevel = Literal["info", "warn", "error"]
_IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,255}$")
_EVENT = re.compile(r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+){0,15}$")
_DIAGNOSTIC_LOG_NAME = re.compile(
r"^diagnostic-(runtime|core|host-bridge|desktop|cli)\.jsonl(?:\.\d+)?$"
_DETAIL_KEYS = frozenset(
{
"action",
"attempt",
"count",
"method",
"mode",
"phase",
"platform",
"reason",
"result_count",
"source_id",
"spawned",
"status",
"version",
"parent_session_id",
"invalid_lines",
"truncated",
"open_file_count",
"memory_bytes",
"thread_count",
"database_error_count",
"route_path",
"sqlite_errorcode",
"sqlite_errorname",
"file_count",
"sqlite_file_count",
"byte_file_count",
"total_bytes",
"copied_bytes",
"bytes_per_second",
"stored_count",
"revised_count",
"skipped_count",
"error_count",
"pending_count",
"prepared_count",
"queue_depth",
"in_flight_count",
"call_count",
"created_count",
"updated_count",
"superseded_count",
"deleted_count",
"operation_kind",
"index_complete",
"lexical_hits",
"semantic_hits",
"generation",
"timed_out",
"failure_site",
"error_class",
"file_kind",
"file_size",
"file_mtime_ns",
"file_inode",
"cursor_new",
"cursor_old",
"cursor_read_to_end",
"cursor_rewind_branch",
"summed_ms",
"model_ms",
"accounted_ms",
"unaccounted_ms",
"conversation_hash",
# The sedimentation switch: a bounded boolean, the same shape as
# `spawned` and `index_complete`. Left out of this list it did not
# drop a log line, it turned the switch's own request into a 500.
"enabled",
}
)
_DETAIL_KEY = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,63}$")
_CODE_FRAME = re.compile(
r"^(?:(?:apps|packages|scripts)/(?:[A-Za-z0-9_-]+/)*[A-Za-z0-9_.-]+:\d+:\d+"
r"|one_memory/(?:[A-Za-z0-9_-]+/)*[A-Za-z0-9_.-]+:\d+:[A-Za-z_][A-Za-z0-9_]*"
r"|[A-Za-z0-9_][A-Za-z0-9_.-]*\.py:\d+:[A-Za-z_][A-Za-z0-9_]*)$"
)
_SHORT_CREDENTIAL_TOKEN = re.compile(
r"(?<![A-Za-z0-9])([A-Za-z0-9]{24,47})(?:[_-])?(?![A-Za-z0-9])"
)
_REFUSED_DETAIL_KEYS = frozenset({"__proto__", "prototype", "constructor"})
_MAX_DETAIL_KEYS = 32
_MAX_SAFE_INTEGER = (1 << 53) - 1
_EXPORT_SECRET_MARKER = re.compile(
r"(?:^|[._:-])(?:authorization|credential|password|prompt|query|secret|token)(?:$|[._:-])",
re.IGNORECASE,
)
_TRACE_STRING_DETAIL_KEYS = frozenset({"phase", "mode", "status", "reason", "failure_site"})
_TRACE_NUMBER_DETAIL_KEYS = frozenset({"count"})
_trace_id: ContextVar[str | None] = ContextVar("one_memory_trace_id", default=None)
_active: DiagnosticLog | None = None
_database_errors: Counter[tuple[str, str]] = Counter()
@@ -216,12 +264,7 @@ class DiagnosticLog:
}
value.update({key: _identifier(item) for key, item in identifiers.items() if item})
if duration_ms is not None:
if (
not isinstance(duration_ms, int)
or isinstance(duration_ms, bool)
or duration_ms < 0
or duration_ms > _MAX_SAFE_INTEGER
):
if not isinstance(duration_ms, int) or duration_ms < 0:
raise ValueError("diagnostic_duration_invalid")
value["duration_ms"] = duration_ms
for field_name, item in {
@@ -233,9 +276,7 @@ class DiagnosticLog:
if item is not None:
value[field_name] = _identifier(item)
if safe_stack:
if len(safe_stack) > 16:
raise ValueError("diagnostic_stack_invalid")
value["safe_stack"] = [_safe_stack_line(line) for line in safe_stack]
value["safe_stack"] = [_safe_stack_line(line) for line in safe_stack[:256]]
if details:
value["details"] = _details(details)
self._submit(_serialize(value))
@@ -248,8 +289,6 @@ class DiagnosticLog:
event: str,
outcome: str,
) -> dict[str, object]:
if level not in ("debug", "info", "warn", "error"):
raise ValueError("diagnostic_level_invalid")
return {
"schema_version": 1,
"timestamp": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
@@ -612,17 +651,24 @@ def _drain_diagnostics_at_exit() -> None:
def diagnostic_event(**values: object) -> None:
active = _active
if active is None:
return
if values.get("level") == "debug" and not logging.getLogger().isEnabledFor(logging.DEBUG):
return
try:
active = _active
if active is None:
return
if values.get("level") == "debug" and not logging.getLogger().isEnabledFor(logging.DEBUG):
return
active.write(**values) # type: ignore[arg-type]
except Exception:
# Diagnostics are best-effort: validation, serialization, logging, and
# filesystem failures must never become a product outage. BaseException
# deliberately remains visible so process-control signals still work.
except OSError:
# Diagnostics must not become a new product outage when a disk is full
# or the log directory is temporarily unavailable.
return
except ValueError:
# Same promise, same reason, for a rejected payload instead of a failed
# write. An unlisted `details` key once turned every external recall
# into a 500 and then killed the process, and the sedimentation switch
# into a permanent failure. The payload itself is still wrong and is
# still caught in CI by tests/test_diagnostic_payload_keys.py; what
# must not happen is the caller paying for it.
return
@@ -645,95 +691,46 @@ def read_trace_events(
if (request_id is None) == (trace_id is None):
raise ValueError("exactly_one_trace_identity_required")
key, expected = ("request_id", request_id) if request_id is not None else ("trace_id", trace_id)
if expected is None or not _export_identifier(expected):
if expected is None or not _IDENTIFIER.fullmatch(expected):
raise ValueError("invalid_trace_identity")
selected: list[dict[str, object]] = []
directory = data_dir / "logs"
try:
directory_info = directory.lstat()
except FileNotFoundError:
return ()
if stat.S_ISLNK(directory_info.st_mode) or not stat.S_ISDIR(directory_info.st_mode):
raise ValueError("diagnostic_directory_unsafe")
files: list[tuple[Path, float]] = []
for path in directory.iterdir():
if not _DIAGNOSTIC_LOG_NAME.fullmatch(path.name):
allowed_details = {"phase", "mode", "status", "reason", "failure_site", "count"}
files = sorted(
(data_dir / "logs").glob("diagnostic-*.jsonl*"),
key=lambda path: path.stat().st_mtime,
)
for path in files:
if path.is_symlink() or not path.is_file():
continue
try:
info = path.lstat()
except OSError:
continue
if stat.S_ISREG(info.st_mode) and not stat.S_ISLNK(info.st_mode):
files.append((path, info.st_mtime))
for path, _mtime in sorted(files, key=lambda item: item[1]):
for line in _read_regular_file_no_follow(path).splitlines():
for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
try:
event = json.loads(line)
except (TypeError, json.JSONDecodeError):
continue
if not isinstance(event, dict) or event.get(key) != expected:
continue
timestamp = event.get("timestamp")
if not isinstance(timestamp, str):
continue
try:
datetime.fromisoformat(timestamp.replace("Z", "+00:00"))
except ValueError:
continue
component = event.get("component")
event_name = event.get("event")
outcome = event.get("outcome")
projected: dict[str, object] = {
"timestamp": timestamp,
"component": (
component
if isinstance(component, str) and _export_identifier(component)
else "diagnostic.redacted"
),
"event": (
event_name
if isinstance(event_name, str) and _export_identifier(event_name)
else "diagnostic.redacted"
),
"outcome": (
outcome
if isinstance(outcome, str) and _export_identifier(outcome)
else "redacted"
),
}
duration_ms = event.get("duration_ms")
if (
isinstance(duration_ms, int)
and not isinstance(duration_ms, bool)
and 0 <= duration_ms <= _MAX_SAFE_INTEGER
):
projected["duration_ms"] = duration_ms
for name in ("error_code", "error_type"):
value = event.get(name)
if isinstance(value, str) and _export_identifier(value):
projected[name] = value
details = event.get("details")
projected = {
name: event[name]
for name in (
"timestamp",
"component",
"event",
"outcome",
"duration_ms",
"error_code",
"error_type",
)
if name in event
}
if isinstance(details, dict):
safe_details = _trace_details(details)
if safe_details:
projected["details"] = safe_details
projected["details"] = {
name: value for name, value in details.items() if name in allowed_details
}
selected.append(projected)
return tuple(selected[-max(1, min(limit, 500)) :])
def _read_regular_file_no_follow(path: Path) -> str:
descriptor = os.open(path, os.O_RDONLY | _O_NOFOLLOW)
try:
if not stat.S_ISREG(os.fstat(descriptor).st_mode):
raise ValueError("diagnostic_file_unsafe")
with os.fdopen(descriptor, encoding="utf-8", errors="replace") as handle:
descriptor = -1
return handle.read()
finally:
if descriptor >= 0:
os.close(descriptor)
def current_trace_id() -> str | None:
return _trace_id.get()
@@ -840,77 +837,8 @@ def diagnostic_exception(error: BaseException) -> dict[str, object]:
}
def _contains_credential_material(value: str) -> bool:
return bool(
re.search(
r"(?:^|[^A-Za-z0-9])(?:sk-|sk_|omhg_|ghp_|github_pat_)[A-Za-z0-9._-]+",
value,
re.IGNORECASE,
)
or re.search(r"(?:^|[^A-Za-z0-9])xox[baprs]-[A-Za-z0-9-]+", value, re.IGNORECASE)
or re.search(r"(?:^|[^A-Za-z0-9])(?:AKIA|ASIA)[A-Z0-9]{16}(?![A-Za-z0-9])", value)
or re.search(r"(?:^|[^A-Za-z0-9])AKID[A-Za-z0-9]{12,}", value, re.IGNORECASE)
or re.search(r"(?:^|[^A-Za-z0-9])AIza[A-Za-z0-9_-]{20,}", value)
or re.search(r"(?:^|[^A-Za-z0-9])GOCSPX-[A-Za-z0-9_-]+", value)
or any(
_is_conservative_short_credential(match.group(1))
for match in _SHORT_CREDENTIAL_TOKEN.finditer(value)
)
or re.search(
r"\b[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\.[A-Za-z0-9_-]{10,}\b",
value,
)
or any(
len(candidate) >= 48
and any(character.isalpha() for character in candidate)
and any(character.isdigit() for character in candidate)
for candidate in re.findall(r"\b[A-Za-z0-9._-]{48,}\b", value)
)
)
def _is_conservative_short_credential(value: str) -> bool:
return bool(
re.search(r"[A-Z]", value)
and re.search(r"[a-z]", value)
and re.search(r"\d", value)
and not re.fullmatch(r"[A-Fa-f0-9]+", value)
)
def _export_identifier(value: str) -> bool:
return bool(
_IDENTIFIER.fullmatch(value)
and not _EXPORT_SECRET_MARKER.search(value)
and not _contains_credential_material(value)
)
def _trace_details(values: Mapping[object, object]) -> dict[str, str | int | float]:
result: dict[str, str | int | float] = {}
for key, value in values.items():
if (
key in _TRACE_STRING_DETAIL_KEYS
and isinstance(value, str)
and _export_identifier(value)
):
result[str(key)] = value
elif key in _TRACE_NUMBER_DETAIL_KEYS and isinstance(value, (int, float)):
if isinstance(value, bool):
continue
safe_integer = isinstance(value, int) and abs(value) <= _MAX_SAFE_INTEGER
safe_float = (
isinstance(value, float)
and math.isfinite(value)
and (not value.is_integer() or abs(value) <= _MAX_SAFE_INTEGER)
)
if safe_integer or safe_float:
result[str(key)] = value
return result
def _serialize(value: Mapping[str, object]) -> str:
return json.dumps(value, ensure_ascii=True, allow_nan=False, separators=(",", ":")) + "\n"
return json.dumps(value, ensure_ascii=True, separators=(",", ":")) + "\n"
def _uuid(value: str | None) -> str | None:
@@ -936,47 +864,24 @@ def _event_name(value: str) -> str:
def _safe_stack_line(value: str) -> str:
if (
not value
or len(value) > 256
or any(character in value for character in "\0\r\n")
or _contains_credential_material(value)
):
if not value or len(value) > 256 or "\n" in value or "\r" in value:
raise ValueError("diagnostic_stack_invalid")
if _CODE_FRAME.fullmatch(value):
return value
if value.startswith("thread|"):
frames = value.removeprefix("thread|").split(" > ")
if frames and all(_CODE_FRAME.fullmatch(frame) for frame in frames):
return value
raise ValueError("diagnostic_stack_invalid")
if Path(value.split(":", 1)[0]).is_absolute() or "file://" in value:
raise ValueError("diagnostic_stack_invalid")
return value
def _details(
values: Mapping[str, str | int | float | bool | None],
) -> dict[str, str | int | float | bool | None]:
if len(values) > _MAX_DETAIL_KEYS:
raise ValueError("diagnostic_detail_count_invalid")
result: dict[str, str | int | float | bool | None] = {}
for key, value in values.items():
if (
not isinstance(key, str)
or not _DETAIL_KEY.fullmatch(key)
or key in _REFUSED_DETAIL_KEYS
):
if key not in _DETAIL_KEYS:
raise ValueError("diagnostic_detail_key_invalid")
if isinstance(value, str):
if not value or len(value) > 256 or any(character in value for character in "\0\r\n"):
if not value or len(value) > 256 or "\n" in value or "\r" in value:
raise ValueError("diagnostic_detail_value_invalid")
elif isinstance(value, float):
if not math.isfinite(value) or (value.is_integer() and abs(value) > _MAX_SAFE_INTEGER):
raise ValueError("diagnostic_detail_value_invalid")
elif isinstance(value, bool) or value is None:
pass
elif isinstance(value, int):
if abs(value) > _MAX_SAFE_INTEGER:
raise ValueError("diagnostic_detail_value_invalid")
else:
elif value is not None and not isinstance(value, (int, float, bool)):
raise ValueError("diagnostic_detail_value_invalid")
result[key] = value
return result
@@ -3,8 +3,8 @@
Every site that catches `Exception` broadly must call `report_exception`
before (or instead of) continuing, so a failure never looks like success
and a stack is always on disk. Content (conversation text, keys, SQL
parameters, and exception messages) never reaches the diagnostic log: only
exception type, a repeat count, and file:line:function frames.
parameters) never reaches the log: only exception type, a truncated
message, and file:line:function frames.
"""
from __future__ import annotations
@@ -67,7 +67,7 @@ def report_exception(
event=event,
outcome="failed",
**payload,
details={"count": _RECENT[key]},
details={"reason": _safe_message(error), "count": _RECENT[key]},
)
if full:
_logger.error(
@@ -14,7 +14,7 @@ from pathlib import Path
from typing import Literal, cast
HY_MEMORY_DISTRIBUTION = "hy-memory"
HY_MEMORY_VERSION = "2.0.5"
HY_MEMORY_VERSION = "2.0.5.dev3"
HY_MEMORY_NAMESPACE = "governed-v2"
DiagnosticLogLevel = Literal["error", "warning", "info", "debug"]
@@ -1,13 +1,18 @@
"""Offline snapshot and restore of the HY SDK data plane for the App migration.
"""Privileged offline access to the HY SDK data plane during App migration.
Runtime product code must use public ``HyMemoryClient`` methods. The startup
migration bundle is the isolated exception: all writers are stopped, so it may
copy and restore the SDK-owned local stores as files.
Runtime product code must use public ``HyMemoryClient`` methods. This module
is the deliberately isolated exception: App startup has stopped all writers,
so the one-time legacy migration may access SDK stores directly without
turning migration mechanics into public Client APIs.
"""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Mapping
from datetime import datetime
from pathlib import Path
from typing import Any, Protocol
from hy_memory import ( # type: ignore[import-untyped]
DataPlaneRestore,
@@ -15,6 +20,380 @@ from hy_memory import ( # type: ignore[import-untyped]
restore_local_data_plane,
snapshot_local_data_plane,
)
from hy_memory.models.memory import ( # type: ignore[import-untyped]
MemoryLayer,
MemoryNode,
MemoryStatus,
SourceType,
)
from hy_memory.utils.source_filter import normalize_source # type: ignore[import-untyped]
from one_memory.hy_sdk_types import (
HySdkMemoryPage,
HySdkMemoryRecord,
HySdkMigrationError,
HySdkRawPage,
HySdkRawRecord,
)
class MigrationClient(Protocol):
_loop_thread: Any
_vector_store: Any
_embed_service: Any
_graph_store: Any
def list_memories(
self,
*,
user_id: str,
agent_id: str | None = None,
statuses: list[str] | None = None,
limit: int = 100,
offset: int = 0,
order: str = "desc",
) -> Mapping[str, object]: ...
def delete(self, memory_id: str) -> Mapping[str, object]: ...
def get_write_outcome(self, request_id: str) -> Mapping[str, object] | None: ...
class HySdkMigration:
"""Typed offline adapter over one initialized SDK data plane."""
def __init__(self, client: MigrationClient) -> None:
self.client = client
self._loop_thread = client._loop_thread
self._vector_store = client._vector_store
self._embed_service = client._embed_service
self._graph_store = client._graph_store
async def list_memories(
self,
*,
user_id: str,
agent_id: str | None = None,
statuses: tuple[str, ...] = ("active",),
limit: int = 100,
offset: int = 0,
order: str = "desc",
) -> HySdkMemoryPage:
raw = await _sdk_call(
"hy_sdk_memory_list_failed",
self.client.list_memories,
user_id=user_id,
agent_id=agent_id,
statuses=list(statuses),
limit=limit,
offset=offset,
order=order,
)
return _memory_page(raw, limit=limit, offset=offset)
async def write_legacy_memory(
self,
*,
content: str,
memory_id: str,
user_id: str,
agent_id: str,
session_id: str,
source: str,
layer: str,
status: str,
owner: str | None,
tags: tuple[str, ...],
supersedes: tuple[str, ...],
superseded_by: tuple[str, ...],
is_latest: bool,
memory_at: datetime | None,
metadata: Mapping[str, object],
request_id: str,
) -> dict[str, object]:
raw = await _sdk_call(
"hy_sdk_memory_import_failed",
self._loop_thread.run,
self._write_legacy_memory(
content=content,
memory_id=memory_id,
user_id=user_id,
agent_id=agent_id,
session_id=session_id,
source=source,
layer=layer,
status=status,
owner=owner,
tags=tags,
supersedes=supersedes,
superseded_by=superseded_by,
is_latest=is_latest,
memory_at=memory_at,
metadata=metadata,
request_id=request_id,
),
)
return _successful_result(raw, "hy_sdk_memory_import_failed")
async def _write_legacy_memory(
self,
*,
content: str,
memory_id: str,
user_id: str,
agent_id: str,
session_id: str,
source: str,
layer: str,
status: str,
owner: str | None,
tags: tuple[str, ...],
supersedes: tuple[str, ...],
superseded_by: tuple[str, ...],
is_latest: bool,
memory_at: datetime | None,
metadata: Mapping[str, object],
request_id: str,
) -> dict[str, object]:
if not content or not memory_id or not user_id:
raise ValueError("content, memory_id and user_id are required")
normalized_layer = MemoryLayer.from_string(layer)
if normalized_layer is MemoryLayer.L1_RAW:
raise ValueError("raw content must use the raw migration lane")
normalized_status = MemoryStatus(status.lower())
normalized_source = normalize_source(source)
existing = await self._vector_store.get_by_id(memory_id)
if existing is not None:
identity = (
existing.content == content,
existing.user_id == user_id,
existing.agent_id == agent_id,
existing.session_id == session_id,
existing.source == normalized_source,
existing.layer == normalized_layer,
existing.status == normalized_status,
existing.owner == owner,
list(existing.tags or []) == list(tags),
list(existing.supersedes or []) == list(supersedes),
list(existing.superseded_by or []) == list(superseded_by),
existing.is_latest == is_latest,
)
if not all(identity):
raise ValueError("memory_id is already bound to different memory content")
await self._project_graph(existing)
return {
"success": True,
"memory_id": memory_id,
"memory_ids": [memory_id],
"request_id": request_id,
"replayed": True,
}
custom = dict(metadata)
custom["migration_request_id"] = request_id
embedding = await self._embed_service.embed(content)
node = MemoryNode(
node_id=memory_id,
user_id=user_id,
agent_id=agent_id,
session_id=session_id,
source=normalized_source,
layer=normalized_layer,
content=content,
source_type=SourceType.EXPLICIT,
status=normalized_status,
owner=owner,
tags=list(tags),
supersedes=list(supersedes),
superseded_by=list(superseded_by),
is_latest=is_latest,
embedding=embedding,
memory_at=memory_at,
custom=custom,
)
stored_id = await self._vector_store.upsert(node)
if stored_id != memory_id:
raise RuntimeError("exact migration changed the requested memory identity")
await self._project_graph(node)
return {
"success": True,
"memory_id": memory_id,
"memory_ids": [memory_id],
"request_id": request_id,
}
async def _project_graph(self, node: MemoryNode) -> None:
if self._graph_store is not None:
await self._graph_store.upsert_memory_node(node)
async def read_raw_inventory(
self,
*,
user_id: str,
agent_id: str | None = None,
session_id: str | None = None,
source: str | None = None,
limit: int = 100,
offset: int = 0,
order: str = "desc",
) -> HySdkRawPage:
raw = await _sdk_call(
"hy_sdk_raw_list_failed",
self._loop_thread.run,
self._read_raw_inventory(
user_id=user_id,
agent_id=agent_id,
session_id=session_id,
source=source,
limit=limit,
offset=offset,
order=order,
),
)
return _raw_page(raw, limit=limit, offset=offset)
async def _read_raw_inventory(
self,
*,
user_id: str,
agent_id: str | None,
session_id: str | None,
source: str | None,
limit: int,
offset: int,
order: str,
) -> dict[str, object]:
if not user_id:
raise ValueError("user_id is required")
nodes = await self._vector_store.list_by_user(
user_id=user_id,
agent_id=agent_id,
status_filter=[MemoryStatus.ACTIVE],
layers=[MemoryLayer.L1_RAW],
include_sources=[source] if source else None,
limit=100_000,
only_latest=False,
with_vectors=False,
allow_raw_layer=True,
)
if session_id:
nodes = [node for node in nodes if node.session_id == session_id]
nodes.sort(
key=_node_timestamp,
reverse=order.lower() != "asc",
)
return {
"raw": [_raw_node(node) for node in nodes[offset : offset + limit]],
"total": len(nodes),
"limit": limit,
"offset": offset,
}
async def write_legacy_raw(
self,
*,
content: str,
raw_id: str,
user_id: str,
agent_id: str,
session_id: str,
source: str,
memory_at: datetime | None,
metadata: Mapping[str, object],
request_id: str,
) -> dict[str, object]:
raw = await _sdk_call(
"hy_sdk_raw_import_failed",
self._loop_thread.run,
self._write_legacy_raw(
content=content,
raw_id=raw_id,
user_id=user_id,
agent_id=agent_id,
session_id=session_id,
source=source,
memory_at=memory_at,
metadata=metadata,
request_id=request_id,
),
)
return _successful_result(raw, "hy_sdk_raw_import_failed")
async def _write_legacy_raw(
self,
*,
content: str,
raw_id: str,
user_id: str,
agent_id: str,
session_id: str,
source: str,
memory_at: datetime | None,
metadata: Mapping[str, object],
request_id: str,
) -> dict[str, object]:
if not content or not raw_id or not user_id:
raise ValueError("content, raw_id and user_id are required")
normalized_source = normalize_source(source)
existing = await self._vector_store.get_by_id(raw_id)
if existing is not None:
identity = (
existing.layer is MemoryLayer.L1_RAW,
existing.content == content,
existing.user_id == user_id,
existing.agent_id == agent_id,
existing.session_id == session_id,
existing.source == normalized_source,
)
if not all(identity):
raise ValueError("raw_id is already bound to different source content")
return {
"success": True,
"raw_id": raw_id,
"request_id": request_id,
"replayed": True,
}
custom = dict(metadata)
custom["migration_request_id"] = request_id
embedding = await self._embed_service.embed(content)
node = MemoryNode(
node_id=raw_id,
user_id=user_id,
agent_id=agent_id,
session_id=session_id,
source=normalized_source,
layer=MemoryLayer.L1_RAW,
content=content,
source_type=SourceType.EXPLICIT,
status=MemoryStatus.ACTIVE,
is_latest=True,
embedding=embedding,
memory_at=memory_at,
custom=custom,
)
stored_id = await self._vector_store.upsert(node)
if stored_id != raw_id:
raise RuntimeError("exact migration changed the requested raw identity")
return {"success": True, "raw_id": raw_id, "request_id": request_id}
async def delete(self, memory_id: str) -> dict[str, object]:
raw = await _sdk_call(
"hy_sdk_memory_delete_failed",
self.client.delete,
memory_id,
)
return _successful_result(raw, "hy_sdk_memory_delete_failed")
async def get_write_outcome(self, request_id: str) -> dict[str, object] | None:
raw = await _sdk_call(
"hy_sdk_write_outcome_failed",
self.client.get_write_outcome,
request_id,
)
if raw is None:
return None
if not isinstance(raw, Mapping):
raise HySdkMigrationError("hy_sdk_write_outcome_shape_invalid")
return dict(raw)
def snapshot_data_plane(
@@ -45,3 +424,176 @@ def restore_data_plane(
writers_stopped=writers_stopped,
protective_root=protective_root,
)
async def _sdk_call[T](
code: str,
call: Callable[..., T],
*args: object,
**kwargs: object,
) -> T:
pending = asyncio.create_task(asyncio.to_thread(call, *args, **kwargs))
try:
return await asyncio.shield(pending)
except asyncio.CancelledError:
await _finish_cancelled_call(pending)
raise
except Exception as error:
raise HySdkMigrationError(code) from error
async def _finish_cancelled_call[T](pending: asyncio.Task[T]) -> None:
while not pending.done():
try:
await asyncio.shield(pending)
except asyncio.CancelledError:
continue
except Exception:
return
if not pending.cancelled():
pending.exception()
def _memory_page(raw: object, *, limit: int, offset: int) -> HySdkMemoryPage:
if not isinstance(raw, Mapping):
raise HySdkMigrationError("hy_sdk_memory_page_shape_invalid")
rows: list[object] = []
total = 0
for bucket_name, row_name in (("vdb", "memories"), ("graph", "nodes")):
bucket = raw.get(bucket_name)
if bucket is None and bucket_name == "graph":
continue
if not isinstance(bucket, Mapping):
raise HySdkMigrationError("hy_sdk_memory_page_shape_invalid")
bucket_rows = bucket.get(row_name)
bucket_total = bucket.get("total")
if not isinstance(bucket_rows, list) or type(bucket_total) is not int or bucket_total < 0:
raise HySdkMigrationError("hy_sdk_memory_page_shape_invalid")
rows.extend(bucket_rows)
total += bucket_total
records: dict[str, HySdkMemoryRecord] = {}
for row in rows:
record = _memory_record(row)
records.setdefault(record.memory_id, record)
return HySdkMemoryPage(
memories=tuple(records.values()),
total=total,
limit=limit,
offset=offset,
)
def _memory_record(raw: object) -> HySdkMemoryRecord:
if not isinstance(raw, Mapping):
raise HySdkMigrationError("hy_sdk_memory_record_shape_invalid")
required = ("memory_id", "content", "layer", "status", "user_id", "agent_id")
if any(not isinstance(raw.get(name), str) or not raw[name] for name in required):
raise HySdkMigrationError("hy_sdk_memory_record_shape_invalid")
custom = raw.get("custom")
source_raw_ids: list[object] = []
if isinstance(custom, Mapping) and isinstance(custom.get("source_raw_memory_ids"), list):
source_raw_ids.extend(custom["source_raw_memory_ids"])
if isinstance(raw.get("source_raw_memory_id"), str):
source_raw_ids.append(raw["source_raw_memory_id"])
try:
return HySdkMemoryRecord(
memory_id=raw["memory_id"],
content=raw["content"],
layer=raw["layer"],
status=raw["status"],
is_latest=bool(raw.get("is_latest", True)),
supersedes=_string_tuple(raw.get("supersedes")),
superseded_by=_string_tuple(raw.get("superseded_by")),
tags=_string_tuple(raw.get("tags")),
owner=raw.get("owner") if raw.get("owner") in {"user", "agent"} else None,
source_raw_ids=_string_tuple(source_raw_ids),
user_id=raw["user_id"],
agent_id=raw["agent_id"],
session_id=_optional_string(raw.get("session_id")),
source=_optional_string(raw.get("source")),
memory_at=_optional_int(raw.get("memory_at")),
gmt_created=_optional_int(raw.get("gmt_created")),
)
except (TypeError, ValueError) as error:
raise HySdkMigrationError("hy_sdk_memory_record_shape_invalid") from error
def _raw_page(raw: object, *, limit: int, offset: int) -> HySdkRawPage:
if not isinstance(raw, Mapping):
raise HySdkMigrationError("hy_sdk_raw_page_shape_invalid")
rows = raw.get("raw")
total = raw.get("total")
if not isinstance(rows, list) or type(total) is not int or total < 0:
raise HySdkMigrationError("hy_sdk_raw_page_shape_invalid")
return HySdkRawPage(
raw=tuple(_raw_record(row) for row in rows),
total=total,
limit=limit,
offset=offset,
)
def _node_timestamp(node: Any) -> float:
value = getattr(node, "memory_at", None) or getattr(node, "gmt_created", None)
return value.timestamp() if isinstance(value, datetime) else 0.0
def _raw_node(node: Any) -> dict[str, object]:
def timestamp(value: object) -> int | None:
return int(value.timestamp()) if isinstance(value, datetime) else None
return {
"raw_id": node.node_id,
"content": node.content,
"user_id": node.user_id,
"agent_id": node.agent_id,
"session_id": node.session_id,
"source": node.source,
"memory_at": timestamp(node.memory_at),
"gmt_created": timestamp(node.gmt_created),
"custom": node.custom or {},
}
def _raw_record(raw: object) -> HySdkRawRecord:
if not isinstance(raw, Mapping):
raise HySdkMigrationError("hy_sdk_raw_record_shape_invalid")
required = ("raw_id", "content", "user_id", "agent_id", "session_id", "source")
if any(not isinstance(raw.get(name), str) or not raw[name] for name in required):
raise HySdkMigrationError("hy_sdk_raw_record_shape_invalid")
custom = raw.get("custom")
try:
return HySdkRawRecord(
raw_id=raw["raw_id"],
content=raw["content"],
user_id=raw["user_id"],
agent_id=raw["agent_id"],
session_id=raw["session_id"],
source=raw["source"],
memory_at=_optional_int(raw.get("memory_at")),
gmt_created=_optional_int(raw.get("gmt_created")),
custom=dict(custom) if isinstance(custom, Mapping) else {},
)
except (TypeError, ValueError) as error:
raise HySdkMigrationError("hy_sdk_raw_record_shape_invalid") from error
def _successful_result(raw: object, code: str) -> dict[str, object]:
if not isinstance(raw, Mapping) or raw.get("success") is not True:
raise HySdkMigrationError(code)
return dict(raw)
def _string_tuple(value: object) -> tuple[str, ...]:
if not isinstance(value, list | tuple):
return ()
return tuple(item for item in value if isinstance(item, str) and item)
def _optional_string(value: object) -> str | None:
return value if isinstance(value, str) else None
def _optional_int(value: object) -> int | None:
return value if type(value) is int else None
@@ -557,10 +557,12 @@ class PollingLiveSyncSupervisor:
)
def _note_internal_error(self, source_id: str, error: Exception) -> None:
# The ordinary local log keeps the exception message for debugging.
# The structured diagnostic event deliberately keeps only the stable
# error code, source id, type, and relative stack: exception messages
# can contain paths, query text, or credentials and are exportable data.
# Type alone used to be all this said. It is not enough to act on:
# `OperationalError` covers a locked database, a missing table and a
# corrupt page alike, and without the message and the frames two dozen
# of these in a log could only be guessed at. The stack is the
# redacted, `one_memory`-relative one, so it names the call site
# without printing anyone's paths or query text.
logging.getLogger(__name__).warning(
"live_sync_internal_error %s: %s", type(error).__name__, error
)
@@ -570,7 +572,10 @@ class PollingLiveSyncSupervisor:
event="live_sync.internal_error",
outcome="failed",
error_code="live_sync_internal_error",
details={"source_id": source_id},
# `reason` is the listed key that can carry the message; see
# `diagnostic_log._DETAIL_KEYS`. Truncated because a detail value
# longer than 256 characters is rejected.
details={"source_id": source_id, "reason": str(error)[:200]},
**diagnostic_exception(error),
)
if isinstance(error, sqlite3.Error):
@@ -867,7 +867,10 @@ class ImportRunner:
outcome="failed",
job_id=job_id,
error_code="import_heartbeat_failed",
details={"phase": "heartbeat"},
# `reason` is the listed key for this; see
# `diagnostic_log._DETAIL_KEYS`. Truncated: a detail value
# over 256 characters is rejected.
details={"reason": str(error)[:200]},
**diagnostic_exception(error),
)
@@ -1,7 +1,6 @@
"""Write visible hook text into Evidence. Tool output stays out."""
import json
import os
import time
from datetime import UTC, datetime
from enum import StrEnum
@@ -75,7 +74,6 @@ _FILE_FALLBACK = {
}
_FALLBACK_EVENTS = frozenset({"Stop", "stop", "SubagentStop", "StopFailure"})
_TEXT_BLOCKS = frozenset({"text"})
SDK_MIGRATION_CAPTURE_FENCE = ".sdk-migration-capture-fence.json"
def database_path() -> Path:
@@ -90,42 +88,17 @@ def capture_stdio(source_id: str, stdin: TextIO, output: TextIO) -> int:
del output
if source_id not in CAPTURE_SOURCES:
return 2
database = database_path()
if sdk_migration_capture_fenced(database.parent):
return 0
try:
payload = json.loads(stdin.read() or "")
except json.JSONDecodeError:
return 0
try:
capture_hook(source_id, payload, database)
capture_hook(source_id, payload, database_path())
except Exception:
return 0
return 0
def sdk_migration_capture_fenced(data_dir: Path) -> bool:
"""Ignore stale fences; only a live Desktop migration pauses hook writes."""
marker = data_dir / SDK_MIGRATION_CAPTURE_FENCE
try:
document = json.loads(marker.read_text(encoding="utf-8"))
pid = int(document["pid"])
if pid < 1:
raise ValueError("invalid migration fence pid")
os.kill(pid, 0)
return True
except FileNotFoundError:
return False
except ProcessLookupError:
marker.unlink(missing_ok=True)
return False
except PermissionError:
return True
except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError):
marker.unlink(missing_ok=True)
return False
def capture_hook(source_id: str, payload: object, database: Path) -> bool:
if source_id not in CAPTURE_SOURCES or not isinstance(payload, dict):
return False
@@ -14,7 +14,6 @@ stuck in it.
"""
import logging
import re
import sys
import threading
import time
@@ -24,7 +23,6 @@ from types import FrameType
logger = logging.getLogger("one_memory.loop_watchdog")
DEFAULT_STALL_SECONDS = 2.0
_SAFE_FALLBACK_FILE = re.compile(r"^[A-Za-z0-9_][A-Za-z0-9_.-]*\.py$")
class LoopWatchdog:
@@ -122,28 +120,22 @@ def _safe_stack_for_main_thread() -> tuple[str, ...]:
def _safe_stacks_for_all_threads() -> tuple[str, ...]:
"""Return bounded stack summaries without recording thread names."""
"""Return one content-free, bounded stack summary for every live thread."""
names = {thread.ident: thread.name for thread in threading.enumerate()}
lines: list[str] = []
for _ident, frame in sorted(sys._current_frames().items()):
for ident, frame in sorted(sys._current_frames().items()):
stack: list[str] = []
# Most specific frame first; append only complete frames so truncation
# can never turn a valid location into free text.
# Most specific frame first so the 256-character privacy bound keeps
# the likely culprit rather than only the thread bootstrap frames.
for entry in reversed(traceback.extract_stack(frame)[-8:]):
parts = entry.filename.replace("\\", "/").split("/")
try:
root = parts.index("one_memory")
relative = "/".join(parts[root:])
except ValueError:
fallback = parts[-1]
relative = fallback if _SAFE_FALLBACK_FILE.fullmatch(fallback) else "unknown.py"
relative = parts[-1]
name = entry.name if entry.name.isidentifier() else "call"
candidate = f"{relative}:{entry.lineno}:{name}"
separator = "" if not stack else " > "
if len("thread|") + len(" > ".join(stack)) + len(separator) + len(candidate) > 256:
break
stack.append(candidate)
if stack:
lines.append(f"thread|{' > '.join(stack)}")
if len(lines) == 16:
break
stack.append(f"{relative}:{entry.lineno}:{name}")
thread_name = names.get(ident, "unknown").replace("\n", " ").replace("\r", " ")
lines.append(f"thread={thread_name[:48]}|{' > '.join(stack)}"[:256])
return tuple(lines)
@@ -64,6 +64,7 @@ EXEMPT_PATHS = frozenset(
"memory_server/sqlite_semantics.py",
"memory_tasks.py",
"sdk_data_plane_migration.py",
"sdk_migration_inventory.py",
"target_schema.py",
# D93 A: the visibility cache is SQLite-only by construction. It is a
# derived table with triggers on the six tables the read check reads —
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
import sqlite3
from dataclasses import dataclass
from datetime import UTC, datetime
@@ -39,6 +40,10 @@ class MigrationProgress:
migration_id: str
state: str
phase: str
counts: dict[str, int]
TERMINAL_ITEM_STATES = frozenset({"verified", "adopted", "deleted", "skipped"})
class SdkDataPlaneMigrationStore:
@@ -131,14 +136,37 @@ class SdkDataPlaneMigrationStore:
""",
values,
)
elif backup_bundle_id and existing["backup_bundle_id"] is None:
connection.execute(
"UPDATE sdk_data_plane_migrations "
"SET backup_bundle_id=?,updated_at=? WHERE migration_id=?",
(backup_bundle_id, now, migration_id),
else:
identity = (
source_runtime_build,
target_sdk_version,
source_schema_version,
namespace_fingerprint,
backend_fingerprint,
)
elif backup_bundle_id and existing["backup_bundle_id"] != backup_bundle_id:
raise DataPlaneMigrationError("sdk_data_plane_migration_bundle_conflict")
existing_identity = tuple(
existing[key]
for key in (
"source_runtime_build",
"target_sdk_version",
"source_schema_version",
"namespace_fingerprint",
"backend_fingerprint",
)
)
if existing_identity != identity:
raise DataPlaneMigrationError("sdk_data_plane_migration_identity_conflict")
if backup_bundle_id and existing["backup_bundle_id"] not in {
None,
backup_bundle_id,
}:
raise DataPlaneMigrationError("sdk_data_plane_migration_bundle_conflict")
if backup_bundle_id and existing["backup_bundle_id"] is None:
connection.execute(
"UPDATE sdk_data_plane_migrations "
"SET backup_bundle_id=?,updated_at=? WHERE migration_id=?",
(backup_bundle_id, now, migration_id),
)
connection.execute("COMMIT")
return self.progress(migration_id)
@@ -154,18 +182,211 @@ class SdkDataPlaneMigrationStore:
raise DataPlaneMigrationError("sdk_data_plane_migration_not_mutable")
connection.execute("COMMIT")
def upsert_item(
self,
*,
migration_id: str,
item_key: str,
source_kind: str,
source_id: str,
source_hash: str,
operation_key: str,
source_revision: str | None = None,
state: str = "pending",
) -> None:
now = self._now()
with main_database(self.database_path).write(
autocommit=True, row_factory=sqlite3.Row
) as connection:
connection.execute("BEGIN IMMEDIATE")
row = connection.execute(
"SELECT * FROM sdk_data_plane_migration_items WHERE migration_id=? AND item_key=?",
(migration_id, item_key),
).fetchone()
if row is None:
try:
connection.execute(
"""
INSERT INTO sdk_data_plane_migration_items (
migration_id,item_key,source_kind,source_id,
source_revision,source_hash,operation_key,state,updated_at
) VALUES (?,?,?,?,?,?,?,?,?)
""",
(
migration_id,
item_key,
source_kind,
source_id,
source_revision,
source_hash,
operation_key,
state,
now,
),
)
except sqlite3.IntegrityError as error:
raise DataPlaneMigrationError(
"sdk_data_plane_migration_operation_conflict"
) from error
else:
if str(row["state"]) in TERMINAL_ITEM_STATES and (
source_id,
source_revision,
source_hash,
) == (
row["source_id"],
row["source_revision"],
row["source_hash"],
):
# A crash after an SDK side effect can make the next
# inventory classify the same source as verified/adopted
# instead of import/delete. Keep the settled operation as
# the durable audit identity.
connection.execute("COMMIT")
return
identity = (
source_kind,
source_id,
source_revision,
source_hash,
operation_key,
)
if identity != tuple(
row[key]
for key in (
"source_kind",
"source_id",
"source_revision",
"source_hash",
"operation_key",
)
):
raise DataPlaneMigrationError("sdk_data_plane_migration_item_conflict")
connection.execute("COMMIT")
def settle_item(
self,
*,
migration_id: str,
item_key: str,
state: str,
target_memory_id: str | None = None,
target_hash: str | None = None,
error_code: str | None = None,
) -> None:
with main_database(self.database_path).write(autocommit=True) as connection:
connection.execute("BEGIN IMMEDIATE")
cursor = connection.execute(
"""
UPDATE sdk_data_plane_migration_items
SET state=?,target_memory_id=?,target_hash=?,error_code=?,
attempts=attempts+1,updated_at=?
WHERE migration_id=? AND item_key=?
""",
(
state,
target_memory_id,
target_hash,
error_code,
self._now(),
migration_id,
item_key,
),
)
if cursor.rowcount != 1:
raise DataPlaneMigrationError("sdk_data_plane_migration_item_not_found")
connection.execute("COMMIT")
def record_alias(
self,
*,
migration_id: str,
legacy_memory_id: str,
canonical_memory_id: str,
) -> None:
with main_database(self.database_path).write(
autocommit=True, row_factory=sqlite3.Row
) as connection:
connection.execute("BEGIN IMMEDIATE")
existing = connection.execute(
"SELECT canonical_memory_id,migration_id FROM memory_id_aliases "
"WHERE legacy_memory_id=?",
(legacy_memory_id,),
).fetchone()
if existing is None:
connection.execute(
"""
INSERT INTO memory_id_aliases (
legacy_memory_id,canonical_memory_id,migration_id,created_at
) VALUES (?,?,?,?)
""",
(
legacy_memory_id,
canonical_memory_id,
migration_id,
self._now(),
),
)
elif existing["canonical_memory_id"] != canonical_memory_id:
raise DataPlaneMigrationError("sdk_data_plane_alias_conflict")
connection.execute("COMMIT")
def record_raw_alias(
self,
*,
migration_id: str,
legacy_evidence_id: str,
canonical_raw_id: str,
item_index: int = 0,
) -> None:
with main_database(self.database_path).write(
autocommit=True,
row_factory=sqlite3.Row,
) as connection:
connection.execute("BEGIN IMMEDIATE")
existing = connection.execute(
"SELECT canonical_raw_id FROM raw_id_aliases "
"WHERE legacy_evidence_id=? AND item_index=?",
(legacy_evidence_id, item_index),
).fetchone()
if existing is None:
connection.execute(
"INSERT INTO raw_id_aliases "
"(legacy_evidence_id,item_index,canonical_raw_id,migration_id,created_at) "
"VALUES (?,?,?,?,?)",
(
legacy_evidence_id,
item_index,
canonical_raw_id,
migration_id,
self._now(),
),
)
elif existing["canonical_raw_id"] != canonical_raw_id:
raise DataPlaneMigrationError("sdk_data_plane_raw_alias_conflict")
connection.execute("COMMIT")
def progress(self, migration_id: str) -> MigrationProgress:
with main_database(self.database_path).read(row_factory=sqlite3.Row) as connection:
migration = connection.execute(
"SELECT state,phase FROM sdk_data_plane_migrations WHERE migration_id=?",
(migration_id,),
).fetchone()
if migration is None:
raise DataPlaneMigrationError("sdk_data_plane_migration_not_found")
if migration is None:
raise DataPlaneMigrationError("sdk_data_plane_migration_not_found")
counts = {
str(row["state"]): int(row["count"])
for row in connection.execute(
"SELECT state,COUNT(*) AS count "
"FROM sdk_data_plane_migration_items WHERE migration_id=? GROUP BY state",
(migration_id,),
)
}
return MigrationProgress(
migration_id=migration_id,
state=str(migration["state"]),
phase=str(migration["phase"]),
counts=counts,
)
def commit_cutover(
@@ -188,8 +409,23 @@ class SdkDataPlaneMigrationStore:
).fetchone()
if migration is None:
raise DataPlaneMigrationError("sdk_data_plane_migration_not_found")
if migration["phase"] != "prepared" or migration["state"] != "ready":
if migration["phase"] != "reconciled" or migration["state"] != "ready":
raise DataPlaneMigrationError("sdk_data_plane_migration_not_ready")
unresolved = connection.execute(
"SELECT state,COUNT(*) AS count FROM sdk_data_plane_migration_items "
"WHERE migration_id=? GROUP BY state",
(migration_id,),
).fetchall()
blocked = {
str(row["state"]): int(row["count"])
for row in unresolved
if str(row["state"]) not in TERMINAL_ITEM_STATES
}
if blocked:
raise DataPlaneMigrationError(
"sdk_data_plane_migration_items_unresolved:"
+ json.dumps(blocked, sort_keys=True, separators=(",", ":"))
)
if migration["backup_bundle_id"] != backup_bundle_id:
raise DataPlaneMigrationError("sdk_data_plane_migration_bundle_conflict")
existing = connection.execute(
@@ -3,9 +3,6 @@
from __future__ import annotations
import asyncio
import logging
import re
from contextlib import suppress
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
@@ -28,21 +25,6 @@ from one_memory.memory_server.durable_queue import COMPLETED, DurableWriteQueue
from one_memory.sdk_write_processor import SdkWriteCoordinator
from one_memory.semantic_deadline import SemanticMemoryHit
# Only a code-shaped message is a code; anything else may carry provider text.
_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:
"""Product services that remain after legacy task/governance writes retire."""
@@ -61,68 +43,6 @@ class SdkMemoryRuntime:
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:
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="local",
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__,
)
async def semantic_recall(
self,
@@ -240,22 +160,8 @@ class SdkMemoryRuntime:
decision=decision,
)
async def _mutation_client(self) -> Any:
"""The write client, with no-model and keychain refusals as SDK codes.
Callers record these against the edit or deletion instead of failing
the request that already committed it.
"""
try:
return await self.coordinator.client()
except RuntimeError as error:
code = str(error)
if not _STABLE_CODE.fullmatch(code):
code = "hy_memory_client_unavailable"
raise HySdkError(code) from error
async def update_hy_memory(self, memory_id: str, content: str) -> dict[str, object]:
client = await self._mutation_client()
client = await self.coordinator.client()
try:
result = await asyncio.to_thread(client.update, memory_id, content)
except Exception as error:
@@ -265,7 +171,7 @@ class SdkMemoryRuntime:
return result
async def finish_deletion(self, memory_id: str) -> None:
client = await self._mutation_client()
client = await self.coordinator.client()
try:
result = await asyncio.to_thread(client.delete, memory_id)
except Exception as error:
@@ -373,12 +279,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, ...]:
@@ -0,0 +1,684 @@
"""Legacy Governance to SDK inventory classification for cutover planning."""
from __future__ import annotations
import hashlib
import json
import sqlite3
from collections.abc import Awaitable, Mapping
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Protocol
from one_memory.database_access import main_database
from one_memory.hy_sdk_types import HySdkMemoryPage, HySdkRawPage
from one_memory.sdk_data_plane_migration import SdkDataPlaneMigrationStore
def content_hash(content: str) -> str:
return hashlib.sha256(content.encode("utf-8")).hexdigest()
@dataclass(frozen=True)
class LegacyMemoryInventory:
legacy_memory_id: str
canonical_memory_id: str
user_id: str
agent_id: str
session_id: str
content: str
content_hash: str
status: str
layer: str | None
owner: str | None
tags: tuple[str, ...]
supersedes: tuple[str, ...]
superseded_by: tuple[str, ...]
@dataclass(frozen=True)
class SdkMemoryInventory:
memory_id: str
user_id: str
agent_id: str
content_hash: str
status: str
@dataclass(frozen=True)
class MemoryMigrationDecision:
item_key: str
action: str
legacy_memory_id: str | None
canonical_memory_id: str
reason: str
@dataclass(frozen=True)
class LegacyRawInventory:
evidence_id: str
user_id: str
agent_id: str
session_id: str
source: str
content: str
content_hash: str
memory_at: datetime | None
metadata: dict[str, object]
@dataclass(frozen=True)
class SdkRawInventory:
raw_id: str
user_id: str
agent_id: str
session_id: str
source: str
content_hash: str
@dataclass(frozen=True)
class RawMigrationDecision:
item_key: str
action: str
evidence_id: str | None
raw_id: str
reason: str
_KIND_LAYER = {
"chat": "l2_fact",
"schema": "l6_schema",
"intention": "l7_intention",
}
SDK_LIFECYCLE_STATUSES = (
"active",
"superseded",
"shadow",
"negated",
"conflicted",
"archived",
)
class SdkMigrationClient(Protocol):
def list_memories(
self,
*,
user_id: str,
agent_id: str | None = None,
statuses: tuple[str, ...] = ("active",),
limit: int = 100,
offset: int = 0,
order: str = "desc",
) -> Awaitable[HySdkMemoryPage]: ...
def write_legacy_memory(
self,
*,
content: str,
memory_id: str,
user_id: str,
agent_id: str,
session_id: str,
layer: str,
status: str,
owner: str | None,
tags: tuple[str, ...],
supersedes: tuple[str, ...],
superseded_by: tuple[str, ...],
is_latest: bool,
request_id: str,
) -> Awaitable[dict[str, object]]: ...
def read_raw_inventory(
self,
*,
user_id: str,
agent_id: str | None = None,
session_id: str | None = None,
source: str | None = None,
limit: int = 100,
offset: int = 0,
order: str = "desc",
) -> Awaitable[HySdkRawPage]: ...
def write_legacy_raw(
self,
*,
content: str,
raw_id: str,
user_id: str,
agent_id: str,
session_id: str,
source: str,
memory_at: datetime | None,
metadata: Mapping[str, object],
request_id: str,
) -> Awaitable[dict[str, object]]: ...
def delete(self, memory_id: str) -> Awaitable[dict[str, object]]: ...
def read_legacy_memory_inventory(database_path: Path) -> tuple[LegacyMemoryInventory, ...]:
"""Read legacy memory intent without constructing a Governance writer."""
with main_database(database_path).read(row_factory=sqlite3.Row) as connection:
mappings: dict[str, tuple[str, str | None, str | None]] = {}
for row in connection.execute("SELECT body FROM memory_task_outputs"):
try:
body = json.loads(row["body"])
except (TypeError, ValueError):
continue
legacy_id = body.get("memory_id")
sdk_id = body.get("hy_memory_id")
if isinstance(legacy_id, str) and isinstance(sdk_id, str):
mappings[legacy_id] = (
sdk_id,
body.get("layer") if isinstance(body.get("layer"), str) else None,
body.get("owner") if body.get("owner") in {"user", "agent"} else None,
)
supersedes: dict[str, list[str]] = {}
superseded_by: dict[str, list[str]] = {}
for row in connection.execute(
"SELECT memory_id,superseded_by FROM governed_memory_supersessions "
"ORDER BY memory_id,superseded_by"
):
older, newer = str(row["memory_id"]), str(row["superseded_by"])
superseded_by.setdefault(older, []).append(newer)
supersedes.setdefault(newer, []).append(older)
records: list[LegacyMemoryInventory] = []
for row in connection.execute(
"SELECT memory_id,user_id,kind,status,content,tags_json "
"FROM governed_memories ORDER BY memory_id"
):
legacy_id = str(row["memory_id"])
mapped_id, mapped_layer, mapped_owner = mappings.get(legacy_id, (legacy_id, None, None))
try:
tags_value = json.loads(row["tags_json"] or "[]")
except (TypeError, ValueError):
tags_value = []
tags = tuple(value for value in tags_value if isinstance(value, str))
kind = str(row["kind"])
status = str(row["status"])
records.append(
LegacyMemoryInventory(
legacy_memory_id=legacy_id,
canonical_memory_id=mapped_id,
user_id=str(row["user_id"]),
agent_id="one-memory",
session_id="default_session",
content=str(row["content"]),
content_hash=content_hash(str(row["content"])),
status=(
"active"
if status == "confirmed"
else "superseded"
if status == "superseded"
else "deleted"
),
layer=mapped_layer or _KIND_LAYER.get(kind),
owner=mapped_owner,
tags=tags,
supersedes=tuple(supersedes.get(legacy_id, ())),
superseded_by=tuple(superseded_by.get(legacy_id, ())),
)
)
return tuple(records)
def _parse_datetime(value: object) -> datetime | None:
if not isinstance(value, str) or not value:
return None
try:
return datetime.fromisoformat(value)
except ValueError:
return None
def read_legacy_raw_inventory(database_path: Path) -> tuple[LegacyRawInventory, ...]:
with main_database(database_path).read(row_factory=sqlite3.Row) as connection:
rows = connection.execute(
"""
SELECT evidence_id,user_id,source_id,workspace,project,
conversation_id,message_id,source_ordinal,content_kind,
role,written_by,created_at,imported_at,verbatim,content_hash,revision
FROM evidence ORDER BY COALESCE(ingest_seq,rowid),evidence_id
"""
).fetchall()
return tuple(
LegacyRawInventory(
evidence_id=str(row["evidence_id"]),
user_id=str(row["user_id"]),
agent_id="one-memory",
session_id=str(row["conversation_id"]),
source=str(row["source_id"]),
content=str(row["verbatim"]),
content_hash=str(row["content_hash"]),
memory_at=_parse_datetime(row["created_at"] or row["imported_at"]),
metadata={
key: row[key]
for key in (
"workspace",
"project",
"message_id",
"source_ordinal",
"content_kind",
"role",
"written_by",
"revision",
)
if row[key] is not None
},
)
for row in rows
)
def classify_memory_inventory(
legacy: tuple[LegacyMemoryInventory, ...],
sdk: tuple[SdkMemoryInventory, ...],
) -> tuple[MemoryMigrationDecision, ...]:
"""Classify every legacy and SDK memory; no item may silently disappear."""
sdk_by_id = {item.memory_id: item for item in sdk}
claimed_sdk_ids: set[str] = set()
decisions: list[MemoryMigrationDecision] = []
for item in legacy:
current = sdk_by_id.get(item.canonical_memory_id)
if current is not None:
claimed_sdk_ids.add(current.memory_id)
if item.status == "deleted":
decisions.append(
MemoryMigrationDecision(
item_key=f"memory:{item.legacy_memory_id}",
action="delete" if current is not None else "verified_deleted",
legacy_memory_id=item.legacy_memory_id,
canonical_memory_id=item.canonical_memory_id,
reason="legacy_user_delete",
)
)
continue
if item.layer is None:
decisions.append(
MemoryMigrationDecision(
item_key=f"memory:{item.legacy_memory_id}",
action="quarantine",
legacy_memory_id=item.legacy_memory_id,
canonical_memory_id=item.canonical_memory_id,
reason="legacy_layer_unknown",
)
)
continue
if current is None:
decisions.append(
MemoryMigrationDecision(
item_key=f"memory:{item.legacy_memory_id}",
action="import_exact",
legacy_memory_id=item.legacy_memory_id,
canonical_memory_id=item.canonical_memory_id,
reason="governance_only",
)
)
continue
if (
current.content_hash == item.content_hash
and current.status == item.status
and current.user_id == item.user_id
and current.agent_id == item.agent_id
):
action = (
"matched" if item.legacy_memory_id == item.canonical_memory_id else "record_alias"
)
reason = "same_identity" if action == "matched" else "mapped_identity"
else:
action, reason = "quarantine", "content_owner_or_lifecycle_diverged"
decisions.append(
MemoryMigrationDecision(
item_key=f"memory:{item.legacy_memory_id}",
action=action,
legacy_memory_id=item.legacy_memory_id,
canonical_memory_id=item.canonical_memory_id,
reason=reason,
)
)
for sdk_item in sdk:
if sdk_item.memory_id not in claimed_sdk_ids:
decisions.append(
MemoryMigrationDecision(
item_key=f"sdk:{sdk_item.memory_id}",
action="adopt",
legacy_memory_id=None,
canonical_memory_id=sdk_item.memory_id,
reason="sdk_only",
)
)
return tuple(decisions)
async def read_sdk_memory_inventory(
client: SdkMigrationClient,
*,
user_id: str = "local",
page_size: int = 500,
) -> tuple[SdkMemoryInventory, ...]:
records: dict[str, SdkMemoryInventory] = {}
offset = 0
while True:
page = await client.list_memories(
user_id=user_id,
statuses=SDK_LIFECYCLE_STATUSES,
limit=page_size,
offset=offset,
order="asc",
)
for memory in page.memories:
records[memory.memory_id] = SdkMemoryInventory(
memory_id=memory.memory_id,
user_id=memory.user_id,
agent_id=memory.agent_id,
content_hash=content_hash(memory.content),
status=memory.status,
)
consumed = len(page.memories)
offset += consumed
if consumed == 0 or offset >= page.total:
break
return tuple(records[key] for key in sorted(records))
def record_memory_migration_plan(
store: SdkDataPlaneMigrationStore,
*,
migration_id: str,
legacy: tuple[LegacyMemoryInventory, ...],
sdk: tuple[SdkMemoryInventory, ...],
) -> tuple[MemoryMigrationDecision, ...]:
decisions = classify_memory_inventory(legacy, sdk)
legacy_by_id = {item.legacy_memory_id: item for item in legacy}
sdk_by_id = {item.memory_id: item for item in sdk}
for decision in decisions:
source_hash = (
legacy_by_id[decision.legacy_memory_id].content_hash
if decision.legacy_memory_id is not None
else sdk_by_id[decision.canonical_memory_id].content_hash
)
operation_key = hashlib.sha256(
f"{migration_id}:{decision.item_key}:{decision.action}:{source_hash}".encode()
).hexdigest()
store.upsert_item(
migration_id=migration_id,
item_key=decision.item_key,
source_kind=f"memory:{decision.action}",
source_id=(
decision.legacy_memory_id
if decision.legacy_memory_id is not None
else decision.canonical_memory_id
),
source_hash=source_hash,
operation_key=operation_key,
)
return decisions
async def execute_memory_migration_plan(
client: SdkMigrationClient,
store: SdkDataPlaneMigrationStore,
*,
migration_id: str,
legacy: tuple[LegacyMemoryInventory, ...],
decisions: tuple[MemoryMigrationDecision, ...],
) -> None:
legacy_by_id = {item.legacy_memory_id: item for item in legacy}
for decision in decisions:
item = (
legacy_by_id.get(decision.legacy_memory_id)
if decision.legacy_memory_id is not None
else None
)
if decision.action == "quarantine":
store.settle_item(
migration_id=migration_id,
item_key=decision.item_key,
state="quarantine",
error_code=decision.reason,
)
continue
if decision.action == "adopt":
store.settle_item(
migration_id=migration_id,
item_key=decision.item_key,
state="adopted",
target_memory_id=decision.canonical_memory_id,
)
continue
if decision.action in {"delete", "verified_deleted"}:
if decision.action == "delete":
await client.delete(decision.canonical_memory_id)
store.settle_item(
migration_id=migration_id,
item_key=decision.item_key,
state="deleted",
target_memory_id=decision.canonical_memory_id,
)
continue
if item is None:
raise RuntimeError(f"missing legacy inventory for {decision.item_key}")
if decision.action == "import_exact":
assert item.layer is not None
request_id = hashlib.sha256(
f"{migration_id}:{decision.item_key}:import_exact".encode()
).hexdigest()
await client.write_legacy_memory(
content=item.content,
memory_id=item.canonical_memory_id,
user_id=item.user_id,
agent_id=item.agent_id,
session_id=item.session_id,
source="legacy-governance",
layer=item.layer,
status=item.status,
owner=item.owner,
tags=item.tags,
supersedes=item.supersedes,
superseded_by=item.superseded_by,
is_latest=item.status == "active",
memory_at=None,
metadata={},
request_id=request_id,
)
if decision.action == "record_alias":
store.record_alias(
migration_id=migration_id,
legacy_memory_id=item.legacy_memory_id,
canonical_memory_id=item.canonical_memory_id,
)
store.settle_item(
migration_id=migration_id,
item_key=decision.item_key,
state="verified",
target_memory_id=decision.canonical_memory_id,
target_hash=item.content_hash,
)
async def read_sdk_raw_inventory(
client: SdkMigrationClient,
*,
user_id: str = "local",
page_size: int = 500,
) -> tuple[SdkRawInventory, ...]:
records: dict[str, SdkRawInventory] = {}
offset = 0
while True:
page = await client.read_raw_inventory(
user_id=user_id,
limit=page_size,
offset=offset,
order="asc",
)
for raw in page.raw:
raw_id = str(raw.raw_id)
records[raw_id] = SdkRawInventory(
raw_id=raw_id,
user_id=str(raw.user_id),
agent_id=str(raw.agent_id),
session_id=str(raw.session_id),
source=str(raw.source),
content_hash=content_hash(str(raw.content)),
)
consumed = len(page.raw)
offset += consumed
if consumed == 0 or offset >= page.total:
break
return tuple(records[key] for key in sorted(records))
def classify_raw_inventory(
legacy: tuple[LegacyRawInventory, ...],
sdk: tuple[SdkRawInventory, ...],
) -> tuple[RawMigrationDecision, ...]:
sdk_by_identity: dict[tuple[str, str, str, str, str], list[SdkRawInventory]] = {}
for raw in sdk:
identity = (
raw.user_id,
raw.agent_id,
raw.session_id,
raw.source,
raw.content_hash,
)
sdk_by_identity.setdefault(identity, []).append(raw)
claimed: set[str] = set()
decisions: list[RawMigrationDecision] = []
for item in legacy:
identity = (
item.user_id,
item.agent_id,
item.session_id,
item.source,
item.content_hash,
)
candidates = sorted(
sdk_by_identity.get(identity, ()),
key=lambda candidate: candidate.raw_id,
)
existing = next(
(candidate for candidate in candidates if candidate.raw_id not in claimed),
None,
)
if existing is not None:
claimed.add(existing.raw_id)
decisions.append(
RawMigrationDecision(
item_key=f"raw:{item.evidence_id}",
action="matched",
evidence_id=item.evidence_id,
raw_id=existing.raw_id,
reason="same_content_and_scope",
)
)
else:
decisions.append(
RawMigrationDecision(
item_key=f"raw:{item.evidence_id}",
action="import_exact",
evidence_id=item.evidence_id,
raw_id=f"evidence:{item.evidence_id}",
reason="evidence_only",
)
)
for raw in sdk:
if raw.raw_id not in claimed:
decisions.append(
RawMigrationDecision(
item_key=f"sdk-raw:{raw.raw_id}",
action="adopt",
evidence_id=None,
raw_id=raw.raw_id,
reason="sdk_only",
)
)
return tuple(decisions)
def record_raw_migration_plan(
store: SdkDataPlaneMigrationStore,
*,
migration_id: str,
legacy: tuple[LegacyRawInventory, ...],
sdk: tuple[SdkRawInventory, ...],
) -> tuple[RawMigrationDecision, ...]:
decisions = classify_raw_inventory(legacy, sdk)
legacy_by_id = {item.evidence_id: item for item in legacy}
sdk_by_id = {item.raw_id: item for item in sdk}
for decision in decisions:
source_hash = (
legacy_by_id[decision.evidence_id].content_hash
if decision.evidence_id is not None
else sdk_by_id[decision.raw_id].content_hash
)
operation_key = hashlib.sha256(
f"{migration_id}:{decision.item_key}:{decision.action}:{source_hash}".encode()
).hexdigest()
store.upsert_item(
migration_id=migration_id,
item_key=decision.item_key,
source_kind=f"raw:{decision.action}",
source_id=decision.evidence_id or decision.raw_id,
source_hash=source_hash,
operation_key=operation_key,
)
return decisions
async def execute_raw_migration_plan(
client: SdkMigrationClient,
store: SdkDataPlaneMigrationStore,
*,
migration_id: str,
legacy: tuple[LegacyRawInventory, ...],
decisions: tuple[RawMigrationDecision, ...],
) -> None:
legacy_by_id = {item.evidence_id: item for item in legacy}
for decision in decisions:
if decision.action == "adopt":
store.settle_item(
migration_id=migration_id,
item_key=decision.item_key,
state="adopted",
target_memory_id=decision.raw_id,
)
continue
item = legacy_by_id.get(decision.evidence_id or "")
if item is None:
raise RuntimeError(f"missing legacy raw inventory for {decision.item_key}")
if decision.action == "import_exact":
request_id = hashlib.sha256(
f"{migration_id}:{decision.item_key}:write_legacy_raw".encode()
).hexdigest()
await client.write_legacy_raw(
content=item.content,
raw_id=decision.raw_id,
user_id=item.user_id,
agent_id=item.agent_id,
session_id=item.session_id,
source=item.source,
memory_at=item.memory_at,
metadata=item.metadata,
request_id=request_id,
)
store.settle_item(
migration_id=migration_id,
item_key=decision.item_key,
state="verified",
target_memory_id=decision.raw_id,
target_hash=item.content_hash,
)
store.record_raw_alias(
migration_id=migration_id,
legacy_evidence_id=item.evidence_id,
canonical_raw_id=decision.raw_id,
)
@@ -1,137 +1,210 @@
"""Offline startup cutover from the legacy Governance mirror to SDK authority.
Every released build wrote memory and raw turns through the SDK first; the
Evidence/Governance tables only mirrored them. The cutover therefore copies no
memory. It makes the surviving write queue independent of the legacy tables,
commits SDK authority and drops those tables.
"""
"""Offline, resumable backfill runner for SDK memory authority."""
from __future__ import annotations
import asyncio
import hashlib
import json
import os
import shutil
import sqlite3
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from collections.abc import Callable
from contextlib import suppress
from dataclasses import dataclass
from pathlib import Path
from one_memory.database_access import main_database
from one_memory.imports.models import EvidenceRecord
from one_memory.imports.storage import EvidenceStore
from one_memory.memory_server.contracts import (
BatchIdentity,
WriteRequest,
internal_session_id_for,
)
from one_memory.memory_server.durable_queue import (
CANCELLED,
COMPLETED,
FAILED,
REAPED_PAYLOAD_MARKER,
DurableWriteQueue,
)
from one_memory.memory_server.unified_entry import UnifiedWriteIntake
from one_memory.memory_server.durable_queue import DurableWriteQueue
from one_memory.sdk_data_plane_migration import (
CutoverMarker,
DataPlaneMigrationError,
MemoryAuthority,
MigrationProgress,
SdkDataPlaneMigrationStore,
)
from one_memory.sdk_migration_bundle import inspect_sdk_migration_bundle
from one_memory.target_schema import drop_legacy_tables, legacy_tables_present
PREPARED_PHASE = "prepared"
from one_memory.sdk_migration_inventory import (
SdkMigrationClient,
execute_memory_migration_plan,
execute_raw_migration_plan,
read_legacy_memory_inventory,
read_legacy_raw_inventory,
read_sdk_memory_inventory,
read_sdk_raw_inventory,
record_memory_migration_plan,
record_raw_migration_plan,
)
from one_memory.target_schema import drop_legacy_tables
@dataclass(frozen=True)
class CutoverPreparation:
"""What the cutover did to legacy work, for the receipt and the dialog."""
frozen_jobs: int
moved_tasks: int
moved_dispatches: int
abandoned: dict[str, int]
def as_document(self) -> dict[str, object]:
return {
"frozen_jobs": self.frozen_jobs,
"moved_tasks": self.moved_tasks,
"moved_dispatches": self.moved_dispatches,
"abandoned": dict(self.abandoned),
}
class BackfillResult:
migration_id: str
state: str
phase: str
counts: dict[str, int]
memory_items: int
raw_items: int
def _tables(database: sqlite3.Connection) -> set[str]:
return {
str(row[0])
for row in database.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
@dataclass(frozen=True)
class SdkMigrationStaging:
migration_id: str
staging_home: Path
live_root: Path
staging_root: Path
protective_root: Path
state_file: Path
def recover_abandoned_staging(*, data_dir: Path, database_path: Path, namespace: str) -> None:
"""Undo or discard the SDK staging copy that builds through 19 created.
Those builds staged a full SDK copy and swapped directories at cutover. The
live root was untouched until the swap, so before SDK authority the
protective root is the original store; after it, the protective root is
stale.
"""
def sdk_migration_staging(
*,
data_dir: Path,
migration_id: str,
namespace: str,
) -> SdkMigrationStaging:
"""Return same-filesystem paths without trusting the migration ID as a path."""
token = hashlib.sha256(migration_id.encode()).hexdigest()[:16]
home = data_dir.resolve()
staging_home = home / ".sdk-migration" / token
live_root = home / "hy-memory" / namespace
authority = SdkDataPlaneMigrationStore(database_path).authority()
protective_roots = sorted(live_root.parent.glob(f".{namespace}.pre-migration-*"))
if authority.value != "sdk" and len(protective_roots) > 1:
raise DataPlaneMigrationError("sdk_migration_swap_state_ambiguous")
for protective_root in protective_roots:
if authority.value == "sdk":
if not live_root.is_dir():
raise DataPlaneMigrationError("sdk_migration_committed_store_missing")
shutil.rmtree(protective_root)
continue
if live_root.exists():
shutil.rmtree(live_root)
protective_root.replace(live_root)
shutil.rmtree(home / ".sdk-migration", ignore_errors=True)
def abandoned_staging_present(*, data_dir: Path, namespace: str) -> bool:
home = data_dir.resolve()
return (home / ".sdk-migration").exists() or any(
(home / "hy-memory").glob(f".{namespace}.pre-migration-*")
return SdkMigrationStaging(
migration_id=migration_id,
staging_home=staging_home,
live_root=live_root,
staging_root=staging_home / "hy-memory" / namespace,
protective_root=live_root.parent / f".{namespace}.pre-migration-{token}",
state_file=staging_home / "swap-state.json",
)
def _write_staging_state(paths: SdkMigrationStaging, phase: str) -> None:
paths.staging_home.mkdir(parents=True, exist_ok=True)
temporary = paths.state_file.with_suffix(".tmp")
temporary.write_text(
json.dumps(
{"schema_version": 1, "migration_id": paths.migration_id, "phase": phase},
sort_keys=True,
),
encoding="utf-8",
)
temporary.replace(paths.state_file)
def recover_sdk_migration_swap(
*,
paths: SdkMigrationStaging,
database_path: Path,
) -> None:
"""Recover an interrupted directory swap before opening any SDK store."""
if not paths.protective_root.exists():
return
authority = SdkDataPlaneMigrationStore(database_path).authority()
if authority.value == "sdk":
if not paths.live_root.is_dir():
raise DataPlaneMigrationError("sdk_migration_committed_store_missing")
shutil.rmtree(paths.protective_root)
shutil.rmtree(paths.staging_home, ignore_errors=True)
return
if paths.live_root.exists():
if paths.staging_root.exists():
raise DataPlaneMigrationError("sdk_migration_swap_state_ambiguous")
paths.staging_root.parent.mkdir(parents=True, exist_ok=True)
paths.live_root.replace(paths.staging_root)
paths.protective_root.replace(paths.live_root)
_write_staging_state(paths, "recovered")
def prepare_sdk_migration_staging(
*,
paths: SdkMigrationStaging,
database_path: Path,
) -> Path:
"""Build a private copy; incomplete retries restart from the unchanged live root."""
recover_sdk_migration_swap(paths=paths, database_path=database_path)
migration_store = SdkDataPlaneMigrationStore(database_path)
try:
progress = migration_store.progress(paths.migration_id)
reuse_ready = progress.state == "ready" and progress.phase == "reconciled"
except Exception:
reuse_ready = False
if paths.staging_root.exists():
if paths.staging_root.is_symlink() or not paths.staging_root.is_dir():
raise DataPlaneMigrationError("sdk_migration_staging_invalid")
if reuse_ready:
return paths.staging_home
shutil.rmtree(paths.staging_root)
paths.live_root.mkdir(parents=True, exist_ok=True)
from one_memory.hy_sdk_migration import snapshot_data_plane
snapshot_data_plane(
source_root=paths.live_root,
target_root=paths.staging_root,
writers_stopped=True,
)
_write_staging_state(paths, "prepared")
return paths.staging_home
def activate_sdk_migration_staging(
*,
paths: SdkMigrationStaging,
database_path: Path,
commit: Callable[[], CutoverMarker],
) -> CutoverMarker:
"""Atomically expose staged SDK files and roll them back if cutover fails."""
recover_sdk_migration_swap(paths=paths, database_path=database_path)
if not paths.staging_root.is_dir():
raise DataPlaneMigrationError("sdk_migration_staging_missing")
paths.live_root.mkdir(parents=True, exist_ok=True)
_write_staging_state(paths, "activating")
paths.live_root.replace(paths.protective_root)
try:
paths.staging_root.replace(paths.live_root)
_write_staging_state(paths, "staging_live")
marker = commit()
except BaseException:
if paths.live_root.exists():
paths.staging_root.parent.mkdir(parents=True, exist_ok=True)
paths.live_root.replace(paths.staging_root)
if paths.protective_root.exists():
paths.protective_root.replace(paths.live_root)
_write_staging_state(paths, "rolled_back")
raise
with suppress(OSError):
_write_staging_state(paths, "committed")
# Authority is already committed; leftover paths are detected by
# sdk-migration-status and removed by sdk-cleanup on the next startup.
shutil.rmtree(paths.protective_root, ignore_errors=True)
shutil.rmtree(paths.staging_home, ignore_errors=True)
return marker
def prepare_legacy_write_queue_for_cutover(database_path: Path) -> tuple[int, int]:
"""Freeze old jobs and move unstarted task rows into the surviving queue."""
with main_database(database_path).read(row_factory=sqlite3.Row) as database:
tables = _tables(database)
tables = {
str(row["name"])
for row in database.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
if "evidence" not in tables:
return (0, 0)
jobs = (
tuple(
database.execute("SELECT job_id,state,payload_json FROM memory_write_jobs")
)
tuple(database.execute("SELECT job_id,payload_json FROM memory_write_jobs"))
if "memory_write_jobs" in tables
else ()
)
evidence = EvidenceStore(database_path)
frozen = 0
for row in jobs:
decoded = json.loads(str(row["payload_json"]))
if isinstance(decoded, dict) and REAPED_PAYLOAD_MARKER in decoded:
# A finished job whose input the queue already released.
continue
request = WriteRequest.model_validate(decoded)
request = WriteRequest.model_validate_json(str(row["payload_json"]))
if request.input_records:
continue
records = tuple(evidence.get(identifier) for identifier in request.evidence_ids)
if any(record is None for record in records):
if str(row["state"]) in {COMPLETED, FAILED, CANCELLED}:
continue
raise DataPlaneMigrationError(f"sdk_data_plane_write_job_input_missing:{row['job_id']}")
materialized = tuple(record for record in records if record is not None)
updated = request.model_copy(update={"input_records": materialized})
@@ -204,55 +277,30 @@ def prepare_legacy_write_queue_for_cutover(database_path: Path) -> tuple[int, in
return (frozen, moved)
def move_pending_dispatches(database_path: Path) -> int:
"""Queue Evidence that was waiting for its source file to finish.
Its source cursor has already moved past these rows, so History catch-up
after the cutover would not read them again.
"""
def sdk_cutover_blockers(database_path: Path) -> dict[str, int]:
"""Count legacy work that could write or duplicate data after cutover."""
blockers: dict[str, int] = {}
with main_database(database_path).read(row_factory=sqlite3.Row) as database:
if not {"evidence", "evidence_pending_memory_dispatch"} <= _tables(database):
return 0
identifiers = tuple(
str(row["evidence_id"])
for row in database.execute(
"SELECT DISTINCT pending.evidence_id "
"FROM evidence_pending_memory_dispatch pending "
"JOIN evidence USING(evidence_id) "
"ORDER BY COALESCE(evidence.created_at,evidence.imported_at),"
"COALESCE(evidence.ingest_seq,evidence.rowid),pending.evidence_id"
)
)
evidence = EvidenceStore(database_path)
conversations: dict[tuple[str, str, str, str | None], list[EvidenceRecord]] = {}
for identifier in identifiers:
record = evidence.get(identifier)
if record is None:
continue
key = (record.user_id, record.source_id, record.conversation_id, record.workspace)
conversations.setdefault(key, []).append(record)
intake = UnifiedWriteIntake(DurableWriteQueue(database_path))
for records in conversations.values():
intake.submit_records(tuple(records))
return len(conversations)
def abandoned_legacy_work(database_path: Path) -> dict[str, int]:
"""Count legacy work that ends with the cutover instead of blocking it.
A task that may have reached the SDK is not replayed: the SDK already holds
whatever it committed. Pending user deletions are not carried over.
"""
abandoned: dict[str, int] = {}
with main_database(database_path).read(row_factory=sqlite3.Row) as database:
tables = _tables(database)
tables = {
str(row["name"])
for row in database.execute("SELECT name FROM sqlite_master WHERE type='table'")
}
checks = (
(
"invoked_tasks",
"memory_tasks",
"memory_tasks",
"status='unknown' OR (status='running' AND hy_invocation_count > 0)",
),
("deletions", "memory_deletions", "status IN ('pending','running')"),
(
"deletions",
"memory_deletions",
"status IN ('pending','running')",
),
(
"pending_dispatch",
"evidence_pending_memory_dispatch",
"1=1",
),
)
for name, table, where in checks:
if table not in tables:
@@ -261,23 +309,11 @@ def abandoned_legacy_work(database_path: Path) -> dict[str, int]:
database.execute(f"SELECT COUNT(*) FROM {table} WHERE {where}").fetchone()[0]
)
if count:
abandoned[name] = count
return abandoned
blockers[name] = count
return blockers
def prepare_legacy_work_for_cutover(database_path: Path) -> CutoverPreparation:
"""Make the surviving queue self-contained. Safe to repeat."""
frozen, moved = prepare_legacy_write_queue_for_cutover(database_path)
dispatches = move_pending_dispatches(database_path)
return CutoverPreparation(
frozen_jobs=frozen,
moved_tasks=moved,
moved_dispatches=dispatches,
abandoned=abandoned_legacy_work(database_path),
)
def commit_sdk_cutover(
def commit_reconciled_cutover(
*,
data_dir: Path,
database_path: Path,
@@ -286,25 +322,38 @@ def commit_sdk_cutover(
sdk_version: str,
store_version: str = "1",
) -> CutoverMarker:
"""Verify the backup, settle legacy work and commit SDK authority."""
"""Verify backup/work quiescence and atomically commit SDK authority."""
inspect_sdk_migration_bundle(data_dir=data_dir, bundle_id=bundle_id)
# Repeated here because a CLI-managed Runtime may have run on legacy
# authority since `sdk-migrate`.
preparation = prepare_legacy_work_for_cutover(database_path)
prepare_legacy_write_queue_for_cutover(database_path)
blockers = sdk_cutover_blockers(database_path)
if blockers:
raise DataPlaneMigrationError(
"sdk_data_plane_cutover_work_pending:"
+ json.dumps(blockers, sort_keys=True, separators=(",", ":"))
)
with main_database(database_path).read(row_factory=sqlite3.Row) as database:
rows = database.execute(
"SELECT item_key,state,target_memory_id,target_hash "
"FROM sdk_data_plane_migration_items WHERE migration_id=? "
"ORDER BY item_key",
(migration_id,),
).fetchall()
digest = hashlib.sha256(
json.dumps(
preparation.as_document(),
[dict(row) for row in rows],
sort_keys=True,
separators=(",", ":"),
).encode()
).hexdigest()
return SdkDataPlaneMigrationStore(database_path).commit_cutover(
marker = SdkDataPlaneMigrationStore(database_path).commit_cutover(
migration_id=migration_id,
sdk_version=sdk_version,
store_version=store_version,
backup_bundle_id=bundle_id,
reconciliation_digest=digest,
)
inspect_sdk_migration_bundle(data_dir=data_dir, bundle_id=marker.backup_bundle_id)
return marker
def resume_legacy_cleanup(*, data_dir: Path, database_path: Path) -> tuple[str, ...]:
@@ -318,175 +367,87 @@ def resume_legacy_cleanup(*, data_dir: Path, database_path: Path) -> tuple[str,
data_dir=data_dir,
bundle_id=marker.backup_bundle_id,
)
from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
from one_memory.hy_runtime_config import HY_MEMORY_NAMESPACE
recover_abandoned_staging(
data_dir=data_dir,
database_path=database_path,
namespace=HY_MEMORY_NAMESPACE,
)
recover_sdk_migration_swap(
paths=sdk_migration_staging(
data_dir=data_dir,
migration_id=marker.migration_id,
namespace=HY_MEMORY_NAMESPACE,
),
database_path=database_path,
)
return drop_legacy_tables(database_path)
MIGRATION_PREFIX = "sdk-authority-v2"
CAPTURE_FENCE = ".sdk-migration-capture-fence.json"
@dataclass(frozen=True)
class StartupMigrationResult:
"""`not_required`, `cutover` (this run switched authority) or `complete`."""
state: str
migration_id: str | None = None
bundle_id: str | None = None
preparation: CutoverPreparation | None = None
dropped_tables: tuple[str, ...] = ()
def as_document(self) -> dict[str, object]:
document: dict[str, object] = {"schema_version": 1, "state": self.state}
if self.migration_id is not None:
document["migration_id"] = self.migration_id
document["bundle_id"] = self.bundle_id
if self.preparation is not None:
document.update(self.preparation.as_document())
if self.dropped_tables:
document["dropped_tables"] = list(self.dropped_tables)
return document
def database_generation(database_path: Path) -> str:
stat = database_path.stat()
return hashlib.sha256(f"{stat.st_size}:{stat.st_mtime_ns}".encode()).hexdigest()[:16]
def _resumable_identity(database_path: Path) -> tuple[str, str]:
"""Resume the newest unfinished run, including one build 19 left blocked."""
with main_database(database_path).read(row_factory=sqlite3.Row) as database:
row = (
database.execute(
"SELECT migration_id,backup_bundle_id FROM sdk_data_plane_migrations "
"WHERE state IN ('running','blocked','ready') AND migration_id LIKE ? "
"ORDER BY created_at DESC LIMIT 1",
(f"{MIGRATION_PREFIX}-%",),
).fetchone()
if "sdk_data_plane_migrations" in _tables(database)
else None
)
if row is not None and row["backup_bundle_id"] == f"{row['migration_id']}-pre-cutover":
return str(row["migration_id"]), str(row["backup_bundle_id"])
migration_id = f"{MIGRATION_PREFIX}-{database_generation(database_path)}"
return migration_id, f"{migration_id}-pre-cutover"
@contextmanager
def capture_fence(data_dir: Path) -> Iterator[None]:
"""Host capture hooks return silently while this process is alive."""
marker = data_dir / CAPTURE_FENCE
temporary = marker.with_name(f"{marker.name}.{os.getpid()}.tmp")
temporary.write_text(
json.dumps({"schema_version": 1, "pid": os.getpid()}) + "\n",
encoding="utf-8",
)
os.chmod(temporary, 0o600)
temporary.replace(marker)
try:
yield
finally:
marker.unlink(missing_ok=True)
temporary.unlink(missing_ok=True)
def _ensure_bundle(*, data_dir: Path, bundle_id: str, runtime_build: str) -> dict[str, object]:
from one_memory.sdk_migration_bundle import (
MigrationBundleRejected,
create_sdk_migration_bundle,
)
try:
create_sdk_migration_bundle(
data_dir=data_dir,
bundle_id=bundle_id,
runtime_build=runtime_build,
runtime_stopped=True,
)
except MigrationBundleRejected as error:
if error.code != "sdk_migration_bundle_exists":
raise
return inspect_sdk_migration_bundle(data_dir=data_dir, bundle_id=bundle_id)
def run_startup_migration(
async def run_sdk_data_plane_backfill(
*,
data_dir: Path,
runtime_build: str,
sdk_version: str,
namespace: str,
progress: Callable[[str], None] = lambda _step: None,
) -> StartupMigrationResult:
"""Bring one stopped home to SDK authority. Safe to repeat after any failure.
The caller must stop Runtime and keep its service from restarting.
"""
database_path = data_dir / "one-memory.sqlite3"
if not database_path.is_file():
return StartupMigrationResult(state="not_required")
database_path: Path,
client: SdkMigrationClient,
migration_id: str,
bundle_id: str,
source_runtime_build: str,
target_sdk_version: str,
source_schema_version: int,
namespace_fingerprint: str,
backend_fingerprint: str,
) -> BackfillResult:
"""Backfill both lanes but never commit the authority cutover marker."""
store = SdkDataPlaneMigrationStore(database_path)
if store.authority() is MemoryAuthority.SDK:
if not legacy_tables_present(database_path) and not abandoned_staging_present(
data_dir=data_dir, namespace=namespace
):
return StartupMigrationResult(state="not_required")
progress("cleanup")
return StartupMigrationResult(
state="complete",
dropped_tables=resume_legacy_cleanup(data_dir=data_dir, database_path=database_path),
)
from one_memory.database_access import close_main_databases
from one_memory.memory_server.migration_runner import SqliteMigrationStore, migrate
from one_memory.memory_server.migrations import MIGRATIONS
migration_id, bundle_id = _resumable_identity(database_path)
with capture_fence(data_dir):
progress("bundle")
manifest = _ensure_bundle(
data_dir=data_dir,
bundle_id=bundle_id,
runtime_build=runtime_build,
)
progress("prepare")
asyncio.run(migrate(SqliteMigrationStore(database_path), MIGRATIONS))
close_main_databases()
recover_abandoned_staging(
data_dir=data_dir,
database_path=database_path,
namespace=namespace,
)
store.start(
migration_id=migration_id,
source_runtime_build=str(manifest.get("runtime_build") or ""),
target_sdk_version=sdk_version,
source_schema_version=MIGRATIONS[-1].version,
namespace_fingerprint=hashlib.sha256(namespace.encode()).hexdigest(),
backend_fingerprint=hashlib.sha256(b"chroma:kuzu:sqlite").hexdigest(),
backup_bundle_id=bundle_id,
)
preparation = prepare_legacy_work_for_cutover(database_path)
store.set_phase(migration_id, phase=PREPARED_PHASE, state="ready")
progress("cutover")
commit_sdk_cutover(
data_dir=data_dir,
database_path=database_path,
migration_id=migration_id,
bundle_id=bundle_id,
sdk_version=sdk_version,
)
progress("cleanup")
dropped = resume_legacy_cleanup(data_dir=data_dir, database_path=database_path)
return StartupMigrationResult(
state="cutover",
store.start(
migration_id=migration_id,
bundle_id=bundle_id,
preparation=preparation,
dropped_tables=dropped,
source_runtime_build=source_runtime_build,
target_sdk_version=target_sdk_version,
source_schema_version=source_schema_version,
namespace_fingerprint=namespace_fingerprint,
backend_fingerprint=backend_fingerprint,
backup_bundle_id=bundle_id,
)
store.set_phase(migration_id, phase="inventory", state="running")
legacy_memories = read_legacy_memory_inventory(database_path)
sdk_memories = await read_sdk_memory_inventory(client)
memory_decisions = record_memory_migration_plan(
store,
migration_id=migration_id,
legacy=legacy_memories,
sdk=sdk_memories,
)
legacy_raw = read_legacy_raw_inventory(database_path)
sdk_raw = await read_sdk_raw_inventory(client)
raw_decisions = record_raw_migration_plan(
store,
migration_id=migration_id,
legacy=legacy_raw,
sdk=sdk_raw,
)
store.set_phase(migration_id, phase="backfill", state="running")
await execute_memory_migration_plan(
client,
store,
migration_id=migration_id,
legacy=legacy_memories,
decisions=memory_decisions,
)
await execute_raw_migration_plan(
client,
store,
migration_id=migration_id,
legacy=legacy_raw,
decisions=raw_decisions,
)
progress: MigrationProgress = store.progress(migration_id)
if progress.counts.get("quarantine", 0):
store.set_phase(migration_id, phase="review_required", state="blocked")
else:
store.set_phase(migration_id, phase="reconciled", state="ready")
progress = store.progress(migration_id)
return BackfillResult(
migration_id=migration_id,
state=progress.state,
phase=progress.phase,
counts=progress.counts,
memory_items=len(memory_decisions),
raw_items=len(raw_decisions),
)
@@ -54,8 +54,6 @@ LEGACY_TABLES_IN_DROP_ORDER = (
"evidence_origins",
"evidence_versions",
"evidence_repairs",
# FTS5 index over Evidence text; dropping it removes its shadow tables.
"evidence_search",
"evidence",
"memory_id_aliases",
"raw_id_aliases",
+1 -2
View File
@@ -54,7 +54,6 @@
| `test_claude_adoption_probe.py` | `contract` | The external-use verifier must distinguish a read from merely mentioning it. |
| `test_cli_commands.py` | `contract` | 覆盖 status reports the verified runtime as json、doctor fails when runtime identity cannot be verified、serve and legacy empty command share the same entry 等 4 个场景。 |
| `test_cli_lifecycle.py` | `integration` | 覆盖 cli starts and stops without residual runtime 场景。 |
| `test_cli_migration_error_codes.py` | `unit` | The startup dialog can only act on the code it is handed. |
| `test_cli_restore.py` | `contract` | 覆盖 restore requires a source、restore refuses while the runtime is running、restore previews without confirmation 等 4 个场景。 |
| `test_codex_header_provenance.py` | `contract` | 覆盖 header workspace versions and native ids reach file origin、new adapter rereads header after a real checkpoint、machine and unknown session sources do not establish human or agent sender 等 14 个场景。 |
| `test_connection_lifetime.py` | `integration` | Every SQLite connection a request opens must be closed when the request ends. |
@@ -328,4 +327,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. |
共 323 个测试文件。
共 322 个测试文件。
@@ -14,14 +14,13 @@ from pathlib import Path
import pytest
PRIMARY_MARKERS = ("unit", "contract", "integration", "performance", "live")
EXPECTED_TEST_FILE_COUNT = 323
EXPECTED_TEST_FILES_SHA256 = "10a2327bd047a281511980f5b558c1ea64d857f0fff550e4bd96cb53800205ba"
EXPECTED_TEST_FILE_COUNT = 322
EXPECTED_TEST_FILES_SHA256 = "d242783adb4015ebdbb5c099f3b5aa88a99c828ca6fddc8ca8d6c74c56e48945"
# Pure transformations, validation models, parity checks and state machines.
# This list is intentionally small: uncertain files default to contract.
UNIT_FILES = frozenset(
{
"test_cli_migration_error_codes.py",
"test_context_v2_contract_parity.py",
"test_daemon_contract.py",
"test_diagnostic_payload_keys.py",
@@ -1,101 +0,0 @@
"""The startup dialog can only act on the code it is handed.
A full disk and a damaged backup are different next steps, so they must reach
the person under different names. These tests pin those names.
"""
from __future__ import annotations
import errno
from pathlib import Path
import pytest
from one_memory.cli import migration_failure_code
class _Coded(RuntimeError):
def __init__(self, code: str) -> None:
super().__init__(code)
self.code = code
def test_running_out_of_disk_is_named_however_deep_it_is_wrapped() -> None:
error = OSError(errno.ENOSPC, "No space left on device")
wrapped: BaseException = error
for _ in range(3):
try:
raise RuntimeError("wrapper") from wrapped
except RuntimeError as caught:
wrapped = caught
assert migration_failure_code(wrapped) == "sdk_migration_disk_space_insufficient"
# The explicit context chain counts too: `raise ... from None` discards the
# cause but Python still keeps the original as __context__.
try:
try:
raise OSError(errno.ENOSPC, "No space left on device")
except OSError:
raise RuntimeError("sdk_snapshot_failed") from None
except RuntimeError as caught:
assert migration_failure_code(caught) == "sdk_migration_disk_space_insufficient"
def test_quota_exhaustion_counts_as_no_space(tmp_path: Path) -> None:
quota = getattr(errno, "EDQUOT", errno.ENOSPC)
assert (
migration_failure_code(OSError(quota, "Disk quota exceeded"))
== "sdk_migration_disk_space_insufficient"
)
def test_migration_state_codes_pass_through_because_they_are_already_actionable() -> None:
assert (
migration_failure_code(_Coded("sdk_data_plane_migration_bundle_conflict"))
== "sdk_data_plane_migration_bundle_conflict"
)
assert (
migration_failure_code(_Coded("sdk_migration_swap_state_ambiguous"))
== "sdk_migration_swap_state_ambiguous"
)
def test_unknown_failures_keep_their_message_out_of_the_code() -> None:
error = RuntimeError("/Users/someone/private/path blew up")
assert migration_failure_code(error) == "sdk_migration_preflight_failed"
assert (
migration_failure_code(error, fallback="sdk_migration_bundle_copy_failed")
== "sdk_migration_bundle_copy_failed"
)
def test_a_non_code_attribute_never_becomes_a_code() -> None:
class Odd(RuntimeError):
code = 17
assert migration_failure_code(Odd("x")) == "sdk_migration_preflight_failed"
def test_the_chain_survives_a_cycle() -> None:
first = RuntimeError("first")
second = RuntimeError("second")
first.__cause__ = second
second.__cause__ = first
assert migration_failure_code(first) == "sdk_migration_preflight_failed"
@pytest.mark.parametrize(
"code",
[
"sdk_migration_disk_space_insufficient",
"sdk_migration_preflight_failed",
"sdk_migration_bundle_copy_failed",
"sdk_migration_cutover_failed",
],
)
def test_the_codes_the_desktop_translates_stay_in_sync(code: str) -> None:
"""Desktop maps these exact names to sentences; renaming one blinds it."""
desktop = Path(__file__).resolve().parents[2] / "desktop" / "src" / "stable-runtime.ts"
assert desktop.is_file(), desktop
assert f'"{code}"' in desktop.read_text(encoding="utf-8")
+302 -80
View File
@@ -1,14 +1,10 @@
import asyncio
import io
import json
import sqlite3
from types import SimpleNamespace
from one_memory import cli
from one_memory.config import Settings
from one_memory.memory_server.migration_runner import SqliteMigrationStore, migrate
from one_memory.memory_server.migrations import MIGRATIONS
from one_memory.runtime_client import RuntimeStatus
from one_memory.sdk_data_plane_migration import SdkDataPlaneMigrationStore
def test_restore_requires_a_source(tmp_path, monkeypatch) -> None:
@@ -114,7 +110,56 @@ def test_restore_sdk_bundle_uses_full_data_plane_restore(tmp_path, monkeypatch)
assert '"restored_files":4' in document
def _stopped(monkeypatch, tmp_path) -> None:
def test_migration_bundle_refuses_while_runtime_is_running(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
"inspect_runtime",
lambda _data_dir: RuntimeStatus("running", "runtime_verified"),
)
output = io.StringIO()
code = cli.run(
[
"migration-bundle",
"--bundle-id",
"migration-1",
"--runtime-build",
"build-13",
"--yes",
"--json",
],
output=output,
)
assert code == 4
assert "sdk_migration_bundle_runtime_running" in output.getvalue()
def test_migration_bundle_previews_before_writing(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
"inspect_runtime",
lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"),
)
output = io.StringIO()
code = cli.run(
[
"migration-bundle",
"--bundle-id",
"migration-1",
"--runtime-build",
"build-13",
"--json",
],
output=output,
)
assert code == 4
assert '"state": "preview"' in output.getvalue() or '"state":"preview"' in output.getvalue()
def test_migration_bundle_success_reports_bundle_identity(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
@@ -122,22 +167,236 @@ def _stopped(monkeypatch, tmp_path) -> None:
lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"),
)
class Result:
bundle_id = "migration-1"
bundle_path = tmp_path / "backups" / "migration-1.sdk-bundle"
file_count = 4
total_bytes = 123
def _legacy_home(tmp_path) -> None:
database = tmp_path / "one-memory.sqlite3"
asyncio.run(migrate(SqliteMigrationStore(database), MIGRATIONS))
with sqlite3.connect(database) as connection:
connection.execute(
"INSERT INTO evidence ("
"evidence_id,idempotency_key,user_id,source_id,conversation_id,"
"content_kind,role,written_by,verbatim,content_hash,revision,"
"imported_at,updated_at"
") VALUES ('e1','k','local','cursor','s','raw_conversation',"
"'user','person','hello','hash',1,'now','now')"
monkeypatch.setattr(
"one_memory.sdk_migration_bundle.create_sdk_migration_bundle",
lambda **_kwargs: Result(),
)
output = io.StringIO()
code = cli.run(
[
"migration-bundle",
"--bundle-id",
"migration-1",
"--runtime-build",
"build-13",
"--yes",
"--json",
],
output=output,
)
assert code == 0
document = output.getvalue().replace(" ", "")
assert '"state":"created"' in document
assert '"bundle_id":"migration-1"' in document
def test_migration_bundle_reuses_an_existing_verified_bundle(tmp_path, monkeypatch) -> None:
from one_memory.sdk_migration_bundle import MigrationBundleRejected
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
"inspect_runtime",
lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"),
)
def exists(**_kwargs):
raise MigrationBundleRejected("sdk_migration_bundle_exists")
monkeypatch.setattr(
"one_memory.sdk_migration_bundle.create_sdk_migration_bundle",
exists,
)
monkeypatch.setattr(
"one_memory.sdk_migration_bundle.inspect_sdk_migration_bundle",
lambda **_kwargs: {"file_count": 4, "total_bytes": 123},
)
output = io.StringIO()
code = cli.run(
[
"migration-bundle",
"--bundle-id",
"migration-1",
"--runtime-build",
"build-13",
"--yes",
"--json",
],
output=output,
)
assert code == 0
assert '"state": "reused"' in output.getvalue()
def test_sdk_migrate_preview_never_commits_cutover(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
"inspect_runtime",
lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"),
)
output = io.StringIO()
code = cli.run(
[
"sdk-migrate",
"--migration-id",
"migration-1",
"--bundle-id",
"bundle-1",
"--json",
],
output=output,
)
assert code == 4
document = output.getvalue().replace(" ", "")
assert '"state":"preview"' in document
assert '"commits_cutover":false' in document
def test_sdk_migrate_runs_offline_backfill_without_cutover(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
"inspect_runtime",
lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"),
)
monkeypatch.setattr(
"one_memory.sdk_migration_bundle.inspect_sdk_migration_bundle",
lambda **_kwargs: {"runtime_build": "v1.0.4"},
)
async def fake_migrate(*_args, **_kwargs):
return None
monkeypatch.setattr("one_memory.memory_server.migration_runner.migrate", fake_migrate)
profile = SimpleNamespace(
has_api_key=False,
embedding_provider_kind="openai",
embedding_model="embedding",
embedding_dimensions=1024,
)
class Profiles:
def __init__(self, _database):
pass
def active_identity(self):
return ("profile", 1)
def snapshot(self, *_identity):
return profile
def resolve_snapshot_secret(self, *_args):
return None
class Secrets:
pass
class Lease:
client = SimpleNamespace(
_loop_thread=None,
_vector_store=None,
_embed_service=None,
_graph_store=None,
)
def close(self):
return None
def test_sdk_migrate_never_runs_while_runtime_is_live(tmp_path, monkeypatch) -> None:
async def backfill(**_kwargs):
return SimpleNamespace(
state="ready",
phase="reconciled",
migration_id="migration-1",
counts={"verified": 2},
memory_items=1,
raw_items=1,
)
monkeypatch.setattr("one_memory.model_profiles.ModelProfileStore", Profiles)
monkeypatch.setattr("one_memory.secrets.KeyringSecretStore", Secrets)
monkeypatch.setattr(
"one_memory.hy_sdk_runtime.acquire_client",
lambda *_args, **_kwargs: Lease(),
)
monkeypatch.setattr(
"one_memory.sdk_migration_runner.run_sdk_data_plane_backfill",
backfill,
)
output = io.StringIO()
code = cli.run(
[
"sdk-migrate",
"--migration-id",
"migration-1",
"--bundle-id",
"bundle-1",
"--yes",
"--json",
],
output=output,
)
assert code == 0
document = output.getvalue().replace(" ", "")
assert '"state":"ready"' in document
assert '"commits_cutover":false' in document
def test_sdk_cutover_is_offline_confirmed_and_returns_marker(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
"inspect_runtime",
lambda _data_dir: RuntimeStatus("stopped", "runtime_receipt_missing"),
)
monkeypatch.setattr(
"one_memory.sdk_migration_runner.commit_reconciled_cutover",
lambda **_kwargs: SimpleNamespace(
migration_id="migration-1",
backup_bundle_id="bundle-1",
sdk_version="2.0.5.dev3",
completed_at="2026-10-21T00:00:00Z",
),
)
monkeypatch.setattr(
"one_memory.sdk_migration_runner.activate_sdk_migration_staging",
lambda *, commit, **_kwargs: commit(),
)
monkeypatch.setattr(
"one_memory.sdk_migration_runner.resume_legacy_cleanup",
lambda **_kwargs: (),
)
output = io.StringIO()
code = cli.run(
[
"sdk-cutover",
"--migration-id",
"migration-1",
"--bundle-id",
"bundle-1",
"--yes",
"--json",
],
output=output,
)
assert code == 0
document = output.getvalue().replace(" ", "")
assert '"state":"cutover"' in document
assert '"sdk_version":"2.0.5.dev3"' in document
def test_sdk_cutover_never_runs_while_runtime_is_live(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
monkeypatch.setattr(
cli,
@@ -145,72 +404,35 @@ def test_sdk_migrate_never_runs_while_runtime_is_live(tmp_path, monkeypatch) ->
lambda _data_dir: RuntimeStatus("running", "healthy"),
)
output = io.StringIO()
code = cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output)
assert code == 4
assert json.loads(output.getvalue())["code"] == "sdk_migration_runtime_running"
def test_sdk_migrate_previews_without_confirmation(tmp_path, monkeypatch) -> None:
_stopped(monkeypatch, tmp_path)
_legacy_home(tmp_path)
output = io.StringIO()
assert cli.run(["sdk-migrate", "--runtime-build", "20", "--json"], output=output) == 4
assert json.loads(output.getvalue())["state"] == "preview"
assert SdkDataPlaneMigrationStore(tmp_path / "one-memory.sqlite3").authority().value == (
"legacy"
code = cli.run(
[
"sdk-cutover",
"--migration-id",
"migration-1",
"--bundle-id",
"bundle-1",
"--yes",
"--json",
],
output=output,
)
def test_sdk_migrate_requires_the_runtime_build(tmp_path, monkeypatch) -> None:
_stopped(monkeypatch, tmp_path)
assert cli.run(["sdk-migrate", "--yes"], output=io.StringIO()) == 2
assert code == 4
assert '"code": "sdk_cutover_runtime_running"' in output.getvalue()
def test_sdk_migrate_reports_each_step_then_the_result(tmp_path, monkeypatch) -> None:
_stopped(monkeypatch, tmp_path)
_legacy_home(tmp_path)
output = io.StringIO()
code = cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output)
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 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()
again = io.StringIO()
assert cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=again) == 0
assert json.loads(again.getvalue())["state"] == "not_required"
def test_sdk_migrate_failure_is_one_named_code(tmp_path, monkeypatch) -> None:
import errno
_stopped(monkeypatch, tmp_path)
def full_disk(**_kwargs):
raise OSError(errno.ENOSPC, "No space left on device")
monkeypatch.setattr("one_memory.sdk_migration_runner.run_startup_migration", full_disk)
output = io.StringIO()
assert cli.run(["sdk-migrate", "--runtime-build", "20", "--yes", "--json"], output=output) == 1
assert json.loads(output.getvalue()) == {
"schema_version": 1,
"state": "failed",
"code": "sdk_migration_disk_space_insufficient",
}
def test_sdk_migration_status_says_whether_migrate_has_work(tmp_path, monkeypatch) -> None:
def test_sdk_migration_status_distinguishes_empty_and_legacy_data(tmp_path, monkeypatch) -> None:
monkeypatch.setattr(cli, "load_settings", lambda: Settings(_env_file=None, data_dir=tmp_path))
empty = io.StringIO()
assert cli.run(["sdk-migration-status", "--json"], output=empty) == 0
assert '"state": "empty"' in empty.getvalue()
def status() -> dict[str, object]:
output = io.StringIO()
assert cli.run(["sdk-migration-status", "--json"], output=output) == 0
return json.loads(output.getvalue())
assert status()["state"] == "not_required"
_legacy_home(tmp_path)
assert status() == {"schema_version": 1, "authority": "legacy", "state": "required"}
database = tmp_path / "one-memory.sqlite3"
with sqlite3.connect(database) as connection:
connection.execute("CREATE TABLE evidence (evidence_id TEXT PRIMARY KEY)")
connection.execute("INSERT INTO evidence VALUES ('e1')")
required = io.StringIO()
assert cli.run(["sdk-migration-status", "--json"], output=required) == 0
assert '"state": "required"' in required.getvalue()
assert '"raw": 1' in required.getvalue()
+18 -310
View File
@@ -18,13 +18,11 @@ from one_memory.diagnostic_log import (
diagnostic_exception,
diagnostics_enabled,
diagnostics_requested,
read_trace_events,
record_database_error,
reset_diagnostics,
reset_trace_id,
resource_snapshot,
)
from one_memory.exception_reporting import report_exception
def test_debug_events_are_collected_only_at_debug_level(tmp_path, monkeypatch) -> None:
@@ -107,31 +105,7 @@ def test_structured_log_is_private_correlated_and_owner_only(tmp_path):
assert log.path.stat().st_mode & 0o777 == 0o600
def test_arbitrary_valid_detail_keys_and_scalars_are_recorded(tmp_path):
log = DiagnosticLog(tmp_path)
details = {
"Custom.metric-v1": "ready",
"noneValue": None,
"boolValue": False,
"intValue": 7,
"floatValue": 1.5,
}
log.write(
level="info",
component="core.lifecycle",
event="core.started",
outcome="succeeded",
details=details,
)
assert log.flush()
assert json.loads(log.path.read_text())["details"] == details
@pytest.mark.parametrize(
"key",
["__proto__", "prototype", "constructor", "_private", "9startsWithDigit", "a" * 65],
)
def test_dangerous_or_malformed_detail_keys_are_rejected(tmp_path, key):
def test_arbitrary_detail_and_absolute_stack_are_rejected(tmp_path):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_detail_key_invalid"):
log.write(
@@ -139,76 +113,8 @@ def test_dangerous_or_malformed_detail_keys_are_rejected(tmp_path, key):
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={key: "value"},
details={"prompt": "private"},
)
@pytest.mark.parametrize("value", [{}, [], object()])
def test_complex_detail_values_are_rejected(tmp_path, value):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_detail_value_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={"payload": value},
)
@pytest.mark.parametrize("value", [float("nan"), float("inf"), float("-inf")])
def test_non_finite_detail_values_are_rejected(tmp_path, value):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_detail_value_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={"metric": value},
)
@pytest.mark.parametrize("value", [2**53, -(2**53), float(2**53), -float(2**53)])
def test_numeric_detail_values_outside_json_safe_range_are_rejected(tmp_path, value):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_detail_value_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={"count": value},
)
@pytest.mark.parametrize("value", ["", "nul\0byte", "carriage\rreturn", "line\nfeed", "x" * 257])
def test_invalid_detail_strings_are_rejected(tmp_path, value):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_detail_value_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={"message": value},
)
def test_more_than_32_detail_keys_are_rejected(tmp_path):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_detail_count_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={f"key{index}": index for index in range(33)},
)
def test_absolute_or_credential_bearing_stack_is_rejected(tmp_path):
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_stack_invalid"):
log.write(
level="error",
@@ -217,90 +123,16 @@ def test_absolute_or_credential_bearing_stack_is_rejected(tmp_path):
outcome="failed",
safe_stack=("/Users/me/private.py:1",),
)
with pytest.raises(ValueError, match="diagnostic_stack_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
safe_stack=("one_memory/sk-live-secret-token.py:1:call",),
)
def test_safe_stack_uses_strict_shared_grammar_and_sixteen_line_limit(tmp_path):
log = DiagnosticLog(tmp_path)
for line in (
"arbitrary stack text",
"thread=worker|threading.py:12:wait",
"FILE:///Users/me/private.py:1:call",
"one_memory/worker.py:9:call /Users/private",
"one_memory/GOCSPX-privateCredential.py:9:call",
"packages/ASIA1234567890ABCDEF/file.ts:1:1",
"packages/Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8/file.ts:1:1",
):
with pytest.raises(ValueError, match="diagnostic_stack_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
safe_stack=(line,),
)
def test_a_rejected_payload_does_not_break_the_caller(tmp_path, monkeypatch) -> None:
"""Rejection protects the log; it must not become the caller's outage.
accepted = (
"apps/desktop/src/main.ts:80:1",
"one_memory/diagnostic_log.py:895:_safe_stack_line",
"threading.py:1045:_bootstrap_inner",
"thread|threading.py:1045:_bootstrap_inner > one_memory/worker.py:9:call",
)
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
safe_stack=accepted,
)
with pytest.raises(ValueError, match="diagnostic_stack_invalid"):
log.write(
level="error",
component="hy.bridge",
event="hy.failed",
outcome="failed",
safe_stack=tuple(f"one_memory/worker.py:{index + 1}:call" for index in range(17)),
)
assert log.flush()
assert json.loads(log.path.read_text())["safe_stack"] == list(accepted)
def test_invalid_level_is_strict_but_diagnostic_event_remains_fail_open(
tmp_path, monkeypatch
) -> None:
log = DiagnosticLog(tmp_path)
with pytest.raises(ValueError, match="diagnostic_level_invalid"):
log.write(
level="fatal", # type: ignore[arg-type]
component="core.lifecycle",
event="core.failed",
outcome="failed",
)
monkeypatch.setenv("ONE_MEMORY_DIAGNOSTICS", "1")
receipt = configure_diagnostics(tmp_path / "fail-open")
try:
diagnostic_event(
level="fatal",
component="core.lifecycle",
event="core.failed",
outcome="failed",
)
finally:
reset_diagnostics()
assert not receipt.path.exists()
def test_a_rejected_payload_does_not_break_the_caller_or_later_events(
tmp_path, monkeypatch
) -> None:
Two of our own nodes shipped with unlisted `details` keys. Both sat on a
request path, so the rejection surfaced as a 500 to the person using the
app — including the sedimentation switch, whose every request failed
because logging the toggle raised before the endpoint could answer.
"""
monkeypatch.setenv("ONE_MEMORY_DIAGNOSTICS", "1")
receipt = configure_diagnostics(tmp_path)
try:
@@ -309,122 +141,20 @@ def test_a_rejected_payload_does_not_break_the_caller_or_later_events(
component="memory.sedimentation",
event="memory_sedimentation.toggled",
outcome="succeeded",
details={"payload": ["not", "scalar"]},
)
diagnostic_event(
level="info",
component="memory.sedimentation",
event="memory_sedimentation.toggled",
outcome="succeeded",
details={"custom.metric": 1},
details={"prompt": "private"},
)
finally:
reset_diagnostics()
events = [json.loads(line) for line in receipt.path.read_text().splitlines()]
assert [event["details"] for event in events] == [{"custom.metric": 1}]
def test_trace_projection_does_not_expose_arbitrary_local_detail_keys(tmp_path):
log = DiagnosticLog(tmp_path)
log.write(
level="info",
component="core.api",
event="memory.request.completed",
outcome="succeeded",
request_id="request-1",
details={
"phase": "recall",
"reason": "/Users/private/query.txt",
"status": "sk-live-secret-token",
"custom.private": "local-only",
},
)
assert log.flush()
events = read_trace_events(tmp_path, request_id="request-1")
assert events[0]["details"] == {"phase": "recall"}
def test_trace_projection_rejects_symlinked_directory_and_redacts_top_level_secrets(
tmp_path,
) -> None:
request_id = "request-1"
logs = tmp_path / "logs"
logs.mkdir()
event = {
"timestamp": "2026-10-23T01:00:00Z",
"request_id": request_id,
"component": "secret.token",
"event": "credential.failed",
"outcome": "Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8",
"error_code": "GOCSPX-clientSecret_A1",
"error_type": "ASIA1234567890ABCDEF",
"details": {
"phase": "recall",
"reason": "x_ASIA1234567890ABCDEF",
"mode": "x_GOCSPX-clientSecret_A1",
"status": "Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8",
"failure_site": "AIzaSyExampleCredential1234567890",
"count": 1,
},
}
(logs / "diagnostic-core.jsonl").write_text(json.dumps(event) + "\n")
projected = read_trace_events(tmp_path, request_id=request_id)
assert projected == (
{
"timestamp": "2026-10-23T01:00:00Z",
"component": "diagnostic.redacted",
"event": "diagnostic.redacted",
"outcome": "redacted",
"details": {"phase": "recall", "count": 1},
},
)
external = tmp_path / "external"
external.mkdir()
(logs / "diagnostic-core.jsonl").unlink()
logs.rmdir()
logs.symlink_to(external, target_is_directory=True)
with pytest.raises(ValueError, match="diagnostic_directory_unsafe"):
read_trace_events(tmp_path, request_id=request_id)
def test_credential_detection_is_conservative_and_matches_export_rules() -> None:
for value in (
"ASIA1234567890ABCDEF",
"GOCSPX-clientSecret_A1",
"Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8",
"Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8_",
):
assert diagnostic_log._contains_credential_material(value)
for value in (
"11111111-1111-4111-8111-111111111111",
"abcdef1234567890abcdef1234567890abcdef12",
"cursor",
):
assert not diagnostic_log._contains_credential_material(value)
def test_trace_reader_never_follows_a_diagnostic_file_symlink(tmp_path) -> None:
logs = tmp_path / "logs"
logs.mkdir()
outside = tmp_path / "outside.jsonl"
outside.write_text(
json.dumps(
{
"timestamp": "2026-10-23T01:00:00Z",
"request_id": "request-1",
"component": "core.api",
"event": "memory.request.completed",
"outcome": "succeeded",
}
)
+ "\n"
)
(logs / "diagnostic-core.jsonl").symlink_to(outside)
assert read_trace_events(tmp_path, request_id="request-1") == ()
# Rejected, so nothing is written — but the call returned.
assert not receipt.path.exists() or receipt.path.read_text() == ""
def test_the_sedimentation_toggle_payload_is_recorded(tmp_path, monkeypatch) -> None:
"""`enabled` is a bounded boolean and has to be a legal detail key.
Omitting it from the allow-list is what made the switch's own endpoint
answer 500 while still saving the new value.
"""
monkeypatch.setenv("ONE_MEMORY_DIAGNOSTICS", "1")
receipt = configure_diagnostics(tmp_path)
try:
@@ -451,7 +181,7 @@ def test_exception_diagnostic_keeps_code_location_not_message():
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={"prompt": f"{secret}\n"},
details={"prompt": secret},
)
except ValueError as error:
safe = diagnostic_exception(error)
@@ -461,28 +191,6 @@ def test_exception_diagnostic_keeps_code_location_not_message():
assert str(__file__) not in json.dumps(safe)
def test_report_exception_never_writes_the_exception_message_to_diagnostics(
tmp_path, monkeypatch
) -> None:
monkeypatch.setenv("ONE_MEMORY_DIAGNOSTICS", "1")
receipt = configure_diagnostics(tmp_path)
secret = "sk-live-private-provider-response"
try:
report_exception(
"core.test",
"core.test.failed",
RuntimeError(f"provider failed with {secret} at /Users/private/query.txt"),
)
finally:
reset_diagnostics()
payload = receipt.path.read_text()
assert secret not in payload
assert "/Users/private" not in payload
event = json.loads(payload)
assert event["details"] == {"count": 1}
assert event["error_type"] == "RuntimeError"
def test_old_logs_are_removed_but_symlinks_are_never_followed(tmp_path):
log = DiagnosticLog(tmp_path)
log.path.parent.mkdir(parents=True)
@@ -449,7 +449,7 @@ def test_invalid_input_is_still_rejected_on_the_caller_side(tmp_path):
component="hy.bridge",
event="hy.failed",
outcome="failed",
details={"invalid key": "private"},
details={"prompt": "private"},
)
log.close()
@@ -1,57 +1,22 @@
"""An instrument must never be able to break the thing it measures.
`diagnostic_event` validates its payload: a malformed `details` key, a missing
`level`, or an unsafe detail value is rejected. A diagnostic payload must never
be able to break the request path or loop heartbeat it is intended to measure.
`diagnostic_event` validates its payload: an unlisted `details` key, a missing
`level`, or a detail value over 256 characters all raise. Two of our own nodes
shipped with unlisted keys, and because they sat on a request path and on the
loop heartbeat, they turned every external recall into a 500 and then killed
the process. Both were caught only by running the built runtime, not by tests.
So this checks the payloads at their source, by walking every literal call.
"""
import ast
import re
from pathlib import Path
import pytest
from one_memory.diagnostic_log import _DETAIL_KEYS
ROOT = Path(__file__).resolve().parent.parent / "src" / "one_memory"
DETAIL_KEY = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]{0,63}$")
REFUSED_DETAIL_KEYS = {"__proto__", "prototype", "constructor"}
EXPORTED_STRING_DETAIL_KEYS = {
"action",
"method",
"mode",
"phase",
"platform",
"reason",
"source_id",
"status",
"version",
"parent_session_id",
"runtime_state",
"core_state",
"broker_state",
"instance_id",
"operation_kind",
"file_kind",
"last_probe",
"signal",
}
PRIVATE_DETAIL_KEYS = {
"api_key",
"authorization",
"content",
"credential",
"memory",
"memory_text",
"message",
"messages",
"password",
"prompt",
"query",
"raw_text",
"secret",
"token",
}
#: Every module, not a hand-picked list. The two payloads that caused real
#: outages were in files nobody thought to add here, so the only shape of this
@@ -68,19 +33,6 @@ def _literal_calls(path: Path):
yield node
def _contains_free_text_conversion(node: ast.AST) -> bool:
for item in ast.walk(node):
if (
isinstance(item, ast.Call)
and isinstance(item.func, ast.Name)
and item.func.id in {"str", "repr", "_safe_message"}
):
return True
if isinstance(item, ast.Attribute) and item.attr in {"args", "message"}:
return True
return False
@pytest.mark.parametrize("relative", SOURCES)
def test_every_diagnostic_payload_is_valid(relative: str) -> None:
path = ROOT / relative
@@ -93,22 +45,7 @@ def test_every_diagnostic_payload_is_valid(relative: str) -> None:
for keyword in node.keywords:
if keyword.arg != "details" or not isinstance(keyword.value, ast.Dict):
continue
for key, expression in zip(keyword.value.keys, keyword.value.values, strict=True):
if not isinstance(key, ast.Constant):
continue
value = key.value
if (
not isinstance(value, str)
or not DETAIL_KEY.fullmatch(value)
or value in REFUSED_DETAIL_KEYS
):
problems.append(f"{relative}:{node.lineno}: invalid details key {value!r}")
elif value.lower() in PRIVATE_DETAIL_KEYS:
problems.append(f"{relative}:{node.lineno}: private details key {value!r}")
elif value in EXPORTED_STRING_DETAIL_KEYS and _contains_free_text_conversion(
expression
):
problems.append(
f"{relative}:{node.lineno}: exported detail {value!r} uses free text"
)
for key in keyword.value.keys:
if isinstance(key, ast.Constant) and key.value not in _DETAIL_KEYS:
problems.append(f"{relative}:{node.lineno}: unlisted details key {key.value!r}")
assert not problems, "\n".join(problems)
+2 -4
View File
@@ -92,13 +92,11 @@ def test_each_fault_outcome_is_realised_by_the_durable_queue(tmp_path) -> None:
self.error = error
self.seen: list[str] = []
async def process_sdk_job(self, records, **_kwargs):
# Refusals are classified before `before_hy_invocation`, as the
# SDK write processor does, so these never cross into HY.
async def process_records(self, records, **_kwargs):
self.seen.append("called")
if self.error is not None:
raise self.error
return type("Result", (), {"raw_ids": (), "memory_ids": ()})()
return ()
# hy_succeeded_then_crash: marked invoked, then the lease lapses -> unknown.
path = tmp_path / "fault.sqlite3"
+4 -30
View File
@@ -2,7 +2,6 @@
import asyncio
import json
import os
import sqlite3
from contextlib import closing
from io import StringIO
@@ -21,7 +20,6 @@ from one_memory.memory_server.durable_queue import (
DurableWriteQueue,
execute_job,
)
from one_memory.sdk_write_processor import SdkWriteReceipt
def _grant(tmp_path, source_id: str, *, sync: bool) -> EvidenceStore:
@@ -235,14 +233,14 @@ def test_hook_job_completes_after_file_identity_adoption(tmp_path) -> None:
).fetchone()
class IdentityCheckingSupervisor:
async def process_sdk_job(
async def process_records(
self,
records,
*,
request_id,
before_hy_invocation,
source_admitted,
):
assert request_id
assert source_admitted is True
with closing(sqlite3.connect(database)) as connection:
connection.execute("BEGIN")
for record in records:
@@ -251,7 +249,7 @@ def test_hook_job_completes_after_file_identity_adoption(tmp_path) -> None:
assert ingest_journal.matches_current(connection, current)
connection.rollback()
await before_hy_invocation()
return SdkWriteReceipt()
return ()
drain = DurableWriteDrain(queue, execute=lambda _job: None)
assert drain.claim(limit=1) == (job_id,)
@@ -562,30 +560,6 @@ def test_capture_failure_stays_silent(tmp_path, monkeypatch) -> None:
assert output.getvalue() == ""
def test_capture_is_a_silent_noop_during_sdk_migration(tmp_path, monkeypatch) -> None:
database = tmp_path / "one-memory.sqlite3"
marker = tmp_path / ".sdk-migration-capture-fence.json"
marker.write_text(json.dumps({"schema_version": 1, "pid": os.getpid()}))
monkeypatch.setattr("one_memory.live_capture.database_path", lambda: database)
monkeypatch.setattr(
"sys.stdin",
StringIO(
json.dumps(
{
"hook_event_name": "beforeSubmitPrompt",
"conversation_id": "session",
"prompt": "not persisted during migration",
}
)
),
)
output = StringIO()
assert cli.run(["capture", "cursor"], output=output) == 0
assert output.getvalue() == ""
assert not database.exists()
def test_codex_harness_preamble_is_not_a_user_message(tmp_path) -> None:
database = tmp_path / "one-memory.sqlite3"
_grant(tmp_path, "codex", sync=True)
+2 -4
View File
@@ -102,8 +102,6 @@ def test_stall_captures_a_safe_stack_for_every_thread() -> None:
assert watchdog.stall_stacks
captured = watchdog.stall_stacks[0]
assert captured
assert len(captured) <= 16
assert all(line.startswith("thread|") for line in captured)
assert all("source-reader-test" not in line and "MainThread" not in line for line in captured)
assert any(line.startswith("thread=MainThread|") for line in captured)
assert any(line.startswith("thread=source-reader-test|") for line in captured)
assert all("/Users/" not in line and "\n" not in line for line in captured)
@@ -6,7 +6,6 @@ import asyncio
import json
import sqlite3
from datetime import UTC, datetime
from types import SimpleNamespace
import pytest
@@ -250,17 +249,16 @@ def test_old_branch_job_executes_its_frozen_revision_after_harness_rewind(tmp_pa
captured = []
class Supervisor:
async def process_sdk_job(
async def process_records(
self,
records,
*,
request_id,
before_hy_invocation,
source_admitted,
):
assert request_id == old_job.request_id
assert source_admitted is True
captured.extend(records)
await before_hy_invocation()
return SimpleNamespace(raw_ids=(), memory_ids=())
asyncio.run(
execute_job(
File diff suppressed because it is too large. Load diff
@@ -91,7 +91,7 @@ def test_bundle_contains_management_sdk_config_and_verified_manifest(tmp_path):
assert manifest["bundle_id"] == "migration-1"
assert manifest["sdk_namespace"] == "governed-v2"
assert manifest["runtime_build"] == "build-13"
assert manifest["snapshot_sdk_version"] == "2.0.5"
assert manifest["snapshot_sdk_version"] == "2.0.5.dev3"
paths = {item["path"]: item for item in manifest["files"]}
assert {
"management/one-memory.sqlite3",
+5 -5
View File
@@ -991,7 +991,7 @@ wheels = [
[[package]]
name = "hy-memory"
version = "2.0.5"
version = "2.0.5.dev3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "chromadb" },
@@ -1008,9 +1008,9 @@ dependencies = [
{ name = "scikit-learn" },
{ name = "tiktoken" },
]
sdist = { url = "https://files.pythonhosted.org/packages/94/56/7a279122bb56b067c390b748b8a83b6a90bb2a8f237c1d3989b6317681cd/hy_memory-2.0.5.tar.gz", hash = "sha256:bfb1e9e9815377477702327fa318331d555b8256a67b5bad8d70d6a682bd9a4f", size = 839060, upload-time = "2026-10-22T12:59:30.502Z" }
sdist = { url = "https://files.pythonhosted.org/packages/cb/83/67c261bc379eb45f1cc460cf55eb057867d485695c2799bd861827dbcedb/hy_memory-2.0.5.dev3.tar.gz", hash = "sha256:8ae8c2b2d9736b304d26654a5c2903160f430ec5065206ae724d5cb841820d75", size = 836317, upload-time = "2026-10-21T09:59:27.311Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c5/97/090ca18b6a13d0216d2df607cb67732ff7a8eba2d0359b9ad95283b2cbdc/hy_memory-2.0.5-py3-none-any.whl", hash = "sha256:c24649d188eaa03d3b4e4e4ec20e818da884170b5f775f3cf23ecfa6b60523d6", size = 902923, upload-time = "2026-10-22T12:59:28.018Z" },
{ url = "https://files.pythonhosted.org/packages/b6/5f/30a88b0278c070c63d11f0c25a360e1cfe130d90a53a5eece9359c4ca7ec/hy_memory-2.0.5.dev3-py3-none-any.whl", hash = "sha256:621193ac3a527c0b8736d66f17c4cce4956dc4a882d9dfc724b9df4d34fe816e", size = 900779, upload-time = "2026-10-21T09:59:24.716Z" },
]
[[package]]
@@ -1801,7 +1801,7 @@ wheels = [
[[package]]
name = "one-memory"
version = "1.0.5"
version = "1.0.4"
source = { editable = "." }
dependencies = [
{ name = "aiomysql" },
@@ -1837,7 +1837,7 @@ requires-dist = [
{ name = "fastapi", specifier = ">=0.141.1" },
{ name = "filelock", specifier = ">=3.32.4" },
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "hy-memory", specifier = "==2.0.5" },
{ name = "hy-memory", specifier = "==2.0.5.dev3" },
{ name = "keyring", specifier = ">=25.7.0" },
{ name = "numpy", specifier = "==2.5.2" },
{ name = "onnxruntime", specifier = "==1.23.2" },
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/desktop",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"main": "dist/main.js",
"scripts": {
@@ -94,14 +94,12 @@ it("returns only diagnostic events written during this desktop launch", async ()
await mkdir(logs, { recursive: true });
const target = path.join(logs, "diagnostic-core.jsonl");
const old = {
schema_version: 1,
timestamp: "2026-10-14T09:00:00.000Z",
level: "info",
service: "core",
component: "core.startup",
event: "core.old",
outcome: "succeeded",
session_id: "11111111-1111-4111-8111-111111111111",
};
await writeFile(target, `${JSON.stringify(old)}\n`);
const cursor = await captureDebugLogCursor(dataDir);
@@ -111,11 +109,6 @@ it("returns only diagnostic events written during this desktop launch", async ()
timestamp: "2026-10-14T10:01:00.000Z",
event: "core.current",
level: "warn",
details: {
count: 1,
"custom.private": "/Users/private/query.txt",
credential_hint: "ghp_private_token",
},
};
await appendFile(target, `${JSON.stringify(current)}\nnot-json\n`);
@@ -125,11 +118,8 @@ it("returns only diagnostic events written during this desktop launch", async ()
expect(snapshot.entries[0]).toMatchObject({
event: "core.current",
level: "warn",
details: { count: 1 },
source: "diagnostic",
});
expect(JSON.stringify(snapshot)).not.toContain("/Users/private");
expect(JSON.stringify(snapshot)).not.toContain("ghp_private_token");
});
it("tracks diagnostic JSONL across rotation without rereading the old prefix", async () => {
@@ -140,14 +130,12 @@ it("tracks diagnostic JSONL across rotation without rereading the old prefix", a
const target = path.join(logs, "diagnostic-core.jsonl");
const event = (name: string, timestamp: string) =>
JSON.stringify({
schema_version: 1,
timestamp,
level: "info",
service: "core",
component: "core.test",
event: name,
outcome: "succeeded",
session_id: "22222222-2222-4222-8222-222222222222",
});
await writeFile(target, `${event("before", "2026-10-14T09:00:00.000Z")}\n`);
const cursor = await captureDebugLogCursor(dataDir);
+51 -51
View File
@@ -1,13 +1,10 @@
import { lstat, open, readdir } from "node:fs/promises";
import path from "node:path";
import {
type DiagnosticEventV1,
type DiagnosticLogger,
diagnosticEvent,
diagnosticZipFilename,
exportDiagnosticZip,
inspectBackgroundHealth,
projectDiagnosticEventForSafeOutput,
recordDiagnosticMarker,
} from "@one-memory/runtime-control";
import type { BrowserWindow, IpcMainInvokeEvent } from "electron";
@@ -373,58 +370,42 @@ function debugLogEntry(
const timestamp = typeof value.timestamp === "string" ? value.timestamp : "";
const timestampMs = Date.parse(timestamp);
if (!Number.isFinite(timestampMs) || timestampMs < startedAt) return null;
let projected: DiagnosticEventV1;
try {
projected = projectDiagnosticEventForSafeOutput(
diagnosticEvent({
timestamp,
level: value.level as never,
service: value.service as never,
component: value.component as never,
event: value.event as never,
outcome: value.outcome as never,
session_id: value.session_id as never,
...(typeof value.trace_id === "string"
? { trace_id: value.trace_id }
: {}),
...(typeof value.duration_ms === "number"
? { duration_ms: value.duration_ms }
: {}),
...(typeof value.error_code === "string"
? { error_code: value.error_code }
: {}),
...(typeof value.error_type === "string"
? { error_type: value.error_type }
: {}),
...(value.details && typeof value.details === "object"
? { details: value.details as never }
: {}),
}),
);
} catch {
const level = value.level;
const service = value.service;
const component = value.component;
const event = value.event;
const outcome = value.outcome;
if (
!["info", "warn", "error"].includes(String(level)) ||
!["runtime", "core", "host-bridge", "desktop", "cli"].includes(
String(service),
) ||
typeof component !== "string" ||
typeof event !== "string" ||
typeof outcome !== "string"
) {
return null;
}
const details = safeDebugDetails(value.details);
return {
id: `${filename}:${lineNumber}:${timestamp}`,
timestamp: projected.timestamp,
level: projected.level,
service: projected.service,
component: projected.component,
event: projected.event,
outcome: projected.outcome,
...(projected.duration_ms === undefined
? {}
: { duration_ms: projected.duration_ms }),
...(projected.error_code === undefined
? {}
: { error_code: projected.error_code }),
...(projected.error_type === undefined
? {}
: { error_type: projected.error_type }),
...(projected.trace_id === undefined
? {}
: { trace_id: projected.trace_id }),
...(projected.details === undefined ? {} : { details: projected.details }),
timestamp,
level: level as DebugLogEntry["level"],
service: service as DebugLogEntry["service"],
component,
event,
outcome,
...(typeof value.duration_ms === "number"
? { duration_ms: value.duration_ms }
: {}),
...(typeof value.error_code === "string"
? { error_code: value.error_code }
: {}),
...(typeof value.error_type === "string"
? { error_type: value.error_type }
: {}),
...(typeof value.trace_id === "string" ? { trace_id: value.trace_id } : {}),
...(details === undefined ? {} : { details }),
source: "diagnostic",
};
}
@@ -542,6 +523,25 @@ function logLineTimestamp(
return new Date(Math.max(startedAt, fallback)).toISOString();
}
function safeDebugDetails(
value: unknown,
): Record<string, string | number | boolean | null> | undefined {
if (!value || typeof value !== "object" || Array.isArray(value))
return undefined;
const details: Record<string, string | number | boolean | null> = {};
for (const [key, item] of Object.entries(value as Record<string, unknown>)) {
if (
item === null ||
typeof item === "string" ||
typeof item === "number" ||
typeof item === "boolean"
) {
details[key] = item;
}
}
return Object.keys(details).length ? details : undefined;
}
async function pipelineLogFiles(root: string): Promise<PipelineLogCursor[]> {
const rootInfo = await lstat(root).catch(() => null);
if (!rootInfo?.isDirectory() || rootInfo.isSymbolicLink()) return [];
+2 -3
View File
@@ -113,7 +113,6 @@ import {
installServiceWithDisabledFallback,
isInstalledApplicationsApp,
stableRuntimeErrorCode,
stableRuntimeFailureDetail,
startStableRuntimeRepair,
} from "./stable-runtime";
import { STARTUP_SCREEN_URL, startupScreenHtml } from "./startup-screen";
@@ -1099,7 +1098,7 @@ async function initialize(isQuitting: () => boolean): Promise<void> {
? "blocked"
: "skipped",
...(migration.code === undefined ? {} : { error_code: migration.code }),
details: { runtime_state: migration.state },
details: { state: migration.state },
});
requireCompletedStartupSdkMigration(migration);
};
@@ -1672,7 +1671,7 @@ if (desktopVersion !== null) {
languagePreferences?.read().locale ?? "zh-CN",
"nativeErrorTitle",
),
stableRuntimeFailureDetail(error),
error instanceof Error ? error.message : "desktop_unavailable",
);
},
});
@@ -16,7 +16,6 @@ import {
isDisabledServiceInstallError,
isInstalledApplicationsApp,
stableRuntimeErrorCode,
stableRuntimeFailureDetail,
stableRuntimeUserMessage,
startStableRuntimeRepair,
} from "./stable-runtime";
@@ -275,34 +274,4 @@ describe("startup install errors", () => {
"MemHub 启动失败",
);
});
it("says why a data upgrade failed, and what the person can do", () => {
const noSpace = stableRuntimeUserMessage(
new Error("sdk_migration_disk_space_insufficient"),
);
// The only failure here the person fixes themselves: name the action.
expect(noSpace).toContain("磁盘空间不足");
expect(noSpace).toContain("重新启动");
expect(noSpace).toContain("不会丢失");
expect(noSpace).not.toContain("sdk_migration_disk_space_insufficient");
});
});
describe("startup dialog body", () => {
it("keeps a code we have no copy for, so support can see which failure fired", () => {
expect(stableRuntimeFailureDetail(new Error("page_load_failed"))).toBe(
"page_load_failed",
);
expect(
stableRuntimeFailureDetail(new Error("runtime_broker_unavailable")),
).toBe("runtime_broker_unavailable");
});
it("shows the sentence we wrote for a migration failure", () => {
const detail = stableRuntimeFailureDetail(
new Error("sdk_migration_disk_space_insufficient"),
);
expect(detail).toContain("磁盘空间不足");
expect(detail).not.toContain("sdk_migration_disk_space_insufficient");
});
});
-37
View File
@@ -126,16 +126,6 @@ const SAFE_STABLE_RUNTIME_ERROR_CODES = new Set([
"installed_version_conflict",
"distribution_downgrade_refused",
"install_current_unmanaged",
// Migration runs before the app window exists, so these are the only names
// a first launch can be given. They stay codes here and become sentences in
// STABLE_RUNTIME_USER_MESSAGES: a person reading the dialog never sees them.
"sdk_migration_preflight_failed",
"sdk_migration_disk_space_insufficient",
"sdk_migration_cutover_failed",
"sdk_migration_bundle_copy_failed",
"sdk_migration_response_invalid",
"sdk_migration_status_unavailable",
"sdk_migration_incomplete",
]);
export function stableRuntimeErrorCode(error: unknown): string {
@@ -201,20 +191,6 @@ export const STABLE_RUNTIME_USER_MESSAGES: Record<string, string> = {
"已安装的运行环境比这个安装包更新,无法用旧安装包启动。请安装最新版本的 MemHub;你的记忆数据不会受到影响。",
install_current_unmanaged:
"运行环境的安装目录被改动过,无法安全接管。请重新安装 MemHub,你的记忆数据不会受到影响。",
sdk_migration_disk_space_insufficient:
"磁盘空间不足,升级记忆数据格式时中断了。请至少腾出与记忆数据同等大小的可用空间(可在「关于本机 → 储存空间」查看),然后重新启动 MemHub;升级会自动继续,原有记忆数据不会丢失。",
sdk_migration_bundle_copy_failed:
"升级前备份没有写完,通常是磁盘空间不足。请清理磁盘空间后重新启动;原有记忆数据不会丢失。",
sdk_migration_cutover_failed:
"切换数据源时失败,原有记忆数据仍然完整可用。请重新启动重试;如果仍然失败,请把这个提示反馈给我们。",
sdk_migration_preflight_failed:
"升级前的检查没有通过,升级已停止。请重新启动重试;如果仍然失败,请把这个提示反馈给我们,原有记忆数据不会丢失。",
sdk_migration_status_unavailable:
"读不到升级状态,无法判断是否需要升级。请重新启动 MemHub;如果仍然失败,请重新安装,你的记忆数据不会受到影响。",
sdk_migration_response_invalid:
"升级过程返回了无法识别的内容,已停止。原有记忆数据不会丢失;请重新启动重试,如果仍然失败,请把这个提示反馈给我们。",
sdk_migration_incomplete:
"记忆数据格式升级没有完成,MemHub 暂时无法打开。请重新启动重试;如果仍然失败,请把这个提示反馈给我们,原有记忆数据不会丢失。",
desktop_start_failed:
"MemHub 启动失败。请重新启动;如果仍然打不开,请重新安装,你的记忆数据不会受到影响。",
};
@@ -227,16 +203,3 @@ export function stableRuntimeUserMessage(error: unknown): string {
""
);
}
/** What the startup dialog body says.
*
* A code we wrote copy for becomes a sentence. Every other code is shown as it
* is: this dialog is the last thing a person sees before the app closes, and an
* unexpected failure is more useful as the code support can look up than as a
* reassuring sentence that erases which failure actually fired. */
export function stableRuntimeFailureDetail(error: unknown): string {
if (!(error instanceof Error)) {
return STABLE_RUNTIME_USER_MESSAGES.desktop_start_failed ?? "";
}
return STABLE_RUNTIME_USER_MESSAGES[error.message] ?? error.message;
}
@@ -21,21 +21,6 @@ describe("startup screen", () => {
);
expect(spanish).not.toMatch(/[一-鿿]/);
});
it("says an upgrade may take a while instead of leaving a silent wait", () => {
expect(startupScreenHtml()).toContain(
"首次启动或升级记忆数据可能需要几分钟,完成后会自动进入应用。",
);
expect(startupScreenHtml("en")).toContain(
"The first launch or a memory-data upgrade can take a few minutes.",
);
});
it("keeps a draggable strip so the waiting window can be moved", () => {
const html = startupScreenHtml();
// CSS alone does not move a frameless window: the region has to be a real
// element in the document, and it has to be the top strip.
expect(html).toContain('<div class="window-drag-region"');
expect(html).toContain("-webkit-app-region: drag");
});
it("is served from the app origin, not from the web bundle", () => {
expect(STARTUP_SCREEN_URL).toBe("one-memory://app/__startup");
});
-22
View File
@@ -37,19 +37,6 @@ export function startupScreenHtml(locale: GuiLocale = "zh-CN"): string {
"Helvetica Neue", Arial, sans-serif;
-webkit-font-smoothing: antialiased;
}
/* A frameless window still needs somewhere to grab; without this the
whole wait is undraggable. */
.window-drag-region {
position: absolute;
z-index: 1;
top: 0;
right: 0;
left: 0;
height: 58px;
-webkit-app-region: drag;
-webkit-user-select: none;
user-select: none;
}
@media (prefers-color-scheme: dark) {
body {
background: #15181d;
@@ -74,13 +61,6 @@ export function startupScreenHtml(locale: GuiLocale = "zh-CN"): string {
font-size: 13px;
opacity: 0.62;
}
p.hint {
max-width: 320px;
font-size: 12px;
line-height: 1.55;
text-align: center;
opacity: 0.45;
}
.bar {
position: relative;
width: 168px;
@@ -121,11 +101,9 @@ export function startupScreenHtml(locale: GuiLocale = "zh-CN"): string {
</style>
</head>
<body>
<div class="window-drag-region" aria-hidden="true"></div>
<main>
<h1>MemHub</h1>
<p>${translate(locale, "startupPreparing")}</p>
<p class="hint">${translate(locale, "startupUpgradeHint")}</p>
<div class="bar"><span></span></div>
</main>
</body>
@@ -1,17 +1,155 @@
import { expect, it } from "vitest";
import { requireCompletedStartupSdkMigration } from "./startup-sdk-migration";
import { expect, it, vi } from "vitest";
import {
requireCompletedStartupSdkMigration,
runStartupSdkMigration,
} from "./startup-sdk-migration";
it("refuses to open the app on a deferred migration, naming its code", () => {
expect(() =>
requireCompletedStartupSdkMigration({
state: "deferred",
code: "sdk_migration_disk_space_insufficient",
const core = { command: "/core/one-memory", args: [] };
it("runs bundle backfill and cutover inside the startup maintenance fence", async () => {
const calls: string[][] = [];
const execute = vi.fn(async (_core, args: string[]) => {
calls.push(args);
const command = args[0];
return {
exitCode: 0,
document:
command === "sdk-migration-status"
? {
authority: "legacy",
state: "required",
generation: "0123456789abcdef",
}
: command === "sdk-migrate"
? { state: "ready" }
: command === "sdk-cutover"
? { state: "cutover" }
: { state: "created" },
};
});
const fenceCalls: string[] = [];
const fence = async <T>(
dataDir: string,
action: () => Promise<T>,
): Promise<T> => {
fenceCalls.push(dataDir);
return action();
};
const result = await runStartupSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "build-13",
execute,
fence,
});
expect(result).toEqual({ state: "cutover" });
expect(fenceCalls).toEqual(["/data"]);
expect(calls.map((args) => args[0])).toEqual([
"sdk-migration-status",
"migration-bundle",
"sdk-migrate",
"sdk-cutover",
]);
});
it("does not stop Runtime when authority is already SDK or the home is empty", async () => {
for (const document of [
{ authority: "sdk", state: "complete" },
{ authority: "legacy", state: "empty" },
]) {
let fenced = false;
const fence = async <T>(
_dataDir: string,
action: () => Promise<T>,
): Promise<T> => {
fenced = true;
return action();
};
const result = await runStartupSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "build-13",
execute: vi.fn(async () => ({ exitCode: 0, document })),
fence,
});
expect(result).toEqual({ state: "not_required" });
expect(fenced).toBe(false);
}
});
it("reports an unreadable migration status as deferred", async () => {
const result = await runStartupSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "build-13",
execute: vi.fn(async () => ({ exitCode: 1, document: {} })),
fence: async (_dataDir, action) => action(),
});
expect(result).toEqual({
state: "deferred",
code: "sdk_migration_status_unavailable",
});
expect(() => requireCompletedStartupSdkMigration(result)).toThrow(
"sdk_migration_status_unavailable",
);
});
it("resumes legacy table cleanup under the maintenance fence", async () => {
const calls: string[] = [];
const result = await runStartupSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "build-13",
execute: vi.fn(async (_core, args: string[]) => {
calls.push(args[0]);
return args[0] === "sdk-migration-status"
? {
exitCode: 0,
document: {
authority: "sdk",
state: "cleanup_required",
},
}
: { exitCode: 0, document: { state: "clean" } };
}),
).toThrow("sdk_migration_disk_space_insufficient");
expect(() =>
requireCompletedStartupSdkMigration({ state: "deferred" }),
).toThrow("sdk_migration_incomplete");
expect(() =>
requireCompletedStartupSdkMigration({ state: "cutover" }),
).not.toThrow();
fence: async (_dataDir, action) => action(),
});
expect(result).toEqual({ state: "cutover" });
expect(calls).toEqual(["sdk-migration-status", "sdk-cleanup"]);
});
it("keeps legacy authority when backfill is blocked", async () => {
const execute = vi.fn(async (_core, args: string[]) => {
const command = args[0];
return {
exitCode: command === "sdk-migrate" ? 4 : 0,
document:
command === "sdk-migration-status"
? {
authority: "legacy",
state: "required",
generation: "0123456789abcdef",
}
: command === "sdk-migrate"
? { state: "blocked", code: "quarantine" }
: { state: "created" },
};
});
const result = await runStartupSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "build-13",
execute,
fence: async (_dataDir, action) => action(),
});
expect(result).toEqual({ state: "deferred", code: "quarantine" });
expect(execute.mock.calls.map((call) => call[1][0])).not.toContain(
"sdk-cutover",
);
});
@@ -1,10 +1,13 @@
import {
runSdkMigration,
type SdkMigrationResult,
} from "@one-memory/runtime-control";
import { spawn } from "node:child_process";
import { withRuntimeMaintenanceFence } from "@one-memory/runtime-control";
export type StartupMigrationResult = SdkMigrationResult;
export const runStartupSdkMigration = runSdkMigration;
const MIGRATION_PREFIX = "sdk-authority-v1";
const MAX_OUTPUT_BYTES = 1024 * 1024;
export interface StartupMigrationResult {
state: "not_required" | "cutover" | "deferred";
code?: string;
}
export function requireCompletedStartupSdkMigration(
result: StartupMigrationResult,
@@ -13,3 +16,179 @@ export function requireCompletedStartupSdkMigration(
throw new Error(result.code ?? "sdk_migration_incomplete");
}
}
interface CommandResult {
exitCode: number;
document: Record<string, unknown>;
}
interface CoreCommand {
command: string;
args?: string[];
cwd?: string;
env?: NodeJS.ProcessEnv;
}
interface StartupMigrationOptions {
dataDir: string;
core: CoreCommand;
runtimeBuild: string;
execute?: (
launch: CoreCommand,
args: string[],
dataDir: string,
) => Promise<CommandResult>;
fence?: <T>(dataDir: string, action: () => Promise<T>) => Promise<T>;
}
export async function runStartupSdkMigration(
options: StartupMigrationOptions,
): Promise<StartupMigrationResult> {
const execute = options.execute ?? executeCoreCommand;
const fence = options.fence ?? withRuntimeMaintenanceFence;
const status = await execute(
options.core,
["sdk-migration-status", "--json"],
options.dataDir,
);
if (status.exitCode !== 0) {
return { state: "deferred", code: "sdk_migration_status_unavailable" };
}
if (
status.document.authority === "sdk" &&
status.document.state === "cleanup_required"
) {
return fence(options.dataDir, async () => {
const cleanup = await execute(
options.core,
["sdk-cleanup", "--yes", "--json"],
options.dataDir,
);
return cleanup.exitCode === 0 && cleanup.document.state === "clean"
? { state: "cutover" }
: deferred(cleanup);
});
}
if (
status.document.authority === "sdk" ||
status.document.state !== "required"
) {
return { state: "not_required" };
}
const generation = status.document.generation;
if (typeof generation !== "string" || !/^[a-f0-9]{16}$/.test(generation)) {
return { state: "deferred", code: "sdk_migration_generation_invalid" };
}
const migrationId = `${MIGRATION_PREFIX}-${generation}`;
const bundleId = `${migrationId}-pre-cutover`;
return fence(options.dataDir, async () => {
const bundle = await execute(
options.core,
[
"migration-bundle",
"--bundle-id",
bundleId,
"--runtime-build",
options.runtimeBuild,
"--yes",
"--json",
],
options.dataDir,
);
if (bundle.exitCode !== 0) return deferred(bundle);
const backfill = await execute(
options.core,
[
"sdk-migrate",
"--migration-id",
migrationId,
"--bundle-id",
bundleId,
"--yes",
"--json",
],
options.dataDir,
);
if (backfill.exitCode !== 0 || backfill.document.state !== "ready") {
return deferred(backfill);
}
const cutover = await execute(
options.core,
[
"sdk-cutover",
"--migration-id",
migrationId,
"--bundle-id",
bundleId,
"--yes",
"--json",
],
options.dataDir,
);
if (cutover.exitCode !== 0 || cutover.document.state !== "cutover") {
return deferred(cutover);
}
return { state: "cutover" };
});
}
function deferred(result: CommandResult): StartupMigrationResult {
return {
state: "deferred",
code:
typeof result.document.code === "string"
? result.document.code
: "sdk_migration_incomplete",
};
}
async function executeCoreCommand(
launch: CoreCommand,
args: string[],
dataDir: string,
): Promise<CommandResult> {
return new Promise((resolve, reject) => {
const child = spawn(launch.command, [...(launch.args ?? []), ...args], {
cwd: launch.cwd,
env: { ...process.env, ...launch.env, ONE_MEMORY_DATA_DIR: dataDir },
stdio: ["ignore", "pipe", "pipe"],
});
let stdout = "";
let stderrBytes = 0;
child.stdout.setEncoding("utf8");
child.stdout.on("data", (chunk: string) => {
stdout += chunk;
if (Buffer.byteLength(stdout) > MAX_OUTPUT_BYTES) {
child.kill("SIGKILL");
reject(new Error("sdk_migration_output_too_large"));
}
});
child.stderr.on("data", (chunk: Buffer) => {
stderrBytes += chunk.length;
if (stderrBytes > MAX_OUTPUT_BYTES) {
child.kill("SIGKILL");
reject(new Error("sdk_migration_error_output_too_large"));
}
});
child.once("error", reject);
child.once("close", (code) => {
try {
const line = stdout.trim().split("\n").at(-1) ?? "{}";
const document = JSON.parse(line) as unknown;
if (
!document ||
typeof document !== "object" ||
Array.isArray(document)
) {
throw new Error("sdk_migration_response_invalid");
}
resolve({
exitCode: code ?? 1,
document: document as Record<string, unknown>,
});
} catch {
reject(new Error("sdk_migration_response_invalid"));
}
});
});
}
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/site",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"scripts": {
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/web",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"scripts": {
+11 -11
View File
@@ -1,6 +1,6 @@
# MemHub Architecture
Updated: 2026-10-24.
Updated: 2026-10-21.
Hy Memory 是 SDK / 引擎;MemHub 是产品与发布名。`one-memory` 仍可能出现在
兼容路径、模块名和服务标识中。
@@ -129,15 +129,15 @@ directly for low-latency Cypher projection and store-width administration.
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 不受该窗口影响。
- Queue:pending `memory_write_jobs` 只表示尚未调用 SDK 的待处理请求,因此没有 SDK raw/memory ID。worker 单次调用 SDK `add()`;SDK 内部完成 raw 写入与 extraction/reconciliation,成功后 MemHub 仅将返回的 raw ID / memory IDs 记作 job outcome。旧用户启动迁移在 Runtime writers 停止期间直接操作 SDK 存储,不向公开 `HyMemoryClient` 增加迁移专用方法,也不拆分正常写路径。
- 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。
- Matter:未形成已发布产品合同,不迁移占位数据;所有 `matter*` 表在升级清理时物理删除,fresh install 不创建,Runtime 不注册 Matter API。
- 旧用户迁移:由 App 在可见的启动 loading 阶段统一执行;此时 Core 尚未对外提供产品读写。所有已发布版本都先经 SDK 写入,Governance/Evidence 只是镜像,所以迁移**不复制、不比对任何 memory 或 raw**:只创建并校验完整 bundle、让写入队列脱离旧表、提交 marker、删除旧表。由一个 Core 命令 `sdk-migrate` 完成;Desktop 启动与 CLI `memhub install`/`upgrade`(macOS、Linux)在启动新 Runtime 前调用它,普通用户不需要另跑命令。App 不得在 migration `deferred` 状态继续启动 Runtime。
- 迁移期间宿主 capture hook 由 Desktop 活跃 PID marker 静默挡住,不再追加
Evidence;cutover 后先从宿主 History checkpoint 补齐遗漏 turn,再恢复 live sync。
- 旧用户迁移:由 App 在可见的启动 loading 阶段统一执行;此时 Core 尚未对外提供产品读写。启动流程创建/校验完整 bundle、续跑 ledger、完成 SDK backfill 与 reconciliation,满足全部门禁后才提交 marker 并进入主界面。普通用户不需要另跑 CLI,也不按旧 App 内 SDK 版本分叉迁移。
- SDK backfill 只写同文件系统的私有 staging 副本;验证后先原子切换 SDK
root,再提交 authority marker。切换或 marker 提交失败必须恢复原 live root;
App 不得在 migration `deferred` 状态继续启动 Runtime。
- Matter 若未来重新开发,需以新的 SDK-native 合同重新设计,不继承当前占位 schema。
Pinned SDK 版本见发布锁文件 / `MEMORY.md`(随发布更新)。
@@ -282,7 +282,7 @@ Support is declared per immutable GitHub Release manifest:
- Linux x64/arm64 Runtime on native GHA runners.
- macOS arm64/x64 App/Runtime: Developer ID, notarization, native acceptance.
- Windows is not supported.
- `hy-memory==2.0.5` is the current stable SDK dependency; the release
- `hy-memory==2.0.5.dev3` is the current development-preview SDK dependency; the release
lock and [`../MEMORY.md`](../MEMORY.md) are authoritative when the pin changes.
See [`INSTALLATION.md`](INSTALLATION.md).
@@ -303,9 +303,9 @@ Test entry points: [`../CONTRIBUTING.md`](../CONTRIBUTING.md).
## Upgrade Compatibility
Released installations require the offline compatibility protocol in
Released installations require the offline, resumable compatibility protocol in
[`contracts/sdk-data-plane-migration-v1.md`](contracts/sdk-data-plane-migration-v1.md).
The App runs it during the startup loading phase before Core exposes product
reads or writes. A cutover requires a verified full bundle and a write queue
that no longer depends on legacy tables; it copies no memory, because the SDK
stores already hold every memory and raw turn.
reads or writes. A cutover requires a complete bundle, resumable backfill,
reconciliation, and verified cleanup; a SQLite-only backup or replaying legacy
rows through extraction is not sufficient.
+2 -2
View File
@@ -1,6 +1,6 @@
# MemHub 安装与发行
MemHub `v1.0.5`(build 14)的发行目标如下:
MemHub `v1.0.4`(build 13)的发行目标如下:
| 平台 | 发行物 | 构建与验收 |
| --- | --- | --- |
@@ -100,7 +100,7 @@ Downloader 使用标准 `HTTPS_PROXY`/`https_proxy` 与 `NO_PROXY`/`no_proxy`
- `npm uninstall -g memhub-cli`:只删除 npm CLI;不停止或删除受管 Runtime,不删除 Memory。
完整 fake Release 验收会在 Runtime 运行中执行真实 npm uninstall,并再次验证
systemd、`current`、受管 `memhub`、doctor 和 SQLite 均仍可用。
- `memhub upgrade`:下载并校验新 Runtime;旧版本写过的数据会先迁移到 SDK 数据面(先做完整备份,终端显示进度),再启动新 Runtime。迁移失败时恢复旧版本;迁移完成后旧版本无法打开数据,健康检查失败时保留新版本并提示 `memhub restart` / `memhub doctor`。
- `memhub upgrade`:下载并校验新 Runtime,健康检查失败时恢复旧版本。
- `memhub uninstall`:停止并移除受管服务、Runtime versions/current 和受管 CLI 链接;
默认保留数据。
- `memhub delete-data`:唯一删除 Memory 的命令;Runtime 必须已停止,并要求 `--yes` 与
+5 -9
View File
@@ -6,13 +6,10 @@
2. 记忆沉淀有没有持续运行、产出了多少条;
3. 如果停住,停在导入、模型调用还是最终写入。
默认 `info` 下的官方结构化事件不应写入会话正文、模型输入输出或 API Key;但本地原始日志允许
新增受限标量标签,仍应按私人数据处理,不能直接对外发送。`debug` 会额外启用 HY pipeline
trace,其中可能包含完整 prompt/response 和获授权的对话正文。
排查时只通过 App 的“导出日志”提供文件:导出过程会按安全白名单重新生成诊断事件,不会打包
原始日志。不要直接发送 `diagnostic-*.jsonl`、整个 `~/.one-memory` 数据目录或 debug pipeline
日志;后者如确需分享,必须由开发人员另行确认并脱敏。
默认 `info` 下的结构化诊断日志不包含会话正文、模型输入输出或 API Key。`debug`
会额外启用 HY pipeline trace,其中可能包含完整 prompt/response 和获授权的对话正文。
排查时优先通过 App 的“导出日志”提供文件,不要直接发送整个 `~/.one-memory` 数据目录;
debug 期间产生的 pipeline 日志在分享前必须由开发人员确认并脱敏。
## 1. Log level 设什么
@@ -42,8 +39,7 @@ debug 可能把完整模型 prompt/response 写入本机调试日志,不得长
## 2. 日志在哪里
App 内推荐直接使用“运行诊断 → 导出日志”。开发人员远程指导时,优先使用
`memhub logs`,它会再次过滤不安全字段。原始结构化日志只用于本机现场查看,不得作为附件发送。
Core 日志默认位于:
`memhub logs`,它会再次过滤不安全字段。需要现场查看原始结构化日志时,Core 日志默认位于:
```text
~/.one-memory/logs/diagnostic-core.jsonl
+8 -24
View File
@@ -15,17 +15,9 @@ Desktop -> Runtime -> Core -> HY -> Host 的实际执行链。日志不是用户
可选关联字段为 `trace_id`、`incident_id`、`instance_id`、`operation_id`、`request_id`、
`job_id`、`task_id`、`duration_ms`、状态、稳定错误码、错误类型和清理后的仓库相对栈帧。
`details` 的本地写入合同只约束结构,不维护业务字段白名单:最多 32 个 key;key 必须是
1–64 字符的字母、数字、点、下划线或连字符并以字母开头;value 只接受 `null`、布尔值、
跨 Python/JavaScript 可无损表示的有限数字(整数限于 ±(2^53-1))或不超过 256 字符且不含
控制换行的短字符串,不接受对象和数组。这样新增诊断标签
不需要先修改中央白名单。诊断日志的校验、序列化、排队或落盘发生任何异常时都只能丢弃该条
日志,不能阻断产品主链。
本地写入放宽不代表允许官方代码记录用户内容。对话原文、查询、回答、模型输入输出、请求体、
响应体、token、API Key、secret ref、完整用户路径和异常 message/cause 仍禁止进入诊断事件,
由类型、静态检查和隐私测试共同防止误用;确需保存完整用户输入的调试能力必须使用独立的本地
trace/session 文件,不得混入可导出的诊断日志。
`details` 只接受代码声明的标量白名单。禁止记录对话原文、查询、回答、模型输入输出、
请求体、响应体、token、API Key、secret ref、完整用户路径或异常 message/cause。未知字段和
不合规标识拒绝写入。日志写入失败不能阻断产品主链。
## 关联规则
@@ -42,16 +34,9 @@ trace/session 文件,不得混入可导出的诊断日志。
host-bridge、desktop 和 cli。单文件 4 MiB,保留 3 份轮转;额外清理 14 天以前的文件并将
总量限制在 128 MiB。POSIX 目录为 0700、文件为 0600。
读取器只枚举上述固定文件名的 regular file,不跟随符号链接,不接受任意路径。日常本地查询
遇到损坏行时跳过并计数,不让一行损坏遮蔽其余证据。
导出是独立的安全边界:它重新读取并校验全部候选事件,只把明确列入导出白名单的字段投影为
新的 `diagnostics/events.jsonl`;details 不只检查 key,还按字段约束 value 类型,并拒绝自由文本、
路径和凭证形态的字符串。绝不复制原始 `diagnostic-*.jsonl`、普通运行日志或崩溃报告。
导出包还可以包含固定 schema 的系统摘要和问题时间点,但不接受调用方传入的任意对象展开。
任何事件、marker、安全投影或日志目录读取失败时,导出必须 fail-closed:删除临时文件、不发布
诊断包,更不能回退为打包原始日志。发布使用已设好权限的临时文件原子创建最终目标,已存在的
目标不得覆盖,发布后也不存在再 chmod 失败留下伪成功文件的窗口。
读取器只枚举上述固定文件名的 regular file,不跟随符号链接,不接受任意路径。损坏行跳过
并计数,不让一行损坏遮蔽其余证据。诊断导出是单个脱敏 JSON,包含版本、平台、架构、
非秘密状态和最多 5000 条已验证事件,不包含数据库或配置文件。
## 覆盖与节流
@@ -71,8 +56,7 @@ host-bridge、desktop 和 cli。单文件 4 MiB,保留 3 份轮转;额外清
## 验收
隐私测试必须先向本地日志植入带未知 key 的可检测秘密作为正向对照,再证明本地读取仍保留该
证据、导出安全投影必定剔除它;另需证明损坏行会阻止导出且不会留下伪成功文件。至少用隔离
数据制造 Core 启动失败、HY 失败和宿主连接失败,并仅凭允许导出的字段还原失败组件、时间、
隐私测试必须先植入一条可检测的秘密作为正向对照,再证明真实链路日志和导出均不含该秘密。
至少用隔离数据制造 Core 启动失败、HY 失败和宿主连接失败,并仅凭日志还原失败组件、时间、
关联 ID、稳定错误码和代码位置。自动测试通过只能标记 verified,正式安装 App 的真实体验仍需
单独验收。
@@ -1,155 +1,296 @@
# SDK Data Plane Migration v1
Updated: 2026-10-24.
Updated: 2026-10-21.
This contract moves released MemHub installations from the historical
This contract migrates released MemHub installations from the historical
Evidence/Governance authority to the Hy Memory SDK data plane defined by
[`../ARCHITECTURE.md`](../ARCHITECTURE.md).
It is an offline migration executed inside the App's visible startup loading
phase, before Core accepts product traffic. It is not a dual-read or dual-write
design and not a separate manual upgrade flow for ordinary users.
It is an offline, resumable migration executed inside the App's visible startup
loading phase, before Core accepts product traffic. It is not a permanent
dual-read or dual-write design and is not a separate manual upgrade flow for
ordinary users.
## 1. Why no memory is copied
## 1. Authority before and after cutover
Every released build (`v1.0.0` through build 19) wrote memory through SDK
`add()` / `add_extracted()` first. The SDK stored the raw turn (L1 raw) and the
extracted Chat / Coding Experience / Working Fact memories in its own stores.
The Governance and Evidence tables were a mirror that MemHub read back from the
SDK afterwards.
Before the cutover marker is committed:
The SDK stores are therefore already complete. The migration copies no memory
and no raw turn, and does not compare the mirror with the SDK:
- the mirror's layer labels did not record which Coding store held a row, and
the SDK physically deletes experiences its reconciler retires, so the mirror
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.
## 2. Authority before and after cutover
Before the cutover marker is committed, legacy tables remain the product read
authority; a Runtime that was never upgraded through Desktop or `memhub
install`/`upgrade` still serves them. Desktop does not start Runtime while
migration is incomplete.
- `evidence*` and `governed_*` remain the product read authority;
- SDK stores are inspected and backfilled, but are not merged into live search;
- new imports, writes, queue drains, corrections, deletes and session delivery
are paused.
After the cutover marker is committed:
- SDK stores are the only memory data authority;
- SDK memory/raw/vector/graph/coding stores are the only memory data authority;
- MemHub stores management state and `memory_id` / raw references only;
- legacy Evidence/Governance/retrieval/session-shadow and Matter tables are
- legacy Evidence/Governance/retrieval/session-shadow tables are physically
dropped before the normal App window is revealed;
- startup never falls back to legacy authority.
- startup must not fall back to legacy authority because an SDK read failed.
The surviving `memory_authority_state` singleton, not table existence, selects
the authority.
The surviving `memory_authority_state` singleton, not legacy table existence,
selects the authority.
A fresh home bootstraps the latest management schema with SDK authority
directly and never creates legacy tables.
### Fresh install
## 3. Who runs it, and how
A fresh home does not execute the historical application schema chain. It
bootstraps the latest management-plane schema directly with SDK authority and
never creates Evidence, Governance, legacy retrieval, any Matter, or
session-shadow tables. Historical schema migrations are an input format reader
for old homes only.
One Core command does the whole migration:
## 2. Supported source state
```text
one-memory sdk-migration-status --json # read-only: is there work?
one-memory sdk-migrate --runtime-build <build> --yes --json
```
The migration preflight records and validates:
`sdk-migrate` requires a stopped Runtime. It takes the host capture fence
(hooks return silently instead of writing Evidence while its PID is alive),
then:
- source MemHub Runtime tag/build and management schema version;
- migration Runtime SDK capabilities and the actual on-disk store shape;
- SDK namespace, vector/graph/cache/history/coding providers and locations;
- embedding model identity and dimensions;
- source grants, model profiles, config, queue leases, and pending operations.
1. **bundle**: copies `one-memory.sqlite3`, the SDK data plane and non-secret
configuration into a versioned bundle and verifies it by an isolated
restore; an existing verified bundle of the same run is reused;
2. **prepare**: applies the management schema migrations, discards SDK staging
left by builds through 19 (§5) and makes legacy queue work self-contained
(§4); it needs no model profile and no SDK client;
3. **cutover**: re-verifies the bundle, repeats §4, commits SDK authority;
4. **cleanup**: drops the legacy tables (including the `evidence_search` index).
The migration fails closed when the installed SDK lacks:
With `--json` each step prints `{"state":"running","step":…}` and the last line
is the result: `cutover`, `complete` (only cleanup was left), `not_required`, or
`failed` with a stable `code`. Every step is safe to repeat; a retry resumes the
newest unfinished `sdk-authority-v2-*` run and its bundle, including a run that
build 19 left in `review_required/blocked`.
- complete lifecycle/raw inventory;
- exact raw and exact memory import;
- durable request outcome lookup;
- consistent snapshot/export and restore.
Callers only keep the Runtime stopped, and its service from restarting, around
that command (`runSdkMigration` in `packages/runtime-control`):
MemHub tags `v1.0.0` through `v1.0.4` all use the `governed-v2` namespace and
the same SDK SQL table families (`memory_operations_v2`, `pipeline_logs_v2`,
`system_metrics`, `episode_state`, `memory_history`, `coding_memory_meta`,
`working_fact`, and `coding_trace`). They therefore use one migration selected
by inspected schema/store capabilities, not one branch per historical SDK pin.
The old SDK version is informational only and is not a migration discriminator.
- **Desktop** runs it in the startup loading phase before attaching the UI; a
failure keeps the App closed with the code's sentence.
- **CLI `memhub install` / `memhub upgrade`** (macOS and Linux) run it with the
new Runtime's Core after switching `current` and before starting the service.
A failure before cutover rolls back to the previous version. After cutover an
older Runtime refuses the home, so a later readiness failure keeps the new
version and reports `runtime_not_ready_after_sdk_migration`.
The migration Runtime itself must expose the required APIs. Its exact SDK
version is defined by the release lockfiles; old data is not rejected merely
because an earlier App wrote it.
After cutover, History catch-up scans forward from each source checkpoint and
recovers turns skipped while capture was fenced.
## 3. Legacy-state classification
## 4. Queue work
While Core and Runtime writers are stopped, build four inventories:
Queue rows remain MemHub management state. Before the legacy tables disappear:
- `E`: Evidence rows and revisions;
- `G`: current Governance rows, statuses, supersession links, and latest content;
- `M`: task-output mappings from legacy `memory_id` to SDK `hy_memory_id`;
- `S`: complete SDK inventory, including raw and non-active lifecycle records.
- each `memory_write_jobs` row that only names Evidence IDs gets those records
frozen into `payload_json`;
- unstarted legacy `memory_tasks` become
`memory_write_jobs(idempotency_key="legacy-task:<task_id>")`;
- Evidence waiting in `evidence_pending_memory_dispatch` (its source cursor has
already passed it) becomes one write job per conversation.
Every item must end in exactly one class:
Legacy work that cannot continue does not block cutover; its counts are
reported in the `sdk-migrate` / cutover receipt as `abandoned`:
| Class | Meaning | Required action |
| --- | --- | --- |
| matched | Governance/mapping and SDK identity, content, owner and lifecycle agree | verify only |
| mapped mismatch | legacy ID maps to a different existing SDK ID | keep SDK ID; record alias |
| Governance-only | current legacy memory has no SDK entity | exact import without extraction |
| SDK-only | SDK entity has no legacy mirror | retain and adopt |
| divergent | both exist but content, owner or lifecycle differs | reconcile explicit user intent or quarantine |
| raw-only | Evidence has no SDK raw receipt | exact raw import |
| unknown outcome | a task may have reached SDK without a confirmed result | SDK outcome inquiry; never blind retry |
- a legacy task that may already have reached the SDK (`unknown`, or `running`
with an SDK invocation) is not replayed: the SDK already holds whatever it
committed;
- pending legacy user deletions are not carried over.
An unclassified item blocks cutover.
Surviving `memory_write_jobs` with `hy_invoked=1` keep their normal
post-cutover handling through SDK `get_write_outcome(request_id)`.
## 4. User edits, deletion and supersession
## 5. Staging left by earlier builds
The current product has no MemHub revision history or Undo.
Builds through 19 copied the SDK root to `<home>/.sdk-migration/` and swapped
directories at cutover. Before SDK authority, a leftover
`hy-memory/.<namespace>.pre-migration-*` directory is the original store and is
restored; after SDK authority it is stale and removed. More than one such
directory before authority is ambiguous and fails closed
(`sdk_migration_swap_state_ambiguous`).
- The latest non-deleted user correction is imported as the current SDK body.
- Historical `governed_memory_versions` remain in the compatibility backup, but
are not recreated as product-visible SDK versions.
- A legacy user deletion is applied as an SDK physical delete. Deleted
Governance tombstones are not imported and no deleted-list UI survives.
- SDK supersession remains an SDK lifecycle relationship. Existing SDK chains
are retained. Governance links are used only to repair a missing SDK chain
during exact import.
- Pending correction/delete records are settled before ordinary backfill.
## 6. Failure and rollback
`update(memory_id, content)` modifies the same SDK ID. There is no product CAS,
revision number, or restore operation.
Any failure leaves the marker absent and reports a stable code to the startup
dialog (for example `sdk_migration_disk_space_insufficient`,
`sdk_migration_bundle_copy_failed`, `sdk_migration_preflight_failed`,
`sdk_migration_cutover_failed`). The next startup retries.
## 5. Evidence to SDK raw
After cutover but before any SDK-only write, rollback restores the bundle
together with the matching old Runtime. After an SDK-only write, automatic
binary-only rollback is forbidden; restoring the pre-cutover bundle loses the
post-cutover writes and must be explicit. An old Runtime must fail closed on a
post-cutover home. Changing only the Runtime `current` symlink is not data
rollback.
Evidence must not be replayed through ordinary `add()`: doing so would extract
and reconcile again, changing semantic memories.
## 7. Acceptance
The SDK exact-raw import preserves:
- source, conversation/session, workspace, role and author;
- original event time and stable source ordering;
- verbatim content and source identity;
- a deterministic migration key.
The operation is idempotent and returns an SDK raw ID. Evidence revisions and
repairs are preserved in the migration bundle for audit; only the current
accepted raw record becomes the normal product read.
## 6. Full migration bundle
A successful preflight creates one versioned bundle containing:
- an online backup of `one-memory.sqlite3`;
- an SDK-consistent export/snapshot of raw, vector, graph, cache, history and
coding stores;
- `config.json` and non-secret configuration needed to locate those stores;
- Runtime, SDK, schema, namespace, backend and embedding fingerprints;
- object/file hashes, row counts, object counts and creation time;
- migration ledger version and bundle ID.
Secrets are not copied into a plaintext manifest. The bundle records only
secret references and the secure-store identity required for restore.
The bundle is published only after an isolated restore verifies hashes, counts
and SDK reads. A remote backend without a supported snapshot/export mechanism
blocks migration; SQLite-only backup is not reported as success.
## 7. Authority state and migration ledger
The MemHub management database owns:
- `memory_authority_state`: the surviving singleton authority and migration ID;
- `sdk_data_plane_migrations`: one run, source/target fingerprints, bundle,
phase, cutover state and reconciliation digest;
- `sdk_data_plane_migration_items`: source identity/revision/hash, target SDK
identity/hash, deterministic operation key, state, attempts and error code;
- `memory_id_aliases`: temporary historical ID to canonical SDK ID mapping for
call-events, write-job receipts, and links;
- `raw_id_aliases`: legacy Evidence ID/item index to canonical SDK raw ID.
- `sdk_data_plane_cutover`: a temporary detailed cutover receipt containing
SDK/backend/bundle/reconciliation fingerprints.
Progress is committed only after SDK verification. Retrying an item uses the
same deterministic SDK operation key and must not create another memory.
Before cleanup, every surviving reference is rewritten to its canonical SDK
ID. Alias, item-ledger, and detailed cutover tables are then dropped; the
authority singleton and verified bundle remain.
Backfill never mutates the live SDK root. Startup first copies the complete
local SDK data plane, including SDK-only records, to a same-filesystem staging
home. Exact legacy writes and reconciliation run there. Cutover renames the
live root to a protective path, renames staging into place, and only then
commits SDK authority. A failed commit restores the previous root; restart
recovery resolves any interrupted rename before opening SDK stores.
## 8. Queue and unknown outcomes
Queue rows and leases remain MemHub management state.
- A pending job has not called the SDK and therefore has no SDK raw or memory
ID yet.
- Its worker calls SDK `add()` once. The SDK owns raw persistence plus
extraction/reconciliation; a successful job stores only the returned raw ID,
canonical memory IDs and terminal outcome.
- Legacy raw and semantic backfill is an App-owned privileged offline
operation while Runtime writers are stopped. It writes SDK stores directly;
no migration-only import/get/list method is added to public
`HyMemoryClient`, and this is not the first half of a two-stage normal write
path.
- `hy_invoked=0` may be safely retried after normal lease recovery.
- `memory_write_jobs.hy_invoked=1`, `state=unknown`, or any non-terminal SDK
outcome must use `get_write_outcome(request_id)`.
- `running`/`unknown` is not absence and must not be replayed.
- confirmed SDK IDs become write-job outcomes/receipts; Governance publication is not part of
completion after cutover.
## 9. Call events and session delivery
- Matter was an unpublished placeholder. Its rows are not migrated and every
`matter*` table is deleted during cleanup.
- Before cleanup, historical IDs resolve through `memory_id_aliases`; surviving
records are rewritten to canonical SDK IDs before that table is dropped.
- `call_events` remains metadata-only; reference titles/content resolve through
SDK APIs.
- Evidence-based session inbox/outbox is disabled at cutover unless replaced by
a management-event or SDK-event contract. It must not send shadow memory
bodies.
## 10. Cutover
The packaged App owns orchestration. While the startup placeholder is visible,
it starts no product writer, creates or verifies the full bundle, resumes the
durable migration ledger, reconciles SDK state, and only then commits cutover.
The normal App window is revealed after this phase. A retry resumes the same
migration and deterministic operation keys; it does not create a second run.
If migration cannot safely complete, the marker remains absent and startup must
report the durable blocked/deferred reason. It must not silently continue into
an SDK/legacy dual-write mode. A missing model profile may defer migration while
legacy authority is still intact so the user can configure the App; it never
authorizes a partial cutover.
Cutover requires:
- no unclassified or quarantined item;
- complete raw coverage;
- no unsafe unknown task;
- pending corrections and deletions settled;
- call-event references resolvable;
- exact counts and hashes reconciled;
- deterministic full-ID get verification;
- a verified full migration bundle.
The final transaction writes the detailed cutover receipt and changes
`memory_authority_state` to SDK authority. Cleanup may remove the detailed
receipt after bundle verification; the authority singleton remains.
After that transaction, legacy memory/raw writes are rejected.
## 11. Rollback and version skew
Before cutover, a failed run resumes from the ledger or restores the full
bundle.
After cutover but before any SDK-only write, rollback restores the full bundle
and matching old Runtime together.
After an SDK-only write:
- automatic binary-only rollback is forbidden;
- operators either forward-fix, or explicitly restore the pre-cutover bundle
while acknowledging loss of post-cutover writes;
- an old Runtime must fail closed on the post-cutover marker/schema.
Changing only the Runtime `current` symlink is not data rollback.
## 12. Legacy cleanup
Legacy tables are not a compatibility runtime. After every memory/raw
reference resolves, the verified bundle has passed restore validation, and the
SDK cutover marker is durable, the offline startup transaction drops:
- every `evidence*` and `governed_*` table;
- legacy retrieval/index tables;
- every `matter*` placeholder table;
- legacy deletion/index-sync/task-output and session-shadow tables;
- migration-only aliases and ledgers after their canonical references have
been materialized into the surviving management tables.
Before that DROP, every surviving `memory_write_jobs` row is made
self-contained by freezing its source records into `payload_json`. Unstarted
legacy `memory_tasks` rows are converted to deterministic
`memory_write_jobs(idempotency_key="legacy-task:<task_id>")`; possibly-sent
task rows remain a cutover blocker until SDK request-ID reconciliation proves
their outcome.
The verified bundle is the audit and rollback artifact; keeping duplicate
plaintext tables in the live database is not. A cleanup failure keeps startup
blocked and never exposes a half-cleaned schema.
## 13. Acceptance matrix
The release gate covers:
- a fresh install with no legacy table names;
- a legacy home reaching cutover with its SDK stores byte-for-byte unchanged;
- a home left blocked by build 19 (quarantined ledger and staging copy)
completing on the next startup;
- CLI upgrade migrating before restart, rolling back on a failed migration,
and keeping the new version when a migrated home is slow to come up;
- host capture fenced for the whole command;
- pending job freezing, legacy task and pending dispatch conversion;
- invoked/unknown legacy tasks and pending deletions not blocking cutover;
- interrupted directory swap recovery before and after authority;
- bundle creation, reuse and restore;
- physical absence of every legacy table after cutover.
- a clean install with no legacy rows;
- a clean install whose schema contains no legacy table names;
- Governance-only, SDK-only, mapped-ID and divergent fixtures;
- Evidence/raw-only content;
- pending correction and physical deletion;
- unknown SDK outcome and crash after SDK commit;
- interruption before/after each ledger transition;
- full bundle restore and hash/count mismatch;
- call-event alias preservation;
- old Runtime against a post-cutover home;
- successful migration followed by physical absence of every legacy table;
- local and supported remote backend snapshot behavior.
No narrow unit test may be used as evidence that the full migration is safe.
+2 -2
View File
@@ -1,8 +1,8 @@
{
"name": "one-memory",
"version": "1.0.5",
"version": "1.0.4",
"release": {
"build_number": 20,
"build_number": 13,
"macos_min_version": "13.0",
"app_update": "manual",
"runtime_manifest_schema": 1,
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/contracts",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"types": "./src/index.ts",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/host-bridge",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"exports": {
@@ -192,12 +192,16 @@ describe("RecallBroker", () => {
level: "error",
component: "host.broker",
outcome: "failed",
error_code: "connection_control_unavailable",
error_type: "Error",
details: { action: "host" },
details: expect.objectContaining({
action: "host",
code: "connection_control_unavailable",
error_type: "Error",
}),
});
expect(exception?.details).not.toHaveProperty("message");
expect(exception?.details).not.toHaveProperty("stack_origin");
expect(typeof exception?.details).toBe("object");
const details = exception?.details as Record<string, unknown>;
expect(typeof details.stack_origin).toBe("string");
expect(details.stack_origin).toContain("recall-broker");
});
it.each(["recall", "ask"])(
"answers %s without waiting for a slow diagnostic write",
@@ -702,10 +702,20 @@ export class RecallBroker {
event: "connection_control.exception",
outcome: "failed",
instance_id: this.options.instanceId,
error_code: code,
error_type:
error instanceof Error ? error.constructor.name : typeof error,
details: { action: action ?? "invalid" },
details: {
action: action ?? "invalid",
code,
error_type:
error instanceof Error ? error.constructor.name : typeof error,
message:
error instanceof Error
? error.message.slice(0, 256)
: String(error).slice(0, 256),
stack_origin:
error instanceof Error && error.stack
? (error.stack.split("\n").at(1)?.trim() ?? "").slice(0, 256)
: null,
},
});
}
const fingerprint = `error:${code}`;
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/i18n",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"types": "./src/index.ts",
@@ -1933,8 +1933,6 @@ const baseEnglishMessages = {
languageLoadError: "Could not load the language setting. Please retry.",
nativeErrorTitle: "MemHub could not start",
startupPreparing: "Starting and preparing the memory service…",
startupUpgradeHint:
"The first launch or a memory-data upgrade can take a few minutes. MemHub opens automatically when it is done.",
} as const;
export const englishMessages = {
@@ -52,8 +52,6 @@ const languageChineseMessages = {
languageLoadError: "无法加载语言设置,请重试。",
nativeErrorTitle: "MemHub 无法启动",
startupPreparing: "正在启动,正在准备记忆服务…",
startupUpgradeHint:
"首次启动或升级记忆数据可能需要几分钟,完成后会自动进入应用。",
} as const;
const languageSpanishMessages: Record<
@@ -73,8 +71,6 @@ const languageSpanishMessages: Record<
languageLoadError: "No se pudo cargar el idioma. Inténtalo de nuevo.",
nativeErrorTitle: "MemHub no pudo iniciarse",
startupPreparing: "Iniciando y preparando el servicio de memorias…",
startupUpgradeHint:
"El primer inicio o la actualización de tus memorias puede tardar unos minutos. La aplicación se abrirá automáticamente al terminar.",
};
type Catalog = {
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/operations",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"types": "./src/index.ts",
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@one-memory/runtime-control",
"version": "1.0.5",
"version": "1.0.4",
"private": true,
"type": "module",
"bin": {
@@ -680,7 +680,7 @@ export class CoreSupervisor {
outcome: "failed",
instance_id: this.activeInstanceId,
error_code: "core_restart_circuit_open",
details: { restart_streak: this.restartStreak },
details: { streak: this.restartStreak },
});
return;
}
@@ -765,7 +765,7 @@ export class CoreSupervisor {
outcome: "failed",
instance_id: this.activeInstanceId,
error_code: "core_restart_exhausted",
details: { attempts: budget, restart_streak: this.restartStreak },
details: { attempts: budget, streak: this.restartStreak },
});
await this.options.diagnostics?.log({
level: "error",
@@ -774,7 +774,7 @@ export class CoreSupervisor {
outcome: "failed",
instance_id: this.activeInstanceId,
error_code: "core_restart_circuit_open",
details: { restart_streak: this.restartStreak },
details: { streak: this.restartStreak },
});
}
} finally {
@@ -1,5 +1,4 @@
import {
chmod,
lstat,
mkdir,
mkdtemp,
@@ -22,10 +21,8 @@ import {
diagnosticLogFiles,
exportDiagnosticBundle,
exportDiagnosticZip,
projectDiagnosticEventForSafeOutput,
pruneDiagnosticLogs,
readDiagnosticEvents,
recordDiagnosticMarker,
writeDiagnosticEventSync,
} from "./diagnostic-log.js";
@@ -69,30 +66,6 @@ describe("diagnostic log", () => {
});
});
it("keeps the synchronous writer fail-open for invalid input", async () => {
const dataDir = await root();
expect(() =>
writeDiagnosticEventSync(
{
dataDir,
service: "desktop",
sessionId: "55555555-5555-4555-8555-555555555555",
},
{
level: "error",
component: "desktop.lifecycle",
event: "desktop.early_start_failed",
outcome: "failed",
details: { nested: ["private"] } as never,
},
),
).not.toThrow();
expect(await readDiagnosticEvents(dataDir)).toEqual({
events: [],
invalid_lines: 0,
});
});
it("writes structured, private events and reads them by correlation identity", async () => {
const dataDir = await root();
const session = "11111111-1111-4111-8111-111111111111";
@@ -116,8 +89,6 @@ describe("diagnostic log", () => {
open_file_count: 7,
memory_bytes: 4096,
thread_count: 4,
"cache.hit-rate": 0.75,
custom_label: "warm",
},
});
const result = await readDiagnosticEvents(dataDir, { traceId: trace });
@@ -130,8 +101,6 @@ describe("diagnostic log", () => {
event: "runtime.ready",
details: expect.objectContaining({
open_file_count: 7,
"cache.hit-rate": 0.75,
custom_label: "warm",
}),
});
const file = (await diagnosticLogFiles(dataDir))[0];
@@ -215,7 +184,7 @@ describe("diagnostic log", () => {
}
});
it("strictly rejects complex detail values, dangerous keys and absolute stacks", () => {
it("rejects arbitrary detail fields and absolute stack paths", () => {
const base = {
timestamp: new Date().toISOString(),
level: "error" as const,
@@ -226,142 +195,11 @@ describe("diagnostic log", () => {
session_id: "11111111-1111-4111-8111-111111111111",
};
expect(() =>
diagnosticEvent({
...base,
details: { nested: { secret: "private" } } as never,
}),
).toThrow("diagnostic_detail_value_invalid");
expect(() =>
diagnosticEvent({ ...base, details: { constructor: "private" } }),
diagnosticEvent({ ...base, details: { prompt: "private" } }),
).toThrow("diagnostic_detail_key_invalid");
expect(() =>
diagnosticEvent({ ...base, details: { "9invalid": "private" } }),
).toThrow("diagnostic_detail_key_invalid");
expect(() =>
diagnosticEvent({
...base,
details: Object.fromEntries(
Array.from({ length: 33 }, (_, index) => [`key_${index}`, index]),
),
}),
).toThrow("diagnostic_detail_count_invalid");
expect(() =>
diagnosticEvent({ ...base, details: { message: "line1\nline2" } }),
).toThrow("diagnostic_text_invalid");
expect(() =>
diagnosticEvent({ ...base, details: { metric: Number.NaN } }),
).toThrow("diagnostic_detail_value_invalid");
expect(() =>
diagnosticEvent({
...base,
details: { count: Number.MAX_SAFE_INTEGER + 1 },
}),
).toThrow("diagnostic_detail_value_invalid");
expect(() =>
diagnosticEvent({ ...base, details: { list: ["private"] } as never }),
).toThrow("diagnostic_detail_value_invalid");
expect(() =>
diagnosticEvent({ ...base, safe_stack: ["/Users/me/private.py:1"] }),
).toThrow("diagnostic_stack_invalid");
expect(() =>
diagnosticEvent({
...base,
safe_stack: ["packages/runtime/sk-live-secret-token.ts:1:1"],
}),
).toThrow("diagnostic_stack_invalid");
for (const line of [
"arbitrary stack text",
"thread=worker|threading.py:12:wait",
"FILE:///Users/me/private.py:1:call",
"one_memory/worker.py:9:call /Users/private",
"one_memory/GOCSPX-privateCredential.py:9:call",
"packages/ASIA1234567890ABCDEF/file.ts:1:1",
"packages/Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8/file.ts:1:1",
]) {
expect(() => diagnosticEvent({ ...base, safe_stack: [line] })).toThrow(
"diagnostic_stack_invalid",
);
}
expect(() =>
diagnosticEvent({
...base,
safe_stack: Array.from(
{ length: 17 },
(_, index) => `one_memory/worker.py:${index + 1}:call`,
),
}),
).toThrow("diagnostic_stack_invalid");
const accepted = diagnosticEvent({
...base,
safe_stack: [
"apps/desktop/src/main.ts:80:1",
"one_memory/diagnostic_log.py:895:_safe_stack_line",
"threading.py:1045:_bootstrap_inner",
"thread|threading.py:1045:_bootstrap_inner > one_memory/worker.py:9:call",
],
});
expect(accepted.safe_stack).toHaveLength(4);
const projected = projectDiagnosticEventForSafeOutput({
...accepted,
safe_stack: [
"free text",
"thread=worker|one_memory/worker.py:9:call",
"packages/runtime-control/src/diagnostic-log.ts:42:7",
"thread|threading.py:1045:_bootstrap_inner > one_memory/worker.py:9:call",
],
});
expect(projected.safe_stack).toEqual([
"packages/runtime-control/src/diagnostic-log.ts:42:7",
"thread|threading.py:1045:_bootstrap_inner > one_memory/worker.py:9:call",
]);
const conservative = projectDiagnosticEventForSafeOutput(
diagnosticEvent({
...base,
details: {
source_id: "cursor",
reason: "abcdef1234567890abcdef1234567890abcdef12",
status: "11111111-1111-4111-8111-111111111111",
},
}),
);
expect(conservative.details).toEqual({
source_id: "cursor",
reason: "abcdef1234567890abcdef1234567890abcdef12",
status: "11111111-1111-4111-8111-111111111111",
});
});
it("drops malformed logger events without interrupting later writes", async () => {
const dataDir = await root();
const logger = new DiagnosticLogger({
dataDir,
service: "desktop",
sessionId: "11111111-1111-4111-8111-111111111111",
});
await expect(
logger.log({
level: "info",
component: "desktop.sdk_migration",
event: "sdk_migration.startup",
outcome: "succeeded",
details: { nested: { secret: "private" } } as never,
}),
).resolves.toBeUndefined();
await logger.log({
level: "info",
component: "desktop.sdk_migration",
event: "sdk_migration.startup",
outcome: "succeeded",
details: { arbitrary_state: "cutover" },
});
const result = await readDiagnosticEvents(dataDir);
expect(result.invalid_lines).toBe(0);
expect(result.events).toHaveLength(1);
expect(result.events[0].details).toEqual({ arbitrary_state: "cutover" });
});
it("keeps code locations but strips exception messages and absolute prefixes", () => {
@@ -377,162 +215,29 @@ describe("diagnostic log", () => {
expect(JSON.stringify(safe)).not.toContain("/Users/me");
});
it("exports events through an independent detail whitelist", async () => {
it("marks incidents and exports only validated events", async () => {
const dataDir = await root();
const logger = new DiagnosticLogger({ dataDir, service: "desktop" });
const incident = await logger.markIncident({
component: "desktop.diagnostics",
status: "degraded",
});
await logger.log({
level: "info",
component: "desktop.diagnostics",
event: "diagnostics.checked",
outcome: "succeeded",
details: {
count: 1,
reason: "eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ1c2VyIn0.signaturevalue",
status: "ASIA1234567890ABCDEF",
source_id: "GOCSPX-clientSecret_A1",
mode: "Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8",
action: "AIzaSyExampleCredential1234567890",
platform: "x_ASIA1234567890ABCDEF",
phase: "x_GOCSPX-clientSecret_A1",
local_only_secret: "must-not-export",
},
});
await logger.log({
level: "error",
component: "credential.private",
event: "token.failed",
outcome: "ghp_private_token",
trace_id: "ghp_private_token",
instance_id: "ASIA1234567890ABCDEF",
request_id: "GOCSPX-clientSecret_A1",
task_id: "Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8",
error_code: "secret.token",
});
const target = path.join(await root(), "bundle.json");
const result = await exportDiagnosticBundle({
dataDir,
target,
productVersion: "1.2.3",
status: { status: "degraded", local_only_secret: "must-not-export" },
});
expect(result).toEqual({ event_count: 3, invalid_lines: 0 });
const payload = JSON.parse(await readFile(target, "utf8"));
expect(payload.events[0].incident_id).toBe(incident.incident_id);
expect(payload.events[1].details).toEqual({ count: 1 });
expect(payload.events[2]).toMatchObject({
component: "diagnostic.redacted",
event: "diagnostic.redacted",
outcome: "redacted",
});
expect(payload.events[2].trace_id).toBeUndefined();
expect(payload.events[2].instance_id).toBeUndefined();
expect(payload.events[2].request_id).toBeUndefined();
expect(payload.events[2].task_id).toBeUndefined();
expect(payload.events[2].error_code).toBeUndefined();
expect(JSON.stringify(payload)).not.toContain("local_only_secret");
expect(JSON.stringify(payload)).not.toContain("must-not-export");
expect(JSON.stringify(payload)).not.toContain("eyJhbGciOiJIUzI1NiJ9");
expect(JSON.stringify(payload)).not.toContain("ASIA1234567890ABCDEF");
expect(JSON.stringify(payload)).not.toContain("GOCSPX-clientSecret_A1");
expect(JSON.stringify(payload)).not.toContain("x_ASIA1234567890ABCDEF");
expect(JSON.stringify(payload)).not.toContain("x_GOCSPX-clientSecret_A1");
expect(JSON.stringify(payload)).not.toContain("Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8");
expect(JSON.stringify(payload)).not.toContain(
"AIzaSyExampleCredential1234567890",
);
expect(JSON.stringify(payload)).not.toContain("ghp_private_token");
});
it("fails closed when exported build identity has a credential shape", async () => {
const dataDir = await root();
const outputDir = await root();
await expect(
exportDiagnosticBundle({
dataDir,
target: path.join(outputDir, "bundle.json"),
productVersion: "xoxb-123456-abcdef",
}),
).rejects.toThrow("diagnostic_product_version_invalid");
await expect(
exportDiagnosticZip({
dataDir,
target: path.join(outputDir, "commit.zip"),
productVersion: "1.2.3",
commit: "ghp_private_token",
}),
).rejects.toThrow("diagnostic_commit_invalid");
await expect(
exportDiagnosticZip({
dataDir,
target: path.join(outputDir, "sha.zip"),
productVersion: "1.2.3",
coreSha256: "AKIDExampleSecretId1234567890",
}),
).rejects.toThrow("diagnostic_sha256_invalid");
expect(await readdir(outputDir)).toEqual([]);
});
it("fails closed without publishing a JSON bundle when any line is invalid", async () => {
const dataDir = await root();
await mkdir(path.join(dataDir, "logs"), { recursive: true });
await writeFile(
path.join(dataDir, "logs", "diagnostic-core.jsonl"),
"not-json\n",
{ mode: 0o600 },
);
const outputDir = await root();
const target = path.join(outputDir, "bundle.json");
await expect(
exportDiagnosticBundle({
dataDir,
target,
productVersion: "1.2.3",
}),
).rejects.toThrow("diagnostic_export_invalid_lines");
expect(await readdir(outputDir)).toEqual([]);
});
it("fails closed when the diagnostic directory cannot be read", async () => {
if (process.platform === "win32") return;
const dataDir = await root();
const logs = path.join(dataDir, "logs");
await mkdir(logs, { recursive: true, mode: 0o700 });
await writeFile(path.join(logs, "diagnostic-core.jsonl"), "{}\n");
const outputDir = await root();
const target = path.join(outputDir, "bundle.json");
await chmod(logs, 0o000);
try {
await expect(
exportDiagnosticBundle({
dataDir,
target,
productVersion: "1.2.3",
}),
).rejects.toMatchObject({ code: "EACCES" });
} finally {
await chmod(logs, 0o700);
}
expect(await readdir(outputDir)).toEqual([]);
});
it("never overwrites an existing export target or leaves temporary files", async () => {
const dataDir = await root();
const outputDir = await root();
const target = path.join(outputDir, "bundle.json");
await writeFile(target, "existing-artifact", { mode: 0o600 });
await expect(
exportDiagnosticBundle({
dataDir,
target,
productVersion: "1.2.3",
}),
).rejects.toMatchObject({ code: "EEXIST" });
expect(await readFile(target, "utf8")).toBe("existing-artifact");
expect(await readdir(outputDir)).toEqual(["bundle.json"]);
const target = path.join(await root(), "bundle.json");
const result = await exportDiagnosticBundle({
dataDir,
target,
productVersion: "1.2.3",
status: { status: "degraded" },
});
expect(result).toEqual({ event_count: 1, invalid_lines: 1 });
const payload = JSON.parse(await readFile(target, "utf8"));
expect(payload.events[0].incident_id).toBe(incident.incident_id);
expect(JSON.stringify(payload)).not.toContain(dataDir);
});
it("preserves microsecond ordering across service log files", async () => {
@@ -575,7 +280,7 @@ describe("diagnostic log", () => {
expect(result.events[1].timestamp).toBe("2026-10-14T00:00:00.123999Z");
});
it("exports only reprojected events, a strict summary and README", async () => {
it("exports a private seven-day zip with summary, logs, crashes and markers", async () => {
const dataDir = await root();
const outputDir = await root();
const crashDir = await root();
@@ -621,178 +326,39 @@ describe("diagnostic log", () => {
markers: [
{
timestamp: "2026-10-05T01:30:00.000Z",
description:
"蓝莓没出来 /Users/private/query.txt ASIA1234567890ABCDEF GOCSPX-clientSecret_A1 Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8 token=ghp_private_token xoxb-123456-abcdef AKIDExampleSecretId1234567890 AIzaSyExampleCredential1234567890 eyJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ1c2VyIn0.signaturevalue",
description: "蓝莓没出来",
},
],
summary: {
sources: [{ source_id: "workbuddy", imported_messages: 12 }],
summary_secret: conversation,
background: {
status: "exported",
runtime_state: "ready",
core_state: "healthy",
broker_state: "ready",
instance_id: "instance-1",
secret: conversation,
},
memory_tasks: { unknown: 1, last_error_codes: { model_failed: 1 } },
},
});
expect(result).toEqual({ filename: "MemHub.zip", file_count: 3 });
expect(result.filename).toBe("MemHub.zip");
const files = unzipSync(await readFile(target));
expect(Object.keys(files).sort()).toEqual([
"README.txt",
"diagnostics/events.jsonl",
"summary.json",
]);
const eventFile = files["diagnostics/events.jsonl"];
const summaryFile = files["summary.json"];
expect(Object.keys(files)).toEqual(
expect.arrayContaining([
"README.txt",
"summary.json",
"logs/diagnostic-core.jsonl",
"logs/core.log",
"crashes/MemHub-2026-10-05.ips",
]),
);
const summary = files["summary.json"];
const readme = files["README.txt"];
if (!eventFile || !summaryFile || !readme)
throw new Error("missing support bundle metadata");
const exportedEvent = JSON.parse(strFromU8(eventFile).trim());
expect(exportedEvent.details).toEqual({ count: 1 });
const summary = JSON.parse(strFromU8(summaryFile));
expect(summary.background).toEqual({
status: "exported",
runtime_state: "ready",
core_state: "healthy",
broker_state: "ready",
instance_id: "instance-1",
});
expect(summary.sources).toBeUndefined();
expect(summary.summary_secret).toBeUndefined();
expect(summary.problem_markers).toEqual([
{
timestamp: "2026-10-05T01:30:00.000Z",
description: "user_marked_incident",
},
]);
expect(strFromU8(readme)).toContain("user_marked_incident");
expect(strFromU8(readme)).not.toContain("蓝莓没出来");
if (!summary || !readme) throw new Error("missing support bundle metadata");
expect(strFromU8(summary)).toContain("model_failed");
expect(strFromU8(readme)).toContain("蓝莓没出来");
const all = Object.values(files)
.map((value) => strFromU8(value))
.join("\n");
expect(all).not.toContain(secret);
expect(all).not.toContain("ghp_private_token");
expect(all).not.toContain("xoxb-123456-abcdef");
expect(all).not.toContain("AKIDExampleSecretId1234567890");
expect(all).not.toContain("ASIA1234567890ABCDEF");
expect(all).not.toContain("GOCSPX-clientSecret_A1");
expect(all).not.toContain("Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8");
expect(all).not.toContain("AIzaSyExampleCredential1234567890");
expect(all).not.toContain("eyJhbGciOiJIUzI1NiJ9");
expect(all).not.toContain(conversation);
expect(all).not.toContain(home);
expect(all).not.toContain("/Users/");
});
it("fails closed without publishing a ZIP when a diagnostic line is invalid", async () => {
const dataDir = await root();
await mkdir(path.join(dataDir, "logs"), { recursive: true });
await writeFile(
path.join(dataDir, "logs", "diagnostic-core.jsonl"),
"not-json\n",
);
const outputDir = await root();
const target = path.join(outputDir, "MemHub.zip");
await expect(
exportDiagnosticZip({
dataDir,
target,
productVersion: "3800",
}),
).rejects.toThrow("diagnostic_export_invalid_lines");
expect(await readdir(outputDir)).toEqual([]);
});
it("stores and exports only the fixed marker description", async () => {
const dataDir = await root();
const markerFile = path.join(dataDir, "logs", "problem-markers.jsonl");
await recordDiagnosticMarker(
dataDir,
{
timestamp: "2026-10-05T01:00:00.000Z",
description:
"蓝莓 /Users/private/query.txt ASIA1234567890ABCDEF GOCSPX-clientSecret_A1 Aa1Bb2Cc3Dd4Ee5Ff6Gg7Hh8",
},
"/Users/private",
);
expect(await readFile(markerFile, "utf8")).toBe(
`${JSON.stringify({ timestamp: "2026-10-05T01:00:00.000Z", description: "user_marked_incident" })}\n`,
);
await writeFile(
markerFile,
`${await readFile(markerFile, "utf8")}${JSON.stringify({ timestamp: "2026-10-05T01:30:00.000Z", description: "legacy arbitrary text" })}\n`,
{ mode: 0o600 },
);
const target = path.join(await root(), "markers.zip");
await exportDiagnosticZip({
dataDir,
target,
productVersion: "3800",
now: new Date("2026-10-05T02:00:00.000Z"),
});
const files = unzipSync(await readFile(target));
const summaryFile = files["summary.json"];
if (!summaryFile) throw new Error("missing summary.json");
const payload = JSON.stringify(JSON.parse(strFromU8(summaryFile)));
expect(payload.match(/user_marked_incident/g)).toHaveLength(2);
expect(payload).not.toContain("legacy arbitrary text");
expect(payload).not.toContain("蓝莓");
expect(payload).not.toContain("/Users/");
});
it("never follows a problem marker symlink for reads or writes", async () => {
const dataDir = await root();
const logs = path.join(dataDir, "logs");
await mkdir(logs, { recursive: true });
const outside = path.join(await root(), "outside-marker");
await writeFile(outside, "private");
await symlink(outside, path.join(logs, "problem-markers.jsonl"));
await expect(
recordDiagnosticMarker(dataDir, {
timestamp: "2026-10-05T01:00:00.000Z",
description: "anything",
}),
).rejects.toThrow("diagnostic_marker_file_unsafe");
await expect(
exportDiagnosticZip({
dataDir,
target: path.join(await root(), "markers.zip"),
productVersion: "3800",
}),
).rejects.toThrow("diagnostic_marker_file_unsafe");
expect(await readFile(outside, "utf8")).toBe("private");
});
it("rejects invalid explicit and persisted markers without publishing a ZIP", async () => {
const dataDir = await root();
const logs = path.join(dataDir, "logs");
await mkdir(logs, { recursive: true });
const outputDir = await root();
const explicitTarget = path.join(outputDir, "explicit.zip");
await expect(
exportDiagnosticZip({
dataDir,
target: explicitTarget,
productVersion: "3800",
markers: [{ timestamp: "invalid", description: "marker" }],
}),
).rejects.toThrow("diagnostic_marker_invalid");
await writeFile(path.join(logs, "problem-markers.jsonl"), "not-json\n");
const persistedTarget = path.join(outputDir, "persisted.zip");
await expect(
exportDiagnosticZip({
dataDir,
target: persistedTarget,
productVersion: "3800",
}),
).rejects.toThrow("diagnostic_marker_invalid");
expect(await readdir(outputDir)).toEqual([]);
expect(all).toContain("~/MemHub");
});
it("an incident retrieves its surrounding session rather than only its marker", async () => {
File diff suppressed because it is too large. Load diff
@@ -59,7 +59,6 @@ export {
diagnosticZipFilename,
exportDiagnosticBundle,
exportDiagnosticZip,
projectDiagnosticEventForSafeOutput,
pruneDiagnosticLogs,
readDiagnosticEvents,
recordDiagnosticMarker,
@@ -174,13 +173,6 @@ export {
RuntimeSupervisor,
type RuntimeSupervisorOptions,
} from "./runtime-supervisor.js";
export {
type CoreCommand,
runSdkMigration,
type SdkMigrationCommandResult,
type SdkMigrationOptions,
type SdkMigrationResult,
} from "./sdk-migration.js";
export {
installUserService,
parseDisabledOverride,
@@ -1,203 +0,0 @@
import { mkdtemp, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { expect, it, vi } from "vitest";
import { runSdkMigration } from "./sdk-migration.js";
const core = { command: "/core/one-memory", args: [] };
const noFence = async <T>(
_dataDir: string,
action: () => Promise<T>,
): Promise<T> => action();
function statusThen(
migrate: { exitCode: number; document: Record<string, unknown> },
status: Record<string, unknown> = { authority: "legacy", state: "required" },
) {
return vi.fn(async (_core, args: string[]) =>
args[0] === "sdk-migration-status"
? { exitCode: 0, document: status }
: migrate,
);
}
it("runs the one migration command inside the runtime maintenance fence", async () => {
const ordering: string[] = [];
const execute = vi.fn(async (_core, args: string[]) => {
ordering.push(args[0] ?? "");
return args[0] === "sdk-migration-status"
? { exitCode: 0, document: { authority: "legacy", state: "required" } }
: { exitCode: 0, document: { state: "cutover" } };
});
const fence = async <T>(
dataDir: string,
action: () => Promise<T>,
): Promise<T> => {
ordering.push(`fence:start:${dataDir}`);
const result = await action();
ordering.push("fence:end");
return result;
};
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute,
fence,
});
expect(result).toEqual({ state: "cutover" });
expect(ordering).toEqual([
"sdk-migration-status",
"fence:start:/data",
"sdk-migrate",
"fence:end",
]);
expect(execute.mock.calls[1]?.[1]).toEqual([
"sdk-migrate",
"--runtime-build",
"20",
"--yes",
"--json",
]);
});
it("does not stop Runtime when nothing needs migrating", async () => {
let fenced = false;
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute: statusThen(
{ exitCode: 0, document: {} },
{ authority: "sdk", state: "not_required" },
),
fence: async (_dataDir, action) => {
fenced = true;
return action();
},
});
expect(result).toEqual({ state: "not_required" });
expect(fenced).toBe(false);
});
it("treats a finished cleanup as a completed migration", async () => {
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute: statusThen(
{ exitCode: 0, document: { state: "complete" } },
{ authority: "sdk", state: "required" },
),
fence: noFence,
});
expect(result).toEqual({ state: "cutover" });
});
it("reports an unreadable migration status as deferred", async () => {
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute: vi.fn(async () => ({ exitCode: 1, document: {} })),
fence: noFence,
});
expect(result).toEqual({
state: "deferred",
code: "sdk_migration_status_unavailable",
});
});
it("passes the migration failure code to the startup dialog", async () => {
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute: statusThen({
exitCode: 1,
document: {
state: "failed",
code: "sdk_migration_disk_space_insufficient",
},
}),
fence: noFence,
});
expect(result).toEqual({
state: "deferred",
code: "sdk_migration_disk_space_insufficient",
});
});
it("never reports success for a result it does not recognise", async () => {
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute: statusThen({ exitCode: 0, document: { state: "running" } }),
fence: noFence,
});
expect(result).toEqual({
state: "deferred",
code: "sdk_migration_incomplete",
});
});
it("names a Core that could not be run instead of throwing", async () => {
const result = await runSdkMigration({
dataDir: "/data",
core,
runtimeBuild: "20",
execute: vi.fn(async (_core, args: string[]) => {
if (args[0] === "sdk-migration-status") {
return {
exitCode: 0,
document: { authority: "legacy", state: "required" },
};
}
throw new Error("sdk_migration_response_invalid");
}),
fence: noFence,
});
expect(result).toEqual({
state: "deferred",
code: "sdk_migration_response_invalid",
});
});
it("streams Core's progress lines to the caller", async () => {
const directory = await mkdtemp(path.join(os.tmpdir(), "sdk-migrate-"));
const fake = path.join(directory, "core.sh");
await writeFile(
fake,
[
"#!/bin/sh",
'if [ "$1" = "sdk-migration-status" ]; then',
' echo \'{"authority":"legacy","state":"required"}\'; exit 0',
"fi",
'echo \'{"state":"running","step":"bundle"}\'',
'echo \'{"state":"running","step":"cutover"}\'',
'echo \'{"state":"cutover"}\'',
].join("\n"),
{ mode: 0o755 },
);
const steps: string[] = [];
const result = await runSdkMigration({
dataDir: directory,
core: { command: fake },
runtimeBuild: "20",
fence: noFence,
onProgress: (step) => steps.push(step),
});
expect(result).toEqual({ state: "cutover" });
expect(steps).toEqual(["bundle", "cutover"]);
});
@@ -1,174 +0,0 @@
import { spawn } from "node:child_process";
import { withRuntimeMaintenanceFence } from "./runtime-maintenance.js";
const MAX_OUTPUT_BYTES = 1024 * 1024;
export interface SdkMigrationResult {
state: "not_required" | "cutover" | "deferred";
code?: string;
}
export interface SdkMigrationCommandResult {
exitCode: number;
document: Record<string, unknown>;
}
export interface CoreCommand {
command: string;
args?: string[];
cwd?: string;
env?: NodeJS.ProcessEnv;
}
export interface SdkMigrationOptions {
dataDir: string;
core: CoreCommand;
runtimeBuild: string;
execute?: (
launch: CoreCommand,
args: string[],
dataDir: string,
onProgress?: (step: string) => void,
) => Promise<SdkMigrationCommandResult>;
fence?: <T>(dataDir: string, action: () => Promise<T>) => Promise<T>;
/** Called with `bundle`, `prepare`, `cutover`, `cleanup` as Core reaches them. */
onProgress?: (step: string) => void;
}
/**
* Bring a home to SDK memory authority before a Runtime serves it.
*
* Core's `sdk-migrate` does the whole migration and resumes after any failure;
* the caller (Desktop startup, CLI install/upgrade) only keeps the Runtime
* stopped, and its service from restarting, while it runs.
*/
export async function runSdkMigration(
options: SdkMigrationOptions,
): Promise<SdkMigrationResult> {
const execute = options.execute ?? executeCoreCommand;
const fence = options.fence ?? withRuntimeMaintenanceFence;
const status = await execute(
options.core,
["sdk-migration-status", "--json"],
options.dataDir,
).catch(() => null);
if (status === null || status.exitCode !== 0) {
return { state: "deferred", code: "sdk_migration_status_unavailable" };
}
if (status.document.state !== "required") {
return { state: "not_required" };
}
return fence(options.dataDir, async () => {
let migrated: SdkMigrationCommandResult;
try {
migrated = await execute(
options.core,
[
"sdk-migrate",
"--runtime-build",
options.runtimeBuild,
"--yes",
"--json",
],
options.dataDir,
options.onProgress,
);
} catch (error) {
const code = error instanceof Error ? error.message : "";
return {
state: "deferred",
code: /^[a-z][a-z0-9_]{0,95}$/.test(code)
? code
: "sdk_migration_response_invalid",
};
}
if (migrated.exitCode !== 0) return deferred(migrated);
const state = migrated.document.state;
if (state === "cutover" || state === "complete")
return { state: "cutover" };
if (state === "not_required") return { state: "not_required" };
return deferred(migrated);
});
}
function deferred(result: SdkMigrationCommandResult): SdkMigrationResult {
return {
state: "deferred",
code:
typeof result.document.code === "string"
? result.document.code
: "sdk_migration_incomplete",
};
}
async function executeCoreCommand(
launch: CoreCommand,
args: string[],
dataDir: string,
onProgress?: (step: string) => void,
): Promise<SdkMigrationCommandResult> {
return new Promise((resolve, reject) => {
const child = spawn(launch.command, [...(launch.args ?? []), ...args], {
cwd: launch.cwd,
env: { ...process.env, ...launch.env, ONE_MEMORY_DATA_DIR: dataDir },
stdio: ["ignore", "pipe", "pipe"],
});
let stdout = "";
let reported = 0;
let stderrBytes = 0;
child.stdout.setEncoding("utf8");
child.stdout.on("data", (chunk: string) => {
stdout += chunk;
const complete = stdout.lastIndexOf("\n");
if (onProgress && complete > reported) {
for (const line of stdout.slice(reported, complete).split("\n")) {
const step = progressStep(line);
if (step !== null) onProgress(step);
}
reported = complete + 1;
}
if (Buffer.byteLength(stdout) > MAX_OUTPUT_BYTES) {
child.kill("SIGKILL");
reject(new Error("sdk_migration_output_too_large"));
}
});
child.stderr.on("data", (chunk: Buffer) => {
stderrBytes += chunk.length;
if (stderrBytes > MAX_OUTPUT_BYTES) {
child.kill("SIGKILL");
reject(new Error("sdk_migration_error_output_too_large"));
}
});
child.once("error", reject);
child.once("close", (code) => {
try {
const line = stdout.trim().split("\n").at(-1) ?? "{}";
const document = JSON.parse(line) as unknown;
if (
!document ||
typeof document !== "object" ||
Array.isArray(document)
) {
throw new Error("sdk_migration_response_invalid");
}
resolve({
exitCode: code ?? 1,
document: document as Record<string, unknown>,
});
} catch {
reject(new Error("sdk_migration_response_invalid"));
}
});
});
}
function progressStep(line: string): string | null {
try {
const document = JSON.parse(line) as { state?: unknown; step?: unknown };
return document.state === "running" && typeof document.step === "string"
? document.step
: null;
} catch {
return null;
}
}
+1 -1
View File
@@ -36,7 +36,7 @@ MEMMY_LICENSE = (
MEMMY_PROVENANCE = MEMMY_LICENSE.with_name("PROVENANCE.md")
MEMMY_PIN = "25acd5f29fc7df73b943713abe61f1b2f9839887"
HY_MEMORY_NAME = "hy-memory"
HY_MEMORY_VERSION = "2.0.5"
HY_MEMORY_VERSION = "2.0.5.dev3"
class SbomError(RuntimeError):
+2 -2
View File
@@ -12,8 +12,8 @@ test("installation docs disclose publication and lifecycle boundaries", async ()
"utf8",
);
assert.match(readme, /`hy-memory==2\.0\.5`/);
assert.match(readme, /`v1\.0\.5` 的发行目标是 Linux x64\/arm64 Runtime/);
assert.match(readme, /`hy-memory==2\.0\.5\.dev3`/);
assert.match(readme, /`v1\.0\.4` 的发行目标是 Linux x64\/arm64 Runtime/);
assert.match(readme, /Linux 制品由 GitHub Actions/);
assert.match(readme, /macOS.*Developer ID 签名、公证/s);
assert.match(readme, /release-manifest\.json/);
+3 -3
View File
@@ -13,7 +13,7 @@ spec.loader.exec_module(module)
def test_hy_memory_notice_comes_from_the_installed_dist_info(tmp_path: Path) -> None:
dist_info = tmp_path / "hy_memory-2.0.5.dist-info"
dist_info = tmp_path / "hy_memory-2.0.5.dev3.dist-info"
licenses = dist_info / "licenses"
licenses.mkdir(parents=True)
(dist_info / "METADATA").write_text(
@@ -21,7 +21,7 @@ def test_hy_memory_notice_comes_from_the_installed_dist_info(tmp_path: Path) ->
(
"Metadata-Version: 2.4",
"Name: hy-memory",
"Version: 2.0.5",
"Version: 2.0.5.dev3",
"License-Expression: Apache-2.0",
"",
)
@@ -38,7 +38,7 @@ def test_hy_memory_notice_comes_from_the_installed_dist_info(tmp_path: Path) ->
notice = module.hy_memory_notice(dist_info)
assert notice["name"] == "hy-memory"
assert notice["version"] == "2.0.5"
assert notice["version"] == "2.0.5.dev3"
assert notice["license"] == "Apache-2.0"
assert notice["copyright"] == "Copyright (C) 2026 Tencent. All rights reserved."
assert "Licensed under the Apache License" in notice["text"]
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hy-memory"
version = "2.0.5"
version = "2.0.5.dev3"
description = "HY Memory - Industrial-grade dual-system cognitive memory framework"
readme = "README.md"
requires-python = ">=3.9"