/**
 * `tail` — follow the newest run for an issue the way `tail -f` follows a log, finishing once the run has left the live states.
 *
 * Moved out of cli.ts's switch by the per-verb module split (#462);
 * only the case wrapper, the injected `ctx` lookups and the imports
 * changed from the original bodies.
 */

import type { CommandContext } from "./context.ts";
import { closeSync, openSync, readdirSync, readSync, statSync } from "node:fs";
import { join } from "node:path";
import { findProject, loadConfig } from "../config.ts";
import { dbPath, LIVE_STATES, openStore } from "../store.ts";
import { formatTranscriptLine, prop } from "../transcript.ts";
import type { Store } from "../types.ts";
import { dim, fail, ok, warn } from "../ui/style.ts";

/** How often `tail` re-stats the transcripts it is following. */
const TAIL_POLL_MS = 1_000;

/**
 * How long the transcripts must stay unchanged, after its run has left the live
 * states, before `tail` calls it over. The state flips from the daemon's thread
 * while the harness may still be flushing its last message, so exiting on the
 * state alone truncates the ending an operator ran this command to watch.
 */
const TAIL_QUIET_MS = 5_000;

/** The harness records advisor turns under this reserved stem. */
const ADVISOR_TRANSCRIPT_PREFIX = "__advisor";

/**
 * One followed file: its open descriptor, how many bytes have been rendered,
 * and whatever byte tail did not end on a newline yet. Holding the partial line
 * across polls is what stops a UTF-8 sequence straddling a read boundary (or
 * the newline itself) from being mangled by decoding each chunk on its own.
 */
interface FollowedStream {
  fd: number;
  offset: number;
  pending: Buffer;
}

/**
 * Return the complete lines appended to `file` since the last poll, plus
 * whether any bytes changed at all (the quiet-timer needs the latter even when
 * a delta rendered no displayable line). A file that shrank (truncated or
 * replaced) restarts from zero; a final line without a newline stays in
 * `pending` until its newline lands.
 */
function followLines(
  state: FollowedStream,
  file: string,
): { lines: Buffer[]; changed: boolean } {
  const lines: Buffer[] = [];
  let changed = false;
  let size = 0;
  try {
    size = statSync(file).size;
  } catch {
    // A transcript that vanishes mid-follow is not worth crashing over.
    return { lines, changed };
  }
  // Shorter than what we have already read means truncated or replaced;
  // resuming from the old offset would read the middle of another file.
  if (size < state.offset) {
    state.offset = 0;
    state.pending = Buffer.alloc(0);
  }
  if (size > state.offset) {
    const chunk = Buffer.allocUnsafe(size - state.offset);
    const read = readSync(state.fd, chunk, 0, chunk.length, state.offset);
    state.offset += read;
    state.pending = Buffer.concat([state.pending, chunk.subarray(0, read)]);
    for (;;) {
      const nl = state.pending.indexOf(0x0a);
      if (nl < 0) break;
      lines.push(state.pending.subarray(0, nl));
      state.pending = state.pending.subarray(nl + 1);
    }
    changed = read > 0;
  }
  return { lines, changed };
}

/**
 * One advisor-transcript line rendered for the watcher, or `undefined` for the
 * lines not worth a row. An advisor speaks in two forms, and an operator
 * watching a run wants both:
 *
 *   - plain assistant text, rendered exactly like the primary transcript's
 *     (one spelling, so the two cannot drift), and
 *   - `advise` tool calls — the reviewer's actual advisory, whose note and
 *     severity live in the call arguments and would otherwise be invisible.
 *
 * The reviewer's other tool calls (`glob`/`read` probes into the workspace)
 * render nothing: they are the investigation, not the advice. Every rendered
 * line is marked so the watcher can tell which reviewer produced it. Exported
 * so a unit test pins the surface without a store.
 */
