Skip to content

Builds on 1. Chat app. Store the transcript in Postgres, restart the host, and recover the same thread. Completed replies survive unchanged. An interrupted echo restarts from its first word under the same message ID.

1. Postgres

  • A database: createdb chat, or without a local server docker run -d -p 5432:5432 -e POSTGRES_HOST_AUTH_METHOD=trust -e POSTGRES_DB=chat postgres.
  • uv add "pinned[postgres]" for PostgresStateStore.
  • DATABASE_URL, for example postgresql://postgres@localhost/chat.

2. The host

Edit the chat tutorial's app.py. Replace lifespan, append_message, and runs with the versions below, add ChatLog, log, and persist, and replace the module-level store, worker, and app setup. Keep the five command handlers and the other helper methods, including transcript(). Drop the MemoryStateStore import:

import os
from contextlib import asynccontextmanager

from pinned import Tray, persistent, tray
from pinned.durability.postgres import PostgresStateStore


class ChatLog(Tray):
    messages = persistent(None)

    def items(self):
        yield from ()


class ChatHost(Statewire):
    @tray(ChatLog)
    async def log(self, item, ctx):
        raise RuntimeError("the chat log admits no work items")

    async def lifespan(self):
        self.chat_log = await self.log.get()
        self.state = {
            "messages": plain(self.chat_log.messages)
            or {"mainBranchId": MAIN, "branches": {MAIN: {"headId": MAIN}}, "messages": {}},
            "threads": {MAIN: {"status": "idle"}},
        }
        self.thread = linear_thread(
            messages=self.transcript,
            role=lambda m: m["role"],
        )
        yield

    def persist(self) -> None:
        self.chat_log.messages.value = plain(self.state["messages"])

    def append_message(self, message: dict) -> str:
        branch = self.state["messages"]["branches"][MAIN]
        entry = {**message, "parentId": branch["headId"]}
        self.state["messages"]["messages"][entry["id"]] = entry
        branch["headId"] = entry["id"]
        self.persist()
        return entry["id"]

    @run_manager(capabilities=("files",))
    async def runs(self, ctx: RunManager.RunContext):
        if not ctx.resume:
            for message in ctx.messages:
                self.append_message(message)
            ctx.applied()
        messages = self.transcript()
        user = next(m for m in reversed(messages) if m["role"] == "user")
        self.state["threads"][MAIN]["status"] = "streaming"
        if messages[-1]["role"] == "assistant":
            reply_id = messages[-1]["id"]
        else:
            reply_id = self.append_message(
                {"id": f"a-{uuid.uuid4()}", "role": "assistant", "parts": [{"type": "text", "text": "", "state": "streaming"}]}
            )
        part = self.state["messages"]["messages"][reply_id]["parts"][0]
        part["text"] = ""
        part["state"] = "streaming"
        words = ["echo:", *(w for p in user["parts"] if p["type"] == "text" for w in p["text"].split())]
        for word in words:
            if ctx.stop_requested.is_set():
                break
            await asyncio.sleep(0.05)
            append(part, "text", word + " ")
        part["state"] = "done"
        self.state["threads"][MAIN]["status"] = "idle"
        self.persist()
        return RunManager.Stop(dispatch_queue=True) if ctx.stop_requested.is_set() else RunManager.Complete()


store = PostgresStateStore(dsn=os.environ["DATABASE_URL"])
worker = PinnedWorker(hosts={"/threads": ChatHost}, state_store=store)


@asynccontextmanager
async def lifespan(app):
    await store.setup()
    yield


app = FastAPI(lifespan=lifespan)
app.include_router(worker)

How it works:

  • A Tray is a durable slot on the instance. ChatLog has one persistent field; its items yields nothing and its handler never runs, since the log admits no work items (trays, persistent).
  • lifespan restores the transcript. transcript() follows the stored parent links to recover conversation order.
  • persist() writes through .value after each message is added and when a run ends. The assistant message ID is saved before streaming starts; text streamed during a run is saved at its end.
  • runs now handles a redelivery. After a crash following ctx.applied(), the machine calls it again with ctx.resume set. The run reads the last user message from the saved transcript, reuses its assistant message ID, and streams the echo again (resumes and failures).
  • await store.setup() at boot creates the tables (state stores).

3. Start it

DATABASE_URL=postgresql://postgres@localhost/chat uv run uvicorn app:app --port 8000

4. Restart and resume

Create attach.mjs: tutorial 1's smoke.mjs with the await h.sendMessage(...) line removed. Send a message, stop the server with Ctrl+C, start it again, and attach. The last transcript line from each command is shown:

$ MESSAGE="remember me" node smoke.mjs # before the restart
user: remember me | assistant: echo: remember me
$ node attach.mjs                      # after the restart
user: remember me | assistant: echo: remember me

attach.mjs reads the saved conversation. Check that the completed reply is unchanged.

Crash during a reply

Keep the completed conversation above, then test another turn:

  1. Send a message long enough to interrupt its echo:

    MESSAGE="$(node -e 'console.log("recover ".repeat(200))')" node smoke.mjs

  2. Once several words have streamed, run kill -9 <uvicorn-pid> in another terminal. Use the PID in Uvicorn's Started server process [...] log line. Stop smoke.mjs with Ctrl+C.

  3. Start the host again with the same DATABASE_URL, then run node attach.mjs.

  4. Check that the earlier reply stays unchanged and the interrupted reply streams from its first word to completion. The thread still has two user messages and two assistant messages.

  5. Run MESSAGE="after recovery" node smoke.mjs. A new turn extends the recovered conversation.

What survives

  • The transcript: pinned_tray_index holds one row per tray: ns='/threads', id='t1', tray_ns='log', tray_id='~', marked=false for the chat log.
  • Each client's command pool, row ('/threads', 't1', 'client_commands', '<client id>', ...): admitted commands awaiting execution remain queued across a crash.
  • A run in flight: its tray redelivers the work on the next activation (requirements). ctx.resume is true when the batch was durably applied before the crash (Runs). A settled run's tray deletes itself.

The durability model is on durable objects. The hosting knobs are on fleet.

What you built

A thread that outlives its process: completed turns are restored, and an interrupted echo restarts under its saved message ID.

  • Next: 3. Fleet puts pinboard in front, so threads spread across many machines.