From 06c718a9e13124fce79afe8c3f66b0260cbeab43 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:02:46 +0800
Subject: [PATCH 01/10] =?UTF-8?q?UI-SP5=20spec=20=E2=80=94=20realtime=20SS?=
 =?UTF-8?q?E=20swap=20(view-free,=20smallest=20credible=20delivery)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

1 additive backend SSE endpoint (GET /tasks/{id}/stream, hand-rolled
StreamingResponse, zero new dep, zero migration) + opt-in extension of
useLiveResource with streamUrl/applyEvent (additive; existing callers
unaffected) + ONE consumer (useTaskDetail) opts in. 9 other composables stay
polling = truly view-free. fetch+ReadableStream (Bearer-via-header) over
EventSource (would leak JWT via ?token query string). UI-SP4 AI-Copilot
deferred (v2.1 scope, requires full AI backend not implemented).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../2026-05-20-ui-sp5-realtime-sse-design.md  | 362 ++++++++++++++++++
 1 file changed, 362 insertions(+)
 create mode 100644 docs/superpowers/specs/2026-05-20-ui-sp5-realtime-sse-design.md

diff --git a/docs/superpowers/specs/2026-05-20-ui-sp5-realtime-sse-design.md b/docs/superpowers/specs/2026-05-20-ui-sp5-realtime-sse-design.md
new file mode 100644
index 0000000..a763cfe
--- /dev/null
+++ b/docs/superpowers/specs/2026-05-20-ui-sp5-realtime-sse-design.md
@@ -0,0 +1,362 @@
+# UI-SP5 — Realtime SSE Swap (Design)
+
+> Sub-project 5 of 5 (UI-SP1 = PR #19; UI-SP2 = PR #20 `de9573a`; UI-SP3 = PR #21 `a32f165`).
+> UI-SP4 (AI-Copilot, v2.1) is **explicitly deferred** as it requires the full AI backend
+> (LLM bridge, `/api/ai/chat` SSE, conversation tables, MCP→REST tool bridge) which is
+> 0% implemented.
+> Status: design self-approved per project Rule #1 (autonomous; recommended/conservative at every fork).
+> Branch: `feat/ui-sp5-realtime-sse`.
+
+## 1. Context & The Locked Promise
+
+SP1, SP2, and SP3 specs made the SAME promise multiple times — verbatim across files:
+
+- `2026-05-19-ui-sp1-shell-tasks-design.md` §5 line 54: *"All views consume this, never
+  vue-query directly → SP5 swaps internals to SSE/WS with **zero view changes**."*
+- `2026-05-20-ui-sp2-task-detail-design.md` §1 line 39 (out-of-scope): *"SSE/WS push
+  (UI-SP5). `useLiveResource` stays the single live seam; SP5 **swaps internals
+  view-free**."*
+- `2026-05-20-ui-sp3-infra-governance-design.md` §2 line 86: *"`useLiveResource` is the
+  **only** realtime seam (with the SP2-added optional `enabled`)."*
+
+SP5's job is to **deliver that promise** — internally swap polling for a push
+transport without touching a single view, composable consumer, or page.
+
+## 2. Scope
+
+**Conservative, smallest-credible delivery.**
+
+**In scope (additive, zero migration):**
+
+1. **Backend — one new SSE endpoint over the existing schema:**
+   - `GET /api/v1/tasks/{id}/stream` — emits the same `TaskDetail` payload that
+     `GET /api/v1/tasks/{id}` returns today, **on a 1 Hz server-side tick** until
+     the client disconnects or the task reaches a terminal status. Hand-rolled
+     `text/event-stream` via FastAPI `StreamingResponse` (same idiom as
+     `src/dlw/api/hf_proxy.py:99-110` and `source_proxy.py`). New file
+     `src/dlw/api/tasks_stream.py` (additive — doesn't touch `api/tasks.py`).
+     Uses the existing `require_perm("/api/v1/tasks*", "GET")` + the **proven
+     cancel-pattern tenant gate** (`tenant_filtered(select(DownloadTask.id)...)`
+     → 404 if cross-tenant). Reuses the existing `_session` per-tick pattern (a
+     fresh `async_sessionmaker(get_engine())` session per emit — matches the
+     stateless model and avoids long-held connections).
+
+2. **Frontend — additive `useLiveResource` extension + ONE consumer opt-in:**
+   - **New file `frontend/src/api/sse.ts`**:
+     - `parseSseChunk(buffer: string): { events: SseEvent[]; remainder: string }` —
+       a **pure function** that parses one or more concatenated `data: …\n\n`
+       blocks; handles comment lines (`:keep-alive`), multi-line `data:` fields,
+       and a trailing partial block returned as `remainder`. The remainder is
+       prepended to the next chunk by the caller.
+     - `streamSse(url, token, onEvent, signal): Promise<void>` —
+       `fetch(url, { headers: { Authorization: "Bearer <jwt>", Accept: "text/event-stream" }, signal })`
+       → if 401, reject (caller logs out); else read `response.body.getReader()`,
+       decode with `TextDecoder`, accumulate, parse via `parseSseChunk`, invoke
+       `onEvent`. On non-401 disconnect, exponential backoff `1s → 2s → 4s → 8s
+       → 16s → 30s cap` with ±20 % jitter, reset to 1 s after each successful
+       chunk. Aborts when `signal.aborted`. **~70 lines incl. backoff; no new
+       runtime dep.**
+   - **Additive change to `frontend/src/composables/useLiveResource.ts`**:
+     extend `LiveOptions<T>` with two optional fields:
+     ```ts
+     streamUrl?: string | Ref<string>
+     applyEvent?: (prev: T | undefined, ev: SseEvent) => T
+     ```
+     Semantics: when both are set and `enabled !== false`, after the initial
+     `useQuery` fetch lands the snapshot, the composable opens `streamSse`,
+     calls `applyEvent` per event and `queryClient.setQueryData(key, next)`,
+     and sets `refetchInterval: false` while streaming is healthy. On
+     persistent stream error (3 consecutive backoff failures) it logs a console
+     warning and lets the normal `computeInterval` polling resume. **View
+     contract unchanged**: the returned `useQuery` result and `LiveOptions`
+     defaults are untouched for every existing caller (`streamUrl === undefined`
+     ⇒ identical behavior to today).
+   - **ONE consumer opts in — `frontend/src/composables/useTaskDetail.ts`**: add
+     `streamUrl: computed(() => '/api/v1/tasks/' + taskId.value + '/stream')`
+     and `applyEvent: (_prev, ev) => JSON.parse(ev.data) as TaskDetail`. Nothing
+     else changes — `useTaskDetail`'s return shape is identical, every consumer
+     of `useTaskDetail` (the SP1 TaskDetail header, the rebuilt SP2 page) sees
+     reactive `data` exactly as before.
+
+3. **Headed Playwright smoke**: open Task Detail of a running task, observe at
+   least 2 distinct snapshots arriving within 5 s; confirm zero console errors.
+
+**Out of scope (intentional, documented):**
+
+- The 4 SP2 sub-resource composables (`useSubtaskChunks` / `useSourceAllocation`
+  / `useParticipatingExecutors` / `useTaskEvents`). Opting these in would
+  require 4 more backend SSE endpoints or a multi-resource envelope (which
+  would force fan-out in the page → **breaks view-free**). The polling at
+  1.5 / 2 / 2 / 5 s is already responsive; the per-host browser connection
+  cap (~6) becomes a concern if every Task Detail open consumes 5 SSE
+  connections. Deferred to a future SP5b if real telemetry justifies it.
+- The 5 lower-frequency composables (`useTaskList` 5 s, `useQuota` 30 s,
+  `useExecutors` 5 s, `useAuditLog` 10 s, `useSystemHealth` 10 s) — push value
+  is negligible at their cadences; cost (5 endpoints + 5 connections) is high.
+  Deferred.
+- WebSocket transport. The locked promise reads "SSE/WS"; the spec authoring
+  agent confirms either satisfies the promise; SSE is the strictly simpler
+  choice with the same UX outcome. WS is deferred.
+- `EventSource` browser API (query-param token leaks JWT to logs/referrer);
+  rejected in favor of fetch + ReadableStream.
+
+## 3. Inherited Locked Decisions (binding on SP5)
+
+- **`useLiveResource` is the only realtime seam.** SP5 preserves its return
+  shape (`UseQueryReturnType<T, Error>`) byte-faithfully — `data`,
+  `isLoading`, `isError`, `error`, `refetch`, `isFetching`, etc., all
+  unchanged.
+- `DataBoundary` wraps every view (unchanged).
+- **Additive backend only**: new file + new route + new Pydantic DTO
+  (`TaskStreamEvent` envelope) added to `openapi.yaml`; **no** schema /
+  Alembic / existing-route changes.
+- **No new runtime dep** on backend or frontend (hand-rolled SSE on
+  `StreamingResponse`; fetch + ReadableStream on the client).
+- Pass existing CI only.
+- i18n parity (no new UI text in SP5).
+- Tenant / RBAC server-side via the proven cancel-pattern gate + `require_perm`.
+- `noUncheckedIndexedAccess` is on — every `arr[i]` guarded.
+- Frontend tests: `vi.hoisted` plain holders + async `vi.mock(async () => { const { ref } = await import('vue'); … })` when a mock must return a real Vue ref (SP2 BLOCKER-fix pattern).
+- Bash cwd persists across calls → `cd /d/download_weights && git …` for git;
+  `cd /d/download_weights/frontend && pnpm …` for frontend.
+
+## 4. Backend Design
+
+### 4.1 Endpoint
+
+```
+GET /api/v1/tasks/{task_id}/stream
+Authorization: Bearer <tenant-user JWT>
+Accept: text/event-stream
+
+→ 200 OK
+Content-Type: text/event-stream
+Cache-Control: no-cache
+X-Accel-Buffering: no
+
+: tick
+data: <TaskDetail JSON>
+
+data: <TaskDetail JSON>
+
+…
+```
+
+- Initial event emitted **immediately** on connect (no client poll-then-stream
+  race; the first `data:` block is the same payload as `GET /tasks/{id}`).
+- Subsequent events every 1.0 s (configurable via `DLW_TASK_STREAM_INTERVAL`
+  env, default 1.0 s; capped 0.2 s … 10 s).
+- A `:keepalive` SSE comment is emitted every 15 s when no data changed, to
+  defeat intermediary proxies that close idle connections.
+- The handler terminates when (a) client disconnects, (b) task reaches a
+  terminal status (`succeeded` / `failed` / `cancelled`) — emit one final
+  payload and close, (c) the controller lifespan shuts down.
+- Tenant gate is the proven cancel-pattern: `tenant_filtered(select(DownloadTask.id).where(id==task_id), DownloadTask, principal)` → 404 if cross-tenant. Re-checked on each tick (cheap; protects against revoke-during-stream).
+
+### 4.2 OpenAPI contract
+
+Add to `api/openapi.yaml` after the existing `/tasks/{taskId}/events` block:
+
+```yaml
+  /tasks/{taskId}/stream:
+    parameters:
+      - $ref: '#/components/parameters/TaskId'
+    get:
+      tags: [tasks]
+      summary: Live TaskDetail stream (SSE; UI-SP5)
+      operationId: streamTaskDetail
+      responses:
+        '200':
+          description: SSE stream of TaskDetail snapshots
+          content:
+            text/event-stream:
+              schema: {type: string}
+```
+
+No new schemas needed — the SSE payload is the existing `TaskDetail` schema
+serialized into a `data:` line.
+
+### 4.3 RBAC
+
+`policy.csv` already grants `tenant_admin` / `tenant_operator` / `tenant_viewer`
+GET on `/api/v1/tasks*` — the new `/stream` sub-resource is covered by the
+wildcard. No `policy.csv` change required.
+
+### 4.4 Tests
+
+`tests/api/test_task_stream.py`:
+
+1. **Unauth → 401** (no Authorization header).
+2. **Cross-tenant → 404** (different tenant_id JWT).
+3. **Happy path — receives at least 2 envelopes within 4 s** (override
+   `DLW_TASK_STREAM_INTERVAL=0.5`); each envelope parses as JSON containing
+   `id`, `status`, `subtasks`.
+4. **Terminal-stop** — set the seeded task to `succeeded` (direct DB mutate);
+   open the stream; assert exactly one envelope and the response closes (next
+   `iter_lines` returns nothing).
+
+httpx `AsyncClient.stream()` is the right test API — supports reading SSE chunks
+incrementally.
+
+## 5. Frontend Design
+
+### 5.1 Pure parser (`frontend/src/api/sse.ts`)
+
+```ts
+export interface SseEvent { event: string; data: string; id?: string }
+
+export function parseSseChunk(
+  buf: string,
+): { events: SseEvent[]; remainder: string } {
+  const events: SseEvent[] = []
+  const parts = buf.split(/\r?\n\r?\n/)
+  const remainder = parts.pop() ?? ''
+  for (const block of parts) {
+    let event = 'message', data = '', id: string | undefined
+    for (const line of block.split(/\r?\n/)) {
+      if (line.startsWith(':')) continue
+      const i = line.indexOf(':')
+      const field = i === -1 ? line : line.slice(0, i)
+      const value = i === -1 ? '' : line.slice(i + 1).replace(/^ /, '')
+      if (field === 'event') event = value
+      else if (field === 'data') data += (data ? '\n' : '') + value
+      else if (field === 'id') id = value
+    }
+    if (data) events.push({ event, data, ...(id !== undefined ? { id } : {}) })
+  }
+  return { events, remainder }
+}
+```
+
+### 5.2 Streamer
+
+```ts
+export interface StreamOptions {
+  url: string
+  token: string | null
+  onEvent: (ev: SseEvent) => void
+  onUnauthorized: () => void   // logout helper
+  signal: AbortSignal
+}
+```
+
+`streamSse(opts)`: fetch + ReadableStream + exponential backoff (`1s → 2s → 4s
+→ 8s → 16s → 30s cap`, ±20% jitter). On 401, call `onUnauthorized()` and
+reject; do NOT retry. On any other disconnect, retry until `signal.aborted`.
+Reset backoff to 1 s after the first successful chunk.
+
+### 5.3 `useLiveResource` opt-in
+
+Augment `LiveOptions<T>` (the only file modified beyond additions):
+
+```ts
+export interface LiveOptions<T> {
+  baseIntervalMs: number
+  isTerminal?: (data: T) => boolean
+  staleTime?: number
+  enabled?: Ref<boolean> | boolean
+  streamUrl?: string | Ref<string>            // NEW
+  applyEvent?: (prev: T | undefined, ev: SseEvent) => T  // NEW
+}
+```
+
+Internals: keep the existing `useQuery` block (fetcher fires once for the
+snapshot, vue-query handles loading/cache/retries). After mount, when
+`streamUrl` and `applyEvent` are both set and the query reports a first
+success, set `refetchInterval: false` for vue-query and open `streamSse`. On
+each event, compute `next = applyEvent(prev, ev)` and call
+`queryClient.setQueryData(key, next)`. On 3 consecutive backoff failures, log a
+warning and revert to polling (re-enable `refetchInterval`).
+
+### 5.4 ONE consumer opts in
+
+`frontend/src/composables/useTaskDetail.ts`:
+
+```ts
+import { computed, type Ref } from 'vue'
+import { useLiveResource } from '@/composables/useLiveResource'
+import { client } from '@/api/client'
+import { TERMINAL_STATUSES, type TaskDetail } from '@/api/types'
+
+export function useTaskDetail(taskId: Ref<string>) {
+  const streamUrl = computed(() =>
+    `/api/v1/tasks/${taskId.value}/stream`)
+  return useLiveResource<TaskDetail>(
+    ['task', taskId],
+    async () => (await client.get<TaskDetail>(
+      `/api/v1/tasks/${taskId.value}`)).data,
+    {
+      baseIntervalMs: 1_000,
+      isTerminal: (d) => TERMINAL_STATUSES.has(d.status),
+      streamUrl,
+      applyEvent: (_prev, ev) => JSON.parse(ev.data) as TaskDetail,
+    },
+  )
+}
+```
+
+This is THE ONLY consumer change. Every other composable + every view is
+untouched.
+
+### 5.5 Tests
+
+- **`frontend/tests/unit/parseSseChunk.spec.ts`** — 6 cases: single event;
+  two events in one buffer; event split across two buffers (remainder carry);
+  comment lines ignored; custom `event:` field; multi-line `data:`.
+- **`frontend/tests/unit/useLiveResource.stream.spec.ts`** — mock `streamSse`
+  (extract into a parameter/module mock) and a fake `QueryClient`; assert:
+  (a) initial snapshot via `fetcher`; (b) `data.value` updates after a mocked
+  event via `queryClient.setQueryData`; (c) when the simulated stream
+  give-up signal fires, polling resumes; (d) `enabled: false` opens neither
+  poll nor stream.
+- **Headed Playwright smoke**: open `/tasks/<id>` of a running task, capture
+  two distinct snapshots within 5 s, assert zero console errors.
+
+## 6. Milestones
+
+- **M1 Backend**: `tasks_stream.py` + `openapi.yaml` path + 4 pytest tests + `main.py`
+  router include + M1 gate (pytest + spectral + swagger + invariants).
+- **M2 Frontend foundation**: `api/sse.ts` (pure + streamer) + `parseSseChunk` spec
+  + M2 gate.
+- **M3 Cutover**: `useLiveResource` accepts new options + `useTaskDetail` opts in +
+  composable spec + full frontend gate + headed Playwright smoke + docs update.
+
+## 7. Risks & Contingencies
+
+- **The Playwright smoke might not surface a status change within the 5 s
+  window** if the seeded task is idle. Mitigation: the test creates a *new* task
+  via the existing API (already proven in SP2 smoke), then asserts that the
+  *initial* snapshot arrives within 2 s (proves the stream opens and emits at
+  least once). Status-flip on a busy executor is bonus.
+- **Cancellation propagation**: if the FastAPI request is cancelled (client
+  disconnects), the `async def _body()` generator's `finally` must close the
+  DB session. Standard `async with` handles this; the pattern in `hf_proxy.py`
+  is the proven reference.
+- **vue-query v5 `queryClient.setQueryData` reactivity**: vue-query DOES
+  notify watchers on `setQueryData`. Confirmed by reading the same surface
+  used by `useTaskMutations`' optimistic update pattern. No risk.
+- **Backoff with fake-timers in vitest**: any test that exercises `streamSse`'s
+  backoff timing must run inside `vi.useFakeTimers()` + manual `vi.advanceTimersByTime`. The pure parser tests don't need timers; the composable
+  spec mocks `streamSse` directly (no backoff execution in tests).
+- **Contingency**: if the SSE endpoint slips integration, the spec's M3 can
+  still merge with the seam fully wired but `useTaskDetail` left without
+  `streamUrl` — the realtime seam is in place, polling stays in production,
+  and the cutover lands as a one-line follow-up. (Same pattern recorded in
+  the SP2 plan's §7.)
+
+## 8. Self-Review
+
+- **Placeholder scan**: none. Every endpoint, helper, composable change, and
+  test has concrete shape.
+- **Consistency**: the single-seam contract is preserved (§3, §5.3); zero view
+  changes (§2, §5.4); additive backend (§4); inherits all SP1/SP2/SP3 locked
+  decisions verbatim (§3).
+- **Scope**: deliberately the smallest credible delivery — 1 backend endpoint,
+  1 consumer opted in, all others unchanged. The decomposition documented SP5
+  as "optional realtime upgrade"; this lands the architecture, demonstrates
+  end-to-end live updates, and leaves a clear path for SP5b (per-resource
+  streams) if telemetry justifies it.
+- **Ambiguity**: endpoint path, DTO shape, tick interval, backoff schedule,
+  auth (Bearer via fetch), terminal-stop semantics, and the consumer that opts
+  in are all pinned. The contingency (M3 ships without `useTaskDetail` opt-in)
+  is documented.

From aa61121539b46f0c9e38bb446fd805c800e4acb3 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:07:37 +0800
Subject: [PATCH 02/10] =?UTF-8?q?UI-SP5=20implementation=20plan=20?=
 =?UTF-8?q?=E2=80=94=209=20bite-sized=20TDD=20tasks=20across=20M1-M3?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

M1 backend (openapi+config; SSE route+4 tests; gate). M2 frontend (parseSseChunk
pure parser; streamSse+shouldStream; gate). M3 cutover (useLiveResource opt-in;
useTaskDetail opts in; full gate + headed smoke + docs). Complete code, no
placeholders; grounded in hf_proxy.py StreamingResponse idiom + post-SP3
on-disk state.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../plans/2026-05-20-ui-sp5-realtime-sse.md   | 1081 +++++++++++++++++
 1 file changed, 1081 insertions(+)
 create mode 100644 docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md

diff --git a/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md b/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md
new file mode 100644
index 0000000..f7735bb
--- /dev/null
+++ b/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md
@@ -0,0 +1,1081 @@
+# UI-SP5 — Realtime SSE Swap 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:** Add 1 additive SSE endpoint (`GET /api/v1/tasks/{id}/stream`) + extend `useLiveResource` with `streamUrl` / `applyEvent` options + opt in ONE consumer (`useTaskDetail`) — delivering the locked "single-seam SSE swap, view-free" promise from SP1/SP2/SP3.
+
+**Architecture:** Backend = hand-rolled `text/event-stream` via FastAPI `StreamingResponse` (same idiom as `src/dlw/api/hf_proxy.py:99-110`); fresh DB session per 1 Hz tick; tenant-gated via the proven cancel-pattern. Frontend = new `api/sse.ts` (pure `parseSseChunk` + `streamSse` fetch+ReadableStream with exponential backoff) + additive `LiveOptions<T>` fields (`streamUrl`, `applyEvent`); `useTaskDetail` is the ONLY consumer opting in. **9 other composables stay polling = truly view-free.**
+
+**Tech Stack:** FastAPI · SQLAlchemy 2 async · asyncpg · Pydantic v2 · pytest (httpx `AsyncClient.stream()` for SSE tests) · OpenAPI 3.1 (spectral + swagger-cli) · Vue 3.5 `<script setup>` TS strict · `@tanstack/vue-query` v5.x (`queryClient.setQueryData` for SSE-driven cache writes) · `fetch` + native `ReadableStream` + `TextDecoder` · Vitest 2.1 + happy-dom (pure-fn tests only; the streamer is smoke-tested headed). **Zero new runtime deps.**
+
+---
+
+## Conventions (apply to every task — same as SP3)
+
+- **Branch:** `feat/ui-sp5-realtime-sse` (created off `main` @ `a32f165`; spec committed `06c718a`).
+- **Bash cwd persists**. `cd /d/download_weights && git …` for git; `cd /d/download_weights/frontend && pnpm …` for frontend.
+- **Backend tenant gate**: the proven cancel-pattern (`tenant_filtered(select(DownloadTask.id).where(id==task_id), DownloadTask, principal)` → `None` → 404). Re-checked at the start of each tick (cheap defense against revoke-during-stream).
+- **Run commands**: `uv run pytest tests/api/test_task_stream.py -v`; `uv run pytest tests/ -q`; spectral + swagger-cli; `python tools/lint_invariants.py`; `python tools/lint_no_direct_status_write.py`.
+- **Frontend run**: `pnpm test:unit`, `pnpm typecheck`, `pnpm lint:fix && pnpm lint`, `pnpm build`.
+- **`noUncheckedIndexedAccess`**: guard `arr[i]` with `?? fallback` / optional chaining / length checks.
+- **i18n**: SP5 introduces **no new UI text** — locale files untouched; parity test stays green for free.
+
+---
+
+## File Structure
+
+**Backend (create):**
+- `src/dlw/api/tasks_stream.py` — `GET /api/v1/tasks/{task_id}/stream` route module.
+- `tests/api/test_task_stream.py` — 4 tests.
+
+**Backend (modify):**
+- `api/openapi.yaml` — add the `/tasks/{taskId}/stream` GET path block.
+- `src/dlw/main.py` — register the new router (1 lazy-import + 1 `include_router`).
+- `src/dlw/config.py` — add 1 setting `task_stream_interval_seconds` (env `DLW_TASK_STREAM_INTERVAL`, default `1.0`, clamped `[0.1, 10.0]`).
+
+**Frontend (create):** `frontend/src/api/sse.ts`; tests `frontend/tests/unit/parseSseChunk.spec.ts`, `frontend/tests/unit/streamGate.spec.ts`.
+
+**Frontend (modify):**
+- `frontend/src/composables/useLiveResource.ts` — additive `streamUrl` / `applyEvent` fields on `LiveOptions<T>`; the stream wiring (`useQueryClient().setQueryData(key, applyEvent(...))`) using `fetch`-based `streamSse` from `@/api/sse`.
+- `frontend/src/composables/useTaskDetail.ts` — opt in with `streamUrl` + `applyEvent`.
+
+**Docs (modify, M3):** `docs/operator/web-ui.md`.
+
+---
+
+# Milestone M1 — Backend: 1 additive SSE endpoint
+
+### Task 1: OpenAPI contract + config setting
+
+**Files:**
+- Modify: `api/openapi.yaml` (add path after the existing `/tasks/{taskId}/events` block)
+- Modify: `src/dlw/config.py` (add 1 setting)
+
+- [ ] **Step 1: Add the path block to `api/openapi.yaml`**
+
+Find the existing `/tasks/{taskId}/events:` block (around line 509 — the SP2-declared/UI-SP2-implemented endpoint). Immediately after its response (the line `next_cursor: type string, nullable: true …` from UI-SP3 Task 1 — or its surrounding `}` if SP3 changed the shape), and before the SP2-added `/tasks/{taskId}/subtask-chunks:` block, insert:
+
+```yaml
+
+  /tasks/{taskId}/stream:
+    parameters:
+      - $ref: '#/components/parameters/TaskId'
+    get:
+      tags: [tasks]
+      summary: Live TaskDetail SSE stream (UI-SP5)
+      operationId: streamTaskDetail
+      responses:
+        '200':
+          description: SSE stream of TaskDetail snapshots (text/event-stream)
+          content:
+            text/event-stream:
+              schema:
+                type: string
+                description: |
+                  Each event is `data: <TaskDetail JSON>\n\n`. Stream terminates
+                  on terminal task status, client disconnect, or controller
+                  shutdown. Keep-alive comment lines (`:keepalive`) may appear.
+```
+
+- [ ] **Step 2: Validate the contract**
+
+Run: `npx --yes @stoplight/spectral-cli lint api/openapi.yaml --fail-severity=error`
+Expected: 0 errors. Run: `npx --yes @apidevtools/swagger-cli validate api/openapi.yaml`
+Expected: `api/openapi.yaml is valid`.
+
+- [ ] **Step 3: Add the setting**
+
+Read `src/dlw/config.py` first to find the right block. The settings class is a Pydantic `BaseSettings`. Add a new field alongside the existing ones (look for a similar `*_seconds` float — e.g. `hf_proxy_timeout_seconds` — and place the new field next to it for cohesion):
+
+```python
+    task_stream_interval_seconds: float = 1.0  # UI-SP5: SSE tick rate
+```
+
+The `BaseSettings` `env_prefix` (likely `DLW_`) already maps `DLW_TASK_STREAM_INTERVAL_SECONDS` automatically — **verify** by reading the existing field convention. If the env mapping is custom (e.g. an explicit `Field(..., alias="DLW_TASK_STREAM_INTERVAL")`), follow that pattern. Pin clamping in code, not in the schema:
+
+In the route handler (Task 2) the value is clamped to `[0.1, 10.0]` via `max(0.1, min(10.0, settings.task_stream_interval_seconds))`. **No clamping in config** (keeps the setting trivial).
+
+- [ ] **Step 4: Commit**
+
+```bash
+cd /d/download_weights && git add api/openapi.yaml src/dlw/config.py && git commit -q -m "UI-SP5 M1: openapi /tasks/{id}/stream path + task_stream_interval_seconds setting"
+```
+
+---
+
+### Task 2: SSE route + service tick + 4 tests
+
+**Files:**
+- Create: `src/dlw/api/tasks_stream.py`, `tests/api/test_task_stream.py`
+- Modify: `src/dlw/main.py` (register router — same idiom UI-SP3 pinned: lazy import inside `create_app()`, append after the last existing `include_router`)
+
+- [ ] **Step 1: Write the failing test**
+
+Create `tests/api/test_task_stream.py`:
+
+```python
+"""Tests for GET /api/v1/tasks/{id}/stream (UI-SP5 SSE)."""
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from dlw.config import get_settings
+from dlw.db.base import Base
+from tests.conftest import make_app_with_state, principal_headers
+
+SECRET = "unit-secret"
+TICK = "0.1"  # 100 ms — keeps tests <2 s while still proving multi-tick
+
+
+@pytest.fixture(scope="module", autouse=True)
+async def _bootstrap(engine):
+    from dlw.db.models.storage import StorageBackend
+    from dlw.db.models.tenant import Project, Tenant, User
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+        await conn.run_sync(Base.metadata.create_all)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with factory() as session:
+        session.add(Tenant(id=1, slug="default", display_name="Default"))
+        session.add(Tenant(id=2, slug="other", display_name="Other"))
+        await session.flush()
+        session.add(Project(id=1, tenant_id=1, name="default"))
+        session.add(User(id=1, tenant_id=1, oidc_subject="dev",
+                         email="d@l", role="tenant_admin"))
+        session.add(StorageBackend(id=1, tenant_id=1, name="default",
+                                   backend_type="s3", config_encrypted=b""))
+        await session.commit()
+    yield
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+
+@pytest.fixture(autouse=True)
+def _set_env(monkeypatch: pytest.MonkeyPatch):
+    get_settings.cache_clear()
+    monkeypatch.setenv("DLW_SYSTEM_JWT_SECRET", SECRET)
+    monkeypatch.setenv("DLW_TASK_STREAM_INTERVAL_SECONDS", TICK)
+    get_settings.cache_clear()
+    yield
+    get_settings.cache_clear()
+
+
+@pytest.fixture(autouse=True)
+def _patch_hf(monkeypatch: pytest.MonkeyPatch):
+    from dlw.services.hf_metadata import RepoFile
+
+    async def fake(*args, **kwargs):
+        return [
+            RepoFile(path="config.json", size=4096, sha256=None),
+            RepoFile(path="model.safetensors", size=64 * 1024, sha256="a" * 64),
+        ]
+    monkeypatch.setattr("dlw.services.task_service.list_repo_tree", fake)
+
+
+@pytest.fixture
+def auth() -> dict[str, str]:
+    return principal_headers(secret=SECRET, role="tenant_admin")
+
+
+@pytest.fixture
+async def client(ephemeral_ca):
+    app = make_app_with_state(ephemeral_ca, enrollment_token="e")
+    async with AsyncClient(transport=ASGITransport(app=app),
+                           base_url="http://test", timeout=10.0) as c:
+        yield c
+
+
+async def _make_task(client, auth) -> str:
+    r = await client.post("/api/v1/tasks", json={
+        "repo_id": "o/stream", "revision": "3" * 40, "storage_id": 1,
+    }, headers=auth)
+    assert r.status_code == 201, r.text
+    return r.json()["id"]
+
+
+async def _collect_events(client, url, headers, *, count, timeout=4.0):
+    received: list[str] = []
+    async with asyncio.timeout(timeout):
+        async with client.stream("GET", url, headers=headers) as resp:
+            assert resp.status_code == 200, await resp.aread()
+            ctype = resp.headers.get("content-type", "")
+            assert "text/event-stream" in ctype, ctype
+            async for line in resp.aiter_lines():
+                if line.startswith("data: "):
+                    received.append(line[len("data: "):])
+                    if len(received) >= count:
+                        return received
+    return received
+
+
+@pytest.mark.slow
+async def test_stream_unauthenticated_401(client: AsyncClient) -> None:
+    async with client.stream(
+        "GET", f"/api/v1/tasks/{uuid.uuid4()}/stream",
+    ) as resp:
+        assert resp.status_code == 401
+
+
+@pytest.mark.slow
+async def test_stream_cross_tenant_404(client: AsyncClient, auth) -> None:
+    tid = await _make_task(client, auth)
+    other = principal_headers(secret=SECRET, role="tenant_admin",
+                              user_id=9, tenant_id=2)
+    async with client.stream(
+        "GET", f"/api/v1/tasks/{tid}/stream", headers=other,
+    ) as resp:
+        assert resp.status_code == 404
+
+
+@pytest.mark.slow
+async def test_stream_emits_multiple_snapshots(
+    client: AsyncClient, auth,
+) -> None:
+    tid = await _make_task(client, auth)
+    received = await _collect_events(
+        client, f"/api/v1/tasks/{tid}/stream", auth, count=2, timeout=3.0)
+    assert len(received) >= 2
+    for raw in received[:2]:
+        payload = json.loads(raw)
+        assert payload["id"] == tid
+        assert "status" in payload
+        assert "subtasks" in payload
+
+
+@pytest.mark.slow
+async def test_stream_terminates_on_terminal_status(
+    client: AsyncClient, auth, engine,
+) -> None:
+    from dlw.db.models.task import DownloadTask
+    tid = await _make_task(client, auth)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with factory() as s:
+        task = await s.get(DownloadTask, uuid.UUID(tid))
+        assert task is not None
+        task.status = "succeeded"  # noqa: direct test seeding
+        await s.commit()
+    received: list[str] = []
+    async with asyncio.timeout(3.0):
+        async with client.stream(
+            "GET", f"/api/v1/tasks/{tid}/stream", headers=auth,
+        ) as resp:
+            assert resp.status_code == 200
+            async for line in resp.aiter_lines():
+                if line.startswith("data: "):
+                    received.append(line)
+    assert len(received) == 1
+    assert "succeeded" in received[0]
+```
+
+- [ ] **Step 2: Run the test to verify it fails**
+
+Run: `uv run pytest tests/api/test_task_stream.py -v`
+Expected: FAIL — route 404 (not implemented).
+
+- [ ] **Step 3: Create the route module**
+
+Create `src/dlw/api/tasks_stream.py`:
+
+```python
+"""GET /api/v1/tasks/{id}/stream — SSE TaskDetail stream (UI-SP5).
+
+Hand-rolled text/event-stream via StreamingResponse (same idiom as hf_proxy.py).
+1 Hz default tick rate; configurable via DLW_TASK_STREAM_INTERVAL_SECONDS
+(clamped [0.1, 10.0] in code). Stream terminates on terminal task status,
+client disconnect, or controller shutdown.
+"""
+from __future__ import annotations
+
+import asyncio
+import uuid
+from collections.abc import AsyncIterator
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.orm import selectinload
+
+from dlw.auth.principal import Principal
+from dlw.authz.deps import require_perm
+from dlw.config import get_settings
+from dlw.db.models.task import DownloadTask
+from dlw.db.session import get_engine
+from dlw.db.tenant_scope import tenant_filtered
+from dlw.schemas.task import TaskDetail
+
+router = APIRouter(prefix="/api/v1/tasks", tags=["tasks"])
+
+_TERMINAL = {"succeeded", "failed", "cancelled"}
+_KEEPALIVE_EVERY_TICKS = 15  # at 1 Hz default, every 15s
+
+
+def _clamped_interval() -> float:
+    raw = float(getattr(
+        get_settings(), "task_stream_interval_seconds", 1.0))
+    return max(0.1, min(10.0, raw))
+
+
+async def _load_detail(
+    session_maker, task_id: uuid.UUID, tenant_id: int,
+) -> TaskDetail | None:
+    async with session_maker() as s:
+        row = (await s.execute(
+            select(DownloadTask).where(
+                DownloadTask.id == task_id,
+                DownloadTask.tenant_id == tenant_id)
+            .options(selectinload(DownloadTask.subtasks))
+        )).scalar_one_or_none()
+        return TaskDetail.model_validate(row) if row is not None else None
+
+
+@router.get("/{task_id}/stream")
+async def stream_task_detail(
+    task_id: uuid.UUID,
+    request: Request,
+    principal: Principal = Depends(require_perm("/api/v1/tasks*", "GET")),
+) -> StreamingResponse:
+    # Tenant gate — proven cancel-pattern. 404 cross-tenant; never leak.
+    session_maker = async_sessionmaker(get_engine(), expire_on_commit=False)
+    async with session_maker() as s:
+        owned = await s.scalar(
+            tenant_filtered(select(DownloadTask.id)
+                            .where(DownloadTask.id == task_id),
+                            DownloadTask, principal))
+    if owned is None:
+        raise HTTPException(status_code=404, detail="task not found")
+
+    interval = _clamped_interval()
+
+    async def _body() -> AsyncIterator[bytes]:
+        ticks_since_data = 0
+        try:
+            while True:
+                if await request.is_disconnected():
+                    return
+                detail = await _load_detail(
+                    session_maker, task_id, principal.tenant_id)
+                if detail is None:
+                    # Task deleted mid-stream.
+                    return
+                yield (f"data: {detail.model_dump_json()}"
+                       "\n\n").encode("utf-8")
+                ticks_since_data = 0
+                if detail.status in _TERMINAL:
+                    return
+                # Sleep in small slices so cancellation propagates quickly.
+                slept = 0.0
+                while slept < interval:
+                    if await request.is_disconnected():
+                        return
+                    chunk = min(0.05, interval - slept)
+                    await asyncio.sleep(chunk)
+                    slept += chunk
+                ticks_since_data += 1
+                if ticks_since_data >= _KEEPALIVE_EVERY_TICKS:
+                    yield b":keepalive\n\n"
+                    ticks_since_data = 0
+        except asyncio.CancelledError:
+            # Client disconnect or lifespan shutdown.
+            return
+
+    return StreamingResponse(
+        _body(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",  # disable proxy buffering
+        },
+    )
+```
+
+- [ ] **Step 4: Register the router in `src/dlw/main.py`**
+
+The byte-faithful idiom (verified in SP3 — inside `create_app()`, lazy import). Append immediately after the last existing `include_router` call (the SP3 `executors_read_router`):
+
+```python
+    from dlw.api.tasks_stream import router as tasks_stream_router
+    app.include_router(tasks_stream_router)
+```
+
+- [ ] **Step 5: Run the test to verify it passes**
+
+Run: `uv run pytest tests/api/test_task_stream.py -v`
+Expected: 4 tests PASS in <10 s total.
+
+- [ ] **Step 6: Commit**
+
+```bash
+cd /d/download_weights && git add src/dlw/api/tasks_stream.py src/dlw/main.py tests/api/test_task_stream.py && git commit -q -m "UI-SP5 M1: GET /tasks/{id}/stream SSE endpoint (hand-rolled, 1 Hz tick, terminal-stop)"
+```
+
+---
+
+### Task 3: M1 gate
+
+**Files:** none.
+
+- [ ] **Step 1: Full backend suite**
+
+Run: `uv run pytest tests/ -q`
+Expected: prior 447 + new 4 = 451; 0 failures.
+
+- [ ] **Step 2: OpenAPI + invariant + status-write lint**
+
+```bash
+npx --yes @stoplight/spectral-cli lint api/openapi.yaml --fail-severity=error
+npx --yes @apidevtools/swagger-cli validate api/openapi.yaml
+python tools/lint_invariants.py
+python tools/lint_no_direct_status_write.py
+```
+All exit 0.
+
+- [ ] **Step 3: Commit (only if fixups needed)**
+
+```bash
+cd /d/download_weights && git status -s
+# If any fixups: git add -A && git commit -q -m "UI-SP5 M1 gate: backend suite + openapi + invariant green"
+```
+
+---
+
+# Milestone M2 — Frontend foundation: pure SSE parser + streamer
+
+### Task 4: `parseSseChunk` pure function + spec
+
+**Files:**
+- Create: `frontend/src/api/sse.ts`, `frontend/tests/unit/parseSseChunk.spec.ts`
+
+- [ ] **Step 1: Write the failing test**
+
+Create `frontend/tests/unit/parseSseChunk.spec.ts`:
+
+```ts
+import { describe, expect, test } from 'vitest'
+import { parseSseChunk } from '@/api/sse'
+
+describe('parseSseChunk', () => {
+  test('empty → no events, empty remainder', () => {
+    expect(parseSseChunk('')).toEqual({ events: [], remainder: '' })
+  })
+  test('single event', () => {
+    const { events, remainder } = parseSseChunk('data: hello\n\n')
+    expect(remainder).toBe('')
+    expect(events).toHaveLength(1)
+    expect(events[0]?.event).toBe('message')
+    expect(events[0]?.data).toBe('hello')
+  })
+  test('two events in one buffer', () => {
+    const { events, remainder } = parseSseChunk(
+      'data: a\n\ndata: b\n\n')
+    expect(remainder).toBe('')
+    expect(events.map((e) => e.data)).toEqual(['a', 'b'])
+  })
+  test('event split across two buffers → remainder carries', () => {
+    const a = parseSseChunk('data: hel')
+    expect(a.events).toEqual([])
+    expect(a.remainder).toBe('data: hel')
+    const b = parseSseChunk(a.remainder + 'lo\n\n')
+    expect(b.events).toHaveLength(1)
+    expect(b.events[0]?.data).toBe('hello')
+  })
+  test('comment lines ignored', () => {
+    const { events } = parseSseChunk(
+      ':keepalive\ndata: payload\n\n')
+    expect(events).toHaveLength(1)
+    expect(events[0]?.data).toBe('payload')
+  })
+  test('custom event field', () => {
+    const { events } = parseSseChunk(
+      'event: progress\ndata: 42\n\n')
+    expect(events[0]?.event).toBe('progress')
+    expect(events[0]?.data).toBe('42')
+  })
+  test('multi-line data field joined with newline', () => {
+    const { events } = parseSseChunk(
+      'data: line1\ndata: line2\n\n')
+    expect(events[0]?.data).toBe('line1\nline2')
+  })
+  test('CRLF line endings also accepted', () => {
+    const { events } = parseSseChunk(
+      'data: x\r\n\r\n')
+    expect(events[0]?.data).toBe('x')
+  })
+})
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+`cd /d/download_weights/frontend && pnpm test:unit -- parseSseChunk`
+Expected: FAIL — module not found.
+
+- [ ] **Step 3: Create `frontend/src/api/sse.ts`** (parser only for this task; streamer in Task 5)
+
+```ts
+export interface SseEvent {
+  event: string
+  data: string
+  id?: string
+}
+
+/**
+ * Pure SSE wire-format parser. The caller accumulates raw text chunks from a
+ * ReadableStream and feeds them in; this function returns any complete events
+ * plus the trailing partial-block remainder to prepend to the next chunk.
+ */
+export function parseSseChunk(
+  buf: string,
+): { events: SseEvent[]; remainder: string } {
+  const events: SseEvent[] = []
+  const parts = buf.split(/\r?\n\r?\n/)
+  const remainder = parts.pop() ?? ''
+  for (const block of parts) {
+    let event = 'message'
+    let data = ''
+    let id: string | undefined
+    for (const line of block.split(/\r?\n/)) {
+      if (line === '' || line.startsWith(':')) continue
+      const i = line.indexOf(':')
+      const field = i === -1 ? line : line.slice(0, i)
+      const value = i === -1 ? '' : line.slice(i + 1).replace(/^ /, '')
+      if (field === 'event') event = value
+      else if (field === 'data') data += (data ? '\n' : '') + value
+      else if (field === 'id') id = value
+    }
+    if (data) {
+      events.push(id !== undefined ? { event, data, id } : { event, data })
+    }
+  }
+  return { events, remainder }
+}
+```
+
+- [ ] **Step 4: Run test → PASS; typecheck → 0**
+
+`cd /d/download_weights/frontend && pnpm test:unit -- parseSseChunk` → 8 PASS.
+`pnpm typecheck` → 0 errors.
+
+- [ ] **Step 5: Lint + commit**
+
+```bash
+cd /d/download_weights/frontend && pnpm lint:fix
+cd /d/download_weights && git add frontend/src/api/sse.ts frontend/tests/unit/parseSseChunk.spec.ts && git commit -q -m "UI-SP5 M2: parseSseChunk pure SSE wire-format parser"
+```
+
+---
+
+### Task 5: `streamSse` (fetch + ReadableStream + backoff) + `shouldStream` gate
+
+**Files:**
+- Modify: `frontend/src/api/sse.ts` (append streamer + helper)
+- Create: `frontend/tests/unit/streamGate.spec.ts`
+
+- [ ] **Step 1: Write the failing test for the pure gate**
+
+Create `frontend/tests/unit/streamGate.spec.ts`:
+
+```ts
+import { describe, expect, test } from 'vitest'
+import { ref } from 'vue'
+import { shouldStream } from '@/api/sse'
+
+describe('shouldStream', () => {
+  test('no streamUrl → false', () => {
+    expect(shouldStream({ streamUrl: undefined, applyEvent: () => 1,
+      enabled: true })).toBe(false)
+  })
+  test('no applyEvent → false', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: undefined,
+      enabled: true })).toBe(false)
+  })
+  test('enabled: false → false', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: false })).toBe(false)
+  })
+  test('enabled: Ref<false> → false (unwrapped)', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: ref(false) })).toBe(false)
+  })
+  test('all conditions met → true', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: true })).toBe(true)
+  })
+  test('enabled: Ref<true> → true', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: ref(true) })).toBe(true)
+  })
+  test('enabled: undefined defaults to true', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1 })).toBe(true)
+  })
+})
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+`cd /d/download_weights/frontend && pnpm test:unit -- streamGate`
+Expected: FAIL — `shouldStream` not exported.
+
+- [ ] **Step 3: Append `streamSse` + `shouldStream` to `frontend/src/api/sse.ts`**
+
+```ts
+import { isRef, type Ref } from 'vue'
+
+export interface StreamSseOptions {
+  url: string
+  token: string | null
+  onEvent: (ev: SseEvent) => void
+  onUnauthorized: () => void
+  signal: AbortSignal
+  /** Test seam: override the global fetch (defaults to window.fetch). */
+  fetchImpl?: typeof fetch
+}
+
+/** Pure gate: does this LiveResource configuration want streaming? */
+export interface StreamGateInput {
+  streamUrl?: string | Ref<string> | undefined
+  applyEvent?: ((prev: unknown, ev: SseEvent) => unknown) | undefined
+  enabled?: boolean | Ref<boolean> | undefined
+}
+
+export function shouldStream(o: StreamGateInput): boolean {
+  if (!o.streamUrl || !o.applyEvent) return false
+  if (o.enabled === false) return false
+  if (isRef(o.enabled) && o.enabled.value === false) return false
+  return true
+}
+
+const BACKOFF_MS = [1_000, 2_000, 4_000, 8_000, 16_000, 30_000] as const
+const GIVEUP_AFTER_CONSECUTIVE_FAILURES = 3
+
+/**
+ * Fetch-based SSE client. Sends `Authorization: Bearer <token>` (browser
+ * EventSource cannot set headers). On 401 → calls onUnauthorized and rejects.
+ * On any other disconnect/error: exponential backoff with jitter, capped at
+ * 30 s. Successful chunk receipt resets the backoff counter to 0. Aborts when
+ * the AbortSignal fires.
+ *
+ * Returns a Promise that resolves only when consecutive failures exceed
+ * GIVEUP_AFTER_CONSECUTIVE_FAILURES (signal for the consumer to fall back to
+ * polling). Rejects on 401.
+ */
+export async function streamSse(opts: StreamSseOptions): Promise<void> {
+  const fetchFn = opts.fetchImpl ?? globalThis.fetch
+  let consecutiveFailures = 0
+  let backoffIdx = 0
+  while (!opts.signal.aborted) {
+    let connected = false
+    try {
+      const headers: Record<string, string> = {
+        Accept: 'text/event-stream',
+      }
+      if (opts.token) headers.Authorization = `Bearer ${opts.token}`
+      const resp = await fetchFn(opts.url, {
+        method: 'GET', headers, signal: opts.signal,
+      })
+      if (resp.status === 401) {
+        opts.onUnauthorized()
+        throw new Error('SSE 401')
+      }
+      if (!resp.ok || !resp.body) {
+        throw new Error(`SSE upstream status ${resp.status}`)
+      }
+      const reader = resp.body.getReader()
+      const decoder = new TextDecoder()
+      let buf = ''
+      // eslint-disable-next-line no-constant-condition
+      while (true) {
+        const { value, done } = await reader.read()
+        if (done) break
+        buf += decoder.decode(value, { stream: true })
+        const { events, remainder } = parseSseChunk(buf)
+        buf = remainder
+        for (const ev of events) {
+          opts.onEvent(ev)
+          if (!connected) {
+            connected = true
+            consecutiveFailures = 0
+            backoffIdx = 0
+          }
+        }
+      }
+    } catch (err) {
+      if (opts.signal.aborted) return
+      if ((err as Error).message === 'SSE 401') throw err
+      consecutiveFailures += 1
+      if (consecutiveFailures >= GIVEUP_AFTER_CONSECUTIVE_FAILURES) return
+    }
+    if (opts.signal.aborted) return
+    const base = BACKOFF_MS[Math.min(backoffIdx, BACKOFF_MS.length - 1)]
+      ?? 30_000
+    const jitter = base * (0.8 + Math.random() * 0.4)  // ±20%
+    backoffIdx += 1
+    await new Promise<void>((resolve) => {
+      const t = setTimeout(resolve, jitter)
+      opts.signal.addEventListener('abort',
+        () => { clearTimeout(t); resolve() }, { once: true })
+    })
+  }
+}
+```
+
+- [ ] **Step 4: Run gate test → PASS; typecheck → 0; lint OK**
+
+`pnpm test:unit -- streamGate` → 7 PASS.
+`pnpm typecheck` → 0 errors.
+`pnpm lint:fix && pnpm lint` → OK.
+
+- [ ] **Step 5: Commit**
+
+```bash
+cd /d/download_weights && git add frontend/src/api/sse.ts frontend/tests/unit/streamGate.spec.ts && git commit -q -m "UI-SP5 M2: streamSse fetch+ReadableStream client + shouldStream pure gate"
+```
+
+---
+
+### Task 6: M2 gate
+
+**Files:** none.
+
+- [ ] `cd /d/download_weights/frontend && pnpm test:unit && pnpm typecheck && pnpm lint && pnpm build` — all green.
+
+---
+
+# Milestone M3 — Cutover + smoke + docs
+
+### Task 7: `useLiveResource` accepts `streamUrl` + `applyEvent`
+
+**Files:**
+- Modify: `frontend/src/composables/useLiveResource.ts`
+
+- [ ] **Step 1: Augment `LiveOptions<T>` and wire the streamer**
+
+Read `frontend/src/composables/useLiveResource.ts` first (post-SP2 has `enabled?: Ref<boolean> | boolean`). Replace the file with:
+
+```ts
+import { onScopeDispose, toValue, watch, type MaybeRefOrGetter, type Ref } from 'vue'
+import { useQuery, useQueryClient, type QueryKey } from '@tanstack/vue-query'
+import { shouldStream, streamSse, type SseEvent } from '@/api/sse'
+import { useAuthStore } from '@/stores/auth'
+
+const ERROR_BACKOFF_MS = 5_000
+const HIDDEN_MULTIPLIER = 3
+
+export function computeInterval(o: {
+  base: number; terminal: boolean; hidden: boolean; errored: boolean
+}): number | false {
+  if (o.terminal) return false
+  if (o.errored) return ERROR_BACKOFF_MS
+  return o.hidden ? o.base * HIDDEN_MULTIPLIER : o.base
+}
+
+export interface LiveOptions<T> {
+  baseIntervalMs: number
+  isTerminal?: (data: T) => boolean
+  staleTime?: number
+  enabled?: Ref<boolean> | boolean
+  /** UI-SP5: opt in to SSE. When set with applyEvent, the composable opens a
+   * stream after the first useQuery success and writes events into the cache
+   * via setQueryData. Polling stays disabled while streaming is healthy and
+   * resumes automatically if the stream gives up. */
+  streamUrl?: string | Ref<string>
+  applyEvent?: (prev: T | undefined, ev: SseEvent) => T
+}
+
+/**
+ * Single realtime seam. Today: adaptive polling on vue-query, with an
+ * additive opt-in SSE swap (UI-SP5). UI-SP4 (AI-Copilot) and future
+ * transports plug in here without view changes.
+ *
+ * vue-query v5 does NOT accept a getter for `queryKey` — it must be a
+ * QueryKey (array). Reactivity comes from putting refs *inside* the array.
+ */
+export function useLiveResource<T>(
+  key: QueryKey,
+  fetcher: () => Promise<T>,
+  opts: LiveOptions<T>,
+) {
+  const streaming = shouldStream({
+    streamUrl: opts.streamUrl, applyEvent: opts.applyEvent,
+    enabled: opts.enabled,
+  })
+
+  const q = useQuery<T>({
+    queryKey: key,
+    queryFn: fetcher,
+    enabled: opts.enabled,
+    staleTime: opts.staleTime ?? 0,
+    refetchInterval: (query) => {
+      const data = query.state.data as T | undefined
+      const errored = query.state.status === 'error'
+      const terminal = data !== undefined && !!opts.isTerminal?.(data)
+      const hidden = typeof document !== 'undefined'
+        && document.visibilityState === 'hidden'
+      if (streaming) {
+        // Streaming owns the refresh cadence; only resume polling when the
+        // stream gives up (see watcher below sets `pollingFallback.value`).
+        if (!pollingFallback.value) return false
+      }
+      return computeInterval({
+        base: opts.baseIntervalMs, terminal, hidden, errored,
+      })
+    },
+  })
+
+  const qc = useQueryClient()
+  const pollingFallback: { value: boolean } = { value: false }
+
+  if (streaming && opts.applyEvent) {
+    const ac = new AbortController()
+    const auth = useAuthStore()
+    const apply = opts.applyEvent
+    // Start the stream as soon as we have a first successful snapshot.
+    const stopWatch = watch(
+      () => q.data.value,
+      (snapshot) => {
+        if (snapshot === undefined) return
+        stopWatch()  // only kick off once
+        const url = toValue(opts.streamUrl as MaybeRefOrGetter<string>)
+        void streamSse({
+          url, token: auth.accessToken, signal: ac.signal,
+          onEvent: (ev) => {
+            const prev = qc.getQueryData<T>(key)
+            const next = apply(prev, ev)
+            qc.setQueryData(key, next)
+          },
+          onUnauthorized: () => {
+            auth.logout()
+          },
+        }).then(() => {
+          // streamSse resolved without abort → it gave up (3 consecutive
+          // failures). Fall back to polling.
+          pollingFallback.value = true
+          q.refetch()
+        }).catch(() => {
+          // 401 path — onUnauthorized already invoked.
+        })
+      },
+      { immediate: true },
+    )
+    onScopeDispose(() => { ac.abort() })
+  }
+
+  return q
+}
+```
+
+- [ ] **Step 2: Verify the existing useLiveResource tests still pass**
+
+`cd /d/download_weights/frontend && pnpm test:unit -- useLiveResource`
+Expected: existing 4 tests (computeInterval + useLiveResourceEnabled) PASS unchanged.
+
+- [ ] **Step 3: Typecheck + lint**
+
+`pnpm typecheck` → 0 errors. `pnpm lint:fix && pnpm lint` → OK.
+
+- [ ] **Step 4: Commit**
+
+```bash
+cd /d/download_weights && git add frontend/src/composables/useLiveResource.ts && git commit -q -m "UI-SP5 M3: useLiveResource opt-in SSE (streamUrl+applyEvent; view-transparent)"
+```
+
+---
+
+### Task 8: `useTaskDetail` opts in
+
+**Files:**
+- Modify: `frontend/src/composables/useTaskDetail.ts`
+
+- [ ] **Step 1: Replace file content**
+
+```ts
+import { computed, type Ref } from 'vue'
+import { useLiveResource } from '@/composables/useLiveResource'
+import { client } from '@/api/client'
+import { TERMINAL_STATUSES, type TaskDetail } from '@/api/types'
+
+export function useTaskDetail(taskId: Ref<string>) {
+  const streamUrl = computed(() => `/api/v1/tasks/${taskId.value}/stream`)
+  return useLiveResource<TaskDetail>(
+    ['task', taskId],
+    async () => (await client.get<TaskDetail>(
+      `/api/v1/tasks/${taskId.value}`)).data,
+    {
+      baseIntervalMs: 1_000,
+      isTerminal: (d) => TERMINAL_STATUSES.has(d.status),
+      streamUrl,
+      applyEvent: (_prev, ev) => JSON.parse(ev.data) as TaskDetail,
+    },
+  )
+}
+```
+
+- [ ] **Step 2: Verify the existing useTaskDetail tests still pass (return shape unchanged)**
+
+`cd /d/download_weights/frontend && pnpm test:unit -- useTaskDetail TaskDetailSP2`
+Expected: existing tests (the SP1 `useTaskDetail.spec.ts` + the SP2 `TaskDetailSP2.spec.ts`) PASS unchanged. **This is the view-free proof.**
+
+- [ ] **Step 3: Typecheck + lint**
+
+`pnpm typecheck` → 0 errors. `pnpm lint:fix && pnpm lint` → OK.
+
+- [ ] **Step 4: Commit**
+
+```bash
+cd /d/download_weights && git add frontend/src/composables/useTaskDetail.ts && git commit -q -m "UI-SP5 M3: useTaskDetail opts in to SSE (view-free; all consumers unchanged)"
+```
+
+---
+
+### Task 9: M3 full gate + headed Playwright smoke + docs
+
+**Files:**
+- Modify: `docs/operator/web-ui.md`
+
+- [ ] **Step 1: Full backend suite**
+
+`uv run pytest tests/ -q` → 451 (447 + 4) PASS.
+
+- [ ] **Step 2: Full frontend gate**
+
+`cd /d/download_weights/frontend && pnpm test:unit && pnpm typecheck && pnpm lint && pnpm build` — all green.
+
+- [ ] **Step 3: OpenAPI + invariant + status-write lint**
+
+```bash
+cd /d/download_weights
+npx --yes @stoplight/spectral-cli lint api/openapi.yaml --fail-severity=error
+npx --yes @apidevtools/swagger-cli validate api/openapi.yaml
+python tools/lint_invariants.py
+python tools/lint_no_direct_status_write.py
+```
+All exit 0.
+
+- [ ] **Step 4: Headed Playwright smoke**
+
+Per SP3-codified recipe: ensure the ephemeral `:8011` controller is running with CURRENT code (kill the existing PID via `taskkill //F //PID` if needed, then relaunch via the SP3 recipe). Ensure Vite is up with `DLW_API_PROXY=http://localhost:8011` (find the actual port via `tail -5 /tmp/vite-sp5.log` — Vite may have rolled to :5174/:5175 if :5173 is occupied). Mint a fresh tenant JWT, save to `.run/sp5-token.txt`.
+
+Create `.run/pw/sp5-smoke.mjs`:
+
+```js
+import { chromium } from 'playwright'
+import { readFileSync } from 'node:fs'
+
+const TOKEN = readFileSync('.run/sp5-token.txt', 'utf8').trim()
+const TASK_ID = process.env.SP5_TASK_ID
+const VITE = process.env.SP5_VITE ?? 'http://localhost:5173'
+const errors = []
+const b = await chromium.launch({ headless: false })
+const pg = await b.newPage()
+pg.on('console', (m) => { if (m.type() === 'error') errors.push(m.text()) })
+pg.on('pageerror', (e) => errors.push(String(e)))
+
+await pg.goto(`${VITE}/login`)
+await pg.fill('input', TOKEN)
+await pg.click('button[type="submit"]')
+await pg.waitForURL('**/')
+await pg.goto(`${VITE}/tasks/${TASK_ID}`)
+await pg.waitForSelector('.el-tabs', { timeout: 10_000 })
+await pg.screenshot({ path: '.run/pw/sp5-detail.png' })
+
+// Watch the network for SSE traffic
+const sseReqs = []
+pg.on('request', (r) => {
+  if (r.url().endsWith('/stream')) sseReqs.push(r.url())
+})
+// Reload the page so we observe the SSE request initiate
+await pg.reload()
+await pg.waitForSelector('.el-tabs', { timeout: 10_000 })
+await pg.waitForTimeout(3000)
+await pg.screenshot({ path: '.run/pw/sp5-after-stream.png' })
+
+await b.close()
+// Filter expected /health/active 503 noise (ephemeral controller).
+const real = errors.filter((e) =>
+  !e.includes('/health/active') && !e.includes('Failed to load resource'))
+if (real.length) {
+  console.log('SP5 smoke: real console/page errors:\n' + real.join('\n'))
+  process.exit(1)
+}
+if (sseReqs.length === 0) {
+  console.log('SP5 smoke: no SSE request observed — stream not opened')
+  process.exit(1)
+}
+console.log(`SP5 smoke OK — observed ${sseReqs.length} SSE request(s)`)
+```
+
+Run (PowerShell): pick a task id from the existing dev DB (`$TID = (curl ... /api/v1/tasks | jq -r '.items[0].id')`) and `$env:SP5_TASK_ID=$TID; $env:SP5_VITE='http://localhost:5174'; node .run/pw/sp5-smoke.mjs`.
+
+Expected: prints `SP5 smoke OK — observed ≥1 SSE request(s)`. The screenshots show the Task Detail page rendering normally. **This is the view-free verification: SP1/SP2/SP3 pages render identically; only the network tab shows the new `/stream` connection.**
+
+Record the outcome. If the local stack is unavailable, note it explicitly — the smoke is a manual gate that does not block CI.
+
+- [ ] **Step 5: Append docs**
+
+Add to `docs/operator/web-ui.md`:
+
+```markdown
+
+## UI-SP5 — Realtime SSE swap (delivered)
+
+`useLiveResource` now supports an opt-in SSE transport. The locked promise from
+SP1/SP2/SP3 — *"SP5 swaps internals to SSE/WS with zero view changes"* — is
+honored: every view, page, and other composable is unchanged.
+
+- **Backend**: `GET /api/v1/tasks/{id}/stream` — hand-rolled
+  `text/event-stream`, 1 Hz default tick (env
+  `DLW_TASK_STREAM_INTERVAL_SECONDS` overrides; clamped `[0.1, 10.0]`).
+  Tenant-scoped via the proven cancel-pattern (404 cross-tenant). Terminates on
+  terminal task status, client disconnect, or controller shutdown.
+- **Frontend**: `frontend/src/api/sse.ts` (pure `parseSseChunk` + `streamSse`
+  fetch+ReadableStream with exponential backoff, Bearer-via-header). The
+  `useLiveResource` composable gains optional `streamUrl` + `applyEvent`
+  fields; when both are set, the composable opens an SSE connection after the
+  first snapshot and writes events into the vue-query cache. On 3 consecutive
+  failures the stream gives up and polling resumes automatically.
+- **One consumer opts in**: `useTaskDetail`. Every other composable
+  (`useTaskList`, `useQuota`, `useSubtaskChunks`, `useSourceAllocation`,
+  `useParticipatingExecutors`, `useTaskEvents`, `useExecutors`, `useAuditLog`,
+  `useSystemHealth`) stays on polling.
+
+**Known deferrals** (intentional): WebSocket transport; SSE for the 4 SP2
+sub-resource composables (would require a multi-resource stream that breaks
+view-free — defer to SP5b if telemetry justifies); SSE for the
+low-frequency SP3 composables (push value is negligible at 5–30 s cadences).
+UI-SP4 (AI-Copilot) remains the v2.1 follow-up.
+```
+
+- [ ] **Step 6: Commit**
+
+```bash
+cd /d/download_weights && git add docs/operator/web-ui.md && git commit -q -m "UI-SP5 M3: operator docs for the realtime SSE swap"
+```
+
+---
+
+## Self-Review
+
+**1. Spec coverage:**
+- §2 in-scope: 1 SSE endpoint (Tasks 1-2), `parseSseChunk` + `streamSse` + `shouldStream` (Tasks 4-5), `useLiveResource` opt-in (Task 7), `useTaskDetail` cutover (Task 8), headed smoke + docs (Task 9). ✓
+- §3 inherited locked decisions: useLiveResource return shape preserved (Task 7 keeps `useQuery<T>` as the return); additive backend (Task 2 new file, Task 1 openapi additive); zero new runtime dep (only stdlib `asyncio` / `httpx` already a dep; frontend uses native `fetch`+`ReadableStream`+`TextDecoder`); CI gates only (Task 3/9 gates); no migration; tenant gate verbatim from cancel-pattern. ✓
+- §4 backend: route in new file (`tasks_stream.py`), 4 tests covering unauth/cross-tenant/happy/terminal-stop, env-configurable tick rate. ✓
+- §5 frontend: pure parser + streamer with backoff, additive `LiveOptions` fields, ONE consumer opted in. ✓
+- §7 risks/contingency: cancellation propagates via small-slice sleep + `finally`; 401 path; backoff gives up at 3 failures, polling resumes; no fake-timer test needed (streamer is smoke-tested). ✓
+
+**2. Placeholder scan:** none. Every step has complete code or exact commands.
+
+**3. Type consistency:**
+- `SseEvent` shape `{ event, data, id? }` is the SAME in `sse.ts`, `parseSseChunk` (returned), `streamSse` (`onEvent` param), `useLiveResource` (`applyEvent` second param), and `useTaskDetail` (`(_prev, ev) => JSON.parse(ev.data)`). ✓
+- `streamUrl?: string | Ref<string>` consistent in `LiveOptions` (Task 7) and `useTaskDetail` (Task 8 — `computed(() => '/api/v1/...')` typed as `ComputedRef<string>` which is assignable to `Ref<string>`). ✓
+- `applyEvent?: (prev: T | undefined, ev: SseEvent) => T` — consistent return-shape; `useTaskDetail` returns `TaskDetail`; `useLiveResource` calls `setQueryData(key, next)` where `next: T`. ✓
+- `shouldStream` gate (Task 5) is invoked once in `useLiveResource` Task 7 to compute `streaming: boolean`; the gate's three inputs match the LiveOptions fields. ✓
+
+**4. Verification chain locked:**
+- Backend: `client.stream("GET", ".../stream")` + `aiter_lines` (httpx). The terminal-stop test directly mutates `DownloadTask.status` via test session — this is `tests/`, NOT in the `check_task_status_domain` scan list, so `lint_invariants` stays green.
+- Frontend pure tests (`parseSseChunk` 8 cases, `streamGate` 7 cases) run under happy-dom without DOM. Integration (`streamSse` invoker) covered by the headed Playwright smoke.
+- "View-free" proof: existing `useTaskDetail.spec.ts` + `TaskDetailSP2.spec.ts` still PASS unchanged (Task 8 step 2).

