/**
 * Self-tick for the fleet orchestrator session.
 *
 * The orchestrator is a 24/7 omp session with a standing brief (composed ORCHESTRATOR.md + POLICY.md)
 * and no user typing into it. A session that is never prompted never runs its
 * loop, so this extension is the heartbeat: every `intervalSeconds` it injects
 * one message that starts a turn.
 *
 * Three properties are worth protecting, and each one is a branch in
 * `tickDecision()`:
 *
 * - **A disarmed fleet is not woken.** The arm marker is a file the operator
 *   controls; missing means "not armed", and a tick then does nothing.
 * - **The human channel must still be there.** Autonomous dispatch is only
 *   defensible while a tier-2 escalation can reach a person, so every tick
 *   re-reads the Telegram bridge's access file and requires it enabled with
 *   exactly one paired owner. This is fail-closed: an unreadable, unparseable or
 *   ambiguous access file stops ticking. A stale arm marker must not outlive the
 *   channel that makes running unattended safe.
 * - **Ticks coalesce.** A tick that lands while an earlier one is still queued
 *   would stack prompts on a session that is already behind. `hasPendingMessages()`
 *   makes the tick idempotent under slow turns — and, because it is the only
 *   signal in the process that says whether the last prompt was ever consumed,
 *   it doubles as the wedge detector: see {@link STALL_MARKER_FILE}.
 *
 * Beyond those three gates, every tick carries the project's `reporting.scope`
 * as one explicit constraint line, re-read from the conductor config on each
 * tick so an `omp-conductor setup` change binds the next heartbeat rather than
 * waiting for a session restart — and one delivery rule
 * ({@link TICK_DELIVERY_RULE}), because a tick is injected locally and a report
 * written as end-of-turn text on such a turn reaches nobody. An operator's own
 * `message` replaces both, and is re-read per tick for the same reason. One
 * clause is not the operator's to replace: a tick composed on a surface that
 * has no `telegram_ask` says so ({@link TICK_APPROVAL_UNAVAILABLE_RULE}),
 * custom prompt included, because the floor's amendment approval names a tool
 * that surface cannot call.
 *
 * The extension is inert unless `<cwd>/.conductor-tick.json` exists, so shipping
 * it inside `omp-conductor` costs an ordinary session nothing. That file is a
 * property of the *directory*, though, which is why arming is gated on one more
 * question — {@link resolveTickOwnership}: is this session the orchestrator, or
 * merely a session standing in its directory? Without it, a shell opened in the
 * fleet's cwd armed a second heartbeat and, with merge and release delegated in
 * config, believed it held both.
 */

import { createHash, randomUUID } from "node:crypto";
import { spawnSync } from "node:child_process";
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from "node:fs";
import { dirname, isAbsolute, join, resolve } from "node:path";
import { availabilityPrompt, interruptDisposition } from "./availability.ts";
import {
  findProject,
  loadConfig,
  resolveReleaseGrants,
  resolveSharedInstallAuthority,
  stateDir,
} from "./config.ts";
import {
  bridgeTokenBound,
  hasBotToken,
  readApprovalSurface,
  readDaemonProfile,
  TELEGRAM_APPROVAL_TOOL,
} from "./approval-surface.ts";
import {
  briefPathForProject,
  policyPathForProject,
  reportScopeFromPolicy,
  refreshComposedBriefForProject,
} from "./setup.ts";
import { sharedPolicyPath } from "./brief-upgrade.ts";
import {
  recordReleaseBlock,
  redactReleaseArgs,
  releaseBlockedInvocation,
  releaseDriftDigestLine,
  releaseRefusal,
  releaseToolMatch,
  sharedHostGuardDigestLine,
  type ReleaseDecision,
} from "./release-policy.ts";
import {
  DEFAULT_REPORT_POLICY,
  DEFAULT_REPORT_SCOPE,
  DENIED_RELEASE_GRANTS,
  INTERRUPT_CATEGORIES,
  type DispatchSummary,
  type DigestBacklog,
  type FrictionSignal,
  type GroomingRecord,
  type HeldNotice,
  type GroomTrigger,
  type InterruptCategory,
  type IntakeItem,
  type IssueState,
  type MaterialEvent,
  type ProjectConfig,
  type ReadyIssue,
  type ReportScopeChoice,
  type ReportingPolicy,
  type ResolvedGrants,
  type Store,
} from "./types.ts";
import { makeTracker } from "./tracker/github.ts";
import { formatDecisionDigest } from "./decisions.ts";
import { installSurfaceMismatch } from "./status-render.ts";
import {
  ASK_TOOL,
  type AskDeliveryResult,
  MAX_QUESTIONNAIRE_ITEMS,
  QUESTIONNAIRE_TOOL,
  parseQuestionnaireRequest,
  performQuestionnaire,
  questionnaireParameterSchema,
  askParameterSchema,
  parseAskRequest,
  performAsk,
  type AskResult,
} from "./ask.ts";
import { deliverOperatorMessage } from "./reports.ts";
import { effectiveLabels } from "./routing.ts";
import type { FailureClass, RecoveryAction, RunRecord } from "./types.ts";
import { dbPath, openStore } from "./store.ts";
import { digestDue, localDayKey } from "./digest-schedule.ts";
import {
  DEFAULT_GROOM_BELOW,
  groomingDue,
  parseReadyGateRejection,
  TO_SPEC_IN_FLIGHT_REASON,
  toSpecDurableVerdict,
  type ToSpecTrackerSeam,
} from "./groom.ts";
import { heldNoticeId } from "./notices.ts";
import {
  clearArmTransaction,
  expiredArmChallenges,
  FLEET_ARM_KEY,
  resolveArmReply,
  type ArmChallengeSighting,
  type ArmReplyMatch,
} from "./arm-challenge.ts";
import {
  readTelegramChannel,
  readTelegramToken,
  resolveProjectTopicId,
  sendTelegram,
} from "./escalate.ts";

/** The activation file. Absent means "this is not an orchestrator session". */
export const TICK_CONFIG_FILE = ".conductor-tick.json";

/** Namespaced so a renderer or a session-log reader can pick ticks out. */
export const TICK_CUSTOM_TYPE = "omp-conductor.tick";

/**
 * A tick costs a whole turn of a frontier model, and the orchestrator's loop is
 * about minutes of latency, not seconds. Anything under a minute is a
 * misconfiguration worth refusing rather than obeying.
 */
export const MIN_INTERVAL_SECONDS = 60;
/** Repetition threshold and windows for Learning-loop friction signals. */
export const FRICTION_MIN_OBSERVATIONS = 3;
export const FRICTION_LOOKBACK_MS = 7 * 24 * 60 * 60 * 1_000;
export const FRICTION_COOLDOWN_MS = 7 * 24 * 60 * 60 * 1_000;
const FRICTION_DIGEST_LIMIT = 3;

/**
 * How often to re-ask who owns the fleet tick after herdr failed to answer.
 *
 * Retrying rather than latching is the whole point: a declined identity is
 * permanent, an unanswered one is a blip. Cheap enough to run every minute — one
 * short-lived `herdr agent list` — and it stops the moment the answer is
 * definitive, either way.
 */
const RETRY_OWNERSHIP_MS = 60_000;
/** A failed digest turn must release working-hours catch-up ownership promptly. */
const MIN_AVAILABILITY_DIGEST_LEASE_MS = 30 * 60 * 1_000;

/**
 * The stall marker — written beside the activation file, in the session cwd —
 * and the number of consecutive coalesced ticks that earn it.
 *
 * This detects a failure that happened. 2026-08-07: the dogfood fleet's
 * orchestrator finished a turn, logged `ui.loop-blocked` immediately after an
 * auto-compaction threshold decision, and never started another. The process
 * stayed alive, so `herdr-conductor`'s recovery — agent listed AND a
 * non-shell foreground process — read healthy, and the dispatch daemon is a
 * different process entirely, so `/healthz` stayed green too. A tick injected
 * two minutes later and an operator's Telegram message five minutes after that
 * both sat unconsumed for 23 minutes, until a manual SIGTERM. This extension
 * had the evidence and threw it away: coalescing logged `tick skipped: tick
 * already pending`, which reads exactly like healthy backpressure.
 *
 * Two in a row is a full hour at the reference 1800s interval — generous by
 * construction, because a turn that legitimately runs an hour belongs to a
 * fleet with larger problems than one spurious marker. The escalation has to
 * leave the session, since a wedged loop cannot report on itself: a file any
 * out-of-process watchdog can stat, plus an error-level line a log scraper can
 * match.
 */
export const STALL_MARKER_FILE = ".conductor-stalled";
export const STALL_TICKS = 2;

/** A turn older than this has its remaining tool calls refused (#189). */
const DEFAULT_TICK_BUDGET_SECONDS = 600;
/** Grace before a queued operator message preempts the turn's tool calls. */
const PENDING_MESSAGE_GRACE_MS = 60_000;
/**
 * Written by herdr-conductor `recover.sh` *before* `agent start`, and by the
 * dispatch daemon when a watched decision condition transitions false→true
 * (#329), so the orchestrator can reconcile without waiting a full
 * `intervalSeconds`. Cleared only after a tick is actually sent — a disarmed or
 * channel-down fleet keeps the request until gates pass (or a human removes the
 * file). Mid-interval the heartbeat polls for this file on a short cadence.
 */
export const TICK_REQUESTED_FILE = ".conductor-tick-requested";

/** How often a live heartbeat looks for {@link TICK_REQUESTED_FILE} between ticks. */
const TICK_REQUEST_POLL_MS = 10_000;

/** Runtime heartbeat schedule consumed by `omp-conductor status`. */
export const TICK_STATUS_FILE = ".conductor-tick-status.json";

export interface TickRuntimeStatus {
  pid: number;
  intervalSeconds: number;
  nextTickAt: string;
}

export function readTickRuntimeStatus(cwd: string): TickRuntimeStatus | undefined {
  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(join(cwd, TICK_STATUS_FILE), "utf8"));
  } catch {
    return undefined;
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return undefined;
  const row = parsed as Record<string, unknown>;
  if (
    typeof row["pid"] !== "number" ||
    !Number.isInteger(row["pid"]) ||
    row["pid"] <= 1 ||
    typeof row["intervalSeconds"] !== "number" ||
    !Number.isInteger(row["intervalSeconds"]) ||
    row["intervalSeconds"] < MIN_INTERVAL_SECONDS ||
    typeof row["nextTickAt"] !== "string" ||
    !Number.isFinite(Date.parse(row["nextTickAt"]))
  ) {
    return undefined;
  }
  return {
    pid: row["pid"],
    intervalSeconds: row["intervalSeconds"],
    nextTickAt: row["nextTickAt"],
  };
}

/**
 * The marker's one line after its ISO timestamp, and the middle of the error
 * log. Shared so the file and the log can never describe different failures.
 */
const STALL_DIAGNOSIS = `${STALL_TICKS} ticks queued unconsumed — the agent loop is not draining`;

/**
 * The one skip reason the stall counter reacts to, shared with the decision so
 * a reworded log line cannot silently disarm the detector.
 */
const PENDING_REASON = "tick already pending";

/**
 * The slice of the omp extension API this entry touches, mirroring
 * `ExtensionAPI` / `ExtensionContext` from `@oh-my-pi/pi-coding-agent`.
 *
 * Declared here rather than imported because the harness is a peer dependency and
 * the package has to type-check without it installed — the same reason ./omp.ts
 * loads the SDK dynamically. Structural typing means the real objects satisfy these
 * on the way in.
 */
interface TickLogger {
  info(message: string, context?: Record<string, unknown>): void;
  error(message: string, context?: Record<string, unknown>): void;
}

interface TickContext {
  /** Session cwd — where the activation file is looked for. */
  cwd: string;
  /**
   * Whether a UI is attached — false in print/RPC mode, and false for every
   * subagent. Typed as the SDK types it: `ExtensionContext.hasUI: boolean`,
   * `@oh-my-pi/pi-coding-agent/src/extensibility/extensions/types.ts:424-425`.
   */
  hasUI: boolean;
  ui: {
    notify(message: string, type?: "info" | "warning" | "error"): void;
  };
  /** True while steering, follow-up or next-turn messages are still queued. */
  hasPendingMessages(): boolean;
  /**
   * The session's own transcript path, or `undefined` before anything is written
   * to it. Typed as the SDK types it: `ExtensionContext.sessionManager` is a
   * `ReadonlySessionManager`, whose `getSessionFile(): string | undefined` is at
   * `@oh-my-pi/pi-coding-agent/src/session/session-manager.ts`. Read for one
   * reason — a directory claim that names only a pid tells an operator which
   * process holds the tick but not which session it is.
   */
  sessionManager: { getSessionFile(): string | undefined };
  /**
   * Managed timer: throws inside `callback` are contained and surfaced on the
   * extension error channel, the handle is `unref`'d, and it is cleared on
   * `session_shutdown`. Raw `setInterval` has none of that and a throwing tick
   * would take the session down, so it is never used here.
   */
  setInterval(callback: () => void, ms?: number): unknown;
}

interface TickApi {
  logger: TickLogger;
  /**
   * The currently active tool names. Typed as the SDK types it:
   * `ExtensionAPI.getActiveTools(): string[]`,
   * `@oh-my-pi/pi-coding-agent/src/extensibility/extensions/types.ts:1267-1268`.
   */
  getActiveTools(): string[];
  /**
   * Replace the session's active tool set by name. The bounded ask surface
   * ({@link ASK_TOOL}) is registered with `defaultInactive` (#594) and brought
   * into the live set here only when the current tick config resolves a
   * project — so an unusable tool schema is never model-visible on a degraded
   * heartbeat. Signature matches the SDK: `setActiveTools(toolNames)`.
   */
  setActiveTools(toolNames: string[]): Promise<void>;
  /**
   * Register one tool on this session — the same shape the verb client uses
   * (`verbs/client.ts`). The bounded ask surface ({@link ASK_TOOL}) is mounted
   * with it, so the orchestrator's only waiting ask can never outlive its
   * ceiling. `defaultInactive` has the SDK meaning: registered but not in the
   * initial active set, activated later via {@link setActiveTools}.
   */
  registerTool(tool: {
    name: string;
    label: string;
    description: string;
    parameters: unknown;
    approval?: "read" | "write" | "exec";
    defaultInactive?: boolean;
    execute(
      toolCallId: string,
      params: Record<string, unknown>,
    ): Promise<{ content: { type: "text"; text: string }[]; isError?: boolean }>;
  }): void;
  on(event: "session_start", handler: (event: { type: "session_start" }, ctx: TickContext) => void): void;
  on(event: "turn_start", handler: (event: { type: "turn_start" }, ctx: TickContext) => void): void;
  on(event: "turn_end", handler: (event: { type: "turn_end" }, ctx: TickContext) => void): void;
  on(
    event: "message_start",
    handler: (
      event: {
        type: "message_start";
        message: {
          role: string;
          customType?: string;
          synthetic?: boolean;
          attribution?: "user" | "agent";
          timestamp?: number;
          /** Full-text content parts, as the runtime's `AgentMessage` carries
           *  them. The arming-proof classifier hashes the reply's tokens
           *  against persisted pending-challenge state, so it needs the text
           *  here — content absent means "no reply text", which never matches. */
          content?: readonly { type?: string; text?: string }[];
        };
      },
      ctx: TickContext,
    ) => void,
  ): void;
  on(event: "agent_end", handler: (event: { type: "agent_end" }, ctx: TickContext) => void): void;
  /**
   * `deliverAs: "followUp"` + `triggerTurn: true`, verified against
   * `AgentSession.sendCustomMessage` rather than assumed:
   *
   * - idle: the `nextTurn` branch is skipped, `triggerTurn` prompts immediately,
   *   so the tick starts a turn now;
   * - streaming: the message is queued as a follow-up and drained when the
   *   current turn ends, which is exactly the intended "tick after this
   *   finishes" and is also what `hasPendingMessages()` sees, so the next tick
   *   coalesces instead of stacking.
   *
   * `attribution: "user"` bills the turn as operator-initiated work, which is
   * what a heartbeat prompt is.
   */
  sendMessage(
    message: {
      customType: string;
      content: string;
      display: boolean;
      attribution: "user" | "agent";
    },
    options: { triggerTurn?: boolean; deliverAs?: "steer" | "followUp" | "nextTurn" },
  ): void;
}

/** Validated activation config. `armedFile` and `accessFile` are absolute once
 *  they get here. */
export interface TickConfig {
  intervalSeconds: number;
  /**
   * The conductor project this fleet session ticks for, stamped by `setup host`
   * (one tick config per fleet cwd). It is what lets a host with several
   * configured projects resolve *this* fleet's brief, reporting policy, digest
   * ledger and release grants, instead of asking `findProject` a question it
   * refuses to guess through.
   *
   * Omitted is the pre-multi-project spelling. Resolution then falls back to the
   * un-named `findProject`: exact for a single-project fleet — which must keep
   * behaving exactly as it did — and ambiguous, therefore degrading, for more
   * than one.
   */
  project?: string;
  armedFile?: string;
  accessFile?: string;
  message?: string;
  /**
   * Seconds a turn may run before the tick guard refuses its remaining tool
   * calls (#189). Optional; defaults to {@link DEFAULT_TICK_BUDGET_SECONDS} (10
   * minutes, at or below a normal tick interval so a runaway turn is caught
   * before the next tick).
   */
  budgetSeconds?: number;
  /**
   * The herdr agent name this fleet's orchestrator pane is registered under, and
   * the whole of {@link resolveTickOwnership}'s identity test under herdr.
   * Omitted means {@link DEFAULT_FLEET_AGENT_NAME}, which is
   * `herdr/bin/recover.sh`'s own `AGENT_NAME=${AGENT_NAME:-fleet}` default — the
   * recovery half and the ticking half key on one identity or neither is safe.
   */
  agentName?: string;
}

/**
 * Three outcomes, none of them an exception: an absent file is the normal case
 * for every other session, and a broken one must not stop the rest of the
 * extension host from loading.
 */
export type TickConfigResult =
  | { kind: "absent"; path: string }
  | { kind: "invalid"; path: string; problem: string }
  | { kind: "ok"; path: string; config: TickConfig };

/**
 * The prompt when the config names none. The timestamp is what makes two
 * consecutive ticks distinguishable in the session log.
 *
 * "Re-read … from disk" is an order, not colour. A 24/7 session holds a copy of
 * the brief from its own start (or its last compaction), and a tick that merely
 * says "run your loop" was observed acting on that cached copy first — which
 * means an operator's amendment, or a dated standing task added between ticks,
 * may never bind. The whole point of a file the operator can edit is that the
 * next tick obeys the file, not the memory of it.
 *
 * Deliberately silent about reporting volume: that clause is
 * {@link TICK_SCOPE_CONSTRAINTS}, appended per tick from the configured scope.
 * A second spelling of it here would contradict the first inside one prompt the
 * moment a fleet chose `escalations`.
 */
export function defaultTickMessage(
  now: Date,
  briefPath = "ORCHESTRATOR.md",
  policyPath = "POLICY.md",
  sharedPath?: string,
): string {
  const sources =
    sharedPath === undefined ? "package floor + policy" : "package floor + shared policy + policy";
  return (
    `Tick ${now.toISOString()}: re-read ${briefPath} (composed ${sources}) and ` +
    `${policyPath} (editable fleet policy) from disk, then run your standing loop from them. ` +
    `Learning-loop amendments edit only ${policyPath} — never the package floor.`
  );
}

/**
 * The reporting contract, one line per scope, appended to the tick prompt.
 *
 * This is the whole of what `reporting.scope` does at tick time: it constrains
 * what the turn is allowed to say, in the session that reads it. It is not an
 * outbound filter — nothing downstream drops a report the orchestrator decides
 * to send anyway.
 *
 * A mapped type rather than a plain object, so adding a member to
 * `REPORT_SCOPES` fails to compile here instead of resolving to `undefined` at
 * the point of use.
 */
/**
 * Which recovery actions resolve a run without leaving human work behind.
 * Exhaustive over {@link RecoveryAction} so adding an action to the vocabulary
 * forces a decision here: a recovered row carrying `true` is genuinely done and
 * may be summarised as "already handled"; `false` — `escalate`, `hold`, `none` —
 * still needs the orchestrator's Duty 1 attention, so it must never inherit the
 * suppressive "do not re-triage" sentence (#610).
 */
const AUTONOMOUS_RECOVERY_ACTIONS: Record<RecoveryAction, boolean> = {
  requeue: true,
  continue: true,
  "rerun-checks": true,
  settle: true,
  escalate: false,
  // An observation row never stamps `recoveredAt` (the recovery IS the later
  // sweep pass), so it never lands in this digest; `false` is the honest value
  // if one ever did — it is not "already handled", it is waiting (#1068).
  observe: false,
  hold: false,
  none: false,
};

/** The failure-class family that names a turns/wall-clock cap kill. Only these
 *  rows carry an ordinal: a second cap kill is a decomposition verdict rather
 *  than a retry candidate, so the prompt must say which attempt it was without
 *  the orchestrator re-querying the runs table (#711). */
const CAP_KILL_CLASSES: Record<string, true> = {
  "turn-cap-progress": true,
  "turn-cap-spinning": true,
  "wall-clock-cap-progress": true,
  "wall-clock-cap-spinning": true,
};

/** Issue-history lookups answering the two questions about a recovered row:
 *  has the issue moved on since that row, and how many attempts has it spent.
 *  Lazy callbacks so the tick can hand the store straight through and the
 *  digest only pays for the rows it actually reports.
 */
export interface RecoveryDigestContext {
  /** Every attempt of the issue, oldest first — exactly what
   *  `Store.runsForIssue` returns. The recovered row itself is among the
   *  answers, so "has the issue moved on" is "is the last element this row". */
  runsForIssue(issue: number): readonly RunRecord[];
}

/** Bounded name-and-count summary shared by the two recovered-row groups. A
 *  cap-kill row also names which attempt it was of the issue's total, taken
 *  from the issue's own attempt history — never from its position in this
 *  digest's list (#711).
 */
function recoveredSummary(
  group: readonly RunRecord[],
  runsForIssue: (issue: number) => readonly RunRecord[],
): string {
  const named = group
    .slice(0, 5)
    .map((r) => {
      const label = `${r.failureClass ?? "unknown"} #${r.issue}`;
      if (r.failureClass === undefined || CAP_KILL_CLASSES[r.failureClass] !== true) return label;
      const attempts = runsForIssue(r.issue).length;
      if (attempts === 0) return label;
      return `${label} (attempt ${r.attempt} of ${attempts})`;
    })
    .join(", ");
  const rest = group.length > 5 ? `, +${group.length - 5} more` : "";
  return `${group.length} (${named}${rest})`;
}

/**
 * One line naming what the daemon recovered without asking (#132).
 *
 * The orchestrator used to write this paragraph by re-deriving it from the run
 * rows every tick. Undefined when nothing was recovered: a line reading "0" is
 * one nobody reads on the day it says 4.
 *
 * With {@link RecoveryDigestContext} supplied, a recovered row whose issue has
 * since gained a newer run — live, pushed-green, or settled to any terminal
 * state — is history, not news, and is dropped before rendering, so a finished
 * recovery is never re-reported as this tick's own (#711).
 */
