import * as plugins from './plugins.js';
import type { IControllerCodexAccount, IControllerCodexActivity, IControllerCodexActivitySummary, IControllerCodexCollaborationMode, IControllerCodexModelChoice, IControllerCodexModelOption, IControllerEvent, IControllerMessagePage, IControllerMessageBundle, IControllerPermission, IControllerQuestion, IControllerSession, TControllerCodexCollaborationMode, TControllerCodexCollaborationModeAuthority } from '../ts_interfaces/index.js';
import { codexAccount, codexRateLimits, codexNonnegativeInteger } from './functions.codexaccount.js';
import { controllerMaxToolEventBytes } from '../ts_interfaces/index.js';
import { CodexSupervisor } from './classes.codexsupervisor.js';
import { codexArray, codexBoundedText, codexItemBundle, codexRecord, codexRuntimeId, codexSession, codexSessionId, codexString, codexText, codexThreadId, isCodexTerminalItemStatus, type TCodexTerminalTurnStatus } from './classes.codexprojection.js';
import { codexPublicThreadId, codexQualifiedIdentity } from './functions.codexidentity.js';

interface ICodexSessionScope { projectId: string; directory: string; session: IControllerSession; revision: number; statusRevision: number; closedRevision: number }
interface ICodexLiveItem { turnId: string; item: Record<string, unknown>; completed: boolean; bytes: number; terminalTurnStatus?: TCodexTerminalTurnStatus }
interface ICodexPendingRequest { bytes: number; request: plugins.crossharness.ICodexAppServerRequest; permission?: IControllerPermission; question?: IControllerQuestion; questionIds?: string[] }
interface ICodexTurnAdmission { turnId?: string; clientMessageId: string; onUserMessage?: (idArg: string) => void; terminal: Set<string>; armed: boolean; complete: () => void }
interface ICodexSteeredInput { clientMessageId: string; turnId: string; armed: boolean; terminal: boolean; onUserMessage: (idArg: string) => void; complete: () => void }
interface ICodexNativeCollaborationMode {
  mode: TControllerCodexCollaborationMode;
  settings: { model: string; reasoning_effort: string | null; developer_instructions: null };
}
export interface ICodexPreparedCollaborationMode {
  native: ICodexNativeCollaborationMode;
  public: IControllerCodexCollaborationMode;
}
interface ICodexModeWaiter {
  expected: string;
  observed?: IControllerCodexCollaborationMode;
  resolve: (modeArg: IControllerCodexCollaborationMode) => void;
}
type TCodexStoredActivity = Omit<IControllerCodexActivity, 'canInterrupt' | 'collaborationModeAuthority'>;
/**
 * What a turn read proved about the conversation's in-progress turn. `unknown` is its own outcome:
 * an unanswered read is evidence of nothing and must never be read as `none`.
 */
type TCodexActiveTurn = { state: 'inProgress'; turnId: string } | { state: 'none' } | { state: 'unknown' };
export interface ICodexEventClock { streamEpoch: number; revision: number }

/**
 * Byte budget for the output of a live tool event. Well below `controllerMaxToolEventBytes`,
 * because the same event also carries the command, the working directory and the identities, and
 * an event that misses the budget is not sent at all — the browser is told to re-read the history
 * instead, which is far more expensive than a shorter prefix.
 */
const codexLiveToolOutputBytes = 3 * 1024;

/** A dispatched mode mutation completed without exact native settings evidence. */
export class CodexCollaborationModeOutcomeUnknownError extends Error {
  constructor(optionsArg?: ErrorOptions) {
    super('The Codex mode update outcome is unknown. Do not retry it automatically.', optionsArg);
    this.name = 'CodexCollaborationModeOutcomeUnknownError';
  }
}

/** Native app-server projection. Only explicitly admitted AGL threads may subscribe or answer requests. */
export class CodexClientAdapter {
  private readonly scopes = new Map<string, ICodexSessionScope>();
  private readonly loaded = new Set<string>();
  private readonly emptyThreads = new Set<string>();
  private readonly loading = new Map<string, Promise<void>>();
  private readonly liveItems = new Map<string, Map<string, ICodexLiveItem>>();
  private readonly pending = new Map<string, ICodexPendingRequest>();
  private readonly turns = new Map<string, ICodexTurnAdmission>();
  private readonly activities = new Map<string, TCodexStoredActivity>();
  private readonly interruptibleTurns = new Map<string, string>();
  private readonly steeredInputs = new Map<string, Map<string, ICodexSteeredInput>>();
  private readonly modeWaiters = new Map<string, ICodexModeWaiter>();
  private readonly modeUpdates = new Set<string>();
  private readonly modeAuthorities = new Map<string, TControllerCodexCollaborationModeAuthority>();
  private readonly released = new Set<string>();
  private liveBytes = 0;
  constructor(public readonly supervisor: CodexSupervisor, private readonly emit: (eventArg: IControllerEvent) => void,
    private readonly clock: ICodexEventClock = { streamEpoch: Date.now(), revision: 0 },
    private readonly identity?: { profileId: string; legacyThreadIds: ReadonlySet<string> }) {}

  public get profileId(): string | undefined { return this.identity?.profileId; }

  public rawThreadId(publicIdArg: string): string {
    const qualified = codexQualifiedIdentity(publicIdArg);
    if (qualified) {
      if (qualified.profileId !== this.identity?.profileId) throw new Error('Codex conversation belongs to another server profile.');
      return qualified.rawThreadId;
    }
    if (this.identity && !this.identity.legacyThreadIds.has(publicIdArg)) throw new Error('Codex conversation has no legacy binding to this server.');
    return codexThreadId(publicIdArg);
  }

  private publicThreadId(rawIdArg: unknown): string {
    const id = codexThreadId(rawIdArg);
    return !this.identity || this.identity.legacyThreadIds.has(id) ? id : codexPublicThreadId(this.identity.profileId, id);
  }

  private publicThread(valueArg: unknown): Record<string, unknown> {
    const thread = codexRecord(valueArg);
    return { ...thread, id: this.publicThreadId(thread.id),
      ...(typeof thread.parentThreadId === 'string' ? { parentThreadId: this.publicThreadId(thread.parentThreadId) } : {}),
      ...(typeof thread.forkedFromId === 'string' ? { forkedFromId: this.publicThreadId(thread.forkedFromId) } : {}) };
  }

  public get streamEpoch(): number { return this.clock.streamEpoch; }
  public get cursor(): { streamEpoch: number; revision: number } { return { ...this.clock }; }

  public activity(threadIdArg: string): IControllerCodexActivity {
    const scope = this.scope(threadIdArg);
    const writer = this.released.has(threadIdArg) ? 'released' : this.turns.has(threadIdArg)
      ? 'agl' : scope.session.status === 'busy' ? 'external' : 'idle';
    const activity = this.activities.get(threadIdArg);
    const modeAuthority = this.modeAuthorities.get(threadIdArg);
    let projectedActivity = activity;
    if (modeAuthority && activity) {
      const { collaborationMode: _collaborationMode, ...activityWithoutMode } = activity;
      projectedActivity = activityWithoutMode;
    }
    const turnId = writer === 'agl' || writer === 'external' ? activity?.turnId : undefined;
    return structuredClone({ ...projectedActivity, turnId, writer,
      ...(modeAuthority ? { collaborationModeAuthority: modeAuthority } : {}),
      canInterrupt: writer !== 'released' && scope.session.status === 'busy'
        && turnId !== undefined && this.interruptibleTurns.get(threadIdArg) === turnId });
  }

  public activitySummary(threadIdArg: string): IControllerCodexActivitySummary {
    const { writer, turnId, canInterrupt, collaborationModeAuthority } = this.activity(threadIdArg);
    return { writer, ...(turnId === undefined ? {} : { turnId }), canInterrupt,
      ...(collaborationModeAuthority ? { collaborationModeAuthority } : {}) };
  }

  /** Prove an exact active turn only through this connected app-server. */
  public async refreshActivity(threadIdArg: string, signalArg?: AbortSignal): Promise<IControllerCodexActivitySummary> {
    const current = this.activitySummary(threadIdArg);
    if (current.writer === 'released' || current.writer === 'idle' || current.canInterrupt) return current;
    this.interruptibleTurns.delete(threadIdArg);
    const result = await this.request('thread/turns/list', {
      threadId: threadIdArg, limit: 1, sortDirection: 'desc', itemsView: 'notLoaded',
    }, signalArg);
    const latest = codexArray(result.data, 1)[0];
    if (!latest || codexRecord(latest).status !== 'inProgress') {
      const previous = this.activities.get(threadIdArg);
      if (previous?.turnId !== undefined) this.activities.set(threadIdArg, { ...previous, turnId: undefined });
      return this.activitySummary(threadIdArg);
    }
    const turnId = codexThreadId(codexRecord(latest).id);
    const previous = this.activities.get(threadIdArg);
    if (previous?.turnId !== turnId) this.activities.set(threadIdArg, {
      ...previous, writer: current.writer, turnId, reroute: undefined, plan: undefined,
      diff: undefined, diffTruncated: undefined,
    });
    this.interruptibleTurns.set(threadIdArg, turnId);
    return this.activitySummary(threadIdArg);
  }

