fix

apps/api/internal/processing/openai.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
 	"math"
 	"math/rand"
 	"net"
@@ -39,6 +40,16 @@ type OpenAIClient struct {
 
 const maxOpenAIRetries = 8
 
+// openAIHTTPTimeout is the full-request limit for OpenAI-compatible chat/embeddings.
+// Reasoning models (e.g. code-fast) often need 60–180s; 240s reduces false timeouts.
+// Timeouts are not retried (see classifyOpenAITransportErr) so a hung upstream fails
+// the item in ~4m instead of MaxRetries×240s of silent PROCESSING.
+const openAIHTTPTimeout = 240 * time.Second
+
+// openAIWaitLogInterval emits progress while HTTPClient.Do blocks (cancel is only
+// checked between job items, so mid-call silence looks like a stuck job).
+const openAIWaitLogInterval = 30 * time.Second
+
 func NewOpenAIClient(apiKey, baseURL, model string, rpm, maxRetries int) *OpenAIClient {
 	if baseURL == "" {
 		baseURL = "https://api.openai.com/v1"
@@ -57,14 +68,12 @@ func NewOpenAIClient(apiKey, baseURL, model string, rpm, maxRetries int) *OpenAI
 	}
 	// Dial-time SSRF. Loopback when base URL is local; RFC1918 only in non-prod.
 	policy := openAIDialPolicy(baseURL)
-	// Header/body timeout: code-fast often exceeds 180s; 240s reduces false timeouts.
-	// Tradeoff: slower fail on hung upstream. Prefer synthesize fallback over infinite wait
-	// (RunSteps always synthesizes on timeout/error/empty — do not raise unboundedly).
+	// Prefer synthesize fallback on timeout/error/empty (RunSteps) over raising this unboundedly.
 	return &OpenAIClient{
 		APIKey:      apiKey,
 		BaseURL:     strings.TrimRight(baseURL, "/"),
 		Model:       model,
-		HTTPClient:  security.SafeHTTPClientPolicy(240*time.Second, policy),
+		HTTPClient:  security.SafeHTTPClientPolicy(openAIHTTPTimeout, policy),
 		MinInterval: interval,
 		MaxRetries:  maxRetries,
 		ModeLabel:   AIProviderInternal,
@@ -141,6 +150,79 @@ func IsMockOrLoopbackBaseURL(baseURL string) bool {
 	return strings.Contains(lower, "mock-llm")
 }
 
+// startWaitLogger logs request start and periodic progress while Do blocks.
+// Call the returned stop with the Do error (or nil) when the request finishes.
+func (c *OpenAIClient) startWaitLogger(op string, maxTokens int) func(error) {
+	if c == nil {
+		return func(error) {}
+	}
+	start := time.Now()
+	log.Printf("openai: request start op=%s model=%s base=%s max_tokens=%d timeout=%s",
+		op, c.Model, c.BaseURL, maxTokens, openAIHTTPTimeout)
+	done := make(chan struct{})
+	var once sync.Once
+	go func() {
+		t := time.NewTicker(openAIWaitLogInterval)
+		defer t.Stop()
+		for {
+			select {
+			case <-done:
+				return
+			case <-t.C:
+				log.Printf("openai: still waiting op=%s model=%s base=%s elapsed=%s timeout=%s",
+					op, c.Model, c.BaseURL, time.Since(start).Round(time.Second), openAIHTTPTimeout)
+			}
+		}
+	}()
+	return func(err error) {
+		once.Do(func() {
+			close(done)
+			elapsed := time.Since(start).Round(time.Millisecond)
+			if err != nil {
+				log.Printf("openai: request done op=%s model=%s elapsed=%s err=%s",
+					op, c.Model, elapsed, TruncateError(err))
+				return
+			}
+			log.Printf("openai: request done op=%s model=%s elapsed=%s ok=1", op, c.Model, elapsed)
+		})
+	}
+}
+
+// isOpenAITimeoutErr reports HTTP client / context deadline timeouts.
+// These must not be retried: each attempt costs openAIHTTPTimeout and leaves the job looking stuck.
+func isOpenAITimeoutErr(err error) bool {
+	if err == nil {
+		return false
+	}
+	if errors.Is(err, context.DeadlineExceeded) {
+		return true
+	}
+	var ne net.Error
+	if errors.As(err, &ne) && ne.Timeout() {
+		return true
+	}
+	s := err.Error()
+	return strings.Contains(s, "Client.Timeout exceeded") ||
+		strings.Contains(s, "context deadline exceeded")
+}
+
+// classifyOpenAITransportErr maps Do/Read errors to (wrappedErr, retryable).
+// Timeouts and cancel are fatal for this attempt chain; other network errors may retry.
+func classifyOpenAITransportErr(model string, err error) (error, bool) {
+	if err == nil {
+		return nil, false
+	}
+	if errors.Is(err, context.Canceled) {
+		return err, false
+	}
+	if isOpenAITimeoutErr(err) {
+		return fmt.Errorf(
+			"openai timed out waiting for model %q after %s (not retrying; use a faster model or fix upstream latency): %w",
+			model, openAIHTTPTimeout, err), false
+	}
+	return err, true
+}
+
 type chatRequest struct {
 	Model       string        `json:"model"`
 	Messages    []chatMessage `json:"messages"`
@@ -218,9 +300,18 @@ func (c *OpenAIClient) CompleteWithOptions(ctx context.Context, system, user str
 			return comp, nil
 		}
 		lastErr = err
-		if errors.Is(err, errEmptyModelResponse) && maxTok > 0 && maxTok < maxTokensReasoningBudget {
-			maxTok = maxTokensReasoningBudget
-			retryable = true
+		if errors.Is(err, errEmptyModelResponse) {
+			if maxTok <= 0 || maxTok < maxTokensReasoningBudget {
+				if maxTok < maxTokensReasoningBudget {
+					maxTok = maxTokensReasoningBudget
+				}
+				retryable = true
+			} else {
+				// Already at reasoning budget — another 240s call will not help.
+				return Completion{}, fmt.Errorf(
+					"openai empty response at max_tokens=%d for model %q (try a faster non-reasoning model): %w",
+					maxTok, c.Model, err)
+			}
 		}
 		if !retryable {
 			return Completion{}, err
@@ -325,14 +416,18 @@ func (c *OpenAIClient) doEmbed(ctx context.Context, texts []string) ([][]float32
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Authorization", "Bearer "+c.APIKey)
 
+	stopWait := c.startWaitLogger("embeddings", 0)
 	res, err := c.HTTPClient.Do(req)
+	stopWait(err)
 	if err != nil {
-		return nil, true, err
+		wrapped, retryable := classifyOpenAITransportErr(c.Model, err)
+		return nil, retryable, wrapped
 	}
 	defer res.Body.Close()
 	raw, err := io.ReadAll(io.LimitReader(res.Body, 4<<20))
 	if err != nil {
-		return nil, true, err
+		wrapped, retryable := classifyOpenAITransportErr(c.Model, err)
+		return nil, retryable, wrapped
 	}
 	var parsed embeddingResponse
 	if err := json.Unmarshal(raw, &parsed); err != nil {
@@ -391,14 +486,18 @@ func (c *OpenAIClient) doComplete(ctx context.Context, system, user string, temp
 	req.Header.Set("Content-Type", "application/json")
 	req.Header.Set("Authorization", "Bearer "+c.APIKey)
 
+	stopWait := c.startWaitLogger("chat/completions", maxTokens)
 	res, err := c.HTTPClient.Do(req)
+	stopWait(err)
 	if err != nil {
-		return Completion{}, true, err
+		wrapped, retryable := classifyOpenAITransportErr(c.Model, err)
+		return Completion{}, retryable, wrapped
 	}
 	defer res.Body.Close()
 	raw, err := io.ReadAll(io.LimitReader(res.Body, 1<<20))
 	if err != nil {
-		return Completion{}, true, err
+		wrapped, retryable := classifyOpenAITransportErr(c.Model, err)
+		return Completion{}, retryable, wrapped
 	}
 	var parsed chatResponse
 	if err := json.Unmarshal(raw, &parsed); err != nil {
@@ -426,7 +525,9 @@ func (c *OpenAIClient) doComplete(ctx context.Context, system, user string, temp
 	}
 	if text == "" {
 		// Reasoning models often return empty content when max_tokens cuts mid-thought.
-		retryable := strings.EqualFold(finishReason, "length")
+		// Only retry when CompleteWithOptions can still raise max_tokens.
+		canBump := maxTokens <= 0 || maxTokens < maxTokensReasoningBudget
+		retryable := strings.EqualFold(finishReason, "length") && canBump
 		return Completion{}, retryable, errEmptyModelResponse
 	}
 	return Completion{

apps/api/internal/processing/openai_test.go

@@ -3,6 +3,7 @@ package processing
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"strings"
@@ -226,3 +227,86 @@ func TestOpenAIClient_doComplete_emptyContentLengthRetriesWithBudget(t *testing.
 		t.Fatalf("text=%q", comp.Text)
 	}
 }
+
+func TestOpenAIClient_Complete_timeoutNotRetried(t *testing.T) {
+	calls := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		deadline := time.After(300 * time.Millisecond)
+		for {
+			select {
+			case <-r.Context().Done():
+				return
+			case <-deadline:
+				_ = json.NewEncoder(w).Encode(map[string]any{
+					"choices": []map[string]any{
+						{"message": map[string]any{"content": "late"}},
+					},
+				})
+				return
+			case <-time.After(10 * time.Millisecond):
+			}
+		}
+	}))
+	defer srv.Close()
+	c := NewOpenAIClient("test-key", srv.URL, "code-fast", 0, 3)
+	c.MaxRetries = 3
+	c.HTTPClient = &http.Client{Timeout: 60 * time.Millisecond}
+	start := time.Now()
+	_, err := c.Complete(context.Background(), "sys", "user")
+	elapsed := time.Since(start)
+	if err == nil {
+		t.Fatal("expected timeout error")
+	}
+	if !strings.Contains(err.Error(), "timed out waiting for model") {
+		t.Fatalf("want clear timeout message, got: %v", err)
+	}
+	if calls != 1 {
+		t.Fatalf("calls=%d want 1 (timeout must not retry)", calls)
+	}
+	if elapsed > time.Second {
+		t.Fatalf("timeout too slow: %s", elapsed)
+	}
+}
+
+func TestOpenAIClient_Complete_emptyAtReasoningBudgetNotRetried(t *testing.T) {
+	t.Parallel()
+	calls := 0
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		calls++
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"model": "code-fast",
+			"choices": []map[string]any{{
+				"finish_reason": "length",
+				"message":       map[string]any{"role": "assistant", "content": "", "reasoning_content": "still thinking"},
+			}},
+			"usage": map[string]int{"prompt_tokens": 1, "completion_tokens": 4096, "total_tokens": 4097},
+		})
+	}))
+	defer srv.Close()
+	c := NewOpenAIClient("test-key", srv.URL, "code-fast", 0, 3)
+	c.HTTPClient = srv.Client()
+	_, err := c.CompleteWithOptions(context.Background(), "sys", "user", CompleteOptions{MaxTokens: maxTokensReasoningBudget})
+	if err == nil {
+		t.Fatal("expected empty response error")
+	}
+	if !strings.Contains(err.Error(), "empty response") {
+		t.Fatalf("err=%v", err)
+	}
+	if calls != 1 {
+		t.Fatalf("calls=%d want 1 (no retry at reasoning budget)", calls)
+	}
+}
+
+func TestIsOpenAITimeoutErr(t *testing.T) {
+	t.Parallel()
+	if !isOpenAITimeoutErr(context.DeadlineExceeded) {
+		t.Fatal("DeadlineExceeded")
+	}
+	if !isOpenAITimeoutErr(fmt.Errorf("Get \"http://x\": context deadline exceeded (Client.Timeout exceeded while awaiting headers)")) {
+		t.Fatal("Client.Timeout message")
+	}
+	if isOpenAITimeoutErr(fmt.Errorf("connection refused")) {
+		t.Fatal("connection refused should not be timeout")
+	}
+}

apps/api/internal/processing/pipeline.go

@@ -602,6 +602,10 @@ func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error {
 		return nil
 	}
 
+	// Log before AI resolve so a hung provider lookup still leaves a start line.
+	log.Printf("processing: start job=%s company=%s type=%s prior_processed=%d",
+		jobID, companyID, processingType, alreadyProcessed)
+
 	modeLabel := AIProviderInternal
 	usingBYOK := false
 	jobEngine := p.Engine
@@ -625,9 +629,7 @@ func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error {
 	} else if label := jobEngine.EngineProviderMode(); label != "" {
 		modeLabel = label
 	}
-
-	log.Printf("processing: start job=%s company=%s type=%s mode=%s prior_processed=%d",
-		jobID, companyID, processingType, modeLabel, alreadyProcessed)
+	log.Printf("processing: provider job=%s mode=%s byok=%v", jobID, modeLabel, usingBYOK)
 
 	progress := InitialStepProgress(processingType)
 	if len(progress) > 0 {

apps/api/internal/processing/stuck_cleanup.go

@@ -26,8 +26,9 @@ type OrphanReclaimResult struct {
 }
 
 // ReclaimOrphanedRunning resets in-flight work left by a dead worker process so
-// ClaimNext / feed sync claim can pick it up again. Call only when the processing
-// worker heartbeat is missing or stale (single-worker architecture) — never while
+// ClaimNext / feed sync claim can pick it up again. Call on every worker process
+// start (before TouchHeartbeat) under the single-worker architecture — a live
+// heartbeat from a just-killed process must not skip reclaim. Never call while
 // another live worker may own the rows.
 //
 // Unlike CleanupStuck (age-gated fail), this requeues immediately: running → pending