export function recoveryDigestLine(
  recovered: readonly RunRecord[],
  ctx?: RecoveryDigestContext,
): string | undefined {
  if (recovered.length === 0) return undefined;
  // Memoized per issue: a window can hold two recovered attempts of one issue,
  // and each row must not re-query the store for the same history.
  const runsByIssue = new Map<number, readonly RunRecord[]>();
  const runsForIssue = (issue: number): readonly RunRecord[] => {
    let runs = runsByIssue.get(issue);
    if (runs === undefined) {
      runs = ctx?.runsForIssue(issue) ?? [];
      runsByIssue.set(issue, runs);
    }
    return runs;
  };
  const fresh =
    ctx === undefined
      ? recovered
      : recovered.filter((r) => {
          const runs = runsForIssue(r.issue);
          // The recovered row is still its issue's newest attempt. Any newer
          // run means the issue moved on after this row recovered.
          const latest = runs[runs.length - 1];
          return latest === undefined || latest.id === r.id;
        });
  if (fresh.length === 0) return undefined;
  const handled: RunRecord[] = [];
  const triage: RunRecord[] = [];
  for (const r of fresh) {
    if (r.recoveryAction !== undefined && AUTONOMOUS_RECOVERY_ACTIONS[r.recoveryAction]) {
      handled.push(r);
    } else {
      triage.push(r);
    }
  }
  const lines: string[] = [];
  if (handled.length > 0) {
    lines.push(`Auto-recovered since last tick: ${recoveredSummary(handled, runsForIssue)} — already handled, do not re-triage these.`);
  }
  if (triage.length > 0) {
    lines.push(`Recovered but needs Duty 1 triage: ${recoveredSummary(triage, runsForIssue)} — inspect these runs.`);
  }
  return lines.join("\n");
}

/** Per-reason counts for a verdict slice — "file-lane 2, depends-on 1". */
function groomingGroupCounts(records: readonly GroomingRecord[]): string {
  const counts = new Map<string, number>();
  for (const record of records) {
    counts.set(record.reason, (counts.get(record.reason) ?? 0) + 1);
  }
  return [...counts.entries()].map(([reason, count]) => `${reason} ${count}`).join(", ");
}

/** Per-verdict counts for a slice — "promotable 1, considered 1". */
function groomingVerdictCounts(records: readonly GroomingRecord[]): string {
  const counts = new Map<string, number>();
  for (const record of records) {
    counts.set(record.verdict, (counts.get(record.verdict) ?? 0) + 1);
  }
  return [...counts.entries()].map(([verdict, count]) => `${verdict} ${count}`).join(", ");
}

/**
 * The queue-label inventory read fresh from the tracker on one tick, with the
 * pending label ops applied exactly as the dispatch pass judges eligibility
 * (#848). Queue verdicts in the present tense ("empty", "running low") are
 * only ever drawn from this observation; a dispatch summary alone is dated
 * history and must never be rendered as the live queue.
 */
export interface QueueObservation {
  /** Open issues carrying the queue label under the *effective* label set —
   *  the count dispatch would route against on its next pass. */
  queued: number;
  /** The instant the tracker observation was made, epoch ms. */
  observedAt: number;
}

/**
 * One line telling the orchestrator the routable queue is running dry (#181),
 * or `undefined` when healthy — no dispatch recorded yet, or the routable count
 * at/above a numeric grooming trigger (`"always"` renders on every dispatched
 * tick, #988, stating what remains ungroomed instead of a threshold). `grooming`
 * is the durable per-issue verdict
 * table (#735): `blocked` rows are admission's lane/dependency holds, so the
 * line tells claimable candidates apart from a runway that cannot move —
 * instead of inviting Duty 2 to groom work whose last pass held it, which is
 * exactly the re-derivation this store exists to stop. A to-spec launch row
 * (#777, written by the daemon's grooming launcher since #1041) carries the
 * same `blocked` verdict as its durable in-flight marker, so it is told apart
 * from the mechanical holds: it says "a batch is running", not "the lane
 * cannot move", and counts neither as claimable nor as known-blocked.
 *
 * `queue` is the tracker observation the caller made THIS tick (#848). When
 * present, the queue verdict reads in the present tense from that inventory,
 * and dispatch-derived counts are only ever dated context. When absent the
 * tracker could not be read, and this never asserts that the queue is empty:
 * the dispatch row is rendered explicitly as-of instead.
 */
export function queueDigestLine(
  summary: DispatchSummary | undefined,
  queueLabel: string,
  labelPrefix: string,
  groomBelow: GroomTrigger,
  grooming: readonly GroomingRecord[] = [],
  queue: QueueObservation | undefined = undefined,
  /** Observation time for the durability of the grooming rows this line
   *  describes — the same clock the daemon's selection judges them against, so
   *  the inventory and the next grooming pass cannot disagree about what is
   *  still groomed (#887). */
  now: number = Date.now(),
): string | undefined {
  if (summary === undefined) return undefined;
  if (queue !== undefined) {
    return liveQueueDigestLine(summary, queue, queueLabel, labelPrefix, groomBelow, grooming, now);
  }
  return datedQueueDigestLine(summary, queueLabel, labelPrefix, groomBelow, grooming, now);
}

/** The dated rendering: the last dispatch row is the only evidence, so every
 *  count carries its snapshot date and nothing is asserted as current. */
function datedQueueDigestLine(
  summary: DispatchSummary,
  queueLabel: string,
  labelPrefix: string,
  groomBelow: GroomTrigger,
  grooming: readonly GroomingRecord[],
  now: number,
): string | undefined {
  const dated = new Date(summary.completedAt).toISOString();
  if (summary.ready === 0) {
    if (summary.paused === true) {
      return `Queue: unchecked — the last pass (${dated}) was a hold that counted no queue and admitted nothing; the tracker could not be re-read this tick, so the live queue is NOT claimed empty. Groom only after a fresh read.`;
    }
    return `Queue: empty (as of the last dispatch, ${dated}) — nobody carried "${queueLabel}" on that count, and the live tracker was not re-read this tick, so this is a snapshot, not a present-tense claim. Groom the backlog (Duty 2) once a fresh read confirms it.`;
  }
  if (summary.routed === 0) {
    // `ready` counts claimed (in-flight) issues too; route() drops those with a
    // state label silently, so "0 routable" must not blanket-blame missing
    // `repo:` labels (#228). Spare depth is what dispatch can actually claim.
    const claimed = summary.claimed ?? 0;
    const unroutable = summary.holds
      .filter((h) => h.reason.startsWith("unroutable:"))
      .reduce((n, h) => n + h.count, 0);
    // A residual lifecycle label — a settled run whose state label was never
    // cleared — is neither spare nor in flight, and `claimed` counts only
    // live/settling ownership, so name it as Duty 1 reconciliation work rather
    // than folding it into the grooming instruction (#611).
    const staleLifecycle = summary.holds
      .filter((h) => h.reason === "stale-lifecycle")
      .reduce((n, h) => n + h.count, 0);
    let line = `As of the last dispatch (${dated}): Queue: ${summary.ready} ready, 0 spare — ${claimed} in flight`;
    if (unroutable > 0) {
      line += `, ${unroutable} unroutable (each unroutable issue needs exactly one "${labelPrefix}<repo>" label)`;
    }
    if (staleLifecycle > 0) {
      line += `, ${staleLifecycle} with a residual lifecycle label (Duty 1: newest run terminal — reconcile the stale agent:in-progress/blocked/failed state)`;
    }
    if (unroutable === 0 && staleLifecycle === 0) {
      line += `. Spare depth is what dispatch can actually claim: groom the backlog (Duty 2) before the live runs settle.`;
    } else {
      line += ".";
    }
    return line;
  }
  if (!groomingDue(summary.routed, groomBelow)) return undefined;
  return lowQueueTail(summary, groomBelow, grooming, `As of the last dispatch (${dated}): Queue: `, now);
}

/** The live rendering: the tracker observation is the queue label inventory,
 *  the dispatch summary supplies the claimability reading, and the shared
 *  low-queue diagnostics are composed underneath (#848 review 2). The raw
 *  carrier count appears only as a present-tense inventory fact — never as
 *  the grooming threshold, which must be claimability (`summary.routed`, the
 *  same count the to-spec offer gates on), or parked/in-flight/state-labelled
 *  carriers would silence a queue that dispatch cannot move. */
function liveQueueDigestLine(
  summary: DispatchSummary,
  queue: QueueObservation,
  queueLabel: string,
  labelPrefix: string,
  groomBelow: GroomTrigger,
  grooming: readonly GroomingRecord[],
  now: number,
): string | undefined {
  const queued = queue.queued;
  const observed = new Date(queue.observedAt).toISOString();
  // The present-tense empty verdict is the tracker's own word: nothing open
  // carries the queue label right now. No dispatch snapshot may produce a
  // live "empty" claim — that is the whole defect (#848).
  if (queued === 0) {
    return (
      `Queue: empty — nothing open carries "${queueLabel}" right now (tracker ${observed}). ` +
      "Groom the backlog (Duty 2): the daemon runs the to-spec passes and promotes what its ready gate " +
      "passes, so file or sharpen the next issues, or say in this tick's report why there is nothing to do."
    );
  }
  // The grooming threshold is claimability — summary.routed — never raw
  // carrier volume. Four in-flight/parked/unroutable carriers must not
  // suppress the low-claimable signal, because the to-spec offer gates on
  // the very same routed count and the two would split.
  if (!groomingDue(summary.routed, groomBelow)) return undefined;
  const dated = new Date(summary.completedAt).toISOString();
  const inventory = `Queue: ${queued} open issue${queued === 1 ? "" : "s"} carry "${queueLabel}" right now (tracker ${observed}); `;
  if (summary.routed === 0) {
    // Dispatch's own claimability read: `ready` counts claimed (in-flight)
    // issues too, so "0 routable" must not blanket-blame missing `repo:`
    // labels (#228), and a residual lifecycle label is Duty 1 reconciliation
    // work, not spare depth (#611).
    if (summary.paused === true) {
      return (
        inventory +
        `the last dispatch (${dated}) was a hold that admitted nothing — the queued issues above are the work that resumes when claiming reopens.`
      );
    }
    const claimed = summary.claimed ?? 0;
    let line = inventory + `at the last dispatch (${dated}) ${summary.ready} ready, 0 spare — ${claimed} in flight`;
    const unroutable = summary.holds
      .filter((h) => h.reason.startsWith("unroutable:"))
      .reduce((n, h) => n + h.count, 0);
    const staleLifecycle = summary.holds
      .filter((h) => h.reason === "stale-lifecycle")
      .reduce((n, h) => n + h.count, 0);
    if (unroutable > 0) {
      line += `, ${unroutable} unroutable (each unroutable issue needs exactly one "${labelPrefix}<repo>" label)`;
    }
    if (staleLifecycle > 0) {
      line += `, ${staleLifecycle} with a residual lifecycle label (Duty 1: newest run terminal — reconcile the stale agent:in-progress/blocked/failed state)`;
    }
    if (unroutable === 0 && staleLifecycle === 0) {
      line += `. Spare depth is what dispatch can actually claim: groom the backlog (Duty 2) before the live runs settle.`;
    } else {
      line += ".";
    }
    return line;
  }
  // routed > 0 below the trigger: the live inventory leads, then the same
  // low-queue diagnostics the dated path ships — the claimable/known-blocked
  // split, in-flight to-spec batches, the considered backlog and this pass's
  // holds (#735, #777, #679) — so a tracker read never hides them.
  return inventory + lowQueueTail(summary, groomBelow, grooming, "", now);
}

/** The shared low-queue diagnostic tail (#735/#777/#679): the routable-count →
 *  claimable/known-blocked split, in-flight to-spec batches, the considered
 *  backlog and this pass's holds. `lead` is the sentence opener the caller
 *  chooses — the dated path opens with the dispatch timestamp and "Queue:",
 *  the live path with its tracker inventory (#848 review 2). Sharing one tail
 *  means a tracker read can never hide these distinctions. */
function lowQueueTail(
  summary: DispatchSummary,
  groomBelow: GroomTrigger,
  grooming: readonly GroomingRecord[],
  lead: string,
  /** The clock the durability of each verdict is judged against — the same one
   *  the daemon's selection uses (#887). */
  now: number,
): string {
  // The durable per-issue verdicts, not this pass's one-shot hold groups: a
  // lane-blocked runway must read as "cannot move" even after a restart, and
  // the orchestrator's own prior verdicts (#679) must not be re-derived.
  //
  // An in-flight to-spec launch (#777) is backlog inventory, not a routed
  // ready candidate, a durable hold, or a verdict. Name it separately without
  // subtracting it from the routed queue or printing lane/dependency guidance.
  const inFlight = grooming.filter((g) => g.reason === TO_SPEC_IN_FLIGHT_REASON);
  const knownBlocked = grooming.filter(
    (g) => g.verdict === "blocked" && (g.reason === "file-lane" || g.reason === "depends-on"),
  );
  const considered = grooming.filter((g) => g.verdict === "promotable" || g.verdict === "considered");
  const claimable = Math.max(0, summary.routed - knownBlocked.length);
  const inFlightNames = inFlight.map((r) => `#${r.issue}`).join(", ");
  const busy = inFlight.length === 0 ? "" : `, ${inFlight.length} in a to-spec batch (${inFlightNames})`;
  let tail: string;
  if (knownBlocked.length > 0 && claimable === 0) {
    tail =
      `${lead}running low — ${summary.routed} routable candidate(s), all known-blocked ` +
      `(${groomingGroupCounts(knownBlocked)})${busy} — no grooming moves them; the holds clear by themselves` +
      `${inFlight.length === 0 ? "" : " and the to-spec batch's results land when it settles"}.`;
  } else if (knownBlocked.length > 0 || inFlight.length > 0) {
    // Who grooms is the daemon's business now (#1041): the tail reports which
    // candidates the pipeline can actually move, and stops instructing a
    // session that no longer holds the launch.
    tail =
      groomBelow === "always"
        ? `${lead}grooming runs every pass ("always") — ${summary.routed} routable candidate(s): ` +
          `${claimable} claimable, ${knownBlocked.length} known-blocked (${groomingGroupCounts(knownBlocked)})` +
          `${busy} — only the claimable are groomable.`
        : `${lead}running low — ${summary.routed} routable candidate(s), below the grooming trigger of ${groomBelow}: ` +
          `${claimable} claimable, ${knownBlocked.length} known-blocked (${groomingGroupCounts(knownBlocked)})` +
          `${busy} — only the claimable are groomable.`;
  } else if (groomBelow === "always") {
    // No numeric trigger exists to be "below", so the tail states the actual
    // condition and what remains ungroomed (#988) instead of inventing one.
    tail =
      `${lead}grooming runs every pass ("always") — ${summary.routed} routable candidate(s); ` +
      "the daemon keeps grooming while ungroomed, unrefused, unparked candidates remain.";
  } else {
    tail = `${lead}running low — ${summary.routed} routable candidate(s), below the grooming trigger of ${groomBelow}.`;
  }
  if (considered.length > 0) {
    // "Already considered" is not "will not be re-groomed": selection only
    // withholds a candidate whose verdict is still durable, so the inventory
    // says which of the two each row is. Claiming "never re-groom these" over
    // rows the mechanical selection was simultaneously offering is the #887
    // defect — the same predicate now answers both.
    const durable = considered.filter((g) => toSpecDurableVerdict(g, now) !== undefined);
    const regroomable = considered.length - durable.length;
    tail +=
      ` Backlog already-considered: ${considered.length} (${groomingVerdictCounts(considered)}) — ` +
      `${durable.length} still durable (never re-groom these), ${regroomable} re-groomable ` +
      "(no readable to-spec source, or observed past the freshness ceiling). " +
      "The daemon promotes what its ready gate passes; audit those promotions and sharpen what it rejected.";
  }
  if (summary.admitted === 0 && summary.holds.length > 0) {
    const held = summary.holds
      .map((h) => {
        const first = h.details?.[0];
        return `${h.reason} ${h.count}${first === undefined ? "" : ` (${first})`}`;
      })
      .join(", ");
    tail += ` All held: ${held}.`;
  }
  return tail;
}

// ============================================================ promotion audit
// (#1041/#1040). #777 gave the tick a per-tick capability token: a launch block
// naming the mechanically selected batch, a `task` tool-call gate that redeemed
// exactly one batch, and promotion as an orchestrator act. That made grooming
// tick-bound — one batch per low-queue tick, and an operator message that
// consumed the tick consumed the batch (#1040) — so the whole mechanism is
// gone. The daemon watches depth, launches its own to-spec sessions, and
// promotes a PROMOTABLE verdict whose spec passes the pure ready gate.
//
// What the tick owes Duty 2 instead is review, and review needs exactly two
// facts the store already holds: what the daemon queued unattended since the
// last beat, and which specs its gate refused. Neither is derivable from
// session memory, and both must arrive with the command that acts on them —
// an audit line an operator cannot act on is a line that gets skipped.

/** How many promotions and gate rejections one audit block names. A tick's
 *  prompt is a budget, and an unbounded burst of promotions must not push the
 *  duties off the end of it; the count always states the full total. */
const PROMOTION_AUDIT_LIMIT = 6;

/**
 * The Duty-2 promotion audit: what the daemon promoted on its own since the
 * previous tick, each with the one command that undoes it, and the specs its
 * ready gate refused, each with what the gate found missing.
 *
 * `undefined` when there is nothing to say — the common state of a fleet whose
 * queue is deep enough that no grooming pass ran. A block that reported "0
 * promotions" every beat would be permanent noise, the same convention
 * {@link queueDigestLine} and {@link formatPendingIntake} keep.
 *
 * The demote command is spelled from the real verb surface rather than
 * invented: `conductor_label` (verbs/protocol.ts) takes the full issue URL,
 * the label, `action`, and a `reason` from {@link LABEL_REASONS} — `needs-human`
 * is the member that means "parked: it needs a decision only a human can
 * make", which is exactly what an overruled auto-promotion is.
 */
export function formatPromotionAudit(input: {
  /** Grooming rows promoted since {@link since}, as `Store.promotionsSince`
   *  returns them (newest or oldest first — the block does not reorder, so the
   *  store's order is the audit's order). */
  promotions: readonly GroomingRecord[];
  /** Every current grooming row, from which the gate rejections are read. The
   *  full set rather than a window: a spec the gate refused three ticks ago is
   *  still unqueued work, and dropping it would let a rejected spec go quiet. */
  grooming: readonly GroomingRecord[];
  project: string;
  tracker: string;
  queueLabel: string;
  /** The lower bound the promotions were read from, named so the orchestrator
   *  can tell "nothing happened" from "a narrow window". */
  since: number;
}): string | undefined {
  const rejected = input.grooming.flatMap((row) => {
    const rejection = parseReadyGateRejection(row.evidence);
    return rejection === undefined ? [] : [{ row, rejection }];
  });
  if (input.promotions.length === 0 && rejected.length === 0) return undefined;
  const lines: string[] = [
    `Promotion audit (Duty 2) — ${input.promotions.length} auto-promotion(s) since ` +
      `${new Date(input.since).toISOString()}, ${rejected.length} spec(s) the ready gate refused:`,
  ];
  for (const row of input.promotions.slice(0, PROMOTION_AUDIT_LIMIT)) {
    const at = row.promotedAt === undefined ? "unknown time" : new Date(row.promotedAt).toISOString();
    const by = row.promotedBy ?? "unrecorded";
    // The gate that passed is named from the provenance the promoter stored,
    // never guessed: the audit exists to review what the daemon did
    // unattended, and a guessed attribution would make that review circular.
    const gate =
      by === "daemon"
        ? "ready gate passed"
        : by === "orchestrator"
          ? "promoted by an earlier tick"
          : "promoted by an operator";
    lines.push(
      `- #${row.issue} — verdict ${row.verdict} (${row.reason}); ${gate}; queued ${at} by ${by}. ` +
        `Demote: omp-conductor verb conductor_label --project ${input.project} ` +
        `--arg issueUrl=https://github.com/${input.tracker}/issues/${row.issue} ` +
        `--arg label=${input.queueLabel} --arg action=remove --arg reason=needs-human ` +
        `--arg rationale="<why this spec is not ready>"`,
    );
  }
  if (input.promotions.length > PROMOTION_AUDIT_LIMIT) {
    lines.push(`- … ${input.promotions.length - PROMOTION_AUDIT_LIMIT} further promotion(s) not shown.`);
  }
  if (rejected.length > 0) {
    lines.push(
      "Gate-rejected specs — never queued, and the pass that produced them is already spent, so the " +
        "issue itself is what has to change. These repeat every tick until they are fixed, which is why " +
        "each carries the date its gate ran:",
    );
    // Oldest rejection first: a spec refused days ago and never edited is the
    // one the audit keeps re-printing, so it leads rather than sinking under
    // whatever the last pass produced.
    const ordered = [...rejected].sort((a, b) => a.rejection.checkedAt - b.rejection.checkedAt);
    for (const { row, rejection } of ordered.slice(0, PROMOTION_AUDIT_LIMIT)) {
      const checked =
        rejection.checkedAt === 0 ? "unknown date" : new Date(rejection.checkedAt).toISOString();
      lines.push(`- #${row.issue} (gate ran ${checked}) missing: ${rejection.missing.join("; ")}`);
    }
    if (rejected.length > PROMOTION_AUDIT_LIMIT) {
      lines.push(`- … ${rejected.length - PROMOTION_AUDIT_LIMIT} further rejected spec(s) not shown.`);
    }
  }
  lines.push(
    "Duty 2 is now three things and no launch: propose net-new scope through a decision row, audit the " +
      "promotions above (demote what you disagree with, using the command on its line), and edit the " +
      "gate-rejected issues until the missing fields are there — the daemon re-runs the gate by itself.",
  );
  return lines.join("\n");
}

export const TICK_SCOPE_CONSTRAINTS: { readonly [K in ReportScopeChoice]: string } = {
  material: "Report material events per your brief.",
  escalations:
    "Interrupt only for: tier2, fleet-stopped; everything else — releases included — waits for the daily digest.",
  decisions:
    "Reporting scope decisions: interrupt only for a decision you need (tier-2) or a condition that stops the fleet. Every other material event waits for the configured digest; nothing between digests.",
  quiet:
    "Interrupt only for: tier2, fleet-stopped, confirmed-failure; everything else waits for the daily digest.",
};

/**
 * Collapses one durable row's prose to a single bounded prompt line.
 *
 * Shared by every appended block rather than re-declared inside each: they all
 * render operator- or model-authored text into a prompt where one pathological
 * row must not push the duties out of the window, and three private copies of
 * the same four lines is how one of them ends up unbounded.
 */
function flattened(text: string, limit: number): string {
  const flat = text.replace(/\s+/g, " ").trim();
  return flat.length <= limit ? flat : `${flat.slice(0, limit - 1)}…`;
}

/** Bounded, durable source material for one due digest. Row ids are part of the
 * handoff contract: the report and exactly the rows it consumed settle in one
 * SQLite transaction, so a crash cannot lose an outcome between those writes. */
export function formatDigestLedger(
  events: readonly MaterialEvent[],
  notices: readonly HeldNotice[],
  backlog: DigestBacklog,
): string {
  const lines = [
    `Durable digest ledger — ${backlog.materialCount} material event(s), ${backlog.heldNoticeCount} held notice(s); oldest first:`,
    ...events.map(
      (event) =>
        `  event ${event.id} | ${new Date(event.occurredAt).toISOString()} | ${event.category} | ` +
        `${flattened(event.summary, 180)} | evidence: ${flattened(event.evidence, 240)}`,
    ),
    ...notices.map(
      (notice) =>
        `  notice ${notice.id} | ${new Date(notice.createdAt).toISOString()} | ${notice.category} | ` +
        `${flattened(notice.summary, 180)} | detail: ${flattened(notice.detail, 240)}`,
    ),
  ];
  if (backlog.materialCount > events.length) {
    lines.push(`  … ${backlog.materialCount - events.length} newer material event(s) remain owed.`);
  }
  if (backlog.heldNoticeCount > notices.length) {
    lines.push(`  … ${backlog.heldNoticeCount - notices.length} additional held notice(s) remain owed.`);
  }
  const eventIds = events.map((event) => event.id).join(",");
  const noticeIds = notices.map((notice) => notice.id).join(",");
  lines.push(
    "Compose the digest from the relevant rows above. Hand it off with omp-conductor report --kind digest " +
      `${eventIds.length === 0 ? "" : `--events ${eventIds} `}` +
      `${noticeIds.length === 0 ? "" : `--notices ${noticeIds} `}` +
      "--text TEXT. Remove any row id you did not actually include; omitted rows remain owed. Never invent an id.",
  );
  return lines.join("\n");
}