  /**
   * The conversation's in-progress turn, as the app-server itself states it. A thread has at most
   * one such turn and it is always the newest, so one turn read answers it — without loading the
   * turn's items, without depending on the notifications AGL may have missed, and regardless of
   * which client owns the turn.
   *
   * The read is auxiliary: it refines a page AGL can already render, so a request that fails, is
   * aborted, or answers with something that is not a turn identity leaves the active turn unknown
   * instead of failing the page. Reading a conversation must not depend on it.
   */
  private async readActiveTurn(threadIdArg: string, signalArg?: AbortSignal): Promise<TCodexActiveTurn> {
    try {
      const result = await this.request('thread/turns/list', {
        threadId: threadIdArg, limit: 1, sortDirection: 'desc', itemsView: 'notLoaded',
      }, signalArg);
      const latest = codexArray(result.data, 1)[0];
      if (!latest || codexRecord(latest).status !== 'inProgress') return { state: 'none' };
      return { state: 'inProgress', turnId: codexThreadId(codexRecord(latest).id) };
    } catch { return { state: 'unknown' }; }
  }

  /**
   * Records the thread's model and reasoning effort as the native app-server reports them. They
   * are what the conversation subheading states answered last, so they are tracked from every
   * source that can change them: a resumed thread, a turn AGL started, and a native settings
   * change or reroute made outside AGL.
   */
  public observeSettings(threadIdArg: string, resultArg: Record<string, unknown>): void {
    this.scope(threadIdArg);
    const previous = this.activities.get(threadIdArg) ?? { writer: 'idle' as const };
    const effort = resultArg.reasoningEffort ?? resultArg.effort;
    const collaborationMode = resultArg.collaborationMode === undefined
      ? undefined
      : this.parseCollaborationMode(resultArg.collaborationMode).public;
    this.activities.set(threadIdArg, { ...previous,
      ...(typeof resultArg.model === 'string' ? { model: codexString(resultArg.model, 512) } : {}),
      ...(typeof effort === 'string' ? { effort: codexString(effort, 128) } : {}),
      ...(collaborationMode === undefined ? {} : { collaborationMode }) });
    const waiter = this.modeWaiters.get(threadIdArg);
    if (collaborationMode) {
      if (waiter) waiter.observed = collaborationMode;
      const authority = this.modeAuthorities.get(threadIdArg);
      if (authority?.status === 'unconfirmed') {
        this.modeAuthorities.delete(threadIdArg);
      } else if (authority?.status === 'pending'
        && waiter?.expected === JSON.stringify(collaborationMode)) {
        this.modeAuthorities.delete(threadIdArg);
        waiter.resolve(collaborationMode);
      }
    }
  }

  public canDispatchModeDependent(threadIdArg: string, explicitModeArg = false): boolean {
    this.scope(threadIdArg);
    return !this.modeUpdates.has(threadIdArg)
      && (explicitModeArg || !this.modeAuthorities.has(threadIdArg));
  }

  private assertModeDependentDispatchAllowed(threadIdArg: string, explicitModeArg = false): void {
    if (this.canDispatchModeDependent(threadIdArg, explicitModeArg)) return;
    throw new plugins.crossharness.CodexAppServerRequestError(
      'The Codex collaboration mode is changing or its outcome is unconfirmed.',
      false,
    );
  }

  private parseCollaborationMode(valueArg: unknown): ICodexPreparedCollaborationMode {
    const mode = codexRecord(valueArg);
    if (mode.mode !== 'default' && mode.mode !== 'plan') throw new Error('Invalid Codex collaboration mode.');
    const settings = codexRecord(mode.settings);
    const model = codexString(settings.model, 512);
    const reasoningEffort = settings.reasoning_effort;
    if (reasoningEffort !== null && typeof reasoningEffort !== 'string') {
      throw new Error('Invalid Codex collaboration mode effort.');
    }
    const effort = reasoningEffort === null ? undefined : codexString(reasoningEffort, 128);
    return {
      native: {
        mode: mode.mode,
        settings: { model, reasoning_effort: effort ?? null, developer_instructions: null },
      },
      public: { mode: mode.mode, model, ...(effort === undefined ? {} : { effort }) },
    };
  }

  public async prepareCollaborationMode(
    modeArg: TControllerCodexCollaborationMode,
    modelArg: IControllerCodexModelChoice,
    signalArg?: AbortSignal,
  ): Promise<ICodexPreparedCollaborationMode> {
    await this.validateModel(modelArg, signalArg);
    const result = await this.request('collaborationMode/list', {}, signalArg);
    const matches = codexArray(result.data, 32).map((valueArg) => codexRecord(valueArg))
      .filter((valueArg) => valueArg.mode === modeArg);
    if (matches.length !== 1) throw new Error(`Codex did not return one ${modeArg} collaboration mode preset.`);
    const preset = matches[0]!;
    const presetModel = preset.model === null || preset.model === undefined
      ? modelArg.modelID
      : codexString(preset.model, 512);
    const presetEffort = preset.reasoning_effort === null || preset.reasoning_effort === undefined
      ? modelArg.variant
      : codexString(preset.reasoning_effort, 128);
    return {
      native: {
        mode: modeArg,
        settings: {
          model: presetModel,
          reasoning_effort: presetEffort ?? null,
          developer_instructions: null,
        },
      },
      public: {
        mode: modeArg,
        model: presetModel,
        ...(presetEffort === undefined ? {} : { effort: presetEffort }),
      },
    };
  }

  public async setCollaborationMode(
    threadIdArg: string,
    modeArg: TControllerCodexCollaborationMode,
    modelArg: IControllerCodexModelChoice,
    signalArg?: AbortSignal,
    beforeDispatchArg?: () => void,
  ): Promise<IControllerCodexCollaborationMode> {
    this.scope(threadIdArg);
    if (this.modeUpdates.has(threadIdArg)) throw new Error('A Codex mode update is already pending.');
    this.modeUpdates.add(threadIdArg);
    try {
      await this.ensureLoaded(threadIdArg, signalArg);
      if ((await this.readSession(this.scope(threadIdArg).directory, threadIdArg, signalArg)).status === 'busy') {
        throw new Error('A Codex turn is already running. Wait for it to finish before changing mode.');
      }
      const prepared = await this.prepareCollaborationMode(modeArg, modelArg, signalArg);
      if (this.turns.has(threadIdArg)) {
        throw new Error('A Codex turn is already active or its outcome is unknown.');
      }
      const previousAuthority = this.modeAuthorities.get(threadIdArg);
      const expected = JSON.stringify(prepared.public);
      const waitSignal = signalArg
        ? AbortSignal.any([signalArg, this.supervisor.signal, AbortSignal.timeout(30_000)])
        : AbortSignal.any([this.supervisor.signal, AbortSignal.timeout(30_000)]);
      let confirmed: IControllerCodexCollaborationMode | undefined;
      let resolveConfirmation!: (modeArg: IControllerCodexCollaborationMode) => void;
      let abortConfirmation!: () => void;
      const notification = new Promise<IControllerCodexCollaborationMode>((resolve, reject) => {
        resolveConfirmation = (modeValueArg) => { confirmed = modeValueArg; resolve(modeValueArg); };
        abortConfirmation = () => reject(waitSignal.reason);
        waitSignal.addEventListener('abort', abortConfirmation, { once: true });
        if (waitSignal.aborted) abortConfirmation();
      });
      const waiter: ICodexModeWaiter = { expected, resolve: resolveConfirmation };
      this.modeWaiters.set(threadIdArg, waiter);
      this.modeAuthorities.set(threadIdArg, { status: 'pending', requested: prepared.public });
      this.changed(threadIdArg, 'session.changed');
      let requestStarted = false;
      try {
        try {
          beforeDispatchArg?.();
          requestStarted = true;
          await this.request('thread/settings/update', {
            threadId: threadIdArg,
            collaborationMode: prepared.native,
          }, waitSignal);
        } catch (error) {
          if (confirmed) return confirmed;
          const definitivelyRejected = !requestStarted
            || (error instanceof plugins.crossharness.CodexAppServerRequestError
              && (!error.dispatched || error.code !== undefined));
          if (definitivelyRejected) {
            if (this.modeAuthorities.get(threadIdArg)?.status === 'pending') {
              if (waiter.observed) this.modeAuthorities.delete(threadIdArg);
              else if (previousAuthority) this.modeAuthorities.set(threadIdArg, previousAuthority);
              else this.modeAuthorities.delete(threadIdArg);
              this.changed(threadIdArg, 'session.changed');
            }
            throw error;
          }
          this.modeAuthorities.set(threadIdArg, {
            status: 'unconfirmed',
            requested: prepared.public,
          });
          this.changed(threadIdArg, 'session.changed');
          throw new CodexCollaborationModeOutcomeUnknownError({ cause: error });
        }
        try {
          return await notification;
        } catch (error) {
          if (confirmed) return confirmed;
          this.modeAuthorities.set(threadIdArg, {
            status: 'unconfirmed',
            requested: prepared.public,
          });
          this.changed(threadIdArg, 'session.changed');
          throw new CodexCollaborationModeOutcomeUnknownError({ cause: error });
        }
      } finally {
        if (this.modeWaiters.get(threadIdArg) === waiter) this.modeWaiters.delete(threadIdArg);
        waitSignal.removeEventListener('abort', abortConfirmation);
        void notification.catch(() => undefined);
      }
    } finally {
      if (this.modeUpdates.delete(threadIdArg) && this.scopes.has(threadIdArg)) {
        this.changed(threadIdArg, 'session.changed');
      }
    }
  }

