diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 7c5d805..9822dc0 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -68,6 +68,11 @@ import ( "github.com/kami/maven/internal/webauthn" ) +// stepUpTTL is how long one passkey assertion keeps the session stepped up. +// Long enough for the unlock call that follows it, short enough that a walked +// away laptop does not stay authorized. +const stepUpTTL = 5 * time.Minute + var errLocked = errors.New("mavend: daemon locked — complete passkey assertion first") // daemonLock tracks whether the daemon is in locked (pre-unlock) mode, and @@ -249,42 +254,12 @@ func run(args []string) error { if !locked { rules = wireRules(cfg) - gatherer = loop.NewGatherer(st, rules) - if cfg.QuietHours != nil { - gatherer.SetQuietHours(cfg.QuietHours.Start, cfg.QuietHours.End) - } + gatherer = wireGatherer(st, cfg, rules) // 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, - CacheRAMMiB: cacheRAMMiB(cfg.Phraser.CacheRAMMiB), - 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) - } + phr, err = wirePhraser(ctx, cfg) + if err != nil { + return fmt.Errorf("phraser: %w", err) } // ecosystem — nexus + hexis + praxis (all over HTTP; no direct DB access) @@ -297,48 +272,13 @@ func run(args []string) error { } // 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 + dispatcher, err = wireDispatcher(st, cfg, voiceW) + if err != nil { + return err } - 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) + tl = wireTickLoop(st, gatherer, dispatcher, phr, rules, cfg) factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval)) evalWorker = newMemoryEvalWorker(st, phr, cfg) feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg) @@ -380,7 +320,7 @@ func run(args []string) error { return fmt.Errorf("ipc listen: %w", err) } - passkeySess := webauthn.NewPasskeySession(5 * time.Minute) + passkeySess := webauthn.NewPasskeySession(stepUpTTL) // Set Server.Check — the single authorization guard, run once by // Server.dispatch before any CoreAPI method is called (see @@ -528,40 +468,11 @@ func run(args []string) error { // Wire everything. rules = wireRules(cfg) - gatherer = loop.NewGatherer(st, rules) - if cfg.QuietHours != nil { - gatherer.SetQuietHours(cfg.QuietHours.Start, cfg.QuietHours.End) - } + gatherer = wireGatherer(st, cfg, rules) - 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, - CacheRAMMiB: cacheRAMMiB(cfg.Phraser.CacheRAMMiB), - 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) - } + phr, err = wirePhraser(ctx, cfg) + if err != nil { + return fmt.Errorf("phraser: %w", err) } eco = wireEcosystem(cfg) @@ -571,43 +482,12 @@ func run(args []string) error { 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 + dispatcher, err = wireDispatcher(st, cfg, voiceW) + if err != nil { + return err } - 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) + tl = wireTickLoop(st, gatherer, dispatcher, phr, rules, cfg) factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval)) evalWorker = newMemoryEvalWorker(st, phr, cfg) feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg) @@ -850,6 +730,115 @@ func waitWorkers(wg *sync.WaitGroup, d time.Duration) bool { } } +// Phraser defaults, applied when the config block leaves a field unset. They +// are the daemon's, not the library's: phraser.Config carries no defaults of +// its own, so an empty field here would reach llama-server as an empty flag. +const ( + defaultLlamaBin = "llama-server" + defaultPhraserListen = "127.0.0.1:0" + defaultPhraserNCtx = 2048 + defaultPhraserTimeout = 30 * time.Second +) + +// wirePhraser builds the phrasing seam. No phraser block means the +// deterministic stub, which is the floor and not an error: the daemon answers +// without a model, in fixed words. +func wirePhraser(ctx context.Context, cfg *config.Config) (phraser.Phraser, error) { + if cfg.Phraser == nil { + return phraser.NewStub(), nil + } + pc := phraser.Config{ + ModelPath: cfg.Phraser.ModelPath, + BinPath: cfg.Phraser.BinPath, + Listen: cfg.Phraser.Listen, + NGpuLayers: cfg.Phraser.NGpuLayers, + NCtx: cfg.Phraser.NCtx, + CacheRAMMiB: cacheRAMMiB(cfg.Phraser.CacheRAMMiB), + Timeout: time.Duration(cfg.Phraser.Timeout), + LLMNudges: cfg.Phraser.LLMNudges, + ContextBlock: contextBlockFn(cfg, time.Now), + } + if pc.BinPath == "" { + pc.BinPath = defaultLlamaBin + } + if pc.Listen == "" { + pc.Listen = defaultPhraserListen + } + if pc.NCtx <= 0 { + pc.NCtx = defaultPhraserNCtx + } + if pc.Timeout <= 0 { + pc.Timeout = defaultPhraserTimeout + } + return phraser.NewLLMPhraser(ctx, pc) +} + +// wireGatherer builds the nudge gatherer over the given rule set and applies +// the configured quiet hours. +func wireGatherer(st *store.Store, cfg *config.Config, rules []loop.Rule) *loop.Gatherer { + g := loop.NewGatherer(st, rules) + if cfg.QuietHours != nil { + g.SetQuietHours(cfg.QuietHours.Start, cfg.QuietHours.End) + } + return g +} + +// wireDispatcher builds the delivery fan-out. Each sink stays nil unless its +// config block is present, and a sink that fails to build fails the boot +// rather than going quiet. +// +// A crashed prior run may have left "pending" delivery attempts (send may have +// landed externally, then the process died before recording it). They are +// reconciled to "unknown" here, before the tick loop resumes sending, so +// nothing auto-resends into that ambiguity. +func wireDispatcher(st *store.Store, cfg *config.Config, voiceW *voiceWiring) (*delivery.Dispatcher, error) { + var ntfy delivery.Sink + if cfg.Ntfy != nil { + s, err := ntfysink.New(*cfg.Ntfy) + if err != nil { + return nil, 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 nil, 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) + } + return delivery.NewDispatcher(delivery.Config{ + Ntfy: ntfy, + Telegram: telegram, + Voice: voiceSink, + Ack: st, + Nudges: st, + Reminders: st, + Outbox: st, + }), nil +} + +// wireTickLoop reads the loop's three intervals and its schedules out of the +// config, so the two boot paths cannot disagree about them. +func wireTickLoop(st *store.Store, gatherer *loop.Gatherer, dispatcher *delivery.Dispatcher, phr phraser.Phraser, rules []loop.Rule, cfg *config.Config) *tickLoop { + return newTickLoop(st, gatherer, dispatcher, phr, rules, + time.Duration(cfg.TickInterval), + time.Duration(cfg.RepeatInterval), + time.Duration(cfg.AutotuneInterval), + cfg.Digest, + routinesFromConfig(cfg.Routines), + config.MorningRoutinesFromConfig(cfg.MorningRoutines), + cfg.PatternProposals) +} + // 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