diff --git a/.gitignore b/.gitignore index c3476b5..37d5c7e 100644 --- a/.gitignore +++ b/.gitignore @@ -1,10 +1,13 @@ -# Built binaries -mavend -mavenclient -mavsttd -mavttsd -mavweb -mavpoll +# Built binaries (root only) +/mavend +/mavenclient +/mavsttd +/mavttsd +/mavweb +/mavpoll + +# Certs (private keys, don't commit) +certs/ # Dependencies (fetch/build, not vendored) deps/ @@ -15,8 +18,8 @@ models/ # Runtime data *.db -# Certs (private keys, don't commit) -certs/ - # Temp files /tmp/ + +# opencode config (per-project, not for sharing) +opencode.json diff --git a/Makefile b/Makefile index 7e41550..eb3e487 100644 --- a/Makefile +++ b/Makefile @@ -41,7 +41,7 @@ run-web: build-web test: CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" LD_LIBRARY_PATH="$(shell pwd)/deps/lib" \ - $(GO) test ./internal/... + $(GO) test ./internal/... ./cmd/... run-stt: build-stt LD_LIBRARY_PATH="$(shell pwd)/deps/lib" \ diff --git a/cmd/mavenclient/main.go b/cmd/mavenclient/main.go new file mode 100644 index 0000000..fa6ef8a --- /dev/null +++ b/cmd/mavenclient/main.go @@ -0,0 +1,163 @@ +// Package main is mavenclient — maven's reference client. +// +// Per the spec (maven.md § stt/tts): capture lives on the client; the +// server transcribes + synthesises on demand. The PC client runs vosk-ru +// (wake-word + stage-0 grammar, real-time on a Pi) + VAD, ships ONE clean +// audio blob per utterance on activation. The server never owns a mic. +// +// This binary is the floor reference: there is NO wake-word / VAD here +// (production PC client libraries); it ships ONE wav file from disk per +// invocation, posts it via voice.PushToTalk, and writes the reply audio to +// a wav file (or stdout). The point is to round-trip the daemon's reactive +// path end-to-end with the real TCP wire, not to be the production client. +// +// usage: +// mavenclient -in audio.wav -out reply.wav +// mavenclient -in audio.wav # reply written to ./reply.wav +// mavenclient -addr 127.0.0.1:9100 # default; production = wg-tunnel addr +// +// -listen mode keeps the conn open and writes incoming Push frames +// (proactive nudge audio) to disk sequentially — exercise proactive +// delivery end-to-end. The reference for "the most-recently-active client +// plays it": run one, fire a tick, see the file appear. +// +// mavenclient -listen -out-prefix /tmp/maven-nudge- +// # then trigger a tick; /tmp/maven-nudge-1.wav, -2.wav ... appear +package main + +import ( + "context" + "encoding/json" + "errors" + "flag" + "fmt" + "io" + "log" + "net" + "os" + "os/signal" + "path/filepath" + "strconv" + "syscall" + "time" + + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/voice" +) + +func main() { + if err := run(os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, "mavenclient:", err) + os.Exit(1) + } +} + +func run(args []string) error { + addr := flag.String("addr", "127.0.0.1:9100", "mavend voice address (TCP; inside wg tunnel in prod)") + inPath := flag.String("in", "", "input WAV (canonical 16k mono int16); required for one-shot") + outPath := flag.String("out", "", "output WAV (default ./reply.wav or nudge-N.wav for -listen)") + lang := flag.String("lang", "ru", "BCP-47 lang hint for stt ('ru' | 'en' | 'mixed')") + listen := flag.Bool("listen", false, "stay open + write incoming Push frames to disk") + outPrefix := flag.String("out-prefix", "", "-listen: prefix for received-nudge wav files (default ./nudge-)") + flag.CommandLine.Parse(args) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + defer stop() + + c := voice.Dial(*addr) + defer c.Close() + + if *listen { + return runListen(ctx, c, *outPrefix) + } + if *inPath == "" { + flag.Usage() + return errors.New("-in is required for one-shot mode (or use -listen)") + } + return runOneShot(ctx, c, *inPath, *outPath, *lang) +} + +func runOneShot(ctx context.Context, c *voice.Client, inPath, outPath, lang string) error { + wav, err := os.ReadFile(inPath) + if err != nil { + return fmt.Errorf("read input: %w", err) + } + format, pcm, err := audio.PCMFromWAV(wav) + if err != nil { + return err + } + log.Printf("mavenclient: loaded %s (%.2fs, %+v)", inPath, float64(len(pcm))/float64(format.SampleRate)/float64(format.SampleBits/8), format) + + resp, err := c.PushToTalk(ctx, audio.Audio{Format: format, Bytes: pcm}, lang) + if err != nil { + return fmt.Errorf("voice round-trip: %w", err) + } + if outPath == "" { + outPath = "reply.wav" + } + if err := writeWAV(outPath, resp.ReplyAudio); err != nil { + return err + } + log.Printf("mavenclient: reply (%.2fs, %q) -> %s", resp.ReplyAudio.Duration(), resp.ReplyText, outPath) + if len(resp.RoutedChannels) > 0 { + log.Printf("mavenclient: also routed to: %v", resp.RoutedChannels) + } + return nil +} + +func runListen(ctx context.Context, c *voice.Client, prefix string) error { + if prefix == "" { + prefix = "nudge-" + } + h := &filePushHandler{prefix: prefix, counter: 0} + log.Printf("mavenclient: listening for server pushes; writing to %s*.wav", prefix) + return c.RunPushReceiver(ctx, h) +} + +type filePushHandler struct { + prefix string + counter int +} + +func (h *filePushHandler) OnPush(p voice.Push) { + switch p.Kind { + case voice.PushKindAudioNudge: + var ap voice.AudioNudgePush + if err := jsonUnmarshal(p.Params, &ap); err != nil { + log.Printf("mavenclient: bad audio_nudge push: %v", err) + return + } + h.counter++ + name := fmt.Sprintf("%s%d.wav", h.prefix, h.counter) + if err := writeWAV(name, ap.Audio); err != nil { + log.Printf("mavenclient: write %s: %v", name, err) + return + } + log.Printf("mavenclient: nudge %q (sev %d) -> %s (%.2fs) %q", ap.RuleName, ap.Severity, name, ap.Audio.Duration(), ap.Text) + case voice.PushKindPing: + // liveness; ignore. + default: + log.Printf("mavenclient: unknown push kind %q", p.Kind) + } +} + +func writeWAV(path string, a audio.Audio) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); filepath.Dir(path) != "" && err != nil { + return err + } + wav, err := audio.WAVFromPCM(a.Format, a.Bytes) + if err != nil { + return err + } + return os.WriteFile(path, wav, 0o644) +} + +// jsonUnmarshal — kept local rather than pulling encoding/json into main.go +// top-level space. +func jsonUnmarshal(b []byte, v any) error { return json.Unmarshal(b, v) } + +// keep strconv + io + net + time alive for future duration/size helpers. +var _ = strconv.Atoi +var _ io.Reader = (io.Reader)(nil) +var _ = net.IPv4 +var _ = time.Second \ No newline at end of file diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go new file mode 100644 index 0000000..d19a8dc --- /dev/null +++ b/cmd/mavend/main.go @@ -0,0 +1,212 @@ +// 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). +// +// Everything wired here is swappable at the construction seam — every module +// behind an interface — without changing the loop. Wire a real phraser, voice +// sink, or session verifier, and the daemon's main loop is unchanged. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net" + "os" + "os/signal" + "sync" + "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/phraser" + "github.com/kami/maven/internal/store" +) + +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") + flag.CommandLine.Parse(args) + cfg, err := config.Load(*cfgPath) + if err != nil { + return err + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + defer stop() + + // ----- store (the unlocked handle; core = the only key-holder) ----- + st, err := store.Open(ctx, cfg.DBPath) + if err != nil { + return fmt.Errorf("open store: %w", err) + } + defer st.Close() + + // ----- loop: rules + gatherer ----- + + rules := loop.DefaultRules() + gatherer := loop.NewGatherer(st, rules) + + // ----- phraser (LLM-backed when configured, Stub floor otherwise) ----- + var phr phraser.Phraser = 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), + } + 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) + } + } + defer phr.Close() + + // ----- voice: reactive audio path (TCP listener + stt/router/tts) ----- + voiceW, err := wireVoice(cfg, ipc.NewStoreAPI(st)) + if err != nil { + return fmt.Errorf("wire voice: %w", err) + } + defer voiceW.close() + + // ----- delivery: sinks + dispatcher ----- + 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 + } + // Voice sink: nil when voice is not enabled — the dispatcher's nil-sink + // path skips ChannelVoice silently, just like the pre-voice floor). + var voiceSink delivery.Sink + if voiceW != nil { + voiceSink = voiceW.voiceSink + } + dispatcher := delivery.NewDispatcher(delivery.Config{ + Ntfy: ntfy, + Telegram: telegram, + Voice: voiceSink, + // AckTracker nil ⇒ repeat-til-ack disabled in the dispatcher. We + // drive repeats from store.UnackedTelegramRules + Dispatcher.RepeatUnacked + // below, which uses the nudges table's outcome=pending row itself as + // the ack-or-not state — the production ack source. The AckTracker + // interface stays reserved for an in-memory cache if the daemon ever + // wants to drive repeats without the SQL hit; the table IS the truth. + Nudges: st, // *store.Store satisfies delivery.NudgeRecorder + Reminders: st, // *store.Store satisfies delivery.ReminderCompleter + }) + + // ----- the proactive loop driver (60s ticker, lives HERE per spec) ----- + tickInterval := time.Duration(cfg.TickInterval) + repeatInterval := time.Duration(cfg.RepeatInterval) + autotuneInterval := time.Duration(cfg.AutotuneInterval) + loop := newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval) + + // ----- IPC boundary (core ↔ modules) ----- + coreAPI := ipc.NewStoreAPI(st) + srv, err := ipc.Listen(cfg.SocketPath, coreAPI) + if err != nil { + return fmt.Errorf("ipc listen: %w", err) + } + // auth floor: any same-uid caller is fully trusted (FloorEnrollment + + // FloorSession — L3, step-up satisfied). The cold-start unlock dance and a + // real passkey Session are the open spec items; today the daemon runs + // unlocked — plain sqlite, sqlcipher deferred. FloorSession keeps the floor + // consistent so the authed mavweb /tools page can EnableTool (AuthStepUp) + // against the local socket; the passkey verifier swaps FloorSession later. + srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: auth.FloorSession{}}).Check + + 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 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()) + } + + wg.Add(1) + go func() { + defer wg.Done() + loop.run(ctx) + }() + + <-ctx.Done() + log.Printf("mavend: shutdown signal received") + if err := srv.Close(); err != nil { + log.Printf("ipc close: %v", err) + } + wg.Wait() + log.Printf("mavend: bye") + return nil +} \ No newline at end of file diff --git a/cmd/mavend/seedtest/main.go b/cmd/mavend/seedtest/main.go new file mode 100644 index 0000000..d0c8d3c --- /dev/null +++ b/cmd/mavend/seedtest/main.go @@ -0,0 +1,43 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "os" + "time" + + "github.com/kami/maven/internal/store" +) + +func main() { + ctx := context.Background() + st, err := store.Open(ctx, "/tmp/maven-test.db") + if err != nil { + fmt.Fprintln(os.Stderr, "open:", err) + os.Exit(1) + } + defer st.Close() + + now := time.Now() + _, err = st.WriteFact(ctx, now.Add(-4*time.Hour), store.KindSelf, "water", "1", "tap:voice", 1.0, sql.NullInt64{}) + if err != nil { + fmt.Fprintln(os.Stderr, "write water fact:", err) + os.Exit(1) + } + fmt.Println("seeded water fact 4h ago") + + _, err = st.WriteFact(ctx, now.Add(-time.Minute), store.KindSelf, "desk_active", "1", "presence", 1.0, sql.NullInt64{}) + if err != nil { + fmt.Fprintln(os.Stderr, "write desk_active fact:", err) + os.Exit(1) + } + fmt.Println("seeded desk_active (presence) 1m ago") + + _, err = st.SetValue(ctx, store.KindConfig, "quiet_hours", "config", "false", now) + if err != nil { + fmt.Fprintln(os.Stderr, "write quiet_hours:", err) + os.Exit(1) + } + fmt.Println("seeded quiet_hours=false") +} diff --git a/cmd/mavend/tick.go b/cmd/mavend/tick.go new file mode 100644 index 0000000..bde17aa --- /dev/null +++ b/cmd/mavend/tick.go @@ -0,0 +1,249 @@ +// mavend/tick.go — the proactive loop driver. +// +// Per spec the 60s schedule loop (`for { tick; sleep }`) lives in the daemon +// main, NOT in `internal/loop/` — that keeps the loop package pure + unit- +// testable without time side effects. the driver here is the ONE impure +// orchestrator: it gathers state under the store lock, runs the pure Tick, +// phrases the candidate, dispatches it, then handles reminders + sev4 repeats +// and runs the feedback auto-tuner on its own slow cadence. +package main + +import ( + "context" + "fmt" + "log" + "os" + "path/filepath" + "sync" + "time" + + "github.com/kami/maven/internal/delivery" + "github.com/kami/maven/internal/loop" + "github.com/kami/maven/internal/phraser" + "github.com/kami/maven/internal/store" +) + +// tickLoop — the impure driver. holds everything wired at daemon construction +// that the per-tick path needs. the rules slice is read-only here; the gatherer +// already captured it, but we keep it for a possible future re-seed path. +type tickLoop struct { + store *store.Store + gatherer *loop.Gatherer + dispatcher *delivery.Dispatcher + phraser phraser.Phraser + rules []loop.Rule + + tickInterval time.Duration + repeatInterval time.Duration + autotuneInterval time.Duration // 0 ⇒ autotune disabled (gatherer falls back to static Base) + + // lastPhrase caches the phraser output per rule so the sev4 repeat path + // can re-send roughly what the user was first alerted with (an alarm + // that re-phrases differently every 5m is hostile; the same terse body + // IS the insistence signal). keyed by rule name. nil phrase for a rule + // = no successful initial dispatch yet (cold-start edge — fall back to + // a generic body). + mu sync.Mutex + lastPhrase map[string]delivery.PhrasedNudge +} + +func newTickLoop( + st *store.Store, + g *loop.Gatherer, + d *delivery.Dispatcher, + p phraser.Phraser, + rules []loop.Rule, + tickInterval, repeatInterval, autotuneInterval time.Duration, +) *tickLoop { + return &tickLoop{ + store: st, + gatherer: g, + dispatcher: d, + phraser: p, + rules: rules, + tickInterval: tickInterval, + repeatInterval: repeatInterval, + autotuneInterval: autotuneInterval, + lastPhrase: make(map[string]delivery.PhrasedNudge), + } +} + +// run drives the loop until ctx is canceled. one tick per tickInterval; +// the first tick fires immediately so a freshly-started daemon doesn't sit +// idle for 60s before its first evaluation (cold-start responsiveness). the +// feedback auto-tuner runs on its own slower ticker (autotuneInterval) so it +// doesn't write a feedback fact every tick — append-only facts would churn. +func (t *tickLoop) run(ctx context.Context) { + t.tick(ctx, time.Now()) + ticker := time.NewTicker(t.tickInterval) + defer ticker.Stop() + + var autotune *time.Ticker + var autotuneC <-chan time.Time + if t.autotuneInterval > 0 { + autotune = time.NewTicker(t.autotuneInterval) + defer autotune.Stop() + autotuneC = autotune.C + } + + for { + select { + case <-ctx.Done(): + return + case now := <-ticker.C: + t.tick(ctx, now) + case <-autotuneC: + t.tune(ctx) + } + } +} + +// tick — one pass of the proactive loop. gathers, decides, phrases, delivers. +// errors at any sub-step are logged and the tick continues / aborts as the +// layer warrants: a gather failure aborts (no consistent snapshot ⇒ no +// decisions); a phrase/dispatch failure logs the failure and continues so a +// transient delivery fault doesn't kill the whole loop. +func (t *tickLoop) tick(ctx context.Context, now time.Time) { + state, due, err := t.gatherer.GatherState(ctx, now) + if err != nil { + log.Printf("tick: gather: %v", err) + return + } + + // proactive: at most one candidate, max severity. + if cand := loop.Tick(state, t.rules); cand != nil { + pn, err := t.phraser.PhraseNudge(ctx, *cand) + if err != nil { + log.Printf("tick: phrase nudge %s: %v", cand.Rule.Name, err) + } else { + t.cachePhrase(pn) + if _, err := t.dispatcher.DispatchNudge(ctx, pn, now); err != nil { + log.Printf("tick: dispatch nudge %s: %v", cand.Rule.Name, err) + } + } + } + + // reminders: gate-bypassing class. fired once, marked after a successful + // delivery. a failed send leaves the reminder pending — the next tick + // re-gathers and re-attempts. + for _, d := range loop.RemindDecisions(state, due) { + pr, err := t.phraser.PhraseReminder(ctx, d) + if err != nil { + log.Printf("tick: phrase reminder %d: %v", d.Reminder.ID, err) + continue + } + if _, err := t.dispatcher.DispatchReminder(ctx, pr, now); err != nil { + log.Printf("tick: dispatch reminder %d: %v", d.Reminder.ID, err) + } + } + + // sev4-away repeats: re-send un-acked telegram nudges per repeatInterval. + // the source of ack truth IS the nudges table (outcome=pending ⇒ not + // acked); store.UnackedTelegramRules surfaces the keys. body/summary come + // from the cached phrase from the initial dispatch — see lastPhrase notes. + // if not cached (cold-start mid-alarm), fall back to a terse generic body. + keys, err := t.store.UnackedTelegramRules(ctx) + if err != nil { + log.Printf("tick: unacked telegram rules: %v", err) + return + } + if len(keys) == 0 { + return + } + for _, key := range keys { + body, summary := t.repeatPhrase(key) + if _, err := t.dispatcher.RepeatUnacked(ctx, []string{key}, now, t.repeatInterval, body, summary); err != nil { + log.Printf("tick: repeat telegram %s: %v", key, err) + } + } +} + +// cachePhrase keeps the latest phrased nudge per rule for the sev4-repeat +// path. writing under a mutex; the repeat path reads under the same. the +// cache is bounded by the rule count (≤ ~30 per spec) so eviction is not a +// concern at this scale. +func (t *tickLoop) cachePhrase(pn delivery.PhrasedNudge) { + t.mu.Lock() + t.lastPhrase[pn.Candidate.Rule.Name] = pn + t.mu.Unlock() +} + +func (t *tickLoop) repeatPhrase(rule string) (body, summary string) { + t.mu.Lock() + pn, ok := t.lastPhrase[rule] + t.mu.Unlock() + if !ok || pn.Summary == "" { + // cold-start mid-alarm: no cached phrase. a deliberately terse generic + // body — the alarm IS the insistence; the wording repeats, the ring + // is what changes. the LLM phraser impl will refresh this on its next + // tick when the rule re-fires through Tick. + return fmt.Sprintf("maven: %s still active", rule), rule + } + return pn.Body, pn.Summary +} + +// tune — the feedback auto-tuner's impure step. runs on a slow cadence +// (autotuneInterval, see run) so it doesn't write a fact every tick. for each +// rule: +// 1. read store.RecentOutcomes for the last TuneSampleN resolved outcomes. +// 2. if there's not enough signal (TuneMinOutcomes), leave Base alone. +// 3. compute the tuned cooldown with loop.TuneCooldown (pure). +// 4. read the currently-persisted feedback fact; if the tuned value equals +// it, skip the write (RecentOutcomes is itself steady ⇒ no churn). +// 5. else write a `facts (kind=config, source=feedback, key=cooldown:)` +// row. the Gatherer reads it next tick. +// +// Error at any step logs + continues to the next rule — a transient store +// fault on one rule must not abort tuning for the rest. +func (t *tickLoop) tune(ctx context.Context) { + now := time.Now() + for _, r := range t.rules { + outcomes, err := t.store.RecentOutcomes(ctx, r.Name, loop.TuneSampleN) + if err != nil { + log.Printf("tune: outcomes %s: %v", r.Name, err) + continue + } + if len(outcomes) < loop.TuneMinOutcomes { + continue // sparse — no signal yet, don't whipsaw on first sight. + } + tuned := loop.TuneCooldown(r, outcomes) + + // the currently-persisted tuned base, if any. equal ⇒ skip the write + // (RecentOutcomes is monotone-steady between resolved outcomes). + if cur, ok := t.currentTunedBase(ctx, r); ok && cur == tuned { + continue + } + if _, err := t.store.SetValue( + ctx, store.KindConfig, loop.FeedbackKey(r), loop.FeedbackSource, + tuned, now, + ); err != nil { + log.Printf("tune: persist %s: %v", r.Name, err) + continue + } + log.Printf("tune: %s cooldown -> %v", r.Name, tuned) + } +} + +// currentTunedBase — read the persisted feedback cooldown fact back into a +// duration. (dur, false) when no feedback fact exists yet OR it's malformed; +// the caller treats that as "differ from anything we'd write — write." +func (t *tickLoop) currentTunedBase(ctx context.Context, r loop.Rule) (time.Duration, bool) { + f, err := t.store.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource) + if err != nil { + return 0, false + } + return loop.ParseCooldownFact(f) +} + +// defaultConfigPath — the config file path the daemon loads if -config wasn't +// passed. XDG_CONFIG_HOME/maven/mavend.json, falling back to ~/.config/maven. +func defaultConfigPath() string { + if x := os.Getenv("XDG_CONFIG_HOME"); x != "" { + return filepath.Join(x, "maven", "mavend.json") + } + home, err := os.UserHomeDir() + if err != nil || home == "" { + return "mavend.json" + } + return filepath.Join(home, ".config", "maven", "mavend.json") +} \ No newline at end of file diff --git a/cmd/mavend/tick_test.go b/cmd/mavend/tick_test.go new file mode 100644 index 0000000..250129c --- /dev/null +++ b/cmd/mavend/tick_test.go @@ -0,0 +1,288 @@ +package main + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/kami/maven/internal/delivery" + "github.com/kami/maven/internal/loop" + "github.com/kami/maven/internal/phraser" + "github.com/kami/maven/internal/store" +) + +// fakeSink — captures every Send for assertion. implements delivery.Sink. +type fakeSink struct { + sends []delivery.Sendable +} + +func (f *fakeSink) Send(_ context.Context, s delivery.Sendable) error { + f.sends = append(f.sends, s) + return nil +} + +func newTestStore(t *testing.T) *store.Store { + t.Helper() + path := filepath.Join(t.TempDir(), "mavend_test.db") + st, err := store.Open(context.Background(), path) + if err != nil { + t.Fatalf("store.Open: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + return st +} + +func newTestTickLoop(t *testing.T, st *store.Store, sink delivery.Sink) *tickLoop { + t.Helper() + rules := loop.DefaultRules() + g := loop.NewGatherer(st, rules) + d := delivery.NewDispatcher(delivery.Config{ + Voice: sink, + Ntfy: sink, + Telegram: sink, + Nudges: st, + Reminders: st, + }) + return newTickLoop(st, g, d, phraser.NewStub(), rules, time.Second, 5*time.Minute, 0) +} + +// refNow — fixed tick time so presence decay + since durations are deterministic. +func refNow() time.Time { return time.Date(2026, 6, 30, 12, 0, 0, 0, time.UTC) } + +// markPresent seeds desk_active + page_heartbeat with fresh ts so presence +// resolves to Present for the given tick time (cold-start is away; ENTER at +// 0.55 — a fresh desk_active alone gives 0.90, well over). +func markPresent(t *testing.T, st *store.Store, ctx context.Context, now time.Time) { + t.Helper() + for _, key := range []string{"desk_active", "page_heartbeat"} { + if _, err := st.SetValue(ctx, store.KindSelf, key, "tap:desk", map[string]bool{key: true}, now); err != nil { + t.Fatalf("seed %s: %v", key, err) + } + } +} + +func TestTickColdStoreSendsNothing(t *testing.T) { + // The "shuts up when uncertain" floor: no facts ⇒ every rule's + // InertWhenNoData keys are missing ⇒ gate skips them. the loop's silence + // is the default outcome of a tick on an empty store. + st := newTestStore(t) + ctx := context.Background() + sink := &fakeSink{} + tl := newTestTickLoop(t, st, sink) + + tl.tick(ctx, refNow()) + + if len(sink.sends) != 0 { + t.Fatalf("cold-store tick sent %d; want 0 (shuts up when no data)", len(sink.sends)) + } +} + +func TestTickWaterFiresWhenDueAndPresent(t *testing.T) { + // water fact 4h ago ⇒ since(water)=4h ≥ 3h ⇒ predicate true. presence + // present ⇒ sev1 care gate holds (no quiet/cal/cooldown). routing for + // sev1 present is [voice] — one send captured. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + } + sink := &fakeSink{} + tl := newTestTickLoop(t, st, sink) + + tl.tick(ctx, now) + + if len(sink.sends) != 1 { + t.Fatalf("tick sends = %d, want 1 (water, voice only)", len(sink.sends)) + } + if got, want := sink.sends[0].RuleName, "water"; got != want { + t.Errorf("send rule = %q, want %q", got, want) + } + if got, want := sink.sends[0].Channel, delivery.ChannelVoice; got != want { + t.Errorf("send channel = %v, want voice (sev1 present)", got) + } + if sink.sends[0].Body == "" { + t.Error("phraser Stub produced an empty body for the water nudge") + } + + // one nudge row recorded with channel=voice — verify the dispatch path + // wrote through to the store (the feedback loop's only input). RecentOutcomes + // filters to resolved rows, so confirm the recorded nudge exists via LastNudge. + n, err := st.LastNudge(ctx, "water") + if err != nil { + t.Fatalf("LastNudge: %v", err) + } + if n.Channel != string(delivery.ChannelVoice) { + t.Errorf("recorded nudge channel = %q, want %q", n.Channel, delivery.ChannelVoice) + } +} + +func TestTickCooldownSuppressesSecondSend(t *testing.T) { + // After a water nudge, the gate's cooldown (DefaultRules sets water + // base cooldown = 30m) suppresses the same rule on the next tick. + st := newTestStore(t) + ctx := context.Background() + now := refNow() + markPresent(t, st, ctx, now) + if _, err := st.SetValue(ctx, store.KindSelf, "water", "tap:water", map[string]int{"ml": 0}, now.Add(-4*time.Hour)); err != nil { + t.Fatalf("seed water: %v", err) + _ = err + } + sink := &fakeSink{} + tl := newTestTickLoop(t, st, sink) + + tl.tick(ctx, now) // fires + tl.tick(ctx, now.Add(time.Minute)) // still within 30m cooldown ⇒ suppressed + + if len(sink.sends) != 1 { + t.Fatalf("sends after second tick = %d, want 1 (cooldown should suppress)", len(sink.sends)) + } +} + +func TestTickReminderFiresOnceAndMarkedFired(t *testing.T) { + // A due reminder bypasses the gate. away (no presence probes ⇒ cold + // start away) routes the reminder to [ntfy]. the dispatcher marks the + // reminder fired only after at least one channel succeeded; verify by + // re-ticking and confirming it isn't re-dispatched (DueReminders returns + // only `status == pending AND fire_ts <= now`). + st := newTestStore(t) + ctx := context.Background() + now := refNow() + if _, err := st.CreateReminder(ctx, now.Add(-time.Minute), `{"text":"stand up"}`); err != nil { + t.Fatalf("CreateReminder: %v", err) + } + sink := &fakeSink{} + tl := newTestTickLoop(t, st, sink) + + tl.tick(ctx, now) + if got, want := len(sink.sends), 1; got != want { + t.Fatalf("reminder tick sends = %d, want 1 (away → ntfy)", got) + } + if sink.sends[0].Channel != delivery.ChannelNtfy { + t.Errorf("reminder channel = %v, want ntfy (away)", sink.sends[0].Channel) + } + if sink.sends[0].Body != "stand up" { + t.Errorf("reminder body = %q, want %q (router payload text)", sink.sends[0].Body, "stand up") + } + if sink.sends[0].Kind != delivery.KindReminder { + t.Errorf("reminder kind = %v, want %v", sink.sends[0].Kind, delivery.KindReminder) + } + + // re-tick: the reminder is no longer pending (marked fired) ⇒ not in + // DueReminders ⇒ the reminder path is silent. + sink.sends = nil + tl.tick(ctx, now.Add(time.Minute)) + if len(sink.sends) != 0 { + t.Fatalf("second reminder tick sends = %d, want 0 (fired once)", len(sink.sends)) + } +} + +// TestTuneWritesFeedbackCooldownToStore — the daemon's impure tune() step end +// to end: seed a rule with ≥ TuneMinOutcomes resolved `ignored` outcomes, +// call tune(), assert it wrote a `cooldown:` (source=feedback) fact that +// the gatherer then reads back as the active cooldown base. This is the closed +// feedback loop: outcomes → tune → write → gather → gate sees the tuned base. +func TestTuneWritesFeedbackCooldownToStore(t *testing.T) { + st := newTestStore(t) + ctx := context.Background() + now := refNow() + + sink := &fakeSink{} + tl := newTestTickLoop(t, st, sink) + + // seed enough resolved `ignored` nudge outcomes to trip the tuner (above + // TuneMinOutcomes). all-ignored ⇒ factor 1.5 ⇒ base × 1.5; clamped to Max. + r := loop.WaterRule() + for i := 0; i < loop.TuneSampleN; i++ { + id, err := st.RecordNudge(ctx, r.Name, "voice", "drink water", now.Add(-time.Hour)) + if err != nil { + t.Fatalf("RecordNudge %d: %v", i, err) + } + if err := st.ResolveNudge(ctx, id, store.NudgeIgnored, now); err != nil { + t.Fatalf("ResolveNudge %d: %v", i, err) + } + } + + tl.tune(ctx) + + // the gatherer should now use the tuned base (clamped to WaterRule.Max = + // 6h) as the cooldown base for `water`. verify via the cooldown-until + // field by seeding a water nudge + asserting cooldown = sendTs + Max. + // (we read the persisted tuned base directly rather than going via gather + // so the assertion isolates tune()'s write from the gatherer path.) + fb, err := st.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource) + if err != nil { + t.Fatalf("LatestFactBySource: %v (no feedback fact written?)", err) + } + tuned, ok := loop.ParseCooldownFact(fb) + if !ok { + t.Fatalf("ParseCooldownFact: not ok (value %q)", fb.Value) + } + if tuned != 45*time.Minute { + t.Fatalf("all-ignored: base 30m × 1.5 = 45m (no — under WaterRule.Max 6h so unclamped): want 45m, got %v", tuned) + } + + // idempotent: a second tune() with the same outcomes writes nothing new + // (RecentOutcomes is steady between resolves; the persisted value equals + // the computed one ⇒ skip). + fb1 := fb + tl.tune(ctx) + fb2, err := st.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource) + if err != nil { + t.Fatalf("LatestFactBySource second call: %v", err) + } + if fb2.ID != fb1.ID { + t.Fatalf("tune() re-wrote identical value: fact id %d → %d (should skip when unchanged)", + fb1.ID, fb2.ID) + } + + // flip the outcomes pattern to `acted`: next tune() writes a new value, + // shorter than Max. RecentOutcomes is sorted DESC ts,id, so re-seeding + // newer-acted nudges makes them dominate the older-ignored set. + for i := 0; i < loop.TuneSampleN; i++ { + id, err := st.RecordNudge(ctx, r.Name, "voice", "drink water", now.Add(time.Minute+time.Duration(i)*time.Second)) + if err != nil { + t.Fatalf("RecordNudge acted %d: %v", i, err) + } + if err := st.ResolveNudge(ctx, id, store.NudgeActed, now); err != nil { + t.Fatalf("ResolveNudge acted %d: %v", i, err) + } + } + tl.tune(ctx) + fb3, err := st.LatestFactBySource(ctx, loop.FeedbackKey(r), loop.FeedbackSource) + if err != nil { + t.Fatalf("LatestFactBySource post-flip: %v", err) + } + tuned3, ok := loop.ParseCooldownFact(fb3) + if !ok { + t.Fatalf("ParseCooldownFact post-flip: not ok (value %q)", fb3.Value) + } + if tuned3 >= r.Cooldown.Max { + t.Fatalf("acted-dominated outcomes should shrink cooldown below Max: got %v (Max %v)", + tuned3, r.Cooldown.Max) + } + if tuned3 < r.Cooldown.Min { + t.Fatalf("tuned cooldown below Min envelope: got %v (Min %v) — clamp broken", + tuned3, r.Cooldown.Min) + } + + // the gatherer actually reads it back: assert CooldownUntil for water is + // derived from the tuned base (not the rule's static Base) when a real + // nudge exists. seed a water nudge now and gather. + nudgeTs := now.Add(2 * time.Minute) + if _, err := st.RecordNudge(ctx, r.Name, "voice", "drink water", nudgeTs); err != nil { + t.Fatalf("final RecordNudge: %v", err) + } + g := loop.NewGatherer(st, loop.DefaultRules()) + snap, _, err := g.GatherState(ctx, now.Add(3*time.Minute)) + if err != nil { + t.Fatalf("GatherState: %v", err) + } + wantUntil := nudgeTs.Add(tuned3) + if got := snap.CooldownUntil[r.Name]; got != wantUntil { + t.Fatalf("gatherer used tuned base: CooldownUntil want %v, got %v", + wantUntil, got) + } +} \ No newline at end of file diff --git a/cmd/mavend/voice.go b/cmd/mavend/voice.go new file mode 100644 index 0000000..e5f819b --- /dev/null +++ b/cmd/mavend/voice.go @@ -0,0 +1,853 @@ +// Package main is mavend's voice wiring + reactive handler. +// +// Two responsibilities for the audio path: +// +// 1. CONSTRUCTION: read cfg.Voice, build the stt/tts/transcribers (Stub +// in-process by default, Remote via worker socket when configured), +// the router (stage-0 grammar + HashEmbedder classifier seeded with +// floor examples — production swaps in the ONNX multilingual model +// later), the voice TCP listener, the sessions registry, the +// voicesink, and wire the voicesink into the dispatcher's Voice slot. +// +// 2. HANDLER: a concrete voice.Handler that processes PushToTalk +// requests: stt → router → action → replier → tts → reply. The +// handler is what makes the audio round-trip "live". It wires to the +// CoreAPI in-process (the daemon already has it as ipc.NewStoreAPI(st) +// for module-IPC — the reactive path uses the same CoreAPI off the +// same store; both are the "core = the only key-holder" path through +// the daemon-embedded adapter). +// +// The "actions" handled today (per spec order; some deferred): +// +// - IntentFact: WriteFact via CoreAPI. The router's Slots.Key/Value feed +// the write; Source = "tap:voice" (the voice path is a tap, value=1.0 +// confidence — the user said it out loud, maven trusts the capture). +// - IntentReminder: CreateReminder via CoreAPI. The router already +// resolved relative→absolute at capture ("in 4h" → fire_ts); the +// CoreAPI stores it as-is. +// - IntentAct: the tool executor runs the matched fn against the store's +// ENABLED allowlist (internal/tool). A verb not on it is scaffolded as a +// 'proposed' tool a human enables on the authed mavweb surface (never +// voice). Destructive tools run only after a spoken confirm turn. +// - IntentNote: chroma/vector-store deferred. The handler replies +// "saved" without persisting — a stub on the way to chroma. +// - IntentQuery: RAG-over-chroma deferred. The handler replies "I'll +// look that up later" — same shape as the other deferred slots. +// - Clarify: the router's stage-3 confidence gate fired; reply "didn't +// catch that, can you rephrase?" +// +// The Replier (voice.StubReplier today) renders the reply TEXT across all +// these branches. The TTS synthesiser (tts.Stub today) renders that text +// to audio. The PushToTalkResp carries BOTH so the client can play (audio) +// AND log (text) for tests asserting the round-trip. +package main + +import ( + "bufio" + "context" + "errors" + "fmt" + "log" + "os" + "path/filepath" + "strings" + "sync" + "time" + + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/config" + "github.com/kami/maven/internal/delivery" + "github.com/kami/maven/internal/delivery/voicesink" + "github.com/kami/maven/internal/ipc" + "github.com/kami/maven/internal/router" + "github.com/kami/maven/internal/stt" + "github.com/kami/maven/internal/tool" + "github.com/kami/maven/internal/tts" + "github.com/kami/maven/internal/voice" + "github.com/kami/maven/internal/worker" +) + +// voiceWiring — everything the daemon needs to run the audio path. Held by +// cmd/mavend/main.go alongside the other wirings; closed on shutdown. +type voiceWiring struct { + server *voice.Server + sessions *voice.Sessions + voiceSink delivery.Sink + embedder router.Embedder + // worker clients (set when configured as Remote): closed on shutdown so + // mavsttd / mavttsd don't keep a stale conn into a restarting daemon. + sttClient *worker.Client + ttsClient *worker.Client +} + +// close releases the listener + worker conns. Safe to call on nil (when +// voice is not wired — wireVoice returns nil,nil). +func (w *voiceWiring) close() { + if w == nil { + return + } + if w.embedder != nil { + _ = w.embedder.Close() + } + if w.server != nil { + _ = w.server.Close() + } + if w.sttClient != nil { + _ = w.sttClient.Close() + } + if w.ttsClient != nil { + _ = w.ttsClient.Close() + } +} + +// wireVoice builds the audio path from cfg + a CoreAPI + a router. Returns +// nil wiring + nil error when voice isn't enabled (the caller's voice sink +// stays nil; the dispatcher's ChannelVoice routing drops silently). +// +// When voice is enabled, MUST wire a voicesink into the dispatcher's Voice +// slot using w.sessions (the caller does that — see main.go). +func wireVoice(cfg *config.Config, coreAPI ipc.CoreAPI) (*voiceWiring, error) { + if cfg.Voice == nil || !cfg.Voice.Enabled { + return nil, nil + } + w := &voiceWiring{} + + // ----- stt (Stub in-process OR Remote via worker socket) ----- + var transcriber stt.Transcriber + if cfg.Voice.Stt != nil && cfg.Voice.Stt.Socket != "" { + c := worker.Dial(cfg.Voice.Stt.Socket) + w.sttClient = c + lang := cfg.Voice.Stt.Lang + if lang == "" { + lang = cfg.Voice.Lang + } + transcriber = stt.NewRemote(c, lang) + } else { + transcriber = stt.NewStub() + } + + // ----- tts (Stub in-process OR Remote) ----- + var synthesizer tts.Synthesizer + if cfg.Voice.Tts != nil && cfg.Voice.Tts.Socket != "" { + c := worker.Dial(cfg.Voice.Tts.Socket) + w.ttsClient = c + lang := cfg.Voice.Tts.Lang + if lang == "" { + lang = cfg.Voice.Lang + } + synthesizer = tts.NewRemote(c, lang, cfg.Voice.Tts.Voice) + } else { + synthesizer = tts.NewStub() + } + + // ----- router: embedder (ONNX when configured, floor HashEmbedder otherwise) ----- + var emb router.Embedder + if cfg.Voice.Embedder != nil { + onnx, err := router.NewONNXEmbedder( + cfg.Voice.Embedder.ModelPath, + cfg.Voice.Embedder.TokenizerPath, + cfg.Voice.Embedder.LibPath, + ) + if err != nil { + w.close() + return nil, fmt.Errorf("embedder: %w", err) + } + log.Printf("voice: onnx embedder loaded (%d dim)", onnx.Dim()) + emb = onnx + } else { + emb = router.NewHashEmbedder(1024) + } + w.embedder = emb + + // ----- tool executor (the enabled act allowlist, store-backed) ----- + // Config tools are the declarative bootstrap: seed them into the store as + // enabled (editing mavend.json IS the human enable act). Ad-hoc tools are + // enabled later through the authed mavweb surface. The executor + matcher + // both read the store live, so a newly-enabled tool is runnable without a + // daemon restart. + seedTools(coreAPI, cfg.Voice.Tools) + exec := tool.NewExecutor(coreAPI, time.Duration(cfg.Voice.ToolTimeout)) + matcher := tool.NewMatcher(coreAPI) + + // ----- router (the cascade; floor examples seed the classifier) ----- + // The act matcher's allowlist is exactly the enabled tool names — the + // router only matches acts the executor can run (one source of truth). + threshold := cfg.Voice.RouterThreshold + if threshold <= 0 { + threshold = config.DefaultRouterThreshold + } + rtr := buildRouter(emb, matcher, threshold) + + // ----- sessions registry (shared with voicesink) ----- + sessions := voice.NewSessions() + w.sessions = sessions + + // ----- voice sink (proactive nudges: dispatcher → voicesink → tts → push to client) ----- + w.voiceSink = voicesink.New(synthesizer, sessions) + + // ----- the handler (the reactive path; closes over stt / tts / router / coreAPI) ----- + h := &reactiveHandler{ + stt: transcriber, + tts: synthesizer, + router: rtr, + embedder: emb, + api: coreAPI, + tools: exec, + replier: voice.NewStubReplier(), + now: time.Now, + } + + // ----- the server (TCP listener) ----- + srv := voice.NewServer(cfg.Voice.Bind, h, sessions) + if err := srv.Listen(); err != nil { + w.close() + return nil, fmt.Errorf("voice listen: %w", err) + } + w.server = srv + + return w, nil +} + +// reactiveHandler — voice.Handler implementation. One method: turn a +// PushToTalkReq into a reply (audio + text). The handler is concurrency- +// safe (the wired stt/tts/router/api all are); called from per-conn +// goroutines on the voice.Server. +type reactiveHandler struct { + stt stt.Transcriber + tts tts.Synthesizer + router *router.Router + embedder router.Embedder // reused for note write/query (same model as the classifier) + api ipc.CoreAPI + tools *tool.Executor + replier voice.Replier + now func() time.Time + + // pending destructive-act confirmation. A destructive act replies with a + // "выполнить X? да/нет" prompt and parks here; the NEXT utterance is read as + // the y/n answer. ponytail: single slot, single-user box — a second act + // while one waits overwrites it (last-asked wins); expires after confirmTTL. + mu sync.Mutex + pending *pendingAct +} + +// pendingAct — a destructive act awaiting a spoken confirm. +type pendingAct struct { + fn string + args []string + phrase string + expiry time.Time +} + +// confirmTTL — how long a parked destructive confirm stays answerable. Short: +// a confirm is a same-breath gesture; a stale prompt shouldn't fire on an +// unrelated later "да". +const confirmTTL = 90 * time.Second + +// HandlePushToTalk — the full reactive round-trip. Each step's failure +// surfaces as a short reply text + empty audio OR an error; the voice +// server translates an error into a wire RpcError. Today the handler +// prefers a canned error-reply over an error return (a user-facing "didn't +// catch that" is better than a wire error the client surfaces as +// "internal"); the only error returned is a synthesizer fault (no audio +// to ship back). +func (h *reactiveHandler) HandlePushToTalk(ctx context.Context, req voice.PushToTalkReq, _ uint64) (voice.PushToTalkResp, error) { + // 1. stt — transcribe the audio. + text, _, err := h.stt.Transcribe(ctx, req.Audio) + if err != nil { + log.Printf("voice: stt error: %v", err) + return h.reply(ctx, "не получилось разобрать речь — попробуй ещё раз.", nil) + } + if text == "" { + return h.reply(ctx, "ничего не услышала — попробуй ещё раз.", nil) + } + log.Printf("voice: stt → %q", text) + + // 1b. confirm turn — if a destructive act is parked, this utterance is its + // y/n answer, not a fresh command. Handled before routing so "да" doesn't + // get classified as some other intent. + if reply, handled := h.resolveConfirm(ctx, text); handled { + return h.reply(ctx, reply, nil) + } + + // 2. router — classify the utterance. + dec, err := h.router.Route(ctx, text, h.now()) + if err != nil { + // ErrNoIntents ⇒ classifier unseeded (cold boot). reply with a + // "still warming up" rather than a wire error. + if errors.Is(err, router.ErrNoIntents) { + return h.reply(ctx, "я ещё не понимаю свободную речь — скоро научусь.", nil) + } + log.Printf("voice: router error: %v", err) + return h.reply(ctx, "не получилось разобрать команду.", nil) + } + + // 3. action — execute the decision's intent. errors here surface as + // short reply text (the user wants to know the action didn't land); + // the round-trip stays alive. + replyText := h.applyAction(ctx, dec) + + // 4. replier — phrase the reply across the router decision. + if replyText == "" { + replyText = h.replier.Reply(dec) + } + + // 5. tts — synthesise the reply text; return to the voice server which + // ships it back on the conn. + return h.reply(ctx, replyText, nil) +} + +// applyAction — executes the router's Decision. Intent-by-intent: +// +// - IntentFact: WriteFact via CoreAPI. Source = "tap:voice" (a voice +// capture is a tap; confidence 1.0). +// - IntentReminder: CreateReminder via CoreAPI. +// - IntentAct: tool-executor deferred (no-op today; the reply says so). +// - IntentNote / IntentQuery: chroma/RAG deferred (no-op; reply says so). +// - Clarify: the router's stage-3 fired; no action. +// +// Returns "" when the Replier should phrase the reply (the default path); +// returns a non-empty string when the action path wants to OVERRIDE the +// reply text (e.g. an action error the user should hear SPECIFICALLY, not +// a generic "ok"). Errors surface as a short reply text the user hears. +// queryMinScore — the note-recall confidence gate. Top cosine below this ⇒ +// "no note" instead of a guess. Hand-tuned for the ONNX embedder; a knob, not +// load-bearing math (same posture as the presence thresholds). +const queryMinScore = 0.55 + +func (h *reactiveHandler) applyAction(ctx context.Context, dec router.Decision) string { + if dec.Clarify { + return "" // the Replier phrases clarify + } + switch dec.Intent { + case router.IntentFact: + if !dec.Slots.HasKey { + return "не разобрала, что записать — попробуй иначе." + } + now := h.now() + req := ipc.WriteFactReq{ + Ts: now, + Kind: "self", + Key: dec.Slots.Key, + Value: dec.Slots.Value, + Source: "tap:voice", + Confidence: 1.0, + } + if _, err := h.api.WriteFact(ctx, req); err != nil { + log.Printf("voice: write fact: %v", err) + return "не получилось сохранить факт." + } + return "" // replier phrases the success reply + + case router.IntentReminder: + if !dec.Slots.HasTime { + return "не получилось разобрать время напоминания." + } + payload := `{"text":` + jsonString(dec.Utterance) + `}` + if _, err := h.api.CreateReminder(ctx, dec.Slots.Time, payload); err != nil { + log.Printf("voice: create reminder: %v", err) + return "не получилось поставить напоминание." + } + return "" + + case router.IntentAct: + // tool executor: run the matched fn against the enabled allowlist. + // HasFn=false ⇒ no allowlist match: scaffold a 'proposed' tool the user + // can enable on the authed surface ("earn the right to ask"). + if !dec.Slots.HasFn { + return h.proposeGap(ctx, dec) + } + out, err := h.tools.Exec(ctx, dec.Slots.Fn, dec.Slots.Args, false) + if err != nil { + switch { + case errors.Is(err, tool.ErrNeedsConfirm): + // destructive: park it and ask. The next utterance answers. + phrase := actPhrase(dec.Slots.Fn, dec.Slots.Args) + h.park(dec.Slots.Fn, dec.Slots.Args, phrase) + return "выполнить «" + phrase + "»? скажи «да» или «нет»." + case errors.Is(err, tool.ErrNotEnabled): + return h.proposeGap(ctx, dec) + } + log.Printf("voice: tool %s: %v", dec.Slots.Fn, err) + if out != "" { + return "не получилось выполнить команду: " + firstLine(out) + } + return "не получилось выполнить команду." + } + if out != "" { + return "готово: " + firstLine(out) + } + return "готово." + + case router.IntentSystem: + // Quiet-hours toggle — "quiet on" / "тихий режим" — writes + // a config fact the gate reads. Check before the query-only path. + if reply := h.handleQuietToggle(ctx, dec); reply != "" { + return reply + } + // System-status queries return to the Replier for phrasing. + // The handler emits the current answer inline (no DB / RAG needed). + return h.replySystem(ctx, dec) + + case router.IntentNote: + // embed the note text with the same model the classifier uses, persist + // via CoreAPI (source=tap:voice). Semantic recall lives in `notes`, not + // facts — no predicate reads it (spec's two-memory split). + vec, err := h.embedder.Embed(ctx, dec.Utterance) + if err != nil { + log.Printf("voice: embed note: %v", err) + return "не получилось сохранить заметку." + } + if _, err := h.api.WriteNote(ctx, h.now(), dec.Utterance, vec, "tap:voice"); err != nil { + log.Printf("voice: write note: %v", err) + return "не получилось сохранить заметку." + } + return "" // replier phrases the "saved" reply + + case router.IntentQuery: + vec, err := h.embedder.Embed(ctx, dec.Utterance) + if err != nil { + log.Printf("voice: embed query: %v", err) + return "не получилось найти ответ." + } + notes, err := h.api.QueryNotes(ctx, vec, 5) + if err != nil { + log.Printf("voice: query notes: %v", err) + return "не получилось найти ответ." + } + // Confidence gate: below threshold, say "I don't know" rather than read + // back the least-unrelated note — a confident wrong recall is worse than + // a gap (spec's "not a guesser-of-truth"). Same instinct as the loop's + // since(key)==null → don't fire. Tuned for the ONNX embedder; the Hash + // floor scores lexically and may rarely clear it. + if len(notes) == 0 || notes[0].Score < queryMinScore { + return "у меня нет заметок по этому вопросу." + } + // Full RAG (phraser-composed) is deferred — this is the browse surface. + // Return a summary of the best match(es) so the user gets context, not just + // one verbatim snippet. The phraser seam in the replier will natural-language + // the results when the LLM-backed Replier swaps in. + if len(notes) == 1 { + return "ты записал: " + notes[0].Text + } + var b strings.Builder + b.WriteString("вот что нашла: ") + for i, n := range notes { + if i > 0 { + b.WriteString("; ") + } + b.WriteString(n.Text) + } + return b.String() + } + return "" +} + +var ruWeekdays = []string{ + "воскресенье", "понедельник", "вторник", "среда", + "четверг", "пятница", "суббота", +} + +var ruMonths = []string{ + "января", "февраля", "марта", "апреля", "мая", "июня", + "июля", "августа", "сентября", "октября", "ноября", "декабря", +} + +func ruPlural(n int, one, two, many string) string { + n = n % 100 + if n > 10 && n < 20 { + return many + } + n = n % 10 + switch n { + case 1: + return one + case 2, 3, 4: + return two + default: + return many + } +} + +// handleQuietToggle — checks if the utterance toggles quiet hours. +// Writes a `quiet_hours` config fact (value "true"/"false") so the loop +// gate reads it next tick. Returns a reply text, or "" if no match. +func (h *reactiveHandler) handleQuietToggle(ctx context.Context, dec router.Decision) string { + u := strings.ToLower(dec.Utterance) + // Match: "quiet on" / "quiet off" / "тихий режим" / "не беспокоить" etc. + var on, off bool + for _, kw := range []string{"quiet on", "quiet mode", "тихий режим", "не беспокоить", "не шуми"} { + if strings.Contains(u, kw) { + on = true + break + } + } + for _, kw := range []string{"quiet off", "quiet end", "выключи тихий", "отключи тихий", "шумный режим"} { + if strings.Contains(u, kw) { + off = true + break + } + } + if !on && !off { + return "" + } + now := h.now() + val := "false" + reply := "тихий режим выключен." + if on { + val = "true" + reply = "тихий режим включён. буду реже напоминать." + } + if _, err := h.api.WriteFact(ctx, ipc.WriteFactReq{ + Ts: now, + Kind: "config", + Key: "quiet_hours", + Value: val, + Source: "tap:voice", + Confidence: 1.0, + }); err != nil { + log.Printf("voice: write quiet_hours: %v", err) + return "не получилось переключить тихий режим." + } + return reply +} + +// replySystem answers system-observable queries using the handler's clock +// and (in future) system interfaces. The decision's utterance is parsed +// for keywords to determine what the user is asking about. +func (h *reactiveHandler) replySystem(ctx context.Context, dec router.Decision) string { + u := strings.ToLower(dec.Utterance) + now := h.now() + + switch { + case strings.Contains(u, "час") || strings.Contains(u, "врем"): + h := now.Hour() + m := now.Minute() + hourWord := ruPlural(h, "час", "часа", "часов") + if m == 0 { + return fmt.Sprintf("сейчас %d %s ровно", h, hourWord) + } + minWord := ruPlural(m, "минута", "минуты", "минут") + return fmt.Sprintf("сейчас %d %s %d %s", h, hourWord, m, minWord) + case strings.Contains(u, "день") || strings.Contains(u, "числ"): + dow := ruWeekdays[now.Weekday()] + month := ruMonths[now.Month()-1] + return fmt.Sprintf("сегодня %s, %d %s %d года", dow, now.Day(), month, now.Year()) + case strings.Contains(u, "погод") || strings.Contains(u, "градус") || strings.Contains(u, "дожд") || strings.Contains(u, "холод") || strings.Contains(u, "тепл"): + return "погода пока не подключена — нужен внешний сервис." + case strings.Contains(u, "кто дома") || strings.Contains(u, "человек дома"): + return "присутствие пока не подключено к голосовому запросу." + case strings.Contains(u, "памят") || strings.Contains(u, "процессор") || strings.Contains(u, "загрузк") || strings.Contains(u, "статус") || strings.Contains(u, "работа") || strings.Contains(u, "сервис") || strings.Contains(u, "диск") || strings.Contains(u, "ip") || strings.Contains(u, "аптайм") || strings.Contains(u, "трафик") || strings.Contains(u, "интернет"): + return "системная статистика пока не подключена." + default: + return "пока не умею отвечать на этот вопрос." + } +} + +// reply wraps a text reply through TTS to produce a PushToTalkResp. If TTS +// fails, the response carries an empty audio + the text — the client can +// still display text if it can't play. The routedChannels field is +// reserved for a future "the dispatcher also forwarded to ntfy/telegram" +// reply (today the reactive path doesn't dispatch nudges; that's the loop +// tick's job). +func (h *reactiveHandler) reply(ctx context.Context, text string, _ []string) (voice.PushToTalkResp, error) { + log.Printf("voice: reply → %q", text) + audioOut, err := h.tts.Synthesize(ctx, text) + if err != nil { + log.Printf("voice: tts error: %v", err) + return voice.PushToTalkResp{ReplyText: text, ReplyAudio: audio.Audio{}}, nil + } + return voice.PushToTalkResp{ReplyText: text, ReplyAudio: audioOut}, nil +} + +// buildRouter constructs the reactive-path router with the given embedder +// and confidence threshold. +// - stage-0 grammars from DefaultActMatcher whose fn allowlist is exactly +// the enabled tool names (actFns) — the router only matches acts the +// executor can run. Empty ⇒ every act refuses at the matcher. +// - The embedder is provided by wireVoice: HashEmbedder (floor) when no +// embedder config is present, or the ONNX multilingual model when +// configured — same interface, one constructor change. +// - 6 bootstrap examples covering the 5 intents + one compound-capture +// placeholder. Spec calls for ~10 per intent at production; this is the +// bootstrapping floor swapped by tuning the seed set later. +// - Threshold is from voice.router_threshold config (default 0.35). +func buildRouter(emb router.Embedder, acts router.ActMatcher, threshold float64) *router.Router { + cls := router.NewClassifier(emb) + seedClassifier(cls) + return router.New(router.Config{ + Grammars: router.DefaultGrammars(acts), + Classifier: cls, + Extractor: router.Extractor{ + Time: router.StubDateTimeParser{}, + Acts: acts, + Facts: router.DefaultFactParser{}, + }, + Threshold: threshold, + }) +} + +// seedDir is the directory containing intent seed files. Each file is named +// .txt and contains one training example per line (blank lines and +// lines starting with # are ignored). Relative to the working directory. +const seedDir = "models/seeds" + +// seedClassifier floors the embedded examples so the cold-boot path +// doesn't return ErrNoIntents. Loads examples from seedDir — one file per +// intent (act.txt, reminder.txt, fact.txt, note.txt, query.txt). When the +// classifier can't decide it falls through to Clarify — the last-resort +// path asks the user to rephrase rather than guessing wrong. +func seedClassifier(c *router.Classifier) { + intents := []router.Intent{ + router.IntentAct, + router.IntentReminder, + router.IntentFact, + router.IntentNote, + router.IntentQuery, + router.IntentSystem, + } + total := 0 + for _, intent := range intents { + n, err := loadSeedFile(c, intent) + if err != nil { + log.Printf("voice: seed %s: %v", intent, err) + continue + } + total += n + } + log.Printf("voice: loaded %d seed examples from %s", total, seedDir) +} + +func loadSeedFile(c *router.Classifier, intent router.Intent) (int, error) { + path := filepath.Join(seedDir, string(intent)+".txt") + f, err := os.Open(path) + if err != nil { + return 0, fmt.Errorf("open %s: %w", path, err) + } + defer f.Close() + + var count int + sc := bufio.NewScanner(f) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + if err := c.AddExample(context.Background(), intent, line); err != nil { + log.Printf("voice: seed %s: skipping %q: %v", intent, line, err) + continue + } + count++ + } + if err := sc.Err(); err != nil { + return count, fmt.Errorf("scan %s: %w", path, err) + } + return count, nil +} + +// jsonString — a one-line JSON string encoder without dragging encoding/json +// into the top of this file. Used to wrap a reminder payload's text field; +// the router's reminder Slots are already absolute (DateTimeParser resolved +// relative→absolute), the payload shape is conventional {"text":...}. +func jsonString(s string) string { + return jsonStringImpl(s) +} + +// park stores a destructive act awaiting confirmation. Overwrites any prior +// pending (last-asked wins — single-user box). +func (h *reactiveHandler) park(fn string, args []string, phrase string) { + h.mu.Lock() + h.pending = &pendingAct{fn: fn, args: args, phrase: phrase, expiry: h.now().Add(confirmTTL)} + h.mu.Unlock() +} + +// resolveConfirm interprets an utterance as the answer to a parked destructive +// act. Returns (reply, true) when it consumed the utterance as a y/n answer; +// (\"\", false) when there's nothing pending (or the parked act expired), so the +// caller routes the utterance normally. An unrecognised answer cancels the +// pending act and routes normally — a confirm that can't be answered clearly is +// safer abandoned than left armed. +func (h *reactiveHandler) resolveConfirm(ctx context.Context, text string) (string, bool) { + h.mu.Lock() + p := h.pending + if p == nil { + h.mu.Unlock() + return "", false + } + if h.now().After(p.expiry) { + h.pending = nil + h.mu.Unlock() + return "", false + } + switch classifyConfirm(text) { + case confirmYes: + h.pending = nil + h.mu.Unlock() + out, err := h.tools.Exec(ctx, p.fn, p.args, true) // confirmed + if err != nil { + log.Printf("voice: tool %s (confirmed): %v", p.fn, err) + if out != "" { + return "не получилось выполнить команду: " + firstLine(out), true + } + return "не получилось выполнить команду.", true + } + if out != "" { + return "готово: " + firstLine(out), true + } + return "готово.", true + case confirmNo: + h.pending = nil + h.mu.Unlock() + return "отменила.", true + default: + // unclear answer: abandon the confirm, route this utterance normally. + h.pending = nil + h.mu.Unlock() + return "", false + } +} + +// proposeGap scaffolds a 'proposed' tool for an act whose verb isn't enabled. +// maven drafts the registration (name = the verb, provenance = the utterance); +// a human enables it on the authed surface. She suggests, never enables. +func (h *reactiveHandler) proposeGap(ctx context.Context, dec router.Decision) string { + name := firstWord(stripWake(dec.Utterance)) + if name == "" { + return "не разобрала команду — попробуй иначе." + } + newly, err := h.api.ProposeTool(ctx, name, dec.Utterance, h.now()) + if err != nil { + log.Printf("voice: propose tool %q: %v", name, err) + return "команды «" + name + "» нет в списке разрешённых." + } + if newly { + return "команды «" + name + "» нет в списке. Предложила её добавить — включи через клиент." + } + return "команды «" + name + "» пока нет в списке — она уже предложена, включи через клиент." +} + +// confirmVerdict — the parse of a y/n confirm answer. +type confirmVerdict int + +const ( + confirmUnknown confirmVerdict = iota + confirmYes + confirmNo +) + +// classifyConfirm reads a short ru/en yes-or-no answer. Substring match on the +// stems so inflections/fillers ("да, давай", "нет, отмени") still land. +func classifyConfirm(text string) confirmVerdict { + t := strings.ToLower(strings.TrimSpace(text)) + // negatives first — "не надо" contains no "да", but check no-stems before + // yes so a leading "нет" isn't shadowed. + for _, no := range []string{"нет", "не надо", "отмен", "стоп", "no", "cancel", "stop", "don't"} { + if strings.Contains(t, no) { + return confirmNo + } + } + for _, yes := range []string{"да", "ага", "давай", "подтвер", "конечно", "yes", "yeah", "yep", "confirm", "ок", "okay", "ok"} { + if strings.Contains(t, yes) { + return confirmYes + } + } + return confirmUnknown +} + +// actPhrase renders "fn arg1 arg2" for the confirm prompt. +func actPhrase(fn string, args []string) string { + if len(args) == 0 { + return fn + } + return fn + " " + strings.Join(args, " ") +} + +// stripWake removes a leading "maven," wake token so the verb is the first word. +func stripWake(u string) string { + u = strings.TrimSpace(u) + low := strings.ToLower(u) + if strings.HasPrefix(low, "maven") { + u = strings.TrimSpace(u[len("maven"):]) + u = strings.TrimLeft(u, ",:; ") + } + return u +} + +// firstWord returns the first whitespace-delimited token (lowercased) — the +// proposed tool's name. +func firstWord(s string) string { + f := strings.Fields(s) + if len(f) == 0 { + return "" + } + return strings.ToLower(f[0]) +} + +// seedTools upserts the config-declared tools into the store as enabled. Editing +// mavend.json is a human act, so a config tool is enabled by definition; this +// makes the declarative config the reproducible bootstrap while the store stays +// the single runtime source of truth (mavweb enables ad-hoc ones on top). +func seedTools(api ipc.CoreAPI, tools []config.ToolConfig) { + ctx := context.Background() + now := time.Now() + n := 0 + for _, tc := range tools { + if tc.Name == "" || len(tc.Cmd) == 0 { + log.Printf("voice: skipping malformed tool config %+v", tc) + continue + } + if err := api.EnableTool(ctx, tc.Name, tc.Cmd, tc.Destructive, now); err != nil { + log.Printf("voice: seed tool %q: %v", tc.Name, err) + continue + } + n++ + } + log.Printf("voice: seeded %d act tools from config", n) +} + +// firstLine — the first non-empty line of a tool's output, for a short spoken +// reply (the full output goes to the log, not the TTS). Trimmed to keep the +// utterance sane if a command dumps a wall of text. +func firstLine(s string) string { + for _, line := range strings.Split(s, "\n") { + line = strings.TrimSpace(line) + if line != "" { + if len(line) > 200 { + line = line[:200] + } + return line + } + } + return "" +} + +func jsonStringImpl(s string) string { + // minimal JSON string escape — quotes + backslash + control chars. + // adequate for the reminder payload's text field; not a general JSON + // encoder. The chroma / RAG modules (when they land) use a real json + // encoder for richer payloads. Keep it inline here so the import + // direction stays narrow. + var b []byte + b = append(b, '"') + for _, r := range s { + switch r { + case '"': + b = append(b, '\\', '"') + case '\\': + b = append(b, '\\', '\\') + case '\n': + b = append(b, '\\', 'n') + case '\r': + b = append(b, '\\', 'r') + case '\t': + b = append(b, '\\', 't') + default: + if r < 0x20 { + b = append(b, []byte(fmt.Sprintf("\\u%04x", r))...) + } else { + b = append(b, []byte(string(r))...) + } + } + } + b = append(b, '"') + return string(b) +} \ No newline at end of file diff --git a/cmd/mavpoll/main.go b/cmd/mavpoll/main.go new file mode 100644 index 0000000..6a98e44 --- /dev/null +++ b/cmd/mavpoll/main.go @@ -0,0 +1,341 @@ +// mavpoll — the env poller module. +// +// maven doesn't collect metrics; netdata and uptime-kuma already do, tuned to +// the box. This is a thin adapter: it reads their alarms/status and writes +// `facts (kind=env, source=poll:*)` through core's IPC socket. Key-free, +// restart-free, fail-independent — a crashing poller can't touch the store key +// (it never had it), worst case a stale env fact until the next tick. +// +// Two sources, each its own provenance (the loop's rules trust source): +// - netdata → poll:netdata resource alarms (disk/mem/cert/temp) +// - kuma → poll:uptimekuma service up/down (the source of truth for it) +// +// Netdata needs no auth over the wg-fronted net. Kuma's /metrics needs an API +// key (basic-auth); without -kuma the whole kuma path is skipped (netdata-only +// still lights up an end-to-end nudge). +// +// Append-only discipline: a fact is written only when its value CHANGED vs the +// latest for that key+source. A poller that wrote every 60s would churn the +// facts table for nothing; the store is the audit trail, not a metrics sink. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "io" + "log" + "net/http" + "os" + "os/exec" + "os/signal" + "regexp" + "strconv" + "strings" + "syscall" + "time" + + "github.com/kami/maven/internal/ipc" +) + +func main() { + if err := run(os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, "mavpoll:", err) + os.Exit(1) + } +} + +func run(args []string) error { + fs := flag.NewFlagSet("mavpoll", flag.ContinueOnError) + socket := fs.String("socket", "", "core IPC socket path (required)") + netdataURL := fs.String("netdata", "http://127.0.0.1:19999", "netdata base URL ('' to disable)") + kumaURL := fs.String("kuma", "", "uptime-kuma metrics URL, e.g. http://127.0.0.1:3001/metrics ('' to disable)") + kumaKey := fs.String("kuma-key", "", "uptime-kuma API key (basic-auth username)") + wgIface := fs.String("wg", "", "wireguard interface for the presence signal, e.g. wg0 or 'all' ('' to disable)") + wgCmd := fs.String("wg-cmd", "wg", "wg binary (use e.g. 'sudo wg' if the poller lacks CAP_NET_ADMIN)") + interval := fs.Duration("interval", 60*time.Second, "poll cadence") + timeout := fs.Duration("timeout", 8*time.Second, "per-request HTTP timeout") + if err := fs.Parse(args); err != nil { + return err + } + if *socket == "" { + return fmt.Errorf("-socket is required") + } + if *netdataURL == "" && *kumaURL == "" && *wgIface == "" { + return fmt.Errorf("nothing to poll: set -netdata, -kuma and/or -wg") + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + core, err := ipc.Dial(*socket) + if err != nil { + return err + } + defer core.Close() + + p := &poller{ + core: core, + http: &http.Client{Timeout: *timeout}, + netdataURL: strings.TrimRight(*netdataURL, "/"), + kumaURL: *kumaURL, + kumaKey: *kumaKey, + wgIface: *wgIface, + wgCmd: *wgCmd, + } + + log.Printf("mavpoll: polling every %s (netdata=%q kuma=%q wg=%q)", *interval, *netdataURL, *kumaURL, *wgIface) + p.pollOnce(ctx) // fire immediately; don't idle a full interval on start + t := time.NewTicker(*interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + log.Printf("mavpoll: bye") + return nil + case <-t.C: + p.pollOnce(ctx) + } + } +} + +type poller struct { + core ipc.CoreAPI + http *http.Client + netdataURL string + kumaURL string + kumaKey string + wgIface string + wgCmd string +} + +// pollOnce — one sweep of both sources. A failure in one source logs and does +// NOT abort the other: netdata being down shouldn't blind kuma and vice versa. +func (p *poller) pollOnce(ctx context.Context) { + now := time.Now() + if p.netdataURL != "" { + if err := p.pollNetdata(ctx, now); err != nil { + log.Printf("mavpoll: netdata: %v", err) + } + } + if p.kumaURL != "" { + if err := p.pollKuma(ctx, now); err != nil { + log.Printf("mavpoll: kuma: %v", err) + } + } + if p.wgIface != "" { + if err := p.pollWg(ctx); err != nil { + log.Printf("mavpoll: wg: %v", err) + } + } +} + +// ---- wireguard: latest handshake → presence signal ------------------------- + +// pollWg reads `wg show latest-handshakes` and writes a wg_handshake +// fact (source=infer:wg) stamped with the MOST RECENT peer handshake time — not +// now(). Presence decays from the real handshake instant, so the fact's ts must +// be that instant. We write only when the handshake ADVANCES vs the last fact, +// so a quiet tunnel produces no churn (and presence just decays out, τ=20min). +// +// `wg show` needs CAP_NET_ADMIN; run mavpoll with the cap or set -wg-cmd "sudo wg". +func (p *poller) pollWg(ctx context.Context) error { + fields := strings.Fields(p.wgCmd) + args := append(fields[1:], "show", p.wgIface, "latest-handshakes") + out, err := exec.CommandContext(ctx, fields[0], args...).Output() + if err != nil { + return fmt.Errorf("run %s: %w", p.wgCmd, err) + } + maxTs := parseMaxHandshake(string(out)) + if maxTs == 0 { + return nil // no peer has ever handshaked → drop out of presence + } + hs := time.Unix(maxTs, 0) + prev, err := p.core.LatestFactBySource(ctx, "wg_handshake", "infer:wg") + if err == nil && !hs.After(prev.Ts) { + return nil // not newer → no churn + } + if err != nil && err != ipc.ErrNoFact && !isNoFact(err) { + return fmt.Errorf("read wg_handshake: %w", err) + } + if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{ + Ts: hs, Kind: "env", Key: "wg_handshake", Value: `"up"`, + Source: "infer:wg", Confidence: 1.0, + }); err != nil { + return fmt.Errorf("write wg_handshake: %w", err) + } + log.Printf("mavpoll: wg_handshake @ %s (infer:wg)", hs.Format(time.RFC3339)) + return nil +} + +// parseMaxHandshake — max last-field unix ts across `wg show latest-handshakes` +// lines. Handles both the per-iface form (`\t`) and the `all` form +// (`\t\t`); the timestamp is always the last field. 0 = none. +func parseMaxHandshake(out string) int64 { + var max int64 + for _, line := range strings.Split(out, "\n") { + f := strings.Fields(line) + if len(f) == 0 { + continue + } + ts, err := strconv.ParseInt(f[len(f)-1], 10, 64) + if err == nil && ts > max { + max = ts + } + } + return max +} + +// ---- netdata: active alarms → aggregate severity --------------------------- + +// netdata /api/v1/alarms?active=true returns {"alarms": {"": {..., +// "status": "WARNING"|"CRITICAL"|"CLEAR"|...}}}. We only need the max active +// severity; a rule fires on "critical". The per-alarm detail lives in netdata's +// own UI — we don't re-store it (YAGNI; add a per-alarm fact when a rule needs +// one specific alarm by name). +type netdataAlarms struct { + Alarms map[string]struct { + Status string `json:"status"` + } `json:"alarms"` +} + +func (p *poller) pollNetdata(ctx context.Context, now time.Time) error { + body, err := p.get(ctx, p.netdataURL+"/api/v1/alarms?active=true", "") + if err != nil { + return err + } + var a netdataAlarms + if err := json.Unmarshal(body, &a); err != nil { + return fmt.Errorf("decode alarms: %w", err) + } + return p.writeIfChanged(ctx, "netdata_alarm", "poll:netdata", maxSeverity(a), now) +} + +// maxSeverity reduces active alarms to the aggregate the rule consumes. +func maxSeverity(a netdataAlarms) string { + sev := "clear" + for _, al := range a.Alarms { + switch strings.ToUpper(al.Status) { + case "CRITICAL": + return "critical" // highest — short-circuit + case "WARNING": + sev = "warning" + } + } + return sev +} + +// ---- kuma: monitor_status gauge → aggregate service_down ------------------- + +// Kuma exposes Prometheus text: `monitor_status{...,monitor_name="X"} V` where +// V is 1=up 0=down 2=pending 3=maintenance. We reduce to one aggregate the +// existing ServiceDownRule consumes: "down" if ANY monitor reads 0, else "up". +// Per-service granularity is a later add (a fact per monitor) — the MVP nudge +// only needs "something is down". +var kumaLine = regexp.MustCompile(`^monitor_status\{([^}]*)\}\s+([0-9.eE+-]+)`) + +func (p *poller) pollKuma(ctx context.Context, now time.Time) error { + body, err := p.get(ctx, p.kumaURL, p.kumaKey) + if err != nil { + return err + } + down, seen := kumaAnyDown(body) + if !seen { + return fmt.Errorf("no monitor_status metrics (auth/endpoint wrong?)") + } + val := "up" + if down { + val = "down" + } + return p.writeIfChanged(ctx, "service_down", "poll:uptimekuma", val, now) +} + +// kumaAnyDown parses kuma's Prometheus text: down=true if any monitor reads 0 +// (pending=2/maintenance=3 are not "down"). seen=false ⇒ no monitor_status +// lines matched at all (wrong endpoint or auth rejected before the body). +func kumaAnyDown(body []byte) (down, seen bool) { + for _, line := range strings.Split(string(body), "\n") { + m := kumaLine.FindStringSubmatch(strings.TrimSpace(line)) + if m == nil { + continue + } + seen = true + v, err := strconv.ParseFloat(m[2], 64) + if err != nil { + continue + } + if v == 0 { + down = true + } + } + return down, seen +} + +// ---- helpers --------------------------------------------------------------- + +// writeIfChanged writes a `facts(kind=env)` row only when val differs from the +// latest fact for (key, source). Values are stored JSON-encoded (the store's +// convention: `"down"`, `"critical"`), matching how rules compare f.Value. +func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error { + jv, _ := json.Marshal(val) // string never fails to marshal + prev, err := p.core.LatestFactBySource(ctx, key, source) + switch { + case err == nil && prev.Value == string(jv): + return nil // unchanged → no churn + case err != nil && err != ipc.ErrNoFact && !isNoFact(err): + return fmt.Errorf("read %s: %w", key, err) + } + _, err = p.core.WriteFact(ctx, ipc.WriteFactReq{ + Ts: now, + Kind: "env", + Key: key, + Value: string(jv), + Source: source, + Confidence: 1.0, // a direct reading, not an inference + }) + if err != nil { + return fmt.Errorf("write %s: %w", key, err) + } + log.Printf("mavpoll: %s=%s (%s)", key, val, source) + return nil +} + +// isNoFact — ErrNoFact rehydrated over the wire is wrapped (fmt.Errorf %w), so +// errors.Is is the right check; keep a helper so the switch above reads clean. +func isNoFact(err error) bool { + for e := err; e != nil; { + if e == ipc.ErrNoFact { + return true + } + u, ok := e.(interface{ Unwrap() error }) + if !ok { + return false + } + e = u.Unwrap() + } + return false +} + +func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + if basicUser != "" { + req.SetBasicAuth(basicUser, "") // kuma: API key as username, empty password + } + resp, err := p.http.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20)) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("GET %s: %s", url, resp.Status) + } + return body, nil +} diff --git a/cmd/mavpoll/main_test.go b/cmd/mavpoll/main_test.go new file mode 100644 index 0000000..09bf3cd --- /dev/null +++ b/cmd/mavpoll/main_test.go @@ -0,0 +1,64 @@ +package main + +import ( + "encoding/json" + "testing" +) + +func TestMaxSeverity(t *testing.T) { + parse := func(s string) netdataAlarms { + var a netdataAlarms + if err := json.Unmarshal([]byte(s), &a); err != nil { + t.Fatal(err) + } + return a + } + cases := []struct{ body, want string }{ + {`{"alarms":{}}`, "clear"}, + {`{"alarms":{"a":{"status":"WARNING"}}}`, "warning"}, + {`{"alarms":{"a":{"status":"WARNING"},"b":{"status":"CRITICAL"}}}`, "critical"}, + {`{"alarms":{"a":{"status":"CLEAR"}}}`, "clear"}, + } + for _, c := range cases { + if got := maxSeverity(parse(c.body)); got != c.want { + t.Errorf("maxSeverity(%s) = %q, want %q", c.body, got, c.want) + } + } +} + +func TestKumaAnyDown(t *testing.T) { + cases := []struct { + body string + down, seen bool + }{ + {"", false, false}, + {`monitor_status{monitor_name="web"} 1`, false, true}, + {`monitor_status{monitor_name="web"} 1` + "\n" + `monitor_status{monitor_name="db"} 0`, true, true}, + {`monitor_status{monitor_name="mnt"} 3`, false, true}, // maintenance ≠ down + {`# HELP monitor_status ...`, false, false}, + } + for _, c := range cases { + down, seen := kumaAnyDown([]byte(c.body)) + if down != c.down || seen != c.seen { + t.Errorf("kumaAnyDown(%q) = (%v,%v), want (%v,%v)", c.body, down, seen, c.down, c.seen) + } + } +} + +func TestParseMaxHandshake(t *testing.T) { + cases := []struct { + out string + want int64 + }{ + {"", 0}, + {"pubkeyAAA\t0\n", 0}, // never handshaked + {"pubkeyAAA\t1700000000\n", 1700000000}, + {"pubkeyAAA\t1700000000\npubkeyBBB\t1700000500\n", 1700000500}, // max wins + {"wg0\tpubkeyAAA\t1700000000\nwg0\tpubkeyBBB\t0\n", 1700000000}, // 'all' 3-field form + } + for _, c := range cases { + if got := parseMaxHandshake(c.out); got != c.want { + t.Errorf("parseMaxHandshake(%q) = %d, want %d", c.out, got, c.want) + } + } +} diff --git a/cmd/mavsttd/main.go b/cmd/mavsttd/main.go new file mode 100644 index 0000000..5faabc3 --- /dev/null +++ b/cmd/mavsttd/main.go @@ -0,0 +1,131 @@ +// Package main is mavsttd — maven's stt module process. +// +// Per spec: stt/tts are restart-free, key-free, fail-independent modules — +// separate processes from core, reachable over the worker boundary +// (internal/worker). Core dials mavsttd's unix socket and ships audio bytes +// for transcription; mavsttd ships text back. +// +// With -model : loads a whisper.cpp ggml model (e.g. ggml-small.bin) +// for real transcription. Without -model: serves the stub transcriber +// (deterministic no-model floor) so the loop is exercisable end-to-end +// without weights. +// +// Module topology: +// +// $ mavsttd -socket /run/user/$UID/maven/stt.sock +// +// The daemon's config points at this socket: +// +// "stt": { "socket": "/run/user/1000/maven/stt.sock" } +// +// Both processes are same-user on the box ⇒ the 0600 socket floor (same +// unix user) is sufficient today; the wg / mTLS cuts in internal/auth are +// for the NETWORK radius (client↔core), not the local module radius. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net" + "os" + "os/signal" + "syscall" + + "github.com/kami/maven/internal/worker" +) + +func main() { + if err := run(os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, "mavsttd:", err) + os.Exit(1) + } +} + +func run(args []string) error { + sock := flag.String("socket", defaultSocket("stt.sock"), "unix socket path") + model := flag.String("model", "", "path to whisper ggml model file") + flag.CommandLine.Parse(args) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + defer stop() + + var t worker.Transcriber + if *model != "" { + w, err := newWhisperHandler(*model) + if err != nil { + return fmt.Errorf("whisper: %w", err) + } + t = w + defer func() { + log.Printf("mavsttd: closing whisper model") + w.Close() + }() + log.Printf("mavsttd: loaded whisper model from %s", *model) + } else { + log.Printf("mavsttd: no model specified, using stub handler") + t = &stubHandler{} + } + + srv := worker.NewServer(*sock, t) + if err := srv.Listen(); err != nil { + return err + } + defer srv.Close() + log.Printf("mavsttd: worker listening on %s", srv.Path()) + + errCh := make(chan error, 1) + go func() { errCh <- srv.Serve() }() + select { + case <-ctx.Done(): + log.Printf("mavsttd: shutdown signal received") + srv.Close() + return nil + case err := <-errCh: + if err != nil && !errors.Is(err, net.ErrClosed) { + return err + } + return nil + } +} + +// stubHandler — worker.Transcriber that delegates to the package Stub. Tiny +// now; the production swap replaces this whole struct with a faster-whisper +// / vosk-backed struct (the same Worker.Transcriber interface). +type stubHandler struct{} + +func (h *stubHandler) Transcribe(ctx context.Context, req worker.TranscribeReq) (worker.TranscribeResp, error) { + // delegate to the same deterministic Stub the daemon could have wired + // in-process; mavsttd is the "separate process" equivalent. + _ = req + // hash for variation; same approach as stt.Stub. + if len(req.Audio.Bytes) == 0 { + return worker.TranscribeResp{Text: "maven, что у меня сегодня", Confidence: 1.0}, nil + } + // vary phrase by first byte for visibility in logs/tests. + phrases := []string{ + "maven, отметь что я выпил воды", + "maven, напомни через 4 часа размяться", + "maven, restart nginx", + "maven, что у меня сегодня по календарю", + "note: staggered cooldown by time of day", + "slept 6h", + } + idx := int(req.Audio.Bytes[0]) % len(phrases) + return worker.TranscribeResp{Text: phrases[idx], Confidence: 1.0}, nil +} + +// defaultSocket returns XDG_RUNTIME_DIR/maven/ if set, falling back +// to a homedir-relative path (mirrors config.defaultRuntimeDir). +func defaultSocket(name string) string { + if x := os.Getenv("XDG_RUNTIME_DIR"); x != "" { + return x + "/maven/" + name + } + home, err := os.UserHomeDir() + if err != nil || home == "" { + return name + } + return home + "/.local/share/maven/" + name +} diff --git a/cmd/mavsttd/whisper_handler.go b/cmd/mavsttd/whisper_handler.go new file mode 100644 index 0000000..2ed64cb --- /dev/null +++ b/cmd/mavsttd/whisper_handler.go @@ -0,0 +1,119 @@ +package main + +/* +#cgo CFLAGS: -I${SRCDIR}/../../deps/include -I${SRCDIR}/../../deps/whisper.cpp/ggml/include +#cgo LDFLAGS: -L${SRCDIR}/../../deps/lib -Wl,-rpath,${SRCDIR}/../../deps/lib -lwhisper -lggml -lggml-base -lggml-cpu -lggml-vulkan -lm -lstdc++ -fopenmp +#include +#include +*/ +import "C" + +import ( + "context" + "fmt" + "math" + "unsafe" + + "github.com/kami/maven/internal/worker" +) + +type whisperHandler struct { + ctx *C.struct_whisper_context +} + +func newWhisperHandler(modelPath string) (*whisperHandler, error) { + cparams := C.whisper_context_default_params() + cPath := C.CString(modelPath) + defer C.free(unsafe.Pointer(cPath)) + + ctx := C.whisper_init_from_file_with_params(cPath, cparams) + if ctx == nil { + return nil, fmt.Errorf("whisper: failed to init from %s", modelPath) + } + return &whisperHandler{ctx: ctx}, nil +} + +func (h *whisperHandler) Transcribe(ctx context.Context, req worker.TranscribeReq) (worker.TranscribeResp, error) { + a := req.Audio + if len(a.Bytes) == 0 { + return worker.TranscribeResp{}, fmt.Errorf("whisper: empty audio") + } + + nSamples := len(a.Bytes) / 2 + samples := make([]float32, nSamples) + for i := 0; i < nSamples; i++ { + s := int16(a.Bytes[i*2]) | int16(a.Bytes[i*2+1])<<8 + samples[i] = float32(s) / 32768.0 + } + + params := C.whisper_full_default_params(C.WHISPER_SAMPLING_GREEDY) + params.print_progress = false + params.print_realtime = false + params.print_timestamps = false + params.print_special = false + params.n_threads = C.int(4) + params.single_segment = true + + lang := C.CString(req.Lang) + defer C.free(unsafe.Pointer(lang)) + params.language = lang + params.detect_language = false + + cSamples := (*C.float)(unsafe.Pointer(&samples[0])) + res := C.whisper_full(h.ctx, params, cSamples, C.int(nSamples)) + if res != 0 { + return worker.TranscribeResp{}, fmt.Errorf("whisper: full failed: %d", int(res)) + } + + nSegments := int(C.whisper_full_n_segments(h.ctx)) + if nSegments == 0 { + return worker.TranscribeResp{Text: "", Confidence: 0}, nil + } + + var text string + totalLogProb := float64(0) + totalTokens := 0 + + for i := 0; i < nSegments; i++ { + cSeg := C.whisper_full_get_segment_text(h.ctx, C.int(i)) + if cSeg != nil { + text += C.GoString(cSeg) + } + + nTokens := int(C.whisper_full_n_tokens(h.ctx, C.int(i))) + for j := 0; j < nTokens; j++ { + p := float64(C.whisper_full_get_token_p(h.ctx, C.int(i), C.int(j))) + if p > 0 { + totalLogProb += math.Log(p) + totalTokens++ + } + } + } + + confidence := 0.0 + if totalTokens > 0 { + avgLogProb := totalLogProb / float64(totalTokens) + confidence = math.Exp(avgLogProb) + } + + noSpeechProb := float64(C.whisper_full_get_segment_no_speech_prob(h.ctx, 0)) + if noSpeechProb > 0.9 { + confidence = 0 + } + + if math.IsNaN(confidence) || math.IsInf(confidence, 0) { + confidence = 0 + } + + return worker.TranscribeResp{ + Text: text, + Confidence: confidence, + }, nil +} + +func (h *whisperHandler) Close() { + if h.ctx != nil { + C.whisper_free(h.ctx) + h.ctx = nil + } +} diff --git a/cmd/mavttsd/main.go b/cmd/mavttsd/main.go new file mode 100644 index 0000000..92df454 --- /dev/null +++ b/cmd/mavttsd/main.go @@ -0,0 +1,129 @@ +// Package main is mavttsd — maven's tts module process. +// +// Sibling to cmd/mavsttd: same worker boundary, opposite job (synthesize vs +// transcribe). Same restart-free, key-free, fail-independent invariant. +// +// With -piper -model : calls piper for real TTS (ru_RU +// voice at models/tts/ru_RU-irina-medium.onnx). Without flags: serves the +// stub synthesizer (200ms tone) for exercisable end-to-end testing. +// +// $ mavttsd -socket /run/user/$UID/maven/tts.sock +// "tts": { "socket": "/run/user/1000/maven/tts.sock", "lang": "ru" } +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "log" + "net" + "os" + "os/signal" + "syscall" + + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/worker" +) + +func main() { + if err := run(os.Args[1:]); err != nil { + fmt.Fprintln(os.Stderr, "mavttsd:", err) + os.Exit(1) + } +} + +func run(args []string) error { + sock := flag.String("socket", defaultSocket("tts.sock"), "unix socket path") + piperBin := flag.String("piper", "", "path to piper binary") + model := flag.String("model", "", "path to piper onnx model file") + espeakData := flag.String("espeak_data", "", "path to espeak-ng data directory") + tashkeelModel := flag.String("tashkeel_model", "", "path to libtashkeel onnx model") + + flag.CommandLine.Parse(args) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM, syscall.SIGHUP) + defer stop() + + var s worker.Synthesizer + if *piperBin != "" && *model != "" { + s = newPiperHandler(*piperBin, *model, *espeakData, *tashkeelModel) + log.Printf("mavttsd: using piper tts (%s, model=%s)", *piperBin, *model) + } else { + log.Printf("mavttsd: no piper/model specified, using stub handler") + s = &stubHandler{} + } + + srv := worker.NewSynthesizerServer(*sock, s) + if err := srv.Listen(); err != nil { + return err + } + defer srv.Close() + log.Printf("mavttsd: worker listening on %s", srv.Path()) + + errCh := make(chan error, 1) + go func() { errCh <- srv.Serve() }() + select { + case <-ctx.Done(): + log.Printf("mavttsd: shutdown signal received") + srv.Close() + return nil + case err := <-errCh: + if err != nil && !errors.Is(err, net.ErrClosed) { + return err + } + return nil + } +} + +// stubHandler — worker.Synthesizer that delegates to the tts Stub. The +// production swap replaces this struct with a silero / piper-backed handler. +type stubHandler struct{} + +func (h *stubHandler) Synthesize(ctx context.Context, req worker.SynthesizeReq) (worker.SynthesizeResp, error) { + _ = ctx + // 200ms tone, freq keyed by first byte of text — same shape as tts.Stub, + // kept locally so this module has zero coupling to the daemon package + // (mavttsd running shouldn't drag stt/tts package symbols here; they're + // siblings in the topology). + const samples = 3200 // 200ms @ 16k + pcm := make([]byte, samples*2) + freq := 220.0 + if len(req.Text) > 0 { + freq = 180.0 + float64(req.Text[0]%6)*60 + } + for i := 0; i < samples; i++ { + t := float64(i) / 16000.0 + v := int16(12000 * sin(2*pi*freq*t)) + pcm[i*2] = byte(v) + pcm[i*2+1] = byte(v >> 8) + } + return worker.SynthesizeResp{Audio: audio.Audio{Format: audio.PCM16kMono, Bytes: pcm}}, nil +} + +const pi = 3.141592653589793 + +// tiny stdlib-free sin approximation — keeps mavttsd out of math import. +// Adequate for a tone generator; the production model returns real audio. +func sin(x float64) float64 { + // reduce to [-pi, +pi] + mod := x - pi*2*float64(int(x/(pi*2))) + if mod > pi { + mod -= pi * 2 + } else if mod < -pi { + mod += pi * 2 + } + // 4-term Taylor series around 0; decent for the small amplitudes here. + return mod - mod*mod*mod/6 + mod*mod*mod*mod*mod/120 - mod*mod*mod*mod*mod*mod*mod/5040 +} + +func defaultSocket(name string) string { + if x := os.Getenv("XDG_RUNTIME_DIR"); x != "" { + return x + "/maven/" + name + } + home, err := os.UserHomeDir() + if err != nil || home == "" { + return name + } + return home + "/.local/share/maven/" + name +} \ No newline at end of file diff --git a/cmd/mavttsd/piper_handler.go b/cmd/mavttsd/piper_handler.go new file mode 100644 index 0000000..308a193 --- /dev/null +++ b/cmd/mavttsd/piper_handler.go @@ -0,0 +1,129 @@ +package main + +import ( + "bytes" + "context" + "encoding/binary" + "fmt" + "io" + "os/exec" + + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/worker" +) + +type piperHandler struct { + piperPath string + modelPath string + configPath string + espeakData string + tashkeelModel string +} + +func newPiperHandler(piperPath, modelPath, espeakData, tashkeelModel string) *piperHandler { + return &piperHandler{ + piperPath: piperPath, + modelPath: modelPath, + configPath: modelPath + ".json", + espeakData: espeakData, + tashkeelModel: tashkeelModel, + } +} + +func (h *piperHandler) Synthesize(ctx context.Context, req worker.SynthesizeReq) (worker.SynthesizeResp, error) { + var stderr bytes.Buffer + + args := []string{ + "--model", h.modelPath, + "--config", h.configPath, + "--output_raw", + "--quiet", + } + if h.espeakData != "" { + args = append(args, "--espeak_data", h.espeakData) + } + if h.tashkeelModel != "" { + args = append(args, "--tashkeel_model", h.tashkeelModel) + } + + cmd := exec.CommandContext(ctx, h.piperPath, args...) + cmd.Stderr = &stderr + + stdin, err := cmd.StdinPipe() + if err != nil { + return worker.SynthesizeResp{}, fmt.Errorf("piper: stdin pipe: %w", err) + } + + stdout, err := cmd.StdoutPipe() + if err != nil { + return worker.SynthesizeResp{}, fmt.Errorf("piper: stdout pipe: %w", err) + } + + if err := cmd.Start(); err != nil { + return worker.SynthesizeResp{}, fmt.Errorf("piper: start: %w", err) + } + + _, _ = io.WriteString(stdin, req.Text) + stdin.Close() + + rawPCM, readErr := io.ReadAll(stdout) + + waitErr := cmd.Wait() + if waitErr != nil { + errMsg := stderr.String() + if errMsg != "" { + return worker.SynthesizeResp{}, fmt.Errorf("piper: %s: %s", waitErr, errMsg) + } + return worker.SynthesizeResp{}, fmt.Errorf("piper: %w", waitErr) + } + if readErr != nil { + return worker.SynthesizeResp{}, fmt.Errorf("piper: read stdout: %w", readErr) + } + + if len(rawPCM) == 0 { + return worker.SynthesizeResp{}, fmt.Errorf("piper: no audio output") + } + + resampled := resample22050To16000(rawPCM) + + return worker.SynthesizeResp{ + Audio: audio.Audio{ + Format: audio.PCM16kMono, + Bytes: resampled, + }, + }, nil +} + +// resample22050To16000 converts raw 16-bit PCM from 22050 Hz to 16000 Hz +// using linear interpolation. +func resample22050To16000(input []byte) []byte { + if len(input) < 2 { + return nil + } + + nSamples := len(input) / 2 + outSamples := int(float64(nSamples) * 16000.0 / 22050.0) + output := make([]byte, outSamples*2) + + ratio := 22050.0 / 16000.0 + + for i := 0; i < outSamples; i++ { + srcPos := float64(i) * ratio + srcIdx := int(srcPos) + frac := srcPos - float64(srcIdx) + + if srcIdx >= nSamples-1 { + v := int16(binary.LittleEndian.Uint16(input[(nSamples-1)*2:])) + binary.LittleEndian.PutUint16(output[i*2:], uint16(v)) + continue + } + + v0 := int16(binary.LittleEndian.Uint16(input[srcIdx*2:])) + v1 := int16(binary.LittleEndian.Uint16(input[(srcIdx+1)*2:])) + + interpolated := int16(float64(v0)*(1-frac) + float64(v1)*frac) + binary.LittleEndian.PutUint16(output[i*2:], uint16(interpolated)) + } + + return output +} diff --git a/cmd/mavweb/dash.html b/cmd/mavweb/dash.html new file mode 100644 index 0000000..594867f --- /dev/null +++ b/cmd/mavweb/dash.html @@ -0,0 +1,36 @@ + +maven dash + +

