From 5ccca2fac90c4ec70c624a8748816452578f47f9 Mon Sep 17 00:00:00 2001 From: Andriy Oblivantsev Date: Fri, 14 Aug 2026 16:35:20 +0000 Subject: [PATCH 1/2] feat(mail): add Microsoft 365 Graph sync source with delta state Adds a GET-only Microsoft Graph source to bin/mail/sync (client-credentials, delta query). Wires it into the CLI (--source m365), the worker pool and the compose stack: - m365.go: M365Client (token cache, delta pagination, message normalization, attachment download) + m365Source adapter (per-mailbox folders + delta link) - sync.go: M365Credentials in SyncConfig; Committer interface so delta links only advance after a fully successful run (no skips on failure) - cli.go: --source m365 with M365_/MS_ env passthrough - Dockerfile: mail-build stage produces /mail-sync into the index image - docker-entrypoint: mail-sync loop (sync -> import -> index, default 10s) - compose.yaml: mail-sync service (index image, kb-var volume, secrets ro) --- .gitignore | 6 + Dockerfile | 9 + bin/docker-entrypoint | 20 ++ bin/mail/sync.go | 5 +- bin/mail/sync/cli.go | 25 ++- bin/mail/sync/m365.go | 382 +++++++++++++++++++++++++++++++++++++ bin/mail/sync/m365_test.go | 188 ++++++++++++++++++ bin/mail/sync/sync.go | 44 ++++- compose.yaml | 27 +++ 9 files changed, 700 insertions(+), 6 deletions(-) create mode 100644 bin/mail/sync/m365.go create mode 100644 bin/mail/sync/m365_test.go diff --git a/.gitignore b/.gitignore index 001fb10..5cc73a0 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,9 @@ go.work.local models/ # Purged from git history. Do not re-add. docs/crm-associations-proof.md + +# mount scaffold for the 8TB volume, never part of the repo +mnt/ + +# go build ./bin/mail/sync.go drops a binary named `sync` in cwd +/sync diff --git a/Dockerfile b/Dockerfile index 61366a6..20cc43b 100644 --- a/Dockerfile +++ b/Dockerfile @@ -28,6 +28,7 @@ RUN python -m pip install --no-cache-dir -r /tmp/requirements.lock.txt \ COPY . . RUN chmod +x /app/bin/docker-entrypoint \ && chown -R 2dph:2dph /app +COPY --from=mail-build /mail-sync /app/bin/mail-sync USER 2dph ENV PATH="/app/bin:${PATH}" \ @@ -37,6 +38,14 @@ HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ CMD python -c "import model2vec, ladybug, mistune; print('ok')" || exit 1 ENTRYPOINT ["/app/bin/docker-entrypoint"] +# --- mail-sync: standalone M365/OnlyOffice/Gmail puller (pure Go, no CGO) --- +FROM golang:1.26-bookworm AS mail-build +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY bin/mail ./bin/mail +RUN CGO_ENABLED=0 go build -o /mail-sync ./bin/mail/sync.go + # --- Go API: CGO with Zig, not gcc --- FROM golang:1.26-bookworm AS api-build WORKDIR /src diff --git a/bin/docker-entrypoint b/bin/docker-entrypoint index a7e3256..9cdacb4 100755 --- a/bin/docker-entrypoint +++ b/bin/docker-entrypoint @@ -5,6 +5,7 @@ # serve | search | watch # Index image (Python write path, compose profile `index`): # index | extract | audit | search (deprecated python wrapper) +# mail-sync [N] pull loop: sync -> import -> index every N s (default 10) # # Usage comment starts at line 2 (self-describing convention). set -euo pipefail @@ -34,5 +35,24 @@ case "$CMD" in serve) exec /app/bin/serve "$@" ;; extract) exec "$KB_PY" /app/bin/facts/extract "$@" ;; audit) exec "$KB_PY" /app/bin/facts/audit "$@" ;; + mail-sync) + # loop: pull mail (M365/OnlyOffice/Gmail) every N s, convert to md, + # rebuild the brain index. Index only when something new arrived. + interval="${1:-10}" + [ "$interval" -gt 0 ] 2>/dev/null || interval=10 + : "${MAIL_SYNC_ENV:=/secret/m365.env}" + : "${MAIL_SYNC_SRC:=m365}" + : "${MAIL_SYNC_OUT:=/app/var/mail}" + while true; do + out="$("/app/bin/mail-sync" --source "$MAIL_SYNC_SRC" --env "$MAIL_SYNC_ENV" --out "$MAIL_SYNC_OUT" 2>&1)" + echo "$out" + new="$(printf '%s\n' "$out" | sed -n 's/.*new=\([0-9]*\).*/\1/p' | tail -1)" + if [ -n "$new" ] && [ "$new" -gt 0 ] 2>/dev/null; then + "$KB_PY" /app/bin/mail/import --from-raw "$MAIL_SYNC_OUT" 2>&1 | tail -1 + "$KB_PY" /app/bin/kb/index --rebuild --with-mail 2>&1 | tail -1 + fi + sleep "$interval" + done + ;; *) echo "unknown command: $CMD" >&2; exit 2 ;; esac diff --git a/bin/mail/sync.go b/bin/mail/sync.go index f03d989..82aa64e 100755 --- a/bin/mail/sync.go +++ b/bin/mail/sync.go @@ -1,8 +1,9 @@ //usr/bin/env go run "$0" "$@"; exit -// bin/mail/sync.go - async download of OnlyOffice and Gmail mail to var/mail/. +// bin/mail/sync.go - async download of OnlyOffice, Gmail and M365 mail to var/mail/. // -// ./bin/mail/sync.go --source onlyoffice,gmail --limit 50 --workers 8 +// ./bin/mail/sync.go --source onlyoffice,gmail,m365 --limit 50 --workers 8 // ./bin/mail/sync.go --source gmail --force +// ./bin/mail/sync.go --source m365 --env .secrets/m365.env // ./bin/mail/sync.go --dry-run // // Writes raw message.json + attachments under var/mail///; run diff --git a/bin/mail/sync/cli.go b/bin/mail/sync/cli.go index 9ec41ec..eec710d 100644 --- a/bin/mail/sync/cli.go +++ b/bin/mail/sync/cli.go @@ -55,7 +55,7 @@ func bind(v *flagVals) *flaggy.Parser { p.Bool(&v.force, "", "force", "overwrite existing message.json") p.Bool(&v.dryRun, "", "dry-run", "list counts without writing") p.String(&v.query, "", "query", "Gmail search query") - p.String(&v.srcs, "", "source", "comma list: onlyoffice,gmail") + p.String(&v.srcs, "", "source", "comma list: onlyoffice,gmail,m365") return p } @@ -110,6 +110,24 @@ func ParseCLI(args []string) (CLIConfig, int, error) { CredentialsPath: filepath.Join(home, ".gmail-mcp", "credentials.json"), KeysPath: filepath.Join(home, ".gmail-mcp", "gcp-oauth.keys.json"), } + case "m365": + tenant := pick(envVars["M365_TENANT"], envVars["MS_TENANT"]) + cid := pick(envVars["M365_CLIENT_ID"], envVars["MS_CLIENT_ID"]) + sec := pick(envVars["M365_CLIENT_SECRET"], envVars["MS_CLIENT_SECRET"]) + users := pick(envVars["M365_USERS"], envVars["MS_USERS"]) + if tenant == "" || cid == "" || sec == "" || users == "" { + return CLIConfig{}, 2, fmt.Errorf("m365 source needs M365_TENANT/CLIENT_ID/CLIENT_SECRET/USERS in %s", v.env) + } + var userList []string + for _, u := range strings.Split(users, ",") { + if u = strings.TrimSpace(u); u != "" { + userList = append(userList, u) + } + } + if len(userList) == 0 { + return CLIConfig{}, 2, fmt.Errorf("m365 source: M365_USERS empty") + } + cfg.M365 = &M365Credentials{Tenant: tenant, ClientID: cid, ClientSecret: sec, Users: userList} default: return CLIConfig{}, 2, fmt.Errorf("unknown source %q", s) } @@ -126,7 +144,7 @@ func Main(args []string) int { return code } if cfg.Help { - fmt.Fprintln(os.Stderr, "usage: bin/mail/sync.go [--source onlyoffice,gmail] [--query GMAIL_Q] [--limit N] [--offset N] [--workers N] [--force] [--dry-run]") + fmt.Fprintln(os.Stderr, "usage: bin/mail/sync.go [--source onlyoffice,gmail,m365] [--query GMAIL_Q] [--limit N] [--offset N] [--workers N] [--force] [--dry-run]") return 0 } ctx, cancel := context.WithTimeout(context.Background(), 6*time.Hour) @@ -169,7 +187,8 @@ func readEnv(path string) map[string]string { if !ok { continue } - if strings.HasPrefix(k, "ONLYOFFICE_") || strings.HasPrefix(k, "OO_") { + if strings.HasPrefix(k, "ONLYOFFICE_") || strings.HasPrefix(k, "OO_") || + strings.HasPrefix(k, "M365_") || strings.HasPrefix(k, "MS_") { out[k] = v } } diff --git a/bin/mail/sync/m365.go b/bin/mail/sync/m365.go new file mode 100644 index 0000000..f3add04 --- /dev/null +++ b/bin/mail/sync/m365.go @@ -0,0 +1,382 @@ +package sync + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "time" +) + +// M365Credentials holds a Microsoft Graph app registration with the Mail.Read +// application permission. Client credentials are read from env/.env, never +// committed. +type M365Credentials struct { + Tenant string + ClientID string + ClientSecret string + Users []string // mailbox addresses to sync, e.g. info@example.com +} + +// m365Token is the cached access token with expiry. +type m365Token struct { + AccessToken string + Expiry time.Time +} + +// M365Client talks to the Microsoft Graph API using the client-credentials +// flow (app registration with Mail.Read application permission). GET-only: +// messages are never marked as read or deleted. +type M365Client struct { + creds M365Credentials + base string // graph base URL; default https://graph.microsoft.com + tokenEndpoint string // login endpoint; default https://login.microsoftonline.com + client *http.Client + mu chan struct{} + token *m365Token +} + +func NewM365Client(creds M365Credentials) (*M365Client, error) { + if creds.Tenant == "" || creds.ClientID == "" || creds.ClientSecret == "" { + return nil, errors.New("m365 needs tenant, client id and client secret") + } + c := &M365Client{ + creds: creds, + base: "https://graph.microsoft.com", + client: &http.Client{Timeout: 90 * time.Second}, + mu: make(chan struct{}, 1), + } + c.mu <- struct{}{} + return c, nil +} + +// accessToken returns a fresh bearer token, refreshing via the Azure AD token +// endpoint when the cached one is missing or about to expire (within 2 min). +func (c *M365Client) accessToken(ctx context.Context) (string, error) { + select { + case <-c.mu: + case <-ctx.Done(): + return "", ctx.Err() + } + defer func() { c.mu <- struct{}{} }() + if c.token != nil && c.token.AccessToken != "" && time.Now().Before(c.token.Expiry.Add(-2*time.Minute)) { + return c.token.AccessToken, nil + } + return c.refreshLocked(ctx) +} + +func (c *M365Client) refreshLocked(ctx context.Context) (string, error) { + form := url.Values{} + form.Set("grant_type", "client_credentials") + form.Set("client_id", c.creds.ClientID) + form.Set("client_secret", c.creds.ClientSecret) + form.Set("scope", "https://graph.microsoft.com/.default") + + endpoint := c.tokenEndpoint + if endpoint == "" { + endpoint = fmt.Sprintf("https://login.microsoftonline.com/%s/oauth2/v2.0/token", c.creds.Tenant) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, strings.NewReader(form.Encode())) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + resp, err := c.client.Do(req) + if err != nil { + return "", fmt.Errorf("m365 token: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) + if resp.StatusCode != http.StatusOK { + var e struct { + Error string `json:"error"` + Desc string `json:"error_description"` + } + _ = json.Unmarshal(body, &e) + return "", fmt.Errorf("m365 token status %d: %s (%s)", resp.StatusCode, e.Error, truncate(e.Desc, 200)) + } + var out struct { + AccessToken string `json:"access_token"` + ExpiresIn int64 `json:"expires_in"` + } + if err := json.Unmarshal(body, &out); err != nil { + return "", fmt.Errorf("m365 token parse: %w", err) + } + c.token = &m365Token{ + AccessToken: out.AccessToken, + Expiry: time.Now().Add(time.Duration(out.ExpiresIn) * time.Second), + } + return out.AccessToken, nil +} + +// deltaPage is one response page of the Graph delta query. +type deltaPage struct { + Value []struct { + ID string `json:"id"` + RemovedReason string `json:"@odata.removedReason"` + } `json:"value"` + NextLink string `json:"@odata.nextLink"` + DeltaLink string `json:"@odata.deltaLink"` +} + +// ListDeltaIDs walks the inbox delta query and returns live message ids since +// the previous deltaLink (or the full inbox when deltaLink is empty). Returns +// the new deltaLink for the next run. GET-only; nothing is mutated server-side. +func (c *M365Client) ListDeltaIDs(ctx context.Context, mailbox, deltaLink string, limit int) ([]string, string, error) { + var ( + ids []string + url string + link = deltaLink + ) + if link == "" { + url = fmt.Sprintf("/v1.0/users/%s/mailFolders/inbox/messages/delta", pathEscape(mailbox)) + } else { + url = link + } + for url != "" { + var page deltaPage + if err := c.getJSON(ctx, url, &page); err != nil { + return ids, link, err + } + for _, m := range page.Value { + if m.RemovedReason != "" { + continue + } + if m.ID == "" { + continue + } + ids = append(ids, m.ID) + if limit > 0 && len(ids) >= limit { + if page.DeltaLink != "" { + link = page.DeltaLink + } + return ids, link, nil + } + } + if page.DeltaLink != "" { + link = page.DeltaLink + url = "" + break + } + url = page.NextLink + } + return ids, link, nil +} + +// GetMessage fetches a single message by id and normalizes it to the Message +// contract. GET-only. +func (c *M365Client) GetMessage(ctx context.Context, mailbox, id string) (*Message, error) { + path := fmt.Sprintf("/v1.0/users/%s/messages/%s?$expand=attachments($select=id,name,contentType,size,isInline)", + pathEscape(mailbox), pathEscape(id)) + var raw struct { + ID string `json:"id"` + Subject string `json:"subject"` + From m365Recipient `json:"from"` + ToRecipients []m365Recipient `json:"toRecipients"` + CCRecipients []m365Recipient `json:"ccRecipients"` + BCCRecipients []m365Recipient `json:"bccRecipients"` + ReceivedDateTime string `json:"receivedDateTime"` + Body m365Body `json:"body"` + BodyPreview string `json:"bodyPreview"` + InternetMessageID string `json:"internetMessageId"` + Attachments []m365Attachment `json:"attachments"` + } + if err := c.getJSON(ctx, path, &raw); err != nil { + return nil, err + } + m := &Message{ + Source: "m365", + ID: raw.ID, + Folder: "m365", + Subject: raw.Subject, + From: formatRecipient(raw.From), + To: formatRecipients(raw.ToRecipients), + CC: formatRecipients(raw.CCRecipients), + BCC: formatRecipients(raw.BCCRecipients), + MimeMessageID: raw.InternetMessageID, + } + if t, err := time.Parse(time.RFC3339, raw.ReceivedDateTime); err == nil { + m.ReceivedAt = t + } + switch strings.ToLower(raw.Body.ContentType) { + case "html": + m.HTMLBody = raw.Body.Content + if raw.BodyPreview != "" { + m.TextBody = raw.BodyPreview + } + default: + m.TextBody = raw.Body.Content + if raw.BodyPreview != "" { + m.HTMLBody = raw.BodyPreview + } + } + for _, a := range raw.Attachments { + if a.IsInline || a.ID == "" || a.Name == "" { + continue + } + m.Attachments = append(m.Attachments, Attachment{ + FileID: a.ID, + FileName: a.Name, + StoredName: a.Name, + Size: a.Size, + ContentType: a.ContentType, + }) + } + m.HasAttachments = len(m.Attachments) > 0 + return m, nil +} + +type m365Recipient struct { + EmailAddress struct { + Name string `json:"name"` + Address string `json:"address"` + } `json:"emailAddress"` +} + +type m365Body struct { + ContentType string `json:"contentType"` + Content string `json:"content"` +} + +type m365Attachment struct { + ID string `json:"id"` + Name string `json:"name"` + ContentType string `json:"contentType"` + Size int64 `json:"size"` + IsInline bool `json:"isInline"` + ContentID string `json:"contentId"` +} + +func formatRecipients(rs []m365Recipient) string { + var parts []string + for _, r := range rs { + if s := formatRecipient(r); s != "" { + parts = append(parts, s) + } + } + return strings.Join(parts, ", ") +} + +func formatRecipient(r m365Recipient) string { a := r.EmailAddress.Address + n := r.EmailAddress.Name + switch { + case n == "" || n == a: + return a + case a == "": + return n + default: + return fmt.Sprintf("%s <%s>", n, a) + } +} + +// DownloadAttachment fetches an attachment's raw bytes via the /$value stream. +func (c *M365Client) DownloadAttachment(ctx context.Context, mailbox, msgID, attID string) ([]byte, error) { + path := fmt.Sprintf("/v1.0/users/%s/messages/%s/attachments/%s/$value", + pathEscape(mailbox), pathEscape(msgID), pathEscape(attID)) + tok, err := c.accessToken(ctx) + if err != nil { + return nil, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.base+path, nil) + if err != nil { + return nil, err + } + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := c.client.Do(req) + if err != nil { + return nil, fmt.Errorf("m365 attachment: %w", err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 256<<20)) + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("m365 attachment %s: status %d: %s", attID, resp.StatusCode, truncate(string(body), 300)) + } + return body, nil +} + +func (c *M365Client) getJSON(ctx context.Context, path string, out any) error { + tok, err := c.accessToken(ctx) + if err != nil { + return err + } + u := path + if !strings.HasPrefix(u, "http") { + u = c.base + u + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+tok) + resp, err := c.client.Do(req) + if err != nil { + return fmt.Errorf("m365 %s: %w", path, err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(io.LimitReader(resp.Body, 16<<20)) + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("m365 %s: status %d: %s", path, resp.StatusCode, truncate(string(body), 300)) + } + if out != nil { + return json.Unmarshal(body, out) + } + return nil +} + +// m365Source adapts a mailbox to the Source worker-pool contract. Each mailbox +// gets its own folder under var/mail/m365// and a delta state file. +type m365Source struct { + c *M365Client + mailbox string + localpart string + stateDir string + pending string // delta link to persist on Commit() + hasPending bool +} + +func (s *m365Source) Folder() string { return filepath.Join("m365", s.localpart) } + +func (s *m365Source) ListIDs(ctx context.Context, limit int, cursor string) ([]string, string, error) { + link, _ := os.ReadFile(filepath.Join(s.stateDir, s.localpart+".deltalink")) + ids, newLink, err := s.c.ListDeltaIDs(ctx, s.mailbox, strings.TrimSpace(string(link)), limit) + if err != nil { + return nil, "", err + } + // Buffer the new delta link; persist it only in Commit() after the full + // batch downloaded, so a failed run stays retryable without gaps. + if newLink != "" { + s.pending = newLink + s.hasPending = true + } + return ids, "", nil +} + +// Commit persists the buffered delta link. Called by the sync runner only when +// every listed message downloaded successfully. +func (s *m365Source) Commit() error { + if !s.hasPending || s.pending == "" { + return nil + } + if err := os.MkdirAll(s.stateDir, 0o755); err != nil { + return err + } + return os.WriteFile(filepath.Join(s.stateDir, s.localpart+".deltalink"), []byte(s.pending), 0o644) +} + +func (s *m365Source) Get(ctx context.Context, id string) (*Message, error) { + return s.c.GetMessage(ctx, s.mailbox, id) +} + +func (s *m365Source) DownloadAttachment(ctx context.Context, msg *Message, att Attachment) ([]byte, error) { + return s.c.DownloadAttachment(ctx, s.mailbox, msg.ID, att.FileID) +} + +func pathEscape(s string) string { + return url.PathEscape(s) +} diff --git a/bin/mail/sync/m365_test.go b/bin/mail/sync/m365_test.go new file mode 100644 index 0000000..9e52686 --- /dev/null +++ b/bin/mail/sync/m365_test.go @@ -0,0 +1,188 @@ +package sync + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "testing" +) + +// newM365TestClient serves the Graph delta + message endpoints against a fake +// token endpoint, so unit tests never touch the network. +func newM365TestClient(t *testing.T, graph http.Handler) *M365Client { + t.Helper() + tok := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"access_token":"test-token","expires_in":3600}`)) + })) + gr := httptest.NewServer(graph) + t.Cleanup(func() { + tok.Close() + gr.Close() + }) + client, err := NewM365Client(M365Credentials{Tenant: "t.onmicrosoft.com", ClientID: "c", ClientSecret: "s"}) + if err != nil { + t.Fatal(err) + } + client.base = gr.URL + client.tokenEndpoint = tok.URL + return client +} + +func TestM365AccessToken(t *testing.T) { + c := newM365TestClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + got, err := c.accessToken(context.Background()) + if err != nil { + t.Fatalf("accessToken: %v", err) + } + if got != "test-token" { + t.Errorf("token = %q", got) + } + // Second call must reuse the cached token (no token request). + again, err := c.accessToken(context.Background()) + if err != nil || again != "test-token" { + t.Fatalf("cached token: %q, %v", again, err) + } +} + +func TestM365DeltaSkipsTombstones(t *testing.T) { + first := true + graph := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if first { + first = false + w.Write([]byte(`{"value":[ + {"id":"m1"}, + {"id":"m2","@odata.removedReason":"deleted"}, + {"id":"m3"} + ],"@odata.deltaLink":"` + deltaNext + `"}`)) + return + } + // Second call must use the stored deltaLink (points at this server). + if r.URL.Path != "/v1.0/delta-next" { + w.WriteHeader(500) + w.Write([]byte(`{"error":{"message":"unexpected path"}}`)) + return + } + w.Write([]byte(`{"value":[{"id":"m4"}],"@odata.deltaLink":"` + deltaFinal + `"}`)) + }) + c := newM365TestClient(t, graph) + deltaNext = c.base + "/v1.0/delta-next" + deltaFinal = c.base + "/v1.0/delta-final" + + ids, link, err := c.ListDeltaIDs(context.Background(), "a@x.de", "", 0) + if err != nil { + t.Fatalf("delta: %v", err) + } + if len(ids) != 2 || ids[0] != "m1" || ids[1] != "m3" { + t.Errorf("ids = %v", ids) + } + if link == "" { + t.Error("expected new deltaLink") + } + // Incremental: pass the deltaLink, get only the new id. + ids2, link2, err := c.ListDeltaIDs(context.Background(), "a@x.de", link, 0) + if err != nil { + t.Fatalf("delta incremental: %v", err) + } + if len(ids2) != 1 || ids2[0] != "m4" { + t.Errorf("ids2 = %v", ids2) + } + if link2 == "" { + t.Error("expected updated deltaLink") + } +} + +func TestM365GetMessageNormalizes(t *testing.T) { + graph := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if pathLast(r.URL.Path) == "messages" { + w.Write([]byte(`{"value":[{"id":"m1"}]}`)) + return + } + w.Write([]byte(`{ + "id":"m1", + "subject":"Hallo", + "from":{"emailAddress":{"name":"Max","address":"max@x.de"}}, + "toRecipients":[{"emailAddress":{"address":"a@x.de"}}], + "receivedDateTime":"2026-08-14T08:15:00Z", + "body":{"contentType":"html","content":"

