mathias/tapir
1
0
Fork 0
Code Issues 25 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(runner): bound auto-summarize to a recency window

Browse Source 
In automatic mode the scheduler now only summarizes videos published within
TAPIR_AUTO_SUMMARIZE_WINDOW (default ~7d). Older videos are still discovered
and listed — they keep the manual "Summarize" affordance — but are not
auto-processed, so a large back-catalogue (the maintainer's ~256-deep queue)
stops self-inflicting 429s against the per-IP caption gate each cycle (UX
review B1, recency design).

- runner.WithAutoWindow + Stats.SkippedTooOld; tooOld() treats a zero window
  as disabled and an undated video as never-aged-out (processed, not stranded).
- An explicit manual request bypasses the bound even in auto mode (requested
  videos are loaded in auto mode when a window is active).
- Wired through cmdRun, the scheduler's per-user runner, sumStats, and pass
  logging. config: TAPIR_AUTO_SUMMARIZE_WINDOW (default 168h), .env.example.
- Account copy (A7) updated to match: "Automatic summarizes new videos from
  about the last week; older videos stay browsable — summarize on demand."

The rate gate is untouched; the manual path still serialises through it. This
bounds auto LOAD, it does not fetch harder.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
mathias
2026-06-08 13:48:04 +02:00
co-authored by Claude Opus 4.8
parent 4a0a56e152
commit 2384c47b81
9 changed files with 199 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/runner/runner.go
+41 -9
View File
@@ -85,13 +85,14 @@ type Processor interface {
// Runner walks a user's subscriptions, persists each candidate video, skips the
// ones already summarized (durably), and processes the rest through the engine.
type Runner struct {
src ports.VideoSource
store VideoStore
engine Processor
userID string
log *slog.Logger
backoff time.Duration // rate-limit retry window; 0 = always retry
now func() time.Time // injectable clock (tests); defaults to time.Now
src ports.VideoSource
store VideoStore
engine Processor
userID string
log *slog.Logger
backoff time.Duration // rate-limit retry window; 0 = always retry
autoWindow time.Duration // recency bound for auto-summarize; 0 = no bound
now func() time.Time // injectable clock (tests); defaults to time.Now
}
// Option configures a Runner at construction. Variadic so existing call sites
@@ -106,6 +107,13 @@ func WithBackoff(d time.Duration) Option { return func(r *Runner) { r.backoff =
// fixed time; production leaves the time.Now default.
func WithClock(now func() time.Time) Option { return func(r *Runner) { r.now = now } }
// WithAutoWindow bounds auto-summarization to videos published within d of now.
// In auto mode a video older than d is discovered and listed but not summarized
// automatically — it waits for an explicit manual request — so a large
// back-catalogue does not self-inflict 429s. An explicitly requested video
// bypasses the bound. 0 (the default) disables it (summarize every unseen video).
func WithAutoWindow(d time.Duration) Option { return func(r *Runner) { r.autoWindow = d } }
// New builds a Runner. A nil logger falls back to slog.Default.
func New(src ports.VideoSource, store VideoStore, engine Processor, userID string, log *slog.Logger, opts ...Option) *Runner {
if log == nil {
@@ -128,11 +136,23 @@ type Stats struct {
SkippedSeen int
SkippedNoText int
SkippedManual int // discovered but not queued, in manual mode
SkippedTooOld int // auto mode: published outside the recency window (not requested)
SkippedRateLimited int // 429'd previously and still inside the backoff window
Errors int
ChannelUnavailable int // channels that returned HTTP 404 (deleted/private)
}
// tooOld reports whether a video published at publishedAt falls outside the
// auto-summarize recency window. A zero window disables the bound, and a zero
// publishedAt (undated video) is never aged out — it cannot be dated, so it is
// processed rather than silently stranded.
func (r *Runner) tooOld(publishedAt time.Time) bool {
if r.autoWindow <= 0 || publishedAt.IsZero() {
return false
}
return r.now().Sub(publishedAt) > r.autoWindow
}
// RunOnce performs a single pass over the user's subscriptions in three phases:
//
// 1. Discovery: walk all channels, persist each candidate video (UpsertVideo),
@@ -172,8 +192,10 @@ func (r *Runner) RunOnce(ctx context.Context) (Stats, error) {
if err != nil {
return stats, fmt.Errorf("runner: load summarize mode: %w", err)
}
// requested is needed in manual mode (the queue) and in auto mode when a
// recency window is active (an explicit request bypasses the bound).
var requested map[string]bool
if !auto {
if !auto || r.autoWindow > 0 {
requested, err = r.store.RequestedVideoIDs(ctx, r.userID)
if err != nil {
return stats, fmt.Errorf("runner: load requested videos: %w", err)
@@ -242,6 +264,15 @@ func (r *Runner) RunOnce(ctx context.Context) (Stats, error) {
continue
}
// Recency bound (auto mode): summarize only recent videos automatically;
// older ones are discovered + listed (UpsertVideo above) but wait for an
// explicit manual request, so a large back-catalogue does not self-inflict
// 429s against the caption rate gate. A requested video bypasses the bound.
if auto && !requested[id] && r.tooOld(v.PublishedAt) {
stats.SkippedTooOld++
continue
}
// Still inside the rate-limit backoff window: skip without fetching.
if at, ok := rateLimited[id]; ok && r.now().Sub(at) < r.backoff {
stats.SkippedRateLimited++
@@ -321,7 +352,8 @@ func (r *Runner) Loop(ctx context.Context, interval time.Duration) error {
r.log.Info("run pass complete",
"candidates", stats.Candidates, "summarized", stats.Summarized,
"skipped_seen", stats.SkippedSeen, "skipped_no_text", stats.SkippedNoText,
"skipped_manual", stats.SkippedManual, "skipped_rate_limited", stats.SkippedRateLimited,
"skipped_manual", stats.SkippedManual, "skipped_too_old", stats.SkippedTooOld,
"skipped_rate_limited", stats.SkippedRateLimited,
"channel_unavailable", stats.ChannelUnavailable, "errors", stats.Errors)
if err != nil {
r.log.Warn("run pass had errors", "err", err)
internal/runner/runner_test.go
+90
View File
@@ -234,6 +234,96 @@ func TestRunOnce_ManualMode_ProcessesRequested(t *testing.T) {
require.Equal(t, []string{"id-v1"}, st.cleared, "the queue flag is cleared after summarizing")
}
// --- recency window (B1) ---------------------------------------------------
// TestRunOnce_AutoMode_SkipsOldVideos: with a recency window set, auto mode
// summarizes only videos published within the window; older ones are discovered
// (upserted) but not auto-summarized — they wait for a manual request.
func TestRunOnce_AutoMode_SkipsOldVideos(t *testing.T) {
base := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
src := &fakeSource{
subs: []domain.Subscription{sub("chan1", "Channel One")},
videos: map[string][]domain.Video{"chan1": {
vidAt("recent", "Recent", base.Add(-24*time.Hour)), // 1d old → in window
vidAt("old", "Old", base.Add(-30*24*time.Hour)), // 30d old → out of window
}},
}
st := &fakeStore{seen: map[string]bool{}, auto: true}
sink := &recordingSink{}
eng := usecase.NewEngine(src, fakeSummarizer{}, sink)
r := runner.New(src, st, eng, testUser, quietLogger(),
runner.WithAutoWindow(7*24*time.Hour), runner.WithClock(func() time.Time { return base }))
stats, err := r.RunOnce(context.Background())
require.NoError(t, err)
require.Equal(t, 1, stats.Summarized, "only the recent video is auto-summarized")
require.Equal(t, 1, stats.SkippedTooOld, "the old video is skipped by the recency bound")
require.Len(t, sink.delivered, 1)
require.Equal(t, "id-recent", sink.delivered[0].VideoID)
require.Len(t, st.upserted, 2, "both videos are still discovered and listed")
}
// TestRunOnce_AutoMode_OldVideoRequestedBypassesWindow: an explicit manual
// request (summarize_requested) overrides the recency bound even in auto mode.
func TestRunOnce_AutoMode_OldVideoRequestedBypassesWindow(t *testing.T) {
base := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
src := &fakeSource{
subs: []domain.Subscription{sub("chan1", "Channel One")},
videos: map[string][]domain.Video{"chan1": {vidAt("old", "Old", base.Add(-30*24*time.Hour))}},
}
st := &fakeStore{seen: map[string]bool{}, auto: true, requested: map[string]bool{"id-old": true}}
sink := &recordingSink{}
eng := usecase.NewEngine(src, fakeSummarizer{}, sink)
r := runner.New(src, st, eng, testUser, quietLogger(),
runner.WithAutoWindow(7*24*time.Hour), runner.WithClock(func() time.Time { return base }))
stats, err := r.RunOnce(context.Background())
require.NoError(t, err)
require.Equal(t, 1, stats.Summarized, "a requested old video is summarized despite the window")
require.Equal(t, 0, stats.SkippedTooOld)
require.Len(t, sink.delivered, 1)
}
// TestRunOnce_AutoWindowZero_SummarizesOld: a zero window disables the bound —
// the pre-recency behaviour (summarize every unseen video) is preserved.
func TestRunOnce_AutoWindowZero_SummarizesOld(t *testing.T) {
base := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
src := &fakeSource{
subs: []domain.Subscription{sub("chan1", "Channel One")},
videos: map[string][]domain.Video{"chan1": {vidAt("old", "Old", base.Add(-365*24*time.Hour))}},
}
st := &fakeStore{seen: map[string]bool{}, auto: true}
sink := &recordingSink{}
eng := usecase.NewEngine(src, fakeSummarizer{}, sink)
r := runner.New(src, st, eng, testUser, quietLogger(),
runner.WithClock(func() time.Time { return base })) // no WithAutoWindow → 0
stats, err := r.RunOnce(context.Background())
require.NoError(t, err)
require.Equal(t, 1, stats.Summarized, "window disabled → old video summarized")
require.Equal(t, 0, stats.SkippedTooOld)
}
// TestRunOnce_AutoMode_UndatedVideoSummarized: a video with no published_at
// cannot be aged out — it is processed, not silently stranded.
func TestRunOnce_AutoMode_UndatedVideoSummarized(t *testing.T) {
base := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
src := &fakeSource{
subs: []domain.Subscription{sub("chan1", "Channel One")},
videos: map[string][]domain.Video{"chan1": {vid("undated", "Undated")}}, // zero PublishedAt
}
st := &fakeStore{seen: map[string]bool{}, auto: true}
sink := &recordingSink{}
eng := usecase.NewEngine(src, fakeSummarizer{}, sink)
r := runner.New(src, st, eng, testUser, quietLogger(),
runner.WithAutoWindow(7*24*time.Hour), runner.WithClock(func() time.Time { return base }))
stats, err := r.RunOnce(context.Background())
require.NoError(t, err)
require.Equal(t, 1, stats.Summarized, "an undated video is processed, not aged out")
require.Equal(t, 0, stats.SkippedTooOld)
}
// noFetchSource fails the test if a transcript fetch happens — used to prove the
// runner skips a rate-limited video before touching the caption endpoint.
type noFetchSource struct{ *fakeSource }