Piik/internal/server/route/controller_test.go

package route

// Oracle port of tests/room-route-controller.test.ts. Each test keeps the TS
// title and line as a comment; every nowMs is the explicit literal the TS
// used, so nothing here needs a clock.

import (
	"encoding/json"
	"fmt"
	"regexp"
	"slices"
	"strings"
	"testing"

	"github.com/TNTcraftHIM/Piik/internal/server/protocol"
)

// TS 359: keeps one diagnostic label until a departed relay is finally pruned
func TestKeepsOneDiagnosticLabelUntilDepartedRelayIsPruned(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(A, "b_from_a"))
	label := routes.DiagnosticParticipantLabel(A)
	if !regexp.MustCompile(`^viewer-\d+$`).MatchString(label) {
		t.Fatalf("label %q is not viewer-<n>", label)
	}

	routes.UpsertParticipant(viewerInput(A, "a_replacement_session", 1), nil)
	eq(t, routes.DiagnosticParticipantLabel(A), label)
	eq(t, routes.DisconnectSession(A, "a_replacement_session"), true)
	eq(t, routes.DiagnosticParticipantLabel(A), label)
	eq(t, routes.ConfirmDeparture(A, ms(0)), true)
	eq(t, routes.DiagnosticParticipantLabel(A), label)

	repair := must(t, routes.Reconcile(1).Operation)
	eq(t, repair.ChildPeerID, B)
	eq(t, repair.Reason, DemandParentDeparted)
	commitCurrent(t, routes, 2, "b_from_host")
	pruned := routes.Reconcile(4)
	if !slices.Contains(pruned.RemovedPeerIDs, A) {
		t.Fatalf("removedPeerIds %v does not contain A", pruned.RemovedPeerIDs)
	}
	eq(t, routes.DiagnosticParticipantLabel(A), "viewer-unknown")

	addViewer(routes, C, 0, nil)
	if routes.DiagnosticParticipantLabel(C) == label {
		t.Fatalf("C reused label %q", label)
	}
}

// TS 394: clears an unknown sender run without rearming its healthy baseline
func TestClearsUnknownSenderRunWithoutRearmingHealthyBaseline(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	for _, seed := range [][2]string{{A, "a_from_host"}, {B, "b_from_host"}} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: seed[0], connectionID: seed[1], state: SenderQualityHealthy, acceptedAtMs: 0,
		}).Accepted, true)
	}
	eq(t, observeSenderState(routes, senderState{
		childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: 1,
	}).Accepted, true)
	eq(t, routes.ObserveSenderQualityEvidence(unknownSample(HOST, A, routes.Snapshot().Revision, "a_from_host", 2), nil).Accepted, true)

	for _, acceptedAtMs := range []int64{3, 4, 5} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	operation := must(t, routes.Reconcile(6).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandQualityConvergence)
}

// TS 448: consumes a quality candidate rejected at preparation instead of rebuilding it
func TestConsumesQualityCandidateRejectedAtPreparation(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 0)
	first := must(t, routes.Reconcile(10).Operation)
	eq(t, first.ChildPeerID, A)
	eq(t, first.Reason, DemandQualityConvergence)
	rejected := first.Candidates[first.Cursor].Tuple

	operation := first
	for round := 0; operation != nil && round < 4; round++ {
		if round > 0 && slices.Contains(tuplesFrom(operation, operation.Cursor), rejected) {
			t.Fatalf("round %d rebuilt the rejected tuple %+v", round, rejected)
		}
		eq(t, routes.SkipCurrentCandidate(cursorGuard(operation), int64(11+round), RejectionStale).Accepted, true)
		operation = routes.Reconcile(int64(12 + round)).Operation
	}
	noOperation(t, operation)
}

// TS 480: lets a newly committed availability edge report persistent degradation before healthy
func TestNewlyCommittedEdgeReportsPersistentDegradationBeforeHealthy(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, A, 2, ms(0))
	joined := must(t, routes.Reconcile(0).Operation)
	eq(t, joined.ChildPeerID, A)
	commitCurrent(t, routes, 1, "a_from_host")

	for _, acceptedAtMs := range []int64{2, 3, 4} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}

	operation := must(t, routes.Reconcile(5).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandQualityConvergence)
}

// TS 506: starts a fresh degradation run when an availability sender identity changes
func TestStartsFreshDegradationRunWhenSenderIdentityChanges(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, A, 2, ms(0))
	routes.Reconcile(0)
	commitCurrent(t, routes, 1, "a_from_host")

	observe := func(identity string, at int64) {
		t.Helper()
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", senderIdentity: identity, state: SenderQualityDegraded, acceptedAtMs: at,
		}).Accepted, true)
	}
	observe("old-rtp\x00track", 2)
	observe("old-rtp\x00track", 3)
	observe("new-rtp\x00track", 4)
	noOperation(t, routes.Reconcile(4).Operation)
	observe("new-rtp\x00track", 5)
	observe("new-rtp\x00track", 6)
	operation := must(t, routes.Reconcile(7).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandQualityConvergence)
}

// TS 566: lets a reset sender begin a new degradation run without a healthy sample
func TestResetSenderBeginsNewDegradationRunWithoutHealthySample(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, A, 2, ms(0))
	routes.Reconcile(0)
	commitCurrent(t, routes, 1, "a_from_host")
	eq(t, observeSenderState(routes, senderState{
		childPeerID: A, connectionID: "a_from_host", state: SenderQualityHealthy, acceptedAtMs: 2,
	}).Accepted, true)

	routes.ResetSenderQuality(HOST, "host_session", 3)
	for _, acceptedAtMs := range []int64{4, 5, 6} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	operation := must(t, routes.Reconcile(7).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandQualityConvergence)
}

// TS 599: regenerates a persistently degraded edge on the same parent
func TestRegeneratesPersistentlyDegradedEdgeOnSameParent(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 10)

	operation := must(t, routes.Reconcile(14).Operation)
	eq(t, len(operation.Candidates), 1)
	eq(t, operation.Candidates[0].Tuple, regenerateTuple(HOST))
	expectTransition(t, operation.Candidates[0].EndpointTransition, TransitionNone, HOST)

	begin := beginOperation(t, routes, BeginInput{NowMs: 15, ConnectionID: "a_regenerated", Reservation: direct()})
	edge := edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_from_host")
	eq(t, edge.Usable, true)
	eq(t, edge.PhysicalActive, true)
	guard := guardFor(A, begin.Current.Revision, "a_regenerated")
	ready := routes.CandidateReady(guard, 16, nil, CandidateProof{RelativeQualityApproved: true})
	eq(t, ready.Accepted, true)
	eq(t, ready.Committed, false)

	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		HOST, A, begin.Current.Revision, "a_regenerated", "a-regenerated-rtp\x00track", SenderQualityHealthy, 17), nil).Committed, true)
	edge = edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_regenerated")
	eq(t, observeSenderState(routes, senderState{
		childPeerID: A, connectionID: "a_regenerated", senderIdentity: "a-regenerated-rtp\x00track",
		state: SenderQualityHealthy, acceptedAtMs: 18,
	}).Accepted, true)
	for _, acceptedAtMs := range []int64{19, 20, 21} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_regenerated", senderIdentity: "a-regenerated-rtp\x00track",
			state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	eq(t, must(t, routes.Reconcile(22).Operation).Candidates[0].Tuple, regenerateTuple(HOST))
}

// TS 686: keeps same-edge regeneration damped while allowing a different parent
func TestKeepsSameEdgeRegenerationDampedWhileAllowingDifferentParent(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 10)

	routes.Reconcile(14)
	begin := beginOperation(t, routes, BeginInput{NowMs: 15, ConnectionID: "a_regenerated", Reservation: direct()})
	guard := guardFor(A, begin.Current.Revision, "a_regenerated")
	routes.CandidateReady(guard, 16, nil, CandidateProof{RelativeQualityApproved: true})
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		HOST, A, begin.Current.Revision, "a_regenerated", "a-regenerated-rtp\x00track", SenderQualityHealthy, 17), nil).Committed, true)

	for _, acceptedAtMs := range []int64{18, 19, 20} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_regenerated", senderIdentity: "a-regenerated-rtp\x00track",
			state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	noOperation(t, routes.Reconcile(21).Operation)

	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	eq(t, observeSenderState(routes, senderState{
		childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 22,
	}).Accepted, true)
	withAlternate := must(t, routes.Reconcile(23).Operation)
	eq(t, withAlternate.ChildPeerID, A)
	eq(t, withAlternate.Reason, DemandQualityConvergence)
	eqSlice(t, tuples(withAlternate), []CandidateTuple{peerTuple(B)})
}

// TS 754: uses the existing one-slot overlap for same-parent regeneration
func TestUsesExistingOneSlotOverlapForSameParentRegeneration(t *testing.T) {
	routes := newController(1, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 1, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 10)

	operation := must(t, routes.Reconcile(14).Operation)
	expectTransition(t, operation.Candidates[0].EndpointTransition, TransitionOverlap, HOST)
	begin := beginOperation(t, routes, BeginInput{NowMs: 15, ConnectionID: "a_regenerated", Reservation: directOverlap("a_overlap")})
	guard := guardFor(A, begin.Current.Revision, "a_regenerated")
	routes.CandidateReady(guard, 16, nil, CandidateProof{RelativeQualityApproved: true})
	committed := routes.ObserveSenderQualityEvidence(senderSample(
		HOST, A, begin.Current.Revision, "a_regenerated", "a-regenerated-rtp\x00track", SenderQualityHealthy, 17), nil)
	eq(t, committed.Committed, true)
	containsLabels(t, committed.Released, "a_overlap")
}

// TS 794: keeps one quality experiment through inconclusive current windows
func TestKeepsOneQualityExperimentThroughInconclusiveWindows(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 9})
	observePersistentDegraded(t, routes, A, "a_from_host", 10)

	operation := must(t, routes.Reconcile(14).Operation)
	eqSlice(t, tuples(operation), []CandidateTuple{peerTuple(B), regenerateTuple(HOST)})
	beginCandidate(t, routes, BeginInput{NowMs: 15, ConnectionID: "a_from_b_failed", Reservation: direct()})

	currentRevision := routes.Snapshot().Revision
	eq(t, routes.ObserveSenderQualityEvidence(unknownSample(HOST, A, currentRevision, "a_from_host", 16), nil).Accepted, true)
	pending := must(t, routes.Snapshot().Operation)
	eq(t, pending.Reason, DemandQualityConvergence)
	eq(t, pending.Cursor, 0)

	afterFirst := routes.OperationExpired(must(t, routes.Snapshot().Operation).WakeAtMs)
	eq(t, afterFirst.Accepted, true)
	pending = must(t, routes.Snapshot().Operation)
	eq(t, pending.Reason, DemandQualityConvergence)
	eq(t, pending.Cursor, 1)
	regeneration := beginOperation(t, routes, BeginInput{
		NowMs: 20, ConnectionID: "a_regeneration_failed", Reservation: directOverlap("a_regeneration_overlap"),
	})
	eq(t, routes.CandidateFailed(guardFor(A, regeneration.Current.Revision, "a_regeneration_failed"), 21).Accepted, true)
	noOperation(t, routes.Snapshot().Operation)

	for _, acceptedAtMs := range []int64{22, 23, 24} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	noOperation(t, routes.Reconcile(25).Operation)
	routes.TouchExternalFacts()
	noOperation(t, routes.Reconcile(26).Operation)

	addViewer(routes, C, 2, nil)
	routes.hydrateEdge(C, peerEdge(B, "c_from_b"))
	observeSenderState(routes, senderState{
		childPeerID: C, connectionID: "c_from_b", parentPeerID: B, state: SenderQualityHealthy, acceptedAtMs: 27,
	})
	newOpportunity := must(t, routes.Reconcile(28).Operation)
	eq(t, newOpportunity.ChildPeerID, A)
	eq(t, newOpportunity.Reason, DemandQualityConvergence)
	eqSlice(t, tuples(newOpportunity), []CandidateTuple{peerTuple(C)})
}

// TS 902: does not restart a quality candidate set after its total deadline
func TestDoesNotRestartQualityCandidateSetAfterTotalDeadline(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 9})
	observePersistentDegraded(t, routes, A, "a_from_host", 10)

	operation := must(t, routes.Reconcile(14).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 15, ConnectionID: "a_from_b_pending", Reservation: direct()})
	eq(t, routes.OperationExpired(operation.DeadlineAtMs).Accepted, true)
	noOperation(t, routes.Snapshot().Operation)
	noOperation(t, routes.Reconcile(operation.DeadlineAtMs+1).Operation)
	routes.TouchExternalFacts()
	noOperation(t, routes.Reconcile(operation.DeadlineAtMs+2).Operation)

	observeSenderState(routes, senderState{
		childPeerID: A, connectionID: "a_from_host", state: SenderQualityHealthy, acceptedAtMs: operation.DeadlineAtMs + 3,
	})
	for _, acceptedAtMs := range []int64{operation.DeadlineAtMs + 4, operation.DeadlineAtMs + 5, operation.DeadlineAtMs + 6} {
		observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		})
	}
	eq(t, must(t, routes.Reconcile(operation.DeadlineAtMs+7).Operation).Reason, DemandQualityConvergence)
}

// TS 951: keeps isolated limitation windows diagnostic
func TestKeepsIsolatedLimitationWindowsDiagnostic(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 99})
	observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", state: SenderQualityHealthy, acceptedAtMs: 99})
	observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: 100})
	repeated := senderSample(HOST, A, routes.Snapshot().Revision, "a_from_host", "a_from_host-rtp\x00track", SenderQualityDegraded, 101)
	repeated.SampleTimestampMs = ms(100) // TS: sampleTimestampMs 100 with acceptedAtMs 101
	eq(t, routes.ObserveSenderQualityEvidence(repeated, nil).Accepted, true)
	noOperation(t, routes.Reconcile(101).Operation)
	for _, acceptedAtMs := range []int64{102} {
		observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs})
		noOperation(t, routes.Reconcile(acceptedAtMs).Operation)
	}
	observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: 103})
	operation := must(t, routes.Reconcile(104).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandQualityConvergence)
}

// TS 1010: never treats one degraded Host edge as SFU fanout pressure
func TestNeverTreatsOneDegradedHostEdgeAsSfuFanoutPressure(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true, QualityConvergenceEnabled: true})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateHostPublication("publication_generation_12345678", res("publication_resource"), "")
	routes.hydrateEdge(B, sfuEdge("publication_generation_12345678", "b_from_sfu", "b_subscription"))
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	observePersistentSfuHealthy(t, routes, "publication_generation_12345678", routes.Snapshot().Revision, 103)
	operation := must(t, routes.Reconcile(104).Operation)
	eq(t, operation.Reason, DemandQualityConvergence)
	eq(t, len(operation.Candidates), 1)
	eq(t, operation.Candidates[0].Tuple, regenerateTuple(HOST))
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, HOST)
}

// TS 1052: commits only after the Viewer approves a P2P receive improvement
func TestCommitsOnlyAfterViewerApprovesP2PReceiveImprovement(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	eq(t, observeSenderState(routes, senderState{
		childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 100,
	}).Accepted, true)
	observePersistentDegraded(t, routes, A, "a_from_host", 99)

	operation := must(t, routes.Reconcile(103).Operation)
	eq(t, operation.Reason, DemandQualityConvergence)
	eq(t, operation.Candidates[0].Tuple, peerTuple(B))
	begin := beginCandidate(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b", Reservation: direct()})
	current := must(t, begin.Operation).Current
	guard := guardFor(A, current.Revision, "a_from_b")
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_from_host")
	ready := routes.CandidateReady(guard, 105, nil, CandidateProof{RelativeQualityApproved: true})
	eq(t, ready.Accepted, true)
	eq(t, ready.Committed, false)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, "a_from_b", "a-from-b-rtp\x00a-track", SenderQualityHealthy, 106), nil).Committed, true)
	edge = edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, B)
	eq(t, edge.ConnectionID, "a_from_b")

	migratedRevision := routes.Snapshot().Revision
	observeMigrated := func(state SenderQualityState, at int64) SenderQualityEvidenceResult {
		return routes.ObserveSenderQualityEvidence(senderSample(
			B, A, migratedRevision, "a_from_b", "a-from-b-rtp\x00a-track", state, at), nil)
	}
	for _, acceptedAtMs := range []int64{107, 108, 109} {
		eq(t, observeMigrated(SenderQualityDegraded, acceptedAtMs).Accepted, true)
	}
	noOperation(t, routes.Reconcile(110).Operation)
	routes.TouchExternalFacts()
	noOperation(t, routes.Reconcile(110).Operation)
	eq(t, observeMigrated(SenderQualityHealthy, 111).Accepted, true)
	eq(t, observeMigrated(SenderQualityDegraded, 112).Accepted, true)
	eq(t, observeMigrated(SenderQualityDegraded, 113).Accepted, true)
	eq(t, observeMigrated(SenderQualityDegraded, 114).Accepted, true)
	eq(t, must(t, routes.Reconcile(115).Operation).Reason, DemandQualityConvergence)
}

// TS 1147: uses one healthy serial operation to distribute a newly committed Host root
func TestUsesOneHealthySerialOperationToDistributeNewHostRoot(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, B, 2, nil)
	addViewer(routes, C, 0, nil)
	addViewer(routes, D, 0, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	routes.hydrateEdge(C, peerEdge(B, "c_from_b"))
	routes.hydrateEdge(D, peerEdge(B, "d_from_b"))
	addViewer(routes, A, 2, ms(10))

	joined := must(t, routes.Reconcile(10).Operation)
	eq(t, joined.Candidates[0].Tuple, peerTuple(HOST))
	commitCurrent(t, routes, 11, "a_from_host")

	convergence := must(t, routes.Reconcile(13).Operation)
	eq(t, convergence.Reason, DemandRootConvergence)
	eq(t, len(convergence.Candidates), 1)
	eq(t, convergence.Candidates[0].Tuple, peerTuple(A))
	eq(t, edgeOf(t, routes, convergence.ChildPeerID).ParentPeerID, B)

	prepared := beginOperation(t, routes, BeginInput{NowMs: 14, ConnectionID: "root_convergence_candidate", Reservation: direct()})
	guard := CandidateGuard{
		ChildPeerID:    convergence.ChildPeerID,
		ChildSessionID: convergence.ChildSessionID,
		Revision:       prepared.Current.Revision,
		ConnectionID:   "root_convergence_candidate",
	}
	ready := routes.CandidateReady(guard, 15, nil, CandidateProof{})
	eq(t, ready.Accepted, true)
	eq(t, ready.Committed, false)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		A, convergence.ChildPeerID, prepared.Current.Revision, "root_convergence_candidate", "root-rtp\x00track",
		SenderQualityHealthy, 16), nil).Committed, true)
	eq(t, edgeOf(t, routes, convergence.ChildPeerID).ParentPeerID, A)
	migratedChildPeerID := convergence.ChildPeerID
	for _, acceptedAtMs := range []int64{17, 18, 19} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: migratedChildPeerID, connectionID: "root_convergence_candidate", parentPeerID: A,
			state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	noOperation(t, routes.Reconcile(20).Operation)
	eq(t, observeSenderState(routes, senderState{
		childPeerID: migratedChildPeerID, connectionID: "root_convergence_candidate", parentPeerID: A,
		state: SenderQualityHealthy, acceptedAtMs: 21,
	}).Accepted, true)
	for _, acceptedAtMs := range []int64{22, 23, 24} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: migratedChildPeerID, connectionID: "root_convergence_candidate", parentPeerID: A,
			state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	eq(t, must(t, routes.Reconcile(25).Operation).Reason, DemandQualityConvergence)
}

