Compare commits

...
14 Commits
Author SHA1 Message Date
mathiasandClaude Opus 4.8 cd461b95f8 feat(observability): instrument AI + HTTP paths, serve /metrics on a side port (ADR-030, #15)
CI / Build & Import (push) Successful in 11s
CI / Lint / Test / Vet (push) Successful in 10s
Wire the metrics package into the live paths and serve it:
- summarizer: per-endpoint latency by model/outcome(success|error|parse_error)/fallback + slog.
- youtube.FetchTranscript: latency by outcome (captions|none|rate_limited) + slog.
- chat: answer latency by model + slog.
- llm usage hook → token counts (prompt|completion) per model, wired in buildSummarizer/buildChat.
- oidc callback: login counter.
- cmdServe: wrap Router in metrics.HTTPMiddleware (request count + latency by bounded
  route pattern) and serve /metrics on TAPIR_METRICS_ADDR (default :9090), a SEPARATE
  port — never on the public app mux.

BDD: observability.feature scenarios un-pended + mapped. TDD: summarizer wiring tested
black-box via the /metrics scrape; metrics-not-on-public-mux asserted.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-12 08:49:51 +02:00
mathiasandClaude Opus 4.8 9c2a04406b feat(llm): usage hook to surface token counts (ADR-030, #15)
WithUsageHook callback fires with model + prompt/completion tokens parsed from the
response usage block. Keeps the copied stdlib-only llm package decoupled from
metrics (ADR-004) — the caller wires it to internal/metrics. TDD covered.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-12 08:36:02 +02:00
mathiasandClaude Opus 4.8 a5a8cf6f6d feat(metrics): Prometheus collectors + typed API + HTTP middleware (ADR-030, #15)
New internal/metrics package: AI metrics (summarize/caption/chat latency, LLM
tokens), session metrics (requests+latency by bounded route pattern, logins), and
a /metrics Handler. Adapters call a typed API; never touch prometheus types.
Dep: github.com/prometheus/client_golang (standard Go client; cluster runs
prometheus-operator). TDD: collectors + middleware route-pattern cardinality
covered.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-12 08:32:41 +02:00
mathias 64d11af9ef docs(bdd): observability.feature scenarios (@pending until TDD, #15) 2026-06-12 08:28:46 +02:00
mathias b590d2708d docs(adr): ADR-030 observability — slog + Prometheus metrics (issue #15) 2026-06-12 08:27:56 +02:00
mathiasandClaude Opus 4.8 7314895ec4 fix(auth): stateless session cookie — stop logging users out on deploy (ADR-029)
CI / Lint / Test / Vet (push) Successful in 10s
CI / Build & Import (push) Successful in 11s
Pilot feedback: lots of re-logging-in on iPhone. Three causes: sessions lived in
an in-memory map (wiped on every pod restart/deploy), a 1h TTL (idle >1h forced
re-login on a check-back-tomorrow reader), and a session cookie with no Max-Age
(dropped on Safari close). Each re-login is the full IdP redirect dance.

Make sessions stateless: identity + absolute expiry live inside the existing
HMAC-signed cookie (no server table), TTL 1h → 30 days sliding, cookie now
persistent (Max-Age). Survives restarts (test: a cookie from one instance is
accepted by a fresh instance with the same secret), browser-close, and idle.
Trade: no server-side revocation — logout clears the cookie client-side; rotating
tapir-session-secret is the global logout lever. Accepted for the Stage-0 reader.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-11 23:07:13 +02:00
mathiasandClaude Opus 4.8 36dd182fb5 feat(report): Stage-0 usage gate counts from a baseline date (default 2026-06-11)
CI / Build & Import (push) Successful in 11s
CI / Lint / Test / Vet (push) Successful in 10s
The return-usage gate (ADR-016) counted distinct active weeks over ALL history,
so pre-launch noise — testing churn and the period the pilot sat blocked on zero
summaries — would inflate the signal. Add a baseline: ActiveWeeks(ctx, since)
filters login_events + summary_actions to seen_at/acted_at >= since. The report
command sets it to TAPIR_USAGE_GATE_START (YYYY-MM-DD, default 2026-06-11 — the
morning the pilot was unblocked) and prints the baseline. The gate now measures
whether users RETURN once it genuinely works.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-11 22:52:15 +02:00
mathias 40808f2d4b docs(onboard): BDD scenarios + architecture for the ADR-028 burst
CI / Lint / Test / Vet (push) Successful in 10s
CI / Build & Import (push) Successful in 10s
- connect_account.feature: refine the burst scenario to "best recent" (likely-good
  selection, skips too-short/too-long) and add a scenario for the stronger model.
- scenario_coverage: remap to TestOnboardBurstVideoIDs + add the model scenario
  (the old NewestUnsummarizedVideoIDs test was removed).
- architecture.md: new "Connect-time onboarding burst" subsection — junk-avoiding
  selection over persisted duration, the burst-only stronger-model chain, and why
  has-captions/cached-first are not selection signals.
2026-06-11 19:54:29 +02:00
mathias 9232f49555 feat(onboard): burst picks likely-good videos, summarizes with stronger model (ADR-028)
Wire the onboarding burst to its quality-aware selection and a stronger model:
- main.go onboard uses OnboardBurstVideoIDs (junk-avoiding) instead of pure
  newest-first, bounded by MinVideoSeconds / OnboardMaxVideoSeconds.
- buildBurstProcessor builds a burst-only summarizer chain led by the onboard
  model (burstChainModels: onboard -> standard ADR-022 chain, deduped, NDA lever
  intact), over the same store/cache/sink. Collapses onto the shared Processor
  when the onboard model is empty/equal-to-primary or config is incomplete.
- summarizerEndpoint + newYouTubeSource extracted so standard and burst wiring
  share one definition.
- Remove now-superseded NewestUnsummarizedVideoIDs: OnboardBurstVideoIDs(.,0,0)
  is identical pure-newest behaviour and its test covers RLS + ordering.
2026-06-11 19:54:29 +02:00
mathias 582c1a2065 feat(store): OnboardBurstVideoIDs — junk-avoiding burst selection (ADR-028)
Newest-first but quality-aware: excludes a video when its duration is KNOWN and
outside [minSeconds, maxSeconds], dropping Shorts and multi-hour livestream VODs
that waste a scarce caption fetch on a poor first impression. NULL/unknown
duration is kept (degrade-open) but ranked after known-good rows. 0/0 bounds
disable the filter (pure newest-first, the reversibility lever). RLS-scoped.
NewestUnsummarizedVideoIDs is left in place for callers that want pure-newest.
2026-06-11 19:54:29 +02:00
mathias 9f0d8cf198 feat(discovery): persist video duration_s instead of discarding it (ADR-028)
ADR-023's filterLowValue already fetches each candidate's duration via the
cheap videos.list quota call to drop Shorts/live, then threw it away — the
videos.duration_s column (migration 001) was never written. Carry it onto the
kept domain.Video and have UpsertVideo persist it, COALESCE-preserving a known
value so an unknown (0) re-upsert never clobbers it (the channel_title backfill
stance, migration 014). This is the enabling change for length-aware burst
selection. No new migration — the column already exists.
2026-06-11 19:54:29 +02:00
mathias d21077303d feat(config): add OnboardSummarizerModel + OnboardMaxVideoSeconds (ADR-028)
Two knobs for the onboarding-burst quality work:
- TAPIR_ONBOARD_SUMMARIZER_MODEL (default iguana/gemma4-26b): the stronger model
  the burst leads its chain with; empty collapses the burst onto the shared
  processor (lookupOr, so explicit-empty disables).
- TAPIR_ONBOARD_MAX_VIDEO_SECONDS (default 14400/4h): upper duration bound for
  burst picks; 0 disables, negative clamps to 0.

Table-driven tests cover defaults, explicit, disable, and invalid input.
2026-06-11 19:54:29 +02:00
mathias 9b2ee2e765 docs: spec onboarding-wow-burst + ADR-028 (better picks, stronger model)
Phase-1 live-DB investigation found the connect-time burst (ADR-018) fires
but delivers a weak first impression: pure newest-first selection picks junk
(a livestream + regional news for pilot user Jonte), and all burst summaries
run on the weak koala/phi4-mini instead of the validated iguana/gemma4-26b.
Cached-first was investigated and rejected — ~3% cross-user overlap, 0
cached-and-unsummarized, newest-20 all uncached (newest-first and cached-first
are structurally incompatible).

ADR-028 records: persist duration_s at discovery (ADR-023 already fetches it,
just stops discarding), junk-avoiding burst selection (drop known too-short/
too-long), and a stronger burst-only summarizer chain. Not a throughput change.
2026-06-11 19:54:29 +02:00
mathias b8fbc5a805 docs: spec first-session wow — verify & improve onboarding summary burst (investigate-first)
CI / Lint / Test / Vet (push) Successful in 11s
CI / Build & Import (push) Successful in 10s
The concern is new-user first-contact: see some GOOD summaries fast or they
don't return. Reframed as curation/latency for ~3 videos, NOT a 429/throughput
problem (3 fetches is nowhere near the wall). Phase 1 (report-and-stop) verifies
whether the existing cap-3 onboarding burst even fires today, what it delivers,
and — critically — how much transcript-cache overlap exists between users (drives
the blend). Phase 2 levers: cached-transcript-first (instant, zero-fetch),
likely-good selection (has-captions/good-length, not just newest), and optionally
the stronger model for the burst's few summaries. Blend deferred to the
maintainer post-Phase-1. Explicitly NOT bulk-fetch, NOT credentials (ADR-010/026
dead end), NOT a client extension.
2026-06-11 15:36:09 +00:00
35 changed files with 1394 additions and 214 deletions
+160
View File
@@ -1060,6 +1060,165 @@ no migration, no stored state to unwind. Spec: `docs/specs/chat-with-transcript.
---
## ADR-028 — Onboarding burst: pick likely-good videos, summarize them with a stronger model
**Status:** Accepted (2026-06-11). **Refines ADR-018** (the connect-time burst) and **ADR-020**
(recency-bounded auto-summarize). **Builds on ADR-022** (the endpoint chain), **ADR-023**
(the discovery-time `videos.list` enrichment), and **ADR-021** (the shared transcript cache).
Triggered by a Phase-1 investigation of the live pilot DB.
**Context.** A new user's first session decides whether they return (the Stage-0 gate, ADR-016).
The connect-time burst (ADR-018: summarize ≤`TAPIR_ONBOARD_SUMMARIZE_COUNT` newest videos so the
feed isn't empty) *fires* in production, but a live-DB investigation of the second pilot user
("Jonte") found it delivers a weak first impression for two reasons, and ruled out a third idea:
1. **Junk picks.** Selection was pure newest-first (`NewestUnsummarizedVideoIDs`,
`ORDER BY published_at DESC`) with **no quality signal**. Jonte's live burst-3 were a
stock-ticker **livestream** + two regional news clips — newest, not best. The cheap signals
that *could* gate this (duration, live status) are fetched by ADR-023's `videos.list`
enrichment at discovery and then **thrown away**: the `videos.duration_s` column (migration
001) was never written.
2. **Weakest model on the first impression.** All burst summaries ran on `koala/phi4-mini` — the
documented weak link (ADR-022 was born from its failures). The stronger, brain-validated
`iguana/gemma4-26b` was never used, even though the burst is only ~3 summaries.
3. **Cached-first instant summaries — REJECTED.** The idea: skip the fetch, summarize
already-cached transcripts (ADR-021) instantly. The pilot numbers kill it — only **11 videos**
overlap between the two users (~3% of each ~350400-video library), **0** cached-and-
unsummarized, and a new user's newest-20 unsummarized are **20/20 NOT cached**. Newest-first
and cached-first are structurally incompatible: fresh uploads are exactly what nobody has
fetched. An empty lever at pilot scale.
**Decision.**
1. **Persist `duration_s` at discovery.** `filterLowValue` (ADR-023) already has each candidate's
duration in hand; carry it onto the kept `domain.Video` and have `UpsertVideo` write it,
COALESCE-preserving a known value (the channel-title backfill stance, migration 014). No new
migration — the column exists. The connect-triggered discovery pass runs *before* the burst,
so a fresh user's candidates are enriched in time.
2. **Junk-avoiding selection.** A new `OnboardBurstVideoIDs(userID, limit, minSeconds, maxSeconds)`
keeps the newest-first order but drops a video when its duration is *known* and outside
`[minSeconds, maxSeconds]``minSeconds` = `TAPIR_MIN_VIDEO_SECONDS` (60, the Shorts floor),
`maxSeconds` = new `TAPIR_ONBOARD_MAX_VIDEO_SECONDS` (default 14400 = 4h, to drop multi-hour
livestream VODs that pass the live filter once ended). A NULL duration is **unknown** — kept
(degrade-open) but ranked after known-good rows. **has-captions stays un-gateable pre-fetch**
(only knowable after a gate fetch or a ~0-probability cache hit); selection only *avoids
known-junk*, it does not *promise* captions.
3. **Stronger model for the burst only.** `TAPIR_ONBOARD_SUMMARIZER_MODEL` (default
`iguana/gemma4-26b`) leads a burst-specific summarizer chain (onboard model first, then the
standard ADR-022 chain as resilience, deduped), wrapped in a burst-specific processor over the
*same* store/cache/sink — a pure wiring choice; the engine and ports are unchanged (ADR-003).
Empty or equal-to-primary collapses the burst back onto the shared processor.
**Not a throughput change.** The caption rate gate (ADR-014) and the foreground priority lane
(ADR-026) are untouched — same pacing, same cap. This changes *which* ≤3 videos the burst spends
its fetches on and *which model* summarizes them, never how fast or how many. The engine's
existing read-stored-first (ADR-021) is unchanged and still yields a free instant summary on the
rare cache hit — we simply do not *select* for cache hits.
**Consequences.** Better odds of a strong first session: the burst avoids the obvious junk and
runs the better model on the one impression that decides return. The selection improvement is
forward-looking — existing rows have NULL `duration_s` until their next discovery pass backfills
it (lazy, like channel_title); a brand-new user benefits immediately because connect-discovery
runs first. `duration_s` becoming live also unblocks future length-aware features (feed sorting,
"long read" badges) for free.
**Reversibility.** Pure config + wiring + one column write + one query, no migration.
`TAPIR_ONBOARD_MAX_VIDEO_SECONDS=0` (and `TAPIR_MIN_VIDEO_SECONDS=0`) restores pure newest-first;
`TAPIR_ONBOARD_SUMMARIZER_MODEL=""` collapses the burst back to the shared processor.
Spec: `docs/specs/onboarding-wow-burst.md`.
---
## ADR-029 — Stateless session cookie (survives restarts, browser-close, idle)
**Status:** Accepted (2026-06-11). Triggered by pilot feedback: "lots of clicking to log in again
on iPhone." Supersedes the in-memory session store in the ADR-011 login.
**Context.** Three compounding causes made users re-login constantly:
1. **In-memory session store** (`sessionStore` map) — wiped on every pod restart, so each deploy
logged everyone out. During the active build period that was ~15 logouts.
2. **1-hour session TTL** — for a "check back tomorrow" reader, idle > 1h forced a re-login on
nearly every visit.
3. **No cookie Max-Age** — a session cookie (deleted on browser/app close); iPhone Safari closing
the tab dropped it.
Each re-login is the full Dex/Authentik redirect dance — many taps on mobile.
**Decision.** Make the session **stateless**: the identity (subject + email) and an absolute
expiry live INSIDE the existing HMAC-signed (HS256) cookie — no server-side table. Plus:
- **30-day sliding TTL** (was 1h), re-signed on each request so an active user never lapses.
- **Persistent cookie** (`Max-Age` set) so it survives browser/app close.
The cookie is HttpOnly + Secure + SameSite=Lax; the HMAC (keyed by the stable ESO
`tapir-session-secret`, which does NOT rotate per deploy) makes it tamper-proof. The payload is
identity, not secrets — the OIDC access/ID tokens are still discarded after callback.
**Consequences.** A deploy/restart no longer logs anyone out (proven by a test: a cookie issued by
one instance is accepted by a fresh instance with the same secret); works across replicas for
free. **Trade:** no server-side revocation — `logout` clears the cookie client-side, but a copied
cookie stays valid until expiry. Accepted for the Stage-0 reader pilot; revisit (server-side
revocation list, or shorter TTL + refresh) if it ever holds sensitive actions. Rotating
`tapir-session-secret` invalidates all sessions — the global logout lever.
**Not addressed here:** the tap-count of the IdP login page itself is Authentik's UX; with
re-login now rare (30-day idle or explicit logout), it matters far less.
---
## ADR-030 — Observability: slog timing + Prometheus metrics (AI-focused)
**Status:** Proposed (2026-06-11). Issue #15. **Draft for review — no code yet.**
**Context / requirements.** Nothing measures the activities that drive Tapir's performance and
UX, and the Stage-0 eval gate (ADR-016) needs a *performance* dimension to sit beside the
return-usage one. We need timing for: caption fetches (the scarce op), summarization (which model
won, how long, fallbacks), Q&A latency, LLM token spend, and basic session/usage (request rate,
latency by route, logins). Requirements:
- R1: structured `slog` timing at each AI call site (human-readable, already the logging stack).
- R2: Prometheus metrics for the same, scrapeable by the cluster's prometheus-operator.
- R3: **AI metrics are the priority** — summarize latency by `model`/`outcome`/`fallback`,
caption-fetch latency by `outcome`, chat latency by `model`, and LLM `tokens` by model+kind.
- R4: HTTP/session metrics via middleware — request count + latency by route, logins.
- R5: bounded label cardinality (no per-user, no raw-path labels).
- R6: `/metrics` must NOT be publicly exposed.
**Decision / architecture.**
1. **New package `internal/metrics`** owns all Prometheus collectors + a typed API
(`ObserveSummarize`, `ObserveCaptionFetch`, `ObserveChat`, `RecordTokens`, `IncLogin`,
`HTTPMiddleware`, `Handler`). Adapters call this API; they never import prometheus types.
2. **New dependency `github.com/prometheus/client_golang`.** Justification: it is *the* standard
Go Prometheus client and the cluster already runs prometheus-operator; hand-rolling exposition
is not worth it. (Needs the dep-justification note in the commit per repo rules.)
3. **The copied `llm` package stays stdlib-only (ADR-004).** It must not import `internal/metrics`.
Token usage is surfaced via an **optional callback** `llm.WithUsageHook(func(model string, prompt, completion int))`
set at wiring time (`buildSummarizer`/`buildChat`) to `metrics.RecordTokens`; `llm.Client` only
gains parsing of the response `usage` block. Our own adapters (`summarizer`, `youtube`, `chat`)
may import `internal/metrics` directly.
4. **HTTP middleware** reads `r.Pattern` AFTER routing (Go 1.22 sets it during ServeMux match), so
the `route` label is the bounded registered pattern (`GET /v/{videoId}`), satisfying R5;
unmatched → `other`.
5. **Dedicated metrics port** (`TAPIR_METRICS_ADDR`, default `:9090`) served by a second
`http.Server` in `cmdServe`; `/metrics` is never on the public app mux (R6). A **PodMonitor**
in `mathias/infra` scrapes it; the deployment exposes the port.
6. **slog** elapsed fields are emitted alongside each metric at the call sites (R1).
**Hook points (where the instrumentation lands).**
- `summarizer.Summarize` — per-endpoint timing + outcome (`success`/`parse_error`/`error`) + fallback flag.
- `youtube.FetchTranscript` — fetch timing + outcome from `domain.Transcript.Source`.
- `chat.Service` answer — timing by model.
- `llm.Client.Complete` — parse `usage`, fire the usage hook.
- `oidc.handleCallback``IncLogin`.
- `cmdServe` — wrap `Router()` in `metrics.HTTPMiddleware`; start the metrics server.
**Out of scope / later.** Persisting per-summary latency into Postgres for `tapir report`
(derive UX latency — publish/discovery → summary — from existing timestamps first; only persist
op-latency if the scrape proves insufficient). SPA view (#16) and visual refresh (#17).
**Reversibility.** Additive: a new package + middleware + a metrics port. Removing the PodMonitor
stops scraping; the app is unaffected. No schema change.
**Next steps (gated):** on approval of this ADR → BDD scenarios (`docs/use-cases/observability.feature`
+ scenario-coverage map) → TDD → implement → SemVer + docs + PodMonitor.
---
## Rejected alternatives
Approaches considered during the 2026-06-02 planning + grill session and **deliberately not
@@ -1083,6 +1242,7 @@ maps to the ADR that settles it.
| Feedback-based Stage 0 gate (friends saying it's useful) | Politeness bias makes asked-for feedback the least reliable signal; return-usage is the real test | ADR-016 |
| Reverse the Dex-write invite flow (Google OIDC only) | Some intended Future-B users won't use Google; OIDC-only leaves them with no onboarding path — invite flow is load-bearing | ADR-017 |
| k8s CronJob for scheduled discovery (vs in-process) | At Future-B scale the in-process scheduler is simpler to deploy; CronJob's failure-isolation benefit was weighed and traded away knowingly (revisit if >1 replica or load grows) | ADR-018 |
| Cached-transcript-first onboarding burst (instant, zero-fetch picks) | Live pilot DB: ~3% cross-user video overlap, 0 cached-and-unsummarized, a new user's newest-20 are 20/20 uncached — newest-first and cached-first are structurally incompatible. Empty lever at pilot scale | ADR-028 |
If a future case genuinely reopens one of these, that's a new ADR superseding the relevant one —
not a silent reversal.
+43 -11
View File
@@ -28,6 +28,7 @@ import (
"gitea.d-ma.be/mathias/tapir/internal/adapters/youtube"
"gitea.d-ma.be/mathias/tapir/internal/auth"
"gitea.d-ma.be/mathias/tapir/internal/config"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
"gitea.d-ma.be/mathias/tapir/internal/runner"
"gitea.d-ma.be/mathias/tapir/internal/web"
"gitea.d-ma.be/mathias/tapir/internal/web/oidc"
@@ -262,23 +263,36 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
// One lock shared by the scheduler and connect-triggered passes (#6) so
// they never fetch concurrently — the single-fetcher invariant (ADR-018).
runUser := serialize(&sync.Mutex{}, rawRunUser)
// Onboarding burst (Feature 1): after the connect-triggered discovery pass,
// summarize up to OnboardSummarizeCount of the user's NEWEST unsummarized
// videos so a fresh account gets real summaries in its first session. Hard
// cap; explicit, so it bypasses the recency window — but every fetch still
// goes through globalFetchGate via the Processor. No-op when disabled
// (count 0) or queue-only (no Processor).
// Onboarding burst (Feature 1, refined by ADR-028): after the connect-triggered
// discovery pass, summarize up to OnboardSummarizeCount of the user's newest
// LIKELY-GOOD unsummarized videos so a fresh account gets a strong first
// session. Selection avoids known-junk (Shorts/over-long/livestream VODs via
// the persisted duration); the burst leads its chain with the stronger onboard
// model. Hard cap; explicit, so it bypasses the recency window — but every
// fetch still goes through globalFetchGate. No-op when disabled (count 0) or
// queue-only (no processor).
//
// burstProcessor leads with the stronger model (ADR-028); it collapses onto the
// shared Processor when the onboard model is empty/equal-to-primary or the
// engine config is incomplete.
burstProcessor := app.Processor
if burstEngine, berr := buildBurstProcessor(cfg, st); berr != nil {
return berr
} else if burstEngine != nil {
burstProcessor = &engineProcessor{engine: burstEngine, store: st}
log.Info("onboarding burst uses a stronger model", "onboard_model", cfg.OnboardSummarizerModel)
}
onboard := func(ctx context.Context, userID string) {
if cfg.OnboardSummarizeCount <= 0 || app.Processor == nil {
if cfg.OnboardSummarizeCount <= 0 || burstProcessor == nil {
return
}
ids, err := st.NewestUnsummarizedVideoIDs(ctx, userID, cfg.OnboardSummarizeCount)
ids, err := st.OnboardBurstVideoIDs(ctx, userID, cfg.OnboardSummarizeCount, cfg.MinVideoSeconds, cfg.OnboardMaxVideoSeconds)
if err != nil {
log.Warn("onboarding: list newest unsummarized", "user", userID, "err", err)
log.Warn("onboarding: list burst candidates", "user", userID, "err", err)
return
}
for _, id := range ids {
if err := app.Processor.ProcessVideo(ctx, userID, id); err != nil {
if err := burstProcessor.ProcessVideo(ctx, userID, id); err != nil {
log.Warn("onboarding: summarize", "user", userID, "video", id, "err", err)
}
}
@@ -297,16 +311,34 @@ func cmdServe(ctx context.Context, log *slog.Logger) error {
srv := &http.Server{
Addr: cfg.HTTPAddr,
Handler: app.Router(),
Handler: metrics.HTTPMiddleware(app.Router()),
ReadHeaderTimeout: 10 * time.Second,
}
// Prometheus /metrics on a SEPARATE port (ADR-030) — never on the public app
// mux, so a scrape is in-cluster only. Empty TAPIR_METRICS_ADDR disables it.
var metricsSrv *http.Server
if cfg.MetricsAddr != "" {
mmux := http.NewServeMux()
mmux.Handle("GET /metrics", metrics.Handler())
metricsSrv = &http.Server{Addr: cfg.MetricsAddr, Handler: mmux, ReadHeaderTimeout: 10 * time.Second}
go func() {
log.Info("serving metrics", "addr", cfg.MetricsAddr)
if err := metricsSrv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
log.Error("metrics server", "err", err)
}
}()
}
// Graceful shutdown on signal: stop accepting, drain in-flight requests.
go func() {
<-ctx.Done()
shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
_ = srv.Shutdown(shutdownCtx)
if metricsSrv != nil {
_ = metricsSrv.Shutdown(shutdownCtx)
}
}()
log.Info("serving web ui", "addr", cfg.HTTPAddr, "user", cfg.UserID)
+85 -17
View File
@@ -13,6 +13,7 @@ import (
"gitea.d-ma.be/mathias/tapir/internal/adapters/youtube"
"gitea.d-ma.be/mathias/tapir/internal/config"
"gitea.d-ma.be/mathias/tapir/internal/domain"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
"gitea.d-ma.be/mathias/tapir/internal/ports"
"gitea.d-ma.be/mathias/tapir/internal/usecase"
"gitea.d-ma.be/mathias/tapir/internal/web"
@@ -52,13 +53,7 @@ func (f videoFetcher) FetchVideo(ctx context.Context, userID, videoID string) (d
// alias, not a second client config. Empty model entries are skipped, so a
// client deployment can set the cloud fallback empty to keep content local.
func buildSummarizer(cfg config.Config) *summarizer.Summarizer {
mk := func(model string) summarizer.Endpoint {
return summarizer.Endpoint{
Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens)),
Provider: providerOf(model),
Model: model,
}
}
mk := summarizerEndpoint(cfg)
eps := []summarizer.Endpoint{mk(cfg.SummarizerModel)}
if cfg.FallbackModel != "" && cfg.FallbackModel != cfg.SummarizerModel {
eps = append(eps, mk(cfg.FallbackModel))
@@ -69,6 +64,55 @@ func buildSummarizer(cfg config.Config) *summarizer.Summarizer {
return summarizer.NewChain(eps, cfg.MaxTranscriptChars)
}
// summarizerEndpoint returns a constructor for a chain endpoint over the one
// LiteLLM gateway, varying only the model alias (the gateway fronts both
// llama-swap and berget). Shared by the standard and burst chains.
func summarizerEndpoint(cfg config.Config) func(model string) summarizer.Endpoint {
return func(model string) summarizer.Endpoint {
return summarizer.Endpoint{
Client: llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens), llm.WithUsageHook(metrics.RecordTokens)),
Provider: providerOf(model),
Model: model,
}
}
}
// burstChainModels is the ordered, deduped model list for the onboarding burst
// (ADR-028): the stronger onboard model leads, then the standard ADR-022 chain
// (primary → local fallback → cloud) follows as resilience. Empty entries are
// dropped and duplicates collapsed, so the NDA lever (empty cloud fallback) keeps
// the burst chain fully local exactly as the standard chain does.
func burstChainModels(cfg config.Config) []string {
var models []string
add := func(m string) {
if m == "" {
return
}
for _, e := range models {
if e == m {
return
}
}
models = append(models, m)
}
add(cfg.OnboardSummarizerModel)
add(cfg.SummarizerModel)
add(cfg.FallbackModel)
add(cfg.CloudFallbackModel)
return models
}
// buildBurstSummarizer builds the onboarding-burst summarizer chain (ADR-028):
// the onboard model first, then the standard chain as fallback, deduped.
func buildBurstSummarizer(cfg config.Config) *summarizer.Summarizer {
mk := summarizerEndpoint(cfg)
var eps []summarizer.Endpoint
for _, m := range burstChainModels(cfg) {
eps = append(eps, mk(m))
}
return summarizer.NewChain(eps, cfg.MaxTranscriptChars)
}
// chatModels is the ordered, local-first set of models offered in the chat
// switcher (ADR-027), reusing the ADR-022 chain: primary → local fallback →
// cloud. Empty entries are dropped and duplicates collapsed, so a client/NDA
@@ -107,7 +151,7 @@ func buildChat(cfg config.Config) *chat.Service {
return nil
}
newClient := func(model string) chat.Completer {
return llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens))
return llm.New(cfg.GatewayURL, cfg.GatewayKey, model, cfg.SummarizerTimeout, llm.WithMaxTokens(cfg.SummaryMaxTokens), llm.WithUsageHook(metrics.RecordTokens))
}
return chat.New(newClient, models, cfg.MaxTranscriptChars)
}
@@ -126,15 +170,7 @@ func buildProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error)
return nil, nil
}
secretStore := secrets.NewFileStore(cfg.SecretsFile)
src := youtube.New(youtube.Config{
ClientID: cfg.YTClientID,
ClientSecret: cfg.YTClientSecret,
TokenSecretRef: cfg.YTTokenRef,
PreferredLanguages: []string{"en"},
MinVideoSeconds: cfg.MinVideoSeconds,
}, secretStore)
src := newYouTubeSource(cfg, secrets.NewFileStore(cfg.SecretsFile))
sum := buildSummarizer(cfg)
// The store is both the summary sink and the shared transcript cache (ADR-021):
@@ -145,6 +181,38 @@ func buildProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error)
return eng, nil
}
// newYouTubeSource builds the captions-first VideoSource shared by the standard
// and burst processors — same per-process YouTube credentials and ADR-023 Shorts
// filter; only the summarizer chain differs between them.
func newYouTubeSource(cfg config.Config, secretStore ports.SecretStore) ports.VideoSource {
return youtube.New(youtube.Config{
ClientID: cfg.YTClientID,
ClientSecret: cfg.YTClientSecret,
TokenSecretRef: cfg.YTTokenRef,
PreferredLanguages: []string{"en"},
MinVideoSeconds: cfg.MinVideoSeconds,
}, secretStore)
}
// buildBurstProcessor wires a processor whose summarizer leads with the stronger
// onboard model (ADR-028), used only by the connect-time burst over the SAME
// store / transcript cache / sink — a wiring choice; the engine and ports are
// unchanged. Returns (nil, nil) — the collapse lever — when the onboard model is
// empty or equal to the primary (the burst then reuses the shared processor), or
// when the engine config is incomplete (queue-only, same as buildProcessor).
func buildBurstProcessor(cfg config.Config, st *store.Store) (*usecase.Engine, error) {
if cfg.OnboardSummarizerModel == "" || cfg.OnboardSummarizerModel == cfg.SummarizerModel {
return nil, nil
}
if cfg.GatewayURL == "" || cfg.YTClientID == "" || cfg.YTClientSecret == "" || cfg.SecretsFile == "" {
return nil, nil
}
src := newYouTubeSource(cfg, secrets.NewFileStore(cfg.SecretsFile))
eng := usecase.NewEngine(src, buildBurstSummarizer(cfg), st)
eng.Transcripts = st
return eng, nil
}
// engineProcessor adapts the engine (which works in terms of a domain.Video) to
// the web.Processor port (which works in terms of a stored video id): it loads the
// video row, runs the engine, and — on a produced summary — clears the manual
+62
View File
@@ -45,3 +45,65 @@ func TestBuildProcessorNilOnIncompleteConfig(t *testing.T) {
})
}
}
// TestBurstChainModelsLeadsWithOnboardModel: the onboarding burst chain (ADR-028)
// leads with the stronger onboard model, then falls back through the standard
// ADR-022 chain (primary -> local fallback -> cloud), deduped.
func TestBurstChainModelsLeadsWithOnboardModel(t *testing.T) {
got := burstChainModels(config.Config{
OnboardSummarizerModel: "iguana/gemma4-26b",
SummarizerModel: "koala/phi4-mini",
FallbackModel: "iguana/gemma4-26b", // also the onboard model -> dedup
CloudFallbackModel: "berget/mistral-small",
})
want := []string{"iguana/gemma4-26b", "koala/phi4-mini", "berget/mistral-small"}
if len(got) != len(want) {
t.Fatalf("burstChainModels = %v, want %v", got, want)
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("burstChainModels = %v, want %v", got, want)
}
}
}
// TestBurstChainModelsCloudAbsentWhenDisabled: the NDA lever holds for the burst
// too — empty cloud fallback keeps the burst chain fully local.
func TestBurstChainModelsCloudAbsentWhenDisabled(t *testing.T) {
got := burstChainModels(config.Config{
OnboardSummarizerModel: "iguana/gemma4-26b",
SummarizerModel: "koala/phi4-mini",
CloudFallbackModel: "",
})
for _, m := range got {
if m == "" || m == "berget/mistral-small" {
t.Fatalf("cloud model leaked into burst chain: %v", got)
}
}
}
// TestBuildBurstProcessorNilWhenCollapsed: an empty or primary-equal onboard model
// collapses the burst onto the shared processor (buildBurstProcessor returns nil).
func TestBuildBurstProcessorNilWhenCollapsed(t *testing.T) {
base := config.Config{
GatewayURL: "http://gw/v1",
YTClientID: "id",
YTClientSecret: "secret",
SecretsFile: "/tmp/secrets.json",
SummarizerModel: "koala/phi4-mini",
}
t.Run("empty onboard model", func(t *testing.T) {
base.OnboardSummarizerModel = ""
eng, err := buildBurstProcessor(base, nil)
if err != nil || eng != nil {
t.Fatalf("buildBurstProcessor = (%v, %v), want (nil, nil)", eng, err)
}
})
t.Run("onboard model equals primary", func(t *testing.T) {
base.OnboardSummarizerModel = "koala/phi4-mini"
eng, err := buildBurstProcessor(base, nil)
if err != nil || eng != nil {
t.Fatalf("buildBurstProcessor = (%v, %v), want (nil, nil)", eng, err)
}
})
}
+33 -3
View File
@@ -6,6 +6,7 @@ import (
"io"
"os"
"text/tabwriter"
"time"
"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
)
@@ -14,6 +15,27 @@ import (
// weeks. The gate passes when any user reaches it.
const gateThreshold = 2
// defaultGateStart is the date Stage-0 return-usage tracking begins: the morning
// the pilot was actually unblocked and summaries started flowing (2026-06-11).
// Activity before this — testing, the period the pilot was stuck on zero — is
// noise and must not count toward the gate. Override with TAPIR_USAGE_GATE_START
// (YYYY-MM-DD). The gate measures whether users RETURN once it genuinely works.
const defaultGateStart = "2026-06-11"
// gateStart resolves the baseline date from TAPIR_USAGE_GATE_START or the default,
// parsed as a UTC calendar day.
func gateStart() (time.Time, error) {
v := os.Getenv("TAPIR_USAGE_GATE_START")
if v == "" {
v = defaultGateStart
}
t, err := time.Parse("2006-01-02", v)
if err != nil {
return time.Time{}, fmt.Errorf("TAPIR_USAGE_GATE_START=%q: want YYYY-MM-DD: %w", v, err)
}
return t, nil
}
// runReport prints the Stage-0 usage gate: per-user distinct active weeks (reads
// UNION acts) and the pass/fail verdict. Read-only, cross-user — needs only
// TAPIR_DB_DSN (not TAPIR_USER_ID; the report enumerates all users itself).
@@ -28,16 +50,24 @@ func runReport(ctx context.Context, _ []string) error {
}
defer s.Close()
rows, err := s.ActiveWeeks(ctx)
since, err := gateStart()
if err != nil {
return err
}
return formatReport(os.Stdout, rows)
rows, err := s.ActiveWeeks(ctx, since)
if err != nil {
return err
}
return formatReport(os.Stdout, rows, since)
}
// formatReport renders the per-user week counts and the gate verdict. Pure: no DB,
// no env — so the layout and verdict logic are unit-testable without Postgres.
func formatReport(w io.Writer, rows []store.UserActiveWeeks) error {
func formatReport(w io.Writer, rows []store.UserActiveWeeks, since time.Time) error {
if _, err := fmt.Fprintf(w, "Counting usage since %s (Stage-0 gate baseline)\n\n", since.Format("2006-01-02")); err != nil {
return err
}
if len(rows) == 0 {
_, err := fmt.Fprintln(w, "no users yet")
return err
+7 -3
View File
@@ -3,12 +3,15 @@ package main
import (
"strings"
"testing"
"time"
"github.com/stretchr/testify/require"
"gitea.d-ma.be/mathias/tapir/internal/adapters/store"
)
var testSince = time.Date(2026, 6, 11, 0, 0, 0, 0, time.UTC)
func TestFormatReportColumnsAndGatePass(t *testing.T) {
rows := []store.UserActiveWeeks{
{UserID: "user-a", DisplayName: "Ada", ActiveWeeks: 3},
@@ -16,9 +19,10 @@ func TestFormatReportColumnsAndGatePass(t *testing.T) {
}
var b strings.Builder
require.NoError(t, formatReport(&b, rows))
require.NoError(t, formatReport(&b, rows, testSince))
out := b.String()
require.Contains(t, out, "since 2026-06-11", "report states the gate baseline date")
require.Contains(t, out, "USER")
require.Contains(t, out, "ACTIVE_WEEKS")
require.Contains(t, out, "Ada")
@@ -34,12 +38,12 @@ func TestFormatReportGateNotMet(t *testing.T) {
rows := []store.UserActiveWeeks{{UserID: "user-a", ActiveWeeks: 1}}
var b strings.Builder
require.NoError(t, formatReport(&b, rows))
require.NoError(t, formatReport(&b, rows, testSince))
require.Contains(t, b.String(), "NOT YET MET", "no user at >= 2 weeks fails the gate")
}
func TestFormatReportEmpty(t *testing.T) {
var b strings.Builder
require.NoError(t, formatReport(&b, nil))
require.NoError(t, formatReport(&b, nil, testSince))
require.Contains(t, b.String(), "no users yet")
}
+26
View File
@@ -250,6 +250,32 @@ After (newest-first): `[chanB-new, chanA-mid, chanA-old, chanB-null]`
The set of *processed* videos now also excludes auto-mode back-catalogue beyond the recency
window (those stay listed, summarised on demand); within the processed set, only order changes.
### Connect-time onboarding burst (ADR-018 → ADR-028)
On a successful YouTube connect, `ConnectHandler` enqueues a connect-triggered discovery pass;
the `discoveryTrigger` runs that pass and then fires the **onboarding burst** — a third entry path
that summarises up to `TAPIR_ONBOARD_SUMMARIZE_COUNT` (default 3, hard-capped) of the new user's
videos so the first session is not empty. The burst still flows through `globalFetchGate` (it is
not a throughput change); ADR-028 sharpened *which* videos and *which model*:
- **Selection** is `OnboardBurstVideoIDs`, not pure newest-first. It keeps newest-first order but
excludes a video whose **known** duration is outside `[TAPIR_MIN_VIDEO_SECONDS,
TAPIR_ONBOARD_MAX_VIDEO_SECONDS]` (drops Shorts and multi-hour livestream VODs). An unknown
(NULL) duration is degrade-open — kept, but ranked after known-good rows. The connect-triggered
discovery pass runs *before* the burst, and ADR-023's `videos.list` enrichment now **persists**
`duration_s` (instead of discarding it after the Shorts filter), so a fresh user's candidates
carry a duration in time for selection.
- **Model**: the burst runs through a dedicated summarizer chain led by
`TAPIR_ONBOARD_SUMMARIZER_MODEL` (default `iguana/gemma4-26b`, the stronger local model), with
the standard ADR-022 chain following as fallback. This is a wiring choice — a second
`engineProcessor` over the same store / transcript cache / sink; the engine and ports are
unchanged. Empty / equal-to-primary collapses it back onto the shared processor.
`has-captions` is deliberately **not** a selection signal — it is only knowable after a gate fetch
(or a ~0-probability cache hit at pilot scale), so the burst can avoid known-junk but cannot
promise captions. Cached-transcript-first selection was investigated and rejected (ADR-028:
~3% cross-user overlap).
---
## Sequence — core use case: new video summarized
+3
View File
@@ -224,6 +224,9 @@ knobs plus one load-bearing deployment constraint:
- `TAPIR_DISCOVERY_INTERVAL` — Go duration, e.g. `2h`. The cadence the serve process runs a
discovery pass for every registered user (run-once-on-startup, then every interval).
**Unset or `0` = disabled** (dev/tests never auto-fetch).
- `TAPIR_USAGE_GATE_START``YYYY-MM-DD`, default **`2026-06-11`** (the morning the pilot was
unblocked and summaries started flowing). `tapir report` counts return-usage (distinct active
weeks, ADR-016) only from this date, so pre-launch testing and the blocked period are excluded.
- `TAPIR_FETCH_RATE` — Go duration, default `2s`. The **process-wide per-egress-IP caption-fetch
rate gate** (ADR-014 item 2). Every caption fetch — scheduler runners *and* the web "Summarize"
click-path — serialises through this one limiter so the pod cannot collectively trip 429s. `0`
+128
View File
@@ -0,0 +1,128 @@
# Spec — Onboarding "wow" burst: better picks, stronger model
**Repo:** tapir · **Size:** medium · **Solo session** (not a swarm).
> **Status: built (v0.25.0, ADR-028).** This supersedes the original investigate-first brief
> (committed as the prior version of this file): Phase 1 was run against the live pilot DB and its
> findings are folded into "Why this exists" below; Phase 2 was built as described here. The one
> brief lever NOT built — the honest "the rest fill in over the coming days" framing copy — is
> listed under *Explicitly NOT in this slice*.
**Why this exists.** A new user's first session decides whether they return (the Stage-0 gate,
VISION.md). On connect, Tapir fires a capped burst (≤`TAPIR_ONBOARD_SUMMARIZE_COUNT`, default 3)
that summarizes the user's newest unsummarized videos so the feed isn't empty (the burst itself
works — wired in `cmd/tapir/discovery.go``cmd/tapir/main.go` `onboard`). A Phase-1
investigation of the live pilot DB found the burst *fires* but delivers a **weak first
impression** for two concrete reasons, and ruled out a third idea:
1. **Picks are junk.** Selection is pure newest-first (`videos.NewestUnsummarizedVideoIDs`,
`ORDER BY published_at DESC`) with **zero quality signal**. Pilot user "Jonte"'s live burst-3
were a stock-ticker **livestream** + two regional news clips — the newest, not the best.
2. **Weakest model on the first impression.** All of Jonte's summaries ran on
`koala/phi4-mini` (the documented weak link — ADR-022 was born from its failures). The
stronger, brain-validated `iguana/gemma4-26b` was never used for the burst.
3. **Cached-first is empty at pilot scale — REJECTED.** The idea (summarize already-cached
transcripts instantly, zero fetch) dies on the numbers: only **11 videos** overlap between the
two pilot users (~3% of each library), **0** cached-and-unsummarized, and a new user's
newest-20 unsummarized are **20/20 NOT cached** — newest-first and cached-first are
structurally incompatible (fresh uploads are exactly what nobody has fetched yet). Not built.
This is a **curation/latency problem for ~3 videos, NOT a throughput/429 problem** — fetching 3
captions is nowhere near the rate limit. Nothing here fetches harder or pressures the rate gate;
it picks the right few videos and runs a better model on them.
Read `CLAUDE.md`, `DECISIONS.md` (esp. ADR-014, ADR-018, ADR-020, ADR-021, ADR-022, ADR-023,
and the new **ADR-028**), and `VISION.md` (the Stage-0 gate) first. TBD — commit directly to
`main`, one logical change per commit, conventional commits, `task check` green before each
commit, `templ generate` if any view changes (none expected).
## Decisions already made (do not reopen)
- **Not a throughput change.** The caption rate gate (ADR-014) is untouched — same pacing, same
priority lane (ADR-026). This slice changes *which* ≤3 videos the burst spends its fetches on
and *which model* summarizes them, never how fast or how many.
- **Cached-first is dropped** (ADR-028, the 3% overlap). The engine's existing read-stored-first
(ADR-021, `resolveTranscript`) stays — it already gives a free instant summary on the rare
cache hit, transparently. We do not *select* for cache hits.
- **has-captions is not a pre-fetch signal.** It is only knowable after a gate fetch (or a cache
hit, ~0 for new videos). Selection can only *avoid known-junk* (Shorts/live/over-long) — it
cannot *guarantee* captions. The spec is honest about this: better odds, not a promise.
- **No credentialed caption fetch** (ADR-010/ADR-026 dead end). **No client extension.**
## 1. Persist `duration_s` at discovery (the enabling change)
The `videos.duration_s` column exists (migration 001) but is **never written** — ADR-023's
`filterLowValue` (`internal/adapters/youtube/youtube.go`) already fetches each candidate's
duration via the cheap quota `videos.list` call, uses it to drop Shorts/live, then **discards
it**. Stop discarding:
- Add `DurationSeconds int` to `domain.Video`.
- In `filterLowValue`, set `DurationSeconds` on each kept video from the `videos.list` `meta`.
- `UpsertVideo` writes `duration_s`, **COALESCE-preserving** a known value (never overwrite a
real duration with 0/unknown), mirroring the `channel_title` backfill stance (migration 014).
- No new migration — the column is already there.
Consequence: a fresh user's connect-triggered discovery pass runs **before** the onboard burst
(`Enqueue`: `run()` then `onboard()`), so duration is populated for the burst's candidates at
connect. Existing rows backfill on their next discovery pass; until then their `duration_s` is
NULL and treated as "unknown" (§2).
## 2. Junk-avoiding burst selection
New store method, RLS-scoped via `withUser`:
```
OnboardBurstVideoIDs(ctx, userID string, limit, minSeconds, maxSeconds int) ([]string, error)
```
- Same base as the old `NewestUnsummarizedVideoIDs`: the user's videos with no summary yet,
`ORDER BY published_at DESC NULLS LAST, seen_at DESC`, `LIMIT limit`.
- **Exclude known-junk**: a row is dropped only when `duration_s IS NOT NULL` **and**
(`duration_s < minSeconds` OR `duration_s > maxSeconds`). A NULL duration is **unknown** — kept
(degrade-open: never starve the burst because metadata is missing), but ordered *after* rows
with a known-good duration so a freshly-enriched good pick wins when both exist.
- `minSeconds` reuses `TAPIR_MIN_VIDEO_SECONDS` (default 60 — the Shorts floor, ADR-023).
`maxSeconds` is new: `TAPIR_ONBOARD_MAX_VIDEO_SECONDS` (default 14400 = 4h) — drops the
multi-hour livestream VODs that pass the live filter once ended.
- `minSeconds<=0` and `maxSeconds<=0` each disable that bound (so `0/0` == the old
newest-first behaviour, the reversibility lever).
- The burst switches to this method; `NewestUnsummarizedVideoIDs` is removed (fully superseded —
`OnboardBurstVideoIDs(., 0, 0)` is identical pure-newest behaviour).
## 3. Stronger model for the burst
The burst summarizes only ≤3 videos, so a slower, stronger model is affordable exactly here.
- New config `TAPIR_ONBOARD_SUMMARIZER_MODEL` (default `iguana/gemma4-26b` — the brain-validated
homelab general-purpose model, already the ADR-022 fallback).
- Build a **burst-specific summarizer chain** that puts the onboard model **first**, then the
standard chain (primary → local fallback → cloud) as resilience, deduped. Wrap it in a
burst-specific `engineProcessor` reusing the same store/transcript-cache/sink — a pure wiring
choice, engine and ports unchanged (Clean Architecture, ADR-003).
- The `onboard` closure uses the burst processor instead of `app.Processor`.
- **Collapse cleanly**: when `OnboardSummarizerModel` is empty or equals `SummarizerModel`, the
onboard path reuses `app.Processor` (no separate chain) — the reversibility lever.
- Local-first preserved: the onboard model is a local alias; the cloud endpoint stays last in the
chain, so a client/NDA deployment with `TAPIR_CLOUD_FALLBACK_MODEL=""` keeps burst content
local too.
## 4. Behaviour spec + docs
- Add scenarios to `docs/use-cases/connect_account.feature` (the connect → burst flow): burst
skips a too-long/live video in favour of a reasonable-length one; burst summarizes with the
stronger model first. Map them in `scenarioCoverage` so `TestScenarioCoverage` stays green.
- Update `docs/architecture/architecture.md` (the onboarding-burst section) to describe the
junk-avoiding selection + the burst model override.
- ADR-028 in `DECISIONS.md` records the rationale (incl. the rejected cached-first lever).
## Success criteria
- `task check` green (fmt, vet, lint, `go test -p 1 ./...`).
- A unit test proves `OnboardBurstVideoIDs` drops a known too-long / sub-min video and keeps a
good one, newest-first, RLS-scoped, unsummarized-only.
- A test proves discovery persists `duration_s` and does not clobber it on re-upsert.
- A test proves the burst chain leads with the onboard model (then the standard chain).
- Config defaults + bounds tested (`OnboardMaxVideoSeconds`, `OnboardSummarizerModel`).
- No change to the rate gate, fetch pacing, or burst cap. `0/0` + empty model == prior behaviour.
## Explicitly NOT in this slice
- Cached-first selection (rejected, ADR-028).
- Any caption-availability *guarantee* (impossible pre-fetch).
- **Honest "taster" framing copy** ("summaries of a few of your videos to get you started — the
rest fill in over the coming days"). A good lever from the original brief, but it's a UI/copy
change with no backend dependency; deferred to a UI pass, tracked as an issue.
- Backfilling `duration_s` for existing rows via a migration (it backfills lazily on discovery).
- Return-nudges / digests (ADR-020: poisons the unprompted-return signal).
- Raising fetch throughput, multi-IP, or Whisper (out of scope; the gate is deliberate).
+9 -2
View File
@@ -24,12 +24,19 @@ Feature: Connect and manage video accounts
Then a discovery pass for my account is triggered right away
And I do not have to wait for the next scheduled pass to see my videos
Scenario: Connecting summarizes my newest videos right away
Scenario: Connecting summarizes my best recent videos right away
Given I have no connected video accounts
When I connect my YouTube account
Then up to the onboarding cap of my newest videos are summarized through the rate gate
Then up to the onboarding cap of my newest likely-good videos are summarized through the rate gate
And videos whose known duration is too short or too long are skipped
And the rest are left to the scheduled recency-bounded pass
Scenario: The onboarding burst summarizes with a stronger model
Given I have no connected video accounts
When I connect my YouTube account
Then the burst summarizes with the stronger onboarding model first
And the standard summarizer chain still follows as a fallback
Scenario: Tokens are never stored in the clear
When I connect any video account
Then no OAuth token value is stored in the database
+46
View File
@@ -0,0 +1,46 @@
Feature: Observability — timing and metrics for performance and UX (ADR-030, #15)
As the maintainer running Tapir for pilot users
I want timing and Prometheus metrics for the activities that drive performance and UX
So that I can see latency, model behaviour, and usage and feed the Stage-0 eval gate
# AI metrics are the priority (ADR-030 R3). Each scenario maps to a Go test in
# test/acceptance/scenario_coverage_test.go (the BDD name-coverage gate).
Scenario: Summarization latency is recorded per endpoint
Given the summarizer runs a transcript through its endpoint chain
When an endpoint returns a parseable summary
Then the summarize latency is recorded with the model, outcome "success", and whether it was a fallback
Scenario: A failing summarizer endpoint records its failure outcome
Given the summarizer runs a transcript through its endpoint chain
When an endpoint errors or returns unparseable output
Then the summarize latency is recorded with outcome "error" or "parse_error" before the chain advances
Scenario: Caption fetch latency is recorded by outcome
Given a caption fetch is attempted for a video
When it resolves to captions, no captions, or a rate limit
Then the caption-fetch latency is recorded labelled by that outcome
Scenario: LLM token usage is recorded from the completion
Given an LLM completion returns a usage block with prompt and completion tokens
When the client finishes the call
Then the prompt and completion tokens are recorded for that model
Scenario: Q&A answer latency is recorded
Given a user asks a question about a video
When the answer is produced from the stored transcript
Then the chat answer latency is recorded for the answering model
Scenario: HTTP requests are counted by route, method, and status
Given the metrics HTTP middleware wraps the app
When a request is served against a registered route
Then it is counted and timed under the bounded route pattern, not the raw path
Scenario: A successful login is counted
Given a user completes the OIDC callback and a session is established
Then the login counter is incremented
Scenario: The metrics endpoint is not on the public app port
Given the service is running
When the public app mux is inspected
Then it exposes no /metrics route metrics are served on the dedicated metrics port only
+11 -1
View File
@@ -9,23 +9,33 @@ require (
github.com/go-jose/go-jose/v4 v4.1.4
github.com/golang-migrate/migrate/v4 v4.19.1
github.com/jackc/pgx/v5 v5.9.2
github.com/prometheus/client_golang v1.23.2
github.com/prometheus/client_model v0.6.2
github.com/stretchr/testify v1.11.1
golang.org/x/crypto v0.45.0
golang.org/x/oauth2 v0.36.0
golang.org/x/time v0.15.0
)
require (
github.com/beorn7/perks v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
github.com/kylelemons/godebug v1.1.0 // indirect
github.com/lib/pq v1.10.9 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
github.com/prometheus/common v0.66.1 // indirect
github.com/prometheus/procfs v0.16.1 // indirect
github.com/rogpeppe/go-internal v1.15.0 // indirect
github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect
go.yaml.in/yaml/v2 v2.4.2 // indirect
golang.org/x/sync v0.18.0 // indirect
golang.org/x/sys v0.41.0 // indirect
golang.org/x/text v0.31.0 // indirect
google.golang.org/protobuf v1.36.8 // indirect
gopkg.in/yaml.v3 v3.0.1 // indirect
)
+26 -6
View File
@@ -4,6 +4,10 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERo
github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
github.com/a-h/templ v0.3.1020 h1:ypAT/L5ySWEnZ6Zft/5yfoWXYYkhFNvEFOeeqecg4tw=
github.com/a-h/templ v0.3.1020/go.mod h1:A2DlK61v+K+NRoGnhmYbNYVmtYHcFO5/AisMvBdDxTM=
github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M=
github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE=
@@ -37,8 +41,8 @@ github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
github.com/golang-migrate/migrate/v4 v4.19.1 h1:OCyb44lFuQfYXYLx1SCxPZQGU7mcaZ7gH9yH4jSFbBA=
github.com/golang-migrate/migrate/v4 v4.19.1/go.mod h1:CTcgfjxhaUtsLipnLoQRWCrjYXycRz/g5+RWDuYgPrE=
github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa h1:s+4MhCQ6YrzisK6hFJUX53drDT4UsSW3DEhKn0ifuHw=
github.com/jackc/pgerrcode v0.0.0-20220416144525-469b46aa5efa/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
@@ -49,10 +53,14 @@ github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0=
github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk=
github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
github.com/lib/pq v1.10.9 h1:YXG7RB+JIjhP29X+OtkiDnYaXQwpS4JEWq7dtCCRUEw=
github.com/lib/pq v1.10.9/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
@@ -61,6 +69,8 @@ github.com/moby/term v0.5.0 h1:xt8Q1nalod/v7BqbG21f8mQPqH+xAaC9C3N3wfWbVP0=
github.com/moby/term v0.5.0/go.mod h1:8FzsFHVUBGZdbDsJw/ot+X+d5HLUbvklYLJ9uGfcI3Y=
github.com/morikuni/aec v1.0.0 h1:nP9CBfwrvYnBRgY6qfDQkygYDmYwOilePFkwzv4dU8A=
github.com/morikuni/aec v1.0.0/go.mod h1:BbKIizmSmc5MMPqRYbxO4ZU0S0+P200+tUnFx7PXmsc=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
github.com/opencontainers/image-spec v1.1.0 h1:8SG7/vwALn54lVB/0yZ/MMwhFrPYtpEHQb2IpWsCzug=
@@ -70,6 +80,14 @@ github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
github.com/rogpeppe/go-internal v1.15.0 h1:D0RCU5rMAp+SpgkiNdrjfJ+LX4J1M32V2NeCY7EJ6hc=
github.com/rogpeppe/go-internal v1.15.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -91,8 +109,8 @@ go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mx
go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0=
go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
golang.org/x/crypto v0.45.0 h1:jMBrvKuj23MTlT0bQEOBcAE0mjg8mK9RXFhRH6nyF3Q=
golang.org/x/crypto v0.45.0/go.mod h1:XTGrrkGJve7CYK7J8PEww4aY7gM3qMCElcJQ8n8JdX4=
go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I=
@@ -103,6 +121,8 @@ golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM=
golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc=
google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+8
View File
@@ -13,8 +13,12 @@ package chat
import (
"context"
"fmt"
"log/slog"
"strings"
"time"
"unicode/utf8"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
)
// Completer is the minimal LLM chat surface the Service needs. *llm.Client
@@ -113,10 +117,14 @@ func (s *Service) Answer(ctx context.Context, req Request) (Reply, error) {
system := buildSystem(transcript, truncated)
user := buildUser(req.History, req.Question)
start := time.Now()
out, err := s.newClient(model).Complete(ctx, system, user)
if err != nil {
return Reply{}, fmt.Errorf("chat: %s: %w", model, err)
}
dur := time.Since(start)
metrics.ObserveChat(model, dur)
slog.Default().Info("chat answer", "model", model, "elapsed_ms", dur.Milliseconds())
answer := strings.TrimSpace(out)
if answer == "" {
return Reply{}, fmt.Errorf("chat: %s returned an empty answer", model)
+16
View File
@@ -32,6 +32,7 @@ type Client struct {
model string
maxTokens int
httpClient *http.Client
usageHook func(model string, prompt, completion int)
}
// Option configures a Client at construction. Variadic so the existing 4-arg
@@ -50,6 +51,14 @@ func WithMaxTokens(n int) Option {
}
}
// WithUsageHook registers a callback fired after a successful completion with the
// model and the prompt/completion token counts from the response usage block. It
// keeps this copied, stdlib-only package (ADR-004) decoupled from metrics: the
// caller wires it to internal/metrics, the client imports nothing. nil is ignored.
func WithUsageHook(fn func(model string, prompt, completion int)) Option {
return func(c *Client) { c.usageHook = fn }
}
// New constructs a Client.
func New(baseURL, apiKey, model string, timeout time.Duration, opts ...Option) *Client {
c := &Client{
@@ -81,6 +90,10 @@ type chatResponse struct {
Choices []struct {
Message message `json:"message"`
} `json:"choices"`
Usage struct {
PromptTokens int `json:"prompt_tokens"`
CompletionTokens int `json:"completion_tokens"`
} `json:"usage"`
}
// Complete sends a system + user message and returns the assistant's reply.
@@ -152,5 +165,8 @@ func (c *Client) Complete(ctx context.Context, system, user string) (string, err
if len(cr.Choices) == 0 {
return "", fmt.Errorf("LLM returned no choices")
}
if c.usageHook != nil {
c.usageHook(c.model, cr.Usage.PromptTokens, cr.Usage.CompletionTokens)
}
return cr.Choices[0].Message.Content, nil
}
+24
View File
@@ -85,6 +85,30 @@ func TestClient_WithMaxTokens(t *testing.T) {
}
}
// TestClient_UsageHookRecordsTokens: the usage hook fires with the model and the
// prompt/completion token counts parsed from the response usage block.
func TestClient_UsageHookRecordsTokens(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
_ = json.NewEncoder(w).Encode(map[string]any{
"choices": []map[string]any{{"message": map[string]any{"content": "ok"}}},
"usage": map[string]any{"prompt_tokens": 123, "completion_tokens": 45},
})
}))
defer srv.Close()
var gotModel string
var gotPrompt, gotCompletion int
c := New(srv.URL, "", "test-model", 10*time.Second, WithUsageHook(func(model string, p, comp int) {
gotModel, gotPrompt, gotCompletion = model, p, comp
}))
if _, err := c.Complete(context.Background(), "sys", "user"); err != nil {
t.Fatalf("Complete: %v", err)
}
if gotModel != "test-model" || gotPrompt != 123 || gotCompletion != 45 {
t.Errorf("usage hook got (%q, %d, %d), want (test-model, 123, 45)", gotModel, gotPrompt, gotCompletion)
}
}
func TestClient_ReturnsErrorOnNon200(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "overloaded", http.StatusServiceUnavailable)
+12 -6
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"sort"
"time"
"github.com/jackc/pgx/v5"
)
@@ -32,7 +33,12 @@ type UserActiveWeeks struct {
// Scope note: the enumeration covers users with a Dex identity (the web users the
// gate is about). A CLI-only user created by the store sink without an identity
// row would not appear — out of scope for this gate.
func (s *Store) ActiveWeeks(ctx context.Context) ([]UserActiveWeeks, error) {
// ActiveWeeks counts each user's distinct active weeks from `since` onward. A zero
// `since` means no lower bound (count all history). The Stage-0 gate baseline is
// set by the caller (the report command) to the date real usage tracking began,
// so pre-launch noise — testing, the period the pilot was blocked — does not count
// toward the return-usage signal (ADR-016).
func (s *Store) ActiveWeeks(ctx context.Context, since time.Time) ([]UserActiveWeeks, error) {
userIDs, err := s.identityUserIDs(ctx)
if err != nil {
return nil, err
@@ -40,7 +46,7 @@ func (s *Store) ActiveWeeks(ctx context.Context) ([]UserActiveWeeks, error) {
out := make([]UserActiveWeeks, 0, len(userIDs))
for _, uid := range userIDs {
row, err := s.activeWeeksFor(ctx, uid)
row, err := s.activeWeeksFor(ctx, uid, since)
if err != nil {
return nil, err
}
@@ -85,18 +91,18 @@ func (s *Store) identityUserIDs(ctx context.Context) ([]string, error) {
// activeWeeksFor counts one user's distinct active weeks (reads UNION acts) and
// reads their display name, RLS-scoped via withUser. The UNION dedups a week that
// has both a login and an action so it counts once.
func (s *Store) activeWeeksFor(ctx context.Context, userID string) (UserActiveWeeks, error) {
func (s *Store) activeWeeksFor(ctx context.Context, userID string, since time.Time) (UserActiveWeeks, error) {
res := UserActiveWeeks{UserID: userID}
if err := s.withUser(ctx, userID, func(tx pgx.Tx) error {
if err := tx.QueryRow(ctx,
`WITH weeks AS (
SELECT date_trunc('week', seen_at) AS wk
FROM login_events WHERE user_id = $1
FROM login_events WHERE user_id = $1 AND seen_at >= $2
UNION
SELECT date_trunc('week', acted_at)
FROM summary_actions WHERE user_id = $1
FROM summary_actions WHERE user_id = $1 AND acted_at >= $2
)
SELECT count(DISTINCT wk) FROM weeks`, userID).Scan(&res.ActiveWeeks); err != nil {
SELECT count(DISTINCT wk) FROM weeks`, userID, since).Scan(&res.ActiveWeeks); err != nil {
return fmt.Errorf("store: count active weeks: %w", err)
}
if err := tx.QueryRow(ctx,
+29 -2
View File
@@ -3,6 +3,7 @@ package store_test
import (
"context"
"testing"
"time"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/stretchr/testify/require"
@@ -54,7 +55,7 @@ func TestActiveWeeksCountsDistinctWeeksAcrossReadsAndActs(t *testing.T) {
($1, 'vid-2', 'saved', '2026-01-19T18:00:00Z')`, userA)
require.NoError(t, err)
got, err := s.ActiveWeeks(ctx)
got, err := s.ActiveWeeks(ctx, time.Time{}) // zero since = no lower bound
require.NoError(t, err)
require.Len(t, got, 2, "both identity users must appear")
@@ -72,7 +73,33 @@ func TestActiveWeeksEmptyWhenNoUsers(t *testing.T) {
s := newStore(t)
resetDB(t, rawPool(t))
got, err := s.ActiveWeeks(ctx)
got, err := s.ActiveWeeks(ctx, time.Time{})
require.NoError(t, err)
require.Empty(t, got)
}
// TestActiveWeeksExcludesBeforeGateStart proves the baseline cutoff: activity
// before `since` does not count, so pre-launch noise (testing, the pilot's blocked
// period) is excluded from the Stage-0 return-usage gate (ADR-016).
func TestActiveWeeksExcludesBeforeGateStart(t *testing.T) {
ctx := context.Background()
s := newStore(t)
p := rawPool(t)
resetDB(t, p)
seedReportUser(t, p, userA, "subject-a", "Ada")
// One read well before the baseline, two reads in distinct weeks after it.
_, err := p.Exec(ctx,
`INSERT INTO login_events (user_id, seen_at) VALUES
($1, '2026-05-01T09:00:00Z'),
($1, '2026-06-12T09:00:00Z'),
($1, '2026-06-19T09:00:00Z')`, userA)
require.NoError(t, err)
since := time.Date(2026, 6, 11, 0, 0, 0, 0, time.UTC)
got, err := s.ActiveWeeks(ctx, since)
require.NoError(t, err)
require.Len(t, got, 1)
require.Equal(t, 2, got[0].ActiveWeeks, "only the two post-baseline weeks count; the May read is excluded")
}
+35 -14
View File
@@ -46,15 +46,16 @@ func (s *Store) UpsertVideo(ctx context.Context, v domain.Video) (string, error)
}
if err := tx.QueryRow(ctx,
`INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at, channel_title)
VALUES ($1, $2, $3, $4, $5, $6, $7)
`INSERT INTO videos (user_id, provider, provider_video_id, title, url, published_at, channel_title, duration_s)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
ON CONFLICT (user_id, provider, provider_video_id) DO UPDATE SET
title = EXCLUDED.title,
url = EXCLUDED.url,
published_at = EXCLUDED.published_at,
channel_title = COALESCE(NULLIF(EXCLUDED.channel_title, ''), videos.channel_title)
channel_title = COALESCE(NULLIF(EXCLUDED.channel_title, ''), videos.channel_title),
duration_s = COALESCE(EXCLUDED.duration_s, videos.duration_s)
RETURNING id`,
v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt), v.ChannelTitle,
v.UserID, provider, v.ProviderVideoID, v.Title, v.URL, nullTime(v.PublishedAt), v.ChannelTitle, nullDuration(v.DurationSeconds),
).Scan(&id); err != nil {
return fmt.Errorf("store: upsert video: %w", err)
}
@@ -74,12 +75,27 @@ func nullTime(t time.Time) *time.Time {
return &t
}
// NewestUnsummarizedVideoIDs returns up to limit of the user's videos that have
// no summary yet, newest first (published_at DESC, NULLS LAST). It caps the
// connect-time onboarding burst (Feature 1) at a fixed count: the caller marks
// these for summarization through the shared rate gate. RLS-scoped via withUser,
// so it only ever sees the requesting user's rows. limit <= 0 returns nil.
func (s *Store) NewestUnsummarizedVideoIDs(ctx context.Context, userID string, limit int) ([]string, error) {
// nullDuration maps an unknown duration (0) to SQL NULL so the upsert's
// COALESCE(EXCLUDED.duration_s, videos.duration_s) preserves a previously-known
// value instead of clobbering it with 0 (ADR-028; the channel_title backfill
// stance, migration 014).
func nullDuration(seconds int) *int {
if seconds <= 0 {
return nil
}
return &seconds
}
// OnboardBurstVideoIDs returns up to limit of the user's unsummarized videos for
// the connect-time onboarding burst (ADR-028), newest-first but quality-aware: a
// video is excluded when its duration is KNOWN and outside [minSeconds, maxSeconds]
// — dropping Shorts (below min) and multi-hour livestream VODs (above max) that
// would waste a scarce caption fetch on a poor first impression. A NULL/unknown
// duration is kept (degrade-open) but ranked AFTER known-good rows, so a freshly
// enriched good pick wins when both exist. minSeconds<=0 / maxSeconds<=0 each
// disable that bound (0/0 == pure newest-first, the reversibility lever).
// RLS-scoped via withUser; limit <= 0 returns nil.
func (s *Store) OnboardBurstVideoIDs(ctx context.Context, userID string, limit, minSeconds, maxSeconds int) ([]string, error) {
if limit <= 0 {
return nil, nil
}
@@ -92,16 +108,21 @@ func (s *Store) NewestUnsummarizedVideoIDs(ctx context.Context, userID string, l
AND NOT EXISTS (
SELECT 1 FROM summaries su
WHERE su.user_id = v.user_id AND su.video_id = v.id)
ORDER BY v.published_at DESC NULLS LAST, v.seen_at DESC
LIMIT $2`, userID, limit)
AND NOT (
v.duration_s IS NOT NULL
AND ( ($3 > 0 AND v.duration_s < $3)
OR ($4 > 0 AND v.duration_s > $4) ))
ORDER BY (v.duration_s IS NOT NULL) DESC,
v.published_at DESC NULLS LAST, v.seen_at DESC
LIMIT $2`, userID, limit, minSeconds, maxSeconds)
if err != nil {
return fmt.Errorf("store: newest unsummarized: %w", err)
return fmt.Errorf("store: onboard burst videos: %w", err)
}
defer rows.Close()
for rows.Next() {
var id string
if err := rows.Scan(&id); err != nil {
return fmt.Errorf("store: scan newest unsummarized: %w", err)
return fmt.Errorf("store: scan onboard burst video: %w", err)
}
ids = append(ids, id)
}
+63 -14
View File
@@ -52,6 +52,36 @@ func TestUpsertVideo_ReturnsStableID(t *testing.T) {
require.Equal(t, 1, count, "must not duplicate the row")
}
func TestUpsertVideo_PersistsAndPreservesDuration(t *testing.T) {
ctx := context.Background()
s := newStore(t)
resetDB(t, rawPool(t))
// First upsert carries a known duration (ADR-028: discovery enriches it).
v := ytVideo(userA, "dur0000001x", "with duration")
v.DurationSeconds = 750
id, err := s.UpsertVideo(ctx, v)
require.NoError(t, err)
p := rawPool(t)
readDuration := func() *int {
var d *int
require.NoError(t, p.QueryRow(ctx, `SELECT duration_s FROM videos WHERE id = $1`, id).Scan(&d))
return d
}
require.NotNil(t, readDuration())
require.Equal(t, 750, *readDuration(), "duration must persist")
// A later upsert that does NOT know the duration (0) must not clobber it —
// the channel_title backfill stance (migration 014): COALESCE-preserve.
v2 := ytVideo(userA, "dur0000001x", "title updated, duration unknown")
v2.DurationSeconds = 0
_, err = s.UpsertVideo(ctx, v2)
require.NoError(t, err)
require.NotNil(t, readDuration(), "a 0/unknown re-upsert must not erase a known duration")
require.Equal(t, 750, *readDuration())
}
func TestUpsertVideo_IDMatchesSummaryDedup(t *testing.T) {
ctx := context.Background()
s := newStore(t)
@@ -82,36 +112,55 @@ func TestUpsertVideo_PerUserIsolation(t *testing.T) {
require.NotEqual(t, idA, idB, "same provider video for two users must be two distinct rows")
}
func TestNewestUnsummarizedVideoIDs(t *testing.T) {
func TestOnboardBurstVideoIDs(t *testing.T) {
ctx := context.Background()
s := newStore(t)
resetDB(t, rawPool(t))
mk := func(user, pid string, day int) string {
mk := func(user, pid string, day, dur int) string {
v := ytVideo(user, pid, pid)
v.PublishedAt = time.Date(2026, 6, day, 12, 0, 0, 0, time.UTC)
v.DurationSeconds = dur // 0 == unknown (NULL)
id, err := s.UpsertVideo(ctx, v)
require.NoError(t, err)
return id
}
_ = mk(userA, "a1vid000001", 1)
id2 := mk(userA, "a2vid000002", 2)
id3 := mk(userA, "a3vid000003", 3)
id4 := mk(userA, "a4vid000004", 4)
mk(userB, "b1vid000009", 9) // userB's newest — must never leak via RLS
summarized := mk(userA, "summ0000001", 6, 600) // newest known-good, but already summarized
good1 := mk(userA, "good0000001", 5, 600) // 10m, newest UNsummarized known-good
tooLong := mk(userA, "toolong0001", 4, 20000) // > maxSeconds -> dropped
tooShort := mk(userA, "tooshort001", 3, 30) // < minSeconds -> dropped
unknown := mk(userA, "unknown0001", 2, 0) // NULL duration -> kept, ranked last
good2 := mk(userA, "good0000002", 1, 800) // known-good but oldest
mk(userB, "bvid0000009", 9, 600) // userB -> must not leak via RLS
// The newest (v4) is summarized, so it's excluded from "unsummarized".
require.NoError(t, s.Deliver(ctx, summary(userA, id4, "done")))
// The newest video is summarized, so it is excluded from the burst.
require.NoError(t, s.Deliver(ctx, summary(userA, summarized, "done")))
// Cap 2, newest-first unsummarized: v3 then v2 (v4 excluded; userB excluded).
got, err := s.NewestUnsummarizedVideoIDs(ctx, userA, 2)
const minSec, maxSec = 60, 14400
// Known-good ranked before unknown, each newest-first within its group; the
// too-long and too-short videos are excluded by their known duration.
got, err := s.OnboardBurstVideoIDs(ctx, userA, 5, minSec, maxSec)
require.NoError(t, err)
require.Equal(t, []string{id3, id2}, got)
require.Equal(t, []string{good1, good2, unknown}, got,
"known-good first (newest-first), then unknown-duration; junk excluded")
none, err := s.NewestUnsummarizedVideoIDs(ctx, userA, 0)
// Cap is honoured.
capped, err := s.OnboardBurstVideoIDs(ctx, userA, 2, minSec, maxSec)
require.NoError(t, err)
require.Empty(t, none, "limit 0 returns nothing")
require.Equal(t, []string{good1, good2}, capped)
// Bounds disabled (0/0) == pure newest-first, nothing excluded.
all, err := s.OnboardBurstVideoIDs(ctx, userA, 10, 0, 0)
require.NoError(t, err)
require.ElementsMatch(t, []string{good1, tooLong, tooShort, unknown, good2}, all,
"0/0 bounds disable the duration filter (prior newest-first behaviour)")
// limit <= 0 returns nothing.
none, err := s.OnboardBurstVideoIDs(ctx, userA, 0, minSec, maxSec)
require.NoError(t, err)
require.Empty(t, none)
}
func TestUpsertVideoPersistsChannelAndDistinctChannels(t *testing.T) {
+10 -1
View File
@@ -13,11 +13,13 @@ import (
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
"time"
"unicode/utf8"
"gitea.d-ma.be/mathias/tapir/internal/domain"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
)
// Completer is the minimal LLM chat surface the Summarizer needs.
@@ -95,16 +97,23 @@ func (s *Summarizer) Summarize(ctx context.Context, v domain.Video, t domain.Tra
var errs []error
for i, ep := range s.endpoints {
fallback := i > 0
start := time.Now()
out, err := ep.Client.Complete(ctx, systemPrompt, user)
dur := time.Since(start)
if err != nil {
metrics.ObserveSummarize(ep.Model, "error", fallback, dur)
errs = append(errs, fmt.Errorf("%s/%s call: %w", ep.Provider, ep.Model, err))
continue
}
sum, perr := s.build(v, ep, i > 0, out)
sum, perr := s.build(v, ep, fallback, out)
if perr != nil {
metrics.ObserveSummarize(ep.Model, "parse_error", fallback, dur)
errs = append(errs, fmt.Errorf("%s/%s output: %w", ep.Provider, ep.Model, perr))
continue
}
metrics.ObserveSummarize(ep.Model, "success", fallback, dur)
slog.Default().Info("summarized", "model", ep.Model, "fallback", fallback, "elapsed_ms", dur.Milliseconds())
return sum, nil
}
return domain.Summary{}, fmt.Errorf("summarize: all %d endpoint(s) failed: %w", len(s.endpoints), errors.Join(errs...))
@@ -7,13 +7,36 @@ package summarizer
import (
"context"
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
"gitea.d-ma.be/mathias/tapir/internal/domain"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
"gitea.d-ma.be/mathias/tapir/internal/ports"
)
// TestSummarizerRecordsMetric verifies the summarizer→metrics wiring (ADR-030)
// black-box: after a successful summarize, the public /metrics scrape shows a
// success observation for that endpoint's model.
func TestSummarizerRecordsMetric(t *testing.T) {
const model = "metrics-test-model"
s := New(Endpoint{Client: &fakeClient{reply: goodReply}, Provider: "local", Model: model}, nil)
if _, err := s.Summarize(context.Background(), testVideo(), testTranscript()); err != nil {
t.Fatalf("Summarize: %v", err)
}
rec := httptest.NewRecorder()
metrics.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil))
body := rec.Body.String()
if !strings.Contains(body, `tapir_summarize_duration_seconds`) ||
!strings.Contains(body, `model="`+model+`"`) ||
!strings.Contains(body, `outcome="success"`) {
t.Errorf("metrics scrape missing summarize success for %s", model)
}
}
// compile-time check: Summarizer satisfies the port.
var _ ports.Summarizer = (*Summarizer)(nil)
+31
View File
@@ -7,10 +7,13 @@ import (
"encoding/xml"
"fmt"
"io"
"log/slog"
"net/http"
"strings"
"time"
"gitea.d-ma.be/mathias/tapir/internal/domain"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
)
// defaultPlayerBaseURL is the InnerTube / watch-page host. Overridable via
@@ -43,7 +46,35 @@ const maxCaptionBytes = 16 << 20 // 16 MiB
// fetch, or an unparseable body all yield SourceNone rather than an error. Only
// genuine transport (network) faults return an error. Audio download and
// speech-to-text remain absent (ADR-007).
// FetchTranscript times the caption fetch and records its latency by outcome
// (ADR-030) before returning. Transport errors are surfaced to the caller and not
// recorded as an outcome (logged upstream); the three resolved outcomes
// captions|none|rate_limited are the ones that consume the scarce fetch budget.
func (a *Adapter) FetchTranscript(ctx context.Context, v domain.Video) (domain.Transcript, error) {
start := time.Now()
tr, err := a.fetchTranscript(ctx, v)
if err == nil {
dur := time.Since(start)
outcome := captionOutcome(tr.Source)
metrics.ObserveCaptionFetch(outcome, dur)
slog.Default().Info("caption fetch", "video", v.ProviderVideoID, "outcome", outcome, "elapsed_ms", dur.Milliseconds())
}
return tr, err
}
// captionOutcome maps a transcript source to the metric outcome label.
func captionOutcome(s domain.TranscriptSource) string {
switch s {
case domain.SourceCaptions:
return "captions"
case domain.SourceRateLimited:
return "rate_limited"
default:
return "none"
}
}
func (a *Adapter) fetchTranscript(ctx context.Context, v domain.Video) (domain.Transcript, error) {
client := a.plainClient()
tracks, err := a.captionTracks(ctx, client, v.ProviderVideoID)
+4
View File
@@ -302,6 +302,10 @@ func (a *Adapter) filterLowValue(ctx context.Context, client *http.Client, video
if m.seconds > 0 && m.seconds < a.cfg.MinVideoSeconds {
continue // Short / sub-threshold clip
}
// Carry the duration we already fetched onto the kept video so the store
// can persist it (ADR-028) — the burst's length-aware selection depends on
// it. Discarding it here was the gap the onboarding investigation found.
v.DurationSeconds = m.seconds
kept = append(kept, v)
}
return kept
@@ -224,6 +224,11 @@ func TestNewVideosFiltersShortsAndLive(t *testing.T) {
if len(vids) != 1 || vids[0].ProviderVideoID != "long1" {
t.Fatalf("expected only long1 to survive the filter, got %+v", vids)
}
// The duration fetched for the filter is carried onto the kept video so the
// store can persist it (ADR-028) instead of discarding it.
if vids[0].DurationSeconds != 750 {
t.Fatalf("kept video DurationSeconds = %d, want 750 (PT12M30S)", vids[0].DurationSeconds)
}
}
// TestNewVideosNoFilterWhenDisabled: MinVideoSeconds=0 keeps the pre-ADR-023
+34
View File
@@ -120,6 +120,21 @@ type Config struct {
// caption rate gate (ADR-014) — the cap bounds count, never the pacing. Default 3.
OnboardSummarizeCount int
// OnboardSummarizerModel is the summarizer alias the connect-time burst leads
// its chain with (ADR-028) — a stronger model is affordable on the ≤3 summaries
// that form a new user's first impression. It heads a burst-specific chain;
// the standard chain (ADR-022) follows as resilience. Empty (or equal to
// SummarizerModel) collapses the burst back onto the shared processor — the
// reversibility lever. Default iguana/gemma4-26b (the brain-validated model).
OnboardSummarizerModel string
// OnboardMaxVideoSeconds upper-bounds the duration of a video the onboarding
// burst will pick (ADR-028), so the burst does not spend a scarce caption fetch
// on a multi-hour livestream VOD that passed the live filter once it ended. Only
// a KNOWN duration outside [MinVideoSeconds, this] is dropped; a NULL/unknown
// duration is kept (degrade-open). 0 disables the upper bound. Default 14400 (4h).
OnboardMaxVideoSeconds int
// DiscoveryInterval, when > 0, makes `serve` run in-process scheduled discovery
// for ALL users on that cadence (ADR-018). Zero/unset = disabled, so dev and
// tests never auto-fetch. Single-replica assumption — see cmdServe.
@@ -128,6 +143,11 @@ type Config struct {
// HTTPAddr is the listen address for `tapir serve` (the Stage-0 web UI).
HTTPAddr string
// MetricsAddr is the listen address for the Prometheus /metrics endpoint
// (ADR-030). A SEPARATE port from HTTPAddr so /metrics is never exposed on the
// public app — only scraped in-cluster. Empty disables the metrics server.
MetricsAddr string
// PublicURL is the externally-reachable base URL of the deployed service,
// e.g. "https://tapir.d-ma.be". Used to build absolute links handed to humans
// (the `tapir invite` URL). No trailing slash is assumed — callers trim it.
@@ -163,12 +183,15 @@ const (
defaultYTConnectRedirectURL = "https://tapir.d-ma.be/oauth/youtube/callback"
defaultOAuthRedirectAddr = "localhost:8080"
defaultHTTPAddr = ":8080"
defaultMetricsAddr = ":9090"
defaultFetchBackoff = time.Hour
defaultFetchRate = 2 * time.Second
defaultPublicURL = "https://tapir.d-ma.be"
defaultAutoSummarizeWindow = 7 * 24 * time.Hour
defaultOnboardSummarizeCount = 3
maxOnboardSummarizeCount = 5
defaultOnboardSummarizerModel = "iguana/gemma4-26b"
defaultOnboardMaxVideoSeconds = 14400 // 4h
)
// Load reads the environment into a Config, applying defaults. It does not
@@ -181,6 +204,7 @@ func Load() (Config, error) {
GatewayURL: envOr("TAPIR_GATEWAY_URL", defaultGatewayURL),
GatewayKey: os.Getenv("TAPIR_GATEWAY_KEY"),
SummarizerModel: envOr("TAPIR_SUMMARIZER_MODEL", defaultSummarizerModel),
OnboardSummarizerModel: lookupOr("TAPIR_ONBOARD_SUMMARIZER_MODEL", defaultOnboardSummarizerModel),
FallbackModel: lookupOr("TAPIR_FALLBACK_MODEL", defaultFallbackModel),
CloudFallbackModel: lookupOr("TAPIR_CLOUD_FALLBACK_MODEL", defaultCloudFallbackModel),
DBDSN: os.Getenv("TAPIR_DB_DSN"),
@@ -191,6 +215,7 @@ func Load() (Config, error) {
SecretsFile: envOr("TAPIR_SECRETS_FILE", defaultSecretsFile()),
OAuthRedirectAddr: envOr("TAPIR_OAUTH_REDIRECT_ADDR", defaultOAuthRedirectAddr),
HTTPAddr: envOr("TAPIR_HTTP_ADDR", defaultHTTPAddr),
MetricsAddr: lookupOr("TAPIR_METRICS_ADDR", defaultMetricsAddr),
PublicURL: envOr("TAPIR_PUBLIC_URL", defaultPublicURL),
OIDCIssuer: os.Getenv("TAPIR_OIDC_ISSUER"),
DexClientID: os.Getenv("TAPIR_DEX_CLIENT_ID"),
@@ -286,6 +311,15 @@ func Load() (Config, error) {
}
c.OnboardSummarizeCount = onboard
onboardMax, err := intOr("TAPIR_ONBOARD_MAX_VIDEO_SECONDS", defaultOnboardMaxVideoSeconds)
if err != nil {
return Config{}, err
}
if onboardMax < 0 {
onboardMax = 0
}
c.OnboardMaxVideoSeconds = onboardMax
return c, nil
}
+59
View File
@@ -214,3 +214,62 @@ func TestValidateForAuth_PassesWhenComplete(t *testing.T) {
t.Errorf("ValidateForAuth: unexpected error %v", err)
}
}
func TestLoad_OnboardSummarizerModel(t *testing.T) {
cases := []struct {
name, env string
set bool
want string
}{
{"default", "", false, defaultOnboardSummarizerModel},
{"explicit", "koala/some-model", true, "koala/some-model"},
{"empty disables (collapses to shared processor)", "", true, ""},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
env := map[string]string{}
if c.set {
env["TAPIR_ONBOARD_SUMMARIZER_MODEL"] = c.env
}
setEnv(t, env)
cfg, err := Load()
if err != nil {
t.Fatalf("Load: %v", err)
}
if cfg.OnboardSummarizerModel != c.want {
t.Fatalf("OnboardSummarizerModel = %q, want %q", cfg.OnboardSummarizerModel, c.want)
}
})
}
}
func TestLoad_OnboardMaxVideoSeconds(t *testing.T) {
cases := []struct {
name, env string
want int
}{
{"default", "", defaultOnboardMaxVideoSeconds},
{"explicit", "7200", 7200},
{"zero disables", "0", 0},
{"negative clamps to zero", "-9", 0},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
setEnv(t, map[string]string{"TAPIR_ONBOARD_MAX_VIDEO_SECONDS": c.env})
cfg, err := Load()
if err != nil {
t.Fatalf("Load: %v", err)
}
if cfg.OnboardMaxVideoSeconds != c.want {
t.Fatalf("OnboardMaxVideoSeconds = %d, want %d", cfg.OnboardMaxVideoSeconds, c.want)
}
})
}
}
func TestLoad_OnboardMaxVideoSecondsInvalid(t *testing.T) {
setEnv(t, map[string]string{"TAPIR_ONBOARD_MAX_VIDEO_SECONDS": "long"})
if _, err := Load(); err == nil {
t.Fatal("Load: want error for non-numeric TAPIR_ONBOARD_MAX_VIDEO_SECONDS")
}
}
+5
View File
@@ -77,6 +77,11 @@ type Video struct {
URL string
PublishedAt time.Time
SeenAt time.Time
// DurationSeconds is the video length in seconds, when known (fetched by the
// ADR-023 videos.list enrichment at discovery). 0 means unknown — the store
// preserves a previously-known value rather than overwriting it with 0, and
// the onboarding burst (ADR-028) treats unknown as degrade-open (kept).
DurationSeconds int
}
// Transcript is the text of a video (or a record that none was available).
+139
View File
@@ -0,0 +1,139 @@
// Package metrics is Tapir's Prometheus instrumentation (ADR-030, issue #15). It
// owns the collectors and a small typed API the rest of the app calls — adapters
// never touch prometheus types directly. Two themes:
//
// - HTTP/session: request count + latency by route (the matched pattern, so
// cardinality stays bounded), and logins.
// - AI (the priority): summarization latency by model/outcome/fallback, caption
// fetch latency by outcome, chat latency by model, and LLM token usage.
//
// Handler() is served on a dedicated port (never the public app port) so a scrape
// is in-cluster only. slog timing lines are emitted at the call sites too.
package metrics
import (
"net/http"
"strconv"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/prometheus/client_golang/prometheus/promhttp"
)
// latencyBuckets spans sub-second UI calls up to multi-minute model calls (a cold
// local model load is tens of seconds; the cloud fallback can be longer).
var latencyBuckets = []float64{0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 20, 30, 60, 120, 300}
var (
httpRequests = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "tapir_http_requests_total",
Help: "HTTP requests by method, matched route pattern, and status code.",
}, []string{"method", "route", "code"})
httpDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "tapir_http_request_duration_seconds",
Help: "HTTP request latency by method and matched route pattern.",
Buckets: []float64{0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5},
}, []string{"method", "route"})
logins = promauto.NewCounter(prometheus.CounterOpts{
Name: "tapir_logins_total",
Help: "Successful OIDC logins (session established).",
})
summarizeDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "tapir_summarize_duration_seconds",
Help: "Per-endpoint summarization latency by model, outcome (success|parse_error|error), and whether it was a fallback.",
Buckets: latencyBuckets,
}, []string{"model", "outcome", "fallback"})
captionFetchDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "tapir_caption_fetch_duration_seconds",
Help: "Caption fetch latency by outcome (captions|none|rate_limited).",
Buckets: latencyBuckets,
}, []string{"outcome"})
chatDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
Name: "tapir_chat_duration_seconds",
Help: "Per-video Q&A answer latency by model.",
Buckets: latencyBuckets,
}, []string{"model"})
llmTokens = promauto.NewCounterVec(prometheus.CounterOpts{
Name: "tapir_llm_tokens_total",
Help: "LLM tokens consumed by model and kind (prompt|completion).",
}, []string{"model", "kind"})
)
// Handler serves the Prometheus exposition format. Mount on the dedicated metrics
// port, never the public app mux.
func Handler() http.Handler { return promhttp.Handler() }
// IncLogin records a successful login.
func IncLogin() { logins.Inc() }
// ObserveSummarize records one summarization endpoint attempt.
func ObserveSummarize(model, outcome string, fallback bool, d time.Duration) {
summarizeDuration.WithLabelValues(model, outcome, strconv.FormatBool(fallback)).Observe(d.Seconds())
}
// ObserveCaptionFetch records one caption fetch by outcome.
func ObserveCaptionFetch(outcome string, d time.Duration) {
captionFetchDuration.WithLabelValues(outcome).Observe(d.Seconds())
}
// ObserveChat records one Q&A answer latency.
func ObserveChat(model string, d time.Duration) {
chatDuration.WithLabelValues(model).Observe(d.Seconds())
}
// RecordTokens records LLM token usage from a completion's usage block. Zero
// counts are skipped so a provider that omits usage adds nothing.
func RecordTokens(model string, prompt, completion int) {
if prompt > 0 {
llmTokens.WithLabelValues(model, "prompt").Add(float64(prompt))
}
if completion > 0 {
llmTokens.WithLabelValues(model, "completion").Add(float64(completion))
}
}
// HTTPMiddleware records request count + latency. It reads r.Pattern AFTER the
// inner handler routes (Go 1.22 sets it during ServeMux matching), so the label is
// the bounded registered pattern (e.g. "GET /v/{videoId}"), never the raw path
// with its high-cardinality ids. Unmatched requests bucket as "other".
func HTTPMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
sw := &statusWriter{ResponseWriter: w, code: http.StatusOK}
next.ServeHTTP(sw, r)
route := r.Pattern
if route == "" {
route = "other"
}
httpRequests.WithLabelValues(r.Method, route, strconv.Itoa(sw.code)).Inc()
httpDuration.WithLabelValues(r.Method, route).Observe(time.Since(start).Seconds())
})
}
// statusWriter captures the response status for the request-count label.
type statusWriter struct {
http.ResponseWriter
code int
wroteHeader bool
}
func (s *statusWriter) WriteHeader(code int) {
if !s.wroteHeader {
s.code = code
s.wroteHeader = true
}
s.ResponseWriter.WriteHeader(code)
}
func (s *statusWriter) Write(b []byte) (int, error) {
s.wroteHeader = true // an implicit 200
return s.ResponseWriter.Write(b)
}
+92
View File
@@ -0,0 +1,92 @@
package metrics
import (
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/testutil"
dto "github.com/prometheus/client_model/go"
"github.com/stretchr/testify/require"
)
// histCount reads a histogram child's observation count (testutil.ToFloat64 only
// works on counters/gauges; a histogram's WithLabelValues child is an Observer).
func histCount(t *testing.T, o prometheus.Observer) uint64 {
t.Helper()
m, ok := o.(prometheus.Metric)
require.True(t, ok, "histogram child must be a prometheus.Metric")
var d dto.Metric
require.NoError(t, m.Write(&d))
return d.GetHistogram().GetSampleCount()
}
// TestObserveSummarizeRecordsModelOutcomeFallback: a success observation lands on
// the right model/outcome/fallback series.
func TestObserveSummarizeRecordsModelOutcomeFallback(t *testing.T) {
before := histCount(t, summarizeDuration.WithLabelValues("koala/phi4-mini", "success", "false"))
ObserveSummarize("koala/phi4-mini", "success", false, 1200*time.Millisecond)
after := histCount(t, summarizeDuration.WithLabelValues("koala/phi4-mini", "success", "false"))
require.Equal(t, before+1, after, "one success observation recorded for the model")
}
// TestObserveSummarizeRecordsFailureOutcomes: error and parse_error are distinct
// series so a fallback chain's failures are visible.
func TestObserveSummarizeRecordsFailureOutcomes(t *testing.T) {
e0 := histCount(t, summarizeDuration.WithLabelValues("m", "error", "false"))
p0 := histCount(t, summarizeDuration.WithLabelValues("m", "parse_error", "false"))
ObserveSummarize("m", "error", false, time.Second)
ObserveSummarize("m", "parse_error", false, time.Second)
require.Equal(t, e0+1, histCount(t, summarizeDuration.WithLabelValues("m", "error", "false")))
require.Equal(t, p0+1, histCount(t, summarizeDuration.WithLabelValues("m", "parse_error", "false")))
}
func TestObserveCaptionFetchByOutcome(t *testing.T) {
b := histCount(t, captionFetchDuration.WithLabelValues("captions"))
ObserveCaptionFetch("captions", 3*time.Second)
require.Equal(t, b+1, histCount(t, captionFetchDuration.WithLabelValues("captions")))
}
func TestChatAnswerLatencyRecorded(t *testing.T) {
b := histCount(t, chatDuration.WithLabelValues("iguana/gemma4-26b"))
ObserveChat("iguana/gemma4-26b", 2*time.Second)
require.Equal(t, b+1, histCount(t, chatDuration.WithLabelValues("iguana/gemma4-26b")))
}
// TestRecordTokens: prompt + completion land on their kind series; zero is skipped.
func TestRecordTokens(t *testing.T) {
p0 := testutil.ToFloat64(llmTokens.WithLabelValues("m", "prompt"))
c0 := testutil.ToFloat64(llmTokens.WithLabelValues("m", "completion"))
RecordTokens("m", 100, 40)
RecordTokens("m", 0, 0) // skipped, no panic
require.Equal(t, p0+100, testutil.ToFloat64(llmTokens.WithLabelValues("m", "prompt")))
require.Equal(t, c0+40, testutil.ToFloat64(llmTokens.WithLabelValues("m", "completion")))
}
func TestLoginCounted(t *testing.T) {
b := testutil.ToFloat64(logins)
IncLogin()
require.Equal(t, b+1, testutil.ToFloat64(logins))
}
// TestHTTPMiddlewareRecordsByRoutePattern: the request is counted under the bounded
// registered pattern (r.Pattern after routing), not the raw path with its ids.
func TestHTTPMiddlewareRecordsByRoutePattern(t *testing.T) {
mux := http.NewServeMux()
mux.HandleFunc("GET /v/{videoId}", func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusTeapot)
})
h := HTTPMiddleware(mux)
before := testutil.ToFloat64(httpRequests.WithLabelValues("GET", "GET /v/{videoId}", "418"))
rec := httptest.NewRecorder()
h.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/v/abc-123", nil))
require.Equal(t, http.StatusTeapot, rec.Code)
after := testutil.ToFloat64(httpRequests.WithLabelValues("GET", "GET /v/{videoId}", "418"))
require.Equal(t, before+1, after, "counted under the pattern, not /v/abc-123")
require.Equal(t, float64(0), testutil.ToFloat64(httpRequests.WithLabelValues("GET", "/v/abc-123", "418")),
"raw path must never be a label value")
}
+9
View File
@@ -524,3 +524,12 @@ func TestListAutoModeBannerCopy(t *testing.T) {
require.Contains(t, html, "land gradually")
require.NotContains(t, html, "are not summarized automatically")
}
// TestMetricsNotOnPublicMux: the public app router exposes no /metrics route —
// Prometheus is served on the dedicated metrics port only (ADR-030, security R6).
func TestMetricsNotOnPublicMux(t *testing.T) {
app := newApp(t)
resetDB(t, rawPool(t))
rec := do(t, app, httptest.NewRequest(http.MethodGet, "/metrics", nil))
require.Equal(t, http.StatusNotFound, rec.Code, "/metrics must not be on the public mux")
}
+38 -44
View File
@@ -7,10 +7,11 @@
// Authentication is real (Dex OIDC) and is the only gate: any Dex-authenticated
// subject may sign in (ADR-012 dropped ADR-011's single-subject allowlist).
// Authorization/registration is layered on top in internal/web (an authenticated
// subject with no tapir user is routed to registration). Sessions are server-side
// (in-memory, fine for the single Stage-1 replica) addressed by an HMAC-signed
// (HS256) HttpOnly Secure SameSite=Lax cookie with a short TTL and sliding
// refresh. Tokens are never logged.
// subject with no tapir user is routed to registration). Sessions are STATELESS
// (ADR-029): the identity + expiry live inside an HMAC-signed (HS256) HttpOnly
// Secure SameSite=Lax persistent cookie with a long sliding TTL — no server-side
// table, so a deploy/restart never logs anyone out and the cookie also survives
// browser-close. Tokens are never logged; logout clears the cookie client-side.
//
// This is mcp-chassis's cousin but NOT the same code: mcp-chassis validates
// inbound Bearer JWTs for MCP APIs; this is a browser session login.
@@ -26,6 +27,7 @@ import (
"github.com/coreos/go-oidc/v3/oidc"
"golang.org/x/oauth2"
"gitea.d-ma.be/mathias/tapir/internal/metrics"
"gitea.d-ma.be/mathias/tapir/internal/web"
)
@@ -47,7 +49,11 @@ type Config struct {
}
const (
defaultSessionTTL = time.Hour
// defaultSessionTTL is generous and sliding: Tapir is a "check back tomorrow"
// reader, so a short TTL meant a re-login (full IdP redirect dance) on almost
// every visit. 30 days, slid forward on each request, keeps a regular user
// logged in indefinitely while an abandoned session still lapses.
defaultSessionTTL = 30 * 24 * time.Hour
pendingTTL = 10 * time.Minute
sessionCookie = "tapir_session"
loginPath = "/auth/login"
@@ -59,7 +65,6 @@ type DexAuth struct {
oauth *oauth2.Config
verifier *oidc.IDTokenVerifier
sessions *sessionStore
pending *pendingStore
secret []byte
sessionTTL time.Duration
@@ -127,7 +132,6 @@ func New(ctx context.Context, cfg Config, opts ...Option) (*DexAuth, error) {
RedirectURL: cfg.RedirectURL,
Scopes: []string{oidc.ScopeOpenID, "profile", "email"},
},
sessions: newSessionStore(),
pending: newPendingStore(),
secret: []byte(cfg.SessionSecret),
sessionTTL: defaultSessionTTL,
@@ -159,31 +163,31 @@ func (d *DexAuth) Middleware(h http.Handler) http.Handler {
h.ServeHTTP(w, r)
return
}
sid, ok := d.sessionID(r)
c, err := r.Cookie(sessionCookie)
if err != nil {
d.redirectUnauthenticated(w, r)
return
}
user, _, ok := d.decodeSession(c.Value, d.now())
if !ok {
d.redirectUnauthenticated(w, r)
return
}
if _, ok := d.sessions.get(sid, d.now()); !ok {
d.redirectUnauthenticated(w, r)
return
}
d.sessions.refresh(sid, d.now().Add(d.sessionTTL)) // sliding refresh
// Sliding refresh: re-issue the cookie with a fresh expiry so an active
// user never lapses (the expiry lives in the cookie, so sliding = re-sign).
d.setSessionCookie(w, d.encodeSession(user, d.now().Add(d.sessionTTL)))
h.ServeHTTP(w, r)
})
}
// CurrentUser resolves the authenticated principal from the session cookie.
// CurrentUser resolves the authenticated principal from the stateless cookie.
func (d *DexAuth) CurrentUser(r *http.Request) (web.User, bool) {
sid, ok := d.sessionID(r)
if !ok {
c, err := r.Cookie(sessionCookie)
if err != nil {
return web.User{}, false
}
data, ok := d.sessions.get(sid, d.now())
if !ok {
return web.User{}, false
}
return data.user, true
user, _, ok := d.decodeSession(c.Value, d.now())
return user, ok
}
func (d *DexAuth) handleLogin(w http.ResponseWriter, r *http.Request) {
@@ -248,23 +252,16 @@ func (d *DexAuth) handleCallback(w http.ResponseWriter, r *http.Request) {
}
_ = idToken.Claims(&claims) // email is best-effort; subject is the identity
sid, err := randToken()
if err != nil {
http.Error(w, "internal error", http.StatusInternalServerError)
return
}
d.sessions.put(sid, sessionData{
user: web.User{Subject: idToken.Subject, Email: claims.Email},
expiry: d.now().Add(d.sessionTTL),
})
d.setSessionCookie(w, sid)
user := web.User{Subject: idToken.Subject, Email: claims.Email}
d.setSessionCookie(w, d.encodeSession(user, d.now().Add(d.sessionTTL)))
metrics.IncLogin()
http.Redirect(w, r, "/", http.StatusFound)
}
func (d *DexAuth) handleLogout(w http.ResponseWriter, r *http.Request) {
if sid, ok := d.sessionID(r); ok {
d.sessions.delete(sid)
}
// Stateless sessions: clearing the cookie logs the browser out. There is no
// server-side record to delete (ADR-029); a copy of the cookie stays valid
// until its expiry — an accepted trade for the Stage-0 reader app.
d.clearSessionCookie(w)
// Land on the public landing page, not the login endpoint: a just-logged-out
// visitor should see /welcome, not be bounced straight back into a Dex login.
@@ -287,22 +284,19 @@ func (d *DexAuth) redirectToLogin(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, loginPath, http.StatusFound)
}
func (d *DexAuth) sessionID(r *http.Request) (string, bool) {
c, err := r.Cookie(sessionCookie)
if err != nil {
return "", false
}
return d.unsign(c.Value)
}
func (d *DexAuth) setSessionCookie(w http.ResponseWriter, sid string) {
// setSessionCookie writes the signed session value as a PERSISTENT cookie
// (Max-Age set), so it survives the browser/app being closed — a session cookie
// (no Max-Age) was dropped on iPhone Safari close, forcing re-login. value is the
// already-signed payload from encodeSession.
func (d *DexAuth) setSessionCookie(w http.ResponseWriter, value string) {
http.SetCookie(w, &http.Cookie{
Name: sessionCookie,
Value: d.sign(sid),
Value: value,
Path: "/",
HttpOnly: true,
Secure: !d.insecure,
SameSite: http.SameSiteLaxMode,
MaxAge: int(d.sessionTTL.Seconds()),
})
}
+33 -4
View File
@@ -306,13 +306,42 @@ func TestLogoutClearsSession(t *testing.T) {
require.Equal(t, http.StatusFound, rec.Code)
require.Equal(t, "/welcome", rec.Header().Get("Location"), "logout lands on the public page")
cleared := sessionCookie(t, rec.Result())
require.Less(t, cleared.MaxAge, 0, "logout expires the cookie")
require.Less(t, cleared.MaxAge, 0, "logout expires the cookie so the browser drops it")
require.Empty(t, cleared.Value, "logout blanks the cookie value")
// The server-side session is gone: the original cookie no longer resolves.
// Sessions are stateless (ADR-029): logout clears the cookie client-side, so a
// request carrying the cleared (empty) cookie is unauthenticated. The original
// signed cookie remains technically valid until its expiry — the accepted
// trade for no server-side store; the browser no longer holds it.
check := httptest.NewRequest(http.MethodGet, "/", nil)
check.AddCookie(cookie)
check.AddCookie(cleared)
_, ok := auth.CurrentUser(check)
require.False(t, ok)
require.False(t, ok, "the cleared cookie does not authenticate")
}
// TestSessionSurvivesRestart is the core of ADR-029: a cookie issued by one
// process is accepted by a FRESH instance with the same session secret — so a
// deploy/pod-restart no longer logs users out (the old in-memory store did).
func TestSessionSurvivesRestart(t *testing.T) {
f := newFakeIssuer(t)
auth1 := newAuth(t, f)
cookie := authenticate(t, auth1, f)
auth2 := newAuth(t, f) // simulate a redeploy: new process, same SessionSecret
req := httptest.NewRequest(http.MethodGet, "/", nil)
req.AddCookie(cookie)
user, ok := auth2.CurrentUser(req)
require.True(t, ok, "a session must survive a restart (stateless signed cookie)")
require.Equal(t, testSubject, user.Subject)
}
// TestSessionCookieIsPersistent: the cookie carries a positive Max-Age so it
// survives the browser/app being closed (a session cookie was dropped on iOS).
func TestSessionCookieIsPersistent(t *testing.T) {
f := newFakeIssuer(t)
auth := newAuth(t, f)
cookie := authenticate(t, auth, f)
require.Greater(t, cookie.MaxAge, 0, "session cookie must be persistent (Max-Age set)")
}
func TestExpiredSessionRejected(t *testing.T) {
+31 -42
View File
@@ -6,6 +6,7 @@ import (
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"fmt"
"strings"
"sync"
@@ -53,56 +54,44 @@ func (d *DexAuth) unsign(signed string) (string, bool) {
return value, true
}
// sessionData is the server-side session record.
type sessionData struct {
user web.User
expiry time.Time
// sessionClaims is the self-contained session payload carried INSIDE the signed
// cookie — there is no server-side session table. This is deliberate (ADR-029):
// an in-memory store was wiped on every pod restart, logging every user out on
// each deploy, and a stateless cookie also survives browser-close and works
// across replicas. It holds only the identity (subject + email, not secret) and
// an absolute expiry; the HMAC tag (sign/unsign) makes it tamper-proof.
type sessionClaims struct {
Sub string `json:"s"`
Email string `json:"e"`
Exp int64 `json:"x"` // unix seconds; absolute expiry
}
// sessionStore is an in-memory session table. Single replica at Stage 0, so an
// in-process map is sufficient; it is safe for concurrent use.
type sessionStore struct {
mu sync.Mutex
m map[string]sessionData
// encodeSession produces the signed cookie value for a user with the given expiry.
func (d *DexAuth) encodeSession(u web.User, exp time.Time) string {
b, _ := json.Marshal(sessionClaims{Sub: u.Subject, Email: u.Email, Exp: exp.Unix()})
return d.sign(base64.RawURLEncoding.EncodeToString(b))
}
func newSessionStore() *sessionStore { return &sessionStore{m: make(map[string]sessionData)} }
func (s *sessionStore) put(id string, d sessionData) {
s.mu.Lock()
defer s.mu.Unlock()
s.m[id] = d
}
// get returns the session if present and unexpired; expired entries are evicted.
func (s *sessionStore) get(id string, now time.Time) (sessionData, bool) {
s.mu.Lock()
defer s.mu.Unlock()
d, ok := s.m[id]
// decodeSession verifies the cookie's HMAC, parses the claims, and checks expiry.
// It returns the user and the absolute expiry on success.
func (d *DexAuth) decodeSession(cookieValue string, now time.Time) (web.User, time.Time, bool) {
payload, ok := d.unsign(cookieValue)
if !ok {
return sessionData{}, false
return web.User{}, time.Time{}, false
}
if !now.Before(d.expiry) {
delete(s.m, id)
return sessionData{}, false
raw, err := base64.RawURLEncoding.DecodeString(payload)
if err != nil {
return web.User{}, time.Time{}, false
}
return d, true
}
// refresh slides an existing session's expiry forward; a no-op for unknown ids.
func (s *sessionStore) refresh(id string, expiry time.Time) {
s.mu.Lock()
defer s.mu.Unlock()
if d, ok := s.m[id]; ok {
d.expiry = expiry
s.m[id] = d
var c sessionClaims
if err := json.Unmarshal(raw, &c); err != nil {
return web.User{}, time.Time{}, false
}
}
func (s *sessionStore) delete(id string) {
s.mu.Lock()
defer s.mu.Unlock()
delete(s.m, id)
exp := time.Unix(c.Exp, 0)
if !now.Before(exp) {
return web.User{}, time.Time{}, false // expired
}
return web.User{Subject: c.Sub, Email: c.Email}, exp, true
}
// pendingData holds the nonce bound to an in-flight authorization request.
+12 -1
View File
@@ -25,6 +25,16 @@ import (
// fails if a scenario is unmapped, a mapped test is missing, or an entry no
// longer matches a real non-pending scenario.
var scenarioCoverage = map[string]string{
// observability.feature (ADR-030, #15)
"Summarization latency is recorded per endpoint": "TestSummarizerRecordsMetric",
"A failing summarizer endpoint records its failure outcome": "TestObserveSummarizeRecordsFailureOutcomes",
"Caption fetch latency is recorded by outcome": "TestObserveCaptionFetchByOutcome",
"LLM token usage is recorded from the completion": "TestClient_UsageHookRecordsTokens",
"Q&A answer latency is recorded": "TestChatAnswerLatencyRecorded",
"HTTP requests are counted by route, method, and status": "TestHTTPMiddlewareRecordsByRoutePattern",
"A successful login is counted": "TestLoginCounted",
"The metrics endpoint is not on the public app port": "TestMetricsNotOnPublicMux",
// ai_routing.feature
"Local AI produces the summary": "TestSummarize_LocalSucceeds",
"Local AI fails and the user has a BYO provider configured": "TestSummarize_FallsBackToBYO",
@@ -40,7 +50,8 @@ var scenarioCoverage = map[string]string{
// connect_account.feature
"Connect a YouTube account": "TestCallbackExchangesAndRecordsConnection",
"Connecting an account discovers videos immediately": "TestCallbackTriggersDiscovery",
"Connecting summarizes my newest videos right away": "TestNewestUnsummarizedVideoIDs",
"Connecting summarizes my best recent videos right away": "TestOnboardBurstVideoIDs",
"The onboarding burst summarizes with a stronger model": "TestBurstChainModelsLeadsWithOnboardModel",
"Tokens are never stored in the clear": "TestCallbackExchangesAndRecordsConnection",
// paste_url.feature