/**
 * Reserved `material_events` category for a mechanical amendment the
 * orchestrator applied on its own authority (Phase 4 — the floor's
 * mechanical-vs-judgment split).
 *
 * There is no amendments table, and neither other durable candidate can carry
 * this audit honestly. `POLICY.md`'s own `## Amendments` section is written by
 * step 4 of the floor's *judgment* protocol only — a mechanical fix skips
 * straight to applying and reporting, so it never lands there — and that
 * section's line shape (date, trigger, one-sentence summary) has nowhere to put
 * a revert command. `friction_rollups` counts admission holds and report
 * classifications and knows nothing about policy edits. `material_events` is
 * the append-only outbox the floor already uses for every outcome that is not
 * sent as an interrupt, its category is a free lowercase slug (`event record`
 * validates `^[a-z0-9][a-z0-9-]{0,31}$`), and both its text columns are
 * mandatory — `--summary` 1-240 characters, `--evidence` 1-500 — which makes
 * `evidence` exactly the right home for a one-line revert. So the audit is a
 * category convention over the ledger that already exists, and needs no schema
 * change. Note where the cap bites: a revert longer than 500 characters is
 * refused by `event record` at write time rather than truncated by the renderer
 * below, so an over-long one is a revert to shorten, never a silently clipped
 * command an operator would paste.
 *
 * The consequence to keep in view: this surface holds what the orchestrator
 * wrote to it and nothing else. An amendment applied without the ledger write
 * is invisible here, which is why the floor's mechanical bullet and the
 * friction digest both name the exact command rather than leaving the shape to
 * be guessed.
 */
export const AMENDMENT_EVENT_CATEGORY = "amendment";

/** Amendment rows rendered in one audit. A window with more self-applied policy
 *  edits than this is a fleet problem the prompt cannot fix by listing them. */
const AMENDMENT_AUDIT_LIMIT = 5;

/**
 * How deep into the outbox the audit reads.
 *
 * Deliberately far past {@link DIGEST_BACKLOG_LIMIT}, which bounds one digest's
 * worth of source material: an audit inheriting that bound would hide a
 * self-applied policy edit behind twenty merge events, and that row is the one
 * that must never go unseen.
 */
export const AMENDMENT_AUDIT_SCAN_LIMIT = 200;

/**
 * The one sentence in the tick prompt that states the amendment cap.
 *
 * A constant, and used in exactly one rendered block, because the cap is the
 * clause the floor just re-scoped: it counts *judgment* proposals, and a
 * mechanical drift fix is not a proposal at all. A second copy of this sentence
 * anywhere in the prompt is how the old undifferentiated wording comes back —
 * the audit block below therefore states the duty and never restates the cap.
 */
export const AMENDMENT_THROTTLE_LINE =
  "The cap is at most one new orchestrator-originated judgment amendment proposal per autonomous tick, and it " +
  "counts judgment proposals only — duties, boundaries, release policy, caps, reporting scope. A mechanical " +
  "drift fix is not a proposal and is not throttled: apply it, report it with its revert, and apply the next one.";

/**
 * Mechanical amendments the orchestrator applied without asking, read back to
 * it before they reach its operator (Phase 4).
 *
 * The floor's mechanical branch trades an ask for a promise — apply now, then
 * say in the next report what changed, why, and how to revert it. This block is
 * what makes that promise auditable by something other than session memory,
 * which is exactly the memory the whole ledger discipline exists to distrust.
 *
 * The window is "recorded and not yet carried into a delivered digest", not
 * "since the previous tick". That is the honest predicate rather than the
 * approximate one: the tick has no durable record of its own previous beat, and
 * an interval-width window silently drops the audit when a turn runs long or a
 * digest fails — the two occasions where an unreviewed policy edit is most
 * likely. Undigested is also self-clearing: composing the report retires the
 * row, so a reported amendment stops appearing without anyone marking it.
 *
 * On a due-digest tick this deliberately overlaps {@link formatDigestLedger},
 * which lists the same row for a different purpose — the ledger hands over row
 * ids to consume, the audit hands over a policy edit to re-justify or revert.
 * Suppressing one because the other rendered would mean a self-applied
 * amendment goes unreviewed precisely on the tick that ships it.
 */
export function formatAmendmentAudit(events: readonly MaterialEvent[]): string | undefined {
  const amendments = events.filter((event) => event.category === AMENDMENT_EVENT_CATEGORY);
  if (amendments.length === 0) return undefined;
  const shown = amendments.slice(0, AMENDMENT_AUDIT_LIMIT);
  const lines = [
    `Mechanical amendments you applied without asking — ${amendments.length}, not yet carried into a report. ` +
      "Each one is a POLICY.md edit made on your own authority, so the audit is yours before it is your operator's:",
    ...shown.map(
      (event) =>
        `- ${new Date(event.occurredAt).toISOString()} | ${flattened(event.summary, 180)} | ` +
        `revert: ${flattened(event.evidence, 240)}`,
    ),
  ];
  if (amendments.length > shown.length) {
    lines.push(`- … ${amendments.length - shown.length} further amendment(s) not shown.`);
  }
  lines.push(
    "Re-check each against the source that made it mechanical — the file, config key or ledger row that proves " +
      "the value. Still true: carry the line into your next report exactly as it reads here, revert command " +
      "included. No longer true, or a source you can no longer state: run the revert now, and propose the change " +
      "as a judgment amendment if it is still worth making.",
  );
  return lines.join("\n");
}

/**
 * The reporting constraint appended to a default tick prompt (#229, #242).
 *
 * Material and decisions keep their legacy preset words. The escalations
 * preset instead names the policy's actual interrupt categories, so the prompt
 * cannot drift from the gate. Every daily cadence gets an explicit due state:
 * model-timed digests need that instruction just as scheduled digests do.
 */
export function tickReportingConstraint(
  policy: ReportingPolicy | undefined,
  digest: {
    due: boolean;
    cadence: ReportingPolicy["digest"]["cadence"];
    at?: string;
    timezone?: string;
  },
  held: number,
): string {
  const preset = policy?.scopePreset;
  const allowed = policy?.interruptOn ?? [];
  let base: string;
  if (preset === "escalations") {
    base = `Interrupt only for: ${allowed.join(", ")}; everything else — releases included — waits for the daily digest.`;
  } else if (preset !== undefined) {
    base = TICK_SCOPE_CONSTRAINTS[preset];
  } else {
    base =
      allowed.length === 0
        ? "Report nothing that would interrupt this turn; everything else accumulates for the digest."
        : `Interrupt only for: ${allowed.join(", ")}. Everything else accumulates for the digest.`;
  }
  if (policy !== undefined && !allowed.includes("material")) {
    base +=
      " Record every material outcome that is not sent as an interrupt immediately with omp-conductor event record; the durable ledger, not session memory, feeds the next digest.";
  }
  if (digest.cadence !== "daily") return base;
  if (digest.due) {
    return `${base} The daily digest is DUE now — compose it from the durable accumulated events and ${held} held notice(s), then send via omp-conductor report --kind digest.`;
  }
  return digest.at === undefined
    ? `${base} The daily digest was already sent today; do not send another.`
    : `${base} The daily digest is not due (scheduled ${digest.at}${digest.timezone === undefined ? "" : ` ${digest.timezone}`}); do not send one.`;
}

/**
 * The delivery clause, appended to every default tick prompt.
 *
 * A tick's end-of-turn text does not reach the operator as a *report*, and the
 * rule's job is to stop a session believing otherwise: on 2026-08-06 the fleet
 * this extension runs produced a release report and two tier-2 escalations as
 * end-of-turn text, and not one of the three reached anybody.
 *
 * This comment used to explain that by saying end-of-turn text streams to
 * Telegram only on a turn that *began* as an inbound Telegram message. That
 * premise is false, and the correction matters because it flips the reason:
 * omp-telegram's `agent_end` handler posts `finalText` to the notify chat on
 * every run that did *not* come from Telegram, whenever `notifyMode` is set —
 * and the fleet must set it, because {@link readApprovalSurface} needs it for
 * the approval surface. So a tick's closing prose *was* reaching the operator
 * all along: not as a delivered report, as an unlogged, unretried, untracked
 * ping in the middle of whatever else was in that chat. Both the void and the
 * ping are the same fault seen from two sides — text is not a delivery channel.
 *
 * omp-telegram 0.11.0's `profile: "daemon"` removes it in the other direction:
 * explicit-only outbound, and the idle/final notify post suppressed outright, so
 * a tick's visible text goes nowhere at all. That is the configuration the fleet
 * runs (see {@link TICK_NARRATION_RULE}, which warns when it is absent), and it
 * makes this rule's demand literal rather than merely prudent.
 *
 * The clause used to name `telegram_send`, and that held — but it was still an
 * instruction where a mechanism was needed: the same miss recurs whenever the
 * session is distracted, compacted or interrupted mid-report, and an unsent
 * report is indistinguishable from a quiet tick. #123 moved delivery to the
 * daemon, so the clause now names the handover instead. `omp-conductor report`
 * persists the text before anything is sent and the daemon retries it until it
 * lands; a handover that returns a durable report or held-notice id has
 * therefore *already* survived the failure mode this rule exists for, which a
 * watched `telegram_send` never could. `telegram_send` remains the right call
 * for an interactive reply to a person who is waiting — that is a
 * conversation, not a report.
 */
export const TICK_DELIVERY_RULE =
  "This tick was injected locally, not sent from Telegram, so your end-of-turn text does NOT reach your operator. Hand anything reportable this turn to the durable outbox by running `omp-conductor report --text \"<the whole report>\"` (add `--kind digest` for the daily digest) and confirming it printed a durable handoff id (a report id, or a held-notice id during quiet hours); the daemon then owns delivery and retries until it lands. Never claim a report was sent otherwise, and never use telegram_send for a report -- that path leaves no record that it went out.";

/** The {@link TICK_DELIVERY_RULE} variant for a fleet whose bridge actually
 *  delivers the tick's ending text (#169). That is a narrower class than the
 *  approval surface being ready: `notifyMode: "always"` has to be set, and the
 *  profile has to be `interactive` — a `daemon` profile suppresses the
 *  `agent_end` notify post, so its end-of-turn text reaches nobody and the
 *  durable-outbox rule above is the honest one there. Everything here keeps the
 *  outbox as the message; only the opening claim about delivery differs. */
export const TICK_DELIVERY_RULE_BRIDGED =
  "Your end-of-turn text IS delivered to the operator by the Telegram bridge (notifyMode: always). Still hand anything reportable to the durable outbox with omp-conductor report — and then close with at most one short line. The report is the message; never restate it in your closing text.";

/** Re-exported so the tick's own contract stays readable from one file: the
 *  constant itself lives beside the check that decides whether it is callable. */
export { TELEGRAM_APPROVAL_TOOL };

/**
 * Appended to every tick — the shipped prompt or the operator's own — composed
 * on a surface where {@link TELEGRAM_APPROVAL_TOOL} is not mounted.
 *
 * The floor mandates a call the ordinary tick cannot make. On 2026-08-09 06:17Z
 * a locally injected veltrosecurity tick reached the amendment step and found
 * no `telegram_ask` at all (`read xd://telegram_ask` answered `No such tool`);
 * the very next turn, which began as an inbound Telegram message at 06:33Z,
 * found the same tool mounted. Availability tracks turn origin, and an
 * amendment proposal normally arises on the locally injected half — issue #114.
 *
 * Fails closed in the only sense that helps a fleet: the duties still run, but
 * the turn is told the approval primitive is missing *before* it can reach the
 * step that needs one. The fallback named here is the one the floor already
 * documents for a `telegram_ask` that never delivered — re-deliver the same
 * question — so a session has one answer to "the ask did not happen", not two.
 * It names `omp-conductor message` rather than `telegram_send` because this
 * rule only ever reaches a locally injected turn: there is no inbound message
 * to inherit a chat and topic from, and a bare send would answer a forum fleet
 * in its main chat (#366).
 *
 * The last clause is the actual hazard #114 exposed: a turn that knows
 * it must ask, and cannot, is one inference away from recording an approval
 * nobody gave. Phase 4 narrows its noun to a *judgment* amendment, and only its
 * noun: a mechanical drift fix is applied without an ask by design, so an
 * undifferentiated "never apply an amendment without an answer" here would
 * contradict the floor on the branch that has no ask to be missing. The hazard
 * this rule guards — recording an unanswered yes/no as approved — is unchanged.
 *
 * Unlike {@link TICK_DELIVERY_RULE} this is appended to a configured `message`
 * too. An operator's prompt owns the reporting contract and is theirs to get
 * wrong; it cannot consent, on the orchestrator's behalf, to a tool being
 * absent from the surface the turn actually runs on.
 */
export const TICK_APPROVAL_UNAVAILABLE_RULE =
  `The ${TELEGRAM_APPROVAL_TOOL} tool is NOT mounted on this tick, so the package floor's yes/no amendment approval cannot be asked here. ` +
  `If you have a judgment amendment to propose, deliver the question with \`omp-conductor message --category decision-needed --text "<the question>"\` — ` +
  `it resolves this project's own Telegram chat and topic, records the question as an open decision row, and applies the ` +
  `same availability policy as the tick: an unanswered yes/no stays pending — re-surfaced in every tick until answered ` +
  `or the seven-day expiry, never recorded as approved. ` +
  `Wait for the operator's reply on a later turn, then resolve the row with \`omp-conductor decision resolve <id> --answer "..."\`. ` +
  `A returned answer proves an answer, not Telegram delivery. ` +
  `Never apply a judgment amendment, or record one as approved, without an explicit answer you actually received.`;

/**
 * Appended to every tick — the shipped prompt or the operator's own — because
 * the durable ask surface is the mechanical contract of a local tick (#438).
 *
 * No ask waits any more. This session is the headless tick brain: operator
 * turns land in the 24/7 console session, so an answer physically cannot
 * arrive here, and a duty cycle that waited would stop the fleet for nothing —
 * on 2026-08-16 that was six stopped hours. So the rule states the whole
 * contract up front: the question becomes a durable row, the console resolves
 * it later, and parking the blocked work is part of the same turn rather than
 * something the model discovers when a wait expires.
 */
export const TICK_ASK_RULE =
  `On this locally injected tick, questions to your operator go through ${ASK_TOOL}: ` +
  `call it with the question, its options and your recommendation. It does NOT wait — nothing here does. ` +
  `The question is filed as a durable decision row and delivered, and the answer arrives later, when your ` +
  `operator resolves that row from the console session (\`omp-conductor decision resolve <id> --answer "..."\`). ` +
  `So in the SAME turn you ask, park the work the question blocks: take it out of the claimable queue, record ` +
  `its state, and name the row id in your report. Never wait for a reply, never re-ask on the next tick, and ` +
  `never proceed as though your recommendation had been approved. ` +
  `Several judgement calls about ONE issue go as a single ${QUESTIONNAIRE_TOOL} instead: one delivery, ` +
  `each item a durable row bound to that issue, resolved independently and in any order. ` +
  `The ${TELEGRAM_APPROVAL_TOOL} tool is refused here: it would wait for your operator for as long as the answer ` +
  `takes, and an unanswered question must never hold the loop.`;

/**
 * The gate's refusal for a raw {@link TELEGRAM_APPROVAL_TOOL} / `write
 * xd://telegram_ask` call on a locally injected tick (#438). Fixed wording the
 * way {@link TICK_ASK_RULE} is: a model that must act on a refusal reads it the
 * same way every time.
 */
export const ASK_DIVERSION_REASON =
  `Blocked: on this locally injected tick, ${TELEGRAM_APPROVAL_TOOL} would wait for your operator for as long as ` +
  `the answer takes, and an unanswered question must never stop the loop (#438). Nothing was sent or recorded ` +
  `by this call. Ask instead with ${ASK_TOOL} — it records the question as a durable decision row, delivers it, ` +
  `and returns immediately with the row id and the command your operator's console resolves it with.`;

/**
 * Appended to every tick — the shipped prompt or the operator's own — composed
 * on a bridge that is not in omp-telegram's `profile: "daemon"`.
 *
 * The running commentary an operator sees is *mechanical*, not disobedience, and
 * that is why a prompt line is the wrong permanent fix and the right interim one.
 * Two verified paths carry visible text out without anyone calling a tool:
 * `outbound.ts` `onTurnEnd()` finalizes one real Telegram message per assistant
 * turn for as long as the chat is marked active — so a multi-step answer arrives
 * as several messages, and a message that lands mid-tick keeps the chat active
 * for the rest of that run, relaying every subsequent tick-internal turn — and
 * the `agent_end` handler posts the closing text of every *local* run to the
 * notify chat whenever `notifyMode` is set. The fleet has to set `notifyMode`
 * (see {@link TICK_APPROVAL_UNAVAILABLE_RULE}), so the correctly-askable fleet is
 * exactly the narrating one. `profile: "daemon"` closes both at the transport:
 * explicit-only outbound, notify post suppressed.
 *
 * So this rule is a stopgap with an expiry date, and it says so: it names the
 * one command that removes it. Until then the only defence is discipline the
 * turn can actually exercise — no visible text between tool calls, and one
 * `telegram_send` for a human answer rather than prose that leaks a turn at a
 * time. It fires *because* the mechanism is absent, so a fleet that has run
 * `/telegram set profile daemon` never sees it.
 *
 * Appended to a configured `message` too, for {@link TICK_APPROVAL_UNAVAILABLE_RULE}'s
 * reason exactly: an operator's prompt owns the reporting contract, but it cannot
 * consent on the orchestrator's behalf to what the transport does with its text.
 * Transport truth is not the operator's prompt to waive.
 */
export const TICK_NARRATION_RULE =
  "This session's Telegram bridge is NOT in daemon profile, so visible assistant text can auto-relay to your operator's chat (per-turn messages while a Telegram conversation is active, and end-of-run notify posts on local runs). Until the operator runs `/telegram set profile daemon`: produce no visible text between tool calls, and answer any human message with a single telegram_send call only.";

function frictionLabel(kind: FrictionSignal["kind"]): string {
  if (kind.startsWith("admission:")) return `admission hold ${kind.slice("admission:".length)}`;
  if (kind === "feedback:escalation-should-digest") return "escalations classified as digest material";
  if (kind === "feedback:report-noise") return "tick reports classified as noise";
  return "tick reports classified as surprising";
}

/**
 * Bounded evidence for the existing approval protocol — never an automatic edit.
 *
 * The closing instruction is the one place the tick prompt routes into the
 * Learning loop, so it is also where the floor's mechanical-vs-judgment split
 * has to be visible (Phase 4): the gate sentence is unchanged — friction with no
 * safe POLICY.md remedy still leaves policy alone — and the two branches under
 * it say which kind of amendment the remedy is, and what a mechanical one owes
 * the ledger. It carries {@link AMENDMENT_THROTTLE_LINE} and is the only block
 * that does.
 */
export function formatFrictionDigest(signals: readonly FrictionSignal[]): string {
  const shown = signals.slice(0, FRICTION_DIGEST_LIMIT);
  const lines = [
    "Repeated friction observed over the last 7 days (evidence only; not permission to edit policy):",
    ...shown.map((signal) => {
      const issues =
        signal.issues.length === 0 ? "" : `; issues ${signal.issues.map((issue) => `#${issue}`).join(", ")}`;
      const samples = signal.samples.length === 0 ? "" : `; examples: ${signal.samples.join(" | ")}`;
      return (
        `- ${frictionLabel(signal.kind)} — ${signal.observations} observations, ` +
        `${signal.occurrences} affected${issues}${samples}`
      );
    }),
  ];
  if (signals.length > shown.length) lines.push(`- ${signals.length - shown.length} more signal(s) deferred`);
  lines.push(
    "After the tick duties, investigate at most one signal. Use the existing Learning loop only if the recurring cause has a safe POLICY.md remedy; otherwise leave policy unchanged and report or file the underlying product/infra issue through the existing rules.",
    "When there is a remedy, route it by kind. A drift fact you can verify against the repo, the config or " +
      "this fleet's own ledger — a renamed label, a moved path, a changed schedule, a caps figure the config " +
      "already carries — is a MECHANICAL amendment: edit POLICY.md now, then record it with " +
      '`omp-conductor event record --category ' +
      AMENDMENT_EVENT_CATEGORY +
      ' --summary "<what changed and why>" --evidence "<the one-line revert>"`, which is what puts it in the ' +
      "next tick's amendment audit and in your next report. A remedy that is a choice about how the fleet " +
      "should behave is a JUDGMENT amendment and goes through the proposal protocol unchanged. " +
      AMENDMENT_THROTTLE_LINE,
  );
  return lines.join("\n");
}

/** Bounds one pending idea's text in the tick prompt, like the digest ledger. */
const INTAKE_TEXT_LIMIT = 160;
/** Bounds a mined item's provenance key beside its text. Shorter than the text
 *  because a source is an identifier, not prose — a longer one is malformed. */
const INTAKE_SOURCE_LIMIT = 80;

/**
 * Pending intake items, each waiting to become an issue (#300).
 *
 * The one appended block that is a *duty* rather than a read-out: every pending
 * item is listed with its id and text (oldest first, as the store returns
 * them), followed by the standing instruction for grooming one into an issue —
 * file it without the queue label, mark provenance with `intake groomed`, and
 * name the grooming in the digest ledger. Absent when nothing is pending: an
 * empty intake is the common state of a healthy fleet, and a block that said
 * "0 pending" every tick would be permanent noise — the same convention as
 * {@link queueDigestLine}. Project-specific grooming taste (priority scales,
 * template wording) stays in POLICY.md; this block carries only the floor duty.
 *
 * Provenance is rendered per item because the two kinds are not the same work
 * (Phase 4 signal mining). An item with no `source` is an idea a person typed:
 * it is scope, and grooming it means writing down what they asked for. A mined
 * item is a machine's inference from settlement flags, failure classes or
 * friction — evidence to check against the code before anything is filed, and
 * the one kind that can be honestly dismissed on the orchestrator's own
 * judgement when the evidence does not hold up. Rendering them
 * indistinguishably invited exactly one error: filing a mined guess as though
 * an operator had asked for it.
 */
export function formatPendingIntake(
  items: readonly IntakeItem[],
  instruction: { tracker: string; queueLabel: string; labelPrefix: string },
): string | undefined {
  if (items.length === 0) return undefined;
  // Bounded like the text: a mined source is a key, and a pathological one must
  // not push the grooming instruction out of the prompt window.
  const provenance = (item: IntakeItem): string =>
    item.source === undefined
      ? "your operator's own idea"
      : `mined signal ${flattened(item.source, INTAKE_SOURCE_LIMIT)}`;
  const lines = [
    `Pending intake — ${items.length} idea(s) captured, waiting to be groomed into issues, oldest first:`,
    ...items.map((item) => `- ${item.id} (${provenance(item)}) — ${flattened(item.text, INTAKE_TEXT_LIMIT)}`),
    "",
    `Groom each into exactly one issue on ${instruction.tracker}: a title stating the problem; a body ` +
      `carrying the product rationale and the acceptance criteria as a checklist; the routing ` +
      `"${instruction.labelPrefix}<repo>" label and a priority per POLICY.md. File it WITHOUT the ` +
      `"${instruction.queueLabel}" queue label — filing is not promoting, and never add ` +
      `${instruction.queueLabel} to an issue you groomed from intake. Then mark provenance with ` +
      "`omp-conductor intake groomed <id> --issue <url>` — an already-groomed id is a no-op, and an " +
      "idea is filed exactly once — and record the outcome with `omp-conductor event record --category " +
      'intake --summary "groomed intake <id> → #<issue-number>" --evidence <url>` so the digest names ' +
      "the grooming. An item that is malformed or empty is dismissed with `omp-conductor intake dismiss " +
      "<id>` and noted in the digest rather than filed.",
    "The two provenances are groomed differently. Your operator's own idea is scope: file what they asked " +
      "for, and never dismiss it on your own judgement — if it looks wrong, that is a question for them. A " +
      "mined signal is evidence, not scope: open what its source names, confirm from the code that the " +
      "problem is real, and file the issue against what you actually find — its wording is a machine's " +
      "inference and yours to correct. A mined signal the evidence does not support is dismissed, with the " +
      "reason in the digest.",
  ];
  return lines.join("\n");
}

