Add transactional conversations

README.md

@@ -9,8 +9,11 @@ events into ordered assistant deltas.
 
 The persistent Pi runtime resolves one explicitly configured model, disables
 tools and ambient resources, stores its JSONL under a private session directory,
-and never substitutes a fallback model silently. Conversation persistence,
-transport adapters, product prompts, and product behavior remain separate review
+and never substitutes a fallback model silently. The conversation layer adds
+private identity, visible-history continuity, one active turn, and rollback when
+a turn does not reach its visible-history commit.
+
+Transport adapters, product prompts, and product behavior remain separate review
 boundaries.
 
 ## Run one session
@@ -35,6 +38,19 @@ The task streams assistant text to standard output and keeps the Pi session at
 session. The task does not discover tools, extensions, skills, prompt templates,
 themes, or ambient context.
 
+## Conversation transactions
+
+`ConversationRegistry` resolves user and chat identity to one `Conversation`.
+The conversation checks the caller's visible history, reserves one turn, streams
+through its `SessionAgent`, and commits by saving the next visible history.
+Model failure, active abort or cancellation, and history-save failure restore the
+prior Pi branch before the registry permits a new session for that identity.
+
+`FileConversationHistoryStore` atomically replaces mode-0600 history files in a
+mode-0700 directory. It stores only visible user and assistant role/content
+pairs. This library boundary does not add transport, deployment, or product
+policy.
+
 ## Local checks
 
 Install the declared tools and run the stable local check surface:

src/conversation/conversation-registry.ts

@@ -0,0 +1,79 @@
+import { createHash } from "node:crypto";
+import { SessionAgent } from "../session/session-agent.ts";
+import {
+  Conversation,
+  type ConversationHistoryStore,
+  type ConversationMessage,
+  type ConversationTurn,
+} from "./conversation.ts";
+
+export type ConversationIdentity = Readonly<{
+  conversationId: string;
+  userId: string;
+  chatId: string;
+}>;
+
+export type ConversationAgentFactory = (
+  identity: ConversationIdentity,
+) => Promise<SessionAgent>;
+
+export class ConversationRegistry {
+  readonly #createAgent: ConversationAgentFactory;
+  readonly #historyStore: ConversationHistoryStore;
+  readonly #conversations = new Map<string, Promise<Conversation>>();
+
+  constructor(createAgent: ConversationAgentFactory, historyStore: ConversationHistoryStore) {
+    this.#createAgent = createAgent;
+    this.#historyStore = historyStore;
+  }
+
+  async start(
+    userId: string,
+    chatId: string,
+    messages: readonly ConversationMessage[],
+  ): Promise<ConversationTurn> {
+    const identity = conversationIdentity(userId, chatId);
+    const conversation = await this.#conversation(identity);
+    return conversation.start(messages);
+  }
+
+  #conversation(identity: ConversationIdentity): Promise<Conversation> {
+    const existing = this.#conversations.get(identity.conversationId);
+    if (existing) return existing;
+
+    let created: Promise<Conversation>;
+    created = this.#historyStore.load(identity.conversationId).then((history) =>
+      new Conversation({
+        conversationId: identity.conversationId,
+        history,
+        historyStore: this.#historyStore,
+        createAgent: () => this.#createAgent(identity),
+        evict: () => {
+          if (this.#conversations.get(identity.conversationId) === created) {
+            this.#conversations.delete(identity.conversationId);
+          }
+        },
+      })
+    );
+    this.#conversations.set(identity.conversationId, created);
+    created.catch(() => {
+      if (this.#conversations.get(identity.conversationId) === created) {
+        this.#conversations.delete(identity.conversationId);
+      }
+    });
+    return created;
+  }
+}
+
+function conversationIdentity(userId: string, chatId: string): ConversationIdentity {
+  requireIdentityPart("userId", userId);
+  requireIdentityPart("chatId", chatId);
+  const conversationId = createHash("sha256")
+    .update(JSON.stringify([userId, chatId]))
+    .digest("hex");
+  return { conversationId, userId, chatId };
+}
+
+function requireIdentityPart(name: string, value: string): void {
+  if (!value.trim()) throw new Error(`${name} must not be empty`);
+}

src/conversation/conversation.ts

