diff --git a/.env.example b/.env.example index 1d5b2a1..0d7c31c 100644 --- a/.env.example +++ b/.env.example @@ -65,3 +65,9 @@ APPRISE_KEY= APPRISE_URLS= APPRISE_TOKEN= EMAIL_AGENT_SUMMARY_TO=samuelrolfe@gmail.com + +# --- Retention (AGENTS.md §7) ----------------------------------------------- +# `email-agent -retain` clears message bodies older than RETENTION_DAYS +# (metadata stays) and caps the append-only actions_log at MAX_ROWS. +EMAIL_AGENT_RETENTION_DAYS=30 +EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS=50000 diff --git a/README.md b/README.md index 6a9be51..20c8584 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ pre-planning material and is not authoritative. ## Status -**Phase 6 — Unsubscribe and rules editor.** The Go skeleton, session login, +**Phase 7 — Summary and Hub; retention.** The Go skeleton, session login, and HTMX dashboard (Phase 1), real mail over **IMAP with an app password, forced to IPv4** (Phase 2), **JEV classification** (Phase 3), **bulk triage actions** (Phase 4), **reply drafts** (Phase 5), and now **unsubscribe @@ -24,7 +24,14 @@ deliberate UI action with a confirmation — there is no auto-send (§10). The the reply model read. **Unsubscribe detection** (`/unsubscribe`) finds RFC 2369/8058 links and body links, keeps a domain ignore list, and records what Sam marks — it **never follows a link**. Every action, including drafts, sends, -rules edits, and unsubscribe marks, is appended to `actions_log` (FR-13). See +rules edits, and unsubscribe marks, is appended to `actions_log` (FR-13). +**Phase 7** adds the daily **summary** (`-summary`, `-summary-send`; delivered +by email through Apprise) and the read-only **Hub API** (`/capabilities`, +`/ask`, `/notify`). **Retention** (AGENTS.md §7) is enforced by the bounded +`-retain` job: it clears message bodies older than one month, keeps metadata, +and caps the append-only `actions_log`; see +[`docs/retention.md`](docs/retention.md). See +[`docs/phase7-summary-hub.md`](docs/phase7-summary-hub.md) for Phase 7, [`docs/phase6-unsubscribe.md`](docs/phase6-unsubscribe.md) for the phase evidence, [`docs/phase5-replies.md`](docs/phase5-replies.md) for Phase 5, and [`docs/phase4-triage.md`](docs/phase4-triage.md) for Phase 4. @@ -32,14 +39,14 @@ rules edits, and unsubscribe marks, is appended to `actions_log` (FR-13). See ## Layout ``` -cmd/email-agent/ service entry point (`-fetch`, `-classify`, `-classify-all`) +cmd/email-agent/ service entry point (`-fetch`, `-classify`, `-classify-all`, `-retain`) internal/classify/ JEV (TypeSafe System One) client, rule engine, orchestration internal/config/ environment-driven configuration internal/mailfetch/ stdlib-only IMAP client, MIME parsing, FR-2 source address internal/mailops/ Gmail label moves and flag changes, and SMTP reply sending internal/reply/ LLM reply drafts through OmniRoute, reply templates internal/session/ server-side session store + signed cookie values -internal/store/ SQLite open + schema (internal/store/schema.sql) +internal/store/ SQLite open + schema (internal/store/schema.sql), retention internal/server/ HTTP routes, auth middleware, HTMX handlers internal/unsubscribe/ RFC 2369/8058 + body unsubscribe-link detection (no network) internal/version/ build metadata (set via -ldflags) @@ -83,6 +90,8 @@ Key variables: | `OMNIROUTE_BASE_URL` | `http://192.168.20.13:20129/v1` | OmniRoute gateway (all non-JEV models) | | `OMNIROUTE_API_KEY` | _(empty)_ | OmniRoute API key (required to draft replies) | | `OMNIROUTE_MODEL` | `auto/best-chat` | Model alias for reply drafts | +| `EMAIL_AGENT_RETENTION_DAYS` | `30` | Message-body retention window in days (§7) | +| `EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS` | `50000` | Newest `actions_log` rows kept | ## Mail sync (Phase 2) @@ -197,6 +206,30 @@ email-agent -check-smtp # IPv4 + STARTTLS + app-password login, no message sen A real send is a UI action by Sam; it is never triggered by an agent test without explicit approval (§10, AGENTS.md §13). +## Retention (AGENTS.md §7) + +AGENTS.md §7 says: *keep message bodies for one month. Logs are append-only and +size-capped.* The bounded `-retain` job enforces both rules: + +- **Message bodies.** Every message whose `received_at` is older than + `EMAIL_AGENT_RETENTION_DAYS` (default 30) has `body_text` set to `NULL`. All + metadata (sender, subject, source address, classification, folder, labels) + stays. A second run is a no-op. +- **Action log.** Only the newest `EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS` (default + 50000) rows of the append-only `actions_log` are kept; older rows are deleted. + +The job runs in one SQLite transaction, writes one `retain` row to +`actions_log` for the audit trail, and never touches a message inside the +window. It is the CLI `-retain` flag, so it can be scheduled exactly like +`-fetch` and `-classify` and is never run by an HTTP request: + +```sh +# inside the container +docker exec email-agent email-agent -retain +``` + +The decision is recorded in [`docs/retention.md`](docs/retention.md). + ## Run locally ```sh diff --git a/cmd/email-agent/main.go b/cmd/email-agent/main.go index eb345e0..18e2efa 100644 --- a/cmd/email-agent/main.go +++ b/cmd/email-agent/main.go @@ -37,6 +37,7 @@ func main() { summary = flag.Bool("summary", false, "build today's inbox summary, store it, and print it; sends nothing") summarySend = flag.Bool("summary-send", false, "build today's inbox summary and deliver it by email via Apprise") notify = flag.Bool("notify", false, "send one test notification to the Hub (POST /notify) and exit") + retain = flag.Bool("retain", false, "enforce retention (AGENTS.md §7): clear old message bodies and cap actions_log, then exit") ) flag.Parse() @@ -111,6 +112,14 @@ func main() { return } + if *retain { + if err := runRetain(cfg, logger); err != nil { + fmt.Fprintf(os.Stderr, "retain failed: %v\n", err) + os.Exit(1) + } + return + } + if err := run(cfg, logger); err != nil { logger.Error("fatal", "error", err) os.Exit(1) @@ -317,6 +326,44 @@ func runNotify(cfg *config.Config, logger *slog.Logger) error { return nil } +// runRetain enforces the AGENTS.md §7 data rules: message bodies older than the +// configured window are cleared (metadata stays), and the append-only +// actions_log is capped. It is a bounded, explicit job for the n8n schedule; +// the HTTP service never runs it on a request. Running it twice is safe. +func runRetain(cfg *config.Config, logger *slog.Logger) error { + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + st, err := store.Open(ctx, cfg.DBPath) + if err != nil { + return fmt.Errorf("open store: %w", err) + } + defer func() { _ = st.Close() }() + + policy := store.RetentionPolicy{ + BodyMaxAge: time.Duration(cfg.RetentionDays) * 24 * time.Hour, + ActionsLogMaxRows: cfg.ActionsLogMaxRows, + } + res, err := st.Retain(ctx, policy) + if err != nil { + return err + } + if err := st.LogAction(ctx, "retain", "scheduler", + fmt.Sprintf("bodies_cleared=%d actions_deleted=%d retention_days=%d max_rows=%d", + res.BodiesCleared, res.ActionsDeleted, cfg.RetentionDays, cfg.ActionsLogMaxRows)); err != nil { + // Retention already committed; a failed audit row must not fail the job. + logger.Error("log retention action", "error", err) + } + + cutoff := "disabled" + if !res.BodyCutoff.IsZero() { + cutoff = res.BodyCutoff.Format("2006-01-02") + } + fmt.Printf("bodies_cleared=%d body_cutoff=%s actions_deleted=%d actions_max_rows=%d\n", + res.BodiesCleared, cutoff, res.ActionsDeleted, cfg.ActionsLogMaxRows) + return nil +} + // probeHealth requests the local health endpoint. It exists so the container // healthcheck does not need an external HTTP client in the image. func probeHealth(port int) error { diff --git a/docker-compose.yml b/docker-compose.yml index f4b1690..056e150 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -62,6 +62,10 @@ services: APPRISE_URLS: ${APPRISE_URLS:-} APPRISE_TOKEN: ${APPRISE_TOKEN:-} EMAIL_AGENT_SUMMARY_TO: ${EMAIL_AGENT_SUMMARY_TO:-samuelrolfe@gmail.com} + # Retention (AGENTS.md §7): `-retain` clears message bodies older than + # this window and caps the append-only actions_log. + EMAIL_AGENT_RETENTION_DAYS: ${EMAIL_AGENT_RETENTION_DAYS:-30} + EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS: ${EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS:-50000} volumes: - email_agent_data:/data diff --git a/internal/config/config.go b/internal/config/config.go index d9039eb..2a2c261 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -60,6 +60,11 @@ const ( // fans out to email/NTFY/Telegram. SUMMARY_TO is the summary recipient. DefaultAppriseURL = "http://192.168.20.35:8210" DefaultSummaryTo = "samuelrolfe@gmail.com" + + // Retention defaults (AGENTS.md §7): keep message bodies for one month, and + // cap the append-only action log. The retention job is the CLI `-retain`. + DefaultRetentionDays = 30 + DefaultActionsLogMaxRows = 50000 ) // Config is the fully resolved runtime configuration. @@ -153,6 +158,14 @@ type Config struct { AppriseToken string // SummaryTo is the recipient address for the daily summary email. SummaryTo string + + // RetentionDays is the message-body window in days (AGENTS.md §7: one + // month). Bodies of messages received more than this many days ago are + // cleared by the `-retain` job; metadata is kept. + RetentionDays int + // ActionsLogMaxRows caps the append-only actions_log. The oldest rows beyond + // this count are deleted by the `-retain` job. + ActionsLogMaxRows int } // Load reads configuration from the environment, applying defaults. @@ -197,6 +210,8 @@ func Load() (*Config, error) { c.IMAPMaxMessages = getenvDefaultInt("EMAIL_AGENT_IMAP_MAX_MESSAGES", DefaultIMAPMaxMessages) c.ClassifyConcurrency = getenvDefaultInt("EMAIL_AGENT_CLASSIFY_CONCURRENCY", DefaultClassifyConcurrency) c.ClassifyMaxMessages = getenvDefaultInt("EMAIL_AGENT_CLASSIFY_MAX_MESSAGES", DefaultClassifyMaxMessages) + c.RetentionDays = getenvDefaultInt("EMAIL_AGENT_RETENTION_DAYS", DefaultRetentionDays) + c.ActionsLogMaxRows = getenvDefaultInt("EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS", DefaultActionsLogMaxRows) port, err := resolvePort() if err != nil { diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ba2458b..956e057 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -9,6 +9,7 @@ func TestLoadDefaults(t *testing.T) { "EMAIL_AGENT_DB_PATH", "EMAIL_AGENT_LOG_LEVEL", "EMAIL_AGENT_PUBLIC_BASE_URL", "GMAIL_USER", "GMAIL_APP_PASSWORD", "EMAIL_AGENT_IMAP_HOST", "EMAIL_AGENT_IMAP_PORT", "EMAIL_AGENT_IMAP_LOOKBACK_DAYS", "EMAIL_AGENT_IMAP_MAX_MESSAGES", + "EMAIL_AGENT_RETENTION_DAYS", "EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS", } { t.Setenv(k, "") } @@ -46,6 +47,35 @@ func TestLoadDefaults(t *testing.T) { } } +func TestRetentionDefaultsAndOverrides(t *testing.T) { + t.Setenv("EMAIL_AGENT_RETENTION_DAYS", "") + t.Setenv("EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS", "") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if cfg.RetentionDays != DefaultRetentionDays { + t.Errorf("RetentionDays = %d, want %d", cfg.RetentionDays, DefaultRetentionDays) + } + if cfg.ActionsLogMaxRows != DefaultActionsLogMaxRows { + t.Errorf("ActionsLogMaxRows = %d, want %d", cfg.ActionsLogMaxRows, DefaultActionsLogMaxRows) + } + + t.Setenv("EMAIL_AGENT_RETENTION_DAYS", "45") + t.Setenv("EMAIL_AGENT_ACTIONS_LOG_MAX_ROWS", "1234") + cfg, err = Load() + if err != nil { + t.Fatalf("Load() override error = %v", err) + } + if cfg.RetentionDays != 45 { + t.Errorf("RetentionDays = %d, want 45", cfg.RetentionDays) + } + if cfg.ActionsLogMaxRows != 1234 { + t.Errorf("ActionsLogMaxRows = %d, want 1234", cfg.ActionsLogMaxRows) + } +} + func TestGmailConfigured(t *testing.T) { t.Setenv("GMAIL_USER", "samuelrolfe@gmail.com") t.Setenv("GMAIL_APP_PASSWORD", "app-password") diff --git a/internal/store/retention.go b/internal/store/retention.go new file mode 100644 index 0000000..84da690 --- /dev/null +++ b/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 +} diff --git a/internal/store/retention_test.go b/internal/store/retention_test.go new file mode 100644 index 0000000..8e5d198 --- /dev/null +++ b/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) + } +} diff --git a/internal/store/schema.sql b/internal/store/schema.sql index 4cff599..db37645 100644 --- a/internal/store/schema.sql +++ b/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.