From 859bbf750f65b975eb7a29809d4c7250d7f9769e Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 31 Jul 2026 02:34:54 +0400 Subject: [PATCH 1/3] Never send a nudge body off-box when the summary is empty (#368) Away channels (ntfy, telegram) leave the box, so an empty Summary now sends a fixed generic line plus the rule name instead of the full Body. The dispatcher strips detail before any sink sees it, so a sink added later cannot leak by reading the wrong field. Voice is local and unchanged. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ --- internal/delivery/dispatcher.go | 49 +++++++++++++++++++++++++-------- 1 file changed, 38 insertions(+), 11 deletions(-) diff --git a/internal/delivery/dispatcher.go b/internal/delivery/dispatcher.go index 93fc47d..f3ad35d 100644 --- a/internal/delivery/dispatcher.go +++ b/internal/delivery/dispatcher.go @@ -160,6 +160,7 @@ func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now tim RepeatUntilAck: ch == ChannelTelegram && c.Severity >= loop.Sev4, Ts: now, } + s = minimalForAway(s) sink := d.sinkFor(ch) if sink == nil { continue @@ -225,6 +226,7 @@ func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, n Summary: pr.Summary, Ts: now, } + s = minimalForAway(s) sink := d.sinkFor(ch) if sink == nil { continue @@ -315,6 +317,7 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time. RepeatUntilAck: true, Ts: now, } + s = minimalForAway(s) attemptID := d.beginOutbox(ctx, "nudge", key, 0, ChannelTelegram, messageForChannel(s), now) if err := d.cfg.Telegram.Send(ctx, s); err != nil { d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) @@ -342,20 +345,44 @@ func (d *Dispatcher) sinkFor(ch Channel) Sink { } } +// GenericAwayMessage — what an away channel gets when the phraser gave us no +// summary. no gendered forms, so it stays right whoever reads it. +const GenericAwayMessage = "что-то требует внимания" + +// isAway — this channel leaves the box, so it only ever gets a minimal body. +func isAway(ch Channel) bool { + return ch == ChannelNtfy || ch == ChannelTelegram +} + // messageForChannel — away channels get the minimal summary (no shoulder-surf // exfil — "disk low on homesrv," not detail); voice gets the full body (local). -// a missing summary falls back to body — a terse full message is better than -// no message, and the phraser should have produced a summary for away-bound -// severities. this is the "minimal body" rule from the spec, enforced at the -// last mile so a phraser bug can't accidentally exfil via the relay. +// an empty summary must NOT fall back to the body: the resident model is small +// and drops fields often, and the away path crosses the "never phones home" +// boundary. so we send a fixed generic line plus the rule name instead. voice +// is local, so it keeps the full body. func messageForChannel(s Sendable) string { - switch s.Channel { - case ChannelNtfy, ChannelTelegram: - if s.Summary != "" { - return s.Summary - } - return s.Body - default: + if !isAway(s.Channel) { return s.Body } + if s.Summary != "" { + return s.Summary + } + if s.RuleName != "" { + return GenericAwayMessage + ": " + s.RuleName + } + return GenericAwayMessage +} + +// minimalForAway — strips detail from a Sendable bound for an away channel +// before any sink sees it. the sinks pick Summary themselves too, but this is +// where the boundary actually is: a sink added later must not be able to leak +// the full body just by reading the wrong field. +func minimalForAway(s Sendable) Sendable { + if !isAway(s.Channel) { + return s + } + msg := messageForChannel(s) + s.Body = msg + s.Summary = msg + return s } From 215aa331c57af7368f92d5bad8d9147e088a4df1 Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 31 Jul 2026 02:35:54 +0400 Subject: [PATCH 2/3] Recover from a panicking sink so the attempt is always closed (#369) A panic in Send used to unwind past completeOutbox and leave the delivery_attempts row pending forever, since reconciliation only runs at startup. safeSend turns the panic into an error, logs it loudly, records the attempt failed, and lets the other channels for the same nudge still go out. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ --- internal/delivery/dispatcher.go | 38 ++++++++++++++++++++++++++++++--- 1 file changed, 35 insertions(+), 3 deletions(-) diff --git a/internal/delivery/dispatcher.go b/internal/delivery/dispatcher.go index f3ad35d..22b7724 100644 --- a/internal/delivery/dispatcher.go +++ b/internal/delivery/dispatcher.go @@ -166,7 +166,14 @@ func (d *Dispatcher) DispatchNudge(ctx context.Context, pn PhrasedNudge, now tim continue } attemptID := d.beginOutbox(ctx, "nudge", c.Rule.Name, 0, ch, messageForChannel(s), now) - if err := sink.Send(ctx, s); err != nil { + if err := safeSend(ctx, sink, s); err != nil { + if errors.Is(err, ErrSinkPanicked) { + // one broken sink must not eat the other channels for this + // nudge (sev4 present is voice + ntfy). the attempt is closed + // as failed and we move on. + d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) + continue + } if errors.Is(err, ErrVoiceNoSession) { // voice was assumed reachable (presence=present) but no live // session exists — the presence guess was wrong. reroute through @@ -232,7 +239,11 @@ func (d *Dispatcher) DispatchReminder(ctx context.Context, pr PhrasedReminder, n continue } attemptID := d.beginOutbox(ctx, "reminder", "", rd.Reminder.ID, ch, messageForChannel(s), now) - if err := sink.Send(ctx, s); err != nil { + if err := safeSend(ctx, sink, s); err != nil { + if errors.Is(err, ErrSinkPanicked) { + d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) + continue + } if errors.Is(err, ErrVoiceNoSession) { // presence guess was wrong — reroute reminder to the away // channel (ntfy). voice is the only present channel, so nothing @@ -319,8 +330,11 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time. } s = minimalForAway(s) attemptID := d.beginOutbox(ctx, "nudge", key, 0, ChannelTelegram, messageForChannel(s), now) - if err := d.cfg.Telegram.Send(ctx, s); err != nil { + if err := safeSend(ctx, d.cfg.Telegram, s); err != nil { d.completeOutbox(ctx, attemptID, store.DeliveryFailed, now) + if errors.Is(err, ErrSinkPanicked) { + continue + } return out, fmt.Errorf("repeat send telegram %s: %w", key, err) } d.completeOutbox(ctx, attemptID, store.DeliverySent, now) @@ -332,6 +346,24 @@ func (d *Dispatcher) RepeatUnacked(ctx context.Context, keys []string, now time. return out, nil } +// ErrSinkPanicked — a sink panicked mid-send. the send did not happen, so the +// attempt is recorded failed and never silently retried as if it had. +var ErrSinkPanicked = errors.New("delivery: sink panicked mid-send") + +// safeSend calls a sink and turns a panic into an error. without this a +// panicking sink unwinds past completeOutbox and leaves the delivery_attempts +// row pending forever — reconciliation only runs at daemon startup, and core +// is long-lived, so the row would sit there for weeks. +func safeSend(ctx context.Context, sink Sink, s Sendable) (err error) { + defer func() { + if r := recover(); r != nil { + log.Printf("dispatcher: PANIC in %s sink (this is a bug, fix the sink): %v", s.Channel, r) + err = fmt.Errorf("%w: %s: %v", ErrSinkPanicked, s.Channel, r) + } + }() + return sink.Send(ctx, s) +} + func (d *Dispatcher) sinkFor(ch Channel) Sink { switch ch { case ChannelVoice: From 5fd25d7ad7d9e92be5a49f28d6b49f2fcb06d719 Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 31 Jul 2026 02:39:38 +0400 Subject: [PATCH 3/3] Test the away-channel minimal body and the panicking sink (#368, #369) The integration branch names one test TestAwayFallsBackToFullBodyWhenSummaryEmpty, which describes the old bug; it is here as TestAwaySendsGenericLineWhenSummaryEmpty and asserts the generic line instead of the body. Also covers: a normal summary goes out unchanged, voice keeps the full body, and one panicking sink does not eat the other channel for the same nudge. Reformatted one pre-existing struct. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01CGeSZxh1DCtRxmFVSYVGvJ --- internal/delivery/dispatcher_test.go | 229 ++++++++++++++++++++++++++- 1 file changed, 224 insertions(+), 5 deletions(-) diff --git a/internal/delivery/dispatcher_test.go b/internal/delivery/dispatcher_test.go index e60faeb..e3d79a0 100644 --- a/internal/delivery/dispatcher_test.go +++ b/internal/delivery/dispatcher_test.go @@ -639,11 +639,11 @@ func TestDispatchRecurringReminderReschedules(t *testing.T) { // ----------------------------- durable outbox -------------------------------- type outboxAttempt struct { - kind, rule string - reminderID int64 - channel, hash string - status string - begunAt, doneAt time.Time + kind, rule string + reminderID int64 + channel, hash string + status string + begunAt, doneAt time.Time } // fakeOutbox — an in-memory Outbox that also lets a test simulate a crash @@ -784,3 +784,222 @@ func TestDispatchNudge_OutboxBeginFailureDoesNotBlockSend(t *testing.T) { t.Fatalf("send should still happen despite outbox begin failure: sends=%v out=%v", voice.sends, out) } } + +// ----------------------- away channels carry no detail ----------------------- + +// panicSink — a broken sink. models the #369 case: the sink blows up mid-send. +type panicSink struct{ calls int } + +func (p *panicSink) Send(_ context.Context, _ Sendable) error { + p.calls++ + panic("sink is broken") +} + +// TestAwaySendsGenericLineWhenSummaryEmpty — #368. The phraser is a small +// model and drops fields often. An empty Summary must NOT put the full body +// on a channel that leaves the box; the away sendable gets a fixed generic +// line plus the rule name instead. +func TestAwaySendsGenericLineWhenSummaryEmpty(t *testing.T) { + ntfy := &fakeSink{} + rec := &fakeNudgeRecorder{} + d := NewDispatcher(Config{Ntfy: ntfy, Nudges: rec}) + + body := "disk /dev/sda1 at 96%, 2.1G free, largest offender /var/lib/docker" + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("disk-low", loop.Sev3, store.Away), + Body: body, + Summary: "", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(ntfy.sends) != 1 { + t.Fatalf("want 1 ntfy send, got %d", len(ntfy.sends)) + } + want := GenericAwayMessage + ": disk-low" + got := messageForChannel(ntfy.sends[0]) + if got != want { + t.Fatalf("away message: want %q, got %q", want, got) + } + if ntfy.sends[0].Body == body { + t.Fatal("away sendable still carries the full body") + } + if len(rec.rows) != 1 || rec.rows[0].message != want { + t.Fatalf("recorded message: want %q, got %+v", want, rec.rows) + } +} + +// TestSev4AwaySendableCarriesNoDetail — the rule is enforced at the +// dispatcher, not in each sink. A sink added later must not be able to leak +// the body just by reading the wrong field, so neither field may hold detail. +func TestSev4AwaySendableCarriesNoDetail(t *testing.T) { + tg := &fakeSink{} + d := NewDispatcher(Config{Telegram: tg, Ack: newFakeAck()}) + + body := "backup job failed: rsync exit 23 on /home/kami, see /var/log/backup.log" + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("backup-failed", loop.Sev4, store.Away), + Body: body, + Summary: "бэкап не прошёл", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(tg.sends) != 1 { + t.Fatalf("want 1 telegram send, got %d", len(tg.sends)) + } + s := tg.sends[0] + if s.Body != "бэкап не прошёл" || s.Summary != "бэкап не прошёл" { + t.Fatalf("away sendable should hold only the summary, got body=%q summary=%q", s.Body, s.Summary) + } +} + +// TestAwayKeepsANonEmptySummary — the normal path is untouched. +func TestAwayKeepsANonEmptySummary(t *testing.T) { + ntfy := &fakeSink{} + d := NewDispatcher(Config{Ntfy: ntfy}) + + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("cert", loop.Sev3, store.Away), + Body: "cert for maven.local expires in 3 days, issuer letsencrypt", + Summary: "сертификат истекает", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if got := messageForChannel(ntfy.sends[0]); got != "сертификат истекает" { + t.Fatalf("want the summary unchanged, got %q", got) + } +} + +// TestVoiceStillGetsTheFullBody — voice never leaves the box, so it keeps the +// full phrased message even when Summary is empty. +func TestVoiceStillGetsTheFullBody(t *testing.T) { + voice := &fakeSink{} + d := NewDispatcher(Config{Voice: voice}) + + body := "ты не пил воду четыре часа" + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("water", loop.Sev1, store.Present), + Body: body, + Summary: "", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(voice.sends) != 1 || voice.sends[0].Body != body { + t.Fatalf("voice should get the full body, got %+v", voice.sends) + } + if got := messageForChannel(voice.sends[0]); got != body { + t.Fatalf("voice message: want %q, got %q", body, got) + } +} + +// TestReminderAwaySendsGenericLineWhenSummaryEmpty — the reminder path crosses +// the same boundary and has its own Sendable construction. +func TestReminderAwaySendsGenericLineWhenSummaryEmpty(t *testing.T) { + ntfy := &fakeSink{} + d := NewDispatcher(Config{Ntfy: ntfy}) + + _, err := d.DispatchReminder(context.Background(), PhrasedReminder{ + Decision: loop.ReminderDecision{ + Reminder: store.Reminder{ID: 5}, + State: loop.State{Now: refNow(), Presence: store.Away}, + }, + Body: "позвонить в клинику по поводу анализов", + Summary: "", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if len(ntfy.sends) != 1 { + t.Fatalf("want 1 ntfy send, got %d", len(ntfy.sends)) + } + if got := messageForChannel(ntfy.sends[0]); got != GenericAwayMessage { + t.Fatalf("away reminder message: want %q, got %q", GenericAwayMessage, got) + } +} + +// --------------------------- a panicking sink ------------------------------- + +// TestPanicMidSendResolvesTheAttempt — #369. A panic used to unwind past +// completeOutbox and leave the row pending forever, because reconciliation +// only runs at daemon startup and core is long-lived. +func TestPanicMidSendResolvesTheAttempt(t *testing.T) { + ob := &fakeOutbox{} + d := NewDispatcher(Config{Voice: &panicSink{}, Outbox: ob}) + + _, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("water", loop.Sev1, store.Present), + Body: "body", Summary: "sum", + }, refNow()) + if err != nil { + t.Fatalf("a panicking sink must not fail the dispatch: %v", err) + } + if len(ob.attempts) != 1 { + t.Fatalf("want 1 outbox attempt, got %d", len(ob.attempts)) + } + if ob.attempts[0].status != store.DeliveryFailed { + t.Fatalf("want status failed after a panic, got %q", ob.attempts[0].status) + } +} + +// TestPanicInOneSinkStillDeliversTheOther — sev4 present is voice + ntfy. One +// broken sink must not eat the other channel for the same nudge. +func TestPanicInOneSinkStillDeliversTheOther(t *testing.T) { + bad := &panicSink{} + ntfy := &fakeSink{} + ob := &fakeOutbox{} + d := NewDispatcher(Config{Voice: bad, Ntfy: ntfy, Outbox: ob}) + + out, err := d.DispatchNudge(context.Background(), PhrasedNudge{ + Candidate: candidate("backup-failed", loop.Sev4, store.Present), + Body: "long detailed body", Summary: "бэкап не прошёл", + }, refNow()) + if err != nil { + t.Fatalf("dispatch: %v", err) + } + if bad.calls != 1 { + t.Fatalf("want the voice sink called once, got %d", bad.calls) + } + if len(ntfy.sends) != 1 { + t.Fatalf("ntfy should still get the nudge, got %d sends", len(ntfy.sends)) + } + if len(out) != 1 || out[0].Sendable.Channel != ChannelNtfy { + t.Fatalf("want only the ntfy dispatch reported, got %+v", out) + } + if len(ob.attempts) != 2 { + t.Fatalf("want 2 outbox attempts, got %d", len(ob.attempts)) + } + if ob.attempts[0].status != store.DeliveryFailed || ob.attempts[1].status != store.DeliverySent { + t.Fatalf("want [failed, sent], got %q %q", ob.attempts[0].status, ob.attempts[1].status) + } +} + +// TestPanicInReminderSinkResolvesTheAttempt — the reminder path has its own +// send call, and a panic there must not leave the reminder marked fired. +func TestPanicInReminderSinkResolvesTheAttempt(t *testing.T) { + ob := &fakeOutbox{} + rc := &fakeReminderCompleter{} + d := NewDispatcher(Config{Voice: &panicSink{}, Reminders: rc, Outbox: ob}) + + out, err := d.DispatchReminder(context.Background(), PhrasedReminder{ + Decision: loop.ReminderDecision{ + Reminder: store.Reminder{ID: 9}, + State: loop.State{Now: refNow(), Presence: store.Present}, + }, + Body: "звонок", Summary: "звонок", + }, refNow()) + if err != nil { + t.Fatalf("a panicking sink must not fail the dispatch: %v", err) + } + if len(out) != 0 { + t.Fatalf("nothing was delivered, want no dispatches, got %+v", out) + } + if len(ob.attempts) != 1 || ob.attempts[0].status != store.DeliveryFailed { + t.Fatalf("want 1 attempt with status failed, got %+v", ob.attempts) + } + if len(rc.marked) != 0 { + t.Fatalf("reminder must stay pending after a panic, got %+v", rc.marked) + } +}