})

        : undefined;

      // The dev lane records too (OpenAgentsInc/openagents#59): with a

      // credential and the switch not off, it opens the same transcript-only

      // thread the local lane opens — no grant minted or spent, the replies

      // come from the responses surface — so a dev session's history,

      // reasoning, and tool executions land in the threads database exactly

      // as a standard or ollama session's do. `openLocalThread` never throws:

      // no credential or no reachable thread route degrades silently to an

      // unrecorded session, never a broken turn.

      const devThread =

        responsesSource !== undefined && Option.isSome(stored) && threadSyncWanted(process.env)

          ? yield* Effect.promise(() =>

              openLocalThread({

                origin: endpoint.origin,

                token: Redacted.value(stored.value.token),

                objective: `openagents coder in ${workspace.repository} on ${workspace.branch}`,

                repository: workspace.repository,

                // The product id, not a vendor string: the dev server picks

                // what answers, so the record names the surface that did.

                model: responsesSource.modelId,

                reasoning: Option.getOrUndefined(reasoning),

              }),

            )

          : undefined;

      const source: ReplySource =

        wantsZen && zenAsked !== undefined && zenKey !== undefined

          ? new ZenReplySource({ model: zenAsked, key: zenKey })

        responsesSource !== undefined

          ? {

              initial: "auto" as CoderTierId,

              build: (_tier: CoderTierId, _history: ReadonlyArray<unknown>) =>

                Promise.resolve<ReplySource>(

                  new ResponsesReplySource({

                    origin: endpoint.origin,

                    ...(Option.isSome(stored)

                      ? { token: Redacted.value(stored.value.token) }

                      : {}),

                  }),

                ),

              build: (_tier: CoderTierId, _history: ReadonlyArray<unknown>) => {

                const fresh = new ResponsesReplySource({

                  origin: endpoint.origin,

                  ...(Option.isSome(stored) ? { token: Redacted.value(stored.value.token) } : {}),

                });

                // The switched source keeps writing to the same thread: the

                // writer is shared, so the record stays one sequence. Safe to

                // reference here — `build` runs on a tier switch, long after

                // the writer below is constructed.

                if (transcript !== undefined) fresh.useTranscript(transcript);

                return Promise.resolve<ReplySource>(fresh);

              },

            }

          : undefined;

      // durable copy; the offline and stand-in lanes keep no record and

      // attach nothing. A failed post never reaches the turn loop: the

      // writer queues, retries, and says so once on the status line.

      const transcriptThreadId = thread?.threadId ?? localThread?.threadId;

      const transcriptThreadId = thread?.threadId ?? localThread?.threadId ?? devThread?.threadId;

      const transcript =

        transcriptThreadId !== undefined && Option.isSome(stored)

          ? new ThreadTranscriptWriter({

      if (transcript !== undefined) {

        thread?.useTranscript(transcript);

        ollamaSource?.useTranscript(transcript);

        responsesSource?.useTranscript(transcript);

      }

      // The machine-readable announcement (OpenAgentsInc/openagents#38): in

import type { ReplyChunk, ReplySource } from "./coder-session.js";

import { tierLabel } from "./coder-tiers.js";

import type { CoderTool } from "./coder-tools.js";

import type { TranscriptSink } from "./coder-transcript.js";

/** Default backoff ladder for transient responses API failures (5xx or network drops). */

const DEFAULT_RETRY_DELAYS_MS: ReadonlyArray<number> = [250, 500, 1000, 2000];

/**

 * How much of one tool's output reaches the durable `tool.ran` event.

 *

 * The same figure the thread and local lanes use: it bounds a record written

 * once, so it is set where every result a real session has produced fits

 * whole. What the model is re-sent each round is bounded separately.

 */

const EVENT_RESULT_KEPT = 64_000;

/** A long tool result, kept at both ends. */

const bounded = (output: string, keep: number): string => {

  if (output.length <= keep) return output;

  const half = Math.floor(keep / 2);

  const cut = output.length - keep;

  return `${output.slice(0, half)}\n\n[${String(cut)} of ${String(output.length)} characters omitted from the middle; run it again more narrowly if you need them]\n\n${output.slice(-half)}`;

};

export interface ResponsesOptions {

  /** The API origin, such as `http://localhost:4000`. */

  readonly origin: string;

  private tools: ReadonlyArray<CoderTool> = [];

  private standing: string | undefined;

  private readonly retryDelaysMs: ReadonlyArray<number>;

  /**

   * The transcript writer, when the session opened a transcript-only thread

   * for this lane (OpenAgentsInc/openagents#59). Absent, nothing is recorded —

   * a `--dev` session without a credential runs exactly as before, it just

   * leaves no record on the server.

   */

  private sink: TranscriptSink | undefined;

  constructor(private readonly options: ResponsesOptions) {

    this.retryDelaysMs = options.retryDelaysMs ?? DEFAULT_RETRY_DELAYS_MS;

    this.standing = standing;

  }

  /**

   * Attach the writer that puts this session's turns on the server.

   *

   * The same vocabulary the thread and local lanes record — `turn.user`,

   * `turn.reasoning`, `tool.ran`, `turn.assistant` — so `/threads/:id`, the

   * export, and a resume read a dev session exactly as they read a hosted

   * one. Set after construction because the writer needs the thread's id,

   * which does not exist until the transcript-only thread is opened.

   */

  useTranscript(sink: TranscriptSink): void {

    this.sink = sink;

  }

  async *reply(prompt: string, signal: AbortSignal): AsyncIterable<ReplyChunk> {

    this.items.push({ role: "user", content: prompt });

    this.sink?.record("turn.user", { text: prompt });

    let calls = 0;

    /** The answer so far, across steps, for the one `turn.assistant` event. */

    let turnText = "";

    /** How many tools this turn ran, reported on `turn.assistant`. */

    let turnToolCalls = 0;

    for (;;) {

      calls += 1;

      const { text, requested } = yield* this.once(signal);

      const { text, reasoning, requested } = yield* this.once(signal);

      if (text !== "") this.items.push({ role: "assistant", content: text });

      // One event per block, whole, never deltas: the record is what was

      // thought, not the pieces it arrived in.

      if (reasoning.length > 0) this.sink?.record("turn.reasoning", { text: reasoning });

      // Whatever the model said belongs to the thread even when the turn was

      // interrupted, or the next turn answers a question it cannot see it

      // half-answered.

      if (text.length > 0) {

        turnText = turnText.length === 0 ? text : `${turnText}\n\n${text}`;

      }

      if (requested.length === 0) {

        yield { type: "usage", calls };

        this.recordAnswer(turnText, turnToolCalls, calls, signal.aborted);

        return;

      }

          call_id: call.callId,

          output: outcome.output ?? outcome.error ?? "",

        });

        turnToolCalls += 1;

        // Call and result are one fact, so they are one event — the thread

        // lane's shape exactly, bounded far above the model-wire bound so the

        // record keeps what the model was fed a cut of.

        this.sink?.record("tool.ran", {

          call_id: call.callId,

          tool: call.name,

          arguments: bounded(call.args, EVENT_RESULT_KEPT),

          status: outcome.error === undefined ? "succeeded" : "failed",

          ...(outcome.error === undefined

            ? { output: bounded(outcome.output ?? "", EVENT_RESULT_KEPT) }

            : { error: bounded(outcome.error, EVENT_RESULT_KEPT) }),

        });

      }

    }

  }

  /**

   * Record the turn's answer, with what it cost.

   *

   * One event per turn, whatever the turn took to get there, the same shape

   * the thread lane records. This surface reports no token counts, so the

   * usage carries only the call count — absent figures stay absent rather

   * than being written as zeros a reader would take for measurements.

   */

  private recordAnswer(text: string, toolCalls: number, calls: number, interrupted: boolean): void {

    if (this.sink === undefined) return;

    if (text.length === 0 && toolCalls === 0) return;

    this.sink.record("turn.assistant", {

      text,

      usage: { calls },

      tool_calls: toolCalls,

      ...(interrupted ? { interrupted: true } : {}),

    });

  }

  /**

  /** One request: stream the events, yield what renders, return the rest. */

  private async *once(

    signal: AbortSignal,

  ): AsyncGenerator<ReplyChunk, { text: string; requested: Call[] }> {

  ): AsyncGenerator<ReplyChunk, { text: string; reasoning: string; requested: Call[] }> {

    const response = await this.request(signal);

    let text = "";

    let reasoning = "";

    const requested: Call[] = [];

    for await (const data of frames(response.body!, signal)) {

        case "response.reasoning_summary_text.delta": {

          const delta = event["delta"];

          if (typeof delta === "string" && delta.length > 0) {

            reasoning += delta;

            yield { type: "reasoning", value: delta };

          }

          break;

      }

    }

    return { text, requested };

    return { text, reasoning, requested };

  }

  /** Run one tool call; a missing tool or a throw is a result, not a crash. */

  private async run(

    call: Call,

    signal: AbortSignal,

  ): Promise<{ output?: string; error?: string }> {

  private async run(call: Call, signal: AbortSignal): Promise<{ output?: string; error?: string }> {

    const tool = this.tools.find((candidate) => candidate.name === call.name);

    if (tool === undefined) {

      return { error: `No tool named \`${call.name}\` is declared in this session.` };

    let args: Record<string, unknown>;

    try {

      const parsed: unknown = JSON.parse(call.args === "" ? "{}" : call.args);

      args = parsed !== null && typeof parsed === "object" ? (parsed as Record<string, unknown>) : {};

      args =

        parsed !== null && typeof parsed === "object" ? (parsed as Record<string, unknown>) : {};

    } catch {

      return { error: "The call's arguments were not valid JSON." };

    }

import { ResponsesReplySource } from "../src/coder-responses.js";

import type { ReplyChunk } from "../src/coder-session.js";

import { ThreadTranscriptWriter } from "../src/coder-transcript.js";

const sse = (frames: ReadonlyArray<string>) =>

  new Response(

    });

    // Both named: the pipeline negotiates on json, the answer is SSE.

    expect(calls[0]?.accept).toContain("application/json");

    const text = chunks

      .map((chunk) => (chunk.type === "text" ? chunk.value : ""))

      .join("");

    const text = chunks.map((chunk) => (chunk.type === "text" ? chunk.value : "")).join("");

    expect(text).toBe("Acknowledged.");

    expect(chunks.at(-1)).toMatchObject({ type: "usage", calls: 1 });

  });

    });

    const chunks = await collect(source);

    expect(callCount).toBe(3);

    const text = chunks

      .map((chunk) => (chunk.type === "text" ? chunk.value : ""))

      .join("");

    const text = chunks.map((chunk) => (chunk.type === "text" ? chunk.value : "")).join("");

    expect(text).toBe("Acknowledged.");

  });

    });

    const chunks = await collect(source);

    expect(callCount).toBe(2);

    const text = chunks

      .map((chunk) => (chunk.type === "text" ? chunk.value : ""))

      .join("");

    const text = chunks.map((chunk) => (chunk.type === "text" ? chunk.value : "")).join("");

    expect(text).toBe("Acknowledged.");

  });

});

    expect(bodies[0]?.["tools"]).toBeDefined();

  });

});