body

"}, + "bodyPreview":"body", + "internetMessageId":"", + "attachments":[ + {"id":"att1","name":"doc.pdf","contentType":"application/pdf","size":10}, + {"id":"img1","name":"logo.png","contentType":"image/png","isInline":true} + ] + }`)) + }) + c := newM365TestClient(t, graph) + m, err := c.GetMessage(context.Background(), "a@x.de", "m1") + if err != nil { + t.Fatalf("GetMessage: %v", err) + } + if m.ID != "m1" || m.Subject != "Hallo" || m.From != "Max " || m.To != "a@x.de" { + t.Errorf("headers mismatch: %+v", m) + } + if m.HTMLBody != "

body

" { + t.Errorf("html = %q", m.HTMLBody) + } + if m.ReceivedAt.IsZero() { + t.Error("receivedAt zero") + } + if m.MimeMessageID != "" { + t.Errorf("mime id = %q", m.MimeMessageID) + } + if len(m.Attachments) != 1 || m.Attachments[0].FileName != "doc.pdf" || m.Attachments[0].FileID != "att1" { + t.Errorf("atts = %+v", m.Attachments) + } + if !m.HasAttachments { + t.Error("expected hasAttachments") + } +} + +func TestM365SourceDeltaState(t *testing.T) { + graph := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"value":[{"id":"m1"}],"@odata.deltaLink":"` + deltaNext + `"}`)) + }) + c := newM365TestClient(t, graph) + deltaNext = c.base + "/v1.0/delta-next" + stateDir := filepath.Join(t.TempDir(), ".m365") + s := &m365Source{c: c, mailbox: "info@x.de", localpart: "info", stateDir: stateDir} + + if s.Folder() != "m365/info" { + t.Errorf("folder = %q", s.Folder()) + } + ids, _, err := s.ListIDs(context.Background(), 0, "") + if err != nil { + t.Fatalf("ListIDs: %v", err) + } + if len(ids) != 1 || ids[0] != "m1" { + t.Errorf("ids = %v", ids) + } + if err := s.Commit(); err != nil { + t.Fatalf("Commit: %v", err) + } + data, err := os.ReadFile(filepath.Join(stateDir, "info.deltalink")) + if err != nil { + t.Fatalf("read delta state: %v", err) + } + if string(data) != deltaNext { + t.Errorf("delta state = %q, want %q", string(data), deltaNext) + } +} + +func pathLast(p string) string { + for i := len(p) - 1; i >= 0; i-- { + if p[i] == '/' { + return p[i+1:] + } + } + return p +} + +// deltaNext/deltaFinal are set per-test from the fake graph server URL so +// deltaLink values always point back at the fake (never the real Graph). +var deltaNext, deltaFinal string diff --git a/bin/mail/sync/sync.go b/bin/mail/sync/sync.go index 2311278..13c11f8 100644 --- a/bin/mail/sync/sync.go +++ b/bin/mail/sync/sync.go @@ -106,6 +106,7 @@ func Retry(ctx context.Context, policy RetryPolicy, fn func() error) error { type SyncConfig struct { OO *OOConfig // OnlyOffice source (optional) Gmail *GmailCredentials // Gmail source (optional) + M365 *M365Credentials // Microsoft 365 Graph source (optional) Out string // var/mail root; default /var/mail Workers int // concurrency; default 4 Limit int // max messages per source (0 = all) @@ -133,6 +134,14 @@ type Source interface { Folder() string } +// Committer is an optional Source capability: Commit is called after all listed +// ids have been downloaded successfully. Sources that only advance durable state +// on success (e.g. a Graph delta link) implement this so a killed or failed run +// stays retryable without gaps. +type Committer interface { + Commit() error +} + type ooSource struct { c *OOClient page int @@ -222,8 +231,22 @@ func Run(ctx context.Context, cfg SyncConfig) (*SyncStats, error) { } sources = append(sources, &gmailSource{c: gm, query: cfg.Query}) } + if cfg.M365 != nil { + stateDir := filepath.Join(cfg.Out, ".m365") + for _, mb := range cfg.M365.Users { + if !strings.Contains(mb, "@") { + return nil, fmt.Errorf("m365 user %q is not an email address", mb) + } + c, err := NewM365Client(*cfg.M365) + if err != nil { + return nil, fmt.Errorf("m365 init for %s: %w", mb, err) + } + local := strings.SplitN(mb, "@", 2)[0] + sources = append(sources, &m365Source{c: c, mailbox: mb, localpart: strings.ToLower(local), stateDir: stateDir}) + } + } if len(sources) == 0 { - return nil, errors.New("sync: no source configured (need OO, Gmail, or both)") + return nil, errors.New("sync: no source configured (need OO, Gmail, M365, or a combination)") } stats := &SyncStats{} @@ -296,6 +319,25 @@ func Run(ctx context.Context, cfg SyncConfig) (*SyncStats, error) { close(jobsCh) wg.Wait() + // Only advance durable source state (e.g. delta links) when everything + // downloaded. A killed or failed run must be retryable without gaps. + if len(failures) == 0 { + seen := map[Source]bool{} + for _, j := range jobs { + if seen[j.src] { + continue + } + seen[j.src] = true + if c, ok := j.src.(Committer); ok { + if err := c.Commit(); err != nil { + mu.Lock() + failures = append(failures, j.src.Folder()+"/commit: "+err.Error()) + mu.Unlock() + } + } + } + } + if len(failures) > 0 { fmt.Fprintf(os.Stderr, "sync: %d failures:\n %s\n", len(failures), strings.Join(failures, "\n ")) } diff --git a/compose.yaml b/compose.yaml index 19475cc..434376f 100644 --- a/compose.yaml +++ b/compose.yaml @@ -92,6 +92,33 @@ services: tmpfs: - /tmp + # mail-sync: pull M365 (and OnlyOffice/Gmail) mail every 10 s into the shared + # var volume, then rebuild the brain index. Runs on the index image so it can + # convert + index in-process. Needs ~/.config/brain/m365.env for the M365 + # source; override MAIL_SYNC_SRC / MAIL_SYNC_ENV for other providers. + # docker compose up -d mail-sync + mail-sync: + image: ghcr.io/eslider/2dph:index + build: + context: . + dockerfile: Dockerfile + target: index + environment: + HF_HOME: /data/hf + KB_PY: python3 + MAIL_SYNC_SRC: m365 + MAIL_SYNC_ENV: /secret/m365.env + volumes: + - kb-model:/data/hf + - kb-var:/app/var + - ~/.config/brain:/secret:ro + command: ["mail-sync", "10"] + read_only: true + tmpfs: + - /tmp + restart: unless-stopped + stop_grace_period: 20s + # Optional local SearXNG (D3). Skip if BRAIN_SEARCH_URL already points at a # live instance — do not run a second copy on that host. # SEARXNG_SECRET=$(openssl rand -hex 32) docker compose --profile searxng up -d -- 2.54.0 From b612d3cc97a8a2e24bf163a73dd60ab4310ead13 Mon Sep 17 00:00:00 2001 From: Andriy Oblivantsev Date: Fri, 14 Aug 2026 17:59:02 +0100 Subject: [PATCH 2/2] feat(mail): adapt M365 ETL loop for OO+Gmail bots. Default compose mail-sync to onlyoffice,gmail every 300s with import on new mail; gate full --rebuild behind MAIL_SYNC_INDEX. Fix mail-build stage order and add bin/stack/start-mail-sync. --- AGENTS.md | 10 +++++++++- Dockerfile | 17 +++++++++-------- PLAN.md | 13 +++++++++---- README.md | 2 ++ bin/docker-entrypoint | 21 ++++++++++++--------- bin/stack/lib.sh | 19 ++++++++++++++++--- bin/stack/start-mail-sync | 21 +++++++++++++++++++++ bin/stack/stop | 2 +- bin/tools/test_stack.py | 21 ++++++++++++++++++--- compose.yaml | 19 ++++++++++++------- docs/runbook.md | 4 +++- 11 files changed, 112 insertions(+), 37 deletions(-) create mode 100755 bin/stack/start-mail-sync diff --git a/AGENTS.md b/AGENTS.md index 60de48d..d74b014 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -66,12 +66,20 @@ var/ kb.lbug, var/mail/*, caches (gitignored) ```bash bin/mail/sync.go --source onlyoffice,gmail --workers 8 --out var/mail # raw message.json + attachments bin/mail/sync.go --source gmail --query 'from:example.com' --out var/mail # Gmail search (default in:inbox) +bin/mail/sync.go --source m365 --env ~/.config/brain/mail.env --out var/mail # Microsoft Graph (delta) bin/mail/import.go --from-raw var/mail # message.json → message.md (convert only) bin/brain/index.go --rebuild --with-facts --with-chats +bin/stack/start-mail-sync # compose ETL: sync→import every 300s ``` - `sync` (Go) downloads messages + attachments; Gmail uses paginated list + - `body.attachmentId` (not partId) for attachments. + `body.attachmentId` (not partId) for attachments. Sources: `onlyoffice`, + `gmail`, `m365` (client-credentials + delta link; commit after success). +- Compose `mail-sync` / `bin/stack/start-mail-sync`: ETL loop (default + `onlyoffice,gmail`, 300s). On `new>0` runs import; full `--rebuild` only if + `MAIL_SYNC_INDEX=1`. Secrets: `~/.config/brain/mail.env` + `~/.gmail-mcp`. + Case wrappers (e.g. family `gmail-sync-la-quinta.sh`) and ai-bot + `gmail-reauth.sh` reuse this sync/OAuth — do not fork corpus download. - `import` converts body + attachments to markdown. PDFs use poppler `pdftotext -layout` fast path (~15ms); textless/scanned PDFs use `pdftoppm` + tesseract `eng+deu` (`bin/mail/ocr.go`). Optional diff --git a/Dockerfile b/Dockerfile index 20cc43b..ad021b9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,6 +6,15 @@ # API: Go + ladybug via Zig CGO (no CPython). # Index: Python write path (profile `index` until brain/add is v2). +# --- mail-sync: standalone M365/OnlyOffice/Gmail puller (pure Go, no CGO) --- +FROM golang:1.26-bookworm AS mail-build +WORKDIR /src +COPY go.mod go.sum ./ +RUN go mod download +COPY bin/mail ./bin/mail +COPY internal ./internal +RUN CGO_ENABLED=0 go build -o /mail-sync ./bin/mail/sync.go + # --- Python sidecar (Ladybug write / rebuild) --- FROM python:3.12-slim AS index @@ -38,14 +47,6 @@ HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \ CMD python -c "import model2vec, ladybug, mistune; print('ok')" || exit 1 ENTRYPOINT ["/app/bin/docker-entrypoint"] -# --- mail-sync: standalone M365/OnlyOffice/Gmail puller (pure Go, no CGO) --- -FROM golang:1.26-bookworm AS mail-build -WORKDIR /src -COPY go.mod go.sum ./ -RUN go mod download -COPY bin/mail ./bin/mail -RUN CGO_ENABLED=0 go build -o /mail-sync ./bin/mail/sync.go - # --- Go API: CGO with Zig, not gcc --- FROM golang:1.26-bookworm AS api-build WORKDIR /src diff --git a/PLAN.md b/PLAN.md index c98a7f0..06f5153 100644 --- a/PLAN.md +++ b/PLAN.md @@ -86,7 +86,7 @@ detective method: **a fact needs ≥2 independent sources or it is mail/ocr.go tesseract eng+deu (pdftoppm scans) md/import (deprecated; bin/markdown/import.go) brain/extract brain/audit brain/deduce (thinking wrapper) - stack/start start-assistant stop status + stack/start start-assistant start-mail-sync stop status web/search (deprecated shim → web/search.go) db/psql-yq (vendored) ssh-tunnel onlyoffice pg tunnel 5433 @@ -141,8 +141,9 @@ Common props on every node/edge: `root`, `confidence`, `evidence[]`, `how`, ## Mail pipeline (done) -1. `bin/mail/sync.go` (Go, 8 workers) — paginated Gmail/OnlyOffice download. - Gmail attachments key off `body.attachmentId`, not MIME `partId`. +1. `bin/mail/sync.go` (Go, 8 workers) — paginated Gmail / OnlyOffice / M365 + Graph download. Gmail attachments key off `body.attachmentId`, not MIME + `partId`. M365 uses client-credentials + delta link (commit after success). 2. `bin/mail/import.go --from-raw` — message.json → message.md; PDFs via `pdftotext -layout` (~15ms); textless/scanned PDFs `pdftoppm` + tesseract `eng+deu`. ICS sidecars @@ -151,7 +152,11 @@ Common props on every node/edge: `root`, `confidence`, `evidence[]`, `how`, corrupts its WAL on bulk-insert into an already-indexed DB. Conversion and indexing stay separate for crash safety. `bin/mail/index_mail` is a deprecation shim. -4. Result: 17,835 messages → 28,918 info leafs, FTS + HNSW healthy, searchable +4. Compose `mail-sync` / `bin/stack/start-mail-sync` — ETL loop (default + `onlyoffice,gmail`, 300s): sync → import on `new>0`; full rebuild only if + `MAIL_SYNC_INDEX=1`. Bot digests (ai-bot) and case wrappers reuse sync/OAuth; + they do not replace the corpus path. +5. Result: 17,835 messages → 28,918 info leafs, FTS + HNSW healthy, searchable via `bin/brain/search.go`. ## CI/CD pipeline (D15) diff --git a/README.md b/README.md index c7b55b9..b95d146 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,8 @@ Mail is a first-class corpus (retrievable through the same search): ```bash bin/mail/sync.go --source onlyoffice,gmail --workers 8 --out var/mail # raw sync (Go) +bin/mail/sync.go --source m365 --env ~/.config/brain/mail.env # Microsoft 365 Graph +bin/stack/start-mail-sync # compose ETL (300s; no auto-rebuild) bin/mail/import.go --from-raw var/mail # JSON → markdown bin/brain/add.go --text T --root facts --source "a.md x b.md" bin/brain/index.go --rebuild --with-facts --with-chats # facts extract + chats md diff --git a/bin/docker-entrypoint b/bin/docker-entrypoint index 9cdacb4..8041408 100755 --- a/bin/docker-entrypoint +++ b/bin/docker-entrypoint @@ -5,7 +5,7 @@ # serve | search | watch # Index image (Python write path, compose profile `index`): # index | extract | audit | search (deprecated python wrapper) -# mail-sync [N] pull loop: sync -> import -> index every N s (default 10) +# mail-sync [N] ETL loop: sync -> import; optional index (default 300s) # # Usage comment starts at line 2 (self-describing convention). set -euo pipefail @@ -36,20 +36,23 @@ case "$CMD" in extract) exec "$KB_PY" /app/bin/facts/extract "$@" ;; audit) exec "$KB_PY" /app/bin/facts/audit "$@" ;; mail-sync) - # loop: pull mail (M365/OnlyOffice/Gmail) every N s, convert to md, - # rebuild the brain index. Index only when something new arrived. - interval="${1:-10}" - [ "$interval" -gt 0 ] 2>/dev/null || interval=10 - : "${MAIL_SYNC_ENV:=/secret/m365.env}" - : "${MAIL_SYNC_SRC:=m365}" + # ETL: pull mail, convert to md when new>0. Full --rebuild is opt-in + # (MAIL_SYNC_INDEX=1) — ~29k leaf rebuild is minutes, not a 10s loop. + interval="${1:-300}" + [ "$interval" -gt 0 ] 2>/dev/null || interval=300 + : "${MAIL_SYNC_ENV:=/secret/mail.env}" + : "${MAIL_SYNC_SRC:=onlyoffice,gmail}" : "${MAIL_SYNC_OUT:=/app/var/mail}" + : "${MAIL_SYNC_INDEX:=0}" while true; do - out="$("/app/bin/mail-sync" --source "$MAIL_SYNC_SRC" --env "$MAIL_SYNC_ENV" --out "$MAIL_SYNC_OUT" 2>&1)" + out="$("/app/bin/mail-sync" --source "$MAIL_SYNC_SRC" --env "$MAIL_SYNC_ENV" --out "$MAIL_SYNC_OUT" 2>&1)" || true echo "$out" new="$(printf '%s\n' "$out" | sed -n 's/.*new=\([0-9]*\).*/\1/p' | tail -1)" if [ -n "$new" ] && [ "$new" -gt 0 ] 2>/dev/null; then "$KB_PY" /app/bin/mail/import --from-raw "$MAIL_SYNC_OUT" 2>&1 | tail -1 - "$KB_PY" /app/bin/kb/index --rebuild --with-mail 2>&1 | tail -1 + if [ "$MAIL_SYNC_INDEX" = "1" ]; then + "$KB_PY" /app/bin/kb/index --rebuild --with-mail 2>&1 | tail -1 + fi fi sleep "$interval" done diff --git a/bin/stack/lib.sh b/bin/stack/lib.sh index 8db3a52..0a761f2 100644 --- a/bin/stack/lib.sh +++ b/bin/stack/lib.sh @@ -143,12 +143,17 @@ ensure_picoclaw() { wait_health "$PICOCLAW_URL/health" || stack_die "picoclaw health failed at $PICOCLAW_URL/health" } +mail_sync_running() { + compose ps --status running --services 2>/dev/null | grep -qx mail-sync +} + stack_status() { - local bh=down mcp=down ph=down present=false + local bh=down mcp=down ph=down present=false ms=down health_ok "$BRAIN_URL/health" && bh=ok mcp_ok && mcp=ok reasoner_has_model && present=true health_ok "$PICOCLAW_URL/health" && ph=ok + mail_sync_running && ms=ok cat <&2 + compose up -d mail-sync +} + stack_attach_agent() { local opts=() if [[ -t 0 && -t 1 ]]; then @@ -230,6 +243,6 @@ stack_stop() { return 0 ;; esac - echo "stack: stop brain brain-mcp reasoner picoclaw (volumes kept)" >&2 - compose --profile picoclaw --profile reasoner stop picoclaw brain-mcp reasoner brain + echo "stack: stop brain brain-mcp reasoner picoclaw mail-sync (volumes kept)" >&2 + compose --profile picoclaw --profile reasoner stop picoclaw brain-mcp reasoner brain mail-sync } diff --git a/bin/stack/start-mail-sync b/bin/stack/start-mail-sync new file mode 100755 index 0000000..d39e34a --- /dev/null +++ b/bin/stack/start-mail-sync @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# bin/stack/start-mail-sync - compose up mail-sync ETL (sync → import; optional index). +# +# bin/stack/start-mail-sync +# +# Default: onlyoffice,gmail every 300s into kb-var. Full --rebuild only if +# MAIL_SYNC_INDEX=1 in compose/env. Secrets: ~/.config/brain/mail.env + +# ~/.gmail-mcp (mounted). Does not start brain/picoclaw. +set -euo pipefail + +STACK_DIR="$(CDPATH= cd -- "$(dirname "$0")" && pwd)" +# shellcheck source=lib.sh +source "$STACK_DIR/lib.sh" +case "${1:-}" in +-h | --help) + stack_usage "$0" + exit 0 + ;; +esac +stack_start_mail_sync "$@" +stack_status diff --git a/bin/stack/stop b/bin/stack/stop index 880c419..70fc77f 100755 --- a/bin/stack/stop +++ b/bin/stack/stop @@ -1,5 +1,5 @@ #!/usr/bin/env bash -# bin/stack/stop - stop compose brain / brain-mcp / reasoner / picoclaw. +# bin/stack/stop - stop compose brain / brain-mcp / reasoner / picoclaw / mail-sync. # # bin/stack/stop # diff --git a/bin/tools/test_stack.py b/bin/tools/test_stack.py index 6ad25ba..f84b0eb 100644 --- a/bin/tools/test_stack.py +++ b/bin/tools/test_stack.py @@ -9,7 +9,7 @@ import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] -METHODS = ("start", "start-assistant", "stop", "status") +METHODS = ("start", "start-assistant", "start-mail-sync", "stop", "status") class StackLayoutTest(unittest.TestCase): @@ -32,8 +32,10 @@ class StackLayoutTest(unittest.TestCase): lib = (ROOT / "bin" / "stack" / "lib.sh").read_text() self.assertIn("stack_start", lib) self.assertIn("stack_start_assistant", lib) + self.assertIn("stack_start_mail_sync", lib) self.assertIn("stack_stop", lib) self.assertIn("stack_status", lib) + self.assertIn("mail-sync", lib) self.assertIn("qwen3.5:9b", lib) self.assertIn("picoclaw agent", lib) self.assertIn("--no-deps", lib) @@ -49,6 +51,16 @@ class StackLayoutTest(unittest.TestCase): self.assertNotIn("stack_start_assistant", start) self.assertNotIn("picoclaw agent", start) + def test_start_mail_sync_is_etl_only(self) -> None: + src = (ROOT / "bin" / "stack" / "start-mail-sync").read_text() + self.assertIn("stack_start_mail_sync", src) + self.assertNotIn("stack_start_assistant", src) + ep = (ROOT / "bin" / "docker-entrypoint").read_text() + self.assertIn('MAIL_SYNC_SRC:=onlyoffice,gmail', ep) + self.assertIn('MAIL_SYNC_INDEX:=0', ep) + self.assertIn('interval="${1:-300}"', ep) + self.assertIn('MAIL_SYNC_INDEX" = "1"', ep) + def test_start_assistant_attaches_agent(self) -> None: src = (ROOT / "bin" / "stack" / "start-assistant").read_text() self.assertIn("stack_start_assistant", src) @@ -151,7 +163,10 @@ exit 0 check=False, ) self.assertEqual(r.returncode, 0, r.stderr) - self.assertFalse(log.exists(), "healthy brain must not docker compose up") + if log.exists(): + logged = log.read_text() + self.assertNotIn("up -d", logged, "healthy brain must not docker compose up") + self.assertIn("ps", logged) # status probes mail-sync via compose ps def test_start_ups_brain_when_unhealthy(self) -> None: with tempfile.TemporaryDirectory() as raw: @@ -190,7 +205,7 @@ exit 0 logged = log.read_text() self.assertIn("stop", logged) self.assertNotIn(" down", logged) - for svc in ("brain", "brain-mcp", "reasoner", "picoclaw"): + for svc in ("brain", "brain-mcp", "reasoner", "picoclaw", "mail-sync"): self.assertIn(svc, logged) def test_start_assistant_no_attach_starts_picoclaw(self) -> None: diff --git a/compose.yaml b/compose.yaml index 434376f..d1d6c5a 100644 --- a/compose.yaml +++ b/compose.yaml @@ -92,11 +92,13 @@ services: tmpfs: - /tmp - # mail-sync: pull M365 (and OnlyOffice/Gmail) mail every 10 s into the shared - # var volume, then rebuild the brain index. Runs on the index image so it can - # convert + index in-process. Needs ~/.config/brain/m365.env for the M365 - # source; override MAIL_SYNC_SRC / MAIL_SYNC_ENV for other providers. + # mail-sync ETL: sync → import every 300s into shared var. Full brain rebuild + # only when MAIL_SYNC_INDEX=1 (expensive on large mail corpora). Default + # sources: onlyoffice,gmail. For M365 set MAIL_SYNC_SRC=m365 and put + # M365_TENANT/CLIENT_ID/CLIENT_SECRET/USERS in ~/.config/brain/mail.env + # (or m365.env + MAIL_SYNC_ENV). Gmail OAuth: mount ~/.gmail-mcp. # docker compose up -d mail-sync + # bin/stack/start-mail-sync mail-sync: image: ghcr.io/eslider/2dph:index build: @@ -106,13 +108,16 @@ services: environment: HF_HOME: /data/hf KB_PY: python3 - MAIL_SYNC_SRC: m365 - MAIL_SYNC_ENV: /secret/m365.env + MAIL_SYNC_SRC: onlyoffice,gmail + MAIL_SYNC_ENV: /secret/mail.env + MAIL_SYNC_INDEX: "0" + HOME: /home/2dph volumes: - kb-model:/data/hf - kb-var:/app/var - ~/.config/brain:/secret:ro - command: ["mail-sync", "10"] + - ~/.gmail-mcp:/home/2dph/.gmail-mcp:ro + command: ["mail-sync", "300"] read_only: true tmpfs: - /tmp diff --git a/docs/runbook.md b/docs/runbook.md index 8bf1d11..6a8f045 100644 --- a/docs/runbook.md +++ b/docs/runbook.md @@ -71,7 +71,8 @@ delete `var/kb.lbug` then `--rebuild`. ```bash bin/stack/start # brain :8630, wait until MCP search/get/audit -bin/stack/status # YAML: brain / reasoner / picoclaw +bin/stack/status # YAML: brain / reasoner / picoclaw / mail_sync +bin/stack/start-mail-sync # compose ETL: OO+Gmail sync→import (300s; no auto-rebuild) bin/stack/start-assistant # + qwen3.5:9b + PicoClaw agent (ask the brain) bin/stack/start-assistant --no-attach bin/stack/stop # compose stop; volumes kept @@ -81,6 +82,7 @@ Same Compose services by hand: ```bash docker compose up -d brain # :8630 Zig CGO serve +docker compose up -d mail-sync # ETL loop into kb-var docker compose --profile index run --rm index # rebuild docker compose --profile picoclaw up brain-mcp # MCP 127.0.0.1:8630 ``` -- 2.54.0