test: heartbeats are stamped with the application clock, not the database's NOW() (#51), red

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

cmd/tapir/uploads_test.go

@@ -44,7 +44,7 @@ func (f *fakeUploadsStore) SaveProgress(context.Context, string, string, []domai
 	return nil
 }
 
-func (f *fakeUploadsStore) Heartbeat(context.Context, string, string) error { return nil }
+func (f *fakeUploadsStore) Heartbeat(context.Context, string, string, time.Time) error { return nil }
 
 func (f *fakeUploadsStore) FailInterruptedUploads(context.Context) (int, error) {
 	f.sweeps++

docs/homelab-integration.md

@@ -14,7 +14,8 @@ it** — endpoints and aliases drift, and this file is a snapshot (2026-06-06),
     not LiteLLM. Use `30401` for off-cluster LiteLLM. `confirm` resolved.
 - **Speech-to-text (ADR-033):** the same gateway serves `POST /v1/audio/transcriptions` for
   `iguana/kb-whisper` (primary, local) and `berget/kb-whisper` (fallback, EU cloud — private audio
-  leaves the homelab if it is used). **Call it via the NodePort / in-cluster address above, NOT
+  leaves the homelab if it is used: when the primary fails **or returns an empty transcript**,
+  e.g. a silent or music-only file; the transcript page then says so, #52). **Call it via the NodePort / in-cluster address above, NOT
   `https://llm-api.d-ma.be`:** that public ingress (openresty) returns `413` for bodies over
   roughly 8–10 MB (~5 minutes of 16 kHz WAV), while the NodePort took 174 MB (90 minutes) in one
   request (measured 2026-10-01, `docs/research/audio-length-ceiling-2026-10-01.md`). Send multipart

docs/use-cases/audio_upload.feature

@@ -240,6 +240,11 @@ Feature: Upload an audio recording, get a private Swedish transcript, discuss it
     Then it is transcribed anyway and its audio step says how long the audio really is
     And a complete recording whose header over-estimated its length is not refused
 
+  Scenario: A transcript made by the cloud fallback says the audio left the homelab
+    When the transcript was produced by an endpoint that is not local
+    Then its page says in plain words that the audio left the homelab and names the provider
+    And a locally transcribed recording carries no such note
+
   Scenario: A full queue is a retry, not an error
     When too many recordings are already waiting
     Then the upload is answered as busy, its row is failed honestly and its audio is deleted

internal/adapters/llm/bodycap_test.go

@@ -0,0 +1,81 @@
+package llm
+
+// Observes how much of a gateway response the client will read. The body is read before
+// the status is looked at, so a misbehaving gateway or proxy that streams forever was
+// bounded only by the client's overall timeout (#53).
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// endlessGateway answers with the given status and streams filler until the client hangs
+// up; wrote reports how many bytes it managed to send.
+func endlessGateway(t *testing.T, status int, wrote *atomic.Int64) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(status)
+		chunk := []byte(strings.Repeat("a", 4096))
+		for {
+			n, err := w.Write(chunk)
+			wrote.Add(int64(n))
+			if err != nil {
+				return
+			}
+			if f, ok := w.(http.Flusher); ok {
+				f.Flush()
+			}
+		}
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func TestClient_ASuccessBodyLargerThanTheCapIsAnErrorAndTheClientStopsReading(t *testing.T) {
+	var wrote atomic.Int64
+	srv := endlessGateway(t, http.StatusOK, &wrote)
+	c := New(srv.URL, "", "m", 30*time.Second, WithMaxResponseBytes(8192))
+
+	start := time.Now()
+	_, err := c.Complete(context.Background(), "s", "u")
+
+	if err == nil || !strings.Contains(err.Error(), "exceeds") {
+		t.Fatalf("want a response-too-large error, got %v", err)
+	}
+	if time.Since(start) > 5*time.Second {
+		t.Fatalf("the client kept reading: took %s", time.Since(start))
+	}
+	// Hanging up lets the server's writes fail; allow what the transport had already buffered.
+	time.Sleep(100 * time.Millisecond)
+	if got := wrote.Load(); got > 8192+1<<20 {
+		t.Fatalf("server managed to send %d bytes after a cap of 8192", got)
+	}
+}
+
+func TestClient_AFailingBodyIsReadOnlyUpToTheErrorBound(t *testing.T) {
+	var wrote atomic.Int64
+	srv := endlessGateway(t, http.StatusBadGateway, &wrote)
+	c := New(srv.URL, "", "m", 30*time.Second)
+
+	start := time.Now()
+	_, err := c.Complete(context.Background(), "s", "u")
+
+	he, ok := err.(*HTTPError)
+	if !ok {
+		t.Fatalf("want *HTTPError, got %T: %v", err, err)
+	}
+	if he.Code != http.StatusBadGateway {
+		t.Fatalf("code %d", he.Code)
+	}
+	if len(he.Body) > maxErrorBody+1 {
+		t.Fatalf("kept %d bytes of an error body, bound is %d", len(he.Body), maxErrorBody)
+	}
+	if time.Since(start) > 5*time.Second {
+		t.Fatalf("the client kept reading an endless error body: took %s", time.Since(start))
+	}
+}

internal/adapters/llm/client.go

@@ -37,6 +37,21 @@ type Client struct {
 	temp       float64
 	httpClient *http.Client
 	usageHook  func(model string, prompt, completion int)
+	maxBody    int64
+}
+
+// defaultMaxResponseBytes bounds a success body. A completion is a few KB of text; this
+// leaves room for a long one and none for a gateway that streams forever (#53).
+const defaultMaxResponseBytes = 8 << 20
+
+// WithMaxResponseBytes overrides the bound on a success response body. A non-positive n is
+// ignored (keeps the default).
+func WithMaxResponseBytes(n int64) Option {
+	return func(c *Client) {
+		if n > 0 {
+			c.maxBody = n
+		}
+	}
 }
 
 // Option configures a Client at construction. Variadic so the existing 4-arg
@@ -101,6 +116,7 @@ func New(baseURL, apiKey, model string, timeout time.Duration, opts ...Option) *
 		maxTokens:  defaultMaxTokens,
 		temp:       defaultTemperature,
 		httpClient: &http.Client{Timeout: timeout},
+		maxBody:    defaultMaxResponseBytes,
 	}
 	for _, opt := range opts {
 		opt(c)
@@ -184,12 +200,17 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err
 	}
 	defer resp.Body.Close() //nolint:errcheck
 