export function renderAdvisorLine(line: string): string | undefined {
  let entry: unknown;
  try {
    entry = JSON.parse(line);
  } catch {
    return undefined;
  }
  if (prop(entry, "type") !== "message") return undefined;
  const message = prop(entry, "message");
  if (prop(message, "role") !== "assistant") return undefined;

  const content = prop(message, "content");
  // A string-typed message is one plain text block — the same shape the
  // primary transcript's formatter accepts, so an advisor advisory written as
  // a bare string is not silently dropped.
  if (typeof content === "string") {
    return content.trim() === "" ? undefined : `[advisor] assistant: ${content.trim()}`;
  }
  const blocks: readonly unknown[] = Array.isArray(content) ? content : [];
  const out: string[] = [];
  for (const block of blocks) {
    const type = prop(block, "type");
    if (type === "text") {
      const text = prop(block, "text");
      if (typeof text === "string" && text.trim() !== "") out.push(`assistant: ${text.trim()}`);
    } else if (type === "toolCall" && prop(block, "name") === "advise") {
      let args = prop(block, "arguments");
      if (typeof args === "string") {
        try {
          args = JSON.parse(args);
        } catch {
          continue;
        }
      }
      const note = prop(args, "note");
      if (typeof note === "string" && note.trim() !== "") {
        const severity = prop(args, "severity");
        const label = typeof severity === "string" && severity !== "" ? severity : "nit";
        out.push(`${label}: ${note.trim()}`);
      }
    }
  }
  return out.length === 0 ? undefined : `[advisor] ${out.join("\n")}`;
}

function styleTailLine(line: string): string {
  if (line.startsWith("[advisor] blocker:")) return fail(line);
  if (line.startsWith("[advisor] warning:")) return warn(line);
  if (line.startsWith("[advisor] nit:")) return dim(line);
  if (line === "run ended: merged" || line === "run ended: pushed-green") return ok(line);
  if (/^run ended: (?:failed|killed|orphaned)$/.test(line)) return fail(line);
  if (line.startsWith("run ended:")) return warn(line);
  return line;
}

/** A poll delay that ends early when the follow is aborted. */
function sleepUntilAbort(ms: number, signal?: AbortSignal): Promise<void> {
  if (signal === undefined) return new Promise((resolve) => setTimeout(resolve, ms));
  const abort = signal;
  return new Promise((resolve) => {
    const done = (): void => {
      clearTimeout(timer);
      abort.removeEventListener("abort", done);
      resolve();
    };
    const timer = setTimeout(done, ms);
    abort.addEventListener("abort", done, { once: true });
  });
}

/**
 * Follow one run's transcripts the way `tail -f` follows a log: the worker's
 * own transcript plus any `__advisor*.jsonl` sitting beside it (the mid-run
 * reviewer's turns, #542), so an advisory the reviewer raised is visible in
 * the same stream as the work it reacted to.
 *
 * Reads from byte zero rather than from the end: attaching to a worker that is
 * already ten turns in and then showing nothing until turn eleven is not
 * watching the run. Polls `stat` instead of taking a file watcher because the
 * transcript is a plain append-only file that may sit on a filesystem where
 * change events are a polite fiction, and one stat a second costs nothing.
 *
 * SIGINT is deliberately left to its default, which is immediate exit. Nothing
 * here is buffered, and a handler could only add a poll interval of latency to
 * every Ctrl-C.
 */