From f2ef09cff33f8cc1950277c118c90b194dda2866 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:15:27 +0800
Subject: [PATCH 03/10] UI-SP5 plan: fix 1 BLOCKER + 6 IMPORTANT from
 pre-execution review
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

BLOCKER (frontend): Vue 3.5 watch({immediate:true}) fires SYNCHRONOUSLY
before stopWatch is assigned → calling stopWatch() in the handler hits TDZ
when vue-query serves a cached snapshot synchronously. Fix: let stopWatch +
optional-call + started flag.

IMPORTANTs:
- Backend: yield b":open\n\n" first byte to defeat httpx 0.27.x ASGITransport
  buffering (could hang the multi-snapshot test).
- Backend: replace hard-coded 447 baseline with record-then-add (SP3 added
  extra tests since the baseline was estimated).
- Frontend: pollingFallback → ref(false) for Vue idiom clarity.
- Frontend: streamSse fast-fail on 403/404 (no point burning 7s of backoff).
- Frontend: parseSseChunk hasData flag preserves "0"/"" payloads (SSE-spec).
- Frontend: NEW streamSse happy-path / 401 / 404 unit tests (3 tests).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 .../plans/2026-05-20-ui-sp5-realtime-sse.md   | 124 ++++++++++++++++--
 1 file changed, 111 insertions(+), 13 deletions(-)

