resolveOllamaModel,

} from "./coder-ollama.js";

import { openThread, ThreadUnavailable, type ThreadReplySource } from "./coder-thread.js";

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

import { delegateTool, openagentsTool, shellTool, skillTool } from "./coder-tools.js";

import {

  describeLoad,

        standingContext(skills.active(), process.cwd()),

      );

      // The thread lane writes its transcript to the server as the turn loop

      // runs — `POST /api/v3/threads/{id}/events`, on the account token that

      // opened the thread. The server copy is the only durable copy; the

      // offline, Ollama, 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 transcript =

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

          ? new ThreadTranscriptWriter({

              origin: endpoint.origin,

              threadId: thread.threadId,

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

              onTrouble: (message) => {

                session.notice(message);

              },

            })

          : undefined;

      if (transcript !== undefined) thread?.useTranscript(transcript);

      // The model is told what it can do rather than the reader being asked to

      // remember a slash command. A turn that needs three agents asks for them

      // mid-sentence, and `/delegate` stays as the way to launch a fan-out

          // expired an hour later, which is the ninth session refused for a

          // session nobody is in. A process killed outright still leaves it to

          // the server's expiry reap.

          // Flush before revoking: revoking makes the thread terminal, and a

          // terminal thread refuses the events that are still queued.

          if (transcript !== undefined) await transcript.close();

          if (thread !== undefined) await thread.revoke();

          if (childThread?.kind === "opened") await childThread.thread.revoke();

          if (setup !== undefined) await setup.close();

import { merge } from "./coder-merge.js";

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

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

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

const THREADS_PATH = "/api/v3/threads";

/** How much of one tool's output is kept on the transcript. */

const TOOL_RESULT_KEPT = 4_000;

/** A long tool result, kept at both ends, which is what it is read for. */

