stack/packages/conversation/src/turns.mjs

// Run tracking, overlap signals and receipt settlement (#1507, CHAT-03 §3).
//
// Pinned Pi ties no event to a prompt. Under the seal (pi-pin.mjs) the Mosaic
// prompt in the slot is the only thing that can start a run, so a run that
// follows the slot's ack is attributed to it by order. The tracker watches
// for signs that the seal failed (O1–O6) and for lead decision 34's
// `aborted` with no stop in progress, and settles the slot's receipt only on
// evidence tied to its own prompt (§3 rule 4).
//
// Every line read from the engine gets an observation index (`read()`), so
// "before the ack", "since the fence" and "before the last abort" are exact
// comparisons on one ordered observation.
//
// A run is one prompt's run, from its first `agent_start` to its
// `agent_settled`; retries and compaction can add further `agent_start` …
// `agent_end` pairs inside it (agent-session.js 787–810).

export const DONE_STOPS = Object.freeze(new Set(["stop", "length", "toolUse"]));

export class Tracker {
  // `onOverlap(signal, detail)`; `onWorking(slot)`; `onSlotSettled(slot,
  // result)` with result { state, reason, stop? } or { outcomeUnknown };
  // `stopLink()` returns the stop ID an `aborted` may be linked to, or null
  // (lead decision 34).
  constructor({ onOverlap, onWorking, onSlotSettled, stopLink }) {
    this.onOverlap = onOverlap;
    this.onWorking = onWorking;
    this.onSlotSettled = onSlotSettled;
    this.stopLink = stopLink;
    this.idx = 0;
    this.slot = null;
    this.lastSlot = null;
    this.runs = [];
    this.current = null;
    this.overlaps = [];
    this.gap = null;
    this.clearWindow = 0;
    this.runSerial = 0;
    this.waiters = new Set();
    this.ignoredAgentEnds = 0;
    this.looseSettles = [];
  }

  get overlapped() {
    return this.overlaps.length > 0;
  }

  read() {
    return ++this.idx;
  }

  overlap(signal, idx, detail = {}) {
    const entry = { signal, idx, request: (this.slot ?? this.lastSlot)?.request ?? null, ...detail };
    this.overlaps.push(entry);
    this.onOverlap(signal, entry);
  }

  attach(slot) {
    this.slot = slot;
    Object.assign(slot, { ackIdx: null, runId: null, working: false, settleSeen: false, failureNoRun: false });
  }

  acked(idx) {
    if (this.slot) this.slot.ackIdx = idx;
  }

  release() {
    if (this.slot) this.lastSlot = this.slot;
    this.slot = null;
  }

  // Runs whose first agent_start was read after `idx`.
  startedAfter(idx) {
    return this.runs.filter((r) => r.startIdx > idx);
  }

  settledAfter(idx) {
    return this.runs.some((r) => r.settleIdx !== null && r.settleIdx > idx) || this.looseSettles.some((i) => i > idx);
  }

  waitSettle(run, ms) {
    if (run.settleIdx !== null) return Promise.resolve(true);
    return new Promise((resolve) => {
      const w = () => {
        if (run.settleIdx === null) return;
        this.waiters.delete(w);
        clearTimeout(t);
        resolve(true);
      };
      const t = setTimeout(() => {
        this.waiters.delete(w);
        resolve(false);
      }, ms);
      this.waiters.add(w);
    });
  }

  // A run in the slot's window, or null.
  slotRun() {
    return this.slot?.runId ? this.runs.find((r) => r.id === this.slot.runId) ?? null : null;
  }

  event(ev, idx) {
    switch (ev.type) {
      case "agent_start":
        return this.#agentStart(idx);
      case "agent_end":
        if (this.current && this.current.openStarts > 0) this.current.openStarts -= 1;
        else this.ignoredAgentEnds += 1;
        return undefined;
      case "message_start":
        if (ev.message?.role === "user") return this.#userStart(idx);
        return undefined;
      case "message_end":
        if (ev.message?.role === "assistant") return this.#assistantEnd(ev.message, idx);
        return undefined;
      case "agent_settled":
        return this.#settled(idx);
      case "queue_update": {
        const empty = (Array.isArray(ev.steering) ? ev.steering.length : 1) === 0 && (Array.isArray(ev.followUp) ? ev.followUp.length : 1) === 0;
        // Pi's own clear emits an empty queue_update before its response
        // (agent-session.js 1201, rpc-mode.js 334): that one is caused (N25).
        if (empty && this.clearWindow > 0) return undefined;
        return this.overlap("O5", idx, { cause: empty ? "uncaused-queue-update" : "queue-update", items: queueDigestCount(ev) });
      }
      default:
        return undefined;
    }
  }