// TS 1247: advances a silent Peer only when another bounded candidate remains
func TestAdvancesSilentPeerOnlyWhenAnotherBoundedCandidateRemains(t *testing.T) {
	routes := newController(2, Options{OperationTimeoutMs: 20_000})
	addViewer(routes, B, 2, nil)
	addViewer(routes, C, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	routes.hydrateEdge(C, peerEdge(HOST, "c_from_host"))
	addViewer(routes, A, 0, ms(0))

	operation := must(t, routes.Reconcile(0).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 100, ConnectionID: "first_silent_candidate", Reservation: direct()})
	eq(t, must(t, routes.Snapshot().Operation).WakeAtMs, 5_100)
	eq(t, routes.OperationExpired(5_100).Accepted, true)

	beginCandidate(t, routes, BeginInput{NowMs: 5_200, ConnectionID: "second_silent_candidate", Reservation: direct()})
	eq(t, must(t, routes.Snapshot().Operation).WakeAtMs, operation.DeadlineAtMs)
	eq(t, must(t, routes.Snapshot().Operation).DeadlineAtMs, operation.DeadlineAtMs)
	eq(t, routes.OperationExpired(10_200).Accepted, false)
	eq(t, must(t, routes.Snapshot().Operation).Current.ConnectionID, "second_silent_candidate")
}

// TS 1281: retains a peer-only direct attempt through the total operation deadline
func TestRetainsPeerOnlyDirectAttemptThroughTotalDeadline(t *testing.T) {
	routes := newController(1, Options{OperationTimeoutMs: 20_000})
	addViewer(routes, A, 0, ms(0))

	operation := must(t, routes.Reconcile(0).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 100, ConnectionID: "only_direct_candidate", Reservation: direct()})

	eq(t, must(t, routes.Snapshot().Operation).WakeAtMs, operation.DeadlineAtMs)
	eq(t, routes.OperationExpired(5_100).Accepted, false)
	eq(t, must(t, routes.Snapshot().Operation).Current.ConnectionID, "only_direct_candidate")
	expired := routes.OperationExpired(operation.DeadlineAtMs)
	eq(t, expired.Accepted, true)
	eqSlice(t, expired.FailedPeerIDs, []string{A})
}

// TS 1305: keeps only fresh exact-edge quality shadow aggregates
func TestKeepsOnlyFreshExactEdgeQualityShadowAggregates(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 1, ms(0))
	routes.Reconcile(10)
	commitCurrent(t, routes, 20, "a_quality_connection")
	active := routes.Snapshot()

	eq(t, routes.ObserveQualityEvidence(qualityWindow(
		A, active.Revision, "a_quality_connection", peerUpstream(HOST), 1, 1_000, qualityMetrics())), QualityEvidenceObserved)
	eq(t, routes.Snapshot().Revision, active.Revision)
	eq(t, routes.Snapshot().FactVersion, active.FactVersion)
	addViewer(routes, B, 0, ms(1_100))
	routes.Reconcile(1_200)
	commitCurrent(t, routes, 1_210, "b_quality_connection")
	quality := routes.RouteDiagnosticSnapshot(1_300).Children[0].Quality
	if quality == nil {
		t.Fatal("expected quality aggregates")
	}
	eq(t, *quality, protocol.RouteDiagnosticQuality{
		EligibleWindows:    1,
		EligibleDurationMs: 2_000,
		FreezeWindows:      1,
		FreezeCount:        1,
		FreezeDurationMs:   250,
		PauseCount:         0,
		PauseDurationMs:    0,
	})
	if routes.RouteDiagnosticSnapshot(5_999).Children[0].Quality == nil {
		t.Fatal("quality expired before 6_000")
	}
	if routes.RouteDiagnosticSnapshot(6_000).Children[0].Quality != nil {
		t.Fatal("quality did not expire at 6_000")
	}

	nextRevision := routes.Snapshot().Revision
	nullDecoded := qualityMetrics()
	nullDecoded.FramesDecodedDelta = nil
	eq(t, routes.ObserveQualityEvidence(qualityWindow(
		A, nextRevision, "a_quality_connection", peerUpstream(HOST), 2, 7_000, nullDecoded)), QualityEvidenceAccepted)
	if routes.RouteDiagnosticSnapshot(7_100).Children[0].Quality != nil {
		t.Fatal("expected no quality after an accepted-only window")
	}
	eq(t, routes.ObserveQualityEvidence(qualityWindow(
		A, nextRevision, "a_quality_connection", peerUpstream(HOST), 2, 7_100, qualityMetrics())), QualityEvidenceObserved)
	eq(t, routes.ObserveQualityEvidence(qualityWindow(
		A, nextRevision, "a_quality_connection", peerUpstream(HOST), 1, 7_200, qualityMetrics())), QualityEvidenceRejected)

	routes.SetPaused(true, ms(7_300))
	if routes.RouteDiagnosticSnapshot(7_301).Children[0].Quality != nil {
		t.Fatal("expected no quality while paused")
	}
}

// TS 1390: waits for a fresh healthy candidate sender after relative proof
func TestWaitsForFreshHealthyCandidateSenderAfterRelativeProof(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	eq(t, observeSenderState(routes, senderState{
		childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 100,
	}).Accepted, true)
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	routes.Reconcile(103)
	begin := beginCandidate(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b", Reservation: direct()})
	current := must(t, begin.Operation).Current
	eq(t, routes.CandidateTransportConnected(guardFor(A, current.Revision, "a_from_b"), 105).Accepted, true)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, "a_from_b", "a-from-b-rtp\x00track", SenderQualityDegraded, 105), nil).Accepted, true)
	eq(t, routes.CandidateReady(guardFor(A, current.Revision, "a_from_b"), 106, nil, CandidateProof{RelativeQualityApproved: true}).Committed, false)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, "a_from_b", "a-from-b-rtp\x00track", SenderQualityHealthy, 107), nil).Committed, true)
}

// TS 1464: does not commit a relative proof after its freshness window expires
func TestDoesNotCommitRelativeProofAfterFreshnessWindowExpires(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	routes.Reconcile(103)
	begin := beginCandidate(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b_stale_proof", Reservation: direct()})
	current := must(t, begin.Operation).Current
	guard := guardFor(A, current.Revision, "a_from_b_stale_proof")
	eq(t, routes.CandidateTransportConnected(guard, 105).Accepted, true)

	eq(t, routes.CandidateReady(guard, 105, nil, CandidateProof{RelativeQualityApproved: true}).Committed, false)
	eq(t, routes.CandidateReady(guard, 109, nil, CandidateProof{RelativeQualityApproved: true}).Committed, false)
	for _, acceptedAtMs := range []int64{5_105, 5_106, 5_107} {
		eq(t, observeSenderState(routes, senderState{
			childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs,
		}).Accepted, true)
	}
	settled := routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, guard.ConnectionID, "stale-proof-rtp\x00track", SenderQualityHealthy, 5_111), nil)
	eq(t, settled.Accepted, true)
	eq(t, settled.Committed, false)
	operation := must(t, routes.Snapshot().Operation)
	eq(t, operation.Reason, DemandQualityConvergence)
	eq(t, operation.Cursor, 1)
	edge := edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_from_host")
}

// TS 1528: abandons a pending quality move when the incumbent recovers
func TestAbandonsPendingQualityMoveWhenIncumbentRecovers(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	routes.Reconcile(103)
	begin := beginCandidate(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b_incumbent_recovery", Reservation: direct()})
	current := must(t, begin.Operation).Current
	guard := guardFor(A, current.Revision, "a_from_b_incumbent_recovery")
	eq(t, routes.CandidateReady(guard, 105, nil, CandidateProof{RelativeQualityApproved: true}).Committed, false)

	eq(t, observeSenderState(routes, senderState{
		childPeerID: A, connectionID: "a_from_host", state: SenderQualityHealthy, acceptedAtMs: 106,
	}).Accepted, true)
	noOperation(t, routes.Snapshot().Operation)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, guard.ConnectionID, "late-healthy-rtp\x00track", SenderQualityHealthy, 107), nil).Accepted, false)
	edge := edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_from_host")
}

func TestQualityReadinessKeepsInconclusiveIncumbentWithinTheSameOperation(t *testing.T) {
	for _, unknownFirst := range []bool{true, false} {
		t.Run(fmt.Sprintf("unknown before readiness=%v", unknownFirst), func(t *testing.T) {
			routes := newController(2, Options{QualityConvergenceEnabled: true})
			addViewer(routes, A, 2, nil)
			addViewer(routes, B, 2, nil)
			routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
			routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
			observePersistentDegraded(t, routes, A, "a_from_host", 99)
			operation := must(t, routes.Reconcile(103).Operation)
			begin := beginOperation(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b", Reservation: direct()})
			guard := guardFor(A, begin.Current.Revision, "a_from_b")
			unknown := func(at int64) {
				eq(t, routes.ObserveSenderQualityEvidence(unknownSample(HOST, A, routes.Snapshot().Revision, "a_from_host", at), nil).Accepted, true)
			}
			ready := func(at int64) {
				settled := routes.CandidateReady(guard, at, nil, CandidateProof{RelativeQualityApproved: true})
				eq(t, settled.Accepted, true)
				eq(t, settled.Committed, false)
			}
			if unknownFirst {
				unknown(105)
				ready(106)
			} else {
				ready(105)
				unknown(106)
			}
			pending := must(t, routes.Snapshot().Operation)
			eq(t, pending.Current.Revision, guard.Revision)
			eq(t, pending.DeadlineAtMs, operation.DeadlineAtMs)
			eq(t, edgeOf(t, routes, A).ConnectionID, "a_from_host")
			settled := routes.ObserveSenderQualityEvidence(senderSample(
				B, A, guard.Revision, guard.ConnectionID, "candidate-rtp\x00track", SenderQualityHealthy, 107), nil)
			eq(t, settled.Committed, true)
			noOperation(t, routes.Snapshot().Operation)
			eq(t, edgeOf(t, routes, A).ConnectionID, "a_from_b")
		})
	}
}

func TestInconclusiveQualityOperationStillRequiresCurrentAuthority(t *testing.T) {
	for _, cause := range []string{"healthy", "reset", "sender-replaced", "pause", "deadline"} {
		t.Run(cause, func(t *testing.T) {
			routes := newController(2, Options{QualityConvergenceEnabled: true})
			addViewer(routes, A, 2, nil)
			addViewer(routes, B, 2, nil)
			routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
			routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
			observePersistentDegraded(t, routes, A, "a_from_host", 99)
			operation := must(t, routes.Reconcile(103).Operation)
			begin := beginOperation(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b", Reservation: direct()})
			guard := guardFor(A, begin.Current.Revision, "a_from_b")
			eq(t, routes.ObserveSenderQualityEvidence(unknownSample(HOST, A, routes.Snapshot().Revision, "a_from_host", 105), nil).Accepted, true)
			eq(t, routes.CandidateReady(guard, 106, nil, CandidateProof{RelativeQualityApproved: true}).Accepted, true)
			at := int64(107)
			switch cause {
			case "healthy":
				observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", state: SenderQualityHealthy, acceptedAtMs: at})
			case "reset":
				routes.ResetSenderQuality(HOST, "host_session", at)
			case "sender-replaced":
				observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", senderIdentity: "replacement-rtp\x00track", state: SenderQualityDegraded, acceptedAtMs: at})
			case "pause":
				routes.SetPaused(true, &at)
			case "deadline":
				at = operation.DeadlineAtMs
			}
			eq(t, routes.CandidateReady(guard, at, nil, CandidateProof{RelativeQualityApproved: true}).Accepted, false)
			noOperation(t, routes.Snapshot().Operation)
			eq(t, edgeOf(t, routes, A).ConnectionID, "a_from_host")
		})
	}
}

// TS 1582: persists relative proof while candidate sender evidence is unknown
func TestPersistsRelativeProofWhileCandidateSenderEvidenceIsUnknown(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 99})
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	routes.Reconcile(103)
	begin := beginCandidate(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b", Reservation: direct()})
	current := must(t, begin.Operation).Current
	eq(t, routes.ObserveSenderQualityEvidence(unknownSample(B, A, current.Revision, "a_from_b", 105), nil).Accepted, true)
	eq(t, routes.CandidateReady(guardFor(A, current.Revision, "a_from_b"), 106, nil, CandidateProof{RelativeQualityApproved: true}).Committed, false)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, "a_from_b", "a-from-b-rtp\x00track", SenderQualityHealthy, 107), nil).Committed, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, B)
	eq(t, edge.ConnectionID, "a_from_b")
}

// TS 1648: advances past a persistently limited quality candidate
func TestAdvancesPastPersistentlyLimitedQualityCandidate(t *testing.T) {
	routes := newController(3, Options{QualityConvergenceEnabled: true})
	for _, seed := range [][2]string{{A, "a_from_host"}, {B, "b_from_host"}, {C, "c_from_host"}} {
		addViewer(routes, seed[0], 3, nil)
		routes.hydrateEdge(seed[0], peerEdge(HOST, seed[1]))
	}
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 99})
	observeSenderState(routes, senderState{childPeerID: C, connectionID: "c_from_host", state: SenderQualityHealthy, acceptedAtMs: 99})
	observePersistentDegraded(t, routes, A, "a_from_host", 99)

	operation := must(t, routes.Reconcile(103).Operation)
	deadlineAtMs := operation.DeadlineAtMs
	eq(t, operation.Candidates[0].Tuple.Kind, UpstreamPeer)
	firstParent := operation.Candidates[0].Tuple.ParentPeerID
	first := beginOperation(t, routes, BeginInput{NowMs: 104, ConnectionID: "first_quality_candidate", Reservation: direct()})
	var lastResult SenderQualityEvidenceResult
	for _, acceptedAtMs := range []int64{105, 106, 107} {
		lastResult = routes.ObserveSenderQualityEvidence(senderSample(
			firstParent, A, first.Current.Revision, "first_quality_candidate", "first-quality-rtp\x00track",
			SenderQualityDegraded, acceptedAtMs), nil)
		eq(t, lastResult.Accepted, true)
	}
	eq(t, lastResult.Committed, false)
	edge := edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_from_host")
	pending := must(t, routes.Snapshot().Operation)
	eq(t, pending.Cursor, 1)
	eq(t, pending.DeadlineAtMs, deadlineAtMs)
	if pending.Current != nil {
		t.Fatal("expected no live attempt")
	}

	secondParent := pending.Candidates[1].Tuple.ParentPeerID
	if secondParent == firstParent {
		t.Fatalf("second parent %q repeats the first", secondParent)
	}
	second := beginOperation(t, routes, BeginInput{NowMs: 108, ConnectionID: "second_quality_candidate", Reservation: direct()})
	eq(t, routes.ObserveSenderQualityEvidence(unknownSample(
		secondParent, A, second.Current.Revision, "second_quality_candidate", 109), nil).Accepted, true)
	eq(t, routes.CandidateReady(guardFor(A, second.Current.Revision, "second_quality_candidate"), 110, nil,
		CandidateProof{RelativeQualityApproved: true}).Committed, false)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		secondParent, A, second.Current.Revision, "second_quality_candidate", "second-quality-rtp\x00track",
		SenderQualityHealthy, 111), nil).Committed, true)
	edge = edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, secondParent)
	eq(t, edge.ConnectionID, "second_quality_candidate")
}

// TS 1768: preempts quality work for a Viewer without media
func TestPreemptsQualityWorkForViewerWithoutMedia(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 100})
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	eq(t, must(t, routes.Reconcile(103).Operation).Reason, DemandQualityConvergence)
	addViewer(routes, C, 0, ms(104))
	join := must(t, routes.Reconcile(105).Operation)
	eq(t, join.ChildPeerID, C)
	eq(t, join.Reason, DemandJoin)
	commitCurrent(t, routes, 106, "c_join_connection")
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 107})
	observePersistentDegraded(t, routes, A, "a_from_host", 107)
	eq(t, must(t, routes.Reconcile(111).Operation).Reason, DemandQualityConvergence)
	edge := edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, HOST)
	eq(t, edge.ConnectionID, "a_from_host")
}

// TS 1806: keeps a quality operation reason immutable until availability preempts it
func TestKeepsQualityOperationReasonImmutableUntilAvailabilityPreempts(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 99})
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	eq(t, must(t, routes.Reconcile(103).Operation).Reason, DemandQualityConvergence)

	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", routes.Snapshot().Revision, "a_from_host"), ms(104)), true)
	eq(t, must(t, routes.Snapshot().Operation).Reason, DemandQualityConvergence)
	preempted := must(t, routes.Reconcile(105).Operation)
	eq(t, preempted.ChildPeerID, A)
	eq(t, preempted.Reason, DemandEdgeUnavailable)
}

