feat(coding-agent): route prompting through chat service

This commit is contained in:
Armin Ronacher 2026-08-20 11:27:25 +02:00
parent 29c41fc6cd
commit d5e3b47877
8 changed files with 228 additions and 22 deletions

View file

@ -1,6 +1,6 @@
# Coding-Agent Application Hosts and Plugin Facets
> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `RemoteState` substrate now have an experimental implementation, including `Models`, `SessionDirectory`, and `SessionManagement` vertical slices. The application host contexts, plugin kernel, events, references, telemetry propagation, and most other example facets remain illustrative. Reconcile this design with `rpc.md`, `telemetry.md`, and the final harness contract before adding it to `harness.md` or creating a work package.
> **Status:** Tentative design input, not a normative contract or implementation handoff. The transport-neutral service token, provider, singleton `use()`, keyed `observe()`, and `RemoteState` substrate now have an experimental implementation, including `Models`, `Chat`, `SessionDirectory`, and `SessionManagement` vertical slices. The application host contexts, plugin kernel, events, references, telemetry propagation, and most other example facets remain illustrative. Reconcile this design with `rpc.md`, `telemetry.md`, and the final harness contract before adding it to `harness.md` or creating a work package.
This document assumes you already understand `AgentHarness`, `AgentLane`, `Session`, `SessionTree`, `SessionRepo`, invocation `Context`, and telemetry. Read `rpc.md` for wire frames, remote references, subscriptions, and trace carriers, and `telemetry.md` for context propagation and cancellation semantics.

View file