/**
 * The scope this tick carries, where the brief actually lives, and — when the
 * config could not answer — why.
 *
 * Read on every tick rather than cached at session start, for the reason the
 * channel gate is: the operator re-runs `omp-conductor setup` while this session
 * lives, and a heartbeat holding a startup snapshot would keep injecting the
 * old contract until somebody restarted it.
 *
 * `briefPath` exists because the prompt orders a re-read, and an order must
 * name a file that is really there: the brief lives at
 * `<workspaceRoot>/ORCHESTRATOR.md`, not in the session cwd — the cwd usually
 * holds only an `AGENTS.md` symlink to it. The name is spelled here rather than
 * imported from setup.ts, whose import graph drags the session SDK into an
 * extension that must stay cheap to load.
 *
 * Every fault collapses to {@link DEFAULT_REPORT_SCOPE} and a pathless prompt:
 * no config written yet, an unreadable or invalid one, or several projects with
 * none named — the same ambiguity `findProject` refuses to guess through for
 * `status`. Stopping the heartbeat over either preference would be the worse
 * trade.
 *
 * `projectName` is {@link TickConfig.project}, stamped per fleet cwd, and it is
 * the whole difference between a two-project host resolving each fleet's own
 * contract and both of them degrading. Omitted (a pre-multi-project tick config)
 * keeps the un-named lookup, which only a host with more than one project can
 * fail — and that host gets {@link LEGACY_TICK_PROJECT_HINT} naming the one
 * command that fixes it.
 */
export function resolveTickScope(projectName?: string): {
  scope: ReportScopeChoice;
  policy?: ReportingPolicy;
  briefPath?: string;
  policyPath?: string;
  projectName?: string;
  fallback?: string;
} {
  // Counted before `findProject` can throw, because "several projects and this
  // tick config names none" is the one fault with a one-command remedy and the
  // only way to tell it apart from "no config at all" is the count.
  let projects = 0;
  try {
    const config = loadConfig();
    projects = config.projects.length;
    const project = findProject(config, projectName);
    return {
      scope: reportScopeFromPolicy(project.reporting),
      policy: project.reporting,
      briefPath: briefPathForProject(project),
      policyPath: policyPathForProject(project),
      projectName: project.name,
    };
  } catch (err) {
    const problem = err instanceof Error ? err.message : String(err);
    const unstamped = projectName === undefined && projects > 1;
    return { scope: DEFAULT_REPORT_SCOPE, fallback: unstamped ? `${problem} ${LEGACY_TICK_PROJECT_HINT}` : problem };
  }
}

/**
 * The fix for the one degradation an operator can clear in a single command: a
 * tick config written before {@link TickConfig.project} existed, on a host that
 * has since gained a second project.
 */
export const LEGACY_TICK_PROJECT_HINT =
  "Re-run `omp-conductor setup host` to stamp the project into .conductor-tick.json.";

/**
 * Whether a tick config found on disk belongs to the project being asked about.
 *
 * Search roots overlap — `stateDir()` and the shared parent of two fleet cwds
 * are read for every project — so without this a stray config in a shared root
 * answers for whichever project asked first, and `setup host` would restamp one
 * project's file while planning another. An unstamped config matches anything,
 * which is what keeps a single-project fleet on its existing file.
 */
export function tickConfigMatchesProject(config: TickConfig, projectName?: string): boolean {
  if (projectName === undefined || config.project === undefined) return true;
  return config.project === projectName;
}

/**
 * Directory that owns this project's {@link TICK_CONFIG_FILE}, if one is
 * present and stamped for the project (or unstamped on a single-project host).
 * Used by out-of-process writers of {@link TICK_REQUESTED_FILE} — the dispatch
 * daemon must not import `fleet.ts` (fleet already imports daemon).
 */
export function resolveTickConfigCwd(projectName?: string): string | undefined {
  const roots: string[] = [stateDir()];
  try {
    const p = findProject(loadConfig(), projectName);
    const parent = dirname(p.workspaceRoot);
    if (parent !== roots[0]) roots.push(parent);
    if (p.workspaceRoot !== roots[0] && p.workspaceRoot !== parent) roots.push(p.workspaceRoot);
  } catch {
    /* no config */
  }
  for (const cwd of roots) {
    const r = readTickConfig(cwd);
    if (r.kind === "ok") {
      if (!tickConfigMatchesProject(r.config, projectName)) continue;
      return cwd;
    }
    if (r.kind === "invalid") return undefined;
  }
  return undefined;
}

/** The shared, pre-per-project arm marker: `<stateDir>/armed`. */
export function legacyArmedMarkerPath(): string {
  return join(stateDir(), "armed");
}

/** What `status` says about a bare `armed` marker that no tick will honour. */
export const LEGACY_ARM_MARKER_DETAIL =
  "legacy global arm marker — re-run setup host, then arm per project";

/**
 * Whether the arm gate is open, and whether the shared pre-per-project marker is
 * what decided it.
 *
 * Two migrations meet here. A single-project fleet whose `armedFile` has just
 * been restamped from `armed` to `armed-<name>` must not be silently disarmed by
 * the upgrade, so its bare `armed` marker still counts (`legacy: "honoured"`). A
 * host with more than one project cannot let one marker arm every fleet — that is
 * the collision this issue removes — so there the bare marker arms nothing and
 * says so instead (`legacy: "stranded"`), including for a tick config that still
 * names it directly.
 *
 * The fallback reaches exactly two `armedFile` spellings: the shared path itself,
 * and the per-project path the restamp replaces it with. An operator's own gate
 * — any other value, which `setup host` deliberately preserves — is the gate; its
 * absence means disarmed however many stale markers lie around it.
 *
 * The config is read only when a legacy marker is actually in play, so the
 * healthy path costs one string compare. An unreadable or absent config is
 * treated as single-project: a host with no conductor config cannot be a
 * multi-project host, and 0.14 behaviour is the honest default there.
 */
export type ArmState = { armed: boolean; legacy?: "honoured" | "stranded" };

export function resolveArmState(armedFile: string, projectName?: string): ArmState {
  const legacy = legacyArmedMarkerPath();
  const shared = armedFile === legacy;
  if (!shared && existsSync(armedFile)) return { armed: true };
  const migratable =
    shared || (projectName !== undefined && armedFile === join(stateDir(), `armed-${projectName}`));
  if (!migratable || !existsSync(legacy)) return { armed: false };
  let multiProject = false;
  try {
    multiProject = loadConfig().projects.length > 1;
  } catch {
    // No readable config: not a multi-project host, so keep 0.14 behaviour.
  }
  if (multiProject) return { armed: false, legacy: "stranded" };
  return shared ? { armed: true } : { armed: true, legacy: "honoured" };
}

/**
 * The one armed-marker write both settlement paths share — the CLI's
 * `arm --reply` half and the orchestrator session's mechanical reply consumer
 * (#1061): same content, same mode, and the same restamp of the pre-per-project
 * shared marker the heartbeat still honours (#316). It lives here, beside
 * {@link resolveArmState}, so the session consumer can write the gate it reads
 * — the reverse import (into `fleet.ts`) would cycle.
 */
export function writeArmedMarker(path: string, owner: string, arm: ArmState): void {
  mkdirSync(dirname(path), { recursive: true });
  writeFileSync(path, `armed ${new Date().toISOString()} owner=${owner}\n`, { mode: 0o600 });
  if (arm.legacy === "honoured") rmSync(legacyArmedMarkerPath(), { force: true });
}

/**
 * Best-effort recompose of `ORCHESTRATOR.md` from the installed package floor +
 * live `POLICY.md`.
 *
 * Runs on **every** successful send — including ticks that use a custom
 * `message` — so protocol updates land after a package upgrade without waiting for
 * the default prompt path. Failures (no config, no `POLICY.md`, unreadable
 * overlay) are silent: the tick still goes out.
 */
export function refreshComposedBriefBestEffort(projectName?: string): boolean {
  try {
    return refreshComposedBriefForProject(findProject(loadConfig(), projectName));
  } catch {
    return false;
  }
}

/**
 * An optional file-path field. Relative paths resolve against the session cwd
 * so `state/armed` means what it looks like; a present-but-unusable value is a
 * config error rather than something to ignore, because both paths this reads
 * are safety gates and a silently dropped gate is an open one.
 */
function optionalPath(raw: unknown, key: string, cwd: string, problems: string[]): string | undefined {
  if (raw === undefined) return undefined;
  if (typeof raw !== "string" || raw.trim().length === 0) {
    problems.push(`${key} must be a non-empty string when present`);
    return undefined;
  }
  return isAbsolute(raw) ? raw : resolve(cwd, raw);
}

/**
 * Reads and validates `<cwd>/.conductor-tick.json`.
 *
 * Every fault is collected and reported in one message, the way ./config.ts
 * does it: an operator fixing a config wants the whole list, not the first
 * complaint followed by another edit-and-retry cycle.
 */
export function readTickConfig(cwd: string): TickConfigResult {
  const path = join(cwd, TICK_CONFIG_FILE);
  if (!existsSync(path)) return { kind: "absent", path };

  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(path, "utf8"));
  } catch (err) {
    return { kind: "invalid", path, problem: `not valid JSON (${err instanceof Error ? err.message : String(err)})` };
  }

  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
    return { kind: "invalid", path, problem: "must be a JSON object" };
  }

  const raw = parsed as { readonly [key: string]: unknown };
  const problems: string[] = [];

  let intervalSeconds = 0;
  const interval = raw["intervalSeconds"];
  if (typeof interval !== "number" || !Number.isFinite(interval) || !Number.isInteger(interval)) {
    problems.push("intervalSeconds must be a whole number of seconds");
  } else if (interval < MIN_INTERVAL_SECONDS) {
    problems.push(`intervalSeconds must be at least ${MIN_INTERVAL_SECONDS}`);
  } else {
    intervalSeconds = interval;
  }

  // Tolerant like the other optional fields: an integer >= the minimum is the
  // turn budget in seconds, anything else degrades to absent (the shipped
  // default) rather than invalidating the config.
  const budgetRaw = raw["budgetSeconds"];
  const budgetSeconds =
    typeof budgetRaw === "number" && Number.isInteger(budgetRaw) && budgetRaw >= MIN_INTERVAL_SECONDS
      ? budgetRaw
      : undefined;

  // No ask ceiling is read from the config any more: the ask does not wait at
  // all (#438 as re-cut for the console split), so a stale `askTimeoutSeconds`
  // on a deployed fleet's `.conductor-tick.json` is simply an unknown key —
  // ignored like every other one, so an old file on disk can never fail a
  // restart.

  // Relative paths resolve against the session cwd, so the files can sit beside
  // the config that names them (`state/armed`) without hard-coding a deploy path.
  const armedFile = optionalPath(raw["armedFile"], "armedFile", cwd, problems);
  const accessFile = optionalPath(raw["accessFile"], "accessFile", cwd, problems);

  const messageRaw = raw["message"];
  let message: string | undefined;
  if (messageRaw !== undefined) {
    if (typeof messageRaw !== "string" || messageRaw.trim().length === 0) {
      problems.push("message must be a non-empty string when present");
    } else {
      message = messageRaw;
    }
  }

  const agentRaw = raw["agentName"];
  let agentName: string | undefined;
  if (agentRaw !== undefined) {
    if (typeof agentRaw !== "string" || agentRaw.trim().length === 0) {
      problems.push("agentName must be a non-empty string when present");
    } else {
      agentName = agentRaw.trim();
    }
  }

  const projectRaw = raw["project"];
  let project: string | undefined;
  if (projectRaw !== undefined) {
    if (typeof projectRaw !== "string" || projectRaw.trim().length === 0) {
      problems.push("project must be a non-empty string when present");
    } else {
      project = projectRaw.trim();
    }
  }

  if (problems.length > 0) return { kind: "invalid", path, problem: problems.join("; ") };

  return {
    kind: "ok",
    path,
    config: {
      intervalSeconds,
      ...(project === undefined ? {} : { project }),
      ...(budgetSeconds === undefined ? {} : { budgetSeconds }),
      ...(armedFile === undefined ? {} : { armedFile }),
      ...(accessFile === undefined ? {} : { accessFile }),
      ...(message === undefined ? {} : { message }),
      ...(agentName === undefined ? {} : { agentName }),
    },
  };
}

/**
 * The claim file the non-herdr path uses to make "who is the orchestrator here"
 * answerable from disk. A sibling of the activation file, and dot-prefixed like
 * the rest of that family.
 */
export const TICK_OWNER_FILE = ".conductor-tick-owner.json";

/**
 * The agent name a fleet is registered under when its config names none —
 * `herdr/bin/recover.sh`'s own default (`AGENT_NAME=${AGENT_NAME:-fleet}`).
 */
export const DEFAULT_FLEET_AGENT_NAME = "fleet";

/** A herdr query that hangs would hang session startup, so it is bounded. */
const HERDR_QUERY_TIMEOUT_MS = 3000;

/**
 * One entry of `herdr agent list`, narrowed to the two fields that decide
 * identity.
 *
 * `name` is the *registered agent name* — what `herdr agent start fleet --pane`
 * sets, what `recover.sh` keys every identity decision on, and what an ad-hoc
 * shell in the same directory does not have. herdr's `agent` field is
 * deliberately not read: it is the *runtime*, it says `"omp"` for the real
 * orchestrator and for a pane somebody opened to look at state, and reading it
 * is what made this problem look unfixable.
 */
export interface HerdrPaneAgent {
  paneId: string;
  /** Absent on a pane herdr has no registered name for. */
  name?: string;
}

/** Either the list, or why there is none — never an exception: a herdr that does
 *  not answer is a fact to log, not a session that fails to start. */
export type HerdrAgentList =
  | { kind: "ok"; agents: HerdrPaneAgent[] }
  | { kind: "unavailable"; problem: string };

/**
 * Whether this session may tick in this directory.
 *
 * `declined` carries the whole sentence to log, because the point of the check is
 * that an operator can tell which session is driving the fleet — a decline that
 * does not name the holder answers the question no better than silence did.
 */
export type TickOwnership =
  | { kind: "owner"; note?: string }
  /** Proven not to be the fleet's session. Permanent until this session ends. */
  | { kind: "declined"; reason: string }
  /**
   * Could not be determined — herdr did not answer. Deliberately NOT `declined`:
   * both refuse to tick, but only this one is worth retrying, and conflating
   * them means a single 3-second CLI timeout silently disables the real
   * orchestrator until someone restarts the pane. An unproven identity still
   * must not tick; a heartbeat that stopped for a transient blip and never came
   * back is the exact silent stall this package keeps having to fix.
   */
  | { kind: "unresolved"; reason: string };

/**
 * `herdr agent list`, over the socket herdr injected into this pane's
 * environment.
 *
 * Shelled out rather than spoken over the socket directly: the wire protocol is
 * not a published interface, while the CLI's "one JSON line on stdout" is — it is
 * what `herdr/bin/recover.sh` already parses, envelope and all. `HERDR_BIN_PATH`
 * is honoured for the same reason that script honours it, so both halves of this
 * repo have one spelling of "how do we reach herdr".
 *
 * Synchronous on purpose: it feeds a decision taken inside `session_start`, and a
 * heartbeat that armed first and checked afterwards would tick from the wrong
 * session for the length of the window between.
 */
function readHerdrAgents(env: Record<string, string | undefined>): HerdrAgentList {
  const bin = env["HERDR_BIN_PATH"] ?? "herdr";
  let stdout: string;
  try {
    const run = spawnSync(bin, ["agent", "list"], {
      encoding: "utf8",
      timeout: HERDR_QUERY_TIMEOUT_MS,
      stdio: ["ignore", "pipe", "pipe"],
    });
    if (run.error !== undefined) return { kind: "unavailable", problem: run.error.message };
    if (run.status !== 0) {
      // A signal rather than an exit code when herdr was killed mid-answer; both
      // mean the same thing here, and both belong in the log line an operator
      // reads to find out why the heartbeat stopped.
      const how = run.signal === null || run.signal === undefined ? `exited ${String(run.status)}` : `died on ${run.signal}`;
      const detail = (run.stderr ?? "").trim().split("\n")[0] ?? "";
      return { kind: "unavailable", problem: `\`${bin} agent list\` ${how}${detail.length === 0 ? "" : `: ${detail}`}` };
    }
    stdout = run.stdout ?? "";
  } catch (err) {
    return { kind: "unavailable", problem: err instanceof Error ? err.message : String(err) };
  }

  return parseHerdrAgents(stdout);
}

/**
 * The CLI's single JSON line, as `recover.sh` reads it: the payload is either the
 * envelope's `result` or the object itself, and `agents` is inside it. Anything
 * else is reported rather than guessed at — an agent list that cannot be parsed
 * proves nothing about which pane owns the tick.
 */
export function parseHerdrAgents(stdout: string): HerdrAgentList {
  let parsed: unknown;
  try {
    parsed = JSON.parse(stdout);
  } catch (err) {
    return { kind: "unavailable", problem: `agent list was not JSON (${err instanceof Error ? err.message : String(err)})` };
  }
  if (parsed === null || typeof parsed !== "object") return { kind: "unavailable", problem: "agent list was not an object" };

  const envelope = parsed as { readonly [key: string]: unknown };
  const inner = envelope["result"];
  const payload = (inner !== null && typeof inner === "object" ? inner : envelope) as {
    readonly [key: string]: unknown;
  };
  const rows = payload["agents"];
  if (!Array.isArray(rows)) return { kind: "unavailable", problem: "agent list carried no agents array" };

  const agents: HerdrPaneAgent[] = [];
  for (const row of rows) {
    if (row === null || typeof row !== "object") continue;
    const entry = row as { readonly [key: string]: unknown };
    const paneId = entry["pane_id"];
    if (typeof paneId !== "string" || paneId.length === 0) continue;
    const name = entry["name"];
    // `null` is what herdr reports for an unnamed pane, and it means the same
    // thing as the key being absent: this pane is not a registered agent.
    agents.push({ paneId, ...(typeof name === "string" && name.trim().length > 0 ? { name: name.trim() } : {}) });
  }
  return { kind: "ok", agents };
}

/**
 * The herdr half of the decision, with no subprocess in it.
 *
 * Fleet-ness is the *session*, shared by every pane in it, so `HERDR_SESSION` and
 * the cwd cannot distinguish the orchestrator from a shell opened beside it. The
 * pane's registered name can, and it survives a detection gap: herdr counts a
 * pane as an agent terminal on a saved name alone, so the fleet pane is listed
 * with its name even in the moment before its runtime is re-detected.
 */
export function paneOwnership(input: { paneId: string; agentName: string; agents: HerdrPaneAgent[] }): TickOwnership {
  const mine = input.agents.find((a) => a.paneId === input.paneId);
  if (mine?.name === input.agentName) return { kind: "owner" };

  if (mine?.name !== undefined) {
    // A registered agent, just not this fleet's. herdr can name several omp
    // agents in one directory, and requiring merely *a* name would arm each one.
    //
    // The rename hint is conditional because this branch has two very different
    // causes: a scratch pane that is correctly named something else, and the
    // fleet's own pane still carrying the name it had before its tick config was
    // restamped for a project. Only the operator knows which.
    return {
      kind: "declined",
      reason:
        `this pane is agent "${mine.name}", not the fleet agent "${input.agentName}" — ` +
        `if this is the fleet pane under an old name, \`herdr agent rename ${input.paneId} ${input.agentName}\`; ` +
        `this session will not tick`,
    };
  }

  const holder = input.agents.find((a) => a.name === input.agentName);
  if (holder !== undefined) {
    return {
      kind: "declined",
      reason: `pane ${holder.paneId} (agent "${input.agentName}") owns the fleet tick here — this session will not tick`,
    };
  }

  // No name here and nobody else holding it: an ad-hoc pane in the fleet's
  // directory, which is exactly the session that must stay inert. Fail-closed,
  // and the fix is named — an orchestrator that lost its registration is one
  // `herdr agent rename` from ticking again.
  //
  // Deliberately NOT `herdr agent start`: that submits omp *into* the pane's
  // existing shell and requires a pane at a shell prompt hosting no agent
  // (herdr/README.md). This pane is running the omp session reading this line, so
  // `agent start` would either be refused or start a second orchestrator in it.
  // `rename` names the agent herdr already detects, and touches no process.
  return {
    kind: "declined",
    reason:
      `this pane is not a registered herdr agent, and no pane is running the fleet agent ` +
      `"${input.agentName}" — name its running agent with ` +
      `\`herdr agent rename ${input.paneId} ${input.agentName}\`; ` +
      `this session will not tick`,
  };
}

/** The claim on disk. `sessionFile` is for the human reading it — the decision
 *  itself only ever trusts `pid`. */
interface TickOwnerRecord {
  pid: number;
  sessionFile?: string;
  claimedAt: string;
}

/** Whether a pid is running. `EPERM` means it exists and is not ours, which is
 *  still alive; only `ESRCH` proves it is gone. */
