feat(coding-agent): port the experimental client/server to pi-durable

Session workers now host a durable Harness over a per-session sqlite storage.
The server keeps each session in <sessionDir>/<id>/ with meta.json and
session.sqlite and never opens the storage itself. Workers hold retirement
while the Harness task graph has live tasks and resume interrupted work on
open.

Transcript serves the root conversation's durable view directly, Models
follows the conversation's agent document, and AgentController maps prompt,
steer, follow-up, queue cancellation, abort, and compaction onto the root
conversation, plus waitForPrompt. navigate, nextRun, and resume are dropped
(see experimental/services/README.md TODO). The durable TUI and the worker
share harness-setup.ts.

pi-server exports its own SessionMetadata and no longer depends on
pi-agent-core. Chord accepts readonly JSON types in service contracts without
walking them, which recursive JSON types like pi-ai's made too deep.
This commit is contained in:
Mario Zechner 2026-10-01 14:44:50 +02:00
parent 7fd478a2e8
commit 48dd1e2f0f
42 changed files with 1047 additions and 1368 deletions

View file

@ -130,14 +130,23 @@ export interface Service<T> {
readonly [SERVICE_TYPE]?: (value: T) => T;
}
/** JSON with readonly containers, such as other packages' recursive JSON types. */
type ReadonlyJsonValue =
| null
| boolean
| number
| string
| readonly ReadonlyJsonValue[]
| { readonly [key: string]: ReadonlyJsonValue };
// A type assignable to readonly JSON is valid as is. Walking a recursive JSON type instead, for example one with
// readonly arrays, would never terminate.
type InvalidJsonPart<T> = IsAny<T> extends true
? T
: unknown extends T
? never
: [T] extends [JsonValue]
? [JsonValue] extends [T]
? never
: InvalidJsonStructure<T>
: [T] extends [ReadonlyJsonValue]
? never
: InvalidJsonStructure<T>;
type InvalidJsonProperty<T> = [Exclude<T, undefined>] extends [never] ? T : InvalidJsonPart<Exclude<T, undefined>>;

View file

