export: per-feed negative ceiling (--neg-feed-cap-pct)

No feed may hold more than 5% (floor 50) of the negative target, counting
existing history: uncapped, a few firehoses own the negative class because
the read+unstarred pool is sized by feed volume. Enforced by skipping
ceiling-full feeds mid seeded permutation walk; shortfall vs the ratio is
reported, neg_capped added to the run summary. Also makes bulk mark-read
of high-volume feeds safe for training data.

4 files changed, 182 insertions, 21 deletions

diff --git a/cmds/export.go b/cmds/export.go
index 102c94a..203a7ef 100644
--- a/cmds/export.go
+++ b/cmds/export.go
@@ -22,14 +22,15 @@ import (
 )
 
 type ExportCommand struct {
-	DumpRaw   string
-	Labels    string
-	State     string
-	RunOut    string
-	RecFeedID int64
-	NegDays   int
-	NegRatio  int
-	NegSeed   int64
+	DumpRaw       string
+	Labels        string
+	State         string
+	RunOut        string
+	RecFeedID     int64
+	NegDays       int
+	NegRatio      int
+	NegSeed       int64
+	NegFeedCapPct int
 }
 
 func (c *ExportCommand) Name() string { return "export" }
@@ -43,7 +44,9 @@ func (c *ExportCommand) Init(args []string) error {
 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.
+No single feed may hold more than --neg-feed-cap-pct percent of the
+negative target (floored at 50 rows), so a few firehoses can't own the
+negative class. 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.
@@ -58,6 +61,7 @@ and write the raw API responses to DIR for offline development fixtures.
 	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)")
+	fs.IntVar(&c.NegFeedCapPct, "neg-feed-cap-pct", 5, "Per-feed negative ceiling as % of target (0 = uncapped; floor 50)")
 	return fs.Parse(args)
 }
 
@@ -202,11 +206,15 @@ func (c *ExportCommand) runExport(stdout io.Writer) error {
 	}
 
 	posTotal, negTotal := 0, 0
+	// negPerFeed is each feed's existing negative count: the cap is on the
+	// feed's total share of the pool, so history counts against it.
+	negPerFeed := make(map[int64]int)
 	for _, row := range existing {
 		if row.Label == core.LabelPos {
 			posTotal++
 		} else {
 			negTotal++
+			negPerFeed[row.FeedID]++
 		}
 	}
 	posTotal += counts.appended
@@ -218,7 +226,7 @@ func (c *ExportCommand) runExport(stdout io.Writer) error {
 		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)
+	nc, err := c.exportNegatives(client, seenIDs, seenURLs, negPerFeed, posTotal, negTotal, now, stdout)
 	if err != nil {
 		return err
 	}
@@ -226,14 +234,15 @@ func (c *ExportCommand) runExport(stdout io.Writer) error {
 		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,
+		NegTotal: negTotal + nc.appended, NegCapped: nc.capped,
+		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
+	candidates, appended, capped int
 }
 
 // exportNegatives samples read, unstarred entries published within the
@@ -242,17 +251,24 @@ type negCounts struct {
 // (either polarity) is excluded; sampling is seeded so the same candidate
 // pool + seed reproduce the same sample.
 //
+// negPerFeed carries each feed's existing negative count and is updated as
+// rows are added: the --neg-feed-cap-pct ceiling applies to a feed's total
+// pool share, so candidates from feeds already at their ceiling are skipped
+// and the budget flows to other feeds. A capped-out pool can therefore
+// finish short of the ratio target, which is reported, not hidden.
+//
 // 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) {
+func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int64]bool, seenURLs map[string]bool, negPerFeed map[int64]int, 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
 	}
+	cap := feedNegCap(c.NegFeedCapPct, c.NegRatio, posTotal)
 
 	cutoff := now.Add(-time.Duration(c.NegDays) * 24 * time.Hour)
 	unstarred := false
@@ -285,17 +301,27 @@ func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int
 	}
 	nc.candidates = len(cands)
 
+	// Seeded permutation walk: same candidate pool + seed = same sample.
+	// The ceiling filters the walk rather than reordering it, so enabling
+	// or disabling it can't perturb the sampled order of accepted rows.
 	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 := make([]core.LabelRow, 0, min(need, len(cands)))
+	for _, idx := range perm {
+		if len(rows) >= need {
+			break
+		}
+		e := cands[idx]
+		if cap > 0 && negPerFeed[e.Feed.ID] >= cap {
+			nc.capped++
+			continue
+		}
 		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,
 		})
