feat(telemetry): 5b — structural ACTIVITY signal so rig ps ACTIVITY means something

Default `rig ps --nodes` ACTIVITY was 100% hook-based: attachAgentActivity emitted
unknown/no_runtime_hook for hook-less/stale-hook seats, so live seats sat at empty prompts
with nothing in the control plane. Lift the EXISTING structural pane classifier
(session-transport classifyPaneActivity — spinner shapes / esc-to-interrupt / idle-prompt,
never a verb allowlist, so a "Drizzling"-style gerund reads as motion) into ACTIVITY:

- NEW SeatStructuralActivityService: captures + structurally classifies each running tmux
  seat's pane on the 1Hz cadence and CACHES it per session. The request path stays
  CAPTURE-FREE (reads the cache) — the OPR.0.4.3 healthz-wedge storm fix is preserved.
- attachAgentActivity precedence fold: fresh POSITIVE hook wins -> cached structural verdict
  (evidenceSource pane_heuristic, sampledAt = the liveness observation time) -> honest
  stale/unknown hook -> unknown/no_runtime_hook. Structural liveness OVERRIDES an absent or
  stale hook (constraint 2 — liveness beats hook-arrival age).
- Constraint 1 (turn-boundary): idle != quiet; the finding-derivation (baton.dropped / idle-
  gate) already requires held work, so the ACTIVITY column honestly shows idle/running.

Background-path safety (guard MF1/MF2):
- Freshness + invalidation: getStructuralActivity age-gates (evicts past a bounded window)
  and pollSeat invalidates the row on a null/failed capture — a stale positive verdict never
  survives a capture outage or same-name occupant transition.
- Single-flight, held until settle: sweeps never overlap (a stuck tmux can't accumulate
  whole-fleet captures). Deliberately NOT time-out-to-release: execCommand has no abort, so a
  wrapper timeout would release single-flight while the child process lives = a process-level
  storm. A permanently-stuck tmux degrades to no-signal via age-expiry (honest). A kill-backed
  capture primitive is a follow-on.

Wired production startup -> server ctx -> sessions/rigs routes (mirrors seatActivityService),
started post-bind in index.ts. Follow-on: the UI "N days quiet" label (activity-visuals.ts,
off eventAt) is the same arrival-age miscalibration on a separate UI surface.

Reviewed-by: dev-guard@v-openrig-build (accept-to-commit; 13-file tree on 560387c8e; MF1/MF2/HIGH3 close at source; 138/138 + tsc 0 + diff-check clean; CLEAR sha256 9034991f)
Co-Authored-By: Claude Opus 4.8 <[email protected]>

packages/daemon/src/domain/node-inventory.ts

@@ -9,7 +9,8 @@ import type { TranscriptStore } from "./transcript-store.js";
 import type { AgentActivityStore } from "./agent-activity-store.js";
 import type { SeatActivityService } from "./seat-activity-service.js";
 import type { TmuxAdapter } from "../adapters/tmux.js";
-import { probeSessionActivity } from "./session-transport.js";
+import { probeSessionActivity, mapPaneState } from "./session-transport.js";
+import type { StructuralObservation } from "./seat-structural-activity-service.js";
 import { findLatestUsableSnapshot, findLatestUsableSnapshotsForAllRigs } from "./rig-repository.js";
 import { resolveNodeWorkspace } from "./workspace/workspace-resolver.js";
 import { deriveCanonicalSessionName } from "./session-name.js";