presence

+

{{.Presence.Bucket}} — score {{printf "%.2f" .Presence.Score}} ({{ago .Presence.Updated}})

+
обновляется каждые 10с
+

nudges

+ +{{range .Nudges}}{{end}} +
whenrulechanoutcomemessage
{{ago .Ts}}{{.Rule}}{{.Channel}}{{.Outcome}}{{.Message}}
+

facts

+ +{{range .Facts}}{{end}} +
whenkindkeyvaluesourceconf
{{ago .Ts}}{{.Kind}}{{.Key}}{{.Value}}{{.Source}}{{printf "%.2f" .Confidence}}
+

notes

+ +{{range .Notes}}{{end}} +
whensourcetext
{{ago .Ts}}{{.Source}}{{.Text}}
+ \ No newline at end of file diff --git a/cmd/mavweb/main.go b/cmd/mavweb/main.go new file mode 100644 index 0000000..bb42d1b --- /dev/null +++ b/cmd/mavweb/main.go @@ -0,0 +1,496 @@ +package main + +import ( + "cmp" + "context" + "embed" + "encoding/binary" + "encoding/json" + "errors" + "flag" + "fmt" + "html/template" + "io" + "io/fs" + "log" + "net" + "net/http" + "net/url" + "os" + "os/signal" + "strings" + "time" + + "github.com/coder/websocket" + "github.com/kami/maven/internal/audio" + "github.com/kami/maven/internal/ipc" + "github.com/kami/maven/internal/voice" +) + +// presenceSignals — the only fact keys /api/signal may write. mavweb is a +// network-facing surface inside wg; an allowlist keeps a compromised caller +// boxed to forging weak presence signals (reachability, multi-source, never +// truth) — it can't write arbitrary facts. ponytail: floor auth (wg-only); a +// per-signal token belongs here if the tunnel ever hosts untrusted devices. +var presenceSignals = map[string]string{ + "desk_active": "infer:hyprland", + "page_heartbeat": "infer:heartbeat", + "wg_handshake": "infer:wg", +} + +//go:embed static/* +var staticFiles embed.FS + +//go:embed dash.html +var dashHTML string + +// dashTmpl — the monitoring read surface, server-rendered from dash.html (no JS, +// no client fetch); meta-refresh keeps it live. html/template escapes the user +// text in facts/nudges. Read-only: browses the append-only store via CoreAPI, +// never writes — the store IS the audit trail, this just shows it. +var dashTmpl = template.Must(template.New("dash").Funcs(template.FuncMap{ + "ago": func(t time.Time) string { return time.Since(t).Round(time.Second).String() + " ago" }, +}).Parse(dashHTML)) + +func noCache(h http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate") + h.ServeHTTP(w, r) + }) +} + +func main() { + addr := flag.String("addr", ":9200", "HTTP listen address") + voiceAddr := flag.String("voice", "127.0.0.1:9100", "voice server TCP addr (host:port)") + // ntfyWS: the ntfy WebSocket subscribe URL the PWA connects to for in-app + // nudge delivery, e.g. wss://ntfy.kvmx.ru/maven/ws?auth=. The + // client subscribes directly (lowest overhead — mavweb isn't in the path); + // we only serve it the URL so the deny-all auth token stays deployment + // config, never baked into the static JS. Empty ⇒ /api/ntfy returns 204 and + // the PWA skips subscription (voice-only, as before). + ntfyWS := flag.String("ntfy", "", "ntfy WebSocket subscribe URL served to the PWA (e.g. wss://host/topic/ws?auth=...)") + // coreSock: mavend's IPC socket. When set, /api/signal writes presence + // facts through CoreAPI (page heartbeat from the PWA, desk_active from a PC + // script). Empty ⇒ /api/signal returns 503 and presence stays unfed. + coreSock := flag.String("core", "", "mavend IPC socket path for presence-signal ingest (empty = disabled)") + flag.Parse() + + var core ipc.CoreAPI + if *coreSock != "" { + c, err := ipc.Dial(*coreSock) + if err != nil { + log.Fatalf("dial core %s: %v", *coreSock, err) + } + defer c.Close() + core = c + } + + mux := http.NewServeMux() + + sub, err := fs.Sub(staticFiles, "static") + if err != nil { + log.Fatalf("static fs: %v", err) + } + mux.Handle("/", noCache(http.FileServer(http.FS(sub)))) + + mux.HandleFunc("/ws", func(w http.ResponseWriter, r *http.Request) { + handleWS(w, r, *voiceAddr) + }) + mux.HandleFunc("/api/ptt", func(w http.ResponseWriter, r *http.Request) { + handlePTT(w, r, *voiceAddr) + }) + mux.HandleFunc("/api/ping", func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte("pong")) + }) + mux.HandleFunc("/api/ntfy", func(w http.ResponseWriter, r *http.Request) { + if *ntfyWS == "" { + w.WriteHeader(http.StatusNoContent) // not configured → PWA skips + return + } + w.Header().Set("Content-Type", "text/plain") + w.Write([]byte(*ntfyWS)) + }) + mux.HandleFunc("/api/signal", func(w http.ResponseWriter, r *http.Request) { + handleSignal(w, r, core) + }) + mux.HandleFunc("/dash", func(w http.ResponseWriter, r *http.Request) { + handleDash(w, r, core) + }) + // /tools — the authed enable surface. maven proposes acts she can't run; + // this page is where a human reviews and enables them (proposed→enabled). + // Enabling is the boundary-moving act (maven.md), so it lives ONLY here, + // behind wg+nginx+auth — never the voice/chat path. + mux.HandleFunc("/tools", func(w http.ResponseWriter, r *http.Request) { + handleTools(w, r, core) + }) + + srv := &http.Server{Addr: *addr, Handler: mux} + + go func() { + sig := make(chan os.Signal, 1) + signal.Notify(sig, os.Interrupt) + <-sig + log.Println("shutting down...") + srv.Close() + }() + + log.Printf("mavweb listening on %s, voice → %s", *addr, *voiceAddr) + if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatal(err) + } +} + +func handleWS(w http.ResponseWriter, r *http.Request, voiceAddr string) { + conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{ + OriginPatterns: []string{"*"}, + }) + if err != nil { + log.Printf("ws accept: %v", err) + return + } + defer conn.Close(websocket.StatusNormalClosure, "bye") + + ctx := r.Context() + + var d net.Dialer + tc, err := d.DialContext(ctx, "tcp", voiceAddr) + if err != nil { + log.Printf("dial voice: %v", err) + writeWSErr(conn, ctx, "voice unavailable") + return + } + defer tc.Close() + + for { + _, msg, err := conn.Read(ctx) + if err != nil { + log.Printf("ws read: %v", err) + return + } + if len(msg) < 4 { + log.Printf("ws msg too short (%d bytes)", len(msg)) + continue + } + + log.Printf("ws got %d bytes from client", len(msg)) + pcm := audio.Audio{Format: audio.PCM16kMono, Bytes: msg} + + req := voice.Request{ + ID: uint64(time.Now().UnixNano()), + Method: voice.MethodPushToTalk, + Params: mustMarshal(voice.PushToTalkReq{ + Audio: pcm, + Lang: "mixed", + Surface: voice.SurfacePCClient, + }), + } + + if err := writeFrame(tc, &req); err != nil { + log.Printf("write voice req: %v", err) + return + } + + // Read frames until we get the matching Response (handling any interleaved Pushes) + for { + resp, push, err := readOneFrame(tc) + if err != nil { + log.Printf("read voice: %v", err) + return + } + if push != nil { + data, _ := json.Marshal(push) + conn.Write(ctx, websocket.MessageText, data) + continue + } + if resp.Error != nil { + writeWSErr(conn, ctx, resp.Error.Message) + break + } + var pttResp voice.PushToTalkResp + if err := json.Unmarshal(resp.Result, &pttResp); err != nil { + log.Printf("unmarshal resp: %v", err) + break + } + if pttResp.ReplyText != "" { + conn.Write(ctx, websocket.MessageText, []byte(pttResp.ReplyText)) + } + if len(pttResp.ReplyAudio.Bytes) > 0 { + conn.Write(ctx, websocket.MessageBinary, pttResp.ReplyAudio.Bytes) + } + break + } + } +} + +// handleSignal ingests one presence signal and writes a fresh fact through +// CoreAPI. The fact's timestamp (now) is all the presence scorer reads; value +// is a marker. Only allowlisted keys are accepted (see presenceSignals). +func handleSignal(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI) { + if r.Method != http.MethodPost { + http.Error(w, "POST only", http.StatusMethodNotAllowed) + return + } + if core == nil { + http.Error(w, "presence ingest disabled (no -core)", http.StatusServiceUnavailable) + return + } + key := r.URL.Query().Get("key") + source, ok := presenceSignals[key] + if !ok { + http.Error(w, "unknown signal key", http.StatusBadRequest) + return + } + // kind=env: an observation about the device/surface, NOT a self-fact — a + // passive signal never writes truth about you (spec), it only feeds + // presence. confidence 1.0: the reading ("input happened") is certain; + // presence applies its own per-signal weight/decay on top. + if _, err := core.WriteFact(r.Context(), ipc.WriteFactReq{ + Ts: time.Now(), + Kind: "env", + Key: key, + Value: `"active"`, + Source: source, + Confidence: 1.0, + }); err != nil { + log.Printf("signal %s: %v", key, err) + http.Error(w, "write failed", http.StatusBadGateway) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func handleDash(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI) { + if core == nil { + http.Error(w, "dash disabled (no -core)", http.StatusServiceUnavailable) + return + } + ctx := r.Context() + pres, err1 := core.Presence(ctx) + facts, err2 := core.RecentFacts(ctx, 50) + nudges, err3 := core.RecentNudges(ctx, 50) + notes, err4 := core.RecentNotes(ctx, 50) + if err := cmp.Or(err1, err2, err3, err4); err != nil { + log.Printf("dash: %v", err) + http.Error(w, "core read failed", http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + if err := dashTmpl.Execute(w, struct { + Presence ipc.Presence + Facts []ipc.Fact + Nudges []ipc.Nudge + Notes []ipc.Note + }{pres, facts, nudges, notes}); err != nil { + log.Printf("dash render: %v", err) + } +} + +// toolsTmpl — the enable surface. Server-rendered, no JS: a plain HTML form +// POSTs back to /tools to enable a proposal. html/template escapes tool names + +// utterances (they came from voice STT — untrusted text). +var toolsTmpl = template.Must(template.New("tools").Funcs(template.FuncMap{ + "join": strings.Join, +}).Parse(toolsHTML)) + +const toolsHTML = `maven · tools + +

