Compare commits

...
5 Commits
Author SHA1 Message Date
mathiasandClaude Opus 4.8 1aa8a97f95 fix(scheduler): rotate over connected users only for true fair share
CI / Lint / Test / Vet (push) Successful in 12s
CI / Build & Import (push) Successful in 10s
The lead-user rotation rotated the full ListAllUsers set, so a connectionless
orphan identity ate a rotation slot — collapsing onto the next real user and
skewing the lead share (two real users got 2/3 vs 1/3 instead of 50/50). Filter
to connected users BEFORE rotating so the rotation is over exactly the users that
consume the caption budget. A dead identity can no longer skew fairness.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-10 18:28:54 +02:00
mathiasandClaude Opus 4.8 cc69a912f4 fix(scheduler): rotate lead user each pass so caption budget is shared
CI / Lint / Test / Vet (push) Successful in 12s
CI / Build & Import (push) Successful in 10s
Caption fetches share one per-egress-IP rate budget; whoever runs first each pass
spends the pre-throttle window before YouTube starts 429ing. ListAllUsers order
is unspecified and was stable, so the last-listed user was permanently starved —
a friendly-pilot user got 0 fetches in 12h (all rate_limited) while the
first-listed user got every successful fetch. rotateUsers left-rotates the user
order by pass index so each user leads 1/N passes and the lead slot is shared.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-10 18:07:01 +02:00
mathiasandClaude Opus 4.8 f4a0544903 fix(scheduler): cache transcripts on the scheduled path (ADR-021 regression)
buildUserRunner built the engine without engine.Transcripts = st, so the
scheduler — unlike the web "Summarize now" path — never read or wrote the shared
transcript cache. Every discovery pass re-fetched transcripts it had already
fetched, burning the scarce per-egress-IP caption budget (ADR-014) on redundant
work and starving other users' first-time fetches. The transcripts table was
empty despite summaries existing. Wire the cache on this path too.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-10 18:07:01 +02:00
mathiasandClaude Opus 4.8 e2a52789b9 feat(web): mode-aware backlog banner — stop telling manual users summaries auto-arrive
CI / Lint / Test / Vet (push) Successful in 12s
CI / Build & Import (push) Successful in 10s
The first pilot user sat in Manual mode reading "new summaries land gradually,
check back tomorrow" — copy that only makes sense in Automatic mode. Manual mode
never auto-summarizes, so the banner promised delivery that would never come.

The list page now reads the user's summarize mode and shows mode-correct copy:
- Auto: unchanged "land gradually" backlog note.
- Manual: "new videos appear here but are not summarized automatically — use the
  Summarize button" plus a "Switch to Automatic" link to /account.
