vrr/kimi-code
2
0
Fork 0
mirror of https://github.com/MoonshotAI/kimi-code.git synced 2026-08-20 06:05:36 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

refactor(agent-core-v2): decouple persistence from filesystem via Store/Storage layering

Browse source 
- rename IRecordStore → IAppendLogStore and IConfigStore → IAtomicDocumentStore (name Stores by access pattern, not business)
- add IAppendLogStorage / IAtomicDocumentStorage tokens — same IStorageService interface, distinct tokens so each access pattern can bind a different backend
- add Store.acquire(scope, key) → IDisposable for flush-on-dispose; add IStorageService.readStream for streaming reads
- delete WireRecordPersistence; WireRecordService now depends directly on IAppendLogStore (preparePersistentRecord inlined, onError per-append)
- BlobStoreService now depends on IHostFileSystem (new createExclusive primitive) instead of node:fs
- rename sessionStore → sessionIndex as a domain-specific Store over hostFs
- document the "business code never persists directly" rule in AGENTS.md and the new persistence subskill
This commit is contained in:
haozhe.yang 2026-06-27 14:07:42 +08:00
parent a9d67f4514
commit ddab6a2293
29 changed files with 568 additions and 411 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.agents/skills/agent-core-dev/SKILL.md
View file

@ -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.

173
.agents/skills/agent-core-dev/persistence.md Normal file
View file

@ -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<Uint8Array | undefined>;
write(scope: string, key: string, data: Uint8Array, options?: { atomic?: boolean }): Promise<void>;
append(scope: string, key: string, data: Uint8Array, options?: { durable?: boolean }): Promise<void>;
list(scope: string, prefix?: string): Promise<readonly string[]>;
delete(scope: string, key: string): Promise<void>;
flush(): Promise<void>;
close(): Promise<void>;
}
export const IAppendLogStorage = createDecorator<IStorageService>('appendLogStorage');
export const IAtomicDocumentStorage = createDecorator<IStorageService>('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.

11
packages/agent-core-v2/AGENTS.md
View file

@ -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/`.

30
packages/agent-core-v2/src/blobStore/blobStoreService.ts
View file

@ -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<string, number>();
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<string> {
@ -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}`;
}

7
packages/agent-core-v2/src/hostFs/hostFs.ts
View file

@ -29,6 +29,13 @@ export interface IHostFileSystem {
writeText(path: string, data: string): Promise<void>;
readBytes(path: string): Promise<Uint8Array>;
writeBytes(path: string, data: Uint8Array): Promise<void>;
/**
* 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<boolean>;
stat(path: string): Promise<HostFileStat>;
readdir(path: string): Promise<readonly HostDirEntry[]>;
mkdir(path: string, options?: { readonly recursive?: boolean }): Promise<void>;

18
packages/agent-core-v2/src/hostFs/hostFsService.ts
View file

@ -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<boolean> {
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<HostFileStat> {
const s = await stat(path);
return { isFile: s.isFile(), isDirectory: s.isDirectory(), size: s.size };

2
packages/agent-core-v2/src/index.ts
View file

@ -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';

6
packages/agent-core-v2/src/sessionIndex/index.ts Normal file
View file

@ -0,0 +1,6 @@
/**
* `sessionIndex` domain barrel.
*/
export * from './sessionIndex';
export * from './sessionIndexService';

24
packages/agent-core-v2/src/sessionIndex/sessionIndex.ts Normal file
View file

@ -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_<slug>_<hash>` 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<number>;
}
export const ISessionIndex: ServiceIdentifier<ISessionIndex> =
createDecorator<ISessionIndex>('sessionIndex');

20
packages/agent-core-v2/src/sessionStore/sessionStoreService.ts → packages/agent-core-v2/src/sessionIndex/sessionIndexService.ts
View file

@ -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<number> {
async countActive(sessionsRoot: string, workDir: string): Promise<number> {
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',
);

13
packages/agent-core-v2/src/sessionMetaStore/sessionMetaStoreService.ts
View file

@ -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<Record<string, unknown>> {
this.data =
(await this.configStore.get<Record<string, unknown>>(SCOPE, this.key)) ?? {};
(await this.documentStore.get<Record<string, unknown>>(SCOPE, this.key)) ?? {};
return this.data;
}
@ -41,7 +42,7 @@ export class SessionMetaStore extends Disposable implements ISessionMetaStore {
}
async flush(): Promise<void> {
await this.configStore.set(SCOPE, this.key, this.data);
await this.documentStore.set(SCOPE, this.key, this.data);
}
}

6
packages/agent-core-v2/src/sessionStore/index.ts
View file

@ -1,6 +0,0 @@
/**
* `sessionStore` domain barrel.
*/
export * from './sessionStore';
export * from './sessionStoreService';

18
packages/agent-core-v2/src/sessionStore/sessionStore.ts
View file

@ -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_<slug>_<hash>` 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<number>;
}
export const ISessionStore: ServiceIdentifier<ISessionStore> =
createDecorator<ISessionStore>('sessionStore');