maven · tools

+{{if .Msg}}
{{.Msg}}
{{end}} +

proposed ({{len .Proposed}})

+{{if .Proposed}}

maven drafted these from acts she couldn't run. Fill the command (argv, space-separated) and enable.

+ +{{range .Proposed}} + + +{{end}}
namefrom utteranceenable as
{{.Name}}{{.Utterance}}
+ + + +
+{{else}}

none pending.

{{end}} +

enabled ({{len .Enabled}})

+{{if .Enabled}} +{{range .Enabled}} +{{end}}
namecommand
{{.Name}}{{join .Cmd " "}}{{if .Destructive}}destructive{{end}}
+{{else}}

none enabled.

{{end}} +` + +// handleTools serves the enable surface (GET) and applies an enable (POST). +// POST fields: name, cmd (space-separated argv), destructive (checkbox). cmd is +// whitespace-split — argv with embedded spaces isn't supported (ponytail: no +// shell-word parsing; the box owner controls this input, quote a wrapper script +// if an arg needs spaces). +func handleTools(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI) { + if core == nil { + http.Error(w, "tools disabled (no -core)", http.StatusServiceUnavailable) + return + } + ctx := r.Context() + var msg string + if r.Method == http.MethodPost { + name := strings.TrimSpace(r.FormValue("name")) + cmd := strings.Fields(r.FormValue("cmd")) + destructive := r.FormValue("destructive") != "" + if name == "" || len(cmd) == 0 { + http.Error(w, "name and cmd required", http.StatusBadRequest) + return + } + if err := core.EnableTool(ctx, name, cmd, destructive, time.Now()); err != nil { + log.Printf("tools: enable %q: %v", name, err) + http.Error(w, "enable failed: "+err.Error(), http.StatusBadGateway) + return + } + msg = "enabled " + name + } + proposed, err1 := core.ListTools(ctx, "proposed") + enabled, err2 := core.ListTools(ctx, "enabled") + if err := cmp.Or(err1, err2); err != nil { + log.Printf("tools: %v", err) + http.Error(w, "core read failed", http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + if err := toolsTmpl.Execute(w, struct { + Msg string + Proposed []ipc.Tool + Enabled []ipc.Tool + }{msg, proposed, enabled}); err != nil { + log.Printf("tools render: %v", err) + } +} + +func writeWSErr(conn *websocket.Conn, ctx context.Context, msg string) { + conn.Write(ctx, websocket.MessageText, []byte(`{"error":"`+msg+`"}`)) +} + +func writeFrame(w io.Writer, v any) error { + body, err := json.Marshal(v) + if err != nil { + return fmt.Errorf("marshal: %w", err) + } + const maxFrame = 64 << 20 + if len(body) > maxFrame { + return fmt.Errorf("frame too large: %d", len(body)) + } + var hdr [4]byte + binary.BigEndian.PutUint32(hdr[:], uint32(len(body))) + if _, err := w.Write(hdr[:]); err != nil { + return err + } + _, err = w.Write(body) + return err +} + +func readFrame(r io.Reader, v any) error { + var hdr [4]byte + if _, err := io.ReadFull(r, hdr[:]); err != nil { + return err + } + n := binary.BigEndian.Uint32(hdr[:]) + const maxFrame = 64 << 20 + if n > maxFrame { + return fmt.Errorf("frame too large: %d", n) + } + buf := make([]byte, n) + if _, err := io.ReadFull(r, buf); err != nil { + return err + } + return json.Unmarshal(buf, v) +} + +func readOneFrame(r io.Reader) (*voice.Response, *voice.Push, error) { + var raw struct { + ID uint64 `json:"id"` + Result json.RawMessage `json:"r,omitempty"` + Error *voice.RpcError `json:"e,omitempty"` + Kind voice.PushKind `json:"kind,omitempty"` + Params json.RawMessage `json:"p,omitempty"` + } + if err := readFrame(r, &raw); err != nil { + return nil, nil, err + } + if raw.Kind != "" && raw.ID == 0 { + return nil, &voice.Push{Kind: raw.Kind, Params: raw.Params}, nil + } + return &voice.Response{ID: raw.ID, Result: raw.Result, Error: raw.Error}, nil, nil +} + +func handlePTT(w http.ResponseWriter, r *http.Request, voiceAddr string) { + if r.Method != http.MethodPost { + http.Error(w, "POST only", 405) + return + } + body, err := io.ReadAll(r.Body) + if err != nil { + http.Error(w, err.Error(), 400) + return + } + if len(body) < 4 { + http.Error(w, "too short", 400) + return + } + + log.Printf("ptt got %d bytes from client", len(body)) + + pcm := audio.Audio{Format: audio.PCM16kMono, Bytes: body} + + var d net.Dialer + tc, err := d.DialContext(r.Context(), "tcp", voiceAddr) + if err != nil { + log.Printf("ptt dial voice: %v", err) + http.Error(w, "voice unavailable", 503) + return + } + defer tc.Close() + + req := voice.Request{ + ID: uint64(time.Now().UnixNano()), + Method: voice.MethodPushToTalk, + Params: mustMarshal(voice.PushToTalkReq{ + Audio: pcm, + Lang: "mixed", + Surface: voice.SurfacePCClient, + }), + } + if err := writeFrame(tc, &req); err != nil { + log.Printf("ptt write: %v", err) + http.Error(w, err.Error(), 500) + return + } + + for { + resp, push, err := readOneFrame(tc) + if err != nil { + log.Printf("ptt read: %v", err) + http.Error(w, err.Error(), 500) + return + } + if push != nil { + continue + } + if resp.Error != nil { + http.Error(w, resp.Error.Message, 500) + return + } + var pttResp voice.PushToTalkResp + if err := json.Unmarshal(resp.Result, &pttResp); err != nil { + http.Error(w, err.Error(), 500) + return + } + w.Header().Set("Content-Type", "audio/l16;rate=16000;channels=1") + w.Header().Set("X-Reply-Text", url.QueryEscape(pttResp.ReplyText)) + w.Write(pttResp.ReplyAudio.Bytes) + return + } +} + +func mustMarshal(v any) json.RawMessage { + b, err := json.Marshal(v) + if err != nil { + panic(err) + } + return b +} diff --git a/cmd/mavweb/nginx.conf b/cmd/mavweb/nginx.conf new file mode 100644 index 0000000..493c8a7 --- /dev/null +++ b/cmd/mavweb/nginx.conf @@ -0,0 +1,27 @@ +# Optional nginx config: put this in /etc/nginx/sites-available/voice.kvmx.ru +# and symlink to sites-enabled. The phone accesses http://voice.kvmx.ru:9200/. +# Alternatively, mavweb can bind 10.42.0.1:9200 directly (no nginx needed). +# +# sudo ln -sf /etc/nginx/sites-available/voice.kvmx.ru /etc/nginx/sites-enabled/ +# sudo nginx -t && sudo systemctl reload nginx + +server { + listen 10.42.0.1:9200; + listen 192.168.1.104:9200; + server_name voice.kvmx.ru; + + allow 10.42.0.0/24; + allow 192.168.1.0/24; + deny all; + + location / { + proxy_pass http://127.0.0.1:9201; + proxy_http_version 1.1; + proxy_set_header Upgrade $http_upgrade; + proxy_set_header Connection "Upgrade"; + proxy_set_header Host $host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Proto $scheme; + } +} diff --git a/cmd/mavweb/static/app.js b/cmd/mavweb/static/app.js new file mode 100644 index 0000000..b93eccd --- /dev/null +++ b/cmd/mavweb/static/app.js @@ -0,0 +1,242 @@ +(() => { + const btn = document.getElementById("btn"); + const status = document.getElementById("status"); + const log = document.getElementById("log"); + + let mediaRecorder = null; + let recordingChunks = []; + let isRecording = false; + let recordingCancel = false; + let isBusy = false; + + function setBtnIdle() { btn.classList.remove("active"); btn.innerHTML = "🎙"; btn.disabled = false; } + function setBtnActive() { btn.classList.add("active"); btn.innerHTML = "■"; } + + function startRecording() { + if (isRecording || isBusy) return; + isRecording = true; + recordingCancel = false; + recordingChunks = []; + status.textContent = "recording... tap to stop"; + setBtnActive(); + + navigator.mediaDevices.getUserMedia({ + audio: { sampleRate: 48000, channelCount: 1, echoCancellation: true, noiseSuppression: true } + }) + .then((stream) => { + if (recordingCancel) { + stream.getTracks().forEach(t => t.stop()); + return; + } + const mr = new MediaRecorder(stream, { mimeType: "audio/webm;codecs=opus" }); + mediaRecorder = mr; + mr.ondataavailable = (e) => { if (e.data.size > 0) recordingChunks.push(e.data); }; + mr.onstop = () => { + stream.getTracks().forEach(t => t.stop()); + processRecording(); + }; + mr.start(100); + }) + .catch((err) => { + status.textContent = "mic error: " + err.message; + isRecording = false; + setBtnIdle(); + }); + } + + function stopRecording() { + if (!isRecording) return; + isRecording = false; + if (!mediaRecorder) { + recordingCancel = true; + setBtnIdle(); + return; + } + status.textContent = "stopping..."; + mediaRecorder.stop(); + mediaRecorder = null; + } + + function processRecording() { + status.textContent = "processing..."; + const blob = new Blob(recordingChunks, { type: "audio/webm" }); + if (blob.size < 200) { status.textContent = "too short"; setBtnIdle(); return; } + + const ac = new (window.AudioContext || window.webkitAudioContext)({ sampleRate: 16000 }); + blob.arrayBuffer().then((buf) => ac.decodeAudioData(buf)) + .then((audioBuffer) => { + const srcRate = audioBuffer.sampleRate; + const srcChan = audioBuffer.numberOfChannels; + const srcLen = audioBuffer.length; + const ratio = 16000 / srcRate; + const dstLen = Math.floor(srcLen * ratio); + + const srcData = new Float32Array(srcLen); + for (let i = 0; i < srcLen; i++) { + let s = 0; + for (let c = 0; c < srcChan; c++) s += audioBuffer.getChannelData(c)[i]; + srcData[i] = s / srcChan; + } + + const dstData = new Int16Array(dstLen); + for (let i = 0; i < dstLen; i++) { + const srcIdx = i / ratio; + const lo = Math.floor(srcIdx); + const hi = Math.min(lo + 1, srcLen - 1); + const frac = srcIdx - lo; + const sample = srcData[lo] + (srcData[hi] - srcData[lo]) * frac; + const clamped = Math.max(-1, Math.min(1, sample)); + dstData[i] = clamped < 0 ? clamped * 32768 : clamped * 32767; + } + + ac.close(); + sendPCM(new Uint8Array(dstData.buffer)); + }) + .catch((err) => { + status.textContent = "decode error: " + err.message; + setBtnIdle(); + }); + } + + function testFetch() { + fetch("/api/ping").then(r => r.text()).then(t => { + if (t === "pong") appendLog("server reachable", ""); + else appendLog("unexpected ping: " + t, "error"); + }).catch(e => appendLog("fetch failed: " + e.message, "error")); + } + + function sendPCM(pcm) { + isBusy = true; + btn.disabled = true; + status.textContent = "sending..."; + appendLog("sending " + pcm.length + " bytes", ""); + + fetch("/api/ptt", { method: "POST", body: pcm }) + .then(async (res) => { + if (!res.ok) { + const txt = await res.text(); + throw new Error(txt); + } + const replyText = res.headers.get("X-Reply-Text"); + if (replyText) { + const txt = decodeURIComponent(replyText); + appendLog(txt, "reply"); + } + const audioData = await res.arrayBuffer(); + if (audioData.byteLength > 0) { + playPCM(new Uint8Array(audioData)); + } else { + status.textContent = "no reply audio"; + setBtnIdle(); + isBusy = false; + } + }) + .catch((err) => { + status.textContent = "error: " + err.message; + setBtnIdle(); + isBusy = false; + }); + } + + function playPCM(pcm) { + const sampleRate = 16000; + const bitsPerSample = 16; + const channels = 1; + const dataLen = pcm.length; + const headerLen = 44; + const wav = new Uint8Array(headerLen + dataLen); + + const dv = (i, v) => { wav[i] = v & 255; wav[i+1] = (v>>8) & 255; wav[i+2] = (v>>16) & 255; wav[i+3] = (v>>24) & 255; }; + const sv = (i, v) => { wav[i] = v & 255; wav[i+1] = (v>>8) & 255; }; + wav[0] = 0x52; wav[1] = 0x49; wav[2] = 0x46; wav[3] = 0x46; + dv(4, 36 + dataLen); + wav[8] = 0x57; wav[9] = 0x41; wav[10] = 0x56; wav[11] = 0x45; + wav[12] = 0x66; wav[13] = 0x6d; wav[14] = 0x74; wav[15] = 0x20; + dv(16, 16); + sv(20, 1); + sv(22, channels); + dv(24, sampleRate); + dv(28, sampleRate * channels * bitsPerSample / 8); + sv(32, channels * bitsPerSample / 8); + sv(34, bitsPerSample); + wav[36] = 0x64; wav[37] = 0x61; wav[38] = 0x74; wav[39] = 0x61; + dv(40, dataLen); + wav.set(pcm, 44); + + const blob = new Blob([wav], { type: "audio/wav" }); + const audio = new Audio(); + audio.src = URL.createObjectURL(blob); + status.textContent = "playing..."; + audio.onended = () => { status.textContent = "ready"; setBtnIdle(); isBusy = false; }; + audio.play().catch(() => { isBusy = false; setBtnIdle(); }); + } + + function appendLog(msg, cls) { + const el = document.createElement("div"); + el.className = cls; + el.textContent = msg; + log.appendChild(el); + log.scrollTop = log.scrollHeight; + } + + btn.addEventListener("click", () => { + if (isBusy) return; + if (isRecording) { stopRecording(); } + else { startRecording(); } + }); + + // ---- ntfy WS subscribe: proactive nudges land in-app ------------------- + // The PWA connects straight to ntfy's WebSocket (mavweb serves only the URL, + // token included). ntfy streams one JSON object per frame; we care about + // event:"message". Reconnects with backoff — ntfy drops idle sockets and the + // phone sleeps. A missed nudge while disconnected is non-loss: sev>=3 also + // hit the native ntfy push, this is the in-app mirror, not the only channel. + function subscribeNtfy() { + fetch("/api/ntfy").then((r) => (r.status === 204 ? "" : r.text())).then((url) => { + if (!url) return; // not configured + if ("Notification" in window && Notification.permission === "default") { + Notification.requestPermission(); + } + connectNtfy(url, 1000); + }).catch(() => {}); // no ntfy config endpoint → stay voice-only + } + + function connectNtfy(url, backoff) { + let ws; + try { ws = new WebSocket(url); } catch (e) { scheduleReconnect(url, backoff); return; } + ws.onopen = () => { backoff = 1000; appendLog("nudges connected", ""); }; + ws.onmessage = (ev) => { + let m; + try { m = JSON.parse(ev.data); } catch (e) { return; } + if (m.event !== "message") return; // skip open/keepalive/poll_request + const text = (m.title ? m.title + ": " : "") + (m.message || ""); + appendLog(text, "reply"); + if ("Notification" in window && Notification.permission === "granted") { + new Notification(m.title || "maven", { body: m.message || "" }); + } + }; + ws.onclose = () => scheduleReconnect(url, backoff); + ws.onerror = () => { try { ws.close(); } catch (e) {} }; + } + + function scheduleReconnect(url, backoff) { + const next = Math.min(backoff * 2, 30000); // cap at 30s + setTimeout(() => connectNtfy(url, next), backoff); + } + + // ---- presence: page heartbeat ----------------------------------------- + // A surface you have open + alive is a weak presence signal (τ=4min). Ping + // every 30s; the fact's fresh timestamp is what the scorer reads. Fire-and- + // forget — a dropped ping just decays, non-loss. Disabled server-side (503) + // when mavweb has no -core; we ignore the failure and stop pinging isn't + // needed (the scorer just never sees the key). + function heartbeat() { + fetch("/api/signal?key=page_heartbeat", { method: "POST" }).catch(() => {}); + } + heartbeat(); + setInterval(heartbeat, 30000); + + status.textContent = "ready"; + testFetch(); + subscribeNtfy(); +})(); diff --git a/cmd/mavweb/static/index.html b/cmd/mavweb/static/index.html new file mode 100644 index 0000000..7f6c875 --- /dev/null +++ b/cmd/mavweb/static/index.html @@ -0,0 +1,63 @@ + + + + + + + + + + + + +
+

Maven Voice

+
tap & hold to speak
+ +
+
+
+ +
+ + + diff --git a/cmd/mavweb/static/manifest.json b/cmd/mavweb/static/manifest.json new file mode 100644 index 0000000..8b2ef07 --- /dev/null +++ b/cmd/mavweb/static/manifest.json @@ -0,0 +1,10 @@ +{ + "name": "Maven", + "short_name": "Maven", + "start_url": "/", + "display": "standalone", + "background_color": "#111", + "theme_color": "#00aaff", + "icons": [], + "description": "Voice client and dashboard for Maven" +} diff --git a/cmd/mavweb/static/service-worker.js b/cmd/mavweb/static/service-worker.js new file mode 100644 index 0000000..0e09577 --- /dev/null +++ b/cmd/mavweb/static/service-worker.js @@ -0,0 +1,16 @@ +const CACHE = "maven-v2"; +self.addEventListener("install", e => { + e.waitUntil(caches.open(CACHE).then(c => c.addAll(["/", "/manifest.json", "/dash"]))); + self.skipWaiting(); +}); +self.addEventListener("activate", e => { + e.waitUntil( + caches.keys().then(keys => Promise.all(keys.filter(k => k !== CACHE).map(k => caches.delete(k)))) + ); + clients.claim(); +}); +self.addEventListener("fetch", e => { + e.respondWith( + fetch(e.request).catch(() => caches.match(e.request)) + ); +}); diff --git a/internal/config/config.go b/internal/config/config.go index af235c8..18ca515 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -120,6 +120,13 @@ type VoiceConfig struct { Tts *TtsConfig `json:"tts,omitempty"` Embedder *EmbedderConfig `json:"embedder,omitempty"` + // RouterThreshold — the minimum confidence score for the intent classifier + // (stage 3 gate). Below this → clarify, don't guess. 0.0 means permissive + // (never clarify); 0.35 is a reasonable floor for the ONNX embedder. + // The HashEmbedder floor scores lexically and may need a lower value. + // Default 0.35 if unset. + RouterThreshold float64 `json:"router_threshold,omitempty"` + // Tools — the enabled act allowlist. Each is a spoken verb → argv the // executor runs (args from the utterance appended). Editing this set is the // human-only "enable" act (per spec); maven can't add to it from a request. @@ -215,6 +222,7 @@ const ( DefaultTickInterval = 60 * time.Second DefaultRepeatInterval = 5 * time.Minute DefaultAutotuneInterval = 10 * time.Minute + DefaultRouterThreshold = 0.35 ) // Load reads the JSON config at path and applies defaults. A missing file is diff --git a/internal/phraser/llmphraser.go b/internal/phraser/llmphraser.go index e16d6ea..2b56ccc 100644 --- a/internal/phraser/llmphraser.go +++ b/internal/phraser/llmphraser.go @@ -126,12 +126,15 @@ func (p *LLMPhraser) start(ctx context.Context) error { return nil case err := <-errCh: _ = cmd.Process.Kill() + _ = cmd.Wait() return fmt.Errorf("llm: server output: %w", err) case <-ctx.Done(): _ = cmd.Process.Kill() + _ = cmd.Wait() return ctx.Err() case <-time.After(60 * time.Second): _ = cmd.Process.Kill() + _ = cmd.Wait() return fmt.Errorf("llm: server did not start within 60s") } } @@ -140,6 +143,7 @@ func (p *LLMPhraser) Close() error { p.cancel() if p.cmd != nil && p.cmd.Process != nil { _ = p.cmd.Process.Kill() + _ = p.cmd.Wait() // reap the process — without Wait, the child becomes a zombie } p.wg.Wait() return nil diff --git a/internal/store/facts.go b/internal/store/facts.go index 66a23c4..f048770 100644 --- a/internal/store/facts.go +++ b/internal/store/facts.go @@ -36,7 +36,10 @@ func (s *Store) WriteFact(ctx context.Context, ts time.Time, kind FactKind, key, if err != nil { return 0, fmt.Errorf("write fact: %w", err) } - id, _ := res.LastInsertId() + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("last insert id: %w", err) + } return id, nil } @@ -156,7 +159,10 @@ func (s *Store) CorrectValue(ctx context.Context, key, source string, value any, if err := tx.Commit(); err != nil { return 0, err } - newID, _ = res.LastInsertId() + newID, err = res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("last insert id: %w", err) + } return newID, nil } diff --git a/internal/store/notes.go b/internal/store/notes.go index fa52fe6..b2b4179 100644 --- a/internal/store/notes.go +++ b/internal/store/notes.go @@ -28,7 +28,10 @@ func (s *Store) WriteNote(ctx context.Context, ts time.Time, text string, embedd if err != nil { return 0, fmt.Errorf("write note: %w", err) } - id, _ := res.LastInsertId() + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("last insert id: %w", err) + } return id, nil } diff --git a/internal/store/store.go b/internal/store/store.go index 5085403..d73b20f 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -70,7 +70,9 @@ func Open(ctx context.Context, path string) (*Store, error) { // single writer expected; the daemon is the only process touching the db. db.SetMaxOpenConns(1) if _, err := db.ExecContext(ctx, schemaSQL); err != nil { - _ = db.Close() + if closeErr := db.Close(); closeErr != nil { + return nil, fmt.Errorf("apply schema: %w (close: %v)", err, closeErr) + } return nil, fmt.Errorf("apply schema: %w", err) } return &Store{db: db}, nil