Phase 7 follow-up: 1-month body retention and actions_log cap (AUD-75)

Implements the AGENTS.md §7 data rule: keep message bodies for one month, and
size-cap the append-only action log.

- internal/store/retention.go: Store.Retain clears messages.body_text older
  than the window (metadata stays) and deletes actions_log rows beyond the
  newest N, in one transaction.
- internal/config: EMAIL_AGENT_RETENTION_DAYS (30) and
  EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS (50000).
- cmd/email-agent: bounded -retain flag for the n8n schedule.
- Tests for body clearing, metadata retention, log cap, and disabled policy.

internal/store/retention.go

@@ -0,0 +1,97 @@
+// Retention for the data model in AGENTS.md §7: "keep message bodies for one
+// month. Logs are append-only and size-capped."
+//
+// Retention never touches message metadata. It only blanks `messages.body_text`
+// once a message is older than the window, and it trims the oldest rows from
+// the append-only `actions_log` so that log cannot grow without bound. It is a
+// bounded, explicit job (the CLI `-retain` flag), not something the HTTP server
+// runs on a request.
+package store
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// DefaultRetentionDays is the AGENTS.md §7 body window: one month.
+const DefaultRetentionDays = 30
+
+// DefaultActionsLogMaxRows caps the append-only action log. The exact number is
+// a policy choice (see docs/retention.md); it is large enough to hold a long
+// history of triage actions and small enough that the SQLite file stays modest.
+const DefaultActionsLogMaxRows = 50000
+
+// RetentionPolicy configures one retention pass. A non-positive field disables
+// that part of the pass, so a dry run can, for example, cap only the log.
+type RetentionPolicy struct {
+	// BodyMaxAge is how long a message body is kept. Bodies of messages with a
+	// received_at older than now-BodyMaxAge are set to NULL. Non-positive
+	// disables body clearing.
+	BodyMaxAge time.Duration
+	// ActionsLogMaxRows is the number of newest actions_log rows to keep. Older
+	// rows are deleted. Non-positive disables the cap.
+	ActionsLogMaxRows int
+}
+
+// RetentionResult reports what one retention pass changed.
+type RetentionResult struct {
+	// BodiesCleared is the number of messages whose body_text was blanked.
+	BodiesCleared int64
+	// ActionsDeleted is the number of oldest actions_log rows removed.
+	ActionsDeleted int64
+	// BodyCutoff is the received_at cutoff used for body clearing (UTC). It is
+	// the zero time when body clearing is disabled.
+	BodyCutoff time.Time
+}
+
+// Retain enforces the retention policy in a single transaction so a partial
+// pass never leaves the database in a mixed state. It is safe to run on a
+// schedule: every statement is idempotent.
+func (s *Store) Retain(ctx context.Context, p RetentionPolicy) (RetentionResult, error) {
+	var res RetentionResult
+
+	tx, err := s.db.BeginTx(ctx, nil)
+	if err != nil {
+		return res, fmt.Errorf("begin retention transaction: %w", err)
+	}
+	defer func() { _ = tx.Rollback() }()
+
+	if p.BodyMaxAge > 0 {
+		res.BodyCutoff = time.Now().UTC().Add(-p.BodyMaxAge)
+		// received_at is stored in sqlTimeLayout (UTC), so compare with the
+		// same format. body_text is already NULL or empty for a cleared row.
+		r, err := tx.ExecContext(ctx, `
+UPDATE messages
+SET body_text = NULL
+WHERE body_text IS NOT NULL AND body_text <> '' AND received_at < ?`,
+			res.BodyCutoff.Format(sqlTimeLayout))
+		if err != nil {
+			return res, fmt.Errorf("clear message bodies older than %s: %w", p.BodyMaxAge, err)
+		}
+		if res.BodiesCleared, err = r.RowsAffected(); err != nil {
+			return res, fmt.Errorf("count cleared message bodies: %w", err)
+		}
+	}
+
+	if p.ActionsLogMaxRows > 0 {
+		// Keep the newest N rows and delete the rest. id is monotonic, so it
+		// is the stable ordering for an append-only table.
+		r, err := tx.ExecContext(ctx, `
+DELETE FROM actions_log
+WHERE id NOT IN (
+    SELECT id FROM actions_log ORDER BY id DESC LIMIT ?
+)`, p.ActionsLogMaxRows)
+		if err != nil {
+			return res, fmt.Errorf("cap actions_log at %d rows: %w", p.ActionsLogMaxRows, err)
+		}
+		if res.ActionsDeleted, err = r.RowsAffected(); err != nil {
+			return res, fmt.Errorf("count deleted action log rows: %w", err)
+		}
+	}
+
+	if err := tx.Commit(); err != nil {
+		return res, fmt.Errorf("commit retention: %w", err)
+	}
+	return res, nil
+}

internal/store/retention_test.go