function pidIsAlive(pid: number): boolean {
  if (!Number.isInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch (err) {
    return err !== null && typeof err === "object" && "code" in err && err.code === "EPERM";
  }
}

/**
 * The no-herdr half: claim the directory, and tick only while this process is the
 * live claimant.
 *
 * Liveness is a pid check and never a timestamp. A crashed orchestrator leaves
 * its claim behind, and a claim that expired on age alone would either wedge the
 * fleet until someone deleted a file, or hand ownership to a second session while
 * the first was merely slow.
 */
export function claimTickOwner(input: {
  cwd: string;
  pid: number;
  sessionFile?: string;
  now: Date;
  alive?: (pid: number) => boolean;
}): TickOwnership {
  const alive = input.alive ?? pidIsAlive;
  const path = join(input.cwd, TICK_OWNER_FILE);

  const held = readOwnerRecord(path);
  if (held !== undefined && held.pid !== input.pid && alive(held.pid)) {
    return {
      kind: "declined",
      reason:
        `pid ${held.pid} (claimed ${held.claimedAt}${held.sessionFile === undefined ? "" : `, session ${held.sessionFile}`}) ` +
        `owns the fleet tick in ${input.cwd} — this session will not tick`,
    };
  }

  const record: TickOwnerRecord = {
    pid: input.pid,
    ...(input.sessionFile === undefined ? {} : { sessionFile: input.sessionFile }),
    claimedAt: input.now.toISOString(),
  };
  try {
    writeFileSync(path, `${JSON.stringify(record, null, 2)}\n`, { mode: 0o600 });
  } catch (err) {
    // An unwritable claim leaves this session no worse off than it was before the
    // file existed, and refusing to tick over it would silence a fleet that has
    // no rival at all. Said out loud, because the guard is now only advisory.
    return {
      kind: "owner",
      note: `could not write ${path} (${err instanceof Error ? err.message : String(err)}) — ticking anyway, but a second session here would not be detected`,
    };
  }
  return { kind: "owner" };
}

/** A claim that is missing, unreadable or not shaped like one is no claim: the
 *  caller then takes ownership, which is also how a corrupt file heals. */
function readOwnerRecord(path: string): TickOwnerRecord | undefined {
  if (!existsSync(path)) return undefined;
  try {
    const parsed: unknown = JSON.parse(readFileSync(path, "utf8"));
    if (parsed === null || typeof parsed !== "object") return undefined;
    const raw = parsed as { readonly [key: string]: unknown };
    const pid = raw["pid"];
    if (typeof pid !== "number" || !Number.isInteger(pid) || pid <= 0) return undefined;
    const sessionFile = raw["sessionFile"];
    const claimedAt = raw["claimedAt"];
    return {
      pid,
      ...(typeof sessionFile === "string" && sessionFile.length > 0 ? { sessionFile } : {}),
      claimedAt: typeof claimedAt === "string" && claimedAt.length > 0 ? claimedAt : "unknown",
    };
  } catch {
    return undefined;
  }
}

/**
 * Who owns the tick in this directory: the whole answer, both paths.
 *
 * Activation is the presence of `.conductor-tick.json` in the session cwd, which
 * is a property of the *directory* — so every omp session started there became a
 * ticker, and with `authority.merge`/`authority.release` delegated, a shell
 * somebody opened to read state believed it could merge and release. The
 * directory cannot identify a session, so this asks something that can: under
 * herdr, the pane's registered agent name; otherwise, a claim on the directory
 * that is only honoured while its claimant is alive.
 *
 * Every collaborator is injectable so the two paths can be tested without a
 * running herdr and without spawning processes to kill.
 */
export function resolveTickOwnership(input: {
  cwd: string;
  agentName: string;
  env: Record<string, string | undefined>;
  pid: number;
  now: Date;
  sessionFile?: string;
  listAgents?: (env: Record<string, string | undefined>) => HerdrAgentList;
  alive?: (pid: number) => boolean;
}): TickOwnership {
  const paneId = input.env["HERDR_PANE_ID"] ?? "";
  if (input.env["HERDR_ENV"] === "1" && paneId.length > 0) {
    const list = (input.listAgents ?? readHerdrAgents)(input.env);
    if (list.kind === "ok") return paneOwnership({ paneId, agentName: input.agentName, agents: list.agents });

    // Fail closed, but not forever. Under herdr this session is one pane of
    // possibly several in the fleet's directory, and an unproven identity is the
    // case this whole check exists for — so it does not tick. It is `unresolved`
    // rather than `declined` because herdr not answering says nothing about who
    // this pane is: the caller retries, and the moment herdr answers the real
    // orchestrator arms. Latching here would mean one CLI timeout stops the
    // fleet until a human notices and restarts the pane.
    return {
      kind: "unresolved",
      reason:
        `cannot yet prove this pane is the fleet agent "${input.agentName}" — ${list.problem} — not ticking until it can`,
    };
  }

  return claimTickOwner({
    cwd: input.cwd,
    pid: input.pid,
    ...(input.sessionFile === undefined ? {} : { sessionFile: input.sessionFile }),
    now: input.now,
    ...(input.alive === undefined ? {} : { alive: input.alive }),
  });
}

/**
 * Whether this tick sends, and why — the whole decision, with no clock, no
 * filesystem and no session in it. The interesting part of a heartbeat is the
 * precedence between "not armed", "channel down" and "already pending", and
 * that is worth being able to test without a session at all.
 *
 * The pause sentinel is deliberately NOT consulted. `pause` is the dispatch
 * daemon's flag — "stop claiming; in-flight work finishes" — and this heartbeat
 * drives a different brain: the supervising session whose duties (groom the
 * queue, drain escalations, report) are exactly the ones that stay useful while
 * dispatch is stopped. Honouring it here shipped once, when the tick-driven
 * session WAS the dispatcher; the day dispatch moved into the daemon, one flag
 * silencing both brains became a starvation bug: a paused fleet's orchestrator
 * could neither groom nor even say it was paused. The operator's lever for this
 * session is the arm marker — `disarm` stops ticks, and only the operator arms.
 *
 * `armed` and `channelOk` are the *satisfied* gates, not the files behind them:
 * a config with no `armedFile` passes the first, and one with no `accessFile`
 * passes the second. The fleet deploy always configures `accessFile` — an
 * orchestrator that can page nobody must not dispatch — so an unconfigured
 * channel gate means "this session is not the fleet", not "the check is off".
 */
export function tickDecision(input: {
  armed: boolean;
  channelOk: boolean;
  hasPending: boolean;
}): {
  send: boolean;
  reason: string;
} {
  if (!input.armed) return { send: false, reason: "not armed" };
  if (!input.channelOk) return { send: false, reason: "escalation channel down" };
  if (input.hasPending) return { send: false, reason: PENDING_REASON };
  return { send: true, reason: "armed, nothing pending" };
}

/**
 * The tick guard's decision, extracted pure from {@link armTickGuard} so it can
 * be pinned without a fake host: whether a turn's remaining tool calls should be
 * refused. Budget first, then a queued operator message; otherwise nothing is
 * blocked (#189).
 */
export function tickGuardDecision(input: {
  turnStartedAt: number | undefined;
  now: number;
  budgetMs: number;
  hasPending: boolean;
}): { block: true; reason: string } | undefined {
  if (input.turnStartedAt === undefined) return undefined;
  const elapsed = input.now - input.turnStartedAt;
  if (elapsed > input.budgetMs) {
    return {
      block: true,
      reason:
        `Blocked: this turn has run ${Math.round(elapsed / 1000)}s, past the ${Math.round(input.budgetMs / 1000)}s tick budget (#189). ` +
        `Finish the turn NOW with a short report. Park anything you were waiting on as a watch — ` +
        `omp-conductor decision open --question "..." --resolves-when pr-checks-green:<url> | pr-mergeable:<url> | pr-merged:<url> | rate-limit-reset:github — ` +
        `or hand it to a subagent; the daemon flags met conditions in your next tick digest.`,
    };
  }
  if (input.hasPending && elapsed > PENDING_MESSAGE_GRACE_MS) {
    return {
      block: true,
      reason:
        `Blocked: an operator message is queued behind this turn (#189). End the turn now and answer it — the person is waiting. ` +
        `Park any wait as a decision watch (--resolves-when) instead of polling.`,
    };
  }
  return undefined;
}

/**
 * Whether the Telegram bridge can still reach a person: a bot token, enabled,
 * with exactly one paired owner.
 *
 * Fail-closed, and every failure mode collapses to the same answer on purpose —
 * missing file, truncated write, hand-edit that dropped `enabled`, a second
 * chat id pasted in, the pairing revoked, or no token for the bot to send with.
 * Distinguishing them would only tempt a future reader into treating one of them
 * as benign, and none of them are: each one means a tier-2 escalation lands
 * nowhere.
 *
 * The token belongs in this gate rather than further down. Without one nothing
 * outbound works, so a tick would carry a delivery rule ordering a
 * `telegram_send` that cannot happen and an approval fallback naming the same
 * unusable tool — an unattended fleet dispatching with no way to page anybody,
 * which is the one thing the arm handshake exists to prevent. A missing token
 * fails the whole channel; the narrower approval preflight is reserved for a
 * bridge that can send but cannot ask.
 *
 * Re-read on every tick, and specifically *not* snapshotted at session start,
 * because of what this gate is actually about. Tier 2 is paged by conductor
 * itself: `escalate.ts` reads the same `.env` and calls `api.telegram.org`
 * directly, never through omp-telegram's bridge. So the question "can an
 * escalation still reach a person" is answered by the file as it stands now, and
 * an operator who fixes a missing token has fixed paging immediately — a startup
 * snapshot would keep a working fleet silent until somebody restarted the
 * session, which is a worse failure than the one it would prevent. Re-reading
 * also catches the reverse: a channel that goes away mid-session stops the
 * heartbeat on the next tick rather than days later.
 *
 * What the file cannot answer is whether *omp-telegram* holds a token. It binds
 * one in `startBot()` at session start and rebinds only on `/telegram token` and
 * `/telegram on`, so a token added out-of-band leaves the bridge's own tools —
 * `telegram_ask`, `telegram_send` — dead until one of those runs. That is real,
 * and it is deliberately not modelled here: this gate protects paging, the
 * README says to reload the bridge, and `omp-conductor status` shows the row.
 */
function channelIsUp(path: string): boolean {
  if (!hasBotToken(dirname(path))) return false;
  let parsed: unknown;
  try {
    parsed = JSON.parse(readFileSync(path, "utf8"));
  } catch {
    return false;
  }
  if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return false;
  const access = parsed as { readonly [key: string]: unknown };
  if (access["enabled"] !== true) return false;
  const allowFrom = access["allowFrom"];
  return Array.isArray(allowFrom) && allowFrom.length === 1;
}

/**
 * The operator's own prompt for this tick, re-read from the session cwd rather
 * than taken from the startup snapshot — for the reason {@link resolveTickScope}
 * and the channel gate re-read: the operator of a 24/7 session reconfigures it
 * out-of-band, and a heartbeat holding a startup copy would keep injecting last
 * week's prompt until somebody restarted the session.
 *
 * A re-read that succeeds owns the answer, "no `message` key any more" included:
 * deleting the override hands the prompt back to the shipped default. A re-read
 * that fails keeps the startup value — a mid-edit truncation, a file moved away
 * or a fault the validator collects must not stop the heartbeat, and must not
 * silently swap the operator's prompt for ours over a transient bad read.
 *
 * `intervalSeconds` is deliberately *not* re-read here: rescheduling a live
 * managed timer is a different change, so a period edit still needs a restart.
 */
function currentMessage(cwd: string, startup: TickConfig): string | undefined {
  const reread = readTickConfig(cwd);
  return reread.kind === "ok" ? reread.config.message : startup.message;
}

/**
 * The tick config fields honoured live, as they are on disk *now*, same contract as
 * {@link currentMessage}: a successful re-read owns the answer, a failed one
 * keeps the startup value.
 *
 * `setup host` restamps `armedFile` and `project` under a live pane — that is
 * how a fleet moves onto its per-project arm marker (#317). Reading the path
 * captured at session start instead meant the restamp silently killed the
 * heartbeat: `status` reads the file and reports `armed`, while the tick kept
 * checking the path the restamp had just replaced, skipped as "not armed" —
 * a reason that deliberately writes no stall marker — and the fleet stopped
 * ticking until somebody noticed and restarted the pane.
 *
 * `intervalSeconds` stays startup-only: rescheduling a live managed timer is a
 * different change and still needs a restart.
 */
function currentConfig(
  cwd: string,
  startup: TickConfig,
): { armedFile?: string; accessFile?: string; project?: string } {
  const reread = readTickConfig(cwd);
  const live = reread.kind === "ok" ? reread.config : startup;
  return {
    ...(live.armedFile === undefined ? {} : { armedFile: live.armedFile }),
    ...(live.accessFile === undefined ? {} : { accessFile: live.accessFile }),
    ...(live.project === undefined ? {} : { project: live.project }),
  };
}

/**
 * Marker writes are best-effort by construction. This runs inside the loop
 * whose whole job is to keep prompting a session; a read-only filesystem, a
 * full disk, or a cwd deleted out from under a long-lived process is not a
 * reason to stop doing that. Both directions log their own failure and carry
 * on — the error line stands as the record either way, and the marker is only
 * the copy an out-of-process watchdog can see.
 */
function writeStallMarker(pi: TickApi, cwd: string): void {
  const path = join(cwd, STALL_MARKER_FILE);
  try {
    writeFileSync(path, `${new Date().toISOString()} ${STALL_DIAGNOSIS}\n`);
  } catch (err) {
    pi.logger.error(`[omp-conductor] could not write ${path}: ${err instanceof Error ? err.message : String(err)}`);
  }
}

/**
 * Cleared by the first consumed tick, whether or not this process wrote it.
 * Recovery normally arrives as a *new* session — the wedged one was SIGTERM'd
 * and its transcript resumed — which starts with a zero counter and its
 * predecessor's marker on disk. Gating removal on this session's own count
 * would strand that file, and a stall marker nobody clears is a stall marker
 * nobody believes.
 */
function clearStallMarker(pi: TickApi, cwd: string): void {
  const path = join(cwd, STALL_MARKER_FILE);
  if (!existsSync(path)) return;
  try {
    rmSync(path, { force: true });
    pi.logger.info("[omp-conductor] orchestrator stall cleared: a tick was consumed");
  } catch (err) {
    pi.logger.error(`[omp-conductor] could not remove ${path}: ${err instanceof Error ? err.message : String(err)}`);
  }
}

/**
 * Best-effort poke that asks a live orchestrator heartbeat to fire soon.
 *
 * Same file shape as herdr `recover.sh`: one line `${iso} ${reason}`. Writers
 * (recover, the dispatch daemon on a false→true condition) never clear it —
 * only a successful tick send does, so a disarmed or channel-down fleet keeps
 * the request until gates pass.
 */
export function requestImmediateTick(cwd: string, reason: string): boolean {
  const path = join(cwd, TICK_REQUESTED_FILE);
  const label = reason.trim().length > 0 ? reason.trim() : "wake";
  try {
    writeFileSync(path, `${new Date().toISOString()} ${label}\n`);
    return true;
  } catch {
    return false;
  }
}

/** First non-empty token after the optional ISO stamp on a poke line. */
/**
 * The poke's first line verbatim, which is its identity.
 *
 * Both writers stamp an ISO timestamp (`${iso} ${reason}`), so two distinct
 * requests never share a line and a request that is still sitting there
 * unconsumed always does. That is what lets the mid-interval poll act once per
 * request instead of once per poll.
 */
function readTickRequestLine(cwd: string): string | undefined {
  const path = join(cwd, TICK_REQUESTED_FILE);
  if (!existsSync(path)) return undefined;
  let raw: string;
  try {
    raw = readFileSync(path, "utf8");
  } catch {
    return undefined;
  }
  const line = raw.split(/\r?\n/, 1)[0]?.trim() ?? "";
  return line.length === 0 ? undefined : line;
}

export function readTickRequestReason(cwd: string): string | undefined {
  const line = readTickRequestLine(cwd);
  if (line === undefined) return undefined;
  const parts = line.split(/\s+/);
  if (parts.length === 0) return undefined;
  // recover.sh writes `${iso} recover`; a bare reason is also accepted.
  if (parts.length >= 2 && Number.isFinite(Date.parse(parts[0]!))) {
    return parts.slice(1).join(" ");
  }
  return parts.join(" ");
}

/**
 * Best-effort, same posture as {@link clearStallMarker}. Leaving the file on a
 * failed unlink means the next successful send retries the clear; that is
 * preferable to treating a recover/condition poke as fire-and-forget when the
 * tick did land.
 */
function clearTickRequest(pi: TickApi, cwd: string): void {
  const path = join(cwd, TICK_REQUESTED_FILE);
  if (!existsSync(path)) return;
  try {
    rmSync(path, { force: true });
    pi.logger.info("[omp-conductor] tick request cleared: a tick was sent");
  } catch (err) {
    pi.logger.error(`[omp-conductor] could not remove ${path}: ${err instanceof Error ? err.message : String(err)}`);
  }
}

interface PendingLocalTick {
  id: string;
  projectName?: string;
  policy?: ReportingPolicy;
}

interface ActiveLocalTick extends PendingLocalTick {
  /** A person who writes during an autonomous run is awake by construction. */
  humanWaiting: boolean;
}

interface TelegramInterrupt {
  kind: "question" | "message";
  category: HeldNotice["category"];
  summary: string;
  detail: string;
  hasFiles: boolean;
  fingerprint: string;
}

function objectRecord(value: unknown): value is Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value);
}
function canonicalTelegramPayload(value: unknown): unknown {
  if (Array.isArray(value)) return value.map(canonicalTelegramPayload);
  if (!objectRecord(value)) return value;
  return Object.fromEntries(
    Object.keys(value)
      .sort()
      .map((key) => [key, canonicalTelegramPayload(value[key])]),
  );
}


function serialiseTelegramPayload(value: unknown): string {
  try {
    return JSON.stringify(canonicalTelegramPayload(value)) ?? String(value);
  } catch {
    return String(value);
  }
}

/**
 * A direct Telegram call, whatever surface it arrived on: the mounted tool, or
 * the same device written through `xd://`. Both spellings reach the same bot,
 * so every gate below reads them the same way.
 */
function directTelegramCall(
  toolName: string,
  input: Record<string, unknown>,
): { kind: TelegramInterrupt["kind"]; payload: Record<string, unknown>; reaction: boolean } | undefined {
  let kind: TelegramInterrupt["kind"];
  let reaction: boolean;
  let payload: Record<string, unknown> = input;

  if (toolName === TELEGRAM_APPROVAL_TOOL) {
    kind = "question";
    reaction = false;
  } else if (toolName === "telegram_send" || toolName === "telegram_react") {
    kind = "message";
    reaction = toolName === "telegram_react";
  } else if (toolName === "write") {
    const path = input["path"];
    if (
      path !== "xd://telegram_ask" &&
      path !== "xd://telegram_send" &&
      path !== "xd://telegram_react"
    ) {
      return undefined;
    }
    kind = path === "xd://telegram_ask" ? "question" : "message";
    reaction = path === "xd://telegram_react";
    const content = input["content"];
    if (typeof content === "string") {
      try {
        const parsed: unknown = JSON.parse(content);
        payload = objectRecord(parsed) ? parsed : { content };
      } catch {
        payload = { content };
      }
    }
  } else {
    return undefined;
  }
  return { kind, payload, reaction };
}

/** A named target that is really named: an empty string targets nothing. */
function targetNamed(value: unknown): boolean {
  if (typeof value === "number") return Number.isFinite(value);
  return typeof value === "string" && value.trim() !== "";
}

/**
 * The one `telegram_send` shape that silently leaves a forum topic: `thread_id`
 * falls back to the active topic *only while `chat_id` is omitted*, so naming
 * the chat alone delivers into the main chat instead of the thread the operator
 * wrote in (#366). This package cannot rewrite another plugin's arguments, so
 * the unsafe shape is refused with both safe ones spelled out — a reply the
 * operator never sees in the thread they asked in is indistinguishable from
 * silence, and it leaks the answer to everyone reading the main chat.
 *
 * Returns `undefined` for every other shape, including a call that names both.
 */
export function telegramTopicRoutingRefusal(payload: Record<string, unknown>): string | undefined {
  if (!targetNamed(payload["chat_id"])) return undefined;
  if (targetNamed(payload["thread_id"]) || targetNamed(payload["message_thread_id"])) return undefined;
  return (
    "Blocked: this project's Telegram is topic-routed, and thread_id defaults to the active topic only while chat_id is omitted — " +
    "naming chat_id alone delivers into the main chat instead of the topic the message arrived in (#366). " +
    "Nothing was sent. Answer in the active message's own topic by omitting BOTH chat_id and thread_id, or pass both together. " +
    'On a locally injected tick there is no active topic to keep: run `omp-conductor message --text "<the message>"`, ' +
    "which resolves this project's own chat and topic from config."
  );
}

/**
 * The caller's declared interrupt category for one Telegram call, when it names
 * a real one (#347). The tick prompt tells an autonomous turn which categories
 * may interrupt under this scope — a condition that stops the fleet is
 * `fleet-stopped`, a decision the operator must make is `tier2` — and the
 * caller tags the call with `category` so the admission gate below runs that
 * category through the same policy as a daemon escalation. Anything that is
 * not a known category falls back to the historical mapping in
 * {@link telegramInterruptFromTool}: a question is `decision-needed`, an
 * ordinary send is `material`.
 */
function declaredTelegramCategory(payload: Record<string, unknown>): InterruptCategory | undefined {
  const value = payload["category"];
  if (typeof value !== "string") return undefined;
  return (INTERRUPT_CATEGORIES as readonly string[]).includes(value)
    ? (value as InterruptCategory)
    : undefined;
}

/**
 * One Telegram call, whatever surface it arrived on, classified enough for the
 * availability gate: its kind (a question that wants an answer, or a message)
 * plus the interrupt category the caller declared — defaulting to
 * `decision-needed` for questions and `material` for messages when the caller
 * declares none.
 */
function telegramInterruptFromTool(
  toolName: string,
  input: Record<string, unknown>,
): TelegramInterrupt | undefined {
  const call = directTelegramCall(toolName, input);
  if (call === undefined) return undefined;
  let kind = call.kind;
  const payload = call.payload;
  if (
    kind === "message" &&
    typeof payload["text"] === "string" &&
    /^\s*QUESTION:\s/i.test(payload["text"])
  ) {
    // The package floor uses this explicit fallback when telegram_ask is not
    // mounted. Without the marker a plain send remains a material update.
    kind = "question";
  }

  let detail = "";
  if (kind === "question" && Array.isArray(payload["questions"])) {
    detail = payload["questions"]
      .map((value, index) => {
        if (!objectRecord(value)) return `Question ${index + 1}: ${serialiseTelegramPayload(value)}`;
        const question = typeof value["question"] === "string" ? value["question"] : "Question";
        const options = Array.isArray(value["options"])
          ? value["options"]
              .map((option) =>
                objectRecord(option) && typeof option["label"] === "string"
                  ? `${option["label"]}${
                      typeof option["description"] === "string" && option["description"] !== ""
                        ? ` — ${option["description"]}`
                        : ""
                    }`
                  : serialiseTelegramPayload(option),
              )
              .join("; ")
          : "";
        return options === "" ? question : `${question}\nOptions: ${options}`;
      })
      .join("\n\n");
  } else if (typeof payload["text"] === "string") {
    detail = payload["text"];
  }
  if (detail.trim() === "") detail = serialiseTelegramPayload(payload);

  const flat = detail.replace(/\s+/g, " ").trim();
  const fallback = kind === "question" ? "Operator question" : "Telegram message";
  const summary = flat === "" ? fallback : flat.length <= 180 ? flat : `${flat.slice(0, 179)}…`;
  const canonical = serialiseTelegramPayload({ kind, payload });
  return {
    kind,
    category: declaredTelegramCategory(payload) ?? (kind === "question" ? "decision-needed" : "material"),
    summary,
    hasFiles: Array.isArray(payload["files"]) && payload["files"].length > 0,
    detail,
    fingerprint: createHash("sha256").update(canonical).digest("hex"),
  };
}

type TelegramInterruptBlock = { block: true; reason: string };

/**
 * Topic routing is checked for **every** direct Telegram *delivery* this
 * session makes, autonomous or not: the incident was a reply to a waiting
 * human, which the autonomous gate below deliberately exempts (#366). A
 * flat-chat project has no thread to lose and is left exactly as it was, and a
 * reaction addresses an exact `message_id` — it carries no thread argument at
 * all, so there is nothing there to get wrong.
 */
function telegramRoutingBlock(
  configuredProject: string | undefined,
  event: { toolName: string; input: Record<string, unknown> },
): TelegramInterruptBlock | undefined {
  const call = directTelegramCall(event.toolName, event.input);
  if (call === undefined || call.reaction) return undefined;
  let topicRouted: boolean;
  try {
    topicRouted = findProject(loadConfig(), configuredProject).escalation.telegramTopicId !== undefined;
  } catch {
    // An unreadable or ambiguous config says nothing about topics. The
    // autonomous gate already fails closed on it for locally injected ticks,
    // and a human-waiting reply must not become unanswerable because of it.
    return undefined;
  }
  if (!topicRouted) return undefined;
  const reason = telegramTopicRoutingRefusal(call.payload);
  return reason === undefined ? undefined : { block: true, reason };
}

/**
 * A locally injected tick is an autonomous actor. Its direct Telegram calls
 * therefore pass through the same category and availability decision as daemon
 * escalations. Human-originated runs, including a message steered into a tick,
 * are exempt: the operator is already present and waiting for the reply.
 */
