feat(cli): add configurable workers retries and fail-fast mode

Author: Congregalis

internal/cli/run.go

@@ -28,6 +28,7 @@ import (
 const (
 	defaultChunkSize        = 2000
 	defaultTranslateWorkers = 4
+	defaultMaxRetries       = 5
 	defaultOutDir           = "out"
 	summaryFileName         = "_summary.json"
 	stateFileName           = ".transblog.state.json"
@@ -43,6 +44,9 @@ type options struct {
 	Model      string
 	OutPath    string
 	View       bool
+	Workers    int
+	MaxRetries int
+	FailFast   bool
 	Glossary   string
 	Timeout    time.Duration
 	ChunkSize  int
@@ -173,7 +177,7 @@ func Run(args []string, stdout io.Writer, stderr io.Writer) error {
 		return err
 	}
 
-	openAIClient := openai.NewClient(apiKey, baseURL, httpClient)
+	openAIClient := openai.NewClient(apiKey, baseURL, httpClient, opts.MaxRetries)
 	runCtx, stopSignal := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
 	defer stopSignal()
 	runStart := time.Now()
@@ -200,6 +204,10 @@ func Run(args []string, stdout io.Writer, stderr io.Writer) error {
 			summary.FailureCount++
 			summary.Results = append(summary.Results, item)
 			_, _ = fmt.Fprintf(stderr, "Failed [%s]: %s (%s)\n", errorType, compactURL(sourceURL), errorMessage)
+			if opts.FailFast {
+				_, _ = fmt.Fprintln(stderr, "Fail-fast enabled: stop after first failure.")
+				break
+			}
 			continue
 		}
 
@@ -245,6 +253,9 @@ func parseFlags(args []string, stderr io.Writer) (options, error) {
 	fs.StringVar(&opts.Model, "model", "gpt-5.2", "OpenAI model name")
 	fs.StringVar(&opts.OutPath, "out", "", "Output path: file for single URL, directory for multiple URLs (default: ./out/)")
 	fs.BoolVar(&opts.View, "view", false, "Generate HTML split-view with Markdown rendering and synchronized scrolling")
+	fs.IntVar(&opts.Workers, "workers", defaultTranslateWorkers, "Translation worker count")
+	fs.IntVar(&opts.MaxRetries, "max-retries", defaultMaxRetries, "Maximum retries for OpenAI requests")
+	fs.BoolVar(&opts.FailFast, "fail-fast", false, "Stop at first URL failure (default: continue for partial success)")
 	fs.StringVar(&opts.Glossary, "glossary", "", "Path to glossary JSON map, e.g. {\"term\":\"translation\"}")
 	fs.DurationVar(&opts.Timeout, "timeout", 90*time.Second, "HTTP timeout, e.g. 120s")
 	fs.IntVar(&opts.ChunkSize, "chunk-size", defaultChunkSize, "Target chunk size in characters")
@@ -268,6 +279,12 @@ func parseFlags(args []string, stderr io.Writer) (options, error) {
 	if opts.Timeout <= 0 {
 		return options{}, errors.New("--timeout must be positive")
 	}
+	if opts.Workers <= 0 {
+		return options{}, errors.New("--workers must be greater than 0")
+	}
+	if opts.MaxRetries < 0 {
+		return options{}, errors.New("--max-retries must be 0 or greater")
+	}
 	if opts.ChunkSize <= 0 {
 		opts.ChunkSize = defaultChunkSize
 	}
@@ -561,6 +578,8 @@ func processURL(
 			glossaryMap,
 			tasks,
 			progress,
+			opts.Workers,
+			opts.FailFast,
 			func(task translationTask, translated string) error {
 				if err := stateStore.saveChunk(
 					sourceURL,
@@ -754,6 +773,8 @@ func translateAllChunks(
 	glossaryMap map[string]string,
 	tasks []translationTask,
 	progress io.Writer,
+	workers int,
+	failFast bool,
 	onChunkTranslated func(task translationTask, translated string) error,
 ) ([]pair, error) {
 	if len(tasks) == 0 {
@@ -763,7 +784,7 @@ func translateAllChunks(
 		progress = io.Discard
 	}
 
-	workerCount := defaultTranslateWorkers
+	workerCount := workers
 	if workerCount > len(tasks) {
 		workerCount = len(tasks)
 	}
@@ -795,8 +816,11 @@ func translateAllChunks(
 					case results <- translationResult{task: task, err: fmt.Errorf("translate chunk %d for %s: %w", task.chunkNumber, task.sourceURL, err)}:
 					case <-runCtx.Done():
 					}
-					cancel()
-					return
+					if failFast {
+						cancel()
+						return
+					}
+					continue
 				}
 
 				translated = glossary.Apply(translated, glossaryMap)

internal/cli/run_test.go

@@ -13,6 +13,7 @@ import (
 	"strings"
 	"sync/atomic"
 	"testing"
+	"time"
 )
 
 func TestRunMultiURLWritesPerURLOutputsAndSummary(t *testing.T) {
@@ -345,6 +346,223 @@ func TestRunSummaryIncludesOutputErrorType(t *testing.T) {
 	}
 }
 
+func TestRunFailFastStopsAfterFirstFailure(t *testing.T) {
+	contentServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		switch r.URL.Path {
+		case "/bad":
+			http.Error(w, "broken", http.StatusInternalServerError)
+		case "/ok":
+			w.Header().Set("Content-Type", "text/html; charset=utf-8")
+			_, _ = w.Write([]byte(sampleArticle("OK", "good content")))
+		default:
+			http.NotFound(w, r)
+		}
+	}))
+	t.Cleanup(contentServer.Close)
+
+	openAIServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/responses" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"output_text":"译文内容"}`)
+	}))
+	t.Cleanup(openAIServer.Close)
+
+	tmpDir := useTempWorkingDir(t)
+	t.Setenv("OPENAI_API_KEY", "test-key")
+	t.Setenv("OPENAI_BASE_URL", openAIServer.URL)
+
+	badURL := contentServer.URL + "/bad"
+	okURL := contentServer.URL + "/ok"
+
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	err := Run([]string{"--fail-fast", badURL, okURL}, &stdout, &stderr)
+	if err == nil {
+		t.Fatalf("Run() error = nil, want fail-fast error")
+	}
+
+	summaryPath := filepath.Join(tmpDir, "out", "_summary.json")
+	rawSummary, err := os.ReadFile(summaryPath)
+	if err != nil {
+		t.Fatalf("read summary file: %v", err)
+	}
+
+	var summary taskSummary
+	if err := json.Unmarshal(rawSummary, &summary); err != nil {
+		t.Fatalf("unmarshal summary JSON: %v", err)
+	}
+
+	if len(summary.Results) != 1 {
+		t.Fatalf("summary result len=%d, want 1 due to fail-fast stop", len(summary.Results))
+	}
+	if summary.Results[0].SourceURL != badURL {
+		t.Fatalf("summary first source_url=%q, want %q", summary.Results[0].SourceURL, badURL)
+	}
+	if !strings.Contains(stderr.String(), "Fail-fast enabled: stop after first failure.") {
+		t.Fatalf("stderr missing fail-fast message: %s", stderr.String())
+	}
+	if !strings.Contains(stdout.String(), "Done: 0 succeeded, 1 failed") {
+		t.Fatalf("stdout missing final summary: %s", stdout.String())
+	}
+}
+
+func TestRunMaxRetriesFlagControlsOpenAIRetry(t *testing.T) {
+	contentServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/retry" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "text/html; charset=utf-8")
+		_, _ = w.Write([]byte(sampleArticle("Retry", "retry content")))
+	}))
+	t.Cleanup(contentServer.Close)
+
+	var callCount int32
+	openAIServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/responses" {
+			http.NotFound(w, r)
+			return
+		}
+
+		n := atomic.AddInt32(&callCount, 1)
+		if n == 1 {
+			w.WriteHeader(http.StatusInternalServerError)
+			_, _ = io.WriteString(w, `{"error":{"message":"temporary upstream failure"}}`)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"output_text":"译文内容"}`)
+	}))
+	t.Cleanup(openAIServer.Close)
+
+	t.Setenv("OPENAI_API_KEY", "test-key")
+	t.Setenv("OPENAI_BASE_URL", openAIServer.URL)
+	sourceURL := contentServer.URL + "/retry"
+
+	dirNoRetry := t.TempDir()
+	runInWorkingDir(t, dirNoRetry, func() string {
+		var stdout bytes.Buffer
+		var stderr bytes.Buffer
+		atomic.StoreInt32(&callCount, 0)
+
+		err := Run([]string{"--chunk-size", "10000", "--max-retries", "0", sourceURL}, &stdout, &stderr)
+		if err == nil {
+			t.Fatalf("Run() error = nil, want failure when retries are disabled")
+		}
+		if got := atomic.LoadInt32(&callCount); got != 1 {
+			t.Fatalf("OpenAI call count=%d, want 1 with --max-retries=0", got)
+		}
+		return ""
+	})
+
+	dirWithRetry := t.TempDir()
+	runInWorkingDir(t, dirWithRetry, func() string {
+		var stdout bytes.Buffer
+		var stderr bytes.Buffer
+		atomic.StoreInt32(&callCount, 0)
+
+		if err := Run([]string{"--chunk-size", "10000", "--max-retries", "1", sourceURL}, &stdout, &stderr); err != nil {
+			t.Fatalf("Run() error = %v; stderr=%s", err, stderr.String())
+		}
+		if got := atomic.LoadInt32(&callCount); got != 2 {
+			t.Fatalf("OpenAI call count=%d, want 2 with --max-retries=1", got)
+		}
+		return ""
+	})
+}
+
+func TestRunWorkersFlagChangesConcurrency(t *testing.T) {
+	contentServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/parallel" {
+			http.NotFound(w, r)
+			return
+		}
+		w.Header().Set("Content-Type", "text/html; charset=utf-8")
+		_, _ = w.Write([]byte(sampleLongArticle("Parallel")))
+	}))
+	t.Cleanup(contentServer.Close)
+
+	var inFlight int32
+	var maxInFlight int32
+	openAIServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/responses" {
+			http.NotFound(w, r)
+			return
+		}
+
+		current := atomic.AddInt32(&inFlight, 1)
+		for {
+			prev := atomic.LoadInt32(&maxInFlight)
+			if current <= prev {
+				break
+			}
+			if atomic.CompareAndSwapInt32(&maxInFlight, prev, current) {
+				break
+			}
+		}
+		time.Sleep(25 * time.Millisecond)
+		atomic.AddInt32(&inFlight, -1)
+
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"output_text":"译文内容"}`)
+	}))
+	t.Cleanup(openAIServer.Close)
+
+	t.Setenv("OPENAI_API_KEY", "test-key")
+	t.Setenv("OPENAI_BASE_URL", openAIServer.URL)
+	sourceURL := contentServer.URL + "/parallel"
+
+	singleWorkerDir := t.TempDir()
+	runInWorkingDir(t, singleWorkerDir, func() string {
+		var stdout bytes.Buffer
+		var stderr bytes.Buffer
+		atomic.StoreInt32(&inFlight, 0)
+		atomic.StoreInt32(&maxInFlight, 0)
+
+		if err := Run([]string{"--chunk-size", "80", "--workers", "1", sourceURL}, &stdout, &stderr); err != nil {
+			t.Fatalf("Run() with --workers=1 error = %v; stderr=%s", err, stderr.String())
+		}
+		if got := atomic.LoadInt32(&maxInFlight); got != 1 {
+			t.Fatalf("max in-flight=%d, want 1 when workers=1", got)
+		}
+		return ""
+	})
+
+	multiWorkerDir := t.TempDir()
+	runInWorkingDir(t, multiWorkerDir, func() string {
+		var stdout bytes.Buffer
+		var stderr bytes.Buffer
+		atomic.StoreInt32(&inFlight, 0)
+		atomic.StoreInt32(&maxInFlight, 0)
+
+		if err := Run([]string{"--chunk-size", "80", "--workers", "4", sourceURL}, &stdout, &stderr); err != nil {
+			t.Fatalf("Run() with --workers=4 error = %v; stderr=%s", err, stderr.String())
+		}
+		if got := atomic.LoadInt32(&maxInFlight); got <= 1 {
+			t.Fatalf("max in-flight=%d, want >1 when workers=4", got)
+		}
+		return ""
+	})
+}
+
+func TestParseFlagsRejectsInvalidWorkers(t *testing.T) {
+	_, err := parseFlags([]string{"--workers", "0", "https://example.com"}, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "--workers must be greater than 0") {
+		t.Fatalf("parseFlags error=%v, want workers validation error", err)
+	}
+}
+
+func TestParseFlagsRejectsInvalidMaxRetries(t *testing.T) {
+	_, err := parseFlags([]string{"--max-retries", "-1", "https://example.com"}, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "--max-retries must be 0 or greater") {
+		t.Fatalf("parseFlags error=%v, want max-retries validation error", err)
+	}
+}
+
 func TestRunResumeReusesSavedChunksAndMatchesSingleRun(t *testing.T) {
 	contentServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if r.URL.Path != "/long" {

internal/openai/responses.go

@@ -17,30 +17,35 @@ import (
 )
 
 const (
-	defaultBaseURL = "https://api.openai.com/v1"
-	maxErrBody     = 2048
-	maxRetries     = 5
+	defaultBaseURL    = "https://api.openai.com/v1"
+	maxErrBody        = 2048
+	defaultMaxRetries = 5
 )
 
 type Client struct {
 	apiKey     string
 	endpoint   string
 	httpClient *http.Client
+	maxRetries int
 }
 
-func NewClient(apiKey string, baseURL string, httpClient *http.Client) *Client {
+func NewClient(apiKey string, baseURL string, httpClient *http.Client, maxRetries int) *Client {
 	if strings.TrimSpace(baseURL) == "" {
 		baseURL = defaultBaseURL
 	}
 	baseURL = strings.TrimSuffix(strings.TrimSpace(baseURL), "/")
 	if strings.HasSuffix(baseURL, "/v1") {
 		baseURL = strings.TrimSuffix(baseURL, "/v1")
 	}
+	if maxRetries < 0 {
+		maxRetries = defaultMaxRetries
+	}
 
 	return &Client{
 		apiKey:     apiKey,
 		endpoint:   baseURL + "/v1/responses",
 		httpClient: httpClient,
+		maxRetries: maxRetries,
 	}
 }
 
@@ -88,14 +93,14 @@ func (c *Client) TranslateMarkdownChunk(ctx context.Context, model string, mdChu
 	}
 
 	var lastErr error
-	for attempt := 0; attempt <= maxRetries; attempt++ {
+	for attempt := 0; attempt <= c.maxRetries; attempt++ {
 		translated, retry, err := c.callResponses(ctx, body)
 		if err == nil {
 			return translated, nil
 		}
 
 		lastErr = err
-		if !retry || attempt == maxRetries {
+		if !retry || attempt == c.maxRetries {
 			break
 		}