const boundedResult = (output: string): string => {

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

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

  const cut = output.length - TOOL_RESULT_KEPT;

/**

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

 *

 * A separate figure from `TOOL_RESULT_KEPT`, because they answer different

 * questions. The 4,000 above is a context-budget decision made against a

 * model's window: it is re-sent on every round of the turn. This one bounds a

 * record written once, so it is set where every result a real session has

 * produced fits whole — the largest measured was 8.4 KB — and only a

 * pathological dump is cut, kept at both ends the same way.

 */

const EVENT_RESULT_KEPT = 64_000;

/** A long tool output, kept at both ends, which is what it is read for. */

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)} characters omitted from the middle; run it again more narrowly if you need them]\n\n${output.slice(-half)}`;

};

const boundedResult = (output: string): string => bounded(output, TOOL_RESULT_KEPT);

/** What the thread may still spend, as the server last reported it. */

export interface ThreadBudget {

  readonly calls: number;

  private readonly transcript: WireMessage[] = [];

  private remaining: ThreadBudget;

  private tools: ReadonlyArray<CoderTool> = [];

  /**

   * Where the turn loop writes the durable transcript, when the session has

   * one. Absent, nothing is recorded — the offline and local lanes never

   * attach one — and every call below is a no-op through optional chaining.

   */

  private sink: TranscriptSink | undefined;

  /** The running turn's token usage, accumulated across its model calls. */

  private turnUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0, calls: 0 };

  /** Set for the one round that must answer rather than call another tool. */

  private mustAnswer = false;

  /**

    this.tools = tools;

  }

  /**

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

   *

   * Set after construction for the same reason `useTools` is: the writer needs

   * the thread's id, so it cannot exist until the thread does, and the failure

   * notice it surfaces needs the session, which is built later still. The

   * server copy is the only durable copy — this process keeps no transcript

   * file of its own.

   */

  useTranscript(sink: TranscriptSink): void {

    this.sink = sink;

  }

  /**

   * What this lane sends as standing context.

   *

    // Per turn, not per session: a turn that had to answer without tools must

    // not leave the next one without them.

    this.mustAnswer = false;

    this.turnUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0, 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;

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

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

    try {

      for (let step = 0; ; step += 1) {

        // model is asked again.

        for (const said of this.steered.splice(0)) {

          this.transcript.push({ role: "user", content: said });

          // Steered mid-turn rather than asked between turns, and the record

          // says so, or a replay would show a question the answer ignores.

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

        }

        const calls: WireCall[] = [];

        let assistant = "";

        let reasoning = "";

        for await (const chunk of this.stream(signal, calls)) {

          if (signal.aborted) break;

          if (chunk.type === "text") assistant += chunk.value;

          if (chunk.type === "reasoning") reasoning += chunk.value;

          yield chunk;

        }

        // One event per block, whole, never deltas. The proxy carries no

        // reasoning today, so this records nothing against a live model; the

        // day a `reasoning` chunk exists here, its record does too.

        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 (assistant.length > 0) this.transcript.push({ role: "assistant", content: assistant });

        if (signal.aborted || calls.length === 0) return;

        if (assistant.length > 0) {

          this.transcript.push({ role: "assistant", content: assistant });

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

        }

        if (signal.aborted || calls.length === 0) {

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

          return;

        }

        if (step >= MAX_TOOL_STEPS) {

          // Take the tools away for one more round rather than stopping on a

        // Concurrently. A model asking for two tools in one turn is saying they do

        // not depend on each other, and running them in order anyway makes a fan-out

        // to two models cost the sum of both.

        if (signal.aborted) return;

        if (signal.aborted) {

          this.recordAnswer(turnText, turnToolCalls, true);

          return;

        }

        turnToolCalls += calls.length;

        yield* merge(calls.map((call) => this.invoke(call, signal)));

      }

    } finally {

    }

  }

  /**

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

   *

   * One event per turn, whatever the turn took to get there: several model

   * calls, several tool rounds, one figure each. An interrupted turn is

   * recorded too, marked as such, because whatever streamed before Escape was

   * said and the next reader of this thread will be answered against it.

   */

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

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

    if (text.length === 0 && this.turnUsage.calls === 0) return;

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

      text,

      usage: {

        prompt_tokens: this.turnUsage.promptTokens,

        completion_tokens: this.turnUsage.completionTokens,

        total_tokens: this.turnUsage.totalTokens,

        calls: this.turnUsage.calls,

      },

      tool_calls: toolCalls,

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

    });

  }

  /**

   * Run one call, report it, and put the exchange on the thread.

   *

      error: failure,

    };

    // Call and result are one fact, so they are one event. The tool's name is

    // its identity for later attribution — a plugin's tool carries the name it

    // was declared under — and the result is bounded far above the model-wire

    // bound, where every result a real session has produced is stored whole.

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

      call_id: call.id,

      tool: call.name,

      arguments: bounded(call.args, EVENT_RESULT_KEPT),

      status: failure === undefined ? "succeeded" : "failed",

      ...(failure === undefined

        ? { output: bounded(output, EVENT_RESULT_KEPT) }

        : { error: bounded(failure, EVENT_RESULT_KEPT) }),

    });

    this.transcript.push({

      role: "assistant",

      content: `[tool call]\n${call.name}(${call.args})`,

      totalTokens: Math.max(0, this.remaining.totalTokens - total),

      costMicrousd: this.remaining.costMicrousd,

    };

    // The same report feeds the turn's own tally, which `turn.assistant`

    // carries: a turn is several calls, and the record holds their sum with

    // the count rather than the last call's figures presented as the turn's.

    this.turnUsage = {

      promptTokens: this.turnUsage.promptTokens + number(usage["prompt_tokens"]),

      completionTokens: this.turnUsage.completionTokens + number(usage["completion_tokens"]),

      totalTokens: this.turnUsage.totalTokens + total,

      calls: this.turnUsage.calls + 1,

    };

  }

  /** Read what the server says the thread has left. Failure keeps the estimate. */

/**

 * The thread's durable transcript, written as the turn loop runs.

 *

 * `POST /api/v3/threads/{id}/events` is append-only and the server's copy is

 * the only copy: this process keeps no file of its own, so what lands here is

 * what `--resume`, the export, and every other machine reading the thread will

 * ever see. The vocabulary is the one decided in the openagents.com audit of

 * 2026-08-24 — `turn.user`, `turn.reasoning`, `tool.ran`, `turn.assistant` —

 * and deltas and interface notices are deliberately not recorded: a delta is

 * how a reply arrived rather than what it is, and a notice never reached a

 * model.

 *

 * The writer must never cost the session anything. A turn loop that blocked on

 * a slow POST would make every tool call wait on the network twice, and a turn

 * that died because the transcript endpoint was down would have traded the work

 * for the record of it. So `record` is synchronous enqueue, one pump posts the

 * queue in order in the background, a failed post is retried with backoff, and

 * a persistent failure surfaces one notice and keeps queueing rather than ever

 * throwing into the loop that called it.

 */

const THREADS_PATH = "/api/v3/threads";

/**

 * How many consecutive failed posts before the reader is told once.

 *

 * One flaky request is the network being the network. Three in a row is an

 * outage the reader should know about, because from that point the durable

 * record is running behind the session it records.

 */

const TROUBLE_AFTER = 3;

/** Backoff between retries of one event, capped where waiting longer buys nothing. */

const RETRY_DELAYS_MS: ReadonlyArray<number> = [500, 1_000, 2_000, 5_000, 10_000, 30_000];

/**

 * How long `close` waits for the queue to drain before giving up.

 *

 * Flushing runs between the last turn and the revoke that closes the thread,

 * and a server that is down at exit must not hold the terminal open forever: a

 * reader who typed `exit` has left. What cannot be posted by this deadline is

 * lost, and that is the one place loss is accepted.

 */

const CLOSE_DEADLINE_MS = 5_000;

/** What one event needs from the caller. The writer supplies the rest. */

interface QueuedEvent {

  readonly eventType: string;

  readonly payload: Record<string, unknown>;

}

/**

 * The one call shape the writer makes of its transport. Named so a test can

 * hand in a plain function without matching `fetch`'s full overload set.

 */

export type TranscriptTransport = (input: URL, init?: RequestInit) => Promise<Response>;

/**

 * The slice of the writer a reply source calls. Narrow on purpose so a test

 * can stand a recorder in for the whole machinery.

 */

export interface TranscriptSink {

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

}

export interface TranscriptWriterOptions {

  readonly origin: string;

  readonly threadId: string;

  /** The account token, the same authority that opened the thread. */

  readonly token: string;

  /** Where one sentence about persistent failure goes. Usually the status line. */

  readonly onTrouble?: ((message: string) => void) | undefined;

  /** Injection seam for tests. Defaults to the global `fetch`. */

  readonly fetch?: TranscriptTransport | undefined;

  /** Injection seam for tests. Defaults to the production backoff ladder. */

  readonly retryDelaysMs?: ReadonlyArray<number> | undefined;

}

export class ThreadTranscriptWriter implements TranscriptSink {

  private readonly queue: QueuedEvent[] = [];

  private readonly delays: ReadonlyArray<number>;

  private readonly post: TranscriptTransport;

  private pumping = false;

  private consecutiveFailures = 0;

  private toldOfTrouble = false;

  /**

   * Set when the server said the thread is terminal. Nothing can ever land on

   * a closed transcript, so from here events are dropped rather than queued

   * against a refusal that cannot change.

   */

  private threadClosed = false;

  /**

   * Set when `close` has run. The session is leaving: whatever could not be

   * posted by the flush deadline is not retried into a process that no longer

   * exists, and that is the one place loss is accepted.

   */

  private stopped = false;

  /** Resolvers waiting in `close` for the queue to drain. */

  private drainWaiters: Array<() => void> = [];

  constructor(private readonly options: TranscriptWriterOptions) {

    this.delays = options.retryDelaysMs ?? RETRY_DELAYS_MS;

    this.post = options.fetch ?? globalThis.fetch.bind(globalThis);

  }

  /** How many events are queued and not yet on the server. For tests and `close`. */

  get pending(): number {

    return this.queue.length;

  }

  /**

   * Queue one event for the thread's transcript.

   *

   * Returns immediately. Events post in the order they were recorded, whatever

   * the network does in between, because the transcript is a sequence and a

   * reordered one describes a session that never happened.

   */

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

    if (this.threadClosed || this.stopped) return;

    this.queue.push({ eventType, payload });

    // The pump is started here rather than awaited: the caller is the turn

    // loop, and the whole contract is that it never waits on this.

    void this.pump().catch(() => undefined);

  }

  /**

   * Wait for the queue to drain, up to a deadline.

   *

   * Called before the thread is revoked, because revoking closes the

   * transcript and anything still queued would then be refused

   * `thread_terminal`. Failure to drain by the deadline resolves rather than

   * rejects: a session on its way out has nobody left to throw to.

   */

  async close(deadlineMs = CLOSE_DEADLINE_MS): Promise<void> {

    try {

      if (this.queue.length === 0 && !this.pumping) return;

      await new Promise<void>((resolve) => {

        const timer = setTimeout(() => {

          this.drainWaiters = this.drainWaiters.filter((waiter) => waiter !== settle);

          resolve();

        }, deadlineMs);

        const settle = () => {

          clearTimeout(timer);

          resolve();

        };

        this.drainWaiters.push(settle);

      });

    } finally {

      // Whatever the flush achieved, the writer stops here: a retry loop that

      // outlived the session it was recording would hold the process open for

      // a transcript nobody is in.

      this.stopped = true;

    }

  }

  /**

   * Post the queue, in order, one event at a time.

   *

   * One pump runs at a time. A transient failure — the network refusing, a

   * 5xx — retries the same event up the backoff ladder and never skips it,

   * because posting the next event first would reorder the transcript. A

   * refusal that cannot change is treated by kind: `thread_terminal` closes

   * the writer, and any other 4xx drops that one event and moves on, since a

   * payload the server has already called invalid will be invalid tomorrow.

   */

  private async pump(): Promise<void> {

    if (this.pumping) return;

    this.pumping = true;

    try {

      while (this.queue.length > 0 && !this.threadClosed && !this.stopped) {

        const event = this.queue[0];

        if (event === undefined) break;

        // Events post strictly in order; each depends on the one before it

        // being on the server, so there is nothing here to run concurrently.

        // eslint-disable-next-line no-await-in-loop

        const outcome = await this.send(event);

        if (outcome === "posted") {

          this.queue.shift();

          this.consecutiveFailures = 0;

          continue;

        }

        if (outcome === "thread_closed") {

          this.threadClosed = true;

          this.queue.length = 0;

          this.trouble(

            "This thread is closed, so the rest of the session will not reach its transcript.",

          );

          break;

        }

        if (outcome === "refused") {

          // The server named the event invalid. Retrying cannot change that,

          // and holding the queue on it would silently stop the record.

          this.queue.shift();

          this.trouble(

            "The server refused a transcript event, which was dropped. The session continues.",

          );

          continue;

        }

        // Transient. Same event, next rung of the ladder.

        this.consecutiveFailures += 1;

        if (this.consecutiveFailures >= TROUBLE_AFTER) {

          this.trouble(

            "The thread transcript is not reaching the server. " +

              "Events are queued and will keep retrying in the background.",

          );

        }

        const rung = Math.min(this.consecutiveFailures - 1, this.delays.length - 1);

        // eslint-disable-next-line no-await-in-loop

        await sleep(this.delays[rung] ?? 0);

      }

    } finally {

      this.pumping = false;

      if (this.queue.length === 0 || this.threadClosed) {

        for (const waiter of this.drainWaiters.splice(0)) waiter();

      }

    }

  }

  /** One POST, translated to what the pump can act on. Never throws. */

  private async send(

    event: QueuedEvent,

  ): Promise<"posted" | "retry" | "refused" | "thread_closed"> {

    let response: Response;

    try {

      response = await this.post(

        new URL(`${THREADS_PATH}/${this.options.threadId}/events`, this.options.origin),

        {

          method: "POST",

          headers: {

            authorization: `Bearer ${this.options.token}`,

            "content-type": "application/json",

            accept: "application/json",

          },

          body: JSON.stringify({ event_type: event.eventType, payload: event.payload }),

        },

      );

    } catch {

      return "retry";

    }

    if (response.status >= 200 && response.status < 300) return "posted";

    // A server that is down answers 5xx; the event is still good.

    if (response.status >= 500) return "retry";

    const body = (await response.json().catch(() => ({}))) as Record<string, unknown>;

    if (body["code"] === "thread_terminal") return "thread_closed";

    return "refused";

  }

  /** One sentence, once. A status line repeating itself is a status line ignored. */

  private trouble(message: string): void {

    if (this.toldOfTrouble) return;

    this.toldOfTrouble = true;

    this.options.onTrouble?.(message);

  }

}

const sleep = (ms: number) =>

  new Promise<void>((resolve) => {

    // Always a real timer, even at zero: a retry loop that resumed on the

    // microtask queue would starve the event loop the session runs on. The

    // timer is unreferenced so a backoff mid-wait cannot hold the process

    // open after the session that was being recorded has ended.

    const timer = setTimeout(resolve, Math.max(0, ms));

    if (typeof timer === "object" && "unref" in timer) timer.unref();

  });

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

import { openThread, ThreadUnavailable, type ThreadReplySource } from "../src/coder-thread.js";

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

const ORIGIN = "https://openagents.test";

const ACCOUNT_TOKEN = "oa_pat_account";

    await expect((await open()).revoke()).resolves.toBeUndefined();

  });

});

/** 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 });

    },

  };

};

const withTool = (run: (args: Record<string, unknown>) => Promise<string>) => ({

  name: "shell",

  description: "run a command",

  parameters: { type: "object" },

  run,

});

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

  const TOOL_ROUND = [

    `data: {"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"id":"call-1","type":"function","function":{"name":"shell","arguments":"{\\"command\\":\\"ls\\"}"}}]}}]}`,

    `data: {"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}`,

    `data: {"choices":[],"usage":{"completion_tokens":5,"prompt_tokens":40,"total_tokens":45}}`,

    `data: [DONE]`,

    "",

  ].join("\n\n");

  const ANSWER_ROUND = [

    `data: {"choices":[{"delta":{"content":"Two files."},"index":0}]}`,

    `data: {"choices":[],"usage":{"completion_tokens":11,"prompt_tokens":60,"total_tokens":71}}`,

    `data: [DONE]`,

    "",

  ].join("\n\n");

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

    stub({ proxy: [sse([TOOL_ROUND]), sse([ANSWER_ROUND])] });

    const source = await open();

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([withTool(async () => "README.md\nsrc")]);

    await chunks(source, "what is in this repo?");

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

      "turn.user",

      "tool.ran",

      "turn.assistant",

    ]);

    expect(sink.events[0]?.payload).toEqual({ text: "what is in this repo?" });

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

      call_id: "call-1",

      tool: "shell",

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

      status: "succeeded",

      output: "README.md\nsrc",

    });

  });

  it("records the answer with the turn's summed usage and its call count", async () => {

    stub({ proxy: [sse([TOOL_ROUND]), sse([ANSWER_ROUND])] });

    const source = await open();

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([withTool(async () => "README.md\nsrc")]);

    await chunks(source, "what is in this repo?");

    const answer = sink.events.find((event) => event.eventType === "turn.assistant");

    // The turn took two model calls; the record holds their sum with the

    // count, not the last call's figures presented as the turn's.

    expect(answer?.payload).toEqual({

      text: "Two files.",

      usage: { prompt_tokens: 100, completion_tokens: 16, total_tokens: 116, calls: 2 },

      tool_calls: 1,

    });

  });

  it("records a failed tool as one event carrying its error", async () => {

    stub({ proxy: [sse([TOOL_ROUND]), sse([ANSWER_ROUND])] });

    const source = await open();

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([

      withTool(async () => {

        throw new Error("permission denied");

      }),

    ]);

    await chunks(source, "try it");

    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: "permission denied",

    });

  });

  it("bounds a tool result on its way into the record", async () => {

    stub({ proxy: [sse([TOOL_ROUND]), sse([ANSWER_ROUND])] });

    const source = await open();

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([withTool(async () => "x".repeat(200_000))]);

    await chunks(source, "dump it");

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

    const output = ran?.payload["output"] as string;

    // Kept at both ends around a marker, and far above the model-wire bound,

    // so every result a real session has produced is stored whole.

    expect(output.length).toBeLessThan(70_000);

    expect(output).toContain("characters omitted");

  });

  it("records a message steered into the running turn as the reader's", async () => {

    stub({ proxy: [sse([TOOL_ROUND]), sse([ANSWER_ROUND])] });

    const source = await open();

    const sink = recorder();

    source.useTranscript(sink);

    source.useTools([

      withTool(async () => {

        source.steer("only the top level");

        return "README.md\nsrc";

      }),

    ]);

    await chunks(source, "list the files");

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

      "turn.user",

      "tool.ran",

      "turn.user",

      "turn.assistant",

    ]);

    expect(sink.events[2]?.payload).toEqual({ text: "only the top level", steered: true });

  });

  it("records nothing extra for a turn without tools", async () => {

    stub({});

    const source = await open();

    const sink = recorder();

    source.useTranscript(sink);

    await chunks(source, "hello");

    expect(sink.events.map((event) => event.eventType)).toEqual(["turn.user", "turn.assistant"]);

    expect(sink.events[1]?.payload).toMatchObject({ text: "Hello! Nice", tool_calls: 0 });

  });

  it("still answers when every transcript post fails", async () => {

    // The real writer against a server that refuses the events route: the

    // turn loop must neither throw nor wait on it.

    vi.stubGlobal(

      "fetch",

      vi.fn(async (target: URL | string, init?: RequestInit) => {

        const url = typeof target === "string" ? target : target.toString();

        if (url.endsWith("/events")) throw new Error("socket closed");

        if (url.endsWith("/api/inference/proxy")) return sse([LIVE_SSE]);

        if ((init?.method ?? "GET") === "POST") return json(201, CREATED);

        return json(200, { thread: {}, grant: {} });

      }),

    );

    const source = await openThread({

      origin: ORIGIN,

      token: ACCOUNT_TOKEN,

      objective: "coder in repo on main",

    });

    const notices: string[] = [];

    const writer = new ThreadTranscriptWriter({

      origin: ORIGIN,

      threadId: source.threadId,

      token: ACCOUNT_TOKEN,

      retryDelaysMs: [0, 0, 10],

      onTrouble: (message) => {

        notices.push(message);

      },

    });

    source.useTranscript(writer);

    expect(textOf(await chunks(source))).toBe("Hello! Nice");

    await writer.close(100);

    expect(notices).toHaveLength(1);

  });

});

