docs(http-clients): a partial clicked answer does emit an event

The needs-me snapshot paragraph said recording one answer of several
emits no event yet. Since #3045, a click that records or changes an
answer emits queue.answer_recorded even while the set is partial; an
identical repeated choice, or a later click after a pending ask's set is
complete, emits none. Events stay best effort, so a client re-reads the
snapshot or row after a stream gap.

CHANGELOG.md

@@ -24,9 +24,9 @@ committed change. Two repairs aren't in this release: see Known limits.
 
 ### Before you upgrade
 
-- **Back up the database, then restart the daemon:** `rig daemon stop`, copy the
-  database file (`rig config get db.path` prints its path), then
-  `rig daemon start`. Your agents keep running through a daemon restart; don't
+- **Back up the database, install 0.7.13, then restart the daemon:** `rig daemon stop`,
+  copy the database file (`rig config get db.path` prints its path), install
+  OpenRig 0.7.13 using your existing installation method, then `rig daemon start`. Your agents keep running through a daemon restart; don't
   use `rig down`, which ends them. Three migrations run on the daemon's first start: 160 and 161 (for the experimental hub connector below)
   and 162 (indexes for session reads). Going back to 0.7.12 after them isn't
   established. ([#2993](https://github.com/mvschwarz/openrig/pull/2993), [#2997](https://github.com/mvschwarz/openrig/pull/2997))

docs/reference/daemon-http-clients.md

@@ -182,8 +182,9 @@ With `needs-me.read` supported:
   about the open rows blocked on it (at most 100; empty when none are recorded).
 - **A snapshot watermark:** the response's `watermark` is the newest event `seq` read before the snapshot. Follow
   `GET /api/events` from it with `Last-Event-ID: <watermark>`: every event after the snapshot is replayed, some possibly
-  already reflected in it. Recording one answer of several emits no event yet, so re-read the snapshot (or the row)
-  rather than relying on events alone for partial answers.
+  already reflected in it. A click that records or changes an answer emits `queue.answer_recorded` even while the set
+  is partial. An identical repeated choice, or a later click after a pending ask's answer set is complete, records
+  nothing new and emits none. Events are best effort; re-read the snapshot or row after a stream gap.
 - **The answers returned to a request:** `GET /api/queue/<qitemId>` adds `answers`, oldest first, each
   `{ qitemId, by, at, text, episode, recordedIdentity }`. They are found by the answer row's link to the request, not
   by its text, so don't parse the `In reply to:` line in an answer's body. `text` is the person's answer without that

packages/daemon/src/adapters/tmux.ts

@@ -494,7 +494,29 @@ export class TmuxAdapter {
       if (probe.state !== "present") return false;
       const panes = await this.listPanes(target.session);
       return panes.length === 1 && panes[0]!.id === target.pane && await this.isPaneDead(target.pane);
-    });
+    }, target => this.lastClientActivity(target.session));
+  }
+
+  /** When a writable tmux client attached to this session last had activity (epoch ms, whole seconds), or null when no
+   *  writable client is attached. tmux's client_activity is session-wide and covers more than text keys, so this is
+   *  recent writable-client activity, not proof of typing in the pane. A read-only client (a viewer tile) can't type
+   *  into the seat and doesn't count. An attach counts: tmux can't tell a key in the attach's second from the attach,
+   *  so a delivery can be held briefly after one. A tmux failure or an unreadable row throws: the guard then holds
+   *  the paste rather than take it for "no activity" (#3076). */
+  async lastClientActivity(session: string): Promise<number | null> {
+    const exact = exactTarget(session, "session");
+    const format = `#{client_readonly}${TMUX_FIELD_SEPARATOR}#{client_activity}`;
+    const output = await this.run(["tmux", "list-clients", "-t", exact, "-F", format],
+      `tmux list-clients -t ${shellQuote(exact)} -F "${format}"`);
+    let latest: number | null = null;
+    for (const line of output.split("\n").map(row => row.trim()).filter(row => row !== "")) {
+      const [readonly, activity, ...rest] = line.split(TMUX_FIELD_SEPARATOR);
+      if (rest.length || !/^[01]$/.test(readonly ?? "") || !/^[1-9]\d*$/.test(activity ?? "")) {
+        throw new Error(`tmux list-clients: unreadable client row for ${session}`);
+      }
+      if (readonly === "0") latest = Math.max(latest ?? 0, Number(activity) * 1000);
+    }
+    return latest;
   }
   private readonly freshProbes = new Map<string, string>();
   private readonly freshManaged = new Map<string, {nodeId: string; pane: string}>();
@@ -1081,12 +1103,14 @@ export class TmuxAdapter {
         // A paste the guard proves it left unsubmitted (#2429) is submitted first, under this lease, once this send's own
         // checks pass; they then aren't run a second time.
         beforePaste: async () => {
-          let checked = false;
-          const checks = async () => { await options?.beforePaste?.(); beforeWrite(); beforeInput?.(); checked = true; };
+          // This send's own checks can wait (a conversation lookup, an input capture), so they run first: the guard's
+          // read of the input is then the last wait before the paste (#3076). A check that refuses leaves a leftover's
+          // record for the next delivery.
+          await options?.beforePaste?.();
+          const checks = async () => { beforeWrite(); beforeInput?.(); };
           // The synchronous checks run again just before the key: the guard's re-read awaits after `checks`.
           const enter = async () => { beforeWrite(); beforeInput?.(); return (await this.sendKeysUnchecked(pane, ["Enter"])).ok; };
           if (checkDraft) await guard!.beforeInput(identity, false, text, { checks, enter });
-          if (!checked) await options?.beforePaste?.();
         },
         afterPaste: async () => { if (checkDraft) await guard!.notePasted(identity, text); },
       } : undefined;

packages/daemon/src/domain/execution-candidate-reader.ts

@@ -1,8 +1,9 @@
 import Database from "better-sqlite3";
 