  public async account(signalArg?: AbortSignal): Promise<IControllerCodexAccount> {
    const account = codexAccount(await this.request('account/read', { refreshToken: false }, signalArg));
    if (account.type !== 'chatgpt') return account;
    const results = await Promise.allSettled([
      this.request('account/rateLimits/read', {}, signalArg), this.request('account/usage/read', {}, signalArg),
    ]);
    signalArg?.throwIfAborted(); this.supervisor.signal.throwIfAborted();
    if (results[0].status === 'fulfilled') {
      account.rateLimits = codexRateLimits(results[0].value);
      if (results[0].value.accountId !== null && results[0].value.accountId !== undefined) {
        account.accountId = codexString(results[0].value.accountId, 512);
      }
      account.rateLimitsAvailable = true;
    }
    if (results[1].status === 'fulfilled') {
      const usage = codexRecord(results[1].value.summary);
      if (usage.lifetimeTokens !== null) account.lifetimeTokens = codexNonnegativeInteger(usage.lifetimeTokens);
      if (usage.peakDailyTokens !== null) account.peakDailyTokens = codexNonnegativeInteger(usage.peakDailyTokens);
      account.usageAvailable = true;
    }
    return account;
  }

  public async request(methodArg: string, paramsArg: Record<string, unknown>, signalArg?: AbortSignal): Promise<Record<string, unknown>> {
    const params = { ...paramsArg, ...(typeof paramsArg.threadId === 'string' ? { threadId: this.rawThreadId(paramsArg.threadId) } : {}) };
    const result = codexRecord(await this.supervisor.requireClient().request(methodArg, params, 30_000,
      signalArg ? AbortSignal.any([signalArg, this.supervisor.signal]) : this.supervisor.signal));
    if (result.thread !== undefined) result.thread = this.publicThread(result.thread);
    if (methodArg === 'thread/list') result.data = codexArray(result.data, 100).map(value => this.publicThread(value));
    if (methodArg === 'thread/loaded/list') result.data = codexArray(result.data, 100).map(value => this.publicThreadId(value));
    return result;
  }

  public register(projectIdArg: string, directoryArg: string, sessionArg: IControllerSession): IControllerSession {
    const id = codexSessionId(sessionArg.id.nativeId);
    this.rawThreadId(id);
    const previous = this.scopes.get(id);
    if (previous && (previous.projectId !== projectIdArg || previous.directory !== directoryArg || previous.session.createdAt !== sessionArg.createdAt)) throw new Error('Codex thread scope changed.');
    if (!previous && this.scopes.size >= 4096) throw new Error('Codex session capacity reached.');
    // Admission may hold its DTO across asynchronous persistence. Existing
    // metadata belongs to this adapter, never to that caller-held snapshot.
    if (previous) return structuredClone(previous.session);
    this.scopes.set(id, { projectId: projectIdArg, directory: directoryArg, session: structuredClone(sessionArg), revision: 0, statusRevision: 0, closedRevision: 0 });
    return structuredClone(sessionArg);
  }

  private applySessionSnapshot(sessionArg: IControllerSession, observedArg: { scope: ICodexSessionScope; revision: number } | undefined): IControllerSession {
    const id = sessionArg.id.nativeId;
    const current = this.scopes.get(id);
    if (!current) return structuredClone(sessionArg);
    if (current.session.createdAt !== sessionArg.createdAt) throw new Error('Codex snapshot changed thread identity.');
    if (observedArg?.scope === current && observedArg.revision === current.revision && !this.loaded.has(id) && !this.loading.has(id)) {
      current.session = structuredClone(sessionArg);
      current.revision += 1;
      current.statusRevision += 1;
    }
    return structuredClone(current.session);
  }

  private scope(threadIdArg: string): ICodexSessionScope {
    this.supervisor.requireClient();
    const scope = this.scopes.get(threadIdArg);
    if (!scope) throw new Error('Codex thread has no managed AGL admission.');
    return scope;
  }

  public async readSession(directoryArg: string, threadIdArg: string, signalArg?: AbortSignal): Promise<IControllerSession> {
    codexSessionId(threadIdArg);
    const loaded = this.scopes.get(threadIdArg);
    // Loaded metadata is owned by the subscription. First-turn disk metadata can
    // still be unwritten while native thread/status notifications are current.
    if (loaded && this.loaded.has(threadIdArg)) {
      this.scope(threadIdArg);
      if (loaded.directory !== directoryArg) throw new Error('Codex thread belongs to another project directory.');
      signalArg?.throwIfAborted();
      return structuredClone(loaded.session);
    }
    const observed = loaded ? { scope: loaded, revision: loaded.revision } : undefined;
    const result = await this.request('thread/read', { threadId: threadIdArg, includeTurns: false }, signalArg);
    const session = codexSession(result.thread, directoryArg);
    const archived = (await this.listThreadState(directoryArg, true, signalArg)).get(threadIdArg);
    if (archived && archived.createdAt !== session.createdAt) throw new Error('Codex archive metadata changed thread identity.');
    return this.applySessionSnapshot(archived ? { ...session, archivedAt: archived.archivedAt } : session, observed);
  }

  private async listThreadState(directoryArg: string, archivedArg: boolean, signalArg?: AbortSignal): Promise<Map<string, IControllerSession>> {
    const sessions = new Map<string, IControllerSession>();
    const cursors = new Set<string>();
    let cursor: string | undefined;
    do {
      const result = await this.request('thread/list', { cwd: directoryArg, modelProviders: [], sourceKinds: ['cli', 'vscode', 'appServer', 'exec', 'unknown'], archived: archivedArg, useStateDbOnly: true, limit: 100, ...(cursor ? { cursor } : {}) }, signalArg);
      for (const value of codexArray(result.data, 100)) {
        const session = codexSession(value, directoryArg);
        if (archivedArg) session.archivedAt = session.updatedAt;
        sessions.set(session.id.nativeId, session);
      }
      if (sessions.size > 4096) throw new Error('Codex session listing exceeds its capacity.');
      cursor = result.nextCursor === null ? undefined : codexString(result.nextCursor);
      if (cursor && cursors.has(cursor)) throw new Error('Codex session pagination did not advance.');
      if (cursor) cursors.add(cursor);
    } while (cursor);
    return sessions;
  }

  public async listSessions(directoryArg: string, signalArg?: AbortSignal): Promise<IControllerSession[]> {
    const observed = new Map([...this.scopes].filter(([, scope]) => scope.directory === directoryArg).map(([id, scope]) => [id, { scope, revision: scope.revision }]));
    const active = await this.listThreadState(directoryArg, false, signalArg);
    const archived = await this.listThreadState(directoryArg, true, signalArg);
    const sessions = new Map([...active, ...archived]);
    // A native client's empty conversation can exist only in server memory.
    // Read metadata for loaded IDs absent from this folder's persisted listing;
    // no resume/subscription or writer takeover is needed for discovery.
    const cursors = new Set<string>();
    const loadedIds = new Set<string>();
    let cursor: string | undefined;
    do {
      const page = await this.request('thread/loaded/list', { limit: 100, ...(cursor ? { cursor } : {}) }, signalArg);
      const ids = codexArray(page.data, 100).map(value => codexSessionId(value));
      for (const id of ids) {
        if (loadedIds.has(id)) throw new Error('Codex loaded session listing returned a duplicate.');
        loadedIds.add(id);
      }
      if (loadedIds.size > 4096) throw new Error('Codex loaded session listing exceeds its capacity.');
      for (let index = 0; index < ids.length; index += 8) {
        const missing = ids.slice(index, index + 8).filter(id => !sessions.has(id) && !this.scopes.has(id));
        const snapshots = await Promise.all(missing.map(async id => {
          const result = await this.request('thread/read', { threadId: id, includeTurns: false }, signalArg);
          const thread = codexRecord(result.thread);
          if (thread.id !== id) throw new Error('Codex returned a different loaded conversation.');
          return thread.cwd === directoryArg ? codexSession(thread, directoryArg) : undefined;
        }));
        for (const session of snapshots) if (session && !session.parentId) sessions.set(session.id.nativeId, session);
      }
      cursor = page.nextCursor === null ? undefined : codexString(page.nextCursor);
      if (cursor && (cursors.has(cursor) || ids.length === 0)) throw new Error('Codex loaded session pagination did not advance.');
      if (cursor) cursors.add(cursor);
    } while (cursor);
    if (sessions.size > 4096) throw new Error('Codex session listing exceeds its capacity.');
    for (const [id, session] of sessions) sessions.set(id, this.applySessionSnapshot(session, observed.get(id)));
    // New threads are live subscription objects until their first turn persists.
    for (const [id, scope] of this.scopes) {
      if (scope.directory === directoryArg && (loadedIds.has(id) || this.loaded.has(id) || this.loading.has(id))) sessions.set(id, structuredClone(scope.session));
    }
    return [...sessions.values()];
  }

