feat(routing-policy): add experimental 'coupled' preset (MPTCP-style coupled congestion control)

pkg/router/policy/preset/names.go

@@ -15,6 +15,7 @@ func Names() []string {
 		"latency-adaptive",
 		"elastic-mux",
 		"probe-and-prune",
+		"coupled",
 		"adaptive",
 		"geo-avoid",
 		"transport-diverse",

pkg/router/policy/preset/preset.go

@@ -108,6 +108,8 @@ func Decide(name string, ctx Context, cands []Candidate) Spec {
 		return decideElasticMux(ctx)
 	case "probe-and-prune":
 		return decideProbeAndPrune(ctx)
+	case "coupled":
+		return decideCoupled(ctx)
 	case "adaptive":
 		return decideAdaptive(ctx, cands)
 	case "app-mux":
@@ -234,6 +236,36 @@ func decideProbeAndPrune(ctx Context) Spec {
 	return Spec{}
 }
 
+// decideCoupled is the EXPERIMENTAL "coupled" preset's decide logic — a
+// multipath policy inspired by MPTCP's coupled congestion control (LIA/OLIA).
+// It provisions a modest symmetric mux (coupledMux legs) over the multi-hop
+// overlay and asks the host to weight bytes toward the best legs (Distribution
+// "auto" = inverse-latency), NOT spread them equally across a lossy path. The
+// on_tick controller (tickCoupled) then "couples" the aggregate: it grows the
+// active set CAUTIOUSLY — at most one promote per tick, and only when NO active
+// leg shows rising loss (LIA's coupled increase) — and sheds the WORST leg
+// (highest loss, latency-tiebroken) the moment congestion appears, so total
+// aggressiveness stays bounded and traffic concentrates on the good legs.
+// Latency-sensitive chat stays a single lean route; non-target apps inherit the
+// visor default.
+func decideCoupled(ctx Context) Spec {
+	switch ctx.App {
+	case "skychat", "skychat-client":
+		return Spec{Mux: 1}
+	case "vpn-client", "skysocks-client", "skynet-client":
+		return Spec{
+			Mux:                     coupledMux,
+			MinHops:                 2,
+			RotationIntervalSeconds: 20,
+			// Weight bytes toward the lowest-latency/lowest-loss legs (the
+			// "coupled" bias) rather than round-robin's equal spread across a
+			// congested leg.
+			Distribution: "auto",
+		}
+	}
+	return Spec{}
+}
+
 // decideAdaptive is the COMPOSITE "adaptive" preset's decide logic — the
 // intended converged default (the config generator wires "preset:adaptive").
 //

pkg/router/policy/preset/preset_test.go

@@ -532,3 +532,91 @@ func TestEngine_OnTick_LedbatGrowsWhenNoQueuing(t *testing.T) {
 		t.Errorf("grow must not shed; got %+v", got)
 	}
 }
+
+// TestDecide_Coupled pins the coupled preset's decide shape: a modest symmetric
+// mux over the multi-hop overlay with best-leg (auto) byte weighting for target
+// apps, a single lean route for chat, defaults for everything else.
+func TestDecide_Coupled(t *testing.T) {
+	if got := Decide("coupled", Context{App: "skysocks-client"}, nil); !reflect.DeepEqual(
+		got, Spec{Mux: 4, MinHops: 2, RotationIntervalSeconds: 20, Distribution: "auto"}) {
+		t.Errorf("coupled/proxy: Decide=%+v", got)
+	}
+	if got := Decide("coupled", Context{App: "skychat"}, nil); !reflect.DeepEqual(got, Spec{Mux: 1}) {
+		t.Errorf("coupled/chat: Decide=%+v want {Mux:1}", got)
+	}
+	if got := Decide("coupled", Context{App: "other"}, nil); !reflect.DeepEqual(got, Spec{}) {
+		t.Errorf("coupled/other: Decide=%+v want {}", got)
+	}
+}
+
+// TestEngine_OnTick_CoupledShedsWorstOnLoss drives the coupled controller with a
+// clean 4-wide active set, then makes one leg's retransmits rise, and asserts the
+// COUPLED DECREASE: the worst (lossy) leg is shed to standby — a lone demote,
+// concentrating traffic on the good legs.
+func TestEngine_OnTick_CoupledShedsWorstOnLoss(t *testing.T) {
+	e := New()
+	base := func(retrans2 uint64) []LegInfo {
+		return []LegInfo{
+			{Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 50, Alive: true},
+			{Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 50, Alive: true},
+			{Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 50, Alive: true, Retransmits: retrans2},
+			{Index: 3, TransportID: "t3", Kind: "stcpr", LatencyMs: 50, Alive: true},
+		}
+	}
+	// Tick 1 establishes the retransmit baseline; a clean 4-wide set at the ceiling
+	// is a no-op (no grow, no loss to shed).
+	if got := e.OnTick("coupled", base(0)); !reflect.DeepEqual(got, RotationAction{}) {
+		t.Fatalf("baseline tick must be a no-op; got %+v", got)
+	}
+	// Tick 2: leg 2's retransmits jumped → rising loss → shed the worst leg (2).
+	got := e.OnTick("coupled", base(100))
+	if len(got.DemoteToStandby) != 1 || got.DemoteToStandby[0] != 2 {
+		t.Errorf("coupled must shed the lossy leg (2) to standby; got %+v", got)
+	}
+	if got.AddLeg || len(got.PromoteFromStandby) != 0 || len(got.DropLegs) != 0 {
+		t.Errorf("coupled decrease must be a lone demote (no grow/drop); got %+v", got)
+	}
+}
+
+// TestEngine_OnTick_CoupledCautiousPromoteWhenClean asserts the COUPLED INCREASE:
+// a loss-free active set below the ceiling with a warm spare promotes AT MOST one
+// leg (LIA's cautious coupled increase).
+func TestEngine_OnTick_CoupledCautiousPromoteWhenClean(t *testing.T) {
+	e := New()
+	legs := []LegInfo{
+		{Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 50, Alive: true},
+		{Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 50, Alive: true},
+		{Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 50, Alive: true},
+		{Index: 3, TransportID: "s0", Kind: "stcpr", LatencyMs: 45, Alive: true, Standby: true},
+	}
+	got := e.OnTick("coupled", legs)
+	if len(got.PromoteFromStandby) != 1 || got.PromoteFromStandby[0] != 3 {
+		t.Errorf("clean active set below the ceiling must cautiously promote the spare; got %+v", got)
+	}
+	if got.AddLeg || len(got.DemoteToStandby) != 0 || len(got.DropLegs) != 0 {
+		t.Errorf("cautious increase must be a lone promote; got %+v", got)
+	}
+}
+
+// TestEngine_OnTick_CoupledNoGrowUnderLoss is the coupling property: even with a
+// warm spare available and the active set below the ceiling, rising loss on an
+// active leg FORBIDS growth — the controller sheds the lossy leg instead of
+// promoting. (The loss baseline is seeded white-box so the rise reads on the very
+// first tick, without an intervening clean tick consuming the spare.)
+func TestEngine_OnTick_CoupledNoGrowUnderLoss(t *testing.T) {
+	e := New()
+	e.coupledPrevRetransmits["t1"] = 0
+	legs := []LegInfo{
+		{Index: 0, TransportID: "t0", Kind: "stcpr", LatencyMs: 50, Alive: true},
+		{Index: 1, TransportID: "t1", Kind: "stcpr", LatencyMs: 50, Alive: true, Retransmits: 100},
+		{Index: 2, TransportID: "t2", Kind: "stcpr", LatencyMs: 50, Alive: true},
+		{Index: 3, TransportID: "s0", Kind: "stcpr", LatencyMs: 45, Alive: true, Standby: true},
+	}
+	got := e.OnTick("coupled", legs)
+	if len(got.PromoteFromStandby) != 0 {
+		t.Errorf("coupled must NOT grow while an active leg shows rising loss (a spare is available); got %+v", got)
+	}
+	if len(got.DemoteToStandby) != 1 || got.DemoteToStandby[0] != 1 {
+		t.Errorf("coupled should shed the lossy leg (1) instead; got %+v", got)
+	}
+}

pkg/router/policy/preset/tick.go

@@ -60,6 +60,13 @@ type Engine struct {
 	ledbatEWMA map[string]float64 // per-leg EWMA-smoothed one-way-ish delay (ms)
 	ledbatBase map[string]float64 // per-leg base (running-min smoothed) delay (ms)
 	ledbatSeen map[string]bool    // transport_ids present this tick (for GC)
+
+	// coupled (MPTCP-style coupled congestion control)
+	coupledLatEWMA         map[string]float64 // per-leg smoothed latency (worst-leg tiebreak)
+	coupledPrevRetransmits map[string]uint64  // per-leg last-tick retransmit counter (loss delta)
+	coupledPrevSent        map[string]uint64  // per-leg last-tick sent bytes (packet-normalize loss)
+	coupledSeen            map[string]bool    // transport_ids present this tick (stale-state GC)
+	coupledCooldown        int                // ticks held steady after a grow/shed (anti-churn)
 }
 
 // New returns an Engine with initialized state maps and the
@@ -86,6 +93,11 @@ func New() *Engine {
 		ledbatEWMA:      map[string]float64{},
 		ledbatBase:      map[string]float64{},
 		ledbatSeen:      map[string]bool{},
+
+		coupledLatEWMA:         map[string]float64{},
+		coupledPrevRetransmits: map[string]uint64{},
+		coupledPrevSent:        map[string]uint64{},
+		coupledSeen:            map[string]bool{},
 	}
 }
 