function autonomousTelegramInterruptBlock(
  pi: TickApi,
  session: TickSession,
  event: { toolName: string; input: Record<string, unknown> },
): TelegramInterruptBlock | undefined {
  const active = session.activeLocalTick;
  if (active === undefined || active.humanWaiting) return undefined;
  const interrupt = telegramInterruptFromTool(event.toolName, event.input);
  if (interrupt === undefined) return undefined;
  // The raw ask tool would wait for the operator for as long as the answer
  // takes; on an autonomous tick that wait must be bounded and its timeout
  // outcome declared (#438). Divert before the policy read, because the
  // diversion does not depend on availability — the incident was an ask the
  // policy *admitted*: the gate let telegram_ask through at 01:01Z and the loop
  // stopped for six hours. Nothing is persisted here: the bounded surface
  // records the row and the delivery when the model actually asks, and a
  // gate-side duplicate would record questions that were never asked.
  const directAsk =
    event.toolName === TELEGRAM_APPROVAL_TOOL ||
    (event.toolName === "write" && event.input["path"] === `xd://${TELEGRAM_APPROVAL_TOOL}`);
  if (directAsk) {
    return { block: true, reason: ASK_DIVERSION_REASON };
  }
  if (active.projectName === undefined) {
    return {
      block: true,
      reason:
        "Operator availability policy is unavailable for this autonomous tick, so direct Telegram is blocked fail-closed. " +
        "Repair the conductor config; do not retry through another Telegram path.",
    };
  }

  let policy = active.policy;
  let configReadable = true;
  try {
    policy = findProject(loadConfig(), active.projectName).reporting;
  } catch (err) {
    configReadable = false;
    pi.logger.error(
      `[omp-conductor] operator availability config unreadable during Telegram call; blocking fail-closed: ${
        err instanceof Error ? err.message : String(err)
      }`,
    );
  }

  const at = Date.now();
  const disposition = configReadable
    ? interruptDisposition(policy, interrupt.category, at)
    : "availability";
  if (disposition === "interrupt") return undefined;
  const release =
    disposition === "availability"
      ? "the availability window opens or the daily digest runs"
      : "the daily digest runs";
  if (interrupt.hasFiles) {
    return {
      block: true,
      reason:
        `Operator policy forbids this autonomous Telegram ${interrupt.kind} until ${release}, ` +
        "and attachment-bearing Telegram sends cannot be durably replayed. Nothing was sent or held. " +
        "Do not retry through another Telegram path while the policy forbids delivery.",
    };
  }

  const key = `${active.projectName}:autonomous-telegram:${active.id}:${interrupt.fingerprint}`;
  try {
    const store = openStore(dbPath());
    try {
      if (!store.wasNotified(key)) {
        store.addHeldNotice({
          id: heldNoticeId(key),
          project: active.projectName,
          category: interrupt.category,
          summary: interrupt.summary,
          detail: interrupt.detail,
          createdAt: at,
          ...(disposition === "availability" ? { releaseOnAvailable: true } : {}),
        });
        store.markNotified(key);
      }
    } finally {
      store.close();
    }
  } catch (err) {
    const reason = err instanceof Error ? err.message : String(err);
    pi.logger.error(`[omp-conductor] could not persist held Telegram ${interrupt.kind}: ${reason}`);
    return {
      block: true,
      reason:
        `Operator policy forbids this autonomous Telegram ${interrupt.kind}, and the durable hold failed: ${reason}. ` +
        "Do not retry through another Telegram path.",
    };
  }

  return {
    block: true,
    reason:
      `Operator policy durably held this autonomous Telegram ${interrupt.kind} until ${release}. ` +
      "Do not retry through another Telegram path.",
  };
}

/** Everything one tick remembers for the next. */
interface TickSession {
  /**
   * Whether the reporting-scope fallback has been logged. Without it, a host
   * with no conductor config would repeat the same line about the same missing
   * file every interval, for as long as the session lives.
   */
  scopeFallbackLogged: boolean;
  /**
   * Whether the missing {@link TELEGRAM_APPROVAL_TOOL} has been logged. Latched
   * for {@link TickSession.scopeFallbackLogged}'s reason and then some: on the
   * surface #114 reports, the tool is absent on *every* locally injected tick,
   * so an unguarded line would be one error per interval — 144 a day at the
   * ten-minute heartbeat this fleet runs — which is how a real fault becomes
   * background noise.
   */
  approvalToolMissingLogged: boolean;
  /**
   * Whether omp-telegram could have bound a bot token when this session started.
   * False means its `telegram_ask` / `telegram_send` are dead for this session
   * however good `access.json` looks now, because the bridge resolves its token
   * once in `startBot()`. Defaults true so a session that never reaches
   * `session_start` — every unit test that calls `tick()` directly — keeps the
   * behaviour the access file describes.
   */
  bridgeTokenAtStart: boolean;
  /** Consecutive {@link PENDING_REASON} skips — see {@link STALL_MARKER_FILE}. */
  pendingSkips: number;
  /** Ticks sent as follow-ups but not yet observed by the agent loop. */
  pendingLocalTicks: PendingLocalTick[];
  /**
   * Whether the stranded shared arm marker has been named. Latched for
   * {@link TickSession.scopeFallbackLogged}'s reason: a host that gained a
   * second project without re-running setup would otherwise repeat the same
   * line every interval, forever.
   */
  legacyArmLogged: boolean;
  /** The local tick whose agent loop is currently running, if any. */
  activeLocalTick?: ActiveLocalTick;
}

/**
 * Resolve the project an {@link ASK_TOOL} call routes to, re-reading the live
 * tick config so a restamp (un-stamped → stamped, or project A → B) binds the
 * current routing rather than the session-start stamp the ask session closed
 * over. Only the routing is live; the ceiling and turn budget stay startup-only
 * (the same contract `budgetSeconds` already has). Mirrors
 * {@link resolveTickScope}'s decision about when a heartbeat may advertise the
 * ask surface: a project resolves exactly when a tick would append
 * {@link TICK_ASK_RULE}, so the model-visible tool and the prompt line never
 * disagree. Returns the resolved project for delivery, or the failure the tool
 * reports verbatim.
 */
function resolveAskProject(
  cwd: string,
  startup: TickConfig,
): { kind: "ok"; project: ProjectConfig } | { kind: "error"; problem: string } {
  const live = currentConfig(cwd, startup);
  try {
    return { kind: "ok", project: findProject(loadConfig(), live.project) };
  } catch (err) {
    return { kind: "error", problem: err instanceof Error ? err.message : String(err) };
  }
}

/**
 * Reconcile the bounded ask's model-visible presentation with the current
 * routing contract. Registered at extension-factory time with
 * `defaultInactive` (#594), {@link ASK_TOOL} is NOT in the initial active set
 * — this activates it only once (and for as long as) the live tick config
 * resolves a project, and deactivates it when resolution degrades (unreadable
 * config, un-stamped multi-project). That keeps an unusable tool schema out of
 * the model's view rather than merely omitting the prompt line that names it —
 * and a restamp (un-stamped → stamped) brings it back on the next resolved
 * beat. Latched by membership: `setActiveTools` fires only on a state flip,
 * never per tick.
 *
 * Returns whether the requested state now actually holds in the live set
 * (#683). `true` only after the `setActiveTools` reconciliation is awaited AND
 * the live membership read confirms it — an already-reconciled beat is
 * confirmed by the membership read alone. A call that rejected, or one that
 * resolved without the tool landing in the live set, returns `false`; the
 * caller must then treat the beat as degraded and never advertise the ask
 * surface. A flip failure is logged, never fatal — the tool then stays in its
 * prior state, and executing it still fails closed when it cannot route.
 */
async function ensureAskSurface(pi: TickApi, resolvable: boolean): Promise<boolean> {
  // Both operator-question surfaces flip together (#947): they route through the
  // same resolution and fail closed the same way, so a beat that can carry one
  // can carry the other, and a degraded beat must hide both — an ask surface that
  // is half-present is a model choosing between a working tool and a broken one.
  const tools = [ASK_TOOL, QUESTIONNAIRE_TOOL];
  const current = pi.getActiveTools();
  const present = tools.every((tool) => current.includes(tool));
  const absent = tools.every((tool) => !current.includes(tool));
  if (resolvable ? present : absent) return true;
  try {
    await pi.setActiveTools(
      resolvable
        ? [...current.filter((name) => !tools.includes(name)), ...tools]
        : current.filter((name) => !tools.includes(name)),
    );
  } catch (err) {
    pi.logger.error(
      `[omp-conductor] could not ${resolvable ? "activate" : "deactivate"} ${tools.join(" / ")}: ${
        err instanceof Error ? err.message : String(err)
      }`,
    );
    return false;
  }
  // An awaited call is not yet confirmation: the live membership read after the
  // reconciliation is what proves the tools are really model-visible.
  const after = pi.getActiveTools();
  return tools.every((tool) => after.includes(tool) === resolvable);
}

/**
 * One tick: gather the three facts, ask `tickDecision`, log the reason either
 * way. Skips are deliberately silent in the UI — a disarmed fleet would
 * otherwise emit a notification every interval, forever.
 */
async function tick(
  pi: TickApi,
  ctx: TickContext,
  config: TickConfig,
  session: TickSession,
  toSpecTrackerSeam: ToSpecTrackerSeam | undefined,
  armSeam: ArmConsumerSeam | undefined,
): Promise<void> {
  const live = currentConfig(ctx.cwd, config);
  const arm = live.armedFile === undefined ? undefined : resolveArmState(live.armedFile, live.project);
  if (arm?.legacy === "stranded" && !session.legacyArmLogged) {
    session.legacyArmLogged = true;
    pi.logger.error(
      `[omp-conductor] ${legacyArmedMarkerPath()}: ${LEGACY_ARM_MARKER_DETAIL} — this heartbeat stays disarmed`,
    );
  }
  // An unanswered ceremony that passed its window is a dead ceremony, not a
  // silent one (#1061): the tick that notices tells the operator why and how
  // to re-run it, on the surface the challenge was sent to. Deliberately
  // before the tick gate — a disarmed fleet is exactly the state an
  // unanswered challenge leaves, and its death notice must not wait for an
  // arm to be heard. The scan is synchronous so an idle host (no expired
  // records) costs the tick nothing; only a send is awaited.
  const armNow = (armSeam?.now ?? Date.now)();
  for (const expired of expiredArmChallenges(live.project, armNow)) {
    const surface: ArmNoticeSurface = { accessFile: live.accessFile, project: live.project };
    const send = armSeam?.notify ?? sendArmNotice;
    try {
      await send(armExpiryNotice(expired, live.project), surface);
    } catch (err) {
      // Keep the record: the next heartbeat retries, and `doctor` still
      // reports it.
      pi.logger.error(
        `[omp-conductor] arm expiry notice not delivered: ${err instanceof Error ? err.message : String(err)}`,
      );
      continue;
    }
    clearArmTransaction(expired.key, expired.id);
  }
  const decision = tickDecision({
    armed: arm === undefined || arm.armed,
    channelOk: live.accessFile === undefined || channelIsUp(live.accessFile),
    hasPending: ctx.hasPendingMessages(),
  });

  if (!decision.send) {
    pi.logger.info(`[omp-conductor] tick skipped: ${decision.reason}`, { reason: decision.reason });
    // Only a coalesced skip counts. "Not armed" and "channel down" say nothing
    // about the queue — they are gates the operator or the bridge closed, and
    // the session behind them may be perfectly awake.
    if (decision.reason !== PENDING_REASON) return;
    session.pendingSkips += 1;
    // Only the crossing writes. A later skip refreshing the timestamp would
    // keep moving "stalled since" forward, and how long it has been wedged is
    // the fact the file is read for.
    if (session.pendingSkips === STALL_TICKS) {
      pi.logger.error(`[omp-conductor] orchestrator stalled: ${STALL_DIAGNOSIS}; see ${STALL_MARKER_FILE}`, {
        reason: decision.reason,
        pendingSkips: session.pendingSkips,
      });
      writeStallMarker(pi, ctx.cwd);
    }
    return;
  }

  // Floor refresh is independent of which prompt we send: a custom message still
  // expects ORCHESTRATOR.md / AGENTS.md to track the installed package.
  refreshComposedBriefBestEffort(live.project);

  const scope = resolveTickScope(live.project);
  // The transport contract, read once from the same file at the same moment so
  // the approval line, the delivery rule and the narration line cannot disagree
  // (#169, #179). No access file means no fleet channel to judge; a session
  // with no bound bridge token means omp-telegram posts nothing, so both halves
  // fail toward "not delivered" — the durable outbox instruction — rather than
  // asserting a delivery the bridge cannot make.
  const approval =
    live.accessFile === undefined
      ? undefined
      : session.bridgeTokenAtStart
        ? readApprovalSurface(live.accessFile)
        : ({
            kind: "missing",
            reason:
              `${TELEGRAM_APPROVAL_TOOL} unavailable on local ticks: omp-telegram had no bot token when this ` +
              "session started, so it bound none and its tools stay dead however complete the access file looks " +
              "now — run `/telegram on` in this session, or restart it, to rebind the bridge",
          } as const);
  const profile =
    live.accessFile === undefined || !session.bridgeTokenAtStart
      ? undefined
      : readDaemonProfile(live.accessFile);

  // A configured message owns the ordinary reporting and delivery clauses.
  // Mechanical evidence is different: release-policy drift and repeated
  // operational friction must not disappear because an operator customized the
  // ordinary heartbeat wording.
  let content = currentMessage(ctx.cwd, config);
  if (content === undefined) {
    if (scope.fallback !== undefined && !session.scopeFallbackLogged) {
      session.scopeFallbackLogged = true;
      pi.logger.info(`[omp-conductor] tick reporting scope: using ${DEFAULT_REPORT_SCOPE} — ${scope.fallback}`);
    }
    const bridged = approval?.kind === "ready" && approval.notifyMode === "always" && profile?.kind === "interactive";
    let reportingConstraint = TICK_SCOPE_CONSTRAINTS[scope.scope];
    if (scope.projectName !== undefined) {
      const policy = scope.policy;
      const digestPolicy = policy?.digest ?? DEFAULT_REPORT_POLICY.digest;
      const cadence = digestPolicy.cadence;
      const at = Date.now();
      const store = openStore(dbPath());
      try {
        const lastKey = store.lastDigestDedupeKey(scope.projectName);
        const lastDay = lastKey === undefined ? undefined : lastKey.slice("digest:".length);
        const due = digestDue(policy ?? DEFAULT_REPORT_POLICY, lastDay, at);
        // Reserve availability-held rows before reading them. The same SQLite
        // transaction releases pending catch-ups; daemon enqueue attempts then
        // observe the lease, closing both orders of the snapshot-to-handoff race.
        const availabilityReserved =
          due &&
          store.reserveAvailabilityDigest(
            scope.projectName,
            localDayKey(at, digestPolicy.timezone ?? policy?.availability?.timezone),
            at,
            at +
              Math.max(
                MIN_AVAILABILITY_DIGEST_LEASE_MS,
                (config.budgetSeconds ?? DEFAULT_TICK_BUDGET_SECONDS) * 1_000 + 5 * 60_000,
              ),
          );
        const backlog = store.digestBacklog(scope.projectName);
        const visibleBacklog: DigestBacklog = availabilityReserved
          ? backlog
          : {
              ...backlog,
              heldNoticeCount: Math.max(
                0,
                backlog.heldNoticeCount - (backlog.availabilityHeldNoticeCount ?? 0),
              ),
            };
        reportingConstraint = tickReportingConstraint(
          policy,
          {
            due,
            cadence,
            at: digestPolicy.at,
            timezone: digestPolicy.timezone,
          },
          visibleBacklog.heldNoticeCount,
        );
        if (
          due &&
          (visibleBacklog.materialCount > 0 || visibleBacklog.heldNoticeCount > 0)
        ) {
          const notices = store
            .undigestedNotices(scope.projectName)
            .filter((notice) => notice.releaseOnAvailable !== true || availabilityReserved);
          reportingConstraint +=
            `\n${formatDigestLedger(
              store.undigestedMaterialEvents(scope.projectName),
              notices,
              visibleBacklog,
            )}`;
        }
      } finally {
        store.close();
      }
    }
    content = `${defaultTickMessage(
      new Date(),
      scope.briefPath,
      scope.policyPath,
      existsSync(sharedPolicyPath()) ? sharedPolicyPath() : undefined,
    )}\n${reportingConstraint}\n${bridged ? TICK_DELIVERY_RULE_BRIDGED : TICK_DELIVERY_RULE}`;
  }
  if (scope.projectName !== undefined) {
    // Unlike ordinary reporting prose, this is a mechanical clock reading and
    // survives a custom heartbeat message. The model never infers whether the
    // operator can be interrupted. The bounded-ask rule is the same class
    // (#438): even a mounted telegram_ask is refused on a locally injected
    // tick — it would wait unbounded — so the tick names the replacement
    // surface up front, custom message included.
    content = `${content}\n${availabilityPrompt(scope.policy, Date.now())}`;
    // The ask rule names the bounded surface this session can call. It is
    // guarded on a resolved project for the same reason {@link ASK_TOOL} is:
    // the tool needs the project to file the decision row against and resolve
    // the delivery target. On a degraded heartbeat (`resolveTickScope` falls
    // back to DEFAULT_REPORT_SCOPE with no projectName — an unreadable config
    // or a legacy un-stamped multi-project tick) the tool cannot route, so the
    // prompt must not advertise it. There is no prose to fall back to — the CLI
    // `omp-conductor message` fallback is deleted (#594), and a registration
    // failure is a bug to fix, not a surface to name. So the degraded beat
    // simply names nothing about asking, rather than naming a tool that would
    // fail closed. The `telegram_ask` refusal stays: it would wait for the
    // operator for as long as the answer takes (#438).
    //
    // The rule is appended only after the ask surface is CONFIRMED
    // model-visible (#683): a `defaultInactive` registration is not in the
    // live set until this beat's `setActiveTools` reconciliation lands, so
    // appending the rule and then firing the reconciliation off in the
    // background rendered a promise of a tool the next turn might not carry.
    // The reconciliation is awaited and verified by live membership before the
    // rule is appended; when it cannot be confirmed, this beat takes the same
    // degraded branch as an unresolvable project — no rule, tool kept out of
    // the live set, never the deleted CLI prose fallback — so a tick can never
    // advertise an ask surface that may still be inactive.
    const activated = await ensureAskSurface(pi, true);
    if (activated) {
      content = `${content}\n${TICK_ASK_RULE}`;
    } else {
      // Keeps the tool out of the model-visible set on this beat, matching the
      // degraded-project branch below: 'not named in the prompt' must not
      // diverge from 'not callable' (#644).
      await ensureAskSurface(pi, false);
    }
  } else {
    // Degraded heartbeat: no project to route to, so the bounded tool must not
    // be model-visible (omitting the prompt line is not enough while the
    // registration still exposed the schema). A later restamp brings it back on
    // the next resolved beat.
    await ensureAskSurface(pi, false);
  }
  let frictionStore: Store | undefined;
  let frictionSignals: FrictionSignal[] = [];
  const now = Date.now();
  if (scope.projectName !== undefined) {
    const drift = releaseDriftDigestLine(scope.projectName);
    if (drift !== undefined) content = `${content}\n${drift}`;
    // The shared-host gate's refusals are reported on their own line, apart
    // from release drift (#562): a worker being stopped by the interlock is
    // the guard working, never a release-policy divergence, and it must not
    // move the drift counter or borrow the release-policy heading.
    const guard = sharedHostGuardDigestLine(scope.projectName);
    if (guard !== undefined) content = `${content}\n${guard}`;
    if (existsSync(dbPath())) {
      try {
        frictionStore = openStore(dbPath());
        frictionSignals = frictionStore.pendingFriction(
          scope.projectName,
          now - FRICTION_LOOKBACK_MS,
          FRICTION_MIN_OBSERVATIONS,
          now - FRICTION_COOLDOWN_MS,
        );
        if (frictionSignals.length > 0) content = `${content}\n${formatFrictionDigest(frictionSignals)}`;
        // Policy edits the session made on its own authority, read back before
        // they reach the operator (Phase 4). Placed directly after the friction
        // digest and before the decision rows because that is the order of the
        // Learning loop itself: the evidence that provokes an amendment, then
        // the mechanical ones already applied and owed a report, then the
        // judgment ones still waiting on an answer. The scan reaches past one
        // digest's worth of backlog on purpose — see AMENDMENT_AUDIT_SCAN_LIMIT.
        const amendments = formatAmendmentAudit(
          frictionStore.undigestedMaterialEvents(scope.projectName, AMENDMENT_AUDIT_SCAN_LIMIT),
        );
        if (amendments !== undefined) content = `${content}\n${amendments}`;
        // What the session still owes its operator, read from the ledger rather
        // than from what it remembers asking (#136). Appended every tick,
        // because the whole failure was a question surviving in context only.
        const decisions = formatDecisionDigest(frictionStore.openDecisions(scope.projectName), now);
        if (decisions.length > 0) content = `${content}\n${decisions}`;
        // A host running two different releases at once, from the row the
        // dispatch pass recorded (#919). Read only — probing the three surfaces
        // here spawns three children per tick and took this file's own suite
        // from 8.4s to 83.4s, which is why the observation is recorded. Only a
        // proven mismatch appears: an absent surface, a `local:` herdr link and
        // an unverifiable pin are steady states someone chose, and a warning
        // repeated every fifteen minutes trains an operator to ignore the line
        // that matters. `doctor` keeps that nuance.
        const surfaces = installSurfaceMismatch(frictionStore.installSurfaces());
        if (surfaces !== undefined) content = `${content}\n${surfaces}`;
        // What the daemon already fixed, so the session stops re-deriving that
        // paragraph on every tick (#132). Two intervals wide rather than one: a
        // tick that ran long must not drop the window it was meant to report.
        // The issue history answers which of those rows are still news: a
        // recovery whose issue has since moved on is dropped inside the digest
        // (#711). The project name and store are captured consts so the lookup
        // callback keeps the flow narrowing the direct calls in this block rely
        // on.
        const projectName = scope.projectName;
        const store = frictionStore;
        const recovered = recoveryDigestLine(
          store.recoveredSince(projectName, now - 2 * config.intervalSeconds * 1_000),
          {
            runsForIssue: (issue) => store.runsForIssue(projectName, issue),
          },
        );
        if (recovered !== undefined) content = `${content}\n${recovered}`;
        // The dispatch pass already persists the numbers that answer "is the
        // queue running dry", so this reads the store rather than asking the
        // tracker (#181). A failed config read skips the line rather than
        // wedging the tick.
        try {
          // Named: this block is already inside `scope.projectName !== undefined`,
          // so the resolved name is in hand and an un-named lookup would refuse
          // to guess on a host with a second project.
          const project = findProject(loadConfig(), scope.projectName);
          const groomBelow = project.groomBelow ?? DEFAULT_GROOM_BELOW;
          const dispatch = frictionStore.latestDispatch(scope.projectName);
          const grooming = store.groomingVerdicts(projectName);
          // #848: "the queue is empty / running dry" is the tracker's word, not
          // the last dispatch pass's echo. One open-issue snapshot per tick,
          // read live through the authoritative tracker surface, overlaid with
          // the label_ops projection exactly as the dispatch pass judges
          // eligibility — so a promotion, unblock or projection write after the
          // last dispatch pass is visible on the very next tick even while a
          // drain holds claiming. If the tracker read fails, the digest falls
          // back to explicitly dated wording, never a present-tense empty
          // claim.
          //
          // The read is made only for dispatch rows the digest could render:
          // a healthy row (routed at/above a numeric grooming threshold)
          // emits no queue line at all, so no freshness check is spent on it
          // and no stale claim can leak from it. A zero-ready row (or any row
          // below the trigger) is exactly where the old wording lied, so every
          // such tick reads the live queue. Under `groomBelow: "always"`
          // (#988) the duty never clears on volume, so every dispatch row
          // reads the queue.
          const needsQueueRead =
            dispatch !== undefined &&
            (dispatch.ready === 0 || dispatch.routed === 0 || groomingDue(dispatch.routed, groomBelow));
          let queueObservation: QueueObservation | undefined;
          if (toSpecTrackerSeam !== undefined && needsQueueRead) {
            try {
              const open = await toSpecTrackerSeam.listOpenIssues(project);
              const effective = open.map((issue) => {
                const pending = store.pendingLabelOpsFor(projectName, issue.number);
                return pending.length === 0
                  ? issue
                  : { ...issue, labels: effectiveLabels(issue.labels, pending) };
              });
              queueObservation = {
                queued: effective.filter((issue) => issue.labels.includes(project.queueLabel)).length,
                observedAt: now,
              };
            } catch {
              // No current-queue evidence this tick: the digest below renders
              // the dispatch row as dated history instead of claiming empty.
            }
          }
          const queue = queueDigestLine(
            dispatch,
            project.queueLabel,
            project.routing.labelPrefix,
            groomBelow,
            grooming,
            queueObservation,
            // One clock for the inventory and the audit below: the digest must
            // never call a verdict durable that the daemon's own selection is
            // about to re-groom (#887).
            now,
          );
          if (queue !== undefined) content = `${content}\n${queue}`;
          // The Duty-2 audit that replaced #777's launch block (#1041/#1040).
          // The daemon promotes unattended, so what the tick owes is review,
          // and the window is two intervals wide for the same reason the
          // recovery digest's is: a tick that ran long must not drop the
          // promotions it was meant to audit. A tick that reports nothing
          // renders nothing — the store, not this call site, decides whether
          // there is anything to say.
          const auditSince = now - 2 * config.intervalSeconds * 1_000;
          const audit = formatPromotionAudit({
            promotions: store.promotionsSince(projectName, auditSince),
            grooming,
            project: project.name,
            tracker: project.tracker.repo,
            queueLabel: project.queueLabel,
            since: auditSince,
          });
          if (audit !== undefined) content = `${content}\n${audit}`;
          // Pending intake is the same class of standing block as the friction
          // and decisions read-outs: a store-backed duty the orchestrator must
          // not derive from memory. The store answers, the prompt instructs.
          const pendingIntake = formatPendingIntake(frictionStore.pendingIntake(scope.projectName), {
            tracker: project.tracker.repo,
            queueLabel: project.queueLabel,
            labelPrefix: project.routing.labelPrefix,
          });
          if (pendingIntake !== undefined) content = `${content}\n${pendingIntake}`;
        } catch {
          // unreadable config: no queue digest, promotion audit, or
          // pending-intake block this tick
        }
      } catch (err) {
        frictionStore?.close();
        frictionStore = undefined;
        pi.logger.error(
          `[omp-conductor] friction digest unavailable: ${err instanceof Error ? err.message : String(err)}`,
        );
      }
    }
  }

  // The floor's approval primitive, decided from configuration rather than from
  // the live mounted set — and that is a correction, not a shortcut. The first
  // attempt at this read `pi.getActiveTools()` here, which is wrong at exactly
  // this point in the lifecycle: omp-telegram mounts `telegram_ask` in
  // `before_agent_start` and takes it away again in `agent_end`
  // (`restorePromptTools`), so it exists only *during* a turn. This runs between
  // turns, composing the prompt that is about to start one, so the live set
  // never contains the tool — a correctly configured fleet would have been told
  // the approval was unavailable on every single tick, and the error line meant
  // to flag a real fault would have fired 144 times a day saying nothing.
  //
  // What decides the mounting is knowable in advance and is shared with the
  // status row, so the two cannot disagree: a locally injected tick has no
  // `<telegram-message>` wrapper, so omp-telegram can only resolve a target
  // through `notifyTarget()` — `notifyMode` plus a destination, both in the
  // access file. Appended last, after the friction digest: the digest is what
  // provokes an amendment, so the sentence saying the amendment cannot be
  // approved here is the one that should read last.
  //
  // No access file configured means no fleet channel to judge, so nothing is
  // claimed: the channel gate above already treats that as "not the fleet".
  //
  // Two facts have to hold and the access file carries only one. It says whether
  // a destination would resolve; it cannot say whether the bridge holds a token
  // to resolve it with, because omp-telegram binds that once in `startBot()`. A
  // token written out-of-band after this session started opens the channel gate
  // — conductor pages tier 2 itself, so that much is honest — while leaving
  // `telegram_ask` and `telegram_send` dead until `/telegram on`. Trusting the
  // file alone there puts the tick straight back to mandating a call its surface
  // cannot make, which is #114 exactly.
  if (approval?.kind === "missing") {
    content = `${content}\n${TICK_APPROVAL_UNAVAILABLE_RULE}`;
    if (!session.approvalToolMissingLogged) {
      session.approvalToolMissingLogged = true;
      // Error level, and once: `status` reported `telegram ok (@tbcoder_bot;
      // inbound configured)` throughout the incident, so nothing else told the
      // operator the approval contract was unsatisfiable. A line buried at
      // info, beside one "tick sent" per interval, would not be found.
      pi.logger.error(`[omp-conductor] ${approval.reason}`, {
        tool: TELEGRAM_APPROVAL_TOOL,
        accessFile: config.accessFile,
      });
    }
  }

  // The other half of transport truth, read from the same file at the same
  // moment: whether visible text stays out of the operator's chat. Appended
  // after the approval line because it is the weaker instruction of the two —
  // the approval rule stops an invented approval, this one asks the turn to keep
  // quiet — and a prompt should end on the sentence that must not be missed.
  //
  // Gated on `bridgeTokenAtStart` for the same reason the approval read is, and
  // the reasoning is the mirror image of it: with no token bound, omp-telegram
  // sends *nothing*, so the two relay paths this rule warns about cannot fire
  // either. A narration warning on a dead bridge would describe a hazard that
  // does not exist, on top of an approval warning that already names the one
  // command worth running (`/telegram on`) — two remedies on one prompt, and the
  // operator acts on neither. So the rule fires only where the leak is real: a
  // live bridge whose profile is not `daemon`.
  //
  // No `accessFile` means no fleet bridge to judge, exactly as above.
  if (profile?.kind === "interactive") content = `${content}\n${TICK_NARRATION_RULE}`;

  // Recover / condition-met pokes are auditable: the same reason the writer put
  // on the sentinel line lands in the prompt and the "tick sent" log.
  const wakeReason = readTickRequestReason(ctx.cwd);
  if (wakeReason !== undefined) content = `${content}\nWake reason: ${wakeReason}`;

  const pendingLocalTick: PendingLocalTick = {
    id: randomUUID(),
    ...(scope.projectName === undefined ? {} : { projectName: scope.projectName }),
    ...(scope.policy === undefined ? {} : { policy: scope.policy }),
  };
  session.pendingLocalTicks.push(pendingLocalTick);
  try {
    try {
      pi.sendMessage(
        { customType: TICK_CUSTOM_TYPE, content, display: true, attribution: "user" },
        { triggerTurn: true, deliverAs: "followUp" },
      );
    } catch (err) {
      const pendingIndex = session.pendingLocalTicks.indexOf(pendingLocalTick);
      if (pendingIndex >= 0) session.pendingLocalTicks.splice(pendingIndex, 1);
      throw err;
    }
    if (scope.projectName !== undefined && frictionSignals.length > 0) {
      try {
        frictionStore?.markFrictionSurfaced(
          scope.projectName,
          frictionSignals.map((signal) => signal.kind),
          now,
        );
      } catch (err) {
        pi.logger.error(
          `[omp-conductor] friction digest cooldown could not be recorded: ${err instanceof Error ? err.message : String(err)}`,
        );
      }
    }
  } finally {
    frictionStore?.close();
  }
  pi.logger.info(
    wakeReason === undefined
      ? `[omp-conductor] tick sent: ${decision.reason}`
      : `[omp-conductor] tick sent: ${decision.reason} (wake: ${wakeReason})`,
    { reason: decision.reason, ...(wakeReason === undefined ? {} : { wakeReason }) },
  );
  // An empty queue at send time is the proof the previous tick was consumed, so
  // this is the only place either the counter or the marker is cleared.
  session.pendingSkips = 0;
  clearStallMarker(pi, ctx.cwd);
  // Immediate-tick poke is consumed only on a real send — gates still apply above.
  clearTickRequest(pi, ctx.cwd);

}