// TS 1842: keeps a direct convergence reason immutable until availability preempts it
func TestKeepsDirectConvergenceReasonImmutableUntilAvailabilityPreempts(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	routes.hydrateHostPublication("publication", res("publication_resource"), "")
	addViewer(routes, A, 0, ms(0))
	acquisition := must(t, routes.Reconcile(0).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_head_start", Reservation: direct()})
	routes.OperationExpired(acquisition.WakeAtMs)
	sfu := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 1, ConnectionID: "a_sfu", Reservation: sfuReuse("a_subscription")})
	eq(t, routes.CandidateReady(guardFor(A, sfu.Current.Revision, "a_sfu"), acquisition.WakeAtMs+2, nil, CandidateProof{}).Committed, true)
	eq(t, must(t, routes.Reconcile(acquisition.WakeAtMs+3).Operation).Reason, DemandDirectConvergence)

	eq(t, routes.InvalidateEdge(edgeGuard(A, "", routes.Snapshot().Revision, "a_sfu"), ms(acquisition.WakeAtMs+4)), true)
	eq(t, must(t, routes.Snapshot().Operation).Reason, DemandDirectConvergence)
	preempted := must(t, routes.Reconcile(acquisition.WakeAtMs+5).Operation)
	eq(t, preempted.ChildPeerID, A)
	eq(t, preempted.Reason, DemandEdgeUnavailable)
}

// TS 1893: keeps a root convergence reason immutable until availability preempts it
func TestKeepsRootConvergenceReasonImmutableUntilAvailabilityPreempts(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, B, 2, nil)
	addViewer(routes, C, 0, nil)
	addViewer(routes, D, 0, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	routes.hydrateEdge(C, peerEdge(B, "c_from_b"))
	routes.hydrateEdge(D, peerEdge(B, "d_from_b"))
	addViewer(routes, A, 2, ms(10))
	routes.Reconcile(10)
	commitCurrent(t, routes, 11, "a_from_host")
	convergence := must(t, routes.Reconcile(13).Operation)
	eq(t, convergence.Reason, DemandRootConvergence)
	oldEdge := edgeOf(t, routes, convergence.ChildPeerID)
	eq(t, oldEdge.Kind, UpstreamPeer)

	eq(t, routes.InvalidateEdge(EdgeGuard{
		ChildPeerID:     convergence.ChildPeerID,
		ChildSessionID:  convergence.ChildSessionID,
		ParentSessionID: oldEdge.ParentSessionID,
		RouteRevision:   routes.Snapshot().Revision,
		ConnectionID:    oldEdge.ConnectionID,
	}, ms(14)), true)
	eq(t, must(t, routes.Snapshot().Operation).Reason, DemandRootConvergence)
	preempted := must(t, routes.Reconcile(15).Operation)
	eq(t, preempted.ChildPeerID, convergence.ChildPeerID)
	eq(t, preempted.Reason, DemandEdgeUnavailable)
}

// TS 1931: keeps a usable parent eligible before native state becomes clear
func TestKeepsUsableParentEligibleBeforeNativeStateBecomesClear(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	operation := must(t, routes.Reconcile(103).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandQualityConvergence)
	eqSlice(t, tuples(operation), []CandidateTuple{peerTuple(B), regenerateTuple(HOST)})
}

// TS 1957: waits for a fresh candidate sender after prior evidence expires
func TestWaitsForFreshCandidateSenderAfterPriorEvidenceExpires(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeCurrent := func(childPeerID, connectionID string, state SenderQualityState, acceptedAtMs int64) SenderQualityEvidenceResult {
		return routes.ObserveSenderQualityEvidence(senderSample(
			HOST, childPeerID, routes.Snapshot().Revision, connectionID, connectionID+"-rtp\x00track", state, acceptedAtMs), nil)
	}
	eq(t, observeCurrent(B, "b_from_host", SenderQualityHealthy, 100).Accepted, true)
	eq(t, observeCurrent(A, "a_from_host", SenderQualityHealthy, 99).Accepted, true)
	eq(t, observeCurrent(A, "a_from_host", SenderQualityDegraded, 100).Accepted, true)
	eq(t, observeCurrent(A, "a_from_host", SenderQualityDegraded, 101).Accepted, true)
	eq(t, observeCurrent(A, "a_from_host", SenderQualityDegraded, 102).Accepted, true)
	routes.Reconcile(103)
	begin := beginCandidate(t, routes, BeginInput{NowMs: 104, ConnectionID: "a_from_b_stale", Reservation: direct()})
	current := must(t, begin.Operation).Current
	eq(t, routes.CandidateTransportConnected(guardFor(A, current.Revision, "a_from_b_stale"), 105).Accepted, true)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, "a_from_b_stale", "a-stale-rtp\x00a-track", SenderQualityHealthy, 104), nil).Accepted, true)
	for _, acceptedAtMs := range []int64{5_101, 5_102, 5_103} {
		eq(t, observeCurrent(A, "a_from_host", SenderQualityDegraded, acceptedAtMs).Accepted, true)
	}
	eq(t, routes.CandidateReady(guardFor(A, current.Revision, "a_from_b_stale"), 5_104, nil,
		CandidateProof{RelativeQualityApproved: true}).Committed, false)
	eq(t, routes.ObserveSenderQualityEvidence(senderSample(
		B, A, current.Revision, "a_from_b_stale", "a-stale-rtp\x00a-track", SenderQualityHealthy, 5_105), nil).Committed, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.ParentPeerID, B)
	eq(t, edge.ConnectionID, "a_from_b_stale")
}

// TS 2063: keeps a usable candidate when its native source category expires
func TestKeepsUsableCandidateWhenNativeSourceCategoryExpires(t *testing.T) {
	routes := newController(2, Options{QualityConvergenceEnabled: true})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityHealthy, acceptedAtMs: 0})
	observePersistentDegraded(t, routes, A, "a_from_host", 0)
	eq(t, must(t, routes.Reconcile(4).Operation).Reason, DemandQualityConvergence)
	begin := beginCandidate(t, routes, BeginInput{NowMs: 5_001, ConnectionID: "expired_candidate", Reservation: direct()})
	eq(t, begin.Accepted, true)
	eq(t, must(t, begin.Operation).Current.ConnectionID, "expired_candidate")
}

// TS 2086: keeps bounded SFU after direct candidates when every Host root is degraded
func TestKeepsBoundedSfuAfterDirectCandidatesWhenEveryHostRootIsDegraded(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true, QualityConvergenceEnabled: true})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	firstRegeneration := must(t, routes.Reconcile(103).Operation)
	eq(t, firstRegeneration.Candidates[0].Tuple, regenerateTuple(HOST))
	firstAttempt := beginOperation(t, routes, BeginInput{
		NowMs: 104, ConnectionID: "a_regeneration_failed", Reservation: directOverlap("a_regeneration_overlap"),
	})
	eq(t, routes.CandidateFailed(guardFor(A, firstAttempt.Current.Revision, "a_regeneration_failed"), 105).Accepted, true)
	eq(t, routes.SetEffectiveCapacity(B, sessionOf(B), 1, nil), true)
	observePersistentDegraded(t, routes, B, "b_from_host", 104)
	operation := must(t, routes.Reconcile(108).Operation)
	childPeerID := operation.ChildPeerID
	eq(t, childPeerID, A)
	eqSlice(t, tuples(operation), []CandidateTuple{peerTuple(B), sfuTuple(PublicationCreate)})
	directAttempt := beginOperation(t, routes, BeginInput{NowMs: 109, ConnectionID: "quality_peer_candidate", Reservation: direct()})
	failed := routes.CandidateFailed(guardFor(childPeerID, directAttempt.Current.Revision, "quality_peer_candidate"), 110)
	eq(t, failed.Accepted, true)
	eqSlice(t, failed.FailedPeerIDs, []string{})

	begin := beginCandidate(t, routes, BeginInput{
		NowMs:                   111,
		ConnectionID:            "quality_sfu_candidate",
		Reservation:             sfuCreateOverlap("sfu-edge", "sfu-publication", "host-overlap"),
		PublicationGeneration:   "publication_generation_12345678",
		PublicationConnectionID: "publication_connection_12345678",
		HostSessionID:           "host_session",
	})
	current := must(t, begin.Operation).Current
	guard := guardFor(childPeerID, current.Revision, "quality_sfu_candidate")
	ready := routes.CandidateReady(guard, 112, nil, CandidateProof{})
	eq(t, ready.Accepted, true)
	eq(t, ready.Committed, false)
	for _, acceptedAtMs := range []int64{113, 114} {
		observed := routes.ObserveSfuPublisherQualityEvidence(publisherSample(
			"publication_generation_12345678", current.Revision, SenderQualityHealthy, acceptedAtMs), nil)
		eq(t, observed.Accepted, true)
		eq(t, observed.Committed, false)
		ready = routes.CandidateReady(guard, acceptedAtMs, nil, CandidateProof{})
		eq(t, ready.Accepted, true)
		eq(t, ready.Committed, false)
	}
	eq(t, routes.ObserveSfuPublisherQualityEvidence(publisherSample(
		"publication_generation_12345678", current.Revision, SenderQualityHealthy, 115), nil).Committed, true)
	edge := edgeOf(t, routes, childPeerID)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.ConnectionID, "quality_sfu_candidate")
}

// TS 2201: rejects an SFU quality candidate on degraded publisher evidence
func TestRejectsSfuQualityCandidateOnDegradedPublisherEvidence(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true, QualityConvergenceEnabled: true})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	firstRegeneration := must(t, routes.Reconcile(103).Operation)
	eq(t, firstRegeneration.Candidates[0].Tuple, regenerateTuple(HOST))
	firstAttempt := beginOperation(t, routes, BeginInput{
		NowMs: 104, ConnectionID: "a_regeneration_failed", Reservation: directOverlap("a_regeneration_overlap"),
	})
	eq(t, routes.CandidateFailed(guardFor(A, firstAttempt.Current.Revision, "a_regeneration_failed"), 105).Accepted, true)
	eq(t, routes.SetEffectiveCapacity(B, sessionOf(B), 1, nil), true)
	observePersistentDegraded(t, routes, B, "b_from_host", 104)
	operation := must(t, routes.Reconcile(108).Operation)
	childPeerID := operation.ChildPeerID
	eq(t, childPeerID, A)
	peerAttempt := beginOperation(t, routes, BeginInput{NowMs: 109, ConnectionID: "quality_peer_candidate", Reservation: direct()})
	eq(t, routes.CandidateFailed(guardFor(childPeerID, peerAttempt.Current.Revision, "quality_peer_candidate"), 110).Accepted, true)
	sfuAttempt := beginOperation(t, routes, BeginInput{
		NowMs:                   111,
		ConnectionID:            "quality_sfu_degraded",
		Reservation:             sfuCreateOverlap("sfu-edge", "sfu-publication", "host-overlap"),
		PublicationGeneration:   "publication_generation_degraded",
		PublicationConnectionID: "publication_connection_degraded",
		HostSessionID:           "host_session",
	})
	current := sfuAttempt.Current
	ready := routes.CandidateReady(guardFor(childPeerID, current.Revision, current.ConnectionID), 112, nil, CandidateProof{})
	eq(t, ready.Accepted, true)
	eq(t, ready.Committed, false)
	degraded := routes.ObserveSfuPublisherQualityEvidence(publisherSample(
		"publication_generation_degraded", current.Revision, SenderQualityDegraded, 113), nil)
	eq(t, degraded.Accepted, true)
	eq(t, degraded.Committed, false)
	noOperation(t, routes.Reconcile(114).Operation)
	edge := edgeOf(t, routes, childPeerID)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, HOST)
}

func TestAbandonsHostReliefWhenAnotherHostRootIsNoLongerDegraded(t *testing.T) {
	for _, state := range []SenderQualityState{SenderQualityHealthy, SenderQualityUnknown} {
		t.Run(string(state), func(t *testing.T) {
			routes := newController(2, Options{SfuEnabled: true, QualityConvergenceEnabled: true})
			addViewer(routes, A, 1, nil)
			addViewer(routes, B, 0, nil)
			routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
			routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
			observePersistentDegraded(t, routes, A, "a_from_host", 99)
			observePersistentDegraded(t, routes, B, "b_from_host", 99)
			routes.Reconcile(103)
			begin := beginCandidate(t, routes, BeginInput{
				NowMs:                   104,
				ConnectionID:            "a_from_sfu",
				Reservation:             sfuCreateOverlap("sfu-edge", "sfu-publication", "host-overlap"),
				PublicationGeneration:   "publication_generation_12345678",
				PublicationConnectionID: "publication_connection_12345678",
				HostSessionID:           "host_session",
			})
			current := must(t, begin.Operation).Current
			eq(t, observeSenderState(routes, senderState{
				childPeerID: B, connectionID: "b_from_host", state: state, acceptedAtMs: 105,
			}).Accepted, true)
			eq(t, routes.CandidateReady(guardFor(A, current.Revision, "a_from_sfu"), 106, nil, CandidateProof{}).Committed, false)
			noOperation(t, routes.Snapshot().Operation)
			edge := edgeOf(t, routes, A)
			eq(t, edge.Kind, UpstreamPeer)
			eq(t, edge.ParentPeerID, HOST)
			eq(t, edge.ConnectionID, "a_from_host")
		})
	}
}

// TS 2350: uses SFU reuse only for multi-edge Host relief with healthy ingress
func TestUsesSfuReuseOnlyForMultiEdgeHostReliefWithHealthyIngress(t *testing.T) {
	routes := newController(3, Options{SfuEnabled: true, QualityConvergenceEnabled: true})
	addViewer(routes, A, 0, nil)
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	routes.hydrateHostPublication("publication_generation_12345678", res("publication_resource"), "")
	revision := routes.Snapshot().Revision
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	observePersistentDegraded(t, routes, B, "b_from_host", 99)
	noOperation(t, routes.Reconcile(103).Operation)
	observePersistentSfuHealthy(t, routes, "publication_generation_12345678", revision, 104)
	eq(t, must(t, routes.Reconcile(105).Operation).Candidates[0].Tuple, sfuTuple(PublicationReuse))
	begin := beginCandidate(t, routes, BeginInput{NowMs: 106, ConnectionID: "a_from_sfu_reuse", Reservation: sfuReuse("a_subscription")})
	current := must(t, begin.Operation).Current
	unknown := publisherSample("publication_generation_12345678", revision, SenderQualityUnknown, 107)
	unknown.SampleTimestampMs = nil
	eq(t, routes.ObserveSfuPublisherQualityEvidence(unknown, nil).Accepted, true)
	eq(t, routes.CandidateReady(guardFor(A, current.Revision, "a_from_sfu_reuse"), 108, nil, CandidateProof{}).Committed, false)
	for _, acceptedAtMs := range []int64{2_100, 4_100, 6_100} {
		observeSenderState(routes, senderState{childPeerID: A, connectionID: "a_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs})
		observeSenderState(routes, senderState{childPeerID: B, connectionID: "b_from_host", state: SenderQualityDegraded, acceptedAtMs: acceptedAtMs})
		settled := routes.ObserveSfuPublisherQualityEvidence(publisherSample(
			"publication_generation_12345678", revision, SenderQualityHealthy, acceptedAtMs), nil)
		eq(t, settled.Committed, acceptedAtMs == 6_100)
	}
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.ConnectionID, "a_from_sfu_reuse")
}

// TS 2435: accepts an SFU-fed Peer parent from decoded progress without freeze stats
func TestAcceptsSfuFedPeerParentFromDecodedProgressWithoutFreezeStats(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true, QualityConvergenceEnabled: true})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateHostPublication("publication_generation_12345678", res("publication_resource"), "")
	routes.hydrateEdge(B, sfuEdge("publication_generation_12345678", "b_from_sfu", "b_subscription"))
	revision := routes.Snapshot().Revision
	observePersistentSfuHealthy(t, routes, "publication_generation_12345678", revision, 100)
	metrics := qualityMetrics()
	metrics.FreezeCountDelta = nil
	metrics.FreezeDurationMsDelta = nil
	metrics.PauseCountDelta = nil
	metrics.PauseDurationMsDelta = nil
	eq(t, routes.ObserveQualityEvidence(qualityWindow(B, revision, "b_from_sfu", sfuUpstream(), 1, 100, metrics)), QualityEvidenceAccepted)
	observePersistentDegraded(t, routes, A, "a_from_host", 99)
	eq(t, must(t, routes.Reconcile(103).Operation).Candidates[0].Tuple, peerTuple(B))
}

// TS 2489: clears quality shadow for a relay subtree when its source changes
func TestClearsQualityShadowForRelaySubtreeWhenSourceChanges(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	addViewer(routes, C, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(A, "b_from_a"))
	routes.hydrateEdge(C, peerEdge(B, "c_from_b"))
	revision := routes.Snapshot().Revision

	relays := [][3]string{{B, A, "b_from_a"}, {C, B, "c_from_b"}}
	for _, relay := range relays {
		eq(t, routes.ObserveQualityEvidence(qualityWindow(
			relay[0], revision, relay[2], peerUpstream(relay[1]), 1, 1_000, qualityMetrics())), QualityEvidenceObserved)
	}
	withQuality := func(nowMs int64) int {
		count := 0
		for _, child := range routes.RouteDiagnosticSnapshot(nowMs).Children {
			if child.Quality != nil {
				count++
			}
		}
		return count
	}
	eq(t, withQuality(1_001), 2)

	eq(t, routes.AdoptDirectConnection(AdoptDirectConnectionInput{
		EdgeGuard:       edgeGuard(A, "host_session", revision, "a_from_host"),
		NewConnectionID: "a_from_host_recovered",
	}, 1_002).Accepted, true)
	eq(t, withQuality(1_002), 0)

	for _, step := range []struct {
		at       int64
		expected RouteQualityEvidenceResult
	}{{1_100, QualityEvidenceAccepted}, {3_100, QualityEvidenceObserved}} {
		for _, relay := range relays {
			eq(t, routes.ObserveQualityEvidence(qualityWindow(
				relay[0], revision, relay[2], peerUpstream(relay[1]), 1, step.at, qualityMetrics())), step.expected)
		}
	}
	eq(t, withQuality(3_101), 2)
}