@@ -103,6 +115,8 @@ func (e *Engine) OnTick(name string, legs []LegInfo) RotationAction {
 		return e.tickElasticMux(legs)
 	case "probe-and-prune":
 		return e.tickProbeAndPrune(legs)
+	case "coupled":
+		return e.tickCoupled(legs)
 	case "adaptive":
 		return e.tickAdaptive(legs)
 	case "ledbat":
@@ -954,3 +968,155 @@ func (e *Engine) tickLedbat(legs []LegInfo) RotationAction {
 	}
 	return RotationAction{}
 }
+
+// --- coupled (MPTCP-style coupled congestion control) ---
+
+const (
+	// coupledMux is the modest symmetric mux the coupled preset provisions and
+	// the CEILING the cautious coupled-increase grows the active set back up to.
+	coupledMux = 4
+	// coupledFloor is the minimum active width the coupled-decrease will never
+	// shed below — the aggregate keeps at least this many legs carrying traffic.
+	coupledFloor = 2
+	// coupledAlpha smooths per-leg latency (the worst-leg tiebreak signal).
+	coupledAlpha = 0.3
+	// coupledCooldownTicks holds the active set steady for a few ticks after any
+	// grow/shed so a transient loss blip can't churn the mux every tick.
+	coupledCooldownTicks = 3
+)
+
+// tickCoupled is the coupled congestion controller. The two congestion signals
+// are per-leg: the RETRANSMIT delta since last tick (loss, packet-normalized by
+// the sent-bytes delta) and the EWMA-smoothed latency. It arbitrates at most one
+// structural change per tick, in priority order:
+//
+//	(0) no active legs        -> bring one up (promote a spare, else dial)
+//	(1) active < floor        -> restore the floor (recovery; bypasses cooldown)
+//	    cooldown active       -> hold steady (anti-churn)
+//	(2) any active leg's loss RISING -> COUPLED DECREASE: shed the WORST leg
+//	    (highest loss, latency-tiebroken; never leg 0, never below the floor),
+//	    concentrating traffic on the good legs instead of equal-spreading it
+//	    across a lossy one.
+//	(3) NO active leg showing rising loss AND active < ceiling -> COUPLED
+//	    INCREASE (LIA-cautious): promote AT MOST one warm spare.
+//
+// Because a shed fires the instant loss appears and a grow fires only when the
+// WHOLE active set is loss-free, aggregate aggressiveness stays bounded — the
+// "coupling" property. Deterministic (no time.Now/rand) for wasm parity.
+func (e *Engine) tickCoupled(legs []LegInfo) RotationAction {
+	for k := range e.coupledSeen {
+		delete(e.coupledSeen, k)
+	}
+	activeCount := 0
+	standbyCount := 0
+	promotable := -1
+	worstIdx := -1
+	worstScore := -1.0
+	anyRisingLoss := false
+	for _, l := range legs {
+		tid := l.TransportID
+		if tid != "" {
+			e.coupledSeen[tid] = true
+		}
+		if !l.Alive {
+			continue
+		}
+		if l.Standby {
+			standbyCount++
+			if promotable == -1 || l.Index < promotable {
+				promotable = l.Index
+			}
+			continue
+		}
+		activeCount++
+		if tid == "" {
+			continue
+		}
+		if l.LatencyMs > 0 {
+			sample := float64(l.LatencyMs)
+			if prev, ok := e.coupledLatEWMA[tid]; ok {
+				e.coupledLatEWMA[tid] = coupledAlpha*sample + (1-coupledAlpha)*prev
+			} else {
+				e.coupledLatEWMA[tid] = sample
+			}
+		}
+		var retransDelta, sentDelta uint64
+		if prev, ok := e.coupledPrevRetransmits[tid]; ok && l.Retransmits >= prev {
+			retransDelta = l.Retransmits - prev
+		}
+		if prev, ok := e.coupledPrevSent[tid]; ok && l.SentBytes >= prev {
+			sentDelta = l.SentBytes - prev
+		}
+		e.coupledPrevRetransmits[tid] = l.Retransmits
+		e.coupledPrevSent[tid] = l.SentBytes
+		if retransDelta > 0 {
+			anyRisingLoss = true
+		}
+		// Loss ratio = retransmits / sent-segments this tick (a ~1KB segment unit;
+		// +1 avoids divide-by-zero when a leg sent nothing). Loss dominates the
+		// worst-leg score; smoothed latency breaks ties. Leg 0 (the primary) is
+		// never a shed candidate.
+		sentSegs := sentDelta / 1024
+		lossRatio := float64(retransDelta) / float64(sentSegs+1)
+		score := lossRatio*1e6 + e.coupledLatEWMA[tid]
+		if l.Index != 0 && score > worstScore {
+			worstScore = score
+			worstIdx = l.Index
+		}
+	}
+	for tid := range e.coupledLatEWMA {
+		if !e.coupledSeen[tid] {
+			delete(e.coupledLatEWMA, tid)
+		}
+	}
+	for tid := range e.coupledPrevRetransmits {
+		if !e.coupledSeen[tid] {
+			delete(e.coupledPrevRetransmits, tid)
+		}
+	}
+	for tid := range e.coupledPrevSent {
+		if !e.coupledSeen[tid] {
+			delete(e.coupledPrevSent, tid)
+		}
+	}
+
+	if e.coupledCooldown > 0 {
+		e.coupledCooldown--
+	}
+
+	// (0) No active legs — bring capacity up (recovery, bypasses cooldown).
+	if activeCount == 0 {
+		if promotable >= 0 {
+			return RotationAction{PromoteFromStandby: []int{promotable}}
+		}
+		return RotationAction{AddLeg: true}
+	}
+	// (1) Restore the floor if an active leg dropped (recovery, bypasses cooldown).
+	if activeCount < coupledFloor {
+		if promotable >= 0 {
+			return RotationAction{PromoteFromStandby: []int{promotable}}
+		}
+		return RotationAction{AddLeg: true}
+	}
+	// Anti-churn: hold the set steady during cooldown.
+	if e.coupledCooldown > 0 {
+		return RotationAction{}
+	}
+
+	// (2) Coupled DECREASE: congestion (rising loss on any active leg) -> shed the
+	// worst leg so aggregate aggressiveness stays bounded.
+	if anyRisingLoss && activeCount > coupledFloor && worstIdx > 0 {
+		e.coupledCooldown = coupledCooldownTicks
+		if standbyCount < nodipStandbyMax {
+			return RotationAction{DemoteToStandby: []int{worstIdx}}
+		}
+		return RotationAction{DropLegs: []int{worstIdx}}
+	}
+	// (3) Coupled INCREASE (LIA-cautious): the whole active set is loss-free and we
+	// are below the ceiling -> promote at most one warm spare.
+	if !anyRisingLoss && activeCount < coupledMux && promotable >= 0 {
+		e.coupledCooldown = coupledCooldownTicks
+		return RotationAction{PromoteFromStandby: []int{promotable}}
+	}
+	return RotationAction{}
+}