-/** Rowids one statement may cover before the read yields to the event loop. Soft: one large row's conversion, native
- *  I/O or a correlated subquery on a matching row can still make a window take longer. */
-const WINDOW_ROWS = 256;
+/** Rows read per slice, and the time a slice may take, before yielding to the event loop. Both are soft: one native
+ *  step or row conversion can take longer. */
+const SLICE_ROWS = 64;
+const SLICE_MS = 8;
 
 /**
  * The main database file when the execution view's candidate read can run cooperatively on its own connection, or
@@ -17,66 +18,45 @@ export function cooperativeReadFile(db: Database.Database): string | null {
   return db.pragma("journal_mode", { simple: true }) === "wal" ? main.file : null;
 }
 
-/** One cooperative read: its rows, the windows it read, and the longest synchronous stretch it ran between yields. */
-export interface CooperativeRead<Row> {
-  rows: Row[];
-  windows: number;
-  longestStretchMs: number;
-}
-
 /**
- * Reads every row of one SELECT over `table` (a table name from code, never input) on a new readonly connection, one
- * rowid window at a time, yielding to the event loop between windows so unrelated requests progress during a long
- * scan. `sql` reads one window: it binds the window's exclusive lower and inclusive upper rowid first
- * (`table.rowid > ? AND table.rowid <= ?`), then `binds`. Each statement steps over at most `windowRows` rowids however
- * few rows its filter matches; a cursor that yields between returned rows can scan the whole table in one step when
- * matches are rare. One read transaction keeps one snapshot for every window, and rows come back in rowid order. The
- * transaction and the connection end before returning, on every path; a read error is rethrown as-is and no partial
- * rows are returned.
+ * Reads every row of one SELECT on a new readonly connection to `file`, yielding to the event loop between slices so
+ * unrelated requests progress during a long scan. One cursor keeps one snapshot for the whole read. The cursor and
+ * the connection are closed before returning, on every path; a read error is rethrown as-is and no partial rows are
+ * returned.
  */
 export async function readRowsCooperatively<Row>(
   file: string,
-  table: string,
   sql: string,
   binds: unknown[],
   open: (file: string) => Database.Database = (path) => new Database(path, { readonly: true, fileMustExist: true }),
-  windowRows = WINDOW_ROWS,
-): Promise<CooperativeRead<Row>> {
+): Promise<Row[]> {
   let reader: Database.Database | undefined;
+  let cursor: IterableIterator<unknown> | undefined;
   let rows: Row[] = [];
-  let windows = 0, longestStretchMs = 0, stretchStart = performance.now();
-  const endStretch = () => { longestStretchMs = Math.max(longestStretchMs, performance.now() - stretchStart); };
   let failure: { error: unknown } | undefined;
   let cleanupFailure: { error: unknown } | undefined;
   try {
     reader = open(file);
-    reader.exec("BEGIN");
-    // The first read takes the snapshot every window shares. Each scalar subquery seeks one end of the rowid b-tree;
-    // one SELECT of MIN and MAX together would scan a whole index instead.
-    const { lo, hi } = reader.prepare(`SELECT (SELECT MIN(rowid) FROM ${table}) AS lo, (SELECT MAX(rowid) FROM ${table}) AS hi`)
-      .get() as { lo: number | null; hi: number | null };
-    const window = reader.prepare(sql);
-    if (lo !== null && hi !== null) {
-      for (let after = lo - 1; after < hi;) {
-        const through = Math.min(after + windowRows, hi);
-        for (const row of window.all(after, through, ...binds)) rows.push(row as Row);
-        windows++;
-        after = through;
-        if (after < hi) {
-          endStretch();
-          await new Promise<void>((resolve) => setImmediate(resolve));
-          stretchStart = performance.now();
-        }
+    cursor = reader.prepare(sql).iterate(...binds);
+    for (;;) {
+      const sliceStart = performance.now();
+      let read = 0;
+      let step = cursor.next();
+      while (!step.done) {
+        rows.push(step.value as Row);
+        if (++read >= SLICE_ROWS || performance.now() - sliceStart >= SLICE_MS) break;
+        step = cursor.next();
       }
+      if (step.done) break;
+      await new Promise<void>((resolve) => setImmediate(resolve));
     }
   } catch (error) {
     failure = { error };
   } finally {
-    try { if (reader?.inTransaction) reader.exec("ROLLBACK"); } catch (error) { cleanupFailure = { error }; }
+    try { cursor?.return?.(); } catch (error) { cleanupFailure = { error }; }
     try { reader?.close(); } catch (error) { cleanupFailure ??= { error }; }
   }
   if (failure) { rows = []; throw failure.error; }
   if (cleanupFailure) { rows = []; throw cleanupFailure.error; }
-  endStretch();
-  return { rows, windows, longestStretchMs };
+  return rows;
 }

packages/daemon/src/domain/execution-view.ts

@@ -54,7 +54,6 @@ import { resolveLegacyTopologyRigsRoot } from "./user-settings/settings-store.js
 import { BUILD_INFO, type BuildInfo } from "../build-info.js";
 import { validateMissionComposition } from "./lifecycle-manifest.js";
 import { cooperativeReadFile, readRowsCooperatively } from "./execution-candidate-reader.js";
-import { recordSlowDiagnostic } from "./sync-site-wrap.js";
 
 export const INDETERMINATE = "INDETERMINATE" as const;
 export type Indeterminate = typeof INDETERMINATE;
@@ -749,7 +748,6 @@ async function gitAncestor(exec: ExecutionViewDeps["exec"], repoCtx: string, sha
 }
 
 export async function buildExecutionView(deps: ExecutionViewDeps, opts?: { mission?: string; rig?: string; project?: string }): Promise<Record<string, unknown>> {
-  const projectionStarted = performance.now();
   const now = deps.now ?? (() => new Date());
   const exec = deps.exec ?? defaultExec;
   const buildInfo = deps.buildInfo ?? BUILD_INFO;
@@ -808,30 +806,23 @@ export async function buildExecutionView(deps: ExecutionViewDeps, opts?: { missi
   // post_claim_motion feeds only derivePickup's legacy fallback, which pending and in-progress rows reach: a blocked row
   // returns parked and a closed row terminal before the count is read. Every row and body is still selected here (the
   // ladder, EC-3 and wave-map evidence need historical rows); only their unused history count is skipped.
-  const candidateSelect = `SELECT qitem_id, source_session, destination_session, state, tags, body, summary, claimed_at,
+  const candidateSql = `SELECT qitem_id, source_session, destination_session, state, tags, body, summary, claimed_at,
               last_heartbeat, blocked_on, ts_created, ts_updated,
               CASE WHEN queue_items.state IN ('pending','in-progress') THEN (SELECT COUNT(*) FROM queue_transitions t
                  WHERE t.qitem_id = queue_items.qitem_id
                    AND t.ts > queue_items.claimed_at
                    AND t.transition_note IS NOT 'claimed') ELSE 0 END AS post_claim_motion
-         FROM queue_items`;
+         FROM queue_items
+        WHERE (${missionWhere})`;
   const candidateBinds = missionLikes.flatMap((like) => [like, like]);
-  // On a WAL database file this full scan runs on its own readonly connection in rowid windows, yielding between
-  // windows so unrelated work can run even when few rows match; the same filter runs inside each window, so the rows
-  // are the ones the single statement returns. A native open, a large row or a window's matching rows can still
-  // block for a while, since the window bound is soft. It sees committed data as of its first read. Inside a
-  // transaction, or for a memory or non-WAL database, it stays one synchronous read on the main handle (no await on
-  // that branch).
+  // On a WAL database file this full scan runs on its own readonly cursor and yields between slices so unrelated
+  // work can run; a native open, prepare or step can still block, since the slice budget is soft. It sees committed
+  // data as of its first step. Inside a transaction, or for a memory or non-WAL database, it stays a synchronous
+  // read on the main handle (no await on that branch).
   const cooperativeFile = cooperativeReadFile(deps.db);
-  const candidateStarted = performance.now();
-  const candidateRead = cooperativeFile === null
-    ? null
-    : await readRowsCooperatively<QueueRowLite>(cooperativeFile, "queue_items",
-      `${candidateSelect} WHERE queue_items.rowid > ? AND queue_items.rowid <= ? AND (${missionWhere})`, candidateBinds);
-  const candidateRows = candidateRead
-    ? candidateRead.rows
-    : deps.db.prepare(`${candidateSelect} WHERE (${missionWhere})`).all(...candidateBinds) as QueueRowLite[];
-  const candidateMs = performance.now() - candidateStarted;
+  const candidateRows = cooperativeFile === null
+    ? deps.db.prepare(candidateSql).all(...candidateBinds) as QueueRowLite[]
+    : await readRowsCooperatively<QueueRowLite>(cooperativeFile, candidateSql, candidateBinds);
 
   const rows = candidateRows
     .filter(row => mission === INDETERMINATE || rowBindsToMission(row.tags, missionRefs))
@@ -1366,7 +1357,7 @@ export async function buildExecutionView(deps: ExecutionViewDeps, opts?: { missi
     df_margin: dfMargin,
   };
 
-  const view = {
+  return {
     ...(opts?.project ? { project: opts.project, membership: "exact project:<id> queue tags and lifecycle identity; unscoped rows excluded" } : {}),
     view: "execution",
     readiness,
@@ -1419,19 +1410,4 @@ export async function buildExecutionView(deps: ExecutionViewDeps, opts?: { missi
     q5_park: q5,
     q6_parallelism: q6,
   };
-  // Measurement only, for a slow projection: its elapsed time, the candidate read's elapsed time, and the candidate
-  // read's longest synchronous stretch, which a whole-view or awaited span cannot show. The main-handle read is one
-  // stretch. Counts and times only; no row content.
-  const projectionMs = performance.now() - projectionStarted;
-  recordSlowDiagnostic(projectionMs, {
-    schema: "openrig.execution-view/v1",
-    utc: new Date().toISOString(),
-    projectionMs,
-    candidateMs,
-    candidateRows: candidateRows.length,
-    candidateRead: candidateRead ? "windowed" : "main-handle",
-    candidateWindows: candidateRead?.windows ?? null,
-    candidateLongestStretchMs: candidateRead?.longestStretchMs ?? candidateMs,
-  });
-  return view;
 }

packages/daemon/src/domain/queue-repository.ts

@@ -105,6 +105,7 @@ import { armQueueWait, backOffQueueWait, refreshQueueWaits, evaluateQueueWait, r
 import { parseHumanQuestions, unansweredQuestions, describeTypedReplyPlacement, formatHumanAnswers, humanAnswerText, type HumanQuestion, type HumanAnswers, type TypedHumanReply, type RecordHumanAnswerResult } from "./human-questions.js";
 import { slackMicros } from "./gateway/slack/state-store.js";
 import { askBackRoute, RUNNER_CLOSE_WAITS, terminalAnswerMeaning } from "./receiver-context.js";
+import { runSyncSite, runSyncSiteWhenSlow } from "./sync-site-wrap.js";
 
 /** A direct ask's answer row ID: from the ask, or from the ask and the park episode its answer belongs to. */
 function humanAnswerId(askId: string, episodeKey?: string): string {
@@ -821,14 +822,14 @@ export function createLinkedSuccessorLookup(db: Database.Database, sourceQitemId
     const startedAt = performance.now();
     const map = new Map<string, Array<{ qitemId: string; destination: string | null }>>();
     const ids = JSON.stringify([...sources]);
-    const rows = db.prepare(
+    const rows = runSyncSite("queue.linked_successors.build", () => db.prepare(
       `SELECT s.qitem_id, s.destination_session, s.handed_off_from AS source FROM queue_items s
         WHERE s.handed_off_from IN (SELECT value FROM json_each(?))
        UNION
        SELECT s.qitem_id, s.destination_session, j.value AS source FROM queue_items s, json_each(s.chain_of_record) j
         WHERE s.chain_of_record IS NOT NULL AND json_valid(s.chain_of_record)
           AND j.value IN (SELECT value FROM json_each(?))`,
-    ).all(ids, ids) as Array<{ qitem_id: string; destination_session: string | null; source: string }>;
+    ).all(ids, ids)) as Array<{ qitem_id: string; destination_session: string | null; source: string }>;
     for (const r of rows) {
       const list = map.get(r.source);
       const entry = { qitemId: r.qitem_id, destination: r.destination_session };
@@ -2634,8 +2635,13 @@ export class QueueRepository {
   }
 
   /** {@link handoff}'s whole act, synchronously: its transaction commits before this returns, and the wake is delivered
-   *  after commit without being awaited, so a synchronous caller (an answer's close path) can hand a row on. */
+   *  after commit without being awaited, so a synchronous caller (an answer's close path) can hand a row on. Named in
+   *  the slow-operation log, so a stall during a handoff is attributed to it. */
   private handoffNow(input: QueueHandoffInput): { closed: QueueItem; created: QueueItem } {
+    return runSyncSite("queue.handoff", () => this.handoffWithin(input));
+  }
+
+  private handoffWithin(input: QueueHandoffInput): { closed: QueueItem; created: QueueItem } {
     const source = this.getById(input.qitemId);
     if (!source) {
       throw new QueueRepositoryError(
@@ -4563,8 +4569,9 @@ export class QueueRepository {
       typeof opts?.excludeOperationReply === "string" && opts.excludeOperationReply.length > 0
         ? mcOperationTag(opts.excludeOperationReply)
         : null;
+    // Named only when slow: it runs on every row read and on each row of the attention read.
     const replies = this.hasReplyToColumn
-      ? (excludeOp
+      ? runSyncSiteWhenSlow("queue.decision_revision.replies", () => excludeOp
           ? (this.db
               .prepare(
                 "SELECT qitem_id FROM queue_items WHERE reply_to = ? AND NOT (tags IS NOT NULL AND json_valid(tags) AND EXISTS (SELECT 1 FROM json_each(tags) WHERE value = ?)) ORDER BY ts_created, rowid",

packages/daemon/src/domain/seat-delivery-guard.ts

@@ -60,6 +60,15 @@ export interface TypingGuardConfig extends TypingGuardSettings { revision: strin
 const DEFAULT_GUARD: TypingGuardConfig = { mode: "off", revision: "legacy-off" };
 /** How long the guard waits before reading again an input it couldn't read right after its own paste or Enter (#2429). */
 const OWN_INPUT_REREAD_MS = 300;
+/** Right before an automatic paste, the empty input is read once more after this wait: a draft typed (or redrawn by the
+ *  agent's TUI) since the first read is seen before the paste, not after it, where nothing is removed (#2447, #3076). */
+const PASTE_SETTLE_MS = 300;
+/** Recent activity at a writable tmux client of the seat (keys, but also other client input; session-wide, not proof of
+ *  typing in this pane) holds the paste: a later attempt can deliver once it reads quiet and the input empty. tmux
+ *  reports client activity in whole seconds, so this is three of them (#3076). */
+const INPUT_QUIET_MS = 3000;
+/** tmux's client_activity is in whole seconds; the activity may have come up to this much after the second it reports. */
+const CLIENT_ACTIVITY_PRECISION_MS = 999;
 /** A terminal member with nothing set holds OpenRig's own notices and takes deliberate sends. */
 const TERMINAL_DEFAULT: TypingGuardConfig = { mode: "off", level: "sends-only", revision: "terminal-default" };
 /** The actor a guard change records when its request carries no seat identity, as from an ordinary terminal (#2350).
@@ -122,6 +131,9 @@ export class SeatDeliveryGuard {
   /** A paste this guard kept from being submitted because it couldn't read the input, per node (#2429). */
   private readonly leftovers = new Map<string, { target: GuardTarget; humanEpoch: number; text: string; collapsedPaste: string | null }>();
   private inspectInput?: (target: GuardTarget, owned?: ComposerOwnership) => Promise<ComposerInput | null>;
+  /** When a writable tmux client of the seat last had activity (epoch ms, whole seconds); null when no writable client
+   *  is attached; a rejection when that can't be known. */
+  private inputActivity?: (target: GuardTarget) => Promise<number | null>;
   private inputAbsent?: (target: GuardTarget) => Promise<boolean>;
   private readonly hasModeConfig: boolean;
 
@@ -133,9 +145,11 @@ export class SeatDeliveryGuard {
     this.hasModeConfig = (db.prepare("PRAGMA table_info(seat_delivery_guards)").all() as Array<{ name: string }>).some(c => c.name === "desired_config");
   }
 
-  attachInputInspection(inspect: (target: GuardTarget, owned?: ComposerOwnership) => Promise<ComposerInput | null>, absent: (target: GuardTarget) => Promise<boolean>): void {
+  attachInputInspection(inspect: (target: GuardTarget, owned?: ComposerOwnership) => Promise<ComposerInput | null>, absent: (target: GuardTarget) => Promise<boolean>,
+    activity?: (target: GuardTarget) => Promise<number | null>): void {
     this.inspectInput = inspect;
     this.inputAbsent = absent;
+    this.inputActivity = activity;
   }
 
   /** Startup-only, before exposing routes or starting writers. A stopped operation cannot
@@ -483,6 +497,9 @@ export class SeatDeliveryGuard {
     } else if (input.state !== "empty") {
       throw new DeliveryGuardError("draft_input_busy", "The seat has an unsubmitted input. Automatic delivery is held.");
     } else {
+      // Right before the paste (the adapter's check passes the text): the read above is a snapshot, and a draft typed
+      // after it would get the paste appended and then lose the Enter, with the paste left in the person's input (#3076).
+      if (pasteText !== undefined) input = await this.quietBeforePaste(target, read, unknown);
       const uncertainty = pasteText === undefined ? null : composerPasteUncertainty(input, pasteText);
       if (uncertainty) {
         const message = uncertainty === "controls"
@@ -496,6 +513,30 @@ export class SeatDeliveryGuard {
     }
   }
 
+  /** No recent writable-client activity before a short wait, the input still empty when read again after it, and still
+   *  no recent activity after that read. A time tmux can't report holds the paste, as an unreadable input does. The
+   *  read it returns is the one the paste goes on. Typing after this final check and before the paste can still leave
+   *  the notice beside the person's draft: the Enter is then refused, nothing is submitted or deleted (#3076). */
+  private async quietBeforePaste(target: GuardTarget, read: () => Promise<ComposerInput | null>,
+    unknown: (reading: ComposerInput | null) => DeliveryGuardError): Promise<ComposerInput> {
+    const quiet = async () => {
+      if (!this.inputActivity) return;
+      const last = await this.inputActivity(target).then(at => ({ at }), () => null);
+      if (!last) throw new DeliveryGuardError("draft_input_unknown", "OpenRig couldn't read the recent activity of the seat's terminal clients. Automatic delivery is held; nothing was pasted.");
+      // tmux reports whole seconds: activity at N.999 s reads as N, so the quiet window counts from the end of that second.
+      if (last.at !== null && Date.now() - (last.at + CLIENT_ACTIVITY_PRECISION_MS) < INPUT_QUIET_MS) {
+        throw new DeliveryGuardError("draft_input_busy", "A writable client of this seat had recent activity. Automatic delivery is held; nothing was pasted.");
+      }
+    };
+    await quiet();
+    await this.sleep(PASTE_SETTLE_MS);
+    const again = await read();
+    if (!again || again.state === "unknown") throw unknown(again);
+    if (again.state !== "empty") throw new DeliveryGuardError("draft_input_busy", "The input changed just before the paste. Automatic delivery is held; nothing was pasted.");
+    await quiet();
+    return again;
+  }
+
   /** A leftover paste is OpenRig's only while it's exactly there: the same occupant, no person's input through OpenRig
    * since, and its text or the label recorded for it. A label first seen later proves nothing (#1871). */
   private ownsLeftover(target: GuardTarget, input: ComposerInput): boolean {

packages/daemon/src/domain/slow-op-recorder.ts

@@ -48,6 +48,8 @@ export interface SlowOperationInstrumentation {
     classify?: (value: T) => "ok" | "failed",
   ): Promise<T>;
   recordRequest?(site: string, durationMs: number): void;
+  /** A synchronous site's duration, recorded only at or above the slow threshold (runSyncSiteWhenSlow). */
+  recordSlowSync?(site: string, durationMs: number, outcome: "ok" | "failed"): void;
   /** Bounded nonblocking diagnostic append; false means observation was dropped. */
   recordDiagnostic?(record: Record<string, unknown>): boolean;
   snapshot?(): SlowOperationSnapshot;
@@ -262,6 +264,11 @@ export class SlowOpRecorder implements SlowOperationInstrumentation {
     this.recordMeasurement(`request:${site}`, durationMs);
   }
 
+  recordSlowSync(site: string, durationMs: number, outcome: "ok" | "failed" = "ok"): void {
+    if (!Number.isFinite(durationMs) || durationMs < this.slowThresholdMs) return;
+    this.recordMeasurement(site, durationMs, outcome);
+  }
+
   recordDiagnostic(record: Record<string, unknown>): boolean {
     this.diagnostic.offered++;
     const coverage = { ...this.diagnostic, pending: this.diagnosticIds.size,

packages/daemon/src/domain/sync-site-wrap.ts

@@ -1,4 +1,4 @@
-import { SLOW_OPERATION_THRESHOLD_MS, type SlowOperationInstrumentation } from "./slow-op-recorder.js";
+import type { SlowOperationInstrumentation } from "./slow-op-recorder.js";
 
 let recorder: SlowOperationInstrumentation | undefined;
 
@@ -10,6 +10,26 @@ export function runSyncSite<T>(site: string, fn: () => T): T {
   return recorder?.runSync ? recorder.runSync(site, fn) : fn();
 }
 
+/**
+ * runSyncSite for a site on a hot path (per request, per row), where a begin and an end record per call would crowd
+ * the bounded log: it records only a call that took at least the recorder's slow threshold, and writes no begin
+ * record, so a call that wedges the loop is named once it ends rather than while it runs.
+ */
+export function runSyncSiteWhenSlow<T>(site: string, fn: () => T): T {
+  const record = recorder?.recordSlowSync?.bind(recorder);
+  if (!record) return fn();
+  const startedAt = performance.now();
+  let outcome: "ok" | "failed" = "failed";
+  try {
+    const value = fn();
+    outcome = "ok";
+    return value;
+  } finally {
+    // Measurement only: a failing record never replaces fn's own result or error.
+    try { record(site, performance.now() - startedAt, outcome); } catch { /* the recorder reports its own health */ }
+  }
+}
+
 /**
  * Async twin of runSyncSite: instruments a NON-blocking site through the
  * recorder's async `runStage` seam (falls back to a bare call when no recorder
@@ -21,13 +41,3 @@ export function runSyncSite<T>(site: string, fn: () => T): T {
 export async function runAsyncSite<T>(site: string, fn: () => Promise<T>): Promise<T> {
   return recorder?.runStage ? recorder.runStage(site, fn) : fn();
 }
-
-/**
- * Offers one diagnostic record to the configured recorder's bounded log, only when `elapsedMs` reached the slow
- * threshold, so a hot path adds no record per call. Measurement only: no recorder, a dropped record or a throwing
- * recorder changes nothing for the caller.
- */
-export function recordSlowDiagnostic(elapsedMs: number, record: Record<string, unknown>): void {
-  if (!(elapsedMs >= SLOW_OPERATION_THRESHOLD_MS)) return;
-  try { recorder?.recordDiagnostic?.(record); } catch { /* the recorder reports its own health */ }
-}

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

@@ -32,7 +32,7 @@ function signal() {
   return { ready: new Promise<void>(resolve => { release = resolve; }), release: () => release() };
 }
 
-function fixture(saved?: Buffer, opts: { runtime?: string; idleHook?: boolean } = {}) {
+function fixture(saved?: Buffer, opts: { runtime?: string; idleHook?: boolean; appendPaste?: boolean } = {}) {
   const db = new Database(saved ?? ":memory:");
   if (!saved) {
     migrate(db, ALL_MIGRATIONS);
@@ -42,23 +42,27 @@ function fixture(saved?: Buffer, opts: { runtime?: string; idleHook?: boolean }
       INSERT INTO sessions(id,node_id,session_name,status) VALUES ('sa','a','worker@test','running'),('sb','b','sibling@test','running');
       INSERT INTO occupant_tenures(id,node_id,generation_ordinal,generation_uuid,kind) VALUES ('ga','a',1,'g1','fresh'),('gb','b',1,'g2','fresh');`);
   }
-  const guard = new SeatDeliveryGuard(db, name => resolveGuardTarget(db, name));
+  const hooks: { load?: () => void | Promise<void>; paste?: () => void | Promise<void>; sleep?: (ms: number) => void | Promise<void>;
+    settle?: (ms: number) => void | Promise<void>; clients?: () => string } = {};
+  // The guard's own waits (its re-reads) go to `settle`, kept apart from the transport's `sleep`; none is real time.
+  const guard = new SeatDeliveryGuard(db, name => resolveGuardTarget(db, name), async ms => { await hooks.settle?.(ms); });
   guard.recoverActivation();
   const bodies: Record<string, string | null> = { a: "", b: "" };
   const history: Record<string, string> = { a: "", b: "" };
   const writes: string[] = [], submissions: string[] = [];
   const observations: unknown[] = [];
-  const hooks: { load?: () => void | Promise<void>; paste?: () => void | Promise<void>; sleep?: (ms: number) => void | Promise<void> } = {};
   let staged = "";
   // A terminal pane shows a submitted line as typed; an agent pane echoes a sent message under its prompt, as Claude does.
   const echo = (node: string, body: string) => (db.prepare("SELECT runtime FROM nodes WHERE id=?").get(node) as { runtime: string }).runtime === "terminal"
     ? `${body}\n` : `${body.split("\n").map((row, i) => (i ? "  " : "❯ ") + row).join("\n")}\n`;
   const tmux = new TmuxAdapter(async command => {
+    if (command.includes("list-clients")) return hooks.clients?.() ?? "";
     if (command.includes("load-buffer")) await hooks.load?.();
     const node = command.includes("%2") ? "b" : "a";
     if (command.includes("paste-buffer")) {
       await hooks.paste?.();
-      writes.push(`paste:${node}:${staged}`); bodies[node] = staged;
+      // appendPaste: a paste lands after whatever the input already holds, as in a real composer (#3076 tests).
+      writes.push(`paste:${node}:${staged}`); bodies[node] = opts.appendPaste ? (bodies[node] ?? "") + staged : staged;
     }
     if (command.includes("send-keys")) {
       writes.push(`keys:${node}:${command}`);
@@ -1156,3 +1160,158 @@ describe("draft-aware at a full held quota (#2430)", () => {
     expect(f.outbox.retentionUsage("a").count).toBe(100);
   });
 });
+
+
+describe("#3076: a draft typed between the guard's read and the paste", () => {
+  const draft = "Please send it to the team. Also";
+  const notice = "[OpenRig watchdog scheduler · policy: periodic-reminder · job: j]\nWake timer fired.";
+  const pasted = (f: ReturnType<typeof fixture>) => f.writes.filter(w => w.startsWith("paste:"));
+  const setActivity = (f: ReturnType<typeof fixture>, at: () => number | null) => {
+    vi.spyOn(f.tmux, "lastClientActivity").mockImplementation(async () => at());
+  };
+
+  it("is seen by the re-read after the settle wait: the notice is held and nothing is pasted into the person's input", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    const waits: number[] = [];
+    // The person starts typing during the guard's wait right before the paste (the first read saw an empty input).
+    f.hooks.settle = ms => { waits.push(ms); if (!pasted(f).length) f.bodies.a = draft; };
+    const result = await f.send("watchdog-wake", notice, { openrigNotice: true, verify: false });
+    expect(result).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
+    expect(waits[0]).toBe(300);
+    expect(pasted(f)).toEqual([]);
+    expect(f.submissions).toEqual([]);
+    expect(f.bodies.a).toBe(draft);
+  });
+
+  it("holds on recent writable-client activity, also activity during the wait, and a later attempt delivers once quiet", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    let last: number | null = Date.now() - 1000;
+    setActivity(f, () => last);
+    const busy = await f.send("wake-1", "Wake timer fired.", { openrigNotice: true, verify: false });
+    expect(busy).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
+
+    last = Date.now() - 60_000;
+    f.hooks.settle = () => { last = Date.now(); };   // activity during the settle wait, nothing drawn yet
+    f.advance();
+    const during = await f.send("wake-2", "Wake timer fired again.", { openrigNotice: true, verify: false });
+    expect(during).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
+    expect(pasted(f)).toEqual([]);
+
+    last = Date.now() - 60_000;
+    f.hooks.settle = undefined;
+    f.advance();
+    const sent = await f.send("wake-3", "Wake timer fired a third time.", { openrigNotice: true, verify: false });
+    expect(sent).toMatchObject({ ok: true });
+    expect(f.submissions).toHaveLength(1);
+  });
+
+  it("counts the quiet window from the end of the whole second tmux reports: 3.5 s ago still holds, 4.5 s ago delivers", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    let last = Date.now() - 3500;   // activity at N.999 s reads as N: it may have been 2.5 s ago
+    setActivity(f, () => last);
+    const held = await f.send("whole-second", "Wake timer fired.", { openrigNotice: true, verify: false });
+    expect(held).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
+    last = Date.now() - 4500;
+    f.advance();
+    const sent = await f.send("whole-second-later", "Wake timer fired again.", { openrigNotice: true, verify: false });
+    expect(sent).toMatchObject({ ok: true });
+  });
+
+  it("holds as unreadable when tmux can't report recent writable-client activity; with no writable client it delivers", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    vi.spyOn(f.tmux, "lastClientActivity").mockRejectedValue(new Error("command timed out"));
+    const held = await f.send("activity-unreadable", notice, { openrigNotice: true, verify: false });
+    expect(held).toMatchObject({ sent: false, reason: "draft_input_unknown" });
+    expect(pasted(f)).toEqual([]);
+    setActivity(f, () => null);
+    f.advance();
+    const sent = await f.send("no-client", "Wake timer fired again.", { openrigNotice: true, verify: false });
+    expect(sent).toMatchObject({ ok: true });
+  });
+
+  it("holds as unreadable when the input can't be read after the wait", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    f.hooks.settle = () => { if (!pasted(f).length) f.bodies.a = null; };
+    const result = await f.send("unreadable", notice, { openrigNotice: true, verify: false });
+    expect(result).toMatchObject({ sent: false, reason: "draft_input_unknown" });
+    expect(pasted(f)).toEqual([]);
+  });
+
+  it("runs the send's own waiting checks before the guard's last read, so a draft typed while they wait is seen", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    // A conditional send's checks await (a conversation lookup); the person types meanwhile. The adapter returns the
+    // guard's refusal as its result.
+    const result = await f.guard.operation("worker@test", () =>
+      f.tmux.sendText("worker@test", notice, undefined, { beforePaste: async () => { f.bodies.a = draft; } }));
+    expect(result).toMatchObject({ ok: false, code: "draft_input_busy" });
+    expect(pasted(f)).toEqual([]);
+    expect(f.bodies.a).toBe(draft);
+  });
+
+  it("applies to every automatic input under draft-aware, a deliberate send included, as the empty check already did", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    f.hooks.settle = () => { if (!pasted(f).length) f.bodies.a = draft; };
+    const result = await f.send("deliberate", "a message from another seat", { verify: false });
+    expect(result).toMatchObject({ sent: false, reason: "draft_input_busy" });
+    expect(pasted(f)).toEqual([]);
+    expect(f.bodies.a).toBe(draft);
+  });
+
+  it("the remaining gap: typing after the final check and before the paste leaves the notice beside the draft, never submitted or deleted", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    let typed = false;
+    // The person types between the guard's last check and tmux's paste: the paste lands after their text.
+    f.hooks.paste = () => { if (!typed) { typed = true; f.bodies.a = draft; } };
+    const result = await f.send("final-boundary", notice, { openrigNotice: true, verify: false });
+    // Contamination: exactly one paste, and the input is the person's text followed by it, nothing removed.
+    expect(pasted(f)).toHaveLength(1);
+    const staged = pasted(f)[0]!.slice("paste:a:".length);
+    expect(f.bodies.a).toBe(draft + staged);
+    // Submission: none. No key at all reached the pane (no Enter, C-m or C-u), and the guard says why.
+    expect(f.writes.filter(w => w.startsWith("keys:a"))).toEqual([]);
+    expect(f.submissions).toEqual([]);
+    expect(result).toMatchObject({ ok: false });
+    expect(JSON.stringify(result)).toContain("draft_input_changed");
+  });
+
+  it("checks the activity again after the final read: activity that appears at that read holds the paste", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    let last: number | null = Date.now() - 60_000;
+    setActivity(f, () => last);
+    const real = f.tmux.captureComposerSnapshot.getMockImplementation()!;
+    let armed = false;
+    f.hooks.settle = () => { armed = true; };
+    // The composer still reads empty at the final read, but a writable client has activity right then.
+    vi.spyOn(f.tmux, "captureComposerSnapshot").mockImplementation(async (name: string) => {
+      const shot = await real(name);
+      if (armed) { armed = false; last = Date.now(); }
+      return shot;
+    });
+    const result = await f.send("activity-at-final-read", notice, { openrigNotice: true, verify: false });
+    expect(result).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
+    expect(pasted(f)).toEqual([]);
+  });
+
+  it("reads tmux's real list-clients output end to end: an attach in this second holds, a quiet client delivers", async () => {
+    const f = fixture(undefined, { runtime: "claude-code", appendPaste: true });
+    await f.enable();
+    const now = () => Math.floor(Date.now() / 1000);
+    f.hooks.clients = () => `1|${now()}\n0|${now()}\n`;   // a viewer tile and a writable client attached this second
+    const held = await f.send("attach-now", notice, { openrigNotice: true, verify: false });
+    expect(held).toMatchObject({ outcome: "retained", sent: false, reason: "draft_input_busy" });
+    expect(pasted(f)).toEqual([]);
+    f.hooks.clients = () => `1|${now()}\n0|${now() - 10}\n`;   // only the read-only tile is active
+    f.advance();
+    const sent = await f.send("attach-quiet", "Wake timer fired again.", { openrigNotice: true, verify: false });
+    expect(sent).toMatchObject({ ok: true });
+  });
+});

packages/daemon/test/execution-candidate-reader.test.ts

@@ -17,8 +17,6 @@ import { viewsCustomSchema } from "../src/db/migrations/030_views_custom.js";
 import { EventBus } from "../src/domain/event-bus.js";
 import { ViewProjector } from "../src/domain/view-projector.js";
 import { cooperativeReadFile, readRowsCooperatively } from "../src/domain/execution-candidate-reader.js";
-import { buildExecutionView } from "../src/domain/execution-view.js";
-import { configureSyncSiteRecorder, recordSlowDiagnostic } from "../src/domain/sync-site-wrap.js";
 
 const dirs: string[] = [];
 const handles: Database.Database[] = [];
@@ -55,69 +53,42 @@ function turns(): { count: () => number; stop: () => void } {
   return { count: () => count, stop: () => { stopped = true; } };
 }
 
-/** One window of t, the way the execution view binds its windows: exclusive lower and inclusive upper rowid first. */
-const WINDOW_SQL = "SELECT id, v FROM t WHERE id > ? AND id <= ? ORDER BY id";
-
 describe("cooperative candidate reads", () => {
-  it("reads every row in rowid order on its own connection, a window at a time, letting other callbacks run between windows", async () => {
+  it("reads every row in order on its own connection, letting other callbacks run between slices", async () => {
     const { file } = walTable(500);
     const loop = turns(), reader = recordingOpen();
-    const read = await readRowsCooperatively<{ id: number }>(file, "t", WINDOW_SQL, [], reader.open, 64);
+    const rows = await readRowsCooperatively<{ id: number }>(file, "SELECT id, v FROM t ORDER BY id", [], reader.open);
     loop.stop();
-    expect(reader.readers.map((db) => [db.readonly, db.open, db.inTransaction])).toEqual([[true, false, false]]); // readonly, and closed before returning
-    expect(read.rows.map((r) => r.id)).toEqual(Array.from({ length: 500 }, (_, i) => i + 1));
-    expect(read.windows).toBe(8);
+    expect(reader.readers.map((db) => [db.readonly, db.open])).toEqual([[true, false]]); // readonly, and closed before returning
+    expect(rows.map((r) => r.id)).toEqual(Array.from({ length: 500 }, (_, i) => i + 1));
     expect(loop.count()).toBeGreaterThan(1);
   });
 
-  it("yields between windows even when the filter matches almost nothing", async () => {
-    const { file } = walTable(2000);
-    const loop = turns();
-    const read = await readRowsCooperatively<{ id: number }>(file, "t", "SELECT id FROM t WHERE id > ? AND id <= ? AND v LIKE ?", ['%"a":2000}%']);
-    loop.stop();
-    expect(read.rows).toEqual([{ id: 2000 }]);
-    expect(read.windows).toBe(8); // 2000 rowids, 256 per window
-    // A cursor that yields only between returned rows reaches its one match in a single native step, with no turn between.
-    expect(loop.count()).toBeGreaterThan(1);
-  });
-
-  it("reads an empty table as no rows, and steps over rowid gaps", async () => {
-    const { file, db } = walTable(0);
-    expect(await readRowsCooperatively(file, "t", WINDOW_SQL, [])).toMatchObject({ rows: [], windows: 0 });
-    db.prepare("INSERT INTO t(id, v) VALUES (1, 'a'), (1000, 'b')").run();
-    const read = await readRowsCooperatively<{ id: number }>(file, "t", WINDOW_SQL, []);
-    expect(read.rows.map((r) => r.id)).toEqual([1, 1000]);
-    expect(read.windows).toBe(4);
-  });
-
-  it("keeps one snapshot for every window while writers commit between windows; the next read sees them", async () => {
+  it("keeps one snapshot for the whole read while writers commit between slices; the next read sees them", async () => {
     const { file, db } = walTable(500);
-    // Runs at the reader's first yield, after its first window: every change lands in a window it has not read yet.
+    // Runs at the reader's first yield, after its cursor has started.
     setImmediate(() => db.transaction(() => {
-      db.prepare("UPDATE t SET v = 'changed' WHERE id = 500").run();
-      db.prepare("DELETE FROM t WHERE id = 400").run();
+      db.prepare("UPDATE t SET v = 'changed' WHERE id = 1").run();
       db.prepare("INSERT INTO t(id, v) VALUES (501, 'new')").run();
     })());
-    const during = await readRowsCooperatively<{ id: number; v: string }>(file, "t", WINDOW_SQL, [], undefined, 64);
-    expect(during.rows.map((r) => r.id)).toEqual(Array.from({ length: 500 }, (_, i) => i + 1));
-    expect(during.rows.at(-1)).toEqual({ id: 500, v: JSON.stringify({ a: 500 }) });
-    const after = await readRowsCooperatively<{ id: number; v: string }>(file, "t", WINDOW_SQL, []);
-    expect(after.rows).toHaveLength(500);
-    expect(after.rows.some((r) => r.id === 400)).toBe(false);
-    expect(after.rows.slice(-2)).toEqual([{ id: 500, v: "changed" }, { id: 501, v: "new" }]);
+    const during = await readRowsCooperatively<{ id: number; v: string }>(file, "SELECT id, v FROM t ORDER BY id", []);
+    expect(during).toHaveLength(500);
+    expect(during[0]!.v).toBe(JSON.stringify({ a: 1 }));
+    const after = await readRowsCooperatively<{ id: number; v: string }>(file, "SELECT id, v FROM t ORDER BY id", []);
+    expect(after).toHaveLength(501);
+    expect(after[0]!.v).toBe("changed");
   });
 
-  it("rethrows a native error raised in a later window, returns nothing, and leaves no reader pinning the WAL", async () => {
+  it("rethrows a native error raised mid-read, returns nothing, and leaves no reader pinning the WAL", async () => {
     const { file, db } = walTable(500, 300);
     const reader = recordingOpen();
-    const sql = "SELECT json_extract(v, '$.a') AS a FROM t WHERE id > ? AND id <= ? ORDER BY id";
-    const read = readRowsCooperatively(file, "t", sql, [], reader.open, 64);
+    const read = readRowsCooperatively(file, "SELECT json_extract(v, '$.a') AS a FROM t ORDER BY id", [], reader.open);
     await expect(read).rejects.toMatchObject({ name: "SqliteError", code: "SQLITE_ERROR" });
     expect(reader.readers.map((db) => db.open)).toEqual([false]);
     db.prepare("INSERT INTO t(id, v) VALUES (501, '{}')").run();
     expect(db.pragma("wal_checkpoint(TRUNCATE)")).toEqual([{ busy: 0, log: 0, checkpointed: 0 }]);
     db.prepare("DELETE FROM t WHERE id = 300").run();
-    expect((await readRowsCooperatively(file, "t", sql, [])).rows).toHaveLength(500);
+    expect(await readRowsCooperatively(file, "SELECT json_extract(v, '$.a') AS a FROM t ORDER BY id", [])).toHaveLength(500);
   });
 
   it("stays on the main handle inside a transaction and for memory or non-WAL databases", () => {
@@ -139,18 +110,12 @@ describe("the projector's execution queue", () => {
     migrate(db, [coreSchema, bindingsSessionsSchema, rigArchiveSchema, eventsSchema, queueItemSummarySchema, queueItemsSchema, queueTransitionsSchema, queueTransitionWakesSchema, viewsCustomSchema]);
     return db;
   }
-  const insertRow = (db: Database.Database, id: string, state: string, tags: string[] | null, body: string, claimedAt: string | null = "2026-10-25T11:00:00.000Z") =>
+  const insertRow = (db: Database.Database, id: string, state: string, tags: string[], body: string, claimedAt: string | null = "2026-10-25T11:00:00.000Z") =>
     db.prepare(`INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, tier, tags, body, claimed_at, last_heartbeat, blocked_on)
       VALUES (?, '2026-10-25T10:00:00.000Z', '2026-10-25T10:00:00.000Z', 'lead@r', 'builder@r', ?, 'normal', 'light', ?, ?, ?, NULL, ?)`)
-      .run(id, state, tags && JSON.stringify(tags), body, claimedAt, state === "blocked" ? "external:review" : null);
+      .run(id, state, JSON.stringify(tags), body, claimedAt, state === "blocked" ? "external:review" : null);
   const laneIds = (doc: Record<string, unknown>) => (doc.q1_lanes as Array<{ qitem_id: string }>).map((l) => l.qitem_id).sort();
   const tick = () => new Promise<void>((resolve) => setImmediate(resolve));
-  // The free-disk reading is taken live at read time (fs.statfsSync); every other field must match.
-  const withoutLiveDisk = (doc: unknown) => {
-    const copy = structuredClone(doc) as { q6_parallelism?: { df_margin?: { available_kib?: unknown } } };
-    if (copy.q6_parallelism?.df_margin) delete copy.q6_parallelism.df_margin.available_kib;
-    return copy;
-  };
 
   it("builds one execution at a time in arrival order, each reading fresh, and a failed build does not block the next", async () => {
     const db = fixture(tempFile("queue.sqlite"));
@@ -211,90 +176,14 @@ describe("the projector's execution queue", () => {
     seed(memory); seed(file);
     expect(cooperativeReadFile(memory)).toBeNull();
     expect(cooperativeReadFile(file)).not.toBeNull();
+    // The free-disk reading is taken live at read time (fs.statfsSync); every other field must match.
+    const withoutLiveDisk = (doc: unknown) => {
+      const copy = structuredClone(doc) as { q6_parallelism?: { df_margin?: { available_kib?: unknown } } };
+      if (copy.q6_parallelism?.df_margin) delete copy.q6_parallelism.df_margin.available_kib;
+      return copy;
+    };
     const legacy = await documentOf(memory);
     expect(withoutLiveDisk(await documentOf(file))).toEqual(withoutLiveDisk(legacy));
     expect((legacy as { q1_lanes: unknown[] }).q1_lanes.length).toBeGreaterThan(0);
   });
-
-  it("binds the same rows through the windowed read when the mission filter matches few of them, edge cases included", async () => {
-    const missionsRoot = path.dirname(tempFile("missions"));
-    fs.mkdirSync(path.join(missionsRoot, "release-9"));
-    fs.writeFileSync(path.join(missionsRoot, "release-9", "SPEC.md"), "---\nid: OPR_9\n---\n\n# Fixture mission\n");
-    const edges: Array<[id: string, state: string, tags: string[] | null, body: string]> = [
-      ["qitem-tagged", "in-progress", ["mission:release-9", "project:p1", "slice:S1"], "Tagged."],
-      ["qitem-alias-tag", "in-progress", ["mission:OPR_9", "project:p2", "slice:S2"], "Tagged with the mission's SPEC id."],
-      ["qitem-case", "in-progress", ["mission:RELEASE-9", "slice:S3"], "Tagged in another ASCII case."],
-      ["qitem-longer-tag", "in-progress", ["mission:release-90", "slice:S4"], "Another mission whose name contains this one."],
-      ["qitem-other-tag", "in-progress", ["mission:other", "slice:S5"], "Tagged elsewhere; mentions release-9."],
-      ["qitem-body", "in-progress", ["slice:S6", "project:p1"], "Untagged work for release-9."],
-      ["qitem-wildcard", "in-progress", ["slice:S7"], "Untagged; mentions OPRx9, which LIKE's _ matches."],
-      ["qitem-null-tags", "in-progress", null, "Slice: S8\nNo tags; mentions release-9."],
-      ["qitem-empty-body", "in-progress", ["mission:release-9", "slice:S9"], ""],
-      ["qitem-null-tags-empty-body", "in-progress", null, ""],
-      ["qitem-moved", "in-progress", ["mission:release-9", "slice:S10"], "Claimed, then moved three times."],
-    ];
-    // 700 unrelated rows around the edge rows, so matches are sparse and fall in different windows.
-    const seed = (db: Database.Database) => db.transaction(() => {
-      const fill = (from: number, to: number) => {
-        for (let i = from; i < to; i++) insertRow(db, `qitem-filler-${String(i).padStart(4, "0")}`, "done", ["slice:S1"], `Unrelated work ${i}.`);
-      };
-      fill(0, 300);
-      for (const [id, state, tags, body] of edges.slice(0, 4)) insertRow(db, id, state, tags, body);
-      fill(300, 600);
-      for (const [id, state, tags, body] of edges.slice(4)) insertRow(db, id, state, tags, body);
-      fill(600, 700);
-      for (let n = 0; n < 3; n++) {
-        db.prepare("INSERT INTO queue_transitions (qitem_id, ts, state, transition_note, actor_session) VALUES ('qitem-moved', ?, 'in-progress', 'moved', 'builder@r')")
-          .run(`2026-10-25T11:3${n}:00.000Z`);
-      }
-    })();
-    const memory = fixture(), file = fixture(tempFile("sparse.sqlite"));
-    seed(memory); seed(file);
-    expect(cooperativeReadFile(file)).not.toBeNull();
-    const deps = (db: Database.Database) => ({ db, now: () => NOW, slicesRoot: () => missionsRoot, rigsRoot: () => "/nonexistent-rigs-root",
-      exec: async (): Promise<string> => { throw new Error("not a repository"); } });
-    for (const project of [undefined, "p1"]) {
-      const legacy = await buildExecutionView(deps(memory), { mission: "release-9", project });
-      expect(withoutLiveDisk(await buildExecutionView(deps(file), { mission: "release-9", project }))).toEqual(withoutLiveDisk(legacy));
-      expect(laneIds(legacy)).toEqual(project ? ["qitem-body", "qitem-tagged"] : [
-        "qitem-alias-tag", "qitem-body", "qitem-case", "qitem-empty-body", "qitem-moved", "qitem-null-tags", "qitem-tagged", "qitem-wildcard",
-      ]);
-    }
-  });
-});
-
-describe("slow execution projections in the slow-operation log", () => {
-  afterEach(() => configureSyncSiteRecorder(undefined));
-
-  it("offers a diagnostic only at the slow threshold, and a throwing recorder changes nothing for the caller", () => {
-    const offered: unknown[] = [];
-    configureSyncSiteRecorder({ recordDiagnostic: (record) => { offered.push(record); return true; } });
-    recordSlowDiagnostic(249.9, { n: 1 });
-    recordSlowDiagnostic(250, { n: 2 });
-    recordSlowDiagnostic(Number.NaN, { n: 3 });
-    expect(offered).toEqual([{ n: 2 }]);
-    configureSyncSiteRecorder({ recordDiagnostic: () => { throw new Error("recorder broke"); } });
-    expect(() => recordSlowDiagnostic(300, { n: 4 })).not.toThrow();
-  });
-
-  it("records a slow projection's elapsed times and the candidate read's longest stretch, with no row content", async () => {
-    const offered: Record<string, unknown>[] = [];
-    configureSyncSiteRecorder({ recordDiagnostic: (record) => { offered.push(record); return true; } });
-    const db = track(createDb(tempFile("slow.sqlite")));
-    migrate(db, [coreSchema, bindingsSessionsSchema, rigArchiveSchema, eventsSchema, queueItemSummarySchema, queueItemsSchema, queueTransitionsSchema, queueTransitionWakesSchema, viewsCustomSchema]);
-    db.prepare(`INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, tier, tags, body)
-      VALUES ('qitem-slow', '2026-10-25T10:00:00.000Z', '2026-10-25T10:00:00.000Z', 'lead@r', 'builder@r', 'in-progress', 'normal', 'light', ?, ?)`)
-      .run(JSON.stringify(["slice:S1"]), "Private words.\nworktree_path=/nowhere/slow\n");
-    // Each Git probe of the row's worktree takes 260 ms, so the projection crosses the 250 ms threshold.
-    await buildExecutionView({ db, now: () => new Date("2026-10-25T12:00:00.000Z"), slicesRoot: () => null, rigsRoot: () => "/nonexistent-rigs-root",
-      exec: async (): Promise<string> => { await new Promise((resolve) => setTimeout(resolve, 260)); throw new Error("not a repository"); } });
-    expect(offered).toHaveLength(1);
-    const record = offered[0]!;
-    expect(record).toMatchObject({ schema: "openrig.execution-view/v1", candidateRead: "windowed", candidateRows: 1, candidateWindows: 1 });
-    expect(record.projectionMs).toBeGreaterThanOrEqual(250);
-    expect(record.candidateMs).toBeLessThanOrEqual(record.projectionMs as number);
-    expect(record.candidateLongestStretchMs).toBeLessThanOrEqual(record.candidateMs as number);
-    expect(Object.keys(record).sort()).toEqual(["candidateLongestStretchMs", "candidateMs", "candidateRead", "candidateRows", "candidateWindows", "projectionMs", "schema", "utc"]);
-    expect(JSON.stringify(record)).not.toMatch(/Private|nowhere|builder@r|S1/);
-  });
 });

packages/daemon/test/queue-sync-sites.test.ts

@@ -0,0 +1,69 @@
+import { afterEach, beforeEach, describe, expect, it } from "vitest";
+import type Database from "better-sqlite3";
+import { createDb } from "../src/db/connection.js";
+import { migrate } from "../src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository, createLinkedSuccessorLookup } from "../src/domain/queue-repository.js";
+import { configureSyncSiteRecorder, runSyncSiteWhenSlow } from "../src/domain/sync-site-wrap.js";
+
+// The queue's synchronous reads named in the slow-operation log, so an event-loop stall during one is attributed to it.
+// A fake recorder takes every call; the real recorder's slow threshold is pinned in slow-op-recorder.test.ts.
+describe("queue reads named in the slow-operation log", () => {
+  let db: Database.Database;
+  let repo: QueueRepository;
+  const sync: string[] = [];
+  const slow: Array<{ site: string; outcome: string }> = [];
+  beforeEach(() => {
+    db = createDb();
+    migrate(db, ALL_MIGRATIONS);
+    repo = new QueueRepository(db, new EventBus(db));
+    configureSyncSiteRecorder({
+      runSync<T>(site: string, fn: () => T): T { sync.push(site); return fn(); },
+      recordSlowSync(site: string, _durationMs: number, outcome: "ok" | "failed"): void { slow.push({ site, outcome }); },
+    });
+  });
+  afterEach(() => { configureSyncSiteRecorder(undefined); sync.length = 0; slow.length = 0; db.close(); });
+
+  it("names a handoff", async () => {
+    const source = await repo.create({ sourceSession: "planner@r", destinationSession: "relay@r", body: "work", nudge: false });
+    sync.length = 0;
+    const { created } = await repo.handoff({ qitemId: source.qitemId, fromSession: "relay@r", toSession: "worker@r", nudge: false });
+    expect(created.destinationSession).toBe("worker@r");
+    expect(sync).toContain("queue.handoff");
+  });
+
+  it("names decisionRevision's reply read through the slow-only site, once per revision", async () => {
+    const ask = await repo.create({ sourceSession: "planner@r", destinationSession: "relay@r", body: "work", nudge: false });
+    slow.length = 0;
+    const revision = repo.decisionRevision(repo.getById(ask.qitemId)!);
+    expect(typeof revision).toBe("string");
+    expect(slow).toEqual([{ site: "queue.decision_revision.replies", outcome: "ok" }]);
+  });
+
+  it("names the stuck sweep's linked-successor build when it is built", () => {
+    const lookup = createLinkedSuccessorLookup(db, ["qitem-source"]);
+    expect(sync).toEqual([]);
+    expect(lookup.has("qitem-source", "worker@r")).toBe(false);
+    expect(lookup.has("qitem-source", "other@r")).toBe(false);
+    expect(sync).toEqual(["queue.linked_successors.build"]);
+  });
+
+  it("keeps the wrapped call's result and error, and records each outcome", () => {
+    expect(runSyncSiteWhenSlow("test.site", () => 7)).toBe(7);
+    expect(() => runSyncSiteWhenSlow("test.site", () => { throw new Error("read failed"); })).toThrow("read failed");
+    expect(slow).toEqual([{ site: "test.site", outcome: "ok" }, { site: "test.site", outcome: "failed" }]);
+  });
+
+  it("never lets a failing record replace the wrapped call's result", () => {
+    configureSyncSiteRecorder({ recordSlowSync(): void { throw new Error("recorder gone"); } });
+    expect(runSyncSiteWhenSlow("test.site", () => "value")).toBe("value");
+    expect(() => runSyncSiteWhenSlow("test.site", () => { throw new Error("read failed"); })).toThrow("read failed");
+  });
+
+  it("runs the call bare when no recorder is configured", () => {
+    configureSyncSiteRecorder(undefined);
+    expect(runSyncSiteWhenSlow("test.site", () => "bare")).toBe("bare");
+    expect(slow).toEqual([]);
+  });
+});

packages/daemon/test/slow-op-recorder.test.ts

@@ -21,6 +21,7 @@ const expectedSites = new Map<string, string[]>([
   ["domain/codex-thread-id.ts", ["codex_thread_id.resolve_home"]],
   ["domain/review/gather.ts", ["review.gather.git"]],
   ["domain/tmux-option-defaults.ts", ["tmux_options.command_v"]],
+  ["domain/queue-repository.ts", ["queue.decision_revision.replies", "queue.linked_successors.build", "queue.handoff"]],
 ]);
 
 const pluginVendorUrl = "https://github.com/mvschwarz/openrig-plugins/releases/latest/download/openrig-core.tar.gz";
@@ -290,6 +291,23 @@ describe("SlowOpRecorder locked instrumentation contract", () => {
     await recorder.close();
   });
 
+  it("records a slow synchronous site only at or above the slow threshold, with its outcome", async () => {
+    const mod = await loadRecorderModule();
+    expect(mod, "slow-op-recorder production module is missing").not.toBeNull();
+    if (!mod) return;
+
+    const dir = tempDir();
+    const logPath = path.join(dir, "slow-operations.jsonl");
+    const recorder = new mod.SlowOpRecorder({ logPath, slowThresholdMs: 250 });
+    recorder.recordSlowSync("queue.decision_revision.replies", 249, "ok");
+    recorder.recordSlowSync("queue.decision_revision.replies", 250, "ok");
+    recorder.recordSlowSync("queue.decision_revision.replies", 400, "failed");
+    await recorder.flush();
+    const records = (await readRecords(logPath)).filter((record) => record.site === "queue.decision_revision.replies");
+    expect(records.map((record) => [record.phase, record.durationMs, record.outcome])).toEqual([["end", 250, "ok"], ["end", 400, "failed"]]);
+    await recorder.close();
+  });
+
   // The request-timing MIDDLEWARE CONTRACT, hermetic + deterministic. The prior "real server" form
   // ran the full daemon (event-loop monitor + a globalThis.fetch monkey-patch), which is structurally
   // ambient-fragile under the concurrent-workspace gate: it intermittently returned 200 for an unmatched

packages/daemon/test/tmux-adapter.test.ts

@@ -51,6 +51,47 @@ describe("TmuxAdapter.listSessionClientTerminals", () => {
   });
 });
 
+describe("TmuxAdapter.lastClientActivity (#3076)", () => {
+  const activity = (stdout: string) => adapterFor(mockExec({ "list-clients": { stdout } })).lastClientActivity("op@kernel");
+
+  it("is the latest activity of a writable client, in milliseconds", async () => {
+    expect(await activity("0|1790000100\n0|1790000105\n")).toBe(1790000105000);
+  });
+
+  it("does not count a read-only client (a viewer tile)", async () => {
+    expect(await activity("1|1790000200\n0|1790000100\n")).toBe(1790000100000);
+    expect(await activity("1|1790000200\n")).toBeNull();
+  });
+
+  it("counts an attach as recent activity: a key in the same second as the attach can't be told apart from it", async () => {
+    // tmux would report client_created == client_activity for both an attach alone and an attach plus a key that second.
+    expect(await activity("0|1790000300\n")).toBe(1790000300000);
+  });
+
+  it("is null with no client", async () => {
+    expect(await activity("")).toBeNull();
+    expect(await activity("\n")).toBeNull();
+  });
+
+  it("throws on a row it can't read, alone or next to a readable one, so the guard holds", async () => {
+    for (const stdout of ["0|abc\n", "0|0\n", "0\n", "2|1790000100\n", "0|1790000000|1790000100\n", "0|1790000100\n0|abc\n"]) {
+      await expect(activity(stdout)).rejects.toThrow();
+    }
+  });
+
+  it("throws when tmux can't say", async () => {
+    for (const error of [new Error("can't find session: op@kernel"), NO_SERVER_ERROR, new Error("command timed out")]) {
+      await expect(adapterFor(mockExec({ "list-clients": { error } })).lastClientActivity("op@kernel")).rejects.toThrow();
+    }
+  });
+
+  it("asks tmux for the read-only flag and the activity of each client of the session", async () => {
+    const commands: string[] = [];
+    await adapterFor(async command => { commands.push(command); return ""; }).lastClientActivity("op@kernel");
+    expect(commands.some(c => c.includes("list-clients") && c.includes("#{client_readonly}|#{client_activity}"))).toBe(true);
+  });
+});
+
 describe("TmuxAdapter", () => {
   it("starts only an empty terminal server and reconciles repeated requests", async () => {
     let live = false;

packages/daemon/test/tmux-argv-legacy-equivalence.test.ts

@@ -158,6 +158,7 @@ async function driveEveryMethod(): Promise<RunCall[]> {
   await a.readPaneLastActivity(SESSION);
   await a.listClients();
   await a.listSessionClientTerminals(SESSION);
+  await a.lastClientActivity(SESSION);  // #3076: the client activity read right before an automatic paste
   await a.switchClient("/dev/ttys001", SESSION);
   // A runtime switch's exchange: two session-id reads, then the guarded if-shell swap (nested quoting); and its env writes.
   await a.swapRuntimeSwitchPanes({ nodeId: "n1", stagingSession: "staging'q a", canonicalSession: SESSION, canonicalPane: "%1", stagedPane: "%1" });

packages/tui/src/connected-assistants.ts

@@ -10,7 +10,7 @@ export function toConnectedAssistants(body: unknown): ConnectedAssistant[] | nul
   if (!Array.isArray(members)) return null;
   return members.flatMap((member: Record<string, unknown>) => {
     if (typeof member?.seat !== "string" || !member.seat) return [];
-    const known = typeof member.status === "string" && member.status in CHECK_IN_DISPLAY;
+    const known = typeof member.status === "string" && Object.hasOwn(CHECK_IN_DISPLAY, member.status);
     return [{ seat: member.seat, checkIn: {
       status: known ? member.status as CheckInStatus : "unknown",
       lastCheckInAt: typeof member.lastCheckInAt === "string" ? member.lastCheckInAt : null,

packages/tui/src/render.ts

@@ -958,7 +958,11 @@ function contentLines(state: ViewState, snap: FleetSnapshot, contentWidth: numbe
       { text: `Its seats aren't read in this view: rig ps --host ${rig.remoteHost} --nodes --rig ${rig.name}` },
     ];
     if (rig.inventoryNotLoaded) return [{ text: `Reading ${rig.name}…` }];
-    if (rig.inventoryUnavailable) return [{ text: `Inventory unavailable for ${rig.name} · refresh to Retry` }];
+    if (rig.inventoryUnavailable) return [
+      { text: `Inventory unavailable for ${rig.name} · refresh to Retry` },
+      ...(leaf?.kind === "pod" ? [] : connectedAssistantLines(host.connectedAssistants, host.connectedAssistantsUnavailable,
+        rig.name, state.filter, contentWidth, state.timeZone)),
+    ];
     const podFilter = leaf?.kind === "pod" ? leaf.name : null;
     const all = rig.pods.flatMap((p) => p.agents.map((a) => ({ pod: p.name, ...a })));
     const rows = all

packages/tui/test/hydrate.test.ts

@@ -782,6 +782,34 @@ describe("snapshot hydration over the §4.A reads (Phase 2)", () => {
     expect(instanceLines).not.toContain("not shown");
   });
 
+  it("shows inherited-property check-in statuses as unknown", async () => {
+    const hosted = { members: ["constructor", "toString"].map((status, i) => ({
+      seat: `guest${i}@myrig`, status, reason: "unrecognized reason",
+    })) };
+    const snap = await hydrateSnapshot(fixtureClient({}, { "/api/queue/hosted-members": hosted }));
+    const view = createViewState({ instanceId: "t", getSnapshot: () => snap });
+    view.dispatch({ type: "drill", resource: "rig", name: "myrig" });
+    const lines = renderScreen(view.get(), snap, { cols: 160, rows: 60 }).lines.join("\n");
+    expect(lines).toContain("guest0@myrig  unknown");
+    expect(lines).toContain("guest1@myrig  unknown");
+    expect(lines).not.toContain("unrecognized reason");
+  });
+
+  it("shows connected check-ins beside a failed rig nodes read", async () => {
+    const hosted = { members: [
+      { seat: "chat@myrig", status: "on-request", lastCheckInAt: null },
+      { seat: "other@downrig", status: "checking-in", lastCheckInAt: null },
+    ] };
+    const snap = await hydrateSnapshot(fixtureClient({ "/api/rigs/01JRIG/nodes": { status: 500 } },
+      { "/api/queue/hosted-members": hosted }));
+    const view = createViewState({ instanceId: "t", getSnapshot: () => snap });
+    view.dispatch({ type: "drill", resource: "rig", name: "myrig" });
+    const lines = renderScreen(view.get(), snap, { cols: 160, rows: 60 }).lines.join("\n");
+    expect(lines).toContain("Inventory unavailable for myrig");
+    expect(lines).toContain("chat@myrig  on request:");
+    expect(lines).not.toContain("other@downrig");
+  });
+
   it("a known-empty list of connected assistants stays quiet", async () => {
     const snap = await hydrateSnapshot(fixtureClient());
     expect(snap.hosts[0]!.connectedAssistants).toEqual([]);