Agent Memory, State, and Durable Execution with Postgres

CREATE TABLE agent_runs (
  id          text PRIMARY KEY,
  goal        text NOT NULL,
  status      text NOT NULL DEFAULT 'queued',  -- queued | running | done | failed
  attempts    int  NOT NULL DEFAULT 0,
  lease_until timestamptz,                     -- a worker owns the run until this time
  result      jsonb,
  created_at  timestamptz NOT NULL DEFAULT now()
);
 
CREATE TABLE agent_steps (
  run_id     text NOT NULL REFERENCES agent_runs(id),
  step_key   text NOT NULL,
  result     jsonb NOT NULL,
  created_at timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (run_id, step_key)
);

import pg from "pg";
 
export const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });

import { pool } from "./db.ts";
 
export interface Run {
  id: string;
  goal: string;
  attempts: number;
}
 
const LEASE_SECONDS = 120;
const MAX_ATTEMPTS = 5;
 
/** Claim one runnable run: queued, or running with an expired lease (its worker died). */
export async function claimRun(): Promise<Run | null> {
  const { rows } = await pool.query<Run>(
    `UPDATE agent_runs
        SET status = 'running', attempts = attempts + 1, lease_until = now() + make_interval(secs => $1)
      WHERE id = (
        SELECT id FROM agent_runs
         WHERE status IN ('queued', 'running') AND (lease_until IS NULL OR lease_until < now()) AND attempts < $2
         ORDER BY created_at
         LIMIT 1
         FOR UPDATE SKIP LOCKED
      )
      RETURNING id, goal, attempts`,
    [LEASE_SECONDS, MAX_ATTEMPTS],
  );
  return rows[0] ?? null;
}
 
/**
 * Run `fn` once per (run, key). If the step already completed in an earlier attempt,
 * return its saved result without running it again.
 */
export async function step<T>(runId: string, key: string, fn: () => Promise<T>): Promise<T> {
  const saved = await pool.query<{ result: T }>("SELECT result FROM agent_steps WHERE run_id = $1 AND step_key = $2", [runId, key]);
  if (saved.rows[0]) return saved.rows[0].result;
 
  const result = await fn();
  await pool.query(
    "INSERT INTO agent_steps (run_id, step_key, result) VALUES ($1, $2, $3) ON CONFLICT DO NOTHING",
    [runId, key, JSON.stringify(result)],
  );
  // Long runs keep their lease alive as they make progress.
  await pool.query("UPDATE agent_runs SET lease_until = now() + make_interval(secs => $2) WHERE id = $1", [runId, LEASE_SECONDS]);
  return result;
}
 
export async function finishRun(runId: string, status: "done" | "failed", result: unknown) {
  await pool.query("UPDATE agent_runs SET status = $2, result = $3, lease_until = NULL WHERE id = $1", [runId, status, JSON.stringify(result)]);
}

import { pool } from "./db.ts";
 
// Your email provider's API would take the idempotency key directly.
// Here a unique constraint plays that role: a second send with the same key is a no-op.
export async function sendEmail(idempotencyKey: string, to: string, body: string): Promise<{ sent: boolean }> {
  const { rowCount } = await pool.query(
    "INSERT INTO sent_emails (idempotency_key, recipient, body) VALUES ($1, $2, $3) ON CONFLICT (idempotency_key) DO NOTHING",
    [idempotencyKey, to, body],
  );
  if (rowCount === 0) return { sent: false }; // already sent in an earlier attempt
  console.log(`[email] to ${to}: ${body.slice(0, 60)}…`);
  return { sent: true };
}

import Anthropic from "@anthropic-ai/sdk";
import { claimRun, finishRun, step } from "./lib/durable.ts";
import { sendEmail } from "./lib/email.ts";
import { pool } from "./lib/db.ts";
 
const client = new Anthropic();
const text = (m: Anthropic.Message) => m.content.flatMap((b) => (b.type === "text" ? [b.text] : [])).join("");
 
const run = await claimRun();
if (!run) {
  console.log("no runnable runs");
} else {
  console.log(`run ${run.id}, attempt ${run.attempts}`);
  try {
    const research = await step(run.id, "research", async () => {
      console.log("  running step: research");
      return text(await client.messages.create({
        model: "claude-opus-5-5",
        max_tokens: 4096,
        messages: [{ role: "user", content: `List the key facts needed to answer: ${run.goal}` }],
      }));
    });
 
    const draft = await step(run.id, "draft", async () => {
      console.log("  running step: draft");
      return text(await client.messages.create({
        model: "claude-opus-5-5",
        max_tokens: 4096,
        messages: [{ role: "user", content: `Write a short email answering "${run.goal}" using these facts:\n${research}` }],
      }));
    });
 
    if (process.env.CRASH_AFTER_DRAFT) process.exit(1); // simulate a crash for this demo
 
    const delivery = await step(run.id, "send-email", async () => {
      console.log("  running step: send-email");
      return sendEmail(`${run.id}:send-email`, "customer@example.com", draft);
    });
 
    await finishRun(run.id, "done", { delivery });
    console.log("  done", delivery);
  } catch (error) {
    // Leave the run claimable: when the lease expires, another attempt resumes from the last step.
    console.error("  attempt failed:", error instanceof Error ? error.message : error);
  }
}
await pool.end();

$ CRASH_AFTER_DRAFT=1 node worker.ts
run run_1, attempt 1
  running step: research
  running step: draft
$ node worker.ts           # after the lease expires
run run_1, attempt 2
  running step: send-email
[email] to customer@example.com: …
  done { sent: true }

import { pool } from "./db.ts";
 
// CREATE TABLE memories (id bigserial PRIMARY KEY, user_id text NOT NULL, fact text NOT NULL,
//   created_at timestamptz NOT NULL DEFAULT now(),
//   tsv tsvector GENERATED ALWAYS AS (to_tsvector('english', fact)) STORED);
 
export async function remember(userId: string, fact: string): Promise<string> {
  if (fact.length > 300) throw new Error("Memories must be a single short fact (max 300 characters)");
  await pool.query("INSERT INTO memories (user_id, fact) VALUES ($1, $2)", [userId, fact]);
  return "Saved.";
}
 
export async function recall(userId: string, query: string, limit = 5): Promise<string[]> {
  const { rows } = await pool.query<{ fact: string }>(
    `SELECT fact FROM memories
      WHERE user_id = $1 AND tsv @@ websearch_to_tsquery('english', $2)
      ORDER BY ts_rank(tsv, websearch_to_tsquery('english', $2)) DESC, created_at DESC
      LIMIT $3`,
    [userId, query, limit],
  );
  return rows.map((r) => r.fact);
}
 
export async function forgetAll(userId: string): Promise<void> {
  await pool.query("DELETE FROM memories WHERE user_id = $1", [userId]);
}