diff --git a/lxc-frontend/app/api/jobs.py b/lxc-frontend/app/api/jobs.py index db527a2..4a431d9 100644 --- a/lxc-frontend/app/api/jobs.py +++ b/lxc-frontend/app/api/jobs.py @@ -120,6 +120,8 @@ async def retry_job( job.status = JobStatus.QUEUED job.progress = 0 job.error = "" + job.step_started_at = None + job.last_heartbeat_at = None session.add(job) session.commit() session.refresh(job) @@ -128,6 +130,21 @@ async def retry_job( return _job_dict(job, user) +@router.get("/{job_id}/diag") +async def diag(job_id: int, session: Session = Depends(get_session), user: User = Depends(current_user)): + """Diagnose-Bündel: Job-Status + letzte Mac-Worker-Logzeilen für genau diesen Job.""" + from app.core.mac_client import MacClient + job = _require_access(session.get(Job, job_id), user) + client = MacClient() + mac_log = await client.fetch_log(job_id, last=200) + return { + "job": _job_dict(job, user), + "step_started_at": job.step_started_at.isoformat() if job.step_started_at else None, + "last_heartbeat_at": job.last_heartbeat_at.isoformat() if job.last_heartbeat_at else None, + "mac_log": mac_log, + } + + @router.get("/{job_id}/download/{kind}") def download(job_id: int, kind: str, session: Session = Depends(get_session), user: User = Depends(current_user)): job = _require_access(session.get(Job, job_id), user) @@ -158,6 +175,8 @@ def _job_dict(j: Job, user: User, owner_username: str | None = None) -> dict: "profile": j.profile or "meeting", "created_at": j.created_at.isoformat(), "updated_at": j.updated_at.isoformat(), + "step_started_at": j.step_started_at.isoformat() if j.step_started_at else None, + "last_heartbeat_at": j.last_heartbeat_at.isoformat() if j.last_heartbeat_at else None, "has": { "transcript": bool(j.transcript_path), "summary": bool(j.summary_path), diff --git a/lxc-frontend/app/core/mac_client.py b/lxc-frontend/app/core/mac_client.py index 74ec779..c5a8c1d 100644 --- a/lxc-frontend/app/core/mac_client.py +++ b/lxc-frontend/app/core/mac_client.py @@ -1,13 +1,62 @@ +import asyncio +import logging from pathlib import Path + import httpx + from .config import settings +log = logging.getLogger("voice-agent.mac_client") + +# Transient-Fehler, bei denen ein einmaliger Retry sinnvoll ist +# (typisch bei Ollama-Reload oder kurzem Mac-Worker-Restart). +_RETRY_EXCEPTIONS = ( + httpx.RemoteProtocolError, + httpx.ConnectError, + httpx.ReadError, +) + class MacClient: def __init__(self, base_url: str | None = None, timeout: int | None = None): self.base_url = (base_url or settings.mac_api_url).rstrip("/") self.timeout = timeout or settings.mac_api_timeout + @staticmethod + def _hdrs(job_id: int | None) -> dict[str, str] | None: + if job_id is None: + return None + return {"X-Job-Id": str(job_id)} + + async def _post_json_retry(self, url: str, payload: dict, job_id: int | None) -> dict: + """JSON-POST mit einmaligem Retry bei transienten Verbindungsabbrüchen.""" + for attempt in (0, 1): + try: + async with httpx.AsyncClient(timeout=self.timeout) as c: + r = await c.post(url, json=payload, headers=self._hdrs(job_id)) + r.raise_for_status() + return r.json() + except _RETRY_EXCEPTIONS as e: + if attempt == 0: + log.warning("[job %s] %s bei %s — Retry in 3s", job_id, type(e).__name__, url) + await asyncio.sleep(3.0) + continue + raise + + async def _post_bytes_retry(self, url: str, payload: dict, job_id: int | None) -> bytes: + for attempt in (0, 1): + try: + async with httpx.AsyncClient(timeout=self.timeout) as c: + r = await c.post(url, json=payload, headers=self._hdrs(job_id)) + r.raise_for_status() + return r.content + except _RETRY_EXCEPTIONS as e: + if attempt == 0: + log.warning("[job %s] %s bei %s — Retry in 3s", job_id, type(e).__name__, url) + await asyncio.sleep(3.0) + continue + raise + async def health(self) -> dict: async with httpx.AsyncClient(timeout=10) as c: r = await c.get(f"{self.base_url}/health") @@ -20,47 +69,67 @@ class MacClient: r.raise_for_status() return r.json().get("profiles", []) - async def transcribe(self, audio_path: Path, language: str = "de") -> dict: + async def fetch_log(self, job_id: int, last: int = 200) -> list[str]: + """Holt die letzten Mac-Worker-Logzeilen für einen Job (für die Diagnose-Anzeige).""" + try: + async with httpx.AsyncClient(timeout=8) as c: + r = await c.get(f"{self.base_url}/api/jobs/{job_id}/log", params={"last": last}) + r.raise_for_status() + return r.json().get("lines", []) + except Exception as e: # noqa: BLE001 + log.warning("fetch_log job=%s failed: %s", job_id, e) + return [] + + async def preload(self, profile: str | None = None, job_id: int | None = None) -> dict: + """Vor-Heizen: Ollama soll das Modell laden, bevor der echte Call läuft.""" + return await self._post_json_retry( + f"{self.base_url}/api/preload", + {"profile": profile}, + job_id, + ) + + async def transcribe(self, audio_path: Path, language: str = "de", job_id: int | None = None) -> dict: + # Kein Retry: Datei-Upload ist nicht idempotent, und die Transkription + # ist der teuerste Schritt — Fehler hier sollen sichtbar bleiben. async with httpx.AsyncClient(timeout=self.timeout) as c: with audio_path.open("rb") as f: files = {"audio": (audio_path.name, f, "application/octet-stream")} data = {"language": language} - r = await c.post(f"{self.base_url}/api/transcribe", files=files, data=data) + r = await c.post( + f"{self.base_url}/api/transcribe", + files=files, + data=data, + headers=self._hdrs(job_id), + ) r.raise_for_status() return r.json() - async def summarize(self, transcript: str, title: str = "", profile: str | None = None) -> dict: - async with httpx.AsyncClient(timeout=self.timeout) as c: - r = await c.post( - f"{self.base_url}/api/summarize", - json={"transcript": transcript, "title": title, "profile": profile}, - ) - r.raise_for_status() - return r.json() + async def summarize(self, transcript: str, title: str = "", profile: str | None = None, + job_id: int | None = None) -> dict: + return await self._post_json_retry( + f"{self.base_url}/api/summarize", + {"transcript": transcript, "title": title, "profile": profile}, + job_id, + ) - async def protocol(self, transcript: str, summary: dict, title: str = "", profile: str | None = None) -> dict: - async with httpx.AsyncClient(timeout=self.timeout) as c: - r = await c.post( - f"{self.base_url}/api/protocol", - json={"transcript": transcript, "summary": summary, "title": title, "profile": profile}, - ) - r.raise_for_status() - return r.json() + async def protocol(self, transcript: str, summary: dict, title: str = "", profile: str | None = None, + job_id: int | None = None) -> dict: + return await self._post_json_retry( + f"{self.base_url}/api/protocol", + {"transcript": transcript, "summary": summary, "title": title, "profile": profile}, + job_id, + ) - async def export_docx(self, data: dict, profile: str | None = None) -> bytes: - async with httpx.AsyncClient(timeout=self.timeout) as c: - r = await c.post( - f"{self.base_url}/api/export/docx", - json={"data": data, "profile": profile}, - ) - r.raise_for_status() - return r.content + async def export_docx(self, data: dict, profile: str | None = None, job_id: int | None = None) -> bytes: + return await self._post_bytes_retry( + f"{self.base_url}/api/export/docx", + {"data": data, "profile": profile}, + job_id, + ) - async def export_pdf(self, data: dict, profile: str | None = None) -> bytes: - async with httpx.AsyncClient(timeout=self.timeout) as c: - r = await c.post( - f"{self.base_url}/api/export/pdf", - json={"data": data, "profile": profile}, - ) - r.raise_for_status() - return r.content + async def export_pdf(self, data: dict, profile: str | None = None, job_id: int | None = None) -> bytes: + return await self._post_bytes_retry( + f"{self.base_url}/api/export/pdf", + {"data": data, "profile": profile}, + job_id, + ) diff --git a/lxc-frontend/app/core/migrate.py b/lxc-frontend/app/core/migrate.py index fc64811..62f41b4 100644 --- a/lxc-frontend/app/core/migrate.py +++ b/lxc-frontend/app/core/migrate.py @@ -46,6 +46,15 @@ def run_migrations() -> None: log.info("Migration: füge job.pdf_path hinzu") conn.execute(text("ALTER TABLE job ADD COLUMN pdf_path VARCHAR DEFAULT ''")) + # 5) Diagnose-Felder: step_started_at, last_heartbeat_at + cols = _existing_columns(conn, "job") + if "step_started_at" not in cols: + log.info("Migration: füge job.step_started_at hinzu") + conn.execute(text("ALTER TABLE job ADD COLUMN step_started_at DATETIME")) + if "last_heartbeat_at" not in cols: + log.info("Migration: füge job.last_heartbeat_at hinzu") + conn.execute(text("ALTER TABLE job ADD COLUMN last_heartbeat_at DATETIME")) + def assign_orphan_jobs_to(user_id: int) -> int: """Weist alle Jobs ohne owner_id dem angegebenen User zu. Gibt Anzahl zurück.""" diff --git a/lxc-frontend/app/models/job.py b/lxc-frontend/app/models/job.py index a3df7f1..6b4eb20 100644 --- a/lxc-frontend/app/models/job.py +++ b/lxc-frontend/app/models/job.py @@ -28,5 +28,7 @@ class Job(SQLModel, table=True): protocol_path: str = "" docx_path: str = "" pdf_path: str = "" + step_started_at: Optional[datetime] = Field(default=None) + last_heartbeat_at: Optional[datetime] = Field(default=None) created_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) diff --git a/lxc-frontend/app/services/pipeline.py b/lxc-frontend/app/services/pipeline.py index 38e7891..11b900d 100644 --- a/lxc-frontend/app/services/pipeline.py +++ b/lxc-frontend/app/services/pipeline.py @@ -1,7 +1,9 @@ +import asyncio import json import logging from datetime import datetime, timezone from pathlib import Path +from typing import Awaitable, TypeVar from sqlmodel import Session @@ -12,6 +14,10 @@ from app.models.job import Job, JobStatus log = logging.getLogger("voice-agent.pipeline") +T = TypeVar("T") + +HEARTBEAT_INTERVAL = 3.0 # Sekunden + def _update(job_id: int, **fields) -> None: with Session(engine) as s: @@ -32,8 +38,51 @@ def _existing(path_str: str) -> Path | None: return p if p.exists() and p.stat().st_size > 0 else None +def _start_step(job_id: int, status: str, progress: int) -> None: + """Setzt status/progress + step_started_at + initialen Heartbeat.""" + now = datetime.now(timezone.utc) + _update( + job_id, + status=status, + progress=progress, + step_started_at=now, + last_heartbeat_at=now, + error="", + ) + + +async def _with_heartbeat(job_id: int, coro: Awaitable[T]) -> T: + """Führt `coro` aus und pingt parallel alle paar Sekunden last_heartbeat_at. + + Damit sieht die UI live, ob ein Schritt noch lebt — auch wenn der Mac + minutenlang an einer Ollama-Generierung arbeitet. + """ + stop = asyncio.Event() + + async def beat() -> None: + while not stop.is_set(): + try: + _update(job_id, last_heartbeat_at=datetime.now(timezone.utc)) + except Exception: # noqa: BLE001 + pass # Heartbeat darf den Job nie killen + try: + await asyncio.wait_for(stop.wait(), timeout=HEARTBEAT_INTERVAL) + except asyncio.TimeoutError: + pass + + beat_task = asyncio.create_task(beat()) + try: + return await coro + finally: + stop.set() + try: + await beat_task + except Exception: # noqa: BLE001 + pass + + async def run_pipeline(job_id: int) -> None: - """Pipeline: transcribe → summarize → protocol → DOCX. + """Pipeline: transcribe → preload → summarize → protocol → DOCX → PDF. Idempotent: bereits vorhandene Zwischenergebnisse werden weiterverwendet, damit ein 'Erneut versuchen' nicht die teure Transkription wiederholt. @@ -64,9 +113,9 @@ async def run_pipeline(job_id: int) -> None: tx = json.loads(existing_tx_path.read_text(encoding="utf-8")) _update(job_id, status=JobStatus.TRANSCRIBING, progress=45) else: - _update(job_id, status=JobStatus.TRANSCRIBING, progress=10, error="") + _start_step(job_id, JobStatus.TRANSCRIBING, 10) log.info("[job %s] transcribe (profile=%s)", job_id, profile) - tx = await client.transcribe(audio_path, language="de") + tx = await _with_heartbeat(job_id, client.transcribe(audio_path, language="de", job_id=job_id)) transcript_path = job_dir / "transcript.json" transcript_path.write_text(json.dumps(tx, ensure_ascii=False, indent=2), encoding="utf-8") _update(job_id, transcript_path=str(transcript_path), progress=45) @@ -75,15 +124,24 @@ async def run_pipeline(job_id: int) -> None: s.get("text", "") for s in tx.get("segments", []) ) - # 2) SUMMARY + # 2) SUMMARY — vorher Preload, damit Ollama nicht mitten drin lädt if existing_sum_path: log.info("[job %s] reuse summary %s", job_id, existing_sum_path.name) summary = json.loads(existing_sum_path.read_text(encoding="utf-8")) _update(job_id, status=JobStatus.SUMMARIZING, progress=70) else: - _update(job_id, status=JobStatus.SUMMARIZING, progress=55) + _start_step(job_id, JobStatus.SUMMARIZING, 50) + log.info("[job %s] preload model", job_id) + try: + await _with_heartbeat(job_id, client.preload(profile=profile, job_id=job_id)) + except Exception as e: # noqa: BLE001 + log.warning("[job %s] preload failed (non-fatal): %s", job_id, e) + _update(job_id, progress=55) log.info("[job %s] summarize", job_id) - summary = await client.summarize(transcript_text, title=title, profile=profile) + summary = await _with_heartbeat( + job_id, + client.summarize(transcript_text, title=title, profile=profile, job_id=job_id), + ) summary_path = job_dir / "summary.json" summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8") _update(job_id, summary_path=str(summary_path), progress=70) @@ -94,9 +152,12 @@ async def run_pipeline(job_id: int) -> None: protocol = json.loads(existing_pro_path.read_text(encoding="utf-8")) _update(job_id, status=JobStatus.PROTOCOLLING, progress=85) else: - _update(job_id, status=JobStatus.PROTOCOLLING, progress=75) + _start_step(job_id, JobStatus.PROTOCOLLING, 75) log.info("[job %s] protocol", job_id) - protocol = await client.protocol(transcript_text, summary, title=title, profile=profile) + protocol = await _with_heartbeat( + job_id, + client.protocol(transcript_text, summary, title=title, profile=profile, job_id=job_id), + ) protocol_path = job_dir / "protocol.json" protocol_path.write_text(json.dumps(protocol, ensure_ascii=False, indent=2), encoding="utf-8") _update(job_id, protocol_path=str(protocol_path), progress=85) @@ -105,9 +166,12 @@ async def run_pipeline(job_id: int) -> None: if existing_dx_path: log.info("[job %s] reuse docx %s", job_id, existing_dx_path.name) else: - _update(job_id, status=JobStatus.EXPORTING, progress=90) + _start_step(job_id, JobStatus.EXPORTING, 90) log.info("[job %s] export docx", job_id) - docx_bytes = await client.export_docx(protocol, profile=profile) + docx_bytes = await _with_heartbeat( + job_id, + client.export_docx(protocol, profile=profile, job_id=job_id), + ) docx_path = job_dir / "protocol.docx" docx_path.write_bytes(docx_bytes) _update(job_id, docx_path=str(docx_path), progress=95) @@ -119,7 +183,10 @@ async def run_pipeline(job_id: int) -> None: else: _update(job_id, status=JobStatus.EXPORTING, progress=96) log.info("[job %s] export pdf", job_id) - pdf_bytes = await client.export_pdf(protocol, profile=profile) + pdf_bytes = await _with_heartbeat( + job_id, + client.export_pdf(protocol, profile=profile, job_id=job_id), + ) pdf_path = job_dir / "protocol.pdf" pdf_path.write_bytes(pdf_bytes) _update(job_id, pdf_path=str(pdf_path), status=JobStatus.DONE, progress=100, error="") diff --git a/lxc-frontend/app/static/app.js b/lxc-frontend/app/static/app.js index 56174b3..40198f1 100644 --- a/lxc-frontend/app/static/app.js +++ b/lxc-frontend/app/static/app.js @@ -76,6 +76,38 @@ function fmtDate(s) { return d.toLocaleString("de-DE", { dateStyle: "short", timeStyle: "medium" }); } +function fmtSecs(sec) { + if (!isFinite(sec) || sec < 0) return "?"; + if (sec < 60) return `${Math.floor(sec)}s`; + const m = Math.floor(sec / 60), s = Math.floor(sec % 60); + return `${m}m ${s}s`; +} + +const RUNNING_STATES = new Set(["queued", "transcribing", "summarizing", "protocolling", "exporting"]); +const DIAG_STATES = new Set([...RUNNING_STATES, "failed"]); + +// Welche Jobs haben aktuell ein offenes Diag-Panel? Über Re-Renders hinweg merken. +const openDiagIds = new Set(); + +function liveTimingHtml(job) { + if (!RUNNING_STATES.has(job.status) || !job.step_started_at) return ""; + const now = Date.now(); + const stepSec = (now - new Date(job.step_started_at).getTime()) / 1000; + let parts = [`läuft seit ${fmtSecs(stepSec)}`]; + if (job.last_heartbeat_at) { + const beatSec = (now - new Date(job.last_heartbeat_at).getTime()) / 1000; + const stale = beatSec > 10; + parts.push(`letzter Ping vor ${fmtSecs(beatSec)}${stale ? " ⚠" : ""}`); + } + return `${parts.join(" · ")}`; +} + +function diagButtonHtml(job) { + if (!DIAG_STATES.has(job.status)) return ""; + const open = openDiagIds.has(String(job.id)); + return ``; +} + async function api(path, opts = {}) { const r = await fetch(path, { credentials: "same-origin", @@ -294,10 +326,13 @@ function row(job) { const retryBtn = job.status !== "done" ? `` : ""; + const timing = liveTimingHtml(job); const statusCell = ` ${escapeHtml(job.status)} + ${timing ? `
${escapeHtml(lines.join("\n"))}`
+ : `