@@ -1061,6 +1062,10 @@ export async function attachAgentActivity(
   deps: {
     tmuxAdapter: TmuxAdapter;
     activityStore?: AgentActivityStore;
+    /** 5b82324b — the cached STRUCTURAL pane observation (SeatStructuralActivityService). READ-only
+     *  and capture-FREE: lifts structural motion into the ACTIVITY signal on the DEFAULT path so a
+     *  live hook-less/stale-hook seat stops rendering as `unknown`. Absent = the pre-5b behavior. */
+    structuralActivity?: { getStructuralActivity(sessionName: string): StructuralObservation | null };
     now?: Date;
     // OPR.0.4.3 healthz-wedge amplification fix: cheap by default. The per-node
     // tmux `capturePaneContent` fallback (probeSessionActivity) is the storm that
@@ -1084,6 +1089,39 @@ export async function attachAgentActivity(
       sessionName: entry.canonicalSessionName,
       now: sampledAt,
     });
+    // A fresh POSITIVE hook (running/needs_input/idle) is authoritative — use it directly.
+    if (hookActivity && hookActivity.state !== "unknown") {
+      return {
+        ...entry,
+        agentActivity: hookActivity,
+      };
+    }
+
+    // Hook ABSENT or unknown/stale → consult the CACHED structural observation. This read is
+    // capture-FREE: the pane capture already happened on the background SeatStructuralActivityService
+    // tick, never here (the healthz-wedge no-per-request-capture invariant is preserved). A structural
+    // motion verdict is a LIVENESS signal that OVERRIDES an absent/stale hook (constraint 2 — liveness
+    // beats hook-arrival age) and makes ACTIVITY real on the DEFAULT path for hook-less / Codex /
+    // just-finished-a-turn seats: the exact fleet-blindness this fix targets. Structural discrimination
+    // is STRUCTURAL (spinner shapes / esc-to-interrupt / idle prompt), never a verb allowlist.
+    const structural = entry.canonicalSessionName
+      ? deps.structuralActivity?.getStructuralActivity(entry.canonicalSessionName)
+      : null;
+    if (structural && structural.state !== "unknown") {
+      return {
+        ...entry,
+        agentActivity: {
+          state: mapPaneState(structural.state),
+          reason: structural.reason,
+          evidenceSource: "pane_heuristic",
+          sampledAt: structural.observedAt,
+          evidence: structural.evidence,
+        },
+      };
+    }
+
+    // No positive hook and no structural motion. A stale/unknown hook, if one exists, is delivered
+    // HONESTLY as-is (unknown/stale) — never upgraded to a quiet-seat verdict on arrival age alone.
     if (hookActivity) {
       return {
         ...entry,

packages/daemon/src/domain/seat-structural-activity-service.ts

@@ -0,0 +1,133 @@
+import type Database from "better-sqlite3";
+import type { TmuxAdapter } from "../adapters/tmux.js";
+import { classifyPaneActivity, type PaneActivityClassification } from "./session-transport.js";
+
+/** A cached STRUCTURAL pane observation: the classifyPaneActivity verdict plus WHEN the pane was read
+ *  as motion. observedAt is a LIVENESS timestamp (last time we saw the pane), NOT a hook-arrival age —
+ *  that distinction is the whole point of constraint 2 (a real counter is not a real liveness verdict). */
+export interface StructuralObservation {
+  state: PaneActivityClassification["state"];
+  reason: string;
+  evidence: string | null;
+  observedAt: string;
+}
+
+export const DEFAULT_STRUCTURAL_POLL_INTERVAL_MS = 1000;
+// A cached observation is authoritative only while CURRENT. Past this window with no fresh capture — a
+// tmux/capture outage, a stalled poller, or a same-name occupant transition — the READ refuses it and
+// evicts it, so a stale positive verdict can never masquerade as liveness (MUST-FIX 1). 5× the poll
+// tolerates a few missed ticks; a persistently failing/stuck poller ages the row out and the ACTIVITY
+// projection falls back to the honest hook/unknown state.
+export const DEFAULT_STRUCTURAL_STALE_MS = 5000;
+
+/**
+ * 5b82324b — the STRUCTURAL activity cache. Sibling of SeatActivityService (which reads ONLY the tmux
+ * window_activity TIMESTAMP and is deliberately text-blind): this service captures pane TEXT once per
+ * running tmux seat per tick and classifies it STRUCTURALLY via classifyPaneActivity (spinner shapes,
+ * `esc to interrupt`, idle-prompt / status-bar signatures — never a verb allowlist, so a "Drizzling"-
+ * style spinner reads as motion instead of a false park). The cache is READ (capture-FREE) by
+ * attachAgentActivity so the `rig ps` ACTIVITY column reflects real pane motion for hook-less / stale-
+ * hook / turn-boundary seats WITHOUT reintroducing the per-request capture storm the healthz-wedge
+ * cheap-default removed. Two safety rules make the BACKGROUND path itself safe:
+ *   (MF1) a read refuses a stale observation and a failed capture invalidates the prior row — a stale
+ *         positive verdict never survives a capture outage or a same-name occupant transition; and
+ *   (MF2) sweeps are SINGLE-FLIGHT and HELD until the real captures settle — a slow/stuck tmux can
+ *         never accumulate overlapping whole-fleet captures (only one sweep's captures are ever in
+ *         flight). We deliberately do NOT time-out a capture to release the guard: execCommand has no
+ *         AbortSignal, so a wrapper timeout would release single-flight while the child process lives,
+ *         letting a later sweep spawn MORE children — the storm at the process level. A permanently
+ *         stuck tmux therefore degrades to "no structural signal" (honest, via MF1 age-expiry), never
+ *         a storm. A kill-backed capture primitive (bounded abort) is a tmux-adapter follow-on.
+ */
+export class SeatStructuralActivityService {
+  private readonly latestBySession = new Map<string, StructuralObservation>();
+  private timer: ReturnType<typeof setInterval> | null = null;
+  private sweeping = false; // single-flight guard: one whole-fleet sweep at a time (MUST-FIX 2)
+
+  constructor(
+    private readonly tmuxAdapter: Pick<TmuxAdapter, "capturePaneContent">,
+    private readonly now: () => Date = () => new Date(),
+    private readonly captureLines: number = 20,
+    private readonly staleAfterMs: number = DEFAULT_STRUCTURAL_STALE_MS,
+  ) {}
+
+  /** The cached structural observation — ONLY while current (capture-FREE read). Returns null AND evicts
+   *  once the observation is older than the freshness window, so a stalled/failed poller never leaves a
+   *  stale positive verdict authoritative over an honest stale hook (MUST-FIX 1). */
+  getStructuralActivity(sessionName: string): StructuralObservation | null {
+    const obs = this.latestBySession.get(sessionName);
+    if (!obs) return null;
+    if (this.now().getTime() - Date.parse(obs.observedAt) >= this.staleAfterMs) {
+      this.latestBySession.delete(sessionName);
+      return null;
+    }
+    return obs;
+  }
+
+  /** Capture + structurally classify one seat's pane, caching the observation keyed by session name. A
+   *  null or failed capture INVALIDATES the prior row (never leaves a stale positive verdict) and
+   *  returns null (MUST-FIX 1). */
+  async pollSeat(sessionName: string): Promise<StructuralObservation | null> {
+    let content: string | null;
+    try {
+      content = await this.tmuxAdapter.capturePaneContent(sessionName, this.captureLines);
+    } catch {
+      this.latestBySession.delete(sessionName);
+      return null;
+    }
+    if (content === null) {
+      this.latestBySession.delete(sessionName);
+      return null;
+    }
+    const c = classifyPaneActivity(content);
+    const obs: StructuralObservation = {
+      state: c.state,
+      reason: c.reason,
+      evidence: c.evidence,
+      observedAt: this.now().toISOString(),
+    };
+    this.latestBySession.set(sessionName, obs);
+    return obs;
+  }
+
+  /** Refresh every running tmux-bound seat once. SINGLE-FLIGHT and HELD until the captures settle: a new
+   *  sweep never starts while one is in flight (MUST-FIX 2), so a slow/stuck tmux can never accumulate
+   *  overlapping whole-fleet captures — at most one sweep's worth (N seats) is ever in flight. */
+  async pollAllRunningTmuxSeats(db: Database.Database): Promise<void> {
+    if (this.sweeping) return;
+    this.sweeping = true;
+    try {
+      const rows = db.prepare(`
+        SELECT s.session_name as session_name
+        FROM nodes n
+        JOIN sessions s ON s.node_id = n.id
+          AND s.id = (SELECT s2.id FROM sessions s2 WHERE s2.node_id = n.id ORDER BY s2.id DESC LIMIT 1)
+        LEFT JOIN bindings b ON b.node_id = n.id
+        WHERE s.status = 'running'
+          AND s.session_name IS NOT NULL
+          AND COALESCE(b.attachment_type, 'tmux') = 'tmux'
+      `).all() as Array<{ session_name: string }>;
+      const live = new Set(rows.map((r) => r.session_name));
+      for (const s of Array.from(this.latestBySession.keys())) {
+        if (!live.has(s)) this.latestBySession.delete(s); // release memory + never serve a stale read
+      }
+      await Promise.all(rows.map(async (r) => {
+        try { await this.pollSeat(r.session_name); } catch { /* isolate: one seat's failure never crashes the sweep */ }
+      }));
+    } finally {
+      this.sweeping = false;
+    }
+  }
+
+  start(db: Database.Database, intervalMs: number = DEFAULT_STRUCTURAL_POLL_INTERVAL_MS): void {
+    if (this.timer) return; // idempotent
+    this.timer = setInterval(() => { void this.pollAllRunningTmuxSeats(db); }, intervalMs);
+    if (this.timer && typeof this.timer === "object" && "unref" in this.timer) {
+      (this.timer as NodeJS.Timeout).unref();
+    }
+  }
+
+  stop(): void {
+    if (this.timer) { clearInterval(this.timer); this.timer = null; }
+  }
+}

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

@@ -317,7 +317,7 @@ export async function probeSessionActivity(input: {
   }
 }
 
-function mapPaneState(state: PaneActivityClassification["state"]): AgentActivity["state"] {
+export function mapPaneState(state: PaneActivityClassification["state"]): AgentActivity["state"] {
   if (state === "agent_active") return "running";
   if (state === "attention") return "needs_input";
   if (state === "agent_idle") return "idle";

packages/daemon/src/index.ts

@@ -197,6 +197,8 @@ export async function startServer(port?: number) {
         // PsProjectionService + node-inventory enrichment serve
         // fresh data on each request.
         deps.seatActivityService?.start(deps.rigRepo.db);
+        // 5b82324b — the structural pane-activity cache, on the same 1Hz cadence.
+        deps.seatStructuralActivityService?.start(deps.rigRepo.db);
         // OPR.0.4.3.19 — start the liveness identity reconciler (5s default).
         // Persists the per-seat pane PID/command verdict so node-inventory
         // gates the running/active projection on verified process identity.
@@ -233,6 +235,7 @@ export async function startServer(port?: number) {
     }
     try {
       deps.seatActivityService?.stop();
+      deps.seatStructuralActivityService?.stop();
     } catch (err) {
       console.error("[seat-activity] shutdown error", err);
     }

packages/daemon/src/routes/rigs.ts

@@ -18,6 +18,7 @@ import { projectionLane } from "../domain/projection-lane.js";
 import type { TmuxAdapter } from "../adapters/tmux.js";
 import type { AgentActivityStore } from "../domain/agent-activity-store.js";
 import type { SeatActivityService } from "../domain/seat-activity-service.js";
+import type { SeatStructuralActivityService } from "../domain/seat-structural-activity-service.js";
 import { deriveRigLifecycleState } from "../domain/ps-projection.js";
 import { assessCurrentStateRehydrateEligibility } from "../domain/rehydrate-eligibility.js";
 import { buildRestorePlanPreview, collectPreviewSessionRows } from "../domain/restore-plan-preview.js";
@@ -321,8 +322,9 @@ rigsRoutes.get("/:id/graph", async (c) => {
   // capture) — the 30s topology poll colors dots from the snapshot (running/idle) +
   // hook activity; ?full=true opts into the per-node needs_input capture.
   const graphFull = c.req.query("full") === "true";
+  const seatStructuralActivityService = c.get("seatStructuralActivityService" as never) as SeatStructuralActivityService | undefined;
   const inventoryWithActivityOnly = tmuxAdapter
-    ? await attachAgentActivity(inventory, { tmuxAdapter, activityStore: agentActivityStore, captureFallback: graphFull })
+    ? await attachAgentActivity(inventory, { tmuxAdapter, activityStore: agentActivityStore, structuralActivity: seatStructuralActivityService, captureFallback: graphFull })
     : inventory;
   const seatActivityService = c.get("seatActivityService" as never) as SeatActivityService | undefined;
   const inventoryWithActivity = attachTerminalActivityAndWork(inventoryWithActivityOnly, {

packages/daemon/src/routes/sessions.ts

@@ -8,6 +8,7 @@ import type { NodeCmuxService } from "../domain/node-cmux-service.js";
 import type { TranscriptStore } from "../domain/transcript-store.js";
 import type { AgentActivityStore } from "../domain/agent-activity-store.js";
 import type { SeatActivityService } from "../domain/seat-activity-service.js";
+import type { SeatStructuralActivityService } from "../domain/seat-structural-activity-service.js";
 import {
   attachAgentActivity,
   attachTerminalActivityAndWork,
@@ -52,6 +53,7 @@ function getDeps(c: { get: (key: string) => unknown }) {
     cmuxAdapter: c.get("cmuxAdapter" as never) as CmuxAdapter,
     agentActivityStore: c.get("agentActivityStore" as never) as AgentActivityStore | undefined,
     seatActivityService: c.get("seatActivityService" as never) as SeatActivityService | undefined,
+    seatStructuralActivityService: c.get("seatStructuralActivityService" as never) as SeatStructuralActivityService | undefined,
     rigLifecycleService: c.get("rigLifecycleService" as never) as RigLifecycleService | undefined,
     restoreOrchestrator: c.get("restoreOrchestrator" as never) as RestoreOrchestrator | undefined,
   };
@@ -119,6 +121,7 @@ nodesRoutes.get("/", async (c) => {
   const withActivity = await attachAgentActivity(inventory, {
     tmuxAdapter: deps.tmuxAdapter,
     activityStore: deps.agentActivityStore,
+    structuralActivity: deps.seatStructuralActivityService,
     captureFallback: full,
   });
   const withTerminalAndWork = attachTerminalActivityAndWork(withActivity, {

packages/daemon/src/server.ts

@@ -172,6 +172,8 @@ export interface AppDeps {
    * construct AppDeps directly don't need to provide one.
    */
   seatActivityService?: import("./domain/seat-activity-service.js").SeatActivityService;
+  /** 5b82324b — the structural pane-activity cache; feeds attachAgentActivity so ACTIVITY is real. */
+  seatStructuralActivityService?: import("./domain/seat-structural-activity-service.js").SeatStructuralActivityService;
   /** OPR.0.4.3.19 — periodic liveness identity reconciler (started post-bind). */
   seatIdentityReconciler?: import("./domain/seat-identity-reconciler.js").SeatIdentityReconciler;
   psProjectionService: PsProjectionService;
@@ -546,6 +548,7 @@ export function createApp(deps: AppDeps): Hono {
     // /api/rigs/:id/nodes route can enrich entries with terminalActive +
     // hasAssignedWork via attachTerminalActivityAndWork.
     c.set("seatActivityService" as never, deps.seatActivityService);
+    c.set("seatStructuralActivityService" as never, deps.seatStructuralActivityService);
     c.set("activityHookToken" as never, deps.activityHookToken);
     c.set("serviceOrchestrator" as never, deps.serviceOrchestrator);
     c.set("composeAdapter" as never, deps.composeAdapter);

packages/daemon/src/startup.ts

@@ -56,6 +56,7 @@ import { LegacyBundleSourceResolver as BundleSourceResolver } from "./domain/bun
 import { PodBundleSourceResolver } from "./domain/bundle-source-resolver.js";
 import { PsProjectionService } from "./domain/ps-projection.js";
 import { SeatActivityService } from "./domain/seat-activity-service.js";
+import { SeatStructuralActivityService } from "./domain/seat-structural-activity-service.js";
 import { SeatIdentityReconciler, reconcileSelfHostIdentity } from "./domain/seat-identity-reconciler.js";
 import { SelfHostIdentityStore } from "./domain/seat-identity-store.js";
 import { DaemonLifecycleStore } from "./domain/daemon-lifecycle-store.js";
@@ -365,6 +366,10 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     defaultWindowSeconds: 3,
     eventBus,
   });
+  // 5b82324b — the STRUCTURAL activity cache (sibling of SeatActivityService). Captures pane TEXT once
+  // per running seat per tick + classifies motion STRUCTURALLY, so the `rig ps` ACTIVITY column reflects
+  // real liveness for hook-less / stale-hook / turn-boundary seats WITHOUT a per-request capture storm.
+  const seatStructuralActivityService = new SeatStructuralActivityService(tmuxAdapter);
   // OPR.0.4.3.19 — SeatIdentityReconciler owns the liveness identity verdict
   // (the THIRD axis). Reconciles each running seat's pane PID/command against
   // the registered binding and persists the verdict so node-inventory can gate
@@ -969,6 +974,7 @@ export async function createDaemon(opts?: DaemonOptions): Promise<DaemonResult>
     // + UI surfaces read the latest observation per seat. NEVER reads
     // queue/assignment state (non-inference contract; see slice 15 IMPL-PRD §2.3).
     seatActivityService,
+    seatStructuralActivityService,
     seatIdentityReconciler,
     // OPR.0.4.4.21 — agentActivity feeds the rig-rollup attention
     // predicate's needs_input signal (synchronous events lookup only).

packages/daemon/test/activity-route-structural.test.ts

@@ -0,0 +1,45 @@
+// 5b82324b — public-path regression (HIGH-3): the DEFAULT /api/rigs/:id/nodes route (what `rig ps`
+// hits) must consume the structural cache, so a live hook-less seat renders a real ACTIVITY instead of
+// unknown — and it must do so WITHOUT a per-request tmux capture (the healthz-wedge invariant).
+
+import { describe, it, expect } from "vitest";
+import { createFullTestDb, createTestApp } from "./helpers/test-app.js";
+import { RigRepository } from "../src/domain/rig-repository.js";
+import { SessionRegistry } from "../src/domain/session-registry.js";
+import { SeatStructuralActivityService } from "../src/domain/seat-structural-activity-service.js";
+
+describe("5b — default node-route ACTIVITY consumes the structural cache (public-path)", () => {
+  it("hook-less seat + cached structural agent_active → ACTIVITY running on the DEFAULT route, NO per-request capture", async () => {
+    const db = createFullTestDb();
+    const rigRepo = new RigRepository(db);
+    const reg = new SessionRegistry(db);
+    const rig = rigRepo.createRig("r");
+    const node = rigRepo.addNode(rig.id, "dev.impl", { runtime: "claude-code" });
+    const sess = reg.registerSession(node.id, "dev-impl@r");
+    reg.updateStatus(sess.id, "running");
+    reg.updateBinding(node.id, { tmuxSession: "dev-impl@r", attachmentType: "tmux" });
+
+    // A capture-counting tmux behind the structural service; pre-populate one background observation.
+    let captures = 0;
+    const structural = new SeatStructuralActivityService({
+      capturePaneContent: async () => { captures++; return "⠋ Working… esc to interrupt"; },
+    } as never);
+    await structural.pollSeat("dev-impl@r"); // ONE capture on the background path
+    const backgroundCaptures = captures; // 1
+    expect(structural.getStructuralActivity("dev-impl@r")?.state).toBe("agent_active");
+
+    // No runtime hook recorded ⇒ the seat is hook-less; the store returns null and the fold consults
+    // the injected structural cache.
+    const { app } = createTestApp(db, { seatStructuralActivityService: structural });
+
+    const res = await app.request(`/api/rigs/${rig.id}/nodes`); // DEFAULT (no ?full) — the rig ps path
+    expect(res.status).toBe(200);
+    const nodes = (await res.json()) as Array<{ canonicalSessionName?: string; agentActivity?: { state: string; evidenceSource: string } }>;
+    const seat = nodes.find((n) => n.canonicalSessionName === "dev-impl@r");
+    expect(seat?.agentActivity?.state).toBe("running"); // pre-5b this was unknown/no_runtime_hook
+    expect(seat?.agentActivity?.evidenceSource).toBe("pane_heuristic");
+    // Zero-request-capture invariant: the default route READ the cache; it did NOT capture during the request.
+    expect(captures).toBe(backgroundCaptures);
+    db.close();
+  });
+});

packages/daemon/test/activity-structural-signal.test.ts

@@ -0,0 +1,136 @@
+// 5b82324b — make `rig ps` ACTIVITY mean something. RED-first.
+//
+// Today attachAgentActivity's cheap default emits unknown/no_runtime_hook for a hook-less seat
+// (node-inventory.ts:1100-1107), so a LIVE seat producing pane motion shows ACTIVITY=unknown while
+// the control plane says nothing — the founder stall (8 seats at empty prompts, nothing surfaced).
+//
+// The fix lifts STRUCTURAL capture-marker discrimination into the ACTIVITY signal via a CACHED
+// structural observation (populated by a background 1Hz service, so the request path stays CAPTURE-FREE
+// — the healthz-wedge storm fix is preserved). attachAgentActivity gains a `structuralActivity` dep it
+// READS (never captures). Precedence: a fresh POSITIVE hook wins; else the cached structural verdict
+// (agent_active→running / agent_idle→idle / attention→needs_input, evidenceSource pane_heuristic);
+// else the honest unknown/no_runtime_hook. Structural OVERRIDES an absent OR stale/unknown hook —
+// liveness beats hook-arrival age (constraint 2). NEVER a verb allowlist (structural markers catch a
+// "Drizzling"-style spinner that an allowlist misses).
+
+import { describe, it, expect } from "vitest";
+import { attachAgentActivity } from "../src/domain/node-inventory.js";
+import { SeatStructuralActivityService } from "../src/domain/seat-structural-activity-service.js";
+import type { AgentActivity } from "../src/domain/types.js";
+
+function mkTmux(counter: { captures: number }) {
+  return {
+    hasSession: async () => true,
+    getPaneCommand: async () => "claude",
+    capturePaneContent: async () => {
+      counter.captures++;
+      return "some pane output\n> awaiting input";
+    },
+  } as never;
+}
+
+function entries(sessionName: string) {
+  return [{ canonicalSessionName: sessionName, runtime: "claude-code", attachmentType: "tmux", logicalId: "dev.impl" }] as never;
+}
+
+// A cached structural observation reader (what the background service exposes). Reading it does NOT
+// capture tmux — the capture already happened on the service cadence.
+function mkStructural(
+  state: "agent_active" | "agent_idle" | "attention" | "unknown" | null,
+  observedAt = "2026-08-10T20:00:00.000Z",
+) {
+  return {
+    getStructuralActivity: () =>
+      state ? { state, reason: `structural_${state}`, evidence: "pane", observedAt } : null,
+  } as never;
+}
+
+const staleHook: AgentActivity = {
+  state: "unknown",
+  reason: "stale_runtime_hook",
+  evidenceSource: "runtime_hook",
+  sampledAt: "2026-08-10T12:00:00.000Z",
+  evidence: null,
+  stale: true,
+};
+
+describe("5b — structural ACTIVITY signal (lifted into the default derivation, capture-free)", () => {
+  it("hook-less + cached structural agent_active → ACTIVITY running, NO capture (cache read, not probe)", async () => {
+    const counter = { captures: 0 };
+    const out = (await attachAgentActivity(entries("s@rig"), {
+      tmuxAdapter: mkTmux(counter),
+      activityStore: { getLatestForNode: () => null } as never,
+      structuralActivity: mkStructural("agent_active"),
+    } as never)) as Array<{ agentActivity: AgentActivity }>;
+    expect(counter.captures).toBe(0); // storm-free: the structural read is a CACHE read
+    expect(out[0]!.agentActivity.state).toBe("running");
+    expect(out[0]!.agentActivity.evidenceSource).toBe("pane_heuristic");
+  });
+
+  it("hook-less + cached structural agent_idle → idle; attention → needs_input", async () => {
+    const idle = (await attachAgentActivity(entries("s@rig"), {
+      tmuxAdapter: mkTmux({ captures: 0 }),
+      activityStore: { getLatestForNode: () => null } as never,
+      structuralActivity: mkStructural("agent_idle"),
+    } as never)) as Array<{ agentActivity: AgentActivity }>;
+    expect(idle[0]!.agentActivity.state).toBe("idle");
+    const attn = (await attachAgentActivity(entries("s@rig"), {
+      tmuxAdapter: mkTmux({ captures: 0 }),
+      activityStore: { getLatestForNode: () => null } as never,
+      structuralActivity: mkStructural("attention"),
+    } as never)) as Array<{ agentActivity: AgentActivity }>;
+    expect(attn[0]!.agentActivity.state).toBe("needs_input");
+  });
+
+  it("STALE hook + fresh structural agent_active → running (liveness beats hook-arrival age; constraint 2)", async () => {
+    const out = (await attachAgentActivity(entries("s@rig"), {
+      tmuxAdapter: mkTmux({ captures: 0 }),
+      activityStore: { getLatestForNode: () => staleHook } as never,
+      structuralActivity: mkStructural("agent_active"),
+    } as never)) as Array<{ agentActivity: AgentActivity }>;
+    expect(out[0]!.agentActivity.state).toBe("running");
+    expect(out[0]!.agentActivity.evidenceSource).toBe("pane_heuristic");
+  });
+
+  it("a fresh POSITIVE hook wins over structural (hook precedence preserved)", async () => {
+    const hook: AgentActivity = { state: "running", reason: "hook", evidenceSource: "runtime_hook", sampledAt: "2026-08-10T20:00:00.000Z", evidence: null };
+    const out = (await attachAgentActivity(entries("s@rig"), {
+      tmuxAdapter: mkTmux({ captures: 0 }),
+      activityStore: { getLatestForNode: () => hook } as never,
+      structuralActivity: mkStructural("agent_idle"), // disagrees; hook must win
+    } as never)) as Array<{ agentActivity: AgentActivity }>;
+    expect(out[0]!.agentActivity.state).toBe("running");
+    expect(out[0]!.agentActivity.evidenceSource).toBe("runtime_hook");
+  });
+
+  it("NO structural cache (empty) → honest unknown/no_runtime_hook preserved, still NO capture (healthz-wedge default)", async () => {
+    const counter = { captures: 0 };
+    const out = (await attachAgentActivity(entries("s@rig"), {
+      tmuxAdapter: mkTmux(counter),
+      activityStore: { getLatestForNode: () => null } as never,
+      structuralActivity: mkStructural(null),
+    } as never)) as Array<{ agentActivity: AgentActivity }>;
+    expect(counter.captures).toBe(0);
+    expect(out[0]!.agentActivity.state).toBe("unknown");
+    expect(out[0]!.agentActivity.reason).toBe("no_runtime_hook");
+  });
+
+  it("MF1 fold-fallback: once the REAL service invalidates the obs (capture outage), attachAgentActivity falls back to the honest stale hook — no false-live", async () => {
+    let content: string | null = "⠋ Working… esc to interrupt";
+    const svc = new SeatStructuralActivityService({ capturePaneContent: async () => content } as never);
+    await svc.pollSeat("s@rig"); // agent_active cached
+    const call = async () =>
+      (await attachAgentActivity(entries("s@rig"), {
+        tmuxAdapter: mkTmux({ captures: 0 }),
+        activityStore: { getLatestForNode: () => staleHook } as never,
+        structuralActivity: svc,
+      } as never)) as Array<{ agentActivity: AgentActivity }>;
+    const before = await call();
+    expect(before[0]!.agentActivity.state).toBe("running"); // structural overrides the stale hook
+    content = null; // capture outage
+    await svc.pollSeat("s@rig"); // invalidates the row
+    const after = await call();
+    expect(after[0]!.agentActivity.state).toBe("unknown"); // fell back to the honest stale hook
+    expect(after[0]!.agentActivity.reason).toBe("stale_runtime_hook");
+  });
+});

packages/daemon/test/helpers/test-app.ts

@@ -199,6 +199,8 @@ export function createTestApp(
     adapters?: Partial<Record<string, RuntimeAdapter>>;
     activityHookToken?: string;
     activityFreshnessMs?: number;
+    /** 5b82324b — inject a structural activity cache so route tests can prove the default /nodes ACTIVITY consumes it. */
+    seatStructuralActivityService?: import("../../src/domain/seat-structural-activity-service.js").SeatStructuralActivityService;
     // OPR.0.4.3.21 — opt-in event-loop health instrumentation for the health/
     // stress proofs. Omitted by default so every existing test keeps the exact
     // legacy `/healthz` `{ status: "ok" }` body.
@@ -369,6 +371,7 @@ export function createTestApp(
     whoamiService,
     nodeCmuxService,
     agentActivityStore,
+    seatStructuralActivityService: opts?.seatStructuralActivityService,
     activityHookToken: opts?.activityHookToken,
     eventLoopMonitor: opts?.eventLoopMonitor,
     routeTimingRecorder: opts?.routeTimingRecorder,

packages/daemon/test/seat-structural-activity-service.test.ts

@@ -0,0 +1,143 @@
+// 5b82324b — the STRUCTURAL activity cache. Proves structural classification (catching a "Drizzling"
+// gerund a verb allowlist misses), plus the two guard-required safety axes:
+//   MF1 — a stale positive verdict never survives a capture outage: a null/throw capture INVALIDATES
+//         the prior row, and a read REFUSES an observation past the freshness window.
+//   MF2 — sweeps are SINGLE-FLIGHT and held until settle: in-flight captures never exceed one sweep
+//         (N seats) across many ticks with held captures, then resume on release.
+
+import { describe, it, expect } from "vitest";
+import { createFullTestDb } from "./helpers/test-app.js";
+import { RigRepository } from "../src/domain/rig-repository.js";
+import { SessionRegistry } from "../src/domain/session-registry.js";
+import { SeatStructuralActivityService } from "../src/domain/seat-structural-activity-service.js";
+
+function mkTmux(content: string | null, counter?: { captures: number }) {
+  return {
+    capturePaneContent: async () => {
+      if (counter) counter.captures++;
+      return content;
+    },
+  } as never;
+}
+
+describe("SeatStructuralActivityService — classification", () => {
+  it("classifies a mid-work spinner pane (incl. a 'Drizzling' gerund) as agent_active + caches it", async () => {
+    const svc = new SeatStructuralActivityService(mkTmux("some output\n⠋ Drizzling… (esc to interrupt)"));
+    expect((await svc.pollSeat("dev@rig"))?.state).toBe("agent_active"); // structural markers, NOT a verb allowlist
+    expect(svc.getStructuralActivity("dev@rig")?.state).toBe("agent_active");
+    expect(svc.getStructuralActivity("dev@rig")?.observedAt).toBeTruthy();
+  });
+
+  it("classifies an empty idle prompt as agent_idle", async () => {
+    const svc = new SeatStructuralActivityService(mkTmux("output above\n❯ "));
+    expect((await svc.pollSeat("dev@rig"))?.state).toBe("agent_idle");
+  });
+
+  it("getStructuralActivity is capture-FREE (reads the cache, never re-captures)", async () => {
+    const counter = { captures: 0 };
+    const svc = new SeatStructuralActivityService(mkTmux("⠹ Working… esc to interrupt", counter));
+    await svc.pollSeat("dev@rig");
+    svc.getStructuralActivity("dev@rig");
+    svc.getStructuralActivity("dev@rig");
+    expect(counter.captures).toBe(1); // reads do not capture — no per-request storm
+  });
+});
+
+describe("SeatStructuralActivityService — MF1: a stale positive never survives a capture outage", () => {
+  it("null capture after a positive verdict INVALIDATES the row (no false-live)", async () => {
+    let content: string | null = "⠋ Working… esc to interrupt";
+    const svc = new SeatStructuralActivityService({ capturePaneContent: async () => content } as never);
+    await svc.pollSeat("s@rig");
+    expect(svc.getStructuralActivity("s@rig")?.state).toBe("agent_active");
+    content = null; // capture goes unavailable
+    await svc.pollSeat("s@rig");
+    expect(svc.getStructuralActivity("s@rig")).toBeNull(); // invalidated, not a stale positive
+  });
+
+  it("THROWING capture after a positive verdict invalidates the row", async () => {
+    let mode = "ok";
+    const svc = new SeatStructuralActivityService({
+      capturePaneContent: async () => {
+        if (mode === "throw") throw new Error("tmux gone");
+        return "⠋ Working… esc to interrupt";
+      },
+    } as never);
+    await svc.pollSeat("s@rig");
+    expect(svc.getStructuralActivity("s@rig")?.state).toBe("agent_active");
+    mode = "throw";
+    await svc.pollSeat("s@rig");
+    expect(svc.getStructuralActivity("s@rig")).toBeNull();
+  });
+
+  it("a read REFUSES + evicts an observation past the freshness window (poller made no progress)", async () => {
+    let t = Date.parse("2026-08-10T20:00:00.000Z");
+    const svc = new SeatStructuralActivityService(
+      { capturePaneContent: async () => "⠋ Working… esc to interrupt" } as never,
+      () => new Date(t),
+      20,
+      5000, // staleAfterMs
+    );
+    await svc.pollSeat("s@rig");
+    expect(svc.getStructuralActivity("s@rig")?.state).toBe("agent_active"); // fresh
+    t += 5000; // advance past the window; no fresh capture happened
+    expect(svc.getStructuralActivity("s@rig")).toBeNull(); // refused + evicted
+  });
+});
+
+describe("SeatStructuralActivityService — MF2: single-flight, held until settle", () => {
+  function dbWith2RunningSeats() {
+    const db = createFullTestDb();
+    const rigRepo = new RigRepository(db);
+    const reg = new SessionRegistry(db);
+    const rig = rigRepo.createRig("r");
+    for (const m of ["a", "b"]) {
+      const node = rigRepo.addNode(rig.id, `dev.${m}`, { runtime: "claude-code" });
+      const sess = reg.registerSession(node.id, `${m}@r`);
+      reg.updateStatus(sess.id, "running");
+      reg.updateBinding(node.id, { tmuxSession: `${m}@r`, attachmentType: "tmux" });
+    }
+    return db;
+  }
+
+  it("in-flight captures never exceed one sweep (N) across many ticks with held captures; resumes on release", async () => {
+    const db = dbWith2RunningSeats();
+    const releasers: Array<() => void> = [];
+    let started = 0;
+    let inFlight = 0;
+    let maxInFlight = 0;
+    const tmux = {
+      capturePaneContent: async () => {
+        started++;
+        inFlight++;
+        maxInFlight = Math.max(maxInFlight, inFlight);
+        await new Promise<void>((res) => releasers.push(res));
+        inFlight--;
+        return "❯ ";
+      },
+    } as never;
+    const svc = new SeatStructuralActivityService(tmux);
+
+    // Fire three sweeps while the 2 captures are held. Only the first starts captures; the next two
+    // hit the single-flight guard and no-op.
+    const s1 = svc.pollAllRunningTmuxSeats(db);
+    await new Promise((r) => setTimeout(r, 2)); // let the first sweep's captures start
+    const s2 = svc.pollAllRunningTmuxSeats(db);
+    const s3 = svc.pollAllRunningTmuxSeats(db);
+    await new Promise((r) => setTimeout(r, 2));
+    expect(started).toBe(2); // ONE sweep's worth
+    expect(maxInFlight).toBe(2); // never 4 or 6 — no overlapping whole-fleet sweeps
+
+    // Release the held captures → the first sweep settles; the skipped sweeps were already resolved.
+    releasers.splice(0).forEach((fn) => fn());
+    await Promise.all([s1, s2, s3]);
+
+    // A later tick now resumes (single-flight released after settle).
+    const s4 = svc.pollAllRunningTmuxSeats(db);
+    await new Promise((r) => setTimeout(r, 2));
+    expect(started).toBe(4); // resumed: another 2 captures
+    expect(maxInFlight).toBe(2); // still bounded to one sweep
+    releasers.splice(0).forEach((fn) => fn());
+    await s4;
+    db.close();
+  });
+});

packages/daemon/test/startup-wiring-pins.test.ts

@@ -74,6 +74,7 @@ describe("P16 — AppDeps members with safety semantics are composed (definednes
     "watchdogPolicyEngine", "watchdogScheduler", "periodicSnapshotScheduler",
     "workflowRuntime", "selfAttachService", "rigLifecycleService",
     "resumeMetadataRefresher", "runtimeAdapters",
+    "seatStructuralActivityService",
     "streamStore", "askService", "wakeResolveService", "projectClassifier",
     "classifierLeaseManager", "viewProjector", "tmuxOptionDefaults",
   ];