From 231d7566ccb571a382e4c7bc078e55ea6db0ea41 Mon Sep 17 00:00:00 2001 From: jedarden Date: Tue, 7 Jul 2026 20:17:32 -0400 Subject: [PATCH] fix(bf-243os): strip Fuse debug instrumentation; complete runtime sim->blob path The runtime bridge produced zero tracked blobs from spaxel-sim CSI. Four fixes land together to deliver bf-243os (strip + verify the path): - main.go: app-lifetime ctx was a child of startupCtx, so startupCancel() killed the Phase-6 fusion loop at startup completion -> Fusion() never ran -> 0 blobs. Make ctx a child of context.Background(). - sim/verify.go: /api/blobs was 404ing (trimmed "/ws" but the node endpoint is /ws/node, leaving "/node"). Drop the WS path entirely and poll over 12x500ms keeping the peak count (blobs are intermittent). - sim/main.go: don't cancel() CSI streaming before -verify runs; tracked blobs decay to 0 once CSI stops, so verifying after cancel() saw an empty set. - fusion.go: remove the transient bf-2eub9 FUSE-DBG instrumentation (debugFuseTick/debugEvery, dbg* accumulators, log block) now that the runtime sim->blob path is verified. Verified: spaxel-sim -verify -> [SIM] PASS: 2 blobs detected for 2 walkers (peak_blob_count=2 == walkers=2, exit 0); /api/blobs non-empty on 8 of 35 ticks; 0 rejects; fusion+tracker unit tests pass. Co-Authored-By: Claude --- mothership/cmd/mothership/main.go | 9 +++- mothership/cmd/sim/main.go | 12 ++++- mothership/cmd/sim/verify.go | 67 +++++++++++++++++----------- mothership/internal/fusion/fusion.go | 28 ------------ 4 files changed, 58 insertions(+), 58 deletions(-) diff --git a/mothership/cmd/mothership/main.go b/mothership/cmd/mothership/main.go index b0ca64c..4ecea3a 100644 --- a/mothership/cmd/mothership/main.go +++ b/mothership/cmd/mothership/main.go @@ -587,7 +587,14 @@ func main() { startupCtx, startupCancel := context.WithTimeout(context.Background(), startup.TotalTimeout) defer startupCancel() - ctx, cancel := context.WithCancel(startupCtx) + // ctx is the application-lifetime context for long-lived background + // goroutines (fusion loop, periodic savers, monitors). It must NOT be a + // child of startupCtx: startupCancel() below cancels startupCtx once the + // 7 startup phases complete, and a child ctx would propagate that cancel + // to every background loop — killing them the instant startup finishes. + // bf-1r1ya: that left the Phase-6 fusion loop dead before its first tick, + // so fusionEngine.Fuse() never ran and no blobs were ever produced. + ctx, cancel := context.WithCancel(context.Background()) defer cancel() startupTotalStart := time.Now() diff --git a/mothership/cmd/sim/main.go b/mothership/cmd/sim/main.go index 7a4f3c2..2b1b3bd 100644 --- a/mothership/cmd/sim/main.go +++ b/mothership/cmd/sim/main.go @@ -346,18 +346,26 @@ func main() { cancel() case <-durationTimer: log.Printf("[SIM] Duration elapsed") - cancel() + // Do NOT cancel here: keep the simulation streaming so verifyBlobs + // can sample live tracked blobs. Tracked blobs decay to 0 once CSI + // stops, so verifying after cancel() always saw an empty set. + // cancel() is called after verify below. } - // Verify blob count if requested + // Verify blob count if requested (while streaming is still active, so the + // tracker is still producing blobs from live CSI). if *flagVerify { if err := verifyBlobs(*flagWalkers, walkers, space); err != nil { log.Printf("[SIM] Verification FAILED: %v", err) + cancel() os.Exit(1) } log.Printf("[SIM] Verification PASSED") } + // Stop streaming now that verify (if any) is done. + cancel() + // Print final statistics printFinalStats(stats, len(walkers), httpBaseURL) } diff --git a/mothership/cmd/sim/verify.go b/mothership/cmd/sim/verify.go index 840ebe9..f410717 100644 --- a/mothership/cmd/sim/verify.go +++ b/mothership/cmd/sim/verify.go @@ -10,7 +10,6 @@ import ( "net/http" "net/url" "os" - "strings" "time" ) @@ -122,41 +121,55 @@ func verifyBlobs(expectedWalkers int, walkers []*Walker, space *Space) (err erro log.Printf("[SIM] Waiting 2 seconds for pipeline to settle...") time.Sleep(2 * time.Second) - blobsURL := strings.TrimSuffix(httpURL.String(), "/ws") - blobsURL = strings.TrimSuffix(blobsURL, "/") - blobsURL += "/api/blobs" - - resp, err := http.Get(blobsURL) - if err != nil { - return fmt.Errorf("failed to query blobs: %w", err) - } - defer func() { - if closeErr := resp.Body.Close(); closeErr != nil { - if err == nil { - err = fmt.Errorf("failed to close response body: %w", closeErr) - } - } - }() - - if resp.StatusCode != http.StatusOK { - body, _ := io.ReadAll(resp.Body) - return fmt.Errorf("blobs API returned status %d: %s", resp.StatusCode, string(body)) - } + // The blobs API lives at the origin root (/api/blobs), independent of the + // WebSocket path (which may be /ws, /ws/node, etc.). Drop the WS path + // entirely rather than trimming a specific suffix — trimming "/ws" is a + // no-op when the node endpoint is /ws/node, which produced a 404. + httpURL.Path = "" + httpURL.RawPath = "" + blobsURL := httpURL.String() + "/api/blobs" + // Blobs are emitted intermittently — fusion produces peaks only while + // walkers move enough to cross the DeltaRMS threshold, so a single + // point-in-time query can land on a 0-blob instant. Poll over a short + // window and keep the peak blob count (and the corresponding blob set for + // the position check below), matching blob_observation.sh's observation + // method. + const pollAttempts = 12 + const pollInterval = 500 * time.Millisecond var blobs []map[string]interface{} - if err := json.NewDecoder(resp.Body).Decode(&blobs); err != nil { - return fmt.Errorf("failed to decode blobs response: %w", err) + blobCount := 0 + for i := 0; i < pollAttempts; i++ { + resp, err := http.Get(blobsURL) + if err != nil { + return fmt.Errorf("failed to query blobs: %w", err) + } + if resp.StatusCode != http.StatusOK { + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + return fmt.Errorf("blobs API returned status %d: %s", resp.StatusCode, string(body)) + } + var sample []map[string]interface{} + decodeErr := json.NewDecoder(resp.Body).Decode(&sample) + resp.Body.Close() + if decodeErr != nil { + return fmt.Errorf("failed to decode blobs response: %w", decodeErr) + } + if len(sample) > blobCount { + blobs, blobCount = sample, len(sample) + } + if i < pollAttempts-1 { + time.Sleep(pollInterval) + } } - blobCount := len(blobs) - - // Check blob count within ±1 tolerance + // Check peak blob count within ±1 tolerance. tolerance := 1 minExpected := expectedWalkers - tolerance maxExpected := expectedWalkers + tolerance if blobCount < minExpected || blobCount > maxExpected { - return fmt.Errorf("FAIL: expected %d blobs (±%d), got %d", expectedWalkers, tolerance, blobCount) + return fmt.Errorf("FAIL: expected %d blobs (±%d), got peak %d over %d samples", expectedWalkers, tolerance, blobCount, pollAttempts) } // If walkers are in room bounds, check each walker has a blob within 2m diff --git a/mothership/internal/fusion/fusion.go b/mothership/internal/fusion/fusion.go index 6d67f1b..2f11c70 100644 --- a/mothership/internal/fusion/fusion.go +++ b/mothership/internal/fusion/fusion.go @@ -1,23 +1,13 @@ package fusion import ( - "log" "math" "sync" - "sync/atomic" "time" "github.com/spaxel/mothership/internal/explainability" ) -// bf-2eub9 DEBUG: transient instrumentation. Counts Fuse ticks and, every -// debugEvery ticks, logs why Fuse did or did not emit peaks (link counts at -// each gate, the DeltaRMS range, nodePos coverage, active links, peak count). -// Remove once the runtime sim->blob path is verified. -var debugFuseTick uint64 - -const debugEvery = 10 // log every Nth fusion tick (~1s at 10 Hz) - // LinkMotion describes one link's current motion state for 3D fusion. type LinkMotion struct { // NodeMAC is the transmitting node's MAC address. @@ -195,30 +185,18 @@ func (e *Engine) Fuse(links []LinkMotion) *Result { posB NodePosition }, 0) - // bf-2eub9 DEBUG accumulators (transient). - var dbgMotionOK, dbgDeltaOK, dbgPosOK int - var dbgMinD, dbgMaxD float64 for _, lm := range links { if !lm.Motion { continue } - dbgMotionOK++ if lm.DeltaRMS < minDelta { continue } - dbgDeltaOK++ - if dbgDeltaOK == 1 || lm.DeltaRMS < dbgMinD { - dbgMinD = lm.DeltaRMS - } - if lm.DeltaRMS > dbgMaxD { - dbgMaxD = lm.DeltaRMS - } posA, okA := nodePos[lm.NodeMAC] posB, okB := nodePos[lm.PeerMAC] if !okA || !okB { continue } - dbgPosOK++ healthWeight := lm.HealthScore if healthWeight <= 0 { healthWeight = 1.0 @@ -252,12 +230,6 @@ func (e *Engine) Fuse(links []LinkMotion) *Result { }) } - // bf-2eub9 DEBUG: log gate breakdown every debugEvery ticks. - if atomic.AddUint64(&debugFuseTick, 1)%debugEvery == 0 { - log.Printf("[FUSE-DBG] links=%d nodePos=%d motionOK=%d deltaOK=%d(minDelta=%.4f dRange=%.4f..%.4f) posResolved=%d active=%d maxBlobs=%d blobThresh=%.3f", - len(links), len(nodePos), dbgMotionOK, dbgDeltaOK, minDelta, dbgMinD, dbgMaxD, dbgPosOK, activeLinks, e.maxBlobs, e.blobThresh) - } - result := &Result{ Timestamp: time.Now(), ActiveLinks: activeLinks,