char-slop/berry

git clone https://git.t4t.associates/char-slop/berry

Charlotte Somallow project reorderingb4fc6cf2026-10-29

import type { Context, JsonValue } from "@earendil-works/chord";
import { homedir } from "node:os";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { diffRevisions, type Op } from "@earendil-works/chord/delta";
import {
  type AssistantMessage,
  clampThinkingLevel,
  type Context as ModelContext,
  type ImageContent,
  lazyStream,
  type Models,
  type ModelsSimpleStreamOptions,
  type ModelThinkingLevel,
  type MutableModels,
  type Provider,
  type TextContent,
  type ToolCall,
  Type,
  validateToolArguments,
} from "@earendil-works/pi-ai";
import {
  type Agent,
  AgentDoc,
  type AgentState,
  type ConversationId,
  createRegistry,
  type Cursor,
  DEFAULT_COMPACTION_POLICY,
  defineDoc,
  defineExtension as defineDurableExtension,
  defineTool,
  type EntryId,
  type EntryRecord,
  type Extension as DurableExtension,
  GenerationTask,
  Harness,
  InboxDoc,
  LiveDoc,
  type ModelRef,
  ProviderDoc,
  section,
  type Storage,
  type SubmissionId,
  type ToolExecutionApi,
  type Tx,
} from "@earendil-works/pi-durable";
import { CODEMODE_SOURCE_GRAMMAR } from "@earendil-works/pi-codemode/source";
import {
  type Channel,
  type ChannelHandler,
  type Command,
  type CompactionCap,
  type Configuration,
  configurationEvents,
  type ContextHook,
  type ConversationApi,
  defineExtension,
  type Extension,
  type Handler,
  handlerArguments,
  type PayloadHook,
  type PromptProse,
  type PromptRender,
  type SectionRender,
  tightestCap,
  type Tool,
  type ToolContext,
} from "@berry/api";
import { imageResult } from "@berry/api/content.ts";
import { MAX_MESSAGE_BYTES, userContent } from "@berry/api/images.ts";
import { directory, type Environment, LOCAL } from "@berry/api/environment.ts";
import {
  applyStoreWrite,
  DESCRIPTION,
  type NestedTool,
  renderCatalog,
  runScript,
  type StoreWrite,
} from "./codemode.ts";
import { EventEntry, StartEntry } from "./entries.ts";
import { recordSettled, SettledDoc } from "./settled.ts";
import { startTool } from "./tool-timing.ts";
import {
  type Binding,
  type BindingEvent,
  configuredEnvironments,
  ENVIRONMENT,
  ENVIRONMENT_NAME,
  environmentExtension,
  type Environments,
  type EnvironmentSettings,
} from "./environment.ts";
import { projectFor, type ProjectInput, workspaceFor, workspacePath } from "@berry/api/projects.ts";
import { ProjectOrder, Projects, resolveProject, validateRoots } from "./projects.ts";
import { folders } from "./folders.ts";
import { references } from "./references.ts";
import { DEFAULT_SYSTEM_PROMPT, renderGuidelines, type SystemPromptSettings } from "./prompts.ts";
import type { Feed } from "./pubsub.ts";
import { SUBAGENT, type Subagent, subagentExtension } from "./subagent.ts";
import { tid } from "./tid.ts";
import { ShellDoc, shellTask } from "./shell.ts";
import { conversationRegistry } from "./registry.ts";

const context = BACKGROUND_CONTEXT;

export { EventEntry, StartEntry };

export type ConversationInfo = {
  /** The durable conversation it began as. */
  id: number;
  created: number;
  /** A user-chosen sidebar name, otherwise the first message is used. */
  title?: string;
  /** A fork's parent, and the entry of it the fork continues from. */
  parent?: { tid: string; at: number };
  owner?: string;
  /** The branches `/tree` has made of it since, each a durable conversation that continues from one of its entries. */
  branches?: { id: number; at: number; created: number }[];
  /** The branch it continues on, when that is not the one it began as. */
  head?: number;
  /** When it was put out of the way, out of the list of conversations. */
  archived?: number;
  /** The newest message acknowledged by a frontend, shared across devices. */
  readThrough?: number;
};

/** A stored choice of extensions, choosing by `name` the ones it chose by a name in `formerly`. */
function adopt(agent: AgentState, formerly: readonly string[], name: string) {
  const renamed = (names: readonly string[]) => [
    ...new Set(names.map((chosen) => (formerly.includes(chosen) ? name : chosen))),
  ];
  const stale = (names?: readonly string[]) => names?.some((chosen) => formerly.includes(chosen));
  const selection = agent.extensions;
  if (Array.isArray(selection)) {
    if (stale(selection)) agent.extensions = renamed(selection);
  } else if (selection !== undefined) {
    if (stale(selection.add)) selection.add = renamed(selection.add!);
    if (stale(selection.remove)) selection.remove = renamed(selection.remove!);
  }
}

/** The durable conversation a conversation continues on. */
export const headOf = (info: ConversationInfo) => (info.head ?? info.id) as ConversationId;

/** Every branch of a conversation, the one it continues on first. */
const branchesOf = (info: ConversationInfo) =>
  [
    ...new Set([headOf(info), info.id, ...(info.branches ?? []).map((branch) => branch.id)]),
  ] as ConversationId[];

/** A change to a conversation's agent; `null` clears a field. */
export type AgentChange = {
  readonly model?: ModelRef | null;
  readonly thinkingLevel?: ModelThinkingLevel | null;
  /** Exactly these extensions, in order. */
  readonly extensions?: readonly string[];
};

export const Conversations = defineDoc<Record<string, ConversationInfo>>({
  kind: "berry.conversations",
  version: 1,
  scope: "session",
  initial: () => ({}),
  checkpointWhen: () => true,
});

/** The harness's own settings, shared by every conversation. */
export type BerrySettings = {
  /** The models conversations may use, out of those the logged-in providers offer. */
  models: { provider: string; modelId: string }[];
  /** The machines conversations may work on besides this one, by name. */
  environments?: Record<string, EnvironmentSettings>;
  allowLocalEnvironment?: boolean;
  systemPrompt?: SystemPromptSettings;
};

export const Settings = defineDoc<BerrySettings>({
  kind: "berry.settings",
  version: 1,
  scope: "session",
  initial: () => ({
    models: [],
    allowLocalEnvironment: true,
    systemPrompt: { ...DEFAULT_SYSTEM_PROMPT },
  }),
  checkpointWhen: () => true,
});

// deno-lint-ignore no-explicit-any
type AnyExtension = Extension<any, any>;
// deno-lint-ignore no-explicit-any
type AnyTool = Tool<any, any>;
// deno-lint-ignore no-explicit-any
type AnyCommand = Command<any, any>;
// deno-lint-ignore no-explicit-any
type AnyHandler = Handler<any>;
// deno-lint-ignore no-explicit-any
type AnyChannelHandler = ChannelHandler<any>;

/** The calls running in one load of an extension, so a replaced load is disposed only once they end. */
type Calls = { count: number; drained: PromiseWithResolvers<void> | null };
const idle = (): Calls => ({ count: 0, drained: null });

async function running<T>(of: Loaded, work: () => Promise<T>): Promise<T> {
  of.calls.count++;
  try {
    return await work();
  } finally {
    if (--of.calls.count === 0) of.calls.drained?.resolve();
  }
}

type Loaded = {
  readonly extension: AnyExtension;
  readonly providers: readonly Provider[];
  readonly replacedProviders: ReadonlyMap<string, Provider>;
  readonly compactionCaps: readonly (() => CompactionCap | undefined)[];
  readonly tools: readonly AnyTool[];
  readonly handlers: readonly AnyHandler[];
  readonly channels: readonly AnyChannelHandler[];
  /** Channels open to this load, which close with it. */
  readonly open: Set<Channel>;
  readonly prompts: readonly { name: string; render: SectionRender<JsonValue> }[];
  readonly promptProse?: PromptProse;
  readonly configuration?: Configuration<JsonValue, JsonValue>;
  readonly conversation?: (conv: ConversationApi<JsonValue, JsonValue>) => void;
  readonly disposers: readonly (() => void | Promise<void>)[];
  readonly calls: Calls;
};

type Contribution = {
  readonly tools: AnyTool[];
  readonly prompts: PromptRender[];
  readonly payloadHooks: PayloadHook[];
  readonly contextHooks: ContextHook[];
  readonly commands: AnyCommand[];
};

/** The durable tool invocation every nested call runs inside of. */
type Frame = { readonly api: ToolExecutionApi; readonly context: Context };

type CallOptions = {
  readonly signal?: AbortSignal;
  readonly output?: (chunk: string) => void;
  readonly progress?: (details: JsonValue) => void;
};

export class BerryError extends Error {}

/** An extension's failure, for the user to read. */
const surfaced = (error: unknown, prefix = "") =>
  new BerryError(prefix + (error instanceof Error ? error.message : String(error)), {
    cause: error,
  });

export type CreateOptions = {
  /** Selected extensions, in order, with their configuration values. */
  readonly extensions: Readonly<Record<string, JsonValue>>;
  readonly model?: ModelRef;
  readonly thinkingLevel?: ModelThinkingLevel;
};

const NEVER = new AbortController().signal;

function toContent(value: JsonValue | undefined): (TextContent | ImageContent)[] {
  if (value === undefined) return [];
  if (typeof value === "string") return [{ type: "text", text: value }];
  const image = imageResult(value).value;
  if (image && !Array.isArray(value)) {
    return [{ type: "image", data: image.data, mimeType: image.mimeType }];
  }
  return [{ type: "text", text: JSON.stringify(value, null, 2) }];
}

