"""Read-only status history for restart-safe background jobs.""" from __future__ import annotations import time import resource import shutil from pathlib import Path from typing import Any from fastapi import APIRouter, HTTPException from fastapi.responses import FileResponse from core.database import background_job_artifact_path, background_job_get, background_job_list from core.constants import CONFIG_DIR from core.registry import TEMP_DIR router = APIRouter() def _directory_bytes(root: Path) -> int: total = 0 try: for path in root.rglob("*"): if path.is_file(): try: total += path.stat().st_size except OSError: pass except OSError: pass return total def _cgroup_bytes(name: str) -> int | None: """Read the container's cgroup v2 memory metric when available.""" try: value = (Path('/sys/fs/cgroup') / name).read_text(encoding='utf-8').strip() if value == 'max': return None return max(0, int(value)) except (OSError, ValueError): return None def _largest_recent_jobs(limit: int = 5) -> list[dict[str, Any]]: rows = [] now = time.time() for job in background_job_list(limit=100): if not job: continue try: started = float(job.get('created') or now) finished = float(job.get('updated') or now) elapsed = max(0.0, finished - started) except (TypeError, ValueError): continue rows.append({ 'kind': str(job.get('kind') or 'job'), 'job_id': str(job.get('job_id') or '')[:12], 'elapsed': round(elapsed, 1), 'status': 'failed' if job.get('error') else ('completed' if job.get('done') else 'running'), }) return sorted(rows, key=lambda row: row['elapsed'], reverse=True)[:max(1, min(10, limit))] def _public_job(job: dict) -> dict: payload = dict(job) payload.pop("fingerprint", None) payload.pop("created", None) payload.pop("updated", None) retry_available = False if payload.get("done") and payload.get("error"): maps = { "transcription": ("routes.stt", "_STT_JOBS"), "voice_design": ("routes.tts", "_VOICE_DESIGN_JOBS"), "casting": ("routes.conversation", "_CAST_JOBS"), "consistent_images": ("routes.conversation", "_consistent_image_jobs"), } module_name, map_name = maps.get(str(payload.get("kind")), (None, None)) if module_name: try: import importlib live = getattr(importlib.import_module(module_name), map_name).get(str(payload.get("job_id")), {}) retry_available = bool(live.get("retry_data") or live.get("retry_source_id")) except Exception: retry_available = False payload["retry_available"] = retry_available result = payload.get("result") if isinstance(result, dict): result = dict(result) if isinstance(result.get("artifact"), dict) and result["artifact"].get("name"): result["artifact"] = {**result["artifact"], "url": f"/api/jobs/{job.get('job_id')}/artifacts/{result['artifact']['name']}"} if isinstance(result.get("artifacts"), list): result["artifacts"] = [ {**item, "url": f"/api/jobs/{job.get('job_id')}/artifacts/{item['name']}"} if isinstance(item, dict) and item.get("name") else item for item in result["artifacts"] ] payload["result"] = result payload["elapsed"] = round(time.time() - float(job.get("created") or time.time()), 1) return payload @router.get("/api/jobs") async def jobs_list(kind: str = "", limit: int = 50): return {"jobs": [_public_job(job) for job in background_job_list(kind, limit) if job]} @router.get("/api/diagnostics") async def diagnostics(): usage = shutil.disk_usage(str(CONFIG_DIR)) artifacts = CONFIG_DIR / 'background-artifacts' return { "server_rss_mb": round(float(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss) / 1024, 1), "config_disk": {"used_bytes": _directory_bytes(CONFIG_DIR), "free_bytes": usage.free}, "temp_disk": {"used_bytes": _directory_bytes(TEMP_DIR)}, "background_jobs": len(background_job_list(limit=100)), "container_memory": { "current_bytes": _cgroup_bytes('memory.current'), "max_bytes": _cgroup_bytes('memory.max'), }, "cache_storage": {"background_artifacts_bytes": _directory_bytes(artifacts)}, "largest_operations": _largest_recent_jobs(), } @router.get("/api/jobs/{job_id}") async def jobs_get(job_id: str): job = background_job_get(job_id) if job is None: raise HTTPException(404, "Unknown or expired job") return _public_job(job) @router.get("/api/jobs/{job_id}/artifacts/{name}") async def jobs_artifact(job_id: str, name: str): path = background_job_artifact_path(job_id, name) if path is None: raise HTTPException(404, "Unknown or expired job artifact") media_type = "application/octet-stream" suffix = path.suffix.lower() if suffix == ".wav": media_type = "audio/wav" elif suffix == ".mp3": media_type = "audio/mpeg" elif suffix == ".png": media_type = "image/png" elif suffix in {".jpg", ".jpeg"}: media_type = "image/jpeg" return FileResponse(path, media_type=media_type, headers={"Cache-Control": "private, no-store"})