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/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

@@ -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

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/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([]);