@pai/storage

Package: @pai/storage

@pai/storage defines the durable provider boundary used by createAgentRuntime(). Application code normally configures a built-in provider and uses Thread, RunHandle, and lifecycle facades. Provider authors implement ThreadStore.

import { createAgentRuntime } from "@pai/core";
import { createProductionThreadStore } from "./thread-storage";

const runtime = createAgentRuntime({
  agent,
  scopeKey: (identity) => identity.workspaceId,
  storage: createProductionThreadStore({
    url: process.env.DATABASE_URL!,
  }),
});

The store persists thread roots, transcript messages, queued items, run lifecycle records, and thread leases. It does not implement scheduling, cancellation, queue admission, regeneration, recovery, or retry policy. Those decisions belong to the runtime.

ThreadStore

The provider surface has three operations and optional cleanup:

interface ThreadStore {
  readThread(input: ReadThreadInput): Promise<ThreadReadResult>;
  listThreads(input: ListThreadsInput): Promise<Page<ThreadRecord>>;
  commitThread(input: CommitThreadInput): Promise<CommitThreadResult>;
  close?(): Promise<void>;
}

readThread() performs a consistent selective read. listThreads() pages root records within one scope. commitThread() atomically compares the current version and applies one complete thread-scoped write set. The runtime composes these storage primitives into higher-level workflows.

Identity And Versioning

type ThreadKey = {
  scopeKey: ScopeKey;
  threadId: ThreadId;
};

type ThreadVersion = {
  threadInstanceId: string;
  revision: number;
};

type PageToken = string;

Every operation is scoped by both scopeKey and threadId. ThreadId is an opaque application string; providers must safely encode physical keys rather than impose filesystem- or database-specific syntax.

threadInstanceId is immutable for one persisted incarnation. revision increases within that incarnation. Deleting and recreating the same { scopeKey, threadId } produces a new instance id, so stale work cannot cross the deletion boundary.

Lazy model-context reads capture the owning agent, immutable thread instance, and newest physical message. Every later page remains below that upper anchor and validates the thread incarnation. The active run lease serializes model context resolution against regeneration and administrative history rewrites; resolver hooks must not rewrite the message visibility they are selecting. Deletion and recreation fail the old attempt instead of reading the replacement thread.

PageToken is a separate opaque continuation value. It is not a ThreadVersion, revision, or realtime replay cursor. A provider binds each token to its original scope, selection, filters, and ordering.

Selective Reads

type ReadThreadInput = ThreadKey & {
  include?: {
    messages?: MessageSelection;
    runs?: RunSelection;
    queuedItems?: QueuedItemSelection;
  };
};

type ThreadReadResult = {
  /** Timestamp from the provider's authoritative clock. */
  observedAt: string;
  thread: ThreadRecord | null;
  messages?: ThreadPage<PaiStoredMessageEnvelope>;
  runs?: ThreadPage<RunRecord>;
  queuedItems?: ThreadPage<StoredQueuedItemRecord>;
};

type ThreadPage<T> = Page<T> & {
  version: ThreadVersion;
};

An existing result comes from one committed version. Every requested child family is returned, even when empty; unrequested families are omitted. A missing root returns thread: null and no child pages.

Selections support either bounded stable-id lookup or pagination:

type MessageSelection =
  | { type: "ids"; messageIds: readonly MessageId[] }
  | {
      type: "page";
      view: "raw" | "transcript" | "context";
      order?: "oldest" | "newest";
      afterMessageId?: MessageId;
      beforeMessageId?: MessageId;
      page: { limit: number; pageToken?: PageToken };
    };

Run and queued-item selections use the same explicit continuation model. Every page repeats the same filters, order, and boundaries. The opaque token adds only the continuation position and providers reject it when the repeated query does not match. Use THREAD_STORE_MAX_PAGE_SIZE and THREAD_STORE_MAX_ID_SELECTION_SIZE when validating requests.

Listing Threads

listThreads() returns root records newest-update first. Every page repeats the same agent, parentThreadId, category, and includeHidden filters alongside the same scopeKey. Callers may choose a different bounded limit for a continuation page.

type ListThreadsInput = {
  scopeKey: ScopeKey;
  limit: number;
  agent?: string;
  parentThreadId?: ThreadId;
  category?: string;
  includeHidden?: boolean;
  pageToken?: PageToken;
};

Atomic Commits

type CommitThreadInput =
  | (ThreadKey & {
      condition: { type: "missing" };
      mutation: CreateThreadMutation;
    })
  | (ThreadKey & {
      condition: { type: "version"; version: ThreadVersion };
      mutation:
        | UpdateThreadMutation
        | { type: "delete" };
    });

type CommitThreadResult =
  | {
      type: "committed";
      thread: ThreadRecord | null;
      committedAt: string;
    }
  | {
      type: "conflict";
      currentVersion: ThreadVersion | null;
    };

A commit replaces the mutable PAI-owned root fields and applies strict message, run, and queued-item changes as one transaction. The thread's owning agent and directExecutionAllowed authority are supplied only at creation and must be retained for the lifetime of that thread incarnation. Ordered child families support insert, replace, delete, and message/run truncation at an existing boundary. Lease changes are part of the same write set.

