fix(tui): disclose capped queue samples through topology and Pulse

packages/tui/src/hydrate.ts

@@ -755,6 +755,11 @@ export async function hydrateSnapshot(
         }
       : {}),
     attention: (attention ?? []).map(toQueueRead),
+    queueCoverage: !seatQueueReads ? undefined
+      : [attention, blocked, inProgress, pending].some(rows => rows === null) ? "unavailable"
+      // Existing /list defaults cap the first three reads at 100; pending requests 50.
+      : [attention, blocked, inProgress].some(rows => rows!.length >= 100) || pending!.length >= 50 ? "partial"
+      : "complete",
     blocked: blockedResolved,
     inProgress: (inProgress ?? []).map(toQueueRead),
     seatActivity,

packages/tui/src/pulse/pulse-model.ts

@@ -57,6 +57,7 @@ export interface PulseLane {
 }
 
 export interface PulseModel {
+  queueSampleNote?: string;
   exceptions: PulseExceptionSection[]; // EMPTY section omitted entirely (empty-strip-is-silence)
   lanes: [PulseLane, PulseLane, PulseLane]; // NOW, JUST FINISHED, UP NEXT
   footer: { active: number; parked: number; waitingYou: number; updatedAgo: string };
@@ -369,10 +370,11 @@ export function buildPulseModel(snap: FleetSnapshot, nowMs: number = Date.now())
   // idle-duration derived at the renderer. Zero parked = silence (OMITTED), like
   // the other live joins — the incr-2 "read pending" placeholder is retired.
   const seatBySession = new Map(snap.seatActivity.map((s) => [s.session, s]));
-  const parked = parkedRows(snap.inProgress, seatBySession, nowMs);
+  const attentionIds = new Set(snap.attention.map(row => row.qitemId));
+  const parked = parkedRows(snap.inProgress.filter(row => !attentionIds.has(row.qitemId)), seatBySession, nowMs);
   if (parked.length > 0) exceptions.push({ glyph: "◌", token: "warn", label: "PARKED WITH BATON", rows: parked });
 
-  const blocked = blockedRows(snap.blocked, nowMs);
+  const blocked = blockedRows(snap.blocked.filter(row => !attentionIds.has(row.qitemId)), nowMs);
   if (blocked.length > 0) exceptions.push({ glyph: "⧗", token: PULSE_INFO_TOKEN, label: "BLOCKED ON AGENTS", rows: blocked });
 
   const now = nowRows(snap.seatActivity, snap.inProgress, snap);
@@ -391,5 +393,8 @@ export function buildPulseModel(snap: FleetSnapshot, nowMs: number = Date.now())
   const ago = ageLabel(snap.hydratedAt, nowMs);
   const footer = { active: now.length, parked: parkedCount, waitingYou, updatedAgo: ago ? `${ago} ago` : "—" };
 
-  return { exceptions, lanes, footer };
+  return { exceptions, lanes, footer,
+    ...(snap.queueCoverage === "partial" || snap.queueCoverage === "unavailable"
+      ? { queueSampleNote: "Queue counts are samples; unlisted work is unknown." } : {}),
+  };
 }

packages/tui/src/pulse/render-pulse.ts

