feat(durable): add Postgres storage backend

packages/durable/CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## [Unreleased]
 
+### Changed
+
+- Postgres storage is built for a database that is a network hop away. A commit is two round trips: one statement reads the session row, the batch's claimed ids, its documents, and their latest revision versions as JSON; one data-modifying CTE statement writes every table's rows, clears and adds revisions, and advances the session, each part gated on the session still carrying this owner at the sequence the read observed. A single statement is atomic, so no transaction brackets a commit, and a takeover by another process writes nothing and is reported as before.
+- Postgres storage serves repeated reads from memory. It is the Session's only writer, so a record it has read stays valid until it writes that record: conversation records, tasks and submissions by id and request, each conversation's newest entries and head markers, current document scans and addresses, and the materialized current content of documents are cached and kept current by every commit. Measured on one faux-model turn, admitting an input drops from 22 queries to 3, creating a conversation from 37 to 11, a plain turn from 61 to 9, and a turn with one tool call from 189 to 31.
+- `PostgresStorage.open` takes `migrate: false` to skip the schema check for deployments that migrate in an explicit step; an open is then two statements.
+
+
 ### Breaking Changes
 
 - The portable SQLite facade in `@earendil-works/pi-durable/storage/sqlite` is asynchronous: `SqliteDatabase` extends the new `SqliteExecutor` (`exec`, `run`, `get`, `all` by SQL text), `prepare` and `SqliteStatement` are removed, `transaction` takes an async callback that receives a transaction handle, and `close()` returns a promise. Custom adapters must be rewritten ([#10232](https://github.com/earendil-works/pi/pull/10232) by [@christianklotz](https://github.com/christianklotz)).
@@ -16,6 +23,7 @@
 
 ### Added
 
+- `PostgresStorage` in `@earendil-works/pi-durable/storage/postgres`: Sessions in a Postgres schema over a caller-owned pool such as `pg.Pool`, with a per-Session advisory lock and an owner token checked by every commit. The package adds no runtime dependency.
 - `CodingTools` extension in `@earendil-works/pi-durable/tools` with `read`, `write`, `edit`, and `bash`.
 - `AgentDoc`, `configure()`, `DEFAULT_RETRY_POLICY`, and `DEFAULT_COMPACTION_POLICY` exports.
 - `HarnessOptions.conversationCreated(tx, conversation)` runs in every commit that creates or forks a conversation, after the built-in documents, so applications can create their own documents in every conversation.

packages/durable/README.md

@@ -523,8 +523,9 @@ const usage = await harness.usage(context); // { models: { "openai/gpt-6-sol": U
 | Memory | `MemoryStorage` from the package root | Nothing is persisted. |
 | SQLite | `openNodeSqliteStorage(file)` from `@earendil-works/pi-durable/storage/sqlite/node` | One database file. WAL mode with `synchronous = NORMAL`: commits survive process crashes; the newest may be lost on power or host failure. |
 | JSONL | `openNodeJsonlStorage(directory, context)` from `@earendil-works/pi-durable/storage/jsonl/node` | Append-only files in one directory. Pass `{ fsync: true }` to flush before each commit marker. |
+| Postgres | `PostgresStorage.open(pool, { session })` from `@earendil-works/pi-durable/storage/postgres` | One Session per `session` name; many Sessions share one schema. Locked to one process. Built for a remote database: a commit is two round trips and repeated reads are served from memory. Pass `migrate: false` when the schema is migrated in an explicit step, so an open costs two statements. |
 
-One process owns a storage at a time; there is no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`.
+One process owns a storage at a time; SQLite and JSONL do no cross-process locking. The portable SQLite and JSONL cores (`/storage/sqlite`, `/storage/jsonl`) run without Node APIs, for example on Bun or in Cloudflare Durable Objects, given an asynchronous `SqliteDatabase` facade or a `FileSystem` from `@earendil-works/pi-durable/env`.
 
 SQLite adapters implement promise-based `exec`, `run`, `get`, `all`, `transaction`, and `close`. `run`, `get`, and `all` take SQL text plus positional bindings; adapters may cache prepared statements by SQL text. A transaction callback receives a transaction handle; all work in the transaction must use it, and the handle expires when the callback settles. Adapters must queue unrelated operations and other transactions until the transaction finishes, so calling `database` itself inside the callback never settles:
 
@@ -535,6 +536,16 @@ await database.transaction(async (transaction) => {
 });
 ```
 
+Postgres storage runs on a pool you create and own; a `pg.Pool` fits as is, and this package imports no driver. The first open creates the `pi_durable` schema (or `schema`) and its tables. Each storage keeps one pooled client checked out to hold its Session's advisory lock, so give the pool at least two connections. A second open of a held Session rejects. Every commit also checks the owner token written by the latest open, so a process that lost its lock can no longer write. Requires Postgres 11 or newer.
+
+```typescript
+import pg from "pg";
+import { PostgresStorage } from "@earendil-works/pi-durable/storage/postgres";
+
+const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
+const harness = await Harness.open(await PostgresStorage.open(pool, { session: "support-bot" }), { models, registry }, context);
+```
+
 Custom backends can run the shared conformance suite with any Vitest- or Jest-compatible runner:
 
 ```typescript
@@ -590,6 +601,8 @@ Examples that call OpenAI need `OPENAI_API_KEY`; most use the faux provider othe
 - [`docs/pico-v5-handoff.md`](docs/pico-v5-handoff.md): the implementation plan
 - [`docs/pico-v5-chord-usage.md`](docs/pico-v5-chord-usage.md): how the package uses Chord
 
+Postgres storage tests run when `PI_DURABLE_POSTGRES_URL` names a database they may create and drop schemas in; otherwise they are skipped.
+
 Benchmarks: `npm run bench:storage`, `npm run bench:storage:memory`, and `npm run bench:tool-output`.
 
 ## License

packages/durable/package.json

@@ -51,6 +51,11 @@
 			"types": "./dist/storage/sqlite/node.d.ts",
 			"import": "./dist/storage/sqlite/node.js"
 		},
+		"./storage/postgres": {
+			"source": "./src/storage/postgres/index.ts",
+			"types": "./dist/storage/postgres/index.d.ts",
+			"import": "./dist/storage/postgres/index.js"
+		},
 		"./testing": {
 			"source": "./src/testing/index.ts",
 			"types": "./dist/testing/index.d.ts",
@@ -96,6 +101,8 @@
 		"typebox": "1.3.27"
 	},
 	"devDependencies": {
+		"@types/pg": "8.23.1",
+		"pg": "8.23.0",
 		"shx": "0.4.0",
 		"vitest": "4.1.11"
 	}

packages/durable/src/storage/postgres/database.ts

@@ -0,0 +1,69 @@
+/** Values the Postgres storage binds as query parameters. */
+export type PostgresValue = null | boolean | number | string;
+
+export interface PostgresQueryResult<R> {
+	readonly rows: R[];
+	readonly rowCount: number | null;
+}
+
+/** Anything that runs one parameterized statement, such as a pool or a checked-out client. */
+export interface PostgresQueryable {
+	query<R extends object>(text: string, values?: PostgresValue[]): Promise<PostgresQueryResult<R>>;
+}
+
+/** A connection checked out of the pool. `release(error)` discards it instead of returning it. */
+export interface PostgresPoolClient extends PostgresQueryable {
+	release(error?: Error | boolean): void;
+	on(event: "error", listener: (error: Error) => void): unknown;
+	off(event: "error", listener: (error: Error) => void): unknown;
+}
+
+/**
+ * The connection pool `PostgresStorage` runs on. A `pg.Pool` satisfies it structurally, so this package
+ * never imports a driver. The pool belongs to the caller: closing the storage does not end it.
+ *
+ * The storage keeps one client checked out for its whole lifetime to hold the session lock, so the pool
+ * needs at least two connections.
+ */
+export interface PostgresPool extends PostgresQueryable {
+	connect(): Promise<PostgresPoolClient>;
+}
+
+/**
+ * Run `callback` in a transaction on its own pooled client. When BEGIN, COMMIT, or ROLLBACK fails, the client is
+ * discarded so the pool never reuses a connection in an unknown transaction state.
+ */
+export async function inTransaction<T>(
+	pool: PostgresPool,
+	begin: string,
+	callback: (client: PostgresQueryable) => Promise<T>,
+): Promise<T> {
+	const client = await pool.connect();
+	let discard: Error | undefined;
+	const control = async (sql: string) => {
+		try {
+			await client.query(sql);
+		} catch (error) {
+			discard = error as Error;
+			throw error;
+		}
+	};
+	try {
+		await control(begin);
+		let result: T;
+		try {
+			result = await callback(client);
+		} catch (error) {
+			try {
+				await control("ROLLBACK");
+			} catch (rollbackError) {
+				throw new AggregateError([error, rollbackError], "Durable Postgres transaction failed to roll back");
+			}
+			throw error;
+		}
+		await control("COMMIT");
+		return result;
+	} finally {
+		client.release(discard);
+	}
+}

packages/durable/src/storage/postgres/index.ts

@@ -0,0 +1,15 @@
+export type {
+	PostgresPool,
+	PostgresPoolClient,
+	PostgresQueryable,
+	PostgresQueryResult,
+	PostgresValue,
+} from "./database.ts";
+export {
+	applyPostgresMigrations,
+	CURRENT_POSTGRES_SCHEMA_VERSION,
+	DEFAULT_POSTGRES_SCHEMA,
+	POSTGRES_MIGRATIONS,
+	type PostgresMigration,
+} from "./migrations.ts";
+export { PostgresStorage, type PostgresStorageOptions } from "./storage.ts";