kami
b35151418a
wireVoice runs before the tick loop exists, so it could only be handed the bare store adapter — and that adapter answers DayPlan with "not available via direct store API", because a day plan is assembled by the tick loop and is not a table to read. So queryDayPlan, which the query chain reaches for "какие у меня планы на сегодня", failed for every caller on the deployed daemon. main already back-patches the other direction (daemonAPI.chatFn = handler.handleText). This is the same seam in reverse, at both wiring sites. No recursion risk: nothing in the voice path calls api.Chat. With the plan reachable, it recited its reminders as literal JSON. The payload unwrapper existed but was private to the phraser, so the day plan had its own non-unwrapping copy. One owner now, store.ReminderText, with the phraser delegating to it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01TrVSBKe3RFDF4fGYKWYQnX
829 lines
28 KiB
Go
829 lines
28 KiB
Go
// Package main is mavend — maven's daemon.
|
|
//
|
|
// "core = the only key-holder": one process holds the unlocked store + the
|
|
// trigger loop; modules are separate processes, key-free, fail-independent.
|
|
// the daemon wires Store → Gatherer → Tick → phraser → delivery, runs the 60s
|
|
// ticker, owns the cold-start unlock dance, and exposes the CoreAPI boundary
|
|
// over a unix socket for modules (router/delivery/poller/...) to call.
|
|
//
|
|
// Floor (this file): pluggable seams wired with the deterministic Stubs.
|
|
// - phraser Stub (no LLM)
|
|
// - voice sink wired via wireVoice: embedder/classifier seeded with ~10
|
|
// examples across 5 intents; stt + tts stubs in-process by default,
|
|
// remote module sockets when configured; TCP listener on voice.bind.
|
|
// The voice sink (voicesink.Sink via Sessions) is wired into the
|
|
// dispatcher — the reactive path (push-to-talk) AND proactive nudges
|
|
// (care-when-present, sev3/sev4 present) both route through the same
|
|
// stt→router→tts→client pipeline.
|
|
// - auth FloorEnrollment + nil Session — cold-start unlock assumed: today
|
|
// the store opens plain sqlite (sqlcipher deferred). the daemon runs
|
|
// "unlocked" — the locked-until-asserted dance lands with the Session
|
|
// verifier + ask-password transport (open spec item).
|
|
//
|
|
// Cold-start unlock (2026-07-06):
|
|
//
|
|
// When a passkey credential is enrolled AND no env key is set, the daemon
|
|
// starts in LOCKED mode: the IPC server runs but rejects all store methods
|
|
// except MethodAssertStepUp and MethodUnlock. A passkey assertion followed
|
|
// by MethodUnlock (with that credential's WebAuthn PRF output) unwraps the at-rest
|
|
// AES-256 key from a wrapped blob on disk (HKDF-SHA256 + AES-GCM) and opens
|
|
// the encrypted store. After unlock, the daemon wires voice, loop, and
|
|
// delivery and runs normally.
|
|
//
|
|
// Fallback: when db_key_env is set (or no wrapped file exists), the daemon
|
|
// starts unlocked from the env key (pre-unlock behavior). Enrolling a passkey
|
|
// while unlocked calls MethodStoreEncryptionKey to wrap the env key and
|
|
// persist the wrapped blob — enabling cold-start unlock on the next boot
|
|
// after the env key is removed. That write happens once, when no blob
|
|
// exists; replacing an existing one takes an explicit request, see
|
|
// cmd/mavend/keyfile.go.
|
|
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"net"
|
|
"os"
|
|
"os/signal"
|
|
"sync"
|
|
"sync/atomic"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/kami/maven/internal/auth"
|
|
"github.com/kami/maven/internal/config"
|
|
"github.com/kami/maven/internal/delivery"
|
|
"github.com/kami/maven/internal/delivery/ntfysink"
|
|
"github.com/kami/maven/internal/delivery/telegramsink"
|
|
"github.com/kami/maven/internal/ipc"
|
|
"github.com/kami/maven/internal/loop"
|
|
"github.com/kami/maven/internal/persona"
|
|
"github.com/kami/maven/internal/phraser"
|
|
"github.com/kami/maven/internal/store"
|
|
"github.com/kami/maven/internal/webauthn"
|
|
)
|
|
|
|
var errLocked = errors.New("mavend: daemon locked — complete passkey assertion first")
|
|
|
|
// daemonLock tracks whether the daemon is in locked (pre-unlock) mode, and
|
|
// owns the store handle the unlock path creates.
|
|
//
|
|
// The store matters here because of who runs when. In locked mode there is no
|
|
// store at boot; one is opened inside UnlockFn, on an IPC goroutine, minutes
|
|
// or days later. Shutdown runs on the main goroutine. Without a handoff the
|
|
// main goroutine has nothing to close, and store.Close is what re-encrypts
|
|
// the tmpfs working copy back over the ciphertext file — so a daemon that
|
|
// cold-started lost every write of that session, silently, on the next boot.
|
|
type daemonLock struct {
|
|
mu sync.Mutex
|
|
locked bool
|
|
st *store.Store
|
|
}
|
|
|
|
func newDaemonLock(locked bool) *daemonLock {
|
|
return &daemonLock{locked: locked}
|
|
}
|
|
|
|
func (l *daemonLock) isLocked() bool {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
return l.locked
|
|
}
|
|
|
|
// unlock flips the flag and takes ownership of the store opened by UnlockFn.
|
|
func (l *daemonLock) unlock(st *store.Store) {
|
|
l.mu.Lock()
|
|
defer l.mu.Unlock()
|
|
l.locked = false
|
|
l.st = st
|
|
}
|
|
|
|
// closeStore seals the store the unlock path opened, if any. Safe to call
|
|
// when the daemon never unlocked, and safe to call twice.
|
|
func (l *daemonLock) closeStore() error {
|
|
l.mu.Lock()
|
|
st := l.st
|
|
l.st = nil
|
|
l.mu.Unlock()
|
|
if st == nil {
|
|
return nil
|
|
}
|
|
return st.Close()
|
|
}
|
|
|
|
func main() {
|
|
if err := run(os.Args[1:]); err != nil {
|
|
fmt.Fprintln(os.Stderr, "mavend:", err)
|
|
os.Exit(1)
|
|
}
|
|
}
|
|
|
|
func run(args []string) error {
|
|
cfgPath := flag.String("config", defaultConfigPath(), "path to mavend JSON config")
|
|
wrappedKeyPath := flag.String("wrapped-key-file", "", "path to wrapped encryption key blob (enables cold-start unlock)")
|
|
reembed := flag.Bool("reembed", false, "re-embed every stored note and fact with the configured embedder, then serve normally (run once after an embedder swap; the daemon does not answer until it finishes)")
|
|
flag.CommandLine.Parse(args)
|
|
reembedOnStart = *reembed
|
|
cfg, err := config.Load(*cfgPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP)
|
|
defer stop()
|
|
|
|
// ----- encryption key: env key → wrapped key → plaintext -----
|
|
// Priority: env key (from config/docker) > wrapped key (cold-start unlock) > plaintext (dev/CI).
|
|
// When a wrapped key file exists AND no env key is set, the daemon starts
|
|
// LOCKED and waits for a passkey assertion to unwrap it.
|
|
envKey, err := cfg.DBEncryptionKey()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
wrappedExists := false
|
|
if *wrappedKeyPath != "" {
|
|
if _, err := os.Stat(*wrappedKeyPath); err == nil {
|
|
wrappedExists = true
|
|
} else if !errors.Is(err, os.ErrNotExist) {
|
|
return fmt.Errorf("check wrapped key: %w", err)
|
|
}
|
|
} else {
|
|
// default path alongside the config
|
|
defaultWrapped := cfg.DefaultWrappedKeyPath()
|
|
if _, err := os.Stat(defaultWrapped); err == nil {
|
|
*wrappedKeyPath = defaultWrapped
|
|
wrappedExists = true
|
|
}
|
|
}
|
|
|
|
locked := wrappedExists && envKey == nil
|
|
dl := newDaemonLock(locked)
|
|
|
|
var st *store.Store
|
|
var envKeyBytes []byte // kept for WrapKeyFn (enrollment wraps this key)
|
|
|
|
// dbKey — the plaintext at-rest key, once the daemon has one. Set at boot
|
|
// in env-key mode and inside UnlockFn after a cold start. WrapKeyFn reads
|
|
// it from an IPC goroutine, hence the atomic: srv's function fields are
|
|
// installed before Serve and must not be reassigned afterwards.
|
|
var dbKey atomic.Pointer[[]byte]
|
|
|
|
// wrappedPath resolves the blob location the same way for both the read
|
|
// at boot and every write, so a default-path deployment cannot wrap to
|
|
// one file and unwrap from another.
|
|
wrappedPath := func() string {
|
|
if *wrappedKeyPath != "" {
|
|
return *wrappedKeyPath
|
|
}
|
|
return cfg.DefaultWrappedKeyPath()
|
|
}
|
|
|
|
if !locked {
|
|
// Normal boot: env key or plaintext (dev/CI)
|
|
if envKey != nil {
|
|
envKeyBytes = make([]byte, len(envKey))
|
|
copy(envKeyBytes, envKey)
|
|
dbKey.Store(&envKeyBytes)
|
|
st, err = store.OpenEncrypted(ctx, cfg.DBPath, cfg.DBTmpfs, envKey)
|
|
} else {
|
|
st, err = store.Open(ctx, cfg.DBPath)
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("open store: %w", err)
|
|
}
|
|
defer st.Close()
|
|
} else {
|
|
// Locked boot: the store does not exist yet. Seal whatever UnlockFn
|
|
// opened, at shutdown, on this goroutine.
|
|
defer func() {
|
|
if err := dl.closeStore(); err != nil {
|
|
log.Printf("mavend: seal store on shutdown: %v", err)
|
|
}
|
|
}()
|
|
}
|
|
|
|
// ----- daemon components (only wired when unlocked) -----
|
|
// Pre-declare so the unlock path can wire them later.
|
|
var (
|
|
gatherer *loop.Gatherer
|
|
rules []loop.Rule
|
|
phr phraser.Phraser
|
|
voiceW *voiceWiring
|
|
dispatcher *delivery.Dispatcher
|
|
tl *tickLoop
|
|
coreAPI ipc.CoreAPI
|
|
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)
|
|
)
|
|
|
|
// The unified intake journal (Vikunja #283). Built before anything else
|
|
// that holds a CoreAPI, because intakeAPI wraps that one interface and
|
|
// every intake path in the daemon reaches its sink through it. nil (the
|
|
// operator set intake_journal negative) means no decorator at all.
|
|
evBus := newEventBus(cfg)
|
|
// coreFor is what every in-process holder of a CoreAPI now takes, instead
|
|
// of a bare ipc.NewStoreAPI(st). Identical behaviour plus one published
|
|
// envelope per successful intake write.
|
|
coreFor := func() ipc.CoreAPI { return newIntakeAPI(ipc.NewStoreAPI(st), evBus, time.Now) }
|
|
|
|
if !locked {
|
|
rules = wireRules(cfg)
|
|
gatherer = loop.NewGatherer(st, rules)
|
|
if cfg.QuietHours != nil {
|
|
gatherer.SetQuietHours(cfg.QuietHours.Start, cfg.QuietHours.End)
|
|
}
|
|
|
|
// phraser
|
|
phr = phraser.NewStub()
|
|
if cfg.Phraser != nil {
|
|
pc := phraser.Config{
|
|
ModelPath: cfg.Phraser.ModelPath,
|
|
BinPath: cfg.Phraser.BinPath,
|
|
Listen: cfg.Phraser.Listen,
|
|
NGpuLayers: cfg.Phraser.NGpuLayers,
|
|
NCtx: cfg.Phraser.NCtx,
|
|
Timeout: time.Duration(cfg.Phraser.Timeout),
|
|
LLMNudges: cfg.Phraser.LLMNudges,
|
|
ContextBlock: contextBlockFn(cfg, time.Now),
|
|
}
|
|
if pc.BinPath == "" {
|
|
pc.BinPath = "llama-server"
|
|
}
|
|
if pc.Listen == "" {
|
|
pc.Listen = "127.0.0.1:0"
|
|
}
|
|
if pc.NCtx <= 0 {
|
|
pc.NCtx = 2048
|
|
}
|
|
if pc.Timeout <= 0 {
|
|
pc.Timeout = 30 * time.Second
|
|
}
|
|
var err error
|
|
phr, err = phraser.NewLLMPhraser(ctx, pc)
|
|
if err != nil {
|
|
return fmt.Errorf("phraser: %w", err)
|
|
}
|
|
}
|
|
|
|
// ecosystem — nexus + hexis + praxis (all over HTTP; no direct DB access)
|
|
eco = wireEcosystem(cfg)
|
|
|
|
// voice
|
|
voiceW, err = wireVoice(cfg, coreFor(), phr, st.VectorMemory(), st, eco)
|
|
if err != nil {
|
|
return fmt.Errorf("wire voice: %w", err)
|
|
}
|
|
|
|
// delivery
|
|
var ntfy delivery.Sink
|
|
if cfg.Ntfy != nil {
|
|
s, err := ntfysink.New(*cfg.Ntfy)
|
|
if err != nil {
|
|
return fmt.Errorf("wire ntfy sink: %w", err)
|
|
}
|
|
ntfy = s
|
|
}
|
|
var telegram delivery.Sink
|
|
if cfg.Telegram != nil {
|
|
s, err := telegramsink.New(*cfg.Telegram)
|
|
if err != nil {
|
|
return fmt.Errorf("wire telegram sink: %w", err)
|
|
}
|
|
telegram = s
|
|
}
|
|
var voiceSink delivery.Sink
|
|
if voiceW != nil {
|
|
voiceSink = voiceW.voiceSink
|
|
}
|
|
// A crashed prior run may have left "pending" delivery attempts (send
|
|
// may have landed externally, then the process died before recording
|
|
// it) — reconcile them to "unknown" before the tick loop resumes
|
|
// sending, so nothing auto-resends into that ambiguity.
|
|
if _, err := st.ReconcileStaleDeliveryAttempts(context.Background(), time.Now()); err != nil {
|
|
log.Printf("delivery outbox reconcile: %v", err)
|
|
}
|
|
dispatcher = delivery.NewDispatcher(delivery.Config{
|
|
Ntfy: ntfy,
|
|
Telegram: telegram,
|
|
Voice: voiceSink,
|
|
Ack: st,
|
|
Nudges: st,
|
|
Reminders: st,
|
|
Outbox: st,
|
|
})
|
|
|
|
// tick loop
|
|
tickInterval := time.Duration(cfg.TickInterval)
|
|
repeatInterval := time.Duration(cfg.RepeatInterval)
|
|
autotuneInterval := time.Duration(cfg.AutotuneInterval)
|
|
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(coreFor(), embedderOf(voiceW), cfg)
|
|
crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
|
|
|
|
coreAPI = &daemonAPI{
|
|
CoreAPI: coreFor(),
|
|
getTrace: tl.trace,
|
|
getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return tl.morningStatus(ctx, time.Now()) },
|
|
getDayPlan: func(ctx context.Context) ipc.DayPlan { return tl.dayPlan(ctx, time.Now()) },
|
|
getEvents: intakeEventsFn(evBus),
|
|
}
|
|
if voiceW != nil && voiceW.handler != nil {
|
|
api := coreAPI.(*daemonAPI)
|
|
api.chatFn = voiceW.handler.handleText
|
|
// And the reverse: the handler was wired with the bare store
|
|
// adapter, which cannot serve the day plan. See upgradeAPI.
|
|
voiceW.handler.upgradeAPI(api)
|
|
}
|
|
if voiceW != nil && voiceW.mcp != nil {
|
|
coreAPI.(*daemonAPI).getMCPServers = voiceW.mcp.status
|
|
}
|
|
} else {
|
|
// locked mode: no real store yet, so there's no meaningful CoreAPI to
|
|
// serve. srv.Check below is the actual guard — every CoreAPI call is
|
|
// refused before it reaches this value. This is just a safe non-nil
|
|
// placeholder: if the guard is ever bypassed by a bug, calls land
|
|
// here and fail loudly with ipc.ErrNotImplemented instead of a nil
|
|
// dereference or, worse, silently succeeding.
|
|
coreAPI = ipc.UnimplementedCoreAPI{}
|
|
}
|
|
|
|
// ----- IPC boundary (core ↔ modules) -----
|
|
srv, err := ipc.Listen(cfg.SocketPath, coreAPI)
|
|
if err != nil {
|
|
return fmt.Errorf("ipc listen: %w", err)
|
|
}
|
|
|
|
passkeySess := webauthn.NewPasskeySession(5 * time.Minute)
|
|
|
|
// Set Server.Check — the single authorization guard, run once by
|
|
// Server.dispatch before any CoreAPI method is called (see
|
|
// internal/ipc/server.go). In locked mode this is the ONLY thing
|
|
// standing between an unauthenticated caller and the store: it must
|
|
// default-deny, with an explicit allowlist for the two methods the
|
|
// unlock flow itself needs (MethodAssertStepUp, MethodUnlock — neither
|
|
// of which touches CoreAPI; dispatch handles them directly via
|
|
// srv.StepUp/srv.UnlockFn). Forgetting to allowlist a new unlock-path
|
|
// method fails safe (denied); forgetting to guard a new CoreAPI method
|
|
// is impossible because there is nothing left to forget — every method
|
|
// not in the allowlist is refused by construction.
|
|
if locked {
|
|
srv.Check = func(ctx context.Context, m ipc.Method, _ json.RawMessage) error {
|
|
switch m {
|
|
case ipc.MethodAssertStepUp, ipc.MethodUnlock, ipc.MethodPing:
|
|
// Ping is allowed for the same reason the two unlock methods
|
|
// are: it never reaches CoreAPI. It answers "she is up and
|
|
// locked", which is what mavupdate needs to tell a daemon
|
|
// waiting for a passkey apart from one that failed to start.
|
|
return nil // allowed in locked mode
|
|
default:
|
|
return errLocked
|
|
}
|
|
}
|
|
} else {
|
|
srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check
|
|
}
|
|
|
|
srv.StepUp = func(ctx context.Context) error { return passkeySess.Assert(ctx, auth.Scope{}) }
|
|
srv.LockedFn = dl.isLocked
|
|
|
|
// wg is declared here rather than next to srv.Serve because the media
|
|
// retention loop starts on this path too, and shutdown has to wait for a
|
|
// prune in flight: it deletes files.
|
|
var wg sync.WaitGroup
|
|
|
|
// 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, evBus)
|
|
wireModelSwap(srv, phr, cfg)
|
|
// Vision + the media blob store (Vikunja #252). Both stay dark without a
|
|
// media block; MethodDescribeImage answers ErrUnknownMethod then.
|
|
keeper := wireVision(ctx, &wg, srv, st, embedderOf(voiceW), cfg)
|
|
// The meeting recorder (Vikunja #253) shares that blob store and its
|
|
// retention loop. Off unless a capture block enables it, in which case
|
|
// all four capture methods answer ErrUnknownMethod.
|
|
wireCapture(ctx, &wg, srv, keeper, st, voiceW, phr, cfg)
|
|
// Voice identification (Vikunja #255). Enrolment plumbing only until a
|
|
// speaker-embedding model exists on disk; off entirely without a speaker
|
|
// block, so no wire path takes a voiceprint on a default box.
|
|
wireSpeaker(srv, st, cfg)
|
|
}
|
|
|
|
// WrapKeyFn — wraps the at-rest key under the passkey PRF secret and
|
|
// persists the wrapped blob. Called by mavweb after every assertion.
|
|
//
|
|
// It is wired in locked mode too, not only in env-key mode, and that is
|
|
// what makes a v1 blob recoverable. A box enrolled before Vikunja #14
|
|
// cold-starts through the legacy public-key retry in mavweb, and the
|
|
// StoreEncryptionKey that follows rewrites the blob as v2. Without this
|
|
// the only escape from a v1 blob was putting MAVEN_DB_KEY back in the
|
|
// environment, which is the thing cold-start unlock exists to avoid.
|
|
//
|
|
// webauthn.WrapKey refuses anything that is not a 32-byte PRF output, so
|
|
// an authenticator without PRF support produces no wrapped file at all
|
|
// rather than a file that looks protected and is not.
|
|
if envKeyBytes != nil || locked {
|
|
srv.WrapKeyFn = func(ctx context.Context, secret []byte, explicit bool) error {
|
|
kp := dbKey.Load()
|
|
if kp == nil {
|
|
return errors.New("wrap encryption key: the daemon is locked and has no key yet (unlock first)")
|
|
}
|
|
wp := wrappedPath()
|
|
// Asserting a passkey is not a request to rewrite the cold-start
|
|
// key. Without this an assertion carrying a substituted PRF value
|
|
// re-wrapped the real database key under it, and a second
|
|
// authenticator silently replaced the first one's blob.
|
|
if !explicit {
|
|
if _, err := os.Stat(wp); err == nil {
|
|
return nil
|
|
} else if !errors.Is(err, os.ErrNotExist) {
|
|
return fmt.Errorf("check wrapped key: %w", err)
|
|
}
|
|
}
|
|
wrote, err := wrapKeyToFile(wp, *kp, secret)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if wrote {
|
|
log.Printf("mavend: wrapped encryption key under this passkey's PRF output → %s", wp)
|
|
}
|
|
return nil
|
|
}
|
|
}
|
|
|
|
// UnlockFn — cold-start unlock: unwraps the encryption key from the
|
|
// wrapped blob using the passkey PRF secret, opens the store, wires all
|
|
// daemon components, and replaces the locked API.
|
|
if locked {
|
|
var unlockMu sync.Mutex
|
|
srv.UnlockFn = func(ctx context.Context, secret []byte) error {
|
|
// One unlock at a time, and never a second one. Without this a
|
|
// concurrent pair of Unlock calls would each open a store and
|
|
// wire a full daemon, and the loser's goroutines would run
|
|
// against a store nobody closes.
|
|
unlockMu.Lock()
|
|
defer unlockMu.Unlock()
|
|
if !dl.isLocked() {
|
|
return nil // already unlocked; the caller does not need to know
|
|
}
|
|
|
|
// Depth, not a boundary. MethodAssertStepUp is AuthRead, so
|
|
// anything that can open the same-uid socket can flip the
|
|
// session and reach MethodUnlock. What actually stops a local
|
|
// attacker is the 32-byte PRF output they do not have, and that
|
|
// was true before this check. What this check stops is an
|
|
// accidental unlock attempt from an unrelated local caller.
|
|
if !passkeySess.IsStepUp() {
|
|
return errors.New("unlock: no verified passkey assertion (assert first)")
|
|
}
|
|
|
|
wp := wrappedPath()
|
|
blob, err := os.ReadFile(wp)
|
|
if err != nil {
|
|
return fmt.Errorf("read wrapped key: %w", err)
|
|
}
|
|
key, version, err := webauthn.UnwrapKey(blob, secret)
|
|
if err != nil {
|
|
return fmt.Errorf("unwrap key: %w", err)
|
|
}
|
|
if version == webauthn.BlobV1 {
|
|
log.Printf("SECURITY: %s was unwrapped from a %s blob. The wrapping key is derived from the credential PUBLIC key, which mavweb also writes to its passkeys.json — anyone holding both files can recover the database key with no authenticator. Use the \"rewrite cold-start key\" button on /auth/webauthn with a PRF-capable authenticator to replace it with a v2 blob.", wp, version)
|
|
}
|
|
// WrapKeyFn needs the key to be able to rewrite the blob later.
|
|
keyCopy := bytes.Clone(key)
|
|
dbKey.Store(&keyCopy)
|
|
// Open the store with the unwrapped key.
|
|
st, err = store.OpenEncrypted(ctx, cfg.DBPath, cfg.DBTmpfs, key)
|
|
if err != nil {
|
|
return fmt.Errorf("unlock store: %w", err)
|
|
}
|
|
|
|
// Wire everything.
|
|
rules = wireRules(cfg)
|
|
gatherer = loop.NewGatherer(st, rules)
|
|
if cfg.QuietHours != nil {
|
|
gatherer.SetQuietHours(cfg.QuietHours.Start, cfg.QuietHours.End)
|
|
}
|
|
|
|
phr = phraser.NewStub()
|
|
if cfg.Phraser != nil {
|
|
pc := phraser.Config{
|
|
ModelPath: cfg.Phraser.ModelPath,
|
|
BinPath: cfg.Phraser.BinPath,
|
|
Listen: cfg.Phraser.Listen,
|
|
NGpuLayers: cfg.Phraser.NGpuLayers,
|
|
NCtx: cfg.Phraser.NCtx,
|
|
Timeout: time.Duration(cfg.Phraser.Timeout),
|
|
LLMNudges: cfg.Phraser.LLMNudges,
|
|
ContextBlock: contextBlockFn(cfg, time.Now),
|
|
}
|
|
if pc.BinPath == "" {
|
|
pc.BinPath = "llama-server"
|
|
}
|
|
if pc.Listen == "" {
|
|
pc.Listen = "127.0.0.1:0"
|
|
}
|
|
if pc.NCtx <= 0 {
|
|
pc.NCtx = 2048
|
|
}
|
|
if pc.Timeout <= 0 {
|
|
pc.Timeout = 30 * time.Second
|
|
}
|
|
phr, err = phraser.NewLLMPhraser(ctx, pc)
|
|
if err != nil {
|
|
return fmt.Errorf("phraser: %w", err)
|
|
}
|
|
}
|
|
|
|
eco = wireEcosystem(cfg)
|
|
|
|
voiceW, err = wireVoice(cfg, coreFor(), phr, st.VectorMemory(), st, eco)
|
|
if err != nil {
|
|
return fmt.Errorf("wire voice: %w", err)
|
|
}
|
|
|
|
var ntfy delivery.Sink
|
|
if cfg.Ntfy != nil {
|
|
s, err := ntfysink.New(*cfg.Ntfy)
|
|
if err != nil {
|
|
return fmt.Errorf("wire ntfy sink: %w", err)
|
|
}
|
|
ntfy = s
|
|
}
|
|
var telegram delivery.Sink
|
|
if cfg.Telegram != nil {
|
|
s, err := telegramsink.New(*cfg.Telegram)
|
|
if err != nil {
|
|
return fmt.Errorf("wire telegram sink: %w", err)
|
|
}
|
|
telegram = s
|
|
}
|
|
var voiceSink delivery.Sink
|
|
if voiceW != nil {
|
|
voiceSink = voiceW.voiceSink
|
|
}
|
|
if _, err := st.ReconcileStaleDeliveryAttempts(context.Background(), time.Now()); err != nil {
|
|
log.Printf("delivery outbox reconcile: %v", err)
|
|
}
|
|
dispatcher = delivery.NewDispatcher(delivery.Config{
|
|
Ntfy: ntfy,
|
|
Telegram: telegram,
|
|
Voice: voiceSink,
|
|
Ack: st,
|
|
Nudges: st,
|
|
Reminders: st,
|
|
Outbox: st,
|
|
})
|
|
|
|
tickInterval := time.Duration(cfg.TickInterval)
|
|
repeatInterval := time.Duration(cfg.RepeatInterval)
|
|
autotuneInterval := time.Duration(cfg.AutotuneInterval)
|
|
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(coreFor(), embedderOf(voiceW), cfg)
|
|
crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
|
|
|
|
// Swap the CoreAPI from the locked placeholder to the real store adapter.
|
|
newAPI := &daemonAPI{
|
|
CoreAPI: coreFor(),
|
|
getTrace: tl.trace,
|
|
getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return tl.morningStatus(ctx, time.Now()) },
|
|
getDayPlan: func(ctx context.Context) ipc.DayPlan { return tl.dayPlan(ctx, time.Now()) },
|
|
getEvents: intakeEventsFn(evBus),
|
|
}
|
|
if voiceW != nil && voiceW.handler != nil {
|
|
newAPI.chatFn = voiceW.handler.handleText
|
|
voiceW.handler.upgradeAPI(newAPI)
|
|
}
|
|
srv.SetAPI(newAPI)
|
|
srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check
|
|
wireMailIntake(srv, st, phr, cfg, evBus)
|
|
wireModelSwap(srv, phr, cfg)
|
|
keeper := wireVision(ctx, &wg, srv, st, embedderOf(voiceW), cfg)
|
|
wireCapture(ctx, &wg, srv, keeper, st, voiceW, phr, cfg)
|
|
// Voice identification (Vikunja #255). Enrolment plumbing only until a
|
|
// speaker-embedding model exists on disk; off entirely without a speaker
|
|
// block, so no wire path takes a voiceprint on a default box.
|
|
wireSpeaker(srv, st, cfg)
|
|
|
|
// Start voice server.
|
|
if voiceW != nil {
|
|
var wg sync.WaitGroup
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if err := voiceW.server.Serve(); err != nil && !errors.Is(err, net.ErrClosed) {
|
|
log.Printf("voice serve: %v", err)
|
|
}
|
|
}()
|
|
log.Printf("mavend: voice listening on %s", voiceW.server.Addr())
|
|
}
|
|
|
|
// Start tick loop.
|
|
go func() {
|
|
tl.run(ctx)
|
|
}()
|
|
|
|
// Start fact-entity enrichment worker.
|
|
go func() {
|
|
factWorker.run(ctx)
|
|
}()
|
|
|
|
// Start background memory evaluation (nil unless configured).
|
|
if evalWorker != nil {
|
|
go func() {
|
|
evalWorker.run(ctx)
|
|
}()
|
|
}
|
|
|
|
// 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)
|
|
}()
|
|
}
|
|
|
|
// Keep MCP connections alive (nil unless configured).
|
|
if voiceW != nil && voiceW.mcp != nil {
|
|
go voiceW.mcp.run(ctx)
|
|
}
|
|
|
|
// Re-enumerate the house for new devices (nil unless configured).
|
|
if voiceW != nil && voiceW.home != nil {
|
|
go voiceW.home.run(ctx)
|
|
}
|
|
|
|
dl.unlock(st)
|
|
log.Printf("mavend: unlocked via passkey assertion")
|
|
return nil
|
|
}
|
|
}
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if err := srv.Serve(); err != nil && !errors.Is(err, net.ErrClosed) {
|
|
log.Printf("ipc serve: %v", err)
|
|
}
|
|
}()
|
|
log.Printf("mavend: ipc listening on %s", srv.Path())
|
|
|
|
if !locked && voiceW != nil {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if err := voiceW.server.Serve(); err != nil && !errors.Is(err, net.ErrClosed) {
|
|
log.Printf("voice serve: %v", err)
|
|
}
|
|
}()
|
|
log.Printf("mavend: voice listening on %s", voiceW.server.Addr())
|
|
}
|
|
|
|
if !locked {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
tl.run(ctx)
|
|
}()
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
factWorker.run(ctx)
|
|
}()
|
|
if evalWorker != nil {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
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)
|
|
}()
|
|
}
|
|
if voiceW != nil && voiceW.mcp != nil {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
voiceW.mcp.run(ctx)
|
|
}()
|
|
}
|
|
if voiceW != nil && voiceW.home != nil {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
voiceW.home.run(ctx)
|
|
}()
|
|
}
|
|
}
|
|
|
|
<-ctx.Done()
|
|
log.Printf("mavend: shutdown signal received")
|
|
if err := srv.Close(); err != nil {
|
|
log.Printf("ipc close: %v", err)
|
|
}
|
|
if voiceW != nil {
|
|
voiceW.close()
|
|
}
|
|
// Bounded. Every worker below watches ctx, but one parked in a model call
|
|
// or an HTTP fetch can outlast the supervisor's patience, and run() has to
|
|
// return for `defer st.Close()` to seal the database. A worker abandoned
|
|
// mid-tick loses one tick; a shutdown that never returns loses every write
|
|
// since the last clean stop — which is how the deployed ciphertext went
|
|
// eleven days stale in July 2026.
|
|
if !waitWorkers(&wg, workerGrace) {
|
|
log.Printf("mavend: workers still running after %s, sealing anyway", workerGrace)
|
|
}
|
|
log.Printf("mavend: bye")
|
|
return nil
|
|
}
|
|
|
|
// personaFacts reads the optional, deployment-specific facts (his name, his
|
|
// city, the free-text persona string) out of the config. Everything here may
|
|
// be empty — the context block is correct without any of it.
|
|
func personaFacts(cfg *config.Config) persona.Facts {
|
|
f := persona.Facts{
|
|
// Telegram lives outside the voice block, so it counts either way.
|
|
Telegram: cfg.Telegram != nil && cfg.Telegram.BotToken != "" && cfg.Telegram.ChatID != "",
|
|
}
|
|
if cfg.Voice == nil {
|
|
return f
|
|
}
|
|
f.OwnerName = cfg.Voice.OwnerName
|
|
f.City = cfg.Voice.City
|
|
f.Static = cfg.Voice.Persona
|
|
// Same test wireVoice uses to pick the real provider over the stub.
|
|
f.Weather = cfg.Voice.Weather != nil && cfg.Voice.Weather.Provider == "open-meteo"
|
|
f.Tools = len(cfg.Voice.Tools) > 0
|
|
return f
|
|
}
|
|
|
|
// contextBlockFn returns the per-turn renderer of the shared context block.
|
|
// Per turn, not once at startup, because the block states the current time.
|
|
func contextBlockFn(cfg *config.Config, now func() time.Time) func() string {
|
|
f := personaFacts(cfg)
|
|
return func() string { return f.Block(now()) }
|
|
}
|
|
|
|
// workerGrace — how long shutdown waits for the background workers before it
|
|
// goes ahead and seals without them. Comfortably inside docker's ten-second
|
|
// default so the seal still lands before SIGKILL.
|
|
const workerGrace = 4 * time.Second
|
|
|
|
// waitWorkers waits on wg for at most d. Reports whether they all finished.
|
|
func waitWorkers(wg *sync.WaitGroup, d time.Duration) bool {
|
|
done := make(chan struct{})
|
|
go func() {
|
|
wg.Wait()
|
|
close(done)
|
|
}()
|
|
select {
|
|
case <-done:
|
|
return true
|
|
case <-time.After(d):
|
|
return false
|
|
}
|
|
}
|
|
|
|
// wireRules builds the nudge rule set, minus anything config turned off. The
|
|
// drop is logged because a rule vanishing silently is indistinguishable from a
|
|
// rule that is broken, and the next person to wonder why she stopped nudging
|
|
// should find the answer in the boot log.
|
|
func wireRules(cfg *config.Config) []loop.Rule {
|
|
rules, dropped := loop.RulesExcept(cfg.DisabledRules)
|
|
for _, name := range dropped {
|
|
log.Printf("loop: rule %q disabled by config", name)
|
|
}
|
|
return rules
|
|
}
|