compact-labels: rebalance negative history to the per-feed ceiling

The single documented exception to labels.jsonl's append-only rule:
trims feeds over the ceiling (keeping each one's NEWEST negatives),
positives untouched, LABELS.bak written first, idempotent, --dry-run.
Existence justified by the ceiling only bounding the future, while
pre-cap history would otherwise never age out.

2 files changed, 295 insertions, 0 deletions

diff --git a/cmds/compact.go b/cmds/compact.go
new file mode 100644
index 0000000..2fc8b6f
--- /dev/null
+++ b/cmds/compact.go
@@ -0,0 +1,164 @@
+// Compact-labels command: one-time (but safely re-runnable) rebalance of
+// labels.jsonl against the same per-feed negative ceiling `export` now
+// enforces. Existing labels were written before the ceiling existed, so a
+// handful of firehoses can already own a large share of the negative class;
+// this trims them back.
+//
+// This is the one deliberate exception to labels.jsonl's append-only rule
+// (see core/labels.go). It exists because the ceiling only bounds the
+// future: without a compaction, pre-cap history keeps distorting training
+// until it ages out — which it never does. Rules that keep it honest:
+// positives are never touched; a feed's newest negatives are the ones kept
+// (recency carries the signal); the original file is copied to a backup
+// before anything is rewritten; and running it twice is a no-op.
+package cmds
+
+import (
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"os"
+	"sort"
+
+	"fluxrec/core"
+)
+
+type CompactLabelsCommand struct {
+	Labels        string
+	NegRatio      int
+	NegFeedCapPct int
+	Backup        string
+	DryRun        bool
+}
+
+func (c *CompactLabelsCommand) Name() string { return "compact-labels" }
+
+func (c *CompactLabelsCommand) Init(args []string) error {
+	fs := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	fs.Usage = func() {
+		fmt.Fprint(fs.Output(), `Usage: fluxrec compact-labels [flags]
+
+Trim per-feed negative over-representation in labels.jsonl down to the
+same ceiling export enforces (--neg-feed-cap-pct of the target, floor 50).
+Positives are untouched; each trimmed feed keeps its NEWEST negatives; the
+original is copied to --backup first; re-running is a no-op. Use --dry-run
+to see the effect without writing anything.
+`)
+		fs.PrintDefaults()
+	}
+	fs.StringVar(&c.Labels, "labels", "labels.jsonl", "Path to labels.jsonl")
+	fs.IntVar(&c.NegRatio, "neg-ratio", 20, "Negatives-per-positive ratio the ceiling is computed against")
+	fs.IntVar(&c.NegFeedCapPct, "neg-feed-cap-pct", 5, "Per-feed negative ceiling as % of target (floor 50)")
+	fs.StringVar(&c.Backup, "backup", "", "Backup path for the original file (default: LABELS.bak)")
+	fs.BoolVar(&c.DryRun, "dry-run", false, "Report what would change without writing")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	if c.NegRatio < 1 {
+		return fmt.Errorf("--neg-ratio must be >= 1, got %d", c.NegRatio)
+	}
+	if c.NegFeedCapPct < 1 {
+		return fmt.Errorf("--neg-feed-cap-pct must be >= 1, got %d", c.NegFeedCapPct)
+	}
+	return nil
+}
+
+func (c *CompactLabelsCommand) Run(stdin io.Reader, stdout io.Writer) error {
+	rows, err := core.ReadLabels(c.Labels)
+	if err != nil {
+		return err
+	}
+	if len(rows) == 0 {
+		fmt.Fprintf(stdout, "compact-labels: %s is empty or missing; nothing to do\n", c.Labels)
+		return nil
+	}
+
+	posTotal := 0
+	for _, row := range rows {
+		if row.Label == core.LabelPos {
+			posTotal++
+		}
+	}
+	cap := feedNegCap(c.NegFeedCapPct, c.NegRatio, posTotal)
+	if cap <= 0 {
+		return fmt.Errorf("computed per-feed ceiling is 0 (positives=%d); check --neg-ratio/--neg-feed-cap-pct", posTotal)
+	}
+
+	// Per-feed negative row indexes, to decide what each overfull feed keeps.
+	negIdx := make(map[int64][]int)
+	for i, row := range rows {
+		if row.Label == core.LabelNeg {
+			negIdx[row.FeedID] = append(negIdx[row.FeedID], i)
+		}
+	}
+
+	drop := make(map[int]bool)
+	trimmed := 0
+	for _, idxs := range negIdx {
+		if len(idxs) <= cap {
+			continue
+		}
+		trimmed++
+		// Keep the newest `cap` rows for this feed; newest-first sort with a
+		// deterministic tiebreak, so repeated runs agree.
+		sorted := make([]int, len(idxs))
+		copy(sorted, idxs)
+		sort.Slice(sorted, func(a, b int) bool {
+			ra, rb := rows[sorted[a]], rows[sorted[b]]
+			if ra.CapturedAt.Equal(rb.CapturedAt) {
+				return ra.EntryID > rb.EntryID
+			}
+			return ra.CapturedAt.After(rb.CapturedAt)
+		})
+		for _, idx := range sorted[cap:] {
+			drop[idx] = true
+		}
+	}
+
+	kept := make([]core.LabelRow, 0, len(rows)-len(drop))
+	for i, row := range rows {
+		if !drop[i] {
+			kept = append(kept, row)
+		}
+	}
+
+	negTotal := len(rows) - posTotal
+	fmt.Fprintf(stdout, "compact-labels: rows %d → %d (positives %d kept; negatives %d → %d; feeds trimmed %d; cap %d/feed)\n",
+		len(rows), len(kept), posTotal, negTotal, negTotal-len(drop), trimmed, cap)
+	if len(drop) == 0 {
+		fmt.Fprintln(stdout, "nothing over the ceiling; labels.jsonl unchanged")
+		return nil
+	}
+	if c.DryRun {
+		fmt.Fprintf(stdout, "dry-run: would drop %d negatives (no files written)\n", len(drop))
+		return nil
+	}
+
+	backup := c.Backup
+	if backup == "" {
+		backup = c.Labels + ".bak"
+	}
+	orig, err := os.ReadFile(c.Labels)
+	if err != nil {
+		return fmt.Errorf("reading %s for backup: %w", c.Labels, err)
+	}
+	if err := os.WriteFile(backup, orig, 0o644); err != nil {
+		return fmt.Errorf("writing backup %s: %w", backup, err)
+	}
+
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	for _, row := range kept {
+		if err := enc.Encode(row); err != nil {
+			return fmt.Errorf("encoding label row: %w", err)
+		}
+	}
+	if err := os.WriteFile(c.Labels, buf.Bytes(), 0o644); err != nil {
+		return fmt.Errorf("rewriting %s: %w", c.Labels, err)
+	}
+	fmt.Fprintf(stdout, "wrote %s (backup: %s); re-running export will sample fresh negatives under the ceiling\n",
+		c.Labels, backup)
+	return nil
+}
diff --git a/cmds/compact_test.go b/cmds/compact_test.go
new file mode 100644
index 0000000..c525fd3
--- /dev/null
+++ b/cmds/compact_test.go
@@ -0,0 +1,131 @@
+// `fluxrec compact-labels` against a fixture file: overfull feeds are
+// trimmed to the ceiling keeping their newest negatives, positives are
+// untouched, a backup is written first, and a second run is a no-op.
+package cmds
+
+import (
+	"bytes"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"fluxrec/core"
+)
+
+func compactFixture(t *testing.T, dir string) (string, time.Time) {
+	t.Helper()
+	now := time.Date(2024, 3, 1, 0, 0, 0, 0, time.UTC)
+	rows := []core.LabelRow{}
+	// 20 positives (feed 7) → target neg pool 400 → cap = 20/feed (5%).
+	for i := 0; i < 20; i++ {
+		rows = append(rows, core.LabelRow{
+			EntryID: int64(100 + i), URL: "https://pos.example/p", Title: "P",
+			FeedID: 7, FeedTitle: "Good Feed", Label: core.LabelPos,
+			CapturedAt: now.Add(time.Duration(i) * time.Hour),
+		})
+	}
+	// Firehose: 80 negatives, staggered times so "newest kept" is testable.
+	for i := 0; i < 80; i++ {
+		rows = append(rows, core.LabelRow{
+			EntryID: int64(5000 + i), URL: "https://fire.example/f", Title: "F",
+			FeedID: 8, FeedTitle: "Firehose", Label: core.LabelNeg,
+			CapturedAt: now.Add(time.Duration(i) * time.Hour),
+		})
+	}
+	// Quiet feed: 5 negatives, under the cap — must survive untouched.
+	for i := 0; i < 5; i++ {
+		rows = append(rows, core.LabelRow{
+			EntryID: int64(9000 + i), URL: "https://quiet.example/q", Title: "Q",
+			FeedID: 9, FeedTitle: "Quiet", Label: core.LabelNeg,
+			CapturedAt: now.Add(time.Duration(i) * time.Hour),
+		})
+	}
+	path := filepath.Join(dir, "labels.jsonl")
+	if err := core.AppendLabels(path, rows); err != nil {
+		t.Fatal(err)
+	}
+	return path, now
+}
+
+func TestCompactLabels(t *testing.T) {
+	dir := t.TempDir()
+	path, _ := compactFixture(t, dir)
+
+	// 20 positives → target neg pool 400 → 5% of 400 = 20 < floor 50 →
+	// the ceiling is 50/feed (the floor doing its job for a young pool).
+	cmd := &CompactLabelsCommand{Labels: path, NegRatio: 20, NegFeedCapPct: 5}
+	var out bytes.Buffer
+	if err := cmd.Run(nil, &out); err != nil {
+		t.Fatalf("compact-labels: %v", err)
+	}
+	if !strings.Contains(out.String(), "rows 105 → 75") {
+		t.Errorf("unexpected summary:\n%s", out.String())
+	}
+
+	rows, err := core.ReadLabels(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pos, perFeedNeg := 0, map[int64]int{}
+	keptIDs := map[int64]bool{}
+	for _, row := range rows {
+		if row.Label == core.LabelPos {
+			pos++
+		} else {
+			perFeedNeg[row.FeedID]++
+			keptIDs[row.EntryID] = true
+		}
+	}
+	if pos != 20 {
+		t.Errorf("positives changed: %d, want 20", pos)
+	}
+	if perFeedNeg[8] != 50 {
+		t.Errorf("firehose negatives = %d, want 50 (floor cap)", perFeedNeg[8])
+	}
+	if perFeedNeg[9] != 5 {
+		t.Errorf("quiet feed negatives = %d, want 5 (untouched)", perFeedNeg[9])
+	}
+	// Newest 50 of the firehose's negatives are the ones kept (IDs 5030-5079).
+	if !keptIDs[5079] || keptIDs[5029] {
+		t.Errorf("kept set should be the newest 50 (5030-5079); 5079 kept=%v 5029 kept=%v",
+			keptIDs[5079], keptIDs[5029])
+	}
+
+	// Backup holds the original 105 rows.
+	backup, err := core.ReadLabels(path + ".bak")
+	if err != nil || len(backup) != 105 {
+		t.Errorf("backup should hold the original 105 rows (got %d, err %v)", len(backup), err)
+	}
+
+	// Second run: nothing over the ceiling → no-op, and no backup rewrite.
+	var out2 bytes.Buffer
+	if err := cmd.Run(nil, &out2); err != nil {
+		t.Fatalf("second run: %v", err)
+	}
+	if !strings.Contains(out2.String(), "unchanged") {
+		t.Errorf("second run should report no change:\n%s", out2.String())
+	}
+}
+
+func TestCompactLabelsDryRun(t *testing.T) {
+	dir := t.TempDir()
+	path, _ := compactFixture(t, dir)
+
+	cmd := &CompactLabelsCommand{Labels: path, NegRatio: 20, NegFeedCapPct: 5, DryRun: true}
+	var out bytes.Buffer
+	if err := cmd.Run(nil, &out); err != nil {
+		t.Fatalf("dry-run: %v", err)
+	}
+	if !strings.Contains(out.String(), "dry-run") {
+		t.Errorf("dry-run note missing:\n%s", out.String())
+	}
+	rows, err := core.ReadLabels(path)
+	if err != nil || len(rows) != 105 {
+		t.Errorf("dry-run must not modify the file (rows=%d, err=%v)", len(rows), err)
+	}
+	if _, err := os.Stat(path + ".bak"); !os.IsNotExist(err) {
+		t.Error("dry-run must not write a backup")
+	}
+}