Add job detail API, resource history, and UI tests

Author: haasonsaas

src/nimbus/control_plane/app.py

@@ -1663,6 +1663,18 @@ async def recent_jobs(
         )
         return [JobRecord.model_validate(row) for row in rows]
 
+    @app.get("/api/jobs/{job_id}", response_model=JobRecord)
+    async def job_detail(
+        job_id: int,
+        _: str = Depends(verify_agent_token),
+        session: AsyncSession = Depends(get_session),
+    ) -> JobRecord:
+        REQUEST_COUNTER.inc()
+        row = await db.get_job(session, job_id)
+        if not row:
+            raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Job not found")
+        return JobRecord.model_validate(row)
+
     @app.get("/api/jobs/metadata/summary")
     async def job_metadata_summary(
         key: str,

src/nimbus/host_agent/agent.py

@@ -5,6 +5,7 @@
 import asyncio
 import time
 from datetime import UTC, datetime, timezone, timedelta
+import json
 from contextlib import asynccontextmanager
 from pathlib import Path
 from typing import AsyncIterator, Dict, Optional
@@ -626,6 +627,12 @@ async def _collect_job_metrics_metadata(
             metadata["executor.name"] = executor_name
         if warm_instance_used is not None:
             metadata["executor.warm_instance"] = "true" if warm_instance_used else "false"
+        history = self._resource_tracker.get_usage_history(job_id)
+        if history:
+            try:
+                metadata["resource.timeline"] = json.dumps(history)
+            except (TypeError, ValueError):  # pragma: no cover - defensive
+                LOGGER.debug("Failed to serialize resource timeline", job_id=job_id)
         return metadata
 
     async def _submit_status(

src/nimbus/runners/resource_manager.py

@@ -5,8 +5,9 @@
 import asyncio
 import os
 from dataclasses import dataclass
+from datetime import datetime, timezone
 from pathlib import Path
-from typing import Optional, Dict
+from typing import Optional, Dict, List
 
 import structlog
 
@@ -211,18 +212,19 @@ async def cleanup_job_cgroup(self, job_id: int) -> None:
                           job_id=job_id,
                           error=str(exc))
 
-    async def update_metrics(self, job_id: int, executor_name: str) -> None:
+    async def update_metrics(self, job_id: int, executor_name: str) -> Optional[ResourceUsage]:
         """Update Prometheus metrics for a job."""
         usage = await self.get_job_usage(job_id)
         if not usage:
-            return
-            
+            return None
+
         labels = [str(job_id), executor_name]
         self._cpu_usage_gauge.set(usage.cpu_seconds, labels=labels)
         self._memory_usage_gauge.set(usage.memory_bytes, labels=labels)
-        
+
         # Counters need to track deltas, but for simplicity we'll just use current values
         # In production, we'd track previous values and report deltas
+        return usage
 
 
 class ResourceTracker:
@@ -232,6 +234,7 @@ def __init__(self) -> None:
         self._cgroup_manager = CGroupManager()
         self._tracking_tasks: Dict[int, asyncio.Task] = {}
         self._running = False
+        self._usage_history: Dict[int, List[dict[str, float | str]]] = {}
 
     async def start(self) -> None:
         """Start the resource tracking system."""
@@ -293,9 +296,10 @@ async def stop_job_tracking(self, job_id: int) -> None:
                 await task
             except asyncio.CancelledError:
                 pass
-        
+
         # Clean up cgroup
         await self._cgroup_manager.cleanup_job_cgroup(job_id)
+        self._usage_history.pop(job_id, None)
 
     async def add_process(self, job_id: int, pid: int) -> None:
         """Add a process to job tracking."""
@@ -304,15 +308,29 @@ async def add_process(self, job_id: int, pid: int) -> None:
     async def get_usage(self, job_id: int) -> Optional[ResourceUsage]:
         """Get resource usage for a job."""
         return await self._cgroup_manager.get_job_usage(job_id)
+
+    def get_usage_history(self, job_id: int) -> List[dict[str, float | str]]:
+        return list(self._usage_history.get(job_id, []))
 
     async def _track_job_metrics(self, job_id: int, executor_name: str) -> None:
         """Periodically update metrics for a job."""
         while self._running:
             try:
-                await self._cgroup_manager.update_metrics(job_id, executor_name)
+                usage = await self._cgroup_manager.update_metrics(job_id, executor_name)
+                if usage:
+                    history = self._usage_history.setdefault(job_id, [])
+                    history.append(
+                        {
+                            "ts": datetime.now(timezone.utc).isoformat(),
+                            "cpu_seconds": usage.cpu_seconds,
+                            "memory_bytes": usage.memory_bytes,
+                        }
+                    )
+                    if len(history) > 120:
+                        history.pop(0)
             except Exception as exc:
-                LOGGER.warning("Metrics update failed", 
+                LOGGER.warning("Metrics update failed",
                               job_id=job_id,
                               error=str(exc))
-            
+
             await asyncio.sleep(5)  # Update every 5 seconds

tests/test_resource_manager.py

@@ -397,7 +397,7 @@ async def test_resource_tracker_metrics_tracking():
 
     mock_usage = ResourceUsage(cpu_seconds=1.0, memory_bytes=2048)
 
-    with patch.object(tracker._cgroup_manager, 'update_metrics') as mock_update, \
+    with patch.object(tracker._cgroup_manager, 'update_metrics', return_value=mock_usage) as mock_update, \
          patch('asyncio.sleep') as mock_sleep:
 
         # Mock sleep to raise exception after first iteration to exit loop
@@ -410,6 +410,10 @@ async def test_resource_tracker_metrics_tracking():
 
     # Should be called at least once before cancellation
     mock_update.assert_called_with(123, "test")
+    history = tracker.get_usage_history(123)
+    assert history
+    assert history[0]["cpu_seconds"] == mock_usage.cpu_seconds
+    assert history[0]["memory_bytes"] == mock_usage.memory_bytes
 
 
 @pytest.mark.asyncio