-	out, err := io.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		// Only the bounded head is ever used (HTTPError.Error truncates to it), so read no more.
+		head, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBody+1))
+		return "", &HTTPError{Code: resp.StatusCode, Body: string(head)}
+	}
+	out, err := io.ReadAll(io.LimitReader(resp.Body, c.maxBody+1))
 	if err != nil {
 		return "", fmt.Errorf("read response: %w", err)
 	}
-	if resp.StatusCode != http.StatusOK {
+	if int64(len(out)) > c.maxBody {
-		return "", &HTTPError{Code: resp.StatusCode, Body: string(out)}
+		return "", fmt.Errorf("LLM response exceeds %d bytes", c.maxBody)
 	}
 
 	var cr chatResponse

internal/adapters/store/uploaded_transcripts_progress_test.go

@@ -68,13 +68,13 @@ func TestUploadProgress_AHeartbeatMovesAliveAtAndNotProgressAt(t *testing.T) {
 	ctx := context.Background()
 	at := time.Date(2026, 10, 2, 8, 30, 0, 0, time.UTC)
 	require.NoError(t, s.SaveProgress(ctx, userA, id, []domain.ProgressStep{{Key: "audio", Label: "a", State: domain.StepRunning}}, at))
-	before := time.Now().Add(-time.Second)
+	beatAt := at.Add(90 * time.Second) // deliberately not the database's NOW()
 
-	require.NoError(t, s.Heartbeat(ctx, userA, id))
+	require.NoError(t, s.Heartbeat(ctx, userA, id, beatAt))
 
 	got, err := s.GetUpload(ctx, userA, id)
 	require.NoError(t, err)
-	assert.True(t, got.AliveAt.After(before), "the heartbeat must stamp the row with now")
+	assert.True(t, got.AliveAt.Equal(beatAt), "the heartbeat is stamped with the application's clock, the same one progress_at uses, not the database's NOW() (#51)")
 	assert.True(t, got.ProgressAt.Equal(at), "alive is not progress: a heartbeat must not move progress_at")
 }
 
@@ -99,7 +99,7 @@ func TestUploadProgress_ByAnotherUserChangesNothing(t *testing.T) {
 	ctx := context.Background()
 
 	errSave := s.SaveProgress(ctx, userB, id, []domain.ProgressStep{{Key: "x", Label: "X", State: domain.StepDone}}, time.Now())
-	errBeat := s.Heartbeat(ctx, userB, id)
+	errBeat := s.Heartbeat(ctx, userB, id, time.Now())
 
 	assert.ErrorIs(t, errSave, store.ErrUploadNotFound)
 	assert.ErrorIs(t, errBeat, store.ErrUploadNotFound)

internal/progress/tracker.go

@@ -42,6 +42,7 @@ type Sink interface {
 
 const (
 	defaultMinSave     = 2 * time.Second
+	defaultSaveTimeout = 5 * time.Second
 	logAdvanceEach     = time.Minute
 )
 
@@ -55,6 +56,11 @@ func WithClock(now func() time.Time) Option { return func(t *Tracker) { t.now =
 // Done, Fail and Finish always save.
 func WithMinSave(d time.Duration) Option { return func(t *Tracker) { t.minSave = d } }
 
+// WithSaveTimeout bounds one save or heartbeat write. Progress is cosmetic, so a database
+// that stops answering costs a dropped write, never a stalled job: saves run under the
+// tracker's lock, which Begin and Advance (called by the engines) need (#51).
+func WithSaveTimeout(d time.Duration) Option { return func(t *Tracker) { t.saveTimeout = d } }
+
 // Tracker implements Reporter for one job.
 type Tracker struct {
 	ctx     context.Context
@@ -63,6 +69,8 @@ type Tracker struct {
 	now     func() time.Time
 	minSave time.Duration
 
+	saveTimeout time.Duration
+
 	mu              sync.Mutex
 	steps           []domain.ProgressStep
 	lastAdvanceSave time.Time
@@ -72,7 +80,7 @@ type Tracker struct {
 // NewTracker plans a job's phases. ctx is used for the saves, so it should outlive the
 // job's own timeout: a job that timed out still gets its failure recorded.
 func NewTracker(ctx context.Context, sink Sink, log *slog.Logger, plan []Planned, opts ...Option) *Tracker {
-	t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave}
+	t := &Tracker{ctx: ctx, sink: sink, log: log, now: time.Now, minSave: defaultMinSave, saveTimeout: defaultSaveTimeout}
 	for _, o := range opts {
 		o(t)
 	}
@@ -212,7 +220,10 @@ func (t *Tracker) Heartbeat(ctx context.Context, beatEvery, stillEvery time.Dura
 		case <-ctx.Done():
 			return
 		case <-beat.C:
-			if err := t.sink.Beat(ctx); err != nil && ctx.Err() == nil {
+			beatCtx, cancel := context.WithTimeout(ctx, t.saveTimeout)
+			err := t.sink.Beat(beatCtx)
+			cancel()
+			if err != nil && ctx.Err() == nil {
 				t.log.Warn("progress: heartbeat not saved", "err", err)
 			}
 		case <-still.C:
@@ -254,7 +265,9 @@ func (t *Tracker) find(key string) *domain.ProgressStep {
 // dropped: progress is cosmetic and must never fail the job it describes.
 func (t *Tracker) save() {
 	snapshot := append([]domain.ProgressStep(nil), t.steps...)
-	if err := t.sink.SaveProgress(t.ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil {
+	ctx, cancel := context.WithTimeout(t.ctx, t.saveTimeout)
+	defer cancel()
+	if err := t.sink.SaveProgress(ctx, snapshot, t.now()); err != nil && t.ctx.Err() == nil {
 		t.log.Warn("progress: not saved", "err", err)
 	}
 }

internal/progress/tracker_stall_test.go

@@ -0,0 +1,98 @@
+package progress_test
+
+// Observes what a database that stops answering does to the job it reports on (#51):
+// progress is cosmetic, so a save that never returns must not stall the engine calls that
+// report through the tracker, nor the heartbeat that says the process is alive.
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"git.d-ma.be/mathias/tapir/internal/domain"
+	"git.d-ma.be/mathias/tapir/internal/progress"
+)
+
+// stallSink never answers until its context ends, like a database connection that hung.
+type stallSink struct{ beats atomic.Int32 }
+
+func (s *stallSink) SaveProgress(ctx context.Context, _ []domain.ProgressStep, _ time.Time) error {
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func (s *stallSink) Beat(ctx context.Context, _ time.Time) error {
+	s.beats.Add(1)
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+type lockedBuf struct {
+	mu sync.Mutex
+	b  bytes.Buffer
+}
+
+func (l *lockedBuf) Write(p []byte) (int, error) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.b.Write(p)
+}
+
+func (l *lockedBuf) String() string {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	return l.b.String()
+}
+
+func stallTracker(sink progress.Sink, buf *lockedBuf) *progress.Tracker {
+	log := slog.New(slog.NewTextHandler(buf, nil))
+	return progress.NewTracker(context.Background(), sink, log,
+		[]progress.Planned{{Key: "a", Label: "A"}}, progress.WithSaveTimeout(50*time.Millisecond))
+}
+
+func TestTracker_AHungSaveDoesNotBlockTheReporterPastItsTimeout(t *testing.T) {
+	buf := &lockedBuf{}
+	tr := stallTracker(&stallSink{}, buf)
+
+	done := make(chan struct{})
+	go func() { tr.Begin("a", 0); tr.Advance("a", 1); tr.Done("a"); close(done) }()
+
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("the reporter was blocked by a save that never returned")
+	}
+	require.Contains(t, buf.String(), "progress: not saved", "a dropped save is logged")
+}
+
+func TestTracker_AHungBeatDoesNotStopTheHeartbeat(t *testing.T) {
+	sink := &stallSink{}
+	tr := stallTracker(sink, &lockedBuf{})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
+
+	require.Eventually(t, func() bool { return sink.beats.Load() >= 3 }, 3*time.Second, 5*time.Millisecond,
+		"each beat is abandoned at its timeout and the next one is tried")
+}
+
+func TestTracker_TheStillRunningLineIsNotHeldUpBehindAHungSave(t *testing.T) {
+	buf := &lockedBuf{}
+	tr := stallTracker(&stallSink{}, buf)
+	go tr.Begin("a", 0) // holds the tracker's lock inside a save that hangs until its timeout
+	time.Sleep(10 * time.Millisecond)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go tr.Heartbeat(ctx, time.Hour, 5*time.Millisecond)
+
+	require.Eventually(t, func() bool { return strings.Contains(buf.String(), "still running") }, 3*time.Second, 5*time.Millisecond)
+}

internal/progress/tracker_test.go

@@ -21,6 +21,7 @@ type fakeSink struct {
 	saves  [][]domain.ProgressStep
 	at     []time.Time
 	beats  int
+	beatAt []time.Time
 	err    error
 }
 
@@ -32,10 +33,11 @@ func (f *fakeSink) SaveProgress(_ context.Context, steps []domain.ProgressStep,
 	return f.err
 }
 
-func (f *fakeSink) Beat(context.Context) error {
+func (f *fakeSink) Beat(_ context.Context, at time.Time) error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	f.beats++
+	f.beatAt = append(f.beatAt, at)
 	return f.err
 }
 
@@ -297,7 +299,7 @@ func (r *retainingSink) SaveProgress(_ context.Context, steps []domain.ProgressS
 	r.got = append(r.got, steps)
 	return nil
 }
-func (r *retainingSink) Beat(context.Context) error { return nil }
+func (r *retainingSink) Beat(context.Context, time.Time) error { return nil }
 
 func TestTracker_ASnapshotHandedToTheSinkIsNotChangedByLaterTransitions(t *testing.T) {
 	sink := &retainingSink{}
@@ -400,3 +402,20 @@ func TestTracker_AnnotateAttachesANoteToAStepAndSavesItAtOnce(t *testing.T) {
 	require.Equal(t, "20 min of the 79 the file claimed", sink.last(t)[0].Note)
 	require.NotPanics(t, func() { tr.Annotate("no-such-step", "x") })
 }
+
+func TestTracker_BeatIsStampedWithTheTrackersClock(t *testing.T) {
+	// alive_at used the database's NOW() while the page compared it with the application
+	// clock; a skewed database clock showed a healthy job as lost (#51).
+	tr, sink, clk, _ := newTracker(t)
+	clk.tick(42 * time.Second)
+	want := clk.now()
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go tr.Heartbeat(ctx, 5*time.Millisecond, time.Hour)
+
+	require.Eventually(t, func() bool { return sink.beatCount() >= 1 }, 2*time.Second, 5*time.Millisecond)
+	sink.mu.Lock()
+	defer sink.mu.Unlock()
+	require.True(t, sink.beatAt[0].Equal(want), "beat stamped %v, the tracker's clock says %v", sink.beatAt[0], want)
+}

internal/uploads/worker_test.go

@@ -70,7 +70,7 @@ func (s *wStore) SaveProgress(_ context.Context, _, _ string, steps []domain.Pro
 	return nil
 }
 
-func (s *wStore) Heartbeat(context.Context, string, string) error {
+func (s *wStore) Heartbeat(context.Context, string, string, time.Time) error {
 	time.Sleep(s.beatDelay)
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/web/uploads.go

@@ -18,6 +18,7 @@ import (
 
 	"git.d-ma.be/mathias/tapir/internal/adapters/chat"
 	"git.d-ma.be/mathias/tapir/internal/adapters/store"
+	"git.d-ma.be/mathias/tapir/internal/config"
 	"git.d-ma.be/mathias/tapir/internal/domain"
 	"git.d-ma.be/mathias/tapir/internal/errkind"
 	"git.d-ma.be/mathias/tapir/internal/uploads"
@@ -243,6 +244,17 @@ func titleFromFilename(name string) string {
 	return t
 }
 
+// cloudProvider names the cloud service that produced a transcript, or false when it was
+// served locally (or by an unknown endpoint). "Local" is config.IsLocalModel: the same
+// definition the start-up validation uses, not a second string match here (#52).
+func cloudProvider(endpoint string) (string, bool) {
+	if endpoint == "" || config.IsLocalModel(endpoint) {
+		return "", false
+	}
+	provider, _, _ := strings.Cut(endpoint, "/")
+	return provider, true
+}
+
 // hasNote reports whether any phase carries a note. A note is the only warning for a
 // recording that is shorter than its file claimed, so a record that has one is shown open.
 func hasNote(steps []domain.ProgressStep) bool {

internal/web/uploads.templ

@@ -66,6 +66,12 @@ templ uploadStatus(u domain.UploadedTranscript, chatEnabled bool, now time.Time)
 					Transcribed { uploads.FormatDuration(time.Duration(u.AudioSeconds * float64(time.Second))) }
 					of audio by <span class="chip">{ u.Endpoint }</span>.
 				</p>
+				if provider, cloud := cloudProvider(u.Endpoint); cloud {
+					<p class="cloud-note">
+						This recording's audio left the homelab: it was transcribed by a cloud service ({ provider })
+						because the local engine could not.
+					</p>
+				}
 				@reconciliationNote(u.Reconciliation)
 				if len(u.Progress) > 0 {
 					<details class="what-was-done" if hasNote(u.Progress) { open }>

internal/web/uploads_cloud_note_test.go

@@ -0,0 +1,55 @@
+package web_test
+
+// A transcript produced by the cloud fallback means this recording's audio left the
+// homelab, which the reader is told in plain words (#52). "Local" is config.IsLocalModel:
+// one definition, not a second string match in the view.
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func setEndpoint(t *testing.T, id, endpoint string) {
+	t.Helper()
+	_, err := rawPool(t).Exec(context.Background(), `UPDATE uploaded_transcripts SET endpoint = $2 WHERE id = $1`, id, endpoint)
+	require.NoError(t, err)
+}
+
+func TestUploadStatus_ACloudServedTranscriptSaysTheAudioLeftTheHomelab(t *testing.T) {
+	app, _ := newUploadApp(t, audioOK, 200<<20)
+	id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
+	setEndpoint(t, id, "berget/kb-whisper")
+
+	b := body(t, do(t, app, get("/uploads/"+id+"/status")))
+
+	assert.Contains(t, b, "left the homelab")
+	assert.Contains(t, b, "cloud service (berget)")
+}
+
+func TestUploadStatus_ALocallyServedTranscriptCarriesNoCloudNote(t *testing.T) {
+	for _, endpoint := range []string{"iguana/kb-whisper", "koala/whisper", ""} {
+		t.Run(endpoint, func(t *testing.T) {
+			app, _ := newUploadApp(t, audioOK, 200<<20)
+			id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
+			setEndpoint(t, id, endpoint)
+
+			b := body(t, do(t, app, get("/uploads/"+id+"/status")))
+
+			assert.NotContains(t, b, "left the homelab")
+		})
+	}
+}
+
+func TestUploadStatus_AnEndpointThatOnlyLooksLocalIsStillNamedAsCloud(t *testing.T) {
+	// The prefix alone must not decide it: IsLocalModel refuses path tricks.
+	app, _ := newUploadApp(t, audioOK, 200<<20)
+	id := seedUpload(t, userID, "m", "done", "Synthetic transcript text.")
+	setEndpoint(t, id, "iguana/../berget/kb-whisper")
+
+	b := body(t, do(t, app, get("/uploads/"+id+"/status")))
+
+	assert.Contains(t, b, "left the homelab")
+}

test/acceptance/scenario_coverage_test.go

@@ -122,6 +122,8 @@ var scenarioCoverage = map[string]string{
 	"I can ask about my recording and get a grounded answer":                    "TestUploadChatPost_SendsTheOwnersTranscriptWithTheUploadBudget",
 	"A long recording is not cut to its first third":                            "TestAnswerMaxCharsOverridesTheServiceCap",
 
+	"A transcript made by the cloud fallback says the audio left the homelab": "TestUploadStatus_ACloudServedTranscriptSaysTheAudioLeftTheHomelab",
+
 	// audio_upload.feature, ADR-034
 	"Two engines transcribe the audio and their disagreements are reconciled": "TestEnsemble_BothEnginesRunInOrderAndTheirDisagreementIsReconciled",
 	"Trivial disagreements never reach the model":                             "TestReconcile_TrivialDisputesAreResolvedWithoutAskingTheModel",