Close the tool loop on the dev lane

ac1c7221713f · AtlantisPleb · · parent 4bb3f52256ea

Close the tool loop on the dev lane

The dev lane's source now holds the conversation as OpenResponses
items, declares the session's tools on every call, runs the
function_call items the model asks for, and replays call and output so
the next request answers from them — the thread lane's division of
labor, spoken in the OpenResponses grammar. A session on coder --dev
can now reach the capability catalog and everything behind it, instead
of explaining that it cannot access the filesystem.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E1mRkPGYmTVvMKqAzmQvy5
Co-Authored-By
Claude Fable 5 <noreply@anthropic.com>

Deploy story

What this commit did to the running system — joined from the forge receipt chain, the part a commit page elsewhere cannot show.

Not deployed through the forge lane

No push, promotion, build, or deploy receipt references this commit (receipts are scanned over a bounded recent window). Changes shipped by full node replacement carry their proof in the release gate receipt instead.

Changed files

  • modified packages/openagents-cli/src/coder-responses.ts
  • modified packages/openagents-cli/test/coder-responses.test.ts

Diff

2 files changed, +262 -21

packages/openagents-cli/src/coder-responses.ts modified +193 -20

@@ -1,31 +1,62 @@

1 1
/**
2 2
 * A reply source speaking the OpenResponses surface at `POST /api/v1/responses`.
3 3
 *
4
 * The dev lane's source: `openagents coder --dev` sends each prompt as an
4
 * The dev lane's source: `openagents coder --dev` sends each turn as an
5 5
 * OpenResponses request with `stream: true` and renders the semantic events
6
 * that come back — today, from the server's acknowledgement stub, which
7
 * answers every prompt "Acknowledged." in two deltas. No model stands behind
8
 * it yet; the point is that the client-side turn loop is built against the
9
 * OpenResponses event grammar before a provider is, so swapping the stub for
10
 * a real loop changes the server and nothing here.
6
 * that come back. The conversation is client-held — every call carries the
7
 * item history — and the tool runtime is the client's: a `function_call`
8
 * item the model asks for is run here, its output replayed as a
9
 * `function_call_output` item, and the loop continues until the model
10
 * answers in text. That is the same division of labor as the thread lane,
11
 * spoken in the OpenResponses grammar instead of chat-completions.
11 12
 *
12
 * Only `response.output_text.delta` is rendered. The rest of the sequence —
13
 * created, item and part boundaries, completed — is parsed and passed over,
14
 * which is exactly what the grammar is for: a client reads the events it
15
 * understands and survives the ones it does not.
13
 * Rendered events are the ones this client understands —
14
 * `response.output_text.delta`, `response.reasoning_summary_text.delta`,
15
 * `response.output_item.done` for function calls, `response.completed`,
16
 * `response.failed` — and the rest of the sequence passes over it, which is
17
 * exactly what the grammar is for.
16 18
 */
17 19
18 20
import type { ReplyChunk, ReplySource } from "./coder-session.js";
19 21
import { tierLabel } from "./coder-tiers.js";
22
import type { CoderTool } from "./coder-tools.js";
20 23
21 24
export interface ResponsesOptions {
22 25
  /** The API origin, such as `http://localhost:4000`. */
23 26
  readonly origin: string;
24
  /** The account bearer, sent when held; the stub also answers without one. */
27
  /** The account bearer, sent when held; the surface also answers without one. */
25 28
  readonly token?: string | undefined;
26 29
}
27 30
31
/** One conversation item, in the OpenResponses input shape. */
32
type Item =
33
  | { readonly role: "user" | "assistant" | "system"; readonly content: string }
34
  | {
35
      readonly type: "function_call";
36
      readonly call_id: string;
37
      readonly name: string;
38
      readonly arguments: string;
39
    }
40
  | { readonly type: "function_call_output"; readonly call_id: string; readonly output: string };
