Files
Maven/cmd/mavend/main.go
T
kami 2bf11f052d Merge branch 'fix/g07' into fix/integrated
# Conflicts:
#	internal/ipc/api.go
#	internal/ipc/client.go
#	internal/llm/client.go
2026-08-01 14:36:48 +04:00

781 lines
26 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 = loop.DefaultRules()
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
}
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
// 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, 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(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 = loop.DefaultRules()
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
}
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, srv, st, embedderOf(voiceW), cfg)
wireCapture(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
}
}
var wg sync.WaitGroup
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()
}
wg.Wait()
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()) }
}