mirror of
https://github.com/mvschwarz/openrig.git
synced 2026-11-07 05:56:21 -10:00
fix(ps): count active assigned work by state
This commit is contained in:
1 parent
1d2594d5af
commit
f5fd42e833
12 files changed
+321
-86
No files matched your search
@@ -22,9 +22,9 @@ interface PsEntry {
|
||||
* window. Sourced from the daemon's SeatActivityService; NEVER derived
|
||||
* from queue/assignment state. */
|
||||
activeCount?: number;
|
||||
/** Slice 15 — subset of nodes with at least one pending qitem assigned
|
||||
* to their canonical session name. Sourced from queue_items; NEVER
|
||||
* derived from tmux output. */
|
||||
/** Slice 17 — subset of nodes with at least one pending, in-progress,
|
||||
* or blocked qitem assigned. Sourced from queue_items; NEVER derived
|
||||
* from tmux output. */
|
||||
hasWorkCount?: number;
|
||||
status: "running" | "partial" | "stopped";
|
||||
/** OPR.0.4.4.21 — additive: seats needing attention (lifecycle/startup
|
||||
@@ -77,8 +77,12 @@ interface NodeEntry {
|
||||
/** Slice 15 — `has-work-to-do` primitive. Derived from queue_items;
|
||||
* NEVER derived from terminalActive. */
|
||||
hasAssignedWork?: boolean;
|
||||
/** Total active assignments (pending + in-progress + blocked). */
|
||||
assignedWorkCount?: number;
|
||||
/** Slice 15 — pending qitem count for this seat (cheap aggregate). */
|
||||
pendingWorkCount?: number;
|
||||
inProgressWorkCount?: number;
|
||||
blockedWorkCount?: number;
|
||||
/** OPR.0.3.4.11 — held reason from node.held event. */
|
||||
heldReason?: string | null;
|
||||
agentActivity?: {
|
||||
@@ -203,7 +207,10 @@ const ALLOWED_NODE_FIELDS = new Set([
|
||||
"latestError",
|
||||
"terminalActive",
|
||||
"hasAssignedWork",
|
||||
"assignedWorkCount",
|
||||
"pendingWorkCount",
|
||||
"inProgressWorkCount",
|
||||
"blockedWorkCount",
|
||||
"agentActivity",
|
||||
"contextUsage",
|
||||
"heldReason",
|
||||
@@ -630,8 +637,11 @@ export function compactNodeProjection(nodes: NodeEntry[]): Array<Record<string,
|
||||
agentActivity: attention
|
||||
? { state: n.agentActivity?.state ?? "unknown", reason: n.agentActivity?.reason }
|
||||
: { state: n.agentActivity?.state ?? "unknown" },
|
||||
// work counts.
|
||||
// Work counts. Compact keeps the base pending-only key plus the active
|
||||
// total needed for honest rendering; exact state siblings remain on
|
||||
// --full and explicit --fields.
|
||||
hasAssignedWork: n.hasAssignedWork,
|
||||
assignedWorkCount: n.assignedWorkCount,
|
||||
pendingWorkCount: n.pendingWorkCount,
|
||||
// resume summary — type + PRESENT boolean only (never the token value).
|
||||
resumeType: n.resumeType ?? null,
|
||||
@@ -799,7 +809,8 @@ latestSnapshot.
|
||||
podNamespace, canonicalSessionName, nodeKind, runtime, sessionStatus,
|
||||
startupStatus, restoreOutcome, lifecycleState, tmuxAttachCommand,
|
||||
resumeCommand, latestError, terminalActive, hasAssignedWork,
|
||||
pendingWorkCount, agentActivity, contextUsage, heldReason.
|
||||
assignedWorkCount, pendingWorkCount, inProgressWorkCount, blockedWorkCount,
|
||||
agentActivity, contextUsage, heldReason.
|
||||
|
||||
--host runs on a remote host declared in ~/.openrig/hosts.yaml (no current-rig
|
||||
default applied; the remote host's rigs are shown).
|
||||
@@ -1180,7 +1191,7 @@ async function handleNodes(
|
||||
n.canonicalSessionName ?? "—",
|
||||
abbrevNodeLifecycle(n.lifecycleState),
|
||||
formatActivity(n),
|
||||
formatHasWork(n.hasAssignedWork, n.pendingWorkCount),
|
||||
formatHasWork(n.hasAssignedWork, n.assignedWorkCount ?? n.pendingWorkCount),
|
||||
reason,
|
||||
));
|
||||
}
|
||||
@@ -1204,7 +1215,7 @@ async function handleNodes(
|
||||
n.oriented ?? "—",
|
||||
abbrevNodeLifecycle(n.lifecycleState),
|
||||
formatTerminalActive(n.terminalActive),
|
||||
formatHasWork(n.hasAssignedWork, n.pendingWorkCount),
|
||||
formatHasWork(n.hasAssignedWork, n.assignedWorkCount ?? n.pendingWorkCount),
|
||||
formatActivity(n),
|
||||
formatContextUsage(n.contextUsage),
|
||||
n.restoreOutcome,
|
||||
@@ -1312,9 +1323,10 @@ function formatTerminalActive(t: boolean | null | undefined): string {
|
||||
return "—"; // null / undefined → no signal
|
||||
}
|
||||
|
||||
// Slice 15 — render has-work with a hint of how much.
|
||||
// Pending count appended in parens when known and > 1 keeps the column
|
||||
// compact while remaining honest.
|
||||
// Slice 17 — render the total active assignment count. pendingWorkCount remains
|
||||
// pending-only and is never substituted here because claimed/blocked work must
|
||||
// stay visible after it leaves pending. pendingWorkCount is only a compatibility
|
||||
// fallback for an older daemon that predates assignedWorkCount.
|
||||
function formatHasWork(has: boolean | undefined, count: number | undefined): string {
|
||||
if (has === undefined) return "—";
|
||||
if (!has) return "no";
|
||||
|
||||
@@ -70,7 +70,10 @@ function makeFatNode(rigId: string, rigName: string, logicalId: string, opts?: {
|
||||
activityState?: string;
|
||||
activityReason?: string;
|
||||
hasAssignedWork?: boolean;
|
||||
assignedWorkCount?: number;
|
||||
pendingWorkCount?: number;
|
||||
inProgressWorkCount?: number;
|
||||
blockedWorkCount?: number;
|
||||
latestError?: string | null;
|
||||
contextPercent?: number | null;
|
||||
}) {
|
||||
@@ -96,7 +99,10 @@ function makeFatNode(rigId: string, rigName: string, logicalId: string, opts?: {
|
||||
latestError: opts?.latestError ?? null,
|
||||
terminalActive: true,
|
||||
hasAssignedWork: opts?.hasAssignedWork ?? false,
|
||||
assignedWorkCount: opts?.assignedWorkCount ?? opts?.pendingWorkCount ?? 0,
|
||||
pendingWorkCount: opts?.pendingWorkCount ?? 0,
|
||||
inProgressWorkCount: opts?.inProgressWorkCount ?? 0,
|
||||
blockedWorkCount: opts?.blockedWorkCount ?? 0,
|
||||
agentActivity: {
|
||||
state: opts?.activityState ?? "running",
|
||||
reason: opts?.activityReason ?? "mid_work_pattern",
|
||||
@@ -134,7 +140,10 @@ const RIG_A_NODES = [
|
||||
activityReason: "approval_pending",
|
||||
latestError: "Compaction imminent",
|
||||
hasAssignedWork: true,
|
||||
assignedWorkCount: 6,
|
||||
pendingWorkCount: 3,
|
||||
inProgressWorkCount: 2,
|
||||
blockedWorkCount: 1,
|
||||
contextPercent: 92,
|
||||
}),
|
||||
];
|
||||
@@ -150,7 +159,10 @@ const RIG_B_NODES = [
|
||||
const RIG_C_NODES = [
|
||||
makeFatNode("rig-c", "openrig-velocity", "orch1.lead", {
|
||||
hasAssignedWork: true,
|
||||
assignedWorkCount: 7,
|
||||
pendingWorkCount: 5,
|
||||
inProgressWorkCount: 1,
|
||||
blockedWorkCount: 1,
|
||||
}),
|
||||
makeFatNode("rig-c", "openrig-velocity", "dev1.driver"),
|
||||
makeFatNode("rig-c", "openrig-velocity", "dev1.qa"),
|
||||
@@ -298,7 +310,12 @@ describe("OPR.0.4.0.25 — rig ps token-safe defaults", () => {
|
||||
expect(node.agentActivity).toBeDefined();
|
||||
expect(node.agentActivity).toHaveProperty("state");
|
||||
expect(node).toHaveProperty("hasAssignedWork");
|
||||
expect(node).toHaveProperty("assignedWorkCount");
|
||||
expect(node).toHaveProperty("pendingWorkCount");
|
||||
// Detailed state siblings stay available through --full/--fields; compact
|
||||
// keeps every base key plus the total needed to render claimed work honestly.
|
||||
expect(node.inProgressWorkCount).toBeUndefined();
|
||||
expect(node.blockedWorkCount).toBeUndefined();
|
||||
|
||||
// Excluded heavy fields absent
|
||||
expect(node.contextUsage).toBeUndefined();
|
||||
@@ -420,6 +437,8 @@ describe("OPR.0.4.0.25 — rig ps token-safe defaults", () => {
|
||||
// Full preserves all fields; compact omits heavy ones
|
||||
expect(full[0].contextUsage).toBeDefined();
|
||||
expect(compact[0].contextUsage).toBeUndefined();
|
||||
expect(full[0]).toHaveProperty("inProgressWorkCount");
|
||||
expect(full[0]).toHaveProperty("blockedWorkCount");
|
||||
});
|
||||
|
||||
// -- Composition: --fields overrides compact (existing behavior preserved) --
|
||||
@@ -434,6 +453,25 @@ describe("OPR.0.4.0.25 — rig ps token-safe defaults", () => {
|
||||
expect(Object.keys(entries[0])).toEqual(["rigName", "canonicalSessionName"]);
|
||||
});
|
||||
|
||||
it("--fields exposes the exact assigned-work state breakdown omitted from compact rows", async () => {
|
||||
const { logs } = await captureLogs(async () => {
|
||||
await makeCmd().parseAsync([
|
||||
"node", "rig", "ps", "--nodes", "-A", "--json", "--fields",
|
||||
"canonicalSessionName,assignedWorkCount,pendingWorkCount,inProgressWorkCount,blockedWorkCount",
|
||||
]);
|
||||
});
|
||||
const entries = JSON.parse(logs.join("")).entries;
|
||||
const assigned = entries.find((entry: Record<string, unknown>) =>
|
||||
entry.canonicalSessionName === "dev1-design@openrig-build");
|
||||
expect(assigned).toEqual({
|
||||
canonicalSessionName: "dev1-design@openrig-build",
|
||||
assignedWorkCount: 6,
|
||||
pendingWorkCount: 3,
|
||||
inProgressWorkCount: 2,
|
||||
blockedWorkCount: 1,
|
||||
});
|
||||
});
|
||||
|
||||
// -- --verbose alias --
|
||||
it("--verbose is an alias for --full", async () => {
|
||||
const { logs: fullLogs } = await captureLogs(async () => {
|
||||
@@ -479,6 +517,11 @@ describe("OPR.0.4.0.25 — rig ps token-safe defaults", () => {
|
||||
// All nodes rendered (same breadth)
|
||||
expect(output).toContain("orch1-lead@openrig-build");
|
||||
expect(output).toContain("dev1-driver@openrig-velocity");
|
||||
const assignedLine = output.split("\n").find((line) => line.includes("dev1-design@openrig-build"));
|
||||
expect(assignedLine).toBeDefined();
|
||||
// WORK begins after the fixed RIG/SESSION/LIFECYCLE/ACTIVITY columns.
|
||||
// It renders the total active assignment count (6), never the pending-only count (3).
|
||||
expect(assignedLine!.slice(22 + 38 + 11 + 14, 22 + 38 + 11 + 14 + 6).trim()).toBe("6");
|
||||
});
|
||||
|
||||
it("--full human table uses full-detail columns", async () => {
|
||||
@@ -494,5 +537,10 @@ describe("OPR.0.4.0.25 — rig ps token-safe defaults", () => {
|
||||
expect(output).toContain("TERMINAL");
|
||||
expect(output).toContain("POD");
|
||||
expect(output).toContain("MEMBER");
|
||||
const assignedLine = output.split("\n").find((line) => line.includes("dev1-design@openrig-build"));
|
||||
expect(assignedLine).toBeDefined();
|
||||
// WORK begins after the fixed columns through TERMINAL in the full table.
|
||||
expect(assignedLine!.slice(30 + 10 + 14 + 34 + 12 + 17 + 10 + 10 + 9 + 11 + 9,
|
||||
30 + 10 + 14 + 34 + 12 + 17 + 10 + 10 + 9 + 11 + 9 + 6).trim()).toBe("6");
|
||||
});
|
||||
});
|
||||
@@ -82,7 +82,10 @@ function makeNode(rigId: string, rigName: string, logicalId: string, opts?: {
|
||||
resumeCommand: opts?.resumeCommand ?? null,
|
||||
latestError: null,
|
||||
hasAssignedWork: false,
|
||||
assignedWorkCount: 0,
|
||||
pendingWorkCount: 0,
|
||||
inProgressWorkCount: 0,
|
||||
blockedWorkCount: 0,
|
||||
resumeType: opts?.resumeType ?? null,
|
||||
resumeToken: opts?.resumeToken ?? null,
|
||||
heldReason: opts?.heldReason ?? null,
|
||||
@@ -361,7 +364,7 @@ describe("OPR.0.4.0.34 — rig ps current-rig default", () => {
|
||||
for (const key of [
|
||||
"rigId", "rigName", "logicalId", "canonicalSessionName",
|
||||
"sessionStatus", "startupStatus", "lifecycleState",
|
||||
"agentActivity", "hasAssignedWork", "pendingWorkCount",
|
||||
"agentActivity", "hasAssignedWork", "assignedWorkCount", "pendingWorkCount",
|
||||
"resumeType", "resumeTokenPresent", "lastActivity",
|
||||
]) {
|
||||
expect(driver, `compact row missing FR-4 key: ${key}`).toHaveProperty(key);
|
||||
|
||||
@@ -42,7 +42,10 @@ interface RFNodeData {
|
||||
currentQitems?: CurrentQitemSummary[];
|
||||
terminalActive?: boolean | null;
|
||||
hasAssignedWork?: boolean;
|
||||
assignedWorkCount?: number;
|
||||
pendingWorkCount?: number;
|
||||
inProgressWorkCount?: number;
|
||||
blockedWorkCount?: number;
|
||||
/** OPR.0.4.3.19 — liveness identity verdict threaded through so the graph
|
||||
* surfaces the same non-green mismatch/missing evidence as node inventory. */
|
||||
identityVerdict?: SeatIdentityVerdict | null;
|
||||
@@ -86,7 +89,10 @@ export interface InventoryOverlay {
|
||||
currentQitems?: CurrentQitemSummary[];
|
||||
terminalActive?: boolean | null;
|
||||
hasAssignedWork?: boolean;
|
||||
assignedWorkCount?: number;
|
||||
pendingWorkCount?: number;
|
||||
inProgressWorkCount?: number;
|
||||
blockedWorkCount?: number;
|
||||
identityVerdict?: SeatIdentityVerdict | null;
|
||||
heldReason?: string | null;
|
||||
}
|
||||
@@ -165,7 +171,10 @@ export function projectRigToGraph(input: RigGraphInput, inventoryOverlay?: Inven
|
||||
currentQitems: overlay?.currentQitems ?? [],
|
||||
terminalActive: overlay?.terminalActive,
|
||||
hasAssignedWork: overlay?.hasAssignedWork ?? false,
|
||||
assignedWorkCount: overlay?.assignedWorkCount ?? 0,
|
||||
pendingWorkCount: overlay?.pendingWorkCount ?? 0,
|
||||
inProgressWorkCount: overlay?.inProgressWorkCount ?? 0,
|
||||
blockedWorkCount: overlay?.blockedWorkCount ?? 0,
|
||||
identityVerdict: overlay?.identityVerdict ?? null,
|
||||
heldReason: overlay?.heldReason ?? null,
|
||||
},
|
||||
@@ -208,7 +217,10 @@ export function projectRigToGraph(input: RigGraphInput, inventoryOverlay?: Inven
|
||||
currentQitems: [],
|
||||
terminalActive: null,
|
||||
hasAssignedWork: false,
|
||||
assignedWorkCount: 0,
|
||||
pendingWorkCount: 0,
|
||||
inProgressWorkCount: 0,
|
||||
blockedWorkCount: 0,
|
||||
identityVerdict: null,
|
||||
heldReason: null,
|
||||
},
|
||||
|
||||
@@ -951,8 +951,8 @@ export function getNodeDetailWithContext(
|
||||
* Two orthogonal enrichments computed independently (non-inference
|
||||
* contract per IMPL-PRD §2.3):
|
||||
* - `terminalActive`: read from SeatActivityService (tmux signal)
|
||||
* - `hasAssignedWork` + `pendingWorkCount`: derived from queue_items
|
||||
* where destination_session matches the seat's canonical session name
|
||||
* - assigned-work counts: pending, in-progress, and blocked queue_items
|
||||
* where destination_session matches either canonical coordinate for the seat
|
||||
*
|
||||
* Pure / synchronous — keeps the projection cheap for both `rig ps` and
|
||||
* the UI which both fetch this per-request. The two enrichments do not
|
||||
@@ -963,7 +963,7 @@ export function attachTerminalActivityAndWork(
|
||||
deps: { db: Database.Database; seatActivity?: SeatActivityService },
|
||||
): NodeInventoryEntry[] {
|
||||
const seatActivity = deps.seatActivity ?? null;
|
||||
const pendingByDest = readPendingWorkBySession(deps.db);
|
||||
const assignedByDest = readAssignedWorkBySession(deps.db);
|
||||
return entries.map((entry) => {
|
||||
let terminalActive: boolean | null | undefined = undefined;
|
||||
// ARCH RULING 3a947fb1 (FR-7 additive): project the RAW lastActivityAt fact
|
||||
@@ -994,14 +994,14 @@ export function attachTerminalActivityAndWork(
|
||||
}
|
||||
: null;
|
||||
}
|
||||
const pendingCount = countPendingForEntry(entry, pendingByDest);
|
||||
const work = countAssignedWorkForEntry(entry, assignedByDest);
|
||||
return {
|
||||
...entry,
|
||||
terminalActive,
|
||||
lastActivityAt,
|
||||
activityState,
|
||||
hasAssignedWork: pendingCount > 0,
|
||||
pendingWorkCount: pendingCount,
|
||||
hasAssignedWork: work.assignedWorkCount > 0,
|
||||
...work,
|
||||
};
|
||||
});
|
||||
}
|
||||
@@ -1037,19 +1037,36 @@ export function attachTerminalActivityAndWork(
|
||||
* when both forms are identical (managed seats whose
|
||||
* canonicalSessionName already equals the derived canonical form).
|
||||
*/
|
||||
function countPendingForEntry(
|
||||
export interface AssignedWorkCounts {
|
||||
assignedWorkCount: number;
|
||||
pendingWorkCount: number;
|
||||
inProgressWorkCount: number;
|
||||
blockedWorkCount: number;
|
||||
}
|
||||
|
||||
export function countAssignedWorkForEntry(
|
||||
entry: NodeInventoryEntry,
|
||||
pendingByDest: Map<string, number>,
|
||||
): number {
|
||||
assignedByDest: Map<string, AssignedWorkCounts>,
|
||||
): AssignedWorkCounts {
|
||||
const keys = new Set<string>();
|
||||
if (entry.canonicalSessionName) keys.add(entry.canonicalSessionName);
|
||||
const derived = deriveCanonicalFromEntry(entry);
|
||||
if (derived) keys.add(derived);
|
||||
let count = 0;
|
||||
const total: AssignedWorkCounts = {
|
||||
assignedWorkCount: 0,
|
||||
pendingWorkCount: 0,
|
||||
inProgressWorkCount: 0,
|
||||
blockedWorkCount: 0,
|
||||
};
|
||||
for (const key of keys) {
|
||||
count += pendingByDest.get(key) ?? 0;
|
||||
const counts = assignedByDest.get(key);
|
||||
if (!counts) continue;
|
||||
total.assignedWorkCount += counts.assignedWorkCount;
|
||||
total.pendingWorkCount += counts.pendingWorkCount;
|
||||
total.inProgressWorkCount += counts.inProgressWorkCount;
|
||||
total.blockedWorkCount += counts.blockedWorkCount;
|
||||
}
|
||||
return count;
|
||||
return total;
|
||||
}
|
||||
|
||||
/** EXPORTED (OPR.0.4.6.FAC1): the derived canonical coordinate
|
||||
@@ -1065,15 +1082,32 @@ export function deriveCanonicalFromEntry(entry: NodeInventoryEntry): string | nu
|
||||
return deriveCanonicalSessionName(pod, member, entry.rigName);
|
||||
}
|
||||
|
||||
export function readPendingWorkBySession(db: Database.Database): Map<string, number> {
|
||||
export function readAssignedWorkBySession(db: Database.Database): Map<string, AssignedWorkCounts> {
|
||||
const rows = db.prepare(`
|
||||
SELECT destination_session, COUNT(*) as c
|
||||
SELECT destination_session,
|
||||
COUNT(*) as assigned_count,
|
||||
SUM(CASE WHEN state = 'pending' THEN 1 ELSE 0 END) as pending_count,
|
||||
SUM(CASE WHEN state = 'in-progress' THEN 1 ELSE 0 END) as in_progress_count,
|
||||
SUM(CASE WHEN state = 'blocked' THEN 1 ELSE 0 END) as blocked_count
|
||||
FROM queue_items
|
||||
WHERE state = 'pending'
|
||||
WHERE state IN ('pending', 'in-progress', 'blocked')
|
||||
GROUP BY destination_session
|
||||
`).all() as Array<{ destination_session: string; c: number }>;
|
||||
const out = new Map<string, number>();
|
||||
for (const r of rows) out.set(r.destination_session, r.c);
|
||||
`).all() as Array<{
|
||||
destination_session: string;
|
||||
assigned_count: number;
|
||||
pending_count: number;
|
||||
in_progress_count: number;
|
||||
blocked_count: number;
|
||||
}>;
|
||||
const out = new Map<string, AssignedWorkCounts>();
|
||||
for (const r of rows) {
|
||||
out.set(r.destination_session, {
|
||||
assignedWorkCount: r.assigned_count,
|
||||
pendingWorkCount: r.pending_count,
|
||||
inProgressWorkCount: r.in_progress_count,
|
||||
blockedWorkCount: r.blocked_count,
|
||||
});
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,11 @@
|
||||
import type Database from "better-sqlite3";
|
||||
import type { AgentActivity, NodeInventoryEntry, NodeLifecycleState, RigLifecycleState } from "./types.js";
|
||||
import { getNodeInventoryForAllRigs, readPendingWorkBySession } from "./node-inventory.js";
|
||||
import {
|
||||
countAssignedWorkForEntry,
|
||||
getNodeInventoryForAllRigs,
|
||||
readAssignedWorkBySession,
|
||||
type AssignedWorkCounts,
|
||||
} from "./node-inventory.js";
|
||||
import { archiveWhereClause, type RigArchiveFilter } from "./rig-repository.js";
|
||||
import type { SeatActivityService } from "./seat-activity-service.js";
|
||||
import type { AgentActivityStore } from "./agent-activity-store.js";
|
||||
@@ -29,9 +34,9 @@ export interface PsEntry {
|
||||
activeCount: number;
|
||||
/**
|
||||
* Slice 15 — count of nodes whose `has-work-to-do` primitive reports
|
||||
* `hasAssignedWork === true`. Derived from the pending queue items
|
||||
* with `destination_session` matching the node's canonical session
|
||||
* name. Independent from `activeCount` (non-inference contract).
|
||||
* `hasAssignedWork === true`. Derived from active queue items
|
||||
* with `destination_session` matching either of the node's canonical
|
||||
* coordinates. Independent from `activeCount` (non-inference contract).
|
||||
*/
|
||||
hasWorkCount: number;
|
||||
status: "running" | "partial" | "stopped";
|
||||
@@ -198,11 +203,11 @@ export class PsProjectionService {
|
||||
|
||||
const now = Date.now();
|
||||
|
||||
// FS-1 W1.2 (rig-level N+1 collapse): the pending-work map is host-global and
|
||||
// FS-1 W1.2 (rig-level N+1 collapse): the assigned-work map is host-global and
|
||||
// bounded by distinct destination sessions (seat count), so read it ONCE here
|
||||
// instead of one countNodesWithPendingWork query per rig. Per-rig hasWorkCount
|
||||
// is then a pure JS derivation over the already-fetched inventory below.
|
||||
const pendingByDest = readPendingWorkBySession(this.db);
|
||||
const assignedByDest = readAssignedWorkBySession(this.db);
|
||||
// FS-1 W1.2: node inventory for ALL rigs built in ONE batched pass (was one
|
||||
// getNodeInventory query-set PER RIG — the rig-level N+1); indexed per rig below.
|
||||
const inventoryByRig = getNodeInventoryForAllRigs(this.db);
|
||||
@@ -254,31 +259,14 @@ export class PsProjectionService {
|
||||
}
|
||||
|
||||
// Slice 15 — `has-work-to-do` count. Subset of nodes with at
|
||||
// least one pending qitem whose `destination_session` matches the
|
||||
// node's `canonicalSessionName`. Sourced ONLY from the queue
|
||||
// least one pending, in-progress, or blocked qitem whose destination
|
||||
// matches either canonical coordinate for the node. Sourced ONLY from the queue
|
||||
// projection; never from tmux output.
|
||||
// FS-1 W1.2: derived in JS from the single global `pendingByDest` map,
|
||||
// byte-identical to the deleted per-rig `countNodesWithPendingWork` query.
|
||||
// That query matched `q.destination_session = s.session_name` (SINGLE-KEY)
|
||||
// where `s` is the node's latest session; `canonicalSessionName` IS that
|
||||
// latest `session_name` (node-inventory.ts return), and the map is
|
||||
// `destination_session -> pending count`, so `> 0` reproduces its `EXISTS`.
|
||||
// NAMED DIVERGENCE (arch pin, FS-1): this derivation DELIBERATELY keeps the
|
||||
// legacy rollup's SINGLE-KEY match (canonicalSessionName == the latest
|
||||
// session_name) and does NOT use the per-node DUAL-KEY sibling
|
||||
// `countPendingForEntry` (node-inventory.ts) — the canonical + derived
|
||||
// `{pod}-{member}@{rig}` match that is the BLOCKING-A2 adopted-seat fix.
|
||||
// The rollup and the per-node enrichment already diverged here BEFORE FS-1;
|
||||
// FS-1 is a perf slice whose proof spine is byte-identity, so it preserves
|
||||
// the rollup's exact prior semantics on purpose — this is preserved, not a
|
||||
// bug. KNOWN CONSEQUENCE: the rollup undercounts pending work for adopted
|
||||
// seats. The future correctness fix now sits ONE LINE away — swap this loop
|
||||
// for the shared dual-key `countPendingForEntry` derivation + update the
|
||||
// byte-identity harness. Arch-routed to pm as a named small correctness
|
||||
// candidate (a glance-layer undercount).
|
||||
// Slice 17: this is the SAME dual-key resolver and active-state map used by
|
||||
// per-node inventory. The former single-key/pending-only divergence is gone.
|
||||
let hasWorkCount = 0;
|
||||
for (const node of inventory) {
|
||||
if (node.canonicalSessionName && (pendingByDest.get(node.canonicalSessionName) ?? 0) > 0) hasWorkCount++;
|
||||
if (countAssignedWorkForEntry(node, assignedByDest).assignedWorkCount > 0) hasWorkCount++;
|
||||
}
|
||||
|
||||
// OPR.0.4.4.21 — attention fold, same inventory pass. Hook activity
|
||||
@@ -318,15 +306,23 @@ export class PsProjectionService {
|
||||
}
|
||||
|
||||
/**
|
||||
* Slice 15 — count of pending qitems assigned to a specific canonical
|
||||
* session name. Exposed for per-node enrichment in node-inventory.
|
||||
* Compatibility helper: count pending-only qitems for one canonical session.
|
||||
* The wider assigned-work projection uses countAssignedWorkForSession below.
|
||||
*/
|
||||
export function countPendingWorkForSession(db: Database.Database, canonicalSessionName: string): number {
|
||||
const row = db.prepare(`
|
||||
SELECT COUNT(*) as c FROM queue_items
|
||||
WHERE destination_session = ? AND state = 'pending'
|
||||
`).get(canonicalSessionName) as { c: number } | undefined;
|
||||
return row?.c ?? 0;
|
||||
return countAssignedWorkForSession(db, canonicalSessionName).pendingWorkCount;
|
||||
}
|
||||
|
||||
export function countAssignedWorkForSession(
|
||||
db: Database.Database,
|
||||
canonicalSessionName: string,
|
||||
): AssignedWorkCounts {
|
||||
return readAssignedWorkBySession(db).get(canonicalSessionName) ?? {
|
||||
assignedWorkCount: 0,
|
||||
pendingWorkCount: 0,
|
||||
inProgressWorkCount: 0,
|
||||
blockedWorkCount: 0,
|
||||
};
|
||||
}
|
||||
|
||||
function formatDuration(ms: number): string {
|
||||
|
||||
@@ -649,15 +649,21 @@ export interface NodeInventoryEntry {
|
||||
lastActivityAt?: string | null;
|
||||
/**
|
||||
* Slice 15 — `has-work-to-do` primitive. Derived from queue/assignment
|
||||
* projection (pending qitems whose `destination_session` matches this
|
||||
* seat's `canonicalSessionName`).
|
||||
* projection (pending, in-progress, or blocked qitems whose
|
||||
* `destination_session` matches this seat's canonical coordinate).
|
||||
*
|
||||
* MUST NOT be derived from `terminalActive` or tmux output. Two
|
||||
* orthogonal primitives, never one inferred from the other.
|
||||
*/
|
||||
hasAssignedWork?: boolean;
|
||||
/** Total active qitems assigned to this seat (pending + in-progress + blocked). */
|
||||
assignedWorkCount?: number;
|
||||
/** Optional count of pending qitems assigned to this seat (cheap aggregate). */
|
||||
pendingWorkCount?: number;
|
||||
/** Optional count of claimed in-progress qitems assigned to this seat. */
|
||||
inProgressWorkCount?: number;
|
||||
/** Optional count of blocked qitems still assigned to this seat. */
|
||||
blockedWorkCount?: number;
|
||||
/**
|
||||
* OPR.0.4.3.19 — liveness identity verdict (the THIRD axis). Present when
|
||||
* the SeatIdentityReconciler has recorded a verdict for this node; absent
|
||||
|
||||
@@ -403,7 +403,10 @@ rigsRoutes.get("/:id/graph", async (c) => {
|
||||
: [],
|
||||
terminalActive: n.terminalActive,
|
||||
hasAssignedWork: n.hasAssignedWork ?? false,
|
||||
assignedWorkCount: n.assignedWorkCount ?? 0,
|
||||
pendingWorkCount: n.pendingWorkCount ?? 0,
|
||||
inProgressWorkCount: n.inProgressWorkCount ?? 0,
|
||||
blockedWorkCount: n.blockedWorkCount ?? 0,
|
||||
identityVerdict: n.identityVerdict ?? null,
|
||||
}));
|
||||
const projectedPods: Pod[] = pods.map((pod) => ({
|
||||
|
||||
@@ -17,7 +17,12 @@ import { createFullTestDb } from "./helpers/test-app.js";
|
||||
import {
|
||||
attachTerminalActivityAndWork,
|
||||
getNodeInventory,
|
||||
readAssignedWorkBySession,
|
||||
} from "../src/domain/node-inventory.js";
|
||||
import {
|
||||
countAssignedWorkForSession,
|
||||
countPendingWorkForSession,
|
||||
} from "../src/domain/ps-projection.js";
|
||||
|
||||
function seedRig(db: Database.Database, name: string): string {
|
||||
const id = `rig-${name}`;
|
||||
@@ -34,13 +39,23 @@ function seedSession(db: Database.Database, nodeId: string, sessionName: string,
|
||||
db.prepare("INSERT INTO sessions (id, node_id, session_name, status, created_at) VALUES (?, ?, ?, ?, ?)")
|
||||
.run(id, nodeId, sessionName, status, new Date().toISOString().replace("T", " ").slice(0, 19));
|
||||
}
|
||||
function seedPendingQitem(db: Database.Database, destinationSession: string, body = "test-body"): void {
|
||||
function seedQitem(
|
||||
db: Database.Database,
|
||||
destinationSession: string,
|
||||
state: string,
|
||||
body = "test-body",
|
||||
): string {
|
||||
const id = `q-${Date.now()}-${Math.random()}`;
|
||||
const ts = new Date().toISOString().replace("T", " ").slice(0, 19);
|
||||
db.prepare(`
|
||||
INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, tier, body)
|
||||
VALUES (?, ?, ?, ?, ?, 'pending', 'routine', 'routine', ?)
|
||||
`).run(id, ts, ts, "op@test", destinationSession, body);
|
||||
VALUES (?, ?, ?, ?, ?, ?, 'routine', 'routine', ?)
|
||||
`).run(id, ts, ts, "op@test", destinationSession, state, body);
|
||||
return id;
|
||||
}
|
||||
|
||||
function seedPendingQitem(db: Database.Database, destinationSession: string, body = "test-body"): void {
|
||||
seedQitem(db, destinationSession, "pending", body);
|
||||
}
|
||||
|
||||
function makeSeatActivityFor(activeBySession: Record<string, boolean | null>) {
|
||||
@@ -326,4 +341,71 @@ describe("attachTerminalActivityAndWork — slice 15 per-node enrichment", () =>
|
||||
expect(entry!.pendingWorkCount).toBe(1);
|
||||
expect(entry!.hasAssignedWork).toBe(true);
|
||||
});
|
||||
|
||||
it("slice 17 RED specimen — a claimed in-progress row remains assigned without changing pendingWorkCount", () => {
|
||||
const rig = seedRig(db, "claimed");
|
||||
const n = seedNode(db, rig, "dev");
|
||||
seedSession(db, n, "dev@rig", "running");
|
||||
const qitemId = seedQitem(db, "dev@rig", "in-progress");
|
||||
db.prepare("UPDATE queue_items SET claimed_at = ? WHERE qitem_id = ?")
|
||||
.run("2026-08-28 09:24:44", qitemId);
|
||||
|
||||
const [entry] = attachTerminalActivityAndWork(getNodeInventory(db, rig), { db });
|
||||
expect(entry!.hasAssignedWork).toBe(true);
|
||||
expect(entry!.assignedWorkCount).toBe(1);
|
||||
expect(entry!.pendingWorkCount).toBe(0);
|
||||
expect(entry!.inProgressWorkCount).toBe(1);
|
||||
expect(entry!.blockedWorkCount).toBe(0);
|
||||
});
|
||||
|
||||
it.each([
|
||||
["pending", true, 1, 1, 0, 0],
|
||||
["in-progress", true, 1, 0, 1, 0],
|
||||
["blocked", true, 1, 0, 0, 1],
|
||||
["done", false, 0, 0, 0, 0],
|
||||
["canceled", false, 0, 0, 0, 0],
|
||||
["handed-off", false, 0, 0, 0, 0],
|
||||
] as const)(
|
||||
"slice 17 state matrix — %s projects assigned=%s with exact per-state counts",
|
||||
(state, hasAssignedWork, assigned, pending, inProgress, blocked) => {
|
||||
const rig = seedRig(db, `matrix-${state}`);
|
||||
const n = seedNode(db, rig, "dev");
|
||||
seedSession(db, n, "dev@rig", "running");
|
||||
seedQitem(db, "dev@rig", state);
|
||||
|
||||
const [entry] = attachTerminalActivityAndWork(getNodeInventory(db, rig), { db });
|
||||
expect(entry!.hasAssignedWork).toBe(hasAssignedWork);
|
||||
expect(entry!.assignedWorkCount).toBe(assigned);
|
||||
expect(entry!.pendingWorkCount).toBe(pending);
|
||||
expect(entry!.inProgressWorkCount).toBe(inProgress);
|
||||
expect(entry!.blockedWorkCount).toBe(blocked);
|
||||
},
|
||||
);
|
||||
|
||||
it("slice 17 mixed-state breakdown keeps pendingWorkCount pending-only and ignores terminal rows", () => {
|
||||
const rig = seedRig(db, "mixed");
|
||||
const n = seedNode(db, rig, "dev");
|
||||
seedSession(db, n, "dev@rig", "running");
|
||||
for (const state of ["pending", "pending", "in-progress", "blocked", "done", "canceled", "handed-off"]) {
|
||||
seedQitem(db, "dev@rig", state);
|
||||
}
|
||||
|
||||
const [entry] = attachTerminalActivityAndWork(getNodeInventory(db, rig), { db });
|
||||
expect(entry!.hasAssignedWork).toBe(true);
|
||||
expect(entry!.assignedWorkCount).toBe(4);
|
||||
expect(entry!.pendingWorkCount).toBe(2);
|
||||
expect(entry!.inProgressWorkCount).toBe(1);
|
||||
expect(entry!.blockedWorkCount).toBe(1);
|
||||
expect(countPendingWorkForSession(db, "dev@rig")).toBe(2);
|
||||
});
|
||||
|
||||
it("slice 17 batch and per-session computation sites return identical state counts", () => {
|
||||
for (const state of ["pending", "pending", "in-progress", "blocked", "done"]) {
|
||||
seedQitem(db, "dev@rig", state);
|
||||
}
|
||||
|
||||
expect(readAssignedWorkBySession(db).get("dev@rig")).toEqual(
|
||||
countAssignedWorkForSession(db, "dev@rig"),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -151,15 +151,21 @@ describe("PL-019 projectRigToGraph: agentActivity + currentQitems", () => {
|
||||
canonicalSessionName: "r01-alpha",
|
||||
restoreOutcome: "n-a",
|
||||
terminalActive: true,
|
||||
hasAssignedWork: false,
|
||||
pendingWorkCount: 0,
|
||||
hasAssignedWork: true,
|
||||
assignedWorkCount: 4,
|
||||
pendingWorkCount: 1,
|
||||
inProgressWorkCount: 2,
|
||||
blockedWorkCount: 1,
|
||||
},
|
||||
];
|
||||
const result = projectRigToGraph(input, overlay);
|
||||
const data = result.nodes[0].data;
|
||||
expect(data.terminalActive).toBe(true);
|
||||
expect(data.hasAssignedWork).toBe(false);
|
||||
expect(data.pendingWorkCount).toBe(0);
|
||||
expect(data.hasAssignedWork).toBe(true);
|
||||
expect(data.assignedWorkCount).toBe(4);
|
||||
expect(data.pendingWorkCount).toBe(1);
|
||||
expect(data.inProgressWorkCount).toBe(2);
|
||||
expect(data.blockedWorkCount).toBe(1);
|
||||
});
|
||||
|
||||
it("defaults currentQitems to [] when overlay omits them", () => {
|
||||
|
||||
@@ -332,13 +332,17 @@ describe("PsProjectionService", () => {
|
||||
};
|
||||
}
|
||||
|
||||
function seedPendingQitem(destinationSession: string): void {
|
||||
function seedQitem(destinationSession: string, state: string): void {
|
||||
const id = `qitem-${Date.now()}-${Math.random()}`;
|
||||
const ts = new Date().toISOString().replace("T", " ").slice(0, 19);
|
||||
db.prepare(`
|
||||
INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, tier, body)
|
||||
VALUES (?, ?, ?, ?, ?, 'pending', 'routine', 'routine', ?)
|
||||
`).run(id, ts, ts, "operator@test", destinationSession, "test-body");
|
||||
VALUES (?, ?, ?, ?, ?, ?, 'routine', 'routine', ?)
|
||||
`).run(id, ts, ts, "operator@test", destinationSession, state, "test-body");
|
||||
}
|
||||
|
||||
function seedPendingQitem(destinationSession: string): void {
|
||||
seedQitem(destinationSession, "pending");
|
||||
}
|
||||
|
||||
it("runningCount stays process-alive semantics; activeCount + hasWorkCount default to 0 when no signals wired", () => {
|
||||
@@ -433,22 +437,26 @@ describe("PsProjectionService", () => {
|
||||
expect(entries[0]!.hasWorkCount).toBe(1);
|
||||
});
|
||||
|
||||
it("done/closed qitems do NOT count toward hasWorkCount (only 'pending' state)", () => {
|
||||
it("active qitems count toward hasWorkCount while terminal qitems do not", () => {
|
||||
const rigId = seedRig("only-pending");
|
||||
const n1 = seedNode(rigId, "dev");
|
||||
seedSession(n1, "running");
|
||||
const paneId = `tmux-${n1}`;
|
||||
// Insert one pending + one done qitem for the same seat.
|
||||
seedPendingQitem(paneId);
|
||||
const id = `qitem-done-${Date.now()}`;
|
||||
const ts = new Date().toISOString().replace("T", " ").slice(0, 19);
|
||||
db.prepare(`
|
||||
INSERT INTO queue_items (qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, tier, body)
|
||||
VALUES (?, ?, ?, ?, ?, 'done', 'routine', 'routine', ?)
|
||||
`).run(id, ts, ts, "operator@test", paneId, "test-body");
|
||||
for (const state of ["pending", "in-progress", "blocked", "done", "canceled", "handed-off"]) {
|
||||
seedQitem(paneId, state);
|
||||
}
|
||||
|
||||
const entries = ps.getEntries();
|
||||
expect(entries[0]!.hasWorkCount).toBe(1); // only the pending one counts
|
||||
expect(entries[0]!.hasWorkCount).toBe(1); // one seat, regardless of its three active rows
|
||||
});
|
||||
|
||||
it.each(["in-progress", "blocked"])("slice 17 — a seat with only %s work counts as assigned", (state) => {
|
||||
const rigId = seedRig(`assigned-${state}`);
|
||||
const n1 = seedNode(rigId, "dev");
|
||||
seedSession(n1, "running");
|
||||
seedQitem(`tmux-${n1}`, state);
|
||||
|
||||
expect(ps.getEntries()[0]!.hasWorkCount).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -310,6 +310,31 @@ describe("Rig CRUD routes", () => {
|
||||
expect(nodeData.data.status).toBe("running");
|
||||
});
|
||||
|
||||
it("GET /api/rigs/:id/graph -> node.data carries the honest assigned-work breakdown", async () => {
|
||||
const rig = repo.createRig("r01");
|
||||
const node = repo.addNode(rig.id, "dev.impl", { role: "worker" });
|
||||
const session = sessionRegistry.registerSession(node.id, "dev-impl@r01");
|
||||
sessionRegistry.updateStatus(session.id, "running");
|
||||
const ts = "2026-08-28 10:00:00";
|
||||
for (const [index, state] of ["pending", "in-progress", "blocked", "done"].entries()) {
|
||||
db.prepare(`
|
||||
INSERT INTO queue_items
|
||||
(qitem_id, ts_created, ts_updated, source_session, destination_session, state, priority, tier, body)
|
||||
VALUES (?, ?, ?, 'op@test', 'dev-impl@r01', ?, 'routine', 'routine', 'body')
|
||||
`).run(`q-graph-${index}`, ts, ts, state);
|
||||
}
|
||||
|
||||
const res = await app.request(`/api/rigs/${rig.id}/graph`);
|
||||
expect(res.status).toBe(200);
|
||||
const body = await res.json();
|
||||
const data = body.nodes.find((n: { data: { logicalId: string } }) => n.data.logicalId === "dev.impl").data;
|
||||
expect(data.hasAssignedWork).toBe(true);
|
||||
expect(data.assignedWorkCount).toBe(3);
|
||||
expect(data.pendingWorkCount).toBe(1);
|
||||
expect(data.inProgressWorkCount).toBe(1);
|
||||
expect(data.blockedWorkCount).toBe(1);
|
||||
});
|
||||
|
||||
it("GET /api/rigs/:id/graph -> unbound node has binding: null in data", async () => {
|
||||
const rig = repo.createRig("r01");
|
||||
repo.addNode(rig.id, "worker");
|
||||
|
||||
Reference in new issue
Block a user