History

gateway

Rhys Sullivan committed 9 months ago

Commit f358be31 parent e00332d17 files · +603 −294Browse files

Unified · Split

17 files
  1. .context/openclaw +0 −1
  2. .context/opensrc-mcp +0 −1
  3. apps/discord-bot/package.json +1 −0
  4. apps/discord-bot/src/agent-session-view.tsx +239 −0
  5. apps/discord-bot/src/approval-bridge.tsx +0 −100
  6. apps/discord-bot/src/approval-registry.test.ts +0 −38
  7. apps/discord-bot/src/approval-registry.ts +0 −67
  8. apps/discord-bot/src/gateway-client.ts +29 −2
  9. apps/discord-bot/src/index.tsx +17 −37
  10. apps/gateway/package.json +2 −1
  11. apps/gateway/src/index.ts +44 −20
  12. apps/gateway/src/rpc.ts +61 −0
  13. apps/gateway/src/turn-manager.ts +209 −0
  14. bun.lock +1 −1
  15. packages/core/package.json +0 −1
  16. packages/core/src/agent-rpc.ts +0 −24
  17. packages/core/src/index.ts +0 −1
.context/openclaw+0−1
@@ -1 +0,0 @@
1Subproject commit 6f4665dda3af787c50685e440cca21a57816b5a6
.context/opensrc-mcp+0−1
@@ -1 +0,0 @@
1Subproject commit 43c45b2cbe0e64fb7859164a28e29da427c37930
apps/discord-bot/package.json+1−0
@@ -16,6 +16,7 @@ · 15 unmodified lines
16 "@effect/platform": "latest",16 "@effect/platform": "latest",
17 "@effect/rpc": "latest",17 "@effect/rpc": "latest",
18 "@openassistant/core": "workspace:*",18 "@openassistant/core": "workspace:*",
19 "@openassistant/gateway": "workspace:*",
19 "@openassistant/reacord": "workspace:*",20 "@openassistant/reacord": "workspace:*",
20 "ai": "^6.0.73",21 "ai": "^6.0.73",
21 "discord.js": "^14.17.3",22 "discord.js": "^14.17.3",
apps/discord-bot/src/agent-session-view.tsx+239−0
@@ -0,0 +1,239 @@
1import type { TurnResult } from "@openassistant/gateway/rpc";
2import type { ButtonInteraction } from "discord.js";
3import { useCallback, useEffect, useRef, useState } from "react";
4import { ApprovalRequestView, AssistantReplyView, AssistantWorkingView } from "./discord-views.js";
5import { formatDiscordResponse } from "./format-response.js";
6import { continueGatewayTurn, resolveGatewayApproval, runGatewayTurn } from "./gateway-client.js";
7
8type ApprovalState = NonNullable<TurnResult["approval"]>;
9
10type SessionState =
11 | { phase: "starting" }
12 | {
13 phase: "awaiting";
14 turnId: string;
15 approval: ApprovalState;
16 resolving: boolean;
17 resolved?: { decision: "approved" | "denied"; actorId?: string };
18 }
19 | { phase: "completed"; message: string; footer?: string }
20 | { phase: "failed"; message: string };
21
22export function AgentSessionView(params: {
23 prompt: string;
24 requesterId: string;
25 channelId: string;
26 approvalTimeoutMs: number;
27}) {
28 const { prompt, requesterId, channelId, approvalTimeoutMs } = params;
29 const [state, setState] = useState<SessionState>({ phase: "starting" });
30 const isMounted = useRef(true);
31
32 useEffect(() => {
33 return () => {
34 isMounted.current = false;
35 };
36 }, []);
37
38 const applyTurnResult = useCallback((turn: TurnResult) => {
39 if (!isMounted.current) {
40 return;
41 }
42 if (turn.status === "awaiting_approval") {
43 if (!turn.approval) {
44 setState({
45 phase: "failed",
46 message: "Approval was required but no approval payload was returned.",
47 });
48 return;
49 }
50 setState({
51 phase: "awaiting",
52 turnId: turn.turnId,
53 approval: turn.approval,
54 resolving: false,
55 });
56 return;
57 }
58
59 if (turn.status === "failed") {
60 setState({
61 phase: "failed",
62 message: turn.error ? `I couldn't complete that request: ${turn.error}` : "I couldn't complete that request.",
63 });
64 return;
65 }
66
67 const response = formatDiscordResponse({
68 text: turn.message ?? "",
69 footer: turn.footer,
70 });
71 setState({
72 phase: "completed",
73 message: response.message,
74 ...(response.footer ? { footer: response.footer } : {}),
75 });
76 }, []);
77
78 const continueAfterDecision = useCallback(
79 async (params: {
80 turnId: string;
81 callId: string;
82 decision: "approved" | "denied";
83 actorId: string;
84 }) => {
85 const { turnId, callId, decision, actorId } = params;
86 setState((current) => {
87 if (
88 current.phase !== "awaiting" ||
89 current.turnId !== turnId ||
90 current.approval.callId !== callId
91 ) {
92 return current;
93 }
94 return {
95 ...current,
96 resolving: true,
97 resolved: { decision, actorId },
98 };
99 });
100
101 try {
102 const resolved = await resolveGatewayApproval({
103 turnId,
104 callId,
105 actorId,
106 decision,
107 });
108 if (resolved.status !== "resolved") {
109 if (!isMounted.current) {
110 return;
111 }
112 setState({
113 phase: "failed",
114 message: `Approval could not be applied: ${resolved.status}.`,
115 });
116 return;
117 }
118 const next = await continueGatewayTurn(turnId);
119 applyTurnResult(next);
120 } catch (error) {
121 if (!isMounted.current) {
122 return;
123 }
124 setState({
125 phase: "failed",
126 message: `I hit an unexpected error while continuing this request: ${describeUnknown(error)}`,
127 });
128 }
129 },
130 [applyTurnResult],
131 );
132
133 useEffect(() => {
134 void (async () => {
135 try {
136 const turn = await runGatewayTurn({
137 prompt,
138 requesterId,
139 channelId,
140 });
141 applyTurnResult(turn);
142 } catch (error) {
143 if (!isMounted.current) {
144 return;
145 }
146 setState({
147 phase: "failed",
148 message: `I hit an unexpected error while processing that request: ${describeUnknown(error)}`,
149 });
150 }
151 })();
152 }, [prompt, requesterId, channelId, applyTurnResult]);
153
154 useEffect(() => {
155 if (state.phase !== "awaiting" || state.resolving) {
156 return;
157 }
158 const timer = setTimeout(() => {
159 void continueAfterDecision({
160 turnId: state.turnId,
161 callId: state.approval.callId,
162 decision: "denied",
163 actorId: requesterId,
164 });
165 }, approvalTimeoutMs);
166
167 return () => clearTimeout(timer);
168 }, [approvalTimeoutMs, continueAfterDecision, requesterId, state]);
169
170 if (state.phase === "starting") {
171 return <AssistantWorkingView />;
172 }
173
174 if (state.phase === "failed") {
175 return <AssistantReplyView message={state.message} />;
176 }
177
178 if (state.phase === "completed") {
179 return <AssistantReplyView message={state.message} footer={state.footer} />;
180 }
181
182 return (
183 <ApprovalRequestView
184 toolPath={state.approval.toolPath}
185 callId={state.approval.callId}
186 inputPreview={state.approval.inputPreview}
187 requesterId={requesterId}
188 resolved={state.resolved}
189 onApprove={async (interaction) => {
190 await handleDecisionInteraction({
191 interaction,
192 requesterId,
193 onAllowed: () =>
194 continueAfterDecision({
195 turnId: state.turnId,
196 callId: state.approval.callId,
197 decision: "approved",
198 actorId: interaction.user.id,
199 }),
200 });
201 }}
202 onDeny={async (interaction) => {
203 await handleDecisionInteraction({
204 interaction,
205 requesterId,
206 onAllowed: () =>
207 continueAfterDecision({
208 turnId: state.turnId,
209 callId: state.approval.callId,
210 decision: "denied",
211 actorId: interaction.user.id,
212 }),
213 });
214 }}
215 />
216 );
217}
218
219async function handleDecisionInteraction(params: {
220 interaction: ButtonInteraction;
221 requesterId: string;
222 onAllowed: () => Promise<void>;
223}): Promise<void> {
224 if (params.interaction.user.id !== params.requesterId) {
225 await params.interaction.reply({
226 content: "Only the requesting user can resolve this approval.",
227 ephemeral: true,
228 });
229 return;
230 }
231 await params.onAllowed();
232}
233
234function describeUnknown(error: unknown): string {
235 if (error instanceof Error) {
236 return error.message;
237 }
238 return String(error);
239}
apps/discord-bot/src/approval-bridge.tsx+0−100
@@ -1,100 +0,0 @@
1import type { ApprovalDecision, ApprovalRequest } from "@openassistant/core";
2import { type ReacordInstance, makeReacord } from "@openassistant/reacord";
3import type { ButtonInteraction, Client, SendableChannels } from "discord.js";
4import { Effect } from "effect";
5import { ApprovalRegistry } from "./approval-registry.js";
6import { ApprovalRequestView } from "./discord-views.js";
7
8type ReacordApi = ReturnType<typeof makeReacord>;
9
10export class DiscordApprovalBridge {
11 private readonly registry: ApprovalRegistry;
12 private readonly reacord: ReacordApi;
13
14 constructor(client: Client, timeoutMs: number = 5 * 60_000) {
15 this.registry = new ApprovalRegistry(timeoutMs);
16 this.reacord = makeReacord(client);
17 }
18
19 async requestApproval(params: {
20 request: ApprovalRequest;
21 channel: SendableChannels;
22 requesterId: string;
23 }): Promise<ApprovalDecision> {
24 const { request, channel, requesterId } = params;
25 const pending = this.registry.open(request.callId, requesterId);
26 let instance: ReacordInstance | null = null;
27 let resolvedActorId: string | undefined;
28
29 try {
30 const handleDecision = async (interaction: ButtonInteraction, decision: ApprovalDecision): Promise<void> => {
31 const status = this.registry.resolve(request.callId, interaction.user.id, decision);
32 if (status === "unauthorized") {
33 await interaction.reply({
34 content: "Only the requesting user can resolve this approval.",
35 ephemeral: true,
36 });
37 return;
38 }
39
40 if (status === "not_found") {
41 await interaction.reply({
42 content: "This approval is no longer pending.",
43 ephemeral: true,
44 });
45 return;
46 }
47
48 resolvedActorId = interaction.user.id;
49 instance?.render(
50 <ApprovalRequestView
51 toolPath={request.toolPath}
52 callId={request.callId}
53 inputPreview={request.inputPreview}
54 requesterId={requesterId}
55 resolved={{ decision, actorId: interaction.user.id }}
56 onApprove={handleApprove}
57 onDeny={handleDeny}
58 />,
59 );
60 };
61 const handleApprove = (interaction: ButtonInteraction) => handleDecision(interaction, "approved");
62 const handleDeny = (interaction: ButtonInteraction) => handleDecision(interaction, "denied");
63
64 instance = await Effect.runPromise(
65 this.reacord.send(
66 channel,
67 <ApprovalRequestView
68 toolPath={request.toolPath}
69 callId={request.callId}
70 inputPreview={request.inputPreview}
71 requesterId={requesterId}
72 onApprove={handleApprove}
73 onDeny={handleDeny}
74 />,
75 ),
76 );
77 } catch {
78 this.registry.cancel(request.callId, "denied");
79 }
80
81 const decision = await pending;
82
83 if (instance) {
84 instance.render(
85 <ApprovalRequestView
86 toolPath={request.toolPath}
87 callId={request.callId}
88 inputPreview={request.inputPreview}
89 requesterId={requesterId}
90 resolved={{ decision, actorId: resolvedActorId }}
91 onApprove={() => {}}
92 onDeny={() => {}}
93 />,
94 );
95 instance.deactivate();
96 }
97
98 return decision;
99 }
100}
apps/discord-bot/src/approval-registry.test.ts+0−38
@@ -1,38 +0,0 @@
1import { describe, expect, it } from "vitest";
2import { ApprovalRegistry } from "./approval-registry.js";
3
4describe("ApprovalRegistry", () => {
5 it("resolves for requester", async () => {
6 const registry = new ApprovalRegistry(1_000);
7 const pending = registry.open("call_1", "user_a");
8
9 const result = registry.resolve("call_1", "user_a", "approved");
10 const decision = await pending;
11
12 expect(result).toBe("resolved");
13 expect(decision).toBe("approved");
14 expect(registry.size()).toBe(0);
15 });
16
17 it("rejects unauthorized actor", async () => {
18 const registry = new ApprovalRegistry(1_000);
19 const pending = registry.open("call_2", "user_a");
20
21 const unauthorized = registry.resolve("call_2", "user_b", "approved");
22 expect(unauthorized).toBe("unauthorized");
23 expect(registry.size()).toBe(1);
24
25 registry.cancel("call_2");
26 const decision = await pending;
27 expect(decision).toBe("denied");
28 });
29
30 it("times out to denied", async () => {
31 const registry = new ApprovalRegistry(10);
32 const pending = registry.open("call_3", "user_a");
33
34 const decision = await pending;
35 expect(decision).toBe("denied");
36 expect(registry.size()).toBe(0);
37 });
38});
apps/discord-bot/src/approval-registry.ts+0−67
@@ -1,67 +0,0 @@
1import type { ApprovalDecision } from "@openassistant/core";
2
3type PendingApproval = {
4 requesterId: string;
5 resolve: (decision: ApprovalDecision) => void;
6 timeout: ReturnType<typeof setTimeout>;
7};
8
9export type ResolveApprovalResult = "resolved" | "not_found" | "unauthorized";
10
11export class ApprovalRegistry {
12 private readonly pending = new Map<string, PendingApproval>();
13
14 constructor(private readonly timeoutMs: number = 5 * 60_000) {}
15
16 open(callId: string, requesterId: string): Promise<ApprovalDecision> {
17 if (this.pending.has(callId)) {
18 throw new Error(`Approval already pending: ${callId}`);
19 }
20
21 return new Promise<ApprovalDecision>((resolve) => {
22 const timeout = setTimeout(() => {
23 this.finish(callId, "denied");
24 }, this.timeoutMs);
25
26 this.pending.set(callId, {
27 requesterId,
28 resolve,
29 timeout,
30 });
31 });
32 }
33
34 resolve(callId: string, actorId: string, decision: ApprovalDecision): ResolveApprovalResult {
35 const entry = this.pending.get(callId);
36 if (!entry) {
37 return "not_found";
38 }
39 if (entry.requesterId !== actorId) {
40 return "unauthorized";
41 }
42
43 this.finish(callId, decision);
44 return "resolved";
45 }
46
47 cancel(callId: string, decision: ApprovalDecision = "denied"): void {
48 if (!this.pending.has(callId)) {
49 return;
50 }
51 this.finish(callId, decision);
52 }
53
54 size(): number {
55 return this.pending.size;
56 }
57
58 private finish(callId: string, decision: ApprovalDecision): void {
59 const entry = this.pending.get(callId);
60 if (!entry) {
61 return;
62 }
63 this.pending.delete(callId);
64 clearTimeout(entry.timeout);
65 entry.resolve(decision);
66 }
67}
apps/discord-bot/src/gateway-client.ts+29−2
@@ -1,6 +1,10 @@
1import { FetchHttpClient } from "@effect/platform";1import { FetchHttpClient } from "@effect/platform";
2import { RpcClient, RpcSerialization } from "@effect/rpc";2import { RpcClient, RpcSerialization } from "@effect/rpc";
3import { AgentRpcs, type RunTurnOutput } from "@openassistant/core";3import {
4 AgentRpcs,
5 type ResolveApprovalOutput,
6 type TurnResult,
7} from "@openassistant/gateway/rpc";
4import { Effect, Layer } from "effect";8import { Effect, Layer } from "effect";
5 9
6const gatewayUrl = readEnv("OPENASSISTANT_GATEWAY_URL") ?? "http://127.0.0.1:8787/rpc";10const gatewayUrl = readEnv("OPENASSISTANT_GATEWAY_URL") ?? "http://127.0.0.1:8787/rpc";
@@ -15,7 +19,7 @@ export interface RunGatewayTurnInput { · 8 unmodified lines
15 channelId: string;19 channelId: string;
16}20}
17 21
18export async function runGatewayTurn(input: RunGatewayTurnInput): Promise<RunTurnOutput> {22export async function runGatewayTurn(input: RunGatewayTurnInput): Promise<TurnResult> {
19 return Effect.runPromise(23 return Effect.runPromise(
20 Effect.gen(function* () {24 Effect.gen(function* () {
21 const client = yield* RpcClient.make(AgentRpcs);25 const client = yield* RpcClient.make(AgentRpcs);
@@ -29,6 +33,29 @@ export async function runGatewayTurn(input: RunGatewayTurnInput): Promise<RunTur · 7 unmodified lines
29 );33 );
30}34}
31 35
36export async function continueGatewayTurn(turnId: string): Promise<TurnResult> {
37 return Effect.runPromise(
38 Effect.gen(function* () {
39 const client = yield* RpcClient.make(AgentRpcs);
40 return yield* client.ContinueTurn({ turnId });
41 }).pipe(Effect.scoped, Effect.provide(GatewayRpcClientLive)),
42 );
43}
44
45export async function resolveGatewayApproval(input: {
46 turnId: string;
47 callId: string;
48 actorId: string;
49 decision: "approved" | "denied";
50}): Promise<ResolveApprovalOutput> {
51 return Effect.runPromise(
52 Effect.gen(function* () {
53 const client = yield* RpcClient.make(AgentRpcs);
54 return yield* client.ResolveApproval(input);
55 }).pipe(Effect.scoped, Effect.provide(GatewayRpcClientLive)),
56 );
57}
58
32function readEnv(key: string): string | undefined {59function readEnv(key: string): string | undefined {
33 const bun = (globalThis as { Bun?: { env?: Record<string, string | undefined> } }).Bun;60 const bun = (globalThis as { Bun?: { env?: Record<string, string | undefined> } }).Bun;
34 return bun?.env?.[key] ?? process.env[key];61 return bun?.env?.[key] ?? process.env[key];
apps/discord-bot/src/index.tsx+17−37
@@ -9,14 +9,13 @@ import { · 8 unmodified lines
9 type SendableChannels,9 type SendableChannels,
10} from "discord.js";10} from "discord.js";
11import { Effect } from "effect";11import { Effect } from "effect";
12import { AssistantReplyView, AssistantWorkingView } from "./discord-views.js";12import { AgentSessionView } from "./agent-session-view.js";
13import { formatDiscordResponse } from "./format-response.js";
14import { runGatewayTurn } from "./gateway-client.js";
15 13
16const token = Bun.env.DISCORD_BOT_TOKEN;14const token = Bun.env.DISCORD_BOT_TOKEN;
17if (!token) {15if (!token) {
18 throw new Error("Missing DISCORD_BOT_TOKEN");16 throw new Error("Missing DISCORD_BOT_TOKEN");
19}17}
18const approvalTimeoutMs = Number(Bun.env.OPENASSISTANT_APPROVAL_TIMEOUT_MS ?? 300_000);
20 19
21const client = new Client({20const client = new Client({
22 intents: [21 intents: [
@@ -48,38 +47,26 @@ function shouldIgnore(message: Message): boolean { · 25 unmodified lines
48}47}
49 48
50async function handleMessage(message: Message): Promise<void> {49async function handleMessage(message: Message): Promise<void> {
51 const approvalChannel = asSendableChannel(message.channel);50 const replyChannel = asSendableChannel(message.channel);
52 if (!approvalChannel) {51 if (!replyChannel) {
53 await message.reply("This channel does not support assistant responses.");52 await message.reply("This channel does not support assistant responses.");
54 return;53 return;
55 }54 }
56 55
57 const instance = await Effect.runPromise(56 await Effect.runPromise(
58 reacord.send(approvalChannel, <AssistantWorkingView />, {57 reacord.send(
59 reply: { messageReference: message.id },58 replyChannel,
60 }),59 <AgentSessionView
61 );60 prompt={message.content}
62 61 requesterId={message.author.id}
63 try {62 channelId={message.channelId}
64 const generated = await runGatewayTurn({63 approvalTimeoutMs={approvalTimeoutMs}
65 prompt: message.content,
66 requesterId: message.author.id,
67 channelId: message.channelId,
68 });
69
70 const response = formatDiscordResponse({
71 text: generated.message,
72 footer: generated.footer,
73 });
74
75 instance.render(<AssistantReplyView message={response.message} footer={response.footer} />);
76 } catch (error) {
77 instance.render(
78 <AssistantReplyView
79 message={`I hit an unexpected error while processing that request: ${describeUnknown(error)}`}
80 />,64 />,
81 );65 {
82 }66 reply: { messageReference: message.id },
67 },
68 ),
69 );
83}70}
84 71
85function asSendableChannel(channel: unknown): SendableChannels | null {72function asSendableChannel(channel: unknown): SendableChannels | null {
@@ -91,10 +78,3 @@ function asSendableChannel(channel: unknown): SendableChannels | null { · 5 unmodified lines
91 }78 }
92 return channel as SendableChannels;79 return channel as SendableChannels;
93}80}
94
95function describeUnknown(error: unknown): string {
96 if (error instanceof Error) {
97 return error.message;
98 }
99 return String(error);
100}
apps/gateway/package.json+2−1
@@ -3,7 +3,8 @@ · 2 unmodified lines
3 "private": true,3 "private": true,
4 "type": "module",4 "type": "module",
5 "exports": {5 "exports": {
6 ".": "./src/index.ts"6 ".": "./src/index.ts",
7 "./rpc": "./src/rpc.ts"
7 },8 },
8 "scripts": {9 "scripts": {
9 "with-env": "bun run --env-file=../../.env --",10 "with-env": "bun run --env-file=../../.env --",
apps/gateway/src/index.ts+44−20
@@ -1,43 +1,67 @@
1import { HttpRouter } from "@effect/platform";1import { HttpRouter } from "@effect/platform";
2import { BunHttpServer, BunRuntime } from "@effect/platform-bun";2import { BunHttpServer, BunRuntime } from "@effect/platform-bun";
3import { RpcSerialization, RpcServer } from "@effect/rpc";3import { RpcSerialization, RpcServer } from "@effect/rpc";
4import { AgentRpcs, RunTurnOutput, createCodeModeRunner } from "@openassistant/core";
5import { Effect, Layer } from "effect";4import { Effect, Layer } from "effect";
6import { runAgentLoop } from "./agent-loop.js";
7import { InMemoryCalendarStore } from "./calendar-store.js";5import { InMemoryCalendarStore } from "./calendar-store.js";
6import { AgentRpcs, ResolveApprovalOutput, type TurnResult } from "./rpc.js";
8import { createToolTree } from "./tools.js";7import { createToolTree } from "./tools.js";
8import { TurnManager } from "./turn-manager.js";
9 9
10const port = Number(readEnv("OPENASSISTANT_GATEWAY_PORT") ?? "8787");10const port = Number(readEnv("OPENASSISTANT_GATEWAY_PORT") ?? "8787");
11 11
12const calendarStore = new InMemoryCalendarStore();12const calendarStore = new InMemoryCalendarStore();
13const tools = createToolTree(calendarStore);13const tools = createToolTree(calendarStore);
14const runner = createCodeModeRunner({14const turnManager = new TurnManager(tools, isVerboseMode());
15 tools,
16 requestApproval: () => Effect.succeed("approved"),
17});
18 15
19const AgentHandlersLive = AgentRpcs.toLayer(16const AgentHandlersLive = AgentRpcs.toLayer(
20 Effect.succeed({17 Effect.succeed({
21 RunTurn: (input: { prompt: string; nowIso: string }) =>18 RunTurn: (input: { prompt: string; requesterId: string; channelId: string; nowIso: string }) =>
22 Effect.tryPromise({19 Effect.tryPromise({
23 try: async () => {20 try: async () => {
24 const generated = await runAgentLoop(21 const turnId = turnManager.start({
25 input.prompt,22 prompt: input.prompt,
26 (code) => Effect.runPromise(runner.run({ code })),23 requesterId: input.requesterId,
27 {24 channelId: input.channelId,
28 now: new Date(input.nowIso),25 now: new Date(input.nowIso),
29 },
30 );
31
32 return new RunTurnOutput({
33 message: generated.text,
34 planner: generated.planner,
35 codeRuns: generated.runs.length,
36 ...(isVerboseMode() ? { footer: generated.planner } : {}),
37 });26 });
27 const event = await turnManager.waitForNext(turnId);
28 if (!event) {
29 return {
30 status: "failed",
31 turnId,
32 error: "Turn not found.",
33 } as TurnResult;
34 }
35 return event;
38 },36 },
39 catch: (error) => `RunTurn failed: ${describeUnknown(error)}`,37 catch: (error) => `RunTurn failed: ${describeUnknown(error)}`,
40 }),38 }),
39 ContinueTurn: (input: { turnId: string }) =>
40 Effect.tryPromise({
41 try: async () => {
42 const event = await turnManager.waitForNext(input.turnId);
43 if (!event) {
44 return {
45 status: "failed",
46 turnId: input.turnId,
47 error: "Turn not found.",
48 } as TurnResult;
49 }
50 return event;
51 },
52 catch: (error) => `ContinueTurn failed: ${describeUnknown(error)}`,
53 }),
54 ResolveApproval: (input: { turnId: string; callId: string; actorId: string; decision: "approved" | "denied" }) =>
55 Effect.succeed(
56 new ResolveApprovalOutput({
57 status: turnManager.resolveApproval({
58 turnId: input.turnId,
59 callId: input.callId,
60 actorId: input.actorId,
61 decision: input.decision,
62 }),
63 }),
64 ),
41 }),65 }),
42);66);
43 67
apps/gateway/src/rpc.ts+61−0
@@ -0,0 +1,61 @@
1import { Rpc, RpcGroup } from "@effect/rpc";
2import { Schema } from "effect";
3
4export class RunTurnInput extends Schema.Class<RunTurnInput>("RunTurnInput")({
5 prompt: Schema.String,
6 requesterId: Schema.String,
7 channelId: Schema.String,
8 nowIso: Schema.String,
9}) {}
10
11export class ContinueTurnInput extends Schema.Class<ContinueTurnInput>("ContinueTurnInput")({
12 turnId: Schema.String,
13}) {}
14
15export class ResolveApprovalInput extends Schema.Class<ResolveApprovalInput>("ResolveApprovalInput")({
16 turnId: Schema.String,
17 callId: Schema.String,
18 actorId: Schema.String,
19 decision: Schema.Literal("approved", "denied"),
20}) {}
21
22export class ResolveApprovalOutput extends Schema.Class<ResolveApprovalOutput>("ResolveApprovalOutput")({
23 status: Schema.Literal("resolved", "not_found", "unauthorized"),
24}) {}
25
26export class ApprovalPrompt extends Schema.Class<ApprovalPrompt>("ApprovalPrompt")({
27 callId: Schema.String,
28 toolPath: Schema.String,
29 inputPreview: Schema.optional(Schema.String),
30}) {}
31
32export const TurnResultSchema = Schema.Struct({
33 status: Schema.Literal("awaiting_approval", "completed", "failed"),
34 turnId: Schema.String,
35 approval: Schema.optional(ApprovalPrompt),
36 message: Schema.optional(Schema.String),
37 planner: Schema.optional(Schema.String),
38 codeRuns: Schema.optional(Schema.Number),
39 footer: Schema.optional(Schema.String),
40 error: Schema.optional(Schema.String),
41});
42
43export type TurnResult = Schema.Schema.Type<typeof TurnResultSchema>;
44
45export class AgentRpcs extends RpcGroup.make(
46 Rpc.make("RunTurn", {
47 payload: RunTurnInput,
48 success: TurnResultSchema,
49 error: Schema.String,
50 }),
51 Rpc.make("ContinueTurn", {
52 payload: ContinueTurnInput,
53 success: TurnResultSchema,
54 error: Schema.String,
55 }),
56 Rpc.make("ResolveApproval", {
57 payload: ResolveApprovalInput,
58 success: ResolveApprovalOutput,
59 error: Schema.String,
60 }),
61) {}
apps/gateway/src/turn-manager.ts+209−0
@@ -0,0 +1,209 @@
1import {
2 createCodeModeRunner,
3 type ApprovalDecision,
4 type ApprovalRequest,
5 type ToolTree,
6} from "@openassistant/core";
7import type { AgentLoopResult } from "./agent-loop.js";
8import { runAgentLoop } from "./agent-loop.js";
9import { type TurnResult } from "./rpc.js";
10import { Effect } from "effect";
11
12type PendingApproval = {
13 callId: string;
14 toolPath: string;
15 inputPreview?: string;
16 resolve: (decision: ApprovalDecision) => void;
17};
18
19type TurnEvent = TurnResult;
20
21type TurnSession = {
22 id: string;
23 requesterId: string;
24 channelId: string;
25 queue: TurnEvent[];
26 waitingResolver: ((event: TurnEvent) => void) | null;
27 pendingApproval: PendingApproval | null;
28 approvalWaiters: Array<() => void>;
29 completed: boolean;
30};
31
32export type ResolveApprovalStatus = "resolved" | "not_found" | "unauthorized";
33
34export class TurnManager {
35 private readonly sessions = new Map<string, TurnSession>();
36
37 constructor(
38 private readonly tools: ToolTree,
39 private readonly verboseFooter: boolean,
40 ) {}
41
42 start(params: {
43 prompt: string;
44 requesterId: string;
45 channelId: string;
46 now: Date;
47 }): string {
48 const id = newTurnId();
49 const session: TurnSession = {
50 id,
51 requesterId: params.requesterId,
52 channelId: params.channelId,
53 queue: [],
54 waitingResolver: null,
55 pendingApproval: null,
56 approvalWaiters: [],
57 completed: false,
58 };
59 this.sessions.set(id, session);
60 this.runSession(session, params.prompt, params.now);
61 return id;
62 }
63
64 async waitForNext(turnId: string): Promise<TurnEvent | null> {
65 const session = this.sessions.get(turnId);
66 if (!session) {
67 return null;
68 }
69 if (session.queue.length > 0) {
70 const next = session.queue.shift()!;
71 this.cleanupIfTerminal(session, next);
72 return next;
73 }
74
75 const event = await new Promise<TurnEvent>((resolve) => {
76 session.waitingResolver = resolve;
77 });
78 this.cleanupIfTerminal(session, event);
79 return event;
80 }
81
82 resolveApproval(params: {
83 turnId: string;
84 callId: string;
85 actorId: string;
86 decision: ApprovalDecision;
87 }): ResolveApprovalStatus {
88 const session = this.sessions.get(params.turnId);
89 if (!session) {
90 return "not_found";
91 }
92 if (session.requesterId !== params.actorId) {
93 return "unauthorized";
94 }
95 if (!session.pendingApproval || session.pendingApproval.callId !== params.callId) {
96 return "not_found";
97 }
98
99 const pending = session.pendingApproval;
100 session.pendingApproval = null;
101 pending.resolve(params.decision);
102 const nextWaiter = session.approvalWaiters.shift();
103 if (nextWaiter) {
104 nextWaiter();
105 }
106 return "resolved";
107 }
108
109 private async runSession(session: TurnSession, prompt: string, now: Date): Promise<void> {
110 try {
111 const runner = createCodeModeRunner({
112 tools: this.tools,
113 requestApproval: (request) =>
114 Effect.tryPromise({
115 try: () => this.requestApproval(session, request),
116 catch: (error) => error,
117 }),
118 });
119
120 const generated = await runAgentLoop(
121 prompt,
122 (code) => Effect.runPromise(runner.run({ code })),
123 { now },
124 );
125
126 this.emitEvent(session, toCompletedEvent(session.id, generated, this.verboseFooter));
127 } catch (error) {
128 this.emitEvent(session, {
129 status: "failed",
130 turnId: session.id,
131 error: describeUnknown(error),
132 });
133 } finally {
134 session.completed = true;
135 }
136 }
137
138 private async requestApproval(session: TurnSession, request: ApprovalRequest): Promise<ApprovalDecision> {
139 await this.waitForApprovalSlot(session);
140
141 return new Promise<ApprovalDecision>((resolve) => {
142 session.pendingApproval = {
143 callId: request.callId,
144 toolPath: request.toolPath,
145 ...(request.inputPreview ? { inputPreview: request.inputPreview } : {}),
146 resolve,
147 };
148
149 this.emitEvent(session, {
150 status: "awaiting_approval",
151 turnId: session.id,
152 approval: {
153 callId: request.callId,
154 toolPath: request.toolPath,
155 ...(request.inputPreview ? { inputPreview: request.inputPreview } : {}),
156 },
157 });
158 });
159 }
160
161 private async waitForApprovalSlot(session: TurnSession): Promise<void> {
162 while (session.pendingApproval) {
163 await new Promise<void>((resolve) => {
164 session.approvalWaiters.push(resolve);
165 });
166 }
167 }
168
169 private emitEvent(session: TurnSession, event: TurnEvent): void {
170 if (session.waitingResolver) {
171 const resolve = session.waitingResolver;
172 session.waitingResolver = null;
173 resolve(event);
174 return;
175 }
176 session.queue.push(event);
177 }
178
179 private cleanupIfTerminal(session: TurnSession, event: TurnEvent): void {
180 if ((event.status === "completed" || event.status === "failed") && session.completed) {
181 this.sessions.delete(session.id);
182 }
183 }
184}
185
186function toCompletedEvent(turnId: string, generated: AgentLoopResult, includeFooter: boolean): TurnResult {
187 return {
188 status: "completed",
189 turnId,
190 message: generated.text,
191 planner: generated.planner,
192 codeRuns: generated.runs.length,
193 ...(includeFooter ? { footer: generated.planner } : {}),
194 };
195}
196
197function newTurnId(): string {
198 if (typeof crypto !== "undefined" && typeof crypto.randomUUID === "function") {
199 return crypto.randomUUID();
200 }
201 return `turn_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`;
202}
203
204function describeUnknown(error: unknown): string {
205 if (error instanceof Error) {
206 return error.message;
207 }
208 return String(error);
209}
bun.lock+1−1
@@ -20,6 +20,7 @@ · 19 unmodified lines
20 "@effect/platform": "latest",20 "@effect/platform": "latest",
21 "@effect/rpc": "latest",21 "@effect/rpc": "latest",
22 "@openassistant/core": "workspace:*",22 "@openassistant/core": "workspace:*",
23 "@openassistant/gateway": "workspace:*",
23 "@openassistant/reacord": "workspace:*",24 "@openassistant/reacord": "workspace:*",
24 "ai": "^6.0.73",25 "ai": "^6.0.73",
25 "discord.js": "^14.17.3",26 "discord.js": "^14.17.3",
@@ -47,7 +48,6 @@ · 21 unmodified lines
47 "packages/core": {48 "packages/core": {
48 "name": "@openassistant/core",49 "name": "@openassistant/core",
49 "dependencies": {50 "dependencies": {
50 "@effect/rpc": "latest",
51 "effect": "latest",51 "effect": "latest",
52 },52 },
53 },53 },
packages/core/package.json+0−1
@@ -12,7 +12,6 @@ · 11 unmodified lines
12 "clean": "bun -e \"import { rmSync } from 'node:fs'; for (const p of ['dist','node_modules','.turbo','coverage','.tsbuildinfo']) rmSync(p,{recursive:true,force:true});\""12 "clean": "bun -e \"import { rmSync } from 'node:fs'; for (const p of ['dist','node_modules','.turbo','coverage','.tsbuildinfo']) rmSync(p,{recursive:true,force:true});\""
13 },13 },
14 "dependencies": {14 "dependencies": {
15 "@effect/rpc": "latest",
16 "effect": "latest"15 "effect": "latest"
17 }16 }
18}17}
packages/core/src/agent-rpc.ts+0−24
@@ -1,24 +0,0 @@
1import { Rpc, RpcGroup } from "@effect/rpc";
2import { Schema } from "effect";
3
4export class RunTurnInput extends Schema.Class<RunTurnInput>("RunTurnInput")({
5 prompt: Schema.String,
6 requesterId: Schema.String,
7 channelId: Schema.String,
8 nowIso: Schema.String,
9}) {}
10
11export class RunTurnOutput extends Schema.Class<RunTurnOutput>("RunTurnOutput")({
12 message: Schema.String,
13 planner: Schema.String,
14 codeRuns: Schema.Number,
15 footer: Schema.optional(Schema.String),
16}) {}
17
18export class AgentRpcs extends RpcGroup.make(
19 Rpc.make("RunTurn", {
20 payload: RunTurnInput,
21 success: RunTurnOutput,
22 error: Schema.String,
23 }),
24) {}
packages/core/src/index.ts+0−1
@@ -2,4 +2,3 @@ import { Effect } from "effect"; · 1 unmodified line
2 2
3export const healthcheck = Effect.succeed("ok");3export const healthcheck = Effect.succeed("ok");
4export * from "./codemode/runner.js";4export * from "./codemode/runner.js";
5export * from "./agent-rpc.js";