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

@@ -460,6 +460,8 @@ export interface ObservedCompaction {
   detail: string;
 }
 export interface PreparationState {
+  /** The reading that held this preparation send; never includes draft text. */
+  inputBox?: SendResult["heldInputBoxReading"];
   policyActor?: TerminalActor;
   attemptId: string;
   occupantGeneration: string | null;
@@ -536,6 +538,7 @@ export const RESTART_RECOVERY_WINDOW_MS_DEFAULT = 6 * 60 * 60_000;
 const DAEMON_PROCESS_ID = randomUUID();
 
 interface CompactionRecord {
+  inputBox?: PreparationState["inputBox"];
   /** The daemon process that last wrote it. */
   daemon: string;
   attemptId: string;
@@ -1566,7 +1569,8 @@ export class ClaudeCompactionEnforcer {
     // Record this attempt's effect even if cancellation/retry happened during the await.
     // Neither a failed response nor its loss proves no paste/Enter. Never replay uncertainty.
     attempt.delivery = prep?.ok && prep.outcome !== "retained" ? "delivered" : notSent ? "not_sent" : "uncertain";
-    this.persist(input.sessionName, attempt.attemptId, { delivery: attempt.delivery, sends: attempt.sends });
+    attempt.inputBox = prep?.heldInputBoxReading;
+    this.persist(input.sessionName, attempt.attemptId, { delivery: attempt.delivery, sends: attempt.sends, inputBox: attempt.inputBox });
     // The replacement request went out (or may have): the ended attempt's gate is used up.
     if (!notSent && attempt.afterEnded !== undefined && this.pendingPreCompactPrep.get(input.sessionName) === attempt) this.lastEnded.delete(input.sessionName);
     if (this.pendingPreCompactPrep.get(input.sessionName) !== attempt || attempt.controller.signal.aborted) return;
@@ -1876,7 +1880,7 @@ export class ClaudeCompactionEnforcer {
         compact: "not_sent", sends: record.sends,
         deadlineAt: record.deadlineAt ?? (record.delivery === "not_sent" ? null : record.updatedAt + AUTO_PREP_WAIT_MS_DEFAULT),
         policyWasEnabled: true, preparationRoot: record.preparationRoot, controller: new AbortController(),
-        compactInstruction: record.compactInstruction,
+        compactInstruction: record.compactInstruction, inputBox: record.inputBox,
         // A held attempt comes back held, with its recorded hold (never stamped again) and its map's age and usage.
         ...(held ? { prepared: true as const, preparedAt: record.preparedAt ?? record.updatedAt, preparedReason: record.preparedReason ?? "map_complete_seat_never_idle" } : {}),
         ...(held ? { mapReadyAt: record.mapReadyAt ?? record.preparedAt ?? record.updatedAt } : record.mapReadyAt !== undefined ? { mapReadyAt: record.mapReadyAt } : {}),

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

@@ -2099,6 +2099,8 @@ export interface SendResult {
   pastedOverInputBox?: InputBoxAnnotation["status"];
   /** A `preserveInput` hold: the input box reading it held on. */
   heldInputBox?: InputBoxAnnotation["status"];
+  /** The existing preserve-input reading, without the draft or suggestion text. No extra capture. */
+  heldInputBoxReading?: Pick<InputBoxAnnotation, "status" | "confidence" | "evidence" | "cursor">;
 }
 
 /** One reading of a Claude Code or Codex input box: `exact` (the sent text, word boundaries kept), `ambiguous`
@@ -3035,11 +3037,12 @@ export class SessionTransport {
     } : null;
     let existingInput: ExistingInput | undefined;
     let pastedOverInputBox: InputBoxAnnotation["status"] | undefined, heldInputBox: InputBoxAnnotation["status"] | undefined;
+    let heldInputBoxReading: SendResult["heldInputBoxReading"];
     let overrideBeforePaste: (() => Promise<void>) | undefined;
     const observe = (result: SendResult): SendResult => {
       if (checkedIdentity) result = { ...result, checkedIdentity, identityContract: 1 };
       if (pastedOverInputBox) result = { ...result, pastedOverInputBox };
-      if (heldInputBox && !result.ok) result = { ...result, heldInputBox };
+      if (heldInputBox && !result.ok) result = { ...result, heldInputBox, ...(heldInputBoxReading ? { heldInputBoxReading } : {}) };
       if (existingInput) result = { ...result, existingInput };
       if (result.ok && runtimeAdvisory) result = { ...result, warning: [runtimeAdvisory, result.warning].filter(Boolean).join(" ") };
       if (observed && this.captureObserver) {
@@ -3547,6 +3550,10 @@ export class SessionTransport {
       const typed = box?.status === "typed" || box?.status === "typed-with-suggestion";
       if (!clear && (opts?.preserveInput !== "typed" || typed)) {
         heldInputBox = box?.status ?? "unknown";
+        if (box) {
+          const { status, confidence, evidence, cursor } = box;
+          heldInputBoxReading = { status, confidence, evidence, cursor };
+        }
         throw new DeliveryGuardError(box?.status === "typed" ? "draft_input_busy" : "draft_input_unknown",
           `Managed input waits for an empty composer (it reads ${box?.status ?? "unknown"}). Your draft was left untouched; inspect the pane before retrying.`);
       }

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/test/compaction-draft-aware.test.ts

@@ -4,7 +4,7 @@
 // in-memory database; the tmux adapter records commands instead of running them.
 import Database from "better-sqlite3";
 import stringWidth from "string-width";
-import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs";
+import { mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { dirname, join } from "node:path";
 import { afterEach, describe, expect, it, vi } from "vitest";
@@ -12,7 +12,7 @@ import { TmuxAdapter } from "../src/adapters/tmux.js";
 import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
 import { migrate } from "../src/db/migrate.js";
 import type { ComposerSnapshot } from "../src/domain/composer-prompts.js";
-import { ClaudeCompactionEnforcer } from "../src/domain/claude-compaction-enforcer.js";
+import { ClaudeCompactionEnforcer, COMPACTION_RECORD_FILE } from "../src/domain/claude-compaction-enforcer.js";
 import { RigRepository } from "../src/domain/rig-repository.js";
 import { SeatDeliveryGuard, resolveGuardTarget } from "../src/domain/seat-delivery-guard.js";
 import { SessionRegistry } from "../src/domain/session-registry.js";
@@ -108,11 +108,17 @@ it("checks the composer after buffer setup, immediately before the preparation p
   expect(f.writes).toEqual([]);
 });
 
-it("an unreadable composer holds managed preparation without input", async () => {
+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(null);
+  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({ 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"]);
+  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([]);
 });
 

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({