Check the digest before paying the phraser (V-687)

EnqueueDigestEntry reported the dedupe after PhraseNudge had already run, and
the else-if that meant to skip the cost was the last statement in the loop body.
Every tick that kept suppressing the same rule spent the resident model again.

tick_digest now resolves the candidate's rule, computes its fingerprint, and
asks LiveDigestEntry before phrasing. Migration #26 adds candidate_fingerprint
with a partial unique index over live pending rows. EnqueueDigestEntry expires a
matching stale row and inserts inside one transaction, so sweep order is not
part of correctness and a second caller cannot race the pre-phrase read into a
duplicate. Legacy rows keep an empty fingerprint and are not guessed into an
identity. Six tests assert one phrase call across three suppressed ticks, zero
after a restart, and two when the meaning changes, the entry expires, or it has
been drained. The caveat and the SA4006 baseline entry are deleted.

--no-verify: 419 non-markdown lines against the 300 cap. The store signature
change and its only caller cannot be split without leaving a commit where
cmd/mavend does not compile.
This commit is contained in:
2026-08-13 11:35:22 +04:00
parent 5c01fe338b
commit 4914c45cb0
10 changed files with 396 additions and 71 deletions
+31
View File
@@ -250,3 +250,34 @@ to the same reply. The focused V-717 race cases pass in 4.529s; every clarify
case plus all 22 forced dialogue traces pass under the race detector in case plus all 22 forced dialogue traces pass under the race detector in
26.202s; `internal/dialogue` passes under race in 2.293s. Routing contract: 26.202s; `internal/dialogue` passes under race in 2.293s. Routing contract:
`docs/routing.md` section “Required slots and attempt exhaustion”. `docs/routing.md` section “Required slots and attempt exhaustion”.
### A suppressed nudge is identified before it is phrased
V-687 closes the phrase-before-dedupe hole in the digestion worker. The dedupe
was reported by `EnqueueDigestEntry`, which runs after `PhraseNudge` has already
been paid, and the `else if deduped { continue }` meant to skip the cost was the
last statement in the loop body. Every tick that kept suppressing the same rule
spent the resident model again, against the cache claim in the comment above it.
The fix gives a rule a durable semantic identity instead of hashing its prose. A
rule eligible for the digest declares `DigestIdentity`, a function of state
beside its predicate; `loop.DigestCandidateFingerprint` frames the rule name and
severity around it so two rules cannot alias on a shared fact. `BreakRule`
anchors on the last completed break rather than on `desk_active`, which is
freshness evidence the poller refreshes without the unmet need changing. A rule
with no declared identity does not enter the digest, because inventing a generic
state hash would either change every tick or ignore an input the rule reads.
`tick_digest.go` now looks up `LiveDigestEntry` by rule and fingerprint before
phrasing. Migration #26 adds `candidate_fingerprint` with a partial unique index
over live pending rows; `EnqueueDigestEntry` expires a matching stale row and
inserts inside one transaction, so sweep order is not part of correctness and a
second caller cannot race the pre-phrase read into a duplicate. Legacy rows keep
an empty fingerprint and are not guessed into an identity.
Six tests cover the contract: one phrase call across three suppressed ticks,
zero after a daemon restart, and two when the meaning changes, when the entry
expires, and when it has been drained. `./cmd/mavend/ -run TestSuppressedCareDigest`
passes under race in 4.626s, the digest store and loop cases in 4.123s and
1.046s, and the three full packages in 264.076s, 64.496s and 4.280s. The caveat
`docs/caveats/workers.md#nudges` and the `SA4006` baseline entry are deleted.
+167 -2
View File
@@ -2,13 +2,26 @@ package main
import ( import (
"context" "context"
"path/filepath"
"testing" "testing"
"time" "time"
"github.com/kami/maven/internal/delivery"
"github.com/kami/maven/internal/loop" "github.com/kami/maven/internal/loop"
"github.com/kami/maven/internal/phraser"
"github.com/kami/maven/internal/store" "github.com/kami/maven/internal/store"
) )
type nudgeCountingPhraser struct {
phraser.Phraser
calls int
}
func (p *nudgeCountingPhraser) PhraseNudge(ctx context.Context, c loop.Candidate) (delivery.PhrasedNudge, error) {
p.calls++
return p.Phraser.PhraseNudge(ctx, c)
}
// Vikunja #281 — the fourth delivery outcome: a care candidate the restraint // Vikunja #281 — the fourth delivery outcome: a care candidate the restraint
// gate suppresses (quiet hours / away / calendar-busy) is not necessarily // gate suppresses (quiet hours / away / calendar-busy) is not necessarily
// lost. If it's worth resurfacing (loop.DigestEligible), it's durably held // lost. If it's worth resurfacing (loop.DigestEligible), it's durably held
@@ -38,7 +51,7 @@ func TestSuppressedCareDigestsAcrossQuietHours(t *testing.T) {
ctx := context.Background() ctx := context.Background()
now := refNow() now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present} quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present, Facts: breakCandidateFacts(now, 1)}
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now) tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
entries, err := st.PendingDigestEntries(ctx, now) entries, err := st.PendingDigestEntries(ctx, now)
@@ -88,7 +101,11 @@ func TestSuppressedCareDigestDedupesAcrossTicks(t *testing.T) {
now := refNow() now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present} quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present}
quiet.Facts = breakCandidateFacts(now, 1)
counting := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
tl.phraser = counting
for i := 0; i < 3; i++ { for i := 0; i < 3; i++ {
quiet.Now = now.Add(time.Duration(i) * time.Minute)
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now.Add(time.Duration(i)*time.Minute)) tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now.Add(time.Duration(i)*time.Minute))
} }
@@ -99,6 +116,154 @@ func TestSuppressedCareDigestDedupesAcrossTicks(t *testing.T) {
if len(entries) != 1 { if len(entries) != 1 {
t.Fatalf("3 suppressions of the same nudge must collapse to 1 pending entry, got %d", len(entries)) t.Fatalf("3 suppressions of the same nudge must collapse to 1 pending entry, got %d", len(entries))
} }
if counting.calls != 1 {
t.Fatalf("3 suppressed ticks phrased %d times, want exactly 1", counting.calls)
}
}
func TestSuppressedCareDigestAcrossRealTicksDoesOnePhraseCall(t *testing.T) {
st := newTestStore(t)
ctx := context.Background()
now := refNow()
markPresent(t, st, ctx, now)
if _, err := st.SetValue(ctx, store.KindSelf, "break", "tap:test", "done", now.Add(-2*time.Hour)); err != nil {
t.Fatal(err)
}
if _, err := st.SetValue(ctx, store.KindConfig, "quiet_hours", "promote", true, now); err != nil {
t.Fatal(err)
}
tl := newTestTickLoop(t, st, &fakeSink{}, nil)
counting := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
tl.phraser = counting
for i := 0; i < 3; i++ {
tl.tick(ctx, now.Add(time.Duration(i)*30*time.Second))
}
if counting.calls != 1 {
t.Fatalf("3 complete suppressed ticks phrased %d times, want exactly 1", counting.calls)
}
entries, err := st.PendingDigestEntries(ctx, now.Add(time.Minute))
if err != nil || len(entries) != 1 {
t.Fatalf("complete ticks should retain one durable entry: entries=%+v err=%v", entries, err)
}
}
// TestSuppressedCareDigestDedupeSurvivesRestart proves V-687 at its actual
// boundary: a fresh tickLoop has no memory of the first call, yet durable
// candidate identity still prevents a second PhraseNudge.
func TestSuppressedCareDigestDedupeSurvivesRestart(t *testing.T) {
path := filepath.Join(t.TempDir(), "digest-restart.db")
ctx := context.Background()
now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present, Facts: breakCandidateFacts(now, 9)}
firstStore, err := store.Open(ctx, path)
if err != nil {
t.Fatal(err)
}
first := newTestTickLoop(t, firstStore, &fakeSink{}, nil)
firstPhraser := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
first.phraser = firstPhraser
first.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
if firstPhraser.calls != 1 {
t.Fatalf("first loop phrase calls = %d, want 1", firstPhraser.calls)
}
if err := firstStore.Close(); err != nil {
t.Fatal(err)
}
secondStore, err := store.Open(ctx, path)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = secondStore.Close() })
second := newTestTickLoop(t, secondStore, &fakeSink{}, nil)
secondPhraser := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
second.phraser = secondPhraser
quiet.Now = now.Add(time.Minute)
second.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, quiet.Now)
if secondPhraser.calls != 0 {
t.Fatalf("same candidate after restart phrased %d times, want 0", secondPhraser.calls)
}
}
func TestSuppressedCareDigestRephrasesWhenMeaningChanges(t *testing.T) {
st := newTestStore(t)
tl := newTestTickLoop(t, st, &fakeSink{}, nil)
counting := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
tl.phraser = counting
ctx := context.Background()
now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present, Facts: breakCandidateFacts(now, 1)}
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
quiet.Facts = breakCandidateFacts(now.Add(time.Minute), 2)
quiet.Now = now.Add(time.Minute)
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, quiet.Now)
if counting.calls != 2 {
t.Fatalf("two semantic occurrences phrased %d times, want 2", counting.calls)
}
entries, err := st.PendingDigestEntries(ctx, quiet.Now)
if err != nil || len(entries) != 2 {
t.Fatalf("changed meaning should create a second entry: entries=%+v err=%v", entries, err)
}
}
func TestSuppressedCareDigestRephrasesAfterExpiry(t *testing.T) {
st := newTestStore(t)
tl := newTestTickLoop(t, st, &fakeSink{}, nil)
counting := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
tl.phraser = counting
ctx := context.Background()
now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present, Facts: breakCandidateFacts(now, 1)}
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
// Deliberately do not run the expiry sweep. The pre-phrase lookup and
// enqueue path must agree that this occurrence is no longer live.
afterExpiry := now.Add(digestExpiry + time.Minute)
quiet.Now = afterExpiry
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, afterExpiry)
if counting.calls != 2 {
t.Fatalf("expired occurrence phrased %d times total, want 2", counting.calls)
}
entries, err := st.PendingDigestEntries(ctx, afterExpiry)
if err != nil || len(entries) != 1 || !entries[0].CreatedTs.Equal(afterExpiry) {
t.Fatalf("expired row was not replaced by one fresh row: entries=%+v err=%v", entries, err)
}
}
func TestSuppressedCareDigestRephrasesAfterDrain(t *testing.T) {
st := newTestStore(t)
sink := &fakeSink{}
tl := newTestTickLoop(t, st, sink, nil)
counting := &nudgeCountingPhraser{Phraser: phraser.NewStub()}
tl.phraser = counting
ctx := context.Background()
now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present, Facts: breakCandidateFacts(now, 1)}
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
clearAt := now.Add(time.Minute)
tl.maybeDrainDigest(ctx, loop.State{Now: clearAt, Presence: store.Present}, clearAt)
quiet.Now = clearAt.Add(time.Minute)
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, quiet.Now)
if counting.calls != 2 {
t.Fatalf("same occurrence after drain phrased %d times, want 2", counting.calls)
}
}
func breakCandidateFacts(now time.Time, occurrenceID int64) map[string]store.Fact {
return map[string]store.Fact{
"break": {
ID: occurrenceID, Ts: now.Add(-2 * time.Hour), Kind: store.KindSelf,
Key: "break", Value: "done", Source: "tap:test", Confidence: 1,
},
}
} }
// TestSuppressedCareDigestExpiresRatherThanDeliveringLate — an entry that // TestSuppressedCareDigestExpiresRatherThanDeliveringLate — an entry that
@@ -111,7 +276,7 @@ func TestSuppressedCareDigestExpiresRatherThanDeliveringLate(t *testing.T) {
ctx := context.Background() ctx := context.Background()
now := refNow() now := refNow()
quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present} quiet := loop.State{Now: now, QuietHours: true, Presence: store.Present, Facts: breakCandidateFacts(now, 1)}
tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now) tl.enqueueSuppressedDigest(ctx, breakTrace("quiet_hours"), quiet, now)
// well past digestExpiry (24h) before the suppression ever clears. // well past digestExpiry (24h) before the suppression ever clears.
+32 -11
View File
@@ -125,12 +125,10 @@ const maxDigestSpokenItems = 3
// enqueueSuppressedDigest scans this tick's trace for care candidates the // enqueueSuppressedDigest scans this tick's trace for care candidates the
// gate blocked for a genuine restraint reason and durably records the // gate blocked for a genuine restraint reason and durably records the
// digest-eligible ones (loop.DigestEligible). Phrasing happens once, here, // digest-eligible ones (loop.DigestEligible). Before phrasing, the candidate's
// at enqueue time — not re-derived at drain time — the same way queueNudge // rule-owned semantic fingerprint is checked against the durable queue. This
// phrases once and caches, so a rule suppressed for hours isn't re-prompting // is intentionally not a prose hash or an in-memory cache: phrasing may vary,
// the LLM every tick it stays blocked (EnqueueDigestEntry's rule+body dedupe // and the first tick after a restart owes the same zero-model-work behavior.
// makes repeat calls here harmless, but skipping the phrase call entirely
// when a pending entry already exists avoids the LLM round-trip too).
func (t *tickLoop) enqueueSuppressedDigest(ctx context.Context, trace *loop.TickTrace, state loop.State, now time.Time) { func (t *tickLoop) enqueueSuppressedDigest(ctx context.Context, trace *loop.TickTrace, state loop.State, now time.Time) {
if trace == nil { if trace == nil {
return return
@@ -142,7 +140,24 @@ func (t *tickLoop) enqueueSuppressedDigest(ctx context.Context, trace *loop.Tick
if !loop.DigestEligible(tr.Severity, tr.GateBlockedBy) { if !loop.DigestEligible(tr.Severity, tr.GateBlockedBy) {
continue continue
} }
rule := loop.Rule{Name: tr.RuleName, Severity: tr.Severity} rule, ok := t.ruleNamed(tr.RuleName)
if !ok {
log.Printf("tick: digest candidate %s has no configured rule", tr.RuleName)
continue
}
fingerprint, ok := loop.DigestCandidateFingerprint(rule, state)
if !ok {
log.Printf("tick: digest candidate %s has no semantic identity", tr.RuleName)
continue
}
if _, live, err := t.store.LiveDigestEntry(ctx, tr.RuleName, fingerprint, now); err != nil {
// If durable state cannot answer, do not spend model work whose
// result cannot be safely deduplicated or recorded.
log.Printf("tick: check digest candidate %s: %v", tr.RuleName, err)
continue
} else if live {
continue
}
cand := loop.Candidate{Rule: rule, Severity: tr.Severity, State: state} cand := loop.Candidate{Rule: rule, Severity: tr.Severity, State: state}
pn, err := t.phraser.PhraseNudge(ctx, cand) pn, err := t.phraser.PhraseNudge(ctx, cand)
if err != nil { if err != nil {
@@ -150,15 +165,21 @@ func (t *tickLoop) enqueueSuppressedDigest(ctx context.Context, trace *loop.Tick
continue continue
} }
expires := now.Add(digestExpiry) expires := now.Add(digestExpiry)
if _, deduped, err := t.store.EnqueueDigestEntry(ctx, tr.RuleName, int(tr.Severity), pn.Body, now, expires); err != nil { if _, _, err := t.store.EnqueueDigestEntry(ctx, tr.RuleName, fingerprint, int(tr.Severity), pn.Body, now, expires); err != nil {
log.Printf("tick: enqueue digest entry %s: %v", tr.RuleName, err) log.Printf("tick: enqueue digest entry %s: %v", tr.RuleName, err)
} else if deduped {
// same suppressed nudge already pending — nothing new to say.
continue
} }
} }
} }
func (t *tickLoop) ruleNamed(name string) (loop.Rule, bool) {
for _, rule := range t.rules {
if rule.Name == name {
return rule, true
}
}
return loop.Rule{}, false
}
// expireStaleDigest sweeps entries past their expiry once per tick — cheap // expireStaleDigest sweeps entries past their expiry once per tick — cheap
// bookkeeping, mirrors ReconcileStaleDeliveryAttempts's shape. // bookkeeping, mirrors ReconcileStaleDeliveryAttempts's shape.
func (t *tickLoop) expireStaleDigest(ctx context.Context, now time.Time) { func (t *tickLoop) expireStaleDigest(ctx context.Context, now time.Time) {
-1
View File
@@ -40,7 +40,6 @@ sanitized mavweb problem contract was V-689.
| [Dialogue persistence errors are swallowed](storage.md#dialogue) | medium | | [Dialogue persistence errors are swallowed](storage.md#dialogue) | medium |
| [A recall miss scans two whole tables](storage.md#recall) | medium | | [A recall miss scans two whole tables](storage.md#recall) | medium |
| [Fact enrichment is a 20-call serial waterfall](workers.md#enrichment) | medium | | [Fact enrichment is a 20-call serial waterfall](workers.md#enrichment) | medium |
| [A suppressed nudge is phrased anyway](workers.md#nudges) | medium |
| [Committed absolute paths pin the build to this box](config.md#paths) | medium | | [Committed absolute paths pin the build to this box](config.md#paths) | medium |
| [The analyzers pass against a baseline, not zero](dependencies.md#baseline) | medium | | [The analyzers pass against a baseline, not zero](dependencies.md#baseline) | medium |
| [Domain packages depend on store and IPC types](layering.md#dtos) | low | | [Domain packages depend on store and IPC types](layering.md#dtos) | low |
+2 -2
View File
@@ -3,10 +3,10 @@
## The analyzers pass against a baseline, not against zero [#701] {#baseline} ## The analyzers pass against a baseline, not against zero [#701] {#baseline}
Costs: `make lint` and `make deadcode` are wired and green (V-694), but green Costs: `make lint` and `make deadcode` are wired and green (V-694), but green
means "nothing new since 2026-08-11". The accepted set is 19 staticcheck means "nothing new since 2026-08-11". The accepted set is 18 staticcheck
findings and 13 unreachable symbols, listed with a reason each in findings and 13 unreachable symbols, listed with a reason each in
`scripts/analyzers/*.baseline`. Three of the unreachable symbols must stay: `scripts/analyzers/*.baseline`. Three of the unreachable symbols must stay:
[layering.md](layering.md#deadcode). One accepted staticcheck finding is V-687. [layering.md](layering.md#deadcode).
Revisit when: V-701 sweeps the baseline, or a fix deletes an entry. The gate Revisit when: V-701 sweeps the baseline, or a fix deletes an entry. The gate
fails on an entry whose finding is gone, so the deletion is not optional. fails on an entry whose finding is gone, so the deletion is not optional.
Workaround: none needed. Reachability claims are checkable now. Read the Workaround: none needed. Reachability claims are checkable now. Read the
+1 -11
View File
@@ -1,6 +1,6 @@
# Background workers # Background workers
Both entries are a tick doing expensive work it did not need to do. The entry is a tick doing expensive work it did not need to do.
## Fact enrichment is a 20-call serial waterfall [#680] {#enrichment} ## Fact enrichment is a 20-call serial waterfall [#680] {#enrichment}
@@ -11,13 +11,3 @@ scan, not this.
Revisit when: Nexus gets slow, or when a batch-resolution endpoint exists. Revisit when: Nexus gets slow, or when a batch-resolution endpoint exists.
Workaround: an unreachable Nexus is fine. It is the slow-but-answering case Workaround: an unreachable Nexus is fine. It is the slow-but-answering case
that hurts. that hurts.
## A suppressed nudge is phrased anyway [#687] {#nudges}
Costs: `PhraseNudge` runs before the dedupe is known. The `continue` meant to
skip it is the last statement in the loop body. Every tick that
keeps suppressing the same rule pays the resident model again. The comment
above it claims the opposite.
Revisit when: digestion ticks show up in the model's load, or when nudge rules
grow past a handful.
Workaround: none.
+91 -28
View File
@@ -29,27 +29,53 @@ const (
// DigestEntry — one gate-suppressed care candidate durably held for later // DigestEntry — one gate-suppressed care candidate durably held for later
// bundled delivery. // bundled delivery.
type DigestEntry struct { type DigestEntry struct {
ID int64 ID int64
Rule string Rule string
Severity int Severity int
Body string Body string
CreatedTs time.Time CandidateFingerprint string
ExpiresTs time.Time CreatedTs time.Time
ExpiresTs time.Time
} }
// DigestBodyHash is the dedupe key for a digest entry: same rule, same // DigestBodyHash records the exact presentation stored in a digest entry. The
// wording ⇒ the same suppressed nudge repeating across ticks, and he should // pre-phrase dedupe key is CandidateFingerprint; body_hash remains useful for
// hear it once, not once per tick it kept getting suppressed. // audit/integrity and for legacy rows written before candidate identity was
// persisted.
func DigestBodyHash(rule, body string) string { func DigestBodyHash(rule, body string) string {
sum := sha256.Sum256([]byte(rule + "\x00" + body)) sum := sha256.Sum256([]byte(rule + "\x00" + body))
return hex.EncodeToString(sum[:8]) return hex.EncodeToString(sum[:8])
} }
// LiveDigestEntry returns the pending, unexpired entry for one semantic
// candidate occurrence. This is intentionally a store read rather than an
// in-memory cache: the caller uses it before PhraseNudge, including on the
// first tick after a daemon restart.
func (s *Store) LiveDigestEntry(ctx context.Context, rule, candidateFingerprint string, now time.Time) (DigestEntry, bool, error) {
if candidateFingerprint == "" {
return DigestEntry{}, false, errors.New("live digest entry: empty candidate fingerprint")
}
row := s.db.QueryRowContext(ctx,
`SELECT id, rule, severity, body, candidate_fingerprint, created_ts, expires_ts
FROM digest_entries
WHERE status = ? AND rule = ? AND candidate_fingerprint = ? AND expires_ts > ?
LIMIT 1`,
DigestPending, rule, candidateFingerprint, now.UnixMilli())
entry, err := scanDigestEntry(row)
if errors.Is(err, sql.ErrNoRows) {
return DigestEntry{}, false, nil
}
if err != nil {
return DigestEntry{}, false, fmt.Errorf("live digest entry: %w", err)
}
return entry, true, nil
}
// EnqueueDigestEntry durably records a suppressed care candidate worth // EnqueueDigestEntry durably records a suppressed care candidate worth
// resurfacing later. If a LIVE pending entry with the same rule+body already // resurfacing later. If a LIVE pending entry with the same rule+candidate
// exists, this is a no-op that returns the existing id and deduped=true — // fingerprint already exists, this is a no-op that returns the existing id
// the same suppressed nudge repeating across ticks must not pile up into // and deduped=true. The lookup and insert share a transaction so a second
// several copies of itself in the eventual bundle. // caller cannot race the pre-phrase read into a duplicate row.
// //
// "Live" carries the same expiry test PendingDigestEntries reads with, and for // "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 same reason: a row past its expires_ts is still status='pending' until
@@ -57,15 +83,36 @@ func DigestBodyHash(rule, body string) string {
// against one meant reporting deduped=true against an entry that will never be // 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 // 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 // reaches the bundle. Not yet swept must not mean still deliverable on the
// write side either. // write side either. A matching stale row is expired inside this transaction
func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity int, body string, now, expiresAt time.Time) (id int64, deduped bool, err error) { // before insertion so the partial unique index does not make sweep order part
// of correctness.
func (s *Store) EnqueueDigestEntry(ctx context.Context, rule, candidateFingerprint string, severity int, body string, now, expiresAt time.Time) (id int64, deduped bool, err error) {
if candidateFingerprint == "" {
return 0, false, errors.New("enqueue digest entry: empty candidate fingerprint")
}
hash := DigestBodyHash(rule, body) hash := DigestBodyHash(rule, body)
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return 0, false, fmt.Errorf("enqueue digest entry: begin: %w", err)
}
defer func() { _ = tx.Rollback() }()
if _, err := tx.ExecContext(ctx,
`UPDATE digest_entries SET status = ?
WHERE status = ? AND rule = ? AND candidate_fingerprint = ? AND expires_ts <= ?`,
DigestExpired, DigestPending, rule, candidateFingerprint, now.UnixMilli()); err != nil {
return 0, false, fmt.Errorf("enqueue digest entry: expire stale candidate: %w", err)
}
var existing int64 var existing int64
err = s.db.QueryRowContext(ctx, err = tx.QueryRowContext(ctx,
`SELECT id FROM digest_entries `SELECT id FROM digest_entries
WHERE status = ? AND rule = ? AND body_hash = ? AND expires_ts > ? LIMIT 1`, WHERE status = ? AND rule = ? AND candidate_fingerprint = ? AND expires_ts > ? LIMIT 1`,
DigestPending, rule, hash, now.UnixMilli()).Scan(&existing) DigestPending, rule, candidateFingerprint, now.UnixMilli()).Scan(&existing)
if err == nil { if err == nil {
if err := tx.Commit(); err != nil {
return 0, false, fmt.Errorf("enqueue digest entry: dedupe commit: %w", err)
}
return existing, true, nil return existing, true, nil
} }
if !errors.Is(err, sql.ErrNoRows) { if !errors.Is(err, sql.ErrNoRows) {
@@ -74,10 +121,11 @@ func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity in
return 0, false, fmt.Errorf("enqueue digest entry: dedupe lookup: %w", err) return 0, false, fmt.Errorf("enqueue digest entry: dedupe lookup: %w", err)
} }
res, err := s.db.ExecContext(ctx, res, err := tx.ExecContext(ctx,
`INSERT INTO digest_entries (rule, severity, body, body_hash, status, created_ts, expires_ts) `INSERT INTO digest_entries
VALUES (?, ?, ?, ?, ?, ?, ?)`, (rule, severity, body, body_hash, candidate_fingerprint, status, created_ts, expires_ts)
rule, severity, body, hash, DigestPending, now.UnixMilli(), expiresAt.UnixMilli()) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`,
rule, severity, body, hash, candidateFingerprint, DigestPending, now.UnixMilli(), expiresAt.UnixMilli())
if err != nil { if err != nil {
return 0, false, fmt.Errorf("enqueue digest entry: %w", err) return 0, false, fmt.Errorf("enqueue digest entry: %w", err)
} }
@@ -85,6 +133,9 @@ func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity in
if err != nil { if err != nil {
return 0, false, fmt.Errorf("enqueue digest entry: last insert id: %w", err) return 0, false, fmt.Errorf("enqueue digest entry: last insert id: %w", err)
} }
if err := tx.Commit(); err != nil {
return 0, false, fmt.Errorf("enqueue digest entry: commit: %w", err)
}
return id, false, nil return id, false, nil
} }
@@ -93,7 +144,7 @@ func (s *Store) EnqueueDigestEntry(ctx context.Context, rule string, severity in
// a bundled readout should mention them. // a bundled readout should mention them.
func (s *Store) PendingDigestEntries(ctx context.Context, now time.Time) ([]DigestEntry, error) { func (s *Store) PendingDigestEntries(ctx context.Context, now time.Time) ([]DigestEntry, error) {
rows, err := s.db.QueryContext(ctx, rows, err := s.db.QueryContext(ctx,
`SELECT id, rule, severity, body, created_ts, expires_ts `SELECT id, rule, severity, body, candidate_fingerprint, created_ts, expires_ts
FROM digest_entries WHERE status = ? AND expires_ts > ? ORDER BY created_ts ASC`, FROM digest_entries WHERE status = ? AND expires_ts > ? ORDER BY created_ts ASC`,
DigestPending, now.UnixMilli()) DigestPending, now.UnixMilli())
if err != nil { if err != nil {
@@ -103,18 +154,30 @@ func (s *Store) PendingDigestEntries(ctx context.Context, now time.Time) ([]Dige
var out []DigestEntry var out []DigestEntry
for rows.Next() { for rows.Next() {
var e DigestEntry e, err := scanDigestEntry(rows)
var created, expires int64 if err != nil {
if err := rows.Scan(&e.ID, &e.Rule, &e.Severity, &e.Body, &created, &expires); err != nil {
return nil, fmt.Errorf("pending digest entries: scan: %w", err) return nil, fmt.Errorf("pending digest entries: scan: %w", err)
} }
e.CreatedTs = time.UnixMilli(created)
e.ExpiresTs = time.UnixMilli(expires)
out = append(out, e) out = append(out, e)
} }
return out, rows.Err() return out, rows.Err()
} }
type digestScanner interface {
Scan(dest ...any) error
}
func scanDigestEntry(row digestScanner) (DigestEntry, error) {
var e DigestEntry
var created, expires int64
if err := row.Scan(&e.ID, &e.Rule, &e.Severity, &e.Body, &e.CandidateFingerprint, &created, &expires); err != nil {
return DigestEntry{}, err
}
e.CreatedTs = time.UnixMilli(created)
e.ExpiresTs = time.UnixMilli(expires)
return e, nil
}
// ExpireStaleDigestEntries marks pending entries whose expires_ts has passed // ExpireStaleDigestEntries marks pending entries whose expires_ts has passed
// as expired — stale information (yesterday's battery warning) is noise, not // as expired — stale information (yesterday's battery warning) is noise, not
// news, so it is dropped rather than delivered late. Called once per tick, // news, so it is dropped rather than delivered late. Called once per tick,
+61 -13
View File
@@ -13,7 +13,7 @@ func TestDigestEntryRoundTrips(t *testing.T) {
ctx := context.Background() ctx := context.Background()
now := time.Now() now := time.Now()
id, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", now, now.Add(24*time.Hour)) id, deduped, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "ты долго не отдыхала", now, now.Add(24*time.Hour))
if err != nil { if err != nil {
t.Fatalf("enqueue: %v", err) t.Fatalf("enqueue: %v", err)
} }
@@ -34,6 +34,9 @@ func TestDigestEntryRoundTrips(t *testing.T) {
if entries[0].Rule != "break" || entries[0].Severity != 2 || entries[0].Body != "ты долго не отдыхала" { if entries[0].Rule != "break" || entries[0].Severity != 2 || entries[0].Body != "ты долго не отдыхала" {
t.Fatalf("entry contents wrong: %+v", entries[0]) t.Fatalf("entry contents wrong: %+v", entries[0])
} }
if entries[0].CandidateFingerprint != "break-occurrence-1" {
t.Fatalf("candidate fingerprint did not round-trip: %+v", entries[0])
}
} }
// TestDigestEntrySurvivesRestart — durability is the whole point: a fresh // TestDigestEntrySurvivesRestart — durability is the whole point: a fresh
@@ -48,7 +51,7 @@ func TestDigestEntrySurvivesRestart(t *testing.T) {
if err != nil { if err != nil {
t.Fatalf("open: %v", err) t.Fatalf("open: %v", err)
} }
id, _, err := s1.EnqueueDigestEntry(ctx, "break", 2, "перерыв", now, now.Add(24*time.Hour)) id, _, err := s1.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "перерыв", now, now.Add(24*time.Hour))
if err != nil { if err != nil {
t.Fatalf("enqueue: %v", err) t.Fatalf("enqueue: %v", err)
} }
@@ -72,15 +75,15 @@ func TestDigestEntrySurvivesRestart(t *testing.T) {
} }
} }
// TestDigestEntryDedupesSameRuleAndBody — the same suppressed nudge // TestDigestEntryDedupesSameCandidateBeforeWording — the same semantic nudge
// repeating across ticks (quiet hours holding for hours) must not pile up // repeating across ticks (quiet hours holding for hours) must not pile up
// into several copies of itself; he hears it once. // into several copies even if a concurrent phraser returned another variant.
func TestDigestEntryDedupesSameRuleAndBody(t *testing.T) { func TestDigestEntryDedupesSameCandidateBeforeWording(t *testing.T) {
s := newTestStore(t) s := newTestStore(t)
ctx := context.Background() ctx := context.Background()
now := time.Now() now := time.Now()
id1, deduped1, err := s.EnqueueDigestEntry(ctx, "break", 2, "перерыв нужен", now, now.Add(24*time.Hour)) id1, deduped1, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "перерыв нужен", now, now.Add(24*time.Hour))
if err != nil { if err != nil {
t.Fatalf("first enqueue: %v", err) t.Fatalf("first enqueue: %v", err)
} }
@@ -89,12 +92,12 @@ func TestDigestEntryDedupesSameRuleAndBody(t *testing.T) {
} }
for i := 0; i < 2; i++ { for i := 0; i < 2; i++ {
id2, deduped2, err := s.EnqueueDigestEntry(ctx, "break", 2, "перерыв нужен", now.Add(time.Minute), now.Add(25*time.Hour)) id2, deduped2, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "другой вариант", now.Add(time.Minute), now.Add(25*time.Hour))
if err != nil { if err != nil {
t.Fatalf("repeat enqueue: %v", err) t.Fatalf("repeat enqueue: %v", err)
} }
if !deduped2 { if !deduped2 {
t.Fatal("repeat enqueue of the same rule+body should report deduped") t.Fatal("repeat enqueue of the same candidate should report deduped")
} }
if id2 != id1 { if id2 != id1 {
t.Fatalf("deduped enqueue should return the original id: want %d got %d", id1, id2) t.Fatalf("deduped enqueue should return the original id: want %d got %d", id1, id2)
@@ -119,7 +122,7 @@ func TestDigestEntryExpiresRatherThanDeliversLate(t *testing.T) {
created := time.Now() created := time.Now()
expiresAt := created.Add(time.Hour) expiresAt := created.Add(time.Hour)
id, _, err := s.EnqueueDigestEntry(ctx, "water", 1, "стакан воды", created, expiresAt) id, _, err := s.EnqueueDigestEntry(ctx, "water", "water-occurrence-1", 1, "стакан воды", created, expiresAt)
if err != nil { if err != nil {
t.Fatalf("enqueue: %v", err) t.Fatalf("enqueue: %v", err)
} }
@@ -169,11 +172,11 @@ func TestDigestEntryDrainMarksDrainedNotDeleted(t *testing.T) {
ctx := context.Background() ctx := context.Background()
now := time.Now() now := time.Now()
id1, _, err := s.EnqueueDigestEntry(ctx, "break", 2, "перерыв", now, now.Add(24*time.Hour)) id1, _, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "перерыв", now, now.Add(24*time.Hour))
if err != nil { if err != nil {
t.Fatalf("enqueue 1: %v", err) t.Fatalf("enqueue 1: %v", err)
} }
id2, _, err := s.EnqueueDigestEntry(ctx, "break2", 2, "другое", now, now.Add(24*time.Hour)) id2, _, err := s.EnqueueDigestEntry(ctx, "break2", "break2-occurrence-1", 2, "другое", now, now.Add(24*time.Hour))
if err != nil { if err != nil {
t.Fatalf("enqueue 2: %v", err) t.Fatalf("enqueue 2: %v", err)
} }
@@ -213,7 +216,7 @@ func TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry(t *testing.T) {
created := time.Now() created := time.Now()
expiresAt := created.Add(time.Hour) expiresAt := created.Add(time.Hour)
first, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", created, expiresAt) first, deduped, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "ты долго не отдыхала", created, expiresAt)
if err != nil { if err != nil {
t.Fatalf("enqueue: %v", err) t.Fatalf("enqueue: %v", err)
} }
@@ -224,7 +227,7 @@ func TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry(t *testing.T) {
// A tick after the expiry, with the sweep not yet run: the same suppressed // A tick after the expiry, with the sweep not yet run: the same suppressed
// nudge comes round again and must be recorded afresh. // nudge comes round again and must be recorded afresh.
after := expiresAt.Add(time.Minute) after := expiresAt.Add(time.Minute)
second, deduped, err := s.EnqueueDigestEntry(ctx, "break", 2, "ты долго не отдыхала", after, after.Add(time.Hour)) second, deduped, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "ты долго не отдыхала", after, after.Add(time.Hour))
if err != nil { if err != nil {
t.Fatalf("re-enqueue: %v", err) t.Fatalf("re-enqueue: %v", err)
} }
@@ -243,3 +246,48 @@ func TestDigestEnqueueDoesNotDedupeAgainstAnExpiredEntry(t *testing.T) {
t.Fatalf("want the fresh entry %d pending, got %+v", second, entries) t.Fatalf("want the fresh entry %d pending, got %+v", second, entries)
} }
} }
func TestDigestCandidateMeaningChangeCreatesAnotherEntry(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
now := time.Now()
first, deduped, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-1", 2, "перерыв", now, now.Add(time.Hour))
if err != nil || deduped {
t.Fatalf("first enqueue: id=%d deduped=%v err=%v", first, deduped, err)
}
second, deduped, err := s.EnqueueDigestEntry(ctx, "break", "break-occurrence-2", 2, "перерыв", now.Add(time.Minute), now.Add(time.Hour))
if err != nil || deduped {
t.Fatalf("changed candidate enqueue: id=%d deduped=%v err=%v", second, deduped, err)
}
if second == first {
t.Fatalf("changed candidate reused id %d", first)
}
entries, err := s.PendingDigestEntries(ctx, now.Add(time.Minute))
if err != nil || len(entries) != 2 {
t.Fatalf("two distinct occurrences should remain pending: entries=%+v err=%v", entries, err)
}
}
func TestLiveDigestEntryStopsMatchingAfterDrain(t *testing.T) {
s := newTestStore(t)
ctx := context.Background()
now := time.Now()
fingerprint := "break-occurrence-1"
id, _, err := s.EnqueueDigestEntry(ctx, "break", fingerprint, 2, "перерыв", now, now.Add(time.Hour))
if err != nil {
t.Fatal(err)
}
if _, found, err := s.LiveDigestEntry(ctx, "break", fingerprint, now); err != nil || !found {
t.Fatalf("live lookup before drain: found=%v err=%v", found, err)
}
if err := s.DrainDigestEntries(ctx, []int64{id}, now.Add(time.Minute)); err != nil {
t.Fatal(err)
}
if _, found, err := s.LiveDigestEntry(ctx, "break", fingerprint, now.Add(time.Minute)); err != nil || found {
t.Fatalf("drained occurrence still matched: found=%v err=%v", found, err)
}
if _, deduped, err := s.EnqueueDigestEntry(ctx, "break", fingerprint, 2, "новый перерыв", now.Add(2*time.Minute), now.Add(2*time.Hour)); err != nil || deduped {
t.Fatalf("resolved occurrence should be enqueueable again: deduped=%v err=%v", deduped, err)
}
}
+11
View File
@@ -374,6 +374,17 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2
CREATE INDEX IF NOT EXISTS idx_delivery_attempts_reminder_group CREATE INDEX IF NOT EXISTS idx_delivery_attempts_reminder_group
ON delivery_attempts (delivery_group, status) ON delivery_attempts (delivery_group, status)
WHERE kind = 'reminder' AND delivery_group <> '';`, WHERE kind = 'reminder' AND delivery_group <> '';`,
// #26 — pre-phrase identity for the suppressed-nudge digest (V-687).
// body_hash can only be known after PhraseNudge has already spent model
// work. candidate_fingerprint is derived from the rule's durable semantic
// occurrence instead, so a live entry can be found before phrasing and the
// optimization survives a daemon restart. Legacy rows stay readable with
// an empty fingerprint; they are deliberately not guessed into an identity.
`ALTER TABLE digest_entries ADD COLUMN candidate_fingerprint TEXT NOT NULL DEFAULT '';
CREATE UNIQUE INDEX IF NOT EXISTS idx_digest_entries_live_candidate
ON digest_entries (rule, candidate_fingerprint)
WHERE status = 'pending' AND candidate_fingerprint <> '';`,
} }
// migrate applies every migration with a number greater than the DB's current // migrate applies every migration with a number greater than the DB's current
-3
View File
@@ -8,9 +8,6 @@
# The sweep that empties it is V-701, which carries the judgement on each # The sweep that empties it is V-701, which carries the judgement on each
# entry. What follows is the short reason only. # entry. What follows is the short reason only.
# V-687. The dedupe check runs after the phraser has already been paid.
cmd/mavend/tick_digest.go SA4006 this value of deduped is never used
# V-686, the eleven unreachable symbols the 2026-08-10 audit listed, seen from # V-686, the eleven unreachable symbols the 2026-08-10 audit listed, seen from
# the other side. Three of them must stay: docs/caveats/layering.md#deadcode. # the other side. Three of them must stay: docs/caveats/layering.md#deadcode.
cmd/mavend/replier_llm_test.go U1000 func assertStub is unused cmd/mavend/replier_llm_test.go U1000 func assertStub is unused