func TestActiveRebuildRetiresBootstrapReservationWithoutLosingWaitingDemand(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	for _, child := range []string{B, C} {
		addViewer(routes, child, 0, nil)
		routes.hydrateEdge(child, peerEdge(HOST, child+"_from_host"))
	}
	addViewer(routes, A, 0, ms(0))
	bootstrap := must(t, routes.Reconcile(0).Operation)
	eq(t, bootstrap.Reason, DemandSfuBootstrap)
	retained := edgeOf(t, routes, bootstrap.ChildPeerID)
	attempt := beginOperation(t, routes, BeginInput{
		NowMs: 1, ConnectionID: "carrier_sfu", PublicationGeneration: "publication",
		PublicationConnectionID: "publication_connection",
		Reservation:             sfuCreateOverlap("subscription", "publication", "overlap"),
	})
	replaced := routes.AdoptDirectConnection(AdoptDirectConnectionInput{
		EdgeGuard:       edgeGuard(bootstrap.ChildPeerID, "host_session", routes.Revision(), retained.ConnectionID),
		NewConnectionID: "recovered_connection",
	}, 2)
	eq(t, replaced.Accepted, true)
	eqLabels(t, replaced.Released, "subscription", "publication", "overlap")
	noOperation(t, routes.Snapshot().Operation)
	eq(t, edgeOf(t, routes, bootstrap.ChildPeerID).ConnectionID, "recovered_connection")
	eq(t, routes.CandidateReady(guardFor(bootstrap.ChildPeerID, attempt.Current.Revision, "carrier_sfu"), 3, nil, CandidateProof{}).Accepted, false)
	next := must(t, routes.Reconcile(4).Operation)
	eq(t, next.Reason, DemandSfuBootstrap)
	eq(t, next.DemandPeerID, A)
}

// TS 2569: retains one latest timing sample for a 20-Viewer burst
func TestRetainsOneLatestTimingSampleForTwentyViewerBurst(t *testing.T) {
	routes := newController(2, Options{})
	viewerPeerIDs := make([]string, 20)
	for index := range viewerPeerIDs {
		viewerPeerIDs[index] = fmt.Sprintf("burst_%02d_12345678", index)
	}
	for _, peerID := range viewerPeerIDs {
		addViewer(routes, peerID, 2, ms(0))
	}

	for index, peerID := range viewerPeerIDs {
		operationStartedAtMs := int64(index+1) * 10
		eq(t, must(t, routes.Reconcile(operationStartedAtMs).Operation).ChildPeerID, peerID)
		commitCurrent(t, routes, operationStartedAtMs+1, peerID+"_connection")
	}

	snapshot := routes.RouteDiagnosticSnapshot(250)
	eq(t, len(snapshot.Children), 20)
	for index, child := range snapshot.Children {
		eqDeep(t, child.QueueWaitMs, optInt(int64(index+1)*10))
		eq(t, child.FinalRoute, string(FinalRouteDirect))
	}
}

// TS 2598: projects latest route timing through snapshot-local ordinals
func TestProjectsLatestRouteTimingThroughSnapshotLocalOrdinals(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 1, ms(100))

	eqDeep(t, routes.RouteDiagnosticSnapshot(150), protocol.RouteDiagnosticSnapshot{
		Children: []protocol.RouteDiagnosticChild{{
			Ordinal:             1,
			Parent:              protocol.RouteDiagnosticParent{Kind: "none"},
			EffectiveCapacity:   1,
			ChildCount:          0,
			DemandAgeMs:         optInt(50),
			QueueWaitMs:         nil,
			CandidateStartMs:    nil,
			FirstDecodedFrameMs: nil,
			FinalMs:             nil,
			FinalRoute:          "waiting",
			RejectionBucket:     "none",
			Quality:             nil,
		}},
		Operation: nil,
	})

	operation := must(t, routes.Reconcile(160).Operation)
	eqDeep(t, routes.RouteDiagnosticSnapshot(165).Operation, &protocol.RouteDiagnosticOperation{
		ChildOrdinal:   1,
		Reason:         "join",
		Stage:          "admission",
		Cursor:         0,
		CandidateCount: protocol.Int(len(operation.Candidates)),
	})
	prepared := beginOperation(t, routes, BeginInput{NowMs: 170, ConnectionID: "a_connection", Reservation: direct()})
	pending := routes.RouteDiagnosticSnapshot(180)
	eq(t, len(pending.Children), 1)
	eqDeep(t, pending.Children[0].QueueWaitMs, optInt(60))
	eqDeep(t, pending.Children[0].CandidateStartMs, optInt(70))
	eqDeep(t, pending.Children[0].FirstDecodedFrameMs, (*protocol.Int)(nil))
	eqDeep(t, pending.Children[0].FinalMs, (*protocol.Int)(nil))
	if pending.Operation == nil {
		t.Fatal("expected a diagnostic operation")
	}
	eq(t, pending.Operation.ChildOrdinal, 1)
	eq(t, pending.Operation.Stage, "first-frame")
	eq(t, routes.CandidateReady(guardFor(A, prepared.Current.Revision, "a_connection"), 230, nil, CandidateProof{}).Accepted, true)
	settled := routes.RouteDiagnosticSnapshot(240)
	eq(t, len(settled.Children), 1)
	eq(t, settled.Children[0].Parent, protocol.RouteDiagnosticParent{Kind: "host"})
	eqDeep(t, settled.Children[0].QueueWaitMs, optInt(60))
	eqDeep(t, settled.Children[0].CandidateStartMs, optInt(70))
	eqDeep(t, settled.Children[0].FirstDecodedFrameMs, optInt(130))
	eqDeep(t, settled.Children[0].FinalMs, optInt(130))
	eq(t, settled.Children[0].FinalRoute, "direct")
	eq(t, settled.Children[0].RejectionBucket, "none")
	if settled.Operation != nil {
		t.Fatal("expected no diagnostic operation")
	}
	serialized, err := json.Marshal(settled)
	if err != nil {
		t.Fatal(err)
	}
	if strings.Contains(string(serialized), A) {
		t.Fatalf("diagnostic snapshot leaks the peer ID: %s", serialized)
	}
	if strings.Contains(string(serialized), "a_connection") {
		t.Fatalf("diagnostic snapshot leaks the connection ID: %s", serialized)
	}

	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", routes.Snapshot().Revision, "a_connection"), ms(300)), true)
	waiting := routes.RouteDiagnosticSnapshot(320).Children[0]
	eqDeep(t, waiting.DemandAgeMs, optInt(20))
	eqDeep(t, waiting.QueueWaitMs, (*protocol.Int)(nil))
	eqDeep(t, waiting.CandidateStartMs, (*protocol.Int)(nil))
	eqDeep(t, waiting.FirstDecodedFrameMs, (*protocol.Int)(nil))
	eqDeep(t, waiting.FinalMs, (*protocol.Int)(nil))
	eq(t, waiting.FinalRoute, "waiting")
	eq(t, waiting.RejectionBucket, "none")

	eq(t, routes.ConfirmDeparture(A, ms(330)), true)
	departed := routes.RouteDiagnosticSnapshot(330)
	eq(t, len(departed.Children), 0)
	if departed.Operation != nil {
		t.Fatal("expected no diagnostic operation")
	}
}

// TS 2704: records a bounded candidate failure without retaining route identity
func TestRecordsBoundedCandidateFailureWithoutRetainingRouteIdentity(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, A, 0, ms(0))
	routes.Reconcile(10)
	prepared := beginOperation(t, routes, BeginInput{NowMs: 20, ConnectionID: "failed_candidate", Reservation: direct()})
	failed := routes.CandidateFailed(guardFor(A, prepared.Current.Revision, "failed_candidate"), 30)
	eq(t, failed.Accepted, true)
	eqSlice(t, failed.FailedPeerIDs, []string{A})
	child := routes.RouteDiagnosticSnapshot(30).Children[0]
	eqDeep(t, child.FinalMs, optInt(30))
	eq(t, child.FinalRoute, "failed")
	eq(t, child.RejectionBucket, "candidate-failed")

	routes.Dispose()
	disposed := routes.RouteDiagnosticSnapshot(40)
	eq(t, len(disposed.Children), 0)
	if disposed.Operation != nil {
		t.Fatal("expected no diagnostic operation after dispose")
	}
}

// TS 2737: reports a current child that has no usable candidate
func TestReportsCurrentChildThatHasNoUsableCandidate(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, A, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_connection"))
	addViewer(routes, B, 0, ms(0))

	result := routes.Reconcile(10)
	noOperation(t, result.Operation)
	eqSlice(t, result.FailedPeerIDs, []string{B})
	child := routes.RouteDiagnosticSnapshot(10).Children[1]
	eqDeep(t, child.FinalMs, optInt(10))
	eq(t, child.FinalRoute, "failed")
	eq(t, child.RejectionBucket, "candidate-failed")
}

// TS 2753: records SFU admission waiting without advancing the route cursor
func TestRecordsSfuAdmissionWaitingWithoutAdvancingRouteCursor(t *testing.T) {
	routes := newController(1, Options{SfuEnabled: true})
	addViewer(routes, A, 0, ms(0))
	directOp := must(t, routes.Reconcile(10).Operation)
	eqSlice(t, tupleKinds(directOp), []UpstreamKind{UpstreamPeer, UpstreamSfu})
	eq(t, routes.SkipCurrentCandidate(cursorGuard(directOp), 20, RejectionCandidateFailed).Accepted, true)
	sfu := must(t, routes.Snapshot().Operation)
	eq(t, routes.NoteCurrentCandidateRejection(cursorGuard(sfu), RejectionSfuAdmission), true)
	snapshot := routes.RouteDiagnosticSnapshot(30)
	eq(t, len(snapshot.Children), 1)
	eq(t, snapshot.Children[0].FinalRoute, "waiting")
	eq(t, snapshot.Children[0].RejectionBucket, "sfu-admission")
	if snapshot.Operation == nil {
		t.Fatal("expected a diagnostic operation")
	}
	eq(t, *snapshot.Operation, protocol.RouteDiagnosticOperation{
		ChildOrdinal: 1, Reason: "join", Stage: "admission", Cursor: 1, CandidateCount: 2,
	})
}

// TS 2791: distinguishes first-frame timeout from the total operation deadline
func TestDistinguishesFirstFrameTimeoutFromTotalOperationDeadline(t *testing.T) {
	routes := newController(1, Options{SfuEnabled: true})
	addViewer(routes, A, 0, ms(0))
	directOp := must(t, routes.Reconcile(10).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 20, ConnectionID: "silent_direct", Reservation: direct()})

	eq(t, routes.OperationExpired(directOp.WakeAtMs).Accepted, true)
	snapshot := routes.RouteDiagnosticSnapshot(directOp.WakeAtMs)
	eq(t, len(snapshot.Children), 1)
	eqDeep(t, snapshot.Children[0].CandidateStartMs, (*protocol.Int)(nil))
	eqDeep(t, snapshot.Children[0].FinalMs, (*protocol.Int)(nil))
	eq(t, snapshot.Children[0].FinalRoute, "waiting")
	eq(t, snapshot.Children[0].RejectionBucket, "first-frame-timeout")
	if snapshot.Operation == nil {
		t.Fatal("expected a diagnostic operation")
	}
	eq(t, snapshot.Operation.Cursor, 1)
	eq(t, snapshot.Operation.Stage, "admission")

	deadline := must(t, routes.Snapshot().Operation).DeadlineAtMs
	expired := routes.OperationExpired(deadline)
	eq(t, expired.Accepted, true)
	eqSlice(t, expired.FailedPeerIDs, []string{A})
	child := routes.RouteDiagnosticSnapshot(deadline).Children[0]
	eqDeep(t, child.FinalMs, optInt(deadline))
	eq(t, child.FinalRoute, "failed")
	eq(t, child.RejectionBucket, "operation-deadline")
}

// TS 2828: admits 20 Viewers through one bounded event-driven graph at C=%i
func TestAdmitsTwentyViewersThroughOneBoundedGraph(t *testing.T) {
	for _, capacity := range []int{1, 2, 3} {
		t.Run(fmt.Sprintf("C=%d", capacity), func(t *testing.T) {
			routes := newController(capacity, Options{})
			viewerPeerIDs := make([]string, 20)
			for index := range viewerPeerIDs {
				viewerPeerIDs[index] = fmt.Sprintf("viewer_%02d_12345678", index)
			}
			for index, peerID := range viewerPeerIDs {
				addViewer(routes, peerID, capacity, nil)
				operation := must(t, routes.Reconcile(int64(index*2)).Operation)
				eq(t, operation.ChildPeerID, peerID)
				commitCurrent(t, routes, int64(index*2+1), peerID+"_connection")
			}
			snapshot := routes.Snapshot()
			eq(t, snapshot.UpstreamByViewer.Len(), 20)
			for _, parentPeerID := range append([]string{HOST}, viewerPeerIDs...) {
				childCount := 0
				for _, edge := range snapshot.UpstreamByViewer.Values() {
					if edge.Kind == UpstreamPeer && edge.PhysicalActive && edge.ParentPeerID == parentPeerID {
						childCount++
					}
				}
				if childCount > capacity {
					t.Fatalf("parent %q has %d children, capacity %d", parentPeerID, childCount, capacity)
				}
			}
		})
	}
}

// TS 2856: admits waiting Viewers through existing SFU before direct convergence
func TestAdmitsWaitingViewersThroughExistingSfuBeforeDirectConvergence(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateHostPublication("publication_generation", res("publication_resource"), "")
	routes.hydrateEdge(B, sfuEdge("publication_generation", "b_sfu", "b_subscription"))
	addViewer(routes, C, 0, nil)
	routes.hydrateEdge(C, peerEdge(HOST, "c_from_host"))
	waiting := make([]string, 18)
	for index := range waiting {
		waiting[index] = fmt.Sprintf("silent_%02d_12345678", index)
	}
	for _, peerID := range waiting {
		addViewer(routes, peerID, 0, nil)
	}

	nowMs := int64(0)
	for index, peerID := range waiting {
		directOperation := must(t, routes.Reconcile(nowMs).Operation)
		eq(t, directOperation.ChildPeerID, peerID)
		eq(t, directOperation.Candidates[0].Tuple.Kind, UpstreamPeer)
		beginCandidate(t, routes, BeginInput{NowMs: nowMs + 1, ConnectionID: peerID + "_direct", Reservation: direct()})
		routes.OperationExpired(directOperation.WakeAtMs)
		sfuOperation := must(t, routes.Snapshot().Operation)
		eq(t, sfuOperation.Candidates[sfuOperation.Cursor].Tuple, sfuTuple(PublicationReuse))
		prepared := beginOperation(t, routes, BeginInput{
			NowMs: directOperation.WakeAtMs + 1, ConnectionID: peerID + "_sfu", Reservation: sfuReuse(peerID + "_subscription"),
		})
		eq(t, routes.CandidateReady(guardFor(peerID, prepared.Current.Revision, peerID+"_sfu"),
			directOperation.WakeAtMs+2, nil, CandidateProof{}).Accepted, true)
		nowMs = directOperation.WakeAtMs + 3

		bEdge := edgeOf(t, routes, B)
		eq(t, bEdge.Kind, UpstreamSfu)
		eq(t, bEdge.ConnectionID, "b_sfu")
		cEdge := edgeOf(t, routes, C)
		eq(t, cEdge.Kind, UpstreamPeer)
		eq(t, cEdge.ParentPeerID, HOST)
		eq(t, cEdge.ConnectionID, "c_from_host")
		admitted := edgeOf(t, routes, peerID)
		eq(t, admitted.Kind, UpstreamSfu)
		eq(t, admitted.Usable, true)
		eq(t, routes.Snapshot().UpstreamByViewer.Len(), index+3)
	}

	eq(t, routes.Snapshot().UpstreamByViewer.Len(), 20)
	publication := routes.Snapshot().HostPublication
	if publication == nil {
		t.Fatal("expected a host publication")
	}
	eq(t, publication.Generation, "publication_generation")
	eq(t, publication.Usable, true)
	eq(t, must(t, routes.Reconcile(nowMs).Operation).Reason, DemandDirectConvergence)
}

// TS 2943: builds a deterministic acyclic route without a depth cap at C=%i
func TestBuildsDeterministicAcyclicRouteWithoutDepthCap(t *testing.T) {
	for _, capacity := range []int{1, 2, 3} {
		t.Run(fmt.Sprintf("C=%d", capacity), func(t *testing.T) {
			routes := newController(capacity, Options{})
			for _, peerID := range []string{A, B, C, D} {
				addViewer(routes, peerID, capacity, nil)
				result := routes.Reconcile(0)
				eq(t, must(t, result.Operation).ChildPeerID, peerID)
				commitCurrent(t, routes, 1, peerID+"_connection")
			}

			snapshot := routes.Snapshot()
			eq(t, snapshot.UpstreamByViewer.Len(), 4)
			for childPeerID, edge := range snapshot.UpstreamByViewer.All() {
				eq(t, edge.Kind, UpstreamPeer)
				if edge.ParentPeerID == childPeerID {
					t.Fatalf("%q is its own parent", childPeerID)
				}
			}
			if capacity == 1 {
				// Pins the stablePairRank/depth ordering: at C=1 D hangs below C.
				dEdge := edgeOf(t, routes, D)
				eq(t, dEdge.Kind, UpstreamPeer)
				eq(t, dEdge.ParentPeerID, C)
			}
		})
	}
}

// TS 2969: keeps the old edge until exact candidate first-frame ready
func TestKeepsOldEdgeUntilExactCandidateFirstFrameReady(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	routes.Reconcile(0)
	commitCurrent(t, routes, 1, "old_connection")
	routes.Reconcile(2)
	commitCurrent(t, routes, 3, "b_connection")
	revision := routes.Snapshot().Revision
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", revision, "old_connection"), nil), true)

	routes.Reconcile(100)
	prepared := beginOperation(t, routes, BeginInput{NowMs: 101, ConnectionID: "new_connection", Reservation: direct()})
	eq(t, edgeOf(t, routes, A).ConnectionID, "old_connection")
	eq(t, routes.CandidateReady(guardFor(A, prepared.Current.Revision-1, "new_connection"), 102, nil, CandidateProof{}).Accepted, false)
	eq(t, routes.CandidateReady(guardFor(A, prepared.Current.Revision, "new_connection"), 102, nil, CandidateProof{}).Accepted, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.ConnectionID, "new_connection")
	eq(t, edge.ParentPeerID, B)
	eq(t, edge.Usable, true)
}

// TS 3020: invalidates one exact active Peer edge from its parent idempotently
func TestInvalidatesOneExactActivePeerEdgeFromParentIdempotently(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 0, nil)
	addViewer(routes, B, 1, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_from_host"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	report := ParentEdgeGuard{
		ParentPeerID:    HOST,
		ParentSessionID: "host_session",
		RouteRevision:   routes.Snapshot().Revision,
		ConnectionID:    "a_from_host",
	}

	eq(t, routes.InvalidateDirectEdgeFromParent(report, ms(10)), true)
	afterFirst := routes.Snapshot()
	aEdge, _ := afterFirst.UpstreamByViewer.Get(A)
	eq(t, aEdge.Usable, false)
	eq(t, aEdge.ConnectionID, "a_from_host")
	bEdge, _ := afterFirst.UpstreamByViewer.Get(B)
	eq(t, bEdge.Usable, true)
	eq(t, bEdge.ConnectionID, "b_from_host")
	eq(t, routes.InvalidateDirectEdgeFromParent(report, ms(11)), true)
	eq(t, routes.Snapshot().FactVersion, afterFirst.FactVersion)
	unknown := report
	unknown.ConnectionID = "unknown_connection"
	eq(t, routes.InvalidateDirectEdgeFromParent(unknown, nil), false)
	operation := must(t, routes.Reconcile(12).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.Reason, DemandEdgeUnavailable)
	eqSlice(t, tuples(operation), []CandidateTuple{peerTuple(B)})
}