+		negPerFeed[e.Feed.ID]++
 		seenIDs[e.ID] = true
 		seenURLs[core.NormalizeURL(e.URL)] = true
 	}
@@ -304,15 +330,23 @@ func (c *ExportCommand) exportNegatives(client *miniflux.Client, seenIDs map[int
 	}
 	nc.appended = len(rows)
 
-	fmt.Fprintf(stdout, "negatives: candidates=%d appended=%d (target=%d:%d)\n",
-		nc.candidates, nc.appended, c.NegRatio, posTotal)
+	fmt.Fprintf(stdout, "negatives: candidates=%d appended=%d capped=%d (target=%d:%d cap=%s)\n",
+		nc.candidates, nc.appended, nc.capped, c.NegRatio, posTotal, capLabel(cap))
 	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)
+		fmt.Fprintf(stdout, "note: negatives short by %d (candidates=%d in the last %d days, capped-out=%d)\n",
+			need-nc.appended, nc.candidates, c.NegDays, nc.capped)
 	}
 	return nc, nil
 }
 
+// capLabel renders a per-feed ceiling for the run line ("off" when uncapped).
+func capLabel(cap int) string {
+	if cap <= 0 {
+		return "off"
+	}
+	return fmt.Sprintf("%d/feed", cap)
+}
+
 // runSummary is the per-run JSON summary written for visibility; kept to
 // the latest run (overwrite, not append).
 type runSummary struct {
@@ -324,6 +358,7 @@ type runSummary struct {
 	NegCandidates int       `json:"neg_candidates"`
 	NegNew        int       `json:"neg_new"`
 	NegTotal      int       `json:"neg_total"`
+	NegCapped     int       `json:"neg_capped"`
 	IDDup         int       `json:"id_dup"`
 	URLDup        int       `json:"url_dup"`
 	RecHit        int       `json:"rec_hit"`
diff --git a/cmds/export_test.go b/cmds/export_test.go
index 0190d39..25eadbb 100644
--- a/cmds/export_test.go
+++ b/cmds/export_test.go
@@ -115,10 +115,13 @@ func runExportAgainst(t *testing.T, p *pagedServer, dir string, extra ...string)
 	t.Setenv("MINIFLUX_URL", srv.URL)
 	t.Setenv("MINIFLUX_TOKEN", "test-token")
 
+	// Ratio/dedup semantics are tested uncapped; the per-feed ceiling has
+	// its own dedicated test (TestExportNegativesFeedCap).
 	args := append([]string{
 		"--labels", filepath.Join(dir, "labels.jsonl"),
 		"--state", filepath.Join(dir, "export_state.json"),
 		"--run-out", filepath.Join(dir, "export_run.json"),
+		"--neg-feed-cap-pct", "0",
 	}, extra...)
 
 	cmd := &ExportCommand{}
@@ -347,6 +350,70 @@ func TestExportNegatives(t *testing.T) {
 	}
 }
 
+// The per-feed ceiling: one feed's negatives are bounded by the cap, the
+// budget then flows to other feeds, and the shortfall (vs. the ratio
+// target) is reported rather than silently padded.
+func TestExportNegativesFeedCap(t *testing.T) {
+	now := time.Now()
+	starred := makeEntries(10, 1000, true, 7)
+	for i := range starred {
+		starred[i].PublishedAt = now.Add(-24 * time.Hour)
+	}
+	// A big firehose pool and a small quiet feed's worth, both in-window.
+	big := makeEntries(500, 5000, false, 8)
+	small := makeEntries(30, 9000, false, 9)
+	for i := range big {
+		big[i].PublishedAt = now.Add(-10 * 24 * time.Hour)
+	}
+	for i := range small {
+		small[i].PublishedAt = now.Add(-10 * 24 * time.Hour)
+	}
+	p := &pagedServer{starred: starred, read: append(big, small...)}
+	dir := t.TempDir()
+
+	// 10 positives × ratio 20 = target 200 negatives. Ceiling = 5% × 200 =
+	// 10 → floor 50 wins; cap is 50/feed.
+	out := runExportAgainst(t, p, dir, "--neg-days", "100000", "--neg-feed-cap-pct", "5")
+
+	labels, err := core.ReadLabels(filepath.Join(dir, "labels.jsonl"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	perFeed := map[int64]int{}
+	negN := 0
+	for _, row := range labels {
+		if row.Label == core.LabelNeg {
+			perFeed[row.FeedID]++
+			negN++
+		}
+	}
+	if perFeed[8] != 50 {
+		t.Errorf("firehose feed should be capped at 50, got %d", perFeed[8])
+	}
+	if perFeed[9] != 30 {
+		t.Errorf("quiet feed should give all 30 of its candidates, got %d", perFeed[9])
+	}
+	if negN != 80 {
+		t.Errorf("total negatives = %d, want 80 (50 capped + 30 quiet)", negN)
+	}
+	if !strings.Contains(out, "capped=") || !strings.Contains(out, "negatives short by 120") {
+		t.Errorf("shortfall from capping should be reported:\n%s", out)
+	}
+	body, err := os.ReadFile(filepath.Join(dir, "export_run.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var rs runSummary
+	if err := json.Unmarshal(body, &rs); err != nil {
+		t.Fatal(err)
+	}
+	// All 500 firehose candidates walk the permutation; 50 are accepted,
+	// the other 450 are skipped by its ceiling (the quiet feed never caps).
+	if rs.NegCapped != 450 {
+		t.Errorf("neg_capped = %d, want 450", rs.NegCapped)
+	}
+}
+
 func TestExportNegativesSeededDeterministic(t *testing.T) {
 	dir1, dir2 := t.TempDir(), t.TempDir()
 	runExportAgainst(t, negFixture(4, 120, 0), dir1)
diff --git a/cmds/negcap.go b/cmds/negcap.go
new file mode 100644
index 0000000..f531b14
--- /dev/null
+++ b/cmds/negcap.go
@@ -0,0 +1,34 @@
+// Per-feed negative ceiling, shared by `export` (enforced while sampling)
+// and `compact-labels` (enforced against existing history).
+//
+// Why: negatives are sampled from the read+unstarred pool, whose size is
+// dominated by high-volume feeds. Without a ceiling, a handful of firehoses
+// own most of the negative class, and the model learns "wire headline" as
+// its main axis instead of "headline samsc liked". The ceiling is on each
+// feed's *total* negatives (existing history + this run's additions), not
+// just the run's sample: a per-run cap would only slow domination, not
+// bound it.
+package cmds
+
+// negFeedCapFloor is the absolute minimum ceiling. It keeps a young,
+// still-small negative pool from capping feeds at a handful of rows:
+// early training needs volume more than it needs balance.
+const negFeedCapFloor = 50
+
+// feedNegCap returns the maximum negatives any single feed may hold in the
+// pool: pct% of the target (negRatio × posTotal), floored at negFeedCapFloor
+// and never above the target itself. pct <= 0 means uncapped (0).
+func feedNegCap(pct, negRatio, posTotal int) int {
+	if pct <= 0 {
+		return 0
+	}
+	target := negRatio * posTotal
+	cap := pct * target / 100
+	if cap < negFeedCapFloor {
+		cap = negFeedCapFloor
+	}
+	if cap > target {
+		cap = target
+	}
+	return cap
+}
diff --git a/cmds/negcap_test.go b/cmds/negcap_test.go
new file mode 100644
index 0000000..5076163
--- /dev/null
+++ b/cmds/negcap_test.go
@@ -0,0 +1,25 @@
+// The shared per-feed negative ceiling: uncapped at pct 0, floored so a
+// young pool isn't strangled, and clamped to the target itself.
+package cmds
+
+import "testing"
+
+func TestFeedNegCap(t *testing.T) {
+	for _, tc := range []struct {
+		name                  string
+		pct, ratio, pos, want int
+	}{
+		{"uncapped at 0", 0, 20, 500, 0},
+		{"normal: 5% of 10000-target", 5, 20, 500, 500},
+		{"floor wins for young pools", 5, 20, 10, 50},    // target 200, 5% = 10 → floor 50
+		{"clamped to the target", 90, 20, 3, 54},         // target 60, 90% = 54
+		{"floor can exceed a tiny target", 5, 20, 1, 20}, // target 20 → clamp to 20
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := feedNegCap(tc.pct, tc.ratio, tc.pos); got != tc.want {
+				t.Errorf("feedNegCap(%d, %d, %d) = %d, want %d",
+					tc.pct, tc.ratio, tc.pos, got, tc.want)
+			}
+		})
+	}
+}