From 73441bcc6cb3b8458caa1b1978247e8d1c803ee2 Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 6 Aug 2026 23:33:26 +0400 Subject: [PATCH 1/2] one boot path for the workers and the API (V-639) run() wired the daemon twice and the lists drifted: seven workers started untracked on the unlock path, a shadowed WaitGroup hid the voice server, and nexus and getMCPServers were never set after a passkey unlock. newDaemonAPI and startBackground in cmd/mavend/boot.go are the one place both paths go through now. backgroundWorkers is the pure list behind the second, so the set is assertable without a running daemon. --- cmd/mavend/boot.go | 120 ++++++++++++++++++++++++ cmd/mavend/main.go | 136 +++++----------------------- docs/plans/25-the-two-boot-paths.md | 19 +++- 3 files changed, 162 insertions(+), 113 deletions(-) create mode 100644 cmd/mavend/boot.go diff --git a/cmd/mavend/boot.go b/cmd/mavend/boot.go new file mode 100644 index 0000000..7e754be --- /dev/null +++ b/cmd/mavend/boot.go @@ -0,0 +1,120 @@ +package main + +import ( + "context" + "errors" + "log" + "net" + "sync" + "time" + + "github.com/kami/maven/internal/event" + "github.com/kami/maven/internal/ipc" + "github.com/kami/maven/internal/store" +) + +// The two boot paths meet here. run() wires the daemon twice: once at boot +// when a key is in the environment, and once inside UnlockFn after a passkey +// assertion, minutes or days later. Listing the same wiring in both places is +// what let them drift — seven workers started untracked on the unlock path and +// two daemonAPI fields were never set there, silently, for as long as anyone +// had been cold-starting (V-639). +// +// So both paths call newDaemonAPI and startBackground and nothing else. A +// field or a worker added later reaches both paths or neither. + +// bootDeps is everything the two constructors below read. It is filled from +// the same variables on both paths, by depsNow in run(). +type bootDeps struct { + coreFor func() ipc.CoreAPI + tl *tickLoop + evBus *event.Bus + voiceW *voiceWiring + st *store.Store + factWorker *factEnrichmentWorker + evalWorker *memoryEvalWorker // nil ⇒ memory evaluation off (the default) + feedWkr *feedWorker // nil ⇒ no feed is read (the default) + crawlWkr *crawlWorker // nil ⇒ no page is watched (the default) +} + +// newDaemonAPI builds the real CoreAPI, with every field set. The unlock path +// used to leave nexus and getMCPServers nil, so after a cold start +// ResolveEntity refused with a nexus block configured and /tools rendered +// "not configured" with an mcp block configured. Empty is a wrong answer +// there, not a degraded one. +func newDaemonAPI(d bootDeps) *daemonAPI { + api := &daemonAPI{ + CoreAPI: d.coreFor(), + getTrace: d.tl.trace, + getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return d.tl.morningStatus(ctx, time.Now()) }, + getDayPlan: func(ctx context.Context) ipc.DayPlan { return d.tl.dayPlan(ctx, time.Now()) }, + getEvents: intakeEventsFn(d.evBus), + getDecisions: turnDecisionsFn(d.voiceW), + seedStore: seedStoreIfAllowed(d.st), + nexus: nexusOf(d.voiceW), + } + if d.voiceW != nil && d.voiceW.handler != nil { + api.chatFn = d.voiceW.handler.handleText + // And the reverse: the handler was wired with the bare store adapter, + // which cannot serve the day plan. See upgradeAPI. + d.voiceW.handler.upgradeAPI(api) + } + if d.voiceW != nil && d.voiceW.mcp != nil { + api.getMCPServers = d.voiceW.mcp.status + } + return api +} + +// namedWorker is one long-running goroutine. The name exists so the set is +// assertable from a test and readable in a log; nothing dispatches on it. +type namedWorker struct { + name string + run func(ctx context.Context) +} + +// backgroundWorkers lists what this deployment runs. It is pure — it starts +// nothing — so a test can compare the set the two paths would start without +// standing a daemon up. +func backgroundWorkers(d bootDeps) []namedWorker { + var ws []namedWorker + if d.voiceW != nil && d.voiceW.server != nil { + ws = append(ws, namedWorker{"voice", func(context.Context) { + if err := d.voiceW.server.Serve(); err != nil && !errors.Is(err, net.ErrClosed) { + log.Printf("voice serve: %v", err) + } + }}) + } + ws = append(ws, + namedWorker{"tick", d.tl.run}, + namedWorker{"fact-enrichment", d.factWorker.run}, + ) + if d.evalWorker != nil { + ws = append(ws, namedWorker{"memory-eval", d.evalWorker.run}) + } + if d.feedWkr != nil { + ws = append(ws, namedWorker{"feed", d.feedWkr.run}) + } + if d.crawlWkr != nil { + ws = append(ws, namedWorker{"crawl", d.crawlWkr.run}) + } + if d.voiceW != nil && d.voiceW.mcp != nil { + ws = append(ws, namedWorker{"mcp", d.voiceW.mcp.run}) + } + if d.voiceW != nil && d.voiceW.home != nil { + ws = append(ws, namedWorker{"home", d.voiceW.home.run}) + } + return ws +} + +// startBackground starts every worker through goWorker, so waitWorkers can +// wait for it at shutdown. A worker started as a bare `go func()` is the +// shutdown bug documented at the end of run(): run() never returns, the +// deferred Close never seals the database, and the ciphertext goes stale. +func startBackground(ctx context.Context, wg *sync.WaitGroup, d bootDeps) { + for _, w := range backgroundWorkers(d) { + goWorker(wg, func() { w.run(ctx) }) + } + if d.voiceW != nil && d.voiceW.server != nil { + log.Printf("mavend: voice listening on %s", d.voiceW.server.Addr()) + } +} diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index 391bd71..75c4f76 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -252,6 +252,23 @@ func run(args []string) error { // envelope per successful intake write. coreFor := func() ipc.CoreAPI { return newIntakeAPI(ipc.NewStoreAPI(st), evBus, time.Now) } + // depsNow reads whatever the current path has wired. Both boot paths build + // the CoreAPI and start the workers from this one value, so neither can + // hold a field the other misses. See cmd/mavend/boot.go. + depsNow := func() bootDeps { + return bootDeps{ + coreFor: coreFor, + tl: tl, + evBus: evBus, + voiceW: voiceW, + st: st, + factWorker: factWorker, + evalWorker: evalWorker, + feedWkr: feedWkr, + crawlWkr: crawlWkr, + } + } + if !locked { rules = wireRules(cfg) gatherer = wireGatherer(st, cfg, rules) @@ -284,26 +301,7 @@ func run(args []string) error { feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg) crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg) - coreAPI = &daemonAPI{ - CoreAPI: coreFor(), - getTrace: tl.trace, - getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return tl.morningStatus(ctx, time.Now()) }, - getDayPlan: func(ctx context.Context) ipc.DayPlan { return tl.dayPlan(ctx, time.Now()) }, - getEvents: intakeEventsFn(evBus), - getDecisions: turnDecisionsFn(voiceW), - seedStore: seedStoreIfAllowed(st), - nexus: nexusOf(voiceW), - } - if voiceW != nil && voiceW.handler != nil { - api := coreAPI.(*daemonAPI) - api.chatFn = voiceW.handler.handleText - // And the reverse: the handler was wired with the bare store - // adapter, which cannot serve the day plan. See upgradeAPI. - voiceW.handler.upgradeAPI(api) - } - if voiceW != nil && voiceW.mcp != nil { - coreAPI.(*daemonAPI).getMCPServers = voiceW.mcp.status - } + coreAPI = newDaemonAPI(depsNow()) } else { // locked mode: no real store yet, so there's no meaningful CoreAPI to // serve. srv.Check below is the actual guard — every CoreAPI call is @@ -497,19 +495,7 @@ func run(args []string) error { crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg) // Swap the CoreAPI from the locked placeholder to the real store adapter. - newAPI := &daemonAPI{ - CoreAPI: coreFor(), - getTrace: tl.trace, - getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return tl.morningStatus(ctx, time.Now()) }, - getDayPlan: func(ctx context.Context) ipc.DayPlan { return tl.dayPlan(ctx, time.Now()) }, - getEvents: intakeEventsFn(evBus), - getDecisions: turnDecisionsFn(voiceW), - seedStore: seedStoreIfAllowed(st), - } - if voiceW != nil && voiceW.handler != nil { - newAPI.chatFn = voiceW.handler.handleText - voiceW.handler.upgradeAPI(newAPI) - } + newAPI := newDaemonAPI(depsNow()) srv.SetAPI(newAPI) srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check wireMailIntake(srv, st, phr, cfg, evBus) @@ -524,59 +510,10 @@ func run(args []string) error { // block, so no wire path takes a voiceprint on a default box. wireSpeaker(srv, st, cfg) - // Start voice server. - if voiceW != nil { - var wg sync.WaitGroup - wg.Add(1) - go func() { - defer wg.Done() - if err := voiceW.server.Serve(); err != nil && !errors.Is(err, net.ErrClosed) { - log.Printf("voice serve: %v", err) - } - }() - log.Printf("mavend: voice listening on %s", voiceW.server.Addr()) - } - - // Start tick loop. - go func() { - tl.run(ctx) - }() - - // Start fact-entity enrichment worker. - go func() { - factWorker.run(ctx) - }() - - // Start background memory evaluation (nil unless configured). - if evalWorker != nil { - go func() { - evalWorker.run(ctx) - }() - } - - // Start feed reading (nil unless configured). - if feedWkr != nil { - go func() { - feedWkr.run(ctx) - }() - } - - // Start the watched-page crawls (nil unless configured). - if crawlWkr != nil { - go func() { - crawlWkr.run(ctx) - }() - } - - // Keep MCP connections alive (nil unless configured). - if voiceW != nil && voiceW.mcp != nil { - go voiceW.mcp.run(ctx) - } - - // Re-enumerate the house for new devices (nil unless configured). - if voiceW != nil && voiceW.home != nil { - go voiceW.home.run(ctx) - } + // The voice server and every background worker, on the outer wg + // so shutdown waits for them. This used to be nine bare + // `go func()` calls and a shadowed WaitGroup (V-639). + startBackground(ctx, &wg, depsNow()) dl.unlock(st) log.Printf("mavend: unlocked via passkey assertion") @@ -591,33 +528,8 @@ func run(args []string) error { }) log.Printf("mavend: ipc listening on %s", srv.Path()) - if !locked && voiceW != nil { - goWorker(&wg, func() { - if err := voiceW.server.Serve(); err != nil && !errors.Is(err, net.ErrClosed) { - log.Printf("voice serve: %v", err) - } - }) - log.Printf("mavend: voice listening on %s", voiceW.server.Addr()) - } - if !locked { - goWorker(&wg, func() { tl.run(ctx) }) - goWorker(&wg, func() { factWorker.run(ctx) }) - if evalWorker != nil { - goWorker(&wg, func() { evalWorker.run(ctx) }) - } - if feedWkr != nil { - goWorker(&wg, func() { feedWkr.run(ctx) }) - } - if crawlWkr != nil { - goWorker(&wg, func() { crawlWkr.run(ctx) }) - } - if voiceW != nil && voiceW.mcp != nil { - goWorker(&wg, func() { voiceW.mcp.run(ctx) }) - } - if voiceW != nil && voiceW.home != nil { - goWorker(&wg, func() { voiceW.home.run(ctx) }) - } + startBackground(ctx, &wg, depsNow()) } <-ctx.Done() diff --git a/docs/plans/25-the-two-boot-paths.md b/docs/plans/25-the-two-boot-paths.md index 72d2b45..cd99d91 100644 --- a/docs/plans/25-the-two-boot-paths.md +++ b/docs/plans/25-the-two-boot-paths.md @@ -1,9 +1,26 @@ # The two boot paths have drifted -Last verified: 06-08-2026 @ 06c1cf2 +Last verified: 06-08-2026 @ 69d0f5e V-639. Reads with `docs/operations.md`. +## What landed + +`cmd/mavend/boot.go`. `newDaemonAPI(deps)` builds the CoreAPI with every field +set, and `startBackground(ctx, &wg, deps)` starts the voice server and every +worker through `goWorker`. `backgroundWorkers(deps)` is the pure list behind it, +so a test can compare the set without standing a daemon up. Both paths in +`run()` now read `coreAPI = newDaemonAPI(depsNow())` and one +`startBackground(...)`, where `depsNow` reads whatever the current path wired. + +The shadowed `wg` is gone. Four tests in `cmd/mavend/boot_test.go`. Every +`daemonAPI` field is set on a fully wired deployment. The handler gets the API +it was built with. The worker set is asserted by name, at the full set and at +the floor. + +Still by hand: unlock a locked box by passkey, ask something that needs Nexus, +and check `/tools` lists the MCP servers. + ## What is wrong `run()` in `cmd/mavend/main.go` brings the daemon up two ways. A box with a key in the -- 2.52.0 From 896c32882a96b61980b015e070c2fff00982e48a Mon Sep 17 00:00:00 2001 From: claude Date: Thu, 6 Aug 2026 23:33:26 +0400 Subject: [PATCH 2/2] assert the two paths wire the same set (V-639) Without this it drifts again on the next wiring line: every daemonAPI field set on a fully wired deployment, and the worker set by name at the full set and at the floor. --- cmd/mavend/boot_test.go | 96 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 96 insertions(+) create mode 100644 cmd/mavend/boot_test.go diff --git a/cmd/mavend/boot_test.go b/cmd/mavend/boot_test.go new file mode 100644 index 0000000..47276f6 --- /dev/null +++ b/cmd/mavend/boot_test.go @@ -0,0 +1,96 @@ +package main + +import ( + "reflect" + "testing" + + "github.com/kami/maven/internal/decision" + "github.com/kami/maven/internal/event" + "github.com/kami/maven/internal/ipc" + "github.com/kami/maven/internal/store" + "github.com/kami/maven/internal/voice" +) + +// fullDeps — a deployment with every optional piece present. Nothing here is +// run: newDaemonAPI takes method values and backgroundWorkers is pure, so +// zero-value wirings are enough to say what WOULD be started. +func fullDeps() bootDeps { + h := &reactiveHandler{ + ecosystem: &ecosystemWiring{nexus: &nexusClient{}}, + decisions: decision.NewRing(), + } + return bootDeps{ + coreFor: func() ipc.CoreAPI { return ipc.UnimplementedCoreAPI{} }, + tl: &tickLoop{}, + evBus: event.NewBus(4), + st: &store.Store{}, + factWorker: &factEnrichmentWorker{}, + evalWorker: &memoryEvalWorker{}, + feedWkr: &feedWorker{}, + crawlWkr: &crawlWorker{}, + voiceW: &voiceWiring{ + server: &voice.Server{}, + handler: h, + mcp: &mcpWiring{}, + home: &homeWiring{}, + }, + } +} + +// The unlock path used to build its own daemonAPI literal and leave nexus and +// getMCPServers nil (V-639). Both paths call newDaemonAPI now, so the drift +// that can still happen is a field added to the struct and not to the +// constructor. This catches that one, by name. +func TestNewDaemonAPISetsEveryField(t *testing.T) { + prev := allowSeedOnStart + allowSeedOnStart = true + defer func() { allowSeedOnStart = prev }() + + api := newDaemonAPI(fullDeps()) + v := reflect.ValueOf(*api) + for i := range v.NumField() { + if v.Field(i).IsZero() { + t.Errorf("newDaemonAPI left %s unset — a fully wired deployment must fill every field", v.Type().Field(i).Name) + } + } +} + +// The handler is wired with the bare store adapter and cannot serve the day +// plan until upgradeAPI hands it the real one. The unlocked path did that and +// the unlock path did it too; keep it a property of the constructor. +func TestNewDaemonAPIUpgradesTheHandler(t *testing.T) { + d := fullDeps() + api := newDaemonAPI(d) + if d.voiceW.handler.api != ipc.CoreAPI(api) { + t.Fatal("newDaemonAPI did not hand the handler the API it built") + } +} + +// Every worker the daemon runs goes through startBackground, so shutdown can +// wait for it. The unlock path used to start seven of these as bare +// `go func()` under a shadowed WaitGroup. +func TestBackgroundWorkersFullSet(t *testing.T) { + want := []string{"voice", "tick", "fact-enrichment", "memory-eval", "feed", "crawl", "mcp", "home"} + var got []string + for _, w := range backgroundWorkers(fullDeps()) { + got = append(got, w.name) + } + if !reflect.DeepEqual(got, want) { + t.Errorf("workers = %v, want %v", got, want) + } +} + +// A default box configures none of the optional blocks. Two workers always run +// and the rest stay dark, rather than a nil run being scheduled. +func TestBackgroundWorkersFloor(t *testing.T) { + d := fullDeps() + d.evalWorker, d.feedWkr, d.crawlWkr, d.voiceW = nil, nil, nil, nil + want := []string{"tick", "fact-enrichment"} + var got []string + for _, w := range backgroundWorkers(d) { + got = append(got, w.name) + } + if !reflect.DeepEqual(got, want) { + t.Errorf("workers = %v, want %v", got, want) + } +} -- 2.52.0