diff --git a/memhub/apps/cli/src/cli.ts b/memhub/apps/cli/src/cli.ts index 159bb5c..7c7ca90 100644 --- a/memhub/apps/cli/src/cli.ts +++ b/memhub/apps/cli/src/cli.ts @@ -2018,12 +2018,17 @@ async function runCommand( return sourceProgressExitCode(document); } if (named === "profile") { + const profileFlags = arguments_.slice(1); if ( - arguments_.filter((item) => item === "--json").length > 1 || - arguments_.slice(1).some((value) => value !== "--json") + profileFlags.filter((item) => item === "--json").length > 1 || + profileFlags.filter((item) => item === "--versions").length > 1 || + profileFlags.some((value) => value !== "--json" && value !== "--versions") ) { return usage(); } + if (profileFlags.includes("--versions")) { + return await profileVersionsCommand(environment, { asJson, userId }); + } const connected = await connectRuntime(environment, { readerTimeoutMs: 15_000, }); @@ -3888,6 +3893,108 @@ async function emitModelReceipt( /** Attach to a verified Runtime, telling a stopped one (exit 3) from one * that is up but unusable (exit 1). */ +/** + * `memhub profile --versions`: the long-term profile's L0 version chains, + * latest first, so a value that replaced a better one can be found by id. + */ +async function profileVersionsCommand( + environment: NodeJS.ProcessEnv, + options: { asJson: boolean; userId?: string }, +): Promise { + const fail = (state: string, code: string, exitCode: number): number => { + emit( + options.asJson + ? `${JSON.stringify({ schema_version: 2, product: PRODUCT, state, code })}\n` + : `画像版本读取失败:${code}\n`, + ); + return exitCode; + }; + const connected = await connectRuntime(environment, { + readerTimeoutMs: 15_000, + }); + if (!connected.ok) { + return fail("unavailable", connected.code, connected.exitCode); + } + const outcome = await connected.reader( + "/api/v2/user-profile/versions", + userIdParameters(options.userId), + ); + const chains = outcome.state === "ok" ? profileChains(outcome.data) : null; + if (outcome.state !== "ok" || chains === null) { + return fail( + outcome.state === "ok" ? "invalid" : outcome.state, + outcome.state === "ok" ? "response_invalid" : outcome.code, + EXIT_FAILED, + ); + } + if (options.asJson) { + emit( + `${JSON.stringify({ + schema_version: 2, + product: PRODUCT, + state: "ok", + ...(outcome.data as object), + })}\n`, + ); + return EXIT_OK; + } + if (chains.length === 0) { + emit("尚未形成用户画像版本。\n"); + return EXIT_OK; + } + const lines: string[] = []; + for (const chain of chains) { + lines.push(`画像 ${chain.memory_id}(${chain.status})`); + for (const version of chain.versions) { + const when = + typeof version.gmt_created === "number" + ? new Date(version.gmt_created * 1000).toISOString() + : "时间未知"; + lines.push( + ` ${version.current ? "*" : " "} ${version.memory_id} ${when} ${version.content}`, + ); + } + } + lines.push( + "* 表示当前生效的版本;删除错误版本:memhub memories delete --yes", + ); + emit(`${lines.join("\n")}\n`); + return EXIT_OK; +} + +type ProfileChainView = { + memory_id: string; + status: string; + versions: { + memory_id: string; + content: string; + gmt_created?: number | null; + current: boolean; + }[]; +}; + +function profileChains(value: unknown): ProfileChainView[] | null { + if (!value || typeof value !== "object") return null; + const chains = (value as { chains?: unknown }).chains; + if (!Array.isArray(chains)) return null; + const valid = chains.every( + (chain) => + chain && + typeof chain === "object" && + typeof (chain as ProfileChainView).memory_id === "string" && + typeof (chain as ProfileChainView).status === "string" && + Array.isArray((chain as ProfileChainView).versions) && + (chain as ProfileChainView).versions.every( + (version) => + version && + typeof version.memory_id === "string" && + typeof version.content === "string" && + typeof version.current === "boolean", + ), + ); + return valid ? (chains as ProfileChainView[]) : null; +} + async function connectRuntime( environment: NodeJS.ProcessEnv, options?: { readerTimeoutMs?: number }, diff --git a/memhub/apps/cli/src/command-catalog.ts b/memhub/apps/cli/src/command-catalog.ts index 1929ba8..9bbb0aa 100644 --- a/memhub/apps/cli/src/command-catalog.ts +++ b/memhub/apps/cli/src/command-catalog.ts @@ -434,9 +434,18 @@ export const COMMANDS: readonly CommandDescriptor[] = [ }, { name: "profile", - summary: "读取 SDK 组装后的用户长期画像(L0 + 已启用的长期画像补充)。", - usage: "memhub profile [--user-id ] [--json]", - options: [USER_ID_OPTION, JSON_OPTION], + summary: + "读取 SDK 组装后的用户长期画像(L0 + 已启用的长期画像补充);--versions 列出 L0 画像的版本链(最新在前,* 为当前版本)。", + usage: "memhub profile [--versions] [--user-id ] [--json]", + options: [ + { + flag: "--versions", + summary: + "列出 L0 基本信息的版本链(get_user_basic_info),用于找出覆盖了正确值的错误版本。", + }, + USER_ID_OPTION, + JSON_OPTION, + ], exit_codes: RUNTIME_EXITS, }, { diff --git a/memhub/apps/cli/src/core-client.ts b/memhub/apps/cli/src/core-client.ts index dbc846e..52d64f6 100644 --- a/memhub/apps/cli/src/core-client.ts +++ b/memhub/apps/cli/src/core-client.ts @@ -17,6 +17,7 @@ const READ_ONLY_PATHS = [ "/api/v2/home-summary", "/api/v2/memory-library", "/api/v2/user-profile", + "/api/v2/user-profile/versions", ] as const; const SDK_MEMORY_DETAIL = /^\/api\/v2\/memories\/[A-Za-z0-9][\w.-]{0,127}$/; /** Status of one `POST /api/v2/memories` addition, by its request id. */ diff --git a/memhub/apps/cli/src/profile-versions-cli.test.ts b/memhub/apps/cli/src/profile-versions-cli.test.ts new file mode 100644 index 0000000..f673960 --- /dev/null +++ b/memhub/apps/cli/src/profile-versions-cli.test.ts @@ -0,0 +1,166 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { run } from "./cli.js"; + +const transport = vi.hoisted(() => ({ + get: vi.fn(), + post: vi.fn(), + put: vi.fn(), + patch: vi.fn(), + delete: vi.fn(), +})); +vi.mock("axios", () => ({ + default: { + create: vi.fn(() => transport), + get: vi.fn(), + post: vi.fn(), + isAxiosError: () => false, + }, +})); + +const runtime = vi.hoisted(() => ({ + inspectRuntimeStatus: vi.fn(), + discoverRuntimeAttachment: vi.fn(), +})); +vi.mock("@one-memory/runtime-control", async (importOriginal) => ({ + ...(await importOriginal()), + ...runtime, +})); + +const VERSIONS = { + user_id: "wife", + request_id: "r-versions", + elapsed_ms: 2, + chains: [ + { + memory_id: "l0-head", + content: "The user's name is Fei.", + status: "active", + memory_at: null, + gmt_created: 1_790_000_100, + basic_info: { name: "Fei" }, + versions: [ + { + memory_id: "l0-head", + content: "The user's name is Fei.", + memory_at: null, + gmt_created: 1_790_000_100, + current: true, + }, + { + memory_id: "l0-old", + content: "The user's name is 费天翔.", + memory_at: null, + gmt_created: 1_790_000_000, + current: false, + }, + ], + }, + ], +}; + +beforeEach(() => { + vi.clearAllMocks(); + for (const method of Object.values(transport)) method.mockReset(); + runtime.inspectRuntimeStatus.mockResolvedValue({ + state: "running", + code: "runtime_verified", + instanceId: "instance-1", + host: "127.0.0.1", + port: 8765, + version: "1.2.3", + }); + runtime.discoverRuntimeAttachment.mockResolvedValue({ + instanceId: "instance-1", + apiToken: "synthetic-runtime-token", + host: "127.0.0.1", + port: 8765, + servicePid: 4242, + health: { + status: "ok", + service: "one-memory-core", + version: "1.2.3", + instance_id: "instance-1", + }, + }); + transport.get.mockImplementation(async (url: string) => + url === "/api/v2/user-profile/versions" + ? { status: 200, data: VERSIONS } + : { status: 404, data: {} }, + ); +}); + +describe("memhub profile --versions", () => { + it("reads the L0 version chain for the named user", async () => { + const result = await capturedRun([ + "profile", + "--versions", + "--user-id", + "wife", + "--json", + ]); + expect(result.code).toBe(0); + expect(transport.get).toHaveBeenCalledWith( + "/api/v2/user-profile/versions", + { params: { user_id: "wife" } }, + ); + const document = JSON.parse(result.stdout); + expect(document.state).toBe("ok"); + expect(document.chains[0].versions).toHaveLength(2); + }); + + it("marks the current version in text and names no user for local", async () => { + const result = await capturedRun(["profile", "--versions"]); + expect(result.code).toBe(0); + const [url, config] = transport.get.mock.calls[0] ?? []; + expect(url).toBe("/api/v2/user-profile/versions"); + expect(config?.params?.user_id).toBeUndefined(); + expect(result.stdout).toContain("* l0-head"); + expect(result.stdout).toContain(" l0-old"); + expect(result.stdout).toContain("费天翔"); + }); + + it("refuses an unknown flag and a repeated --versions", async () => { + expect((await capturedRun(["profile", "--version-chain"])).code).toBe(2); + expect( + (await capturedRun(["profile", "--versions", "--versions"])).code, + ).toBe(2); + expect(transport.get).not.toHaveBeenCalled(); + }); + + it("reports a malformed answer as invalid", async () => { + transport.get.mockResolvedValue({ status: 200, data: { chains: "x" } }); + const result = await capturedRun(["profile", "--versions", "--json"]); + expect(result.code).toBe(1); + expect(JSON.parse(result.stdout).code).toBe("response_invalid"); + }); +}); + +async function capturedRun(arguments_: string[]) { + let stdout = ""; + let stderr = ""; + const stdoutSpy = vi + .spyOn(process.stdout, "write") + .mockImplementation((chunk) => { + stdout += String(chunk); + return true; + }); + const stderrSpy = vi + .spyOn(process.stderr, "write") + .mockImplementation((chunk) => { + stderr += String(chunk); + return true; + }); + try { + return { + code: await run(arguments_, { + ONE_MEMORY_DATA_DIR: "/synthetic/data", + }), + stdout, + stderr, + }; + } finally { + stdoutSpy.mockRestore(); + stderrSpy.mockRestore(); + } +} diff --git a/memhub/apps/core/src/one_memory/api.py b/memhub/apps/core/src/one_memory/api.py index 419814f..8275507 100644 --- a/memhub/apps/core/src/one_memory/api.py +++ b/memhub/apps/core/src/one_memory/api.py @@ -1,4 +1,5 @@ import asyncio +import hashlib import json import logging import os @@ -103,6 +104,7 @@ from one_memory.hy_sdk_types import ( HySdkError, HySdkMemoryPage, HySdkMemoryRecord, + HySdkProfileVersions, HySdkRawHit, HySdkRawPage, HySdkRawRecord, @@ -110,8 +112,10 @@ from one_memory.hy_sdk_types import ( HySdkUserProfile, LibraryHit, library_keyword_hits, + parse_sdk_lane_search, parse_sdk_raw_search, parse_sdk_search, + raw_turn_messages, ) from one_memory.imports.access import ( AuthorizedSourceRoot, @@ -230,6 +234,13 @@ from one_memory.sdk_memory_qa import ( from one_memory.sdk_memory_runtime import SdkMemoryRuntime from one_memory.sdk_migration_log import error_chain_text, error_frames from one_memory.sdk_read_handle import ActiveProfileReadHandle +from one_memory.sdk_search_options import ( + MEMORY_LAYERS, + MEMORY_STATUSES, + SdkSearchFilters, + SearchLane, + split_csv, +) from one_memory.sdk_write_processor import SdkWriteCoordinator from one_memory.secrets import ( KeyringSecretStore, @@ -237,6 +248,7 @@ from one_memory.secrets import ( SecretStoreErrorCode, SecretStoreUnavailableError, ) +from one_memory.semantic_deadline import SemanticMemoryHit from one_memory.service_levels import ( ResourceLimit, ServiceLevel, @@ -399,6 +411,23 @@ class CreateImportJobRequest(BaseModel): operation: Literal["history", "sync"] = "history" +class CreateImportJobV2Request(CreateImportJobRequest): + """`POST /api/v2/import-jobs`: v1's body plus whose lane it runs on.""" + + model_config = ConfigDict(extra="forbid") + + #: Only Hermes holds roots for a user other than `local`. + user_id: UserId = DEFAULT_USER_ID + + +class SourceSyncNowRequest(BaseModel): + """Optional body of `POST /api/v2/sources/{id}/sync`.""" + + model_config = ConfigDict(extra="forbid") + + user_id: UserId = DEFAULT_USER_ID + + class ConnectSourcesRequest(BaseModel): source_ids: tuple[str, ...] = Field(min_length=1, max_length=32) #: Whose grant this is. Only Hermes may connect a root for a user other @@ -454,13 +483,83 @@ class SdkUserProfileResponse(BaseModel): elapsed_ms: float = Field(ge=0) +class SdkProfileVersionsResponse(HySdkProfileVersions): + model_config = ConfigDict(frozen=True, extra="forbid") + + user_id: str + + +class SdkRawMessage(BaseModel): + model_config = ConfigDict(extra="forbid") + + #: user / assistant (tool and system if the SDK kept them); "" if unnamed. + role: str + content: str + + +class SdkRawItem(HySdkRawRecord): + """One SDK raw turn as `/api/v2/raw` lists it. + + The SDK record's fields, plus what tracing an import needs: the turn's + messages with their roles (parsed from the SDK's `[role]: text` lines), + a `sha256:` hash of the stored content computed by Core (the SDK keeps + none), and `created`, when the SDK stored it. + """ + + model_config = ConfigDict(frozen=True, extra="forbid") + + messages: tuple[SdkRawMessage, ...] = () + #: The one role of a single-message turn; None for a mixed turn. + role: str | None = None + content_hash: str + created: datetime | None = None + + @classmethod + def of(cls, record: HySdkRawRecord) -> "SdkRawItem": + messages = tuple( + SdkRawMessage(role=role, content=text) + for role, text in raw_turn_messages(record.content) + ) + roles = {message.role for message in messages} + return cls( + **record.model_dump(), + messages=messages, + role=next(iter(roles)) if len(roles) == 1 and "" not in roles else None, + content_hash="sha256:" + hashlib.sha256(record.content.encode("utf-8")).hexdigest(), + created=( + datetime.fromtimestamp(record.gmt_created, tz=UTC) + if record.gmt_created is not None + else None + ), + ) + + class SdkRawPage(BaseModel): model_config = ConfigDict(extra="forbid") - items: tuple[HySdkRawRecord, ...] = () + items: tuple[SdkRawItem, ...] = () total: int = Field(ge=0) offset: int = Field(ge=0) limit: int = Field(ge=1, le=500) + #: Set when a time filter was applied to this page in Core (the SDK list + #: has none): `items` is the page's rows inside the range, `total` stays + #: the SDK's count before that filter. + time_filtered: bool = False + + +class SdkLayerCounts(BaseModel): + model_config = ConfigDict(extra="forbid") + + #: Every SDK layer, zero when empty. + layers: dict[str, int] + total: int = Field(ge=0) + #: The statuses counted, or None for the defaults (active; raw: active+shadow). + statuses: tuple[str, ...] | None = None + + +#: Raw lifecycle states that are still a kept conversation. Pro marks a raw +#: SHADOW once extraction succeeds; lite leaves it ACTIVE. +RAW_STATUSES = ("active", "shadow") class MemoryProcessingBlock(BaseModel): @@ -605,8 +704,15 @@ class ManualExtractReceipt(BaseModel): batches_current: int -class AskMemoryRequest(BaseModel): - model_config = ConfigDict(extra="forbid") +class AskMemoryRequest(SdkSearchFilters): + """Ask over one user's memories. + + The `SdkSearchFilters` fields (floors, rerank, source/session/agent/time + and layer filters) reach the retrieval search. Scene, reader and search + mode stay fixed: the answer cites chat memories read back by id. + """ + + model_config = ConfigDict(extra="forbid", frozen=False) question: str = Field(min_length=1, max_length=2_000) scope: str = Field(default="all-apps", min_length=1, max_length=512) @@ -1964,7 +2070,11 @@ def create_app( dependencies=protected, ) async def list_sources() -> list[SourceResponse]: - detection = _detection(app) + return await sources_for(DEFAULT_USER_ID) + + async def sources_for(user_id: str) -> list[SourceResponse]: + """One user's sources. `local`'s are the App's; another user sees Hermes only.""" + detection = _detection(app) if user_id == DEFAULT_USER_ID else {} # Read in threads. The desktop polls this every few seconds, and both # reads below are synchronous SQLite — the totals across every source, # then a grant receipt per source. Made on the loop, as they were, @@ -1974,7 +2084,7 @@ def create_app( # The totals are also not counted again for each ask: counting is a # pass over every row, and answering this took up to 6.70s during a # catch-up, nearly all of it the same count repeated. - totals_by_source = await app.state.source_totals.by_source(user_id="local") + totals_by_source = await app.state.source_totals.by_source(user_id=user_id) # Status polling must not trigger history discovery: detect can walk an # entire host tree and recheck the grant in SQLite for every file, with # concurrent polls duplicating that work. Only the explicit detect @@ -1984,28 +2094,49 @@ def create_app( def describe() -> list[SourceResponse]: grant_store = _grant_store(app) config_store = _config_store(app) - return [ - _source_response( - registration, - grant_store.receipt(user_id="local", source_id=source_id), - config_store.sync_enabled(source_id), - detection.get(source_id), - runtime_settings.source_home, - totals_by_source.get(source_id, (0, 0)), - candidate_change_reason=app.state.source_candidate_store.change_reason( - registration - ), - authorization_state=grant_store.authorization_state( - user_id="local", - source_id=source_id, - candidate=registration.candidate.path, - ), + responses = [] + for source_id, registration in _catalog(app).items(): + if user_id != DEFAULT_USER_ID and source_id not in MULTI_USER_SOURCES: + continue + if user_id != DEFAULT_USER_ID: + # Another user's Hermes is the profile root they were + # granted, not the App's candidate. + registration = _registration_at_grant(grant_store, registration, user_id) + responses.append( + _source_response( + registration, + grant_store.receipt(user_id=user_id, source_id=source_id), + _lane_sync_enabled(config_store, lane_key(user_id, source_id)), + detection.get(source_id), + runtime_settings.source_home, + totals_by_source.get(source_id, (0, 0)), + candidate_change_reason=( + app.state.source_candidate_store.change_reason(registration) + if user_id == DEFAULT_USER_ID + else "multi_user_grant" + ), + authorization_state=grant_store.authorization_state( + user_id=user_id, + source_id=source_id, + candidate=registration.candidate.path, + ), + ) ) - for source_id, registration in _catalog(app).items() - ] + return responses return await asyncio.to_thread(describe) + @app.get( + "/api/v2/sources", + response_model=list[SourceResponse], + dependencies=strict, + ) + async def list_sources_v2( + user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + ) -> list[SourceResponse]: + """`GET /api/v1/sources` for one user: `local` (the default) reads the App's.""" + return await sources_for(user_id) + @app.get( "/api/v1/adapters", dependencies=protected, @@ -2523,20 +2654,37 @@ def create_app( dependencies=protected, ) async def create_import_job(request: CreateImportJobRequest) -> ImportJob: - _require_registration(app, request.source_id) - if request.source_id not in _supervisor(app).adapters: + return await create_job_on(request.source_id, request.operation, DEFAULT_USER_ID) + + async def create_job_on( + source_id: str, operation: Literal["history", "sync"], user_id: str + ) -> ImportJob: + """Start an import on one user's lane of a source (`local`: the bare id).""" + _require_registration(app, source_id) + _require_multi_user_source(source_id, user_id) + lane = lane_key(user_id, source_id) + if lane not in _supervisor(app).adapters: raise HTTPException(status_code=409, detail="source_not_authorized") try: job = await asyncio.to_thread( _supervisor(app).create_job, - request.source_id, - operation=request.operation, + lane, + operation=operation, ) except SourceAccessError as error: raise HTTPException(status_code=409, detail=error.code) from None _supervisor(app).enqueue(job.job_id) return job + @app.post( + "/api/v2/import-jobs", + response_model=ImportJob, + dependencies=strict, + ) + async def create_import_job_v2(request: CreateImportJobV2Request) -> ImportJob: + """`POST /api/v1/import-jobs` on one user's lane (Hermes only for another user).""" + return await create_job_on(request.source_id, request.operation, request.user_id) + @app.post( "/api/v1/sources/connect", response_model=ConnectSourcesResponse, @@ -2786,14 +2934,55 @@ def create_app( dependencies=protected, ) async def trigger_live_sync(source_id: str) -> ImportJob: + return await sync_lane(source_id, DEFAULT_USER_ID) + + async def sync_lane(source_id: str, user_id: str) -> ImportJob: _require_registration(app, source_id) + _require_multi_user_source(source_id, user_id) # Asked for by hand, so it reports a job even when nothing was new; # `live_sync_unavailable` keeps meaning the source cannot be synced. - job = await _live_sync(app).trigger_now(source_id) + job = await _live_sync(app).trigger_now(lane_key(user_id, source_id)) if job is None: raise HTTPException(status_code=409, detail="live_sync_unavailable") return job + @app.post( + "/api/v2/sources/{source_id}/sync", + response_model=ImportJob, + dependencies=strict, + ) + async def trigger_live_sync_v2( + source_id: str, + request: SourceSyncNowRequest | None = None, + ) -> ImportJob: + """`POST /api/v1/sources/{id}/sync` on one user's lane; the body is optional.""" + return await sync_lane(source_id, request.user_id if request else DEFAULT_USER_ID) + + @app.get( + "/api/v2/import-jobs", + response_model=list[ImportJob], + dependencies=strict, + ) + def list_import_jobs_v2( + user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + ) -> list[ImportJob]: + return list_import_jobs(user_id=user_id) + + @app.get( + "/api/v2/import-jobs/{job_id}", + response_model=ImportJob, + dependencies=strict, + ) + def get_import_job_v2( + job_id: str, + user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + ) -> ImportJob: + """One job; another user's job answers exactly like a missing one.""" + job = _require_job(app, job_id) + if job.user_id != user_id: + raise HTTPException(status_code=404, detail="job_not_found") + return job + @app.get( "/api/v1/import-jobs/{job_id}", response_model=ImportJob, @@ -3156,6 +3345,11 @@ def create_app( window_id: str | None = Query(default=None, min_length=1, max_length=64), scene: Literal["all", "chat", "working", "coding"] = Query(default="all"), user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + layers: list[str] | None = Query(default=None), # noqa: B008 + layer: list[str] | None = Query(default=None), # noqa: B008 + exclude_sources: list[str] | None = Query(default=None), # noqa: B008 + session_ids: list[str] | None = Query(default=None), # noqa: B008 + agent_ids: list[str] | None = Query(default=None), # noqa: B008 ) -> HySdkMemoryPage: """One page of memories across chat, working and coding. @@ -3170,7 +3364,29 @@ def create_app( and are cut from the same matches, so pages never overlap or skip. A `window_id` that expired or was searched with other filters searches again and answers page 1 with `window_reset: true`. + + `layers` (or `layer`), `statuses`, `exclude_sources`, `session_ids` + and `agent_ids` may each repeat or be comma separated. Layers exist in + the chat store only, so a layer filter lists the chat store (graph + layers included) and refuses `scene=working|coding`. `l1_raw` belongs + to `/api/v2/raw`. """ + statuses = _library_values(statuses, MEMORY_STATUSES, "memory_status_unknown") + chosen_layers = _library_values( + [*(layers or ()), *(layer or ())], MEMORY_LAYERS, "memory_layer_unknown" + ) + if chosen_layers and scene in {"working", "coding"}: + raise HTTPException(status_code=422, detail="memory_layers_require_chat_scene") + filters = { + name: values + for name, values in ( + ("layers", chosen_layers), + ("exclude_sources", split_csv(exclude_sources)), + ("session_ids", split_csv(session_ids)), + ("agent_ids", split_csv(agent_ids)), + ) + if values + } try: client = await _sdk_client_handle(app).client() search_query = (query or "").strip() @@ -3191,16 +3407,20 @@ def create_app( window_id=window_id, scene=scene, user_id=user_id, + filters=filters, ) raw = await asyncio.to_thread( client.list_memories, user_id=user_id, - scene=scene, + # Layers are chat-store rows: `scene="all"` would still add + # every working/coding experience to a layer-filtered page. + scene=None if chosen_layers else scene, statuses=list(statuses), include_sources=[source] if source else None, offset=offset, limit=limit, order="desc", + **filters, ) page, skipped = HySdkMemoryPage.from_sdk_skipping_invalid( raw, limit=limit, offset=offset @@ -3212,6 +3432,51 @@ def create_app( except Exception as error: raise _sdk_http_error(error, "hy_memory_list_failed") from None + @app.get( + "/api/v2/memory-library/layers", + response_model=SdkLayerCounts, + dependencies=strict, + ) + async def sdk_memory_layers( + statuses: list[str] | None = Query(default=None), # noqa: B008 + user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + ) -> SdkLayerCounts: + """How many memories each SDK layer holds for one user. + + One `list_memories(layers=[layer], limit=1)` per layer, read for its + total only. Without `statuses` every layer counts its active rows, + and `l1_raw` counts active and shadow raws: pro shadows a raw once + it is extracted, and both are the user's kept conversations. + """ + chosen = _library_values(statuses or [], MEMORY_STATUSES, "memory_status_unknown") + try: + client = await _sdk_client_handle(app).client() + + def count(layer: str) -> int: + layer_statuses = chosen or (list(RAW_STATUSES) if layer == "l1_raw" else ["active"]) + raw = client.list_memories( + user_id=user_id, + layers=[layer], + statuses=layer_statuses, + limit=1, + offset=0, + order="desc", + ) + return _list_total(raw) + + counts = await asyncio.to_thread( + lambda: {layer: count(layer) for layer in MEMORY_LAYERS} + ) + except HTTPException: + raise + except Exception as error: + raise _sdk_http_error(error, "hy_memory_list_failed") from None + return SdkLayerCounts( + layers=counts, + total=sum(counts.values()), + statuses=tuple(chosen) if chosen else None, + ) + @app.get( "/api/v2/memories/graph", response_model=MemoryGraphResponse, @@ -3222,13 +3487,30 @@ def create_app( offset: int = Query(default=0, ge=0), source: str | None = Query(default=None, min_length=1), user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + layers: list[str] | None = Query(default=None), # noqa: B008 + session_ids: list[str] | None = Query(default=None), # noqa: B008 + exclude_sources: list[str] | None = Query(default=None), # noqa: B008 ) -> MemoryGraphResponse: """The memory map on SDK authority: SDK memories plus the Kuzu overlay. Memories come from the public client; only the overlay reads Kuzu (`hy_graph_projection`, the visualization exception). Shape matches the legacy map so the same view draws both. + + `layers` maps the chat store's chosen layers only (as on the library); + `session_ids` and `exclude_sources` narrow the listed memories. Each + may repeat or be comma separated. The overlay is not filtered. """ + chosen_layers = _library_values(layers or [], MEMORY_LAYERS, "memory_layer_unknown") + filters = { + name: values + for name, values in ( + ("layers", chosen_layers), + ("session_ids", split_csv(session_ids)), + ("exclude_sources", split_csv(exclude_sources)), + ) + if values + } try: client = await _sdk_client_handle(app).client() # The map visualizes the active source scope, independently of the @@ -3236,17 +3518,20 @@ def create_app( raw = await asyncio.to_thread( client.list_memories, user_id=user_id, - scene="all", + scene=None if chosen_layers else "all", statuses=["active"], include_sources=[source] if source else None, limit=limit, offset=offset, order="desc", + **filters, ) page, skipped = HySdkMemoryPage.from_sdk_skipping_invalid( raw, limit=limit, offset=offset ) _log_skipped_sdk_memories(skipped) + except HTTPException: + raise except Exception as error: raise _sdk_http_error(error, "hy_memory_list_failed") from None vdb = raw.get("vdb") if isinstance(raw, dict) else None @@ -3370,6 +3655,27 @@ def create_app( raise _sdk_http_error(error, "hy_memory_profile_read_failed") from None return SdkUserProfileResponse.model_validate(profile.model_dump()) + @app.get( + "/api/v2/user-profile/versions", + response_model=SdkProfileVersionsResponse, + dependencies=strict, + ) + async def sdk_user_profile_versions( + user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + ) -> SdkProfileVersionsResponse: + """The long-term profile's L0 version chains (`get_user_basic_info`). + + Each chain is one L0 head with its versions latest first, so a value + that replaced a better one can be found and deleted by id. + """ + try: + client = await _sdk_client_handle(app).client() + value = await asyncio.to_thread(client.get_user_basic_info, user_id=user_id) + versions = HySdkProfileVersions.from_sdk(value) + except Exception as error: + raise _sdk_http_error(error, "hy_memory_profile_read_failed") from None + return SdkProfileVersionsResponse(user_id=user_id, **versions.model_dump()) + @app.get( "/api/v2/memories/{memory_id}", response_model=HySdkMemoryRecord, @@ -3502,25 +3808,69 @@ def create_app( dependencies=strict, ) async def sdk_raw_library( - source: str | None = Query(default=None, min_length=1), + source: list[str] | None = Query(default=None), # noqa: B008 offset: int = Query(default=0, ge=0), limit: int = Query(default=100, ge=1, le=500), user_id: str = Query(default=DEFAULT_USER_ID, pattern=USER_ID_PATTERN), + exclude_sources: list[str] | None = Query(default=None), # noqa: B008 + session_id: list[str] | None = Query(default=None), # noqa: B008 + statuses: list[str] | None = Query(default=None), # noqa: B008 + after: float | None = Query(default=None, ge=0), + before: float | None = Query(default=None, ge=0), + order: Literal["desc", "asc"] = Query(default="desc"), ) -> SdkRawPage: + """One page of a user's SDK raw turns: what was imported, as stored. + + The SDK lists raw only when asked for `layers=["l1_raw"]`, and only + active rows unless told otherwise. Pro shadows a raw once extraction + succeeds, so a raw page reads active and shadow (`statuses` narrows + it). `source` and `session_id` (a conversation) may repeat or be comma + separated. `after`/`before` (unix seconds, against the conversation + time, else the stored time) have no SDK list filter: they cut this + page in Core and mark it `time_filtered`. + """ + chosen_statuses = _library_values( + statuses or [], MEMORY_STATUSES, "memory_status_unknown" + ) or list(RAW_STATUSES) + sources = split_csv(source) + excluded = split_csv(exclude_sources) + sessions = split_csv(session_id) try: client = await _sdk_client_handle(app).client() value = await asyncio.to_thread( client.list_memories, user_id=user_id, layers=["l1_raw"], - include_sources=[source] if source else None, + statuses=chosen_statuses, + include_sources=sources or None, + exclude_sources=excluded or None, + session_ids=sessions or None, offset=offset, limit=limit, + order=order, ) page = HySdkRawPage.from_memory_list(value, limit=limit, offset=offset) + except HTTPException: + raise except Exception as error: raise _sdk_http_error(error, "hy_memory_raw_list_failed") from None - return SdkRawPage(items=page.raw, total=page.total, offset=offset, limit=limit) + records = page.raw + time_filtered = after is not None or before is not None + if time_filtered: + records = tuple( + record + for record in records + if (at := record.memory_at or record.gmt_created) is not None + and (after is None or at >= after) + and (before is None or at < before) + ) + return SdkRawPage( + items=tuple(SdkRawItem.of(record) for record in records), + total=page.total, + offset=offset, + limit=limit, + time_filtered=time_filtered, + ) @app.post( "/api/v2/memories/ask", @@ -3543,7 +3893,7 @@ def create_app( ) try: answer = await _sdk_request_answerer(app).answer( - request.question, user_id=request.user_id + request.question, user_id=request.user_id, **_ask_filters(request) ) except MemoryAnswerError as error: raise HTTPException(status_code=503, detail=str(error)) from None @@ -3584,7 +3934,7 @@ def create_app( final: dict[str, object] | None = None try: async for event in answerer.answer_stream( - request.question, user_id=request.user_id + request.question, user_id=request.user_id, **_ask_filters(request) ): if event.get("type") == "final": final = event @@ -3629,11 +3979,15 @@ def create_app( await _require_external_recall_scope( app, http_request, request.scope, request.workspace_path ) + # A caller's pass-through search options need the SDK target, which + # carries them; without any, recall is the call it always was. + options = request if request.search_overrides() else None target = ( _recall_target(app) - if request.user_id == DEFAULT_USER_ID - else _sdk_recall_target(app, request.user_id) + if request.user_id == DEFAULT_USER_ID and options is None + else _sdk_recall_target(app, request.user_id, options) ) + raw_options = {"filters": options} if options is not None else {} access = external_access(http_request.headers) receipt = await recall_with_fallback( request, @@ -3642,7 +3996,12 @@ def create_app( has_memories=lambda: _library_has_memories(app, request.user_id), raw_sources=lambda: _recall_raw_sources(app, request, access, started), search_raw=lambda query, limit, sources: _search_sdk_raw( - app, query, limit=limit, sources=sources, user_id=request.user_id + app, + query, + limit=limit, + sources=sources, + user_id=request.user_id, + **raw_options, ), ) event_store = getattr(app.state, "call_event_store", None) @@ -3932,6 +4291,11 @@ async def _require_external_recall_scope( access.require(tool, frozenset({scope})) +def _ask_filters(request: AskMemoryRequest) -> dict[str, SdkSearchFilters]: + """The answerer's `filters` argument, only when the caller set any.""" + return {"filters": request} if request.search_overrides() else {} + + def _sdk_request_answerer(app: FastAPI) -> SdkMemoryAnswerer: return SdkMemoryAnswerer( client_handle=_sdk_client_handle(app), @@ -4108,6 +4472,25 @@ def _active_adapters( return adapters +def _registration_at_grant( + grant_store: SourceGrantStore, + registration: SourceRegistration, + user_id: str, +) -> SourceRegistration: + """The registration with its candidate at `user_id`'s granted root, if any.""" + for receipt, root_path in grant_store.grants(): + if receipt.user_id == user_id and receipt.source_id == registration.descriptor.source_id: + return replace( + registration, + candidate=replace( + registration.candidate, + path=Path(root_path), + kind=RootCandidateKind.USER_SELECTED, + ), + ) + return registration + + def _source_response( registration: SourceRegistration, receipt: SourceGrantReceipt | None, @@ -4192,12 +4575,21 @@ def _require_registration(app: FastAPI, source_id: str) -> SourceRegistration: return registration +def _require_multi_user_source(source_id: str, user_id: str) -> None: + """Refuse another user's lane of a source that holds `local`'s roots only.""" + if user_id != DEFAULT_USER_ID and source_id not in MULTI_USER_SOURCES: + raise HTTPException(status_code=422, detail="user_id_requires_multi_user_source") + + def _diagnostic_request_event(method: str, request_path: str) -> str | None: if method == "GET" or request_path in {"/api/v1/health", "/api/v1/runtime/identity"}: return None families = ( ("/api/v1/import", "import.request"), ("/api/v1/sources", "source.request"), + ("/api/v2/import", "import.request"), + ("/api/v2/sources", "source.request"), + ("/api/v2/evolve", "memory.request"), ("/api/v1/model", "model.request"), ("/api/v1/memor", "memory.request"), ("/api/v2/context", "recall.request"), @@ -4777,23 +5169,33 @@ async def _recall_model_ready(app: FastAPI) -> bool: return profile is not None and bool(profile.capabilities.model_available) -def _sdk_recall_target(app: FastAPI, user_id: str) -> SdkRecallTarget: - """Recall over one user's SDK memories; `get` results are kept only if theirs.""" +def _sdk_recall_target( + app: FastAPI, + user_id: str, + options: SdkSearchFilters | None = None, +) -> SdkRecallTarget: + """Recall over one user's SDK memories; `get` results are kept only if theirs. + + With a caller's pass-through `options`, the search carries them and a + working/coding hit is read back from its own namespace. + """ handle = _sdk_client_handle(app) + namespaces: dict[str, dict[str, str]] = {} async def owned(memory_id: str) -> HySdkMemoryRecord | None: - memory = await _get_sdk_memory(handle, memory_id) + memory = await _get_sdk_memory(handle, memory_id, **namespaces.get(memory_id, {})) return memory if memory is not None and owns(memory.user_id, user_id) else None - return SdkRecallTarget( - semantic_search=lambda query: _search_sdk_memories( - handle, - query, - user_id=user_id, - agent_id="*", - ), - get_memory=owned, - ) + async def search(query: str) -> tuple[object, ...]: + hits = await _search_sdk_memories( + handle, query, user_id=user_id, agent_id="*", options=options + ) + for hit in hits: + if isinstance(hit, SemanticMemoryHit) and hit.variant is not None: + namespaces[hit.memory_id] = {"kind": hit.kind, "variant": hit.variant} + return hits + + return SdkRecallTarget(semantic_search=search, get_memory=owned) async def _library_has_memories(app: FastAPI, user_id: str = DEFAULT_USER_ID) -> bool: @@ -4861,25 +5263,47 @@ async def _search_sdk_raw( limit: int, sources: frozenset[str] | None, user_id: str = DEFAULT_USER_ID, + filters: SdkSearchFilters | None = None, ) -> tuple[HySdkRawHit, ...]: - """The imported conversations nearest the query: the SDK raw layer.""" + """The imported conversations nearest the query: the SDK raw layer. + + `filters` are a caller's pass-through options: its floor and row filters + apply, and its `include_sources` only narrow `sources`, never widen them. + """ + from one_memory.context_routes import raw_lane_overrides + + include = sorted(sources) if sources else None + if filters is not None and filters.include_sources is not None: + include = [ + source + for source in dict.fromkeys(filters.include_sources) + if sources is None or source in sources + ] + if not include: + return () client = await _sdk_client_handle(app).client() raw = await asyncio.to_thread( client.search, query, - user_ids=[user_id], - agent_ids=[], - include_sources=sorted(sources) if sources else None, - limit=limit, - min_score=0.0, - schema_min_score=0.0, - schema_limit=0, - intention_limit=0, - scene="chat", - reader=HY_MEMORY_READER, - layers=["l1_raw"], + **{ + "user_ids": [user_id], + "agent_ids": [], + "include_sources": include, + "limit": limit, + "min_score": 0.0, + "schema_min_score": 0.0, + "schema_limit": 0, + "intention_limit": 0, + "scene": "chat", + "reader": HY_MEMORY_READER, + "layers": ["l1_raw"], + **(raw_lane_overrides(filters) if filters is not None else {}), + }, ) - return parse_sdk_raw_search(raw) + hits = parse_sdk_raw_search(raw) + if filters is None: + return hits + return tuple(hit for hit in hits if filters.passes_floor("raw", hit.score)) def _sdk_client_handle(app: FastAPI) -> SdkClientHandle: @@ -4961,6 +5385,30 @@ def _library_search_windows(app: FastAPI) -> LibrarySearchWindows: return cast(LibrarySearchWindows, windows) +def _library_values(values: list[str], allowed: tuple[str, ...], code: str) -> list[str]: + """Repeated or comma-separated query values, each one the SDK knows.""" + chosen = split_csv(values) + if any(value not in allowed for value in chosen): + raise HTTPException(status_code=422, detail=code) + return chosen + + +def _list_total(raw: object) -> int: + """A chat-store `list_memories` reply's total: its vector and graph buckets.""" + if not isinstance(raw, dict): + raise HySdkReadError("hy_memory_list_shape_unknown") + total = 0 + for name in ("vdb", "graph"): + bucket = raw.get(name) + if bucket is None: + continue + value = bucket.get("total") if isinstance(bucket, dict) else None + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise HySdkReadError("hy_memory_list_shape_unknown") + total += value + return total + + async def _library_search_page( app: FastAPI, client: Any, @@ -4973,6 +5421,7 @@ async def _library_search_page( window_id: str | None, scene: str = "all", user_id: str = DEFAULT_USER_ID, + filters: dict[str, list[str]] | None = None, ) -> HySdkMemoryPage: """One page of a keyword search, cut from a cached result window. @@ -4982,11 +5431,16 @@ async def _library_search_page( contain every word, and caches those matches under a `window_id`. Every page of that search is a slice of the same matches, read fresh from the SDK so edits and deletes show. The whole library is never listed in Core. + + `filters` (layers, exclude_sources, session_ids, agent_ids) reach the SDK + search and are part of the window's identity. A layer filter keeps chat + matches only: the SDK applies layers to the chat lane alone. """ windows = _library_search_windows(app) + filters = filters or {} # The user is part of the window's identity: one user's `window_id` must # never cut a page from another user's matches. - key = window_key(query, source, statuses, scene, user_id) + key = window_key(query, source, statuses, scene, user_id, filters) window = windows.get(window_id, key) if window_id else None window_reset = window_id is not None and window is None if window is None: @@ -4997,7 +5451,7 @@ async def _library_search_page( scene="all", rerank=False, user_ids=[user_id], - agent_ids=[], + agent_ids=filters.get("agent_ids", []), include_sources=[source] if source else None, limit=fetch, min_score=0.0, @@ -5009,8 +5463,11 @@ async def _library_search_page( # recall-time lane the list never shows. intention_limit=0, reader=HY_MEMORY_READER, + **{name: values for name, values in filters.items() if name != "agent_ids"}, ) hits, held = library_keyword_hits(search_raw, query) + if "layers" in filters: + hits = tuple(hit for hit in hits if hit.kind == "chat") if scene != "all": # One scene: keep that scene's matches from the same fused window. hits = tuple( @@ -5053,22 +5510,33 @@ async def _search_sdk_memories( user_id: str, agent_id: str, limit: int = 12, + options: SdkSearchFilters | None = None, ) -> tuple[object, ...]: client = await handle.client() + overrides = options.search_overrides() if options is not None else {} raw = await asyncio.to_thread( client.search, query, - user_ids=[user_id], - agent_ids=[] if agent_id == "*" else [agent_id], - limit=limit, - min_score=sdk_relevance_floor(), - schema_min_score=sdk_relevance_floor(), - schema_limit=limit, - intention_limit=limit, - scene="normal", - reader=HY_MEMORY_READER, + **{ + "user_ids": [user_id], + "agent_ids": [] if agent_id == "*" else [agent_id], + "limit": limit, + "min_score": sdk_relevance_floor(), + "schema_min_score": sdk_relevance_floor(), + "schema_limit": limit, + "intention_limit": limit, + "scene": "normal", + "reader": HY_MEMORY_READER, + **overrides, + }, + ) + if options is None or not overrides: + return parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown") + return tuple( + hit + for hit in parse_sdk_lane_search(raw, error_code="hy_memory_search_shape_unknown") + if options.passes_floor(cast(SearchLane, hit.lane), hit.score) ) - return parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown") def _sdk_http_error(error: Exception, fallback: str) -> HTTPException: diff --git a/memhub/apps/core/src/one_memory/context_models.py b/memhub/apps/core/src/one_memory/context_models.py index 25fce74..281750d 100644 --- a/memhub/apps/core/src/one_memory/context_models.py +++ b/memhub/apps/core/src/one_memory/context_models.py @@ -5,6 +5,7 @@ from typing import Annotated, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator +from one_memory.sdk_search_options import SdkSearchOptions from one_memory.user_scope import DEFAULT_USER_ID, UserId ContextKind = Literal["raw", "memory"] @@ -49,12 +50,17 @@ class ContextSearchRequest(BaseModel): return self -class SdkContextSearchRequest(ContextSearchRequest): +class SdkContextSearchRequest(ContextSearchRequest, SdkSearchOptions): """`POST /api/v2/context/search`: the three SDK recall lanes are set apart. `limit` is the normal lane's top-k and the cap on returned items; `schema_limit` and `intention_limit` are the SDK's schema (L6) and proactive intention (L7) lanes. A caller's values are passed through. + + The `SdkSearchOptions` fields reach the memory lane's `client.search`; + the raw lane takes the floor and the source, session, agent and time + filters (never wider than the caller's raw authority). Absent, both + lanes make the call they always made. """ limit: int = Field(default=10, ge=1, le=20, strict=True) diff --git a/memhub/apps/core/src/one_memory/context_routes.py b/memhub/apps/core/src/one_memory/context_routes.py index 500ad4d..3084d7a 100644 --- a/memhub/apps/core/src/one_memory/context_routes.py +++ b/memhub/apps/core/src/one_memory/context_routes.py @@ -24,14 +24,57 @@ from one_memory.hy_runtime_config import HY_MEMORY_READER from one_memory.hy_sdk_runtime import sdk_relevance_floor from one_memory.hy_sdk_types import ( HySdkMemoryRecord, + parse_sdk_lane_search, parse_sdk_raw_search, parse_sdk_search, ) from one_memory.imports.storage import EvidenceStore +from one_memory.sdk_search_options import SdkSearchFilters, SearchLane +from one_memory.semantic_deadline import SemanticMemoryHit from one_memory.user_scope import owns CALL_EVENT_RESPONSE_BUDGET_MS = 250 +#: Pass-through filters the raw lane honours. Scene, layers, reader and the +#: schema/working floors describe memory lanes; the raw lane is always the +#: chat store's L1 raw layer. +_RAW_LANE_FILTERS = ("created_after", "session_ids", "agent_ids", "exclude_sources", "rerank") + + +def raw_lane_sources( + allowed: Sequence[str] | frozenset[str] | None, + requested: Sequence[str] | None, +) -> list[str] | None: + """The raw lane's source filter: the caller's, never wider than its authority. + + `allowed` empty or None means every source (the call Core always made). + None back means nothing is left to search. + """ + allowed_list = list(allowed or ()) + if requested is None: + return allowed_list + chosen = [ + source for source in dict.fromkeys(requested) if not allowed_list or source in allowed_list + ] + return chosen or None + + +def raw_lane_overrides(value: SdkSearchFilters) -> dict[str, object]: + """The caller's raw-lane overrides: its floor and its row filters.""" + overrides = value.filter_kwargs() + found = {name: overrides[name] for name in _RAW_LANE_FILTERS if name in overrides} + floor = value.lane_floor("raw") + if floor is not None: + found["min_score"] = floor + return found + + +def hit_namespace(hit: SemanticMemoryHit) -> dict[str, str]: + """The `get` namespace of a hit: nothing for chat, kind and variant otherwise.""" + if hit.kind in {"experience", "fact"} and hit.variant is not None: + return {"kind": hit.kind, "variant": hit.variant} + return {} + def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter: """SDK-native context: no Evidence/Governance ids or fabricated revisions.""" @@ -71,22 +114,30 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter: async def raw_lane() -> tuple[SdkRawContextItem, ...]: if "raw" not in value.kinds or not authority.raw_allowed: return () + sources = raw_lane_sources(authority.raw_sources, value.include_sources) + if sources is None: + return () raw = await asyncio.to_thread( client.search, value.query, - user_ids=[value.user_id], - agent_ids=[], - include_sources=list(authority.raw_sources or ()), - limit=value.limit, - min_score=0.0, - schema_min_score=0.0, - schema_limit=0, - intention_limit=0, - scene="chat", - reader=HY_MEMORY_READER, - layers=["l1_raw"], + **{ + "user_ids": [value.user_id], + "agent_ids": [], + "include_sources": sources, + "limit": value.limit, + "min_score": 0.0, + "schema_min_score": 0.0, + "schema_limit": 0, + "intention_limit": 0, + "scene": "chat", + "reader": HY_MEMORY_READER, + "layers": ["l1_raw"], + **raw_lane_overrides(value), + }, + ) + hits = tuple( + hit for hit in parse_sdk_raw_search(raw) if value.passes_floor("raw", hit.score) ) - hits = parse_sdk_raw_search(raw) return tuple( SdkRawContextItem( raw_id=hit.raw_id, @@ -106,23 +157,37 @@ def sdk_context_router(dependencies: Sequence[Depends]) -> APIRouter: async def memory_lane() -> tuple[SdkMemoryContextItem, ...]: if "memory" not in value.kinds or not authority.memory_allowed: return () + overrides = value.search_overrides() raw = await asyncio.to_thread( client.search, value.query, - user_ids=[value.user_id], - agent_ids=[], - limit=value.limit, - min_score=sdk_relevance_floor(), - schema_min_score=sdk_relevance_floor(), - schema_limit=value.schema_limit, - intention_limit=value.intention_limit, - scene="normal", - reader=HY_MEMORY_READER, + **{ + "user_ids": [value.user_id], + "agent_ids": [], + "limit": value.limit, + "min_score": sdk_relevance_floor(), + "schema_min_score": sdk_relevance_floor(), + "schema_limit": value.schema_limit, + "intention_limit": value.intention_limit, + "scene": "normal", + "reader": HY_MEMORY_READER, + **overrides, + }, + ) + hits = ( + tuple( + hit + for hit in parse_sdk_lane_search( + raw, error_code="hy_memory_search_shape_unknown" + ) + if value.passes_floor(cast(SearchLane, hit.lane), hit.score) + ) + if overrides + else parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown") ) - hits = parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown") items: list[SdkMemoryContextItem] = [] for hit in hits: - record = await asyncio.to_thread(client.get, hit.memory_id) + record = await asyncio.to_thread(client.get, hit.memory_id, **hit_namespace(hit)) memory = None if record is None else HySdkMemoryRecord.from_sdk(record) # `get` takes no user: a hit is kept only if it is the caller's. if ( diff --git a/memhub/apps/core/src/one_memory/hy_sdk_types.py b/memhub/apps/core/src/one_memory/hy_sdk_types.py index d510a79..a700cec 100644 --- a/memhub/apps/core/src/one_memory/hy_sdk_types.py +++ b/memhub/apps/core/src/one_memory/hy_sdk_types.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from collections.abc import Mapping from datetime import UTC, datetime from typing import Literal, NamedTuple, cast @@ -391,6 +392,8 @@ class HySdkRawRecord(BaseModel): memory_at: int | None = None gmt_created: int | None = None custom: dict[str, object] = Field(default_factory=dict) + #: active, or shadow once pro extracted it; None when the SDK gave none. + status: str | None = None @classmethod def from_sdk(cls, raw: object) -> HySdkRawRecord: @@ -417,11 +420,34 @@ class HySdkRawRecord(BaseModel): raw.get("gmt_created") if isinstance(raw.get("gmt_created"), int) else None ), custom=dict(custom), + status=str(raw["status"]) if isinstance(raw.get("status"), str) else None, ) except (KeyError, TypeError, ValueError) as error: raise HySdkReadError("hy_memory_raw_shape_unknown") from error +_RAW_TURN_LINE = re.compile(r"^\[(user|assistant|tool|system)\]:\s?(.*)$") + + +def raw_turn_messages(content: str) -> tuple[tuple[str, str], ...]: + """The `(role, text)` messages of one SDK raw turn. + + The SDK stores a turn as `[user]: …\\n[assistant]: …`; a line without a + role prefix continues the message above it. Content that names no role + at all is one message of unknown role (`""`). + """ + messages: list[list[str]] = [] + for line in content.splitlines(): + match = _RAW_TURN_LINE.match(line) + if match is not None: + messages.append([match.group(1), match.group(2)]) + elif messages: + messages[-1][1] += "\n" + line + else: + messages.append(["", line]) + return tuple((role, text) for role, text in messages) + + class HySdkRawHit(HySdkRawRecord): score: float = Field(ge=0, le=1) @@ -516,6 +542,96 @@ class HySdkUserProfile(BaseModel): ) +class HySdkProfileVersion(BaseModel): + """One version on an L0 basic-info chain.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + memory_id: str + content: str + #: Conversation time as the SDK renders it on a chain (epoch or text). + memory_at: int | str | None = None + gmt_created: int | None = None + #: The chain head: the version the profile shows now. + current: bool = False + + +class HySdkProfileChain(BaseModel): + """One L0 basic-info head and its versions, latest first.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + memory_id: str + content: str + status: str + memory_at: int | None = None + gmt_created: int | None = None + #: The structured fields this head holds (`basic_info_kv`). + basic_info: dict[str, object] = Field(default_factory=dict) + versions: tuple[HySdkProfileVersion, ...] + + +class HySdkProfileVersions(BaseModel): + """`get_user_basic_info`: the long-term profile's L0 version chains.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + request_id: str + chains: tuple[HySdkProfileChain, ...] + elapsed_ms: float = Field(ge=0) + + @classmethod + def from_sdk(cls, raw: object) -> HySdkProfileVersions: + if not isinstance(raw, Mapping): + raise HySdkReadError("hy_memory_profile_versions_shape_unknown") + request_id = raw.get("request_id") + rows = raw.get("memories") + elapsed_ms = raw.get("elapsed_ms") + if ( + not isinstance(request_id, str) + or not isinstance(rows, list) + or not isinstance(elapsed_ms, int | float) + or isinstance(elapsed_ms, bool) + ): + raise HySdkReadError("hy_memory_profile_versions_shape_unknown") + chains: list[HySdkProfileChain] = [] + for row in rows: + if not isinstance(row, Mapping) or not isinstance(row.get("memory_id"), str): + raise HySdkReadError("hy_memory_profile_versions_shape_unknown") + head_id = str(row["memory_id"]) + chain = row.get("evolution_chain") + entries = chain if isinstance(chain, list) and chain else [{**row, "node_id": head_id}] + versions = tuple( + HySdkProfileVersion( + memory_id=str(entry.get("node_id") or entry.get("memory_id")), + content=str(entry.get("content") or ""), + memory_at=( + entry.get("memory_at") + if isinstance(entry.get("memory_at"), int | str) + and not isinstance(entry.get("memory_at"), bool) + else None + ), + gmt_created=_epoch(entry.get("gmt_created")), + current=str(entry.get("node_id") or entry.get("memory_id")) == head_id, + ) + for entry in entries + if isinstance(entry, Mapping) and (entry.get("node_id") or entry.get("memory_id")) + ) + basic_info = row.get("basic_info_kv") + chains.append( + HySdkProfileChain( + memory_id=head_id, + content=str(row.get("content") or ""), + status=str(row.get("status") or "active"), + memory_at=_epoch(row.get("memory_at")), + gmt_created=_epoch(row.get("gmt_created")), + basic_info=dict(basic_info) if isinstance(basic_info, Mapping) else {}, + versions=versions, + ) + ) + return cls(request_id=request_id, chains=tuple(chains), elapsed_ms=float(elapsed_ms)) + + def parse_sdk_search(raw: object, *, error_code: str) -> tuple[SemanticMemoryHit, ...]: rows = _reply_rows(raw, error_code) found: list[SemanticMemoryHit] = [] @@ -542,6 +658,120 @@ def parse_sdk_search(raw: object, *, error_code: str) -> tuple[SemanticMemoryHit return tuple(found) +def parse_sdk_lane_search(raw: object, *, error_code: str) -> tuple[SemanticMemoryHit, ...]: + """Every scene's search reply as hits that name their lane and namespace. + + `parse_sdk_search` reads the chat reply only and forgets which lane a hit + came from. A caller may now pick the scene and a floor per lane, so this + reads all three reply shapes: + + - chat: `memories` as `{schema, proactive, normal}` (or one list, normal); + - working/coding: `facts` and `experiences` of that variant; + - all: one fused `memories` list whose rows carry `scene`, `kind` and + their own lane's score (`lane_score`). The fused `score` is a rank + fusion (1/(60+rank)), not a relevance, so the lane score is reported. + + Highest score first, first occurrence of an id kept. + """ + if not isinstance(raw, Mapping): + raise HySdkReadError(error_code) + scene = raw.get("scene") + found: list[SemanticMemoryHit] = [] + if scene == "all": + rows = raw.get("memories") + if not isinstance(rows, list) or any(not isinstance(row, Mapping) for row in rows): + raise HySdkReadError(error_code) + for row in rows: + hit = _lane_hit(row, scene=str(row.get("scene") or "chat"), score_key="lane_score") + if hit is not None: + found.append(hit) + elif scene in ("productivity", "working", "coding"): + variant = raw.get("variant") if raw.get("variant") in {"working", "coding"} else None + lane = variant or (scene if scene in {"working", "coding"} else "working") + for key, kind in (("facts", "fact"), ("experiences", "experience")): + rows = raw.get(key) or [] + if not isinstance(rows, list) or any(not isinstance(row, Mapping) for row in rows): + raise HySdkReadError(error_code) + for row in rows: + hit = _lane_hit({**row, "kind": kind}, scene=lane, score_key="score") + if hit is not None: + found.append(hit) + else: + memories = raw.get("memories") + if isinstance(memories, Mapping): + if set(memories) != {"schema", "proactive", "normal"}: + raise HySdkReadError(error_code) + channels = tuple(memories.items()) + elif isinstance(memories, list): + channels = (("normal", memories),) + else: + raise HySdkReadError(error_code) + for lane, rows in channels: + if not isinstance(rows, list) or any(not isinstance(row, Mapping) for row in rows): + raise HySdkReadError(error_code) + for row in rows: + hit = _lane_hit(row, scene="chat", score_key="score", lane=str(lane)) + if hit is not None: + found.append(hit) + ordered: list[SemanticMemoryHit] = [] + seen: set[tuple[str, str, str | None]] = set() + for hit in sorted(found, key=lambda value: value.score, reverse=True): + identity = (hit.memory_id, hit.kind, hit.variant) + if identity in seen: + continue + seen.add(identity) + ordered.append(hit) + return tuple(ordered) + + +def _lane_hit( + row: Mapping[str, object], + *, + scene: str, + score_key: str, + lane: str | None = None, +) -> SemanticMemoryHit | None: + kind = str(row.get("kind")) if row.get("kind") in {"experience", "fact"} else "chat" + if kind == "chat": + memory_id = row.get("memory_id") or row.get("id") + variant = None + hit_lane = lane or "normal" + else: + id_key = "experience_id" if kind == "experience" else "fact_id" + memory_id = row.get("memory_id") or row.get(id_key) + variant = scene if scene in {"working", "coding"} else None + if variant is None: + return None + hit_lane = variant + if not isinstance(memory_id, str) or not memory_id: + return None + value = row.get(score_key) + if not isinstance(value, int | float) or isinstance(value, bool): + value = row.get("score") + score = ( + max(0.0, min(1.0, float(value))) + if isinstance(value, int | float) and not isinstance(value, bool) + else 0.0 + ) + if kind != "chat": + return SemanticMemoryHit( + memory_id=memory_id, score=score, lane=hit_lane, kind=kind, variant=variant + ) + content = row.get("content") + member_ids, formatting_complete = _formatted_member_ids(row) + return SemanticMemoryHit( + memory_id=memory_id, + score=score, + formatted_content=( + content.strip() if isinstance(content, str) and content.strip() else None + ), + member_ids=member_ids, + sdk_formatted=True, + formatting_complete=formatting_complete, + lane=hit_lane, + ) + + class LibraryHit(NamedTuple): """One keyword match in a library search window, with its namespace.""" diff --git a/memhub/apps/core/src/one_memory/library_search_window.py b/memhub/apps/core/src/one_memory/library_search_window.py index 8c17477..b2dd968 100644 --- a/memhub/apps/core/src/one_memory/library_search_window.py +++ b/memhub/apps/core/src/one_memory/library_search_window.py @@ -28,7 +28,10 @@ from one_memory.hy_sdk_types import LibraryHit DEFAULT_WINDOW_TTL_SECONDS = 300.0 DEFAULT_MAX_WINDOWS = 32 -WindowKey = tuple[str, str | None, tuple[str, ...], str, str] +#: (query, source, statuses, scene, user, other filters as sorted name/values pairs) +WindowKey = tuple[ + str, str | None, tuple[str, ...], str, str, tuple[tuple[str, tuple[str, ...]], ...] +] def window_key( @@ -37,8 +40,12 @@ def window_key( statuses: list[str], scene: str = "all", user_id: str = "local", + filters: dict[str, list[str]] | None = None, ) -> WindowKey: - return (query, source, tuple(sorted(set(statuses))), scene, user_id) + extra = tuple( + sorted((name, tuple(sorted(set(values)))) for name, values in (filters or {}).items()) + ) + return (query, source, tuple(sorted(set(statuses))), scene, user_id, extra) @dataclass(frozen=True) diff --git a/memhub/apps/core/src/one_memory/recall.py b/memhub/apps/core/src/one_memory/recall.py index fa275a4..2f85685 100644 --- a/memhub/apps/core/src/one_memory/recall.py +++ b/memhub/apps/core/src/one_memory/recall.py @@ -14,6 +14,7 @@ from pydantic import BaseModel, ConfigDict, Field from one_memory.hy_sdk_types import HySdkMemoryRecord from one_memory.memory_governance import GovernedMemory, MemoryStatus +from one_memory.sdk_search_options import SdkSearchOptions from one_memory.semantic_deadline import ( SemanticMemoryHit, SemanticSearch, @@ -50,7 +51,15 @@ class TruncationBudget(StrEnum): TIMEOUT = "timeout_budget" -class RecallRequest(BaseModel): +class RecallRequest(SdkSearchOptions): + """`POST /api/v1/recall`. + + The `SdkSearchOptions` fields, `schema_limit` and `intention_limit` pass + through to the SDK memory search (and the floor and row filters to the + raw fallback). Absent, recall makes the search it always made: the + relevance floor, schema and intention lanes sized to `limit`. + """ + model_config = ConfigDict(frozen=True, extra="forbid") query: str = Field(min_length=1, max_length=2_000) @@ -70,6 +79,17 @@ class RecallRequest(BaseModel): timeout_ms: int = Field(default=2_000, ge=50, le=30_000) #: Whose memories to recall. Omitted, it is `local`, the App's library. user_id: UserId = DEFAULT_USER_ID + #: The SDK's schema (L6) and proactive intention (L7) lanes. + schema_limit: int | None = Field(default=None, ge=0, le=20) + intention_limit: int | None = Field(default=None, ge=0, le=20) + + def search_overrides(self) -> Mapping[str, object]: + values = dict(super().search_overrides()) + if self.schema_limit is not None: + values["schema_limit"] = self.schema_limit + if self.intention_limit is not None: + values["intention_limit"] = self.intention_limit + return values class RecallHit(BaseModel): diff --git a/memhub/apps/core/src/one_memory/sdk_memory_qa.py b/memhub/apps/core/src/one_memory/sdk_memory_qa.py index 2f39645..b1ff68d 100644 --- a/memhub/apps/core/src/one_memory/sdk_memory_qa.py +++ b/memhub/apps/core/src/one_memory/sdk_memory_qa.py @@ -6,7 +6,7 @@ import asyncio import json from collections.abc import AsyncIterator, Awaitable, Callable from contextlib import aclosing -from typing import Any, Literal, Protocol +from typing import Any, Literal, Protocol, cast import httpx from pydantic import BaseModel, ConfigDict @@ -14,7 +14,7 @@ from pydantic import BaseModel, ConfigDict from one_memory.http_client import create_async_client from one_memory.hy_runtime_config import HY_MEMORY_READER from one_memory.hy_sdk_runtime import sdk_relevance_floor -from one_memory.hy_sdk_types import HySdkMemoryRecord, parse_sdk_search +from one_memory.hy_sdk_types import HySdkMemoryRecord, parse_sdk_lane_search, parse_sdk_search from one_memory.memory_qa import ( MemoryAnswerError, _DeferredSecretStore, @@ -31,6 +31,7 @@ from one_memory.model_profiles import ( ProviderKind, attach_disable_thinking, ) +from one_memory.sdk_search_options import SdkSearchFilters, SearchLane from one_memory.secrets import SecretStore, SecretStoreUnavailableError from one_memory.user_scope import owns @@ -96,22 +97,39 @@ class SdkMemoryAnswerer: question: str, *, user_id: str = "local", + filters: SdkSearchFilters | None = None, ) -> tuple[SdkAskMemory, ...]: + """The user's active memories nearest the question. + + `filters` are a caller's pass-through search options; absent, the + search is the one ask always made. + """ client = await self.client_handle.client() + overrides = filters.search_overrides() if filters is not None else {} raw = await asyncio.to_thread( client.search, question, - user_ids=[user_id], - agent_ids=[], - limit=12, - min_score=sdk_relevance_floor(), - schema_min_score=sdk_relevance_floor(), - schema_limit=12, - intention_limit=12, - scene="normal", - reader=HY_MEMORY_READER, + **{ + "user_ids": [user_id], + "agent_ids": [], + "limit": 12, + "min_score": sdk_relevance_floor(), + "schema_min_score": sdk_relevance_floor(), + "schema_limit": 12, + "intention_limit": 12, + "scene": "normal", + "reader": HY_MEMORY_READER, + **overrides, + }, ) - hits = parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown") + if filters is not None and overrides: + hits = tuple( + hit + for hit in parse_sdk_lane_search(raw, error_code="hy_memory_search_shape_unknown") + if filters.passes_floor(cast(SearchLane, hit.lane), hit.score) + ) + else: + hits = parse_sdk_search(raw, error_code="hy_memory_search_shape_unknown") found: list[SdkAskMemory] = [] seen: set[str] = set() for hit in hits: @@ -134,8 +152,9 @@ class SdkMemoryAnswerer: question: str, *, user_id: str = "local", + filters: SdkSearchFilters | None = None, ) -> SdkMemoryAnswer: - candidates = await self.retrieve(question, user_id=user_id) + candidates = await self.retrieve(question, user_id=user_id, filters=filters) if not candidates: return SdkMemoryAnswer(answer=_NO_GROUNDING, memories=(), mode="empty") profile = await self._active_profile() @@ -166,8 +185,9 @@ class SdkMemoryAnswerer: question: str, *, user_id: str = "local", + filters: SdkSearchFilters | None = None, ) -> AsyncIterator[dict[str, object]]: - candidates = await self.retrieve(question, user_id=user_id) + candidates = await self.retrieve(question, user_id=user_id, filters=filters) await self._require_current(candidates) yield {"type": "recall", "memories": candidates} if not candidates: diff --git a/memhub/apps/core/src/one_memory/sdk_search_options.py b/memhub/apps/core/src/one_memory/sdk_search_options.py new file mode 100644 index 0000000..2bc2a54 --- /dev/null +++ b/memhub/apps/core/src/one_memory/sdk_search_options.py @@ -0,0 +1,179 @@ +"""SDK search parameters a host may pass through to `client.search`. + +`POST /api/v2/context/search`, `POST /api/v1/recall` and the v2 ask routes +used to fix every SDK search knob, so a host that wanted a stricter relevance +floor filtered the answer itself. These fields are the public `search` +parameters of hy-memory 2.0.9 that are safe to hand a caller. Every field is +optional: absent, the route keeps exactly the call it made before. + +Never exposed: `user_ids` (a route searches its one scoped `user_id`), +`embed_override` and `query_embedding` (they bypass the configured embedder), +and `request_id` (Core's own correlation id). + +A caller's floor is also applied to what Core returns, not only handed to the +SDK: a hit below the floor that names its lane is dropped (`passes_floor`). +With `min_score` alone that floor covers every lane, so `min_score=0.9` +answers only results scoring at least 0.9. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Annotated, Literal + +from pydantic import BaseModel, ConfigDict, Field, StrictBool, model_validator + +#: The SDK's `MemoryLayer` values (hy-memory 2.0.9). +MemoryLayerName = Literal[ + "l0_basic_info", + "l1_raw", + "l2_fact", + "l3_summary", + "l4_identity", + "l5_knowledge", + "l6_schema", + "l7_intention", +] +MEMORY_LAYERS: tuple[str, ...] = ( + "l0_basic_info", + "l1_raw", + "l2_fact", + "l3_summary", + "l4_identity", + "l5_knowledge", + "l6_schema", + "l7_intention", +) +#: The SDK's `MemoryStatus` values. +MEMORY_STATUSES: tuple[str, ...] = ( + "active", + "superseded", + "negated", + "conflicted", + "shadow", + "archived", +) +SearchScene = Literal["chat", "working", "coding", "all"] +#: Which recall lane produced a hit, for the floor that applies to it. +SearchLane = Literal["normal", "schema", "proactive", "working", "coding", "raw"] + +Score = Annotated[float, Field(ge=0.0, le=1.0)] +SourceName = Annotated[str, Field(min_length=1, max_length=128)] +ScopeId = Annotated[str, Field(min_length=1, max_length=512)] + + +class SdkSearchFilters(BaseModel): + """Filters and floors every search route accepts (recall, context, ask).""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + #: Chat lane floor; with nothing more specific, every lane's. + min_score: Score | None = None + #: Schema (L6) lane floor; defaults to `min_score` when only that is set. + schema_min_score: Score | None = None + #: False skips the SDK's reranking stages for this call. + rerank: StrictBool | None = None + include_sources: tuple[SourceName, ...] | None = Field(default=None, max_length=64) + exclude_sources: tuple[SourceName, ...] | None = Field(default=None, max_length=64) + #: Unix seconds: only memories created at or after it. + created_after: float | None = Field(default=None, ge=0) + session_ids: tuple[ScopeId, ...] | None = Field(default=None, max_length=64) + agent_ids: tuple[ScopeId, ...] | None = Field(default=None, max_length=64) + #: Only these layers (pro/ultra; lite always searches raw). + layers: tuple[MemoryLayerName, ...] | None = Field(default=None, min_length=1, max_length=8) + + def filter_kwargs(self) -> dict[str, object]: + """The SDK keyword arguments this request sets, and only those.""" + values: dict[str, object] = {} + for name in ( + "min_score", + "schema_min_score", + "rerank", + "created_after", + ): + value = getattr(self, name) + if value is not None: + values[name] = value + for name in ("include_sources", "exclude_sources", "session_ids", "agent_ids", "layers"): + value = getattr(self, name) + if value is not None: + values[name] = list(dict.fromkeys(value)) + if self.min_score is not None and self.schema_min_score is None: + values["schema_min_score"] = self.min_score + return values + + def lane_floor(self, lane: SearchLane) -> float | None: + """The caller's floor for one lane, or None when the caller set none.""" + if lane == "schema": + return self.schema_min_score if self.schema_min_score is not None else self.min_score + return self.min_score + + def passes_floor(self, lane: SearchLane, score: float) -> bool: + floor = self.lane_floor(lane) + return floor is None or score >= floor + + def search_overrides(self) -> Mapping[str, object]: + return self.filter_kwargs() + + +class SdkSearchOptions(SdkSearchFilters): + """Every SDK search knob a recall or context caller may set.""" + + #: working/coding fact-layer floor (SDK default 0.30). + working_fact_min_score: Score | None = None + #: working/coding experience-layer floor (SDK default fact floor + 0.10). + working_experience_min_score: Score | None = None + reader: Literal["legacy", "hybrid"] | None = None + #: chat (the default), working, coding, or all three fused. + scene: SearchScene | None = None + #: agentic spends one model call to classify and rewrite the query. + search_mode: Literal["default", "agentic"] | None = None + #: working/coding lanes only; `branch` needs `workspace_id`. + workspace_id: ScopeId | None = None + branch: ScopeId | None = None + + @model_validator(mode="after") + def branch_needs_workspace(self) -> SdkSearchOptions: + if self.branch is not None and self.workspace_id is None: + raise ValueError("branch requires workspace_id") + return self + + def search_overrides(self) -> Mapping[str, object]: + values = self.filter_kwargs() + for name in ( + "working_fact_min_score", + "working_experience_min_score", + "reader", + "scene", + "search_mode", + "workspace_id", + "branch", + ): + value = getattr(self, name) + if value is not None: + values[name] = value + if self.min_score is not None and self.working_fact_min_score is None: + values["working_fact_min_score"] = self.min_score + if self.min_score is not None and self.working_experience_min_score is None: + values["working_experience_min_score"] = self.min_score + return values + + def lane_floor(self, lane: SearchLane) -> float | None: + if lane in {"working", "coding"}: + explicit = [ + value + for value in (self.working_fact_min_score, self.working_experience_min_score) + if value is not None + ] + # Facts and experiences share one score field on a fused hit, so + # the looser of the two floors is the one Core can enforce. + return min(explicit) if explicit else self.min_score + return super().lane_floor(lane) + + +def split_csv(values: list[str] | None) -> list[str]: + """Query values given repeated (`a&a`) or comma separated (`a,b`).""" + found: list[str] = [] + for value in values or (): + found.extend(part.strip() for part in value.split(",") if part.strip()) + return list(dict.fromkeys(found)) diff --git a/memhub/apps/core/src/one_memory/semantic_deadline.py b/memhub/apps/core/src/one_memory/semantic_deadline.py index 1b2f883..a3c14fe 100644 --- a/memhub/apps/core/src/one_memory/semantic_deadline.py +++ b/memhub/apps/core/src/one_memory/semantic_deadline.py @@ -27,6 +27,11 @@ class SemanticMemoryHit: member_ids: tuple[str, ...] = () sdk_formatted: bool = False formatting_complete: bool = False + #: The recall lane that produced the hit (`sdk_search_options.SearchLane`) + #: and the namespace to read it back from. Defaults are a chat memory. + lane: str = "normal" + kind: str = "chat" + variant: str | None = None def as_dict(self) -> dict[str, object]: return { diff --git a/memhub/apps/core/tests/TEST_INDEX.md b/memhub/apps/core/tests/TEST_INDEX.md index b29b94a..bbaf9c7 100644 --- a/memhub/apps/core/tests/TEST_INDEX.md +++ b/memhub/apps/core/tests/TEST_INDEX.md @@ -248,6 +248,7 @@ | `test_schema_gate.py` | `contract` | 覆盖 fresh database records the build schema、future schema refuses to open、legacy database without a ledger is left untouched 场景。 | | `test_schema_migrations.py` | `contract` | 覆盖 evidence store adds source ordinal to previous schema、import store migrates previous job and error schema、old active job index lets a sync sit beside paused history 等 4 个场景。 | | `test_schema_ownership_gate.py` | `contract` | A1-03/A1-04: schema ownership, and the hard part of telling DDL from reads. | +| `test_sdk_api_surface.py` | `contract` | The SDK surface MemHub exposes: search pass-through, layers, raw, v2 aliases. | | `test_sdk_client_construction.py` | `contract` | SDK clients of one home are constructed one at a time. | | `test_sdk_data_plane_migration.py` | `contract` | SDK-authority migration ledger, aliases and atomic cutover marker. | | `test_sdk_memory_qa.py` | `contract` | 覆盖 sdk ask propagates sdk search failure without legacy fallback、sdk ask revalidates memory before returning extract、sdk ask uses native sdk memory without governance fields 场景。 | @@ -311,4 +312,4 @@ | `test_workspace_recall.py` | `contract` | Workspace memories were unreachable from every host; these pin the way in. | | `test_write_queue_lease_renewal.py` | `contract` | Rule 8: a lease must outlive the work it covers, and be renewed while it runs. | -共 306 个测试文件。 +共 307 个测试文件。 diff --git a/memhub/apps/core/tests/core_test_classification.py b/memhub/apps/core/tests/core_test_classification.py index a441cc5..23d2315 100644 --- a/memhub/apps/core/tests/core_test_classification.py +++ b/memhub/apps/core/tests/core_test_classification.py @@ -14,8 +14,8 @@ from pathlib import Path import pytest PRIMARY_MARKERS = ("unit", "contract", "integration", "performance", "live") -EXPECTED_TEST_FILE_COUNT = 306 -EXPECTED_TEST_FILES_SHA256 = "e616dcf9fea288bf15a4445045f7b4278517f3304091c04fffd55be80308d481" +EXPECTED_TEST_FILE_COUNT = 307 +EXPECTED_TEST_FILES_SHA256 = "96a9ed11e358791723fb5ad827f72da2e9caeb673841119a5d9856a2d76fd5b1" # Pure transformations, validation models, parity checks and state machines. # This list is intentionally small: uncertain files default to contract. diff --git a/memhub/apps/core/tests/test_sdk_api_surface.py b/memhub/apps/core/tests/test_sdk_api_surface.py new file mode 100644 index 0000000..a907328 --- /dev/null +++ b/memhub/apps/core/tests/test_sdk_api_surface.py @@ -0,0 +1,744 @@ +"""The SDK surface MemHub exposes: search pass-through, layers, raw, v2 aliases. + +P2-1 (R8): `context/search`, `recall` and the v2 ask routes pass the public +`client.search` knobs through, validated, and keep a caller's floor on what +they return. P2-2 (R9): the library honours `layers`, `statuses`, `scene` and +the source/session/agent filters, and counts memories per layer. P2-4: raw +lists the SDK's L1 raw turns, active and shadow. P2-3: v2 aliases of the +source and import routes, per user. Profile versions read the L0 chain. + +Every route stays scoped to its one `user_id`; the stand-in client records +each call so a test can see exactly what reached the SDK. +""" + +from __future__ import annotations + +import asyncio +import hashlib +import sqlite3 +from contextlib import asynccontextmanager +from pathlib import Path +from uuid import uuid4 + +import pytest +from httpx import ASGITransport, AsyncClient +from pydantic import SecretStr + +from one_memory.api import create_app +from one_memory.config import Settings +from one_memory.imports.sources.registry import build_source_catalog +from one_memory.secrets import InMemorySecretStore + +TOKEN = "sdk-surface-test" + +CHAT_ROWS = { + "local": [ + ("m-high", "local likes green tea", 0.95, "l2_fact", "hermes"), + ("m-mid", "local walks at dawn", 0.6, "l2_fact", "codex"), + ], + "wife": [("w-high", "wife likes coffee", 0.92, "l2_fact", "hermes")], +} +RAW_ROWS = { + "local": [ + { + "memory_id": "raw-1", + "content": "[user]: 我叫费天翔\n[assistant]: 好的,记住了。", + "status": "shadow", + "session_id": "s-1", + "source": "hermes", + "memory_at": 1_790_000_000, + "gmt_created": 1_790_000_100, + }, + { + "memory_id": "raw-2", + "content": "[user]: 你好", + "status": "active", + "session_id": "s-2", + "source": "codex", + "memory_at": 1_790_100_000, + "gmt_created": 1_790_100_100, + }, + ], + "wife": [ + { + "memory_id": "raw-w", + "content": "[user]: I am the wife", + "status": "active", + "session_id": "s-w", + "source": "hermes", + "memory_at": 1_790_000_000, + "gmt_created": 1_790_000_000, + } + ], +} + + +def _chat(user_id: str, memory_id: str, content: str, layer: str, source: str) -> dict: + return { + "memory_id": memory_id, + "content": content, + "layer": layer, + "status": "active", + "user_id": user_id, + "agent_id": "all-apps", + "session_id": "s-1", + "source": source, + "tags": [], + "gmt_created": 1_790_000_000, + } + + +class RecordingSdk: + """The public client surface, recording every call's keyword arguments.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + + def _record(self, name: str, kwargs: dict) -> None: + self.calls.append((name, dict(kwargs))) + + def last(self, name: str) -> dict: + return [kwargs for called, kwargs in self.calls if called == name][-1] + + def all(self, name: str) -> list[dict]: + return [kwargs for called, kwargs in self.calls if called == name] + + # The SDK answers rows below a caller's floor in some lanes (a schema + # floor, a fused rank score), so Core must filter what it returns. + def search(self, query, **kwargs): + self._record("search", {"query": query, **kwargs}) + [user_id] = kwargs["user_ids"] + if kwargs.get("layers") == ["l1_raw"]: + hits = [ + {**row, "user_id": user_id, "agent_id": "all-apps", "score": score} + for row, score in zip(RAW_ROWS.get(user_id, []), (0.97, 0.3), strict=False) + ] + return {"request_id": "s", "memories": hits} + rows = [ + {**_chat(user_id, mid, text, layer, source), "score": score} + for mid, text, score, layer, source in CHAT_ROWS.get(user_id, []) + ] + scene = kwargs.get("scene") + if scene == "all": + fused = [ + {**row, "scene": "chat", "kind": "chat", "lane_score": row["score"], "score": 0.016} + for row in rows + ] + return {"request_id": "s", "scene": "all", "memories": fused, "lanes": {}, "errors": {}} + if scene in {"working", "coding"}: + return { + "request_id": "s", + "scene": "productivity", + "variant": scene, + "facts": [ + {"fact_id": "f-1", "statement": "use uv", "score": 0.91, "user_id": user_id} + ], + "experiences": [], + } + return { + "request_id": "s", + "scene": "normal", + "memories": {"normal": rows, "schema": [], "proactive": []}, + } + + def get(self, memory_id, kind=None, variant=None): + self._record("get", {"memory_id": memory_id, "kind": kind, "variant": variant}) + if kind == "fact": + return { + "kind": "fact", + "variant": variant, + "fact_id": memory_id, + "statement": "use uv", + "status": "active", + "user_id": "local", + } + for user_id, rows in CHAT_ROWS.items(): + for mid, text, _score, layer, source in rows: + if mid == memory_id: + return {**_chat(user_id, mid, text, layer, source), "kind": "chat"} + return None + + def list_memories(self, **kwargs): + self._record("list_memories", kwargs) + user_id = kwargs["user_id"] + layers = kwargs.get("layers") + if layers == ["l1_raw"]: + statuses = set(kwargs.get("statuses") or ["active"]) + rows = [ + {**row, "layer": "l1_raw", "user_id": user_id, "agent_id": "all-apps"} + for row in RAW_ROWS.get(user_id, []) + if row["status"] in statuses + and ( + not kwargs.get("include_sources") or row["source"] in kwargs["include_sources"] + ) + and (not kwargs.get("session_ids") or row["session_id"] in kwargs["session_ids"]) + ] + return {"vdb": {"memories": rows, "total": len(rows)}} + rows = [_chat(user_id, *row[:2], *row[3:]) for row in CHAT_ROWS.get(user_id, [])] + if layers: + rows = [row for row in rows if row["layer"] in layers] + if "l0_basic_info" in layers: + rows.append(_chat(user_id, "l0-1", "name is Fei", "l0_basic_info", "hermes")) + if kwargs.get("scene") == "all": + return { + "scene": "all", + "memories": [{**row, "scene": "chat", "kind": "chat"} for row in rows], + "total": len(rows), + "totals": {"chat": len(rows), "working": 0, "coding": 0}, + "errors": {}, + } + return {"vdb": {"memories": rows, "total": len(rows)}} + + def memory_stats(self, *, user_id, statuses=None): + return {"total": 0, "request_id": "stats", "elapsed_ms": 1.0} + + def get_user_basic_info(self, *, user_id): + self._record("get_user_basic_info", {"user_id": user_id}) + return { + "request_id": f"basic-{user_id}", + "elapsed_ms": 1.5, + "memories": [ + { + "memory_id": f"{user_id}-l0-head", + "content": "The user's name is Fei.", + "layer": "l0_basic_info", + "status": "active", + "memory_at": 1_790_000_100, + "gmt_created": 1_790_000_100, + "tags": [], + "basic_info_kv": {"name": "Fei"}, + "evolution_chain": [ + { + "node_id": f"{user_id}-l0-head", + "content": "The user's name is Fei.", + "memory_at": "2026-10-31 10:00", + "gmt_created": 1_790_000_100, + "layer": "l0_basic_info", + }, + { + "node_id": f"{user_id}-l0-old", + "content": "The user's name is 费天翔.", + "memory_at": "2026-10-30 10:00", + "gmt_created": 1_790_000_000, + "layer": "l0_basic_info", + }, + ], + } + ], + } + + +class Handle: + def __init__(self, sdk: RecordingSdk) -> None: + self.sdk = sdk + + async def client(self) -> RecordingSdk: + return self.sdk + + async def close(self) -> None: + return None + + +def application(tmp_path: Path, *, home: Path | None = None): + settings = Settings( + data_dir=tmp_path / "data", + source_home=home or tmp_path, + api_token=SecretStr(TOKEN), + imports_enabled=home is not None, + automatic_memory_processing=False, + live_sync_enabled=False, + ) + return create_app( + settings, + secret_store=InMemorySecretStore(), + source_catalog=( + build_source_catalog(home=home, environment={}, platform="darwin") + if home is not None + else None + ), + ) + + +def client_for(app) -> AsyncClient: + return AsyncClient( + transport=ASGITransport(app=app), + base_url="http://test", + headers={"Authorization": f"Bearer {TOKEN}"}, + ) + + +async def _model_ready(_app) -> bool: + return True + + +def run_with_sdk(tmp_path, monkeypatch, scenario) -> None: + async def main() -> None: + app = application(tmp_path) + sdk = RecordingSdk() + + @asynccontextmanager + async def write_client(): + yield sdk + + async with app.router.lifespan_context(app): + app.state.sdk_client_handle = Handle(sdk) + monkeypatch.setattr(app.state.sdk_write_coordinator, "use_client", write_client) + monkeypatch.setattr("one_memory.api._recall_model_ready", _model_ready) + async with client_for(app) as client: + await scenario(app, sdk, client) + + asyncio.run(main()) + + +PASS_THROUGH = { + "min_score": 0.5, + "schema_min_score": 0.6, + "working_fact_min_score": 0.3, + "working_experience_min_score": 0.45, + "schema_limit": 4, + "intention_limit": 2, + "rerank": False, + "reader": "legacy", + "scene": "chat", + "search_mode": "agentic", + "include_sources": ["hermes", "codex"], + "exclude_sources": ["cursor"], + "created_after": 1_780_000_000, + "session_ids": ["s-1"], + "agent_ids": ["all-apps"], + "workspace_id": "ws-1", + "branch": "main", + "layers": ["l2_fact", "l0_basic_info"], +} +#: How each field reaches `client.search` (lists stay lists). +EXPECTED_SEARCH = {**PASS_THROUGH} + + +def _memory_search(sdk: RecordingSdk) -> dict: + return [call for call in sdk.all("search") if call.get("layers") != ["l1_raw"]][-1] + + +@pytest.mark.fresh_sdk +def test_every_context_search_option_reaches_the_sdk_and_absent_ones_keep_the_old_call( + tmp_path, monkeypatch +) -> None: + async def scenario(app, sdk, client): + baseline = await client.post( + "/api/v2/context/search", json={"query": "tea", "kinds": ["memory"]} + ) + assert baseline.status_code == 200, baseline.text + before = _memory_search(sdk) + assert before["min_score"] == before["schema_min_score"] == 0.35 + assert before["scene"] == "normal" and before["reader"] == "hybrid" + assert before["schema_limit"] == 3 and before["intention_limit"] == 1 + for absent in ("rerank", "search_mode", "layers", "session_ids", "created_after"): + assert absent not in before + + response = await client.post( + "/api/v2/context/search", + json={"query": "tea", "kinds": ["memory"], **PASS_THROUGH}, + ) + assert response.status_code == 200, response.text + sent = _memory_search(sdk) + for name, value in EXPECTED_SEARCH.items(): + assert sent[name] == value, name + assert sent["user_ids"] == ["local"] + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_every_recall_and_ask_option_reaches_the_sdk(tmp_path, monkeypatch) -> None: + async def scenario(app, sdk, client): + recalled = await client.post( + "/api/v1/recall", json={"query": "tea", "user_id": "wife", **PASS_THROUGH} + ) + assert recalled.status_code == 200, recalled.text + sent = _memory_search(sdk) + for name, value in EXPECTED_SEARCH.items(): + assert sent[name] == value, name + assert sent["user_ids"] == ["wife"] + + ask_options = { + name: value + for name, value in PASS_THROUGH.items() + if name + not in { + "working_fact_min_score", + "working_experience_min_score", + "schema_limit", + "intention_limit", + "reader", + "scene", + "search_mode", + "workspace_id", + "branch", + } + } + asked = await client.post( + "/api/v2/memories/ask", json={"question": "tea?", "user_id": "wife", **ask_options} + ) + assert asked.status_code == 200, asked.text + sent = _memory_search(sdk) + for name, value in ask_options.items(): + assert sent[name] == value, name + assert sent["user_ids"] == ["wife"] and sent["scene"] == "normal" + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_min_score_is_enforced_on_what_core_returns(tmp_path, monkeypatch) -> None: + """Acceptance item 9: `min_score=0.9` answers only results scoring ≥ 0.9.""" + + async def scenario(app, sdk, client): + context = await client.post( + "/api/v2/context/search", json={"query": "tea", "min_score": 0.9} + ) + assert context.status_code == 200, context.text + items = context.json()["items"] + assert items and all(item["score"] >= 0.9 for item in items) + assert {item.get("memory_id") or item.get("raw_id") for item in items} == { + "m-high", + "raw-1", + } + raw_call = sdk.last("search") if sdk.last("search").get("layers") else None + assert raw_call is None or raw_call["min_score"] == 0.9 + + # The fused scene ranks by 1/(60+rank); the lane's own score is the floor's. + fused = await client.post( + "/api/v2/context/search", + json={"query": "tea", "kinds": ["memory"], "scene": "all", "min_score": 0.9}, + ) + assert [item["memory_id"] for item in fused.json()["items"]] == ["m-high"] + assert fused.json()["items"][0]["score"] == 0.95 + + recalled = await client.post("/api/v1/recall", json={"query": "tea", "min_score": 0.9}) + assert recalled.status_code == 200, recalled.text + assert [hit["memory_id"] for hit in recalled.json()["hits"]] == ["m-high"] + + # A working-scene fact is read back from its own namespace. + working = await client.post( + "/api/v2/context/search", + json={"query": "uv", "kinds": ["memory"], "scene": "working", "min_score": 0.9}, + ) + assert [item["memory_id"] for item in working.json()["items"]] == ["f-1"] + assert sdk.last("get") == {"memory_id": "f-1", "kind": "fact", "variant": "working"} + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +@pytest.mark.parametrize( + "extra", + [ + {"min_score": 1.5}, + {"min_score": -0.1}, + {"schema_min_score": "high"}, + {"scene": "normal"}, + {"search_mode": "fast"}, + {"reader": "vector"}, + {"layers": ["l9_nothing"]}, + {"layers": []}, + {"branch": "main"}, + {"rerank": "no"}, + {"created_after": -1}, + {"user_ids": ["wife"]}, + {"embed_override": {"model": "x"}}, + {"query_embedding": [0.1]}, + {"request_id": "r-1"}, + ], +) +def test_out_of_range_and_unexposed_search_fields_answer_422(tmp_path, extra) -> None: + async def scenario(): + app = application(tmp_path) + async with app.router.lifespan_context(app), client_for(app) as client: + for path, body in ( + ("/api/v2/context/search", {"query": "q"}), + ("/api/v1/recall", {"query": "q"}), + ("/api/v2/memories/ask", {"question": "q"}), + ): + response = await client.post(path, json={**body, **extra}) + assert response.status_code == 422, (path, response.text) + + asyncio.run(scenario()) + + +@pytest.mark.fresh_sdk +def test_raw_lane_takes_the_callers_sources_only_inside_its_authority( + tmp_path, monkeypatch +) -> None: + async def scenario(app, sdk, client): + response = await client.post( + "/api/v2/context/search", + json={ + "query": "tea", + "kinds": ["raw"], + "include_sources": ["hermes"], + "session_ids": ["s-1"], + "created_after": 1_700_000_000, + }, + ) + assert response.status_code == 200, response.text + raw_call = sdk.last("search") + assert raw_call["layers"] == ["l1_raw"] + assert raw_call["include_sources"] == ["hermes"] + assert raw_call["session_ids"] == ["s-1"] + assert raw_call["created_after"] == 1_700_000_000 + # Memory-lane knobs never reach the raw search. + assert raw_call["scene"] == "chat" and raw_call["schema_limit"] == 0 + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_library_honours_layers_statuses_and_filters(tmp_path, monkeypatch) -> None: + """Acceptance item 10: `layer=l0_basic_info` returns only L0 records.""" + + async def scenario(app, sdk, client): + page = await client.get("/api/v2/memory-library", params={"layer": "l0_basic_info"}) + assert page.status_code == 200, page.text + assert {memory["layer"] for memory in page.json()["memories"]} == {"l0_basic_info"} + sent = sdk.last("list_memories") + assert sent["layers"] == ["l0_basic_info"] and sent["scene"] is None + + await client.get( + "/api/v2/memory-library", + params=[ + ("layers", "l2_fact,l6_schema"), + ("layers", "l7_intention"), + ("statuses", "active,superseded"), + ("session_ids", "s-1"), + ("agent_ids", "all-apps"), + ("exclude_sources", "cursor"), + ("user_id", "wife"), + ], + ) + sent = sdk.last("list_memories") + assert sent["layers"] == ["l2_fact", "l6_schema", "l7_intention"] + assert sent["statuses"] == ["active", "superseded"] + assert sent["session_ids"] == ["s-1"] and sent["agent_ids"] == ["all-apps"] + assert sent["exclude_sources"] == ["cursor"] and sent["user_id"] == "wife" + + # Without a layer the page is what it always was. + await client.get("/api/v2/memory-library") + sent = sdk.last("list_memories") + assert sent["scene"] == "all" and "layers" not in sent + + searched = await client.get( + "/api/v2/memory-library", params={"query": "tea", "layers": "l2_fact"} + ) + assert searched.status_code == 200, searched.text + assert sdk.last("search")["layers"] == ["l2_fact"] + + for params in ( + {"layers": "l9_nothing"}, + {"statuses": "gone"}, + {"layers": "l2_fact", "scene": "working"}, + ): + refused = await client.get("/api/v2/memory-library", params=params) + assert refused.status_code == 422, (params, refused.text) + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_layer_counts_read_each_layer_for_one_user(tmp_path, monkeypatch) -> None: + async def scenario(app, sdk, client): + counts = await client.get("/api/v2/memory-library/layers", params={"user_id": "wife"}) + assert counts.status_code == 200, counts.text + body = counts.json() + assert body["layers"]["l1_raw"] == 1 + assert body["layers"]["l2_fact"] == 1 + assert body["layers"]["l0_basic_info"] == 1 + assert body["total"] == sum(body["layers"].values()) + calls = sdk.all("list_memories") + assert {call["user_id"] for call in calls} == {"wife"} + [raw_call] = [call for call in calls if call["layers"] == ["l1_raw"]] + assert raw_call["statuses"] == ["active", "shadow"] + assert ( + await client.get("/api/v2/memory-library/layers", params={"layer": "x"}) + ).status_code == 422 + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_raw_lists_active_and_shadow_turns_with_roles_and_hashes(tmp_path, monkeypatch) -> None: + async def scenario(app, sdk, client): + page = await client.get("/api/v2/raw") + assert page.status_code == 200, page.text + body = page.json() + assert body["total"] == 2 and len(body["items"]) == 2 + sent = sdk.last("list_memories") + assert sent["layers"] == ["l1_raw"] and sent["statuses"] == ["active", "shadow"] + first = next(item for item in body["items"] if item["raw_id"] == "raw-1") + assert first["status"] == "shadow" + assert [message["role"] for message in first["messages"]] == ["user", "assistant"] + assert first["messages"][0]["content"] == "我叫费天翔" + assert first["role"] is None + expected = hashlib.sha256(first["content"].encode()).hexdigest() + assert first["content_hash"] == f"sha256:{expected}" + assert first["session_id"] == "s-1" and first["source"] == "hermes" + assert first["created"].startswith("2026-") + second = next(item for item in body["items"] if item["raw_id"] == "raw-2") + assert second["role"] == "user" + + await client.get( + "/api/v2/raw", + params=[("source", "hermes,codex"), ("session_id", "s-1"), ("order", "asc")], + ) + sent = sdk.last("list_memories") + assert sent["include_sources"] == ["hermes", "codex"] + assert sent["session_ids"] == ["s-1"] and sent["order"] == "asc" + + windowed = (await client.get("/api/v2/raw", params={"after": 1_790_050_000})).json() + assert windowed["time_filtered"] is True + assert [item["raw_id"] for item in windowed["items"]] == ["raw-2"] + + wife = (await client.get("/api/v2/raw", params={"user_id": "wife"})).json() + assert [item["raw_id"] for item in wife["items"]] == ["raw-w"] + assert sdk.last("list_memories")["user_id"] == "wife" + assert (await client.get("/api/v2/raw", params={"statuses": "bogus"})).status_code == 422 + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_profile_versions_read_the_l0_chain_for_one_user(tmp_path, monkeypatch) -> None: + async def scenario(app, sdk, client): + response = await client.get("/api/v2/user-profile/versions", params={"user_id": "wife"}) + assert response.status_code == 200, response.text + body = response.json() + assert body["user_id"] == "wife" + assert sdk.last("get_user_basic_info") == {"user_id": "wife"} + [chain] = body["chains"] + assert chain["memory_id"] == "wife-l0-head" + assert chain["basic_info"] == {"name": "Fei"} + assert [(v["memory_id"], v["current"]) for v in chain["versions"]] == [ + ("wife-l0-head", True), + ("wife-l0-old", False), + ] + local = (await client.get("/api/v2/user-profile/versions")).json() + assert local["user_id"] == "local" + assert sdk.last("get_user_basic_info") == {"user_id": "local"} + refused = await client.get("/api/v2/user-profile/versions", params={"owner": "wife"}) + assert refused.status_code == 422 + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +@pytest.mark.fresh_sdk +def test_graph_takes_layers_and_session_filters(tmp_path, monkeypatch) -> None: + async def scenario(app, sdk, client): + response = await client.get( + "/api/v2/memories/graph", + params={"layers": "l2_fact", "session_ids": "s-1", "user_id": "wife"}, + ) + assert response.status_code == 200, response.text + sent = sdk.last("list_memories") + assert sent["layers"] == ["l2_fact"] and sent["scene"] is None + assert sent["session_ids"] == ["s-1"] and sent["user_id"] == "wife" + + run_with_sdk(tmp_path, monkeypatch, scenario) + + +def _hermes_state_db(root: Path, text: str) -> None: + root.mkdir(parents=True, exist_ok=True) + connection = sqlite3.connect(root / "state.db") + connection.execute( + "CREATE TABLE messages (id INTEGER PRIMARY KEY, session_id TEXT, role TEXT," + " content TEXT, timestamp REAL, tool_name TEXT, active INTEGER)" + ) + connection.executemany( + "INSERT INTO messages VALUES (?, ?, ?, ?, 1790000000, NULL, 1)", + [(1, "session-1", "user", text), (2, "session-1", "assistant", "ok")], + ) + connection.commit() + connection.close() + + +@pytest.mark.fresh_sdk +def test_v2_source_and_import_aliases_behave_like_v1_per_user(tmp_path) -> None: + async def scenario(): + home = tmp_path / "home" + hermes = home / ".hermes" + wife = hermes / "profiles" / "wife" + _hermes_state_db(hermes, "I am the husband") + _hermes_state_db(wife, "I am the wife") + app = application(tmp_path, home=home) + async with app.router.lifespan_context(app), client_for(app) as client: + location = (await client.post("/api/v1/sources/hermes/location")).json() + local = await client.post( + "/api/v1/sources/connect", + json={ + "source_ids": ["hermes"], + "request_id": str(uuid4()), + "source_roots": {"hermes": location["root_id"]}, + "enable_sync": True, + }, + ) + assert local.status_code == 200, local.text + connected = await client.post( + "/api/v1/sources/connect", + json={ + "source_ids": ["hermes"], + "request_id": str(uuid4()), + "user_id": "wife", + "root": str(wife), + }, + ) + assert connected.status_code == 200, connected.text + for job in (*local.json()["jobs"], *connected.json()["jobs"]): + await app.state.import_supervisor.wait(job["job_id"]) + + v1 = (await client.get("/api/v1/sources")).json() + v2 = (await client.get("/api/v2/sources")).json() + assert v1 == v2 + wife_sources = (await client.get("/api/v2/sources", params={"user_id": "wife"})).json() + assert [source["source_id"] for source in wife_sources] == ["hermes"] + assert wife_sources[0]["authorized"] is True + assert (await client.get("/api/v2/sources", params={"u": "x"})).status_code == 422 + + created = await client.post( + "/api/v2/import-jobs", + json={"source_id": "hermes", "operation": "sync", "user_id": "wife"}, + ) + assert created.status_code == 200, created.text + wife_job = created.json() + assert wife_job["user_id"] == "wife" and wife_job["source_id"] == "hermes" + await app.state.import_supervisor.wait(wife_job["job_id"]) + refused = await client.post( + "/api/v2/import-jobs", + json={"source_id": "claude-code", "user_id": "wife"}, + ) + assert refused.status_code in {404, 422} + unknown = await client.post( + "/api/v2/import-jobs", + json={"source_id": "hermes", "from_beginning": True}, + ) + assert unknown.status_code == 422 + + listed = (await client.get("/api/v2/import-jobs", params={"user_id": "wife"})).json() + assert {job["user_id"] for job in listed} == {"wife"} + assert (await client.get("/api/v2/import-jobs")).json() == ( + await client.get("/api/v1/import-jobs") + ).json() + path = f"/api/v2/import-jobs/{wife_job['job_id']}" + assert (await client.get(path)).status_code == 404 + own = await client.get(path, params={"user_id": "wife"}) + assert own.status_code == 200 and own.json()["job_id"] == wife_job["job_id"] + + synced = await client.post("/api/v2/sources/hermes/sync", json={"user_id": "wife"}) + v1_synced = await client.post("/api/v1/sources/hermes/sync") + assert synced.status_code == v1_synced.status_code, (synced.text, v1_synced.text) + assert synced.status_code == 200 + assert synced.json()["user_id"] == "wife" + assert v1_synced.json()["user_id"] == "local" + other = await client.post("/api/v2/sources/claude-code/sync", json={"user_id": "wife"}) + assert other.status_code in {404, 422} + assert ( + await client.post("/api/v2/sources/hermes/sync", json={"owner": "wife"}) + ).status_code == 422 + + asyncio.run(scenario()) diff --git a/memhub/docs/cli/EXPERIENCE.md b/memhub/docs/cli/EXPERIENCE.md index 09c024a..a2233a4 100644 --- a/memhub/docs/cli/EXPERIENCE.md +++ b/memhub/docs/cli/EXPERIENCE.md @@ -142,6 +142,16 @@ node apps/cli/dist/one-memory.cjs write status --user-id wife - 不带 `--user-id`(或写 `--user-id local`)时请求里不出现 `user_id`,和以前完全一样。 - 用户之间互相隔离:读、改、删另一个用户的记忆 ID 返回「未找到」;write 的 request id 也只属于提交它的用户。 +### 画像版本(`profile --versions`) + +```bash +node apps/cli/dist/one-memory.cjs profile --versions # L0 画像版本链,最新在前,* 为当前版本 +node apps/cli/dist/one-memory.cjs profile --versions --user-id wife --json +``` + +用于找出覆盖了正确值的错误版本(例如 `用户` / `The user`),再用 +`memories delete --yes` 删除;对应 `GET /api/v2/user-profile/versions`。 + ## 8. 召回(recall) `recall` 两条路径(raw / memory)此前已真机验证。文本输出会标明每条是「原始对话摘录」 diff --git a/memhub/packages/contracts/src/index.ts b/memhub/packages/contracts/src/index.ts index 1d443ba..0686fab 100644 --- a/memhub/packages/contracts/src/index.ts +++ b/memhub/packages/contracts/src/index.ts @@ -636,6 +636,16 @@ export interface SdkRawRecord { memory_at: number | null; gmt_created: number | null; custom: Record; + /** active, or shadow once pro extracted it. */ + status?: string | null; + /** The turn's messages, parsed from the SDK's `[role]: text` lines. */ + messages?: { role: string; content: string }[]; + /** The one role of a single-message turn; null for a mixed turn. */ + role?: string | null; + /** `sha256:` of the stored content, computed by Core. */ + content_hash?: string; + /** ISO time the SDK stored the raw. */ + created?: string | null; } export interface SdkRawPage { @@ -643,6 +653,37 @@ export interface SdkRawPage { total: number; limit: number; offset: number; + /** A time range was applied to this page in Core (the SDK list has none). */ + time_filtered?: boolean; +} + +/** `GET /api/v2/memory-library/layers`: memories per SDK layer. */ +export interface SdkLayerCounts { + layers: Record; + total: number; + statuses: string[] | null; +} + +/** `GET /api/v2/user-profile/versions`: the L0 basic-info version chains. */ +export interface SdkProfileVersions { + user_id: string; + request_id: string; + elapsed_ms: number; + chains: { + memory_id: string; + content: string; + status: string; + memory_at: number | null; + gmt_created: number | null; + basic_info: Record; + versions: { + memory_id: string; + content: string; + memory_at: number | string | null; + gmt_created: number | null; + current: boolean; + }[]; + }[]; } export interface FileImportReceipt {