  #agentStart(idx) {
    if (this.current) {
      this.current.openStarts += 1;
      return;
    }
    const run = { id: `run-${++this.runSerial}`, startIdx: idx, settleIdx: null, openStarts: 1, userStarts: 0, lastStop: null, lastError: null, failureBeforeUser: false, slot: false };
    this.runs.push(run);
    this.current = run;
    const s = this.slot;
    if (!s || s.ackIdx === null || s.runId !== null) {
      this.overlap("O1", idx, { run: run.id, why: !s ? "no slot held" : s.ackIdx === null ? "before the slot's ack" : "after the slot's run" });
      return;
    }
    s.runId = run.id;
    run.slot = true;
  }

  #userStart(idx) {
    const run = this.current;
    if (!run) return;
    run.userStarts += 1;
    if (run.userStarts === 2) this.overlap("O6", idx, { run: run.id });
    const s = this.slot;
    if (s && run.id === s.runId && run.userStarts === 1 && !s.working && !this.overlapped && !this.gap) {
      s.working = true;
      this.onWorking(s);
    }
  }

  #assistantEnd(message, idx) {
    const stopReason = typeof message.stopReason === "string" ? message.stopReason : null;
    // Lead decision 34: only the controller's abort produces `aborted`.
    if (stopReason === "aborted" && !this.stopLink()) this.overlap("aborted-without-stop", idx, { run: this.current?.id ?? null });
    const run = this.current;
    if (!run) {
      // Pi's run-failure handler can emit a failure message with no
      // agent_start (pi-agent-core agent.js 349–364).
      if (this.slot && this.slot.ackIdx !== null && (stopReason === "error" || stopReason === "aborted")) this.slot.failureNoRun = true;
      return;
    }
    run.lastStop = stopReason;
    run.lastError = typeof message.errorMessage === "string" ? message.errorMessage.slice(0, 2000) : null;
    if ((stopReason === "error" || stopReason === "aborted") && run.userStarts === 0) run.failureBeforeUser = true;
  }

  #settled(idx) {
    const run = this.current;
    const s = this.slot;
    if (!run) {
      this.looseSettles.push(idx);
      if (!s || s.ackIdx === null) return this.overlap("O2", idx, { why: !s ? "no slot held" : "before the slot's ack" });
      if (s.settleSeen) return this.overlap("O2", idx, { why: "a second settle for one ack" });
      s.settleSeen = true;
      if (s.runId !== null) return this.overlap("O2", idx, { why: "a second settle for one ack" });
      if (!s.failureNoRun) return this.overlap("O4", idx, { why: "settle after the ack with no agent_start and no failure message" });
      if (this.overlapped || this.gap) return undefined;
      return this.onSlotSettled(s, { state: "delivery-unknown", reason: "ack-without-start" });
    }
    if (run.openStarts > 0) {
      // O3: the settle does not close this run; the run stays current.
      return this.overlap("O3", idx, { run: run.id });
    }
    run.settleIdx = idx;
    this.current = null;
    for (const w of [...this.waiters]) w();
    if (!run.slot) {
      if (!s || s.ackIdx === null) this.overlap("O2", idx, { run: run.id, why: "settle of a run that is not the slot's" });
      return undefined;
    }
    if (!s || s.runId !== run.id) return this.overlap("O2", idx, { run: run.id, why: "the slot's run settled after the slot moved on" });
    if (s.settleSeen) return this.overlap("O2", idx, { why: "a second settle for one ack" });
    s.settleSeen = true;
    if (this.overlapped || this.gap) return undefined;
    return this.onSlotSettled(s, classify(s, run, this.stopLink));
  }
}

// §3 rule 4, for a settle that closes the slot's own run with no overlap.
export function classify(slot, run, stopLink) {
  if (!slot.working) return { state: "delivery-unknown", reason: "ack-without-start" };
  const s = run.lastStop;
  if (DONE_STOPS.has(s)) return { state: "finished", reason: null };
  if (s === "error") return { state: "failed", reason: null, nativeError: run.lastError };
  if (s === "aborted") {
    const stop = stopLink();
    // Unreachable without a stop: #assistantEnd raised the overlap already.
    return stop ? { state: "failed", reason: "interrupted", stop } : { outcomeUnknown: "aborted-without-stop" };
  }
  return { outcomeUnknown: s === null ? "no-final-assistant-message" : "unrecognised-stop-reason" };
}

function queueDigestCount(ev) {
  const all = [...(Array.isArray(ev.steering) ? ev.steering : []), ...(Array.isArray(ev.followUp) ? ev.followUp : [])];
  return all.length;
}