Phase 5: reply drafts — template/LLM draft, edit, SMTP send (AUD-71)

Implements Phase 5 of AGENTS.md §11 (FR-8, FR-15):

- Message detail pane (GET /partials/message) with body, recommended
  action, reply identity, template picker, editable draft, and history.
- internal/reply: OmniRoute (OpenAI-compatible) chat client and a
  template-based draft generator. All non-JEV models go through
  OmniRoute (§10).
- internal/mailops: SMTP send over smtp.gmail.com:587 with STARTTLS and
  the app password, forcing IPv4 (§9); FR-15 reply identity (send as the
  arrived-at X-Pm-Original-To address, else "Sam Rolfe"); threading via
  In-Reply-To/References; -check-smtp safe verification; and a
  [Gmail]/Sent Mail thread check.
- Migration v5 stores messages.message_id_header and drafts.sent_at;
  mailfetch parses Message-ID.
- Send is user-triggered only: CSRF + UI confirm + confirm=yes guard,
  no auto-send, duplicate-send refused. A sent reply files the original
  under Replied-to/<address> and logs every step (FR-13).
- Tests for reply, mailops SMTP/identity/thread check, store drafts and
  migration v5, and the server draft/edit/send guards.
- docs/phase5-replies.md, README, .env.example, docker-compose updated.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Paperclip Builder
2026-10-11 22:05:35 +11:00
parent ae53374ad1
commit 7ca888f0c6
31 changed files with 2556 additions and 42 deletions

View File