@@ -0,0 +1,142 @@
+package store
+
+import (
+	"context"
+	"testing"
+	"time"
+)
+
+// seedBodyMessage stores one message with a body and a chosen received_at.
+func seedBodyMessage(t *testing.T, s *Store, ctx context.Context, gmailID, body string, receivedAt time.Time) int64 {
+	t.Helper()
+	id, err := s.UpsertMessage(ctx, &Message{
+		GmailID:       gmailID,
+		FromAddr:      "sender@example.test",
+		ToAddr:        "hello@where-woof.com",
+		Subject:       "Subject " + gmailID,
+		BodyText:      body,
+		SourceAddress: "hello@where-woof.com",
+		DateSent:      receivedAt,
+		ReceivedAt:    receivedAt,
+	})
+	if err != nil {
+		t.Fatalf("seed body message: %v", err)
+	}
+	return id
+}
+
+// bodyOf reads body_text directly, distinguishing NULL from empty.
+func bodyOf(t *testing.T, s *Store, ctx context.Context, id int64) (string, bool) {
+	t.Helper()
+	var body *string
+	if err := s.DB().QueryRowContext(ctx,
+		`SELECT body_text FROM messages WHERE id = ?`, id).Scan(&body); err != nil {
+		t.Fatalf("read body of message %d: %v", id, err)
+	}
+	if body == nil {
+		return "", true
+	}
+	return *body, false
+}
+
+func TestRetainClearsOldBodiesKeepsMetadata(t *testing.T) {
+	s, ctx := testStore(t)
+
+	old := seedBodyMessage(t, s, ctx, "old-1", "an old body", time.Now().UTC().Add(-45*24*time.Hour))
+	recent := seedBodyMessage(t, s, ctx, "recent-1", "a recent body", time.Now().UTC().Add(-2*24*time.Hour))
+
+	res, err := s.Retain(ctx, RetentionPolicy{BodyMaxAge: 30 * 24 * time.Hour})
+	if err != nil {
+		t.Fatalf("Retain() error = %v", err)
+	}
+	if res.BodiesCleared != 1 {
+		t.Errorf("BodiesCleared = %d, want 1", res.BodiesCleared)
+	}
+
+	if body, null := bodyOf(t, s, ctx, old); !null || body != "" {
+		t.Errorf("old body = %q (null=%v), want NULL", body, null)
+	}
+	if body, null := bodyOf(t, s, ctx, recent); null || body != "a recent body" {
+		t.Errorf("recent body = %q (null=%v), want unchanged", body, null)
+	}
+
+	// Metadata must survive the body clear.
+	m, err := s.MessageByID(ctx, old)
+	if err != nil {
+		t.Fatalf("MessageByID() error = %v", err)
+	}
+	if m.Subject != "Subject old-1" || m.SourceAddress != "hello@where-woof.com" {
+		t.Errorf("metadata lost after retention: %+v", m)
+	}
+	if m.BodyText != "" {
+		t.Errorf("MessageByID().BodyText = %q, want empty", m.BodyText)
+	}
+
+	// A second pass is a no-op: the row is already NULL, not an empty string.
+	res2, err := s.Retain(ctx, RetentionPolicy{BodyMaxAge: 30 * 24 * time.Hour})
+	if err != nil {
+		t.Fatalf("second Retain() error = %v", err)
+	}
+	if res2.BodiesCleared != 0 {
+		t.Errorf("second BodiesCleared = %d, want 0", res2.BodiesCleared)
+	}
+}
+
+func TestRetainCapsActionsLog(t *testing.T) {
+	s, ctx := testStore(t)
+
+	for i := 0; i < 5; i++ {
+		if err := s.LogAction(ctx, "fetch", "scheduler", "pass"); err != nil {
+			t.Fatalf("LogAction() %d error = %v", i, err)
+		}
+	}
+
+	res, err := s.Retain(ctx, RetentionPolicy{ActionsLogMaxRows: 2})
+	if err != nil {
+		t.Fatalf("Retain() error = %v", err)
+	}
+	if res.ActionsDeleted != 3 {
+		t.Errorf("ActionsDeleted = %d, want 3", res.ActionsDeleted)
+	}
+
+	entries, err := s.RecentActions(ctx, 100)
+	if err != nil {
+		t.Fatalf("RecentActions() error = %v", err)
+	}
+	if len(entries) != 2 {
+		t.Fatalf("actions_log rows = %d, want 2", len(entries))
+	}
+	// RecentActions is newest-first; the two kept rows are the newest inserts.
+	if entries[0].ID <= entries[1].ID {
+		t.Errorf("kept rows not the newest: %d, %d", entries[0].ID, entries[1].ID)
+	}
+
+	// A smaller table than the cap is untouched.
+	res2, err := s.Retain(ctx, RetentionPolicy{ActionsLogMaxRows: 50})
+	if err != nil {
+		t.Fatalf("second Retain() error = %v", err)
+	}
+	if res2.ActionsDeleted != 0 {
+		t.Errorf("second ActionsDeleted = %d, want 0", res2.ActionsDeleted)
+	}
+}
+
+func TestRetainDisabledPolicyIsNoOp(t *testing.T) {
+	s, ctx := testStore(t)
+
+	old := seedBodyMessage(t, s, ctx, "old-1", "an old body", time.Now().UTC().Add(-400*24*time.Hour))
+	if err := s.LogAction(ctx, "fetch", "scheduler", "pass"); err != nil {
+		t.Fatalf("LogAction() error = %v", err)
+	}
+
+	res, err := s.Retain(ctx, RetentionPolicy{})
+	if err != nil {
+		t.Fatalf("Retain() error = %v", err)
+	}
+	if res.BodiesCleared != 0 || res.ActionsDeleted != 0 {
+		t.Errorf("disabled policy changed data: %+v", res)
+	}
+	if body, null := bodyOf(t, s, ctx, old); null || body != "an old body" {
+		t.Errorf("old body = %q (null=%v), want unchanged", body, null)
+	}
+}

internal/store/schema.sql

@@ -1,7 +1,8 @@
 -- Email Agent schema (AGENTS.md §7).
 --
 -- Every statement is idempotent (CREATE ... IF NOT EXISTS) so this file can be
--- executed on every start. Retention/capping jobs belong to later phases.
+-- executed on every start. The retention/capping job (AGENTS.md §7) lives in
+-- internal/store/retention.go and runs from the CLI `-retain` flag.
 --
 -- Attribution: created for the Email Agent Phase 1 skeleton.