WebGUI für die Whisper-Pipeline · Implementation Plan¶
For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (
- [ ]) syntax for tracking.
Goal: Build a browser-based WebGUI on FastAPI + Jinja2 + HTMX + SSE that replaces the Textual TUI as the primary interface, adds Run-Historie, MP4-Vorschau and manuellen Upload-Trigger. Single-User, localhost only.
Architecture: FastAPI server runs pipeline.py and upload_youtube.py as subprocesses (one slot at a time). Stdout is streamed via Server-Sent Events to the browser, plus persisted to per-run logfiles for reconnect. Shared TUI/WebGUI logic (build_command, match_line, audio-path-resolution) is extracted into a new pipeline_core.py. HTMX swaps phase-indicator and progress fragments on phase/progress events; full-page refresh on done/aborted. Mockups in design-concepts/ are ported into webgui/templates/ and webgui/static/style.css.
Tech Stack: Python 3.12 · FastAPI · uvicorn · sse-starlette · Jinja2 · HTMX (vendored) · pytest · httpx (TestClient).
Spec: docs/superpowers/specs/2026-05-20-webgui-design.md (17 sections, approved 2026-05-20)
Mockups: design-concepts/ — Kuro Signal Protocol, Strategist aspect. Class-based components. tweaks.js and the aspect-switcher are NOT ported into production.
File Structure¶
Created¶
| Path | Responsibility |
|---|---|
pipeline_core.py |
Shared helpers (PipelineConfig, build_command, resolve_audio_path, can_diarize, is_pyannote_cached, match_line, StreamEvent dataclasses). Used by TUI and WebGUI. No textual or fastapi deps. |
webgui/__init__.py |
Package marker. |
webgui/app.py |
FastAPI app instance, all route handlers, Jinja2-Environment. Imports from runner/runs/probe/settings. |
webgui/runner.py |
JobRegistry (single slot), Subprocess spawning, log-file writer thread, SSE event queue. |
webgui/runs.py |
list_runs() scanner over output/*/run-state.json, RunSummary dataclass, filter helpers. |
webgui/probe.py |
audio_probe() (ffprobe), disk_free, eta_estimate constants + helper. |
webgui/settings.py |
Load/save ~/.whisper-pipeline-ui.json. |
webgui/templates/base.html |
Topbar + nav + theme wrapper + slot-block. |
webgui/templates/index.html |
Start screen (extends base). |
webgui/templates/runs.html |
Run history (extends base). |
webgui/templates/run_detail.html |
Run detail with 4 status variants in one template, conditional includes. |
webgui/templates/_partials/resume_banner.html |
Resume banner (3 variants). |
webgui/templates/_partials/config_form.html |
Config form. |
webgui/templates/_partials/phase_indicator.html |
Stepper variant of phase indicator. |
webgui/templates/_partials/progress_bar.html |
Overall progress bar fragment. |
webgui/templates/_partials/log_panel.html |
Log panel (initial empty rows, SSE fills). |
webgui/templates/_partials/transcript_preview.html |
First 4 lines of transcript. |
webgui/templates/_partials/metadata_card.html |
YouTube draft (title, description, tags, chapters). |
webgui/templates/_partials/upload_card.html |
Upload-trigger with privacy radio. |
webgui/templates/_partials/runs_row.html |
One row in the runs table. |
webgui/templates/_partials/confirm_run_modal.html |
Confirm-run modal body. |
webgui/static/style.css |
Ported from design-concepts/style.css (mood-tinting layer added). |
webgui/static/app.js |
SSE listener, tail-toggle, modal wiring, fetch wrappers. |
webgui/static/htmx.min.js |
Vendored HTMX (v1.9+). |
webgui/static/fonts/*.woff2 |
Vendored EB Garamond, Space Grotesk, Inter, JetBrains Mono. |
webgui.py |
Entry-point: starts uvicorn, opens browser. |
tests/__init__.py |
Package marker. |
tests/conftest.py |
pytest fixtures (tmp output dir, mock pipeline subprocess, sample run-state.json). |
tests/test_pipeline_core.py |
Tests for match_line, build_command, audio-path. |
tests/test_probe.py |
Tests for audio_probe (real ffprobe on tiny fixture) + helpers. |
tests/test_runs.py |
Tests for list_runs with fixture directory. |
tests/test_runner.py |
Tests for JobRegistry, subprocess spawning (mock pipeline.py). |
tests/test_routes.py |
Tests for FastAPI endpoints via TestClient. |
tests/fixtures/run-states/*.json |
6 fingierte run-state.json files for runs tests. |
tests/fixtures/sample.m4a |
Tiny 1-sec audio file (ffmpeg-generated) for probe tests. |
tests/fixtures/stdout-snippets/*.txt |
Real stdout-line snippets per phase. |
pytest.ini |
pytest config (testpaths, asyncio_mode). |
Modified¶
| Path | Change |
|---|---|
tui_cmd.py |
All functions/dataclasses re-export from pipeline_core. Keep file as thin shim during transition. |
tui_progress.py |
match_line re-exported from pipeline_core. Thin shim. |
requirements.txt |
+ fastapi, uvicorn[standard], jinja2, sse-starlette, python-multipart, pytest, httpx, pytest-asyncio |
README.md |
WebGUI as primary entry, TUI mentioned as fallback. Update alias. |
.gitignore |
+ tests/fixtures/sample.m4a if generated; + ~/.whisper-pipeline-ui.json is outside repo, no entry needed. |
Deleted (later spec, not this plan)¶
tui_app.py, tui_cmd.py, tui_progress.py, tui.py, tui.tcss will be removed in a separate spec once WebGUI runs stable for a week.
Task 1: Extract pipeline_core.py (shared TUI/WebGUI helpers)¶
Files:
- Create: pipeline_core.py
- Create: tests/__init__.py
- Create: tests/conftest.py
- Create: tests/test_pipeline_core.py
- Create: tests/fixtures/stdout-snippets/transcribe.txt
- Create: tests/fixtures/stdout-snippets/render.txt
- Create: pytest.ini
- Modify: tui_cmd.py (turn into re-export shim)
- Modify: tui_progress.py (turn into re-export shim)
- Modify: requirements.txt (+ pytest, pytest-asyncio)
- [ ] Step 1: Add test deps to requirements.txt
Open requirements.txt, append:
Then install: uv pip install -r requirements.txt
- [ ] Step 2: Create pytest.ini
Write pytest.ini:
[pytest]
testpaths = tests
asyncio_mode = auto
python_files = test_*.py
addopts = -ra --strict-markers
- [ ] Step 3: Create tests/init.py and tests/conftest.py
Write tests/__init__.py:
(empty)
Write tests/conftest.py:
"""Shared pytest fixtures."""
import json
from pathlib import Path
import pytest
@pytest.fixture
def fixtures_dir() -> Path:
return Path(__file__).parent / "fixtures"
@pytest.fixture
def stdout_snippet(fixtures_dir: Path):
def _read(name: str) -> list[str]:
path = fixtures_dir / "stdout-snippets" / f"{name}.txt"
return [line for line in path.read_text(encoding="utf-8").splitlines() if line]
return _read
- [ ] Step 4: Create stdout fixtures
Create tests/fixtures/stdout-snippets/transcribe.txt:
────────────────────────────────────────────────────────────
SCHRITT 1: Transkription
────────────────────────────────────────────────────────────
[1/4] Modell laden (large-v3-turbo, de)
[2/4] Transkribieren
[3/4] Wort-Alignment
[4/4] Speaker-Diarization
✓ Transkription fertig
Create tests/fixtures/stdout-snippets/render.txt:
────────────────────────────────────────────────────────────
SCHRITT 3: Video rendern (Remotion)
────────────────────────────────────────────────────────────
Rendering 12.0%
Rendering 24.0%
Rendering 53.4%
Rendering 87.2%
Rendering 100.0%
✓ Render fertig: output/folge-082-dialogue.mp4
Create tests/fixtures/stdout-snippets/meta.txt:
────────────────────────────────────────────────────────────
SCHRITT 2: YouTube-Metadaten generieren
────────────────────────────────────────────────────────────
Metadaten generieren via MLX (qwen2.5-7b-instruct-4bit) …
✓ Metadaten geschrieben
Create tests/fixtures/stdout-snippets/upload.txt:
────────────────────────────────────────────────────────────
SCHRITT 4: YouTube-Upload
────────────────────────────────────────────────────────────
Upload zu YouTube als private …
✓ Hochgeladen: https://youtu.be/abc123
- [ ] Step 5: Write failing tests for
pipeline_core.match_line
Create tests/test_pipeline_core.py:
"""Tests for pipeline_core shared helpers."""
import os
import sys
from pathlib import Path
import pytest
def test_match_line_detects_transcribe_phase(stdout_snippet):
from pipeline_core import match_line
lines = stdout_snippet("transcribe")
events = [match_line(line, 0) for line in lines]
assert any(e and e.step == 1 for e in events)
schritt_event = next(e for e in events if e and "Transkription" in e.label)
assert schritt_event.progress == 2
def test_match_line_extracts_render_progress(stdout_snippet):
from pipeline_core import match_line
lines = stdout_snippet("render")
pct_events = [match_line(line, 3) for line in lines if "Rendering" in line and "%" in line]
assert len(pct_events) >= 4
progresses = [e.progress for e in pct_events if e]
assert progresses == sorted(progresses)
def test_match_line_returns_none_for_unmatched():
from pipeline_core import match_line
assert match_line("random log line that means nothing", 0) is None
def test_build_command_minimal_config():
from pipeline_core import PipelineConfig, build_command
cfg = PipelineConfig(
audio="/tmp/test.m4a", viz="dialogue", language="de",
model="large-v3-turbo", diarize="auto", episode="EP 01",
show_name="Signal", skip_transcribe=False, skip_meta=False,
skip_render=False, skip_upload=False,
)
cmd = build_command(cfg, Path("/repo"))
assert "pipeline.py" in cmd[1]
assert cmd[2] == "/tmp/test.m4a"
assert "--viz" in cmd and "dialogue" in cmd
def test_build_command_skip_flags():
from pipeline_core import PipelineConfig, build_command
cfg = PipelineConfig(
audio="/tmp/test.m4a", viz="dialogue", language="de",
model="large-v3-turbo", diarize="off", episode="EP 01",
show_name="Signal", skip_transcribe=True, skip_meta=False,
skip_render=False, skip_upload=True,
)
cmd = build_command(cfg, Path("/repo"))
assert "--skip-transcribe" in cmd
assert "--skip-upload" in cmd
assert "--no-diarize" in cmd
assert "--skip-meta" not in cmd
def test_resolve_audio_path_absolute():
from pipeline_core import resolve_audio_path
p = resolve_audio_path("/abs/path.m4a", Path("/some/fallback"))
assert p == Path("/abs/path.m4a")
def test_resolve_audio_path_user_home(tmp_path, monkeypatch):
from pipeline_core import resolve_audio_path
monkeypatch.setenv("HOME", str(tmp_path))
audio = tmp_path / "foo.m4a"
audio.write_bytes(b"")
p = resolve_audio_path("~/foo.m4a", Path("/some/fallback"))
assert p == audio
- [ ] Step 6: Run tests, verify they fail
Run: pytest tests/test_pipeline_core.py -v
Expected: FAIL (ModuleNotFoundError or 6+ failures since pipeline_core doesn't exist)
- [ ] Step 7: Write
pipeline_core.py— extract from tui_cmd + tui_progress
Create pipeline_core.py:
"""Shared helpers used by both the TUI and the WebGUI.
Pure-Python module with no UI dependencies (no textual, no fastapi).
Owns the pipeline-config dataclass, command builder, audio-path resolution,
diarization-availability check, and the stdout-line classifier.
"""
import os
import re
import sys
from dataclasses import dataclass
from pathlib import Path
# ── Pipeline-config dataclass ─────────────────────────────────────────────
@dataclass
class PipelineConfig:
audio: str
viz: str
language: str
model: str
diarize: str # "auto" | "off" | "2" | "3" | ...
episode: str
show_name: str
skip_transcribe: bool
skip_meta: bool
skip_render: bool
skip_upload: bool
# ── Diarization availability ──────────────────────────────────────────────
def is_pyannote_cached() -> bool:
hf_home = os.environ.get("HF_HOME", os.path.expanduser("~/.cache/huggingface"))
return os.path.isdir(
os.path.join(hf_home, "hub", "models--pyannote--speaker-diarization-3.1")
)
def can_diarize() -> bool:
return bool(os.environ.get("HF_TOKEN")) or is_pyannote_cached()
# ── Audio-path resolution ─────────────────────────────────────────────────
def resolve_audio_path(audio: str, fallback_dir: Path) -> Path:
"""Expand ~, make absolute. CWD wins; fallback_dir tried if CWD-relative fails."""
p = Path(audio).expanduser()
if p.is_absolute():
return p
cwd_resolved = (Path.cwd() / p).resolve()
if cwd_resolved.exists():
return cwd_resolved
return (fallback_dir / p).resolve()
# ── Command builder ───────────────────────────────────────────────────────
def build_command(config: PipelineConfig, pipeline_dir: Path) -> list[str]:
cmd = [
sys.executable, str(pipeline_dir / "pipeline.py"), config.audio,
"--viz", config.viz, "--language", config.language, "--model", config.model,
"--episode", config.episode, "--show-name", config.show_name,
]
if config.diarize == "off":
cmd.append("--no-diarize")
elif config.diarize.isdigit():
cmd.extend(["--speakers", config.diarize])
if config.skip_transcribe: cmd.append("--skip-transcribe")
if config.skip_meta: cmd.append("--skip-meta")
if config.skip_render: cmd.append("--skip-render")
if config.skip_upload: cmd.append("--skip-upload")
return cmd
# ── Progress event dataclass ──────────────────────────────────────────────
@dataclass
class ProgressEvent:
progress: float # 0–100
label: str # display label
step: int # 1–4
# ── Stdout-line classifier ────────────────────────────────────────────────
_STEP_MARKERS = [
(re.compile(r"SCHRITT 1:"), 1, "Schritt 1/4 · Transkription", 2),
(re.compile(r"\[1/4\] Modell"), 1, "Schritt 1/4 · Modell laden …", 5),
(re.compile(r"\[2/4\] Transkrib"), 1, "Schritt 1/4 · Transkribieren …", 15),
(re.compile(r"\[3/4\] Wort"), 1, "Schritt 1/4 · Wort-Alignment …", 32),
(re.compile(r"\[4/4\]"), 1, "Schritt 1/4 · Speaker-Erkennung …", 39),
(re.compile(r"SCHRITT 2:"), 2, "Schritt 2/4 · Metadaten generieren …", 44),
(re.compile(r"Metadaten generieren via"), 2, "Schritt 2/4 · LLM generiert …", 47),
(re.compile(r"SCHRITT 3:"), 3, "Schritt 3/4 · Video rendern …", 53),
(re.compile(r"SCHRITT 4:"), 4, "Schritt 4/4 · YouTube-Upload …", 96),
]
_RENDER_PCT_RE = re.compile(r"(\d+\.?\d*)%")
def match_line(line: str, current_step: int) -> ProgressEvent | None:
"""Map a single stdout line to a progress event, or None if it doesn't match.
`current_step` is the most recently observed step (1–4) so we can
interpret render-percent lines correctly while a render is in progress.
"""
for pattern, step, label, progress in _STEP_MARKERS:
if pattern.search(line):
return ProgressEvent(progress, label, step)
if current_step == 3:
m = _RENDER_PCT_RE.search(line)
if m:
pct = float(m.group(1))
overall = 53.0 + pct * 42.0 / 100.0
return ProgressEvent(overall, f"Schritt 3/4 · Rendering {pct:.0f}%", 3)
return None
- [ ] Step 8: Convert tui_cmd.py and tui_progress.py to re-export shims
Replace contents of tui_cmd.py:
"""Backwards-compatibility shim — re-exports from pipeline_core."""
from pipeline_core import (
PipelineConfig,
is_pyannote_cached,
can_diarize,
resolve_audio_path,
build_command,
)
__all__ = [
"PipelineConfig", "is_pyannote_cached", "can_diarize",
"resolve_audio_path", "build_command",
]
Replace contents of tui_progress.py:
"""Backwards-compatibility shim — re-exports from pipeline_core."""
from pipeline_core import ProgressEvent, match_line
__all__ = ["ProgressEvent", "match_line"]
- [ ] Step 9: Run tests, verify they pass
Run: pytest tests/test_pipeline_core.py -v
Expected: PASS — 7/7 tests green.
- [ ] Step 10: Run the TUI manually to verify it still launches
Run: python tui.py /tmp/nonexistent.m4a
Expected: TUI starts, shows error about file not found, doesn't crash. Press Ctrl+Q to quit.
- [ ] Step 11: Commit
git add pipeline_core.py tui_cmd.py tui_progress.py requirements.txt pytest.ini tests/
git commit -m "refactor(core): extract pipeline_core.py — shared helpers for TUI + WebGUI
Move PipelineConfig, build_command, resolve_audio_path, can_diarize,
is_pyannote_cached, ProgressEvent, match_line from tui_cmd/tui_progress
into a new pipeline_core.py. The two tui_*.py files become thin re-export
shims for backwards compat. Adds pytest setup with 7 unit tests covering
match_line classification and build_command flag-mapping.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 2: WebGUI Skeleton — Directory, Requirements, FastAPI Stub¶
Files:
- Create: webgui/__init__.py
- Create: webgui/app.py
- Create: webgui/templates/base.html
- Create: webgui/templates/index.html
- Create: webgui/static/style.css (placeholder; full CSS in Task 22)
- Create: webgui/static/app.js (placeholder)
- Modify: requirements.txt
- Create: tests/test_routes.py (first endpoint test)
- [ ] Step 1: Add WebGUI deps to requirements.txt
Append to requirements.txt:
fastapi>=0.110.0
uvicorn[standard]>=0.27.0
jinja2>=3.1.0
sse-starlette>=2.0.0
python-multipart>=0.0.9
Install: uv pip install -r requirements.txt
- [ ] Step 2: Create webgui package marker and minimal app
Write webgui/__init__.py:
(empty)
Write webgui/app.py:
"""FastAPI app for the Whisper-Pipeline WebGUI.
Single-User, localhost only. No auth, no CSRF.
"""
from pathlib import Path
from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
REPO_ROOT = Path(__file__).parent.parent
TEMPLATES_DIR = Path(__file__).parent / "templates"
STATIC_DIR = Path(__file__).parent / "static"
app = FastAPI(title="Whisper-Pipeline WebGUI")
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
templates = Jinja2Templates(directory=TEMPLATES_DIR)
@app.get("/")
async def index(request: Request):
return templates.TemplateResponse(
"index.html",
{"request": request, "page_mood": "neutral"},
)
@app.get("/healthz")
async def healthz():
return {"status": "ok"}
- [ ] Step 3: Create minimal base.html and index.html
Write webgui/templates/base.html:
<!doctype html>
<html lang="en" data-aspect="gunshi" data-theme="dark">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=1200">
<title>{% block title %}Whisper-Pipeline{% endblock %}</title>
<link rel="stylesheet" href="{{ url_for('static', path='style.css') }}">
<script>
// Set theme before first paint to avoid flash
(function() {
const saved = localStorage.getItem('theme');
const prefers = window.matchMedia('(prefers-color-scheme: light)').matches ? 'light' : 'dark';
document.documentElement.dataset.theme = saved || prefers;
})();
</script>
</head>
<body data-page-mood="{{ page_mood | default('neutral') }}">
<div class="app">
<header class="topbar">
<div class="topbar-inner">
<a class="brand" href="/">
<span class="brand-mark"></span>
<span class="brand-name">Whisper<span class="dim">·Pipeline</span></span>
</a>
<nav class="topnav">
<a href="/" {% if request.url.path == '/' %}class="is-active"{% endif %}>Start</a>
<a href="/runs" {% if request.url.path.startswith('/runs') %}class="is-active"{% endif %}>Runs</a>
</nav>
<div class="topbar-meta">
<span class="host-pill"><span class="dot"></span>localhost</span>
</div>
</div>
</header>
<main>
<div class="page">
{% block content %}{% endblock %}
</div>
</main>
</div>
<script src="{{ url_for('static', path='htmx.min.js') }}"></script>
<script src="{{ url_for('static', path='app.js') }}"></script>
</body>
</html>
Write webgui/templates/index.html:
{% extends "base.html" %}
{% block title %}Whisper-Pipeline · Start{% endblock %}
{% block content %}
<h1>Drop an episode. Let it process.</h1>
<p class="lede">Placeholder — full config form lands in Task 12.</p>
{% endblock %}
- [ ] Step 4: Placeholder static files
Write webgui/static/style.css:
/* Placeholder — full CSS in Task 22 */
body { font-family: -apple-system, BlinkMacSystemFont, sans-serif; background: #0b0d11; color: #e8e4d8; margin: 0; padding: 24px; }
.app { max-width: 1180px; margin: 0 auto; }
.topbar-inner { display: flex; gap: 24px; align-items: center; padding: 12px 0; border-bottom: 1px solid #22272f; }
.brand-name { font-weight: 600; }
.topnav { display: flex; gap: 16px; flex: 1; }
.topnav a { color: inherit; text-decoration: none; }
.topnav a.is-active { color: #a878ff; }
.host-pill { font-family: ui-monospace, monospace; font-size: 11px; color: #828a97; }
Write webgui/static/app.js:
// Placeholder — SSE listener + interactivity in later tasks
console.log('Whisper-Pipeline WebGUI');
- [ ] Step 5: Vendor HTMX
Run:
Verify it landed: wc -l webgui/static/htmx.min.js — should be a single minified line, ~50 KB.
- [ ] Step 6: Write tests for
/and/healthz
Create tests/test_routes.py:
"""FastAPI endpoint tests via TestClient."""
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client():
from webgui.app import app
return TestClient(app)
def test_index_returns_html(client):
r = client.get("/")
assert r.status_code == 200
assert "text/html" in r.headers["content-type"]
assert "Whisper" in r.text
def test_healthz_returns_ok(client):
r = client.get("/healthz")
assert r.status_code == 200
assert r.json() == {"status": "ok"}
def test_static_css_served(client):
r = client.get("/static/style.css")
assert r.status_code == 200
assert "text/css" in r.headers["content-type"]
- [ ] Step 7: Run tests
Run: pytest tests/test_routes.py -v
Expected: PASS — 3/3 tests green.
- [ ] Step 8: Start the server manually to sanity-check
Run: uvicorn webgui.app:app --reload --host 127.0.0.1 --port 8765
Open http://localhost:8765 in a browser. Expected: page renders with the placeholder text and basic styling. Press Ctrl+C to stop.
- [ ] Step 9: Commit
git add webgui/ requirements.txt tests/test_routes.py
git commit -m "feat(webgui): scaffold FastAPI app + skeleton templates
Adds webgui/ package with app.py (FastAPI instance, static-mount,
templates), base.html + index.html stubs, placeholder style.css/app.js,
vendored htmx.min.js. Two routes (/ and /healthz) covered by tests via
TestClient.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 3: Audio-Probe — webgui/probe.py + /api/audio/probe route¶
Files:
- Create: webgui/probe.py
- Create: tests/test_probe.py
- Create: tests/fixtures/sample.m4a (generated via ffmpeg)
- Modify: webgui/app.py (add route)
- [ ] Step 1: Generate test audio fixture
Run:
ffmpeg -f lavfi -i "sine=frequency=440:duration=1" -ar 48000 -ac 2 -c:a aac \
tests/fixtures/sample.m4a
Verify: ffprobe -v error -show_format tests/fixtures/sample.m4a — should show duration ≈ 1.0s.
- [ ] Step 2: Write tests for probe helpers
Create tests/test_probe.py:
"""Tests for webgui.probe."""
import json
from pathlib import Path
import pytest
def test_probe_returns_valid_for_existing_audio(fixtures_dir):
from webgui.probe import audio_probe
result = audio_probe(fixtures_dir / "sample.m4a", output_root=fixtures_dir / "_output")
assert result["valid"] is True
assert result["stem"] == "sample"
assert result["format"] in ("m4a", "mp4", "mov,mp4,m4a,3gp,3g2,mj2")
assert result["duration_s"] == pytest.approx(1.0, abs=0.2)
assert result["channels"] == 2
assert result["sample_rate"] == 48000
assert result["resume_state"] is None
assert result["disk_free_bytes"] > 0
def test_probe_returns_invalid_for_missing_file(tmp_path):
from webgui.probe import audio_probe
result = audio_probe(tmp_path / "missing.m4a", output_root=tmp_path)
assert result["valid"] is False
assert result["error"] == "file_not_found"
def test_probe_returns_invalid_for_unsupported_format(tmp_path):
from webgui.probe import audio_probe
bad = tmp_path / "doc.flac"
bad.write_bytes(b"fake")
result = audio_probe(bad, output_root=tmp_path)
assert result["valid"] is False
assert result["error"] == "format_unsupported"
def test_probe_picks_up_resume_state(fixtures_dir, tmp_path):
"""When output/<stem>/run-state.json exists, probe returns it."""
from webgui.probe import audio_probe
stem = "sample"
output_dir = tmp_path / stem
output_dir.mkdir()
state = {
"schema_version": 1, "stem": stem, "audio": str(fixtures_dir / "sample.m4a"),
"phases": {
"transcribe": {"status": "done"},
"meta": {"status": "done"},
"render": {"status": "aborted"},
"upload": {"status": "pending"},
},
}
(output_dir / "run-state.json").write_text(json.dumps(state))
result = audio_probe(fixtures_dir / "sample.m4a", output_root=tmp_path)
assert result["valid"] is True
assert result["resume_state"]["phases"]["render"]["status"] == "aborted"
def test_eta_estimate_scales_with_duration():
from webgui.probe import eta_estimate
e_short = eta_estimate(duration_s=60, model="large-v3-turbo", viz="dialogue")
e_long = eta_estimate(duration_s=600, model="large-v3-turbo", viz="dialogue")
assert e_long["total"] > e_short["total"]
assert e_long["transcribe"] / e_short["transcribe"] == pytest.approx(10.0, abs=0.1)
- [ ] Step 3: Run tests, verify they fail
Run: pytest tests/test_probe.py -v
Expected: FAIL — ModuleNotFoundError for webgui.probe.
- [ ] Step 4: Write
webgui/probe.py
Create webgui/probe.py:
"""Audio-probe + pre-flight helpers (ffprobe, disk-free, ETA-heuristic)."""
import json
import shutil
import subprocess
from pathlib import Path
SUPPORTED_EXTS = {".m4a", ".mp3", ".wav"}
# ETA factors per phase (× audio-duration-seconds, on M-series Mac)
_MODEL_FACTORS = {
"large-v3-turbo": 0.045,
"large-v3": 0.18,
"large-v2": 0.16,
"medium": 0.08,
"small": 0.04,
"base": 0.02,
"tiny": 0.01,
}
_VIZ_FACTORS = {"dialogue": 0.17, "monologue": 0.15}
_META_FIXED_S = 45
_UPLOAD_FACTOR = 0.03 # bandwidth-dependent — heuristic only
def audio_probe(audio_path: Path, output_root: Path) -> dict:
"""Probe audio file. Returns dict with 'valid' bool + audio/resume metadata."""
if not audio_path.exists():
return {"valid": False, "error": "file_not_found"}
if audio_path.suffix.lower() not in SUPPORTED_EXTS:
return {"valid": False, "error": "format_unsupported"}
try:
ff = subprocess.run(
["ffprobe", "-v", "error", "-show_format", "-show_streams",
"-of", "json", str(audio_path)],
capture_output=True, text=True, check=True, timeout=5,
)
except (subprocess.CalledProcessError, subprocess.TimeoutExpired, FileNotFoundError):
return {"valid": False, "error": "ffprobe_failed"}
probe = json.loads(ff.stdout)
fmt = probe.get("format", {})
audio_stream = next(
(s for s in probe.get("streams", []) if s.get("codec_type") == "audio"),
{},
)
stem = audio_path.stem
output_dir = output_root / stem
state_file = output_dir / "run-state.json"
resume_state = None
if state_file.exists():
try:
resume_state = json.loads(state_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
resume_state = None
duration_s = float(fmt.get("duration", 0))
return {
"valid": True,
"stem": stem,
"size_bytes": audio_path.stat().st_size,
"duration_s": duration_s,
"format": fmt.get("format_name", "unknown"),
"channels": int(audio_stream.get("channels", 0)),
"sample_rate": int(audio_stream.get("sample_rate", 0)),
"resume_state": resume_state,
"disk_free_bytes": shutil.disk_usage(output_root.parent if not output_root.exists() else output_root).free,
"eta_estimate_s": eta_estimate(duration_s, model="large-v3-turbo", viz="dialogue"),
}
def eta_estimate(duration_s: float, model: str, viz: str) -> dict[str, float]:
"""Heuristic ETA per phase. Returns floats — callers can round() for display."""
transcribe = duration_s * _MODEL_FACTORS.get(model, 0.045)
meta = float(_META_FIXED_S)
render = duration_s * _VIZ_FACTORS.get(viz, 0.17)
upload = duration_s * _UPLOAD_FACTOR
return {
"transcribe": transcribe,
"meta": meta,
"render": render,
"upload": upload,
"total": transcribe + meta + render + upload,
}
- [ ] Step 5: Run tests, verify they pass
Run: pytest tests/test_probe.py -v
Expected: PASS — 5/5 tests green.
- [ ] Step 6: Add
/api/audio/proberoute to webgui/app.py
Open webgui/app.py. Add at the top (alongside the other imports):
Add constants near the existing ones:
Add a new route handler before index:
class AudioProbeRequest(BaseModel):
path: str
@app.post("/api/audio/probe")
async def api_audio_probe(req: AudioProbeRequest):
from pipeline_core import resolve_audio_path
audio_path = resolve_audio_path(req.path, REPO_ROOT)
return audio_probe(audio_path, OUTPUT_ROOT)
- [ ] Step 7: Write endpoint test
Append to tests/test_routes.py:
def test_api_audio_probe_with_valid_file(client, fixtures_dir):
r = client.post("/api/audio/probe", json={"path": str(fixtures_dir / "sample.m4a")})
assert r.status_code == 200
body = r.json()
assert body["valid"] is True
assert body["stem"] == "sample"
def test_api_audio_probe_with_missing_file(client, tmp_path):
r = client.post("/api/audio/probe", json={"path": str(tmp_path / "nope.m4a")})
assert r.status_code == 200
assert r.json()["valid"] is False
assert r.json()["error"] == "file_not_found"
- [ ] Step 8: Run all route tests
Run: pytest tests/test_routes.py -v
Expected: PASS — 5/5 tests green.
- [ ] Step 9: Commit
git add webgui/probe.py webgui/app.py tests/test_probe.py tests/test_routes.py tests/fixtures/sample.m4a
git commit -m "feat(webgui): audio probe + /api/audio/probe endpoint
webgui/probe.py wraps ffprobe, reads resume-state.json if present, and
computes ETA per phase via the heuristic from spec §9. Pulls the audio
path through pipeline_core.resolve_audio_path so the WebGUI accepts the
same path syntax as the TUI.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 4: Run-Historie Backend — webgui/runs.py + list_runs¶
Files:
- Create: webgui/runs.py
- Create: tests/test_runs.py
- Create: tests/fixtures/run-states/done.json
- Create: tests/fixtures/run-states/running.json
- Create: tests/fixtures/run-states/aborted.json
- Create: tests/fixtures/run-states/skipped-upload.json
- Create: tests/fixtures/run-states/partial-resume.json
- Create: tests/fixtures/run-states/empty.json
- [ ] Step 1: Create fixture run-state.json files
Create tests/fixtures/run-states/done.json:
{
"schema_version": 1,
"audio": "/tmp/folge-081.m4a",
"stem": "folge-081",
"started_at": "2026-05-18T09:14:00Z",
"updated_at": "2026-05-18T09:36:18Z",
"config": {"show_name": "Notizen aus der Kammer", "episode": "Folge 81", "language": "de", "model": "large-v3-turbo", "viz_type": "dialogue", "diarize": true, "num_speakers": null, "privacy": "private"},
"phases": {
"transcribe": {"status": "done", "started_at": "2026-05-18T09:14:00Z", "finished_at": "2026-05-18T09:17:14Z"},
"meta": {"status": "done", "started_at": "2026-05-18T09:17:14Z", "finished_at": "2026-05-18T09:17:56Z", "title": "Warum Aufmerksamkeit keine Ressource ist"},
"render": {"status": "done", "started_at": "2026-05-18T09:17:56Z", "finished_at": "2026-05-18T09:30:11Z", "size_mb": 198.0, "output": "folge-081-dialogue.mp4"},
"upload": {"status": "done", "started_at": "2026-05-18T09:30:11Z", "finished_at": "2026-05-18T09:36:18Z", "url": "https://youtu.be/qC7w-2hL", "video_id": "qC7w-2hL"}
}
}
Create tests/fixtures/run-states/running.json:
{
"schema_version": 1,
"audio": "/tmp/folge-082.m4a",
"stem": "folge-082",
"started_at": "2026-05-20T14:02:11Z",
"updated_at": "2026-05-20T14:09:53Z",
"config": {"show_name": "Notizen aus der Kammer", "episode": "Folge 82", "language": "de", "model": "large-v3-turbo", "viz_type": "dialogue"},
"phases": {
"transcribe": {"status": "done", "finished_at": "2026-05-20T14:05:25Z"},
"meta": {"status": "done", "finished_at": "2026-05-20T14:06:08Z"},
"render": {"status": "running", "started_at": "2026-05-20T14:06:08Z"},
"upload": {"status": "pending"}
}
}
Create tests/fixtures/run-states/aborted.json:
{
"schema_version": 1,
"audio": "/tmp/probefolge.m4a",
"stem": "probefolge",
"started_at": "2026-05-16T21:08:00Z",
"updated_at": "2026-05-16T21:19:44Z",
"config": {"show_name": "Pilot", "episode": "Test", "viz_type": "dialogue"},
"phases": {
"transcribe": {"status": "done", "finished_at": "2026-05-16T21:10:18Z"},
"meta": {"status": "done", "finished_at": "2026-05-16T21:11:02Z"},
"render": {"status": "aborted", "started_at": "2026-05-16T21:11:02Z", "finished_at": "2026-05-16T21:19:44Z", "error": "RuntimeError: render exit 1"},
"upload": {"status": "skipped"}
}
}
Create tests/fixtures/run-states/skipped-upload.json:
{
"schema_version": 1,
"audio": "/tmp/interview.m4a",
"stem": "interview",
"started_at": "2026-05-12T22:01:00Z",
"updated_at": "2026-05-12T22:26:09Z",
"config": {"show_name": "Notizen aus der Kammer", "episode": "Spezial", "viz_type": "dialogue"},
"phases": {
"transcribe": {"status": "done"}, "meta": {"status": "done"},
"render": {"status": "done", "output": "interview-dialogue.mp4"},
"upload": {"status": "skipped"}
}
}
Create tests/fixtures/run-states/partial-resume.json:
{
"schema_version": 1,
"audio": "/tmp/folge-079.m4a",
"stem": "folge-079",
"started_at": "2026-05-09T11:24:00Z",
"updated_at": "2026-05-09T11:25:14Z",
"config": {"show_name": "Notizen aus der Kammer", "episode": "Folge 79"},
"phases": {
"transcribe": {"status": "done"}, "meta": {"status": "running"},
"render": {"status": "pending"}, "upload": {"status": "pending"}
}
}
Create tests/fixtures/run-states/empty.json:
- [ ] Step 2: Write tests for list_runs
Create tests/test_runs.py:
"""Tests for webgui.runs."""
import json
import shutil
from pathlib import Path
import pytest
@pytest.fixture
def output_root(tmp_path, fixtures_dir):
"""Build a tmp output/ tree from the run-states fixtures."""
root = tmp_path / "output"
root.mkdir()
for src in (fixtures_dir / "run-states").glob("*.json"):
stem = src.stem
if stem == "empty":
(root / stem).mkdir()
(root / stem / "run-state.json").write_text(src.read_text())
continue
(root / stem).mkdir()
shutil.copy(src, root / stem / "run-state.json")
return root
def test_list_runs_returns_all_valid(output_root):
from webgui.runs import list_runs
runs = list_runs(output_root)
valid_stems = {r.stem for r in runs}
assert "done" in valid_stems
assert "running" in valid_stems
assert "aborted" in valid_stems
assert "empty" not in valid_stems # invalid schema → skipped
def test_list_runs_sorted_newest_first(output_root):
from webgui.runs import list_runs
runs = list_runs(output_root)
timestamps = [r.updated_at for r in runs]
assert timestamps == sorted(timestamps, reverse=True)
def test_run_summary_includes_phase_statuses(output_root):
from webgui.runs import list_runs
runs = list_runs(output_root)
done = next(r for r in runs if r.stem == "done")
assert done.phases["transcribe"] == "done"
assert done.phases["upload"] == "done"
assert done.youtube_url == "https://youtu.be/qC7w-2hL"
def test_aborted_run_has_no_youtube_url(output_root):
from webgui.runs import list_runs
runs = list_runs(output_root)
aborted = next(r for r in runs if r.stem == "aborted")
assert aborted.youtube_url is None
assert aborted.phases["render"] == "aborted"
def test_filter_runs_done(output_root):
from webgui.runs import list_runs, filter_runs
all_runs = list_runs(output_root)
done = filter_runs(all_runs, "done")
assert {r.stem for r in done} == {"done"}
def test_filter_runs_aborted(output_root):
from webgui.runs import list_runs, filter_runs
aborted = filter_runs(list_runs(output_root), "aborted")
assert {r.stem for r in aborted} == {"aborted"}
def test_filter_runs_not_uploaded(output_root):
from webgui.runs import list_runs, filter_runs
not_uploaded = filter_runs(list_runs(output_root), "not-uploaded")
stems = {r.stem for r in not_uploaded}
# done has upload.done, aborted has upload.skipped, skipped-upload has upload.skipped
# so only running + partial-resume are not-yet-uploaded
assert "running" in stems
assert "partial-resume" in stems
assert "done" not in stems
- [ ] Step 3: Run tests, verify they fail
Run: pytest tests/test_runs.py -v
Expected: FAIL — ModuleNotFoundError for webgui.runs.
- [ ] Step 4: Write
webgui/runs.py
Create webgui/runs.py:
"""Run-Historie scanner — lists past runs from output/*/run-state.json."""
import hashlib
import json
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Literal
PhaseName = Literal["transcribe", "meta", "render", "upload"]
PhaseStatus = Literal["pending", "running", "done", "aborted", "skipped"]
PHASES: tuple[PhaseName, ...] = ("transcribe", "meta", "render", "upload")
@dataclass
class RunSummary:
stem: str
audio_path: str
started_at: datetime
updated_at: datetime
show_name: str
episode: str
phases: dict[PhaseName, PhaseStatus]
youtube_url: str | None
video_path: str | None
duration_s: int | None # total pipeline duration, only if all phases done/skipped
waveform_seed: int # for procedural Run-Card thumbnail
raw: dict = field(repr=False)
def list_runs(output_root: Path) -> list[RunSummary]:
"""Scan output_root/*/run-state.json. Returns runs sorted by updated_at desc."""
if not output_root.exists():
return []
out = []
for state_file in output_root.glob("*/run-state.json"):
try:
data = json.loads(state_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
continue
try:
out.append(_summarize(data))
except (KeyError, ValueError, TypeError):
continue
out.sort(key=lambda r: r.updated_at, reverse=True)
return out
def _summarize(data: dict) -> RunSummary:
stem = data["stem"]
phases_raw = data["phases"]
phases: dict[PhaseName, PhaseStatus] = {
p: phases_raw.get(p, {}).get("status", "pending") for p in PHASES
}
upload = phases_raw.get("upload", {})
youtube_url = upload.get("url")
video_path = phases_raw.get("render", {}).get("output")
# Total duration only when fully done (or skipped)
duration_s = None
if all(phases[p] in ("done", "skipped") for p in PHASES):
try:
started = datetime.fromisoformat(data["started_at"].replace("Z", "+00:00"))
ended_str = phases_raw.get("upload", {}).get("finished_at") \
or phases_raw.get("render", {}).get("finished_at") \
or data["updated_at"]
ended = datetime.fromisoformat(ended_str.replace("Z", "+00:00"))
duration_s = int((ended - started).total_seconds())
except (KeyError, ValueError):
duration_s = None
cfg = data.get("config", {})
return RunSummary(
stem=stem,
audio_path=data.get("audio", ""),
started_at=datetime.fromisoformat(data["started_at"].replace("Z", "+00:00")),
updated_at=datetime.fromisoformat(data["updated_at"].replace("Z", "+00:00")),
show_name=cfg.get("show_name", ""),
episode=cfg.get("episode", ""),
phases=phases,
youtube_url=youtube_url,
video_path=video_path,
duration_s=duration_s,
waveform_seed=int(hashlib.md5(stem.encode()).hexdigest()[:8], 16),
raw=data,
)
def filter_runs(runs: list[RunSummary], filt: str) -> list[RunSummary]:
"""Apply UI filter chip to a list of RunSummary."""
if filt in ("all", ""):
return runs
if filt == "done":
return [r for r in runs if all(r.phases[p] in ("done", "skipped") for p in PHASES)
and r.phases["upload"] == "done"]
if filt == "aborted":
return [r for r in runs if any(r.phases[p] == "aborted" for p in PHASES)]
if filt == "unfinished":
return [r for r in runs if any(r.phases[p] in ("pending", "running") for p in PHASES)
and not any(r.phases[p] == "aborted" for p in PHASES)]
if filt == "not-uploaded":
return [r for r in runs if r.phases["upload"] != "done"]
return runs
- [ ] Step 5: Run tests, verify they pass
Run: pytest tests/test_runs.py -v
Expected: PASS — 7/7 tests green.
- [ ] Step 6: Commit
git add webgui/runs.py tests/test_runs.py tests/fixtures/run-states/
git commit -m "feat(webgui): run-historie scanner + filter helpers
webgui/runs.py iterates output/*/run-state.json, builds RunSummary
dataclasses (stem, phase-statuses, youtube_url, video_path, duration),
sorts newest-first. filter_runs() implements the four UI chips
(done / aborted / unfinished / not-uploaded). 6 fixture run-states cover
all status combinations.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 5: Run-Historie Frontend — /runs route + runs.html¶
Files:
- Modify: webgui/app.py (add /runs route)
- Create: webgui/templates/runs.html
- Create: webgui/templates/_partials/runs_row.html
- Modify: tests/test_routes.py (add /runs tests)
- [ ] Step 1: Write test for /runs
Append to tests/test_routes.py:
@pytest.fixture
def populated_output(tmp_path, fixtures_dir, monkeypatch):
"""Re-point webgui.app.OUTPUT_ROOT to a tmp tree with fixture runs."""
import shutil
out = tmp_path / "output"
out.mkdir()
for src in (fixtures_dir / "run-states").glob("*.json"):
stem = src.stem
if stem == "empty":
continue
(out / stem).mkdir()
shutil.copy(src, out / stem / "run-state.json")
from webgui import app as app_mod
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", out)
return out
def test_runs_returns_html(client, populated_output):
r = client.get("/runs")
assert r.status_code == 200
assert "Past runs" in r.text
# We rendered three of the fixture stems
assert "folge-081" in r.text
assert "folge-082" in r.text
assert "probefolge" in r.text
def test_runs_empty_state_when_no_runs(client, tmp_path, monkeypatch):
from webgui import app as app_mod
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "empty")
r = client.get("/runs")
assert r.status_code == 200
assert "No signal" in r.text or "No runs" in r.text
def test_runs_filter_done(client, populated_output):
r = client.get("/runs?filter=done")
assert r.status_code == 200
assert "folge-081" in r.text # done
assert "folge-082" not in r.text # still running
assert "probefolge" not in r.text # aborted
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_routes.py::test_runs_returns_html -v
Expected: FAIL — /runs route doesn't exist (404).
- [ ] Step 3: Add /runs route to webgui/app.py
Add to webgui/app.py:
Add route handler:
@app.get("/runs")
async def runs_view(request: Request, filter: str = "all"):
all_runs = list_runs(OUTPUT_ROOT)
runs = filter_runs(all_runs, filter)
return templates.TemplateResponse(
"runs.html",
{
"request": request,
"runs": runs,
"active_filter": filter,
"total_count": len(all_runs),
"page_mood": "neutral",
},
)
- [ ] Step 4: Create runs.html (using
design-concepts/runs.htmlas reference)
Create webgui/templates/runs.html:
{% extends "base.html" %}
{% block title %}Whisper-Pipeline · Past runs{% endblock %}
{% block content %}
<div class="page-header">
<div>
<div class="eyebrow">Output · {{ total_count }} runs</div>
<h1>Past runs</h1>
<p class="lede">Everything the chamber has seen. Newest first. Click to open.</p>
</div>
<div class="row-end">
<a class="btn btn--primary" href="/">+ New run</a>
</div>
</div>
{% if runs %}
<div class="runs-toolbar">
{% for chip, label in [("all","All"), ("done","Done"), ("aborted","Aborted"), ("unfinished","Unfinished"), ("not-uploaded","Not uploaded")] %}
<a class="filter-chip {% if active_filter == chip %}is-active{% endif %}" href="?filter={{ chip }}">{{ label }}</a>
{% endfor %}
</div>
<div class="runs-table">
<div class="runs-row is-header">
<span>Date</span>
<span>Stem · Episode</span>
<span>Phases</span>
<span>YouTube</span>
<span style="text-align:right;">Time</span>
<span></span>
</div>
{% for run in runs %}
{% include "_partials/runs_row.html" %}
{% endfor %}
</div>
{% else %}
<div class="empty">
<h3>No signal. The chamber is quiet.</h3>
<p>No runs yet. Drop a podcast file on the start screen and the pipeline will leave a trail here.</p>
<a class="btn btn--primary" href="/">+ Start your first run</a>
</div>
{% endif %}
{% endblock %}
- [ ] Step 5: Create _partials/runs_row.html
Create webgui/templates/_partials/runs_row.html:
<a class="runs-row" href="/runs/{{ run.stem }}">
<span class="date">{{ run.started_at.strftime("%Y-%m-%d · %H:%M") }}</span>
<span style="min-width:0;">
<div class="row" style="gap:12px;">
<span class="thumb {{ 'wave' if run.phases.render == 'done' else 'empty' }}"></span>
<span style="min-width:0;">
<span class="stem">{{ run.stem }}</span>
<div class="stem-meta">{{ run.show_name }}{% if run.episode %} · {{ run.episode }}{% endif %}</div>
</span>
</div>
</span>
<span>
<span class="phases--compact">
{% for phase_name in ["transcribe", "meta", "render", "upload"] %}
{% set status = run.phases[phase_name] %}
<span class="phase-pill" data-status="{{ status }}">
{% if status == "done" %}✓
{% elif status == "running" %}⟳
{% elif status == "aborted" %}✗
{% elif status == "skipped" %}·
{% else %}○{% endif %}
</span>
{% endfor %}
</span>
</span>
<span class="mono youtube-cell">
{% if run.youtube_url %}{{ run.youtube_url.replace("https://", "") }}
{% elif run.phases.upload == "skipped" and run.phases.render == "done" %}Local only
{% elif run.phases | selectattr("==", "aborted") | list %}Aborted
{% elif run.phases.render == "running" or run.phases.transcribe == "running" or run.phases.meta == "running" %}Running…
{% else %}—{% endif %}
</span>
<span class="duration">
{% if run.duration_s %}{{ "%d:%02d" | format(run.duration_s // 60, run.duration_s % 60) }}
{% else %}—{% endif %}
</span>
<span class="chev">→</span>
</a>
- [ ] Step 6: Run tests, verify they pass
Run: pytest tests/test_routes.py -v
Expected: PASS — all green.
- [ ] Step 7: Manual sanity-check
Run server: uvicorn webgui.app:app --reload --port 8765
Create a couple of fake runs:
mkdir -p output/test-run-1
cp tests/fixtures/run-states/done.json output/test-run-1/run-state.json
mkdir -p output/test-run-2
cp tests/fixtures/run-states/aborted.json output/test-run-2/run-state.json
Visit http://localhost:8765/runs — both rows should appear, sorted by date. Visit http://localhost:8765/runs?filter=done — only one row.
Clean up: rm -rf output/test-run-1 output/test-run-2
- [ ] Step 8: Commit
git add webgui/app.py webgui/templates/runs.html webgui/templates/_partials/runs_row.html tests/test_routes.py
git commit -m "feat(webgui): /runs route + runs.html + runs_row partial
GET /runs lists past runs (newest-first) with filter-chip toolbar (all /
done / aborted / unfinished / not-uploaded). Empty-state when no runs.
Filter changes via query param; current chip gets is-active class.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 6: JobRegistry — single-slot in-memory job tracker¶
Files:
- Modify: webgui/runner.py (created here)
- Create: tests/test_runner.py
- [ ] Step 1: Write tests for JobRegistry
Create tests/test_runner.py:
"""Tests for webgui.runner — JobRegistry + subprocess spawning."""
import asyncio
from pathlib import Path
import pytest
def test_registry_starts_empty():
from webgui.runner import JobRegistry
reg = JobRegistry()
assert reg.current is None
def test_registry_claim_succeeds_when_empty():
from webgui.runner import JobRegistry, ActiveJob
reg = JobRegistry()
job = ActiveJob(
stem="folge-082", audio_path=Path("/tmp/a.m4a"),
output_dir=Path("/tmp/out"), process=None, log_file=Path("/tmp/log"),
kind="pipeline",
)
claimed = reg.try_claim(job)
assert claimed is True
assert reg.current is job
assert reg.current.stem == "folge-082"
def test_registry_claim_fails_when_busy():
from webgui.runner import JobRegistry, ActiveJob
reg = JobRegistry()
first = ActiveJob(stem="first", audio_path=Path("/a"), output_dir=Path("/b"),
process=None, log_file=Path("/c"), kind="pipeline")
second = ActiveJob(stem="second", audio_path=Path("/a"), output_dir=Path("/b"),
process=None, log_file=Path("/c"), kind="pipeline")
reg.try_claim(first)
assert reg.try_claim(second) is False
assert reg.current.stem == "first"
def test_registry_release_frees_slot():
from webgui.runner import JobRegistry, ActiveJob
reg = JobRegistry()
job = ActiveJob(stem="x", audio_path=Path("/a"), output_dir=Path("/b"),
process=None, log_file=Path("/c"), kind="pipeline")
reg.try_claim(job)
reg.release(job)
assert reg.current is None
def test_registry_release_ignores_wrong_job():
"""Releasing a job that doesn't match current should not free the slot."""
from webgui.runner import JobRegistry, ActiveJob
reg = JobRegistry()
real = ActiveJob(stem="real", audio_path=Path("/a"), output_dir=Path("/b"),
process=None, log_file=Path("/c"), kind="pipeline")
other = ActiveJob(stem="other", audio_path=Path("/a"), output_dir=Path("/b"),
process=None, log_file=Path("/c"), kind="pipeline")
reg.try_claim(real)
reg.release(other)
assert reg.current is real
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_runner.py -v
Expected: FAIL — ModuleNotFoundError.
- [ ] Step 3: Create webgui/runner.py with JobRegistry only
Create webgui/runner.py:
"""Subprocess lifecycle + SSE event queue.
JobRegistry is in-memory, single-slot. Both pipeline runs and uploads
share this slot — only one subprocess can run at a time.
"""
import asyncio
import subprocess
import threading
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Literal
JobKind = Literal["pipeline", "upload"]
@dataclass
class StreamEvent:
"""One SSE event to push to subscribers."""
type: Literal["log", "phase", "progress", "done"]
data: dict[str, Any]
seq: int = 0
@dataclass
class ActiveJob:
stem: str
audio_path: Path
output_dir: Path
process: subprocess.Popen | None
log_file: Path
kind: JobKind
started_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
seq: int = 0
queue: asyncio.Queue[StreamEvent] | None = None
class JobRegistry:
"""Single-slot job registry. Thread-safe."""
def __init__(self) -> None:
self._slot: ActiveJob | None = None
self._lock = threading.Lock()
@property
def current(self) -> ActiveJob | None:
return self._slot
def try_claim(self, job: ActiveJob) -> bool:
with self._lock:
if self._slot is not None:
return False
self._slot = job
return True
def release(self, job: ActiveJob) -> None:
with self._lock:
if self._slot is job:
self._slot = None
# Module-level singleton (FastAPI app gets it via Depends or import)
registry = JobRegistry()
- [ ] Step 4: Run tests, verify they pass
Run: pytest tests/test_runner.py -v
Expected: PASS — 5/5 green.
- [ ] Step 5: Commit
git add webgui/runner.py tests/test_runner.py
git commit -m "feat(webgui): JobRegistry — single-slot subprocess tracker
In-memory, thread-safe slot for either a pipeline or upload subprocess.
try_claim returns False when busy. release only frees the slot if the
caller owns it.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 7: Subprocess Runner — spawn_pipeline + log capture¶
Files:
- Modify: webgui/runner.py (add spawn function + reader thread)
- Modify: tests/test_runner.py (subprocess tests with mock pipeline.py)
- Create: tests/fixtures/mock_pipeline.py (plays back pre-recorded stdout)
- [ ] Step 1: Create mock_pipeline.py fixture
Create tests/fixtures/mock_pipeline.py:
#!/usr/bin/env python3
"""Mock pipeline.py — emits a fixed stdout script then exits 0.
Used by tests that need a subprocess to attach to without running the real pipeline.
"""
import sys
import time
SCRIPT = [
("─" * 60, 0),
("SCHRITT 1: Transkription", 0),
("─" * 60, 0),
("[1/4] Modell laden (large-v3-turbo, de)", 0.05),
("[2/4] Transkribieren", 0.05),
("[3/4] Wort-Alignment", 0.05),
("✓ Transkription fertig", 0.05),
("", 0),
("─" * 60, 0),
("SCHRITT 3: Video rendern (Remotion)", 0.05),
("─" * 60, 0),
("Rendering 50.0%", 0.05),
("Rendering 100.0%", 0.05),
("✓ Render fertig", 0.05),
]
for line, delay in SCRIPT:
print(line, flush=True)
if delay:
time.sleep(delay)
sys.exit(0)
- [ ] Step 2: Write tests for spawn_pipeline
Append to tests/test_runner.py:
import sys
def test_spawn_pipeline_writes_to_logfile_and_queue(tmp_path):
"""End-to-end: spawn mock pipeline, verify log file and queue get populated."""
from webgui.runner import spawn_pipeline, JobRegistry, ActiveJob
reg = JobRegistry()
output_dir = tmp_path / "stem-x"
output_dir.mkdir()
log_file = output_dir / "run-test.log"
mock_cmd = [sys.executable, str(Path("tests/fixtures/mock_pipeline.py"))]
job = spawn_pipeline(
cmd=mock_cmd,
stem="stem-x",
audio_path=Path("/tmp/a.m4a"),
output_dir=output_dir,
log_file=log_file,
registry=reg,
)
# Wait for subprocess to exit
job.process.wait(timeout=10)
# Logfile must contain the mock script
log_contents = log_file.read_text()
assert "SCHRITT 1" in log_contents
assert "Rendering 100.0%" in log_contents
# Registry must release slot once process exited (reader thread joins)
import time
for _ in range(30):
if reg.current is None:
break
time.sleep(0.1)
assert reg.current is None
def test_spawn_pipeline_fails_when_slot_busy(tmp_path):
from webgui.runner import spawn_pipeline, JobRegistry, ActiveJob
reg = JobRegistry()
existing = ActiveJob(
stem="busy", audio_path=Path("/a"), output_dir=Path("/b"),
process=None, log_file=Path("/c"), kind="pipeline",
)
reg.try_claim(existing)
with pytest.raises(RuntimeError) as exc:
spawn_pipeline(
cmd=[sys.executable, "-c", "pass"],
stem="new", audio_path=Path("/tmp/a.m4a"),
output_dir=tmp_path, log_file=tmp_path / "x.log",
registry=reg,
)
assert "busy" in str(exc.value).lower() or "slot" in str(exc.value).lower()
- [ ] Step 3: Run tests, verify failure
Run: pytest tests/test_runner.py::test_spawn_pipeline_writes_to_logfile_and_queue -v
Expected: FAIL — spawn_pipeline not yet defined.
- [ ] Step 4: Add spawn_pipeline + reader thread to webgui/runner.py
Append to webgui/runner.py:
def spawn_pipeline(
cmd: list[str],
stem: str,
audio_path: Path,
output_dir: Path,
log_file: Path,
registry: JobRegistry,
kind: JobKind = "pipeline",
on_line: callable | None = None,
) -> ActiveJob:
"""Spawn a pipeline subprocess. Returns the ActiveJob.
Raises RuntimeError if the registry slot is already taken.
A background thread reads stdout, writes each line to log_file, and
invokes on_line(line) for every non-empty line (default: no-op).
Once the subprocess exits, the registry is released.
"""
job = ActiveJob(
stem=stem, audio_path=audio_path, output_dir=output_dir,
process=None, log_file=log_file, kind=kind,
)
if not registry.try_claim(job):
raise RuntimeError(f"Slot is busy with {registry.current.stem!r}")
output_dir.mkdir(parents=True, exist_ok=True)
log_file.parent.mkdir(parents=True, exist_ok=True)
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
text=True,
bufsize=1,
)
job.process = proc
def _reader():
try:
with log_file.open("w", encoding="utf-8", buffering=1) as f:
f.write(f"# Pipeline-Run started {datetime.now().isoformat()}\n")
f.write(f"# Command: {' '.join(cmd)}\n")
f.write("# " + ("─" * 60) + "\n\n")
for line in proc.stdout:
line = line.rstrip("\n")
f.write(line + "\n")
if on_line and line:
on_line(line)
proc.wait()
finally:
registry.release(job)
threading.Thread(target=_reader, daemon=True, name=f"run-{stem}").start()
return job
- [ ] Step 5: Run tests
Run: pytest tests/test_runner.py -v
Expected: PASS — 7/7 green.
- [ ] Step 6: Commit
git add webgui/runner.py tests/test_runner.py tests/fixtures/mock_pipeline.py
git commit -m "feat(webgui): spawn_pipeline — subprocess + reader thread
spawn_pipeline starts a pipeline.py subprocess, claims the registry slot,
and runs a daemon reader thread that mirrors stdout to a per-run logfile.
Caller can pass on_line for fan-out to SSE queue (used in next task).
Slot is auto-released once the process exits.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 8: SSE Basic Stream — /runs/{stem}/stream (log events only)¶
Files:
- Modify: webgui/app.py (add /api/runs POST + /runs/{stem}/stream GET)
- Modify: webgui/runner.py (add queue per ActiveJob)
- Modify: tests/test_routes.py
- [ ] Step 1: Write tests for POST /api/runs and SSE stream
Append to tests/test_routes.py:
import sys
def test_api_runs_starts_pipeline_mock(client, populated_output, tmp_path, monkeypatch):
"""POST /api/runs spawns a subprocess; client receives 303 to /runs/{stem}."""
from webgui import app as app_mod, runner
# Reset registry for test isolation
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
# Patch build_command to return our mock
monkeypatch.setattr(
app_mod,
"build_command",
lambda cfg, _dir: [sys.executable, str(Path("tests/fixtures/mock_pipeline.py"))],
)
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
body = {
"audio": str(populated_output.parent.parent / "fixtures" / "sample.m4a"),
"viz": "dialogue", "language": "de", "model": "large-v3-turbo",
"diarize": "off", "episode": "EP X", "show_name": "Test",
"skip_transcribe": False, "skip_meta": False,
"skip_render": False, "skip_upload": True,
}
r = client.post("/api/runs", json=body, follow_redirects=False)
assert r.status_code == 303
assert r.headers["location"].startswith("/runs/")
def test_api_runs_returns_409_when_busy(client, tmp_path, monkeypatch):
from webgui import app as app_mod, runner
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
# Pre-claim the slot
runner.registry.try_claim(runner.ActiveJob(
stem="busy-stem", audio_path=Path("/x"), output_dir=Path("/y"),
process=None, log_file=Path("/z"), kind="pipeline",
))
body = {
"audio": "/whatever.m4a", "viz": "dialogue", "language": "de",
"model": "large-v3-turbo", "diarize": "off", "episode": "X",
"show_name": "X", "skip_transcribe": False, "skip_meta": False,
"skip_render": False, "skip_upload": True,
}
r = client.post("/api/runs", json=body)
assert r.status_code == 409
assert "busy-stem" in r.text
- [ ] Step 2: Run test, verify failure (404 since route doesn't exist)
Run: pytest tests/test_routes.py::test_api_runs_starts_pipeline_mock -v
Expected: FAIL.
- [ ] Step 3: Add ActiveJob queue + on_line wiring to spawn_pipeline
Modify webgui/runner.py — change spawn_pipeline to attach an asyncio.Queue to the job and feed log events:
import json
def spawn_pipeline(
cmd: list[str],
stem: str,
audio_path: Path,
output_dir: Path,
log_file: Path,
registry: JobRegistry,
kind: JobKind = "pipeline",
) -> ActiveJob:
"""Spawn a pipeline subprocess and attach an SSE queue to the ActiveJob."""
job = ActiveJob(
stem=stem, audio_path=audio_path, output_dir=output_dir,
process=None, log_file=log_file, kind=kind,
queue=asyncio.Queue(),
)
if not registry.try_claim(job):
raise RuntimeError(f"Slot is busy with {registry.current.stem!r}")
output_dir.mkdir(parents=True, exist_ok=True)
log_file.parent.mkdir(parents=True, exist_ok=True)
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, bufsize=1,
)
job.process = proc
# The reader runs in a thread but pushes events into the asyncio queue
# via thread-safe call_soon_threadsafe. We capture the loop now.
loop = asyncio.get_running_loop()
def _put(event: StreamEvent) -> None:
loop.call_soon_threadsafe(job.queue.put_nowait, event)
def _reader():
try:
with log_file.open("w", encoding="utf-8", buffering=1) as f:
f.write(f"# Pipeline-Run started {datetime.now().isoformat()}\n")
f.write(f"# Command: {' '.join(cmd)}\n")
f.write("# " + ("─" * 60) + "\n\n")
for line in proc.stdout:
line = line.rstrip("\n")
f.write(line + "\n")
if line:
job.seq += 1
_put(StreamEvent(
type="log",
seq=job.seq,
data={"msg": line, "level": _classify_level(line)},
))
proc.wait()
job.seq += 1
_put(StreamEvent(type="done", seq=job.seq,
data={"exit_code": proc.returncode, "kind": kind}))
finally:
registry.release(job)
threading.Thread(target=_reader, daemon=True, name=f"run-{stem}").start()
return job
def _classify_level(line: str) -> str:
"""Trivial log-level inference for styling."""
low = line.lower()
if "✓" in line or "fertig" in low or "complete" in low:
return "success"
if "✗" in line or "fehler" in low or "error" in low or "traceback" in low:
return "error"
if "warn" in low or "slow" in low:
return "warn"
if line.startswith("─") or line.startswith("SCHRITT") or low.startswith("phase"):
return "phase"
return "info"
- [ ] Step 4: Add /api/runs POST + /runs/{stem}/stream GET to webgui/app.py
Add imports to webgui/app.py:
from datetime import datetime
from fastapi.responses import RedirectResponse
from sse_starlette.sse import EventSourceResponse
import json
from pipeline_core import PipelineConfig, build_command
from .runner import registry, spawn_pipeline, JobRegistry
Replace the module-level registry import with a re-export so monkeypatch can override:
Add request model and routes:
class RunRequest(BaseModel):
audio: str
viz: str
language: str
model: str
diarize: str
episode: str
show_name: str
skip_transcribe: bool
skip_meta: bool
skip_render: bool
skip_upload: bool
@app.post("/api/runs")
async def api_create_run(req: RunRequest):
if registry.current is not None:
from fastapi import HTTPException
raise HTTPException(
status_code=409,
detail={
"error": "slot_busy",
"stem": registry.current.stem,
"kind": registry.current.kind,
},
)
from pipeline_core import resolve_audio_path
audio_path = resolve_audio_path(req.audio, REPO_ROOT)
stem = audio_path.stem
output_dir = OUTPUT_ROOT / stem
ts = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
log_file = output_dir / f"run-{ts}.log"
cfg = PipelineConfig(
audio=str(audio_path), viz=req.viz, language=req.language, model=req.model,
diarize=req.diarize, episode=req.episode, show_name=req.show_name,
skip_transcribe=req.skip_transcribe, skip_meta=req.skip_meta,
skip_render=req.skip_render, skip_upload=req.skip_upload,
)
cmd = build_command(cfg, REPO_ROOT)
spawn_pipeline(
cmd=cmd, stem=stem, audio_path=audio_path,
output_dir=output_dir, log_file=log_file, registry=registry,
)
return RedirectResponse(url=f"/runs/{stem}", status_code=303)
@app.get("/runs/{stem}/stream")
async def runs_stream(stem: str):
"""SSE stream of log events for the active run (if any)."""
job = registry.current
if job is None or job.stem != stem:
async def _empty():
yield {"event": "done", "data": json.dumps({"exit_code": 0, "kind": "static"})}
return EventSourceResponse(_empty())
async def event_gen():
while True:
event: StreamEvent = await job.queue.get()
yield {
"id": str(event.seq),
"event": event.type,
"data": json.dumps(event.data),
}
if event.type == "done":
break
return EventSourceResponse(event_gen())
Make sure from webgui.runner import StreamEvent is at the top.
- [ ] Step 5: Run tests
Run: pytest tests/test_routes.py -v
Expected: PASS — including the new tests. Note that test_api_runs_starts_pipeline_mock will leave a subprocess running; assertions only check the redirect.
- [ ] Step 6: Commit
git add webgui/app.py webgui/runner.py tests/test_routes.py
git commit -m "feat(webgui): POST /api/runs + SSE /runs/{stem}/stream (log events)
POST /api/runs spawns pipeline subprocess via webgui.runner.spawn_pipeline,
redirects to /runs/{stem}. 409 with active stem if slot is busy.
GET /runs/{stem}/stream opens an SSE channel: log events come from
the per-job asyncio.Queue, populated by the subprocess reader thread.
Closes on event: done.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 9: SSE Reconnect — replay from persisted logfile with Last-Event-ID¶
Files:
- Modify: webgui/runner.py (helper replay_logfile)
- Modify: webgui/app.py (extend /stream handler)
- Modify: tests/test_routes.py
- [ ] Step 1: Write tests for reconnect
Append to tests/test_routes.py:
def test_stream_replays_logfile_when_no_active_job(client, tmp_path, monkeypatch):
"""If a run has finished, /stream replays the logfile then closes."""
from webgui import app as app_mod, runner
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
stem = "finished-run"
output_dir = tmp_path / "output" / stem
output_dir.mkdir(parents=True)
log_file = output_dir / "run-test.log"
log_file.write_text(
"# Pipeline-Run started …\n"
"# Command: …\n"
"# ────────\n"
"\n"
"SCHRITT 1: Transkription\n"
"✓ Transkription fertig\n"
)
state = output_dir / "run-state.json"
state.write_text('{"schema_version":1,"stem":"finished-run","audio":"/x","started_at":"2026-05-20T10:00:00Z","updated_at":"2026-05-20T10:30:00Z","phases":{"transcribe":{"status":"done"},"meta":{"status":"done"},"render":{"status":"done"},"upload":{"status":"done"}}}')
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
# SSE responses are streaming; use the iter-lines helper of httpx
with client.stream("GET", f"/runs/{stem}/stream") as r:
assert r.status_code == 200
body = "".join(r.iter_text())
assert "SCHRITT 1" in body
assert "event: done" in body
def test_stream_with_last_event_id_skips_already_seen_lines(client, tmp_path, monkeypatch):
from webgui import app as app_mod, runner
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
stem = "midrun"
output_dir = tmp_path / "output" / stem
output_dir.mkdir(parents=True)
log_file = output_dir / "run-test.log"
log_file.write_text("\n".join(f"line {i}" for i in range(1, 11)) + "\n")
state = output_dir / "run-state.json"
state.write_text('{"schema_version":1,"stem":"midrun","audio":"/x","started_at":"2026-05-20T10:00:00Z","updated_at":"2026-05-20T10:30:00Z","phases":{"transcribe":{"status":"done"},"meta":{"status":"done"},"render":{"status":"done"},"upload":{"status":"done"}}}')
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
with client.stream("GET", f"/runs/{stem}/stream", headers={"Last-Event-ID": "5"}) as r:
body = "".join(r.iter_text())
assert "line 7" in body
assert "line 4" not in body
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_routes.py::test_stream_replays_logfile_when_no_active_job -v
Expected: FAIL — the current empty-generator returns no log content.
- [ ] Step 3: Add
replay_logfileto webgui/runner.py
Append to webgui/runner.py:
def latest_logfile(output_dir: Path) -> Path | None:
"""Return the newest run-*.log in output_dir, or None."""
logs = sorted(output_dir.glob("run-*.log"), key=lambda p: p.stat().st_mtime)
return logs[-1] if logs else None
def replay_logfile(log_file: Path, start_seq: int = 0):
"""Yield StreamEvent log lines from a logfile, starting after start_seq."""
if not log_file.exists():
return
seq = 0
for raw_line in log_file.read_text(encoding="utf-8").splitlines():
# skip header comments and blank lines
if not raw_line or raw_line.startswith("#"):
continue
seq += 1
if seq <= start_seq:
continue
yield StreamEvent(
type="log",
seq=seq,
data={"msg": raw_line, "level": _classify_level(raw_line)},
)
- [ ] Step 4: Extend /stream handler to use replay
Modify webgui/app.py — replace the runs_stream handler:
@app.get("/runs/{stem}/stream")
async def runs_stream(stem: str, request: Request):
"""SSE stream — live from active job, or replay from persisted logfile."""
last_id_str = request.headers.get("Last-Event-ID", "0")
try:
last_seq = int(last_id_str)
except ValueError:
last_seq = 0
job = registry.current
async def replay_then_done():
output_dir = OUTPUT_ROOT / stem
log_file = latest_logfile(output_dir)
if log_file is not None:
for event in replay_logfile(log_file, start_seq=last_seq):
yield {
"id": str(event.seq),
"event": event.type,
"data": json.dumps(event.data),
}
yield {
"event": "done",
"data": json.dumps({"exit_code": 0, "kind": "replay"}),
}
if job is None or job.stem != stem:
return EventSourceResponse(replay_then_done())
async def live_then_drain():
# First, replay any backlog from the logfile up to current job.seq
log_file = job.log_file
if log_file.exists():
for event in replay_logfile(log_file, start_seq=last_seq):
if event.seq > job.seq:
break
yield {
"id": str(event.seq),
"event": event.type,
"data": json.dumps(event.data),
}
# Then drain the live queue
while True:
event: StreamEvent = await job.queue.get()
if event.seq <= last_seq:
continue
yield {
"id": str(event.seq),
"event": event.type,
"data": json.dumps(event.data),
}
if event.type == "done":
break
return EventSourceResponse(live_then_drain())
Add to the top imports:
- [ ] Step 5: Run tests
Run: pytest tests/test_routes.py -v
Expected: PASS — both new reconnect tests green.
- [ ] Step 6: Commit
git add webgui/runner.py webgui/app.py tests/test_routes.py
git commit -m "feat(webgui): SSE reconnect — Last-Event-ID + logfile replay
Two paths in /runs/{stem}/stream:
• No active job: stream from persisted logfile, skip events <= Last-Event-ID,
send 'done' and close.
• Active job: backfill from logfile up to current seq, then drain live queue.
Reconnects after browser-reload or transient connection drops no longer
duplicate or drop log lines.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 10: Phase + Progress events via pipeline_core.match_line¶
Files:
- Modify: webgui/runner.py (extend _reader with match_line)
- Modify: webgui/app.py (add /runs/{stem}/phases + /progress fragment routes)
- Create: webgui/templates/_partials/phase_indicator.html
- Create: webgui/templates/_partials/progress_bar.html
- [ ] Step 1: Replace
spawn_pipeline._readerwith the match-line-aware version
Open webgui/runner.py. Find the inner function _reader defined inside spawn_pipeline (added in Task 8). Replace its for-loop body with the version below that runs each stdout line through pipeline_core.match_line. The full replacement of the for-loop:
from pipeline_core import match_line
current_step = 0
for line in proc.stdout:
line = line.rstrip("\n")
f.write(line + "\n")
if not line:
continue
job.seq += 1
_put(StreamEvent(
type="log",
seq=job.seq,
data={"msg": line, "level": _classify_level(line)},
))
evt = match_line(line, current_step)
if evt:
current_step = evt.step
job.seq += 1
_put(StreamEvent(
type="phase",
seq=job.seq,
data={"step": evt.step, "label": evt.label},
))
job.seq += 1
_put(StreamEvent(
type="progress",
seq=job.seq,
data={"value": evt.progress, "label": evt.label},
))
- [ ] Step 2: Create phase_indicator.html partial
Create webgui/templates/_partials/phase_indicator.html:
<div data-phases-wrapper class="phases">
<div class="phases--stepper">
{% for phase_name in ["transcribe", "meta", "render", "upload"] %}
{% set status = phases.get(phase_name, {}).get("status", "pending") %}
{% set label_map = {"transcribe":"Transcribe","meta":"Metadata","render":"Render","upload":"Upload"} %}
{% set status_meta = {
"done":"Done", "running":"Running",
"aborted":"Aborted", "skipped":"Skipped", "pending":"Pending"
} %}
{% set glyph = {"done":"✓","running":"⟳","aborted":"✗","skipped":"·","pending":"○"} %}
<div class="phase" data-status="{{ status }}">
<div class="glyph">{% if status == 'running' %}<span class="ico-running">⟳</span>{% else %}{{ glyph[status] }}{% endif %}</div>
<div class="phase-meta">
<span class="phase-name">{{ label_map[phase_name] }}</span>
<span class="phase-status">{{ status_meta[status] }}</span>
</div>
</div>
{% endfor %}
</div>
</div>
- [ ] Step 3: Create progress_bar.html partial
Create webgui/templates/_partials/progress_bar.html:
<div class="progress" id="progress" hx-swap-oob="true">
<div class="progress-track"><div class="progress-fill" style="--progress: {{ progress.value | default(0) }}%"></div></div>
<div class="progress-meta">
<span>{{ progress.label | default("Bereit") }}</span>
<span>{{ progress.value | default(0) | round(0) }} %</span>
</div>
</div>
- [ ] Step 4: Add /runs/{stem}/phases + /progress fragment routes to webgui/app.py
def _load_state(stem: str) -> dict:
state_file = OUTPUT_ROOT / stem / "run-state.json"
if not state_file.exists():
return {"phases": {p: {"status": "pending"} for p in ("transcribe","meta","render","upload")}}
try:
return json.loads(state_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return {"phases": {p: {"status": "pending"} for p in ("transcribe","meta","render","upload")}}
@app.get("/runs/{stem}/phases", response_class=HTMLResponse)
async def runs_phases_fragment(stem: str, request: Request):
state = _load_state(stem)
return templates.TemplateResponse(
"_partials/phase_indicator.html",
{"request": request, "phases": state.get("phases", {})},
)
@app.get("/runs/{stem}/progress", response_class=HTMLResponse)
async def runs_progress_fragment(stem: str, request: Request, value: int = 0, label: str = ""):
return templates.TemplateResponse(
"_partials/progress_bar.html",
{"request": request, "progress": {"value": value, "label": label}},
)
Add from fastapi.responses import HTMLResponse at the top.
- [ ] Step 5: Update app.js to react to phase/progress events
Replace webgui/static/app.js:
// SSE listener + HTMX-OOB swaps for phase/progress fragments
function startEventSource(stem) {
const url = `/runs/${stem}/stream`;
const es = new EventSource(url);
const logEl = document.getElementById('log');
es.addEventListener('log', (e) => {
const data = JSON.parse(e.data);
if (!logEl) return;
const row = document.createElement('div');
row.className = `row ${data.level || ''}`;
row.innerHTML = `<span class="mark">·</span><span class="msg"></span>`;
row.querySelector('.msg').textContent = data.msg;
logEl.appendChild(row);
const tail = document.getElementById('tail-toggle');
if (!tail || tail.getAttribute('aria-pressed') === 'true') {
logEl.scrollTop = logEl.scrollHeight;
}
});
es.addEventListener('phase', (e) => {
htmx.ajax('GET', `/runs/${stem}/phases`, { target: '[data-phases-wrapper]', swap: 'outerHTML' });
});
es.addEventListener('progress', (e) => {
const data = JSON.parse(e.data);
const params = new URLSearchParams({ value: Math.round(data.value), label: data.label });
htmx.ajax('GET', `/runs/${stem}/progress?${params}`, { target: '#progress', swap: 'outerHTML' });
});
es.addEventListener('done', (e) => {
es.close();
// Trigger full reload to swap to ready-to-upload / done / aborted variant
htmx.ajax('GET', `/runs/${stem}`, { target: 'body', swap: 'innerHTML' });
});
}
window.startEventSource = startEventSource;
// Tail-toggle wiring
document.addEventListener('click', (e) => {
if (e.target.id !== 'tail-toggle') return;
const pressed = e.target.getAttribute('aria-pressed') === 'true';
e.target.setAttribute('aria-pressed', pressed ? 'false' : 'true');
});
- [ ] Step 6: Run tests + manual sanity
Run: pytest -v
Expected: all green.
Manually start the server and a mock run to verify SSE wiring works end-to-end. Skipped here — the integration test in Task 7 already covers the queue.
- [ ] Step 7: Commit
git add webgui/runner.py webgui/app.py webgui/templates/_partials/ webgui/static/app.js
git commit -m "feat(webgui): phase + progress events via match_line — HTMX OOB swaps
Reader thread runs every stdout line through pipeline_core.match_line.
On hit, emits phase + progress events on the SSE queue. Frontend uses
HTMX ajax to swap [data-phases-wrapper] and #progress with freshly
rendered fragments. On 'done' event, frontend triggers a full reload
of /runs/{stem} (variant switches automatically based on run-state.json).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 11: Run-Detail Template — 4 variants in one template¶
Files:
- Create: webgui/templates/run_detail.html
- Modify: webgui/app.py (add /runs/{stem} GET)
- Modify: tests/test_routes.py
- [ ] Step 1: Write tests for /runs/{stem}
Append to tests/test_routes.py:
def test_runs_detail_running_variant(client, populated_output, monkeypatch):
"""Running state shows live-log section and Abort button."""
r = client.get("/runs/folge-082")
assert r.status_code == 200
assert "Running" in r.text
assert "stdout · pipeline" in r.text.lower() or "log" in r.text.lower()
def test_runs_detail_done_variant(client, populated_output):
"""Done state shows YouTube URL."""
r = client.get("/runs/folge-081")
assert r.status_code == 200
assert "youtu.be" in r.text or "youtube" in r.text.lower()
def test_runs_detail_aborted_variant(client, populated_output):
"""Aborted state shows error message."""
r = client.get("/runs/probefolge")
assert r.status_code == 200
assert "aborted" in r.text.lower() or "error" in r.text.lower()
def test_runs_detail_404_unknown_stem(client, populated_output):
r = client.get("/runs/does-not-exist")
assert r.status_code == 404
- [ ] Step 2: Run tests, verify they fail
Run: pytest tests/test_routes.py::test_runs_detail_running_variant -v
Expected: FAIL.
- [ ] Step 3: Add /runs/{stem} GET route
Add to webgui/app.py:
@app.get("/runs/{stem}", response_class=HTMLResponse)
async def runs_detail(stem: str, request: Request):
from fastapi import HTTPException
state_file = OUTPUT_ROOT / stem / "run-state.json"
if not state_file.exists():
raise HTTPException(status_code=404, detail="Run not found")
state = _load_state(stem)
phases = state.get("phases", {})
# Determine variant
if any(phases.get(p, {}).get("status") == "aborted" for p in ("transcribe","meta","render","upload")):
variant = "aborted"
page_mood = "error"
elif phases.get("upload", {}).get("status") == "done":
variant = "done"
page_mood = "success"
elif phases.get("render", {}).get("status") == "done":
variant = "ready-to-upload"
page_mood = "neutral"
elif any(phases.get(p, {}).get("status") == "running" for p in ("transcribe","meta","render","upload")):
variant = "running"
page_mood = "neutral"
else:
# Phases pending — partial resume
variant = "running"
page_mood = "warning"
# Is this run currently active in registry?
active = registry.current is not None and registry.current.stem == stem
return templates.TemplateResponse(
"run_detail.html",
{
"request": request,
"stem": stem,
"state": state,
"phases": phases,
"variant": variant,
"page_mood": page_mood,
"active": active,
},
)
- [ ] Step 4: Create run_detail.html
Create webgui/templates/run_detail.html:
{% extends "base.html" %}
{% block title %}Run · {{ stem }} · {{ variant }}{% endblock %}
{% block content %}
<header class="run-header">
<div class="left">
<div class="crumbs"><a href="/runs">Runs</a> · <span>{{ stem }}</span></div>
<h1><span class="accent-mark">⟩</span>{{ stem }}</h1>
<p class="muted mono" style="margin-top:8px; font-size: 13px;">
{{ state.audio | default("") }} · started {{ state.started_at }}
</p>
</div>
<div class="timer">
<small>Elapsed</small>
<span id="elapsed">{% if active %}00:00:00{% else %}—{% endif %}</span>
</div>
</header>
{% include "_partials/phase_indicator.html" %}
<div id="progress" class="progress" style="margin-top: 16px;">
<div class="progress-track"><div class="progress-fill" style="--progress: 0%"></div></div>
<div class="progress-meta"><span>—</span><span>—</span></div>
</div>
<div class="run-grid" style="margin-top: 32px;">
<div class="stack">
{% if variant in ("running", "ready-to-upload", "done", "aborted") %}
{% include "_partials/log_panel.html" %}
{% endif %}
{% if variant == "ready-to-upload" %}
{% include "_partials/upload_card.html" %}
{% endif %}
{% if variant == "aborted" %}
<section class="card" data-tone="error">
<header class="card-header"><span class="caps">Run aborted</span></header>
<div class="card-body">
<p>The pipeline stopped before reaching upload. Check the live-log tail.</p>
{% for phase_name, phase in phases.items() %}
{% if phase.get("status") == "aborted" %}
<p class="mono dim">Phase: {{ phase_name }} — {{ phase.get("error", "") }}</p>
{% endif %}
{% endfor %}
</div>
</section>
{% endif %}
{% if variant == "done" and state.phases.upload.url %}
<section class="card" data-tone="success">
<header class="card-header"><span class="caps">Uploaded to YouTube</span></header>
<div class="card-body">
<a class="btn btn--primary" href="{{ state.phases.upload.url }}" target="_blank">Open on YouTube ↗</a>
<p class="mono dim" style="margin-top: 12px;">{{ state.phases.upload.url }}</p>
</div>
</section>
{% endif %}
</div>
<aside class="stack">
<section class="card card--padded">
<div class="caps" style="margin-bottom: 12px;">Run details</div>
<dl class="kv">
<dt>Stem</dt><dd>{{ stem }}</dd>
<dt>Visualizer</dt><dd>{{ state.config.viz_type | default("—") }}</dd>
<dt>Model</dt><dd>{{ state.config.model | default("—") }}</dd>
<dt>Language</dt><dd>{{ state.config.language | default("—") }}</dd>
<dt>Output</dt><dd>output/{{ stem }}</dd>
</dl>
</section>
</aside>
</div>
{% if active %}
<script>
window.startEventSource("{{ stem }}");
</script>
{% endif %}
{% endblock %}
- [ ] Step 5: Create log_panel.html partial (minimal — full styling in Task 22)
Create webgui/templates/_partials/log_panel.html:
<section class="logpanel">
<header class="logpanel-head">
<span class="dots"><span></span><span></span><span></span></span>
<span class="title">stdout · pipeline</span>
<span class="meta">
<button id="tail-toggle" class="tail" type="button" aria-pressed="true" title="Auto-scroll to newest line">Tail</button>
<span class="live">live · sse</span>
</span>
</header>
<div class="logpanel-body" id="log">
</div>
</section>
- [ ] Step 6: Stub upload_card.html (full version in Task 15)
Create webgui/templates/_partials/upload_card.html:
<section class="upload-card">
<div style="min-width:0;">
<div class="caps" style="color: var(--accent);">↑ Final phase — upload to YouTube</div>
<p class="lede">Render done — upload trigger lands in Task 15.</p>
</div>
</section>
- [ ] Step 7: Run tests
Run: pytest tests/test_routes.py -v
Expected: PASS — 4 new tests green.
- [ ] Step 8: Commit
git add webgui/app.py webgui/templates/run_detail.html webgui/templates/_partials/log_panel.html webgui/templates/_partials/upload_card.html tests/test_routes.py
git commit -m "feat(webgui): /runs/{stem} detail view + 4-variant template
GET /runs/{stem} renders run_detail.html with variant inferred from
run-state.json phases: running · ready-to-upload · done · aborted.
Variant drives page_mood (neutral/success/error/warning), header
sections (upload card, error card, done card). Live SSE only kicks in
when the run is the registry's current job.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 12: Index Page — Config form, Resume banner, Confirm-run modal¶
Files:
- Modify: webgui/templates/index.html (full implementation)
- Create: webgui/templates/_partials/config_form.html
- Create: webgui/templates/_partials/resume_banner.html
- Create: webgui/templates/_partials/confirm_run_modal.html
- Modify: webgui/static/app.js (audio-probe + modal wiring)
Drag-and-Drop note¶
The dropzone in this task is visual-only by default. Browsers cannot reveal an OS path from a drop event for security reasons. The drop handler in app.js (Step 5 below) reads file.name and uses it as a starting hint inside the input — the user still has to type or paste the absolute path. That matches the spec's "Pfad-Hint aus Drop-Event" wording.
- [ ] Step 1: Replace index.html with full structure
Replace webgui/templates/index.html:
{% extends "base.html" %}
{% block title %}Whisper-Pipeline · Start{% endblock %}
{% block content %}
<div class="page-header">
<div>
<div class="eyebrow">Local · Single user</div>
<h1>Drop an episode. Let it process.</h1>
<p class="lede">Transcription · Metadata · Render · Upload — four phases, one queue.</p>
</div>
<div class="row-end">
<a class="btn btn--ghost" href="/runs">View past runs →</a>
</div>
</div>
<div class="section-title"><span class="caps">01 · Source</span><span class="rule"></span></div>
<div class="dropzone" data-state="idle" id="dropzone">
<div>
<div class="dropzone-icon">↓ Drop file</div>
<h3 class="dropzone-title">Drop a podcast file here</h3>
<p class="dropzone-sub">Accepts <code>.m4a</code> · <code>.mp3</code> · <code>.wav</code> — or paste the absolute path below.</p>
<div class="dropzone-divider">— or —</div>
<div class="field" style="max-width:640px">
<label class="field-label" for="audio-path">Audio path <span class="field-hint">(absolute)</span></label>
<input class="input mono" id="audio-path" type="text" placeholder="/Users/.../folge-XYZ.m4a">
<span class="field-error" id="audio-error" hidden></span>
</div>
</div>
</div>
<div id="resume-banner-slot"></div>
<div class="section-title"><span class="caps">03 · Configuration</span><span class="rule"></span></div>
<div class="card card--padded">
{% include "_partials/config_form.html" %}
</div>
<div class="section-title"><span class="caps">04 · Run</span><span class="rule"></span></div>
<div class="card card--padded">
<div class="spread">
<div class="stack-2">
<div class="caps">Pre-flight</div>
<div class="eta-block" id="eta-block">
<span class="eta-item"><span>Est. time</span><span class="v" id="eta-total">—</span></span>
<span class="sep">·</span>
<span class="eta-item"><span>Disk free</span><span class="v" id="disk-free">—</span></span>
</div>
</div>
<div class="row">
<button class="btn btn--primary btn--lg" id="start-btn" type="button" disabled>
<span style="font-family: monospace;">▶</span>
<span>Start pipeline</span>
<kbd class="kbd-hint">Ctrl + R</kbd>
</button>
</div>
</div>
</div>
{% include "_partials/confirm_run_modal.html" %}
{% endblock %}
- [ ] Step 2: Create config_form.html
Create webgui/templates/_partials/config_form.html:
<form id="config-form" class="form-grid">
<div class="field">
<label class="field-label" for="cfg-visualizer">Visualizer</label>
<select class="select" id="cfg-visualizer" name="viz">
<option value="dialogue">Dialogue · two-column waveform</option>
<option value="monologue">Monologue · single waveform</option>
</select>
</div>
<div class="field">
<label class="field-label" for="cfg-language">Language</label>
<select class="select" id="cfg-language" name="language">
<option value="auto">Auto-detect</option>
<option value="de">de — German</option>
<option value="en">en — English</option>
</select>
</div>
<div class="field">
<label class="field-label" for="cfg-model">Whisper model</label>
<select class="select" id="cfg-model" name="model">
<option value="large-v3-turbo">large-v3-turbo · recommended</option>
<option value="large-v3">large-v3</option>
<option value="large-v2">large-v2</option>
<option value="medium">medium</option>
<option value="small">small</option>
<option value="base">base</option>
<option value="tiny">tiny</option>
</select>
</div>
<div class="field">
<label class="field-label">Speakers</label>
<div class="segmented" role="radiogroup">
<label><input type="radio" name="diarize" value="auto" checked><span>Auto</span></label>
<label><input type="radio" name="diarize" value="2"><span>2</span></label>
<label><input type="radio" name="diarize" value="3"><span>3</span></label>
<label><input type="radio" name="diarize" value="4"><span>4</span></label>
<label><input type="radio" name="diarize" value="5"><span>5</span></label>
<label><input type="radio" name="diarize" value="off"><span>Off</span></label>
</div>
</div>
<div class="field">
<label class="field-label" for="cfg-episode">Episode</label>
<input class="input" id="cfg-episode" name="episode" type="text" placeholder="EP 01">
</div>
<div class="field">
<label class="field-label" for="cfg-channel">Channel · series</label>
<input class="input" id="cfg-channel" name="show_name" type="text" placeholder="Signal">
</div>
<div class="field field--full">
<label class="field-label">Skip phases</label>
<div class="row" style="gap:24px; flex-wrap:wrap;">
<label class="check"><input type="checkbox" name="skip_transcribe"><span class="box"></span><span>Skip transcribe</span></label>
<label class="check"><input type="checkbox" name="skip_meta"><span class="box"></span><span>Skip metadata</span></label>
<label class="check"><input type="checkbox" name="skip_render"><span class="box"></span><span>Skip render</span></label>
<label class="check"><input type="checkbox" name="skip_upload" checked><span class="box"></span><span>Skip upload</span></label>
</div>
</div>
</form>
- [ ] Step 3: Create resume_banner.html
Create webgui/templates/_partials/resume_banner.html:
<div class="banner" data-variant="{{ variant }}">
<div class="banner-icon">
{% if variant == "aborted" %}⚠{% elif variant == "complete" %}✓{% else %}⟳{% endif %}
</div>
<div class="banner-body">
<div class="banner-title">
{% if variant == "aborted" %}Last run aborted during <span class="mono">{{ aborted_phase }}</span>
{% elif variant == "complete" %}Last run complete — loop closed
{% else %}Existing run state found
{% endif %}
</div>
<div class="banner-sub">Skip-boxes for finished phases pre-checked.</div>
</div>
<div class="banner-glyphs">
{% for phase_name in ["transcribe", "meta", "render", "upload"] %}
{% set status = phases.get(phase_name, {}).get("status", "pending") %}
<span class="g-{{ status }}">
{% if status == "done" %}✓{% elif status == "running" %}⟳{% elif status == "aborted" %}✗{% else %}○{% endif %}
{{ {"transcribe":"Transcribe","meta":"Metadata","render":"Render","upload":"Upload"}[phase_name] }}
</span>
{% endfor %}
</div>
</div>
- [ ] Step 4: Create confirm_run_modal.html
Create webgui/templates/_partials/confirm_run_modal.html:
<div class="modal-backdrop" id="modal-confirm-run" hidden>
<div class="modal" role="dialog" aria-modal="true">
<header class="modal-head">
<span class="caps">Pre-flight</span>
<button class="close" type="button" data-close-modal>✕</button>
</header>
<div class="modal-body">
<h2 class="modal-title" id="confirm-modal-title">Start pipeline</h2>
<p class="mono dim">Four phases will run in sequence. Upload requires a second confirmation after render.</p>
<div class="will-list" id="confirm-will-list"></div>
<div class="eta-block" id="confirm-eta-block"></div>
</div>
<footer class="modal-foot">
<button class="btn btn--ghost" type="button" data-close-modal>Cancel</button>
<button class="btn btn--primary" id="confirm-start" type="button">
<span style="font-family: monospace;">▶</span>
<span>Start pipeline</span>
</button>
</footer>
</div>
</div>
- [ ] Step 5: Add probe + modal + start logic to app.js
Append to webgui/static/app.js:
// Audio-path probe + start-button enable
async function probeAudio(path) {
if (!path.trim()) return null;
try {
const r = await fetch('/api/audio/probe', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ path }),
});
return await r.json();
} catch (e) {
return { valid: false, error: 'network' };
}
}
function updateResumeBanner(stem, resumeState) {
const slot = document.getElementById('resume-banner-slot');
if (!slot) return;
if (!resumeState) { slot.innerHTML = ''; return; }
// Fetch the partial rendered server-side
const params = new URLSearchParams({ stem });
fetch(`/runs/${stem}/resume-banner`)
.then(r => r.ok ? r.text() : '')
.then(html => { slot.innerHTML = html; });
}
function updateEtaBlock(eta, diskFree) {
const totalEl = document.getElementById('eta-total');
const diskEl = document.getElementById('disk-free');
if (totalEl && eta) totalEl.textContent = `~ ${Math.round(eta.total / 60)} min`;
if (diskEl && diskFree) diskEl.textContent = `${(diskFree / 1e9).toFixed(0)} GB`;
}
function setStartEnabled(enabled) {
const btn = document.getElementById('start-btn');
if (btn) btn.disabled = !enabled;
}
document.addEventListener('input', async (e) => {
if (e.target.id !== 'audio-path') return;
const path = e.target.value.trim();
if (!path) {
setStartEnabled(false);
updateResumeBanner(null, null);
return;
}
const probe = await probeAudio(path);
const errEl = document.getElementById('audio-error');
if (!probe || !probe.valid) {
errEl.textContent = probe?.error === 'file_not_found' ? '✗ File not found.'
: probe?.error === 'format_unsupported' ? '✗ Format not supported (use .m4a, .mp3, .wav).'
: '✗ Audio check failed.';
errEl.hidden = false;
setStartEnabled(false);
return;
}
errEl.hidden = true;
updateResumeBanner(probe.stem, probe.resume_state);
updateEtaBlock(probe.eta_estimate_s, probe.disk_free_bytes);
setStartEnabled(true);
// store probe for modal
window.__lastProbe = probe;
});
// Modal open / close
document.addEventListener('click', (e) => {
const openTrigger = e.target.closest('[data-open-modal]');
if (openTrigger) {
const id = `modal-${openTrigger.dataset.openModal}`;
document.getElementById(id)?.removeAttribute('hidden');
}
if (e.target.closest('[data-close-modal]')) {
e.target.closest('.modal-backdrop')?.setAttribute('hidden', '');
}
if (e.target.id === 'start-btn') {
openConfirmModal();
}
if (e.target.id === 'confirm-start') {
submitRun();
}
});
function openConfirmModal() {
const probe = window.__lastProbe;
if (!probe) return;
document.getElementById('confirm-modal-title').textContent = `Start pipeline · ${probe.stem}`;
const willList = document.getElementById('confirm-will-list');
const eta = probe.eta_estimate_s;
const phases = [
{ name: 'Transcribe', t: eta.transcribe, skip: false },
{ name: 'Metadata', t: eta.meta, skip: false },
{ name: 'Render', t: eta.render, skip: false },
{ name: 'Upload', t: eta.upload, skip: document.querySelector('input[name="skip_upload"]').checked },
];
willList.innerHTML = phases.map(p => `
<div class="row" data-state="${p.skip ? 'skip' : ''}">
<span class="mark">${p.skip ? '·' : '▶'}</span>
<span class="label">${p.name}</span>
<span class="est">~ ${Math.floor(p.t/60)}:${String(p.t%60).padStart(2,'0')}</span>
</div>`).join('');
const totalNonSkipped = phases.filter(p => !p.skip).reduce((s,p)=>s+p.t,0);
document.getElementById('confirm-eta-block').innerHTML = `
<span class="eta-item"><span>Total est.</span><span class="v">~ ${Math.round(totalNonSkipped/60)} min</span></span>`;
document.getElementById('modal-confirm-run').removeAttribute('hidden');
}
async function submitRun() {
const form = document.getElementById('config-form');
const audio = document.getElementById('audio-path').value.trim();
const body = {
audio,
viz: form.viz.value,
language: form.language.value,
model: form.model.value,
diarize: form.querySelector('input[name="diarize"]:checked').value,
episode: form.episode.value || 'EP 01',
show_name: form.show_name.value || 'Signal',
skip_transcribe: form.skip_transcribe.checked,
skip_meta: form.skip_meta.checked,
skip_render: form.skip_render.checked,
skip_upload: form.skip_upload.checked,
};
const r = await fetch('/api/runs', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify(body),
redirect: 'manual',
});
if (r.status === 303 || r.type === 'opaqueredirect') {
// Browser fetch with redirect:'manual' may give opaqueredirect; use the predictable Location
window.location.href = `/runs/${window.__lastProbe.stem}`;
} else if (r.status === 409) {
const data = await r.json();
alert(`Slot is busy with ${data.detail.stem}`);
} else {
alert(`Start failed: ${r.status}`);
}
}
// Drag-and-Drop — read file.name as starting hint
const dropzone = document.getElementById('dropzone');
if (dropzone) {
['dragenter','dragover'].forEach(ev => dropzone.addEventListener(ev, (e) => {
e.preventDefault();
dropzone.dataset.state = 'hover';
}));
['dragleave','drop'].forEach(ev => dropzone.addEventListener(ev, (e) => {
e.preventDefault();
dropzone.dataset.state = 'idle';
}));
dropzone.addEventListener('drop', (e) => {
const f = e.dataTransfer?.files?.[0];
if (!f) return;
const input = document.getElementById('audio-path');
if (input && !input.value.trim()) {
// browsers don't expose OS path; the file name is a hint only
input.value = f.name;
input.dispatchEvent(new Event('input', { bubbles: true }));
}
});
}
// Keyboard shortcut Ctrl+R / Cmd+R
document.addEventListener('keydown', (e) => {
if ((e.ctrlKey || e.metaKey) && e.key.toLowerCase() === 'r') {
const startBtn = document.getElementById('start-btn');
if (startBtn && !startBtn.disabled) {
e.preventDefault();
openConfirmModal();
}
}
});
- [ ] Step 6: Add /runs/{stem}/resume-banner fragment endpoint
Add to webgui/app.py:
@app.get("/runs/{stem}/resume-banner", response_class=HTMLResponse)
async def runs_resume_banner(stem: str, request: Request):
state = _load_state(stem)
phases = state.get("phases", {})
aborted_phase = next(
(p for p in ("transcribe","meta","render","upload") if phases.get(p, {}).get("status") == "aborted"),
None,
)
if aborted_phase:
variant = "aborted"
elif all(phases.get(p, {}).get("status") in ("done", "skipped") for p in ("transcribe","meta","render","upload")):
variant = "complete"
else:
variant = "inprogress"
return templates.TemplateResponse(
"_partials/resume_banner.html",
{"request": request, "phases": phases, "variant": variant, "aborted_phase": aborted_phase},
)
- [ ] Step 7: Manual sanity check
Run server. Open http://localhost:8765. Type a path in the audio field — verify:
- empty: start-button disabled
- bad path: error message shown
- good path (use tests/fixtures/sample.m4a): start-button enables, ETA appears
- click start → modal opens with phase list and estimates
- click "Start pipeline" → redirects to /runs/sample (which is a 404 since the sample run doesn't have a state file yet — that's fine for now)
- [ ] Step 8: Run all tests
Run: pytest -v
Expected: PASS.
- [ ] Step 9: Commit
git add webgui/templates/index.html webgui/templates/_partials/config_form.html webgui/templates/_partials/resume_banner.html webgui/templates/_partials/confirm_run_modal.html webgui/static/app.js webgui/app.py
git commit -m "feat(webgui): index page — config form + resume banner + confirm modal
Full Start screen: drop-zone with path input, live audio-probe (file-found,
format-check, disk-free, ETA), resume-banner (3 variants from prior runs),
config form (visualizer, language, model, speakers, episode, channel, skip
flags), pre-flight block, confirm-run modal with phase-by-phase ETA list.
Ctrl+R triggers modal. Submit posts to /api/runs and redirects to detail.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 13: MP4 Preview — /runs/{stem}/preview.mp4 with Range support¶
Files:
- Modify: webgui/app.py
- Modify: webgui/templates/run_detail.html (add <video> tag when MP4 exists)
- Modify: tests/test_routes.py
- [ ] Step 1: Write tests for /preview.mp4
Append to tests/test_routes.py:
def test_preview_mp4_404_when_missing(client, tmp_path, monkeypatch):
from webgui import app as app_mod
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
r = client.get("/runs/no-such-run/preview.mp4")
assert r.status_code == 404
def test_preview_mp4_serves_file(client, tmp_path, monkeypatch):
from webgui import app as app_mod
output = tmp_path / "output" / "stem-1"
output.mkdir(parents=True)
mp4 = output / "stem-1-dialogue.mp4"
mp4.write_bytes(b"\x00\x00\x00\x20ftypmp42" + b"\x00" * 1000) # fake mp4
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
r = client.get("/runs/stem-1/preview.mp4")
assert r.status_code == 200
assert r.headers["content-type"] == "video/mp4"
assert len(r.content) > 1000
def test_preview_mp4_range_request(client, tmp_path, monkeypatch):
from webgui import app as app_mod
output = tmp_path / "output" / "stem-2"
output.mkdir(parents=True)
mp4 = output / "stem-2-dialogue.mp4"
content = bytes(range(256)) * 4 # 1024 deterministic bytes
mp4.write_bytes(content)
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
r = client.get("/runs/stem-2/preview.mp4", headers={"Range": "bytes=100-199"})
assert r.status_code == 206
assert r.headers["content-range"] == "bytes 100-199/1024"
assert r.content == content[100:200]
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_routes.py::test_preview_mp4_404_when_missing -v
Expected: FAIL.
- [ ] Step 3: Add /preview.mp4 route with range support
Add to webgui/app.py:
from fastapi.responses import StreamingResponse, Response
def _find_mp4(stem: str) -> Path | None:
output_dir = OUTPUT_ROOT / stem
if not output_dir.exists():
return None
mp4s = sorted(output_dir.glob(f"{stem}-*.mp4"))
if not mp4s:
mp4s = sorted(output_dir.glob("*.mp4"))
return mp4s[-1] if mp4s else None
def _parse_range(header: str | None, size: int) -> tuple[int, int] | None:
if not header or not header.startswith("bytes="):
return None
spec = header[6:].split(",")[0].strip()
if "-" not in spec:
return None
start_s, end_s = spec.split("-", 1)
start = int(start_s) if start_s else 0
end = int(end_s) if end_s else size - 1
if start < 0 or end >= size or start > end:
return None
return start, end
@app.get("/runs/{stem}/preview.mp4")
async def runs_preview_mp4(stem: str, request: Request):
from fastapi import HTTPException
mp4 = _find_mp4(stem)
if mp4 is None:
raise HTTPException(status_code=404, detail="MP4 not found")
size = mp4.stat().st_size
range_header = request.headers.get("Range")
rng = _parse_range(range_header, size)
def file_iter(start: int, end: int, chunk_size: int = 65536):
with mp4.open("rb") as f:
f.seek(start)
remaining = end - start + 1
while remaining > 0:
chunk = f.read(min(chunk_size, remaining))
if not chunk:
break
yield chunk
remaining -= len(chunk)
if rng is None:
return StreamingResponse(
file_iter(0, size - 1),
media_type="video/mp4",
headers={"Content-Length": str(size), "Accept-Ranges": "bytes"},
)
start, end = rng
return StreamingResponse(
file_iter(start, end),
status_code=206,
media_type="video/mp4",
headers={
"Content-Range": f"bytes {start}-{end}/{size}",
"Accept-Ranges": "bytes",
"Content-Length": str(end - start + 1),
},
)
- [ ] Step 4: Add
Modify webgui/templates/run_detail.html — replace the upload_card include block:
{% if variant in ("ready-to-upload", "done") and state.phases.render.status == "done" %}
<section class="card">
<header class="card-header">
<span class="caps">Rendered video · ready</span>
<span class="muted mono" style="margin-left:auto; font-size: 11px;">output/{{ stem }}/{{ state.phases.render.output | default("video.mp4") }}</span>
</header>
<div class="card-body" style="padding: 0;">
<video controls preload="metadata" style="width: 100%; max-height: 540px;" src="/runs/{{ stem }}/preview.mp4"></video>
</div>
</section>
{% endif %}
{% if variant == "ready-to-upload" %}
{% include "_partials/upload_card.html" %}
{% endif %}
- [ ] Step 5: Run tests
Run: pytest tests/test_routes.py -v
Expected: PASS — including 3 new MP4 tests.
- [ ] Step 6: Commit
git add webgui/app.py webgui/templates/run_detail.html tests/test_routes.py
git commit -m "feat(webgui): /runs/{stem}/preview.mp4 with Range support
StreamingResponse over the rendered MP4 with HTTP Range parsing (206
Partial Content for video seeking). Detail-template adds <video controls>
in ready-to-upload and done variants. 65536-byte chunks; no caching.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 14: Upload-Workflow — /runs/{stem}/upload + upload_card¶
Files:
- Modify: webgui/runner.py (spawn_upload helper)
- Modify: webgui/app.py
- Modify: webgui/templates/_partials/upload_card.html (full version)
- Modify: tests/test_routes.py
- Modify: tests/test_runner.py
- [ ] Step 1: Write tests for upload
Append to tests/test_routes.py:
def test_upload_returns_404_if_no_mp4(client, tmp_path, monkeypatch):
from webgui import app as app_mod, runner
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
(tmp_path / "output" / "stem-z").mkdir(parents=True)
r = client.post("/runs/stem-z/upload", json={"privacy": "private"})
assert r.status_code == 404
def test_upload_rejects_public_privacy(client, tmp_path, monkeypatch):
from webgui import app as app_mod
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
(tmp_path / "output" / "stem-q").mkdir(parents=True)
(tmp_path / "output" / "stem-q" / "stem-q-dialogue.mp4").write_bytes(b"x")
r = client.post("/runs/stem-q/upload", json={"privacy": "public"})
assert r.status_code == 400
def test_skip_upload_marks_state(client, tmp_path, monkeypatch):
import json as _json
from webgui import app as app_mod
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
output = tmp_path / "output" / "stem-r"
output.mkdir(parents=True)
state = {
"schema_version": 1, "stem": "stem-r", "audio": "/x",
"started_at": "2026-05-20T10:00:00Z", "updated_at": "2026-05-20T10:30:00Z",
"phases": {"transcribe": {"status":"done"}, "meta": {"status":"done"},
"render": {"status":"done"}, "upload": {"status":"pending"}},
}
(output / "run-state.json").write_text(_json.dumps(state))
r = client.post("/runs/stem-r/skip-upload")
assert r.status_code == 204
fresh = _json.loads((output / "run-state.json").read_text())
assert fresh["phases"]["upload"]["status"] == "skipped"
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_routes.py::test_upload_returns_404_if_no_mp4 -v
Expected: FAIL.
- [ ] Step 3: Add spawn_upload to webgui/runner.py
Append to webgui/runner.py:
def spawn_upload(
video_path: Path,
stem: str,
privacy: str,
output_dir: Path,
log_file: Path,
registry: JobRegistry,
) -> ActiveJob:
"""Spawn upload_youtube.py as a subprocess. Same SSE-queue plumbing as spawn_pipeline."""
import sys
cmd = [
sys.executable,
str(Path(__file__).parent.parent / "upload_youtube.py"),
str(video_path),
"--privacy", privacy,
]
job = ActiveJob(
stem=stem, audio_path=video_path, output_dir=output_dir,
process=None, log_file=log_file, kind="upload",
queue=asyncio.Queue(),
)
if not registry.try_claim(job):
raise RuntimeError(f"Slot is busy with {registry.current.stem!r}")
output_dir.mkdir(parents=True, exist_ok=True)
log_file.parent.mkdir(parents=True, exist_ok=True)
proc = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, bufsize=1,
)
job.process = proc
loop = asyncio.get_running_loop()
def _put(event): loop.call_soon_threadsafe(job.queue.put_nowait, event)
def _reader():
try:
with log_file.open("a", encoding="utf-8", buffering=1) as f:
f.write(f"\n# Upload started {datetime.now().isoformat()}\n")
f.write(f"# Command: {' '.join(cmd)}\n")
f.write("# " + ("─" * 60) + "\n\n")
for line in proc.stdout:
line = line.rstrip("\n")
f.write(line + "\n")
if line:
job.seq += 1
_put(StreamEvent(type="log", seq=job.seq,
data={"msg": line, "level": _classify_level(line)}))
proc.wait()
job.seq += 1
_put(StreamEvent(type="done", seq=job.seq,
data={"exit_code": proc.returncode, "kind": "upload"}))
finally:
registry.release(job)
threading.Thread(target=_reader, daemon=True, name=f"upload-{stem}").start()
return job
- [ ] Step 4: Add /upload and /skip-upload routes
Add to webgui/app.py:
from .runner import spawn_upload
class UploadRequest(BaseModel):
privacy: str = "private"
@app.post("/runs/{stem}/upload", status_code=202)
async def runs_upload(stem: str, req: UploadRequest):
from fastapi import HTTPException
if req.privacy not in ("private", "unlisted"):
raise HTTPException(status_code=400, detail="Privacy must be 'private' or 'unlisted'")
mp4 = _find_mp4(stem)
if mp4 is None:
raise HTTPException(status_code=404, detail="MP4 not found")
if registry.current is not None:
raise HTTPException(
status_code=409,
detail={"error": "slot_busy", "stem": registry.current.stem,
"kind": registry.current.kind},
)
output_dir = OUTPUT_ROOT / stem
ts = datetime.now().strftime("%Y-%m-%dT%H-%M-%S")
log_file = output_dir / f"run-{ts}.log"
spawn_upload(
video_path=mp4, stem=stem, privacy=req.privacy,
output_dir=output_dir, log_file=log_file, registry=registry,
)
return {"status": "accepted", "stem": stem}
@app.post("/runs/{stem}/skip-upload", status_code=204)
async def runs_skip_upload(stem: str):
state_file = OUTPUT_ROOT / stem / "run-state.json"
if not state_file.exists():
from fastapi import HTTPException
raise HTTPException(status_code=404, detail="Run not found")
state = json.loads(state_file.read_text(encoding="utf-8"))
state.setdefault("phases", {})["upload"] = {"status": "skipped"}
state["updated_at"] = datetime.now().isoformat() + "Z"
state_file.write_text(json.dumps(state, indent=2))
return Response(status_code=204)
- [ ] Step 5: Full upload_card.html
Replace webgui/templates/_partials/upload_card.html:
<section class="upload-card">
<div style="min-width:0;">
<div class="caps" style="color: var(--accent);">↑ Final phase — upload to YouTube</div>
<div class="meta-title" style="margin-top: 8px;">
{{ state.phases.meta.title | default(stem) }}
</div>
<div class="mono dim" style="margin-top: 6px; font-size: 11px;">
Channel · {{ state.config.show_name | default("—") }}
· {{ state.phases.render.size_mb | default("—") }} MB
</div>
<div class="privacy" style="margin-top: 16px;">
<span class="caps" style="margin-right: 12px;">Visibility</span>
<div class="segmented" role="radiogroup" id="upload-privacy">
<label><input type="radio" name="vis" value="private" checked><span>Private</span></label>
<label><input type="radio" name="vis" value="unlisted"><span>Unlisted</span></label>
<label aria-disabled="true" style="opacity:0.5; cursor: not-allowed;"><input type="radio" name="vis" value="public" disabled><span>Public</span></label>
</div>
</div>
<p class="mono" style="margin-top: 12px; font-size: 11px; color: var(--fg-tertiary);">
Public is disabled at the application level. Flip it from YouTube Studio after upload.
</p>
</div>
<div class="stack-3" style="align-self: stretch; justify-content: center;">
<button class="btn btn--primary btn--lg" id="upload-btn" type="button">
<span style="font-family: monospace;">↑</span>
<span>Upload to YouTube</span>
</button>
<button class="btn btn--ghost btn--sm" id="skip-upload-btn" type="button">Skip upload · keep local</button>
</div>
</section>
- [ ] Step 6: Wire upload buttons in app.js
Append to webgui/static/app.js:
document.addEventListener('click', async (e) => {
if (e.target.id === 'upload-btn') {
const privacy = document.querySelector('#upload-privacy input:checked').value;
const stem = window.location.pathname.split('/').pop();
const r = await fetch(`/runs/${stem}/upload`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ privacy }),
});
if (r.ok) window.startEventSource(stem);
else alert(`Upload failed: ${r.status}`);
}
if (e.target.id === 'skip-upload-btn') {
const stem = window.location.pathname.split('/').pop();
await fetch(`/runs/${stem}/skip-upload`, { method: 'POST' });
window.location.reload();
}
});
- [ ] Step 7: Run all tests
Run: pytest -v
Expected: PASS.
- [ ] Step 8: Commit
git add webgui/runner.py webgui/app.py webgui/templates/_partials/upload_card.html webgui/static/app.js tests/test_routes.py
git commit -m "feat(webgui): upload trigger + skip-upload — second subprocess via spawn_upload
POST /runs/{stem}/upload validates privacy (private/unlisted only,
public rejected), checks MP4 exists, spawns upload_youtube.py through
the same registry slot. Stdout appends to the same per-run logfile
with an upload header. POST /runs/{stem}/skip-upload sets phases.upload
to skipped without invoking the subprocess. Upload-card renders title,
privacy radio (public disabled), two action buttons.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 15: Abort Run + OS-Hooks (Finder, QuickTime)¶
Files:
- Modify: webgui/app.py
- Modify: tests/test_routes.py
- [ ] Step 1: Write tests
Append to tests/test_routes.py:
def test_abort_run_sigterms_subprocess(client, tmp_path, monkeypatch):
"""When a job is active, /abort sends SIGTERM and frees the slot."""
import sys, time
from webgui import app as app_mod, runner
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
output = tmp_path / "output" / "abort-stem"
output.mkdir(parents=True)
long_cmd = [sys.executable, "-c", "import time; print('hi', flush=True); time.sleep(30)"]
runner.spawn_pipeline(
cmd=long_cmd, stem="abort-stem",
audio_path=Path("/x"), output_dir=output, log_file=output / "x.log",
registry=runner.registry,
)
# Wait for subprocess to actually start
time.sleep(0.5)
r = client.post("/runs/abort-stem/abort")
assert r.status_code == 204
time.sleep(0.5)
assert runner.registry.current is None
def test_abort_unknown_run_returns_404(client, tmp_path, monkeypatch):
from webgui import app as app_mod, runner
runner.registry = runner.JobRegistry()
monkeypatch.setattr(app_mod, "registry", runner.registry)
r = client.post("/runs/no-such-stem/abort")
assert r.status_code == 404
def test_open_finder_path_must_be_inside_repo(client, tmp_path, monkeypatch):
"""Reject paths outside REPO_ROOT/output."""
from webgui import app as app_mod
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
r = client.post("/open/finder", json={"path": "/etc/passwd"})
assert r.status_code == 400
def test_open_finder_calls_open_minus_R(client, tmp_path, monkeypatch):
"""Valid path: open -R is invoked."""
import subprocess
from webgui import app as app_mod
calls = []
monkeypatch.setattr(app_mod, "OUTPUT_ROOT", tmp_path / "output")
(tmp_path / "output" / "stem-a").mkdir(parents=True)
target = tmp_path / "output" / "stem-a" / "video.mp4"
target.write_bytes(b"x")
def fake_run(args, **kwargs):
calls.append(args)
return subprocess.CompletedProcess(args, 0)
monkeypatch.setattr(app_mod.subprocess, "run", fake_run)
r = client.post("/open/finder", json={"path": str(target)})
assert r.status_code == 204
assert calls and calls[0][:2] == ["open", "-R"]
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_routes.py::test_abort_run_sigterms_subprocess -v
Expected: FAIL.
- [ ] Step 3: Add abort + OS-hooks routes
Add to webgui/app.py:
import subprocess
import signal
class OpenRequest(BaseModel):
path: str
def _path_is_safe(path: Path) -> bool:
try:
resolved = path.resolve()
return resolved.is_relative_to(REPO_ROOT.resolve()) or resolved.is_relative_to(OUTPUT_ROOT.resolve())
except (OSError, ValueError):
return False
@app.post("/runs/{stem}/abort", status_code=204)
async def runs_abort(stem: str):
from fastapi import HTTPException
job = registry.current
if job is None or job.stem != stem:
raise HTTPException(status_code=404, detail="No active run for this stem")
if job.process and job.process.poll() is None:
try:
job.process.send_signal(signal.SIGTERM)
except ProcessLookupError:
pass
return Response(status_code=204)
@app.post("/open/finder", status_code=204)
async def open_finder(req: OpenRequest):
from fastapi import HTTPException
path = Path(req.path)
if not _path_is_safe(path):
raise HTTPException(status_code=400, detail="Path outside repo")
subprocess.run(["open", "-R", str(path)], check=False)
return Response(status_code=204)
@app.post("/open/quicktime", status_code=204)
async def open_quicktime(req: OpenRequest):
from fastapi import HTTPException
path = Path(req.path)
if not _path_is_safe(path):
raise HTTPException(status_code=400, detail="Path outside repo")
subprocess.run(["open", "-a", "QuickTime Player", str(path)], check=False)
return Response(status_code=204)
- [ ] Step 4: Run tests
Run: pytest tests/test_routes.py -v
Expected: PASS.
- [ ] Step 5: Commit
git add webgui/app.py tests/test_routes.py
git commit -m "feat(webgui): abort + OS-hooks (finder, quicktime)
POST /runs/{stem}/abort SIGTERMs the active subprocess if it matches.
POST /open/finder + /open/quicktime invoke /usr/bin/open with safe-path
check (only paths inside REPO_ROOT or OUTPUT_ROOT are accepted).
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 16: Settings — Theme persistence + /api/settings¶
Files:
- Create: webgui/settings.py
- Create: tests/test_settings.py
- Modify: webgui/app.py
- [ ] Step 1: Write tests for settings
Create tests/test_settings.py:
import json
from pathlib import Path
def test_load_settings_returns_defaults_when_file_missing(tmp_path):
from webgui.settings import load_settings
s = load_settings(tmp_path / "nope.json")
assert s["theme"] == "dark"
assert s["tail_default"] is True
def test_save_then_load_roundtrips(tmp_path):
from webgui.settings import load_settings, save_settings
p = tmp_path / "settings.json"
save_settings(p, {"theme": "light"})
loaded = load_settings(p)
assert loaded["theme"] == "light"
assert loaded["tail_default"] is True # default still applies
def test_save_settings_partial_merges(tmp_path):
from webgui.settings import save_settings, load_settings
p = tmp_path / "settings.json"
save_settings(p, {"theme": "dark", "tail_default": False})
save_settings(p, {"theme": "light"}) # don't touch tail_default
loaded = load_settings(p)
assert loaded["theme"] == "light"
assert loaded["tail_default"] is False
- [ ] Step 2: Run tests, verify failure
Run: pytest tests/test_settings.py -v
Expected: FAIL — module not found.
- [ ] Step 3: Create webgui/settings.py
Create webgui/settings.py:
"""Settings persistence — Theme / tail-default / preferred-defaults."""
import json
from pathlib import Path
DEFAULTS = {
"theme": "dark",
"tail_default": True,
"preferred_visualizer": "dialogue",
"preferred_model": "large-v3-turbo",
}
def load_settings(path: Path) -> dict:
if not path.exists():
return dict(DEFAULTS)
try:
on_disk = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return dict(DEFAULTS)
merged = dict(DEFAULTS)
merged.update(on_disk)
return merged
def save_settings(path: Path, partial: dict) -> None:
current = load_settings(path)
current.update(partial)
path.write_text(json.dumps(current, indent=2), encoding="utf-8")
- [ ] Step 4: Add /api/settings routes
Add to webgui/app.py:
import os
from .settings import load_settings, save_settings
SETTINGS_PATH = Path(os.path.expanduser("~/.whisper-pipeline-ui.json"))
class SettingsPatch(BaseModel):
theme: str | None = None
tail_default: bool | None = None
preferred_visualizer: str | None = None
preferred_model: str | None = None
@app.get("/api/settings")
async def api_get_settings():
return load_settings(SETTINGS_PATH)
@app.post("/api/settings", status_code=204)
async def api_post_settings(patch: SettingsPatch):
payload = {k: v for k, v in patch.model_dump().items() if v is not None}
save_settings(SETTINGS_PATH, payload)
return Response(status_code=204)
- [ ] Step 5: Add theme-toggle UI to base.html topbar
Modify webgui/templates/base.html — inside .topbar-meta, add:
<button class="btn btn--ghost btn--sm" id="theme-toggle" type="button" title="Toggle theme">
<span id="theme-icon">☾</span>
</button>
Add to webgui/static/app.js:
document.addEventListener('click', async (e) => {
if (e.target.closest('#theme-toggle')) {
const html = document.documentElement;
const next = html.dataset.theme === 'dark' ? 'light' : 'dark';
html.dataset.theme = next;
localStorage.setItem('theme', next);
document.getElementById('theme-icon').textContent = next === 'dark' ? '☾' : '☀';
fetch('/api/settings', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({ theme: next }),
});
}
});
- [ ] Step 6: Run all tests
Run: pytest -v
Expected: PASS.
- [ ] Step 7: Commit
git add webgui/settings.py webgui/app.py webgui/templates/base.html webgui/static/app.js tests/test_settings.py
git commit -m "feat(webgui): settings persistence + theme-toggle
webgui/settings.py loads/saves ~/.whisper-pipeline-ui.json with sane
defaults (dark theme, tail-on, dialogue+large-v3-turbo as preferred
defaults). GET/POST /api/settings expose it. Topbar gets a moon/sun
toggle that flips data-theme on <html>, persists to localStorage AND
server.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 17: Mood-Tinting CSS¶
Files:
- Modify: webgui/static/style.css (mood layer)
- [ ] Step 1: Add mood-tinting tokens + selectors
The full style.css ship lands in Task 18 (port from design-concepts/style.css). For now, append a mood-layer stub to the existing placeholder webgui/static/style.css:
/* --- Mood tinting --------------------------------------------------- */
:root {
--mood-tint-strength: 7%;
--role-error: #d4203a;
--role-success: #39ff7a;
--role-warning: #ffb442;
--accent: #a878ff;
}
body[data-page-mood="error"] {
background:
radial-gradient(1200px 800px at 70% -20%, color-mix(in oklab, var(--role-error) var(--mood-tint-strength), transparent), transparent 60%),
#060709;
}
body[data-page-mood="success"] {
background:
radial-gradient(1200px 800px at 70% -20%, color-mix(in oklab, var(--role-success) var(--mood-tint-strength), transparent), transparent 60%),
#060709;
}
body[data-page-mood="warning"] {
background:
radial-gradient(1200px 800px at 70% -20%, color-mix(in oklab, var(--role-warning) var(--mood-tint-strength), transparent), transparent 60%),
#060709;
}
body[data-page-mood="error"] .topbar { background: color-mix(in oklab, var(--role-error) 5%, #0b0d11); }
body[data-page-mood="success"] .topbar { background: color-mix(in oklab, var(--role-success) 5%, #0b0d11); }
body[data-page-mood="warning"] .topbar { background: color-mix(in oklab, var(--role-warning) 5%, #0b0d11); }
- [ ] Step 2: Commit
git add webgui/static/style.css
git commit -m "feat(webgui): mood-tinting layer (error/success/warning)
Body[data-page-mood] gets a subtle radial-gradient tint (7%) and topbar
gets a 5% color-mix on its base surface. Accent and identity remain
Strategist Spectre. The variable --mood-tint-strength keeps the dial
in one place.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 18: Port full CSS from design-concepts/¶
Files:
- Replace: webgui/static/style.css
- Create: webgui/static/fonts/*.woff2 (downloaded from Google Fonts)
- [ ] Step 1: Copy style.css from design-concepts/
- [ ] Step 2: Vendor the fonts
The placeholder @import from Google Fonts works fine in dev. For production, download the woff2 files into webgui/static/fonts/. Use gfonts-helper or download manually from https://fonts.google.com (EB Garamond, Space Grotesk, Inter, JetBrains Mono).
For this MVP keep the @import. Just add a note in the spec.
Add to top of webgui/static/style.css (replacing the existing @import):
/* In production: vendor woff2 files into static/fonts/ and replace this @import. */
@import url('https://fonts.googleapis.com/css2?family=EB+Garamond:ital,wght@0,400;0,500;0,600;1,400;1,500;1,600&family=Space+Grotesk:wght@400;500;600;700&family=Inter:wght@400;500;600;700&family=JetBrains+Mono:wght@400;500;600;700&display=swap');
- [ ] Step 3: Re-append the mood-layer block (Task 17 changes were overwritten)
Append the mood-tinting block from Task 17 to the end of the new style.css. Bullet from earlier task step 1 (just paste it after the last existing rule).
- [ ] Step 4: Remove aspect-switching CSS rules from style.css (production: Strategist only)
Find the block starting with /* Aspect overrides — set on <html data-aspect="…"> */ and remove the lines for [data-aspect="shugo"], [data-aspect="kantoku"], [data-aspect="sensei"]. Keep [data-aspect="gunshi"] since that's the default and matches what base.html sets.
- [ ] Step 5: Manual visual smoke
Run server: uvicorn webgui.app:app --reload --port 8765. Open http://localhost:8765/ and /runs (after putting one fixture state in output/). Verify the design matches design-concepts/index.html and design-concepts/runs.html close enough.
- [ ] Step 6: Commit
git add webgui/static/style.css
git commit -m "feat(webgui): port full CSS from design-concepts/ + drop aspect-switch
Replaces placeholder style.css with the full Kuro Signal Protocol port.
Aspect overrides for shugo/kantoku/sensei removed (production: Strategist
default only). Mood-layer block from Task 17 re-applied at the bottom.
Fonts still loaded via Google Fonts @import — vendoring left as future
maintenance task.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 19: Transcript + Metadata Preview Partials¶
Files:
- Create: webgui/templates/_partials/transcript_preview.html
- Create: webgui/templates/_partials/metadata_card.html
- Modify: webgui/templates/run_detail.html
- Modify: webgui/app.py (load transcript snippet)
- [ ] Step 1: Add transcript-snippet loader to webgui/app.py
Add helper:
def _load_transcript_snippet(stem: str, max_lines: int = 8) -> list[dict]:
txt = OUTPUT_ROOT / stem / f"{stem}.txt"
if not txt.exists():
return []
lines = []
for raw in txt.read_text(encoding="utf-8").splitlines():
s = raw.strip()
if not s:
continue
if ":" in s:
spk, _, text = s.partition(":")
lines.append({"spk": spk.strip(), "text": text.strip()})
else:
lines.append({"spk": "", "text": s})
if len(lines) >= max_lines:
break
return lines
def _load_metadata(stem: str) -> dict | None:
meta_file = OUTPUT_ROOT / stem / f"{stem}.youtube-meta.json"
if not meta_file.exists():
return None
try:
return json.loads(meta_file.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
return None
In the runs_detail handler, pass them through:
return templates.TemplateResponse(
"run_detail.html",
{
"request": request,
"stem": stem,
"state": state,
"phases": phases,
"variant": variant,
"page_mood": page_mood,
"active": active,
"transcript_lines": _load_transcript_snippet(stem),
"youtube_meta": _load_metadata(stem),
},
)
- [ ] Step 2: Create transcript_preview.html
Create webgui/templates/_partials/transcript_preview.html:
{% if transcript_lines %}
<section class="card">
<header class="card-header">
<span class="caps">Transcript preview</span>
<span class="pill" data-tone="success"><span class="dot"></span>Phase 1 complete</span>
</header>
<div class="card-body">
<div class="transcript">
{% for line in transcript_lines %}
<div class="line" data-spk="{{ 'a' if loop.index0 % 2 == 0 else 'b' }}">
{% if line.spk %}<span class="spk">{{ line.spk }}</span>{% endif %}
<span class="text">{{ line.text }}</span>
</div>
{% endfor %}
</div>
</div>
</section>
{% endif %}
- [ ] Step 3: Create metadata_card.html
Create webgui/templates/_partials/metadata_card.html:
{% if youtube_meta %}
<section class="card" data-tone="accent">
<header class="card-header">
<span class="caps">YouTube metadata</span>
<span class="pill" data-tone="success"><span class="dot"></span>Generated</span>
</header>
<div class="card-body">
<div class="meta-title">{{ youtube_meta.title | default(stem) }}</div>
<p class="meta-desc" style="margin-top: 12px;">{{ youtube_meta.description | default("") | truncate(280) }}</p>
{% if youtube_meta.tags %}
<div class="caps" style="margin-top: 16px; margin-bottom: 8px;">Tags · {{ youtube_meta.tags | length }}</div>
<div class="tags">
{% for tag in youtube_meta.tags[:12] %}<span class="tag">{{ tag }}</span>{% endfor %}
{% if youtube_meta.tags | length > 12 %}<span class="tag">+{{ youtube_meta.tags | length - 12 }} more</span>{% endif %}
</div>
{% endif %}
{% if youtube_meta.chapters %}
<div class="caps" style="margin-top: 16px; margin-bottom: 8px;">Chapters · {{ youtube_meta.chapters | length }}</div>
<div class="chapters">
{% for chapter in youtube_meta.chapters[:8] %}
<div class="chapter"><span class="ts">{{ chapter.timestamp | default("00:00") }}</span><span>{{ chapter.title | default(chapter.label | default("")) }}</span></div>
{% endfor %}
</div>
{% endif %}
</div>
</section>
{% endif %}
- [ ] Step 4: Wire them into run_detail.html
Modify webgui/templates/run_detail.html — inside the main .stack left column (after log_panel include), add:
And inside the right <aside class="stack">, add:
- [ ] Step 5: Run all tests
Run: pytest -v
Expected: PASS.
- [ ] Step 6: Commit
git add webgui/templates/_partials/transcript_preview.html webgui/templates/_partials/metadata_card.html webgui/templates/run_detail.html webgui/app.py
git commit -m "feat(webgui): transcript + metadata preview cards
Run detail loads {stem}.txt (first 8 non-empty lines, split on first colon
into spk:text) and {stem}.youtube-meta.json. Transcript preview shows
speaker-alternation styling via data-spk='a|b'. Metadata card shows title,
truncated description, first 12 tags (overflow chip), first 8 chapters.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 20: Entry-Point — webgui.py + browser auto-open¶
Files:
- Create: webgui.py (top-level entry)
- Modify: README.md
- [ ] Step 1: Write webgui.py
Create webgui.py:
#!/usr/bin/env python3
"""WebGUI entry-point.
Usage:
python webgui.py # opens browser at http://localhost:8765
python webgui.py --port 9000 # custom port, no browser
python webgui.py --no-open # don't open browser
"""
import argparse
import threading
import time
import webbrowser
import uvicorn
def _open_browser_after_delay(url: str, delay: float = 1.0) -> None:
time.sleep(delay)
webbrowser.open(url)
def main() -> None:
parser = argparse.ArgumentParser(description="Whisper-Pipeline WebGUI")
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--port", type=int, default=8765)
parser.add_argument("--no-open", action="store_true", help="Don't auto-open browser")
parser.add_argument("--reload", action="store_true", help="Dev mode: reload on file change")
args = parser.parse_args()
url = f"http://{args.host}:{args.port}"
if not args.no_open:
threading.Thread(
target=_open_browser_after_delay, args=(url,), daemon=True,
).start()
print(f"\n Whisper-Pipeline WebGUI → {url}\n")
uvicorn.run(
"webgui.app:app",
host=args.host, port=args.port,
reload=args.reload, log_level="info",
)
if __name__ == "__main__":
main()
- [ ] Step 2: Update README.md
Open README.md. Replace the ## TUI (Empfohlen) section with ## WebGUI (Empfohlen):
## WebGUI (Empfohlen)
```bash
podcast-video-upload # Alias — startet WebGUI + öffnet Browser
# oder ohne Alias:
python webgui.py
Im Browser auf http://localhost:8765. Drop deinen Podcast in die Source-Zone (oder Pfad eingeben),
Optionen wählen, Start pipeline klicken. Live-Log + Phase-Progress per Server-Sent Events.
Nach dem Render-Schritt zeigt die WebGUI die MP4-Vorschau direkt im Browser — Upload wird nicht automatisch gestartet. Privacy wählen (Private/Unlisted), dann Upload to YouTube klicken.
TUI (Fallback)¶
Die alte Textual-TUI bleibt erstmal als Fallback verfügbar:
Also adjust the shell-alias section:
```markdown
## Shell-Alias
Im `~/.zshrc` eintragen:
```bash
alias podcast-video-upload='cd <repo-root> && source .venv/bin/activate && python webgui.py'
- [ ] **Step 3: Manual sanity**
Run: `python webgui.py --no-open`. Open http://localhost:8765 in a browser. Verify the page renders. Ctrl+C to stop.
Then: `python webgui.py`. Verify a browser tab opens automatically after ~1 second.
- [ ] **Step 4: Commit**
```bash
git add webgui.py README.md
git commit -m "feat(webgui): entry-point + README — WebGUI as primary
webgui.py starts uvicorn on localhost:8765 and auto-opens the default
browser (--no-open to skip). README promotes WebGUI to primary, TUI to
fallback. Shell-alias updated to point at webgui.py.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>"
Task 21: End-to-End Smoke Test (manual) + DoD checklist¶
Files:
- Modify: docs/superpowers/specs/2026-05-20-webgui-design.md (only DoD checkbox marks; optional)
This is a manual verification task — no code, just a checklist.
- [ ] Step 1: Run a real Pipeline through WebGUI
Provide a real 5-min .m4a (sample podcast). In the browser:
1. ✓ Audio path validates, ETA appears, start-button enables
2. ✓ Confirm modal opens, phase list shows, start triggers redirect
3. ✓ Run-detail page shows live log streaming
4. ✓ Phase indicator updates as phases complete
5. ✓ Progress bar advances
6. ✓ Transcript preview appears after phase 1
7. ✓ Metadata card appears after phase 2
8. ✓ MP4 video plays after phase 3
9. ✓ Upload card appears, privacy radio shows (public disabled)
10. ✓ Click Upload → upload subprocess starts, logs append
11. ✓ After upload: variant flips to done, YouTube URL shown
12. ✓ Visit /runs — the run appears, sorted newest first
13. ✓ Filter chip done shows only this run
14. ✓ Theme toggle works, persists across reloads
15. ✓ Mood tinting: on a separately-fabricated aborted run, body has crimson radial
- [ ] Step 2: Reconnect test
While a run is mid-render, close the browser tab. Reopen /runs/{stem}. Verify the log replays from the start and live updates resume.
- [ ] Step 3: Single-slot test
Open two browser tabs. Trigger a run in tab A. In tab B, try to start a second run on a different audio. Verify: tab B alerts with "Slot is busy with {first-stem}" and stays on /.
- [ ] Step 4: TUI still works
Verify the TUI launches, accepts the file, and still does the right thing through the shared pipeline_core.
- [ ] Step 5: Run full test suite one last time
Expected: all green. If anything broke, fix it.
- [ ] Step 6: Commit + tag (optional)
Nothing to commit unless fixes were needed. If happy, tag:
Summary of expected file count after this plan¶
- New Python files: 7 (
pipeline_core.py,webgui/__init__.py,webgui/app.py,webgui/runner.py,webgui/runs.py,webgui/probe.py,webgui/settings.py,webgui.py) - New templates: 4 base + 9 partials = 13
- New static assets: 3 (style.css, app.js, htmx.min.js)
- New tests: 5 files (
test_pipeline_core.py,test_probe.py,test_runs.py,test_runner.py,test_routes.py,test_settings.py) + ~50 test functions - New fixtures: 6 run-states + 4 stdout-snippets + 1 sample.m4a + 1 mock_pipeline.py = 12
Total: roughly 30 new files, ~3000 LOC, fully tested up to the SSE-stream layer.