# Workshop Auto-Download 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:** Make freshly-added workshop items download automatically, give owners a per-overlay refresh button, and run a daily global refresh from a systemd timer.

**Architecture:** The per-overlay `build_overlay` job already runs after every add — extend its `WorkshopBuilder` to download missing/stale `.vpk` files before symlinking, with bounded retries and cancel-aware backoff. Add a `POST /overlays/{id}/refresh` route, a `flask workshop-refresh` CLI subcommand, and a `left4me-workshop-refresh.{service,timer}` pair that enqueues the existing `refresh_workshop_items` job daily.

**Tech Stack:** Python 3 / Flask / SQLAlchemy 2 / Click (Flask CLI) / SQLite / systemd / pytest.

**Spec:** [`docs/superpowers/specs/2026-05-11-workshop-auto-download-design.md`](../specs/2026-05-11-workshop-auto-download-design.md).

**Prerequisites already verified:**
- `Job.user_id` is **already nullable** (`l4d2web/models.py:154`) — no migration needed.
- `steam_workshop.download_to_cache` accepts `should_cancel` and is `(mtime, size)`-idempotent (`l4d2web/services/steam_workshop.py:202`).
- The add route already enqueues `build_overlay` (`l4d2web/routes/workshop_routes.py:95`).

---

## File Structure

**Create:**
- `deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.service`
- `deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.timer`

**Modify:**
- `l4d2web/services/overlay_builders.py` — new `_download_with_retry` helper + new download phase in `WorkshopBuilder.build`. Item data tuple grows to carry the fields the downloader needs.
- `l4d2web/routes/workshop_routes.py` — new `refresh_overlay` route.
- `l4d2web/cli.py` — new `workshop-refresh` click command and registration.
- `l4d2web/templates/overlay_detail.html` — Refresh button next to the existing Add form.
- `l4d2web/tests/test_overlay_builders.py` — new tests for download-on-build, retry, cancellation, no-file_url skip.
- `l4d2web/tests/test_workshop_routes.py` — new tests for `/overlays/{id}/refresh`.
- `l4d2web/tests/test_cli.py` (create if absent) — tests for `workshop-refresh` CLI.
- `deploy/deploy-test-server.sh` — `systemctl enable --now left4me-workshop-refresh.timer` line.

**Do NOT modify:**
- `l4d2web/services/steam_workshop.py` — `download_to_cache` stays a single-shot primitive; retry policy is a caller concern.
- `l4d2web/services/job_worker.py` — scheduler rules and `_run_refresh_workshop_items` are unchanged.
- `l4d2web/models.py` — no schema changes.

---

## Task 1: Builder helper — `_download_with_retry` + `_sleep_with_cancel`

**Goal:** Introduce two small helpers in `overlay_builders.py` and prove they behave correctly in isolation. No call sites yet.

**Files:**
- Modify: `l4d2web/services/overlay_builders.py` (add helpers near the top of the file, after imports, before `WorkshopBuilder`)
- Test: `l4d2web/tests/test_overlay_builders.py` (append new test cases)

- [ ] **Step 1.1: Write the failing tests for `_sleep_with_cancel`**

Append at the bottom of `l4d2web/tests/test_overlay_builders.py`:

```python
def test_sleep_with_cancel_returns_normally_when_not_cancelled():
    from l4d2web.services.overlay_builders import _sleep_with_cancel

    cancelled = _sleep_with_cancel(0.05, lambda: False, poll_interval=0.01)
    assert cancelled is False


def test_sleep_with_cancel_returns_early_when_cancelled():
    import time
    from l4d2web.services.overlay_builders import _sleep_with_cancel

    flag = {"cancel": False}
    def cancel_check():
        return flag["cancel"]

    # Flip cancel after a short delay using a thread.
    import threading
    threading.Timer(0.05, lambda: flag.update(cancel=True)).start()

    start = time.monotonic()
    cancelled = _sleep_with_cancel(5.0, cancel_check, poll_interval=0.01)
    elapsed = time.monotonic() - start

    assert cancelled is True
    assert elapsed < 0.5, f"should have woken up promptly, slept {elapsed:.3f}s"
```

- [ ] **Step 1.2: Run the tests to verify they fail**

```bash
cd /Users/mwiegand/Projekte/left4me
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py::test_sleep_with_cancel_returns_normally_when_not_cancelled \
                 l4d2web/tests/test_overlay_builders.py::test_sleep_with_cancel_returns_early_when_cancelled -v
```

Expected: both FAIL with `ImportError: cannot import name '_sleep_with_cancel'`.

- [ ] **Step 1.3: Implement `_sleep_with_cancel`**

In `l4d2web/services/overlay_builders.py`, add after the existing imports and before `class BuildError`:

```python
import time as _time


def _sleep_with_cancel(
    seconds: float,
    should_cancel: CancelCheck,
    *,
    poll_interval: float = 0.25,
) -> bool:
    """Sleep up to `seconds`, returning early (True) if `should_cancel` becomes
    True. Returns False on a full uninterrupted sleep. Polls every
    `poll_interval` seconds."""
    deadline = _time.monotonic() + seconds
    while True:
        if should_cancel():
            return True
        remaining = deadline - _time.monotonic()
        if remaining <= 0:
            return False
        _time.sleep(min(poll_interval, remaining))
```

- [ ] **Step 1.4: Run the tests to verify they pass**

```bash
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py::test_sleep_with_cancel_returns_normally_when_not_cancelled \
                 l4d2web/tests/test_overlay_builders.py::test_sleep_with_cancel_returns_early_when_cancelled -v
```

Expected: both PASS.

- [ ] **Step 1.5: Write the failing tests for `_download_with_retry`**

Append at the bottom of `l4d2web/tests/test_overlay_builders.py`:

```python
def test_download_with_retry_succeeds_on_first_attempt(env, tmp_path, monkeypatch):
    from l4d2web.services import overlay_builders, steam_workshop

    monkeypatch.setattr(overlay_builders, "_sleep_with_cancel", lambda *a, **kw: False)
    calls = []
    def fake_download(meta, cache_root, *, should_cancel=None):
        calls.append(1)
        return cache_root / f"{meta.steam_id}.vpk"
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="x", filename="x.vpk", file_url="http://e/x.vpk",
        file_size=1, time_updated=1700000000, preview_url="", consumer_app_id=550, result=1,
    )
    out, err, on_stdout, on_stderr = _capture_logs()
    overlay_builders._download_with_retry(
        meta, tmp_path / "cache",
        on_stderr=on_stderr, should_cancel=lambda: False,
    )
    assert calls == [1]
    assert err == []


def test_download_with_retry_retries_then_succeeds(env, tmp_path, monkeypatch):
    import requests
    from l4d2web.services import overlay_builders, steam_workshop

    monkeypatch.setattr(overlay_builders, "_sleep_with_cancel", lambda *a, **kw: False)
    attempts = {"n": 0}
    def fake_download(meta, cache_root, *, should_cancel=None):
        attempts["n"] += 1
        if attempts["n"] < 3:
            raise requests.ConnectionError("boom")
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="x", filename="x.vpk", file_url="http://e/x.vpk",
        file_size=1, time_updated=1700000000, preview_url="", consumer_app_id=550, result=1,
    )
    out, err, on_stdout, on_stderr = _capture_logs()
    overlay_builders._download_with_retry(
        meta, tmp_path / "cache",
        on_stderr=on_stderr, should_cancel=lambda: False,
    )
    assert attempts["n"] == 3
    # Two failure messages on stderr, one per failed attempt.
    assert sum(1 for line in err if "attempt" in line and "failed" in line) == 2


def test_download_with_retry_exhausts_and_raises(env, tmp_path, monkeypatch):
    import requests
    from l4d2web.services import overlay_builders, steam_workshop

    monkeypatch.setattr(overlay_builders, "_sleep_with_cancel", lambda *a, **kw: False)
    def fake_download(meta, cache_root, *, should_cancel=None):
        raise requests.ConnectionError("permanent")
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="x", filename="x.vpk", file_url="http://e/x.vpk",
        file_size=1, time_updated=1700000000, preview_url="", consumer_app_id=550, result=1,
    )
    out, err, on_stdout, on_stderr = _capture_logs()
    with pytest.raises(requests.ConnectionError):
        overlay_builders._download_with_retry(
            meta, tmp_path / "cache",
            on_stderr=on_stderr, should_cancel=lambda: False,
        )


def test_download_with_retry_propagates_interrupted(env, tmp_path, monkeypatch):
    from l4d2web.services import overlay_builders, steam_workshop

    def fake_download(meta, cache_root, *, should_cancel=None):
        raise InterruptedError("cancelled")
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="x", filename="x.vpk", file_url="http://e/x.vpk",
        file_size=1, time_updated=1700000000, preview_url="", consumer_app_id=550, result=1,
    )
    out, err, on_stdout, on_stderr = _capture_logs()
    with pytest.raises(InterruptedError):
        overlay_builders._download_with_retry(
            meta, tmp_path / "cache",
            on_stderr=on_stderr, should_cancel=lambda: False,
        )


def test_download_with_retry_bails_when_cancelled_during_backoff(env, tmp_path, monkeypatch):
    import requests
    from l4d2web.services import overlay_builders, steam_workshop

    def fake_download(meta, cache_root, *, should_cancel=None):
        raise requests.ConnectionError("boom")
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)
    # _sleep_with_cancel reports cancellation -> retry loop exits early.
    monkeypatch.setattr(overlay_builders, "_sleep_with_cancel", lambda *a, **kw: True)

    meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="x", filename="x.vpk", file_url="http://e/x.vpk",
        file_size=1, time_updated=1700000000, preview_url="", consumer_app_id=550, result=1,
    )
    out, err, on_stdout, on_stderr = _capture_logs()
    with pytest.raises(InterruptedError):
        overlay_builders._download_with_retry(
            meta, tmp_path / "cache",
            on_stderr=on_stderr, should_cancel=lambda: False,
        )
```

- [ ] **Step 1.6: Run the tests to verify they fail**

```bash
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py -k "download_with_retry" -v
```

Expected: all five FAIL with `AttributeError: module 'l4d2web.services.overlay_builders' has no attribute '_download_with_retry'`.

- [ ] **Step 1.7: Implement `_download_with_retry`**

In `l4d2web/services/overlay_builders.py`, add after `_sleep_with_cancel`:

```python
import requests as _requests

from l4d2web.services.steam_workshop import (
    WorkshopMetadata,
    download_to_cache,
)


DOWNLOAD_RETRY_ATTEMPTS = 3
DOWNLOAD_RETRY_BACKOFF_SECONDS = (1.0, 2.0, 4.0)


def _download_with_retry(
    meta: WorkshopMetadata,
    cache_root: Path,
    *,
    on_stderr: LogSink,
    should_cancel: CancelCheck,
) -> None:
    """Wrap `download_to_cache` with bounded retries and cancel-aware backoff.
    Raises the last exception after `DOWNLOAD_RETRY_ATTEMPTS` failures.
    Raises `InterruptedError` if cancelled during a backoff sleep."""
    last_exc: BaseException | None = None
    for attempt in range(1, DOWNLOAD_RETRY_ATTEMPTS + 1):
        try:
            download_to_cache(meta, cache_root, should_cancel=should_cancel)
            return
        except InterruptedError:
            raise
        except (_requests.RequestException, OSError) as exc:
            last_exc = exc
            if attempt == DOWNLOAD_RETRY_ATTEMPTS:
                raise
            on_stderr(
                f"workshop {meta.steam_id} attempt {attempt}/"
                f"{DOWNLOAD_RETRY_ATTEMPTS} failed: {exc}"
            )
            delay = DOWNLOAD_RETRY_BACKOFF_SECONDS[attempt - 1]
            if _sleep_with_cancel(delay, should_cancel):
                raise InterruptedError("download cancelled during backoff") from last_exc
```

- [ ] **Step 1.8: Run the tests to verify they pass**

```bash
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py -k "download_with_retry or sleep_with_cancel" -v
```

Expected: all seven PASS.

- [ ] **Step 1.9: Commit**

```bash
git add l4d2web/services/overlay_builders.py l4d2web/tests/test_overlay_builders.py
git commit -m "overlay_builders: add _download_with_retry + _sleep_with_cancel helpers"
```

---

## Task 2: WorkshopBuilder downloads missing/stale items