@@ -149,7 +149,7 @@ export interface PulseLaneTarget {
 export function pulseLaneTargets(model: PulseModel): PulseLaneTarget[] {
   const exceptionLines = model.exceptions.reduce((n, s) => n + renderExceptionSection(s).length, 0);
   // renderPulseView layout: [tabStrip, blank] + exceptions + [blank, laneRule] + laneRows + footer
-  const laneStart = 2 + exceptionLines + 2;
+  const laneStart = 2 + (model.queueSampleNote ? 1 : 0) + exceptionLines + 2;
   const g = LANE_GUTTER.length;
   const starts = [0, COL[0] + g, COL[0] + g + COL[1] + g]; // text col where each lane's cell begins
   const widths = [COL[0], COL[1], UP_NEXT_WIDTH];
@@ -170,6 +170,7 @@ export function renderPulseFooter(f: PulseModel["footer"]): Line {
 /** The whole PULSE view (increment 1: static from demoPulseModel). REUSABLE parts above. */
 export function renderPulseView(model: PulseModel): Line[] {
   const out: Line[] = [renderPulseTabStrip(), { text: "" }];
+  if (model.queueSampleNote) out.push(line([{ text: model.queueSampleNote, token: "dim" }]));
   // EMPTY section is omitted entirely by the model (empty-strip-is-silence); we
   // render exactly the sections present.
   for (const section of model.exceptions) out.push(...renderExceptionSection(section));

packages/tui/src/render.ts

@@ -166,7 +166,7 @@ function queueFacts(snap: FleetSnapshot, session: string | null | undefined): {
     rows.push({ role, row });
   }
   const primary = rows[0];
-  if (!primary) return { count: 0, work: "—", now: "—" };
+  if (!primary) return { count: 0, work: queueIncomplete(snap) ? "unknown" : "—", now: queueIncomplete(snap) ? "not in sample" : "—" };
   const slice = primary.row.tags?.find((tag) => tag.startsWith("slice:"))?.slice("slice:".length);
   const work = slice?.match(/^OPR(?:\.\d+){3}\.(\d+)$/)?.[1]
     ? `S${slice.slice(slice.lastIndexOf(".") + 1)}`
@@ -176,6 +176,21 @@ function queueFacts(snap: FleetSnapshot, session: string | null | undefined): {
   return { count: rows.length, work, now };
 }
 
+function queueIncomplete(snap: FleetSnapshot): boolean {
+  return snap.queueCoverage === "partial" || snap.queueCoverage === "unavailable";
+}
+
+function queueCount(snap: FleetSnapshot, count: number): string | number {
+  return queueIncomplete(snap) ? count ? `${count}+` : "?" : count || "·";
+}
+
+function queueSampleLines(snap: FleetSnapshot, width: number): ContentLine[] {
+  return queueIncomplete(snap) ? wrapDetailLines([
+    { text: snap.queueCoverage === "unavailable" ? "Queue sample incomplete: a list read failed." : "Queue sample may be incomplete: a list reached its cap." },
+    { text: "Counts are lower bounds; unlisted work is unknown." },
+  ], width) : [];
+}
+
 function number(value: number | null | undefined): string {
   return value == null ? "—" : value.toLocaleString("en-US");
 }
@@ -214,8 +229,8 @@ function rowsForAgent(snap: FleetSnapshot, session: string, sources: Array<Fleet
   return rows;
 }
 
-function workRows(rows: FleetSnapshot["attention"], width: number): ContentLine[] {
-  if (rows.length === 0) return [fieldLine({ label: "rows", value: "none in the served bounded lists" })];
+function workRows(rows: FleetSnapshot["attention"], width: number, incomplete = false): ContentLine[] {
+  if (rows.length === 0) return [fieldLine({ label: "rows", value: incomplete ? "not in sample; work unknown" : "none in the served bounded lists" })];
   return rows.flatMap((row) => {
     const summary = row.summary?.trim() || row.body.split("\n").find((line) => line.trim())?.trim() || "no summary served";
     return [
@@ -239,7 +254,7 @@ function agentDetailLines(
   const pendingRows = session ? rowsForAgent(snap, session, [snap.pending]) : [];
   const recentRows = session ? rowsForAgent(snap, session, [snap.recentlyFinished]) : [];
   const needs = session ? snap.needs.filter((item) => item.target === session) : [];
-  const visibleAssigned = currentRows.length + pendingRows.length;
+  const visibleAssigned = new Set([...currentRows, ...pendingRows].map(row => row.qitemId)).size;
   const assigned = agent.assignedWorkCount ?? visibleAssigned;
   const pending = agent.pendingWorkCount ?? pendingRows.length;
   const inProgress = agent.inProgressWorkCount ?? currentRows.filter((row) => row.state === "in-progress").length;
@@ -300,15 +315,16 @@ function agentDetailLines(
           local: found.host === snap.hosts[0],
         }, contentWidth),
       },
-      { title: `CURRENT WORK · ${currentRows.length}`, lines: workRows(currentRows, contentWidth) },
+      { title: `CURRENT WORK · ${currentRows.length}${queueIncomplete(snap) ? " sampled" : ""}`, lines: workRows(currentRows, contentWidth, queueIncomplete(snap)) },
       {
         title: "QUEUE",
         lines: [
-          ...wrapDetailValue("depth", `${assigned} assigned · ${pending} pending · ${inProgress} in progress · ${blocked} blocked`, contentWidth),
+          ...queueSampleLines(snap, contentWidth),
+          ...wrapDetailValue("depth", `${agent.assignedWorkCount == null && queueIncomplete(snap) ? `${assigned}+` : assigned} assigned · ${agent.pendingWorkCount == null && queueIncomplete(snap) ? `${pending}+` : pending} pending · ${agent.inProgressWorkCount == null && queueIncomplete(snap) ? `${inProgress}+` : inProgress} in progress · ${agent.blockedWorkCount == null && queueIncomplete(snap) ? `${blocked}+` : blocked} blocked`, contentWidth),
           ...(agent.assignedWorkCount == null ? wrapDetailValue("basis", `${visibleAssigned} rows visible in bounded list reads; complete count not served`, contentWidth) : []),
         ],
       },
-      { title: `UP NEXT · ${pending}`, lines: workRows(pendingRows, contentWidth) },
+      { title: `UP NEXT · ${pending}${queueIncomplete(snap) ? " · sampled rows" : ""}`, lines: workRows(pendingRows, contentWidth, queueIncomplete(snap)) },
       { title: `NEEDS YOU · ${needs.length || agent.activity?.needsInput?.count || 0}`, lines: needsRows },
       ...(recentRows.length ? [{ title: "RECENTLY FINISHED · bounded window", lines: workRows(recentRows, contentWidth) }] : []),
       {
@@ -636,6 +652,7 @@ function instanceContentLines(
   }
 
   lines.push(healthSummaryLine(snap, scope, contentWidth));
+  lines.push(...queueSampleLines(snap, contentWidth));
   lines.push({ text: state.filter ? `/ filter instance rows: ${state.filter} · / replace · esc clear` : "/ filter instance rows…" });
   const columns = instanceAgentColumns(contentWidth);
   lines.push({ text: tableRow(columns, { rig: "RIG", pod: "POD", seat: "SEAT", runtime: "RT", context: "CTX", status: "STATE", queue: "Q", work: "WORK", now: "NOW" }) });
@@ -675,7 +692,7 @@ function instanceContentLines(
           runtime: runtimeShort(agent.runtime),
           context: contextCompact(agent.context, contentWidth < 110),
           status: stateCell.word,
-          queue: queue.count || "·",
+          queue: queueCount(snap, queue.count),
           work: queue.work,
           now: queue.now,
         }),
@@ -684,7 +701,7 @@ function instanceContentLines(
       });
     }
   }
-  lines.push({ text: "" }, { text: host.rigs.some(r => r.inventoryUnavailable) ? `${host.rigs.length} rigs · inventory incomplete · ${seatCount} seats read` : `${host.rigs.length} rigs · ${seatCount} seats · ${workingCount} working · ${attentionCount} need attention · ${openCount} open rows` });
+  lines.push({ text: "" }, { text: host.rigs.some(r => r.inventoryUnavailable) ? `${host.rigs.length} rigs · inventory incomplete · ${seatCount} seats read` : `${host.rigs.length} rigs · ${seatCount} seats · ${workingCount} working · ${attentionCount} need attention · ${openCount}${queueIncomplete(snap) ? "+ sampled" : ""} open rows` });
   lines.push(...recentLines(snap, scope, contentWidth, false, state.timeZone));
   return lines;
 }
@@ -853,6 +870,7 @@ function contentLines(state: ViewState, snap: FleetSnapshot, contentWidth: numbe
       return lines;
     }
     const agentCols = agentColumns(contentWidth);
+    lines.push(...queueSampleLines(snap, contentWidth));
     const narrowFactory = !agentCols.some(([key]) => key === "model");
     if (narrowFactory) lines.push({ text: "MODEL/NOW/ACTIONS on drill (enter)" });
     lines.push({
@@ -898,7 +916,7 @@ function contentLines(state: ViewState, snap: FleetSnapshot, contentWidth: numbe
           model: tableModel(a.model),
           context: contextCompact(a.context, narrowFactory),
           status: stateCell.word,
-          queue: queue.count || "·",
+          queue: queueCount(snap, queue.count),
           work: queue.work,
           now: queue.now,
           actions: actionsCell,
@@ -910,7 +928,7 @@ function contentLines(state: ViewState, snap: FleetSnapshot, contentWidth: numbe
     lines.push({ text: "" });
     const working = rows.filter((agent) => ["active", "working", "running"].includes(agent.status)).length;
     const attention = rows.filter((agent) => /attention|needs|blocked|unknown|failed/.test(agent.status)).length;
-    lines.push({ text: `${rows.length} seats · ${working} working · ${attention} need attention · ${rows.reduce((n, agent) => n + queueFacts(snap, agent.session).count, 0)} open rows` });
+    lines.push({ text: `${rows.length} seats · ${working} working · ${attention} need attention · ${rows.reduce((n, agent) => n + queueFacts(snap, agent.session).count, 0)}${queueIncomplete(snap) ? "+ sampled" : ""} open rows` });
     if (!podFilter) lines.push(...recentLines(snap, { kind: "rig", rig: rig.name }, contentWidth, false, state.timeZone));
     return lines;
   }

packages/tui/src/types.ts

@@ -394,6 +394,9 @@ export interface FleetSnapshot {
   /** PULSE ▲ NEEDS YOU source — the shipped attention read (already the exact
    * human-facing set); empty array = the join ran and yielded zero (silence). */
   attention: QueueRead[];
+  /** Coverage of the four active queue lists, before filtering by rig/pod/seat.
+   * A cap is possible truncation; it does not prove an omitted row. */
+  queueCoverage?: "complete" | "partial" | "unavailable";
   /** PULSE ⧗ BLOCKED ON AGENTS source — the shipped state=blocked read (ALL
    * blocked qitems); the render filters to NON-human blockedOn. */
   blocked: QueueRead[];