@@ -0,0 +1,208 @@
+import { SessionAgent, type SessionCheckpoint } from "../session/session-agent.ts";
+import type { SessionAttachment } from "../session/session-turn.ts";
+
+export type ConversationMessage = Readonly<{
+  role: "user" | "assistant";
+  content: string;
+  attachments: readonly SessionAttachment[];
+}>;
+
+export type VisibleConversationMessage = Readonly<{
+  role: "user" | "assistant";
+  content: string;
+}>;
+
+export interface ConversationHistoryStore {
+  load(conversationId: string): Promise<readonly VisibleConversationMessage[]>;
+  save(
+    conversationId: string,
+    history: readonly VisibleConversationMessage[],
+  ): Promise<void>;
+}
+
+export type ConversationTurn = Readonly<{
+  conversationId: string;
+  deltas: AsyncIterable<string>;
+  abort(): Promise<void>;
+}>;
+
+export class ConversationConflictError extends Error {
+  override readonly name = "ConversationConflictError";
+}
+
+type ConversationOptions = Readonly<{
+  conversationId: string;
+  history: readonly VisibleConversationMessage[];
+  historyStore: ConversationHistoryStore;
+  createAgent(): Promise<SessionAgent>;
+  evict(): void;
+}>;
+
+export class Conversation {
+  readonly conversationId: string;
+  readonly #historyStore: ConversationHistoryStore;
+  readonly #createAgent: () => Promise<SessionAgent>;
+  readonly #evict: () => void;
+  #history: VisibleConversationMessage[];
+  #agent: SessionAgent | undefined;
+  #active = false;
+
+  constructor(options: ConversationOptions) {
+    this.conversationId = options.conversationId;
+    this.#history = options.history.map(({ role, content }) => ({ role, content }));
+    this.#historyStore = options.historyStore;
+    this.#createAgent = options.createAgent;
+    this.#evict = options.evict;
+  }
+
+  async start(messages: readonly ConversationMessage[]): Promise<ConversationTurn> {
+    if (this.#active) {
+      throw new ConversationConflictError("Conversation already has an active turn");
+    }
+
+    const lastMessage = messages.at(-1);
+    if (!lastMessage || lastMessage.role !== "user") {
+      throw new Error("Last conversation message must be a user message");
+    }
+    if (!sameVisibleHistory(messages.slice(0, -1), this.#history)) {
+      throw new ConversationConflictError(
+        "Visible conversation path does not match the persistent session",
+      );
+    }
+    const userMessage = snapshotUserMessage(lastMessage);
+
+    this.#active = true;
+    try {
+      this.#agent ??= await this.#createAgent();
+      if (this.#agent.conversationId !== this.conversationId) {
+        throw new Error("Conversation agent identity does not match the conversation");
+      }
+      return new ActiveConversationTurn(this, this.#agent, userMessage, this.#agent.checkpoint());
+    } catch (error) {
+      try {
+        this.#agent?.dispose();
+      } finally {
+        this.#agent = undefined;
+        this.#active = false;
+        this.#evict();
+      }
+      throw error;
+    }
+  }
+
+  async commit(userMessage: ConversationMessage, assistantContent: string): Promise<void> {
+    const history: VisibleConversationMessage[] = [
+      ...this.#history,
+      { role: "user", content: userMessage.content },
+      { role: "assistant", content: assistantContent },
+    ];
+    await this.#historyStore.save(this.conversationId, history);
+    this.#history = history;
+    this.#active = false;
+  }
+
+  async rollback(agent: SessionAgent, checkpoint: SessionCheckpoint): Promise<void> {
+    try {
+      await agent.rollback(checkpoint);
+    } finally {
+      try {
+        agent.dispose();
+      } finally {
+        this.#agent = undefined;
+        this.#evict();
+      }
+    }
+  }
+}
+
+class ActiveConversationTurn implements ConversationTurn {
+  readonly conversationId: string;
+  readonly deltas: AsyncIterable<string>;
+  readonly #conversation: Conversation;
+  readonly #agent: SessionAgent;
+  readonly #userMessage: ConversationMessage;
+  readonly #checkpoint: SessionCheckpoint;
+  #commitStarted = false;
+  #finished = false;
+  #rollbackPromise: Promise<void> | undefined;
+
+  constructor(
+    conversation: Conversation,
+    agent: SessionAgent,
+    userMessage: ConversationMessage,
+    checkpoint: SessionCheckpoint,
+  ) {
+    this.conversationId = conversation.conversationId;
+    this.#conversation = conversation;
+    this.#agent = agent;
+    this.#userMessage = userMessage;
+    this.#checkpoint = checkpoint;
+    this.deltas = this.#respond();
+  }
+
+  abort(): Promise<void> {
+    if (this.#rollbackPromise) return this.#rollbackPromise;
+    if (this.#commitStarted) return Promise.resolve();
+    return this.#rollback();
+  }
+
+  async *#respond(): AsyncIterable<string> {
+    if (this.#finished) return;
+    let output = "";
+    try {
+      for await (const delta of this.#agent.respond({
+        conversationId: this.conversationId,
+        userText: this.#userMessage.content,
+        attachments: this.#userMessage.attachments,
+      })) {
+        output += delta;
+        yield delta;
+      }
+
+      if (this.#finished) return;
+      this.#commitStarted = true;
+      await this.#conversation.commit(this.#userMessage, output);
+      this.#finished = true;
+    } catch (error) {
+      try {
+        await this.#rollback();
+      } catch (rollbackError) {
+        throw new AggregateError(
+          [error, rollbackError],
+          "Conversation turn failed and Pi session rollback also failed",
+        );
+      }
+      throw error;
+    } finally {
+      if (!this.#finished && !this.#commitStarted) await this.#rollback();
+    }
+  }
+
+  #rollback(): Promise<void> {
+    if (this.#rollbackPromise) return this.#rollbackPromise;
+    if (this.#finished) return Promise.resolve();
+
+    this.#finished = true;
+    this.#rollbackPromise = this.#conversation.rollback(this.#agent, this.#checkpoint);
+    return this.#rollbackPromise;
+  }
+}
+
+function snapshotUserMessage(message: ConversationMessage): ConversationMessage {
+  return {
+    role: "user",
+    content: message.content,
+    attachments: message.attachments.map(({ name, text }) => ({ name, text })),
+  };
+}
+
+function sameVisibleHistory(
+  incoming: readonly ConversationMessage[],
+  stored: readonly VisibleConversationMessage[],
+): boolean {
+  if (incoming.length !== stored.length) return false;
+  return incoming.every((message, index) => {
+    const expected = stored[index];
+    return expected?.role === message.role && expected.content === message.content;
+  });
+}