**Goal:** Replace the current skip-uncached behavior with: download what's missing/stale (via the retry helper from Task 1), stamp `last_downloaded_at`, then run the existing symlink phase. Items with no `file_url` get a clear skip line and do not fail the build.

**Files:**
- Modify: `l4d2web/services/overlay_builders.py` (rewrite `WorkshopBuilder.build`)
- Test: `l4d2web/tests/test_overlay_builders.py`

- [ ] **Step 2.1: Read the existing WorkshopBuilder.build**

```bash
sed -n '71,192p' l4d2web/services/overlay_builders.py
```

Note: the existing `items_data` tuple is `(steam_id, last_downloaded_at)` — needs to grow to carry every field the downloader/decision logic needs.

- [ ] **Step 2.2: Write the failing test for "downloads uncached items, stamps last_downloaded_at"**

Append at the bottom of `l4d2web/tests/test_overlay_builders.py`:

```python
def _make_meta_from_db_row(steam_id: str, *, file_size: int, time_updated: int):
    from l4d2web.services import steam_workshop
    return steam_workshop.WorkshopMetadata(
        steam_id=steam_id, title=f"item-{steam_id}", filename=f"orig-{steam_id}.vpk",
        file_url=f"https://example.com/{steam_id}.vpk", file_size=file_size,
        time_updated=time_updated, preview_url="", consumer_app_id=550, result=1,
    )


def test_workshop_build_downloads_uncached_and_stamps_timestamp(env, tmp_path, monkeypatch):
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    cache_root = tmp_path / "workshop_cache"
    user_id, overlay_id = _create_user_and_overlay("ws", "workshop")
    item_id = _add_workshop_item("2001", downloaded=False, cache_root=cache_root)
    _associate(overlay_id, item_id)

    download_calls = []
    def fake_download(meta, cache_root_arg, *, should_cancel=None):
        download_calls.append(meta.steam_id)
        cache_root_arg.mkdir(parents=True, exist_ok=True)
        (cache_root_arg / f"{meta.steam_id}.vpk").write_bytes(b"data")
        os.utime(cache_root_arg / f"{meta.steam_id}.vpk", (meta.time_updated, meta.time_updated))

    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    out, err, on_stdout, on_stderr = _capture_logs()
    with session_scope() as s:
        overlay = s.scalar(__import__("sqlalchemy").select(Overlay).where(Overlay.id == overlay_id))
        s.expunge(overlay)
    overlay_builders.WorkshopBuilder().build(
        overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False,
    )

    assert download_calls == ["2001"]
    with session_scope() as s:
        from sqlalchemy import select as _select
        wi = s.scalar(_select(WorkshopItem).where(WorkshopItem.id == item_id))
        assert wi.last_downloaded_at is not None
        assert wi.last_error == ""

    addons = tmp_path / "overlays" / str(overlay_id) / "left4dead2" / "addons"
    assert (addons / "2001.vpk").is_symlink()
```

- [ ] **Step 2.3: Run the test to verify it fails**

```bash
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py::test_workshop_build_downloads_uncached_and_stamps_timestamp -v
```

Expected: FAIL — current builder skips uncached items rather than downloading them. The assertion `download_calls == ["2001"]` will fail with `[] != ['2001']`.

- [ ] **Step 2.4: Rewrite `WorkshopBuilder.build`**

In `l4d2web/services/overlay_builders.py`, replace the entire body of `class WorkshopBuilder`'s `build` method (currently lines ~77–191 — find it by its docstring and signature). The new body:

```python
    def build(
        self,
        overlay: Overlay,
        *,
        on_stdout: LogSink,
        on_stderr: LogSink,
        should_cancel: CancelCheck,
    ) -> None:
        addons_dir = _overlay_root(overlay) / "left4dead2" / "addons"
        addons_dir.mkdir(parents=True, exist_ok=True)

        # Snapshot every field the decision logic + downloader will need.
        with session_scope() as db:
            rows = db.scalars(
                select(WorkshopItem)
                .join(
                    OverlayWorkshopItem,
                    OverlayWorkshopItem.workshop_item_id == WorkshopItem.id,
                )
                .where(OverlayWorkshopItem.overlay_id == overlay.id)
            ).all()
            items_data = [
                (
                    it.id,
                    it.steam_id,
                    it.title,
                    it.filename,
                    it.file_url,
                    it.file_size,
                    it.time_updated,
                    it.preview_url,
                    it.last_downloaded_at,
                    it.last_error,
                )
                for it in rows
            ]

        cache_root = workshop_cache_root()
        cache_root.mkdir(parents=True, exist_ok=True)

        downloaded = 0
        cached = 0
        skipped: list[str] = []

        # Download phase.
        for (
            item_id, steam_id, title, filename, file_url, file_size,
            time_updated, preview_url, last_downloaded_at, last_error,
        ) in items_data:
            if should_cancel():
                on_stderr("workshop build cancelled during download phase")
                return
            if not file_url:
                on_stderr(
                    f"workshop item {steam_id} skipped: no file_url "
                    f"(steam result: {last_error or 'unknown'})"
                )
                skipped.append(steam_id)
                continue
            target = cache_path(steam_id)
            needs_download = (
                last_downloaded_at is None
                or not target.exists()
                or int(target.stat().st_mtime) != int(time_updated)
                or int(target.stat().st_size) != int(file_size)
            )
            if not needs_download:
                cached += 1
                continue
            meta = WorkshopMetadata(
                steam_id=steam_id,
                title=title,
                filename=filename,
                file_url=file_url,
                file_size=file_size,
                time_updated=time_updated,
                preview_url=preview_url,
                consumer_app_id=550,
                result=1,
            )
            on_stdout(f"workshop {steam_id} downloading")
            try:
                _download_with_retry(
                    meta, cache_root,
                    on_stderr=on_stderr, should_cancel=should_cancel,
                )
            except InterruptedError:
                raise
            except Exception as exc:
                with session_scope() as db:
                    wi = db.scalar(select(WorkshopItem).where(WorkshopItem.id == item_id))
                    if wi is not None:
                        wi.last_error = f"download failed: {exc}"
                raise
            with session_scope() as db:
                wi = db.scalar(select(WorkshopItem).where(WorkshopItem.id == item_id))
                if wi is not None:
                    wi.last_downloaded_at = datetime.now(UTC)
                    wi.last_error = ""
            downloaded += 1

        # Re-snapshot for symlink phase: only items that have a cache file now
        # belong in the desired set. Items skipped above stay out.
        desired: dict[str, Path] = {}
        for (
            _item_id, steam_id, _title, _filename, _file_url, _file_size,
            _time_updated, _preview_url, _last_downloaded_at, _last_error,
        ) in items_data:
            if steam_id in skipped:
                continue
            target = cache_path(steam_id)
            if not target.exists():
                continue  # shouldn't happen post-download; safety net
            desired[f"{steam_id}.vpk"] = target.resolve()

        if should_cancel():
            on_stderr("workshop build cancelled before applying symlinks")
            return

        # existing: symlink-name -> link target (only symlinks pointing at our cache)
        existing: dict[str, Path] = {}
        for entry in os.scandir(addons_dir):
            if not entry.is_symlink():
                continue
            try:
                link_target = Path(os.readlink(entry.path))
            except OSError:
                continue
            try:
                resolved = link_target.resolve(strict=False)
            except OSError:
                continue
            if not _is_under(resolved, cache_root):
                continue
            existing[entry.name] = resolved

        created = 0
        removed = 0
        unchanged = 0

        for name, current_target in existing.items():
            if should_cancel():
                on_stderr("workshop build cancelled mid-removal")
                return
            desired_target = desired.get(name)
            if desired_target is None:
                os.unlink(addons_dir / name)
                removed += 1
            elif current_target != desired_target:
                os.unlink(addons_dir / name)
            else:
                unchanged += 1

        post_removal_existing = {
            name for name in existing
            if name in desired and existing[name] == desired[name]
        }

        for name, target in desired.items():
            if should_cancel():
                on_stderr("workshop build cancelled mid-creation")
                return
            if name in post_removal_existing:
                continue
            os.symlink(target, addons_dir / name)
            created += 1

        on_stdout(
            f"workshop overlay {overlay.name!r}: "
            f"downloaded={downloaded} cached={cached} skipped={len(skipped)} "
            f"created={created} removed={removed} unchanged={unchanged}"
        )
```