pkg/router/policy/wasm/presets/manifest.json

@@ -4,6 +4,7 @@
   "latency-adaptive": "latency-adaptive — mux=4 multi-hop that evicts the slowest leg each 30s (when its EWMA-smoothed latency is a >=1.5x-median outlier) until the leg set converges to low-latency disjoint paths, then holds (hysteresis-damped; no churn once converged).",
   "elastic-mux": "elastic-mux — AIMD scaling of the mux leg count to load: grows a leg (up to 6) when the group is saturated, releases one (down to 2) when idle.",
   "probe-and-prune": "probe-and-prune — periodically adds one speculative leg over a fresh path, observes its EWMA latency a few ticks, and keeps it only if it beats the current worst leg (continuous explore/exploit).",
+  "coupled": "coupled (experimental) — MPTCP-style coupled congestion control over a modest mux (4): weights bytes toward the best legs (auto), grows the active set cautiously (at most one promote per tick, only when NO active leg shows rising loss — LIA's coupled increase), and sheds the worst (highest-loss, latency-tiebroken) leg the instant congestion appears, so aggregate aggressiveness stays bounded and traffic concentrates on the good legs.",
   "adaptive": "adaptive — the app-agnostic performance+stability default (chat excepted): a single lean forward leg and a reverse mux that stays at ONE healthy leg for interactive/idle flows and only widens under sustained bulk load; holds an always-on warm-standby reserve for instant dip-free promotion, keeps gross-latency-outlier and dead legs OUT of the active mux, and rate-limits reshapes (hysteresis + cooldown) so the active set stays stable — one arbitrated action per tick.",
   "geo-avoid": "geo-avoid — chooses the lowest-latency route whose hops transit NONE of the countries in cli_overrides.avoid_geo (comma-separated ISO codes); defers rather than pick a violating path when none is clean.",
   "transport-diverse": "transport-diverse — chooses the route whose hops span the most distinct transport types (ties broken by latency), so no single transport technology failing takes the path down; seeds a mux of 2.",

