Pi Durable

Vandaag hebben Earendil en de Pi-community Pi 1.0 uitgebracht. Dit weerspiegelt onze overtuiging dat Pi, na talloze uren van hardening, onderhoud en actieve ontwikkeling, nu een solide fundament is waarop gebouwd kan worden. Pi blijft echter evolueren. Samen met Pi 1.0 brengen we een nieuw experimenteel pakket uit genaamd Pi Durable. Pi Durable is specifiek gebouwd voor langlopende, bestendige (durable) en kneedbare (malleable) agents die overal kunnen draaien. We nodigen jullie uit om mee te doen en ons te helpen er de beste durable harness van dit moment van te maken.

Waarom Pi Durable?

Pi, de coding agent, is gebouwd om op jouw (remote) machine te draaien, binnen een terminal, aangestuurd door één persoon. Als het proces stopt, bekijk je wat er is gebeurd en instrueer je de agent om verder te gaan. Dat is waar Pi 1.0 zich op richt en waarin het uitblinkt, en dat zal niet veranderen.

Bij Earendil willen we deze technologie toegankelijk maken voor iedereen, in de vorm die het beste bij hun behoeften past. Hiervoor hebben we een harness nodig die overal kan draaien, bereikbaar is via verschillende interfaces, oneindig lange gesprekken ondersteunt, catastrofale interne en externe storingen overleeft en toestaat dat meerdere mensen dezelfde agents aansturen.

Pi Durable is die harness. Het vervangt de Pi coding agent niet, maar is een framework voor het bouwen van elke agentische applicatie, inclusief coding agents. Het deelt niet alleen code met de Pi coding agent (zoals pi-ai), maar ook de principes: minimalisme en kneedbaarheid.

Het stelt ons daarnaast in staat om ontwerpen in deze ruimte te verkennen zonder de Pi coding agent te verstoren. De lessen die we leren bij het bouwen van agentische applicaties op Pi Durable zullen terugvloeien naar de Pi coding agent zodra ze hun waarde hebben bewezen.

Wat is een harness?

Iedereen heeft zijn eigen definitie van een harness. We hebben hier eerder over geschreven, maar laten we het concept opnieuw introduceren voor Pi Durable.