// TS 3062: advances one candidate cursor without resetting the operation deadline
func TestAdvancesOneCandidateCursorWithoutResettingOperationDeadline(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, A, 2, nil)
	operation := must(t, routes.Reconcile(1_000).Operation)
	deadline := operation.DeadlineAtMs
	first := beginOperation(t, routes, BeginInput{NowMs: 1_001, ConnectionID: "candidate_1", Reservation: direct()})
	failed := routes.CandidateFailed(guardFor(A, first.Current.Revision, "candidate_1"), 1_001)
	eq(t, failed.Accepted, true)
	if failed.ActiveRevision <= first.Current.Revision {
		t.Fatalf("activeRevision %d did not advance past %d", failed.ActiveRevision, first.Current.Revision)
	}
	eqSlice(t, failed.FailedPeerIDs, []string{})
	eqLabels(t, failed.Released)
	pending := must(t, routes.Snapshot().Operation)
	eq(t, pending.Cursor, 1)
	eq(t, pending.DeadlineAtMs, deadline)

	second := beginOperation(t, routes, BeginInput{
		NowMs:                   1_002,
		ConnectionID:            "candidate_2",
		PublicationGeneration:   "publication_2",
		PublicationConnectionID: "publication_connection_2",
		Reservation:             sfuCreate("candidate_2_edge", "candidate_2_publication"),
	})
	lateReady := routes.CandidateReady(guardFor(A, second.Current.Revision, "candidate_2"), deadline, nil, CandidateProof{})
	eq(t, lateReady.Accepted, false)
	eqSlice(t, lateReady.FailedPeerIDs, []string{A})
	eqLabels(t, lateReady.Released, "candidate_2_edge", "candidate_2_publication")
	noOperation(t, routes.Snapshot().Operation)
	eq(t, routes.CandidateReady(guardFor(A, second.Current.Revision, "candidate_2"), deadline+1, nil, CandidateProof{}).Accepted, false)
	noOperation(t, routes.Reconcile(deadline+1).Operation)
	routes.TouchExternalFacts()
	eq(t, must(t, routes.Reconcile(deadline+2).Operation).ChildPeerID, A)
}

// TS 3123: reserves one derived deadline stage for direct and SFU
func TestReservesOneDerivedDeadlineStageForDirectAndSfu(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, A, 0, nil)

	operation := must(t, routes.Reconcile(0).Operation)
	eq(t, operation.DeadlineAtMs, 10_000)
	eq(t, operation.WakeAtMs, 5_000)
	for _, want := range []CandidateTuple{peerTuple(B), sfuTuple(PublicationCreate)} {
		if !slices.Contains(tuples(operation), want) {
			t.Fatalf("candidates %+v do not contain %+v", tuples(operation), want)
		}
	}

	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "silent_direct", Reservation: direct()})
	directExpired := routes.OperationExpired(operation.WakeAtMs)
	eq(t, directExpired.Accepted, true)
	eqSlice(t, directExpired.FailedPeerIDs, []string{})
	pending := must(t, routes.Snapshot().Operation)
	eq(t, pending.DeadlineAtMs, 10_000)
	eq(t, pending.WakeAtMs, 10_000)
	if pending.Current != nil {
		t.Fatal("expected no live attempt")
	}
	eq(t, pending.Candidates[pending.Cursor].Tuple.Kind, UpstreamSfu)

	prepared := beginOperation(t, routes, BeginInput{
		NowMs:                   7_000,
		ConnectionID:            "sfu_subscription",
		PublicationGeneration:   "publication_generation",
		PublicationConnectionID: "publication_connection",
		Reservation:             sfuCreate("sfu_edge", "sfu_publication"),
	})
	eq(t, routes.CandidateReady(guardFor(A, prepared.Current.Revision, "sfu_subscription"), 9_000, nil, CandidateProof{}).Accepted, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.Usable, true)
}

// TS 3186: uses the same direct head-start after an early hard failure
func TestUsesSameDirectHeadStartAfterEarlyHardFailure(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, D, 0, nil)
	routes.hydrateEdge(D, peerEdge(HOST, "host_full"))
	routes.hydrateHostPublication("publication_1", res("publication_resource"), "")
	for _, peerID := range []string{B, C} {
		addViewer(routes, peerID, 1, nil)
		routes.hydrateEdge(peerID, sfuEdge("publication_1", peerID+"_sfu", peerID+"_subscription"))
	}
	addViewer(routes, A, 0, nil)

	operation := must(t, routes.Reconcile(0).Operation)
	eqSlice(t, tupleKinds(operation), []UpstreamKind{UpstreamPeer, UpstreamSfu, UpstreamPeer})
	first := beginOperation(t, routes, BeginInput{NowMs: 1, ConnectionID: "first_direct", Reservation: direct()})
	eq(t, routes.CandidateFailed(guardFor(A, first.Current.Revision, "first_direct"), 2).Accepted, true)
	pending := must(t, routes.Snapshot().Operation)
	eq(t, pending.Cursor, 1)
	eq(t, pending.WakeAtMs, operation.WakeAtMs)
	eq(t, pending.Candidates[1].Tuple.Kind, UpstreamPeer)
}

// TS 3231: keeps a transport-connected direct candidate until the total deadline
func TestKeepsTransportConnectedDirectCandidateUntilTotalDeadline(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, D, 0, nil)
	routes.hydrateEdge(D, peerEdge(HOST, "host_full"))
	routes.hydrateHostPublication("publication", res("publication_resource"), "")
	addViewer(routes, B, 1, nil)
	routes.hydrateEdge(B, sfuEdge("publication", "b_sfu", "b_subscription"))
	addViewer(routes, A, 0, nil)

	operation := must(t, routes.Reconcile(0).Operation)
	directOp := beginOperation(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_direct", Reservation: direct()})
	eq(t, routes.CandidateTransportConnected(guardFor(A, directOp.Current.Revision, "a_direct"), 1_000).Accepted, true)
	eq(t, routes.CandidateTransportConnected(guardFor(A, directOp.Current.Revision, "a_direct"), 1_001).Accepted, true)
	eq(t, routes.CandidateTransportConnected(guardFor(A, directOp.Current.Revision, "stale_direct"), 1_002).Accepted, false)
	eq(t, routes.Snapshot().Revision, directOp.Current.Revision-1)
	eq(t, must(t, routes.Snapshot().Operation).WakeAtMs, operation.DeadlineAtMs)
	eq(t, routes.OperationExpired(operation.WakeAtMs).Accepted, false)
	eq(t, must(t, routes.Snapshot().Operation).Current.ConnectionID, "a_direct")
	eq(t, routes.CandidateReady(guardFor(A, directOp.Current.Revision, "a_direct"), operation.WakeAtMs+1, nil, CandidateProof{}).Accepted, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ConnectionID, "a_direct")
}

// TS 3288: keeps SFU active while converging through an SFU-fed parent
func TestKeepsSfuActiveWhileConvergingThroughSfuFedParent(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateHostPublication("publication_generation", res("publication_resource"), "")
	routes.hydrateEdge(B, sfuEdge("publication_generation", "b_sfu", "b_subscription"))
	addViewer(routes, C, 0, nil)
	routes.hydrateEdge(C, peerEdge(HOST, "c_from_host"))
	addViewer(routes, A, 0, nil)

	acquisition := must(t, routes.Reconcile(0).Operation)
	eq(t, acquisition.Candidates[0].Tuple, peerTuple(B))
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_direct_head_start", Reservation: direct()})
	routes.OperationExpired(acquisition.WakeAtMs)
	prepared := beginOperation(t, routes, BeginInput{
		NowMs: acquisition.WakeAtMs + 1, ConnectionID: "a_sfu_reuse", Reservation: sfuReuse("a_subscription"),
	})
	eq(t, routes.CandidateReady(guardFor(A, prepared.Current.Revision, "a_sfu_reuse"), acquisition.WakeAtMs+2, nil, CandidateProof{}).Accepted, true)
	publication := routes.Snapshot().HostPublication
	if publication == nil {
		t.Fatal("expected a host publication")
	}
	eq(t, publication.Generation, "publication_generation")
	eq(t, publication.Usable, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.PublicationGeneration, "publication_generation")
	eq(t, edge.Usable, true)
	convergence := must(t, routes.Reconcile(acquisition.WakeAtMs+3).Operation)
	eq(t, convergence.ChildPeerID, A)
	eq(t, convergence.Reason, DemandDirectConvergence)
	eq(t, convergence.Candidates[0].Tuple, peerTuple(B))
	directOp := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 4, ConnectionID: "a_from_b", Reservation: direct()})
	eq(t, must(t, routes.Snapshot().Operation).WakeAtMs, convergence.DeadlineAtMs)
	edge = edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.ConnectionID, "a_sfu_reuse")
	eq(t, routes.CandidateReady(guardFor(A, directOp.Current.Revision, "a_from_b"), acquisition.WakeAtMs+5, nil, CandidateProof{}).Accepted, true)
	edge = edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ParentPeerID, B)
	eq(t, edge.ConnectionID, "a_from_b")
}

// TS 3380: advances exact direct convergence failures without failing SFU
func TestAdvancesExactDirectConvergenceFailuresWithoutFailingSfu(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, D, 0, nil)
	routes.hydrateEdge(D, peerEdge(HOST, "host_full"))
	routes.hydrateHostPublication("publication_1", res("publication_resource"), "")
	for _, peerID := range []string{B, C} {
		addViewer(routes, peerID, 1, nil)
		routes.hydrateEdge(peerID, sfuEdge("publication_1", peerID+"_sfu", peerID+"_subscription"))
	}
	addViewer(routes, A, 0, ms(0))
	acquisition := must(t, routes.Reconcile(0).Operation)
	headStartTuple := acquisition.Candidates[0].Tuple
	eq(t, headStartTuple.Kind, UpstreamPeer)
	headStartParent := headStartTuple.ParentPeerID
	otherParent := B
	if headStartParent == B {
		otherParent = C
	}
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_direct_head_start", Reservation: direct()})
	routes.OperationExpired(acquisition.WakeAtMs)
	sfu := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 1, ConnectionID: "a_sfu", Reservation: sfuReuse("a_subscription")})
	eq(t, routes.CandidateReady(guardFor(A, sfu.Current.Revision, "a_sfu"), acquisition.WakeAtMs+2, nil, CandidateProof{}).Accepted, true)

	first := must(t, routes.Reconcile(acquisition.WakeAtMs+3).Operation)
	eq(t, first.Candidates[0].Tuple.Kind, UpstreamPeer)
	eq(t, first.Candidates[0].Tuple.ParentPeerID, otherParent)
	failed := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 4, ConnectionID: "a_direct_followup", Reservation: direct()})
	eqSlice(t, routes.CandidateFailed(guardFor(A, failed.Current.Revision, "a_direct_followup"), acquisition.WakeAtMs+5).FailedPeerIDs, []string{})
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.ConnectionID, "a_sfu")
	retry := must(t, routes.Reconcile(acquisition.WakeAtMs+6).Operation)
	eq(t, retry.Candidates[0].Tuple.Kind, UpstreamPeer)
	eq(t, retry.Candidates[0].Tuple.ParentPeerID, headStartParent)
}

// TS 3451: round-robins background direct convergence between SFU Viewers
func TestRoundRobinsBackgroundDirectConvergenceBetweenSfuViewers(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	routes.hydrateHostPublication("publication", res("publication_resource"), "")
	for _, peerID := range []string{B, C} {
		addViewer(routes, peerID, 2, nil)
		routes.hydrateEdge(peerID, sfuEdge("publication", peerID+"_sfu", peerID+"_subscription"))
	}

	nowMs := int64(0)
	for _, peerID := range []string{A, D} {
		addViewer(routes, peerID, 0, ms(nowMs))
		acquisition := must(t, routes.Reconcile(nowMs).Operation)
		beginCandidate(t, routes, BeginInput{NowMs: nowMs + 1, ConnectionID: peerID + "_head_start", Reservation: direct()})
		routes.OperationExpired(acquisition.WakeAtMs)
		sfu := beginOperation(t, routes, BeginInput{
			NowMs: acquisition.WakeAtMs + 1, ConnectionID: peerID + "_sfu", Reservation: sfuReuse(peerID + "_subscription"),
		})
		eq(t, routes.CandidateReady(guardFor(peerID, sfu.Current.Revision, peerID+"_sfu"), acquisition.WakeAtMs+2, nil, CandidateProof{}).Accepted, true)
		nowMs = acquisition.WakeAtMs + 3
	}

	first := must(t, routes.Reconcile(nowMs).Operation)
	eq(t, first.ChildPeerID, A)
	eq(t, first.Reason, DemandDirectConvergence)
	firstAttempt := beginOperation(t, routes, BeginInput{NowMs: nowMs + 1, ConnectionID: "a_background_direct", Reservation: direct()})
	eq(t, routes.CandidateFailed(guardFor(A, firstAttempt.Current.Revision, "a_background_direct"), nowMs+2).Accepted, true)
	next := must(t, routes.Reconcile(nowMs+3).Operation)
	eq(t, next.ChildPeerID, D)
	eq(t, next.Reason, DemandDirectConvergence)
}

// TS 3516: preempts direct convergence when another Viewer needs a route
func TestPreemptsDirectConvergenceWhenAnotherViewerNeedsRoute(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateHostPublication("publication_1", res("publication_resource"), "")
	routes.hydrateEdge(B, sfuEdge("publication_1", "b_sfu", "b_subscription"))
	addViewer(routes, C, 0, nil)
	routes.hydrateEdge(C, peerEdge(HOST, "c_from_host"))
	addViewer(routes, A, 0, ms(0))
	acquisition := must(t, routes.Reconcile(0).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_direct_head_start", Reservation: direct()})
	routes.OperationExpired(acquisition.WakeAtMs)
	sfu := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 1, ConnectionID: "a_sfu", Reservation: sfuReuse("a_subscription")})
	routes.CandidateReady(guardFor(A, sfu.Current.Revision, "a_sfu"), acquisition.WakeAtMs+2, nil, CandidateProof{})
	eq(t, must(t, routes.Reconcile(acquisition.WakeAtMs+3).Operation).Reason, DemandDirectConvergence)
	beginCandidate(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 4, ConnectionID: "a_direct_pending", Reservation: directOverlap("a_overlap")})

	activeRevision := routes.Snapshot().Revision
	eq(t, routes.InvalidateEdge(edgeGuard(C, "host_session", activeRevision, "c_from_host"), ms(acquisition.WakeAtMs+5)), true)
	reconciled := routes.Reconcile(acquisition.WakeAtMs + 6)
	eqLabels(t, reconciled.Released, "a_overlap")
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamSfu)
	eq(t, edge.ConnectionID, "a_sfu")
	preempted := must(t, reconciled.Operation)
	eq(t, preempted.ChildPeerID, C)
	eq(t, preempted.Reason, DemandEdgeUnavailable)
}

// TS 3579: owns Host-full SFU bootstrap failure by the waiting demand
func TestOwnsHostFullSfuBootstrapFailureByWaitingDemand(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, C, 0, nil)
	routes.hydrateEdge(C, peerEdge(HOST, "c_from_host"))
	addViewer(routes, A, 0, nil)
	addViewer(routes, D, 0, nil)

	operation := must(t, routes.Reconcile(0).Operation)
	eq(t, hasTuple(operation, isSfu), false)
	eq(t, operation.WakeAtMs, 5_000)
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_silent_direct", Reservation: direct()})
	blocked := routes.OperationExpired(operation.WakeAtMs)
	eq(t, blocked.Accepted, true)
	eqSlice(t, blocked.FailedPeerIDs, []string{})
	eqLabels(t, blocked.Released)
	noOperation(t, routes.Snapshot().Operation)

	bootstrap := must(t, routes.Reconcile(operation.WakeAtMs+1).Operation)
	eq(t, bootstrap.ChildPeerID, C)
	eq(t, bootstrap.DemandPeerID, A)
	eq(t, bootstrap.Candidates[0].Tuple, sfuTuple(PublicationCreate))
	eq(t, bootstrap.Candidates[0].EndpointTransition.Kind, TransitionOverlap)
	firstCarrier := beginOperation(t, routes, BeginInput{
		NowMs:                   operation.WakeAtMs + 2,
		ConnectionID:            "c_bootstrap_sfu",
		PublicationGeneration:   "c_bootstrap_publication",
		PublicationConnectionID: "c_bootstrap_ingress",
		Reservation:             sfuCreateOverlap("c_bootstrap_subscription", "c_bootstrap_publication_resource", "c_bootstrap_overlap"),
	})
	eqSlice(t, routes.CandidateFailed(guardFor(C, firstCarrier.Current.Revision, "c_bootstrap_sfu"), operation.WakeAtMs+3).FailedPeerIDs, []string{})
	cEdge := edgeOf(t, routes, C)
	eq(t, cEdge.ConnectionID, "c_from_host")
	eq(t, cEdge.Usable, true)
	eq(t, cEdge.PhysicalActive, true)

	secondBootstrap := must(t, routes.Reconcile(operation.WakeAtMs+4).Operation)
	eq(t, secondBootstrap.ChildPeerID, B)
	eq(t, secondBootstrap.DemandPeerID, A)
	eq(t, secondBootstrap.Reason, DemandSfuBootstrap)
	secondCarrier := beginOperation(t, routes, BeginInput{
		NowMs:                   operation.WakeAtMs + 5,
		ConnectionID:            "b_bootstrap_sfu",
		PublicationGeneration:   "b_bootstrap_publication",
		PublicationConnectionID: "b_bootstrap_ingress",
		Reservation:             sfuCreateOverlap("b_bootstrap_subscription", "b_bootstrap_publication_resource", "b_bootstrap_overlap"),
	})
	eqSlice(t, routes.CandidateFailed(guardFor(B, secondCarrier.Current.Revision, "b_bootstrap_sfu"), operation.WakeAtMs+6).FailedPeerIDs, []string{A})
	for _, seed := range [][2]string{{B, "b_from_host"}, {C, "c_from_host"}} {
		edge := edgeOf(t, routes, seed[0])
		eq(t, edge.ConnectionID, seed[1])
		eq(t, edge.Usable, true)
		eq(t, edge.PhysicalActive, true)
	}
	nextDemand := must(t, routes.Reconcile(operation.WakeAtMs+7).Operation)
	eq(t, nextDemand.ChildPeerID, D)
	eq(t, nextDemand.Reason, DemandJoin)
	beginCandidate(t, routes, BeginInput{NowMs: operation.WakeAtMs + 8, ConnectionID: "d_direct", Reservation: direct()})
	eqSlice(t, routes.OperationExpired(nextDemand.DeadlineAtMs).FailedPeerIDs, []string{D})
	noOperation(t, routes.Reconcile(nextDemand.DeadlineAtMs+1).Operation)
}

