feat: add fabric session protocol with topic remapping and HAL integration

.gitignore

@@ -1,2 +1,2 @@
 build/
-.vscode/settings.json
+.vscode/settings.json

.vscode/tasks.json

@@ -23,7 +23,7 @@
       "args": [
         "build",
         "-o", "${workspaceFolder}/build/devicecode.elf",
-        "-stack-size=3KB",
+        "-stack-size=8KB",
         "-serial=none",
         "-target=pico2",
         "-tags", "pico_bb_proto_1",

README.md

@@ -4,7 +4,7 @@
 tinygo flash -stack-size=3KB -monitor -scheduler tasks -target=pico -tags "pico_bb_proto_1" main.go
 
 ## Flashing ISOC Power Board via USB port on Pico2
-tinygo flash -stack-size=3KB -monitor -scheduler tasks -target=pico2 -tags "pico_bb_proto_1" main.go
+tinygo flash -stack-size=8KB -monitor -scheduler tasks -target=pico2 -tags "pico_bb_proto_1" main.go
 
 -------------------
 

go.mod

@@ -1,11 +1,14 @@
 module devicecode-go
 
-go 1.25.0
+go 1.25.1
 
 require (
+	pico2-a-b v0.0.0
 	github.com/jangala-dev/tinygo-uartx v0.0.0-20251028085354-58b6258234b3
 	golang.org/x/exp v0.0.0-20251002181428-27f1f14c8bb9
 	tinygo.org/x/drivers v0.33.0
 )
 
 require github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
+
+replace pico2-a-b => ../pico2-a-b

go.sum

@@ -1,7 +1,5 @@
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
 github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
-github.com/jangala-dev/tinygo-uartx v0.0.0-20251008020047-bc80b114e3cc h1:HU2VI0lw5wlu1rUgjzSuVH7IWQMNdZEbpDaoxCTVMmY=
-github.com/jangala-dev/tinygo-uartx v0.0.0-20251008020047-bc80b114e3cc/go.mod h1:e3HxjGzBZBIsn/oYvWr707ug3IbkglEyivyYVxHRph4=
 github.com/jangala-dev/tinygo-uartx v0.0.0-20251028085354-58b6258234b3 h1:b6mCDQEeeICoGpsbKyh/kfIRnr2DMK/wACLLi0t8uoU=
 github.com/jangala-dev/tinygo-uartx v0.0.0-20251028085354-58b6258234b3/go.mod h1:e3HxjGzBZBIsn/oYvWr707ug3IbkglEyivyYVxHRph4=
 golang.org/x/exp v0.0.0-20251002181428-27f1f14c8bb9 h1:TQwNpfvNkxAVlItJf6Cr5JTsVZoC/Sj7K3OZv2Pc14A=

main.go

@@ -27,7 +27,12 @@ func main() {
 	ctx := context.Background()
 
 	log.Println("[main] bootstrapping bus …")
-	b := bus.NewBus(3, "+", "#")
+	// Queue length must cover the retained replay burst when fabric
+	// subscribes to wildcard export patterns (hal/cap/env/#,
+	// hal/cap/power/#). Each capability publishes retained info +
+	// status + value; pico_bb_proto_1 has ~26 retained topics across
+	// env and power domains. 32 provides margin for growth.
+	b := bus.NewBus(32, "+", "#")
 	halConn := b.NewConnection("hal")
 	uiConn := b.NewConnection("ui")
 
@@ -43,7 +48,7 @@ func main() {
 	}
 
 	// Reactor
-	r := reactor.NewReactor(uiConn)
+	r := reactor.NewReactor(b, uiConn)
 	r.Run(ctx)
 }
 

services/fabric/config.go

@@ -0,0 +1,124 @@
+package fabric
+
+import (
+	"encoding/json"
+
+	"devicecode-go/types"
+)
+
+// decodeHALConfig extracts a HALConfig from an arbitrary payload,
+// normalizing Lua empty-table encoding ({} → []) for known slice fields.
+func decodeHALConfig(payload any) (types.HALConfig, string) {
+	switch v := payload.(type) {
+	case types.HALConfig:
+		return v, ""
+	case *types.HALConfig:
+		if v == nil {
+			return types.HALConfig{}, "nil_hal_config"
+		}
+		return *v, ""
+	case json.RawMessage:
+		return decodeHALConfigBytes(v)
+	case []byte:
+		return decodeHALConfigBytes(v)
+	default:
+		b, err := json.Marshal(v)
+		if err != nil {
+			return types.HALConfig{}, "payload_marshal_failed: " + err.Error()
+		}
+		return decodeHALConfigBytes(b)
+	}
+}
+
+func decodeHALConfigBytes(b []byte) (types.HALConfig, string) {
+	var probe map[string]json.RawMessage
+	if err := json.Unmarshal(b, &probe); err != nil {
+		return types.HALConfig{}, "json_unmarshal_failed: " + err.Error() + "; raw=" + truncateRawJSON(b)
+	}
+	if _, ok := probe["devices"]; !ok {
+		return types.HALConfig{}, "missing_devices_field; raw=" + truncateRawJSON(b)
+	}
+
+	// Lua encodes empty tables as {} (object) not [] (array).
+	// Normalize known slice fields so Go unmarshal accepts them.
+	for _, key := range []string{"devices", "pollers"} {
+		if raw, ok := probe[key]; ok && len(raw) == 2 && raw[0] == '{' && raw[1] == '}' {
+			probe[key] = json.RawMessage("[]")
+		}
+	}
+	fixed, err := json.Marshal(probe)
+	if err != nil {
+		return types.HALConfig{}, "normalize_failed: " + err.Error()
+	}
+
+	var out types.HALConfig
+	if err := json.Unmarshal(fixed, &out); err != nil {
+		return types.HALConfig{}, "hal_config_unmarshal_failed: " + err.Error() + "; raw=" + truncateRawJSON(fixed)
+	}
+	return out, ""
+}
+
+func decodeHALState(payload any) (types.HALState, bool) {
+	switch v := payload.(type) {
+	case types.HALState:
+		return v, true
+	case *types.HALState:
+		if v == nil {
+			return types.HALState{}, false
+		}
+		return *v, true
+	case json.RawMessage:
+		var out types.HALState
+		return out, json.Unmarshal(v, &out) == nil
+	case []byte:
+		var out types.HALState
+		return out, json.Unmarshal(v, &out) == nil
+	default:
+		b, err := json.Marshal(v)
+		if err != nil {
+			return types.HALState{}, false
+		}
+		var out types.HALState
+		return out, json.Unmarshal(b, &out) == nil
+	}
+}
+
+func decodePayload(payload any) any {
+	switch v := payload.(type) {
+	case nil:
+		return nil
+	case json.RawMessage:
+		if len(v) == 0 {
+			return nil
+		}
+		var out any
+		if err := json.Unmarshal(v, &out); err == nil {
+			return out
+		}
+		return []byte(v)
+	case []byte:
+		if len(v) == 0 {
+			return nil
+		}
+		var out any
+		if err := json.Unmarshal(v, &out); err == nil {
+			return out
+		}
+		cp := make([]byte, len(v))
+		copy(cp, v)
+		return cp
+	default:
+		return v
+	}
+}
+
+func truncateRawJSON(b []byte) string {
+	if len(b) == 0 {
+		return ""
+	}
+	const max = 160
+	if len(b) <= max {
+		return string(b)
+	}
+	return string(b[:max]) + "..."
+}

services/fabric/fabric.go

@@ -0,0 +1,119 @@
+package fabric
+
+import (
+	"context"
+	"sync/atomic"
+	"time"
+
+	"devicecode-go/bus"
+	"devicecode-go/x/strconvx"
+)
+
+// Transport abstracts the byte stream as newline-delimited JSON lines.
+type Transport interface {
+	ReadLine() ([]byte, error)
+	WriteLine(data []byte) error
+	Close() error
+}
+
+const protoVersion = 1
+const defaultLinkID = "mcu0"
+
+// LinkConfig carries the fabric link parameters that the CM5 publishes
+// alongside its own session/transfer-mgr instances. Mirrors the relevant
+// keys in `bigbox-v1-cm-2.json` `service.fabric.links.<id>` for the
+// MCU-facing link. Missing fields fall back to release defaults via
+// applyDefaults so callers can pass `LinkConfig{}` to mean "release".
+type LinkConfig struct {
+	// ChunkSize is the expected raw-byte payload per xfer_chunk. The MCU
+	// is receive-only for transfers, so this is informational/validation
+	// only on the Go side. Release: 2048 bytes.
+	ChunkSize uint32
+	// PhaseTimeout is the idle-chunk watchdog: an active inbound transfer
+	// is aborted with reason="timeout" if no xfer_chunk arrives within
+	// this window. Mirrors transfer_mgr.lua's `phase_timeout`.
+	// Release: 15s.
+	PhaseTimeout time.Duration
+	// PingInterval drives the unconditional outbound ping cadence after
+	// the link is established (`session_ctl.lua` resets next_ping_at =
+	// now + ping_interval after every send; not TX-activity-based).
+	// Release: 10s.
+	PingInterval time.Duration
+	// LivenessTimeout tears the link down if no frame arrives within
+	// this window once established. Mirrors session_ctl.lua's
+	// liveness_timeout_s. Release: 30s.
+	LivenessTimeout time.Duration
+	// MaxInboundHelpers caps the number of in-flight inbound RPC calls.
+	// Excess inbound calls reply `{ok=false, err="busy"}` per
+	// rpc_bridge.lua's `spawn_local_call_helper`. Lua default is 64
+	// (falls back to max_pending_calls); we keep that for parity.
+	MaxInboundHelpers int
+	// RPCQuantum and BulkQuantum control the writer's weighted
+	// round-robin between the rpc and bulk lanes after the control
+	// lane drains. Mirrors writer.lua's lane scheduler. Release: 4 and 1.
+	RPCQuantum  int
+	BulkQuantum int
+}
+
+func DefaultLinkConfig() LinkConfig {
+	return LinkConfig{
+		ChunkSize:         2048,
+		PhaseTimeout:      15 * time.Second,
+		PingInterval:      10 * time.Second,
+		LivenessTimeout:   30 * time.Second,
+		MaxInboundHelpers: 64,
+		RPCQuantum:        4,
+		BulkQuantum:       1,
+	}
+}
+
+func (c *LinkConfig) applyDefaults() {
+	d := DefaultLinkConfig()
+	if c.ChunkSize == 0 {
+		c.ChunkSize = d.ChunkSize
+	}
+	if c.PhaseTimeout == 0 {
+		c.PhaseTimeout = d.PhaseTimeout
+	}
+	if c.PingInterval == 0 {
+		c.PingInterval = d.PingInterval
+	}
+	if c.LivenessTimeout == 0 {
+		c.LivenessTimeout = d.LivenessTimeout
+	}
+	if c.MaxInboundHelpers == 0 {
+		c.MaxInboundHelpers = d.MaxInboundHelpers
+	}
+	if c.RPCQuantum == 0 {
+		c.RPCQuantum = d.RPCQuantum
+	}
+	if c.BulkQuantum == 0 {
+		c.BulkQuantum = d.BulkQuantum
+	}
+}
+
+var nextSessionID atomic.Uint64
+
+func newLocalSID() string {
+	return "mcu-sid-" + strconvx.Utoa64(nextSessionID.Add(1))
+}
+
+// Run starts the fabric session. Blocks until ctx is cancelled or the
+// transport returns an unrecoverable error. The MCU is a hello
+// responder (CM5 always initiates hello/hello_ack), but otherwise
+// runs the symmetric session_ctl semantics: once established, it
+// sends pings every PingInterval and tears the link down if no frame
+// arrives within LivenessTimeout. Mirrors session_ctl.lua at
+// devicecode-lua@2c88090.
+func Run(ctx context.Context, tr Transport, conn *bus.Connection, nodeID, peerID string, cfg LinkConfig) {
+	s := session{
+		linkID:   defaultLinkID,
+		nodeID:   nodeID,
+		peerID:   peerID,
+		localSID: newLocalSID(),
+		tr:       tr,
+		conn:     conn,
+		cfg:      cfg,
+	}
+	s.run(ctx)
+}