@ -2,10 +2,11 @@ import { basename } from "node:path";
import { BACKGROUND_CONTEXT, type RemoteServiceNamespace, type RemoteState } from "@earendil-works/pi-agent-core";
import { Client } from "@earendil-works/pi-client";
import { createUnixTransportFactory, discoverUnixServers, type UnixServerRoute } from "@earendil-works/pi-client/unix";
import { isServerId, type LaneEvent, type SessionAddress } from "@earendil-works/pi-protocol";
import { isServerId, type LaneEvent, type PromptMessage, type SessionAddress } from "@earendil-works/pi-protocol";
import type { ClientCommand } from "../cli/experimental/commands/client.ts";
import { activateServer, ENV_SERVER_ID, resolveServerDirectory, resolveSessionDirectory } from "./server.ts";
import { createServerServiceNamespace } from "./services/connection.ts";
import { Chat, type ChatPromptResponse } from "./services/chat.ts";
import { createServerServiceNamespace, createSessionServiceNamespace } from "./services/connection.ts";
import { SessionDirectory, SessionManagement, type SessionManagementService } from "./services/sessions.ts";
export type ClientResult =
@ -19,7 +20,7 @@ export type ClientResult =
export interface RunClientOptions {
/** Directory searched when --connect is omitted. Defaults to PI_SERVER_DIR or ~/.pi/server. */
readonly directory?: string;
/** Reserved for snapshot-ordered main-lane events once remote harness prompting is implemented. */
/** Receives snapshot-ordered main-lane events while a prompt is active. */
readonly onEvent?: (event: LaneEvent) => void | Promise<void>;
}
@ -129,30 +130,47 @@ export async function runClient(command: ClientCommand, options: RunClientOption
return { kind: "attached", serverId: match.route.serverId, sessionId };
}
const watch = options.onEvent === undefined ? undefined : await match.client.watchSession(sessionId);
if (watch && options.onEvent) await watch.start(options.onEvent);
const sessionServices = createSessionServiceNamespace(match.client, { services: [Chat] });
serviceNamespaces.add(sessionServices);
const chat = sessionServices.use(Chat);
const completedText = new Map<string, string>();
// Chat deliberately returns no transcript content. Keep the compatibility watch until Transcript uses RemoteEvents.
const watch = await match.client.watchSession(sessionId);
await watch.start(async (event) => {
if (event.type === "message_end" && event.runId !== undefined && event.message.role === "assistant") {
completedText.set(event.runId, messageText(event.message));
} else if (event.type === "run_end" && "finalMessage" in event) {
completedText.set(event.runId, messageText(event.finalMessage));
}
await options.onEvent?.(event);
});
let response: ChatPromptResponse;
try {
const result = await match.client.promptSession(sessionId, command.prompt);
if (!result.ok) throw new Error(result.error.message);
const value = result.value;
if (value.kind === "failed") throw new Error(value.error.message);
const text =
"finalMessage" in value
? value.finalMessage.content
.filter((content) => content.type === "text")
.map((content) => content.text)
.join("")
: "";
return { kind: "prompted", serverId: match.route.serverId, sessionId, text };
response = await chat.prompt({ message: command.prompt, images: null }, BACKGROUND_CONTEXT);
} finally {
await watch?.dispose();
await watch.dispose();
}
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) ?? "",
};
} finally {
await Promise.all([...serviceNamespaces].map((services) => services.dispose(BACKGROUND_CONTEXT)));
await Promise.all([...openedClients].map((client) => client.dispose()));
}
}
function messageText(message: Extract<PromptMessage, { role: "assistant" }>): string {
return message.content
.filter((content) => content.type === "text")
.map((content) => content.text)
.join("");
}
function waitForState<T>(state: RemoteState<T>): Promise<T> {
if (state.value !== undefined) return Promise.resolve(state.value);
return new Promise((resolve) => {

View file

@ -0,0 +1,53 @@
import type { AgentHarness, RunResult as HarnessRunResult, RemoteServiceProvider } from "@earendil-works/pi-agent-core";
import type { PromptArguments } from "@earendil-works/pi-protocol";
import { toHarnessPromptArguments } from "../harness-wire-adapter.ts";
import { Chat, type ChatPromptRequest, type ChatPromptResponse } from "./chat.ts";
export function provideChatService(provider: RemoteServiceProvider, harness: AgentHarness): void {
provider.provide(Chat, {
async prompt(request, context) {
const prompt = toHarnessPromptArguments(toPromptArguments(request));
const result =
typeof prompt[0] === "string"
? await harness.prompt(prompt[0], prompt[1], context)
: await harness.prompt(prompt[0], context);
return toChatPromptResponse(result);
},
async requestAbort(operationId, context) {
const result = await harness.requestAbort(operationId, context);
if (!result.ok) throw new Error(result.error.message);
},
});
}
export function toChatPromptResponse(result: HarnessRunResult): ChatPromptResponse {
if (result.ok) {
return {
accepted: true,
operationId: result.value.runId,
error:
result.value.kind === "failed"
? {
code: result.value.error.code,
message: result.value.error.message,
}
: null,
};
}
const code = {
LaneBusy: "lane_busy",
InvalidMessage: "invalid_message",
UnknownSkill: "unknown_skill",
UnknownTemplate: "unknown_template",
Closed: "closed",
}[result.error._tag];
return {
accepted: false,
operationId: result.error._tag === "LaneBusy" ? result.error.operationId : null,
error: { code, message: result.error.message },
};
}
function toPromptArguments(request: ChatPromptRequest): PromptArguments {
return request.images === null ? [request.message] : [request.message, request.images];
}

View file

@ -0,0 +1,23 @@
import { type Context, defineRemoteService } from "@earendil-works/pi-agent-core";
import type { PromptImage } from "@earendil-works/pi-protocol";
export interface ChatPromptRequest {
message: string;
images: PromptImage[] | null;
}
export interface ChatPromptError {
code: string;
message: string;
}
export type ChatPromptResponse =
| { accepted: true; operationId: string; error: ChatPromptError | null }
| { accepted: false; operationId: string | null; error: ChatPromptError };
export interface ChatService {
prompt(request: ChatPromptRequest, context: Context): Promise<ChatPromptResponse>;
requestAbort(operationId: string, context: Context): Promise<void>;
}
export const Chat = defineRemoteService<ChatService>("chat");

View file

@ -16,6 +16,8 @@ import {
import Type, { type Static } from "typebox";
import { Check } from "typebox/value";
import type { ModelRuntime } from "../../core/model-runtime.ts";
import { Chat } from "./chat.ts";
import { provideChatService } from "./chat-provider.ts";
import { Models } from "./models.ts";
import { provideModelsService } from "./models-provider.ts";
@ -55,8 +57,9 @@ export async function createSessionWorkerServices(options: {
readonly configureServices: ((provider: RemoteServiceProvider) => void | Promise<void>) | undefined;
publish(scope: WorkerServiceScope, subscriptionId: string, update: ProtocolServiceProviderUpdate): Promise<void>;
}): Promise<SessionWorkerServices> {
const provider = new RemoteServiceProvider([Models, ...options.serviceTokens]);
const provider = new RemoteServiceProvider([Chat, Models, ...options.serviceTokens]);
try {
provideChatService(provider, options.harness);
await provideModelsService(provider, options.harness, options.modelRuntime);
await options.configureServices?.(provider);
} catch (error) {

View file

@ -0,0 +1,104 @@
import {
type AgentHarness,
BACKGROUND_CONTEXT,
Closed,
type RunResult as HarnessRunResult,
InvalidMessage,
LaneBusy,
RemoteServiceProvider,
UnknownSkill,
UnknownTemplate,
} from "@earendil-works/pi-agent-core";
import { describe, expect, test, vi } from "vitest";
import { Chat } from "../src/experimental/services/chat.ts";
import { provideChatService, toChatPromptResponse } from "../src/experimental/services/chat-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][];
describe("Chat service", () => {
test("provides prompt and durable abort methods", async () => {
const prompt = vi.fn(async () => ({
ok: true as const,
value: {
operation: "run" as const,
kind: "completed" as const,
runId: "operation-1",
leafId: "entry-1",
},
}));
const requestAbort = vi.fn(async () => ({
ok: true as const,
value: {
operationId: "operation-1",
newlyRequested: true,
steer: [],
followUp: [],
},
}));
const provider = new RemoteServiceProvider([Chat]);
provideChatService(provider, { prompt, requestAbort } as unknown as AgentHarness);
try {
await expect(
provider.invoke(
{ serviceId: "chat", member: "prompt", args: [{ message: "hello", images: null }] },
BACKGROUND_CONTEXT,
),
).resolves.toEqual({ accepted: true, operationId: "operation-1", error: null });
await expect(
provider.invoke({ serviceId: "chat", member: "requestAbort", args: ["operation-1"] }, BACKGROUND_CONTEXT),
).resolves.toBeUndefined();
expect(prompt).toHaveBeenCalledWith("hello", undefined, BACKGROUND_CONTEXT);
expect(requestAbort).toHaveBeenCalledWith("operation-1", BACKGROUND_CONTEXT);
} finally {
provider.dispose();
}
});
test("reports accepted successful and failed operations", () => {
expect(
toChatPromptResponse({
ok: true,
value: { operation: "run", kind: "completed", runId: "operation-1", leafId: "entry-1" },
}),
).toEqual({ accepted: true, operationId: "operation-1", error: null });
expect(
toChatPromptResponse({
ok: true,
value: {
operation: "run",
kind: "failed",
runId: "operation-2",
leafId: "entry-1",
error: { code: "provider", message: "failed", details: { status: 500 } },
},
}),
).toEqual({
accepted: true,
operationId: "operation-2",
error: { code: "provider", message: "failed" },
});
});
test.each(admissionErrors)("maps admission error %# to a stable response", (error, code, operationId) => {
expect(toChatPromptResponse({ ok: false, error })).toEqual({
accepted: false,
operationId,
error: { code, message: error.message },
});
});
});

View file

@ -444,12 +444,17 @@ describe("experimental durable server composition", () => {
await expect(services.dispose(BACKGROUND_CONTEXT)).resolves.toBeUndefined();
});
test("routes prompting through the worker-owned Harness and sanitizes runtime failures", async () => {
test("routes Chat through the worker-owned service provider and sanitizes runtime failures", async ({
onTestFinished,
}) => {
const legacyPrompt = vi.spyOn(Client.prototype, "promptSession");
onTestFinished(() => legacyPrompt.mockRestore());
const { directory } = await makeServer();
await expect(
runClient({ command: "client", sessionId: "demo-1", prompt: "question" }, { directory }),
).rejects.toThrow("Internal server error");
expect(legacyPrompt).not.toHaveBeenCalled();
});
// Runtime2 cannot complete no-tool runs yet; keep the framed prompt path covered as a sanitized failure above.

View file

@ -14,7 +14,7 @@ The transport carries `{ serviceId, instance?, member, args }` calls. A server t
`PromptArguments` contains one serializable `AgentLane.prompt()` overload. `PromptMessage` is the protocol's closed set of built-in message DTOs; application-defined `AgentMessage` extensions are not accepted implicitly. `RunResult` is the wire-safe structural equivalent of the Harness result and contains no JavaScript `Error` instances.
The temporary `ServiceRpc` manifest still backs client compatibility methods and transcript operations while those features migrate to plugin services. Experimental presentations consume server-owned `SessionDirectory` and `SessionManagement` services; Session plugin calls route opaquely to the attached worker, where the provider validates and invokes them.
The temporary `ServiceRpc` manifest still backs low-level client compatibility methods and transcript operations while those features migrate to plugin services. Experimental presentations consume server-owned `SessionDirectory` and `SessionManagement` services plus the worker-owned `Chat` service; Session plugin calls route opaquely to the attached worker, where the provider validates and invokes them.
A lane watch uses three RPCs so snapshot ordering does not depend on response/event scheduling: `watch()` creates a buffering watch and returns its authoritative snapshot, `startWatch()` flushes buffered events and begins live delivery, and `stopWatch()` releases it. Events carry their watch ID. Streaming message updates carry compact assistant-message frames rather than cumulative partial messages. Reconnection creates a new watch and snapshot rather than replaying old events.