From 20d86c3657e90e857309d908b13adcf7fb8d9611 Mon Sep 17 00:00:00 2001
From: Mario Zechner
Date: Thu, 5 Nov 2026 18:07:24 +0000
Subject: [PATCH] feat(p2): Chord v2 cleanup pass
p2 rewritten by the Chord v2 authoring rules (decision "App authoring rules confirmed
by the p2 audit"), in four parts merged together:
- core: session lifecycle (relocate, clone, fork, archive) and search rebuild are
persisted jobs on a new server store; Sessions v6 with client-chosen ids and visible
operations; Retry answers with the worker's real outcome; workers push branch,
subagent and status state to the server; roots take the method's ctx.
- services: models, envs, ports, login, mcp, logs, sharing and voice follow external
state (self.follow, followNative) instead of copy loops; methods answer after their
state commits; presence from host.references.
- chat: chat mode, artifacts and prompt resources follow Durable and native state;
shell commands are Chord jobs (shellId is the requestId); status pushed before
send and stop answer; race-free draft claim.
- ui: optimistic updates on web (useOptimistic) and terminal (TuiScope.optimistic,
extended in Chord with a reducer and a pending flag), pending states for long work,
visible rollback, load waterfalls removed; merged with the one composer and
immediate first send (ccff9876e) and the folder + button (5a31a6d5f).
docs/optimistic-ui.md describes the pattern.
---
apps/p2/docs/handoff.md | 1 +
apps/p2/docs/mcp-durable.md | 4 +-
apps/p2/docs/optimistic-ui.md | 137 ++++
apps/p2/docs/p2-notes.md | 127 ++--
apps/p2/docs/p2-tui-port.md | 21 +-
apps/p2/docs/plugin-hooks.md | 2 +-
apps/p2/docs/tree.md | 7 +-
apps/p2/plugins/artifacts/worker.ts | 67 +-
apps/p2/plugins/chat-mode/contract.ts | 2 +-
apps/p2/plugins/chat-mode/tui.ts | 59 +-
apps/p2/plugins/chat-mode/web.tsx | 55 +-
apps/p2/plugins/chat-mode/worker.ts | 132 ++--
apps/p2/plugins/chat/contract.ts | 47 +-
apps/p2/plugins/chat/session-start.ts | 24 +-
apps/p2/plugins/chat/shell.ts | 305 +++++++++
apps/p2/plugins/chat/tui.ts | 407 ++++++-----
apps/p2/plugins/chat/web.tsx | 236 ++++---
apps/p2/plugins/chat/worker.ts | 396 +++--------
apps/p2/plugins/codemode/calls.ts | 32 +-
apps/p2/plugins/codemode/worker.ts | 4 +-
apps/p2/plugins/conversations/server.ts | 14 +-
apps/p2/plugins/conversations/web.tsx | 8 +-
apps/p2/plugins/conversations/worker.ts | 40 +-
apps/p2/plugins/devices/server.ts | 13 +-
apps/p2/plugins/drafts/worker.ts | 13 +-
apps/p2/plugins/envs/EnvsPanel.tsx | 65 +-
apps/p2/plugins/envs/server.ts | 48 +-
apps/p2/plugins/envs/store.ts | 2 +-
apps/p2/plugins/envs/worker.ts | 9 -
apps/p2/plugins/file-index/worker.ts | 14 +-
apps/p2/plugins/history/tools.ts | 27 +-
apps/p2/plugins/history/worker.ts | 7 +-
apps/p2/plugins/locations/worker.ts | 7 +-
apps/p2/plugins/login/contract.ts | 5 +-
apps/p2/plugins/login/logins.ts | 97 ++-
apps/p2/plugins/login/server.ts | 148 +---
apps/p2/plugins/login/tui.ts | 5 +-
apps/p2/plugins/login/web.tsx | 91 ++-
apps/p2/plugins/logs/server.ts | 36 +-
apps/p2/plugins/logs/tui.ts | 38 +-
apps/p2/plugins/logs/web.tsx | 13 +-
apps/p2/plugins/mcp/config.ts | 38 +-
apps/p2/plugins/mcp/routes.ts | 26 +-
apps/p2/plugins/mcp/server.ts | 192 +++---
apps/p2/plugins/mcp/tui.ts | 84 +--
apps/p2/plugins/mcp/web.tsx | 79 ++-
apps/p2/plugins/mcp/worker.ts | 5 +-
apps/p2/plugins/models/server.ts | 53 +-
apps/p2/plugins/models/tui.ts | 221 +++---
apps/p2/plugins/models/worker.ts | 33 +-
apps/p2/plugins/ports/PortsPanel.tsx | 107 +--
apps/p2/plugins/ports/server.ts | 29 +-
apps/p2/plugins/ports/tui.ts | 144 ++--
apps/p2/plugins/ports/worker.ts | 2 +-
apps/p2/plugins/powershell/worker.ts | 10 +-
apps/p2/plugins/prompt/tui.ts | 9 +-
apps/p2/plugins/prompt/worker.ts | 129 ++--
apps/p2/plugins/search/contract.ts | 36 +-
apps/p2/plugins/search/server.ts | 148 ++--
apps/p2/plugins/search/tui.ts | 57 +-
apps/p2/plugins/search/worker.ts | 18 +-
apps/p2/plugins/sessions/BranchSheet.tsx | 17 +-
apps/p2/plugins/sessions/SessionOptions.tsx | 65 --
apps/p2/plugins/sessions/SessionSheet.tsx | 15 +
apps/p2/plugins/sessions/contract.ts | 108 ++-
apps/p2/plugins/sessions/copy.ts | 15 +-
apps/p2/plugins/sessions/server.ts | 647 +++++++++++++-----
apps/p2/plugins/sessions/tui.ts | 238 +++++--
apps/p2/plugins/sessions/web.tsx | 196 +++++-
apps/p2/plugins/sessions/worker.ts | 31 +-
apps/p2/plugins/sharing/contract.ts | 9 +-
apps/p2/plugins/sharing/db.ts | 39 +-
apps/p2/plugins/sharing/server.ts | 140 ++--
apps/p2/plugins/sharing/tui.ts | 11 +-
apps/p2/plugins/sharing/web.tsx | 17 +-
apps/p2/plugins/status/contract.ts | 37 +-
apps/p2/plugins/status/server.ts | 121 ++--
apps/p2/plugins/status/tui.ts | 4 +-
apps/p2/plugins/status/worker.ts | 75 +-
apps/p2/plugins/subagents/tui.ts | 56 +-
apps/p2/plugins/subagents/worker.ts | 140 ++--
apps/p2/plugins/terminals/server.ts | 8 +-
apps/p2/plugins/terminals/worker.ts | 2 +-
apps/p2/plugins/tree/worker.ts | 53 +-
apps/p2/plugins/voice/server.ts | 4 +-
apps/p2/plugins/voice/web.tsx | 20 +-
apps/p2/plugins/voice/worker.ts | 83 +--
apps/p2/plugins/workspace/server.ts | 24 +-
apps/p2/plugins/workspace/tui.ts | 12 +-
apps/p2/plugins/workspace/web.tsx | 32 +-
apps/p2/plugins/workspace/worker.ts | 37 +-
apps/p2/src/api/v1/index.ts | 8 +
apps/p2/src/hosts/server.ts | 17 +-
apps/p2/src/hosts/session-workers.ts | 62 +-
apps/p2/src/hosts/tui-client.ts | 3 +
apps/p2/src/hosts/tui-worker-runtime.ts | 43 +-
apps/p2/src/hosts/tui.ts | 1 +
apps/p2/src/hosts/worker.ts | 6 +-
apps/p2/src/runtime/chord.ts | 31 +-
apps/p2/src/runtime/refs.ts | 2 +-
apps/p2/src/runtime/server.ts | 65 +-
apps/p2/src/runtime/tui-context.ts | 7 +
apps/p2/src/runtime/tui.ts | 108 ++-
apps/p2/src/runtime/web.tsx | 97 +--
apps/p2/src/runtime/worker.ts | 9 +-
apps/p2/src/web/AppLayout.tsx | 10 +-
apps/p2/src/web/Drawer.tsx | 28 +-
apps/p2/src/web/composer-state.tsx | 7 +-
apps/p2/src/web/history.ts | 5 +-
apps/p2/src/web/navigation.ts | 27 +
apps/p2/src/web/style.css | 17 +
apps/p2/test/fixtures/optimistic.ts | 54 ++
apps/p2/test/p2/chat-contracts.test.ts | 8 +-
apps/p2/test/p2/chat-mode-host.ts | 3 +
apps/p2/test/p2/chat-mode.test.ts | 20 +-
apps/p2/test/p2/contract-versions.test.ts | 52 +-
apps/p2/test/p2/drafts.test.ts | 11 +
apps/p2/test/p2/envs.test.ts | 4 +-
apps/p2/test/p2/fork-clone.test.ts | 32 +-
apps/p2/test/p2/helpers.ts | 12 +-
apps/p2/test/p2/lifecycle.test.ts | 7 +-
apps/p2/test/p2/login.test.ts | 29 +-
apps/p2/test/p2/logs.test.ts | 3 +-
apps/p2/test/p2/m1.test.ts | 6 +-
apps/p2/test/p2/mcp-sign-in.test.ts | 9 +-
apps/p2/test/p2/mcp.test.ts | 7 +-
apps/p2/test/p2/plugins.test.ts | 3 +-
apps/p2/test/p2/ports.test.ts | 10 +
apps/p2/test/p2/rebuild-index.test.ts | 9 +-
apps/p2/test/p2/relocate.test.ts | 21 +-
apps/p2/test/p2/session-operations.test.ts | 106 +++
apps/p2/test/p2/session-start.test.ts | 37 +-
apps/p2/test/p2/sharing.test.ts | 8 +-
apps/p2/test/p2/shell.test.ts | 18 +-
apps/p2/test/p2/subagents.test.ts | 7 +-
apps/p2/test/p2/test-host.test.ts | 14 +-
apps/p2/test/p2/tui-fork.test.ts | 29 +-
apps/p2/test/p2/tui-location.test.ts | 11 +-
apps/p2/test/p2/tui-optimistic.test.ts | 262 +++++++
apps/p2/test/p2/tui-p4.test.ts | 16 +-
apps/p2/test/p2/tui-render.test.ts | 6 +-
apps/p2/test/p2/tui-tree.test.ts | 2 +
apps/p2/test/p2/tui.test.ts | 2 +
apps/p2/test/p2/ui-optimistic.test.ts | 42 ++
apps/p2/test/p2/ui-sheets.test.ts | 2 +-
apps/p2/test/p2/voice.test.ts | 7 +-
packages/chord/docs/design.md | 5 +-
.../chord/docs/guide-plugins-hosts/book.md | 13 +-
.../chord/docs/guide-plugins-hosts/index.html | 13 +-
packages/chord/src/tui/README.md | 5 +-
packages/chord/src/tui/index.ts | 2 +-
packages/chord/src/tui/scope.ts | 47 +-
.../test/examples/63-optimistic-answers.ts | 4 +-
packages/chord/test/tui/scope.test.ts | 46 +-
154 files changed, 4996 insertions(+), 3002 deletions(-)
create mode 100644 apps/p2/docs/optimistic-ui.md
create mode 100644 apps/p2/plugins/chat/shell.ts
delete mode 100644 apps/p2/plugins/sessions/SessionOptions.tsx
create mode 100644 apps/p2/test/fixtures/optimistic.ts
create mode 100644 apps/p2/test/p2/session-operations.test.ts
create mode 100644 apps/p2/test/p2/tui-optimistic.test.ts
create mode 100644 apps/p2/test/p2/ui-optimistic.test.ts
diff --git a/apps/p2/docs/handoff.md b/apps/p2/docs/handoff.md
index 9bdb83cb9..230f9be81 100644
--- a/apps/p2/docs/handoff.md
+++ b/apps/p2/docs/handoff.md
@@ -173,6 +173,7 @@ From `AGENTS.md` and this project's history:
| `README.md`, `DEV.md` | running, deployment behind a proxy, TUI sign-in and isolation |
| `docs/p2-notes.md` | decisions, configuration, plugin-by-plugin notes |
| `docs/p2-tui-port.md` | TUI design, as-built notes, open phases |
+| `docs/optimistic-ui.md` | optimistic UI: the pattern for actions, pending states and loads (web and terminal) |
| `docs/workflows.md` | code mode workflows with one-shot agents |
| `docs/tree.md` | `/tree`: branches of the main conversation as Durable forks |
| `docs/remote-envs.md` | SSH environments |
diff --git a/apps/p2/docs/mcp-durable.md b/apps/p2/docs/mcp-durable.md
index 4792570eb..fab382895 100644
--- a/apps/p2/docs/mcp-durable.md
+++ b/apps/p2/docs/mcp-durable.md
@@ -225,8 +225,8 @@ export class RemoteProcess {
(`mcpNamespace(name)|url`). The refresh lock is a promise chain per server in the process, not `proper-lockfile`: one
p2 server owns a data directory (process lock), and nothing else reads this file.
- **Flows:** `sign-in.ts` (`SignIns`, a native of the manager), in memory, one per server (a new one replaces it),
- 10 minutes each; listed (`signIns`: id, server, `via: "loopback" | "route"`, `waiting | failed`, error) in the
- manager's `Listed` document. Not a Chord flow object: the manager's methods are `beginSignIn`, `completeSignIn`
+ 10 minutes each; listed (`signIns`: id, server, `via: "loopback" | "route"`, `waiting | failed`, error) as the
+ manager follows them (`self.follow`: a method's change is listed when it answers). Not a Chord flow object: the manager's methods are `beginSignIn`, `completeSignIn`
(callback route), `pasteSignIn`, `cancelSignIn`, `signOut`, `probe`. A done flow disappears; the instances of its
server restart (`Instances.reconnect`). A pre-registered client (`oauth.clientId`) always uses loopback.
- **Links by server, not flow:** `GET /mcp/sign-in/` begins (or replaces) the server's flow, so the link is
diff --git a/apps/p2/docs/optimistic-ui.md b/apps/p2/docs/optimistic-ui.md
new file mode 100644
index 000000000..9c6d49aa9
--- /dev/null
+++ b/apps/p2/docs/optimistic-ui.md
@@ -0,0 +1,137 @@
+# Optimistic UI in p2
+
+Every action whose effect the user waits to see shows that effect at once: the switch flips, the row is renamed, the
+message is in the transcript. This page is the pattern p2's plugins follow, on the web and in the terminal, and the
+reference for plugin authors.
+
+## Why it works without flicker
+
+Two rules make an optimistic value safe to drop the moment its call settles:
+
+1. **A method that should feel instant awaits its own effects** and keeps them on the host that answers (Chord's
+ guide; decision "App authoring rules confirmed by the p2 audit"). `Sessions.rename` commits the catalog before it
+ returns; `ChatMode.set` returns once the followed Durable document changed.
+2. **A call's answer follows its state** (Chord decision of 2026-11-05): before a host sends a `result`, it sends the
+ page the `state` it committed. When the call resolves on the page, the new value is already there.
+
+So the page shows a guess while the call is in flight and simply forgets it when the call settles: on success the
+real value has replaced it in the same frame; on failure the real value is the old one, so the guess rolls back by
+itself, and the error is shown. Nothing on the page reconciles a guess with the server's value, compares ids or
+waits for state after an answer.
+
+A method that returns before its effect (a background job, a native that reports later) breaks rule 1: its guess
+would vanish before the real value arrives. Such work shows a **pending state** instead (below), never a guess of
+its outcome.
+
+## Web: `useOptimisticCall`
+
+`useOptimisticCall(value, apply, onError)` (`src/runtime/web.tsx`, from `@p2/app/v1/web`) is React's
+`useOptimistic` plus a transition: it returns `[shown, run, pending]`.
+
+```tsx
+// plugins/chat-mode/web.tsx: the chat mode chip
+const { fail } = useComposerState(); // the app's toast
+const [{ enabled, name }, guess] = useOptimisticCall>(
+ { enabled: useReadable(mode?.enabled) ?? false, name: useReadable(mode?.name) ?? DEFAULT_AGENT_NAME },
+ (shown, change) => ({ ...shown, ...change }),
+ fail,
+);
+// ...
+
- {error && {error}
}
@@ -142,7 +147,7 @@ function ChatModeChip() {
{renaming && (
mode.rename({ name: next }, BACKGROUND_CONTEXT)}
+ onSave={(next) => guess({ name: next }, () => mode.rename({ name: next }, BACKGROUND_CONTEXT))}
onClose={() => setRenaming(false)}
/>
)}
diff --git a/apps/p2/plugins/chat-mode/worker.ts b/apps/p2/plugins/chat-mode/worker.ts
index 8a999710d..9d15f2c94 100644
--- a/apps/p2/plugins/chat-mode/worker.ts
+++ b/apps/p2/plugins/chat-mode/worker.ts
@@ -9,8 +9,9 @@ import {
type EntryId,
type EntryRecord,
InboxDoc,
+ type InboxState,
} from "@earendil-works/pi-durable";
-import { bool, num, obj } from "@earendil-works/pi-schema";
+import { num, obj } from "@earendil-works/pi-schema";
import {
type Context,
type Ctx,
@@ -18,17 +19,15 @@ import {
contribute,
defineFacet,
defineState,
+ deriveTyped,
forbidden,
invalid,
isOwner,
- signalOf,
} from "@p2/app/v1";
-import { durableContext } from "@p2/app/v1/node";
-import { agentConversation, CommitSignal, harnessOf } from "@p2/app/v1/worker";
+import { agentConversation, harnessOf } from "@p2/app/v1/worker";
import { Conversation, SendRouters } from "../chat/contract.ts";
import { Attachments, MAX_ATTACHMENTS } from "../files/contract.ts";
import {
- AgentNameSchema,
CHAT_ENTRY,
type ChatMessage,
ChatMode,
@@ -57,16 +56,11 @@ export const ChatModeDoc = defineDoc<{ enabled: boolean; name: string }>({
/** A conversation's newest entries, newest first, until `unmentioned` has what it needs (at most `limit`). */
async function recentEntries(durable: DurableHarness, id: ConversationId, ctx: Context, limit = 2000) {
- const conversation = await durable.conversation(id, durableContext(ctx));
+ const conversation = await durable.conversation(id, ctx);
const entries: EntryRecord[] = [];
let before = Number.MAX_SAFE_INTEGER;
while (conversation && entries.length < limit && !readEnough(entries)) {
- const page = await conversation.entries(
- { maxEntryId: (before - 1) as EntryId },
- 256,
- undefined,
- durableContext(ctx),
- );
+ const page = await conversation.entries({ maxEntryId: (before - 1) as EntryId }, 256, undefined, ctx);
for (const entry of page.items) {
entries.push(entry);
before = entry.id;
@@ -81,20 +75,20 @@ async function recentEntries(durable: DurableHarness, id: ConversationId, ctx: C
* conversation's own entries, so an ID from another conversation is not found.
*/
export async function replyTarget(durable: DurableHarness, id: ConversationId, entryId: number, ctx: Context) {
- const conversation = await durable.conversation(id, durableContext(ctx));
+ const conversation = await durable.conversation(id, ctx);
const only = { minEntryId: entryId as EntryId, maxEntryId: entryId as EntryId };
- const page = await conversation?.entries(only, 1, undefined, durableContext(ctx));
+ const page = await conversation?.entries(only, 1, undefined, ctx);
const entry = page?.items[0];
const quote: ReplyTo | undefined = entry?.id === entryId ? quoteOf(entry) : undefined;
if (!quote) throw invalid("args", "That message cannot be replied to");
return quote;
}
-/** What the cells show of the document, and how many messages wait for a mention. */
-const Shown = defineState("p2.chat-mode-setting", {
- schema: obj({ enabled: bool(), name: AgentNameSchema, waiting: num() }),
+/** How many chat messages a mention would pass on now: computed from the transcript, which the view only windows. */
+const Waiting = defineState("p2.chat-mode-setting.waiting", {
+ schema: obj({ count: num() }),
version: 1,
- initial: () => ({ enabled: false, name: DEFAULT_AGENT_NAME, waiting: 0 }),
+ initial: () => ({ count: 0 }),
});
export default defineFacet((host) => {
@@ -103,73 +97,47 @@ export default defineFacet((host) => {
({ conversation }, self) => {
const id = conversation as ConversationId;
const harness = harnessOf(self);
- /** Where the agent's chat mode is (the main agent's on conversation 1); set when the object opens. */
+ // The conversation's view (shared with its viewers): its entries and queue change what waits for a mention.
+ const chat = self.get(Conversation, { conversation });
+ /** Where the agent's chat mode is (the main agent's on conversation 1); found when the document is followed. */
let holder = id;
- const read = async (ctx: Context) =>
- (await harness().durable.snapshot(ChatModeDoc, holder, durableContext(ctx))) ?? {
- enabled: false,
- name: DEFAULT_AGENT_NAME,
- };
- const shown = self.state(Shown);
- const commits = self.open(
- async (ctx) => {
- const durable = harness().durable;
- holder = await agentConversation(durable, id, durableContext(ctx));
- const doc = await read(ctx);
- await shown.change((draft) => {
- draft.enabled = doc.enabled;
- draft.name = doc.name;
+ const settings = self.follow(async (observer, ctx) => {
+ const durable = harness().durable;
+ holder = await agentConversation(durable, id, ctx);
+ const observed = await durable.observeDoc(ChatModeDoc, holder, observer, ctx);
+ if (observed !== undefined) return observed;
+ // Nobody switched it yet: the document is absent, and an absent address cannot be followed. Create it as it
+ // reads (off, the default name).
+ await durable.commit(async (tx) => {
+ await tx.doc(ChatModeDoc, holder);
+ }, ctx);
+ return durable.observeDoc(ChatModeDoc, holder, observer, ctx);
+ });
+ const waiting = self.state(Waiting);
+ const counted = deriveTyped(self, [settings, chat.view], ([doc, view]) => ({
+ enabled: doc.enabled,
+ newest: view?.entries.at(-1)?.id ?? 0,
+ inbox: (view?.docs[InboxDoc.definition.kind] ?? null) as InboxState | null,
+ }));
+ // The view holds only a window of the entries: the router's backwards scan counts what waits.
+ self.on(counted, async ({ value }, ctx) => {
+ const count = value.enabled
+ ? unmentioned(await recentEntries(harness().durable, id, ctx), value.inbox ?? undefined).length
+ : 0;
+ if (waiting.value.count !== count)
+ await waiting.change((draft) => {
+ draft.count = count;
}, ctx);
- if (!(await durable.conversation(id, ctx))) throw new Error(`Unknown conversation ${conversation}`);
- // The waiting count also follows the entries and the queue.
- return new CommitSignal(durable, (publication) =>
- publication.changes.some(
- (change) =>
- (change.type === "entry" && change.value.conversationId === id) ||
- (change.type === "document" &&
- change.record.kind === ChatModeDoc.definition.kind &&
- change.conversationId === holder) ||
- (change.type === "document" &&
- change.record.kind === InboxDoc.definition.kind &&
- change.conversationId === id),
- ),
- );
- },
- (signal) => signal.close(),
- );
- // The document is the truth; the cells show it. Subscribed before reading, so no commit is missed. The view
- // holds only a window of the entries, so the router's backwards scan counts what waits, while chat mode is on.
- self.onLive(async (ctx) => {
- const signal = signalOf(ctx);
- while (!signal.aborted) {
- const seen = commits.value.version;
- const doc = await read(ctx);
- const durable = harness().durable;
- const count = doc.enabled
- ? unmentioned(
- await recentEntries(durable, id, ctx),
- await durable.snapshot(InboxDoc, id, durableContext(ctx)),
- ).length
- : 0;
- const current = shown.value;
- if (current.enabled !== doc.enabled || current.name !== doc.name || current.waiting !== count)
- await shown.change((draft) => {
- draft.enabled = doc.enabled;
- draft.name = doc.name;
- draft.waiting = count;
- }, ctx);
- await commits.value.next(seen, signal).catch(() => {});
- }
});
const commit = async (ctx: Ctx, change: (doc: { enabled: boolean; name: string }) => void) => {
- const durable = await harness().durable.conversation(id, durableContext(ctx));
+ const durable = await harness().durable.conversation(id, ctx);
if (!durable) throw invalid("args", `Unknown conversation ${conversation}`);
- await durable.commit(async (tx) => change(await tx.doc(ChatModeDoc, holder)), durableContext(ctx));
+ await durable.commit(async (tx) => change(await tx.doc(ChatModeDoc, holder)), ctx);
};
return {
- enabled: self.derive([shown], ([value]) => value.enabled),
- name: self.derive([shown], ([value]) => value.name),
- waiting: self.derive([shown], ([value]) => value.waiting),
+ enabled: self.derive([settings], ([doc]) => doc.enabled),
+ name: self.derive([settings], ([doc]) => doc.name),
+ waiting: self.derive([waiting], ([value]) => value.count),
set: async ({ enabled: on }, ctx) => {
await commit(ctx, (doc) => {
doc.enabled = on;
@@ -203,11 +171,7 @@ export default defineFacet((host) => {
"chat-mode": async ({ conversation, author, text, requestId, whenBusy, attachments, replyTo }, ctx) => {
const durable = harness().durable;
const id = conversation as ConversationId;
- const settings = await durable.snapshot(
- ChatModeDoc,
- await agentConversation(durable, id, durableContext(ctx)),
- durableContext(ctx),
- );
+ const settings = await durable.snapshot(ChatModeDoc, await agentConversation(durable, id, ctx), ctx);
if (!settings?.enabled) return undefined;
const chat = await ctx.get(Conversation, { conversation });
// The client names the entry; whom it quotes and what comes from the entry.
@@ -222,7 +186,7 @@ export default defineFacet((host) => {
};
return chat.write({ kind: CHAT_ENTRY, data, requestId, author }, ctx);
}
- const inbox = await durable.snapshot(InboxDoc, id, durableContext(ctx));
+ const inbox = await durable.snapshot(InboxDoc, id, ctx);
const earlier = unmentioned(await recentEntries(durable, id, ctx), inbox);
const files = await resolve(digestFiles(earlier, attachments), conversation, ctx);
const own = files.filter((file) => attachments.includes(file.id));
diff --git a/apps/p2/plugins/chat/contract.ts b/apps/p2/plugins/chat/contract.ts
index c2617b158..7af4435ee 100644
--- a/apps/p2/plugins/chat/contract.ts
+++ b/apps/p2/plugins/chat/contract.ts
@@ -83,25 +83,23 @@ const ConversationConfigureSchema = obj({
const RestoredSchema = arr(obj({ author: AuthorSchema, text: str() }));
-/**
- * The conversation's user shell command (`!cmd`/`!!cmd`) while it runs and until its `p2.shell` entry is placed:
- * the tail of its output (src/runtime/shell.ts limits). Null when none.
- */
-export const ShellStateSchema = union([
- obj({
- id: str(),
- command: str(),
- author: AuthorSchema,
- includeInContext: bool(),
- output: str(),
- truncated: bool(),
- running: bool(),
- exitCode: opt(num()),
- cancelled: bool(),
- error: opt(str()),
- }),
- nul(),
-]);
+/** A user shell command (`!cmd`/`!!cmd`) while it runs and until its `p2.shell` entry is placed. */
+export const LiveShellSchema = obj({
+ id: str(),
+ command: str(),
+ author: AuthorSchema,
+ includeInContext: bool(),
+ /** The tail of its output (src/runtime/shell.ts limits). */
+ output: str(),
+ truncated: bool(),
+ running: bool(),
+ exitCode: opt(num()),
+ cancelled: bool(),
+ error: opt(str()),
+});
+export type LiveShell = Static;
+/** The conversation's user shell command; null when none. */
+export const ShellStateSchema = union([LiveShellSchema, nul()]);
export type ShellState = Static;
/** pi's text when a second command is started while one runs (interactive-mode `isBashRunning`). */
export const SHELL_BUSY = "A bash command is already running. Press Esc to cancel it first.";
@@ -109,7 +107,7 @@ export const ShellSchema = obj({
command: str({ minLength: 1, maxLength: 100_000 }),
/** `!` (true): the output joins the agent's context; `!!` (false): display only. */
includeInContext: bool(),
- /** Client-chosen; a retried call with the same ID runs once (scoped to the author). */
+ /** Client-chosen, the command's `shellId` (`liveShell.id`); a retried call with the same ID runs once (per author). */
requestId: str({ minLength: 1, maxLength: 200 }),
});
@@ -165,7 +163,8 @@ export type Submit = Static;
* watch, and the operations that need an author, which callers pass explicitly.
*/
export const Conversation = local("p2.conversation", {
- version: 1,
+ /** 2: user shell commands moved to `ShellCommands` (chat/shell.ts). */
+ version: 2,
params: { conversation: param.int() },
members: {
/** The view's window, as every viewer holds it. */
@@ -202,12 +201,6 @@ export const Conversation = local("p2.conversation", {
abortCompaction: method(undefined, obj({ aborted: int({ minimum: 0 }) })),
/** Withdraws `authorId`'s queued inputs (by the server-stamped author) in submission order. */
dequeueOwn: method(obj({ authorId: str() }), obj({ restored: RestoredSchema, skipped: int({ minimum: 0 }) })),
- /** The running (or just finished, not yet placed) user shell command. */
- shell: ShellStateSchema,
- /** Runs a user shell command on the conversation's environment; one at a time per conversation. */
- runShell: method(obj({ ...propertiesOf(ShellSchema), author: AuthorSchema }), obj({ shellId: str() })),
- /** Cancels the running command `shellId`; false when it is not the running one. */
- abortShell: method(obj({ shellId: str() }), obj({ aborted: bool() })),
},
});
diff --git a/apps/p2/plugins/chat/session-start.ts b/apps/p2/plugins/chat/session-start.ts
index d911f59b5..4ad48ae7c 100644
--- a/apps/p2/plugins/chat/session-start.ts
+++ b/apps/p2/plugins/chat/session-start.ts
@@ -4,16 +4,21 @@
* exists, the messages move to its main conversation's outbox, which delivers them (uploads, template expansion,
* retries), and the page opens the session.
*
- * One start per page, kept for the page's lifetime: messages sent while the session is being created join it. A
- * creation the server refuses (a directory that does not exist, an unknown model) puts the messages back into the
- * composer; a failure of the connection is tried again, at most `GIVE_UP` times while connected. A creation whose answer
- * was lost with the connection is tried again too and may leave an empty session behind.
+ * One start per page, kept for the page's lifetime: messages sent while the session is being created join it. The page
+ * chooses the session's ID at the first Send (`Sessions.create` takes it), so a creation whose answer was lost with
+ * the connection is tried again with the same ID and answers the same session. A creation the server refuses (a
+ * directory that does not exist, an unknown model) puts the messages back into the composer; a failure of the
+ * connection is tried again, at most `GIVE_UP` times while connected.
*/
import type { ModelChoice } from "@p2/app/v1/web";
import { GIVE_UP, type Outgoing, outboxOf, retryDelay } from "./outbox.ts";
-/** Where and with which model the session starts (as chosen at the first Send), and the messages it starts with. */
+/**
+ * The session's ID, where and with which model it starts (as chosen at the first Send), and the messages it starts
+ * with.
+ */
export type Start = {
+ readonly session: string;
readonly location: { readonly env: string | null; readonly cwd: string };
readonly choice: ModelChoice;
readonly messages: readonly Outgoing[];
@@ -21,8 +26,8 @@ export type Start = {
/** How the connected new-session screen creates the session, and what it does with the outcome. */
export type Creator = {
- /** Creates the session; resolves with its ID. */
- create(start: Start): Promise;
+ /** Creates the session `start.session` (again, after a lost answer: the server answers the same row). */
+ create(start: Start): Promise;
/** Whether `error` refuses the creation for good; anything else is retried. */
refused(error: unknown): boolean;
/** The session exists and its main conversation's outbox has the messages. */
@@ -51,7 +56,7 @@ export class SessionStart {
};
snapshot = (): Start | undefined => this.#start;
- /** Adds `message`; the first one's `place` decides where and with which model the session starts. */
+ /** Adds `message`; the first one's `place` decides the session's ID, where and with which model it starts. */
add(message: Outgoing, place: Omit): void {
this.#set(
this.#start
@@ -83,7 +88,8 @@ export class SessionStart {
this.#running = true;
clearTimeout(this.#retry);
try {
- const session = await creator.create(start);
+ await creator.create(start);
+ const { session } = start;
this.#attempt = 0;
// Read after the creation: messages sent meanwhile go with the first.
const messages = this.#start?.messages ?? [];
diff --git a/apps/p2/plugins/chat/shell.ts b/apps/p2/plugins/chat/shell.ts
new file mode 100644
index 000000000..d751df267
--- /dev/null
+++ b/apps/p2/plugins/chat/shell.ts
@@ -0,0 +1,305 @@
+// User shell commands (`!cmd`/`!!cmd`) of the worker's conversations: each command is a job whose state shows its
+// output while it runs; the run places the command's `p2.shell` entry before it ends, so the live block gives way to
+// the entry without a gap. One command at a time per conversation; the session neither moves nor clones while one runs.
+
+import type { ConversationId, Harness as DurableHarness } from "@earendil-works/pi-durable";
+import { bool, int, map, nul, num, obj, opt, type Static, str, union } from "@earendil-works/pi-schema";
+import {
+ authorSender,
+ defineJob,
+ defineState,
+ defineTypedState,
+ deriveTyped,
+ type FacetHost,
+ type JobCtx,
+ type JsonValue,
+ literal,
+ local,
+ method,
+ type P2Principal,
+ SHELL_ENTRY,
+ SHELL_MAX_BYTES,
+ SHELL_MAX_LINES,
+ type ShellData,
+ senderText,
+ shellContextText,
+ shellTail,
+ until,
+ withoutAbortSignal,
+} from "@p2/app/v1";
+import { conversationEnvironment, type ExecutionEnvironments, harnessOf, workerOf } from "@p2/app/v1/worker";
+import { type LiveShell, LiveShellSchema, SHELL_BUSY } from "./contract.ts";
+
+const CommandSchema = obj({
+ conversation: int({ minimum: 1 }),
+ command: str({ minLength: 1, maxLength: 100_000 }),
+ /** `!` (true): the output joins the agent's context; `!!` (false): display only. */
+ includeInContext: bool(),
+ /** Closed, unlike the stored entry's author: job arguments are checked like method arguments. */
+ author: obj({ kind: union([literal("owner"), literal("guest")]), id: str(), name: str() }),
+ /** The caller's ID for the command, its `shellId`: a client shows the command under it before the state arrives. */
+ requestId: str({ minLength: 1, maxLength: 200 }),
+});
+type Command = Static;
+
+/**
+ * The worker's user shell commands (host-internal): each conversation's running command, and how to start and cancel
+ * one. Chat's views show their conversation's command and pass the caller as its author.
+ */
+export const ShellCommands = local("p2.shell-commands", {
+ version: 1,
+ params: {},
+ members: {
+ /** The running (or finished, its entry not placed yet) command of each conversation, by conversation. */
+ shells: map(LiveShellSchema),
+ /**
+ * Starts a command; refuses while its conversation runs one (`SHELL_BUSY`). Its `shellId` is the caller's
+ * `requestId`; a retried `requestId` of the same author starts nothing (among the last 100 starts).
+ */
+ start: method(CommandSchema, obj({ shellId: str() })),
+ /** Cancels the running command `shellId`; resolves once it shows as cancelled, false when it is not running. */
+ abort: method(obj({ shellId: str() }), obj({ aborted: bool() })),
+ },
+});
+
+/** How often the live output is published at most (the bash tool's sampling, 100 KiB/s). */
+const SAMPLE_MS = 200;
+
+const OutputSchema = obj({
+ output: str(),
+ truncated: bool(),
+ running: bool(),
+ cancelled: bool(),
+ exitCode: opt(num()),
+ /** Where the environment kept the complete output, when it spilled. */
+ fullOutputPath: opt(str()),
+ error: opt(str()),
+});
+type Output = Static;
+
+/** A command's live output: the run's state document. */
+const OutputDoc = defineTypedState("p2.shell-commands.output", {
+ schema: OutputSchema,
+ version: 1,
+ initial: () => ({ output: "", truncated: false, running: true, cancelled: false }),
+});
+
+/** 2: the command's `requestId` (its `shellId`). */
+const ShellJob = defineJob("p2.shell-commands.run", {
+ version: 2,
+ args: CommandSchema,
+ result: nul(),
+ state: OutputDoc,
+});
+
+type RunCtx = JobCtx;
+
+/** Request keys (`author:requestId`) of recent starts and the command each started, oldest first. */
+const Admitted = defineState("p2.shell-commands.admitted", {
+ schema: obj({ starts: map(str()) }),
+ version: 1,
+ initial: () => ({ starts: {} }),
+});
+
+const MAX_ADMITTED = 100;
+
+/** Serves `ShellCommands` (the chat worker facet). */
+export function serveShellCommands(host: FacetHost): void {
+ host.serve(ShellCommands, (_params, self) => {
+ const harness = harnessOf(self);
+ const worker = workerOf(self);
+ const admitted = self.state(Admitted);
+ const job = self.job(ShellJob, {
+ run: async (command, ctx) => {
+ try {
+ await runCommand(command, harness().environments, ctx);
+ await placeEntry(command, ctx, harness().durable);
+ return null;
+ } finally {
+ worker().liveness.add("activeCalls", -1);
+ }
+ },
+ });
+ const shells = deriveTyped(self, [job.running], ([runs]) => {
+ const byConversation: Record = {};
+ for (const run of runs)
+ if (run.phase !== "blocked") byConversation[run.args.conversation] = liveShell(run.args, run.state);
+ return byConversation;
+ });
+ const running = (shellId: string) =>
+ Object.values(shells.value).some((shell) => shell.id === shellId && shell.running);
+ return {
+ shells,
+ start: async (command, ctx) => {
+ // A running command is work the session's fence waits for (a move, a clone), as a call is: counted from
+ // before its start (the caller's own call still counts) until its run ends.
+ worker().liveness.add("activeCalls", 1);
+ let started = false;
+ try {
+ const result = await self.tx(async (tx) => {
+ const starts = (await tx.draft(admitted)).starts;
+ const key = `${command.author.id}:${command.requestId}`;
+ if (starts[key] !== undefined) return { started: false };
+ if (shells.value[command.conversation] !== undefined) throw new Error(SHELL_BUSY);
+ starts[key] = tx.start(job, command).id;
+ const keys = Object.keys(starts);
+ for (const old of keys.slice(0, Math.max(0, keys.length - MAX_ADMITTED))) delete starts[old];
+ return { started: true };
+ }, ctx);
+ started = result.started;
+ return { shellId: command.requestId };
+ } finally {
+ if (!started) worker().liveness.add("activeCalls", -1);
+ }
+ },
+ abort: async ({ shellId }, ctx) => {
+ if (!running(shellId)) return { aborted: false };
+ // `abort` resolves once the run ended, after placing its entry (at the next boundary while the agent works):
+ // the caller waits only until the command shows as cancelled.
+ const run = job.running.value.find((each) => each.phase !== "blocked" && each.args.requestId === shellId);
+ if (run) await Promise.race([job.abort(run.id, ctx), until(shells, () => !running(shellId), ctx)]);
+ return { aborted: true };
+ },
+ };
+ });
+}
+
+/** A run as Chat shows it. */
+function liveShell(command: Command, state: Output): LiveShell {
+ return {
+ id: command.requestId,
+ command: command.command,
+ author: command.author,
+ includeInContext: command.includeInContext,
+ output: state.output,
+ truncated: state.truncated,
+ running: state.running,
+ cancelled: state.cancelled,
+ ...(state.exitCode === undefined ? {} : { exitCode: state.exitCode }),
+ ...(state.error === undefined ? {} : { error: state.error }),
+ };
+}
+
+/**
+ * Runs the command where its conversation works (a subagent's own location may be on another machine than the
+ * session's), sampling its output into the run's state, which ends with how it ended.
+ */
+async function runCommand(command: Command, environments: ExecutionEnvironments, ctx: RunCtx): Promise {
+ let output = "";
+ let truncated = false;
+ let timer: ReturnType | undefined;
+ // State changes are commits: chained, so the last one is the last written, and awaited before the run moves on.
+ let writes: Promise = Promise.resolve();
+ const write = (change: (draft: Output) => void) => {
+ writes = writes.then(() => ctx.state.change(change, ctx));
+ };
+ const sample = () => {
+ timer = undefined;
+ const [text, cut] = [output, truncated];
+ write((draft) => {
+ draft.output = text;
+ draft.truncated = cut;
+ });
+ };
+ let ended: Pick
{message && }
- {text || (status === "live" ? "No log lines" : status === "gone" ? "Logs unavailable" : "Loading logs…")}
+ {text ||
+ (status === "live"
+ ? "No log lines"
+ : status === "failed" || status === "ended"
+ ? "Logs unavailable"
+ : "Loading logs…")}
);
diff --git a/apps/p2/plugins/mcp/config.ts b/apps/p2/plugins/mcp/config.ts
index 4a94c4a2f..57bf79b78 100644
--- a/apps/p2/plugins/mcp/config.ts
+++ b/apps/p2/plugins/mcp/config.ts
@@ -38,14 +38,8 @@ export type OwnerServer =
/** The owner's servers and why entries were skipped. */
export type OwnerConfig = { readonly servers: readonly OwnerServer[]; readonly errors: readonly string[] };
-/**
- * The file as last read: not yet, a valid configuration, or why it is not one (its readers keep the last valid servers,
- * also from before this reader existed: a reload's new reader may first see a save caught half-way).
- */
-type OwnerConfigRead =
- | { readonly kind: "unread" }
- | { readonly kind: "valid"; readonly config: OwnerConfig }
- | { readonly kind: "invalid"; readonly failure: string };
+/** The file as last read: the last valid servers and their errors, and why the last read failed (null: it did not). */
+export type OwnerConfigState = OwnerConfig & { readonly failure: string | null };
/**
* The owner's configuration in `text` (`{ "mcpServers": { name: entry } }`, comments allowed): invalid entries are
@@ -158,32 +152,40 @@ export function expandHostVariables(value: string): string {
export class OwnerConfigFile {
/** Watches the file and reads it at each change; `start()` reads it the first time, `close()` stops. */
readonly observer: Observer;
- #read: OwnerConfigRead = { kind: "unread" };
- readonly #listeners = new Set<(read: OwnerConfigRead) => void>();
+ #state: OwnerConfigState;
+ readonly #listeners = new Set<() => void>();
- constructor(dataDir: string) {
+ /**
+ * `last`: the servers last read validly before this reader existed (a restarted manager's: its new reader may first
+ * see a save caught half-way, and keeps them then).
+ */
+ constructor(dataDir: string, last: OwnerConfig) {
const path = join(dataDir, "mcp.json");
+ this.#state = { ...last, failure: null };
this.observer = new Observer(
[{ env: new NodeExecutionEnv({ cwd: dataDir }), targets: () => [{ path }] }],
async (retrying) => {
try {
- this.#read = { kind: "valid", config: parseOwnerConfig((await optionalText(path)) ?? "{}") };
+ this.#state = { ...parseOwnerConfig((await optionalText(path)) ?? "{}"), failure: null };
} catch (error) {
- // A save caught half-way is read once more before it counts.
+ // A save caught half-way is read once more before it counts; an invalid read keeps the last valid servers.
if (!retrying) return true;
- this.#read = { kind: "invalid", failure: error instanceof Error ? error.message : String(error) };
+ this.#state = { ...this.#state, failure: error instanceof Error ? error.message : String(error) };
}
- for (const listener of this.#listeners) listener(this.#read);
+ for (const listener of this.#listeners) listener();
return false;
},
2000,
);
}
- /** `listener` gets the last read at once (once there is one), then every read's. */
- subscribe(listener: (read: OwnerConfigRead) => void): () => void {
+ get state(): OwnerConfigState {
+ return this.#state;
+ }
+
+ /** `listener` hears of every read from now on; the result stops it. */
+ subscribe(listener: () => void): () => void {
this.#listeners.add(listener);
- if (this.#read.kind !== "unread") listener(this.#read);
return () => this.#listeners.delete(listener);
}
}
diff --git a/apps/p2/plugins/mcp/routes.ts b/apps/p2/plugins/mcp/routes.ts
index 463c47dc6..7ab0a2281 100644
--- a/apps/p2/plugins/mcp/routes.ts
+++ b/apps/p2/plugins/mcp/routes.ts
@@ -4,16 +4,15 @@
// p2 host; for any other owner a probe page finds out first (`LoopbackProbe`).
import type { IncomingMessage, ServerResponse } from "node:http";
-import { type Ctx, isOwner, type StoreSelf, until } from "@p2/app/v1";
-import { type RequestAccess, Routes, type ServerParts, serverOf } from "@p2/app/v1/server";
+import { type Ctx, isOwner, type StoreSelf } from "@p2/app/v1";
+import { type RequestAccess, Routes } from "@p2/app/v1/server";
import { McpManager } from "./contract.ts";
import { resultPage, sendPage, sendProbePage } from "./pages.ts";
/** Contributes the sign-in routes. */
export function signInRoutes(self: StoreSelf): void {
- const server = serverOf(self);
self.contribute(Routes, {
- "GET /mcp/sign-in/*": (request, response, access, ctx) => signIn(request, response, access, server, ctx),
+ "GET /mcp/sign-in/*": (request, response, access, ctx) => signIn(request, response, access, ctx),
"GET /mcp/callback": (request, response, access, ctx) => callback(request, response, access, ctx),
});
}
@@ -29,7 +28,6 @@ async function signIn(
request: IncomingMessage,
response: ServerResponse,
{ principal, origin }: RequestAccess,
- server: () => ServerParts,
ctx: Ctx,
): Promise {
if (!isOwner(principal)) return refuse(response, principal === undefined);
@@ -64,28 +62,10 @@ async function signIn(
}
if (start.kind === "authorized")
return sendPage(response, 200, resultPage(`Signed in to ${name}`, "You may close this page."));
- keepManager(name, server);
response.writeHead(302, { Location: start.url });
response.end();
}
-/**
- * Keeps the manager, which holds the flow, while the sign-in to `name` waits for its browser step (until it finishes,
- * fails or expires), also when no view holds it meanwhile.
- */
-function keepManager(name: string, server: () => ServerParts): void {
- void server()
- .root(async (ctx) => {
- const manager = await ctx.get(McpManager, {});
- await until(
- manager.signIns,
- (signIns) => !signIns.some((entry) => entry.server === name && entry.status === "waiting"),
- ctx,
- );
- })
- .catch(() => {});
-}
-
/** Completes a route sign-in with the authorization response the browser brings back. */
async function callback(
request: IncomingMessage,
diff --git a/apps/p2/plugins/mcp/server.ts b/apps/p2/plugins/mcp/server.ts
index deb156d36..378ec7d36 100644
--- a/apps/p2/plugins/mcp/server.ts
+++ b/apps/p2/plugins/mcp/server.ts
@@ -6,7 +6,7 @@
import { readFile } from "node:fs/promises";
import { join } from "node:path";
-import { arr, int, nul, obj, str, union, unknown } from "@earendil-works/pi-schema";
+import { arr, int, obj, str, union, unknown } from "@earendil-works/pi-schema";
import {
type Ctx,
contribute,
@@ -15,6 +15,7 @@ import {
deriveTyped,
type Facet,
type Factory,
+ followNative,
forbidden,
type HandleOf,
isOwner,
@@ -23,7 +24,6 @@ import {
LocationSchema,
literal,
locationText,
- type Native,
nativeOf,
PluginError,
type Ref,
@@ -32,7 +32,6 @@ import {
signalOf,
unsafe,
until,
- untilAborted,
type Watchable,
type WorkerPrincipal,
} from "@p2/app/v1";
@@ -41,8 +40,8 @@ import type { EnvConfig } from "../envs/config.ts";
import { EnvManagers, EnvService } from "../envs/contract.ts";
import { Catalog, type Session } from "../sessions/contract.ts";
import { type Child, ChildStore } from "../subagents/contract.ts";
-import { type OwnerConfig, OwnerConfigFile, type OwnerServer } from "./config.ts";
-import { InstanceSchema, Mcp, McpManager, type ProjectFile, SignInSchema } from "./contract.ts";
+import { type OwnerConfig, OwnerConfigFile, type OwnerConfigState, type OwnerServer } from "./config.ts";
+import { type Instance, InstanceSchema, Mcp, McpManager, type ProjectFile } from "./contract.ts";
import { McpCredentials } from "./credentials.ts";
import type { ConnectEnv } from "./environments.ts";
import { type ClientInfo, Instances, RECONNECT } from "./instances.ts";
@@ -88,16 +87,6 @@ export function mcpServerFacet(viewLingerMs: number, connector: EnvConnector): F
});
}
-/**
- * The owner's configuration as last validly read, and why the last read failed (null: it did not). In memory, kept
- * across a reload: a new reader seeing a save caught half-way keeps these servers.
- */
-const Configured = defineTypedState("p2.mcp.config", {
- schema: obj({ servers: unsafe(unknown()), errors: arr(str()), failure: union([str(), nul()]) }),
- version: 1,
- initial: () => ({ servers: [], errors: [], failure: null }),
-});
-
/** The leases of session workers; kept across a reload of the plugin, so the new supervisor starts what they use. */
const Leased = defineTypedState("p2.mcp.leases", {
schema: obj({
@@ -120,46 +109,41 @@ const Leased = defineTypedState("p2.mcp.leases", {
});
/**
- * The supervisor's instances as it last listed them (a reload's new supervisor starts with their tools), and the
- * sign-ins under way.
+ * What a restarted manager's natives start from (in memory, kept across a reload's restart; never shown): the servers
+ * last read validly (a new reader may first see a save caught half-way) and the instances as last listed (their tools
+ * stay listed until their servers answer again).
*/
-const Listed = defineTypedState("p2.mcp.instances", {
- schema: obj({ instances: arr(InstanceSchema), signIns: arr(SignInSchema) }),
- version: 2,
- initial: () => ({ instances: [], signIns: [] }),
+const Memento = defineTypedState("p2.mcp.memento", {
+ schema: obj({ servers: unsafe(unknown()), errors: arr(str()), instances: arr(InstanceSchema) }),
+ version: 1,
+ initial: () => ({ servers: [], errors: [], instances: [] }),
});
-/** The manager: its documents, the file and the supervisor as natives, and the routines that keep them in step. */
+/**
+ * The manager: the file, the supervisor and the sign-ins as natives it follows, the leases as its document, and the
+ * routine that starts and closes instances as they change.
+ */
function managerFactory(connector: EnvConnector): Factory {
return (_params, self) => {
const server = self.get(Server, {});
const envs = self.get(EnvService, {});
- const work = workEnvironments(self.get(Catalog, {}), self.get(ChildStore, {}), self);
- const configured = self.state(Configured);
+ const catalog = self.get(Catalog, {});
+ const store = self.get(ChildStore, {});
+ const work = deriveTyped(self, [catalog.sessions, store.children], ([sessions, children]) =>
+ workEnvironments(sessions, children),
+ );
const leased = self.state(Leased);
- const listed = self.state(Listed);
+ const memento = self.state(Memento);
const placed = placements();
- const placement = () => placed(configured.value, leased.value.leases, envs.configs.value, work.value);
const connect = connector(self);
const file = self.open(
async () => {
- const opened = new OwnerConfigFile(server.info.value.dataDir);
+ const opened = new OwnerConfigFile(server.info.value.dataDir, memento.value);
await opened.observer.start();
return opened;
},
(opened) => opened.observer.close(),
);
- const instances = self.open(
- async () => {
- // After a reload the leases are still here: the routine below starts what they use, with the tools listed
- // before. Not here: natives open before derives compute.
- const info = await clientInfo(server.info.value.app);
- return new Instances(info, RECONNECT, listed.value.instances, connect, (oauth) =>
- signIns.value.authProvider(oauth),
- );
- },
- (opened) => opened.close(),
- );
const signIns = self.open(
async () =>
new SignIns(
@@ -170,11 +154,42 @@ function managerFactory(connector: EnvConnector): Factory {
),
(opened) => opened.close(),
);
+ const instances = self.open(
+ async () => {
+ const info = await clientInfo(server.info.value.app);
+ const opened = new Instances(info, RECONNECT, memento.value.instances, connect, (oauth) =>
+ signIns.value.authProvider(oauth),
+ );
+ // What the leases use now starts at once (after a restart with the tools listed before), so the supervisor's
+ // first list already shows it; derives compute only after the natives opened, so placed from the sources.
+ const at = workEnvironments(catalog.sessions.value, store.children.value);
+ opened.reconcile(placed(file.value.state, leased.value.leases, envs.configs.value, at));
+ return opened;
+ },
+ (opened) => opened.close(),
+ );
const probe = self.open(
async () => new LoopbackProbe(),
(opened) => opened.close(),
);
- self.onLive((ctx) => followNatives(file, instances, signIns, configured, listed, ctx));
+ // The file, the supervisor and the sign-ins as they are: every method's change is made before it returns.
+ const configured = followNative(
+ self,
+ (changed) => file.value.subscribe(changed),
+ () => file.value.state,
+ );
+ const listed = followNative(
+ self,
+ (changed) => {
+ const stops = [instances.value.subscribe(changed), signIns.value.subscribe(changed)];
+ return () => {
+ for (const stop of stops) stop();
+ };
+ },
+ () => ({ instances: instances.value.list(), signIns: signIns.value.list() }),
+ );
+ keepMemento(self, memento, configured, listed);
+ const placement = () => placed(configured.value, leased.value.leases, envs.configs.value, work.value);
// The current configuration, leases, environments and places of work, then every change of them, start and close
// instances; methods that change leases reconcile at once too, so a call right after `setLocations` finds its
// instance.
@@ -209,7 +224,6 @@ function managerFactory(connector: EnvConnector): Factory {
...managerMethods(leased, () => instances.value, placement),
...signInMethods(
configured,
- listed,
() => signIns.value,
() => probe.value,
),
@@ -217,14 +231,36 @@ function managerFactory(connector: EnvConnector): Factory {
};
}
+/**
+ * Keeps the memento a restarted manager's natives start from: the servers of each valid read, and the instances once
+ * one of them lists other tools (instances closing change no tools, so a manager ending writes nothing).
+ */
+function keepMemento(
+ self: Self,
+ memento: HandleOf,
+ configured: Watchable,
+ listed: Watchable<{ readonly instances: readonly Instance[] }>,
+): void {
+ self.on(configured, async ({ value: { servers, errors, failure } }, ctx) => {
+ if (failure === null) await memento.change((draft) => Object.assign(draft, { servers, errors }), ctx);
+ });
+ self.on(listed, async ({ value: { instances } }, ctx) => {
+ const remembered = new Map(memento.value.instances.map((instance) => [instance.id, toolNames(instance)]));
+ if (instances.some((instance) => remembered.get(instance.id) !== toolNames(instance)))
+ await memento.change((draft) => Object.assign(draft, { instances }), ctx);
+ });
+}
+
+/** The names of the tools `instance` lists, as one string. */
+const toolNames = (instance: Instance): string => instance.tools.map((tool) => tool.name).join("\n");
+
/** The environments each session's conversations work on, as the catalog and the subagent store record them. */
function workEnvironments(
- catalog: Ref,
- store: Ref,
- self: Self,
-): Watchable {
- return deriveTyped(self, [catalog.sessions, store.children], ([sessions, children]) =>
- Object.fromEntries(sessions.map((session) => [session.id, sessionEnvironments(session, children[session.id])])),
+ sessions: readonly Session[],
+ children: Readonly>,
+): WorkEnvironments {
+ return Object.fromEntries(
+ sessions.map((session) => [session.id, sessionEnvironments(session, children[session.id])]),
);
}
@@ -254,52 +290,6 @@ function placements(): (
};
}
-/** Copies what the file reads, the supervisor lists and the sign-ins are into the manager's documents until `ctx` ends. */
-async function followNatives(
- file: Native,
- instances: Native,
- signIns: Native,
- configured: HandleOf,
- listed: HandleOf,
- ctx: Ctx,
-): Promise {
- // Changes heard while this routine stops would change state with an ended ctx.
- const show = (): void => {
- if (signalOf(ctx).aborted) return;
- void listed
- .change((draft) => {
- draft.instances = instances.value.list();
- draft.signIns = signIns.value.list();
- }, ctx)
- .catch(() => {});
- };
- const stops = [
- file.value.subscribe((read) => {
- if (signalOf(ctx).aborted) return;
- void configured
- .change((draft) => {
- // An invalid read keeps the servers last read validly.
- if (read.kind === "invalid") draft.failure = read.failure;
- else if (read.kind === "valid")
- Object.assign(draft, {
- servers: [...read.config.servers],
- errors: [...read.config.errors],
- failure: null,
- });
- }, ctx)
- .catch(() => {});
- }),
- instances.value.subscribe(show),
- signIns.value.subscribe(show),
- ];
- show();
- try {
- await untilAborted(ctx);
- } finally {
- for (const stop of stops) stop();
- }
-}
-
/** The manager's methods: the lease rules applied on the leases' line, calls through the leased instance, restarts. */
function managerMethods(leased: HandleOf, instances: () => Instances, placement: () => Placement) {
return {
@@ -341,12 +331,7 @@ function managerMethods(leased: HandleOf, instances: () => Instan
}
/** The manager's sign-in methods: flows of the owner's OAuth servers, the callback route's completion, the probe. */
-function signInMethods(
- configured: HandleOf,
- listed: HandleOf,
- signIns: () => SignIns,
- probe: () => LoopbackProbe,
-) {
+function signInMethods(configured: Watchable, signIns: () => SignIns, probe: () => LoopbackProbe) {
const oauthServer = (name: string): OAuthServer => {
const server = configured.value.servers.find((candidate) => candidate.name === name);
const oauth = server === undefined ? undefined : oauthServerOf(server);
@@ -354,14 +339,9 @@ function signInMethods(
return oauth;
};
return {
- beginSignIn: async ({ server, redirect }: { server: string; redirect: SignInRedirect }, ctx: Ctx) => {
- const start = await asPluginError(signIns().begin(oauthServer(server), redirect));
- // Listed before the answer: a caller following `signIns` for this flow's end sees it under way first.
- await listed.change((draft) => {
- draft.signIns = signIns().list();
- }, ctx);
- return start;
- },
+ // The flow is listed before the answer: a caller following `signIns` for its end sees it under way first.
+ beginSignIn: ({ server, redirect }: { server: string; redirect: SignInRedirect }) =>
+ asPluginError(signIns().begin(oauthServer(server), redirect)),
completeSignIn: ({ state, code, iss, error }: { state: string; code?: string; iss?: string; error?: string }) =>
signIns().complete(
error !== undefined || code === undefined
diff --git a/apps/p2/plugins/mcp/tui.ts b/apps/p2/plugins/mcp/tui.ts
index 95d665a03..b4fe6389a 100644
--- a/apps/p2/plugins/mcp/tui.ts
+++ b/apps/p2/plugins/mcp/tui.ts
@@ -7,7 +7,6 @@ import {
BACKGROUND_CONTEXT,
contribute,
defineFacet,
- follow,
type Held,
type HeldIn,
liveIn,
@@ -39,48 +38,51 @@ const messageOf = (error: unknown): string => (error instanceof Error ? error.me
/** `/mcp [paste ]`: the overlay, or a pasted redirect URL for the sign-in waiting for it. */
function openMcp(ctx: TuiContext, args: string | undefined): void {
- const ref = ctx.host.hold(Mcp, {}, BACKGROUND_CONTEXT);
const words = args?.trim().split(/\s+/).filter(Boolean) ?? [];
if (words[0] === "paste") {
+ const ref = ctx.host.hold(Mcp, {}, BACKGROUND_CONTEXT);
void paste(ctx, ref, words[1]).finally(() => ref.release());
return;
}
- let instances: Instance[] = [];
+ const scope = ctx.scope();
+ let received: Instance[] = [];
let signIns: SignIn[] = [];
let status = "";
- let closed = false;
- const stops: (() => void)[] = [];
- void whenLive(ref)
- .then(() => {
- const view = liveIn(ref, "owner");
- if (closed || view === undefined) return;
- stops.push(
- follow(view.instances, (value) => {
- instances = value;
- ctx.requestRender();
- }),
- follow(view.signIns, (value) => {
- signIns = value;
- ctx.requestRender();
- }),
- );
- })
- .catch(() => {});
- const run = (label: string, action: (view: OwnerView) => Promise) => {
- const view = liveIn(ref, "owner");
+ const held = scope.watch(Mcp, {}, (mcp, follow) => {
+ const view = liveIn(mcp, "owner");
+ if (!view) return;
+ follow(view.instances, (value) => {
+ received = value;
+ });
+ follow(view.signIns, (value) => {
+ signIns = value;
+ });
+ });
+ // A restart shows the server starting and a sign-out signed out at once; the status line follows the call.
+ type StatusGuess = { instance: string; status: Instance["status"] };
+ const [instances, changing] = scope.optimistic(
+ () => received,
+ (list, guess: StatusGuess) =>
+ list.map((instance) => (instance.id === guess.instance ? { ...instance, status: guess.status } : instance)),
+ );
+ const run = (label: string, action: (view: OwnerView) => Promise, guess?: StatusGuess) => {
+ const view = liveIn(held, "owner");
if (view === undefined) return;
status = `${label}…`;
- ctx.requestRender();
- void action(view)
- .then(() => {
+ const call = (async () => {
+ try {
+ await action(view);
status = `${label}: done`;
- })
- .catch((error: unknown) => {
+ } catch (error) {
status = `${label}: ${messageOf(error)}`;
- })
- .finally(() => ctx.requestRender());
+ throw error;
+ }
+ })();
+ // The status line shows how it ended, a failure included.
+ void (guess ? changing(guess, call) : call).catch(() => ctx.requestRender());
+ ctx.requestRender();
};
- const byRow = (row: ListRow | undefined) => instances.find((instance) => instance.id === row?.key);
+ const byRow = (row: ListRow | undefined) => instances().find((instance) => instance.id === row?.key);
const signIn = (view: OwnerView, server: string) =>
view.signIn({ server }, BACKGROUND_CONTEXT).then(async (start) => {
if (start.kind === "authorized") return;
@@ -92,9 +94,9 @@ function openMcp(ctx: TuiContext, args: string | undefined): void {
const overlay = ctx.overlay(
"mcp",
new ListOverlay(ctx, {
- title: () => `MCP servers (${instances.length})`,
+ title: () => `MCP servers (${instances().length})`,
rows: () =>
- instances.map((instance) => {
+ instances().map((instance) => {
const flow = signIns.find((entry) => entry.server === instance.server);
const place = [
instance.location ? locationText(instance.location) : "",
@@ -127,8 +129,10 @@ function openMcp(ctx: TuiContext, args: string | undefined): void {
run: (row) => {
const instance = byRow(row);
if (instance)
- run(`Restarting ${instance.server}`, (view) =>
- view.restart({ instance: instance.id }, BACKGROUND_CONTEXT),
+ run(
+ `Restarting ${instance.server}`,
+ (view) => view.restart({ instance: instance.id }, BACKGROUND_CONTEXT),
+ { instance: instance.id, status: "starting" },
);
},
},
@@ -147,17 +151,17 @@ function openMcp(ctx: TuiContext, args: string | undefined): void {
run: (row) => {
const instance = byRow(row);
if (instance?.auth === "oauth")
- run(`Signing out of ${instance.server}`, (view) =>
- view.signOut({ server: instance.server }, BACKGROUND_CONTEXT),
+ run(
+ `Signing out of ${instance.server}`,
+ (view) => view.signOut({ server: instance.server }, BACKGROUND_CONTEXT),
+ { instance: instance.id, status: "sign-in" },
);
},
},
],
footer: () => (status ? [ctx.theme.current.fg("muted", ` ${lineText(status)}`)] : []),
onClose: () => {
- closed = true;
- for (const stop of stops) stop();
- ref.release();
+ scope.dispose();
overlay.close();
},
}),
diff --git a/apps/p2/plugins/mcp/web.tsx b/apps/p2/plugins/mcp/web.tsx
index 49cc59d51..fdf27d698 100644
--- a/apps/p2/plugins/mcp/web.tsx
+++ b/apps/p2/plugins/mcp/web.tsx
@@ -6,11 +6,13 @@ import { BACKGROUND_CONTEXT, contribute, defineFacet, type HeldIn, locationText
import {
ChevronIcon,
closeLayers,
+ errorText,
openLayer,
Sheet,
Slots,
useComposerState,
useObject,
+ useOptimisticCall,
useOwner,
useReadable,
useSheet,
@@ -20,7 +22,29 @@ import { type Instance, Mcp, type SignIn, signInLink } from "./contract.ts";
type OwnerView = HeldIn;
-const errorText = (reason: unknown): string => (reason instanceof Error ? reason.message : String(reason));
+/** A change of the instances while its call is in flight: a restart starts it, a sign-out leaves it signed out. */
+type InstanceGuess =
+ | { readonly kind: "restart"; readonly instance: string }
+ | { readonly kind: "signOut"; readonly server: string };
+
+function applyInstance(instances: readonly Instance[], guess: InstanceGuess): readonly Instance[] {
+ return instances.map((instance) =>
+ guess.kind === "restart"
+ ? instance.id === guess.instance
+ ? { ...instance, status: "starting", error: undefined }
+ : instance
+ : instance.server === guess.server && instance.auth === "oauth"
+ ? { ...instance, status: "sign-in" }
+ : instance,
+ );
+}
+
+/** What a server's entry does; each shows its effect at once and rolls back with the error in the toast. */
+type Actions = {
+ readonly restart: (instance: string) => void;
+ readonly signOut: (server: string) => void;
+ readonly cancelSignIn: (id: string) => void;
+};
const STATUS: Record = {
starting: "Starting",
@@ -32,10 +56,22 @@ const STATUS: Record = {
/** The sheet: configuration errors, then each server with its instances and sign-in. */
function McpSheet({ onClose }: { onClose: () => void }) {
const view = useObject(Mcp, {}, { variant: "owner" });
- const instances = useReadable(view?.instances) ?? [];
const errors = useReadable(view?.errors) ?? [];
- const signIns = useReadable(view?.signIns) ?? [];
+ const { fail } = useComposerState();
+ const [instances, changeInstance] = useOptimisticCall(useReadable(view?.instances) ?? [], applyInstance, fail);
+ const [signIns, cancel] = useOptimisticCall(
+ useReadable(view?.signIns) ?? [],
+ (list, id: string) => list.filter((entry) => entry.id !== id),
+ fail,
+ );
const servers = [...new Set(instances.map((instance) => instance.server))];
+ const actions: Actions | undefined = view && {
+ restart: (instance) =>
+ changeInstance({ kind: "restart", instance }, () => view.restart({ instance }, BACKGROUND_CONTEXT)),
+ signOut: (server) =>
+ changeInstance({ kind: "signOut", server }, () => view.signOut({ server }, BACKGROUND_CONTEXT)),
+ cancelSignIn: (id) => cancel(id, () => view.cancelSignIn({ id }, BACKGROUND_CONTEXT)),
+ };
return (
@@ -48,12 +84,13 @@ function McpSheet({ onClose }: { onClose: () => void }) {
))}
{view && servers.length === 0 && No MCP server is running.
}
- {view && (
+ {view && actions && (
{servers.map((server) => (
instance.server === server)}
signIn={signIns.find((entry) => entry.server === server)}
@@ -68,17 +105,17 @@ function McpSheet({ onClose }: { onClose: () => void }) {
/** One server: its instances (status, place, tools, error, restart), and signing in or out if it uses OAuth. */
function ServerEntry({
view,
+ actions,
server,
instances,
signIn,
}: {
view: OwnerView;
+ actions: Actions;
server: string;
instances: readonly Instance[];
signIn: SignIn | undefined;
}) {
- const { setError } = useComposerState();
- const perform = (action: Promise) => void action.catch((reason: unknown) => setError(errorText(reason)));
const oauth = instances.some((instance) => instance.auth === "oauth");
const needsSignIn = instances.some((instance) => instance.status === "sign-in");
const connected = instances.some((instance) => instance.status === "connected");
@@ -100,25 +137,19 @@ function ServerEntry({
{instance.error && instance.status !== "sign-in" && {instance.error}
}
- perform(view.restart({ instance: instance.id }, BACKGROUND_CONTEXT))}
- >
+ actions.restart(instance.id)}>
Restart
))}
- {oauth && signIn?.status === "waiting" && }
+ {oauth && signIn?.status === "waiting" && (
+ actions.cancelSignIn(signIn.id)} />
+ )}
{oauth && signIn?.status === "failed" && (
{signIn.error ?? "The sign-in failed."}
-
perform(view.cancelSignIn({ id: signIn.id }, BACKGROUND_CONTEXT))}
- >
+ actions.cancelSignIn(signIn.id)}>
Dismiss
@@ -132,11 +163,7 @@ function ServerEntry({
) : null}
{connected && (
- perform(view.signOut({ server }, BACKGROUND_CONTEXT))}
- >
+ actions.signOut(server)}>
Sign out
)}
@@ -147,7 +174,7 @@ function ServerEntry({
}
/** A sign-in waiting for the browser: paste the final redirect URL, or cancel. */
-function PendingSignIn({ view, signIn }: { view: OwnerView; signIn: SignIn }) {
+function PendingSignIn({ view, signIn, onCancel }: { view: OwnerView; signIn: SignIn; onCancel: () => void }) {
const [value, setValue] = useState("");
const [sending, setSending] = useState(false);
const [error, setError] = useState();
@@ -184,11 +211,7 @@ function PendingSignIn({ view, signIn }: { view: OwnerView; signIn: SignIn }) {
Continue
- void view.cancelSignIn({ id: signIn.id }, BACKGROUND_CONTEXT).catch(() => {})}
- >
+
Cancel
diff --git a/apps/p2/plugins/mcp/worker.ts b/apps/p2/plugins/mcp/worker.ts
index b266d09bf..d7af81e5d 100644
--- a/apps/p2/plugins/mcp/worker.ts
+++ b/apps/p2/plugins/mcp/worker.ts
@@ -64,10 +64,7 @@ function mcpWorker(self: StoreSelf): void {
const view = self.get(Mcp, {});
const call: McpCall = async ({ conversationId, read, ...request }, context) => {
const location = await harness().environments.locate(conversationId, context, read);
- return worker().root(
- (ctx) => expectVariant(view, "worker").call({ location, ...request }, ctx),
- context.abortSignal,
- );
+ return worker().root((ctx) => expectVariant(view, "worker").call({ location, ...request }, ctx), context);
};
// Code mode switched off or not loaded: the tools would be unreachable, so `tool_search` loads them, as pi falls back.
const reach = self.derive(
diff --git a/apps/p2/plugins/models/server.ts b/apps/p2/plugins/models/server.ts
index 7c4d69271..84bc3ee57 100644
--- a/apps/p2/plugins/models/server.ts
+++ b/apps/p2/plugins/models/server.ts
@@ -1,36 +1,14 @@
-import { obj, unknown } from "@earendil-works/pi-schema";
-import {
- defineFacet,
- defineState,
- forbidden,
- isOwner,
- type JsonValue,
- unsafe,
- untilAborted,
- type Watchable,
-} from "@p2/app/v1";
+import { defineFacet, deriveTyped, followNative, forbidden, isOwner } from "@p2/app/v1";
import { Server, serverOf } from "@p2/app/v1/server";
import { openResources, resourceState } from "./config.ts";
-import { ModelConfig, ModelResources, ModelSummarySchema, Models } from "./contract.ts";
-import type { ResourceSnapshot } from "./resources.ts";
+import { ModelConfig, ModelResources, Models } from "./contract.ts";
import { resolveModelScope } from "./scope.ts";
-/** What `ModelConfig` publishes of the configuration: the picker's summary and the secret-free snapshot. */
-const Published = defineState("p2.model-config", {
- schema: obj({ summary: ModelSummarySchema, snapshot: unsafe(unknown()) }),
- version: 1,
- initial: () => ({
- summary: { models: [], defaultModel: null, errors: [] },
- snapshot: { version: 0, settings: {}, credentials: [], providers: [], models: [], errors: [] },
- }),
-});
-
export default defineFacet((host) => {
host.serve(ModelResources, { compose: (table) => table[0]?.value });
host.serve(ModelConfig, (_params, self) => {
const server = self.get(Server, {});
const natives = serverOf(self);
- const published = self.state(Published);
const resources = self.open(
() => openResources(server.info.value.dataDir, server.info.value.fake, natives().resourceOptions),
(opened) => opened.close(),
@@ -40,25 +18,16 @@ export default defineFacet((host) => {
return resources.value;
},
});
- self.onLive(async (ctx) => {
- const publish = () =>
- published.change((draft) => {
- const state = resourceState(resources.value, server.info.value.fake);
- draft.summary = state.summary;
- draft.snapshot = state.snapshot as unknown as JsonValue;
- }, ctx);
- await publish();
- const stop = resources.value.subscribe(() => void publish().catch(() => {}));
- try {
- await untilAborted(ctx);
- } finally {
- stop();
- }
- });
+ // The configuration files' state as the resources hold it: a save below resolves after its reload, which
+ // told this source, so the caller's answer comes with the new state.
+ const state = followNative(
+ self,
+ (changed) => resources.value.subscribe(changed),
+ () => resourceState(resources.value, server.info.value.fake),
+ );
return {
- summary: self.derive([published], ([value]) => value.summary),
- // The snapshot is JSON (`Resources.snapshot()`); its type is the one the contract names.
- snapshot: self.derive([published], ([value]) => value.snapshot) as unknown as Watchable,
+ summary: self.derive([state], ([value]) => value.summary),
+ snapshot: deriveTyped(self, [state], ([value]) => value.snapshot),
setDefault: async ({ provider, modelId, thinkingLevel }) => {
const opened = resources.value;
if (!opened.models.getModel(provider, modelId)) throw new Error("Unknown model");
diff --git a/apps/p2/plugins/models/tui.ts b/apps/p2/plugins/models/tui.ts
index d3fbc83e4..08875aaca 100644
--- a/apps/p2/plugins/models/tui.ts
+++ b/apps/p2/plugins/models/tui.ts
@@ -10,6 +10,7 @@ import type { TuiContext } from "@p2/app/v1/tui";
import {
lineText,
ModelSelectorComponent,
+ type OptimisticAdd,
ScopedModelsSelectorComponent,
type SelectorModel,
ThinkingSelectorComponent,
@@ -25,6 +26,22 @@ import { type ModelChoice, type ModelInfo, Models } from "./contract.ts";
type Scoped = { provider: string; modelId: string; thinkingLevel?: string };
+/** A choice in flight: the conversation's model and level, the default, or the scope. */
+type Guess = {
+ readonly model?: ModelMeta | null;
+ readonly thinkingLevel?: string;
+ readonly defaultModel?: ModelChoice | null;
+ readonly scoped?: Scoped[];
+};
+
+/** What the commands read: the state with the choices in flight applied. */
+type Shown = {
+ readonly model: ModelMeta | null;
+ readonly thinkingLevel: string | undefined;
+ readonly defaultModel: ModelChoice | null;
+ readonly scoped: Scoped[];
+};
+
/**
* What the commands need, held per runtime (a non-screen slot): the owner's catalog, scope and default, and the viewed
* conversation's model and thinking level.
@@ -35,18 +52,22 @@ export class ModelState implements TuiSurface {
private owner: TuiScope | undefined;
private chat: TuiScope | undefined;
private identity: string | undefined;
- private models: () => Held | undefined = () => undefined;
- private chatRef: () => Held | undefined = () => undefined;
+ private models: Held | undefined;
+ private chatRef: Held | undefined;
+ /** The state as shown: the choices in flight over what the server last sent. */
+ private readonly shownState: () => Shown;
+ /** Shows a choice while its call is in flight: the selectors and cycling start from it (Ctrl+P twice moves twice). */
+ readonly choosing: OptimisticAdd;
catalog: ModelInfo[] = [];
- scoped: Scoped[] = [];
- defaultModel: ModelChoice | null = null;
- /** The viewed conversation's model (guest-safe metadata) and thinking level. */
- model: ModelMeta | null = null;
- thinkingLevel: string | undefined;
+ private received: Shown = { model: null, thinkingLevel: undefined, defaultModel: null, scoped: [] };
private stop: () => void;
constructor(runtime: TuiRuntime) {
this.runtime = runtime;
this.scope = new TuiScope(runtime);
+ [this.shownState, this.choosing] = this.scope.optimistic(
+ () => this.received,
+ (shown, guess: Guess) => ({ ...shown, ...guess }),
+ );
const changed = (): void => {
// The catalog is owner-only: guests never request it.
if (runtime.role === "owner" && !this.owner) {
@@ -56,10 +77,10 @@ export class ModelState implements TuiSurface {
this.catalog = models ?? [];
});
follow(ref.scopedModels, (scoped) => {
- this.scoped = scoped ?? [];
+ this.received = { ...this.received, scoped: scoped ?? [] };
});
follow(ref.defaultModel, (value) => {
- this.defaultModel = value ?? null;
+ this.received = { ...this.received, defaultModel: value ?? null };
});
});
}
@@ -67,19 +88,18 @@ export class ModelState implements TuiSurface {
this.identity = runtime.identity;
this.chat?.dispose();
this.chat = undefined;
- this.chatRef = () => undefined;
- this.model = null;
- this.thinkingLevel = undefined;
+ this.chatRef = undefined;
+ this.received = { ...this.received, model: null, thinkingLevel: undefined };
const session = runtime.session;
if (!session) return;
this.chat = new TuiScope(runtime);
this.chatRef = this.chat.watch(Chat, { session, conversation: runtime.conversation }, (ref, follow) => {
follow(ref.model, (model) => {
- this.model = model ?? null;
+ this.received = { ...this.received, model: model ?? null };
});
follow(ref.view, (view) => {
const agent = view?.docs["pi.agent"] as { thinkingLevel?: string } | undefined;
- this.thinkingLevel = agent?.thinkingLevel;
+ this.received = { ...this.received, thinkingLevel: agent?.thinkingLevel };
});
});
};
@@ -87,15 +107,30 @@ export class ModelState implements TuiSurface {
this.stop = () => runtime.listeners.delete(changed);
changed();
}
+ /** The state as shown: the choices in flight over what the server last sent. */
+ private get shown(): Shown {
+ return this.shownState();
+ }
+ /** The viewed conversation's model (guest-safe metadata). */
+ get model(): ModelMeta | null {
+ return this.shown.model;
+ }
+ get thinkingLevel(): string | undefined {
+ return this.shown.thinkingLevel;
+ }
+ get defaultModel(): ModelChoice | null {
+ return this.shown.defaultModel;
+ }
+ get scoped(): Scoped[] {
+ return this.shown.scoped;
+ }
/** The owner's Models reference once live. */
service(): Held | undefined {
- const ref = this.models();
- return liveIn(ref, "owner");
+ return liveIn(this.models, "owner");
}
/** The viewed conversation's participant reference once live. */
participant(): HeldIn | undefined {
- const ref = this.chatRef();
- return liveIn(ref, "participant");
+ return liveIn(this.chatRef, "participant");
}
selectorModels(): SelectorModel[] {
return this.catalog.map(({ provider, modelId, name }) => ({ provider, modelId, name }));
@@ -122,39 +157,79 @@ const describe = (state: ModelState): string =>
? `${lineText(state.model.provider)}/${lineText(state.model.modelId)} · thinking ${lineText(state.thinkingLevel ?? "off")}`
: "no model";
-/** Applies a model and/or level: the conversation's (Chat.configure) with a session, else the default. */
-async function apply(ctx: TuiContext, state: ModelState, change: { model?: SelectorModel; thinkingLevel?: string }) {
- const chat = ctx.identity.session ? state.participant() : undefined;
+/** A model and/or level a command chose. */
+type Change = { readonly model?: SelectorModel; readonly thinkingLevel?: string };
+
+/** The conversation's model as shown while a change to `model` is in flight: the catalog's facts about it. */
+function metaOf(state: ModelState, model: SelectorModel): ModelMeta {
+ const info = state.catalog.find((entry) => entry.provider === model.provider && entry.modelId === model.modelId);
+ const thinkingLevels = info?.thinkingLevels ?? [];
+ return {
+ provider: model.provider,
+ modelId: model.modelId,
+ name: info?.name ?? model.name,
+ contextWindow: info?.contextWindow ?? 0,
+ reasoning: thinkingLevels.length > 1,
+ thinkingLevels,
+ };
+}
+
+/**
+ * Applies a change: the conversation's (Chat.configure) with a session, else the default. The choice shows at once
+ * (the notice, and the selectors and cycling start from it); a refused call takes it back and says why.
+ */
+function choose(ctx: TuiContext, state: ModelState, change: Change, done: string): void {
if (ctx.identity.session) {
- if (!chat) throw new Error("This conversation is not connected or read-only");
- await chat.configure(
- {
- ...(change.model ? { model: { provider: change.model.provider, modelId: change.model.modelId } } : {}),
- ...(change.thinkingLevel ? { thinkingLevel: change.thinkingLevel } : {}),
- },
- BACKGROUND_CONTEXT,
+ const chat = state.participant();
+ if (!chat) {
+ ctx.notify("This conversation is not connected or read-only", "error");
+ return;
+ }
+ const guess = {
+ ...(change.model ? { model: metaOf(state, change.model) } : {}),
+ ...(change.thinkingLevel ? { thinkingLevel: change.thinkingLevel } : {}),
+ };
+ ctx.notify(done);
+ ctx.perform(
+ state.choosing(
+ guess,
+ chat.configure(
+ {
+ ...(change.model
+ ? { model: { provider: change.model.provider, modelId: change.model.modelId } }
+ : {}),
+ ...(change.thinkingLevel ? { thinkingLevel: change.thinkingLevel } : {}),
+ },
+ BACKGROUND_CONTEXT,
+ ),
+ ),
);
return;
}
- const service = state.service();
const model = change.model ?? state.defaultModel;
- if (!service || !model) throw new Error("Models are not available");
- await service.setDefault(
- {
- provider: model.provider,
- modelId: model.modelId,
- thinkingLevel: change.thinkingLevel ?? state.defaultModel?.thinkingLevel ?? "off",
- },
- BACKGROUND_CONTEXT,
- );
+ if (model) setDefault(ctx, state, { ...model, thinkingLevel: change.thinkingLevel }, done);
}
-const perform = (ctx: TuiContext, action: Promise, done?: string) =>
- void action
- .then(() => done && ctx.notify(done))
- .catch((error: unknown) => {
- ctx.notify(error instanceof Error ? error.message : String(error), "error");
- });
+/** Sets the default for new sessions (keeping the default's level unless `thinkingLevel` is given), shown at once. */
+function setDefault(
+ ctx: TuiContext,
+ state: ModelState,
+ choice: { provider: string; modelId: string; thinkingLevel: string | undefined },
+ done: string,
+): void {
+ const service = state.service();
+ if (!service) {
+ ctx.notify("Models are not available", "error");
+ return;
+ }
+ const defaultModel = {
+ provider: choice.provider,
+ modelId: choice.modelId,
+ thinkingLevel: choice.thinkingLevel ?? state.defaultModel?.thinkingLevel ?? "off",
+ };
+ ctx.notify(done);
+ ctx.perform(state.choosing({ defaultModel }, service.setDefault(defaultModel, BACKGROUND_CONTEXT)));
+}
/** Guests (and the owner without the catalog yet) see the model read-only. */
function readOnly(ctx: TuiContext, state: ModelState | undefined): boolean {
@@ -184,24 +259,13 @@ function openModelSelector(ctx: TuiContext, initialSearch?: string): void {
...(initialSearch ? { initialSearch } : {}),
onSelect: (model) => {
handle.close();
- perform(ctx, apply(ctx, state, { model }), `Model: ${model.provider}/${model.modelId}`);
+ choose(ctx, state, { model }, `Model: ${model.provider}/${model.modelId}`);
},
onSelectAsDefault: (model) => {
handle.close();
- const service = state.service();
- if (!service) return;
const thinkingLevel = state.thinkingLevel ?? state.defaultModel?.thinkingLevel ?? "off";
- perform(
- ctx,
- Promise.all([
- service.setDefault(
- { provider: model.provider, modelId: model.modelId, thinkingLevel },
- BACKGROUND_CONTEXT,
- ),
- ...(ctx.identity.session ? [apply(ctx, state, { model })] : []),
- ]),
- `Default model: ${model.provider}/${model.modelId}`,
- );
+ setDefault(ctx, state, { ...model, thinkingLevel }, `Default model: ${model.provider}/${model.modelId}`);
+ if (ctx.identity.session) choose(ctx, state, { model }, `Model: ${model.provider}/${model.modelId}`);
},
onCancel: () => handle.close(),
}),
@@ -222,24 +286,14 @@ function openThinkingSelector(ctx: TuiContext): void {
...(state.defaultModel ? { defaultLevel: state.defaultModel.thinkingLevel } : {}),
onSelect: (level) => {
handle.close();
- perform(ctx, apply(ctx, state, { thinkingLevel: level }), `Thinking level: ${level}`);
+ choose(ctx, state, { thinkingLevel: level }, `Thinking level: ${level}`);
},
onSelectAsDefault: (level) => {
handle.close();
- const service = state.service();
const model = state.model ?? state.defaultModel;
- if (!service || !model) return;
- perform(
- ctx,
- Promise.all([
- service.setDefault(
- { provider: model.provider, modelId: model.modelId, thinkingLevel: level },
- BACKGROUND_CONTEXT,
- ),
- ...(ctx.identity.session ? [apply(ctx, state, { thinkingLevel: level })] : []),
- ]),
- `Default thinking level: ${level}`,
- );
+ if (!model) return;
+ setDefault(ctx, state, { ...model, thinkingLevel: level }, `Default thinking level: ${level}`);
+ if (ctx.identity.session) choose(ctx, state, { thinkingLevel: level }, `Thinking level: ${level}`);
},
onCancel: () => handle.close(),
}),
@@ -260,9 +314,10 @@ function cycleModel(ctx: TuiContext, direction: 1 | -1): void {
const next = ring[(index + direction + ring.length) % ring.length]!;
const info = state.catalog.find((model) => model.provider === next.provider && model.modelId === next.modelId);
const model = { provider: next.provider, modelId: next.modelId, name: info?.name ?? next.modelId };
- perform(
+ choose(
ctx,
- apply(ctx, state, { model, ...(next.thinkingLevel ? { thinkingLevel: next.thinkingLevel } : {}) }),
+ state,
+ { model, ...(next.thinkingLevel ? { thinkingLevel: next.thinkingLevel } : {}) },
`Model: ${next.provider}/${next.modelId}${next.thinkingLevel ? ` · thinking ${next.thinkingLevel}` : ""}`,
);
}
@@ -278,7 +333,7 @@ function cycleThinking(ctx: TuiContext): void {
}
const current = state.thinkingLevel ?? "off";
const next = levels[(levels.indexOf(current) + 1) % levels.length]!;
- perform(ctx, apply(ctx, state, { thinkingLevel: next }), `Thinking level: ${next}`);
+ choose(ctx, state, { thinkingLevel: next }, `Thinking level: ${next}`);
}
function openScopedModels(ctx: TuiContext): void {
@@ -292,12 +347,14 @@ function openScopedModels(ctx: TuiContext): void {
onSave: (patterns) => {
handle.close();
const service = state.service();
- if (service)
- perform(
- ctx,
- service.setEnabledModels({ patterns }, BACKGROUND_CONTEXT),
- patterns.length ? `Scoped models: ${patterns.length}` : "Model scope cleared",
- );
+ if (!service) return;
+ ctx.notify(patterns.length ? `Scoped models: ${patterns.length}` : "Model scope cleared");
+ // The selector saves exact `provider/modelId` patterns: cycling uses them at once.
+ const scoped = patterns.map((pattern) => {
+ const slash = pattern.indexOf("/");
+ return { provider: pattern.slice(0, slash), modelId: pattern.slice(slash + 1) };
+ });
+ ctx.perform(state.choosing({ scoped }, service.setEnabledModels({ patterns }, BACKGROUND_CONTEXT)));
},
onCancel: () => handle.close(),
}),
@@ -331,7 +388,7 @@ export const MODEL_COMMANDS: Record = {
? state?.catalog.find((model) => `${model.provider}/${model.modelId}`.toLowerCase() === query.toLowerCase())
: undefined;
if (state && exact && !readOnly(ctx, state))
- perform(ctx, apply(ctx, state, { model: exact }), `Model: ${exact.provider}/${exact.modelId}`);
+ choose(ctx, state, { model: exact }, `Model: ${exact.provider}/${exact.modelId}`);
else if (!exact) openModelSelector(ctx, query);
},
},
diff --git a/apps/p2/plugins/models/worker.ts b/apps/p2/plugins/models/worker.ts
index 7807cb5b0..1de156ce2 100644
--- a/apps/p2/plugins/models/worker.ts
+++ b/apps/p2/plugins/models/worker.ts
@@ -1,17 +1,8 @@
-import { obj, unknown } from "@earendil-works/pi-schema";
-import { defineFacet, defineState, unsafe, untilAborted } from "@p2/app/v1";
-import type { MobileSettings } from "@p2/app/v1/node";
+import { defineFacet, followNative } from "@p2/app/v1";
import { HarnessConfig, workerOf } from "@p2/app/v1/worker";
import { openResources } from "./config.ts";
import { WorkerModels } from "./contract.ts";
-/** The worker's current, valid settings (hot-reloaded). */
-const Settings = defineState("p2.worker-models", {
- schema: obj({ settings: unsafe(unknown()) }),
- version: 1,
- initial: () => ({ settings: {} }),
-});
-
export default defineFacet((host) => {
host.serve(WorkerModels, (_params, self) => {
const worker = workerOf(self);
@@ -19,26 +10,18 @@ export default defineFacet((host) => {
() => openResources(worker().init.dataDir, worker().init.fake),
(opened) => opened.close(),
);
- const settings = self.state(Settings);
- self.onLive(async (ctx) => {
- const publish = () =>
- settings.change((draft) => {
- draft.settings = resources.value.settings;
- }, ctx);
- await publish();
- const stop = resources.value.subscribe(() => void publish().catch(() => {}));
- try {
- await untilAborted(ctx);
- } finally {
- stop();
- }
- });
self.contribute(HarnessConfig, {
default: {
models: () => resources.value.models,
settings: () => resources.value.harnessSettings(),
},
});
- return { settings: self.derive([settings], ([value]) => value.settings) };
+ return {
+ settings: followNative(
+ self,
+ (changed) => resources.value.subscribe(changed),
+ () => resources.value.settings,
+ ),
+ };
});
});
diff --git a/apps/p2/plugins/ports/PortsPanel.tsx b/apps/p2/plugins/ports/PortsPanel.tsx
index 554ad37c0..10d39104d 100644
--- a/apps/p2/plugins/ports/PortsPanel.tsx
+++ b/apps/p2/plugins/ports/PortsPanel.tsx
@@ -1,29 +1,51 @@
+import type { Static } from "@earendil-works/pi-schema";
import { BACKGROUND_CONTEXT, liveIn } from "@p2/app/v1";
-import { CopyButton, Sheet, useObject, useReadable } from "@p2/app/v1/web";
+import { CopyButton, errorText, Sheet, useObject, useOptimisticCall, useReadable } from "@p2/app/v1/web";
import { useState } from "react";
-import { Ports } from "./contract.ts";
+import { type ForwardSchema, Ports } from "./contract.ts";
-/** Every forward is visible to the owner. Making one public requires a second explicit tap. */
+type Forward = Static;
+
+/** A change of the list while its call is in flight. */
+type PortGuess =
+ | { readonly kind: "add"; readonly key: string; readonly port: number; readonly name: string | undefined }
+ | { readonly kind: "remove"; readonly id: string }
+ | { readonly kind: "public"; readonly id: string; readonly open: boolean };
+
+/**
+ * The list with a change applied. An added forward's id and addresses are the server's: until it answers, the row
+ * shows the port and name only (no Open, no link).
+ */
+function applyPort(forwards: Forward[] | undefined, guess: PortGuess): Forward[] | undefined {
+ if (!forwards) return forwards;
+ if (guess.kind === "add")
+ return [...forwards, { id: guess.key, port: guess.port, ...(guess.name ? { name: guess.name } : {}) }];
+ if (guess.kind === "remove") return forwards.filter((forward) => forward.id !== guess.id);
+ return forwards.map((forward) =>
+ forward.id === guess.id
+ ? { ...forward, public: guess.open, ...(guess.open ? {} : { publicUrl: undefined }) }
+ : forward,
+ );
+}
+
+/**
+ * Every forward is visible to the owner. Making one public requires a second explicit tap. Adding, removing and the
+ * public switch show at once; a refused one rolls back with its error.
+ */
export function PortsPanel({ onClose }: { onClose: () => void }) {
const service = liveIn(useObject(Ports, {}), "owner");
- const forwards = useReadable(service?.forwards);
const origin = useReadable(service?.radiusOrigin);
const [port, setPort] = useState("");
const [name, setName] = useState("");
const [removing, setRemoving] = useState();
const [publishing, setPublishing] = useState();
- const [busy, setBusy] = useState(false);
const [error, setError] = useState();
- const run = async (action: () => Promise) => {
- setBusy(true);
+ const [forwards, change] = useOptimisticCall(useReadable(service?.forwards), applyPort, (reason) =>
+ setError(errorText(reason)),
+ );
+ const run = (guess: PortGuess, call: () => Promise) => {
setError(undefined);
- try {
- await action();
- } catch (reason) {
- setError(reason instanceof Error ? reason.message : String(reason));
- } finally {
- setBusy(false);
- }
+ change(guess, call);
};
const onDevice = ["127.0.0.1", "localhost", "[::1]"].includes(location.hostname);
return (
@@ -42,11 +64,14 @@ export function PortsPanel({ onClose }: { onClose: () => void }) {
{forwards?.map((forward) => {
const url = forward.url ?? (onDevice ? `http://localhost:${forward.port}/` : undefined);
- const setPublic = (open: boolean) =>
- void run(async () => {
- await service?.setPublic({ id: forward.id, public: open }, BACKGROUND_CONTEXT);
- setPublishing(undefined);
- });
+ const pending = forward.id.startsWith("adding-");
+ const setPublic = (open: boolean) => {
+ setPublishing(undefined);
+ if (service)
+ run({ kind: "public", id: forward.id, open }, () =>
+ service.setPublic({ id: forward.id, public: open }, BACKGROUND_CONTEXT),
+ );
+ };
return (
@@ -63,10 +88,13 @@ export function PortsPanel({ onClose }: { onClose: () => void }) {
- void run(() => service!.remove({ id: forward.id }, BACKGROUND_CONTEXT))
- }
+ onClick={() => {
+ setRemoving(undefined);
+ if (service)
+ run({ kind: "remove", id: forward.id }, () =>
+ service.remove({ id: forward.id }, BACKGROUND_CONTEXT),
+ );
+ }}
>
Remove
@@ -76,7 +104,12 @@ export function PortsPanel({ onClose }: { onClose: () => void }) {
>
) : (
<>
-
setRemoving(forward.id)}>
+ setRemoving(forward.id)}
+ >
Remove
{forward.url && }
@@ -94,12 +127,7 @@ export function PortsPanel({ onClose }: { onClose: () => void }) {
Anyone with the link can open localhost:{forward.port} without signing in.
- setPublic(true)}
- >
+ setPublic(true)}>
Make public
setPublishing(undefined)}>
@@ -111,7 +139,7 @@ export function PortsPanel({ onClose }: { onClose: () => void }) {
event.target.checked ? setPublishing(forward.id) : setPublic(false)
}
@@ -134,14 +162,13 @@ export function PortsPanel({ onClose }: { onClose: () => void }) {
className="inline-form"
onSubmit={(event) => {
event.preventDefault();
- void run(async () => {
- await service?.add(
- { port: Number(port.trim()), ...(name.trim() ? { name: name.trim() } : {}) },
- BACKGROUND_CONTEXT,
- );
- setPort("");
- setName("");
- });
+ if (!service) return;
+ const added = { port: Number(port.trim()), ...(name.trim() ? { name: name.trim() } : {}) };
+ run({ kind: "add", key: `adding-${crypto.randomUUID()}`, port: added.port, name: added.name }, () =>
+ service.add(added, BACKGROUND_CONTEXT),
+ );
+ setPort("");
+ setName("");
}}
>
void }) {
placeholder="Name (optional)"
aria-label="Name"
/>
-
+
Add
diff --git a/apps/p2/plugins/ports/server.ts b/apps/p2/plugins/ports/server.ts
index afff5f9b4..2bd4ac45a 100644
--- a/apps/p2/plugins/ports/server.ts
+++ b/apps/p2/plugins/ports/server.ts
@@ -1,9 +1,7 @@
-import { arr, nul, obj, str, union } from "@earendil-works/pi-schema";
import {
- BACKGROUND_CONTEXT,
defineFacet,
- defineTypedState,
deriveTyped,
+ followNative,
forbidden,
isOwner,
isWorker,
@@ -12,43 +10,30 @@ import {
} from "@p2/app/v1";
import { Hosts, Server } from "@p2/app/v1/server";
import { Catalog, SessionArchived } from "../sessions/contract.ts";
-import { ForwardSchema, ForwardStore, ForwardsNative, Ports } from "./contract.ts";
+import { ForwardStore, ForwardsNative, Ports } from "./contract.ts";
import { forwardView } from "./protocol.ts";
import { Forwards } from "./store.ts";
-/** The forwards' list and Radius origin as the store shows them (in memory). */
-const Listed = defineTypedState("p2.forward-store", {
- schema: obj({ list: arr(ForwardSchema), radiusOrigin: union([str(), nul()]) }),
- version: 1,
- initial: () => ({ list: [], radiusOrigin: null }),
-});
-
export default defineFacet((host) => {
serveNative(host, ForwardsNative);
host.serve(ForwardStore, (_params, self) => {
const server = self.get(Server, {});
const catalog = self.get(Catalog, {});
- const listed = self.state(Listed);
- const show = (store: Forwards) =>
- listed.change((draft) => {
- // A draft's arrays are mutable; the store's list is the same JSON.
- draft.list = store.list() as typeof draft.list;
- draft.radiusOrigin = store.radiusOrigin;
- }, BACKGROUND_CONTEXT);
const forwards = self.open(
async () => {
const store = await Forwards.open(server.info.value.dataDir);
// Forwards of archived sessions close first, so the list never shows them.
for (const session of catalog.sessions.value)
if (session.archivedAt !== null) await store.closeSession(session.id);
- await show(store);
return store;
},
() => {},
);
- self.open(
- async () => forwards.value.events.subscribe(() => void show(forwards.value).catch(() => {})),
- (stop) => stop(),
+ // The store's list and Radius origin: a method's write resolves after the store told this source.
+ const listed = followNative(
+ self,
+ (changed) => forwards.value.events.subscribe(changed),
+ () => ({ list: forwards.value.list(), radiusOrigin: forwards.value.radiusOrigin }),
);
self.contribute(ForwardsNative, {
get forwards() {
diff --git a/apps/p2/plugins/ports/tui.ts b/apps/p2/plugins/ports/tui.ts
index 6fa75c108..bbc8f9fe9 100644
--- a/apps/p2/plugins/ports/tui.ts
+++ b/apps/p2/plugins/ports/tui.ts
@@ -3,7 +3,7 @@
// (p2-tui-port.md section 3.5); OSC 8 hyperlinks stay stripped. `/ports` (P11f) is the web's ports panel.
import { Text } from "@earendil-works/pi-tui";
-import { BACKGROUND_CONTEXT, contribute, defineFacet, follow, liveIn, whenLive } from "@p2/app/v1";
+import { BACKGROUND_CONTEXT, contribute, defineFacet, type HeldIn, liveIn, whenLive } from "@p2/app/v1";
import type { TuiContext } from "@p2/app/v1/tui";
import {
confirm,
@@ -58,70 +58,62 @@ export const portsRenderers: ToolRenderers = {
},
};
+/** A change of the list while its call is in flight; an added forward's id and addresses are the server's. */
+type PortGuess =
+ | { kind: "add"; key: string; port: number; name?: string; public?: boolean }
+ | { kind: "remove"; id: string }
+ | { kind: "public"; id: string; open: boolean };
+
+function applyPort(forwards: ForwardView[], guess: PortGuess): ForwardView[] {
+ if (guess.kind === "add")
+ return [
+ ...forwards,
+ { id: guess.key, port: guess.port, ...(guess.name ? { name: guess.name } : {}), public: guess.public },
+ ];
+ if (guess.kind === "remove") return forwards.filter((forward) => forward.id !== guess.id);
+ return forwards.map((forward) => (forward.id === guess.id ? { ...forward, public: guess.open } : forward));
+}
+
/**
* `/ports [open [name] [public]]` (P11f), after the web's ports panel: the owner's forwards with their
* addresses; Ctrl+P makes one public or private (public only after a confirmation: anyone with the address can open
* it), Ctrl+D closes one. `open` adds a forward; a public one is confirmed first.
*/
function openPorts(ctx: TuiContext, args: string | undefined): void {
- const ref = ctx.host.hold(Ports, {}, BACKGROUND_CONTEXT);
- const owner = () => liveIn(ref, "owner");
- let forwards: ForwardView[] = [];
- let status = "";
- let closed = false;
- let stop = (): void => {};
- // The forwards are the owner variant's: subscribed once the reference knows its variant.
- void whenLive(ref)
- .then(() => {
- if (closed || ref.variant !== "owner") return;
- stop = follow(ref.forwards, (value) => {
- forwards = value;
- ctx.requestRender();
+ const scope = ctx.scope();
+ let received: ForwardView[] = [];
+ const ref = scope.watch(Ports, {}, (held, follow) => {
+ if (held.variant === "owner")
+ follow(held.forwards, (value) => {
+ received = value;
});
- })
- .catch(() => {});
- const run = (label: string, action: () => Promise) => {
- status = `${label}…`;
- void action()
- .then(() => {
- status = `${label}: done`;
- })
- .catch((error: unknown) => {
- status = `${label}: ${error instanceof Error ? error.message : String(error)}`;
- })
- .finally(() => ctx.requestRender());
- };
+ });
+ // Adding, closing and the public switch show at once; a refused one rolls back with its error.
+ const [forwards, changing] = scope.optimistic(() => received, applyPort);
+ const change = (guess: PortGuess, call: (service: HeldIn) => Promise) =>
+ ctx.perform(
+ changing(
+ guess,
+ (async () => {
+ // `/ports open` runs before the list arrived: the call waits for the reference.
+ await whenLive(ref);
+ const service = liveIn(ref, "owner");
+ if (!service) throw new Error("Ports are the owner's");
+ await call(service);
+ })(),
+ ),
+ );
const publicWarning = (port: number) => [
`Make port ${port} public?`,
"Anyone with its public address can open it without signing in.",
];
- const words = args?.trim().split(/\s+/).filter(Boolean) ?? [];
- if (words[0] === "open") {
- const port = Number(words[1]);
- const isPublic = words.includes("public");
- const name = words.slice(2).find((word) => word !== "public");
- if (!Number.isInteger(port) || port < 1 || port > 65535) ctx.notify("Usage: /ports open [name] [public]");
- else
- void whenLive(ref)
- .then(async () => {
- if (isPublic && !(await confirm(ctx, publicWarning(port)))) return;
- const service = owner();
- if (!service) return;
- await service.add(
- { port, ...(name ? { name } : {}), ...(isPublic ? { public: true } : {}) },
- BACKGROUND_CONTEXT,
- );
- ctx.notify(`Forwarding port ${port}`);
- })
- .catch((error: unknown) => ctx.notify(error instanceof Error ? error.message : String(error), "error"));
- }
- const byRow = (row: ListRow | undefined) => forwards.find((forward) => forward.id === row?.key);
+ const byRow = (row: ListRow | undefined) => forwards().find((forward) => forward.id === row?.key);
const overlay = ctx.overlay(
"ports",
new ListOverlay(ctx, {
- title: () => `Ports (${forwards.length})`,
+ title: () => `Ports (${forwards().length})`,
rows: () =>
- forwards.map((forward) => ({
+ forwards().map((forward) => ({
key: forward.id,
label: `${forward.name ?? `Port ${forward.port}`} localhost:${forward.port}${forward.public ? " public" : ""}`,
detail: forward.session ? `by ${forward.sessionTitle ?? "another session"}` : "",
@@ -138,20 +130,13 @@ function openPorts(ctx: TuiContext, args: string | undefined): void {
label: "public/private",
run: (row) => {
const forward = byRow(row);
- const service = owner();
- if (!forward || service?.variant !== "owner") return;
- if (forward.public)
- run(`Making ${forward.port} private`, () =>
- service.setPublic({ id: forward.id, public: false }, BACKGROUND_CONTEXT),
- );
- else
- void confirm(ctx, publicWarning(forward.port)).then(
- (yes) =>
- yes &&
- run(`Making ${forward.port} public`, () =>
- service.setPublic({ id: forward.id, public: true }, BACKGROUND_CONTEXT),
- ),
+ if (!forward || forward.id.startsWith("adding-")) return;
+ const setPublic = (open: boolean) =>
+ change({ kind: "public", id: forward.id, open }, (service) =>
+ service.setPublic({ id: forward.id, public: open }, BACKGROUND_CONTEXT),
);
+ if (forward.public) setPublic(false);
+ else void confirm(ctx, publicWarning(forward.port)).then((yes) => yes && setPublic(true));
},
},
{
@@ -159,25 +144,38 @@ function openPorts(ctx: TuiContext, args: string | undefined): void {
label: "close",
run: (row) => {
const forward = byRow(row);
- const service = owner();
- if (forward && service?.variant === "owner")
- run(`Closing ${forward.port}`, () => service.remove({ id: forward.id }, BACKGROUND_CONTEXT));
+ if (forward && !forward.id.startsWith("adding-"))
+ change({ kind: "remove", id: forward.id }, (service) =>
+ service.remove({ id: forward.id }, BACKGROUND_CONTEXT),
+ );
},
},
],
- footer: () => [
- ctx.theme.current.fg("dim", " /ports open [name] [public] adds one"),
- ...(status ? [ctx.theme.current.fg("muted", ` ${lineText(status)}`)] : []),
- ],
+ footer: () => [ctx.theme.current.fg("dim", " /ports open [name] [public] adds one")],
onClose: () => {
- closed = true;
- stop();
- ref.release();
+ scope.dispose();
overlay.close();
},
}),
{ width: "100%", anchor: "bottom-center" },
);
+ // After the overlay: a confirmation opens on top of it.
+ const words = args?.trim().split(/\s+/).filter(Boolean) ?? [];
+ if (words[0] === "open") {
+ const port = Number(words[1]);
+ const isPublic = words.includes("public");
+ const name = words.slice(2).find((word) => word !== "public");
+ if (!Number.isInteger(port) || port < 1 || port > 65535) ctx.notify("Usage: /ports open [name] [public]");
+ else
+ void (async () => {
+ if (isPublic && !(await confirm(ctx, publicWarning(port)))) return;
+ const added = { port, ...(name ? { name } : {}), ...(isPublic ? { public: true } : {}) };
+ change({ kind: "add", key: `adding-${crypto.randomUUID()}`, ...added }, async (service) => {
+ await service.add(added, BACKGROUND_CONTEXT);
+ ctx.notify(`Forwarding port ${port}`);
+ });
+ })();
+ }
}
const PORT_COMMANDS: Record = {
diff --git a/apps/p2/plugins/ports/worker.ts b/apps/p2/plugins/ports/worker.ts
index 708db0c41..2326675b4 100644
--- a/apps/p2/plugins/ports/worker.ts
+++ b/apps/p2/plugins/ports/worker.ts
@@ -10,7 +10,7 @@ export default defineFacet((host) => {
(query, context) =>
worker().root(
async (ctx) => expectVariant(await ctx.get(Ports, {}), "worker").query({ conversation: 1, query }, ctx),
- context.abortSignal,
+ context,
),
() => worker().init.values.session.id,
);
diff --git a/apps/p2/plugins/powershell/worker.ts b/apps/p2/plugins/powershell/worker.ts
index 1791a7ceb..5d0f592a6 100644
--- a/apps/p2/plugins/powershell/worker.ts
+++ b/apps/p2/plugins/powershell/worker.ts
@@ -12,7 +12,6 @@ import {
} from "@earendil-works/pi-durable";
import { createPowerShellTool } from "@earendil-works/pi-durable/tools";
import { defineFacet, type Location, signalOf } from "@p2/app/v1";
-import { durableContext } from "@p2/app/v1/node";
import { CommitSignal, Extensions, harnessOf, workerOf } from "@p2/app/v1/worker";
import { Subagents } from "../subagents/contract.worker.ts";
import { POWERSHELL_EXTENSION, POWERSHELL_TOOL, PowerShell } from "./contract.ts";
@@ -94,8 +93,7 @@ export default defineFacet((host) => {
const seen = signal.version;
let unknown = false;
try {
- const context = durableContext(ctx);
- const registry = await durable.snapshot(Subagents, ROOT_CONVERSATION_ID, context);
+ const registry = await durable.snapshot(Subagents, ROOT_CONVERSATION_ID, ctx);
const ids: ConversationId[] = [
ROOT_CONVERSATION_ID,
...Object.values(registry?.agents ?? {}).map((agent) => agent.conversationId),
@@ -104,8 +102,8 @@ export default defineFacet((host) => {
const offered: Record = {};
const changes: { id: ConversationId; on: boolean }[] = [];
for (const id of ids) {
- const state = await durable.snapshot(AgentDoc, id, context);
- const windows = await isWindows(await harness().environments.locate(id, context));
+ const state = await durable.snapshot(AgentDoc, id, ctx);
+ const windows = await isWindows(await harness().environments.locate(id, ctx));
if (windows === undefined) unknown = true;
// Unknown keeps what it has: an unreachable machine does not take PowerShell away.
const on = windows === undefined ? offersPowerShell(state) : windows && enables(state, "bash");
@@ -115,7 +113,7 @@ export default defineFacet((host) => {
if (changes.length)
await durable.commit(async (tx) => {
for (const { id, on } of changes) setPowerShell(await tx.doc(AgentDoc, id), on);
- }, context);
+ }, ctx);
offeredState.set(offered);
} catch (error) {
if (signalOf(ctx).aborted) return;
diff --git a/apps/p2/plugins/prompt/tui.ts b/apps/p2/plugins/prompt/tui.ts
index 0c1457bcf..08a117a1d 100644
--- a/apps/p2/plugins/prompt/tui.ts
+++ b/apps/p2/plugins/prompt/tui.ts
@@ -2,7 +2,7 @@
// the viewed conversation's `PromptResources`, and their expansion on the session worker when the editor submits.
import type { Held } from "@p2/app/v1";
-import { BACKGROUND_CONTEXT, contribute, defineFacet, statusOf } from "@p2/app/v1";
+import { BACKGROUND_CONTEXT, contribute, defineFacet, liveIn } from "@p2/app/v1";
import type { TuiContext } from "@p2/app/v1/tui";
import {
lineText,
@@ -21,7 +21,7 @@ export class PromptSurface implements TuiSurface {
readonly runtime: TuiRuntime;
private scope: TuiScope | undefined;
private viewed: string | undefined;
- private ref: (() => Held) | undefined;
+ private ref: Held | undefined;
templates: readonly TemplateInfo[] = [];
skills: readonly SkillInfo[] = [];
private stop: () => void;
@@ -66,9 +66,8 @@ export class PromptSurface implements TuiSurface {
];
}
async expand(text: string): Promise {
- const ref = this.ref?.();
- if (ref === undefined || statusOf(ref) !== "live" || ref.variant !== "participant") return undefined;
- return (await ref.expand({ text }, BACKGROUND_CONTEXT)).text;
+ const ref = liveIn(this.ref, "participant");
+ return ref && (await ref.expand({ text }, BACKGROUND_CONTEXT)).text;
}
render(): string[] {
return [];
diff --git a/apps/p2/plugins/prompt/worker.ts b/apps/p2/plugins/prompt/worker.ts
index c2fb32527..5f680bf24 100644
--- a/apps/p2/plugins/prompt/worker.ts
+++ b/apps/p2/plugins/prompt/worker.ts
@@ -11,14 +11,13 @@ import {
} from "@earendil-works/pi-durable";
import type { ExecutionEnv } from "@earendil-works/pi-durable/env";
import { RemoteExecutionEnv } from "@earendil-works/pi-env";
-import { arr, num, obj, opt, str } from "@earendil-works/pi-schema";
+import { num, obj, opt, str } from "@earendil-works/pi-schema";
import {
BACKGROUND_CONTEXT,
type Context,
canSend,
defineFacet,
defineRegistry,
- defineTypedState,
deriveTyped,
forbidden,
local,
@@ -26,6 +25,8 @@ import {
nativeOf,
type Subscribable,
serveNative,
+ signalOf,
+ untilAborted,
} from "@p2/app/v1";
import { remotePaths } from "@p2/app/v1/node";
import {
@@ -38,7 +39,7 @@ import {
workerOf,
} from "@p2/app/v1/worker";
import type { LoadedPrompt } from "./context-files.ts";
-import { PromptResources, SkillInfoSchema, TemplateInfoSchema } from "./contract.ts";
+import { PromptResources, type SkillInfo, type TemplateInfo } from "./contract.ts";
import { envFiles, promptObserver } from "./environment.ts";
import { expandPrompt } from "./templates.ts";
@@ -75,7 +76,7 @@ type PromptLoads = {
/** Where conversations find their prompt resources: the session's (following its moves) or their own location's. */
type PromptFiles = {
- readonly session: PromptLoads;
+ readonly session: MovingPromptLoads;
of: (id: ConversationId, context: Context, read?: DocumentReader) => Promise;
};
@@ -140,11 +141,20 @@ function promptLoads(dataDir: string, at: Location, environment: Promise;
+};
+
/**
- * The prompt resources of the session's current location: loaded for the location it is at, and loaded anew (from the
- * new cwd, on the new machine) after each move. `loads` follows the current location's loads.
+ * The prompt resources of the session's current location: loaded for the location it is at, and loaded anew after
+ * each move. `loads` follows the current location's loads.
*/
-function followedPromptLoads(location: SessionLocation, open: (at: Location) => PromptLoads): PromptLoads {
+function movingPromptLoads(location: SessionLocation, open: (at: Location) => PromptLoads): MovingPromptLoads {
const listeners = new Set<(loaded: Loaded) => void>();
let latest: Loaded | undefined;
const forward = (loaded: Loaded): void => {
@@ -157,15 +167,9 @@ function followedPromptLoads(location: SessionLocation, open: (at: Location) =>
loads.loads.subscribe((loaded) => {
if (inner === loads) forward(loaded);
});
+ let version = location.version;
let inner = open(location.value);
let unsubscribe = follow(inner);
- const stop = location.subscribe((at) => {
- const previous = inner;
- unsubscribe();
- inner = open(at);
- unsubscribe = follow(inner);
- void previous.close().catch((error: unknown) => console.error("p2 prompt: closing old watches:", error));
- });
return {
current: () => inner.current(),
started: () => inner.started(),
@@ -177,8 +181,16 @@ function followedPromptLoads(location: SessionLocation, open: (at: Location) =>
return () => listeners.delete(listener);
},
},
+ move: async (at, moved) => {
+ if (moved <= version) return;
+ version = moved;
+ const previous = inner;
+ unsubscribe();
+ inner = open(at);
+ unsubscribe = follow(inner);
+ await previous.close().catch((error: unknown) => console.error("p2 prompt: closing old watches:", error));
+ },
close: async () => {
- stop();
unsubscribe();
await inner.close();
},
@@ -199,12 +211,18 @@ const Prompt = local("p2.prompt", {
/** The open `Prompt`'s files (the native it contributes), for `PromptResources`. */
const PromptFilesNative = defineRegistry("p2.prompt-files", { version: 1 });
-/** The templates and skills where one conversation works, as `PromptResources` shows them. */
-const Listed = defineTypedState("p2.prompt-resources", {
- schema: obj({ templates: arr(TemplateInfoSchema), skills: arr(SkillInfoSchema) }),
- version: 1,
- initial: () => ({ templates: [], skills: [] }),
-});
+/** The templates and skills of a load, as `PromptResources` shows them. */
+function listedOf(loaded: LoadedPrompt): { templates: TemplateInfo[]; skills: SkillInfo[] } {
+ return {
+ templates: loaded.templates.map(({ name, description, argumentHint, source }) => ({
+ name,
+ description,
+ ...(argumentHint ? { argumentHint } : {}),
+ source,
+ })),
+ skills: loaded.skillFiles.map(({ name, description }) => ({ name, description })),
+ };
+}
export default defineFacet((host) => {
serveNative(host, PromptFilesNative);
@@ -216,7 +234,7 @@ export default defineFacet((host) => {
const environments = harness().environments;
const dataDir = worker().init.dataDir;
const open = (at: Location): PromptLoads => promptLoads(dataDir, at, environments.at(at));
- const session = followedPromptLoads(worker().location, open);
+ const session = movingPromptLoads(worker().location, open);
const own = new Map();
const files: PromptFiles = {
session,
@@ -241,13 +259,29 @@ export default defineFacet((host) => {
},
(value) => value.close(),
);
- self.open(
- async () =>
- filesResource.value.files.session.loads.subscribe(({ error }) => {
- if (error !== undefined) console.error(`p2 prompt: ${error}`);
- }),
- (stop) => stop(),
- );
+ // The session's resources follow its moves (one committed while the files opened included).
+ self.onLive(async (ctx) => {
+ const location = worker().location;
+ const session = filesResource.value.files.session;
+ for (;;) {
+ // Listening before moving: a move committed while the old watches close is not missed.
+ const moved = location.next(signalOf(ctx));
+ moved.catch(() => {}); // The abort ends this routine below.
+ await session.move(location.value, location.version);
+ await moved;
+ }
+ });
+ // Load failures of the session's resources go to the worker's log (they also leave the last good load in use).
+ self.onLive(async (ctx) => {
+ const stop = filesResource.value.files.session.loads.subscribe(({ error }) => {
+ if (error !== undefined) console.error(`p2 prompt: ${error}`);
+ });
+ try {
+ await untilAborted(ctx);
+ } finally {
+ stop();
+ }
+ });
self.contribute(PromptFilesNative, {
get session() {
return filesResource.value.files.session;
@@ -312,31 +346,18 @@ export default defineFacet((host) => {
if (!canSend(principal, session)) throw forbidden("participants only");
const prompt = self.get(Prompt, {});
const files = nativeOf(self, PromptFilesNative, "The prompt files");
- const listed = self.state(Listed);
- let fingerprint: string | undefined;
- const show = (loaded: LoadedPrompt, context: Context) => {
- fingerprint = loaded.fingerprint;
- void listed
- .change((draft) => {
- draft.templates = loaded.templates.map(({ name, description, argumentHint, source }) => ({
- name,
- description,
- ...(argumentHint ? { argumentHint } : {}),
- source,
- }));
- draft.skills = loaded.skillFiles.map(({ name, description }) => ({ name, description }));
- }, context)
- .catch(() => {});
- };
- self.open(
- async (ctx) => {
- const loads = await files().of(conversation as ConversationId, ctx);
- return loads.loads.subscribe(({ prompt: loaded }) => {
- if (loaded && loaded.fingerprint !== fingerprint) show(loaded, BACKGROUND_CONTEXT);
- });
- },
- (stop) => stop(),
- );
+ // The loads where the conversation works, followed: each new load (a changed fingerprint) is the next value.
+ const listed = self.follow(async (observer, ctx) => {
+ const loads = await files().of(conversation as ConversationId, ctx);
+ let fingerprint = loads.current().fingerprint;
+ const stop = loads.loads.subscribe(({ prompt: loaded }) => {
+ if (!loaded || loaded.fingerprint === fingerprint) return;
+ fingerprint = loaded.fingerprint;
+ const value = listedOf(loaded);
+ observer.advance({ kind: "value", value, ops: [["r", value]], retired: false }, BACKGROUND_CONTEXT);
+ });
+ return { value: listedOf(loads.current()), detach: stop };
+ });
return {
variant: "participant",
members: {
diff --git a/apps/p2/plugins/search/contract.ts b/apps/p2/plugins/search/contract.ts
index 2572b4939..5d9dafccb 100644
--- a/apps/p2/plugins/search/contract.ts
+++ b/apps/p2/plugins/search/contract.ts
@@ -1,5 +1,16 @@
import { arr, bool, int, nul, num, obj, opt, type Static, str, union } from "@earendil-works/pi-schema";
-import { literal, local, method, param, propertiesOf, record, remote, serverHost, sessionHost } from "@p2/app/v1";
+import {
+ evolve,
+ literal,
+ local,
+ method,
+ param,
+ propertiesOf,
+ record,
+ remote,
+ serverHost,
+ sessionHost,
+} from "@p2/app/v1";
/** One committed user or assistant message of a session conversation, as indexed. */
export const IndexedMessageSchema = obj({
@@ -66,6 +77,13 @@ const HistoryQuerySchema = obj({
limit: int({ minimum: 1, maximum: 20 }),
});
+/** A session's index rebuild: running, or failed with why (kept until the next rebuild of that session). */
+export const RebuildSchema = union([
+ obj({ session: str(), phase: literal("running") }),
+ obj({ session: str(), phase: literal("failed"), error: str() }),
+]);
+export type Rebuild = Static;
+
/** Root conversation entry IDs by session. */
const EntryIdsSchema = record(num());
@@ -92,6 +110,10 @@ export const SearchIndex = local("p2.search-index", {
history: method(obj({ session: str(), ...propertiesOf(HistoryQuerySchema) }), HistoryHitsSchema),
/** Message matches among `sessions` (literal terms, all required). */
messages: method(obj({ query: str(), sessions: arr(str()) }), arr(MessageHitSchema)),
+ /** The index rebuilds running or failed (a job each: they survive a server restart). */
+ rebuilds: arr(RebuildSchema),
+ /** Starts re-indexing `session` from the start (`Search.rebuildIndex`); refuses an unknown session. */
+ rebuild: method(obj({ session: str() })),
},
});
@@ -116,8 +138,8 @@ export const SearchFeed = remote("p2.search-feed", {
},
});
-/** Search over session titles, working directories and message text (owner only). */
-export const Search = remote("p2.search", {
+/** Search over session titles, working directories and message text (owner only): the base of the history. */
+const SearchV1 = remote("p2.search", {
version: 1,
params: {},
hostAddress: serverHost,
@@ -128,7 +150,7 @@ export const Search = remote("p2.search", {
/**
* Re-indexes one session from the start (Pi Mobile's `rebuildIndex`): its worker stops, its index entries
* go, and a fresh worker scans its history; an archived session's worker scans without resuming work and
- * stops again. Returns once the scan committed.
+ * stops again.
*/
rebuildIndex: method(obj({ session: str() })),
},
@@ -136,6 +158,12 @@ export const Search = remote("p2.search", {
},
});
+/**
+ * Search over session titles, working directories and message text (owner only). 2: `rebuildIndex` answers once the
+ * rebuild started (a job); `rebuilds` shows it running, and why it failed. The provider still serves 1.
+ */
+export const Search = evolve(SearchV1, (e) => e.addState("owner", "rebuilds", arr(RebuildSchema)));
+
/**
* Unread answers (owners only): a session is unread while its latest root answer (`answers`) is newer than what an owner
* has read (`reads`), on any device. A browser marks a session read when its root conversation is shown.
diff --git a/apps/p2/plugins/search/server.ts b/apps/p2/plugins/search/server.ts
index 6c1bf10a7..e2ec17130 100644
--- a/apps/p2/plugins/search/server.ts
+++ b/apps/p2/plugins/search/server.ts
@@ -1,10 +1,10 @@
// The message index, the workers' feed into it, and owner search over sessions and messages.
-import { num, obj } from "@earendil-works/pi-schema";
-import { type Ctx, defineFacet, defineState, forbidden, isOwner, isWorker, record } from "@p2/app/v1";
-import { Hosts, Server, serverOf } from "@p2/app/v1/server";
+import { arr, num, obj, str } from "@earendil-works/pi-schema";
+import { defineFacet, defineJob, defineState, deriveTyped, forbidden, isOwner, isWorker, record } from "@p2/app/v1";
+import { Server, serverOf } from "@p2/app/v1/server";
import { ArchivedStarts, Catalog, type Session } from "../sessions/contract.ts";
-import { IndexScan, Search, SearchFeed, SearchIndex, SessionReads } from "./contract.ts";
+import { IndexScan, type Rebuild, RebuildSchema, Search, SearchFeed, SearchIndex, SessionReads } from "./contract.ts";
import { SearchDb } from "./db.ts";
import { ftsQuery } from "./query.ts";
@@ -19,11 +19,29 @@ const Markers = defineState("p2.search-index", {
initial: () => ({ answers: {}, reads: {} }),
});
+/** Pi Mobile's rebuildIndex as a job: a rebuild interrupted by a restart runs again (it starts over). */
+const RebuildJob = defineJob("p2.search.rebuild", {
+ version: 1,
+ args: obj({ session: str() }),
+ result: obj({}),
+ persist: { resume: "rerun" },
+});
+
+/** The latest failed rebuild per session, until it is rebuilt again. */
+const RebuildFailures = defineState("p2.search.rebuild-failures", {
+ schema: obj({ list: arr(RebuildSchema) }),
+ version: 1,
+ initial: () => ({ list: [] }),
+});
+
export default defineFacet((host) => {
host.serve(SearchIndex, (_params, self) => {
const server = self.get(Server, {});
const catalog = self.get(Catalog, {});
+ const archivedStarts = self.get(ArchivedStarts, {});
+ const natives = serverOf(self);
const markers = self.state(Markers);
+ const failures = self.state(RebuildFailures);
const dbResource = self.open(
async (ctx) => {
const db = new SearchDb(server.info.value.dataDir);
@@ -35,7 +53,48 @@ export default defineFacet((host) => {
},
(db) => db.close(),
);
+ // Stop the worker, reset the index, scan from a fresh worker (an archived one without resuming its work), and stop
+ // an archived session's worker again. This local object's `ctx.get` acts as the server, whom `IndexScan` serves.
+ const rebuilds = self.job(RebuildJob, {
+ run: async ({ session }, ctx) => {
+ const archived = (await catalog.get({ id: session }, ctx)).archivedAt !== null;
+ await natives().workers.stop(session);
+ dbResource.value.reset(session);
+ if (archived) await archivedStarts.allow({ session }, ctx);
+ try {
+ await (await ctx.get(IndexScan, { session })).scanned(ctx);
+ } finally {
+ if (archived) {
+ await archivedStarts.end({ session }, ctx);
+ await natives().workers.suspend(session);
+ }
+ }
+ return {};
+ },
+ finish: async (outcome, tx) => {
+ const draft = await tx.draft(failures);
+ draft.list = draft.list.filter((each) => each.session !== outcome.args.session);
+ if (outcome.kind === "failed")
+ draft.list.push({
+ session: outcome.args.session,
+ phase: "failed",
+ error: outcome.error.kind === "threw" ? outcome.error.error.message : "The rebuild was interrupted",
+ });
+ },
+ });
return {
+ rebuilds: deriveTyped(self, [rebuilds.running, failures], ([running, { list }]): Rebuild[] => [
+ ...running.flatMap((run) =>
+ run.phase === "blocked" ? [] : [{ session: run.args.session, phase: "running" as const }],
+ ),
+ ...list,
+ ]),
+ rebuild: async ({ session }, ctx) => {
+ // Refused at once for a session the catalog does not know.
+ await catalog.get({ id: session }, ctx);
+ await rebuilds.start({ session }, ctx, { key: session });
+ return null;
+ },
answers: self.derive([markers], ([value]) => value.answers),
reads: self.derive([markers], ([value]) => value.reads),
markRead: async ({ session, through }, ctx) => {
@@ -121,55 +180,36 @@ export default defineFacet((host) => {
};
});
- host.serve(Search, (_params, principal, self) => {
- if (!isOwner(principal)) throw forbidden("search is owner-only");
- const catalog = self.get(Catalog, {});
- const index = self.get(SearchIndex, {});
- const hosts = self.get(Hosts, {});
- const archivedStarts = self.get(ArchivedStarts, {});
- const server = serverOf(self);
- // Pi Mobile's rebuildIndex: stop the worker, reset the index, scan from a fresh worker (an archived one without
- // resuming its work), and stop an archived session's worker again.
- const rebuild = async (session: string, ctx: Ctx) => {
- const archived = (await catalog.get({ id: session }, ctx)).archivedAt !== null;
- await hosts.stop({ session }, ctx);
- await index.reset({ session }, ctx);
- if (archived) await archivedStarts.allow({ session }, ctx);
- try {
- // As the server: the worker answers only the server's own scan requests.
- await server().root(async (root) => {
- const scan = await root.get(IndexScan, { session });
- await scan.scanned(root);
- });
- } finally {
- if (archived) {
- await archivedStarts.end({ session }, ctx);
- await server().workers.suspend(session);
- }
- }
- };
- return {
- variant: "owner",
- members: {
- rebuildIndex: async ({ session }, ctx) => {
- await rebuild(session, ctx);
- return null;
+ host.serve(
+ Search,
+ (_params, principal, self) => {
+ if (!isOwner(principal)) throw forbidden("search is owner-only");
+ const catalog = self.get(Catalog, {});
+ const index = self.get(SearchIndex, {});
+ return {
+ variant: "owner",
+ members: {
+ rebuilds: index.rebuilds,
+ rebuildIndex: (args, ctx) => index.rebuild(args, ctx),
+ search: async ({ query, archived = false }, ctx) => {
+ const scope = catalog.sessions.value.filter((session) => (session.archivedAt !== null) === archived);
+ const needle = query.toLowerCase();
+ const sessions = scope
+ .filter((session) => `${session.title}\n${session.cwd}`.toLowerCase().includes(needle))
+ .sort(sidebarOrder)
+ .map((session) => session.id);
+ const titles = new Map(scope.map((session) => [session.id, session.title]));
+ const messages = query.trim()
+ ? await index.messages({ query, sessions: [...titles.keys()] }, ctx)
+ : [];
+ return {
+ sessions,
+ messages: messages.map((hit) => ({ ...hit, title: titles.get(hit.session) ?? "" })),
+ };
+ },
},
- search: async ({ query, archived = false }, ctx) => {
- const scope = catalog.sessions.value.filter((session) => (session.archivedAt !== null) === archived);
- const needle = query.toLowerCase();
- const sessions = scope
- .filter((session) => `${session.title}\n${session.cwd}`.toLowerCase().includes(needle))
- .sort(sidebarOrder)
- .map((session) => session.id);
- const titles = new Map(scope.map((session) => [session.id, session.title]));
- const messages = query.trim() ? await index.messages({ query, sessions: [...titles.keys()] }, ctx) : [];
- return {
- sessions,
- messages: messages.map((hit) => ({ ...hit, title: titles.get(hit.session) ?? "" })),
- };
- },
- },
- };
- });
+ };
+ },
+ { accepts: { 1: {} } },
+ );
});
diff --git a/apps/p2/plugins/search/tui.ts b/apps/p2/plugins/search/tui.ts
index f981d4240..481d3a3fc 100644
--- a/apps/p2/plugins/search/tui.ts
+++ b/apps/p2/plugins/search/tui.ts
@@ -1,8 +1,15 @@
// The session selector's message search in the TUI (p2-tui-port.md section 1.17): `Search` is literal and bounded, and
-// the owner's; the selector shows its hits as a separate group.
+// the owner's; the selector shows its hits as a separate group. `/reindex` rebuilds the viewed session's index.
-import { BACKGROUND_CONTEXT, contribute, defineFacet } from "@p2/app/v1";
-import { type TuiSessionSource, TuiSessionSources } from "@p2/app/v1/tui";
+import { BACKGROUND_CONTEXT, contribute, defineFacet, liveIn, whenLive } from "@p2/app/v1";
+import {
+ lineText,
+ type TuiCommand,
+ TuiCommands,
+ type TuiContext,
+ type TuiSessionSource,
+ TuiSessionSources,
+} from "@p2/app/v1/tui";
import { Search } from "./contract.ts";
const source: TuiSessionSource = {
@@ -14,8 +21,52 @@ const source: TuiSessionSource = {
}, BACKGROUND_CONTEXT),
};
+/**
+ * `/reindex`: re-indexes the viewed session's history (`Search.rebuildIndex`). Long work (a job on the server): a
+ * line says it runs at once, another when it is done or why it failed (`Search.rebuilds`).
+ */
+function reindex(ctx: TuiContext): void {
+ const session = ctx.identity.session;
+ if (!session) return;
+ const scope = ctx.scope();
+ let started = false;
+ const search = scope.watch(Search, {}, (ref, follow) => {
+ if (ref.variant !== "owner") return;
+ follow(ref.rebuilds, (rebuilds) => {
+ const rebuild = rebuilds.find((each) => each.session === session);
+ if (!started || rebuild?.phase === "running") return;
+ if (rebuild?.phase === "failed")
+ ctx.notify(`Search index rebuild failed: ${lineText(rebuild.error)}`, "error");
+ else ctx.notify("Search index rebuilt");
+ scope.dispose();
+ });
+ });
+ ctx.notify("Rebuilding the search index…");
+ void (async () => {
+ await whenLive(search);
+ const owner = liveIn(search, "owner");
+ if (!owner) throw new Error("Only the owner rebuilds the search index");
+ await owner.rebuildIndex({ session }, BACKGROUND_CONTEXT);
+ started = true;
+ })().catch((error: unknown) => {
+ ctx.fail(error);
+ scope.dispose();
+ });
+}
+
+const COMMANDS: Record = {
+ "p2.search.reindex": {
+ title: "Rebuild the session's search index",
+ category: "Owner",
+ when: "chat && owner && !overlay",
+ slash: { name: "reindex" },
+ run: (ctx) => reindex(ctx),
+ },
+};
+
export default defineFacet((host) =>
contribute(host, "p2.tui-search", (self) => {
self.contribute(TuiSessionSources, { search: source });
+ self.contribute(TuiCommands, COMMANDS);
}),
);
diff --git a/apps/p2/plugins/search/worker.ts b/apps/p2/plugins/search/worker.ts
index 1b79cc5ba..7bcb30ad1 100644
--- a/apps/p2/plugins/search/worker.ts
+++ b/apps/p2/plugins/search/worker.ts
@@ -24,7 +24,6 @@ import {
signalOf,
until,
} from "@p2/app/v1";
-import { durableContext } from "@p2/app/v1/node";
import { CommitSignal, harnessOf, topLevelRecords } from "@p2/app/v1/worker";
import { Subagents } from "../subagents/contract.worker.ts";
import { type IndexedMessage, IndexScan, SearchFeed } from "./contract.ts";
@@ -89,17 +88,11 @@ async function catchUp(
through: number,
ctx: Ctx,
): Promise {
- const context = durableContext(ctx);
- const upper = (await root.entries({}, 1, undefined, context)).items[0]?.id;
+ const upper = (await root.entries({}, 1, undefined, ctx)).items[0]?.id;
if (upper === undefined || upper <= through) return through;
let cursor: Cursor | undefined;
do {
- const page = await root.entries(
- { minEntryId: (through + 1) as EntryId, maxEntryId: upper },
- PAGE,
- cursor,
- context,
- );
+ const page = await root.entries({ minEntryId: (through + 1) as EntryId, maxEntryId: upper }, PAGE, cursor, ctx);
cursor = page.next;
const messages = page.items.map(indexedMessage).filter((message) => message !== undefined);
await feed.index({ conversation: root.id, messages, through: cursor === undefined ? upper : through }, ctx);
@@ -137,16 +130,15 @@ export default defineFacet((host) => {
while (!signalOf(ctx).aborted) {
const seen = signal.version;
try {
- const context = durableContext(ctx);
- const registry = await durable.snapshot(Subagents, ROOT_CONVERSATION_ID, context);
+ const registry = await durable.snapshot(Subagents, ROOT_CONVERSATION_ID, ctx);
// A branch's history starts after its fork point: what it inherits is indexed with the conversation
// storing it.
const starts = new Map(
- (await topLevelRecords(durable, context)).map((record) => [record.id, record.parent?.at ?? 0]),
+ (await topLevelRecords(durable, ctx)).map((record) => [record.id, record.parent?.at ?? 0]),
);
for (const agent of Object.values(registry?.agents ?? {})) starts.set(agent.conversationId, 0);
for (const [id, start] of starts) {
- const conversation = await durable.conversation(id as ConversationId, context);
+ const conversation = await durable.conversation(id as ConversationId, ctx);
if (!conversation) continue;
const through = await feed.cursor({ conversation: id }, ctx);
await catchUp(conversation, feed, Math.max(through, start), ctx);
diff --git a/apps/p2/plugins/sessions/BranchSheet.tsx b/apps/p2/plugins/sessions/BranchSheet.tsx
index 931b7d01b..e156ada7f 100644
--- a/apps/p2/plugins/sessions/BranchSheet.tsx
+++ b/apps/p2/plugins/sessions/BranchSheet.tsx
@@ -3,7 +3,10 @@ import { Sheet } from "@p2/app/v1/web";
import { useState } from "react";
import { PREVIEW_LENGTH } from "../tree/contract.ts";
-/** Renames a branch; `label` null shows `title` (conversation 1's default), and only conversation 1 may go back to it. */
+/**
+ * Renames a branch; `label` null shows `title` (conversation 1's default), and only conversation 1 may go back to it.
+ * Saving closes the sheet: the row shows the new label while the rename is in flight.
+ */
export function BranchSheet({
title,
conversation,
@@ -15,18 +18,17 @@ export function BranchSheet({
title: string;
conversation: number;
label: string | null;
- onSave: (label: string | null) => Promise;
+ onSave: (label: string | null) => void;
onClose: () => void;
}) {
const [draft, setDraft] = useState(label ?? title);
- const [error, setError] = useState();
const next = draft.replace(/\s+/g, " ").trim();
const resettable = conversation === 1;
const unchanged = next === (label ?? title);
- const save = (value: string | null) =>
- void onSave(value)
- .then(onClose)
- .catch((failure: unknown) => setError(failure instanceof Error ? failure.message : String(failure)));
+ const save = (value: string | null) => {
+ onSave(value);
+ onClose();
+ };
return (