// TS 3680: abandons bootstrap when a Host slot becomes available
func TestAbandonsBootstrapWhenHostSlotBecomesAvailable(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	eq(t, routes.SetEffectiveCapacity(HOST, "host_session", 1, nil), true)
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, A, 0, nil)

	bootstrap := must(t, routes.Reconcile(0).Operation)
	eq(t, bootstrap.ChildPeerID, B)
	eq(t, bootstrap.DemandPeerID, A)
	eq(t, bootstrap.Reason, DemandSfuBootstrap)
	prepared := beginOperation(t, routes, BeginInput{
		NowMs:                   1,
		ConnectionID:            "stale_bootstrap",
		PublicationGeneration:   "stale_publication",
		PublicationConnectionID: "stale_ingress",
		Reservation:             sfuCreateOverlap("stale_subscription", "stale_publication_resource", "stale_overlap"),
	})

	eq(t, routes.SetEffectiveCapacity(HOST, "host_session", 2, nil), true)
	stale := routes.CandidateReady(guardFor(B, prepared.Current.Revision, "stale_bootstrap"), 2, nil, CandidateProof{})
	eq(t, stale.Accepted, false)
	containsLabels(t, stale.Released, "stale_subscription", "stale_publication_resource", "stale_overlap")
	bEdge := edgeOf(t, routes, B)
	eq(t, bEdge.ConnectionID, "b_from_host")
	eq(t, bEdge.Usable, true)
	eq(t, bEdge.PhysicalActive, true)
	join := must(t, routes.Reconcile(3).Operation)
	eq(t, join.ChildPeerID, A)
	eq(t, join.DemandPeerID, A)
	eq(t, join.Reason, DemandJoin)
}

// TS 3731: tries SFU first after bootstrap and retains the direct suffix
func TestTriesSfuFirstAfterBootstrapAndRetainsDirectSuffix(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	for _, seed := range [][2]string{{B, "b_from_host"}, {C, "c_from_host"}} {
		addViewer(routes, seed[0], 1, nil)
		routes.hydrateEdge(seed[0], peerEdge(HOST, seed[1]))
	}
	addViewer(routes, A, 0, nil)

	acquisition := must(t, routes.Reconcile(0).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_direct_head_start", Reservation: direct()})
	routes.OperationExpired(acquisition.WakeAtMs)
	bootstrap := must(t, routes.Reconcile(acquisition.WakeAtMs+1).Operation)
	carrier := beginOperation(t, routes, BeginInput{
		NowMs:                   acquisition.WakeAtMs + 2,
		ConnectionID:            "bootstrap_sfu",
		PublicationGeneration:   "bootstrap_publication",
		PublicationConnectionID: "bootstrap_ingress",
		Reservation:             sfuCreateOverlap("carrier_subscription", "bootstrap_publication_resource", "bootstrap_overlap"),
	})
	eq(t, routes.CandidateReady(CandidateGuard{
		ChildPeerID:    bootstrap.ChildPeerID,
		ChildSessionID: bootstrap.ChildSessionID,
		Revision:       carrier.Current.Revision,
		ConnectionID:   "bootstrap_sfu",
	}, acquisition.WakeAtMs+3, nil, CandidateProof{}).Accepted, true)

	demand := must(t, routes.Reconcile(acquisition.WakeAtMs+4).Operation)
	eq(t, demand.ChildPeerID, A)
	eq(t, demand.DemandPeerID, A)
	eq(t, demand.Reason, DemandJoin)
	eq(t, demand.Candidates[0].Tuple, sfuTuple(PublicationReuse))
	for _, tuple := range tuplesFrom(demand, 1) {
		eq(t, tuple.Kind, UpstreamPeer)
	}
	subscription := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 5, ConnectionID: "a_sfu", Reservation: sfuReuse("a_subscription")})
	eq(t, routes.CandidateFailed(guardFor(A, subscription.Current.Revision, "a_sfu"), acquisition.WakeAtMs+6).Accepted, true)
	pending := must(t, routes.Snapshot().Operation)
	eq(t, pending.Candidates[pending.Cursor].Tuple.Kind, UpstreamPeer)
	directOp := beginOperation(t, routes, BeginInput{NowMs: acquisition.WakeAtMs + 7, ConnectionID: "a_deferred_direct", Reservation: direct()})
	eq(t, routes.CandidateReady(guardFor(A, directOp.Current.Revision, "a_deferred_direct"), acquisition.WakeAtMs+8, nil, CandidateProof{}).Accepted, true)
	edge := edgeOf(t, routes, A)
	eq(t, edge.Kind, UpstreamPeer)
	eq(t, edge.ConnectionID, "a_deferred_direct")
	convergence := must(t, routes.Reconcile(acquisition.WakeAtMs+9).Operation)
	eq(t, convergence.ChildPeerID, bootstrap.ChildPeerID)
	eq(t, convergence.Reason, DemandDirectConvergence)
}

// TS 3819: replans when the current candidate parent fails
func TestReplansWhenCurrentCandidateParentFails(t *testing.T) {
	routes := newController(3, Options{})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 1, nil)
	addViewer(routes, C, 1, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_old"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_old"))
	routes.hydrateEdge(C, peerEdge(HOST, "c_old"))
	revision := routes.Snapshot().Revision
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", revision, "a_old"), nil), true)
	routes.Reconcile(0)
	current := beginOperation(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_new", Reservation: direct()})
	currentTuple := current.Current.Tuple
	eq(t, currentTuple.Kind, UpstreamPeer)
	failedParentPeerID := currentTuple.ParentPeerID
	failedParentConnectionID := "c_old"
	if failedParentPeerID == B {
		failedParentConnectionID = "b_old"
	}
	eq(t, routes.InvalidateEdge(edgeGuard(failedParentPeerID, "host_session", revision, failedParentConnectionID), nil), true)
	eq(t, routes.CandidateReady(guardFor(A, current.Current.Revision, "a_new"), 2, nil, CandidateProof{}).Accepted, false)
	replacement := beginOperation(t, routes, BeginInput{NowMs: 2, ConnectionID: "a_new_2", Reservation: direct()})
	eq(t, routes.CandidateReady(guardFor(A, replacement.Current.Revision, "a_new_2"), 3, nil, CandidateProof{}).Accepted, true)
	eq(t, must(t, routes.Reconcile(3).Operation).ChildPeerID, failedParentPeerID)
}

// TS 3886: keeps an active candidate when an unrelated session is replaced
func TestKeepsActiveCandidateWhenUnrelatedSessionIsReplaced(t *testing.T) {
	routes := newController(2, Options{})
	for _, seed := range [][2]string{{B, "b_from_host"}, {C, "c_from_host"}} {
		addViewer(routes, seed[0], 1, nil)
		routes.hydrateEdge(seed[0], peerEdge(HOST, seed[1]))
	}
	addViewer(routes, A, 0, nil)
	operation := must(t, routes.Reconcile(0).Operation)
	active := beginOperation(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_active_candidate", Reservation: direct()})
	tuple := active.Current.Tuple
	eq(t, tuple.Kind, UpstreamPeer)
	unrelatedPeerID := B
	if tuple.ParentPeerID == B {
		unrelatedPeerID = C
	}

	eqLabels(t, routes.UpsertParticipant(viewerInput(unrelatedPeerID, unrelatedPeerID+"_replacement", 1), ms(2)))
	current := must(t, routes.Snapshot().Operation).Current
	if current == nil {
		t.Fatal("expected the live attempt to survive")
	}
	eq(t, current.Revision, operation.BaseRevision+1)
	eq(t, current.ConnectionID, "a_active_candidate")
}

func TestUsesDepartedRelaysFreedHostSlotWithoutMovingHealthyDescendants(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 2, nil)
	addViewer(routes, C, 0, nil)
	addViewer(routes, D, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_root"))
	routes.hydrateEdge(B, peerEdge(HOST, "b_root"))
	routes.hydrateEdge(C, peerEdge(A, "c_from_a"))
	routes.hydrateEdge(D, peerEdge(B, "d_from_b"))

	eq(t, routes.ConfirmDeparture(B, nil), true)
	repair := must(t, routes.Reconcile(0).Operation)
	eq(t, repair.ChildPeerID, D)
	eq(t, repair.Candidates[0].Tuple, peerTuple(HOST))
	commitCurrent(t, routes, 1, "d_from_host")
	noOperation(t, routes.Reconcile(2).Operation)
	eq(t, hasEdge(routes, B), false)

	addViewer(routes, E, 0, nil)
	join := must(t, routes.Reconcile(3).Operation)
	eq(t, join.ChildPeerID, E)
	eq(t, join.Candidates[0].Tuple, peerTuple(A))
	commitCurrent(t, routes, 4, "e_from_a")

	cEdge := edgeOf(t, routes, C)
	eq(t, cEdge.ParentPeerID, A)
	eq(t, cEdge.ConnectionID, "c_from_a")
	dEdge := edgeOf(t, routes, D)
	eq(t, dEdge.ParentPeerID, HOST)
	eq(t, dEdge.ConnectionID, "d_from_host")
	eEdge := edgeOf(t, routes, E)
	eq(t, eEdge.ParentPeerID, A)
	eq(t, eEdge.ConnectionID, "e_from_a")
}

// TS 3967: keeps grace media, then reparents a confirmed relay departure
func TestKeepsGraceMediaThenReparentsConfirmedRelayDeparture(t *testing.T) {
	routes := newController(2, Options{})
	addViewer(routes, A, 2, nil)
	addViewer(routes, B, 1, nil)
	addViewer(routes, C, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_connection"))
	routes.hydrateEdge(B, peerEdge(A, "b_connection"))
	routes.hydrateEdge(C, peerEdge(B, "c_connection"))

	eq(t, routes.DisconnectSession(A, sessionOf(A)), true)
	noOperation(t, routes.Reconcile(0).Operation)
	eq(t, routes.ConfirmDeparture(A, nil), true)
	eq(t, must(t, routes.Reconcile(1).Operation).ChildPeerID, B)
	commitCurrent(t, routes, 2, "b_reparented")
	beforePrune := routes.Snapshot().Revision
	after := routes.Reconcile(3)
	if !slices.Contains(after.RemovedPeerIDs, A) {
		t.Fatalf("removedPeerIds %v does not contain A", after.RemovedPeerIDs)
	}
	if routes.Snapshot().Revision <= beforePrune {
		t.Fatalf("revision %d did not advance past %d", routes.Snapshot().Revision, beforePrune)
	}
	cEdge := edgeOf(t, routes, C)
	eq(t, cEdge.ParentPeerID, B)
	eq(t, cEdge.ConnectionID, "c_connection")
}

// TS 3991: moves newest overflow children and supports effective capacity zero
func TestMovesNewestOverflowChildrenAndSupportsEffectiveCapacityZero(t *testing.T) {
	routes := newController(3, Options{})
	addViewer(routes, A, 3, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "a_connection"))
	for _, peerID := range []string{B, C, D} {
		addViewer(routes, peerID, 0, nil)
		routes.hydrateEdge(peerID, peerEdge(A, peerID+"_connection"))
	}
	eq(t, routes.SetEffectiveCapacity(A, sessionOf(A), 1, nil), true)
	eq(t, must(t, routes.Reconcile(0).Operation).ChildPeerID, D)
	eqLabels(t, routes.SetPaused(true, nil))
	eq(t, routes.SetEffectiveCapacity(A, sessionOf(A), 0, nil), true)
	eqLabels(t, routes.SetPaused(false, nil))
	eq(t, must(t, routes.Reconcile(1).Operation).ChildPeerID, D)

	blocked := newController(1, Options{})
	addViewer(blocked, A, 0, nil)
	addViewer(blocked, B, 0, nil)
	blocked.hydrateEdge(A, peerEdge(HOST, "blocked_a"))
	blocked.hydrateEdge(B, peerEdge(A, "blocked_b"))
	noOperation(t, blocked.Reconcile(0).Operation)
	noOperation(t, blocked.Reconcile(1).Operation)
}

// TS 4015: aborts pending SFU resources on pause and repairs publication
func TestAbortsPendingSfuResourcesOnPauseAndRepairsPublication(t *testing.T) {
	routes := newController(1, Options{SfuEnabled: true})
	addViewer(routes, A, 1, nil)
	addViewer(routes, B, 0, nil)
	routes.hydrateHostPublication("publication_1", res("publication_resource"), "")
	routes.hydrateEdge(A, sfuEdge("publication_1", "a_sfu", "a_subscription"))
	operation := must(t, routes.Reconcile(0).Operation)
	eq(t, operation.Candidates[0].Tuple, peerTuple(A))
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "b_direct_head_start", Reservation: direct()})
	routes.OperationExpired(operation.WakeAtMs)
	beginCandidate(t, routes, BeginInput{NowMs: operation.WakeAtMs + 1, ConnectionID: "b_candidate", Reservation: sfuReuse("b_subscription")})
	eqLabels(t, routes.SetPaused(true, nil), "b_subscription")
	noOperation(t, routes.Snapshot().Operation)
	eq(t, edgeOf(t, routes, A).ConnectionID, "a_sfu")
	routes.SetPaused(false, nil)
	eq(t, routes.InvalidateHostPublication(PublicationGuard{
		HostSessionID: "host_session",
		RouteRevision: routes.Snapshot().Revision,
		Generation:    "publication_1",
		ConnectionID:  "publication:publication_1",
	}, nil), true)
	routes.TouchExternalFacts()
	repair := must(t, routes.Reconcile(2).Operation)
	eq(t, repair.ChildPeerID, A)
	eq(t, repair.Candidates[0].Tuple, peerTuple(HOST))
	skipCandidate(t, routes, 3)
	eq(t, must(t, routes.Snapshot().Operation).Candidates[1].Tuple, sfuTuple(PublicationReplace))
	expectPanic(t, "endpoint overlap reservation", func() {
		beginCandidate(t, routes, BeginInput{
			NowMs:                   3,
			ConnectionID:            "a_sfu_2",
			PublicationGeneration:   "publication_2",
			PublicationConnectionID: "publication_connection_2",
			Reservation:             sfuCreate("a_subscription_2", "publication_resource_2"),
		})
	})
	publicationRepair := beginOperation(t, routes, BeginInput{
		NowMs:                   3,
		ConnectionID:            "a_sfu_2",
		PublicationGeneration:   "publication_2",
		PublicationConnectionID: "publication_connection_2",
		Reservation:             sfuCreateOverlap("a_subscription_2", "publication_resource_2", "publication_overlap"),
	})
	repaired := routes.CandidateReady(guardFor(A, publicationRepair.Current.Revision, "a_sfu_2"), 4, nil, CandidateProof{})
	eq(t, repaired.Accepted, true)
	eq(t, len(repaired.Released), 3)
	containsLabels(t, repaired.Released, "publication_overlap", "a_subscription", "publication_resource")
	publication := routes.Snapshot().HostPublication
	if publication == nil {
		t.Fatal("expected a repaired publication")
	}
	eq(t, publication.Generation, "publication_2")
	eq(t, publication.Usable, true)
	eq(t, publication.PhysicalActive, true)

	bootstrap := newController(1, Options{SfuEnabled: true})
	addViewer(bootstrap, A, 0, nil)
	addViewer(bootstrap, B, 0, nil)
	bootstrap.hydrateEdge(A, peerEdge(HOST, "bootstrap_a"))
	bootstrapOperation := must(t, bootstrap.Reconcile(0).Operation)
	eq(t, bootstrapOperation.ChildPeerID, A)
	eq(t, bootstrapOperation.Candidates[0].Tuple, sfuTuple(PublicationCreate))
	expectPanic(t, "endpoint overlap reservation", func() {
		beginCandidate(t, bootstrap, BeginInput{
			NowMs:                   2,
			ConnectionID:            "bootstrap_sfu",
			PublicationGeneration:   "bootstrap_publication",
			PublicationConnectionID: "bootstrap_ingress_connection",
			Reservation:             sfuCreate("bootstrap_subscription", "bootstrap_ingress"),
		})
	})
	preparedBootstrap := beginOperation(t, bootstrap, BeginInput{
		NowMs:                   2,
		ConnectionID:            "bootstrap_sfu",
		PublicationGeneration:   "bootstrap_publication",
		PublicationConnectionID: "bootstrap_ingress_connection",
		Reservation:             sfuCreateOverlap("bootstrap_subscription", "bootstrap_ingress", "bootstrap_overlap"),
	})
	eq(t, bootstrap.CandidateReady(guardFor(A, preparedBootstrap.Current.Revision, "bootstrap_sfu"), 3, nil, CandidateProof{}).Accepted, true)
	eq(t, must(t, bootstrap.Reconcile(4).Operation).ChildPeerID, B)
}

// TS 4153: reopens SFU bootstrap after the physical publication retires
func TestReopensSfuBootstrapAfterPhysicalPublicationRetires(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, A, 0, nil)
	routes.hydrateHostPublication("retired_publication", res("publication_resource"), "")
	routes.hydrateEdge(A, sfuEdge("retired_publication", "a_from_sfu", "a_subscription"))

	publication := routes.Snapshot().HostPublication
	if publication == nil {
		t.Fatal("expected a host publication")
	}
	containsLabels(t, routes.RetireHostPublication(PublicationGuard{
		HostSessionID: publication.HostSessionID,
		RouteRevision: routes.Snapshot().Revision,
		Generation:    publication.Generation,
		ConnectionID:  publication.ConnectionID,
	}), "a_subscription", "publication_resource")
	if routes.Snapshot().HostPublication != nil {
		t.Fatal("expected no host publication")
	}

	for _, seed := range [][2]string{{B, "b_from_host"}, {C, "c_from_host"}} {
		addViewer(routes, seed[0], 0, nil)
		routes.hydrateEdge(seed[0], peerEdge(HOST, seed[1]))
	}
	reconciled := routes.Reconcile(1)
	eqSlice(t, reconciled.FailedPeerIDs, []string{})
	operation := must(t, reconciled.Operation)
	eq(t, operation.DemandPeerID, A)
	eq(t, operation.Reason, DemandSfuBootstrap)
	eqSlice(t, tuples(operation), []CandidateTuple{sfuTuple(PublicationCreate)})
}