import { describe, expect, it, vi } from "vitest";

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

const ORIGIN = "https://openagents.test";

const TOKEN = "oa_pat_account";

const THREAD_ID = "9bb19447-ecf4-4f1b-b44e-6b128664da9c";

const json = (status: number, body: unknown) =>

  new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });

const posted = () => json(201, { thread: { id: THREAD_ID, event_count: 1 } });

interface Recorded {

  readonly url: string;

  readonly method: string;

  readonly authorization: string;

  readonly body: Record<string, unknown>;

}

/**

 * A transport that answers from a script, one response or failure per call,

 * and records what was sent. The last entry repeats, so a drain after the

 * interesting part needs no padding.

 */

const transport = (script: ReadonlyArray<Response | Error>) => {

  const calls: Recorded[] = [];

  let at = 0;

  const fetch = vi.fn(async (target: URL | string, init?: RequestInit) => {

    const headers = (init?.headers ?? {}) as Record<string, string>;

    calls.push({

      url: typeof target === "string" ? target : target.toString(),

      method: init?.method ?? "GET",

      authorization: headers["authorization"] ?? "",

      body: JSON.parse(typeof init?.body === "string" ? init.body : "{}") as Record<

        string,

        unknown

      >,

    });

    const answer = script[Math.min(at, script.length - 1)];

    at += 1;

    if (answer instanceof Error) throw answer;

    return (answer ?? posted()).clone();

  });

  return { calls, fetch };

};