function writeTickRuntimeStatus(pi: TickApi, cwd: string, intervalSeconds: number): void {
  const path = join(cwd, TICK_STATUS_FILE);
  const tmp = `${path}.${process.pid}.tmp`;
  const status: TickRuntimeStatus = {
    pid: process.pid,
    intervalSeconds,
    nextTickAt: new Date(Date.now() + intervalSeconds * 1000).toISOString(),
  };
  try {
    writeFileSync(tmp, `${JSON.stringify(status)}\n`);
    renameSync(tmp, path);
  } catch (err) {
    rmSync(tmp, { force: true });
    pi.logger.error(`[omp-conductor] could not write ${path}: ${err instanceof Error ? err.message : String(err)}`);
  }
}

/**
 * Arm the interval heartbeat, honour a poke already waiting at arm time, and
 * poll mid-interval for later pokes (recover resume, condition-met #329).
 * Extracted so the ownership-retry path and the immediate-accept path cannot
 * drift: both must fire the same "do not wait a full interval after a poke"
 * behaviour. Every path still runs {@link tick} → {@link tickDecision}, so a
 * live turn coalesces rather than stacking concurrent ticks.
 *
 * The two fire-and-forget sites below (mid-interval poll, arm-time poke)
 * attach their own `.catch`: `tick` is async, the poll callback returns
 * nothing for ManagedTimers to route, and the arm-time call's handler
 * dispatch cannot see a rejection — an escaped one reaches the process-level
 * unhandledRejection handler and takes the session down. The primary
 * heartbeat interval keeps handing its promise to the harness, which routes
 * rejections to the extension error channel on its own.
 */
function armTickHeartbeat(
  pi: TickApi,
  ctx: TickContext,
  config: TickConfig,
  session: TickSession,
  toSpecTrackerSeam: ToSpecTrackerSeam | undefined,
  armSeam: ArmConsumerSeam | undefined,
): void {
  writeTickRuntimeStatus(pi, ctx.cwd, config.intervalSeconds);
  const runScheduledTick = async (): Promise<void> => {
    try {
      await tick(pi, ctx, config, session, toSpecTrackerSeam, armSeam);
    } finally {
      writeTickRuntimeStatus(pi, ctx.cwd, config.intervalSeconds);
    }
  };
  ctx.setInterval(runScheduledTick, config.intervalSeconds * 1000);
  // Short mid-interval poll so a condition that flips between heartbeats does
  // not wait a full cycle. Cap at the heartbeat itself so a misconfigured short
  // interval cannot arm a second timer faster than the primary.
  const pollMs = Math.min(TICK_REQUEST_POLL_MS, config.intervalSeconds * 1000);
  if (pollMs < config.intervalSeconds * 1000) {
    // One attempt per request, not one per poll. `clearTickRequest` only fires
    // on a successful send, so a request the gates refuse — disarmed fleet,
    // bridge down — stays on disk by design. Acting on it every poll turned that
    // into a 10s hot loop that re-entered `tick()` forever and kept rewriting the
    // runtime status. Remembering the line we already tried leaves the fixed
    // heartbeat as the fallback for that request, and still wakes promptly for
    // the next distinct one.
    let attempted: string | undefined;
    ctx.setInterval(() => {
      const line = readTickRequestLine(ctx.cwd);
      if (line === undefined) {
        // Cleared: a send landed, or an operator removed it. Re-arm for the next.
        attempted = undefined;
        return;
      }
      // A live turn owns the session. Coalesce by leaving the request on disk and
      // trying again next poll — without re-entering `tick()`, which would drive
      // the pending-skip counter at poll cadence and forge a stall diagnosis in
      // seconds when STALL_TICKS is calibrated for heartbeat intervals.
      if (ctx.hasPendingMessages()) return;
      if (line === attempted) return;
      attempted = line;
      const reason = readTickRequestReason(ctx.cwd) ?? "wake";
      pi.logger.info(
        `[omp-conductor] tick requested by ${reason} — firing without waiting for the interval`,
      );
      // Fire-and-forget like the arm-time poke, with the same containment: the
      // callback is not async, so the harness has no promise to attach a catch
      // to — a rejected tick would escape to the process-level
      // unhandledRejection handler. Log through the same channel instead, and
      // the fixed heartbeat stays the fallback for this request line.
      runScheduledTick().catch((err) => {
        pi.logger.error(
          `[omp-conductor] poked tick failed: ${err instanceof Error ? err.message : String(err)}`,
        );
      });
    }, pollMs);
  }
  if (!existsSync(join(ctx.cwd, TICK_REQUESTED_FILE))) return;
  const reason = readTickRequestReason(ctx.cwd) ?? "wake";
  pi.logger.info(`[omp-conductor] tick requested by ${reason} — firing without waiting for the interval`);
  // Fire-and-forget with the same containment as the mid-interval poll: the
  // call runs inside the session_start handler dispatch, which cannot see an
  // async rejection — an escaped one reaches the process-level
  // unhandledRejection handler and takes the session down.
  void tick(pi, ctx, config, session, toSpecTrackerSeam, armSeam).catch((err) => {
    pi.logger.error(
      `[omp-conductor] arm-time tick failed: ${err instanceof Error ? err.message : String(err)}`,
    );
  });
}

/**
 * The mechanical guard against a turn that never ends (#189).
 *
 * The harness delivers `turn_start`/`turn_end`, so the guard times a turn from
 * the first to the last tool-eligible phase and refuses further tool calls once
 * it exceeds `budgetMs` — or once an operator message has been queued behind it
 * for longer than {@link PENDING_MESSAGE_GRACE_MS}. Refusing every tool forces
 * the model to end the turn with text; nothing else unblocks the queue. `yield`
 * stays callable so a session that is also a task can still return.
 */
function armTickGuard(pi: TickApi, ctx: TickContext, budgetMs: number): void {
  let turnStartedAt: number | undefined;
  pi.on("turn_start", () => {
    turnStartedAt = Date.now();
  });
  pi.on("turn_end", () => {
    turnStartedAt = undefined;
  });
  (pi as TickApi & {
    on(
      event: "tool_call",
      handler: (
        event: { toolName: string; input: Record<string, unknown> },
        ctx: unknown,
      ) => { block: true; reason: string } | undefined,
    ): void;
  }).on("tool_call", (event) => {
    if (event.toolName === "yield") return undefined;
    return tickGuardDecision({
      turnStartedAt,
      now: Date.now(),
      budgetMs,
      hasPending: ctx.hasPendingMessages(),
    });
  });
}

/**
 * Factory-time seams for the extension, used by tests. Production runs
 * `orchestratorTickExtension(pi)` with no options and gets the module's own
 * real-time defaults; the `ask` seam hands the registered {@link ASK_TOOL} an
 * injected clock (`now`) so the filed row's timestamp is deterministic. There
 * is no wait seam any more: the ask does not wait (#438 as re-cut for the
 * console split), so there is no timer for a test to control.
 */
export interface OrchestratorTickExtensionOptions {
  ask?: {
    now?: () => number;
  };
  /**
   * The authoritative tracker surface the queue digest reads (#848).
   * Production omits it and the extension builds the real seam over the
   * existing Tracker adapter for each project; tests inject deterministic
   * fakes so the digest's live inventory is provable without a network. An
   * unreadable snapshot leaves the digest dated rather than claiming a
   * present-tense empty queue.
   */
  toSpec?: { tracker?: ToSpecTrackerSeam };
  /**
   * The arm ceremony's mechanical consumer seams (#1061). Production omits
   * them and the extension classifies inbound turns against the pending
   * challenge records itself, sending confirmations and expiry notices over
   * the same bridge transport `armTicks` sends the challenge on. Tests inject
   * a recorder (`notify`) and a fixed clock (`now`) so the settle and the
   * notices are provable without Telegram or timers.
   */
  arm?: {
    now?: () => number;
    notify?: ArmNoticeFn;
  };
}

/** The send surface an arm notice rides: the live tick config's channel and project. */
export interface ArmNoticeSurface {
  accessFile?: string;
  project?: string;
}

/**
 * One mechanical Telegram send the arm consumer raises: the confirmation when
 * a reply settles the ceremony, or the expiry notice when a challenge dies
 * unanswered. The default is {@link sendArmNotice}; tests record instead.
 */
export type ArmNoticeFn = (text: string, surface: ArmNoticeSurface) => Promise<void> | void;

/** The consumer's injectable clock and transport, carried into the tick's helpers. */
export interface ArmConsumerSeam {
  now?: () => number;
  notify?: ArmNoticeFn;
}

/**
 * The plain text of an inbound message, for the arm classifier. Content blocks
 * join with newlines so a code split across block boundaries still tokenises
 * whole; non-text blocks contribute nothing. `undefined` for an absent or
 * empty message, which is ordinary chat either way.
 */
function inboundMessageText(content: readonly { type?: string; text?: string }[] | undefined): string | undefined {
  if (content === undefined || content.length === 0) return undefined;
  const text = content.map((block) => block.text ?? "").join("\n");
  return text.length === 0 ? undefined : text;
}

/**
 * The ready-to-send expiry notice: the reason — never answered, or answered
 * but never settled — and the exact command that re-runs the ceremony. The
 * code is the proof and is never named in a notice; only the window is.
 */
function armExpiryNotice(expired: { key: string; sighting: ArmChallengeSighting }, project: string | undefined): string {
  const reissue =
    expired.key === FLEET_ARM_KEY || project === undefined || project === ""
      ? "omp-conductor arm"
      : `omp-conductor arm --project ${project}`;
  const sentAt =
    expired.sighting.sentAt === undefined
      ? ""
      : ` sent ${new Date(expired.sighting.sentAt).toISOString()}`;
  const reason =
    expired.sighting.acknowledgedAt === undefined
      ? `The arming challenge${sentAt} expired without being answered — nothing is armed.`
      : `The arming challenge${sentAt} was answered but never settled before its window closed — nothing is armed.`;
  return `${reason} Send a fresh one: ${reissue}`;
}

/**
 * The default transport for the arm consumer's confirmation and expiry
 * notices: the same bot token, paired channel and project topic `armTicks`
 * sends the challenge on, so the notice lands on the surface the operator was
 * asked in. Deliberately not availability-gated — the ceremony is the
 * operator's own interaction, exactly like the challenge send itself, and a
 * notice held for quiet hours would be the silent failure #1061 removes. Any
 * missing or down fact throws; the caller logs and keeps the challenge record
 * so the next heartbeat retries and `doctor` still reports it.
 */
export async function sendArmNotice(text: string, surface: ArmNoticeSurface): Promise<void> {
  if (surface.accessFile === undefined) {
    throw new Error("no accessFile — the arm notice has no channel");
  }
  const channel = readTelegramChannel(surface.accessFile);
  if (channel.kind === "down") {
    throw new Error(`escalation channel is not up (${surface.accessFile}): ${channel.reason}`);
  }
  const token = readTelegramToken();
  if (token === undefined) {
    throw new Error("no Telegram bot token readable — the arm notice cannot send");
  }
  let topicId: number | undefined;
  let project: { name: string; workspaceRoot?: string } | undefined;
  if (surface.project !== undefined) {
    try {
      const p = findProject(loadConfig(), surface.project);
      topicId = resolveProjectTopicId(p);
      project = { name: p.name, workspaceRoot: p.workspaceRoot };
    } catch {
      /* no project config — flat chat, exactly like armTicks */
    }
  }
  await sendTelegram(token, channel.owner, text, {
    topicId,
    ...(project === undefined ? {} : { project }),
  });
}

/**
 * The settle half of the mechanical consumer: write the markers the matched
 * challenge recorded and clear its transaction, then confirm on the
 * challenge's surface. The recorded targets are the contract — the exact rule
 * `armReply` applies — so nothing is ever armed that the challenge did not
 * name. A pre-targets record (one release of overlap) is not settled here:
 * nothing in-session may guess a marker the challenge never recorded; the CLI
 * half still settles it, and the operator is told so.
 */
function settleInboundArmMatch(
  pi: TickApi,
  match: ArmReplyMatch,
  notify: (text: string) => void,
): void {
  if (match.targets === undefined || match.owner === undefined) {
    pi.logger.error(
      "[omp-conductor] an inbound arming code matched a pre-targets challenge — nothing armed in-session; " +
        'run `omp-conductor arm --reply "<the reply>"` to settle it',
    );
    return;
  }
  for (const target of match.targets) {
    const state = resolveArmState(target.armedFile, target.project);
    writeArmedMarker(target.armedFile, match.owner, state);
  }
  clearArmTransaction(match.key, match.id);
  const label =
    match.targets.length === 1 && match.targets[0]!.project !== undefined
      ? `project ${match.targets[0]!.project}`
      : `${match.targets.length} project(s)`;
  notify(`Arming verified — ${label} armed. The reply was consumed; nothing else to do.`);
}

/**
 * The arm ceremony's mechanical consumer (#1061): classify one inbound user
 * turn against this host's pending challenges and, on a match, settle the
 * ceremony exactly as `omp-conductor arm --reply` would — markers for the
 * recorded targets, transaction cleared, a confirmation on the surface the
 * challenge was sent to. This is the real owner the ceremony needed on a host
 * with no console session: the challenge goes to the project topic, which is
 * this pane's own claimed surface, so the reply lands here and the ceremony
 * completes with no second human command. A code past its window is answered
 * with the re-run command and the dead record is cleared; a lookalike matching
 * nothing stays inert (#415's non-disclosure bound is unchanged).
 *
 * Never throws: an inbound message must not take the session down. Failures
 * are logged and the CLI half of the ceremony stays available.
 */
function consumeInboundArmReply(
  pi: TickApi,
  cwd: string,
  startupProject: string | undefined,
  text: string,
  seam: ArmConsumerSeam | undefined,
): void {
  // The state key the send half derived from the same file, read fresh so a
  // re-stamp between the send and the reply cannot split the ceremony. An
  // invalid config cannot be trusted to name either the key or a marker —
  // fail closed exactly like `armReply`, but without throwing at a message.
  const reread = readTickConfig(cwd);
  if (reread.kind === "invalid") {
    pi.logger.error(
      `[omp-conductor] inbound arm reply not classified — tick config invalid at ${reread.path}: ` +
        `${reread.problem}; use \`omp-conductor arm --reply\` on the host`,
    );
    return;
  }
  const project = reread.kind === "ok" ? reread.config.project : startupProject;
  const surface: ArmNoticeSurface = {
    ...(reread.kind === "ok" ? { accessFile: reread.config.accessFile } : {}),
    project,
  };
  const now = (seam?.now ?? Date.now)();
  const resolved = resolveArmReply(project, text, now);
  const notify = (notice: string): void => {
    const send = seam?.notify ?? sendArmNotice;
    try {
      // The send is async on the production path; message_start handlers are
      // not, so delivery rides the microtask queue and its failure lands in
      // the log, never on the session.
      void Promise.resolve(send(notice, surface)).catch((err: unknown) => {
        pi.logger.error(
          `[omp-conductor] arm notice not delivered: ${err instanceof Error ? err.message : String(err)}`,
        );
      });
    } catch (err) {
      pi.logger.error(
        `[omp-conductor] arm notice not delivered: ${err instanceof Error ? err.message : String(err)}`,
      );
    }
  };
  if (resolved.verdict === "matched") {
    settleInboundArmMatch(pi, resolved.match, notify);
    return;
  }
  // A code past its window: the ceremony is dead, so answer the person who is
  // clearly trying to complete it — and clear the expired record so neither
  // this path nor the heartbeat ever notifies twice.
  if (resolved.verdict === "expired") {
    for (const expired of expiredArmChallenges(project, now)) {
      notify(armExpiryNotice(expired, project));
      clearArmTransaction(expired.key, expired.id);
    }
  }
}