const exposure = (tool: AnyTool) => tool.expose ?? "codemode";

const declaration = (of: Loaded, tool: AnyTool) => ({
  name: tool.name,
  extension: of.extension.name,
  description: tool.description,
  parameters: tool.parameters,
  ...(tool.returns === undefined ? {} : { returns: tool.returns }),
  direct: exposure(tool) === "both",
});

/** How an extension's last load went. */
export type ExtensionStatus = {
  readonly name: string;
  /** The source directory when known, or null for an extension without a unique source. */
  readonly directory?: string | null;
  readonly description: string | null;
  readonly status: "loaded" | "failed";
  /** Why the last load failed; the version before it, if any, still runs. */
  readonly error: string | null;
};

/** What the process running berry offers its self-modification tools. */
export type Host = {
  /** Where berry's source is. */
  readonly root: string;
  /** Where extensions on disk are discovered and watched. */
  readonly extensionDirectories?: readonly string[];
  /** What `deno check` says of the code on disk, or `null` when it passes. */
  check(): Promise<string | null>;
  /** Stop gracefully, and come back running the code on disk. */
  restart(): void;
  extensions(): readonly ExtensionStatus[];
  /** Load an extension's files again. */
  reload(name: string): Promise<ExtensionStatus>;
};

/** The harness's own extension: codemode, and with a host the /reload command. Its state is the store. */
export const Builtin = defineExtension<Record<string, JsonValue>, StoreWrite>({
  name: "berry",
  alwaysOn: true,
  initial: () => ({}),
  apply: applyStoreWrite,
  setup() {},
});

/** The user's `/reload`: every extension on disk, or the one named. */
function reloadCommand(host: Host): AnyCommand {
  return {
    name: "reload",
    description: "reload all extensions or one extension by name",
    async execute(args) {
      const names = args.trim() ? [args.trim()] : host.extensions().map((status) => status.name);
      const statuses = [];
      for (const name of names) statuses.push(await host.reload(name));
      const failed = statuses.filter((status) => status.status === "failed");
      if (failed.length > 0) {
        throw new BerryError(failed.map(({ name, error }) => `${name}: ${error}`).join("\n\n"));
      }
      return `reloaded ${names.join(", ")}`;
    },
  };
}

function guide(root: string, directories: readonly string[] = [`${root}/extensions`]): string {
  return [
    `the assistant's operating environment is berry. its source is at ${root}, a deno project.`,
    `- extensions live in ${directories.join(" and ")}, each under <name>/: ` +
    "mod.ts (`defineExtension` from @berry/api), an optional " +
    "web.tsx (renderers, from @berry/api/web.ts), and tests using @berry/api/testing.ts. " +
    "an extension imports only its own directory's files and bare specifiers.",
    "- saving an extension's files auto-reloads it once `deno check` passes.",
    "- events are kept forever: a reducer must keep folding every event shape it has ever recorded.",
    "- changes outside extension directories need the header's restart button, which checks the code first.",
  ].join("\n");
}