// TS 4199: binds reserve and skip results to the exact candidate cursor
func TestBindsReserveAndSkipResultsToExactCandidateCursor(t *testing.T) {
	reserve := newController(2, Options{})
	addViewer(reserve, A, 2, nil)
	reserve.hydrateEdge(A, peerEdge(HOST, "a_active"))
	addViewer(reserve, B, 0, nil)
	first := must(t, reserve.Reconcile(0).Operation)
	eq(t, first.Candidates[0].Tuple.Kind, UpstreamPeer)
	eq(t, first.Candidates[0].Tuple.ParentPeerID, HOST)
	reserve.SetEffectiveCapacity(HOST, "host_session", 0, nil)
	staleReserve := reserve.BeginCurrentCandidate(BeginInput{
		Guard:        cursorGuard(first),
		NowMs:        1,
		ConnectionID: "reserved_for_old_cursor",
		Reservation:  directOverlap("stale_reservation"),
	})
	eq(t, staleReserve.Accepted, false)
	eqLabels(t, staleReserve.Released, "stale_reservation")
	rebuilt := must(t, reserve.Snapshot().Operation)
	eq(t, rebuilt.Cursor, 0)
	eqSlice(t, tuples(rebuilt), []CandidateTuple{peerTuple(A)})
	if rebuilt.Current != nil {
		t.Fatal("expected no live attempt")
	}

	skip := newController(2, Options{})
	addViewer(skip, A, 2, nil)
	skip.hydrateEdge(A, peerEdge(HOST, "a_active"))
	addViewer(skip, B, 0, nil)
	staleCursor := must(t, skip.Reconcile(0).Operation)
	skip.SetEffectiveCapacity(HOST, "host_session", 0, nil)
	eq(t, skip.SkipCurrentCandidate(cursorGuard(staleCursor), 1, RejectionStale).Accepted, false)
	pending := must(t, skip.Snapshot().Operation)
	eq(t, pending.Cursor, 0)
	eq(t, pending.Candidates[0].Tuple.Kind, UpstreamPeer)
	eq(t, pending.Candidates[0].Tuple.ParentPeerID, A)
}

// TS 4243: keeps only old SFU roots that still anchor peer descendants
func TestKeepsOnlyOldSfuRootsThatStillAnchorPeerDescendants(t *testing.T) {
	routes := newController(1, Options{SfuEnabled: true})
	addViewer(routes, A, 0, nil)
	addViewer(routes, B, 1, nil)
	addViewer(routes, C, 0, nil)
	addViewer(routes, D, 0, nil)
	routes.hydrateHostPublication("publication_1", res("publication_resource_1"), "")
	for _, seed := range [][2]string{{A, "subscription_a"}, {B, "subscription_b"}, {D, "subscription_d"}} {
		routes.hydrateEdge(seed[0], sfuEdge("publication_1", seed[0]+"_sfu_1", seed[1]))
	}
	routes.hydrateEdge(C, peerEdge(B, "c_from_b"))
	eq(t, routes.InvalidateHostPublication(PublicationGuard{
		HostSessionID: "host_session",
		RouteRevision: 0,
		Generation:    "publication_1",
		ConnectionID:  "publication:publication_1",
	}, ms(10)), true)
	routes.TouchExternalFacts()
	operation := must(t, routes.Reconcile(20).Operation)
	eq(t, operation.Candidates[0].Tuple, sfuTuple(PublicationReplace))
	expectTransition(t, operation.Candidates[0].EndpointTransition, TransitionOverlap, HOST)
	prepared := beginOperation(t, routes, BeginInput{
		NowMs:                   21,
		ConnectionID:            "a_sfu_2",
		PublicationGeneration:   "publication_2",
		PublicationConnectionID: "publication_connection_2",
		Reservation:             sfuCreateOverlap("subscription_a_2", "publication_resource_2", "publication_overlap"),
	})
	settled := routes.CandidateReady(guardFor(A, prepared.Current.Revision, "a_sfu_2"), 22, nil, CandidateProof{})
	eq(t, settled.Accepted, true)
	containsLabels(t, settled.Released, "publication_overlap", "subscription_a", "subscription_b", "subscription_d", "publication_resource_1")
	eq(t, edgeOf(t, routes, A).PublicationGeneration, "publication_2")
	bEdge := edgeOf(t, routes, B)
	eq(t, bEdge.Kind, UpstreamSfu)
	eq(t, bEdge.PublicationGeneration, "publication_1")
	eq(t, bEdge.Usable, false)
	eq(t, bEdge.PhysicalActive, false)
	cEdge := edgeOf(t, routes, C)
	eq(t, cEdge.Kind, UpstreamPeer)
	eq(t, cEdge.ParentPeerID, B)
	eq(t, hasEdge(routes, D), false)
	child := routes.RouteDiagnosticSnapshot(22).Children[1]
	eqDeep(t, child.DemandAgeMs, optInt(12))
	eq(t, child.FinalRoute, "waiting")
	next := must(t, routes.Reconcile(23).Operation)
	eq(t, next.ChildPeerID, B)
	eq(t, next.Reason, DemandEdgeUnavailable)
}

// TS 4328: rebinds direct and SFU media to replacement sessions
func TestRebindsDirectAndSfuMediaToReplacementSessions(t *testing.T) {
	peerRoutes := newController(2, Options{})
	addViewer(peerRoutes, A, 1, nil)
	addViewer(peerRoutes, B, 0, nil)
	peerRoutes.hydrateEdge(A, peerEdge(HOST, "a_direct"))
	peerRoutes.hydrateEdge(B, peerEdge(A, "b_direct"))
	eqLabels(t, peerRoutes.UpsertParticipant(viewerInput(A, "a_replacement", 1), nil))
	aEdge := edgeOf(t, peerRoutes, A)
	eq(t, aEdge.ChildSessionID, "a_replacement")
	eq(t, aEdge.Usable, true)
	eq(t, aEdge.PhysicalActive, true)
	eq(t, edgeOf(t, peerRoutes, B).ParentSessionID, "a_replacement")
	eqLabels(t, peerRoutes.UpsertParticipant(viewerInput(B, "b_replacement", 0), nil))
	bEdge := edgeOf(t, peerRoutes, B)
	eq(t, bEdge.ChildSessionID, "b_replacement")
	eq(t, bEdge.Usable, true)
	eq(t, bEdge.PhysicalActive, true)

	sfuRoutes := newController(1, Options{SfuEnabled: true})
	addViewer(sfuRoutes, A, 0, nil)
	sfuRoutes.hydrateHostPublication("publication", res("publication_resource"), "publication_connection")
	sfuRoutes.hydrateEdge(A, sfuEdge("publication", "subscription_connection", "subscription_resource"))
	eqLabels(t, sfuRoutes.UpsertParticipant(hostInput("host_replacement", 1), nil))
	publication := sfuRoutes.Snapshot().HostPublication
	if publication == nil {
		t.Fatal("expected a host publication")
	}
	eq(t, publication.HostSessionID, "host_replacement")
	eq(t, publication.Usable, true)
	eq(t, publication.PhysicalActive, true)
	eqLabels(t, sfuRoutes.UpsertParticipant(viewerInput(A, "a_replacement", 0), nil))
	sfuEdgeA := edgeOf(t, sfuRoutes, A)
	eq(t, sfuEdgeA.ChildSessionID, "a_replacement")
	eq(t, sfuEdgeA.PhysicalActive, true)
}

// TS 4399: retries a bootstrap carrier only when its transition improves
func TestRetriesBootstrapCarrierOnlyWhenTransitionImproves(t *testing.T) {
	routes := newController(1, Options{SfuEnabled: true})
	addViewer(routes, A, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "failed_direct"))
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", 0, "failed_direct"), nil), true)
	operation := must(t, routes.Reconcile(0).Operation)
	prepared := beginOperation(t, routes, BeginInput{
		NowMs:                   1,
		ConnectionID:            "failed_sfu",
		PublicationGeneration:   "failed_publication",
		PublicationConnectionID: "failed_publication_connection",
		Reservation:             sfuCreateOverlap("failed_subscription_resource", "failed_publication_resource", "failed_overlap"),
	})
	failed := routes.CandidateFailed(CandidateGuard{
		ChildPeerID:    operation.ChildPeerID,
		ChildSessionID: operation.ChildSessionID,
		Revision:       prepared.Current.Revision,
		ConnectionID:   "failed_sfu",
	}, 2)
	eq(t, failed.Accepted, true)
	eqSlice(t, failed.FailedPeerIDs, []string{A})
	improved := must(t, routes.Reconcile(3).Operation)
	eq(t, improved.Candidates[0].Tuple, sfuTuple(PublicationCreate))
	expectTransition(t, improved.Candidates[0].EndpointTransition, TransitionNone, HOST)
	improvedAttempt := beginOperation(t, routes, BeginInput{
		NowMs:                   4,
		ConnectionID:            "improved_sfu",
		PublicationGeneration:   "improved_publication",
		PublicationConnectionID: "improved_publication_connection",
		Reservation:             sfuCreate("improved_subscription_resource", "improved_publication_resource"),
	})
	failedAgain := routes.CandidateFailed(CandidateGuard{
		ChildPeerID:    improved.ChildPeerID,
		ChildSessionID: improved.ChildSessionID,
		Revision:       improvedAttempt.Current.Revision,
		ConnectionID:   "improved_sfu",
	}, 5)
	eq(t, failedAgain.Accepted, true)
	eqSlice(t, failedAgain.FailedPeerIDs, []string{A})
	factVersion := routes.Snapshot().FactVersion
	noOperation(t, routes.Reconcile(6).Operation)
	eq(t, routes.SetEffectiveCapacity(A, sessionOf(A), 0, nil), true)
	eq(t, routes.Snapshot().FactVersion, factVersion)
	noOperation(t, routes.Reconcile(7).Operation)
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", routes.Snapshot().Revision, "failed_direct"), nil), true)
	eq(t, routes.Snapshot().FactVersion, factVersion)
	noOperation(t, routes.Reconcile(8).Operation)

	addViewer(routes, B, 0, ms(9))
	eq(t, must(t, routes.Reconcile(9).Operation).ChildPeerID, B)
	// TS: `while (routes.snapshot().operation) skipCandidate(routes, 10);` bounded here so a broken port cannot hang.
	for rounds := 0; routes.Snapshot().Operation != nil; rounds++ {
		if rounds > 100 {
			t.Fatal("skipping never exhausted the operation")
		}
		skipCandidate(t, routes, 10)
	}
	noOperation(t, routes.Reconcile(11).Operation)
}

// TS 4472: requires one overlap slot for an SFU-to-Host transition at C=%i
func TestRequiresOneOverlapSlotForSfuToHostTransition(t *testing.T) {
	for _, capacity := range []int{1, 2} {
		t.Run(fmt.Sprintf("C=%d", capacity), func(t *testing.T) {
			routes := newController(capacity, Options{SfuEnabled: true})
			addViewer(routes, A, 0, nil)
			if capacity == 2 {
				addViewer(routes, E, 0, nil)
				routes.hydrateEdge(E, peerEdge(HOST, "e_direct"))
			}
			routes.hydrateHostPublication("publication_1", res("publication_resource_1"), "")
			routes.hydrateEdge(A, sfuEdge("publication_1", "a_sfu", "a_subscription"))
			routes.InvalidateEdge(edgeGuard(A, "", 0, "a_sfu"), nil)
			operation := must(t, routes.Reconcile(0).Operation)
			eq(t, operation.Candidates[0].Tuple, peerTuple(HOST))
			expectTransition(t, operation.Candidates[0].EndpointTransition, TransitionOverlap, HOST)
			expectPanic(t, "endpoint overlap reservation", func() {
				beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_host_direct", Reservation: direct()})
			})
		})
	}
}

// TS 4510: uses one bounded gap instead of a fourth Host copy at C=3
func TestUsesOneBoundedGapInsteadOfFourthHostCopyAtCapacityThree(t *testing.T) {
	routes := newController(3, Options{SfuEnabled: true})
	addViewer(routes, A, 0, nil)
	for _, peerID := range []string{E, F} {
		addViewer(routes, peerID, 0, nil)
		routes.hydrateEdge(peerID, peerEdge(HOST, peerID+"_direct"))
	}
	routes.hydrateHostPublication("publication_1", res("publication_resource_1"), "")
	routes.hydrateEdge(A, sfuEdge("publication_1", "a_sfu", "a_subscription"))
	routes.InvalidateEdge(edgeGuard(A, "", 0, "a_sfu"), nil)
	operation := must(t, routes.Reconcile(10).Operation)
	deadline := operation.DeadlineAtMs
	eq(t, operation.Candidates[0].Tuple, peerTuple(HOST))
	transition := operation.Candidates[0].EndpointTransition
	expectTransition(t, transition, TransitionBoundedGap, HOST)
	if transition.Retire == nil {
		t.Fatal("expected a retirement")
	}
	eq(t, transition.Retire.Kind, RetirePublication)
	eq(t, transition.Retire.Generation, "publication_1")
	gap := routes.RetireCurrentCandidateProducer(cursorGuard(operation), 11)
	eq(t, gap.Accepted, true)
	eqLabels(t, gap.Released, "a_subscription", "publication_resource_1")
	replanned := must(t, gap.Operation)
	eq(t, replanned.DeadlineAtMs, deadline)
	eq(t, len(replanned.Candidates), 1)
	expectTransition(t, replanned.Candidates[0].EndpointTransition, TransitionNone, HOST)
	prepared := beginOperation(t, routes, BeginInput{NowMs: 12, ConnectionID: "a_host_direct", Reservation: direct()})
	eq(t, routes.CandidateReady(guardFor(A, prepared.Current.Revision, "a_host_direct"), 13, nil, CandidateProof{}).Accepted, true)
	if routes.Snapshot().HostPublication != nil {
		t.Fatal("expected no host publication")
	}
}

// TS 4564: does not reopen consumed none transitions after capacity reduction
func TestDoesNotReopenConsumedNoneTransitionsAfterCapacityReduction(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	addViewer(routes, A, 0, nil)
	routes.hydrateHostPublication("publication", res("publication_resource"), "")
	routes.hydrateEdge(A, sfuEdge("publication", "a_from_sfu", "a_subscription"))
	eq(t, routes.InvalidateEdge(edgeGuard(A, "", 0, "a_from_sfu"), nil), true)
	operation := must(t, routes.Reconcile(0).Operation)
	eq(t, operation.Candidates[0].Tuple, peerTuple(HOST))
	expectTransition(t, operation.Candidates[0].EndpointTransition, TransitionNone, HOST)

	eq(t, routes.SetEffectiveCapacity(HOST, "host_session", 1, nil), true)
	eqSlice(t, routes.OperationExpired(operation.DeadlineAtMs).FailedPeerIDs, []string{A})
	noOperation(t, routes.Reconcile(operation.DeadlineAtMs+1).Operation)
}

// TS 4596: does not unblock an exhausted demand for an unrelated join
func TestDoesNotUnblockExhaustedDemandForUnrelatedJoin(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, B, 0, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "b_from_host"))
	addViewer(routes, A, 0, ms(0))
	eqSlice(t, routes.Reconcile(0).FailedPeerIDs, []string{A})

	eq(t, routes.SetEffectiveCapacity(A, sessionOf(A), 1, nil), true)
	noOperation(t, routes.Reconcile(1).Operation)

	addViewer(routes, C, 0, ms(2))
	eqSlice(t, routes.Reconcile(2).FailedPeerIDs, []string{C})
	noOperation(t, routes.Snapshot().Operation)

	eq(t, routes.SetEffectiveCapacity(B, sessionOf(B), 1, nil), true)
	operation := must(t, routes.Reconcile(3).Operation)
	eq(t, operation.ChildPeerID, A)
	eqSlice(t, tuples(operation), []CandidateTuple{peerTuple(B)})
}

// TS 4621: does not revive an exact failed Host tuple when a new parent joins
func TestDoesNotReviveExactFailedHostTupleWhenNewParentJoins(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, A, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "failed_direct"))
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", 0, "failed_direct"), nil), true)
	noOperation(t, routes.Reconcile(0).Operation)

	addViewer(routes, B, 1, ms(1))
	join := must(t, routes.Reconcile(2).Operation)
	eq(t, join.ChildPeerID, B)
	eq(t, join.Reason, DemandJoin)
	commitCurrent(t, routes, 2, "b_from_host")
	retry := must(t, routes.Reconcile(4).Operation)
	eq(t, retry.ChildPeerID, A)
	eqSlice(t, tuples(retry), []CandidateTuple{peerTuple(B)})
	eq(t, hasTuple(retry, func(tuple CandidateTuple) bool { return isPeer(tuple) && tuple.ParentPeerID == HOST }), false)
}

// TS 4660: does not unblock an exhausted Viewer when its candidate set shrinks
func TestDoesNotUnblockExhaustedViewerWhenCandidateSetShrinks(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, A, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "failed_host"))
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", 0, "failed_host"), nil), true)
	noOperation(t, routes.Reconcile(0).Operation)

	addViewer(routes, B, 1, ms(1))
	routes.Reconcile(1)
	commitCurrent(t, routes, 2, "b_from_host")
	candidate := must(t, routes.Reconcile(3).Operation)
	attempt := beginOperation(t, routes, BeginInput{NowMs: 4, ConnectionID: "a_from_b", Reservation: direct()})
	eqSlice(t, routes.CandidateFailed(guardFor(A, attempt.Current.Revision, "a_from_b"), 5).FailedPeerIDs, []string{A})
	eq(t, len(candidate.Candidates), 1)

	eq(t, routes.DisconnectSession(B, sessionOf(B)), true)
	noOperation(t, routes.Reconcile(6).Operation)
}