Version or lease-precondition mismatch is an expected conflict result; no part of the mutation is applied. Malformed mutations and impossible persisted state remain invariant errors. Providers stamp threadInstanceId, ThreadVersion.revision, commit timestamps, and lease timestamps from provider-owned state. Runtime callers request a lease duration rather than calculate an expiry using their own clock. Acquisition may replace a lease that is expired at commit time, while renewal requires the matching lease to remain unexpired at commit time. Run acquisition also supplies the current execution operation (send, trigger, regenerate, or resume); the provider persists it on the lease so stale recovery reports the correct episode. Every renewal and release is owner-fenced. A release also declares whether it represents active work, which requires an unexpired lease, or stale-state cleanup, which may release an expired lease:

changes: {
  lease: {
    type: "release",
    lease: { kind: "run", runId },
    leaseOwnerId: workerId,
    expiry: "unexpired", // use "any" only for explicit recovery cleanup
  },
}

Run-centered records add aggregate invariants to that atomic boundary:

• every run-associated message references an existing run;
• messages for one admitted run form one contiguous transcript interval;
• that interval starts with exactly the run's ordered inputMessageIds, all carrying relation: "input";
• context and output may follow inputs, but input cannot appear later;
• queued and cancelled-before-admission runs own no transcript messages;
• a running or waiting run owns the transcript tail, and at most one such open-tail run exists;
• run lifecycle, queue admission, and lease changes remain mutually consistent.

Providers validate these relationships over the post-mutation aggregate, not merely record by record. Indexed backends may use targeted relationship checks when a mutation cannot affect ordering or ownership.

Persisted Values

Provider-facing stored values are strict JSON. Optional properties are omitted instead of written as undefined; non-finite numbers, negative zero, functions, symbols, cycles, class instances, raw JSON wrappers, and other values that do not round-trip losslessly are rejected before commit. Wire/client DTOs are separate projections and should not be persisted as a shortcut.

The exported portable failures are:

• ThreadStorePageTokenError for invalid or mismatched continuation tokens;
• ThreadStoreInvariantError for invalid mechanical changes;
• ThreadStoreSerializationError for non-JSON persisted input;
• ThreadStoreCommitUnknownError when a remote provider cannot determine whether a commit succeeded.

After ThreadStoreCommitUnknownError, reread and replan. Do not blindly repeat the same stale write set.

Provider Responsibilities

A conforming provider must:

• isolate every operation by scope and opaque thread id;
• make selected reads consistent at one ThreadVersion;
• keep pages stable and bind tokens to their original authority-bearing filters, order, and boundaries while allowing a different bounded page size;
• preserve requested ID order and omit missing IDs;
• compare the full version, including threadInstanceId;
• retain creation-time agent and directExecutionAllowed values on updates;
• apply every root, child, and lease change atomically or none;
• preserve message producer and explicit run/standalone association;
• preserve run initiator, admission, ordered inputMessageIds, and both strict-JSON metadata bags;
• preserve thread-root server-only privateMetadata and client-readable metadata while omitting the private bag from public projections;
• enforce the cross-record run/message/queue/lease invariants above;
• return the ordinary conflict result when a well-formed version or lease precondition is false;
• advance ThreadVersion.revision exactly once for a successful create or update;
• use its authoritative clock for observations, commits, and lease expiry;
• reject invalid JSON and persisted invariants without partial writes;
• create a new instance id after deletion and recreation.

Remote backends that cannot atomically compare and apply a thread write set cannot safely implement this contract.

Helpers And Conformance

@pai/storage/thread-store exports assertThreadCommitInput() for mechanical input validation. Aggregate-style implementations can use applyThreadChanges() to validate and apply a complete changeset, while normalized implementations can use applyThreadRootUpdate() to stamp a root update before applying child changes in their own transaction. These helpers do not provide locking or durable atomicity; the provider still owns those. The same entrypoint exports strict parsers for individual logical records and complete thread aggregates.

For bounded normalized writes, planThreadRelationshipValidation() owns the framework's topology-diff and affected-ID rules. Providers pass canonical projections of the previous rows touched by the write plus truncation impacts, then keep the planned database queries and final relationship assertions local to their atomic transaction.

Providers can also share canonical selection planning, result finishing, and the authority-bound pagination codec:

import { createHash } from "node:crypto";
import { createThreadStoreSelectionHelpers } from "@pai/storage/thread-store";

const selections = createThreadStoreSelectionHelpers({
  tokenPrefix: "my-store-page:",
  providerName: "My store",
  scopeBinding: {
    field: "scopeDigest",
    digest: (scopeKey) =>
      createHash("sha256").update(JSON.stringify(scopeKey)).digest("base64url"),
  },
});

Create one planRead(input) or planThreadList(input) before loading records. Normalized backends apply the validated filters and cursor in a bounded query, load at most candidateLimit, and pass those provider-filtered records to the corresponding finish* helper. Aggregate-style backends can instead use the select* helpers to filter loaded canonical families. The provider supplies a deterministic, stable scope binding and still owns consistent reads, physical queries, and atomic commits.

Every provider should run the shared conformance suite against its real backend:

import { createThreadStoreConformanceSuite } from "@pai/storage/test";
import { createMyThreadStore } from "../src/index.js";

createThreadStoreConformanceSuite({
  name: "my provider",
  createStore: () => createMyThreadStore(),
});

The test entrypoint is for provider tests only and should not be imported by runtime code.