@ -1,5 +1,5 @@
import type { AgentMessage, Entry, LaneSnapshot } from "@earendil-works/pi-agent-core";
import type { AssistantMessage } from "@earendil-works/pi-ai";
import type { AssistantMessage, ToolResultMessage, UserMessage } from "@earendil-works/pi-ai";
import type { ConversationView, EntryRecord, InboxState, LiveState } from "@earendil-works/pi-durable";
import { Container, Spacer, Text, TruncatedText, type TUI } from "@earendil-works/pi-tui";
import { createAllToolRenderers } from "../core/tools/renderers/index.ts";
import { AssistantMessageComponent } from "../modes/interactive/components/assistant-message.ts";
@ -8,16 +8,20 @@ import { ToolExecutionComponent, type ToolRenderers } from "../modes/interactive
import { UserMessageComponent } from "../modes/interactive/components/user-message.ts";
import { theme } from "../modes/interactive/theme/theme.ts";
function userMessageText(message: AgentMessage): string {
if (message.role !== "user") return "";
if (typeof message.content === "string") return message.content;
return message.content
.filter((content) => content.type === "text")
.map((content) => content.text)
/** The `pi.live` document of a view: the active run, the streaming answer, and running tools. */
export function liveOf(view: ConversationView): LiveState {
return (view.docs["pi.live"] ?? {}) as LiveState;
}
function userText(content: UserMessage["content"]): string {
if (typeof content === "string") return content;
return content
.filter((block) => block.type === "text")
.map((block) => block.text)
.join("");
}
/** Snapshot-driven transcript used by the service-only experimental presentation. */
/** Renders the root conversation's durable view for the service-only experimental presentation. */
export class ExperimentalChatView {
static readonly #renderers: Record<string, ToolRenderers> = createAllToolRenderers();
@ -26,128 +30,179 @@ export class ExperimentalChatView {
readonly status = new Container();
readonly #ui: TUI;
readonly #cwd: string;
/** The newest card per call ID; provider call IDs may repeat across turns. */
readonly #tools = new Map<string, ToolExecutionComponent>();
#renderedEntryIds: string[] = [];
/** Every card shown, also older ones whose call ID a later turn reused. */
readonly #cards: ToolExecutionComponent[] = [];
/** Call IDs whose cards the streaming answer created; its entry takes them over. */
readonly #streamingCalls = new Set<string>();
#renderedEntryIds: number[] = [];
#streaming: AssistantMessageComponent | undefined;
#indicator: StatusIndicator | undefined;
#working = false;
#statusText = "";
constructor(ui: TUI, cwd: string) {
this.#ui = ui;
this.#cwd = cwd;
}
apply(snapshot: LaneSnapshot): void {
this.#syncTranscript(snapshot.transcript);
this.#syncStreaming(snapshot.operation?.streamingMessage);
for (const tool of snapshot.operation?.runningTools ?? []) {
const component = this.#tool(tool.toolName, tool.toolCallId, tool.args);
if (tool.status === "running") {
component.markExecutionStarted();
if (tool.result !== undefined) component.updateResult({ ...tool.result, isError: false }, true);
} else {
component.updateResult({ ...tool.result, isError: tool.isError }, false);
apply(view: ConversationView): void {
const live = liveOf(view);
this.#syncTranscript(view.entries);
const message = live.generation?.message as AssistantMessage | undefined;
// A partial without its entry was dropped, for example by a retry: render the transcript again.
if (message === undefined && this.#streaming !== undefined) this.#rebuild(view.entries);
if (message !== undefined) this.#syncStreaming(message);
for (const slot of live.tools ?? []) {
if (slot.status === "pending") continue;
const component = this.#tool(slot.name, slot.callId);
component.setArgsComplete();
if (slot.status !== "running") continue;
component.markExecutionStarted();
if (slot.output !== undefined) {
component.updateResult(
{ content: [{ type: "text", text: slot.output }], details: slot.details, isError: false },
true,
);
}
}
this.#syncQueues(snapshot.queues);
this.#setWorking(snapshot.operation !== null);
this.#syncQueue((view.docs["pi.inbox"] ?? { items: [] }) as InboxState);
this.#syncStatus(live);
this.transcript.invalidate();
this.pendingMessages.invalidate();
this.status.invalidate();
}
refreshTheme(snapshot: LaneSnapshot): void {
refreshTheme(view: ConversationView): void {
this.#indicator?.dispose();
this.#indicator = undefined;
this.#working = false;
this.transcript.clear();
this.pendingMessages.clear();
this.#statusText = "";
this.status.clear();
this.#tools.clear();
this.#renderedEntryIds = [];
this.#streaming = undefined;
this.apply(snapshot);
this.#rebuild(view.entries);
this.apply(view);
}
dispose(): void {
this.#discardTools();
this.#indicator?.dispose();
}
#syncQueues(queues: LaneSnapshot["queues"]): void {
/** Finish every card: a running bash card keeps a timer until it gets a final result. */
#discardTools(): void {
for (const component of this.#cards) component.updateResult({ content: [], isError: false }, false);
this.#cards.length = 0;
this.#tools.clear();
this.#streamingCalls.clear();
}
#syncQueue(inbox: InboxState): void {
this.pendingMessages.clear();
for (const item of queues) {
for (const item of inbox.items) {
const text =
item.type === "message" ? userMessageText(item.message).replace(/\s+/g, " ") : `<${item.customType}>`;
this.pendingMessages.addChild(new TruncatedText(theme.fg("muted", `[${item.kind}] ${text}`), 1, 0));
item.mode === "write"
? `<${String(item.entry.kind)}>`
: userText(item.content as UserMessage["content"]).replace(/\s+/g, " ");
this.pendingMessages.addChild(new TruncatedText(theme.fg("muted", `[${item.mode}] ${text}`), 1, 0));
}
}
#syncTranscript(transcript: readonly Entry[]): void {
const diverged = this.#renderedEntryIds.some((id, index) => transcript[index]?.id !== id);
if (diverged) {
this.transcript.clear();
this.#tools.clear();
this.#renderedEntryIds = [];
this.#streaming = undefined;
#syncStatus(live: LiveState): void {
const generation = live.generation;
const compaction = live.compactions?.[0];
const runningTool = live.tools?.find((slot) => slot.status === "running");
let text = "";
if (generation?.retry !== undefined) {
text = `Retrying (attempt ${generation.attempt + 1}): ${generation.retry.error}`;
} else if (generation?.deferred !== undefined) text = "Waiting for deferred response...";
else if (compaction !== undefined) {
text = compaction.retry
? `Retrying ${compaction.reason} compaction (attempt ${compaction.attempt + 1})...`
: `Compacting (${compaction.reason})...`;
} else if (runningTool !== undefined) text = `Running ${runningTool.name}... (esc to abort)`;
else if (live.run !== undefined) text = "Working... (esc to abort)";
if (text === this.#statusText) return;
this.#statusText = text;
this.#indicator?.dispose();
this.#indicator = undefined;
this.status.clear();
if (text.length > 0) {
this.#indicator = new WorkingStatusIndicator(this.#ui, text);
this.status.addChild(this.#indicator);
}
for (const entry of transcript.slice(this.#renderedEntryIds.length)) {
}
#syncTranscript(entries: readonly EntryRecord[]): void {
// Compaction and resets replace the head of the active transcript.
if (this.#renderedEntryIds.some((id, index) => entries[index]?.id !== id)) this.#rebuild(entries);
for (const entry of entries.slice(this.#renderedEntryIds.length)) {
this.#addEntry(entry);
this.#renderedEntryIds.push(entry.id);
}
}
#addEntry(entry: Entry): void {
if (entry.type === "compaction") {
this.#addText(theme.fg("muted", `[compaction] compacted from ${entry.tokensBefore} tokens`));
for (const retained of entry.retainedTail) this.#addMessage(retained);
return;
#rebuild(entries: readonly EntryRecord[]): void {
this.transcript.clear();
this.#discardTools();
this.#renderedEntryIds = [];
this.#streaming = undefined;
for (const entry of entries) {
this.#addEntry(entry);
this.#renderedEntryIds.push(entry.id);
}
if (entry.type === "branch_summary") {
this.#addText(theme.fg("muted", "[branch summary]"));
this.#addText(entry.summary);
return;
}
if (entry.type === "custom") {
this.#addText(theme.fg("muted", `[${entry.customType}]`));
return;
}
this.#addMessage(entry.message);
}
#addMessage(message: AgentMessage): void {
if (message.role === "user") {
#addEntry(entry: EntryRecord): void {
const message = entry.model?.[0];
if (entry.kind === "pi.user" && message?.role === "user") {
this.transcript.addChild(new Spacer(1));
this.transcript.addChild(new UserMessageComponent(userMessageText(message)));
return;
}
if (message.role === "assistant") {
this.transcript.addChild(new UserMessageComponent(userText(message.content)));
} else if (entry.kind === "pi.assistant" && message?.role === "assistant") {
const component = this.#streaming ?? new AssistantMessageComponent();
if (!this.#streaming) this.transcript.addChild(component);
if (this.#streaming === undefined) this.transcript.addChild(component);
this.#streaming = undefined;
component.updateContent(message, false);
// Only a tool-calling answer runs its calls; an aborted, failed, or truncated one never does.
const ran = message.stopReason === "toolUse";
for (const content of message.content) {
if (content.type === "toolCall") this.#tool(content.name, content.id, content.arguments).setArgsComplete();
if (content.type !== "toolCall") continue;
const streamed = this.#streamingCalls.has(content.id);
if (!ran && !streamed) continue;
const card = this.#tool(content.name, content.id, content.arguments, !streamed);
card.setArgsComplete();
if (!ran) {
const text = "Not run: the answer was interrupted.";
card.updateResult({ content: [{ type: "text", text }], isError: true }, false);
}
}
return;
this.#streamingCalls.clear();
} else if (entry.kind === "pi.tool-result" && message?.role === "toolResult") {
const result = message as ToolResultMessage;
this.#tool(result.toolName, result.toolCallId).updateResult(result);
} else if (entry.kind === "pi.compaction") {
this.#addText(theme.fg("muted", "[compaction]"));
if (message?.role === "user") this.#addText(userText(message.content));
} else if (entry.kind === "pi.reset") {
this.#addText(theme.fg("muted", "[new context]"));
}
if (message.role === "toolResult") this.#tool(message.toolName, message.toolCallId).updateResult(message);
}
#syncStreaming(message: AssistantMessage | undefined): void {
if (!message) return;
if (!this.#streaming) {
#syncStreaming(message: AssistantMessage): void {
if (this.#streaming === undefined) {
this.#streaming = new AssistantMessageComponent();
this.transcript.addChild(this.#streaming);
}
this.#streaming.updateContent(message, true);
for (const content of message.content) {
if (content.type === "toolCall") this.#tool(content.name, content.id, content.arguments);
if (content.type !== "toolCall") continue;
this.#tool(content.name, content.id, content.arguments, !this.#streamingCalls.has(content.id));
this.#streamingCalls.add(content.id);
}
}
#tool(toolName: string, toolCallId: string, args?: unknown): ToolExecutionComponent {
const existing = this.#tools.get(toolCallId);
if (existing) {
/** The card of a call; `fresh` starts a new one for a call ID an earlier turn used. */
#tool(toolName: string, toolCallId: string, args?: unknown, fresh = false): ToolExecutionComponent {
const existing = fresh ? undefined : this.#tools.get(toolCallId);
if (existing !== undefined) {
if (args !== undefined) existing.updateArgs(args);
return existing;
}
@ -161,6 +216,7 @@ export class ExperimentalChatView {
this.#cwd,
);
this.transcript.addChild(component);
this.#cards.push(component);
this.#tools.set(toolCallId, component);
return component;
}
@ -169,16 +225,4 @@ export class ExperimentalChatView {
this.transcript.addChild(new Spacer(1));
this.transcript.addChild(new Text(text, 1, 0));
}
#setWorking(working: boolean): void {
if (working === this.#working) return;
this.#working = working;
this.#indicator?.dispose();
this.#indicator = undefined;
this.status.clear();
if (working) {
this.#indicator = new WorkingStatusIndicator(this.#ui, "Working... (esc to abort)");
this.status.addChild(this.#indicator);
}
}
}

View file

@ -9,6 +9,7 @@ import {
type LoadedFacets,
} from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { AgentState, ConversationView } from "@earendil-works/pi-durable";
import {
CombinedAutocompleteProvider,
type Component,
@ -30,7 +31,7 @@ import { getEditorTheme, setRegisteredThemes, stopThemeWatcher, theme } from "..
import { InteractiveThemeController } from "../modes/interactive/theme/theme-controller.ts";
import { createInteractiveTui } from "../modes/interactive/tui-renderer.ts";
import { type OpenClientRuntimeOptions, openClientRuntime } from "./client-runtime.ts";
import { ExperimentalChatView } from "./client-tui-chat.ts";
import { ExperimentalChatView, liveOf } from "./client-tui-chat.ts";
import { createPresentationFacetLoaders } from "./plugins/bundled.ts";
import { AgentController, type AgentOperationResponse, type AgentQueueResponse } from "./services/agent-controller.ts";
import type {
@ -220,8 +221,8 @@ export class ExperimentalClientTui implements Component {
}
refreshTheme(): void {
const snapshot = this.#laneSnapshot();
if (snapshot !== undefined) this.#chatView?.refreshTheme(snapshot);
const view = this.#conversationView();
if (view !== undefined) this.#chatView?.refreshTheme(view);
this.#rebuild();
}
@ -499,13 +500,12 @@ export class ExperimentalClientTui implements Component {
this.#documentContainer.addChild(view.transcript);
this.#pendingMessagesContainer.addChild(view.pendingMessages);
this.#laneUnsubscribe = feature.transcript.state.subscribe((value) => {
if (value.snapshot === null) return;
view.apply(value.snapshot);
view.apply(value);
this.#rebuild();
});
if (feature.transcript.state.value?.snapshot === null || feature.transcript.state.value?.snapshot === undefined) {
if (feature.transcript.state.value === undefined) {
await this.#closeLane();
throw new Error("Transcript has no initialized snapshot");
throw new Error("Transcript has no initialized view");
}
}
@ -563,8 +563,8 @@ export class ExperimentalClientTui implements Component {
async #submitPrompt(prompt: string): Promise<void> {
const controller = this.#selectedController();
if (controller === undefined) throw new Error("No Session AgentController service is available");
const operation = this.#laneSnapshot()?.operation;
const running = operation !== null && operation !== undefined;
const view = this.#conversationView();
const running = view !== undefined && liveOf(view).run !== undefined;
this.#status = running ? "Queueing steering message…" : "Running turn…";
this.#rebuild();
if (running) this.#reportQueue(await controller.steer({ message: prompt, images: null }, BACKGROUND_CONTEXT));
@ -585,11 +585,7 @@ export class ExperimentalClientTui implements Component {
}
#reportOperation(response: AgentOperationResponse): void {
this.#status = response.accepted
? response.error === null
? ""
: `Operation failed: ${response.error.message}`
: `Operation rejected: ${response.error.message}`;
this.#status = response.accepted ? "" : `Operation rejected: ${response.error.message}`;
this.#rebuild();
}
@ -599,30 +595,38 @@ export class ExperimentalClientTui implements Component {
}
#interrupt(): void {
const operation = this.#laneSnapshot()?.operation;
const view = this.#conversationView();
const controller = this.#selectedController();
if (operation === null || operation === undefined || controller === undefined) return;
this.#status = `Aborting ${operation.id}…`;
if (view === undefined || liveOf(view).run === undefined || controller === undefined) return;
this.#status = "Aborting…";
this.#rebuild();
void controller.requestAbort(operation.id, BACKGROUND_CONTEXT).catch((error: unknown) => {
this.#status = `Error: ${message(error)}`;
this.#rebuild();
});
void controller.abort(BACKGROUND_CONTEXT).then(
() => {
if (this.#status !== "Aborting…") return;
this.#status = "";
this.#rebuild();
},
(error: unknown) => {
this.#status = `Error: ${message(error)}`;
this.#rebuild();
},
);
}
#selectedController(): AgentController | undefined {
return this.#controller;
}
#laneSnapshot() {
const snapshot = this.#session?.transcript.state.value?.snapshot;
return snapshot === null ? undefined : snapshot;
#conversationView(): ConversationView | undefined {
return this.#session?.transcript.state.value;
}
#footer(): string {
const snapshot = this.#laneSnapshot();
if (!snapshot) return "/model · /thinking · /compact · /reload";
return `${snapshot.configuration.model.provider}/${snapshot.configuration.model.modelId} · thinking:${snapshot.configuration.thinkingLevel} · ${snapshot.stats.messageCount} messages · /model · /thinking · /compact · /reload`;
const view = this.#conversationView();
if (!view) return "/model · /thinking · /compact · /reload";
const agent = (view.docs["pi.agent"] ?? {}) as AgentState;
const model = agent.model === undefined ? "no model" : `${agent.model.provider}/${agent.model.modelId}`;
return `${model} · thinking:${agent.thinkingLevel ?? "off"} · ${view.entries.length} entries · /model · /thinking · /compact · /reload`;
}
}

View file

@ -1,9 +1,7 @@
import { resolve } from "node:path";
import { BACKGROUND_CONTEXT, type LaneWatchEvent } from "@earendil-works/pi-agent-core";
import type { AssistantMessage } from "@earendil-works/pi-ai";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { ClientCommand } from "../cli/experimental/commands/client.ts";
import { activateBuiltinClientServices, openClientRuntime } from "./client-runtime.ts";
import type { AgentOperationResponse } from "./services/agent-controller.ts";
import type { SessionAddress } from "./services/sessions.ts";
export type ClientResult =
@ -17,8 +15,6 @@ export type ClientResult =
export interface RunClientOptions {
/** Directory searched when --connect is omitted. Defaults to PI_SERVER_DIR or ~/.pi/server. */
readonly directory?: string;
/** Receives snapshot-ordered main-lane events while a prompt is active. */
readonly onEvent?: (event: LaneWatchEvent) => void | Promise<void>;
}
/** Discover servers, then list Sessions, attach to one, or create one for a prompt. */
@ -79,61 +75,12 @@ export async function runClient(command: ClientCommand, options: RunClientOption
return { kind: "attached", serverId: match.route.serverId, sessionId };
}
const agent = match.agent;
const completedText = new Map<string, string>();
const operationBoundaries = new Set<string>();
const boundaryWaiters = new Map<string, () => void>();
let deliveryTail = Promise.resolve();
const unsubscribe = match.transcript.state.subscribe((value, _context, delivery) => {
if (delivery.kind !== "update" || value.event === null) return;
const event = value.event;
deliveryTail = deliveryTail.then(async () => {
if (event.type === "message_end" && event.runId !== undefined && event.message.role === "assistant") {
completedText.set(event.runId, messageText(event.message));
}
await options.onEvent?.(event);
});
if (event.type === "run_end" || event.type === "run_suspend") {
operationBoundaries.add(event.runId);
boundaryWaiters.get(event.runId)?.();
boundaryWaiters.delete(event.runId);
}
});
if (match.transcript.state.value?.snapshot === null || match.transcript.state.value?.snapshot === undefined) {
unsubscribe();
throw new Error("Transcript has no initialized snapshot");
}
let response: AgentOperationResponse;
try {
response = await agent.prompt({ message: command.prompt, images: null }, BACKGROUND_CONTEXT);
// The operation response and transcript updates use independent protocol
// messages, so the response can arrive before its terminal event.
if (response.accepted) {
const operationId = response.operationId;
if (!operationBoundaries.has(operationId)) {
await new Promise<void>((resolve) => boundaryWaiters.set(operationId, resolve));
}
}
} finally {
unsubscribe();
await deliveryTail;
}
const response = await match.agent.prompt({ message: command.prompt, images: null }, BACKGROUND_CONTEXT);
if (!response.accepted) throw new Error(response.error.message);
if (response.error !== null) throw new Error(response.error.message);
return {
kind: "prompted",
serverId: match.route.serverId,
sessionId,
text: completedText.get(response.operationId) ?? "",
};
const result = await match.agent.waitForPrompt(response.operationId, BACKGROUND_CONTEXT);
if (result.status === "unanswered") throw new Error(`Prompt was not answered: ${result.reason}`);
return { kind: "prompted", serverId: match.route.serverId, sessionId, text: result.text };
} finally {
await runtime.dispose();
}
}
function messageText(message: AssistantMessage): string {
return message.content
.filter((content) => content.type === "text")
.map((content) => content.text)
.join("");
}

View file

@ -69,21 +69,13 @@ async function runClientCommand(command: ClientCommand): Promise<void> {
await runClientTui(command);
return;
}
let streamedText = false;
const result = await runClient(command, {
onEvent(event) {
if (event.type !== "message_update" || event.frame?.type !== "text_delta") return;
streamedText = true;
process.stdout.write(event.frame.delta);
},
});
const result = await runClient(command);
if (result.kind === "attached") {
console.log(`${result.serverId}\t${result.sessionId}\tattached`);
return;
}
if (result.kind === "prompted") {
if (streamedText) process.stdout.write("\n");
else console.log(result.text);
console.log(result.text);
return;
}
for (const session of result.sessions) console.log(`${session.serverId}\t${session.sessionId}`);

View file

@ -0,0 +1,123 @@
import type { Context } from "@earendil-works/chord";
import type { ModelThinkingLevel } from "@earendil-works/pi-ai";
import {
createRegistry,
type EnvTarget,
type HarnessSettings,
type ModelRef,
type Registry,
} from "@earendil-works/pi-durable";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { applyHttpProxySettings, configureHttpDispatcher } from "../../core/http-dispatcher.ts";
import { findInitialModel, resolveCliModel } from "../../core/model-resolver.ts";
import type { ModelRuntime } from "../../core/model-runtime.ts";
import type { SettingsManager } from "../../core/settings-manager.ts";
import { createPiPrompt } from "./prompt.ts";
/** pi's HTTP setup: proxy, idle timeouts, and one undici for fetch. Without it, some provider streams break off. */
export function configureHarnessHttp(settingsManager: SettingsManager): void {
applyHttpProxySettings(settingsManager.getGlobalSettings().httpProxy);
configureHttpDispatcher(settingsManager.getHttpIdleTimeoutMs());
}
/** Harness settings read at every use from pi's settings as loaded at startup. */
export function createHarnessSettings(settingsManager: SettingsManager): HarnessSettings {
return {
get stream() {
const provider = settingsManager.getProviderRetrySettings();
const idle = settingsManager.getHttpIdleTimeoutMs();
return {
timeoutMs: provider.timeoutMs ?? (idle === 0 ? 2147483647 : idle),
maxRetryDelayMs: provider.maxRetryDelayMs,
...(provider.maxRetries === undefined ? {} : { maxRetries: provider.maxRetries }),
};
},
get compaction() {
return settingsManager.getCompactionSettings();
},
get retry() {
return settingsManager.getRetrySettings();
},
get steeringMode() {
return settingsManager.getSteeringMode();
},
get followUpMode() {
return settingsManager.getFollowUpMode();
},
};
}
/** A registry with pi's coding tools and system prompt. */
export function createCodingRegistry(settingsManager: SettingsManager, cwd: string): Registry {
const registry = createRegistry();
registry.install(CodingTools);
registry.install(createPiPrompt(settingsManager, cwd));
return registry;
}
/** One execution environment per directory, shared by every conversation in it. */
export class ExecutionEnvs {
readonly #defaultCwd: string;
readonly #envs = new Map<string, NodeExecutionEnv>();
constructor(defaultCwd: string) {
this.#defaultCwd = defaultCwd;
}
readonly env = ({ cwd = this.#defaultCwd }: EnvTarget): NodeExecutionEnv => {
let env = this.#envs.get(cwd);
if (env === undefined) {
env = new NodeExecutionEnv({ cwd });
this.#envs.set(cwd, env);
}
return env;
};
async cleanup(context: Context): Promise<void> {
const envs = [...this.#envs.values()];
this.#envs.clear();
for (const env of envs) await env.cleanup(context);
}
}
export interface InitialModel {
readonly model?: ModelRef;
readonly thinkingLevel?: ModelThinkingLevel;
readonly fallbackMessage?: string;
}
/** The model a new root conversation starts with: an explicit `--provider`/`--model`, or pi's default resolution. */
export async function findInitialAgentModel(
settingsManager: SettingsManager,
modelRuntime: ModelRuntime,
cli?: { readonly provider?: string; readonly model: string },
): Promise<InitialModel> {
if (cli !== undefined) {
const resolved = resolveCliModel({ cliProvider: cli.provider, cliModel: cli.model, modelRuntime });
if (resolved.error !== undefined || resolved.model === undefined) {
throw new Error(`Could not resolve model: ${resolved.error ?? cli.model}`);
}
return {
model: { provider: resolved.model.provider, modelId: resolved.model.id },
...(resolved.thinkingLevel === undefined ? {} : { thinkingLevel: resolved.thinkingLevel }),
};
}
const initial = await findInitialModel({
scopedModels: [],
isContinuing: false,
defaultProvider: settingsManager.getDefaultProvider(),
defaultModelId: settingsManager.getDefaultModel(),
defaultThinkingLevel: settingsManager.getDefaultThinkingLevel(),
modelRuntime,
});
return {
...(initial.model === undefined
? {}
: {
model: { provider: initial.model.provider, modelId: initial.model.id },
thinkingLevel: initial.thinkingLevel,
}),
...(initial.fallbackMessage === undefined ? {} : { fallbackMessage: initial.fallbackMessage }),
};
}

View file

@ -7,23 +7,23 @@ import {
type ConversationId,
type ConversationView,
type Cursor,
createRegistry,
type EntryRecord,
Harness,
type HarnessSettings,
type ModelRef,
ROOT_CONVERSATION_ID,
type Submission,
type TaskGraph,
} from "@earendil-works/pi-durable";
import { NodeExecutionEnv } from "@earendil-works/pi-durable/env/node";
import { openNodeSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/node";
import { CodingTools } from "@earendil-works/pi-durable/tools";
import { applyHttpProxySettings, configureHttpDispatcher } from "../../core/http-dispatcher.ts";
import { findInitialModel } from "../../core/model-resolver.ts";
import { ModelRuntime } from "../../core/model-runtime.ts";
import { SettingsManager } from "../../core/settings-manager.ts";
import { createPiPrompt } from "./prompt.ts";
import {
configureHarnessHttp,
createCodingRegistry,
createHarnessSettings,
ExecutionEnvs,
findInitialAgentModel,
} from "./harness-setup.ts";
import { selectSession } from "./sessions.ts";
import { Subagent } from "./subagent.ts";
@ -123,41 +123,14 @@ function titleOf(entry: EntryRecord | undefined): { title?: string } {
export async function openDurable(options: OpenDurableOptions = {}): Promise<OpenDurableResult> {
const location = await selectSession(options.cwd ?? process.cwd(), options.continueSession ?? false);
const envs = new Map<string, NodeExecutionEnv>();
const envs = new ExecutionEnvs(location.cwd);
let harness: Harness | undefined;
try {
const modelRuntime = await ModelRuntime.create();
const settingsManager = SettingsManager.create(location.cwd);
// pi's HTTP setup: proxy, idle timeouts, and one undici for fetch. Without it, some provider streams break off.
applyHttpProxySettings(settingsManager.getGlobalSettings().httpProxy);
configureHttpDispatcher(settingsManager.getHttpIdleTimeoutMs());
// Read at every use from pi's settings as loaded at startup.
const settings: HarnessSettings = {
get stream() {
const provider = settingsManager.getProviderRetrySettings();
const idle = settingsManager.getHttpIdleTimeoutMs();
return {
timeoutMs: provider.timeoutMs ?? (idle === 0 ? 2147483647 : idle),
maxRetryDelayMs: provider.maxRetryDelayMs,
...(provider.maxRetries === undefined ? {} : { maxRetries: provider.maxRetries }),
};
},
get compaction() {
return settingsManager.getCompactionSettings();
},
get retry() {
return settingsManager.getRetrySettings();
},
get steeringMode() {
return settingsManager.getSteeringMode();
},
get followUpMode() {
return settingsManager.getFollowUpMode();
},
};
const registry = createRegistry();
registry.install(CodingTools);
registry.install(createPiPrompt(settingsManager, location.cwd));
configureHarnessHttp(settingsManager);
const settings = createHarnessSettings(settingsManager);
const registry = createCodingRegistry(settingsManager, location.cwd);
registry.install(Subagent);
const pendingReports: unknown[] = [];
@ -168,38 +141,17 @@ export async function openDurable(options: OpenDurableOptions = {}): Promise<Ope
models: modelRuntime,
registry,
settings,
// One environment per directory, shared by every conversation in it.
env: ({ cwd = location.cwd }) => {
let env = envs.get(cwd);
if (env === undefined) {
env = new NodeExecutionEnv({ cwd });
envs.set(cwd, env);
}
return env;
},
env: envs.env,
onReport: (error) => report(error),
},
context,
);
const initial = location.created
? await findInitialModel({
scopedModels: [],
isContinuing: false,
defaultProvider: settingsManager.getDefaultProvider(),
defaultModelId: settingsManager.getDefaultModel(),
defaultThinkingLevel: settingsManager.getDefaultThinkingLevel(),
modelRuntime,
})
: undefined;
const initial = location.created ? await findInitialAgentModel(settingsManager, modelRuntime) : undefined;
const root = await harness.root(context, {
agent: {
cwd: location.cwd,
...(initial?.model === undefined
? {}
: {
model: { provider: initial.model.provider, modelId: initial.model.id },
thinkingLevel: initial.thinkingLevel,
}),
...(initial?.model === undefined ? {} : { model: initial.model }),
...(initial?.thinkingLevel === undefined ? {} : { thinkingLevel: initial.thinkingLevel }),
},
});
const label = (id: ConversationId): string => (id === root.id ? "main" : `subagent ${id}`);
@ -399,7 +351,7 @@ export async function openDurable(options: OpenDurableOptions = {}): Promise<Ope
try {
// Close writes no outcome: a running turn resumes with --continue.
await opened.close(context);
for (const env of envs.values()) await env.cleanup(context);
await envs.cleanup(context);
} finally {
await location.release();
}

View file

@ -1,10 +1,10 @@
export {
type AgentCompactionRequest,
AgentController,
type AgentNavigationRequest,
type AgentOperationError,
type AgentOperationResponse,
type AgentPromptRequest,
type AgentPromptResult,
type AgentQueueResponse,
} from "./services/agent-controller.ts";
export { type PresentationSelectItem, PresentationUI } from "./services/presentation-ui.ts";

View file

@ -3,14 +3,8 @@ import { chmod, lstat, mkdir, readFile, writeFile } from "node:fs/promises";
import { homedir } from "node:os";
import { isAbsolute, join } from "node:path";
import type { Context } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { FacetBundleArtifact } from "@earendil-works/chord/node";
import {
BACKGROUND_CONTEXT,
type JsonlSessionMetadata,
JsonlSessionRepo,
TODO_CONTEXT,
} from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import { Client, ServerError as ClientServerError, DisconnectedError } from "@earendil-works/pi-client";
import { createUnixTransportFactory, type UnixServerRoute } from "@earendil-works/pi-client/unix";
import { isServerId, type ServerId } from "@earendil-works/pi-protocol";
@ -18,7 +12,6 @@ import {
ServerError as RoutedServerError,
type Server,
type ServerHost,
SessionAmbiguousError,
SessionNotFoundError,
} from "@earendil-works/pi-server";
import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix";
@ -46,6 +39,13 @@ import { RadiusRelayAuthResolver } from "./radius-auth.ts";
import { RadiusRelayHost, type RadiusRelayHostStatus } from "./radius-relay.ts";
import { createExperimentalServerServices } from "./services/server.ts";
import type { SessionCreateOptions, SessionSummary } from "./services/sessions.ts";
import {
createSession as createCatalogSession,
deleteSession,
listSessions as listCatalogSessions,
readSession,
type SessionCatalogMetadata,
} from "./session-catalog.ts";
import { SessionPluginSelectionConflictError, SessionWorkerManager } from "./session-worker-manager.ts";
export const ENV_SERVER_DIR = "PI_SERVER_DIR";
@ -361,11 +361,11 @@ interface StartServerBackendOptions {
readonly serverId: ServerId;
readonly sessionDir?: string;
resolveSessionPlugins(
metadata: JsonlSessionMetadata,
metadata: SessionCatalogMetadata,
packagePaths: readonly string[] | undefined,
context: Context,
): Promise<ResolvedSessionPlugins>;
removeSessionPlugins(metadata: JsonlSessionMetadata): Promise<void>;
removeSessionPlugins(metadata: SessionCatalogMetadata): Promise<void>;
reloadPresentationFacetBundles(packagePaths: readonly string[]): Promise<readonly FacetBundleArtifact[]>;
}
@ -380,51 +380,35 @@ async function startServerBackend(
): Promise<RunningServerBackend> {
const serverId = options.serverId;
const sessionDir = resolveSessionDirectory(options.sessionDir);
const executionEnv = new NodeExecutionEnv({ cwd: process.cwd() });
const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir });
const listSessions = async (context: Context): Promise<JsonlSessionMetadata[]> => {
const sessions = new Map((await repo.list(undefined, context)).map((metadata) => [metadata.path, metadata]));
const listSessions = async (): Promise<SessionCatalogMetadata[]> => {
const sessions = new Map((await listCatalogSessions(sessionDir)).map((metadata) => [metadata.path, metadata]));
for (const metadata of workers.trackedSessions) sessions.set(metadata.path, metadata);
return [...sessions.values()];
};
const resolveSession = async (sessionId: string, context: Context): Promise<JsonlSessionMetadata> => {
const matches = (await listSessions(context)).filter((metadata) => metadata.id === sessionId);
if (matches.length === 0) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
if (matches.length > 1) throw new SessionAmbiguousError();
return matches[0]!;
const resolveSession = async (sessionId: string, _context: Context): Promise<SessionCatalogMetadata> => {
const metadata =
workers.trackedSessions.find((candidate) => candidate.id === sessionId) ??
(await readSession(sessionDir, sessionId));
if (metadata === undefined) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
return metadata;
};
const createSession = async (
createOptions: SessionCreateOptions,
context: Context,
): Promise<JsonlSessionMetadata> => {
const session = await repo.create({ ...createOptions, cwd: process.cwd() }, context);
try {
return session.metadata;
} finally {
await session.close(context);
}
};
const summarize = (metadata: JsonlSessionMetadata): SessionSummary => ({
const createSession = (createOptions: SessionCreateOptions): Promise<SessionCatalogMetadata> =>
createCatalogSession(sessionDir, { ...createOptions, cwd: process.cwd() });
const summarize = (metadata: SessionCatalogMetadata): SessionSummary => ({
serverId,
sessionId: metadata.id,
createdAt: metadata.createdAt,
});
const closeStorage = async (): Promise<void> => {
const cleanup = await Promise.allSettled([repo.close(TODO_CONTEXT), executionEnv.cleanup(TODO_CONTEXT)]);
const errors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Experimental session storage cleanup failed");
};
const serverServices = await createExperimentalServerServices({
list: async (context) =>
(await listSessions(context))
list: async () =>
(await listSessions())
.map(summarize)
.sort((left, right) => left.sessionId.localeCompare(right.sessionId) || left.createdAt - right.createdAt),
create: async (createOptions, context) => summarize(await createSession(createOptions, context)),
create: async (createOptions) => summarize(await createSession(createOptions)),
remove: async (sessionId, context) => {
const metadata = await resolveSession(sessionId, context);
await workers.closeSession(metadata, context);
await repo.delete(metadata, context);
await deleteSession(metadata);
await options.removeSessionPlugins(metadata);
},
async prepareSessionPlugins(sessionId, packagePaths, context) {
@ -447,15 +431,8 @@ async function startServerBackend(
async reloadPresentationPlugins(packagePaths) {
return createPresentationFacetData(await options.reloadPresentationFacetBundles(packagePaths));
},
}).catch(async (error: unknown) => {
try {
await closeStorage();
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Server service startup and cleanup failed");
}
throw error;
});
const host: ServerHost<JsonlSessionMetadata> = {
const host: ServerHost<SessionCatalogMetadata> = {
serverServices: serverServices.host,
resolveSession,
openSession: async (metadata, context) => {
@ -464,12 +441,7 @@ async function startServerBackend(
},
};
const socketPath = options.path;
const closeCatalog = async (): Promise<void> => {
const cleanup = await Promise.allSettled([serverServices.dispose(), closeStorage()]);
const errors = cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
if (errors.length === 1) throw errors[0];
if (errors.length > 1) throw new AggregateError(errors, "Experimental session catalog cleanup failed");
};
const closeCatalog = (): Promise<void> => serverServices.dispose();
const server = createUnixServer(host, {
serverId,
path: socketPath,
@ -560,7 +532,7 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
let defaultPluginSelection = await buildPluginSelection(pluginPackagePaths);
const sessionPluginSelections = new Map<string, ResolvedSessionPlugins>();
const resolveSessionPlugins = async (
metadata: JsonlSessionMetadata,
metadata: SessionCatalogMetadata,
requestedPackagePaths: readonly string[] | undefined,
_context: Context,
): Promise<ResolvedSessionPlugins> => {
@ -589,7 +561,7 @@ export async function startServer(options: StartServerOptions = {}): Promise<Run
sessionPluginSelections.set(metadata.path, selected);
return selected;
};
const removeSessionPlugins = async (metadata: JsonlSessionMetadata): Promise<void> => {
const removeSessionPlugins = async (metadata: SessionCatalogMetadata): Promise<void> => {
sessionPluginSelections.delete(metadata.path);
await removeSessionPluginPackageProfile(directory, serverId, metadata.path);
};

View file

@ -18,17 +18,28 @@ PI_EXPERIMENTAL=1 ./pi-test.sh client
| server | `PresentationPlugins` | prepares the selected Session branch's matching TUI artifacts and reloads that branch | add authenticated plugin policy |
| session | `SessionPlugins` | reloads the configured Session facet generation | add coordinated multi-worker reload reporting |
| session | `Models` | state, default-persisting selection, thinking, refresh implemented | move provider/auth composition behind plugin facets |
| session | `AgentController` | presentation-safe `AgentLane` facade for prompting, queueing, abort, resume, compaction, and navigation | add new lane operations only when a presentation requires them |
| session | `Transcript` | replicated lane state with source-event metadata | add projections only when another presentation needs them |
| session | `AgentController` | presentation-safe facade over the root durable conversation for prompting, steering, follow-ups, queue cancellation, abort, compaction, and waiting for a prompt's answer | add new conversation operations only when a presentation requires them |
| session | `Transcript` | the root conversation's durable `ConversationView` as replicated state | add projections only when another presentation needs them |
| presentation | `SlashCommands` | process-local contribution registry with model, thinking, compact, reload, and example hello commands | add more presentation hookpoints only as concrete plugin slices require them |
| presentation | `PresentationUI` | process-local selection and status capabilities | add narrowly scoped UI capabilities only when commands require them |
`ServerServiceSource.connection` and `SessionServiceSource.attachment` are implemented local control states. Session directory, creation, and address DTOs are owned by these coding-agent service contracts rather than `pi-protocol`; the transport treats their payloads as opaque service data.
With `PI_EXPERIMENTAL=1`, an interactive `pi client` creates and attaches a Session before opening the service-only chat TUI. `pi client -c` and `pi client -r` attach the newest existing Session instead, preserving its durable model and thinking configuration. Model selection is available on demand through `/model`; it is never a startup screen. The presentation always uses the stable coding agent's alternate-screen renderer and shared transcript/dock viewport. It loads configured theme resources and uses the stable terminal light/dark detection and appearance-change notifications. Its replicated state feeds the stable editor, message, tool, status, theme, and tool-renderer components. The presentation drives the worker-owned main lane through `AgentController` and renders the `Transcript` service's complete replicated value while controller calls are pending. The provider remains the sole Harness reducer and mutates Transcript's tracked snapshot before publishing each event. Chord flushes compact operations once per publication, while each client/state pairing encodes them with an independent path dictionary. Chord reconstructs presentation replicas and owns hydration, sequencing, and gap detection.
With `PI_EXPERIMENTAL=1`, an interactive `pi client` creates and attaches a Session before opening the service-only chat TUI. `pi client -c` and `pi client -r` attach the newest existing Session instead, preserving its durable model and thinking configuration. Model selection is available on demand through `/model`; it is never a startup screen. The presentation always uses the stable coding agent's alternate-screen renderer and shared transcript/dock viewport. It loads configured theme resources and uses the stable terminal light/dark detection and appearance-change notifications. Its replicated state feeds the stable editor, message, tool, status, theme, and tool-renderer components. The presentation drives the worker-owned root conversation through `AgentController` and renders the `Transcript` service's complete replicated value while controller calls are pending. `Transcript` serves `Conversation.viewState()` directly: the durable Harness publishes exact operations per commit, so the worker keeps no reducer of its own. Chord flushes compact operations once per publication, while each client/state pairing encodes them with an independent path dictionary. Chord reconstructs presentation replicas and owns hydration, sequencing, and gap detection.
A foreground server uses repeatable `-e` options to establish its default Session and TUI facets. A local client may instead select packages for the Session it creates or resumes; that branch selection is persisted with the Session and does not alter other workers or the server default. Before attaching, the server asks Chord to build conventional `src/session.ts` and `src/tui.ts` entries into separate `plugin-builds/` directories, passes the generated manifest paths to that Session worker, and returns the matching TUI artifacts. The Session worker loads built-in facets and separately owned plugin generations, then creates one active `FacetHost`. `/reload` atomically rebuilds the packages, loads fresh plugin candidates, cuts them over through `FacetHost.reload()`, and disposes the retired generations. Host-created implementation dependencies such as `AgentLane`, `ModelRuntime`, and `SettingsManager` are passed directly to built-in facet factories; they are not exposed as services. The TUI loads configured presentation facets and adds one private bridge facet that consumes the connected server and selected-Session services for `ExperimentalClientTui`. Its local presentation services are `SlashCommands`, which owns command contributions, and `PresentationUI`, which exposes selection and status rendering without exposing the raw TUI. Command callbacks receive Chord `Context` directly. Facets consume `AgentController` explicitly for prompting, steering, or queueing and return structured controller results to the command dispatcher.
The server keeps each Session in its own directory under the session directory: `meta.json` holds the ID, creation time, and working directory, and `session.sqlite` is the `@earendil-works/pi-durable` storage. The server lists and creates Sessions from `meta.json` only. The Session worker locks the directory, opens the storage, and owns it until it retires; the worker stays alive while the Harness task graph has live tasks.
A foreground server uses repeatable `-e` options to establish its default Session and TUI facets. A local client may instead select packages for the Session it creates or resumes; that branch selection is persisted with the Session and does not alter other workers or the server default. Before attaching, the server asks Chord to build conventional `src/session.ts` and `src/tui.ts` entries into separate `plugin-builds/` directories, passes the generated manifest paths to that Session worker, and returns the matching TUI artifacts. The Session worker loads built-in facets and separately owned plugin generations, then creates one active `FacetHost`. `/reload` atomically rebuilds the packages, loads fresh plugin candidates, cuts them over through `FacetHost.reload()`, and disposes the retired generations. Host-created implementation dependencies such as the durable `Harness`, its root `Conversation`, `ModelRuntime`, and `SettingsManager` are passed directly to built-in facet factories; they are not exposed as services. The TUI loads configured presentation facets and adds one private bridge facet that consumes the connected server and selected-Session services for `ExperimentalClientTui`. Its local presentation services are `SlashCommands`, which owns command contributions, and `PresentationUI`, which exposes selection and status rendering without exposing the raw TUI. Command callbacks receive Chord `Context` directly. Facets consume `AgentController` explicitly for prompting, steering, or queueing and return structured controller results to the command dispatcher.
A host may reload selected facets when their declared service shape is unchanged; retained consumers keep the same service facades while replacement implementations and singleton snapshots are installed behind them. Synchronous setup-time `env.provide()`, `env.provideMany()`, `env.use()`, and `env.observe()` calls produce the internal dependency graph; setup does not repeat a declarative dependency list, and service handles remain disconnected until the complete graph validates. Providers activate before consumers, observations connect with their consuming facet, and facet replacement or host shutdown disposes affected lifecycles in reverse dependency order. Server and Session service tokens are non-local and automatically published by their providing host. Presentation-only hookpoints such as `SlashCommands` are explicitly local and never enter an RPC catalogue.
A facet always calls unqualified `env.use()` or `env.observe()`. The host resolves each token across facet-provided and connected services. The example plugin package contributes `/hello` through its TUI facet. `examples/plugins/pi-example-plugin/` is an actual `@earendil-works/pi-example-plugin` package: repeatable `-e` options select plugin packages as server defaults or for one Session branch, Chord discovers and builds their conventional facets, and the attached presentation receives matching TUI artifacts through `PresentationPlugins`. Other host facets from the same plugin remain separate bundle entries rather than one aggregate plugin object. Transport bindings remain internal machinery rather than part of the facet environment. `ExperimentalClientTui` currently owns terminal rendering, state subscriptions, navigation, and action dispatch directly. The server provider remains directly assembled until its complete host environment exists. The canvas, diff-review, and task-observation examples in `packages/durable/docs/pico-v5-chord-usage.md` are extension patterns, not built-in coding-agent services. Private references, trace carriers, and flow control remain protocol/host infrastructure slices rather than presentation service tokens.
## TODO
Deferred while the worker moves from the removed `AgentHarness` to `@earendil-works/pi-durable`:
- Tree navigation: `AgentController.navigate()` is dropped. Durable branches by forking into a new conversation, so navigation needs a fork, a summary entry that starts the new context, and a pointer to the Session's current conversation.
- Next-run queue: `AgentController.nextRun()` is dropped. Durable queues only steering and follow-up input. `resume()` is dropped as well: the worker resumes interrupted work when it opens the Session.
- Subagents: the services cover only the root conversation. Exposing subagent conversations needs keyed service instances per conversation.
- Transcript history: the `Transcript` state is the conversation's view, which holds only the active context since the newest reset or compaction. Older entries need a paging method over `Conversation.entries()`.

View file

@ -1,109 +1,95 @@
import type { AgentLane, OperationResultRecord, SuspendedRun } from "@earendil-works/pi-agent-core";
import type { ImageContent } from "@earendil-works/pi-ai";
import type { Context } from "@earendil-works/chord";
import type { ImageContent, TextContent } from "@earendil-works/pi-ai";
import {
type Conversation,
ConversationBusy,
type Harness,
type SubmissionId,
type UserInput,
} from "@earendil-works/pi-durable";
import type {
AgentController as AgentControllerService,
AgentOperationError,
AgentOperationResponse,
AgentPromptRequest,
AgentPromptResult,
AgentQueueResponse,
} from "./agent-controller.ts";
export function createAgentController(lane: AgentLane): AgentControllerService {
export function createAgentController(harness: Harness, conversation: Conversation): AgentControllerService {
const queue = async (
operation: "steer" | "followUp" | "nextRun",
whenBusy: "steer" | "followUp",
request: AgentPromptRequest,
context: Parameters<AgentControllerService["steer"]>[1],
context: Context,
): Promise<AgentQueueResponse> => {
const [message, images] = toTextPrompt(request);
const result = await lane[operation](message, images, context);
return result.ok
? { accepted: true, entryId: result.value.entryId, error: null }
: { accepted: false, entryId: null, error: toAgentError(result.error) };
try {
const submission = await conversation.submit({ type: "input", content: toInput(request), whenBusy }, context);
return { accepted: true, entryId: String(submission.id), error: null };
} catch (error) {
return { accepted: false, entryId: null, error: toAgentError(error) };
}
};
return {
async prompt(request, context) {
const [message, images] = toTextPrompt(request);
const result = await lane.prompt(message, images, context);
return result.ok
? toOperationResponse(result.value)
: { accepted: false, operationId: operationId(result.error), error: toAgentError(result.error) };
},
async requestAbort(operationId, context) {
const result = await lane.requestAbort(operationId, context);
if (!result.ok) throw new Error(result.error.message);
try {
const submission = await conversation.submit(
{ type: "input", content: toInput(request), whenBusy: "reject" },
context,
);
return { accepted: true, operationId: String(submission.id), error: null };
} catch (error) {
return { accepted: false, operationId: null, error: toAgentError(error) };
}
},
steer: (request, context) => queue("steer", request, context),
followUp: (request, context) => queue("followUp", request, context),
nextRun: (request, context) => queue("nextRun", request, context),
async cancelQueued(entryId, context) {
const result = await lane.cancelQueued(entryId, context);
if (!result.ok) throw new Error(result.error.message);
return { outcome: result.value.kind };
},
async resume(context) {
const result = await lane.resume(context);
return result.ok
? toOperationResponse(result.value)
: { accepted: false, operationId: null, error: toAgentError(result.error) };
const id = parseSubmissionId(entryId);
if (id === undefined) return { outcome: "not_found" };
const result = await harness.abortSubmission(id, context, conversation.id);
return {
outcome: result === "aborted" ? "cancelled" : result === "not_found" ? "not_found" : "already_consumed",
};
},
abort: (context) => conversation.abort(context),
async compact(request, context) {
const result = await lane.compact(
request.customInstructions === null ? undefined : { customInstructions: request.customInstructions },
context,
);
return result.ok
? toOperationResponse(result.value.compaction)
: { accepted: false, operationId: operationId(result.error), error: toAgentError(result.error) };
try {
const id = await conversation.compact(request.customInstructions ?? undefined, context);
return { accepted: true, operationId: String(id), error: null };
} catch (error) {
return { accepted: false, operationId: null, error: toAgentError(error) };
}
},
async navigate(request, context) {
const result = await lane.navigateTree(
request.targetId,
{
summarize: request.summarize,
...(request.label === null ? {} : { label: request.label }),
...(request.customInstructions === null ? {} : { customInstructions: request.customInstructions }),
},
context,
);
return result.ok
? toOperationResponse(result.value.navigation)
: { accepted: false, operationId: operationId(result.error), error: toAgentError(result.error) };
async waitForPrompt(operationId, context): Promise<AgentPromptResult> {
const id = parseSubmissionId(operationId);
const submission = id === undefined ? undefined : await harness.submission(id, context);
if (submission === undefined) throw new Error(`Unknown prompt: ${operationId}`);
const settled = await submission.wait(context);
if (settled.status === "unanswered") return { status: "unanswered", text: null, reason: settled.reason };
const answer = settled.type === "input" ? settled.answer : undefined;
const message =
answer === undefined ? undefined : (await harness.commit((tx) => tx.entry(answer), context))?.model?.[0];
const text =
message?.role === "assistant"
? message.content.flatMap((content) => (content.type === "text" ? [content.text] : [])).join("")
: "";
return { status: "done", text, reason: null };
},
};
}
function toOperationResponse(value: OperationResultRecord | SuspendedRun): AgentOperationResponse {
return {
accepted: true,
operationId: value.operationId,
error:
"status" in value && value.status === "failed" && value.error !== undefined
? { code: value.error.code, message: value.error.message }
: null,
};
function parseSubmissionId(value: string): SubmissionId | undefined {
const id = Number(value);
return /^[1-9][0-9]*$/.test(value) && Number.isSafeInteger(id) ? (id as SubmissionId) : undefined;
}
function operationId(error: { readonly _tag: string }): string | null {
return "operationId" in error && typeof error.operationId === "string" ? error.operationId : null;
function toInput(request: AgentPromptRequest): UserInput {
if (request.images === null || request.images.length === 0) return request.message;
const content: (TextContent | ImageContent)[] = [{ type: "text", text: request.message }, ...request.images];
return content;
}
function toAgentError(error: { readonly _tag: string; readonly message: string }): AgentOperationError {
const code =
{
LaneBusy: "lane_busy",
InvalidMessage: "invalid_message",
UnknownSkill: "unknown_skill",
UnknownTemplate: "unknown_template",
NothingToCompact: "nothing_to_compact",
NothingToResume: "nothing_to_resume",
InvalidNavigation: "invalid_navigation",
UnknownTarget: "unknown_target",
Closed: "closed",
}[error._tag] ?? "operation_failed";
return { code, message: error.message };
}
function toTextPrompt(request: AgentPromptRequest): [message: string, images: ImageContent[] | undefined] {
return [request.message, request.images ?? undefined];
function toAgentError(error: unknown): AgentOperationError {
if (error instanceof ConversationBusy) return { code: "busy", message: error.message };
return { code: "operation_failed", message: error instanceof Error ? error.message : String(error) };
}

View file

@ -16,39 +16,42 @@ export interface AgentOperationError {
message: string;
}
/** `operationId` identifies the durable submission of a prompt, or the task of a compaction. */
export type AgentOperationResponse =
| { accepted: true; operationId: string; error: AgentOperationError | null }
| { accepted: false; operationId: string | null; error: AgentOperationError };
| { accepted: true; operationId: string; error: null }
| { accepted: false; operationId: null; error: AgentOperationError };
/** `entryId` identifies the durable submission; `cancelQueued()` withdraws it while it is still queued. */
export type AgentQueueResponse =
| { accepted: true; entryId: string; error: null }
| { accepted: false; entryId: null; error: AgentOperationError };
/** The settled outcome of a prompt: the answer's text, or why it got none. */
export type AgentPromptResult =
| { status: "done"; text: string; reason: null }
| { status: "unanswered"; text: null; reason: string };
export interface AgentCompactionRequest {
customInstructions: string | null;
}
export interface AgentNavigationRequest {
targetId: string | null;
summarize: boolean;
label: string | null;
customInstructions: string | null;
}
/** Presentation-safe command facade over the worker-owned main AgentLane. */
/** Presentation-safe command facade over the worker-owned root conversation. */
export interface AgentController {
/** Start a run; rejected with `busy` while one is active. */
prompt(request: AgentPromptRequest, context: Context): Promise<AgentOperationResponse>;
requestAbort(operationId: string, context: Context): Promise<void>;
/** Steer the active run, or start one when idle. */
steer(request: AgentPromptRequest, context: Context): Promise<AgentQueueResponse>;
/** Queue input for after the active run, or start one when idle. */
followUp(request: AgentPromptRequest, context: Context): Promise<AgentQueueResponse>;
nextRun(request: AgentPromptRequest, context: Context): Promise<AgentQueueResponse>;
cancelQueued(
entryId: string,
context: Context,
): Promise<{ outcome: "cancelled" | "already_consumed" | "not_found" }>;
resume(context: Context): Promise<AgentOperationResponse>;
/** Withdraw queued input and abort the active run and compaction. */
abort(context: Context): Promise<void>;
compact(request: AgentCompactionRequest, context: Context): Promise<AgentOperationResponse>;
navigate(request: AgentNavigationRequest, context: Context): Promise<AgentOperationResponse>;
/** Wait until the prompt with this `operationId` is answered or settles unanswered. */
waitForPrompt(operationId: string, context: Context): Promise<AgentPromptResult>;
}
export const AgentController = defineService<AgentController>("pi.agent-controller");

View file

@ -1,7 +1,13 @@
import { type Context, defineFacet, type Facet, type MutableReplicatedState } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { AgentLane, ThinkingLevel } from "@earendil-works/pi-agent-core";
import { getSupportedThinkingLevels } from "@earendil-works/pi-ai";
import { clampThinkingLevel, getSupportedThinkingLevels, type ModelThinkingLevel } from "@earendil-works/pi-ai";
import {
AgentDoc,
type AgentState,
type Conversation,
type DocumentState,
type Harness,
} from "@earendil-works/pi-durable";
import type { ModelRuntime } from "../../core/model-runtime.ts";
import type { SettingsManager } from "../../core/settings-manager.ts";
import { Models, type Models as ModelsService, type ModelsState } from "./models.ts";
@ -9,33 +15,41 @@ import { Models, type Models as ModelsService, type ModelsState } from "./models
export interface ModelsServiceRuntime {
readonly service: ModelsService;
activate(context: Context): Promise<void>;
/** Publish the agent document's model and thinking level when they changed. */
syncConfiguration(context: Context): void;
}
/**
* The Models service over one conversation. `agent` is the conversation's replicated `pi.agent` document: its
* configuration follows every change, also those made by other clients.
*/
export function createModelsService(
lane: AgentLane,
conversation: Conversation,
agent: DocumentState<AgentState>,
modelRuntime: ModelRuntime | undefined,
settingsManager: SettingsManager | undefined,
createState: (initial: ModelsState) => MutableReplicatedState<ModelsState>,
): ModelsServiceRuntime {
let catalogRevision = 0;
const configurationOf = (value: Readonly<AgentState> | null): ModelsState["configuration"] => ({
model: value?.model === undefined ? null : { provider: value.model.provider, modelId: value.model.modelId },
thinkingLevel: value?.thinkingLevel ?? "off",
});
const state = createState({
catalog: { revision: 0, availableModels: [] },
configuration: { model: null, thinkingLevel: "off" },
configuration: configurationOf(agent.value),
refresh: { status: "idle" },
});
const readConfiguration = async (context: Context): Promise<ModelsState["configuration"]> => {
const [selected, thinkingLevel] = await Promise.all([lane.getModel(context), lane.getThinkingLevel(context)]);
return {
model: selected === undefined ? null : { provider: selected.provider, modelId: selected.id },
thinkingLevel,
};
const selectedModel = () => {
const ref = agent.value?.model;
return ref === undefined ? undefined : modelRuntime?.getModel(ref.provider, ref.modelId);
};
const readThinkingLevels = async (context: Context): Promise<ThinkingLevel[]> => {
const selected = await lane.getModel(context);
const readThinkingLevels = (): ModelThinkingLevel[] => {
const selected = selectedModel();
return selected === undefined ? ["off"] : getSupportedThinkingLevels(selected);
};
const readCatalog = async (context: Context): Promise<ModelsState["catalog"]> => {
const selected = await lane.getModel(context);
const readCatalog = (): ModelsState["catalog"] => {
const selected = selectedModel();
const available = modelRuntime?.getAvailableSnapshot() ?? [];
const catalog =
selected === undefined || includesModel(available, selected) ? available : [...available, selected];
@ -53,94 +67,100 @@ export function createModelsService(
const service: ModelsService = {
state,
async cycleThinking(context) {
const levels = await readThinkingLevels(context);
const current = await lane.getThinkingLevel(context);
const index = levels.indexOf(current);
const next = levels[(index + 1) % levels.length] ?? "off";
await lane.setThinkingLevel(next, context);
const configuration = await readConfiguration(context);
state.change(context, (draft) => {
draft.configuration = configuration;
});
const levels = readThinkingLevels();
const current = agent.value?.thinkingLevel ?? "off";
const next = levels[(levels.indexOf(current) + 1) % levels.length] ?? "off";
await conversation.configure({ thinkingLevel: next }, context);
},
async getThinkingLevels(context) {
return [...(await readThinkingLevels(context))];
async getThinkingLevels() {
return readThinkingLevels();
},
async refresh(context) {
state.change(context, (draft) => {
draft.refresh = { status: "refreshing" };
});
const refresh = modelRuntime?.refresh({ signal: context.abortSignal });
if (refresh === undefined) {
const [catalog, configuration] = await Promise.all([readCatalog(context), readConfiguration(context)]);
state.change(context, (draft) => {
draft.catalog = catalog;
draft.configuration = configuration;
draft.refresh = { status: "done" };
});
return;
}
const result = await refresh;
const errors = Object.fromEntries([...result.errors].map(([id, error]) => [id, error.message]));
const [catalog, configuration] = await Promise.all([readCatalog(context), readConfiguration(context)]);
const errors =
refresh === undefined
? {}
: Object.fromEntries([...(await refresh).errors].map(([id, error]) => [id, error.message]));
const catalog = readCatalog();
state.change(context, (draft) => {
draft.catalog = catalog;
draft.configuration = configuration;
draft.refresh = Object.keys(errors).length === 0 ? { status: "done" } : { status: "warning", errors };
});
},
async select(model, context) {
const selected = modelRuntime?.getModel(model.provider, model.modelId);
if (selected === undefined) throw new Error(`Unknown model: ${model.provider}/${model.modelId}`);
await lane.setModel({ provider: selected.provider, modelId: selected.id }, context);
const thinkingLevel = clampThinkingLevel(selected, agent.value?.thinkingLevel ?? "off");
await conversation.configure(
{ model: { provider: selected.provider, modelId: selected.id }, thinkingLevel },
context,
);
settingsManager?.setDefaultModelAndProvider(selected.provider, selected.id);
await settingsManager?.flush();
const configuration = await readConfiguration(context);
state.change(context, (draft) => {
draft.configuration = configuration;
});
},
async selectThinking(level, context) {
const levels = await readThinkingLevels(context);
const levels = readThinkingLevels();
if (!levels.includes(level)) {
throw new Error(`Thinking level ${level} is unavailable; choose one of: ${levels.join(", ")}`);
}
await lane.setThinkingLevel(level, context);
const configuration = await readConfiguration(context);
state.change(context, (draft) => {
draft.configuration = configuration;
});
await conversation.configure({ thinkingLevel: level }, context);
},
};
return {
service,
async activate(context) {
const [catalog, configuration] = await Promise.all([readCatalog(context), readConfiguration(context)]);
const catalog = readCatalog();
state.change(context, (draft) => {
draft.catalog = catalog;
draft.configuration = configuration;
draft.refresh = { status: "idle" };
});
},
syncConfiguration(context) {
const next = configurationOf(agent.value);
const current = state.value.configuration;
if (
current.model?.provider === next.model?.provider &&
current.model?.modelId === next.model?.modelId &&
current.thinkingLevel === next.thinkingLevel
) {
return;
}
state.change(context, (draft) => {
draft.configuration = next;
});
},
};
}
export function createModelsServiceFacet(options: {
readonly lane: AgentLane;
/** Acquire the conversation's agent document, then build the facet that owns it. */
export async function createModelsServiceFacet(options: {
readonly harness: Harness;
readonly conversation: Conversation;
readonly modelRuntime: ModelRuntime | undefined;
readonly settingsManager?: SettingsManager;
}): Facet {
readonly context: Context;
}): Promise<Facet> {
const agent = await options.harness.documentState(AgentDoc, options.conversation.id, options.context);
if (agent === undefined) throw new Error(`Conversation ${options.conversation.id} has no agent document`);
return defineFacet({
id: "@pi/models",
setup(env) {
env.own(() => agent.dispose());
const runtime = createModelsService(
options.lane,
options.conversation,
agent,
options.modelRuntime,
options.settingsManager,
env.replicatedState,
);
env.provide(Models, runtime.service);
env.onActivate(() => runtime.activate(BACKGROUND_CONTEXT));
env.onActivate(async () => {
await runtime.activate(BACKGROUND_CONTEXT);
env.own(agent.subscribe((_value, context) => runtime.syncConfiguration(context)));
});
},
});
}

View file

@ -1,5 +1,5 @@
import { type Context, defineService, type ReplicatedState } from "@earendil-works/chord";
import type { ThinkingLevel } from "@earendil-works/pi-agent-core";
import type { ModelThinkingLevel } from "@earendil-works/pi-ai";
export interface ModelRef {
provider: string;
@ -18,7 +18,7 @@ export interface ModelsState {
};
configuration: {
model: ModelRef | null;
thinkingLevel: ThinkingLevel;
thinkingLevel: ModelThinkingLevel;
};
refresh: { status: "idle" | "refreshing" | "done" } | { status: "warning"; errors: Record<string, string> };
}
@ -26,10 +26,10 @@ export interface ModelsState {
export interface Models {
readonly state: ReplicatedState<ModelsState>;
cycleThinking(context: Context): Promise<void>;
getThinkingLevels(context: Context): Promise<ThinkingLevel[]>;
getThinkingLevels(context: Context): Promise<ModelThinkingLevel[]>;
refresh(context: Context): Promise<void>;
select(model: ModelRef, context: Context): Promise<void>;
selectThinking(level: ThinkingLevel, context: Context): Promise<void>;
selectThinking(level: ModelThinkingLevel, context: Context): Promise<void>;
}
export const Models = defineService<Models>("pi.models");

View file

@ -1,12 +1,12 @@
import { defineFacet, type Facet, type JsonValue } from "@earendil-works/chord";
import type { ThinkingLevel } from "@earendil-works/pi-agent-core";
import type { ModelThinkingLevel } from "@earendil-works/pi-ai";
import { AgentController } from "./agent-controller.ts";
import { type ModelSummary, Models, type Models as ModelsService } from "./models.ts";
import { PresentationPlugins, SessionPlugins } from "./plugins.ts";
import { PresentationUI } from "./presentation-ui.ts";
import { type SlashCommandContribution, SlashCommands } from "./slash-commands.ts";
const THINKING_DESCRIPTIONS: Record<ThinkingLevel, string> = {
const THINKING_DESCRIPTIONS: Record<ModelThinkingLevel, string> = {
off: "No reasoning",
minimal: "Very brief reasoning",
low: "Light reasoning",

View file

@ -1,124 +1,15 @@
import { type Context, defineFacet, type Facet, type MutableReplicatedState } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import {
type AgentLane,
type HarnessEvent,
type LaneSnapshot,
type LaneTranscriptSnapshot,
type LaneWatchEvent,
reduceLaneSnapshot,
type WatchHandle,
} from "@earendil-works/pi-agent-core";
import { Transcript, type Transcript as TranscriptService, type TranscriptState } from "./transcript.ts";
import { type Context, defineFacet, type Facet } from "@earendil-works/chord";
import type { Conversation } from "@earendil-works/pi-durable";
import { Transcript } from "./transcript.ts";
interface TranscriptRuntime {
readonly service: TranscriptService;
activate(): Promise<void>;
dispose(): Promise<void>;
}
export function createTranscriptService(
lane: AgentLane,
createState: (initial: TranscriptState) => MutableReplicatedState<TranscriptState>,
): TranscriptRuntime {
const state = createState({ snapshot: null, event: null });
let watch: WatchHandle<LaneSnapshot> | undefined;
let rebase: Promise<void> | undefined;
let rebaseError: Error | undefined;
const publishSnapshot = (next: LaneSnapshot, event: LaneWatchEvent | null, context: Context): void => {
state.replace(context, { snapshot: next as LaneTranscriptSnapshot, event });
};
const scheduleRebase = (context: Context): void => {
if (rebase !== undefined) return;
const activeWatch = watch;
if (activeWatch === undefined) return;
const pending = (async () => {
const refreshed = await activeWatch.resnapshot(context);
publishSnapshot(refreshed, null, context);
})();
rebase = pending;
void pending.then(
() => {
if (rebase === pending) rebase = undefined;
},
(error: unknown) => {
rebaseError = error instanceof Error ? error : new Error(String(error));
if (rebase === pending) rebase = undefined;
},
);
};
const onEvent = (event: HarnessEvent, context: Context): void => {
if (rebaseError !== undefined) throw rebaseError;
const forwarded = toLaneWatchEvent(event);
if (forwarded === undefined) return;
if (state.value.snapshot === null) throw new Error("Transcript service is not active");
let needsRebase = false;
state.change(context, (draft) => {
needsRebase = reduceLaneSnapshot(draft.snapshot as unknown as LaneSnapshot, event) === "rebase";
draft.event = forwarded;
});
if (needsRebase) scheduleRebase(context);
};
return {
service: { state },
async activate() {
if (watch !== undefined) throw new Error("Transcript service is already active");
const opened = await lane.watch(BACKGROUND_CONTEXT);
watch = opened;
publishSnapshot(opened.snapshot, null, BACKGROUND_CONTEXT);
opened.start(onEvent);
},
async dispose() {
let failure: unknown;
try {
await rebase;
} catch (error) {
failure = error;
}
watch?.unsubscribe();
watch = undefined;
if (failure !== undefined) throw failure;
},
};
}
export function createTranscriptServiceFacet(lane: AgentLane): Facet {
/** Serve the conversation's durable view state. The facet owns and disposes the attached state. */
export async function createTranscriptServiceFacet(conversation: Conversation, context: Context): Promise<Facet> {
const state = await conversation.viewState(context);
return defineFacet({
id: "@pi/transcript",
setup(env) {
const runtime = createTranscriptService(lane, env.replicatedState);
env.provide(Transcript, runtime.service);
env.onActivate(() => runtime.activate());
env.own(() => runtime.dispose());
env.own(() => state.dispose());
env.provide(Transcript, { state });
},
});
}
function toLaneWatchEvent(event: HarnessEvent): LaneWatchEvent | undefined {
switch (event.type) {
case "handler_error":
case "turn_start":
case "turn_end":
case "value_update":
case "lane_created":
return undefined;
case "config_update":
if (event.property !== "model" && event.property !== "thinkingLevel" && event.property !== "activeTools") {
return undefined;
}
return event as LaneWatchEvent;
case "message_update": {
if (event.message.role !== "assistant") {
throw new TypeError("Harness message_update did not carry an assistant message");
}
const { event: _providerEvent, ...update } = event;
return update as LaneWatchEvent;
}
default:
return event as LaneWatchEvent;
}
}

View file

@ -1,15 +1,9 @@
import { defineService, type ReplicatedState } from "@earendil-works/chord";
import type { LaneTranscriptSnapshot, LaneWatchEvent } from "@earendil-works/pi-agent-core";
import type { ConversationView } from "@earendil-works/pi-durable";
export interface TranscriptState {
snapshot: LaneTranscriptSnapshot | null;
/** The source event is retained for presentation side effects; hydration does not replay it. */
event: LaneWatchEvent | null;
}
/** Coherent main-lane state replicated through Chord's operation stream. */
/** The root conversation's durable view: active entries and its live, inbox, agent, and usage documents. */
export interface Transcript {
readonly state: ReplicatedState<TranscriptState>;
readonly state: ReplicatedState<ConversationView>;
}
export const Transcript = defineService<Transcript>("pi.transcript");

View file

@ -11,7 +11,8 @@ import {
type ServiceCall,
type ServiceProviderUpdate,
} from "@earendil-works/chord";
import type { AgentHarness, AgentLane } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { Conversation, Harness } from "@earendil-works/pi-durable";
import type { ModelRuntime } from "../../core/model-runtime.ts";
import type { SettingsManager } from "../../core/settings-manager.ts";
import { AgentController } from "./agent-controller.ts";
@ -21,11 +22,14 @@ import { SessionPlugins } from "./plugins.ts";
import { createTranscriptServiceFacet } from "./transcript-provider.ts";
export interface SessionWorkerRuntime {
readonly harness: AgentHarness;
readonly lane?: AgentLane;
readonly harness: Harness;
/** The root conversation the services expose. */
readonly conversation: Conversation;
readonly modelRuntime?: ModelRuntime;
readonly settingsManager?: SettingsManager;
readonly facetLoader?: FacetLoader;
/** Release resources the Harness does not own, such as execution environments, after it closed. */
cleanup?(context: Context): Promise<void>;
}
export interface WorkerServiceScope {
@ -45,7 +49,8 @@ export interface SessionWorkerServices {
}
export async function createSessionWorkerServices(options: {
readonly lane: AgentLane;
readonly harness: Harness;
readonly conversation: Conversation;
readonly modelRuntime: ModelRuntime | undefined;
readonly settingsManager?: SettingsManager;
readonly facetLoader?: FacetLoader;
@ -54,7 +59,7 @@ export async function createSessionWorkerServices(options: {
const agentControllerRuntimeFacet = defineFacet({
id: "@pi/agent-controller-runtime",
setup(env) {
env.provide(AgentController, createAgentController(options.lane));
env.provide(AgentController, createAgentController(options.harness, options.conversation));
},
});
let reloadPlugins = (): Promise<void> => Promise.reject(new Error("Session plugins are not ready"));
@ -67,8 +72,8 @@ export async function createSessionWorkerServices(options: {
const builtins = await createStaticFacetLoader([
agentControllerRuntimeFacet,
pluginRuntimeFacet,
createModelsServiceFacet(options),
createTranscriptServiceFacet(options.lane),
await createModelsServiceFacet({ ...options, context: BACKGROUND_CONTEXT }),
await createTranscriptServiceFacet(options.conversation, BACKGROUND_CONTEXT),
]).load();
const pluginLoader = options.facetLoader ?? createStaticFacetLoader([]);
let loadedPlugins = await pluginLoader.load();

View file

@ -0,0 +1,87 @@
import { randomUUID } from "node:crypto";
import { mkdir, readdir, readFile, rm, writeFile } from "node:fs/promises";
import { join } from "node:path";
/** One server-hosted Session: a directory holding `meta.json` and the worker-owned `session.sqlite`. */
export interface SessionCatalogMetadata {
readonly id: string;
readonly createdAt: number;
/** The working directory the Session's agent runs in. */
readonly cwd: string;
/** The Session directory. Workers lock it and own the storage inside it. */
readonly path: string;
}
const METADATA_FILE = "meta.json";
const STORAGE_FILE = "session.sqlite";
const SESSION_ID = /^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$/;
export function isSessionId(id: string): boolean {
return SESSION_ID.test(id);
}
/** The durable storage file of a Session. Only the Session's worker opens it. */
export function sessionStoragePath(metadata: SessionCatalogMetadata): string {
return join(metadata.path, STORAGE_FILE);
}
/** Every Session in the directory. Entries without valid metadata are skipped. */
export async function listSessions(sessionDir: string): Promise<SessionCatalogMetadata[]> {
let names: string[];
try {
names = await readdir(sessionDir);
} catch (error) {
if (isNotFound(error)) return [];
throw error;
}
const sessions = await Promise.all(names.filter(isSessionId).map((name) => readSession(sessionDir, name)));
return sessions.filter((metadata) => metadata !== undefined);
}
/** One Session by ID, or undefined when it does not exist. */
export async function readSession(sessionDir: string, id: string): Promise<SessionCatalogMetadata | undefined> {
if (!isSessionId(id)) return undefined;
const path = join(sessionDir, id);
let value: unknown;
try {
value = JSON.parse(await readFile(join(path, METADATA_FILE), "utf8"));
} catch {
return undefined;
}
if (typeof value !== "object" || value === null) return undefined;
const { createdAt, cwd } = value as { createdAt?: unknown; cwd?: unknown };
if (typeof createdAt !== "number" || typeof cwd !== "string") return undefined;
return { id, createdAt, cwd, path };
}
/** Create an empty Session. Its worker creates the storage on first open. */
export async function createSession(
sessionDir: string,
options: { readonly id?: string; readonly cwd: string },
): Promise<SessionCatalogMetadata> {
const id = options.id ?? randomUUID();
if (!isSessionId(id)) throw new Error(`Invalid session ID: ${id}`);
const path = join(sessionDir, id);
await mkdir(sessionDir, { recursive: true });
try {
await mkdir(path);
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "EEXIST") throw new Error(`Session ${id} already exists`);
throw error;
}
const metadata: SessionCatalogMetadata = { id, createdAt: Date.now(), cwd: options.cwd, path };
await writeFile(
join(path, METADATA_FILE),
`${JSON.stringify({ createdAt: metadata.createdAt, cwd: metadata.cwd }, null, "\t")}\n`,
);
return metadata;
}
/** Delete a Session directory. Its worker must be closed first. */
export async function deleteSession(metadata: SessionCatalogMetadata): Promise<void> {
await rm(metadata.path, { recursive: true, force: true });
}
function isNotFound(error: unknown): boolean {
return (error as NodeJS.ErrnoException | undefined)?.code === "ENOENT";
}

View file

@ -2,6 +2,7 @@ import type { ChildProcess } from "node:child_process";
import { randomUUID } from "node:crypto";
import { isAbsolute } from "node:path";
import {
type Context,
createServiceUnsubscribeCall,
decodeServiceControlCall,
type JsonValue,
@ -9,16 +10,12 @@ import {
type ServiceCall,
type ServiceProviderUpdate,
} from "@earendil-works/chord";
import {
BACKGROUND_CONTEXT,
type Context,
type JsonlSessionMetadata,
TODO_CONTEXT,
} from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT, TODO_CONTEXT } from "@earendil-works/chord/context";
import { type RoutedSessionAttachment, type RoutedSessionHandle, ServerError } from "@earendil-works/pi-server";
import { Check } from "typebox/value";
import type { CoordinatorConnection, CoordinatorConnectionEvent } from "./coordinator.ts";
import { spawnInternalProcess } from "./process.ts";
import type { SessionCatalogMetadata } from "./session-catalog.ts";
import {
SESSION_WORKER_CONTROL_ADDRESS_ENV,
SESSION_WORKER_CONTROL_TOKEN_ENV,
@ -45,7 +42,7 @@ export class SessionPluginSelectionConflictError extends Error {
interface WorkerRecord {
readonly peerId: string;
readonly metadata: JsonlSessionMetadata;
readonly metadata: SessionCatalogMetadata;
readonly pid: number;
readonly token: string;
readonly pluginManifestPaths: readonly string[];
@ -132,11 +129,11 @@ export class SessionWorkerManager {
this.#removeListener = coordinator.onEvent((event) => this.#handleCoordinatorEvent(event));
}
get trackedSessions(): readonly JsonlSessionMetadata[] {
get trackedSessions(): readonly SessionCatalogMetadata[] {
return [...this.#workersBySession.values()].map((worker) => worker.metadata);
}
assertSessionPluginManifestPaths(metadata: JsonlSessionMetadata, manifestPaths: readonly string[]): void {
assertSessionPluginManifestPaths(metadata: SessionCatalogMetadata, manifestPaths: readonly string[]): void {
const existing = this.#workersBySession.get(metadata.path);
if (existing !== undefined && !sameStrings(existing.pluginManifestPaths, manifestPaths)) {
throw new SessionPluginSelectionConflictError(
@ -176,7 +173,7 @@ export class SessionWorkerManager {
}
async openSession(
metadata: JsonlSessionMetadata,
metadata: SessionCatalogMetadata,
context: Context,
pluginManifestPaths: readonly string[],
): Promise<RoutedSessionHandle> {
@ -189,7 +186,7 @@ export class SessionWorkerManager {
return this.#routedHandle(await this.#launch(metadata, context, pluginManifestPaths));
}
async closeSession(metadata: JsonlSessionMetadata, context: Context): Promise<void> {
async closeSession(metadata: SessionCatalogMetadata, context: Context): Promise<void> {
const worker = this.#workersBySession.get(metadata.path) ?? (await this.#pending.get(metadata.path)?.promise);
if (worker !== undefined) await this.#stopWorker(worker, context);
}
@ -450,7 +447,7 @@ export class SessionWorkerManager {
}
#launch(
metadata: JsonlSessionMetadata,
metadata: SessionCatalogMetadata,
_context: Context,
pluginManifestPaths: readonly string[],
): Promise<WorkerRecord> {
@ -461,15 +458,7 @@ export class SessionWorkerManager {
try {
const options: SessionWorkerOptions = {
sessionDir: this.#sessionDir,
metadata: {
id: metadata.id,
createdAt: metadata.createdAt,
storageVersion: metadata.storageVersion,
cwd: metadata.cwd,
path: metadata.path,
modifiedAt: metadata.modifiedAt,
...(metadata.parentSessionId === undefined ? {} : { parentSessionId: metadata.parentSessionId }),
},
metadata: { id: metadata.id, createdAt: metadata.createdAt, cwd: metadata.cwd, path: metadata.path },
pluginManifestPaths: [...pluginManifestPaths],
...(this.#model ?? {}),
};

View file

@ -1,6 +1,7 @@
import { createConnection, type Socket } from "node:net";
import { isAbsolute } from "node:path";
import {
type AttachedReplicatedState,
isJsonValue,
type JsonValue,
parseServiceProviderUpdate,
@ -10,28 +11,22 @@ import {
type ServiceCall,
type ServiceProviderUpdate,
} from "@earendil-works/chord";
import {
AgentHarness,
type AgentHarness as AgentHarnessInstance,
type AgentLane,
BACKGROUND_CONTEXT,
createBashTool,
createReadTool,
createWriteTool,
type JsonlSessionMetadata,
JsonlSessionRepo,
type Session,
TODO_CONTEXT,
withCancel,
} from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
import { BACKGROUND_CONTEXT, TODO_CONTEXT, withCancel } from "@earendil-works/chord/context";
import { Harness, ROOT_CONVERSATION_ID, type TaskGraph } from "@earendil-works/pi-durable";
import { openNodeSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/node";
import lockfile from "proper-lockfile";
import Type, { type Static } from "typebox";
import { Check } from "typebox/value";
import { findInitialModel, resolveCliModel } from "../core/model-resolver.ts";
import { ModelRuntime } from "../core/model-runtime.ts";
import { SettingsManager } from "../core/settings-manager.ts";
import { COORDINATOR_PROTOCOL_VERSION } from "./coordinator.ts";
import {
configureHarnessHttp,
createCodingRegistry,
createHarnessSettings,
ExecutionEnvs,
findInitialAgentModel,
} from "./durable/harness-setup.ts";
import { createSessionPluginFacetLoader } from "./plugins/bundled.ts";
import {
consumeInternalProcessRole,
@ -45,6 +40,7 @@ import {
type SessionWorkerServices,
type WorkerServiceScope,
} from "./services/worker.ts";
import { sessionStoragePath } from "./session-catalog.ts";
export type { SessionWorkerRuntime } from "./services/worker.ts";
@ -72,12 +68,9 @@ export const SESSION_WORKER_PEER_ID_ENV = "PI_SESSION_WORKER_PEER_ID";
export const SessionWorkerMetadataSchema = StrictObject({
id: Type.String({ minLength: 1 }),
createdAt: Type.Integer(),
storageVersion: Type.Integer(),
createdAt: Type.Number(),
cwd: Type.String(),
path: Type.String(),
modifiedAt: Type.Number(),
parentSessionId: Type.Optional(Type.String()),
});
export const SessionWorkerOptionsSchema = StrictObject({
@ -193,7 +186,7 @@ export class WorkerLifecycle {
readonly #orphanDemandGraceMs: number;
readonly #onRetire: () => void;
readonly #demands = new Map<string, { serverConnectionId: string; attachmentId: string; timer?: NodeJS.Timeout }>();
readonly #activeOperations = new Set<string>();
#harnessActive = false;
#currentServerConnectionId: string | undefined;
#initialTimer: NodeJS.Timeout | undefined;
#demandInitialized: boolean;
@ -282,13 +275,10 @@ export class WorkerLifecycle {
this.#reconcile();
}
operationStarted(kind: "run" | "compaction" | "navigation", lane: string, operationId: string): void {
this.#activeOperations.add(`${kind}\0${lane}\0${operationId}`);
}
operationStopped(kind: "run" | "compaction" | "navigation", lane: string, operationId: string): void {
this.#activeOperations.delete(`${kind}\0${lane}\0${operationId}`);
this.#reconcile();
/** Live Harness tasks, such as runs and compactions, hold the worker until they end. */
setHarnessActive(active: boolean): void {
this.#harnessActive = active;
if (!active) this.#reconcile();
}
close(): void {
@ -304,7 +294,7 @@ export class WorkerLifecycle {
this.#retiring ||
!this.#demandInitialized ||
this.#retirementHolds !== 0 ||
this.#activeOperations.size !== 0 ||
this.#harnessActive ||
this.#demands.size !== 0
) {
return;
@ -478,11 +468,8 @@ function lifecycleDelay(name: string, fallback: number): number {
}
async function closeResources(resources: {
harness?: AgentHarnessInstance;
runtime?: SessionWorkerRuntime;
services?: SessionWorkerServices;
session?: Session<JsonlSessionMetadata>;
repo: JsonlSessionRepo;
executionEnv: NodeExecutionEnv;
releaseOwnership: () => Promise<void>;
}): Promise<void> {
const errors: unknown[] = [];
@ -492,18 +479,12 @@ async function closeResources(resources: {
errors.push(error);
}
try {
if (resources.harness) await resources.harness.close(TODO_CONTEXT);
else await resources.session?.close(TODO_CONTEXT);
await resources.runtime?.harness.close(TODO_CONTEXT);
} catch (error) {
errors.push(error);
}
try {
await resources.repo.close(TODO_CONTEXT);
} catch (error) {
errors.push(error);
}
try {
await resources.executionEnv.cleanup(TODO_CONTEXT);
await resources.runtime?.cleanup?.(TODO_CONTEXT);
} catch (error) {
errors.push(error);
}
@ -516,14 +497,17 @@ async function closeResources(resources: {
if (errors.length > 1) throw new AggregateError(errors, "Session worker cleanup failed");
}
/**
* Open the Session's Harness over `databasePath` and return it with its root conversation. Called while the worker
* holds the Session lock; on failure it closes whatever it opened.
*/
export type CreateSessionWorkerHarness = (
session: Session<JsonlSessionMetadata>,
databasePath: string,
options: SessionWorkerOptions,
executionEnv: NodeExecutionEnv,
) => Promise<SessionWorkerRuntime>;
async function run(options: SessionWorkerOptions, createHarness: CreateSessionWorkerHarness): Promise<void> {
const { sessionDir, metadata } = options;
const { metadata } = options;
const sessionId = metadata.id;
const control = await connectControl();
const token = process.env[SESSION_WORKER_CONTROL_TOKEN_ENV]!;
@ -536,19 +520,15 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
update: 1_000,
retries: { retries: 320, factor: 1, minTimeout: 25, maxTimeout: 25, maxRetryTime: 8_000 },
});
const executionEnv = new NodeExecutionEnv({ cwd: metadata.cwd });
const repo = new JsonlSessionRepo({ fileSystem: executionEnv, sessionsRoot: sessionDir });
let session: Session<JsonlSessionMetadata> | undefined;
let harness: AgentHarnessInstance | undefined;
let lane: AgentLane | undefined;
let runtime: SessionWorkerRuntime | undefined;
let services: SessionWorkerServices | undefined;
let taskGraph: AttachedReplicatedState<TaskGraph> | undefined;
try {
session = await repo.open(metadata, TODO_CONTEXT);
const runtime = await createHarness(session, options, executionEnv);
harness = runtime.harness;
lane = runtime.lane ?? (await harness.lane("main", TODO_CONTEXT));
runtime = await createHarness(sessionStoragePath(metadata), options);
taskGraph = await runtime.harness.taskGraph(TODO_CONTEXT);
services = await createSessionWorkerServices({
lane,
harness: runtime.harness,
conversation: runtime.conversation,
modelRuntime: runtime.modelRuntime,
settingsManager: runtime.settingsManager,
facetLoader: runtime.facetLoader,
@ -563,8 +543,9 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
}),
});
} catch (error) {
taskGraph?.dispose();
try {
await closeResources({ harness, services, session, repo, executionEnv, releaseOwnership });
await closeResources({ runtime, services, releaseOwnership });
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Session worker startup and cleanup failed");
}
@ -575,18 +556,20 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
string,
{ readonly scope: WorkerOperationScope; readonly cancel: (reason?: unknown) => void }
>();
const harness = runtime.harness;
const activity = taskGraph;
let lifecycle: WorkerLifecycle | undefined;
let removeLifecycleListeners: (() => void)[] = [];
let stopActivity = (): void => {};
let closing: Promise<void> | undefined;
const close = (): Promise<void> => {
if (closing) return closing;
lifecycle?.close();
stopActivity();
activity.dispose();
services.removeSubscriptions(() => true);
for (const request of activeRequests.values()) request.cancel(new Error("Session worker is closing"));
activeRequests.clear();
for (const remove of removeLifecycleListeners) remove();
removeLifecycleListeners = [];
closing = closeResources({ harness, services, repo, executionEnv, releaseOwnership });
closing = closeResources({ runtime, services, releaseOwnership });
return closing;
};
const closeAndExit = (): void => {
@ -605,25 +588,10 @@ async function run(options: SessionWorkerOptions, createHarness: CreateSessionWo
orphanDemandGraceMs: lifecycleDelay(SESSION_WORKER_ORPHAN_DEMAND_GRACE_ENV, DEFAULT_ORPHAN_DEMAND_GRACE_MS),
onRetire: closeAndExit,
});
removeLifecycleListeners = [
harness.events.on("run_start", (event) => lifecycle?.operationStarted("run", event.lane, event.runId)),
harness.events.on("run_resume", (event) => lifecycle?.operationStarted("run", event.lane, event.runId)),
harness.events.on("run_suspend", (event) => lifecycle?.operationStopped("run", event.lane, event.runId)),
harness.events.on("run_end", (event) => lifecycle?.operationStopped("run", event.lane, event.runId)),
harness.events.on("compaction_start", (event) =>
lifecycle?.operationStarted("compaction", event.lane, event.runId),
),
harness.events.on("compaction_end", (event) =>
lifecycle?.operationStopped("compaction", event.lane, event.runId),
),
harness.events.on("navigation_start", (event) =>
lifecycle?.operationStarted("navigation", event.lane, event.runId),
),
harness.events.on("navigation_end", (event) =>
lifecycle?.operationStopped("navigation", event.lane, event.runId),
),
harness.events.on("fault", closeAndExit),
];
// Every live task, background work included, keeps the worker and its Session open.
stopActivity = activity.subscribe((graph) => lifecycle?.setHarnessActive(Object.keys(graph.tasks).length > 0));
// Recovered work from an interrupted turn continues now.
harness.resume();
const handleOperation = async (request: WorkerOperationRequest): Promise<void> => {
let releaseRequest = (): void => {};
@ -803,69 +771,57 @@ export async function runSessionWorkerWithHarness(
}
async function createCodingAgentHarness(
session: Session<JsonlSessionMetadata>,
databasePath: string,
options: SessionWorkerOptions,
executionEnv: NodeExecutionEnv,
): Promise<SessionWorkerRuntime> {
const { cwd } = options.metadata;
const modelRuntime = await ModelRuntime.create();
const settingsManager = SettingsManager.create(session.metadata.cwd);
let resolved: Awaited<ReturnType<typeof findInitialModel>> | ReturnType<typeof resolveCliModel>;
if (options.model === undefined) {
resolved = await findInitialModel({
scopedModels: [],
isContinuing: true,
defaultProvider: settingsManager.getDefaultProvider(),
defaultModelId: settingsManager.getDefaultModel(),
defaultThinkingLevel: settingsManager.getDefaultThinkingLevel(),
modelRuntime,
});
} else {
resolved = resolveCliModel({
cliProvider: options.provider,
cliModel: options.model,
modelRuntime,
});
if (resolved.error) throw new Error(`Session worker could not resolve model: ${resolved.error}`);
}
if (!resolved.model) throw new Error("Session worker could not resolve a model");
const tools = [createReadTool(), createWriteTool(), createBashTool()];
const activeToolNames = tools.map((tool) => tool.name);
const harness = (
await AgentHarness.create(
const settingsManager = SettingsManager.create(cwd);
configureHarnessHttp(settingsManager);
const envs = new ExecutionEnvs(cwd);
let harness: Harness | undefined;
try {
harness = await Harness.open(
await openNodeSqliteStorage(databasePath),
{
session,
models: modelRuntime,
model: resolved.model,
thinkingLevel: resolved.thinkingLevel,
tools,
activeToolNames,
toolContext: { env: executionEnv },
resources: {},
registry: createCodingRegistry(settingsManager, cwd),
settings: createHarnessSettings(settingsManager),
env: envs.env,
onReport: (error) => console.error(error),
},
TODO_CONTEXT,
)
).harness;
try {
const lane = await harness.lane("main", TODO_CONTEXT);
const currentActiveToolNames = await lane.getActiveTools(TODO_CONTEXT);
if (
currentActiveToolNames.length !== activeToolNames.length ||
currentActiveToolNames.some((name, index) => name !== activeToolNames[index])
) {
await lane.setActiveTools(activeToolNames, TODO_CONTEXT);
}
);
// The initial model applies only when the root conversation is created; later starts keep its durable model.
const created = (await harness.conversation(ROOT_CONVERSATION_ID, TODO_CONTEXT)) === undefined;
const initial = created
? await findInitialAgentModel(
settingsManager,
modelRuntime,
options.model === undefined ? undefined : { provider: options.provider, model: options.model },
)
: undefined;
const conversation = await harness.root(TODO_CONTEXT, {
agent: {
cwd,
...(initial?.model === undefined ? {} : { model: initial.model }),
...(initial?.thinkingLevel === undefined ? {} : { thinkingLevel: initial.thinkingLevel }),
},
});
return {
harness,
lane,
conversation,
modelRuntime,
settingsManager,
facetLoader: createSessionPluginFacetLoader(options.pluginManifestPaths),
cleanup: (context) => envs.cleanup(context),
};
} catch (error) {
try {
await harness.close(TODO_CONTEXT);
await harness?.close(TODO_CONTEXT);
await envs.cleanup(TODO_CONTEXT);
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], "Session worker model selection and cleanup failed");
throw new AggregateError([error, cleanupError], "Session worker Harness startup and cleanup failed");
}
throw error;
}

View file

@ -1,178 +1,113 @@
import { createFacetHost, defineFacet } from "@earendil-works/chord";
import {
type AgentLane,
BACKGROUND_CONTEXT,
Closed,
type RunResult as HarnessRunResult,
InvalidMessage,
LaneBusy,
UnknownSkill,
UnknownTemplate,
} from "@earendil-works/pi-agent-core";
import { describe, expect, test, vi } from "vitest";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { fauxAssistantMessage } from "@earendil-works/pi-ai";
import type { InboxState, LiveState } from "@earendil-works/pi-durable";
import { describe, expect, test } from "vitest";
import { AgentController } from "../src/experimental/services/agent-controller.ts";
import { createAgentController } from "../src/experimental/services/agent-controller-provider.ts";
const admissionErrors = [
[
new LaneBusy({
lane: "main",
operationId: "operation-1",
operationKind: "run",
message: "busy",
}),
"lane_busy",
"operation-1",
],
[new InvalidMessage({ lane: "main", reason: "invalid", message: "invalid" }), "invalid_message", null],
[new UnknownSkill({ name: "skill", message: "unknown" }), "unknown_skill", null],
[new UnknownTemplate({ name: "template", message: "unknown" }), "unknown_template", null],
[new Closed({ message: "closed" }), "closed", null],
] as const satisfies readonly [Extract<HarnessRunResult, { ok: false }>["error"], string, string | null][];
const completed = {
operationId: "operation-1",
kind: "run" as const,
status: "completed" as const,
fromTipId: null,
tipId: "entry-1",
startedAt: 1,
endedAt: 2,
};
import { openFauxConversation, pendingResponse } from "./experimental-durable-support.ts";
describe("AgentController service", () => {
test("provides the presentation-safe AgentLane command surface", async () => {
const prompt = vi.fn(async () => ({ ok: true as const, value: completed }));
const requestAbort = vi.fn(async () => ({
ok: true as const,
value: { operationId: "operation-1", newlyRequested: true, steer: [], followUp: [] },
}));
const steer = vi.fn(async () => ({ ok: true as const, value: { entryId: "queue-1" } }));
const followUp = vi.fn(async () => ({ ok: true as const, value: { entryId: "queue-2" } }));
const lane = { prompt, requestAbort, steer, followUp } as unknown as AgentLane;
const controllerFacet = defineFacet({
id: "test-agent-controller",
setup(env) {
env.provide(AgentController, createAgentController(lane));
},
test("prompts the root conversation through the service catalogue", async () => {
const { harness, conversation, close } = await openFauxConversation([fauxAssistantMessage("hello back")]);
const host = await createFacetHost({
facets: [
defineFacet({
id: "test-agent-controller",
setup(env) {
env.provide(AgentController, createAgentController(harness, conversation));
},
}),
],
});
const host = await createFacetHost({ facets: [controllerFacet] });
try {
expect(host.services.catalogue).toEqual([{ serviceId: AgentController.id, mode: "singleton" }]);
const response = await host.services.invoke(
{ serviceId: AgentController.id, member: "prompt", args: [{ message: "hello", images: null }] },
BACKGROUND_CONTEXT,
);
expect(response).toEqual({ accepted: true, operationId: expect.any(String), error: null });
const operationId = (response as { operationId: string }).operationId;
await expect(
host.services.invoke(
{ serviceId: AgentController.id, member: "prompt", args: [{ message: "hello", images: null }] },
{ serviceId: AgentController.id, member: "waitForPrompt", args: [operationId] },
BACKGROUND_CONTEXT,
),
).resolves.toEqual({ accepted: true, operationId: "operation-1", error: null });
await expect(
host.services.invoke(
{ serviceId: AgentController.id, member: "requestAbort", args: ["operation-1"] },
BACKGROUND_CONTEXT,
),
).resolves.toBeUndefined();
await expect(
host.services.invoke(
{ serviceId: AgentController.id, member: "steer", args: [{ message: "later", images: null }] },
BACKGROUND_CONTEXT,
),
).resolves.toEqual({ accepted: true, entryId: "queue-1", error: null });
await expect(
host.services.invoke(
{ serviceId: AgentController.id, member: "followUp", args: [{ message: "after", images: null }] },
BACKGROUND_CONTEXT,
),
).resolves.toEqual({ accepted: true, entryId: "queue-2", error: null });
expect(prompt).toHaveBeenCalledWith("hello", undefined, BACKGROUND_CONTEXT);
expect(requestAbort).toHaveBeenCalledWith("operation-1", BACKGROUND_CONTEXT);
expect(steer).toHaveBeenCalledWith("later", undefined, BACKGROUND_CONTEXT);
expect(followUp).toHaveBeenCalledWith("after", undefined, BACKGROUND_CONTEXT);
).resolves.toEqual({ status: "done", text: "hello back", reason: null });
} finally {
await host.dispose();
await close();
}
});
test("wraps queue, resume, compaction, and navigation lane operations", async () => {
const nextRun = vi.fn(async () => ({ ok: true as const, value: { entryId: "queue-3" } }));
const cancelQueued = vi.fn(async () => ({ ok: true as const, value: { kind: "cancelled" as const } }));
const resume = vi.fn(async () => ({ ok: true as const, value: completed }));
const compact = vi.fn(async () => ({
ok: true as const,
value: { compaction: { ...completed, operationId: "compact-1", kind: "compaction" as const } },
}));
const navigateTree = vi.fn(async () => ({
ok: true as const,
value: { navigation: { ...completed, operationId: "navigation-1", kind: "navigation" as const } },
}));
const controller = createAgentController({
nextRun,
cancelQueued,
resume,
compact,
navigateTree,
} as unknown as AgentLane);
test("rejects a prompt while busy and queues steering and follow-up input", async () => {
const pending = pendingResponse();
const { harness, conversation, close } = await openFauxConversation([pending.step]);
const controller = createAgentController(harness, conversation);
try {
const first = await controller.prompt({ message: "first", images: null }, BACKGROUND_CONTEXT);
expect(first.accepted).toBe(true);
await pending.reached;
await expect(controller.nextRun({ message: "next", images: null }, BACKGROUND_CONTEXT)).resolves.toEqual({
accepted: true,
entryId: "queue-3",
error: null,
});
await expect(controller.cancelQueued("queue-3", BACKGROUND_CONTEXT)).resolves.toEqual({
outcome: "cancelled",
});
await expect(controller.resume(BACKGROUND_CONTEXT)).resolves.toMatchObject({ operationId: "operation-1" });
await expect(controller.compact({ customInstructions: "short" }, BACKGROUND_CONTEXT)).resolves.toMatchObject({
operationId: "compact-1",
});
await expect(
controller.navigate(
{ targetId: "entry-1", summarize: true, label: "branch", customInstructions: null },
BACKGROUND_CONTEXT,
),
).resolves.toMatchObject({ operationId: "navigation-1" });
expect(nextRun).toHaveBeenCalledWith("next", undefined, BACKGROUND_CONTEXT);
expect(cancelQueued).toHaveBeenCalledWith("queue-3", BACKGROUND_CONTEXT);
expect(resume).toHaveBeenCalledWith(BACKGROUND_CONTEXT);
expect(compact).toHaveBeenCalledWith({ customInstructions: "short" }, BACKGROUND_CONTEXT);
expect(navigateTree).toHaveBeenCalledWith("entry-1", { summarize: true, label: "branch" }, BACKGROUND_CONTEXT);
await expect(controller.prompt({ message: "second", images: null }, BACKGROUND_CONTEXT)).resolves.toEqual({
accepted: false,
operationId: null,
error: { code: "busy", message: expect.stringContaining("busy") },
});
const steer = await controller.steer({ message: "steer", images: null }, BACKGROUND_CONTEXT);
const followUp = await controller.followUp({ message: "later", images: null }, BACKGROUND_CONTEXT);
expect(steer).toEqual({ accepted: true, entryId: expect.any(String), error: null });
expect(followUp).toEqual({ accepted: true, entryId: expect.any(String), error: null });
const inbox = await conversation.viewState(BACKGROUND_CONTEXT);
try {
expect((inbox.value.docs["pi.inbox"] as InboxState).items.map((item) => item.mode)).toEqual([
"steer",
"followUp",
]);
} finally {
inbox.dispose();
}
if (!followUp.accepted) throw new Error("Follow-up was rejected");
await expect(controller.cancelQueued(followUp.entryId, BACKGROUND_CONTEXT)).resolves.toEqual({
outcome: "cancelled",
});
await expect(controller.cancelQueued(followUp.entryId, BACKGROUND_CONTEXT)).resolves.toEqual({
outcome: "already_consumed",
});
await expect(controller.cancelQueued("999", BACKGROUND_CONTEXT)).resolves.toEqual({ outcome: "not_found" });
await expect(controller.cancelQueued("not-an-id", BACKGROUND_CONTEXT)).resolves.toEqual({
outcome: "not_found",
});
await controller.abort(BACKGROUND_CONTEXT);
const view = await conversation.viewState(BACKGROUND_CONTEXT);
try {
expect((view.value.docs["pi.live"] as LiveState | undefined)?.run).toBeUndefined();
} finally {
view.dispose();
}
if (!first.accepted) throw new Error("Prompt was rejected");
await expect(controller.waitForPrompt(first.operationId, BACKGROUND_CONTEXT)).resolves.toMatchObject({
status: "unanswered",
text: null,
});
} finally {
await close();
}
});
test("reports accepted successful and failed operations", async () => {
const success = createAgentController({
prompt: async () => ({ ok: true, value: completed }),
} as unknown as AgentLane);
await expect(success.prompt({ message: "hello", images: null }, BACKGROUND_CONTEXT)).resolves.toEqual({
accepted: true,
operationId: "operation-1",
error: null,
});
const failed = createAgentController({
prompt: async () => ({
ok: true,
value: {
...completed,
operationId: "operation-2",
status: "failed",
error: { code: "provider", message: "failed", details: { status: 500 } },
},
}),
} as unknown as AgentLane);
await expect(failed.prompt({ message: "hello", images: null }, BACKGROUND_CONTEXT)).resolves.toEqual({
accepted: true,
operationId: "operation-2",
error: { code: "provider", message: "failed" },
});
});
test.each(admissionErrors)("maps admission error %# to a stable response", async (error, code, operationId) => {
const controller = createAgentController({
prompt: async () => ({ ok: false, error }),
} as unknown as AgentLane);
await expect(controller.prompt({ message: "hello", images: null }, BACKGROUND_CONTEXT)).resolves.toEqual({
accepted: false,
operationId,
error: { code, message: error.message },
});
test("starts a compaction task", async () => {
const { harness, conversation, close } = await openFauxConversation();
const controller = createAgentController(harness, conversation);
try {
await expect(controller.compact({ customInstructions: "short" }, BACKGROUND_CONTEXT)).resolves.toEqual({
accepted: true,
operationId: expect.any(String),
error: null,
});
} finally {
await close();
}
});
});

View file

@ -13,13 +13,7 @@ import {
FACET_BUNDLE_ARTIFACT_FORMAT_VERSION,
type FacetBundleArtifact,
} from "@earendil-works/chord/node";
import {
type AgentLane,
type LaneSnapshot,
type LaneTranscriptSnapshot,
type LaneWatchEvent,
reduceLaneSnapshot,
} from "@earendil-works/pi-agent-core";
import { fauxAssistantMessage } from "@earendil-works/pi-ai";
import { ProcessTerminal, TuiMainScreen } from "@earendil-works/pi-tui";
import { beforeAll, describe, expect, test, vi } from "vitest";
import { type ClientTuiServer, ExperimentalClientTui } from "../src/experimental/client-tui.ts";
@ -40,8 +34,9 @@ import {
SessionManagement,
type SessionSummary,
} from "../src/experimental/services/sessions.ts";
import { Transcript, type TranscriptState } from "../src/experimental/services/transcript.ts";
import { Transcript } from "../src/experimental/services/transcript.ts";
import { initTheme } from "../src/modes/interactive/theme/theme.ts";
import { openFauxConversation } from "./experimental-durable-support.ts";
const serverId = "00000000-0000-4000-8000-000000000001";
@ -67,33 +62,6 @@ function publishReplacement<T extends object>(state: MutableReplicatedState<T>,
state.replace(BACKGROUND_CONTEXT, value);
}
function laneSnapshot(): LaneSnapshot {
return {
lane: "main",
transcript: [],
tipId: null,
configuration: {
model: { provider: "test", modelId: "one" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
};
}
describe("experimental client TUI", () => {
beforeAll(() => initTheme("dark"));
@ -136,85 +104,22 @@ describe("experimental client TUI", () => {
draft.configuration.thinkingLevel = thinkingLevel;
});
});
const transcriptState = replicatedState<TranscriptState>({
snapshot: laneSnapshot() as LaneTranscriptSnapshot,
event: null,
});
const emitTranscriptEvent = (event: LaneWatchEvent): void => {
transcriptState.change(BACKGROUND_CONTEXT, (draft) => {
if (reduceLaneSnapshot(draft.snapshot as unknown as LaneSnapshot, event) === "rebase") {
throw new Error("Test transcript event unexpectedly requires a rebase");
}
draft.event = event;
});
};
let finishPrompt!: () => void;
const promptFinished = new Promise<void>((resolve) => {
finishPrompt = resolve;
});
const prompt = vi.fn(async () => {
emitTranscriptEvent({
type: "run_start",
lane: "main",
runId: "run-1",
startedAt: 1,
});
await promptFinished;
emitTranscriptEvent({
type: "entry_added",
lane: "main",
entry: {
id: "entry-user",
parentId: null,
seq: 1,
timestamp: 1,
type: "message",
message: { role: "user", content: [{ type: "text", text: "hello" }], timestamp: 1 },
},
});
emitTranscriptEvent({
type: "entry_added",
lane: "main",
entry: {
id: "entry-assistant",
parentId: "entry-user",
seq: 2,
timestamp: 2,
type: "message",
message: {
role: "assistant",
content: [{ type: "text", text: "remote answer" }],
provider: "test",
model: "one",
api: "test",
usage: transcriptState.value.snapshot!.stats.usage,
stopReason: "stop",
timestamp: 2,
},
},
});
emitTranscriptEvent({
type: "run_end",
lane: "main",
runId: "run-1",
status: "completed",
fromTipId: null,
tipId: "entry-assistant",
endedAt: 2,
});
return {
ok: true as const,
value: {
operationId: "run-1",
kind: "run" as const,
status: "completed" as const,
fromTipId: null,
tipId: "entry-assistant",
startedAt: 1,
endedAt: 2,
},
};
});
const prompts: string[] = [];
const durable = await openFauxConversation([
async (context) => {
const last = context.messages.at(-1);
if (last?.role === "user") {
prompts.push(typeof last.content === "string" ? last.content : JSON.stringify(last.content));
}
await promptFinished;
return fauxAssistantMessage("remote answer");
},
]);
const transcriptState = await durable.conversation.viewState(BACKGROUND_CONTEXT);
const reloadSource =
'"use strict";\nconst { defineFacet, defineService } = require("@earendil-works/chord");\nconst Models = defineService("pi.models");\nmodule.exports = { __esModule: true, default: defineFacet({ id: "test-tui-facet", setup(env) { env.use(Models); } }) };\n';
@ -262,7 +167,7 @@ describe("experimental client TUI", () => {
select,
selectThinking,
});
sessionProvider.provide(AgentController, createAgentController({ prompt } as unknown as AgentLane));
sessionProvider.provide(AgentController, createAgentController(durable.harness, durable.conversation));
sessionProvider.provide(Transcript, { state: transcriptState });
const serverNamespace = createRemoteServiceBinding({
@ -356,7 +261,7 @@ describe("experimental client TUI", () => {
expect(select).not.toHaveBeenCalled();
expect(component.render(80).join("\n")).toContain(`Server: ${serverId}`);
expect(component.render(80).join("\n")).toContain(`Session: ${sessionId}`);
expect(component.render(80).join("\n")).toContain("test/one");
expect(component.render(80).join("\n")).toContain("faux/faux-1");
expect(component.render(80).join("\n")).not.toContain("Experimental Sessions");
expect(component.render(80).join("\n")).not.toContain("Experimental Models");
@ -365,13 +270,12 @@ describe("experimental client TUI", () => {
component.handleInput("hello");
component.handleInput("\r");
await vi.waitFor(() => expect(prompt).toHaveBeenCalledWith("hello", undefined, BACKGROUND_CONTEXT));
await vi.waitFor(() => expect(prompts).toEqual(["hello"]));
await vi.waitFor(() => expect(component.render(80).join("\n")).toContain("Working..."));
finishPrompt();
await vi.waitFor(() => expect(component.render(80).join("\n")).toContain("remote answer"));
expect(component.render(80).join("\n")).toContain("hello");
expect(component.render(80).join("\n")).not.toContain("Working...");
expect(component.render(80).join("\n")).not.toContain("Operation run-1 completed");
component.handleInput("/reload");
component.handleInput("\u001b");
@ -444,6 +348,8 @@ describe("experimental client TUI", () => {
]);
serverProvider.dispose();
sessionProvider.dispose();
transcriptState.dispose();
await durable.close();
}
},
);

View file

@ -0,0 +1,39 @@
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { createModels, type FauxProviderHandle, type FauxResponseStep, fauxProvider } from "@earendil-works/pi-ai";
import { type Conversation, createRegistry, Harness, MemoryStorage } from "@earendil-works/pi-durable";
export interface FauxConversation {
readonly faux: FauxProviderHandle;
readonly harness: Harness;
readonly conversation: Conversation;
close(): Promise<void>;
}
/** An in-memory durable Harness whose root conversation answers with the faux provider. */
export async function openFauxConversation(responses: FauxResponseStep[] = []): Promise<FauxConversation> {
const faux = fauxProvider();
faux.setResponses(responses);
const models = createModels();
models.setProvider(faux.provider);
const harness = await Harness.open(new MemoryStorage(), { models, registry: createRegistry() }, BACKGROUND_CONTEXT);
const model = faux.getModel();
const conversation = await harness.root(BACKGROUND_CONTEXT, {
agent: { model: { provider: model.provider, modelId: model.id } },
});
return { faux, harness, conversation, close: () => harness.close(BACKGROUND_CONTEXT) };
}
/** A faux response that never arrives; it ends only when its generation is aborted. */
export function pendingResponse(): { readonly step: FauxResponseStep; readonly reached: Promise<void> } {
let reach!: () => void;
const reached = new Promise<void>((resolve) => {
reach = resolve;
});
const step: FauxResponseStep = (_context, options) =>
new Promise((_resolve, reject) => {
reach();
const signal = options!.signal!;
signal.addEventListener("abort", () => reject(signal.reason), { once: true });
});
return { step, reached };
}

View file

@ -1,8 +1,9 @@
import { defineFacet, type FacetLoader } from "@earendil-works/chord";
import { type AgentLane, BACKGROUND_CONTEXT, type LaneSnapshot } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { describe, expect, test, vi } from "vitest";
import { SessionPlugins } from "../src/experimental/services/plugins.ts";
import { createSessionWorkerServices } from "../src/experimental/services/worker.ts";
import { openFauxConversation } from "./experimental-durable-support.ts";
describe("experimental plugin reload", () => {
test("loads and cuts over a fresh Session facet generation", async () => {
@ -29,50 +30,10 @@ describe("experimental plugin reload", () => {
};
},
};
const snapshot: LaneSnapshot = {
lane: "main",
transcript: [],
tipId: null,
configuration: {
model: { provider: "test", modelId: "model" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
};
const lane = {
async watch() {
return {
snapshot,
start() {},
async resnapshot() {
return snapshot;
},
unsubscribe() {},
};
},
async getModel() {
return undefined;
},
async getThinkingLevel() {
return "off" as const;
},
} as unknown as AgentLane;
const { harness, conversation, close } = await openFauxConversation();
const services = await createSessionWorkerServices({
lane,
harness,
conversation,
modelRuntime: undefined,
facetLoader,
publish: vi.fn(async () => {}),
@ -88,6 +49,7 @@ describe("experimental plugin reload", () => {
expect(disposals).toEqual([1]);
} finally {
await services.dispose();
await close();
}
expect(disposals).toEqual([1, 2]);
});

View file

@ -589,7 +589,7 @@ describe("experimental durable server composition", () => {
await expect(services.dispose(BACKGROUND_CONTEXT)).resolves.toBeUndefined();
});
test("streams prompt events through the worker-owned service provider", async ({ onTestFinished }) => {
test("answers a client prompt through the worker-owned service provider", async ({ onTestFinished }) => {
const spawn = vi
.spyOn(processRuntime, "spawnInternalProcess")
.mockImplementation((role, args, options) =>
@ -601,32 +601,13 @@ describe("experimental durable server composition", () => {
);
onTestFinished(() => spawn.mockRestore());
const { directory } = await makeServer();
const eventTypes: string[] = [];
const result = await runClient(
{ command: "client", sessionId: "demo-1", prompt: "question" },
{
directory,
onEvent(event) {
eventTypes.push(event.type);
},
},
);
const result = await runClient({ command: "client", sessionId: "demo-1", prompt: "question" }, { directory });
expect(result).toMatchObject({ kind: "prompted", text: "deterministic remote answer" });
expect(eventTypes).toEqual(
expect.arrayContaining([
"run_start",
"message_start",
"message_update",
"message_end",
"entry_added",
"run_end",
]),
);
});
test("replicates terminal operation state after consecutive prompts", async ({ onTestFinished }) => {
test("replicates the transcript after consecutive prompts", async ({ onTestFinished }) => {
const spawn = vi
.spyOn(processRuntime, "spawnInternalProcess")
.mockImplementation((role, args, options) =>
@ -648,11 +629,16 @@ describe("experimental durable server composition", () => {
for (const message of ["first question", "second question"]) {
const response = await controller.prompt({ message, images: null }, BACKGROUND_CONTEXT);
expect(response).toMatchObject({ accepted: true, operationId: expect.any(String) });
if (!response.accepted) throw new Error(response.error.message);
await expect(controller.waitForPrompt(response.operationId, BACKGROUND_CONTEXT)).resolves.toEqual({
status: "done",
text: "deterministic remote answer",
reason: null,
});
await vi.waitFor(() => {
expect(transcript.state.value?.snapshot).toMatchObject({
operation: null,
lastResult: { operationId: response.operationId },
});
expect(transcript.state.value?.entries.filter((entry) => entry.kind === "pi.assistant")).toHaveLength(
message === "first question" ? 1 : 2,
);
});
}
} finally {
@ -803,17 +789,14 @@ describe("experimental durable server composition", () => {
runClient({ command: "client", sessionId: "demo-1" }, { directory: sharedDirectory }),
).rejects.toThrow("Session demo-1 is available from more than one server");
});
test("rejects a duplicate session ID within one durable repository", async () => {
await createExperimentalSessions(
join(agentDir, "experimental", "sessions"),
["demo-1"],
join(agentDir, "other-cwd"),
);
const { directory } = await makeServer();
await expect(runClient({ command: "client", sessionId: "demo-1" }, { directory })).rejects.toMatchObject({
code: "session_ambiguous",
});
test("rejects a duplicate session ID within one session directory", async () => {
await expect(
createExperimentalSessions(
join(agentDir, "experimental", "sessions"),
["demo-1"],
join(agentDir, "other-cwd"),
),
).rejects.toThrow("Session demo-1 already exists");
});
});

View file

@ -1,33 +1,23 @@
import { mkdir, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { AgentDoc, createSession, type ModelRef, ROOT_CONVERSATION_ID } from "@earendil-works/pi-durable";
import { openNodeSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/node";
import {
BACKGROUND_CONTEXT,
type Entry,
type JsonlSessionMetadata,
JsonlSessionRepo,
laneConfig,
} from "@earendil-works/pi-agent-core";
import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/node";
createSession as createCatalogSession,
readSession,
type SessionCatalogMetadata,
sessionStoragePath,
} from "../src/experimental/session-catalog.ts";
export async function createExperimentalSessions(
sessionsRoot: string,
ids: readonly string[],
cwd = process.cwd(),
): Promise<JsonlSessionMetadata[]> {
const fileSystem = new NodeExecutionEnv({ cwd });
const repo = new JsonlSessionRepo({ fileSystem, sessionsRoot });
const metadata: JsonlSessionMetadata[] = [];
try {
for (const id of ids) {
const session = await repo.create({ id, cwd }, BACKGROUND_CONTEXT);
metadata.push(session.metadata);
await session.close(BACKGROUND_CONTEXT);
}
return metadata;
} finally {
await repo.close(BACKGROUND_CONTEXT);
await fileSystem.cleanup(BACKGROUND_CONTEXT);
}
): Promise<SessionCatalogMetadata[]> {
const metadata: SessionCatalogMetadata[] = [];
for (const id of ids) metadata.push(await createCatalogSession(sessionsRoot, { id, cwd }));
return metadata;
}
export async function configureExperimentalWorkerModel(agentDir: string): Promise<void> {
@ -37,31 +27,18 @@ export async function configureExperimentalWorkerModel(agentDir: string): Promis
});
}
/** The root conversation's model, read from the Session's durable storage while no worker owns it. */
export async function readExperimentalSessionState(
sessionsRoot: string,
sessionId: string,
): Promise<{
branch: Entry[];
model: { provider: string; modelId: string } | undefined;
activeTools: string[];
}> {
const fileSystem = new NodeExecutionEnv({ cwd: process.cwd() });
const repo = new JsonlSessionRepo({ fileSystem, sessionsRoot });
let session: Awaited<ReturnType<JsonlSessionRepo["open"]>> | undefined;
): Promise<{ model: ModelRef | undefined }> {
const metadata = await readSession(sessionsRoot, sessionId);
if (metadata === undefined) throw new Error(`Expected Session ${sessionId}`);
const session = createSession(await openNodeSqliteStorage(sessionStoragePath(metadata)));
try {
const matches = (await repo.list(undefined, BACKGROUND_CONTEXT)).filter((metadata) => metadata.id === sessionId);
if (matches.length !== 1) throw new Error(`Expected one Session ${sessionId}, found ${matches.length}`);
session = await repo.open(matches[0]!, BACKGROUND_CONTEXT);
const main = await session.branch("main", BACKGROUND_CONTEXT);
if (main === undefined) throw new Error("Expected Session main Branch");
const [branch, configuration] = await Promise.all([
main.findEntries({ order: "oldestFirst" }, BACKGROUND_CONTEXT),
session.getValue(laneConfig("main"), BACKGROUND_CONTEXT),
]);
return { branch, model: configuration?.value.model, activeTools: configuration?.value.activeToolNames ?? [] };
const agent = await session.snapshot(AgentDoc, ROOT_CONVERSATION_ID, BACKGROUND_CONTEXT);
return { model: agent?.model };
} finally {
await session?.close(BACKGROUND_CONTEXT);
await repo.close(BACKGROUND_CONTEXT);
await fileSystem.cleanup(BACKGROUND_CONTEXT);
await session.close(BACKGROUND_CONTEXT);
}
}

View file

@ -23,11 +23,11 @@ describe("Session worker lifecycle", () => {
vi.useFakeTimers();
const { lifecycle, retire } = createLifecycle();
lifecycle.setDemand(GENERATION, "attachment-1", true);
lifecycle.operationStarted("run", "main", "operation-1");
lifecycle.setHarnessActive(true);
lifecycle.setDemand(GENERATION, "attachment-1", false);
expect(retire).not.toHaveBeenCalled();
lifecycle.operationStopped("run", "main", "operation-1");
lifecycle.setHarnessActive(false);
await vi.runAllTicks();
expect(retire).toHaveBeenCalledOnce();
lifecycle.close();
@ -45,30 +45,25 @@ describe("Session worker lifecycle", () => {
lifecycle.close();
});
test("tracks a nested compaction independently from its enclosing run", () => {
test("does not retire on repeated activity updates while the Harness stays active", () => {
vi.useFakeTimers();
const { lifecycle, retire } = createLifecycle();
lifecycle.setDemand(GENERATION, "attachment-1", true);
lifecycle.operationStarted("run", "main", "operation-1");
lifecycle.operationStarted("compaction", "main", "operation-1");
lifecycle.setHarnessActive(true);
lifecycle.setDemand(GENERATION, "attachment-1", false);
lifecycle.operationStopped("compaction", "main", "operation-1");
lifecycle.setHarnessActive(true);
expect(retire).not.toHaveBeenCalled();
lifecycle.operationStopped("run", "main", "operation-1");
lifecycle.setHarnessActive(false);
expect(retire).toHaveBeenCalledOnce();
lifecycle.close();
});
test.each(["compaction", "navigation"] as const)("clears suspended %s activity", (kind) => {
test("an idle Harness update does not retire while demand remains", () => {
vi.useFakeTimers();
const { lifecycle, retire } = createLifecycle();
lifecycle.setDemand(GENERATION, "attachment-1", true);
lifecycle.operationStarted(kind, "main", "operation-1");
lifecycle.setDemand(GENERATION, "attachment-1", false);
lifecycle.operationStopped(kind, "main", "operation-1");
expect(retire).toHaveBeenCalledOnce();
lifecycle.setHarnessActive(false);
expect(retire).not.toHaveBeenCalled();
lifecycle.close();
});

View file

@ -1,16 +1,15 @@
import type { ServiceCall } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type JsonlSessionMetadata } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { afterEach, describe, expect, test, vi } from "vitest";
import type { CoordinatorConnectionEvent } from "../src/experimental/coordinator.ts";
import type { SessionCatalogMetadata } from "../src/experimental/session-catalog.ts";
import { SessionWorkerManager } from "../src/experimental/session-worker-manager.ts";
const metadata: JsonlSessionMetadata = {
const metadata: SessionCatalogMetadata = {
id: "session-1",
createdAt: 1,
storageVersion: 1,
cwd: "/tmp",
path: "/tmp/session-1.jsonl",
modifiedAt: 1,
path: "/tmp/session-1",
};
class FakeCoordinator {

View file

@ -1,108 +1,34 @@
import { replicatedState } from "@earendil-works/chord";
import { createFacetHost, defineFacet } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { AgentLane, EventListener, HarnessEvent, LaneSnapshot, WatchHandle } from "@earendil-works/pi-agent-core";
import { describe, expect, test, vi } from "vitest";
import type { TranscriptState } from "../src/experimental/services/transcript.ts";
import { createTranscriptService } from "../src/experimental/services/transcript-provider.ts";
function laneSnapshot(tipId: string | null = null): LaneSnapshot {
return {
lane: "main",
transcript: [],
tipId,
configuration: {
model: { provider: "test", modelId: "model" },
thinkingLevel: "off",
activeToolNames: [],
},
stats: {
messageCount: 0,
usage: {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
},
},
operation: null,
queues: [],
faulted: false,
};
}
import { fauxAssistantMessage } from "@earendil-works/pi-ai";
import type { ConversationView } from "@earendil-works/pi-durable";
import { describe, expect, test } from "vitest";
import { Transcript } from "../src/experimental/services/transcript.ts";
import { createTranscriptServiceFacet } from "../src/experimental/services/transcript-provider.ts";
import { openFauxConversation } from "./experimental-durable-support.ts";
describe("Transcript service", () => {
test("publishes coherent state and rebases after navigation", async () => {
let listener: EventListener | undefined;
const replacement = laneSnapshot("replacement-tip");
const resnapshot = vi.fn(async () => replacement);
const unsubscribe = vi.fn();
const handle: WatchHandle<LaneSnapshot> = {
snapshot: laneSnapshot(),
start(next) {
listener = next;
test("replicates the conversation view as it changes", async () => {
const { conversation, close } = await openFauxConversation([fauxAssistantMessage("answer")]);
const views: ConversationView[] = [];
const consumer = defineFacet({
id: "test-transcript-consumer",
setup(env) {
const transcript = env.use(Transcript);
env.onActivate(() => env.own(transcript.state.subscribe((value) => void views.push(value))));
},
resnapshot,
unsubscribe,
};
const lane = { watch: async () => handle } as unknown as AgentLane;
const runtime = createTranscriptService(lane, replicatedState);
const states: TranscriptState[] = [];
runtime.service.state.subscribe((value) => {
if (value.snapshot !== null) states.push(value);
});
await runtime.activate();
expect(states).toHaveLength(1);
expect(states[0]).toMatchObject({ snapshot: { lane: "main", operation: null }, event: null });
await listener?.({ type: "run_start", lane: "main", runId: "run-1", startedAt: 1 }, BACKGROUND_CONTEXT);
expect(states).toHaveLength(2);
expect(states[1]).toMatchObject({
snapshot: { operation: { id: "run-1" } },
event: { type: "run_start", runId: "run-1" },
const host = await createFacetHost({
facets: [await createTranscriptServiceFacet(conversation, BACKGROUND_CONTEXT), consumer],
});
await listener?.(
{
type: "entry_added",
lane: "main",
entry: {
id: "entry-1",
parentId: null,
seq: 1,
timestamp: 2,
type: "message",
message: { role: "user", content: "hello", timestamp: 2 },
},
},
BACKGROUND_CONTEXT,
);
expect(states.at(-1)).toMatchObject({
snapshot: { tipId: "entry-1", transcript: [{ id: "entry-1" }] },
event: { type: "entry_added" },
});
const navigation: HarnessEvent = {
type: "navigation_end",
lane: "main",
runId: "navigation-1",
status: "completed",
fromTipId: "entry-1",
tipId: "replacement-tip",
endedAt: 3,
};
await listener?.(navigation, BACKGROUND_CONTEXT);
await vi.waitFor(() => expect(states.at(-1)?.event).toBeNull());
expect(states.at(-2)).toMatchObject({
snapshot: { tipId: "entry-1" },
event: { type: "navigation_end" },
});
expect(states.at(-1)).toMatchObject({ snapshot: { tipId: "replacement-tip" }, event: null });
expect(runtime.service.state.value).toMatchObject({ snapshot: { tipId: "replacement-tip" }, event: null });
expect(resnapshot).toHaveBeenCalledOnce();
await runtime.dispose();
expect(unsubscribe).toHaveBeenCalledOnce();
try {
expect(views[0]?.entries).toEqual([]);
const submission = await conversation.submit({ type: "input", content: "question" }, BACKGROUND_CONTEXT);
await submission.wait(BACKGROUND_CONTEXT);
await expect.poll(() => views.at(-1)?.entries.map((entry) => entry.kind)).toEqual(["pi.user", "pi.assistant"]);
} finally {
await host.dispose();
await close();
}
});
});

View file

@ -1,8 +1,10 @@
import { resolve } from "node:path";
import { fileURLToPath } from "node:url";
import { createStaticFacetLoader, defineFacet } from "@earendil-works/chord";
import { AgentHarness, BACKGROUND_CONTEXT } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { createModels, fauxAssistantMessage, fauxProvider } from "@earendil-works/pi-ai";
import { createRegistry, Harness } from "@earendil-works/pi-durable";
import { openNodeSqliteStorage } from "@earendil-works/pi-durable/storage/sqlite/node";
import { consumeInternalProcessRole } from "../../src/experimental/process.ts";
import { runSessionWorkerWithHarness } from "../../src/experimental/session-worker.ts";
import { KeyedProbe } from "./keyed-service.ts";
@ -10,26 +12,24 @@ import { KeyedProbe } from "./keyed-service.ts";
if (process.argv[1] !== undefined && resolve(process.argv[1]) === fileURLToPath(import.meta.url)) {
const role = consumeInternalProcessRole();
if (role !== "session-worker") throw new Error("Faux Session worker requires a session-worker invocation");
void runSessionWorkerWithHarness(process.argv.slice(2), async (session, options) => {
void runSessionWorkerWithHarness(process.argv.slice(2), async (databasePath, options) => {
if (options.provider !== "anthropic" || options.model !== "claude-sonnet-4-5") {
throw new Error(`Unexpected faux worker model: ${options.provider}/${options.model}`);
}
const faux = fauxProvider();
faux.setResponses([fauxAssistantMessage("deterministic remote answer", { timestamp: 20 })]);
const answer = fauxAssistantMessage("deterministic remote answer", { timestamp: 20 });
faux.setResponses([answer, answer]);
const models = createModels();
models.setProvider(faux.provider);
const harness = (
await AgentHarness.create(
{
session,
models,
model: faux.getModel(),
tools: [],
resources: {},
},
BACKGROUND_CONTEXT,
)
).harness;
const harness = await Harness.open(
await openNodeSqliteStorage(databasePath),
{ models, registry: createRegistry() },
BACKGROUND_CONTEXT,
);
const model = faux.getModel();
const conversation = await harness.root(BACKGROUND_CONTEXT, {
agent: { cwd: options.metadata.cwd, model: { provider: model.provider, modelId: model.id } },
});
const keyedProbeFacet = defineFacet({
id: "@test/keyed-probe",
setup(env) {
@ -56,7 +56,7 @@ if (process.argv[1] !== undefined && resolve(process.argv[1]) === fileURLToPath(
env.onActivate(() => spawn("first"));
},
});
return { harness, facetLoader: createStaticFacetLoader([keyedProbeFacet]) };
return { harness, conversation, facetLoader: createStaticFacetLoader([keyedProbeFacet]) };
}).catch((error: unknown) => {
console.error(error);
process.exit(1);

View file

@ -14,7 +14,7 @@ A server target contains `{ serverId }`; a Session target contains `{ serverId,
Chord owns the payload semantics carried inside these envelopes: `{ serviceId, instance?, member, args }` calls, the `$chord.service` control vocabulary, service catalogues, subscription snapshots and updates, service error codes, and the independent Delta path codecs for replicated states. `pi-protocol` validates that each opaque payload is strict JSON but does not validate or export its Chord grammar. Clients and servers parse those values through `@earendil-works/chord` at the service adapter boundary.
Session-directory state, management results, transcripts, models, plugins, and all other application values remain opaque service data. The real `Session` and `AgentHarness` remain process-local. Server and Session calls route opaquely to their owning providers, where Chord and the application validate and invoke them.
Session-directory state, management results, transcripts, models, plugins, and all other application values remain opaque service data. The durable Session and its Harness remain local to the Session worker process. Server and Session calls route opaquely to their owning providers, where Chord and the application validate and invoke them.
Server and worker lifecycle is intentionally outside this public protocol. The experimental local coordinator is only an opaque message router; each replaceable server process owns the private lifecycle protocol.

View file

@ -2,6 +2,10 @@
## [Unreleased]
### Breaking Changes
- `SessionMetadata` is now exported by `@earendil-works/pi-server` and requires only `id`; the package no longer depends on `@earendil-works/pi-agent-core`. The testing `TestServerHost` keeps an in-memory session map instead of a `MemorySessionRepo`, and `TestHarness` exposes `metadata` instead of `session`.
## [0.99.2] - 2026-09-30
## [0.99.1] - 2026-09-29

View file

@ -1,6 +1,6 @@
# @earendil-works/pi-server
Experimental local server for the new durable Session and Agent Harness interfaces.
Experimental local server that routes clients to application-hosted durable Sessions.
The current slice supports server- and Session-scoped facet-service routing and multi-presentation attachment. `RoutedServerServiceHost.attachClient()` creates one connection-scoped server service endpoint with narrow attachment-management capabilities. `RoutedSessionHandle.attachClient()` returns a presentation-scoped Session capability. Its `invokeService()` forwards an opaque service/member envelope to the selected Session endpoint; the server validates the attachment route but does not load the facet contract.
@ -16,48 +16,32 @@ A Session may have multiple presentation attachments. Repeating `attach` from on
```ts
import { randomUUID } from "node:crypto";
import { MemorySessionRepo, type Session } from "@earendil-works/pi-agent-core";
import {
type RoutedServerServiceHost,
type RoutedSessionHandle,
type ServerHost,
SessionAmbiguousError,
type SessionMetadata,
SessionNotFoundError,
} from "@earendil-works/pi-server";
import { createUnixServer, getUnixSocketPath } from "@earendil-works/pi-server/unix";
interface StoredSession extends SessionMetadata {
path: string;
}
async function startServer(
serverServices: RoutedServerServiceHost,
openRoutedSession: (session: Session) => Promise<RoutedSessionHandle>,
sessions: Map<string, StoredSession>,
openRoutedSession: (session: StoredSession) => Promise<RoutedSessionHandle>,
) {
const sessions = new MemorySessionRepo();
const host: ServerHost = {
const host: ServerHost<StoredSession> = {
serverServices,
async resolveSession(sessionId, context) {
const matches = (await sessions.list(undefined, context))
.filter((metadata) => metadata.id === sessionId);
if (matches.length === 0) {
throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
}
if (matches.length > 1) throw new SessionAmbiguousError();
return matches[0];
},
async openSession(metadata, context) {
const session = await sessions.open(metadata, context);
try {
return await openRoutedSession(session);
} catch (error) {
try {
await session.close(context);
} catch (cleanupError) {
throw new AggregateError(
[error, cleanupError],
"Harness creation and Session cleanup failed",
);
}
throw error;
}
async resolveSession(sessionId) {
const metadata = sessions.get(sessionId);
if (!metadata) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
return metadata;
},
openSession: (metadata) => openRoutedSession(metadata),
};
const serverId = randomUUID();
@ -70,7 +54,7 @@ async function startServer(
}
```
Applications supply a required server service host, a bounded Session resolver, and a routed Session factory. Session discovery and management are application-owned services; the protocol server only asks the resolver for metadata when routing an attachment. The host owns acquiring the worker-local Session and Harness. Failures are cleaned up in that worker. Neither an open JavaScript Session nor a Harness crosses the process boundary.
Applications supply a required server service host, a bounded Session resolver, and a routed Session factory. `SessionMetadata` only requires an `id`; applications may extend it with their own storage fields. Session discovery and management are application-owned services; the protocol server only asks the resolver for metadata when routing an attachment. The host owns acquiring the worker-local Session and Harness. Failures are cleaned up in that worker. Neither an open JavaScript Session nor a Harness crosses the process boundary.
`serverId` is a logical identity supplied by the launcher, not a socket address. The Unix preset requires an explicit physical `path`; `getUnixSocketPath()` derives one from a caller-selected directory. Choose a short, private runtime directory rather than deriving the route from an unbounded home-directory path. A long-lived launcher can reuse the same ID and path when replacing a server process.

View file

@ -48,7 +48,6 @@
},
"dependencies": {
"@earendil-works/chord": "^0.99.2",
"@earendil-works/pi-agent-core": "^0.99.2",
"@earendil-works/pi-protocol": "^0.99.2"
},
"devDependencies": {

View file

@ -8,7 +8,7 @@ import {
type ServiceCall,
type ServiceProviderUpdate,
} from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type SessionMetadata, TODO_CONTEXT, withAbortSignal } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT, TODO_CONTEXT, withAbortSignal } from "@earendil-works/chord/context";
import {
type CancelEnvelope,
type ClientHello,
@ -37,7 +37,7 @@ import {
import { INTERNAL_SERVER_ERROR_MESSAGE, ServerError, WrongServerError } from "./errors.ts";
import type { ServerListener } from "./listener.ts";
import { SessionRouter } from "./session-router.ts";
import type { ServerHost, ServerOptions } from "./types.ts";
import type { ServerHost, ServerOptions, SessionMetadata } from "./types.ts";
const DEFAULT_HANDSHAKE_TIMEOUT_MS = 5_000;
const MAX_UINT32 = 0xffff_ffff;

View file

@ -1,9 +1,9 @@
import { randomUUID } from "node:crypto";
import type { JsonValue, ServiceCall, ServiceProviderUpdate } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type Context, type SessionMetadata } from "@earendil-works/pi-agent-core";
import type { Context, JsonValue, ServiceCall, ServiceProviderUpdate } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import type { RpcTarget, SessionTarget } from "@earendil-works/pi-protocol";
import { ServerDrainingError, SessionNotAttachedError } from "./errors.ts";
import type { RoutedSessionAttachment, RoutedSessionHandle, ServerHost } from "./types.ts";
import type { RoutedSessionAttachment, RoutedSessionHandle, ServerHost, SessionMetadata } from "./types.ts";
class SessionCleanupError extends AggregateError {}

View file

@ -1,8 +1,6 @@
import type { JsonValue, ServiceCall } from "@earendil-works/chord";
import type { Context, Session, SessionMetadata } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT, MemorySessionRepo } from "@earendil-works/pi-agent-core";
import { SessionAmbiguousError, SessionNotFoundError } from "../errors.ts";
import type { RoutedServerServiceHost, RoutedSessionHandle, ServerHost } from "../types.ts";
import type { Context, JsonValue, ServiceCall } from "@earendil-works/chord";
import { SessionNotFoundError } from "../errors.ts";
import type { RoutedServerServiceHost, RoutedSessionHandle, ServerHost, SessionMetadata } from "../types.ts";
export class Deferred<T> {
readonly promise: Promise<T>;
@ -25,7 +23,7 @@ interface OpenGate {
}
export class TestHarness {
readonly session: Session;
readonly metadata: SessionMetadata;
readonly closed = new Deferred<void>();
readonly #termination = new Deferred<Error | undefined>();
readonly terminated = this.#termination.promise;
@ -40,8 +38,8 @@ export class TestHarness {
private nextCloseGate?: OpenGate;
private nextServiceGate?: OpenGate;
constructor(session: Session) {
this.session = session;
constructor(metadata: SessionMetadata) {
this.metadata = metadata;
}
attachClient(_context: Context): {
@ -80,7 +78,7 @@ export class TestHarness {
return result;
}
async close(context: Context): Promise<void> {
async close(_context: Context): Promise<void> {
this.closeCount += 1;
const gate = this.nextCloseGate;
if (gate) {
@ -93,13 +91,11 @@ export class TestHarness {
this.failClose = undefined;
throw error;
}
await this.session.close(context);
this.closed.resolve(undefined);
this.#termination.resolve(undefined);
}
async terminate(error: Error): Promise<void> {
await this.session.close(BACKGROUND_CONTEXT);
this.#termination.resolve(error);
}
@ -150,21 +146,20 @@ export function createTestServerServices(): RoutedServerServiceHost {
export class TestServerHost implements ServerHost {
readonly serverServices = createTestServerServices();
readonly repo = new MemorySessionRepo({ now: () => 1 });
readonly sessions = new Map<string, SessionMetadata>();
readonly harnesses = new Map<string, TestHarness[]>();
openSessionCount = 0;
nextOpenSessionError?: Error;
nextHarnessCloseError?: Error;
private nextOpenSessionGate?: OpenGate;
async resolveSession(sessionId: string, context: Context): Promise<SessionMetadata> {
const matches = (await this.repo.list(undefined, context)).filter(({ id }) => id === sessionId);
if (matches.length === 0) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
if (matches.length > 1) throw new SessionAmbiguousError();
return matches[0]!;
async resolveSession(sessionId: string, _context: Context): Promise<SessionMetadata> {
const metadata = this.sessions.get(sessionId);
if (!metadata) throw new SessionNotFoundError(`Unknown session: ${sessionId}`);
return metadata;
}
async openSession(metadata: SessionMetadata, context: Context): Promise<RoutedSessionHandle> {
async openSession(metadata: SessionMetadata, _context: Context): Promise<RoutedSessionHandle> {
this.openSessionCount += 1;
const gate = this.nextOpenSessionGate;
if (gate) {
@ -172,32 +167,26 @@ export class TestServerHost implements ServerHost {
gate.entered.resolve(undefined);
await gate.release.promise;
}
const session = await this.repo.open(metadata, context);
try {
if (this.nextOpenSessionError) {
const error = this.nextOpenSessionError;
this.nextOpenSessionError = undefined;
throw error;
}
const harness = new TestHarness(session);
if (this.nextHarnessCloseError) {
harness.failClose = this.nextHarnessCloseError;
this.nextHarnessCloseError = undefined;
}
const harnesses = this.harnesses.get(metadata.id) ?? [];
harnesses.push(harness);
this.harnesses.set(metadata.id, harnesses);
return harness;
} catch (error) {
await session.close(context);
if (!this.sessions.has(metadata.id)) throw new SessionNotFoundError(`Unknown session: ${metadata.id}`);
if (this.nextOpenSessionError) {
const error = this.nextOpenSessionError;
this.nextOpenSessionError = undefined;
throw error;
}
const harness = new TestHarness(metadata);
if (this.nextHarnessCloseError) {
harness.failClose = this.nextHarnessCloseError;
this.nextHarnessCloseError = undefined;
}
const harnesses = this.harnesses.get(metadata.id) ?? [];
harnesses.push(harness);
this.harnesses.set(metadata.id, harnesses);
return harness;
}
async seed(id = "session-1", parentSessionId?: string): Promise<SessionMetadata> {
const session = await this.repo.create({ id, parentSessionId }, BACKGROUND_CONTEXT);
const metadata = session.metadata;
await session.close(BACKGROUND_CONTEXT);
async seed(id = "session-1"): Promise<SessionMetadata> {
const metadata: SessionMetadata = { id };
this.sessions.set(id, metadata);
return metadata;
}

View file

@ -1,6 +1,5 @@
import type { SessionMetadata } from "@earendil-works/pi-agent-core";
import { Server } from "../../server.ts";
import type { ServerHost } from "../../types.ts";
import type { ServerHost, SessionMetadata } from "../../types.ts";
import { createUnixListener } from "./listener.ts";
import type { UnixServerOptions } from "./types.ts";

View file

@ -1,5 +1,4 @@
import type { JsonValue, ServiceCall, ServiceProviderUpdate } from "@earendil-works/chord";
import type { Context, SessionMetadata } from "@earendil-works/pi-agent-core";
import type { Context, JsonValue, ServiceCall, ServiceProviderUpdate } from "@earendil-works/chord";
import type { ServerListener } from "./listener.ts";
export interface ServerOptions {
@ -14,6 +13,11 @@ export interface ServerOptions {
export type MaybePromise<T> = T | Promise<T>;
/** Minimal durable Session metadata the server needs for routing. Applications may extend it. */
export interface SessionMetadata {
readonly id: string;
}
/** One presentation connection's live capability for a hosted Session. */
export interface RoutedSessionAttachment {
/** Route one contract-agnostic service operation to the attached Session endpoint. */

View file

@ -1,5 +1,5 @@
import type { ServiceCall } from "@earendil-works/chord";
import { BACKGROUND_CONTEXT, type SessionMetadata } from "@earendil-works/pi-agent-core";
import { BACKGROUND_CONTEXT } from "@earendil-works/chord/context";
import { afterEach, describe, expect, test } from "vitest";
import type { ByteConnection, ByteConnectionHandler } from "../src/connection.ts";
import { SessionAmbiguousError } from "../src/errors.ts";
@ -11,7 +11,7 @@ import {
TestServerHost,
type WireChannel,
} from "../src/testing/index.ts";
import type { ServerHost } from "../src/types.ts";
import type { ServerHost, SessionMetadata } from "../src/types.ts";
const serverId = "00000000-0000-4000-8000-000000000001";
const servers = new Set<Server>();
@ -105,9 +105,6 @@ describe("Session protocol", () => {
type BackendMetadata = SessionMetadata & { path: string; modifiedAt: number };
const metadata: BackendMetadata = {
id: "session-1",
createdAt: 1,
storageVersion: 1,
cwd: "/workspace",
path: "/sessions/session-1.jsonl",
modifiedAt: 2,
};
@ -399,7 +396,7 @@ describe("Session protocol", () => {
describe("routed Session acquisition failures", () => {
test("releases a lease acquired concurrently with Harness termination", async () => {
const metadata: SessionMetadata = { id: "session-1", createdAt: 1, storageVersion: 1 };
const metadata: SessionMetadata = { id: "session-1" };
const acquiring = new Deferred<void>();
const continueAcquiring = new Deferred<void>();
const terminated = new Deferred<Error | undefined>();

View file

@ -6,11 +6,7 @@
"moduleResolution": "NodeNext",
"types": ["node", "vitest"],
"paths": {
"@earendil-works/pi-agent-core": ["../agent/src/index.ts"],
"@earendil-works/pi-ai": ["../ai/src/index.ts"],
"@earendil-works/pi-ai/*": ["../ai/src/*.ts", "../ai/src/providers/*.ts"],
"@earendil-works/pi-protocol": ["../protocol/src/index.ts"],
"@earendil-works/pi-telemetry": ["../telemetry/src/index.ts"]
"@earendil-works/pi-protocol": ["../protocol/src/index.ts"]
}
},
"include": ["src/**/*.ts", "test/**/*.ts"]