// TS 4694: treats a replacement child session as a new exact opportunity
func TestTreatsReplacementChildSessionAsNewExactOpportunity(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, A, 0, nil)
	routes.hydrateEdge(A, peerEdge(HOST, "failed_host"))
	eq(t, routes.InvalidateEdge(edgeGuard(A, "host_session", 0, "failed_host"), nil), true)
	noOperation(t, routes.Reconcile(0).Operation)

	routes.UpsertParticipant(viewerInput(A, "replacement_session", 0), nil)
	operation := must(t, routes.Reconcile(1).Operation)
	eq(t, operation.ChildPeerID, A)
	eq(t, operation.ChildSessionID, "replacement_session")
	eqSlice(t, tuples(operation), []CandidateTuple{peerTuple(HOST)})
}

// TS 4724: does not retry a failed bootstrap carrier when another carrier leaves
func TestDoesNotRetryFailedBootstrapCarrierWhenAnotherCarrierLeaves(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	for _, seed := range [][2]string{{B, "b_from_host"}, {C, "c_from_host"}} {
		addViewer(routes, seed[0], 1, nil)
		routes.hydrateEdge(seed[0], peerEdge(HOST, seed[1]))
	}
	addViewer(routes, A, 0, ms(0))
	acquisition := must(t, routes.Reconcile(0).Operation)
	beginCandidate(t, routes, BeginInput{NowMs: 1, ConnectionID: "a_head_start", Reservation: direct()})
	routes.OperationExpired(acquisition.WakeAtMs)

	bootstrap := must(t, routes.Reconcile(acquisition.WakeAtMs+1).Operation)
	eq(t, bootstrap.Reason, DemandSfuBootstrap)
	failedCarrier := bootstrap.ChildPeerID
	otherCarrier := B
	if failedCarrier == B {
		otherCarrier = C
	}
	attempt := beginOperation(t, routes, BeginInput{
		NowMs:                   acquisition.WakeAtMs + 2,
		ConnectionID:            "failed_bootstrap",
		PublicationGeneration:   "failed_publication",
		PublicationConnectionID: "failed_publication_connection",
		Reservation:             sfuCreateOverlap("failed_subscription", "failed_publication_resource", "failed_overlap"),
	})
	eq(t, routes.CandidateFailed(guardFor(failedCarrier, attempt.Current.Revision, "failed_bootstrap"), acquisition.WakeAtMs+3).Accepted, true)

	eq(t, routes.DisconnectSession(otherCarrier, sessionOf(otherCarrier)), true)
	next := routes.Reconcile(acquisition.WakeAtMs + 4).Operation
	if next != nil && next.Reason == DemandSfuBootstrap && next.ChildPeerID == failedCarrier {
		t.Fatalf("retried the failed bootstrap carrier %q", failedCarrier)
	}
}

// TS 4775: does not bootstrap a new SFU generation after the last reuse opportunity expires
func TestDoesNotBootstrapNewSfuGenerationAfterLastReuseOpportunityExpires(t *testing.T) {
	routes, nowMs := retireLastSfuPublicationAfterReuseDeadline(t)

	reconciled := routes.Reconcile(nowMs)
	noOperation(t, reconciled.Operation)
	eqSlice(t, reconciled.FailedPeerIDs, []string{})
}

// TS 4783: reopens the exhausted SFU opportunity for a new %s authority
func TestReopensExhaustedSfuOpportunityForNewAuthority(t *testing.T) {
	for _, authority := range []string{"child-session", "host-session", "external-resource"} {
		t.Run(authority, func(t *testing.T) {
			routes, nowMs := retireLastSfuPublicationAfterReuseDeadline(t)
			switch authority {
			case "child-session":
				routes.UpsertParticipant(viewerInput(A, "replacement_child_session", 0), nil)
			case "host-session":
				routes.UpsertParticipant(hostInput("replacement_host_session", 2), nil)
			default:
				routes.TouchExternalFacts()
			}

			operation := must(t, routes.Reconcile(nowMs).Operation)
			eq(t, operation.DemandPeerID, A)
			eq(t, operation.Reason, DemandSfuBootstrap)
			eqSlice(t, tuples(operation), []CandidateTuple{sfuTuple(PublicationCreate)})
		})
	}
}

// TS 4817: consumes the canonical SFU opportunity when an active edge fails
func TestConsumesCanonicalSfuOpportunityWhenActiveEdgeFails(t *testing.T) {
	routes := invalidateActiveSfuEdge(t)

	// TS: `operation?.candidates.some(sfu)` toBe(false) needs an operation.
	operation := must(t, routes.Reconcile(0).Operation)
	eq(t, hasTuple(operation, isSfu), false)
}

// TS 4826: reopens an active-edge SFU opportunity for a new %s authority
func TestReopensActiveEdgeSfuOpportunityForNewAuthority(t *testing.T) {
	for _, authority := range []string{"child-session", "host-session", "external-resource"} {
		t.Run(authority, func(t *testing.T) {
			routes := invalidateActiveSfuEdge(t)
			switch authority {
			case "child-session":
				routes.UpsertParticipant(viewerInput(A, "active_edge_child_replacement", 0), nil)
			case "host-session":
				routes.UpsertParticipant(hostInput("active_edge_host_replacement", 1), nil)
			default:
				routes.TouchExternalFacts()
			}

			operation := must(t, routes.Reconcile(0).Operation)
			eq(t, hasTuple(operation, func(tuple CandidateTuple) bool {
				return isSfu(tuple) && tuple.Publication == PublicationReuse
			}), true)
		})
	}
}

// TS 4860: does not turn a failed same-rank SFU replace into a create retry
func TestDoesNotTurnFailedSameRankSfuReplaceIntoCreateRetry(t *testing.T) {
	routes := newController(2, Options{SfuEnabled: true})
	routes.hydrateHostPublication("failed_publication", res("failed_publication_resource"), "")
	eq(t, routes.InvalidateHostPublication(PublicationGuard{
		HostSessionID: "host_session",
		RouteRevision: routes.Snapshot().Revision,
		Generation:    "failed_publication",
		ConnectionID:  "publication:failed_publication",
	}, nil), true)
	addViewer(routes, A, 0, nil)
	operation := must(t, routes.Reconcile(0).Operation)
	eq(t, operation.Candidates[0].Tuple.Kind, UpstreamPeer)
	skipCandidate(t, routes, 1)
	replaceCandidate := must(t, routes.Snapshot().Operation).Candidates[1]
	eq(t, replaceCandidate.Tuple, sfuTuple(PublicationReplace))
	expectTransition(t, replaceCandidate.EndpointTransition, TransitionNone, HOST)
	replace := beginOperation(t, routes, BeginInput{
		NowMs:                   2,
		ConnectionID:            "failed_replace",
		PublicationGeneration:   "replacement_publication",
		PublicationConnectionID: "replacement_ingress",
		Reservation:             sfuCreate("replacement_subscription", "replacement_publication_resource"),
	})
	eqSlice(t, routes.CandidateFailed(guardFor(A, replace.Current.Revision, "failed_replace"), 3).FailedPeerIDs, []string{A})

	addViewer(routes, B, 0, nil)
	eq(t, routes.ConfirmDeparture(B, nil), true)
	reconciled := routes.Reconcile(4)
	if routes.Snapshot().HostPublication != nil {
		t.Fatal("expected no host publication")
	}
	noOperation(t, reconciled.Operation)
	eqSlice(t, reconciled.FailedPeerIDs, []string{})
}

// TS 4913: reopens only SFU opportunities for a new external fact
func TestReopensOnlySfuOpportunitiesForNewExternalFact(t *testing.T) {
	routes := newController(1, Options{SfuEnabled: true})
	addViewer(routes, A, 0, ms(0))
	routes.Reconcile(0)
	directAttempt := beginOperation(t, routes, BeginInput{NowMs: 1, ConnectionID: "failed_direct", Reservation: direct()})
	eq(t, routes.CandidateFailed(guardFor(A, directAttempt.Current.Revision, "failed_direct"), 2).Accepted, true)
	sfuAttempt := beginOperation(t, routes, BeginInput{
		NowMs:                   3,
		ConnectionID:            "failed_sfu",
		PublicationGeneration:   "failed_publication",
		PublicationConnectionID: "failed_publication_connection",
		Reservation:             sfuCreate("failed_subscription", "failed_publication_resource"),
	})
	eq(t, routes.CandidateFailed(guardFor(A, sfuAttempt.Current.Revision, "failed_sfu"), 4).Accepted, true)
	noOperation(t, routes.Snapshot().Operation)
	noOperation(t, routes.Reconcile(5).Operation)

	routes.TouchExternalFacts()
	retried := must(t, routes.Reconcile(6).Operation)
	eq(t, len(retried.Candidates), 1)
	eq(t, retried.Candidates[0].Tuple, sfuTuple(PublicationCreate))
	expectTransition(t, retried.Candidates[0].EndpointTransition, TransitionNone, HOST)
	eq(t, hasTuple(retried, isPeer), false)
}

// TS 4963: fails all waiting demands without cutting capacity-three Host children
func TestFailsAllWaitingDemandsWithoutCuttingCapacityThreeHostChildren(t *testing.T) {
	routes := newController(3, Options{SfuEnabled: true})
	seeds := [][2]string{{A, "a_direct"}, {B, "b_direct"}, {C, "c_direct"}}
	for _, seed := range seeds {
		addViewer(routes, seed[0], 0, nil)
		routes.hydrateEdge(seed[0], peerEdge(HOST, seed[1]))
	}
	addViewer(routes, D, 0, nil)
	addViewer(routes, E, 0, nil)
	reconciled := routes.Reconcile(0)
	noOperation(t, reconciled.Operation)
	eqSlice(t, reconciled.FailedPeerIDs, []string{D, E})
	for _, seed := range seeds {
		edge := edgeOf(t, routes, seed[0])
		eq(t, edge.ConnectionID, seed[1])
		eq(t, edge.Usable, true)
		eq(t, edge.PhysicalActive, true)
	}
}

// TS 4991: does not promote when synchronous admission commit fails
func TestDoesNotPromoteWhenSynchronousAdmissionCommitFails(t *testing.T) {
	routes := newController(1, Options{})
	addViewer(routes, A, 0, nil)
	operation := must(t, routes.Reconcile(0).Operation)
	prepared := beginOperation(t, routes, BeginInput{NowMs: 1, ConnectionID: "candidate_connection", Reservation: direct()})
	settled := routes.CandidateReady(CandidateGuard{
		ChildPeerID:    operation.ChildPeerID,
		ChildSessionID: operation.ChildSessionID,
		Revision:       prepared.Current.Revision,
		ConnectionID:   "candidate_connection",
	}, 2, func(CandidateReservation) bool { return false }, CandidateProof{})
	eq(t, settled.Accepted, false)
	if settled.ActiveRevision <= prepared.Current.Revision {
		t.Fatalf("activeRevision %d did not advance past %d", settled.ActiveRevision, prepared.Current.Revision)
	}
	eq(t, hasEdge(routes, A), false)
}

// --- describe("bounded NAT candidate acquisition") -------------------------

func TestRunsFourRealGenerationsThenExhaustsUntilNewSession(t *testing.T) {
	for _, failure := range []string{"failed", "timeout"} {
		t.Run(failure, func(t *testing.T) {
			routes := newController(2, Options{NatPredictionEnabled: true, OperationTimeoutMs: 20_000})
			addViewer(routes, A, 0, nil)
			nowMs := int64(0)
			var previous *CandidateGuard
			for attempt := 1; attempt <= 4; attempt++ {
				operation, prepared, guard := preparePeer(t, routes, nowMs, fmt.Sprintf("nat_%d", attempt))
				eqDeep(t, prepared.Current.ConnectionAttempt, &ConnectionAttemptProgress{Current: attempt, Total: 4})
				eq(t, operation.DeadlineAtMs-nowMs, 20_000)
				eq(t, prepared.WakeAtMs, operation.DeadlineAtMs)
				if previous != nil {
					eq(t, routes.CandidateFailed(*previous, nowMs).Accepted, false)
				}
				eq(t, must(t, routes.Snapshot().Operation).Current.ConnectionID, fmt.Sprintf("nat_%d", attempt))
				var result SettleResult
				if failure == "timeout" {
					nowMs = operation.DeadlineAtMs
					result = routes.OperationExpired(nowMs)
				} else {
					nowMs++
					result = routes.CandidateFailed(guard, nowMs)
				}
				if attempt == 4 {
					eqSlice(t, result.FailedPeerIDs, []string{A})
				} else {
					eqSlice(t, result.FailedPeerIDs, []string{})
				}
				eq(t, routes.CandidateFailed(guard, nowMs).Accepted, false)
				previous = &guard
				nowMs++
			}
			noOperation(t, routes.Reconcile(nowMs).Operation)
			routes.UpsertParticipant(viewerInput(A, "replacement_session", 0), ms(nowMs))
			_, prepared, guard := preparePeer(t, routes, nowMs+1, "new_session")
			eq(t, attemptCurrent(t, prepared), 1)
			eq(t, routes.CandidateReady(guard, nowMs+2, nil, CandidateProof{}).Committed, true)
		})
	}
}

// TS 5067: gives waiting viewers their first opportunity before another viewer retries
func TestGivesWaitingViewersFirstOpportunityBeforeAnotherViewerRetries(t *testing.T) {
	routes := newController(2, Options{NatPredictionEnabled: true})
	addViewer(routes, A, 0, nil)
	addViewer(routes, B, 0, nil)
	for index, childPeerID := range []string{A, B, A, B, A, B, A, B} {
		_, prepared, guard := preparePeer(t, routes, int64(index*2), fmt.Sprintf("fair_%d", index))
		eq(t, prepared.ChildPeerID, childPeerID)
		eq(t, attemptCurrent(t, prepared), index/2+1)
		routes.CandidateFailed(guard, int64(index*2+1))
	}
	noOperation(t, routes.Reconcile(20).Operation)
}

// TS 5080: tries a newly usable parent before spending another attempt on an old parent
func TestTriesNewlyUsableParentBeforeSpendingAnotherAttemptOnOldParent(t *testing.T) {
	routes := newController(2, Options{NatPredictionEnabled: true})
	addViewer(routes, B, 2, nil)
	routes.hydrateEdge(B, peerEdge(HOST, "existing_b"))
	addViewer(routes, A, 0, nil)
	_, firstPrepared, firstGuard := preparePeer(t, routes, 0, "first_host")
	eq(t, firstPrepared.Current.Tuple.ParentPeerID, HOST)
	routes.CandidateFailed(firstGuard, 1)
	_, secondPrepared, secondGuard := preparePeer(t, routes, 2, "first_b")
	eq(t, secondPrepared.Current.Tuple.ParentPeerID, B)
	routes.CandidateFailed(secondGuard, 3)
	addViewer(routes, C, 2, ms(4))
	routes.hydrateEdge(C, peerEdge(HOST, "existing_c"))
	_, thirdPrepared, _ := preparePeer(t, routes, 5, "first_c")
	eq(t, thirdPrepared.Current.Tuple.ParentPeerID, C)
	eq(t, attemptCurrent(t, thirdPrepared), 1)
}

// TS 5098: keeps SFU's foreground timing and shares the attempt budget with round-robin direct continuations
func TestKeepsSfuForegroundTimingAndSharesAttemptBudgetWithDirectContinuations(t *testing.T) {
	routes := newController(2, Options{NatPredictionEnabled: true, SfuEnabled: true, OperationTimeoutMs: 20_000})
	nowMs := int64(0)
	for _, childPeerID := range []string{A, B} {
		addViewer(routes, childPeerID, 0, ms(nowMs))
		_, directPrepared, _ := preparePeer(t, routes, nowMs, childPeerID+"_first")
		eq(t, directPrepared.WakeAtMs, nowMs+5_000)
		nowMs += 5_000
		eqSlice(t, routes.OperationExpired(nowMs).FailedPeerIDs, []string{})
		reservation := sfuReuse("subscription_b")
		if childPeerID == A {
			reservation = sfuCreate("subscription_a", "publication")
		}
		sfu := beginOperation(t, routes, BeginInput{
			NowMs:                   nowMs,
			ConnectionID:            childPeerID + "_sfu",
			PublicationGeneration:   "sfu_generation",
			PublicationConnectionID: "sfu_publisher",
			Reservation:             reservation,
		})
		if sfu.Current.ConnectionAttempt != nil {
			t.Fatal("SFU attempts carry no NAT connection attempt")
		}
		nowMs++
		routes.CandidateReady(CandidateGuard{
			ChildPeerID:    childPeerID,
			ChildSessionID: sfu.ChildSessionID,
			Revision:       sfu.Current.Revision,
			ConnectionID:   childPeerID + "_sfu",
		}, nowMs, nil, CandidateProof{})
		nowMs++
	}
	for index, childPeerID := range []string{A, B, A, B, A, B} {
		_, retry, guard := preparePeer(t, routes, nowMs, fmt.Sprintf("background_%d", index))
		eq(t, retry.ChildPeerID, childPeerID)
		eq(t, retry.Reason, DemandDirectConvergence)
		eqDeep(t, retry.Current.ConnectionAttempt, &ConnectionAttemptProgress{Current: index/2 + 2, Total: 4})
		eq(t, retry.WakeAtMs-nowMs, 20_000)
		nowMs++
		eqSlice(t, routes.CandidateFailed(guard, nowMs).FailedPeerIDs, []string{})
		edge := edgeOf(t, routes, childPeerID)
		eq(t, edge.Kind, UpstreamSfu)
		eq(t, edge.Usable, true)
		nowMs++
	}
	noOperation(t, routes.Reconcile(nowMs).Operation)
}

// TS 5139: does not retry without NAT policy or when no candidate was created
func TestDoesNotRetryWithoutNatPolicyOrWhenNoCandidateWasCreated(t *testing.T) {
	ordinary := newController(2, Options{})
	addViewer(ordinary, A, 0, nil)
	_, firstPrepared, firstGuard := preparePeer(t, ordinary, 0, "ordinary")
	if firstPrepared.Current.ConnectionAttempt != nil {
		t.Fatal("expected no connection attempt without NAT prediction")
	}
	eqSlice(t, ordinary.CandidateFailed(firstGuard, 1).FailedPeerIDs, []string{A})
	noOperation(t, ordinary.Reconcile(2).Operation)

	routes := newController(2, Options{NatPredictionEnabled: true})
	addViewer(routes, A, 0, nil)
	operation := must(t, routes.Reconcile(0).Operation)
	routes.SkipCurrentCandidate(cursorGuard(operation), 1, RejectionCandidateFailed)
	noOperation(t, routes.Reconcile(2).Operation)
}