export async function openBerry(
  storage: Storage,
  models: MutableModels,
  host?: Host,
  given?: Environments,
) {
  const machines = given ??
    configuredEnvironments(async () =>
      (await harness.snapshot(Settings, context))?.environments ?? {}
    );
  const environments: Environments = {
    async open(name, cwd, during) {
      await requireEnvironmentAllowed(name);
      return await machines.open(name, cwd, during);
    },
    close: () => machines.close(),
  };
  const registry = createRegistry();
  const loaded = new Map<string, Loaded>();
  const folds = new Map<string, { loaded: Loaded; upTo: number; state: JsonValue }>();
  const tids = new Map<number, string>();
  const providerSessions = new Map<string, ConversationId>();
  /** Live state feeds; a reinstall changes state without a commit. */
  const stateFeeds = new Set<() => void>();

  const own: Loaded = {
    extension: Builtin,
    providers: [],
    replacedProviders: new Map(),
    compactionCaps: [],
    tools: [],
    handlers: [],
    channels: [],
    open: new Set(),
    prompts: [],
    disposers: [],
    calls: idle(),
    ...(host ? { conversation: (conv) => conv.command(reloadCommand(host)) } : {}),
  };
  loaded.set(Builtin.name, own);
  const shell = shellTask(async (command, runtime, during, progress) => {
    const id = runtime.conversationId;
    const of = (await selected(id)).find((of) => of.extension.name === "code");
    const env = await environmentOf(id, during);
    const name = env.platform === "win32" ? "powershell" : "bash";
    if (!of) throw new BerryError("Shell commands require the code extension");
    const tool = [...of.tools, ...(await contribution(id, of)).tools]
      .findLast((tool) => tool.name === name);
    if (!tool) throw new BerryError(`The code extension has no ${name} tool for this conversation`);
    let state = await stateOf(id, of);
    const value = await running(of, async () =>
      await tool.execute({ command }, {
        tid: await tidOf(id),
        get state() {
          return state;
        },
        env,
        signal: runtime.signal,
        emit: async (event: JsonValue) => {
          await runtime.commit(async (tx) => {
            await tx.appendEntry(id, { kind: EventEntry.kind, data: { ext: "code", event } });
          }, during);
          state = await stateOf(id, of);
        },
        output: () => {},
        progress,
      }));
    return { value, text: tool.text ? tool.text(value) : JSON.stringify(value) ?? "(no output)" };
  }, () => harness);
  const builtin = defineDurableExtension({
    name: Builtin.name,
    tasks: [shell],
    tools: [
      defineTool({
        name: "codemode",
        description: DESCRIPTION,
        parameters: Type.Object({ code: Type.String({ description: "Raw JavaScript source." }) }),
        // capable models write the script as raw text instead of a json-escaped string
        constrainedSampling: {
          type: "grammar",
          variants: { openai_lark: CODEMODE_SOURCE_GRAMMAR },
        },
        execute: async ({ code }, api, context) => {
          await startTool(api, context);
          const id = api.conversationId;
          const frame = { api, context };
          const result = await runScript(code, {
            tools: await nested(id, frame),
            store: await stateOf(id, own) as Record<string, JsonValue>,
            signal: context.abortSignal ?? NEVER,
            // a rejection only means the invocation ended first
            progress: (details) => void api.details(details, context).catch(() => {}),
          });
          if (result.storeWrites) await commitEvent(frame, id, Builtin.name, result.storeWrites);
          return { output: result.content, isError: result.isError, details: result.details };
        },
      }),
    ],
    sections: [
      section("lede", async () => {
        const { userName } = await systemPrompt();
        return "we are a multispecialist software dev & sysadmin team. you are assisting the user" +
          (userName ? ` (${userName})` : "");
      }, { tag: false }),
      section("guidelines", async () => {
        const { generalGuidelines, userName } = await systemPrompt();
        return renderGuidelines(generalGuidelines, userName);
      }, { tag: false }),
      ...(host
        ? [
          section("berry", ({ env }) =>
            (env as Environment | undefined)?.name === LOCAL
              ? guide(host.root, host.extensionDirectories)
              : undefined),
        ]
        : []),
      section(
        "codemode",
        async (input) => renderCatalog(await declarations(input.conversationId)),
        { tag: false },
      ),
    ],
  });
  registry.install(builtin);

  // model reads go through the tightest compaction cap, so generation compacts at one
  // threshold whatever the model's window; the stored providers keep their own windows
  const effectiveModels: Models = new Proxy(models, {
    get(target, prop) {
      if (prop === "streamSimple" || prop === "completeSimple") {
        return (...[model, request, options]: Parameters<Models[typeof prop]>) => {
          const source = lazyStream(model, async () => {
            const id = await providerConversation(options?.sessionId);
            return await target.streamSimple(
              model,
              await contextRequest(id, model, request, options),
              payloadOptions(id, await cacheSession(model, id, options)),
            );
          });
          // pi-ai 1.1.0's transient-error classifier misses raw EOF messages.
          const normalizeError = (message: AssistantMessage): AssistantMessage =>
            message.stopReason === "error" &&
              /\b(?:unexpected eof|error reading a body from connection)\b/i.test(
                message.errorMessage ?? "",
              )
              ? { ...message, errorMessage: `Network error: ${message.errorMessage}` }
              : message;
          const stream = lazyStream(model, () =>
            Promise.resolve({
              async *[Symbol.asyncIterator]() {
                for await (const event of source) {
                  yield event.type === "error"
                    ? { ...event, error: normalizeError(event.error) }
                    : event;
                }
              },
              result: async () => normalizeError(await source.result()),
            }));
          return prop === "completeSimple" ? stream.result() : stream;
        };
      }
      if (
        prop === "getModel" || prop === "getModelOfType" || prop === "getModels" ||
        prop === "getModelsOfType" || prop === "getAllModels" || prop === "getAvailable" ||
        prop === "getAvailableOfType" || prop === "getAllAvailable"
      ) {
        const read = Reflect.get(target, prop) as (...args: never[]) => unknown;
        return (...args: never[]) => {
          const found = read.apply(target, args);
          const cap = effectiveCap();
          if (cap === undefined) return found;
          if (found instanceof Promise) {
            return found.then((value) =>
              Array.isArray(value)
                ? value.map((model) => cappedModel(model, cap))
                : cappedModel(value, cap)
            );
          }
          return Array.isArray(found)
            ? found.map((model) => cappedModel(model, cap))
            : cappedModel(found, cap);
        };
      }
      const value = Reflect.get(target, prop);
      return typeof value === "function" ? value.bind(target) : value;
    },
  });

  const harness = await Harness.open(storage, {
    // the proxied catalog keeps generation honest about the cap without touching stored providers
    models: effectiveModels,
    registry: conversationRegistry(registry, resolveTools),
    // every conversation selects the always-on extensions; their stored choices only add to them
    settings: {
      retry: { enabled: true, maxRetries: 3, baseDelayMs: 2_000, maxAgentDelayMs: 60_000 },
      get compaction() {
        const cap = effectiveCap();
        return cap === undefined ? {} : { backgroundTokens: cap.backgroundTokens };
      },
      get extensions() {
        return registry.snapshot().installed().filter((ext) =>
          ext.name === SUBAGENT || loaded.get(ext.name)?.extension.alwaysOn
        );
      },
    },
    env: ({ conversationId }, during) => environmentOf(conversationId, during),
    onReport: (error) => console.error("[berry]", error),
    conversationCreated: async (tx, record) => {
      const index = await tx.doc(Conversations);
      const tidFor = (id: ConversationId) =>
        Object.entries(index).find(([, info]) => branchesOf(info).includes(id))?.[0];
      const parent = record.parent && tidFor(record.parent.conversationId);
      const owner = record.owner && tidFor(record.owner.conversationId);
      index[tid()] = {
        id: record.id,
        created: Date.now(),
        ...(parent === undefined ? {} : {
          parent: { tid: parent, at: record.parent!.at },
          readThrough: record.parent!.at,
        }),
        ...(owner === undefined ? {} : { owner }),
      };
      if (record.parent === undefined) {
        await tx.appendEntry(StartEntry, record.id, {});
        await tx.doc(SettledDoc, record.id);
      }
      // a fork starts with its parent's choice as of the fork point, which may name extensions by what they were
      if (record.parent !== undefined) {
        const agent = await tx.doc(AgentDoc, record.id);
        for (const { extension } of loaded.values()) {
          adopt(agent, extension.formerly ?? [], extension.name);
        }
      }
    },
  }, context);
  const stopSettled = recordSettled(harness);
  // watchDoc never creates a document; the conversations topic needs this one from the start
  await harness.commit(async (tx) => {
    await tx.doc(Conversations);
    const settings = await tx.doc(Settings);
    settings.systemPrompt = { ...DEFAULT_SYSTEM_PROMPT, ...settings.systemPrompt };
    const legacy = settings.systemPrompt as SystemPromptSettings & { codingGuidelines?: unknown };
    if (typeof legacy.codingGuidelines === "string" && legacy.promptProse?.code === undefined) {
      const text = legacy.codingGuidelines;
      legacy.promptProse = {
        ...(legacy.promptProse ?? {}),
        code: text.trim() ? `guidelines:\n${text}` : text,
      };
    }
    delete legacy.codingGuidelines;
    await tx.doc(Projects);
    await tx.doc(ProjectOrder);
  }, context);

  async function systemPrompt(): Promise<SystemPromptSettings> {
    return (await harness.snapshot(Settings, context))?.systemPrompt ?? DEFAULT_SYSTEM_PROMPT;
  }

  /** Editable prose for `name`, with `$USER` replaced and blank prose omitted. */
  async function resolveProse(
    name: string,
    registered: PromptProse | undefined,
  ): Promise<string | undefined> {
    if (registered === undefined) return undefined;
    const { userName, promptProse } = await systemPrompt();
    const raw = Object.hasOwn(promptProse ?? {}, name) ? promptProse![name] : registered.default;
    return raw.trim() ? raw.replaceAll("$USER", () => userName) : undefined;
  }

  /** The level `model` runs `requested` at: a conversation's level is always one its model supports. */
  function supported(model: ModelRef, requested: ModelThinkingLevel): ModelThinkingLevel {
    const known = models.getModel(model.provider, model.modelId);
    return known ? clampThinkingLevel(known, requested) : requested;
  }

  async function requireEnabled(model: ModelRef) {
    const settings = await harness.snapshot(Settings, context);
    const enabled = settings?.models.some((m) =>
      m.provider === model.provider && m.modelId === model.modelId
    );
    if (!enabled) {
      throw new BerryError(`${model.provider}/${model.modelId} is not enabled in settings`);
    }
  }

  async function idOf(tid: string): Promise<ConversationId | undefined> {
    const index = await harness.snapshot(Conversations, context);
    const info = index?.[tid];
    return info && headOf(info);
  }

  async function existing(tid: string): Promise<ConversationId> {
    const id = await idOf(tid);
    if (id === undefined) throw new BerryError(`No conversation ${tid}`);
    return id;
  }

  async function conversationOf(tid: string) {
    const id = await idOf(tid);
    return id === undefined ? undefined : await harness.conversation(id, context);
  }

  /** The name an extension goes by now, given any it went by. */
  function current(name: string): string {
    if (loaded.has(name)) return name;
    return [...loaded.values()].find((of) => of.extension.formerly?.includes(name))?.extension
      .name ??
      name;
  }

  function selection(names: readonly string[]) {
    const snapshot = registry.snapshot();
    const chosen = names.map(current).filter((name) => !loaded.get(name)?.extension.alwaysOn);
    const add = chosen.map((name) => {
      const installed = snapshot.extension(name);
      if (installed === undefined) throw new BerryError(`No extension named ${name}`);
      return installed;
    });
    return {
      add,
      // old conversations inherit delegation; explicit choices override that default
      ...(chosen.includes(SUBAGENT) ? {} : { remove: [subagent.extension] }),
    };
  }

  // Read from committed state only: a failed creating commit may hand its ID out again.
  async function tidOf(id: ConversationId): Promise<string> {
    if (!tids.has(id)) {
      const index = await harness.snapshot(Conversations, context);
      for (const [tid, info] of Object.entries(index ?? {})) {
        for (const id of branchesOf(info)) tids.set(id, tid);
      }
    }
    const found = tids.get(id);
    if (found === undefined) throw new BerryError(`Conversation ${id} has no TID`);
    return found;
  }

  async function conversation(id: ConversationId) {
    const found = await harness.conversation(id, context);
    if (found === undefined) throw new BerryError(`No conversation ${id}`);
    return found;
  }

  /** The branch of `tid` whose history holds its entry `at`, the one it continues on if that does. */
  async function locate(tid: string, at: number) {
    const info = (await harness.snapshot(Conversations, context))?.[tid];
    for (const id of info ? branchesOf(info) : []) {
      const branch = await conversation(id);
      const query = {
        order: "descending",
        minEntryId: at as EntryId,
        maxEntryId: at as EntryId,
      } as const;
      if ((await branch.entries(query, 1, undefined, context)).items.length > 0) return branch;
    }
    return undefined;
  }

  /**
   * Continue `tid` from its entry `at`: on a branch of it that ends there, or else on a new one, which keeps the agent
   * it has now. A busy conversation stays where it is.
   */
  async function branch(tid: string, at: number) {
    const info = (await harness.snapshot(Conversations, context))?.[tid];
    if (info === undefined) throw new BerryError(`No conversation ${tid}`);
    const head = headOf(info);
    const ending = await (async () => {
      for (const id of branchesOf(info)) {
        const query = { order: "descending" } as const;
        const [tip] = (await (await conversation(id)).entries(query, 1, undefined, context)).items;
        if (tip?.id === at) return id;
      }
    })();
    if (ending === head) return;
    if (
      (await harness.snapshot(LiveDoc, head, context))?.run ||
      (await harness.snapshot(ShellDoc, head, context))?.run
    ) {
      throw new BerryError(`${tid} is busy: let it finish, or abort it, first`);
    }
    if (ending !== undefined) {
      return await harness.commit(async (tx) => {
        const moved = (await tx.doc(Conversations))[tid];
        if (ending === moved.id) delete moved.head;
        else moved.head = ending;
      }, context);
    }
    const source = await locate(tid, at);
    if (source === undefined) throw new BerryError(`No entry ${at} in ${tid}`);
    const agent = await harness.snapshot(AgentDoc, head, context);
    await source.fork(at as EntryId, {
      ownership: { kind: "ownerless" },
      init: async (tx, id) => {
        // conversationCreated took the branch for a conversation of its own
        const index = await tx.doc(Conversations);
        for (const [other, { id: of }] of Object.entries(index)) {
          if (of === id) delete index[other];
        }
        const moved = index[tid];
        moved.branches = [...(moved.branches ?? []), { id, at, created: Date.now() }];
        moved.head = id;
        const copy = await tx.doc(AgentDoc, id);
        for (const key of Object.keys(copy)) delete copy[key as keyof typeof copy];
        Object.assign(copy, structuredClone(agent ?? {}));
      },
    }, context);
  }

  /**
   * A feed of `tid` that follows it onto whichever branch it continues on, opened by `open`: a move reads as the diff
   * between the branches' values.
   */
  async function follow(
    tid: string,
    open: (id: ConversationId) => Promise<Feed | undefined>,
  ): Promise<Feed | undefined> {
    let id = await idOf(tid);
    let inner = id === undefined ? undefined : await open(id);
    if (inner === undefined) return undefined;
    let value = inner.value;
    let listener: ((value: unknown, ops: readonly Op[]) => Promise<void>) | undefined;
    const forward = (feed: Feed) =>
      feed.start(async (next, ops) => {
        // a branch moved off may still deliver what it had pending
        if (feed !== inner) return;
        value = next;
        await listener?.(value, ops);
      });
    let moving = Promise.resolve();
    const move = () => {
      moving = moving.then(async () => {
        const next = await idOf(tid);
        if (next === undefined || next === id || listener === undefined) return;
        const feed = await open(next);
        if (feed === undefined) return;
        const previous = inner!;
        id = next;
        inner = feed;
        const ops = diffRevisions(value as JsonValue, feed.value as JsonValue);
        value = feed.value;
        // the move's ops go out before anything the new branch's feed delivers, which is never inline
        forward(feed);
        await listener(value, ops);
        await previous.stop();
      }).catch((error) => console.error(`[berry] following ${tid}:`, error));
    };
    const unsubscribe = harness.subscribeCommits((publication) => {
      const moved = publication.changes.some((change) =>
        change.type === "document" && change.record.kind === Conversations.definition.kind
      );
      if (moved) queueMicrotask(move);
    });
    return {
      get value() {
        return value;
      },
      start(next) {
        listener = next;
        forward(inner!);
      },
      async stop() {
        listener = undefined;
        unsubscribe();
        await moving;
        await inner!.stop();
      },
    };
  }

  /** Conversations that chose an extension by a name it went by choose it by its name now. */
  async function rename(extension: AnyExtension) {
    const formerly = extension.formerly ?? [];
    if (formerly.length === 0) return;
    const index = await harness.snapshot(Conversations, context) ?? {};
    const stale: ConversationId[] = [];
    for (const id of Object.values(index).flatMap(branchesOf)) {
      const agent = structuredClone(await harness.snapshot(AgentDoc, id, context) ?? {});
      const before = JSON.stringify(agent);
      adopt(agent, formerly, extension.name);
      if (JSON.stringify(agent) !== before) stale.push(id);
    }
    if (stale.length === 0) return;
    await harness.commit(async (tx) => {
      for (const id of stale) adopt(await tx.doc(AgentDoc, id), formerly, extension.name);
    }, context);
  }

  function fold(extension: AnyExtension, state: JsonValue, entry: EntryRecord): JsonValue {
    const own = (ext: string) => ext === extension.name || extension.formerly?.includes(ext);
    if (EventEntry.is(entry) && own(entry.data.ext) && extension.apply) {
      state = extension.apply(state, entry.data.event);
    }
    return extension.observe ? extension.observe(state, entry) : state;
  }

  async function stateOf(id: ConversationId, of: Loaded, tx?: Tx): Promise<JsonValue> {
    const key = `${id}:${of.extension.name}`;
    const cached = folds.get(key);
    let { upTo, state } = cached?.loaded === of
      ? cached
      : { upTo: 0, state: of.extension.initial() };
    const history = tx ? undefined : await conversation(id);
    const query = { order: "ascending", minEntryId: (upTo + 1) as EntryId } as const;
    let cursor: Cursor | undefined;
    do {
      const page = tx
        ? await tx.scanEntries({ conversationId: id, ...query }, 256, cursor)
        : await history!.entries(query, 256, cursor, context);
      for (const entry of page.items) {
        state = fold(of.extension, state, entry);
        upTo = entry.id;
      }
      cursor = page.next;
    } while (cursor !== undefined);
    const latest = folds.get(key);
    if (latest?.loaded !== of || latest.upTo < upTo) folds.set(key, { loaded: of, upTo, state });
    return state;
  }

  async function contribution(id: ConversationId, of: Loaded): Promise<Contribution> {
    const state = await stateOf(id, of);
    const env = of.conversation ? await environmentOf(id).catch(() => undefined) : undefined;

    const result: Contribution = {
      tools: [],
      prompts: [],
      commands: [],
      payloadHooks: [],
      contextHooks: [],
    };
    of.conversation?.({
      tid: await tidOf(id),
      state,
      env,
      tool: (tool) => {
        if (tool.name === SUBAGENT || (tool.name === "codemode" && exposure(tool) !== "codemode")) {
          throw new BerryError(
            `${of.extension.name}: tool ${tool.name} is reserved for the harness`,
          );
        }
        result.tools.push(tool as AnyTool);
      },
      prompt: (render) => void result.prompts.push(render),
      onPayload: (hook) => void result.payloadHooks.push(hook),
      onContext: (hook) => void result.contextHooks.push(hook),
      command: (command) => void result.commands.push(command),
    });
    return result;
  }

  async function providerConversation(sessionId?: string): Promise<ConversationId | undefined> {
    if (sessionId === undefined) return undefined;
    // pi-durable exposes the conversation to providers through its stable session identity.
    if (!providerSessions.has(sessionId)) {
      const index = await harness.snapshot(Conversations, context) ?? {};
      for (const id of new Set(Object.values(index).flatMap(branchesOf))) {
        const provider = await harness.snapshot(ProviderDoc, id, context);
        if (provider) providerSessions.set(provider.sessionId, id);
      }
    }
    return providerSessions.get(sessionId);
  }

  async function cacheSession(
    model: Parameters<ContextHook>[1],
    id: ConversationId | undefined,
    options?: ModelsSimpleStreamOptions,
  ): Promise<ModelsSimpleStreamOptions | undefined> {
    // codex pools a websocket per session and continues it with previous_response_id, so its session
    // stays per branch, and payloadOptions patches its cache key in the body instead
    if (id === undefined || model.api === "openai-codex-responses") return options;
    const tid = await tidOf(id).catch(() => undefined);
    return tid === undefined ? options : { ...options, sessionId: tid };
  }

  async function contextRequest(
    id: ConversationId | undefined,
    model: Parameters<ContextHook>[1],
    request: ModelContext,
    options?: ModelsSimpleStreamOptions,
  ): Promise<ModelContext> {
    let messages = request.messages;
    if (id !== undefined) {
      await requireEnvironmentAllowed((await bindingOf(id)).environment);
      const extensions = await selected(id);
      let copied = false;
      for (const of of extensions) {
        const { contextHooks } = await contribution(id, of);
        if (contextHooks.length === 0) continue;
        if (!copied) {
          messages = structuredClone(messages);
          copied = true;
        }
        await running(of, async () => {
          for (const hook of contextHooks) {
            const next = await hook(messages, model);
            if (next !== undefined) messages = [...next];
          }
        });
      }
      options?.signal?.throwIfAborted();
    }
    // anthropic refuses a failed tool result that is not all text, and so every later request with it
    // in the history; one with images goes as a success instead, its text still saying it failed
    if (model.api === "anthropic-messages") {
      messages = messages.map((message) =>
        message.role === "toolResult" && message.isError &&
          message.content.some((block) => block.type === "image")
          ? { ...message, isError: false }
          : message
      );
    }
    return messages === request.messages ? request : { ...request, messages };
  }

  function payloadOptions(
    id: ConversationId | undefined,
    options?: ModelsSimpleStreamOptions,
  ): ModelsSimpleStreamOptions {
    return {
      ...options,
      onPayload: async (payload, model) => {
        const original = await options?.onPayload?.(payload, model);
        if (original !== undefined) payload = original;
        if (id === undefined) return payload;
        for (const of of await selected(id)) {
          const { payloadHooks } = await contribution(id, of);
          await running(of, async () => {
            for (const hook of payloadHooks) {
              const next = await hook(payload, model);
              if (next !== undefined) payload = next;
            }
          });
        }
        // a provider whose session cannot be the conversation's (codex, see cacheSession) keeps the
        // conversation's key for its prompt cache here, so a rewind's new branch does not lose it
        const tid = await tidOf(id).catch(() => undefined);
        if (
          tid !== undefined && typeof payload === "object" && payload !== null &&
          !Array.isArray(payload) &&
          typeof (payload as Record<string, unknown>).prompt_cache_key === "string"
        ) {
          payload = { ...payload, prompt_cache_key: tid };
        }
        return payload;
      },
    };
  }

  async function selected(id: ConversationId): Promise<Loaded[]> {
    const agent = await (await conversation(id)).agent(context);
    return agent.extensions.flatMap(({ name }) => loaded.get(name) ?? []);
  }

  async function resolveTools(agent: Agent, id: ConversationId): Promise<Agent> {
    const tools = new Map<string, Agent["tools"][number]>();
    const sources = new Map<string, AnyTool>();
    for (const extension of agent.extensions) {
      const own = new Map((extension.tools ?? []).map((tool) => [tool.name, tool]));
      const of = loaded.get(extension.name);
      const available = new Map(of?.tools.map((tool) => [tool.name, tool]));
      if (of) {
        for (const tool of (await contribution(id, of)).tools) {
          available.set(tool.name, tool);
          own.delete(tool.name);
          if (exposure(tool) !== "codemode") own.set(tool.name, compileTool(of, tool));
        }
      }
      for (const [name, tool] of own) {
        tools.set(name, tool);
        sources.delete(name);
        const source = available.get(name);
        if (source) sources.set(name, source);
      }
    }
    if ([...sources.values()].some((tool) => tool.enabled)) {
      const env = await environmentOf(id).catch(() => undefined);
      for (const [name, tool] of sources) {
        if (tool.enabled && !await tool.enabled({ env })) tools.delete(name);
      }
    }
    const filter = (await harness.snapshot(AgentDoc, id, context))?.tools;
    const offered = filter === undefined
      ? [...tools.values()]
      : Array.isArray(filter)
      ? [...new Set(filter)].flatMap((name) => tools.get(name) ?? [])
      : [...tools.values()].filter((tool) => !filter.remove.includes(tool.name));
    return { ...agent, tools: offered };
  }

  /** Codemode-callable tools of the selected extensions; a later extension's tool replaces an earlier one. */
  async function callable(id: ConversationId) {
    const byName = new Map<string, { loaded: Loaded; tool: AnyTool }>();
    for (const of of await selected(id)) {
      const own = [
        ...of.tools,
        ...(await contribution(id, of)).tools,
      ];
      for (const tool of own) byName.set(tool.name, { loaded: of, tool });
    }
    for (const [name, { tool }] of byName) {
      if (exposure(tool) === "model") byName.delete(name);
    }
    if ([...byName.values()].some(({ tool }) => tool.enabled)) {
      const env = await environmentOf(id).catch(() => undefined);
      for (const [name, { tool }] of byName) {
        if (tool.enabled && !await tool.enabled({ env })) byName.delete(name);
      }
    }
    return byName;
  }

  /** Commands of the selected extensions; a later extension's command replaces an earlier one. */
  async function commandsOf(id: ConversationId) {
    const byName = new Map<string, { loaded: Loaded; command: AnyCommand }>();
    for (const of of await selected(id)) {
      for (const command of (await contribution(id, of)).commands) {
        byName.set(command.name, { loaded: of, command });
      }
    }
    if ([...byName.values()].some(({ command }) => command.enabled)) {
      const env = await environmentOf(id).catch(() => undefined);
      for (const [name, { command }] of byName) {
        if (command.enabled && !await command.enabled({ env })) byName.delete(name);
      }
    }
    return byName;
  }

  async function writeEvent(id: ConversationId, ext: string, event: JsonValue, content?: string) {
    const target = await conversation(id);
    return await target.submit({
      type: "write",
      entry: {
        kind: EventEntry.kind,
        data: { ext, event },
        ...(content === undefined ? {} : {
          model: [{ role: "user", content, timestamp: Date.now() }],
        }),
      },
    }, context);
  }

  async function delegationOffered(id: ConversationId) {
    const agent = await (await conversation(id)).agent(context);
    return agent.extensions.some((ext) => ext.name === SUBAGENT);
  }

  async function declarations(id: ConversationId) {
    const tools = [...(await callable(id)).values()].map(({ loaded: of, tool }) =>
      declaration(of, tool)
    );
    if (await delegationOffered(id)) {
      tools.push(...subagent.scripted.map(({ declaration }) => declaration));
    }
    return tools;
  }

  /** What a script may call, each tool in its own extension's context, with arguments validated like the model's. */
  async function nested(id: ConversationId, frame: Frame): Promise<NestedTool[]> {
    const tools: NestedTool[] = [...(await callable(id)).values()].map(({ loaded: of, tool }) => ({
      ...declaration(of, tool),
      call: (args, options) => {
        const call = {
          type: "toolCall",
          id: frame.api.callId,
          name: tool.name,
          arguments: args,
        } as const;
        // deno-lint-ignore no-explicit-any
        return invoke(of, tool, id, validateToolArguments(tool, call as any), frame, options);
      },
    }));
    if (await delegationOffered(id)) {
      for (const { declaration, call } of subagent.scripted) {
        tools.push({
          ...declaration,
          call: (args, options) => {
            const toolCall = {
              type: "toolCall",
              id: frame.api.callId,
              name: declaration.name,
              arguments: args as ToolCall["arguments"],
            } as const;
            const validated = validateToolArguments(declaration, toolCall);
            return call(validated, frame.api, frame.context, options);
          },
        });
      }
    }
    return tools;
  }

  async function commitEvent(frame: Frame, id: ConversationId, ext: string, event: JsonValue) {
    await frame.api.commit(
      (tx) => tx.appendEntry(id, { kind: EventEntry.kind, data: { ext, event } }),
      frame.context,
    );
  }

  async function invoke(
    of: Loaded,
    tool: AnyTool,
    id: ConversationId,
    args: unknown,
    frame: Frame,
    options: CallOptions,
  ): Promise<JsonValue | undefined> {
    let state = await stateOf(id, of);
    // the harness's env option builds one for every call
    const env = frame.api.env as Environment | undefined;
    if (env === undefined) throw new BerryError("the call has no environment");
    if (tool.enabled && !await tool.enabled({ env })) {
      throw new BerryError(`Tool ${tool.name} is not enabled in this environment`);
    }
    const ctx: ToolContext<JsonValue, JsonValue> = {
      tid: await tidOf(id),
      get state() {
        return state;
      },
      env,
      signal: options.signal ?? frame.context.abortSignal ?? NEVER,
      emit: async (event) => {
        await commitEvent(frame, id, of.extension.name, event);
        state = await stateOf(id, of);
      },
      output: (chunk) => options.output?.(chunk),
      progress: (details) => options.progress?.(details),
    };
    return await running(of, async () => await tool.execute(args, ctx));
  }

  /** Once the load has no call running; a replaced load takes no new ones. */
  function drained(of: Loaded): Promise<void> {
    if (of.calls.count === 0) return Promise.resolve();
    of.calls.drained ??= Promise.withResolvers();
    return of.calls.drained.promise;
  }

  function compileTool(of: Loaded, tool: AnyTool) {
    return defineTool({
      name: tool.name,
      description: tool.description,
      parameters: tool.parameters,
      execute: async (args, api, context) => {
        await startTool(api, context);
        const value = await invoke(of, tool, api.conversationId, args, { api, context }, {
          output: (chunk) => api.output(chunk),
          // a rejection only means the invocation ended first
          progress: (details) => void api.details(details, context).catch(() => {}),
        });
        return {
          output: tool.text ? [{ type: "text", text: tool.text(value) }] : toContent(value),
        };
      },
    });
  }

  function compile(of: Loaded) {
    const { name } = of.extension;
    return defineDurableExtension({
      name,
      tools: of.tools.filter((tool) => exposure(tool) !== "codemode").map((tool) =>
        compileTool(of, tool)
      ),
      sections: [
        section(name, async (input) => {
          const prose = await resolveProse(name, of.promptProse);
          const { prompts } = await contribution(input.conversationId, of);
          const env = input.env as Environment | undefined;
          const texts = await Promise.all(prompts.map((render) => render({ env })));
          const all = [prose, ...texts].filter((t) => t !== undefined);
          return all.length === 0 ? undefined : all.join("\n\n");
        }),
        ...of.prompts.map((prompt) =>
          section(`${name}_${prompt.name}`, async (input) =>
            await prompt.render({
              tid: await tidOf(input.conversationId),
              state: await stateOf(input.conversationId, of),
              env: input.env as Environment | undefined,
            }))
        ),
      ],
    });
  }

  /** The tightest compaction cap any load reports; a failing getter reads as no cap. */
  function effectiveCap(): CompactionCap | undefined {
    const seen: (CompactionCap | undefined)[] = [];
    for (const of of loaded.values()) {
      for (const get of of.compactionCaps) {
        try {
          seen.push(get());
        } catch (error) {
          console.error(`[berry] compaction cap of ${of.extension.name}:`, error);
        }
      }
    }
    return tightestCap(seen);
  }

  /** A catalog model with its effective context window under a cap, smaller windows untouched. */
  function cappedModel(model: unknown, cap: CompactionCap): unknown {
    if (typeof model !== "object" || model === null) return model;
    const window = (model as { contextWindow?: unknown }).contextWindow;
    const limit = cap.maxContextTokens + DEFAULT_COMPACTION_POLICY.reserveTokens;
    if (typeof window !== "number" || !(window > limit)) return model;
    return { ...(model as Record<string, unknown>), contextWindow: limit };
  }

  function releaseProviders(of: Loaded) {
    for (const provider of of.providers) {
      if (models.getProvider(provider.id) !== provider) continue;
      const original = of.replacedProviders.get(provider.id);
      if (original) models.setProvider(original);
      else models.deleteProvider(provider.id);
    }
  }

  async function dispose(of: Loaded) {
    for (const channel of of.open) channel.close();
    for (const disposer of of.disposers) {
      try {
        await disposer();
      } catch (error) {
        console.error(`[berry] disposing ${of.extension.name}:`, error);
      }
    }
  }

  /** An extension's state in one branch as a feed: a snapshot, then diffs whenever history or the extension changes. */
  async function stateFeed(
    id: ConversationId,
    name: string,
    configuration = false,
  ): Promise<Feed | undefined> {
    const initial = loaded.get(name);
    if (initial === undefined || (configuration && !initial.configuration)) return undefined;
    const read = async (of: Loaded) => {
      const state = await stateOf(id, of);
      return configuration ? of.configuration?.values(state) ?? null : state;
    };
    let value = await read(initial);
    let listener: ((value: JsonValue, ops: readonly Op[]) => Promise<void>) | undefined;
    let dirty = false;
    let refreshing = false;

    async function refresh() {
      refreshing = true;
      try {
        while (dirty && listener) {
          dirty = false;
          const of = loaded.get(name);
          if (of === undefined) continue;
          try {
            const next = await read(of);
            if (next === value) continue;
            const ops = diffRevisions(value, next);
            if (ops.length === 0) continue;
            value = next;
            await listener?.(value, ops);
          } catch (error) {
            console.error(`[berry] state of ${name} in ${id}:`, error);
          }
        }
      } finally {
        refreshing = false;
      }
    }
    const poke = () => {
      dirty = true;
      if (!refreshing) void refresh();
    };

    // commit listeners must not call Session APIs inline
    const unsubscribe = harness.subscribeCommits((publication) => {
      const touched = publication.changes.some((change) =>
        change.type === "entry" && change.value.conversationId === id
      );
      if (touched) queueMicrotask(poke);
    });
    stateFeeds.add(poke);
    return {
      get value() {
        return value;
      },
      start(next) {
        listener = next;
        poke();
      },
      stop() {
        listener = undefined;
        unsubscribe();
        stateFeeds.delete(poke);
        return Promise.resolve();
      },
    };
  }

  /** Load a berry extension, as the durable extension it compiles to unless the harness gives its own. */
  async function install(
    extension: AnyExtension,
    durable?: DurableExtension,
  ): Promise<void> {
    const providers: Provider[] = [];
    const compactionCaps: (() => CompactionCap | undefined)[] = [];
    const tools: AnyTool[] = [];
    const handlers: AnyHandler[] = [];
    const channels: AnyChannelHandler[] = [];
    const prompts: { name: string; render: SectionRender<JsonValue> }[] = [];
    const disposers: (() => void | Promise<void>)[] = [];
    let configuration: Configuration<JsonValue, JsonValue> | undefined;
    let promptProse: PromptProse | undefined;
    const conversation = extension.setup({
      provider: (provider) => void providers.push(provider),
      auth: (provider, signal) => models.getAuth(provider, { signal }),
      tool: (tool) => void tools.push(tool),
      handle: (handler) => void handlers.push(handler as AnyHandler),
      channel: (handler) => void channels.push(handler as AnyChannelHandler),
      prompt: (name, render) => void prompts.push({ name, render }),
      promptProse: (given) => void (promptProse = given),
      compactionCap: (getter) => void compactionCaps.push(getter),
      configuration: (given) =>
        void (configuration = given as unknown as Configuration<JsonValue, JsonValue>),
      onDispose: (disposer) => void disposers.push(disposer),
    });
    const replacedProviders = new Map<string, Provider>();
    const next: Loaded = {
      extension,
      providers,
      replacedProviders,
      compactionCaps,
      tools,
      handlers,
      channels,
      open: new Set(),
      prompts,
      disposers,
      calls: idle(),
      ...(conversation ? { conversation } : {}),
      ...(configuration ? { configuration } : {}),
      ...(promptProse ? { promptProse } : {}),
    };
    // the model resolves same-named tools to the later extension, which would shadow the harness's own
    if (tools.some((tool) => tool.name === "codemode" && exposure(tool) !== "codemode")) {
      await dispose(next);
      throw new BerryError(`${extension.name}: the model's codemode tool is the harness's own`);
    }
    if (tools.some((tool) => tool.name === SUBAGENT)) {
      await dispose(next);
      throw new BerryError(`${extension.name}: the subagent tool is the harness's own`);
    }
    const previous = loaded.get(extension.name);
    const claimed = new Set<string>();
    for (const provider of providers) {
      if (
        claimed.has(provider.id) ||
        [...loaded.values()].some((of) =>
          of !== previous && of.providers.some((owned) => owned.id === provider.id)
        )
      ) {
        await dispose(next);
        throw new BerryError(`${extension.name}: provider ${provider.id} is already registered`);
      }
      claimed.add(provider.id);
    }
    if (previous) releaseProviders(previous);
    for (const provider of providers) {
      const original = models.getProvider(provider.id);
      if (original) replacedProviders.set(provider.id, original);
      models.setProvider(provider);
    }
    loaded.set(extension.name, next);
    registry.install(durable ?? compile(next));
    await rename(extension);
    if (providers.length) {
      const refreshed = await models.refresh({
        providers: providers.map((provider) => provider.id),
      });
      for (const [id, error] of refreshed.errors) {
        console.error(`[berry] refreshing ${id}:`, error);
      }
    }
    for (const refresh of stateFeeds) refresh();
    // calls the old load started finish on it, and may still need what it disposes
    if (previous) void drained(previous).then(() => dispose(previous));
  }

  /** Where a conversation works now. */
  async function bindingOf(id: ConversationId, tx?: Tx): Promise<Binding> {
    return await stateOf(id, loaded.get(ENVIRONMENT)!, tx) as Binding;
  }

  async function requireEnvironmentAllowed(name: string) {
    if (
      name === LOCAL &&
      (await harness.snapshot(Settings, context))?.allowLocalEnvironment === false
    ) {
      throw new BerryError("The local environment is disabled in settings");
    }
  }

  async function environmentOf(id: ConversationId, during: Context = context) {
    const { environment, cwd } = await bindingOf(id);
    return await environments.open(environment, cwd, during);
  }

  /** Move a conversation to another machine, which only an idle one may do. */
  async function rebind(id: ConversationId, events: readonly BindingEvent[]) {
    // mid-run, calls the model made for one machine would reach the other
    await harness.commit(async (tx) => {
      if ((await tx.doc(LiveDoc, id)).run || (await tx.doc(ShellDoc, id)).run) {
        throw new BerryError("busy: let it finish, or abort it, before changing its environment");
      }
      for (const event of events) {
        await tx.appendEntry(id, { kind: EventEntry.kind, data: { ext: ENVIRONMENT, event } });
      }
    }, context);
  }

  await install(environmentExtension(environments));
  const subagent = subagentExtension({
    harness,
    tidOf,
    enabled: async () => (await harness.snapshot(Settings, context))?.models ?? [],
    workers: async (id) => await stateOf(id, loaded.get(SUBAGENT)!) as Subagent,
    async title(tx, id, task) {
      const info = Object.values(await tx.doc(Conversations))
        .find((info) => branchesOf(info).includes(id));
      if (info) info.title = task.replace(/\s+/g, " ").trim();
    },
    supported,
    async prose() {
      return await resolveProse(SUBAGENT, loaded.get(SUBAGENT)?.promptProse);
    },
    async resolve(tid) {
      const info = (await harness.snapshot(Conversations, context))?.[tid];
      return info &&
        { id: headOf(info), ...(info.owner === undefined ? {} : { owner: info.owner }) };
    },
    async prepare(api, model, thinkingLevel, during) {
      const agent = await api.agent(during);
      const extensions = agent.extensions.filter((ext) => ext.name !== SUBAGENT);
      const events: { ext: string; event: JsonValue }[] = [];
      for (const { name } of extensions) {
        const of = loaded.get(name);
        if (!of?.configuration || name === ENVIRONMENT) continue;
        const state = await stateOf(api.conversationId, of);
        const values = of.configuration.values(state);
        const seeds = await configurationEvents(of.configuration, values, of.extension.initial());
        for (const event of seeds) events.push({ ext: name, event });
      }
      const env = api.env as Environment;
      events.push({
        ext: ENVIRONMENT,
        event: { type: "bind", environment: env.name, cwd: env.cwd },
      });
      return {
        model,
        thinkingLevel: supported(model, thinkingLevel ?? agent.thinkingLevel),
        extensions: extensions
          .filter((ext) => !loaded.get(ext.name)?.extension.alwaysOn)
          .map((ext) => ext.name),
        events,
      };
    },
  });
  await install(subagent.workers, subagent.extension);

  async function stateAt(tid: string, name: string, at?: EntryId): Promise<JsonValue | undefined> {
    const id = at === undefined ? await idOf(tid) : (await locate(tid, at))?.id;
    const of = loaded.get(name);
    if (id === undefined || of === undefined) return undefined;
    if (at === undefined) return await stateOf(id, of);
    let state = of.extension.initial();
    let cursor;
    do {
      const page = await (await conversation(id)).entries(
        { order: "ascending", maxEntryId: at },
        256,
        cursor,
        context,
      );
      for (const entry of page.items) state = fold(of.extension, state, entry);
      cursor = page.next;
    } while (cursor !== undefined);
    return state;
  }

  return {
    harness,

    async changeQueuedInput(tid: string, id: SubmissionId, text: string | null) {
      return await harness.commit(async (tx) => {
        const info = (await tx.doc(Conversations))[tid];
        if (info === undefined) return "not_found" as const;
        const conversationId = headOf(info);
        // public Tx omits submission-by-id; this read still holds the harness's commit line.
        const record = await storage.submission(id, context);
        if (record?.conversationId !== conversationId || record.type !== "input") {
          return "not_found" as const;
        }
        if (record.status !== "queued") return "not_queued" as const;
        const inbox = await tx.doc(InboxDoc, conversationId);
        const index = inbox.items.findIndex((item) => item.id === id);
        const item = inbox.items[index];
        if (item === undefined) return "not_queued" as const;
        if (item.mode === "write") return "not_found" as const;
        if (text === null) {
          tx.settleSubmission(id, { status: "unanswered", reason: "aborted" });
          inbox.items.splice(index, 1);
        } else {
          const images = typeof item.content === "string"
            ? []
            : item.content.filter((block) => block.type === "image");
          if (!text.trim() && images.length === 0) return "empty" as const;
          const content = userContent(text, images);
          if (new TextEncoder().encode(JSON.stringify({ content })).length > MAX_MESSAGE_BYTES) {
            return "too_large" as const;
          }
          item.content = content;
        }
        return "ok" as const;
      }, context);
    },

    /** Load an extension, replacing the installed one with its name. A failing setup changes nothing. */
    async install(extension: AnyExtension): Promise<void> {
      if (
        extension.name === Builtin.name || extension.name === ENVIRONMENT ||
        extension.name === SUBAGENT
      ) {
        throw new BerryError(`${extension.name} is the harness's own`);
      }
      await install(extension);
    },

    /** Remove an extension: conversations that select it go without it, until one by its name is installed again. */
    uninstall(name: string) {
      const previous = loaded.get(name);
      if (previous === undefined || previous === own || name === ENVIRONMENT || name === SUBAGENT) {
        return;
      }
      loaded.delete(name);
      releaseProviders(previous);
      const installed = registry.snapshot().extension(name);
      if (installed) registry.uninstall(installed);
      for (const refresh of stateFeeds) refresh();
      void drained(previous).then(() => dispose(previous));
    },

    extensions: () =>
      [...loaded.values()].flatMap((of) =>
        of === own || of.extension.name === ENVIRONMENT || of.extension.name === SUBAGENT
          ? []
          : [of.extension]
      ),

    /** The harness's own extensions that the ui shows as extensions: those with configuration. */
    builtins: (): readonly Pick<
      AnyExtension,
      "name" | "description" | "alwaysOn" | "defaultEnabled" | "sortOrder" | "formerly"
    >[] => [
      loaded.get(ENVIRONMENT)!.extension,
      loaded.get(SUBAGENT)!.extension,
    ],

    /** What a conversation starting with the extension is asked, as a json schema; `null` when nothing. */
    configurationParameters: (name: string) => loaded.get(name)?.configuration?.parameters ?? null,

    /** The tightest compaction cap any installed extension reports, if any. */
    compactionCap: () => effectiveCap(),

    /** Model catalog as generation sees it: effective windows under the tightest cap. */
    effectiveModels,

    async saveProject(id: string | undefined, input: ProjectInput): Promise<string> {
      const projects = await harness.snapshot(Projects, context) ?? {};
      if (id !== undefined && !Object.hasOwn(projects, id)) {
        throw new BerryError(`No project ${id}`);
      }
      const key = id ?? crypto.randomUUID();
      try {
        const project = await resolveProject(
          input,
          id ? projects[id] : undefined,
          environments,
          context,
        );
        await harness.commit(async (tx) => {
          const index = await tx.doc(Projects);
          if (id !== undefined && !Object.hasOwn(index, id)) {
            throw new BerryError(`No project ${id}`);
          }
          validateRoots(index, key, project);
          const previous = index[key];
          if (input.moveConversations && previous) {
            const removed = previous.roots.filter((root) =>
              !project.roots.some((saved) =>
                saved.environment === root.environment &&
                workspacePath(saved.cwd, saved.platform) === workspacePath(root.cwd, root.platform)
              )
            );
            if (removed.length) {
              const destination = project.roots[project.defaultRoot];
              const moving: ConversationId[] = [];
              for (const info of Object.values(await tx.doc(Conversations))) {
                const conversationId = headOf(info);
                const binding = await bindingOf(conversationId, tx);
                if (binding.cwd === null && binding.environment !== LOCAL) continue;
                const location = {
                  environment: binding.environment,
                  cwd: binding.cwd ?? homedir(),
                };
                if (projectFor(index, location) !== key) continue;
                const workspace = workspaceFor(previous, location);
                if (!workspace || !removed.includes(workspace)) continue;
                if (
                  (await tx.doc(LiveDoc, conversationId)).run ||
                  (await tx.doc(ShellDoc, conversationId)).run
                ) {
                  throw new BerryError(
                    "busy: let affected conversations finish, or abort them, before moving their workspaces",
                  );
                }
                moving.push(conversationId);
              }
              const event: BindingEvent = {
                type: "bind",
                environment: destination.environment,
                cwd: destination.cwd,
              };
              for (const conversationId of moving) {
                await tx.appendEntry(conversationId, {
                  kind: EventEntry.kind,
                  data: { ext: ENVIRONMENT, event },
                });
              }
            }
          }
          index[key] = project;
        }, context);
        return key;
      } catch (error) {
        throw surfaced(error);
      }
    },

    async deleteProject(id: string) {
      await harness.commit(async (tx) => {
        const projects = await tx.doc(Projects);
        if (!Object.hasOwn(projects, id)) throw new BerryError(`No project ${id}`);
        delete projects[id];
      }, context);
    },

    async folders(
      environment: string,
      cwd: string | null,
      prefix: string,
      during: Context = context,
    ) {
      try {
        const env = await environments.open(environment, cwd, during);
        return await folders(env, prefix, during);
      } catch (error) {
        throw surfaced(error);
      }
    },

    /** Paths a reference could extend to from `cwd`, before any conversation works there. */
    async references(
      environment: string,
      cwd: string | null,
      prefix: string,
      during: Context = context,
    ) {
      try {
        const env = await environments.open(environment, cwd, during);
        return await references(env, prefix, during);
      } catch (error) {
        throw surfaced(error);
      }
    },

    async create(options: CreateOptions): Promise<string> {
      if (options.model) await requireEnabled(options.model);
      const selected = selection(Object.keys(options.extensions));
      const seeds = await Promise.all(
        Object.entries(options.extensions).map(async ([given, values]) => {
          const name = current(given);
          const of = loaded.get(name);
          try {
            const events = of?.configuration
              ? await configurationEvents(of.configuration, values, of.extension.initial())
              : [];
            return [name, events] as const;
          } catch (error) {
            throw surfaced(error, `${name}: `);
          }
        }),
      );
      const thinkingLevel = options.model
        ? supported(options.model, options.thinkingLevel ?? "high")
        : options.thinkingLevel;
      const created = await harness.createConversation({
        ownership: { kind: "ownerless" },
        agent: {
          extensions: selected,
          ...(options.model ? { model: options.model } : {}),
          ...(thinkingLevel ? { thinkingLevel } : {}),
        },
        init: async (tx, id) => {
          for (const [ext, events] of seeds) {
            for (const event of events) {
              await tx.appendEntry(id, { kind: EventEntry.kind, data: { ext, event } });
            }
          }
        },
      }, context);
      return await tidOf(created.id);
    },

    conversation: conversationOf,

    state: stateAt,

    /** The environment `tid` is bound to, now or as of its entry `at`. */
    async environment(tid: string, at?: EntryId): Promise<Environment> {
      const binding = await stateAt(tid, ENVIRONMENT, at) as Binding | undefined;
      if (binding === undefined) throw new BerryError(`No entry ${at} in ${tid}`);
      return await environments.open(binding.environment, binding.cwd, context);
    },

    async configure(tid: string, change: AgentChange) {
      const target = await conversationOf(tid);
      if (target === undefined) throw new BerryError(`No conversation ${tid}`);
      if (change.model) await requireEnabled(change.model);
      const extensions = change.extensions && selection(change.extensions);
      // a model change carries the level over as far as the new model allows
      const agent = await target.agent(context);
      const model = change.model === undefined ? agent.model : change.model ?? undefined;
      const requested = change.thinkingLevel === undefined
        ? agent.thinkingLevel
        : change.thinkingLevel ?? "off";
      const thinkingLevel = model && (change.model !== undefined || change.thinkingLevel)
        ? supported(model, requested)
        : change.thinkingLevel;
      await target.configure({
        ...(change.model === undefined ? {} : { model: change.model }),
        ...(thinkingLevel === undefined ? {} : { thinkingLevel }),
        ...(extensions === undefined ? {} : { extensions }),
      }, context);
    },

    /** An extension's state as a feed, following the conversation from branch to branch. */
    watchState: (tid: string, name: string) => follow(tid, (id) => stateFeed(id, name)),

    watchConfiguration: (tid: string, name: string) =>
      follow(tid, (id) => stateFeed(id, name, true)),

    /** The durable view of a conversation as a feed, following it from branch to branch. */
    watch: (tid: string) =>
      follow(tid, async (id) => await (await harness.conversation(id, context))?.watch(context)),

    locate,

    /** Acknowledge only the message the frontend saw, never a newer completed turn. */
    async markRead(tid: string, entryId: number) {
      const found = await locate(tid, entryId);
      if (found === undefined) throw new BerryError(`No entry ${entryId} in ${tid}`);
      await harness.commit(async (tx) => {
        const entry = await tx.entry(entryId as EntryId);
        if (entry?.kind !== "pi.user" && entry?.kind !== "pi.assistant") {
          throw new BerryError(`Entry ${entryId} is not a message`);
        }
        const info = (await tx.doc(Conversations))[tid];
        if (info === undefined) throw new BerryError(`No conversation ${tid}`);
        info.readThrough = Math.max(info.readThrough ?? 0, entryId);
      }, context);
    },

    /** Set a sidebar name, or restore the message preview with an empty name. */
    async rename(tid: string, title: string | null) {
      await harness.commit(async (tx) => {
        const info = (await tx.doc(Conversations))[tid];
        if (info === undefined) throw new BerryError(`No conversation ${tid}`);
        const name = title?.trim();
        if (name) info.title = name;
        else delete info.title;
      }, context);
    },

    /** Put a conversation out of the way, or bring it back. */
    async archive(tid: string, archived: boolean) {
      await harness.commit(async (tx) => {
        const info = (await tx.doc(Conversations))[tid];
        if (info === undefined) throw new BerryError(`No conversation ${tid}`);
        if (!archived) delete info.archived;
        else info.archived ??= Date.now();
      }, context);
    },

    branch,

    /**
     * Have the agent write its message `at` afresh: continue `tid` from just before it, on a run without new input,
     * which pi-durable starts as its own `startRun` does.
     */
    async regenerate(tid: string, at: number) {
      const source = await locate(tid, at);
      if (source === undefined) throw new BerryError(`No entry ${at} in ${tid}`);
      const one = async (query: { minEntryId?: EntryId; maxEntryId: EntryId }) =>
        (await source.entries({ order: "descending", ...query }, 1, undefined, context)).items[0];
      const entry = await one({ minEntryId: at as EntryId, maxEntryId: at as EntryId });
      if (entry?.kind !== "pi.assistant") {
        throw new BerryError(`Entry ${at} is not the agent's message`);
      }
      const before = await one({ maxEntryId: (at - 1) as EntryId });
      if (before === undefined) throw new BerryError("Nothing comes before this message");
      await branch(tid, before.id);
      const id = await existing(tid);
      await harness.commit(async (tx) => {
        const live = await tx.doc(LiveDoc, id);
        if (live.run || (await tx.doc(ShellDoc, id)).run) {
          throw new BerryError(`${tid} is busy: let it finish, or abort it, first`);
        }
        const taskId = await tx.createTask(GenerationTask, {}, {
          ownership: { kind: "conversation" },
          conversationId: id,
        });
        live.run = { taskId, inputs: [] };
      }, context);
    },

    /** A host-originated event: a write submission, placed at the next boundary while busy. */
    async emit(tid: string, ext: string, event: JsonValue) {
      const id = await existing(tid);
      return await writeEvent(id, ext, event);
    },

    async configureExtension(tid: string, name: string, change: Record<string, JsonValue>) {
      const id = await existing(tid);
      const of = (await selected(id)).find((of) => of.extension.name === name);
      if (!of?.configuration) throw new BerryError(`${name} has no enabled configuration`);
      const configuration = of.configuration;
      try {
        for (const key of Object.keys(change)) {
          if (!Object.hasOwn(configuration.parameters.properties, key)) {
            throw new Error(`unknown configuration field: ${key}`);
          }
        }
        const state = await stateOf(id, of);
        const events = await configurationEvents(configuration, {
          ...configuration.values(state),
          ...change,
        }, state);
        const moving = name === ENVIRONMENT &&
          (events as readonly BindingEvent[]).some((event) =>
            event.environment !== (state as Binding).environment
          );
        if (moving) await rebind(id, events as readonly BindingEvent[]);
        else {
          const submissions = await Promise.all(
            events.map((event) => writeEvent(id, name, event)),
          );
          await Promise.all(submissions.map((submission) => submission.wait(context)));
        }
        return configuration.values(await stateOf(id, of));
      } catch (error) {
        throw surfaced(error);
      }
    },

    async commands(tid: string) {
      const id = await existing(tid);
      return [...(await commandsOf(id)).values()].map(({ loaded, command }) => ({
        name: command.name,
        extension: loaded.extension.name,
        description: command.description,
      }));
    },

    /** Hand a channel the frontend opened to the extension it names; refusals surface as `BerryError`. */
    async channel(
      tid: string,
      extension: string,
      name: string,
      args: JsonValue | undefined,
      channel: Channel,
      signal: AbortSignal,
    ): Promise<void> {
      const id = await existing(tid);
      const of = (await selected(id)).find((of) => of.extension.name === extension);
      if (of === undefined) throw new BerryError(`${extension} is not in this conversation`);
      const handler = of.channels.findLast((handler) => handler.name === name);
      if (handler === undefined) throw new BerryError(`${extension} has no channel named ${name}`);
      try {
        const validated = handlerArguments(handler.parameters, args);
        const state = await stateOf(id, of);
        const opened = await environmentOf(id).then(
          (env) => ({ env }),
          (error: unknown) => ({ error }),
        );
        of.open.add(channel);
        signal.addEventListener("abort", () => of.open.delete(channel), { once: true });
        await handler.open(validated, channel, {
          tid,
          state,
          get env() {
            if ("error" in opened) throw opened.error;
            return opened.env;
          },
          signal,
        });
      } catch (error) {
        throw surfaced(error);
      }
    },

    /** Answer an extension's frontend with one of its handlers; its failures surface as `BerryError`. */
    async request(
      tid: string,
      extension: string,
      name: string,
      args: JsonValue | undefined,
      signal: AbortSignal = NEVER,
    ): Promise<JsonValue> {
      const id = await existing(tid);
      const of = (await selected(id)).find((of) => of.extension.name === extension);
      if (of === undefined) throw new BerryError(`${extension} is not in this conversation`);
      const handler = of.handlers.findLast((handler) => handler.name === name);
      if (handler === undefined) throw new BerryError(`${extension} has no handler named ${name}`);
      try {
        const validated = handlerArguments(handler.parameters, args);
        const state = await stateOf(id, of);
        // a handler that does not use the environment still answers while it cannot be reached
        const opened = await environmentOf(id).then(
          (env) => ({ env }),
          (error: unknown) => ({ error }),
        );
        return await running(of, async () =>
          await handler.execute(validated, {
            tid,
            state,
            get env() {
              if ("error" in opened) throw opened.error;
              return opened.env;
            },
            signal,
          }) ?? null);
      } catch (error) {
        throw surfaced(error);
      }
    },

    /** Run a command; its failures are the user's to read, so they surface as `BerryError`. */
    async command(tid: string, name: string, args: string): Promise<JsonValue | undefined> {
      const id = await existing(tid);
      const found = (await commandsOf(id)).get(name);
      if (found === undefined) throw new BerryError(`No command /${name}`);
      const { loaded: of, command } = found;
      // a command that does not use the environment still runs while it cannot be reached
      const opened = await environmentOf(id).then(
        (env) => ({ env }),
        (error: unknown) => ({ error }),
      );
      const envOf = () => {
        if ("error" in opened) throw opened.error;
        return opened.env;
      };
      try {
        return await command.execute(args, {
          tid,
          state: await stateOf(id, of),
          get env() {
            return envOf();
          },
          emit: async (event, options) => {
            const submission = await writeEvent(id, of.extension.name, event, options?.content);
            await submission.wait(context);
          },
          async cd(path, options) {
            const env = envOf();
            const cwd = await directory(env, path, context);
            if (options?.inheritProject) {
              await harness.commit(async (tx) => {
                const projects = await tx.doc(Projects);
                const projectId = projectFor(projects, { environment: env.name, cwd: env.cwd });
                if (projectId === null) return;
                const project = projects[projectId];
                if (
                  project.roots.some((root) =>
                    root.environment === env.name &&
                    workspacePath(root.cwd, root.platform) === workspacePath(cwd, env.platform)
                  )
                ) return;
                const updated = {
                  ...project,
                  roots: [...project.roots, { environment: env.name, cwd, platform: env.platform }],
                };
                validateRoots(projects, projectId, updated);
                projects[projectId] = updated;
              }, context);
            }
            const binding = await bindingOf(id);
            if (cwd !== binding.cwd) {
              const event: BindingEvent = { type: "bind", environment: binding.environment, cwd };
              await writeEvent(id, ENVIRONMENT, event);
            }
            return cwd;
          },
          compact: async (instructions) => {
            await (await conversation(id)).compact(instructions, context);
          },
        });
      } catch (error) {
        throw surfaced(error);
      }
    },

    tidOf,

    async shell(tid: string, command: string, excludeFromContext = false) {
      if (!command.trim()) throw new BerryError("Expected a shell command after !");
      const id = await existing(tid);
      if (!(await selected(id)).some((of) => of.extension.name === "code")) {
        throw new BerryError("Shell commands require the code extension");
      }
      const taskId = await harness.commit(async (tx) => {
        const live = await tx.doc(LiveDoc, id);
        const state = await tx.doc(ShellDoc, id);
        if (live.run || state.run) {
          throw new BerryError("Busy: let it finish, or abort it, before running a shell command");
        }
        const taskId = await tx.createTask(shell, { command, excludeFromContext }, {
          ownership: { kind: "conversation" },
          conversationId: id,
        });
        state.run = {
          taskId,
          command,
          excludeFromContext,
          status: "running",
          result: null,
          details: null,
          error: null,
          startedAt: Date.now(),
          durationMs: null,
        };
        return taskId;
      }, context);
      harness.resume();
      return taskId;
    },

    watchShell: (tid: string) =>
      follow(tid, async (id) => {
        await harness.commit(async (tx) => {
          await tx.doc(ShellDoc, id);
        }, context);
        return await harness.watchDoc(ShellDoc, id, context);
      }),

    /** Replace the enabled models; a model listed twice counts once. */
    async enableModels(models: readonly ModelRef[]) {
      const unique = [...new Map(models.map((m) => [`${m.provider}/${m.modelId}`, m])).values()];
      await harness.commit(async (tx) => {
        (await tx.doc(Settings)).models = unique.map(({ provider, modelId }) => ({
          provider,
          modelId,
        }));
      }, context);
    },

    async setAllowLocalEnvironment(allowed: boolean) {
      await harness.commit(async (tx) => {
        (await tx.doc(Settings)).allowLocalEnvironment = allowed;
      }, context);
    },

    /** What an extension's prose is unless settings override it; `null` when it registers none. */
    promptProseDefault: (name: string) => loaded.get(name)?.promptProse?.default ?? null,

    async setSystemPrompt(patch: Partial<SystemPromptSettings>) {
      await harness.commit(async (tx) => {
        const settings = await tx.doc(Settings);
        const current = { ...DEFAULT_SYSTEM_PROMPT, ...settings.systemPrompt };
        settings.systemPrompt = {
          ...current,
          ...patch,
          promptProse: { ...current.promptProse, ...patch.promptProse },
        };
      }, context);
    },

    /** Replace the machines conversations may work on besides this one. */
    async setEnvironments(configured: Readonly<Record<string, EnvironmentSettings>>) {
      for (const [name, { ssh, deno }] of Object.entries(configured)) {
        if (name === LOCAL || !ENVIRONMENT_NAME.test(name)) {
          throw new BerryError(`${JSON.stringify(name)} cannot name an environment`);
        }
        // the destination is an argument of its own, so this is only ssh mistaking it for an option
        if (!ssh.trim() || ssh.startsWith("-")) {
          throw new BerryError(`${name}: ${JSON.stringify(ssh)} is not an ssh destination`);
        }
        if (deno !== undefined && !deno.trim()) throw new BerryError(`${name}: deno is empty`);
      }
      await harness.commit(async (tx) => {
        (await tx.doc(Settings)).environments = Object.fromEntries(
          Object.entries(configured).map(([name, { ssh, deno }]) => [
            name,
            { ssh: ssh.trim(), ...(deno === undefined ? {} : { deno: deno.trim() }) },
          ]),
        );
      }, context);
    },

    async close() {
      await stopSettled();
      await harness.close(context);
      // a browser session, mcp servers, and each machine are independent, so none waits out another
      await Promise.allSettled(
        [...loaded.values()].map(async (of) => {
          releaseProviders(of);
          await dispose(of);
        }),
      );
      await environments.close();
    },
  };
}

export type Berry = Awaited<ReturnType<typeof openBerry>>;