fix(daemon): align ps session selection with inventory

packages/daemon/src/domain/ps-projection.ts

@@ -216,14 +216,6 @@ export class PsProjectionService {
         r.name,
         r.archived_at as archived_at,
         (SELECT COUNT(*) FROM nodes n WHERE n.rig_id = r.id) as node_count,
-        (SELECT COUNT(*) FROM nodes n WHERE n.rig_id = r.id AND
-          (SELECT status FROM sessions s WHERE s.node_id = n.id ORDER BY s.created_at DESC, s.id DESC LIMIT 1) = 'running'
-        ) as running_count,
-        (SELECT MIN(s.created_at) FROM sessions s
-          JOIN nodes n ON n.id = s.node_id
-          WHERE n.rig_id = r.id AND s.status = 'running'
-          AND s.id = (SELECT s2.id FROM sessions s2 WHERE s2.node_id = s.node_id ORDER BY s2.created_at DESC, s2.id DESC LIMIT 1)
-        ) as earliest_running_at,
         (SELECT snap.created_at FROM snapshots snap WHERE snap.rig_id = r.id ORDER BY snap.created_at DESC, snap.id DESC LIMIT 1) as latest_snapshot_at,
         (SELECT COUNT(*) FROM snapshots snap WHERE snap.rig_id = r.id AND snap.kind = 'auto-periodic') as auto_periodic_count
       FROM rigs r
@@ -234,8 +226,6 @@ export class PsProjectionService {
       name: string;
       archived_at: string | null;
       node_count: number;
-      running_count: number;
-      earliest_running_at: string | null;
       latest_snapshot_at: string | null;
       auto_periodic_count: number;
     }>;
@@ -251,29 +241,32 @@ export class PsProjectionService {
     // getNodeInventory query-set PER RIG — the rig-level N+1); indexed per rig below.
     const inventoryByRig = getNodeInventoryForAllRigs(this.db);
 
+    // Read the current registration's timestamp once per node, in one batch.
+    // Uptime membership comes from the same effective-running inventory below.
+    const sessionTimes = this.db.prepare(`
+      SELECT n.id as node_id,
+        (SELECT s.created_at FROM sessions s WHERE s.node_id = n.id ORDER BY s.id DESC LIMIT 1) as created_at
+      FROM nodes n JOIN rigs r ON r.id = n.rig_id
+      ${where}
+    `).all() as Array<{ node_id: string; created_at: string | null }>;
+    const sessionCreatedAt = new Map(sessionTimes.map((row) => [row.node_id, row.created_at]));
+
     return rows.map((r) => {
       // Derive rig-level lifecycleState by folding per-node states. FS-1 W1.2:
       // inventory for ALL rigs was built in ONE batched pass above (previously one
       // getNodeInventory query-set per rig — the rig-level N+1); index it here.
       const inventory = inventoryByRig.get(r.rig_id) ?? [];
 
-      // Slice-05 item-5 (D5) — running honesty. The raw SQL `running_count` is
-      // the running BASE: it selects each node's latest session by
-      // `created_at DESC, id DESC` (the "newest session counts" contract). We
-      // must NOT re-derive that base from `inventory` — node-inventory picks the
-      // latest session by `id DESC` alone, a DIFFERENT ordering that diverges
-      // whenever session ids are not monotonic with created_at (pre-existing
-      // latent inconsistency, orthogonal to D5). Instead, subtract ONLY the
-      // status-running seats that are not EFFECTIVELY running — i.e. those
-      // carrying an applicable `session_missing` verdict — so a torn-down tmux
-      // session is never fabricated as running. The subtraction routes through
-      // the SAME isEffectivelyRunning predicate the activeCount gate uses below,
-      // so the two counts never disagree on liveness. NEVER mutates
-      // `sessions.status`.
-      const sessionMissingRunning = inventory.filter(
-        (n) => n.storedSessionStatus === "running" && !isEffectivelyRunning(n),
-      ).length;
-      const effectiveRunningCount = Math.max(0, r.running_count - sessionMissingRunning);
+      // Count the same current registrations used for lifecycle and activity.
+      // Registry IDs stay monotonic through clock rollback within one daemon;
+      // created_at does not. Reusing inventory also avoids another per-node sort.
+      const running = inventory.filter(isEffectivelyRunning);
+      const effectiveRunningCount = running.length;
+      let earliestRunningAt: string | null = null;
+      for (const node of running) {
+        const createdAt = sessionCreatedAt.get(node.nodeId);
+        if (createdAt != null && (earliestRunningAt === null || createdAt < earliestRunningAt)) earliestRunningAt = createdAt;
+      }
 
       const status: PsEntry["status"] =
         r.node_count === 0 ? "stopped"
@@ -342,7 +335,7 @@ export class PsProjectionService {
         hasWorkCount,
         status,
         lifecycleState,
-        uptime: r.earliest_running_at ? formatDuration(now - new Date(r.earliest_running_at + "Z").getTime()) : null,
+        uptime: earliestRunningAt ? formatDuration(now - new Date(earliestRunningAt + "Z").getTime()) : null,
         latestSnapshot: r.latest_snapshot_at ? formatAge(now - new Date(r.latest_snapshot_at + "Z").getTime()) : null,
         archivedAt: r.archived_at,
         isArchived: r.archived_at !== null,

packages/daemon/test/ps-projection.test.ts

@@ -1,5 +1,6 @@
 import { describe, it, expect, beforeEach, afterEach } from "vitest";
 import type Database from "better-sqlite3";
+import { monotonicFactory } from "ulid";
 import { createFullTestDb } from "./helpers/test-app.js";
 import { PsProjectionService, deriveRigLifecycleState, seatNeedsAttention, arbitratedNeedsInputSignal } from "../src/domain/ps-projection.js";
 import { SeatIdentityStore } from "../src/domain/seat-identity-store.js";
@@ -8,6 +9,8 @@ import { EventBus } from "../src/domain/event-bus.js";
 import type { AgentActivity, NodeInventoryEntry } from "../src/domain/types.js";
 import type { EvidenceRungId } from "../src/domain/activity-taxonomy.js";
 
+const sessionId = monotonicFactory();
+
 describe("PsProjectionService", () => {
   let db: Database.Database;
   let ps: PsProjectionService;
@@ -32,7 +35,7 @@ describe("PsProjectionService", () => {
   }
 
   function seedSession(nodeId: string, status: string, createdAt?: string): string {
-    const id = `sess-${nodeId}-${Date.now()}-${Math.random()}`;
+    const id = sessionId();
     db.prepare("INSERT INTO sessions (id, node_id, session_name, status, created_at) VALUES (?, ?, ?, ?, ?)")
       .run(id, nodeId, `tmux-${nodeId}`, status, createdAt ?? new Date().toISOString().replace("T", " ").slice(0, 19));
     return id;

packages/daemon/test/ps-session-ordering.test.ts

@@ -0,0 +1,109 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import type Database from "better-sqlite3";
+import { decodeTime } from "ulid";
+import { createFullTestDb } from "./helpers/test-app.js";
+import { SessionRegistry } from "../src/domain/session-registry.js";
+import { PsProjectionService } from "../src/domain/ps-projection.js";
+import { getNodeInventory } from "../src/domain/node-inventory.js";
+import { SeatIdentityStore } from "../src/domain/seat-identity-store.js";
+
+describe("ps session ordering", () => {
+  let db: Database.Database;
+  let registry: SessionRegistry;
+  let clock = Date.UTC(2100, 0, 1);
+
+  beforeEach(() => {
+    db = createFullTestDb();
+    clock += 60_000;
+    vi.useFakeTimers({ toFake: ["Date"] });
+    vi.setSystemTime(clock);
+    // Simulate the same wall clock at both production reads: ULID generation
+    // and SQLite's DEFAULT datetime('now'). Session rows/IDs are never patched.
+    db.function("datetime", (value: string) => {
+      expect(value).toBe("now");
+      return new Date().toISOString().replace("T", " ").slice(0, 19);
+    });
+    db.prepare("INSERT INTO rigs (id, name) VALUES ('r', 'ordering')").run();
+    db.prepare("INSERT INTO nodes (id, rig_id, logical_id) VALUES ('n', 'r', 'dev')").run();
+    registry = new SessionRegistry(db);
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    db.close();
+  });
+
+  describe.each([
+    ["forward", 2_000], ["same millisecond", 0], ["rollback", -2_000],
+  ] as const)("after a %s clock", (_name, delta) => {
+    it.each(["running", "exited"])("uses the current %s registration", (status) => {
+      const old = registry.registerClaimedSession("n", "dev@ordering");
+      registry.updateStatus(old.id, status === "running" ? "exited" : "running");
+      vi.setSystemTime(clock + delta);
+      const current = registry.registerClaimedSession("n", "dev@ordering");
+      registry.updateStatus(current.id, status);
+      expect(current.id > old.id).toBe(true);
+      expect(decodeTime(current.id)).toBe(clock + Math.max(delta, 0));
+      if (delta < 0) expect(current.createdAt < old.createdAt).toBe(true);
+      vi.setSystemTime(clock + 5_000);
+
+      expect(getNodeInventory(db, "r")[0]!.sessionStatus).toBe(status);
+      expect(new PsProjectionService({ db }).getEntries()[0]).toMatchObject({
+        runningCount: status === "running" ? 1 : 0,
+        status: status === "running" ? "running" : "stopped",
+        lifecycleState: status === "running" ? "running" : "stopped",
+        uptime: status === "running" ? `${(5_000 - delta) / 1_000}s` : null,
+      });
+    });
+  });
+
+  it("does not subtract a missing replacement from another node after clock rollback", () => {
+    const old = registry.registerClaimedSession("n", "dev@ordering");
+    registry.updateStatus(old.id, "exited");
+    vi.setSystemTime(clock - 2_000);
+    registry.registerClaimedSession("n", "dev@ordering");
+    registry.updateBinding("n", { tmuxSession: "dev@ordering", tmuxPane: "%1" });
+    db.prepare("INSERT INTO nodes (id, rig_id, logical_id) VALUES ('peer', 'r', 'qa')").run();
+    registry.registerClaimedSession("peer", "qa@ordering");
+    vi.setSystemTime(clock + 5_000);
+    new SeatIdentityStore(db).upsert({
+      nodeId: "n", verdict: "pane_missing", reason: "session_missing",
+      evidenceSource: "tmux_session", sessionName: "dev@ordering",
+      evidence: { registeredPane: "%1", observedPid: null, observedCommand: null, matchedLayer: null },
+      observedAt: new Date().toISOString(),
+    });
+
+    expect(getNodeInventory(db, "r").find((node) => node.nodeId === "n")!.sessionStatus).toBe("detached");
+    expect(new PsProjectionService({ db }).getEntries()[0]).toMatchObject({
+      runningCount: 1, nodeCount: 2, status: "partial", lifecycleState: "degraded",
+    });
+  });
+
+  it.each([false, true])("excludes a confirmed-missing replacement from uptime (healthy sibling: %s)", (withSibling) => {
+    const old = registry.registerClaimedSession("n", "dev@ordering");
+    registry.updateStatus(old.id, "exited");
+    vi.setSystemTime(clock - 2_000);
+    registry.registerClaimedSession("n", "dev@ordering");
+    registry.updateBinding("n", { tmuxSession: "dev@ordering", tmuxPane: "%1" });
+    if (withSibling) {
+      vi.setSystemTime(clock + 4_000);
+      db.prepare("INSERT INTO nodes (id, rig_id, logical_id) VALUES ('peer', 'r', 'qa')").run();
+      registry.registerClaimedSession("peer", "qa@ordering");
+    }
+    vi.setSystemTime(clock + 10_000);
+    new SeatIdentityStore(db).upsert({
+      nodeId: "n", verdict: "pane_missing", reason: "session_missing",
+      evidenceSource: "tmux_session", sessionName: "dev@ordering",
+      evidence: { registeredPane: "%1", observedPid: null, observedCommand: null, matchedLayer: null },
+      observedAt: new Date().toISOString(),
+    });
+
+    const stored = db.prepare("SELECT * FROM sessions ORDER BY id").all();
+    expect(new PsProjectionService({ db }).getEntries()[0]).toMatchObject({
+      runningCount: withSibling ? 1 : 0,
+      status: withSibling ? "partial" : "stopped",
+      uptime: withSibling ? "6s" : null,
+    });
+    expect(db.prepare("SELECT * FROM sessions ORDER BY id").all()).toEqual(stored);
+  });
+});

packages/daemon/test/ps-summary-stall-red.test.ts

@@ -124,18 +124,20 @@ describe("slice-04 ps/summary event-loop regression", () => {
 
       // fail-loud: a deterministic prepare throw at the ps boundary returns the EXISTING 500,
       // no partial/stale payload published; a subsequent call recovers fully.
-      const real = db.prepare.bind(db);
-      spy = vi.spyOn(db, "prepare").mockImplementation(((sql: string) => {
-        if (String(sql).includes("running_count")) throw new Error("injected ps-boundary failure");
-        return real(sql);
-      }) as never);
+      const failPrepare = vi.fn(() => { throw new Error("injected ps-boundary failure"); });
+      spy = vi.spyOn(db, "prepare").mockImplementationOnce(failPrepare);
       const failRes = await app.request("/api/ps");
+      expect(failPrepare).toHaveBeenCalledOnce();
+      // Pin the boundary without depending on a computed column's SQL alias.
+      expect(failPrepare).toHaveBeenCalledWith(expect.stringMatching(/^\s*SELECT\s+r\.id\s+as\s+rig_id\b/i));
       expect(failRes.status).toBe(500);
       const body = await failRes.text();
       expect(body).not.toContain("runningCount"); // no partial ps payload
       spy.mockRestore();
       spy = null;
-      const rec = (await (await app.request("/api/ps")).json()) as unknown[];
+      const recRes = await app.request("/api/ps");
+      expect(recRes.status).toBe(200);
+      const rec = (await recRes.json()) as unknown[];
       expect(rec).toHaveLength(9); // no cached partial; full correct response
     } finally {
       if (spy) spy.mockRestore();