import { runCoderPlain } from "./coder-plain.js";

import { CoderSession, DummyReplySource } from "./coder-session.js";

import { runCoderUi } from "./coder-ui.js";

import { OxAlphaReplySource } from "./coder-ox.js";

import { describeWorkspace } from "./coder-workspace.js";

import { ComputerConfiguration, type ComputerConfigurationValues } from "./computer-config.js";

import { ComputerJournal, journalMaxBytes, journalReadTailBytes } from "./computer-journal.js";

const loginHeadlessFlag = Flag.boolean("headless").pipe(

  Flag.withDescription("Return an authorization URL and code without waiting for approval"),

);

const loginScopeFlag = Flag.string("scope").pipe(

  Flag.atLeast(0),

  Flag.withDescription(

    "Request a scope for the new token; repeatable. Omit to take the server's default. Use chat:account to reach the chat API from `openagents coder`.",

  ),

);

const loginResumeFlag = Flag.boolean("resume").pipe(

  Flag.withDescription("Complete the pending device authorization for the selected API"),

);

const authLoginCommand = Command.make(

  "login",

  { tokenStdin: loginTokenStdinFlag, headless: loginHeadlessFlag, resume: loginResumeFlag },

  ({ headless, resume, tokenStdin }) =>

  {

    tokenStdin: loginTokenStdinFlag,

    headless: loginHeadlessFlag,

    resume: loginResumeFlag,

    scope: loginScopeFlag,

  },

  ({ headless, resume, scope, tokenStdin }) =>

    Effect.gen(function* () {

      if ((tokenStdin && (headless || resume)) || (headless && resume)) {

        return yield* new InputError({

      if (!resume) {

        const devices = yield* DeviceClient;

        const terminal = yield* TerminalSession;

        const authorization = yield* devices.start(endpoint.origin);

        const authorization = yield* devices.start(endpoint.origin, scope);

        const returnForApproval = headless || !terminal.interactive || flags.json;

        if (returnForApproval) {

          const now = yield* Clock.currentTimeMillis;

const coderPlainFlag = Flag.boolean("plain").pipe(

  Flag.withDescription("Use line-oriented output with no cursor control, even on a terminal"),

);

const coderOfflineFlag = Flag.boolean("offline").pipe(

  Flag.withDescription("Answer from the built-in stand-in instead of the chat API"),

);

const coderReasoningFlag = Flag.choice("reasoning", [

  "minimal",

  "low",

  "medium",

  "high",

  "max",

]).pipe(Flag.optional, Flag.withDescription("Reasoning effort the server passes to the provider"));

const coderCommand = Command.make(

  "coder",

  { prompt: coderPrompt, plain: coderPlainFlag },

  ({ prompt, plain }) =>

  {

    prompt: coderPrompt,

    plain: coderPlainFlag,

    offline: coderOfflineFlag,

    reasoning: coderReasoningFlag,

  },

  ({ prompt, plain, offline, reasoning }) =>

    Effect.gen(function* () {

      const flags = yield* rootCommand;

      const terminal = yield* TerminalSession;

      const workspace = describeWorkspace();

      const endpoint = yield* resolveApiEndpoint(endpointOverrides(flags));

      // The reply source is a stand-in until the ACP client lands. Nothing

      // else in this command changes when it is replaced.

      const session = new CoderSession(

        new DummyReplySource(),

        workspace.repository,

        workspace.branch,

      );

      // Replies come from the account chat API, which runs Ox Alpha through

      // OpenRouter on the server, so the CLI never holds a provider key and a

      // session costs exactly what the server metered. Without a credential it

      // falls back to the stand-in and says so rather than failing.

      const stored = offline

        ? Option.none()

        : yield* findToken(endpoint.origin).pipe(

            Effect.catchTag("OpenAgentsCli.CredentialPersistenceUnavailable", () =>

              Effect.succeed(Option.none()),

            ),

          );

      const source = Option.isSome(stored)

        ? new OxAlphaReplySource({

            origin: endpoint.origin,

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

            reasoning: Option.getOrUndefined(reasoning),

          })

        : new DummyReplySource();

      const session = new CoderSession(source, workspace.repository, workspace.branch);

      if (Option.isNone(stored) && !offline) {

        session.notice(

          "No stored credential, so replies come from the built-in stand-in. " +

            "Run `openagents auth login --scope chat:account` to reach Ox Alpha.",

        );

      }

      const oneShot = Option.getOrUndefined(prompt);

      const interactive = terminal.interactive && !plain && !flags.json && oneShot === undefined;

    }),

).pipe(

  Command.withDescription(

    "Open a terminal coding session. Development build: the interface, the session loop, and streaming are real; the replies are a stand-in until the agent runtime is attached",

    "Open a terminal coding session. Replies come from Ox Alpha through the account chat API; --offline answers from a built-in stand-in instead",

  ),

);

/**

 * A reply source backed by the account chat API, which runs Ox Alpha through

 * OpenRouter on the server.

 *

 * The CLI never holds a provider key and never talks to OpenRouter. It submits

 * a turn and reads the durable event log the server writes, so a coder session

 * costs exactly what the server metered and leaves the same receipts the web

 * surface leaves.

 *

 * Two properties of the shipped contract shape this file:

 *

 * - The server records one conversation per account (`DATA-002`) and one active

 *   turn per conversation (`TURN-001`). A coder session therefore shares the

 *   account's conversation rather than opening its own, and a second turn while

 *   one is running is refused with `turn_in_progress` rather than queued.

 * - `GET /api/v3/chat/events` returns the conversation's whole event log, not a

 *   stream. This polls it and yields what is new, which is why the reply

 *   appears in pieces rather than at once.

 */

const SUBMIT_PATH = "/api/v3/chat/turns";

const EVENTS_PATH = "/api/v3/chat/events";

const POLL_INTERVAL_MS = 250;

/** Give up rather than poll forever when a turn never reaches a terminal event. */

const TURN_TIMEOUT_MS = 300_000;

export interface OxAlphaOptions {

  readonly origin: string;

  readonly token: string;

  /** Reasoning effort the server passes to the provider. */

  readonly reasoning?: string | undefined;

  readonly model?: string | undefined;

}

interface ChatEvent {

  readonly id?: string;

  readonly run_id?: string;

  readonly sequence?: number;

  readonly type?: string;

  readonly payload?: Record<string, unknown>;

}

export class OxAlphaUnavailable extends Error {

  constructor(

    readonly code: string,

    message: string,

  ) {

    super(message);

    this.name = "OxAlphaUnavailable";

  }

}

/**

 * Submit a turn and yield the assistant text as the server records it.

 *

 * Reasoning deltas are read but not yielded: the transcript shows what the

 * assistant said, and the runtime already treats a thought as something a

 * client may drop.

 */

export class OxAlphaReplySource {

  readonly model: string;

  constructor(private readonly options: OxAlphaOptions) {

    this.model = options.model ?? "stealth/ox-alpha";

  }

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

    const seen = await this.latestSequence();

    const runId = await this.submit(prompt, signal);

    const startedAt = Date.now();

    const delivered = new Map<string, number>();

    if (runId !== undefined) delivered.set(runId, seen.get(runId) ?? -1);

    while (!signal.aborted) {

      if (Date.now() - startedAt > TURN_TIMEOUT_MS) {

        throw new OxAlphaUnavailable(

          "turn_timed_out",

          "The turn produced no terminal event within five minutes.",

        );

      }

      const events = await this.events(signal);

      if (signal.aborted) return;

      // A submit that answered without a run id still identifies its run in the

      // log; take the newest run not already accounted for.

      const target = runId ?? newestRun(events, seen);

      if (target === undefined) {

        await sleep(POLL_INTERVAL_MS, signal);

        continue;

      }

      const floor = delivered.get(target) ?? seen.get(target) ?? -1;

      let highest = floor;

      let finished = false;

      for (const event of events) {

        if (event.run_id !== target) continue;

        const sequence = typeof event.sequence === "number" ? event.sequence : -1;

        if (sequence <= floor) continue;

        highest = Math.max(highest, sequence);

        if (event.type === "text_delta") {

          const value = event.payload?.["value"];

          if (typeof value === "string" && value.length > 0) yield value;

        } else if (event.type === "completed") {

          finished = true;

        } else if (event.type === "failed") {

          const reason = event.payload?.["message"] ?? event.payload?.["error"];

          throw new OxAlphaUnavailable(

            "turn_failed",

            typeof reason === "string" ? reason : "The turn failed on the server.",

          );

        }

      }

      delivered.set(target, highest);

      if (finished) return;

      await sleep(POLL_INTERVAL_MS, signal);

    }

  }

  /** Highest sequence per run before submitting, so old events are not replayed. */

  private async latestSequence(): Promise<Map<string, number>> {

    const seen = new Map<string, number>();

    for (const event of await this.events()) {

      if (typeof event.run_id !== "string") continue;

      const sequence = typeof event.sequence === "number" ? event.sequence : -1;

      seen.set(event.run_id, Math.max(seen.get(event.run_id) ?? -1, sequence));

    }

    return seen;

  }

  private async submit(prompt: string, signal: AbortSignal): Promise<string | undefined> {

    const response = await fetch(new URL(SUBMIT_PATH, this.options.origin), {

      method: "POST",

      signal,

      headers: {

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

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

        accept: "application/json",

      },

      body: JSON.stringify({

        message: prompt,

        ...(this.options.reasoning === undefined ? {} : { reasoning: this.options.reasoning }),

      }),

    });

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

    if (response.status === 401 || response.status === 403) {

      throw new OxAlphaUnavailable(

        "scope_missing",

        "This token cannot reach the chat API. Sign in again with the chat:account scope.",

      );

    }

    if (response.status === 409) {

      throw new OxAlphaUnavailable(

        "turn_in_progress",

        "The account already has a turn running. One turn runs at a time.",

      );

    }

    if (response.status === 429) {

      throw new OxAlphaUnavailable("rate_limited", "The chat API is rate limiting this account.");

    }

    if (response.status < 200 || response.status >= 300) {

      const code = typeof body["error"] === "string" ? body["error"] : `http_${response.status}`;

      throw new OxAlphaUnavailable(code, `The chat API refused the turn (${code}).`);

    }

    const turn = body["turn"];

    if (turn !== null && typeof turn === "object") {

      const id = (turn as Record<string, unknown>)["id"];

      if (typeof id === "string") return id;

    }

    return undefined;

  }

  private async events(signal?: AbortSignal): Promise<ReadonlyArray<ChatEvent>> {

    const response = await fetch(new URL(EVENTS_PATH, this.options.origin), {

      ...(signal === undefined ? {} : { signal }),

      headers: {

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

        accept: "application/json",

      },

    });

    if (response.status === 401 || response.status === 403) {

      throw new OxAlphaUnavailable(

        "scope_missing",

        "This token cannot read chat events. Sign in again with the chat:account scope.",

      );

    }

    if (response.status < 200 || response.status >= 300) return [];

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

    const events = body["events"];

    return Array.isArray(events) ? (events as ReadonlyArray<ChatEvent>) : [];

  }

}

/** The newest run in the log that the pre-submit snapshot did not know about. */

function newestRun(

  events: ReadonlyArray<ChatEvent>,

  seen: ReadonlyMap<string, number>,

): string | undefined {

  for (let index = events.length - 1; index >= 0; index -= 1) {

    const runId = events[index]?.run_id;

    if (typeof runId === "string" && !seen.has(runId)) return runId;

  }

  return undefined;

}

function sleep(ms: number, signal: AbortSignal): Promise<void> {

  return new Promise((resolve) => {

    if (signal.aborted) return resolve();

    const timer = setTimeout(finish, ms);

    signal.addEventListener("abort", finish, { once: true });

    function finish() {

      clearTimeout(timer);

      signal.removeEventListener("abort", finish);

      resolve();

    }

  });

}

  const { stdin, stdout, prompt } = options;

  let written = 0;

  // Notices are tracked by their text rather than by position: a failed turn

  // removes the empty assistant entry, so an index into the transcript can move

  // backwards and skip the very notice that explains the failure.

  const reported = new Set<string>();

  const flush = () => {

    const snapshot = session.snapshot();

    const last = snapshot.entries.at(-1);

    const entries = session.snapshot().entries;

    // Notices carry refusals and failures. Dropping them here is how a failed

    // turn becomes a silent empty reply, so they are written as they arrive.

    for (const entry of entries) {

      if (entry.role !== "notice" || reported.has(entry.text)) continue;

      reported.add(entry.text);

      stdout.write(`${entry.text}\n`);

    }

    const last = entries.at(-1);

    if (last === undefined || last.role !== "assistant") return;

    if (last.text.length > written) {

      stdout.write(last.text.slice(written));

  };

  const unsubscribe = session.onChange(flush);

  flush();

  const answer = async (line: string) => {

    written = 0;

        // failure, so what the agent already said stays on the transcript.

        reply.text += "\n\n[interrupted]";

      }

    } catch (cause) {

      // A failed turn ends the turn, not the session. The reason belongs on the

      // transcript where the prompt that caused it is still visible, and any

      // text the source produced before failing is kept.

      const message = cause instanceof Error ? cause.message : String(cause);

      if (reply.text.length === 0) {

        this.entries.splice(this.entries.indexOf(reply), 1);

      }

      this.entries.push({ role: "notice", text: message, settled: true });

    } finally {

      reply.settled = true;

      this.controller = undefined;

) {}

export interface DeviceClientInterface {

  readonly start: (origin: string) => Effect.Effect<DeviceAuthorization, CliError>;

  readonly start: (

    origin: string,

    scopes?: ReadonlyArray<string>,

  ) => Effect.Effect<DeviceAuthorization, CliError>;

  readonly wait: (

    origin: string,

    authorization: DeviceAuthorization,

  Effect.gen(function* () {

    const transport = yield* ApiTransport;

    const start = Effect.fn("DeviceClient.start")(function* (origin: string) {

    const start = Effect.fn("DeviceClient.start")(function* (

      origin: string,

      scopes?: ReadonlyArray<string>,

    ) {

      const response = yield* transport.request({

        origin,

        method: "POST",

        path: "/api/v3/device/authorizations",

        body: {},

        // The server decides the default scope set. Asking for none keeps that

        // decision on the server; asking names exactly what the approval page

        // must show the person approving it.

        body: scopes === undefined || scopes.length === 0 ? {} : { scope: scopes.join(" ") },

      });

      if (response.status !== 201) {

        return yield* new ApiError({

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

import { OxAlphaReplySource, OxAlphaUnavailable } from "../src/coder-ox.js";

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

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

const collect = async (source: OxAlphaReplySource, prompt = "hello") => {

  const chunks: string[] = [];

  for await (const chunk of source.reply(prompt, new AbortController().signal)) chunks.push(chunk);

  return chunks.join("");

};

const source = () =>

  new OxAlphaReplySource({ origin: "https://openagents.test", token: "test-token" });

afterEach(() => {

  vi.unstubAllGlobals();

});

/** Answer `GET /chat/events` from a script and `POST /chat/turns` with a run id. */

const stubFetch = (pages: ReadonlyArray<ReadonlyArray<unknown>>, submit = json(202, {})) => {

  let page = 0;

  const calls: string[] = [];

  vi.stubGlobal(

    "fetch",

    vi.fn((url: URL, init?: RequestInit) => {

      const path = url.pathname;

      calls.push(`${init?.method ?? "GET"} ${path}`);

      if (path.endsWith("/chat/turns")) return Promise.resolve(submit.clone());

      const events = pages[Math.min(page, pages.length - 1)] ?? [];

      page += 1;

      return Promise.resolve(json(200, { events }));

    }),

  );

  return calls;

};

describe("OxAlphaReplySource", () => {

  it("yields text deltas for the submitted run and stops at completed", async () => {

    stubFetch(

      [

        [],

        [

          { run_id: "run-1", sequence: 1, type: "text_delta", payload: { value: "Hello" } },

          { run_id: "run-1", sequence: 2, type: "text_delta", payload: { value: " there" } },

          { run_id: "run-1", sequence: 3, type: "completed", payload: {} },

        ],

      ],

      json(202, { turn: { id: "run-1" } }),

    );

    expect(await collect(source())).toBe("Hello there");

  });

  it("does not replay events that existed before the turn was submitted", async () => {

    stubFetch(

      [

        [{ run_id: "run-0", sequence: 9, type: "text_delta", payload: { value: "OLD" } }],

        [

          { run_id: "run-0", sequence: 9, type: "text_delta", payload: { value: "OLD" } },

          { run_id: "run-1", sequence: 1, type: "text_delta", payload: { value: "new" } },

          { run_id: "run-1", sequence: 2, type: "completed", payload: {} },

        ],

      ],

      json(202, { turn: { id: "run-1" } }),

    );

    expect(await collect(source())).toBe("new");

  });

  it("does not repeat a delta already delivered on an earlier poll", async () => {

    stubFetch(

      [

        [],

        [{ run_id: "run-1", sequence: 1, type: "text_delta", payload: { value: "one" } }],

        [

          { run_id: "run-1", sequence: 1, type: "text_delta", payload: { value: "one" } },

          { run_id: "run-1", sequence: 2, type: "text_delta", payload: { value: " two" } },

          { run_id: "run-1", sequence: 3, type: "completed", payload: {} },

        ],

      ],

      json(202, { turn: { id: "run-1" } }),

    );

    expect(await collect(source())).toBe("one two");

  });

  it("ignores reasoning deltas, which the transcript does not show", async () => {

    stubFetch(

      [

        [],

        [

          { run_id: "run-1", sequence: 1, type: "reasoning_delta", payload: { value: "thinking" } },

          { run_id: "run-1", sequence: 2, type: "text_delta", payload: { value: "said" } },

          { run_id: "run-1", sequence: 3, type: "completed", payload: {} },

        ],

      ],

      json(202, { turn: { id: "run-1" } }),

    );

    expect(await collect(source())).toBe("said");

  });

  it("reports a missing scope rather than an empty reply", async () => {

    stubFetch([[]], json(401, { error: "invalid_api_token" }));

    await expect(collect(source())).rejects.toThrow(OxAlphaUnavailable);

    await expect(collect(source())).rejects.toThrow(/chat:account/);

  });

  it("names the one-turn-at-a-time rule when the server refuses a concurrent turn", async () => {

    stubFetch([[]], json(409, { error: "turn_in_progress" }));

    await expect(collect(source())).rejects.toThrow(/one turn runs at a time/i);

  });

  it("surfaces a failed turn with the server's reason", async () => {

    stubFetch(

      [

        [],

        [

          {

            run_id: "run-1",

            sequence: 1,

            type: "failed",

            payload: { message: "provider refused" },

          },

        ],

      ],

      json(202, { turn: { id: "run-1" } }),

    );

    await expect(collect(source())).rejects.toThrow(/provider refused/);

  });

  it("sends the prompt and the reasoning effort the caller chose", async () => {

    const seen: Array<Record<string, unknown>> = [];

    let submitted = false;

    vi.stubGlobal(

      "fetch",

      vi.fn((url: URL, init?: RequestInit) => {

        if (url.pathname.endsWith("/chat/turns")) {

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

          submitted = true;

          return Promise.resolve(json(202, { turn: { id: "run-1" } }));

        }

        // The run does not exist until it is submitted, which is what the

        // pre-submit snapshot relies on.

        return Promise.resolve(

          json(200, {

            events: submitted

              ? [{ run_id: "run-1", sequence: 1, type: "completed", payload: {} }]

              : [],

          }),

        );

      }),

    );

    const configured = new OxAlphaReplySource({

      origin: "https://openagents.test",

      token: "test-token",

      reasoning: "high",

    });

    await collect(configured, "do the thing");

    expect(seen[0]).toEqual({ message: "do the thing", reasoning: "high" });

  });

  it("reports the model it runs on", () => {

    expect(source().model).toBe("stealth/ox-alpha");

  });

});