  public async listModels(signalArg?: AbortSignal): Promise<IControllerCodexModelOption[]> {
    const models: IControllerCodexModelOption[] = [];
    const cursors = new Set<string>();
    let cursor: string | undefined;
    do {
      const result = await this.request('model/list', { limit: 100, ...(cursor ? { cursor } : {}) }, signalArg);
      for (const value of codexArray(result.data, 100)) {
        const model = codexRecord(value);
        if (model.hidden === true) continue;
        models.push({ harnessId: 'codex', providerID: 'codex', providerName: 'Codex', modelID: codexString(model.model), modelName: codexString(model.displayName),
          variants: codexArray(model.supportedReasoningEfforts, 32).map((value) => codexString(codexRecord(value).reasoningEffort, 128)) });
      }
      if (models.length > 1024) throw new Error('Codex model catalog exceeds its capacity.');
      cursor = result.nextCursor === null ? undefined : codexString(result.nextCursor);
      if (cursor && cursors.has(cursor)) throw new Error('Codex model pagination did not advance.');
      if (cursor) cursors.add(cursor);
    } while (cursor);
    return models;
  }

  public async validateModel(modelArg: IControllerCodexModelChoice, signalArg?: AbortSignal): Promise<void> {
    const model = (await this.listModels(signalArg)).find((candidate) => candidate.modelID === modelArg.modelID);
    if (!model || (modelArg.variant !== undefined && !model.variants.includes(modelArg.variant))) throw new Error('The selected Codex model or reasoning effort is unavailable.');
  }

  public async ensureLoaded(threadIdArg: string, signalArg?: AbortSignal): Promise<void> {
    this.scope(threadIdArg);
    if (this.released.has(threadIdArg)) throw new Error('This conversation was released to another client. Choose Resume in AGL to join again.');
    if (this.scope(threadIdArg).session.archivedAt !== undefined) throw new Error('An archived Codex conversation cannot run a turn.');
    if (this.loaded.has(threadIdArg)) return;
    let pending = this.loading.get(threadIdArg);
    if (!pending) {
      if (this.loaded.size + this.loading.size >= 128) throw new Error('Codex loaded conversation capacity reached.');
      const observedScope = this.scope(threadIdArg);
      const observedRevision = observedScope.revision;
      pending = this.request('thread/resume', { threadId: threadIdArg, excludeTurns: true }, signalArg).then((result) => {
        const scope = this.scope(threadIdArg);
        const session = codexSession(result.thread, scope.directory);
        if (scope !== observedScope || scope.closedRevision > observedRevision || scope.session.archivedAt !== undefined) throw new Error('Codex subscription closed while resume was pending.');
        if (session.id.nativeId !== threadIdArg || session.createdAt !== scope.session.createdAt) throw new Error('Codex resumed a different thread identity.');
        if (scope.revision === observedRevision) { scope.session = session; scope.revision += 1; scope.statusRevision += 1; }
        this.loaded.add(threadIdArg);
        this.observeSettings(threadIdArg, result);
      }).finally(() => this.loading.delete(threadIdArg));
      this.loading.set(threadIdArg, pending);
    }
    await pending;
  }

  public markCreatedLoaded(threadIdArg: string): void { this.scope(threadIdArg); this.loaded.add(threadIdArg); this.emptyThreads.add(threadIdArg); }
  public markForkedLoaded(threadIdArg: string): void { this.scope(threadIdArg); this.loaded.add(threadIdArg); this.emptyThreads.delete(threadIdArg); }

  public async listMessagePage(threadIdArg: string, optionsArg: { limit?: number; before?: string } = {}, signalArg?: AbortSignal): Promise<IControllerMessagePage> {
    if (this.scope(threadIdArg).session.archivedAt === undefined && !this.released.has(threadIdArg)) await this.ensureLoaded(threadIdArg, signalArg);
    if (this.emptyThreads.has(threadIdArg)) {
      // No older history exists for this newly created runtime object. Its first
      // live items arrive before the paginated rollout becomes readable.
      return { bundles: [...(this.liveItems.get(threadIdArg)?.values() ?? [])].map((entry) => codexItemBundle(entry.item, entry.turnId, entry.completed, entry.terminalTurnStatus)) };
    }
    const limit = Math.min(200, Math.max(1, optionsArg.limit ?? 20));
    const result = await this.request('thread/items/list', { threadId: threadIdArg, limit, sortDirection: 'desc', ...(optionsArg.before ? { cursor: optionsArg.before } : {}) }, signalArg);
    const entries = codexArray(result.data, limit).map(codexRecord).reverse();
    const bundles = entries.map((entry) => {
      const item = codexRecord(entry.item);
      const id = codexString(item.id, 512);
      // The rollout is authoritative the moment it settles an item. Preferring the live snapshot
      // there would pin the item to running for good whenever AGL missed its completion.
      if (isCodexTerminalItemStatus(item.status)) {
        this.settleLiveItem(threadIdArg, id);
        return codexItemBundle(item, codexThreadId(entry.turnId));
      }
      const live = this.liveItems.get(threadIdArg)?.get(id);
      return codexItemBundle(live?.item ?? item, codexThreadId(entry.turnId), live?.completed ?? true, live?.terminalTurnStatus);
    });
    const ids = new Set(bundles.map((bundle) => bundle.sourceMessageId.nativeId));
    if (!optionsArg.before) {
      const liveEntries = [...(this.liveItems.get(threadIdArg)?.entries() ?? [])];
      // Only an item the page does not carry can still claim to be running, so the active turn is
      // read only when there is such a claim — an idle conversation never pays for it, and an
      // unasked question stays unknown.
      const activeTurn: TCodexActiveTurn = liveEntries.some(([id, live]) => !live.completed && !ids.has(id))
        ? await this.readActiveTurn(threadIdArg, signalArg)
        : { state: 'unknown' };
      for (const [liveIndex, [id, live]] of liveEntries.entries()) {
        if (ids.has(id) || live.completed) continue;
        // An item can only be in progress inside the conversation's active turn. One left over
        // from a turn that has ended was settled natively while AGL was not listening; showing it
        // as running would put a card on the newest page that nothing can ever close. It belongs
        // to the older page that carries it, where the rollout states its outcome. An unknown
        // active turn settles nothing: a command that is genuinely running must never be closed on
        // the strength of a read that did not answer, so the page hydrates as it did before this
        // evidence existed — spliced in as running, and healed by a later read that does answer.
        if (activeTurn.state === 'none' || (activeTurn.state === 'inProgress' && live.turnId !== activeTurn.turnId)) {
          this.settleLiveItem(threadIdArg, id); continue;
        }
        const bundle = codexItemBundle(live.item, live.turnId, false);
        const nextKnownId = liveEntries.slice(liveIndex + 1)
          .map(([candidateId]) => candidateId)
          .find(candidateId => ids.has(candidateId));
        if (nextKnownId) {
          const nextIndex = bundles.findIndex(candidate => candidate.sourceMessageId.nativeId === nextKnownId);
          bundles.splice(nextIndex, 0, bundle);
        } else {
          const previousKnownId = liveEntries.slice(0, liveIndex).reverse()
            .map(([candidateId]) => candidateId)
            .find(candidateId => ids.has(candidateId));
          const previousIndex = previousKnownId === undefined
            ? bundles.length - 1
            : bundles.findIndex(candidate => candidate.sourceMessageId.nativeId === previousKnownId);
          bundles.splice(previousIndex + 1, 0, bundle);
        }
        ids.add(id);
      }
    }
    return { bundles, ...(result.nextCursor === null ? {} : { nextCursor: codexString(result.nextCursor) }) };
  }

  public async getMessage(threadIdArg: string, itemIdArg: string, signalArg?: AbortSignal): Promise<IControllerMessageBundle> {
    let before: string | undefined;
    const seen = new Set<string>();
    for (let page = 0; page < 128; page += 1) {
      const result = await this.listMessagePage(threadIdArg, { limit: 200, before }, signalArg);
      const bundle = result.bundles.find((entry) => entry.sourceMessageId.nativeId === itemIdArg);
      if (bundle) return bundle;
      before = result.nextCursor;
      if (!before) break;
      if (seen.has(before)) throw new Error('Codex item pagination did not advance.');
      seen.add(before);
    }
    throw new Error('Codex transcript item was not found within the history limit.');
  }

  public pendingPermissionScopes(): Array<{ projectId: string; sessionId: IControllerSession['id'] }> {
    const result = new Map<string, { projectId: string; sessionId: IControllerSession['id'] }>();
    for (const pending of this.pending.values()) {
      if (!pending.permission) continue;
      const threadId = pending.permission.sessionId.nativeId;
      const scope = this.scopes.get(threadId);
      if (scope) result.set(threadId, { projectId: scope.projectId, sessionId: codexRuntimeId(threadId) });
    }
    return [...result.values()];
  }

  public attention(threadIdArg: string): { permissions: IControllerPermission[]; questions: IControllerQuestion[] } {
    this.scope(threadIdArg);
    const requests = [...this.pending.values()].filter((entry) => entry.request.params.threadId === threadIdArg);
    return { permissions: requests.flatMap((entry) => entry.permission ? [entry.permission] : []), questions: requests.flatMap((entry) => entry.question ? [entry.question] : []) };
  }

