diff --git a/packages/agent/docs/plugins.md b/packages/agent/docs/plugins.md index 93e95e7bb..853b6d63a 100644 --- a/packages/agent/docs/plugins.md +++ b/packages/agent/docs/plugins.md @@ -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. diff --git a/packages/coding-agent/src/experimental/client.ts b/packages/coding-agent/src/experimental/client.ts index 25d68ad5d..d40bf9070 100644 --- a/packages/coding-agent/src/experimental/client.ts +++ b/packages/coding-agent/src/experimental/client.ts @@ -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; } @@ -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(); + // 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): string { + return message.content + .filter((content) => content.type === "text") + .map((content) => content.text) + .join(""); +} + function waitForState(state: RemoteState): Promise { if (state.value !== undefined) return Promise.resolve(state.value); return new Promise((resolve) => { diff --git a/packages/coding-agent/src/experimental/services/chat-provider.ts b/packages/coding-agent/src/experimental/services/chat-provider.ts new file mode 100644 index 000000000..9cc886d9d --- /dev/null +++ b/packages/coding-agent/src/experimental/services/chat-provider.ts @@ -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]; +} diff --git a/packages/coding-agent/src/experimental/services/chat.ts b/packages/coding-agent/src/experimental/services/chat.ts new file mode 100644 index 000000000..23254cf54 --- /dev/null +++ b/packages/coding-agent/src/experimental/services/chat.ts @@ -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; + requestAbort(operationId: string, context: Context): Promise; +} + +export const Chat = defineRemoteService("chat"); diff --git a/packages/coding-agent/src/experimental/services/worker.ts b/packages/coding-agent/src/experimental/services/worker.ts index 65b372d69..a503e9f94 100644 --- a/packages/coding-agent/src/experimental/services/worker.ts +++ b/packages/coding-agent/src/experimental/services/worker.ts @@ -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) | undefined; publish(scope: WorkerServiceScope, subscriptionId: string, update: ProtocolServiceProviderUpdate): Promise; }): Promise { - 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) { diff --git a/packages/coding-agent/test/experimental-chat-service.test.ts b/packages/coding-agent/test/experimental-chat-service.test.ts new file mode 100644 index 000000000..673a283fe --- /dev/null +++ b/packages/coding-agent/test/experimental-chat-service.test.ts @@ -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["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 }, + }); + }); +}); diff --git a/packages/coding-agent/test/experimental-remote-runtime.test.ts b/packages/coding-agent/test/experimental-remote-runtime.test.ts index 6422c26fe..136690a2d 100644 --- a/packages/coding-agent/test/experimental-remote-runtime.test.ts +++ b/packages/coding-agent/test/experimental-remote-runtime.test.ts @@ -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. diff --git a/packages/protocol/README.md b/packages/protocol/README.md index 1ae09df04..d03146b14 100644 --- a/packages/protocol/README.md +++ b/packages/protocol/README.md @@ -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.