aboutsummaryrefslogtreecommitdiff
path: root/cmds
diff options
context:
space:
mode:
Diffstat (limited to 'cmds')
-rw-r--r--cmds/export.go402
-rw-r--r--cmds/export_test.go441
-rw-r--r--cmds/retrostar.go127
-rw-r--r--cmds/retrostar_test.go64
-rw-r--r--cmds/rss.go82
-rw-r--r--cmds/rss_test.go73
-rw-r--r--cmds/score.go103
-rw-r--r--cmds/serve.go452
-rw-r--r--cmds/serve_test.go407
-rw-r--r--cmds/train.go430
-rw-r--r--cmds/train_test.go260
11 files changed, 2841 insertions, 0 deletions
diff --git a/cmds/export.go b/cmds/export.go
new file mode 100644
index 0000000..102c94a
--- /dev/null
+++ b/cmds/export.go
@@ -0,0 +1,402 @@
+// Export command: pull stars and sampled negatives from Miniflux into
+// labels.jsonl, the canonical label store.
+//
+// Positives: starred entries, incremental via a persisted changed_after
+// cursor, deduped by ID then normalized URL. Negatives: seeded random
+// sample of recent read+unstarred entries, topped up to a fixed ratio of
+// the positive pool. Also carries --dump-raw fixture capture.
+package cmds
+
+import (
+ "encoding/json"
+ "flag"
+ "fmt"
+ "io"
+ "math/rand"
+ "os"
+ "path/filepath"
+ "time"
+
+ "fluxrec/core"
+ "fluxrec/miniflux"
+)
+
+type ExportCommand struct {
+ DumpRaw string
+ Labels string
+ State string
+ RunOut string
+ RecFeedID int64
+ NegDays int
+ NegRatio int
+ NegSeed int64
+}
+
+func (c *ExportCommand) Name() string { return "export" }
+
+func (c *ExportCommand) Init(args []string) error {
+ fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+ fs.Usage = func() {
+ fmt.Fprint(fs.Output(), `Usage: fluxrec export [--labels FILE] [--state FILE] [--rec-feed-id ID]
+ fluxrec export --dump-raw DIR
+
+Export labels from Miniflux into labels.jsonl: starred entries become
+positives (incremental via --state cursor), plus a seeded random sample of
+recent read+unstarred entries as negatives at --neg-ratio per positive.
+Each run also writes a JSON summary to --run-out.
+
+With --dump-raw, instead fetch ~25 starred and ~100 read unstarred entries
+and write the raw API responses to DIR for offline development fixtures.
+`)
+ fs.PrintDefaults()
+ }
+ fs.StringVar(&c.DumpRaw, "dump-raw", "", "Write raw API fixtures to DIR and exit")
+ fs.StringVar(&c.Labels, "labels", "labels.jsonl", "Path to labels.jsonl")
+ fs.StringVar(&c.State, "state", "export_state.json", "Path to the export cursor file")
+ fs.StringVar(&c.RunOut, "run-out", "export_run.json", "Path to write the per-run summary JSON")
+ fs.Int64Var(&c.RecFeedID, "rec-feed-id", 0, "Miniflux feed ID of the recommendations feed (0 = not deployed yet)")
+ fs.IntVar(&c.NegDays, "neg-days", 45, "Only sample negatives published within this many days")
+ fs.IntVar(&c.NegRatio, "neg-ratio", 20, "Target negatives per positive")
+ fs.Int64Var(&c.NegSeed, "neg-seed", 20240301, "Seed for negative sampling (same data + same seed = same sample)")
+ return fs.Parse(args)
+}
+
+func (c *ExportCommand) Run(stdin io.Reader, stdout io.Writer) error {
+ if c.DumpRaw != "" {
+ return c.runDumpRaw(stdout)
+ }
+ return c.runExport(stdout)
+}
+
+// exportState is the persisted incremental cursor: the largest changed_at
+// (unix seconds) seen so far. Refetching starts one second before it so a
+// star landing in the same second as the previous run's last entry can't
+// slip through; the overlap is absorbed by dedup.
+type exportState struct {
+ ChangedAfter int64 `json:"changed_after"`
+}
+
+func loadExportState(path string) (exportState, error) {
+ var st exportState
+ body, err := os.ReadFile(path)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return st, nil
+ }
+ return st, fmt.Errorf("reading state %s: %w", path, err)
+ }
+ if err := json.Unmarshal(body, &st); err != nil {
+ return st, fmt.Errorf("parsing state %s: %w", path, err)
+ }
+ return st, nil
+}
+
+func saveExportState(path string, st exportState) error {
+ body, err := json.Marshal(st)
+ if err != nil {
+ return err
+ }
+ if err := os.WriteFile(path, body, 0o644); err != nil {
+ return fmt.Errorf("writing state %s: %w", path, err)
+ }
+ return nil
+}
+
+// exportCounts is what the run reports to stdout.
+type exportCounts struct {
+ seen, appended, idDup, urlDup, recHit, recUnresolved int
+}
+
+func (c *ExportCommand) runExport(stdout io.Writer) error {
+ started := time.Now()
+ cfg, err := miniflux.ConfigFromEnv()
+ if err != nil {
+ return err
+ }
+ client := miniflux.NewClient(cfg)
+
+ state, err := loadExportState(c.State)
+ if err != nil {
+ return err
+ }
+
+ existing, err := core.ReadLabels(c.Labels)
+ if err != nil {
+ return err
+ }
+ seenIDs := make(map[int64]bool, len(existing))
+ seenURLs := make(map[string]bool, len(existing))
+ for _, row := range existing {
+ seenIDs[row.EntryID] = true
+ seenURLs[core.NormalizeURL(row.URL)] = true
+ }
+
+ starred := true
+ f := miniflux.Filters{Starred: &starred, Limit: 200}
+ if state.ChangedAfter > 0 {
+ f.ChangedAfter = state.ChangedAfter - 1 // boundary overlap, see exportState doc
+ }
+
+ var counts exportCounts
+ var newRows []core.LabelRow
+ maxChanged := state.ChangedAfter
+ now := time.Now().UTC()
+
+ for {
+ set, err := client.Entries(f)
+ if err != nil {
+ return err
+ }
+ if len(set.Entries) == 0 {
+ break
+ }
+ for _, e := range set.Entries {
+ counts.seen++
+ if ts := e.ChangedAt.Unix(); ts > maxChanged {
+ maxChanged = ts
+ }
+
+ // Stars on the recommendations feed are not original entries;
+ // resolve by URL against existing labels. If the original was
+ // never exported, there is nothing to attach the star to.
+ if c.RecFeedID != 0 && e.Feed.ID == c.RecFeedID {
+ if seenURLs[core.NormalizeURL(e.URL)] {
+ counts.recHit++
+ } else {
+ counts.recUnresolved++
+ }
+ continue
+ }
+
+ if seenIDs[e.ID] {
+ counts.idDup++
+ continue
+ }
+ nu := core.NormalizeURL(e.URL)
+ if seenURLs[nu] {
+ counts.urlDup++
+ continue
+ }
+ newRows = append(newRows, core.LabelRow{
+ EntryID: e.ID, URL: e.URL, Title: e.Title,
+ FeedID: e.Feed.ID, FeedTitle: e.Feed.Title,
+ Label: core.LabelPos, CapturedAt: now,
+ })
+ seenIDs[e.ID] = true
+ seenURLs[nu] = true
+ counts.appended++
+ }
+ f.Offset += len(set.Entries)
+ if f.Offset >= set.Total {
+ break
+ }
+ }
+
+ if err := core.AppendLabels(c.Labels, newRows); err != nil {
+ return err
+ }
+ if maxChanged > state.ChangedAfter {
+ if err := saveExportState(c.State, exportState{ChangedAfter: maxChanged}); err != nil {
+ return err
+ }
+ }
+
+ posTotal, negTotal := 0, 0
+ for _, row := range existing {
+ if row.Label == core.LabelPos {
+ posTotal++
+ } else {
+ negTotal++
+ }
+ }
+ posTotal += counts.appended
+
+ fmt.Fprintf(stdout, "export: seen=%d appended=%d (id-dup=%d url-dup=%d rec-hit=%d rec-unresolved=%d) pos-total=%d\n",
+ counts.seen, counts.appended, counts.idDup, counts.urlDup,
+ counts.recHit, counts.recUnresolved, posTotal)
+ if counts.recUnresolved > 0 {
+ fmt.Fprintln(stdout, "note: rec-feed stars without a matching labeled original were skipped")
+ }
+
+ nc, err := c.exportNegatives(client, seenIDs, seenURLs, posTotal, negTotal, now, stdout)
+ if err != nil {
+ return err
+ }
+ return writeRunSummary(c.RunOut, runSummary{
+ RanAt: now, DurationMs: time.Since(started).Milliseconds(),
+ PositivesSeen: counts.seen, PositivesNew: counts.appended,
+ PosTotal: posTotal, NegCandidates: nc.candidates, NegNew: nc.appended,
+ NegTotal: negTotal + nc.appended, IDDup: counts.idDup, URLDup: counts.urlDup,
+ RecHit: counts.recHit, RecUnresolved: counts.recUnresolved,
+ })
+}
+
+// negCounts is what the negatives sampler reports back.
+type negCounts struct {
+ candidates, appended int
+}
+
+// exportNegatives samples read, unstarred entries published within the
+// --neg-days window down to labels.jsonl as LabelNeg rows, topping the
+// negative pool up to --neg-ratio × positives. Anything already labeled
+// (either polarity) is excluded; sampling is seeded so the same candidate
+// pool + seed reproduce the same sample.
+//
+// The window is enforced server-side via published_after (Miniflux >= 2.0.49,
+// the same floor the rest of export requires): only in-window entries are
+// fetched, so the deep archive is never walked and no client-side order
+// assumption is needed (Miniflux's default order is published_at asc).
+func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int64]bool, seenURLs map[string]bool, posTotal, negTotal int, now time.Time, stdout io.Writer) (negCounts, error) {
+ var nc negCounts
+ need := c.NegRatio*posTotal - negTotal
+ if need <= 0 || posTotal == 0 {
+ fmt.Fprintf(stdout, "negatives: skipped (need=%d pos=%d)\n", need, posTotal)
+ return nc, nil
+ }
+
+ cutoff := now.Add(-time.Duration(c.NegDays) * 24 * time.Hour)
+ unstarred := false
+ f := miniflux.Filters{
+ Starred: &unstarred, Status: "read", Limit: 200,
+ PublishedAfter: cutoff.Unix(),
+ }
+
+ candIDs := make(map[int64]bool)
+ var cands []miniflux.Entry
+ for {
+ set, err := client.Entries(f)
+ if err != nil {
+ return nc, err
+ }
+ if len(set.Entries) == 0 {
+ break
+ }
+ for _, e := range set.Entries {
+ if seenIDs[e.ID] || candIDs[e.ID] || seenURLs[core.NormalizeURL(e.URL)] {
+ continue
+ }
+ candIDs[e.ID] = true
+ cands = append(cands, e)
+ }
+ f.Offset += len(set.Entries)
+ if f.Offset >= set.Total {
+ break
+ }
+ }
+ nc.candidates = len(cands)
+
+ rng := rand.New(rand.NewSource(c.NegSeed))
+ perm := rng.Perm(len(cands))
+ take := min(need, len(cands))
+ rows := make([]core.LabelRow, 0, take)
+ for i := range take {
+ e := cands[perm[i]]
+ rows = append(rows, core.LabelRow{
+ EntryID: e.ID, URL: e.URL, Title: e.Title,
+ FeedID: e.Feed.ID, FeedTitle: e.Feed.Title,
+ Label: core.LabelNeg, CapturedAt: now,
+ })
+ seenIDs[e.ID] = true
+ seenURLs[core.NormalizeURL(e.URL)] = true
+ }
+ if err := core.AppendLabels(c.Labels, rows); err != nil {
+ return nc, err
+ }
+ nc.appended = len(rows)
+
+ fmt.Fprintf(stdout, "negatives: candidates=%d appended=%d (target=%d:%d)\n",
+ nc.candidates, nc.appended, c.NegRatio, posTotal)
+ if nc.appended < need {
+ fmt.Fprintf(stdout, "note: only %d usable candidates in the last %d days; negatives short by %d\n",
+ nc.candidates, c.NegDays, need-nc.appended)
+ }
+ return nc, nil
+}
+
+// runSummary is the per-run JSON summary written for visibility; kept to
+// the latest run (overwrite, not append).
+type runSummary struct {
+ RanAt time.Time `json:"ran_at"`
+ DurationMs int64 `json:"duration_ms"`
+ PositivesSeen int `json:"positives_seen"`
+ PositivesNew int `json:"positives_new"`
+ PosTotal int `json:"pos_total"`
+ NegCandidates int `json:"neg_candidates"`
+ NegNew int `json:"neg_new"`
+ NegTotal int `json:"neg_total"`
+ IDDup int `json:"id_dup"`
+ URLDup int `json:"url_dup"`
+ RecHit int `json:"rec_hit"`
+ RecUnresolved int `json:"rec_unresolved"`
+}
+
+func writeRunSummary(path string, s runSummary) error {
+ body, err := json.MarshalIndent(s, "", " ")
+ if err != nil {
+ return err
+ }
+ if err := os.WriteFile(path, body, 0o644); err != nil {
+ return fmt.Errorf("writing run summary %s: %w", path, err)
+ }
+ return nil
+}
+
+// ── fixture capture (--dump-raw) ──
+
+// dumpJob is one fixture to capture: output filename + API filters.
+type dumpJob struct {
+ filename string
+ filters miniflux.Filters
+}
+
+func dumpJobs() []dumpJob {
+ starred, unstarred := true, false
+ return []dumpJob{
+ {"starred_sample.json", miniflux.Filters{Starred: &starred, Limit: 25}},
+ // Negatives fixture must NOT overlap stars; a plain status=read
+ // sample can be all-starred after a retroactive starring pass.
+ {"read_sample.json", miniflux.Filters{Starred: &unstarred, Status: "read", Limit: 100}},
+ }
+}
+
+func (c *ExportCommand) runDumpRaw(stdout io.Writer) error {
+ cfg, err := miniflux.ConfigFromEnv()
+ if err != nil {
+ return err
+ }
+ client := miniflux.NewClient(cfg)
+
+ if err := os.MkdirAll(c.DumpRaw, 0o755); err != nil {
+ return fmt.Errorf("creating %s: %w", c.DumpRaw, err)
+ }
+
+ for _, job := range dumpJobs() {
+ body, err := client.RawEntries(job.filters)
+ if err != nil {
+ return fmt.Errorf("fetching %s: %w", job.filename, err)
+ }
+ // Decode only to report a count and catch a malformed response
+ // before it becomes a silent bad fixture.
+ var set miniflux.EntrySet
+ if err := json.Unmarshal(body, &set); err != nil {
+ return fmt.Errorf("%s: response did not decode as entries: %w", job.filename, err)
+ }
+ path := filepath.Join(c.DumpRaw, job.filename)
+ if err := os.WriteFile(path, body, 0o644); err != nil {
+ return fmt.Errorf("writing %s: %w", path, err)
+ }
+ // Contamination guard: the negatives fixture must be star-free,
+ // or dev data for negative sampling teaches the wrong lesson.
+ starredN := 0
+ for _, e := range set.Entries {
+ if e.Starred {
+ starredN++
+ }
+ }
+ if job.filters.Starred != nil && !*job.filters.Starred && starredN > 0 {
+ fmt.Fprintf(stdout, "WARNING: %s contains %d starred entries (API ignored starred=false?)\n", job.filename, starredN)
+ }
+ fmt.Fprintf(stdout, "wrote %s (%d entries, %d bytes)\n", path, len(set.Entries), len(body))
+ }
+ return nil
+}
diff --git a/cmds/export_test.go b/cmds/export_test.go
new file mode 100644
index 0000000..0190d39
--- /dev/null
+++ b/cmds/export_test.go
@@ -0,0 +1,441 @@
+// Tests for `fluxrec export` against an httptest fake Miniflux. Env vars
+// point the command's config resolution at the fake — no real credentials.
+package cmds
+
+import (
+ "bytes"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "fluxrec/core"
+ "fluxrec/miniflux"
+)
+
+// testEntry mirrors the entry fields export consumes.
+type testFeed struct {
+ ID int64 `json:"id"`
+ Title string `json:"title"`
+}
+
+type testEntry struct {
+ ID int64 `json:"id"`
+ URL string `json:"url"`
+ Title string `json:"title"`
+ Content string `json:"content"`
+ Status string `json:"status"`
+ Starred bool `json:"starred"`
+ PublishedAt time.Time `json:"published_at"`
+ ChangedAt time.Time `json:"changed_at"`
+ Feed testFeed `json:"feed"`
+}
+
+func makeEntries(n int, startID int64, starred bool, feedID int64) []testEntry {
+ base := time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC)
+ entries := make([]testEntry, n)
+ for i := range entries {
+ entries[i] = testEntry{
+ ID: startID + int64(i),
+ URL: fmt.Sprintf("https://example.com/post/%d", startID+int64(i)),
+ Title: fmt.Sprintf("Title %d", startID+int64(i)),
+ Status: "read",
+ Starred: starred,
+ PublishedAt: base,
+ ChangedAt: base.Add(time.Duration(i) * time.Hour),
+ Feed: testFeed{ID: feedID, Title: "Fixture Feed"},
+ }
+ }
+ return entries
+}
+
+func entriesJSON(total int, entries []testEntry) string {
+ b, _ := json.Marshal(entries)
+ return fmt.Sprintf(`{"total":%d,"entries":%s}`, total, b)
+}
+
+// pagedServer serves a starred pool and a read pool (chosen per request by
+// the starred filter) with limit/offset paging, and records every query.
+type pagedServer struct {
+ starred []testEntry
+ read []testEntry
+ queries []string
+}
+
+func (p *pagedServer) handler() http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ p.queries = append(p.queries, r.URL.RawQuery)
+ q := r.URL.Query()
+ pool := p.read
+ if q.Get("starred") == "true" {
+ pool = p.starred
+ }
+ // Honor published_after server-side, as real Miniflux does, so the
+ // negatives sampler's window is exercised by these tests.
+ if pa, _ := strconv.ParseInt(q.Get("published_after"), 10, 64); pa > 0 {
+ cutoff := time.Unix(pa, 0)
+ filtered := pool[:0:0]
+ for _, e := range pool {
+ if !e.PublishedAt.Before(cutoff) {
+ filtered = append(filtered, e)
+ }
+ }
+ pool = filtered
+ }
+ limit, _ := strconv.Atoi(q.Get("limit"))
+ offset, _ := strconv.Atoi(q.Get("offset"))
+ if limit <= 0 {
+ limit = len(pool)
+ }
+ if offset > len(pool) {
+ offset = len(pool)
+ }
+ end := offset + limit
+ if end > len(pool) {
+ end = len(pool)
+ }
+ w.Header().Set("Content-Type", "application/json")
+ fmt.Fprint(w, entriesJSON(len(pool), pool[offset:end]))
+ }
+}
+
+// runExportAgainst wires a fake Miniflux via env, then runs the export
+// command with the given extra args, returning stdout.
+func runExportAgainst(t *testing.T, p *pagedServer, dir string, extra ...string) string {
+ t.Helper()
+ srv := httptest.NewServer(p.handler())
+ defer srv.Close()
+ t.Setenv("MINIFLUX_URL", srv.URL)
+ t.Setenv("MINIFLUX_TOKEN", "test-token")
+
+ args := append([]string{
+ "--labels", filepath.Join(dir, "labels.jsonl"),
+ "--state", filepath.Join(dir, "export_state.json"),
+ "--run-out", filepath.Join(dir, "export_run.json"),
+ }, extra...)
+
+ cmd := &ExportCommand{}
+ if err := cmd.Init(args); err != nil {
+ t.Fatalf("Init: %v", err)
+ }
+ var out bytes.Buffer
+ if err := cmd.Run(nil, &out); err != nil {
+ t.Fatalf("Run: %v", err)
+ }
+ return out.String()
+}
+
+func TestExportPositivesIdempotent(t *testing.T) {
+ p := &pagedServer{starred: makeEntries(25, 1000, true, 7)}
+ dir := t.TempDir()
+
+ out1 := runExportAgainst(t, p, dir)
+ if !strings.Contains(out1, "appended=25") {
+ t.Errorf("first run should append 25, got:\n%s", out1)
+ }
+ labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl"))
+ if err != nil {
+ t.Fatalf("read labels: %v", err)
+ }
+ if len(labels) != 25 {
+ t.Fatalf("expected 25 rows, got %d", len(labels))
+ }
+ for _, row := range labels {
+ if row.Label != core.LabelPos {
+ t.Errorf("row label = %q, want pos", row.Label)
+ }
+ if row.EntryID < 1000 || row.EntryID >= 1025 {
+ t.Errorf("unexpected entry id %d", row.EntryID)
+ }
+ }
+
+ // State cursor must have advanced to the newest changed_at.
+ stateBody, err := os.ReadFile(filepath.Join(dir, "export_state.json"))
+ if err != nil {
+ t.Fatalf("state not written: %v", err)
+ }
+ var st exportState
+ if err := json.Unmarshal(stateBody, &st); err != nil {
+ t.Fatalf("state unparseable: %v", err)
+ }
+ if st.ChangedAfter <= 0 {
+ t.Errorf("state cursor = %d, want a positive timestamp", st.ChangedAfter)
+ }
+
+ nQueriesAfterRun1 := len(p.queries)
+ out2 := runExportAgainst(t, p, dir)
+ if !strings.Contains(out2, "appended=0") {
+ t.Errorf("second run must append nothing, got:\n%s", out2)
+ }
+ labels2, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl"))
+ if err != nil {
+ t.Fatalf("read labels after run 2: %v", err)
+ }
+ if len(labels2) != 25 {
+ t.Fatalf("second run changed the file: %d rows", len(labels2))
+ }
+ // Second run must have gone incremental: changed_after in some query
+ // issued by that run (the first run, with no cursor, sends none).
+ incremental := false
+ for _, qq := range p.queries[nQueriesAfterRun1:] {
+ decoded, _ := url.QueryUnescape(qq)
+ if strings.Contains(decoded, "changed_after=") {
+ incremental = true
+ }
+ }
+ if !incremental {
+ t.Errorf("second run sent no changed_after query: %v", p.queries[nQueriesAfterRun1:])
+ }
+}
+
+func TestExportURLDedup(t *testing.T) {
+ p := &pagedServer{starred: makeEntries(3, 2000, true, 7)}
+ dir := t.TempDir()
+
+ // Pre-seed a label whose URL is a normalized twin of entry 2000's URL.
+ pre := core.LabelRow{
+ EntryID: 99999, URL: "HTTPS://www.example.com/post/2000/", Title: "Old",
+ FeedID: 1, FeedTitle: "F", Label: core.LabelPos,
+ CapturedAt: time.Now(),
+ }
+ if err := core.AppendLabels(filepath.Join(dir, "labels.jsonl"), []core.LabelRow{pre}); err != nil {
+ t.Fatal(err)
+ }
+
+ out := runExportAgainst(t, p, dir)
+ if !strings.Contains(out, "appended=2") || !strings.Contains(out, "url-dup=1") {
+ t.Errorf("expected appended=2 url-dup=1, got:\n%s", out)
+ }
+}
+
+func TestExportRecFeedResolution(t *testing.T) {
+ // Two starred entries ON the recommendations feed (feed id 99).
+ rec := makeEntries(2, 3000, true, 99)
+ rec[0].URL = "https://example.com/post/known"
+ rec[1].URL = "https://example.com/post/unknown"
+ p := &pagedServer{starred: rec}
+ dir := t.TempDir()
+
+ // The "known" URL is already labeled (as the original entry's star).
+ pre := core.LabelRow{
+ EntryID: 1234, URL: "https://example.com/post/known", Title: "Known",
+ FeedID: 7, FeedTitle: "Orig", Label: core.LabelPos,
+ CapturedAt: time.Now(),
+ }
+ if err := core.AppendLabels(filepath.Join(dir, "labels.jsonl"), []core.LabelRow{pre}); err != nil {
+ t.Fatal(err)
+ }
+
+ out := runExportAgainst(t, p, dir, "--rec-feed-id", "99")
+ if !strings.Contains(out, "rec-hit=1") || !strings.Contains(out, "rec-unresolved=1") {
+ t.Errorf("expected rec-hit=1 rec-unresolved=1, got:\n%s", out)
+ }
+ if !strings.Contains(out, "appended=0") {
+ t.Errorf("rec-feed entries must never append rows, got:\n%s", out)
+ }
+ labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(labels) != 1 {
+ t.Errorf("labels file should be untouched, has %d rows", len(labels))
+ }
+}
+
+// negFixture builds a server with `nPos` fresh starred entries and a read
+// pool of `nFresh` entries inside the typical 45-day window plus `nOld`
+// entries far outside it (sorted after the fresh ones, as published-desc
+// would order them).
+func negFixture(nPos, nFresh, nOld int) *pagedServer {
+ now := time.Now()
+ starred := makeEntries(nPos, 1000, true, 7)
+ for i := range starred {
+ starred[i].PublishedAt = now.Add(-24 * time.Hour)
+ }
+ fresh := makeEntries(nFresh, 5000, false, 8)
+ for i := range fresh {
+ fresh[i].PublishedAt = now.Add(-10 * 24 * time.Hour)
+ }
+ old := makeEntries(nOld, 6000, false, 8)
+ for i := range old {
+ old[i].PublishedAt = now.Add(-100 * 24 * time.Hour)
+ }
+ return &pagedServer{starred: starred, read: append(fresh, old...)}
+}
+
+func negEntryIDs(t *testing.T, dir string) map[int64]bool {
+ t.Helper()
+ labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ ids := map[int64]bool{}
+ for _, row := range labels {
+ if row.Label == core.LabelNeg {
+ ids[row.EntryID] = true
+ }
+ }
+ return ids
+}
+
+func TestExportNegatives(t *testing.T) {
+ // 5 positives × ratio 20 = 100 negatives from 150 in-window candidates.
+ p := negFixture(5, 150, 10)
+ dir := t.TempDir()
+
+ out := runExportAgainst(t, p, dir)
+ if !strings.Contains(out, "candidates=150") || !strings.Contains(out, "appended=100") {
+ t.Errorf("expected candidates=150 appended=100, got:\n%s", out)
+ }
+
+ labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ posIDs := map[int64]bool{}
+ posURLs := map[string]bool{}
+ negCount := 0
+ for _, row := range labels {
+ if row.Label == core.LabelPos {
+ posIDs[row.EntryID] = true
+ posURLs[core.NormalizeURL(row.URL)] = true
+ }
+ }
+ for _, row := range labels {
+ if row.Label != core.LabelNeg {
+ continue
+ }
+ negCount++
+ if posIDs[row.EntryID] {
+ t.Errorf("negative %d collides with a positive ID", row.EntryID)
+ }
+ if posURLs[core.NormalizeURL(row.URL)] {
+ t.Errorf("negative %d collides with a positive URL", row.EntryID)
+ }
+ if row.EntryID >= 6000 {
+ t.Errorf("old (out-of-window) entry %d sampled as negative", row.EntryID)
+ }
+ }
+ if negCount != 100 {
+ t.Errorf("expected 100 negatives, got %d", negCount)
+ }
+
+ // Run summary (latest run, overwritten each time).
+ body, err := os.ReadFile(filepath.Join(dir, "export_run.json"))
+ if err != nil {
+ t.Fatalf("run summary not written: %v", err)
+ }
+ var rs runSummary
+ if err := json.Unmarshal(body, &rs); err != nil {
+ t.Fatalf("run summary unparseable: %v", err)
+ }
+ if rs.PositivesNew != 5 || rs.NegNew != 100 || rs.PosTotal != 5 || rs.NegTotal != 100 {
+ t.Errorf("run summary counts wrong: %+v", rs)
+ }
+
+ // Second run: pool already at target, so no new negatives.
+ out2 := runExportAgainst(t, p, dir)
+ if !strings.Contains(out2, "appended=0") {
+ t.Errorf("second run should append nothing, got:\n%s", out2)
+ }
+}
+
+func TestExportNegativesSeededDeterministic(t *testing.T) {
+ dir1, dir2 := t.TempDir(), t.TempDir()
+ runExportAgainst(t, negFixture(4, 120, 0), dir1)
+ runExportAgainst(t, negFixture(4, 120, 0), dir2)
+
+ ids1, ids2 := negEntryIDs(t, dir1), negEntryIDs(t, dir2)
+ if len(ids1) != len(ids2) || len(ids1) == 0 {
+ t.Fatalf("negative counts differ or zero: %d vs %d", len(ids1), len(ids2))
+ }
+ for id := range ids1 {
+ if !ids2[id] {
+ t.Fatalf("same seed + same pool picked different samples; %d missing in run 2", id)
+ }
+ }
+}
+
+// ── dump-raw tests ──
+
+// fakeMiniflux serves small responses for fixture capture and records the
+// query of each request.
+type fakeMiniflux struct {
+ queries []string
+}
+
+func (f *fakeMiniflux) handler() http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ f.queries = append(f.queries, r.URL.RawQuery)
+ w.Header().Set("Content-Type", "application/json")
+ if r.URL.Query().Get("starred") == "true" {
+ fmt.Fprint(w, entriesJSON(2, makeEntries(2, 1000, true, 1)))
+ } else {
+ fmt.Fprint(w, entriesJSON(3, makeEntries(3, 2000, false, 1)))
+ }
+ }
+}
+
+func TestExportDumpRawWritesFixtures(t *testing.T) {
+ fake := &fakeMiniflux{}
+ srv := httptest.NewServer(fake.handler())
+ defer srv.Close()
+ t.Setenv("MINIFLUX_URL", srv.URL)
+ t.Setenv("MINIFLUX_TOKEN", "test-token")
+
+ dir := t.TempDir()
+ cmd := &ExportCommand{}
+ if err := cmd.Init([]string{"--dump-raw", dir}); err != nil {
+ t.Fatalf("Init: %v", err)
+ }
+ var out bytes.Buffer
+ if err := cmd.Run(nil, &out); err != nil {
+ t.Fatalf("Run: %v", err)
+ }
+
+ for _, want := range []struct {
+ file string
+ nEntries int
+ }{
+ {"starred_sample.json", 2},
+ {"read_sample.json", 3},
+ } {
+ path := filepath.Join(dir, want.file)
+ body, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatalf("fixture %s not written: %v", want.file, err)
+ }
+ var set miniflux.EntrySet
+ if err := json.Unmarshal(body, &set); err != nil {
+ t.Fatalf("fixture %s is not valid entries JSON: %v", want.file, err)
+ }
+ if len(set.Entries) != want.nEntries {
+ t.Errorf("%s: got %d entries, want %d", want.file, len(set.Entries), want.nEntries)
+ }
+ }
+
+ if len(fake.queries) != 2 {
+ t.Fatalf("expected 2 API calls, got %d", len(fake.queries))
+ }
+ if !strings.Contains(fake.queries[0], "starred=true") {
+ t.Errorf("first call should filter starred=true, got %q", fake.queries[0])
+ }
+ if !strings.Contains(fake.queries[1], "status=read") {
+ t.Errorf("second call should filter status=read, got %q", fake.queries[1])
+ }
+ if !strings.Contains(fake.queries[1], "starred=false") {
+ t.Errorf("negatives call must filter starred=false, got %q", fake.queries[1])
+ }
+
+ if !strings.Contains(out.String(), "starred_sample.json") ||
+ !strings.Contains(out.String(), "read_sample.json") {
+ t.Errorf("stdout should name both fixtures, got:\n%s", out.String())
+ }
+}
diff --git a/cmds/retrostar.go b/cmds/retrostar.go
new file mode 100644
index 0000000..37f9313
--- /dev/null
+++ b/cmds/retrostar.go
@@ -0,0 +1,127 @@
+// Retrostar command: retroactive-starring aid for cold start. Scores the
+// recent read+unstarred entry pool with the current model and prints the
+// highest-ranked candidates so they can be starred by hand in the Miniflux
+// UI. Entries already in labels.jsonl are excluded: an exported negative
+// can never be re-labeled (labels never change) and an exported positive
+// is already known. Read-only — it never writes to Miniflux; starring
+// stays a human action in the UI.
+package cmds
+
+import (
+ "flag"
+ "fmt"
+ "io"
+ "sort"
+ "time"
+
+ "fluxrec/core"
+ "fluxrec/miniflux"
+)
+
+type RetrostarCommand struct {
+ ModelPath string
+ Labels string
+ Days int
+ Limit int
+}
+
+func (c *RetrostarCommand) Name() string { return "retrostar" }
+
+func (c *RetrostarCommand) Init(args []string) error {
+ fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+ fs.Usage = func() {
+ fmt.Fprint(fs.Output(), `Usage: fluxrec retrostar [flags]
+
+Print the highest-scored recent read+unstarred entries (model-ranked) as
+candidates for retroactive starring in the Miniflux UI. Entries already in
+labels.jsonl are skipped. Read-only: nothing is written to Miniflux.
+`)
+ fs.PrintDefaults()
+ }
+ fs.StringVar(&c.ModelPath, "model", "model.json", "Path to model.json")
+ fs.StringVar(&c.Labels, "labels", "labels.jsonl", "Path to labels.jsonl (already-labeled entries are excluded)")
+ fs.IntVar(&c.Days, "days", 60, "Only consider entries published within this many days (inside the deletion horizon)")
+ fs.IntVar(&c.Limit, "limit", 50, "How many top candidates to print")
+ if err := fs.Parse(args); err != nil {
+ return err
+ }
+ if c.Days < 1 {
+ return fmt.Errorf("--days must be >= 1, got %d", c.Days)
+ }
+ if c.Limit < 1 {
+ return fmt.Errorf("--limit must be >= 1, got %d", c.Limit)
+ }
+ return nil
+}
+
+func (c *RetrostarCommand) Run(stdin io.Reader, stdout io.Writer) error {
+ cfg, err := miniflux.ConfigFromEnv()
+ if err != nil {
+ return err
+ }
+ client := miniflux.NewClient(cfg)
+
+ model, err := loadModel(c.ModelPath)
+ if err != nil {
+ return err
+ }
+ vec := core.CreateVectorizerFromModel(model)
+
+ existing, err := core.ReadLabels(c.Labels)
+ if err != nil {
+ return err
+ }
+ labeled := make(map[int64]bool, len(existing))
+ for _, row := range existing {
+ labeled[row.EntryID] = true
+ }
+
+ cutoff := time.Now().Add(-time.Duration(c.Days) * 24 * time.Hour)
+ unstarred := false
+ f := miniflux.Filters{
+ Starred: &unstarred, Status: "read", Limit: 200,
+ PublishedAfter: cutoff.Unix(),
+ }
+ var scored []scoredEntry
+ for {
+ set, err := client.Entries(f)
+ if err != nil {
+ return err
+ }
+ if len(set.Entries) == 0 {
+ break
+ }
+ for _, e := range set.Entries {
+ if labeled[e.ID] {
+ continue
+ }
+ vecs := vec.Transform([]string{core.ScoringText(e.Title, e.Feed.Title)})
+ s, err := core.PredictScore(vecs[0], model.Weights)
+ if err != nil {
+ return fmt.Errorf("scoring entry %d: %w", e.ID, err)
+ }
+ scored = append(scored, scoredEntry{entry: e, score: s})
+ }
+ f.Offset += len(set.Entries)
+ if f.Offset >= set.Total {
+ break
+ }
+ }
+
+ sort.SliceStable(scored, func(i, j int) bool {
+ if scored[i].score == scored[j].score {
+ return scored[i].entry.ID < scored[j].entry.ID
+ }
+ return scored[i].score > scored[j].score
+ })
+
+ shown := min(c.Limit, len(scored))
+ for _, se := range scored[:shown] {
+ e := se.entry
+ fmt.Fprintf(stdout, "%.3f %s %s %s\n %s (entry %d)\n",
+ se.score, e.PublishedAt.Format("2006-01-02"), e.Feed.Title, e.Title, e.URL, e.ID)
+ }
+ fmt.Fprintf(stdout, "retrostar: scanned %d unlabeled candidates, showing %d — star in the Miniflux UI, then run export\n",
+ len(scored), shown)
+ return nil
+}
diff --git a/cmds/retrostar_test.go b/cmds/retrostar_test.go
new file mode 100644
index 0000000..3ad6dca
--- /dev/null
+++ b/cmds/retrostar_test.go
@@ -0,0 +1,64 @@
+// Test for `fluxrec retrostar` against an httptest fake Miniflux: the
+// printed candidates are model-ranked and anything already in labels.jsonl
+// is excluded. Helpers (testEntry, makeEntries, pagedServer) come from
+// export_test.go.
+package cmds
+
+import (
+ "encoding/json"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "fluxrec/core"
+)
+
+func TestRetrostar(t *testing.T) {
+ // pool has 20 read+unstarred candidates (IDs 100-119)
+ fake := &pagedServer{read: makeEntries(20, 100, false, 1)}
+ srv := httptest.NewServer(fake.handler())
+ defer srv.Close()
+ t.Setenv("MINIFLUX_URL", srv.URL)
+ t.Setenv("MINIFLUX_TOKEN", "test-token")
+
+ dir := t.TempDir()
+
+ // Two candidates are already labeled and must be excluded.
+ labelsPath := filepath.Join(dir, "labels.jsonl")
+ now := time.Date(2024, 3, 1, 0, 0, 0, 0, time.UTC)
+ if err := core.AppendLabels(labelsPath, []core.LabelRow{
+ {EntryID: 105, URL: "https://example.com/post/105", Title: "Title 105",
+ FeedID: 1, FeedTitle: "Fixture Feed", Label: core.LabelPos, CapturedAt: now},
+ {EntryID: 110, URL: "https://example.com/post/110", Title: "Title 110",
+ FeedID: 1, FeedTitle: "Fixture Feed", Label: core.LabelNeg, CapturedAt: now},
+ }); err != nil {
+ t.Fatalf("AppendLabels: %v", err)
+ }
+
+ modelPath := filepath.Join(dir, "model.json")
+ body, _ := json.Marshal(testModel())
+ if err := os.WriteFile(modelPath, body, 0o644); err != nil {
+ t.Fatalf("writing model: %v", err)
+ }
+
+ cmd := &RetrostarCommand{ModelPath: modelPath, Labels: labelsPath, Days: 100000, Limit: 5}
+ var out strings.Builder
+ if err := cmd.Run(nil, &out); err != nil {
+ t.Fatalf("retrostar: %v", err)
+ }
+ got := out.String()
+ if !strings.Contains(got, "entry 100") {
+ t.Errorf("output missing top candidate 100:\n%s", got)
+ }
+ for _, excluded := range []string{"entry 105", "entry 110"} {
+ if strings.Contains(got, excluded) {
+ t.Errorf("output contains already-labeled %s:\n%s", excluded, got)
+ }
+ }
+ if !strings.Contains(got, "scanned 18 unlabeled candidates, showing 5") {
+ t.Errorf("summary line wrong:\n%s", got)
+ }
+}
diff --git a/cmds/rss.go b/cmds/rss.go
new file mode 100644
index 0000000..3c49513
--- /dev/null
+++ b/cmds/rss.go
@@ -0,0 +1,82 @@
+// RSS rendering for the recommendations feed: hand-rolled RSS 2.0 via
+// encoding/xml, no feed library. The render source is the runs.jsonl tail;
+// items appear newest batch first, kept order within a batch. Guid embeds
+// the original Miniflux entry ID (mf:<id>) for traceability.
+package cmds
+
+import (
+ "encoding/xml"
+ "fmt"
+ "time"
+
+ "fluxrec/core"
+)
+
+const rssXMLHeader = `<?xml version="1.0" encoding="UTF-8"?>` + "\n"
+
+type rssGUID struct {
+ IsPermaLink bool `xml:"isPermaLink,attr"`
+ Value string `xml:",chardata"`
+}
+
+type rssItem struct {
+ Title string `xml:"title"`
+ Link string `xml:"link"`
+ GUID rssGUID `xml:"guid"`
+ PubDate string `xml:"pubDate"`
+ Description string `xml:"description"`
+}
+
+type rssChannel struct {
+ Title string `xml:"title"`
+ Description string `xml:"description"`
+ Items []rssItem `xml:"item"`
+}
+
+type rssDocument struct {
+ XMLName xml.Name `xml:"rss"`
+ Version string `xml:"version,attr"`
+ Channel rssChannel `xml:"channel"`
+}
+
+// RenderRSS renders runs (oldest to newest, as read from runs.jsonl) as an
+// RSS 2.0 document. The source feed title prefixes each item description so
+// provenance is visible in-reader; item content was already cleaned when
+// the run row was written.
+func RenderRSS(runs []core.RunRow) ([]byte, error) {
+ doc := rssDocument{
+ Version: "2.0",
+ Channel: rssChannel{
+ Title: "fluxrec recommendations",
+ Description: "Top-scored Miniflux entries, refreshed on a schedule.",
+ },
+ }
+ for i := len(runs) - 1; i >= 0; i-- {
+ for _, k := range runs[i].Kept {
+ doc.Channel.Items = append(doc.Channel.Items, rssItem{
+ Title: k.Title,
+ Link: k.URL,
+ GUID: rssGUID{IsPermaLink: false, Value: fmt.Sprintf("mf:%d", k.EntryID)},
+ PubDate: k.PublishedAt.Format(time.RFC1123Z),
+ Description: rssDescription(k),
+ })
+ }
+ }
+ body, err := xml.MarshalIndent(doc, "", " ")
+ if err != nil {
+ return nil, fmt.Errorf("marshaling RSS: %w", err)
+ }
+ return append([]byte(rssXMLHeader), body...), nil
+}
+
+// rssDescription builds the item body: source feed name on the first line,
+// then the cleaned content (if any).
+func rssDescription(k core.KeptItem) string {
+ if k.Content == "" {
+ return k.FeedTitle
+ }
+ if k.FeedTitle == "" {
+ return k.Content
+ }
+ return k.FeedTitle + "\n\n" + k.Content
+}
diff --git a/cmds/rss_test.go b/cmds/rss_test.go
new file mode 100644
index 0000000..406f853
--- /dev/null
+++ b/cmds/rss_test.go
@@ -0,0 +1,73 @@
+// Golden test for the RSS renderer: fixed runs in, exact XML out. Also
+// pins batch ordering (newest first) and XML escaping of item text.
+package cmds
+
+import (
+ "testing"
+ "time"
+
+ "fluxrec/core"
+)
+
+func TestRenderRSSGolden(t *testing.T) {
+ at := func(day, h, m int) time.Time {
+ return time.Date(2024, 3, day, h, m, 0, 0, time.UTC)
+ }
+ runs := []core.RunRow{
+ {
+ RanAt: at(1, 6, 0), EntriesSeen: 40,
+ Kept: []core.KeptItem{
+ {EntryID: 101, URL: "https://example.com/a?x=1&y=2", Title: "Alpha & Omega",
+ FeedTitle: "Feed A", PublishedAt: at(1, 5, 0),
+ Content: "x < y", Score: 0.9, Kind: core.KindRanked},
+ {EntryID: 102, URL: "https://example.com/b", Title: "Beta",
+ FeedTitle: "Feed B", PublishedAt: at(1, 5, 30),
+ Content: "second body", Score: 0.2, Kind: core.KindRandom},
+ },
+ },
+ {
+ RanAt: at(1, 12, 0), EntriesSeen: 12,
+ Kept: []core.KeptItem{
+ {EntryID: 103, URL: "https://example.com/c", Title: "Gamma",
+ FeedTitle: "Feed A", PublishedAt: at(1, 11, 0),
+ Content: "", Score: 0.8, Kind: core.KindRunnerUp},
+ },
+ },
+ }
+
+ got, err := RenderRSS(runs)
+ if err != nil {
+ t.Fatalf("RenderRSS: %v", err)
+ }
+ want := `<?xml version="1.0" encoding="UTF-8"?>
+<rss version="2.0">
+ <channel>
+ <title>fluxrec recommendations</title>
+ <description>Top-scored Miniflux entries, refreshed on a schedule.</description>
+ <item>
+ <title>Gamma</title>
+ <link>https://example.com/c</link>
+ <guid isPermaLink="false">mf:103</guid>
+ <pubDate>Fri, 01 Mar 2024 11:00:00 +0000</pubDate>
+ <description>Feed A</description>
+ </item>
+ <item>
+ <title>Alpha &amp; Omega</title>
+ <link>https://example.com/a?x=1&amp;y=2</link>
+ <guid isPermaLink="false">mf:101</guid>
+ <pubDate>Fri, 01 Mar 2024 05:00:00 +0000</pubDate>
+ <description>Feed A&#xA;&#xA;x &lt; y</description>
+ </item>
+ <item>
+ <title>Beta</title>
+ <link>https://example.com/b</link>
+ <guid isPermaLink="false">mf:102</guid>
+ <pubDate>Fri, 01 Mar 2024 05:30:00 +0000</pubDate>
+ <description>Feed B&#xA;&#xA;second body</description>
+ </item>
+ </channel>
+</rss>`
+ if string(got) != want {
+ t.Errorf("RSS mismatch.\ngot:\n%s", got)
+ }
+}
diff --git a/cmds/score.go b/cmds/score.go
new file mode 100644
index 0000000..da6167a
--- /dev/null
+++ b/cmds/score.go
@@ -0,0 +1,103 @@
+// Score command: debugging utility, one trained model vs. stdin titles.
+//
+// Reads one title per line on stdin, writes {"title", "score"} JSONL on
+// stdout. Exists to make the vendored ML core exercisable end-to-end before
+// the real commands (export/train/serve) land, and stays useful later for
+// poking at model behavior by hand.
+package cmds
+
+import (
+ "bufio"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+
+ "fluxrec/core"
+)
+
+type ScoreCommand struct {
+ ModelPath string
+}
+
+func (c *ScoreCommand) Name() string { return "score" }
+
+func (c *ScoreCommand) Init(args []string) error {
+ fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+ fs.Usage = func() {
+ fmt.Fprint(fs.Output(), `Usage: fluxrec score --model MODEL
+
+Score stdin titles (one per line) with a trained model, JSONL to stdout.
+`)
+ fs.PrintDefaults()
+ }
+ fs.StringVar(&c.ModelPath, "model", "", "Path to model.json (required)")
+ if err := fs.Parse(args); err != nil {
+ return err
+ }
+ if c.ModelPath == "" {
+ return errors.New("--model flag is required")
+ }
+ clean := filepath.Clean(c.ModelPath)
+ if strings.Contains(clean, "..") {
+ return fmt.Errorf("invalid model path %q: must not contain '..'", c.ModelPath)
+ }
+ c.ModelPath = clean
+ return nil
+}
+
+type scoreResult struct {
+ Title string `json:"title"`
+ Score float64 `json:"score"`
+}
+
+func (c *ScoreCommand) Run(stdin io.Reader, stdout io.Writer) error {
+ model, err := loadModel(c.ModelPath)
+ if err != nil {
+ return err
+ }
+ vectorizer := core.CreateVectorizerFromModel(model)
+
+ enc := json.NewEncoder(stdout)
+ scanner := bufio.NewScanner(stdin)
+ scanner.Buffer(make([]byte, 64*1024), 1024*1024)
+ for scanner.Scan() {
+ title := strings.TrimSpace(scanner.Text())
+ if title == "" {
+ continue
+ }
+ if len(title) > 1000 {
+ return fmt.Errorf("title exceeds 1000 chars")
+ }
+ vec := vectorizer.Transform([]string{title})
+ score, err := core.PredictScore(vec[0], model.Weights)
+ if err != nil {
+ return fmt.Errorf("scoring %q: %w", title, err)
+ }
+ if err := enc.Encode(scoreResult{Title: title, Score: score}); err != nil {
+ return err
+ }
+ }
+ return scanner.Err()
+}
+
+func loadModel(path string) (*core.ModelEnvelope, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return nil, fmt.Errorf("opening model: %w", err)
+ }
+ defer func() { _ = f.Close() }()
+
+ var model core.ModelEnvelope
+ if err := json.NewDecoder(f).Decode(&model); err != nil {
+ return nil, fmt.Errorf("parsing model %s: %w", path, err)
+ }
+ if len(model.Weights) == 0 || len(model.OrderedVocab) == 0 {
+ return nil, fmt.Errorf("model %s is missing weights or vocabulary", path)
+ }
+ return &model, nil
+}
diff --git a/cmds/serve.go b/cmds/serve.go
new file mode 100644
index 0000000..9d82d38
--- /dev/null
+++ b/cmds/serve.go
@@ -0,0 +1,452 @@
+// Serve command: the long-running server lane. Polls Miniflux for entries
+// newer than a persisted cursor, scores each one exactly once, assembles a
+// top-N + exploration batch per refresh into runs.jsonl, and serves the
+// recent batches as an RSS feed plus a JSON status endpoint. Strictly
+// read-only toward Miniflux (the client has no mutation methods).
+//
+// Scoring text is the same construction as training: title plus the
+// feed-identity token. Batch = top-N by raw score (no threshold cutoff)
+// with the exploration budget split at a fixed 1/3 runner-up + 2/3 random.
+package cmds
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "flag"
+ "fmt"
+ "io"
+ "math"
+ "math/rand"
+ "net/http"
+ "os"
+ "os/signal"
+ "sort"
+ "strconv"
+ "strings"
+ "sync/atomic"
+ "syscall"
+ "time"
+
+ "fluxrec/core"
+ "fluxrec/miniflux"
+)
+
+type ServeCommand struct {
+ ModelPath string
+ Cursor string
+ Runs string
+ Addr string
+ RecFeedID int64
+ TopN int
+ ExploreFrac float64
+ RSSBatches int
+ RefreshAt string
+
+ pollsOK atomic.Int64
+ pollsFailed atomic.Int64
+}
+
+func (c *ServeCommand) Name() string { return "serve" }
+
+func (c *ServeCommand) Init(args []string) error {
+ fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+ fs.Usage = func() {
+ fmt.Fprint(fs.Output(), `Usage: fluxrec serve [flags]
+
+Poll Miniflux for new entries, score each once, keep a top-N + exploration
+batch per refresh in runs.jsonl, and serve the recent batches as RSS at
+/recommendations.xml plus JSON status at /api/status. Read-only toward
+Miniflux. Refresh times are wall-clock in the process timezone ($TZ,
+default UTC). Exits immediately if the model file is missing or corrupt.
+`)
+ fs.PrintDefaults()
+ }
+ fs.StringVar(&c.ModelPath, "model", "model.json", "Path to model.json")
+ fs.StringVar(&c.Cursor, "cursor", "cursor.json", "Path to the poll cursor file")
+ fs.StringVar(&c.Runs, "runs", "runs.jsonl", "Path to the run log (doubles as the RSS render source)")
+ fs.StringVar(&c.Addr, "addr", ":8080", "Listen address for RSS + status")
+ fs.Int64Var(&c.RecFeedID, "rec-feed-id", 0, "Miniflux feed ID of the recommendations feed itself; its entries are skipped so serve never re-recommends its own output (0 = not subscribed yet)")
+ fs.IntVar(&c.TopN, "top-n", 15, "Batch size: ranked picks plus exploration")
+ fs.Float64Var(&c.ExploreFrac, "explore-frac", 0.2, "Fraction of the batch spent on exploration (1/3 runner-up, 2/3 random)")
+ fs.IntVar(&c.RSSBatches, "rss-batches", 21, "How many recent batches the RSS feed renders")
+ fs.StringVar(&c.RefreshAt, "refresh-at", "06:00,12:00,18:00", "Comma-separated daily refresh times (HH:MM, local time)")
+ if err := fs.Parse(args); err != nil {
+ return err
+ }
+ if c.TopN < 1 {
+ return fmt.Errorf("--top-n must be >= 1, got %d", c.TopN)
+ }
+ if c.ExploreFrac < 0 || c.ExploreFrac >= 1 {
+ return fmt.Errorf("--explore-frac must be in [0,1), got %v", c.ExploreFrac)
+ }
+ if c.RSSBatches < 1 {
+ return fmt.Errorf("--rss-batches must be >= 1, got %d", c.RSSBatches)
+ }
+ if _, err := parseRefreshAt(c.RefreshAt); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (c *ServeCommand) Run(stdin io.Reader, stdout io.Writer) error {
+ cfg, err := miniflux.ConfigFromEnv()
+ if err != nil {
+ return err
+ }
+ client := miniflux.NewClient(cfg)
+
+ model, err := loadModel(c.ModelPath)
+ if err != nil {
+ return err
+ }
+ vec := core.CreateVectorizerFromModel(model)
+ mins, _ := parseRefreshAt(c.RefreshAt) // validated in Init
+
+ rng := rand.New(rand.NewSource(time.Now().UnixNano()))
+ poll := func() {
+ if err := c.pollOnce(client, vec, model, rng, stdout); err != nil {
+ c.pollsFailed.Add(1)
+ fmt.Fprintf(os.Stderr, "serve: poll failed (cursor not advanced, will retry next cycle): %v\n", err)
+ } else {
+ c.pollsOK.Add(1)
+ }
+ }
+
+ // Boot catch-up: if the last logged run is older than half the smallest
+ // gap between refreshes (e.g. the container was down across a slot),
+ // poll immediately rather than waiting for the next scheduled time.
+ due, err := catchUpDue(c.Runs, mins, time.Now())
+ if err != nil {
+ return err
+ }
+ if due {
+ poll()
+ }
+
+ ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+ defer stop()
+
+ srv := &http.Server{
+ Addr: c.Addr,
+ Handler: c.routes(mins, model.CreatedAt),
+ ReadTimeout: core.DefaultReadTimeout,
+ WriteTimeout: core.DefaultWriteTimeout,
+ IdleTimeout: core.DefaultIdleTimeout,
+ }
+ go func() {
+ if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+ fmt.Fprintf(os.Stderr, "serve: http listener failed: %v\n", err)
+ stop()
+ }
+ }()
+
+ go func() {
+ for {
+ timer := time.NewTimer(time.Until(nextRefresh(time.Now(), mins)))
+ select {
+ case <-ctx.Done():
+ timer.Stop()
+ return
+ case <-timer.C:
+ poll()
+ }
+ }
+ }()
+
+ fmt.Fprintf(stdout, "serve: listening on %s (refresh at %s, model %s)\n", c.Addr, c.RefreshAt, c.ModelPath)
+ <-ctx.Done()
+
+ shutdownCtx, cancel := context.WithTimeout(context.Background(), core.DefaultShutdownTimeout)
+ defer cancel()
+ return srv.Shutdown(shutdownCtx)
+}
+
+// pollCursor is the persisted serve cursor: the largest entry ID fully
+// processed so far.
+type pollCursor struct {
+ AfterEntryID int64 `json:"after_entry_id"`
+}
+
+func loadCursor(path string) (int64, error) {
+ body, err := os.ReadFile(path)
+ if err != nil {
+ if os.IsNotExist(err) {
+ return 0, nil
+ }
+ return 0, fmt.Errorf("reading cursor %s: %w", path, err)
+ }
+ var c pollCursor
+ if err := json.Unmarshal(body, &c); err != nil {
+ return 0, fmt.Errorf("parsing cursor %s: %w", path, err)
+ }
+ return c.AfterEntryID, nil
+}
+
+func saveCursor(path string, id int64) error {
+ body, err := json.Marshal(pollCursor{AfterEntryID: id})
+ if err != nil {
+ return err
+ }
+ if err := os.WriteFile(path, body, 0o644); err != nil {
+ return fmt.Errorf("writing cursor %s: %w", path, err)
+ }
+ return nil
+}
+
+// pollOnce fetches every entry newer than the cursor, scores each one,
+// assembles the batch, appends one row to runs.jsonl, and only then
+// advances the cursor — process → append log → save cursor, so a crash at
+// any point re-processes at worst one batch. Nothing is written for an
+// empty poll (a run of empty batches would starve the count-based RSS
+// tail).
+func (c *ServeCommand) pollOnce(client *miniflux.Client, vec *core.TFIDFVectorizer, model *core.ModelEnvelope, rng *rand.Rand, stdout io.Writer) error {
+ started := time.Now()
+ cursor, err := loadCursor(c.Cursor)
+ if err != nil {
+ return err
+ }
+
+ var scored []scoredEntry
+ seen, recSkipped := 0, 0
+ maxID := cursor
+ f := miniflux.Filters{AfterEntryID: cursor, Limit: 200, Direction: "asc"}
+ for {
+ set, err := client.Entries(f)
+ if err != nil {
+ return err
+ }
+ if len(set.Entries) == 0 {
+ break
+ }
+ for _, e := range set.Entries {
+ seen++
+ if e.ID > maxID {
+ maxID = e.ID
+ }
+ // The recommendations feed's own items are our output, not
+ // candidates: skip them or the loop feeds itself.
+ if c.RecFeedID != 0 && e.Feed.ID == c.RecFeedID {
+ recSkipped++
+ continue
+ }
+ vecs := vec.Transform([]string{core.ScoringText(e.Title, e.Feed.Title)})
+ s, err := core.PredictScore(vecs[0], model.Weights)
+ if err != nil {
+ return fmt.Errorf("scoring entry %d: %w", e.ID, err)
+ }
+ scored = append(scored, scoredEntry{entry: e, score: s})
+ }
+ if len(set.Entries) < f.Limit || maxID == f.AfterEntryID {
+ // Short page: done. Unchanged max ID despite a full page means
+ // the server ignored the cursor — break rather than spin.
+ break
+ }
+ f.AfterEntryID = maxID
+ }
+
+ if seen == 0 {
+ fmt.Fprintln(stdout, "serve: no new entries")
+ return nil
+ }
+
+ row := core.RunRow{
+ RanAt: started.UTC(),
+ EntriesSeen: seen,
+ RecFeedSkipped: recSkipped,
+ Kept: assembleBatch(scored, c.TopN, c.ExploreFrac, rng),
+ DurationMs: time.Since(started).Milliseconds(),
+ ModelCreatedAt: model.CreatedAt,
+ }
+ if err := core.AppendRun(c.Runs, row); err != nil {
+ return err
+ }
+ if err := saveCursor(c.Cursor, maxID); err != nil {
+ return err
+ }
+ fmt.Fprintf(stdout, "serve: seen=%d kept=%d skipped-rec=%d cursor=%d\n",
+ seen, len(row.Kept), recSkipped, maxID)
+ return nil
+}
+
+// scoredEntry is an entry with its model score; batch assembly input.
+type scoredEntry struct {
+ entry miniflux.Entry
+ score float64
+}
+
+// assembleBatch ranks scored entries by score descending and selects the
+// batch: the top (topN minus the exploration budget) are ranked; the
+// budget splits at a fixed 1/3 : 2/3 into runner-ups (highest-scored just
+// below the cut — the model's borderline cases) and random picks from the
+// remaining tail (serendipity and a recall-ceiling estimator). With fewer
+// candidates than the ranked cut, everything is kept as ranked.
+func assembleBatch(entries []scoredEntry, topN int, exploreFrac float64, rng *rand.Rand) []core.KeptItem {
+ ranked := make([]scoredEntry, len(entries))
+ copy(ranked, entries)
+ sort.SliceStable(ranked, func(i, j int) bool {
+ if ranked[i].score == ranked[j].score {
+ return ranked[i].entry.ID < ranked[j].entry.ID
+ }
+ return ranked[i].score > ranked[j].score
+ })
+
+ explore := int(math.Round(float64(topN) * exploreFrac))
+ nRanked := topN - explore
+ if len(ranked) <= nRanked {
+ return keptFrom(ranked, core.KindRanked)
+ }
+ kept := keptFrom(ranked[:nRanked], core.KindRanked)
+
+ nRunnerUp := min(explore/3, len(ranked)-nRanked)
+ kept = append(kept, keptFrom(ranked[nRanked:nRanked+nRunnerUp], core.KindRunnerUp)...)
+
+ rest := ranked[nRanked+nRunnerUp:]
+ nRandom := min(explore-nRunnerUp, len(rest))
+ if nRandom > 0 {
+ picks := make([]scoredEntry, 0, nRandom)
+ for _, i := range rng.Perm(len(rest))[:nRandom] {
+ picks = append(picks, rest[i])
+ }
+ // Log random picks in ID order so runs.jsonl stays diff-friendly.
+ sort.Slice(picks, func(i, j int) bool { return picks[i].entry.ID < picks[j].entry.ID })
+ kept = append(kept, keptFrom(picks, core.KindRandom)...)
+ }
+ return kept
+}
+
+// keptFrom converts scored entries into run-log items, cleaning content on
+// the way in so RSS rendering never sees raw feed HTML.
+func keptFrom(entries []scoredEntry, kind string) []core.KeptItem {
+ out := make([]core.KeptItem, len(entries))
+ for i, se := range entries {
+ out[i] = core.KeptItem{
+ EntryID: se.entry.ID,
+ URL: se.entry.URL,
+ Title: se.entry.Title,
+ FeedTitle: se.entry.Feed.Title,
+ PublishedAt: se.entry.PublishedAt.UTC(),
+ Content: core.CleanFeedContent(se.entry.Content),
+ Score: se.score,
+ Kind: kind,
+ }
+ }
+ return out
+}
+
+// parseRefreshAt parses "06:00,12:00,18:00" into ascending
+// minutes-after-midnight.
+func parseRefreshAt(spec string) ([]int, error) {
+ var mins []int
+ for _, part := range strings.Split(spec, ",") {
+ part = strings.TrimSpace(part)
+ hm := strings.Split(part, ":")
+ if len(hm) != 2 {
+ return nil, fmt.Errorf("--refresh-at: %q is not HH:MM", part)
+ }
+ h, errH := strconv.Atoi(hm[0])
+ m, errM := strconv.Atoi(hm[1])
+ if errH != nil || errM != nil || h < 0 || h > 23 || m < 0 || m > 59 {
+ return nil, fmt.Errorf("--refresh-at: %q is not a valid time", part)
+ }
+ mins = append(mins, h*60+m)
+ }
+ sort.Ints(mins)
+ return mins, nil
+}
+
+// nextRefresh returns the next occurrence of any refresh time strictly
+// after now, in now's location, wrapping to tomorrow past the last slot.
+func nextRefresh(now time.Time, mins []int) time.Time {
+ var best time.Time
+ for _, m := range mins {
+ t := time.Date(now.Year(), now.Month(), now.Day(), m/60, m%60, 0, 0, now.Location())
+ if !t.After(now) {
+ t = t.Add(24 * time.Hour)
+ }
+ if best.IsZero() || t.Before(best) {
+ best = t
+ }
+ }
+ return best
+}
+
+// catchUpDue reports whether startup should trigger an immediate poll:
+// nothing logged yet, or the last run is older than half the smallest gap
+// between refresh times (a container down across a slot catches up on boot
+// instead of waiting for the next one).
+func catchUpDue(runsPath string, mins []int, now time.Time) (bool, error) {
+ rows, err := core.ReadRunsTail(runsPath, 1)
+ if err != nil {
+ return false, err
+ }
+ if len(rows) == 0 {
+ return true, nil
+ }
+ half := time.Duration(minGapMinutes(mins)) * time.Minute / 2
+ return now.Sub(rows[len(rows)-1].RanAt) > half, nil
+}
+
+// minGapMinutes returns the smallest gap between consecutive refresh times,
+// including the wrap across midnight (a single time has a 24h gap).
+func minGapMinutes(mins []int) int {
+ if len(mins) < 2 {
+ return 24 * 60
+ }
+ sorted := make([]int, len(mins))
+ copy(sorted, mins)
+ sort.Ints(sorted)
+ gap := 24*60 - sorted[len(sorted)-1] + sorted[0]
+ for i := 1; i < len(sorted); i++ {
+ gap = min(gap, sorted[i]-sorted[i-1])
+ }
+ return gap
+}
+
+// routes wires the two surfaces: the RSS feed and the JSON status.
+func (c *ServeCommand) routes(mins []int, modelCreatedAt time.Time) http.Handler {
+ mux := http.NewServeMux()
+ mux.HandleFunc("/recommendations.xml", func(w http.ResponseWriter, r *http.Request) {
+ runs, err := core.ReadRunsTail(c.Runs, c.RSSBatches)
+ if err != nil {
+ http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ body, err := RenderRSS(runs)
+ if err != nil {
+ http.Error(w, "rendering RSS: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ w.Header().Set("Content-Type", "application/rss+xml; charset=utf-8")
+ _, _ = w.Write(body)
+ })
+ mux.HandleFunc("/api/status", func(w http.ResponseWriter, r *http.Request) {
+ runs, err := core.ReadRunsTail(c.Runs, 1)
+ if err != nil {
+ http.Error(w, "reading runs: "+err.Error(), http.StatusInternalServerError)
+ return
+ }
+ status := map[string]any{
+ "model_created_at": modelCreatedAt.UTC(),
+ "model_age_hours": math.Round(time.Since(modelCreatedAt).Hours()),
+ "next_refresh_at": nextRefresh(time.Now(), mins),
+ "polls_ok": c.pollsOK.Load(),
+ "polls_failed": c.pollsFailed.Load(),
+ "last_run": nil,
+ }
+ if len(runs) > 0 {
+ last := runs[len(runs)-1]
+ status["last_run"] = map[string]any{
+ "ran_at": last.RanAt,
+ "entries_seen": last.EntriesSeen,
+ "rec_feed_skipped": last.RecFeedSkipped,
+ "kept": len(last.Kept),
+ "duration_ms": last.DurationMs,
+ "model_created_at": last.ModelCreatedAt,
+ }
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _ = json.NewEncoder(w).Encode(status)
+ })
+ return mux
+}
diff --git a/cmds/serve_test.go b/cmds/serve_test.go
new file mode 100644
index 0000000..d7df2dd
--- /dev/null
+++ b/cmds/serve_test.go
@@ -0,0 +1,407 @@
+// Tests for `fluxrec serve`: the poll loop against an httptest fake
+// Miniflux (multi-page, mid-run crash), batch assembly split/kinds, and
+// the refresh-time computation including the midnight wrap. Helpers
+// (testEntry, makeEntries, entriesJSON) come from export_test.go.
+package cmds
+
+import (
+ "encoding/json"
+ "fmt"
+ "math/rand"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "fluxrec/core"
+ "fluxrec/miniflux"
+)
+
+// testModel is a minimal hand-built model: one unigram ("title"), so every
+// makeEntries title scores identically and batch ties break by entry ID.
+func testModel() *core.ModelEnvelope {
+ return &core.ModelEnvelope{
+ CreatedAt: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC),
+ Meta: map[string]any{},
+ Vectorizer: map[string]float64{"title": 1.0},
+ OrderedVocab: []string{"title"},
+ Weights: []float64{2.0, 0.1},
+ }
+}
+
+// idServer pages its entry pool by after_entry_id ascending, like the
+// serve poll expects, and can be told to die on the Nth request.
+type idServer struct {
+ entries []testEntry // ascending IDs
+ queries []string
+ calls int
+ failOn int // 1-based request number to 500 on; 0 = never
+}
+
+func (s *idServer) handler() http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ s.calls++
+ s.queries = append(s.queries, r.URL.RawQuery)
+ if s.failOn == s.calls {
+ http.Error(w, "boom", http.StatusInternalServerError)
+ return
+ }
+ q := r.URL.Query()
+ after, _ := strconv.ParseInt(q.Get("after_entry_id"), 10, 64)
+ limit, _ := strconv.Atoi(q.Get("limit"))
+ if limit <= 0 {
+ limit = len(s.entries)
+ }
+ var page []testEntry
+ for _, e := range s.entries {
+ if e.ID > after {
+ page = append(page, e)
+ }
+ }
+ if len(page) > limit {
+ page = page[:limit]
+ }
+ w.Header().Set("Content-Type", "application/json")
+ fmt.Fprint(w, entriesJSON(len(page), page))
+ }
+}
+
+func serveTestClient(t *testing.T, s http.Handler) *miniflux.Client {
+ t.Helper()
+ srv := httptest.NewServer(s)
+ t.Cleanup(srv.Close)
+ return miniflux.NewClientWithHTTP(miniflux.Config{BaseURL: srv.URL, Token: "t"}, srv.Client())
+}
+
+func TestPollMultiPage(t *testing.T) {
+ // 450 entries => three 200/200/50 pages off one ascending cursor.
+ fake := &idServer{entries: makeEntries(450, 1, false, 1)}
+ client := serveTestClient(t, fake.handler())
+
+ dir := t.TempDir()
+ model := testModel()
+ cmd := &ServeCommand{
+ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"),
+ TopN: 15, ExploreFrac: 0.2,
+ }
+ var out strings.Builder
+ rng := rand.New(rand.NewSource(1))
+ if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model, rng, &out); err != nil {
+ t.Fatalf("pollOnce: %v", err)
+ }
+
+ if fake.calls != 3 {
+ t.Fatalf("expected 3 paged requests, got %d", fake.calls)
+ }
+ if !strings.Contains(fake.queries[0], "direction=asc") || !strings.Contains(fake.queries[0], "limit=200") {
+ t.Errorf("first query missing direction/limit: %s", fake.queries[0])
+ }
+ if strings.Contains(fake.queries[0], "after_entry_id") {
+ t.Errorf("first query should have no cursor: %s", fake.queries[0])
+ }
+ if !strings.Contains(fake.queries[1], "after_entry_id=200") ||
+ !strings.Contains(fake.queries[2], "after_entry_id=400") {
+ t.Errorf("cursor paging wrong: %q", fake.queries)
+ }
+
+ cursor, err := loadCursor(cmd.Cursor)
+ if err != nil {
+ t.Fatalf("loadCursor: %v", err)
+ }
+ if cursor != 450 {
+ t.Errorf("cursor = %d, want 450", cursor)
+ }
+
+ rows, err := core.ReadRuns(cmd.Runs)
+ if err != nil {
+ t.Fatalf("ReadRuns: %v", err)
+ }
+ if len(rows) != 1 {
+ t.Fatalf("expected 1 run row, got %d", len(rows))
+ }
+ row := rows[0]
+ if row.EntriesSeen != 450 {
+ t.Errorf("entries_seen = %d, want 450", row.EntriesSeen)
+ }
+ if len(row.Kept) != 15 {
+ t.Errorf("kept = %d, want 15", len(row.Kept))
+ }
+ if !row.ModelCreatedAt.Equal(model.CreatedAt) {
+ t.Errorf("model_created_at = %v, want %v", row.ModelCreatedAt, model.CreatedAt)
+ }
+}
+
+func TestPollCrashLeavesNoState(t *testing.T) {
+ fake := &idServer{entries: makeEntries(450, 1, false, 1), failOn: 2}
+ client := serveTestClient(t, fake.handler())
+
+ dir := t.TempDir()
+ model := testModel()
+ cmd := &ServeCommand{
+ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"),
+ TopN: 15, ExploreFrac: 0.2,
+ }
+ err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model,
+ rand.New(rand.NewSource(1)), &strings.Builder{})
+ if err == nil {
+ t.Fatal("expected pollOnce to fail on the second page")
+ }
+ // Crash-safe ordering: the page-1 work must not be half-persisted.
+ if _, statErr := os.Stat(cmd.Cursor); !os.IsNotExist(statErr) {
+ t.Error("cursor file exists despite the failed poll")
+ }
+ if _, statErr := os.Stat(cmd.Runs); !os.IsNotExist(statErr) {
+ t.Error("runs file exists despite the failed poll")
+ }
+}
+
+func TestPollEmptyWritesNothing(t *testing.T) {
+ fake := &idServer{}
+ client := serveTestClient(t, fake.handler())
+
+ dir := t.TempDir()
+ model := testModel()
+ cmd := &ServeCommand{
+ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"),
+ TopN: 15, ExploreFrac: 0.2,
+ }
+ if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model,
+ rand.New(rand.NewSource(1)), &strings.Builder{}); err != nil {
+ t.Fatalf("pollOnce: %v", err)
+ }
+ if rows, _ := core.ReadRuns(cmd.Runs); len(rows) != 0 {
+ t.Errorf("empty poll wrote %d rows, want 0 (RSS tail starvation)", len(rows))
+ }
+}
+
+func TestPollSkipsRecFeed(t *testing.T) {
+ fake := &idServer{
+ entries: append(makeEntries(10, 1, false, 1), makeEntries(5, 100, false, 99)...),
+ }
+ client := serveTestClient(t, fake.handler())
+
+ dir := t.TempDir()
+ model := testModel()
+ cmd := &ServeCommand{
+ Cursor: filepath.Join(dir, "cursor.json"), Runs: filepath.Join(dir, "runs.jsonl"),
+ TopN: 30, ExploreFrac: 0, RecFeedID: 99,
+ }
+ if err := cmd.pollOnce(client, core.CreateVectorizerFromModel(model), model,
+ rand.New(rand.NewSource(1)), &strings.Builder{}); err != nil {
+ t.Fatalf("pollOnce: %v", err)
+ }
+ rows, _ := core.ReadRuns(cmd.Runs)
+ if rows[0].EntriesSeen != 15 || rows[0].RecFeedSkipped != 5 {
+ t.Errorf("seen=%d skipped=%d, want 15 / 5", rows[0].EntriesSeen, rows[0].RecFeedSkipped)
+ }
+ for _, k := range rows[0].Kept {
+ if k.EntryID >= 100 {
+ t.Errorf("rec-feed entry %d was kept; serve must not re-recommend its own output", k.EntryID)
+ }
+ }
+}
+
+func TestAssembleBatchSplitAndKinds(t *testing.T) {
+ // 30 candidates, scores increasing with ID => deterministic ranks.
+ entries := make([]scoredEntry, 30)
+ for i := range entries {
+ entries[i] = scoredEntry{
+ entry: miniflux.Entry{ID: int64(i + 1), Title: fmt.Sprintf("T%d", i+1)},
+ score: float64(i + 1),
+ }
+ }
+ kept := assembleBatch(entries, 15, 0.2, rand.New(rand.NewSource(7)))
+ if len(kept) != 15 {
+ t.Fatalf("kept = %d, want 15", len(kept))
+ }
+ counts := map[string]int{}
+ var rankedIDs, runnerUpIDs []int64
+ random := map[int64]bool{}
+ for _, k := range kept {
+ counts[k.Kind]++
+ switch k.Kind {
+ case core.KindRanked:
+ rankedIDs = append(rankedIDs, k.EntryID)
+ case core.KindRunnerUp:
+ runnerUpIDs = append(runnerUpIDs, k.EntryID)
+ case core.KindRandom:
+ random[k.EntryID] = true
+ }
+ }
+ if counts[core.KindRanked] != 12 || counts[core.KindRunnerUp] != 1 || counts[core.KindRandom] != 2 {
+ t.Errorf("split = %v, want ranked=12 runnerup=1 random=2", counts)
+ }
+ // Ranked are the top 12 by score, in score order.
+ for i, id := range rankedIDs {
+ want := int64(30 - i)
+ if id != want {
+ t.Fatalf("ranked[%d] = %d, want %d", i, id, want)
+ }
+ }
+ // The runner-up is the highest-scored entry below the cut.
+ if len(runnerUpIDs) != 1 || runnerUpIDs[0] != 18 {
+ t.Errorf("runner-up = %v, want [18]", runnerUpIDs)
+ }
+ // Random picks come from the tail below the runner-up, no duplicates.
+ for id := range random {
+ if id > 17 {
+ t.Errorf("random pick %d not below the runner-up cut", id)
+ }
+ }
+}
+
+func TestAssembleBatchFewCandidates(t *testing.T) {
+ entries := []scoredEntry{
+ {entry: miniflux.Entry{ID: 1}, score: 0.1},
+ {entry: miniflux.Entry{ID: 2}, score: 0.9},
+ {entry: miniflux.Entry{ID: 3}, score: 0.5},
+ }
+ kept := assembleBatch(entries, 15, 0.2, rand.New(rand.NewSource(1)))
+ if len(kept) != 3 {
+ t.Fatalf("kept = %d, want all 3 as ranked", len(kept))
+ }
+ for _, k := range kept {
+ if k.Kind != core.KindRanked {
+ t.Errorf("kind = %s, want ranked when candidates < cut", k.Kind)
+ }
+ }
+ if kept[0].EntryID != 2 { // highest score first
+ t.Errorf("first kept ID = %d, want 2 (highest score)", kept[0].EntryID)
+ }
+}
+
+func TestParseRefreshAt(t *testing.T) {
+ mins, err := parseRefreshAt("18:00,06:00, 12:30")
+ if err != nil {
+ t.Fatalf("parseRefreshAt: %v", err)
+ }
+ want := []int{360, 750, 1080}
+ if fmt.Sprint(mins) != fmt.Sprint(want) {
+ t.Errorf("mins = %v, want %v (sorted)", mins, want)
+ }
+ for _, bad := range []string{"24:00", "abc", "12", "12:60", ""} {
+ if _, err := parseRefreshAt(bad); err == nil {
+ t.Errorf("parseRefreshAt(%q) unexpectedly succeeded", bad)
+ }
+ }
+}
+
+func TestNextRefresh(t *testing.T) {
+ mins := []int{360, 720, 1080} // 06:00, 12:00, 18:00
+ at := func(day, h, m int) time.Time {
+ return time.Date(2024, 3, day, h, m, 0, 0, time.UTC)
+ }
+ cases := []struct {
+ now time.Time
+ want time.Time
+ }{
+ {at(1, 13, 0), at(1, 18, 0)}, // same-day next slot
+ {at(1, 6, 0), at(1, 12, 0)}, // exactly on a slot: strictly after
+ {at(1, 23, 30), at(2, 6, 0)}, // past last slot: midnight wrap
+ {at(1, 0, 0), at(1, 6, 0)}, // before all slots
+ }
+ for _, tc := range cases {
+ if got := nextRefresh(tc.now, mins); !got.Equal(tc.want) {
+ t.Errorf("nextRefresh(%v) = %v, want %v", tc.now, got, tc.want)
+ }
+ }
+}
+
+func TestMinGapMinutes(t *testing.T) {
+ if got := minGapMinutes([]int{360, 720, 1080}); got != 360 {
+ t.Errorf("minGap = %d, want 360", got)
+ }
+ if got := minGapMinutes([]int{1380, 60}); got != 120 {
+ t.Errorf("minGap across midnight = %d, want 120", got)
+ }
+ if got := minGapMinutes([]int{360}); got != 1440 {
+ t.Errorf("single slot minGap = %d, want 1440", got)
+ }
+}
+
+func TestCatchUpDue(t *testing.T) {
+ mins := []int{360, 720, 1080} // min gap 6h => half = 3h
+ now := time.Date(2024, 3, 2, 12, 0, 0, 0, time.UTC)
+ dir := t.TempDir()
+ runs := filepath.Join(dir, "runs.jsonl")
+
+ // No log yet: first boot polls immediately.
+ if due, err := catchUpDue(runs, mins, now); err != nil || !due {
+ t.Errorf("no runs: due=%v err=%v, want true", due, err)
+ }
+ recent := core.RunRow{RanAt: now.Add(-time.Hour), EntriesSeen: 3}
+ if err := core.AppendRun(runs, recent); err != nil {
+ t.Fatalf("AppendRun: %v", err)
+ }
+ if due, _ := catchUpDue(runs, mins, now); due {
+ t.Error("last run 1h ago: due=true, want false (next slot will cover it)")
+ }
+ old := core.RunRow{RanAt: now.Add(-5 * time.Hour), EntriesSeen: 3}
+ if err := core.AppendRun(runs, old); err != nil {
+ t.Fatalf("AppendRun: %v", err)
+ }
+ if due, _ := catchUpDue(runs, mins, now); !due {
+ t.Error("last run 5h ago: due=false, want true (downtime across slots)")
+ }
+}
+
+func TestServeRoutes(t *testing.T) {
+ dir := t.TempDir()
+ runsPath := filepath.Join(dir, "runs.jsonl")
+ row := core.RunRow{
+ RanAt: time.Date(2024, 3, 1, 6, 0, 0, 0, time.UTC),
+ EntriesSeen: 42,
+ Kept: []core.KeptItem{
+ {EntryID: 101, URL: "https://example.com/a", Title: "Alpha", FeedTitle: "Feed A",
+ PublishedAt: time.Date(2024, 3, 1, 5, 0, 0, 0, time.UTC), Content: "body a", Score: 0.9, Kind: core.KindRanked},
+ {EntryID: 102, URL: "https://example.com/b", Title: "Beta", FeedTitle: "Feed B",
+ PublishedAt: time.Date(2024, 3, 1, 5, 30, 0, 0, time.UTC), Content: "body b", Score: 0.2, Kind: core.KindRandom},
+ },
+ }
+ if err := core.AppendRun(runsPath, row); err != nil {
+ t.Fatalf("AppendRun: %v", err)
+ }
+
+ cmd := &ServeCommand{Runs: runsPath, RSSBatches: 21}
+ handler := cmd.routes([]int{360, 720, 1080}, time.Now().Add(-48*time.Hour))
+
+ rec := httptest.NewRecorder()
+ handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/recommendations.xml", nil))
+ if rec.Code != http.StatusOK {
+ t.Fatalf("RSS status = %d", rec.Code)
+ }
+ if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "application/rss+xml") {
+ t.Errorf("RSS content-type = %q", ct)
+ }
+ body := rec.Body.String()
+ if !strings.Contains(body, "<guid isPermaLink=\"false\">mf:101</guid>") {
+ t.Error("RSS missing mf:101 guid")
+ }
+ if strings.Count(body, "<item>") != 2 {
+ t.Errorf("RSS item count = %d, want 2", strings.Count(body, "<item>"))
+ }
+
+ rec = httptest.NewRecorder()
+ handler.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/api/status", nil))
+ if rec.Code != http.StatusOK {
+ t.Fatalf("status = %d", rec.Code)
+ }
+ var status map[string]any
+ if err := json.Unmarshal(rec.Body.Bytes(), &status); err != nil {
+ t.Fatalf("status not JSON: %v", err)
+ }
+ last, ok := status["last_run"].(map[string]any)
+ if !ok {
+ t.Fatalf("status last_run missing: %v", status)
+ }
+ if last["entries_seen"].(float64) != 42 || last["kept"].(float64) != 2 {
+ t.Errorf("last_run = %v, want entries_seen=42 kept=2", last)
+ }
+ if status["model_age_hours"].(float64) <= 0 {
+ t.Errorf("model_age_hours = %v, want positive", status["model_age_hours"])
+ }
+}
diff --git a/cmds/train.go b/cmds/train.go
new file mode 100644
index 0000000..234ef06
--- /dev/null
+++ b/cmds/train.go
@@ -0,0 +1,430 @@
+// Train command: labels.jsonl → model.json + report.json.
+//
+// Time-split validation (train on older labels, evaluate on the newest
+// window — what serve actually faces). Headline metric is precision@15
+// (rank-based: of the 15 highest-scored validation entries, what fraction
+// are positives). Youden's J is computed and reported for information only;
+// it is NOT used as a cutoff — serve ranks by raw score.
+//
+// Reproducible: a fixed seed makes the downsampling deterministic, so the
+// same labels produce the same weights/vocab. IDF is fit on the full
+// (unbalanced) train set to preserve document frequencies, then the
+// majority class is downsampled to 1:1 before LR fit (matches the vendored
+// scholscan approach).
+package cmds
+
+import (
+ "encoding/json"
+ "flag"
+ "fmt"
+ "io"
+ "math"
+ "math/rand"
+ "os"
+ "sort"
+ "time"
+
+ "fluxrec/core"
+)
+
+// trainSeed fixes the downsampling RNG: same labels + same seed → same model.
+const trainSeed int64 = 20240301
+
+type TrainCommand struct {
+ Labels string
+ ModelOut string
+ ReportOut string
+ ValFrac float64
+ Lambda float64
+ MinDF int
+ MaxDF float64
+ NgramMax int
+}
+
+func (c *TrainCommand) Name() string { return "train" }
+
+func (c *TrainCommand) Init(args []string) error {
+ fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+ fs.Usage = func() {
+ fmt.Fprint(fs.Output(), `Usage: fluxrec train LABELS_JSONL [flags]
+
+Train a TF-IDF + logistic regression model from labels.jsonl. Writes the
+model to --model-out and a validation report (precision@15 etc.) to
+--report-out. Time-split: the newest --val-frac of labels (by captured_at)
+are held out for validation; the rest train the model.
+`)
+ fs.PrintDefaults()
+ }
+ fs.StringVar(&c.ModelOut, "model-out", "model.json", "Path to write the trained model JSON")
+ fs.StringVar(&c.ReportOut, "report-out", "report.json", "Path to write the validation report JSON")
+ fs.Float64Var(&c.ValFrac, "val-frac", 0.2, "Fraction of newest labels held out for validation")
+ fs.Float64Var(&c.Lambda, "lambda", 0.001, "L2 regularization strength")
+ fs.IntVar(&c.MinDF, "min-df", 2, "Minimum document frequency (absolute)")
+ fs.Float64Var(&c.MaxDF, "max-df", 0.8, "Maximum document frequency (ratio)")
+ fs.IntVar(&c.NgramMax, "ngram-max", 2, "Max n-gram size (1=unigrams, 2=+bigrams)")
+
+ for _, a := range args {
+ if a == "--help" || a == "-h" {
+ fs.Usage()
+ return flag.ErrHelp
+ }
+ }
+ if len(args) == 0 {
+ return fmt.Errorf("labels.jsonl path argument is required")
+ }
+ c.Labels = args[0]
+ if err := fs.Parse(args[1:]); err != nil {
+ return err
+ }
+ if c.ValFrac <= 0 || c.ValFrac >= 1 {
+ return fmt.Errorf("--val-frac must be between 0 and 1, got %v", c.ValFrac)
+ }
+ return nil
+}
+
+func (c *TrainCommand) Run(stdin io.Reader, stdout io.Writer) error {
+ rows, err := core.ReadLabels(c.Labels)
+ if err != nil {
+ return err
+ }
+ if len(rows) == 0 {
+ return fmt.Errorf("no labels in %s", c.Labels)
+ }
+
+ trainRows, valRows := core.SplitByTime(rows, c.ValFrac)
+ if len(trainRows) == 0 || len(valRows) == 0 {
+ return fmt.Errorf("time split produced an empty side (train=%d val=%d); adjust --val-frac or accumulate more labels",
+ len(trainRows), len(valRows))
+ }
+ valPos := 0
+ for _, r := range valRows {
+ if r.Label == core.LabelPos {
+ valPos++
+ }
+ }
+ if valPos < 5 {
+ return fmt.Errorf("only %d positives in validation (need >=5 for signal, not noise); lower --val-frac or accumulate more labels",
+ valPos)
+ }
+
+ model, report, err := c.trainAndEvaluate(trainRows, valRows, time.Now().UTC())
+ if err != nil {
+ return err
+ }
+
+ if err := writeJSON(c.ModelOut, model); err != nil {
+ return err
+ }
+ if err := writeJSON(c.ReportOut, report); err != nil {
+ return err
+ }
+
+ fmt.Fprintf(stdout, "train: pos=%d neg=%d | val: pos=%d neg=%d | precision@15=%.3f youden_j=%.3f | vocab=%d → %s\n",
+ report.TrainPos, report.TrainNeg, report.ValPos, report.ValNeg,
+ report.PrecisionAt15, report.YoudensJ, report.VocabSize, c.ModelOut)
+ return nil
+}
+
+func writeJSON(path string, v any) error {
+ body, err := json.MarshalIndent(v, "", " ")
+ if err != nil {
+ return err
+ }
+ if err := os.WriteFile(path, body, 0o644); err != nil {
+ return fmt.Errorf("writing %s: %w", path, err)
+ }
+ return nil
+}
+
+// trainAndEvaluate is the pure training core: given split rows and a clock
+// value, returns the model envelope and validation report. Pure w.r.t. the
+// inputs (no I/O, no wall clock) so two calls with the same rows + now are
+// byte-identical — the reproducibility guarantee.
+func (c *TrainCommand) trainAndEvaluate(trainRows, valRows []core.LabelRow, now time.Time) (*core.ModelEnvelope, *trainReport, error) {
+ trainDocs := make([]string, len(trainRows))
+ trainLabels := make([]float64, len(trainRows))
+ trainPos, trainNeg := 0, 0
+ for i, r := range trainRows {
+ trainDocs[i] = core.ScoringText(r.Title, r.FeedTitle)
+ if r.Label == core.LabelPos {
+ trainLabels[i] = 1.0
+ trainPos++
+ } else {
+ trainLabels[i] = 0.0
+ trainNeg++
+ }
+ }
+ valDocs := make([]string, len(valRows))
+ valLabels := make([]float64, len(valRows))
+ valPos, valNeg := 0, 0
+ for i, r := range valRows {
+ valDocs[i] = core.ScoringText(r.Title, r.FeedTitle)
+ if r.Label == core.LabelPos {
+ valLabels[i] = 1.0
+ valPos++
+ } else {
+ valLabels[i] = 0.0
+ valNeg++
+ }
+ }
+
+ // Fit IDF on the full (unbalanced) train set to preserve document
+ // frequencies, then downsample the majority to 1:1 before LR fit.
+ vectorizer := &core.TFIDFVectorizer{
+ NgramMin: 1, NgramMax: c.NgramMax,
+ MinDF: c.MinDF, MaxDF: c.MaxDF, VocabCap: 50000,
+ Vocabulary: make(map[string]float64),
+ }
+ vectorizer.Fit(trainDocs)
+
+ balDocs, balLabels := downsampleBalance(trainDocs, trainLabels, trainSeed)
+ trainVectors := vectorizer.Transform(balDocs)
+ valVectors := vectorizer.Transform(valDocs)
+
+ lr := &core.LogisticRegression{
+ LearningRate: 0.5, Lambda: c.Lambda, Iterations: 500, Tolerance: 0.000001,
+ }
+ lr.Validate()
+ weights, err := lr.Fit(trainVectors, balLabels, map[float64]float64{1.0: 1.0, 0.0: 1.0})
+ if err != nil {
+ return nil, nil, fmt.Errorf("training logistic regression: %w", err)
+ }
+
+ precisionAt15, youdenJ, posStats, negStats := evaluate(valVectors, valLabels, weights)
+
+ minCap, maxCap := trainRows[0].CapturedAt, trainRows[0].CapturedAt
+ for _, r := range trainRows {
+ if r.CapturedAt.Before(minCap) {
+ minCap = r.CapturedAt
+ }
+ if r.CapturedAt.After(maxCap) {
+ maxCap = r.CapturedAt
+ }
+ }
+
+ vecParams := map[string]any{
+ "ngram_min": vectorizer.NgramMin, "ngram_max": vectorizer.NgramMax,
+ "min_df": vectorizer.MinDF, "max_df": vectorizer.MaxDF, "vocab_cap": vectorizer.VocabCap,
+ }
+ modelParams := map[string]any{
+ "learning_rate": lr.LearningRate, "lambda": lr.Lambda,
+ "iterations": lr.Iterations, "tolerance": lr.Tolerance,
+ }
+
+ model := &core.ModelEnvelope{
+ Algorithm: "tfidf-go", Impl: "go", Version: "1",
+ CreatedAt: now,
+ Meta: map[string]any{
+ "scorer": "tfidf",
+ "seed": trainSeed,
+ "train_size": len(trainRows),
+ "val_size": len(valRows),
+ "train_pos": trainPos,
+ "train_neg": trainNeg,
+ "train_window_min": minCap.UTC().Format(time.RFC3339),
+ "train_window_max": maxCap.UTC().Format(time.RFC3339),
+ "val_frac": c.ValFrac,
+ "precision_at_15": precisionAt15,
+ "youdens_j": youdenJ,
+ "vectorizer_params": vecParams,
+ "model_params": modelParams,
+ },
+ Vectorizer: vectorizer.Vocabulary,
+ OrderedVocab: vectorizer.OrderedVocab,
+ Weights: weights,
+ }
+
+ report := &trainReport{
+ Scorer: "tfidf", Seed: trainSeed,
+ TrainSize: len(trainRows), ValSize: len(valRows),
+ TrainPos: trainPos, TrainNeg: trainNeg,
+ ValPos: valPos, ValNeg: valNeg,
+ TrainWindowMin: minCap.UTC(), TrainWindowMax: maxCap.UTC(),
+ ValFrac: c.ValFrac, VocabSize: len(vectorizer.OrderedVocab),
+ PrecisionAt15: precisionAt15, YoudensJ: youdenJ,
+ PositiveScores: posStats, NegativeScores: negStats,
+ VectorizerParams: vecParams, ModelParams: modelParams,
+ }
+ return model, report, nil
+}
+
+// downsampleBalance downsamples the majority class to a 1:1 ratio with the
+// minority, using a seeded RNG for reproducibility. If already balanced,
+// returns the inputs unchanged.
+func downsampleBalance(docs []string, labels []float64, seed int64) ([]string, []float64) {
+ var posDocs, negDocs []string
+ var posLabels, negLabels []float64
+ for i, label := range labels {
+ if label == 1.0 {
+ posDocs = append(posDocs, docs[i])
+ posLabels = append(posLabels, label)
+ } else {
+ negDocs = append(negDocs, docs[i])
+ negLabels = append(negLabels, label)
+ }
+ }
+ if len(posDocs) == len(negDocs) {
+ return docs, labels
+ }
+ var majorityDocs, minorityDocs []string
+ var majorityLabels, minorityLabels []float64
+ if len(negDocs) > len(posDocs) {
+ majorityDocs, minorityDocs = negDocs, posDocs
+ majorityLabels, minorityLabels = negLabels, posLabels
+ } else {
+ majorityDocs, minorityDocs = posDocs, negDocs
+ majorityLabels, minorityLabels = posLabels, negLabels
+ }
+ minoritySize := len(minorityDocs)
+ rng := rand.New(rand.NewSource(seed))
+ indices := make([]int, len(majorityDocs))
+ for i := range indices {
+ indices[i] = i
+ }
+ rng.Shuffle(len(indices), func(i, j int) {
+ indices[i], indices[j] = indices[j], indices[i]
+ })
+ outDocs := make([]string, 0, minoritySize*2)
+ outLabels := make([]float64, 0, minoritySize*2)
+ outDocs = append(outDocs, minorityDocs...)
+ outLabels = append(outLabels, minorityLabels...)
+ for i := 0; i < minoritySize; i++ {
+ outDocs = append(outDocs, majorityDocs[indices[i]])
+ outLabels = append(outLabels, majorityLabels[indices[i]])
+ }
+ return outDocs, outLabels
+}
+
+// evaluate scores the validation set and returns precision@15 (headline),
+// Youden's J (information only), and per-class score stats.
+func evaluate(valVectors [][]float64, valLabels []float64, weights []float64) (precisionAt15, youdenJ float64, posStats, negStats map[string]float64) {
+ type scored struct {
+ score float64
+ label float64
+ idx int
+ }
+ ss := make([]scored, len(valVectors))
+ for i, v := range valVectors {
+ s, err := core.PredictScore(v, weights)
+ if err != nil {
+ s = 0
+ }
+ ss[i] = scored{s, valLabels[i], i}
+ }
+
+ // precision@15: rank by score desc (ties by idx), take top 15, count positives.
+ ranked := make([]scored, len(ss))
+ copy(ranked, ss)
+ sort.SliceStable(ranked, func(i, j int) bool {
+ if ranked[i].score == ranked[j].score {
+ return ranked[i].idx < ranked[j].idx
+ }
+ return ranked[i].score > ranked[j].score
+ })
+ k := 15
+ if k > len(ranked) {
+ k = len(ranked)
+ }
+ topPos := 0
+ for i := 0; i < k; i++ {
+ if ranked[i].label == 1.0 {
+ topPos++
+ }
+ }
+ if k > 0 {
+ precisionAt15 = float64(topPos) / float64(k)
+ }
+
+ // Youden's J (information only): best sensitivity + specificity - 1 over
+ // a threshold sweep. Not used as a cutoff anywhere.
+ bestJ := -1.0
+ for t := 5; t <= 95; t++ {
+ threshold := float64(t) / 100.0
+ var tp, fp, tn, fn int
+ for _, s := range ss {
+ pred := s.score > threshold
+ switch {
+ case pred && s.label == 1.0:
+ tp++
+ case pred && s.label == 0.0:
+ fp++
+ case !pred && s.label == 1.0:
+ fn++
+ default:
+ tn++
+ }
+ }
+ sens := 0.0
+ if tp+fn > 0 {
+ sens = float64(tp) / float64(tp+fn)
+ }
+ spec := 0.0
+ if tn+fp > 0 {
+ spec = float64(tn) / float64(tn+fp)
+ }
+ if j := sens + spec - 1.0; j > bestJ {
+ bestJ = j
+ }
+ }
+ youdenJ = bestJ
+
+ var pos, neg []float64
+ for _, s := range ss {
+ if s.label == 1.0 {
+ pos = append(pos, s.score)
+ } else {
+ neg = append(neg, s.score)
+ }
+ }
+ posStats = scoreStats(pos)
+ negStats = scoreStats(neg)
+ return
+}
+
+// scoreStats returns min/max/mean/std for a slice of scores.
+func scoreStats(scores []float64) map[string]float64 {
+ if len(scores) == 0 {
+ return map[string]float64{"min": 0, "max": 0, "mean": 0, "std": 0}
+ }
+ min, max, sum := scores[0], scores[0], 0.0
+ for _, s := range scores {
+ if s < min {
+ min = s
+ }
+ if s > max {
+ max = s
+ }
+ sum += s
+ }
+ mean := sum / float64(len(scores))
+ variance := 0.0
+ for _, s := range scores {
+ variance += (s - mean) * (s - mean)
+ }
+ variance /= float64(len(scores))
+ return map[string]float64{
+ "min": min, "max": max, "mean": mean, "std": math.Sqrt(variance),
+ }
+}
+
+// trainReport is the validation report written to --report-out.
+type trainReport struct {
+ Scorer string `json:"scorer"`
+ Seed int64 `json:"seed"`
+ TrainSize int `json:"train_size"`
+ ValSize int `json:"val_size"`
+ TrainPos int `json:"train_pos"`
+ TrainNeg int `json:"train_neg"`
+ ValPos int `json:"val_pos"`
+ ValNeg int `json:"val_neg"`
+ TrainWindowMin time.Time `json:"train_window_min"`
+ TrainWindowMax time.Time `json:"train_window_max"`
+ ValFrac float64 `json:"val_frac"`
+ VocabSize int `json:"vocab_size"`
+ PrecisionAt15 float64 `json:"precision_at_15"`
+ YoudensJ float64 `json:"youdens_j"`
+ PositiveScores map[string]float64 `json:"positive_scores"`
+ NegativeScores map[string]float64 `json:"negative_scores"`
+ VectorizerParams map[string]any `json:"vectorizer_params"`
+ ModelParams map[string]any `json:"model_params"`
+}
diff --git a/cmds/train_test.go b/cmds/train_test.go
new file mode 100644
index 0000000..5a2ecee
--- /dev/null
+++ b/cmds/train_test.go
@@ -0,0 +1,260 @@
+// Tests for `fluxrec train`: reproducibility, metrics, and model envelope
+// consumption by the `score` command.
+package cmds
+
+import (
+ "bytes"
+ "encoding/json"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "fluxrec/core"
+)
+
+// syntheticLabels builds a labels.jsonl file interleaving positives and
+// negatives across time (so the newest validation window contains both
+// classes), with varied feed titles so the feed-identity token is exercised.
+// Timestamps are sequential by position: oldest first, newest last, so the
+// time split's "newest val-frac" is a clean tail slice with mixed labels.
+func syntheticLabels(t *testing.T, path string, nPos, nNeg int) {
+ t.Helper()
+ now := time.Now().UTC()
+ total := nPos + nNeg
+ rows := make([]core.LabelRow, total)
+ posTitles := []string{
+ "New evidence on the rate of productivity growth in services",
+ "Why headline-only models are fine actually, a reanalysis",
+ "Field notes from sixty days of careful RSS triage at home",
+ "The economics of attention and the case for reading slower",
+ "A long retrospective on building small personal recommendation tools",
+ "Decade-scale trends in academic publishing, a meta-review",
+ "How I learned to stop worrying and love the title-only classifier",
+ "Notes on serendipity, exploration budgets, and ranked recall",
+ "Reflections on a year of minimal-information inference methods",
+ "The quiet resurgence of logistic regression in small data settings",
+ }
+ negTitles := []string{
+ "Roundup of this week's product launches and minor updates",
+ "Local weather forecast for the coming weekend and beyond",
+ "Five ways to organize your desk for slightly better focus",
+ "Celebrity interview about their new streaming television show",
+ "Sports scores and highlights from last night's important games",
+ "Markets close mostly unchanged as investors await fresh data",
+ "Travel deals for the autumn season across several destinations",
+ "Cooking tips for weeknight dinners under thirty minutes flat",
+ "Opinion piece on city traffic and the eternal parking problem",
+ "Best of the app store this month, ten small utilities reviewed",
+ }
+ feeds := []string{"Marginal Revolution", "ACM Tech News", "Some Other Blog", "Hacker News", "The Atlantic"}
+ pi, ni := 0, 0
+ for i := 0; i < total; i++ {
+ // Place a positive on even indices while positives remain, so
+ // labels interleave across the whole time range including the tail.
+ label := core.LabelNeg
+ title := ""
+ if i%2 == 0 && pi < nPos {
+ label, title = core.LabelPos, posTitles[pi%len(posTitles)]
+ pi++
+ } else {
+ label, title = core.LabelNeg, negTitles[ni%len(negTitles)]
+ ni++
+ }
+ rows[i] = core.LabelRow{
+ EntryID: int64(i + 1), URL: "https://example.com/" + label + "/" + itoa(i),
+ Title: title, FeedID: int64(i%len(feeds) + 1), FeedTitle: feeds[i%len(feeds)],
+ Label: label, CapturedAt: now.Add(-time.Duration(total-i) * 24 * time.Hour),
+ }
+ }
+ if err := core.AppendLabels(path, rows); err != nil {
+ t.Fatalf("AppendLabels: %v", err)
+ }
+}
+
+func itoa(i int) string {
+ b := []byte{}
+ if i == 0 {
+ return "0"
+ }
+ for i > 0 {
+ b = append([]byte{byte('0' + i%10)}, b...)
+ i /= 10
+ }
+ return string(b)
+}
+
+func runTrain(t *testing.T, labelsPath, dir string) (modelPath, reportPath string) {
+ t.Helper()
+ modelPath = filepath.Join(dir, "model.json")
+ reportPath = filepath.Join(dir, "report.json")
+ cmd := &TrainCommand{}
+ if err := cmd.Init([]string{labelsPath, "--model-out", modelPath, "--report-out", reportPath, "--val-frac", "0.3"}); err != nil {
+ t.Fatalf("Init: %v", err)
+ }
+ var out bytes.Buffer
+ if err := cmd.Run(nil, &out); err != nil {
+ t.Fatalf("Run: %v\nstdout: %s", err, out.String())
+ }
+ return
+}
+
+// B4: same labels → byte-identical model.json (seeds make training deterministic).
+func TestTrainReproducible(t *testing.T) {
+ dir := t.TempDir()
+ labelsPath := filepath.Join(dir, "labels.jsonl")
+ // Enough rows that a 0.3 val split has >=5 positives in val.
+ syntheticLabels(t, labelsPath, 50, 50)
+
+ m1, r1 := runTrain(t, labelsPath, dir)
+ body1, err := os.ReadFile(m1)
+ if err != nil {
+ t.Fatalf("read model1: %v", err)
+ }
+ report1, err := os.ReadFile(r1)
+ if err != nil {
+ t.Fatalf("read report1: %v", err)
+ }
+
+ // Second train to a second pair of paths.
+ dir2 := t.TempDir()
+ m2, r2 := runTrain(t, labelsPath, dir2)
+ body2, err := os.ReadFile(m2)
+ if err != nil {
+ t.Fatalf("read model2: %v", err)
+ }
+ report2, err := os.ReadFile(r2)
+ if err != nil {
+ t.Fatalf("read report2: %v", err)
+ }
+
+ // Strip CreatedAt (wall clock) before comparing models — the rest must
+ // be byte-identical. Compare reports fully.
+ var env1, env2 core.ModelEnvelope
+ if err := json.Unmarshal(body1, &env1); err != nil {
+ t.Fatal(err)
+ }
+ if err := json.Unmarshal(body2, &env2); err != nil {
+ t.Fatal(err)
+ }
+ env1.CreatedAt = time.Time{}
+ env2.CreatedAt = time.Time{}
+ c1, _ := json.Marshal(env1)
+ c2, _ := json.Marshal(env2)
+ if string(c1) != string(c2) {
+ t.Fatal("model not reproducible: weights/vocab differ across runs with same labels")
+ }
+ if string(report1) != string(report2) {
+ t.Fatal("report not reproducible: differs across runs with same labels")
+ }
+}
+
+// B5: report.json parses and precision@15 is a valid probability.
+func TestTrainReportValid(t *testing.T) {
+ dir := t.TempDir()
+ labelsPath := filepath.Join(dir, "labels.jsonl")
+ syntheticLabels(t, labelsPath, 50, 50)
+
+ modelPath, reportPath := runTrain(t, labelsPath, dir)
+
+ var report trainReport
+ body, err := os.ReadFile(reportPath)
+ if err != nil {
+ t.Fatalf("read report: %v", err)
+ }
+ if err := json.Unmarshal(body, &report); err != nil {
+ t.Fatalf("report unparseable: %v", err)
+ }
+ if report.PrecisionAt15 < 0 || report.PrecisionAt15 > 1 {
+ t.Errorf("precision@15 = %v, want in [0,1]", report.PrecisionAt15)
+ }
+ if report.VocabSize <= 0 {
+ t.Errorf("vocab size = %d, want > 0", report.VocabSize)
+ }
+ if report.TrainPos == 0 || report.TrainNeg == 0 {
+ t.Errorf("train should have both classes: pos=%d neg=%d", report.TrainPos, report.TrainNeg)
+ }
+ if report.ValPos < 5 {
+ t.Errorf("val pos = %d, guard should have required >=5", report.ValPos)
+ }
+ if report.Seed != trainSeed {
+ t.Errorf("seed = %d, want %d", report.Seed, trainSeed)
+ }
+
+ // Model file exists and has non-empty vocab + weights.
+ mbody, err := os.ReadFile(modelPath)
+ if err != nil {
+ t.Fatalf("read model: %v", err)
+ }
+ var model core.ModelEnvelope
+ if err := json.Unmarshal(mbody, &model); err != nil {
+ t.Fatalf("model unparseable: %v", err)
+ }
+ if len(model.Weights) == 0 || len(model.OrderedVocab) == 0 {
+ t.Fatal("model missing weights or vocab")
+ }
+}
+
+// B6: the produced model.json is consumable by the score command's loader
+// (same ModelEnvelope type). Round-trips through CreateVectorizerFromModel.
+func TestTrainModelConsumedByScore(t *testing.T) {
+ dir := t.TempDir()
+ labelsPath := filepath.Join(dir, "labels.jsonl")
+ syntheticLabels(t, labelsPath, 50, 50)
+
+ modelPath, _ := runTrain(t, labelsPath, dir)
+
+ model, err := loadModel(modelPath) // same loader the score command uses
+ if err != nil {
+ t.Fatalf("loadModel (score's loader): %v", err)
+ }
+ vec := core.CreateVectorizerFromModel(model)
+ title := "A long retrospective on building small personal recommendation tools"
+ out := vec.Transform([]string{core.ScoringText(title, "Marginal Revolution")})
+ score, err := core.PredictScore(out[0], model.Weights)
+ if err != nil {
+ t.Fatalf("PredictScore: %v", err)
+ }
+ if score < 0 || score > 1 {
+ t.Errorf("score = %v, want in [0,1]", score)
+ }
+}
+
+// Guard: too few validation positives must error loudly.
+func TestTrainGuardFewValPositives(t *testing.T) {
+ dir := t.TempDir()
+ labelsPath := filepath.Join(dir, "labels.jsonl")
+ // 5 positives total, val-frac 0.3 → 1-2 val positives < 5.
+ syntheticLabels(t, labelsPath, 5, 20)
+
+ cmd := &TrainCommand{}
+ if err := cmd.Init([]string{labelsPath, "--val-frac", "0.3", "--model-out",
+ filepath.Join(dir, "m.json"), "--report-out", filepath.Join(dir, "r.json")}); err != nil {
+ t.Fatalf("Init: %v", err)
+ }
+ var out bytes.Buffer
+ err := cmd.Run(nil, &out)
+ if err == nil {
+ t.Fatal("expected error for <5 validation positives, got nil")
+ }
+ if !strings.Contains(err.Error(), "positives in validation") {
+ t.Errorf("error should mention validation positives: %v", err)
+ }
+}
+
+// Guard: empty labels file errors.
+func TestTrainEmptyLabels(t *testing.T) {
+ dir := t.TempDir()
+ labelsPath := filepath.Join(dir, "empty.jsonl")
+ if err := os.WriteFile(labelsPath, []byte{}, 0o644); err != nil {
+ t.Fatal(err)
+ }
+ cmd := &TrainCommand{}
+ if err := cmd.Init([]string{labelsPath}); err != nil {
+ t.Fatalf("Init: %v", err)
+ }
+ if err := cmd.Run(nil, &bytes.Buffer{}); err == nil {
+ t.Fatal("expected error for empty labels")
+ }
+}
generated by cgit v1.2.3 (git 2.46.0) at 2026-08-31 08:46:40 +0000