  public async startTurn(threadIdArg: string, textArg: string, clientMessageIdArg: string, modelArg: IControllerCodexModelChoice | undefined, beforeDispatchArg: () => Promise<void>, completeArg: () => void, signalArg?: AbortSignal, onUserMessageArg?: (idArg: string) => void, cancelUndispatchedArg?: () => Promise<void>, collaborationModeArg?: ICodexPreparedCollaborationMode): Promise<() => void> {
    this.scope(threadIdArg);
    this.assertModeDependentDispatchAllowed(threadIdArg, collaborationModeArg !== undefined);
    if (this.turns.has(threadIdArg)) throw new Error('A Codex turn is already active or its outcome is unknown.');
    const admission: ICodexTurnAdmission = { clientMessageId: clientMessageIdArg, onUserMessage: onUserMessageArg, terminal: new Set(), armed: false, complete: completeArg };
    this.turns.set(threadIdArg, admission);
    let dispatched = false;
    try {
      await this.ensureLoaded(threadIdArg, signalArg);
      const session = await this.readSession(this.scope(threadIdArg).directory, threadIdArg, signalArg);
      if (session.status === 'busy') throw new Error('A Codex turn is already running. Wait for it to finish before sending from AGL.');
      if (modelArg) await this.validateModel(modelArg, signalArg);
      await this.supervisor.observeOwnedMembers();
      await beforeDispatchArg();
      signalArg?.throwIfAborted();
      this.supervisor.signal.throwIfAborted();
      this.assertModeDependentDispatchAllowed(threadIdArg, collaborationModeArg !== undefined);
      dispatched = true;
      const scope = this.scope(threadIdArg);
      const dispatchRevision = scope.statusRevision;
      const result = await this.request('turn/start', { threadId: threadIdArg, clientUserMessageId: clientMessageIdArg,
        input: [{ type: 'text', text: textArg, text_elements: [] }],
        ...(collaborationModeArg ? { collaborationMode: collaborationModeArg.native }
          : modelArg ? { model: modelArg.modelID, ...(modelArg.variant ? { effort: modelArg.variant } : {}) } : {}) }, signalArg);
      const turn = codexRecord(result.turn);
      admission.turnId = codexThreadId(turn.id);
      const activity = this.activities.get(threadIdArg);
      const sameTurn = activity?.turnId === admission.turnId;
      this.activities.set(threadIdArg, {
        ...activity, writer: 'agl', turnId: admission.turnId,
        ...(!sameTurn ? { reroute: undefined, plan: undefined, diff: undefined,
          diffTruncated: undefined } : {}),
        ...(collaborationModeArg ? {
          model: collaborationModeArg.public.model,
          effort: collaborationModeArg.public.effort,
        } : modelArg ? { model: sameTurn && activity.reroute ? activity.reroute.toModel : modelArg.modelID, effort: modelArg.variant } : {}),
      });
      if (turn.status === 'inProgress') this.interruptibleTurns.set(threadIdArg, admission.turnId);
      else this.interruptibleTurns.delete(threadIdArg);
      if (turn.status !== 'inProgress') {
        this.completeTurnItems(threadIdArg, admission.turnId, turn.status);
        admission.terminal.add(admission.turnId); this.emptyThreads.delete(threadIdArg);
      }
      if (scope.statusRevision === dispatchRevision) {
        scope.session.status = admission.terminal.has(admission.turnId) ? 'idle' : 'busy';
        scope.revision += 1;
        scope.statusRevision += 1;
      }
      return () => { admission.armed = true; this.completeTurn(threadIdArg, admission); };
    } catch (error) {
      if (!dispatched || (error instanceof plugins.crossharness.CodexAppServerRequestError && !error.dispatched)) {
        // Keep the admission lock until the durable first-turn marker is cleared.
        await cancelUndispatchedArg?.();
        this.turns.delete(threadIdArg);
        if (!(error instanceof plugins.crossharness.CodexAppServerRequestError)) throw new plugins.crossharness.CodexAppServerRequestError(error instanceof Error ? error.message : String(error), false, undefined, { cause: error });
      }
      throw error;
    }
  }

  /** Starts the native inline review operation and owns only the exact turn returned by Codex. */
  public async startReview(
    threadIdArg: string,
    instructionsArg: string | undefined,
    admissionIdArg: string,
    beforeDispatchArg: () => Promise<void>,
    completeArg: () => void,
    signalArg?: AbortSignal,
    cancelUndispatchedArg?: () => Promise<void>,
  ): Promise<() => void> {
    this.scope(threadIdArg);
    if (this.turns.has(threadIdArg)) throw new Error('A Codex turn is already active or its outcome is unknown.');
    const admission: ICodexTurnAdmission = {
      clientMessageId: admissionIdArg,
      terminal: new Set(),
      armed: false,
      complete: completeArg,
    };
    this.turns.set(threadIdArg, admission);
    let dispatched = false;
    try {
      await this.ensureLoaded(threadIdArg, signalArg);
      const scope = this.scope(threadIdArg);
      const session = await this.readSession(scope.directory, threadIdArg, signalArg);
      if (session.status === 'busy') throw new Error('A Codex turn is already running. Wait for it to finish before starting a review.');
      await this.supervisor.observeOwnedMembers();
      await beforeDispatchArg();
      signalArg?.throwIfAborted();
      this.supervisor.signal.throwIfAborted();
      dispatched = true;
      const dispatchRevision = scope.statusRevision;
      const result = await this.request('review/start', {
        threadId: threadIdArg,
        target: instructionsArg === undefined
          ? { type: 'uncommittedChanges' }
          : { type: 'custom', instructions: instructionsArg },
        delivery: 'inline',
      }, signalArg);
      const turn = codexRecord(result.turn);
      admission.turnId = codexThreadId(turn.id);
      const activity = this.activities.get(threadIdArg);
      this.activities.set(threadIdArg, {
        ...activity,
        writer: 'agl',
        turnId: admission.turnId,
        reroute: undefined,
        plan: undefined,
        diff: undefined,
        diffTruncated: undefined,
      });
      if (turn.status === 'inProgress') this.interruptibleTurns.set(threadIdArg, admission.turnId);
      else this.interruptibleTurns.delete(threadIdArg);
      if (turn.status !== 'inProgress') {
        this.completeTurnItems(threadIdArg, admission.turnId, turn.status);
        admission.terminal.add(admission.turnId);
        this.emptyThreads.delete(threadIdArg);
      }
      if (scope.statusRevision === dispatchRevision) {
        scope.session.status = admission.terminal.has(admission.turnId) ? 'idle' : 'busy';
        scope.revision += 1;
        scope.statusRevision += 1;
      }
      return () => {
        admission.armed = true;
        this.completeTurn(threadIdArg, admission);
      };
    } catch (error) {
      if (!dispatched || (error instanceof plugins.crossharness.CodexAppServerRequestError && !error.dispatched)) {
        await cancelUndispatchedArg?.();
        this.turns.delete(threadIdArg);
        if (!(error instanceof plugins.crossharness.CodexAppServerRequestError)) {
          throw new plugins.crossharness.CodexAppServerRequestError(
            error instanceof Error ? error.message : String(error),
            false,
            undefined,
            { cause: error },
          );
        }
      }
      throw error;
    }
  }

  private completeTurn(threadIdArg: string, admissionArg: ICodexTurnAdmission): void {
    if (!admissionArg.armed || !admissionArg.turnId || !admissionArg.terminal.has(admissionArg.turnId)) return;
    if (this.turns.get(threadIdArg) !== admissionArg) return;
    this.turns.delete(threadIdArg);
    admissionArg.complete();
  }

  /** Observe only the exact dispatched input after its caller has fenced the unknown outcome. */
  public observeSubmissionOutcomeUnknown(threadIdArg: string, clientMessageIdArg: string): void {
    const admission = this.turns.get(threadIdArg);
    if (admission?.clientMessageId === clientMessageIdArg) {
      admission.armed = true;
      this.completeTurn(threadIdArg, admission);
    }
    this.armSteeredInput(threadIdArg, clientMessageIdArg);
  }

  private armSteeredInput(threadIdArg: string, clientMessageIdArg: string): void {
    const entries = this.steeredInputs.get(threadIdArg);
    const input = entries?.get(clientMessageIdArg);
    if (!input) return;
    input.armed = true;
    if (input.terminal) { entries!.delete(clientMessageIdArg); input.complete(); }
  }

