diff options
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | README.md | 1 | ||||
| -rw-r--r-- | cmds/compact.go | 164 | ||||
| -rw-r--r-- | cmds/compact_test.go | 131 | ||||
| -rw-r--r-- | main.go | 3 |
5 files changed, 300 insertions, 0 deletions
@@ -3,6 +3,7 @@ fluxrec # Data artifacts (plumbing, not source) labels.jsonl +labels.jsonl.bak model.json cursor.json runs.jsonl @@ -20,6 +20,7 @@ fluxrec train labels.jsonl → model.json + report.json (local) fluxrec serve poll, score, RSS at /recommendations.xml (server) fluxrec retrostar model-ranked candidates to star by hand (cold start) fluxrec feeds per-feed volume + star stats (pruning aid) +fluxrec compact-labels trim per-feed negative skew (one-off, re-runnable) fluxrec score debug: stdin titles → stdout scores ``` 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") + } +} @@ -40,6 +40,8 @@ func main() { cmd = &cmds.RetrostarCommand{} case "feeds": cmd = &cmds.FeedsCommand{} + case "compact-labels": + cmd = &cmds.CompactLabelsCommand{} default: fmt.Fprintf(os.Stderr, "Unknown command: %s\n\n", cmdName) printHelp() @@ -72,6 +74,7 @@ Commands: serve Poll Miniflux, score, serve recommendations RSS (server) retrostar Model-ranked candidates for retroactive starring (cold start) feeds Subscription audit: per-feed volume + star stats (pruning aid) + compact-labels Trim per-feed negative over-representation in labels.jsonl Usage: echo "some title" | fluxrec score --model model.json |