| 1 | import * as Schema from "effect/Schema"; |
| 2 | import { Response, Toolkit } from "effect/ai"; |
| 3 | import * as Rpc from "effect/rpc/Rpc"; |
| 4 | import * as RpcGroup from "effect/rpc/RpcGroup"; |
| 5 | import * as RpcSchema from "effect/rpc/RpcSchema"; |
| 6 | |
| 7 | // `effect/ai` ships a `Schema` for every streaming response part. |
| 8 | // `Response.StreamPart(toolkit)` builds the union codec; with the empty |
| 9 | // toolkit it's exactly what `Chat.streamText` emits for a tool-less |
| 10 | // chat (text-start / text-delta / text-end / reasoning / finish / |
| 11 | // error). Reusing it as the RPC stream element means both ends decode |
| 12 | // to real `effect/ai` part instances instead of hand-rolled structs. |
| 13 | export const ChatStreamPart = Response.StreamPart(Toolkit.empty); |
| 14 | |
| 15 | export const SendResult = Schema.Struct({ |
| 16 | text: Schema.String, |
| 17 | turns: Schema.Number, |
| 18 | }); |
| 19 | |
| 20 | // Served on each Durable Object instance. The DO instance *is* the |
| 21 | // conversation thread, so payloads carry no id — keying happens at |
| 22 | // `getByName(threadId)`. |
| 23 | export class ChatBackendRpcs extends RpcGroup.make( |
| 24 | Rpc.make("send", { |
| 25 | payload: { prompt: Schema.String }, |
| 26 | success: SendResult, |
| 27 | }), |
| 28 | Rpc.make("streamMessage", { |
| 29 | payload: { prompt: Schema.String }, |
| 30 | success: RpcSchema.Stream(ChatStreamPart, Schema.Never), |
| 31 | }), |
| 32 | ) {} |
| 33 | |
| 34 | // Served on the public `RpcWorker`. It fans out to a DO instance by |
| 35 | // thread id, so every payload carries the id. |
| 36 | export class ChatRpcs extends RpcGroup.make( |
| 37 | Rpc.make("send", { |
| 38 | payload: { id: Schema.String, prompt: Schema.String }, |
| 39 | success: SendResult, |
| 40 | }), |
| 41 | Rpc.make("streamMessage", { |
| 42 | payload: { id: Schema.String, prompt: Schema.String }, |
| 43 | success: RpcSchema.Stream(ChatStreamPart, Schema.Never), |
| 44 | }), |
| 45 | ) {} |