fix(ui): make download file deletion explicit in task dialog (#1622) (#1638)

Replace server startup panics with returned errors, print actionable CLI
messages with exit status 1, and clarify the API token flags.

Add regressions for missing web authentication credentials.

Fixes GopeedLab/gopeed#1614

.github/workflows/test.yml

@@ -86,6 +86,32 @@ jobs:
           flutter pub get
           flutter gen-l10n
           flutter test test/api/downloader_config_test.dart
+  test-flutter-task-context-menu:
+    name: Test task context menus and text inputs
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    env:
+      CHROME_EXECUTABLE: /usr/bin/google-chrome
+    steps:
+      - uses: actions/checkout@v3
+      - uses: subosito/flutter-action@v2
+        with:
+          flutter-version: ${{ env.FLUTTER_VERSION }}
+      - name: Prepare Flutter tests
+        working-directory: ui/flutter
+        run: |
+          flutter pub get
+          flutter gen-l10n
+      - name: Test task menu actions and native input policy
+        working-directory: ui/flutter
+        run: |
+          flutter test --no-pub test/features/tasks/presentation/widgets/task_context_menu_test.dart test/shared/widgets/app_text_field_test.dart
+          flutter test --no-pub test/widget_test.dart --plain-name 'task context menu exposes'
+      - name: Test browser menu cancellation and input policy
+        working-directory: ui/flutter
+        run: |
+          "$CHROME_EXECUTABLE" --version
+          flutter test --no-pub --platform chrome test/features/tasks/presentation/widgets/task_context_menu_test.dart test/shared/widgets/app_text_field_test.dart
   test-webview-integration:
     name: Test webview integration (${{ matrix.name }})
     strategy:

README.md

@@ -20,7 +20,7 @@ Beyond core download management, Gopeed offers browser integration, JavaScript e
 
 Visit ✈ [Official Website](https://gopeed.com)
 
-![Application screenshot](_docs/img/ui-concept-en.png)
+![Application screenshot](_docs/img/ui-concept-en.webp)
 
 ## ✨ Features
 

README_ja-JP.md

@@ -20,7 +20,7 @@ Gopeed（**Go Speed** の略）は、Go と Flutter で開発された高速で
 
 ✈ [公式サイト](https://gopeed.com)
 
-![アプリのスクリーンショット](_docs/img/ui-concept-en.png)
+![アプリのスクリーンショット](_docs/img/ui-concept-en.webp)
 
 ## ✨ 主な機能
 

README_vi-VN.md

@@ -20,7 +20,7 @@ Ngoài việc quản lý tải xuống hằng ngày, Gopeed còn hỗ trợ tíc
 
 Truy cập ✈ [Trang web chính thức](https://gopeed.com)
 
-![Ảnh chụp màn hình ứng dụng](_docs/img/ui-concept-en.png)
+![Ảnh chụp màn hình ứng dụng](_docs/img/ui-concept-en.webp)
 
 ## ✨ Tính năng chính
 

README_zh-CN.md

@@ -20,7 +20,7 @@ Gopeed（全称 **Go Speed**，也叫“够快下载器”）是一款使用 Go
 
 访问 ✈ [官方网站](https://gopeed.com)
 
-![应用截图](_docs/img/ui-concept-zh-CN.png)
+![应用截图](_docs/img/ui-concept-zh-CN.webp)
 
 ## ✨ 主要特性
 

README_zh-TW.md

@@ -20,7 +20,7 @@ Gopeed（**Go Speed** 的縮寫）是一款以 Go 與 Flutter 開發的高速、
 
 瀏覽 ✈ [官方網站](https://gopeed.com)
 
-![應用程式截圖](_docs/img/ui-concept-en.png)
+![應用程式截圖](_docs/img/ui-concept-en.webp)
 
 ## ✨ 主要功能
 

_docs/task-batch-api.md

@@ -0,0 +1,52 @@
+# Batch task creation
+
+`POST /api/v1/tasks/batch` accepts ordinary and grouped batches:
+
+```json
+{
+  "group": true,
+  "groupName": "album",
+  "opts": {"path": "/downloads"},
+  "reqs": [
+    {"rid": "resolved-resource-id", "opts": {"selectFiles": [1]}},
+    {"req": {"url": "https://example.com/cover.jpg"}}
+  ]
+}
+```
+
+Each member accepts `rid` or `req`. A nonempty `rid` takes priority and creates a
+task from the cached resolve result. A member's `opts`, when present, replaces
+batch `opts` as a whole; omitted fields do not inherit individual batch values.
+Members without `opts` use batch `opts`. BT file selections use the indexes from
+the resolve result. Requests without `rid` continue to work.
+
+For groups, batch `opts.path` selects the parent directory. Members inheriting
+batch options, or with an empty path, use the reserved group directory. Explicit
+relative member paths are resolved under that directory; do not prefix them with
+the batch parent path. For example, parent `downloads` and member path `nested`
+place the member in `downloads/<group name>/nested`. Absolute paths under the
+parent directory preserve their relative structure within the group. In ordinary
+batches, an empty member path uses the configured default download directory.
+
+All members are attempted. If at least one task is persisted, the response is
+successful and contains the created task IDs. A task that fails to start after
+being persisted is included; use the existing task controls to retry it.
+Successful tasks are kept when other members fail. If every member fails, the
+API returns an error and removes any newly reserved empty group and directory.
+
+Both ordinary and grouped batches now return an object in `data`:
+
+```json
+{"code": 0, "msg": "", "data": {"taskIds": ["task-1", "task-2"], "parentId": "group-id"}}
+```
+
+Ordinary batches omit `parentId`. **This replaces the former array response for
+ordinary batches.** The outer `code`/`msg`/`data` envelope is unchanged. Clients
+must read `data.taskIds`; there are no per-member results. A successful response
+completes the batch, including partial success, so clients must not automatically
+resubmit missing members.
+
+With `group: true`, the server generates the group ID and reserves its directory.
+Creation does not accept a client group ID or append to an existing group. Group
+metadata remains available through the existing task list and detail responses;
+`parentId` identifies membership for group operations.

cmd/api/main.go

@@ -1,6 +1,9 @@
 package main
 
 import (
+	"fmt"
+	"os"
+
 	"github.com/GopeedLab/gopeed/cmd"
 	"github.com/GopeedLab/gopeed/pkg/rest/model"
 )
@@ -13,5 +16,8 @@ func main() {
 		Storage:   model.StorageBolt,
 		WebEnable: true,
 	}
-	cmd.Start(cfg)
+	if err := cmd.Start(cfg); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
 }

cmd/server.go

@@ -16,15 +16,16 @@ import (
 //go:embed banner.txt
 var banner string
 
-func Start(cfg *model.StartConfig) {
+func Start(cfg *model.StartConfig) error {
-	fmt.Println(banner)
 	srv, listener, err := rest.BuildServer(cfg)
 	if err != nil {
-		panic(err)
+		return err
 	}
+	fmt.Println(banner)
 	downloadCfg, err := rest.Downloader.GetConfig()
 	if err != nil {
-		panic(err)
+		listener.Close()
+		return err
 	}
 	if rest.Downloader.IsFirstLoad() {
 		// Set default download config
@@ -62,8 +63,9 @@ func Start(cfg *model.StartConfig) {
 
 	fmt.Printf("Server start success on http://%s\n", listener.Addr().String())
 	if err := srv.Serve(listener); err != nil && err != http.ErrServerClosed {
-		panic(err)
+		return err
 	}
+	return nil
 }
 
 func watchExit() {

cmd/server_test.go

@@ -0,0 +1,35 @@
+package cmd
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/GopeedLab/gopeed/pkg/rest/model"
+)
+
+func TestStartInvalidWebAPITokenReturnsError(t *testing.T) {
+	tests := []struct {
+		name    string
+		webAuth *model.WebAuth
+	}{
+		{name: "no credentials"},
+		{name: "missing username", webAuth: &model.WebAuth{Password: "secret"}},
+		{name: "missing password", webAuth: &model.WebAuth{Username: "gopeed"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := Start(&model.StartConfig{
+				WebEnable: true,
+				ApiToken:  "test-token",
+				WebAuth:   tt.webAuth,
+			})
+			if err == nil {
+				t.Fatal("Start() error = nil, want web authentication validation error")
+			}
+			if !strings.Contains(err.Error(), "web authentication username and password are required") {
+				t.Fatalf("Start() error = %q, want missing web authentication credentials", err)
+			}
+		})
+	}
+}

cmd/web/flags.go

@@ -52,8 +52,8 @@ func loadCliArgs() *args {
 	flag.StringVar(cfg.Username, "username", "gopeed", "Web Authentication Username")
 	cfg.Password = flag.String("p", "", "Web Authentication Password, if no password is set, web authentication will not be enabled")
 	flag.StringVar(cfg.Password, "password", "", "Web Authentication Password, if no password is set, web authentication will not be enabled")
-	cfg.ApiToken = flag.String("T", "", "API token, it must be configured when using HTTP API in the case of enabling web authentication")
+	cfg.ApiToken = flag.String("T", "", "API token (requires web authentication with username and password)")
-	flag.StringVar(cfg.ApiToken, "api-token", "", "API token, it must be configured when using HTTP API in the case of enabling web authentication")
+	flag.StringVar(cfg.ApiToken, "api-token", "", "API token (requires web authentication with username and password)")
 	flag.BoolVar(&cfg.MCPEnable, "mcp-enable", false, "Enable MCP endpoint")
 	cfg.TempDir = flag.String("temp-dir", "", "Temporary directory (defaults to the OS temporary directory)")
 	cfg.StorageDir = flag.String("d", "", "Storage directory")

cmd/web/main.go

@@ -62,7 +62,10 @@ func main() {
 		WebFS:             sub,
 		WebAuth:           webAuth,
 	}
-	cmd.Start(cfg)
+	if err := cmd.Start(cfg); err != nil {
+		fmt.Fprintln(os.Stderr, err)
+		os.Exit(1)
+	}
 }
 
 func isNotBlank(str *string) bool {

go.mod

@@ -10,11 +10,13 @@ require (
 	github.com/bodgit/sevenzip v1.6.1
 	github.com/dop251/goja v0.0.0-20260106131823-651366fbe6e3
 	github.com/dop251/goja_nodejs v0.0.0-20240728170619-29b559befffc
+	github.com/fclairamb/ftpserverlib v0.31.0
 	github.com/go-git/go-git/v5 v5.8.1
 	github.com/google/jsonschema-go v0.4.3
 	github.com/gorilla/handlers v1.5.1
 	github.com/gorilla/mux v1.8.0
 	github.com/imroc/req/v3 v3.52.2
+	github.com/jlaffaye/ftp v0.2.4
 	github.com/matoous/go-nanoid/v2 v2.0.0
 	github.com/mattn/go-ieproxy v0.0.12
 	github.com/mholt/archives v0.1.5
@@ -23,6 +25,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/rs/dnscache v0.0.0-20230804202142-fc85eb664529
 	github.com/rs/zerolog v1.31.0
+	github.com/spf13/afero v1.15.0
 	github.com/tetratelabs/wazero v1.12.0
 	github.com/xiaoqidun/setft v0.0.0-20220310121541-be86327699ad
 	go.etcd.io/bbolt v1.4.3
@@ -59,7 +62,6 @@ require (
 	github.com/segmentio/asm v1.1.3 // indirect
 	github.com/segmentio/encoding v0.5.4 // indirect
 	github.com/sorairolake/lzip-go v0.3.8 // indirect
-	github.com/spf13/afero v1.15.0 // indirect
 	github.com/ulikunitz/xz v0.5.15 // indirect
 	github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect

go.sum

@@ -205,6 +205,8 @@ github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc
 github.com/emirpasic/gods v1.18.1/go.mod h1:8tpGGwCnJ5H4r6BWwaV6OrWmMoPhUl5jm/FMNAnJvWQ=
 github.com/envoyproxy/go-control-plane v0.9.1-0.20191026205805-5f8ba28d4473/go.mod h1:YTl/9mNaCwkRvm6d1a2C3ymFceY/DCBVvsKhRF0iEA4=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
+github.com/fclairamb/ftpserverlib v0.31.0 h1:ws4fLJvuU/mgg1uIgRjCJfc6CrFSgIBAOxVsnawXegE=
+github.com/fclairamb/ftpserverlib v0.31.0/go.mod h1:C5hUT0sd9P5XsHMxsmpcrT6wdZexu3qtUu1z0Ih6aG0=
 github.com/felixge/fgprof v0.9.5 h1:8+vR6yu2vvSKn08urWyEuxx75NWPEvybbkBirEpsbVY=
 github.com/felixge/fgprof v0.9.5/go.mod h1:yKl+ERSa++RYOs32d8K6WEXCB4uXdLls4ZaZPpayhMM=
 github.com/felixge/httpsnoop v1.0.1 h1:lvB5Jl89CsZtGIWuTcDM1E/vkVs49/Ml7JJe07l8SPQ=
@@ -347,6 +349,8 @@ github.com/imroc/req/v3 v3.52.2 h1:xJocr1aIv0a2K9knfBQ4JnZHk+kWTITdjf0mgDg229I=
 github.com/imroc/req/v3 v3.52.2/go.mod h1:dBGsDloOSZJcFs6PnTjZXYBJK70OXbZpizHBLNqcH2k=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 h1:BQSFePA1RWJOlocH6Fxy8MmwDt+yVQYULKfN0RoTN8A=
 github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99/go.mod h1:1lJo3i6rXxKeerYnT8Nvf0QmHCRC1n8sfWVwXF2Frvo=
+github.com/jlaffaye/ftp v0.2.4 h1:JqI85DdkfZj8ntaHk8W9U2SC3jNfiPUU70+wtIWmlfE=
+github.com/jlaffaye/ftp v0.2.4/go.mod h1:Y1ZnkzxownGIuX7xQ1mQzzkZ21+DbjVIyeKL/V+IIz4=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
@@ -527,6 +531,8 @@ github.com/rs/zerolog v1.31.0/go.mod h1:/7mN4D5sKwJLZQ2b/znpjC3/GQWY/xaDXUM0kKWR
 github.com/rwcarlsen/goexif v0.0.0-20190401172101-9e8deecbddbd/go.mod h1:hPqNNc0+uJM6H+SuU8sEs5K5IQeKccPqeSjfgcKGgPk=
 github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 h1:GHRpF1pTW19a8tTFrMLUcfWwyC0pnifVo2ClaLq+hP8=
 github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46/go.mod h1:uAQ5PCi+MFsC7HjREoAz1BU+Mq60+05gifQSsHSDG/8=
+github.com/secsy/goftp v0.0.0-20200609142545-aa2de14babf4 h1:PT+ElG/UUFMfqy5HrxJxNzj3QBOf7dZwupeVC+mG1Lo=
+github.com/secsy/goftp v0.0.0-20200609142545-aa2de14babf4/go.mod h1:MnkX001NG75g3p8bhFycnyIjeQoOjGL6CEIsdE/nKSY=
 github.com/segmentio/asm v1.1.3 h1:WM03sfUOENvvKexOLp+pCqgb/WDjsi7EK8gIsICtzhc=
 github.com/segmentio/asm v1.1.3/go.mod h1:Ld3L4ZXGNcSLRg4JBsZ3//1+f/TjYl0Mzen/DQy1EJg=
 github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0=
@@ -566,6 +572,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWDWE=
+github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg=
 github.com/tetratelabs/wazero v1.12.0 h1:DuWcpNu/FzgEXgGBDp8J1Spc+CWOvvtvVyjKlaZopYU=
 github.com/tetratelabs/wazero v1.12.0/go.mod h1:LvKtzl2RqO4gyF27BiXU+nKAjcV8f38U+kP/q2vgxh0=
 github.com/tidwall/btree v1.8.1 h1:27ehoXvm5AG/g+1VxLS1SD3vRhp/H7LuEfwNvddEdmA=
@@ -612,6 +620,8 @@ go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
 go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
 go.uber.org/mock v0.5.1 h1:ASgazW/qBmR+A32MYFDB6E2POoTgOwT509VP0CT/fjs=
 go.uber.org/mock v0.5.1/go.mod h1:ge71pBPLYDk7QIi1LupWxdAykm7KIEFchiOqd6z7qMM=
+go.yaml.in/yaml/v3 v3.0.5 h1:N6y/pJk8buWs9NY5ERU2HSMfm+IuD/OtfdAnq6kESPw=
+go.yaml.in/yaml/v3 v3.0.5/go.mod h1:HVTZu1O7/Vkt2N+BFy8Zza+lnLsABggaTM2ZpNIGuKg=
 go4.org v0.0.0-20230225012048-214862532bf5 h1:nifaUDeh+rPaBCMPMQHZmvJf+QdpLFnuQPwx+LxVmtc=
 go4.org v0.0.0-20230225012048-214862532bf5/go.mod h1:F57wTi5Lrj6WLyswp5EYV1ncrEbFGHD4hhz6S1ZYeaU=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=

internal/controller/controller.go

@@ -3,6 +3,7 @@ package controller
 import (
 	"github.com/GopeedLab/gopeed/internal/tempfiles"
 	"github.com/GopeedLab/gopeed/pkg/base"
+	"github.com/rs/zerolog"
 	"net/http"
 	"net/url"
 	"os"
@@ -15,6 +16,9 @@ type Controller struct {
 	TempDir           string
 	GetConfig         func(v any)
 	GetProxy          func(requestProxy *base.RequestProxy) func(*http.Request) (*url.URL, error)
+	// Logger is the downloader's core logger. It may be nil, in which case
+	// fetchers log nothing.
+	Logger *zerolog.Logger
 	FileController
 	//ContextDialer() (proxy.Dialer, error)
 }

internal/protocol/bt/fetcher.go

@@ -67,16 +67,11 @@ func (f *Fetcher) Setup(ctl *controller.Controller) {
 	return
 }
 
+// initClient must be called with lock held.
 func (f *Fetcher) initClient() (err error) {
-	lock.Lock()
-	defer lock.Unlock()
-
 	if client != nil {
 		return
 	}
-	if closeCtx == nil {
-		closeCtx, closeFunc = context.WithCancel(context.Background())
-	}
 
 	cfg = torrent.NewDefaultClientConfig()
 	cfg.Seed = true
@@ -94,12 +89,23 @@ func (f *Fetcher) initClient() (err error) {
 	}
 
 	closeCtx, closeFunc = context.WithCancel(context.Background())
-	go func() {
+	// closeClient may reset closeCtx before the goroutine runs, so pass it as an argument.
-		dnsResolver.Run(closeCtx)
+	go dnsResolver.Run(closeCtx)
-	}()
 	return
 }
 
+// addTorrentSpec holds lock from client initialization until the torrent is added,
+// so closeClient cannot close or reset the client in between.
+func (f *Fetcher) addTorrentSpec(spec *torrent.TorrentSpec) (*torrent.Torrent, bool, error) {
+	lock.Lock()
+	defer lock.Unlock()
+
+	if err := f.initClient(); err != nil {
+		return nil, false, err
+	}
+	return client.AddTorrentSpec(spec)
+}
+
 func (f *Fetcher) Resolve(req *base.Request) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
@@ -748,9 +754,6 @@ func (f *Fetcher) addTorrent(req *base.Request) (err error) {
 	if err = base.ParseReqExtra[bt.ReqExtra](req); err != nil {
 		return
 	}
-	if err = f.initClient(); err != nil {
-		return
-	}
 	schema := util.ParseSchema(req.URL)
 	privateTorrent := false
 	var spec *torrent.TorrentSpec
@@ -829,7 +832,7 @@ func (f *Fetcher) addTorrent(req *base.Request) (err error) {
 	s := &deferredFileStorage{}
 	spec.Storage = s
 	var added bool
-	f.torrent, added, err = client.AddTorrentSpec(spec)
+	f.torrent, added, err = f.addTorrentSpec(spec)
 	if err != nil {
 		return err
 	}

internal/protocol/bt/fetcher_test.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
+	"sync"
 	"testing"
 
 	"github.com/GopeedLab/gopeed/internal/controller"
@@ -71,6 +72,82 @@ func TestFetcher_ResolveWithProxy(t *testing.T) {
 	}))
 }
 
+func TestFetcher_CloseAfterInit(t *testing.T) {
+	if err := closeClient(); err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 100; i++ {
+		f := buildFetcher(t).(*Fetcher)
+		f.meta.Req = &base.Request{}
+		lock.Lock()
+		err := f.initClient()
+		lock.Unlock()
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := closeClient(); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestFetcher_CloseWithoutClient(t *testing.T) {
+	if err := closeClient(); err != nil {
+		t.Fatal(err)
+	}
+
+	t.Run("restored task", func(t *testing.T) {
+		_, build := new(FetcherManager).Restore()
+		f := build(&fetcher.FetcherMeta{
+			Req:  &base.Request{URL: "./testdata/test.torrent"},
+			Opts: &base.Options{},
+		}, &fetcherData{})
+		f.Setup(controller.NewController())
+		if err := f.Close(); err != nil {
+			t.Errorf("Close() got = %v, want nil", err)
+		}
+	})
+
+	t.Run("failed resolve", func(t *testing.T) {
+		f := buildFetcher(t)
+		if err := f.Resolve(&base.Request{URL: "./testdata/not-exist.torrent"}); err == nil {
+			t.Fatal("Resolve() got = nil, want error")
+		}
+		if err := f.Close(); err != nil {
+			t.Errorf("Close() got = %v, want nil", err)
+		}
+	})
+}
+
+func TestFetcher_ResolveWhileClientCloses(t *testing.T) {
+	if err := closeClient(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { closeClient() })
+
+	var wg sync.WaitGroup
+	for _, u := range []string{"./testdata/test.torrent", "./testdata/ubuntu-22.04-live-server-amd64.iso.torrent"} {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 10; i++ {
+				f := buildFetcher(t)
+				// The other goroutine may close the client after the torrent was added.
+				if err := f.Resolve(&base.Request{URL: u}); err != nil && err.Error() != "torrent closed during resolution" {
+					t.Errorf("Resolve(%s) got = %v", u, err)
+					return
+				}
+				f.Close()
+				if err := closeClient(); err != nil {
+					t.Errorf("closeClient() got = %v", err)
+					return
+				}
+			}
+		}()
+	}
+	wg.Wait()
+}
+
 func doResolve(t *testing.T, fetcher fetcher.Fetcher) {
 	t.Run("Resolve Single File", func(t *testing.T) {
 		err := fetcher.Resolve(&base.Request{

internal/protocol/ftp/conn.go

@@ -0,0 +1,404 @@
+package ftp
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/textproto"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/jlaffaye/ftp"
+	"golang.org/x/net/proxy"
+
+	pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
+)
+
+const (
+	connectTimeout = 15 * time.Second
+	readTimeout    = 15 * time.Second
+	quitTimeout    = 2 * time.Second
+)
+
+// target is a parsed FTP URL. It holds the password, so it must never be
+// formatted into an error or a log line.
+type target struct {
+	host    string // host name or IP, the TLS server name
+	addr    string // host:port
+	user    string
+	pass    string
+	path    string // path relative to the login directory, "" for the directory itself
+	tlsMode string
+}
+
+// parseTarget reads an ftp, ftps or ftpes URL. The path follows RFC 1738: it
+// is relative to the login directory, and an encoded leading slash
+// (ftp://host/%2Fpub) makes it absolute.
+func parseTarget(rawURL, optTLS string) (*target, error) {
+	u, err := url.Parse(rawURL)
+	if err != nil {
+		// url.Error repeats the whole URL, password included.
+		return nil, errors.New("ftp: invalid URL")
+	}
+	t := &target{}
+	switch strings.ToLower(u.Scheme) {
+	case "ftps":
+		t.tlsMode = pftp.TLSImplicit
+	case "ftpes":
+		t.tlsMode = pftp.TLSExplicit
+	case "ftp":
+		switch strings.ToLower(optTLS) {
+		case "", pftp.TLSNone:
+			t.tlsMode = pftp.TLSNone
+		case pftp.TLSExplicit:
+			t.tlsMode = pftp.TLSExplicit
+		case pftp.TLSImplicit:
+			t.tlsMode = pftp.TLSImplicit
+		default:
+			return nil, fmt.Errorf("ftp: unknown tls mode %q", optTLS)
+		}
+	default:
+		return nil, fmt.Errorf("ftp: unsupported scheme %q", u.Scheme)
+	}
+	t.host = u.Hostname()
+	if t.host == "" {
+		return nil, errors.New("ftp: URL has no host")
+	}
+	port := u.Port()
+	if port == "" {
+		port = "21"
+		if t.tlsMode == pftp.TLSImplicit {
+			port = "990"
+		}
+	}
+	t.addr = net.JoinHostPort(t.host, port)
+	if u.User != nil {
+		// Username and Password are already percent-decoded.
+		t.user = u.User.Username()
+		t.pass, _ = u.User.Password()
+	}
+	if t.user == "" {
+		t.user, t.pass = "anonymous", "anonymous@"
+	}
+	t.path = strings.TrimPrefix(u.Path, "/")
+	return t, nil
+}
+
+type dialContextFunc func(ctx context.Context, network, addr string) (net.Conn, error)
+
+// session is one logged-in control connection. It owns the network
+// connections it opens, so abort can unblock any read or write in progress.
+type session struct {
+	*ftp.ServerConn
+
+	ctrl     net.Conn // control connection as dialed, under any TLS layer
+	dataHost string   // where data connections go
+
+	mu     sync.Mutex
+	data   net.Conn // current data connection as dialed
+	closed bool
+	stop   func() bool
+}
+
+// deadline bounds the next control-connection exchange. The control
+// connection is idle during a transfer, so every exchange sets a new one.
+func (s *session) deadline() {
+	_ = s.ctrl.SetDeadline(time.Now().Add(readTimeout))
+}
+
+// clearDeadline lets the control connection stay idle through a transfer.
+func (s *session) clearDeadline() {
+	_ = s.ctrl.SetDeadline(time.Time{})
+}
+
+// abort closes every connection at once. Pending calls fail immediately.
+func (s *session) abort() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.closed = true
+	_ = s.ctrl.Close()
+	if s.data != nil {
+		_ = s.data.Close()
+	}
+}
+
+// quit logs out and waits, briefly, for the server to hang up. Servers that
+// cap logins count a session until they have seen it end, so the next login
+// would otherwise be refused.
+func (s *session) quit() {
+	if s.stop != nil {
+		s.stop()
+	}
+	_ = s.ctrl.SetDeadline(time.Now().Add(quitTimeout))
+	if s.ServerConn != nil {
+		_ = s.ServerConn.Quit()
+	}
+	s.abort()
+}
+
+func (s *session) setData(c net.Conn) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.closed {
+		_ = c.Close()
+		return net.ErrClosed
+	}
+	s.data = c
+	return nil
+}
+
+// ctrlConn is the control connection handed to the ftp library.
+type ctrlConn struct {
+	net.Conn
+}
+
+// RemoteAddr must be a *net.TCPAddr: the library asserts it. A connection
+// through a proxy may report something else, and its address is not used
+// for data connections anyway (see dialData).
+func (c ctrlConn) RemoteAddr() net.Addr {
+	if a, ok := c.Conn.RemoteAddr().(*net.TCPAddr); ok {
+		return a
+	}
+	return &net.TCPAddr{}
+}
+
+// Close is called by the library's Quit right after it sends QUIT. Reading
+// until the server hangs up (bounded by the deadline quit set) tells us the
+// server has ended the session.
+func (c ctrlConn) Close() error {
+	_, _ = io.Copy(io.Discard, c.Conn)
+	return c.Conn.Close()
+}
+
+// dial opens a control connection, upgrades it to TLS when asked, and logs in.
+//
+// The library is given a dial function for every connection. That is how
+// data connections go through the same proxy as the control connection, but
+// it also means the library does no TLS on them: with a dial function it
+// neither wraps an implicit-TLS control connection nor any data connection.
+// We do both here. Explicit TLS (AUTH TLS) is still done by the library, and
+// so are PBSZ and PROT P after login. For the same reason the library's
+// DialWithTimeout and DialWithContext would have no effect: connectTimeout
+// and ctx are applied to our own dials instead.
+func (f *Fetcher) dial(ctx context.Context, t *target) (*session, error) {
+	netDial, proxied, err := f.netDialer(t)
+	if err != nil {
+		return nil, err
+	}
+	var tlsConf *tls.Config
+	if t.tlsMode != pftp.TLSNone {
+		tlsConf = f.tlsConfig(t.host)
+	}
+
+	dialCtx, cancel := context.WithTimeout(ctx, connectTimeout)
+	defer cancel()
+	raw, err := netDial(dialCtx, "tcp", t.addr)
+	if err != nil {
+		return nil, err
+	}
+	ctrl := raw
+	if t.tlsMode == pftp.TLSImplicit {
+		tc := tls.Client(raw, tlsConf)
+		if err := tc.HandshakeContext(dialCtx); err != nil {
+			_ = raw.Close()
+			return nil, err
+		}
+		ctrl = tc
+	}
+
+	s := &session{ctrl: raw, dataHost: t.host}
+	if a, ok := raw.RemoteAddr().(*net.TCPAddr); ok && !proxied {
+		// Same IP as the control connection, as the library itself does,
+		// so a name with several addresses cannot send us elsewhere.
+		s.dataHost = a.IP.String()
+	}
+	s.stop = context.AfterFunc(ctx, s.abort)
+	s.deadline()
+
+	handedOut := false
+	opts := []ftp.DialOption{
+		ftp.DialWithShutTimeout(readTimeout),
+		ftp.DialWithDialFunc(func(network, addr string) (net.Conn, error) {
+			if !handedOut {
+				// Dial asks for the control connection first, once.
+				handedOut = true
+				return ctrlConn{ctrl}, nil
+			}
+			return s.dialData(ctx, netDial, tlsConf, addr)
+		}),
+	}
+	switch t.tlsMode {
+	case pftp.TLSImplicit:
+		opts = append(opts, ftp.DialWithTLS(tlsConf))
+	case pftp.TLSExplicit:
+		opts = append(opts, ftp.DialWithExplicitTLS(tlsConf))
+	}
+	c, err := ftp.Dial(t.addr, opts...)
+	if err != nil {
+		s.stop()
+		s.abort()
+		return nil, err
+	}
+	s.ServerConn = c
+	s.deadline()
+	if err := c.Login(t.user, t.pass); err != nil {
+		s.stop()
+		s.abort()
+		return nil, err
+	}
+	return s, nil
+}
+
+// dialData opens a passive data connection. The library takes the host from
+// the control connection's peer, which through a proxy is the proxy, so only
+// the port is used here.
+func (s *session) dialData(ctx context.Context, netDial dialContextFunc, tlsConf *tls.Config, addr string) (net.Conn, error) {
+	_, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		return nil, err
+	}
+	dialCtx, cancel := context.WithTimeout(ctx, connectTimeout)
+	defer cancel()
+	raw, err := netDial(dialCtx, "tcp", net.JoinHostPort(s.dataHost, port))
+	if err != nil {
+		return nil, err
+	}
+	if err := s.setData(raw); err != nil {
+		return nil, err
+	}
+	if tlsConf != nil {
+		// The handshake runs on the first read, as in the library, because
+		// some servers only start it once the transfer command is answered.
+		return tls.Client(raw, tlsConf), nil
+	}
+	return raw, nil
+}
+
+// tlsConfig returns the TLS settings for one control connection and its data
+// connections. Certificates are verified unless the request sets
+// skipVerifyCert, as for HTTP.
+//
+// Each call gets its own one-entry session cache, so a data connection
+// resumes the TLS session of its own control connection. vsftpd
+// (require_ssl_reuse) only checks that some session was resumed, but ProFTPD's
+// mod_tls also checks that it is the control connection's; a cache shared by
+// several workers would hand a data connection another worker's session.
+// A cache in the manager's base config is kept as given.
+func (f *Fetcher) tlsConfig(host string) *tls.Config {
+	cfg := &tls.Config{}
+	if f.manager != nil && f.manager.baseTLS != nil {
+		cfg = f.manager.baseTLS.Clone()
+	}
+	cfg.ServerName = host
+	if f.meta.Req != nil && f.meta.Req.SkipVerifyCert {
+		cfg.InsecureSkipVerify = true
+	}
+	if cfg.ClientSessionCache == nil {
+		cfg.ClientSessionCache = tls.NewLRUClientSessionCache(1)
+	}
+	return cfg
+}
+
+// netDialer picks how to reach the server, from the same proxy settings the
+// HTTP protocol reads (the task's proxy, otherwise the global one).
+func (f *Fetcher) netDialer(t *target) (dialContextFunc, bool, error) {
+	direct := (&net.Dialer{Timeout: connectTimeout}).DialContext
+	pu, err := f.proxyURL(t)
+	if err != nil {
+		return nil, false, err
+	}
+	if pu == nil {
+		return direct, false, nil
+	}
+	switch strings.ToLower(pu.Scheme) {
+	case "socks5", "socks5h":
+		d, err := proxy.FromURL(pu, &net.Dialer{Timeout: connectTimeout})
+		if err != nil {
+			return nil, false, fmt.Errorf("ftp: socks5 proxy: %w", err)
+		}
+		if cd, ok := d.(proxy.ContextDialer); ok {
+			return cd.DialContext, true, nil
+		}
+		return func(_ context.Context, network, addr string) (net.Conn, error) {
+			return d.Dial(network, addr)
+		}, true, nil
+	default:
+		// An HTTP proxy is not used. Every passive data connection goes to
+		// a port the server picks, and HTTP proxies usually allow CONNECT to
+		// port 443 only, so the transfer would fail after the login worked.
+		f.warnOnce.Do(func() {
+			if l := f.logger(); l != nil {
+				l.Warn().Msgf("ftp: the %s proxy at %s cannot carry FTP data connections (HTTP proxy), connecting directly; use a SOCKS5 proxy for FTP", strings.ToLower(pu.Scheme), pu.Host)
+			}
+		})
+		return direct, false, nil
+	}
+}
+
+func (f *Fetcher) proxyURL(t *target) (*url.URL, error) {
+	if f.ctl == nil || f.ctl.GetProxy == nil {
+		return nil, nil
+	}
+	handler := f.ctl.GetProxy(f.meta.Req.Proxy)
+	if handler == nil {
+		return nil, nil
+	}
+	// The proxy chooser reads only the URL; the credentials stay out.
+	req := &http.Request{URL: &url.URL{Scheme: "ftp", Host: t.addr, Path: "/" + t.path}, Header: http.Header{}}
+	pu, err := handler(req)
+	if err != nil {
+		return nil, errors.New("ftp: cannot read the proxy settings")
+	}
+	return pu, nil
+}
+
+// isLoginRefusal tells a server that is full apart from bad credentials.
+// 421 always means "not now". 530 means a refused login; it is only a
+// capacity limit when these credentials have worked before. The library
+// drops the code when USER itself is refused, so a "too many" message counts
+// as well.
+func isLoginRefusal(err error, credentialsWorked bool) bool {
+	var te *textproto.Error
+	if errors.As(err, &te) {
+		switch te.Code {
+		case ftp.StatusNotAvailable:
+			return true
+		case ftp.StatusNotLoggedIn:
+			return credentialsWorked || strings.Contains(strings.ToLower(te.Msg), "too many")
+		}
+		return false
+	}
+	return strings.Contains(strings.ToLower(err.Error()), "too many")
+}
+
+// isPermanent reports errors a retry cannot fix: a 5xx reply and a
+// certificate that does not verify.
+func isPermanent(err error) bool {
+	var te *textproto.Error
+	if errors.As(err, &te) {
+		return te.Code >= 500
+	}
+	var verifyErr *tls.CertificateVerificationError
+	if errors.As(err, &verifyErr) {
+		return true
+	}
+	var recordErr tls.RecordHeaderError
+	if errors.As(err, &recordErr) {
+		return true
+	}
+	return errors.Is(err, errSizeChanged)
+}
+
+func replyCode(err error) int {
+	var te *textproto.Error
+	if errors.As(err, &te) {
+		return te.Code
+	}
+	return 0
+}

internal/protocol/ftp/download_test.go

@@ -0,0 +1,146 @@
+package ftp_test
+
+import (
+	"bytes"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/GopeedLab/gopeed/internal/logger"
+	iftp "github.com/GopeedLab/gopeed/internal/protocol/ftp"
+	"github.com/GopeedLab/gopeed/pkg/base"
+	"github.com/GopeedLab/gopeed/pkg/download"
+	pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
+)
+
+// syncBuffer is a log sink that the downloader's goroutines can share.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *syncBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
+// taskEvents records the first done or error event of every task, so the
+// test never reads a task's status while the downloader writes it.
+type taskEvents struct {
+	mu    sync.Mutex
+	final map[string]download.EventKey
+}
+
+func (e *taskEvents) listen(ev *download.Event) {
+	if ev.Key != download.EventKeyDone && ev.Key != download.EventKeyError {
+		return
+	}
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if _, seen := e.final[ev.Task.ID]; !seen {
+		e.final[ev.Task.ID] = ev.Key
+	}
+}
+
+func (e *taskEvents) wait(t *testing.T, id string) download.EventKey {
+	t.Helper()
+	deadline := time.Now().Add(60 * time.Second)
+	for time.Now().Before(deadline) {
+		e.mu.Lock()
+		key, ok := e.final[id]
+		e.mu.Unlock()
+		if ok {
+			return key
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("task %s did not finish", id)
+	return ""
+}
+
+// TestPasswordNeverLogged runs FTP tasks through the downloader with its core
+// logger captured: a success, a missing file and a wrong password, all behind
+// an HTTP proxy that FTP must skip with a warning.
+func TestPasswordNeverLogged(t *testing.T) {
+	srv := iftp.StartFTPServer(t, iftp.ServerOptions{User: "u", Pass: "p@ss"})
+	want := srv.WriteFile(t, "secret.bin", iftp.RandomBytes(300000, 21))
+
+	d := download.NewDownloader(&download.DownloaderConfig{Storage: download.NewMemStorage(), StorageDir: t.TempDir()})
+	logs := &syncBuffer{}
+	d.Logger = &logger.Logger{Logger: zerolog.New(logs)}
+	events := &taskEvents{final: map[string]download.EventKey{}}
+	d.Listener(events.listen)
+	if err := d.Setup(); err != nil {
+		t.Fatal(err)
+	}
+	// Close, not Clear: Clear resets the extension list without a lock while
+	// the last task's done hooks may still read it.
+	defer d.Close()
+	cfg, err := d.GetConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	cfg.Proxy = &base.DownloaderProxyConfig{Enable: true, Scheme: "http", Host: "127.0.0.1:1", Usr: "proxyuser", Pwd: "proxysecret"}
+	if err := d.PutConfig(cfg); err != nil {
+		t.Fatal(err)
+	}
+
+	dir := t.TempDir()
+	create := func(rawURL string) string {
+		id, err := d.CreateDirect(&base.Request{URL: rawURL}, &base.Options{Path: dir, Extra: &pftp.OptsExtra{Connections: 2}})
+		if err != nil {
+			if strings.Contains(err.Error(), "p@ss") || strings.Contains(err.Error(), "p%40ss") {
+				t.Fatalf("the password leaked into an error: %v", err)
+			}
+			t.Fatal(err)
+		}
+		return id
+	}
+	root := "ftp://u:p%40ss@" + srv.Addr + "/"
+
+	okID := create(root + "secret.bin")
+	if key := events.wait(t, okID); key != download.EventKeyDone {
+		t.Fatalf("download ended with %q, logs:\n%s", key, logs.String())
+	}
+	ok := d.GetTask(okID)
+	if ok.Protocol != "ftp" {
+		t.Fatalf("protocol %q, want ftp", ok.Protocol)
+	}
+	if got := iftp.FileSHA256(t, ok.Meta.SingleFilepath()); got != want {
+		t.Fatalf("sha256 %s, want %s", got, want)
+	}
+	missingID := create(root + "missing.bin")
+	if key := events.wait(t, missingID); key != download.EventKeyError {
+		t.Fatalf("a missing file must fail, got %q", key)
+	}
+	wrongID := create("ftp://u:wr0ng%40pw@" + srv.Addr + "/secret.bin")
+	if key := events.wait(t, wrongID); key != download.EventKeyError {
+		t.Fatalf("a wrong password must fail, got %q", key)
+	}
+	missing, wrong := d.GetTask(missingID), d.GetTask(wrongID)
+	for _, task := range []*download.Task{ok, missing, wrong} {
+		if name := task.Name(); strings.Contains(name, "p@ss") || strings.Contains(name, "p%40ss") || strings.Contains(name, "wr0ng") {
+			t.Fatalf("task name %q shows the password", name)
+		}
+	}
+
+	out := logs.String()
+	if !strings.Contains(out, "task id") || !strings.Contains(out, "HTTP proxy") {
+		t.Fatalf("the captured log misses the expected lines, so it proves nothing:\n%s", out)
+	}
+	for _, secret := range []string{"p@ss", "p%40ss", "wr0ng", "proxysecret"} {
+		if strings.Contains(out, secret) {
+			t.Fatalf("%q appears in the log:\n%s", secret, out)
+		}
+	}
+}

internal/protocol/ftp/fetcher_manager.go

@@ -0,0 +1,109 @@
+package ftp
+
+import (
+	"crypto/tls"
+	"net/url"
+	"path"
+	"strings"
+
+	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/pkg/base"
+	pftp "github.com/GopeedLab/gopeed/pkg/protocol/ftp"
+)
+
+// fetcherData is what a task saves to resume later. Segments has the same
+// Begin, End and Downloaded fields every version writes; the other fields are
+// optional, so a save without them still restores.
+type fetcherData struct {
+	Segments []*segment
+	// MaxConns is the login ceiling the server enforced, 0 when none was hit.
+	MaxConns int `json:",omitempty"`
+	// NoRest records that the server refused REST, so the task downloads each
+	// file in one sequential stream.
+	NoRest bool `json:",omitempty"`
+}
+
+// FetcherManager handles ftp://, ftps:// (implicit TLS) and ftpes:// (explicit
+// TLS) URLs.
+type FetcherManager struct {
+	// baseTLS is the base for every TLS connection. Nil means the system
+	// roots; the package's tests set it to trust their own certificate. The
+	// task's skipVerifyCert still turns verification off.
+	baseTLS *tls.Config
+}
+
+func (fm *FetcherManager) Name() string {
+	return "ftp"
+}
+
+func (fm *FetcherManager) Filters() []*fetcher.SchemeFilter {
+	return []*fetcher.SchemeFilter{
+		{Type: fetcher.FilterTypeUrl, Pattern: "FTP"},
+		{Type: fetcher.FilterTypeUrl, Pattern: "FTPS"},
+		{Type: fetcher.FilterTypeUrl, Pattern: "FTPES"},
+	}
+}
+
+func (fm *FetcherManager) Build() fetcher.Fetcher {
+	return &Fetcher{manager: fm}
+}
+
+// ParseName names a task before it is resolved. It never includes the
+// userinfo of the URL.
+func (fm *FetcherManager) ParseName(u string) string {
+	parsed, err := url.Parse(u)
+	if err != nil {
+		return ""
+	}
+	name := path.Base(strings.TrimSuffix(parsed.Path, "/"))
+	if name == "" || name == "/" || name == "." {
+		name = parsed.Hostname()
+	}
+	return name
+}
+
+func (fm *FetcherManager) AutoRename() bool {
+	return true
+}
+
+func (fm *FetcherManager) DefaultConfig() any {
+	return &config{Connections: defaultConnections}
+}
+
+func (fm *FetcherManager) Store(f fetcher.Fetcher) (any, error) {
+	ff := f.(*Fetcher)
+	ff.mu.Lock()
+	defer ff.mu.Unlock()
+	segments := make([]*segment, len(ff.segments))
+	for i, s := range ff.segments {
+		c := *s
+		segments[i] = &c
+	}
+	return &fetcherData{
+		Segments: segments,
+		MaxConns: ff.maxConns,
+		NoRest:   ff.noRest,
+	}, nil
+}
+
+func (fm *FetcherManager) Restore() (v any, f func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher) {
+	return &fetcherData{}, func(meta *fetcher.FetcherMeta, v any) fetcher.Fetcher {
+		fd := v.(*fetcherData)
+		ff := fm.Build().(*Fetcher)
+		ff.meta = meta
+		if meta != nil && meta.Req != nil {
+			_ = base.ParseReqExtra[pftp.ReqExtra](meta.Req)
+		}
+		if meta != nil && meta.Opts != nil {
+			_ = base.ParseOptExtra[pftp.OptsExtra](meta.Opts)
+		}
+		ff.segments = fd.Segments
+		ff.maxConns = fd.MaxConns
+		ff.noRest = fd.NoRest
+		return ff
+	}
+}
+
+func (fm *FetcherManager) Close() error {
+	return nil
+}

internal/protocol/ftp/walk.go

@@ -0,0 +1,187 @@
+package ftp
+
+import (
+	"cmp"
+	"fmt"
+	"path"
+	"slices"
+	"strings"
+	"time"
+
+	"github.com/jlaffaye/ftp"
+
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+// Limits of a directory download. A tree past either one is refused with an
+// error that names the limit; it is never cut short.
+const maxWalkDepth = 8
+
+// maxWalkFiles is a variable so tests can lower it.
+var maxWalkFiles = 10000
+
+// resolveTarget describes the URL's path: one file, or a directory walked
+// into a multi-file resource whose paths are relative to it.
+func (f *Fetcher) resolveTarget(s *session, t *target) (*base.Resource, error) {
+	isDir, size, mtime, err := f.stat(s, t.path)
+	if err != nil {
+		return nil, err
+	}
+	if !isDir {
+		return &base.Resource{
+			Size:  size,
+			Range: true,
+			Files: []*base.FileInfo{{
+				Name:  path.Base(t.path),
+				Size:  size,
+				Ctime: mtime,
+			}},
+		}, nil
+	}
+
+	files, err := f.walk(s, t.path)
+	if err != nil {
+		return nil, err
+	}
+	name := path.Base(strings.TrimSuffix(t.path, "/"))
+	if name == "" || name == "." || name == "/" {
+		name = t.host
+	}
+	res := &base.Resource{Name: name, Range: true, Files: files}
+	res.CalcSize(nil)
+	return res, nil
+}
+
+// stat tells a file from a directory. MLST answers both at once. Without
+// it, SIZE answers for a file, and a directory is one we can change into.
+func (f *Fetcher) stat(s *session, p string) (isDir bool, size int64, mtime *time.Time, err error) {
+	if p == "" || strings.HasSuffix(p, "/") {
+		return true, 0, nil, nil
+	}
+	s.deadline()
+	if entry, mlstErr := s.GetEntry(p); mlstErr == nil {
+		switch entry.Type {
+		case ftp.EntryTypeFolder:
+			return true, 0, nil, nil
+		case ftp.EntryTypeFile:
+			if !entry.Time.IsZero() {
+				t := entry.Time
+				mtime = &t
+			}
+			return false, int64(entry.Size), mtime, nil
+		}
+		// A link: SIZE below follows it on the server.
+	}
+	// MLST is missing or refused the path: the older commands decide. A
+	// broken connection fails them too, with its own error.
+
+	s.deadline()
+	size, sizeErr := s.FileSize(p)
+	if sizeErr == nil {
+		if s.IsGetTimeSupported() {
+			s.deadline()
+			if t, err := s.GetTime(p); err == nil {
+				mtime = &t
+			}
+		}
+		return false, size, mtime, nil
+	}
+	if replyCode(sizeErr) == 0 {
+		return false, 0, nil, sizeErr
+	}
+
+	// SIZE failed: a directory, or a missing path.
+	s.deadline()
+	cwd, err := s.CurrentDir()
+	if err != nil {
+		return false, 0, nil, err
+	}
+	s.deadline()
+	if err := s.ChangeDir(p); err != nil {
+		return false, 0, nil, sizeErr
+	}
+	s.deadline()
+	if err := s.ChangeDir(cwd); err != nil {
+		return false, 0, nil, err
+	}
+	return true, 0, nil, nil
+}
+
+// safeEntryName reports whether a name from a listing can become a local
+// path element. A hostile server could otherwise list "../x" and write
+// outside the task directory.
+func safeEntryName(name string) bool {
+	if name == "" || name == "." || name == ".." {
+		return false
+	}
+	return !strings.ContainsAny(name, "/\\\x00")
+}
+
+// walk lists a directory tree, at most maxWalkDepth levels below root and
+// maxWalkFiles files. Files come back sorted by path, then name.
+func (f *Fetcher) walk(s *session, root string) ([]*base.FileInfo, error) {
+	type dir struct {
+		remote string // path on the server
+		rel    string // path below root, "" for root itself
+		depth  int
+	}
+	var files []*base.FileInfo
+	stack := []dir{{remote: strings.TrimSuffix(root, "/"), rel: "", depth: 0}}
+	for len(stack) > 0 {
+		d := stack[len(stack)-1]
+		stack = stack[:len(stack)-1]
+
+		s.deadline()
+		entries, err := s.List(d.remote)
+		if err != nil {
+			return nil, err
+		}
+		for _, e := range entries {
+			if e.Name == "." || e.Name == ".." {
+				continue
+			}
+			if !safeEntryName(e.Name) {
+				f.warn("ftp: skipping a listed name that is not a safe file name: %q", e.Name)
+				continue
+			}
+			remote := path.Join(d.remote, e.Name)
+			rel := path.Join(d.rel, e.Name)
+			if d.remote == "" {
+				remote = e.Name
+			}
+			entryType, size := e.Type, int64(e.Size)
+			if entryType == ftp.EntryTypeLink {
+				// A link to a file has a size; a link to a directory is
+				// not followed, so a loop cannot make the walk endless.
+				s.deadline()
+				linkSize, err := s.FileSize(remote)
+				if err != nil {
+					f.warn("ftp: skipping link %q: %v", rel, err)
+					continue
+				}
+				entryType, size = ftp.EntryTypeFile, linkSize
+			}
+			switch entryType {
+			case ftp.EntryTypeFolder:
+				if d.depth+1 > maxWalkDepth {
+					return nil, fmt.Errorf("ftp: the folder is deeper than %d levels, the limit for a folder download", maxWalkDepth)
+				}
+				stack = append(stack, dir{remote: remote, rel: rel, depth: d.depth + 1})
+			case ftp.EntryTypeFile:
+				if len(files) >= maxWalkFiles {
+					return nil, fmt.Errorf("ftp: the folder has more than %d files, the limit for a folder download", maxWalkFiles)
+				}
+				fi := &base.FileInfo{Name: e.Name, Path: d.rel, Size: size}
+				if !e.Time.IsZero() {
+					t := e.Time
+					fi.Ctime = &t
+				}
+				files = append(files, fi)
+			}
+		}
+	}
+	slices.SortFunc(files, func(a, b *base.FileInfo) int {
+		return cmp.Or(strings.Compare(a.Path, b.Path), strings.Compare(a.Name, b.Name))
+	})
+	return files, nil
+}

internal/protocol/http/fetcher.go

@@ -304,6 +304,7 @@ type Fetcher struct {
 
 	// Resolve response - kept open for one-time URLs
 	resolveResp       *http.Response
+	resolveRespCancel context.CancelFunc // Cancels the Resolve request
 	resolveRespLock   sync.Mutex
 
 	// Async prefetch during resolve phase
@@ -425,8 +426,11 @@ func (f *Fetcher) Resolve(req *base.Request) error {
 
 	// Build HTTP request WITHOUT Range header (normal request)
 	// This allows the response to be reused for downloading (important for one-time URLs)
-	httpReq, err := f.buildRequest(context.TODO(), req)
+	// The body outlives Resolve, so the read timeout gets its own cancel func.
+	reqCtx, reqCancel := context.WithCancel(context.Background())
+	httpReq, err := f.buildRequest(reqCtx, req)
 	if err != nil {
+		reqCancel()
 		f.setState(stateError)
 		return err
 	}
@@ -438,6 +442,7 @@ func (f *Fetcher) Resolve(req *base.Request) error {
 	connStartTime := time.Now()
 	resp, err := client.Do(httpReq)
 	if err != nil {
+		reqCancel()
 		f.setState(stateError)
 		return err
 	}
@@ -452,6 +457,7 @@ func (f *Fetcher) Resolve(req *base.Request) error {
 
 	if resp.StatusCode != base.HttpCodeOK && resp.StatusCode != base.HttpCodePartialContent {
 		resp.Body.Close()
+		reqCancel()
 		f.setState(stateError)
 		return NewRequestError(resp.StatusCode)
 	}
@@ -469,6 +475,7 @@ func (f *Fetcher) Resolve(req *base.Request) error {
 		start, _, total, err := parseRangeResponse(resp)
 		if err != nil {
 			resp.Body.Close()
+			reqCancel()
 			f.setState(stateError)
 			return err
 		}
@@ -536,8 +543,10 @@ func (f *Fetcher) Resolve(req *base.Request) error {
 	f.resolveRespLock.Lock()
 	if reuseResolveBody {
 		f.resolveResp = resp
+		f.resolveRespCancel = reqCancel
 	} else {
 		resp.Body.Close()
+		reqCancel()
 		f.prefetchDone.Store(true)
 	}
 	f.resolveRespLock.Unlock()
@@ -574,7 +583,7 @@ func (f *Fetcher) asyncPrefetch() {
 
 	// Get the resolve response
 	f.resolveRespLock.Lock()
-	resp := f.resolveResp
+	resp, cancel := f.resolveResp, f.resolveRespCancel
 	f.resolveRespLock.Unlock()
 
 	if resp == nil {
@@ -585,10 +594,7 @@ func (f *Fetcher) asyncPrefetch() {
 		// Close the response even if temporary-file setup fails or shutdown
 		// rejects a late prefetch. Otherwise the server can remain blocked writing.
 		f.resolveRespLock.Lock()
-		if f.resolveResp != nil {
+		f.closeResolveRespLocked()
-			f.resolveResp.Body.Close()
-			f.resolveResp = nil
-		}
 		f.resolveRespLock.Unlock()
 	}()
 
@@ -618,7 +624,7 @@ func (f *Fetcher) asyncPrefetch() {
 	f.prefetchFilePath = tmpFile.Name()
 
 	buf := make([]byte, 32*1024) // 32KB buffer
-	reader := f.responseReader(resp)
+	reader := f.responseReader(resp, cancel)
 
 	for {
 		select {
@@ -881,10 +887,7 @@ func (f *Fetcher) doStart() error {
 		if !f.resolveFallback.Load() {
 			// Also close resolve response if still open.
 			f.resolveRespLock.Lock()
-			if f.resolveResp != nil {
+			f.closeResolveRespLocked()
-				f.resolveResp.Body.Close()
-				f.resolveResp = nil
-			}
 			f.resolveRespLock.Unlock()
 		}
 	}
@@ -1337,7 +1340,11 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
 	}
 	f.connMu.Unlock()
 
-	httpReq, err := f.buildRequest(conn.ctx, f.meta.Req)
+	// The read timeout cancels this request through its own context.
+	reqCtx, reqCancel := context.WithCancel(conn.ctx)
+	defer reqCancel()
+
+	httpReq, err := f.buildRequest(reqCtx, f.meta.Req)
 	if err != nil {
 		return err
 	}
@@ -1365,7 +1372,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
 		// Check if this might be a redirect URL expiration error
 		// If so, try falling back to the original URL
 		if f.hasRedirectURL() && isRedirectExpiredError(originalErr) {
-			fallbackResp, fallbackErr := f.tryFallbackToOriginalURL(conn.ctx, client, rangeStart, rangeEnd, rangeRequested, ifRange)
+			fallbackResp, fallbackErr := f.tryFallbackToOriginalURL(reqCtx, client, rangeStart, rangeEnd, rangeRequested, ifRange)
 			if fallbackErr == nil && fallbackResp != nil {
 				// Fallback succeeded, use this response instead
 				resp = fallbackResp
@@ -1486,7 +1493,7 @@ func (f *Fetcher) downloadChunkOnce(conn *connection, client *http.Client, buf [
 		f.slowStart.onConnectSuccess()
 	}
 
-	reader := f.responseReader(resp)
+	reader := f.responseReader(resp, reqCancel)
 	var responseBytesRead int64
 	for {
 		if conn.ctx.Err() != nil {
@@ -1840,8 +1847,8 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
 
 	// Get the resolve response
 	f.resolveRespLock.Lock()
-	resp := f.resolveResp
+	resp, cancel := f.resolveResp, f.resolveRespCancel
-	f.resolveResp = nil // Take ownership
+	f.resolveResp, f.resolveRespCancel = nil, nil // Take ownership
 	f.resolveRespLock.Unlock()
 
 	if resp == nil {
@@ -1850,6 +1857,9 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
 		return
 	}
 
+	if cancel != nil {
+		defer cancel()
+	}
 	defer resp.Body.Close()
 
 	f.connMu.Lock()
@@ -1866,7 +1876,7 @@ func (f *Fetcher) runConnectionWithResolveResp(conn *connection) {
 	}
 
 	// Download data from resolve response
-	reader := f.responseReader(resp)
+	reader := f.responseReader(resp, cancel)
 	for {
 		if conn.ctx.Err() != nil {
 			return
@@ -2481,10 +2491,7 @@ func (f *Fetcher) Pause() error {
 
 	// Clean up resolve response if still held
 	f.resolveRespLock.Lock()
-	if f.resolveResp != nil {
+	f.closeResolveRespLocked()
-		f.resolveResp.Body.Close()
-		f.resolveResp = nil
-	}
 	f.resolveRespLock.Unlock()
 
 	f.fileMu.Lock()
@@ -2602,9 +2609,25 @@ func (f *Fetcher) Wait() error {
 
 // Generated output can pause while its producer waits for processing capacity.
 // Its producer owns upstream deadlines; cancellation still closes the response.
-func (f *Fetcher) responseReader(resp *http.Response) io.Reader {
+// Otherwise a stalled read cancels the request via cancel, which must belong
+// to the request that produced resp.
+func (f *Fetcher) responseReader(resp *http.Response, cancel context.CancelFunc) io.Reader {
 	if f.ctl != nil && f.ctl.ManagedProduction != nil && resp.Request != nil && f.ctl.ManagedProduction(resp.Request.URL.String()) {
 		return resp.Body
 	}
-	return NewTimeoutReader(resp.Body, readTimeout)
+	return NewTimeoutReader(resp.Body, readTimeout, cancel)
+}
+
+// closeResolveRespLocked closes the kept Resolve response, if any, and
+// releases its request context. The caller must hold resolveRespLock.
+func (f *Fetcher) closeResolveRespLocked() {
+	if f.resolveResp == nil {
+		return
+	}
+	f.resolveResp.Body.Close()
+	f.resolveResp = nil
+	if f.resolveRespCancel != nil {
+		f.resolveRespCancel()
+		f.resolveRespCancel = nil
+	}
 }

internal/protocol/http/timeout_reader.go

@@ -6,35 +6,56 @@ import (
 	"time"
 )
 
+// TimeoutReader fails a Read that makes no progress within timeout.
+//
+// One timer per reader is armed when Read starts and stopped before it
+// returns, so time the caller spends between reads is not counted. If the
+// timer fires, it calls cancel, which should cancel the HTTP request the
+// reader belongs to so the transport unblocks the pending Read. Read still
+// waits for the underlying Read to return; that Read and every later one
+// return context.DeadlineExceeded. Errors from any other cancellation are
+// returned unchanged, and closing the response body is left to the caller.
+//
+// With a nil cancel the pending Read cannot be interrupted; it still reports
+// context.DeadlineExceeded, but only once it returns.
+//
+// TimeoutReader is not safe for concurrent use.
 type TimeoutReader struct {
 	reader   io.Reader
 	timeout  time.Duration
+	cancel   context.CancelFunc
+	timer    *time.Timer
+	timedOut bool
 }
 
-func NewTimeoutReader(r io.Reader, timeout time.Duration) *TimeoutReader {
+func NewTimeoutReader(r io.Reader, timeout time.Duration, cancel context.CancelFunc) *TimeoutReader {
+	if cancel == nil {
+		cancel = func() {}
+	}
 	return &TimeoutReader{
 		reader:  r,
 		timeout: timeout,
+		cancel:  cancel,
 	}
 }
 
 func (tr *TimeoutReader) Read(p []byte) (n int, err error) {
-	ctx, cancel := context.WithTimeout(context.Background(), tr.timeout)
+	if tr.timedOut {
-	defer cancel()
+		return 0, context.DeadlineExceeded
-
-	done := make(chan struct{})
-	var readErr error
-	var bytesRead int
-
-	go func() {
-		bytesRead, readErr = tr.reader.Read(p)
-		close(done)
-	}()
-
-	select {
-	case <-done:
-		return bytesRead, readErr
-	case <-ctx.Done():
-		return 0, ctx.Err()
 	}
+	if tr.timer == nil {
+		tr.timer = time.AfterFunc(tr.timeout, tr.cancel)
+	} else {
+		tr.timer.Reset(tr.timeout)
+	}
+
+	n, err = tr.reader.Read(p)
+
+	if !tr.timer.Stop() {
+		// The timer fired during this Read and cancelled the request, so err is
+		// whatever the interrupted read produced. Report the timeout instead.
+		tr.timedOut = true
+		return n, context.DeadlineExceeded
+	}
+	return n, err
 }

internal/protocol/http/timeout_reader_test.go

@@ -2,17 +2,26 @@ package http
 
 import (
 	"bytes"
+	"compress/gzip"
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	"errors"
 	"io"
+	gohttp "net/http"
+	"net/http/httptest"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/GopeedLab/gopeed/internal/httpclient"
 )
 
 func TestTimeoutReader_Read(t *testing.T) {
 	data := []byte("Hello, World!")
 	reader := bytes.NewReader(data)
-	timeoutReader := NewTimeoutReader(reader, 1*time.Second)
+	timeoutReader := NewTimeoutReader(reader, 1*time.Second, nil)
 
 	buf := make([]byte, len(data))
 	n, err := timeoutReader.Read(buf)
@@ -29,7 +38,7 @@ func TestTimeoutReader_Read(t *testing.T) {
 
 func TestTimeoutReader_ReadTimeout(t *testing.T) {
 	reader := &slowReader{delay: 2 * time.Second}
-	timeoutReader := NewTimeoutReader(reader, 1*time.Second)
+	timeoutReader := NewTimeoutReader(reader, 1*time.Second, nil)
 
 	buf := make([]byte, 8192)
 	_, err := timeoutReader.Read(buf)
@@ -49,3 +58,340 @@ func (sr *slowReader) Read(p []byte) (n int, err error) {
 	time.Sleep(sr.delay)
 	return 0, io.EOF
 }
+
+// The timer must only cover time spent inside Read, not caller work between reads.
+func TestTimeoutReader_IdleBetweenReads(t *testing.T) {
+	var cancelled atomic.Bool
+	timeoutReader := NewTimeoutReader(bytes.NewReader([]byte("abcdef")), 50*time.Millisecond, func() {
+		cancelled.Store(true)
+	})
+
+	buf := make([]byte, 3)
+	if _, err := timeoutReader.Read(buf); err != nil {
+		t.Fatalf("first read: %v", err)
+	}
+	time.Sleep(200 * time.Millisecond)
+	if _, err := timeoutReader.Read(buf); err != nil {
+		t.Fatalf("second read after idle: %v", err)
+	}
+	if cancelled.Load() {
+		t.Fatal("request was cancelled while idle between reads")
+	}
+}
+
+// A timed-out Read must not return while the underlying Read can still write
+// into the caller's buffer, and must leave closing the reader to the caller.
+func TestTimeoutReader_TimedOutReadDoesNotOutliveCaller(t *testing.T) {
+	reader := &lateWriteReader{cancelled: make(chan struct{})}
+	timeoutReader := NewTimeoutReader(reader, 50*time.Millisecond, reader.cancel)
+
+	buf := make([]byte, 8)
+	_, err := timeoutReader.Read(buf)
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("expected %v, got %v", context.DeadlineExceeded, err)
+	}
+	if !reader.returned.Load() {
+		t.Fatal("Read returned before the underlying Read finished")
+	}
+	if _, err := timeoutReader.Read(buf); !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("read after timeout: expected %v, got %v", context.DeadlineExceeded, err)
+	}
+	if reader.closed.Load() {
+		t.Fatal("reader was closed by the timeout")
+	}
+}
+
+// lateWriteReader blocks until cancelled, then writes into p before returning.
+type lateWriteReader struct {
+	cancelOnce sync.Once
+	cancelled  chan struct{}
+	returned   atomic.Bool
+	closed     atomic.Bool
+}
+
+func (r *lateWriteReader) Read(p []byte) (int, error) {
+	<-r.cancelled
+	time.Sleep(20 * time.Millisecond)
+	p[0] = 'x'
+	r.returned.Store(true)
+	return 0, context.Canceled
+}
+
+func (r *lateWriteReader) cancel() {
+	r.cancelOnce.Do(func() { close(r.cancelled) })
+}
+
+func (r *lateWriteReader) Close() error {
+	r.closed.Store(true)
+	return nil
+}
+
+type timeoutReaderClientCase struct {
+	name   string
+	http2  bool
+	client func(t *testing.T, server *httptest.Server) *gohttp.Client
+}
+
+func timeoutReaderClientCases() []timeoutReaderClientCase {
+	newClient := func(mode httpclient.ImpersonationMode) func(t *testing.T, server *httptest.Server) *gohttp.Client {
+		return func(t *testing.T, server *httptest.Server) *gohttp.Client {
+			var tlsConfig *tls.Config
+			if server.TLS != nil {
+				roots := x509.NewCertPool()
+				roots.AddCert(server.Certificate())
+				tlsConfig = &tls.Config{RootCAs: roots}
+			}
+			client, err := httpclient.NewClient(httpclient.Options{
+				Transport: httpclient.TransportOptions{
+					TLSClientConfig: tlsConfig,
+				},
+				Impersonation: httpclient.ImpersonationOptions{
+					Mode:    mode,
+					Browser: httpclient.BrowserChrome,
+				},
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			return client
+		}
+	}
+	var cases []timeoutReaderClientCase
+	for _, mode := range []httpclient.ImpersonationMode{httpclient.ImpersonationNative, httpclient.ImpersonationFixed} {
+		cases = append(cases,
+			timeoutReaderClientCase{name: string(mode) + "/http1", client: newClient(mode)},
+			timeoutReaderClientCase{name: string(mode) + "/http2", http2: true, client: newClient(mode)},
+		)
+	}
+	return cases
+}
+
+func startTimeoutReaderServer(t *testing.T, useHTTP2 bool, handler gohttp.HandlerFunc) *httptest.Server {
+	server := httptest.NewUnstartedServer(handler)
+	if useHTTP2 {
+		server.EnableHTTP2 = true
+		server.StartTLS()
+	} else {
+		server.Start()
+	}
+	t.Cleanup(server.Close)
+	return server
+}
+
+// readBodyWithTimeout sends the request under its own context derived from
+// ctx, as the fetcher does, and reads the body through a TimeoutReader.
+func readBodyWithTimeout(t *testing.T, ctx context.Context, client *gohttp.Client, url string, timeout time.Duration, wantProto int) ([]byte, error) {
+	t.Helper()
+	reqCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	req, err := gohttp.NewRequestWithContext(reqCtx, gohttp.MethodGet, url, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.ProtoMajor != wantProto {
+		t.Fatalf("expected HTTP/%d, got %s", wantProto, resp.Proto)
+	}
+
+	type result struct {
+		data []byte
+		err  error
+	}
+	done := make(chan result, 1)
+	go func() {
+		data, err := io.ReadAll(NewTimeoutReader(resp.Body, timeout, cancel))
+		done <- result{data, err}
+	}()
+	select {
+	case r := <-done:
+		return r.data, r.err
+	case <-time.After(timeout + 10*time.Second):
+		t.Fatal("Read did not return after the read timeout")
+		return nil, nil
+	}
+}
+
+func TestTimeoutReader_StalledHTTPBodyTimesOut(t *testing.T) {
+	for _, tc := range timeoutReaderClientCases() {
+		t.Run(tc.name, func(t *testing.T) {
+			server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
+				w.Header().Set("Content-Length", "1048576")
+				w.Write(bytes.Repeat([]byte("a"), 1024))
+				w.(gohttp.Flusher).Flush()
+				<-r.Context().Done()
+			})
+
+			wantProto := 1
+			if tc.http2 {
+				wantProto = 2
+			}
+			const timeout = 200 * time.Millisecond
+			start := time.Now()
+			data, err := readBodyWithTimeout(t, context.Background(), tc.client(t, server), server.URL, timeout, wantProto)
+			if !errors.Is(err, context.DeadlineExceeded) {
+				t.Fatalf("expected %v, got %v", context.DeadlineExceeded, err)
+			}
+			if len(data) != 1024 {
+				t.Fatalf("expected the 1024 bytes sent before the stall, got %d", len(data))
+			}
+			if elapsed := time.Since(start); elapsed < timeout {
+				t.Fatalf("timed out after %v, before the %v read timeout", elapsed, timeout)
+			}
+		})
+	}
+}
+
+func TestTimeoutReader_SlowSteadyHTTPBodyDoesNotTimeOut(t *testing.T) {
+	const (
+		timeout  = 300 * time.Millisecond
+		interval = 30 * time.Millisecond
+		chunks   = 40 // 1.2s in total, four times the read timeout
+	)
+	for _, tc := range timeoutReaderClientCases() {
+		t.Run(tc.name, func(t *testing.T) {
+			server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
+				for i := 0; i < chunks; i++ {
+					w.Write([]byte("0123456789"))
+					w.(gohttp.Flusher).Flush()
+					time.Sleep(interval)
+				}
+			})
+
+			wantProto := 1
+			if tc.http2 {
+				wantProto = 2
+			}
+			data, err := readBodyWithTimeout(t, context.Background(), tc.client(t, server), server.URL, timeout, wantProto)
+			if err != nil {
+				t.Fatalf("expected no error, got %v", err)
+			}
+			if len(data) != chunks*10 {
+				t.Fatalf("expected %d bytes, got %d", chunks*10, len(data))
+			}
+		})
+	}
+}
+
+// A gzip-encoded body that stalls must time out without racing the caller's
+// Body.Close against the decompressing reader.
+func TestTimeoutReader_StalledGzipHTTPBodyTimesOut(t *testing.T) {
+	payload := bytes.Repeat([]byte("gopeed"), 1024)
+	stages := []struct {
+		name  string
+		write func(w io.Writer)
+		want  int
+	}{
+		{
+			// The first bytes of a gzip header, so the client blocks inside
+			// gzip.NewReader.
+			name:  "incomplete-header",
+			write: func(w io.Writer) { w.Write([]byte{0x1f, 0x8b, 0x08, 0x00}) },
+			want:  0,
+		},
+		{
+			// A complete header and some compressed data, flushed without
+			// finishing the gzip stream.
+			name: "stalled-body",
+			write: func(w io.Writer) {
+				gz := gzip.NewWriter(w)
+				gz.Write(payload)
+				gz.Flush()
+			},
+			want: len(payload),
+		},
+	}
+	for _, stage := range stages {
+		for _, tc := range timeoutReaderClientCases() {
+			t.Run(stage.name+"/"+tc.name, func(t *testing.T) {
+				server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
+					w.Header().Set("Content-Encoding", "gzip")
+					stage.write(w)
+					w.(gohttp.Flusher).Flush()
+					<-r.Context().Done()
+				})
+
+				wantProto := 1
+				if tc.http2 {
+					wantProto = 2
+				}
+				data, err := readBodyWithTimeout(t, context.Background(), tc.client(t, server), server.URL, 200*time.Millisecond, wantProto)
+				if !errors.Is(err, context.DeadlineExceeded) {
+					t.Fatalf("expected %v, got %v", context.DeadlineExceeded, err)
+				}
+				if len(data) != stage.want {
+					t.Fatalf("expected the %d bytes sent before the stall, got %d", stage.want, len(data))
+				}
+			})
+		}
+	}
+}
+
+// Cancelling the caller's context must not be reported as a read timeout.
+func TestTimeoutReader_CallerCancellationIsNotTimeout(t *testing.T) {
+	for _, tc := range timeoutReaderClientCases() {
+		t.Run(tc.name, func(t *testing.T) {
+			server := startTimeoutReaderServer(t, tc.http2, func(w gohttp.ResponseWriter, r *gohttp.Request) {
+				w.Header().Set("Content-Length", "1048576")
+				w.Write(bytes.Repeat([]byte("a"), 1024))
+				w.(gohttp.Flusher).Flush()
+				<-r.Context().Done()
+			})
+
+			wantProto := 1
+			if tc.http2 {
+				wantProto = 2
+			}
+			// Cancel once the client has read the first bytes, so the next
+			// Read is the one blocked when the caller cancels.
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			client := tc.client(t, server)
+			next := client.Transport
+			client.Transport = roundTripFunc(func(r *gohttp.Request) (*gohttp.Response, error) {
+				resp, err := next.RoundTrip(r)
+				if err == nil {
+					resp.Body = &cancelAfterFirstRead{ReadCloser: resp.Body, cancel: cancel}
+				}
+				return resp, err
+			})
+
+			const timeout = 5 * time.Second
+			start := time.Now()
+			data, err := readBodyWithTimeout(t, ctx, client, server.URL, timeout, wantProto)
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("expected %v, got %v", context.Canceled, err)
+			}
+			if len(data) == 0 {
+				t.Fatal("cancelled before any data was read")
+			}
+			if elapsed := time.Since(start); elapsed >= timeout {
+				t.Fatalf("read returned after %v, not on cancellation", elapsed)
+			}
+		})
+	}
+}
+
+type roundTripFunc func(*gohttp.Request) (*gohttp.Response, error)
+
+func (f roundTripFunc) RoundTrip(r *gohttp.Request) (*gohttp.Response, error) {
+	return f(r)
+}
+
+// cancelAfterFirstRead calls cancel once a Read has returned data.
+type cancelAfterFirstRead struct {
+	io.ReadCloser
+	once   sync.Once
+	cancel context.CancelFunc
+}
+
+func (b *cancelAfterFirstRead) Read(p []byte) (int, error) {
+	n, err := b.ReadCloser.Read(p)
+	if n > 0 {
+		b.once.Do(b.cancel)
+	}
+	return n, err
+}

pkg/api/service.go

@@ -180,11 +180,11 @@ func (s *Service) createTaskBatch(ctx *Context) *Response {
 	if len(request.Reqs) == 0 {
 		return newErrorResponse("param invalid: reqs", restmodel.CodeInvalidParam)
 	}
-	taskIDs, err := s.Downloader.CreateDirectBatch(&request)
+	result, err := s.Downloader.CreateDirectBatchWithResult(&request)
 	if err != nil {
 		return newErrorResponse(err.Error())
 	}
-	return newOKResponse(taskIDs)
+	return newOKResponse(result)
 }
 
 func (s *Service) patchTask(ctx *Context) *Response {

pkg/api/service_test.go

@@ -267,3 +267,65 @@ func TestFullConfigAndJobTriggerShareProtectedProtocolState(t *testing.T) {
 		t.Fatalf("job state overwritten: %s", raw)
 	}
 }
+
+func TestBatchResolvedAndDirectCreation(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Length", "4")
+		_, _ = w.Write([]byte("test"))
+	}))
+	t.Cleanup(upstream.Close)
+	d := download.NewDownloader(&download.DownloaderConfig{StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&httpProtocol.FetcherManager{}}})
+	if err := d.Setup(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { d.Close() })
+	cfg, err := d.GetConfig()
+	if err != nil {
+		t.Fatal(err)
+	}
+	cfg.MaxRunning = 0
+	if err := d.PutConfig(cfg); err != nil {
+		t.Fatal(err)
+	}
+	service, err := NewService(d)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, grouped := range []bool{false, true} {
+		resolved := serviceTestRequest[*download.ResolveResult](t, service, http.MethodPost, "/api/v1/resolve", &restmodel.ResolveTask{Req: &base.Request{URL: upstream.URL + "/resolved"}})
+		result := serviceTestRequest[map[string]json.RawMessage](t, service, http.MethodPost, "/api/v1/tasks/batch", &base.CreateTaskBatch{
+			Group: grouped, GroupName: "folder", Opts: &base.Options{Path: t.TempDir()},
+			Reqs: []*base.CreateTaskBatchItem{
+				{Rid: "invalid"},
+				{Rid: resolved.ID, Req: &base.Request{URL: "unsupported://ignored"}, Opts: &base.Options{SelectFiles: []int{0}}},
+				{Req: &base.Request{URL: upstream.URL + "/direct"}},
+			},
+		})
+		var ids []string
+		if err := json.Unmarshal(result["taskIds"], &ids); err != nil || len(ids) != 2 {
+			t.Fatalf("unexpected task IDs: %+v, %v", result, err)
+		}
+		wantFields := 1
+		if grouped {
+			wantFields = 2
+		}
+		if len(result) != wantFields {
+			t.Fatalf("unexpected response fields: %+v", result)
+		}
+		var parentID string
+		if grouped {
+			if err := json.Unmarshal(result["parentId"], &parentID); err != nil || parentID == "" {
+				t.Fatalf("missing parent ID: %+v", result)
+			}
+		}
+		task := serviceTestRequest[*download.Task](t, service, http.MethodGet, "/api/v1/tasks/"+ids[0], nil)
+		if task.ParentID != parentID || !reflect.DeepEqual(task.Meta.Opts.SelectFiles, []int{0}) {
+			t.Fatalf("resolved task lost membership or selection: %+v", task)
+		}
+		status, response := service.JSON(http.MethodPost, "/api/v1/tasks/batch", nil, []byte(`{"reqs":[{"rid":"missing"},{"req":{"url":"unsupported://bad"}}]}`), nil)
+		var failed restmodel.Result[any]
+		if err := json.Unmarshal(response, &failed); err != nil || status != http.StatusOK || failed.Code == restmodel.CodeOk {
+			t.Fatalf("all-failed batch succeeded: %d %s", status, response)
+		}
+	}
+}

pkg/base/model.go

@@ -27,6 +27,33 @@ type Request struct {
 	SkipVerifyCert bool `json:"skipVerifyCert"`
 }
 
+// SetLabels replaces all labels on the request.
+//
+// Deprecated: In onResolve and onDone, assign request labels directly.
+// In onStart and onError, use task.meta.req.setLabels instead.
+func (r *Request) SetLabels(labels map[string]string) {
+	r.Labels = labels
+}
+
+// PutLabel sets a label on the request.
+//
+// Deprecated: In onResolve and onDone, assign request labels directly.
+// In onStart and onError, use task.meta.req.putLabel instead.
+func (r *Request) PutLabel(key, value string) {
+	if r.Labels == nil {
+		r.Labels = make(map[string]string)
+	}
+	r.Labels[key] = value
+}
+
+// DelLabel deletes a label from the request.
+//
+// Deprecated: In onResolve and onDone, delete request labels directly.
+// In onStart and onError, use task.meta.req.delLabel instead.
+func (r *Request) DelLabel(key string) {
+	delete(r.Labels, key)
+}
+
 func (r *Request) Validate() error {
 	if r.URL == "" {
 		return fmt.Errorf("invalid request url")
@@ -175,9 +202,12 @@ func ParseOptExtra[E any](opts *Options) error {
 type CreateTaskBatch struct {
 	Reqs      []*CreateTaskBatchItem `json:"reqs"`
 	Opts      *Options               `json:"opts"`
+	Group     bool                   `json:"group"`
+	GroupName string                 `json:"groupName"`
 }
 
 type CreateTaskBatchItem struct {
+	Rid  string   `json:"rid,omitempty"`
 	Req  *Request `json:"req"`
 	Opts *Options `json:"opts"`
 }

pkg/download/category.go

@@ -32,6 +32,10 @@ func categoryForResource(cfg *base.DownloaderStoreConfig, res *base.Resource) *b
 // runs on the result, so a category directory the white list rejects falls
 // back to the base directory of this task instead of failing it.
 func (d *Downloader) decideDownloadDir(opts *base.Options, res *base.Resource) (string, error) {
+	return d.decideDownloadDirForTask(opts, res, nil)
+}
+
+func (d *Downloader) decideDownloadDirForTask(opts *base.Options, res *base.Resource, group *TaskGroup) (string, error) {
 	if opts == nil {
 		opts = &base.Options{}
 	}
@@ -48,6 +52,9 @@ func (d *Downloader) decideDownloadDir(opts *base.Options, res *base.Resource) (
 	if base == "" {
 		return "", errors.New("download directory is empty")
 	}
+	if group != nil && opts.SkipAutoCategorize {
+		return d.initDownloadPathForTask(base, group)
+	}
 	target := base
 	if !opts.SkipAutoCategorize {
 		if category := categoryForResource(cfg, res); category != nil {

pkg/download/category_test.go

@@ -234,6 +234,38 @@ func TestDecideDownloadDirWhiteList(t *testing.T) {
 	})
 }
 
+func TestDecideGroupDownloadDirWhiteList(t *testing.T) {
+	parent := t.TempDir()
+	root := filepath.Join(parent, "folder")
+	nested := filepath.Join(root, "nested")
+	for _, test := range []struct {
+		name    string
+		path    string
+		group   *TaskGroup
+		allowed string
+		wantErr bool
+	}{
+		{name: "group root", path: root, group: &TaskGroup{Path: root}, allowed: filepath.Join(parent, "*")},
+		{name: "nested member", path: nested, group: &TaskGroup{Path: root}, allowed: filepath.Join(parent, "*")},
+		{name: "ordinary nested task", path: nested, allowed: filepath.Join(parent, "*"), wantErr: true},
+		{name: "member outside group", path: filepath.Join(parent, "other"), group: &TaskGroup{Path: root}, allowed: filepath.Join(parent, "*"), wantErr: true},
+		{name: "group permission revoked", path: nested, group: &TaskGroup{Path: root}, allowed: nested, wantErr: true},
+		{name: "literal group name", path: filepath.Join(parent, "%year%", "nested"), group: &TaskGroup{Path: filepath.Join(parent, "%year%")}, allowed: filepath.Join(parent, "*")},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			d := newUnitCategoryDownloader(newTestCategoryConfig())
+			d.cfg.WhiteDownloadDirs = []string{test.allowed}
+			got, err := d.decideDownloadDirForTask(&base.Options{Path: test.path, SkipAutoCategorize: true}, singleFileResource("file.txt"), test.group)
+			if (err != nil) != test.wantErr {
+				t.Fatalf("decideDownloadDirForTask() = %q, %v; want error = %v", got, err, test.wantErr)
+			}
+			if err == nil && got != test.path {
+				t.Fatalf("path = %q, want %q", got, test.path)
+			}
+		})
+	}
+}
+
 func TestSuggestedDownloadDir(t *testing.T) {
 	programDir := filepath.Join("./downloads", "Program")
 

pkg/download/extension.go

@@ -825,6 +825,13 @@ func newOnDoneExtensionTask(task *Task) *Task {
 	return cloneExtensionTask(task)
 }
 
+// SetUrl replaces the task request URL.
+//
+// Deprecated: Use task.meta.req.setUrl instead.
+func (t *ExtensionTask) SetUrl(url string) {
+	t.Meta.Req.SetUrl(url)
+}
+
 func (t *OnErrorExtensionTask) Continue() error {
 	return t.download.Continue(&TaskFilter{
 		IDs: []string{t.ID},

pkg/download/extension_compat_test.go

@@ -0,0 +1,89 @@
+package download
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"reflect"
+	"testing"
+
+	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+func TestExtensionDeprecatedRequestAPIs(t *testing.T) {
+	for _, event := range []ActivationEvent{EventOnResolve, EventOnStart, EventOnError, EventOnDone} {
+		for _, withLabels := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/labels=%v", event, withLabels), func(t *testing.T) {
+				d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir()})
+				if err := d.Setup(); err != nil {
+					t.Fatal(err)
+				}
+				defer d.Clear()
+
+				extDir := t.TempDir()
+				manifest := fmt.Sprintf(`{"name":"legacy-request","title":"Legacy request APIs","version":"0.0.1","scripts":[{"event":%q,"match":{"urls":["*://example.com/*"]},"entry":"index.js"}]}`, event)
+				request := "ctx.task.meta.req"
+				urlMutation := ""
+				if event == EventOnResolve {
+					request = "ctx.req"
+				} else if event == EventOnStart || event == EventOnError {
+					urlMutation = `await ctx.task.setUrl('https://example.com/legacy');
+  if (req.url !== 'https://example.com/legacy') throw new Error('legacy URL did not write through');
+  await req.setUrl('https://example.com/new');
+  if (req.url !== 'https://example.com/new') throw new Error('new URL did not write through');
+  await ctx.task.setUrl('https://example.com/final');`
+				}
+				script := fmt.Sprintf(`gopeed.events.%s(async ctx => {
+  const req = %s;
+  await req.delLabel('missing');
+  await req.putLabel('initial', 'yes');
+  if (req.labels.initial !== 'yes') throw new Error('putLabel did not initialize labels');
+  await req.setLabels({keep: 'yes', remove: 'yes'});
+  if (req.labels.initial !== undefined || req.labels.old !== undefined) throw new Error('setLabels did not replace labels');
+  await req.putLabel('added', 'yes');
+  await req.delLabel('remove');
+  if (req.labels.keep !== 'yes' || req.labels.added !== 'yes' || req.labels.remove !== undefined) throw new Error('label mutation failed');
+  %s
+  await req.putLabel('completed', 'yes');
+});`, event, request, urlMutation)
+				for name, content := range map[string]string{"manifest.json": manifest, "index.js": script} {
+					if err := os.WriteFile(filepath.Join(extDir, name), []byte(content), 0600); err != nil {
+						t.Fatal(err)
+					}
+				}
+				if _, err := d.InstallExtensionByFolder(extDir, false); err != nil {
+					t.Fatal(err)
+				}
+				req := &base.Request{URL: "https://example.com/original"}
+				if withLabels {
+					req.Labels = map[string]string{"old": "yes"}
+				}
+				task := &Task{Protocol: "http", Meta: &fetcher.FetcherMeta{Req: req, Opts: &base.Options{}}}
+				switch event {
+				case EventOnResolve:
+					if _, err := d.triggerOnResolve(req); err != nil {
+						t.Fatal(err)
+					}
+				case EventOnStart:
+					d.triggerOnStart(task)
+				case EventOnError:
+					d.triggerOnError(task, errors.New("retry"))
+				case EventOnDone:
+					d.triggerOnDone(task)
+				}
+				if !reflect.DeepEqual(req.Labels, map[string]string{"keep": "yes", "added": "yes", "completed": "yes"}) {
+					t.Fatalf("legacy handler did not update the original request: %#v", req.Labels)
+				}
+				expectedURL := "https://example.com/original"
+				if urlMutation != "" {
+					expectedURL = "https://example.com/final"
+				}
+				if req.URL != expectedURL {
+					t.Fatalf("URL = %q, want %q", req.URL, expectedURL)
+				}
+			})
+		}
+	}
+}

pkg/download/extension_extra_mutation_test.go

@@ -152,7 +152,7 @@ func TestExtensionOnErrorRequestWritesToOriginalTask(t *testing.T) {
    request.setHeaders({Keep:'new'});
    request.putHeader('Authorization','Bearer new');
    request.setMethod('POST');
-   if (typeof task.setUrl !== 'undefined' || typeof task.extra !== 'undefined') throw new Error('legacy task API');
+   if (typeof task.setUrl !== 'function' || typeof task.extra !== 'undefined') throw new Error('unexpected legacy task API');
    request.setUrl('https://example.com/new');
    request.setLabels({keep:'yes',remove:'yes'});
    request.putLabel('new','value');

pkg/download/extension_match_test.go

@@ -0,0 +1,56 @@
+package download
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/GopeedLab/gopeed/pkg/base"
+)
+
+func TestDownloader_Extension_MagnetMatch(t *testing.T) {
+	const magnetURL = "magnet:?xt=urn:btih:0123456789abcdef0123456789abcdef01234567"
+	for _, pattern := range []string{"magnet:*", "magnet:?*"} {
+		t.Run(pattern, func(t *testing.T) {
+			d := NewDownloader(&DownloaderConfig{
+				Storage:    NewMemStorage(),
+				StorageDir: t.TempDir(),
+				TempDir:    t.TempDir(),
+			})
+			if err := d.Setup(); err != nil {
+				t.Fatal(err)
+			}
+			defer d.Clear()
+
+			extDir := t.TempDir()
+			manifest := fmt.Sprintf(`{"name":"magnet-match","author":"test","title":"Magnet match","version":"1.0.0","scripts":[{"event":"onResolve","match":{"urls":[%q]},"entry":"index.js"}]}`, pattern)
+			script := `gopeed.events.onResolve(ctx => {
+  ctx.res = {files: [{name: "resolved.txt", size: 1, req: {url: "https://example.com/resolved.txt"}}]};
+});`
+			for name, content := range map[string]string{"manifest.json": manifest, "index.js": script} {
+				if err := os.WriteFile(filepath.Join(extDir, name), []byte(content), 0600); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if _, err := d.InstallExtensionByFolder(extDir, true); err != nil {
+				t.Fatal(err)
+			}
+
+			for _, req := range []*base.Request{
+				{URL: magnetURL},
+				{URL: "https://example.com/rewritten", RawURL: magnetURL},
+			} {
+				// Exercise extension activation without falling back to a live
+				// BitTorrent metadata lookup when the pattern fails to match.
+				res, err := d.triggerOnResolve(req)
+				if err != nil {
+					t.Fatal(err)
+				}
+				if res == nil || len(res.Files) != 1 || res.Files[0].Name != "resolved.txt" {
+					t.Fatalf("magnet extension did not resolve request %+v: %+v", req, res)
+				}
+			}
+		})
+	}
+}

pkg/download/extension_resolve_size_test.go

@@ -70,10 +70,11 @@ func TestExtensionResolveSizeThroughTaskCreation(t *testing.T) {
 				if err := json.Unmarshal(wire, &submitted); err != nil {
 					t.Fatal(err)
 				}
-				ids, err := d.CreateDirectBatch(&submitted)
+				batchResult, err := d.CreateDirectBatch(&submitted)
 				if err != nil {
 					t.Fatal(err)
 				}
+				ids := batchResult
 				for i, id := range ids {
 					task := d.GetTask(id)
 					if task.Meta.Res != nil {

pkg/download/extension_test.go

@@ -2044,18 +2044,20 @@ func waitForTaskStatus(t *testing.T, downloader *Downloader, id string, status b
 	deadline := time.Now().Add(timeout)
 	for time.Now().Before(deadline) {
 		task := downloader.GetTask(id)
-		if task != nil && task.Status == status {
+		if task != nil && downloader.taskStatus(task) == status {
+			if status != base.DownloadStatusDone {
 				return
 			}
+			if _, watching := downloader.watchedTasks.Load(id); !watching {
+				return
+			}
+		}
 		time.Sleep(10 * time.Millisecond)
 	}
 	task := downloader.GetTask(id)
 	if task == nil {
 		t.Fatalf("timeout waiting for task %s status %s: task not found", id, status)
 	}
-	total := int64(0)
+	snapshot, err := downloader.RuntimeStatus(id)
-	if task.Meta != nil && task.Meta.Res != nil {
+	t.Fatalf("timeout waiting for task %s status %s: got %+v, error=%v", id, status, snapshot, err)
-		total = task.Meta.Res.Size
-	}
-	t.Fatalf("timeout waiting for task %s status %s: got %s downloaded=%d total=%d", id, status, task.Status, task.Progress.Downloaded, total)
 }

pkg/download/ftp_test.go

@@ -0,0 +1,57 @@
+package download
+
+import (
+	"testing"
+
+	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/pkg/base"
+	"github.com/GopeedLab/gopeed/pkg/protocol/ftp"
+	"github.com/GopeedLab/gopeed/pkg/protocol/http"
+)
+
+// TestTorrentNameIsFile checks the default manager order: an FTP URL whose
+// name looks like a torrent or a playlist is still an FTP download.
+func TestTorrentNameIsFile(t *testing.T) {
+	d := NewDownloader(nil)
+	for u, want := range map[string]string{
+		"ftp://h/x.torrent":    "ftp",
+		"ftp://h/v.m3u8":       "ftp",
+		"ftps://h/x.torrent":   "ftp",
+		"ftpes://h/a.bin":      "ftp",
+		"https://h/v.m3u8":     "hls",
+		"https://h/x.torrent":  "http",
+		"magnet:?xt=urn:btih:": "bt",
+	} {
+		fm, err := d.parseFm(u)
+		if err != nil {
+			t.Fatalf("%s: %v", u, err)
+		}
+		if fm.Name() != want {
+			t.Fatalf("%s goes to %q, want %q", u, fm.Name(), want)
+		}
+	}
+}
+
+func TestPostDownloadExtraAcceptsFTP(t *testing.T) {
+	yes := true
+	file := &base.Resource{Files: []*base.FileInfo{{Name: "a.zip"}}}
+	folder := &base.Resource{Name: "dir", Files: []*base.FileInfo{{Name: "a.zip"}}}
+
+	httpExtra := &http.OptsExtra{AutoExtract: &yes}
+	if got := postDownloadExtra(&fetcher.FetcherMeta{Res: file, Opts: &base.Options{Extra: httpExtra}}); got != httpExtra {
+		t.Fatalf("HTTP options must pass through unchanged, got %+v", got)
+	}
+
+	ftpExtra := &ftp.OptsExtra{Connections: 4, AutoTorrent: &yes, DeleteTorrentAfterDownload: &yes, AutoExtract: &yes, ArchivePassword: "pw", DeleteAfterExtract: true}
+	got := postDownloadExtra(&fetcher.FetcherMeta{Res: file, Opts: &base.Options{Extra: ftpExtra}})
+	if got == nil || got.AutoTorrent != &yes || got.DeleteTorrentAfterDownload != &yes || got.AutoExtract != &yes ||
+		got.ArchivePassword != "pw" || !got.DeleteAfterExtract {
+		t.Fatalf("FTP options were not carried over: %+v", got)
+	}
+	if got := postDownloadExtra(&fetcher.FetcherMeta{Res: folder, Opts: &base.Options{Extra: ftpExtra}}); got != nil {
+		t.Fatalf("an FTP folder has no single file to extract, got %+v", got)
+	}
+	if got := postDownloadExtra(&fetcher.FetcherMeta{Res: file, Opts: &base.Options{}}); got != nil {
+		t.Fatalf("no options, got %+v", got)
+	}
+}

pkg/download/model.go

@@ -10,6 +10,7 @@ import (
 	"github.com/GopeedLab/gopeed/internal/production"
 	"github.com/GopeedLab/gopeed/internal/protocol/bt"
 	"github.com/GopeedLab/gopeed/internal/protocol/ed2k"
+	"github.com/GopeedLab/gopeed/internal/protocol/ftp"
 	"github.com/GopeedLab/gopeed/internal/protocol/hls"
 	"github.com/GopeedLab/gopeed/internal/protocol/http"
 	"github.com/GopeedLab/gopeed/pkg/base"
@@ -28,6 +29,20 @@ type ResolveResult struct {
 	SuggestedDownloadDir string `json:"suggestedDownloadDir,omitempty"`
 }
 
+// TaskGroup stores the directory shared by independently scheduled tasks.
+type TaskGroup struct {
+	ID        string    `json:"id"`
+	Name      string    `json:"name"`
+	Path      string    `json:"path"`
+	CreatedAt time.Time `json:"createdAt"`
+}
+
+// CreateTaskBatchResult is shared by ordinary and grouped batches.
+type CreateTaskBatchResult struct {
+	TaskIDs  []string `json:"taskIds"`
+	ParentID string   `json:"parentId,omitempty"`
+}
+
 // Task represents a download unit.
 // Options flow path:
 // REST Request (model.CreateTask/ResolveTask) -> pkg/download Downloader -> Task.Meta.Opts (base.Options) -> Fetcher.Resolve/Start (e.g. internal/protocol/http/fetcher.go)
@@ -35,6 +50,8 @@ type Task struct {
 	ID       string               `json:"id"`
 	Protocol string               `json:"protocol"`
 	Meta     *fetcher.FetcherMeta `json:"meta"`
+	// ParentID identifies the independently stored group.
+	ParentID  string      `json:"parentId,omitempty"`
 	Status    base.Status `json:"status"`
 	Uploading bool        `json:"uploading"`
 	Progress  *Progress   `json:"progress"`
@@ -53,6 +70,8 @@ type Task struct {
 	// does not have it and keeps its directory and name.
 	NeedsPreparation bool `json:"needsPreparation,omitempty"`
 
+	group *TaskGroup
+
 	producer         production.Source
 	producerReceived int64
 
@@ -144,9 +163,11 @@ func (t *Task) MarshalJSON() ([]byte, error) {
 	jsonTask := struct {
 		*rawTaskType
 		Name  string     `json:"name"`
+		Group *TaskGroup `json:"group,omitempty"`
 	}{
 		(*rawTaskType)(t),
 		t.Name(),
+		t.group,
 	}
 	return json.Marshal(jsonTask)
 }
@@ -229,6 +250,10 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
 	}
 	if len(cfg.FetchManagers) == 0 {
 		cfg.FetchManagers = []fetcher.FetcherManager{
+			// FTP matches by URL scheme only. It comes before the managers that
+			// match by file extension, so ftp://host/x.torrent or x.m3u8 is
+			// downloaded as a file over FTP.
+			new(ftp.FetcherManager),
 			new(hls.FetcherManager),
 			new(http.FetcherManager),
 			new(bt.FetcherManager),

pkg/download/testdata/extensions/basic/index.js

@@ -1,5 +1,5 @@
 gopeed.events.onResolve(async function (ctx) {
-    for (const method of ['setUrl', 'setLabels', 'putLabel', 'delLabel', 'setMethod', 'putHeader', 'setTrackers']) {
+    for (const method of ['setUrl', 'setMethod', 'putHeader', 'setTrackers']) {
         if (typeof ctx.req[method] !== 'undefined') throw new Error(`onResolve exposes ${method}`);
     }
     ctx.req.labels = { replaced: "true", removed: "true" };

pkg/mcpserver/output_schema_test.go

@@ -0,0 +1,148 @@
+package mcpserver
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/GopeedLab/gopeed/internal/fetcher"
+	"github.com/GopeedLab/gopeed/pkg/base"
+	"github.com/GopeedLab/gopeed/pkg/download"
+	"github.com/GopeedLab/gopeed/pkg/protocol/bt"
+	"github.com/GopeedLab/gopeed/pkg/protocol/ed2k"
+	phttp "github.com/GopeedLab/gopeed/pkg/protocol/http"
+	"github.com/google/jsonschema-go/jsonschema"
+)
+
+func TestToolSchemaPropertiesAreObjects(t *testing.T) {
+	downloader := download.NewDownloader(&download.DownloaderConfig{Storage: download.NewMemStorage()})
+	req := httptest.NewRequest(http.MethodPost, "http://localhost/mcp", bytes.NewBufferString(
+		`{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}`,
+	))
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "application/json, text/event-stream")
+	recorder := httptest.NewRecorder()
+	NewHandler(downloader).ServeHTTP(recorder, req)
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("tools/list status = %d: %s", recorder.Code, recorder.Body.String())
+	}
+	var response struct {
+		Result struct {
+			Tools []struct {
+				Name         string         `json:"name"`
+				InputSchema  map[string]any `json:"inputSchema"`
+				OutputSchema map[string]any `json:"outputSchema"`
+			} `json:"tools"`
+		} `json:"result"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
+		t.Fatal(err)
+	}
+	if len(response.Result.Tools) == 0 {
+		t.Fatal("tools/list returned no tools")
+	}
+	for _, tool := range response.Result.Tools {
+		t.Run(tool.Name, func(t *testing.T) {
+			checkSchemaProperties(t, tool.InputSchema, "inputSchema")
+			checkSchemaProperties(t, tool.OutputSchema, "outputSchema")
+		})
+	}
+}
+
+// Check the serialized definitions, including nested file requests and union
+// branches. Boolean additionalProperties remains valid; boolean entries inside
+// properties are what clients with object-only schema parsers cannot consume.
+func checkSchemaProperties(t *testing.T, value any, path string) {
+	t.Helper()
+	switch value := value.(type) {
+	case map[string]any:
+		for key, child := range value {
+			if key == "properties" {
+				properties, ok := child.(map[string]any)
+				if !ok {
+					t.Fatalf("%s.properties = %#v, want object", path, child)
+				}
+				for name, property := range properties {
+					if _, ok := property.(map[string]any); !ok {
+						t.Errorf("%s.properties.%s = %#v, want schema object", path, name, property)
+					}
+				}
+			}
+			checkSchemaProperties(t, child, path+"."+key)
+		}
+	case []any:
+		for i, child := range value {
+			checkSchemaProperties(t, child, fmt.Sprintf("%s[%d]", path, i))
+		}
+	}
+}
+
+func TestToolOutputSchemasPreserveDynamicValues(t *testing.T) {
+	statsSchema := toolOutputSchema[getTaskStatsOutput]()
+	for _, test := range []struct {
+		name  string
+		stats any
+	}{
+		{"unavailable", nil},
+		{"typed nil", (*fetcher.Stats)(nil)},
+		{"http", &fetcher.Stats{Snapshot: &phttp.Stats{Connections: []*phttp.StatsConnection{{Downloaded: 123}}}}},
+		{"bittorrent", &fetcher.Stats{Snapshot: &bt.StatsSnapshot{}, Runtime: &bt.StatsRuntime{}}},
+		{"ed2k", &fetcher.Stats{Snapshot: &ed2k.StatsSnapshot{}, Runtime: &ed2k.StatsRuntime{}}},
+		{"restored snapshot", &fetcher.Stats{Snapshot: json.RawMessage(`{"segmentsTotal":3,"segmentsDone":2,"segmentsFailed":0}`)}},
+		{"empty object", map[string]any{}},
+		{"nested JSON", map[string]any{"values": []any{nil, true, "text", 1.5, map[string]any{"key": "value"}}}},
+		{"array", []any{1, "two", false}},
+		{"string", "protocol value"},
+		{"number", 1.5},
+		{"boolean", true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			validateToolOutput(t, statsSchema, &getTaskStatsOutput{Stats: test.stats})
+		})
+	}
+
+	// Output extras can be empty, null, or extension-specific. In particular,
+	// they need not satisfy the required settings in the download input schema.
+	for _, extra := range []any{nil, map[string]any{}, &phttp.ReqExtra{}, &bt.ReqExtra{},
+		map[string]any{"custom": []any{true, nil, 42}}, "custom value"} {
+		req := &base.Request{URL: "https://example.com/file", Extra: extra}
+		res := &base.Resource{Files: []*base.FileInfo{{Name: "file", Req: req}}}
+		validateToolOutput(t, toolOutputSchema[resolveTaskOutput](), &resolveTaskOutput{Res: res})
+		validateToolOutput(t, toolOutputSchema[getTaskOutput](), &getTaskOutput{Task: taskDetails{
+			Req: req, Res: res, Opts: &base.Options{Extra: extra},
+		}})
+	}
+
+	// Supplying an explicit schema must retain validation of the output envelope.
+	resolved, err := statsSchema.Resolve(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, invalid := range []map[string]any{{}, {"stats": nil, "unexpected": true}} {
+		if err := resolved.Validate(invalid); err == nil {
+			t.Errorf("accepted invalid output envelope: %#v", invalid)
+		}
+	}
+}
+
+func validateToolOutput(t *testing.T, schema *jsonschema.Schema, output any) {
+	t.Helper()
+	data, err := json.Marshal(output)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var value any
+	if err := json.Unmarshal(data, &value); err != nil {
+		t.Fatal(err)
+	}
+	resolved, err := schema.Resolve(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := resolved.Validate(value); err != nil {
+		t.Fatalf("output %s does not match schema: %v", data, err)
+	}
+}