114
packages/agent-core-v2/src/storage/recordStore.ts → packages/agent-core-v2/src/storage/appendLogStore.ts
View file

@ -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<R>(scope: string, key: string, record: R): void;
append<R>(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<void>;
/**
* 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<IRecordStore> =
createDecorator<IRecordStore>('recordStore');
export const IAppendLogStore: ServiceIdentifier<IAppendLogStore> =
createDecorator<IAppendLogStore>('appendLogStore');
interface LogState {
pending: unknown[];
flushPromise: Promise<void> | 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<string, LogState>();
constructor(@IStorageService private readonly storage: IStorageService) {}
constructor(@IAppendLogStorage private readonly storage: IStorageService) {}
append<R>(scope: string, key: string, record: R): void {
append<R>(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<R>(scope: string, key: string): AsyncIterable<R> {
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<R>(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<R>(pending, scope, key, lineNumber, true);
if (record !== undefined) yield record;
}
}
private parseLine<R>(
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<R>(scope: string, key: string, records: readonly R[]): Promise<void> {
@ -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',
);

47
packages/agent-core-v2/src/storage/configStore.ts → packages/agent-core-v2/src/storage/atomicDocumentStore.ts
View file

@ -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<readonly string[]>;
/**
* 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<IConfigStore> =
createDecorator<IConfigStore>('configStore');
export const IAtomicDocumentStore: ServiceIdentifier<IAtomicDocumentStore> =
createDecorator<IAtomicDocumentStore>('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<T>(scope: string, key: string): Promise<T | undefined> {
const bytes = await this.storage.read(scope, key);
@ -63,12 +74,16 @@ export class ConfigStore implements IConfigStore {
async list(scope: string, prefix?: string): Promise<readonly string[]> {
return this.storage.list(scope, prefix);
}
acquire(scope: string, key: string): IDisposable {
return toDisposable(() => {});
}
}
registerScopedService(
LifecycleScope.Session,
IConfigStore,
ConfigStore,
IAtomicDocumentStore,
AtomicDocumentStore,
InstantiationType.Delayed,
'storage',
);

13
packages/agent-core-v2/src/storage/fileStorageService.ts
View file

@ -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<Uint8Array> {
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,

23
packages/agent-core-v2/src/storage/inMemoryStorageService.ts
View file

@ -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<Uint8Array> {
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',
);

4
packages/agent-core-v2/src/storage/index.ts
View file

@ -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';

9
packages/agent-core-v2/src/storage/queryStore.ts
View file

@ -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;

27
packages/agent-core-v2/src/storage/storageService.ts
View file

@ -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<Uint8Array | undefined>;
/**
* 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<Uint8Array>;
/** Atomically replace the whole value. */
write(scope: string, key: string, data: Uint8Array, options?: StorageWriteOptions): Promise<void>;
@ -69,3 +76,21 @@ export interface IStorageService {
export const IStorageService: ServiceIdentifier<IStorageService> =
createDecorator<IStorageService>('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<IStorageService> =
createDecorator<IStorageService>('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<IStorageService> =
createDecorator<IStorageService>('atomicDocumentStorage');

1
packages/agent-core-v2/src/wireRecord/index.ts
View file

@ -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';

216
packages/agent-core-v2/src/wireRecord/persistence.ts
View file

@ -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<PersistedWireRecord>;
}
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<PersistedWireRecord> {
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<void> {}
async close(): Promise<void> {}
}
export class FileSystemWireRecordPersistence implements WireRecordPersistence {
private readonly pendingRecords: PersistedWireRecord[] = [];
private shouldClear = false;
private directorySynced = false;
private flushPromise: Promise<void> | undefined;
private error: unknown;
constructor(
private readonly filePath: string,
private readonly options: FileSystemWireRecordPersistenceOptions = {},
) {}
async *read(): AsyncIterable<PersistedWireRecord> {
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<void> {
this.throwIfError();
while (
this.flushPromise !== undefined ||
this.shouldClear ||
this.pendingRecords.length > 0
) {
await this.ensureFlush();
this.throwIfError();
}
}
async close(): Promise<void> {
await this.flush();
}
private scheduleFlush(): void {
void this.ensureFlush().catch((error) => {
this.options.onError?.(error);
});
}
private ensureFlush(): Promise<void> {
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<void> {
while (this.shouldClear || this.pendingRecords.length > 0) {
await this.drainBatch();
}
}
private async drainBatch(): Promise<void> {
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 },
);
}
}

9
packages/agent-core-v2/src/wireRecord/wireRecord.ts
View file

@ -19,14 +19,6 @@ export interface WireRecordMetadata {
export type PersistedWireRecord = WireRecord | WireRecordMetadata | WireMigrationRecord;
export interface WireRecordPersistence {
read(): AsyncIterable<PersistedWireRecord>;
append(input: PersistedWireRecord): void;
rewrite(records: readonly PersistedWireRecord[]): void;
flush(): Promise<void>;
close(): Promise<void>;
}
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,

78
packages/agent-core-v2/src/wireRecord/wireRecordService.ts
View file

@ -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<keyof WireRecordMap>[]
>();
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<WireRecordRestoreResult> {
const fromPersistence = records === undefined;
const source = records ?? this.persistence?.read();
const source =
records ??
(this.log !== undefined && this.persistKey !== undefined
? this.log.read<PersistedWireRecord>('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<void> {
await this.persistence?.flush();
await this.log?.flush();
}
async close(): Promise<void> {
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<boolean> {
@ -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);
}

2
packages/agent-core-v2/src/workspaceRegistry/workspaceRegistryService.ts
View file

@ -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';

20
packages/agent-core-v2/test/sessionStore/sessionStore.test.ts → packages/agent-core-v2/test/sessionIndex/sessionIndex.test.ts
View file

@ -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);
});
});

6
packages/agent-core-v2/test/sessionMetaStore/sessionMetaStore.test.ts
View file

@ -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));
});

67
packages/agent-core-v2/test/storage/recordStore.test.ts → packages/agent-core-v2/test/storage/appendLogStore.test.ts
View file

@ -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<Rec>(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<Rec>(SCOPE, KEY)).rejects.toBeInstanceOf(RecordCorruptedError);
await expect(collect<Rec>(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<Rec>(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<Rec & { s?: string }> = [];
for await (const r of log.read<Rec & { s?: string }>(SCOPE, KEY)) out.push(r);
expect(out).toEqual([
{ n: 1, s: '中文' },
{ n: 2, s: '日本語' },
]);
});
});

14
packages/agent-core-v2/test/storage/configStore.test.ts → packages/agent-core-v2/test/storage/atomicDocumentStore.test.ts
View file

@ -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());