diff --git a/.agents/skills/agent-core-dev/SKILL.md b/.agents/skills/agent-core-dev/SKILL.md index 7c53ec574..f7d55db5c 100644 --- a/.agents/skills/agent-core-dev/SKILL.md +++ b/.agents/skills/agent-core-dev/SKILL.md @@ -33,6 +33,7 @@ End-to-end procedures that span the stages. Reach for these before reading the s - [Stage 1 — Orient](orient.md): the DI black box (identity / dependencies / lifetime), the four `LifecycleScope` tiers and visibility, and the file-header comment convention. Read before touching business code. - [Stage 2 — Design a service](design.md): pick a scope, split a domain across scopes, choose a calling style (direct call vs event vs hook), and direct dependencies. Decide *where things live and who knows whom* before coding. + - Topic: [Persistence layering](persistence.md) — the three-layer `Store → Storage → backend` model, naming Stores by access pattern, and which layer business code should depend on. - [Stage 3 — Implement](implement.md): the standard Service recipe and the DI building blocks — interface + identity, constructor injection, scoped registration, `Disposable`, eager vs delayed, `invokeFunction`, `createInstance`, child scopes, and the cycle-refactor playbook. - Topic: [Service authoring](service-authoring.md) — file layout, naming, contract vs impl contents, interface style, constructor/field conventions, events, multi-Service domains, comment rules. - Topic: [Config](config.md) — the section-registry model, Core vs Session split, owning a config section, the TOML format, and the env overlay. diff --git a/.agents/skills/agent-core-dev/persistence.md b/.agents/skills/agent-core-dev/persistence.md new file mode 100644 index 000000000..454217832 --- /dev/null +++ b/.agents/skills/agent-core-dev/persistence.md @@ -0,0 +1,173 @@ +# Topic — Persistence layering + +How business code persists data in `agent-core-v2`: the three-layer model (`Store → Storage → backend`), the naming rules for each layer, and how to decide which layer a domain should depend on. Read this before adding any persistence to a domain. + +## The three-layer model + +Persistence is split into three layers, each hiding one kind of change: + +```text +Business Service + │ inject + ▼ +┌────────────────────────────────────────┐ +│ Store (semantic layer) │ ← access-pattern facade +│ IAppendLogStore / IAtomicDocumentStore│ append-log / atomic-doc / blob +└────────────────────────────────────────┘ + │ inject + ▼ +┌────────────────────────────────────────┐ +│ Storage (byte layer) │ ← byte primitives +│ IStorageService │ read/write/append/list/delete +│ IAppendLogStorage / IAtomicDocumentStorage │ (same interface, distinct tokens) +└────────────────────────────────────────┘ + │ implements + ▼ +┌────────────────────────────────────────┐ +│ Backend (deployment-specific) │ ← File / Postgres / Redis / S3 +│ FileStorageService / PostgresStorage │ +└────────────────────────────────────────┘ + │ uses + ▼ +┌────────────────────────────────────────┐ +│ Platform primitives │ ← hostFs / dbClient / redisClient +└────────────────────────────────────────┘ +``` + +Each layer hides exactly one concern: + +| Layer | Hides | Business code sees | +|---|---|---| +| **Store** | how an access pattern works (append-log reads, atomic-doc serialization) | "append this record" / "save this document" | +| **Storage** | byte primitives (atomic write, ordered append, prefix list) | `read/write/append/list/delete` over `(scope, key)` | +| **Backend** | deployment environment (file vs DB vs Redis vs S3) | nothing — chosen at the composition root | + +## The one-sentence rule + +> **Business code expresses *what* to store or fetch, never *how* to store it.** + +If business code contains any "how to persist" detail, it has punched through the layer it should depend on: + +| Business code contains | It has punched through | Depend on instead | +|---|---|---| +| `INSERT INTO …` / `SELECT …` | Storage + backend | a Store | +| file paths / `rename` / `fsync` | Storage | Storage or a Store | +| `JSON.parse` / `JSON.stringify` | Store (serialization) | `IAtomicDocumentStore` | +| append offsets / sequential cursors | Store (log semantics) | `IAppendLogStore` | +| `hash(data)` used as a key | Store (blob semantics) | `IBlobStore` | +| only `read/write/list/delete` on bytes | nothing — this is the byte layer | `IStorageService` directly ✅ | + +## Which layer to depend on — decision tree + +```text +Need to persist + │ + ├─ read-whole / write-whole, JSON-serializable? + │ └─ IAtomicDocumentStore + │ + ├─ append-only writes / sequential reads, independent records? + │ └─ IAppendLogStore + │ + ├─ large object, addressed by content hash? + │ └─ IBlobStore + │ + ├─ custom byte layout (index / cache / binary) that read/write/list cover? + │ └─ IStorageService directly + │ + ├─ new, reusable access semantics (multi-field query / time-range / graph)? + │ └─ add a new Store; business depends on the Store + │ + └─ business-specific, trivial, one or two lines? + └─ IStorageService directly; if it grows, extract a private Store +``` + +## Naming — Store by access pattern, not by business + +A Store abstracts an **access pattern**, not a business data type. Name it after the pattern so its reusability is obvious from the name. + +| Access pattern | Store name | Backend examples | +|---|---|---| +| append-log (append / sequential read) | `IAppendLogStore` | `FileAppendLogStore` / `PostgresAppendLogStore` | +| atomic-document (read/write whole) | `IAtomicDocumentStore` | `FileDocumentStore` / `RedisDocumentStore` | +| blob (hash-addressed large object) | `IBlobStore` | `FileBlobStore` / `S3BlobStore` | + +**Do not name a generic Store after a business concept.** `IRecordStore` / `IConfigStore` make a reusable access pattern look like a private store for one feature. Any domain that needs an append-log uses `IAppendLogStore`; any domain that needs an atomic document uses `IAtomicDocumentStore`. + +**Exception — business-specific Stores are named after the business.** When a Store captures one domain's unique query semantics (not a generic access pattern), name it after the domain: + +```text +ISessionIndex query / enumerate sessions by workspace ← business-specific +``` + +Test: is the Store's semantics a *generic access pattern* (append-log / atomic-doc / blob) or *one domain's unique query*? Generic → name by pattern; unique → name by domain. + +## Storage — one interface, distinct tokens per backend role + +The byte layer is a **single `IStorageService` interface** (read/write/append/list/delete). Different backends (File / Postgres / Redis) all implement it. To route different Stores to different backends, declare **distinct tokens of the same interface type**: + +```ts +export interface IStorageService { + read(scope: string, key: string): Promise; + write(scope: string, key: string, data: Uint8Array, options?: { atomic?: boolean }): Promise; + append(scope: string, key: string, data: Uint8Array, options?: { durable?: boolean }): Promise; + list(scope: string, prefix?: string): Promise; + delete(scope: string, key: string): Promise; + flush(): Promise; + close(): Promise; +} + +export const IAppendLogStorage = createDecorator('appendLogStorage'); +export const IAtomicDocumentStorage = createDecorator('atomicDocumentStorage'); +``` + +`IAppendLogStorage` and `IAtomicDocumentStorage` share the `IStorageService` type (so `AppendLogStore` / `AtomicDocumentStore` code is unchanged) but are distinct DI tokens, so the composition root can bind each to a different backend: + +```ts +// Local profile — both on the local filesystem +collection.set(IAppendLogStorage, fileStorageService); +collection.set(IAtomicDocumentStorage, fileStorageService); + +// Server profile — append-logs on Postgres, atomic documents on Redis +collection.set(IAppendLogStorage, new PostgresStorageService(db, 'records')); +collection.set(IAtomicDocumentStorage, new RedisStorageService(redis, 'config')); +``` + +Use a token to express **backend role** (append-log / atomic-document / blob); use the `scope` parameter to express **business namespace** within a backend. Do not overload `scope` to route backends. + +## Store `acquire(scope, key)` — flush-on-dispose handle + +Stores that buffer writes expose an `acquire(scope, key)` handle so a business can flush them on disposal: + +```ts +export interface IAppendLogStore { + // … + /** + * Acquire a disposable handle for `(scope, key)`. Register it with your + * `Disposable` (via `this._register(...)`); when you are disposed, pending + * appends for that log are flushed. The shared store itself is not disposed. + */ + acquire(scope: string, key: string): IDisposable; +} +``` + +`IAppendLogStore.acquire` flushes the log's pending appends on dispose — it exists because `append` is fire-and-forget. `IAtomicDocumentStore.acquire` is a no-op today (atomic documents are durable on write) and exists for interface symmetry. Businesses that do not need flush-on-dispose simply do not call `acquire`. + +## When Storage primitives may diverge + +Keep `IStorageService` unified for byte storage. Diverge only when the semantics genuinely do not fit: + +- **Blobs** do not fit `IStorageService` (large objects, hash-addressed, S3 has no native append) → `IBlobStore` is a separate interface with its own backends. +- **A backend has a fast primitive the unified interface cannot express** (e.g. Postgres `COPY`) → as an exception, let that backend implement the Store interface directly, bypassing `IStorageService`. This is an exception, not the default. + +## Platform primitives are deployment-coupled, not core abstractions + +`hostFs` (local filesystem) is a **platform primitive** used only by local backends (`FileStorageService`, `LocalFileSystemBackend`, `LocalSkillCatalog`, `HostFolderBrowser`). It is **not** a core abstraction and must not appear in L2/L3 dependency graphs. A server deployment swaps those backends for DB / S3 implementations and never registers `hostFs`. + +## Red lines (this topic) + +- Business code never contains "how to persist" details (serialization / paths / SQL / append offsets) — if it does, drop a layer. +- Name generic Stores by access pattern (`IAppendLogStore` / `IAtomicDocumentStore` / `IBlobStore`), never by business concept (`IRecordStore` / `IConfigStore`). +- Business-specific Stores (unique query semantics) are named after the domain (`ISessionIndex`). +- `IStorageService` is the single byte-layer interface; route backends with **distinct tokens of the same type** (`IAppendLogStorage` / `IAtomicDocumentStorage`), not by overloading `scope`. +- `hostFs` is a local-only platform primitive; L2/L3 domains must not import `node:fs` or `hostFs` directly. +- Do not create a pass-through `Store` that only forwards `read/write` — a Store must hide a real access-pattern concern, or it is noise; use `IStorageService` directly instead. diff --git a/packages/agent-core-v2/AGENTS.md b/packages/agent-core-v2/AGENTS.md index a44f90464..de5f05765 100644 --- a/packages/agent-core-v2/AGENTS.md +++ b/packages/agent-core-v2/AGENTS.md @@ -33,6 +33,17 @@ Barrel (`src/session/index.ts`): */ ``` +## Persistence + +Business domains **do not implement persistence themselves** — they depend on a Service that owns the access pattern. Business code expresses *what* to store or fetch, never *how*. + +- Append-log → `IAppendLogStore` +- Atomic document → `IAtomicDocumentStore` +- Blob → `IBlobStore` +- Domain-specific query → a dedicated Store (e.g. `ISessionIndex`) + +Business code must not `import 'node:fs'`, write SQL, hand-roll append-logs / atomic writes, or hold file handles. Generic Stores are named by **access pattern** (`IAppendLogStore`, `IAtomicDocumentStore`); only domain-unique Stores are named after the domain (`ISessionIndex`). See `.agents/skills/agent-core-dev/persistence.md` for the full layering rules and decision tree. + ## Docs Per-domain references live in `docs/`. diff --git a/packages/agent-core-v2/src/blobStore/blobStoreService.ts b/packages/agent-core-v2/src/blobStore/blobStoreService.ts index e16c3500b..dcfc08fab 100644 --- a/packages/agent-core-v2/src/blobStore/blobStoreService.ts +++ b/packages/agent-core-v2/src/blobStore/blobStoreService.ts @@ -1,12 +1,10 @@ import { createHash } from 'node:crypto'; -import { mkdir, - open, - readFile } from 'node:fs/promises'; import { join } from 'pathe'; import type { ContentPart } from '@moonshot-ai/kosong'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; +import { IHostFileSystem } from '#/hostFs'; import { BLOBREF_PROTOCOL, @@ -27,7 +25,10 @@ export class BlobStoreService implements IBlobStoreService { private readonly cacheSizes = new Map(); private currentCacheSize = 0; - constructor(options: BlobStoreServiceOptions = {}) { + constructor( + options: BlobStoreServiceOptions = {}, + @IHostFileSystem private readonly hostFs: IHostFileSystem, + ) { this.blobsDir = options.blobsDir; this.threshold = options.threshold ?? DEFAULT_THRESHOLD; this.maxCacheSize = options.maxCacheSize ?? DEFAULT_MAX_CACHE_SIZE; @@ -121,11 +122,11 @@ export class BlobStoreService implements IBlobStoreService { } if (this.blobsDir === undefined) return undefined; - const payload = await readFile(join(this.blobsDir, hash)).catch(() => undefined); + const payload = await this.hostFs.readBytes(join(this.blobsDir, hash)).catch(() => undefined); if (payload !== undefined) { - this.setCache(hash, payload); + this.setCache(hash, Buffer.from(payload)); } - return payload; + return payload !== undefined ? Buffer.from(payload) : undefined; } private async maybeOffloadString(value: string): Promise { @@ -145,22 +146,11 @@ export class BlobStoreService implements IBlobStoreService { const blobsDir = this.blobsDir; if (blobsDir === undefined) return `data:${mimeType};base64,${base64Payload}`; - await mkdir(blobsDir, { recursive: true, mode: 0o700 }); + await this.hostFs.mkdir(blobsDir, { recursive: true }); const hash = createHash('sha256').update(base64Payload, 'utf8').digest('hex'); const blobPath = join(blobsDir, hash); const binary = Buffer.from(base64Payload, 'base64'); - try { - const fh = await open(blobPath, 'wx'); - try { - await fh.writeFile(binary); - await fh.sync(); - } finally { - await fh.close(); - } - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code !== 'EEXIST') throw error; - } + await this.hostFs.createExclusive(blobPath, binary); this.setCache(hash, binary); return `${BLOBREF_PROTOCOL}${mimeType};${hash}`; } diff --git a/packages/agent-core-v2/src/hostFs/hostFs.ts b/packages/agent-core-v2/src/hostFs/hostFs.ts index 28674df5e..fd1de31d8 100644 --- a/packages/agent-core-v2/src/hostFs/hostFs.ts +++ b/packages/agent-core-v2/src/hostFs/hostFs.ts @@ -29,6 +29,13 @@ export interface IHostFileSystem { writeText(path: string, data: string): Promise; readBytes(path: string): Promise; writeBytes(path: string, data: Uint8Array): Promise; + /** + * Create a file exclusively with `data`. Returns `true` when the file was + * created, `false` when it already existed (EEXIST) — the existing content is + * left untouched. Used by content-addressed stores where a collision means + * the same bytes are already present. + */ + createExclusive(path: string, data: Uint8Array): Promise; stat(path: string): Promise; readdir(path: string): Promise; mkdir(path: string, options?: { readonly recursive?: boolean }): Promise; diff --git a/packages/agent-core-v2/src/hostFs/hostFsService.ts b/packages/agent-core-v2/src/hostFs/hostFsService.ts index 3df079c58..87d89c8a5 100644 --- a/packages/agent-core-v2/src/hostFs/hostFsService.ts +++ b/packages/agent-core-v2/src/hostFs/hostFsService.ts @@ -5,7 +5,7 @@ * `node:fs/promises`. Bound at Core scope. */ -import { readFile, readdir, stat, mkdir, rm, writeFile } from 'node:fs/promises'; +import { open, readFile, readdir, stat, mkdir, rm, writeFile } from 'node:fs/promises'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; @@ -32,6 +32,22 @@ export class HostFileSystem implements IHostFileSystem { await writeFile(path, data); } + async createExclusive(path: string, data: Uint8Array): Promise { + try { + const fh = await open(path, 'wx'); + try { + await fh.writeFile(data); + await fh.sync(); + } finally { + await fh.close(); + } + return true; + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'EEXIST') return false; + throw error; + } + } + async stat(path: string): Promise { const s = await stat(path); return { isFile: s.isFile(), isDirectory: s.isDirectory(), size: s.size }; diff --git a/packages/agent-core-v2/src/index.ts b/packages/agent-core-v2/src/index.ts index 20b4e7018..0db1e1713 100644 --- a/packages/agent-core-v2/src/index.ts +++ b/packages/agent-core-v2/src/index.ts @@ -12,7 +12,7 @@ export * from './environment/index'; export * from './hostFs/index'; export * from './kosong/index'; -export * from './sessionStore/index'; +export * from './sessionIndex/index'; export * from './sessionMetaStore/index'; export * from './config/index'; diff --git a/packages/agent-core-v2/src/sessionIndex/index.ts b/packages/agent-core-v2/src/sessionIndex/index.ts new file mode 100644 index 000000000..07fa5bb2b --- /dev/null +++ b/packages/agent-core-v2/src/sessionIndex/index.ts @@ -0,0 +1,6 @@ +/** + * `sessionIndex` domain barrel. + */ + +export * from './sessionIndex'; +export * from './sessionIndexService'; diff --git a/packages/agent-core-v2/src/sessionIndex/sessionIndex.ts b/packages/agent-core-v2/src/sessionIndex/sessionIndex.ts new file mode 100644 index 000000000..e8eee6d44 --- /dev/null +++ b/packages/agent-core-v2/src/sessionIndex/sessionIndex.ts @@ -0,0 +1,24 @@ +/** + * `sessionIndex` domain (L2) — session index contract. + * + * `ISessionIndex` is a domain-specific persistence Store: it knows how to + * locate and enumerate session directories under a `sessionsRoot`. Business + * code depends on `ISessionIndex` rather than touching the filesystem directly. + * Backends are deployment-specific (local filesystem today; database on a + * server). + */ + +import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; + +export interface ISessionIndex { + readonly _serviceBrand: undefined; + /** Absolute directory for a given session under `sessionsRoot`. */ + sessionDir(sessionsRoot: string, workDir: string, sessionId: string): string; + /** Stable workspace id (the `wd__` key) derived from a work dir. */ + workspaceIdFor(workDir: string): string; + /** Count non-archived session directories for a work dir under `sessionsRoot`. */ + countActive(sessionsRoot: string, workDir: string): Promise; +} + +export const ISessionIndex: ServiceIdentifier = + createDecorator('sessionIndex'); diff --git a/packages/agent-core-v2/src/sessionStore/sessionStoreService.ts b/packages/agent-core-v2/src/sessionIndex/sessionIndexService.ts similarity index 78% rename from packages/agent-core-v2/src/sessionStore/sessionStoreService.ts rename to packages/agent-core-v2/src/sessionIndex/sessionIndexService.ts index ba83ebd53..eb9fdad97 100644 --- a/packages/agent-core-v2/src/sessionStore/sessionStoreService.ts +++ b/packages/agent-core-v2/src/sessionIndex/sessionIndexService.ts @@ -1,8 +1,10 @@ /** - * `sessionStore` domain (L2) — `ISessionStore` implementation. + * `sessionIndex` domain (L2) — `FileSessionIndex` implementation. * - * Enumerates session directories on the real local disk through the program - * side `hostFs` primitives. Bound at Core scope. + * Enumerates session directories on the local filesystem through the program + * side `hostFs` primitives. This is the local-deployment backend of + * `ISessionIndex`; a server deployment would substitute a database-backed + * `DbSessionIndex`. Bound at Core scope. */ import { createHash } from 'node:crypto'; @@ -13,7 +15,7 @@ import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { slugifyWorkDirName } from '#/_base/utils/workdir-slug'; import { IHostFileSystem } from '#/hostFs'; -import { ISessionStore } from './sessionStore'; +import { ISessionIndex } from './sessionIndex'; const WORKDIR_KEY_PREFIX = 'wd_'; const HASH_LENGTH = 12; @@ -26,7 +28,7 @@ export function encodeWorkDirKey(workDir: string): string { return `${WORKDIR_KEY_PREFIX}${slug}_${hash}`; } -export class SessionStore implements ISessionStore { +export class FileSessionIndex implements ISessionIndex { declare readonly _serviceBrand: undefined; constructor(@IHostFileSystem private readonly hostFs: IHostFileSystem) {} @@ -39,7 +41,7 @@ export class SessionStore implements ISessionStore { return encodeWorkDirKey(workDir); } - async countActiveSessions(sessionsRoot: string, workDir: string): Promise { + async countActive(sessionsRoot: string, workDir: string): Promise { const dir = join(sessionsRoot, encodeWorkDirKey(workDir)); let entries; try { @@ -74,8 +76,8 @@ export class SessionStore implements ISessionStore { registerScopedService( LifecycleScope.Core, - ISessionStore, - SessionStore, + ISessionIndex, + FileSessionIndex, InstantiationType.Delayed, - 'records', + 'sessionIndex', ); diff --git a/packages/agent-core-v2/src/sessionMetaStore/sessionMetaStoreService.ts b/packages/agent-core-v2/src/sessionMetaStore/sessionMetaStoreService.ts index 325fcbc24..e4746662d 100644 --- a/packages/agent-core-v2/src/sessionMetaStore/sessionMetaStoreService.ts +++ b/packages/agent-core-v2/src/sessionMetaStore/sessionMetaStoreService.ts @@ -1,15 +1,16 @@ /** * `sessionMetaStore` domain (L2) — `ISessionMetaStore` implementation. * - * Persists session metadata as a single atomic document through the program - * side `storage` (`IConfigStore`). Bound at Session scope. + * Persists session metadata as a single atomic document through the + * `storage` access-pattern store (`IAtomicDocumentStore`). Bound at Session + * scope. */ import { Disposable } from '#/_base/di/lifecycle'; import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { ILogService } from '#/log'; -import { IConfigStore } from '#/storage'; +import { IAtomicDocumentStore } from '#/storage'; import { ISessionMetaStore } from './sessionMetaStore'; @@ -21,7 +22,7 @@ export class SessionMetaStore extends Disposable implements ISessionMetaStore { private readonly key: string; constructor( - @IConfigStore private readonly configStore: IConfigStore, + @IAtomicDocumentStore private readonly documentStore: IAtomicDocumentStore, @ILogService _log: ILogService, key: string = 'state.json', ) { @@ -31,7 +32,7 @@ export class SessionMetaStore extends Disposable implements ISessionMetaStore { async read(): Promise> { this.data = - (await this.configStore.get>(SCOPE, this.key)) ?? {}; + (await this.documentStore.get>(SCOPE, this.key)) ?? {}; return this.data; } @@ -41,7 +42,7 @@ export class SessionMetaStore extends Disposable implements ISessionMetaStore { } async flush(): Promise { - await this.configStore.set(SCOPE, this.key, this.data); + await this.documentStore.set(SCOPE, this.key, this.data); } } diff --git a/packages/agent-core-v2/src/sessionStore/index.ts b/packages/agent-core-v2/src/sessionStore/index.ts deleted file mode 100644 index 207fb2ea5..000000000 --- a/packages/agent-core-v2/src/sessionStore/index.ts +++ /dev/null @@ -1,6 +0,0 @@ -/** - * `sessionStore` domain barrel. - */ - -export * from './sessionStore'; -export * from './sessionStoreService'; diff --git a/packages/agent-core-v2/src/sessionStore/sessionStore.ts b/packages/agent-core-v2/src/sessionStore/sessionStore.ts deleted file mode 100644 index 00f8cc823..000000000 --- a/packages/agent-core-v2/src/sessionStore/sessionStore.ts +++ /dev/null @@ -1,18 +0,0 @@ -/** - * `sessionStore` domain — core-scope session directory store contract. - */ - -import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; - -export interface ISessionStore { - readonly _serviceBrand: undefined; - /** Absolute directory for a given session under `sessionsRoot`. */ - sessionDir(sessionsRoot: string, workDir: string, sessionId: string): string; - /** Stable workspace id (the `wd__` key) derived from a work dir. */ - workspaceIdFor(workDir: string): string; - /** Count non-archived session directories for a work dir under `sessionsRoot`. */ - countActiveSessions(sessionsRoot: string, workDir: string): Promise; -} - -export const ISessionStore: ServiceIdentifier = - createDecorator('sessionStore'); diff --git a/packages/agent-core-v2/src/storage/recordStore.ts b/packages/agent-core-v2/src/storage/appendLogStore.ts similarity index 58% rename from packages/agent-core-v2/src/storage/recordStore.ts rename to packages/agent-core-v2/src/storage/appendLogStore.ts index bd80c8dc0..91fbe24b6 100644 --- a/packages/agent-core-v2/src/storage/recordStore.ts +++ b/packages/agent-core-v2/src/storage/appendLogStore.ts @@ -1,5 +1,5 @@ /** - * `IRecordStore` / `RecordStore` — the typed append-log service. + * `IAppendLogStore` / `AppendLogStore` — the append-log access-pattern store. * * Sits on top of `IStorageService` and turns a byte stream into an ordered * sequence of typed JSON records. Owns the concerns the storage service @@ -7,38 +7,44 @@ * batching of appends into a single durable `append`, and crash-tolerant * decoding (a torn final line is dropped; corruption anywhere else throws). * - * It is a DI service: domains inject `IRecordStore` and call - * `append/read/rewrite` with the `(scope, key)` of the log they own. Buffering - * is kept per log inside the service, so many appends within a synchronous - * block collapse into one durable write. + * It is a DI service: any domain that needs an append-log injects + * `IAppendLogStore` and calls `append/read/rewrite` with the `(scope, key)` of + * the log it owns. Buffering is kept per log inside the service, so many + * appends within a synchronous block collapse into one durable write. */ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; import { InstantiationType } from '#/_base/di/extensions'; +import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IStorageService } from './storageService'; +import { IAppendLogStorage, IStorageService } from './storageService'; const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); -export class RecordCorruptedError extends Error { +export class AppendLogCorruptedError extends Error { constructor( readonly scope: string, readonly key: string, readonly lineNumber: number, cause: unknown, ) { - super(`record log ${scope}/${key}: corrupted line ${lineNumber}: ${String(cause)}`); - this.name = 'RecordCorruptedError'; + super(`append-log ${scope}/${key}: corrupted line ${lineNumber}: ${String(cause)}`); + this.name = 'AppendLogCorruptedError'; } } -export interface IRecordStore { +export interface AppendLogOptions { + /** Called when a background flush fails. */ + readonly onError?: (error: unknown) => void; +} + +export interface IAppendLogStore { readonly _serviceBrand: undefined; /** Buffer a record for the next durable append. Resolves immediately. */ - append(scope: string, key: string, record: R): void; + append(scope: string, key: string, record: R, options?: AppendLogOptions): void; /** * Replay the log in order. Flushes pending appends first. A torn final line @@ -54,50 +60,82 @@ export interface IRecordStore { /** Flush and release resources. */ close(): Promise; + + /** + * Acquire a disposable handle for `(scope, key)`. Register it with your + * `Disposable` (via `this._register(...)`); when you are disposed, pending + * appends for that log are flushed. The shared store itself is not disposed. + */ + acquire(scope: string, key: string): IDisposable; } -export const IRecordStore: ServiceIdentifier = - createDecorator('recordStore'); +export const IAppendLogStore: ServiceIdentifier = + createDecorator('appendLogStore'); interface LogState { pending: unknown[]; flushPromise: Promise | undefined; flushScheduled: boolean; + onError?: (error: unknown) => void; } -export class RecordStore implements IRecordStore { +export class AppendLogStore implements IAppendLogStore { declare readonly _serviceBrand: undefined; private readonly logs = new Map(); - constructor(@IStorageService private readonly storage: IStorageService) {} + constructor(@IAppendLogStorage private readonly storage: IStorageService) {} - append(scope: string, key: string, record: R): void { + append(scope: string, key: string, record: R, options?: AppendLogOptions): void { const state = this.state(scope, key); state.pending.push(record); + if (options?.onError !== undefined && state.onError === undefined) { + state.onError = options.onError; + } this.scheduleFlush(scope, key, state); } async *read(scope: string, key: string): AsyncIterable { await this.flushLog(scope, key); - const bytes = await this.storage.read(scope, key); - if (bytes === undefined) return; - - const lines = textDecoder.decode(bytes).split('\n'); - for (let i = 0; i < lines.length; i++) { - const raw = lines[i]!; - const line = raw.endsWith('\r') ? raw.slice(0, -1) : raw; - if (line.length === 0) continue; - const isLast = i === lines.length - 1; - try { - yield JSON.parse(line) as R; - } catch (error) { - // A crash can leave a half-written last line; drop it. Corruption - // anywhere before the end is real and must surface. - if (isLast) return; - throw new RecordCorruptedError(scope, key, i + 1, error); + let pending = ''; + let lineNumber = 0; + for await (const chunk of this.storage.readStream(scope, key)) { + pending += textDecoder.decode(chunk, { stream: true }); + let newlineIndex = pending.indexOf('\n'); + while (newlineIndex !== -1) { + const raw = pending.slice(0, newlineIndex); + pending = pending.slice(newlineIndex + 1); + lineNumber++; + const record = this.parseLine(raw, scope, key, lineNumber, false); + if (record !== undefined) yield record; + newlineIndex = pending.indexOf('\n'); } } + pending += textDecoder.decode(); + if (pending.length > 0) { + lineNumber++; + // A crash can leave a half-written last line (no trailing newline); drop + // it. Corruption anywhere before the end is real and must surface. + const record = this.parseLine(pending, scope, key, lineNumber, true); + if (record !== undefined) yield record; + } + } + + private parseLine( + raw: string, + scope: string, + key: string, + lineNumber: number, + allowTruncated: boolean, + ): R | undefined { + const line = raw.endsWith('\r') ? raw.slice(0, -1) : raw; + if (line.length === 0) return undefined; + try { + return JSON.parse(line) as R; + } catch (error) { + if (allowTruncated) return undefined; + throw new AppendLogCorruptedError(scope, key, lineNumber, error); + } } async rewrite(scope: string, key: string, records: readonly R[]): Promise { @@ -118,6 +156,12 @@ export class RecordStore implements IRecordStore { await this.flush(); } + acquire(scope: string, key: string): IDisposable { + return toDisposable(() => { + void this.flushLog(scope, key); + }); + } + private state(scope: string, key: string): LogState { const id = logId(scope, key); let state = this.logs.get(id); @@ -137,7 +181,7 @@ export class RecordStore implements IRecordStore { state.flushScheduled = true; queueMicrotask(() => { state.flushScheduled = false; - void this.flushLog(scope, key); + void this.flushLog(scope, key).catch((error) => state.onError?.(error)); }); } @@ -183,8 +227,8 @@ function encodeBatch(records: readonly unknown[]): Uint8Array { registerScopedService( LifecycleScope.Session, - IRecordStore, - RecordStore, + IAppendLogStore, + AppendLogStore, InstantiationType.Delayed, 'storage', ); diff --git a/packages/agent-core-v2/src/storage/configStore.ts b/packages/agent-core-v2/src/storage/atomicDocumentStore.ts similarity index 53% rename from packages/agent-core-v2/src/storage/configStore.ts rename to packages/agent-core-v2/src/storage/atomicDocumentStore.ts index d96a8ab1d..f2256d52c 100644 --- a/packages/agent-core-v2/src/storage/configStore.ts +++ b/packages/agent-core-v2/src/storage/atomicDocumentStore.ts @@ -1,27 +1,29 @@ /** - * `IConfigStore` / `ConfigStore` — the typed atomic-document service. + * `IAtomicDocumentStore` / `AtomicDocumentStore` — the atomic-document + * access-pattern store. * * Sits on top of `IStorageService` and stores one typed JSON value per - * `(scope, key)`, replaced atomically on every write. This is the `Config` - * access pattern: `state.json`, `upcoming-goals.json`, per-id cron/background - * records, etc. + * `(scope, key)`, replaced atomically on every write. This is the atomic- + * document access pattern: `state.json`, `upcoming-goals.json`, per-id + * cron/background records, etc. * - * It is a DI service: domains inject `IConfigStore` and call `get/set` with - * the scope they own — they do not construct stores themselves. JSON - * (de)serialization and atomic replacement are centralized here so domains - * do not reimplement them. + * It is a DI service: any domain that needs an atomic document injects + * `IAtomicDocumentStore` and calls `get/set` with the scope it owns — it does + * not construct stores itself. JSON (de)serialization and atomic replacement + * are centralized here so domains do not reimplement them. */ import { createDecorator, type ServiceIdentifier } from '#/_base/di/instantiation'; import { InstantiationType } from '#/_base/di/extensions'; +import { toDisposable, type IDisposable } from '#/_base/di/lifecycle'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; -import { IStorageService } from './storageService'; +import { IAtomicDocumentStorage, IStorageService } from './storageService'; const textEncoder = new TextEncoder(); const textDecoder = new TextDecoder(); -export interface IConfigStore { +export interface IAtomicDocumentStore { readonly _serviceBrand: undefined; /** Read the value at `(scope, key)`, or `undefined` when absent. */ @@ -35,15 +37,24 @@ export interface IConfigStore { /** List the keys under `scope`, optionally filtered by `prefix`. */ list(scope: string, prefix?: string): Promise; + + /** + * Acquire a disposable handle for `(scope, key)`. Register it with your + * `Disposable`; when you are disposed, the handle is released. The shared + * store itself is not disposed. Atomic documents are durable on write, so + * the handle currently releases no resources; it exists for interface + * symmetry with `IAppendLogStore`. + */ + acquire(scope: string, key: string): IDisposable; } -export const IConfigStore: ServiceIdentifier = - createDecorator('configStore'); +export const IAtomicDocumentStore: ServiceIdentifier = + createDecorator('atomicDocumentStore'); -export class ConfigStore implements IConfigStore { +export class AtomicDocumentStore implements IAtomicDocumentStore { declare readonly _serviceBrand: undefined; - constructor(@IStorageService private readonly storage: IStorageService) {} + constructor(@IAtomicDocumentStorage private readonly storage: IStorageService) {} async get(scope: string, key: string): Promise { const bytes = await this.storage.read(scope, key); @@ -63,12 +74,16 @@ export class ConfigStore implements IConfigStore { async list(scope: string, prefix?: string): Promise { return this.storage.list(scope, prefix); } + + acquire(scope: string, key: string): IDisposable { + return toDisposable(() => {}); + } } registerScopedService( LifecycleScope.Session, - IConfigStore, - ConfigStore, + IAtomicDocumentStore, + AtomicDocumentStore, InstantiationType.Delayed, 'storage', ); diff --git a/packages/agent-core-v2/src/storage/fileStorageService.ts b/packages/agent-core-v2/src/storage/fileStorageService.ts index 11c34dbd8..367f41b50 100644 --- a/packages/agent-core-v2/src/storage/fileStorageService.ts +++ b/packages/agent-core-v2/src/storage/fileStorageService.ts @@ -17,6 +17,7 @@ * which the agent-execution-environment abstraction does not expose. */ +import { createReadStream } from 'node:fs'; import { mkdir, open, readFile, readdir, unlink } from 'node:fs/promises'; import { dirname, join } from 'pathe'; @@ -48,6 +49,18 @@ export class FileStorageService implements IStorageService { } } + async *readStream(scope: string, key: string): AsyncIterable { + const stream = createReadStream(this.path(scope, key)); + try { + for await (const chunk of stream) { + yield chunk as Uint8Array; + } + } catch (error) { + if (isEnoent(error)) return; + throw error; + } + } + async write( scope: string, key: string, diff --git a/packages/agent-core-v2/src/storage/inMemoryStorageService.ts b/packages/agent-core-v2/src/storage/inMemoryStorageService.ts index 3c351d817..ccbfde65e 100644 --- a/packages/agent-core-v2/src/storage/inMemoryStorageService.ts +++ b/packages/agent-core-v2/src/storage/inMemoryStorageService.ts @@ -16,6 +16,8 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, registerScopedService } from '#/_base/di/scope'; import { + IAppendLogStorage, + IAtomicDocumentStorage, IStorageService, type StorageAppendOptions, type StorageWriteOptions, @@ -30,6 +32,11 @@ export class InMemoryStorageService implements IStorageService { return this.scopes.get(scope)?.get(key); } + async *readStream(scope: string, key: string): AsyncIterable { + const data = this.scopes.get(scope)?.get(key); + if (data !== undefined) yield data; + } + async write( scope: string, key: string, @@ -89,3 +96,19 @@ registerScopedService( InstantiationType.Delayed, 'storage', ); + +registerScopedService( + LifecycleScope.Session, + IAppendLogStorage, + InMemoryStorageService, + InstantiationType.Delayed, + 'storage', +); + +registerScopedService( + LifecycleScope.Session, + IAtomicDocumentStorage, + InMemoryStorageService, + InstantiationType.Delayed, + 'storage', +); diff --git a/packages/agent-core-v2/src/storage/index.ts b/packages/agent-core-v2/src/storage/index.ts index 10e101d40..42509e8c9 100644 --- a/packages/agent-core-v2/src/storage/index.ts +++ b/packages/agent-core-v2/src/storage/index.ts @@ -1,6 +1,6 @@ export * from './storageService'; export * from './fileStorageService'; export * from './inMemoryStorageService'; -export * from './recordStore'; -export * from './configStore'; +export * from './appendLogStore'; +export * from './atomicDocumentStore'; export * from './queryStore'; diff --git a/packages/agent-core-v2/src/storage/queryStore.ts b/packages/agent-core-v2/src/storage/queryStore.ts index 7bf04802a..232eb0008 100644 --- a/packages/agent-core-v2/src/storage/queryStore.ts +++ b/packages/agent-core-v2/src/storage/queryStore.ts @@ -1,10 +1,11 @@ /** * `IQueryStore` — the indexed, queryable read-model facade. * - * A peer of `IRecordStore` and `IConfigStore`. Where `IRecordStore` is the - * authoritative append-only write model and `IConfigStore` holds atomic - * documents, `IQueryStore` serves fast, indexed, paginated reads over a - * *derived* dataset — typically materialized from a record log by a projector. + * A peer of `IAppendLogStore` and `IAtomicDocumentStore`. Where + * `IAppendLogStore` is the authoritative append-only write model and + * `IAtomicDocumentStore` holds atomic documents, `IQueryStore` serves fast, + * indexed, paginated reads over a *derived* dataset — typically materialized + * from an append log by a projector. * * This file intentionally ships the interface only. A concrete implementation * (e.g. backed by `minidb`) and the projector that feeds it are a follow-up; diff --git a/packages/agent-core-v2/src/storage/storageService.ts b/packages/agent-core-v2/src/storage/storageService.ts index 5ff7cf230..490721751 100644 --- a/packages/agent-core-v2/src/storage/storageService.ts +++ b/packages/agent-core-v2/src/storage/storageService.ts @@ -16,7 +16,7 @@ * * The service is intentionally byte-oriented and scope/key-addressed: it knows * nothing about JSON, records, configs, versions or framing. Those concerns - * live in the typed facades above it (`IRecordStore`, `IConfigStore`). + * live in the typed facades above it (`IAppendLogStore`, `IAtomicDocumentStore`). * * `scope`/`key` are trusted internal path segments for the file implementation * (e.g. scope `"agents/main"`, key `"wire.jsonl"`); they are not user input. @@ -48,6 +48,13 @@ export interface IStorageService { /** Read the whole value, or `undefined` when the key does not exist. */ read(scope: string, key: string): Promise; + /** + * Stream the bytes of `(scope, key)` as chunks. Yields nothing when the key + * does not exist. Implementations may back this with a real stream (file) or + * a single chunk (memory / DB). + */ + readStream(scope: string, key: string): AsyncIterable; + /** Atomically replace the whole value. */ write(scope: string, key: string, data: Uint8Array, options?: StorageWriteOptions): Promise; @@ -69,3 +76,21 @@ export interface IStorageService { export const IStorageService: ServiceIdentifier = createDecorator('storageService'); + +/** + * Token for the byte-storage backend dedicated to the append-log access + * pattern. Shares the `IStorageService` interface; the distinct token lets the + * composition root bind it to a different backend (e.g. Postgres) than the + * atomic-document backend. + */ +export const IAppendLogStorage: ServiceIdentifier = + createDecorator('appendLogStorage'); + +/** + * Token for the byte-storage backend dedicated to the atomic-document access + * pattern. Shares the `IStorageService` interface; the distinct token lets the + * composition root bind it to a different backend (e.g. Redis) than the + * append-log backend. + */ +export const IAtomicDocumentStorage: ServiceIdentifier = + createDecorator('atomicDocumentStorage'); diff --git a/packages/agent-core-v2/src/wireRecord/index.ts b/packages/agent-core-v2/src/wireRecord/index.ts index 5a7c299bb..4a020e60a 100644 --- a/packages/agent-core-v2/src/wireRecord/index.ts +++ b/packages/agent-core-v2/src/wireRecord/index.ts @@ -2,7 +2,6 @@ * `wireRecord` domain barrel - re-exports the wireRecord service contract and implementation. */ -export * from './persistence'; export * from './wireRecord'; export * from './wireRecordService'; export * from './migration/index'; diff --git a/packages/agent-core-v2/src/wireRecord/persistence.ts b/packages/agent-core-v2/src/wireRecord/persistence.ts deleted file mode 100644 index ddcef699d..000000000 --- a/packages/agent-core-v2/src/wireRecord/persistence.ts +++ /dev/null @@ -1,216 +0,0 @@ -import { createReadStream } from 'node:fs'; -import { mkdir, open } from 'node:fs/promises'; -import { dirname } from 'pathe'; - -import { syncDir } from "#/_base/utils/fs"; -import type { PersistedWireRecord, WireRecordPersistence } from './wireRecord'; - -export interface FileSystemWireRecordPersistenceOptions { - readonly onError?: (error: unknown) => void; - readonly beforeWrite?: ( - record: PersistedWireRecord, - ) => PersistedWireRecord | Promise; -} - -export interface InMemoryWireRecordPersistenceOptions { - readonly onRecord?: (record: PersistedWireRecord) => void; -} - -export class InMemoryWireRecordPersistence implements WireRecordPersistence { - readonly records: PersistedWireRecord[] = []; - - constructor( - records: readonly PersistedWireRecord[] = [], - private readonly options: InMemoryWireRecordPersistenceOptions = {}, - ) { - this.records.push(...records); - } - - async *read(): AsyncIterable { - for (const record of this.records) { - yield record; - } - } - - append(input: PersistedWireRecord): void { - this.records.push(input); - this.options.onRecord?.(input); - } - - rewrite(records: readonly PersistedWireRecord[]): void { - this.records.splice(0, this.records.length, ...records); - } - - async flush(): Promise {} - - async close(): Promise {} -} - -export class FileSystemWireRecordPersistence implements WireRecordPersistence { - private readonly pendingRecords: PersistedWireRecord[] = []; - private shouldClear = false; - private directorySynced = false; - private flushPromise: Promise | undefined; - private error: unknown; - - constructor( - private readonly filePath: string, - private readonly options: FileSystemWireRecordPersistenceOptions = {}, - ) {} - - async *read(): AsyncIterable { - await this.flush(); - - let line = ''; - let lineNumber = 0; - const stream = createReadStream(this.filePath, { encoding: 'utf8' }); - try { - for await (const chunk of stream) { - line += chunk; - let newlineIndex = line.indexOf('\n'); - while (newlineIndex !== -1) { - const rawLine = line.slice(0, newlineIndex); - line = line.slice(newlineIndex + 1); - lineNumber++; - - const record = parseRecordLine( - rawLine.endsWith('\r') ? rawLine.slice(0, -1) : rawLine, - lineNumber, - this.filePath, - false, - ); - if (record !== undefined) yield record; - - newlineIndex = line.indexOf('\n'); - } - } - } catch (error) { - const code = (error as NodeJS.ErrnoException).code; - if (code === 'ENOENT') return; - // oxlint-disable-next-line typescript-eslint/only-throw-error - throw error; - } - - if (line.length > 0) { - lineNumber++; - const record = parseRecordLine(line, lineNumber, this.filePath, true); - if (record !== undefined) yield record; - } - } - - append(input: PersistedWireRecord): void { - this.throwIfError(); - this.pendingRecords.push(input); - this.scheduleFlush(); - } - - rewrite(records: readonly PersistedWireRecord[]): void { - this.throwIfError(); - this.shouldClear = true; - this.pendingRecords.splice(0, this.pendingRecords.length, ...records); - this.scheduleFlush(); - } - - async flush(): Promise { - this.throwIfError(); - while ( - this.flushPromise !== undefined || - this.shouldClear || - this.pendingRecords.length > 0 - ) { - await this.ensureFlush(); - this.throwIfError(); - } - } - - async close(): Promise { - await this.flush(); - } - - private scheduleFlush(): void { - void this.ensureFlush().catch((error) => { - this.options.onError?.(error); - }); - } - - private ensureFlush(): Promise { - if (this.flushPromise !== undefined) return this.flushPromise; - - const promise = this.drainPendingRecords() - .catch((error: unknown) => { - this.error = error; - // oxlint-disable-next-line typescript-eslint/only-throw-error - throw error; - }) - .finally(() => { - if (this.flushPromise === promise) { - this.flushPromise = undefined; - } - if ( - this.error === undefined && - (this.shouldClear || this.pendingRecords.length > 0) - ) { - this.scheduleFlush(); - } - }); - this.flushPromise = promise; - return promise; - } - - private throwIfError(): void { - // oxlint-disable-next-line typescript-eslint/only-throw-error - if (this.error !== undefined) throw this.error; - } - - private async drainPendingRecords(): Promise { - while (this.shouldClear || this.pendingRecords.length > 0) { - await this.drainBatch(); - } - } - - private async drainBatch(): Promise { - const shouldClear = this.shouldClear; - const batch = this.pendingRecords.splice(0); - this.shouldClear = false; - - const writable = this.options.beforeWrite === undefined - ? batch - : await Promise.all(batch.map((record) => Promise.resolve(this.options.beforeWrite!(record)))); - const content = writable.map((e) => JSON.stringify(e) + '\n').join(''); - const directory = dirname(this.filePath); - await mkdir(directory, { recursive: true }); - - const fh = await open(this.filePath, shouldClear ? 'w' : 'a'); - try { - if (content.length > 0) { - await fh.writeFile(content, 'utf8'); - } - await fh.sync(); - } finally { - await fh.close(); - } - - if (!this.directorySynced) { - await syncDir(directory); - this.directorySynced = true; - } - } -} - -function parseRecordLine( - line: string, - lineNumber: number, - filePath: string, - allowTruncated: boolean, -): PersistedWireRecord | undefined { - if (line.length === 0) return undefined; - try { - return JSON.parse(line) as PersistedWireRecord; - } catch (parseError) { - if (allowTruncated) return undefined; - throw new Error( - `wire.jsonl: corrupted line ${lineNumber} in ${filePath}: ${String(parseError)}`, - { cause: parseError }, - ); - } -} diff --git a/packages/agent-core-v2/src/wireRecord/wireRecord.ts b/packages/agent-core-v2/src/wireRecord/wireRecord.ts index b64844294..4e61570c1 100644 --- a/packages/agent-core-v2/src/wireRecord/wireRecord.ts +++ b/packages/agent-core-v2/src/wireRecord/wireRecord.ts @@ -19,14 +19,6 @@ export interface WireRecordMetadata { export type PersistedWireRecord = WireRecord | WireRecordMetadata | WireMigrationRecord; -export interface WireRecordPersistence { - read(): AsyncIterable; - append(input: PersistedWireRecord): void; - rewrite(records: readonly PersistedWireRecord[]): void; - flush(): Promise; - close(): Promise; -} - export interface WireRecordRestoringContext { readonly time?: number; } @@ -46,7 +38,6 @@ export interface WireRecordRestoreResult { export interface WireRecordServiceOptions { readonly homedir?: string; - readonly persistence?: WireRecordPersistence; readonly blobStore?: IBlobStoreService; readonly onPersistenceError?: ( error: unknown, diff --git a/packages/agent-core-v2/src/wireRecord/wireRecordService.ts b/packages/agent-core-v2/src/wireRecord/wireRecordService.ts index 0f70862cc..9a58d7c14 100644 --- a/packages/agent-core-v2/src/wireRecord/wireRecordService.ts +++ b/packages/agent-core-v2/src/wireRecord/wireRecordService.ts @@ -4,6 +4,8 @@ import { join } from 'pathe'; +import { createHash } from 'node:crypto'; + import { Disposable, toDisposable, @@ -12,6 +14,7 @@ import { IBlobStoreService, type BlobStoreServiceOptions, } from '#/blobStore'; +import { IAppendLogStore } from '#/storage'; import { BlobStoreService } from '../blobStore/blobStoreService'; import { OrderedHookSlot } from '../hooks'; import type { WireRecord, WireRecordMap } from '../wireRecord'; @@ -23,13 +26,11 @@ import { type WireMigration, type WireMigrationRecord, } from './migration'; -import { FileSystemWireRecordPersistence } from './persistence'; import { IWireRecord, type PersistedWireRecord, type WireRecordBlobSelector, type WireRecordMetadata, - type WireRecordPersistence, type WireRecordRegisterOptions, type WireRecordRestoredContext, type WireRecordRestoreOptions, @@ -47,7 +48,7 @@ export class WireRecordService extends Disposable implements IWireRecord { keyof WireRecordMap, BlobSelector[] >(); - private readonly persistence: WireRecordPersistence | undefined; + private readonly persistKey: string | undefined; private readonly blobStore: IBlobStoreService | undefined; private _restoring: { time?: number } | null = null; private _postRestoring = false; @@ -60,19 +61,14 @@ export class WireRecordService extends Disposable implements IWireRecord { constructor( private readonly options: WireRecordServiceOptions = {}, @IBlobStoreService injectedBlobStore?: IBlobStoreService, + @IAppendLogStore private readonly log?: IAppendLogStore, ) { super(); this.blobStore = this.resolveBlobStore(options, injectedBlobStore); - this.persistence = - options.persistence ?? - (options.homedir === undefined - ? undefined - : new FileSystemWireRecordPersistence(join(options.homedir, 'wire.jsonl'), { - beforeWrite: (record) => this.preparePersistentRecord(record), - onError: (error) => { - this.reportPersistenceError(error); - }, - })); + this.persistKey = options.homedir === undefined ? undefined : hashKey(options.homedir); + if (this.log !== undefined && this.persistKey !== undefined) { + this._register(this.log.acquire('wire', this.persistKey)); + } } get restoring() { @@ -119,7 +115,11 @@ export class WireRecordService extends Disposable implements IWireRecord { options: WireRecordRestoreOptions = {}, ): Promise { const fromPersistence = records === undefined; - const source = records ?? this.persistence?.read(); + const source = + records ?? + (this.log !== undefined && this.persistKey !== undefined + ? this.log.read('wire', this.persistKey) + : undefined); if (source === undefined) { await this.runResumeEndedHooks(); return {}; @@ -129,7 +129,8 @@ export class WireRecordService extends Disposable implements IWireRecord { fromPersistence && (options.rewriteMigratedRecords ?? true); const restoredRecords: PersistedWireRecord[] | undefined = rewriteMigratedRecords ? [] : undefined; - const requireMetadata = fromPersistence && this.persistence !== undefined; + const requireMetadata = + fromPersistence && this.log !== undefined && this.persistKey !== undefined; let migrations: readonly WireMigration[] = []; let shouldRewrite = false; let completed = true; @@ -181,9 +182,15 @@ export class WireRecordService extends Disposable implements IWireRecord { } } - if (completed && shouldRewrite && restoredRecords !== undefined) { - this.persistence?.rewrite(restoredRecords); - await this.persistence?.flush(); + if ( + completed && + shouldRewrite && + restoredRecords !== undefined && + this.log !== undefined && + this.persistKey !== undefined + ) { + this.log.rewrite('wire', this.persistKey, restoredRecords); + await this.log.flush(); } if (completed) { await this.runResumeEndedHooks(); @@ -192,40 +199,35 @@ export class WireRecordService extends Disposable implements IWireRecord { } async flush(): Promise { - await this.persistence?.flush(); + await this.log?.flush(); } async close(): Promise { - await this.persistence?.close(); + await this.log?.close(); } private appendPersistent(record: PersistedWireRecord): void { - if (this.persistence === undefined) return; + if (this.log === undefined || this.persistKey === undefined) return; if (!this.metadataInitialized && record.type !== 'metadata') { const metadata: WireRecordMetadata = { type: 'metadata', protocol_version: AGENT_WIRE_PROTOCOL_VERSION, created_at: Date.now(), }; - try { - this.persistence.append(metadata); - this.metadataInitialized = true; - } catch (error) { - this.reportPersistenceError(error, metadata); - // oxlint-disable-next-line typescript-eslint/only-throw-error - throw error; - } + this.log.append('wire', this.persistKey, metadata, { + onError: (error) => this.reportPersistenceError(error, metadata), + }); + this.metadataInitialized = true; } if (record.type === 'metadata') { this.metadataInitialized = true; } - try { - this.persistence.append(record); - } catch (error) { - this.reportPersistenceError(error, record); - // oxlint-disable-next-line typescript-eslint/only-throw-error - throw error; - } + void this.preparePersistentRecord(record).then((prepared) => { + if (this.log === undefined || this.persistKey === undefined) return; + this.log.append('wire', this.persistKey, prepared, { + onError: (error) => this.reportPersistenceError(error, prepared), + }); + }); } private async restoreRecord(record: WireRecord): Promise { @@ -354,3 +356,7 @@ registerScopedService( function isWireRecordMetadata(record: PersistedWireRecord): record is WireRecordMetadata { return record.type === 'metadata' && typeof record['protocol_version'] === 'string'; } + +function hashKey(homedir: string): string { + return createHash('sha256').update(homedir).digest('hex').slice(0, 16); +} diff --git a/packages/agent-core-v2/src/workspaceRegistry/workspaceRegistryService.ts b/packages/agent-core-v2/src/workspaceRegistry/workspaceRegistryService.ts index d5d2ef2e2..20aacc73c 100644 --- a/packages/agent-core-v2/src/workspaceRegistry/workspaceRegistryService.ts +++ b/packages/agent-core-v2/src/workspaceRegistry/workspaceRegistryService.ts @@ -2,7 +2,7 @@ * `workspaceRegistry` domain (L1) — `IWorkspaceRegistry` implementation. * * In-memory skeleton of the known-workspaces catalog; persistence through - * `IConfigStore` will replace the map in a later phase. Bound at Core scope. + * `IAtomicDocumentStore` will replace the map in a later phase. Bound at Core scope. */ import { createHash } from 'node:crypto'; diff --git a/packages/agent-core-v2/test/sessionStore/sessionStore.test.ts b/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts similarity index 73% rename from packages/agent-core-v2/test/sessionStore/sessionStore.test.ts rename to packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts index 5c4839d73..f6b977528 100644 --- a/packages/agent-core-v2/test/sessionStore/sessionStore.test.ts +++ b/packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts @@ -8,8 +8,8 @@ import { InstantiationType } from '#/_base/di/extensions'; import { LifecycleScope, _clearScopedRegistryForTests, registerScopedService } from '#/_base/di/scope'; import { createScopedTestHost, stubPair } from '#/_base/di/test'; import { HostFileSystem, IHostFileSystem } from '#/hostFs'; -import { ISessionStore } from '#/sessionStore/sessionStore'; -import { SessionStore, encodeWorkDirKey } from '#/sessionStore/sessionStoreService'; +import { ISessionIndex } from '#/sessionIndex/sessionIndex'; +import { FileSessionIndex, encodeWorkDirKey } from '#/sessionIndex/sessionIndexService'; describe('encodeWorkDirKey', () => { it('is deterministic and path-sensitive', () => { @@ -22,13 +22,13 @@ describe('encodeWorkDirKey', () => { }); }); -describe('SessionStore workspace helpers', () => { +describe('FileSessionIndex workspace helpers', () => { let sessionsRoot: string; let disposeHost: (() => void) | undefined; beforeEach(async () => { _clearScopedRegistryForTests(); - registerScopedService(LifecycleScope.Core, ISessionStore, SessionStore, InstantiationType.Delayed, 'records'); + registerScopedService(LifecycleScope.Core, ISessionIndex, FileSessionIndex, InstantiationType.Delayed, 'sessionIndex'); sessionsRoot = await fsp.mkdtemp(join(os.tmpdir(), 'ws-sessions-')); }); @@ -38,10 +38,10 @@ describe('SessionStore workspace helpers', () => { await fsp.rm(sessionsRoot, { recursive: true, force: true }); }); - function build(): ISessionStore { + function build(): ISessionIndex { const host = createScopedTestHost([stubPair(IHostFileSystem, new HostFileSystem())]); disposeHost = () => host.dispose(); - return host.core.accessor.get(ISessionStore); + return host.core.accessor.get(ISessionIndex); } it('workspaceIdFor matches encodeWorkDirKey', () => { @@ -50,7 +50,7 @@ describe('SessionStore workspace helpers', () => { expect(store.workspaceIdFor(workDir)).toBe(encodeWorkDirKey(workDir)); }); - it('countActiveSessions counts non-archived session dirs', async () => { + it('countActive counts non-archived session dirs', async () => { const store = build(); const workDir = '/home/user/repo'; const wsDir = join(sessionsRoot, encodeWorkDirKey(workDir)); @@ -63,11 +63,11 @@ describe('SessionStore workspace helpers', () => { await fsp.mkdir(join(wsDir, 'no-state'), { recursive: true }); - expect(await store.countActiveSessions(sessionsRoot, workDir)).toBe(2); + expect(await store.countActive(sessionsRoot, workDir)).toBe(2); }); - it('countActiveSessions returns 0 when the work dir has no sessions yet', async () => { + it('countActive returns 0 when the work dir has no sessions yet', async () => { const store = build(); - expect(await store.countActiveSessions(sessionsRoot, '/home/user/never-created')).toBe(0); + expect(await store.countActive(sessionsRoot, '/home/user/never-created')).toBe(0); }); }); diff --git a/packages/agent-core-v2/test/sessionMetaStore/sessionMetaStore.test.ts b/packages/agent-core-v2/test/sessionMetaStore/sessionMetaStore.test.ts index 8a895b0fe..0b575c8f8 100644 --- a/packages/agent-core-v2/test/sessionMetaStore/sessionMetaStore.test.ts +++ b/packages/agent-core-v2/test/sessionMetaStore/sessionMetaStore.test.ts @@ -6,7 +6,7 @@ import { TestInstantiationService } from '#/_base/di/test'; import { ILogService } from '#/log'; import { ISessionMetaStore } from '#/sessionMetaStore'; import { SessionMetaStore } from '#/sessionMetaStore/sessionMetaStoreService'; -import { ConfigStore, IConfigStore, IStorageService, InMemoryStorageService } from '#/storage'; +import { AtomicDocumentStore, IAtomicDocumentStorage, IAtomicDocumentStore, InMemoryStorageService } from '#/storage'; import { stubLog } from '../log/stubs'; @@ -18,8 +18,8 @@ describe('SessionMetaStore', () => { disposables = new DisposableStore(); ix = disposables.add(new TestInstantiationService()); ix.stub(ILogService, stubLog()); - ix.set(IStorageService, new SyncDescriptor(InMemoryStorageService)); - ix.set(IConfigStore, new SyncDescriptor(ConfigStore)); + ix.set(IAtomicDocumentStorage, new SyncDescriptor(InMemoryStorageService)); + ix.set(IAtomicDocumentStore, new SyncDescriptor(AtomicDocumentStore)); ix.set(ISessionMetaStore, new SyncDescriptor(SessionMetaStore)); }); diff --git a/packages/agent-core-v2/test/storage/recordStore.test.ts b/packages/agent-core-v2/test/storage/appendLogStore.test.ts similarity index 54% rename from packages/agent-core-v2/test/storage/recordStore.test.ts rename to packages/agent-core-v2/test/storage/appendLogStore.test.ts index 94463b46d..b3b23ca70 100644 --- a/packages/agent-core-v2/test/storage/recordStore.test.ts +++ b/packages/agent-core-v2/test/storage/appendLogStore.test.ts @@ -3,9 +3,9 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; -import { IRecordStore, IStorageService, RecordCorruptedError } from '#/storage'; +import { AppendLogCorruptedError, IAppendLogStorage, IAppendLogStore, IStorageService } from '#/storage'; +import { AppendLogStore } from '#/storage/appendLogStore'; import { InMemoryStorageService } from '#/storage/inMemoryStorageService'; -import { RecordStore } from '#/storage/recordStore'; const enc = new TextEncoder(); @@ -16,19 +16,35 @@ interface Rec { const SCOPE = 'agents/main'; const KEY = 'wire.jsonl'; -describe('RecordStore', () => { +function chunkedStorage(chunks: Uint8Array[]): IStorageService { + return { + _serviceBrand: undefined, + read: async () => undefined, + readStream: async function* () { + for (const c of chunks) yield c; + }, + write: async () => {}, + append: async () => {}, + list: async () => [], + delete: async () => {}, + flush: async () => {}, + close: async () => {}, + }; +} + +describe('AppendLogStore', () => { let disposables: DisposableStore; let ix: TestInstantiationService; let storage: InMemoryStorageService; - let record: IRecordStore; + let record: IAppendLogStore; beforeEach(() => { disposables = new DisposableStore(); ix = disposables.add(new TestInstantiationService()); storage = new InMemoryStorageService(); - ix.stub(IStorageService, storage); - ix.set(IRecordStore, new SyncDescriptor(RecordStore)); - record = ix.get(IRecordStore); + ix.stub(IAppendLogStorage, storage); + ix.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); + record = ix.get(IAppendLogStore); }); afterEach(() => disposables.dispose()); @@ -91,10 +107,43 @@ describe('RecordStore', () => { expect(await collect(SCOPE, KEY)).toEqual([{ n: 1 }]); }); - it('throws RecordCorruptedError on a corrupted middle line', async () => { + it('throws AppendLogCorruptedError on a corrupted middle line', async () => { const raw = `${JSON.stringify({ n: 1 })}\nGARBAGE\n${JSON.stringify({ n: 3 })}\n`; await storage.append(SCOPE, KEY, enc.encode(raw)); - await expect(collect(SCOPE, KEY)).rejects.toBeInstanceOf(RecordCorruptedError); + await expect(collect(SCOPE, KEY)).rejects.toBeInstanceOf(AppendLogCorruptedError); + }); + + it('reads across chunk boundaries (stream read splits lines)', async () => { + const full = `${JSON.stringify({ n: 1 })}\n${JSON.stringify({ n: 2 })}\n${JSON.stringify({ n: 3 })}\n`; + const bytes = enc.encode(full); + // Split into chunks that cut through the middle of lines. + const chunks = [bytes.slice(0, 7), bytes.slice(7, 23), bytes.slice(23)]; + const localIx = disposables.add(new TestInstantiationService()); + localIx.stub(IAppendLogStorage, chunkedStorage(chunks)); + localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); + const log = localIx.get(IAppendLogStore); + + const out: Rec[] = []; + for await (const r of log.read(SCOPE, KEY)) out.push(r); + expect(out).toEqual([{ n: 1 }, { n: 2 }, { n: 3 }]); + }); + + it('reads across chunk boundaries with multi-byte UTF-8 split', async () => { + const full = `${JSON.stringify({ n: 1, s: '中文' })}\n${JSON.stringify({ n: 2, s: '日本語' })}\n`; + const bytes = enc.encode(full); + // Split at every byte to maximally stress multi-byte decode across chunks. + const chunks = Array.from(bytes, (b) => new Uint8Array([b])); + const localIx = disposables.add(new TestInstantiationService()); + localIx.stub(IAppendLogStorage, chunkedStorage(chunks)); + localIx.set(IAppendLogStore, new SyncDescriptor(AppendLogStore)); + const log = localIx.get(IAppendLogStore); + + const out: Array = []; + for await (const r of log.read(SCOPE, KEY)) out.push(r); + expect(out).toEqual([ + { n: 1, s: '中文' }, + { n: 2, s: '日本語' }, + ]); }); }); diff --git a/packages/agent-core-v2/test/storage/configStore.test.ts b/packages/agent-core-v2/test/storage/atomicDocumentStore.test.ts similarity index 89% rename from packages/agent-core-v2/test/storage/configStore.test.ts rename to packages/agent-core-v2/test/storage/atomicDocumentStore.test.ts index 37d509183..d1bb2dd1c 100644 --- a/packages/agent-core-v2/test/storage/configStore.test.ts +++ b/packages/agent-core-v2/test/storage/atomicDocumentStore.test.ts @@ -3,8 +3,8 @@ import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; -import { IConfigStore, IStorageService } from '#/storage'; -import { ConfigStore } from '#/storage/configStore'; +import { IAtomicDocumentStorage, IAtomicDocumentStore } from '#/storage'; +import { AtomicDocumentStore } from '#/storage/atomicDocumentStore'; import { InMemoryStorageService } from '#/storage/inMemoryStorageService'; interface State { @@ -12,19 +12,19 @@ interface State { readonly count?: number; } -describe('ConfigStore', () => { +describe('AtomicDocumentStore', () => { let disposables: DisposableStore; let ix: TestInstantiationService; let storage: InMemoryStorageService; - let config: IConfigStore; + let config: IAtomicDocumentStore; beforeEach(() => { disposables = new DisposableStore(); ix = disposables.add(new TestInstantiationService()); storage = new InMemoryStorageService(); - ix.stub(IStorageService, storage); - ix.set(IConfigStore, new SyncDescriptor(ConfigStore)); - config = ix.get(IConfigStore); + ix.stub(IAtomicDocumentStorage, storage); + ix.set(IAtomicDocumentStore, new SyncDescriptor(AtomicDocumentStore)); + config = ix.get(IAtomicDocumentStore); }); afterEach(() => disposables.dispose());