@pai/storage-fs

Package: @pai/storage-fs

Persists each thread as a single JSON file under a root directory. Suitable for local development, single-process services, and scripts that need state to survive across restarts.

Single-host only. Lock directories coordinate store instances that share the same local filesystem. Use a transactional managed backend for multi-host workers, network filesystems with uncertain rename/lock semantics, or stronger operational recovery guarantees.

Quick start

import { createAgentRuntime } from "@pai/core";
import { createFileSystemThreadStore } from "@pai/storage-fs";

const runtime = createAgentRuntime({
  agent,
  scopeKey: (identity) => identity.workspaceId,
  storage: createFileSystemThreadStore({
    rootDir: "./.pai/threads",
  }),
});

Options

type FileSystemThreadStoreOptions = {
  /**
   * Root directory for persisted threads. Created on demand. Threads are
   * stored below SHA-256 hashes of both scopeKey and opaque threadId.
   */
  rootDir: string;

  /** Override clock for tests. Defaults to `() => new Date()`. */
  now?: () => Date;
};

The factory throws if rootDir is missing or empty.

On-disk layout

<rootDir>/
  <sha256(scopeKey)>/
    <sha256(threadId)>.json

Each file contains a schema-versioned aggregate: the thread record, transcript messages, queued items, run lifecycle records, and optional run/admin lease. Writes use temp-file replacement, so concurrent readers never see a partial document.

The current and only accepted document schemaVersion is 1. The provider validates the complete run-centered logical aggregate on every read. This development format is strict: the schema number alone does not make an older document compatible with the current required record shape.

The thread root and every run in the aggregate carry both required strict-JSON objects: server-only privateMetadata and server-written, client-readable metadata. A file missing either field on either record is invalid current data.

Both identity components are hashed, while the original values remain inside the validated document. ThreadId is therefore an opaque portable string, not a filesystem path or filename.

Concurrency

Each commitThread() acquires a per-thread lock directory. Concurrent compare-and-swap commits serialize, then re-read the current version before applying their write set. Reads do not take the lock: atomic replacement means a reader sees either the previous aggregate or the next one.

Recovering a lock whose same-host process has died leaves an owner-specific .stale-* directory as a durable race fence. Do not remove these directories while another process may still be accessing the same rootDir; deleting one could let a delayed recovery attempt move a newer live lock.

The store seeds its monotonic clock from the largest persisted updatedAt it observes on load. New writes after a restart always produce timestamps strictly greater than anything already on disk, even if the system clock has moved backwards.

Conformance

createFileSystemThreadStore passes the full @pai/storage/test conformance suite. The package's tests create a fresh temp directory per createStore call:

import { mkdtemp, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { createThreadStoreConformanceSuite } from "@pai/storage/test";
import { createFileSystemThreadStore } from "@pai/storage-fs";

const dirs: string[] = [];
afterAll(() =>
  Promise.all(dirs.map((d) => rm(d, { recursive: true, force: true }))),
);

createThreadStoreConformanceSuite({
  name: "createFileSystemThreadStore",
  createStore: async () => {
    const dir = await mkdtemp(join(tmpdir(), "pai-storage-fs-"));
    dirs.push(dir);
    return createFileSystemThreadStore({ rootDir: dir });
  },
});