/** A sink that just remembers, standing in for the writer. */

const recorder = () => {

  const events: Array<{ eventType: string; payload: Record<string, unknown> }> = [];

  return {

    events,

    record(eventType: string, payload: Record<string, unknown>) {

      events.push({ eventType, payload });

    },

  };

};

describe("the dev lane's durable transcript", () => {

  const callStream = [

    'event: response.reasoning_summary_text.delta\ndata: {"type":"response.reasoning_summary_text.delta","sequence_number":0,"delta":"Listing "}\n\n',

    'event: response.reasoning_summary_text.delta\ndata: {"type":"response.reasoning_summary_text.delta","sequence_number":1,"delta":"first."}\n\n',

    'event: response.output_item.done\ndata: {"type":"response.output_item.done","sequence_number":2,"output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"shell","arguments":"{\\"command\\":\\"ls\\"}","status":"completed"}}\n\n',

    'event: response.completed\ndata: {"type":"response.completed","sequence_number":3}\n\n',

  ];

  const answerStream = [

    'event: response.output_text.delta\ndata: {"type":"response.output_text.delta","sequence_number":0,"delta":"Two files."}\n\n',

    'event: response.completed\ndata: {"type":"response.completed","sequence_number":1}\n\n',

  ];

  it("records the turn in order: what was asked, the reasoning, each tool run, the answer", async () => {

    const streams = [callStream, answerStream];

    vi.stubGlobal(

      "fetch",

      vi.fn(async () => sse(streams.shift() ?? [])),

    );

    const source = new ResponsesReplySource({ origin: "http://localhost:4000", token: "oa_pat_x" });

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([

      {

        name: "shell",

        description: "run a command",

        parameters: { type: "object" },

        run: () => Promise.resolve("README.md\nsrc"),

      },

    ]);

    await collect(source);

    expect(sink.events.map((event) => event.eventType)).toEqual([

      "turn.user",

      "turn.reasoning",

      "tool.ran",

      "turn.assistant",

    ]);

    expect(sink.events[0]?.payload).toEqual({ text: "hello" });

    // One event per block, whole, never the deltas it arrived in.

    expect(sink.events[1]?.payload).toEqual({ text: "Listing first." });

    expect(sink.events[2]?.payload).toEqual({

      call_id: "call_1",

      tool: "shell",

      arguments: `{"command":"ls"}`,

      status: "succeeded",

      output: "README.md\nsrc",

    });

    // This surface reports no token counts, so the usage carries only the

    // call count rather than zeros a reader would take for measurements.

    expect(sink.events[3]?.payload).toEqual({

      text: "Two files.",

      usage: { calls: 2 },

      tool_calls: 1,

    });

  });

  it("records a failed tool run as failed, not as a crash", async () => {

    const streams = [callStream, answerStream];

    vi.stubGlobal(

      "fetch",

      vi.fn(async () => sse(streams.shift() ?? [])),

    );

    const source = new ResponsesReplySource({ origin: "http://localhost:4000" });

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([

      {

        name: "shell",

        description: "run a command",

        parameters: { type: "object" },

        run: () => Promise.reject(new Error("no such directory")),

      },

    ]);

    await collect(source);

    const ran = sink.events.find((event) => event.eventType === "tool.ran");

    expect(ran?.payload).toEqual({

      call_id: "call_1",

      tool: "shell",

      arguments: `{"command":"ls"}`,

      status: "failed",

      error: "no such directory",

    });

  });

  it("posts the recorded events to the thread's transcript on the server", async () => {

    const streams = [callStream, answerStream];

    vi.stubGlobal(

      "fetch",

      vi.fn(async () => sse(streams.shift() ?? [])),

    );

    // The writer's own transport is injected, so this test sees exactly what

    // a server at the other end would: the events route, in order.

    const posted: Array<{ url: string; body: Record<string, unknown> }> = [];

    const writer = new ThreadTranscriptWriter({

      origin: "http://localhost:4000",

      threadId: "9bb19447-ecf4-4f1b-b44e-6b128664da9c",

      token: "oa_pat_x",

      fetch: async (url, init) => {

        posted.push({

          url: url.toString(),

          body: JSON.parse(String(init?.body)) as Record<string, unknown>,

        });

        return new Response("{}", { status: 201 });

      },

    });

    const source = new ResponsesReplySource({ origin: "http://localhost:4000", token: "oa_pat_x" });

    source.useTranscript(writer);

    source.useTools([

      {

        name: "shell",

        description: "run a command",

        parameters: { type: "object" },

        run: () => Promise.resolve("README.md\nsrc"),

      },

    ]);

    await collect(source);

    await writer.close();

    expect(posted[0]?.url).toBe(

      "http://localhost:4000/api/v1/threads/9bb19447-ecf4-4f1b-b44e-6b128664da9c/events",

    );

    expect(posted.map((post) => post.body["event_type"])).toEqual([

      "turn.user",

      "turn.reasoning",

      "tool.ran",

      "turn.assistant",

    ]);

  });

  it("runs exactly as before when no transcript is attached, recording nothing", async () => {

    // The unauthenticated case: `--dev` without a credential opens no thread

    // and attaches no writer, and the turn must not know the difference.

    const streams = [callStream, answerStream];

    const fetchMock = vi.fn(async () => sse(streams.shift() ?? []));

    vi.stubGlobal("fetch", fetchMock);

    const source = new ResponsesReplySource({ origin: "http://localhost:4000" });

    source.useTools([

      {

        name: "shell",

        description: "run a command",

        parameters: { type: "object" },

        run: () => Promise.resolve("README.md\nsrc"),

      },

    ]);

    const chunks = await collect(source);

    // The turn ran whole — call, result, answer — and the only requests made

    // were the two responses calls; nothing went to a threads route.

    expect(chunks.at(-1)).toMatchObject({ type: "usage", calls: 2 });

    expect(fetchMock).toHaveBeenCalledTimes(2);

  });

});