@@ -29,6 +29,11 @@ const (
DefaultIMAPLookbackDays = 14
DefaultIMAPMaxMessages = 100
// SMTP defaults (AGENTS.md §9): Gmail submission on 587 with STARTTLS. The
// dialer forces IPv4, matching IMAP.
DefaultSMTPHost = "smtp.gmail.com"
DefaultSMTPPort = 587
// JEV (TypeSafe System One) defaults. The base URL is fixed by the TypeSafe
// docs (https://docs.typesafe.ai/api.md); JEV_BASE_URL overrides it.
DefaultJEVBaseURL = "https://api.typesafe.ai"
@@ -38,6 +43,12 @@ const (
// max messages bounds one run so a scheduled scan stays cheap.
DefaultClassifyConcurrency = 4
DefaultClassifyMaxMessages = 200
// OmniRoute defaults (AGENTS.md §8/§10). Reply drafts are the first caller.
// The gateway runs on .13, the host that runs this service; the other house
// apps use the same API endpoint. OMNIROUTE_BASE_URL overrides it.
DefaultOmniRouteBaseURL = "http://192.168.20.13:20129/v1"
DefaultOmniRouteModel = "auto/best-chat"
)
// Config is the fully resolved runtime configuration.
@@ -83,10 +94,13 @@ type Config struct {
// ClassifyMaxMessages caps how many messages one classify run processes.
ClassifyMaxMessages int
// OmniRouteBaseURL is the base URL of the OmniRoute gateway.
// OmniRouteBaseURL is the base URL of the OmniRoute gateway. All non-JEV
// models go through it (AGENTS.md §10).
OmniRouteBaseURL string
// OmniRouteAPIKey authenticates to OmniRoute. Secret.
OmniRouteAPIKey string
// OmniRouteModel is the model alias used for reply drafts.
OmniRouteModel string
// GmailUser is the Gmail account that receives the forwarded mail.
// Secret (half of the app-password pair).
@@ -102,6 +116,10 @@ type Config struct {
IMAPLookbackDays int
// IMAPMaxMessages caps how many messages one scan fetches.
IMAPMaxMessages int
// SMTPHost is the Gmail submission host (default smtp.gmail.com).
SMTPHost string
// SMTPPort is the Gmail submission port (default 587, STARTTLS).
SMTPPort int
// HubToken authenticates the read-only Hub API. Secret.
HubToken string
@@ -118,8 +136,9 @@ func Load() (*Config, error) {
JEVBaseURL: strings.TrimRight(getenvDefault("JEV_BASE_URL", DefaultJEVBaseURL), "/"),
JEVAPIKey: os.Getenv("JEV_API_KEY"),
JEVModel: getenvDefault("JEV_MODEL", DefaultJEVModel),
OmniRouteBaseURL: strings.TrimRight(os.Getenv("OMNIROUTE_BASE_URL"), "/"),
OmniRouteBaseURL: strings.TrimRight(getenvDefault("OMNIROUTE_BASE_URL", DefaultOmniRouteBaseURL), "/"),
OmniRouteAPIKey: os.Getenv("OMNIROUTE_API_KEY"),
OmniRouteModel: getenvDefault("OMNIROUTE_MODEL", DefaultOmniRouteModel),
GmailUser: os.Getenv("GMAIL_USER"),
GmailAppPassword: os.Getenv("GMAIL_APP_PASSWORD"),
HubToken: os.Getenv("HUB_TOKEN"),
@@ -131,6 +150,12 @@ func Load() (*Config, error) {
}
c.IMAPHost = getenvDefault("EMAIL_AGENT_IMAP_HOST", DefaultIMAPHost)
c.IMAPPort = imapPort
smtpPort, err := resolvePortSetting("EMAIL_AGENT_SMTP_PORT", DefaultSMTPPort)
if err != nil {
return nil, err
}
c.SMTPHost = getenvDefault("EMAIL_AGENT_SMTP_HOST", DefaultSMTPHost)
c.SMTPPort = smtpPort
c.IMAPLookbackDays = getenvDefaultInt("EMAIL_AGENT_IMAP_LOOKBACK_DAYS", DefaultIMAPLookbackDays)
c.IMAPMaxMessages = getenvDefaultInt("EMAIL_AGENT_IMAP_MAX_MESSAGES", DefaultIMAPMaxMessages)
c.ClassifyConcurrency = getenvDefaultInt("EMAIL_AGENT_CLASSIFY_CONCURRENCY", DefaultClassifyConcurrency)
@@ -152,13 +177,19 @@ func Load() (*Config, error) {
// resolveIMAPPort reads EMAIL_AGENT_IMAP_PORT, falling back to DefaultIMAPPort.
func resolveIMAPPort() (int, error) {
raw := strings.TrimSpace(os.Getenv("EMAIL_AGENT_IMAP_PORT"))
return resolvePortSetting("EMAIL_AGENT_IMAP_PORT", DefaultIMAPPort)
}
// resolvePortSetting reads a port-valued environment variable, falling back to
// def when unset.
func resolvePortSetting(key string, def int) (int, error) {
raw := strings.TrimSpace(os.Getenv(key))
if raw == "" {
return DefaultIMAPPort, nil
return def, nil
}
p, err := strconv.Atoi(raw)
if err != nil {
return 0, fmt.Errorf("EMAIL_AGENT_IMAP_PORT %q is not a number: %w", raw, err)
return 0, fmt.Errorf("%s %q is not a number: %w", key, raw, err)
}
return p, nil
}
@@ -229,6 +260,12 @@ func (c *Config) validate() error {
if c.IMAPPort < 1 || c.IMAPPort > 65535 {
return fmt.Errorf("EMAIL_AGENT_IMAP_PORT %d is out of range", c.IMAPPort)
}
if strings.TrimSpace(c.SMTPHost) == "" {
return fmt.Errorf("EMAIL_AGENT_SMTP_HOST must not be empty")
}
if c.SMTPPort < 1 || c.SMTPPort > 65535 {
return fmt.Errorf("EMAIL_AGENT_SMTP_PORT %d is out of range", c.SMTPPort)
}
return nil
}

View File

@@ -104,6 +104,7 @@ func (f *Fetcher) Fetch(ctx context.Context) (Summary, error) {
msg := &store.Message{
GmailID: gmailID(r, uidValidity),
ThreadID: r.GmailThreadID,
MessageIDHeader: parsed.MessageID,
FromAddr: parsed.From,
ToAddr: parsed.To,
Subject: parsed.Subject,

View File

@@ -10,6 +10,7 @@ package mailfetch
import (
"errors"
"fmt"
"strconv"
"strings"
)
@@ -88,17 +89,29 @@ func (c *Client) StoreFlags(uid uint32, op string, flag string) error {
// id (X-GM-MSGID) in the currently selected mailbox. Gmail assigns a different
// UID per label, so this is how a message is located after a move.
func (c *Client) SearchGmailID(gmailID string) ([]uint32, error) {
gmailID = strings.TrimSpace(gmailID)
if gmailID == "" {
return nil, errors.New("empty x-gm-msgid")
return c.searchNumeric("X-GM-MSGID", gmailID)
}
// SearchGmailThreadID returns the UIDs of the messages in the given Gmail
// thread (X-GM-THRID) in the currently selected mailbox. Phase 5 uses it to
// confirm that a reply landed in the same thread as the original message.
func (c *Client) SearchGmailThreadID(threadID string) ([]uint32, error) {
return c.searchNumeric("X-GM-THRID", threadID)
}
// searchNumeric runs a UID SEARCH for a numeric Gmail extension key.
func (c *Client) searchNumeric(key, value string) ([]uint32, error) {
value = strings.TrimSpace(value)
if value == "" {
return nil, fmt.Errorf("empty %s", strings.ToLower(key))
}
for _, r := range gmailID {
for _, r := range value {
if r < '0' || r > '9' {
return nil, errors.New("x-gm-msgid must be numeric")
return nil, fmt.Errorf("%s must be numeric", strings.ToLower(key))
}
}
var uids []uint32
err := c.exec("UID SEARCH X-GM-MSGID "+gmailID, func(resp *response) error {
err := c.exec("UID SEARCH "+key+" "+value, func(resp *response) error {
if !strings.HasPrefix(strings.ToUpper(resp.line), "* SEARCH") {
return nil
}

View File

@@ -21,10 +21,13 @@ const maxBodyBytes = 1 << 20 // 1 MiB
// ParsedMessage is the subset of an RFC822 message the agent stores.
type ParsedMessage struct {
From string
To string
Subject string
Date time.Time
From string
To string
Subject string
Date time.Time
// MessageID is the RFC822 Message-ID header, used for reply threading
// (Phase 5, FR-8).
MessageID string
SourceAddress string
// SourceHeader records which header the source address came from, so FR-2
// decisions are auditable.
@@ -40,10 +43,11 @@ func ParseMessage(raw []byte) (*ParsedMessage, error) {
}
h := msg.Header
pm := &ParsedMessage{
From: decodeHeader(h.Get("From")),
To: decodeHeader(h.Get("To")),
Subject: decodeHeader(h.Get("Subject")),
Date: parseDate(h.Get("Date")),
From: decodeHeader(h.Get("From")),
To: decodeHeader(h.Get("To")),
Subject: decodeHeader(h.Get("Subject")),
Date: parseDate(h.Get("Date")),
MessageID: strings.TrimSpace(h.Get("Message-Id")),
}
pm.SourceAddress, pm.SourceHeader = SourceAddress(textproto.MIMEHeader(h))
pm.BodyText = extractBody(textproto.MIMEHeader(h), msg.Body)

View File

@@ -31,6 +31,9 @@ const (
LabelDelete = "Delete"
LabelSaved = "Saved"
repliedPrefix = "Replied-to/"
// LabelSent is Gmail's IMAP name for the Sent mailbox. Phase 5 checks it to
// confirm a reply joined the original thread.
LabelSent = "[Gmail]/Sent Mail"
)
// ErrNotConfigured means Gmail credentials are missing, so no mailbox action can
@@ -46,25 +49,30 @@ type Conn interface {
Move(uid uint32, dest string) error
StoreFlags(uid uint32, op string, flag string) error
SearchGmailID(gmailID string) ([]uint32, error)
SearchGmailThreadID(threadID string) ([]uint32, error)
Logout() error
}
// dialFunc opens an IMAP connection. It is a field so tests can substitute one.
type dialFunc func(ctx context.Context, host string, port int, timeout time.Duration) (Conn, error)
// Client performs triage actions against Gmail IMAP.
// Client performs triage actions against Gmail IMAP and sends replies over
// Gmail SMTP.
type Client struct {
cfg *config.Config
logger *slog.Logger
dial dialFunc
// smtpDial opens the SMTP connection (IPv4 only). It is a field so tests can
// substitute one; the production value is dialSMTP4.
smtpDial smtpDialFunc
}
// New builds a Client using the real Gmail IMAP dialer.
// New builds a Client using the real Gmail IMAP dialer and IPv4 SMTP dialer.
func New(cfg *config.Config, logger *slog.Logger) *Client {
if logger == nil {
logger = slog.New(slog.DiscardHandler)
}
return &Client{cfg: cfg, logger: logger, dial: defaultDial}
return &Client{cfg: cfg, logger: logger, dial: defaultDial, smtpDial: dialSMTP4}
}
// defaultDial dials Gmail over IPv4 (mailfetch.Dial forces tcp4).
@@ -143,6 +151,29 @@ func (c *Client) Mark(ctx context.Context, msg store.Message, mark string) error
return nil
}
// VerifyThreadUpdated reports whether the Gmail thread now contains a message in
// the Sent mailbox. Phase 5 calls it after a reply is sent so the DoD "the
// thread updates" can be checked without touching the recipient.
func (c *Client) VerifyThreadUpdated(ctx context.Context, threadID string) (bool, error) {
if strings.TrimSpace(threadID) == "" {
return false, fmt.Errorf("message has no Gmail thread id")
}
conn, err := c.connect(ctx)
if err != nil {
return false, err
}
defer func() { _ = conn.Logout() }()
if _, err := conn.Select(LabelSent); err != nil {
return false, fmt.Errorf("select %q: %w", LabelSent, err)
}
uids, err := conn.SearchGmailThreadID(threadID)
if err != nil {
return false, fmt.Errorf("search thread %s in %q: %w", threadID, LabelSent, err)
}
return len(uids) > 0, nil
}
// connect dials and logs in. Callers must Logout when done.
func (c *Client) connect(ctx context.Context) (Conn, error) {
if !c.cfg.GmailConfigured() {

View File

@@ -45,6 +45,10 @@ func (f *fakeConn) SearchGmailID(id string) ([]uint32, error) {
f.searched = append(f.searched, id)
return f.searchUIDs, f.searchErr
}
func (f *fakeConn) SearchGmailThreadID(id string) ([]uint32, error) {
f.searched = append(f.searched, "thrid:"+id)
return f.searchUIDs, f.searchErr
}
func (f *fakeConn) Logout() error { f.loggedOut = true; return nil }
func configured() *config.Config {
@@ -54,6 +58,8 @@ func configured() *config.Config {
IMAPPort: config.DefaultIMAPPort,
GmailUser: "user@example.test",
GmailAppPassword: "app-password",
SMTPHost: config.DefaultSMTPHost,
SMTPPort: config.DefaultSMTPPort,
}
}

264
internal/mailops/smtp.go Normal file
View File

@@ -0,0 +1,264 @@
// SMTP send path for Phase 5 replies (FR-8, FR-15).
//
// Gmail SMTP is reached over IPv4 only, exactly like the IMAP read path
// (AGENTS.md §9): IPv6 to Gmail is broken from .13. The connection upgrades with
// STARTTLS on port 587 and authenticates with the same app password as IMAP.
//
// Reply identity (FR-15): the From header is the address the mail arrived at
// (X-Pm-Original-To). When that is unknown, it falls back to Sam Rolfe at the
// Gmail account. The SMTP envelope sender is always the authenticated Gmail
// account, which is what Gmail expects; the From header carries the alias.
package mailops
import (
"context"
"crypto/rand"
"crypto/tls"
"encoding/hex"
"fmt"
"mime"
"net"
"net/mail"
"net/smtp"
"strings"
"time"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/mailfetch"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/store"
)
// ReplyFromName is the display name used when the arrived-at address is unknown
// (AGENTS.md §13.1: "Otherwise send as Sam Rolfe").
const ReplyFromName = "Sam Rolfe"
// OutgoingMessage is one reply to send.
type OutgoingMessage struct {
// To is the recipient (the original sender).
To string
// From is the From header value, normally "Sam Rolfe <arrived-at@address>".
From string
// Subject is the reply subject (already prefixed with Re: when appropriate).
Subject string
// InReplyTo and References carry the original Message-ID so Gmail threads
// the reply.
InReplyTo string
References string
// Body is the plain-text reply body.
Body string
}
// Sender sends a reply. The live implementation is *Client; tests substitute a
// fake.
type Sender interface {
Send(ctx context.Context, msg OutgoingMessage) error
}
// SMTPConfig is the connection detail for the Gmail SMTP submission endpoint.
type SMTPConfig struct {
Host string
Port int
User string
Password string
}
// smtpDialFunc opens a TCP connection. It is a field so tests can assert the
// IPv4-only network is used.
type smtpDialFunc func(ctx context.Context, host string, port int, timeout time.Duration) (net.Conn, error)
// dialSMTP4 forces IPv4, matching mailfetch.Dial on the read path.
func dialSMTP4(ctx context.Context, host string, port int, timeout time.Duration) (net.Conn, error) {
d := net.Dialer{Timeout: timeout}
return d.DialContext(ctx, "tcp4", net.JoinHostPort(host, fmt.Sprintf("%d", port)))
}
// ReplyIdentity returns the From header for a reply (FR-15): the address the
// mail arrived at, otherwise Sam Rolfe at the Gmail account.
func ReplyIdentity(msg store.Message, gmailUser string) string {
addr := strings.TrimSpace(msg.SourceAddress)
if addr == "" {
addr = strings.TrimSpace(gmailUser)
}
if addr == "" {
return ReplyFromName
}
return (&mail.Address{Name: ReplyFromName, Address: addr}).String()
}
// ReplyTo returns the recipient of a reply: the original sender.
func ReplyTo(msg store.Message) string {
addr := strings.TrimSpace(msg.FromAddr)
if addr == "" {
return ""
}
if parsed, err := mail.ParseAddress(addr); err == nil {
return parsed.Address
}
return addr
}
// ReplySubject prefixes the original subject with "Re: " once.
func ReplySubject(subject string) string {
s := strings.TrimSpace(subject)
if s == "" {
return "Re: (no subject)"
}
if strings.HasPrefix(strings.ToLower(s), "re:") {
return s
}
return "Re: " + s
}
// CheckSMTP verifies the submission path without sending mail: it dials Gmail
// over IPv4, upgrades with STARTTLS, and authenticates with the app password.
// It stops before MAIL FROM, so no message is sent. It is the safe part of the
// Phase 5 send-path verification.
func (c *Client) CheckSMTP(ctx context.Context) error {
if !c.cfg.GmailConfigured() {
return ErrNotConfigured
}
host := c.cfg.SMTPHost
port := c.cfg.SMTPPort
conn, err := c.smtpDial(ctx, host, port, mailfetch.DefaultTimeout)
if err != nil {
return fmt.Errorf("dial smtp %s:%d (ipv4): %w", host, port, err)
}
client, err := smtp.NewClient(conn, host)
if err != nil {
_ = conn.Close()
return fmt.Errorf("smtp handshake: %w", err)
}
defer func() { _ = client.Close() }()
if ok, _ := client.Extension("STARTTLS"); ok {
if err := client.StartTLS(&tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}); err != nil {
return fmt.Errorf("smtp starttls: %w", err)
}
}
if err := client.Auth(smtp.PlainAuth("", c.cfg.GmailUser, c.cfg.GmailAppPassword, host)); err != nil {
return fmt.Errorf("smtp auth: %w", err)
}
c.logger.Info("smtp check ok", "host", host, "port", port, "user", c.cfg.GmailUser)
return client.Quit()
}
// Send builds and sends one reply over IPv4 + STARTTLS + the app password.
func (c *Client) Send(ctx context.Context, msg OutgoingMessage) error {
if !c.cfg.GmailConfigured() {
return ErrNotConfigured
}
if strings.TrimSpace(msg.To) == "" {
return fmt.Errorf("reply has no recipient")
}
host := c.cfg.SMTPHost
port := c.cfg.SMTPPort
conn, err := c.smtpDial(ctx, host, port, mailfetch.DefaultTimeout)
if err != nil {
return fmt.Errorf("dial smtp %s:%d (ipv4): %w", host, port, err)
}
client, err := smtp.NewClient(conn, host)
if err != nil {
_ = conn.Close()
return fmt.Errorf("smtp handshake: %w", err)
}
defer func() { _ = client.Close() }()
if ok, _ := client.Extension("STARTTLS"); ok {
if err := client.StartTLS(&tls.Config{ServerName: host, MinVersion: tls.VersionTLS12}); err != nil {
return fmt.Errorf("smtp starttls: %w", err)
}
}
if err := client.Auth(smtp.PlainAuth("", c.cfg.GmailUser, c.cfg.GmailAppPassword, host)); err != nil {
return fmt.Errorf("smtp auth: %w", err)
}
// The envelope sender is the authenticated account; the From header carries
// the arrived-at alias (FR-15).
if err := client.Mail(c.cfg.GmailUser); err != nil {
return fmt.Errorf("smtp mail from: %w", err)
}
if err := client.Rcpt(msg.To); err != nil {
return fmt.Errorf("smtp rcpt to %s: %w", msg.To, err)
}
w, err := client.Data()
if err != nil {
return fmt.Errorf("smtp data: %w", err)
}
raw := BuildRFC822(msg, c.cfg.GmailUser, time.Now().UTC())
if _, err := w.Write(raw); err != nil {
_ = w.Close()
return fmt.Errorf("write reply: %w", err)
}
if err := w.Close(); err != nil {
return fmt.Errorf("finish reply: %w", err)
}
if err := client.Quit(); err != nil {
return fmt.Errorf("smtp quit: %w", err)
}
c.logger.Info("reply sent", "to", msg.To, "from", msg.From, "subject", msg.Subject)
return nil
}
// BuildRFC822 renders the reply as an RFC822 message. The Message-ID is
// generated and the headers are CRLF-terminated for SMTP. It is exported so the
// wire format can be tested without a live server.
func BuildRFC822(msg OutgoingMessage, gmailUser string, now time.Time) []byte {
var b strings.Builder
writeHeader(&b, "From", msg.From)
writeHeader(&b, "To", msg.To)
writeHeader(&b, "Subject", msg.Subject)
writeHeader(&b, "Date", now.Format(time.RFC1123Z))
writeHeader(&b, "Message-ID", generateMessageID(now, gmailUser))
if v := strings.TrimSpace(msg.InReplyTo); v != "" {
writeHeader(&b, "In-Reply-To", ensureAngle(v))
}
if v := strings.TrimSpace(msg.References); v != "" {
writeHeader(&b, "References", ensureAngle(v))
}
writeHeader(&b, "MIME-Version", "1.0")
writeHeader(&b, "Content-Type", "text/plain; charset=UTF-8")
writeHeader(&b, "Content-Transfer-Encoding", "8bit")
b.WriteString("\r\n")
b.WriteString(strings.ReplaceAll(strings.ReplaceAll(msg.Body, "\r\n", "\n"), "\n", "\r\n"))
if !strings.HasSuffix(b.String(), "\r\n") {
b.WriteString("\r\n")
}
return []byte(b.String())
}
// writeHeader encodes a header value when it contains non-ASCII text.
func writeHeader(b *strings.Builder, key, value string) {
value = strings.TrimSpace(value)
for _, r := range value {
if r > 127 {
value = mime.QEncoding.Encode("UTF-8", value)
break
}
}
b.WriteString(key)
b.WriteString(": ")
b.WriteString(value)
b.WriteString("\r\n")
}
// generateMessageID builds a unique Message-ID for an outgoing reply.
func generateMessageID(now time.Time, user string) string {
domain := "email-agent.local"
if user != "" {
if i := strings.LastIndex(user, "@"); i >= 0 && i < len(user)-1 {
domain = user[i+1:]
}
}
buf := make([]byte, 8)
_, _ = rand.Read(buf)
return fmt.Sprintf("<%d.%s@%s>", now.UnixNano(), hex.EncodeToString(buf), domain)
}
// ensureAngle wraps a Message-ID in angle brackets when it is not already.
func ensureAngle(id string) string {
id = strings.TrimSpace(id)
if strings.HasPrefix(id, "<") && strings.HasSuffix(id, ">") {
return id
}
return "<" + strings.Trim(id, "<>") + ">"
}

View File

@@ -0,0 +1,140 @@
package mailops
import (
"context"
"net"
"strings"
"testing"
"time"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/config"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/store"
)
func TestReplyIdentityUsesArrivedAtAddress(t *testing.T) {
msg := store.Message{SourceAddress: "Hello@where-woof.com"}
if got := ReplyIdentity(msg, "samuelrolfe@gmail.com"); got != `"Sam Rolfe" <Hello@where-woof.com>` {
t.Errorf("ReplyIdentity() = %q", got)
}
fallback := ReplyIdentity(store.Message{}, "samuelrolfe@gmail.com")
if fallback != `"Sam Rolfe" <samuelrolfe@gmail.com>` {
t.Errorf("fallback = %q", fallback)
}
if got := ReplyIdentity(store.Message{}, ""); got != "Sam Rolfe" {
t.Errorf("bare identity = %q", got)
}
}
func TestReplyToParsesSender(t *testing.T) {
if got := ReplyTo(store.Message{FromAddr: "Customer Name <cust@example.test>"}); got != "cust@example.test" {
t.Errorf("ReplyTo() = %q", got)
}
if got := ReplyTo(store.Message{FromAddr: "plain@example.test"}); got != "plain@example.test" {
t.Errorf("ReplyTo() = %q", got)
}
if got := ReplyTo(store.Message{}); got != "" {
t.Errorf("ReplyTo() empty = %q", got)
}
}
func TestReplySubjectPrefixesOnce(t *testing.T) {
if got := ReplySubject("Booking"); got != "Re: Booking" {
t.Errorf("ReplySubject() = %q", got)
}
if got := ReplySubject("RE: Booking"); got != "RE: Booking" {
t.Errorf("ReplySubject() should not double-prefix: %q", got)
}
}
func TestBuildRFC822ThreadsReply(t *testing.T) {
raw := string(BuildRFC822(OutgoingMessage{
To: "cust@example.test",
From: `"Sam Rolfe" <hello@where-woof.com>`,
Subject: "Re: Booking",
InReplyTo: "<orig-123@mail.example>",
References: "orig-123@mail.example",
Body: "Saturday works.\n\nSam",
}, "samuelrolfe@gmail.com", time.Unix(0, 0).UTC()))
for _, want := range []string{
"From: \"Sam Rolfe\" <hello@where-woof.com>\r\n",
"To: cust@example.test\r\n",
"Subject: Re: Booking\r\n",
"In-Reply-To: <orig-123@mail.example>\r\n",
"References: <orig-123@mail.example>\r\n",
"Content-Type: text/plain; charset=UTF-8\r\n",
"\r\nSaturday works.\r\n\r\nSam\r\n",
} {
if !strings.Contains(raw, want) {
t.Errorf("message missing %q\n%s", want, raw)
}
}
if strings.Contains(strings.ReplaceAll(raw, "\r\n", ""), "\n") {
t.Errorf("message contains a bare LF; SMTP needs CRLF line endings")
}
if !strings.Contains(raw, "Message-ID: <") {
t.Errorf("message has no generated Message-ID:\n%s", raw)
}
}
func TestSendWithoutCredentials(t *testing.T) {
c := New(&config.Config{Env: "test"}, nil)
out := OutgoingMessage{To: "a@b.test", From: "Sam Rolfe <x@y.test>", Body: "hi"}
if err := c.Send(context.Background(), out); err != ErrNotConfigured {
t.Fatalf("Send() error = %v, want ErrNotConfigured", err)
}
}
func TestDialSMTP4ConnectsOverIPv4(t *testing.T) {
ln, err := net.Listen("tcp4", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
defer ln.Close()
port := ln.Addr().(*net.TCPAddr).Port
// "localhost" may resolve to ::1 first; the tcp4 network must still reach the
// IPv4 listener.
conn, err := dialSMTP4(context.Background(), "localhost", port, 2*time.Second)
if err != nil {
t.Fatalf("dialSMTP4() error = %v", err)
}
_ = conn.Close()
}
func TestVerifyThreadUpdatedSearchesSentMailbox(t *testing.T) {
fc := &fakeConn{uidValidity: 3, searchUIDs: []uint32{11}}
c := newFakeClient(fc)
ok, err := c.VerifyThreadUpdated(context.Background(), "987654321")
if err != nil {
t.Fatalf("VerifyThreadUpdated() error = %v", err)
}
if !ok {
t.Error("VerifyThreadUpdated() = false, want true when the thread has a sent message")
}
if fc.selected != LabelSent {
t.Errorf("selected = %q, want %q", fc.selected, LabelSent)
}
if len(fc.searched) != 1 || fc.searched[0] != "thrid:987654321" {
t.Errorf("searched = %v, want [thrid:987654321]", fc.searched)
}
}
func TestVerifyThreadUpdatedEmptyThread(t *testing.T) {
c := newFakeClient(&fakeConn{searchUIDs: nil})
if _, err := c.VerifyThreadUpdated(context.Background(), ""); err == nil {
t.Fatal("VerifyThreadUpdated() error = nil, want an error for a missing thread id")
}
}
func TestVerifyThreadUpdatedNoSentMatch(t *testing.T) {
c := newFakeClient(&fakeConn{searchUIDs: nil})
ok, err := c.VerifyThreadUpdated(context.Background(), "42")
if err != nil {
t.Fatalf("VerifyThreadUpdated() error = %v", err)
}
if ok {
t.Error("VerifyThreadUpdated() = true, want false when no sent message matches")
}
}

184
internal/reply/client.go Normal file
View File

@@ -0,0 +1,184 @@
// Package reply drafts short replies with an LLM (FR-8).
//
// The model is reached through OmniRoute, the house AI gateway (AGENTS.md §8,
// §10): every non-JEV model goes through it. OmniRoute exposes an
// OpenAI-compatible /chat/completions endpoint, so this is a tiny client rather
// than a vendor SDK.
package reply
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"io"
"log/slog"
"math/rand"
"net/http"
"strings"
"time"
)
// DefaultBaseURL is the OmniRoute API on the host that runs this service (.13).
// It matches the URL the other house apps use (dsh, worldmonitor). The
// OMNIROUTE_BASE_URL environment variable overrides it.
const DefaultBaseURL = "http://192.168.20.13:20129/v1"
// DefaultModel is the OmniRoute router alias for a good general chat model.
const DefaultModel = "auto/best-chat"
// ErrNotConfigured means no OmniRoute credentials are available, so no draft can
// be generated.
var ErrNotConfigured = errors.New("OmniRoute is not configured: set OMNIROUTE_API_KEY")
// Client is a minimal OpenAI-compatible chat-completions client.
type Client struct {
baseURL string
apiKey string
model string
session string
http *http.Client
logger *slog.Logger
}
// NewClient builds a Client. Empty baseURL/model fall back to the defaults.
func NewClient(baseURL, apiKey, model string, logger *slog.Logger) *Client {
baseURL = strings.TrimRight(strings.TrimSpace(baseURL), "/")
if baseURL == "" {
baseURL = DefaultBaseURL
}
if strings.TrimSpace(model) == "" {
model = DefaultModel
}
if logger == nil {
logger = slog.New(slog.DiscardHandler)
}
return &Client{
baseURL: baseURL,
apiKey: strings.TrimSpace(apiKey),
model: model,
session: fmt.Sprintf("email-agent-%d-%d", time.Now().Unix(), rand.Int63()),
http: &http.Client{Timeout: 90 * time.Second},
logger: logger,
}
}
// Configured reports whether an API key is present.
func (c *Client) Configured() bool { return c.apiKey != "" }
// Model returns the configured model alias.
func (c *Client) Model() string { return c.model }
// BaseURL returns the configured gateway URL (used in diagnostics only).
func (c *Client) BaseURL() string { return c.baseURL }
type chatMessage struct {
Role string `json:"role"`
Content string `json:"content"`
}
type chatRequest struct {
Model string `json:"model"`
Messages []chatMessage `json:"messages"`
Stream bool `json:"stream"`
Temperature float64 `json:"temperature"`
}
type chatResponse struct {
Model string `json:"model"`
Choices []struct {
Message chatMessage `json:"message"`
} `json:"choices"`
Error *struct {
Message string `json:"message"`
Type string `json:"type"`
} `json:"error"`
}
// Complete sends a system and user prompt and returns the assistant text.
func (c *Client) Complete(ctx context.Context, system, user string) (string, error) {
if !c.Configured() {
return "", ErrNotConfigured
}
payload, err := json.Marshal(chatRequest{
Model: c.model,
Messages: []chatMessage{
{Role: "system", Content: system},
{Role: "user", Content: user},
},
Stream: false,
Temperature: 0.3,
})
if err != nil {
return "", fmt.Errorf("encode OmniRoute request: %w", err)
}
const maxAttempts = 3
var lastErr error
for attempt := 1; attempt <= maxAttempts; attempt++ {
text, retryable, err := c.do(ctx, payload)
if err == nil {
return text, nil
}
lastErr = err
if !retryable || attempt == maxAttempts {
break
}
delay := time.Duration(1<<(attempt-1)) * time.Second
c.logger.Warn("omniroute request failed; retrying",
"attempt", attempt, "delay_ms", delay.Milliseconds(), "error", err)
select {
case <-ctx.Done():
return "", ctx.Err()
case <-time.After(delay):
}
}
return "", lastErr
}
func (c *Client) do(ctx context.Context, payload []byte) (string, bool, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost,
c.baseURL+"/chat/completions", bytes.NewReader(payload))
if err != nil {
return "", false, fmt.Errorf("build OmniRoute request: %w", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "application/json")
req.Header.Set("Authorization", "Bearer "+c.apiKey)
// Some OmniRoute upstreams (opencode-go) require a session id.
req.Header.Set("x-opencode-session", c.session)
resp, err := c.http.Do(req)
if err != nil {
return "", true, fmt.Errorf("call OmniRoute: %w", err)
}
defer func() { _ = resp.Body.Close() }()
raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
if err != nil {
return "", true, fmt.Errorf("read OmniRoute response: %w", err)
}
if resp.StatusCode != http.StatusOK {
retryable := resp.StatusCode == http.StatusTooManyRequests ||
resp.StatusCode == http.StatusServiceUnavailable || resp.StatusCode >= 500
return "", retryable, fmt.Errorf("OmniRoute returned %d: %s",
resp.StatusCode, strings.TrimSpace(string(raw)))
}
var out chatResponse
if err := json.Unmarshal(raw, &out); err != nil {
return "", false, fmt.Errorf("decode OmniRoute response: %w", err)
}
if out.Error != nil && out.Error.Message != "" {
return "", false, fmt.Errorf("OmniRoute error: %s", out.Error.Message)
}
if len(out.Choices) == 0 {
return "", false, errors.New("OmniRoute response contained no choices")
}
text := strings.TrimSpace(out.Choices[0].Message.Content)
if text == "" {
return "", false, errors.New("OmniRoute returned an empty reply")
}
return text, false, nil
}

View File

@@ -0,0 +1,84 @@
package reply
import (
"context"
"encoding/json"
"log/slog"
"net/http"
"net/http/httptest"
"testing"
)
func TestCompleteSendsOpenAIRequestAndParsesReply(t *testing.T) {
var (
gotAuth string
gotSession string
gotPath string
gotBody chatRequest
)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotAuth = r.Header.Get("Authorization")
gotSession = r.Header.Get("x-opencode-session")
gotPath = r.URL.Path
if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil {
t.Errorf("decode request: %v", err)
}
_ = json.NewEncoder(w).Encode(chatResponse{
Model: "test-model",
Choices: []struct {
Message chatMessage `json:"message"`
}{{Message: chatMessage{Role: "assistant", Content: " Thanks, got it. "}}},
})
}))
defer srv.Close()
c := NewClient(srv.URL, "secret-key", "auto/best-chat", slog.New(slog.DiscardHandler))
if !c.Configured() {
t.Fatal("client should be configured with a key")
}
got, err := c.Complete(context.Background(), "system prompt", "user prompt")
if err != nil {
t.Fatalf("Complete() error = %v", err)
}
if got != "Thanks, got it." {
t.Errorf("Complete() = %q, want trimmed reply", got)
}
if gotPath != "/chat/completions" {
t.Errorf("path = %q, want /chat/completions", gotPath)
}
if gotAuth != "Bearer secret-key" {
t.Errorf("Authorization = %q", gotAuth)
}
if gotSession == "" {
t.Error("x-opencode-session header was not set")
}
if gotBody.Model != "auto/best-chat" || gotBody.Stream {
t.Errorf("request = %+v, want non-streaming auto/best-chat", gotBody)
}
if len(gotBody.Messages) != 2 || gotBody.Messages[0].Role != "system" || gotBody.Messages[1].Role != "user" {
t.Errorf("messages = %+v", gotBody.Messages)
}
}
func TestCompleteNotConfigured(t *testing.T) {
c := NewClient("http://127.0.0.1:1/v1", "", "auto/best-chat", slog.New(slog.DiscardHandler))
if c.Configured() {
t.Fatal("client should not be configured without a key")
}
if _, err := c.Complete(context.Background(), "s", "u"); err != ErrNotConfigured {
t.Fatalf("Complete() error = %v, want ErrNotConfigured", err)
}
}
func TestCompleteReportsHTTPError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "nope", http.StatusUnauthorized)
}))
defer srv.Close()
c := NewClient(srv.URL, "key", "m", slog.New(slog.DiscardHandler))
_, err := c.Complete(context.Background(), "s", "u")
if err == nil {
t.Fatal("Complete() error = nil, want an error")
}
}

160
internal/reply/draft.go Normal file
View File

@@ -0,0 +1,160 @@
// Draft generation for Phase 5 (FR-8): turn a message into a short reply using
// a template and the OmniRoute gateway.
package reply
import (
"context"
"fmt"
"log/slog"
"strings"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/store"
)
// maxPromptBodyChars bounds how much of the original message is sent to the
// model. Personal mail is usually short; this keeps the request cheap.
const maxPromptBodyChars = 4000
// maxDraftChars bounds the generated body before it is stored.
const maxDraftChars = 8000
// Request is one draft request.
type Request struct {
// Message is the mail being replied to.
Message store.Message
// Category, Importance, and Action come from the latest classification.
Category string
Importance string
Action string
// TemplateID selects a built-in template. Empty picks one from Action.
TemplateID string
// Instruction is optional extra guidance Sam typed (for example "say the
// invoice is paid"). It is appended to the template instruction.
Instruction string
// Signer is the name used for the sign-off (default "Sam").
Signer string
}
// Generator produces reply drafts from an LLM client.
type Generator struct {
client *Client
logger *slog.Logger
}
// NewGenerator builds a Generator.
func NewGenerator(client *Client, logger *slog.Logger) *Generator {
if logger == nil {
logger = slog.New(slog.DiscardHandler)
}
return &Generator{client: client, logger: logger}
}
// Configured reports whether the LLM gateway is configured.
func (g *Generator) Configured() bool { return g.client.Configured() }
// Model returns the configured model alias.
func (g *Generator) Model() string { return g.client.Model() }
// Draft generates a reply body and returns it with the template that was used.
func (g *Generator) Draft(ctx context.Context, req Request) (body, templateID string, err error) {
tmpl, ok := TemplateByID(req.TemplateID)
if !ok {
tmpl = TemplateForAction(req.Action)
if strings.TrimSpace(req.TemplateID) != "" {
g.logger.Warn("unknown reply template; using the action default",
"template", req.TemplateID, "used", tmpl.ID)
}
}
system, user := BuildPrompt(req, tmpl)
body, err = g.client.Complete(ctx, system, user)
if err != nil {
return "", "", err
}
body = cleanDraft(body, maxDraftChars)
if body == "" {
return "", "", fmt.Errorf("the model returned an empty draft")
}
return body, tmpl.ID, nil
}
// BuildPrompt builds the system and user prompts for a draft. It is exported so
// the prompt shape can be tested without a live gateway.
func BuildPrompt(req Request, tmpl Template) (system, user string) {
signer := strings.TrimSpace(req.Signer)
if signer == "" {
signer = "Sam"
}
system = strings.Join([]string{
"You write short, plain-text email replies on behalf of " + signer + ".",
"Return only the reply body: no subject line, no markdown, no placeholders, no quoted history.",
"Match the language of the incoming message.",
"Keep replies under 120 words and end with a sign-off of \"" + signer + "\".",
"If the incoming message is unclear, write a brief acknowledgement and ask one short question.",
"Never invent facts, prices, dates, or commitments. If a detail is missing, leave it out or ask about it.",
}, " ")
var b strings.Builder
b.WriteString("TEMPLATE: ")
b.WriteString(tmpl.Name)
b.WriteString("\n")
b.WriteString("INSTRUCTION: ")
b.WriteString(tmpl.Instruction)
b.WriteString("\n")
if extra := strings.TrimSpace(req.Instruction); extra != "" {
b.WriteString("SAM'S NOTE: ")
b.WriteString(extra)
b.WriteString("\n")
}
if cat := strings.TrimSpace(req.Category); cat != "" {
b.WriteString("CLASSIFICATION: category=")
b.WriteString(cat)
if imp := strings.TrimSpace(req.Importance); imp != "" {
b.WriteString(" importance=")
b.WriteString(imp)
}
if act := strings.TrimSpace(req.Action); act != "" {
b.WriteString(" suggested_action=")
b.WriteString(act)
}
b.WriteString("\n")
}
b.WriteString("\n--- ORIGINAL MESSAGE ---\n")
b.WriteString("From: ")
b.WriteString(strings.TrimSpace(req.Message.FromAddr))
b.WriteString("\n")
b.WriteString("Arrived at: ")
b.WriteString(strings.TrimSpace(req.Message.SourceAddress))
b.WriteString("\n")
b.WriteString("Subject: ")
b.WriteString(strings.TrimSpace(req.Message.Subject))
b.WriteString("\n\n")
b.WriteString(truncate(req.Message.BodyText, maxPromptBodyChars))
b.WriteString("\n--- END ---\n\nWrite the reply body now.")
return system, b.String()
}
// cleanDraft trims a generated draft. Some models wrap output in a code fence or
// prefix it with a "Subject:" line even when told not to; strip those.
func cleanDraft(s string, max int) string {
s = strings.TrimSpace(s)
s = strings.TrimPrefix(s, "```")
s = strings.TrimSuffix(s, "```")
s = strings.TrimSpace(s)
if strings.HasPrefix(strings.ToLower(s), "subject:") {
if i := strings.IndexByte(s, '\n'); i >= 0 {
s = strings.TrimSpace(s[i+1:])
}
}
return truncate(s, max)
}
func truncate(s string, max int) string {
s = strings.TrimSpace(s)
if len(s) <= max {
return s
}
return strings.TrimSpace(s[:max]) + "…"
}

View File

@@ -0,0 +1,118 @@
package reply
import (
"context"
"encoding/json"
"log/slog"
"net/http"
"net/http/httptest"
"strings"
"testing"
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/store"
)
func TestTemplateForAction(t *testing.T) {
if got := TemplateForAction("reply").ID; got != "answer" {
t.Errorf("TemplateForAction(reply) = %q, want answer", got)
}
if got := TemplateForAction("keep").ID; got != "quick-ack" {
t.Errorf("TemplateForAction(keep) = %q, want quick-ack", got)
}
if _, ok := TemplateByID("DECLINE"); !ok {
t.Error("TemplateByID should match case-insensitively")
}
if _, ok := TemplateByID("nonsense"); ok {
t.Error("TemplateByID should reject an unknown template")
}
}
func TestBuildPromptIncludesContext(t *testing.T) {
req := Request{
Message: store.Message{
FromAddr: "customer@example.test",
SourceAddress: "hello@where-woof.com",
Subject: "Booking for Saturday",
BodyText: "Can we book Saturday at 10?",
},
Category: "Action",
Importance: "high",
Action: "reply",
Instruction: "Confirm 10am works",
Signer: "Sam",
}
tmpl, _ := TemplateByID("confirm")
system, user := BuildPrompt(req, tmpl)
if !strings.Contains(system, "Return only the reply body") {
t.Errorf("system prompt missing the output constraint: %q", system)
}
for _, want := range []string{
"Confirm an arrangement",
"Confirm 10am works",
"category=Action",
"importance=high",
"suggested_action=reply",
"customer@example.test",
"hello@where-woof.com",
"Booking for Saturday",
"Can we book Saturday at 10?",
} {
if !strings.Contains(user, want) {
t.Errorf("user prompt missing %q\n%s", want, user)
}
}
}
func TestCleanDraftStripsFenceAndSubject(t *testing.T) {
if got := cleanDraft("```\nHello there.\n```", 1000); got != "Hello there." {
t.Errorf("cleanDraft fence = %q", got)
}
if got := cleanDraft("Subject: Re: hi\n\nHello there.", 1000); got != "Hello there." {
t.Errorf("cleanDraft subject = %q", got)
}
}
func TestGeneratorDraftUsesTemplate(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(chatResponse{
Choices: []struct {
Message chatMessage `json:"message"`
}{{Message: chatMessage{Role: "assistant", Content: "Sure, Saturday works.\n\nSam"}}},
})
}))
defer srv.Close()
g := NewGenerator(NewClient(srv.URL, "key", "auto/best-chat", slog.New(slog.DiscardHandler)),
slog.New(slog.DiscardHandler))
body, templateID, err := g.Draft(context.Background(), Request{
Message: store.Message{FromAddr: "a@b.test", Subject: "hi", BodyText: "?"},
Action: "reply",
})
if err != nil {
t.Fatalf("Draft() error = %v", err)
}
if templateID != "answer" {
t.Errorf("template = %q, want answer (the default for action=reply)", templateID)
}
if !strings.Contains(body, "Saturday") {
t.Errorf("body = %q", body)
}
}
func TestGeneratorDraftEmptyReplyIsAnError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(chatResponse{
Choices: []struct {
Message chatMessage `json:"message"`
}{{Message: chatMessage{Role: "assistant", Content: " "}}},
})
}))
defer srv.Close()
g := NewGenerator(NewClient(srv.URL, "key", "m", slog.New(slog.DiscardHandler)),
slog.New(slog.DiscardHandler))
if _, _, err := g.Draft(context.Background(), Request{Message: store.Message{Subject: "x"}}); err == nil {
t.Fatal("Draft() error = nil, want an error for an empty reply")
}
}

View File

@@ -0,0 +1,71 @@
// Reply templates for Phase 5 (FR-8).
//
// A template is a short instruction that shapes the LLM draft. It is not the
// final text: the model fills it in from the message, the classification, and
// any extra note Sam types. Templates are built in for v1; a template editor is
// not in the Phase 5 requirements.
package reply
import "strings"
// Template is one reply shape.
type Template struct {
ID string
Name string
Description string
// Instruction is the template-specific part of the prompt.
Instruction string
}
// Templates is the built-in template list, in menu order.
var Templates = []Template{
{
ID: "quick-ack",
Name: "Quick acknowledgement",
Description: "Confirm receipt in one or two short sentences.",
Instruction: "Acknowledge that the message was received. If the sender asked for something, say when Sam will look at it. Do not promise anything Sam has not agreed to.",
},
{
ID: "answer",
Name: "Answer the question",
Description: "Answer the sender's question directly.",
Instruction: "Answer the sender's question directly and briefly. If the message does not contain a clear question, summarise what Sam understood and ask one clarifying question.",
},
{
ID: "confirm",
Name: "Confirm an arrangement",
Description: "Confirm a booking, time, or arrangement.",
Instruction: "Confirm the arrangement in the message: restate the key detail (date, time, or reference) and say it works for Sam.",
},
{
ID: "decline",
Name: "Politely decline",
Description: "Decline or say no thanks, politely.",
Instruction: "Politely decline the request or offer. Thank the sender and keep it warm but brief. Do not over-explain.",
},
}
// TemplateByID returns the template with the given id. ok is false when it is
// unknown. The id match is case-insensitive.
func TemplateByID(id string) (Template, bool) {
id = strings.ToLower(strings.TrimSpace(id))
for _, t := range Templates {
if t.ID == id {
return t, true
}
}
return Template{}, false
}
// TemplateForAction picks a sensible default template from the classifier's
// recommended action. An unknown action falls back to a quick acknowledgement.
func TemplateForAction(action string) Template {
switch strings.ToLower(strings.TrimSpace(action)) {
case "reply":
t, _ := TemplateByID("answer")
return t
default:
t, _ := TemplateByID("quick-ack")
return t
}
}

369
internal/server/replies.go Normal file
View File

@@ -0,0 +1,369 @@
// Reply drafting and sending for Phase 5 (FR-8, FR-15).
//
// The flow is: open a message, generate a draft with the LLM (through
// OmniRoute) from a template, edit it, then send it. Sending is always a
// deliberate UI action (AGENTS.md §10): the handler requires a CSRF token and an
// explicit confirmation, and it never auto-sends. A sent reply moves the
// original to its Replied-to/<address> label and is logged (FR-13).
package server
import (
"context"
"database/sql"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"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/store"
)
// ReplyGenerator drafts a reply body. The live implementation is
// *reply.Generator; tests substitute a fake.
type ReplyGenerator interface {
Draft(ctx context.Context, req reply.Request) (body, templateID string, err error)
Configured() bool
Model() string
}
// MailSender sends a reply and can check that it joined the original thread.
// The live implementation is *mailops.Client; tests substitute a fake.
type MailSender interface {
Send(ctx context.Context, msg mailops.OutgoingMessage) error
VerifyThreadUpdated(ctx context.Context, threadID string) (bool, error)
}
// replyData is the template model for the message detail pane.
type replyData struct {
Message store.Message
Category string
Importance string
Action string
Confidence int
Classified bool
HasDraft bool
Draft store.Draft
Templates []reply.Template
TemplateID string
Instruction string
From string
To string
Subject string
CSRFToken string
Note string
Error string
History []store.ActionLogEntry
LLMReady bool
LLMModel string
Sent bool
ThreadUpdated bool
RepliedFolder string
// NoThreadHeader is true when the original message has no stored Message-ID,
// so the reply cannot set In-Reply-To/References.
NoThreadHeader bool
}
// handleMessageDetail renders the detail pane for one message.
func (s *Server) handleMessageDetail(w http.ResponseWriter, r *http.Request) {
id, ok := parseMessageID(r)
if !ok {
http.Error(w, "unknown message", http.StatusBadRequest)
return
}
s.renderMessage(w, r, id, "", "", false, false)
}
// handleDraft generates a reply draft with the LLM and stores it.
func (s *Server) handleDraft(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 {
http.Error(w, "could not read the form", http.StatusBadRequest)
return
}
id, ok := parseMessageID(r)
if !ok {
http.Error(w, "unknown message", http.StatusBadRequest)
return
}
if !s.replies.Configured() {
s.renderMessage(w, r, id, "", "Reply drafting is not connected yet: set OMNIROUTE_API_KEY.", false, false)
return
}
ctx := r.Context()
msg, err := s.store.MessageByID(ctx, id)
if err != nil {
s.renderMessage(w, r, id, "", "Could not load that message.", false, false)
return
}
class := s.latestClass(ctx, id)
body, templateID, err := s.replies.Draft(ctx, reply.Request{
Message: msg,
Category: class.Category,
Importance: class.Importance,
Action: class.Action,
TemplateID: r.FormValue("template"),
Instruction: r.FormValue("instruction"),
Signer: "Sam",
})
if err != nil {
s.logger.Warn("draft generation failed", "message_id", id, "error", err)
s.renderMessage(w, r, id, "", "Draft generation failed: "+err.Error(), false, false)
return
}
if _, err := s.store.SaveDraft(ctx, id, body, templateID); err != nil {
s.logger.Error("save draft", "message_id", id, "error", err)
s.renderMessage(w, r, id, "", "Could not save the draft.", false, false)
return
}
s.logMessageAction(ctx, id, "draft", "Sam",
fmt.Sprintf("template=%s model=%s chars=%d", templateID, s.replies.Model(), len(body)))
s.renderMessage(w, r, id,
fmt.Sprintf("Draft generated with the %q template. Edit it, then send.", templateID), "", false, false)
}
// handleSaveDraft stores Sam's edits to a draft.
func (s *Server) handleSaveDraft(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 {
http.Error(w, "could not read the form", http.StatusBadRequest)
return
}
id, ok := parseMessageID(r)
if !ok {
http.Error(w, "unknown message", http.StatusBadRequest)
return
}
draftID, err := strconv.ParseInt(strings.TrimSpace(r.FormValue("draft_id")), 10, 64)
if err != nil || draftID <= 0 {
s.renderMessage(w, r, id, "", "There is no draft to save.", false, false)
return
}
if err := s.store.UpdateDraftBody(r.Context(), draftID, r.FormValue("body")); err != nil {
s.logger.Error("update draft", "draft_id", draftID, "error", err)
s.renderMessage(w, r, id, "", "Could not save the draft.", false, false)
return
}
s.logMessageAction(r.Context(), id, "draft_edit", "Sam", fmt.Sprintf("draft_id=%d", draftID))
s.renderMessage(w, r, id, "Draft saved.", "", false, false)
}
// handleSendReply sends the (edited) draft. It is the only handler that can send
// mail, it needs the CSRF token and confirm=yes, and it never runs from a
// schedule.
func (s *Server) handleSendReply(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 {
http.Error(w, "could not read the form", http.StatusBadRequest)
return
}
id, ok := parseMessageID(r)
if !ok {
http.Error(w, "unknown message", http.StatusBadRequest)
return
}
if r.FormValue("confirm") != "yes" {
s.renderMessage(w, r, id, "", "Sending needs confirmation. Press Send reply again and confirm.", false, false)
return
}
ctx := r.Context()
msg, err := s.store.MessageByID(ctx, id)
if err != nil {
s.renderMessage(w, r, id, "", "Could not load that message.", false, false)
return
}
draftID, err := strconv.ParseInt(strings.TrimSpace(r.FormValue("draft_id")), 10, 64)
if err != nil || draftID <= 0 {
s.renderMessage(w, r, id, "", "There is no draft to send.", false, false)
return
}
draft, err := s.store.DraftByID(ctx, draftID)
if err != nil {
s.renderMessage(w, r, id, "", "Could not load the draft.", false, false)
return
}
if draft.Status == store.DraftStatusSent {
s.renderMessage(w, r, id, "", "That draft has already been sent.", false, false)
return
}
// Save any last edits before sending.
body := strings.TrimSpace(r.FormValue("body"))
if body == "" {
body = draft.Body
}
if body == "" {
s.renderMessage(w, r, id, "", "The reply is empty.", false, false)
return
}
if body != draft.Body {
if err := s.store.UpdateDraftBody(ctx, draftID, body); err != nil {
s.logger.Error("update draft before send", "draft_id", draftID, "error", err)
}
}
to := mailops.ReplyTo(msg)
if to == "" {
s.renderMessage(w, r, id, "", "This message has no sender address to reply to.", false, false)
return
}
out := mailops.OutgoingMessage{
To: to,
From: mailops.ReplyIdentity(msg, s.cfg.GmailUser),
Subject: mailops.ReplySubject(msg.Subject),
InReplyTo: msg.MessageIDHeader,
References: msg.MessageIDHeader,
Body: body,
}
if err := s.sender.Send(ctx, out); err != nil {
s.logger.Error("send reply failed", "message_id", id, "to", to, "error", err)
s.renderMessage(w, r, id, "", "Send failed: "+err.Error(), false, false)
return
}
if err := s.store.MarkDraftSent(ctx, draftID); err != nil {
s.logger.Error("mark draft sent", "draft_id", draftID, "error", err)
}
s.logMessageAction(ctx, id, "reply_sent", "Sam",
fmt.Sprintf("to=%s from=%s template=%s", to, out.From, draft.TemplateID))
// File the original under Replied-to/<arrived-at> (FR-9). Best effort: the
// reply is already sent, so a label failure is logged, not fatal.
label, merr := s.actions.Move(ctx, msg, store.FolderReplied)
if merr != nil {
s.logger.Warn("move replied message failed", "message_id", id, "error", merr)
s.logMessageAction(ctx, id, "reply_filed", "agent", "failed="+merr.Error())
} else if serr := s.store.SetMessageLocation(ctx, id, store.FolderReplied, label); serr != nil {
s.logger.Error("record replied location", "message_id", id, "error", serr)
} else {
s.logMessageAction(ctx, id, "move", "Sam",
fmt.Sprintf("to=%s label=%q from=%q", store.FolderReplied, label, msg.Mailbox))
}
// Confirm the thread updated. This is read-only and best effort.
threadUpdated := false
if strings.TrimSpace(msg.ThreadID) != "" {
if ok, verr := s.sender.VerifyThreadUpdated(ctx, msg.ThreadID); verr != nil {
s.logger.Warn("verify thread update failed", "message_id", id, "error", verr)
} else {
threadUpdated = ok
}
}
note := fmt.Sprintf("Reply sent to %s as %s.", to, out.From)
if !threadUpdated && strings.TrimSpace(msg.ThreadID) != "" {
note += " Could not yet confirm the thread update in Gmail; it may take a moment to appear."
}
s.renderMessage(w, r, id, note, "", true, threadUpdated)
}
// renderMessage loads and renders the message detail pane.
func (s *Server) renderMessage(w http.ResponseWriter, r *http.Request, id int64, note, errMsg string, sent, threadUpdated bool) {
ctx := r.Context()
msg, err := s.store.MessageByID(ctx, id)
if errors.Is(err, sql.ErrNoRows) {
http.Error(w, "message not found", http.StatusNotFound)
return
}
if err != nil {
s.logger.Error("load message detail", "message_id", id, "error", err)
http.Error(w, "could not load message", http.StatusInternalServerError)
return
}
class := s.latestClass(ctx, id)
data := replyData{
Message: msg,
Category: class.Category,
Importance: class.Importance,
Action: class.Action,
Confidence: confidencePercent(class.Confidence),
Classified: class.Category != "",
Templates: reply.Templates,
From: mailops.ReplyIdentity(msg, s.cfg.GmailUser),
To: mailops.ReplyTo(msg),
Subject: mailops.ReplySubject(msg.Subject),
CSRFToken: s.csrfToken(w, r),
Note: note,
Error: errMsg,
LLMReady: s.replies.Configured(),
LLMModel: s.replies.Model(),
Sent: sent,
ThreadUpdated: threadUpdated,
}
if data.Action == "" {
data.Action = "Review"
}
data.NoThreadHeader = strings.TrimSpace(msg.MessageIDHeader) == ""
if d, ok, derr := s.store.LatestDraft(ctx, id); derr != nil {
s.logger.Error("load draft", "message_id", id, "error", derr)
} else if ok {
data.HasDraft = true
data.Draft = d
templateID := d.TemplateID
if templateID == "" {
templateID = reply.TemplateForAction(class.Action).ID
}
data.TemplateID = templateID
}
if history, herr := s.store.MessageActions(ctx, id, 20); herr != nil {
s.logger.Error("load message history", "message_id", id, "error", herr)
} else {
data.History = history
}
if !data.HasDraft {
data.TemplateID = reply.TemplateForAction(class.Action).ID
}
w.Header().Set("Content-Type", "text/html; charset=utf-8")
if err := s.tmpl.ExecuteTemplate(w, "message.html", data); err != nil {
s.logger.Error("render message detail", "error", err)
}
}
// latestClass returns the newest classification for a message.
func (s *Server) latestClass(ctx context.Context, id int64) store.Classification {
classes, err := s.store.LatestClassifications(ctx, []int64{id})
if err != nil {
s.logger.Error("load latest classification", "message_id", id, "error", err)
return store.Classification{}
}
return classes[id]
}
// parseMessageID reads the message id from the query or the posted form.
func parseMessageID(r *http.Request) (int64, bool) {
raw := strings.TrimSpace(r.FormValue("id"))
if raw == "" {
return 0, false
}
id, err := strconv.ParseInt(raw, 10, 64)
if err != nil || id <= 0 {
return 0, false
}
return id, true
}
// Ensure the live implementations satisfy the interfaces at compile time.
var (
_ ReplyGenerator = (*reply.Generator)(nil)
_ MailSender = (*mailops.Client)(nil)
)

View File

@@ -0,0 +1,300 @@
package server
import (
"context"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"testing"
"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/store"
)
type fakeGenerator struct {
body string
templateID string
err error
calls int
lastReq reply.Request
}
func (f *fakeGenerator) Draft(_ context.Context, req reply.Request) (string, string, error) {
f.calls++
f.lastReq = req
if f.err != nil {
return "", "", f.err
}
return f.body, f.templateID, nil
}
func (f *fakeGenerator) Configured() bool { return true }
func (f *fakeGenerator) Model() string { return "fake-model" }
type fakeSender struct {
sent []mailops.OutgoingMessage
err error
threadUpdated bool
verifyErr error
verifyCalls int
}
func (f *fakeSender) Send(_ context.Context, msg mailops.OutgoingMessage) error {
if f.err != nil {
return f.err
}
f.sent = append(f.sent, msg)
return nil
}
func (f *fakeSender) VerifyThreadUpdated(_ context.Context, _ string) (bool, error) {
f.verifyCalls++
return f.threadUpdated, f.verifyErr
}
func newReplyServer(t *testing.T) (*Server, *fakeActions, *fakeGenerator, *fakeSender) {
t.Helper()
srv, fa := newActionsServer(t)
fg := &fakeGenerator{body: "Thanks, Saturday works.\n\nSam", templateID: "confirm"}
fs := &fakeSender{threadUpdated: true}
srv.replies = fg
srv.sender = fs
return srv, fa, fg, fs
}
func seedReplyable(t *testing.T, srv *Server, gmailID, source, messageID string) int64 {
t.Helper()
ctx := context.Background()
id, err := srv.store.UpsertMessage(ctx, &store.Message{
GmailID: gmailID,
ThreadID: "1234567890",
MessageIDHeader: messageID,
FromAddr: "Customer Name <cust@example.test>",
ToAddr: source,
Subject: "Booking for Saturday",
BodyText: "Can we book Saturday at 10?",
SourceAddress: source,
})
if err != nil {
t.Fatalf("seed message: %v", err)
}
if _, err := srv.store.SaveClassification(ctx, store.Classification{
MessageID: id, Category: "Action", Importance: "high", Action: "reply", Confidence: 0.8,
}); err != nil {
t.Fatalf("seed classification: %v", err)
}
return id
}
func TestMessageDetailRendersReplyIdentity(t *testing.T) {
srv, _, _, _ := newReplyServer(t)
session, _, _ := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
rec := getWithSession(t, srv, "/partials/message?id="+itoa(id), session)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}
body := rec.Body.String()
for _, want := range []string{
"hello@where-woof.com",
"Sam Rolfe",
"cust@example.test",
"Re: Booking for Saturday",
"Can we book Saturday at 10?",
"Draft reply",
} {
if !strings.Contains(body, want) {
t.Errorf("detail pane missing %q", want)
}
}
}
func TestMessageDetailRequiresAuth(t *testing.T) {
srv, _, _, _ := newReplyServer(t)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
req := httptest.NewRequest(http.MethodGet, "/partials/message?id="+itoa(id), nil)
rec := httptest.NewRecorder()
srv.Handler().ServeHTTP(rec, req)
if rec.Code != http.StatusSeeOther {
t.Fatalf("status = %d, want %d", rec.Code, http.StatusSeeOther)
}
}
func TestDraftRequiresCSRF(t *testing.T) {
srv, _, fg, _ := newReplyServer(t)
session, _, _ := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
form := url.Values{"id": {itoa(id)}, "template": {"quick-ack"}}
rec := postForm(t, srv, "/actions/draft", session, nil, form)
if rec.Code != http.StatusForbidden {
t.Fatalf("status = %d, want 403", rec.Code)
}
if fg.calls != 0 {
t.Errorf("generator called %d times, want 0", fg.calls)
}
}
func TestDraftGeneratesAndSaves(t *testing.T) {
srv, _, fg, _ := newReplyServer(t)
session, csrfCookie, csrf := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
form := url.Values{"csrf": {csrf}, "id": {itoa(id)}, "template": {"confirm"}, "instruction": {"10am works"}}
rec := postForm(t, srv, "/actions/draft", session, csrfCookie, form)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200 (body %s)", rec.Code, rec.Body.String())
}
if fg.calls != 1 {
t.Fatalf("generator calls = %d, want 1", fg.calls)
}
if fg.lastReq.Instruction != "10am works" || fg.lastReq.TemplateID != "confirm" {
t.Errorf("generator request = %+v", fg.lastReq)
}
if !strings.Contains(rec.Body.String(), "Thanks, Saturday works.") {
t.Errorf("response does not show the generated draft")
}
d, ok, err := srv.store.LatestDraft(context.Background(), id)
if err != nil || !ok {
t.Fatalf("LatestDraft() ok=%v err=%v", ok, err)
}
if d.TemplateID != "confirm" || d.Status != store.DraftStatusDraft {
t.Errorf("stored draft = %+v", d)
}
}
func TestSaveDraftEditsBody(t *testing.T) {
srv, _, _, _ := newReplyServer(t)
session, csrfCookie, csrf := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
d, err := srv.store.SaveDraft(context.Background(), id, "original", "quick-ack")
if err != nil {
t.Fatalf("SaveDraft() error = %v", err)
}
form := url.Values{"csrf": {csrf}, "id": {itoa(id)}, "draft_id": {itoa(d.ID)}, "body": {"edited by Sam"}}
rec := postForm(t, srv, "/actions/draft/save", session, csrfCookie, form)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}
got, _, err := srv.store.LatestDraft(context.Background(), id)
if err != nil {
t.Fatalf("LatestDraft() error = %v", err)
}
if got.Body != "edited by Sam" {
t.Errorf("body = %q, want the edit", got.Body)
}
}
func TestSendRequiresConfirmation(t *testing.T) {
srv, _, _, fs := newReplyServer(t)
session, csrfCookie, csrf := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
d, _ := srv.store.SaveDraft(context.Background(), id, "body", "quick-ack")
form := url.Values{"csrf": {csrf}, "id": {itoa(id)}, "draft_id": {itoa(d.ID)}, "body": {"body"}}
rec := postForm(t, srv, "/actions/send", session, csrfCookie, form)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}
if len(fs.sent) != 0 {
t.Fatalf("sent %d messages without confirmation", len(fs.sent))
}
if !strings.Contains(rec.Body.String(), "needs confirmation") {
t.Errorf("body did not explain the confirmation requirement")
}
}
func TestSendReplySendsFilesAndVerifiesThread(t *testing.T) {
srv, fa, _, fs := newReplyServer(t)
session, csrfCookie, csrf := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
d, _ := srv.store.SaveDraft(context.Background(), id, "Saturday works.\n\nSam", "confirm")
form := url.Values{
"csrf": {csrf}, "id": {itoa(id)}, "draft_id": {itoa(d.ID)},
"body": {"Saturday works.\n\nSam"}, "confirm": {"yes"},
}
rec := postForm(t, srv, "/actions/send", session, csrfCookie, form)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200 (body %s)", rec.Code, rec.Body.String())
}
if len(fs.sent) != 1 {
t.Fatalf("sent = %d, want 1", len(fs.sent))
}
out := fs.sent[0]
if out.To != "cust@example.test" {
t.Errorf("To = %q", out.To)
}
if out.From != `"Sam Rolfe" <hello@where-woof.com>` {
t.Errorf("From = %q", out.From)
}
if out.Subject != "Re: Booking for Saturday" {
t.Errorf("Subject = %q", out.Subject)
}
if out.InReplyTo != "<orig-1@mail.example>" || out.References != "<orig-1@mail.example>" {
t.Errorf("threading headers = %q / %q", out.InReplyTo, out.References)
}
if fs.verifyCalls != 1 {
t.Errorf("verifyCalls = %d, want 1", fs.verifyCalls)
}
if !strings.Contains(rec.Body.String(), "Reply sent") {
t.Errorf("response does not confirm the send")
}
// The draft is marked sent and the message is filed under Replied-to.
got, _, err := srv.store.LatestDraft(context.Background(), id)
if err != nil {
t.Fatalf("LatestDraft() error = %v", err)
}
if got.Status != store.DraftStatusSent {
t.Errorf("draft status = %q, want sent", got.Status)
}
if fa.moveCount() != 1 {
t.Errorf("moves = %d, want 1 (file the original)", fa.moveCount())
}
msgs, _ := srv.store.MessagesByIDs(context.Background(), []int64{id})
if len(msgs) != 1 || msgs[0].Folder != store.FolderReplied {
t.Errorf("message = %+v, want folder replied", msgs)
}
}
func TestSendReplyAlreadySentRefused(t *testing.T) {
srv, _, _, fs := newReplyServer(t)
session, csrfCookie, csrf := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "<orig-1@mail.example>")
d, _ := srv.store.SaveDraft(context.Background(), id, "body", "quick-ack")
if err := srv.store.MarkDraftSent(context.Background(), d.ID); err != nil {
t.Fatalf("MarkDraftSent() error = %v", err)
}
form := url.Values{
"csrf": {csrf}, "id": {itoa(id)}, "draft_id": {itoa(d.ID)},
"body": {"body"}, "confirm": {"yes"},
}
rec := postForm(t, srv, "/actions/send", session, csrfCookie, form)
if rec.Code != http.StatusOK {
t.Fatalf("status = %d, want 200", rec.Code)
}
if len(fs.sent) != 0 {
t.Fatalf("sent = %d, want 0 for an already-sent draft", len(fs.sent))
}
if !strings.Contains(rec.Body.String(), "already been sent") {
t.Errorf("body did not explain the duplicate guard")
}
}
func TestSendReplyNotesMissingThreadHeader(t *testing.T) {
srv, _, _, _ := newReplyServer(t)
session, _, _ := login(t, srv)
id := seedReplyable(t, srv, "1", "hello@where-woof.com", "")
rec := getWithSession(t, srv, "/partials/message?id="+itoa(id), session)
if !strings.Contains(rec.Body.String(), "cannot set threading headers") {
t.Errorf("detail pane did not warn about the missing Message-ID")
}
}