The connected-but-empty first-run state is likewise mode-aware.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-10 09:14:52 +02:00
mathias e696b6405b docs(build-state): v0.15.0, summarizer is now a resilient chain (ADR-022)
CI / Lint / Test / Vet (push) Successful in 12s
CI / Build & Import (push) Successful in 10s
2026-06-10 08:59:26 +02:00
8 changed files with 437 additions and 269 deletions
+3 -2
View File
@@ -86,7 +86,7 @@ Skills live in the canonical library `mathias/skills` and are wired into this re
## Current build state (start here for the first task)
The repo is **green and shipping** — last tag `v0.14.0`. `task check` passes (fmt, vet, lint,
The repo is **green and shipping** — last tag `v0.15.0`. `task check` passes (fmt, vet, lint,
`go test -p 1 ./...`). Go is `1.26.1` (see `go.mod`).
- Clean Architecture core is implemented: `internal/domain` (entities), `internal/ports`
@@ -95,7 +95,8 @@ The repo is **green and shipping** — last tag `v0.14.0`. `task check` passes (
green against it.
- Adapters present under `internal/adapters/`: `youtube` (captions-first `VideoSource`,
timedtext/InnerTube acquisition per ADR-010), `summarizer` + `llm` (the copied AI router,
Primary→Fallback per ADR-004), `store` (Postgres, golang-migrate migrations 001015),
now a resilient endpoint chain — local primary → local fallback → external worst-case,
parse-failure-aware, ADR-004 + ADR-022), `store` (Postgres, golang-migrate migrations 001015),
`secrets` (file-backed `SecretStore`). The brain HTTP sink (ADR-005) is the remaining
optional sink.
- Stage 1 is open (ADR-012): multi-user with **DB-enforced** isolation — Postgres RLS `FORCE`d
+54 -8
View File
@@ -36,6 +36,11 @@ func buildUserRunner(cfg config.Config, st *store.Store, secretStore ports.Secre
}, secretStore)
engine := usecase.NewEngine(src, buildSummarizer(cfg), st)
// Share the transcript cache (ADR-021) on the scheduler path too — without
// this every scheduled pass re-fetches transcripts it already had, burning the
// scarce per-IP caption budget (ADR-014) and starving other users. The web
// "Summarize now" path already sets this; the scheduler omitting it was a bug.
engine.Transcripts = st
return runner.New(src, st, engine, userID, log,
runner.WithBackoff(cfg.FetchBackoff),
@@ -57,6 +62,7 @@ type userLister interface {
// isolation). Returns the stats summed across users.
func runDiscoveryPass(
ctx context.Context,
pass int,
lister userLister,
runUser func(context.Context, string) (runner.Stats, error),
log *slog.Logger,
@@ -67,15 +73,17 @@ func runDiscoveryPass(
return runner.Stats{}
}
log.Info("scheduler: starting discovery pass", "users", len(users))
var total runner.Stats
// Keep only users with a video connection. A pass for a connectionless user
// (e.g. a stale Dex-era orphan identity) only tries to resolve a token that
// was never minted, logging a spurious "ref not found" every tick. Filtering
// here — BEFORE rotation — also keeps fairness honest: rotation is over the
// users that actually consume the caption budget, so a dead identity can't eat
// a rotation slot and skew the lead share.
var connected []store.UserIdentity
for _, u := range users {
if ctx.Err() != nil {
break // shutting down: stop enumerating
return runner.Stats{} // shutting down
}
// Skip users with no video connection. A discovery pass for them only
// attempts to resolve a token that was never minted, logging a spurious
// "ref not found" every tick (e.g. stale Dex-era orphan identities).
conns, err := lister.ConnectionsForUser(ctx, u.UserID)
if err != nil {
log.Warn("scheduler: list connections failed", "user", u.UserID, "err", err)
@@ -85,6 +93,22 @@ func runDiscoveryPass(
log.Debug("scheduler: skipping user with no video connections", "user", u.UserID)
continue
}
connected = append(connected, u)
}
// Rotate who goes first each pass. Caption fetches share one per-egress-IP
// rate budget (ADR-014); whoever runs first each pass spends the pre-throttle
// window, so a FIXED order permanently starves whoever is last (a new pilot
// user got 0 fetches for 12h while the first-listed user got all of them).
// Rotation over the connected set gives each real user the lead in turn.
connected = rotateUsers(connected, pass)
log.Info("scheduler: starting discovery pass", "users", len(connected))
var total runner.Stats
for _, u := range connected {
if ctx.Err() != nil {
break // shutting down: stop enumerating
}
stats, err := runUser(ctx, u.UserID)
total = sumStats(total, stats)
if err != nil {
@@ -116,7 +140,8 @@ func runScheduler(
return // disabled
}
runDiscoveryPass(ctx, lister, runUser, log)
pass := 0
runDiscoveryPass(ctx, pass, lister, runUser, log)
ticker := time.NewTicker(interval)
defer ticker.Stop()
@@ -125,11 +150,32 @@ func runScheduler(
case <-ctx.Done():
return
case <-ticker.C:
runDiscoveryPass(ctx, lister, runUser, log)
pass++
runDiscoveryPass(ctx, pass, lister, runUser, log)
}
}
}
// rotateUsers left-rotates users by pass positions so a different user leads each
// pass. With n users, user i leads on every pass where pass ≡ i (mod n). A pass
// offset that is negative or exceeds n is normalised. Order within the rotation
// is otherwise preserved, so the set of users run is unchanged — only who is
// first (and thus wins the scarce caption-fetch budget) rotates.
func rotateUsers(users []store.UserIdentity, pass int) []store.UserIdentity {
n := len(users)
if n <= 1 {
return users
}
off := ((pass % n) + n) % n
if off == 0 {
return users
}
out := make([]store.UserIdentity, 0, n)
out = append(out, users[off:]...)
out = append(out, users[:off]...)
return out
}
// sumStats adds two passes' stats field-wise, so runDiscoveryPass can report a
// per-tick aggregate across all users.
func sumStats(a, b runner.Stats) runner.Stats {
+45 -4
View File
@@ -45,6 +45,7 @@ func (f fakeLister) ConnectionsForUser(_ context.Context, userID string) ([]stor
type countingRunUser struct {
mu sync.Mutex
calls map[string]int
order []string // userIDs in the order they were run, across all passes
failFor map[string]bool
}
@@ -60,12 +61,19 @@ func (c *countingRunUser) run(_ context.Context, userID string) (runner.Stats, e
c.mu.Lock()
defer c.mu.Unlock()
c.calls[userID]++
c.order = append(c.order, userID)
if c.failFor[userID] {
return runner.Stats{Errors: 1}, errors.New("boom")
}
return runner.Stats{Summarized: 1}, nil
}
func (c *countingRunUser) runOrder() []string {
c.mu.Lock()
defer c.mu.Unlock()
return append([]string(nil), c.order...)
}
func (c *countingRunUser) count(userID string) int {
c.mu.Lock()
defer c.mu.Unlock()
@@ -94,7 +102,7 @@ func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) {
lister := fakeLister{users: usersN("a", "b", "c")}
rc := newCountingRunUser()
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
require.Equal(t, 1, rc.count("a"))
require.Equal(t, 1, rc.count("b"))
@@ -102,13 +110,46 @@ func TestDiscoveryPassRunsEveryUserOnce(t *testing.T) {
require.Equal(t, 3, stats.Summarized, "stats are summed across users")
}
// Caption fetches share one per-IP budget; a fixed user order starves whoever is
// last. Each pass must rotate which user leads so the lead slot is shared.
func TestDiscoveryPassRotatesLeadUser(t *testing.T) {
lister := fakeLister{users: usersN("a", "b", "c")}
rc := newCountingRunUser()
runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())
runDiscoveryPass(context.Background(), 2, lister, rc.run, quietLog())
require.Equal(t, []string{"a", "b", "c", "b", "c", "a", "c", "a", "b"}, rc.runOrder(),
"each pass left-rotates the user order so every user leads in turn")
// Fairness: over a full rotation cycle every user ran the same number of times.
require.Equal(t, 3, rc.count("a"))
require.Equal(t, 3, rc.count("b"))
require.Equal(t, 3, rc.count("c"))
}
// A connectionless orphan must not consume a rotation slot: rotation is over the
// connected users only, so two real users alternate the lead 50/50 even with a
// dead identity listed between them.
func TestDiscoveryPassRotationIgnoresConnectionlessUsers(t *testing.T) {
lister := fakeLister{users: usersN("a", "orphan", "c"), noConn: map[string]bool{"orphan": true}}
rc := newCountingRunUser()
runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
runDiscoveryPass(context.Background(), 1, lister, rc.run, quietLog())
require.Equal(t, []string{"a", "c", "c", "a"}, rc.runOrder(),
"only connected users rotate; the orphan never runs and never holds a slot")
require.Equal(t, 0, rc.count("orphan"))
}
func TestDiscoveryPassSkipsUsersWithoutConnections(t *testing.T) {
// b never connected a video source (e.g. a stale Dex-era orphan identity).
// It must be skipped silently — not run and logged as a token error every pass.
lister := fakeLister{users: usersN("a", "b", "c"), noConn: map[string]bool{"b": true}}
rc := newCountingRunUser()
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
require.Equal(t, 1, rc.count("a"))
require.Equal(t, 0, rc.count("b"), "a user with no connection must be skipped, not run")
@@ -121,7 +162,7 @@ func TestDiscoveryPassOneUserFailureDoesNotStopOthers(t *testing.T) {
lister := fakeLister{users: usersN("a", "b", "c")}
rc := newCountingRunUser("b") // user b's pass errors
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
require.Equal(t, 1, rc.count("a"))
require.Equal(t, 1, rc.count("b"))
@@ -134,7 +175,7 @@ func TestDiscoveryPassListerErrorIsContained(t *testing.T) {
lister := fakeLister{err: errors.New("db down")}
rc := newCountingRunUser()
stats := runDiscoveryPass(context.Background(), lister, rc.run, quietLog())
stats := runDiscoveryPass(context.Background(), 0, lister, rc.run, quietLog())
require.Equal(t, 0, rc.total(), "no users enumerated → no passes")
require.Equal(t, runner.Stats{}, stats)
+12 -3
View File
@@ -233,11 +233,20 @@ func (a *App) handleList(w http.ResponseWriter, r *http.Request) {
}
hasConnected := len(conns) > 0
if isHTMX(r) {
a.render(w, r, summaryList(buckets, hasConnected))
// Summarization mode drives the backlog copy: an auto user is told summaries
// land gradually; a manual user is told to click Summarize (the first pilot
// user sat in manual mode reading "land automatically" and waited forever).
autoSummarize, err := a.Store.GetAutoSummarize(r.Context(), userID)
if err != nil {
a.serverError(w, r, "summarize mode", err)
return
}
a.render(w, r, ListPage(buckets, f, stats, takeFlash(w, r), hasConnected, channels))
if isHTMX(r) {
a.render(w, r, summaryList(buckets, hasConnected, autoSummarize))
return
}
a.render(w, r, ListPage(buckets, f, stats, takeFlash(w, r), hasConnected, channels, autoSummarize))
}
// handleDetail renders one summary in full (highlights, takeaways, action group).
+33
View File
@@ -477,3 +477,36 @@ func postAction(t *testing.T, app *web.App, videoID, action string, htmx bool) *
}
return do(t, app, req)
}
// TestListManualModeBannerCopy: a manual-mode user with un-summarized videos
// sees the manual prompt (click Summarize), NOT the "summaries land
// automatically" copy that misled the first pilot user into waiting forever.
func TestListManualModeBannerCopy(t *testing.T) {
ctx := context.Background()
app := newApp(t)
p := rawPool(t)
resetDB(t, p)
seedVideo(t, p, videoX, "X Title", "https://x", time.Time{}) // pending, un-summarized
require.NoError(t, app.Store.SetAutoSummarize(ctx, userID, false))
html := body(t, do(t, app, httptest.NewRequest(http.MethodGet, "/", nil)))
require.Contains(t, html, "Manual mode")
require.Contains(t, html, "are not summarized automatically")
require.NotContains(t, html, "land gradually",
"manual-mode user must not be told summaries arrive automatically")
}
// TestListAutoModeBannerCopy: an auto-mode user with a backlog sees the
// gradual-delivery copy, not the manual prompt.
func TestListAutoModeBannerCopy(t *testing.T) {
ctx := context.Background()
app := newApp(t)
p := rawPool(t)
resetDB(t, p)
seedVideo(t, p, videoX, "X Title", "https://x", time.Time{})
require.NoError(t, app.Store.SetAutoSummarize(ctx, userID, true))
html := body(t, do(t, app, httptest.NewRequest(http.MethodGet, "/", nil)))
require.Contains(t, html, "land gradually")
require.NotContains(t, html, "are not summarized automatically")
}
+1 -1
View File
@@ -65,7 +65,7 @@ func TestParseYouTubeVideoID(t *testing.T) {
func TestListPageShowsPasteFormOnlyWhenConnected(t *testing.T) {
render := func(connected bool) string {
var buf bytes.Buffer
if err := ListPage(listBuckets{}, Filter{}, PipelineStats{}, "", connected, nil).Render(context.Background(), &buf); err != nil {
if err := ListPage(listBuckets{}, Filter{}, PipelineStats{}, "", connected, nil, true).Render(context.Background(), &buf); err != nil {
t.Fatalf("render: %v", err)
}
return buf.String()
+17 -4
View File
@@ -104,7 +104,7 @@ templ flashBanner(code string) {
// #summary-list region; a non-HTMX request renders the whole page. flash carries
// a one-shot notification (e.g. "connected", "registered") surfaced on arrival
// after a POST→redirect.
templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasConnected bool, channels []string) {
templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasConnected bool, channels []string, autoSummarize bool) {
@Layout("Tapir — Summaries") {
@flashBanner(flash)
if hasConnected {
@@ -116,14 +116,23 @@ templ ListPage(b listBuckets, f Filter, stats PipelineStats, flash string, hasCo
if stats.RateLimited > 0 || stats.Pending > 0 || stats.NoText > 0 {
@pipelineBar(stats)
}
if stats.RateLimited+stats.Pending > 0 {
if (stats.RateLimited+stats.Pending) > 0 && autoSummarize {
<p class="pipeline-note muted">
Tapir fetches captions slowly on purpose, to respect YouTube's limits
new summaries land gradually. Check back tomorrow.
</p>
}
if (stats.RateLimited+stats.Pending) > 0 && !autoSummarize {
<p class="pipeline-note muted">
You are in Manual mode: new videos appear here but are not summarized
automatically. Use the Summarize button on the ones you want.
</p>
<p class="pipeline-note muted">
<a href="/account">Switch to Automatic</a> to have new videos summarized for you.
</p>
}
<div id="summary-list">
@summaryList(b, hasConnected)
@summaryList(b, hasConnected, autoSummarize)
</div>
}
}
@@ -203,12 +212,16 @@ templ filterForm(f Filter, channels []string) {
// and a single disclosure holding the older un-summarized back-catalogue. Cards
// reflow to a single column on mobile; an empty list shows a friendly first-run
// state instead of a blank table.
templ summaryList(b listBuckets, hasConnected bool) {
templ summaryList(b listBuckets, hasConnected bool, autoSummarize bool) {
if b.empty() {
if hasConnected {
<div class="empty empty-connected">
<strong>Your account is connected</strong>
if autoSummarize {
<span>Tapir is finding your subscriptions and fetching captions summaries appear here gradually. Check back later.</span>
} else {
<span>Tapir is finding your subscriptions. You are in Manual mode, so videos appear here with a Summarize button pick the ones you want, or switch to Automatic in your account.</span>
}
</div>
} else {
<div class="empty">
File diff suppressed because it is too large Load Diff