  /** Uses the exact acknowledged active turn. A lost response is never retried. */
  public async steer(threadIdArg: string, textArg: string, clientMessageIdArg: string,
    onUserMessageArg: (idArg: string) => void, completeArg: () => void,
    expectedTurnIdArg?: string, signalArg?: AbortSignal): Promise<() => void> {
    this.scope(threadIdArg);
    this.assertModeDependentDispatchAllowed(threadIdArg);
    const admission = this.turns.get(threadIdArg);
    if (!admission?.armed || !admission.turnId || admission.terminal.has(admission.turnId)) {
      throw new plugins.crossharness.CodexAppServerRequestError('There is no acknowledged AGL turn to steer. Use Send or Queue next.', false);
    }
    if (expectedTurnIdArg !== undefined && expectedTurnIdArg !== admission.turnId) {
      throw new plugins.crossharness.CodexAppServerRequestError(
        'The active Codex turn changed before the steering input was submitted.', false,
      );
    }
    let entries = this.steeredInputs.get(threadIdArg);
    if (!entries) { entries = new Map(); this.steeredInputs.set(threadIdArg, entries); }
    if (entries.size >= 16 || entries.has(clientMessageIdArg)) throw new plugins.crossharness.CodexAppServerRequestError('Codex steering capacity reached.', false);
    const input: ICodexSteeredInput = { clientMessageId: clientMessageIdArg, turnId: admission.turnId,
      armed: false, terminal: false, onUserMessage: onUserMessageArg, complete: completeArg };
    entries.set(clientMessageIdArg, input);
    try {
      const result = await this.request('turn/steer', { threadId: threadIdArg, expectedTurnId: admission.turnId,
        clientUserMessageId: clientMessageIdArg, input: [{ type: 'text', text: textArg, text_elements: [] }] }, signalArg);
      if (result.turnId !== admission.turnId) throw new Error('Codex steering returned another turn identity.');
      return () => this.armSteeredInput(threadIdArg, clientMessageIdArg);
    } catch (error) {
      if (error instanceof plugins.crossharness.CodexAppServerRequestError) {
        if (!error.dispatched) entries.delete(clientMessageIdArg);
        else if (this.isDefinitiveSteerRejection(error)) {
          entries.delete(clientMessageIdArg);
          throw new plugins.crossharness.CodexAppServerRequestError(
            'Codex did not accept the steering input because its active turn changed or no longer permits steering.',
            false,
            error.code,
            { cause: error },
          );
        }
      }
      throw error;
    }
  }

  /** Native validation errors are returned before Codex enqueues the steering input. */
  private isDefinitiveSteerRejection(errorArg: plugins.crossharness.CodexAppServerRequestError): boolean {
    return [
      'no active turn to steer',
      'expected active turn id',
      'cannot steer a review turn',
      'cannot steer a compact turn',
      'input must not be empty',
      'active turn uses a different output schema',
    ].some((message) => errorArg.message.includes(message));
  }

  public async releaseWriter(threadIdArg: string, signalArg?: AbortSignal): Promise<void> {
    this.scope(threadIdArg);
    if (this.turns.has(threadIdArg) || this.loading.has(threadIdArg)
      || [...this.pending.values()].some((entry) => entry.request.params.threadId === threadIdArg)) {
      throw new Error('Stop the AGL turn and resolve its pending requests before releasing this conversation.');
    }
    // The fence precedes dispatch, so an uncertain unsubscribe cannot implicitly rejoin.
    this.released.add(threadIdArg);
    const result = await this.request('thread/unsubscribe', { threadId: threadIdArg }, signalArg);
    if (!['notLoaded', 'notSubscribed', 'unsubscribed'].includes(String(result.status))) throw new Error('Codex did not confirm conversation release.');
    this.loaded.delete(threadIdArg);
    // An AGL turn is fenced out above, so nothing settled by `turn/completed` can be lost here.
    if (this.dropLiveItems(threadIdArg)) this.changed(threadIdArg, 'session.history.changed');
    this.changed(threadIdArg, 'session.changed');
  }

  public async resumeWriter(threadIdArg: string, signalArg?: AbortSignal): Promise<void> {
    this.scope(threadIdArg); this.released.delete(threadIdArg);
    try { await this.ensureLoaded(threadIdArg, signalArg); }
    catch (error) { this.released.add(threadIdArg); throw error; }
    this.changed(threadIdArg, 'session.changed');
  }

  /**
   * Marks one live entry settled once the rollout has proven it settled. The entry keeps its place
   * so the existing capacity eviction releases it in insertion order, and the completion guard in
   * `updateItem` stops a later replay from reopening it.
   */
  private settleLiveItem(threadIdArg: string, itemIdArg: string): void {
    const live = this.liveItems.get(threadIdArg)?.get(itemIdArg);
    if (live) live.completed = true;
  }

  /**
   * Drops the live entries of a conversation whose notification stream AGL has left for good. A
   * conversation AGL created is served from that snapshot alone until its rollout becomes readable;
   * once the snapshot is gone the rollout is by definition its only remaining source, so the thread
   * stops being treated as empty here rather than serving a blank transcript.
   */
  private dropLiveItems(threadIdArg: string): boolean {
    this.emptyThreads.delete(threadIdArg);
    const items = this.liveItems.get(threadIdArg);
    if (!items) return false;
    for (const item of items.values()) this.liveBytes -= item.bytes;
    this.liveItems.delete(threadIdArg);
    return items.size > 0;
  }

  private completeTurnItems(threadIdArg: string, turnIdArg: string, statusArg: unknown): void {
    if (statusArg !== 'completed' && statusArg !== 'interrupted' && statusArg !== 'failed') throw new Error('Invalid Codex terminal turn status.');
    // Codex can end a turn without item/completed notifications for in-flight
    // commands or partial text. The terminal turn closes those exact streams.
    for (const live of this.liveItems.get(threadIdArg)?.values() ?? []) {
      if (live.turnId !== turnIdArg || live.completed) continue;
      this.updateItem(threadIdArg, turnIdArg, live.item, true, statusArg);
    }
    for (const [id, input] of this.steeredInputs.get(threadIdArg) ?? []) {
      if (input.turnId !== turnIdArg) continue;
      input.terminal = true;
      if (input.armed) { this.steeredInputs.get(threadIdArg)!.delete(id); input.complete(); }
    }
  }

  public async abort(threadIdArg: string, expectedTurnIdArg?: string,
    signalArg?: AbortSignal): Promise<{ turnId: string; writer: 'agl' | 'external' }> {
    const activity = await this.refreshActivity(threadIdArg, signalArg);
    if (!activity.canInterrupt || !activity.turnId
      || (activity.writer !== 'agl' && activity.writer !== 'external')) {
      throw new Error('Codex has no exact interruptible turn on this connection.');
    }
    if (activity.writer === 'external' && expectedTurnIdArg === undefined) {
      throw new Error('Stopping an externally started Codex turn requires its observed turn ID.');
    }
    const expectedTurnId = expectedTurnIdArg ?? activity.turnId;
    if (expectedTurnId !== activity.turnId) {
      throw new Error('The active Codex turn changed before it could be stopped.');
    }
    await this.request('turn/interrupt', { threadId: threadIdArg, turnId: expectedTurnId }, signalArg);
    return { turnId: expectedTurnId, writer: activity.writer };
  }

  public async waitForTurnTerminal(threadIdArg: string, turnIdArg: string,
    signalArg: AbortSignal): Promise<void> {
    while (true) {
      signalArg.throwIfAborted();
      const result = await this.request('thread/turns/list', {
        threadId: threadIdArg, limit: 1, sortDirection: 'desc', itemsView: 'notLoaded',
      }, signalArg);
      const latest = codexArray(result.data, 1)[0];
      if (!latest || codexRecord(latest).id !== turnIdArg
        || codexRecord(latest).status !== 'inProgress') return;
      await new Promise<void>((resolve, reject) => {
        const abort = () => { clearTimeout(timer); reject(signalArg.reason); };
        const timer = setTimeout(() => { signalArg.removeEventListener('abort', abort); resolve(); }, 250);
        signalArg.addEventListener('abort', abort, { once: true });
        if (signalArg.aborted) abort();
      });
    }
  }

  public async waitForIdle(threadIdArg: string, signalArg: AbortSignal): Promise<void> {
    while (true) {
      signalArg.throwIfAborted();
      const scope = this.scope(threadIdArg);
      const session = await this.readSession(scope.directory, threadIdArg, signalArg);
      if (session.status !== 'busy') {
        const admission = this.turns.get(threadIdArg);
        if (admission?.turnId) {
          const result = await this.request('thread/turns/list', { threadId: threadIdArg, limit: 1, sortDirection: 'desc', itemsView: 'notLoaded' }, signalArg);
          const latest = codexArray(result.data, 1)[0];
          if (latest && codexRecord(latest).id === admission.turnId && codexRecord(latest).status !== 'inProgress') {
            this.completeTurnItems(threadIdArg, admission.turnId, codexRecord(latest).status);
            admission.terminal.add(admission.turnId);
            this.completeTurn(threadIdArg, admission);
          }
        }
        return;
      }
      await new Promise<void>((resolve, reject) => {
        const abort = () => { clearTimeout(timer); reject(signalArg.reason); };
        const timer = setTimeout(() => { signalArg.removeEventListener('abort', abort); resolve(); }, 250);
        signalArg.addEventListener('abort', abort, { once: true });
        if (signalArg.aborted) abort();
      });
    }
  }

  public async rename(threadIdArg: string, titleArg: string, signalArg?: AbortSignal): Promise<void> {
    this.scope(threadIdArg);
    await this.request('thread/name/set', { threadId: threadIdArg, name: titleArg }, signalArg);
    this.scope(threadIdArg).session.title = titleArg;
    this.scope(threadIdArg).revision += 1;
  }

  public async readIfPresent(directoryArg: string, threadIdArg: string, signalArg?: AbortSignal): Promise<IControllerSession | undefined> {
    try { return await this.readSession(directoryArg, threadIdArg, signalArg); }
    catch (error) {
      if (error instanceof plugins.crossharness.CodexAppServerRequestError && error.code === -32600
        && /thread (not loaded|not found):|no rollout found for thread/i.test(error.message)) return undefined;
      throw error;
    }
  }

