From a349b42b1fb9edf3a6a726e7c4c5f52a0914b58b Mon Sep 17 00:00:00 2001 From: Sam Rolfe Date: Sun, 11 Oct 2026 22:38:49 +1100 Subject: [PATCH] Phase 7: daily summary and read-only Hub API (AUD-73) - internal/hub: project side of the frozen ask-contract. GET /capabilities, read-only POST /ask answerer (no LLM, <3s), and the outbound POST /notify client. - internal/apprise: daily-summary delivery through the house Apprise API. - internal/summary: deterministic plain-text daily summary builder. - internal/store: CountByCategory, MessagesByCategory, and summary CRUD. - internal/server: /capabilities and /ask (bearer-token protected), plus the Reports screen (/reports, build, send, delete). - main.go: -summary, -summary-send, and -notify CLI entry points for n8n. Co-Authored-By: Paperclip --- .env.example | 15 ++ cmd/email-agent/main.go | 82 ++++++++++ docker-compose.yml | 12 ++ internal/apprise/apprise.go | 103 ++++++++++++ internal/apprise/apprise_test.go | 91 +++++++++++ internal/config/config.go | 73 ++++++++- internal/config/config_test.go | 52 ++++++ internal/hub/ask.go | 260 ++++++++++++++++++++++++++++++ internal/hub/ask_test.go | 224 ++++++++++++++++++++++++++ internal/hub/hub.go | 93 +++++++++++ internal/hub/notify.go | 104 ++++++++++++ internal/hub/notify_test.go | 87 ++++++++++ internal/server/hub.go | 98 ++++++++++++ internal/server/hub_test.go | 262 +++++++++++++++++++++++++++++++ internal/server/reports.go | 165 +++++++++++++++++++ internal/server/server.go | 23 +++ internal/store/summary.go | 217 +++++++++++++++++++++++++ internal/store/summary_test.go | 106 +++++++++++++ internal/summary/summary.go | 150 ++++++++++++++++++ internal/summary/summary_test.go | 106 +++++++++++++ web/static/app.css | 22 +++ web/templates/dashboard.html | 1 + web/templates/reports.html | 108 +++++++++++++ web/templates/rules.html | 1 + 24 files changed, 2454 insertions(+), 1 deletion(-) create mode 100644 internal/apprise/apprise.go create mode 100644 internal/apprise/apprise_test.go create mode 100644 internal/hub/ask.go create mode 100644 internal/hub/ask_test.go create mode 100644 internal/hub/hub.go create mode 100644 internal/hub/notify.go create mode 100644 internal/hub/notify_test.go create mode 100644 internal/server/hub.go create mode 100644 internal/server/hub_test.go create mode 100644 internal/server/reports.go create mode 100644 internal/store/summary.go create mode 100644 internal/store/summary_test.go create mode 100644 internal/summary/summary.go create mode 100644 internal/summary/summary_test.go create mode 100644 web/templates/reports.html diff --git a/.env.example b/.env.example index fa514a9..1d5b2a1 100644 --- a/.env.example +++ b/.env.example @@ -49,4 +49,19 @@ EMAIL_AGENT_CLASSIFY_MAX_MESSAGES=200 OMNIROUTE_BASE_URL= OMNIROUTE_API_KEY= OMNIROUTE_MODEL=auto/best-chat + +# --- Hub (Phase 7, FR-12) --------------------------------------------------- +# HUB_TOKEN is the shared bearer token the Hub sends to /capabilities and /ask. +# The Hub side uses HUB_PROJECT_EMAIL_AGENT_TOKEN with the same value. HUB_TOKEN= +HUB_BASE_URL=http://192.168.20.13:8788 +HUB_NOTIFY_TIMEOUT=5s + +# --- Daily summary (Phase 7, FR-10) ----------------------------------------- +# Delivery goes through the house Apprise API on .35. Set APPRISE_KEY for a +# stored Apprise configuration, or APPRISE_URLS for the stateless form. +APPRISE_URL=http://192.168.20.35:8210 +APPRISE_KEY= +APPRISE_URLS= +APPRISE_TOKEN= +EMAIL_AGENT_SUMMARY_TO=samuelrolfe@gmail.com diff --git a/cmd/email-agent/main.go b/cmd/email-agent/main.go index eb18a82..eb345e0 100644 --- a/cmd/email-agent/main.go +++ b/cmd/email-agent/main.go @@ -13,12 +13,15 @@ import ( "syscall" "time" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/apprise" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/classify" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/config" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/hub" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/mailfetch" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/mailops" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/server" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/summary" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/version" ) @@ -31,6 +34,9 @@ func main() { classifyAll = flag.Bool("classify-all", false, "re-classify every stored message with JEV and exit") checkSMTP = flag.Bool("check-smtp", false, "verify the Gmail SMTP connection (IPv4 + STARTTLS + login) and exit; sends no mail") checkThread = flag.Int64("check-thread", 0, "verify that a stored message's Gmail thread has a sent reply (read-only), then exit") + 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") ) flag.Parse() @@ -89,6 +95,22 @@ func main() { return } + if *summary || *summarySend { + if err := runSummary(cfg, logger, *summarySend); err != nil { + fmt.Fprintf(os.Stderr, "summary failed: %v\n", err) + os.Exit(1) + } + return + } + + if *notify { + if err := runNotify(cfg, logger); err != nil { + fmt.Fprintf(os.Stderr, "notify failed: %v\n", err) + os.Exit(1) + } + return + } + if err := run(cfg, logger); err != nil { logger.Error("fatal", "error", err) os.Exit(1) @@ -235,6 +257,66 @@ func runCheckThread(cfg *config.Config, logger *slog.Logger, messageID int64) er return nil } +// runSummary builds today's inbox summary (FR-10). With send=false it stores and +// prints the summary and sends nothing; with send=true it also delivers it by +// email through Apprise. The n8n schedule runs the send form once a day. +func runSummary(cfg *config.Config, logger *slog.Logger, send bool) 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() }() + + data, err := summary.New(st).Build(ctx) + if err != nil { + return fmt.Errorf("build summary: %w", err) + } + id, err := st.SaveSummary(ctx, data.Date, data.Text(), "") + if err != nil { + return err + } + fmt.Print(data.Text()) + + if !send { + fmt.Printf("\nsummary stored id=%d (not sent)\n", id) + return nil + } + if !cfg.AppriseConfigured() { + return fmt.Errorf("apprise is not configured: set APPRISE_URL and APPRISE_KEY (or APPRISE_URLS)") + } + if err := apprise.New(cfg.AppriseURL, cfg.AppriseKey, cfg.AppriseURLs, cfg.AppriseToken, 0). + Send(ctx, data.Subject(), data.Text()); err != nil { + return fmt.Errorf("deliver summary: %w", err) + } + if err := st.MarkSummaryDelivered(ctx, id, cfg.SummaryTo); err != nil { + logger.Error("mark summary delivered", "error", err) + } + fmt.Printf("\nsummary delivered id=%d to=%s\n", id, cfg.SummaryTo) + return nil +} + +// runNotify sends one test notification to the Hub's POST /notify. It exists so +// the outbound notify path (ask-contract.md) can be exercised without the UI. +func runNotify(cfg *config.Config, logger *slog.Logger) error { + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + n := hub.NewNotifier(cfg.HubBaseURL, cfg.HubToken, cfg.HubNotifyTimeout, nil) + if err := n.Notify(ctx, hub.NotifyRequest{ + From: hub.ProjectName, + Level: "info", + Message: "Email Agent notify path check.", + Person: "sam", + }); err != nil { + return err + } + fmt.Printf("notify ok: %s/notify\n", cfg.HubBaseURL) + 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 7f5ed9f..f4b1690 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -49,7 +49,19 @@ services: OMNIROUTE_BASE_URL: ${OMNIROUTE_BASE_URL:-} OMNIROUTE_API_KEY: ${OMNIROUTE_API_KEY:-} OMNIROUTE_MODEL: ${OMNIROUTE_MODEL:-auto/best-chat} + # Phase 7 (FR-12): the Hub contract. HUB_TOKEN is the shared bearer token + # the Hub sends; /capabilities and /ask refuse an unsigned request. HUB_TOKEN: ${HUB_TOKEN:-} + HUB_BASE_URL: ${HUB_BASE_URL:-http://192.168.20.13:8788} + HUB_NOTIFY_TIMEOUT: ${HUB_NOTIFY_TIMEOUT:-5s} + # Phase 7 (FR-10): daily summary delivery through the house Apprise API + # on .35. Set APPRISE_KEY (stored config) or APPRISE_URLS (stateless), + # and EMAIL_AGENT_SUMMARY_TO for the recipient. + APPRISE_URL: ${APPRISE_URL:-http://192.168.20.35:8210} + APPRISE_KEY: ${APPRISE_KEY:-} + APPRISE_URLS: ${APPRISE_URLS:-} + APPRISE_TOKEN: ${APPRISE_TOKEN:-} + EMAIL_AGENT_SUMMARY_TO: ${EMAIL_AGENT_SUMMARY_TO:-samuelrolfe@gmail.com} volumes: - email_agent_data:/data diff --git a/internal/apprise/apprise.go b/internal/apprise/apprise.go new file mode 100644 index 0000000..830a4ee --- /dev/null +++ b/internal/apprise/apprise.go @@ -0,0 +1,103 @@ +// Package apprise delivers notifications through the house Apprise API. +// +// Apprise is the shared notification gateway (AGENTS.md §8). The Email Agent +// uses it only to deliver the daily summary by email (FR-10). The client speaks +// the standard Apprise API server shape: +// +// POST {base}/notify/{key} JSON {"title","body","type"} +// POST {base}/notify JSON {"urls","title","body","type"} (stateless) +// +// A key names a stored Apprise configuration. When no key is set the client +// uses the stateless form with an explicit list of Apprise URLs. +package apprise + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +// Client posts a notification to one Apprise API server. +type Client struct { + BaseURL string + // Key names a stored Apprise configuration. Empty means stateless mode. + Key string + // URLs is the stateless destination list, used when Key is empty. + URLs []string + // Token is sent as a bearer token when the Apprise API is protected. + Token string + HTTP *http.Client +} + +// New builds a client. baseURL is the Apprise API root, e.g. +// http://192.168.20.35:8210. A zero timeout defaults to ten seconds. +func New(baseURL, key string, urls []string, token string, timeout time.Duration) *Client { + if timeout <= 0 { + timeout = 10 * time.Second + } + return &Client{ + BaseURL: strings.TrimRight(baseURL, "/"), + Key: strings.TrimSpace(key), + URLs: urls, + Token: token, + HTTP: &http.Client{Timeout: timeout}, + } +} + +// Configured reports whether a delivery can be attempted. +func (c *Client) Configured() bool { + if c == nil || strings.TrimSpace(c.BaseURL) == "" { + return false + } + return c.Key != "" || len(c.URLs) > 0 +} + +// notifyBody is the Apprise API request body. `type` selects the notification +// class; the daily summary is informational. +type notifyBody struct { + URLs []string `json:"urls,omitempty"` + Title string `json:"title,omitempty"` + Body string `json:"body"` + Type string `json:"type,omitempty"` +} + +// Send delivers one notification titled title with the given text body. +func (c *Client) Send(ctx context.Context, title, body string) error { + if !c.Configured() { + return fmt.Errorf("apprise is not configured") + } + payload := notifyBody{Title: title, Body: body, Type: "info"} + url := c.BaseURL + "/notify" + if c.Key != "" { + url += "/" + c.Key + } else { + payload.URLs = c.URLs + } + + raw, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("encode apprise request: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(raw)) + if err != nil { + return fmt.Errorf("build apprise request: %w", err) + } + req.Header.Set("Content-Type", "application/json; charset=utf-8") + if c.Token != "" { + req.Header.Set("Authorization", "Bearer "+c.Token) + } + + resp, err := c.HTTP.Do(req) + if err != nil { + return fmt.Errorf("post apprise: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("post apprise: status %d", resp.StatusCode) + } + return nil +} diff --git a/internal/apprise/apprise_test.go b/internal/apprise/apprise_test.go new file mode 100644 index 0000000..2932f99 --- /dev/null +++ b/internal/apprise/apprise_test.go @@ -0,0 +1,91 @@ +package apprise + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +type captured struct { + path string + auth string + body notifyBody +} + +func TestSendWithKey(t *testing.T) { + var got captured + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got.path = r.URL.Path + got.auth = r.Header.Get("Authorization") + _ = json.NewDecoder(r.Body).Decode(&got.body) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + c := New(srv.URL, "email-summary", nil, "apprise-token", time.Second) + if err := c.Send(context.Background(), "Subject", "Body text"); err != nil { + t.Fatalf("Send() error = %v", err) + } + if got.path != "/notify/email-summary" { + t.Errorf("path = %q, want /notify/email-summary", got.path) + } + if got.auth != "Bearer apprise-token" { + t.Errorf("auth = %q", got.auth) + } + if got.body.Title != "Subject" || got.body.Body != "Body text" { + t.Errorf("body = %+v", got.body) + } + if got.body.Type != "info" { + t.Errorf("type = %q, want info", got.body.Type) + } + if len(got.body.URLs) != 0 { + t.Errorf("urls = %v, want none in key mode", got.body.URLs) + } +} + +func TestSendStatelessURLs(t *testing.T) { + var got captured + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + got.path = r.URL.Path + _ = json.NewDecoder(r.Body).Decode(&got.body) + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + c := New(srv.URL, "", []string{"mailto://user:pass@example.com"}, "", time.Second) + if err := c.Send(context.Background(), "T", "B"); err != nil { + t.Fatalf("Send() error = %v", err) + } + if got.path != "/notify" { + t.Errorf("path = %q, want /notify", got.path) + } + if len(got.body.URLs) != 1 || got.body.URLs[0] != "mailto://user:pass@example.com" { + t.Errorf("urls = %v", got.body.URLs) + } +} + +func TestSendErrorStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "bad", http.StatusUnauthorized) + })) + defer srv.Close() + c := New(srv.URL, "key", nil, "", time.Second) + if err := c.Send(context.Background(), "T", "B"); err == nil { + t.Fatal("expected an error for a 401 reply") + } +} + +func TestConfigured(t *testing.T) { + if New("http://x", "", nil, "", time.Second).Configured() { + t.Error("keyless and urlless client reports configured") + } + if New("http://x", "k", nil, "", time.Second).Configured() != true { + t.Error("key client should report configured") + } + if New("http://x", "", []string{"mailto://a@b"}, "", time.Second).Configured() != true { + t.Error("url client should report configured") + } +} diff --git a/internal/config/config.go b/internal/config/config.go index ddaf99d..d9039eb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -13,6 +13,7 @@ import ( "os" "strconv" "strings" + "time" ) // Defaults for the service. Port 8790 is recorded in AGENTS.md §9. @@ -49,6 +50,16 @@ const ( // apps use the same API endpoint. OMNIROUTE_BASE_URL overrides it. DefaultOmniRouteBaseURL = "http://192.168.20.13:20129/v1" DefaultOmniRouteModel = "auto/best-chat" + + // Hub defaults (Phase 7, FR-12). The central Hub runs on .13:8788. All + // non-JEV model calls and project notifications leave through it. + DefaultHubBaseURL = "http://192.168.20.13:8788" + DefaultHubNotifyTimout = 5 * time.Second + + // Apprise defaults (FR-10). The house Apprise API runs on .35:8210 and + // fans out to email/NTFY/Telegram. SUMMARY_TO is the summary recipient. + DefaultAppriseURL = "http://192.168.20.35:8210" + DefaultSummaryTo = "samuelrolfe@gmail.com" ) // Config is the fully resolved runtime configuration. @@ -121,8 +132,27 @@ type Config struct { // SMTPPort is the Gmail submission port (default 587, STARTTLS). SMTPPort int - // HubToken authenticates the read-only Hub API. Secret. + // HubToken authenticates the read-only Hub API (FR-12). Secret. The Hub + // sends it as a bearer token; this service never trusts an unsigned request. HubToken string + // HubBaseURL is the base URL of the central Hub, used by the outbound + // POST /notify path (ask-contract.md). Not secret. + HubBaseURL string + // HubNotifyTimeout bounds one outbound notify call. + HubNotifyTimeout time.Duration + + // Apprise (FR-10): the daily summary is delivered by email through Apprise. + // AppriseURL is the Apprise API base URL (default the house instance on + // .35). AppriseKey names a stored Apprise configuration. AppriseURLs is the + // stateless fallback used when no key is set. AppriseToken is an optional + // bearer token for a protected Apprise API. All values come from the host + // environment; none are committed. + AppriseURL string + AppriseKey string + AppriseURLs []string + AppriseToken string + // SummaryTo is the recipient address for the daily summary email. + SummaryTo string } // Load reads configuration from the environment, applying defaults. @@ -142,7 +172,14 @@ func Load() (*Config, error) { GmailUser: os.Getenv("GMAIL_USER"), GmailAppPassword: os.Getenv("GMAIL_APP_PASSWORD"), HubToken: os.Getenv("HUB_TOKEN"), + HubBaseURL: strings.TrimRight(getenvDefault("HUB_BASE_URL", DefaultHubBaseURL), "/"), + AppriseURL: strings.TrimRight(getenvDefault("APPRISE_URL", DefaultAppriseURL), "/"), + AppriseKey: strings.TrimSpace(os.Getenv("APPRISE_KEY")), + AppriseURLs: splitList(os.Getenv("APPRISE_URLS")), + AppriseToken: os.Getenv("APPRISE_TOKEN"), + SummaryTo: getenvDefault("EMAIL_AGENT_SUMMARY_TO", DefaultSummaryTo), } + c.HubNotifyTimeout = durationSetting("HUB_NOTIFY_TIMEOUT", DefaultHubNotifyTimout) imapPort, err := resolveIMAPPort() if err != nil { @@ -288,3 +325,37 @@ func getenvDefault(key, def string) string { } return def } + +// splitList parses a comma-separated environment value, trimming blanks. +func splitList(v string) []string { + var out []string + for _, part := range strings.Split(v, ",") { + if p := strings.TrimSpace(part); p != "" { + out = append(out, p) + } + } + return out +} + +// durationSetting reads a Go duration setting, falling back to def when unset +// or malformed. +func durationSetting(key string, def time.Duration) time.Duration { + raw := strings.TrimSpace(os.Getenv(key)) + if raw == "" { + return def + } + d, err := time.ParseDuration(raw) + if err != nil || d <= 0 { + return def + } + return d +} + +// HubConfigured reports whether the Hub API can serve authenticated requests. +func (c *Config) HubConfigured() bool { return strings.TrimSpace(c.HubToken) != "" } + +// AppriseConfigured reports whether a daily summary can be delivered. +func (c *Config) AppriseConfigured() bool { + return strings.TrimSpace(c.AppriseURL) != "" && + (strings.TrimSpace(c.AppriseKey) != "" || len(c.AppriseURLs) > 0) +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go index a3065af..ba2458b 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -109,3 +109,55 @@ func TestLoadInvalidEnv(t *testing.T) { t.Fatal("Load() error = nil, want error for invalid env") } } + +func TestHubAndAppriseConfig(t *testing.T) { + for _, k := range []string{"HUB_TOKEN", "HUB_BASE_URL", "APPRISE_URL", "APPRISE_KEY", "APPRISE_URLS", "EMAIL_AGENT_SUMMARY_TO", "HUB_NOTIFY_TIMEOUT"} { + t.Setenv(k, "") + } + cfg, err := Load() + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if cfg.HubConfigured() { + t.Error("HubConfigured() = true with no HUB_TOKEN") + } + if cfg.HubBaseURL != DefaultHubBaseURL { + t.Errorf("HubBaseURL = %q, want %q", cfg.HubBaseURL, DefaultHubBaseURL) + } + if cfg.AppriseURL != DefaultAppriseURL { + t.Errorf("AppriseURL = %q, want %q", cfg.AppriseURL, DefaultAppriseURL) + } + if cfg.SummaryTo != DefaultSummaryTo { + t.Errorf("SummaryTo = %q, want %q", cfg.SummaryTo, DefaultSummaryTo) + } + if cfg.AppriseConfigured() { + t.Error("AppriseConfigured() = true with no key or URLs") + } + + t.Setenv("HUB_TOKEN", "hub-secret") + t.Setenv("APPRISE_KEY", "email-summary") + cfg, err = Load() + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if !cfg.HubConfigured() { + t.Error("HubConfigured() = false with HUB_TOKEN set") + } + if !cfg.AppriseConfigured() { + t.Error("AppriseConfigured() = false with a key set") + } + + // A stateless URL list also configures Apprise. + t.Setenv("APPRISE_KEY", "") + t.Setenv("APPRISE_URLS", "mailto://a:b@example.com, ntfy://topic") + cfg, err = Load() + if err != nil { + t.Fatalf("Load() error = %v", err) + } + if !cfg.AppriseConfigured() { + t.Error("AppriseConfigured() = false with APPRISE_URLS set") + } + if len(cfg.AppriseURLs) != 2 { + t.Errorf("AppriseURLs = %v, want 2 entries", cfg.AppriseURLs) + } +} diff --git a/internal/hub/ask.go b/internal/hub/ask.go new file mode 100644 index 0000000..b6b6066 --- /dev/null +++ b/internal/hub/ask.go @@ -0,0 +1,260 @@ +package hub + +import ( + "context" + "fmt" + "strings" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" +) + +// ImportantCategories are the categories that mean "needs Sam's attention". They +// come from AGENTS.md §13.2 (action, important). +var ImportantCategories = []string{"Action", "Important"} + +// answerer decisions. /ask is a small read-only question-answering surface, not +// an LLM call: it must answer in under three seconds and never guess. +type Answerer struct { + Reader Reader + // Now is injectable so expiry timestamps are deterministic in tests. + Now func() time.Time +} + +// NewAnswerer builds an Answerer over a read-only Reader. +func NewAnswerer(r Reader) *Answerer { + return &Answerer{Reader: r, Now: func() time.Time { return time.Now().UTC() }} +} + +// now returns the current time, tolerating a nil Now (as in a zero Answerer). +func (a *Answerer) now() time.Time { + if a.Now == nil { + return time.Now().UTC() + } + return a.Now() +} + +// Answer returns a short, speakable answer for one question. It never writes +// state. A question it cannot answer returns answer=null plus a reason. +func (a *Answerer) Answer(ctx context.Context, req AskRequest) AskResponse { + now := a.now() + + asker := strings.ToLower(strings.TrimSpace(req.Asker)) + person := strings.ToLower(strings.TrimSpace(req.Person)) + if person == "" { + person = asker + } + switch { + case asker == "": + return cannotAnswer("no asker was given") + case asker != "sam": + return cannotAnswer("I only answer for Sam") + case person != "sam": + return cannotAnswer("no data for that person") + } + + q := normalizeQuestion(req.Question) + if q == "" { + return cannotAnswer("no question was given") + } + + // The service does not track read state; say so instead of inventing one. + if containsAny(q, "unread", "not read", "mark as read") { + return cannotAnswer("I do not track read or unread mail") + } + + counts, err := a.Reader.CountByCategory(ctx, "inbox") + if err != nil { + return cannotAnswer("the inbox could not be read") + } + byCategory := map[string]int{} + total := 0 + for _, c := range counts { + byCategory[strings.ToLower(strings.TrimSpace(c.Category))] = c.Count + total += c.Count + } + + switch { + case containsAny(q, "important", "urgent", "needs attention", "need attention", + "needs a decision", "needs a reply", "needs action", "action required", + "needs doing", "follow up", "follow-up", "to do", "todo"): + return a.answerImportant(ctx, now, byCategory) + case containsAny(q, "spam", "junk"): + return a.answerSpam(ctx, now, byCategory) + case containsAny(q, "newsletter", "unsubscribe", "subscription", "marketing", "digest"): + return a.answerNewsletter(now, byCategory) + case containsAny(q, "how many", "how much", "count", "number of"): + return a.answerCount(now, total, counts) + case containsAny(q, "summary", "overview", "what's in", "whats in", "what is in", + "anything new", "catch me up", "update me", "anything in my inbox", "anything"): + return a.answerSummary(now, total, byCategory) + default: + return cannotAnswer("I can answer questions about important mail, spam, newsletters, and inbox counts") + } +} + +// answerImportant reports how many inbox messages need attention. +func (a *Answerer) answerImportant(ctx context.Context, now time.Time, byCategory map[string]int) AskResponse { + n := byCategory["action"] + byCategory["important"] + resp := AskResponse{Expires: expiresIn(now, 10*time.Minute)} + if n == 0 { + resp.Answer = answerText("Nothing important is waiting in your inbox.") + resp.Data = map[string]any{"count": 0} + return resp + } + + items, err := a.Reader.MessagesByCategory(ctx, "inbox", ImportantCategories, 6) + if err != nil { + return cannotAnswer("the inbox could not be read") + } + + switch { + case n == 1 && len(items) > 0: + resp.Answer = answerText(fmt.Sprintf("One email needs your attention: %q from %s.", + truncate(items[0].Subject, 60), senderName(items[0].FromAddr))) + case len(items) > 0: + resp.Answer = answerText(fmt.Sprintf("%d emails need your attention, including %q from %s.", + n, truncate(items[0].Subject, 60), senderName(items[0].FromAddr))) + default: + resp.Answer = answerText(fmt.Sprintf("%d emails need your attention.", n)) + } + resp.Data = map[string]any{ + "count": n, + "messages": messageData(items), + } + return resp +} + +// answerSpam reports spam. Spam is always recommended, never auto-deleted +// (AGENTS.md §10.5), so the wording stays neutral. +func (a *Answerer) answerSpam(ctx context.Context, now time.Time, byCategory map[string]int) AskResponse { + n := byCategory["spam"] + resp := AskResponse{Expires: expiresIn(now, 10*time.Minute)} + if n == 0 { + resp.Answer = answerText("There is no spam in your inbox right now.") + resp.Data = map[string]any{"count": 0} + return resp + } + items, err := a.Reader.MessagesByCategory(ctx, "inbox", []string{"Spam"}, 6) + if err != nil { + return cannotAnswer("the inbox could not be read") + } + resp.Answer = answerText(fmt.Sprintf("%d spam emails are in your inbox.", n)) + resp.Data = map[string]any{"count": n, "messages": messageData(items)} + return resp +} + +func (a *Answerer) answerNewsletter(now time.Time, byCategory map[string]int) AskResponse { + n := byCategory["newsletter"] + answer := fmt.Sprintf("Your inbox has %d newsletters.", n) + if n == 1 { + answer = "Your inbox has 1 newsletter." + } + return AskResponse{ + Answer: answerText(answer), + Data: map[string]any{"count": n}, + Expires: expiresIn(now, 10*time.Minute), + } +} + +func (a *Answerer) answerCount(now time.Time, total int, counts []store.CategoryCount) AskResponse { + return AskResponse{ + Answer: answerText(fmt.Sprintf("Your inbox has %d emails.", total)), + Data: map[string]any{"total": total, "by_category": categoryMap(counts)}, + Expires: expiresIn(now, 10*time.Minute), + } +} + +func (a *Answerer) answerSummary(now time.Time, total int, byCategory map[string]int) AskResponse { + needs := byCategory["action"] + byCategory["important"] + news := byCategory["newsletter"] + answer := fmt.Sprintf("Your inbox has %d emails: %d need attention and %d are newsletters.", + total, needs, news) + return AskResponse{ + Answer: answerText(answer), + Data: map[string]any{ + "total": total, + "needs_action": needs, + "newsletters": news, + "spam": byCategory["spam"], + }, + Expires: expiresIn(now, 10*time.Minute), + } +} + +// messageData projects a message list into the compact `data` payload the Hub +// may ignore (ask-contract.md). +func messageData(items []store.MessageListItem) []map[string]any { + out := make([]map[string]any, 0, len(items)) + for _, m := range items { + out = append(out, map[string]any{ + "id": m.ID, + "from": m.FromAddr, + "subject": m.Subject, + "category": m.Category, + "importance": m.Importance, + "action": m.Action, + }) + } + return out +} + +// categoryMap turns the count rows into a JSON-friendly map keyed by the +// category name. The unclassified bucket uses the name "unclassified". +func categoryMap(counts []store.CategoryCount) map[string]int { + out := make(map[string]int, len(counts)) + for _, c := range counts { + name := strings.ToLower(strings.TrimSpace(c.Category)) + if name == "" { + name = "unclassified" + } + out[name] = c.Count + } + return out +} + +// normalizeQuestion lowercases and collapses whitespace so intent matching is +// stable. +func normalizeQuestion(q string) string { + return strings.Join(strings.Fields(strings.ToLower(q)), " ") +} + +func containsAny(q string, subs ...string) bool { + for _, s := range subs { + if strings.Contains(q, s) { + return true + } + } + return false +} + +// senderName extracts a display name from a From header, falling back to the +// address or the raw value. +func senderName(addr string) string { + addr = strings.TrimSpace(addr) + if i := strings.Index(addr, "<"); i > 0 { + name := strings.Trim(strings.TrimSpace(addr[:i]), `"'`) + if name != "" { + return name + } + } + if i := strings.Index(addr, "<"); i >= 0 { + if j := strings.Index(addr[i:], ">"); j > i { + return strings.TrimSpace(addr[i+1 : i+j]) + } + } + return addr +} + +// truncate shortens s to at most n runes so a spoken answer stays short. +func truncate(s string, n int) string { + s = strings.TrimSpace(s) + if s == "" { + return "(no subject)" + } + r := []rune(s) + if len(r) <= n { + return s + } + return strings.TrimSpace(string(r[:n-1])) + "…" +} diff --git a/internal/hub/ask_test.go b/internal/hub/ask_test.go new file mode 100644 index 0000000..02dc8b9 --- /dev/null +++ b/internal/hub/ask_test.go @@ -0,0 +1,224 @@ +package hub + +import ( + "context" + "errors" + "strings" + "testing" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" +) + +// fakeReader is a read-only in-memory Reader. It deliberately has no write +// method: /ask cannot change state through this interface. +type fakeReader struct { + counts []store.CategoryCount + items []store.MessageListItem + countErr error + msgErr error + countCalls int + msgCalls int +} + +func (f *fakeReader) CountByCategory(_ context.Context, _ string) ([]store.CategoryCount, error) { + f.countCalls++ + return f.counts, f.countErr +} + +func (f *fakeReader) MessagesByCategory(_ context.Context, _ string, _ []string, _ int) ([]store.MessageListItem, error) { + f.msgCalls++ + return f.items, f.msgErr +} + +func fixedNow() time.Time { + return time.Date(2026, 10, 11, 8, 0, 0, 0, time.UTC) +} + +func newTestAnswerer(r Reader) *Answerer { + a := NewAnswerer(r) + a.Now = fixedNow + return a +} + +func TestAnswerImportant(t *testing.T) { + r := &fakeReader{ + counts: []store.CategoryCount{ + {Category: "Action", Count: 2}, + {Category: "Important", Count: 1}, + {Category: "Newsletter", Count: 40}, + }, + items: []store.MessageListItem{{ + Message: store.Message{ID: 96, FromAddr: "Google ", Subject: "Security alert"}, + Category: "Action", + Importance: "high", + }}, + } + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{ + Question: "anything important in my inbox?", + Asker: "sam", + }) + + if resp.Answer == nil { + t.Fatalf("Answer = nil, reason = %q", resp.Reason) + } + if !strings.Contains(*resp.Answer, "3 emails need your attention") { + t.Errorf("answer = %q, want it to mention 3 emails", *resp.Answer) + } + if !strings.Contains(*resp.Answer, "Security alert") { + t.Errorf("answer = %q, want the top subject", *resp.Answer) + } + if got := resp.Data["count"]; got != 3 { + t.Errorf("data.count = %v, want 3", got) + } + if resp.Expires == "" { + t.Error("answered response has no expiry") + } +} + +func TestAnswerImportantNone(t *testing.T) { + r := &fakeReader{counts: []store.CategoryCount{{Category: "Info", Count: 4}}} + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{ + Question: "is anything urgent?", + Asker: "sam", + }) + if resp.Answer == nil || !strings.Contains(*resp.Answer, "Nothing important") { + t.Fatalf("answer = %v, want the empty-inbox message", resp.Answer) + } +} + +func TestAnswerCannotAnswerForOtherPerson(t *testing.T) { + r := &fakeReader{} + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{ + Question: "anything important?", + Asker: "sam", + Person: "harry", + }) + if resp.Answer != nil { + t.Fatalf("Answer = %v, want nil", *resp.Answer) + } + if !strings.Contains(resp.Reason, "no data") { + t.Errorf("reason = %q", resp.Reason) + } +} + +func TestAnswerRequiresAsker(t *testing.T) { + r := &fakeReader{} + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{Question: "anything important?"}) + if resp.Answer != nil || resp.Reason == "" { + t.Fatalf("answer = %v reason = %q, want a reason", resp.Answer, resp.Reason) + } +} + +func TestAnswerDoesNotTrackUnread(t *testing.T) { + r := &fakeReader{} + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{ + Question: "how many unread emails do I have?", + Asker: "sam", + }) + if resp.Answer != nil { + t.Fatalf("Answer = %v, want nil", *resp.Answer) + } + if !strings.Contains(resp.Reason, "read") { + t.Errorf("reason = %q, want it to mention read state", resp.Reason) + } +} + +func TestAnswerSpamAndNewsletter(t *testing.T) { + r := &fakeReader{ + counts: []store.CategoryCount{ + {Category: "Spam", Count: 1}, + {Category: "Newsletter", Count: 12}, + }, + items: []store.MessageListItem{{ + Message: store.Message{ID: 5, FromAddr: "spammer@example.com", Subject: "Cheap pills"}, + Category: "Spam", + }}, + } + a := newTestAnswerer(r) + spam := a.Answer(context.Background(), AskRequest{Question: "any spam?", Asker: "sam"}) + if spam.Answer == nil || !strings.Contains(*spam.Answer, "1 spam") { + t.Errorf("spam answer = %v", spam.Answer) + } + news := a.Answer(context.Background(), AskRequest{Question: "how many newsletters?", Asker: "sam"}) + if news.Answer == nil || !strings.Contains(*news.Answer, "12 newsletters") { + t.Errorf("newsletter answer = %v", news.Answer) + } +} + +func TestAnswerCountAndSummary(t *testing.T) { + r := &fakeReader{counts: []store.CategoryCount{ + {Category: "Action", Count: 19}, + {Category: "Important", Count: 1}, + {Category: "Newsletter", Count: 40}, + {Category: "Auto-reply", Count: 34}, + {Category: "Info", Count: 7}, + {Category: "Spam", Count: 1}, + }} + a := newTestAnswerer(r) + count := a.Answer(context.Background(), AskRequest{Question: "how many emails are in my inbox?", Asker: "sam"}) + if count.Answer == nil || !strings.Contains(*count.Answer, "102 emails") { + t.Errorf("count answer = %v", count.Answer) + } + sum := a.Answer(context.Background(), AskRequest{Question: "give me an overview", Asker: "sam"}) + if sum.Answer == nil || !strings.Contains(*sum.Answer, "102 emails") { + t.Errorf("summary answer = %v", sum.Answer) + } +} + +func TestAnswerUnknownQuestion(t *testing.T) { + r := &fakeReader{} + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{ + Question: "what time is it?", + Asker: "sam", + }) + if resp.Answer != nil || resp.Reason == "" { + t.Fatalf("answer = %v reason = %q, want a cannot-answer reply", resp.Answer, resp.Reason) + } +} + +func TestAnswerReadErrorIsReported(t *testing.T) { + r := &fakeReader{countErr: errors.New("boom")} + resp := newTestAnswerer(r).Answer(context.Background(), AskRequest{Question: "anything important?", Asker: "sam"}) + if resp.Answer != nil || !strings.Contains(resp.Reason, "could not be read") { + t.Fatalf("answer = %v reason = %q", resp.Answer, resp.Reason) + } +} + +func TestSenderName(t *testing.T) { + cases := map[string]string{ + "Google ": "Google", + `"api@minimaxi.com" `: "api@minimaxi.com", + "": "no-reply@mail.minimax.io", + "plain@example.com": "plain@example.com", + } + for in, want := range cases { + if got := senderName(in); got != want { + t.Errorf("senderName(%q) = %q, want %q", in, got, want) + } + } +} + +func TestDefaultCapabilities(t *testing.T) { + c := DefaultCapabilities() + if c.Project != ProjectName { + t.Errorf("project = %q, want %q", c.Project, ProjectName) + } + for _, topic := range []string{"email", "inbox", "mail"} { + if !containsString(c.Topics, topic) { + t.Errorf("topics missing %q", topic) + } + } + if containsString(c.Topics, "unread") { + t.Error("topics must not claim read/unread support") + } +} + +func containsString(list []string, v string) bool { + for _, x := range list { + if x == v { + return true + } + } + return false +} diff --git a/internal/hub/hub.go b/internal/hub/hub.go new file mode 100644 index 0000000..54e7411 --- /dev/null +++ b/internal/hub/hub.go @@ -0,0 +1,93 @@ +// Package hub implements the project side of the Hub contract +// (central_ai_llm_master/docs/ask-contract.md) and the outbound notify call. +// +// The contract is one-directional by design: the Hub asks a project about its +// own data and the project answers. A project never learns the Hub's data model +// and the Hub never learns the project's. The project endpoints are +// GET /capabilities and POST /ask (read only); POST /notify is the +// project-to-Hub direction. +// +// Phase 7 (FR-12). Answers are computed from SQLite; /ask performs no writes. +package hub + +import ( + "context" + "encoding/json" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" +) + +// ProjectName is the value the Hub matches in routing.md. +const ProjectName = "email_agent" + +// Capabilities is the GET /capabilities reply. The Hub reads it to route a +// question to this project. Keep the topic list short and plain; it holds no +// private data. +type Capabilities struct { + Project string `json:"project"` + Topics []string `json:"topics"` + People []string `json:"people"` +} + +// DefaultCapabilities returns the topics this service can answer. "email", +// "inbox", "mail", and "message" match the routing.md topic map. Read/unread is +// deliberately absent: the service does not track read state. +func DefaultCapabilities() Capabilities { + return Capabilities{ + Project: ProjectName, + Topics: []string{"email", "inbox", "mail", "message"}, + People: []string{"sam"}, + } +} + +// AskRequest is the POST /ask body (ask-contract.md). +type AskRequest struct { + Question string `json:"question"` + Asker string `json:"asker"` + Person string `json:"person,omitempty"` + Source string `json:"source,omitempty"` +} + +// AskResponse is the POST /ask reply. A nil Answer means the project cannot +// answer and Reason explains why. The shape is frozen by the contract. +type AskResponse struct { + Answer *string `json:"answer"` + Data map[string]any `json:"data,omitempty"` + Expires string `json:"expires,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// answerText is a small helper so the pointer handling stays in one place. +func answerText(s string) *string { return &s } + +// cannotAnswer builds the contract's "cannot answer" reply. +func cannotAnswer(reason string) AskResponse { + return AskResponse{Answer: nil, Reason: reason} +} + +// expiresIn formats an RFC3339 expiry used by data-bearing answers. An inbox +// answer goes stale quickly because mail arrives and is triaged. +func expiresIn(now time.Time, d time.Duration) string { + return now.UTC().Add(d).Format(time.RFC3339) +} + +// NotifyRequest is the project-to-Hub POST /notify body (ask-contract.md). The +// Hub may use it to speak or notify. +type NotifyRequest struct { + From string `json:"from"` + Level string `json:"level"` + Message string `json:"message"` + Person string `json:"person,omitempty"` +} + +// jsonBody marshals a notify payload, kept small and deterministic for tests. +func jsonBody(v any) ([]byte, error) { return json.Marshal(v) } + +// Reader is the read-only data surface the answerer uses. *store.Store +// satisfies it; tests can supply a fake. Keeping it an interface makes the +// read-only guarantee explicit and easy to assert. +type Reader interface { + CountByCategory(ctx context.Context, folder string) ([]store.CategoryCount, error) + MessagesByCategory(ctx context.Context, folder string, categories []string, limit int) ([]store.MessageListItem, error) +} diff --git a/internal/hub/notify.go b/internal/hub/notify.go new file mode 100644 index 0000000..9c8efdb --- /dev/null +++ b/internal/hub/notify.go @@ -0,0 +1,104 @@ +// Outbound notify path (ask-contract.md, "Notify (project to Hub)"). +// +// A project may tell the Hub something so the Hub can speak or notify. This is +// the one call this service makes into the Hub. It is best-effort: the Hub's +// /notify receiver is not part of its frozen v1 routes (docs/hub-api.md §12), +// so a failure is logged and never blocks the local work. +package hub + +import ( + "bytes" + "context" + "fmt" + "net/http" + "strings" + "time" +) + +// Notifier posts project notifications to the Hub. The nil zero value is not +// usable; build one with NewNotifier. +type Notifier struct { + BaseURL string + Token string + HTTP *http.Client +} + +// NewNotifier builds a notifier for the Hub at baseURL. timeout defaults to +// five seconds when it is zero or negative. +func NewNotifier(baseURL, token string, timeout time.Duration, httpClient *http.Client) *Notifier { + if timeout <= 0 { + timeout = 5 * time.Second + } + if httpClient == nil { + httpClient = &http.Client{Timeout: timeout} + } + return &Notifier{ + BaseURL: strings.TrimRight(baseURL, "/"), + Token: token, + HTTP: httpClient, + } +} + +// Configured reports whether the outbound notify path can be attempted. +func (n *Notifier) Configured() bool { + return n != nil && strings.TrimSpace(n.BaseURL) != "" +} + +// Notify posts one notification. A nil notifier is a no-op so callers do not +// need to branch on configuration. +func (n *Notifier) Notify(ctx context.Context, req NotifyRequest) error { + if !n.Configured() { + return fmt.Errorf("hub notify is not configured") + } + if strings.TrimSpace(req.From) == "" { + req.From = ProjectName + } + body, err := jsonBody(req) + if err != nil { + return fmt.Errorf("encode notify: %w", err) + } + url := n.BaseURL + "/notify" + httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return fmt.Errorf("build notify request: %w", err) + } + httpReq.Header.Set("Content-Type", "application/json; charset=utf-8") + if n.Token != "" { + httpReq.Header.Set("Authorization", "Bearer "+n.Token) + } + + resp, err := n.HTTP.Do(httpReq) + if err != nil { + return fmt.Errorf("post notify: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return fmt.Errorf("post notify: status %d", resp.StatusCode) + } + return nil +} + +// NotifySummaryReady is the convenience call used after a daily summary is +// built. +func (n *Notifier) NotifySummaryReady(ctx context.Context, message string) error { + return n.Notify(ctx, NotifyRequest{ + From: ProjectName, + Level: "info", + Message: message, + Person: "sam", + }) +} + +// NotifyImportantCount tells the Hub how many messages need attention. It is +// used by the daily summary job when the count is non-zero. +func (n *Notifier) NotifyImportantCount(ctx context.Context, n2 int) error { + if n2 <= 0 { + return nil + } + return n.Notify(ctx, NotifyRequest{ + From: ProjectName, + Level: "important", + Message: fmt.Sprintf("%d emails need a decision.", n2), + Person: "sam", + }) +} diff --git a/internal/hub/notify_test.go b/internal/hub/notify_test.go new file mode 100644 index 0000000..89ad140 --- /dev/null +++ b/internal/hub/notify_test.go @@ -0,0 +1,87 @@ +package hub + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +func TestNotifyPostsContractPayload(t *testing.T) { + var ( + gotReq NotifyRequest + gotAuth string + hits int + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + hits++ + if r.URL.Path != "/notify" { + t.Errorf("path = %q, want /notify", r.URL.Path) + } + gotAuth = r.Header.Get("Authorization") + if err := json.NewDecoder(r.Body).Decode(&gotReq); err != nil { + t.Errorf("decode body: %v", err) + } + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + n := NewNotifier(srv.URL, "hub-token", time.Second, nil) + if err := n.NotifyImportantCount(context.Background(), 3); err != nil { + t.Fatalf("NotifyImportantCount() error = %v", err) + } + if hits != 1 { + t.Fatalf("server hits = %d, want 1", hits) + } + if gotAuth != "Bearer hub-token" { + t.Errorf("Authorization = %q", gotAuth) + } + if gotReq.From != ProjectName { + t.Errorf("from = %q, want %q", gotReq.From, ProjectName) + } + if gotReq.Level != "important" { + t.Errorf("level = %q, want important", gotReq.Level) + } + if gotReq.Person != "sam" { + t.Errorf("person = %q, want sam", gotReq.Person) + } + if gotReq.Message != "3 emails need a decision." { + t.Errorf("message = %q", gotReq.Message) + } +} + +func TestNotifyImportantCountZeroIsNoop(t *testing.T) { + hits := 0 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { hits++ })) + defer srv.Close() + n := NewNotifier(srv.URL, "", time.Second, nil) + if err := n.NotifyImportantCount(context.Background(), 0); err != nil { + t.Fatalf("error = %v", err) + } + if hits != 0 { + t.Errorf("server hits = %d, want 0", hits) + } +} + +func TestNotifyErrorStatus(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "nope", http.StatusNotFound) + })) + defer srv.Close() + n := NewNotifier(srv.URL, "", time.Second, nil) + if err := n.NotifySummaryReady(context.Background(), "x"); err == nil { + t.Fatal("expected an error for a 404 reply") + } +} + +func TestNotifierNotConfigured(t *testing.T) { + n := NewNotifier("", "", time.Second, nil) + if n.Configured() { + t.Error("empty notifier reports configured") + } + if err := n.NotifySummaryReady(context.Background(), "x"); err == nil { + t.Fatal("expected an error when the notifier is not configured") + } +} diff --git a/internal/server/hub.go b/internal/server/hub.go new file mode 100644 index 0000000..c6b16ba --- /dev/null +++ b/internal/server/hub.go @@ -0,0 +1,98 @@ +// Hub API for Phase 7 (FR-12). +// +// These are the project side of the frozen contract in +// central_ai_llm_master/docs/ask-contract.md: +// +// GET /capabilities -> the topics this project answers (Hub routing) +// POST /ask -> answer a question about this project's data +// +// Both endpoints require the shared bearer token. /ask is read-only: it calls +// only store read methods and writes nothing, including no actions_log row. +// This is a deliberate, testable property (see TestAskWritesNothing). +package server + +import ( + "context" + "crypto/subtle" + "encoding/json" + "net/http" + "strings" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/hub" +) + +// hubAuthRequired reports whether the Hub API has a token configured. With no +// token the endpoints refuse to serve rather than trust an unsigned request. +func (s *Server) hubAuthRequired(w http.ResponseWriter) bool { + if !s.cfg.HubConfigured() { + writeJSON(w, http.StatusServiceUnavailable, map[string]any{ + "answer": nil, + "reason": "the Hub API is not configured", + }) + return false + } + return true +} + +// requireHub enforces the project bearer token on the Hub endpoints. +func (s *Server) requireHub(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if !s.hubAuthRequired(w) { + return + } + if subtle.ConstantTimeCompare([]byte(bearerToken(r)), []byte(s.cfg.HubToken)) != 1 { + writeJSON(w, http.StatusUnauthorized, map[string]any{ + "answer": nil, + "reason": "unauthorized", + }) + return + } + next(w, r) + } +} + +// bearerToken reads the project token from the Authorization header. +func bearerToken(r *http.Request) string { + h := strings.TrimSpace(r.Header.Get("Authorization")) + if h == "" { + return "" + } + if len(h) > 7 && strings.EqualFold(h[:7], "Bearer ") { + return strings.TrimSpace(h[7:]) + } + return h +} + +// handleCapabilities returns the routing topics for the Hub. It holds no +// private data. +func (s *Server) handleCapabilities(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, hub.DefaultCapabilities()) +} + +// handleAsk answers one question from this project's read-only data. The reply +// is always HTTP 200 with either an answer or a reason, so the Hub can relay a +// "cannot answer" without treating the project as down. +func (s *Server) handleAsk(w http.ResponseWriter, r *http.Request) { + ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second) + defer cancel() + + var req hub.AskRequest + dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 64<<10)) + if err := dec.Decode(&req); err != nil { + writeJSON(w, http.StatusOK, hub.AskResponse{ + Answer: nil, + Reason: "the request body is not valid JSON", + }) + return + } + + resp := s.answerer.Answer(ctx, req) + s.logger.Debug("hub ask", + "asker", req.Asker, + "person", req.Person, + "source", req.Source, + "answered", resp.Answer != nil, + ) + writeJSON(w, http.StatusOK, resp) +} diff --git a/internal/server/hub_test.go b/internal/server/hub_test.go new file mode 100644 index 0000000..01d345e --- /dev/null +++ b/internal/server/hub_test.go @@ -0,0 +1,262 @@ +package server + +import ( + "context" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "net/url" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/config" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" +) + +const testHubToken = "test-hub-token" + +// newServerWithConfig builds a server from an explicit config so a test can set +// the Hub token and the Apprise URL. +func newServerWithConfig(t *testing.T, cfg *config.Config) *Server { + t.Helper() + st, err := store.Open(context.Background(), cfg.DBPath) + if err != nil { + t.Fatalf("store.Open() error = %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + srv, err := New(cfg, st, slog.New(slog.DiscardHandler)) + if err != nil { + t.Fatalf("New() error = %v", err) + } + return srv +} + +func hubTestConfig(t *testing.T) *config.Config { + t.Helper() + return &config.Config{ + Env: "test", + Addr: ":0", + DBPath: filepath.Join(t.TempDir(), "email_agent.db"), + LogLevel: "error", + SessionSecret: "test-session-secret", + Password: testPassword, + HubToken: testHubToken, + HubBaseURL: "http://127.0.0.1:0", + AppriseURL: "http://127.0.0.1:0", + AppriseKey: "test-key", + SummaryTo: "sam@example.com", + } +} + +func TestCapabilities(t *testing.T) { + srv := newServerWithConfig(t, hubTestConfig(t)) + + // Without a token the endpoint refuses. + rec := httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/capabilities", nil)) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("unauthenticated status = %d, want 401", rec.Code) + } + + req := httptest.NewRequest(http.MethodGet, "/capabilities", nil) + req.Header.Set("Authorization", "Bearer "+testHubToken) + rec = httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 (body %s)", rec.Code, rec.Body.String()) + } + var body struct { + Project string `json:"project"` + Topics []string `json:"topics"` + People []string `json:"people"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil { + t.Fatalf("decode: %v", err) + } + if body.Project != "email_agent" { + t.Errorf("project = %q", body.Project) + } + for _, topic := range []string{"email", "inbox", "mail"} { + if !containsString(body.Topics, topic) { + t.Errorf("topics missing %q: %v", topic, body.Topics) + } + } +} + +func TestAskRequiresToken(t *testing.T) { + srv := newServerWithConfig(t, hubTestConfig(t)) + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/ask", strings.NewReader(`{"question":"anything important?","asker":"sam"}`)) + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Fatalf("status = %d, want 401", rec.Code) + } +} + +func TestAskAnswersImportant(t *testing.T) { + srv := newServerWithConfig(t, hubTestConfig(t)) + seedHubMessage(t, srv.store, "a1", "Google ", "Security alert", "Action", "high") + seedHubMessage(t, srv.store, "a2", "Store ", "Sale", "Newsletter", "low") + + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/ask", + strings.NewReader(`{"question":"anything important in my inbox?","asker":"sam"}`)) + req.Header.Set("Authorization", "Bearer "+testHubToken) + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 (body %s)", rec.Code, rec.Body.String()) + } + var resp struct { + Answer *string `json:"answer"` + Reason string `json:"reason"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode: %v", err) + } + if resp.Answer == nil || !strings.Contains(*resp.Answer, "needs your attention") { + t.Fatalf("answer = %v reason = %q", resp.Answer, resp.Reason) + } +} + +// TestAskWritesNothing proves the frozen read-only property of /ask: repeated +// calls leave the message, classification, and action-log row counts unchanged. +func TestAskWritesNothing(t *testing.T) { + srv := newServerWithConfig(t, hubTestConfig(t)) + seedHubMessage(t, srv.store, "w1", "Google ", "Security alert", "Action", "high") + seedHubMessage(t, srv.store, "w2", "Store ", "Sale", "Newsletter", "low") + + before := tableCounts(t, srv) + + questions := []string{ + `{"question":"anything important in my inbox?","asker":"sam"}`, + `{"question":"how many emails?","asker":"sam"}`, + `{"question":"any spam?","asker":"sam"}`, + `{"question":"what time is it?","asker":"sam"}`, + } + for _, body := range questions { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/ask", strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+testHubToken) + srv.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d for %s", rec.Code, body) + } + } + + after := tableCounts(t, srv) + for _, table := range []string{"messages", "classifications", "actions_log", "summaries"} { + if before[table] != after[table] { + t.Errorf("%s row count changed: %d -> %d (an /ask call wrote state)", table, before[table], after[table]) + } + } +} + +func TestReportsGenerateStoresWithoutSending(t *testing.T) { + srv := newServerWithConfig(t, hubTestConfig(t)) + seedHubMessage(t, srv.store, "r1", "Google ", "Security alert", "Action", "high") + sessionCookie, csrfCookie, csrf := login(t, srv) + + rec := postForm(t, srv, "/reports/generate", sessionCookie, csrfCookie, url.Values{"csrf": {csrf}}) + if rec.Code != http.StatusSeeOther { + t.Fatalf("status = %d, want 303 (body %s)", rec.Code, rec.Body.String()) + } + summaries, err := srv.store.ListSummaries(context.Background(), 10) + if err != nil { + t.Fatalf("ListSummaries() error = %v", err) + } + if len(summaries) != 1 { + t.Fatalf("summaries = %d, want 1", len(summaries)) + } + if summaries[0].DeliveredTo != "" { + t.Errorf("delivered_to = %q, want empty (generate must not send)", summaries[0].DeliveredTo) + } +} + +func TestReportsSendDeliversThroughApprise(t *testing.T) { + var ( + appriseHits int + apprisePath string + ) + appriseSrv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + appriseHits++ + apprisePath = r.URL.Path + w.WriteHeader(http.StatusOK) + })) + defer appriseSrv.Close() + + cfg := hubTestConfig(t) + cfg.AppriseURL = appriseSrv.URL + cfg.AppriseKey = "summary" + srv := newServerWithConfig(t, cfg) + seedHubMessage(t, srv.store, "s1", "Google ", "Security alert", "Action", "high") + sessionCookie, csrfCookie, csrf := login(t, srv) + + rec := postForm(t, srv, "/reports/send", sessionCookie, csrfCookie, url.Values{"csrf": {csrf}}) + if rec.Code != http.StatusSeeOther { + t.Fatalf("status = %d, want 303 (body %s)", rec.Code, rec.Body.String()) + } + if appriseHits != 1 { + t.Fatalf("apprise hits = %d, want 1", appriseHits) + } + if apprisePath != "/notify/summary" { + t.Errorf("apprise path = %q, want /notify/summary", apprisePath) + } + summaries, _ := srv.store.ListSummaries(context.Background(), 10) + if len(summaries) != 1 || summaries[0].DeliveredTo != "sam@example.com" { + t.Errorf("summaries = %+v, want one delivered to sam@example.com", summaries) + } +} + +func TestReportsRequireLogin(t *testing.T) { + srv := newServerWithConfig(t, hubTestConfig(t)) + rec := httptest.NewRecorder() + srv.Handler().ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/reports", nil)) + if rec.Code != http.StatusSeeOther { + t.Fatalf("status = %d, want a redirect to login", rec.Code) + } +} + +func seedHubMessage(t *testing.T, st *store.Store, gmailID, from, subject, category, importance string) { + t.Helper() + id, err := st.UpsertMessage(context.Background(), &store.Message{ + GmailID: gmailID, + FromAddr: from, + Subject: subject, + DateSent: time.Now().UTC(), + }) + if err != nil { + t.Fatalf("seed message: %v", err) + } + if _, err := st.SaveClassification(context.Background(), store.Classification{ + MessageID: id, Category: category, Importance: importance, + }); err != nil { + t.Fatalf("classify: %v", err) + } +} + +// tableCounts reads row counts for the tables /ask must never touch. +func tableCounts(t *testing.T, srv *Server) map[string]int { + t.Helper() + out := map[string]int{} + for _, table := range []string{"messages", "classifications", "actions_log", "summaries"} { + var n int + if err := srv.store.DB().QueryRowContext(context.Background(), + "SELECT COUNT(*) FROM "+table).Scan(&n); err != nil { + t.Fatalf("count %s: %v", table, err) + } + out[table] = n + } + return out +} + +func containsString(list []string, v string) bool { + for _, x := range list { + if x == v { + return true + } + } + return false +} diff --git a/internal/server/reports.go b/internal/server/reports.go new file mode 100644 index 0000000..b407ea7 --- /dev/null +++ b/internal/server/reports.go @@ -0,0 +1,165 @@ +// Reports screen and daily-summary delivery for Phase 7 (FR-10). +// +// The Reports screen (AGENTS.md §5) lists stored daily summaries, can build +// today's summary on demand, can deliver it by email through Apprise, and can +// delete a stored report. Building never sends mail; only the explicit "Send +// now" action does, and only when Apprise is configured. +package server + +import ( + "net/http" + "net/url" + "strconv" + "strings" +) + +type reportRow struct { + ID int64 + Date string + Text string + Delivered string + CreatedAt string +} + +type reportsData struct { + Summaries []reportRow + Note string + Error string + CSRFToken string + AppriseReady bool + HubReady bool +} + +// handleReports renders the list of stored summaries. +func (s *Server) handleReports(w http.ResponseWriter, r *http.Request) { + s.renderReports(w, r, http.StatusOK, r.URL.Query().Get("note"), r.URL.Query().Get("error")) +} + +func (s *Server) renderReports(w http.ResponseWriter, r *http.Request, code int, note, errMsg string) { + summaries, err := s.store.ListSummaries(r.Context(), 60) + if err != nil { + s.logger.Error("list summaries", "error", err) + http.Error(w, "could not load reports", http.StatusInternalServerError) + return + } + rows := make([]reportRow, 0, len(summaries)) + for _, sum := range summaries { + rows = append(rows, reportRow{ + ID: sum.ID, + Date: sum.Date, + Text: sum.Text, + Delivered: sum.DeliveredTo, + CreatedAt: sum.CreatedAt, + }) + } + data := reportsData{ + Summaries: rows, + Note: note, + Error: errMsg, + CSRFToken: s.csrfToken(w, r), + AppriseReady: s.cfg.AppriseConfigured(), + HubReady: s.cfg.HubConfigured(), + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + w.WriteHeader(code) + if err := s.tmpl.ExecuteTemplate(w, "reports.html", data); err != nil { + s.logger.Error("render reports", "error", err) + } +} + +// handleSummaryGenerate builds and stores today's summary. It never sends mail. +func (s *Server) handleSummaryGenerate(w http.ResponseWriter, r *http.Request) { + if !s.csrfOK(r) { + http.Error(w, "invalid CSRF token", http.StatusForbidden) + return + } + data, err := s.summaries.Build(r.Context()) + if err != nil { + s.logger.Error("build summary", "error", err) + s.redirectReports(w, r, "", "Could not build the summary: "+err.Error()) + return + } + id, err := s.store.SaveSummary(r.Context(), data.Date, data.Text(), "") + if err != nil { + s.logger.Error("save summary", "error", err) + s.redirectReports(w, r, "", "Could not store the summary: "+err.Error()) + return + } + s.logAction(r.Context(), "summary_generate", "Sam", + "date="+data.Date+" id="+strconv.FormatInt(id, 10)+ + " inbox="+strconv.Itoa(data.InboxTotal)+ + " needs_attention="+strconv.Itoa(data.NeedsAttention())) + s.redirectReports(w, r, "Built the summary for "+data.Date+". Nothing was sent.", "") +} + +// handleSummarySend builds, stores, and emails today's summary through Apprise. +// It is the only path that sends mail, and it runs only when Sam clicks Send. +func (s *Server) handleSummarySend(w http.ResponseWriter, r *http.Request) { + if !s.csrfOK(r) { + http.Error(w, "invalid CSRF token", http.StatusForbidden) + return + } + if !s.cfg.AppriseConfigured() { + s.redirectReports(w, r, "", "Apprise is not configured: set APPRISE_URL and APPRISE_KEY (or APPRISE_URLS).") + return + } + data, err := s.summaries.Build(r.Context()) + if err != nil { + s.logger.Error("build summary", "error", err) + s.redirectReports(w, r, "", "Could not build the summary: "+err.Error()) + return + } + id, err := s.store.SaveSummary(r.Context(), data.Date, data.Text(), "") + if err != nil { + s.logger.Error("save summary", "error", err) + s.redirectReports(w, r, "", "Could not store the summary: "+err.Error()) + return + } + if err := s.apprise.Send(r.Context(), data.Subject(), data.Text()); err != nil { + s.logger.Error("deliver summary", "error", err) + s.redirectReports(w, r, "", "Apprise delivery failed: "+err.Error()) + return + } + if err := s.store.MarkSummaryDelivered(r.Context(), id, s.cfg.SummaryTo); err != nil { + s.logger.Error("mark summary delivered", "error", err) + } + s.logAction(r.Context(), "summary_send", "Sam", + "date="+data.Date+" id="+strconv.FormatInt(id, 10)+" to="+s.cfg.SummaryTo) + s.redirectReports(w, r, "Sent the "+data.Date+" summary to "+s.cfg.SummaryTo+".", "") +} + +// handleSummaryDelete removes one stored report (AGENTS.md §5 delete-report). +func (s *Server) handleSummaryDelete(w http.ResponseWriter, r *http.Request) { + if !s.csrfOK(r) { + http.Error(w, "invalid CSRF token", http.StatusForbidden) + return + } + if err := r.ParseForm(); err != nil { + s.redirectReports(w, r, "", "Could not read the form.") + return + } + id, err := strconv.ParseInt(strings.TrimSpace(r.FormValue("id")), 10, 64) + if err != nil || id < 1 { + s.redirectReports(w, r, "", "Unknown report.") + return + } + if err := s.store.DeleteSummary(r.Context(), id); err != nil { + s.logger.Error("delete summary", "error", err) + s.redirectReports(w, r, "", "Could not delete the report: "+err.Error()) + return + } + s.logAction(r.Context(), "summary_delete", "Sam", "id="+strconv.FormatInt(id, 10)) + s.redirectReports(w, r, "Deleted report "+strconv.FormatInt(id, 10)+".", "") +} + +// redirectReports is POST/redirect/GET so a refresh does not resubmit. +func (s *Server) redirectReports(w http.ResponseWriter, r *http.Request, note, errMsg string) { + q := url.Values{} + if note != "" { + q.Set("note", note) + } + if errMsg != "" { + q.Set("error", errMsg) + } + http.Redirect(w, r, "/reports?"+q.Encode(), http.StatusSeeOther) +} diff --git a/internal/server/server.go b/internal/server/server.go index a9bb2d3..c2f8a27 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -17,13 +17,16 @@ import ( "net/http" "time" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/apprise" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/classify" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/config" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/hub" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/mailfetch" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/mailops" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/reply" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/session" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/summary" "gitea.lab.audasmedia.com.au/sam/email_agent/internal/version" "gitea.lab.audasmedia.com.au/sam/email_agent/web" ) @@ -46,6 +49,12 @@ type Server struct { actions MailActions replies ReplyGenerator sender MailSender + // Phase 7 (FR-10/FR-12): the read-only Hub answerer, the outbound + // notifier, the daily-summary builder, and the Apprise delivery client. + answerer *hub.Answerer + notifier *hub.Notifier + summaries *summary.Builder + apprise *apprise.Client } // New builds a Server. It does not start listening. @@ -82,6 +91,10 @@ func New(cfg *config.Config, st *store.Store, logger *slog.Logger) (*Server, err reply.NewClient(cfg.OmniRouteBaseURL, cfg.OmniRouteAPIKey, cfg.OmniRouteModel, logger), logger, ), + answerer: hub.NewAnswerer(st), + notifier: hub.NewNotifier(cfg.HubBaseURL, cfg.HubToken, cfg.HubNotifyTimeout, nil), + summaries: summary.New(st), + apprise: apprise.New(cfg.AppriseURL, cfg.AppriseKey, cfg.AppriseURLs, cfg.AppriseToken, 0), } s.http = &http.Server{ Addr: cfg.Addr, @@ -161,6 +174,16 @@ func (s *Server) Handler() http.Handler { mux.HandleFunc("POST /unsubscribe/unignore", s.requireAuth(s.handleUnsubscribeUnignore)) mux.HandleFunc("POST /unsubscribe/mark", s.requireAuth(s.handleUnsubscribeMark)) + // Phase 7 (FR-12): the Hub contract. /ask is read-only and token-protected. + mux.HandleFunc("GET /capabilities", s.requireHub(s.handleCapabilities)) + mux.HandleFunc("POST /ask", s.requireHub(s.handleAsk)) + + // Phase 7 (FR-10): daily summary and the Reports screen. + mux.HandleFunc("GET /reports", s.requireAuth(s.handleReports)) + mux.HandleFunc("POST /reports/generate", s.requireAuth(s.handleSummaryGenerate)) + mux.HandleFunc("POST /reports/send", s.requireAuth(s.handleSummarySend)) + mux.HandleFunc("POST /reports/delete", s.requireAuth(s.handleSummaryDelete)) + return s.recoverPanic(s.logRequests(mux)) } diff --git a/internal/store/summary.go b/internal/store/summary.go new file mode 100644 index 0000000..e62745e --- /dev/null +++ b/internal/store/summary.go @@ -0,0 +1,217 @@ +// Summary and reporting persistence for Phase 7 (FR-10). +// +// The daily summary is built from the same classification data the triage list +// uses; it is stored once per date in the `summaries` table (AGENTS.md §7) and +// then delivered by email through Apprise. +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "strings" +) + +// CategoryCount is one bucket of the inbox overview: a category name and how +// many messages currently sit in it. The empty category is the unclassified +// bucket. +type CategoryCount struct { + Category string + Count int +} + +// CountByCategory returns message counts for one folder, grouped by each +// message's latest classification category. Messages with no usable +// classification are counted under the empty category name. +func (s *Store) CountByCategory(ctx context.Context, folder string) ([]CategoryCount, error) { + const q = ` +SELECT COALESCE(c.category, ''), COUNT(*) +FROM messages m +LEFT JOIN classifications c ON c.id = ( + SELECT id FROM classifications WHERE message_id = m.id ORDER BY id DESC LIMIT 1 +) +WHERE (? = '' OR m.folder = ?) +GROUP BY 1 +ORDER BY 2 DESC, 1 ASC` + rows, err := s.db.QueryContext(ctx, q, folder, folder) + if err != nil { + return nil, fmt.Errorf("count by category: %w", err) + } + defer rows.Close() + + var out []CategoryCount + for rows.Next() { + var c CategoryCount + if err := rows.Scan(&c.Category, &c.Count); err != nil { + return nil, fmt.Errorf("scan category count: %w", err) + } + out = append(out, c) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("count by category: %w", err) + } + return out, nil +} + +// MessagesByCategory returns the newest messages in a folder whose latest +// classification is one of categories, highest importance first. It backs the +// "what needs attention" list in the daily summary and the Hub answer. +func (s *Store) MessagesByCategory(ctx context.Context, folder string, categories []string, limit int) ([]MessageListItem, error) { + if len(categories) == 0 { + return nil, nil + } + if limit < 1 { + limit = 20 + } + placeholders := make([]string, len(categories)) + args := make([]any, 0, len(categories)+2) + for i, c := range categories { + placeholders[i] = "?" + args = append(args, c) + } + // The importance rank puts high first; an unset importance sorts last. + rank := `CASE LOWER(COALESCE(c.importance, '')) + WHEN 'high' THEN 0 WHEN 'medium' THEN 1 WHEN 'low' THEN 2 ELSE 3 END` + folderClause := "" + if folder != "" { + folderClause = "AND m.folder = ?" + args = append(args, folder) + } + args = append(args, limit) + + q := ` +SELECT + m.id, m.gmail_id, COALESCE(m.thread_id, ''), COALESCE(m.from_addr, ''), + COALESCE(m.to_addr, ''), COALESCE(m.subject, ''), COALESCE(m.body_text, ''), + COALESCE(m.source_address, ''), COALESCE(m.date_sent, ''), m.received_at, + COALESCE(m.imap_uid, 0), COALESCE(m.imap_uidvalidity, 0), + COALESCE(m.folder, 'inbox'), COALESCE(m.mailbox, 'INBOX'), + COALESCE(c.category, ''), COALESCE(c.importance, ''), + COALESCE(c.action, ''), COALESCE(c.confidence, 0) +FROM messages m +LEFT JOIN classifications c ON c.id = ( + SELECT id FROM classifications WHERE message_id = m.id ORDER BY id DESC LIMIT 1 +) +WHERE c.category IN (` + strings.Join(placeholders, ",") + `) ` + folderClause + ` +ORDER BY ` + rank + `, COALESCE(NULLIF(m.date_sent, ''), m.received_at) DESC, m.id DESC +LIMIT ?` + rows, err := s.db.QueryContext(ctx, q, args...) + if err != nil { + return nil, fmt.Errorf("messages by category: %w", err) + } + defer rows.Close() + + var out []MessageListItem + for rows.Next() { + var ( + item MessageListItem + dateSent string + received string + ) + if err := rows.Scan( + &item.ID, &item.GmailID, &item.ThreadID, &item.FromAddr, + &item.ToAddr, &item.Subject, &item.BodyText, + &item.SourceAddress, &dateSent, &received, + &item.IMAPUID, &item.IMAPUIDValidity, &item.Folder, &item.Mailbox, + &item.Category, &item.Importance, &item.Action, &item.Confidence, + ); err != nil { + return nil, fmt.Errorf("scan category message: %w", err) + } + item.DateSent = parseSQLTime(dateSent) + item.ReceivedAt = parseSQLTime(received) + item.HasClassification = item.Category != "" + out = append(out, item) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("messages by category: %w", err) + } + return out, nil +} + +// Summary is one stored daily summary (FR-10). +type Summary struct { + ID int64 + Date string + Text string + DeliveredTo string + CreatedAt string +} + +// SaveSummary stores the summary for a date, replacing an earlier build for the +// same date so re-running the job does not duplicate rows. It returns the row id. +func (s *Store) SaveSummary(ctx context.Context, date, text, deliveredTo string) (int64, error) { + const q = ` +INSERT INTO summaries (date, text, delivered_to) VALUES (?, ?, ?) +ON CONFLICT (date) DO UPDATE SET + text = excluded.text, + delivered_to = excluded.delivered_to` + if _, err := s.db.ExecContext(ctx, q, date, text, nullString(deliveredTo)); err != nil { + return 0, fmt.Errorf("save summary %s: %w", date, err) + } + var id int64 + if err := s.db.QueryRowContext(ctx, + `SELECT id FROM summaries WHERE date = ?`, date).Scan(&id); err != nil { + return 0, fmt.Errorf("read summary id for %s: %w", date, err) + } + return id, nil +} + +// MarkSummaryDelivered records the recipient once a summary has been emailed. +func (s *Store) MarkSummaryDelivered(ctx context.Context, id int64, deliveredTo string) error { + if _, err := s.db.ExecContext(ctx, + `UPDATE summaries SET delivered_to = ? WHERE id = ?`, nullString(deliveredTo), id); err != nil { + return fmt.Errorf("mark summary %d delivered: %w", id, err) + } + return nil +} + +// SummaryByDate loads one summary by its YYYY-MM-DD date. ok is false when none +// exists. +func (s *Store) SummaryByDate(ctx context.Context, date string) (Summary, bool, error) { + row := s.db.QueryRowContext(ctx, ` +SELECT id, date, text, COALESCE(delivered_to, ''), created_at +FROM summaries WHERE date = ?`, date) + var sum Summary + if err := row.Scan(&sum.ID, &sum.Date, &sum.Text, &sum.DeliveredTo, &sum.CreatedAt); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return Summary{}, false, nil + } + return Summary{}, false, fmt.Errorf("read summary %s: %w", date, err) + } + return sum, true, nil +} + +// ListSummaries returns stored summaries, newest date first. +func (s *Store) ListSummaries(ctx context.Context, limit int) ([]Summary, error) { + if limit < 1 { + limit = 30 + } + rows, err := s.db.QueryContext(ctx, ` +SELECT id, date, text, COALESCE(delivered_to, ''), created_at +FROM summaries ORDER BY date DESC LIMIT ?`, limit) + if err != nil { + return nil, fmt.Errorf("list summaries: %w", err) + } + defer rows.Close() + var out []Summary + for rows.Next() { + var sum Summary + if err := rows.Scan(&sum.ID, &sum.Date, &sum.Text, &sum.DeliveredTo, &sum.CreatedAt); err != nil { + return nil, fmt.Errorf("scan summary: %w", err) + } + out = append(out, sum) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("list summaries: %w", err) + } + return out, nil +} + +// DeleteSummary removes one stored report (AGENTS.md §5 "delete-report"). +func (s *Store) DeleteSummary(ctx context.Context, id int64) error { + if _, err := s.db.ExecContext(ctx, `DELETE FROM summaries WHERE id = ?`, id); err != nil { + return fmt.Errorf("delete summary %d: %w", id, err) + } + return nil +} diff --git a/internal/store/summary_test.go b/internal/store/summary_test.go new file mode 100644 index 0000000..58bca32 --- /dev/null +++ b/internal/store/summary_test.go @@ -0,0 +1,106 @@ +package store + +import ( + "context" + "testing" +) + +func TestCountByCategory(t *testing.T) { + s, ctx := testStore(t) + a := seedMessage(t, s, ctx, "m1", "Security alert", "hello@where-woof.com") + b := seedMessage(t, s, ctx, "m2", "News", "hello@where-woof.com") + if _, err := s.SaveClassification(ctx, Classification{MessageID: a, Category: "Action", Importance: "high"}); err != nil { + t.Fatalf("classify a: %v", err) + } + if _, err := s.SaveClassification(ctx, Classification{MessageID: b, Category: "Newsletter"}); err != nil { + t.Fatalf("classify b: %v", err) + } + + counts, err := s.CountByCategory(ctx, FolderInbox) + if err != nil { + t.Fatalf("CountByCategory() error = %v", err) + } + got := map[string]int{} + for _, c := range counts { + got[c.Category] = c.Count + } + if got["Action"] != 1 || got["Newsletter"] != 1 { + t.Errorf("counts = %v, want Action=1 Newsletter=1", got) + } +} + +func TestMessagesByCategoryOrder(t *testing.T) { + s, ctx := testStore(t) + low := seedMessage(t, s, ctx, "low", "Routine", "hello@where-woof.com") + high := seedMessage(t, s, ctx, "high", "Urgent", "hello@where-woof.com") + info := seedMessage(t, s, ctx, "info", "FYI", "hello@where-woof.com") + mustClassify(t, s, ctx, low, "Action", "low") + mustClassify(t, s, ctx, high, "Action", "high") + mustClassify(t, s, ctx, info, "Info", "low") + + items, err := s.MessagesByCategory(ctx, FolderInbox, []string{"Action", "Important"}, 10) + if err != nil { + t.Fatalf("MessagesByCategory() error = %v", err) + } + if len(items) != 2 { + t.Fatalf("items = %d, want 2", len(items)) + } + if items[0].ID != high { + t.Errorf("first item id = %d, want the high-importance message %d", items[0].ID, high) + } +} + +func mustClassify(t *testing.T, s *Store, ctx context.Context, msgID int64, category, importance string) { + t.Helper() + if _, err := s.SaveClassification(ctx, Classification{MessageID: msgID, Category: category, Importance: importance}); err != nil { + t.Fatalf("classify %d: %v", msgID, err) + } +} + +func TestSummaryStore(t *testing.T) { + s, ctx := testStore(t) + + id, err := s.SaveSummary(ctx, "2026-10-11", "first text", "") + if err != nil { + t.Fatalf("SaveSummary() error = %v", err) + } + // Saving the same date replaces rather than duplicates. + id2, err := s.SaveSummary(ctx, "2026-10-11", "second text", "sam@example.com") + if err != nil { + t.Fatalf("SaveSummary() second error = %v", err) + } + if id2 != id { + t.Errorf("id = %d, want the existing row %d", id2, id) + } + + sum, ok, err := s.SummaryByDate(ctx, "2026-10-11") + if err != nil || !ok { + t.Fatalf("SummaryByDate() ok=%v err=%v", ok, err) + } + if sum.Text != "second text" || sum.DeliveredTo != "sam@example.com" { + t.Errorf("summary = %+v", sum) + } + + list, err := s.ListSummaries(ctx, 10) + if err != nil { + t.Fatalf("ListSummaries() error = %v", err) + } + if len(list) != 1 { + t.Fatalf("list = %d, want 1", len(list)) + } + + if err := s.MarkSummaryDelivered(ctx, id, "other@example.com"); err != nil { + t.Fatalf("MarkSummaryDelivered() error = %v", err) + } + sum, _, _ = s.SummaryByDate(ctx, "2026-10-11") + if sum.DeliveredTo != "other@example.com" { + t.Errorf("delivered_to = %q", sum.DeliveredTo) + } + + if err := s.DeleteSummary(ctx, id); err != nil { + t.Fatalf("DeleteSummary() error = %v", err) + } + if _, ok, _ := s.SummaryByDate(ctx, "2026-10-11"); ok { + t.Error("summary still present after delete") + } +} diff --git a/internal/summary/summary.go b/internal/summary/summary.go new file mode 100644 index 0000000..d907f24 --- /dev/null +++ b/internal/summary/summary.go @@ -0,0 +1,150 @@ +// Package summary builds the daily inbox summary (FR-10). +// +// The summary is a plain-text email body. It is built from the same read-only +// classification data the triage screen uses, then stored in the `summaries` +// table and delivered through Apprise. No LLM is involved: the summary must be +// cheap, deterministic, and answer in under a second. +package summary + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" +) + +// Builder reads the inbox and formats one day's summary. +type Builder struct { + Store *store.Store + // Now is injectable for deterministic tests. + Now func() time.Time +} + +// New builds a Builder. +func New(st *store.Store) *Builder { + return &Builder{Store: st, Now: func() time.Time { return time.Now() }} +} + +// Data is the structured summary, kept so a caller can render text or a UI. +type Data struct { + Date string + GeneratedAt time.Time + InboxTotal int + Counts []store.CategoryCount + Important []store.MessageListItem + Newsletters int + Spam int +} + +// Build collects the summary data for the current day. +func (b *Builder) Build(ctx context.Context) (Data, error) { + now := b.now() + counts, err := b.Store.CountByCategory(ctx, store.FolderInbox) + if err != nil { + return Data{}, err + } + important, err := b.Store.MessagesByCategory(ctx, store.FolderInbox, []string{"Action", "Important"}, 10) + if err != nil { + return Data{}, err + } + + d := Data{Date: now.Format("2006-01-02"), GeneratedAt: now, Counts: counts, Important: important} + for _, c := range counts { + d.InboxTotal += c.Count + switch strings.ToLower(strings.TrimSpace(c.Category)) { + case "newsletter": + d.Newsletters += c.Count + case "spam": + d.Spam += c.Count + } + } + return d, nil +} + +// Subject returns the email subject line for the summary. +func (d Data) Subject() string { + return fmt.Sprintf("Email Agent daily summary — %s", d.Date) +} + +// Text renders the plain-text email body. +func (d Data) Text() string { + var sb strings.Builder + fmt.Fprintf(&sb, "Email Agent — daily summary for %s\n\n", d.Date) + fmt.Fprintf(&sb, "Inbox: %d message(s)\n", d.InboxTotal) + for _, c := range sortedCounts(d.Counts) { + name := c.Category + if strings.TrimSpace(name) == "" { + name = "Unclassified" + } + fmt.Fprintf(&sb, " %-12s %d\n", name, c.Count) + } + + needs := d.NeedsAttention() + fmt.Fprintf(&sb, "\nNeeds attention: %d\n", needs) + if len(d.Important) == 0 { + sb.WriteString(" (nothing waiting)\n") + } + for _, m := range d.Important { + fmt.Fprintf(&sb, " - %s — %s [%s/%s]\n", + strings.TrimSpace(m.Subject), senderName(m.FromAddr), m.Category, m.Importance) + } + + fmt.Fprintf(&sb, "\nNewsletters: %d\n", d.Newsletters) + fmt.Fprintf(&sb, "Spam (recommend, never auto-delete): %d\n", d.Spam) + fmt.Fprintf(&sb, "\nGenerated %s by email_agent.\n", d.generatedAt()) + return sb.String() +} + +// NeedsAttention counts the Action and Important buckets. +func (d Data) NeedsAttention() int { + n := 0 + for _, c := range d.Counts { + switch strings.ToLower(strings.TrimSpace(c.Category)) { + case "action", "important": + n += c.Count + } + } + return n +} + +func (d Data) generatedAt() string { + if !d.GeneratedAt.IsZero() { + return d.GeneratedAt.Format("2006-01-02 15:04 MST") + } + return "" +} + +func (b *Builder) now() time.Time { + if b.Now == nil { + return time.Now() + } + return b.Now() +} + +// sortedCounts orders category rows by count descending, then name, so the text +// summary is stable run to run. +func sortedCounts(counts []store.CategoryCount) []store.CategoryCount { + out := append([]store.CategoryCount(nil), counts...) + sort.SliceStable(out, func(i, j int) bool { + if out[i].Count != out[j].Count { + return out[i].Count > out[j].Count + } + return out[i].Category < out[j].Category + }) + return out +} + +// senderName extracts a display name from a From header. +func senderName(addr string) string { + addr = strings.TrimSpace(addr) + if i := strings.Index(addr, "<"); i > 0 { + name := strings.Trim(strings.TrimSpace(addr[:i]), `"'`) + if name != "" { + return name + } + } + return addr +} diff --git a/internal/summary/summary_test.go b/internal/summary/summary_test.go new file mode 100644 index 0000000..5e361dc --- /dev/null +++ b/internal/summary/summary_test.go @@ -0,0 +1,106 @@ +package summary + +import ( + "context" + "path/filepath" + "strings" + "testing" + "time" + + "gitea.lab.audasmedia.com.au/sam/email_agent/internal/store" +) + +func testStore(t *testing.T) (*store.Store, context.Context) { + t.Helper() + ctx := context.Background() + st, err := store.Open(ctx, filepath.Join(t.TempDir(), "email_agent.db")) + if err != nil { + t.Fatalf("store.Open() error = %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + return st, ctx +} + +func seed(t *testing.T, st *store.Store, ctx context.Context, gmailID, from, subject, category, importance string) { + t.Helper() + id, err := st.UpsertMessage(ctx, &store.Message{ + GmailID: gmailID, + FromAddr: from, + Subject: subject, + DateSent: time.Now().UTC(), + }) + if err != nil { + t.Fatalf("seed message: %v", err) + } + if _, err := st.SaveClassification(ctx, store.Classification{ + MessageID: id, Category: category, Importance: importance, + }); err != nil { + t.Fatalf("classify: %v", err) + } +} + +func TestBuildAndText(t *testing.T) { + st, ctx := testStore(t) + seed(t, st, ctx, "1", "Google ", "Security alert", "Action", "high") + seed(t, st, ctx, "2", "Store ", "Spring sale", "Newsletter", "low") + seed(t, st, ctx, "3", "Store ", "More sale", "Newsletter", "low") + seed(t, st, ctx, "4", "Spammer ", "Cheap pills", "Spam", "low") + + b := New(st) + b.Now = func() time.Time { return time.Date(2026, 10, 11, 8, 0, 0, 0, time.UTC) } + data, err := b.Build(ctx) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if data.Date != "2026-10-11" { + t.Errorf("date = %q", data.Date) + } + if data.InboxTotal != 4 { + t.Errorf("InboxTotal = %d, want 4", data.InboxTotal) + } + if data.NeedsAttention() != 1 { + t.Errorf("NeedsAttention = %d, want 1", data.NeedsAttention()) + } + if data.Newsletters != 2 { + t.Errorf("Newsletters = %d, want 2", data.Newsletters) + } + if data.Spam != 1 { + t.Errorf("Spam = %d, want 1", data.Spam) + } + if len(data.Important) != 1 || data.Important[0].Subject != "Security alert" { + t.Errorf("Important = %+v", data.Important) + } + + text := data.Text() + for _, want := range []string{ + "daily summary for 2026-10-11", + "Inbox: 4 message(s)", + "Security alert", + "Google", + "Newsletters: 2", + "Spam (recommend, never auto-delete): 1", + } { + if !strings.Contains(text, want) { + t.Errorf("Text() missing %q:\n%s", want, text) + } + } + if sub := data.Subject(); !strings.Contains(sub, "2026-10-11") { + t.Errorf("Subject() = %q", sub) + } +} + +func TestBuildEmptyInbox(t *testing.T) { + st, ctx := testStore(t) + b := New(st) + b.Now = func() time.Time { return time.Date(2026, 10, 11, 8, 0, 0, 0, time.UTC) } + data, err := b.Build(ctx) + if err != nil { + t.Fatalf("Build() error = %v", err) + } + if data.InboxTotal != 0 || data.NeedsAttention() != 0 { + t.Errorf("data = %+v, want empty", data) + } + if !strings.Contains(data.Text(), "nothing waiting") { + t.Errorf("Text() = %s", data.Text()) + } +} diff --git a/web/static/app.css b/web/static/app.css index 23a4f71..b61bfce 100644 --- a/web/static/app.css +++ b/web/static/app.css @@ -531,3 +531,25 @@ a:hover { text-decoration: underline; } .unsub-row { flex-direction: column; align-items: stretch; } .rules-version { flex-direction: column; align-items: stretch; } } + +/* Phase 7 (FR-10): Reports screen and daily summary. */ +.report { + border: 1px solid var(--hairline); + border-radius: 4px; + padding: var(--space-sm); + margin-bottom: var(--space-md); + display: grid; + gap: var(--space-xs); +} +.report:last-child { margin-bottom: 0; } +.report-head { display: flex; align-items: center; gap: var(--space-sm); flex-wrap: wrap; } +.report-text { + margin: 0; + padding: var(--space-sm); + background: var(--surface-2, var(--canvas)); + border-radius: 4px; + font: var(--type-caption); + color: var(--ink-secondary); + white-space: pre-wrap; + overflow-wrap: anywhere; +} diff --git a/web/templates/dashboard.html b/web/templates/dashboard.html index ecb346b..fcbcb48 100644 --- a/web/templates/dashboard.html +++ b/web/templates/dashboard.html @@ -28,6 +28,7 @@
Rules Unsubscribe + Reports +
+ + +
+
+ + +
+

