From 63cbb08ea01ffd2959f5081a842d87502f13941b Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Fri, 25 Sep 2026 16:37:56 +0300 Subject: [PATCH] feat(rover): add drive profiles and leased remote command channel --- apps/node-agent/internal/node/board_layout.go | 151 +++++++++++ .../internal/node/board_layout_test.go | 56 ++++ .../node-agent/internal/node/rover_channel.go | 249 ++++++++++++++++++ .../internal/node/rover_diagnostics.go | 61 +++++ .../internal/node/rover_diagnostics_test.go | 63 +++++ .../internal/node/rover_stream_state.go | 199 ++++++++++++++ .../internal/node/rover_stream_state_test.go | 235 +++++++++++++++++ packages/rover-control/README.md | 107 ++++++++ packages/rover-control/src/authority.ts | 134 ++++++++++ packages/rover-control/src/profile.ts | 72 +++++ packages/sensor-ui/src/BoardSections.tsx | 23 ++ packages/sensor-ui/src/boardLayout.ts | 66 +++++ src/k1link/fleet/board_layout.py | 45 ++++ src/k1link/fleet/rover_control.py | 135 ++++++++++ src/k1link/web/rover_api.py | 40 +++ tests/fleet/test_board_layout.py | 42 +++ tests/fleet/test_rover_control.py | 115 ++++++++ tests/fleet/test_rover_stream_transport.py | 52 ++++ 18 files changed, 1845 insertions(+) create mode 100644 apps/node-agent/internal/node/board_layout.go create mode 100644 apps/node-agent/internal/node/board_layout_test.go create mode 100644 apps/node-agent/internal/node/rover_channel.go create mode 100644 apps/node-agent/internal/node/rover_diagnostics.go create mode 100644 apps/node-agent/internal/node/rover_diagnostics_test.go create mode 100644 apps/node-agent/internal/node/rover_stream_state.go create mode 100644 apps/node-agent/internal/node/rover_stream_state_test.go create mode 100644 packages/rover-control/README.md create mode 100644 packages/rover-control/src/authority.ts create mode 100644 packages/rover-control/src/profile.ts create mode 100644 packages/sensor-ui/src/BoardSections.tsx create mode 100644 packages/sensor-ui/src/boardLayout.ts create mode 100644 src/k1link/fleet/board_layout.py create mode 100644 src/k1link/fleet/rover_control.py create mode 100644 src/k1link/web/rover_api.py create mode 100644 tests/fleet/test_board_layout.py create mode 100644 tests/fleet/test_rover_control.py create mode 100644 tests/fleet/test_rover_stream_transport.py diff --git a/apps/node-agent/internal/node/board_layout.go b/apps/node-agent/internal/node/board_layout.go new file mode 100644 index 0000000..b2d7aaa --- /dev/null +++ b/apps/node-agent/internal/node/board_layout.go @@ -0,0 +1,151 @@ +package node + +import ( + "encoding/json" + "errors" + "io" + "net/http" + "os" + "path/filepath" +) + +const boardLayoutSchema = "missioncore.board-layout/v1" + +var boardSectionIDs = []string{"computer", "settings", "devices"} + +type BoardLayout struct { + Schema string `json:"schema"` + Revision int64 `json:"revision"` + OpenSections []string `json:"open_sections"` +} + +func validBoardSection(id string) bool { + for _, section := range boardSectionIDs { + if id == section { + return true + } + } + return false +} + +func (p *PresentationStore) readBoardLayout() (BoardLayout, error) { + value := BoardLayout{Schema: boardLayoutSchema, OpenSections: append([]string{}, boardSectionIDs...)} + file, err := os.Open(filepath.Join(p.dir, "board-layout.json")) + if errors.Is(err, os.ErrNotExist) { + return value, nil + } + if err != nil { + return BoardLayout{}, err + } + defer file.Close() + d := json.NewDecoder(io.LimitReader(file, 4097)) + d.DisallowUnknownFields() + value = BoardLayout{} + if d.Decode(&value) != nil || d.Decode(new(any)) != io.EOF || value.Schema != boardLayoutSchema || value.Revision < 0 || value.Revision >= 1<<53-1 || value.OpenSections == nil { + return BoardLayout{}, errors.New("invalid board layout") + } + seen := map[string]bool{} + for _, id := range value.OpenSections { + if !validBoardSection(id) || seen[id] { + return BoardLayout{}, errors.New("invalid board section") + } + seen[id] = true + } + return value, nil +} + +func (p *PresentationStore) saveBoardLayout(value BoardLayout) error { + data, err := json.MarshalIndent(value, "", " ") + if err != nil { + return err + } + file, err := os.CreateTemp(p.dir, ".board-layout-*") + if err != nil { + return err + } + defer os.Remove(file.Name()) + if _, err = file.Write(append(data, '\n')); err == nil { + err = file.Sync() + } + closeErr := file.Close() + if err != nil { + return err + } + if closeErr != nil { + return closeErr + } + if err = os.Rename(file.Name(), filepath.Join(p.dir, "board-layout.json")); err != nil { + return err + } + dir, err := os.Open(p.dir) + if err != nil { + return err + } + defer dir.Close() + return dir.Sync() +} + +func (s *Server) boardLayoutRoutes(mux *http.ServeMux) { + p := s.Presentation + mux.HandleFunc("GET /api/presentation/board-layout", func(w http.ResponseWriter, r *http.Request) { + if !s.authorized(w, r) { + return + } + p.mu.Lock() + defer p.mu.Unlock() + value, err := p.readBoardLayout() + if err != nil { + reply(w, 500, map[string]string{"error": "Не удалось прочитать раскладку борта"}) + return + } + reply(w, 200, value) + }) + mux.HandleFunc("PATCH /api/presentation/board-layout", func(w http.ResponseWriter, r *http.Request) { + if !s.authorized(w, r) { + return + } + if r.Header.Get("Content-Type") != "application/json" { + reply(w, 415, map[string]string{"error": "Ожидался JSON"}) + return + } + var change struct { + Section string `json:"section"` + Open *bool `json:"open"` + } + d := json.NewDecoder(http.MaxBytesReader(w, r.Body, 1024)) + d.DisallowUnknownFields() + if d.Decode(&change) != nil || d.Decode(new(any)) != io.EOF || change.Open == nil || !validBoardSection(change.Section) { + reply(w, 400, map[string]string{"error": "Некорректная раскладка борта"}) + return + } + p.mu.Lock() + defer p.mu.Unlock() + value, err := p.readBoardLayout() + if err != nil { + reply(w, 500, map[string]string{"error": "Не удалось прочитать раскладку борта"}) + return + } + next := []string{} + for _, id := range boardSectionIDs { + open := false + for _, saved := range value.OpenSections { + if saved == id { + open = true + } + } + if id == change.Section { + open = *change.Open + } + if open { + next = append(next, id) + } + } + value.OpenSections = next + value.Revision++ + if p.saveBoardLayout(value) != nil { + reply(w, 500, map[string]string{"error": "Не удалось сохранить раскладку борта"}) + return + } + reply(w, 200, value) + }) +} diff --git a/apps/node-agent/internal/node/board_layout_test.go b/apps/node-agent/internal/node/board_layout_test.go new file mode 100644 index 0000000..60e5f6d --- /dev/null +++ b/apps/node-agent/internal/node/board_layout_test.go @@ -0,0 +1,56 @@ +package node + +import ( + "os" + "path/filepath" + "sync" + "testing" +) + +func TestBoardLayoutPersistsEmptyAndConcurrentSectionChanges(t *testing.T) { + s, cookie := presentationServer(t) + var wg sync.WaitGroup + for _, id := range boardSectionIDs { + wg.Add(1) + go func(id string) { + defer wg.Done() + w := call(s, "PATCH", "/api/presentation/board-layout", `{"section":"`+id+`","open":false}`, cookie) + if w.Code != 200 { + t.Error(w.Code, w.Body.String()) + } + }(id) + } + wg.Wait() + value, err := NewPresentationStore(s.Presentation.dir).readBoardLayout() + if err != nil || value.Revision != 3 || len(value.OpenSections) != 0 || value.OpenSections == nil { + t.Fatal(value, err) + } + info, err := os.Stat(filepath.Join(s.Presentation.dir, "board-layout.json")) + if err != nil || info.Mode().Perm() != 0600 { + t.Fatal(info, err) + } + if w := call(s, "GET", "/api/presentation/board-layout", "", nil); w.Code != 401 { + t.Fatal(w.Code) + } +} + +func TestBoardLayoutRejectsInvalidAndPreservesCorruptFile(t *testing.T) { + s, cookie := presentationServer(t) + for _, body := range []string{`{"section":"motor","open":true}`, `{"section":"computer"}`, `{"section":"computer","open":"true"}`, `{"section":"computer","open":true,"extra":0}`} { + if w := call(s, "PATCH", "/api/presentation/board-layout", body, cookie); w.Code != 400 { + t.Fatal(w.Code, w.Body.String()) + } + } + path := filepath.Join(s.Presentation.dir, "board-layout.json") + raw := []byte(`{"schema":"missioncore.board-layout/v1","revision":0,"open_sections":["motor"]}`) + if err := os.WriteFile(path, raw, 0600); err != nil { + t.Fatal(err) + } + if w := call(s, "PATCH", "/api/presentation/board-layout", `{"section":"computer","open":false}`, cookie); w.Code != 500 { + t.Fatal(w.Code) + } + got, _ := os.ReadFile(path) + if string(got) != string(raw) { + t.Fatal("Corrupt file replaced") + } +} diff --git a/apps/node-agent/internal/node/rover_channel.go b/apps/node-agent/internal/node/rover_channel.go new file mode 100644 index 0000000..329b6e2 --- /dev/null +++ b/apps/node-agent/internal/node/rover_channel.go @@ -0,0 +1,249 @@ +package node + +// Commands stream independently of telemetry. Only the local 20 Hz loop may +// deliver the latest unexpired intent to the single VESC owner. +import ( + "bufio" + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "errors" + "io" + "log" + "net" + "net/http" + "sync" + "time" +) + +func (p *Pairing) roverBinding() *CoreBinding { + p.mu.Lock() + defer p.mu.Unlock() + if p.state.Phase != "paired" || p.state.Binding == nil { + return nil + } + b := *p.state.Binding + return &b +} +func sameRoverBinding(a, b *CoreBinding) bool { + return a != nil && b != nil && a.BindingID == b.BindingID && a.Endpoint == b.Endpoint && a.ClientPEM == b.ClientPEM && a.EndpointRevision == b.EndpointRevision +} +func roverWait(ctx context.Context, delay time.Duration) bool { + if delay < 0 { + delay = 0 + } + select { + case <-ctx.Done(): + return false + case <-time.After(delay): + return true + } +} +func (p *Pairing) roverChannel(ctx context.Context) { + for ctx.Err() == nil { + binding := p.roverBinding() + if binding != nil && p.Sensors != nil { + p.runRoverChannel(ctx, *binding) + } + if !roverWait(ctx, time.Second) { + return + } + } +} +func (p *Pairing) runRoverChannel(parent context.Context, b CoreBinding) { + config, err := bindingTLS(b, p.store.pairingKey()) + if err != nil { + return + } + transport := func() *http.Transport { + return &http.Transport{TLSClientConfig: config, Proxy: nil, + MaxConnsPerHost: 1, MaxIdleConnsPerHost: 1, IdleConnTimeout: 10 * time.Second, + TLSHandshakeTimeout: time.Second, ResponseHeaderTimeout: 2 * time.Second, + DialContext: (&net.Dialer{Timeout: time.Second}).DialContext} + } + telemetry := &http.Client{Transport: transport(), Timeout: 300 * time.Millisecond, CheckRedirect: func(*http.Request, []*http.Request) error { return errors.New("redirect forbidden") }} + stream := &http.Client{Transport: transport(), CheckRedirect: telemetry.CheckRedirect} + defer telemetry.CloseIdleConnections() + defer stream.CloseIdleConnections() + random := make([]byte, 16) + if _, err = rand.Read(random); err != nil { + return + } + relay := hex.EncodeToString(random) + id, _ := p.store.Public() + base := map[string]any{"schema": PairSchema, "node_id": id, "binding_id": b.BindingID, + "core_endpoint": b.Endpoint, "endpoint_revision": b.EndpointRevision, "relay_id": relay} + state := newRoverStreamState(time.Now()) + ctx, cancel := context.WithCancel(parent) + var workers sync.WaitGroup + defer func() { cancel(); workers.Wait() }() + workers.Add(2) + go func() { defer workers.Done(); roverTelemetry(ctx, telemetry, b.Endpoint, base, state) }() + go func() { defer workers.Done(); roverCommands(ctx, stream, b.Endpoint, base, state) }() + var model *sensorModel + for i := range sensorModels { + if sensorModels[i].ID == "vesc.controller" { + model = &sensorModels[i] + break + } + } + if model == nil { + return + } + var diagnostic roverDiagnostics + for ctx.Err() == nil && sameRoverBinding(&b, p.roverBinding()) { + select { + case <-state.wake: + default: + } + started := time.Now() + watch, command, stop := state.delivery(started) + frame := roverFrame{} + if command != nil { + frame.Session, _ = command["id"].(string) + frame.Sequence, _ = command["sequence"].(float64) + frame.TTLMS, _ = command["ttl_ms"].(float64) + } + call, done := context.WithTimeout(ctx, 25*time.Millisecond) + result, driverErr := p.Sensors.modelDriver(call, model, "/remote", map[string]any{"watch": watch, "command": command, "relay_id": relay}) + done() + frame.DriverMS = time.Since(started).Milliseconds() + if driverErr != nil { + state.stopStream() + frame.Stage = "driver_transport" + state.delivered(map[string]any{}, 0) + } else { + state.delivered(result, stop) + frame.State, _ = result["state"].(string) + } + diagnostic.record(started, frame) + // New intent and stop signals bypass the periodic watchdog tick. The + // buffered wake channel coalesces arrivals; it never queues commands. + timer := time.NewTimer(max(0, 50*time.Millisecond-time.Since(started))) + select { + case <-ctx.Done(): + case <-state.wake: + case <-timer.C: + } + timer.Stop() + } + // Binding changes and channel shutdown explicitly retire any current intent. + call, done := context.WithTimeout(context.Background(), 25*time.Millisecond) + _, _ = p.Sensors.modelDriver(call, model, "/remote", map[string]any{"watch": false, "command": nil, "relay_id": relay}) + done() +} +func roverTelemetry(ctx context.Context, client *http.Client, endpoint string, base map[string]any, state *roverStreamState) { + var diagnostic roverDiagnostics + for ctx.Err() == nil { + started := time.Now() + snapshot, _ := state.view() + payload := make(map[string]any, len(base)+1) + for k, v := range base { + payload[k] = v + } + payload["rover"] = snapshot + raw, _ := json.Marshal(payload) + req, _ := http.NewRequestWithContext(ctx, "POST", endpoint+"/v1/node/rover", bytes.NewReader(raw)) + req.Header.Set("Content-Type", "application/json") + frame := roverFrame{} + frame.State, _ = snapshot["state"].(string) + if frame.State == "preparing" || frame.State == "ready" || frame.State == "driving" || frame.State == "stopping" { + frame.Session, _ = snapshot["session_id"].(string) + } + response, err := client.Do(req) + if err != nil { + frame.Stage = "telemetry_transport" + } else { + frame.Status = response.StatusCode + var out roverStreamReply + readErr := json.NewDecoder(io.LimitReader(response.Body, 32768)).Decode(&out) + response.Body.Close() + if readErr != nil || response.StatusCode != 200 { + frame.Stage = "telemetry_response" + } else if !state.sampleClock(out.Clock, started, time.Now()) { + frame.Stage = "clock_invalid" + state.stopStream() + } else { + state.setWatch(out.Watch) + } + } + frame.CoreMS = time.Since(started).Milliseconds() + diagnostic.record(started, frame) + _, watch := state.view() + delay := 500 * time.Millisecond + if watch { + delay = 100 * time.Millisecond + } + if !roverWait(ctx, delay-time.Since(started)) { + return + } + } +} +func roverCommands(ctx context.Context, client *http.Client, endpoint string, base map[string]any, state *roverStreamState) { + for ctx.Err() == nil { + _, watch := state.view() + if !watch { + if !roverWait(ctx, 50*time.Millisecond) { + return + } + continue + } + err := roverReadStream(ctx, client, endpoint, base, state) + state.stopStream() + if err != nil && ctx.Err() == nil { + log.Printf("{\"event\":\"rover-command-stream\",\"state\":\"disconnected\"}") + } + if !roverWait(ctx, 250*time.Millisecond) { + return + } + } +} +func roverReadStream(parent context.Context, client *http.Client, endpoint string, base map[string]any, state *roverStreamState) error { + ctx, cancel := context.WithCancel(parent) + defer cancel() + raw, _ := json.Marshal(base) + req, _ := http.NewRequestWithContext(ctx, "POST", endpoint+"/v1/node/rover-stream", bytes.NewReader(raw)) + req.Header.Set("Content-Type", "application/json") + sent := time.Now() + response, err := client.Do(req) + if err != nil { + return err + } + defer response.Body.Close() + if response.StatusCode != 200 { + return errors.New("stream rejected") + } + // The native 200 ms output watchdog is unchanged. A silent stream also + // cancels its network read and retires intent; reconnect never rearms it. + guard := time.AfterFunc(350*time.Millisecond, cancel) + defer guard.Stop() + reader := bufio.NewReaderSize(response.Body, 32768) + first := true + for { + line, err := reader.ReadSlice('\n') + if err != nil { + return err + } + var out roverStreamReply + if json.Unmarshal(line, &out) != nil { + return errors.New("invalid command frame") + } + if first { + if !state.sampleClock(out.Clock, sent, time.Now()) { + return errors.New("invalid stream clock") + } + first = false + } + if !state.accept(out) { + return errors.New("invalid stream intent") + } + if !out.Watch { + return nil + } + if !guard.Reset(350 * time.Millisecond) { + return errors.New("stream deadline") + } + } +} diff --git a/apps/node-agent/internal/node/rover_diagnostics.go b/apps/node-agent/internal/node/rover_diagnostics.go new file mode 100644 index 0000000..aa600df --- /dev/null +++ b/apps/node-agent/internal/node/rover_diagnostics.go @@ -0,0 +1,61 @@ +package node + +// A bounded memory flight recorder. It never stores endpoints, certificates, +// payloads or raw HTTP errors, and never changes admission or lease deadlines. +import ( + "encoding/json" + "log" + "time" +) + +type roverFrame struct { + AtMS int64 `json:"at_ms"` + GapMS int64 `json:"gap_ms"` + BindingMS int64 `json:"binding_ms"` + CoreMS int64 `json:"core_ms"` + DriverMS int64 `json:"driver_ms"` + Status int `json:"http_status"` + Stage string `json:"error_stage,omitempty"` + Session string `json:"session,omitempty"` + Sequence float64 `json:"sequence"` + TTLMS float64 `json:"ttl_ms"` + State string `json:"state"` +} + +type roverDiagnostics struct { + frames []roverFrame + last time.Time + started time.Time + state string + session string + emit func([]byte) +} + +func (d *roverDiagnostics) record(at time.Time, frame roverFrame) { + if d.started.IsZero() { + d.started = at + } + frame.AtMS = at.Sub(d.started).Milliseconds() + if !d.last.IsZero() { + frame.GapMS = at.Sub(d.last).Milliseconds() + } + d.last = at + d.frames = append(d.frames, frame) + if len(d.frames) > 128 { + d.frames = d.frames[len(d.frames)-128:] + } + active := frame.Session != "" || d.session != "" + changed := frame.State != d.state || frame.Session != d.session + if active && (changed || frame.Stage != "" || frame.GapMS >= 200) { + raw, _ := json.Marshal(struct { + Event string `json:"event"` + Frames []roverFrame `json:"frames"` + }{"rover-channel", d.frames}) + if d.emit != nil { + d.emit(raw) + } else { + log.Printf("%s", raw) + } + } + d.state, d.session = frame.State, frame.Session +} diff --git a/apps/node-agent/internal/node/rover_diagnostics_test.go b/apps/node-agent/internal/node/rover_diagnostics_test.go new file mode 100644 index 0000000..20e23a4 --- /dev/null +++ b/apps/node-agent/internal/node/rover_diagnostics_test.go @@ -0,0 +1,63 @@ +package node + +import ( + "encoding/json" + "testing" + "time" +) + +func TestRoverDiagnosticsRetainsCancellationCauseWithoutIdleLogSpam(t *testing.T) { + var events [][]byte + d := roverDiagnostics{emit: func(b []byte) { events = append(events, b) }} + at := time.Unix(100, 0) + for i := 0; i < 200; i++ { + d.record(at.Add(time.Duration(i)*50*time.Millisecond), roverFrame{State: "observing"}) + } + if len(events) != 0 || len(d.frames) != 128 { + t.Fatal("idle must remain bounded and quiet") + } + at = at.Add(11 * time.Second) + d.record(at, roverFrame{Session: "session", Sequence: 1, TTLMS: 350, State: "preparing"}) + d.record(at.Add(100*time.Millisecond), roverFrame{Session: "session", Sequence: 2, TTLMS: 320, State: "preparing"}) + if len(events) != 1 { + t.Fatal("unchanged healthy frames should remain in memory") + } + d.record(at.Add(450*time.Millisecond), roverFrame{CoreMS: 300, Stage: "core_transport", State: "preparing"}) + if len(events) != 2 { + t.Fatal("lost command needs diagnostic evidence") + } + var event struct { + Frames []roverFrame `json:"frames"` + } + if err := json.Unmarshal(events[1], &event); err != nil { + t.Fatal(err) + } + last := event.Frames[len(event.Frames)-1] + if last.Stage != "core_transport" || last.GapMS != 350 || last.CoreMS != 300 || last.Session != "" { + t.Fatal(last) + } + if event.Frames[len(event.Frames)-2].Sequence != 2 { + t.Fatal("last accepted frame lost") + } +} + +func TestRoverDiagnosticsRecordsBindingWaitAndDriverTerminalState(t *testing.T) { + var events [][]byte + d := roverDiagnostics{emit: func(b []byte) { events = append(events, b) }} + at := time.Unix(100, 0) + d.record(at, roverFrame{Session: "session", State: "preparing"}) + d.record(at.Add(500*time.Millisecond), roverFrame{Session: "session", State: "stopped", BindingMS: 400, DriverMS: 2}) + if len(events) != 2 { + t.Fatal("terminal state must flush preceding transport history") + } + var event struct { + Frames []roverFrame `json:"frames"` + } + if err := json.Unmarshal(events[1], &event); err != nil { + t.Fatal(err) + } + last := event.Frames[len(event.Frames)-1] + if last.BindingMS != 400 || last.DriverMS != 2 || last.State != "stopped" { + t.Fatal(last) + } +} diff --git a/apps/node-agent/internal/node/rover_stream_state.go b/apps/node-agent/internal/node/rover_stream_state.go new file mode 100644 index 0000000..692aa86 --- /dev/null +++ b/apps/node-agent/internal/node/rover_stream_state.go @@ -0,0 +1,199 @@ +package node + +import ( + "encoding/hex" + "math" + "sync" + "time" +) + +type roverClock struct { + Instance string `json:"instance"` + MonotonicMS float64 `json:"monotonic_ms"` +} +type roverStreamReply struct { + Watch bool `json:"watch"` + Command map[string]any `json:"command"` + Clock roverClock `json:"control_clock"` +} + +// Clock bounds use the request-send time, never RTT/2 or synchronized wall +// clocks. Server time was sampled after that send, so it gives an upper bound +// on server-minus-local monotonic offset even on an asymmetric network. +type roverStreamState struct { + mu sync.Mutex + origin time.Time + clockID string + upperMS float64 + anchored time.Time + clockSeen time.Time + command map[string]any + watch bool + stopPending bool + stopRevision uint64 + snapshot map[string]any + retired map[string]bool + inhibited bool + wake chan struct{} +} + +func newRoverStreamState(now time.Time) *roverStreamState { + return &roverStreamState{origin: now, stopPending: true, stopRevision: 1, snapshot: map[string]any{}, retired: map[string]bool{}, wake: make(chan struct{}, 1)} +} +func finiteNumber(v any) (float64, bool) { + n, ok := v.(float64) + return n, ok && !math.IsNaN(n) && !math.IsInf(n, 0) +} +func (s *roverStreamState) sampleClock(c roverClock, sent, received time.Time) bool { + if len(c.Instance) != 32 { + return false + } + if _, err := hex.DecodeString(c.Instance); err != nil { + return false + } + if math.IsNaN(c.MonotonicMS) || math.IsInf(c.MonotonicMS, 0) || c.MonotonicMS < 0 || received.Before(sent) { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + upper := c.MonotonicMS - float64(sent.Sub(s.origin).Microseconds())/1000 + 1 + if c.Instance != s.clockID { + s.invalidate() + s.clockID = c.Instance + s.upperMS = upper + s.anchored = received + } else if s.clockSeen.IsZero() || received.Sub(s.clockSeen) > 2*time.Second || upper < s.upperMS+float64(received.Sub(s.anchored).Microseconds())/1e6 { + s.upperMS = upper + s.anchored = received + } + s.clockSeen = received + return true +} +func (s *roverStreamState) accept(reply roverStreamReply) bool { + s.mu.Lock() + defer s.mu.Unlock() + if reply.Clock.Instance != s.clockID || s.clockID == "" { + return false + } + if reply.Command == nil { + if s.command != nil { + s.invalidate() + } + return true + } + command := reply.Command + if len(command) != 6 { + return false + } + for _, k := range []string{"id", "sequence", "left", "right", "settings", "expires_mono_ms"} { + if _, ok := command[k]; !ok { + return false + } + } + expires, ok := finiteNumber(command["expires_mono_ms"]) + if !ok || expires > reply.Clock.MonotonicMS+400.001 { + return false + } + id, ok := command["id"].(string) + if !ok || len(id) != 32 { + return false + } + if _, err := hex.DecodeString(id); err != nil { + return false + } + sequence, ok := finiteNumber(command["sequence"]) + if !ok || sequence < 0 || sequence >= 1<<53 || math.Trunc(sequence) != sequence { + return false + } + if s.inhibited || s.retired[id] { + return true + } + // The native driver independently validates identity, settings and sequence. + if s.stopPending { + s.retire(id) + return true + } + changed := s.command == nil || s.command["id"] != id || s.command["sequence"] != sequence + s.command = command + if changed { + s.notify() + } + return true +} +func (s *roverStreamState) notify() { + select { + case s.wake <- struct{}{}: + default: + } +} +func (s *roverStreamState) stopStream() { + s.mu.Lock() + defer s.mu.Unlock() + s.invalidate() +} +func (s *roverStreamState) retire(id string) { + if len(s.retired) >= 1024 { + s.inhibited = true // Fail closed instead of forgetting interrupted sessions. + return + } + s.retired[id] = true +} +func (s *roverStreamState) invalidate() { + wake := !s.stopPending || s.command != nil + if s.command != nil { + if id, ok := s.command["id"].(string); ok { + s.retire(id) + } + } + s.command = nil + s.stopPending = true + s.stopRevision++ + if wake { + s.notify() + } +} +func (s *roverStreamState) delivery(now time.Time) (bool, map[string]any, uint64) { + s.mu.Lock() + defer s.mu.Unlock() + if s.stopPending { + return s.watch, nil, s.stopRevision + } + if s.command == nil { + return s.watch, nil, 0 + } + if now.Sub(s.clockSeen) > 2*time.Second { + s.invalidate() + return s.watch, nil, s.stopRevision + } + expires, _ := finiteNumber(s.command["expires_mono_ms"]) + // Reserve 25 ms for the private driver RPC and 5 ms plus 1000 ppm for clock + // quantization/rate uncertainty. Expiry is never extended by receipt time. + serverUpper := float64(now.Sub(s.origin).Microseconds())/1000 + s.upperMS + 5 + float64(now.Sub(s.anchored).Microseconds())/1e6 + ttl := math.Min(400, expires-serverUpper-25) + if ttl <= 0 { + s.invalidate() + return s.watch, nil, s.stopRevision + } + out := make(map[string]any, 6) + for k, v := range s.command { + if k != "expires_mono_ms" { + out[k] = v + } + } + out["ttl_ms"] = ttl + return s.watch, out, 0 +} +func (s *roverStreamState) delivered(snapshot map[string]any, stop uint64) { + s.mu.Lock() + defer s.mu.Unlock() + s.snapshot = snapshot + if stop != 0 && stop == s.stopRevision { + s.stopPending = false + } +} +func (s *roverStreamState) view() (map[string]any, bool) { + s.mu.Lock() + defer s.mu.Unlock() + return s.snapshot, s.watch +} +func (s *roverStreamState) setWatch(watch bool) { s.mu.Lock(); s.watch = watch; s.mu.Unlock() } diff --git a/apps/node-agent/internal/node/rover_stream_state_test.go b/apps/node-agent/internal/node/rover_stream_state_test.go new file mode 100644 index 0000000..5948426 --- /dev/null +++ b/apps/node-agent/internal/node/rover_stream_state_test.go @@ -0,0 +1,235 @@ +package node + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +const testClockID = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + +func streamIntent(sequence int, serverNow, expires float64) roverStreamReply { + return roverStreamReply{Watch: true, Clock: roverClock{testClockID, serverNow}, Command: map[string]any{ + "id": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "sequence": float64(sequence), "left": float64(1), "right": float64(1), + "settings": map[string]any{"standstill_confirmed": true, "current_a": float64(30), "max_erpm": float64(2000)}, "expires_mono_ms": expires}} +} +func acknowledgeStop(s *roverStreamState, at time.Time) { + _, _, revision := s.delivery(at) + s.delivered(map[string]any{"state": "observing"}, revision) +} +func TestRoverStreamAsymmetricDelayRetainsOriginalExpiry(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + // Core's clock is +10 seconds. Request took 70 ms outbound and 130 ms + // inbound: dividing RTT in half would be incorrect on this connection. + if !s.sampleClock(roverClock{testClockID, 10070}, origin, origin.Add(200*time.Millisecond)) { + t.Fatal("clock") + } + acknowledgeStop(s, origin.Add(200*time.Millisecond)) + frame := streamIntent(1, 10200, 10600) + if !s.accept(frame) { + t.Fatal("intent") + } + _, command, _ := s.delivery(origin.Add(280 * time.Millisecond)) + if command == nil { + t.Fatal("fresh streamed intent rejected") + } + ttl := command["ttl_ms"].(float64) + if ttl <= 0 || 280+ttl > 600 { + t.Fatal("delay extended original browser deadline", ttl) + } + // Replaying the same frame does not receive a new deadline. + if !s.accept(frame) { + t.Fatal("repeat") + } + _, late, _ := s.delivery(origin.Add(650 * time.Millisecond)) + if late != nil { + t.Fatal("expired frame revived motor intent") + } +} +func TestRoverStreamContinuousInputSurvivesMeasuredRelayJitter(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + s.sampleClock(roverClock{testClockID, 10050}, origin, origin.Add(150*time.Millisecond)) + acknowledgeStop(s, origin.Add(150*time.Millisecond)) + priorUntil := float64(0) + for i, delay := range []int{60, 90, 45, 110, 80, 50, 100, 60, 90, 50} { + sent := 200 + i*100 + arrival := sent + delay + s.accept(streamIntent(i+1, float64(10000+sent), float64(10400+sent))) + if i > 0 && float64(arrival) >= priorUntil { + t.Fatal("lease gap under measured jitter", i, arrival, priorUntil) + } + _, command, _ := s.delivery(origin.Add(time.Duration(arrival) * time.Millisecond)) + if command == nil { + t.Fatal("fresh command lost", i) + } + priorUntil = float64(arrival) + command["ttl_ms"].(float64) + if priorUntil > float64(sent+400) { + t.Fatal("end-to-end deadline expanded") + } + } +} +func TestRoverStreamDisconnectNeedsAcknowledgedStopBeforeNewFrames(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + s.sampleClock(roverClock{testClockID, 10000}, origin, origin) + acknowledgeStop(s, origin) + s.accept(streamIntent(1, 10000, 10400)) + s.stopStream() + _, _, oldStop := s.delivery(origin) + s.stopStream() // Another disconnect races the local driver's response. + s.delivered(map[string]any{}, oldStop) + s.accept(streamIntent(2, 10000, 10400)) + _, command, newStop := s.delivery(origin) + if command != nil || newStop == 0 || newStop == oldStop { + t.Fatal("stale ACK removed stop barrier") + } + s.delivered(map[string]any{}, newStop) + _, command, _ = s.delivery(origin) + if command != nil { + t.Fatal("frame received before acknowledged stop was queued") + } +} +func TestRoverStreamClockChangeAndStaleClockDisarm(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + s.sampleClock(roverClock{testClockID, 10000}, origin, origin) + acknowledgeStop(s, origin) + s.accept(streamIntent(1, 10000, 10400)) + s.sampleClock(roverClock{"cccccccccccccccccccccccccccccccc", 2}, origin, origin) + _, command, stop := s.delivery(origin) + if command != nil || stop == 0 { + t.Fatal("Core restart retained intent") + } + if s.accept(streamIntent(2, 10000, 10400)) { + t.Fatal("old clock accepted") + } + acknowledgeStop(s, origin) + s.sampleClock(roverClock{testClockID, 10000}, origin, origin) + acknowledgeStop(s, origin) + // A fresh-looking frame alone cannot renew the clock calibration. + fresh := streamIntent(3, 13000, 13400) + fresh.Command["id"] = "dddddddddddddddddddddddddddddddd" + s.accept(fresh) + _, command, stop = s.delivery(origin.Add(3 * time.Second)) + if command != nil || stop == 0 { + t.Fatal("stale clock allowed output") + } +} + +func TestRoverStreamUnsentIntentCannotStartAfterReconnect(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + s.sampleClock(roverClock{testClockID, 10000}, origin, origin) + acknowledgeStop(s, origin) + frame := streamIntent(1, 10000, 10400) + s.accept(frame) + // The network fails before the local driver ever sees this session. + s.stopStream() + acknowledgeStop(s, origin) + s.accept(streamIntent(2, 10010, 10410)) + _, command, _ := s.delivery(origin) + if command != nil { + t.Fatal("undelivered old session started after reconnect") + } + frame.Command["id"] = "dddddddddddddddddddddddddddddddd" + s.accept(frame) + _, command, _ = s.delivery(origin) + if command == nil { + t.Fatal("new explicit session rejected after stop") + } +} + +func TestRoverStreamIntentWakesImmediatelyAndCoalescesWithoutReplayWake(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + s.sampleClock(roverClock{testClockID, 10000}, origin, origin) + acknowledgeStop(s, origin) + for i := 1; i <= 3; i++ { + s.accept(streamIntent(i, 10000, 10400)) + } + if len(s.wake) != 1 { + t.Fatal("intent notifications must coalesce") + } + <-s.wake + _, command, _ := s.delivery(origin) + if command["sequence"] != float64(3) { + t.Fatal("queued an intermediate command") + } + s.accept(streamIntent(3, 10000, 10400)) + if len(s.wake) != 0 { + t.Fatal("replayed frame woke driver again") + } + s.stopStream() + if len(s.wake) != 1 { + t.Fatal("stop did not wake driver") + } + <-s.wake + s.stopStream() + if len(s.wake) != 0 { + t.Fatal("repeated failure would spin the local loop") + } +} +func TestRoverStreamRejectsExtendedDeadlineAndUnknownCommandFields(t *testing.T) { + origin := time.Unix(100, 0) + s := newRoverStreamState(origin) + s.sampleClock(roverClock{testClockID, 10000}, origin, origin) + acknowledgeStop(s, origin) + if s.accept(streamIntent(1, 10000, 10401)) { + t.Fatal("extended deadline accepted") + } + frame := streamIntent(1, 10000, 10400) + frame.Command["queued"] = true + if s.accept(frame) { + t.Fatal("unknown field accepted") + } + for _, sequence := range []any{map[string]any{}, []any{1}, "1", 1.5, float64(1 << 53)} { + bad := streamIntent(1, 10000, 10400) + bad.Command["sequence"] = sequence + if s.accept(bad) { + t.Fatal("malformed sequence accepted") + } + } +} +func TestRoverStreamSilenceCancelsReadWithoutWaitingForTelemetry(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/x-ndjson") + json.NewEncoder(w).Encode(roverStreamReply{Watch: true, Clock: roverClock{testClockID, 10000}}) + w.(http.Flusher).Flush() + <-r.Context().Done() + })) + defer server.Close() + state := newRoverStreamState(time.Now()) + started := time.Now() + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + if roverReadStream(ctx, server.Client(), server.URL, map[string]any{}, state) == nil { + t.Fatal("silent stream accepted") + } + if time.Since(started) > time.Second { + t.Fatal("silent stream did not cancel its read") + } +} +func TestRoverStreamReadsMultipleFramesOnOneResponse(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/node/rover-stream" { + t.Error("wrong path") + } + w.Header().Set("Content-Type", "application/x-ndjson") + for i := 0; i < 3; i++ { + raw, _ := json.Marshal(roverStreamReply{Watch: i < 2, Clock: roverClock{testClockID, 10000 + float64(i)}}) + fmt.Fprintln(w, string(raw)) + w.(http.Flusher).Flush() + } + })) + defer server.Close() + state := newRoverStreamState(time.Now()) + if err := roverReadStream(context.Background(), server.Client(), server.URL, map[string]any{}, state); err != nil { + t.Fatal(err) + } +} diff --git a/packages/rover-control/README.md b/packages/rover-control/README.md new file mode 100644 index 0000000..8fa845b --- /dev/null +++ b/packages/rover-control/README.md @@ -0,0 +1,107 @@ +# Rover control behaviour prototype + +Status: **offline behavioural reference; no production actuation adapter**. +Requested by the owner while unavailable for physical testing, 2026-09-24. +Neither Core composition nor Node/VESC runtime imports these modules. No new +driver, service, firmware, motor settings or authority capability is installed. + +`src/profile.ts` contains the versioned desired-profile parser, Tank and Arcade +mixers, continuous deadband, linear/squared response, relative output scaling, +and fan-out to any number of UUID-bound motors on both sides. UUIDs must be +unique; a missing side is rejected. `forwardSign` is the verified sign at the +future actuator adapter, not a copy of `m_invert_direction` and not a second +automatic inversion of the current VESC configuration. + +The normalized result has no electrical units. It cannot be sent as amperes, +watts, duty, ERPM or speed without a separately qualified adapter and individual +motor, battery/BMS and braking limits. A profile output scale of 80% is **not** +the owner's proposed 20% safety margin against verified equipment ratings. +The current real RC path is PPM Duty Cycle; this prototype does not change it. + +Arcade uses continuous diamond desaturation, with forward positive and right +yaw positive. For shaped inputs `v` and `r`, the pair is `(v+r, v-r)` multiplied +by `max(abs(v),abs(r))/(abs(v)+abs(r))`, or zero at the origin. This follows the +WPILib ArcadeDriveIK geometry with the steering sign adapted to the UI. +Reverse plus right still requests right yaw; it is not a car steering-wheel +convention, curvature drive, a turn-radius controller or omnidirectional motion. +No VESC Tool calibration algorithm is reproduced. + +References inspected 2026-09-24: + +- [WPILib drive classes](https://docs.wpilib.org/en/stable/docs/software/hardware-apis/motors/wpi-drive-classes.html) +- [ArcadeDriveIK source](https://github.com/wpilibsuite/allwpilib/blob/main/wpilibc/src/main/native/cpp/drive/DifferentialDrive.cpp) + +## Authority model + +`src/authority.ts` is a pure deterministic model. Its output contains **intent** +to stop all drives, revoke the Core epoch, flush queued motion and cancel +autonomous motion tasks; it does not actually stop a motor or cancel a process. +The future actuator must enforce the gate synchronously before asynchronous +task cancellation. RC reception, drive supervision and telemetry must survive. + +Boot, input loss, controller loss, stale/invalid data, a timing gap or Core +command loss enter hold. First RC deflection while Core owns motion consumes +the gesture and revokes the old token. Every assigned input must then be neutral +and every motor must have trusted physical-stop evidence for the full configured +interval. A held first gesture or repeated packet does not qualify. The next +gesture starts RC manual operation. Neutral never resumes the previous Core +task. Reacquisition requires a new explicit, nonreplayed request in neutral. +Old Core commands cannot cross an epoch or process boot identity. + +Policy values are mandatory constructor arguments. The demo uses 100 ms +freshness / 200 ms neutral solely as synthetic fixtures, **not accepted rover +reaction limits**. Use one monotonic clock domain from an authenticated local +producer and a fresh unpredictable boot identity per instance. A token is a +stale-command fence, not authentication or a replacement for access control. +Calls and input types belong to a trusted model harness; this is not a public +network request parser. Run a supervisory tick even when no new input arrives. + +`link.live`, sample acquisition timestamps/sequences and `drive.stopped` must +come from qualified evidence. A fresh USB response with old decoded PPM does +not meet this contract. Zero motor current/duty alone is not physical stop. +The current FW 5.02 input API does not supply all required evidence. In +particular, stop-first independent of Mini needs an enforcement point outside +Mini and coordination of **all** motor controllers. No such qualified mechanism +is claimed by these tests. Receiver failsafe is also still awaiting acceptance. + +Input axes are semantic controls, not invented receiver channels. The mixer for +Tank requires two Y axes; Arcade requires both axes of the selected stick. The +authority policy separately declares all `monitoredAxes`: the demo watches all +four axes, so the other stick also stops Core in Arcade. Missing monitored axes +are rejected, not synthesized as zero; all must return to neutral. The current two +separate receiver-to-VESC PWM outputs do not establish access to those Arcade +axes. The channel map, radio-link semantics and independent mixed RC path remain +hardware integration work. Profile revision or binding changes require a fresh +model initialized in hold, not an in-place live change. + +## Preview and validation + +Use the already installed Control Station toolchain; no new dependencies: + +```sh +cd apps/control-station +node --test test/roverControl.test.mjs +./node_modules/.bin/tsc --project tools/rover-control-preview/tsconfig.json +node tools/rover-control-preview/build.mjs /absolute/artifact/directory +``` + +The self-contained HTML uses canonical Design Guideline components. Its CSP +disables all network connections, and no transport/serial code exists in the +bundle. It is an owner-review artifact, outside production navigation. Browser +storage and JSON export contain a **draft**, never confirmed applied state. +It does not add USB controls, device-specific phantom entities or a runtime +source-selection switch to the product. The synthetic takeover trace is +engineering review content, not a proposed operator control panel. + +Intended product placement remains the existing shared «Настройки борта» section +on Core and Node, between computer details and devices. Alternatives (a new +workspace or separate per-VESC control-mode selectors) would fragment a single +vehicle-wide profile and are not used. Existing `Inspector`, `SettingsCard`, +`InspectorSelectField`, `RangeControl`, `Button`, `StatusBadge` cover the review; +no Design Guideline extensions or new visual primitives were needed. + +Before production integration: verified input mapping and radio failsafe, +independent stop-first actuator mechanism, desired/applied profile storage on +Node with optimistic revision checking, all-member application receipts and +failure recovery, then shared UI and supervised unloaded tests. Do not expose +an enabled Apply button based only on successful model tests. diff --git a/packages/rover-control/src/authority.ts b/packages/rover-control/src/authority.ts new file mode 100644 index 0000000..59dcc27 --- /dev/null +++ b/packages/rover-control/src/authority.ts @@ -0,0 +1,134 @@ +/** Behavioural reference ONLY. Not connected to the current VESC PPM path. */ +import {axisKeys, bounded, mix, parseProfile, type Axes, type ControlProfile, type Sides} from './profile'; + +export interface Sample { value: number; at: number; sequence: number } +export interface DriveEvidence { at: number; healthy: boolean; stopped: boolean } +export interface CoreCommand { + token: string; sequence: number; at: number; expires: number; demand: Sides; +} +export interface Observation { + now: number; + // These are trusted receiver timestamps and link status, NOT USB read times. + link: { state: 'live' | 'lost' | 'unknown'; at: number }; + axes: Partial>; + drives: Record; + command?: CoreCommand; + requestCore?: {owner: 'remote' | 'autonomy'; sequence: number; at: number}; +} +export interface Policy { + maxAgeMs: number; neutralMs: number; maxCommandMs: number; + /** Verified RC controls that can take over, including non-driving stick axes. */ + monitoredAxes: readonly (keyof Axes)[]; +} +export type State = 'hold' | 'rc-ready' | 'rc-manual' | 'core'; +export interface Decision { + state: State; reason: string; demand: Sides; token: string | null; + owner: 'remote' | 'autonomy' | 'rc' | null; + stopAll: boolean; flushMotionQueue: boolean; cancelMotionTasks: boolean; +} +const zero = (): Sides => ({left: 0, right: 0}); + +export class AuthorityModel { + private state: State = 'hold'; + private epoch = 0; + private token: string | null = null; + private owner: Decision['owner'] = null; + private neutralSince: number | null = null; + private lastNow = -Infinity; + private lastCommand = -1; + private lastRequest = -1; + private samples: Partial> = {}; + private readonly profile: ControlProfile; + private readonly motors: string[]; + private readonly policy: Policy; + + constructor(profile: ControlProfile, motors: readonly string[], policy: Policy, private readonly bootId: string) { + this.profile = parseProfile(profile); + this.motors = [...motors]; this.policy = {...policy, monitoredAxes:[...policy.monitoredAxes]}; + if (!bootId || !motors.length || new Set(motors).size !== motors.length || motors.some(id => !id) + || !bounded(policy.maxAgeMs, 1, 10000) || !bounded(policy.neutralMs, 1, 10000) + || !bounded(policy.maxCommandMs, 1, 10000) + || new Set(policy.monitoredAxes).size !== policy.monitoredAxes.length + || policy.monitoredAxes.some(key => !['leftY','rightY','leftX','rightX'].includes(key)) + || axisKeys(this.profile).some(key => !policy.monitoredAxes.includes(key))) throw Error('Invalid authority policy'); + } + private result(reason: string, demand = zero(), revoke = false): Decision { + return {state: this.state, reason, demand, token: this.token, owner: this.owner, + stopAll: this.state === 'hold', flushMotionQueue: revoke, cancelMotionTasks: revoke}; + } + private hold(reason: string): Decision { + const revoke = this.state === 'core'; + if (this.state !== 'hold') this.epoch++; + this.state = 'hold'; this.token = null; this.owner = null; + this.neutralSince = null; this.lastCommand = -1; + return this.result(reason, zero(), revoke); + } + step(input: Observation): Decision { + const now = input.now; + if (!Number.isFinite(now) || now < 0 || now < this.lastNow) return this.hold('clock-invalid'); + const gap = now - this.lastNow; this.lastNow = now; + const fresh = (at: number) => Number.isFinite(at) && at <= now && now - at <= this.policy.maxAgeMs; + const request = input.requestCore; + const newRequest = !!request && Number.isSafeInteger(request.sequence) && request.sequence > this.lastRequest; + // Consume even a premature request: it must never become effective later. + if (newRequest) this.lastRequest = request.sequence; + // A gap cannot count towards continuous observed neutral. + if (gap > this.policy.maxAgeMs) { + const first = gap === Infinity; + if (!first) return this.hold('observation-gap'); + this.neutralSince = null; + } + if (input.link?.state !== 'live' || !fresh(input.link.at)) return this.hold('receiver-unverified'); + const axes = {leftY: 0, rightY: 0, leftX: 0, rightX: 0}; + let neutral = true; + let observedThrough = input.link.at; + const candidates: Partial> = {}; + for (const key of this.policy.monitoredAxes) { + const sample = input.axes[key], prev = this.samples[key]; + if (!sample || !bounded(sample.value, -1, 1) || !fresh(sample.at) + || !Number.isSafeInteger(sample.sequence) || sample.sequence < 0 + || (prev && (sample.sequence < prev.sequence || sample.at < prev.at + || (sample.sequence === prev.sequence && (sample.value !== prev.value || sample.at !== prev.at))))) + return this.hold('axis-invalid'); + candidates[key] = {...sample}; axes[key] = sample.value; + observedThrough = Math.min(observedThrough, sample.at); + neutral &&= Math.abs(sample.value) <= this.profile.deadband; + } + this.samples = candidates; + let stopped = true; + for (const id of this.motors) { + const drive = input.drives[id]; + if (!drive || drive.healthy !== true || !fresh(drive.at)) return this.hold('drive-unverified'); + observedThrough = Math.min(observedThrough, drive.at); + stopped &&= drive.stopped === true; + } + // RC intent is evaluated before any Core command or reacquisition request. + if (this.state === 'core' && !neutral) return this.hold('rc-takeover'); + if (neutral && stopped) this.neutralSince ??= now; + else this.neutralSince = null; + const stableNeutral = this.neutralSince !== null && observedThrough - this.neutralSince >= this.policy.neutralMs; + if (this.state === 'hold') { + if (!stableNeutral) return this.result(stopped ? 'await-neutral' : 'await-stop'); + this.state = 'rc-ready'; this.owner = 'rc'; + // A request queued before reaching neutral cannot acquire Core authority. + return this.result('rc-ready'); + } + if (newRequest && request && fresh(request.at) && this.state !== 'core' && stableNeutral) { + if (!['remote', 'autonomy'].includes(request.owner)) return this.hold('source-invalid'); + this.epoch++; this.token = `${this.bootId}:${this.epoch}:${this.profile.revision}`; + this.owner = request.owner; this.state = 'core'; this.lastCommand = -1; + return this.result('core-granted'); + } + if (this.state === 'core') { + const c = input.command; + if (!c || c.token !== this.token || !Number.isSafeInteger(c.sequence) || c.sequence <= this.lastCommand + || !fresh(c.at) || !Number.isFinite(c.expires) || c.expires <= now + || c.expires - c.at > this.policy.maxCommandMs || c.expires < c.at + || !bounded(c.demand.left, -1, 1) || !bounded(c.demand.right, -1, 1)) return this.hold('core-command-invalid'); + this.lastCommand = c.sequence; + return this.result('core-command', {left: c.demand.left * this.profile.outputScale, right: c.demand.right * this.profile.outputScale}); + } + if (this.state === 'rc-ready' && !neutral) this.state = 'rc-manual'; + return this.result(neutral ? 'rc-neutral' : 'rc-command', mix(this.profile, axes)); + } +} diff --git a/packages/rover-control/src/profile.ts b/packages/rover-control/src/profile.ts new file mode 100644 index 0000000..b5ba470 --- /dev/null +++ b/packages/rover-control/src/profile.ts @@ -0,0 +1,72 @@ +/** Executable profile proposal. Pure calculations; no hardware or transport. */ +export interface ControlProfile { + schema: 'missioncore.rover-control/v1'; + revision: number; + mode: 'tank' | 'arcade'; + stick: 'left' | 'right'; + deadband: number; + response: 'linear' | 'squared'; + outputScale: number; +} +export interface Axes { leftY: number; rightY: number; leftX: number; rightX: number } +export interface Sides { left: number; right: number } +export interface MotorBinding { uuid: string; side: 'left' | 'right'; forwardSign: 1 | -1 } + +export const defaultProfile: Readonly = Object.freeze({ + schema: 'missioncore.rover-control/v1', revision: 0, mode: 'tank', stick: 'right', + deadband: 0.15, response: 'linear', outputScale: 1, +}); +export function bounded(value: unknown, min: number, max: number): value is number { + return typeof value === 'number' && Number.isFinite(value) && value >= min && value <= max; +} +export function parseProfile(value: unknown): ControlProfile { + if (!value || typeof value !== 'object' || Array.isArray(value)) throw Error('Invalid profile'); + const p = value as ControlProfile; + if (Object.keys(p).sort().join() !== Object.keys(defaultProfile).sort().join() + || p.schema !== defaultProfile.schema || !Number.isSafeInteger(p.revision) || p.revision < 0 + || !['tank', 'arcade'].includes(p.mode) || !['left', 'right'].includes(p.stick) + || !['linear', 'squared'].includes(p.response) || !bounded(p.deadband, 0, 0.5) + || !bounded(p.outputScale, 0, 1)) throw Error('Invalid profile'); + return {...p}; +} +export function axisKeys(profile: ControlProfile): (keyof Axes)[] { + return profile.mode === 'tank' ? ['leftY', 'rightY'] + : profile.stick === 'right' ? ['rightY', 'rightX'] : ['leftY', 'leftX']; +} +export function shapeAxis(value: number, profile: ControlProfile): number { + if (!bounded(value, -1, 1)) throw Error('Invalid axis'); + const magnitude = Math.max(0, (Math.abs(value) - profile.deadband) / (1 - profile.deadband)); + return Math.sign(value) * (profile.response === 'squared' ? magnitude * magnitude : magnitude); +} +/** +Y = forward, +X = turn right, including while reversing (yaw convention). + * Arcade diamond desaturation follows the documented WPILib ArcadeDriveIK + * convention, with clockwise steering sign adapted here. Output is normalized + * demand, NOT amps, watts, ERPM, physical velocity, or a guaranteed turn radius. + */ +export function mix(profile: ControlProfile, axes: Axes): Sides { + parseProfile(profile); + for (const key of axisKeys(profile)) if (!bounded(axes[key], -1, 1)) throw Error('Missing or invalid axis'); + let left: number, right: number; + if (profile.mode === 'tank') { + left = shapeAxis(axes.leftY, profile); right = shapeAxis(axes.rightY, profile); + } else { + const throttle = shapeAxis(profile.stick === 'right' ? axes.rightY : axes.leftY, profile); + const turn = shapeAxis(profile.stick === 'right' ? axes.rightX : axes.leftX, profile); + const peak = Math.max(Math.abs(throttle), Math.abs(turn)); + const scale = peak === 0 ? 0 : peak / (Math.abs(throttle) + Math.abs(turn)); + left = (throttle + turn) * scale; right = (throttle - turn) * scale; + } + return {left: left * profile.outputScale || 0, right: right * profile.outputScale || 0}; +} +/** All members of both sides are required, irrespective of 1x1/2x2/6x6. */ +export function motorDemands(sides: Sides, bindings: readonly MotorBinding[]): Record { + if (!bounded(sides.left, -1, 1) || !bounded(sides.right, -1, 1) + || !bindings.some(b => b.side === 'left') || !bindings.some(b => b.side === 'right')) throw Error('Incomplete drive'); + const values: Record = Object.create(null); + for (const b of bindings) { + if (!/^[0-9a-f]{24}$/.test(b.uuid) || b.uuid in values || !['left', 'right'].includes(b.side) + || ![1, -1].includes(b.forwardSign)) throw Error('Invalid motor binding'); + values[b.uuid] = sides[b.side] * b.forwardSign || 0; + } + return values; +} diff --git a/packages/sensor-ui/src/BoardSections.tsx b/packages/sensor-ui/src/BoardSections.tsx new file mode 100644 index 0000000..9f21564 --- /dev/null +++ b/packages/sensor-ui/src/BoardSections.tsx @@ -0,0 +1,23 @@ +import {useEffect,useSyncExternalStore,type ReactNode} from 'react'; +import {Button,Icon,Inspector,LoadingRegion} from '@nodedc/ui-react'; +import type {BoardLayoutStore} from './boardLayout'; + +export interface BoardSectionsProps { + layout:BoardLayoutStore; + computer:ReactNode; + description?:string; +} +export function BoardSections({layout,computer,description,settings,devices}:BoardSectionsProps&{settings:ReactNode;devices:ReactNode}){ + const state=useSyncExternalStore(layout.subscribe,layout.getSnapshot); + useEffect(()=>{void layout.load();},[layout]); + return
+ {state.error&&

{state.error}

} + + ,disabled:!state.ready,content:computer}, + {id:'settings',label:'Настройки борта',icon:,disabled:!state.ready,content:settings}, + {id:'devices',label:'Устройства аппарата',icon:,disabled:!state.ready,content:devices}, + ]}/> + +
; +} diff --git a/packages/sensor-ui/src/boardLayout.ts b/packages/sensor-ui/src/boardLayout.ts new file mode 100644 index 0000000..c1681ed --- /dev/null +++ b/packages/sensor-ui/src/boardLayout.ts @@ -0,0 +1,66 @@ +export const boardSections = ['computer', 'settings', 'devices'] as const; +export type BoardSection = typeof boardSections[number]; +export interface BoardLayout {schema:'missioncore.board-layout/v1';revision:number;open_sections:BoardSection[]} +export interface BoardLayoutTransport { + read:()=>Promise; + patch:(section:BoardSection,open:boolean)=>Promise; +} +export const defaultBoardLayout:BoardLayout={schema:'missioncore.board-layout/v1',revision:0,open_sections:[...boardSections]}; +type Change={section:BoardSection;open:boolean}; +type Snapshot={value:BoardLayout;ready:boolean;error:string|null;saving:boolean}; +function apply(value:BoardLayout,change:Change):BoardLayout { + const open=new Set(value.open_sections); + if(change.open)open.add(change.section);else open.delete(change.section); + return {...value,open_sections:boardSections.filter(section=>open.has(section))}; +} +function validate(value:BoardLayout):BoardLayout { + if(value.schema!==defaultBoardLayout.schema||!Number.isSafeInteger(value.revision)||value.revision<0|| + !Array.isArray(value.open_sections)||new Set(value.open_sections).size!==value.open_sections.length|| + value.open_sections.some(id=>!boardSections.includes(id)))throw new Error('Не удалось прочитать раскладку блоков.'); + return value; +} +// The queue belongs to the application resource, not a mounted accordion. +// Navigation cannot discard a pending save or let an older reply win a toggle. +export function createBoardLayoutStore(transport:BoardLayoutTransport){ + let saved=defaultBoardLayout; + let snapshot:Snapshot={value:saved,ready:false,error:null,saving:false}; + let pending:Change[]=[]; + let reading:Promise|null=null; + let writing=false; + const listeners=new Set<()=>void>(); + const emit=(patch:Partial={})=>{ + snapshot={...snapshot,...patch,value:pending.reduce(apply,saved),saving:writing||pending.length>0}; + listeners.forEach(listener=>listener()); + }; + const load=():Promise=>{ + if(reading)return reading; + if(writing)return Promise.resolve(); + reading=transport.read().then(value=>{saved=validate(value);emit({ready:true,error:null});}) + .catch(()=>emit({error:'Раскладка блоков не загружена. Повторите подключение.'})) + .finally(()=>{reading=null;}); + return reading; + }; + const flush=async()=>{ + if(writing)return; + writing=true;emit({error:null}); + while(pending.length){ + const change=pending[0]; + try{saved=validate(await transport.patch(change.section,change.open));pending.shift();emit();} + catch{pending=[];emit({error:'Не удалось сохранить раскладку блоков. Изменение отменено.'});break;} + } + writing=false;emit(); + }; + return { + getSnapshot:()=>snapshot, + subscribe:(listener:()=>void)=>{listeners.add(listener);return()=>{listeners.delete(listener);};}, + load, + change:(open:string[])=>{ + if(!snapshot.ready||reading)return; + for(const section of boardSections){ + if(open.includes(section)!==snapshot.value.open_sections.includes(section))pending.push({section,open:open.includes(section)}); + } + emit();void flush(); + }, + }; +} +export type BoardLayoutStore=ReturnType; diff --git a/src/k1link/fleet/board_layout.py b/src/k1link/fleet/board_layout.py new file mode 100644 index 0000000..ad3415b --- /dev/null +++ b/src/k1link/fleet/board_layout.py @@ -0,0 +1,45 @@ +"""Operator presentation only; never transported to a motor controller.""" +import hashlib +import json +from pathlib import Path + +from .trust import atomic_private + +SCHEMA = "missioncore.board-layout/v1" +SECTIONS = ("computer", "settings", "devices") + + +def path(root: Path, vehicle: str) -> Path: + return root / "board-layouts" / (hashlib.sha256(vehicle.encode()).hexdigest() + ".json") + + +def read(root: Path, vehicle: str) -> dict: + target = path(root, vehicle) + if not target.exists(): + return {"schema": SCHEMA, "revision": 0, "open_sections": list(SECTIONS)} + if target.is_symlink() or target.stat().st_size > 4096: + raise ValueError("Invalid layout file") + value = json.loads(target.read_bytes()) + if (not isinstance(value, dict) or set(value) != {"schema", "revision", "open_sections"} + or value["schema"] != SCHEMA or type(value["revision"]) is not int + or not 0 <= value["revision"] < 2**53-1 + or not isinstance(value["open_sections"], list) + or any(item not in SECTIONS for item in value["open_sections"]) + or len(set(value["open_sections"])) != len(value["open_sections"])): + raise ValueError("Invalid layout document") + return value + + +def update(root: Path, vehicle: str, section: str, opened: bool) -> dict: + # Caller holds the fleet writer lock. A patch cannot lose another section. + if section not in SECTIONS or type(opened) is not bool: + raise ValueError("Invalid layout change") + value = read(root, vehicle) + opened_sections = set(value["open_sections"]) + if opened: + opened_sections.add(section) + else: + opened_sections.discard(section) + value.update(revision=value["revision"]+1, open_sections=[s for s in SECTIONS if s in opened_sections]) + atomic_private(path(root, vehicle), (json.dumps(value, indent=2)+"\n").encode()) + return value diff --git a/src/k1link/fleet/rover_control.py b/src/k1link/fleet/rover_control.py new file mode 100644 index 0000000..fd3f5f9 --- /dev/null +++ b/src/k1link/fleet/rover_control.py @@ -0,0 +1,135 @@ +"""Ephemeral operator leases. No motor calls, command queue or persisted motion.""" +from __future__ import annotations + +import copy +import math +import secrets +import threading +import time + + +class RoverControl: + def __init__(self, clock=time.monotonic): + self.clock = clock + self.lock = threading.RLock() + self.changed = threading.Condition(self.lock) + self.boards = {} + self.clock_id = secrets.token_hex(16) + + def _board(self, node): + return self.boards.setdefault(node, {"seen": -1e9, "watch": 0, "snapshot": {}, + "session": None, "relay": None}) + + def view(self, node): + with self.lock: + b = self._board(node) + b["watch"] = self.clock() + 2 + fresh = self.clock() - b["seen"] < 1 + self._expire(b) + return {"fresh": fresh, "snapshot": copy.deepcopy(b["snapshot"]) if fresh else {}, + "controlling": b["session"] is not None} + + def _expire(self, b): + if b["session"] and self.clock() >= b["session"]["until"]: + b["session"] = None + + def arm(self, node, body): + if (set(body) != {"standstill_confirmed", "current_a", "max_erpm"} + or body["standstill_confirmed"] is not True + or type(body["current_a"]) not in (int, float) + or not .5 <= body["current_a"] <= 30 + or type(body["max_erpm"]) not in (int, float) + or not 300 <= body["max_erpm"] <= 3000): + raise ValueError("Подтвердите остановку и выберите допустимые пределы.") + with self.lock: + b = self._board(node) + self._expire(b) + if b["session"]: + raise ValueError("Аппаратом уже управляют. Сначала остановите управление.") + snapshot = b["snapshot"] + if self.clock() - b["seen"] >= 1 or snapshot.get("supported") is not True: + raise ValueError("Канал управления бортом недоступен.") + if snapshot.get("state") in ("preparing", "ready", "driving", "stopping"): + raise ValueError("Дождитесь завершения предыдущего управления.") + identifier = secrets.token_hex(16) + b["session"] = {"id": identifier, "sequence": 0, "until": self.clock()+.4, + "left": 0, "right": 0, "settings": copy.deepcopy(body)} + self.changed.notify_all() + return {"session_id": identifier} + + def command(self, node, body): + if set(body) != {"session_id", "sequence", "left", "right", "stop"}: + raise ValueError("Некорректная команда управления.") + if (type(body["sequence"]) is not int or not 1 <= body["sequence"] < 2**53 + or type(body["stop"]) is not bool + or any(type(body[k]) not in (int,float) or not math.isfinite(body[k]) + or abs(body[k]) > 1 for k in ("left", "right"))): + raise ValueError("Некорректные значения команды.") + with self.lock: + b = self._board(node) + self._expire(b) + s = b["session"] + if not s or not secrets.compare_digest(str(body["session_id"]), s["id"]): + raise ValueError("Управление завершено. Включите его заново.") + if body["sequence"] <= s["sequence"]: + raise ValueError("Устаревшая команда отклонена.") + if body["stop"]: + b["session"] = None + else: + s.update(sequence=body["sequence"], left=body["left"], right=body["right"], + until=self.clock()+.4) + self.changed.notify_all() + return {"accepted_sequence": body["sequence"]} + + def exchange(self, node, body): + """Called only after normal paired-certificate/binding authentication.""" + snapshot = body.get("rover") + relay = body.get("relay_id") + if not isinstance(snapshot, dict) or not isinstance(relay, str) or len(relay) != 32: + raise ValueError("Invalid rover exchange") + with self.lock: + b = self._board(node) + if b["relay"] != relay: + b["session"] = None # No resume after relay restart/rebind. + if b["snapshot"].get("instance") != snapshot.get("instance"): + b["session"] = None # Driver restarts also revoke the browser lease. + b.update(relay=relay, snapshot=copy.deepcopy(snapshot), seen=self.clock()) + self._expire(b) + s = b["session"] + if s and snapshot.get("session_id") == s["id"] and snapshot.get("state") in ( + "stopped", "fault", "receiver"): + b["session"] = s = None + command = None + if s: + command = {k: copy.deepcopy(v) for k,v in s.items() if k != "until"} + command["ttl_ms"] = max(0, min(400, int((s["until"]-self.clock())*1000))) + return {"watch": self.clock() < b["watch"] or s is not None, "command": command, + "control_clock": self._clock()} + + def _clock(self): + return {"instance": self.clock_id, "monotonic_ms": self.clock()*1000} + + def wait_for_intent(self, node, relay, previous, timeout=.05): + """Wake on a new intent, including updates between write and wait.""" + previous = previous or {} + def changed(): + board = self._board(node) + current = (board["session"] if board["relay"] == relay else None) or {} + return (current.get("id"), current.get("sequence")) != (previous.get("id"), previous.get("sequence")) + with self.changed: + self.changed.wait_for(changed, timeout) + + def stream(self, node, relay): + """Latest intent only. Telemetry round trips never pace command delivery.""" + with self.lock: + b = self._board(node) + self._expire(b) + if self.clock() - b["seen"] >= 1: + b["session"] = None + s = b["session"] if b["relay"] == relay else None + command = None + if s: + command = {k: copy.deepcopy(v) for k,v in s.items() if k != "until"} + command["expires_mono_ms"] = s["until"]*1000 + return {"watch": self.clock() < b["watch"] or s is not None, + "command": command, "control_clock": self._clock()} diff --git a/src/k1link/web/rover_api.py b/src/k1link/web/rover_api.py new file mode 100644 index 0000000..edd4988 --- /dev/null +++ b/src/k1link/web/rover_api.py @@ -0,0 +1,40 @@ +"""Local operator BFF for the paired Node's bounded control channel.""" +from typing import Annotated +from fastapi import APIRouter, Depends, HTTPException, Response +from k1link.fleet.registry import FleetRegistry +from k1link.fleet.trust import PairingError +from .fleet_api import local_operator + +router = APIRouter(prefix="/api/v1/fleet", tags=["fleet"]) + +def node(fleet, vehicle): + with fleet.lock: + row = fleet.find(vehicle) + if row["enrollment"] != "paired": + raise ValueError("Борт не привязан.") + return row["node_id"] + +@router.get("/{vehicle_id}/rover") +def rover_state(vehicle_id: str, response: Response, + fleet: Annotated[FleetRegistry, Depends(local_operator)]): + response.headers["Cache-Control"] = "no-store" + try: + return fleet.rover_control.view(node(fleet, vehicle_id)) + except (ValueError, PairingError) as error: + raise HTTPException(409, str(error)) from None + +@router.post("/{vehicle_id}/rover/arm") +def rover_arm(vehicle_id: str, body: dict, + fleet: Annotated[FleetRegistry, Depends(local_operator)]): + try: + return fleet.rover_control.arm(node(fleet, vehicle_id), body) + except (ValueError, PairingError) as error: + raise HTTPException(409, str(error)) from None + +@router.post("/{vehicle_id}/rover/command") +def rover_command(vehicle_id: str, body: dict, + fleet: Annotated[FleetRegistry, Depends(local_operator)]): + try: + return fleet.rover_control.command(node(fleet, vehicle_id), body) + except (ValueError, PairingError) as error: + raise HTTPException(409, str(error)) from None diff --git a/tests/fleet/test_board_layout.py b/tests/fleet/test_board_layout.py new file mode 100644 index 0000000..5e50780 --- /dev/null +++ b/tests/fleet/test_board_layout.py @@ -0,0 +1,42 @@ +import json +import threading +from types import SimpleNamespace + +from fastapi import FastAPI +from fastapi.testclient import TestClient +import pytest + +from k1link.fleet import board_layout +from k1link.fleet.trust import PairingError +from k1link.web.fleet_api import router, local_operator + + +def test_layout_api_persists_empty_and_isolates_vehicles(tmp_path): + app = FastAPI() + app.include_router(router) + def find(identifier): + if identifier not in ("rover-a", "rover-b"): + raise PairingError("Unknown") + fleet = SimpleNamespace(root=tmp_path, lock=threading.RLock(), find=find) + app.dependency_overrides[local_operator] = lambda: fleet + client = TestClient(app) + for section in board_layout.SECTIONS: + response = client.patch('/api/v1/fleet/rover-a/board-layout', json={"section": section, "open": False}) + assert response.status_code == 200 + assert client.get('/api/v1/fleet/rover-a/board-layout').json()['open_sections'] == [] + assert client.get('/api/v1/fleet/rover-b/board-layout').json()['open_sections'] == list(board_layout.SECTIONS) + assert board_layout.read(tmp_path, 'rover-a')['revision'] == 3 + assert board_layout.path(tmp_path, 'rover-a').stat().st_mode & 0o777 == 0o600 + for body in ({"section":"motor","open":True},{"section":"computer","open":"true"},{"section":"computer","open":True,"extra":0}): + assert client.patch('/api/v1/fleet/rover-a/board-layout', json=body).status_code == 422 + assert client.get('/api/v1/fleet/missing/board-layout').status_code == 404 + + +def test_corrupt_layout_is_preserved(tmp_path): + board_layout.update(tmp_path, "a", "computer", False) + target = board_layout.path(tmp_path, "a") + raw = json.dumps({"schema":board_layout.SCHEMA,"revision":3,"open_sections":["motor"]}) + target.write_text(raw) + with pytest.raises(ValueError): + board_layout.update(tmp_path, "a", "devices", False) + assert target.read_text() == raw diff --git a/tests/fleet/test_rover_control.py b/tests/fleet/test_rover_control.py new file mode 100644 index 0000000..a535114 --- /dev/null +++ b/tests/fleet/test_rover_control.py @@ -0,0 +1,115 @@ +import pytest +from k1link.fleet.rover_control import RoverControl + +@pytest.fixture +def hub(): + now=[10.] + h=RoverControl(lambda:now[0]) + h.exchange('node',{'relay_id':'a'*32,'rover':{'supported':True,'instance':'driver','state':'observing'}}) + return h,now + +def arm(h): + return h.arm('node',{'standstill_confirmed':True,'current_a':30,'max_erpm':2000})['session_id'] + +def cmd(id,seq=1,**kw): + return {'session_id':id,'sequence':seq,'left':1,'right':1,'stop':False,**kw} + +def exchange(h,**kw): + return h.exchange('node',{'relay_id':'a'*32,'rover':{'supported':True,'instance':'driver',**kw}}) + +def test_expired_browser_cannot_be_revived_by_late_packet(hub): + h,now=hub;id=arm(h);h.command('node',cmd(id));now[0]+=.401 + assert exchange(h)['command'] is None + with pytest.raises(ValueError):h.command('node',cmd(id,2)) + +def test_sequence_stop_and_single_owner(hub): + h,_=hub;id=arm(h) + with pytest.raises(ValueError):arm(h) + h.command('node',cmd(id,2)) + with pytest.raises(ValueError):h.command('node',cmd(id,1)) + assert exchange(h)['command']['sequence']==2 + h.command('node',cmd(id,3,stop=True)) + with pytest.raises(ValueError):h.command('node',cmd(id,4)) + assert exchange(h)['command'] is None + +@pytest.mark.parametrize('snapshot',[{'state':'receiver'},{'state':'fault'},{'instance':'new-driver'}]) +def test_takeover_fault_or_driver_restart_revokes(hub,snapshot): + h,_=hub;id=arm(h) + assert exchange(h,session_id=id,**snapshot)['command'] is None + +def test_relay_restart_and_cross_board_do_not_inherit_authority(hub): + h,_=hub;id=arm(h) + with pytest.raises(ValueError):h.command('other',cmd(id)) + assert h.exchange('node',{'relay_id':'b'*32,'rover':{}})['command'] is None + +@pytest.mark.parametrize('value',[True,float('nan'),float('inf'),1.01,-1.01,'1']) +def test_invalid_demand_cannot_refresh(hub,value): + h,_=hub;id=arm(h) + with pytest.raises(ValueError):h.command('node',cmd(id,left=value)) + +def test_stale_telemetry_is_unavailable_and_arm_requires_current_board(hub): + h,now=hub;now[0]+=1.1 + assert h.view('node')['snapshot']=={} + with pytest.raises(ValueError):arm(h) + +def test_stream_keeps_absolute_deadline_and_never_renews_held_frame(hub): + h, now = hub + identifier = arm(h) + h.command('node', cmd(identifier, 1)) + first = h.stream('node', 'a'*32) + now[0] += .2 + repeated = h.stream('node', 'a'*32) + assert repeated['command'] == first['command'] + assert repeated['control_clock']['monotonic_ms'] > first['control_clock']['monotonic_ms'] + assert first['command']['expires_mono_ms'] == pytest.approx(10400) + now[0] += .201 + assert h.stream('node', 'a'*32)['command'] is None + with pytest.raises(ValueError): h.command('node', cmd(identifier, 2)) + +def test_stream_delivers_new_intent_without_waiting_for_telemetry(hub): + h, now = hub + identifier = arm(h) + h.command('node', cmd(identifier, 1)) + h.command('node', cmd(identifier, 2, left=-1)) + assert h.stream('node', 'a'*32)['command']['left'] == -1 + assert h.stream('node', 'b'*32)['command'] is None + h.command('node', cmd(identifier, 3, stop=True)) + assert h.stream('node', 'a'*32)['command'] is None + +def test_new_core_has_distinct_clock_epoch(hub): + h, _ = hub + assert h.stream('node', 'a'*32)['control_clock']['instance'] != RoverControl().clock_id + +def test_stream_retires_control_if_return_telemetry_is_lost(hub): + h, now = hub + identifier = arm(h) + for seq in range(1, 13): + h.command('node', cmd(identifier, seq)) + now[0] += .09 + assert h.stream('node', 'a'*32)['command'] is None + with pytest.raises(ValueError): h.command('node', cmd(identifier, 13)) + +def test_intent_wait_wakes_on_update_and_does_not_miss_prior_update(hub): + import threading + from unittest.mock import patch + h, _ = hub + identifier = arm(h) + previous = h.stream('node', 'a'*32)['command'] + entered, done = threading.Event(), threading.Event() + def waiter(): + entered.set() + h.wait_for_intent('node', 'a'*32, previous, timeout=2) + done.set() + worker = threading.Thread(target=waiter) + worker.start() + assert entered.wait(1) + h.command('node', cmd(identifier, 1)) + assert done.wait(.5), 'new intent waited for the periodic keepalive' + worker.join(2) + # A command arriving after socket write but before wait must not be lost. + with patch.object(h.changed, 'wait', side_effect=AssertionError('missed prior update')): + h.wait_for_intent('node', 'a'*32, previous, timeout=2) + latest = h.stream('node', 'a'*32)['command'] + h.command('node', cmd(identifier, 2, stop=True)) + with patch.object(h.changed, 'wait', side_effect=AssertionError('missed stop')): + h.wait_for_intent('node', 'a'*32, latest, timeout=2) diff --git a/tests/fleet/test_rover_stream_transport.py b/tests/fleet/test_rover_stream_transport.py new file mode 100644 index 0000000..56e23a6 --- /dev/null +++ b/tests/fleet/test_rover_stream_transport.py @@ -0,0 +1,52 @@ +"""Exercise HTTP framing and per-frame binding checks without a real board.""" +import http.client +import json +import threading +import time +from http.server import ThreadingHTTPServer +from types import SimpleNamespace + +from k1link.fleet.transport import NodeChannelHandler + + +def test_stream_rechecks_binding_and_closes_when_revoked(): + calls = [] + + def receive(certificate, path, body, *, address): + calls.append((certificate, path, body, address)) + if len(calls) == 3: + return 410, {"error": "Binding revoked"} + return 200, {"watch": True, "command": {"sequence": len(calls)}} + + class Handler(NodeChannelHandler): + def setup(self): + super().setup() + # Only TLS certificate extraction is substituted. HTTP reads, + # writes, streaming, loop and reauthentication use production code. + self.connection = SimpleNamespace(getpeercert=lambda **_: b"synthetic-cert") + + server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) + server.address = "127.0.0.1" + server.registry = SimpleNamespace(receive=receive, stop=threading.Event(), + rover_control=SimpleNamespace(wait_for_intent=lambda *_: time.sleep(.01))) + worker = threading.Thread(target=server.serve_forever, daemon=True) + worker.start() + client = http.client.HTTPConnection(*server.server_address, timeout=2) + try: + body = {"relay_id": "a" * 32, "node_id": "synthetic-node"} + client.request("POST", "/v1/node/rover-stream", json.dumps(body), + {"Content-Type": "application/json"}) + response = client.getresponse() + assert response.status == 200 + assert response.getheader("Content-Type") == "application/x-ndjson" + assert response.getheader("Content-Length") is None + frames = [json.loads(line) for line in response.read().splitlines()] + assert [frame["command"]["sequence"] for frame in frames] == [1, 2] + assert len(calls) == 3 + assert all(call == (b"synthetic-cert", "/v1/node/rover-stream", body, "127.0.0.1") for call in calls) + finally: + client.close() + server.shutdown() + server.server_close() + worker.join(timeout=2) + assert not worker.is_alive()