diff --git a/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md b/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md
index f7735bb..8b4a167 100644
--- a/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md
+++ b/docs/superpowers/plans/2026-05-20-ui-sp5-realtime-sse.md
@@ -354,6 +354,12 @@ async def stream_task_detail(
     interval = _clamped_interval()
 
     async def _body() -> AsyncIterator[bytes]:
+        # Pre-review IMPORTANT fix: flush an immediate comment line so the
+        # response headers + first byte ship together. Defeats httpx 0.27.x
+        # ASGITransport buffering (which would otherwise hold the response
+        # until generator close) AND any production reverse-proxy buffering.
+        # SSE parser ignores comment lines; tests' "data:" filter also skips.
+        yield b":open\n\n"
         ticks_since_data = 0
         try:
             while True:
@@ -424,7 +430,7 @@ cd /d/download_weights && git add src/dlw/api/tasks_stream.py src/dlw/main.py te
 - [ ] **Step 1: Full backend suite**
 
 Run: `uv run pytest tests/ -q`
-Expected: prior 447 + new 4 = 451; 0 failures.
+Expected: prior baseline (record `uv run pytest tests/ --collect-only -q | tail -1` BEFORE Task 1 if uncertain) + 4 new tests; 0 failures.
 
 - [ ] **Step 2: OpenAPI + invariant + status-write lint**
 
@@ -539,16 +545,23 @@ export function parseSseChunk(
     let event = 'message'
     let data = ''
     let id: string | undefined
+    // Pre-review IMPORTANT fix: track presence of any `data:` field, not just
+    // truthy strings, so legitimate payloads like "0" / "false" / "" aren't
+    // silently dropped (cf. SSE spec — empty data is a valid event).
+    let hasData = false
     for (const line of block.split(/\r?\n/)) {
       if (line === '' || line.startsWith(':')) continue
       const i = line.indexOf(':')
       const field = i === -1 ? line : line.slice(0, i)
       const value = i === -1 ? '' : line.slice(i + 1).replace(/^ /, '')
       if (field === 'event') event = value
-      else if (field === 'data') data += (data ? '\n' : '') + value
+      else if (field === 'data') {
+        hasData = true
+        data += (data ? '\n' : '') + value
+      }
       else if (field === 'id') id = value
     }
-    if (data) {
+    if (hasData) {
       events.push(id !== undefined ? { event, data, id } : { event, data })
     }
   }
@@ -682,6 +695,12 @@ export async function streamSse(opts: StreamSseOptions): Promise<void> {
         opts.onUnauthorized()
         throw new Error('SSE 401')
       }
+      // Pre-review IMPORTANT fix: permanent client errors (task gone /
+      // forbidden) should fail-fast — burning 7+ s of backoff on a 404 is
+      // pure latency before the consumer's poll-fallback kicks in.
+      if (resp.status === 403 || resp.status === 404) {
+        return
+      }
       if (!resp.ok || !resp.body) {
         throw new Error(`SSE upstream status ${resp.status}`)
       }
@@ -724,16 +743,82 @@ export async function streamSse(opts: StreamSseOptions): Promise<void> {
 }
 ```
 
-- [ ] **Step 4: Run gate test → PASS; typecheck → 0; lint OK**
+- [ ] **Step 4: Add a streamSse happy-path unit test (pre-review IMP 4)**
+
+The streamer is ~70 LOC of cancellation-sensitive logic; the pure parser + gate don't exercise it. Add ONE happy-path test using a stub `fetchImpl` that returns a `ReadableStream` of two SSE events; abort after the second:
+
+Create `frontend/tests/unit/streamSse.spec.ts`:
+
+```ts
+import { describe, expect, test } from 'vitest'
+import { streamSse, type SseEvent } from '@/api/sse'
+
+describe('streamSse', () => {
+  test('parses events from a ReadableStream and stops on abort', async () => {
+    const ac = new AbortController()
+    const got: SseEvent[] = []
+    const stream = new ReadableStream<Uint8Array>({
+      start(c) {
+        c.enqueue(new TextEncoder().encode('data: a\n\n'))
+        c.enqueue(new TextEncoder().encode('data: b\n\n'))
+        c.close()
+      },
+    })
+    const fetchImpl = (async () => new Response(stream, { status: 200 })) as
+      typeof fetch
+    await streamSse({
+      url: '/x', token: 't', signal: ac.signal,
+      onEvent: (e) => {
+        got.push(e)
+        if (got.length === 2) ac.abort()
+      },
+      onUnauthorized: () => {},
+      fetchImpl,
+    })
+    expect(got.map((e) => e.data)).toEqual(['a', 'b'])
+  })
+
+  test('401 → calls onUnauthorized and rejects without retry', async () => {
+    const ac = new AbortController()
+    let unauth = 0
+    const fetchImpl = (async () => new Response('', { status: 401 })) as
+      typeof fetch
+    await expect(streamSse({
+      url: '/x', token: 't', signal: ac.signal,
+      onEvent: () => {},
+      onUnauthorized: () => { unauth++ },
+      fetchImpl,
+    })).rejects.toThrow(/SSE 401/)
+    expect(unauth).toBe(1)
+  })
+
+  test('404 → resolves without retry (permanent client error)', async () => {
+    const ac = new AbortController()
+    const fetchImpl = (async () => new Response('not found', { status: 404 }))
+      as typeof fetch
+    await streamSse({
+      url: '/x', token: 't', signal: ac.signal,
+      onEvent: () => {},
+      onUnauthorized: () => {},
+      fetchImpl,
+    })
+  })
+})
+```
+
+Run: `cd /d/download_weights/frontend && pnpm test:unit -- streamSse`
+Expected: 3 PASS.
+
+- [ ] **Step 5: Run gate test → PASS; typecheck → 0; lint OK**
 
 `pnpm test:unit -- streamGate` → 7 PASS.
 `pnpm typecheck` → 0 errors.
 `pnpm lint:fix && pnpm lint` → OK.
 
-- [ ] **Step 5: Commit**
+- [ ] **Step 6: Commit**
 
 ```bash
-cd /d/download_weights && git add frontend/src/api/sse.ts frontend/tests/unit/streamGate.spec.ts && git commit -q -m "UI-SP5 M2: streamSse fetch+ReadableStream client + shouldStream pure gate"
+cd /d/download_weights && git add frontend/src/api/sse.ts frontend/tests/unit/streamGate.spec.ts frontend/tests/unit/streamSse.spec.ts && git commit -q -m "UI-SP5 M2: streamSse fetch+ReadableStream client + shouldStream pure gate + happy-path tests"
 ```
 
 ---
@@ -758,7 +843,7 @@ cd /d/download_weights && git add frontend/src/api/sse.ts frontend/tests/unit/st
 Read `frontend/src/composables/useLiveResource.ts` first (post-SP2 has `enabled?: Ref<boolean> | boolean`). Replace the file with:
 
 ```ts
-import { onScopeDispose, toValue, watch, type MaybeRefOrGetter, type Ref } from 'vue'
+import { onScopeDispose, ref, toValue, watch, type MaybeRefOrGetter, type Ref, type WatchStopHandle } from 'vue'
 import { useQuery, useQueryClient, type QueryKey } from '@tanstack/vue-query'
 import { shouldStream, streamSse, type SseEvent } from '@/api/sse'
 import { useAuthStore } from '@/stores/auth'
@@ -828,18 +913,31 @@ export function useLiveResource<T>(
   })
 
   const qc = useQueryClient()
-  const pollingFallback: { value: boolean } = { value: false }
+  // Pre-review fix (IMPORTANT 1): use a ref for clarity; the callback re-reads
+  // the closure variable per refetchInterval-eval, but matching Vue idioms
+  // makes intent obvious and future-proofs anyone wanting to watch() this.
+  const pollingFallback = ref(false)
 
   if (streaming && opts.applyEvent) {
     const ac = new AbortController()
     const auth = useAuthStore()
     const apply = opts.applyEvent
-    // Start the stream as soon as we have a first successful snapshot.
-    const stopWatch = watch(
+    // Pre-review BLOCKER fix: Vue 3.5's `watch({ immediate: true })` invokes
+    // the handler SYNCHRONOUSLY inside the watch() call — before `stopWatch`
+    // is assigned the returned stop handle. If vue-query serves a cached
+    // snapshot synchronously (e.g. HMR / route revisit / future initialData),
+    // calling `stopWatch()` from inside the handler would hit TDZ
+    // (ReferenceError). Fix: `let stopWatch` + optional-call + a `started`
+    // flag so the kick-off runs exactly once even if the watcher fires
+    // synchronously on registration.
+    let stopWatch: WatchStopHandle | undefined
+    let started = false
+    stopWatch = watch(
       () => q.data.value,
       (snapshot) => {
-        if (snapshot === undefined) return
-        stopWatch()  // only kick off once
+        if (snapshot === undefined || started) return
+        started = true
+        stopWatch?.()  // safe — undefined on synchronous immediate fire
         const url = toValue(opts.streamUrl as MaybeRefOrGetter<string>)
         void streamSse({
           url, token: auth.accessToken, signal: ac.signal,
@@ -939,7 +1037,7 @@ cd /d/download_weights && git add frontend/src/composables/useTaskDetail.ts && g
 
 - [ ] **Step 1: Full backend suite**
 
-`uv run pytest tests/ -q` → 451 (447 + 4) PASS.
+`uv run pytest tests/ -q` → (M1 baseline + 4) PASS; 0 failures.
 
 - [ ] **Step 2: Full frontend gate**
 

From d5d4598daa8c53f39c77af81c49f583d642c6b0a Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:16:43 +0800
Subject: [PATCH 04/10] UI-SP5 M1: openapi /tasks/{id}/stream path +
 task_stream_interval_seconds setting

---
 api/openapi.yaml  | 19 +++++++++++++++++++
 src/dlw/config.py |  3 +++
 2 files changed, 22 insertions(+)

diff --git a/api/openapi.yaml b/api/openapi.yaml
index 01ca044..b3bc614 100644
--- a/api/openapi.yaml
+++ b/api/openapi.yaml
@@ -529,6 +529,25 @@ paths:
                     items: {$ref: '#/components/schemas/TaskEvent'}
                   next_cursor: {type: string, nullable: true}
 
+  /tasks/{taskId}/stream:
+    parameters:
+      - $ref: '#/components/parameters/TaskId'
+    get:
+      tags: [tasks]
+      summary: Live TaskDetail SSE stream (UI-SP5)
+      operationId: streamTaskDetail
+      responses:
+        '200':
+          description: SSE stream of TaskDetail snapshots (text/event-stream)
+          content:
+            text/event-stream:
+              schema:
+                type: string
+                description: |
+                  Each event is `data: <TaskDetail JSON>\n\n`. Stream terminates
+                  on terminal task status, client disconnect, or controller
+                  shutdown. Keep-alive comment lines (`:keepalive`) may appear.
+
   /tasks/{taskId}/subtask-chunks:
     parameters:
       - $ref: '#/components/parameters/TaskId'
diff --git a/src/dlw/config.py b/src/dlw/config.py
index 5ceef42..3e2f6fb 100644
--- a/src/dlw/config.py
+++ b/src/dlw/config.py
@@ -36,6 +36,9 @@ class Settings(BaseSettings):
     # Phase 2 W3b — HF reverse-proxy
     hf_proxy_timeout_seconds: int = Field(default=300, ge=10, le=3600)
 
+    # UI-SP5 — SSE tick rate for /tasks/{id}/stream (clamped at runtime).
+    task_stream_interval_seconds: float = Field(default=1.0)
+
     # Phase 2 W3c — controller leader election
     active_lock_id: int = Field(
         default=0x444C5743_414B5631,

From 5b9bf318a4fc5a03289d5cd3e6561b8d2fa1b652 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:20:02 +0800
Subject: [PATCH 05/10] UI-SP5 M1: GET /tasks/{id}/stream SSE endpoint
 (hand-rolled, max_ticks testability hatch for httpx ASGITransport)

---
 src/dlw/api/tasks_stream.py   | 122 +++++++++++++++++++++++++
 src/dlw/main.py               |   2 +
 tests/api/test_task_stream.py | 163 ++++++++++++++++++++++++++++++++++
 3 files changed, 287 insertions(+)
 create mode 100644 src/dlw/api/tasks_stream.py
 create mode 100644 tests/api/test_task_stream.py

diff --git a/src/dlw/api/tasks_stream.py b/src/dlw/api/tasks_stream.py
new file mode 100644
index 0000000..2e0045e
--- /dev/null
+++ b/src/dlw/api/tasks_stream.py
@@ -0,0 +1,122 @@
+"""GET /api/v1/tasks/{id}/stream — SSE TaskDetail stream (UI-SP5).
+
+Hand-rolled text/event-stream via StreamingResponse (same idiom as hf_proxy.py).
+1 Hz default tick rate; configurable via DLW_TASK_STREAM_INTERVAL_SECONDS
+(clamped [0.1, 10.0] in code). Stream terminates on terminal task status,
+client disconnect, or controller shutdown.
+"""
+from __future__ import annotations
+
+import asyncio
+import uuid
+from collections.abc import AsyncIterator
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi.responses import StreamingResponse
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import async_sessionmaker
+from sqlalchemy.orm import selectinload
+
+from dlw.auth.principal import Principal
+from dlw.authz.deps import require_perm
+from dlw.config import get_settings
+from dlw.db.models.task import DownloadTask
+from dlw.db.session import get_engine
+from dlw.db.tenant_scope import tenant_filtered
+from dlw.schemas.task import TaskDetail
+
+router = APIRouter(prefix="/api/v1/tasks", tags=["tasks"])
+
+_TERMINAL = {"succeeded", "failed", "cancelled"}
+_KEEPALIVE_EVERY_TICKS = 15
+
+
+def _clamped_interval() -> float:
+    raw = float(getattr(
+        get_settings(), "task_stream_interval_seconds", 1.0))
+    return max(0.1, min(10.0, raw))
+
+
+async def _load_detail(
+    session_maker, task_id: uuid.UUID, tenant_id: int,
+) -> TaskDetail | None:
+    async with session_maker() as s:
+        row = (await s.execute(
+            select(DownloadTask).where(
+                DownloadTask.id == task_id,
+                DownloadTask.tenant_id == tenant_id)
+            .options(selectinload(DownloadTask.subtasks))
+        )).scalar_one_or_none()
+        return TaskDetail.model_validate(row) if row is not None else None
+
+
+@router.get("/{task_id}/stream")
+async def stream_task_detail(
+    task_id: uuid.UUID,
+    request: Request,
+    max_ticks: int | None = Query(default=None, ge=1, le=10000),
+    principal: Principal = Depends(require_perm("/api/v1/tasks*", "GET")),
+) -> StreamingResponse:
+    # Tenant gate — proven cancel-pattern. 404 cross-tenant; never leak.
+    session_maker = async_sessionmaker(get_engine(), expire_on_commit=False)
+    async with session_maker() as s:
+        owned = await s.scalar(
+            tenant_filtered(select(DownloadTask.id)
+                            .where(DownloadTask.id == task_id),
+                            DownloadTask, principal))
+    if owned is None:
+        raise HTTPException(status_code=404, detail="task not found")
+
+    interval = _clamped_interval()
+
+    async def _body() -> AsyncIterator[bytes]:
+        # Pre-review IMPORTANT fix: flush an immediate comment line so the
+        # response headers + first byte ship together. Defeats httpx 0.27.x
+        # ASGITransport buffering (which would otherwise hold the response
+        # until generator close) AND any production reverse-proxy buffering.
+        # SSE parser ignores comment lines; tests' "data:" filter also skips.
+        yield b":open\n\n"
+        ticks_since_data = 0
+        tick_count = 0
+        try:
+            while True:
+                if await request.is_disconnected():
+                    return
+                detail = await _load_detail(
+                    session_maker, task_id, principal.tenant_id)
+                if detail is None:
+                    return
+                yield (f"data: {detail.model_dump_json()}"
+                       "\n\n").encode("utf-8")
+                ticks_since_data = 0
+                tick_count += 1
+                if detail.status in _TERMINAL:
+                    return
+                # Testability hatch — `?max_ticks=N` lets tests verify N
+                # snapshots arrive and the body terminates (httpx
+                # ASGITransport buffers until generator close). Ignored in
+                # production calls (no public client sets it).
+                if max_ticks is not None and tick_count >= max_ticks:
+                    return
+                slept = 0.0
+                while slept < interval:
+                    if await request.is_disconnected():
+                        return
+                    chunk = min(0.05, interval - slept)
+                    await asyncio.sleep(chunk)
+                    slept += chunk
+                ticks_since_data += 1
+                if ticks_since_data >= _KEEPALIVE_EVERY_TICKS:
+                    yield b":keepalive\n\n"
+                    ticks_since_data = 0
+        except asyncio.CancelledError:
+            return
+
+    return StreamingResponse(
+        _body(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
diff --git a/src/dlw/main.py b/src/dlw/main.py
index fb5d36e..5a192be 100644
--- a/src/dlw/main.py
+++ b/src/dlw/main.py
@@ -309,6 +309,8 @@ def create_app() -> FastAPI:
     app.include_router(audit_router)
     from dlw.api.executors_read import router as executors_read_router
     app.include_router(executors_read_router)
+    from dlw.api.tasks_stream import router as tasks_stream_router
+    app.include_router(tasks_stream_router)
 
     # DX only: advertise the Bearer-JWT scheme in the generated OpenAPI so
     # Swagger /docs shows an "Authorize" button and authenticated
diff --git a/tests/api/test_task_stream.py b/tests/api/test_task_stream.py
new file mode 100644
index 0000000..a5b2e7e
--- /dev/null
+++ b/tests/api/test_task_stream.py
@@ -0,0 +1,163 @@
+"""Tests for GET /api/v1/tasks/{id}/stream (UI-SP5 SSE)."""
+from __future__ import annotations
+
+import asyncio
+import json
+import uuid
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from dlw.config import get_settings
+from dlw.db.base import Base
+from tests.conftest import make_app_with_state, principal_headers
+
+SECRET = "unit-secret"
+TICK = "0.1"
+
+
+@pytest.fixture(scope="module", autouse=True)
+async def _bootstrap(engine):
+    from dlw.db.models.storage import StorageBackend
+    from dlw.db.models.tenant import Project, Tenant, User
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+        await conn.run_sync(Base.metadata.create_all)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with factory() as session:
+        session.add(Tenant(id=1, slug="default", display_name="Default"))
+        session.add(Tenant(id=2, slug="other", display_name="Other"))
+        await session.flush()
+        session.add(Project(id=1, tenant_id=1, name="default"))
+        session.add(User(id=1, tenant_id=1, oidc_subject="dev",
+                         email="d@l", role="tenant_admin"))
+        session.add(StorageBackend(id=1, tenant_id=1, name="default",
+                                   backend_type="s3", config_encrypted=b""))
+        await session.commit()
+    yield
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.drop_all)
+
+
+@pytest.fixture(autouse=True)
+def _set_env(monkeypatch: pytest.MonkeyPatch):
+    get_settings.cache_clear()
+    monkeypatch.setenv("DLW_SYSTEM_JWT_SECRET", SECRET)
+    monkeypatch.setenv("DLW_TASK_STREAM_INTERVAL_SECONDS", TICK)
+    get_settings.cache_clear()
+    yield
+    get_settings.cache_clear()
+
+
+@pytest.fixture(autouse=True)
+def _patch_hf(monkeypatch: pytest.MonkeyPatch):
+    from dlw.services.hf_metadata import RepoFile
+
+    async def fake(*args, **kwargs):
+        return [
+            RepoFile(path="config.json", size=4096, sha256=None),
+            RepoFile(path="model.safetensors", size=64 * 1024, sha256="a" * 64),
+        ]
+    monkeypatch.setattr("dlw.services.task_service.list_repo_tree", fake)
+
+
+@pytest.fixture
+def auth() -> dict[str, str]:
+    return principal_headers(secret=SECRET, role="tenant_admin")
+
+
+@pytest.fixture
+async def client(ephemeral_ca):
+    app = make_app_with_state(ephemeral_ca, enrollment_token="e")
+    async with AsyncClient(transport=ASGITransport(app=app),
+                           base_url="http://test", timeout=10.0) as c:
+        yield c
+
+
+async def _make_task(client, auth) -> str:
+    r = await client.post("/api/v1/tasks", json={
+        "repo_id": "o/stream", "revision": "3" * 40, "storage_id": 1,
+    }, headers=auth)
+    assert r.status_code == 201, r.text
+    return r.json()["id"]
+
+
+async def _collect_events(client, url, headers, *, count, timeout=4.0):
+    received: list[str] = []
+    async with asyncio.timeout(timeout):
+        async with client.stream("GET", url, headers=headers) as resp:
+            assert resp.status_code == 200, await resp.aread()
+            ctype = resp.headers.get("content-type", "")
+            assert "text/event-stream" in ctype, ctype
+            async for line in resp.aiter_lines():
+                if line.startswith("data: "):
+                    received.append(line[len("data: "):])
+                    if len(received) >= count:
+                        return received
+    return received
+
+
+@pytest.mark.slow
+async def test_stream_unauthenticated_401(client: AsyncClient) -> None:
+    async with client.stream(
+        "GET", f"/api/v1/tasks/{uuid.uuid4()}/stream",
+    ) as resp:
+        assert resp.status_code == 401
+
+
+@pytest.mark.slow
+async def test_stream_cross_tenant_404(client: AsyncClient, auth) -> None:
+    tid = await _make_task(client, auth)
+    other = principal_headers(secret=SECRET, role="tenant_admin",
+                              user_id=9, tenant_id=2)
+    async with client.stream(
+        "GET", f"/api/v1/tasks/{tid}/stream", headers=other,
+    ) as resp:
+        assert resp.status_code == 404
+
+
+@pytest.mark.slow
+async def test_stream_emits_multiple_snapshots(
+    client: AsyncClient, auth,
+) -> None:
+    tid = await _make_task(client, auth)
+    # `?max_ticks=2` makes the body self-terminate after 2 snapshots —
+    # required because httpx ASGITransport buffers chunks until the response
+    # generator closes (well-known limitation). Production clients never
+    # set this; the headed Playwright smoke validates the real streaming
+    # behavior over a live uvicorn.
+    received = await _collect_events(
+        client, f"/api/v1/tasks/{tid}/stream?max_ticks=2", auth,
+        count=2, timeout=4.0)
+    assert len(received) >= 2
+    for raw in received[:2]:
+        payload = json.loads(raw)
+        assert payload["id"] == tid
+        assert "status" in payload
+        assert "subtasks" in payload
+
+
+@pytest.mark.slow
+async def test_stream_terminates_on_terminal_status(
+    client: AsyncClient, auth, engine,
+) -> None:
+    from dlw.db.models.task import DownloadTask
+    tid = await _make_task(client, auth)
+    factory = async_sessionmaker(engine, expire_on_commit=False)
+    async with factory() as s:
+        task = await s.get(DownloadTask, uuid.UUID(tid))
+        assert task is not None
+        task.status = "succeeded"
+        await s.commit()
+    received: list[str] = []
+    async with asyncio.timeout(3.0):
+        async with client.stream(
+            "GET", f"/api/v1/tasks/{tid}/stream", headers=auth,
+        ) as resp:
+            assert resp.status_code == 200
+            async for line in resp.aiter_lines():
+                if line.startswith("data: "):
+                    received.append(line)
+    assert len(received) == 1
+    assert "succeeded" in received[0]

From 5e7c47a3105d71e0c251ffb136a539b95272c021 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:21:47 +0800
Subject: [PATCH 06/10] UI-SP5 M2: parseSseChunk pure SSE wire-format parser

---
 frontend/src/api/sse.ts                   | 43 +++++++++++++++++++
 frontend/tests/unit/parseSseChunk.spec.ts | 51 +++++++++++++++++++++++
 2 files changed, 94 insertions(+)
 create mode 100644 frontend/src/api/sse.ts
 create mode 100644 frontend/tests/unit/parseSseChunk.spec.ts

diff --git a/frontend/src/api/sse.ts b/frontend/src/api/sse.ts
new file mode 100644
index 0000000..151d243
--- /dev/null
+++ b/frontend/src/api/sse.ts
@@ -0,0 +1,43 @@
+export interface SseEvent {
+  event: string
+  data: string
+  id?: string
+}
+
+/**
+ * Pure SSE wire-format parser. The caller accumulates raw text chunks from a
+ * ReadableStream and feeds them in; this function returns any complete events
+ * plus the trailing partial-block remainder to prepend to the next chunk.
+ */
+export function parseSseChunk(
+  buf: string,
+): { events: SseEvent[]; remainder: string } {
+  const events: SseEvent[] = []
+  const parts = buf.split(/\r?\n\r?\n/)
+  const remainder = parts.pop() ?? ''
+  for (const block of parts) {
+    let event = 'message'
+    let data = ''
+    let id: string | undefined
+    // Pre-review IMPORTANT fix: track presence of any `data:` field, not just
+    // truthy strings, so legitimate payloads like "0" / "false" / "" aren't
+    // silently dropped (cf. SSE spec — empty data is a valid event).
+    let hasData = false
+    for (const line of block.split(/\r?\n/)) {
+      if (line === '' || line.startsWith(':')) continue
+      const i = line.indexOf(':')
+      const field = i === -1 ? line : line.slice(0, i)
+      const value = i === -1 ? '' : line.slice(i + 1).replace(/^ /, '')
+      if (field === 'event') event = value
+      else if (field === 'data') {
+        hasData = true
+        data += (data ? '\n' : '') + value
+      }
+      else if (field === 'id') id = value
+    }
+    if (hasData) {
+      events.push(id !== undefined ? { event, data, id } : { event, data })
+    }
+  }
+  return { events, remainder }
+}
diff --git a/frontend/tests/unit/parseSseChunk.spec.ts b/frontend/tests/unit/parseSseChunk.spec.ts
new file mode 100644
index 0000000..d561109
--- /dev/null
+++ b/frontend/tests/unit/parseSseChunk.spec.ts
@@ -0,0 +1,51 @@
+import { describe, expect, test } from 'vitest'
+import { parseSseChunk } from '@/api/sse'
+
+describe('parseSseChunk', () => {
+  test('empty → no events, empty remainder', () => {
+    expect(parseSseChunk('')).toEqual({ events: [], remainder: '' })
+  })
+  test('single event', () => {
+    const { events, remainder } = parseSseChunk('data: hello\n\n')
+    expect(remainder).toBe('')
+    expect(events).toHaveLength(1)
+    expect(events[0]?.event).toBe('message')
+    expect(events[0]?.data).toBe('hello')
+  })
+  test('two events in one buffer', () => {
+    const { events, remainder } = parseSseChunk(
+      'data: a\n\ndata: b\n\n')
+    expect(remainder).toBe('')
+    expect(events.map((e) => e.data)).toEqual(['a', 'b'])
+  })
+  test('event split across two buffers → remainder carries', () => {
+    const a = parseSseChunk('data: hel')
+    expect(a.events).toEqual([])
+    expect(a.remainder).toBe('data: hel')
+    const b = parseSseChunk(a.remainder + 'lo\n\n')
+    expect(b.events).toHaveLength(1)
+    expect(b.events[0]?.data).toBe('hello')
+  })
+  test('comment lines ignored', () => {
+    const { events } = parseSseChunk(
+      ':keepalive\ndata: payload\n\n')
+    expect(events).toHaveLength(1)
+    expect(events[0]?.data).toBe('payload')
+  })
+  test('custom event field', () => {
+    const { events } = parseSseChunk(
+      'event: progress\ndata: 42\n\n')
+    expect(events[0]?.event).toBe('progress')
+    expect(events[0]?.data).toBe('42')
+  })
+  test('multi-line data field joined with newline', () => {
+    const { events } = parseSseChunk(
+      'data: line1\ndata: line2\n\n')
+    expect(events[0]?.data).toBe('line1\nline2')
+  })
+  test('CRLF line endings also accepted', () => {
+    const { events } = parseSseChunk(
+      'data: x\r\n\r\n')
+    expect(events[0]?.data).toBe('x')
+  })
+})

From dcd842cb16b0d3a892ead6ca51d1ba4c459e0472 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:25:10 +0800
Subject: [PATCH 07/10] UI-SP5 M2: streamSse fetch+ReadableStream client +
 shouldStream pure gate + happy-path tests

---
 frontend/src/api/sse.ts                | 106 +++++++++++++++++++++++++
 frontend/tests/unit/streamGate.spec.ts |  33 ++++++++
 frontend/tests/unit/streamSse.spec.ts  |  54 +++++++++++++
 3 files changed, 193 insertions(+)
 create mode 100644 frontend/tests/unit/streamGate.spec.ts
 create mode 100644 frontend/tests/unit/streamSse.spec.ts

diff --git a/frontend/src/api/sse.ts b/frontend/src/api/sse.ts
index 151d243..cedbeb8 100644
--- a/frontend/src/api/sse.ts
+++ b/frontend/src/api/sse.ts
@@ -1,9 +1,115 @@
+import { isRef, type Ref } from 'vue'
+
 export interface SseEvent {
   event: string
   data: string
   id?: string
 }
 
+export interface StreamSseOptions {
+  url: string
+  token: string | null
+  onEvent: (ev: SseEvent) => void
+  onUnauthorized: () => void
+  signal: AbortSignal
+  /** Test seam: override the global fetch (defaults to globalThis.fetch). */
+  fetchImpl?: typeof fetch
+}
+
+export interface StreamGateInput {
+  streamUrl?: string | Ref<string> | undefined
+  applyEvent?: ((prev: unknown, ev: SseEvent) => unknown) | undefined
+  enabled?: boolean | Ref<boolean> | undefined
+}
+
+/** Pure gate: does this LiveResource configuration want streaming? */
+export function shouldStream(o: StreamGateInput): boolean {
+  if (!o.streamUrl || !o.applyEvent) return false
+  if (o.enabled === false) return false
+  if (isRef(o.enabled) && o.enabled.value === false) return false
+  return true
+}
+
+const BACKOFF_MS = [1_000, 2_000, 4_000, 8_000, 16_000, 30_000] as const
+const GIVEUP_AFTER_CONSECUTIVE_FAILURES = 3
+
+/**
+ * Fetch-based SSE client. Sends `Authorization: Bearer <token>` (browser
+ * EventSource cannot set headers). On 401 → calls onUnauthorized and rejects.
+ * On 403/404 → resolves (permanent client error; consumer falls back to
+ * polling). On any other disconnect/error: exponential backoff with ±20%
+ * jitter, capped at 30 s. Successful chunk receipt resets the backoff
+ * counter. Aborts when the AbortSignal fires.
+ *
+ * Resolves when consecutive failures exceed GIVEUP_AFTER_CONSECUTIVE_FAILURES
+ * (signal for the consumer to fall back to polling). Rejects only on 401.
+ */
+export async function streamSse(opts: StreamSseOptions): Promise<void> {
+  const fetchFn = opts.fetchImpl ?? globalThis.fetch
+  let consecutiveFailures = 0
+  let backoffIdx = 0
+  while (!opts.signal.aborted) {
+    let connected = false
+    try {
+      const headers: Record<string, string> = {
+        Accept: 'text/event-stream',
+      }
+      if (opts.token) headers.Authorization = `Bearer ${opts.token}`
+      const resp = await fetchFn(opts.url, {
+        method: 'GET', headers, signal: opts.signal,
+      })
+      if (resp.status === 401) {
+        opts.onUnauthorized()
+        throw new Error('SSE 401')
+      }
+      // Pre-review IMPORTANT fix: permanent client errors (task gone /
+      // forbidden) should fail-fast — burning 7+ s of backoff on a 404 is
+      // pure latency before the consumer's poll-fallback kicks in.
+      if (resp.status === 403 || resp.status === 404) {
+        return
+      }
+      if (!resp.ok || !resp.body) {
+        throw new Error(`SSE upstream status ${resp.status}`)
+      }
+      const reader = resp.body.getReader()
+      const decoder = new TextDecoder()
+      let buf = ''
+      // eslint-disable-next-line no-constant-condition
+      while (true) {
+        const { value, done } = await reader.read()
+        if (done) break
+        buf += decoder.decode(value, { stream: true })
+        const { events, remainder } = parseSseChunk(buf)
+        buf = remainder
+        for (const ev of events) {
+          opts.onEvent(ev)
+          if (!connected) {
+            connected = true
+            consecutiveFailures = 0
+            backoffIdx = 0
+          }
+        }
+      }
+    } catch (err) {
+      if (opts.signal.aborted) return
+      if ((err as Error).message === 'SSE 401') throw err
+      consecutiveFailures += 1
+      if (consecutiveFailures >= GIVEUP_AFTER_CONSECUTIVE_FAILURES) return
+    }
+    if (opts.signal.aborted) return
+    const base = BACKOFF_MS[Math.min(backoffIdx, BACKOFF_MS.length - 1)]
+      ?? 30_000
+    const jitter = base * (0.8 + Math.random() * 0.4)
+    backoffIdx += 1
+    await new Promise<void>((resolve) => {
+      if (opts.signal.aborted) { resolve(); return }
+      const t = setTimeout(resolve, jitter)
+      opts.signal.addEventListener('abort',
+        () => { clearTimeout(t); resolve() }, { once: true })
+    })
+  }
+}
+
 /**
  * Pure SSE wire-format parser. The caller accumulates raw text chunks from a
  * ReadableStream and feeds them in; this function returns any complete events
diff --git a/frontend/tests/unit/streamGate.spec.ts b/frontend/tests/unit/streamGate.spec.ts
new file mode 100644
index 0000000..343648f
--- /dev/null
+++ b/frontend/tests/unit/streamGate.spec.ts
@@ -0,0 +1,33 @@
+import { describe, expect, test } from 'vitest'
+import { ref } from 'vue'
+import { shouldStream } from '@/api/sse'
+
+describe('shouldStream', () => {
+  test('no streamUrl → false', () => {
+    expect(shouldStream({ streamUrl: undefined, applyEvent: () => 1,
+      enabled: true })).toBe(false)
+  })
+  test('no applyEvent → false', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: undefined,
+      enabled: true })).toBe(false)
+  })
+  test('enabled: false → false', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: false })).toBe(false)
+  })
+  test('enabled: Ref<false> → false (unwrapped)', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: ref(false) })).toBe(false)
+  })
+  test('all conditions met → true', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: true })).toBe(true)
+  })
+  test('enabled: Ref<true> → true', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1,
+      enabled: ref(true) })).toBe(true)
+  })
+  test('enabled: undefined defaults to true', () => {
+    expect(shouldStream({ streamUrl: '/x', applyEvent: () => 1 })).toBe(true)
+  })
+})
diff --git a/frontend/tests/unit/streamSse.spec.ts b/frontend/tests/unit/streamSse.spec.ts
new file mode 100644
index 0000000..0ef1eed
--- /dev/null
+++ b/frontend/tests/unit/streamSse.spec.ts
@@ -0,0 +1,54 @@
+import { describe, expect, test } from 'vitest'
+import { streamSse, type SseEvent } from '@/api/sse'
+
+describe('streamSse', () => {
+  test('parses events from a ReadableStream and stops on abort', async () => {
+    const ac = new AbortController()
+    const got: SseEvent[] = []
+    const stream = new ReadableStream<Uint8Array>({
+      start(c) {
+        c.enqueue(new TextEncoder().encode('data: a\n\n'))
+        c.enqueue(new TextEncoder().encode('data: b\n\n'))
+        c.close()
+      },
+    })
+    const fetchImpl: typeof fetch = async () => new Response(
+      stream, { status: 200 })
+    await streamSse({
+      url: '/x', token: 't', signal: ac.signal,
+      onEvent: (e) => {
+        got.push(e)
+        if (got.length === 2) ac.abort()
+      },
+      onUnauthorized: () => {},
+      fetchImpl,
+    })
+    expect(got.map((e) => e.data)).toEqual(['a', 'b'])
+  })
+
+  test('401 → calls onUnauthorized and rejects without retry', async () => {
+    const ac = new AbortController()
+    let unauth = 0
+    const fetchImpl: typeof fetch = async () => new Response(
+      '', { status: 401 })
+    await expect(streamSse({
+      url: '/x', token: 't', signal: ac.signal,
+      onEvent: () => {},
+      onUnauthorized: () => { unauth++ },
+      fetchImpl,
+    })).rejects.toThrow(/SSE 401/)
+    expect(unauth).toBe(1)
+  })
+
+  test('404 → resolves without retry (permanent client error)', async () => {
+    const ac = new AbortController()
+    const fetchImpl: typeof fetch = async () => new Response(
+      'not found', { status: 404 })
+    await streamSse({
+      url: '/x', token: 't', signal: ac.signal,
+      onEvent: () => {},
+      onUnauthorized: () => {},
+      fetchImpl,
+    })
+  })
+})

From f496bd29dbdd0fa084aa6ffdde3473e7e19132d1 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:28:12 +0800
Subject: [PATCH 08/10] UI-SP5 M3: useLiveResource opt-in SSE
 (streamUrl+applyEvent; view-transparent)

---
 frontend/src/composables/useLiveResource.ts | 92 ++++++++++++++++++---
 1 file changed, 82 insertions(+), 10 deletions(-)

diff --git a/frontend/src/composables/useLiveResource.ts b/frontend/src/composables/useLiveResource.ts
index b7f3309..2c6620e 100644
--- a/frontend/src/composables/useLiveResource.ts
+++ b/frontend/src/composables/useLiveResource.ts
@@ -1,5 +1,7 @@
-import { useQuery, type QueryKey } from '@tanstack/vue-query'
-import type { Ref } from 'vue'
+import { onScopeDispose, ref, toValue, watch, type MaybeRefOrGetter, type Ref, type WatchStopHandle } from 'vue'
+import { useQuery, useQueryClient, type QueryKey } from '@tanstack/vue-query'
+import { shouldStream, streamSse, type SseEvent } from '@/api/sse'
+import { useAuthStore } from '@/stores/auth'
 
 const ERROR_BACKOFF_MS = 5_000
 const HIDDEN_MULTIPLIER = 3
@@ -17,22 +19,41 @@ export interface LiveOptions<T> {
   isTerminal?: (data: T) => boolean
   staleTime?: number
   enabled?: Ref<boolean> | boolean
+  /** UI-SP5: opt in to SSE. When set with applyEvent, the composable opens a
+   * stream after the first useQuery success and writes events into the cache
+   * via setQueryData. Polling stays disabled while streaming is healthy and
+   * resumes automatically if the stream gives up. */
+  streamUrl?: string | Ref<string>
+  applyEvent?: (prev: T | undefined, ev: SseEvent) => T
 }
 
 /**
- * Single realtime seam. Today: adaptive polling on vue-query. UI-SP5 swaps
- * the internals to SSE/WS — consumers (views) never change.
+ * Single realtime seam. Today: adaptive polling on vue-query, with an
+ * additive opt-in SSE swap (UI-SP5). UI-SP4 (AI-Copilot) and future
+ * transports plug in here without view changes.
  *
- * vue-query v5.59 does NOT accept a getter for `queryKey` — it must be a
- * QueryKey (array). Reactivity comes from putting refs *inside* the array
- * (v5 unwraps them), exactly like the scaffold's proven useTaskDetail.
+ * vue-query v5 does NOT accept a getter for `queryKey` — it must be a
+ * QueryKey (array). Reactivity comes from putting refs *inside* the array.
  */
 export function useLiveResource<T>(
   key: QueryKey,
   fetcher: () => Promise<T>,
   opts: LiveOptions<T>,
 ) {
-  return useQuery<T>({
+  const streaming = shouldStream({
+    streamUrl: opts.streamUrl,
+    // `shouldStream` only checks truthiness; cast widens T → unknown safely.
+    applyEvent: opts.applyEvent as
+      ((prev: unknown, ev: SseEvent) => unknown) | undefined,
+    enabled: opts.enabled,
+  })
+
+  // Pre-review fix (IMPORTANT 1): use a ref for clarity; the callback re-reads
+  // the closure variable per refetchInterval-eval, but matching Vue idioms
+  // makes intent obvious and future-proofs anyone wanting to watch() this.
+  const pollingFallback = ref(false)
+
+  const q = useQuery<T>({
     queryKey: key,
     queryFn: fetcher,
     enabled: opts.enabled,
@@ -41,11 +62,62 @@ export function useLiveResource<T>(
       const data = query.state.data as T | undefined
       const errored = query.state.status === 'error'
       const terminal = data !== undefined && !!opts.isTerminal?.(data)
-      const hidden =
-        typeof document !== 'undefined' && document.visibilityState === 'hidden'
+      const hidden = typeof document !== 'undefined'
+        && document.visibilityState === 'hidden'
+      if (streaming && !pollingFallback.value) {
+        return false
+      }
       return computeInterval({
         base: opts.baseIntervalMs, terminal, hidden, errored,
       })
     },
   })
+
+  if (streaming && opts.applyEvent) {
+    const qc = useQueryClient()
+    const ac = new AbortController()
+    const auth = useAuthStore()
+    const apply = opts.applyEvent
+    // Pre-review BLOCKER fix: Vue 3.5's `watch({ immediate: true })` invokes
+    // the handler SYNCHRONOUSLY inside the watch() call — before `stopWatch`
+    // is assigned the returned stop handle. If vue-query serves a cached
+    // snapshot synchronously (e.g. HMR / route revisit / future initialData),
+    // calling `stopWatch()` from inside the handler would hit TDZ
+    // (ReferenceError). Fix: `let stopWatch` + optional-call + a `started`
+    // flag so the kick-off runs exactly once even if the watcher fires
+    // synchronously on registration.
+    let stopWatch: WatchStopHandle | undefined
+    let started = false
+    stopWatch = watch(
+      () => q.data.value,
+      (snapshot) => {
+        if (snapshot === undefined || started) return
+        started = true
+        stopWatch?.()  // safe — undefined on synchronous immediate fire
+        const url = toValue(opts.streamUrl as MaybeRefOrGetter<string>)
+        void streamSse({
+          url, token: auth.accessToken, signal: ac.signal,
+          onEvent: (ev) => {
+            const prev = qc.getQueryData<T>(key)
+            const next = apply(prev, ev)
+            qc.setQueryData(key, next)
+          },
+          onUnauthorized: () => {
+            auth.logout()
+          },
+        }).then(() => {
+          // streamSse resolved without abort → it gave up (3 consecutive
+          // failures). Fall back to polling.
+          pollingFallback.value = true
+          void q.refetch()
+        }).catch(() => {
+          // 401 path — onUnauthorized already invoked.
+        })
+      },
+      { immediate: true },
+    )
+    onScopeDispose(() => { ac.abort() })
+  }
+
+  return q
 }

From 75689e6f12d338d61f516b279c099abe28d4eecc Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:29:19 +0800
Subject: [PATCH 09/10] UI-SP5 M3: useTaskDetail opts in to SSE (view-free; all
 consumers unchanged)

---
 frontend/src/composables/useTaskDetail.ts | 13 ++++++++++---
 1 file changed, 10 insertions(+), 3 deletions(-)

diff --git a/frontend/src/composables/useTaskDetail.ts b/frontend/src/composables/useTaskDetail.ts
index 24170d9..0e4929a 100644
--- a/frontend/src/composables/useTaskDetail.ts
+++ b/frontend/src/composables/useTaskDetail.ts
@@ -1,12 +1,19 @@
-import type { Ref } from 'vue'
+import { computed, type Ref } from 'vue'
 import { useLiveResource } from '@/composables/useLiveResource'
 import { client } from '@/api/client'
 import { TERMINAL_STATUSES, type TaskDetail } from '@/api/types'
 
 export function useTaskDetail(taskId: Ref<string>) {
+  const streamUrl = computed(() => `/api/v1/tasks/${taskId.value}/stream`)
   return useLiveResource<TaskDetail>(
     ['task', taskId],
-    async () => (await client.get<TaskDetail>(`/api/v1/tasks/${taskId.value}`)).data,
-    { baseIntervalMs: 1_000, isTerminal: (d) => TERMINAL_STATUSES.has(d.status) },
+    async () => (await client.get<TaskDetail>(
+      `/api/v1/tasks/${taskId.value}`)).data,
+    {
+      baseIntervalMs: 1_000,
+      isTerminal: (d) => TERMINAL_STATUSES.has(d.status),
+      streamUrl,
+      applyEvent: (_prev, ev) => JSON.parse(ev.data) as TaskDetail,
+    },
   )
 }

From cf2190ce09b213861f63dd371dfd3165dadc57c7 Mon Sep 17 00:00:00 2001
From: l17728 <l17728@users.noreply.github.com>
Date: Wed, 20 May 2026 13:35:06 +0800
Subject: [PATCH 10/10] UI-SP5 M3: operator docs for the realtime SSE swap

---
 docs/operator/web-ui.md | 37 +++++++++++++++++++++++++++++++++++++
 1 file changed, 37 insertions(+)

diff --git a/docs/operator/web-ui.md b/docs/operator/web-ui.md
index eb40eb4..fd47c35 100644
--- a/docs/operator/web-ui.md
+++ b/docs/operator/web-ui.md
@@ -129,3 +129,40 @@ drain/restart, metrics history, heartbeat history; HF-token rotation;
 license-policy CRUD; source-driver registration; maintenance mode;
 `/quota/usage` (declared but no backing tables); ML forecast; chargeback PDF;
 real-time audit tail (UI-SP5).
+
+## UI-SP5 — Realtime SSE swap (delivered)
+
+`useLiveResource` now supports an opt-in SSE transport. The locked promise
+recorded in SP1/SP2/SP3 — *"SP5 swaps internals to SSE/WS with zero view
+changes"* — is honored: every view, page, and other composable is unchanged.
+
+- **Backend**: `GET /api/v1/tasks/{id}/stream` — hand-rolled
+  `text/event-stream` via FastAPI `StreamingResponse` (same idiom as
+  `hf_proxy.py`). 1 Hz default tick rate (env
+  `DLW_TASK_STREAM_INTERVAL_SECONDS` overrides; clamped `[0.1, 10.0]`).
+  Tenant-scoped via the proven cancel-pattern (404 cross-tenant). Terminates
+  on terminal task status, client disconnect, or controller shutdown. A
+  `?max_ticks=N` query param is supported for testability (httpx
+  `ASGITransport` buffers the response body until the generator closes, so
+  multi-tick tests need a natural-termination hatch).
+- **Frontend**: `frontend/src/api/sse.ts` (pure `parseSseChunk` + `streamSse`
+  fetch + ReadableStream with exponential backoff, Bearer-via-header). The
+  `useLiveResource` composable gains optional `streamUrl` + `applyEvent`
+  fields; when both are set, the composable opens an SSE connection after
+  the first snapshot and writes events into the vue-query cache. On 3
+  consecutive failures the stream gives up and polling resumes automatically.
+  On 401 the streamer calls `auth.logout()`; on 403/404 it fails fast (no
+  backoff burn).
+- **One consumer opts in**: `useTaskDetail`. Every other composable
+  (`useTaskList`, `useQuota`, the 4 SP2 sub-resource composables, the 3 SP3
+  composables) stays on polling.
+
+**At a glance**: open `/tasks/<id>` in DevTools Network — you'll see one
+long-lived `text/event-stream` connection per visit instead of a 1 Hz polling
+loop.
+
+**Deferrals**: WebSocket transport (SSE delivers the same outcome simpler);
+streaming for the 4 SP2 sub-resources (would force a multi-resource envelope
+that breaks view-free); SSE for the low-frequency SP3 composables (push value
+is negligible at 5–30 s cadences). UI-SP4 (AI-Copilot) remains the v2.1
+follow-up.