fix(compaction): retain input-box reading on held preparation (#2797)

A preparation held by the input-box check now records the reading on its attempt: status, confidence, the fixed evidence strings and the cursor, with no draft or suggestion text. It appears in getPreparationState (and so rig compact --state), the attempt state file and a recovered preparation. The transport passes the selected metadata through its existing preserveInput hold; the existing status strings are unchanged. Captures, refusals, retries, waits, Enter and compaction decisions are unchanged.

docs/reference/slack-app-setup.md

@@ -405,9 +405,11 @@ budget window when exhausted. Slack's per-method `Retry-After` is retained acros
 Only an acknowledged edit gets a success receipt. `rig slack status` shows queued reconciliation
 and the last edit failure. Transient failures retry with increasing waits, capped at an hour,
 behind other queued work. Permanent Slack failures (such as missing scope, a deleted original,
-an archived channel or an uneditable message) remain unfinished and are retried only on the next
-connector start. They do not keep consuming edit capacity. These edit calls have their own
-budget, separate from missed-message scanning below.
+an archived channel or an uneditable message) remain unfinished. A deleted original is recorded
+on the row and never asked about again; a new post of the same ask is tried afresh. The others,
+which a scope, membership, a different bot token or an upgrade can fix, are retried on the next
+connector start. A message is uneditable when the configured token didn't post it. They do not keep consuming edit capacity. These edit calls have their own budget,
+separate from missed-message scanning below.
 
 The baseline already includes public-channel history access and `chat:write`. Private-channel
 reads may require `groups:history`, which the baseline does not grant; a refused read never
@@ -454,9 +456,18 @@ not covered by that margin. A corrupt or unreadable checkpoint leaves recovery u
 and preserves the existing file; live inbound continues.
 
 Each pass admits at most four history calls, 100 entries and 15 seconds of work, with a
-five-second request ceiling. Channels and threads take turns within that one budget, so with many
-tracked threads each thread is scanned every few passes; `rig slack status` shows how far behind
-the oldest waiting thread is. Work already started on a message, such as
+five-second request ceiling. Channels and threads take turns within that one budget. Each tracked
+thread is scanned once after each start and after each Socket Mode reconnect that followed a gap
+(not a refresh Slack overlaps with the old connection), which catches replies sent while no
+connection was up; replies while connected arrive live. A thread the pass at the connect itself
+reaches is scanned once more, since that pass stops five seconds short of the connect. Those scans go in need order: a thread
+leaving the seven-day window within six hours first, since once it leaves a missed reply is never
+recovered; then threads whose ask still waits on the person (its queue row is pending, in progress
+or blocked) before the rest, each newest first. The order comes from the saved checkpoints, so a
+restart starts with the newest open ask, not the oldest thread. After that a thread is scanned again
+only as a backstop, hourly while its ask waits on the person and daily otherwise. A machine the hub
+relays to can't see a gap in relayed events, so there a settled thread's backstop is three hours. `rig slack status` shows how far behind the oldest waiting thread is and how
+many threads are settled. Work already started on a message, such as
 an attachment transfer, is not cancelled by these limits or by a stop, and a new
 pass waits for it to finish. Rate limits retain Slack's Retry-After across restart, and hold back
 only the method Slack limited (channel history or thread replies).
@@ -465,8 +476,8 @@ When Slack reports a plan history limit, recovery still lands the available page
 and advances normally, retaining an older-history limitation in status across restarts.
 Missing bot credentials, history scope, membership, retention limits and API
 failures appear as recovery limitations. A tracked thread Slack can't find stays incomplete in
-status, with no further call, until its thread mapping changes or the daemon restarts or rewires
-the connector; that isn't proof the thread was deleted. No additional scope is required to keep
+status and is asked again at most once a day, across restarts; that isn't proof the thread was
+deleted. No additional scope is required to keep
 live delivery working. Threads nobody tracks, older open threads, threads closed more than seven
 days ago and other channels remain outside this recovery scope; global chronological order is
 not promised. This change does not replay intervals already checkpointed before broadcast-reply

packages/cli/src/commands/slack.ts