41
42
/** A call the model asked for, as the stream's item events carry it. */
43
interface Call {
44
  readonly callId: string;
45
  readonly name: string;
46
  readonly args: string;
47
}
48
49
/**
50
 * Rounds of tool calls one turn may take before it must answer. A backstop
51
 * against a loop, not a budget; the reader can stop a turn at any time.
52
 */
53
const MAX_TOOL_ROUNDS = 24;
54
28 55
export class ResponsesReplySource implements ReplySource {
56
  private readonly items: Item[] = [];
57
  private tools: ReadonlyArray<CoderTool> = [];
58
  private standing: string | undefined;
59
29 60
  constructor(private readonly options: ResponsesOptions) {}
30 61
31 62
  /** The dev lane defers to the server, which is what Coder Auto names. */

@@ -33,12 +64,66 @@ export class ResponsesReplySource implements ReplySource {

33 64
    return tierLabel("auto");
34 65
  }
35 66
36
  /** The product id: no vendor stands behind the stub, so none is named. */
67
  /** The product id: the server picks what answers, so none is promised. */
37 68
  get modelId(): string {
38 69
    return "openagents-coder";
39 70
  }
40 71
72
  useTools(tools: ReadonlyArray<CoderTool>): void {
73
    this.tools = tools;
74
  }
75
76
  useContext(standing: string): void {
77
    this.standing = standing;
78
  }
79
41 80
  async *reply(prompt: string, signal: AbortSignal): AsyncIterable<ReplyChunk> {
81
    this.items.push({ role: "user", content: prompt });
82
    let calls = 0;
83
84
    for (let round = 0; round < MAX_TOOL_ROUNDS; round += 1) {
85
      calls += 1;
86
      const { text, requested } = yield* this.once(signal);
87
      if (text !== "") this.items.push({ role: "assistant", content: text });
88
89
      if (requested.length === 0) {
90
        yield { type: "usage", calls };
91
        return;
92
      }
93
94
      for (const call of requested) {
95
        this.items.push({
96
          type: "function_call",
97
          call_id: call.callId,
98
          name: call.name,
99
          arguments: call.args,
100
        });
101
        const outcome = await this.run(call, signal);
102
        yield {
103
          type: "tool_result",
104
          callId: call.callId,
105
          output: outcome.output,
106
          error: outcome.error,
107
        };
108
        this.items.push({
109
          type: "function_call_output",
110
          call_id: call.callId,
111
          output: outcome.output ?? outcome.error ?? "",
112
        });
113
      }
114
    }
115
116
    yield {
117
      type: "text",
118
      value: `\n\nStopped after ${String(MAX_TOOL_ROUNDS)} rounds of tool calls without an answer.`,
119
    };
120
    yield { type: "usage", calls };
121
  }
122
123
  /** One request: stream the events, yield what renders, return the rest. */
124
  private async *once(
125
    signal: AbortSignal,
126
  ): AsyncGenerator<ReplyChunk, { text: string; requested: Call[] }> {
42 127
    const response = await fetch(new URL("/api/v1/responses", this.options.origin), {
43 128
      method: "POST",
44 129
      headers: {

@@ -49,7 +134,21 @@ export class ResponsesReplySource implements ReplySource {

49 134
        // Both named: the pipeline negotiates on json, the answer is SSE.
50 135
        accept: "text/event-stream, application/json",
51 136
      },
52
      body: JSON.stringify({ input: prompt, stream: true }),
137
      body: JSON.stringify({
138
        input: this.items,
139
        stream: true,
140
        ...(this.standing === undefined ? {} : { instructions: this.standing }),
141
        ...(this.tools.length === 0
142
          ? {}
143
          : {
144
              tools: this.tools.map((tool) => ({
145
                type: "function",
146
                name: tool.name,
147
                description: tool.description,
148
                parameters: tool.parameters,
149
              })),
150
            }),
151
      }),
53 152
      signal,
54 153
    });
55 154

@@ -59,23 +158,97 @@ export class ResponsesReplySource implements ReplySource {

59 158
      );
60 159
    }
61 160
62
    let calls = 0;
161
    let text = "";
162
    const requested: Call[] = [];
163
63 164
    for await (const data of frames(response.body, signal)) {
64 165
      const event = parse(data);
65 166
      if (event === undefined) continue;
66
      if (event["type"] === "response.output_text.delta") {
67
        const delta = event["delta"];
68
        if (typeof delta === "string" && delta.length > 0) {
69
          yield { type: "text", value: delta };
167
168
      switch (event["type"]) {
169
        case "response.output_text.delta": {
170
          const delta = event["delta"];
171
          if (typeof delta === "string" && delta.length > 0) {
172
            text += delta;
173
            yield { type: "text", value: delta };
174
          }
175
          break;
176
        }
177
        case "response.reasoning_summary_text.delta": {
178
          const delta = event["delta"];
179
          if (typeof delta === "string" && delta.length > 0) {
180
            yield { type: "reasoning", value: delta };
181
          }
182
          break;
183
        }
184
        case "response.output_item.done": {
185
          const item = event["item"];
186
          const call = functionCall(item);
187
          if (call !== undefined) {
188
            requested.push(call);
189
            yield { type: "tool_call", callId: call.callId, name: call.name, arguments: call.args };
190
          }
191
          break;
70 192
        }
193
        case "response.failed": {
194
          const failure = failureOf(event);
195
          throw new Error(`The responses API reported a failure: ${failure}`);
196
        }
197
        default:
198
          break;
71 199
      }
72
      if (event["type"] === "response.completed") calls += 1;
73 200
    }
74 201
75
    yield { type: "usage", promptTokens: 0, completionTokens: 0, calls: Math.max(calls, 1) };
202
    return { text, requested };
203
  }
204
205
  /** Run one tool call; a missing tool or a throw is a result, not a crash. */
206
  private async run(
207
    call: Call,
208
    signal: AbortSignal,
209
  ): Promise<{ output?: string; error?: string }> {
210
    const tool = this.tools.find((candidate) => candidate.name === call.name);
211
    if (tool === undefined) {
212
      return { error: `No tool named \`${call.name}\` is declared in this session.` };
213
    }
214
    let args: Record<string, unknown>;
215
    try {
216
      const parsed: unknown = JSON.parse(call.args === "" ? "{}" : call.args);
217
      args = parsed !== null && typeof parsed === "object" ? (parsed as Record<string, unknown>) : {};
218
    } catch {
219
      return { error: "The call's arguments were not valid JSON." };
220
    }
221
    try {
222
      return { output: await tool.run(args, signal) };
223
    } catch (cause) {
224
      return { error: cause instanceof Error ? cause.message : String(cause) };
225
    }
76 226
  }
77 227
}
78 228
229
const functionCall = (item: unknown): Call | undefined => {
230
  if (item === null || typeof item !== "object") return undefined;
231
  const record = item as Record<string, unknown>;
232
  if (record["type"] !== "function_call") return undefined;
233
  const callId = record["call_id"];
234
  const name = record["name"];
235
  const args = record["arguments"];
236
  if (typeof callId !== "string" || typeof name !== "string") return undefined;
237
  return { callId, name, args: typeof args === "string" ? args : "{}" };
238
};
239
240
const failureOf = (event: Record<string, unknown>): string => {
241
  const response = event["response"];
242
  if (response !== null && typeof response === "object") {
243
    const error = (response as Record<string, unknown>)["error"];
244
    if (error !== null && typeof error === "object") {
245
      const message = (error as Record<string, unknown>)["message"];
246
      if (typeof message === "string") return message;
247
    }
248
  }
249
  return "no reason was given";
250
};
251
79 252
/** Each SSE frame's `data:` payload, in order. */
80 253
async function* frames(
81 254
  body: ReadableStream<Uint8Array>,
packages/openagents-cli/test/coder-responses.test.ts modified +69 -1

@@ -56,7 +56,11 @@ describe("ResponsesReplySource", () => {

56 56
    const chunks = await collect(source);
57 57
58 58
    expect(calls[0]?.url).toBe("http://localhost:4000/api/v1/responses");
59
    expect(calls[0]?.body).toEqual({ input: "hello", stream: true });
59
    // The conversation is client-held: the input is the item history.
60
    expect(calls[0]?.body).toEqual({
61
      input: [{ role: "user", content: "hello" }],
62
      stream: true,
63
    });
60 64
    // Both named: the pipeline negotiates on json, the answer is SSE.
61 65
    expect(calls[0]?.accept).toContain("application/json");
62 66
    const text = chunks

@@ -79,3 +83,67 @@ describe("ResponsesReplySource", () => {

79 83
    await expect(collect(source)).rejects.toThrow("http://localhost:4000 answered HTTP 503");
80 84
  });
81 85
});
86
87
// The agentic loop over the surface: the model asks for a tool, the client
88
// runs it, replays the output, and the next call answers in text.
89
describe("ResponsesReplySource tools", () => {
90
  const callStream = [
91
    'event: response.output_item.added\ndata: {"type":"response.output_item.added","sequence_number":0}\n\n',
92
    'event: response.output_item.done\ndata: {"type":"response.output_item.done","sequence_number":1,"output_index":1,"item":{"type":"function_call","call_id":"call_1","name":"read_conversation","arguments":"{\\"max_turns\\":4}","status":"completed"}}\n\n',
93
    'event: response.completed\ndata: {"type":"response.completed","sequence_number":2}\n\n',
94
  ];
95
  const answerStream = [
96
    'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","sequence_number":0,"delta":"Four turns, read."}\n\n',
97
    'event: response.completed\ndata: {"type":"response.completed","sequence_number":1}\n\n',
98
  ];
99
100
  it("runs the requested tool and continues to the text answer", async () => {
101
    const bodies: Array<Record<string, unknown>> = [];
102
    const streams = [callStream, answerStream];
103
    vi.stubGlobal(
104
      "fetch",
105
      vi.fn(async (_target: URL | string, init?: RequestInit) => {
106
        bodies.push(
107
          JSON.parse(typeof init?.body === "string" ? init.body : "{}") as Record<string, unknown>,
108
        );
109
        return sse(streams.shift() ?? []);
110
      }),
111
    );
112
113
    const ran: Array<Record<string, unknown>> = [];
114
    const source = new ResponsesReplySource({ origin: "http://localhost:4000" });
115
    source.useTools([
116
      {
117
        name: "read_conversation",
118
        description: "Read a conversation back.",
119
        parameters: { type: "object" },
120
        run: (args) => {
121
          ran.push(args);
122
          return Promise.resolve("four turns of text");
123
        },
124
      },
125
    ]);
126
127
    const chunks = await collect(source);
128
129
    // The tool ran with the model's arguments.
130
    expect(ran).toEqual([{ max_turns: 4 }]);
131
    // The call and its result both rendered, then the answer.
132
    expect(chunks.map((chunk) => chunk.type)).toEqual([
133
      "tool_call",
134
      "tool_result",
135
      "text",
136
      "usage",
137
    ]);
138
    // The second request replayed the call and its output as items.
139
    const replayed = bodies[1]?.["input"] as Array<Record<string, unknown>>;
140
    expect(replayed.at(-2)).toMatchObject({ type: "function_call", call_id: "call_1" });
141
    expect(replayed.at(-1)).toMatchObject({
142
      type: "function_call_output",
143
      call_id: "call_1",
144
      output: "four turns of text",
145
    });
146
    // And declared the tools both times.
147
    expect(bodies[0]?.["tools"]).toBeDefined();
148
  });
149
});

This page updates live while a promote is in flight · changelog