diff --git a/dashboard/README.md b/dashboard/README.md index e981ae3c..23f97a0e 100644 --- a/dashboard/README.md +++ b/dashboard/README.md @@ -36,6 +36,7 @@ A modern Next.js dashboard for visualizing and managing the Exosphere State Mana - **Secret Management**: View and manage node secrets securely - **Schema Validation**: JSON schema rendering with type information - **Node Details Modal**: Comprehensive node information display +- **Eligibility Time**: Delayed nodes show when they become eligible to run (local timezone), and past-due nodes show how long they have been waiting for a worker ## 🚀 Getting Started diff --git a/dashboard/package.json b/dashboard/package.json index b737a5e9..31129570 100644 --- a/dashboard/package.json +++ b/dashboard/package.json @@ -6,7 +6,8 @@ "dev": "next dev --turbopack", "build": "next build", "start": "next start", - "lint": "next lint" + "lint": "next lint", + "test": "node --test \"src/**/*.test.ts\"" }, "dependencies": { "@radix-ui/react-slot": "^1.2.4", diff --git a/dashboard/src/components/NodeDetailsModal.tsx b/dashboard/src/components/NodeDetailsModal.tsx index 68b25e4c..9a91c8ea 100644 --- a/dashboard/src/components/NodeDetailsModal.tsx +++ b/dashboard/src/components/NodeDetailsModal.tsx @@ -14,6 +14,10 @@ import { Button } from '@/components/ui/button'; import { Badge } from '@/components/ui/badge'; import { GraphNode as GraphNodeType, NodeRunDetailsResponse } from '@/types/state-manager'; import { clientApiService } from '@/services/clientApi'; +import { getNodeEligibility, formatEligibilityTime } from '@/lib/nodeEligibility'; + +/** How often the eligibility row re-checks the clock so "scheduled" turns into "waiting" on time. */ +const ELIGIBILITY_TICK_MS = 5000; interface NodeDetailsModalProps { selectedNode: GraphNodeType | null; @@ -37,6 +41,24 @@ export const NodeDetailsModal: React.FC = ({ const [retryState, setRetryState] = useState<'idle' | 'confirm' | 'loading' | 'success' | 'error'>('idle'); const [retryError, setRetryError] = useState(null); const [countdown, setCountdown] = useState(null); + // Clock reading paired with the details it was taken for, so a node opened long after page + // load is classified against "now" and never against a stale reading for a previous node. + const [clock, setClock] = useState<{ details: NodeRunDetailsResponse | null; nowMs: number }>({ details: null, nowMs: 0 }); + + // Tick only while a node is waiting on an eligibility time; every timer is cleared when the + // details change (node picked up, node changed, modal closed) or the component unmounts. + useEffect(() => { + if (!selectedNodeDetails || getNodeEligibility(selectedNodeDetails, Date.now()) === null) return; + const tick = () => setClock({ details: selectedNodeDetails, nowMs: Date.now() }); + const first = setTimeout(tick, 0); + const timer = setInterval(tick, ELIGIBILITY_TICK_MS); + return () => { + clearTimeout(first); + clearInterval(timer); + }; + }, [selectedNodeDetails]); + + const eligibility = clock.details === selectedNodeDetails ? getNodeEligibility(selectedNodeDetails, clock.nowMs) : null; // Reset retry state when modal closes or node changes useEffect(() => { @@ -310,6 +332,21 @@ export const NodeDetailsModal: React.FC = ({ {new Date(selectedNodeDetails.updated_at).toLocaleString()} )} + {eligibility && ( +
+
+ + {eligibility.kind === 'scheduled' ? 'Eligible after:' : 'Eligible since:'} + + {formatEligibilityTime(eligibility.at)} +
+

+ {eligibility.kind === 'scheduled' + ? 'Delayed by a start delay, retry policy or requeue signal. A worker can pick this node up any time after this moment; it is not a guaranteed start time.' + : 'Eligible to run and waiting for a worker to pick it up.'} +

