Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 39d44bb384 | |||
| 65ee0f9c61 | |||
| 76938e206d | |||
| 0b3d81ecbf | |||
| 4be6852b94 | |||
| f7b76c572f | |||
| 05ddc5c92e | |||
| b55e68f98d | |||
| beaa24754c | |||
| aed8cac439 |
@@ -101,9 +101,15 @@ protocol; the config in `deploy/mavend.json` (with `${VAR}` env expansion from g
|
|||||||
Count against compose, not against the table. Four of the nine daemons are absent, and each
|
Count against compose, not against the table. Four of the nine daemons are absent, and each
|
||||||
absence has a different reason.
|
absence has a different reason.
|
||||||
|
|
||||||
`mavmaild` is commented out in compose, with the reason written beside it: it needs a mail
|
`mavmaild` and `mavcaldav` are commented out in compose, each with the reason written
|
||||||
account and this box has none. `mavcaldav` appears nowhere at all, and unlike the other
|
beside it: the first needs a mail account, the second a CalDAV account, and this box has
|
||||||
three that is an oversight rather than a decision (V-644).
|
neither. `mavcaldav` used to appear nowhere at all, which was an oversight; it became a
|
||||||
|
recorded decision on 07-08-2026 (V-644). Two things ride on that absence and the block
|
||||||
|
names them. Agenda questions route to `IntentQuery` at stage 0 (V-498) and the `calendar`
|
||||||
|
query source then reads a table nobody writes. And `loop.State.CalendarBusy` is fed by the
|
||||||
|
same facts, so the gate's "do not nag mid-meeting" is permanently false. Its password is
|
||||||
|
read from a file (`-pass-file`, and `-render-pass-file` for the render collection), never
|
||||||
|
taken as a flag value, which is the rule `mavpoll` and `mavmaild` follow too.
|
||||||
|
|
||||||
**`mavwaked` and `mavenclient` are absent by decision, not oversight** (Vikunja #463,
|
**`mavwaked` and `mavenclient` are absent by decision, not oversight** (Vikunja #463,
|
||||||
`docs/plans/17-where-the-voice-loop-runs.md`).
|
`docs/plans/17-where-the-voice-loop-runs.md`).
|
||||||
@@ -454,8 +460,11 @@ start of a session rather than one lookup per first use:
|
|||||||
ToolSearch("select:mcp__vikunja__list_tasks,mcp__vikunja__get_task_details,mcp__vikunja__create_task,mcp__vikunja__update_task")
|
ToolSearch("select:mcp__vikunja__list_tasks,mcp__vikunja__get_task_details,mcp__vikunja__create_task,mcp__vikunja__update_task")
|
||||||
```
|
```
|
||||||
|
|
||||||
`update_task` carrying a `description` resets `done` to false, so closing a task with a
|
**Close a finished task with `done: true` and nothing else** (owner's call, 07-08-2026).
|
||||||
write-up takes two calls: the description, then `done: true`.
|
Do not write a completion summary into the description on the way out. It is lost anyway,
|
||||||
|
and the durable record is the commit messages and the merged PR. Note that `update_task`
|
||||||
|
carrying a `description` resets `done` to false, which is why a write-up ever took two
|
||||||
|
calls.
|
||||||
|
|
||||||
## Session workflow
|
## Session workflow
|
||||||
|
|
||||||
|
|||||||
+35
-8
@@ -50,10 +50,10 @@ func run(args []string) error {
|
|||||||
socket := fs.String("socket", "", "core IPC socket path (required)")
|
socket := fs.String("socket", "", "core IPC socket path (required)")
|
||||||
url := fs.String("url", "", "CalDAV calendar URL, e.g. http://localhost:5232/kami/personal (required)")
|
url := fs.String("url", "", "CalDAV calendar URL, e.g. http://localhost:5232/kami/personal (required)")
|
||||||
user := fs.String("user", "", "CalDAV basic-auth username (required)")
|
user := fs.String("user", "", "CalDAV basic-auth username (required)")
|
||||||
pass := fs.String("pass", "", "CalDAV basic-auth password (required)")
|
passFile := fs.String("pass-file", "", "file holding the CalDAV basic-auth password (required — never passed as a flag value)")
|
||||||
renderURL := fs.String("render-url", "", "CalDAV collection maven publishes her own reminders to; empty disables rendering")
|
renderURL := fs.String("render-url", "", "CalDAV collection maven publishes her own reminders to; empty disables rendering")
|
||||||
renderUser := fs.String("render-user", "", "basic-auth username for -render-url (defaults to -user)")
|
renderUser := fs.String("render-user", "", "basic-auth username for -render-url (defaults to -user)")
|
||||||
renderPass := fs.String("render-pass", "", "basic-auth password for -render-url (defaults to -pass)")
|
renderPassFile := fs.String("render-pass-file", "", "file holding the password for -render-url (defaults to -pass-file)")
|
||||||
renderDur := fs.Duration("render-duration", calendar.DefaultReminderDuration, "how long a rendered reminder occupies")
|
renderDur := fs.Duration("render-duration", calendar.DefaultReminderDuration, "how long a rendered reminder occupies")
|
||||||
interval := fs.Duration("interval", 5*time.Minute, "poll cadence")
|
interval := fs.Duration("interval", 5*time.Minute, "poll cadence")
|
||||||
timeout := fs.Duration("timeout", 10*time.Second, "per-request HTTP timeout")
|
timeout := fs.Duration("timeout", 10*time.Second, "per-request HTTP timeout")
|
||||||
@@ -63,13 +63,22 @@ func run(args []string) error {
|
|||||||
if *socket == "" {
|
if *socket == "" {
|
||||||
return fmt.Errorf("-socket is required")
|
return fmt.Errorf("-socket is required")
|
||||||
}
|
}
|
||||||
if *url == "" || *user == "" || *pass == "" {
|
if *url == "" || *user == "" || *passFile == "" {
|
||||||
return fmt.Errorf("-url, -user, -pass are required")
|
return fmt.Errorf("-url, -user, -pass-file are required")
|
||||||
}
|
}
|
||||||
if err := checkRenderTarget([]string{*url}, *renderURL); err != nil {
|
if err := checkRenderTarget([]string{*url}, *renderURL); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 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. Same rule mavmaild and mavpoll follow. Read
|
||||||
|
// once at start, so a rotated password means a restart.
|
||||||
|
pass, err := readSecret(*passFile)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||||
defer stop()
|
defer stop()
|
||||||
|
|
||||||
@@ -85,17 +94,20 @@ func run(args []string) error {
|
|||||||
http: hc,
|
http: hc,
|
||||||
url: strings.TrimRight(*url, "/"),
|
url: strings.TrimRight(*url, "/"),
|
||||||
user: *user,
|
user: *user,
|
||||||
pass: *pass,
|
pass: pass,
|
||||||
}
|
}
|
||||||
|
|
||||||
var rend *renderer
|
var rend *renderer
|
||||||
if *renderURL != "" {
|
if *renderURL != "" {
|
||||||
ru, rp := *renderUser, *renderPass
|
ru, rp := *renderUser, pass
|
||||||
if ru == "" {
|
if ru == "" {
|
||||||
ru = *user
|
ru = *user
|
||||||
}
|
}
|
||||||
if rp == "" {
|
if *renderPassFile != "" {
|
||||||
rp = *pass
|
rp, err = readSecret(*renderPassFile)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
}
|
}
|
||||||
rend = newRenderer(core, hc, *renderURL, ru, rp, *renderDur)
|
rend = newRenderer(core, hc, *renderURL, ru, rp, *renderDur)
|
||||||
log.Printf("mavcaldav: rendering reminders to %s", *renderURL)
|
log.Printf("mavcaldav: rendering reminders to %s", *renderURL)
|
||||||
@@ -131,6 +143,21 @@ func run(args []string) error {
|
|||||||
// It takes the whole read set, not one URL. The guarantee in the package
|
// It takes the whole read set, not one URL. The guarantee in the package
|
||||||
// comment is about every calendar maven reads, and a second read target added
|
// comment is about every calendar maven reads, and a second read target added
|
||||||
// later must not quietly fall outside the check.
|
// later must not quietly fall outside the check.
|
||||||
|
// readSecret reads one credential from a file and refuses an empty one. An
|
||||||
|
// empty file is a deployment mistake, not a password, and CalDAV basic auth
|
||||||
|
// would send it and get a 401 every poll.
|
||||||
|
func readSecret(path string) (string, error) {
|
||||||
|
raw, err := os.ReadFile(path)
|
||||||
|
if err != nil {
|
||||||
|
return "", fmt.Errorf("read password file: %w", err)
|
||||||
|
}
|
||||||
|
secret := strings.TrimSpace(string(raw))
|
||||||
|
if secret == "" {
|
||||||
|
return "", fmt.Errorf("password file %s is empty", path)
|
||||||
|
}
|
||||||
|
return secret, nil
|
||||||
|
}
|
||||||
|
|
||||||
func checkRenderTarget(readURLs []string, renderURL string) error {
|
func checkRenderTarget(readURLs []string, renderURL string) error {
|
||||||
if renderURL == "" {
|
if renderURL == "" {
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -5,12 +5,38 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/kami/maven/internal/ipc"
|
"github.com/kami/maven/internal/ipc"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// The password comes from a file so it never reaches argv. An empty or missing
|
||||||
|
// file must fail at start rather than authenticate as "" against his calendar.
|
||||||
|
func TestReadSecret(t *testing.T) {
|
||||||
|
dir := t.TempDir()
|
||||||
|
good := filepath.Join(dir, "ok")
|
||||||
|
if err := os.WriteFile(good, []byte(" hunter2\n"), 0o600); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if got, err := readSecret(good); err != nil || got != "hunter2" {
|
||||||
|
t.Fatalf("readSecret(good) = %q, %v; want \"hunter2\", nil", got, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
empty := filepath.Join(dir, "empty")
|
||||||
|
if err := os.WriteFile(empty, []byte("\n \n"), 0o600); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, err := readSecret(empty); err == nil {
|
||||||
|
t.Fatal("readSecret(empty) = nil error, want refusal")
|
||||||
|
}
|
||||||
|
if _, err := readSecret(filepath.Join(dir, "absent")); err == nil {
|
||||||
|
t.Fatal("readSecret(absent) = nil error, want refusal")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
type fakeCore struct {
|
type fakeCore struct {
|
||||||
ipc.UnimplementedCoreAPI
|
ipc.UnimplementedCoreAPI
|
||||||
facts map[string]ipc.Fact // composite key "key|source" → Fact
|
facts map[string]ipc.Fact // composite key "key|source" → Fact
|
||||||
|
|||||||
@@ -157,6 +157,44 @@ services:
|
|||||||
# - maildata:/var/lib/mavmaild
|
# - maildata:/var/lib/mavmaild
|
||||||
# - ./deploy/imap.password:/run/secrets/imap.password:ro
|
# - ./deploy/imap.password:/run/secrets/imap.password:ro
|
||||||
|
|
||||||
|
# The calendar reader (Vikunja #644) is OFF and commented out: it needs a
|
||||||
|
# CalDAV account, and there is none on this box. It was built, listed in
|
||||||
|
# `make build`, and deployed nowhere, which is the worst of the three states —
|
||||||
|
# this block records the decision instead.
|
||||||
|
#
|
||||||
|
# What its absence costs, so the cost is visible from here:
|
||||||
|
# - Agenda questions route correctly and answer from nothing. Stage 0 sends
|
||||||
|
# "что у меня сегодня" to IntentQuery (V-498) and the `calendar` query
|
||||||
|
# source reads facts(kind=env, source=caldav:*) that nobody writes.
|
||||||
|
# - The nudge gate loses a suppressor. loop.State.CalendarBusy is fed by
|
||||||
|
# those same facts, so "do not nag mid-meeting" is permanently false.
|
||||||
|
#
|
||||||
|
# Core never sees the CalDAV password: the reader polls the collection itself
|
||||||
|
# and hands core one fact per event over WriteFact. Nothing here can create a
|
||||||
|
# reminder, so a misread event cannot fire.
|
||||||
|
#
|
||||||
|
# The password is read from a FILE, so it never appears in `ps`, in this file,
|
||||||
|
# or in shell history — the same rule mavpoll and mavmaild follow.
|
||||||
|
#
|
||||||
|
# To enable: write the password to deploy/caldav.password (0600, gitignored),
|
||||||
|
# point -url at the collection, and uncomment this service. No mavend.json
|
||||||
|
# block is needed — events arrive over IPC as facts. -render-url is optional
|
||||||
|
# and OFF here: it publishes Maven's own reminders back as events, and it must
|
||||||
|
# not name the collection -url reads, or the poller reads its own writes back
|
||||||
|
# in (checkRenderTarget refuses that). It takes -render-pass-file, and falls
|
||||||
|
# back to this password when that is not given.
|
||||||
|
# mavcaldav:
|
||||||
|
# <<: *image
|
||||||
|
# command: ["mavcaldav", "-socket", "/run/maven/mavend.sock",
|
||||||
|
# "-url", "http://localhost:5232/kami/personal",
|
||||||
|
# "-user", "kami",
|
||||||
|
# "-pass-file", "/run/secrets/caldav.password",
|
||||||
|
# "-interval", "5m"]
|
||||||
|
# depends_on: [mavend]
|
||||||
|
# volumes:
|
||||||
|
# - sockets:/run/maven
|
||||||
|
# - ./deploy/caldav.password:/run/secrets/caldav.password:ro
|
||||||
|
|
||||||
volumes:
|
volumes:
|
||||||
dbdata:
|
dbdata:
|
||||||
sockets:
|
sockets:
|
||||||
|
|||||||
+37
-6
@@ -89,8 +89,8 @@ type Poller struct {
|
|||||||
ranker Ranker
|
ranker Ranker
|
||||||
cfg Config
|
cfg Config
|
||||||
nextDue map[string]time.Time
|
nextDue map[string]time.Time
|
||||||
seen map[string]map[string]bool // feed → item ID, for items with no date
|
seen map[string]*seenIDs // feed → item IDs, for items with no date
|
||||||
polled map[string]bool // feed → polled at least once in THIS process
|
polled map[string]bool // feed → polled at least once in THIS process
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewPoller wires a poller. Returns nil when there is nothing to poll — a
|
// NewPoller wires a poller. Returns nil when there is nothing to poll — a
|
||||||
@@ -122,7 +122,7 @@ func NewPoller(feeds []FeedConfig, fetch Fetcher, notes Notes, marks Marks, embe
|
|||||||
feeds: valid, fetch: fetch, notes: notes, marks: marks,
|
feeds: valid, fetch: fetch, notes: notes, marks: marks,
|
||||||
embed: embed, ranker: ranker, cfg: cfg,
|
embed: embed, ranker: ranker, cfg: cfg,
|
||||||
nextDue: map[string]time.Time{},
|
nextDue: map[string]time.Time{},
|
||||||
seen: map[string]map[string]bool{},
|
seen: map[string]*seenIDs{},
|
||||||
polled: map[string]bool{},
|
polled: map[string]bool{},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -283,6 +283,38 @@ func (p *Poller) mark(ctx context.Context, feed string, now time.Time) (time.Tim
|
|||||||
return at, true
|
return at, true
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// maxSeenPerFeed bounds the undated-item set. It has to stay comfortably above
|
||||||
|
// any one feed's front page, or an item still listed there would fall out of the
|
||||||
|
// set and be written a second time. A few hundred entries covers the largest
|
||||||
|
// page anyone publishes, and the set only has to span one poll window plus the
|
||||||
|
// resync guard, not all of history.
|
||||||
|
const maxSeenPerFeed = 512
|
||||||
|
|
||||||
|
// seenIDs is a bounded insertion-ordered set. The map answers the lookup, the
|
||||||
|
// slice remembers what to drop first, so an undated feed cannot grow the poller
|
||||||
|
// for as long as mavend runs.
|
||||||
|
type seenIDs struct {
|
||||||
|
ids map[string]bool
|
||||||
|
order []string
|
||||||
|
}
|
||||||
|
|
||||||
|
// add records id and reports whether it was new.
|
||||||
|
func (s *seenIDs) add(id string) bool {
|
||||||
|
if s.ids == nil {
|
||||||
|
s.ids = make(map[string]bool, maxSeenPerFeed)
|
||||||
|
}
|
||||||
|
if s.ids[id] {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
s.ids[id] = true
|
||||||
|
s.order = append(s.order, id)
|
||||||
|
if len(s.order) > maxSeenPerFeed {
|
||||||
|
delete(s.ids, s.order[0])
|
||||||
|
s.order = s.order[1:]
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
// fresh — two dedup rules, because feeds are inconsistent about dates. A dated
|
// 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
|
// item must be newer than the mark; an undated one is kept once per process by
|
||||||
// ID.
|
// ID.
|
||||||
@@ -308,12 +340,11 @@ func (p *Poller) fresh(f FeedConfig, it Item, mark, now time.Time, resync bool)
|
|||||||
id = it.Title
|
id = it.Title
|
||||||
}
|
}
|
||||||
if p.seen[f.Name] == nil {
|
if p.seen[f.Name] == nil {
|
||||||
p.seen[f.Name] = map[string]bool{}
|
p.seen[f.Name] = &seenIDs{}
|
||||||
}
|
}
|
||||||
if p.seen[f.Name][id] {
|
if !p.seen[f.Name].add(id) {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
p.seen[f.Name][id] = true
|
|
||||||
return !resync
|
return !resync
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -3,6 +3,7 @@ package rss
|
|||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
@@ -208,3 +209,27 @@ func TestNoFeedsMeansNoPoller(t *testing.T) {
|
|||||||
t.Fatal("a feed with no name or url is not a configuration")
|
t.Fatal("a feed with no name or url is not a configuration")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// An undated feed used to grow p.seen for as long as mavend ran. The set is
|
||||||
|
// bounded now, and the bound must not cost the dedupe an item still on the
|
||||||
|
// front page — only ids far older than any page fall out.
|
||||||
|
func TestSeenIDsBounded(t *testing.T) {
|
||||||
|
var s seenIDs
|
||||||
|
for i := 0; i < maxSeenPerFeed*3; i++ {
|
||||||
|
if !s.add(fmt.Sprintf("item-%d", i)) {
|
||||||
|
t.Fatalf("item-%d read as already seen", i)
|
||||||
|
}
|
||||||
|
if len(s.ids) > maxSeenPerFeed || len(s.order) > maxSeenPerFeed {
|
||||||
|
t.Fatalf("after %d inserts: ids=%d order=%d, cap is %d",
|
||||||
|
i+1, len(s.ids), len(s.order), maxSeenPerFeed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// The newest insert is still deduped; the oldest was evicted.
|
||||||
|
last := fmt.Sprintf("item-%d", maxSeenPerFeed*3-1)
|
||||||
|
if s.add(last) {
|
||||||
|
t.Fatalf("%s read as new, so the most recent id was dropped", last)
|
||||||
|
}
|
||||||
|
if !s.add("item-0") {
|
||||||
|
t.Fatal("item-0 survived, so nothing was evicted")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
+135
-11
@@ -68,6 +68,13 @@ func (m *MemoryStore) Insert(ctx context.Context, id string, vec []float32, meta
|
|||||||
// Rows under memory.NonRecallPrefix are excluded in SQL. They are speaker
|
// Rows under memory.NonRecallPrefix are excluded in SQL. They are speaker
|
||||||
// voiceprints sharing this table, and note recall must not rank them; see that
|
// voiceprints sharing this table, and note recall must not rank them; see that
|
||||||
// constant for why the previous arrangement only appeared to do this.
|
// constant for why the previous arrangement only appeared to do this.
|
||||||
|
//
|
||||||
|
// Every row is still scored, because a full scan is what picks the winners.
|
||||||
|
// What the scan does NOT do is pay for a row it is about to discard: the score
|
||||||
|
// is read straight off the stored bytes without materializing a []float32, and
|
||||||
|
// the meta blob is copied and unmarshalled only for a row that has entered the
|
||||||
|
// topK. Losers cost one dot product and nothing else. Ranking is unchanged —
|
||||||
|
// same scores, same order, same ties.
|
||||||
func (m *MemoryStore) Search(ctx context.Context, vec []float32, topK int) ([]memory.Result, error) {
|
func (m *MemoryStore) Search(ctx context.Context, vec []float32, topK int) ([]memory.Result, error) {
|
||||||
if topK <= 0 {
|
if topK <= 0 {
|
||||||
topK = 10
|
topK = 10
|
||||||
@@ -80,30 +87,127 @@ func (m *MemoryStore) Search(ctx context.Context, vec []float32, topK int) ([]me
|
|||||||
}
|
}
|
||||||
defer rows.Close()
|
defer rows.Close()
|
||||||
|
|
||||||
var out []memory.Result
|
// sql.RawBytes hands us the driver's own buffer, valid only until the next
|
||||||
|
// Next(). Nothing here outlives the row except what topK.offer copies on a
|
||||||
|
// survivor, so the three columns cost no allocation per row.
|
||||||
|
var id, blob, metaJSON sql.RawBytes
|
||||||
|
top := newTopK(topK)
|
||||||
for rows.Next() {
|
for rows.Next() {
|
||||||
var id, metaJSON string
|
|
||||||
var blob []byte
|
|
||||||
if err := rows.Scan(&id, &blob, &metaJSON); err != nil {
|
if err := rows.Scan(&id, &blob, &metaJSON); err != nil {
|
||||||
return nil, fmt.Errorf("memory: row: %w", err)
|
return nil, fmt.Errorf("memory: row: %w", err)
|
||||||
}
|
}
|
||||||
meta := map[string]string{}
|
top.offer(dotBlob(vec, blob), id, metaJSON)
|
||||||
if err := json.Unmarshal([]byte(metaJSON), &meta); err != nil {
|
|
||||||
return nil, fmt.Errorf("memory: unmarshal meta for %q: %w", id, err)
|
|
||||||
}
|
|
||||||
out = append(out, memory.Result{ID: id, Score: dot(vec, decodeVec(blob)), Meta: meta})
|
|
||||||
}
|
}
|
||||||
if err := rows.Err(); err != nil {
|
if err := rows.Err(); err != nil {
|
||||||
return nil, fmt.Errorf("memory: rows: %w", err)
|
return nil, fmt.Errorf("memory: rows: %w", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
sort.Slice(out, func(i, j int) bool { return out[i].Score > out[j].Score })
|
survivors := top.sorted()
|
||||||
if topK < len(out) {
|
out := make([]memory.Result, 0, len(survivors))
|
||||||
out = out[:topK]
|
for _, c := range survivors {
|
||||||
|
meta := map[string]string{}
|
||||||
|
if err := json.Unmarshal(c.meta, &meta); err != nil {
|
||||||
|
return nil, fmt.Errorf("memory: unmarshal meta for %q: %w", c.id, err)
|
||||||
|
}
|
||||||
|
out = append(out, memory.Result{ID: c.id, Score: c.score, Meta: meta})
|
||||||
}
|
}
|
||||||
return out, nil
|
return out, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// candidate is one row that is currently in the topK: its score, its id, and
|
||||||
|
// its meta blob copied out of the driver's buffer. The copy is the price of
|
||||||
|
// surviving, and only survivors pay it.
|
||||||
|
type candidate struct {
|
||||||
|
score float64
|
||||||
|
id string
|
||||||
|
meta []byte
|
||||||
|
}
|
||||||
|
|
||||||
|
// topK keeps the k highest-scoring candidates seen so far as a min-heap, so the
|
||||||
|
// weakest survivor is always heap[0] and one comparison decides whether a new
|
||||||
|
// row is worth copying. k is 10 in practice, so the heap is tiny and the whole
|
||||||
|
// structure fits in cache.
|
||||||
|
//
|
||||||
|
// It is a plain slice with hand-written sift operations rather than
|
||||||
|
// container/heap, because that interface boxes every element into an `any` on
|
||||||
|
// Push and costs an allocation per surviving row.
|
||||||
|
type topK struct {
|
||||||
|
k int
|
||||||
|
heap []candidate
|
||||||
|
}
|
||||||
|
|
||||||
|
func newTopK(k int) *topK {
|
||||||
|
return &topK{k: k, heap: make([]candidate, 0, k)}
|
||||||
|
}
|
||||||
|
|
||||||
|
// offer admits a row if it beats the weakest survivor, or if the heap is not
|
||||||
|
// full yet. id and meta are the driver's buffers and are copied here, never
|
||||||
|
// retained.
|
||||||
|
//
|
||||||
|
// A row that only ties the weakest survivor does not displace it, so among
|
||||||
|
// equal scores the earliest k rows are kept. The full sort this replaced used
|
||||||
|
// sort.Slice, which is not stable, so it broke such a tie arbitrarily. That is
|
||||||
|
// the ONE observable difference between the two, and it is deliberate:
|
||||||
|
// deterministic beats arbitrary.
|
||||||
|
//
|
||||||
|
// It is not academic. Under the real embedder an exact tie means duplicate
|
||||||
|
// vectors and nothing in the recall eval moved (V-643). Under the hash
|
||||||
|
// embedder the eval's deterministic floor uses, ties are everywhere — it is
|
||||||
|
// bag-of-words, so every note sharing no word with the query scores exactly 0
|
||||||
|
// — and recall@3 on that run moved 74.1% to 81.5% purely because the zeros now
|
||||||
|
// come out in a fixed order. Neither number measures retrieval. recall@1 and
|
||||||
|
// false recall, which the eval actually asserts, are unchanged on both runs.
|
||||||
|
func (t *topK) offer(score float64, id, meta []byte) {
|
||||||
|
if t.k == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if len(t.heap) < t.k {
|
||||||
|
t.heap = append(t.heap, candidate{score: score, id: string(id), meta: append([]byte(nil), meta...)})
|
||||||
|
t.up(len(t.heap) - 1)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if score <= t.heap[0].score {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
t.heap[0] = candidate{score: score, id: string(id), meta: append([]byte(nil), meta...)}
|
||||||
|
t.down(0)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *topK) up(i int) {
|
||||||
|
for i > 0 {
|
||||||
|
parent := (i - 1) / 2
|
||||||
|
if t.heap[parent].score <= t.heap[i].score {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
t.heap[parent], t.heap[i] = t.heap[i], t.heap[parent]
|
||||||
|
i = parent
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (t *topK) down(i int) {
|
||||||
|
for {
|
||||||
|
l, r, small := 2*i+1, 2*i+2, i
|
||||||
|
if l < len(t.heap) && t.heap[l].score < t.heap[small].score {
|
||||||
|
small = l
|
||||||
|
}
|
||||||
|
if r < len(t.heap) && t.heap[r].score < t.heap[small].score {
|
||||||
|
small = r
|
||||||
|
}
|
||||||
|
if small == i {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
t.heap[small], t.heap[i] = t.heap[i], t.heap[small]
|
||||||
|
i = small
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// sorted drains the heap into descending score order — what Search returns.
|
||||||
|
func (t *topK) sorted() []candidate {
|
||||||
|
out := t.heap
|
||||||
|
sort.Slice(out, func(i, j int) bool { return out[i].score > out[j].score })
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
// ByPrefix returns every row whose id starts with prefix, vectors included.
|
// ByPrefix returns every row whose id starts with prefix, vectors included.
|
||||||
//
|
//
|
||||||
// This is not a similarity query and deliberately does not score anything:
|
// This is not a similarity query and deliberately does not score anything:
|
||||||
@@ -240,6 +344,26 @@ func decodeVec(b []byte) []float32 {
|
|||||||
return v
|
return v
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// dotBlob is dot against a vector still in its stored encoding, so scoring a
|
||||||
|
// row the query is about to discard does not allocate the []float32 that
|
||||||
|
// decodeVec would build. Same arithmetic, same order of operations, so it
|
||||||
|
// returns bit-identical scores to dot(a, decodeVec(b)).
|
||||||
|
//
|
||||||
|
// A blob whose length isn't a multiple of 4 is truncated to the whole-element
|
||||||
|
// prefix, matching decodeVec, and a length mismatch is 0, matching dot.
|
||||||
|
func dotBlob(a []float32, b []byte) float64 {
|
||||||
|
n := len(b) / 4
|
||||||
|
if len(a) != n || n == 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
var sum float64
|
||||||
|
for i := 0; i < n; i++ {
|
||||||
|
f := math.Float32frombits(binary.LittleEndian.Uint32(b[4*i:]))
|
||||||
|
sum += float64(a[i]) * float64(f)
|
||||||
|
}
|
||||||
|
return sum
|
||||||
|
}
|
||||||
|
|
||||||
// dot is the cosine similarity for L2-normalized vectors (mismatched lengths ⇒
|
// dot is the cosine similarity for L2-normalized vectors (mismatched lengths ⇒
|
||||||
// 0, matching internal/memory's cosine).
|
// 0, matching internal/memory's cosine).
|
||||||
func dot(a, b []float32) float64 {
|
func dot(a, b []float32) float64 {
|
||||||
|
|||||||
@@ -0,0 +1,77 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"math"
|
||||||
|
"math/rand"
|
||||||
|
"path/filepath"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
// benchDim is the resident embedder's width (multilingual-e5-small, 384), so
|
||||||
|
// the per-row decode cost the benchmark measures is the real one.
|
||||||
|
const benchDim = 384
|
||||||
|
|
||||||
|
// seedMemVectors fills a fresh store with n L2-normalized rows carrying a meta
|
||||||
|
// blob the size recall actually stores — the note text plus its type — because
|
||||||
|
// the cost this benchmark exists to measure is unmarshalling that blob for
|
||||||
|
// every row when only topK survivors need it.
|
||||||
|
func seedMemVectors(tb testing.TB, n int) *MemoryStore {
|
||||||
|
tb.Helper()
|
||||||
|
path := filepath.Join(tb.TempDir(), "mem_bench.db")
|
||||||
|
st, err := Open(context.Background(), path)
|
||||||
|
if err != nil {
|
||||||
|
tb.Fatalf("Open: %v", err)
|
||||||
|
}
|
||||||
|
tb.Cleanup(func() { _ = st.Close() })
|
||||||
|
m := st.VectorMemory()
|
||||||
|
|
||||||
|
rng := rand.New(rand.NewSource(1))
|
||||||
|
ctx := context.Background()
|
||||||
|
for i := 0; i < n; i++ {
|
||||||
|
if err := m.Insert(ctx, fmt.Sprintf("note:%d", i), randUnitVec(rng, benchDim), map[string]string{
|
||||||
|
"type": "note",
|
||||||
|
"text": fmt.Sprintf("заметка номер %d о том, что надо не забыть сделать на неделе", i),
|
||||||
|
}); err != nil {
|
||||||
|
tb.Fatalf("Insert %d: %v", i, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return m
|
||||||
|
}
|
||||||
|
|
||||||
|
func randUnitVec(rng *rand.Rand, dim int) []float32 {
|
||||||
|
v := make([]float32, dim)
|
||||||
|
var norm float64
|
||||||
|
for i := range v {
|
||||||
|
f := rng.NormFloat64()
|
||||||
|
v[i] = float32(f)
|
||||||
|
norm += f * f
|
||||||
|
}
|
||||||
|
norm = math.Sqrt(norm)
|
||||||
|
for i := range v {
|
||||||
|
v[i] = float32(float64(v[i]) / norm)
|
||||||
|
}
|
||||||
|
return v
|
||||||
|
}
|
||||||
|
|
||||||
|
// BenchmarkMemoryStoreSearch measures one recall query against a store of n
|
||||||
|
// rows. Row counts bracket the documented scale: 1000 is a plausible today,
|
||||||
|
// 10000 is the "thousands, not millions" ceiling the type doc claims a full
|
||||||
|
// scan is fine at.
|
||||||
|
func BenchmarkMemoryStoreSearch(b *testing.B) {
|
||||||
|
for _, n := range []int{1000, 10000} {
|
||||||
|
b.Run(fmt.Sprintf("rows=%d", n), func(b *testing.B) {
|
||||||
|
m := seedMemVectors(b, n)
|
||||||
|
q := randUnitVec(rand.New(rand.NewSource(2)), benchDim)
|
||||||
|
ctx := context.Background()
|
||||||
|
b.ReportAllocs()
|
||||||
|
b.ResetTimer()
|
||||||
|
for i := 0; i < b.N; i++ {
|
||||||
|
if _, err := m.Search(ctx, q, 10); err != nil {
|
||||||
|
b.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,107 @@
|
|||||||
|
package store
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"math/rand"
|
||||||
|
"sort"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/kami/maven/internal/memory"
|
||||||
|
)
|
||||||
|
|
||||||
|
// naiveSearch is the implementation Search replaced: score every row into a
|
||||||
|
// slice, sort the whole slice, truncate. It stays in the test file as the
|
||||||
|
// reference the bounded-heap version is judged against, because "recall must
|
||||||
|
// not change" is a claim about output, not about the code that produces it.
|
||||||
|
func naiveSearch(t *testing.T, m *MemoryStore, vec []float32, topK int) []memory.Result {
|
||||||
|
t.Helper()
|
||||||
|
rows, err := m.db.QueryContext(context.Background(),
|
||||||
|
`SELECT id, vec FROM memory_vectors WHERE id NOT LIKE ? ESCAPE '\'`,
|
||||||
|
escapeLike(memory.NonRecallPrefix)+"%")
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("naive scan: %v", err)
|
||||||
|
}
|
||||||
|
defer rows.Close()
|
||||||
|
var out []memory.Result
|
||||||
|
for rows.Next() {
|
||||||
|
var id string
|
||||||
|
var blob []byte
|
||||||
|
if err := rows.Scan(&id, &blob); err != nil {
|
||||||
|
t.Fatalf("naive row: %v", err)
|
||||||
|
}
|
||||||
|
out = append(out, memory.Result{ID: id, Score: dot(vec, decodeVec(blob))})
|
||||||
|
}
|
||||||
|
if err := rows.Err(); err != nil {
|
||||||
|
t.Fatalf("naive rows: %v", err)
|
||||||
|
}
|
||||||
|
sort.Slice(out, func(i, j int) bool { return out[i].Score > out[j].Score })
|
||||||
|
if topK < len(out) {
|
||||||
|
out = out[:topK]
|
||||||
|
}
|
||||||
|
return out
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestMemoryStoreSearchMatchesNaive is the constraint on V-643: the bounded
|
||||||
|
// heap must return exactly what a full scan and sort returned. Distinct random
|
||||||
|
// vectors, so no two scores tie and the ranking is total — a mismatch here is
|
||||||
|
// arithmetic or heap logic, not a tie-break difference.
|
||||||
|
func TestMemoryStoreSearchMatchesNaive(t *testing.T) {
|
||||||
|
ctx := context.Background()
|
||||||
|
m := newMemTestStore(t).VectorMemory()
|
||||||
|
|
||||||
|
rng := rand.New(rand.NewSource(7))
|
||||||
|
const rows, dim = 500, 64
|
||||||
|
for i := 0; i < rows; i++ {
|
||||||
|
if err := m.Insert(ctx, fmt.Sprintf("n%d", i), randUnitVec(rng, dim), map[string]string{
|
||||||
|
"text": fmt.Sprintf("note %d", i),
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("Insert %d: %v", i, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, topK := range []int{1, 3, 10, 50, rows, rows + 100} {
|
||||||
|
q := randUnitVec(rng, dim)
|
||||||
|
got, err := m.Search(ctx, q, topK)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Search topK=%d: %v", topK, err)
|
||||||
|
}
|
||||||
|
want := naiveSearch(t, m, q, topK)
|
||||||
|
if len(got) != len(want) {
|
||||||
|
t.Fatalf("topK=%d: got %d results, naive returned %d", topK, len(got), len(want))
|
||||||
|
}
|
||||||
|
for i := range want {
|
||||||
|
if got[i].ID != want[i].ID {
|
||||||
|
t.Errorf("topK=%d rank %d: got %q, naive says %q", topK, i, got[i].ID, want[i].ID)
|
||||||
|
}
|
||||||
|
if got[i].Score != want[i].Score {
|
||||||
|
t.Errorf("topK=%d rank %d (%s): score %v, naive says %v",
|
||||||
|
topK, i, got[i].ID, got[i].Score, want[i].Score)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if len(got) > 0 && got[0].Meta["text"] == "" {
|
||||||
|
t.Errorf("topK=%d: survivor %s has no meta — it was never unmarshalled", topK, got[0].ID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestDotBlobMatchesDot pins the claim in dotBlob's doc comment: reading the
|
||||||
|
// vector out of its stored bytes is bit-identical to decoding it first. Scores
|
||||||
|
// feed a gate with a 0.008 margin, so "close enough" is not the bar.
|
||||||
|
func TestDotBlobMatchesDot(t *testing.T) {
|
||||||
|
rng := rand.New(rand.NewSource(11))
|
||||||
|
for i := 0; i < 200; i++ {
|
||||||
|
a := randUnitVec(rng, 384)
|
||||||
|
b := randUnitVec(rng, 384)
|
||||||
|
if got, want := dotBlob(a, encodeVec(b)), dot(a, b); got != want {
|
||||||
|
t.Fatalf("dotBlob = %v, dot = %v", got, want)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Length mismatch is 0 in both, and so is an empty vector.
|
||||||
|
if got := dotBlob([]float32{1, 0}, encodeVec([]float32{1, 0, 0})); got != 0 {
|
||||||
|
t.Errorf("mismatched lengths scored %v, want 0", got)
|
||||||
|
}
|
||||||
|
if got := dotBlob(nil, nil); got != 0 {
|
||||||
|
t.Errorf("empty scored %v, want 0", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -295,6 +295,27 @@ func (f *Fetcher) checkURL(u *url.URL) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// pruneHostsAbove is when pruneLocked bothers to walk the map. Below it the
|
||||||
|
// walk costs more than the entries do, and `crawl.on_demand` means the host set
|
||||||
|
// is whatever he names out loud, so it grows slowly.
|
||||||
|
const pruneHostsAbove = 64
|
||||||
|
|
||||||
|
// pruneLocked drops hosts whose last dial is further back than HostInterval.
|
||||||
|
// Such an entry cannot delay anything — waitTurn would let the next request
|
||||||
|
// through immediately — so keeping it only holds memory for the life of the
|
||||||
|
// process. Caller holds f.mu.
|
||||||
|
func (f *Fetcher) pruneLocked(now time.Time) {
|
||||||
|
if len(f.last) <= pruneHostsAbove {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
cutoff := now.Add(-f.cfg.HostInterval)
|
||||||
|
for h, at := range f.last {
|
||||||
|
if at.Before(cutoff) {
|
||||||
|
delete(f.last, h)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// waitTurn blocks until this host's rate-limit interval has elapsed. It holds
|
// 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.
|
// no lock while sleeping, so two hosts never wait on each other.
|
||||||
func (f *Fetcher) waitTurn(ctx context.Context, host string) error {
|
func (f *Fetcher) waitTurn(ctx context.Context, host string) error {
|
||||||
@@ -304,6 +325,7 @@ func (f *Fetcher) waitTurn(ctx context.Context, host string) error {
|
|||||||
earliest := f.last[host].Add(f.cfg.HostInterval)
|
earliest := f.last[host].Add(f.cfg.HostInterval)
|
||||||
if !now.Before(earliest) {
|
if !now.Before(earliest) {
|
||||||
f.last[host] = now
|
f.last[host] = now
|
||||||
|
f.pruneLocked(now)
|
||||||
f.mu.Unlock()
|
f.mu.Unlock()
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
@@ -318,3 +319,37 @@ func TestPostObeysDenylist(t *testing.T) {
|
|||||||
t.Fatalf("error = %v, want ErrBlocked", err)
|
t.Fatalf("error = %v, want ErrBlocked", err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// f.last used to hold one entry per host ever dialed, for the life of the
|
||||||
|
// process. A host whose last dial is older than HostInterval cannot delay
|
||||||
|
// anything, so it is dropped once the map is worth walking.
|
||||||
|
func TestHostRateMapIsPruned(t *testing.T) {
|
||||||
|
f := New(Config{HostInterval: time.Minute, AllowPrivate: true})
|
||||||
|
stale := time.Now().Add(-time.Hour)
|
||||||
|
for i := 0; i < pruneHostsAbove*2; i++ {
|
||||||
|
f.last[fmt.Sprintf("h%d.example", i)] = stale
|
||||||
|
}
|
||||||
|
|
||||||
|
// One real turn is what triggers the sweep.
|
||||||
|
if err := f.waitTurn(context.Background(), "fresh.example"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if len(f.last) != 1 {
|
||||||
|
t.Fatalf("len(f.last) = %d after the sweep, want 1 (only the host just dialed)", len(f.last))
|
||||||
|
}
|
||||||
|
if _, ok := f.last["fresh.example"]; !ok {
|
||||||
|
t.Fatal("the host just dialed was pruned, so its own rate limit is lost")
|
||||||
|
}
|
||||||
|
|
||||||
|
// A host inside the interval is kept: pruning must not hand out a free turn.
|
||||||
|
f.last["recent.example"] = time.Now()
|
||||||
|
for i := 0; i < pruneHostsAbove*2; i++ {
|
||||||
|
f.last[fmt.Sprintf("g%d.example", i)] = stale
|
||||||
|
}
|
||||||
|
if err := f.waitTurn(context.Background(), "other.example"); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if _, ok := f.last["recent.example"]; !ok {
|
||||||
|
t.Fatal("a host dialed inside HostInterval was pruned")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user