export default function orchestratorTickExtension(
  pi: TickApi,
  options: OrchestratorTickExtensionOptions = {},
): void {
  // Scoped to this registration rather than the module, so a second
  // `session_start` can neither install a second heartbeat on the same session
  // nor repeat the ownership decline — which is logged exactly once, because it
  // is the line that tells an operator which session is driving the fleet.
  let decided = false;
  // Held per registration for the same reason: the "using the default reporting
  // scope, because ..." line and the missing-approval-tool line are each logged
  // once for this heartbeat, and the stall counter is about this session's own
  // queue. A second session in the same process starts with all three at zero.
  const session: TickSession = {
    scopeFallbackLogged: false,
    approvalToolMissingLogged: false,
    bridgeTokenAtStart: true,
    pendingSkips: 0,
    pendingLocalTicks: [],
    legacyArmLogged: false,
  };
  let releaseGateArmed = false;
  let availabilityGateArmed = false;
  let guardArmed = false;
  // The authoritative tracker seam the queue digest reads (#848): production
  // runs without options and builds the real seam lazily over the existing
  // Tracker adapter (one tracker read per low-queue tick); tests inject a
  // deterministic fake. Read-only — the tick never mutates tracker state.
  // Grooming itself no longer launches from here: the daemon owns the
  // selection and the launch (#1041), so there is no launch token to mint and
  // no `task` gate to arm (#1040).
  const toSpecTrackerSeam: ToSpecTrackerSeam | undefined =
    options.toSpec?.tracker ?? {
      listOpenIssues: (project) => makeTracker(project).listOpenIssues(),
      childrenOf: (project, issue) => makeTracker(project).childrenOf(issue),
    };
  // The bounded ask tool's session state. The tool itself is registered at
  // extension-factory time, before any session exists, because OMP snapshots
  // the extension's active tool set before it emits `session_start` — a tool
  // registered there never reaches the session's model-visible set. So the
  // registration is unconditional, and the state it needs to act is filled
  // here — by `session_start`, and only for a session whose ownership is
  // accepted (the fleet agent) — because a non-owner or unresolved session in
  // the fleet cwd must not be able to execute the model-visible tool and
  // create or deliver decisions. `config` is the startup tick config that
  // owns the ceiling (timeout/budget is startup-only); `cwd` lets the tool
  // re-read the *live* tick config at execution so routing follows the current
  // project post-restamp. A call before that — or on a session that never
  // composed a tick — finds `undefined` and fails closed.
  let askSession: { cwd: string; config: TickConfig } | undefined;
  // An activation file makes this a fleet directory before Herdr can prove
  // which pane owns it. The gate therefore starts closed and only honours a
  // configured grant after ownership is accepted.
  let releaseAuthorityAccepted = false;
  /**
   * `configuredProject` is {@link TickConfig.project}, and it is what stops a
   * second configured project from collapsing every release grant this fleet
   * really holds to {@link DENIED_RELEASE_GRANTS}. Undefined for a tick config
   * that names none — including an invalid one, which arms this gate before the
   * config can be trusted at all.
   */
  const armReleaseGate = (configuredProject?: string): void => {
    if (releaseGateArmed) return;
    releaseGateArmed = true;
    (pi as TickApi & {
      on(
        event: "tool_call",
        handler: (
          event: { toolName: string; input: Record<string, unknown> },
          ctx: unknown,
        ) => ReleaseDecision | undefined,
      ): void;
    }).on("tool_call", (event) => {
      // Most tool calls are ordinary tracker/file work. Classify first so they
      // do not re-read config or emit policy diagnostics.
      const match = releaseToolMatch(event.toolName, event.input);
      if (match === undefined) return undefined;
      const shape = match.shape;
      let projectName: string | undefined;
      let grants: ResolvedGrants = DENIED_RELEASE_GRANTS;
      let external = true;
      try {
        const config = loadConfig();
        const project = findProject(config, configuredProject);
        const installAuthority = resolveSharedInstallAuthority(config.projects);
        projectName = project.name;
        grants = {
          ...resolveReleaseGrants(project),
          install: installAuthority.holder ?? "human",
        };
        external = project.escalation.orchestrator === "external";
      } catch (err) {
        // A missing/unreadable config cannot open a release gate. Log only when
        // a release-shaped call actually reaches this handler.
        pi.logger.error(
          `[omp-conductor] release policy unreadable; denying every release shape: ${
            err instanceof Error ? err.message : String(err)
          }`,
        );
      }
      // Per shape since #122: a fleet may hold `git-tag` and still be refused
      // `deploy`, so this asks about the shape in hand, not about one policy word.
      const refusal = releaseRefusal(grants, "orchestrator", shape);
      if (releaseAuthorityAccepted && refusal === undefined) return undefined;
      // Embedded orchestrators carry the same tripwire inline through
      // `createSession`; suppress this second copy only after this session has
      // proved it owns the external heartbeat.
      if (releaseAuthorityAccepted && !external) return undefined;
      // Before ownership is proved this session holds no grant at all, so a
      // covered shape still refuses with the deny-all wording.
      const decision = refusal ?? releaseRefusal(DENIED_RELEASE_GRANTS, "orchestrator", shape);
      if (decision === undefined) return undefined;
      if (projectName !== undefined) {
        try {
          const invocation = releaseBlockedInvocation(event.toolName, event.input);
          recordReleaseBlock(projectName, "orchestrator", shape, {
            tool: event.toolName,
            reason: decision.reason,
            ...(invocation === undefined ? {} : { invocation }),
            matched: match.matched,
            args: redactReleaseArgs(event.input),
          });
        } catch (err) {
          pi.logger.error(
            `[omp-conductor] could not record release-policy block: ${
              err instanceof Error ? err.message : String(err)
            }`,
          );
        }
      }
      return decision;
    });
  };

  /** `configuredProject` carries {@link TickConfig.project} for the same reason
   *  {@link armReleaseGate} takes it: a recovered tick must reconstruct *this*
   *  fleet's availability policy, not refuse to guess between two projects.
   *  `cwd` is the fleet directory itself, which the arm consumer re-reads the
   *  tick config from at reply time (#1061). */
  const armAvailabilityGate = (configuredProject?: string, cwd?: string): void => {
    if (availabilityGateArmed) return;
    availabilityGateArmed = true;

    // Custom messages sent with `pi.sendMessage` bypass `before_agent_start`.
    // `message_start` is the first lifecycle event that carries their exact
    // custom type, so it is the reliable boundary between a local tick and an
    // inbound human turn.
    pi.on("message_start", (event) => {
      const message = event.message;
      if (message.role === "custom" && message.customType === TICK_CUSTOM_TYPE) {
        const pending = session.pendingLocalTicks.shift();
        if (pending !== undefined) {
          session.activeLocalTick = { ...pending, humanWaiting: false };
        } else {
          // A queued custom tick can survive a host restart after the in-memory
          // enqueue record does not. Reconstruct its scope so recovery cannot
          // silently turn an autonomous run into an interactive one.
          const scope = resolveTickScope(configuredProject);
          const id =
            typeof message.timestamp === "number"
              ? createHash("sha256").update(`recovered-tick\0${message.timestamp}`).digest("hex")
              : randomUUID();
          session.activeLocalTick = {
            id,
            ...(scope.projectName === undefined ? {} : { projectName: scope.projectName }),
            ...(scope.policy === undefined ? {} : { policy: scope.policy }),
            humanWaiting: false,
          };
        }
        return;
      }
      // An inbound user turn on THIS session is an anomaly, not a
      // conversation: operator DMs belong to the 24/7 console session, so a
      // turn arriving here is an accidental topic post. Two things happen,
      // and both are mechanisms — nothing here is left to model behaviour:
      //
      // 1. The arm ceremony's reply is classified against the pending
      //    challenge records (#1061). The pre-#415 tombstone is retired: it
      //    reasoned the reply lands "where no tick extension runs", but the
      //    send half delivers the challenge to the project topic — this
      //    pane's own claimed surface — so the reply lands exactly here, and
      //    on a host with no console session the CLI's `arm --reply` step had
      //    no owner at all. A matching code settles the ceremony
      //    (markers written, transaction cleared) and answers with a
      //    confirmation; a code past its window is answered with the re-run
      //    command; a lookalike matching nothing stays inert, exactly as
      //    #415 required.
      // 2. {@link tickGuardDecision}'s preemption, unchanged: a person who
      //    posts into the fleet topic by accident is still a person waiting,
      //    and the autonomous-interrupt refusal must not fire at them.
      if (message.role === "user" && message.synthetic !== true && message.attribution !== "agent") {
        if (cwd !== undefined) {
          const text = inboundMessageText(message.content);
          if (text !== undefined) consumeInboundArmReply(pi, cwd, configuredProject, text, options.arm);
        }
        if (session.activeLocalTick !== undefined) {
          session.activeLocalTick.humanWaiting = true;
        }
      }
    });
    pi.on("agent_end", () => {
      session.activeLocalTick = undefined;
    });
    (pi as TickApi & {
      on(
        event: "tool_call",
        handler: (
          event: { toolName: string; input: Record<string, unknown> },
          ctx: unknown,
        ) => TelegramInterruptBlock | undefined,
      ): void;
    }).on(
      "tool_call",
      (event) =>
        // Routing first: a call that cannot reach the right thread is refused
        // whether or not the availability policy would have let it through.
        telegramRoutingBlock(configuredProject, event) ??
        autonomousTelegramInterruptBlock(pi, session, event),
    );
  };

  // Mount the durable ask surface (#438) at extension-factory time, not in
  // `session_start`: OMP 17.2.9 snapshots the extension's active tool set
  // before it emits `session_start`, so a tool registered there mutates the
  // registry but never reaches this session's model-visible set. Registering
  // here is the only seam that lands the tool in the real snapshot. Next to
  // the gates it complements: raw {@link TELEGRAM_APPROVAL_TOOL} is refused on
  // autonomous ticks, so the session must own the tool that asks instead.
  //
  // It registers `defaultInactive`: the OMP snapshot auto-includes extension
  // tools in the initial active set unless the definition opts out, so without
  // this flag an unreadable or un-stamped multi-project session would expose
  // the tool's schema even though it cannot route. Inactive by default, the
  // surface is brought into the live set by {@link tick} — and only for as
  // long as the current tick config resolves a project.
  //
  // The session state the tools need — the cwd and the startup config that
  // routes a call to a project — is read lazily from {@link askSession},
  // filled by `session_start` only once ownership is accepted for a session
  // that goes on to compose a tick; a call on any other session fails closed.
  //
  // The tool is inert unless the config resolves this fleet's project: it needs
  // the project to file the decision row against and to resolve the delivery
  // target. An unreadable/ambiguous config makes the tool say so and record
  // nothing, which is the same fail-closed posture the autonomous gate takes.
  /**
   * The plumbing both operator-question tools share (#947): which project this
   * call records against, and the one sanctioned delivery path.
   *
   * Extracted rather than copied, because the questionnaire needs exactly the
   * live-config routing, the fail-closed refusals and the held-notice fallback
   * the single ask already got right — and a second copy of the fallback is how
   * one surface quietly starts dropping questions the other one holds.
   */
  const resolveOperatorAskContext = (
    tool: string,
  ): { ok: true; project: ProjectConfig } | { ok: false; text: string } => {
    const session = askSession;
    if (session === undefined) {
      return {
        ok: false,
        text: `${tool}: not available in this session (no orchestrator tick); nothing was asked or recorded.`,
      };
    }
    const routed = resolveAskProject(session.cwd, session.config);
    if (routed.kind === "error") {
      return {
        ok: false,
        text:
          `${tool}: conductor config unreadable (${routed.problem}); ` +
          "nothing was asked or recorded. Repair the config, do not ask through another path.",
      };
    }
    return { ok: true, project: routed.project };
  };

  const operatorAskDelivery =
    (tool: string, projectConfig: ProjectConfig, store: Store) =>
    async (text: string, category: InterruptCategory): Promise<AskDeliveryResult> => {
      const at = Date.now();
      const noticeId = randomUUID();
      try {
        const delivered = await deliverOperatorMessage(projectConfig, text, {
          store,
          at,
          noticeId,
          category,
        });
        return delivered.kind === "sent"
          ? { kind: "sent", category: delivered.category }
          : { kind: "held", category: delivered.category, noticeId: delivered.noticeId };
      } catch (err) {
        // A failed immediate send must not drop the question: fall back to the
        // durable hold exactly like the gate's own hold path, and let the daemon
        // retry with the digest.
        store.addHeldNotice({
          id: noticeId,
          project: projectConfig.name,
          category,
          summary: text.split("\n", 1)[0]!.slice(0, 240),
          detail: text,
          createdAt: at,
        });
        pi.logger.error(
          `[omp-conductor] ${tool} could not deliver the ask directly (${err instanceof Error ? err.message : String(err)}); held durably`,
        );
        return { kind: "held", category, noticeId };
      }
    };

  pi.registerTool({
    name: ASK_TOOL,
    label: ASK_TOOL,
    defaultInactive: true,
    description:
      `Ask your operator one question. It does NOT wait for the answer: the question is recorded ` +
      `durably (a decision row + the same delivery path as \`omp-conductor message\`), delivered per the ` +
      `reporting policy, and the call returns immediately with the row id and the exact command that ` +
      `resolves it. Your operator answers in their console session — this tick session is headless and no ` +
      `reply can land here — and an unanswered row stays open and pending, re-surfaced in every tick ` +
      `prompt, until it is answered, withdrawn, or swept by the seven-day decision expiry. When ` +
      `"options" are supplied, "recommended" must be one of their labels — the label as delivered, never ` +
      `an index. Park the work the question blocks in the same turn you ask: out of the claimable queue, ` +
      `state recorded, row id in your report. Never treat your own recommendation as an approval.`,
    parameters: askParameterSchema(),
    approval: "write",
    execute: async (_toolCallId, params) => {
      const parsed = parseAskRequest(params);
      if (!parsed.ok) {
        return { content: [{ type: "text", text: parsed.problem }], isError: true };
      }
      // Routing follows the *live* tick config, not the session-start stamp: a
      // restamp (un-stamped → stamped, or project A → B) must make the next
      // tick's toolbox land on the project the turn actually ticks for.
      const context = resolveOperatorAskContext(ASK_TOOL);
      if (!context.ok) {
        return { content: [{ type: "text", text: context.text }], isError: true };
      }
      const projectConfig = context.project;
      const store = openStore(dbPath());
      let result: AskResult;
      try {
        result = await performAsk(parsed.request, {
          store,
          project: projectConfig.name,
          // Test seam: production omits `now` and `performAsk` falls back to
          // the module's real clock; a test hands one in so the filed row's
          // timestamp is deterministic.
          ...(options.ask?.now === undefined ? {} : { now: options.ask.now }),
          deliver: operatorAskDelivery(ASK_TOOL, projectConfig, store),
        });
      } finally {
        store.close();
      }
      return { content: [{ type: "text", text: result.text }] };
    },
  });

  // The batched spec-out questionnaire (#947): the same durable contract as the
  // single ask, asked once for several judgement calls about one issue. Mounted
  // and routed through exactly the shared helpers above, so the two surfaces
  // cannot come to disagree about which project a question belongs to or what
  // happens when delivery fails.
  pi.registerTool({
    name: QUESTIONNAIRE_TOOL,
    label: QUESTIONNAIRE_TOOL,
    defaultInactive: true,
    description:
      `Ask your operator several questions about ONE issue as a single message. Like conductor_ask it ` +
      `does NOT wait: every item is recorded as its own durable decision row before anything is ` +
      `delivered, all bound to the issue you name in "spec-issue" — so the answers become that issue's ` +
      `provenance and a later reader sees why a slice is shaped the way it is — and the call returns the ` +
      `group id, every row id, and the command that resolves each one. Your operator answers in their ` +
      `console session, item by item, in any order. Ask ONLY the judgement calls that genuinely belong ` +
      `to your operator — anything a repo read can answer is your own work — and at most ` +
      `${MAX_QUESTIONNAIRE_ITEMS} items. Park the work every item blocks in the same turn you ask, and ` +
      `name the group id in your report.`,
    parameters: questionnaireParameterSchema(),
    approval: "write",
    execute: async (_toolCallId, params) => {
      const parsed = parseQuestionnaireRequest(params);
      if (!parsed.ok) {
        return { content: [{ type: "text", text: parsed.problem }], isError: true };
      }
      const context = resolveOperatorAskContext(QUESTIONNAIRE_TOOL);
      if (!context.ok) {
        return { content: [{ type: "text", text: context.text }], isError: true };
      }
      const projectConfig = context.project;
      const store = openStore(dbPath());
      try {
        const result = await performQuestionnaire(parsed.request, {
          store,
          project: projectConfig.name,
          ...(options.ask?.now === undefined ? {} : { now: options.ask.now }),
          deliver: operatorAskDelivery(QUESTIONNAIRE_TOOL, projectConfig, store),
        });
        return { content: [{ type: "text", text: result.text }] };
      } finally {
        store.close();
      }
    },
  });

  pi.on("session_start", (_event, ctx) => {
    if (decided) return;

    // A subagent inherits the orchestrator's cwd, so it finds the same
    // activation file and would arm a heartbeat of its own — one extra tick
    // per worker, each prompting a session whose whole contract is to finish
    // and yield. The discriminator is the one omp-telegram has been running in
    // production: task sessions are headless *and* always carry the `yield`
    // tool. Neither half suffices alone — a headless root session (print/RPC
    // mode) has no `yield`, and an
    // interactive session may well have one. Checked before the config read
    // so the overwhelmingly common case never touches the filesystem.
    if (!ctx.hasUI && pi.getActiveTools().includes("yield")) {
      pi.logger.info("[omp-conductor] tick inert: subagent session");
      return;
    }

    const result = readTickConfig(ctx.cwd);

    if (result.kind === "absent") {
      pi.logger.info(`[omp-conductor] orchestrator tick inactive: no ${TICK_CONFIG_FILE} in ${ctx.cwd}`);
      return;
    }

    // Present but invalid still identifies a fleet directory. Install the
    // fail-closed handler before validation or ownership can return early.
    // The project this fleet cwd ticks for, so both gates resolve its own policy
    // rather than collapsing fail-closed on a host with a second project. An
    // invalid config names none — it cannot be trusted to.
    const configuredProject = result.kind === "ok" ? result.config.project : undefined;
    armReleaseGate(configuredProject);
    armAvailabilityGate(configuredProject, ctx.cwd);
    // The tool is already registered (extension-factory time); only the session
    // state it acts on is filled, and only once ownership is accepted below.
    // `configuredProject` isn't closed over at all — routing is re-read from
    // the live tick config at execution, so a restamp binds the current
    // project. Ownership gating is the point: a declined or unresolved session
    // in the fleet cwd must never be able to execute the model-visible tool
    // and create or deliver decisions.

    if (result.kind === "invalid") {
      const detail = `${result.path}: ${result.problem}`;
      pi.logger.error(`[omp-conductor] orchestrator tick disabled — ${detail}`);
      // The one notification this extension ever raises: a broken heartbeat
      // config is silent failure otherwise, and silence is what it is for.
      ctx.ui.notify(`omp-conductor: orchestrator tick disabled — ${detail}`, "error");
      return;
    }

    const config = result.config;

    // Whether omp-telegram could have bound a token when this session started,
    // which is a different question from whether one exists now and is the only
    // one the approval preflight can honestly ask. The bridge resolves its token
    // once, in `startBot()`, and rebinds only on `/telegram token` or
    // `/telegram on` — so a token written into `.env` out-of-band leaves
    // `telegram_ask` and `telegram_send` dead for the life of this session even
    // though the file now looks perfect. Sampling here, next to the same
    // `session_start` omp-telegram binds on, is as close as another package can
    // get to that fact.
    //
    // Conservative on the other transition: an operator who ran `/telegram on`
    // after start really does have a working bridge, and this snapshot will keep
    // saying otherwise until the session restarts. That costs a fallback
    // instruction the orchestrator can follow, where guessing the other way
    // costs an amendment recorded as approved that nobody ever answered.
    session.bridgeTokenAtStart =
      config.accessFile === undefined ? true : bridgeTokenBound(config.accessFile);

    // Activation is a property of the directory, so every omp session started in
    // the fleet's cwd used to become a ticker — and with merge and release
    // delegated in config, a shell opened beside the orchestrator believed it
    // held both. Asked after the config read because the config names the agent,
    // and before the timer because arming first is the bug.
    const agentName = config.agentName ?? DEFAULT_FLEET_AGENT_NAME;
    const resolve = (): TickOwnership =>
      resolveTickOwnership({
        cwd: ctx.cwd,
        agentName,
        env: process.env,
        pid: process.pid,
        now: new Date(),
        ...(ctx.sessionManager.getSessionFile() === undefined
          ? {}
          : { sessionFile: ctx.sessionManager.getSessionFile() }),
      });

    const ownership = resolve();

    // Only a definitive answer is final. "You are not the fleet agent" cannot
    // become untrue while this session lives, so it latches. "herdr did not
    // answer" says nothing about identity, so it must not — otherwise one CLI
    // timeout at session start is indistinguishable from a fleet that was never
    // meant to tick, and the heartbeat is gone until a human notices.
    if (ownership.kind === "unresolved") {
      pi.logger.info(`[omp-conductor] orchestrator tick pending: ${ownership.reason}`, { agentName });
      // Faster than the tick interval so a blip costs a minute rather than a
      // whole cycle, and never slower than one — a fleet on a short interval
      // should not wait longer to recover than it would to tick.
      const retryMs = Math.min(RETRY_OWNERSHIP_MS, config.intervalSeconds * 1000);
      // Disarmed by a flag, not `clearInterval`: `ctx.setInterval` hands back an
      // opaque handle precisely because the harness owns timer lifecycle and
      // clears them on `session_shutdown`. A settled retry is a no-op that costs
      // one boolean per minute until the session ends.
      let settled = false;
      ctx.setInterval(() => {
        if (settled) return;
        const next = resolve();
        if (next.kind === "unresolved") return; // already logged once; stay quiet
        settled = true;
        if (next.kind === "declined") {
          pi.logger.info(`[omp-conductor] orchestrator tick inactive: ${next.reason}`, { agentName });
          return;
        }
        if (next.note !== undefined) pi.logger.error(`[omp-conductor] tick ownership: ${next.note}`);
        releaseAuthorityAccepted = true;
        // Ownership finally resolved to the fleet agent — only now may the
        // model-visible ask be executable (defect: non-owner sessions must
        // not create or deliver decisions). `cwd` is for re-reading the live
        // tick config at execution, `config` for the startup-only ceiling.
        askSession = { cwd: ctx.cwd, config };
        armTickHeartbeat(pi, ctx, config, session, toSpecTrackerSeam, options.arm);
        if (!guardArmed) {
          guardArmed = true;
          armTickGuard(pi, ctx, (config.budgetSeconds ?? DEFAULT_TICK_BUDGET_SECONDS) * 1000);
        }
        pi.logger.info(`[omp-conductor] orchestrator tick active: ownership resolved on retry`, { agentName });
      }, retryMs);
      decided = true;
      return;
    }

    if (ownership.kind === "declined") {
      decided = true;
      pi.logger.info(`[omp-conductor] orchestrator tick inactive: ${ownership.reason}`, { agentName });
      return;
    }
    if (ownership.note !== undefined) pi.logger.error(`[omp-conductor] tick ownership: ${ownership.note}`);
    releaseAuthorityAccepted = true;
    // Only the fleet agent may execute the bounded ask (see the retry path for
    // why): a declined or unresolved session keeps `askSession` undefined and
    // the tool fails closed.
    askSession = { cwd: ctx.cwd, config };
    armTickHeartbeat(pi, ctx, config, session, toSpecTrackerSeam, options.arm);
    if (!guardArmed) {
      guardArmed = true;
      armTickGuard(pi, ctx, (config.budgetSeconds ?? DEFAULT_TICK_BUDGET_SECONDS) * 1000);
    }
    decided = true;
    // Both gates are named at startup: "why is it not ticking?" is answered by
    // looking at the files this line lists, and an unset channel gate on a fleet
    // host is visible here rather than only in its absence.
    const gates = [
      config.armedFile === undefined ? undefined : `armed marker ${config.armedFile}`,
      config.accessFile === undefined ? "no escalation-channel gate" : `escalation channel ${config.accessFile}`,
    ].filter((g) => g !== undefined);
    pi.logger.info(
      `[omp-conductor] orchestrator tick active: every ${config.intervalSeconds}s, gated on ${gates.join(" + ")}`,
    );
  });
}