Merge the store sweep: the repeat-til-ack loop never took a first step (#260)

LastSent scanned MAX(sent_at) into a bare int64. MAX over an empty set
is one row holding NULL, so it errored where its own doc promised a zero
time. ack_sends is written only by MarkSent, which runs only after a
repeat has been sent, so the first repeat for every rule read an empty
table and RepeatUnacked returned on the error and aborted the whole
sweep. The sev4 repeat-til-ack loop could never take its first step for
any rule. nudges.go:213 documents this exact trap for MIN; ack.go never
got the same treatment.

EnqueueDigestEntry deduped on status='pending' alone. A row past its
expires_ts stays pending until the sweep marks it, and tick.go enqueues
before it sweeps, so on the tick after an expiry a suppressed nudge
deduped against a row PendingDigestEntries will never return, and the
phrasing already paid for was discarded. The read side already treated
not-yet-swept as not-deliverable; the write side did not. It also
treated any read error as no-row and inserted anyway.

ack.go had no test file at all. It has one now.

internal/memory was read and is clean, and every embedder call site
correctly passes EmbedPassage for a stored text.

(V-617)

internal/store/ack.go

@@ -2,6 +2,7 @@ package store
 
 import (
 	"context"
+	"database/sql"
 	"fmt"
 	"time"
 )
@@ -38,16 +39,21 @@ func (s *Store) MarkSent(ctx context.Context, key string, ts time.Time) error {
 // send goes through RecordNudge, not MarkSent, so the first MarkSent comes on
 // the repeat path — LastSent may legitimately be zero until then).
 func (s *Store) LastSent(ctx context.Context, key string) (time.Time, error) {
-	var millis int64
+	// MAX over an empty set is one row holding NULL, not zero rows, so this
+	// scans into a NullInt64 — the same trap OldestPendingTelegram spells out.
+	// A bare int64 turned the ordinary "nothing sent yet" case into a scan
+	// error, and RepeatUnacked aborts its whole sweep on one, so the first
+	// repeat could never go out for any rule.
+	var millis sql.NullInt64
 	err := s.db.QueryRowContext(ctx,
 		`SELECT MAX(sent_at) FROM ack_sends WHERE rule = ?`, key).Scan(&millis)
 	if err != nil {
 		return time.Time{}, fmt.Errorf("last sent %s: %w", key, err)
 	}
-	if millis == 0 {
+	if !millis.Valid {
 		return time.Time{}, nil
 	}
-	return time.UnixMilli(millis).UTC(), nil
+	return time.UnixMilli(millis.Int64).UTC(), nil
 }
 
 // MarkAcked marks ALL pending telegram nudges for the rule as "acted" —

internal/store/ack_test.go

@@ -0,0 +1,60 @@
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// TestLastSentOnEmptyTableIsZero pins the aggregate-over-nothing trap that
+// nudges.go's OldestPendingTelegram already documents: MAX over an empty set is
+// one row holding NULL, not zero rows. Scanning that into a bare int64 is an
+// error, and the doc on LastSent promises a zero time instead.
+//
+// It is not a cosmetic promise. ack_sends is written only by MarkSent, and
+// MarkSent is called only after a repeat has already gone out, so the first
+// repeat for every rule reads an empty table. delivery.Dispatcher.RepeatUnacked
+// aborts the whole sweep on that error, which means the repeat-til-ack loop can
+// never take its first step for any rule.
+func TestLastSentOnEmptyTableIsZero(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+
+	last, err := s.LastSent(ctx, "service_down")
+	if err != nil {
+		t.Fatalf("last sent on an empty table must not error: %v", err)
+	}
+	if !last.IsZero() {
+		t.Fatalf("want the zero time before anything was sent, got %v", last)
+	}
+}
+
+// TestLastSentIsScopedToItsRule — a send for another rule must not answer for
+// this one, or the repeat interval is clocked off somebody else's alarm.
+func TestLastSentIsScopedToItsRule(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	at := time.UnixMilli(1_700_000_000_000).UTC()
+
+	if err := s.MarkSent(ctx, "other_rule", at); err != nil {
+		t.Fatalf("mark sent: %v", err)
+	}
+	last, err := s.LastSent(ctx, "service_down")
+	if err != nil {
+		t.Fatalf("last sent: %v", err)
+	}
+	if !last.IsZero() {
+		t.Fatalf("want zero for a rule with no sends, got %v", last)
+	}
+
+	if err := s.MarkSent(ctx, "service_down", at); err != nil {
+		t.Fatalf("mark sent: %v", err)
+	}
+	last, err = s.LastSent(ctx, "service_down")
+	if err != nil {
+		t.Fatalf("last sent: %v", err)
+	}
+	if !last.Equal(at) {
+		t.Fatalf("want %v, got %v", at, last)
+	}
+}

internal/store/digest.go

@@ -3,7 +3,9 @@ package store
 import (
 	"context"
 	"crypto/sha256"
+	"database/sql"
 	"encoding/hex"
+	"errors"
 	"fmt"
 	"time"
 )
@@ -44,19 +46,33 @@ func DigestBodyHash(rule, body string) string {
 }
 
 // EnqueueDigestEntry durably records a suppressed care candidate worth
-// resurfacing later. If a pending entry with the same rule+body already
+// resurfacing later. If a LIVE pending entry with the same rule+body already
 // exists, this is a no-op that returns the existing id and deduped=true —
 // the same suppressed nudge repeating across ticks must not pile up into
 // several copies of itself in the eventual bundle.
+//
+// "Live" carries the same expiry test PendingDigestEntries reads with, and for
+// the same reason: a row past its expires_ts is still status='pending' until
+// the sweep gets to it, and the tick enqueues before it sweeps. Deduping
+// against one meant reporting deduped=true against an entry that will never be
+// spoken — the caller drops the phrasing it just paid the LLM for and nothing
+// reaches the bundle. Not yet swept must not mean still deliverable on the
+// write side either.
 func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity int, body string, now, expiresAt time.Time) (id int64, deduped bool, err error) {
 	hash := DigestBodyHash(rule, body)
 	var existing int64
 	err = s.db.QueryRowContext(ctx,
-		`SELECT id FROM digest_entries WHERE status = ? AND rule = ? AND body_hash = ? LIMIT 1`,
-		DigestPending, rule, hash).Scan(&existing)
+		`SELECT id FROM digest_entries
+		  WHERE status = ? AND rule = ? AND body_hash = ? AND expires_ts > ? LIMIT 1`,
+		DigestPending, rule, hash, now.UnixMilli()).Scan(&existing)
 	if err == nil {
 		return existing, true, nil
 	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		// A real read failure is not "nothing there". Inserting anyway would
+		// duplicate an entry whose existence we never established.
+		return 0, false, fmt.Errorf("enqueue digest entry: dedupe lookup: %w", err)
+	}
 
 	res, err := s.db.ExecContext(ctx,
 		`INSERT INTO digest_entries (rule, severity, body, body_hash, status, created_ts, expires_ts)

internal/store/digest_test.go

@@ -200,3 +200,46 @@ func TestDigestEntryDrainMarksDrainedNotDeleted(t *testing.T) {
 		}
 	}
 }
+
+// TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry — an entry past its
+// expires_ts is still status='pending' until the sweep runs, and the tick
+// enqueues before it sweeps. Deduping against one reports deduped=true for a
+// row PendingDigestEntries will never hand back, so the suppressed nudge is
+// dropped instead of held: the entry says the thing was recorded when nothing
+// was.
+func TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry(t *testing.T) {
+	s := newTestStore(t)
+	ctx := context.Background()
+	created := time.Now()
+	expiresAt := created.Add(time.Hour)
+
+	first, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", created, expiresAt)
+	if err != nil {
+		t.Fatalf("enqueue: %v", err)
+	}
+	if deduped {
+		t.Fatal("first enqueue must not report deduped")
+	}
+
+	// A tick after the expiry, with the sweep not yet run: the same suppressed
+	// nudge comes round again and must be recorded afresh.
+	after := expiresAt.Add(time.Minute)
+	second, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", after, after.Add(time.Hour))
+	if err != nil {
+		t.Fatalf("re-enqueue: %v", err)
+	}
+	if deduped {
+		t.Fatal("an expired entry must not swallow a fresh one")
+	}
+	if second == first {
+		t.Fatalf("want a new row, got the expired one back: id=%d", second)
+	}
+
+	entries, err := s.PendingDigestEntries(ctx, after)
+	if err != nil {
+		t.Fatalf("pending: %v", err)
+	}
+	if len(entries) != 1 || entries[0].ID != second {
+		t.Fatalf("want the fresh entry %d pending, got %+v", second, entries)
+	}
+}