Merge pull request 'worker を追加（ステップ6a）' (#9) from worker-download into main

Reviewed-on: #9

.env.example

@@ -1,3 +1,4 @@
 DATABASE_URL=
 VIDEO_HOST_DIR=
 VIDEO_DIR=/videos
+WORKER_CONCURRENCY=1

Dockerfile.worker

@@ -0,0 +1,21 @@
+FROM golang:1.27.1-alpine AS build
+
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY cmd ./cmd
+COPY internal ./internal
+RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /worker ./cmd/worker
+
+FROM python:3.13-slim
+ARG YT_DLP_VERSION=2026.08.19
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ca-certificates ffmpeg tzdata \
+    && pip install --no-cache-dir "yt-dlp[default]==${YT_DLP_VERSION}" \
+    && rm -rf /var/lib/apt/lists/*
+WORKDIR /app
+COPY --from=build /worker /app/worker
+ENV HOME=/tmp \
+    XDG_CACHE_HOME=/tmp/yt-dlp-cache
+USER 65532:65532
+ENTRYPOINT ["/app/worker"]

PROJECT.md

@@ -3,7 +3,7 @@ title: tver の現状
 tags:
   - homelab
   - tver
-status: ライブラリ・再生・削除を実装中
+status: worker のダウンロード本体を実装中
 ---
 
 # tver の現状
@@ -13,8 +13,8 @@ status: ライブラリ・再生・削除を実装中
 
 ## 今の状態
 
-- Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web コンテナの定義がある
-- **ライブラリ・再生・削除を実装中。** 完了したジョブを一覧し、ブラウザで Range 対応の再生とファイル・ジョブ記録の削除ができる。worker がまだ無いため、実際のダウンロード実行は行わない
+- Go の HTTP サーバー、TVer API クライアント、ドラマ・バラエティの新着表示、テスト、web・worker コンテナの定義がある
+- **worker のダウンロード本体を実装中。** 待ち行列から yt-dlp を起動し、進捗・heartbeat・取消・出力検査・保存・失敗分類を扱う。登録番組の定期確認はまだ行わない
 - リポジトリ: `https://gitea.kazu634.com/kazu634/tver.git`
 
 ## 目的
@@ -45,7 +45,12 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 | 番組の一時停止 | 登録は残し、自動ダウンロードだけ止める。再開後は停止中に公開された未確認の回も対象にする |
 | 動画のマウント先 | web と worker は `VIDEO_DIR`（既定 `/videos`）に同じ NAS 共有をマウントする。`download_jobs.file_path` は `VIDEO_DIR` からの相対パスを保存する |
 | ファイル名 | `<番組名>/<番組名> <回タイトル> [<エピソードID>].mp4`。禁止文字と制御文字を `_` に置換し、空の要素は `名称未設定`、各入力要素は UTF-8 で60バイトまでに切り詰める。エピソード ID で同名衝突を避ける |
+| 同名ファイル | 移動先に同名ファイルがあれば上書きする。ファイル名にエピソード ID が入り `episode_id` は一意なので、同名は同じ回の前回処理の残りに限られる。再実行で同じ結果になることを優先する |
 | 動画の削除 | ファイルを先に削除してからジョブ記録を削除する。ファイルが既に無い場合も記録は削除する。登録番組の `known_episodes` は残す |
+| 同時ダウンロード数 | 既定は **1**。`WORKER_CONCURRENCY` で **2まで**増やせる |
+| ダウンロードの再試行 | 自動では再試行しない。失敗を `failed` で残し、画面から手動で再試行する |
+| 実行中の取り消し | web で `running` を `canceled` にし、worker が検知して yt-dlp のプロセスグループを停止する |
+| yt-dlp の作業場所 | `VIDEO_DIR/.incoming/<job id>/` に `%(id)s.%(ext)s` で保存し、検査後に Go が `media.RelativeVideoPath` の場所へ同一ファイルシステム内で移動する。終了状態にかかわらず作業場所を片付ける |
 
 ## 要件
 
@@ -84,13 +89,15 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 登録と公開中の回の既知記録・任意の待ち行列投入は1トランザクションで行い、登録時刻を最終確認日時に設定する。`known_episodes` は番組解除時に連動して削除する。worker 用の未確認回の抽出と、未確認回だけの既知記録・投入も `internal/store` に用意した。一時停止中は投入を拒否する
 - 新着の登録ボタンはエピソード ID とジャンルだけを送る。web が有効な新着キャッシュから番組情報を引き、`episode_id` の一意制約で重複登録を防ぐ。新着一覧はジョブ状態を一括取得し、失敗・取り消しには再試行を表示する
 - 待ち行列は実行中・待機中・失敗・取り消し・完了を状態別に直近50件ずつ表示する。段階別の進捗と失敗理由を表示し、待機中の取り消しと失敗・取り消しの再試行ができる。実行中または待機中のジョブがある間だけ5秒ごとに一覧を更新する
+- 実行中のジョブにも取り消しボタンを表示する。worker は DB の `canceled` を定期確認し、yt-dlp をプロセスグループごと停止する
 - 新着はジャンルごとにメモリへ 10 分間キャッシュする。キャッシュ中は TVer セッションの再発行と一覧の再取得を行わない。TVer の取得に失敗した場合、一覧の場所にエラーを表示して詳細をログへ記録する
 - 「予告を隠す」は既定でオンとし、180 秒以下の件を非表示にする。長さは秒から分へ切り上げて表示し、配信終了日時は Asia/Tokyo で表示する。ジャンルと予告表示の状態は URL に残す
 - 画面は `html/template` と htmx 2.0.11 を使い、htmx は Go の `embed` から `/static/` でローカル配信する
-- `compose.yaml` は web のみを定義し、dockage VM の LAN アドレス `192.168.10.155:8080` で公開する。待ち受けは `WEB_ADDR` で変更でき、既定は `:8080`。DB 接続先は `DATABASE_URL`、NAS のホスト側マウント先は `VIDEO_HOST_DIR`、コンテナ内の保存先は `VIDEO_DIR`（既定 `/videos`）として `.env` から渡す（値の無い雛形は `.env.example`）
+- `compose.yaml` は web と worker を定義する。web は dockage VM の LAN アドレス `192.168.10.155:8080` で公開し、待ち受けは `WEB_ADDR` で変更できる。両方へ `DATABASE_URL` と同じ動画ボリュームを渡し、worker の並行数は `WORKER_CONCURRENCY`（既定1、最大2）で指定する
 - DB は NAS 上に tver 専用 Postgres 17 コンテナを置く。`internal/store` の埋め込み SQL を `schema_migrations` で管理し、`pg_advisory_lock` で複数プロセスからの同時適用を直列化する。web は起動時に `pgxpool` で接続してマイグレーションを適用する
 - `download_jobs` は `episode_id` を一意にして二重登録を防ぎ、ジョブの状態・段階別進捗・結果を持つ。`subscriptions` はシリーズ単位の登録と最終確認日時を持つ。worker からの操作と画面用の一覧取得を `internal/store` にまとめる
-- worker は `yt-dlp[default]` と ffmpeg を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を扱う。正確な全体パーセントは保存しない。`after_move:filepath` で実パスを取り、ffprobe で H.264 と AAC の MP4 を確認して成功とする。失敗は DRM、地域制限、ネットワーク、配信不可、抽出失敗、出力不正、その他の順に分類する
+- worker は `yt-dlp[default]` と ffmpeg を外部コマンドとして使い、映像・音声・結合・移動の段階別進捗を1秒間隔に間引いて保存する。独立した heartbeat を5秒ごとに更新し、正確な全体パーセントは保存しない。`after_move:filepath` で実パスを取り、ffprobe で H.264 と AAC の MP4 を確認して成功とする。失敗は取り消し、DRM、地域制限、ネットワーク、配信不可、抽出失敗、出力不正、その他の順に扱い、自動再試行しない
+- worker は起動時にマイグレーションと停止中ジョブの `RequeueStale` を行う。SIGTERM では yt-dlp を止め、担当中ジョブを `queued` に戻してから終了する。`Dockerfile.worker` は yt-dlp 2026.08.19 を ARG で固定し、ffmpeg・ffprobe とともに非 root で実行する
 - 保存先の NAS 共有フォルダは dockage VM に NFS でマウントし、`worker`（書き込み）と `web`（読み取り・削除）に同じ `VIDEO_DIR` でバインドマウントする
 
 ## 外部とのつながり
@@ -121,7 +128,7 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 新着 API が 100 件で打ち切りか、続きを取得できるか [要確認]
 - 新着のうち DRM 付きの作品の割合 [要確認]
 - 非公開 API と yt-dlp の TVer 対応は、TVer 側の変更で予告なく壊れうる。壊れたことを検知する手段が要る
-- 登録番組の24時間ごとの確認と、待ち行列のジョブを実行する worker は未実装
+- 登録番組の24時間ごとの確認は未実装
 - NAS の準備（NFS 共有フォルダ、tver 専用 Postgres 17 コンテナ、DSM ファイアウォール）は DSM での手作業になる。DB の公開ポート（候補 15434）は本番前に実機で確認する [要確認]
 - DRM・地域制限の yt-dlp 失敗時の文言と終了コードは未実測 [要確認]
 
@@ -134,5 +141,6 @@ TVer（`https://tver.jp/`）で公開中の動画情報を取得し、選んだ
 - 番組一覧の応答は `internal/tver/testdata/series_seasons.json` と `season_episodes.json`（2026-10-03 の実応答を各1件に縮小）に記録した。テストは `httptest` を使い、実 API にアクセスしない
 - 新着画面のテストは偽物の TVer クライアント・store と差し替え可能な時計を使い、表示項目、絞り込みの境界、10 分キャッシュ、取得失敗、JST 表示、登録・状態表示・待ち行列の操作を確認する
 - ライブラリのテストは一時ディレクトリの小さな動画を使い、Range の 206・範囲外の 416、相対パスとシンボリックリンクの検査、404、ファイル削除後の記録削除を確認する。共有のファイル名規則は `internal/media` でテストする
+- worker のテストは `internal/worker/testdata` の偽 yt-dlp・ffprobe を使い、本物の TVer へ接続せずに正常完了、失敗分類、取消、終了時の待機復帰、進捗の間引き、heartbeat、出力検査、移動先、作業場所の片付けを確認する
 - `env GOPATH=/private/tmp/tver-gopath GOCACHE=/private/tmp/tver-go-cache GOMODCACHE=/private/tmp/tver-go-modcache GOPROXY=off go vet ./...` で静的解析する
-- `DATABASE_URL`、`VIDEO_DIR`、`WEB_ADDR=127.0.0.1:18080` を設定して `go run ./cmd/web` で起動し、`/healthz`、`/`、`/queue`、`/subscriptions`、`/library` を確認できる。TVer への実アクセスを伴うため手動確認は少数回に留める
+- `DATABASE_URL`、`VIDEO_DIR`、`WEB_ADDR=127.0.0.1:18080` を設定して `go run ./cmd/web` で web を起動できる。worker は同じ `DATABASE_URL` と `VIDEO_DIR` を設定して `go run ./cmd/worker` で起動する。worker の実行は TVer への実アクセスを伴うため手動確認は少数回に留める

cmd/worker/main.go

@@ -0,0 +1,82 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	workerpkg "gitea.kazu634.com/kazu634/tver/internal/worker"
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+const staleAfter = 30 * time.Second
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+	config, err := loadConfig(os.Getenv)
+	if err != nil {
+		log.Fatal(err)
+	}
+	pool, err := initializeDatabase(ctx, os.Getenv("DATABASE_URL"), time.Now(), pgxpool.New,
+		store.Migrate, func(ctx context.Context, pool *pgxpool.Pool, cutoff time.Time) (int64, error) {
+			return store.New(pool).RequeueStale(ctx, cutoff)
+		})
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer pool.Close()
+	runner, err := workerpkg.New(store.New(pool), config, log.Default())
+	if err != nil {
+		log.Fatal(err)
+	}
+	log.Printf("worker started (concurrency=%d)", config.Concurrency)
+	if err := runner.Run(ctx); err != nil {
+		log.Fatal(err)
+	}
+	log.Print("worker stopped")
+}
+
+func loadConfig(getenv func(string) string) (workerpkg.Config, error) {
+	concurrency := 1
+	if value := strings.TrimSpace(getenv("WORKER_CONCURRENCY")); value != "" {
+		parsed, err := strconv.Atoi(value)
+		if err != nil || parsed < 1 || parsed > 2 {
+			return workerpkg.Config{}, errors.New("WORKER_CONCURRENCY は 1 または 2 を指定してください")
+		}
+		concurrency = parsed
+	}
+	return workerpkg.Config{VideoDir: getenv("VIDEO_DIR"), YTDLPPath: getenv("YT_DLP_PATH"),
+		FFprobePath: getenv("FFPROBE_PATH"), Concurrency: concurrency}, nil
+}
+
+func initializeDatabase(ctx context.Context, databaseURL string, now time.Time,
+	connect func(context.Context, string) (*pgxpool.Pool, error),
+	migrate func(context.Context, *pgxpool.Pool) error,
+	requeue func(context.Context, *pgxpool.Pool, time.Time) (int64, error),
+) (*pgxpool.Pool, error) {
+	if strings.TrimSpace(databaseURL) == "" {
+		return nil, errors.New("DATABASE_URL が設定されていません")
+	}
+	pool, err := connect(ctx, databaseURL)
+	if err != nil {
+		return nil, errors.New("DATABASE_URL から接続プールを作成できません")
+	}
+	if err := migrate(ctx, pool); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("DB マイグレーションに失敗: %w", err)
+	}
+	if _, err := requeue(ctx, pool, now.Add(-staleAfter)); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("停止中ジョブの復帰に失敗: %w", err)
+	}
+	return pool, nil
+}

cmd/worker/main_test.go

@@ -0,0 +1,97 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+func TestLoadConfigConcurrency(t *testing.T) {
+	values := map[string]string{"VIDEO_DIR": "/media", "YT_DLP_PATH": "/bin/downloader", "FFPROBE_PATH": "/bin/probe"}
+	getenv := func(key string) string { return values[key] }
+	config, err := loadConfig(getenv)
+	if err != nil || config.Concurrency != 1 || config.VideoDir != "/media" || config.YTDLPPath != "/bin/downloader" || config.FFprobePath != "/bin/probe" {
+		t.Fatalf("既定設定: %+v %v", config, err)
+	}
+	values["WORKER_CONCURRENCY"] = "2"
+	config, err = loadConfig(getenv)
+	if err != nil || config.Concurrency != 2 {
+		t.Fatalf("並行数2: %+v %v", config, err)
+	}
+	for _, invalid := range []string{"0", "3", "bad"} {
+		values["WORKER_CONCURRENCY"] = invalid
+		if _, err := loadConfig(getenv); err == nil {
+			t.Errorf("不正な並行数 %q を許した", invalid)
+		}
+	}
+}
+
+func TestInitializeDatabaseRequiresURLMigratesAndRequeues(t *testing.T) {
+	ctx := context.Background()
+	now := time.Date(2026, 10, 3, 12, 0, 0, 0, time.UTC)
+	connectCalls, migrateCalls, requeueCalls := 0, 0, 0
+	connect := func(_ context.Context, url string) (*pgxpool.Pool, error) {
+		connectCalls++
+		if url != "postgres://test:secret@localhost/test" {
+			t.Errorf("URL=%q", url)
+		}
+		return pgxpool.New(ctx, url)
+	}
+	migrate := func(context.Context, *pgxpool.Pool) error { migrateCalls++; return nil }
+	requeue := func(_ context.Context, _ *pgxpool.Pool, cutoff time.Time) (int64, error) {
+		requeueCalls++
+		if want := now.Add(-staleAfter); !cutoff.Equal(want) {
+			t.Errorf("cutoff=%v want=%v", cutoff, want)
+		}
+		return 2, nil
+	}
+	if _, err := initializeDatabase(ctx, " ", now, connect, migrate, requeue); err == nil || !strings.Contains(err.Error(), "DATABASE_URL") {
+		t.Fatalf("未設定 URL: %v", err)
+	}
+	if connectCalls != 0 || migrateCalls != 0 || requeueCalls != 0 {
+		t.Fatalf("未設定時の呼び出し: %d %d %d", connectCalls, migrateCalls, requeueCalls)
+	}
+	pool, err := initializeDatabase(ctx, "postgres://test:secret@localhost/test", now, connect, migrate, requeue)
+	if err != nil || pool == nil {
+		t.Fatalf("初期化: %v", err)
+	}
+	pool.Close()
+	if connectCalls != 1 || migrateCalls != 1 || requeueCalls != 1 {
+		t.Fatalf("呼び出し回数: %d %d %d", connectCalls, migrateCalls, requeueCalls)
+	}
+}
+
+func TestInitializeDatabaseStopsOnMigrationOrRequeueFailure(t *testing.T) {
+	ctx := context.Background()
+	connect := func(context.Context, string) (*pgxpool.Pool, error) {
+		return pgxpool.New(ctx, "postgres://localhost/test")
+	}
+	requeueCalled := false
+	pool, err := initializeDatabase(ctx, "postgres://localhost/test", time.Now(), connect,
+		func(context.Context, *pgxpool.Pool) error { return errors.New("migration failed") },
+		func(context.Context, *pgxpool.Pool, time.Time) (int64, error) { requeueCalled = true; return 0, nil })
+	if pool != nil || err == nil || !strings.Contains(err.Error(), "migration failed") || requeueCalled {
+		t.Fatalf("migration failure: pool=%v err=%v requeue=%v", pool, err, requeueCalled)
+	}
+	pool, err = initializeDatabase(ctx, "postgres://localhost/test", time.Now(), connect,
+		func(context.Context, *pgxpool.Pool) error { return nil },
+		func(context.Context, *pgxpool.Pool, time.Time) (int64, error) { return 0, errors.New("requeue failed") })
+	if pool != nil || err == nil || !strings.Contains(err.Error(), "requeue failed") {
+		t.Fatalf("requeue failure: pool=%v err=%v", pool, err)
+	}
+}
+
+func TestInitializeDatabaseDoesNotLeakURL(t *testing.T) {
+	_, err := initializeDatabase(context.Background(), "postgres://user:secret@localhost/test", time.Now(),
+		func(context.Context, string) (*pgxpool.Pool, error) {
+			return nil, errors.New("postgres://user:secret@localhost/test refused")
+		}, func(context.Context, *pgxpool.Pool) error { return nil },
+		func(context.Context, *pgxpool.Pool, time.Time) (int64, error) { return 0, nil })
+	if err == nil || strings.Contains(err.Error(), "secret") {
+		t.Fatalf("URL 漏えい: %v", err)
+	}
+}

compose.yaml

@@ -11,3 +11,15 @@ services:
     ports:
       - "192.168.10.155:8080:8080"
     restart: unless-stopped
+  worker:
+    build:
+      context: .
+      dockerfile: Dockerfile.worker
+    env_file: .env
+    environment:
+      TZ: Asia/Tokyo
+      VIDEO_DIR: "${VIDEO_DIR:-/videos}"
+      WORKER_CONCURRENCY: "${WORKER_CONCURRENCY:-1}"
+    volumes:
+      - "${VIDEO_HOST_DIR:?VIDEO_HOST_DIR must be set}:${VIDEO_DIR:-/videos}"
+    restart: unless-stopped

internal/store/store.go

@@ -183,6 +183,13 @@ func (s *Store) UpdateProgress(ctx context.Context, id int64, p Progress) error
 	return transitionResult(tag.RowsAffected(), err)
 }
 
+// Heartbeat は進捗出力がない間も実行中ジョブの生存を記録する。
+func (s *Store) Heartbeat(ctx context.Context, id int64) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=now(), updated_at=now()
+		WHERE id=$1 AND status='running'`, id)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
 func (s *Store) Complete(ctx context.Context, id int64, filePath string, fileSizeBytes int64) error {
 	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='succeeded', phase=NULL,
 		file_path=$2, file_size_bytes=$3, heartbeat_at=NULL, finished_at=now(), updated_at=now()
@@ -222,6 +229,16 @@ func (s *Store) RequeueStale(ctx context.Context, cutoff time.Time) (int64, erro
 	return tag.RowsAffected(), err
 }
 
+// RequeueRunning は worker の正常終了時に担当中ジョブを待機中へ戻す。
+func (s *Store) RequeueRunning(ctx context.Context, id int64) error {
+	tag, err := s.pool.Exec(ctx, `UPDATE download_jobs SET status='queued', phase=NULL,
+		downloaded_bytes=NULL, total_bytes=NULL, total_is_estimate=false, speed_bps=NULL,
+		eta_seconds=NULL, fragment_index=NULL, fragment_count=NULL, format_id=NULL,
+		heartbeat_at=NULL, started_at=NULL, updated_at=now()
+		WHERE id=$1 AND status='running'`, id)
+	return transitionResult(tag.RowsAffected(), err)
+}
+
 func transitionResult(count int64, err error) error {
 	if err != nil {
 		return err

internal/store/store_test.go

@@ -405,6 +405,47 @@ func TestCancelAndStaleBoundary(t *testing.T) {
 	}
 }
 
+func TestHeartbeatAndGracefulRequeue(t *testing.T) {
+	s := New(testPool(t, true))
+	ctx := context.Background()
+	job, _, err := s.Enqueue(ctx, sampleEpisode("heartbeat"), "manual")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Heartbeat(ctx, job.ID); !errors.Is(err, ErrInvalidTransition) {
+		t.Fatalf("待機中の heartbeat: %v", err)
+	}
+	claimed, err := s.ClaimNext(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	old := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC)
+	if _, err := s.pool.Exec(ctx, `UPDATE download_jobs SET heartbeat_at=$2 WHERE id=$1`, claimed.ID, old); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Heartbeat(ctx, claimed.ID); err != nil {
+		t.Fatal(err)
+	}
+	heartbeated, err := s.GetJob(ctx, claimed.ID)
+	if err != nil || heartbeated.HeartbeatAt == nil || !heartbeated.HeartbeatAt.After(old) {
+		t.Fatalf("heartbeat: %+v %v", heartbeated, err)
+	}
+	bytes := int64(10)
+	if err := s.UpdateProgress(ctx, claimed.ID, Progress{Phase: "video", DownloadedBytes: &bytes}); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.RequeueRunning(ctx, claimed.ID); err != nil {
+		t.Fatal(err)
+	}
+	requeued, err := s.GetJob(ctx, claimed.ID)
+	if err != nil || requeued.Status != "queued" || requeued.StartedAt != nil || requeued.HeartbeatAt != nil || requeued.Phase != nil || requeued.DownloadedBytes != nil || requeued.Attempts != 1 {
+		t.Fatalf("graceful requeue: %+v %v", requeued, err)
+	}
+	if err := s.RequeueRunning(ctx, claimed.ID); !errors.Is(err, ErrInvalidTransition) {
+		t.Fatalf("待機中の再復帰: %v", err)
+	}
+}
+
 func TestRejectedTransitions(t *testing.T) {
 	pool := testPool(t, true)
 	s := New(pool)

internal/web/handler.go

@@ -32,7 +32,7 @@ type JobStore interface {
 	GetJob(context.Context, int64) (store.Job, error)
 	Retry(context.Context, int64) error
 	RetryEpisode(context.Context, string) error
-	CancelQueued(context.Context, int64) error
+	Cancel(context.Context, int64) error
 	SubscribedSeries(context.Context, []string) (map[string]bool, error)
 	RegisterSubscription(context.Context, string, string, []store.Episode, bool) (bool, error)
 	ListSubscriptions(context.Context) ([]store.Subscription, error)

internal/web/queue.go

@@ -145,34 +145,37 @@ func errorKindText(kind string) string {
 }
 
 func (s *server) retryJob(w http.ResponseWriter, r *http.Request) {
-	s.queueMutation(w, r, s.jobs.Retry, false)
+	s.queueMutation(w, r, s.jobs.Retry)
 }
 
 func (s *server) cancelJob(w http.ResponseWriter, r *http.Request) {
-	s.queueMutation(w, r, s.jobs.CancelQueued, true)
-}
-
-func (s *server) queueMutation(w http.ResponseWriter, r *http.Request, change func(context.Context, int64) error, queuedOnly bool) {
 	id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
 	if err != nil || id <= 0 {
 		http.Error(w, "ジョブ ID が不正です", http.StatusBadRequest)
 		return
 	}
-	if queuedOnly {
-		job, err := s.jobs.GetJob(r.Context(), id)
-		if errors.Is(err, pgx.ErrNoRows) {
-			http.Error(w, "ジョブが見つかりません", http.StatusNotFound)
-			return
-		}
-		if err != nil {
-			s.logger.Printf("ジョブ状態の取得に失敗: %v", err)
-			http.Error(w, "操作できませんでした", http.StatusInternalServerError)
-			return
-		}
-		if job.Status != "queued" {
-			http.Error(w, "待機中のジョブだけ取り消せます", http.StatusConflict)
-			return
-		}
+	job, err := s.jobs.GetJob(r.Context(), id)
+	if errors.Is(err, pgx.ErrNoRows) {
+		http.Error(w, "ジョブが見つかりません", http.StatusNotFound)
+		return
+	}
+	if err != nil {
+		s.logger.Printf("ジョブ状態の取得に失敗: %v", err)
+		http.Error(w, "操作できませんでした", http.StatusInternalServerError)
+		return
+	}
+	if job.Status != "queued" && job.Status != "running" {
+		http.Error(w, "待機中またはダウンロード中のジョブだけ取り消せます", http.StatusConflict)
+		return
+	}
+	s.queueMutation(w, r, s.jobs.Cancel)
+}
+
+func (s *server) queueMutation(w http.ResponseWriter, r *http.Request, change func(context.Context, int64) error) {
+	id, err := strconv.ParseInt(r.PathValue("id"), 10, 64)
+	if err != nil || id <= 0 {
+		http.Error(w, "ジョブ ID が不正です", http.StatusBadRequest)
+		return
 	}
 	if err := change(r.Context(), id); err != nil {
 		if errors.Is(err, store.ErrInvalidTransition) {

internal/web/queue_test.go

@@ -155,7 +155,7 @@ func TestQueueGroupsProgressAndPolling(t *testing.T) {
 			t.Errorf("待ち行列に %q がない", want)
 		}
 	}
-	if strings.Contains(body, errorMessage) || strings.Contains(body, `action="/jobs/1/cancel"`) || strings.Contains(body, "全体") {
+	if strings.Contains(body, errorMessage) || !strings.Contains(body, `action="/jobs/1/cancel"`) || strings.Contains(body, "全体") {
 		t.Errorf("不正な詳細・操作・全体進捗: %s", body)
 	}
 	fragment := getPage(handler, "/queue/list")
@@ -195,7 +195,7 @@ func TestQueueJobsAppearInCorrectSectionsWithLimit(t *testing.T) {
 		cancelAction := `action="/jobs/` + string(rune('1'+i)) + `/cancel"`
 		retryAction := `action="/jobs/` + string(rune('1'+i)) + `/retry"`
 		switch statuses[i] {
-		case "queued":
+		case "running", "queued":
 			if !strings.Contains(section[1], cancelAction) || strings.Contains(section[1], retryAction) {
 				t.Errorf("%s の操作が不正: %s", statuses[i], section[1])
 			}
@@ -246,9 +246,13 @@ func TestQueueCancelRetryAndStopPolling(t *testing.T) {
 	if got := postForm(handler, "/jobs/bad/cancel", nil).Code; got != http.StatusBadRequest {
 		t.Fatalf("不正な ID: %d", got)
 	}
+	if got := postForm(handler, "/jobs/99/cancel", nil).Code; got != http.StatusNotFound {
+		t.Fatalf("存在しないジョブ: %d", got)
+	}
 	jobs.jobs["ep"] = store.Job{ID: 1, Status: "running"}
-	if got := postForm(handler, "/jobs/1/cancel", nil).Code; got != http.StatusConflict {
-		t.Fatalf("実行中の取り消し: %d", got)
+	runningCanceled := postForm(handler, "/jobs/1/cancel", nil)
+	if runningCanceled.Code != http.StatusOK || jobs.jobs["ep"].Status != "canceled" || !strings.Contains(runningCanceled.Body.String(), `action="/jobs/1/retry"`) {
+		t.Fatalf("実行中の取り消し: %d %+v", runningCanceled.Code, jobs.jobs["ep"])
 	}
 }
 

internal/web/store_test.go

@@ -124,7 +124,7 @@ func (f *fakeJobStore) Cancel(_ context.Context, id int64) error {
 		return f.err
 	}
 	for ep, job := range f.jobs {
-		if job.ID == id && job.Status == "queued" {
+		if job.ID == id && (job.Status == "queued" || job.Status == "running") {
 			job.Status = "canceled"
 			f.jobs[ep] = job
 			return nil

internal/web/templates/queue.html

@@ -38,6 +38,9 @@
       {{if .SpeedText}}／{{.SpeedText}}{{end}}
       {{if .ETAText}}／{{.ETAText}}{{end}}
       {{if .FragmentText}}／{{.FragmentText}}{{end}}</p>
+    <form action="/jobs/{{.ID}}/cancel" method="post" hx-post="/jobs/{{.ID}}/cancel" hx-target="#queue-list" hx-swap="outerHTML">
+      <button type="submit">取り消し</button>
+    </form>
   {{end}}
   {{if eq .Status "failed"}}
     <p>失敗理由: {{.ErrorKindText}}</p>

internal/worker/progress.go

@@ -0,0 +1,81 @@
+package worker
+
+import (
+	"encoding/json"
+	"strings"
+
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+)
+
+type outputEvent struct {
+	Event    string `json:"event"`
+	FormatID string `json:"format_id"`
+	Filepath string `json:"filepath"`
+	Progress struct {
+		DownloadedBytes    *float64 `json:"downloaded_bytes"`
+		TotalBytes         *float64 `json:"total_bytes"`
+		TotalBytesEstimate *float64 `json:"total_bytes_estimate"`
+		Speed              *float64 `json:"speed"`
+		ETA                *float64 `json:"eta"`
+		FragmentIndex      *float64 `json:"fragment_index"`
+		FragmentCount      *float64 `json:"fragment_count"`
+	} `json:"progress"`
+}
+
+func parseProgress(line string) (store.Progress, bool) {
+	var event outputEvent
+	if err := json.Unmarshal([]byte(line), &event); err != nil {
+		return store.Progress{}, false
+	}
+	phase := ""
+	switch event.Event {
+	case "download":
+		if strings.Contains(strings.ToLower(event.FormatID), "audio") {
+			phase = "audio"
+		} else {
+			phase = "video"
+		}
+	case "postprocess":
+		phase = "merging"
+	default:
+		return store.Progress{}, false
+	}
+	total := int64Pointer(event.Progress.TotalBytes)
+	estimate := false
+	if total == nil && event.Progress.TotalBytesEstimate != nil {
+		total = int64Pointer(event.Progress.TotalBytesEstimate)
+		estimate = true
+	}
+	var formatID *string
+	if event.FormatID != "" {
+		formatID = &event.FormatID
+	}
+	return store.Progress{Phase: phase, DownloadedBytes: int64Pointer(event.Progress.DownloadedBytes),
+		TotalBytes: total, TotalIsEstimate: estimate, SpeedBPS: event.Progress.Speed,
+		ETASeconds: event.Progress.ETA, FragmentIndex: int32Pointer(event.Progress.FragmentIndex),
+		FragmentCount: int32Pointer(event.Progress.FragmentCount), FormatID: formatID}, true
+}
+
+func int64Pointer(value *float64) *int64 {
+	if value == nil {
+		return nil
+	}
+	result := int64(*value)
+	return &result
+}
+
+func int32Pointer(value *float64) *int32 {
+	if value == nil {
+		return nil
+	}
+	result := int32(*value)
+	return &result
+}
+
+func parseAfterMove(line string) (string, bool) {
+	var event outputEvent
+	if err := json.Unmarshal([]byte(line), &event); err != nil || event.Event != "after_move" || event.Filepath == "" {
+		return "", false
+	}
+	return event.Filepath, true
+}

internal/worker/progress_test.go

@@ -0,0 +1,27 @@
+package worker
+
+import "testing"
+
+func TestParseProgress(t *testing.T) {
+	video, ok := parseProgress(`{"event":"download","format_id":"hls-698-1","progress":{"downloaded_bytes":25,"total_bytes_estimate":100.0,"speed":10.5,"eta":3,"fragment_index":1,"fragment_count":4}}`)
+	if !ok || video.Phase != "video" || video.DownloadedBytes == nil || *video.DownloadedBytes != 25 || video.TotalBytes == nil || *video.TotalBytes != 100 || !video.TotalIsEstimate || video.SpeedBPS == nil || *video.SpeedBPS != 10.5 || video.ETASeconds == nil || *video.ETASeconds != 3 || video.FragmentIndex == nil || *video.FragmentIndex != 1 || video.FragmentCount == nil || *video.FragmentCount != 4 || video.FormatID == nil || *video.FormatID != "hls-698-1" {
+		t.Fatalf("映像進捗: %+v ok=%v", video, ok)
+	}
+	audio, ok := parseProgress(`{"event":"download","format_id":"hls-ts_AUDIO-0_2","progress":{"downloaded_bytes":50,"total_bytes":50}}`)
+	if !ok || audio.Phase != "audio" || audio.TotalIsEstimate || audio.TotalBytes == nil || *audio.TotalBytes != 50 {
+		t.Fatalf("音声進捗: %+v ok=%v", audio, ok)
+	}
+	merging, ok := parseProgress(`{"event":"postprocess","progress":{"status":"started","postprocessor":"Merger"}}`)
+	if !ok || merging.Phase != "merging" {
+		t.Fatalf("結合進捗: %+v ok=%v", merging, ok)
+	}
+	path, ok := parseAfterMove(`{"event":"after_move","filepath":"/videos/incoming/ep.mp4"}`)
+	if !ok || path != "/videos/incoming/ep.mp4" {
+		t.Fatalf("移動後パス: %q ok=%v", path, ok)
+	}
+	for _, invalid := range []string{"noise", `{"event":"other"}`} {
+		if _, ok := parseProgress(invalid); ok {
+			t.Errorf("不正な進捗を許した: %q", invalid)
+		}
+	}
+}

internal/worker/runner.go

@@ -0,0 +1,500 @@
+package worker
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"strconv"
+	"strings"
+	"sync"
+	"syscall"
+	"time"
+
+	"gitea.kazu634.com/kazu634/tver/internal/media"
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"github.com/jackc/pgx/v5"
+)
+
+var (
+	errCanceled      = errors.New("job canceled")
+	errShutdown      = errors.New("worker shutting down")
+	errInvalidOutput = errors.New("invalid output")
+	headerSecretRE   = regexp.MustCompile(`(?im)(authorization|cookie)([=:][ \t]*)([^\r\n]+)`)
+	paramSecretRE    = regexp.MustCompile(`(?i)(token|platform_uid|platform_token)([=:][[:space:]]*)([^&[:space:]]+)`)
+)
+
+type JobStore interface {
+	ClaimNext(context.Context) (store.Job, error)
+	UpdateProgress(context.Context, int64, store.Progress) error
+	Heartbeat(context.Context, int64) error
+	GetJob(context.Context, int64) (store.Job, error)
+	Complete(context.Context, int64, string, int64) error
+	Fail(context.Context, int64, string, string) error
+	RequeueRunning(context.Context, int64) error
+}
+
+type Config struct {
+	VideoDir          string
+	YTDLPPath         string
+	FFprobePath       string
+	Concurrency       int
+	PollInterval      time.Duration
+	ProgressInterval  time.Duration
+	HeartbeatInterval time.Duration
+	CancelInterval    time.Duration
+}
+
+type Runner struct {
+	store  JobStore
+	config Config
+	logger *log.Logger
+}
+
+type cappedBuffer struct {
+	buffer bytes.Buffer
+	limit  int
+}
+
+func (b *cappedBuffer) Write(p []byte) (int, error) {
+	written := len(p)
+	remaining := b.limit - b.buffer.Len()
+	if remaining > 0 {
+		if remaining < len(p) {
+			p = p[:remaining]
+		}
+		_, _ = b.buffer.Write(p)
+	}
+	return written, nil
+}
+
+func (b *cappedBuffer) String() string { return b.buffer.String() }
+
+func scanLines(reader io.Reader, capture *cappedBuffer) <-chan string {
+	lines := make(chan string)
+	go func() {
+		defer close(lines)
+		scanner := bufio.NewScanner(reader)
+		scanner.Buffer(make([]byte, 4096), 1<<20)
+		for scanner.Scan() {
+			line := scanner.Text()
+			if capture != nil {
+				_, _ = capture.Write([]byte(line + "\n"))
+			}
+			lines <- line
+		}
+	}()
+	return lines
+}
+
+func New(jobs JobStore, config Config, logger *log.Logger) (*Runner, error) {
+	if config.VideoDir == "" {
+		config.VideoDir = "/videos"
+	}
+	if config.YTDLPPath == "" {
+		config.YTDLPPath = "yt-dlp"
+	}
+	if config.FFprobePath == "" {
+		config.FFprobePath = "ffprobe"
+	}
+	if config.Concurrency == 0 {
+		config.Concurrency = 1
+	}
+	if config.Concurrency < 1 || config.Concurrency > 2 {
+		return nil, errors.New("worker concurrency must be 1 or 2")
+	}
+	if config.PollInterval <= 0 {
+		config.PollInterval = time.Second
+	}
+	if config.ProgressInterval <= 0 {
+		config.ProgressInterval = time.Second
+	}
+	if config.HeartbeatInterval <= 0 {
+		config.HeartbeatInterval = 5 * time.Second
+	}
+	if config.CancelInterval <= 0 {
+		config.CancelInterval = 500 * time.Millisecond
+	}
+	if logger == nil {
+		logger = log.Default()
+	}
+	return &Runner{store: jobs, config: config, logger: logger}, nil
+}
+
+func (r *Runner) Run(ctx context.Context) error {
+	sem := make(chan struct{}, r.config.Concurrency)
+	var wg sync.WaitGroup
+	ticker := time.NewTicker(r.config.PollInterval)
+	defer ticker.Stop()
+	for {
+		for len(sem) < cap(sem) && ctx.Err() == nil {
+			job, err := r.store.ClaimNext(ctx)
+			if errors.Is(err, pgx.ErrNoRows) {
+				break
+			}
+			if err != nil {
+				r.logger.Printf("ジョブ取得に失敗: %v", err)
+				break
+			}
+			sem <- struct{}{}
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				defer func() { <-sem }()
+				r.ProcessJob(ctx, job)
+			}()
+		}
+		select {
+		case <-ctx.Done():
+			wg.Wait()
+			return nil
+		case <-ticker.C:
+		}
+	}
+}
+
+func (r *Runner) ProcessJob(ctx context.Context, job store.Job) {
+	workDir := filepath.Join(r.config.VideoDir, ".incoming", strconv.FormatInt(job.ID, 10))
+	if err := os.RemoveAll(workDir); err != nil {
+		r.fail(job.ID, "unknown", fmt.Sprintf("作業ディレクトリを初期化できません: %v", err))
+		return
+	}
+	if err := os.MkdirAll(workDir, 0o755); err != nil {
+		r.fail(job.ID, "unknown", fmt.Sprintf("作業ディレクトリを作成できません: %v", err))
+		return
+	}
+	defer os.RemoveAll(workDir)
+
+	downloaded, stderr, err := r.download(ctx, job, workDir)
+	if errors.Is(err, errShutdown) {
+		r.requeue(job.ID)
+		return
+	}
+	if errors.Is(err, errCanceled) {
+		return
+	}
+	if err != nil {
+		message := redact(stderr)
+		if message == "" {
+			message = redact(err.Error())
+		}
+		kind := classifyFailure(message)
+		if errors.Is(err, errInvalidOutput) {
+			kind = "invalid_output"
+		}
+		r.failIfRunning(job.ID, kind, message)
+		return
+	}
+	if r.canceled(job.ID) {
+		return
+	}
+	if err := r.probe(ctx, downloaded); err != nil {
+		if ctx.Err() != nil {
+			r.requeue(job.ID)
+			return
+		}
+		r.failIfRunning(job.ID, "invalid_output", redact(err.Error()))
+		return
+	}
+	if r.canceled(job.ID) {
+		return
+	}
+	if err := r.store.UpdateProgress(ctx, job.ID, store.Progress{Phase: "moving"}); err != nil {
+		if r.canceled(job.ID) {
+			return
+		}
+		r.failIfRunning(job.ID, "unknown", "移動状態を記録できません: "+err.Error())
+		return
+	}
+	relative := media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID)
+	destination := filepath.Join(r.config.VideoDir, relative)
+	if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil {
+		r.failIfRunning(job.ID, "unknown", "保存先を作成できません: "+err.Error())
+		return
+	}
+	if err := os.Rename(downloaded, destination); err != nil {
+		r.failIfRunning(job.ID, "unknown", "動画を保存先へ移動できません: "+err.Error())
+		return
+	}
+	info, err := os.Stat(destination)
+	if err != nil {
+		r.failIfRunning(job.ID, "unknown", "保存後の動画を確認できません: "+err.Error())
+		return
+	}
+	if err := r.store.Complete(ctx, job.ID, filepath.ToSlash(relative), info.Size()); err != nil {
+		if r.canceled(job.ID) {
+			if removeErr := os.Remove(destination); removeErr != nil && !errors.Is(removeErr, os.ErrNotExist) {
+				r.logger.Printf("取り消されたジョブ %d の保存済みファイル削除に失敗: %v", job.ID, removeErr)
+			}
+			return
+		}
+		r.logger.Printf("ジョブ %d の完了記録に失敗: %v", job.ID, err)
+	}
+}
+
+func (r *Runner) download(ctx context.Context, job store.Job, workDir string) (string, string, error) {
+	output := filepath.Join(workDir, "%(id)s.%(ext)s")
+	args := []string{"--no-playlist", "--no-simulate", "--quiet", "--progress", "--newline",
+		"--continue", "--retries", "0", "--fragment-retries", "0", "--extractor-retries", "0",
+		"-f", "bv*[vcodec^=avc1]+ba/b", "--merge-output-format", "mp4",
+		"--progress-template", `download:{"event":"download","format_id":%(info.format_id)j,"progress":%(progress)j}`,
+		"--progress-template", `postprocess:{"event":"postprocess","progress":%(progress)j}`,
+		"--print", `after_move:{"event":"after_move","filepath":%(filepath)j}`, "-o", output,
+		"https://tver.jp/episodes/" + job.EpisodeID}
+	cmd := exec.Command(r.config.YTDLPPath, args...)
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return "", "", err
+	}
+	stderrPipe, err := cmd.StderrPipe()
+	if err != nil {
+		return "", "", err
+	}
+	if err := cmd.Start(); err != nil {
+		return "", "", err
+	}
+	stderr := cappedBuffer{limit: 64 << 10}
+	stdoutLines := scanLines(stdout, nil)
+	stderrLines := scanLines(stderrPipe, &stderr)
+	waited := make(chan error, 1)
+	go func() { waited <- cmd.Wait() }()
+	heartbeat := time.NewTicker(r.config.HeartbeatInterval)
+	cancelCheck := time.NewTicker(r.config.CancelInterval)
+	defer heartbeat.Stop()
+	defer cancelCheck.Stop()
+	_ = r.store.Heartbeat(ctx, job.ID)
+	var afterMove string
+	var pending *store.Progress
+	nextProgress := time.Time{}
+	lastPhase := ""
+	recordProgress := func(progress store.Progress) {
+		pending = &progress
+		now := time.Now()
+		if progress.Phase != lastPhase || now.After(nextProgress) {
+			_ = r.store.UpdateProgress(ctx, job.ID, progress)
+			nextProgress = now.Add(r.config.ProgressInterval)
+			lastPhase = progress.Phase
+			pending = nil
+		}
+	}
+	stopping := ""
+	var forceKill <-chan time.Time
+	ctxDone := ctx.Done()
+	waitComplete := false
+	var waitErr error
+	for {
+		if waitComplete && stdoutLines == nil && stderrLines == nil {
+			if pending != nil && stopping == "" {
+				updateCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				_ = r.store.UpdateProgress(updateCtx, job.ID, *pending)
+				cancel()
+			}
+			if stopping == "canceled" {
+				return "", stderr.String(), errCanceled
+			}
+			if stopping == "shutdown" {
+				return "", stderr.String(), errShutdown
+			}
+			if waitErr != nil {
+				r.logger.Printf("yt-dlp ジョブ %d 失敗: %s", job.ID, redact(stderr.String()))
+				return "", stderr.String(), waitErr
+			}
+			path, err := pathInside(workDir, afterMove)
+			if err != nil {
+				return "", stderr.String(), err
+			}
+			return path, stderr.String(), nil
+		}
+		select {
+		case line, ok := <-stdoutLines:
+			if !ok {
+				stdoutLines = nil
+				continue
+			}
+			if path, ok := parseAfterMove(line); ok {
+				afterMove = path
+				continue
+			}
+			if progress, ok := parseProgress(line); ok {
+				recordProgress(progress)
+			}
+		case line, ok := <-stderrLines:
+			if !ok {
+				stderrLines = nil
+				continue
+			}
+			if progress, ok := parseProgress(line); ok {
+				recordProgress(progress)
+			}
+		case <-heartbeat.C:
+			_ = r.store.Heartbeat(ctx, job.ID)
+		case <-cancelCheck.C:
+			if r.canceled(job.ID) && stopping == "" {
+				stopping = "canceled"
+				if !waitComplete {
+					stopProcessGroup(cmd)
+					forceKill = time.After(2 * time.Second)
+				}
+			}
+		case <-ctxDone:
+			if stopping == "" {
+				stopping = "shutdown"
+				if !waitComplete {
+					stopProcessGroup(cmd)
+					forceKill = time.After(2 * time.Second)
+				}
+			}
+			ctxDone = nil
+		case <-forceKill:
+			_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+			forceKill = nil
+		case waitErr = <-waited:
+			waitComplete = true
+			waited = nil
+			forceKill = nil
+		}
+	}
+}
+
+func stopProcessGroup(cmd *exec.Cmd) {
+	if cmd.Process == nil {
+		return
+	}
+	_ = syscall.Kill(-cmd.Process.Pid, syscall.SIGTERM)
+}
+
+func pathInside(root, candidate string) (string, error) {
+	if candidate == "" {
+		return "", errors.New("yt-dlp did not report after_move filepath")
+	}
+	rootAbs, err := filepath.Abs(root)
+	if err != nil {
+		return "", err
+	}
+	candidateAbs, err := filepath.Abs(candidate)
+	if err != nil {
+		return "", err
+	}
+	rel, err := filepath.Rel(rootAbs, candidateAbs)
+	if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("%w: yt-dlp filepath leaves working directory", errInvalidOutput)
+	}
+	info, err := os.Lstat(candidateAbs)
+	if err != nil {
+		return "", fmt.Errorf("%w: yt-dlp output file is missing: %v", errInvalidOutput, err)
+	}
+	if !info.Mode().IsRegular() {
+		return "", fmt.Errorf("%w: yt-dlp output is not a regular file", errInvalidOutput)
+	}
+	rootReal, err := filepath.EvalSymlinks(rootAbs)
+	if err != nil {
+		return "", fmt.Errorf("%w: resolve working directory: %v", errInvalidOutput, err)
+	}
+	candidateReal, err := filepath.EvalSymlinks(candidateAbs)
+	if err != nil {
+		return "", fmt.Errorf("%w: resolve yt-dlp output: %v", errInvalidOutput, err)
+	}
+	realRelative, err := filepath.Rel(rootReal, candidateReal)
+	if err != nil || realRelative == ".." || strings.HasPrefix(realRelative, ".."+string(filepath.Separator)) {
+		return "", fmt.Errorf("%w: yt-dlp output resolves outside working directory", errInvalidOutput)
+	}
+	return candidateAbs, nil
+}
+
+func (r *Runner) probe(ctx context.Context, path string) error {
+	cmd := exec.CommandContext(ctx, r.config.FFprobePath, "-v", "error", "-show_entries",
+		"format=format_name", "-show_entries", "stream=codec_type,codec_name", "-of", "json", path)
+	output, err := cmd.Output()
+	if err != nil {
+		return fmt.Errorf("ffprobe failed: %w", err)
+	}
+	var result struct {
+		Format struct {
+			FormatName string `json:"format_name"`
+		} `json:"format"`
+		Streams []struct {
+			CodecType string `json:"codec_type"`
+			CodecName string `json:"codec_name"`
+		} `json:"streams"`
+	}
+	if err := json.Unmarshal(output, &result); err != nil {
+		return fmt.Errorf("ffprobe output is not JSON: %w", err)
+	}
+	video, audio := false, false
+	for _, stream := range result.Streams {
+		video = video || stream.CodecType == "video" && stream.CodecName == "h264"
+		audio = audio || stream.CodecType == "audio" && stream.CodecName == "aac"
+	}
+	if !strings.Contains(result.Format.FormatName, "mp4") || !video || !audio {
+		return fmt.Errorf("expected MP4 with H.264 and AAC, got format=%s video=%t audio=%t", result.Format.FormatName, video, audio)
+	}
+	return nil
+}
+
+func (r *Runner) canceled(id int64) bool {
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	job, err := r.store.GetJob(ctx, id)
+	return err == nil && job.Status == "canceled"
+}
+
+func (r *Runner) failIfRunning(id int64, kind, message string) {
+	if r.canceled(id) {
+		return
+	}
+	r.fail(id, kind, message)
+}
+
+func (r *Runner) fail(id int64, kind, message string) {
+	if len(message) > 8192 {
+		message = message[:8192]
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := r.store.Fail(ctx, id, kind, message); err != nil {
+		r.logger.Printf("ジョブ %d の失敗記録に失敗: %v", id, err)
+	}
+}
+
+func (r *Runner) requeue(id int64) {
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+	if err := r.store.RequeueRunning(ctx, id); err != nil && !errors.Is(err, store.ErrInvalidTransition) {
+		r.logger.Printf("ジョブ %d の待機中への復帰に失敗: %v", id, err)
+	}
+}
+
+func classifyFailure(message string) string {
+	lower := strings.ToLower(message)
+	for _, test := range []struct {
+		kind     string
+		patterns []string
+	}{
+		{"drm", []string{"drm", "digital rights management", "protected by"}},
+		{"geo", []string{"geo-restricted", "not available in your country", "not available from your location", "geo restriction", "region restriction", "geographic restriction"}},
+		{"network", []string{"network", "connection", "timed out", "temporary failure", "unable to download", "proxy error", "http error 5"}},
+		{"unavailable", []string{"media_not_found", "video unavailable", "not available", "no longer available", "http error 404", "episode does not exist"}},
+		{"extract", []string{"extractor error", "unable to extract", "unsupported url", "no video formats", "after_move", "output file is missing"}},
+	} {
+		for _, pattern := range test.patterns {
+			if strings.Contains(lower, pattern) {
+				return test.kind
+			}
+		}
+	}
+	return "unknown"
+}
+
+func redact(message string) string {
+	message = headerSecretRE.ReplaceAllString(message, "$1$2[REDACTED]")
+	return strings.TrimSpace(paramSecretRE.ReplaceAllString(message, "$1$2[REDACTED]"))
+}

internal/worker/runner_test.go

@@ -0,0 +1,643 @@
+package worker
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"gitea.kazu634.com/kazu634/tver/internal/media"
+	"gitea.kazu634.com/kazu634/tver/internal/store"
+	"github.com/jackc/pgx/v5"
+)
+
+type fakeStore struct {
+	mu                        sync.Mutex
+	job                       store.Job
+	progress                  []store.Progress
+	heartbeats                int
+	completed                 bool
+	completePath              string
+	completeSize              int64
+	failedKind, failedMessage string
+	requeued                  bool
+	claimed                   bool
+}
+
+type queueStore struct {
+	mu          sync.Mutex
+	jobs        []store.Job
+	claimCounts map[int64]int
+	active      int
+	maxActive   int
+}
+
+func (f *queueStore) ClaimNext(context.Context) (store.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for i := range f.jobs {
+		if f.jobs[i].Status != "queued" {
+			continue
+		}
+		f.jobs[i].Status = "running"
+		f.claimCounts[f.jobs[i].ID]++
+		f.active++
+		if f.active > f.maxActive {
+			f.maxActive = f.active
+		}
+		return f.jobs[i], nil
+	}
+	return store.Job{}, pgx.ErrNoRows
+}
+
+func (f *queueStore) UpdateProgress(_ context.Context, id int64, _ store.Progress) error {
+	return f.requireRunning(id)
+}
+
+func (f *queueStore) Heartbeat(_ context.Context, id int64) error {
+	return f.requireRunning(id)
+}
+
+func (f *queueStore) requireRunning(id int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for i := range f.jobs {
+		if f.jobs[i].ID == id && f.jobs[i].Status == "running" {
+			return nil
+		}
+	}
+	return store.ErrInvalidTransition
+}
+
+func (f *queueStore) GetJob(_ context.Context, id int64) (store.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for _, job := range f.jobs {
+		if job.ID == id {
+			return job, nil
+		}
+	}
+	return store.Job{}, pgx.ErrNoRows
+}
+
+func (f *queueStore) Complete(_ context.Context, id int64, _ string, _ int64) error {
+	return f.finish(id, "succeeded")
+}
+
+func (f *queueStore) Fail(_ context.Context, id int64, _, _ string) error {
+	return f.finish(id, "failed")
+}
+
+func (f *queueStore) RequeueRunning(_ context.Context, id int64) error {
+	return f.finish(id, "queued")
+}
+
+func (f *queueStore) finish(id int64, status string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for i := range f.jobs {
+		if f.jobs[i].ID == id && f.jobs[i].Status == "running" {
+			f.jobs[i].Status = status
+			f.active--
+			return nil
+		}
+	}
+	return store.ErrInvalidTransition
+}
+
+func (f *queueStore) state() ([]store.Job, map[int64]int, int) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	jobs := append([]store.Job(nil), f.jobs...)
+	counts := make(map[int64]int, len(f.claimCounts))
+	for id, count := range f.claimCounts {
+		counts[id] = count
+	}
+	return jobs, counts, f.maxActive
+}
+
+func (f *fakeStore) ClaimNext(context.Context) (store.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.claimed || f.job.Status != "queued" {
+		return store.Job{}, pgx.ErrNoRows
+	}
+	f.claimed = true
+	f.job.Status = "running"
+	return f.job, nil
+}
+
+func (f *fakeStore) UpdateProgress(_ context.Context, _ int64, p store.Progress) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.job.Status != "running" {
+		return store.ErrInvalidTransition
+	}
+	f.progress = append(f.progress, p)
+	return nil
+}
+
+func (f *fakeStore) Heartbeat(context.Context, int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.job.Status != "running" {
+		return store.ErrInvalidTransition
+	}
+	f.heartbeats++
+	return nil
+}
+
+func (f *fakeStore) GetJob(context.Context, int64) (store.Job, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.job, nil
+}
+
+func (f *fakeStore) Complete(_ context.Context, _ int64, path string, size int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.job.Status != "running" {
+		return store.ErrInvalidTransition
+	}
+	f.job.Status = "succeeded"
+	f.completed, f.completePath, f.completeSize = true, path, size
+	return nil
+}
+
+func (f *fakeStore) Fail(_ context.Context, _ int64, kind, message string) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.job.Status != "running" {
+		return store.ErrInvalidTransition
+	}
+	f.job.Status = "failed"
+	f.failedKind, f.failedMessage = kind, message
+	return nil
+}
+
+func (f *fakeStore) RequeueRunning(context.Context, int64) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	if f.job.Status != "running" {
+		return store.ErrInvalidTransition
+	}
+	f.job.Status = "queued"
+	f.requeued = true
+	return nil
+}
+
+func (f *fakeStore) setStatus(status string) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.job.Status = status
+}
+
+func (f *fakeStore) heartbeatCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.heartbeats
+}
+
+func testRunner(t *testing.T, jobs *fakeStore, videoDir string) *Runner {
+	t.Helper()
+	ytdlp := filepath.Join("testdata", "fake-yt-dlp.sh")
+	ffprobe := filepath.Join("testdata", "fake-ffprobe.sh")
+	for _, path := range []string{ytdlp, ffprobe} {
+		info, err := os.Stat(path)
+		if err != nil || info.Mode().Perm()&0o111 == 0 {
+			t.Fatalf("偽コマンドが実行可能ではありません: %s: %v", path, err)
+		}
+	}
+	runner, err := New(jobs, Config{VideoDir: videoDir, YTDLPPath: ytdlp, FFprobePath: ffprobe,
+		Concurrency: 1, PollInterval: 5 * time.Millisecond, ProgressInterval: time.Millisecond,
+		HeartbeatInterval: 10 * time.Millisecond, CancelInterval: 5 * time.Millisecond}, log.New(io.Discard, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return runner
+}
+
+func runningJob() store.Job {
+	return store.Job{ID: 7, EpisodeID: "ep123", SeriesTitle: "番組名", Title: "第1話", Status: "running"}
+}
+
+func TestProcessJobDownloadsProbesMovesAndCleans(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "success")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	argsFile := filepath.Join(dir, "args")
+	t.Setenv("FAKE_ARGS_FILE", argsFile)
+	jobs := &fakeStore{job: runningJob()}
+	testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
+	want := filepath.ToSlash(media.RelativeVideoPath("番組名", "第1話", "ep123"))
+	if !jobs.completed || jobs.completePath != want || jobs.completeSize == 0 {
+		t.Fatalf("complete: done=%v path=%q size=%d", jobs.completed, jobs.completePath, jobs.completeSize)
+	}
+	if _, err := os.Stat(filepath.Join(dir, filepath.FromSlash(want))); err != nil {
+		t.Fatalf("保存先: %v", err)
+	}
+	if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("作業ディレクトリが残った: %v", err)
+	}
+	args, err := os.ReadFile(argsFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	argumentText := string(args)
+	for _, want := range []string{"--newline", "--no-playlist", "--no-simulate", "--quiet", "--progress", "--continue", "--merge-output-format", "bv*[vcodec^=avc1]+ba/b", `download:{"event":"download","format_id":%(info.format_id)j,"progress":%(progress)j}`, `postprocess:{"event":"postprocess","progress":%(progress)j}`, `after_move:{"event":"after_move","filepath":%(filepath)j}`, filepath.Join(dir, ".incoming", "7", "%(id)s.%(ext)s")} {
+		if !strings.Contains(argumentText, want) {
+			t.Errorf("yt-dlp 引数に %q がない: %s", want, argumentText)
+		}
+	}
+	if strings.Contains(argumentText, "番組名") || strings.Contains(argumentText, "第1話") {
+		t.Errorf("出力テンプレートに番組情報が入った: %s", argumentText)
+	}
+	phases := make(map[string]bool)
+	for _, progress := range jobs.progress {
+		phases[progress.Phase] = true
+	}
+	for _, phase := range []string{"video", "audio", "merging", "moving"} {
+		if !phases[phase] {
+			t.Errorf("進捗段階 %s がない: %+v", phase, jobs.progress)
+		}
+	}
+	if jobs.heartbeats < 2 {
+		t.Errorf("heartbeat が少ない: %d", jobs.heartbeats)
+	}
+}
+
+func TestProcessJobKeepsPercentNamesOutOfYTDLPTemplate(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "success")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	argsFile := filepath.Join(dir, "args")
+	t.Setenv("FAKE_ARGS_FILE", argsFile)
+	job := runningJob()
+	job.SeriesTitle = "100%番組"
+	job.Title = "第%1話"
+	jobs := &fakeStore{job: job}
+	testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
+	if !jobs.completed {
+		t.Fatalf("%% を含む番組を完了できなかった: %+v", jobs)
+	}
+	args, err := os.ReadFile(argsFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(args), job.SeriesTitle) || strings.Contains(string(args), job.Title) {
+		t.Fatalf("番組情報が yt-dlp 引数に入った: %s", args)
+	}
+	want := filepath.ToSlash(media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID))
+	if jobs.completePath != want {
+		t.Fatalf("保存先=%q want=%q", jobs.completePath, want)
+	}
+}
+
+func TestProcessJobReplacesExistingDestination(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "success")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	job := runningJob()
+	relative := media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID)
+	destination := filepath.Join(dir, relative)
+	if err := os.MkdirAll(filepath.Dir(destination), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(destination, []byte("stale file from previous attempt"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	jobs := &fakeStore{job: job}
+	testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
+	got, err := os.ReadFile(destination)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !jobs.completed || string(got) != "fake mp4" || jobs.completeSize != int64(len(got)) {
+		t.Fatalf("既存ファイルの置換: completed=%v content=%q size=%d", jobs.completed, got, jobs.completeSize)
+	}
+}
+
+func TestProcessJobRejectsAfterMoveSymlinkOutsideWorkDirectory(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "symlink")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	outside := filepath.Join(t.TempDir(), "outside.mp4")
+	if err := os.WriteFile(outside, []byte("outside video"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	t.Setenv("FAKE_OUTSIDE_FILE", outside)
+	jobs := &fakeStore{job: runningJob()}
+	testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
+	if jobs.completed || jobs.failedKind != "invalid_output" {
+		t.Fatalf("作業ディレクトリ外を指すシンボリックリンクを受理した: %+v", jobs)
+	}
+	if got, err := os.ReadFile(outside); err != nil || string(got) != "outside video" {
+		t.Fatalf("作業ディレクトリ外のファイルを変更した: %q %v", got, err)
+	}
+}
+
+func TestPathInsideSymlinkDerivedCases(t *testing.T) {
+	root := t.TempDir()
+	insideDir := filepath.Join(root, "real")
+	if err := os.Mkdir(insideDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	insideFile := filepath.Join(insideDir, "video.mp4")
+	if err := os.WriteFile(insideFile, []byte("video"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	outsideDir := t.TempDir()
+	outsideFile := filepath.Join(outsideDir, "outside.mp4")
+	if err := os.WriteFile(outsideFile, []byte("outside"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	internalFileLink := filepath.Join(root, "internal-file-link.mp4")
+	internalDirLink := filepath.Join(root, "internal-dir-link")
+	externalDirLink := filepath.Join(root, "external-dir-link")
+	rootLink := filepath.Join(t.TempDir(), "root-link")
+	for link, target := range map[string]string{
+		internalFileLink: insideFile,
+		internalDirLink:  insideDir,
+		externalDirLink:  outsideDir,
+		rootLink:         root,
+	} {
+		if err := os.Symlink(target, link); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	for _, tt := range []struct {
+		name, testRoot, candidate string
+		wantOK                    bool
+	}{
+		{"通常ファイル", root, insideFile, true},
+		{"作業ディレクトリ内を指すファイルリンク", root, internalFileLink, false},
+		{"after_moveがディレクトリ", root, insideDir, false},
+		{"作業ディレクトリ内を指すディレクトリリンク配下", root, filepath.Join(internalDirLink, "video.mp4"), true},
+		{"作業ディレクトリ外を指すディレクトリリンク配下", root, filepath.Join(externalDirLink, "outside.mp4"), false},
+		{"作業ディレクトリ自体がリンク", rootLink, filepath.Join(rootLink, "real", "video.mp4"), true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := pathInside(tt.testRoot, tt.candidate)
+			if tt.wantOK {
+				want, absErr := filepath.Abs(tt.candidate)
+				if absErr != nil {
+					t.Fatal(absErr)
+				}
+				if err != nil || got != want {
+					t.Fatalf("pathInside()=(%q, %v) want=(%q, nil)", got, err, want)
+				}
+				return
+			}
+			if !errors.Is(err, errInvalidOutput) || got != "" {
+				t.Fatalf("pathInside()=(%q, %v), invalid_output を期待", got, err)
+			}
+		})
+	}
+}
+
+func TestProcessJobDestinationDirectoryFailureCleansIncoming(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "success")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	job := runningJob()
+	seriesDir := filepath.Join(dir, filepath.Dir(media.RelativeVideoPath(job.SeriesTitle, job.Title, job.EpisodeID)))
+	if err := os.WriteFile(seriesDir, []byte("blocks directory"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	jobs := &fakeStore{job: job}
+	testRunner(t, jobs, dir).ProcessJob(context.Background(), job)
+	if jobs.completed || jobs.failedKind != "unknown" {
+		t.Fatalf("保存先作成失敗: %+v", jobs)
+	}
+	if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("作業ディレクトリが残った: %v", err)
+	}
+}
+
+func TestFailureClassificationAndRedaction(t *testing.T) {
+	for _, tt := range []struct{ mode, kind string }{
+		{"drm", "drm"}, {"geo", "geo"}, {"network", "network"},
+		{"unavailable", "unavailable"}, {"extract", "extract"}, {"unknown", "unknown"},
+	} {
+		t.Run(tt.mode, func(t *testing.T) {
+			t.Setenv("FAKE_YTDLP_MODE", tt.mode)
+			dir := t.TempDir()
+			jobs := &fakeStore{job: runningJob()}
+			testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
+			if jobs.failedKind != tt.kind || jobs.completed || strings.Contains(jobs.failedMessage, "secret-value") {
+				t.Fatalf("failure: kind=%q message=%q completed=%v", jobs.failedKind, jobs.failedMessage, jobs.completed)
+			}
+			if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
+				t.Fatalf("作業ディレクトリが残った: %v", err)
+			}
+		})
+	}
+}
+
+func TestInvalidProbeFailsAndCleans(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "success")
+	for _, mode := range []string{"bad-container", "bad-video", "bad-audio", "non-json", "failure"} {
+		t.Run(mode, func(t *testing.T) {
+			t.Setenv("FAKE_FFPROBE_MODE", mode)
+			dir := t.TempDir()
+			jobs := &fakeStore{job: runningJob()}
+			testRunner(t, jobs, dir).ProcessJob(context.Background(), jobs.job)
+			if jobs.failedKind != "invalid_output" || jobs.completed {
+				t.Fatalf("ffprobe 不合格: kind=%q complete=%v", jobs.failedKind, jobs.completed)
+			}
+			if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
+				t.Fatalf("作業ディレクトリが残った: %v", err)
+			}
+		})
+	}
+}
+
+func TestProgressIsThrottledButHeartbeatContinues(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "burst")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	jobs := &fakeStore{job: runningJob()}
+	runner := testRunner(t, jobs, dir)
+	runner.config.ProgressInterval = 50 * time.Millisecond
+	runner.ProcessJob(context.Background(), jobs.job)
+	downloadUpdates := 0
+	for _, progress := range jobs.progress {
+		if progress.Phase == "video" {
+			downloadUpdates++
+		}
+	}
+	if downloadUpdates > 2 || downloadUpdates == 0 {
+		t.Fatalf("進捗の間引き: %d updates (%+v)", downloadUpdates, jobs.progress)
+	}
+	if jobs.heartbeats < 5 {
+		t.Fatalf("進捗が止まった間の heartbeat: %d", jobs.heartbeats)
+	}
+}
+
+func TestRunConcurrencyTwoDoesNotDuplicateAndIsolatesFailure(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "mixed")
+	t.Setenv("FAKE_FFPROBE_MODE", "valid")
+	dir := t.TempDir()
+	jobs := &queueStore{
+		jobs: []store.Job{
+			{ID: 7, EpisodeID: "ep-ok", SeriesTitle: "成功番組", Title: "成功回", Status: "queued"},
+			{ID: 8, EpisodeID: "ep-fail", SeriesTitle: "失敗番組", Title: "失敗回", Status: "queued"},
+		},
+		claimCounts: make(map[int64]int),
+	}
+	runner, err := New(jobs, Config{VideoDir: dir,
+		YTDLPPath: filepath.Join("testdata", "fake-yt-dlp.sh"), FFprobePath: filepath.Join("testdata", "fake-ffprobe.sh"),
+		Concurrency: 2, PollInterval: 5 * time.Millisecond, ProgressInterval: time.Millisecond,
+		HeartbeatInterval: 10 * time.Millisecond, CancelInterval: 5 * time.Millisecond}, log.New(io.Discard, "", 0))
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { done <- runner.Run(ctx) }()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		state, _, _ := jobs.state()
+		if state[0].Status == "succeeded" && state[1].Status == "failed" {
+			break
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	cancel()
+	if err := <-done; err != nil {
+		t.Fatal(err)
+	}
+	state, counts, maxActive := jobs.state()
+	if state[0].Status != "succeeded" || state[1].Status != "failed" {
+		t.Fatalf("片方の失敗がもう片方へ波及した: %+v", state)
+	}
+	if counts[7] != 1 || counts[8] != 1 {
+		t.Fatalf("ジョブを二重取得した: %+v", counts)
+	}
+	if maxActive != 2 {
+		t.Fatalf("並行実行数の最大値=%d want=2", maxActive)
+	}
+	for _, id := range []int64{7, 8} {
+		if _, err := os.Stat(filepath.Join(dir, ".incoming", strconv.FormatInt(id, 10))); !errors.Is(err, os.ErrNotExist) {
+			t.Errorf("ジョブ %d の作業ディレクトリが残った: %v", id, err)
+		}
+	}
+}
+
+func TestCancelStopsProcessGroupAndLeavesCanceled(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "hang-drm")
+	dir := t.TempDir()
+	ready := filepath.Join(dir, "ready")
+	terminated := filepath.Join(dir, "terminated")
+	t.Setenv("FAKE_READY_FILE", ready)
+	t.Setenv("FAKE_TERM_FILE", terminated)
+	jobs := &fakeStore{job: runningJob()}
+	runner := testRunner(t, jobs, dir)
+	job := jobs.job
+	done := make(chan struct{})
+	go func() { runner.ProcessJob(context.Background(), job); close(done) }()
+	waitForFile(t, ready)
+	waitForHeartbeatCount(t, jobs, 2)
+	jobs.setStatus("canceled")
+	waitDone(t, done)
+	if jobs.failedKind != "" || jobs.requeued || jobs.completed || jobs.job.Status != "canceled" {
+		t.Fatalf("取り消し後: %+v", jobs)
+	}
+	waitForFile(t, terminated)
+	if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("作業ディレクトリが残った: %v", err)
+	}
+}
+
+func TestShutdownStopsProcessAndRequeues(t *testing.T) {
+	t.Setenv("FAKE_YTDLP_MODE", "hang")
+	dir := t.TempDir()
+	ready := filepath.Join(dir, "ready")
+	t.Setenv("FAKE_READY_FILE", ready)
+	jobs := &fakeStore{job: runningJob()}
+	ctx, cancel := context.WithCancel(context.Background())
+	runner := testRunner(t, jobs, dir)
+	job := jobs.job
+	done := make(chan struct{})
+	go func() { runner.ProcessJob(ctx, job); close(done) }()
+	waitForFile(t, ready)
+	cancel()
+	waitDone(t, done)
+	if !jobs.requeued || jobs.job.Status != "queued" || jobs.failedKind != "" {
+		t.Fatalf("終了時の復帰: %+v", jobs)
+	}
+	if _, err := os.Stat(filepath.Join(dir, ".incoming", "7")); !errors.Is(err, os.ErrNotExist) {
+		t.Fatalf("作業ディレクトリが残った: %v", err)
+	}
+}
+
+func TestRunnerRejectsConcurrencyOutsideRange(t *testing.T) {
+	for _, concurrency := range []int{-1, 3} {
+		if _, err := New(&fakeStore{}, Config{Concurrency: concurrency}, nil); err == nil {
+			t.Errorf("concurrency=%d を許した", concurrency)
+		}
+	}
+}
+
+func TestFailureClassificationOrderAndRedaction(t *testing.T) {
+	for _, tt := range []struct{ message, want string }{
+		{"DRM geo-restricted connection video unavailable unable to extract", "drm"},
+		{"geo-restricted connection video unavailable unable to extract", "geo"},
+		{"connection video unavailable unable to extract", "network"},
+		{"video unavailable unable to extract", "unavailable"},
+		{"unable to extract", "extract"},
+		{"different error", "unknown"},
+	} {
+		if got := classifyFailure(tt.message); got != tt.want {
+			t.Errorf("classifyFailure(%q)=%q want=%q", tt.message, got, tt.want)
+		}
+	}
+	redacted := redact("Authorization: Bearer bearer-secret\nCookie: session=session-secret\ntoken=token-secret platform_uid=uid-secret")
+	for _, secret := range []string{"bearer-secret", "session-secret", "token-secret", "uid-secret"} {
+		if strings.Contains(redacted, secret) {
+			t.Errorf("秘密が残った: %q", redacted)
+		}
+	}
+}
+
+func waitForFile(t *testing.T, path string) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, err := os.Stat(path); err == nil {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("ファイル待ちがタイムアウト: %s", path)
+}
+
+func waitDone(t *testing.T, done <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("worker の終了待ちがタイムアウト")
+	}
+}
+
+func waitForHeartbeatCount(t *testing.T, jobs *fakeStore, want int) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if jobs.heartbeatCount() >= want {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+	t.Fatalf("heartbeat 待ちがタイムアウト: got=%d want=%d", jobs.heartbeatCount(), want)
+}

internal/worker/testdata/fake-ffprobe.sh

@@ -0,0 +1,17 @@
+#!/bin/sh
+set -eu
+case "${FAKE_FFPROBE_MODE:-valid}" in
+  invalid|bad-container)
+    echo '{"format":{"format_name":"matroska"},"streams":[{"codec_type":"video","codec_name":"h264"},{"codec_type":"audio","codec_name":"aac"}]}' ;;
+  bad-video)
+    echo '{"format":{"format_name":"mp4"},"streams":[{"codec_type":"video","codec_name":"vp9"},{"codec_type":"audio","codec_name":"aac"}]}' ;;
+  bad-audio)
+    echo '{"format":{"format_name":"mp4"},"streams":[{"codec_type":"video","codec_name":"h264"},{"codec_type":"audio","codec_name":"opus"}]}' ;;
+  non-json)
+    echo 'not JSON' ;;
+  failure)
+    echo 'probe failed' >&2
+    exit 1 ;;
+  *)
+    echo '{"format":{"format_name":"mov,mp4,m4a,3gp,3g2,mj2"},"streams":[{"codec_type":"video","codec_name":"h264"},{"codec_type":"audio","codec_name":"aac"}]}' ;;
+esac

internal/worker/testdata/fake-yt-dlp.sh

@@ -0,0 +1,67 @@
+#!/bin/sh
+set -eu
+
+[ -z "${FAKE_ARGS_FILE:-}" ] || printf '%s\n' "$@" > "$FAKE_ARGS_FILE"
+
+output_template=
+episode_url=
+while [ "$#" -gt 0 ]; do
+  if [ "$1" = "-o" ]; then
+    output_template=$2
+    shift 2
+    continue
+  fi
+  case "$1" in
+    https://tver.jp/episodes/*) episode_url=$1 ;;
+  esac
+  shift
+done
+
+mode=${FAKE_YTDLP_MODE:-success}
+if [ "$mode" = "mixed" ]; then
+  case "$episode_url" in
+    */ep-fail) mode=unknown ;;
+    *) mode=success ;;
+  esac
+fi
+case "$mode" in
+  drm) echo 'ERROR: This video is DRM protected token=secret-value' >&2; exit 1 ;;
+  geo) echo 'ERROR: This video is not available from your location due to geo restriction' >&2; exit 1 ;;
+  network) echo 'ERROR: Unable to download JSON metadata: [Errno 61] Connection refused' >&2; exit 1 ;;
+  unavailable) echo 'ERROR: MEDIA_NOT_FOUND: episode does not exist' >&2; exit 1 ;;
+  extract) echo 'ERROR: Unable to extract player data' >&2; exit 1 ;;
+  unknown) echo 'ERROR: surprising failure' >&2; exit 1 ;;
+  hang|hang-drm)
+    [ -z "${FAKE_READY_FILE:-}" ] || : > "$FAKE_READY_FILE"
+    if [ "$mode" = "hang-drm" ]; then
+      trap 'echo "ERROR: DRM protected" >&2; [ -z "${FAKE_TERM_FILE:-}" ] || : > "$FAKE_TERM_FILE"; exit 143' TERM INT
+    else
+      trap '[ -z "${FAKE_TERM_FILE:-}" ] || : > "$FAKE_TERM_FILE"; exit 143' TERM INT
+    fi
+    while :; do sleep 1; done
+    ;;
+esac
+
+output=$(printf '%s' "$output_template" | sed 's/%(id)s/fake/g; s/%(ext)s/mp4/g')
+mkdir -p "$(dirname "$output")"
+printf 'fake mp4' > "$output"
+if [ "$mode" = "symlink" ]; then
+  rm "$output"
+  ln -s "$FAKE_OUTSIDE_FILE" "$output"
+fi
+
+if [ "$mode" = "burst" ]; then
+  i=1
+  while [ "$i" -le 20 ]; do
+    printf '{"event":"download","format_id":"hls-698-1","progress":{"status":"downloading","downloaded_bytes":%s,"total_bytes":100,"speed":10,"eta":2,"fragment_index":%s,"fragment_count":20}}\n' "$i" "$i"
+    i=$((i + 1))
+  done
+  sleep 0.12
+else
+  echo '{"event":"download","format_id":"hls-698-1","progress":{"status":"downloading","downloaded_bytes":1024,"fragment_index":0,"fragment_count":50,"filename":"/private/tmp/tver-spike/epq432nee4.fhls-698-1.mp4","tmpfilename":"/private/tmp/tver-spike/epq432nee4.fhls-698-1.mp4.part","max_progress":null,"progress_idx":null,"elapsed":0,"total_bytes_estimate":29798400.0,"speed":4834.271158821972,"eta":null,"_eta_str":"Unknown","_speed_str":"   4.72KiB/s","_percent":0.003436426116838488,"_percent_str":"  0.0%","_total_bytes_str":"       N/A","_total_bytes_estimate_str":"  28.42MiB","_downloaded_bytes_str":"   1.00KiB","_elapsed_str":"00:00:00","_default_template":"  0.0% of ~  28.42MiB at    4.72KiB/s ETA Unknown (frag 0/50)"}}'
+  sleep 0.02
+  echo '{"event":"download","format_id":"hls-ts_AUDIO-0_2","progress":{"status":"finished","downloaded_bytes":50,"total_bytes":50,"speed":20,"eta":0}}'
+  sleep 0.02
+  echo '{"event":"postprocess","progress":{"status":"started","postprocessor":"Merger"}}' >&2
+fi
+printf '{"event":"after_move","filepath":"%s"}\n' "$output"