+
+ )} diff --git a/dashboard/src/lib/nodeEligibility.test.ts b/dashboard/src/lib/nodeEligibility.test.ts new file mode 100644 index 00000000..cfea38d7 --- /dev/null +++ b/dashboard/src/lib/nodeEligibility.test.ts @@ -0,0 +1,57 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { getNodeEligibility, formatEligibilityTime } from './nodeEligibility.ts'; + +// Fake clock: every case injects `now` instead of reading Date.now(), so no real timers run. +const NOW_MS = Date.UTC(2026, 0, 15, 12, 0, 0); // 2026-01-15T12:00:00Z + +test('future delayed CREATED node is scheduled for its stored eligibility time', () => { + const at = NOW_MS + 90 * 60 * 1000; + const result = getNodeEligibility({ status: 'CREATED', enqueue_after: at }, NOW_MS); + assert.deepEqual(result, { kind: 'scheduled', at: new Date(at) }); +}); + +test('past-due CREATED node is reported as eligible and waiting, not as scheduled', () => { + const at = NOW_MS - 5 * 60 * 1000; + const result = getNodeEligibility({ status: 'CREATED', enqueue_after: at }, NOW_MS); + assert.deepEqual(result, { kind: 'waiting', at: new Date(at) }); +}); + +test('an eligibility time equal to now counts as eligible (matches the server-side <= pick-up rule)', () => { + const result = getNodeEligibility({ status: 'CREATED', enqueue_after: NOW_MS }, NOW_MS); + assert.equal(result?.kind, 'waiting'); +}); + +test('old record without the field shows nothing (no Invalid Date, no epoch zero)', () => { + assert.equal(getNodeEligibility({ status: 'CREATED' }, NOW_MS), null); + assert.equal(getNodeEligibility({ status: 'CREATED', enqueue_after: undefined }, NOW_MS), null); + assert.equal(getNodeEligibility({ status: 'CREATED', enqueue_after: null }, NOW_MS), null); + assert.equal(getNodeEligibility({ status: 'CREATED', enqueue_after: 0 }, NOW_MS), null); + assert.equal(getNodeEligibility({ status: 'CREATED', enqueue_after: Number.NaN }, NOW_MS), null); +}); + +test('already queued, running or finished nodes carry no waiting message', () => { + const at = NOW_MS + 60 * 1000; + for (const status of ['QUEUED', 'EXECUTED', 'SUCCESS', 'ERRORED', 'TIMEDOUT', 'CANCELLED', 'PRUNED', 'NEXT_CREATED_ERROR'] as const) { + assert.equal(getNodeEligibility({ status, enqueue_after: at }, NOW_MS), null, status); + } +}); + +test('missing details yield nothing', () => { + assert.equal(getNodeEligibility(null, NOW_MS), null); + assert.equal(getNodeEligibility(undefined, NOW_MS), null); +}); + +test('the same instant renders in the viewer timezone with a zone label, never re-persisted', () => { + const at = new Date(Date.UTC(2026, 0, 15, 23, 30, 0)); // near a day boundary + const utc = formatEligibilityTime(at, 'en-US', 'UTC'); + const tokyo = formatEligibilityTime(at, 'en-US', 'Asia/Tokyo'); + const la = formatEligibilityTime(at, 'en-US', 'America/Los_Angeles'); + assert.match(utc, /UTC/); + assert.match(tokyo, /GMT\+9|JST/); + assert.match(la, /PST|GMT-8/); + // Different calendar days for the same instant; the instant itself is unchanged. + assert.match(tokyo, /1\/16\/2026/); + assert.match(la, /1\/15\/2026/); + assert.equal(at.getTime(), Date.UTC(2026, 0, 15, 23, 30, 0)); +}); diff --git a/dashboard/src/lib/nodeEligibility.ts b/dashboard/src/lib/nodeEligibility.ts new file mode 100644 index 00000000..bd2bb761 --- /dev/null +++ b/dashboard/src/lib/nodeEligibility.ts @@ -0,0 +1,45 @@ +/** + * Eligibility of a node that has not been picked up by a worker yet. + * + * `enqueue_after` is the stored epoch-millisecond instant after which the state manager + * lets a worker pick the state up (`enqueue_after <= now` on the server). It explains a delay; + * it does not promise execution at that instant. + */ +export type NodeEligibility = + /** The eligibility time is still ahead of the viewer's clock. */ + | { kind: 'scheduled'; at: Date } + /** The eligibility time has passed and the node is still waiting for a worker. */ + | { kind: 'waiting'; at: Date }; + +export interface NodeEligibilityInput { + status: string; + enqueue_after?: number | null; +} + +/** Only states a worker has not claimed yet can be waiting on an eligibility time. */ +const WAITING_STATUSES: ReadonlySet = new Set(['CREATED']); + +/** + * Classify a node's stored eligibility time against an injected clock. + * Returns null when nothing should be shown: the node is not waiting to be picked up, + * or the record carries no usable timestamp (older records, missing field, 0, NaN). + */ +export function getNodeEligibility( + details: NodeEligibilityInput | null | undefined, + nowMs: number, +): NodeEligibility | null { + if (!details || !WAITING_STATUSES.has(details.status)) return null; + + const at = details.enqueue_after; + if (typeof at !== 'number' || !Number.isFinite(at) || at <= 0) return null; + + return at > nowMs ? { kind: 'scheduled', at: new Date(at) } : { kind: 'waiting', at: new Date(at) }; +} + +/** + * Render an instant in the viewer's locale and timezone with an explicit zone label. + * Formatting is display-only; the wire representation stays epoch milliseconds. + */ +export function formatEligibilityTime(at: Date, locale?: string, timeZone?: string): string { + return at.toLocaleString(locale, { timeZone, timeZoneName: 'short' }); +} diff --git a/dashboard/src/types/state-manager.ts b/dashboard/src/types/state-manager.ts index 7ae03134..f2c55db2 100644 --- a/dashboard/src/types/state-manager.ts +++ b/dashboard/src/types/state-manager.ts @@ -217,6 +217,8 @@ export interface NodeRunDetailsResponse { parents: Record; created_at: string; updated_at: string; + /** Epoch milliseconds after which the state is eligible to be enqueued; absent on older records/servers. */ + enqueue_after?: number | null; } // Runs Types diff --git a/dashboard/tsconfig.json b/dashboard/tsconfig.json index 55e2a026..6e972825 100644 --- a/dashboard/tsconfig.json +++ b/dashboard/tsconfig.json @@ -6,6 +6,7 @@ "skipLibCheck": true, "strict": true, "noEmit": true, + "allowImportingTsExtensions": true, "esModuleInterop": true, "module": "esnext", "moduleResolution": "bundler", diff --git a/docs/docs/exosphere/dashboard.md b/docs/docs/exosphere/dashboard.md index d43e09a4..b39899b5 100644 --- a/docs/docs/exosphere/dashboard.md +++ b/docs/docs/exosphere/dashboard.md @@ -169,6 +169,8 @@ View registered nodes, and graph templates on a namespace ![Runs Overview](../assets/DashboardSS-3.jpg) View graph runs and debug each node that was created. +Nodes that are still `CREATED` show when they become eligible to run: a node delayed by a start delay, a retry policy or a `ReQueueAfterSignal`/`RequeueAtSignal` shows **Eligible after** with the time in your local timezone, and a node whose eligibility time has already passed shows **Eligible since** while it waits for a worker. This is the stored eligibility time, not a guaranteed start time; a worker picks the node up on its next poll after that moment. + ## Using the Dashboard 1. **Configure Connection**: diff --git a/state-manager/app/controller/get_node_run_details.py b/state-manager/app/controller/get_node_run_details.py index aa649434..6340328e 100644 --- a/state-manager/app/controller/get_node_run_details.py +++ b/state-manager/app/controller/get_node_run_details.py @@ -71,7 +71,8 @@ async def get_node_run_details(namespace: str, graph_name: str, run_id: str, nod error=state.error, parents=parent_identifiers, created_at=state.created_at.isoformat() if state.created_at else "", - updated_at=state.updated_at.isoformat() if state.updated_at else "" + updated_at=state.updated_at.isoformat() if state.updated_at else "", + enqueue_after=state.enqueue_after ) logger.info(f"Successfully retrieved node run details for node ID: {node_id}", x_exosphere_request_id=request_id) diff --git a/state-manager/app/models/node_run_details_models.py b/state-manager/app/models/node_run_details_models.py index a5997cb6..52485dd5 100644 --- a/state-manager/app/models/node_run_details_models.py +++ b/state-manager/app/models/node_run_details_models.py @@ -16,4 +16,12 @@ class NodeRunDetailsResponse(BaseModel): error: Optional[str] = Field(None, description="Error message if any") parents: Dict[str, str] = Field(..., description="Parent node identifiers") created_at: str = Field(..., description="Creation timestamp") - updated_at: str = Field(..., description="Last update timestamp") \ No newline at end of file + updated_at: str = Field(..., description="Last update timestamp") + enqueue_after: Optional[int] = Field( + None, + description=( + "Unix time in milliseconds after which the state becomes eligible to be enqueued. " + "This is the stored eligibility time (set by start delays, retry policies and requeue signals), " + "not a guaranteed execution time: a worker picks the state up on its next poll after this instant." + ), + ) diff --git a/state-manager/tests/unit/controller/test_get_node_run_details.py b/state-manager/tests/unit/controller/test_get_node_run_details.py index e8ca3d79..7d09d70c 100644 --- a/state-manager/tests/unit/controller/test_get_node_run_details.py +++ b/state-manager/tests/unit/controller/test_get_node_run_details.py @@ -170,4 +170,58 @@ async def test_get_node_run_details_empty_timestamps(self): # Verify the result handles None timestamps assert result.created_at == "" - assert result.updated_at == "" \ No newline at end of file + assert result.updated_at == "" + @pytest.mark.asyncio + async def test_get_node_run_details_exposes_stored_enqueue_after(self): + """A delayed node's stored enqueue_after (epoch ms) is passed through untouched""" + namespace = "test_namespace" + graph_name = "test_graph" + run_id = "test_run_id" + node_id = str(ObjectId()) + request_id = "test_request_id" + stored_enqueue_after = 1_800_000_000_000 # 2027-01-15T08:00:00Z, epoch milliseconds + + mock_state = MagicMock() + mock_state.id = ObjectId(node_id) + mock_state.node_name = "delayed_node" + mock_state.identifier = "delayed_identifier" + mock_state.graph_name = graph_name + mock_state.run_id = run_id + mock_state.status = StateStatusEnum.CREATED + mock_state.inputs = {} + mock_state.outputs = {} + mock_state.error = None + mock_state.parents = {} + mock_state.created_at = datetime.now() + mock_state.updated_at = datetime.now() + mock_state.enqueue_after = stored_enqueue_after + + with patch('app.controller.get_node_run_details.State') as mock_state_class: + mock_state_class.find_one = AsyncMock(return_value=mock_state) + + result = await get_node_run_details(namespace, graph_name, run_id, node_id, request_id) + + assert result.status == StateStatusEnum.CREATED + assert result.enqueue_after == stored_enqueue_after + assert result.model_dump()["enqueue_after"] == stored_enqueue_after + + def test_node_run_details_response_enqueue_after_is_optional(self): + """Older records/clients: the field is optional and serialises as null when absent""" + response = NodeRunDetailsResponse( + id=str(ObjectId()), + node_name="n", + identifier="i", + graph_name="g", + run_id="r", + status=StateStatusEnum.SUCCESS, + inputs={}, + outputs={}, + error=None, + parents={}, + created_at="", + updated_at="", + ) + + assert response.enqueue_after is None + assert "enqueue_after" in response.model_dump() + assert response.model_dump()["enqueue_after"] is None