Compare commits

...

4 Commits

Author SHA1 Message Date
kami 2c1b0eede0 Read a web page when he names one, and watch a few on a timer (#259)
The network fallback behind the local sources, off unless configured.

internal/crawl is pure: a stdlib robots.txt parser (group specificity,
wildcards, Crawl-delay, cached per host), HTML-to-plaintext extraction, and a
watcher that notes a watched page only when its text changed. It has no store
access and no net/http; cmd/mavend/crawls.go is the impure half.

Every limit is code and tested: the guarded fetcher from #258 enforces the host
allowlist/denylist, refuses private addresses in the dialer Control hook (so DNS
rebinding and each redirect hop are covered), caps size and redirects, times out,
and spaces requests per host. A robots.txt Disallow is refused with no override.

On demand, reading is a query source placed last in the chain, after his memory,
his notes, and the local Kiwix ZIMs once those are wired: no URL in the
utterance means no fetch, and only the URL ever leaves the box. Scheduled
watches write notes and announce nothing.

The vendored tree has no x/net/html, goquery or temoto/robotstxt, so the parsers
are stdlib. No new dependency.
2026-08-01 03:40:21 +04:00
kami cb3641e7bb Read RSS and Atom feeds, and speak about them only when asked (#258)
internal/rss parses RSS 2.0 and Atom, and polls each configured feed on its own
interval; internal/webfetch is the one door either of them uses to touch the
network. The poller writes items as notes with source "rss:<feed>" and nothing
else: the answer path reads them back when he asks "что нового в лентах?", and
nothing is announced on arrival. A feed that dispatched would be a nag, which is
why the plan's breaking-news rule was left out rather than built.

webfetch is where the limits live, as code rather than a paragraph: http(s)
only, an allowlist (the configured feeds' hosts) and a denylist, a 2 MiB body
cap, a 3-redirect cap, one request per host per second, and a refusal to connect
to any private address — checked in the dialer's Control hook so it holds for
every resolved address and every redirect hop, not just for a literal IP.

Off unless configured: no "feeds" block, no poller, no outbound request. How far
a feed was read is a config fact (rss:latest:<name>), so a restart does not
re-note yesterday's headlines.
2026-08-01 03:27:45 +04:00
kami ee7bec11e3 Add mavmaild, the read-only IMAP poller that feeds mail intake (#246)
The extraction seam landed on the previous branch but nothing fed it. This
adds the daemon that does: every interval it opens one mailbox read-only
(EXAMINE + BODY.PEEK, so reading leaves no \Seen behind), fetches the UIDs
it has not handed over yet, and posts each message to core over
ingest_mail. Core runs the model and writes task candidates; this daemon
writes nothing and cannot create a reminder.

It is a separate daemon because of the credential. mavpoll set the
precedent with the zenmoney token (#125): the module talking to the third
party holds the secret, reads it from a file so it never lands in argv, in
docker-compose.yml or in shell history, and core never sees it. There is
deliberately no -password flag, and a test asserts that.

Off unless configured at both ends: without -password-file the daemon
refuses to start, and if core has no email block the first ingest returns
ErrUnknownMethod, which disables the reader instead of hammering a socket
that will keep refusing. A seen-UID state file (0600, atomic write) keeps a
restart from re-extracting the whole lookback window; correctness does not
depend on it, since capture dedupes on normalised text. Logs are counts and
UIDs — no subject, sender or body.

Verified with an in-process IMAP server and a fake core: bulk mail is
filtered before core is asked, seen UIDs are not re-fetched, a failed
ingest is retried next poll, ErrUnknownMethod stops at the first message,
and state survives a restart. The live half is untested by design — no IMAP
credential exists on this box; setup is written up as QA steps.

Vikunja #246
2026-08-01 03:13:18 +04:00
kami f42d1594ef Turn a mail into task candidates, and into nothing else (#246)
The extraction half. internal/email.Extractor asks the resident Qwen3-1.7B,
under a GBNF grammar, what one message requires of him, and returns at most
three short candidates with an optional date.

Everything it can produce is a row in `tasks` with status "candidate",
written through the intake seam #130 built for exactly this (Source
"email:<mailbox>", Evidence = the subject line). No reminder, no fact, no
note, no calendar event. That bound is the design: a reminder FIRES, so a
1.7B misreading "встреча была в четверг" as a future appointment would wake
him up about it, whereas a wrong candidate is a line he dismisses in one
click. A due date the model read out of the mail is stored on the candidate,
where no scheduler reads it — the review page sorts by it. Relative wording
("до пятницы") is deliberately left in the text rather than resolved to a
date the model would get wrong.

The prompt is written against the two things a small model does here: it
summarises when asked to extract, and it invents an obligation out of a
polite closing line. Hence the demand for a verb phrase, and an explicit
empty array — most mail contains no task, and a model with no way to say
"nothing" says something.

Wiring: core owns extraction because llama-server lives in core's process,
so the reader hands messages over a new ipc.MethodIngestMail. It is a Server
hook (like StepUp/UnlockFn), not a CoreAPI method — not a store operation,
and no CoreAPI implementation should have to carry it. The hook stays nil
without an `email` config block or without a llama-server phraser, so the
method answers ErrUnknownMethod: off unless configured, twice over. There is
no keyword fallback on purpose — "the subject became a task" is a mailbox
rendered as a to-do list, not extraction.

Privacy: junk is refused before the model is called, mail text is never
search input, extraction errors carry byte counts rather than the reply, the
stored evidence is a truncated subject, and the log line names the mailbox
and the UID only.
2026-08-01 03:06:55 +04:00
48 changed files with 5271 additions and 31 deletions
+3
View File
@@ -7,6 +7,7 @@
/mavpoll
/mavcaldav
/mavwaked
/mavmaild
# Certs (private keys, don't commit)
certs/
@@ -36,6 +37,8 @@ deploy/db_key.env
deploy/telegram.env
# zenmoney API token, read by mavpoll (never in argv, never committed)
deploy/zenmoney.token
# IMAP password, read by mavmaild (never in argv, never committed)
deploy/imap.password
# Temp files
/tmp/
+2 -1
View File
@@ -34,7 +34,7 @@ CGO daemons (`mavend`, `mavsttd`, `mavttsd`, `mavenclient`) need the vendored to
and libs wired through the Makefile — **do not** call `go build` on them bare, use `make`:
```sh
make build # all 8 binaries
make build # all 9 binaries
make build-web # single daemon (pure-Go ones: web/waked/poll/caldav build without CGO)
make test # go test -race across ./internal/... ./cmd/... with CGO env set
```
@@ -62,6 +62,7 @@ Pure-Go packages (`router`, `memory`, `mavweb`, …) run under a plain `go test
| `mavenclient` | Voice loop client (mic → stt → core → tts). |
| `mavpoll` | Telegram long-poll reach. |
| `mavcaldav` | CalDAV calendar sync. |
| `mavmaild` | Mail reader (IMAP, read-only). Holds the IMAP password; core never sees it. |
Daemons are wired socket-to-socket, not linked. `internal/ipc` is the client/server wire
protocol; the config in `deploy/mavend.json` (with `${VAR}` env expansion from gitignored
+2 -1
View File
@@ -51,7 +51,8 @@ RUN go build -o /out/mavend ./cmd/mavend && \
go build -o /out/mavttsd ./cmd/mavttsd && \
go build -o /out/mavweb ./cmd/mavweb && \
go build -o /out/mavpoll ./cmd/mavpoll && \
go build -o /out/mavcaldav ./cmd/mavcaldav
go build -o /out/mavcaldav ./cmd/mavcaldav && \
go build -o /out/mavmaild ./cmd/mavmaild
# llama.cpp Vulkan build — the phraser/router LFM engine (llama-server). Built
# from source (not a prebuilt vendored blob) so the binary's glibc/GLIBCXX match
+5 -2
View File
@@ -20,7 +20,7 @@ PIPER_ESPEAK := $(shell pwd)/deps/piper/espeak-ng-data
all: build
build: build-stt build-tts build-daemon build-client build-waked build-web build-poll build-caldav
build: build-stt build-tts build-daemon build-client build-waked build-web build-poll build-caldav build-mail
build-stt:
CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" LD_LIBRARY_PATH="$(shell pwd)/deps/lib" \
@@ -50,6 +50,9 @@ build-poll:
build-caldav:
$(GO) build $(GOFLAGS) -o mavcaldav ./cmd/mavcaldav/
build-mail:
$(GO) build $(GOFLAGS) -o mavmaild ./cmd/mavmaild/
run-web: build-web
./mavweb -addr :9200 -voice 127.0.0.1:9100
@@ -185,4 +188,4 @@ download-embedder:
@echo ' sudo cp onnxruntime-linux-x64-1.15.1/lib/libonnxruntime.so* /usr/local/lib/'
clean:
rm -f mavend mavenclient mavsttd mavttsd mavweb mavpoll mavcaldav mavwaked
rm -f mavend mavenclient mavsttd mavttsd mavweb mavpoll mavcaldav mavwaked mavmaild
+122
View File
@@ -8,10 +8,12 @@ import (
"strings"
"time"
"github.com/kami/maven/internal/crawl"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/memory"
"github.com/kami/maven/internal/morning"
"github.com/kami/maven/internal/router"
"github.com/kami/maven/internal/rss"
"github.com/kami/maven/internal/weather"
)
@@ -67,11 +69,23 @@ var querySources = []querySource{
// answer it from whatever he once said about spending. Its matcher needs a
// money noun plus an actual ask, so "я потратил весь день" is untouched.
{"money", (*reactiveHandler).queryMoney},
// Before the recall sources and before general knowledge: "что нового?" is
// a question about the feeds she reads, and general knowledge would answer
// it by inventing news. Its matcher needs a feed noun plus an ask, so
// "у меня новая лента в инстаграме" is untouched.
{"feeds", (*reactiveHandler).queryFeeds},
{"calendar", (*reactiveHandler).queryCalendar},
{"weather", (*reactiveHandler).queryWeather},
{"embed", (*reactiveHandler).queryEmbed},
{"memory", (*reactiveHandler).queryMemory},
{"notes", (*reactiveHandler).queryNotes},
// LAST before the model answers from memory, and that position is the whole
// design (Vikunja #259): local sources first. The model, his own notes and
// facts, and — once internal/kiwix is wired into this chain — the offline
// ZIMs all get their turn before anything touches the network. This source
// only claims a turn where he named a URL out loud, so it never competes
// with a local answer.
{"web", (*reactiveHandler).queryWeb},
{"general-knowledge", (*reactiveHandler).queryGeneral},
}
@@ -180,6 +194,63 @@ func (h *reactiveHandler) queryHabits(ctx context.Context, t *queryTurn) (string
return profile.FormatOverallRU(), true
}
// feedNoteWindow — how many recent notes are scanned for feed items, and
// feedReadOut — how many headlines she actually reads back. She summarises the
// top of the pile, she does not recite a river.
const (
feedNoteWindow = 200
feedReadOut = 3
)
// queryFeeds — "что нового в лентах?", "что нового по технологиям?"
// (Vikunja #258).
//
// This is the ONLY way a feed item reaches him. The poller writes notes and
// never speaks; asking is the trigger. If that ever changes, the thing that
// changed is "Maven is not a nag", not a detail of this file.
func (h *reactiveHandler) queryFeeds(ctx context.Context, t *queryTurn) (string, bool) {
q, ok := router.ParseFeedQuery(t.dec.Utterance)
if !ok {
return "", false
}
if !h.feedsOn {
// Claim the turn rather than fall through: "не читаю ленты" is true, and
// letting general knowledge answer "что нового?" would be an invented
// news bulletin.
return "я пока не читаю ленты — они не настроены.", true
}
notes, err := h.api.RecentNotes(ctx, feedNoteWindow)
if err != nil {
log.Printf("voice: feeds: recent notes: %v", err)
return "не получилось посмотреть ленты.", true
}
var picked []string
for _, n := range notes {
if !strings.HasPrefix(n.Source, rss.SourcePrefix) {
continue
}
if !router.CategoryMatches(n.Text, q.Category) {
continue
}
// The note carries title, summary and link; she reads the title.
title := n.Text
if i := strings.IndexByte(title, '\n'); i > 0 {
title = title[:i]
}
picked = append(picked, strings.TrimSpace(title))
if len(picked) == feedReadOut {
break
}
}
if len(picked) == 0 {
if q.Category != "" {
return "по этой теме в лентах пока ничего.", true
}
return "в лентах пока ничего нового.", true
}
return "вот что нового: " + strings.Join(picked, "; "), true
}
// queryCalendar — "что у меня сегодня?", "планы на завтра?"
// h.now(), not time.Now(): the handler's clock is the injected one, so this
// source can be tested at a fixed time like the rest.
@@ -308,6 +379,57 @@ func (h *reactiveHandler) queryNotes(ctx context.Context, t *queryTurn) (string,
return reply, true
}
// webPageContextRunes — how much of a fetched page is handed to the phraser.
// Less than the crawler keeps: the rest of the 4096-token window belongs to the
// prompt, the persona block and the reply.
const webPageContextRunes = 1500
// queryWeb — "посмотри https://example.org/x — что там?" (Vikunja #259).
//
// It claims a turn ONLY when he named a URL, which is what keeps a fallback from
// becoming a habit: no URL, no fetch, and the model answers from what is local.
// What leaves the box is the URL and nothing else — no note, no fact, no history
// travels with it.
func (h *reactiveHandler) queryWeb(ctx context.Context, t *queryTurn) (string, bool) {
link, ok := router.FirstURL(t.dec.Utterance)
if !ok {
return "", false
}
if h.crawler == nil {
// Claim rather than fall through: he asked about a specific page, and
// letting the model answer from the URL's spelling alone is how a small
// model invents a page's contents.
return "я не читаю страницы — это не настроено.", true
}
ctxFetch, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
page, err := h.crawler.Page(ctxFetch, link)
if err != nil {
if errors.Is(err, crawl.ErrRobots) {
return "эта страница закрыта для чтения — robots.txt не разрешает.", true
}
log.Printf("voice: web: %v", err)
return "не получилось прочитать страницу.", true
}
if page.Text == "" {
return "страница открылась, но читать там нечего.", true
}
// The page is handed to the phraser the same way a note is: as context for
// the question he actually asked. She answers the question, she does not
// recite the page.
snippet := page.Title + "\n" + crawl.TrimRunes(page.Text, webPageContextRunes)
reply, perr := h.phraser.PhraseQuery(ctx, t.dec.Utterance, []string{snippet})
if perr != nil {
log.Printf("voice: web: phrase: %v", perr)
}
if reply == "" {
// No phraser (or it failed): read back the top of the page rather than
// pretend the fetch did not happen.
return "вот что на странице: " + crawl.TrimRunes(page.Text, 300), true
}
return reply, true
}
// queryGeneral — general knowledge from the phraser, the last source before
// giving up. It always claims: either the model answers or Maven says she
// doesn't know.
+184
View File
@@ -0,0 +1,184 @@
// mavend/crawls.go — the driver for reading web pages (Vikunja #259,
// docs/plans/14-web-crawler.md). The crawler is pure and lives in
// internal/crawl; this is the impure half: the guarded fetcher, a ticker for the
// scheduled watches, and the fact-backed dedup hashes.
//
// Two paths, one config block, both off unless configured:
//
// - ON DEMAND — he names a URL out loud and she reads it. That is the
// `queryWeb` source in actions_query.go, LAST in the chain: after his
// memory, after the notes, and (once Kiwix is wired into the chain) after
// the local ZIMs. A local read costs nothing and leaks nothing; a fetch puts
// a URL in someone's log, so it goes last.
// - SCHEDULED — a watched page is re-read on its interval, and a page whose
// text changed is written as a note. It does NOT announce itself. Same rule
// as the feed poller: notes, never nudges.
//
// Only the URL goes out. Nothing here reads a note, a fact, the persona block or
// the history, and internal/crawl has no access to the store at all.
package main
import (
"context"
"log"
"net/url"
"time"
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/crawl"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/router"
"github.com/kami/maven/internal/webfetch"
)
// newCrawler builds the crawler from the `crawl` block, or returns nil when
// there is none. Every caller checks for nil, and nil means no page is ever
// fetched.
func newCrawler(cfg *config.Config) *crawl.Crawler {
if cfg.Crawl == nil {
return nil
}
cc := cfg.Crawl
hosts := append([]string(nil), cc.AllowHosts...)
// A watched page's own host is always reachable; otherwise an allowlist and
// a watch list would have to be kept in sync by hand.
for _, w := range cc.Watches {
if u, err := url.Parse(w.URL); err == nil && u.Hostname() != "" {
hosts = append(hosts, u.Hostname())
}
}
// An allowlist plus on-demand is a contradiction worth logging rather than
// silently resolving: he asked for arbitrary pages AND for a fixed list.
// The allowlist wins, because it is the narrower instruction.
if len(hosts) > 0 && cc.OnDemand && len(cc.AllowHosts) > 0 {
log.Printf("crawl: allow_hosts is set, so on-demand reading is limited to those hosts")
}
ua := cc.UserAgent
if ua == "" {
ua = webfetch.DefaultUserAgent
}
fetcher := webfetch.New(webfetch.Config{
AllowHosts: hosts,
DenyHosts: cc.DenyHosts,
Timeout: time.Duration(cc.Timeout),
MaxBytes: cc.MaxBytes,
UserAgent: ua,
})
// The user-agent handed to the crawler is the one the fetcher sends: obeying
// robots rules written for a different name would be a lie.
return crawl.New(&crawlFetcher{f: fetcher}, crawl.Config{
UserAgent: ua,
MaxRunes: cc.MaxRunes,
})
}
// onDemandCrawler returns a crawler for the answer path, or nil when on-demand
// reading is off. The scheduled watches can be on while this is off: reading a
// fixed list of pages on a timer and reading whatever URL is in an utterance are
// different permissions, and the config keeps them separate.
func onDemandCrawler(cfg *config.Config) *crawl.Crawler {
if cfg.Crawl == nil || !cfg.Crawl.OnDemand {
return nil
}
return newCrawler(cfg)
}
// crawlWorker — ticker + watcher for the scheduled half.
type crawlWorker struct {
watcher *crawl.Watcher
interval time.Duration
}
// crawlTickInterval — how often the worker asks what is due. Per-watch cadence
// is the watcher's business.
const crawlTickInterval = 15 * time.Minute
// newCrawlWorker wires the scheduled crawls, or nil when nothing is watched.
func newCrawlWorker(c *crawl.Crawler, api ipc.CoreAPI, emb router.Embedder, cfg *config.Config) *crawlWorker {
if c == nil || cfg.Crawl == nil || len(cfg.Crawl.Watches) == 0 {
return nil
}
watches := make([]crawl.WatchConfig, 0, len(cfg.Crawl.Watches))
for _, w := range cfg.Crawl.Watches {
watches = append(watches, crawl.WatchConfig{
Name: w.Name,
URL: w.URL,
Interval: time.Duration(w.Interval),
})
}
watcher := crawl.NewWatcher(c, watches, api, &factHashes{api: api},
crawlEmbedder(emb), time.Duration(cfg.Crawl.Interval))
if watcher == nil {
log.Printf("crawl: configured but nothing watchable — scheduled crawls disabled")
return nil
}
log.Printf("crawl: watching %d page(s), checking what is due every %s", len(watches), crawlTickInterval)
return &crawlWorker{watcher: watcher, interval: crawlTickInterval}
}
// run checks what is due until ctx is canceled. The first round runs
// immediately; it writes notes only, so an early round startles nobody.
func (w *crawlWorker) run(ctx context.Context) {
w.watcher.CheckDue(ctx, time.Now())
t := time.NewTicker(w.interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case now := <-t.C:
w.watcher.CheckDue(ctx, now)
}
}
}
// crawlFetcher adapts webfetch to crawl.Fetcher, which is the seam that keeps
// net/http out of the crawler package.
type crawlFetcher struct{ f *webfetch.Fetcher }
func (a *crawlFetcher) Get(ctx context.Context, u string) (*crawl.Response, error) {
resp, err := a.f.Get(ctx, u)
if err != nil {
return nil, err
}
return &crawl.Response{URL: resp.URL, ContentType: resp.ContentType, Body: resp.Body}, nil
}
// factHashes stores each watch's last content hash as a config fact, so a
// restart does not re-note an unchanged page. Same mechanism the feed reader
// uses for its marks, and inspectable on /dash.
type factHashes struct{ api ipc.CoreAPI }
func hashKey(name string) string { return "crawl:hash:" + name }
func (h *factHashes) LastHash(ctx context.Context, name string) (string, error) {
f, err := h.api.LatestFact(ctx, hashKey(name))
if err != nil {
// No hash yet is not an error: the watcher treats "" as "never read".
return "", nil
}
return f.Value, nil
}
func (h *factHashes) SetHash(ctx context.Context, name, hash string) error {
_, err := h.api.WriteFact(ctx, ipc.WriteFactReq{
Ts: time.Now(),
Kind: "config",
Key: hashKey(name),
Value: hash,
Source: "poll:crawl",
Confidence: 1.0,
})
return err
}
// crawlEmbedder adapts router.Embedder for the watcher, embedding with
// EmbedPassage (a page is text being searched FOR, and the e5 embedder is
// asymmetric).
func crawlEmbedder(emb router.Embedder) crawl.Embedder {
if emb == nil {
return nil
}
return passageEmbedder{emb}
}
+186
View File
@@ -0,0 +1,186 @@
package main
import (
"context"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/crawl"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/phraser"
"github.com/kami/maven/internal/router"
"github.com/kami/maven/internal/voice"
)
// The default config reads nothing. This is the whole "off unless configured"
// contract for the crawler, asserted at the wiring level rather than trusted.
func TestCrawlOffByDefault(t *testing.T) {
cfg := &config.Config{}
if c := newCrawler(cfg); c != nil {
t.Error("newCrawler with no crawl block returned a crawler")
}
if c := onDemandCrawler(cfg); c != nil {
t.Error("onDemandCrawler with no crawl block returned a crawler")
}
if w := newCrawlWorker(nil, nil, nil, cfg); w != nil {
t.Error("newCrawlWorker with no crawl block returned a worker")
}
// Watches configured but on_demand off ⇒ the answer path still reads
// nothing: a timer over a fixed list is not permission for arbitrary URLs.
withWatch := &config.Config{Crawl: &config.CrawlConfig{
Watches: []config.CrawlWatchConfig{{Name: "p", URL: "https://example.org/p"}},
}}
if c := onDemandCrawler(withWatch); c != nil {
t.Error("onDemandCrawler honoured a watch list as on-demand permission")
}
if c := newCrawler(withWatch); c == nil {
t.Error("newCrawler returned nil for a configured watch")
}
}
// The wired fetcher must refuse a private address, because the crawler on this
// box sits one hop from the whole homelab. Same guard the webfetch tests cover;
// this asserts the daemon actually wires it.
func TestCrawlerRefusesPrivateAddress(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/html")
w.Write([]byte("<html><body>secret</body></html>"))
}))
defer srv.Close()
c := newCrawler(&config.Config{Crawl: &config.CrawlConfig{OnDemand: true}})
if c == nil {
t.Fatal("newCrawler returned nil for an on-demand config")
}
if _, err := c.Page(context.Background(), srv.URL); err == nil {
t.Fatalf("reading %s succeeded; a loopback address must be refused", srv.URL)
}
}
func TestFactHashesRoundTrip(t *testing.T) {
ctx := context.Background()
st := newTestStore(t)
h := &factHashes{api: ipc.NewStoreAPI(st)}
got, err := h.LastHash(ctx, "page")
if err != nil {
t.Fatalf("LastHash on a fresh store: %v", err)
}
if got != "" {
t.Errorf("LastHash = %q, want empty for a never-read page", got)
}
if err := h.SetHash(ctx, "page", "deadbeef"); err != nil {
t.Fatalf("SetHash: %v", err)
}
got, err = h.LastHash(ctx, "page")
if err != nil {
t.Fatalf("LastHash: %v", err)
}
if got != "deadbeef" {
t.Errorf("LastHash = %q, want deadbeef", got)
}
if key := hashKey("page"); key != "crawl:hash:page" {
t.Errorf("hashKey = %q", key)
}
}
// stubCrawlFetcher serves one fixed page to every URL, so queryWeb can be
// exercised without a network or an allowlist.
type stubCrawlFetcher struct{ body, ctype string }
func (s *stubCrawlFetcher) Get(_ context.Context, u string) (*crawl.Response, error) {
ct := s.ctype
if ct == "" {
ct = "text/html"
}
if strings.HasSuffix(u, "/robots.txt") {
return &crawl.Response{URL: u, ContentType: "text/plain", Body: []byte("")}, nil
}
return &crawl.Response{URL: u, ContentType: ct, Body: []byte(s.body)}, nil
}
func buildWebHandler(c *crawl.Crawler) *reactiveHandler {
return &reactiveHandler{
replier: voice.NewStubReplier(),
phraser: phraser.NewStub(),
crawler: c,
}
}
func askWeb(h *reactiveHandler, q string) (string, bool) {
return h.queryWeb(context.Background(), &queryTurn{
dec: router.Decision{Intent: router.IntentQuery, Utterance: q},
})
}
func TestQueryWebPassesWithoutAURL(t *testing.T) {
h := buildWebHandler(crawl.New(&stubCrawlFetcher{body: "<html><body>x</body></html>"}, crawl.Config{}))
if reply, ok := askWeb(h, "почему небо синее?"); ok {
t.Errorf("the web source claimed a question with no URL: %q", reply)
}
}
// Not configured is said out loud rather than falling through, so a small model
// never invents a page's contents from its URL.
func TestQueryWebSaysWhenNotConfigured(t *testing.T) {
h := buildWebHandler(nil)
reply, ok := askWeb(h, "посмотри https://example.org/page")
if !ok {
t.Fatal("the web source did not claim a question with a URL")
}
if !strings.Contains(reply, "не настроено") {
t.Errorf("reply = %q, want the not-configured answer", reply)
}
}
func TestQueryWebReadsThePage(t *testing.T) {
h := buildWebHandler(crawl.New(&stubCrawlFetcher{
body: "<html><head><title>Заголовок</title></head><body><p>текст страницы</p></body></html>",
}, crawl.Config{}))
reply, ok := askWeb(h, "посмотри https://example.org/page — что там?")
if !ok {
t.Fatal("the web source did not claim a question with a URL")
}
if !strings.Contains(reply, "текст страницы") {
t.Errorf("reply = %q, want the page text read back", reply)
}
}
func TestQueryWebRefusesNonHTML(t *testing.T) {
h := buildWebHandler(crawl.New(&stubCrawlFetcher{
body: "\x00\x01binary", ctype: "application/octet-stream",
}, crawl.Config{}))
reply, ok := askWeb(h, "почитай https://example.org/blob.bin")
if !ok {
t.Fatal("the web source did not claim a question with a URL")
}
if !strings.Contains(reply, "не получилось") {
t.Errorf("reply = %q, want the read-failed answer", reply)
}
}
// robots.txt is honoured on the answer path too, and she says so instead of
// reporting a generic failure.
func TestQueryWebObeysRobots(t *testing.T) {
h := buildWebHandler(crawl.New(&robotsDenyFetcher{}, crawl.Config{}))
reply, ok := askWeb(h, "посмотри https://example.org/private")
if !ok {
t.Fatal("the web source did not claim a question with a URL")
}
if !strings.Contains(reply, "robots.txt") {
t.Errorf("reply = %q, want the robots answer", reply)
}
}
type robotsDenyFetcher struct{}
func (robotsDenyFetcher) Get(_ context.Context, u string) (*crawl.Response, error) {
if strings.HasSuffix(u, "/robots.txt") {
return &crawl.Response{URL: u, ContentType: "text/plain",
Body: []byte("User-agent: *\nDisallow: /private\n")}, nil
}
return &crawl.Response{URL: u, ContentType: "text/html", Body: []byte("<html>nope</html>")}, nil
}
+182
View File
@@ -0,0 +1,182 @@
// mavend/feeds.go — the driver for RSS/Atom reading (Vikunja #258,
// docs/plans/13-rss-news-feeds.md). The reader itself is pure and lives in
// internal/rss; this is the impure half: a ticker, the guarded fetcher, and the
// two adapters that let a pure package talk to the store.
//
// Why in-core rather than its own daemon like mavmaild and mavpoll: those two
// hold a CREDENTIAL (an IMAP password, a zenmoney token), and the reason they
// are separate processes is that core must never see it. A feed URL is public,
// there is no secret to isolate, and a whole extra binary and compose service
// would buy nothing. The other half of the mavpoll precedent — off unless
// configured — is kept: no `feeds` block, no poller, no outbound request.
//
// It is its own goroutine, not a step on the tick: the tick has a delivery
// deadline behind it, and a feed read is a network round-trip that nobody is
// waiting on.
//
// Nothing here dispatches. A feed that announced itself would be a nag, so the
// only output is notes with source "rss:<feed>", which the answer path reads
// when he asks ("что нового в лентах?" — see queryFeeds in actions_query.go).
package main
import (
"context"
"log"
"net/url"
"time"
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/router"
"github.com/kami/maven/internal/rss"
"github.com/kami/maven/internal/webfetch"
)
// feedWorker — ticker + poller.
type feedWorker struct {
poller *rss.Poller
interval time.Duration
}
// feedTickInterval — how often the worker asks the poller what is due. Per-feed
// cadence is the poller's business; this is just the granularity.
const feedTickInterval = 5 * time.Minute
// newFeedWorker wires feed reading, or returns nil when it must not run:
// no `feeds` block (the normal case), or nothing valid in it. Every caller
// checks for nil.
func newFeedWorker(api ipc.CoreAPI, emb router.Embedder, cfg *config.Config) *feedWorker {
if cfg.Feeds == nil {
return nil
}
fc := cfg.Feeds
feeds := make([]rss.FeedConfig, 0, len(fc.Sources))
hosts := append([]string(nil), fc.AllowHosts...)
for _, s := range fc.Sources {
feeds = append(feeds, rss.FeedConfig{
Name: s.Name,
URL: s.URL,
Category: s.Category,
Interval: time.Duration(s.Interval),
Include: s.Include,
Exclude: s.Exclude,
})
// Each configured feed's own host is allowed. The allowlist is then
// exactly "the feeds he asked for", so a redirect off to somewhere else
// is refused by the fetcher rather than followed.
if u, err := url.Parse(s.URL); err == nil && u.Hostname() != "" {
hosts = append(hosts, u.Hostname())
}
}
fetcher := webfetch.New(webfetch.Config{
AllowHosts: hosts,
Timeout: time.Duration(fc.Timeout),
MaxBytes: fc.MaxBytes,
})
poller := rss.NewPoller(feeds, &feedFetcher{f: fetcher}, api, &factMarks{api: api},
embedderFor(emb), nil, rss.Config{
DefaultInterval: time.Duration(fc.PollInterval),
MaxItems: fc.MaxItems,
MaxAge: time.Duration(fc.MaxAge),
})
if poller == nil {
log.Printf("feeds: configured but nothing pollable — feed reading disabled")
return nil
}
log.Printf("feeds: reading %d feed(s), checking what is due every %s", len(feeds), feedTickInterval)
return &feedWorker{poller: poller, interval: feedTickInterval}
}
// run polls what is due until ctx is canceled. The first round runs immediately
// so a restart does not blind her for the first interval; it writes notes only,
// so an early round cannot startle anyone.
func (w *feedWorker) run(ctx context.Context) {
w.poller.PollDue(ctx, time.Now())
t := time.NewTicker(w.interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
return
case now := <-t.C:
w.poller.PollDue(ctx, now)
}
}
}
// embedderOf — the voice wiring's embedder, or nil when voice is not wired.
// Feed notes are embedded with the SAME model the rest of the store uses, or not
// at all; a second embedder would write vectors nothing can search.
func embedderOf(w *voiceWiring) router.Embedder {
if w == nil {
return nil
}
return w.embedder
}
// feedFetcher adapts webfetch to rss.Fetcher — the pure package names the two
// fields it needs and stays free of net/http.
type feedFetcher struct{ f *webfetch.Fetcher }
func (a *feedFetcher) Get(ctx context.Context, u string) (*rss.Body, error) {
resp, err := a.f.Get(ctx, u)
if err != nil {
return nil, err
}
return &rss.Body{Bytes: resp.Body}, nil
}
// factMarks stores "how far this feed was read" as a config fact, the same
// mechanism the plan named and the same one the pattern tick uses for its own
// bookkeeping. Durable, inspectable on /dash, and cheap.
type factMarks struct{ api ipc.CoreAPI }
func markKey(feed string) string { return "rss:latest:" + feed }
func (m *factMarks) LastMark(ctx context.Context, feed string) (time.Time, error) {
f, err := m.api.LatestFact(ctx, markKey(feed))
if err != nil {
// No mark yet is not an error worth propagating: the poller treats a
// zero time as a cold start.
return time.Time{}, nil
}
t, err := time.Parse(time.RFC3339, f.Value)
if err != nil {
return time.Time{}, nil
}
return t, nil
}
func (m *factMarks) SetMark(ctx context.Context, feed string, at time.Time) error {
_, err := m.api.WriteFact(ctx, ipc.WriteFactReq{
Ts: time.Now(),
Kind: "config",
Key: markKey(feed),
Value: at.UTC().Format(time.RFC3339),
Source: "poll:rss",
Confidence: 1.0,
})
return err
}
// embedderFor adapts router.Embedder to rss.Embedder, and returns nil when
// there is none — a note without a vector is still a note the recent-notes path
// can read.
//
// EmbedPassage, not Embed: a feed item is text being searched FOR, and the e5
// embedder is asymmetric. Getting this backwards makes the item unfindable by
// the question that should have matched it.
func embedderFor(emb router.Embedder) rss.Embedder {
if emb == nil {
return nil
}
return passageEmbedder{emb}
}
type passageEmbedder struct{ e router.Embedder }
func (p passageEmbedder) Embed(ctx context.Context, text string) ([]float32, error) {
return router.EmbedPassage(ctx, p.e, text)
}
+177
View File
@@ -0,0 +1,177 @@
package main
import (
"context"
"strings"
"testing"
"time"
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/phraser"
"github.com/kami/maven/internal/router"
"github.com/kami/maven/internal/rss"
"github.com/kami/maven/internal/voice"
)
// buildFeedHandler — a handler with the given feed notes already stored. No
// embedder: the feed source answers from recent notes by source, which is what
// makes it work for notes written before an embedder existed.
func buildFeedHandler(t *testing.T, feedsOn bool, notes ...ipc.Note) *reactiveHandler {
t.Helper()
ctx := context.Background()
st := newTestStore(t)
now := time.Now()
for i, n := range notes {
ts := now.Add(time.Duration(i) * time.Minute)
if _, err := st.WriteNote(ctx, ts, n.Text, nil, n.Source); err != nil {
t.Fatalf("WriteNote: %v", err)
}
}
return &reactiveHandler{
api: ipc.NewStoreAPI(st),
replier: voice.NewStubReplier(),
phraser: phraser.NewStub(),
now: func() time.Time { return now },
feedsOn: feedsOn,
embedder: nil,
}
}
func askFeeds(t *testing.T, h *reactiveHandler, q string) (string, bool) {
t.Helper()
return h.queryFeeds(context.Background(), &queryTurn{
dec: router.Decision{Intent: router.IntentQuery, Utterance: q},
})
}
func TestQueryFeedsReadsFeedNotes(t *testing.T) {
h := buildFeedHandler(t, true,
ipc.Note{Text: "Новая уязвимость в ядре [технологии]\nпатч вышел\nhttps://example.org/a", Source: "rss:habr"},
ipc.Note{Text: "что-то он сам сказал", Source: "tap:voice"},
)
reply, ok := askFeeds(t, h, "что нового в лентах?")
if !ok {
t.Fatal("the feed source did not claim the question")
}
if !strings.Contains(reply, "уязвимость") {
t.Errorf("reply = %q, want the headline", reply)
}
if strings.Contains(reply, "он сам сказал") {
t.Errorf("a note he dictated leaked into the feed answer: %q", reply)
}
// She reads the headline, not the summary and not the URL.
if strings.Contains(reply, "https://") || strings.Contains(reply, "патч вышел") {
t.Errorf("reply = %q, want the title line only", reply)
}
}
func TestQueryFeedsByCategory(t *testing.T) {
h := buildFeedHandler(t, true,
ipc.Note{Text: "Релиз ядра [технологии]", Source: "rss:habr"},
ipc.Note{Text: "Выборы отложены [политика]", Source: "rss:news"},
)
reply, ok := askFeeds(t, h, "что нового по технологиям?")
if !ok {
t.Fatal("not claimed")
}
if !strings.Contains(reply, "ядра") || strings.Contains(reply, "Выборы") {
t.Fatalf("reply = %q, want only the технологии item", reply)
}
reply, _ = askFeeds(t, h, "что нового по спорту?")
if !strings.Contains(reply, "ничего") {
t.Fatalf("reply = %q, want an honest empty answer for an unread category", reply)
}
}
// "не настроены" and "ничего нового" are different truths, and neither may be
// answered by the model inventing a bulletin.
func TestQueryFeedsOffAndEmptyDiffer(t *testing.T) {
off := buildFeedHandler(t, false)
reply, ok := askFeeds(t, off, "что нового?")
if !ok || !strings.Contains(reply, "не настроены") {
t.Fatalf("feeds off: reply = %q, ok = %v", reply, ok)
}
on := buildFeedHandler(t, true)
reply, ok = askFeeds(t, on, "что нового?")
if !ok || !strings.Contains(reply, "ничего нового") {
t.Fatalf("feeds on but empty: reply = %q, ok = %v", reply, ok)
}
}
func TestQueryFeedsPassesOnANonFeedQuestion(t *testing.T) {
h := buildFeedHandler(t, true)
if reply, ok := askFeeds(t, h, "напомни полить цветы"); ok {
t.Fatalf("claimed an unrelated question with %q", reply)
}
}
// The mark is what stops a restart from re-noting yesterday's headlines, so the
// fact round-trip is worth a test of its own.
func TestFactMarksRoundTrip(t *testing.T) {
st := newTestStore(t)
m := &factMarks{api: ipc.NewStoreAPI(st)}
ctx := context.Background()
at, err := m.LastMark(ctx, "habr")
if err != nil || !at.IsZero() {
t.Fatalf("no mark yet: got %v, %v — want zero time and no error", at, err)
}
want := time.Date(2026, 7, 28, 10, 0, 0, 0, time.UTC)
if err := m.SetMark(ctx, "habr", want); err != nil {
t.Fatal(err)
}
got, err := m.LastMark(ctx, "habr")
if err != nil {
t.Fatal(err)
}
if !got.Equal(want) {
t.Fatalf("mark = %v, want %v", got, want)
}
}
// Off unless configured, checked at the wiring seam: no `feeds` block ⇒ no
// worker ⇒ no outbound request is possible.
func TestNewFeedWorkerOffByDefault(t *testing.T) {
st := newTestStore(t)
api := ipc.NewStoreAPI(st)
if w := newFeedWorker(api, nil, &config.Config{}); w != nil {
t.Fatal("a config with no feeds block wired a feed worker")
}
// An empty sources list is normalised to "off" by config.Load; the worker
// refuses it too, so a hand-built Config cannot switch it on by accident.
if w := newFeedWorker(api, nil, &config.Config{Feeds: &config.FeedsConfig{}}); w != nil {
t.Fatal("an empty sources list wired a feed worker")
}
cfg := &config.Config{Feeds: &config.FeedsConfig{Sources: []config.FeedSourceConfig{
{Name: "habr", URL: "https://example.org/rss"},
}}}
w := newFeedWorker(api, nil, cfg)
if w == nil {
t.Fatal("a configured feed did not wire a worker")
}
if got := w.poller.Feeds(); len(got) != 1 || got[0].Name != "habr" {
t.Fatalf("feeds = %+v", got)
}
}
// The fetcher the worker builds must be allowlisted to the configured feeds and
// nothing else — the crawler's SSRF guards are only worth as much as the
// allowlist handed to them.
func TestFeedWorkerFetcherIsAllowlisted(t *testing.T) {
cfg := &config.Config{Feeds: &config.FeedsConfig{Sources: []config.FeedSourceConfig{
{Name: "habr", URL: "https://feeds.example.org/rss"},
}}}
w := newFeedWorker(ipc.NewStoreAPI(newTestStore(t)), nil, cfg)
if w == nil {
t.Fatal("no worker")
}
// PollFeed goes through the guarded fetcher; a feed URL pointing at the box
// itself must fail rather than be read.
_, err := w.poller.PollFeed(context.Background(), rss.FeedConfig{
Name: "evil", URL: "http://127.0.0.1:9100/mcp",
}, time.Now())
if err == nil {
t.Fatal("the poller fetched a private address")
}
}
+158
View File
@@ -0,0 +1,158 @@
// mavend/mail.go — core's half of the email reader (Vikunja #246,
// docs/plans/01-email-reader.md).
//
// The split: cmd/mavmaild holds the IMAP credential, connects to the mailbox
// and converts messages to plaintext; it hands each message to core over
// ipc.MethodIngestMail. Core runs the extraction on the resident model —
// llama-server lives in this process, spawned by the phraser — and writes what
// comes back through the one task intake seam.
//
// What this file may produce is exactly one thing: rows in `tasks` with status
// "candidate". No fact, no reminder, no note, no nudge, no calendar event. A
// 1.7B misreading a mail can therefore put a wrong line on a review page and
// nothing else; it can never make Maven speak, and it can never make her
// recite something out of an advert as true.
//
// Off unless configured twice over: no `email` block in mavend.json ⇒ the IPC
// method does not exist; no llama-server phraser ⇒ same. A reader pointed at a
// core that is not set up for mail gets ErrUnknownMethod rather than silence.
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/email"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/llm"
"github.com/kami/maven/internal/phraser"
"github.com/kami/maven/internal/store"
)
// evidenceMaxChars — how much of the subject line is kept as a candidate's
// evidence. Enough to recognise the mail on /tasks, not enough to turn the task
// list into a copy of his mailbox.
const evidenceMaxChars = 160
// mailIntake — extraction + capture for one message at a time.
type mailIntake struct {
st *store.Store
ex *email.Extractor
timeout time.Duration
now func() time.Time
}
// newMailIntake returns nil when mail ingestion must not be available, which is
// the default. Both preconditions are real:
//
// - no cfg.Email ⇒ not configured, and a capability is off unless configured;
// - no llama-server phraser ⇒ nothing to extract with. There is deliberately
// no keyword fallback: "the subject line became a task" is not extraction,
// it is a mailbox rendered as a to-do list, and it would fill the review
// page faster than he could clear it.
func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config) *mailIntake {
if cfg.Email == nil {
return nil
}
lp, ok := phr.(*phraser.LLMPhraser)
if !ok {
log.Printf("mail intake: configured but no llama-server phraser — mail ingestion disabled")
return nil
}
timeout := time.Duration(cfg.Email.Timeout)
if timeout <= 0 {
timeout = config.DefaultEmailTimeout
}
ex := email.NewExtractor(llm.New(lp.BaseURL(), timeout), cfg.Email.MaxTasks, contextBlockFn(cfg, time.Now))
log.Printf("mail intake: enabled (max %d candidates per message, timeout %s)", cfg.Email.MaxTasks, timeout)
return &mailIntake{st: st, ex: ex, timeout: timeout, now: time.Now}
}
// ingest handles one ipc.MethodIngestMail call.
//
// Junk and empty messages are answered Skipped without touching the model — the
// reader's header filter is what keeps the resident model off newsletters.
//
// Every candidate is captured with Status "candidate", Source "email:<mailbox>"
// and the subject as Evidence. CaptureTask dedupes on normalised text among
// live rows, so a mailbox re-read after a restart produces Created=0 rather
// than a second copy of every task.
func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error) {
msg := email.Message{
UID: req.UID,
From: req.From,
Subject: req.Subject,
Date: req.Date,
Body: req.Body,
Junk: req.Junk,
}
if msg.Junk || (msg.Subject == "" && msg.Body == "") {
return ipc.IngestMailResp{Skipped: true}, nil
}
ctx, cancel := context.WithTimeout(ctx, m.timeout)
defer cancel()
cands, err := m.ex.Extract(ctx, msg)
if err != nil {
// The error from internal/email never carries mail text; keep it that way
// by not adding the subject here.
return ipc.IngestMailResp{}, fmt.Errorf("mail intake: uid %d: %w", req.UID, err)
}
if len(cands) == 0 {
return ipc.IngestMailResp{}, nil
}
source := email.SourcePrefix + req.Mailbox
evidence := truncateRunes(req.Subject, evidenceMaxChars)
now := m.now()
var resp ipc.IngestMailResp
for _, c := range cands {
t := store.Task{
CreatedTs: now,
Text: c.Text,
Source: source,
Evidence: evidence,
// The one status this path may ever write. Anything Maven derived from
// something she read is a suggestion until he confirms it on /tasks.
Status: store.TaskCandidate,
}
if due, ok := email.ParseDue(c.Due); ok {
t.Due = &due
}
id, created, err := m.st.CaptureTask(ctx, t)
if err != nil {
return resp, fmt.Errorf("mail intake: capture: %w", err)
}
resp.TaskIDs = append(resp.TaskIDs, id)
if created {
resp.Created++
}
}
// Counts only: the log line names the mailbox and the UID, never the subject,
// the sender or the task text. Reviewing a candidate is what /tasks is for.
log.Printf("mail intake: %s uid %d → %d candidate(s), %d new", source, req.UID, len(cands), resp.Created)
return resp, nil
}
// wireMailIntake installs the IPC hook, or leaves it nil so the method reports
// ErrUnknownMethod. Called on both startup paths (unlocked boot and passkey
// unlock) so mail behaves the same either way.
func wireMailIntake(srv *ipc.Server, st *store.Store, phr phraser.Phraser, cfg *config.Config) {
mi := newMailIntake(st, phr, cfg)
if mi == nil {
return
}
srv.IngestMailFn = mi.ingest
}
// truncateRunes cuts a string to n runes, marking the cut.
func truncateRunes(s string, n int) string {
r := []rune(s)
if len(r) <= n {
return s
}
return string(r[:n]) + "…"
}
+182
View File
@@ -0,0 +1,182 @@
package main
import (
"context"
"strings"
"testing"
"time"
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/email"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/llm"
"github.com/kami/maven/internal/store"
)
// mailLLM — a canned extraction reply.
type mailLLM struct {
reply string
calls int
}
func (m *mailLLM) Complete(_ context.Context, _ llm.Req) (string, error) {
m.calls++
return m.reply, nil
}
func newTestIntake(t *testing.T, reply string) (*mailIntake, *store.Store, *mailLLM) {
t.Helper()
st := newTestStore(t)
fake := &mailLLM{reply: reply}
return &mailIntake{
st: st,
ex: email.NewExtractor(fake, 0, nil),
timeout: 5 * time.Second,
now: func() time.Time { return time.Date(2026, 8, 1, 10, 0, 0, 0, time.UTC) },
}, st, fake
}
func ingestReq() ipc.IngestMailReq {
return ipc.IngestMailReq{
Mailbox: "INBOX", UID: 42,
From: "billing@isp.example",
Subject: "Счёт за интернет",
Body: "Оплатите счёт до 5 августа.",
}
}
// The one property that matters: a mail-derived task is a candidate, attributed
// to the mailbox, with the subject as reviewable evidence — and nothing else is
// written.
func TestIngestCapturesCandidates(t *testing.T) {
mi, st, _ := newTestIntake(t, `[{"text":"оплатить счёт за интернет","due":"2026-08-05"}]`)
resp, err := mi.ingest(context.Background(), ingestReq())
if err != nil {
t.Fatalf("ingest: %v", err)
}
if resp.Created != 1 || len(resp.TaskIDs) != 1 {
t.Fatalf("resp = %+v, want one created task", resp)
}
tasks, err := st.ListTasks(context.Background(), "")
if err != nil {
t.Fatalf("list: %v", err)
}
if len(tasks) != 1 {
t.Fatalf("got %d tasks, want 1", len(tasks))
}
got := tasks[0]
if got.Status != store.TaskCandidate {
t.Errorf("status = %q, want %q — mail may only produce candidates", got.Status, store.TaskCandidate)
}
if got.Source != "email:INBOX" {
t.Errorf("source = %q, want email:INBOX", got.Source)
}
if got.Evidence != "Счёт за интернет" {
t.Errorf("evidence = %q, want the subject line", got.Evidence)
}
if got.Due == nil || got.Due.Format("2006-01-02") != "2026-08-05" {
t.Errorf("due = %v, want 2026-08-05", got.Due)
}
// Nothing else may have been written: no reminder, no fact.
rem, err := st.ListReminders(context.Background(), 10)
if err != nil {
t.Fatalf("list reminders: %v", err)
}
if len(rem) != 0 {
t.Errorf("mail created %d reminders; a misread mail must never be able to fire", len(rem))
}
}
// Re-reading a mailbox must not grow the list — CaptureTask dedupes among live
// rows, and the intake relies on exactly that.
func TestIngestSameMailTwiceIsIdempotent(t *testing.T) {
mi, st, _ := newTestIntake(t, `[{"text":"оплатить счёт","due":""}]`)
if _, err := mi.ingest(context.Background(), ingestReq()); err != nil {
t.Fatalf("first ingest: %v", err)
}
resp, err := mi.ingest(context.Background(), ingestReq())
if err != nil {
t.Fatalf("second ingest: %v", err)
}
if resp.Created != 0 || len(resp.TaskIDs) != 1 {
t.Errorf("resp = %+v, want the existing row and Created=0", resp)
}
tasks, _ := st.ListTasks(context.Background(), "")
if len(tasks) != 1 {
t.Errorf("got %d tasks after two reads, want 1", len(tasks))
}
}
func TestIngestJunkSkipsTheModel(t *testing.T) {
mi, st, fake := newTestIntake(t, `[{"text":"купить со скидкой","due":""}]`)
req := ingestReq()
req.Junk = true
resp, err := mi.ingest(context.Background(), req)
if err != nil {
t.Fatalf("ingest: %v", err)
}
if !resp.Skipped || resp.Created != 0 {
t.Errorf("resp = %+v, want skipped", resp)
}
if fake.calls != 0 {
t.Errorf("model called %d times for junk, want 0", fake.calls)
}
if tasks, _ := st.ListTasks(context.Background(), ""); len(tasks) != 0 {
t.Errorf("junk produced %d tasks, want 0", len(tasks))
}
}
func TestIngestEmptyMessageSkipped(t *testing.T) {
mi, _, fake := newTestIntake(t, "[]")
resp, err := mi.ingest(context.Background(), ipc.IngestMailReq{Mailbox: "INBOX", UID: 1})
if err != nil || !resp.Skipped {
t.Fatalf("resp = %+v, err = %v; want skipped", resp, err)
}
if fake.calls != 0 {
t.Errorf("model called %d times for an empty message, want 0", fake.calls)
}
}
func TestIngestNoTasksWritesNothing(t *testing.T) {
mi, st, _ := newTestIntake(t, "[]")
resp, err := mi.ingest(context.Background(), ingestReq())
if err != nil {
t.Fatalf("ingest: %v", err)
}
if resp.Created != 0 || len(resp.TaskIDs) != 0 || resp.Skipped {
t.Errorf("resp = %+v, want nothing captured and not skipped", resp)
}
if tasks, _ := st.ListTasks(context.Background(), ""); len(tasks) != 0 {
t.Errorf("got %d tasks, want 0", len(tasks))
}
}
func TestIngestTruncatesEvidence(t *testing.T) {
mi, st, _ := newTestIntake(t, `[{"text":"дело","due":""}]`)
req := ingestReq()
req.Subject = strings.Repeat("щ", 400)
if _, err := mi.ingest(context.Background(), req); err != nil {
t.Fatalf("ingest: %v", err)
}
tasks, _ := st.ListTasks(context.Background(), "")
if len(tasks) != 1 {
t.Fatalf("got %d tasks, want 1", len(tasks))
}
if n := len([]rune(tasks[0].Evidence)); n > evidenceMaxChars+1 {
t.Errorf("evidence kept %d runes, want ≤ %d", n, evidenceMaxChars)
}
}
// Off unless configured: no email block ⇒ no intake, so the IPC method does not
// exist at all.
func TestNewMailIntakeOffWithoutConfig(t *testing.T) {
st := newTestStore(t)
if mi := newMailIntake(st, nil, &config.Config{}); mi != nil {
t.Error("no email block must mean no mail intake")
}
// Configured but with a non-LLM phraser: still off — there is no fallback
// extraction, by design.
if mi := newMailIntake(st, nil, &config.Config{Email: &config.EmailConfig{}}); mi != nil {
t.Error("without a llama-server phraser there is nothing to extract with")
}
}
+42
View File
@@ -170,6 +170,8 @@ func run(args []string) error {
eco *ecosystemWiring
factWorker *factEnrichmentWorker
evalWorker *memoryEvalWorker // nil ⇒ memory evaluation off (the default)
feedWkr *feedWorker // nil ⇒ no feed is read (the default)
crawlWkr *crawlWorker // nil ⇒ no page is watched (the default)
)
if !locked {
@@ -265,6 +267,8 @@ func run(args []string) error {
tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines), config.MorningRoutinesFromConfig(cfg.MorningRoutines), cfg.PatternProposals)
factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval))
evalWorker = newMemoryEvalWorker(st, phr, cfg)
feedWkr = newFeedWorker(ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
crawlWkr = newCrawlWorker(newCrawler(cfg), ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
coreAPI = &daemonAPI{
CoreAPI: ipc.NewStoreAPI(st),
@@ -320,6 +324,13 @@ func run(args []string) error {
srv.StepUp = func(ctx context.Context) error { return passkeySess.Assert(ctx, auth.Scope{}) }
// Mail ingestion (Vikunja #246): the hook stays nil unless an email block is
// configured and there is a llama-server to extract with, in which case
// ipc.MethodIngestMail reports ErrUnknownMethod.
if !locked {
wireMailIntake(srv, st, phr, cfg)
}
// WrapKeyFn — wraps the env key with a passkey credential public key and
// persists the wrapped blob. Only wired when the daemon has the key in
// memory (env key mode). Called by mavweb after passkey enrollment.
@@ -444,6 +455,8 @@ func run(args []string) error {
tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines), config.MorningRoutinesFromConfig(cfg.MorningRoutines), cfg.PatternProposals)
factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval))
evalWorker = newMemoryEvalWorker(st, phr, cfg)
feedWkr = newFeedWorker(ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
crawlWkr = newCrawlWorker(newCrawler(cfg), ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
// Swap the CoreAPI from the locked placeholder to the real store adapter.
newAPI := &daemonAPI{
@@ -457,6 +470,7 @@ func run(args []string) error {
}
srv.SetAPI(newAPI)
srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check
wireMailIntake(srv, st, phr, cfg)
// Start voice server.
if voiceW != nil {
@@ -488,6 +502,20 @@ func run(args []string) error {
}()
}
// Start feed reading (nil unless configured).
if feedWkr != nil {
go func() {
feedWkr.run(ctx)
}()
}
// Start the watched-page crawls (nil unless configured).
if crawlWkr != nil {
go func() {
crawlWkr.run(ctx)
}()
}
dl.unlock()
log.Printf("mavend: unlocked via passkey assertion")
return nil
@@ -533,6 +561,20 @@ func run(args []string) error {
evalWorker.run(ctx)
}()
}
if feedWkr != nil {
wg.Add(1)
go func() {
defer wg.Done()
feedWkr.run(ctx)
}()
}
if crawlWkr != nil {
wg.Add(1)
go func() {
defer wg.Done()
crawlWkr.run(ctx)
}()
}
}
<-ctx.Done()
+10
View File
@@ -52,6 +52,7 @@ import (
"time"
"github.com/kami/maven/internal/audio"
"github.com/kami/maven/internal/crawl"
"github.com/kami/maven/internal/dialogue"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/memory"
@@ -82,6 +83,15 @@ type reactiveHandler struct {
replier voice.Replier
now func() time.Time
// crawler reads a web page he names out loud (queryWeb). nil ⇒ on-demand
// page reading is off, which is the default: no `crawl` block, no fetch.
crawler *crawl.Crawler
// feedsOn — whether any RSS feed is configured (config.Feeds). It changes
// only what she SAYS when asked and nothing is there: "ленты не настроены"
// instead of "ничего нового", which are different truths.
feedsOn bool
weatherProvider weather.Provider
weatherLocation string // default location for weather queries
+14 -10
View File
@@ -201,16 +201,20 @@ func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI, phr phraser.Phraser, mem
// ----- the handler (the reactive path; closes over stt / tts / router / coreAPI / memory) -----
h := &reactiveHandler{
stt: transcriber,
tts: synthesizer,
router: rtr,
embedder: emb,
api: coreAPI,
tools: exec,
matcher: matcher,
replier: replier,
phraser: phr,
now: time.Now,
stt: transcriber,
tts: synthesizer,
router: rtr,
embedder: emb,
api: coreAPI,
tools: exec,
matcher: matcher,
replier: replier,
phraser: phr,
now: time.Now,
feedsOn: cfg.Feeds != nil,
// nil unless `crawl.on_demand` is on: reading a page he names is a
// capability, and capabilities are off unless configured.
crawler: onDemandCrawler(cfg),
weatherProvider: weatherProvider,
weatherLocation: weatherLocation,
memStore: memStore,
+332
View File
@@ -0,0 +1,332 @@
// mavmaild — the mail reader module (Vikunja #246,
// docs/plans/01-email-reader.md).
//
// Every so often it opens one IMAP mailbox read-only, fetches the messages it
// has not read yet, and hands each one to core over ipc.MethodIngestMail. Core
// runs the extraction on the resident model and writes what comes back as task
// CANDIDATES he reviews on /tasks. Nothing here writes to the store, nothing
// here can create a reminder, and nothing here speaks.
//
// Why a separate daemon rather than a loop inside mavend, when extraction has
// to happen in mavend anyway: the credential. mavpoll set the precedent with the
// zenmoney token (#125) — the module that talks to a third party holds the
// secret, reads it from a FILE so it never appears in `ps`, in
// docker-compose.yml or in shell history, and core never sees it. Core learns
// that mail exists only as message text on one IPC method; it cannot connect to
// the mailbox even if it wanted to, and a compromised core yields no mail
// password.
//
// Off unless configured: without -password-file there is nothing to run, and
// the daemon says so and exits. If core has no `email` block the very first
// ingest comes back ErrUnknownMethod and this daemon stops polling instead of
// hammering a socket that will keep refusing.
//
// Mail is personal, so the log is counts and UIDs: how many messages were
// fetched, how many were bulk, how many candidates came back. No subject, no
// sender, no body, ever — reviewing a candidate is what /tasks is for.
package main
import (
"context"
"encoding/json"
"errors"
"flag"
"fmt"
"log"
"os"
"os/signal"
"path/filepath"
"sort"
"strings"
"syscall"
"time"
"github.com/kami/maven/internal/email"
"github.com/kami/maven/internal/ipc"
)
func main() {
if err := run(os.Args[1:]); err != nil {
fmt.Fprintln(os.Stderr, "mavmaild:", err)
os.Exit(1)
}
}
func run(args []string) error {
fs := flag.NewFlagSet("mavmaild", flag.ContinueOnError)
socket := fs.String("socket", "", "core IPC socket path (required)")
server := fs.String("imap", "", "IMAP server, host or host:993 (required)")
user := fs.String("user", "", "IMAP username (required)")
passFile := fs.String("password-file", "", "file holding the IMAP password (required — never passed as a flag value)")
mailbox := fs.String("mailbox", "INBOX", "mailbox to read, read-only")
interval := fs.Duration("interval", 15*time.Minute, "how often to read the mailbox")
lookback := fs.Duration("lookback", 72*time.Hour, "how far back to search on each poll")
max := fs.Int("max", 25, "most messages to fetch in one poll")
timeout := fs.Duration("timeout", 30*time.Second, "IMAP network timeout")
statePath := fs.String("state", "", "file remembering which UIDs were read (default: none — every poll re-reads the window)")
if err := fs.Parse(args); err != nil {
return err
}
if *socket == "" {
return fmt.Errorf("-socket is required")
}
if *server == "" || *user == "" || *passFile == "" {
return fmt.Errorf("mail reading is off unless configured: set -imap, -user and -password-file")
}
// The password is read from a file, never taken as a flag value: an argv
// secret is visible in `ps` to every user on the box and lands in the compose
// file and the shell history. Read once at start — a rotated password means a
// restart, which is cheaper than re-reading his credential every quarter hour.
raw, err := os.ReadFile(*passFile)
if err != nil {
return fmt.Errorf("read password file: %w", err)
}
password := strings.TrimSpace(string(raw))
if password == "" {
return fmt.Errorf("password file %s is empty", *passFile)
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
core, err := ipc.DialWait(*socket, 60*time.Second)
if err != nil {
return err
}
defer core.Close()
r := &reader{
core: core,
addr: *server,
user: *user,
mailbox: *mailbox,
lookback: *lookback,
max: *max,
timeout: *timeout,
state: newSeenState(*statePath),
}
if err := r.state.load(); err != nil {
// A missing or corrupt state file must not stop mail from being read: the
// worst case is re-reading the window, and capture dedupes on text.
log.Printf("mavmaild: state: %v (starting from an empty seen-set)", err)
}
// The password is never logged, not even its length.
log.Printf("mavmaild: reading %s on %s every %s (lookback %s, max %d/poll)",
*mailbox, *server, *interval, *lookback, *max)
r.pollOnce(ctx, password) // don't idle a full interval on start
t := time.NewTicker(*interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
log.Printf("mavmaild: bye")
return nil
case <-t.C:
if r.disabled {
// Core told us mail ingestion is not configured. Nothing will change
// without a core restart, and a restart restarts us too.
log.Printf("mavmaild: core does not accept mail — idling")
return nil
}
r.pollOnce(ctx, password)
}
}
}
// mailIngester — the slice of core this daemon uses. One method: hand over a
// message. It cannot write a fact, create a reminder or read the store, and the
// interface says so.
type mailIngester interface {
IngestMail(ctx context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error)
}
type reader struct {
core mailIngester
addr string
user string
mailbox string
lookback time.Duration
max int
timeout time.Duration
state *seenState
// dial — connection seam for the tests; nil ⇒ implicit TLS.
dial func(addr string, timeout time.Duration) (*email.Conn, error)
// disabled — core answered ErrUnknownMethod, i.e. it has no email block.
disabled bool
}
// pollOnce — one read of the mailbox, then one ingest per message.
//
// A fetch error aborts this poll and nothing else; the next tick tries again.
// An ingest error for one message does not skip the rest — one mail the model
// choked on should not hide the four behind it.
func (r *reader) pollOnce(ctx context.Context, password string) {
msgs, err := r.fetch(password)
if err != nil {
// The error may name a UID; it never names a subject or a sender.
log.Printf("mavmaild: fetch: %v", err)
if len(msgs) == 0 {
return
}
}
var junk, candidates, created int
for _, m := range msgs {
if ctx.Err() != nil {
return
}
if m.Junk {
junk++
// Marked seen without a model call: the header filter already decided,
// and re-classifying it every quarter hour would be pure waste.
r.state.mark(m.UID)
continue
}
resp, err := r.core.IngestMail(ctx, ipc.IngestMailReq{
Mailbox: r.mailbox,
UID: m.UID,
From: m.From,
Subject: m.Subject,
Date: m.Date,
Body: m.Body,
})
if errors.Is(err, ipc.ErrUnknownMethod) {
log.Printf("mavmaild: core has no email block configured — mail ingestion is off; stopping")
r.disabled = true
return
}
if err != nil {
// Not marked seen: an ingest that failed should be retried next poll.
log.Printf("mavmaild: ingest uid %d: %v", m.UID, err)
continue
}
r.state.mark(m.UID)
candidates += len(resp.TaskIDs)
created += resp.Created
}
if err := r.state.save(); err != nil {
log.Printf("mavmaild: state: %v", err)
}
log.Printf("mavmaild: %s: %d read, %d bulk, %d candidate(s), %d new", r.mailbox, len(msgs), junk, candidates, created)
}
// fetch reads the mailbox. Messages already in the seen-set are not fetched at
// all, so a steady mailbox costs one SEARCH per poll and nothing else.
func (r *reader) fetch(password string) ([]email.Message, error) {
f := email.FetchSince{
Addr: r.addr,
User: r.user,
Mailbox: r.mailbox,
Timeout: r.timeout,
Since: time.Now().Add(-r.lookback),
Max: r.max,
Skip: r.state.seen,
}
return f.RunWith(password, r.dial)
}
// ---- seen state ------------------------------------------------------------
// seenState — the UIDs already handed to core, persisted so a restart does not
// re-read (and re-extract, at multi-second LLM cost) the whole lookback window.
//
// Correctness does not depend on it: ipc.CaptureTask dedupes on normalised text
// among live tasks, so a re-read produces no duplicate rows. This exists to save
// the model's time, which is why a broken state file is a log line rather than a
// failure.
//
// UIDs are per-mailbox and monotonic, so the set is kept as a high-water mark
// plus the stragglers above it. If the server ever changes UIDVALIDITY, UIDs
// reset and the window is simply re-read once — dedupe absorbs it.
type seenState struct {
path string
high uint32
set map[uint32]bool
dirty bool
}
func newSeenState(path string) *seenState {
return &seenState{path: path, set: map[uint32]bool{}}
}
type seenFile struct {
High uint32 `json:"high"`
UIDs []uint32 `json:"uids,omitempty"`
}
func (s *seenState) seen(uid uint32) bool {
return uid <= s.high || s.set[uid]
}
func (s *seenState) mark(uid uint32) {
if s.seen(uid) {
return
}
s.set[uid] = true
s.dirty = true
// Advance the high-water mark through any contiguous run, so the explicit set
// stays small on a mailbox read in order.
for {
next := s.high + 1
if !s.set[next] {
break
}
delete(s.set, next)
s.high = next
}
}
func (s *seenState) load() error {
if s.path == "" {
return nil
}
b, err := os.ReadFile(s.path)
if errors.Is(err, os.ErrNotExist) {
return nil // first run
}
if err != nil {
return err
}
var f seenFile
if err := json.Unmarshal(b, &f); err != nil {
return fmt.Errorf("parse %s: %w", s.path, err)
}
s.high = f.High
for _, u := range f.UIDs {
s.set[u] = true
}
return nil
}
// save writes the state atomically (temp file + rename), 0600: it is a list of
// message ids from his mailbox, which is metadata about his mail.
func (s *seenState) save() error {
if s.path == "" || !s.dirty {
return nil
}
uids := make([]uint32, 0, len(s.set))
for u := range s.set {
uids = append(uids, u)
}
sort.Slice(uids, func(i, j int) bool { return uids[i] < uids[j] })
b, err := json.Marshal(seenFile{High: s.high, UIDs: uids})
if err != nil {
return err
}
tmp := s.path + ".tmp"
if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil {
return err
}
if err := os.WriteFile(tmp, b, 0o600); err != nil {
return err
}
if err := os.Rename(tmp, s.path); err != nil {
return err
}
s.dirty = false
return nil
}
+254
View File
@@ -0,0 +1,254 @@
package main
import (
"bufio"
"context"
"fmt"
"net"
"os"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
"github.com/kami/maven/internal/email"
"github.com/kami/maven/internal/ipc"
)
// ---- a scripted IMAP server, same shape internal/email's tests use ---------
type fakeIMAP struct {
msgs map[uint32]string
uids []uint32
cmds []string
}
func (f *fakeIMAP) serve(c net.Conn) {
defer c.Close()
fmt.Fprint(c, "* OK fake ready\r\n")
r := bufio.NewReader(c)
for {
line, err := r.ReadString('\n')
if err != nil {
return
}
parts := strings.SplitN(strings.TrimRight(line, "\r\n"), " ", 2)
if len(parts) != 2 {
return
}
tag, cmd := parts[0], parts[1]
f.cmds = append(f.cmds, cmd)
upper := strings.ToUpper(cmd)
switch {
case strings.HasPrefix(upper, "LOGIN"), strings.HasPrefix(upper, "EXAMINE"):
fmt.Fprintf(c, "%s OK\r\n", tag)
case strings.HasPrefix(upper, "UID SEARCH"):
var ids []string
for _, u := range f.uids {
ids = append(ids, strconv.FormatUint(uint64(u), 10))
}
fmt.Fprintf(c, "* SEARCH %s\r\n%s OK\r\n", strings.Join(ids, " "), tag)
case strings.HasPrefix(upper, "UID FETCH"):
uid, _ := strconv.ParseUint(strings.Fields(cmd)[2], 10, 32)
if raw, ok := f.msgs[uint32(uid)]; ok {
fmt.Fprintf(c, "* 1 FETCH (UID %d BODY[] {%d}\r\n%s)\r\n", uid, len(raw), raw)
}
fmt.Fprintf(c, "%s OK\r\n", tag)
case strings.HasPrefix(upper, "LOGOUT"):
fmt.Fprintf(c, "* BYE\r\n%s OK\r\n", tag)
return
default:
fmt.Fprintf(c, "%s BAD\r\n", tag)
}
}
}
func (f *fakeIMAP) dial(_ string, timeout time.Duration) (*email.Conn, error) {
cli, srv := net.Pipe()
go f.serve(srv)
return email.NewConn(cli, timeout)
}
// ---- a fake core -----------------------------------------------------------
type fakeCore struct {
got []ipc.IngestMailReq
resp ipc.IngestMailResp
err error
}
func (c *fakeCore) IngestMail(_ context.Context, req ipc.IngestMailReq) (ipc.IngestMailResp, error) {
c.got = append(c.got, req)
if c.err != nil {
return ipc.IngestMailResp{}, c.err
}
return c.resp, nil
}
func mail(subject, body string, extraHeaders ...string) string {
h := "Subject: " + subject + "\r\nFrom: a@b.c\r\nContent-Type: text/plain; charset=utf-8\r\n"
for _, e := range extraHeaders {
h += e + "\r\n"
}
return h + "\r\n" + body + "\r\n"
}
func newTestReader(t *testing.T, f *fakeIMAP, core *fakeCore, statePath string) *reader {
t.Helper()
return &reader{
core: core, addr: "mail.example:993", user: "kami", mailbox: "INBOX",
lookback: 72 * time.Hour, max: 25, timeout: 5 * time.Second,
state: newSeenState(statePath),
dial: f.dial,
}
}
func TestPollHandsMessagesToCore(t *testing.T) {
f := &fakeIMAP{
uids: []uint32{1, 2},
msgs: map[uint32]string{
1: mail("Счёт", "Оплатить до 5 августа."),
2: mail("Скидки", "Sale!", "List-Unsubscribe: <mailto:u@x>"),
},
}
core := &fakeCore{resp: ipc.IngestMailResp{TaskIDs: []int64{1}, Created: 1}}
r := newTestReader(t, f, core, "")
r.pollOnce(context.Background(), "secret")
// The newsletter is filtered before core is asked: only the real mail crosses.
if len(core.got) != 1 {
t.Fatalf("core saw %d messages, want 1 (the bulk one must not cross): %+v", len(core.got), core.got)
}
got := core.got[0]
if got.UID != 1 || got.Mailbox != "INBOX" || got.Subject != "Счёт" {
t.Errorf("ingest req = %+v", got)
}
if !strings.Contains(got.Body, "Оплатить") {
t.Errorf("body = %q", got.Body)
}
}
// A second poll must not re-send what core already saw — extraction is a
// multi-second LLM call per message.
func TestPollSkipsSeenUIDs(t *testing.T) {
f := &fakeIMAP{uids: []uint32{5}, msgs: map[uint32]string{5: mail("Счёт", "текст")}}
core := &fakeCore{}
r := newTestReader(t, f, core, "")
r.pollOnce(context.Background(), "secret")
r.pollOnce(context.Background(), "secret")
if len(core.got) != 1 {
t.Errorf("core saw %d messages over two polls, want 1", len(core.got))
}
}
// An ingest that failed is NOT marked seen: the next poll retries it.
func TestPollRetriesFailedIngest(t *testing.T) {
f := &fakeIMAP{uids: []uint32{5}, msgs: map[uint32]string{5: mail("Счёт", "текст")}}
core := &fakeCore{err: fmt.Errorf("llama-server is warming up")}
r := newTestReader(t, f, core, "")
r.pollOnce(context.Background(), "secret")
core.err = nil
r.pollOnce(context.Background(), "secret")
if len(core.got) != 2 {
t.Errorf("core saw %d attempts, want 2 (a failed ingest is retried)", len(core.got))
}
}
// Core without an email block ⇒ stop, don't hammer the socket.
func TestPollStopsWhenCoreRefusesMail(t *testing.T) {
f := &fakeIMAP{uids: []uint32{1, 2}, msgs: map[uint32]string{1: mail("a", "b"), 2: mail("c", "d")}}
core := &fakeCore{err: fmt.Errorf("call: %w", ipc.ErrUnknownMethod)}
r := newTestReader(t, f, core, "")
r.pollOnce(context.Background(), "secret")
if !r.disabled {
t.Error("ErrUnknownMethod must disable the reader")
}
if len(core.got) != 1 {
t.Errorf("core saw %d messages, want 1 — stop at the first refusal", len(core.got))
}
}
func TestSeenStatePersists(t *testing.T) {
path := filepath.Join(t.TempDir(), "state", "seen.json")
f := &fakeIMAP{uids: []uint32{9}, msgs: map[uint32]string{9: mail("Счёт", "текст")}}
core := &fakeCore{}
r := newTestReader(t, f, core, path)
r.pollOnce(context.Background(), "secret")
fi, err := os.Stat(path)
if err != nil {
t.Fatalf("state file: %v", err)
}
// A list of message ids from his mailbox is metadata about his mail.
if perm := fi.Mode().Perm(); perm != 0o600 {
t.Errorf("state file mode = %v, want 0600", perm)
}
// A fresh reader with the same state file must not re-read the message.
core2 := &fakeCore{}
r2 := newTestReader(t, f, core2, path)
if err := r2.state.load(); err != nil {
t.Fatalf("load: %v", err)
}
r2.pollOnce(context.Background(), "secret")
if len(core2.got) != 0 {
t.Errorf("after a restart core saw %d messages, want 0", len(core2.got))
}
}
func TestSeenStateHighWaterMark(t *testing.T) {
s := newSeenState("")
s.mark(1)
s.mark(3)
s.mark(2)
if s.high != 3 {
t.Errorf("high = %d, want 3 (contiguous run collapses)", s.high)
}
if len(s.set) != 0 {
t.Errorf("explicit set = %v, want empty", s.set)
}
if !s.seen(2) || s.seen(4) {
t.Errorf("seen(2)=%v seen(4)=%v", s.seen(2), s.seen(4))
}
}
func TestSeenStateCorruptFileIsNotFatal(t *testing.T) {
path := filepath.Join(t.TempDir(), "seen.json")
if err := os.WriteFile(path, []byte("{not json"), 0o600); err != nil {
t.Fatal(err)
}
s := newSeenState(path)
if err := s.load(); err == nil {
t.Error("a corrupt state file should report an error the caller logs")
}
if s.seen(1) {
t.Error("a corrupt state file must leave an empty seen-set, not a poisoned one")
}
}
// Off unless configured, and the credential is never a flag value.
func TestRunRequiresConfig(t *testing.T) {
if err := run([]string{}); err == nil {
t.Error("no -socket must be an error")
}
if err := run([]string{"-socket", "/tmp/nope.sock"}); err == nil {
t.Error("no mailbox configuration must be an error, not a default mailbox")
}
// There is no -password flag at all: only -password-file.
if err := run([]string{"-socket", "/x", "-imap", "h", "-user", "u", "-password", "p"}); err == nil ||
!strings.Contains(err.Error(), "flag provided but not defined") {
t.Errorf("a -password flag must not exist; err = %v", err)
}
}
func TestRunRejectsEmptyPasswordFile(t *testing.T) {
path := filepath.Join(t.TempDir(), "pass")
if err := os.WriteFile(path, []byte(" \n"), 0o600); err != nil {
t.Fatal(err)
}
err := run([]string{"-socket", "/x/y.sock", "-imap", "h", "-user", "u", "-password-file", path})
if err == nil || !strings.Contains(err.Error(), "empty") {
t.Errorf("an empty password file must be refused before dialling; err = %v", err)
}
}
+61
View File
@@ -36,6 +36,67 @@ present (see `.dockerignore`).
| `/var/lib/maven` (volume) | encrypted db at rest |
| `/dev/shm` (tmpfs) | decrypted db working copy (RAM only) |
## Reading the outside world (off by default)
`mavend.json` ships without a `feeds` block, which means no RSS/Atom feed is
fetched and no outbound request is made. Switching it on is adding the block:
```json
"feeds": {
"poll_interval": "30m",
"max_items": 5,
"max_age": "24h",
"sources": [
{ "name": "habr", "url": "https://habr.com/ru/rss/best/daily/",
"category": "технологии", "exclude": ["реклама"] }
]
}
```
What it does and does not do:
- items are written as notes with source `rss:<name>`, visible on `/dash`;
- **nothing is announced.** She reads them back when asked — "что нового в
лентах?", "что нового по технологиям?" — and never on arrival. There is no
severity or channel knob here on purpose;
- the fetcher is allowlisted to the hosts of the configured feeds, plus any
`allow_hosts`. It refuses non-http(s) schemes and every private address
(loopback, the LAN, the `10.42.0.0/24` wg range, cloud metadata). It caps the
response at 2 MiB and redirects at 3, and makes at most one request per host
per second. See `internal/webfetch`;
- how far each feed was read is stored as a config fact `rss:latest:<name>`, so
a restart does not re-note yesterday's headlines.
### Reading a page (`crawl`, also off by default)
There is no `crawl` block either, so no page is fetched. Two halves, separately
switched:
```json
"crawl": {
"on_demand": true,
"interval": "6h",
"max_runes": 4000,
"watches": [
{ "name": "changelog", "url": "https://example.org/changelog", "interval": "12h" }
]
}
```
- `on_demand` lets her read a page he names in the utterance: "посмотри
https://example.org/x — что там?". The page becomes context for his question,
and only the URL leaves the box. Without a URL nothing is fetched, so this is
a fallback and not a habit;
- `watches` re-reads a fixed list on its interval and writes a note when the
text changed. Like the feeds, it announces nothing;
- the answer path sits **last** in the query chain, behind his memory, his notes
and (once wired) the local Kiwix ZIMs. A local read costs nothing;
- `robots.txt` is fetched first and obeyed with no override; a `Disallow` is a
refusal she says out loud. `Crawl-delay` is honoured;
- same guarded fetcher as the feeds: allowlist/denylist, no private addresses,
size cap, redirect cap, timeout, one request per host per second;
- dedup state is the config fact `crawl:hash:<name>`.
## Not yet verified / host-dependent
This stack is correct-by-construction but has **not been build-tested here**
+25
View File
@@ -113,6 +113,31 @@ services:
- sockets:/run/maven
# - ./deploy/zenmoney.token:/run/secrets/zenmoney.token:ro
# The mail reader (Vikunja #246) is OFF and commented out: it needs an IMAP
# account, and there is none on this box. mavmaild reads the password from a
# FILE so it never appears in `ps`, in this file, or in shell history — the
# same rule mavpoll follows for the zenmoney token. Core never sees the
# password: the reader hands core message text on one IPC method, and core
# writes what the model extracts as task CANDIDATES he reviews on /tasks.
# Nothing here can create a reminder, so a misread mail cannot fire.
#
# To enable: write the password to deploy/imap.password (0600, gitignored),
# add an "email": {} block to deploy/mavend.json, and uncomment this service.
# mavmaild:
# <<: *image
# command: ["mavmaild", "-socket", "/run/maven/mavend.sock",
# "-imap", "imap.example.org:993",
# "-user", "kami@example.org",
# "-password-file", "/run/secrets/imap.password",
# "-mailbox", "INBOX",
# "-interval", "15m",
# "-state", "/var/lib/maven/mail-seen.json"]
# depends_on: [mavend]
# volumes:
# - sockets:/run/maven
# - dbdata:/var/lib/maven
# - ./deploy/imap.password:/run/secrets/imap.password:ro
volumes:
dbdata:
sockets:
+25
View File
@@ -27,3 +27,28 @@
7. Add voice query handler — `"что нового?"` queries `RecentNotes` filtered by source prefix `rss:` and phrases via `phraser.PhraseQuery`
8. Add `feeds` block to `config.Config` and `deploy/mavend.json`
9. Test with a live RSS feed (e.g., `https://news.ycombinator.com/rss`) — verify items appear in notes table
---
## Shipped 2026-08-01 (#258)
`internal/webfetch` (the guarded HTTP door: scheme, allow/deny hosts, private-address
refusal in the dialer, size cap, redirect cap, per-host rate limit), `internal/rss`
(RSS 2.0 + Atom parser, poller with durable marks and a keyword filter),
`cmd/mavend/feeds.go` (ticker, fetcher adapter, `rss:latest:<feed>` fact marks),
config block `feeds`, and the `feeds` query source with `router.ParseFeedQuery`.
Deviations from the plan above, both deliberate:
- **Step 5 (breaking-news nudges) was not built.** A feed that dispatches is a nag,
and the one thing Maven is not is a nag. Items are read when asked and nowhere else.
If breaking news is ever wanted, it belongs behind the existing delivery policy
(severity, quiet hours, digest), not in the poller.
- **Step 3 (embedder relevance) is a seam, not an implementation.** `rss.Ranker`
exists and is wired nil. Scoring items against an "interest profile" needs a
profile, and there is none yet; a threshold with nothing to compare against is a
random filter with a confident name. The filter that runs is the per-feed
include/exclude keyword list, which he can read and predict.
No new dependency: stdlib `encoding/xml`, no gofeed. Stock deploy config has no
`feeds` block, so the capability is off.
+38
View File
@@ -28,3 +28,41 @@
7. Add IPC methods `MethodTriggerCrawl(name)`, `MethodListCrawls`, `MethodGetCrawlResult(name)`
8. Add `crawls` block to `config.Config` and `deploy/mavend.json`
9. Test with a static HTML page — verify extraction matches expected values, verify scheduling fires correctly
## Shipped 2026-08-01 (#259)
Built as `internal/crawl` (pure: robots, extraction, watcher) plus
`cmd/mavend/crawls.go` (fetcher, ticker, dedup facts), on top of the guarded
`internal/webfetch` door added with the feed reader (#258). Off unless
configured, in two separately-switched halves: `crawl.on_demand` for a URL he
names, `crawl.watches` for a scheduled re-read.
**Limits are code, not documentation** (`internal/webfetch`, tested one test per
limit): host allowlist/denylist, no private addresses (loopback, RFC1918 —
hence the LAN and the `10.42.0.0/24` wg range —, link-local incl. cloud
metadata, CGNAT, v6 ULA) enforced in the dialer's `Control` hook so DNS
rebinding and every redirect hop are covered, response size cap, redirect cap,
timeout, one request per host per second. `robots.txt` is fetched first, cached
per host, and a `Disallow` is refused with no override.
Deliberate deviations from the plan above:
- **No CSS selectors and no LLM structured extraction** (steps 2). The output is
plaintext handed to the phraser as context for the question he asked. A 1.7B
extracting a JSON price table from 4000 runes is a worse bet than reading, and
`goquery` is not vendored.
- **No `crawl` act verb and no new IPC methods** (steps 5, 7). Reading a page is
a query source (`queryWeb` in `actions_query.go`, last in the chain, behind
Kiwix once that is wired), not an action he commands. Nothing needs a new wire
method to work.
- **Notes, not facts.** A page's text is not a fact about him. Only the dedup
hash is a fact (`crawl:hash:<name>`, kind `config`, source `poll:crawl`).
- **Nothing is dispatched.** A changed page writes a note; it does not nudge.
Not a nag.
- **No `/tools` crawl history page.** The notes and the hash facts are already
visible on `/dash`.
**No new dependency.** The vendored tree has no `x/net/html`, no `goquery` and
no `temoto/robotstxt`, so robots parsing and HTML-to-text are stdlib
(`regexp`, `html`) — RE2 has no backreferences, hence the `pairsRE` builder in
`extract.go`.
+6 -1
View File
@@ -74,7 +74,12 @@ func Requirement(m ipc.Method) Authority {
// existing analogue.
ipc.MethodCaptureTask,
ipc.MethodListTasks,
ipc.MethodSetTaskStatus:
ipc.MethodSetTaskStatus,
// Mail ingestion (Vikunja #246). AuthRead because of what the method can
// produce: candidate tasks and nothing else. It cannot write a fact, set a
// reminder, or touch the tool allowlist, so a compromised mail reader can
// at worst put junk on a review page he clears in one click.
ipc.MethodIngestMail:
return AuthRead
}
// Unknown method ⇒ AuthRead, but ipc.dispatch returns ErrUnknownMethod
+156
View File
@@ -150,6 +150,21 @@ type Config struct {
// absent ⇒ no evaluation loop at all. See MemoryEvalConfig.
MemoryEval *MemoryEvalConfig `json:"memory_eval,omitempty"`
// Email — mail ingestion (Vikunja #246). nil / absent ⇒ core refuses
// ipc.MethodIngestMail outright, so a mail reader cannot make Maven read a
// mailbox by merely existing. See EmailConfig; the IMAP host and credential
// live in the reader (cmd/mavmaild), never here.
Email *EmailConfig `json:"email,omitempty"`
// Feeds — RSS/Atom feed reading (Vikunja #258). nil / absent ⇒ no feed is
// ever fetched: reading the outside world is off unless configured, like
// the weather and telegram. See FeedsConfig.
Feeds *FeedsConfig `json:"feeds,omitempty"`
// Crawl — reading a web page (Vikunja #259). nil / absent ⇒ Maven never
// fetches a page: not on request, not on a schedule. See CrawlConfig.
Crawl *CrawlConfig `json:"crawl,omitempty"`
// Praxis — the ecosystem attention-state service. When configured, maven
// calls the Praxis HTTP tools API for attention listing and item lifecycle.
// Maven never touches Praxis's database directly (ecosystem invariant: no
@@ -396,6 +411,109 @@ func (p *PatternProposalConfig) AnnounceProposals() bool {
return p != nil && p.Notify
}
// FeedsConfig — the RSS/Atom reader (Vikunja #258, docs/plans/13-rss-news-feeds.md).
//
// Absent ⇒ off, and off means no outbound request at all. Present with an empty
// `sources` list is also off — a poller with nothing to poll is not wired.
//
// What a feed may NOT do here: speak. Items are written as notes with source
// "rss:<name>" and read back when he asks; nothing is dispatched, nudged or
// announced on arrival. That is the "not a nag" constraint, and it is why there
// is no severity or channel field in this block to reach for.
type FeedsConfig struct {
// Sources — the feeds to read. Empty ⇒ the reader stays down.
Sources []FeedSourceConfig `json:"sources,omitempty"`
// PollInterval — default per-feed cadence. 0 ⇒ rss.DefaultPollInterval (30m).
PollInterval Duration `json:"poll_interval,omitempty"`
// MaxItems — most items kept from one feed in one poll. 0 ⇒
// rss.DefaultMaxItems (5). This is the "не завали мне /dash" knob.
MaxItems int `json:"max_items,omitempty"`
// MaxAge — on a first poll (no saved mark), how far back to take items.
// 0 ⇒ rss.DefaultMaxAge (24h), so switching a feed on imports today, not
// the archive.
MaxAge Duration `json:"max_age,omitempty"`
// AllowHosts — when set, the reader may only connect to these hosts (and
// their subdomains). The feed URLs' own hosts are added automatically, so
// this is only needed to be stricter than that.
AllowHosts []string `json:"allow_hosts,omitempty"`
// Timeout — per-request budget. 0 ⇒ webfetch.DefaultTimeout.
Timeout Duration `json:"timeout,omitempty"`
// MaxBytes — response size cap. 0 ⇒ webfetch.DefaultMaxBytes (2 MiB).
MaxBytes int64 `json:"max_bytes,omitempty"`
}
// FeedSourceConfig — one feed.
type FeedSourceConfig struct {
Name string `json:"name"` // note source is "rss:<name>"
URL string `json:"url"` // http(s) only
Category string `json:"category,omitempty"` // "технологии" — what "что нового по X?" matches
Interval Duration `json:"interval,omitempty"` // 0 ⇒ FeedsConfig.PollInterval
Include []string `json:"include,omitempty"` // keep only items containing one of these
Exclude []string `json:"exclude,omitempty"` // drop items containing any of these
}
// CrawlConfig — the web crawler (Vikunja #259, docs/plans/14-web-crawler.md).
//
// Absent ⇒ off, and off means no page is ever fetched. Present with neither
// `on_demand` nor a `watches` entry is also off: there would be nothing to do.
//
// The crawler is the LAST place an answer is looked for, behind the model, his
// own memory and the local Kiwix ZIMs. That ordering lives in the query-source
// chain (cmd/mavend/actions_query.go), not here, but it is the reason this block
// is small: it is a fallback, not a search engine.
//
// Only the URL leaves the box. His notes, facts, persona block and history are
// never part of a request — the crawler package cannot even read the store.
type CrawlConfig struct {
// OnDemand — may he ask her to read a page he names out loud
// ("посмотри https://… — что там пишут?"). false ⇒ the on-demand answer
// source stays off and only the watches below run.
OnDemand bool `json:"on_demand,omitempty"`
// Watches — pages re-read on a schedule. A page whose text changed is
// written as a note (source "crawl:<name>"); nothing is announced.
Watches []CrawlWatchConfig `json:"watches,omitempty"`
// Interval — default watch cadence. 0 ⇒ crawl.DefaultWatchInterval (6h).
Interval Duration `json:"interval,omitempty"`
// AllowHosts — when set, the ONLY hosts the crawler may reach (subdomains
// included). Watched pages' own hosts are added automatically. Setting this
// is how "she may read the arch wiki and nothing else" is expressed.
AllowHosts []string `json:"allow_hosts,omitempty"`
// DenyHosts — never reachable, checked first. Private addresses do not need
// to be listed: they are refused unconditionally (see internal/webfetch).
DenyHosts []string `json:"deny_hosts,omitempty"`
// UserAgent — sent on every request AND matched against robots.txt groups.
// Empty ⇒ webfetch.DefaultUserAgent.
UserAgent string `json:"user_agent,omitempty"`
// Timeout — per-request budget. 0 ⇒ webfetch.DefaultTimeout.
Timeout Duration `json:"timeout,omitempty"`
// MaxBytes — response size cap. 0 ⇒ webfetch.DefaultMaxBytes (2 MiB).
MaxBytes int64 `json:"max_bytes,omitempty"`
// MaxRunes — how much extracted text is kept. 0 ⇒ crawl.DefaultMaxRunes
// (4000), which is what fits a 4096-token context alongside a prompt.
MaxRunes int `json:"max_runes,omitempty"`
}
// CrawlWatchConfig — one page kept an eye on.
type CrawlWatchConfig struct {
Name string `json:"name"` // note source is "crawl:<name>"
URL string `json:"url"`
Interval Duration `json:"interval,omitempty"` // 0 ⇒ CrawlConfig.Interval
}
// MemoryEvalConfig — the background memory-evaluation loop (Vikunja #248).
// Absent ⇒ off, like every other capability that costs something the owner did
// not ask for. Each evaluation is a full LLM round-trip on the one resident
@@ -418,6 +536,26 @@ type MemoryEvalConfig struct {
MinConfidence float64 `json:"min_confidence,omitempty"`
}
// EmailConfig — core's half of the email reader: how many task candidates one
// message may produce, and how long the extraction call may take.
//
// There is deliberately nothing about a mailbox here. Core does not connect to
// IMAP, does not know an account exists, and holds no mail credential — the
// reader daemon does, the same split mavpoll uses for the zenmoney token. This
// block only says "extraction is allowed, with these bounds".
type EmailConfig struct {
// MaxTasks — candidates per message. 0 ⇒ email.MaxCandidates (3).
MaxTasks int `json:"max_tasks,omitempty"`
// Timeout — per-message extraction budget. 0 ⇒ DefaultEmailTimeout. This is
// a Thinking model reading a mail; nobody is waiting on the answer, but a
// hung llama-server must not pin the reader's connection forever.
Timeout Duration `json:"timeout,omitempty"`
}
// DefaultEmailTimeout — extraction budget per message.
const DefaultEmailTimeout = 2 * time.Minute
// PhraserConfig — the LLM-backed phraser seam. The daemon spawns llama-server
// as a managed subprocess and sends chat-completion requests to phrase nudge
// and reminder messages. nil ⇒ the template-based Stub is used instead.
@@ -607,6 +745,24 @@ func (c *Config) applyDefaults() {
c.MemoryEval.Interval = Duration(DefaultMemoryEvalInterval)
}
// Same rule again: absent stays nil (⇒ mail ingestion refused), present gets
// the timeout default so `{}` is a valid "on with the defaults".
if c.Email != nil && c.Email.Timeout <= 0 {
c.Email.Timeout = Duration(DefaultEmailTimeout)
}
// A feeds block with no sources is the same as no block: nothing to poll,
// nothing wired. Normalising it to nil keeps that "off" in one place.
if c.Feeds != nil && len(c.Feeds.Sources) == 0 {
c.Feeds = nil
}
// Same rule for the crawler: a block that neither answers on demand nor
// watches anything has nothing to do, so it is normalised to "off".
if c.Crawl != nil && !c.Crawl.OnDemand && len(c.Crawl.Watches) == 0 {
c.Crawl = nil
}
if c.Voice != nil {
if c.Voice.RouterThreshold <= 0 {
c.Voice.RouterThreshold = DefaultRouterThreshold
+174
View File
@@ -0,0 +1,174 @@
// Package crawl reads a web page: fetch, robots check, HTML to text.
//
// It is the LAST place Maven looks for an answer, and that ordering is the whole
// design. "Never phones home" is deprecated, but what replaced it puts local
// sources first: the resident model, then his own memory, then the Kiwix ZIMs on
// the box (internal/kiwix), and only then the network. A local read costs
// nothing and leaks nothing; a fetch costs a round-trip and puts a URL in
// someone's access log. So this package exists to be the fallback, not the
// front door — see the querySources chain in cmd/mavend/actions_query.go for
// where it actually sits.
//
// What never leaves the box: his notes, his facts, the persona block, the
// conversation history. Only the URL is requested and, for the on-demand path,
// only because he said it out loud. Nothing here reads the store.
//
// The limits are not in this package — they are in internal/webfetch, which is
// the only way anything here touches a socket: http(s) only, host allow/deny,
// private-address refusal, size cap, redirect cap, per-host rate limit. What
// this package adds is politeness (robots.txt) and dedup.
package crawl
import (
"context"
"crypto/sha256"
"encoding/hex"
"errors"
"fmt"
"net/url"
"strings"
"time"
)
// Errors callers distinguish.
var (
ErrRobots = errors.New("crawl: robots.txt disallows this path")
ErrNotHTML = errors.New("crawl: response is not html or text")
)
// Fetcher is the guarded HTTP door (internal/webfetch adapted by the daemon). An
// interface so this package constructs no http.Client of its own and can be
// tested without a network.
type Fetcher interface {
Get(ctx context.Context, url string) (*Response, error)
}
// Response is the minimum a crawl needs from a fetch.
type Response struct {
URL string
ContentType string
Body []byte
}
// Config — crawler knobs.
type Config struct {
// UserAgent is the name matched against robots.txt groups. It must be the
// same string the fetcher sends, or Maven would be claiming one identity
// and obeying the rules for another.
UserAgent string
// MaxRunes caps extracted text. 0 ⇒ DefaultMaxRunes.
MaxRunes int
// RobotsTTL — how long a parsed robots.txt is trusted. 0 ⇒ 1h.
RobotsTTL time.Duration
// Now is injectable for tests. nil ⇒ time.Now.
Now func() time.Time
}
// Crawler fetches and extracts pages. Safe for concurrent use.
type Crawler struct {
fetch Fetcher
cfg Config
robots *robotsCache
}
// New builds a crawler. Returns nil when there is no fetcher, which is how the
// daemon expresses "crawling is off unless configured".
func New(fetch Fetcher, cfg Config) *Crawler {
if fetch == nil {
return nil
}
if cfg.UserAgent == "" {
cfg.UserAgent = "Maven"
}
if cfg.MaxRunes <= 0 {
cfg.MaxRunes = DefaultMaxRunes
}
if cfg.RobotsTTL <= 0 {
cfg.RobotsTTL = time.Hour
}
if cfg.Now == nil {
cfg.Now = time.Now
}
return &Crawler{fetch: fetch, cfg: cfg, robots: newRobotsCache(cfg.RobotsTTL)}
}
// Page fetches rawURL and returns its text. It checks robots.txt first and
// refuses a disallowed path with ErrRobots — there is no override.
func (c *Crawler) Page(ctx context.Context, rawURL string) (Page, error) {
u, err := url.Parse(strings.TrimSpace(rawURL))
if err != nil {
return Page{}, fmt.Errorf("crawl: bad url %q: %w", rawURL, err)
}
ok, err := c.allowed(ctx, u)
if err != nil {
return Page{}, err
}
if !ok {
return Page{}, fmt.Errorf("%w: %s", ErrRobots, u.Path)
}
resp, err := c.fetch.Get(ctx, u.String())
if err != nil {
return Page{}, err
}
// A PDF or an image is bytes Maven cannot read; saying so beats storing
// binary garbage as a "note".
ct := strings.ToLower(resp.ContentType)
if ct != "" && !strings.Contains(ct, "html") && !strings.Contains(ct, "text/") &&
!strings.Contains(ct, "xml") && !strings.Contains(ct, "json") {
return Page{}, fmt.Errorf("%w: %s", ErrNotHTML, resp.ContentType)
}
return Extract(resp.URL, resp.Body, c.cfg.MaxRunes), nil
}
// allowed consults robots.txt for u's host, reading it at most once per TTL.
//
// A robots.txt that cannot be fetched (404, a timeout, a blocked host) means
// allow, per the standard. The one thing that is NOT fail-open is an explicit
// Disallow.
func (c *Crawler) allowed(ctx context.Context, u *url.URL) (bool, error) {
host := u.Host
now := c.cfg.Now()
rules, ok := c.robots.get(host, now)
if !ok {
robotsURL := u.Scheme + "://" + host + "/robots.txt"
resp, err := c.fetch.Get(ctx, robotsURL)
switch {
case err != nil:
// Note what is NOT swallowed: a refusal from the guarded fetcher.
// If webfetch says this host is denied or private, the page fetch
// would fail the same way, and reporting the real reason beats
// reporting a robots verdict we never got.
if isFatalFetchError(err) {
return false, err
}
rules = Rules{}
default:
rules = ParseRobots(string(resp.Body), c.cfg.UserAgent)
}
c.robots.put(host, rules, now)
}
path := u.EscapedPath()
if u.RawQuery != "" {
path += "?" + u.RawQuery
}
return rules.Allowed(path), nil
}
// isFatalFetchError — a fetch failure that means "this host is off limits"
// rather than "there is no robots.txt here". The sentinel set is webfetch's, but
// this package must not import it (the interface exists precisely so it does
// not), so the check is on the message. Ugly and honest: the alternative is a
// dependency inversion for two strings.
func isFatalFetchError(err error) bool {
s := err.Error()
return strings.Contains(s, "not allowed") || strings.Contains(s, "private address") ||
strings.Contains(s, "only http and https")
}
// Hash is the dedup key for a crawl result: the sha256 of the extracted text,
// hex, first 16 chars. Text and not raw HTML, because a page whose only change
// is a rotating ad slot or a CSRF token has not changed.
func Hash(text string) string {
sum := sha256.Sum256([]byte(strings.TrimSpace(text)))
return hex.EncodeToString(sum[:])[:16]
}
+155
View File
@@ -0,0 +1,155 @@
package crawl
import (
"context"
"errors"
"strings"
"testing"
"time"
)
// fakeFetcher serves canned pages by URL and counts requests, so a test can
// assert that robots.txt was read once and that a refusal never reached the page.
type fakeFetcher struct {
pages map[string]Response
err error
calls []string
}
func (f *fakeFetcher) Get(_ context.Context, u string) (*Response, error) {
f.calls = append(f.calls, u)
if f.err != nil {
return nil, f.err
}
r, ok := f.pages[u]
if !ok {
return nil, errors.New("http 404")
}
if r.URL == "" {
r.URL = u
}
if r.ContentType == "" {
r.ContentType = "text/html; charset=utf-8"
}
return &r, nil
}
const htmlPage = `<html><head><title>Почему небо синее</title>
<style>body{color:red}</style><script>track()</script></head>
<body><nav>меню</nav><h1>Небо</h1>
<p>Свет рассеивается на молекулах воздуха.</p>
<p>Короткие волны рассеиваются сильнее.</p>
<footer>© 2026</footer></body></html>`
func newTestCrawler(f *fakeFetcher) *Crawler {
return New(f, Config{UserAgent: "Maven/1.0", Now: func() time.Time { return time.Unix(0, 0) }})
}
func TestPageExtractsText(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{
"https://example.org/sky": {Body: []byte(htmlPage)},
}}
page, err := newTestCrawler(f).Page(context.Background(), "https://example.org/sky")
if err != nil {
t.Fatal(err)
}
if page.Title != "Почему небо синее" {
t.Errorf("title = %q", page.Title)
}
if !strings.Contains(page.Text, "Свет рассеивается") {
t.Errorf("body text missing: %q", page.Text)
}
for _, junk := range []string{"track()", "color:red", "меню", "© 2026"} {
if strings.Contains(page.Text, junk) {
t.Errorf("%q survived extraction: %q", junk, page.Text)
}
}
}
func TestRobotsIsCheckedAndObeyed(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{
"https://example.org/robots.txt": {Body: []byte("User-agent: *\nDisallow: /secret\n"), ContentType: "text/plain"},
"https://example.org/secret/x": {Body: []byte(htmlPage)},
"https://example.org/open": {Body: []byte(htmlPage)},
}}
c := newTestCrawler(f)
if _, err := c.Page(context.Background(), "https://example.org/secret/x"); !errors.Is(err, ErrRobots) {
t.Fatalf("error = %v, want ErrRobots", err)
}
for _, u := range f.calls {
if strings.Contains(u, "/secret") {
t.Fatal("the disallowed page was fetched anyway")
}
}
if _, err := c.Page(context.Background(), "https://example.org/open"); err != nil {
t.Fatalf("allowed page: %v", err)
}
// robots.txt was read once for the host, not once per page.
robotsReads := 0
for _, u := range f.calls {
if strings.HasSuffix(u, "/robots.txt") {
robotsReads++
}
}
if robotsReads != 1 {
t.Fatalf("robots.txt read %d times, want 1", robotsReads)
}
}
// No robots.txt means allow — that is the standard, and the alternative makes
// most of the web unreadable.
func TestMissingRobotsAllows(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{
"https://example.org/page": {Body: []byte(htmlPage)},
}}
if _, err := newTestCrawler(f).Page(context.Background(), "https://example.org/page"); err != nil {
t.Fatalf("err = %v, want the page", err)
}
}
// A refusal from the guarded fetcher must surface as itself, not be laundered
// into "no robots.txt, go ahead".
func TestFetcherRefusalIsNotSwallowed(t *testing.T) {
f := &fakeFetcher{err: errors.New("webfetch: refusing to connect to a private address: 127.0.0.1")}
_, err := newTestCrawler(f).Page(context.Background(), "http://127.0.0.1:9100/mcp")
if err == nil || !strings.Contains(err.Error(), "private address") {
t.Fatalf("error = %v, want the fetcher's refusal", err)
}
}
func TestNonTextIsRefused(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{
"https://example.org/f.pdf": {Body: []byte("%PDF-1.7"), ContentType: "application/pdf"},
}}
if _, err := newTestCrawler(f).Page(context.Background(), "https://example.org/f.pdf"); !errors.Is(err, ErrNotHTML) {
t.Fatalf("error = %v, want ErrNotHTML", err)
}
}
func TestMaxRunesCapsText(t *testing.T) {
long := "<html><body><p>" + strings.Repeat("привет ", 2000) + "</p></body></html>"
f := &fakeFetcher{pages: map[string]Response{"https://example.org/l": {Body: []byte(long)}}}
c := New(f, Config{MaxRunes: 50})
page, err := c.Page(context.Background(), "https://example.org/l")
if err != nil {
t.Fatal(err)
}
if n := len([]rune(page.Text)); n > 51 {
t.Fatalf("text = %d runes, want the 50-rune cap", n)
}
}
func TestNewWithoutFetcherIsNil(t *testing.T) {
if New(nil, Config{}) != nil {
t.Fatal("a crawler with no fetcher must be nil — crawling is off unless configured")
}
}
func TestHashIgnoresNothingButText(t *testing.T) {
if Hash("a") == Hash("b") {
t.Fatal("different text hashed the same")
}
if Hash(" same \n") != Hash("same") {
t.Fatal("surrounding whitespace changed the hash")
}
}
+106
View File
@@ -0,0 +1,106 @@
package crawl
import (
"html"
"regexp"
"strings"
)
// HTML → text, with a regexp and no tokenizer.
//
// golang.org/x/net/html is not vendored and the network is not assumed, so this
// is stdlib. That is less of a compromise than it sounds: the unit of context
// here is a few hundred words for a 4096-token model to read, exactly like the
// Kiwix snippet, so what matters is dropping script/style/nav noise and keeping
// paragraph boundaries. A DOM would buy correctness on malformed markup that is
// then thrown away by truncation anyway.
//
// What this deliberately does NOT do: run JavaScript, follow links, or extract
// structured fields with CSS selectors or an LLM prompt. The plan's step 2 asked
// for the last of those; see docs/plans/14-web-crawler.md for why it was left
// out for now.
var (
// RE2 has no backreferences, so each tag pair is spelled out rather than
// captured and matched against itself.
dropRE = regexp.MustCompile(pairsRE("script", "style", "noscript", "svg", "head", "nav", "footer", "form"))
titleRE = regexp.MustCompile(`(?is)<title\b[^>]*>(.*?)</title>`)
h1RE = regexp.MustCompile(`(?is)<h1\b[^>]*>(.*?)</h1>`)
// Block-level tags become newlines so paragraphs survive as paragraphs.
blockRE = regexp.MustCompile(`(?is)</?(p|div|br|li|tr|h[1-6]|section|article|blockquote|pre)\b[^>]*>`)
tagRE = regexp.MustCompile(`(?s)<[^>]*>`)
commentRE = regexp.MustCompile(`(?s)<!--.*?-->`)
spaceRE = regexp.MustCompile(`[ \t\f\v]+`)
blankRE = regexp.MustCompile(`\n{2,}`)
)
// pairsRE builds `(?is)<tag …>…</tag>|…` for the given tags.
func pairsRE(tags ...string) string {
parts := make([]string, 0, len(tags))
for _, t := range tags {
parts = append(parts, `<`+t+`\b[^>]*>.*?</`+t+`>`)
}
return `(?is)` + strings.Join(parts, "|")
}
// Page is an extracted page.
type Page struct {
URL string
Title string
Text string // plain text, paragraphs separated by single newlines
}
// Extract turns a fetched HTML document into a Page. maxRunes caps the text (0 ⇒
// DefaultMaxRunes); the cap is on runes, not bytes, because a Russian page cut
// at a byte boundary ends in half a letter.
func Extract(url string, body []byte, maxRunes int) Page {
if maxRunes <= 0 {
maxRunes = DefaultMaxRunes
}
s := string(body)
s = commentRE.ReplaceAllString(s, " ")
title := firstGroup(titleRE, s)
if title == "" {
title = firstGroup(h1RE, s)
}
s = dropRE.ReplaceAllString(s, "\n")
s = blockRE.ReplaceAllString(s, "\n")
s = tagRE.ReplaceAllString(s, " ")
s = html.UnescapeString(s)
s = spaceRE.ReplaceAllString(s, " ")
var lines []string
for _, l := range strings.Split(s, "\n") {
if l = strings.TrimSpace(l); l != "" {
lines = append(lines, l)
}
}
text := blankRE.ReplaceAllString(strings.Join(lines, "\n"), "\n")
return Page{URL: url, Title: title, Text: TrimRunes(text, maxRunes)}
}
// DefaultMaxRunes — how much of a page is kept. ~4000 runes is a long answer's
// worth of context and still leaves room in a 4096-token window for the prompt
// and the reply.
const DefaultMaxRunes = 4000
func firstGroup(re *regexp.Regexp, s string) string {
m := re.FindStringSubmatch(s)
if len(m) < 2 {
return ""
}
t := tagRE.ReplaceAllString(m[1], " ")
return strings.TrimSpace(strings.Join(strings.Fields(html.UnescapeString(t)), " "))
}
// TrimRunes cuts s to at most max runes, on a rune boundary.
func TrimRunes(s string, max int) string {
r := []rune(s)
if len(r) <= max {
return s
}
return strings.TrimSpace(string(r[:max])) + "…"
}
+211
View File
@@ -0,0 +1,211 @@
package crawl
import (
"regexp"
"strings"
"sync"
"time"
)
// robots.txt, parsed the small way: no wildcards beyond the two the standard
// actually defines (`*` inside a path and `$` at the end), no sitemaps, no
// crawl-delay-per-agent gymnastics. A personal assistant reading a handful of
// pages does not need a spec-complete implementation; it needs to not be rude,
// and to be auditable in one sitting.
//
// Two rules worth stating because they are choices, not accidents:
//
// - a missing or unreadable robots.txt means ALLOW. That is what the standard
// says (404 ⇒ unrestricted), and the alternative would make a site that
// simply has no robots.txt unreadable;
// - an explicit Disallow means REFUSE, and Maven does not offer an override.
// There is no "but he asked me to" flag: the page is not read.
// Rules is a parsed robots.txt for one user-agent.
type Rules struct {
allow []string
disallow []string
// Delay is Crawl-delay in seconds when the group named one, 0 otherwise.
// The fetcher's own per-host rate limit is the floor; this can only make
// Maven slower, never faster.
Delay time.Duration
}
// ParseRobots reads robots.txt and returns the rules that apply to agent.
//
// Group selection follows the standard: the most specific matching group wins,
// which here means an exact user-agent match beats `*`. Lines that are neither
// are ignored rather than guessed at.
func ParseRobots(body string, agent string) Rules {
agent = strings.ToLower(agent)
type group struct {
agents []string
allow []string
disallow []string
delay time.Duration
}
var groups []group
var cur *group
// startNew tracks whether the next User-agent line opens a new group or
// joins the current one: consecutive User-agent lines share their rules.
startNew := true
for _, raw := range strings.Split(body, "\n") {
line := raw
if i := strings.IndexByte(line, '#'); i >= 0 {
line = line[:i]
}
line = strings.TrimSpace(line)
if line == "" {
continue
}
key, val, ok := strings.Cut(line, ":")
if !ok {
continue
}
key = strings.ToLower(strings.TrimSpace(key))
val = strings.TrimSpace(val)
switch key {
case "user-agent":
if startNew || cur == nil {
groups = append(groups, group{})
cur = &groups[len(groups)-1]
startNew = false
}
cur.agents = append(cur.agents, strings.ToLower(val))
case "disallow":
if cur == nil {
continue
}
startNew = true
// "Disallow:" with an empty value allows everything, and is not a
// path rule at all.
if val != "" {
cur.disallow = append(cur.disallow, val)
}
case "allow":
if cur == nil {
continue
}
startNew = true
if val != "" {
cur.allow = append(cur.allow, val)
}
case "crawl-delay":
if cur == nil {
continue
}
startNew = true
if d, err := time.ParseDuration(val + "s"); err == nil && d > 0 {
cur.delay = d
}
}
}
var star, exact *group
for i := range groups {
for _, a := range groups[i].agents {
if a == "*" && star == nil {
star = &groups[i]
}
// A robots.txt names "maven", we send "Maven/1.0 (…)": match on
// prefix, which is how every crawler reads this field.
if a != "*" && a != "" && strings.HasPrefix(agent, a) {
exact = &groups[i]
}
}
}
g := exact
if g == nil {
g = star
}
if g == nil {
return Rules{}
}
return Rules{allow: g.allow, disallow: g.disallow, Delay: g.delay}
}
// Allowed reports whether path may be fetched. Longest matching rule wins, and
// Allow beats Disallow at equal length — the standard's tie-break, and the one
// that makes "Disallow: /" plus "Allow: /public" mean what it looks like.
func (r Rules) Allowed(path string) bool {
if path == "" {
path = "/"
}
best, allowed := -1, true
for _, p := range r.disallow {
if n, ok := matchPath(p, path); ok && n > best {
best, allowed = n, false
}
}
for _, p := range r.allow {
if n, ok := matchPath(p, path); ok && n >= best {
best, allowed = n, true
}
}
return allowed
}
// matchPath applies a robots path pattern and returns the pattern's length as
// the specificity score. `*` matches any run of characters, `$` anchors the end.
// A pattern is a PREFIX match otherwise, which is what "Disallow: /admin" means.
func matchPath(pattern, path string) (int, bool) {
score := len(pattern)
re, err := robotsRegexp(pattern)
if err != nil {
return 0, false
}
return score, re.MatchString(path)
}
// robotsRegexp turns a robots path pattern into an anchored-at-the-start
// regexp. Everything but `*` and a trailing `$` is a literal, so the pattern is
// quoted first and the two metacharacters are put back afterwards.
func robotsRegexp(pattern string) (*regexp.Regexp, error) {
end := ""
if strings.HasSuffix(pattern, "$") {
pattern = strings.TrimSuffix(pattern, "$")
end = "$"
}
parts := strings.Split(pattern, "*")
for i, p := range parts {
parts[i] = regexp.QuoteMeta(p)
}
return regexp.Compile("^" + strings.Join(parts, ".*") + end)
}
// robotsCache holds parsed rules per host so a crawl of ten pages on one site
// reads robots.txt once. TTL because a site may change its mind, and a daemon
// that runs for weeks would otherwise never notice.
type robotsCache struct {
ttl time.Duration
mu sync.Mutex
m map[string]robotsEntry
}
type robotsEntry struct {
rules Rules
at time.Time
}
func newRobotsCache(ttl time.Duration) *robotsCache {
return &robotsCache{ttl: ttl, m: map[string]robotsEntry{}}
}
func (c *robotsCache) get(host string, now time.Time) (Rules, bool) {
c.mu.Lock()
defer c.mu.Unlock()
e, ok := c.m[host]
if !ok || now.Sub(e.at) > c.ttl {
return Rules{}, false
}
return e.rules, true
}
func (c *robotsCache) put(host string, r Rules, now time.Time) {
c.mu.Lock()
defer c.mu.Unlock()
c.m[host] = robotsEntry{rules: r, at: now}
}
+84
View File
@@ -0,0 +1,84 @@
package crawl
import (
"testing"
"time"
)
const robotsBody = `# a comment
User-agent: *
Disallow: /private
Disallow: /tmp/
Crawl-delay: 5
User-agent: Maven
Disallow: /
Allow: /public
`
func TestParseRobotsPicksTheMostSpecificGroup(t *testing.T) {
// The Maven group applies to us even though we send a longer UA string.
r := ParseRobots(robotsBody, "Maven/1.0 (self-hosted personal assistant)")
if r.Allowed("/anything") {
t.Error("Disallow: / in our own group was ignored")
}
if !r.Allowed("/public/page") {
t.Error("Allow: /public must beat the shorter Disallow: /")
}
// A different agent falls into the * group.
star := ParseRobots(robotsBody, "SomeoneElse/2")
if !star.Allowed("/anything") {
t.Error("the * group disallows nothing but /private and /tmp/")
}
if star.Allowed("/private/x") || star.Allowed("/tmp/") {
t.Error("the * group's disallows were not applied")
}
if star.Delay != 5*time.Second {
t.Errorf("crawl-delay = %v, want 5s", star.Delay)
}
}
func TestParseRobotsEmptyMeansAllowAll(t *testing.T) {
for _, body := range []string{"", "# nothing here\n", "User-agent: *\nDisallow:\n"} {
if !ParseRobots(body, "Maven").Allowed("/whatever") {
t.Errorf("body %q must allow everything", body)
}
}
}
func TestRobotsWildcards(t *testing.T) {
r := ParseRobots("User-agent: *\nDisallow: /*.pdf$\nDisallow: /a/*/secret\n", "Maven")
if r.Allowed("/docs/manual.pdf") {
t.Error("*.pdf$ did not match")
}
if !r.Allowed("/docs/manual.pdf.html") {
t.Error("$ must anchor at the end")
}
if r.Allowed("/a/b/secret") {
t.Error("/a/*/secret did not match")
}
if !r.Allowed("/a/b/public") {
t.Error("unrelated path was refused")
}
}
// Consecutive User-agent lines share one group, which is common in the wild.
func TestRobotsSharedGroup(t *testing.T) {
r := ParseRobots("User-agent: Googlebot\nUser-agent: Maven\nDisallow: /x\n", "Maven/1.0")
if r.Allowed("/x/y") {
t.Fatal("a shared group's rules were not applied to the second agent")
}
}
func TestRobotsCacheTTL(t *testing.T) {
c := newRobotsCache(time.Minute)
now := time.Now()
c.put("example.com", ParseRobots("User-agent: *\nDisallow: /\n", "Maven"), now)
if _, ok := c.get("example.com", now.Add(30*time.Second)); !ok {
t.Error("a fresh entry must be served from cache")
}
if _, ok := c.get("example.com", now.Add(2*time.Minute)); ok {
t.Error("an expired entry must be re-read")
}
}
+177
View File
@@ -0,0 +1,177 @@
package crawl
import (
"context"
"fmt"
"log"
"strings"
"time"
)
// Scheduled crawls: a page is re-read on an interval, and when its TEXT changed
// the new text is written as a note. Nothing is dispatched — same rule as the
// feed poller (Vikunja #258). A page that announced its own change would be a
// nag, and "the docs page changed" is not worth interrupting anyone for.
//
// Dedup is by content hash, so a page that re-renders identically writes nothing
// and a rotating ad slot does not count as news.
// WatchConfig — one page to keep an eye on.
type WatchConfig struct {
Name string // note source is "crawl:<Name>"
URL string // http(s), guarded by the fetcher
Interval time.Duration // 0 ⇒ Watcher's default
}
// Notes is core's note-writing half (same shape as ipc.CoreAPI's method).
type Notes interface {
WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error)
}
// Hashes remembers the last text hash per watch, durably, so a restart does not
// re-note an unchanged page. The daemon backs this with config facts
// ("crawl:hash:<name>").
type Hashes interface {
LastHash(ctx context.Context, name string) (string, error)
SetHash(ctx context.Context, name, hash string) error
}
// Embedder embeds a note on its way into the store. nil ⇒ no vector.
type Embedder interface {
Embed(ctx context.Context, text string) ([]float32, error)
}
// DefaultWatchInterval — pages change slowly, and every check is a request in
// someone's log.
const DefaultWatchInterval = 6 * time.Hour
// Watcher re-reads watched pages on their interval.
type Watcher struct {
c *Crawler
watches []WatchConfig
notes Notes
hashes Hashes
embed Embedder
interval time.Duration
nextDue map[string]time.Time
}
// NewWatcher wires the scheduled half, or returns nil when there is nothing to
// watch. Callers check for nil: no watches, no goroutine, no request.
func NewWatcher(c *Crawler, watches []WatchConfig, notes Notes, hashes Hashes, embed Embedder, defaultInterval time.Duration) *Watcher {
if c == nil || notes == nil {
return nil
}
var valid []WatchConfig
for _, w := range watches {
if strings.TrimSpace(w.Name) == "" || strings.TrimSpace(w.URL) == "" {
log.Printf("crawl: skipping a watch with no name or no url")
continue
}
valid = append(valid, w)
}
if len(valid) == 0 {
return nil
}
if defaultInterval <= 0 {
defaultInterval = DefaultWatchInterval
}
return &Watcher{
c: c, watches: valid, notes: notes, hashes: hashes, embed: embed,
interval: defaultInterval, nextDue: map[string]time.Time{},
}
}
// Watches returns the configured watches.
func (w *Watcher) Watches() []WatchConfig { return w.watches }
// CheckDue re-reads every watch whose interval elapsed and returns how many
// notes were written. Errors are logged per watch, never returned: one dead page
// must not stop the others.
func (w *Watcher) CheckDue(ctx context.Context, now time.Time) int {
written := 0
for _, watch := range w.watches {
if due, ok := w.nextDue[watch.Name]; ok && now.Before(due) {
continue
}
interval := watch.Interval
if interval <= 0 {
interval = w.interval
}
w.nextDue[watch.Name] = now.Add(interval)
changed, err := w.Check(ctx, watch, now)
if err != nil {
log.Printf("crawl: watch %s: %v", watch.Name, err)
continue
}
if changed {
log.Printf("crawl: watch %s: page changed, noted", watch.Name)
written++
}
}
return written
}
// Check re-reads one watch now and reports whether it wrote a note.
func (w *Watcher) Check(ctx context.Context, watch WatchConfig, now time.Time) (bool, error) {
page, err := w.c.Page(ctx, watch.URL)
if err != nil {
return false, err
}
// Title included: a page whose headline changed has changed.
h := Hash(page.Title + "\n" + page.Text)
if w.hashes != nil {
prev, err := w.hashes.LastHash(ctx, watch.Name)
if err != nil {
log.Printf("crawl: watch %s: read hash: %v", watch.Name, err)
}
if prev == h {
return false, nil
}
}
text := NoteText(watch, page)
var vec []float32
if w.embed != nil {
v, err := w.embed.Embed(ctx, text)
if err != nil {
log.Printf("crawl: watch %s: embed: %v", watch.Name, err)
} else {
vec = v
}
}
if _, err := w.notes.WriteNote(ctx, now, text, vec, SourceFor(watch.Name)); err != nil {
return false, fmt.Errorf("write note: %w", err)
}
if w.hashes != nil {
if err := w.hashes.SetHash(ctx, watch.Name, h); err != nil {
log.Printf("crawl: watch %s: save hash: %v", watch.Name, err)
}
}
return true, nil
}
// SourceFor is the note source for a watch, and SourcePrefix is what the answer
// path matches to recognise one.
func SourceFor(name string) string { return SourcePrefix + name }
// SourcePrefix — provenance for anything read off the network on a schedule.
const SourcePrefix = "crawl:"
// noteRunes — how much of a watched page goes into a note. Shorter than what the
// on-demand path reads: a note is a record of a change, not an archive.
const noteRunes = 800
// NoteText renders a watched page as a note body.
func NoteText(watch WatchConfig, page Page) string {
var b strings.Builder
if page.Title != "" {
b.WriteString(page.Title)
} else {
b.WriteString(watch.Name)
}
b.WriteString("\n")
b.WriteString(TrimRunes(page.Text, noteRunes))
b.WriteString("\n")
b.WriteString(watch.URL)
return b.String()
}
+117
View File
@@ -0,0 +1,117 @@
package crawl
import (
"context"
"strings"
"testing"
"time"
)
type note struct {
text string
source string
}
type fakeNotes struct{ notes []note }
func (n *fakeNotes) WriteNote(_ context.Context, _ time.Time, text string, _ []float32, source string) (int64, error) {
n.notes = append(n.notes, note{text, source})
return int64(len(n.notes)), nil
}
type fakeHashes struct{ m map[string]string }
func newHashes() *fakeHashes { return &fakeHashes{m: map[string]string{}} }
func (f *fakeHashes) LastHash(_ context.Context, name string) (string, error) {
return f.m[name], nil
}
func (f *fakeHashes) SetHash(_ context.Context, name, h string) error { f.m[name] = h; return nil }
var t0 = time.Date(2026, 8, 1, 9, 0, 0, 0, time.UTC)
func TestWatchNotesAChangedPage(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{
"https://example.org/docs": {Body: []byte(htmlPage)},
}}
notes := &fakeNotes{}
hashes := newHashes()
w := NewWatcher(newTestCrawler(f), []WatchConfig{{Name: "docs", URL: "https://example.org/docs"}},
notes, hashes, nil, time.Hour)
if w == nil {
t.Fatal("NewWatcher returned nil for a configured watch")
}
if n := w.CheckDue(context.Background(), t0); n != 1 {
t.Fatalf("first check wrote %d notes, want 1", n)
}
if notes.notes[0].source != "crawl:docs" {
t.Errorf("source = %q, want crawl:docs", notes.notes[0].source)
}
if !strings.Contains(notes.notes[0].text, "https://example.org/docs") {
t.Errorf("note does not carry the url: %q", notes.notes[0].text)
}
// Unchanged page, interval elapsed: nothing written.
if n := w.CheckDue(context.Background(), t0.Add(2*time.Hour)); n != 0 {
t.Fatalf("an unchanged page wrote %d notes", n)
}
// Changed page: one note.
f.pages["https://example.org/docs"] = Response{Body: []byte(strings.Replace(htmlPage, "синее", "серое", 1))}
if n := w.CheckDue(context.Background(), t0.Add(4*time.Hour)); n != 1 {
t.Fatalf("a changed page wrote %d notes, want 1", n)
}
}
func TestWatchIntervalIsRespected(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{"https://example.org/d": {Body: []byte(htmlPage)}}}
w := NewWatcher(newTestCrawler(f), []WatchConfig{{Name: "d", URL: "https://example.org/d", Interval: time.Hour}},
&fakeNotes{}, newHashes(), nil, 0)
w.CheckDue(context.Background(), t0)
before := len(f.calls)
w.CheckDue(context.Background(), t0.Add(time.Minute))
if len(f.calls) != before {
t.Fatal("the page was re-read inside its interval")
}
}
// The hash is durable so a restart does not re-note an unchanged page.
func TestWatchHashSurvivesRestart(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{"https://example.org/d": {Body: []byte(htmlPage)}}}
hashes := newHashes()
watches := []WatchConfig{{Name: "d", URL: "https://example.org/d"}}
NewWatcher(newTestCrawler(f), watches, &fakeNotes{}, hashes, nil, time.Hour).CheckDue(context.Background(), t0)
notes2 := &fakeNotes{}
NewWatcher(newTestCrawler(f), watches, notes2, hashes, nil, time.Hour).CheckDue(context.Background(), t0.Add(time.Hour))
if len(notes2.notes) != 0 {
t.Fatalf("a fresh watcher re-noted an unchanged page: %q", notes2.notes[0].text)
}
}
func TestWatchDeadPageDoesNotStopTheOthers(t *testing.T) {
f := &fakeFetcher{pages: map[string]Response{"https://example.org/live": {Body: []byte(htmlPage)}}}
notes := &fakeNotes{}
w := NewWatcher(newTestCrawler(f), []WatchConfig{
{Name: "dead", URL: "https://example.org/gone"},
{Name: "live", URL: "https://example.org/live"},
}, notes, newHashes(), nil, time.Hour)
if n := w.CheckDue(context.Background(), t0); n != 1 {
t.Fatalf("wrote %d notes, want 1 (the live page)", n)
}
if notes.notes[0].source != "crawl:live" {
t.Fatalf("source = %q", notes.notes[0].source)
}
}
func TestNoWatchesMeansNoWatcher(t *testing.T) {
c := newTestCrawler(&fakeFetcher{})
if NewWatcher(c, nil, &fakeNotes{}, nil, nil, 0) != nil {
t.Fatal("no watches must mean no watcher")
}
if NewWatcher(nil, []WatchConfig{{Name: "a", URL: "u"}}, &fakeNotes{}, nil, nil, 0) != nil {
t.Fatal("no crawler must mean no watcher")
}
if NewWatcher(c, []WatchConfig{{Name: "", URL: ""}}, &fakeNotes{}, nil, nil, 0) != nil {
t.Fatal("a watch with no name or url is not a configuration")
}
}
+226
View File
@@ -0,0 +1,226 @@
package email
import (
"context"
"encoding/json"
"fmt"
"strings"
"time"
"github.com/kami/maven/internal/llm"
"github.com/kami/maven/internal/persona"
)
// Extraction — turning one mail into task CANDIDATES, and nothing else.
//
// The output of this file can only ever become rows in `tasks` with status
// "candidate" (store.TaskCandidate), written through the one intake seam
// (ipc.CaptureTaskReq, Vikunja #130). That bound is the whole design:
//
// - No reminder. A reminder FIRES; it speaks to him unprompted. A 1.7B that
// misreads "встреча была в четверг" as a future appointment would then wake
// him up about it. A candidate that is wrong is a line on a review page he
// dismisses in one click, which is the correct cost of a model being wrong
// about someone's mail.
// - No fact. A fact is a claim Maven will later recite as true. Nothing read
// out of a marketing mail deserves that standing.
// - No calendar event, no note, no action. Extraction writes candidates or
// writes nothing.
//
// The due date the model may return is stored on the candidate (tasks.due_ts),
// which no scheduler reads — it is there so the review page can sort by it.
//
// Privacy: the mail text goes to the resident model on this box and nowhere
// else. It is never search input (CLAUDE.md: "his notes and facts are never
// search input" — mail is the same class), and Evidence keeps only the subject
// line, so the review page shows him where a candidate came from without the
// store growing a copy of his mailbox.
// MaxCandidates — at most this many candidates per message, enforced by the
// grammar. A mail with four tasks in it is a mail he has to read himself; a
// model allowed ten will produce ten.
const MaxCandidates = 3
// SourcePrefix — provenance for everything this package captures. The mailbox
// name is appended: "email:INBOX". Same vocabulary as tap:voice / poll:netdata.
const SourcePrefix = "email:"
// Candidate — one piece of work the model thinks the mail is asking for.
type Candidate struct {
Text string `json:"text"`
// Due — "YYYY-MM-DD" or empty. A date the model read out of the text, not a
// date it computed: relative wording ("до пятницы") is left in Text, because
// a small model resolving "пятница" against today's date gets it wrong often
// enough that a stored wrong date is worse than no date.
Due string `json:"due"`
}
// Completer — the llama-server seam, same shape memeval and the router use, so
// the one resident model serves this caller too.
type Completer interface {
Complete(ctx context.Context, r llm.Req) (string, error)
}
// Extractor reads a message and returns candidates. It holds no store and no
// writer on purpose: this type cannot persist anything, so "extraction never
// acts" is a property of the code, not of a review.
type Extractor struct {
llm Completer
// MaxCandidates — 0 ⇒ MaxCandidates.
max int
// ContextBlock — the shared persona block, optional. Extraction output is
// not spoken, so the persona matters less here than in the phraser; it is
// wired anyway so a candidate reads in her voice on the review page.
contextBlock func() string
}
func NewExtractor(c Completer, max int, contextBlock func() string) *Extractor {
if max <= 0 || max > MaxCandidates {
max = MaxCandidates
}
return &Extractor{llm: c, max: max, contextBlock: contextBlock}
}
// extractGrammar — GBNF pinning the answer to a bounded array of fixed-shape
// candidates. Same reasoning as memeval's evalGrammar and the router's
// routeGrammar: the shape and the length bound are what keep a small model from
// drifting into prose or spending the token budget repeating one field.
//
// The empty array is reachable, deliberately: most mail contains no task, and a
// model with no way to say "nothing" invents something.
const extractGrammar = `
root ::= "[" ws (item ("," ws item){0,2})? ws "]"
item ::= "{" ws "\"text\"" ws ":" ws text "," ws "\"due\"" ws ":" ws due ws "}"
text ::= "\"" ([^"\\] | "\\" .){1,120} "\""
due ::= "\"\"" | "\"" [0-9]{4} "-" [0-9]{2} "-" [0-9]{2} "\""
ws ::= [ \t\n]*
`
// extractSystem — the extraction prompt.
//
// Written around the two failure modes a small model has on this task: it
// summarises when asked to extract (turning a mail into "письмо от Антона"),
// and it invents an obligation from any polite closing sentence. Hence the
// insistence on a verb phrase, and the explicit permission to return [].
const extractSystem = `Ты читаешь одно письмо из его почты и достаёшь из него дела, которые письмо от него требует.
Правила:
- Отвечай ТОЛЬКО массивом JSON. Каждый элемент: {"text": "...", "due": "ГГГГ-ММ-ДД" или ""}.
- text — короткая формулировка дела по-русски, с глаголом: "оплатить счёт за интернет", "отправить акт". Не пересказывай письмо и не описывай его.
- Дело — это то, что должен сделать ОН. Рассылка, реклама, уведомление, отчёт, письмо «просто к сведению» — дел не содержат.
- Если письмо ничего от него не требует, верни пустой массив []. Это нормальный ответ, так бывает чаще всего.
- Ничего не придумывай. Если срока в письме нет — "".
- due заполняй только когда в письме стоит конкретная дата. Слова вроде «до пятницы» оставь в text, дату не вычисляй.
- Максимум три дела. Лучше одно точное, чем три общих.`
// Extract returns the candidates in one message.
//
// Junk is refused without an LLM call — cheapest possible defence, and the
// reason the header filter exists. An empty message (no subject, no body) is
// likewise not worth a round trip.
//
// A parse failure is an error the caller logs and moves past. It is never
// silently turned into zero candidates, because "the model went off the rails"
// and "the mail contains no task" want different reactions from a human reading
// the log.
func (e *Extractor) Extract(ctx context.Context, msg Message) ([]Candidate, error) {
if msg.Junk {
return nil, nil
}
user := renderForModel(msg)
if user == "" {
return nil, nil
}
raw, err := e.llm.Complete(ctx, llm.Req{
System: persona.Prepend(e.contextBlock, extractSystem),
User: user,
Grammar: extractGrammar,
MaxTokens: 512,
RepeatPenalty: 1.1,
})
if err != nil {
return nil, fmt.Errorf("email: extract: %w", err)
}
items, err := parseCandidates(raw)
if err != nil {
// The raw reply is NOT in the error: it is a transformation of his mail,
// and this error reaches the daemon log.
return nil, fmt.Errorf("email: extract: unparsable reply (%d bytes)", len(raw))
}
out := make([]Candidate, 0, len(items))
seen := map[string]bool{}
for _, it := range items {
it.Text = strings.TrimSpace(it.Text)
if it.Text == "" {
continue
}
key := strings.ToLower(strings.Join(strings.Fields(it.Text), " "))
if seen[key] {
continue // the model repeating itself is not two tasks
}
seen[key] = true
if _, ok := ParseDue(it.Due); !ok {
it.Due = "" // a date the grammar allowed but the calendar does not
}
out = append(out, it)
if len(out) >= e.max {
break
}
}
return out, nil
}
// renderForModel is the user turn: subject, sender and body, labelled. Only
// these three fields — no headers, no recipient list, no message-id, nothing
// that would let the model start reasoning about routing metadata.
func renderForModel(msg Message) string {
var b strings.Builder
if msg.From != "" {
fmt.Fprintf(&b, "От: %s\n", msg.From)
}
if msg.Subject != "" {
fmt.Fprintf(&b, "Тема: %s\n", msg.Subject)
}
if msg.Body != "" {
fmt.Fprintf(&b, "\n%s\n", msg.Body)
}
if msg.Subject == "" && msg.Body == "" {
return ""
}
return b.String()
}
// parseCandidates decodes the grammar-constrained reply, tolerating the
// wrappers a Thinking model sometimes leaves around it (a fenced block, or
// leading reasoning before the array).
func parseCandidates(raw string) ([]Candidate, error) {
s := strings.TrimSpace(raw)
if i := strings.Index(s, "["); i > 0 {
s = s[i:]
}
if j := strings.LastIndex(s, "]"); j >= 0 {
s = s[:j+1]
}
var out []Candidate
if err := json.Unmarshal([]byte(s), &out); err != nil {
return nil, err
}
return out, nil
}
// ParseDue turns the model's "YYYY-MM-DD" into a time in UTC. Exported because
// the daemon-side intake stores it on the candidate.
//
// The zero-value/empty case returns ok=false rather than an error: no date is
// the common answer, not a failure.
func ParseDue(s string) (time.Time, bool) {
s = strings.TrimSpace(s)
if s == "" {
return time.Time{}, false
}
t, err := time.Parse("2006-01-02", s)
if err != nil {
return time.Time{}, false
}
return t, true
}
+143
View File
@@ -0,0 +1,143 @@
package email
import (
"context"
"strings"
"testing"
"github.com/kami/maven/internal/llm"
)
// fakeLLM returns a canned reply and records the request, so a test can assert
// on the grammar and on what of the mail was sent.
type fakeLLM struct {
reply string
err error
got llm.Req
calls int
}
func (f *fakeLLM) Complete(_ context.Context, r llm.Req) (string, error) {
f.calls++
f.got = r
return f.reply, f.err
}
func msgFor(subject, body string) Message {
return Message{UID: 1, From: "anton@example.org", Subject: subject, Body: body}
}
func TestExtractCandidates(t *testing.T) {
f := &fakeLLM{reply: `[{"text":"отправить акт","due":""},{"text":"оплатить счёт","due":"2026-08-05"}]`}
e := NewExtractor(f, 0, nil)
got, err := e.Extract(context.Background(), msgFor("Акт и счёт", "Надо отправить акт и оплатить счёт до 5 августа."))
if err != nil {
t.Fatalf("extract: %v", err)
}
if len(got) != 2 {
t.Fatalf("got %d candidates, want 2: %+v", len(got), got)
}
if got[0].Text != "отправить акт" || got[1].Due != "2026-08-05" {
t.Errorf("candidates = %+v", got)
}
if f.got.Grammar == "" {
t.Error("extraction must be grammar-constrained")
}
// The subject and body go to the model; nothing else about the message does.
if !strings.Contains(f.got.User, "Акт и счёт") || !strings.Contains(f.got.User, "оплатить счёт") {
t.Errorf("user turn = %q", f.got.User)
}
}
func TestExtractEmptyArrayIsNotAnError(t *testing.T) {
f := &fakeLLM{reply: "[]"}
got, err := NewExtractor(f, 0, nil).Extract(context.Background(), msgFor("FYI", "Просто к сведению."))
if err != nil || len(got) != 0 {
t.Fatalf("got (%v, %v), want (empty, nil) — no task is the normal answer", got, err)
}
}
// Junk must never reach the model: the header filter exists so the resident
// model is not spent on newsletters.
func TestExtractSkipsJunkWithoutCallingModel(t *testing.T) {
f := &fakeLLM{reply: `[{"text":"купить всё со скидкой","due":""}]`}
msg := msgFor("Скидки", "Sale!")
msg.Junk = true
got, err := NewExtractor(f, 0, nil).Extract(context.Background(), msg)
if err != nil || got != nil {
t.Fatalf("got (%v, %v), want (nil, nil)", got, err)
}
if f.calls != 0 {
t.Errorf("model called %d times for junk, want 0", f.calls)
}
}
func TestExtractEmptyMessageIsNotSent(t *testing.T) {
f := &fakeLLM{reply: "[]"}
if _, err := NewExtractor(f, 0, nil).Extract(context.Background(), Message{UID: 3}); err != nil {
t.Fatalf("extract: %v", err)
}
if f.calls != 0 {
t.Errorf("model called %d times for an empty message, want 0", f.calls)
}
}
func TestExtractCaps(t *testing.T) {
f := &fakeLLM{reply: `[{"text":"a","due":""},{"text":"b","due":""},{"text":"c","due":""}]`}
got, err := NewExtractor(f, 2, nil).Extract(context.Background(), msgFor("s", "b"))
if err != nil {
t.Fatalf("extract: %v", err)
}
if len(got) != 2 {
t.Errorf("got %d, want the configured cap of 2", len(got))
}
}
func TestExtractDropsRepeatsAndBadDates(t *testing.T) {
f := &fakeLLM{reply: `[{"text":"Отправить акт","due":"2026-02-31"},{"text":"отправить акт","due":""},{"text":" ","due":""}]`}
got, err := NewExtractor(f, 0, nil).Extract(context.Background(), msgFor("s", "b"))
if err != nil {
t.Fatalf("extract: %v", err)
}
if len(got) != 1 {
t.Fatalf("got %d candidates, want 1 (repeat and blank dropped): %+v", len(got), got)
}
if got[0].Due != "" {
t.Errorf("due = %q, want empty — 2026-02-31 is not a date", got[0].Due)
}
}
// A Thinking model sometimes wraps the array; and when it emits something
// unparsable the caller must hear about it rather than see "no tasks".
func TestParseCandidatesTolerance(t *testing.T) {
got, err := parseCandidates("думаю... [{\"text\":\"x\",\"due\":\"\"}] всё")
if err != nil || len(got) != 1 || got[0].Text != "x" {
t.Fatalf("got (%+v, %v)", got, err)
}
if _, err := parseCandidates("нет никакого JSON"); err == nil {
t.Error("unparsable output must be an error")
}
}
func TestExtractParseErrorHidesMailText(t *testing.T) {
f := &fakeLLM{reply: "он просил отправить акт, вот такой ответ"}
_, err := NewExtractor(f, 0, nil).Extract(context.Background(), msgFor("Акт", "секретный текст"))
if err == nil {
t.Fatal("want an error")
}
if strings.Contains(err.Error(), "акт") || strings.Contains(err.Error(), "секретный") {
t.Errorf("error text leaks mail content: %v", err)
}
}
func TestParseDue(t *testing.T) {
if _, ok := ParseDue(""); ok {
t.Error("empty due must be (zero, false)")
}
if got, ok := ParseDue("2026-08-05"); !ok || got.Year() != 2026 || got.Month() != 8 || got.Day() != 5 {
t.Errorf("ParseDue = (%v, %v)", got, ok)
}
if _, ok := ParseDue("05.08.2026"); ok {
t.Error("a non-ISO date must not parse")
}
}
+8 -6
View File
@@ -26,21 +26,23 @@ type FetchSince struct {
Since time.Time
Max int
Skip func(uid uint32) bool
// dial is the connection seam. nil means Dial (implicit TLS); the tests set
// it to an in-process fake. Unexported so no configuration path can point
// the reader at a non-TLS transport.
dial func(addr string, timeout time.Duration) (*Conn, error)
}
// Run performs one read. password is passed here, not stored in the struct, so
// the configuration of a mailbox and the secret for it are never the same value
// sitting in the same place.
func (f FetchSince) Run(password string) ([]Message, error) {
return f.RunWith(password, nil)
}
// RunWith is Run with an explicit connection function, which is how the reader
// daemon and the tests substitute an in-process server. nil ⇒ Dial, i.e.
// implicit TLS with certificate verification; there is no configuration path
// that reaches this, so no deployment can end up talking cleartext IMAP.
func (f FetchSince) RunWith(password string, dial func(addr string, timeout time.Duration) (*Conn, error)) ([]Message, error) {
if f.Addr == "" || f.User == "" || f.Mailbox == "" {
return nil, fmt.Errorf("email: mailbox not configured (addr/user/mailbox)")
}
dial := f.dial
if dial == nil {
dial = Dial
}
+5 -6
View File
@@ -21,13 +21,12 @@ func TestFetchSinceRun(t *testing.T) {
Since: time.Date(2026, 7, 30, 0, 0, 0, 0, time.UTC),
Max: 2,
Skip: func(uid uint32) bool { return uid == 3 },
dial: func(addr string, timeout time.Duration) (*Conn, error) {
cli, srv := net.Pipe()
go f.serve(t, srv)
return NewConn(cli, timeout)
},
}
msgs, err := fs.Run("secret")
msgs, err := fs.RunWith("secret", func(addr string, timeout time.Duration) (*Conn, error) {
cli, srv := net.Pipe()
go f.serve(t, srv)
return NewConn(cli, timeout)
})
if err != nil {
t.Fatalf("run: %v", err)
}
+38
View File
@@ -138,6 +138,44 @@ type CaptureTaskResp struct {
Created bool `json:"created"`
}
// IngestMailReq — one message a mail reader has fetched, handed to core for
// extraction (Vikunja #246).
//
// The mail reader (cmd/mavmaild) holds the IMAP credential and core never sees
// it, the same split mavpoll uses for the zenmoney token. What crosses this
// boundary is only the message text, because extraction runs on the resident
// model and llama-server lives inside core's process.
//
// Body is already plaintext and truncated by internal/email; core does not
// re-parse MIME and never stores the body. Junk means the reader's header
// filter already classified the message as bulk — core is told rather than
// asked, so a junk message can be counted without a model call.
//
// This method is available only when core has an email block configured AND a
// llama-server phraser; otherwise it answers ErrUnknownMethod, which is what
// "off unless configured" looks like at the wire.
type IngestMailReq struct {
Mailbox string `json:"mailbox"`
UID uint32 `json:"uid"`
From string `json:"from,omitempty"`
Subject string `json:"subject,omitempty"`
Date string `json:"date,omitempty"`
Body string `json:"body,omitempty"`
Junk bool `json:"junk,omitempty"`
}
// IngestMailResp — what core did with the message. TaskIDs are the rows
// CaptureTask returned; Created counts the ones that were new (a re-read
// mailbox dedupes to Created=0). Skipped is set when nothing was asked of the
// model at all — junk, or an empty message.
//
// Nothing here echoes the mail back. The reader logs counts.
type IngestMailResp struct {
TaskIDs []int64 `json:"task_ids,omitempty"`
Created int `json:"created"`
Skipped bool `json:"skipped,omitempty"`
}
type listTasksReq struct {
Status string `json:"status"` // "" all | "live" | candidate|open|done|dropped
}
+11
View File
@@ -448,6 +448,17 @@ func (c *Client) SetTaskStatus(ctx context.Context, id int64, status string, ts
return c.call(ctx, MethodSetTaskStatus, setTaskStatusReq{ID: id, Status: status, Ts: ts}, nil)
}
// IngestMail hands one fetched message to core for extraction. ErrUnknownMethod
// means core has no email block configured — the caller should stop asking, not
// retry.
func (c *Client) IngestMail(ctx context.Context, req IngestMailReq) (IngestMailResp, error) {
var r IngestMailResp
if err := c.call(ctx, MethodIngestMail, req, &r); err != nil {
return IngestMailResp{}, err
}
return r, nil
}
func (c *Client) DismissProposedRoutine(ctx context.Context, id int64) error {
return c.call(ctx, MethodDismissProposedRoutine, dismissProposedRoutineReq{ID: id}, nil)
}
+33
View File
@@ -565,3 +565,36 @@ func mustJSON(v any) []byte {
}
return b
}
// TestIngestMail_OffUnlessConfigured — with no IngestMailFn set (the default,
// and what an unconfigured core looks like) the method does not exist. A mail
// reader gets a refusal it can act on rather than a silent success.
func TestIngestMail_OffUnlessConfigured(t *testing.T) {
_, _, cli, _ := newServerWithStore(t)
if _, err := cli.IngestMail(context.Background(), IngestMailReq{Mailbox: "INBOX", UID: 1}); !errors.Is(err, ErrUnknownMethod) {
t.Fatalf("IngestMail error = %v, want ErrUnknownMethod", err)
}
}
// TestIngestMail_Hook — when the daemon wires the hook, the message crosses the
// boundary intact and the response comes back.
func TestIngestMail_Hook(t *testing.T) {
_, srv, cli, _ := newServerWithStore(t)
var got IngestMailReq
srv.IngestMailFn = func(_ context.Context, req IngestMailReq) (IngestMailResp, error) {
got = req
return IngestMailResp{TaskIDs: []int64{7}, Created: 1}, nil
}
resp, err := cli.IngestMail(context.Background(), IngestMailReq{
Mailbox: "INBOX", UID: 12, Subject: "Счёт", Body: "Оплатить.", Junk: false,
})
if err != nil {
t.Fatalf("IngestMail: %v", err)
}
if resp.Created != 1 || len(resp.TaskIDs) != 1 || resp.TaskIDs[0] != 7 {
t.Errorf("resp = %+v", resp)
}
if got.UID != 12 || got.Subject != "Счёт" || got.Body != "Оплатить." {
t.Errorf("req across the wire = %+v", got)
}
}
+33 -4
View File
@@ -421,6 +421,17 @@ type Server struct {
// Set by the daemon; nil ⇒ MethodStoreEncryptionKey returns ErrUnknownMethod.
WrapKeyFn WrapKeyFunc
// IngestMailFn — extracts task candidates from one fetched message. Set by
// the daemon only when an email block is configured AND there is a
// llama-server to extract with; nil ⇒ MethodIngestMail returns
// ErrUnknownMethod, so a mail reader pointed at a core that is not
// configured for mail is refused rather than silently ignored.
//
// Like StepUp/WrapKeyFn/UnlockFn this bypasses CoreAPI: it is not a store
// operation, it needs the resident model, and it must not become a method
// every CoreAPI implementation has to carry.
IngestMailFn IngestMailFunc
// UnlockFn — unwraps the store encryption key from the wrapped blob using
// the passkey credential public key, opens the encrypted store, and wires
// the rest of the daemon (voice, loop, delivery). Set by the daemon when
@@ -439,6 +450,9 @@ type WrapKeyFunc func(ctx context.Context, publicKey []byte) error
// public key and completes daemon initialization.
type UnlockFunc func(ctx context.Context, publicKey []byte) error
// IngestMailFunc — core-side mail extraction. Returns what was captured.
type IngestMailFunc func(ctx context.Context, req IngestMailReq) (IngestMailResp, error)
// CheckFunc — the auth hook signature. Wired by the daemon (auth.Gate.Check
// satisfies this); dispatch calls it once per request after param-unmarshal
// independence (it gets the raw params, may unmarshal what it needs — ipc
@@ -606,9 +620,10 @@ func withoutParams[R any](fn func(ctx context.Context, api CoreAPI) (R, error))
// existed) as an argument — so SetAPI's runtime swap (the unlock transition)
// is still honored on the very next request with no extra plumbing here.
//
// MethodAssertStepUp, MethodStoreEncryptionKey and MethodUnlock are NOT in
// this table: they bypass CoreAPI entirely (s.StepUp / s.WrapKeyFn /
// s.UnlockFn), so dispatch special-cases them before consulting the table.
// MethodAssertStepUp, MethodStoreEncryptionKey, MethodUnlock and
// MethodIngestMail are NOT in this table: they bypass CoreAPI entirely
// (s.StepUp / s.WrapKeyFn / s.UnlockFn / s.IngestMailFn), so dispatch
// special-cases them before consulting the table.
var methodTable = map[Method]handlerFunc{
MethodWriteFact: withParams(func(ctx context.Context, api CoreAPI, p WriteFactReq) (idResp, error) {
id, err := api.WriteFact(ctx, p)
@@ -816,7 +831,7 @@ func (s *Server) dispatch(ctx context.Context, req Request) (json.RawMessage, er
}
}
// These three bypass CoreAPI entirely — they drive Server fields set
// These bypass CoreAPI entirely — they drive Server fields set
// directly by the daemon (StepUp / WrapKeyFn / UnlockFn), not store
// state, so they can never be table entries keyed on a CoreAPI method.
switch req.Method {
@@ -845,6 +860,20 @@ func (s *Server) dispatch(ctx context.Context, req Request) (json.RawMessage, er
return marshalResult(nil), s.UnlockFn(ctx, p.PublicKey)
}
return nil, fmt.Errorf("%w: %s", ErrUnknownMethod, req.Method)
case MethodIngestMail:
if s.IngestMailFn != nil {
var p IngestMailReq
if err := unmarshalParams(req.Params, &p); err != nil {
return nil, err
}
resp, err := s.IngestMailFn(ctx, p)
if err != nil {
return nil, err
}
return marshalResult(resp), nil
}
return nil, fmt.Errorf("%w: %s", ErrUnknownMethod, req.Method)
}
h, ok := methodTable[req.Method]
+1
View File
@@ -50,6 +50,7 @@ const (
MethodCaptureTask Method = "capture_task"
MethodListTasks Method = "list_tasks"
MethodSetTaskStatus Method = "set_task_status"
MethodIngestMail Method = "ingest_mail"
)
// Request — one frame from module to core. Params is the JSON-encoded argument
+109
View File
@@ -0,0 +1,109 @@
package router
import "strings"
// Feed queries — "что нового в лентах?", "что нового по технологиям?"
// (Vikunja #258).
//
// Deterministic matching, like the calendar, plan and habit matchers above it:
// the LLM router says this is a query, and this decides whether it is a question
// about the feeds. A model deciding that would occasionally answer "что нового?"
// out of world knowledge, which is the one thing a feed reader exists to avoid.
// FeedQuery — a parsed "what's new" question. Category is the topic he named
// ("технологии"), empty when he asked about the feeds in general.
type FeedQuery struct {
Category string
}
// feedNouns — the words that make a question about the feeds themselves.
var feedNouns = []string{
"лента", "ленте", "ленты", "лентах", "лентам",
"новости", "новостей", "новостях", "новостям",
"новое", "нового", "новенького",
"feed", "feeds", "news", "headlines",
}
// newnessMarkers — the "что нового" half. "нового" alone is in feedNouns
// because it carries the question on its own ("что нового?"); a bare "лента"
// needs the ask, which is what askMarkers below is for.
var askMarkers = []string{
"что", "какие", "какое", "расскажи", "почитай", "прочитай", "покажи",
"what", "any", "tell", "show", "read",
}
// ParseFeedQuery reports whether an utterance asks what is new in the feeds, and
// which topic if it names one after "по"/"о"/"про"/"about".
//
// Both a feed noun and an ask are required. "у меня новая лента в инстаграме" is
// a statement and must not be read as a request to recite headlines.
func ParseFeedQuery(text string) (FeedQuery, bool) {
toks := planTokens(text)
noun, ask := false, false
for _, t := range toks {
for _, n := range feedNouns {
if t == n {
noun = true
break
}
}
for _, a := range askMarkers {
if t == a {
ask = true
break
}
}
}
if !noun || !ask {
return FeedQuery{}, false
}
return FeedQuery{Category: feedCategory(toks)}, true
}
// categoryPreps — the prepositions a topic follows. Russian marks the topic with
// a preposition ("по технологиям", "про политику"), so the word after one is the
// category; there is no stemming here, and the match against the configured
// category is a prefix comparison for exactly that reason.
var categoryPreps = map[string]bool{"по": true, "о": true, "об": true, "про": true, "about": true, "on": true}
func feedCategory(toks []string) string {
for i, t := range toks {
if categoryPreps[t] && i+1 < len(toks) {
next := toks[i+1]
// "по новостям" names no topic, it repeats the noun.
for _, n := range feedNouns {
if next == n {
return ""
}
}
return next
}
}
return ""
}
// CategoryMatches reports whether a note's text plausibly belongs to the
// category he named. Russian inflects the topic ("технологиям" vs the configured
// "технологии"), and there is no stemmer in this repo, so the comparison is on a
// common prefix — long enough that "полит" and "погод" stay apart, short enough
// to survive a case ending.
func CategoryMatches(text, category string) bool {
if category == "" {
return true
}
stem := categoryStem(category)
if stem == "" {
return false
}
return strings.Contains(strings.ToLower(text), stem)
}
// categoryStem cuts a word down to the part inflection leaves alone. 5 runes is
// the compromise: shorter words are used whole.
func categoryStem(word string) string {
r := []rune(strings.ToLower(strings.TrimSpace(word)))
if len(r) > 5 {
r = r[:5]
}
return string(r)
}
+48
View File
@@ -0,0 +1,48 @@
package router
import "testing"
func TestParseFeedQuery(t *testing.T) {
cases := []struct {
text string
ok bool
category string
}{
{"что нового в лентах?", true, ""},
{"что нового?", true, ""},
{"какие новости?", true, ""},
{"что нового по технологиям?", true, "технологиям"},
{"расскажи новости про политику", true, "политику"},
{"что нового по новостям", true, ""},
{"what's new in the feeds?", true, ""},
{"any news about kubernetes", true, "kubernetes"},
// Statements, not requests.
{"у меня новая лента в инстаграме", false, ""},
{"новости меня утомили", false, ""},
{"напомни полить цветы", false, ""},
{"", false, ""},
}
for _, c := range cases {
q, ok := ParseFeedQuery(c.text)
if ok != c.ok {
t.Errorf("ParseFeedQuery(%q) ok = %v, want %v", c.text, ok, c.ok)
continue
}
if ok && q.Category != c.category {
t.Errorf("ParseFeedQuery(%q) category = %q, want %q", c.text, q.Category, c.category)
}
}
}
func TestCategoryMatches(t *testing.T) {
// The inflected form he says must match the form the config spells.
if !CategoryMatches("Новый релиз [технологии]", "технологиям") {
t.Error("inflected category did not match")
}
if CategoryMatches("Новый релиз [технологии]", "политику") {
t.Error("unrelated category matched")
}
if !CategoryMatches("anything", "") {
t.Error("an empty category must match everything")
}
}
+39
View File
@@ -0,0 +1,39 @@
package router
import (
"regexp"
"strings"
)
// Finding a URL in an utterance (Vikunja #259).
//
// This is deliberately strict: a scheme is required. "посмотри на example.org"
// is not treated as a fetch request, because a bare dotted word is also how
// people say file names, versions and Russian abbreviations, and the cost of a
// false positive here is an outbound request nobody asked for.
//
// Note where this runs: an utterance from STT. Whisper will mangle a spoken URL,
// which is fine — the URL that survives is one he pasted into the web chat, and
// a mangled one simply fails to match.
var urlRE = regexp.MustCompile(`(?i)\bhttps?://[^\s<>"']+`)
// FirstURL returns the first http(s) URL in text.
//
// Trailing punctuation is trimmed: he ends sentences, and "…/page." is not a
// path component. A closing bracket is only trimmed when it has no opener,
// because a wikipedia URL legitimately ends in one.
func FirstURL(text string) (string, bool) {
m := urlRE.FindString(text)
if m == "" {
return "", false
}
m = strings.TrimRight(m, ".,;:!?…")
if strings.HasSuffix(m, ")") && strings.Count(m, "(") == 0 {
m = strings.TrimSuffix(m, ")")
}
// A scheme with nothing after it is not a URL.
if rest := strings.SplitN(m, "//", 2); len(rest) < 2 || rest[1] == "" {
return "", false
}
return m, true
}
+34
View File
@@ -0,0 +1,34 @@
package router
import "testing"
func TestFirstURL(t *testing.T) {
cases := []struct {
text string
want string
}{
{"посмотри https://example.org/page — что там?", "https://example.org/page"},
{"почитай http://example.org/a/b?x=1 и скажи", "http://example.org/a/b?x=1"},
{"вот ссылка: https://example.org/page.", "https://example.org/page"},
{"https://ru.wikipedia.org/wiki/Небо_(значения)", "https://ru.wikipedia.org/wiki/Небо_(значения)"},
// No scheme ⇒ no fetch. A bare dotted word is not an instruction to
// reach out to the network.
{"посмотри на example.org", ""},
{"открой файл config.json", ""},
{"что нового?", ""},
{"https://", ""},
{"", ""},
}
for _, c := range cases {
got, ok := FirstURL(c.text)
if c.want == "" {
if ok {
t.Errorf("FirstURL(%q) = %q, want no match", c.text, got)
}
continue
}
if !ok || got != c.want {
t.Errorf("FirstURL(%q) = %q, %v; want %q", c.text, got, ok, c.want)
}
}
}
+179
View File
@@ -0,0 +1,179 @@
// Package rss reads RSS 2.0 and Atom feeds, and does nothing else with them.
//
// Parsing and polling are split from delivery on purpose: a feed is a source
// Maven can be ASKED about, not a thing that speaks. Nothing in this package
// dispatches, nudges or notifies — the poller writes notes, and the answer path
// reads them when he asks "что нового в лентах?". "Not a nag" is the oldest
// constraint in the spec, and a news feed is the single most tempting way to
// break it.
//
// Stdlib only (encoding/xml). Feeds are XML from strangers, so the parser takes
// what it recognises and ignores the rest rather than failing a whole feed over
// one malformed item.
package rss
import (
"encoding/xml"
"fmt"
"html"
"io"
"regexp"
"strings"
"time"
)
// Item is one feed entry, normalised across RSS and Atom.
type Item struct {
Title string
Link string
Summary string // plain text, tags stripped, entities decoded
Published time.Time // zero when the feed did not say
ID string // guid / atom id, falling back to the link
}
// Feed is a parsed document.
type Feed struct {
Title string
Items []Item
}
// feedDoc covers both dialects in one struct. RSS puts items under
// channel>item, Atom puts entries at the top level, and the field names barely
// overlap — so both sets are declared and whichever the document filled in wins.
type feedDoc struct {
ChannelTitle string `xml:"channel>title"`
AtomTitle string `xml:"title"`
Items []struct {
Title string `xml:"title"`
Link string `xml:"link"`
Description string `xml:"description"`
Encoded string `xml:"encoded"` // content:encoded
GUID string `xml:"guid"`
PubDate string `xml:"pubDate"`
Date string `xml:"date"` // dc:date
} `xml:"channel>item"`
Entries []struct {
Title string `xml:"title"`
Links []struct {
Href string `xml:"href,attr"`
Rel string `xml:"rel,attr"`
} `xml:"link"`
Summary string `xml:"summary"`
Content string `xml:"content"`
ID string `xml:"id"`
Updated string `xml:"updated"`
Published string `xml:"published"`
} `xml:"entry"`
}
// Parse reads a feed document.
func Parse(r io.Reader) (Feed, error) {
var doc feedDoc
dec := xml.NewDecoder(r)
// Feeds in the wild declare windows-1251 and worse. We only ever read
// UTF-8; a charset we cannot decode is a feed we do not read, which is
// better than mojibake in his notes.
dec.Strict = false
if err := dec.Decode(&doc); err != nil {
return Feed{}, fmt.Errorf("rss: bad xml: %w", err)
}
f := Feed{Title: strings.TrimSpace(doc.ChannelTitle)}
if f.Title == "" {
f.Title = strings.TrimSpace(doc.AtomTitle)
}
for _, it := range doc.Items {
item := Item{
Title: PlainText(it.Title),
Link: strings.TrimSpace(it.Link),
Summary: PlainText(firstNonEmpty(it.Description, it.Encoded)),
Published: parseTime(firstNonEmpty(it.PubDate, it.Date)),
ID: strings.TrimSpace(firstNonEmpty(it.GUID, it.Link)),
}
if item.Title != "" || item.Link != "" {
f.Items = append(f.Items, item)
}
}
for _, e := range doc.Entries {
link := ""
for _, l := range e.Links {
if l.Rel == "" || l.Rel == "alternate" {
link = strings.TrimSpace(l.Href)
break
}
}
if link == "" && len(e.Links) > 0 {
link = strings.TrimSpace(e.Links[0].Href)
}
item := Item{
Title: PlainText(e.Title),
Link: link,
Summary: PlainText(firstNonEmpty(e.Summary, e.Content)),
Published: parseTime(firstNonEmpty(e.Published, e.Updated)),
ID: strings.TrimSpace(firstNonEmpty(e.ID, link)),
}
if item.Title != "" || item.Link != "" {
f.Items = append(f.Items, item)
}
}
return f, nil
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if strings.TrimSpace(v) != "" {
return v
}
}
return ""
}
// timeLayouts — RFC1123/822 for RSS, RFC3339 for Atom, plus the near-misses
// real feeds ship (no seconds, numeric zone where a name is expected).
var timeLayouts = []string{
time.RFC1123Z,
time.RFC1123,
time.RFC822Z,
time.RFC822,
time.RFC3339,
"2006-01-02T15:04:05Z0700",
"2006-01-02 15:04:05",
"2006-01-02",
"Mon, 02 Jan 2006 15:04:05 -0700",
"Mon, 2 Jan 2006 15:04:05 -0700",
"Mon, 2 Jan 2006 15:04:05 MST",
}
// parseTime returns the zero time on anything it cannot read. An undated item
// is still an item; the poller dedupes by ID, so a missing date costs nothing.
func parseTime(s string) time.Time {
s = strings.TrimSpace(s)
if s == "" {
return time.Time{}
}
for _, l := range timeLayouts {
if t, err := time.Parse(l, s); err == nil {
return t.UTC()
}
}
return time.Time{}
}
var (
// RE2 has no backreferences, so the two tags are spelled out rather than
// captured and matched against themselves.
scriptRE = regexp.MustCompile(`(?is)<script\b[^>]*>.*?</script>|<style\b[^>]*>.*?</style>`)
tagRE = regexp.MustCompile(`(?s)<[^>]*>`)
)
// PlainText strips markup and decodes entities — feed summaries are HTML, and
// what reaches a note (and possibly the TTS) must be text. Exported because the
// crawler's extractor needs exactly this on a bigger input.
func PlainText(s string) string {
s = scriptRE.ReplaceAllString(s, " ")
s = tagRE.ReplaceAllString(s, " ")
s = html.UnescapeString(s)
return strings.TrimSpace(strings.Join(strings.Fields(s), " "))
}
+112
View File
@@ -0,0 +1,112 @@
package rss
import (
"strings"
"testing"
"time"
)
const rss2 = `<?xml version="1.0"?>
<rss version="2.0">
<channel>
<title>Хабр</title>
<item>
<title>Новая уязвимость в ядре</title>
<link>https://example.org/a</link>
<description>&lt;p&gt;Патч уже &lt;b&gt;вышел&lt;/b&gt;.&lt;/p&gt;</description>
<guid>tag:example.org,a</guid>
<pubDate>Mon, 28 Jul 2026 10:00:00 +0000</pubDate>
</item>
<item>
<title>Без даты</title>
<link>https://example.org/b</link>
</item>
</channel>
</rss>`
const atom = `<?xml version="1.0" encoding="utf-8"?>
<feed xmlns="http://www.w3.org/2005/Atom">
<title>Example Atom</title>
<entry>
<title>Release 2.0</title>
<link rel="alternate" href="https://example.com/rel"/>
<link rel="edit" href="https://example.com/edit"/>
<id>urn:uuid:1</id>
<updated>2026-07-30T12:30:00Z</updated>
<summary>Ships &amp; works</summary>
</entry>
</feed>`
func TestParseRSS2(t *testing.T) {
f, err := Parse(strings.NewReader(rss2))
if err != nil {
t.Fatal(err)
}
if f.Title != "Хабр" {
t.Fatalf("title = %q", f.Title)
}
if len(f.Items) != 2 {
t.Fatalf("items = %d, want 2", len(f.Items))
}
it := f.Items[0]
if it.Title != "Новая уязвимость в ядре" {
t.Errorf("title = %q", it.Title)
}
if it.Summary != "Патч уже вышел ." && it.Summary != "Патч уже вышел." {
t.Errorf("summary = %q — tags must be stripped and entities decoded", it.Summary)
}
if it.ID != "tag:example.org,a" {
t.Errorf("id = %q", it.ID)
}
if want := time.Date(2026, 7, 28, 10, 0, 0, 0, time.UTC); !it.Published.Equal(want) {
t.Errorf("published = %v, want %v", it.Published, want)
}
if !f.Items[1].Published.IsZero() {
t.Errorf("undated item got a date: %v", f.Items[1].Published)
}
if f.Items[1].ID != "https://example.org/b" {
t.Errorf("id falls back to the link, got %q", f.Items[1].ID)
}
}
func TestParseAtom(t *testing.T) {
f, err := Parse(strings.NewReader(atom))
if err != nil {
t.Fatal(err)
}
if f.Title != "Example Atom" || len(f.Items) != 1 {
t.Fatalf("feed = %+v", f)
}
it := f.Items[0]
if it.Link != "https://example.com/rel" {
t.Errorf("link = %q, want the alternate link", it.Link)
}
if it.Summary != "Ships & works" {
t.Errorf("summary = %q", it.Summary)
}
if want := time.Date(2026, 7, 30, 12, 30, 0, 0, time.UTC); !it.Published.Equal(want) {
t.Errorf("published = %v, want %v", it.Published, want)
}
}
func TestParseGarbage(t *testing.T) {
if _, err := Parse(strings.NewReader("<html><body>not a feed")); err == nil {
t.Fatal("want an error on a non-feed document")
}
// A feed with an item that has neither title nor link contributes nothing
// rather than an empty note.
f, err := Parse(strings.NewReader(`<rss><channel><item><description>x</description></item></channel></rss>`))
if err != nil {
t.Fatal(err)
}
if len(f.Items) != 0 {
t.Fatalf("items = %d, want 0", len(f.Items))
}
}
func TestPlainTextDropsScript(t *testing.T) {
got := PlainText(`<p>hi</p><script>alert("x")</script><style>b{}</style> there`)
if got != "hi there" {
t.Fatalf("got %q", got)
}
}
+324
View File
@@ -0,0 +1,324 @@
package rss
import (
"context"
"fmt"
"log"
"strings"
"time"
)
// FeedConfig — one feed to read. A feed with no Name or no URL is ignored.
type FeedConfig struct {
Name string // short id; the note source is "rss:<Name>"
URL string // http(s) only, enforced by the fetcher
Category string // free text ("технологии"), used to answer "что по X?"
Interval time.Duration // 0 ⇒ the poller's default
Include []string // when non-empty, keep only items matching one of these
Exclude []string // drop items matching any of these, even if included
}
// Fetcher is the guarded HTTP door (internal/webfetch). An interface so the
// poller is testable without a network and so it CANNOT fetch by any other
// means: no http.Client is constructed in this package.
type Fetcher interface {
Get(ctx context.Context, url string) (*Body, error)
}
// Body is the minimum the poller needs from a response.
type Body struct{ Bytes []byte }
// Notes is core's note-writing half. Same shape as ipc.CoreAPI's method, so the
// daemon passes its API straight in.
type Notes interface {
WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error)
}
// Marks remembers how far a feed was read. Durable, because the alternative is
// re-writing yesterday's headlines as fresh notes after every restart. The
// daemon backs this with config facts (key "rss:latest:<feed>").
type Marks interface {
LastMark(ctx context.Context, feed string) (time.Time, error)
SetMark(ctx context.Context, feed string, at time.Time) error
}
// Embedder embeds a note on its way into the store so recall can find it. nil ⇒
// notes are written without a vector (still readable by the recent-notes path).
type Embedder interface {
Embed(ctx context.Context, text string) ([]float32, error)
}
// Ranker is the relevance seam. The plan called for scoring each item against
// an interest profile built from his notes; that profile does not exist yet, and
// a threshold over an embedder with no profile to compare to is a random filter
// with a confident name. So the seam is here, nil in the daemon, and the filter
// that actually runs is the per-feed keyword one — a rule he can read and
// predict. When there IS a profile, implement this and pass it.
//
// Note what a Ranker must NOT be: anything that sends his notes outward. The
// scoring happens locally against a local embedder; the feed item is the input,
// his memory is never the payload.
type Ranker interface {
Relevant(ctx context.Context, text string) (bool, error)
}
// Config — poller-wide settings.
type Config struct {
DefaultInterval time.Duration // 0 ⇒ DefaultPollInterval
MaxItems int // most notes written per feed per poll; 0 ⇒ DefaultMaxItems
MaxAge time.Duration // ignore items older than this on a cold start; 0 ⇒ DefaultMaxAge
}
// Defaults chosen to be quiet: a feed read every half hour, at most a handful of
// items kept, and a cold start that does not import a month of history.
const (
DefaultPollInterval = 30 * time.Minute
DefaultMaxItems = 5
DefaultMaxAge = 24 * time.Hour
)
// Poller reads feeds on a schedule and writes what survives filtering as notes.
type Poller struct {
feeds []FeedConfig
fetch Fetcher
notes Notes
marks Marks
embed Embedder
ranker Ranker
cfg Config
nextDue map[string]time.Time
seen map[string]map[string]bool // feed → item ID, for items with no date
}
// NewPoller wires a poller. Returns nil when there is nothing to poll — a
// capability is off unless configured, and callers check for nil.
func NewPoller(feeds []FeedConfig, fetch Fetcher, notes Notes, marks Marks, embed Embedder, ranker Ranker, cfg Config) *Poller {
var valid []FeedConfig
for _, f := range feeds {
if strings.TrimSpace(f.Name) == "" || strings.TrimSpace(f.URL) == "" {
log.Printf("rss: skipping a feed with no name or no url")
continue
}
valid = append(valid, f)
}
if len(valid) == 0 || fetch == nil || notes == nil {
return nil
}
if cfg.DefaultInterval <= 0 {
cfg.DefaultInterval = DefaultPollInterval
}
if cfg.MaxItems <= 0 {
cfg.MaxItems = DefaultMaxItems
}
if cfg.MaxAge <= 0 {
cfg.MaxAge = DefaultMaxAge
}
return &Poller{
feeds: valid, fetch: fetch, notes: notes, marks: marks,
embed: embed, ranker: ranker, cfg: cfg,
nextDue: map[string]time.Time{},
seen: map[string]map[string]bool{},
}
}
// Feeds returns the configured feeds (the answer path lists categories).
func (p *Poller) Feeds() []FeedConfig { return p.feeds }
// PollDue reads every feed whose interval has elapsed and returns how many
// notes were written. Errors are logged per feed, never returned: one dead feed
// must not stop the others, and there is nobody waiting on this.
func (p *Poller) PollDue(ctx context.Context, now time.Time) int {
written := 0
for _, f := range p.feeds {
if due, ok := p.nextDue[f.Name]; ok && now.Before(due) {
continue
}
interval := f.Interval
if interval <= 0 {
interval = p.cfg.DefaultInterval
}
p.nextDue[f.Name] = now.Add(interval)
n, err := p.PollFeed(ctx, f, now)
if err != nil {
// The URL is configured by him and not a secret, so it is loggable;
// item titles are not logged, only counts.
log.Printf("rss: feed %s: %v", f.Name, err)
continue
}
if n > 0 {
log.Printf("rss: feed %s: %d new item(s) noted", f.Name, n)
}
written += n
}
return written
}
// PollFeed reads one feed now, regardless of its schedule.
func (p *Poller) PollFeed(ctx context.Context, f FeedConfig, now time.Time) (int, error) {
body, err := p.fetch.Get(ctx, f.URL)
if err != nil {
return 0, err
}
feed, err := Parse(strings.NewReader(string(body.Bytes)))
if err != nil {
return 0, err
}
mark := p.mark(ctx, f.Name, now)
newest := mark
written := 0
for _, it := range feed.Items {
if written >= p.cfg.MaxItems {
break
}
if !p.fresh(f, it, mark, now) {
continue
}
if !Matches(f, it) {
continue
}
if p.ranker != nil {
ok, err := p.ranker.Relevant(ctx, it.Title+" "+it.Summary)
if err != nil {
log.Printf("rss: feed %s: relevance: %v", f.Name, err)
} else if !ok {
continue
}
}
if err := p.write(ctx, f, it, now); err != nil {
return written, err
}
written++
if it.Published.After(newest) {
newest = it.Published
}
}
if p.marks != nil && newest.After(mark) {
if err := p.marks.SetMark(ctx, f.Name, newest); err != nil {
log.Printf("rss: feed %s: save mark: %v", f.Name, err)
}
}
return written, nil
}
// mark — how far this feed was read. A feed with no mark starts MaxAge ago, so
// a first poll takes today's headlines instead of the whole archive.
func (p *Poller) mark(ctx context.Context, feed string, now time.Time) time.Time {
cold := now.Add(-p.cfg.MaxAge)
if p.marks == nil {
return cold
}
at, err := p.marks.LastMark(ctx, feed)
if err != nil || at.IsZero() {
return cold
}
return at
}
// fresh — two dedup rules, because feeds are inconsistent about dates. A dated
// item must be newer than the mark; an undated one is kept once per process by
// ID. Both are needed: dates alone re-import undated feeds forever, IDs alone
// lose their memory on restart.
func (p *Poller) fresh(f FeedConfig, it Item, mark, now time.Time) bool {
if !it.Published.IsZero() {
if !it.Published.After(mark) {
return false
}
// A feed that dates its items in the future (or a clock skew) must not
// win the mark and mute everything after it.
return !it.Published.After(now.Add(time.Hour))
}
id := it.ID
if id == "" {
id = it.Title
}
if p.seen[f.Name] == nil {
p.seen[f.Name] = map[string]bool{}
}
if p.seen[f.Name][id] {
return false
}
p.seen[f.Name][id] = true
return true
}
// write stores one item as a note. Source "rss:<feed>" is what the answer path
// filters on, and what makes a feed note distinguishable from something he said.
func (p *Poller) write(ctx context.Context, f FeedConfig, it Item, now time.Time) error {
text := NoteText(f, it)
var vec []float32
if p.embed != nil {
v, err := p.embed.Embed(ctx, text)
if err != nil {
log.Printf("rss: feed %s: embed: %v", f.Name, err)
} else {
vec = v
}
}
ts := it.Published
if ts.IsZero() {
ts = now
}
if _, err := p.notes.WriteNote(ctx, ts, text, vec, SourceFor(f.Name)); err != nil {
return fmt.Errorf("write note: %w", err)
}
return nil
}
// SourceFor is the note source for a feed.
func SourceFor(feed string) string { return "rss:" + feed }
// SourcePrefix — what the answer path matches to find feed notes.
const SourcePrefix = "rss:"
// NoteText renders an item as the note body. The category is included because
// "что нового по технологиям?" is answered by reading notes, and a note has to
// carry enough to be recognised as belonging to that category.
func NoteText(f FeedConfig, it Item) string {
var b strings.Builder
b.WriteString(it.Title)
if f.Category != "" {
fmt.Fprintf(&b, " [%s]", f.Category)
}
if it.Summary != "" {
b.WriteString("\n")
b.WriteString(trimRunes(it.Summary, 500))
}
if it.Link != "" {
b.WriteString("\n")
b.WriteString(it.Link)
}
return b.String()
}
// trimRunes cuts on a rune boundary — a note is Russian as often as English and
// half a cyrillic letter is a broken note.
func trimRunes(s string, max int) string {
r := []rune(s)
if len(r) <= max {
return s
}
return strings.TrimSpace(string(r[:max])) + "…"
}
// Matches applies the per-feed keyword filter: keep when Include is empty or one
// include matches, drop when any exclude matches. Case-insensitive substring,
// which for Russian is the honest choice — no stemmer here, so "выборы" does not
// match "выборах", and a filter he writes is a filter he can predict.
func Matches(f FeedConfig, it Item) bool {
hay := strings.ToLower(it.Title + " " + it.Summary)
for _, x := range f.Exclude {
if x = strings.ToLower(strings.TrimSpace(x)); x != "" && strings.Contains(hay, x) {
return false
}
}
if len(f.Include) == 0 {
return true
}
for _, in := range f.Include {
if in = strings.ToLower(strings.TrimSpace(in)); in != "" && strings.Contains(hay, in) {
return true
}
}
return false
}
+210
View File
@@ -0,0 +1,210 @@
package rss
import (
"context"
"errors"
"strings"
"testing"
"time"
)
type fakeFetch struct {
body string
err error
calls int
urls []string
}
func (f *fakeFetch) Get(_ context.Context, url string) (*Body, error) {
f.calls++
f.urls = append(f.urls, url)
if f.err != nil {
return nil, f.err
}
return &Body{Bytes: []byte(f.body)}, nil
}
type writtenNote struct {
ts time.Time
text string
source string
vec []float32
}
type fakeNotes struct{ notes []writtenNote }
func (n *fakeNotes) WriteNote(_ context.Context, ts time.Time, text string, vec []float32, source string) (int64, error) {
n.notes = append(n.notes, writtenNote{ts, text, source, vec})
return int64(len(n.notes)), nil
}
type fakeMarks struct{ m map[string]time.Time }
func newMarks() *fakeMarks { return &fakeMarks{m: map[string]time.Time{}} }
func (f *fakeMarks) LastMark(_ context.Context, feed string) (time.Time, error) {
return f.m[feed], nil
}
func (f *fakeMarks) SetMark(_ context.Context, feed string, at time.Time) error {
f.m[feed] = at
return nil
}
var now = time.Date(2026, 7, 28, 12, 0, 0, 0, time.UTC)
func TestPollWritesNotesWithSource(t *testing.T) {
fetch := &fakeFetch{body: rss2}
notes := &fakeNotes{}
marks := newMarks()
p := NewPoller([]FeedConfig{{Name: "habr", URL: "https://example.org/rss", Category: "технологии"}},
fetch, notes, marks, nil, nil, Config{})
if p == nil {
t.Fatal("NewPoller returned nil for a configured feed")
}
n := p.PollDue(context.Background(), now)
if n != 2 || len(notes.notes) != 2 {
t.Fatalf("wrote %d notes (returned %d), want 2", len(notes.notes), n)
}
if notes.notes[0].source != "rss:habr" {
t.Errorf("source = %q, want rss:habr", notes.notes[0].source)
}
if !strings.Contains(notes.notes[0].text, "технологии") {
t.Errorf("note does not carry its category: %q", notes.notes[0].text)
}
if !strings.Contains(notes.notes[0].text, "https://example.org/a") {
t.Errorf("note does not carry its link: %q", notes.notes[0].text)
}
// The undated item is stamped with now, the dated one with its own date.
if !notes.notes[1].ts.Equal(now) {
t.Errorf("undated item ts = %v, want now", notes.notes[1].ts)
}
}
// The whole point of a mark: polling twice must not re-note the same headlines.
func TestSecondPollIsQuiet(t *testing.T) {
fetch := &fakeFetch{body: rss2}
notes := &fakeNotes{}
p := NewPoller([]FeedConfig{{Name: "habr", URL: "u", Interval: time.Minute}}, fetch, notes, newMarks(), nil, nil, Config{})
p.PollDue(context.Background(), now)
before := len(notes.notes)
p.PollDue(context.Background(), now.Add(2*time.Minute))
if len(notes.notes) != before {
t.Fatalf("second poll wrote %d extra notes", len(notes.notes)-before)
}
}
// A mark that survives a restart is the durable half; simulate one by building a
// fresh poller over the same marks.
func TestMarkSurvivesRestart(t *testing.T) {
marks := newMarks()
fetch := &fakeFetch{body: rss2}
notes := &fakeNotes{}
feeds := []FeedConfig{{Name: "habr", URL: "u"}}
NewPoller(feeds, fetch, notes, marks, nil, nil, Config{}).PollDue(context.Background(), now)
if len(notes.notes) != 2 {
t.Fatalf("first run wrote %d", len(notes.notes))
}
notes2 := &fakeNotes{}
NewPoller(feeds, fetch, notes2, marks, nil, nil, Config{}).PollDue(context.Background(), now.Add(time.Hour))
// The dated item is behind the mark. The undated one has no date to compare,
// so it comes back — accepted and documented in fresh(): an undated feed is
// deduped per process, not forever.
for _, n := range notes2.notes {
if strings.Contains(n.text, "уязвимость") {
t.Fatalf("dated item re-noted after restart: %q", n.text)
}
}
}
func TestIntervalIsRespected(t *testing.T) {
fetch := &fakeFetch{body: rss2}
p := NewPoller([]FeedConfig{{Name: "habr", URL: "u", Interval: time.Hour}}, fetch, &fakeNotes{}, newMarks(), nil, nil, Config{})
p.PollDue(context.Background(), now)
p.PollDue(context.Background(), now.Add(time.Minute))
if fetch.calls != 1 {
t.Fatalf("fetched %d times inside one interval, want 1", fetch.calls)
}
p.PollDue(context.Background(), now.Add(2*time.Hour))
if fetch.calls != 2 {
t.Fatalf("fetched %d times, want 2 after the interval elapsed", fetch.calls)
}
}
func TestColdStartIgnoresOldItems(t *testing.T) {
old := `<rss><channel><item><title>Старое</title><link>l</link>` +
`<pubDate>Mon, 01 Jun 2026 10:00:00 +0000</pubDate></item></channel></rss>`
notes := &fakeNotes{}
p := NewPoller([]FeedConfig{{Name: "f", URL: "u"}}, &fakeFetch{body: old}, notes, newMarks(), nil, nil, Config{MaxAge: 24 * time.Hour})
if n := p.PollDue(context.Background(), now); n != 0 {
t.Fatalf("cold start imported %d old items, want 0", n)
}
}
func TestMaxItemsCap(t *testing.T) {
var b strings.Builder
b.WriteString("<rss><channel>")
for i := 0; i < 10; i++ {
b.WriteString("<item><title>t")
b.WriteByte(byte('0' + i))
b.WriteString("</title><link>https://example.org/")
b.WriteByte(byte('0' + i))
b.WriteString("</link></item>")
}
b.WriteString("</channel></rss>")
notes := &fakeNotes{}
p := NewPoller([]FeedConfig{{Name: "f", URL: "u"}}, &fakeFetch{body: b.String()}, notes, newMarks(), nil, nil, Config{MaxItems: 3})
if n := p.PollDue(context.Background(), now); n != 3 {
t.Fatalf("wrote %d notes, want the cap of 3", n)
}
}
func TestKeywordFilter(t *testing.T) {
f := FeedConfig{Include: []string{"ядр"}, Exclude: []string{"реклама"}}
if !Matches(f, Item{Title: "Новое ядро"}) {
t.Error("include did not match")
}
if Matches(f, Item{Title: "Новое ядро", Summary: "Реклама внутри"}) {
t.Error("exclude must win over include")
}
if Matches(f, Item{Title: "Погода"}) {
t.Error("non-matching item passed the include filter")
}
if !Matches(FeedConfig{}, Item{Title: "что угодно"}) {
t.Error("an unfiltered feed must keep everything")
}
}
type fakeRanker struct{ keep bool }
func (r fakeRanker) Relevant(context.Context, string) (bool, error) { return r.keep, nil }
func TestRankerCanDropEverything(t *testing.T) {
notes := &fakeNotes{}
p := NewPoller([]FeedConfig{{Name: "f", URL: "u"}}, &fakeFetch{body: rss2}, notes, newMarks(), nil, fakeRanker{false}, Config{})
if n := p.PollDue(context.Background(), now); n != 0 {
t.Fatalf("ranker rejected everything but %d notes were written", n)
}
}
func TestFetchErrorIsSurvivable(t *testing.T) {
notes := &fakeNotes{}
p := NewPoller([]FeedConfig{
{Name: "dead", URL: "u1"},
{Name: "live", URL: "u2"},
}, &fakeFetch{err: errors.New("boom")}, notes, newMarks(), nil, nil, Config{})
if n := p.PollDue(context.Background(), now); n != 0 {
t.Fatalf("n = %d", n)
}
// Both feeds were attempted: one dead feed does not abort the round.
if p.nextDue["live"].IsZero() {
t.Fatal("the second feed was never attempted")
}
}
func TestNoFeedsMeansNoPoller(t *testing.T) {
if p := NewPoller(nil, &fakeFetch{}, &fakeNotes{}, nil, nil, nil, Config{}); p != nil {
t.Fatal("NewPoller must return nil when nothing is configured")
}
if p := NewPoller([]FeedConfig{{Name: "", URL: ""}}, &fakeFetch{}, &fakeNotes{}, nil, nil, nil, Config{}); p != nil {
t.Fatal("a feed with no name or url is not a configuration")
}
}
+301
View File
@@ -0,0 +1,301 @@
// Package webfetch is the one door Maven uses to read something off the
// network, and it is a narrow one.
//
// "Never phones home" stopped being a hard constraint on 2026-07-31, but what
// replaced it is not "she may fetch anything": local sources come first (Kiwix
// on the box), external fetching is off unless configured, and only the
// utterance ever leaves — never his notes, facts or history. That policy is
// enforced by the callers. What THIS package enforces is the part that must be
// code rather than a paragraph in a plan, because it protects the homelab from
// its own assistant:
//
// - http/https only — no file://, no ftp://, no gopher;
// - no private address, ever: loopback, RFC1918 (which is what makes the
// 10.42.0.0/24 wireguard tunnel and the 192.168.1.0/24 LAN unreachable),
// link-local incl. the 169.254.169.254 cloud metadata address, CGNAT,
// unique-local v6. Checked in the dialer's Control hook, so it holds for
// every address the resolver returns AND for every hop of a redirect
// chain — a DNS name that resolves to 127.0.0.1 is refused at connect
// time, which a pre-flight lookup could not promise (rebinding);
// - an allowlist, when one is configured, and a denylist that always wins;
// - a response size cap, a total timeout, a redirect cap;
// - one request per host per interval, so a poll loop with a bug is slow
// rather than an outbound flood.
//
// Everything above is on by default with sane numbers: a zero Config is a
// usable, conservative fetcher. There is no cache and no retry — a feed poll
// or a page read that fails is simply not answered this round.
package webfetch
import (
"context"
"errors"
"fmt"
"io"
"net"
"net/http"
"net/url"
"strings"
"sync"
"syscall"
"time"
)
// Defaults. Small on purpose: this reads feeds and article pages, not ISOs.
const (
DefaultTimeout = 20 * time.Second
DefaultMaxBytes = 2 << 20 // 2 MiB
DefaultMaxRedirects = 3
DefaultHostInterval = time.Second
DefaultUserAgent = "Maven/1.0 (self-hosted personal assistant)"
)
// Errors callers distinguish. Everything else is wrapped transport error.
var (
ErrScheme = errors.New("webfetch: only http and https are allowed")
ErrBlocked = errors.New("webfetch: host is not allowed")
ErrPrivate = errors.New("webfetch: refusing to connect to a private address")
ErrTooLarge = errors.New("webfetch: response exceeds the size cap")
ErrRedirects = errors.New("webfetch: too many redirects")
ErrStatus = errors.New("webfetch: non-2xx status")
)
// Config are the limits. Every zero value means "the default above", so
// Config{} is safe; the only field that changes behaviour by being empty is
// AllowHosts (empty ⇒ any public host that is not denied).
type Config struct {
// AllowHosts — when non-empty, the ONLY hosts that may be fetched. An
// entry matches the host itself and its subdomains ("example.com" allows
// "news.example.com"). This is the knob to reach for when a capability
// should read two feeds and nothing else.
AllowHosts []string
// DenyHosts — same matching, checked first and always winning.
DenyHosts []string
Timeout time.Duration // whole request, including redirects and body read
MaxBytes int64 // response body cap
MaxRedirects int // 0 ⇒ default; negative ⇒ no redirects followed
HostInterval time.Duration // minimum spacing between requests to one host
UserAgent string
// AllowPrivate disables the private-address guard. It exists for tests
// (httptest listens on 127.0.0.1) and for an explicitly configured
// on-box mirror. Nothing in deploy/mavend.json sets it, and it should
// stay that way: with it on, any URL Maven is handed becomes an SSRF
// probe of the LAN and the wireguard range.
AllowPrivate bool
}
// Response is a fetched body, already bounded by MaxBytes.
type Response struct {
URL string // final URL after redirects
Status int
ContentType string
Body []byte
}
// Fetcher performs guarded GETs. Safe for concurrent use; the per-host rate
// limiter is shared, which is the point of sharing one Fetcher.
type Fetcher struct {
cfg Config
http *http.Client
mu sync.Mutex
last map[string]time.Time // host → when we last dialed it
}
// New builds a fetcher from cfg, filling in defaults.
func New(cfg Config) *Fetcher {
if cfg.Timeout <= 0 {
cfg.Timeout = DefaultTimeout
}
if cfg.MaxBytes <= 0 {
cfg.MaxBytes = DefaultMaxBytes
}
if cfg.MaxRedirects == 0 {
cfg.MaxRedirects = DefaultMaxRedirects
}
if cfg.HostInterval <= 0 {
cfg.HostInterval = DefaultHostInterval
}
if cfg.UserAgent == "" {
cfg.UserAgent = DefaultUserAgent
}
f := &Fetcher{cfg: cfg, last: map[string]time.Time{}}
dialer := &net.Dialer{Timeout: 10 * time.Second}
if !cfg.AllowPrivate {
// The guard lives here rather than in a pre-flight net.LookupHost so
// that it sees the address actually being connected to: every A/AAAA
// the resolver handed back, on every redirect hop, with no window in
// which the name could be re-pointed at the LAN.
dialer.Control = func(_, address string, _ syscall.RawConn) error {
host, _, err := net.SplitHostPort(address)
if err != nil {
return err
}
ip := net.ParseIP(host)
if ip == nil || IsPrivateIP(ip) {
return fmt.Errorf("%w: %s", ErrPrivate, host)
}
return nil
}
}
f.http = &http.Client{
Timeout: cfg.Timeout,
Transport: &http.Transport{DialContext: dialer.DialContext},
CheckRedirect: func(req *http.Request, via []*http.Request) error {
if len(via) > f.cfg.MaxRedirects {
return ErrRedirects
}
// A redirect is a fresh URL and gets the full check: an allowed
// host must not be able to bounce us onto a denied one.
return f.checkURL(req.URL)
},
}
return f
}
// Get fetches rawURL. The body is capped: a larger response is an error, not a
// truncation, because half an XML document is worse than none.
func (f *Fetcher) Get(ctx context.Context, rawURL string) (*Response, error) {
u, err := url.Parse(strings.TrimSpace(rawURL))
if err != nil {
return nil, fmt.Errorf("webfetch: bad url %q: %w", rawURL, err)
}
if err := f.checkURL(u); err != nil {
return nil, err
}
if err := f.waitTurn(ctx, u.Hostname()); err != nil {
return nil, err
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, u.String(), nil)
if err != nil {
return nil, err
}
req.Header.Set("User-Agent", f.cfg.UserAgent)
req.Header.Set("Accept-Encoding", "identity")
resp, err := f.http.Do(req)
if err != nil {
// http.Client wraps our sentinels in *url.Error; unwrap so callers can
// still tell "blocked" from "the network is down".
for _, sentinel := range []error{ErrPrivate, ErrBlocked, ErrRedirects, ErrScheme} {
if errors.Is(err, sentinel) {
return nil, err
}
}
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, f.cfg.MaxBytes+1))
if err != nil {
return nil, err
}
if int64(len(body)) > f.cfg.MaxBytes {
return nil, fmt.Errorf("%w (%d bytes)", ErrTooLarge, f.cfg.MaxBytes)
}
if resp.StatusCode < 200 || resp.StatusCode > 299 {
return nil, fmt.Errorf("%w: %d", ErrStatus, resp.StatusCode)
}
return &Response{
URL: resp.Request.URL.String(),
Status: resp.StatusCode,
ContentType: resp.Header.Get("Content-Type"),
Body: body,
}, nil
}
// checkURL applies the scheme rule and the host lists. The address rule is the
// dialer's job (see New).
func (f *Fetcher) checkURL(u *url.URL) error {
switch u.Scheme {
case "http", "https":
default:
return fmt.Errorf("%w: %q", ErrScheme, u.Scheme)
}
host := strings.ToLower(u.Hostname())
if host == "" {
return fmt.Errorf("%w: no host", ErrBlocked)
}
if HostMatches(host, f.cfg.DenyHosts) {
return fmt.Errorf("%w: %s is denied", ErrBlocked, host)
}
if len(f.cfg.AllowHosts) > 0 && !HostMatches(host, f.cfg.AllowHosts) {
return fmt.Errorf("%w: %s is not on the allowlist", ErrBlocked, host)
}
// A literal private address is refused here as well as in the dialer, so
// the error is the specific one even when no connection is attempted.
if !f.cfg.AllowPrivate {
if ip := net.ParseIP(host); ip != nil && IsPrivateIP(ip) {
return fmt.Errorf("%w: %s", ErrPrivate, host)
}
}
return nil
}
// waitTurn blocks until this host's rate-limit interval has elapsed. It holds
// no lock while sleeping, so two hosts never wait on each other.
func (f *Fetcher) waitTurn(ctx context.Context, host string) error {
for {
f.mu.Lock()
now := time.Now()
earliest := f.last[host].Add(f.cfg.HostInterval)
if !now.Before(earliest) {
f.last[host] = now
f.mu.Unlock()
return nil
}
f.mu.Unlock()
wait := time.NewTimer(earliest.Sub(now))
select {
case <-ctx.Done():
wait.Stop()
return ctx.Err()
case <-wait.C:
}
}
}
// HostMatches reports whether host equals one of pats or is a subdomain of one.
// Exported because the crawler applies the same rule to links it decides not to
// follow, before it ever builds a request.
func HostMatches(host string, pats []string) bool {
host = strings.ToLower(strings.TrimSuffix(host, "."))
for _, p := range pats {
p = strings.ToLower(strings.TrimSpace(strings.TrimPrefix(p, "*.")))
if p == "" {
continue
}
if host == p || strings.HasSuffix(host, "."+p) {
return true
}
}
return false
}
// cgnat is 100.64.0.0/10 — carrier NAT, not covered by net.IP's helpers and not
// somewhere a personal assistant has business connecting.
var cgnat = &net.IPNet{IP: net.IPv4(100, 64, 0, 0).To4(), Mask: net.CIDRMask(10, 32)}
// IsPrivateIP reports whether ip is somewhere Maven must never reach out to:
// the box itself, the LAN, the wireguard range (10.42.0.0/24 ⊂ 10/8), the cloud
// metadata address (169.254.169.254 ⊂ link-local), or anything unroutable.
func IsPrivateIP(ip net.IP) bool {
if ip.IsLoopback() || ip.IsPrivate() || ip.IsUnspecified() ||
ip.IsLinkLocalUnicast() || ip.IsLinkLocalMulticast() ||
ip.IsInterfaceLocalMulticast() || ip.IsMulticast() {
return true
}
if v4 := ip.To4(); v4 != nil && cgnat.Contains(v4) {
return true
}
// IPv4-mapped/compatible forms of the above are handled by To4() inside the
// stdlib helpers; what is left is v6 unique-local (fc00::/7).
if len(ip) == net.IPv6len && ip.To4() == nil && ip[0]&0xfe == 0xfc {
return true
}
return false
}
+227
View File
@@ -0,0 +1,227 @@
package webfetch
import (
"context"
"errors"
"net"
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
// The limits in this package are the reason a crawler is allowed to exist on
// this box at all, so each one has a test that fails loudly if it is removed.
func TestPrivateAddressesAreRefused(t *testing.T) {
// The wireguard range (10.42.0.0/24), the LAN (192.168.1.0/24) and the
// cloud metadata address are the three that matter here; the rest come
// along for free.
for _, s := range []string{
"127.0.0.1", "127.1.2.3", "10.42.0.7", "10.0.0.5", "192.168.1.104",
"172.16.4.4", "169.254.169.254", "100.64.1.1", "0.0.0.0",
"::1", "fc00::1", "fd12:3456::1", "fe80::1",
} {
if !IsPrivateIP(net.ParseIP(s)) {
t.Errorf("IsPrivateIP(%s) = false, want true", s)
}
}
for _, s := range []string{"8.8.8.8", "1.1.1.1", "93.184.216.34", "2606:2800:220:1::1"} {
if IsPrivateIP(net.ParseIP(s)) {
t.Errorf("IsPrivateIP(%s) = true, want false", s)
}
}
}
func TestGetRefusesPrivateLiteral(t *testing.T) {
f := New(Config{})
for _, u := range []string{
"http://127.0.0.1:8034/search",
"http://10.42.0.1/",
"http://192.168.1.104/dash",
"http://[::1]:9100/mcp",
} {
if _, err := f.Get(context.Background(), u); !errors.Is(err, ErrPrivate) {
t.Errorf("Get(%s) error = %v, want ErrPrivate", u, err)
}
}
}
// A hostname that resolves into private space must fail too — that is the
// rebinding case, and it is why the check lives in the dialer.
func TestGetRefusesPrivateResolution(t *testing.T) {
f := New(Config{})
if _, err := f.Get(context.Background(), "http://localhost:8034/"); !errors.Is(err, ErrPrivate) {
t.Fatalf("Get(localhost) error = %v, want ErrPrivate", err)
}
}
func TestGetRefusesNonHTTPSchemes(t *testing.T) {
f := New(Config{})
for _, u := range []string{"file:///etc/passwd", "ftp://example.com/x", "gopher://example.com"} {
if _, err := f.Get(context.Background(), u); !errors.Is(err, ErrScheme) {
t.Errorf("Get(%s) error = %v, want ErrScheme", u, err)
}
}
}
// testFetcher — a fetcher pointed at an httptest server, which necessarily
// listens on loopback. AllowPrivate is the test-only escape hatch.
func testFetcher(t *testing.T, cfg Config) *Fetcher {
t.Helper()
cfg.AllowPrivate = true
if cfg.HostInterval == 0 {
cfg.HostInterval = time.Nanosecond
}
return New(cfg)
}
func TestAllowAndDenyLists(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("ok"))
}))
defer srv.Close()
f := testFetcher(t, Config{AllowHosts: []string{"example.com"}})
if _, err := f.Get(context.Background(), srv.URL); !errors.Is(err, ErrBlocked) {
t.Fatalf("off-allowlist host: error = %v, want ErrBlocked", err)
}
f = testFetcher(t, Config{DenyHosts: []string{"127.0.0.1"}})
if _, err := f.Get(context.Background(), srv.URL); !errors.Is(err, ErrBlocked) {
t.Fatalf("denied host: error = %v, want ErrBlocked", err)
}
f = testFetcher(t, Config{AllowHosts: []string{"127.0.0.1"}})
if _, err := f.Get(context.Background(), srv.URL); err != nil {
t.Fatalf("allowlisted host: %v", err)
}
}
func TestHostMatchesSubdomains(t *testing.T) {
pats := []string{"example.com", "*.news.org"}
for _, h := range []string{"example.com", "news.example.com", "a.b.example.com", "news.org", "feeds.news.org"} {
if !HostMatches(h, pats) {
t.Errorf("HostMatches(%q) = false, want true", h)
}
}
for _, h := range []string{"notexample.com", "example.com.evil.net", "org"} {
if HostMatches(h, pats) {
t.Errorf("HostMatches(%q) = true, want false", h)
}
}
}
func TestSizeCap(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte(strings.Repeat("x", 5000)))
}))
defer srv.Close()
f := testFetcher(t, Config{MaxBytes: 100})
if _, err := f.Get(context.Background(), srv.URL); !errors.Is(err, ErrTooLarge) {
t.Fatalf("error = %v, want ErrTooLarge", err)
}
f = testFetcher(t, Config{MaxBytes: 6000})
resp, err := f.Get(context.Background(), srv.URL)
if err != nil {
t.Fatalf("under the cap: %v", err)
}
if len(resp.Body) != 5000 {
t.Fatalf("body = %d bytes, want 5000", len(resp.Body))
}
}
func TestRedirectCap(t *testing.T) {
var srv *httptest.Server
srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, srv.URL+"/again", http.StatusFound)
}))
defer srv.Close()
f := testFetcher(t, Config{MaxRedirects: 2})
if _, err := f.Get(context.Background(), srv.URL); !errors.Is(err, ErrRedirects) {
t.Fatalf("error = %v, want ErrRedirects", err)
}
}
// A redirect off the allowlist is the interesting redirect: the first hop is
// permitted, the second must not be.
func TestRedirectRecheckedAgainstDenylist(t *testing.T) {
target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("secret"))
}))
defer target.Close()
hop := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Redirect(w, r, target.URL, http.StatusFound)
}))
defer hop.Close()
// Reach the hop under the name "localhost" and allow only that name; the
// redirect lands on the same box under its literal address, which the
// allowlist does not cover. Without the CheckRedirect hook this fetch
// succeeds and returns "secret".
f := testFetcher(t, Config{AllowHosts: []string{"localhost"}})
viaName := strings.Replace(hop.URL, "127.0.0.1", "localhost", 1)
if _, err := f.Get(context.Background(), viaName); !errors.Is(err, ErrBlocked) {
t.Fatalf("error = %v, want ErrBlocked", err)
}
}
func TestPerHostRateLimit(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Write([]byte("ok"))
}))
defer srv.Close()
f := testFetcher(t, Config{HostInterval: 60 * time.Millisecond})
start := time.Now()
for i := 0; i < 3; i++ {
if _, err := f.Get(context.Background(), srv.URL); err != nil {
t.Fatalf("request %d: %v", i, err)
}
}
if elapsed := time.Since(start); elapsed < 120*time.Millisecond {
t.Fatalf("three requests took %s, want at least 120ms of spacing", elapsed)
}
}
func TestRateLimitHonoursContext(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}))
defer srv.Close()
f := testFetcher(t, Config{HostInterval: 10 * time.Second})
if _, err := f.Get(context.Background(), srv.URL); err != nil {
t.Fatalf("first request: %v", err)
}
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
defer cancel()
if _, err := f.Get(ctx, srv.URL); !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("error = %v, want DeadlineExceeded", err)
}
}
func TestNon2xxIsAnError(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
http.Error(w, "nope", http.StatusInternalServerError)
}))
defer srv.Close()
f := testFetcher(t, Config{})
if _, err := f.Get(context.Background(), srv.URL); !errors.Is(err, ErrStatus) {
t.Fatalf("error = %v, want ErrStatus", err)
}
}
func TestUserAgentIsSent(t *testing.T) {
got := make(chan string, 1)
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
got <- r.Header.Get("User-Agent")
}))
defer srv.Close()
f := testFetcher(t, Config{UserAgent: "Maven/test"})
if _, err := f.Get(context.Background(), srv.URL); err != nil {
t.Fatal(err)
}
if ua := <-got; ua != "Maven/test" {
t.Fatalf("user-agent = %q", ua)
}
}