  public forget(threadIdArg: string): void {
    this.dropLiveItems(threadIdArg);
    this.scopes.delete(threadIdArg); this.loaded.delete(threadIdArg);
    this.turns.delete(threadIdArg);
    this.activities.delete(threadIdArg); this.interruptibleTurns.delete(threadIdArg);
    this.released.delete(threadIdArg); this.steeredInputs.delete(threadIdArg); this.modeWaiters.delete(threadIdArg);
    this.modeUpdates.delete(threadIdArg); this.modeAuthorities.delete(threadIdArg);
  }

  public async deleteThread(directoryArg: string, threadIdArg: string, signalArg?: AbortSignal): Promise<void> {
    await this.supervisor.observeOwnedMembers();
    await this.request('thread/delete', { threadId: codexSessionId(threadIdArg) }, signalArg);
    this.loaded.delete(threadIdArg);
    if (await this.readIfPresent(directoryArg, threadIdArg, signalArg)) throw new Error('Codex deletion did not remove the thread.');
    this.forget(threadIdArg);
  }

  public async archive(threadIdArg: string, signalArg?: AbortSignal): Promise<IControllerSession> {
    const scope = this.scope(threadIdArg);
    await this.supervisor.observeOwnedMembers();
    let mutationError: unknown;
    try { await this.request('thread/archive', { threadId: threadIdArg }, signalArg); }
    catch (error) { mutationError = error; }
    let archived: IControllerSession | undefined;
    try {
      archived = (await this.listThreadState(scope.directory, true, this.supervisor.signal)).get(threadIdArg);
      if (!archived || archived.createdAt !== scope.session.createdAt) throw mutationError ?? new Error('Codex archive was not confirmed by the server.');
    } catch (error) {
      if (!this.supervisor.signal.aborted) this.supervisor.requireClient().close('Codex archive outcome is uncertain.');
      throw error;
    }
    this.forget(threadIdArg);
    return archived;
  }

  public onNotification(notificationArg: plugins.crossharness.ICodexAppServerNotification): void {
    if (this.supervisor.signal.aborted) return;
    let params = notificationArg.params;
    // Native creation is announced with a thread object, before AGL has a scope.
    if (notificationArg.method === 'thread/started') {
      try { this.publicThread(params.thread); } catch { return; }
      this.emit({ type: 'sessions.changed', harnessId: 'codex', timestamp: Date.now() });
      return;
    }
    if (typeof params.threadId === 'string') {
      try { params = { ...params, threadId: this.publicThreadId(params.threadId) }; } catch { return; }
    }
    const threadId = params.threadId;
    if (typeof threadId !== 'string') return;
    if (!this.scopes.has(threadId)) {
      if (['thread/name/updated', 'thread/archived', 'thread/unarchived'].includes(notificationArg.method)) {
        this.emit({ type: 'sessions.changed', harnessId: 'codex', timestamp: Date.now() });
      }
      return;
    }
    try {
      const method = notificationArg.method;
      if (method === 'turn/completed') {
        const turn = codexRecord(params.turn);
        const turnId = codexThreadId(turn.id);
        if (this.interruptibleTurns.get(threadId) === turnId) this.interruptibleTurns.delete(threadId);
        this.completeTurnItems(threadId, turnId, turn.status);
        this.emptyThreads.delete(threadId);
        const admission = this.turns.get(threadId);
        if (this.activities.get(threadId)?.turnId === turnId || admission?.turnId === turnId) {
          const scope = this.scope(threadId);
          scope.session.status = turn.status === 'failed' ? 'error' : 'idle';
          scope.revision += 1; scope.statusRevision += 1;
        }
        if (admission) {
          if (admission.terminal.size >= 64) throw new Error('Codex turn correlation capacity exceeded.');
          admission.terminal.add(turnId);
          this.completeTurn(threadId, admission);
        }
        this.changed(threadId, 'session.changed');
        return;
      }
      if (method === 'turn/started') {
        const turnId = codexThreadId(codexRecord(params.turn).id);
        const previous = this.activities.get(threadId);
        if (previous?.turnId !== turnId) this.activities.set(threadId, { ...previous, writer: 'agl', turnId,
          reroute: undefined, plan: undefined, diff: undefined, diffTruncated: undefined });
        this.interruptibleTurns.set(threadId, turnId);
      }
      if (method === 'thread/settings/updated') this.observeSettings(threadId, codexRecord(params.threadSettings));
      if (method === 'turn/plan/updated' || method === 'turn/diff/updated' || method === 'model/rerouted'
        || method === 'thread/tokenUsage/updated') {
        const turnId = codexThreadId(params.turnId);
        const current = this.activities.get(threadId);
        const activeTurn = this.turns.get(threadId)?.turnId ?? current?.turnId;
        if (activeTurn && activeTurn !== turnId) return;
        const next: TCodexStoredActivity = { ...current, writer: 'agl', turnId };
        if (method === 'turn/plan/updated') next.plan = {
          ...(params.explanation === null ? {} : { explanation: codexText(params.explanation, 8192) }),
          steps: codexArray(params.plan, 128).map((value) => {
            const step = codexRecord(value);
            if (!['pending', 'inProgress', 'completed'].includes(String(step.status))) throw new Error('Invalid Codex plan state.');
            return { content: codexString(step.step, 4096), status: step.status === 'inProgress' ? 'in_progress' : step.status as 'pending' | 'completed' };
          }) };
        if (method === 'turn/diff/updated') {
          const diff = typeof params.diff === 'string' ? params.diff : '';
          next.diff = codexText(diff, 128 * 1024);
          if (Buffer.byteLength(diff, 'utf8') > 128 * 1024) next.diffTruncated = true;
          else delete next.diffTruncated;
        }
        if (method === 'model/rerouted') {
          next.reroute = { fromModel: codexString(params.fromModel, 512), toModel: codexString(params.toModel, 512), reason: codexString(params.reason, 128) };
          next.model = next.reroute.toModel;
        }
        if (method === 'thread/tokenUsage/updated') {
          const usage = codexRecord(params.tokenUsage);
          next.totalTokens = codexNonnegativeInteger(codexRecord(usage.total).totalTokens);
          if (usage.modelContextWindow !== null) next.contextWindow = codexNonnegativeInteger(usage.modelContextWindow);
        }
        this.activities.set(threadId, next);
        this.changed(threadId, 'session.changed');
        return;
      }
      if (method === 'item/started' || method === 'item/completed') {
        this.updateItem(threadId, codexThreadId(params.turnId), codexRecord(params.item), method === 'item/completed');
        return;
      }
      if (method === 'item/agentMessage/delta' || method === 'item/reasoning/summaryTextDelta' || method === 'item/reasoning/textDelta' || method === 'item/commandExecution/outputDelta') {
        const itemId = codexString(params.itemId, 512);
        const live = this.liveItems.get(threadId)?.get(itemId);
        if (!live || live.completed) { this.changed(threadId, 'session.history.changed'); return; }
        const delta = codexString(params.delta, 1024 * 1024);
        if (method === 'item/agentMessage/delta') live.item.text = codexText(String(live.item.text ?? '') + delta);
        else if (method === 'item/commandExecution/outputDelta') live.item.aggregatedOutput = codexText(String(live.item.aggregatedOutput ?? '') + delta, 48 * 1024);
        else {
          const key = method === 'item/reasoning/summaryTextDelta' ? 'summary' : 'content';
          const texts = codexArray(live.item[key], 4096).map((value) => codexText(value));
          const index = Number(params.summaryIndex ?? params.contentIndex ?? 0);
          if (!Number.isSafeInteger(index) || index < 0 || index > 1024) throw new Error('Invalid Codex reasoning index.');
          while (texts.length <= index) texts.push('');
          texts[index] = codexText(texts[index] + delta);
          live.item[key] = texts;
        }
        this.updateItem(threadId, live.turnId, live.item, false);
        return;
      }
      const scope = this.scope(threadId);
      if (method === 'thread/status/changed') {
        const status = codexString(codexRecord(params.status).type);
        if (!['active', 'idle', 'notLoaded', 'systemError'].includes(status)) throw new Error('Invalid native thread status.');
        scope.session = { ...scope.session, status: status === 'active' ? 'busy' : status === 'systemError' ? 'error' : 'idle', updatedAt: Date.now() };
        scope.revision += 1;
        scope.statusRevision += 1;
        if (status !== 'active') this.interruptibleTurns.delete(threadId);
      }
      if (method === 'thread/name/updated' && params.threadName !== undefined) { scope.session.title = codexString(params.threadName, 8192); scope.revision += 1; }
      if (method === 'thread/closed' || method === 'thread/archived') {
        this.loaded.delete(threadId); scope.revision += 1; scope.closedRevision = scope.revision;
        if (this.dropLiveItems(threadId)) this.changed(threadId, 'session.history.changed');
      }
      if (method === 'thread/archived') scope.session.archivedAt = Date.now();
      if (method === 'thread/unarchived') { delete scope.session.archivedAt; scope.revision += 1; }
      if (method.startsWith('thread/') || method === 'turn/started' || method === 'error') this.changed(threadId, 'session.changed');
    } catch { this.changed(threadId, 'session.history.changed'); }
  }