const writer = (

  wire: { fetch: TranscriptTransport },

  extra?: Partial<ConstructorParameters<typeof ThreadTranscriptWriter>[0]>,

) =>

  new ThreadTranscriptWriter({

    origin: ORIGIN,

    threadId: THREAD_ID,

    token: TOKEN,

    fetch: wire.fetch,

    retryDelaysMs: [0, 0, 0],

    ...extra,

  });

describe("ThreadTranscriptWriter", () => {

  it("posts each event to the thread's transcript on the account token", async () => {

    const wire = transport([posted()]);

    const sink = writer(wire);

    sink.record("turn.user", { text: "list the open issues" });

    await sink.close();

    expect(wire.calls).toHaveLength(1);

    const call = wire.calls[0];

    expect(call?.method).toBe("POST");

    expect(call?.url).toBe(`${ORIGIN}/api/v3/threads/${THREAD_ID}/events`);

    expect(call?.authorization).toBe(`Bearer ${TOKEN}`);

    expect(call?.body).toEqual({

      event_type: "turn.user",

      payload: { text: "list the open issues" },

    });

  });

  it("posts events in the order they were recorded", async () => {

    const wire = transport([posted()]);

    const sink = writer(wire);

    sink.record("turn.user", { text: "hello" });

    sink.record("tool.ran", { tool: "shell", status: "succeeded" });

    sink.record("turn.assistant", { text: "done" });

    await sink.close();

    expect(wire.calls.map((call) => call.body["event_type"])).toEqual([

      "turn.user",

      "tool.ran",

      "turn.assistant",

    ]);

  });

  it("retries a refused connection and keeps the order across the retry", async () => {

    const wire = transport([new Error("socket closed"), posted()]);

    const sink = writer(wire);

    sink.record("turn.user", { text: "first" });

    sink.record("turn.assistant", { text: "second" });

    await sink.close();

    // The failed event is retried, not skipped: posting the next one first

    // would reorder the transcript.

    expect(wire.calls.map((call) => call.body["event_type"])).toEqual([

      "turn.user",

      "turn.user",

      "turn.assistant",

    ]);

    expect(sink.pending).toBe(0);

  });

  it("retries a server failure the same way", async () => {

    const wire = transport([json(502, {}), posted()]);

    const sink = writer(wire);

    sink.record("turn.user", { text: "hello" });

    await sink.close();

    expect(wire.calls).toHaveLength(2);

    expect(sink.pending).toBe(0);

  });

  it("never throws into the caller, whatever the transport does", async () => {

    const sink = writer({

      fetch: (() => {

        throw new Error("broken before the promise");

      }) as unknown as TranscriptTransport,

    });

    expect(() => {

      sink.record("turn.user", { text: "hello" });

    }).not.toThrow();

    // Stop the retry loop rather than leaving it running under later tests.

    await sink.close(20);

  });

  it("says so once when posting keeps failing, and keeps queueing", async () => {

    const notices: string[] = [];

    const wire = transport([new Error("down")]);

    const sink = writer(wire, {

      // A real rung after the notice, so the retry loop idles rather than

      // spins while the close deadline runs down.

      retryDelaysMs: [0, 0, 10],

      onTrouble: (message) => {

        notices.push(message);

      },

    });

    sink.record("turn.user", { text: "one" });

    sink.record("turn.user", { text: "two" });

    await sink.close(200);

    // Three consecutive failures are an outage worth one sentence; the fourth

    // and fifth are the same outage and get no second sentence.

    expect(notices).toHaveLength(1);

    expect(notices[0]).toContain("not reaching the server");

    // Nothing was dropped: both events are still queued for retry.

    expect(sink.pending).toBe(2);

  });

  it("drops an event the server called invalid and continues with the next", async () => {

    const notices: string[] = [];

    const wire = transport([json(422, { errors: { event_type: ["cannot be blank"] } }), posted()]);

    const sink = writer(wire, {

      onTrouble: (message) => {

        notices.push(message);

      },

    });

    sink.record("turn.user", { text: "refused" });

    sink.record("turn.assistant", { text: "still recorded" });

    await sink.close();

    // A payload the server has already called invalid will be invalid

    // tomorrow; retrying it would silently stop the whole record.

    expect(wire.calls.map((call) => call.body["event_type"])).toEqual([

      "turn.user",

      "turn.assistant",

    ]);

    expect(notices).toHaveLength(1);

    expect(sink.pending).toBe(0);

  });

  it("stops for good when the thread is terminal", async () => {

    const notices: string[] = [];

    const wire = transport([json(422, { code: "thread_terminal" })]);

    const sink = writer(wire, {

      onTrouble: (message) => {

        notices.push(message);

      },

    });

    sink.record("turn.user", { text: "one" });

    sink.record("turn.user", { text: "two" });

    await sink.close();

    sink.record("turn.user", { text: "three" });

    await sink.close();

    // A closed transcript can never take another event, so nothing is queued

    // against a refusal that cannot change.

    expect(wire.calls).toHaveLength(1);

    expect(notices).toHaveLength(1);

    expect(sink.pending).toBe(0);

  });

  it("gives up the flush at the deadline rather than holding the exit", async () => {

    const sink = writer({

      // A transport that never answers, which is what a dead server looks

      // like from a process that is trying to leave.

      fetch: (() => new Promise<Response>(() => undefined)) as TranscriptTransport,

    });

    sink.record("turn.user", { text: "unsendable" });

    const started = Date.now();

    await sink.close(50);

    expect(Date.now() - started).toBeLessThan(1_000);

  });

});