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.
This commit is contained in:
@@ -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())
|
||||||
|
}
|
||||||
|
}
|
||||||
+24
-112
@@ -252,6 +252,23 @@ func run(args []string) error {
|
|||||||
// envelope per successful intake write.
|
// envelope per successful intake write.
|
||||||
coreFor := func() ipc.CoreAPI { return newIntakeAPI(ipc.NewStoreAPI(st), evBus, time.Now) }
|
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 {
|
if !locked {
|
||||||
rules = wireRules(cfg)
|
rules = wireRules(cfg)
|
||||||
gatherer = wireGatherer(st, cfg, rules)
|
gatherer = wireGatherer(st, cfg, rules)
|
||||||
@@ -284,26 +301,7 @@ func run(args []string) error {
|
|||||||
feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg)
|
feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg)
|
||||||
crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
|
crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
|
||||||
|
|
||||||
coreAPI = &daemonAPI{
|
coreAPI = newDaemonAPI(depsNow())
|
||||||
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
|
|
||||||
}
|
|
||||||
} else {
|
} else {
|
||||||
// locked mode: no real store yet, so there's no meaningful CoreAPI to
|
// 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
|
// 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)
|
crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
|
||||||
|
|
||||||
// Swap the CoreAPI from the locked placeholder to the real store adapter.
|
// Swap the CoreAPI from the locked placeholder to the real store adapter.
|
||||||
newAPI := &daemonAPI{
|
newAPI := newDaemonAPI(depsNow())
|
||||||
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)
|
|
||||||
}
|
|
||||||
srv.SetAPI(newAPI)
|
srv.SetAPI(newAPI)
|
||||||
srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check
|
srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check
|
||||||
wireMailIntake(srv, st, phr, cfg, evBus)
|
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.
|
// block, so no wire path takes a voiceprint on a default box.
|
||||||
wireSpeaker(srv, st, cfg)
|
wireSpeaker(srv, st, cfg)
|
||||||
|
|
||||||
// Start voice server.
|
// The voice server and every background worker, on the outer wg
|
||||||
if voiceW != nil {
|
// so shutdown waits for them. This used to be nine bare
|
||||||
var wg sync.WaitGroup
|
// `go func()` calls and a shadowed WaitGroup (V-639).
|
||||||
wg.Add(1)
|
startBackground(ctx, &wg, depsNow())
|
||||||
go func() {
|
|
||||||
defer wg.Done()
|
|
||||||
if err := voiceW.server.Serve(); err != nil && !errors.Is(err, net.ErrClosed) {
|
|
||||||
log.Printf("voice serve: %v", err)
|
|
||||||
}
|
|
||||||
}()
|
|
||||||
log.Printf("mavend: voice listening on %s", voiceW.server.Addr())
|
|
||||||
}
|
|
||||||
|
|
||||||
// Start tick loop.
|
|
||||||
go func() {
|
|
||||||
tl.run(ctx)
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Start fact-entity enrichment worker.
|
|
||||||
go func() {
|
|
||||||
factWorker.run(ctx)
|
|
||||||
}()
|
|
||||||
|
|
||||||
// Start background memory evaluation (nil unless configured).
|
|
||||||
if evalWorker != nil {
|
|
||||||
go func() {
|
|
||||||
evalWorker.run(ctx)
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Start feed reading (nil unless configured).
|
|
||||||
if feedWkr != nil {
|
|
||||||
go func() {
|
|
||||||
feedWkr.run(ctx)
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Start the watched-page crawls (nil unless configured).
|
|
||||||
if crawlWkr != nil {
|
|
||||||
go func() {
|
|
||||||
crawlWkr.run(ctx)
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
// Keep MCP connections alive (nil unless configured).
|
|
||||||
if voiceW != nil && voiceW.mcp != nil {
|
|
||||||
go voiceW.mcp.run(ctx)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Re-enumerate the house for new devices (nil unless configured).
|
|
||||||
if voiceW != nil && voiceW.home != nil {
|
|
||||||
go voiceW.home.run(ctx)
|
|
||||||
}
|
|
||||||
|
|
||||||
dl.unlock(st)
|
dl.unlock(st)
|
||||||
log.Printf("mavend: unlocked via passkey assertion")
|
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())
|
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 {
|
if !locked {
|
||||||
goWorker(&wg, func() { tl.run(ctx) })
|
startBackground(ctx, &wg, depsNow())
|
||||||
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) })
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
<-ctx.Done()
|
<-ctx.Done()
|
||||||
|
|||||||
@@ -1,9 +1,26 @@
|
|||||||
# The two boot paths have drifted
|
# 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`.
|
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
|
## What is wrong
|
||||||
|
|
||||||
`run()` in `cmd/mavend/main.go` brings the daemon up two ways. A box with a key in the
|
`run()` in `cmd/mavend/main.go` brings the daemon up two ways. A box with a key in the
|
||||||
|
|||||||
Reference in New Issue
Block a user