Een harness is opslag plus de mechanismen die nodig zijn om één of meer gesprekken met grote taalmodellen (LLM's) parallel uit te voeren. Het biedt de tools die deze modellen aanroepen en de uitvoeringsomgevingen waarin die tools draaien.

• Een gesprek is een interactie tussen jou en een agent, vastgelegd als een transcript.
• De agent is het grote taalmodel samen met de instellingen (zoals het denk-niveau) en de tools die het kan aanroepen.
• Tools voeren hun werk uit via een uitvoeringsomgeving, wat jouw laptop, een remote VM of een in-memory sandbox kan zijn. Welke tools en welke uitvoeringsomgeving een agent krijgt, hangt af van het specifieke gesprek.
• Een taak (task) is alles wat de harness uitvoert, van het aanroepen van het model tot het uitvoeren van een tool.

Net als alles in Pi is Pi Durable zo gebouwd dat jouw agent het kan begrijpen. De volledige broncode, zonder tests, bestaat uit ongeveer 15.000 regels, wat neerkomt op ongeveer 150.000 tokens met GPT en ongeveer 250.000 met Claude. Dat is het worst-case scenario. Om op Pi Durable te bouwen heeft je agent zelden alles nodig; de storage backends alleen al beslaan 3.000 regels die meestal overgeslagen kunnen worden.

Hieronder volgt een tour door Pi Durable om te illustreren wat we hebben gebouwd en waarom.

Langlopende processen, overal

We willen dat agents lange tijd kunnen draaien en overal kunnen functioneren, waarbij 'overal' op dit moment betekent: overal waar een JavaScript-runtime beschikbaar is.

In Pi Durable opent een harness zich over een storage backend. Pi Durable wordt geleverd met memory-, SQLite- en JSONL-opslag, plus een conformiteitssuite en benchmarks voor je eigen backend. De SQLite- en JSONL-opslagcode maakt geen gebruik van Node API's, waardoor het met een kleine adapter kan draaien op Bun of binnen een Cloudflare Durable Object. De storage-interface is klein en eenvoudig te implementeren bovenop wat je al hebt, zoals een key-value store of Postgres. Eén proces bezit op een gegeven moment de opslag, en andere clients koppelen zich aan dat proces.

Bij SQLite houdt de harness alleen de actieve set (working set) in het geheugen: de actieve transcripten, live taken en openstaande inzendingen. Alles komt pas van de schijf wanneer het nodig is. Actieve transcripten zijn van nature begrensd door het contextvenster van het model, omdat compaction oudere berichten samenvat voordat ze het venster doen overlopen. Zo past zelfs een gesprek met tienduizenden berichten gemakkelijk in het geheugen.

Tools die bestanden of een shell nodig hebben, krijgen deze van een uitvoeringsomgeving. Pi Durable levert een Node-uitvoeringsomgeving, die tools toegang geeft tot je lokale bestanden. Net als de opslag is de interface voor de uitvoeringsomgeving klein en eenvoudig te implementeren, zodat je ook remote uitvoeringsomgevingen aan je tools kunt ontsluiten. Hierdoor kan de harness op de ene machine draaien, terwijl de tools op een andere machine draaien. Je env-functie bouwt de omgeving voor elke tool-aanroep op basis van de werkmap van het gesprek, zodat elk gesprek op een andere plek kan draaien.

import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { createModels } from "@earendil-works/pi-ai/models";
import { openaiProvider } from "@earendil-works/pi-ai/providers/openai";
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";
import {
  openNodeSqliteStorage,
} from "@earendil-works/pi-durable/storage/sqlite/node";
import { CodingTools } from "@earendil-works/pi-durable/tools";

const context = BACKGROUND_CONTEXT; // elke call neemt een context mee voor annulering
const models = createModels();
models.setProvider(openaiProvider());

const registry = createRegistry();
registry.install(CodingTools); // read, write, edit, bash

const env = ({ cwd }: { cwd?: string }) =>
  new NodeExecutionEnv({ cwd: cwd ?? process.cwd() });

const harness = await Harness.open(
  await openNodeSqliteStorage("./agent.sqlite"),
  { models, registry, env },
  context,
);

// Het hoofdgesprek: aangemaakt bij eerste gebruik, en hetzelfde na elke herstart.
const root = await harness.root(context, {
  agent: {
    model: { provider: "openai", modelId: "gpt-6.1-sol" },
    cwd: "/work/repo",
  },
});

Overleeft crashes

We willen dat een agent overleeft wanneer het proces stopt—of de laptop nu in slaapstand gaat, de container opnieuw wordt uitgerold of de machine door een tekort aan geheugen crasht—en dat hij precies verder gaat waar hij was gebleven.

In Pi Durable is elke stap van een run een taak die een checkpoint opslaat voordat hij verdergaat. Als het proces sterft, opent een nieuw proces dezelfde opslag, vindt de onvoltooide taken en vervolgt elke taak vanaf het laatste checkpoint. Een modelverzoek dat werd afgebroken, wordt opnieuw verzonden; het gedeeltelijke antwoord blijft in het transcript staan, gemarkeerd als geaborteerd. Een tool-aanroep die werd afgebroken, wordt opnieuw uitgevoerd als dat veilig is; anders wordt het model geïnformeerd dat de aanroep werd onderbroken.

Pi Durable heeft geen ingebouwde subagents, maar deze zijn met een paar regels code te bouwen, zoals de onderstaande triage tool laat zien. Een subagent draait in een eigen gesprek, dus hij vervolgt zijn werk ook vanaf waar hij was gebleven. Een tool voor een subagent die veilig is om opnieuw uit te voeren, vindt zijn subagent opnieuw terug en wacht op het antwoord. In de wachtrij geplaatste berichten blijven in de wachtrij staan. Een requestId zorgt ervoor dat een inzending exact één keer gebeurt (exactly-once), zodat een client die opnieuw probeert na een crash het originele resultaat terugkrijgt in plaats van een tweede aanvraag te doen.

const job = {
  type: "input",
  content: "Fix the flaky login test",
  requestId: "job-42",
} as const;

await root.submit(job, context);

// Het proces sterft hier, midden in een tool-aanroep.
// Een nieuw proces opent dezelfde opslag.
const harness = await Harness.open(
  await openNodeSqliteStorage("./agent.sqlite"),
  { models, registry, env },
  context,
);

harness.resume(); // vervolg de onderbroken run
const root = await harness.root(context);

// dezelfde inzending, nu beantwoord
const settled = await (await root.submit(job, context)).wait(context);

Meerdere gesprekken tegelijk

We willen dat één harness veel gesprekken tegelijkertijd kan voeren, zonder dat de één de ander blokkeert.

In Pi Durable voert één harness zoveel gesprekken uit als nodig is, gelijktijdig en met dezelfde garanties. Een gesprek begint vanaf nul of splitst (forks) een ander gesprek op elk gewenst punt in het transcript, waarbij de geschiedenis van de ouder tot dat punt wordt gezien zonder deze te kopiëren.

Stel je een Slack-kanaal voor waar je agent reageert op mentions van iedereen. Iemand opent vervolgens een thread. Het kanaal kan één gesprek zijn, en de thread een fork daarvan op het bericht waarop gereageerd wordt. Beiden draaien tegelijkertijd en blokkeren elkaar niet.

const channel = await harness.root(context);
const question = await channel.submit(
  { type: "input", content: "@agent why did the deploy fail?" },
  context,
);
const answered = await question.wait(context);

// Iemand reageert op het antwoord van de agent in een thread.
// Elk gesprek benoemt zijn eigenaar, wat bepaalt waar een abort op uitmondt.
// De thread heeft geen eigenaar.
const thread = await channel.fork(
  answered.answer!,
  { ownership: { kind: "ownerless" } },
  context,
);

// Beide gesprekken werken tegelijkertijd.
const inThread = await thread.submit(
  { type: "input", content: "@agent can we roll it back?" },
  context,
);
const inChannel = await channel.submit(
  { type: "input", content: "@agent who is on call today?" },
  context,
);
await Promise.all([inThread.wait(context), inChannel.wait(context)]);

Elk gesprek slaat ook zijn eigen agent op: het model, het denk-niveau, de geselecteerde extensies en welke tools daarvan actief zijn, extra instructies en de werkmap in de uitvoeringsomgeving. Een reviewer naast de hoofdagent kan bijvoorbeeld een goedkoper model gebruiken, read-only tools hebben en een eigen checkout gebruiken.

Extensies

We willen dat alles wat een agent kan doen inplugbaar is, en dat elk ingeplugd onderdeel deelneemt aan de bestendigheid (durability).

In Pi Durable is een extensie een benoemd pakket van systeem-promptsecties, tools, hooks en taken. De applicatie installeert extensies in een registry. Elk gesprek selecteert welke extensies en tools het gebruikt en slaat alleen hun namen op.

Systeem-promptsecties

De systeem-prompt wordt voor elk verzoek opnieuw opgebouwd uit de secties van de extensies van het gesprek. Een gewijzigde sectie wordt dus direct opgepikt bij het volgende verzoek. Pi Durable registreert wat er is gewijzigd in het transcript, op de positie waar de wijziging plaatsvond, zodat een herstart of een fork precies ziet wat het model zag. Bij modellen die wijzigingen in de systeem-prompt en tools midden in een gesprek ondersteunen, wordt alleen de wijziging verzonden, waardoor de prompt-cache geldig blijft.

import { defineExtension, section } from "@earendil-works/pi-durable";

const ProjectContext = defineExtension({
  name: "project-context",
  sections: [
    // Lees uit de uitvoeringsomgeving van het gesprek.
    // Bestanden kunnen op de achtergrond worden geladen en bewaakt;
    // elk verzoek rendert de nieuwste status.
    section("agents_md", (input) => agentsMd.latest(input.env)),
    section("skills", (input) => skills.latest(input.env)),
  ],
});

Tools

Elke tool-aanroep wordt uitgevoerd als een eigen bestendige taak, en de intentie wordt opgeslagen voordat deze wordt uitgevoerd. Na een crash wordt een tool alleen opnieuw uitgevoerd als deze aangeeft dat dit veilig is. Anders wordt het model geïnformeerd dat de aanroep is onderbroken, inclusief de tot nu toe opgeslagen output, waarna het model beslist wat te doen. Elk gesprek kan ook een eigen set tools krijgen; de Slack-thread van eerder kan bijvoorbeeld wel zoeken, maar niet deployen.

import { Type } from "@earendil-works/pi-ai";
import { defineTool } from "@earendil-works/pi-durable";

const searchIssues = defineTool({
  name: "search_issues",
  description: "Search the issue tracker",
  parameters: Type.Object({ query: Type.String() }),
  replay: "safe", // alleen lezen, dus een rerun na een crash is prima
  execute: async (args, api) => {
    // gestreamd naar elke client die meekijkt
    api.output(`searching for ${args.query}\n`);
    return {
      content: [{ type: "text", text: await tracker.search(args.query) }],
    };
  },
});

const deploy = defineTool({
  name: "deploy",
  description: "Deploy a version to production",
  parameters: Type.Object({ version: Type.String() }),
  // Geen replay: een deploy die door een crash is onderbroken
  // wordt gemeld aan het model en nooit herhaald.
  execute: async (args) => ({
    content: [{ type: "text", text: await ci.deploy(args.version) }],
  }),
});

registry.install(defineExtension({ name: "ops", tools: [searchIssues, deploy] }));

// De thread mag zoeken, maar niet deployen.
await thread.configure({ tools: { remove: [deploy] } }, context);

Een tool krijgt de harness API voor zijn aanroep: hij kan vermeldingen en documenten commiten, taken en gesprekken starten, en communiceren met andere gesprekken. Hierdoor is een subagent slechts een paar regels code. Een tool maakt een gesprek aan dat hij bezit, geeft het een kleiner model en eigen instructies, en wacht op het antwoord. De subagent is een gesprek zoals ieder ander, dus hij overleeft een crash, houdt zijn eigen kosten bij en een UI kan hem onder de aanroep tonen.

import type { AssistantMessage } from "@earendil-works/pi-ai";
import { AssistantEntry, configure } from "@earendil-works/pi-durable";

const triage = defineTool({
  name: "triage",
  description: "Label an incoming issue as bug, feature, or question",
  parameters: Type.Object({ issue: Type.String() }),
  // een rerun na een crash vindt dezelfde subagent en dezelfde inzending
  replay: "safe",
  execute: async (args, api, context) => {
    const child = await api.commit(async (tx) => {
      const existing = (
        await tx.scanConversations({ ownerTaskId: api.taskId }, 1)
      ).items[0];
      if (existing !== undefined) return existing.id;

      // Eigendom van deze aanroep, dus het aborteren van de aanroep aborteert de subagent.
      const created = await tx.createConversation({
        ownership: { kind: "task", taskId: api.taskId },
      });

      // Begint als een kopie van de agent in dit gesprek.
      // Maak er een klein model van zonder tools.
      await configure(tx, created.id, {
        model: { provider: "openai", modelId: "gpt-6-luna" },
        tools: [],
        instructions: "Answer with one word: bug, feature, or question.",
      });
      return created.id;
    }, context);

    // laat een UI de subagent onder de aanroep tonen
    await api.details({ conversationId: child }, context);
    const subagent = await api.conversation(child, context);
    const request = {
      type: "input",
      content: args.issue,
      requestId: `triage:${api.taskId}`,
    } as const;
    const settled = await (
      await subagent!.submit(request, context)
    ).wait(context);

    // Het antwoord is een vermelding in het transcript van de subagent.
    // Lees dit en neem de tekst over.
    const entry = await api.commit(
      (tx) => tx.entry(AssistantEntry, settled.answer!),
      context,
    );
    const message = entry?.model?.[0] as AssistantMessage;
    const text = message.content
      .flatMap((content) => (content.type === "text" ? [content.text] : []))
      .join("");
    return { content: [{ type: "text", text }] };
  },
});

Extensies kunnen ook tools van andere extensies wijzigen. Een tool met dezelfde naam in een latere extensie vervangt de eerdere tool—bijvoorbeeld een bash die binnen een Python virtualenv draait. Een wrap decoreert de tool die uiteindelijk wint, ongeacht waar de wrapper-extensie is geselecteerd.

import { wrapTool } from "@earendil-works/pi-durable";
import { createBashTool } from "@earendil-works/pi-durable/tools";

// Meet de tijd van elke bash-aanroep, ongeacht welke bash het gesprek uiteindelijk gebruikt.
const Timing = defineExtension({
  name: "timing",
  wraps: [
    wrapTool(createBashTool(), (bash) => ({
      ...bash,
      execute: async (args, api, context) => {
        const start = Date.now();
        try {
          return await bash.execute(args, api, context);
        } finally {
          metrics.record("bash", Date.now() - start);
        }
      },
    })),
  ],
});

Hooks

Hooks laten extensies stappen ondernemen in taken, inclusief de ingebouwde taken voor het genereren van een modelrespons, het aanroepen van een tool of het uitvoeren van compaction. Ze kunnen een verzoek herschrijven voordat het naar het model gaat, een tool-aanroep blokkeren of herschrijven, een resultaat vervangen, een run gaande houden of zelf een samenvatting schrijven. Een hook kan opnieuw draaien na een crash, dus een hook die een beslissing neemt, slaat deze op in een memo: een kleine waarde die bij de taak wordt bewaard, waarbij de eerste schrijfactie wint.

import { hook, ToolTask } from "@earendil-works/pi-durable";

const Approval = defineExtension({
  name: "approval",
  hooks: [
    hook(ToolTask, {
      beforeTool: async (call, api, context) => {
        if (call.name !== "deploy") return undefined;

        // Na een herstart vindt de hook het opgeslagen antwoord
        // in plaats van opnieuw te vragen.
        let approved = await api.memo<boolean>(
          "approval:deploy",
          context,
        );
        approved ??= await api.memo(
          "approval:deploy",
          await askInSlack(call),
          context,
        );
        return approved
          ? undefined
          : { block: "Nobody approved the deploy." };
      },
    }),
  ],
});

Verschillende extensies kunnen aan hetzelfde haken. Hun hooks draaien als een keten, in de volgorde waarin het gesprek de extensies selecteert, en elke hook definieert hoe zijn keten draait. beforeTool geeft herschreven argumenten door aan de volgende in de keten, en de eerste block stopt de keten. afterTool geeft het resultaat door. onYield stopt bij de eerste hook die de run gaande houdt. Observers zoals afterResponse draaien altijd allemaal. Een hook die een fout gooit wordt gemeld, en de keten gaat verder, behalve in beforeTool, waar een fout de aanroep blokkeert.

Taken (Tasks)

De harness voert gesprekken uit met ingebouwde taken: één voor elk modelverzoek, één voor elke tool-aanroep en één voor compaction. Extensies kunnen hun eigen taken meebrengen en krijgen dezelfde mechanismen: een checkpoint na elke stap, timers die herstarts overleven en het wachten op andere taken.

Een checkout die de rekening splitst over verschillende kaarten, belast elke kaart tegelijkertijd. Als één kaart wordt geweigerd, worden de andere betalingen geaborteerd en terugbetaald:

import { defineTask, type TaskId } from "@earendil-works/pi-durable";

const Payment = defineTask<{ card: string }, { phase: "charge" }, string>({
  name: "shop.payment",
  version: 1,
  initial: () => ({ phase: "charge" }),
  phases: {
    charge: async (task, runtime, context) => {
      // De key maakt de afschrijving idempotent: als een crash deze fase
      // opnieuw uitvoert, wordt de kaart slechts één keer belast.
      const charge = await bank.charge(
        task.input.card,
        `payment-${task.id}`,
      );
      await runtime.commit(
        () => ({
          status: "terminal",
          outcome: charge.ok
            ? { status: "completed", result: charge.receipt }
            : { status: "failed", error: { message: charge.error } },
        }),
        context,
      );
    },
  },
  // Een andere betaling is mislukt, of de checkout is geannuleerd: maak deze ongedaan.
  abort: async (task, runtime, context) => {
    await bank.refund(`payment-${task.id}`);
    await runtime.commit(
      () => ({ status: "terminal", outcome: { status: "aborted" } }),
      context,
    );
  },
});

type CheckoutState =
  | { phase: "pay" }
  | { phase: "decide"; payments: TaskId<string>[] };

const Checkout = defineTask<{ cards: string[] }, CheckoutState, string>({
  name: "shop.checkout",
  version: 1,
  initial: () => ({ phase: "pay" }),
  phases: {
    pay: async (task, runtime, context) => {
      await runtime.commit(async (tx) => {
        const payments: TaskId<string>[] = [];
        for (const card of task.input.cards) {
          payments.push(
            await tx.createTask(Payment, { card }, {
              ownership: { kind: "task", taskId: task.id },
            }),
          );
        }
        // Voer geen code uit totdat elke betaling is voltooid.
        // De eerste mislukte betaling aborteert de anderen.
        return {
          status: "waiting",
          checkpoint: { phase: "decide", payments },
          on: payments,
          policy: "failFast",
        };
      }, context);
    },
    decide: async (task, runtime, context) => {
      const outcomes = await runtime.outcomes(
        task.state.checkpoint.payments,
        context,
      );
      const paid = outcomes.every(
        (outcome) => outcome.status === "completed",
      );
      await runtime.commit(
        () => ({
          status: "terminal",
          outcome: paid
            ? { status: "completed", result: "Order placed." }
            : {
              status: "failed",
              error: { message: "A payment failed." },
            },
        }),
        context,
      );
    },
  },
  abort: (_task, runtime, context) =>
    runtime.commit(
      () => ({ status: "terminal", outcome: { status: "aborted" } }),
      context,
    ),
});

// De agent start een checkout met een tool.
const checkout = defineTool({
  name: "checkout",
  description: "Pay for the cart, split across several cards",
  parameters: Type.Object({ cards: Type.Array(Type.String()) }),
  execute: async (args, api, context) => {
    // Eigendom van deze aanroep: het aborteren van de aanroep
    // aborteert de checkout en betaalt de betalingen terug.
    const owner = {
      ownership: { kind: "task", taskId: api.taskId },
    } as const;
    const id = await api.createTask(
      Checkout,
      { cards: args.cards },
      owner,
      context,
    );
    const { outcome } = (await api.waitForTask(id, context)).state;
    const text =
      outcome.status === "completed" ? outcome.result : outcome.status;
    return { content: [{ type: "text", text }] };
  },
});

registry.install(defineExtension({
  name: "shop",
  tools: [checkout],
  tasks: [Payment, Checkout],
}));

Taken en gesprekken vormen samen één eigendomsboom (ownership tree). Het aborteren van een taak aborteert alles wat deze bezit, van onderaf. Zo ruimt elke taak eerst zijn eigen effecten op, en een taak is pas voltooid als het werk dat hij bezit is afgerond. Een subagent volgt hetzelfde patroon: een gesprek dat eigendom is van de tool-aanroep die het startte.

Taken zijn standaard foreground: ze maken deel uit van het huidige werk van het gesprek. Het gesprek is pas inactief (idle) wanneer ze klaar zijn. Het aborteren van het gesprek (bijvoorbeeld wanneer de gebruiker op Esc drukt) aborteert deze taken en alles wat zij bezitten. Een background task hoort bij het gesprek, maar niet bij het huidige werk. Het gesprek wordt inactief terwijl de taak draait, en een gewone abort laat de taak en alles wat deze bezit ongemoeid. Dit is handig voor een subagent die langer moet overleven dan de beurt die hem startte, of een herinnering die morgen afgaat. Het aborteren van de taak zelf, of het gesprek met { background: true }, stopt deze wel.

// Onderdeel van het huidige werk: Esc aborteert dit en het gesprek wacht erop.
await api.createTask(
  Checkout,
  input,
  { ownership: { kind: "task", taskId: api.taskId } },
  context,
);

// Zij-werk: het gesprek wordt inactief terwijl dit draait, en Esc laat het ongemoeid.
await api.createTask(
  Reminder,
  input,
  { ownership: { kind: "conversation" }, background: true },
  context,
);

Compactering (Compaction)

We willen dat lange gesprekken kunnen doorgaan zonder dat de agent telkens moet stoppen om samen te vatten.

In Pi Durable is compactering een taak zoals elke andere, en deze draait terwijl het gesprek doorgaat. Wanneer het contextvenster het limiet van het model nadert, maakt een achtergrondproces een samenvatting van de oudere berichten, en deze samenvatting wordt geplaatst bij de volgende beurtgrens. Het gesprek wacht alleen op een samenvatting wanneer het volgende verzoek anders niet zou passen. Als de provider een verzoek nog steeds weigert omdat het te lang is, voert de harness de compactering uit en probeert het eenmaal opnieuw. Je kunt ook op elk moment handmatig compacteren met je eigen instructies. De oudere berichten blijven altijd bewaard in de opslag.

const harness = await Harness.open(storage, {
  models,
  registry,
  settings: {
    compaction: {
      // voorbij contextWindow - reserveTokens, wacht het volgende verzoek op een samenvatting
      reserveTokens: 16384,
      // hiervoor begint een samenvatting op de achtergrond
      backgroundTokens: 32768,
    },
  },
}, context);

// Handmatig, ook terwijl de agent aan het werk is.
await root.compact("Keep the names of the failing tests", context);

reset() gaat verder: het start een nieuwe context, eventueel vanaf een overdrachtsnotitie (handoff note), en een tool kan hierom vragen door controle terug te geven: { handoff }. Omdat er niets wordt verwijderd, kan een tweede tool nog steeds alles zoeken vóór de overdracht. Dit is alles wat nodig is om een agent te bouwen die zichzelf overdraagt en later zaken opzoekt.

const handoff = defineTool({
  name: "handoff",
  description:
    "Start over from a handoff note. " +
    "Older messages stay searchable with search_history.",
  parameters: Type.Object({ note: Type.String() }),
  execute: async (args, api, context) => {
    // In de wachtrij geplaatst achter de handoff, dus start de volgende run in de nieuwe context.
    const self = await api.conversation(api.conversationId, context);
    await self!.submit(
      {
        type: "input",
        content: "Continue.",
        requestId: `handoff:${api.taskId}`,
      },
      context,
    );
    // Beëindigt deze run en start een nieuwe context vanaf de notitie, zoals reset(note).
    return {
      content: [{ type: "text", text: "Handing off." }],
      control: { handoff: args.note },
    };
  },
});

const searchHistory = defineTool({
  name: "search_history",
  description: "Search older messages, including those before a handoff",
  parameters: Type.Object({ text: Type.String() }),
  replay: "safe",
  execute: async (args, api, context) => {
    // Tools lezen records ook via een transactie. Een transactie die niets schrijft, slaat niets op.
    const page = await api.commit(
      (tx) => tx.scanEntries({ conversationId: api.conversationId }, 200),
      context,
    );
    const hits = page.items.filter((entry) =>
      JSON.stringify(entry.model ?? []).includes(args.text),
    );
    const text = hits.map((entry) => JSON.stringify(entry.model)).join("\n");
    return { content: [{ type: "text", text }] };
  },
});

Bestendige applicatiestatus (Durable application state)

We willen dat de status van de applicatie die op de agent is gebouwd, net zo bestendig is als het gesprek zelf.

In Pi Durable leeft de applicatiestatus—zoals een takenlijst, een plan, een ticket of de sandbox waarin een gesprek draait—in documenten. Documenten zijn getypeerde JSON-bestanden die naast het transcript worden opgeslagen en in dezelfde atomaire commits worden gewijzigd. Hierdoor is de status nooit in strijd met het transcript dat deze heeft geproduceerd. Elk document bepaalt waarmee een fork start: de waarde van de ouder op het punt van de fork, de huidige waarde, of een nieuwe waarde.

import { defineDoc } from "@earendil-works/pi-durable";

const Todos = defineDoc<{ items: string[] }>({
  kind: "app.todos",
  version: 1,
  scope: "conversation",
  history: "rewindable",
  fork: "asOf", // een fork start met de todos die de ouder had op het fork-punt
  initial: () => ({ items: [] }),
});

const Todo = defineExtension({
  name: "todo",
  tools: [
    defineTool({
      name: "todo",
      description: "Add an item to your todo list",
      parameters: Type.Object({ item: Type.String() }),
      execute: async (args, api, context) => {
        await api.commit(async (tx) => {
          const todos = await tx.doc(Todos, api.conversationId);
          todos.items.push(args.item);
        }, context);
        const text = `Added ${args.item}`;
        return { content: [{ type: "text", text }] };
      },
    }),
  ],
  // Het model ziet de lijst voor elk verzoek.
  sections: [
    section("todos", async (input, context) => {
      const todos = await input.read.snapshot(
        Todos,
        input.conversationId,
        context,
      );
      return todos?.items.join("\n") || undefined;
    }),
  ],
});

// Een UI schrijft zich in op de gecommitteerde waarde.
const todos = await harness.documentState(Todos, channel.id, context);
todos?.subscribe((value) => renderTodos(value?.items ?? []));

Kneedbaarheid (Malleability)

We willen de code van een draaiende agent kunnen wijzigen zonder deze te stoppen.

In Pi Durable kan de registry veranderen terwijl gesprekken draaien. Het installeren van een extensie onder een naam die al is geïnstalleerd, vervangt deze in één stap. Een tool-aanroep die al draait, wordt voltooid met de code waarmee hij is gestart; de volgende aanroep gebruikt de nieuwe code. Gesprekken slaan extensie- en toolnamen op, nooit de code zelf, waardoor ze na een herstart alles oppikken wat het nieuwe proces installeert.

// Het bestand van de extensie is gewijzigd op de schijf.
// dezelfde naam "ops": vervangt de geïnstalleerde extensie
registry.install(await loadExtension("./ops.ts"));

Multiplayer

We willen dat veel mensen en clients tegelijkertijd met dezelfde gesprekken kunnen werken: meekijken, later instappen en ze aansturen.

In Pi Durable is alles wat een UI nodig heeft gecommitteerde status, dus elk aantal clients kan zich koppelen aan elk gesprek in de harness. Een client krijgt eerst het huidige overzicht: het transcript, het antwoord dat wordt gestreamd, draaiende tools en hun output, berichten in de wachtrij, de agent en het verbruik. Daarna ontvangt de client alleen de wijzigingen. Een client die later instapt of opnieuw verbinding maakt, begint vanaf het huidige overzicht. Elke client kan een lopend gesprek aansturen of een vervolgvraag in de wachtrij plaatsen.

// Een tweede client voegt zich bij de thread terwijl de agent aan het werk is.
const view = await thread.viewState(context);
render(view.value);
view.subscribe((value) => render(value));

// En stuurt het gesprek aan. Het bericht voegt zich bij het lopende werk
// na de huidige tool-aanroepen.
await thread.submit(
  { type: "input", content: "Check the staging logs first", whenBusy: "steer" },
  context,
);

Voor een remote client levert thread.watch() de exacte operaties van elke commit, klein genoeg om over een socket te verzenden. Als je de voorkeur geeft aan de bekende events van de coding agent, zet watchEvents() de commits om in die events, ten koste van meer bytes over de lijn.

Probeer het uit

Je kunt Pi Durable vandaag nog proberen. Pi Durable is experimenteel en de API kan nog veranderen. Richt je agent op packages/durable in een Pi-checkout, laat hem de README lezen, de meer dan dertig voorbeelden bekijken, de kleine coding agent op Pi Durable testen of deze prachtige vakantieplanner-agent uitproberen, en begin met bouwen.

De vakantieplanner bestaat uit ongeveer 1.300 regels TypeScript, waarvan het grootste deel de TUI (Text User Interface) is. Als het op een coding agent lijkt, komt dat alleen omdat hij TUI-componenten leent van de Pi coding agent.

• Vakantieplanner met TUI, gebouwd op Pi Durable.
• Een subagent voert drie zoekopdrachten parallel uit, elk als een bestendige taak. Ondertussen is de hoofdagent vrij om te chatten.
• Het proces sterft. Weer en musea zijn voltooid, treinen niet.
• Herstart. search is veilig om opnieuw uit te voeren: alleen trains draait opnieuw.
• Schakel over naar de subagent en stuur deze aan.
• Terug in de hoofdagent: vraag, compacteer en stuur aan, terwijl hij doorwerkt.
• Het rapport arriveert als een bericht; de hoofdagent maakt er een plan van.

Om beide demo's uit een Pi-checkout te draaien:

npm install && npm run build
node packages/coding-agent/src/experimental/durable/main.ts
node packages/coding-agent/src/experimental/vacation/main.ts

Om op Pi Durable te bouwen in je eigen project:

npm install @earendil-works/pi-durable @earendil-works/pi-ai @earendil-works/chord

In de komende weken zullen we meer vertellen over Pi Durable en laten zien welke kleine agentische tools we ermee bouwen om ons te helpen werken, zoals een Slack-bot of een GitHub-triage-bot. We willen de details nog niet verklappen. Er komt meer aan naarmate we Pi Durable zelf gebruiken, net zoals we Pi gebruiken.

FAQ

Waarom opnieuw TypeScript? Omdat dit de makkelijkste manier is om dit op te starten. Maar zoals iedereen inmiddels weet, is het zeer eenvoudig om alles te porten naar Rust of assembler. We sluiten dit niet uit voor de toekomst, maar op dit moment focussen we ons op TypeScript.