View File

@@ -21,6 +21,7 @@ import (
"gitea.lab.audasmedia.com.au/sam/email_agent/internal/config"
"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/version"
@@ -43,6 +44,8 @@ type Server struct {
fetcher *mailfetch.Fetcher
classifier *classify.Classifier
actions MailActions
replies ReplyGenerator
sender MailSender
}
// New builds a Server. It does not start listening.
@@ -61,6 +64,7 @@ func New(cfg *config.Config, st *store.Store, logger *slog.Logger) (*Server, err
return nil, fmt.Errorf("session manager: %w", err)
}
mail := mailops.New(cfg, logger)
s := &Server{
cfg: cfg,
store: st,
@@ -72,7 +76,12 @@ func New(cfg *config.Config, st *store.Store, logger *slog.Logger) (*Server, err
classify.NewClient(cfg.JEVBaseURL, cfg.JEVAPIKey, cfg.JEVModel, logger),
st, logger, cfg.ClassifyConcurrency, cfg.ClassifyMaxMessages,
),
actions: mailops.New(cfg, logger),
actions: mail,
sender: mail,
replies: reply.NewGenerator(
reply.NewClient(cfg.OmniRouteBaseURL, cfg.OmniRouteAPIKey, cfg.OmniRouteModel, logger),
logger,
),
}
s.http = &http.Server{
Addr: cfg.Addr,
@@ -128,6 +137,12 @@ func (s *Server) Handler() http.Handler {
// Phase 4 (FR-6/FR-9): bulk triage actions (delete, move, mark) and restore.
mux.HandleFunc("POST /actions/bulk", s.requireAuth(s.handleBulkAction))
// Phase 5 (FR-8/FR-15): message detail, reply draft, edit, and send.
mux.HandleFunc("GET /partials/message", s.requireAuth(s.handleMessageDetail))
mux.HandleFunc("POST /actions/draft", s.requireAuth(s.handleDraft))
mux.HandleFunc("POST /actions/draft/save", s.requireAuth(s.handleSaveDraft))
mux.HandleFunc("POST /actions/send", s.requireAuth(s.handleSendReply))
// FR-14: editable category list.
mux.HandleFunc("GET /categories", s.requireAuth(s.handleCategories))
mux.HandleFunc("POST /categories/add", s.requireAuth(s.handleCategoryAdd))

117
internal/store/drafts.go Normal file
View File

@@ -0,0 +1,117 @@
// Reply-draft persistence for Phase 5 (FR-8).
//
// A draft belongs to one message. It is rewritten in place while Sam edits it
// and marked `sent` after a successful SMTP send. Only the newest draft per
// message is shown in the UI; history is kept in the actions_log (FR-13).
package store
import (
"context"
"database/sql"
"errors"
"fmt"
"time"
)
// Draft statuses. A draft moves draft -> sent once, on a successful send.
const (
DraftStatusDraft = "draft"
DraftStatusSent = "sent"
)
// Draft is one reply draft row.
type Draft struct {
ID int64
MessageID int64
Body string
TemplateID string
Status string
SentAt time.Time
CreatedAt time.Time
UpdatedAt time.Time
}
// SaveDraft inserts a new draft for a message and returns it.
func (s *Store) SaveDraft(ctx context.Context, messageID int64, body, templateID string) (Draft, error) {
const q = `INSERT INTO drafts (message_id, body, template_id, status)
VALUES (?, ?, ?, 'draft')`
res, err := s.db.ExecContext(ctx, q, messageID, body, nullString(templateID))
if err != nil {
return Draft{}, fmt.Errorf("save draft for message %d: %w", messageID, err)
}
id, err := res.LastInsertId()
if err != nil {
return Draft{}, fmt.Errorf("read draft id: %w", err)
}
return s.DraftByID(ctx, id)
}
// DraftByID loads one draft.
func (s *Store) DraftByID(ctx context.Context, id int64) (Draft, error) {
const q = `
SELECT id, message_id, body, COALESCE(template_id, ''), status,
COALESCE(sent_at, ''), created_at, updated_at
FROM drafts WHERE id = ?`
return scanDraft(s.db.QueryRowContext(ctx, q, id))
}
// LatestDraft returns the newest draft for a message. ok is false when the
// message has no draft yet.
func (s *Store) LatestDraft(ctx context.Context, messageID int64) (Draft, bool, error) {
const q = `
SELECT id, message_id, body, COALESCE(template_id, ''), status,
COALESCE(sent_at, ''), created_at, updated_at
FROM drafts WHERE message_id = ? ORDER BY id DESC LIMIT 1`
d, err := scanDraft(s.db.QueryRowContext(ctx, q, messageID))
if errors.Is(err, sql.ErrNoRows) {
return Draft{}, false, nil
}
if err != nil {
return Draft{}, false, err
}
return d, true, nil
}
// UpdateDraftBody replaces the body of a draft Sam edited and bumps updated_at.
func (s *Store) UpdateDraftBody(ctx context.Context, id int64, body string) error {
const q = `UPDATE drafts SET body = ?, updated_at = datetime('now') WHERE id = ?`
res, err := s.db.ExecContext(ctx, q, body, id)
if err != nil {
return fmt.Errorf("update draft %d: %w", id, err)
}
if n, _ := res.RowsAffected(); n == 0 {
return fmt.Errorf("update draft %d: not found", id)
}
return nil
}
// MarkDraftSent records a successful send.
func (s *Store) MarkDraftSent(ctx context.Context, id int64) error {
const q = `UPDATE drafts SET status = 'sent', sent_at = datetime('now'),
updated_at = datetime('now') WHERE id = ?`
res, err := s.db.ExecContext(ctx, q, id)
if err != nil {
return fmt.Errorf("mark draft %d sent: %w", id, err)
}
if n, _ := res.RowsAffected(); n == 0 {
return fmt.Errorf("mark draft %d sent: not found", id)
}
return nil
}
func scanDraft(row *sql.Row) (Draft, error) {
var (
d Draft
sentAt string
created string
updated string
)
if err := row.Scan(&d.ID, &d.MessageID, &d.Body, &d.TemplateID, &d.Status,
&sentAt, &created, &updated); err != nil {
return Draft{}, err
}
d.SentAt = parseSQLTime(sentAt)
d.CreatedAt = parseSQLTime(created)
d.UpdatedAt = parseSQLTime(updated)
return d, nil
}

View File

@@ -0,0 +1,146 @@
package store
import (
"context"
"database/sql"
"path/filepath"
"testing"
)
func TestDraftsRoundTrip(t *testing.T) {
s, ctx := testStore(t)
id := seedMessage(t, s, ctx, "1", "question", "hello@where-woof.com")
d, err := s.SaveDraft(ctx, id, "Thanks, got it.", "quick-ack")
if err != nil {
t.Fatalf("SaveDraft() error = %v", err)
}
if d.ID == 0 || d.Status != DraftStatusDraft || d.TemplateID != "quick-ack" {
t.Fatalf("SaveDraft() = %+v", d)
}
if d.MessageID != id {
t.Errorf("MessageID = %d, want %d", d.MessageID, id)
}
if err := s.UpdateDraftBody(ctx, d.ID, "Edited body."); err != nil {
t.Fatalf("UpdateDraftBody() error = %v", err)
}
latest, ok, err := s.LatestDraft(ctx, id)
if err != nil || !ok {
t.Fatalf("LatestDraft() ok=%v err=%v", ok, err)
}
if latest.Body != "Edited body." || latest.Status != DraftStatusDraft {
t.Errorf("latest = %+v", latest)
}
if err := s.MarkDraftSent(ctx, latest.ID); err != nil {
t.Fatalf("MarkDraftSent() error = %v", err)
}
sent, _, err := s.LatestDraft(ctx, id)
if err != nil {
t.Fatalf("LatestDraft() after send error = %v", err)
}
if sent.Status != DraftStatusSent || sent.SentAt.IsZero() {
t.Errorf("sent draft = %+v, want status=sent and a sent_at", sent)
}
}
func TestLatestDraftMissing(t *testing.T) {
s, ctx := testStore(t)
id := seedMessage(t, s, ctx, "1", "x", "a@b.test")
if _, ok, err := s.LatestDraft(ctx, id); err != nil || ok {
t.Fatalf("LatestDraft() ok=%v err=%v, want ok=false", ok, err)
}
}
func TestMessageByIDIncludesReplyFields(t *testing.T) {
s, ctx := testStore(t)
id, err := s.UpsertMessage(ctx, &Message{
GmailID: "99",
FromAddr: "Customer <cust@example.test>",
Subject: "Booking",
BodyText: "Saturday?",
SourceAddress: "hello@where-woof.com",
MessageIDHeader: "<orig-1@mail.example>",
})
if err != nil {
t.Fatalf("UpsertMessage() error = %v", err)
}
msg, err := s.MessageByID(ctx, id)
if err != nil {
t.Fatalf("MessageByID() error = %v", err)
}
if msg.MessageIDHeader != "<orig-1@mail.example>" {
t.Errorf("MessageIDHeader = %q", msg.MessageIDHeader)
}
if msg.BodyText != "Saturday?" || msg.SourceAddress != "hello@where-woof.com" {
t.Errorf("message = %+v", msg)
}
}
func TestMigrationV5AddsReplyColumns(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "old.db")
// Simulate a v4 database: messages has no message_id_header and drafts has
// no sent_at. Opening it must apply migration v5.
db, err := sql.Open("sqlite", "file:"+path)
if err != nil {
t.Fatalf("open old db: %v", err)
}
stmts := []string{
`CREATE TABLE messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
gmail_id TEXT NOT NULL UNIQUE,
thread_id TEXT, from_addr TEXT, to_addr TEXT, subject TEXT,
body_text TEXT, source_address TEXT, date_sent TEXT,
imap_uid INTEGER, imap_uidvalidity INTEGER,
folder TEXT NOT NULL DEFAULT 'inbox',
mailbox TEXT NOT NULL DEFAULT 'INBOX',
received_at TEXT NOT NULL DEFAULT (datetime('now')),
created_at TEXT NOT NULL DEFAULT (datetime('now')))`,
`CREATE TABLE drafts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
message_id INTEGER NOT NULL,
body TEXT NOT NULL,
template_id TEXT,
status TEXT NOT NULL DEFAULT 'draft',
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now')))`,
`CREATE TABLE schema_migrations (version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL DEFAULT (datetime('now')))`,
`INSERT INTO schema_migrations (version) VALUES (4)`,
}
for _, q := range stmts {
if _, err := db.ExecContext(ctx, q); err != nil {
t.Fatalf("seed old db: %v", err)
}
}
_ = db.Close()
s, err := Open(ctx, path)
if err != nil {
t.Fatalf("Open() on v4 database error = %v", err)
}
defer s.Close()
v, err := s.SchemaVersion(ctx)
if err != nil {
t.Fatalf("SchemaVersion() error = %v", err)
}
if v < 5 {
t.Errorf("SchemaVersion() = %d, want >= 5", v)
}
for _, tc := range []struct{ table, col string }{
{"messages", "message_id_header"},
{"drafts", "sent_at"},
} {
var n int
q := `SELECT COUNT(*) FROM pragma_table_info('` + tc.table + `') WHERE name = ?`
if err := s.DB().QueryRowContext(ctx, q, tc.col).Scan(&n); err != nil {
t.Fatalf("count %s.%s: %v", tc.table, tc.col, err)
}
if n != 1 {
t.Errorf("%s.%s missing after migration v5", tc.table, tc.col)
}
}
}

View File

@@ -14,9 +14,12 @@ import (
// Message is one row of the messages table.
type Message struct {
ID int64
GmailID string
ThreadID string
ID int64
GmailID string
ThreadID string
// MessageIDHeader is the RFC822 Message-ID header (Phase 5, FR-8). A reply
// uses it for In-Reply-To and References so Gmail keeps the thread.
MessageIDHeader string
FromAddr string
ToAddr string
Subject string
@@ -67,11 +70,12 @@ func ValidFolder(folder string) bool {
func (s *Store) UpsertMessage(ctx context.Context, m *Message) (int64, error) {
const q = `
INSERT INTO messages (
gmail_id, thread_id, from_addr, to_addr, subject, body_text,
gmail_id, thread_id, message_id_header, from_addr, to_addr, subject, body_text,
source_address, date_sent, received_at, imap_uid, imap_uidvalidity
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (gmail_id) DO UPDATE SET
thread_id = excluded.thread_id,
message_id_header = excluded.message_id_header,
from_addr = excluded.from_addr,
to_addr = excluded.to_addr,
subject = excluded.subject,
@@ -88,6 +92,7 @@ ON CONFLICT (gmail_id) DO UPDATE SET
_, err := s.db.ExecContext(ctx, q,
m.GmailID,
nullString(m.ThreadID),
nullString(m.MessageIDHeader),
nullString(m.FromAddr),
nullString(m.ToAddr),
nullString(m.Subject),
@@ -110,6 +115,35 @@ ON CONFLICT (gmail_id) DO UPDATE SET
return id, nil
}
// MessageByID loads one message including the reply fields (body, source
// address, and Message-ID header). It is the read path for a reply draft.
func (s *Store) MessageByID(ctx context.Context, id int64) (Message, error) {
const 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(m.message_id_header, '')
FROM messages m WHERE m.id = ?`
var (
m Message
dateSent string
received string
)
err := s.db.QueryRowContext(ctx, q, id).Scan(
&m.ID, &m.GmailID, &m.ThreadID, &m.FromAddr, &m.ToAddr, &m.Subject,
&m.BodyText, &m.SourceAddress, &dateSent, &received,
&m.IMAPUID, &m.IMAPUIDValidity, &m.Folder, &m.Mailbox, &m.MessageIDHeader,
)
if err != nil {
return Message{}, fmt.Errorf("load message %d: %w", id, err)
}
m.DateSent = parseSQLTime(dateSent)
m.ReceivedAt = parseSQLTime(received)
return m, nil
}
// ListMessages returns the most recent messages, newest first, joined to their
// latest classification when one exists.
func (s *Store) ListMessages(ctx context.Context, limit int) ([]MessageListItem, error) {

View File

@@ -18,6 +18,9 @@ CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
gmail_id TEXT NOT NULL UNIQUE,
thread_id TEXT,
-- Phase 5 (FR-8): the RFC822 Message-ID header. A reply sets In-Reply-To and
-- References to it so Gmail files the reply in the same thread.
message_id_header TEXT,
from_addr TEXT,
to_addr TEXT,
subject TEXT,
@@ -105,7 +108,9 @@ CREATE TABLE IF NOT EXISTS drafts (
message_id INTEGER NOT NULL REFERENCES messages (id) ON DELETE CASCADE,
body TEXT NOT NULL,
template_id TEXT,
-- Phase 5 (FR-8): draft -> sent. sent_at is set when the SMTP send succeeds.
status TEXT NOT NULL DEFAULT 'draft',
sent_at TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
);

View File

@@ -144,6 +144,21 @@ func (s *Store) applyMigrations(ctx context.Context) error {
return fmt.Errorf("record schema version 4: %w", err)
}
}
if version < 5 {
// Phase 5 (FR-8): store the RFC822 Message-ID so a reply can set
// In-Reply-To/References and stay in the same Gmail thread, and record
// when a draft was sent.
if err := s.ensureColumn(ctx, "messages", "message_id_header", "TEXT"); err != nil {
return err
}
if err := s.ensureColumn(ctx, "drafts", "sent_at", "TEXT"); err != nil {
return err
}
if _, err := s.db.ExecContext(ctx,
`INSERT OR IGNORE INTO schema_migrations (version) VALUES (5)`); err != nil {
return fmt.Errorf("record schema version 5: %w", err)
}
}
return nil
}

View File

@@ -238,6 +238,33 @@ type ActionLogEntry struct {
CreatedAt string
}
// MessageActions returns the action-log rows for one message, newest first.
// It backs the message detail pane (FR-13).
func (s *Store) MessageActions(ctx context.Context, messageID int64, limit int) ([]ActionLogEntry, error) {
if limit < 1 {
limit = 20
}
rows, err := s.db.QueryContext(ctx, `
SELECT id, message_id, action, COALESCE(actor, ''), COALESCE(detail, ''), created_at
FROM actions_log WHERE message_id = ? ORDER BY id DESC LIMIT ?`, messageID, limit)
if err != nil {
return nil, fmt.Errorf("message actions: %w", err)
}
defer rows.Close()
var out []ActionLogEntry
for rows.Next() {
var e ActionLogEntry
if err := rows.Scan(&e.ID, &e.MessageID, &e.Action, &e.Actor, &e.Detail, &e.CreatedAt); err != nil {
return nil, fmt.Errorf("scan message action: %w", err)
}
out = append(out, e)
}
if err := rows.Err(); err != nil {
return nil, fmt.Errorf("message actions: %w", err)
}
return out, nil
}
// RecentActions returns the newest action-log rows, newest first.
func (s *Store) RecentActions(ctx context.Context, limit int) ([]ActionLogEntry, error) {
if limit < 1 {