@@ -258,7 +258,7 @@ export function slackCommand(deps: SlackDeps = {}): Command {
             coverage?: { coverageStart: string; coveredThrough: string; pending?: { upper: string; nextLatest: string }; nextRetryAt?: number } | null;
             channels?: Array<{ channel?: string | null; state?: string; reason?: string; lastScanAt?: string;
               coverage?: { coverageStart: string; coveredThrough: string; nextRetryAt?: number } | null }>;
-            threads?: { state?: string; reason?: string; tracked?: number; scanned?: number; pending?: number; oldestPendingSeconds?: number | null;
+            threads?: { state?: string; reason?: string; tracked?: number; scanned?: number; pending?: number; settled?: number; settledRescanHours?: number; oldestPendingSeconds?: number | null;
               backoff?: Record<string, number>; incomplete?: Record<string, number> };
             limits?: string[] } } | undefined;
         if (connector) {
@@ -288,6 +288,7 @@ export function slackCommand(deps: SlackDeps = {}): Command {
             const behind = threads.oldestPendingSeconds != null ? ` (oldest ${Math.round(threads.oldestPendingSeconds / 60)} min behind)` : "";
             log(`  Thread recovery: ${threads.state ?? "unknown"}${threads.reason ? ` (${threads.reason})` : ""}; ${threads.tracked ?? 0} tracked (open, or closed within seven days); `
               + `${threads.scanned ?? 0} scanned, ${threads.pending ?? 0} waiting their turn${behind}`
+              + `${threads.settled ? `, ${threads.settled} settled (nobody waiting; rescanned ${!threads.settledRescanHours || threads.settledRescanHours === 24 ? "daily" : `every ${threads.settledRescanHours} hours`})` : ""}`
               + `${reasons(threads.backoff) ? `; backing off: ${reasons(threads.backoff)}` : ""}${reasons(threads.incomplete) ? `; incomplete: ${reasons(threads.incomplete)}` : ""}`);
           }
           log(`  Recovery counts since connector start: accepted ${recovery?.acceptedThisProcess ?? "unknown"}; dead-lettered ${recovery?.deadLetteredThisProcess ?? "unknown"} (custody, not delivery)`);

packages/daemon/src/domain/claude-compaction-enforcer.ts

@@ -435,14 +435,15 @@ export interface RestoreDeliveryAttempt {
   errorClass?: string;
   failures: number;
   noticeErrorClass?: string;
-  /** Legacy unreadable-poll count from records written before restore input followed ordinary delivery defaults. */
+  /** Consecutive idle polls whose input box read neither empty nor typed for this step (#2575). */
   unreadable?: number;
   /** What the input box read: on a hold, the reading held on; on a delivery over a box OpenRig couldn't read (#2575),
    *  that reading. Never "empty" on a delivery. */
   inputBox?: string;
 }
 
-/** Legacy #2575 threshold for old diagnostics. Managed input now follows ordinary draft-delivery defaults. */
+/** #2575: after this many consecutive idle polls that can't read the box, a post-compaction step goes ahead; a typed
+ *  draft still holds it. Waiting doesn't show the box is empty, so the delivery is recorded as over an unreadable box. */
 export const UNREADABLE_POLLS_BEFORE_DELIVERY = 2;
 export interface RestoreDeliveryFailure {
   sessionName: string;
@@ -536,26 +537,7 @@ export const RESTART_RECOVERY_WINDOW_MS_DEFAULT = 6 * 60 * 60_000;
 /** This daemon process. A record this process wrote is its own state, never something to recover. */
 const DAEMON_PROCESS_ID = randomUUID();
 
-export interface PreparationFailure {
-  sessionName: string;
-  occupantGeneration: string;
-  attemptId: string;
-  mapPath: string;
-  statePath: string;
-  endedAttempts: number;
-  reason: string;
-  inputBox?: PreparationState["inputBox"];
-}
-interface PreparationFailureRun {
-  occupantGeneration: string;
-  sessionId: string | null;
-  endedAttempts: number;
-  lastAttemptId: string;
-  reportedAttempts?: number;
-}
-
 interface CompactionRecord {
-  preparationFailures?: PreparationFailureRun;
   inputBox?: PreparationState["inputBox"];
   /** The daemon process that last wrote it. */
   daemon: string;
@@ -639,6 +621,7 @@ function readCompactionRecord(file: string): CompactionRecord | null {
 }
 
 /** Input-box readings that hold nothing typed: an owed step after an uncertain /compact may go in. A harness suggestion isn't input. */
+const INPUT_BOX_CLEAR = new Set(["empty", "likely-suggestion"]);
 
 function strandedSummary(record: CompactionRecord): StrandedCompaction {
   return { attemptId: record.attemptId, mapPath: record.mapPath, phase: record.phase, recordedAt: new Date(record.updatedAt).toISOString() };
@@ -684,6 +667,8 @@ export class ClaudeCompactionEnforcer {
   /** Seats whose owed restore steps follow a /compact whose outcome is unknown: its text may wait unsubmitted in the input
    *  box, where any Enter would submit it, so each step goes only into an empty box. */
   private readonly afterUncertainCompact = new Set<string>();
+  /** What the input box last read for such a seat, before an owed step (absent until the first check). */
+  private readonly lastInputBox = new Map<string, string>();
   private readonly restartRecoveryWindowMs: number;
   private readonly daemonInstance: string;
   private readonly resolveOccupantGeneration?: (sessionName: string) => string | null;
@@ -694,7 +679,6 @@ export class ClaudeCompactionEnforcer {
     saturationBoundPercentage: number;
   }) => Promise<void> | void;
 
-  private readonly onPreparationFailure?: (failure: PreparationFailure) => Promise<void>;
   private readonly onRestoreDeliveryFailure?: (failure: RestoreDeliveryFailure) => Promise<void>;
   private readonly terminalPolicies?: TerminalPolicyStore;
 
@@ -702,7 +686,6 @@ export class ClaudeCompactionEnforcer {
     settingsStore: SettingsStore,
     sessionTransport: SessionTransport,
     opts?: {
-      onPreparationFailure?: (failure: PreparationFailure) => Promise<void>;
       onRestoreDeliveryFailure?: (failure: RestoreDeliveryFailure) => Promise<void>;
       terminalPolicies?: TerminalPolicyStore;
       dedupWindowMs?: number;
@@ -727,7 +710,6 @@ export class ClaudeCompactionEnforcer {
     },
   ) {
     this.onRestoreDeliveryFailure = opts?.onRestoreDeliveryFailure;
-    this.onPreparationFailure = opts?.onPreparationFailure;
     this.terminalPolicies = opts?.terminalPolicies;
     this.settingsStore = settingsStore;
     this.sessionTransport = sessionTransport;
@@ -794,7 +776,7 @@ export class ClaudeCompactionEnforcer {
     // One readiness read, with no wait after it: this runs inside the monitor's serial poll, so a busy seat must not hold
     // the other seats' polls. A busy seat, or a read that fails, waits for a later poll.
     let idle: { ok: boolean };
-    try { idle = await this.sessionTransport.waitUntilIdle(session, CAPTURE_READABLE_BUDGET_MS, undefined, { singleObservation: true, allowDraft: true }); }
+    try { idle = await this.sessionTransport.waitUntilIdle(session, CAPTURE_READABLE_BUDGET_MS, undefined, { singleObservation: true }); }
     catch { idle = { ok: false }; }
     if (!idle.ok) return { triggered: false, reason: "waiting_for_idle" };
     // The gate is used up only when the replacement request goes out (deliverPreparation), and that send checks idle
@@ -865,12 +847,16 @@ export class ClaudeCompactionEnforcer {
     const failure = !lastDelivery ? "" : lastDelivery.reason === "delivered_over_unreadable_input_box"
       ? ` The ${steps[lastDelivery.stage]} went in at ${lastDelivery.at} over an input box OpenRig couldn't read (it read ${lastDelivery.inputBox ?? "unknown"}); a draft there, if any, may hold its text.`
       : ` ${failed ? "Last delivery" : "waiting: seat busy"} at ${lastDelivery.at}: ${[lastDelivery.reason, lastDelivery.activity?.reason, lastDelivery.errorClass, lastDelivery.noticeErrorClass && `owner notice failed (${lastDelivery.noticeErrorClass})`].filter(Boolean).join("; ")}. OpenRig will retry on its existing polls.${lastDelivery.unreadable
-        ? " The operator’s draft-protection settings still apply." : ""}`;
+        ? ` After ${UNREADABLE_POLLS_BEFORE_DELIVERY} idle polls that can't read the input box, the step goes in anyway; a typed draft still holds it.` : ""}`;
     const step = steps[stage];
     if (!this.afterUncertainCompact.has(sessionName)) {
       return { stage, held: failed, ...last, detail: `OpenRig owes this seat its ${step} next, sent on a poll once the pane is idle.${failure}` };
     }
-    return { stage, held: failed, ...last, detail: `OpenRig owes this seat its ${step} next, after a /compact whose result was lost. The command is not replayed; restore input follows ordinary delivery checks and the operator's draft-protection settings.${failure}` };
+    const inputBox = this.lastInputBox.get(sessionName) ?? "not_checked";
+    const held = inputBox !== "not_checked" && !INPUT_BOX_CLEAR.has(inputBox);
+    return { stage, held: held || failed, inputBox, ...last, detail: (held
+      ? `held: OpenRig owes this seat its restore steps after a /compact whose result was lost, and waits while its input box isn't empty (it reads ${inputBox}). The seat may have /compact text waiting, which any Enter would submit, so check the pane. A seat whose own status line keeps the box unreadable stays held.`
+      : `OpenRig owes this seat its ${step} next, after a /compact whose result was lost. Each step goes only into an empty input box, sent on a poll once the pane is idle (the box last read ${inputBox}).`) + failure };
   }
 
   /** An open record from before a restart that was too old to resume, while no later attempt has replaced it. */
@@ -1049,9 +1035,12 @@ export class ClaudeCompactionEnforcer {
       && this.pendingPreCompactPrep.get(input.sessionName) === attempt;
     // #2575: after enough idle polls that couldn't read the box for this step, the step goes in unless the box reads typed.
     // Not after a /compact whose result was lost: there an Enter could submit /compact, so that check below stays strict.
+    const last = this.records.get(input.sessionName)?.record.lastDelivery;
+    const overUnreadable = !this.afterUncertainCompact.has(input.sessionName) && !!last && last.stage === pendingStage && !last.ok
+      && (last.unreadable ?? 0) >= UNREADABLE_POLLS_BEFORE_DELIVERY;
     const sendOpts = {
       openrigNotice: true as const,
-      allowDraftWhileIdle: true as const,
+      preserveInput: overUnreadable ? "typed" as const : true as const,
       beforeWrite: () => {
         if (!isCurrent()) throw new DeliveryGuardError("compaction_cancelled", "Managed compaction was cancelled or replaced; no further input is authorized.");
       },
@@ -1085,6 +1074,15 @@ export class ClaudeCompactionEnforcer {
         return { triggered: false, reason: "stale_generation" };
       }
     }
+    // After a /compact whose outcome is unknown, its text may still wait unsubmitted in the input box, and this step's Enter
+    // would submit it. So each step goes only into a box holding nothing typed (a harness suggestion isn't input); while it
+    // holds text or can't be read, the step waits for the next poll, and the notice and --state say why.
+    if (this.afterUncertainCompact.has(input.sessionName)) {
+      const box = await this.inputBoxStatus(input.sessionName);
+      if (!isCurrent()) return { triggered: false, reason: "preparation_stopped" };
+      this.lastInputBox.set(input.sessionName, box);
+      if (!INPUT_BOX_CLEAR.has(box)) return { triggered: false, reason: "input_box_not_empty" };
+    }
     if (pendingStage === "turn_boundary") {
       // Only this first step after /compact reads past the compaction output that fills the screen above the input box.
       const boundary = await this.sendRestoreStage(
@@ -1164,6 +1162,7 @@ export class ClaudeCompactionEnforcer {
       this.persistStage(input.sessionName, "done", undefined, this.overUnreadableDelivery("compliance_prompt", compliance));
       this.records.delete(input.sessionName);
       this.afterUncertainCompact.delete(input.sessionName);
+      this.lastInputBox.delete(input.sessionName);
       this.restoreCarry.delete(input.sessionName);
       // The attempt is finished, including one stopped after a lost /compact result whose restore this was.
       const finished = this.pendingPreCompactPrep.get(input.sessionName);
@@ -1204,7 +1203,6 @@ export class ClaudeCompactionEnforcer {
       return { triggered: false, reason: "invalid_policy" };
     }
     this.recoverAfterRestart(input, policy.preparationRoot || null);
-    await this.noteEndedPreparation(input, policy);
     // A trigger sample must not block work already owed to this occupant. Restore reads can
     // themselves exceed a low threshold; keep the same stage, authorization and idle gates.
     const drained = await this.drainPendingStage(input, policy);
@@ -1396,7 +1394,7 @@ export class ClaudeCompactionEnforcer {
     do {
       const remaining = attempt.deadlineAt! - this.now();
       if (remaining <= 0) return failed(this.mapReady(attempt) ? "map_complete_seat_never_idle" : "preparation_incomplete");
-      try { idle = await this.sessionTransport.waitUntilIdle(input.sessionName, remaining, attempt.controller.signal, { allowDraft: true }); }
+      try { idle = await this.sessionTransport.waitUntilIdle(input.sessionName, remaining, attempt.controller.signal); }
       catch { idle = { ok: false, reason: "transport_unavailable" }; }
       this.reconcilePreparations();
       if (idle.ok || idle.reason !== "target_needs_input" || attempt.status !== "waiting") break;
@@ -1547,7 +1545,7 @@ export class ClaudeCompactionEnforcer {
         overflowCompactInstruction: composed.instructionInCommand ? null : attempt.compactInstruction,
         usedPercentage: input.usedPercentage!, thresholdPercent: policy.thresholdPercent,
         preCompactInstruction: policy.preCompactInstruction, preparation: attempt, constructedAt: this.now(),
-      }), { openrigNotice: true, allowDraftWhileIdle: true, ...idleGate, beforeWrite: () => {
+      }), { openrigNotice: true, preserveInput: true, ...idleGate, beforeWrite: () => {
         this.reconcilePreparations();
         if (this.pendingPreCompactPrep.get(input.sessionName) !== attempt || attempt.controller.signal.aborted || attempt.status === "stopped")
           throw new DeliveryGuardError("preparation_stopped", "This compaction attempt ended before preparation input.");
@@ -1620,7 +1618,7 @@ export class ClaudeCompactionEnforcer {
       // Idle comes from a live pane read: the newest hook can predate the rest of the preparation turn.
       const idleWait = this.compactSendIdleWaitMs + CAPTURE_READABLE_BUDGET_MS;
       return this.sessionTransport.send(input.sessionName, composed.command,
-        { openrigNotice: true, allowDraftWhileIdle: true, beforeWrite, ...(attempt.deadlineAt !== null ? { waitForIdleMs: attempt.prepared ? idleWait : Math.max(1, Math.min(idleWait, this.compactDeadline(attempt)! - this.now())), readinessFromPaneOnly: true } : {}) });
+        { openrigNotice: true, preserveInput: true, beforeWrite, ...(attempt.deadlineAt !== null ? { waitForIdleMs: attempt.prepared ? idleWait : Math.max(1, Math.min(idleWait, this.compactDeadline(attempt)! - this.now())), readinessFromPaneOnly: true } : {}) });
     };
     let result: Pick<SendResult, "ok" | "reason" | "outcome" | "sent" | "inputPhase">;
     let guardRefused = false;
@@ -1679,6 +1677,15 @@ export class ClaudeCompactionEnforcer {
     return { ok: true };
   }
 
+  /** What the seat's input box holds now, as `rig capture --json` reads it (`empty`, `likely-suggestion`, `typed`, ...), or
+   *  why it couldn't be read. Never throws. */
+  private async inputBoxStatus(sessionName: string): Promise<string> {
+    try {
+      const capture = await this.sessionTransport.capture(sessionName, { inputBox: true, readable: { width: 120, height: 40 } });
+      return capture.inputBox?.status ?? capture.reason ?? "unknown";
+    } catch { return "unknown"; }
+  }
+
   /** /compact was submitted, or may have been: its restore steps are owed next, and no new automatic compaction starts
    *  until they finish. */
   private owePostCompactRestore(sessionName: string, attempt: PreparationAttempt, carry: RestoreCarry | undefined): void {
@@ -1756,12 +1763,9 @@ export class ClaudeCompactionEnforcer {
 
   /** Start the record for a new attempt, beside its map. */
   private startRecord(sessionName: string, attempt: PreparationAttempt): void {
-    const previous = this.records.get(sessionName)?.record;
-    const preparationFailures = attempt.mode === "automatic" && previous?.occupantGeneration === attempt.occupantGeneration
-      ? previous?.preparationFailures : undefined;
     this.records.set(sessionName, { file: path.join(path.dirname(attempt.mapPath), COMPACTION_RECORD_FILE), record: {
       daemon: this.daemonInstance, attemptId: attempt.attemptId, session: sessionName, occupantGeneration: attempt.occupantGeneration, mode: attempt.mode,
-      phase: "preparing", delivery: attempt.delivery, deadlineAt: attempt.deadlineAt, sends: attempt.sends, preparationFailures,
+      phase: "preparing", delivery: attempt.delivery, deadlineAt: attempt.deadlineAt, sends: attempt.sends,
       mapPath: attempt.mapPath, marker: attempt.marker, compactInstruction: attempt.compactInstruction,
       preparationRoot: attempt.preparationRoot, ...(attempt.sessionId ? { sessionId: attempt.sessionId } : {}),
       ...(attempt.afterEnded !== undefined ? { afterEnded: attempt.afterEnded, afterEndedSessionId: attempt.afterEndedSessionId ?? null } : {}),
@@ -1835,8 +1839,6 @@ export class ClaudeCompactionEnforcer {
     if (record.daemon === this.daemonInstance || record.phase === "done") return;
     const live = this.resolveOccupantGeneration?.(session) ?? this.sessionTransport.deliveryGuard?.maybeTarget(session)?.occupant ?? null;
     if (record.phase === "ended") {
-      if (record.occupantGeneration !== null && live === record.occupantGeneration
-        && this.now() - (record.endedAt ?? record.updatedAt) < this.restartRecoveryWindowMs) this.records.set(session, newest);
       // An ended automatic attempt of this occupant whose request was delivered and whose /compact was never sent keeps a
       // busy seat from a fresh request (#2682), as in memory. Only a stop writes endedAt; a /compact that was sent (and then
       // cancelled) never stopped, and one whose result was lost is marked uncertainCompact.
@@ -1914,44 +1916,6 @@ export class ClaudeCompactionEnforcer {
     }
   }
 
-  /** Count ended attempts, never individual send calls. The current attempt record carries the run across restarts. */
-  private async noteEndedPreparation(input: EnforcerInput, policy: ClaudeCompactionPolicy): Promise<void> {
-    const current = this.records.get(input.sessionName);
-    if (!current) return;
-    const r = current.record, generation = this.liveGeneration(input.sessionName);
-    const interrupted = this.pendingPreCompactPrep.get(input.sessionName)?.cancelled || !policy.enabled || input.usedPercentage == null || input.usedPercentage < policy.thresholdPercent
-      || r.mode !== "automatic" || generation === null || generation !== r.occupantGeneration
-      || (r.sessionId && input.sessionId && r.sessionId !== input.sessionId)
-      || r.delivery !== "not_sent" || r.belowSince !== undefined || r.compactSending || r.uncertainCompact
-      || ["preparation_cancelled", "preparation_replaced", "stale_generation", "disabled", "daemon_restarted",
-        "below_threshold_while_prepared", "prepared_map_stale", "occupant_generation_unavailable"].includes(r.reason ?? "");
-    if (interrupted) {
-      if (r.preparationFailures) this.persist(input.sessionName, r.attemptId, { preparationFailures: undefined });
-      return;
-    }
-    if (r.phase !== "ended" || r.endedAt === undefined || r.sends === 0) return;
-    const old = r.preparationFailures;
-    const run: PreparationFailureRun = old && Number.isSafeInteger(old.endedAttempts) && old.endedAttempts >= 0
-      && typeof old.lastAttemptId === "string" && (old.sessionId === null || typeof old.sessionId === "string")
-      && old.occupantGeneration === generation
-      && (!old.sessionId || !input.sessionId || old.sessionId === input.sessionId) ? old
-      : { occupantGeneration: generation, sessionId: input.sessionId ?? null, endedAttempts: 0, lastAttemptId: "" };
-    if (run.lastAttemptId !== r.attemptId) {
-      run.lastAttemptId = r.attemptId; run.endedAttempts++;
-      this.persist(input.sessionName, r.attemptId, { preparationFailures: run });
-    }
-    if (run.endedAttempts < 3 || run.reportedAttempts === run.endedAttempts || !this.onPreparationFailure) return;
-    try {
-      await this.onPreparationFailure({ sessionName: input.sessionName, occupantGeneration: generation,
-        attemptId: r.attemptId, mapPath: r.mapPath, statePath: current.file, endedAttempts: run.endedAttempts,
-        reason: r.reason ?? "send_failed", ...(r.inputBox ? { inputBox: r.inputBox } : {}) });
-      if (this.records.get(input.sessionName) === current && current.record.preparationFailures === run) {
-        run.reportedAttempts = run.endedAttempts;
-        this.persist(input.sessionName, r.attemptId, { preparationFailures: run });
-      }
-    } catch { /* The same diagnostic stays eligible on existing polls; no input effect or new scheduler. */ }
-  }
-
   private stopPreparation(session: string, reason: string, expected?: PreparationAttempt): void {
     const attempt = this.pendingPreCompactPrep.get(session);
     if ((expected && expected !== attempt) || !attempt || attempt.status === "stopped" || attempt.status === "compact-sent") return;
@@ -1969,10 +1933,7 @@ export class ClaudeCompactionEnforcer {
 
   cancelPreparation(session: string): PreparationState | null {
     const attempt = this.pendingPreCompactPrep.get(session);
-    if (attempt) {
-      attempt.cancelled = true;
-      this.persist(session, attempt.attemptId, { preparationFailures: undefined });
-    }
+    if (attempt) attempt.cancelled = true;
     if (attempt?.status === "compact-sent" || this.pendingPostCompactRestore.has(session)) {
       // Abandon management, not the provider command: keep its unconfirmed receipt and map.
       // Do not clear automatic dedup/re-arm state or send any input. A later explicit retry
@@ -1987,6 +1948,7 @@ export class ClaudeCompactionEnforcer {
       this.pendingStageGeneration.delete(session);
       this.restoreCarry.delete(session);
       this.afterUncertainCompact.delete(session);
+      this.lastInputBox.delete(session);
       this.setManualStage(session, "cancelled", "compaction_cancelled");
     } else {
       this.stopPreparation(session, "preparation_cancelled");
@@ -2029,6 +1991,7 @@ export class ClaudeCompactionEnforcer {
     this.records.delete(sessionName);
     this.stranded.delete(sessionName);
     this.afterUncertainCompact.delete(sessionName);
+    this.lastInputBox.delete(sessionName);
     this.lastAutoCompactAt.delete(sessionName);
     this.postCompactRestoreCooldownUntil.delete(sessionName);
     this.triggeredAboveThreshold.delete(sessionName);

packages/daemon/src/domain/compaction-restore-notice.ts

@@ -1,6 +1,5 @@
-import { createHash } from "node:crypto";
 import type Database from "better-sqlite3";
-import { type PreparationFailure, type RestoreDeliveryFailure } from "./claude-compaction-enforcer.js";
+import { UNREADABLE_POLLS_BEFORE_DELIVERY, type RestoreDeliveryFailure } from "./claude-compaction-enforcer.js";
 import type { QueueRepository } from "./queue-repository.js";
 import { defaultResolveOrchestrator } from "./queue-owner.js";
 import { workflowHumanDestination } from "./workflow-human-destination.js";
@@ -26,26 +25,6 @@ export async function reportRestoreDeliveryFailure(db: Database.Database, queue:
     summary, wakeTrigger: "openrig",
     tags: ["compaction", "restore-delivery", `attempt:${attemptId}`],
     evidenceRef: failure.mapPath,
-    body: `${summary}. OpenRig still owes this stage after repeated delivery failures. Please inspect the seat and decide whether intervention is needed. This does not prove native compaction finished. The existing poll retries with its delivery checks. Inspect rig compact ${shellQuote(sessionName)} --state --json and rig seat status ${shellQuote(sessionName)} before acting. Do not issue another /compact. Restore map: ${failure.mapPath}. The latest reason is in this item's summary and transition notes. Restore input follows ordinary delivery defaults: a draft can receive and submit the restore text unless the operator enabled draft protection. The seat's draft-aware mode and transport.draft_delivery=refuse keep their existing protections. A lost /compact result is not replayed. To restore the seat yourself instead: first run rig compact ${shellQuote(sessionName)} --cancel, so OpenRig stops owing the remaining steps (a step it is already pasting at that moment can still land), then send the seat a restore request naming the restore map${failure.markerPath ? `, its restore-pending marker ${failure.markerPath}` : ""} and the packet the marker names.`,
-  });
-}
-
-/** One diagnostic per seat/occupant: an owner's disposition stands across automatic attempt IDs and restarts. */
-export async function reportPreparationFailure(db: Database.Database, queue: QueueRepository, failure: PreparationFailure): Promise<void> {
-  const key = createHash("sha256").update(`${failure.sessionName}\0${failure.occupantGeneration}`).digest("hex").slice(0, 24);
-  const qitemId = `qitem-compaction-preparation-${key}`;
-  const summary = `Managed preparation for ${failure.sessionName}: ${failure.endedAttempts} ended unsent attempts (${failure.reason})`;
-  const evidence = `Attempt ${failure.attemptId}; reason ${failure.reason}; input box ${failure.inputBox ? JSON.stringify(failure.inputBox) : "reading unavailable"}`;
-  const existing = queue.getById(qitemId);
-  if (existing) {
-    if (["pending", "in-progress", "blocked"].includes(existing.state) && existing.summary !== summary)
-      queue.update({ qitemId, actorSession: "system:compaction", summary, transitionNote: `${summary}. ${evidence}` });
-    return;
-  }
-  await queue.create({ qitemId, sourceSession: failure.sessionName,
-    destinationSession: defaultResolveOrchestrator(db, failure.sessionName) ?? workflowHumanDestination(),
-    summary, wakeTrigger: "openrig", tags: ["compaction", "preparation-delivery", `occupant:${failure.occupantGeneration}`],
-    evidenceRef: failure.statePath,
-    body: `${summary}. ${evidence}. These are ended automatic attempts while usage remained above the threshold, not repeated sends in one attempt. Inspect rig compact ${shellQuote(failure.sessionName)} --state --json and the seat before choosing an existing supported recovery route. This finding does not force input, skip a map, reset the seat, or change its account or model. No input is authorized by this notice.`,
+    body: `${summary}. OpenRig still owes this stage after repeated delivery failures. Please inspect the seat and decide whether intervention is needed. This does not prove native compaction finished. The existing poll retries with its delivery checks. Inspect rig compact ${shellQuote(sessionName)} --state --json and rig seat status ${shellQuote(sessionName)} before acting. Do not issue another /compact. Restore map: ${failure.mapPath}. The latest reason is in this item's summary and transition notes. When the seat's input box can't be read but doesn't read typed, OpenRig sends the step itself after ${UNREADABLE_POLLS_BEFORE_DELIVERY} idle polls, recorded as over an unreadable box. That doesn't apply after a /compact whose result was lost, where each step waits for an empty box, or on a seat whose draft-aware delivery or typing guard holds OpenRig's input: those keep holding, draft-aware delivery until the seat's mode is set to off and a typing guard until its level admits OpenRig's input. To restore the seat yourself instead: first run rig compact ${shellQuote(sessionName)} --cancel, so OpenRig stops owing the remaining steps (a step it is already pasting at that moment can still land), then send the seat a restore request naming the restore map${failure.markerPath ? `, its restore-pending marker ${failure.markerPath}` : ""} and the packet the marker names.`,
   });
 }

packages/daemon/src/domain/gateway/slack/settled-asks.ts

@@ -12,6 +12,10 @@ import { TYPED_REPLY_HINT, CHANNEL_REPLY_HINT, HUMAN_REQUIRED_REPLY_HINT, QUESTI
 const UPDATE = "chat.update";
 const EDITED = "slack-ask-settled";
 const RATE_LIMIT = "slack-ask-edit-rate-limit";
+/** The posted message itself is gone or can never be edited by us: recorded, so a restart doesn't ask Slack again. */
+const UNAVAILABLE = "slack-ask-settle-unavailable";
+// cant_update_message means our token didn't author the message, which a token change can repair, so it isn't here.
+const goneErrors = new Set(["message_not_found", "thread_not_found", "original message unavailable"]);
 const STATUS_BLOCK = "or-settled";
 const permanentErrors = new Set(["missing_scope", "not_in_channel", "channel_not_found", "is_archived",
   "cant_update_message", "invalid_blocks", "message_not_found", "thread_not_found", "original message unavailable"]);
@@ -133,7 +137,19 @@ export class SlackAskSettlement {
   private failed(id: string, key: string, method: string, result: WebApiResult): number {
     this.lastFailure = `${id}: ${method} not done (${result.error ?? result.status})`;
     this.opts.log(`slack ask settlement retained: ${this.lastFailure}`);
-    if (permanentErrors.has(result.error ?? "")) { this.unavailable.add(key); return 0; }
+    if (permanentErrors.has(result.error ?? "")) {
+      this.unavailable.add(key);
+      // Scope, membership, the token or a rendering fault can be fixed, so those are asked again after a restart; a gone message can't.
+      if (goneErrors.has(result.error ?? "")) {
+        try {
+          this.opts.queue.update({ qitemId: id, actorSession: "daemon@kernel",
+            transitionNote: `${UNAVAILABLE}${key.slice(EDITED.length)} error=${(result.error ?? "").replace(/\s+/g, "_")}` });
+        } catch (error) {
+          this.opts.log(`slack ask settlement: couldn't record ${id} as unavailable (${(error as Error).message}); skipped until restart`);
+        }
+      }
+      return 0;
+    }
     const count = (this.failures.get(key) ?? 0) + 1;
     this.failures.set(key, count);
     const wait = Math.max(Math.min(30_000 * 2 ** Math.min(count - 1, 7), 3_600_000), (result.retryAfterSeconds ?? 0) * 1000);
@@ -150,7 +166,7 @@ export class SlackAskSettlement {
     const item = this.opts.queue.getById(id);
     if (!item) return undefined;
     const after = transitions.filter((t) => t.transitionId > episode.transitionId && ![
-      "slack-posted ", "slack-owner-notification-posted ", EDITED + " ", RATE_LIMIT + " ",
+      "slack-posted ", "slack-owner-notification-posted ", EDITED + " ", RATE_LIMIT + " ", UNAVAILABLE + " ",
     ].some((prefix) => (t.transitionNote ?? "").startsWith(prefix)));
     const make = (t: QueueTransition, label?: string) => {
       const status = statusLine(t, label);
@@ -204,6 +220,9 @@ export class SlackAskSettlement {
       if (!stamp) continue; // No authoritative channel/timestamp: never guess a replacement post.
       const edited = `${EDITED} notification_key=${key} channel=${stamp.channel} message_ts=${messageTs}`;
       if (this.unavailable.has(edited)) continue;
+      // Recorded as gone for this exact message; a new post of the episode has its own message_ts and is tried.
+      const gone = `${UNAVAILABLE} notification_key=${key} channel=${stamp.channel} message_ts=${messageTs} `;
+      if (transitions.some((t) => t.transitionNote?.startsWith(gone))) continue;
       const desired = this.desired(id, episode, stamp.human, transitions);
       if (!desired) continue;
       const lastAck = transitions.filter((t) => t.transitionNote === edited || t.transitionNote?.startsWith(edited + " ")).at(-1);

packages/daemon/src/domain/gateway/slack/slack-recovery.ts

@@ -5,14 +5,18 @@ import path from "node:path";
 import type { FetchImpl } from "./slack-api.js";
 import type { InboundRouter } from "./inbound.js";
 import { ChannelRecovery } from "./channel-recovery.js";
-import { ThreadRecovery, threadKey } from "./thread-recovery.js";
+import { QUIET_RESCAN_MS, RELAYED_QUIET_RESCAN_MS, ThreadRecovery, threadKey } from "./thread-recovery.js";
 import { ChannelCoverageStore, ThreadCoverageStore, slackMicros, slackTimestamp, type ThreadCoverage } from "./state-store.js";
 import { budgetLeft, HISTORY, MethodBackoff, newRecoveryBudget, REPLIES, type RecoverySource } from "./recovery-budget.js";
+import { ACTIVE_QUEUE_STATES } from "../../queue-repository.js";
 
 /** A thread is tracked while its root was posted, or it closed, within this window. */
 export const THREAD_WINDOW_MS = 7 * 24 * 60 * 60 * 1000;
+/** A thread leaving the window within this long goes first after a gap: once it leaves, a missed reply is never recovered. */
+const EXPIRING_SOON_MS = 6 * 60 * 60 * 1000;
 
-export interface TrackedThread { channel: string; threadTs: string; state?: string; closedAt?: string | null }
+/** `askState`: the state of the queue row the thread's root belongs to; null when that row is gone. */
+export interface TrackedThread { channel: string; threadTs: string; state?: string; closedAt?: string | null; askState?: string | null }
 
 /** The channels this machine recovers: its default channel and its channel map's, except a channel
  *  its configuration relays to the machine that owns it (that machine recovers it). */
@@ -45,7 +49,7 @@ interface SlackRecoveryOptions {
   relayed?: boolean;
 }
 
-interface TrackedSource { source: ThreadRecovery; mapping: string }
+interface TrackedSource { source: ThreadRecovery; mapping: string; waiting: boolean; expiresAt: number }
 
 export class SlackRecovery {
   private readonly channels: ChannelRecovery[];
@@ -57,7 +61,9 @@ export class SlackRecovery {
   private pass?: Promise<void>;
   private stopped = false;
   private lastChannel: string | undefined;
-  private lastThread: string | undefined;
+  /** When live events last began arriving without a gap: the connector's start, or the latest Socket Mode connect that
+   *  followed a gap. A reply after it in a thread nobody waits on arrives live. */
+  private listeningSince: number;
   private threadsUnavailable: string | undefined;
   private initialized = false;
   private readonly adoptedHere = new Set<string>();
@@ -66,6 +72,7 @@ export class SlackRecovery {
 
   constructor(private readonly opts: SlackRecoveryOptions) {
     this.now = opts.now ?? Date.now;
+    this.listeningSince = this.now();
     this.channelStore = opts.channelStore ?? new ChannelCoverageStore(path.join(opts.stateDir, "slack-channel-coverage.json"));
     const channels: Array<{ channel: string | null; seed?: "adoption" }> = opts.channels.length ? opts.channels : [{ channel: null }];
     this.channels = channels.map((c) => new ChannelRecovery({ channel: c.channel, token: opts.token, stateDir: opts.stateDir,
@@ -94,6 +101,12 @@ export class SlackRecovery {
     } catch { /* an unreadable checkpoint file is never replaced; the channel keeps waiting */ }
   }
 
+  /** The Socket Mode connection opened after a gap (not a refresh that overlapped the old one). Replies sent in the gap
+   *  past Slack's own retries weren't delivered live, so every settled thread is due once more. */
+  connected(): void {
+    this.listeningSince = this.now();
+  }
+
   stop(): void {
     this.stopped = true;
     for (const channel of this.channels) channel.stop();
@@ -145,14 +158,13 @@ export class SlackRecovery {
     this.restoreMethodHolds();
     // Channels and threads take turns one step at a time, each kind in its own rotation. A source with
     // more to scan rejoins the back of its kind's queue; a kind with nothing left gives its turns away.
-    const queues: RecoverySource[][] = [rotate(this.activeChannels(), this.lastChannel), rotate(this.trackedThreadSources(), this.lastThread)];
+    const queues: RecoverySource[][] = [rotate(this.activeChannels(), this.lastChannel), this.trackedThreadSources()];
     const budget = newRecoveryBudget(this.now());
     let kind = 0;
     while (!this.stopped && budgetLeft(budget, this.now()) && (queues[0]!.length || queues[1]!.length)) {
       if (!queues[kind]!.length) kind = 1 - kind;
       const source = queues[kind]!.shift()!;
       if (source instanceof ThreadRecovery) {
-        this.lastThread = source.key;
         const before = { accepted: source.accepted, deadLettered: source.deadLettered };
         const more = await source.step(budget, this.backoff);
         this.threadTotals.accepted += source.accepted - before.accepted;
@@ -166,6 +178,8 @@ export class SlackRecovery {
     }
   }
 
+  /** The threads this pass scans, in the order it scans them. Refreshes the tracked set from the seat map (a thread keeps its
+   *  source while its mapping is unchanged), then keeps only the threads that are due and sorts them by `threadOrder`. */
   private trackedThreadSources(): ThreadRecovery[] {
     if (!this.opts.trackedThreads || !this.opts.token) return [];
     let stored: Record<string, ThreadCoverage>;
@@ -178,20 +192,27 @@ export class SlackRecovery {
       return [];
     }
     const owned = this.activeChannels().flatMap((c) => (c.channel ? [c.channel] : []));
-    const since = new Date(this.now() - THREAD_WINDOW_MS).toISOString();
+    const now = this.now();
+    const since = new Date(now - THREAD_WINDOW_MS).toISOString();
     const next = new Map<string, TrackedSource>();
     for (const t of this.opts.trackedThreads(owned, since)) {
       if (!owned.includes(t.channel) || slackMicros(t.threadTs) === null) continue;
       const key = threadKey(t.channel, t.threadTs);
       const mapping = `${t.state ?? ""}|${t.closedAt ?? ""}`;
+      const waiting = (ACTIVE_QUEUE_STATES as readonly string[]).includes(t.askState ?? "");
+      const closedMs = t.closedAt ? Date.parse(t.closedAt) : NaN;
+      const expiresAt = Math.max(Number(slackMicros(t.threadTs)! / 1000n), Number.isFinite(closedMs) ? closedMs : 0) + THREAD_WINDOW_MS;
       const prior = this.threads.get(key);
-      // A changed mapping gets a fresh source (so a thread Slack couldn't find is asked again); saved coverage carries over.
-      next.set(key, prior && prior.mapping === mapping ? prior : { mapping, source: new ThreadRecovery({ channel: t.channel,
-        threadTs: t.threadTs, token: this.opts.token, store: this.threadStore, stored: stored[key], router: this.opts.router,
-        fetchImpl: this.opts.fetchImpl, now: this.now, stopped: () => this.stopped }) });
+      // A changed mapping gets a fresh source; saved coverage, and its turn among retries, carry over.
+      if (prior && prior.mapping === mapping) { next.set(key, { ...prior, waiting, expiresAt }); continue; }
+      const source = new ThreadRecovery({ channel: t.channel, threadTs: t.threadTs, token: this.opts.token, store: this.threadStore,
+        stored: stored[key], router: this.opts.router, fetchImpl: this.opts.fetchImpl, now: this.now, stopped: () => this.stopped });
+      source.lastStalledAt = prior?.source.lastStalledAt;
+      next.set(key, { mapping, waiting, expiresAt, source });
     }
     this.threads = next;
-    return [...next.values()].map((t) => t.source).sort((a, b) => (a.key < b.key ? -1 : 1));
+    const quiet = this.opts.relayed ? RELAYED_QUIET_RESCAN_MS : undefined;
+    return threadOrder([...next.values()].filter((t) => t.source.due(now, t.waiting, this.listeningSince, quiet)), this.listeningSince, now);
   }
 
   status() {
@@ -226,6 +247,8 @@ export class SlackRecovery {
         "coverage means scanned available history; dead letters are custody, not delivery",
         "five-second settle margin; larger clock skew or history visibility lag remains unverified",
         "four calls / 100 entries / 15 seconds per pass, taken in turn by channels and threads; no global chronological ordering",
+        "each thread is scanned once after boot or a reconnect from a gap (twice if the pass at the connect itself reached it, since that pass stops at the settle margin): any leaving the seven-day window within six hours first, then those whose ask waits on the person, then the rest, newest first; "
+          + `after that hourly while its ask waits, otherwise ${this.opts.relayed ? "every three hours (relayed: no gap is visible here)" : "daily"}`,
       ],
       channels: others.map(({ limits: _limits, ...rest }) => rest),
       threads: {
@@ -237,11 +260,41 @@ export class SlackRecovery {
         oldestPendingSeconds: pending.length ? Math.max(...pending.map(behind)) : null,
         backoff: byReason(["backoff"]),
         incomplete: byReason(["incomplete"]),
+        settled: count("settled"),
+        settledRescanHours: (this.opts.relayed ? RELAYED_QUIET_RESCAN_MS : QUIET_RESCAN_MS) / 3_600_000,
       },
     };
   }
 }
 
+/** The order a pass takes the due threads in, derived from saved coverage so a restart keeps it. First every thread not
+ *  yet scanned since live events last began: any about to leave the window, soonest first; then those whose ask waits on
+ *  the person, then the rest, each newest root first. Then the backstop rescans, least recently scanned first. So a boot
+ *  reaches the newest open ask first, after only the few threads that would otherwise never be scanned. A thread whose
+ *  call got nowhere since then (an error, or a page that couldn't be trusted) waits behind every other unscanned thread,
+ *  least recently stalled first, so a thread that keeps failing can't hold the same turn pass after pass. A thread that
+ *  made progress but stopped short, at the settle margin or partway through a long backlog, keeps its place. */
+function threadOrder(threads: TrackedSource[], listeningSince: number, now: number): ThreadRecovery[] {
+  const floor = BigInt(listeningSince) * 1000n;
+  const scannedAt = (t: ThreadRecovery) => { const through = slackMicros(t.coveredThrough)!; return through >= floor ? through : -1n; };
+  const stalledAt = (t: ThreadRecovery) => (t.lastStalledAt !== undefined && t.lastStalledAt >= listeningSince ? t.lastStalledAt : -1);
+  const root = (t: ThreadRecovery) => slackMicros(t.threadTs)!;
+  return threads.sort((a, b) => {
+    const sa = scannedAt(a.source), sb = scannedAt(b.source);
+    if (sa !== sb) return sa < sb ? -1 : 1;
+    if (sa === -1n) {
+      const qa = stalledAt(a.source), qb = stalledAt(b.source);
+      if (qa !== qb) return qa - qb;
+      const ea = a.expiresAt - now < EXPIRING_SOON_MS, eb = b.expiresAt - now < EXPIRING_SOON_MS;
+      if (ea !== eb) return ea ? -1 : 1;
+      if (ea && a.expiresAt !== b.expiresAt) return a.expiresAt - b.expiresAt;
+    }
+    if (a.waiting !== b.waiting) return a.waiting ? -1 : 1;
+    const ra = root(a.source), rb = root(b.source);
+    return ra === rb ? (a.source.key < b.source.key ? -1 : 1) : ra > rb ? -1 : 1;
+  }).map((t) => t.source);
+}
+
 function rotate<T extends { key: string }>(sources: T[], last: string | undefined): T[] {
   const start = last === undefined ? 0 : sources.findIndex((s) => s.key === last) + 1;
   return [...sources.slice(start), ...sources.slice(0, start)];

packages/daemon/src/domain/gateway/slack/socket-inbound.ts

@@ -57,7 +57,7 @@ export interface SocketInboundDeps {
   /** How often liveness is checked (default 5s). */
   livenessCheckMs?: number;
   receipts?: InboundReceiptStore;
-  recovery?: { run(): Promise<void>; stop(): void };
+  recovery?: { run(): Promise<void>; stop(): void; connected?(): void };
   log?: (msg: string) => void;
 }
 
@@ -344,6 +344,10 @@ export function startSocketInbound(appToken: string, router: InboundRouter, deps
         status.unechoedPosts = 0;
         status.autoReconnectSuppressedUntil = undefined; // a new connection: nothing is held back for it
         receipt({ generation: conn.generation, status: "connected" });
+        // Live events had a gap unless this connection overlaps one Slack asked to refresh, which drains until Slack closes it.
+        const overlapped = previous && previous !== conn && previous.retired && !previous.closed &&
+          (previous.retiredFor === "refresh_requested" || previous.retiredFor === "warning");
+        if (!overlapped) deps.recovery?.connected?.();
         // A connection replaced for a failed signal is presumed broken: close it now. One Slack
         // asked to refresh stays open so it drains what it is handed, until Slack closes it.
         if (previous && previous !== conn && previous.retired && !previous.closed) {

packages/daemon/src/domain/gateway/slack/state-store.ts

@@ -402,6 +402,8 @@ export interface ThreadCoverage {
   nextRetryAt?: number;
   /** Whether nextRetryAt came from Slack's rate limit, which holds the whole method. */
   rateLimited?: boolean;
+  /** When Slack last answered thread_not_found (ms). Saved, so a restart doesn't ask again at once. */
+  notFoundAt?: number;
 }
 
 /** Scan boundaries for tracked threads, keyed `channel:thread_ts`. A separate file, so a channel
@@ -409,6 +411,8 @@ export interface ThreadCoverage {
 export class ThreadCoverageStore {
   constructor(private readonly file: string, private readonly fsops: StateFsOps = nodeStateFs) {}
 
+  /** Every saved thread checkpoint, or none if the file doesn't exist yet. Throws on a malformed file or checkpoint, so a
+   *  bad file is never overwritten. */
   read(): Record<string, ThreadCoverage> {
     const raw = readOptional(this.file, this.fsops);
     if (raw === undefined) return {};
@@ -418,7 +422,8 @@ export class ThreadCoverageStore {
       const start = slackMicros(c?.coverageStart), through = slackMicros(c?.coveredThrough);
       if (start === null || through === null || start > through ||
         (c.rateLimited !== undefined && typeof c.rateLimited !== "boolean") ||
-        (c.nextRetryAt !== undefined && (!Number.isFinite(c.nextRetryAt) || c.nextRetryAt < 0))) throw new Error("invalid thread checkpoint");
+        (c.nextRetryAt !== undefined && (!Number.isFinite(c.nextRetryAt) || c.nextRetryAt < 0)) ||
+        (c.notFoundAt !== undefined && (!Number.isFinite(c.notFoundAt) || c.notFoundAt < 0))) throw new Error("invalid thread checkpoint");
       if (c.pending) {
         const upper = slackMicros(c.pending.upper);
         if (upper === null || upper <= through) throw new Error("invalid pending interval");

packages/daemon/src/domain/gateway/slack/thread-recovery.ts

@@ -6,6 +6,15 @@ import { slackMicros, slackTimestamp, type ThreadCoverage, type ThreadCoverageSt
 import { budgetLeft, REPLIES, type MethodBackoff, type RecoveryBudget, type RecoverySource } from "./recovery-budget.js";
 /** Slack's page cap for apps distributed outside the Marketplace. */
 const PAGE = 15;
+/** How long a thread Slack couldn't find, or a settled thread already scanned since live events began, waits before it's
+ *  asked again. */
+export const QUIET_RESCAN_MS = 24 * 60 * 60 * 1000;
+/** The same for a thread whose ask still waits on the person: a shorter backstop, should Slack drop a live event without
+ *  the connection showing a gap. */
+export const WAITING_RESCAN_MS = 60 * 60 * 1000;
+/** A settled thread's backstop on a machine the hub relays to: it has no connection of its own, so it can't see a gap in
+ *  relayed events, and checks about as often as a full rotation of tracked threads used to. */
+export const RELAYED_QUIET_RESCAN_MS = 3 * 60 * 60 * 1000;
 
 export function threadKey(channel: string, threadTs: string): string {
   return `${channel}:${threadTs}`;
@@ -30,10 +39,11 @@ export class ThreadRecovery implements RecoverySource {
   lastScanAt: string | undefined;
   accepted = 0;
   deadLettered = 0;
+  /** When a call for the thread's replies last got nowhere (ms): an error, a page that couldn't be trusted, or a landing
+   *  that couldn't finish. Orders retries fairly. A call that made progress, or stopped short at the settle margin or the
+   *  pass budget, never sets it, so that thread keeps its first-attempt priority. In memory only. */
+  lastStalledAt: number | undefined;
   private coverage: ThreadCoverage;
-  /** Slack can't find the thread. This source stays incomplete, without a call, until the coordinator replaces it because its
-   *  mapping or the configuration changed. Not proof of deletion, and never set by a transient error. */
-  private missing = false;
 
   constructor(private readonly opts: ThreadRecoveryOptions) {
     this.coverage = opts.stored ?? { coverageStart: opts.threadTs, coveredThrough: opts.threadTs };
@@ -41,13 +51,35 @@ export class ThreadRecovery implements RecoverySource {
 
   get key(): string { return `thread:${threadKey(this.opts.channel, this.opts.threadTs)}`; }
   get coveredThrough(): string { return this.coverage.coveredThrough; }
+  get threadTs(): string { return this.opts.threadTs; }
+
+  /** Whether this thread has anything for a pass to do. Partway through an interval: always. Slack couldn't find it: once
+   *  a day (not proof of deletion, and never set by a transient error). Scanned through `listeningSince` (ms, when live
+   *  events last began arriving without a gap): nothing, since replies after that arrive live, until a backstop rescan
+   *  an hour on for an ask still waiting on the person, a day on for any other thread. Coverage must reach
+   *  `listeningSince` itself: a scan in the first seconds after connecting stops at the settle margin, short of it. */
+  due(now: number, waiting: boolean, listeningSince: number, quietRescanMs = QUIET_RESCAN_MS): boolean {
+    if (this.coverage.notFoundAt !== undefined && now - this.coverage.notFoundAt < QUIET_RESCAN_MS) {
+      this.state = "incomplete"; this.reason = "thread_not_found"; return false;
+    }
+    if (this.coverage.pending) return true;
+    const through = Number(slackMicros(this.coverage.coveredThrough)! / 1000n);
+    if (through >= listeningSince && now - through < (waiting ? Math.min(WAITING_RESCAN_MS, quietRescanMs) : quietRescanMs)) {
+      // A waiting ask stays "scanned" in status; only a thread nobody waits on counts as settled.
+      if (!waiting && (this.state === "pending" || this.state === "scanned")) { this.state = "settled"; this.reason = undefined; }
+      if (waiting && this.state === "settled") this.state = "scanned"; // its ask was reopened
+      return false;
+    }
+    if (this.state === "settled") this.state = "pending";
+    return true;
+  }
 
   async step(budget: RecoveryBudget, backoff: MethodBackoff): Promise<"more" | "done"> {
     if (this.opts.stopped()) return "done";
     try {
       return await this.page(budget, backoff);
     } catch {
-      if (!this.opts.stopped()) { this.state = "incomplete"; this.reason = "storage-or-routing-failed"; }
+      if (!this.opts.stopped()) { this.state = "incomplete"; this.reason = "storage-or-routing-failed"; this.lastStalledAt = this.opts.now(); }
       return "done";
     }
   }
@@ -58,9 +90,11 @@ export class ThreadRecovery implements RecoverySource {
     this.coverage = next; // publish only after the atomic write succeeds
   }
 
+  /** One conversations.replies call over the pending interval (opening one up to five seconds ago if none is open), routing
+   *  each reply it returns. "more" when the interval has replies left past this page; "done" when it is covered, or the
+   *  thread is backing off, out of budget, not found or the page couldn't be trusted (each recorded in state and reason). */
   private async page(budget: RecoveryBudget, backoff: MethodBackoff): Promise<"more" | "done"> {
     const now = this.opts.now();
-    if (this.missing) { this.state = "incomplete"; this.reason = "thread_not_found"; return "done"; }
     if (backoff.blocked(REPLIES, now)) { this.state = "backoff"; this.reason = "rate-limited"; return "done"; }
     if (this.coverage.nextRetryAt && this.coverage.nextRetryAt > now) { this.state = "backoff"; return "done"; }
     if (!this.coverage.pending) {
@@ -83,8 +117,12 @@ export class ThreadRecovery implements RecoverySource {
     }, this.opts.fetchImpl, Math.min(5000, budget.deadline - this.opts.now()), "get-query");
     if (this.opts.stopped()) return "done";
     if (!r.ok) {
+      this.lastStalledAt = now;
       const known = ["missing_scope", "not_in_channel", "channel_not_found", "thread_not_found", "invalid_auth", "token_revoked"];
-      if (r.error === "thread_not_found") { this.missing = true; this.state = "incomplete"; this.reason = "thread_not_found"; return "done"; }
+      if (r.error === "thread_not_found") {
+        this.save({ ...c, notFoundAt: this.opts.now() });
+        this.state = "incomplete"; this.reason = "thread_not_found"; return "done";
+      }
       this.reason = r.status === 429 ? "rate-limited" : known.includes(r.error ?? "") ? r.error : "replies-api-unavailable";
       const retrySeconds = r.retryAfterSeconds ?? (r.status === 0 || r.status >= 500 ? 5 : 300);
       if (r.status === 429) backoff.hold(REPLIES, this.opts.now() + retrySeconds * 1000);
@@ -100,11 +138,11 @@ export class ThreadRecovery implements RecoverySource {
         (cursor !== undefined && typeof cursor !== "string") || replies.length > limit ||
         replies.some(m => !m || typeof m !== "object" || m.thread_ts !== this.opts.threadTs || slackMicros(m.ts) === null ||
           slackMicros(m.ts)! < lower || slackMicros(m.ts)! >= upper)) {
-      this.state = "incomplete"; this.reason = "invalid-replies-page"; return "done";
+      this.state = "incomplete"; this.reason = "invalid-replies-page"; this.lastStalledAt = now; return "done";
     }
     const ordered = [...replies].sort((a, b) => slackMicros(a.ts)! < slackMicros(b.ts)! ? -1 : 1);
     const more = r.json.has_more === true || (typeof cursor === "string" && cursor.length > 0);
-    if (more && !ordered.length) { this.state = "incomplete"; this.reason = "replies-page-no-progress"; return "done"; }
+    if (more && !ordered.length) { this.state = "incomplete"; this.reason = "replies-page-no-progress"; this.lastStalledAt = now; return "done"; }
     for (const message of ordered) {
       if (this.opts.stopped()) return "done";
       if (this.opts.now() >= budget.deadline || budget.entries >= 100) { this.state = "incomplete"; this.reason = "pass-budget"; return "done"; }
@@ -117,7 +155,7 @@ export class ThreadRecovery implements RecoverySource {
       const landed = await this.opts.router.route(ev, 0, () => !this.opts.stopped());
       if (this.opts.stopped()) return "done";
       if (landed.reason === "inflight" || landed.reason === "inactive") {
-        this.state = "incomplete"; this.reason = "landing-in-progress"; return "done";
+        this.state = "incomplete"; this.reason = "landing-in-progress"; this.lastStalledAt = now; return "done";
       }
       if (landed.disposition === "accepted") this.accepted++;
       if (landed.disposition === "dead-lettered") this.deadLettered++;
@@ -126,7 +164,7 @@ export class ThreadRecovery implements RecoverySource {
     // refused by sender policy or skipped as not ingestible.
     const through = more ? slackMicros(ordered.at(-1)!.ts)! + 1n : upper;
     if (through >= upper) {
-      this.save({ ...c, coveredThrough: c.pending!.upper, pending: undefined, nextRetryAt: undefined, rateLimited: undefined });
+      this.save({ ...c, coveredThrough: c.pending!.upper, pending: undefined, nextRetryAt: undefined, rateLimited: undefined, notFoundAt: undefined });
       this.state = "scanned"; return "done";
     }
     this.save({ ...c, coveredThrough: slackTimestamp(through), nextRetryAt: undefined, rateLimited: undefined });

packages/daemon/src/domain/gateway/slack/thread-seat-map.ts

@@ -149,14 +149,17 @@ export class ThreadSeatMap {
   }
 
   /** Missed-reply recovery's threads: roots in these channels posted at or after `since` (ISO), by the root's own Slack
-   *  timestamp, or closed at or after it. Never by opened_at: a rebuild from stamps sets that to the rebuild clock. */
-  trackedForRecovery(channels: string[], since: string): Array<{ channel: string; threadTs: string; state: string; closedAt: string | null }> {
+   *  timestamp, or closed at or after it. Never by opened_at: a rebuild from stamps sets that to the rebuild clock.
+   *  `askState` is the state of the queue row the root belongs to (null when that row is gone). The map's own `state`
+   *  says only whether the root can still be posted into; it doesn't follow the ask. */
+  trackedForRecovery(channels: string[], since: string): Array<{ channel: string; threadTs: string; state: string; closedAt: string | null; askState: string | null }> {
     if (!channels.length) return [];
     const rows = this.db
-      .prepare(`SELECT channel, thread_ts, state, closed_at FROM thread_seat_map
-         WHERE channel IN (${channels.map(() => "?").join(", ")}) AND (CAST(thread_ts AS REAL) >= ? OR closed_at >= ?)`)
-      .all(...channels, Date.parse(since) / 1000, since) as Array<{ channel: string; thread_ts: string; state: string; closed_at: string | null }>;
-    return rows.map((r) => ({ channel: r.channel, threadTs: r.thread_ts, state: r.state, closedAt: r.closed_at }));
+      .prepare(`SELECT t.channel, t.thread_ts, t.state, t.closed_at, q.state AS ask_state FROM thread_seat_map t
+         LEFT JOIN queue_items q ON q.qitem_id = t.conversation_id
+         WHERE t.channel IN (${channels.map(() => "?").join(", ")}) AND (CAST(t.thread_ts AS REAL) >= ? OR t.closed_at >= ?)`)
+      .all(...channels, Date.parse(since) / 1000, since) as Array<{ channel: string; thread_ts: string; state: string; closed_at: string | null; ask_state: string | null }>;
+    return rows.map((r) => ({ channel: r.channel, threadTs: r.thread_ts, state: r.state, closedAt: r.closed_at, askState: r.ask_state }));
   }
 
   close(threadTs: string, channel?: string): void {

packages/daemon/src/domain/session-transport.ts

@@ -1919,12 +1919,10 @@ export interface SendOpts {
   onInputEffect?: (phase: "pasted" | "submitted") => void;
   /** Internal managed lifecycle prerequisite; never accepted from HTTP send options. */
   beforeWrite?: () => void;
-  /** Explicit internal composer-preservation request. Managed compaction instead uses ordinary draft-delivery defaults.
+  /** Internal: managed lifecycle input must preserve a person's draft even when the optional typing guard is off.
    *  `"typed"` (#2575) holds only a box that reads typed or typed-with-suggestion; any other reading that isn't empty is
    *  pasted over and named in `pastedOverInputBox`, never called empty. */
   preserveInput?: true | "typed";
-  /** A managed idle wait follows the ordinary draft-delivery setting; menus and running work still wait/refuse. */
-  allowDraftWhileIdle?: true;
   /** The caller's request. Aborted while the send still waits, before any input starts, the send ends with
    *  `caller_cancelled` and never writes, then or later. Once input may have started it is ignored: the result stays
    *  attempted or written. Queue wakes and committed outbox deliveries have no caller and ignore it. */
@@ -2956,7 +2954,6 @@ export class SessionTransport {
       privateInput: opts.dangerouslyInteract === true,
       readinessFromPaneOnly: opts.readinessFromPaneOnly === true,
       postCompactTurnBoundary: opts.postCompactTurnBoundary === true,
-      allowDraft: opts.allowDraftWhileIdle === true,
       signal: callerSignal(opts), cancelledBy: "caller",
     });
     if (!preWait.ok) {
@@ -3241,7 +3238,6 @@ export class SessionTransport {
         privateInput: opts?.dangerouslyInteract === true,
         readinessFromPaneOnly: opts?.readinessFromPaneOnly === true,
         postCompactTurnBoundary: opts?.postCompactTurnBoundary === true,
-        allowDraft: opts?.allowDraftWhileIdle === true,
         signal: callerSignal(opts), cancelledBy: "caller",
       });
       waitEvidence = {
@@ -3302,7 +3298,6 @@ export class SessionTransport {
           binding: observed?.binding,
           readinessFromPaneOnly: opts.readinessFromPaneOnly === true,
           postCompactTurnBoundary: opts.postCompactTurnBoundary === true,
-          allowDraft: opts.allowDraftWhileIdle === true,
           signal: callerSignal(opts), cancelledBy: "caller",
         });
         waitEvidence = {
@@ -3330,7 +3325,7 @@ export class SessionTransport {
           readinessFromPaneOnly: opts.readinessFromPaneOnly === true,
           postCompactTurnBoundary: opts.postCompactTurnBoundary === true,
         }, opts.dangerouslyInteract === true);
-        if (recheck.state !== "idle" && !(opts.allowDraftWhileIdle && this.deliversOverDraft(recheck))) {
+        if (recheck.state !== "idle") {
           const late: { reason: string; error: string } =
             recheck.state === "needs_input"
               ? {
@@ -3526,9 +3521,6 @@ export class SessionTransport {
       const box = await this.captureInputBox(readTarget, boxRuntime, { remainingMs: INPUT_BOX_READ_DEADLINE_MS });
       if (box.status === "typed" || box.status === "typed-with-suggestion") {
         if (draftChoice === "refuse") {
-          heldInputBox = box.status;
-          const { status, confidence, evidence, cursor } = box;
-          heldInputBoxReading = { status, confidence, evidence, cursor };
           const activity: AgentActivity = { state: "needs_input", reason: "prompt_draft", evidenceSource: "pane_heuristic",
             sampledAt: this.now().toISOString(), evidence: `input box reads ${box.status}`, fallback: true };
           // Wait evidence keeps its waitedMs and attempts; the activity is this draft's, not the idle the wait ended on.
@@ -3773,11 +3765,10 @@ export class SessionTransport {
 
   /** Wait on the existing classifier without occupying the input/lifecycle lease. With `singleObservation`, one
    *  observation within `timeoutMs` and no wait after it: a busy reading returns at once. */
-  async waitUntilIdle(sessionName: string, timeoutMs: number, signal?: AbortSignal, opts?: { singleObservation?: boolean; allowDraft?: true }) {
+  async waitUntilIdle(sessionName: string, timeoutMs: number, signal?: AbortSignal, opts?: { singleObservation?: boolean }) {
     const meta = this.getSessionMeta(sessionName);
     return this.waitForIdle({ sessionName, runtime: meta.runtime, attachmentType: meta.attachmentType, timeoutMs, signal,
       binding: { sessionName, nodeId: meta.nodeId, occupant: meta.occupant, pane: meta.pane },
-      ...(opts?.allowDraft ? { allowDraft: true } : {}),
       ...(opts?.singleObservation ? { singleObservation: true } : {}) });
   }
 
@@ -3793,7 +3784,6 @@ export class SessionTransport {
     binding?: ObservedBinding;
     readinessFromPaneOnly?: boolean;
     postCompactTurnBoundary?: boolean;
-    allowDraft?: boolean;
     /** One observation only: a reading that isn't idle ends the wait at once. */
     singleObservation?: boolean;
     // Optional display value for timeout messages. The deadline always comes
@@ -3863,7 +3853,7 @@ export class SessionTransport {
       last = activity;
       const waitedMs = input.timeoutMs - (deadline - Date.now());
 
-      if (activity.state === "idle" || (input.allowDraft && this.deliversOverDraft(activity))) {
+      if (activity.state === "idle") {
         return { ok: true, activity, waitedMs, attempts };
       }
 
@@ -3972,7 +3962,7 @@ export class SessionTransport {
 
   /** transport.draft_delivery = deliver (the default): a person's draft at the prompt is the one needs_input an ordinary send or work
    *  notice goes through. Menus, permission prompts and hook-reported requests still refuse, and so does a send that must
-   *  not join a draft (an explicit preserveInput request). Idle waits opt in to this draft policy separately. */
+   *  not join a draft (preserveInput: managed compaction). A send that waits for idle keeps waiting. */
   private deliversOverDraft(readiness: AgentActivity, opts?: SendOpts): boolean {
     if (readiness.state !== "needs_input" || readiness.reason !== "prompt_draft" || readiness.evidenceSource !== "pane_heuristic") return false;
     if (opts?.preserveInput) return false;

packages/daemon/src/domain/terminal/tile-attach.ts

@@ -69,7 +69,7 @@ export function tileAttachArgv(
     'else view "$t" "" "$@"; seen=$?; fi;',
     'if [ "$seen" -ne 0 ] || ! here "$@"; then printf "\\033[?2004l"; stty sane 2>/dev/null; [ -z "$f" ] || rm -f "$f" "$f".*; return 1; fi;',
     'pz=; if LC_ALL=C grep -aqF -- "$b0" "$f"; then pz=1; put "$@"; fi;',
-    'me=$(tty 2>/dev/null) || me=; w=; if [ -n "$me" ]; then idle "$me" "$@" & w=$!; fi; printf "\\033[?2004l";',
+    'me=$(tty 2>/dev/null) || me=; w=; if [ -n "$me" ]; then idle "$me" "$@" & w=$!; fi;',
     'if [ -z "$pz" ] && [ -s "$f" ]; then go "$@"; else "$@" attach -t "$t"; fi;',
     'if [ -n "$w" ]; then kill "$w" 2>/dev/null; wait "$w" 2>/dev/null; fi; rm -f "$f" "$f".*; return 0; };',
     'take() { i=0; while [ "$i" -lt 256 ]; do',

packages/daemon/src/startup.ts

@@ -4,7 +4,7 @@ import { EnvironmentPlatform } from "./domain/environment-platform.js";
 import { OpenShellDriver } from "./adapters/openshell-driver.js";
 import { isBoundTerminalSeat } from "./domain/terminal-seat.js";
 import { seatRuntimeHomeBySession, seatAccountHome, effectiveRuntimeHome, type HomeRuntime } from "./domain/seat-runtime-home.js";
-import { reportPreparationFailure, reportRestoreDeliveryFailure } from "./domain/compaction-restore-notice.js";
+import { reportRestoreDeliveryFailure } from "./domain/compaction-restore-notice.js";
 import { ClaudeTokenUsage, claudeTokenUsageReader } from "./domain/provider/claude-token-usage.js";
 import type { ClaudeTokenIdentity } from "./domain/provider/provider-types.js";
 import { TerminalPolicyStore } from "./domain/terminal-policy.js";
@@ -2685,7 +2685,6 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
         // so a stage minted by a retired generation is refused for the successor. null = UNKNOWN → the
         // gate is inert (never compares a stale generation as if live).
         {
-          onPreparationFailure: (failure) => reportPreparationFailure(db, queueRepoInstance, failure),
           onRestoreDeliveryFailure: (failure) => reportRestoreDeliveryFailure(db, queueRepoInstance, failure),
           terminalPolicies: new TerminalPolicyStore(db),
           resolveOccupantGeneration: (sessionName) =>

packages/daemon/test/claude-compaction-enforcer.test.ts

@@ -129,7 +129,7 @@ describe("ClaudeCompactionEnforcer", () => {
     expect(send).toHaveBeenLastCalledWith(
       "claude-seat@rig",
       expect.stringContaining("OpenRig automatic compaction preparation is now required"),
-      { openrigNotice: true, allowDraftWhileIdle: true, beforeWrite: expect.any(Function) },
+      { openrigNotice: true, preserveInput: true, beforeWrite: expect.any(Function) },
     );
     expect(send.mock.calls[0]![1]).toContain("Current context usage is 80%; configured compaction threshold is 80%");
     expect(send.mock.calls[0]![1]).toContain("Operator pre-compaction instruction");
@@ -239,7 +239,7 @@ describe("ClaudeCompactionEnforcer", () => {
     expect(send).toHaveBeenLastCalledWith(
       "claude-seat@rig",
       expect.stringContaining("OpenRig post-compaction turn boundary"),
-      { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) },
+      { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) },
     );
 
     const belowRestore = await enforcer.maybeAutoCompact({
@@ -252,7 +252,7 @@ describe("ClaudeCompactionEnforcer", () => {
     expect(send).toHaveBeenLastCalledWith(
       "claude-seat@rig",
       expect.stringContaining(`${TEST_HOME}/compaction/restore-pending/[email protected]`),
-      { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) },
+      { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) },
     );
     expect(send.mock.calls[3]![1]).toContain("/tmp/claude.jsonl");
     expect(send.mock.calls[3]![1]).toContain("Restoration is the current task");
@@ -268,7 +268,7 @@ describe("ClaudeCompactionEnforcer", () => {
     expect(send).toHaveBeenLastCalledWith(
       "claude-seat@rig",
       expect.stringContaining("Now audit your compaction restore"),
-      { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) },
+      { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) },
     );
 
     now += 61_000;
@@ -446,7 +446,7 @@ describe("ClaudeCompactionEnforcer", () => {
     send.mockResolvedValueOnce(held);
     expect(await highTick()).toEqual({ triggered: false, reason: "send_failed" });
     expect(send.mock.calls.at(-1)?.[1]).toContain("restoring this Claude session after compaction");
-    expect(send.mock.calls.at(-1)?.[2]).toEqual({ openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true, beforeWrite: expect.any(Function) });
+    expect(send.mock.calls.at(-1)?.[2]).toEqual({ openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true, beforeWrite: expect.any(Function) });
     expect(await highTick()).toEqual({ triggered: true });
     expect(send.mock.calls.at(-1)?.[1]).toContain("restoring this Claude session after compaction");
     expect(await highTick()).toEqual({ triggered: true });
@@ -499,7 +499,7 @@ describe("ClaudeCompactionEnforcer", () => {
     // Without a sample, every stage's send must read the pane: no hook may stand in for it. Only the turn boundary, the first
     // step after /compact, reads past the compaction output above the input box.
     for (let i = 0; i < 3; i++) {
-      expect(send.mock.calls[queuedSends + i]?.[2]).toEqual({ openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true,
+      expect(send.mock.calls[queuedSends + i]?.[2]).toEqual({ openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true,
         beforeWrite: expect.any(Function), ...(i === 0 ? { postCompactTurnBoundary: true } : {}) });
     }
 
@@ -512,7 +512,7 @@ describe("ClaudeCompactionEnforcer", () => {
     const { enforcer, send, queuedSends } = await queueStageWithGen("gen-uuid-1");
     expect(await belowTick(enforcer)).toEqual({ triggered: true });
     // Not pane-only; the first stage is the turn boundary, which also reads past the compaction output.
-    expect(send.mock.calls[queuedSends]?.[2]).toEqual({ openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
+    expect(send.mock.calls[queuedSends]?.[2]).toEqual({ openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
   });
 
   it("the restore prompt tells a seat that did not compact to say so and skip the restore reading", () => {
@@ -1033,7 +1033,7 @@ describe("ClaudeCompactionEnforcer", () => {
         expect(send).toHaveBeenCalledWith(
           "claude-seat@rig",
           expect.stringContaining("OpenRig automatic compaction preparation"),
-          { openrigNotice: true, allowDraftWhileIdle: true, beforeWrite: expect.any(Function) },
+          { openrigNotice: true, preserveInput: true, beforeWrite: expect.any(Function) },
         );
 
         const compact = await enforcer.maybeAutoCompact({
@@ -1129,28 +1129,28 @@ describe("ClaudeCompactionEnforcer", () => {
       // Phase 1 — prep (a normal send, no wait option).
       expect(send.mock.calls[0]![0]).toBe(SEAT);
       expect(send.mock.calls[0]![1]).toContain("OpenRig manual compaction was requested");
-      expect(send.mock.calls[0]![2]).toEqual({ openrigNotice: true, allowDraftWhileIdle: true, beforeWrite: expect.any(Function) });
+      expect(send.mock.calls[0]![2]).toEqual({ openrigNotice: true, preserveInput: true, beforeWrite: expect.any(Function) });
       // Phase 2 — /compact WITH the trust-bridge AND wait-for-idle (two-phase).
       expect(send.mock.calls[1]![1]).toContain("Keep in the summary: OpenRig's later restore request is operator-authorized");
       expect(send.mock.calls[0]![1]).toContain("Trust-channel note: after compaction");
-      expect(send.mock.calls[1]![2]).toEqual({ openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true, beforeWrite: expect.any(Function) });
+      expect(send.mock.calls[1]![2]).toEqual({ openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true, beforeWrite: expect.any(Function) });
       expect(send).toHaveBeenCalledTimes(2);
 
       // The SAME maybeAutoCompact back-half (single restore path) drains it.
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20, transcriptPath: "/tmp/claude.jsonl",
       })).toEqual({ triggered: true });
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("OpenRig post-compaction turn boundary"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("OpenRig post-compaction turn boundary"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
 
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20, transcriptPath: "/tmp/claude.jsonl",
       })).toEqual({ triggered: true });
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Please respond to this normal user message now"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Please respond to this normal user message now"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
 
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20,
       })).toEqual({ triggered: true });
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Now audit your compaction restore"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Now audit your compaction restore"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
 
       expect(send).toHaveBeenCalledTimes(5);
       expect(enforcer.getManualCompactionState(SEAT)?.stage).toBe("audit-sent");
@@ -1273,14 +1273,14 @@ describe("ClaudeCompactionEnforcer", () => {
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20, transcriptPath: "/tmp/claude.jsonl",
       })).toEqual({ triggered: true });
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("OpenRig post-compaction turn boundary"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("OpenRig post-compaction turn boundary"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20, transcriptPath: "/tmp/claude.jsonl",
       })).toEqual({ triggered: true });
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20,
       })).toEqual({ triggered: true });
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Now audit your compaction restore"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Now audit your compaction restore"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
     });
 
     it("GHOST-STAGE (a) PM-pin: an AUTOMATION-initiated manual trigger is NOT drain-exempt — a disabled tick drains nothing (no laundering bypass)", async () => {
@@ -1333,7 +1333,7 @@ describe("ClaudeCompactionEnforcer", () => {
         return { ok: true, activity: { state: "idle" } as any, attempts: 1, waitedMs: 0 };
       };
       await enforcer.triggerManualCompact({ sessionName: SEAT, runtime: "claude-code", usedPercentage: 20 });
-      expect(callOrder).toEqual(["prep", "idle", `compact:${JSON.stringify({ openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: SEND_READINESS_FRESHNESS_MS + CAPTURE_READABLE_BUDGET_MS, readinessFromPaneOnly: true })}`]);
+      expect(callOrder).toEqual(["prep", "idle", `compact:${JSON.stringify({ openrigNotice: true, preserveInput: true, waitForIdleMs: SEND_READINESS_FRESHNESS_MS + CAPTURE_READABLE_BUDGET_MS, readinessFromPaneOnly: true })}`]);
     });
 
     it.each([0, 75])("wait-for-idle failure means /compact never landed: the back-half is NOT seeded (preparation delay=%ims)", async (delayMs) => {
@@ -1424,14 +1424,14 @@ describe("ClaudeCompactionEnforcer", () => {
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 95,
       })).toEqual({ triggered: true }); // owed turn boundary, not another prep
       expect(send).toHaveBeenCalledTimes(3);
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("OpenRig post-compaction turn boundary"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true, postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("OpenRig post-compaction turn boundary"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), readinessFromPaneOnly: true, postCompactTurnBoundary: true, beforeWrite: expect.any(Function) });
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20, transcriptPath: "/tmp/claude.jsonl",
       })).toEqual({ triggered: true }); // restore
       expect(await enforcer.maybeAutoCompact({
         sessionName: SEAT, runtime: "claude-code", usedPercentage: 20,
       })).toEqual({ triggered: true }); // audit
-      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Now audit your compaction restore"), { openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
+      expect(send).toHaveBeenLastCalledWith(SEAT, expect.stringContaining("Now audit your compaction restore"), { openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) });
       expect(enforcer.getManualCompactionState(SEAT)?.stage).toBe("audit-sent");
       expect(send).toHaveBeenCalledTimes(5);
     });
@@ -1500,7 +1500,7 @@ describe("ClaudeCompactionEnforcer", () => {
       expect(enforcer.getManualCompactionState(SEAT)?.stage).toBe("compact-sent"); // busy timeout does NOT advance to restore-sent
       const restoreCall = send.mock.calls.find(isRestore);
       expect(restoreCall, "restore send must be attempted with an idle gate").toBeDefined();
-      expect(restoreCall![2]).toEqual({ openrigNotice: true, allowDraftWhileIdle: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) }); // idle-gated, not a busy-pane fire-and-forget
+      expect(restoreCall![2]).toEqual({ openrigNotice: true, preserveInput: true, waitForIdleMs: expect.any(Number), beforeWrite: expect.any(Function) }); // idle-gated, not a busy-pane fire-and-forget
       expect(send.mock.calls.some(isAudit)).toBe(false); // audit cannot overtake restore
 
       // Claude goes idle → the SAME restore stage delivers and advances ONCE.

packages/daemon/test/compaction-draft-aware.test.ts

@@ -27,7 +27,7 @@ function frame(body: string): ComposerSnapshot {
     cursor: { x: 2 + stringWidth(body), y: 2, width: 100000, height: 60 }, inMode: false };
 }
 
-async function fixture(opts: { pane: string | null; mode?: "off" | "draft-aware"; footer?: boolean; refuseDraft?: boolean; suggestion?: "faint" | "plain" } = { pane: "%1" }) {
+async function fixture(opts: { pane: string | null; mode?: "off" | "draft-aware"; footer?: boolean; suggestion?: "faint" | "plain" } = { pane: "%1" }) {
   const db = new Database(":memory:");
   migrate(db, ALL_MIGRATIONS);
   db.exec(`INSERT INTO rigs(id,name) VALUES ('rig','test');
@@ -59,7 +59,6 @@ async function fixture(opts: { pane: string | null; mode?: "off" | "draft-aware"
   vi.spyOn(tmux, "capturePaneContent").mockImplementation(async () => snapshot().screen);
   vi.spyOn(tmux, "getPaneCursorPosition").mockImplementation(async () => snapshot().cursor);
   const transport = new SessionTransport({ db, tmuxAdapter: tmux, rigRepo: new RigRepository(db), sessionRegistry: new SessionRegistry(db),
-    deliverOverDraft: () => !opts.refuseDraft,
     sleep: (ms) => new Promise((r) => setTimeout(r, Math.min(ms, 5))), waitForIdlePollMs: 5 });
   // Readiness is the variable under test: the pane says idle or working.
   let activity = "idle";
@@ -90,18 +89,18 @@ async function fixture(opts: { pane: string | null; mode?: "off" | "draft-aware"
 it.each([
   { footer: false, skipMap: false }, { footer: false, skipMap: true },
   { footer: true, skipMap: false }, { footer: true, skipMap: true },
-])("default managed preparation joins a draft with typing guard off (footer=$footer, skipMap=$skipMap)", async ({ footer, skipMap }) => {
+])("managed preparation preserves a draft with typing guard off (footer=$footer, skipMap=$skipMap)", async ({ footer, skipMap }) => {
   const f = await fixture({ pane: "%1", mode: "off", footer });
   f.draft("an unsent message");
   const outcome = await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap });
   f.e.cancelPreparation(seat);
-  expect(outcome).toMatchObject(skipMap ? { triggered: true } : { preparation: { delivery: "delivered", compact: "not_sent" } });
-  expect(f.body()).toBe("");
-  expect(f.effects()).toEqual({ compactPastes: skipMap ? 1 : 0, prepPastes: 1, enters: skipMap ? 2 : 1 });
+  expect(outcome).toMatchObject({ triggered: false, reason: "draft_input_busy", preparation: { delivery: "not_sent", compact: "not_sent" } });
+  expect(f.body()).toBe("an unsent message");
+  expect(f.effects()).toEqual({ compactPastes: 0, prepPastes: 0, enters: 0 });
 });
 
 it("checks the composer after buffer setup, immediately before the preparation paste", async () => {
-  const f = await fixture({ pane: "%1", mode: "draft-aware", footer: true });
+  const f = await fixture({ pane: "%1", mode: "off", footer: true });
   f.onStage(() => f.draft("typed while the buffer was prepared"));
   const outcome = await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true });
   expect(outcome).toMatchObject({ reason: "draft_input_busy", preparation: { delivery: "not_sent" } });
@@ -109,33 +108,35 @@ it("checks the composer after buffer setup, immediately before the preparation p
   expect(f.writes).toEqual([]);
 });
 
-it("an explicit draft refusal records its existing reading without draft text", async () => {
-  const f = await fixture({ pane: "%1", mode: "off", refuseDraft: true });
-  f.draft("an unsent message");
-  await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true });
+it.each([null, { x: 0, y: 0, width: 100000, height: 60 }])("an unreadable composer holds managed preparation without input (cursor=%j)", async cursor => {
+  const f = await fixture({ pane: "%1", mode: "off" });
+  vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(cursor);
+  const outcome = await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true });
+  expect(outcome).toMatchObject({ reason: "draft_input_unknown", preparation: { delivery: "not_sent" } });
   const state = f.e.getPreparationState(seat)!;
-  expect(state).toMatchObject({ delivery: "not_sent", inputBox: { status: "typed", confidence: "medium",
-    cursor: { x: 19, y: 2, width: 100000, height: 60 } } });
+  expect(state).toMatchObject({ inputBox: { status: "unknown", confidence: "unknown", cursor,
+    evidence: expect.arrayContaining([cursor ? "cursor outside the input box" : "cursor unavailable"]) } });
   expect(Object.keys(state.inputBox!).sort()).toEqual(["confidence", "cursor", "evidence", "status"]);
-  expect(JSON.parse(readFileSync(join(dirname(state.mapPath), COMPACTION_RECORD_FILE), "utf8"))).toMatchObject({ delivery: "not_sent", inputBox: state.inputBox });
+  const saved = JSON.parse(readFileSync(join(dirname(state.mapPath), COMPACTION_RECORD_FILE), "utf8"));
+  expect(saved).toMatchObject({ delivery: "not_sent", inputBox: state.inputBox });
   expect(f.writes).toEqual([]);
 });
 
-it("an opted-in draft refusal also protects the managed command", async () => {
-  const f = await fixture({ pane: "%1", mode: "off", refuseDraft: true });
+it("a draft arriving after preparation prevents the managed command paste", async () => {
+  const f = await fixture({ pane: "%1", mode: "off" });
   const send = f.transport.send.bind(f.transport);
   vi.spyOn(f.transport, "send").mockImplementation(async (s, text, opts) => {
     if (text.startsWith("/compact")) f.draft("draft before command");
     return send(s, text, opts);
   });
   const outcome = await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true });
-  expect(outcome).toMatchObject({ preparation: { compact: "not_sent" } });
+  expect(outcome).toMatchObject({ reason: "draft_input_busy", preparation: { compact: "not_sent" } });
   expect(f.body()).toBe("draft before command");
   expect(f.effects()).toEqual({ compactPastes: 0, prepPastes: 1, enters: 1 });
 });
 
-it("an opted-in draft refusal protects owed restore input", async () => {
-  const f = await fixture({ pane: "%1", mode: "off", refuseDraft: true });
+it("an owed restore stage holds its place while the composer has a draft", async () => {
+  const f = await fixture({ pane: "%1", mode: "off" });
   expect(await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true })).toMatchObject({ triggered: true });
   f.draft("draft before restore");
   const before = f.writes.length;
@@ -157,10 +158,10 @@ it("a high-confidence faint suggestion still permits managed input", async () =>
   expect(f.effects()).toEqual({ compactPastes: 1, prepPastes: 1, enters: 2 });
 });
 
-it("default delivery does not hold possible suggestion words without faint styling", async () => {
+it("possible suggestion words without faint styling do not authorize managed input", async () => {
   const f = await fixture({ pane: "%1", mode: "off", suggestion: "plain" });
-  expect(await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true })).toMatchObject({ triggered: true });
-  expect(f.effects()).toEqual({ compactPastes: 1, prepPastes: 1, enters: 2 });
+  expect(await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true })).toMatchObject({ reason: "draft_input_unknown" });
+  expect(f.writes).toEqual([]);
 });
 
 it("automatic, recorded pane (existing-behaviour control, green before this PR's fixes): busy then idle, the lifecycle lease refuses before any send, so no /compact and no held row", async () => {
@@ -246,20 +247,84 @@ it("a draft-aware seat with a recorded pane is not affected by the pane check",
 });
 
 // #2575: a seat whose empty input box OpenRig can't read used to sit unrestored, retried on every poll. After two idle
-describe("managed restore uses ordinary draft defaults", () => {
+// polls that can't read the box, a post-compaction step goes in and is recorded as over an unreadable box; a typed draft
+// still holds it, and so does the stricter check after a /compact whose result was lost.
+describe("#2575: a post-compaction step OpenRig can't read the input box for", () => {
+  const restorePastes = (f: Awaited<ReturnType<typeof fixture>>) =>
+    f.writes.filter((w) => w.startsWith("paste:") && w.includes("Please respond to this normal user message now")).length;
   async function owedRestore(f: Awaited<ReturnType<typeof fixture>>) {
     expect(await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true })).toMatchObject({ triggered: true });
-    expect(await f.e.maybeAutoCompact(input)).toEqual({ triggered: true });
+    expect(await f.e.maybeAutoCompact(input)).toEqual({ triggered: true }); // the turn boundary, into an empty box
     expect(f.e.getOwedRestore(seat)?.stage).toBe("restore_prompt");
   }
-  it.each(["typed", "unreadable"] as const)("delivers the owed restore and audit over %s input without idle-poll counting", async kind => {
-    const f = await fixture({ pane: "%1", mode: "off" }); await owedRestore(f);
-    if (kind === "typed") f.draft("an unsent message");
-    else vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(null);
-    expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: true });
-    expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: true });
-    expect(f.e.getOwedRestore(seat)).toBeNull();
+
+  it("goes in after two idle polls that can't read the box, recorded as over an unreadable box, never as empty", async () => {
+    const f = await fixture({ pane: "%1", mode: "off" });
+    await owedRestore(f);
+    const cursor = vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(null); // an empty box that can't be confirmed live
+    const held: Array<ReturnType<typeof f.e.getOwedRestore>> = [];
+    for (let i = 1; i <= 2; i++) {
+      expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: false });
+      held.push(f.e.getOwedRestore(seat));
+    }
+    expect(restorePastes(f)).toBe(0);
+    expect(await f.e.maybeAutoCompact(input)).toEqual({ triggered: true });
+    expect(restorePastes(f)).toBe(1);
+    // What the two held polls showed: each counted, and the state said what would happen next.
+    held.forEach((owed, i) => expect(owed?.lastDelivery).toMatchObject({ stage: "restore_prompt", ok: false, reason: "draft_input_unknown", unreadable: i + 1 }));
+    expect(held[1]?.detail).toContain("After 2 idle polls that can't read the input box, the step goes in anyway; a typed draft still holds it.");
+    const owed = f.e.getOwedRestore(seat)!;
+    expect(owed.stage).toBe("compliance_prompt");
+    expect(owed.lastDelivery).toEqual({ stage: "restore_prompt", at: expect.any(String), ok: true,
+      reason: "delivered_over_unreadable_input_box", inputBox: "unknown", failures: 0 });
+    expect(owed.detail).toContain("The restore request went in at");
+    expect(owed.detail).toContain("over an input box OpenRig couldn't read (it read unknown); a draft there, if any, may hold its text.");
+    expect(cursor).toHaveBeenCalled();
   });
+
+  it("a typed draft still holds it after two unreadable polls, and the run starts again", async () => {
+    const f = await fixture({ pane: "%1", mode: "off" });
+    await owedRestore(f);
+    const cursor = vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(null);
+    for (let i = 0; i < 2; i++) await f.e.maybeAutoCompact(input);
+    f.draft("half a sentence");
+    cursor.mockResolvedValue({ x: 2 + "half a sentence".length, y: 2, width: 100000, height: 60 }); // typed, cursor after it
+    for (let i = 0; i < 3; i++) expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: false });
+    expect(f.body()).toBe("half a sentence");
+    expect(restorePastes(f)).toBe(0);
+    expect(f.e.getOwedRestore(seat)).toMatchObject({ stage: "restore_prompt", lastDelivery: { ok: false, reason: "draft_input_busy" } });
+    expect(f.e.getOwedRestore(seat)?.lastDelivery?.unreadable).toBeUndefined();
+  });
+
+  it("typed text with Claude's suggestion after it still holds it after two unreadable polls", async () => {
+    const f = await fixture({ pane: "%1", mode: "off" });
+    await owedRestore(f);
+    const cursor = vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(null);
+    for (let i = 0; i < 2; i++) await f.e.maybeAutoCompact(input);
+    f.draft("fix\x1b[2m the typecheck errors\x1b[22m");
+    cursor.mockResolvedValue({ x: 2 + "fix".length, y: 2, width: 100000, height: 60 });
+    for (let i = 0; i < 3; i++) expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: false });
+    expect(restorePastes(f)).toBe(0);
+    expect(f.body()).toBe("fix\x1b[2m the typecheck errors\x1b[22m");
+    expect(f.e.getOwedRestore(seat)?.stage).toBe("restore_prompt");
+  });
+
+  it("held drafts with Claude's suggestion don't count: one unreadable poll after them still holds the step", async () => {
+    const f = await fixture({ pane: "%1", mode: "off" });
+    await owedRestore(f);
+    f.draft("fix\x1b[2m the typecheck errors\x1b[22m");
+    const cursor = vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue({ x: 2 + "fix".length, y: 2, width: 100000, height: 60 });
+    for (let i = 0; i < 2; i++) await f.e.maybeAutoCompact(input);
+    const mixed = f.e.getOwedRestore(seat)?.lastDelivery;
+    f.draft("");
+    cursor.mockResolvedValue(null);
+    expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: false });
+    expect(restorePastes(f)).toBe(0);
+    expect(mixed).toMatchObject({ reason: "draft_input_unknown", inputBox: "typed-with-suggestion" });
+    expect(mixed?.unreadable).toBeUndefined();
+    expect(f.e.getOwedRestore(seat)?.lastDelivery).toMatchObject({ reason: "draft_input_unknown", inputBox: "unknown", unreadable: 1 });
+  });
+
   it("a step held as retained reads as still owed, never as gone in", async () => {
     const f = await fixture({ pane: "%1", mode: "off" });
     await owedRestore(f);
@@ -274,43 +339,18 @@ describe("managed restore uses ordinary draft defaults", () => {
     expect(owed.detail).not.toContain("goes in anyway");
   });
 
-  it("does not replay an uncertain compact, while owed restore follows default delivery", async () => {
-    const f = await fixture({ pane: "%1", mode: "off" }); const real = f.transport.send.bind(f.transport);
-    let commands = 0;
-    vi.spyOn(f.transport, "send").mockImplementation(async (s: string, text: string, opts?: any) => {
-      if (text.startsWith("/compact")) { commands++; return { ok: false, sessionName: s, sent: false, inputPhase: "written", reason: "submit_failed" } as any; }
-      return real(s, text, opts);
-    });
+  it("after a /compact whose result was lost, an unreadable box still holds every step, however many polls", async () => {
+    const f = await fixture({ pane: "%1", mode: "off" });
+    const real = f.transport.send.bind(f.transport);
+    vi.spyOn(f.transport, "send").mockImplementation(async (s: string, text: string, opts?: any) => text.startsWith("/compact")
+      ? { ok: false, sessionName: s, sent: false, inputPhase: "written", reason: "submit_failed" } as any : real(s, text, opts));
     await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true });
     expect(f.e.getPreparationState(seat)).toMatchObject({ compact: "uncertain" });
+    expect(f.e.getOwedRestore(seat)?.stage).toBe("turn_boundary");
     vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(null);
-    for (let i = 0; i < 3; i++) expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: true });
-    expect(commands).toBe(1); expect(f.e.getOwedRestore(seat)).toBeNull();
+    const before = f.writes.length;
+    for (let i = 0; i < 5; i++) expect(await f.e.maybeAutoCompact(input)).toMatchObject({ triggered: false });
+    expect(f.writes.length).toBe(before);
+    expect(f.e.getOwedRestore(seat)).toMatchObject({ stage: "turn_boundary", held: true });
   });
 });
-
-// Managed input follows the ordinary product default, not an internal composer policy.
-it.each(["typed", "unreadable"] as const)("default managed preparation delivers over %s input", async kind => {
-  const f = await fixture({ pane: "%1", mode: "off" });
-  if (kind === "typed") f.draft("an unsent message");
-  else vi.spyOn(f.tmux, "getPaneCursorPosition").mockResolvedValue(null);
-  await f.e.maybeAutoCompact(input);
-  expect(f.e.getPreparationState(seat)).toMatchObject({ delivery: "delivered" });
-  expect(f.effects()).toMatchObject({ prepPastes: 1, compactPastes: 0, enters: 1 });
-});
-
-it("explicit transport draft refusal still protects managed preparation", async () => {
-  const f = await fixture({ pane: "%1", mode: "off", refuseDraft: true });
-  f.draft("an unsent message");
-  await f.e.maybeAutoCompact(input);
-  expect(f.e.getPreparationState(seat)).toMatchObject({ delivery: "not_sent" });
-  expect(f.writes).toEqual([]);
-  expect(f.body()).toBe("an unsent message");
-});
-
-it("the managed idle wait treats only a recognized draft according to the ordinary default", async () => {
-  const f = await fixture({ pane: "%1", mode: "off" }); f.draft("an unsent message");
-  (f.transport as any).classifySendReadiness = async () => ({ state: "needs_input", reason: "prompt_draft", evidenceSource: "pane_heuristic" });
-  expect(await f.e.triggerManualCompact(input, { operatorInitiated: true, skipMap: true })).toMatchObject({ triggered: true });
-  expect(f.effects()).toMatchObject({ prepPastes: 1, compactPastes: 1, enters: 2 });
-});

packages/daemon/test/compaction-preparation-notice.test.ts

@@ -1,70 +0,0 @@
-import { afterEach, expect, it, vi } from "vitest";
-import { mkdtempSync, readFileSync, rmSync } from "node:fs";
-import { tmpdir } from "node:os";
-import { dirname, join } from "node:path";
-import { ClaudeCompactionEnforcer, COMPACTION_RECORD_FILE, type PreparationFailure } from "../src/domain/claude-compaction-enforcer.js";
-import type { SessionTransport } from "../src/domain/session-transport.js";
-const homes: string[] = [];
-afterEach(() => { for (const h of homes.splice(0)) rmSync(h, { recursive: true, force: true }); });
-const input = { sessionName: "writer@test", runtime: "claude-code", usedPercentage: 90 };
-function fixture() {
-  let clock = 10000, generation = "g1";
-  const home = mkdtempSync(join(tmpdir(), "prep-notice-")); homes.push(home);
-  const send = vi.fn(async (): Promise<any> => { clock++; return { ok: false, sent: false, inputPhase: "not_reached", reason: "draft_input_busy" }; });
-  const notice = vi.fn(async (_failure: PreparationFailure) => {});
-  const policy = { enabled: true, thresholdPercent: 80, preCompactInstruction: "Write a map.", compactInstruction: "" };
-  const opts = { openrigHome: home, now: () => clock, resolveOccupantGeneration: () => generation,
-    autoRetryAfterStopMs: 1, onPreparationFailure: notice };
-  const transport = { send, waitUntilIdle: vi.fn(async () => ({ ok: true })) } as unknown as SessionTransport;
-  let e = new ClaudeCompactionEnforcer({ resolveClaudeCompactionPolicy: () => policy } as any, transport, opts);
-  return { get e() { return e; }, send, notice, advance: (n = 2) => { clock += n; },
-    generation: (s: string) => { generation = s; },
-    restart: () => { e = new ClaudeCompactionEnforcer({ resolveClaudeCompactionPolicy: () => policy } as any, transport, { ...opts, daemonInstance: "restarted" }); } };
-}
-async function end(f: ReturnType<typeof fixture>) {
-  for (let i = 0; i < 4; i++) {
-    await f.e.maybeAutoCompact(input);
-    if (f.e.getPreparationState(input.sessionName)?.status === "stopped") {
-      const id = f.e.getPreparationState(input.sessionName)!.attemptId; f.advance(); return id;
-    }
-  }
-  throw new Error("attempt did not end");
-}
-
-it("counts three ended automatic attempts across ids/restart, not their nine send calls", async () => {
-  const f = fixture(); const ids = [await end(f)];
-  expect(f.send).toHaveBeenCalledTimes(3); expect(f.notice).not.toHaveBeenCalled();
-  ids.push(await end(f)); expect(f.notice).not.toHaveBeenCalled();
-  f.restart(); ids.push(await end(f)); expect(new Set(ids).size).toBe(3);
-  await f.e.maybeAutoCompact(input);
-  expect(f.notice).toHaveBeenCalledTimes(1);
-  expect(f.notice).toHaveBeenCalledWith(expect.objectContaining({ occupantGeneration: "g1", attemptId: ids[2], endedAttempts: 3, reason: "draft_input_busy" }));
-  const evidence = JSON.parse(readFileSync(f.notice.mock.calls[0]![0].statePath, "utf8"));
-  expect(evidence).toMatchObject({ phase: "ended", delivery: "not_sent", attemptId: ids[2], preparationFailures: { endedAttempts: 3 } });
-  await f.e.maybeAutoCompact(input); expect(f.notice).toHaveBeenCalledTimes(1);
-});
-
-it.each(["below", "cancel", "generation", "uncertain", "progress"])("%s interrupts the failure run", async kind => {
-  const f = fixture(); await end(f); await end(f);
-  if (kind === "below") await f.e.maybeAutoCompact({ ...input, usedPercentage: 20 });
-  if (kind === "cancel") { f.e.cancelPreparation(input.sessionName); await f.e.maybeAutoCompact(input); expect(f.notice).not.toHaveBeenCalled(); return; }
-  if (kind === "generation") f.generation("g2");
-  if (kind === "uncertain" || kind === "progress") {
-    f.send.mockResolvedValue(kind === "uncertain" ? { ok: false, sent: true, inputPhase: "attempted", reason: "send_failed" } : { ok: true });
-    await f.e.maybeAutoCompact(input); await f.e.maybeAutoCompact(input);
-    const count = f.send.mock.calls.length;
-    await f.e.maybeAutoCompact(input); expect(f.send).toHaveBeenCalledTimes(count);
-    expect(f.notice).not.toHaveBeenCalled(); return;
-  }
-  await end(f); await f.e.maybeAutoCompact(input); expect(f.notice).not.toHaveBeenCalled();
-});
-
-it("a notice callback cannot restore a run that cancellation cleared", async () => {
-  const f = fixture(); await end(f); await end(f); await end(f);
-  const state = f.e.getPreparationState(input.sessionName)!;
-  f.notice.mockImplementationOnce(async () => { f.e.cancelPreparation(input.sessionName); });
-  await f.e.maybeAutoCompact(input);
-  const saved = JSON.parse(readFileSync(join(dirname(state.mapPath), COMPACTION_RECORD_FILE), "utf8"));
-  expect(saved.preparationFailures).toBeUndefined();
-  await f.e.maybeAutoCompact(input); expect(f.notice).toHaveBeenCalledTimes(1);
-});

packages/daemon/test/compaction-preparation.test.ts

@@ -98,8 +98,8 @@ for(const budget of [40_000,9_000])it(`manual skip-map final send keeps its orig
  expect(f.clock()-start).toBeLessThanOrEqual(budget);
  expect(f.e.getPreparationState(seat)?.deadlineAt).toBe(start+budget);
 });
-for(const pane of ['running','permission','unknown'] as const)it(`an aged preparation hook still cannot compact a ${pane} pane`,async()=>{
- const f=await tinyPane(pane==='unknown'?'unknown':'empty');
+for(const pane of ['running','permission','typed','unknown'] as const)it(`an aged preparation hook still cannot compact a ${pane} pane`,async()=>{
+ const f=await tinyPane(pane==='typed'?'typed':pane==='unknown'?'unknown':'empty');
  if(pane==='running'||pane==='permission'){
   const screen=pane==='running'
    ?readFileSync(join(import.meta.dirname,'fixtures/input-box/claude-code-2.1.296/after-enter-0.4s.J.txt'),'utf8')
@@ -124,14 +124,14 @@ it.each(['live-pane','final-composer'])('submits managed /compact from an 11x20
  await f.e.maybeAutoCompact(input);
  expect(f.window).toEqual({width:11,height:20,own:null,recorded:''});
  expect(compacts(f)).toHaveLength(1);
- expect(f.tmux.sendText).toHaveBeenCalledWith(seat,expect.stringMatching(/^\/compact/),expect.any(Function));
+ expect(f.tmux.sendText).toHaveBeenCalledWith(seat,expect.stringMatching(/^\/compact/),expect.any(Function),expect.objectContaining({beforePaste:expect.any(Function)}));
 });
-it.each(['unknown','typed'] as const)('a readable %s pane follows ordinary draft defaults and gives its window back',async kind=>{
+it.each(['unknown','typed'] as const)('keeps a readable %s composer from authorizing /compact and gives its window back',async kind=>{
  const f=await tinyPane(kind);
  const box=(await f.transport.capture(seat,{inputBox:true,readable:{width:120,height:40}})).inputBox;
  expect(box?.status).toBe(kind==='typed'?'typed':'no-input-box');
  await f.e.maybeAutoCompact(input);
- expect(compacts(f)).toHaveLength(kind==='typed'?1:0);
+ expect(compacts(f)).toHaveLength(0);
  expect(f.window).toEqual({width:11,height:20,own:null,recorded:''});
 });
 it('submits managed /compact when readable redraw settling exceeds the ordinary one-second idle wait',async()=>{

packages/daemon/test/compaction-restore-notice.test.ts

@@ -8,8 +8,8 @@ import { RigRepository } from "../src/domain/rig-repository.js";
 import { SessionRegistry } from "../src/domain/session-registry.js";
 import { EventBus } from "../src/domain/event-bus.js";
 import { QueueRepository } from "../src/domain/queue-repository.js";
-import { reportPreparationFailure, reportRestoreDeliveryFailure } from "../src/domain/compaction-restore-notice.js";
-import type { PreparationFailure, RestoreDeliveryFailure } from "../src/domain/claude-compaction-enforcer.js";
+import { reportRestoreDeliveryFailure } from "../src/domain/compaction-restore-notice.js";
+import type { RestoreDeliveryFailure } from "../src/domain/claude-compaction-enforcer.js";
 
 it("notifies the seat's owner once, updates a changed refusal without another wake, and honors the owner's closure", async () => {
   const db = createDb();
@@ -34,8 +34,7 @@ it("notifies the seat's owner once, updates a changed refusal without another wa
     expect(queue.getById(id)?.destinationSession).toBe("owner@restore-notice");
     // #2575: the owner is told how to take the restore over with existing verbs, and that it isn't race-free.
     const body = queue.getById(id)?.body ?? "";
-    expect(body).toContain("Restore input follows ordinary delivery defaults");
-    expect(body).toContain("transport.draft_delivery=refuse");
+    expect(body).toContain("OpenRig sends the step itself after 2 idle polls, recorded as over an unreadable box. That doesn't apply after a /compact whose result was lost, where each step waits for an empty box, or on a seat whose draft-aware delivery or typing guard holds OpenRig's input: those keep holding, draft-aware delivery until the seat's mode is set to off and a typing guard until its level admits OpenRig's input.");
     expect(body).toContain("first run rig compact 'writer@restore-notice' --cancel, so OpenRig stops owing the remaining steps (a step it is already pasting at that moment can still land)");
     expect(body).toContain("naming the restore map, its restore-pending marker /openrig/compaction/restore-pending/writer.json and the packet the marker names.");
     expect(send).toHaveBeenCalledTimes(1);
@@ -71,28 +70,3 @@ it("a human fallback is a plain ask: urgent-only stays on Steering while every-a
     expect(queue.listTransitions("qitem-compaction-restore-human-every-ask")[0]).toMatchObject({ ownerNotificationKind: "human-required", ownerNotificationLevel: "ALERT" });
   } finally { selected.mockRestore(); db.close(); }
 });
-
-it("preparation loop finds its owner once across attempts and honors the owner's disposition", async () => {
-  const db = createDb(); migrate(db, ALL_MIGRATIONS);
-  try {
-    const rigs = new RigRepository(db), sessions = new SessionRegistry(db), rig = rigs.createRig("prep-notice");
-    const owner = rigs.addNode(rig.id, "owner", { runtime: "claude-code" });
-    const seat = rigs.addNode(rig.id, "writer", { runtime: "claude-code" });
-    sessions.registerSession(owner.id, "owner@prep-notice"); sessions.updateBinding(owner.id, { tmuxSession: "owner@prep-notice" });
-    sessions.registerSession(seat.id, "writer@prep-notice"); rigs.addEdge(rig.id, owner.id, seat.id, "delegates_to");
-    const send = vi.fn(async () => ({ ok: true, verified: true }));
-    const queue = new QueueRepository(db, new EventBus(db), { validateRig: () => true, transport: { send } });
-    const f: PreparationFailure = { sessionName: "writer@prep-notice", occupantGeneration: "g1", attemptId: "three", endedAttempts: 3,
-      mapPath: "/maps/three/RESTORE-MAP.md", statePath: "/maps/three/COMPACTION-STATE.json", reason: "typing_guard_enabled" };
-    await reportPreparationFailure(db, queue, f);
-    const rows = db.prepare("SELECT qitem_id AS id FROM queue_items WHERE tags LIKE '%preparation-delivery%'").all() as { id: string }[];
-    expect(rows).toHaveLength(1); const id = rows[0]!.id;
-    expect(queue.getById(id)).toMatchObject({ destinationSession: "owner@prep-notice", evidenceRef: "/maps/three/COMPACTION-STATE.json", summary: expect.stringContaining("3 ended unsent attempts"), body: expect.stringContaining("reading unavailable") });
-    await reportPreparationFailure(db, queue, { ...f, attemptId: "four", endedAttempts: 4, inputBox: { status: "typed", confidence: "high", cursor: null, evidence: ["cursor unavailable"] } });
-    expect(queue.listTransitions(id).at(-1)?.transitionNote).toContain('"status":"typed"');
-    expect(send).toHaveBeenCalledTimes(1);
-    await queue.update({ qitemId: id, actorSession: "owner@prep-notice", state: "done", closureReason: "no-follow-on" });
-    await reportPreparationFailure(db, new QueueRepository(db, new EventBus(db), { validateRig: () => true, transport: { send } }), { ...f, attemptId: "five", endedAttempts: 5 });
-    expect(queue.getById(id)?.state).toBe("done"); expect(send).toHaveBeenCalledTimes(1);
-  } finally { db.close(); }
-});

packages/daemon/test/helpers/tmux-tile-size.py

@@ -238,14 +238,18 @@ def writable_first_input(server):
     server.wait_for(lambda: server.client_ttys() == [], "an idle tile detaches", 8)
     # A key and, at once, a paste: the paste is still framed, and both arrive in order.
     server.wait_for(lambda: server.seen(tile).rfind("\x1b[?2004h") > server.seen(tile).rfind("[detached"), "the tile shows again")
+    mark = len(server.seen(tile))
     os.write(tile[1], b"A")
     time.sleep(0.03)
     framed = server.seen(tile).rfind("\x1b[?2004h") > server.seen(tile).rfind("\x1b[?2004l")
     os.write(tile[1], b"\x1b[200~" + paste + b"\x1b[201~" if framed else paste)
     expected += b"A\x1b[200~" + paste + b"\x1b[201~"
-    # A rare CI failure here is unexplained: say what the seat got.
     server.wait_for(lambda: received(server) == expected, "a key then a paste arrive in order, the paste framed",
                     detail=lambda: "framed=%r; seat got %r; expected %r" % (framed, received(server)[-80:], expected[-80:]))
+    # From the first key until tmux takes the terminal, its paste framing stays on: a paste in that gap stays one paste.
+    server.wait_for(lambda: "\x1b[?2004h" in server.seen(tile)[mark:], "tmux takes the tile's terminal")
+    gap = server.seen(tile)[mark:].split("\x1b[?2004h", 1)[0]
+    assert "\x1b[?2004l" not in gap, "the tile turned paste framing off before tmux attached: " + repr(gap[-80:])
     server.wait_for(lambda: server.client_ttys() == [], "an idle tile detaches", 8)
     # A run of text that ends in ; keeps it, through the client and before a paste alike.
     os.write(tile[1], b"ls;\r")

packages/daemon/test/slack-settled-ask.test.ts

@@ -377,6 +377,113 @@ describe("posted asks follow settlement outside Slack", () => {
     expect(postCount).toBe(7);
   });
 
+  it("remembers a deleted original across restarts: asked once, not again after a restart", async () => {
+    const item = await postAsk();
+    apiHook = (method) => method === "conversations.history" ? response({ ok: true, messages: [] }) : undefined;
+    repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
+    await vi.waitFor(() => expect(logs.join("\n")).toContain("original message unavailable"));
+    await new Promise((r) => setTimeout(r, 25));
+    const reads = () => methods.filter((m) => m === "conversations.history").length;
+    expect(reads()).toBe(1);
+    expect(repo.listTransitions(item.qitemId).filter((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable "))).toHaveLength(1);
+    expect(repo.getById(item.qitemId)?.state).toBe("done");
+
+    wire.stop(); start();
+    await new Promise((r) => setTimeout(r, 50));
+    expect(reads()).toBe(1);
+    expect(methods).not.toContain("chat.update");
+    expect(repo.listTransitions(item.qitemId).filter((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable "))).toHaveLength(1);
+  });
+
+  it("still edits a later post of the same ask after its first post was recorded as gone", async () => {
+    const item = await repo.create({ sourceSession: "author@rig", destinationSession: "worker@rig",
+      summary: "Choose the next step", body: "Original work", nudge: false });
+    const park = () => repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "blocked", blockedOn: human,
+      summary: "May I continue?", transitionNote: "Need the person's answer." });
+    const send = async (ts: string) => {
+      const [alert] = await makeQueuePorts(repo, { loadHumanRegistry: () => registry }).listHumanAlerts({});
+      expect(wire.dispatcher.dispatch("post_message", human, alert)).toMatchObject({ ok: true });
+      await vi.waitFor(() => expect(repo.currentHumanRequiredEpisode(item.qitemId)?.receipt?.messageTs).toBe(ts));
+    };
+    const reads = (ts: string) => requests.filter((r) => new URL(r.url).searchParams.get("latest") === ts
+      || new URL(r.url).searchParams.get("oldest") === ts).length;
+    // The first post was deleted in Slack: reading it finds nothing.
+    apiHook = (method, params) => method === "conversations.history" && params.latest === "1.000001" ? response({ ok: true, messages: [] }) : undefined;
+    park(); await send("1.000001");
+    repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "in-progress", transitionNote: "Answered in the pane; continuing." });
+    await vi.waitFor(() => expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable ")
+      && t.transitionNote.includes("message_ts=1.000001 "))).toBe(true));
+
+    park(); await send("2.000001");
+    repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "in-progress", transitionNote: "Second answer received." });
+
+    await vi.waitFor(() => expect(updates).toHaveLength(1));
+    expect(updates[0]!.ts).toBe("2.000001");
+    expect(reads("1.000001")).toBe(1);
+    wire.stop(); start();
+    await new Promise((r) => setTimeout(r, 50));
+    expect(reads("1.000001")).toBe(1);
+    expect(updates).toHaveLength(1);
+  });
+
+  it("still skips a deleted original until restart when recording it as unavailable fails", async () => {
+    const item = await postAsk();
+    apiHook = (method) => method === "conversations.history" ? response({ ok: true, messages: [] }) : undefined;
+    const update = repo.update.bind(repo);
+    const spy = vi.spyOn(repo, "update").mockImplementation((input) => {
+      if (input.transitionNote?.startsWith("slack-ask-settle-unavailable ")) throw new Error("disk full");
+      return update(input);
+    });
+    repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
+    await vi.waitFor(() => expect(logs.join("\n")).toContain("couldn't record"));
+    spy.mockRestore();
+    const reads = () => methods.filter((m) => m === "conversations.history").length;
+    expect(reads()).toBe(1);
+    expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable "))).toBe(false);
+
+    // Another queue change re-enqueues the ask; the in-memory note keeps it from being read again.
+    repo.update({ qitemId: item.qitemId, actorSession: "author@rig", transitionNote: "Follow-up note." });
+    await new Promise((r) => setTimeout(r, 50));
+    expect(reads()).toBe(1);
+
+    // Nothing durable was written, so a restart asks Slack once more and records it this time.
+    wire.stop(); start();
+    await vi.waitFor(() => expect(repo.listTransitions(item.qitemId).filter((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable "))).toHaveLength(1));
+    expect(reads()).toBe(2);
+    expect(methods).not.toContain("chat.update");
+  });
+
+  it("retries an edit refused as cant_update_message after a restart, once the token can edit it", async () => {
+    const item = await postAsk();
+    apiHook = (method) => method === "chat.update" ? response({ ok: false, error: "cant_update_message" }) : undefined;
+    repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
+    await vi.waitFor(() => expect(logs.join("\n")).toContain("cant_update_message"));
+    const edits = () => methods.filter((m) => m === "chat.update").length;
+    expect(edits()).toBe(1);
+    expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable "))).toBe(false);
+
+    // Skipped until restart: another queue change doesn't try the edit again.
+    repo.update({ qitemId: item.qitemId, actorSession: "author@rig", transitionNote: "Follow-up note." });
+    await new Promise((r) => setTimeout(r, 50));
+    expect(edits()).toBe(1);
+
+    // The token is corrected and the daemon restarts: the same message is edited.
+    wire.stop(); apiHook = undefined; start();
+    await vi.waitFor(() => expect(updates).toHaveLength(1));
+    expect(updates[0]!.ts).toBe("1.000001");
+    expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settled "))).toBe(true);
+  });
+
+  it("keeps retrying after a restart when the failure isn't about the message itself", async () => {
+    const item = await postAsk();
+    apiHook = (method) => method === "conversations.history" ? response({ ok: false, error: "not_in_channel" }) : undefined;
+    repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
+    await vi.waitFor(() => expect(logs.join("\n")).toContain("not_in_channel"));
+    expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settle-unavailable "))).toBe(false);
+    wire.stop(); apiHook = undefined; start();
+    await vi.waitFor(() => expect(updates).toHaveLength(1));
+  });
+
   it("shows a later close's new reason after the same direct ask was reopened", async () => {
     const item = await postAsk();
     wire.stop(); vi.useFakeTimers();

packages/daemon/test/slack-socket-liveness.test.ts

@@ -39,6 +39,7 @@ describe("Socket Mode liveness", () => {
   let openGate: Promise<void> | null;
   let openFails: boolean;
   let openCalls: number;
+  let gapConnects: number;
   const pingChannel: PingChannel = { subscribe: (fn) => { pings.add(fn); }, unsubscribe: (fn) => { pings.delete(fn); } };
   const ping = () => { for (const fn of pings) fn({ payload: Buffer.from("") }); };
 
@@ -50,6 +51,7 @@ describe("Socket Mode liveness", () => {
     openGate = null;
     openFails = false;
     openCalls = 0;
+    gapConnects = 0;
     const fsx = memFs();
     receipts = new InboundReceiptStore("/s/inbound-receipts.jsonl", fsx);
     const router = new InboundRouter({
@@ -74,6 +76,7 @@ describe("Socket Mode liveness", () => {
       },
       pingChannel,
       receipts,
+      recovery: { run: async () => {}, stop: () => {}, connected: () => { gapConnects += 1; } },
       log: () => {},
     });
   });
@@ -159,6 +162,19 @@ describe("Socket Mode liveness", () => {
     expect(handle!.status()).toMatchObject({ state: "connected", generation: 2 });
   });
 
+  it("tells recovery about a connect after a gap, never about a refresh that overlapped the old connection", async () => {
+    const first = await openSocket(1);
+    expect(gapConnects).toBe(1);
+    message(first, { envelope_id: "d-9", type: "disconnect", reason: "refresh_requested" });
+    const second = await openSocket(2);
+    expect(gapConnects).toBe(1);
+    first.ws.onclose!();
+    second.ws.onclose!();
+    await vi.advanceTimersByTimeAsync(5_000);
+    await openSocket(3);
+    expect(gapConnects).toBe(2);
+  });
+
   it("reports disconnected when the retired connection closes before its replacement opens", async () => {
     const first = await openSocket(1);
     for (let i = 0; i < 2; i++) { ping(); await vi.advanceTimersByTimeAsync(10_000); }

packages/daemon/test/slack-thread-recovery.test.ts

@@ -69,23 +69,31 @@ function fixture() {
     const channel = (history[u.searchParams.get("channel")!] ?? []).filter(inRange).sort(byTs).reverse();
     return Response.json({ ok: true, messages: channel.slice(0, limit), has_more: channel.length > limit });
   };
-  const recovery = (opts: { channels?: Array<{ channel: string; seed?: "adoption" }> } = {}) => {
+  const recovery = (opts: { channels?: Array<{ channel: string; seed?: "adoption" }>; relayed?: boolean } = {}) => {
     const r = new SlackRecovery({ channels: opts.channels ?? [{ channel: "C1" }], token: "fixture-token", stateDir: "/s", router,
       trackedThreads: (channels, closedSince) => map.trackedForRecovery(channels, closedSince),
-      fetchImpl, now: () => now, channelStore, threadStore });
+      fetchImpl, now: () => now, channelStore, threadStore, ...(opts.relayed ? { relayed: true } : {}) });
     r.initialize();
     closes.push(() => r.stop());
     return r;
   };
   const track = (root: string, seat = "asker@test", channel = "C1") =>
     map.open({ threadTs: root, channel, human: "human-fixture@external", seat, conversationId: `q-${seat}-${root}` });
+  // An ask: its queue row in `state`, and the thread root it posted.
+  const ask = (root: string, state: string, seat = "asker@test") => {
+    const id = `q-${seat}-${root}`, stamp = new Date(now).toISOString();
+    db.prepare(`INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, body)
+      VALUES (?, ?, ?, ?, 'human-fixture@external', ?, 'ask')`).run(id, stamp, stamp, seat, state);
+    track(root, seat);
+    return { settle: () => db.prepare(`UPDATE queue_items SET state = 'done', ts_updated = ? WHERE qitem_id = ?`).run(new Date(now).toISOString(), id) };
+  };
   const reply = (seconds: number, root = ROOT, extra: Partial<SlackEvent> = {}): SlackEvent =>
     ({ type: "message", user: "U1", channel: "C1", text: `reply ${seconds}`, ts: ts(seconds), thread_ts: root, ...extra });
   const message = (seconds: number, channel = "C1"): SlackEvent =>
     ({ type: "message", user: "U1", channel, text: `task ${seconds}`, ts: ts(seconds) });
   const calls = (method: string) => queries.filter(u => u.pathname.endsWith(method));
   const rows = () => repo.list({ limit: 500 });
-  return { db, files, seen, dead, map, router, deps, routed, resolveHumanReply, history, replies, queries, recovery, track, reply, message,
+  return { db, files, seen, dead, map, router, deps, routed, resolveHumanReply, history, replies, queries, recovery, track, ask, reply, message,
     calls, rows, time: (ms: number) => { now = ms; }, advance: (ms: number) => { now += ms; },
     respond: (fn?: typeof respond) => { respond = fn; }, down: (v: boolean) => { queueDown = v; } };
 }
@@ -144,7 +152,7 @@ it("tracks a thread whose root or close is within seven days, never by the map's
 
   await r.run();
 
-  expect(f.calls("conversations.replies").map(u => u.searchParams.get("ts"))).toEqual([root("oldClosed"), root("recentRoot")]);
+  expect(f.calls("conversations.replies").map(u => u.searchParams.get("ts"))).toEqual([root("recentRoot"), root("oldClosed")]);
   // The old root closed yesterday is scanned from its root: eligibility isn't a seven-day history cap.
   expect(f.rows().map(row => row.destinationSession).sort()).toEqual(["a@test", "c@test"]);
 });
@@ -163,16 +171,18 @@ it("keeps recovery counts for the connector's lifetime after a thread leaves the
   expect(r.status().acceptedThisProcess).toBe(1);
 });
 
-it("a thread Slack can't find stays incomplete without a call each pass until its mapping changes", async () => {
+it("a thread Slack can't find isn't asked again each pass or after a restart, only once a day", async () => {
   const f = fixture(); f.track(ROOT);
   f.respond(u => u.pathname.endsWith("conversations.replies") ? Response.json({ ok: false, error: "thread_not_found" }) : undefined);
   const r = f.recovery();
   await r.run();
   f.time(at(1000)); await r.run();
+  r.stop();
+  f.time(at(2000)); const restarted = f.recovery(); await restarted.run();
 
   expect(f.calls("conversations.replies")).toHaveLength(1);
-  expect(r.status().threads).toMatchObject({ incomplete: { thread_not_found: 1 } });
-  f.map.close(ROOT, "C1"); f.time(at(1100)); await r.run();
+  expect(restarted.status().threads).toMatchObject({ incomplete: { thread_not_found: 1 } });
+  f.time(at(100 + DAY + 1)); await restarted.run();
   expect(f.calls("conversations.replies")).toHaveLength(2);
 });
 
@@ -313,7 +323,7 @@ it.each(["missing_scope", "not_in_channel", "thread_not_found"])("%s leaves the
 
 it("channels and threads take turns under one budget: every thread is reached and the channel is visited every pass", async () => {
   const f = fixture();
-  for (let i = 1; i <= 10; i++) { f.track(ts(i), `s${i}@test`); f.replies[`C1:${ts(i)}`] = [f.reply(i + 20, ts(i))]; }
+  for (let i = 1; i <= 10; i++) { f.ask(ts(i), "pending", `s${i}@test`); f.replies[`C1:${ts(i)}`] = [f.reply(i + 20, ts(i))]; }
   f.history.C1 = [];
   const r = f.recovery();
   for (let pass = 1; pass <= 4; pass++) {
@@ -321,7 +331,8 @@ it("channels and threads take turns under one budget: every thread is reached an
     const before = f.calls("conversations.replies").length;
     await r.run();
     expect(f.calls("conversations.history")).toHaveLength(pass);
-    expect(f.calls("conversations.replies").length - before).toBe(3);
+    // Three thread calls a pass until each thread has had its scan since boot: the last pass has one left.
+    expect(f.calls("conversations.replies").length - before).toBe(pass < 4 ? 3 : 1);
     if (pass === 1) expect(r.status().threads).toMatchObject({ tracked: 10, scanned: 3, pending: 7 });
     if (pass === 1) expect(r.status().threads.oldestPendingSeconds).toBeGreaterThan(100);
   }
@@ -407,3 +418,248 @@ it.each(["thread reply", "broadcast reply in a newly covered channel"])("a recov
   expect(f.resolveHumanReply).not.toHaveBeenCalled();
   expect(f.rows()[0]!.body).toContain("It was not applied as an answer to q-decision");
 });
+
+it("a missed reply on an open ask is recovered in the first pass after boot, ahead of 100 older threads, and again after a restart", async () => {
+  const f = fixture();
+  // A week of settled asks, newer and older than the open one, none scanned since they closed.
+  for (let i = 0; i < 100; i++) {
+    const root = ts(-6 * DAY + i * 3000);
+    f.ask(root, "done", `old${i}@test`); f.replies[`C1:${root}`] = [f.reply(-6 * DAY + i * 3000 + 10, root)];
+  }
+  const open = ts(-3 * DAY - 1);
+  f.ask(open, "pending", "asker@test");
+  f.replies[`C1:${open}`] = [f.reply(50, open, { text: "Yes, ship it" })];
+  const first = f.recovery();
+
+  await first.run();
+
+  expect(f.calls("conversations.replies")[0]!.searchParams.get("ts")).toBe(open);
+  expect(f.rows().filter(row => row.destinationSession === "asker@test")).toHaveLength(1);
+  // The rest follow newest first.
+  expect(f.calls("conversations.replies").slice(1).map(u => u.searchParams.get("ts")))
+    .toEqual([ts(-6 * DAY + 99 * 3000), ts(-6 * DAY + 98 * 3000), ts(-6 * DAY + 97 * 3000)]);
+  f.time(at(400)); await first.run();
+  first.stop();
+
+  // Down for an hour; the person replies again meanwhile.
+  f.replies[`C1:${open}`].push(f.reply(1000, open, { text: "And the docs too" }));
+  f.time(at(3700));
+  const restarted = f.recovery();
+  await restarted.run();
+
+  const asker = f.rows().filter(row => row.destinationSession === "asker@test");
+  expect(asker).toHaveLength(2);
+  expect(asker.map(row => row.body).join("\n")).toContain("And the docs too");
+  expect(f.calls("conversations.replies").at(-3)!.searchParams.get("ts")).toBe(open);
+  expect(f.resolveHumanReply).not.toHaveBeenCalled();
+});
+
+it("asks still waiting on the person go first, newest first, then the rest; nothing is scanned again while connected", async () => {
+  const f = fixture();
+  for (let i = 1; i <= 5; i++) f.ask(ts(i), "pending", `open${i}@test`);
+  f.ask(ts(10), "done", "settled@test");
+  const r = f.recovery();
+  f.advance(5000); // past the settle margin, so a scan covers through boot (the boundary has its own test)
+  const order = async () => { const before = f.calls("conversations.replies").length; await r.run(); f.advance(10_000);
+    return f.calls("conversations.replies").slice(before).map(u => u.searchParams.get("ts")); };
+
+  // (The first pass has four thread calls: the channel has no gap yet.)
+  expect(await order()).toEqual([ts(5), ts(4), ts(3), ts(2)]);
+  expect(await order()).toEqual([ts(1), ts(10)]);
+  // Replies while connected arrive live, so a thread scanned since boot isn't asked again every pass.
+  expect(await order()).toEqual([]);
+});
+
+it("a waiting ask scanned since boot is scanned again within the hour, a settled thread within the day", async () => {
+  const f = fixture();
+  f.ask(ts(1), "pending", "open@test");
+  f.ask(ts(2), "done", "settled@test");
+  const r = f.recovery();
+  f.advance(5000); // past the settle margin, so a scan covers through boot (the boundary has its own test)
+  await r.run();
+  const scanned = () => f.calls("conversations.replies").map(u => u.searchParams.get("ts"));
+  expect(scanned()).toEqual([ts(1), ts(2)]);
+
+  f.time(at(100 + 59 * 60)); await r.run();
+  expect(scanned()).toHaveLength(2);
+  f.time(at(100 + 61 * 60)); await r.run();
+  expect(scanned()).toEqual([ts(1), ts(2), ts(1)]);
+  f.time(at(100 + DAY + 60)); await r.run();
+  expect(scanned().slice(3).sort()).toEqual([ts(1), ts(2)]);
+});
+
+it("a settled thread is due again after the socket reconnects from a gap, not a day later", async () => {
+  const f = fixture();
+  f.ask(ts(1), "done", "settled@test");
+  const r = f.recovery();
+  f.advance(5000); // past the settle margin, so a scan covers through boot (the boundary has its own test)
+  await r.run();
+  f.time(at(400)); await r.run();
+  expect(f.calls("conversations.replies")).toHaveLength(1);
+
+  // The socket drops for longer than Slack retries; a reply lands in the settled thread meanwhile.
+  f.replies[`C1:${ts(1)}`] = [f.reply(500, ts(1), { text: "one more thing" })];
+  f.time(at(1200)); r.connected();
+  f.advance(5000); await r.run();
+
+  expect(f.calls("conversations.replies")).toHaveLength(2);
+  expect(f.rows().filter(row => row.destinationSession === "settled@test")).toHaveLength(1);
+  f.time(at(1500)); await r.run();
+  expect(f.calls("conversations.replies")).toHaveLength(2);
+});
+
+it("a settled thread is scanned once after each boot and then daily, not every pass, and a reply after close still lands", async () => {
+  const f = fixture();
+  const done = f.ask(ts(1), "pending", "settled@test");
+  f.ask(ts(2), "pending", "open@test");
+  const r = f.recovery();
+  f.advance(5000); // past the settle margin, so a scan covers through boot (the boundary has its own test)
+  await r.run();
+  done.settle();
+  const settledCalls = () => f.calls("conversations.replies").filter(u => u.searchParams.get("ts") === ts(1)).length;
+  expect(settledCalls()).toBe(1);
+
+  f.time(at(400)); await r.run();
+  f.time(at(700)); await r.run();
+  expect(settledCalls()).toBe(1); // caught up while this process listened live: nothing to recover
+  expect(r.status().threads).toMatchObject({ tracked: 2, settled: 1, settledRescanHours: 24 });
+  r.stop();
+
+  // Off for a while; a reply arrives in the settled thread meanwhile.
+  f.replies[`C1:${ts(1)}`] = [f.reply(800, ts(1), { text: "one more thing" })];
+  f.time(at(1000));
+  const restarted = f.recovery();
+  f.advance(5000); await restarted.run();
+  expect(settledCalls()).toBe(2);
+  expect(f.rows().filter(row => row.destinationSession === "settled@test")).toHaveLength(1);
+
+  f.time(at(1300)); await restarted.run();
+  expect(settledCalls()).toBe(2);
+  f.time(at(1000 + DAY + 10)); await restarted.run();
+  expect(settledCalls()).toBe(3);
+});
+
+it("a thread about to leave the seven-day window is scanned first after boot, so its missed reply isn't lost", async () => {
+  const f = fixture();
+  for (let i = 1; i <= 5; i++) f.ask(ts(i), "pending", `open${i}@test`);
+  // Posted seven days ago less an hour: it leaves the window within the hour.
+  const expiring = ts(100 - 7 * DAY + 3600);
+  f.ask(expiring, "done", "old@test");
+  f.replies[`C1:${expiring}`] = [f.reply(50, expiring)];
+  const r = f.recovery();
+
+  await r.run();
+
+  expect(f.calls("conversations.replies")[0]!.searchParams.get("ts")).toBe(expiring);
+  expect(f.rows().filter(row => row.destinationSession === "old@test")).toHaveLength(1);
+  // The waiting asks follow, newest first.
+  expect(f.calls("conversations.replies").slice(1).map(u => u.searchParams.get("ts"))).toEqual([ts(5), ts(4), ts(3)]);
+});
+
+it("a reply in the last five seconds before a reconnect is recovered on the next pass, not left for the daily backstop", async () => {
+  const f = fixture();
+  f.ask(ts(1), "done", "settled@test");
+  const r = f.recovery();
+  f.time(at(400)); await r.run();
+  f.time(at(700)); await r.run();
+  expect(f.calls("conversations.replies")).toHaveLength(1);
+
+  // A reply three seconds before live delivery resumes: a scan at the connect stops at the settle margin, before it.
+  f.replies[`C1:${ts(1)}`] = [f.reply(1197, ts(1), { text: "just before" })];
+  f.time(at(1200)); r.connected();
+  await r.run();
+  expect(f.calls("conversations.replies").at(-1)!.searchParams.get("latest")).toBe(`${BASE + 1195}.000000`);
+  expect(f.rows().filter(row => row.destinationSession === "settled@test")).toHaveLength(0);
+
+  f.time(at(1210)); await r.run();
+  expect(f.rows().filter(row => row.destinationSession === "settled@test")).toHaveLength(1);
+  const calls = f.calls("conversations.replies").length;
+  f.time(at(1500)); await r.run();
+  expect(f.calls("conversations.replies")).toHaveLength(calls);
+});
+
+it("threads that keep failing ahead of a reachable waiting ask can't take its turn pass after pass", async () => {
+  const f = fixture();
+  // Four newer waiting asks Slack refuses every time, enough to fill a pass's thread calls, and an older one it answers.
+  const failing = [ts(11), ts(12), ts(13), ts(14)];
+  failing.forEach((root, i) => f.ask(root, "pending", `failing${i}@test`));
+  f.ask(ts(1), "pending", "reachable@test");
+  f.replies[`C1:${ts(1)}`] = [f.reply(50, ts(1), { text: "the answer" })];
+  f.respond(u => u.pathname.endsWith("conversations.replies") && failing.includes(u.searchParams.get("ts")!)
+    ? Response.json({ ok: false, error: "not_in_channel" }) : undefined);
+  const r = f.recovery();
+
+  await r.run();
+  expect(f.calls("conversations.replies").map(u => u.searchParams.get("ts"))).toEqual([ts(14), ts(13), ts(12), ts(11)]);
+  expect(f.rows().filter(row => row.destinationSession === "reachable@test")).toHaveLength(0);
+
+  // Every failing thread's 300-second retry is due again at the next pass; the reachable one still goes first.
+  f.time(at(100 + 301)); await r.run();
+  const second = f.calls("conversations.replies").slice(4).map(u => u.searchParams.get("ts"));
+  expect(second[0]).toBe(ts(1));
+  expect(f.rows().filter(row => row.destinationSession === "reachable@test")).toHaveLength(1);
+});
+
+it("a reply on the newest waiting ask just before a reconnect is recovered on the next pass, though the connect's own pass stopped short", async () => {
+  const f = fixture();
+  // Ten waiting asks, more than a pass's four thread calls.
+  for (let i = 1; i <= 10; i++) f.ask(ts(i), "pending", `open${i}@test`);
+  const r = f.recovery();
+  f.advance(5000);
+  for (let pass = 0; pass < 3; pass++) { await r.run(); f.advance(300_000); }
+  expect(new Set(f.calls("conversations.replies").map(u => u.searchParams.get("ts"))).size).toBe(10);
+
+  // A reply three seconds before live delivery resumes, and a pass at the connect itself, which stops short of it.
+  f.replies[`C1:${ts(10)}`] = [f.reply(1997, ts(10), { text: "just before" })];
+  f.time(at(2000)); r.connected();
+  const before = f.calls("conversations.replies").length;
+  await r.run();
+  const atConnect = f.calls("conversations.replies").slice(before).map(u => u.searchParams.get("ts"));
+  expect(atConnect[0]).toBe(ts(10));
+  expect(atConnect.length).toBeLessThan(10);
+  expect(f.rows().filter(row => row.destinationSession === "open10@test")).toHaveLength(0);
+
+  // Its call made progress, so it keeps its place ahead of the six threads not yet asked since the connect.
+  f.time(at(2300));
+  const next = f.calls("conversations.replies").length;
+  await r.run();
+  expect(f.calls("conversations.replies")[next]!.searchParams.get("ts")).toBe(ts(10));
+  expect(f.rows().filter(row => row.destinationSession === "open10@test")).toHaveLength(1);
+});
+
+it("a waiting ask whose backlog needs a second page keeps its place ahead of threads not yet asked", async () => {
+  const f = fixture();
+  for (let i = 1; i <= 10; i++) f.ask(ts(i), "pending", `open${i}@test`);
+  // The newest ask has twenty missed replies: more than a page.
+  f.replies[`C1:${ts(10)}`] = Array.from({ length: 20 }, (_, i) => f.reply(11 + i, ts(10)));
+  const r = f.recovery();
+  f.advance(5000);
+
+  await r.run();
+  const first = f.calls("conversations.replies").map(u => u.searchParams.get("ts"));
+  expect(first[0]).toBe(ts(10));
+  expect(first.length).toBeLessThan(10);
+  expect(f.rows().filter(row => row.destinationSession === "open10@test").length).toBeLessThan(20);
+
+  f.advance(300_000);
+  await r.run();
+  expect(f.calls("conversations.replies")[first.length]!.searchParams.get("ts")).toBe(ts(10));
+  expect(f.rows().filter(row => row.destinationSession === "open10@test")).toHaveLength(20);
+});
+
+it("on a machine the hub relays to, which can't see a gap in relayed events, a settled thread's backstop is three hours", async () => {
+  const f = fixture();
+  f.ask(ts(1), "done", "settled@test");
+  const r = f.recovery({ channels: [{ channel: "C1", seed: "adoption" }], relayed: true });
+  r.adoptRelayed("C1", at(100));
+  f.advance(5000); // past the settle margin, so a scan covers through boot (the boundary has its own test)
+  await r.run();
+  const scans = () => f.calls("conversations.replies").length;
+  expect(scans()).toBe(1);
+
+  f.time(at(100 + 2 * 3600)); await r.run();
+  expect(scans()).toBe(1);
+  expect(r.status().threads).toMatchObject({ settled: 1, settledRescanHours: 3 });
+  f.time(at(100 + 3 * 3600 + 60)); await r.run();
+  expect(scans()).toBe(2);
+});

packages/daemon/test/terminal-tile-size.test.ts

@@ -16,6 +16,16 @@ const helper = async (shell: string, ...more: string[]) => JSON.parse((await pro
   JSON.stringify(tileAttachArgv("tmux", ["-S", "TILE_SOCKET"], "seat", true, 1, 2)),
   ...more,
 ], { encoding: "utf8", timeout: 150000 })).stdout);
+// tmux 3.4 and later turn paste framing on only when a client starts, so a framing-off write between the first key and
+// the attach leaves a paste in that gap unframed: each of its lines would reach the agent as Enter.
+it("a writable tile turns paste framing off only on the return that doesn't attach", () => {
+  const script = tileAttachArgv("tmux", ["-S", "TILE_SOCKET"], "seat", true, 1, 2).find((arg) => arg.includes("use()")) ?? "";
+  const use = script.slice(script.indexOf("use()"), script.indexOf("take()"));
+  expect(use.length).toBeGreaterThan(0);
+  expect(use.split("\\033[?2004l").length - 1).toBe(1);
+  expect(use).toContain(`then printf "\\033[?2004l"; stty sane 2>/dev/null;`);
+});
+
 describe.skipIf(!available)("a tile views its seat without sizing it, and attaches only while the person types", () => {
   it.each(shells)("under %s: whole lines, the seat's own size, keys reaching the seat, and back to the view when idle", async (shell) => {
     expect(await helper(shell)).toEqual({