Reports

+ + {{if .Note}}
{{.Note}}
{{end}} + {{if .Error}}{{end}} + +
+
+

Daily summary

+ delivered by email via Apprise +
+
+

+ The summary counts today's inbox by category and lists what needs + attention. Building a summary never sends mail. Only Send + now emails it, and only when Apprise is configured. +

+
+
+ + +
+ {{if .AppriseReady}} +
+ + +
+ {{else}} + Sending is unavailable until APPRISE_URL and a key (or APPRISE_URLS) are set. + {{end}} +
+ {{if .HubReady}} +

The Hub API is configured. The outbound /notify path is available to the summary job.

+ {{else}} +

The Hub API is not configured (set HUB_TOKEN to serve /capabilities and /ask).

+ {{end}} +
+
+ +
+

Stored summaries

+
+ {{range .Summaries}} +
+
+ {{.Date}} + {{if .Delivered}}sent to {{.Delivered}}{{else}}not sent{{end}} + {{.CreatedAt}} +
+ + + +
+
+ 
{{.Text}}
+
+ {{else}} +

No summaries stored yet. Build today's summary to create one.

+ {{end}} +
+
+
+ + diff --git a/web/templates/rules.html b/web/templates/rules.html index ee09601..a888e5a 100644 --- a/web/templates/rules.html +++ b/web/templates/rules.html @@ -28,6 +28,7 @@
Back to inbox Unsubscribe + Reports