  private updateItem(threadIdArg: string, turnIdArg: string, itemArg: Record<string, unknown>, completedArg: boolean, terminalTurnStatusArg?: TCodexTerminalTurnStatus): void {
    const id = codexString(itemArg.id, 512);
    if (itemArg.type === 'userMessage' && !terminalTurnStatusArg) {
      const admission = this.turns.get(threadIdArg);
      if (admission && itemArg.clientId === admission.clientMessageId
        && (!admission.turnId || admission.turnId === turnIdArg)) {
        admission.turnId = turnIdArg;
        admission.onUserMessage?.(id);
        this.completeTurn(threadIdArg, admission);
      }
      if (typeof itemArg.clientId === 'string') this.steeredInputs.get(threadIdArg)?.get(itemArg.clientId)?.onUserMessage(id);
    }
    const bundle = codexItemBundle(itemArg, turnIdArg, completedArg, terminalTurnStatusArg);
    const bytes = Buffer.byteLength(JSON.stringify(itemArg), 'utf8');
    let items = this.liveItems.get(threadIdArg);
    if (!items) { items = new Map(); this.liveItems.set(threadIdArg, items); }
    const previous = items.get(id);
    if (previous?.completed && !completedArg) return;
    const nextBytes = this.liveBytes - (previous?.bytes ?? 0) + bytes;
    if (nextBytes > 32 * 1024 * 1024) {
      this.supervisor.requireClient().close('Codex live transcript memory limit exceeded.');
      return;
    }
    this.liveBytes = nextBytes;
    items.set(id, { turnId: turnIdArg, item: structuredClone(itemArg), completed: completedArg, bytes, ...(terminalTurnStatusArg ? { terminalTurnStatus: terminalTurnStatusArg } : {}) });
    while (items.size > 200) {
      const oldest = [...items].find(([, entry]) => entry.completed)?.[0];
      if (!oldest) { this.supervisor.requireClient().close('Codex active transcript item limit exceeded.'); return; }
      this.liveBytes -= items.get(oldest)!.bytes;
      items.delete(oldest);
    }
    const message = bundle.messages[0]!;
    const scope = this.scope(threadIdArg);
    const common = { sessionId: codexRuntimeId(threadIdArg), messageId: codexRuntimeId(id), partId: codexRuntimeId(id), sourceUpdatedAt: Date.now(), revision: ++this.clock.revision, streamEpoch: this.streamEpoch };
    const event = { projectId: scope.projectId, harnessId: 'codex' as const, sessionId: common.sessionId, timestamp: Date.now() };
    if (message.toolCall) {
      const { id: _id, name, input, output, ...tool } = message.toolCall;
      // A live output travels as a marker-free prefix plus the contract's flag: a marker written
      // into the value would never appear in the durable output, so the finished tool call could
      // never cover this snapshot and the card would stay on its first kilobytes for good.
      const boundedOutput = output === undefined ? undefined : codexBoundedText(
        typeof output === 'string' ? output : JSON.stringify(output), codexLiveToolOutputBytes,
      );
      const update: IControllerEvent = { ...event, type: 'session.tool.updated', toolExecution: { ...common, ...tool, callId: common.partId, toolName: codexText(name, 512),
        ...(input === undefined ? {} : { input: structuredClone(input) }),
        ...(boundedOutput === undefined ? {} : { output: boundedOutput.text,
          ...(boundedOutput.truncated ? { outputTruncated: true as const } : {}) }),
      } };
      // Large structured diffs use bounded transcript hydration, without corrupting their JSON shape.
      if (Buffer.byteLength(JSON.stringify(update), 'utf8') <= controllerMaxToolEventBytes) this.emit(update);
      else this.changed(threadIdArg, 'session.history.changed');
    }
    else if (message.reasoning) this.emit({ ...event, type: 'session.reasoning.updated', reasoningUpdate: { ...common, text: codexText(message.reasoning[0]!.text, 16 * 1024), status: completedArg ? 'completed' : 'running' } });
    else if (message.role === 'assistant') this.emit({ ...event, type: 'session.text.updated', textUpdate: { ...common, text: codexText(message.text, 16 * 1024), status: completedArg ? 'completed' : 'running' } });
    if (completedArg || message.role === 'user') this.changed(threadIdArg, 'session.changed');
  }

  public onServerRequest(requestArg: plugins.crossharness.ICodexAppServerRequest): void {
    const client = this.supervisor.requireClient();
    try {
      requestArg = { ...requestArg, params: { ...requestArg.params, threadId: this.publicThreadId(requestArg.params.threadId) } };
      const params = requestArg.params;
      const threadId = codexSessionId(params.threadId);
      this.scope(threadId);
      const bytes = Buffer.byteLength(JSON.stringify(params), 'utf8');
      if (bytes > 64 * 1024 || this.pending.size >= 128 || [...this.pending.values()].reduce((sum, entry) => sum + entry.bytes, bytes) > 256 * 1024) throw new Error('Codex approval capacity reached.');
      const id = plugins.crypto.randomBytes(24).toString('base64url');
      const pending: ICodexPendingRequest = { request: requestArg, bytes };
      if (['item/commandExecution/requestApproval', 'item/fileChange/requestApproval', 'item/permissions/requestApproval'].includes(requestArg.method)) {
        pending.permission = { id: codexRuntimeId(id), sessionId: codexRuntimeId(threadId), type: requestArg.method,
          title: codexText(params.reason, 4096) || 'Codex requests approval',
          patterns: typeof params.command === 'string' ? [codexText(params.command, 48 * 1024)] : [],
          metadata: { ...(params.cwd ? { cwd: params.cwd } : {}), ...(params.permissions ? { permissions: params.permissions } : {}), ...(params.grantRoot ? { grantRoot: params.grantRoot } : {}) }, createdAt: Date.now() };
      } else if (requestArg.method === 'item/tool/requestUserInput') {
        const questions = codexArray(params.questions, 16).map(codexRecord);
        if (questions.some((question) => question.isSecret === true)) throw new Error('AGL does not collect secret Codex input.');
        pending.questionIds = questions.map((question) => codexString(question.id, 512));
        pending.question = { id: codexRuntimeId(id), sessionId: codexRuntimeId(threadId), toolCallId: codexRuntimeId(codexString(params.itemId, 512)),
          questions: questions.map((question) => ({ question: codexString(question.question, 16 * 1024), header: codexString(question.header, 512),
            custom: question.isOther === true || question.options === null,
            options: question.options === null ? [] : codexArray(question.options, 64).map((option) => ({ label: codexString(codexRecord(option).label, 1024), description: codexString(codexRecord(option).description, 4096) })) })) };
      } else { client.rejectRequest(requestArg.id, -32601, 'This Codex host request is not supported by AGL.'); return; }
      if (Buffer.byteLength(JSON.stringify(pending.permission ?? pending.question), 'utf8') > 64 * 1024) throw new Error('Codex attention request exceeds its display capacity.');
      this.pending.set(id, pending);
      requestArg.signal.addEventListener('abort', () => { this.pending.delete(id); this.changed(threadId, 'permissions.changed'); }, { once: true });
      if (requestArg.signal.aborted) { this.pending.delete(id); return; }
      this.changed(threadId, 'permissions.changed');
    } catch { client.rejectRequest(requestArg.id, -32602, 'The Codex request cannot be handled in this managed AGL scope.'); }
  }

  public replyPermission(threadIdArg: string, idArg: string, replyArg: 'once' | 'reject'): void {
    this.scope(threadIdArg);
    const pending = this.pending.get(idArg);
    if (!pending?.permission || pending.request.params.threadId !== threadIdArg) throw new Error('Codex approval is no longer pending in this conversation.');
    let result: Record<string, unknown> = { decision: replyArg === 'once' ? 'accept' : 'decline' };
    if (pending.request.method === 'item/permissions/requestApproval') {
      const requested = codexRecord(pending.request.params.permissions);
      result = { scope: 'turn', permissions: replyArg === 'reject' ? {} : Object.fromEntries(['network', 'fileSystem'].filter((key) => requested[key] !== null && requested[key] !== undefined).map((key) => [key, requested[key]])) };
    }
    if (!this.supervisor.requireClient().respond(pending.request.id, result)) throw new Error('Codex approval expired before the reply.');
  }

  public replyQuestion(threadIdArg: string, idArg: string, answersArg: string[][] | null): void {
    this.scope(threadIdArg);
    const pending = this.pending.get(idArg);
    if (!pending?.question || !pending.questionIds || pending.request.params.threadId !== threadIdArg) throw new Error('Codex question is no longer pending in this conversation.');
    if (answersArg === null) {
      if (!this.supervisor.requireClient().respond(pending.request.id, { answers: {} })) throw new Error('Codex question expired before dismissal.');
      return;
    }
    if (answersArg.length !== pending.questionIds.length) throw new Error('Codex question answer count does not match.');
    const answers = Object.fromEntries(pending.questionIds.map((id, index) => [id, { answers: answersArg[index] }]));
    if (!this.supervisor.requireClient().respond(pending.request.id, { answers })) throw new Error('Codex question expired before the reply.');
  }

  private changed(threadIdArg: string, typeArg: IControllerEvent['type']): void {
    const scope = this.scopes.get(threadIdArg);
    if (scope && !this.supervisor.signal.aborted) this.emit({ type: typeArg, harnessId: 'codex',
      projectId: scope.projectId, sessionId: codexRuntimeId(threadIdArg),
      ...(typeArg === 'session.changed' ? { codexActivity: this.activitySummary(threadIdArg) } : {}),
      timestamp: Date.now() });
  }
}