Add the imports `datetime` and `UTC` at the top of the file if not already present:

```python
from datetime import UTC, datetime
```

(Check existing imports first — only add what's missing.)

- [ ] **Step 2.5: Run the test from Step 2.2 to verify it now passes**

```bash
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py::test_workshop_build_downloads_uncached_and_stamps_timestamp -v
```

Expected: PASS.

- [ ] **Step 2.6: Write the remaining builder behavior tests**

Append at the bottom of `l4d2web/tests/test_overlay_builders.py`:

```python
def test_workshop_build_skips_already_cached(env, tmp_path, monkeypatch):
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    cache_root = tmp_path / "workshop_cache"
    user_id, overlay_id = _create_user_and_overlay("ws", "workshop")
    item_id = _add_workshop_item("2002", downloaded=True, cache_root=cache_root)
    # Make the cache file's (mtime, size) match the DB row exactly.
    file_path = cache_root / "2002.vpk"
    os.utime(file_path, (1700000000, 1700000000))
    with session_scope() as s:
        from sqlalchemy import select as _sel, update as _upd
        s.execute(_upd(WorkshopItem).where(WorkshopItem.id == item_id).values(
            file_size=os.path.getsize(file_path), time_updated=1700000000,
        ))
    _associate(overlay_id, item_id)

    called = []
    monkeypatch.setattr(
        overlay_builders, "download_to_cache",
        lambda *a, **kw: called.append(1),
    )

    out, err, on_stdout, on_stderr = _capture_logs()
    with session_scope() as s:
        from sqlalchemy import select as _sel
        overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id))
        s.expunge(overlay)
    overlay_builders.WorkshopBuilder().build(
        overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False,
    )

    assert called == [], "should not call downloader for an already-cached item"
    addons = tmp_path / "overlays" / str(overlay_id) / "left4dead2" / "addons"
    assert (addons / "2002.vpk").is_symlink()


def test_workshop_build_redownloads_stale_cache(env, tmp_path, monkeypatch):
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    cache_root = tmp_path / "workshop_cache"
    user_id, overlay_id = _create_user_and_overlay("ws", "workshop")
    item_id = _add_workshop_item("2003", downloaded=True, cache_root=cache_root)
    # DB says time_updated is newer than what's on disk.
    with session_scope() as s:
        from sqlalchemy import update as _upd
        s.execute(_upd(WorkshopItem).where(WorkshopItem.id == item_id).values(
            file_size=99, time_updated=1800000000,
        ))
    _associate(overlay_id, item_id)

    download_calls = []
    def fake_download(meta, cache_root_arg, *, should_cancel=None):
        download_calls.append(meta.steam_id)
        (cache_root_arg / f"{meta.steam_id}.vpk").write_bytes(b"99bytes____99bytes____99bytes____99bytes____99bytes____99bytes____99bytes____99bytes____99bytes____")
        os.utime(cache_root_arg / f"{meta.steam_id}.vpk", (meta.time_updated, meta.time_updated))
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    out, err, on_stdout, on_stderr = _capture_logs()
    with session_scope() as s:
        from sqlalchemy import select as _sel
        overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id))
        s.expunge(overlay)
    overlay_builders.WorkshopBuilder().build(
        overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False,
    )

    assert download_calls == ["2003"]


def test_workshop_build_skips_items_with_no_file_url(env, tmp_path, monkeypatch):
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    user_id, overlay_id = _create_user_and_overlay("ws", "workshop")
    with session_scope() as s:
        wi = WorkshopItem(
            steam_id="2004", title="gone", filename="",
            file_url="",  # Steam returned result != 1; no URL.
            file_size=0, time_updated=0, preview_url="",
            last_downloaded_at=None, last_error="steam result 9",
        )
        s.add(wi)
        s.flush()
        item_id = wi.id
    _associate(overlay_id, item_id)

    monkeypatch.setattr(
        overlay_builders, "download_to_cache",
        lambda *a, **kw: (_ for _ in ()).throw(AssertionError("must not be called")),
    )

    out, err, on_stdout, on_stderr = _capture_logs()
    with session_scope() as s:
        from sqlalchemy import select as _sel
        overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id))
        s.expunge(overlay)
    overlay_builders.WorkshopBuilder().build(
        overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False,
    )

    assert any("2004" in line and "skipped" in line for line in err)
    addons = tmp_path / "overlays" / str(overlay_id) / "left4dead2" / "addons"
    assert not (addons / "2004.vpk").exists()


def test_workshop_build_fails_when_all_retries_exhausted(env, tmp_path, monkeypatch):
    import requests
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    user_id, overlay_id = _create_user_and_overlay("ws", "workshop")
    item_id = _add_workshop_item("2005", downloaded=False, cache_root=tmp_path / "workshop_cache")
    _associate(overlay_id, item_id)

    monkeypatch.setattr(overlay_builders, "_sleep_with_cancel", lambda *a, **kw: False)
    monkeypatch.setattr(
        overlay_builders, "download_to_cache",
        lambda *a, **kw: (_ for _ in ()).throw(requests.ConnectionError("net")),
    )

    out, err, on_stdout, on_stderr = _capture_logs()
    with session_scope() as s:
        from sqlalchemy import select as _sel
        overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id))
        s.expunge(overlay)
    with pytest.raises(requests.ConnectionError):
        overlay_builders.WorkshopBuilder().build(
            overlay, on_stdout=on_stdout, on_stderr=on_stderr, should_cancel=lambda: False,
        )
    # last_error must be stamped on the WorkshopItem so the UI can surface why.
    with session_scope() as s:
        from sqlalchemy import select as _sel
        wi = s.scalar(_sel(WorkshopItem).where(WorkshopItem.id == item_id))
        assert "download failed" in wi.last_error


def test_workshop_build_cancels_cleanly_during_download_phase(env, tmp_path, monkeypatch):
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    user_id, overlay_id = _create_user_and_overlay("ws", "workshop")
    item_id = _add_workshop_item("2006", downloaded=False, cache_root=tmp_path / "workshop_cache")
    _associate(overlay_id, item_id)

    cancel_flag = {"v": False}
    def fake_download(meta, cache_root, *, should_cancel=None):
        cancel_flag["v"] = True  # Flip cancel during the download.
        raise InterruptedError("cancelled")
    monkeypatch.setattr(overlay_builders, "download_to_cache", fake_download)

    out, err, on_stdout, on_stderr = _capture_logs()
    with session_scope() as s:
        from sqlalchemy import select as _sel
        overlay = s.scalar(_sel(Overlay).where(Overlay.id == overlay_id))
        s.expunge(overlay)
    with pytest.raises(InterruptedError):
        overlay_builders.WorkshopBuilder().build(
            overlay, on_stdout=on_stdout, on_stderr=on_stderr,
            should_cancel=lambda: cancel_flag["v"],
        )
```

- [ ] **Step 2.7: Run the full builder test file**

```bash
.venv/bin/pytest l4d2web/tests/test_overlay_builders.py -v
```

Expected: all tests in the file PASS (new and pre-existing).

- [ ] **Step 2.8: Run job-worker tests as a regression check**

The builder is called from `_run_build_overlay`. Verify nothing there breaks:

```bash
.venv/bin/pytest l4d2web/tests/test_job_worker.py -v
```

Expected: all PASS.

- [ ] **Step 2.9: Commit**

```bash
git add l4d2web/services/overlay_builders.py l4d2web/tests/test_overlay_builders.py
git commit -m "overlay_builders: download missing/stale workshop items inline"
```

---

## Task 3: Per-overlay refresh route

**Goal:** `POST /overlays/{id}/refresh` fetches fresh Steam metadata for this overlay's items, updates the rows, and enqueues a `build_overlay` (which now downloads any stale items by virtue of Task 2).

**Files:**
- Modify: `l4d2web/routes/workshop_routes.py`
- Test: `l4d2web/tests/test_workshop_routes.py`

- [ ] **Step 3.1: Write the failing tests**

Append at the bottom of `l4d2web/tests/test_workshop_routes.py`:

```python
def test_overlay_refresh_owner_can_refresh_and_enqueues_build(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    # Seed one item via the add route.
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    # Mark the prior add-route build as already-finished so refresh can enqueue
    # a fresh one (enqueue_build_overlay coalesces against queued, not completed).
    with session_scope() as session:
        for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
            job.state = "succeeded"

    fresh_meta = steam_workshop.WorkshopMetadata(
        steam_id="1001", title="Item 1001 (updated)", filename="1001.vpk",
        file_url="https://example.com/1001.vpk", file_size=99, time_updated=1800000000,
        preview_url="https://example.com/preview-1001.jpg", consumer_app_id=550, result=1,
    )
    with patch.object(steam_workshop, "fetch_metadata_batch", return_value=[fresh_meta]) as fetch:
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302
    assert response.headers["Location"].startswith("/jobs/")
    fetch.assert_called_once_with(["1001"], mode="refresh")

    with session_scope() as session:
        wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
        assert wi.title == "Item 1001 (updated)"
        assert wi.time_updated == 1800000000
        # last_downloaded_at intentionally NOT touched by refresh.
        new_jobs = session.query(Job).filter_by(
            operation="build_overlay", overlay_id=overlay_id, state="queued"
        ).all()
        assert len(new_jobs) == 1


def test_overlay_refresh_returns_400_when_overlay_empty(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)

    with patch.object(steam_workshop, "fetch_metadata_batch") as fetch:
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 400
    fetch.assert_not_called()


def test_overlay_refresh_forbidden_for_non_owner(overlay_for, env_user):
    app, login, _user_id, _admin_id, overlay_id = overlay_for
    # Create a second non-admin user and have them attempt the refresh.
    with session_scope() as session:
        bob = User(username="bob", password_digest=hash_password("x"), admin=False)
        session.add(bob)
        session.flush()
        bob_id = bob.id
    bob_client = login(bob_id)
    response = bob_client.post(
        f"/overlays/{overlay_id}/refresh",
        headers={"X-CSRF-Token": "test-token"},
    )
    assert response.status_code == 403


def test_overlay_refresh_admin_can_refresh_anyone(overlay_for):
    app, login, user_id, admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )

    admin_client = login(admin_id)
    with _patch_steam([_meta("1001")]):
        response = admin_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302


def test_overlay_refresh_502_on_steam_error(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    with session_scope() as session:
        for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
            job.state = "succeeded"
        baseline_job_count = session.query(Job).filter_by(
            operation="build_overlay", overlay_id=overlay_id, state="queued"
        ).count()

    with patch.object(steam_workshop, "fetch_metadata_batch", side_effect=Exception("boom")):
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 502
    assert b"steam api error" in response.data

    with session_scope() as session:
        n = session.query(Job).filter_by(
            operation="build_overlay", overlay_id=overlay_id, state="queued"
        ).count()
        assert n == baseline_job_count


def test_overlay_refresh_missing_item_records_last_error(overlay_for):
    app, login, user_id, _admin_id, overlay_id = overlay_for
    user_client = login(user_id)
    with _patch_steam([_meta("1001")]):
        user_client.post(
            f"/overlays/{overlay_id}/items",
            data={"input": "1001", "input_mode": "items"},
            headers={"X-CSRF-Token": "test-token"},
        )
    with session_scope() as session:
        for job in session.query(Job).filter_by(operation="build_overlay", state="queued"):
            job.state = "succeeded"

    # fetch_metadata_batch returns nothing for the requested id.
    with patch.object(steam_workshop, "fetch_metadata_batch", return_value=[]):
        response = user_client.post(
            f"/overlays/{overlay_id}/refresh",
            headers={"X-CSRF-Token": "test-token"},
        )
    assert response.status_code == 302
    with session_scope() as session:
        wi = session.query(WorkshopItem).filter_by(steam_id="1001").one()
        assert "no entry" in wi.last_error
```

- [ ] **Step 3.2: Run the tests to verify they fail**

```bash
.venv/bin/pytest l4d2web/tests/test_workshop_routes.py -k "overlay_refresh" -v
```

Expected: all six FAIL with 404 (the route doesn't exist yet).

- [ ] **Step 3.3: Implement the route**

In `l4d2web/routes/workshop_routes.py`, add a new route. Place it after the existing `remove_item` route and before the `admin_refresh` route:

```python
@bp.post("/overlays/<int:overlay_id>/refresh")
@require_login
def refresh_overlay(overlay_id: int) -> Response:
    user = current_user()
    assert user is not None
    with session_scope() as db:
        overlay, err = _check_workshop_overlay_access(overlay_id, user, db)
        if err is not None:
            return err
        steam_ids = list(
            db.scalars(
                select(WorkshopItem.steam_id)
                .join(
                    OverlayWorkshopItem,
                    OverlayWorkshopItem.workshop_item_id == WorkshopItem.id,
                )
                .where(OverlayWorkshopItem.overlay_id == overlay_id)
            ).all()
        )

    if not steam_ids:
        return Response("overlay has no items", status=400)

    try:
        metas = steam_workshop.fetch_metadata_batch(steam_ids, mode="refresh")
    except Exception as exc:
        return Response(f"steam api error: {exc}", status=502)

    metas_by_id = {m.steam_id: m for m in metas}
    with session_scope() as db:
        overlay, err = _check_workshop_overlay_access(overlay_id, user, db)
        if err is not None:
            return err
        for steam_id in steam_ids:
            wi = db.scalar(select(WorkshopItem).where(WorkshopItem.steam_id == steam_id))
            if wi is None:
                continue
            meta = metas_by_id.get(steam_id)
            if meta is None:
                wi.last_error = "steam returned no entry for this item"
                continue
            wi.title = meta.title
            wi.filename = meta.filename
            wi.file_url = meta.file_url
            wi.file_size = meta.file_size
            wi.time_updated = meta.time_updated
            wi.preview_url = meta.preview_url
            wi.last_error = "" if meta.result == 1 else f"steam result {meta.result}"
        job = enqueue_build_overlay(db, overlay_id=overlay_id, user_id=user.id)
        job_id = job.id

    return redirect(f"/jobs/{job_id}")
```

- [ ] **Step 3.4: Run the tests to verify they pass**

```bash
.venv/bin/pytest l4d2web/tests/test_workshop_routes.py -k "overlay_refresh" -v
```

Expected: all six PASS.

- [ ] **Step 3.5: Run the full workshop-routes test file as a regression check**

```bash
.venv/bin/pytest l4d2web/tests/test_workshop_routes.py -v
```

Expected: all PASS.

- [ ] **Step 3.6: Commit**

```bash
git add l4d2web/routes/workshop_routes.py l4d2web/tests/test_workshop_routes.py
git commit -m "workshop_routes: add per-overlay refresh endpoint"
```

---

## Task 4: Refresh button in the overlay detail template

**Goal:** A "Refresh" button next to the existing "Add items" form on workshop-type overlays. Submits the new route; uses the same CSRF token pattern as the add form.

**Files:**
- Modify: `l4d2web/templates/overlay_detail.html`
- Test: `l4d2web/tests/test_pages.py` (smoke test that the button renders)

- [ ] **Step 4.1: Read the current workshop section of the template**

```bash
sed -n '41,62p' l4d2web/templates/overlay_detail.html
```

Confirm the existing form ends at `</form>` on line 53 and the conditional wrapper ends at `{% endif %}` on line 54.

- [ ] **Step 4.2: Add the Refresh form**

Edit `l4d2web/templates/overlay_detail.html`. Replace:

```html
    <button type="submit">Add</button>
  </form>
  {% endif %}

  <div id="overlay-item-table">
```

with:

```html
    <button type="submit">Add</button>
  </form>
  <form method="post" action="/overlays/{{ overlay.id }}/refresh" class="stack workshop-refresh-form">
    <input type="hidden" name="csrf_token" value="{{ session.get('csrf_token', '') }}">
    <button type="submit">Refresh from Steam</button>
  </form>
  {% endif %}

  <div id="overlay-item-table">
```

(The Refresh button sits inside the same `{% if can_edit and not latest_build_is_running %}` block as Add — same visibility rules.)

- [ ] **Step 4.3: Write a smoke test that the button renders**

Find the existing test that loads an overlay detail page for a workshop overlay. From `grep`:

```bash
grep -n "overlay_detail\|/overlays/<\|test_overlay_detail" l4d2web/tests/test_pages.py | head -20
```

Append (or add near similar tests) in `l4d2web/tests/test_pages.py`:

```python
def test_workshop_overlay_detail_renders_refresh_button(client_with_user_and_workshop_overlay):
    """The owner sees a 'Refresh from Steam' button on their workshop overlay."""
    client, overlay_id = client_with_user_and_workshop_overlay
    response = client.get(f"/overlays/{overlay_id}")
    assert response.status_code == 200
    body = response.get_data(as_text=True)
    assert "Refresh from Steam" in body
    assert f'action="/overlays/{overlay_id}/refresh"' in body
```

If the fixture `client_with_user_and_workshop_overlay` doesn't exist, inline-build the setup using the same pattern as other detail-page tests in that file (look for one that calls `POST /overlays` with `type=workshop` then `GET /overlays/{id}`). The test should be small enough that copying ~10 lines of setup is preferable to inventing a new fixture.

- [ ] **Step 4.4: Run the new test to confirm it passes**

```bash
.venv/bin/pytest l4d2web/tests/test_pages.py::test_workshop_overlay_detail_renders_refresh_button -v
```

Expected: PASS.

- [ ] **Step 4.5: Run the full pages test file as a regression check**

```bash
.venv/bin/pytest l4d2web/tests/test_pages.py -v
```

Expected: all PASS.

- [ ] **Step 4.6: Commit**

```bash
git add l4d2web/templates/overlay_detail.html l4d2web/tests/test_pages.py
git commit -m "overlay_detail: add 'Refresh from Steam' button for workshop overlays"
```

---

## Task 5: `flask workshop-refresh` CLI subcommand

**Goal:** Idempotent CLI that inserts a `refresh_workshop_items` job (or returns the id of one that's already queued/running). The systemd timer calls this.

**Files:**
- Modify: `l4d2web/cli.py`
- Test: `l4d2web/tests/test_cli.py` (create if absent)

- [ ] **Step 5.1: Check whether `test_cli.py` exists**

```bash
ls l4d2web/tests/test_cli.py 2>&1 || echo MISSING
```

If MISSING, the test fixture template lives in `test_seed_script_overlays.py` (uses `CliRunner` + `init_db()` against a temp DB). Mirror that pattern.

- [ ] **Step 5.2: Write the failing tests**

Create or append to `l4d2web/tests/test_cli.py`:

```python
"""Tests for the l4d2web Flask CLI subcommands."""
from __future__ import annotations
from click.testing import CliRunner
import pytest
from sqlalchemy import select

from l4d2web.app import create_app
from l4d2web.cli import workshop_refresh
from l4d2web.db import init_db, session_scope
from l4d2web.models import Job


@pytest.fixture
def app_env(tmp_path, monkeypatch):
    db_url = f"sqlite:///{tmp_path/'cli.db'}"
    monkeypatch.setenv("DATABASE_URL", db_url)
    monkeypatch.setenv("LEFT4ME_ROOT", str(tmp_path))
    app = create_app({"TESTING": True, "DATABASE_URL": db_url, "SECRET_KEY": "test"})
    init_db()
    return app


def test_workshop_refresh_enqueues_job(app_env):
    runner = CliRunner()
    with app_env.app_context():
        result = runner.invoke(workshop_refresh, [])
    assert result.exit_code == 0, result.output
    assert "enqueued refresh_workshop_items job" in result.output
    with session_scope() as db:
        jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all()
        assert len(jobs) == 1
        assert jobs[0].state == "queued"
        assert jobs[0].user_id is None
        assert jobs[0].server_id is None


def test_workshop_refresh_is_idempotent_when_job_queued(app_env):
    runner = CliRunner()
    with app_env.app_context():
        first = runner.invoke(workshop_refresh, [])
        second = runner.invoke(workshop_refresh, [])
    assert first.exit_code == 0
    assert second.exit_code == 0
    assert "already queued" in second.output
    with session_scope() as db:
        jobs = db.scalars(select(Job).where(Job.operation == "refresh_workshop_items")).all()
        assert len(jobs) == 1, "must not insert a second job when one is already queued"
```

- [ ] **Step 5.3: Run the tests to verify they fail**

```bash
.venv/bin/pytest l4d2web/tests/test_cli.py -v
```

Expected: FAIL with `ImportError: cannot import name 'workshop_refresh' from 'l4d2web.cli'`.

- [ ] **Step 5.4: Implement the CLI command**

In `l4d2web/cli.py`, add the imports and command. After the existing `from l4d2web.models import Overlay, User` line, extend the import:

```python
from l4d2web.models import Job, Overlay, User
```

Add the new command after `seed_script_overlays`:

```python
@click.command("workshop-refresh")
def workshop_refresh() -> None:
    """Enqueue a global workshop refresh job. Idempotent: if a refresh is
    already queued or running, prints its id and exits 0."""
    with session_scope() as db:
        existing = db.scalar(
            select(Job)
            .where(
                Job.operation == "refresh_workshop_items",
                Job.state.in_(("queued", "running", "cancelling")),
            )
            .order_by(Job.id.desc())
            .limit(1)
        )
        if existing is not None:
            click.echo(
                f"refresh_workshop_items job {existing.id} already {existing.state}"
            )
            return
        job = Job(
            user_id=None,
            server_id=None,
            operation="refresh_workshop_items",
            state="queued",
        )
        db.add(job)
        db.flush()
        click.echo(f"enqueued refresh_workshop_items job {job.id}")
```

Register the command in `register_cli`:

```python
def register_cli(app) -> None:
    app.cli.add_command(promote_admin)
    app.cli.add_command(create_user)
    app.cli.add_command(seed_script_overlays)
    app.cli.add_command(workshop_refresh)
```

- [ ] **Step 5.5: Run the tests to verify they pass**

```bash
.venv/bin/pytest l4d2web/tests/test_cli.py -v
```

Expected: both PASS.

- [ ] **Step 5.6: Verify the command is registered on the Flask app**

```bash
.venv/bin/flask --app l4d2web.app:create_app --help 2>&1 | grep workshop-refresh
```

Expected: line containing `workshop-refresh   Enqueue a global workshop refresh job.`

- [ ] **Step 5.7: Commit**

```bash
git add l4d2web/cli.py l4d2web/tests/test_cli.py
git commit -m "cli: add workshop-refresh subcommand for scheduled global refresh"
```

---

## Task 6: systemd `.service` and `.timer`

**Goal:** Drop two unit files into `deploy/files/usr/local/lib/systemd/system/` and wire them into the deploy script.

**Files:**
- Create: `deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.service`
- Create: `deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.timer`
- Modify: `deploy/deploy-test-server.sh` (add `systemctl enable --now` for the timer)

- [ ] **Step 6.1: Inspect the existing `left4me-web.service` for the right env/path conventions**

```bash
cat deploy/files/usr/local/lib/systemd/system/left4me-web.service
```

Confirm: `User=left4me`, `EnvironmentFile=/etc/left4me/host.env` + `/etc/left4me/web.env`, `WorkingDirectory=/opt/left4me`, venv at `/opt/left4me/.venv`.

- [ ] **Step 6.2: Create the service unit**

Write `deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.service`:

```ini
[Unit]
Description=left4me daily workshop refresh (enqueue job)
After=network-online.target left4me-web.service
Requires=left4me-web.service

[Service]
Type=oneshot
User=left4me
Group=left4me
WorkingDirectory=/opt/left4me
Environment=HOME=/var/lib/left4me
Environment=PATH=/opt/left4me/.venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
EnvironmentFile=/etc/left4me/host.env
EnvironmentFile=/etc/left4me/web.env
ExecStart=/opt/left4me/.venv/bin/flask --app l4d2web.app:create_app workshop-refresh
```

(No `[Install]` section — the timer pulls the service in via `Unit=`.)

- [ ] **Step 6.3: Create the timer unit**

Write `deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.timer`:

```ini
[Unit]
Description=left4me daily workshop refresh

[Timer]
OnCalendar=*-*-* 04:00:00
Persistent=true
RandomizedDelaySec=15min
Unit=left4me-workshop-refresh.service

[Install]
WantedBy=timers.target
```

- [ ] **Step 6.4: Wire the timer into `deploy-test-server.sh`**

Find the section that enables/starts `left4me-web.service`:

```bash
grep -n "left4me-web.service\|systemctl enable" deploy/deploy-test-server.sh | head -10
```

Find the line that does `systemctl enable --now left4me-web.service`. After it, add:

```bash
ssh "${REMOTE}" -- 'sudo systemctl daemon-reload && sudo systemctl enable --now left4me-workshop-refresh.timer'
```

(Match the existing `ssh "${REMOTE}"` invocation style — whatever exact shell pattern the file uses.)

- [ ] **Step 6.5: Lint the unit files**

If `systemd-analyze` is available locally:

```bash
systemd-analyze verify deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.service
systemd-analyze verify deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.timer
```

Expected: no output (silence = success). If not available, skip — the deploy host validates on `daemon-reload`.

- [ ] **Step 6.6: Commit**

```bash
git add deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.service \
        deploy/files/usr/local/lib/systemd/system/left4me-workshop-refresh.timer \
        deploy/deploy-test-server.sh
git commit -m "deploy: schedule daily workshop refresh via systemd timer"
```

---

## Task 7: Full-suite smoke + manual verification

**Goal:** Confirm nothing broke; smoke-test on the dev host that the timer actually enqueues a job.

- [ ] **Step 7.1: Run the entire pytest suite**

```bash
.venv/bin/pytest l4d2web/tests/ -v
```

Expected: all PASS.

- [ ] **Step 7.2: Type-check (if `mypy` / `ruff` are configured)**

```bash
.venv/bin/ruff check l4d2web/services/overlay_builders.py l4d2web/routes/workshop_routes.py l4d2web/cli.py
```

Expected: clean.

- [ ] **Step 7.3: Manual on-host smoke (after deploy)**

On the dev host (the user knows the host name — `ckn@10.0.4.128` per project memory):

```bash
# Confirm timer is loaded and enabled.
sudo systemctl status left4me-workshop-refresh.timer
sudo systemctl list-timers left4me-workshop-refresh.timer

# Trigger the service once, out-of-schedule, to verify wiring.
sudo systemctl start left4me-workshop-refresh.service
sudo journalctl -u left4me-workshop-refresh.service -n 20 --no-pager
```

Expected: journal contains `enqueued refresh_workshop_items job N`. Visit `/admin/jobs` in the browser; the job should appear and progress through `queued → running → succeeded`.

- [ ] **Step 7.4: Manual on-host smoke (browser, end-to-end on-add download)**

In the browser, against the dev host:

1. Create a fresh workshop overlay.
2. Add one workshop item ID (e.g. `129283155` — Death Aboard 2).
3. Verify the redirect lands on `/jobs/{id}`.
4. Confirm the job log shows `workshop {steam_id} downloading` then a summary line `downloaded=1 cached=0 …` and finishes `succeeded`.
5. Confirm the cache file exists on disk:

```bash
ssh ckn@10.0.4.128 -- ls -la /var/lib/left4me/workshop_cache/
```

Expected: `129283155.vpk` present, non-zero size.

- [ ] **Step 7.5: Final review commit (if anything was tweaked during smoke)**

If smoke testing surfaced no fixes needed, this step is a no-op. Otherwise, commit the fix with a focused message.

---

## Self-Review Notes (author's check, recorded for the executor)

- Spec coverage: every component of the spec (download-on-build, retry/backoff, per-overlay refresh, CLI + timer, no-file_url skip) has a dedicated task.
- Type consistency: `_download_with_retry` signature is identical across Task 1 (definition) and Task 2 (call site). `WorkshopMetadata` import is added once in Task 1.
- Schema check: design listed `Job.user_id` nullability as TBD. Verified during plan authoring — already nullable; no migration step is in the plan.
- `download_to_cache` import in `overlay_builders.py`: Task 1 introduces the import. Task 2 uses it. The existing file already imports from `steam_workshop` for cache helpers, so the new import sits alongside.
- Tests vs. monkeypatching of `download_to_cache`: tests monkeypatch the symbol *on `overlay_builders`* (not on `steam_workshop`) because the builder imports it once at module load and binds the name locally. The Task 1 implementation snippet uses `from l4d2web.services.steam_workshop import download_to_cache, WorkshopMetadata` — this puts a `download_to_cache` name on the `overlay_builders` module, which is exactly what the tests target.