pkg/router/policy/wasm/presets/parity_test.go

@@ -154,6 +154,8 @@ func TestDecideParity_NativeMatchesWazero(t *testing.T) {
 		{"latency-adaptive/other", "latency-adaptive", policy.RoutingContext{App: "skychat"}, nil},
 		{"elastic-mux", "elastic-mux", policy.RoutingContext{App: "skysocks-client"}, nil},
 		{"probe-and-prune", "probe-and-prune", policy.RoutingContext{App: "skynet-client"}, nil},
+		{"coupled", "coupled", policy.RoutingContext{App: "skysocks-client"}, nil},
+		{"coupled/chat", "coupled", policy.RoutingContext{App: "skychat"}, nil},
 		{"adaptive", "adaptive", policy.RoutingContext{App: "vpn-client"}, nil},
 		// adaptive with real transport-kind metadata: both paths must seed the
 		// most transport-diverse forward candidate (cand b, 2 distinct kinds).
@@ -271,11 +273,30 @@ func TestTickParity_NativeMatchesWazero(t *testing.T) {
 		{leg(0, "a", "stcpr", 40, true, false, 0), leg(1, "b", "stcpr", 40, true, true, 0), leg(2, "c", "stcpr", 40, true, true, 0)},
 	}
 
+	// coupled: a clean 4-wide set, then rising loss on one leg (shed), a couple of
+	// cooldown ticks, then a clean below-ceiling set with a warm spare (cautious
+	// promote) — exercises both coupled-decrease and coupled-increase.
+	rl := func(idx int, tid string, lat int, retrans uint64) policy.LegInfo {
+		return policy.LegInfo{Index: idx, TransportID: tid, Kind: "stcpr", LatencyMs: lat, Alive: true, Retransmits: retrans}
+	}
+	sb := func(idx int, tid string, lat int) policy.LegInfo {
+		return policy.LegInfo{Index: idx, TransportID: tid, Kind: "stcpr", LatencyMs: lat, Alive: true, Standby: true}
+	}
+	cpl := [][]policy.LegInfo{
+		{rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(2, "c", 50, 0), rl(3, "d", 50, 0)},
+		{rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(2, "c", 50, 500), rl(3, "d", 50, 0)},
+		{rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)},
+		{rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)},
+		{rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)},
+		{rl(0, "a", 50, 0), rl(1, "b", 50, 0), rl(3, "d", 50, 0), sb(2, "c", 50)},
+	}
+
 	cases := []tickCase{
 		{"rotating-bw", "rotating-bw", rbw},
 		{"latency-adaptive", "latency-adaptive", la},
 		{"elastic-mux", "elastic-mux", em},
 		{"probe-and-prune", "probe-and-prune", pp},
+		{"coupled", "coupled", cpl},
 		{"adaptive", "adaptive", ad},
 		{"ledbat", "ledbat", lb},
 	}