export async function tailRun(
  project: string,
  issue: number,
  deps: {
    /** Read-only in practice: opened WAL with a busy timeout. Tests inject one. */
    store?: Store;
    /** How often the transcripts are re-statted (tests shrink it). */
    pollMs?: number;
    /** How long a terminal run must stay quiet before the watcher exits. */
    quietMs?: number;
    /** Where rendered lines go; the CLI writes to stdout, tests capture. */
    write?: (line: string) => void;
    /**
     * Stops the follow before the run is terminal (#296).
     *
     * The CLI passes none: `tail` runs until the run ends or the operator hits
     * Ctrl-C, and SIGINT's default is immediate exit. The dashboard's SSE stream
     * passes the request's signal, because a browser that closed its tab must
     * not leave a poll loop holding file descriptors until the run finishes —
     * repeated opens would accumulate one follow each.
     */
    signal?: AbortSignal;
  } = {},
): Promise<void> {
  const store = deps.store ?? openStore(dbPath());
  const write =
    deps.write ?? ((line: string) => process.stdout.write(`${styleTailLine(line)}\n`));
  const pollMs = deps.pollMs ?? TAIL_POLL_MS;
  const quietMs = deps.quietMs ?? TAIL_QUIET_MS;
  try {
    const run = store.latestRun(project, issue);
    if (run === undefined) throw new Error(`no run recorded for #${issue}`);
    const path = run.sessionFile;
    // Claimed but not yet started, or an attempt whose session never opened one.
    if (path === undefined) throw new Error(`no transcript yet (state: ${run.state})`);

    const primary: FollowedStream = { fd: openSync(path, "r"), offset: 0, pending: Buffer.alloc(0) };
    // Advisor transcripts live in the directory named after the primary
    // transcript stem; the harness writes them there beside the session file.
    const advisorDir = path.endsWith(".jsonl") ? path.slice(0, -".jsonl".length) : undefined;
    const advisors = new Map<string, FollowedStream>();
    try {
      let lastChange = Date.now();

      for (;;) {
        // Checked at the top as well as after the sleep: a signal that was
        // already aborted must not buy one whole poll's worth of work.
        if (deps.signal?.aborted === true) return;
        let changed = false;
        const primaryDelta = followLines(primary, path);
        changed = primaryDelta.changed || changed;
        for (const raw of primaryDelta.lines) {
          const rendered = formatTranscriptLine(raw.toString("utf8"));
          if (rendered !== undefined) write(rendered);
        }

        // Follow every advisor transcript that appears beside the primary one.
        // Files land mid-run (the reviewer's first turn), so each poll re-lists
        // the directory and opens whatever is new.
        if (advisorDir !== undefined) {
          let names: string[] = [];
          try {
            names = readdirSync(advisorDir).filter(
              (name) =>
                name.startsWith(ADVISOR_TRANSCRIPT_PREFIX) &&
                name.endsWith(".jsonl"),
            );
          } catch {
            // No advisor directory — nothing to surface.
          }
          for (const name of names) {
            const file = join(advisorDir, name);
            let state = advisors.get(file);
            if (state === undefined) {
              try {
                state = { fd: openSync(file, "r"), offset: 0, pending: Buffer.alloc(0) };
                advisors.set(file, state);
              } catch {
                continue;
              }
            }
            const advisorDelta = followLines(state, file);
            changed = advisorDelta.changed || changed;
            for (const raw of advisorDelta.lines) {
              const rendered = renderAdvisorLine(raw.toString("utf8"));
              if (rendered !== undefined) write(rendered);
            }
          }
          // Drop watched files that vanished (mid-run rotation is rare, but a
          // stale descriptor would otherwise pin the old inode forever).
          for (const [file, state] of advisors) {
            if (!names.includes(file.slice(advisorDir.length + 1))) {
              closeSync(state.fd);
              advisors.delete(file);
            }
          }
        }
        if (changed) lastChange = Date.now();

        // Re-read this exact run every poll — not `latestRun`, which would jump
        // to a retry started meanwhile and report its state against the wrong
        // transcript. The daemon writes the row from another process, so looking
        // is the only way to notice the run finished.
        const state = store.getRun(run.id)?.state ?? run.state;
        if (!LIVE_STATES.includes(state) && Date.now() - lastChange >= quietMs) {
          write(`run ended: ${state}`);
          return;
        }
        // Wakes on the signal rather than after the full poll: a closed browser
        // should release the descriptors now, not up to a second later.
        await sleepUntilAbort(pollMs, deps.signal);
      }
    } finally {
      closeSync(primary.fd);
      for (const state of advisors.values()) closeSync(state.fd);
    }
  } finally {
    store.close();
  }
}

export async function tailCommand(ctx: CommandContext): Promise<void> {
  const issue = ctx.issueArg("tail", ctx.argv[1]);
  await tailRun(findProject(loadConfig(), ctx.projectFlag).name, issue);
}