feat(broker): schedule demand-aware re-probe freshness (C2)

RegistryConfig gains DemandFreshForLocal/DemandFreshForRemote: while
Registry.SetDemand reports a live client subscription, apply and
applyLocal schedule NextDue on the faster cadence (~1s local, ~2s
remote) instead of the passive 15s defaultFreshFor. Service.Subscribe
and serviceSubscription.Close wire the demand counter to the actual
subscription lifecycle, and hint() wakes the run loop so a transition
takes effect without waiting out the current cadence.

Also make OpenStream request a local re-probe as soon as a local
Control/Attachment stream opens, not only once it ends, so a session
created or claimed by an in-progress attach appears in `ls --all`/
other pickers promptly while the client stays attached.

commit: 9a44abcf7e67c94c94a6df7ca214149edc057833
author: brice · 30d ago · 2026-09-22 20:18 UTC
parent: 1f5bcd2
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

5 files · +299 -12

M internal/usecase/broker/registry.go +73 -2


@@ -29,6 +29,14 @@ const (
29	29		defaultFreshFor = 15 * time.Second
30	30		defaultRetryBase = 5 * time.Second
31	31		defaultRetryLimit = time.Minute
	32	+	// defaultDemandFreshForLocal and defaultDemandFreshForRemote are the
	33	+	// built-in demand cadences RegistryConfig falls back to when
	34	+	// DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only
	35	+	// while at least one client subscription is live (see Registry.SetDemand);
	36	+	// otherwise defaultFreshFor governs every re-probe schedule exactly as
	37	+	// before.
	38	+	defaultDemandFreshForLocal = 1 * time.Second
	39	+	defaultDemandFreshForRemote = 2 * time.Second
32	40		)
33	41
34	42		// Run lifecycle states. A registry runs exactly once: runIdle is the zero

@@ -67,6 +75,14 @@ type RegistryConfig struct {
67	75		// of every remote, and never persists it. Local is refused together with
68	76		// ObservationDisabled: a read-only registry owns no local producer.
69	77		Local *LocalObservation
	78	+	// DemandFreshForLocal and DemandFreshForRemote are the re-probe cadences
	79	+	// apply/applyLocal schedule into NextDue while Registry.SetDemand reports
	80	+	// at least one live client subscription. A zero value falls back to
	81	+	// defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote
	82	+	// (about 2s). With no live subscription, NextDue keeps the passive
	83	+	// defaultFreshFor (15s) exactly as before this field existed.
	84	+	DemandFreshForLocal time.Duration
	85	+	DemandFreshForRemote time.Duration
70	86		}
71	87
72	88		// Registry owns configured hosts and their immutable observation projection.

@@ -144,6 +160,16 @@ type Registry struct {
144	160		retryBase time.Duration
145	161		retryMax time.Duration
146	162		jitter func(base time.Duration, endpoint string, attempt uint64) time.Duration
	163	+
	164	+	// demandFreshForLocal and demandFreshForRemote are the resolved (non-zero)
	165	+	// demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.
	166	+	demandFreshForLocal time.Duration
	167	+	demandFreshForRemote time.Duration
	168	+	// demand counts live client subscriptions: Registry.SetDemand(true) is one
	169	+	// subscription opening and SetDemand(false) is one closing. Guarded by mu.
	170	+	// While demand > 0, apply and applyLocal schedule NextDue on the faster
	171	+	// demand cadence instead of the passive freshFor.
	172	+	demand int
147	173		}
148	174
149	175		// NewRegistry restores a validated durable snapshot under a fresh broker epoch.

@@ -186,6 +212,14 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
186	212		if generator == nil {
187	213		generator = func() (id [16]byte, err error) { _, err = rand.Read(id[:]); return id, err }
188	214		}
	215	+	demandFreshForLocal := cfg.DemandFreshForLocal
	216	+	if demandFreshForLocal <= 0 {
	217	+	demandFreshForLocal = defaultDemandFreshForLocal
	218	+	}
	219	+	demandFreshForRemote := cfg.DemandFreshForRemote
	220	+	if demandFreshForRemote <= 0 {
	221	+	demandFreshForRemote = defaultDemandFreshForRemote
	222	+	}
189	223		r := &Registry{
190	224		epoch: epoch, probe: probe, clock: clock, log: log,
191	225		observationDisabled: cfg.ObservationDisabled,

@@ -194,7 +228,9 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
194	228		pending: make(map[string]bool), tombstones: make(map[string]ports.BrokerHostTombstone),
195	229		subs: make(map[*subscription]struct{}), wake: make(chan struct{}, 1),
196	230		freshFor: defaultFreshFor, retryBase: defaultRetryBase, retryMax: defaultRetryLimit,
197		-	jitter: jittered,
	231	+	jitter: jittered,
	232	+	demandFreshForLocal: demandFreshForLocal,
	233	+	demandFreshForRemote: demandFreshForRemote,
198	234		}
199	235		if cfg.ObservationDisabled && cfg.Local != nil {
200	236		return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")

@@ -862,6 +898,41 @@ func (r *Registry) RequestProbe(endpoint string) {
862	898		r.hint()
863	899		}
864	900
	901	+	// SetDemand records one live client subscription opening (active=true) or
	902	+	// closing (active=false). internal/usecase/broker/service.go calls it from
	903	+	// Service.Subscribe and serviceSubscription.Close, so the demand count tracks
	904	+	// exactly the connections a client is actively watching. While the count is
	905	+	// above zero, apply and applyLocal schedule NextDue on the faster demand
	906	+	// cadence (demandFreshForLocal/demandFreshForRemote) instead of the passive
	907	+	// defaultFreshFor. The transition takes effect for the next observation to
	908	+	// complete; hint wakes the run loop so a registry idling on its timer
	909	+	// re-evaluates promptly rather than waiting out the current cadence. A
	910	+	// disabled registry never observes, so demand is still counted (Close must
	911	+	// stay balanced) but never changes any schedule.
	912	+	func (r *Registry) SetDemand(active bool) {
	913	+	r.mu.Lock()
	914	+	if active {
	915	+	r.demand++
	916	+	} else if r.demand > 0 {
	917	+	r.demand--
	918	+	}
	919	+	r.mu.Unlock()
	920	+	r.hint()
	921	+	}
	922	+
	923	+	// freshForLocked returns the re-probe cadence to stamp into NextDue: the
	924	+	// faster demand cadence while a client subscription is live, otherwise the
	925	+	// passive default. Callers must hold r.mu.
	926	+	func (r *Registry) freshForLocked(local bool) time.Duration {
	927	+	if r.demand <= 0 {
	928	+	return r.freshFor
	929	+	}
	930	+	if local {
	931	+	return r.demandFreshForLocal
	932	+	}
	933	+	return r.demandFreshForRemote
	934	+	}
	935	+
865	936		func (r *Registry) hint() {
866	937		select {
867	938		case r.wake <- struct{}{}:

@@ -1044,7 +1115,7 @@ func (r *Registry) apply(result probeResult) {
1044	1115		observed.Checking = false
1045	1116		observed.LastAttempt = result.at
1046	1117		observed.LastSuccess = result.at
1047		-	observed.NextDue = result.at.Add(r.jitter(r.freshFor, result.endpoint, result.attempt))
	1118	+	observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
1048	1119		observed.ConsecutiveFailures = 0
1049	1120		observed.LastFailure = domain.RemoteFailure{}
1050	1121		observed.FailureEpisode = current.FailureEpisode


@@ -29,6 +29,14 @@ const (
29		defaultFreshFor = 15 * time.Second	29		defaultFreshFor = 15 * time.Second
30		defaultRetryBase = 5 * time.Second	30		defaultRetryBase = 5 * time.Second
31		defaultRetryLimit = time.Minute	31		defaultRetryLimit = time.Minute
			32	+	// defaultDemandFreshForLocal and defaultDemandFreshForRemote are the
			33	+	// built-in demand cadences RegistryConfig falls back to when
			34	+	// DemandFreshForLocal/DemandFreshForRemote is left zero. They apply only
			35	+	// while at least one client subscription is live (see Registry.SetDemand);
			36	+	// otherwise defaultFreshFor governs every re-probe schedule exactly as
			37	+	// before.
			38	+	defaultDemandFreshForLocal = 1 * time.Second
			39	+	defaultDemandFreshForRemote = 2 * time.Second
32		)	40		)
33			41
34		// Run lifecycle states. A registry runs exactly once: runIdle is the zero	42		// Run lifecycle states. A registry runs exactly once: runIdle is the zero

@@ -67,6 +75,14 @@ type RegistryConfig struct {
67		// of every remote, and never persists it. Local is refused together with	75		// of every remote, and never persists it. Local is refused together with
68		// ObservationDisabled: a read-only registry owns no local producer.	76		// ObservationDisabled: a read-only registry owns no local producer.
69		Local *LocalObservation	77		Local *LocalObservation
			78	+	// DemandFreshForLocal and DemandFreshForRemote are the re-probe cadences
			79	+	// apply/applyLocal schedule into NextDue while Registry.SetDemand reports
			80	+	// at least one live client subscription. A zero value falls back to
			81	+	// defaultDemandFreshForLocal (about 1s) or defaultDemandFreshForRemote
			82	+	// (about 2s). With no live subscription, NextDue keeps the passive
			83	+	// defaultFreshFor (15s) exactly as before this field existed.
			84	+	DemandFreshForLocal time.Duration
			85	+	DemandFreshForRemote time.Duration
70		}	86		}
71			87
72		// Registry owns configured hosts and their immutable observation projection.	88		// Registry owns configured hosts and their immutable observation projection.

@@ -144,6 +160,16 @@ type Registry struct {
144		retryBase time.Duration	160		retryBase time.Duration
145		retryMax time.Duration	161		retryMax time.Duration
146		jitter func(base time.Duration, endpoint string, attempt uint64) time.Duration	162		jitter func(base time.Duration, endpoint string, attempt uint64) time.Duration
			163	+
			164	+	// demandFreshForLocal and demandFreshForRemote are the resolved (non-zero)
			165	+	// demand cadences; see RegistryConfig.DemandFreshForLocal/Remote.
			166	+	demandFreshForLocal time.Duration
			167	+	demandFreshForRemote time.Duration
			168	+	// demand counts live client subscriptions: Registry.SetDemand(true) is one
			169	+	// subscription opening and SetDemand(false) is one closing. Guarded by mu.
			170	+	// While demand > 0, apply and applyLocal schedule NextDue on the faster
			171	+	// demand cadence instead of the passive freshFor.
			172	+	demand int
147		}	173		}
148			174
149		// NewRegistry restores a validated durable snapshot under a fresh broker epoch.	175		// NewRegistry restores a validated durable snapshot under a fresh broker epoch.

@@ -186,6 +212,14 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
186		if generator == nil {	212		if generator == nil {
187		generator = func() (id [16]byte, err error) { _, err = rand.Read(id[:]); return id, err }	213		generator = func() (id [16]byte, err error) { _, err = rand.Read(id[:]); return id, err }
188		}	214		}
			215	+	demandFreshForLocal := cfg.DemandFreshForLocal
			216	+	if demandFreshForLocal <= 0 {
			217	+	demandFreshForLocal = defaultDemandFreshForLocal
			218	+	}
			219	+	demandFreshForRemote := cfg.DemandFreshForRemote
			220	+	if demandFreshForRemote <= 0 {
			221	+	demandFreshForRemote = defaultDemandFreshForRemote
			222	+	}
189		r := &Registry{	223		r := &Registry{
190		epoch: epoch, probe: probe, clock: clock, log: log,	224		epoch: epoch, probe: probe, clock: clock, log: log,
191		observationDisabled: cfg.ObservationDisabled,	225		observationDisabled: cfg.ObservationDisabled,

@@ -194,7 +228,9 @@ func NewRegistryWithConfig(epoch ports.BrokerEpoch, store ports.BrokerHostStore,
194		pending: make(map[string]bool), tombstones: make(map[string]ports.BrokerHostTombstone),	228		pending: make(map[string]bool), tombstones: make(map[string]ports.BrokerHostTombstone),
195		subs: make(map[*subscription]struct{}), wake: make(chan struct{}, 1),	229		subs: make(map[*subscription]struct{}), wake: make(chan struct{}, 1),
196		freshFor: defaultFreshFor, retryBase: defaultRetryBase, retryMax: defaultRetryLimit,	230		freshFor: defaultFreshFor, retryBase: defaultRetryBase, retryMax: defaultRetryLimit,
197	-	jitter: jittered,	231	+	jitter: jittered,
			232	+	demandFreshForLocal: demandFreshForLocal,
			233	+	demandFreshForRemote: demandFreshForRemote,
198		}	234		}
199		if cfg.ObservationDisabled && cfg.Local != nil {	235		if cfg.ObservationDisabled && cfg.Local != nil {
200		return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")	236		return nil, errors.New("broker: observation-disabled registry cannot observe the local daemon")

@@ -862,6 +898,41 @@ func (r *Registry) RequestProbe(endpoint string) {
862		r.hint()	898		r.hint()
863		}	899		}
864			900
			901	+	// SetDemand records one live client subscription opening (active=true) or
			902	+	// closing (active=false). internal/usecase/broker/service.go calls it from
			903	+	// Service.Subscribe and serviceSubscription.Close, so the demand count tracks
			904	+	// exactly the connections a client is actively watching. While the count is
			905	+	// above zero, apply and applyLocal schedule NextDue on the faster demand
			906	+	// cadence (demandFreshForLocal/demandFreshForRemote) instead of the passive
			907	+	// defaultFreshFor. The transition takes effect for the next observation to
			908	+	// complete; hint wakes the run loop so a registry idling on its timer
			909	+	// re-evaluates promptly rather than waiting out the current cadence. A
			910	+	// disabled registry never observes, so demand is still counted (Close must
			911	+	// stay balanced) but never changes any schedule.
			912	+	func (r *Registry) SetDemand(active bool) {
			913	+	r.mu.Lock()
			914	+	if active {
			915	+	r.demand++
			916	+	} else if r.demand > 0 {
			917	+	r.demand--
			918	+	}
			919	+	r.mu.Unlock()
			920	+	r.hint()
			921	+	}
			922	+
			923	+	// freshForLocked returns the re-probe cadence to stamp into NextDue: the
			924	+	// faster demand cadence while a client subscription is live, otherwise the
			925	+	// passive default. Callers must hold r.mu.
			926	+	func (r *Registry) freshForLocked(local bool) time.Duration {
			927	+	if r.demand <= 0 {
			928	+	return r.freshFor
			929	+	}
			930	+	if local {
			931	+	return r.demandFreshForLocal
			932	+	}
			933	+	return r.demandFreshForRemote
			934	+	}
			935	+
865		func (r *Registry) hint() {	936		func (r *Registry) hint() {
866		select {	937		select {
867		case r.wake <- struct{}{}:	938		case r.wake <- struct{}{}:

@@ -1044,7 +1115,7 @@ func (r *Registry) apply(result probeResult) {
1044		observed.Checking = false	1115		observed.Checking = false
1045		observed.LastAttempt = result.at	1116		observed.LastAttempt = result.at
1046		observed.LastSuccess = result.at	1117		observed.LastSuccess = result.at
1047	-	observed.NextDue = result.at.Add(r.jitter(r.freshFor, result.endpoint, result.attempt))	1118	+	observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
1048		observed.ConsecutiveFailures = 0	1119		observed.ConsecutiveFailures = 0
1049		observed.LastFailure = domain.RemoteFailure{}	1120		observed.LastFailure = domain.RemoteFailure{}
1050		observed.FailureEpisode = current.FailureEpisode	1121		observed.FailureEpisode = current.FailureEpisode

M internal/usecase/broker/registry_local.go +1 -1


@@ -160,7 +160,7 @@ func (r *Registry) applyLocal(result probeResult) {
160	160		observed.Checking = false
161	161		observed.LastAttempt = result.at
162	162		observed.LastSuccess = result.at
163		-	observed.NextDue = result.at.Add(r.jitter(r.freshFor, localProbeKey, result.attempt))
	163	+	observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(true), localProbeKey, result.attempt))
164	164		observed.ConsecutiveFailures = 0
165	165		observed.LastFailure = domain.RemoteFailure{}
166	166		observed.FailureEpisode = current.FailureEpisode


@@ -160,7 +160,7 @@ func (r *Registry) applyLocal(result probeResult) {
160		observed.Checking = false	160		observed.Checking = false
161		observed.LastAttempt = result.at	161		observed.LastAttempt = result.at
162		observed.LastSuccess = result.at	162		observed.LastSuccess = result.at
163	-	observed.NextDue = result.at.Add(r.jitter(r.freshFor, localProbeKey, result.attempt))	163	+	observed.NextDue = result.at.Add(r.jitter(r.freshForLocked(true), localProbeKey, result.attempt))
164		observed.ConsecutiveFailures = 0	164		observed.ConsecutiveFailures = 0
165		observed.LastFailure = domain.RemoteFailure{}	165		observed.LastFailure = domain.RemoteFailure{}
166		observed.FailureEpisode = current.FailureEpisode	166		observed.FailureEpisode = current.FailureEpisode

M internal/usecase/broker/registry_test.go +108 -0


@@ -1072,6 +1072,114 @@ func TestRegistrySchedulesFreshObservationAfterSuccess(t *testing.T) {
1072	1072		require.True(t, next.registration.Equal(reg))
1073	1073		}
1074	1074
	1075	+	// TestRegistryAppliesSubscriptionDemandFreshness pins C2: apply/applyLocal
	1076	+	// schedule NextDue on RegistryConfig's demand cadence while Registry.SetDemand
	1077	+	// reports a live client subscription, and keep the passive defaultFreshFor
	1078	+	// once demand drops back to zero. It exercises both the remote and the local
	1079	+	// observation paths on the same registry, since service.go's Subscribe/Close
	1080	+	// wiring feeds one shared demand counter for both.
	1081	+	func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) {
	1082	+	const (
	1083	+	demandLocal = 250 * time.Millisecond
	1084	+	demandRemote = 500 * time.Millisecond
	1085	+	)
	1086	+	tests := []struct {
	1087	+	name string
	1088	+	local bool
	1089	+	demand bool
	1090	+	wantAfter time.Duration
	1091	+	}{
	1092	+	{name: "remote no demand keeps default", local: false, demand: false, wantAfter: defaultFreshFor},
	1093	+	{name: "remote demand uses faster cadence", local: false, demand: true, wantAfter: demandRemote},
	1094	+	{name: "local no demand keeps default", local: true, demand: false, wantAfter: defaultFreshFor},
	1095	+	{name: "local demand uses faster cadence", local: true, demand: true, wantAfter: demandLocal},
	1096	+	}
	1097	+
	1098	+	for _, tt := range tests {
	1099	+	t.Run(tt.name, func(t *testing.T) {
	1100	+	start := time.Unix(100, 0)
	1101	+	clock := newManualClock(start)
	1102	+	remoteProbe := newTestProbe(1)
	1103	+	localProbe := newScriptedLocalProbe(1)
	1104	+	registry, err := NewRegistryWithConfig(1, newTestStore(), remoteProbe, clock, nil, RegistryConfig{
	1105	+	Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: localProbe},
	1106	+	DemandFreshForLocal: demandLocal,
	1107	+	DemandFreshForRemote: demandRemote,
	1108	+	})
	1109	+	require.NoError(t, err)
	1110	+	registry.jitter = identityJitter
	1111	+	reg := registration(t, "example.test", 1)
	1112	+	require.NoError(t, registry.setHosts(hostRecords(reg)))
	1113	+	startRegistry(t, registry)
	1114	+
	1115	+	if tt.demand {
	1116	+	registry.SetDemand(true)
	1117	+	t.Cleanup(func() { registry.SetDemand(false) })
	1118	+	}
	1119	+
	1120	+	if tt.local {
	1121	+	call := receiveLocalCall(t, localProbe)
	1122	+	call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
	1123	+	Identity: "local-daemon",
	1124	+	Incarnation: ports.BrokerDaemonIncarnation{9},
	1125	+	ProtocolVersion: protocol.Version,
	1126	+	Availability: domain.RemoteAvailabilityReachable,
	1127	+	}}
	1128	+	host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
	1129	+	return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
	1130	+	})
	1131	+	require.Equal(t, start.Add(tt.wantAfter), host.NextDue)
	1132	+	return
	1133	+	}
	1134	+	call := receiveCall(t, remoteProbe)
	1135	+	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
	1136	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1137	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
	1138	+	})
	1139	+	require.Equal(t, start.Add(tt.wantAfter), host.NextDue)
	1140	+	})
	1141	+	}
	1142	+	}
	1143	+
	1144	+	// TestRegistrySetDemandIsCountedNotBoolean pins that SetDemand is a balanced
	1145	+	// counter: two openings require two closings before the schedule reverts to
	1146	+	// the passive default, exactly as two concurrent client subscriptions would
	1147	+	// drive it through service.go.
	1148	+	func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) {
	1149	+	start := time.Unix(100, 0)
	1150	+	clock := newManualClock(start)
	1151	+	probe := newTestProbe(2)
	1152	+	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
	1153	+	registry.demandFreshForRemote = 500 * time.Millisecond
	1154	+	reg := registration(t, "example.test", 1)
	1155	+	require.NoError(t, registry.setHosts(hostRecords(reg)))
	1156	+	startRegistry(t, registry)
	1157	+
	1158	+	// Two subscriptions open; only one closes, so demand must stay active.
	1159	+	registry.SetDemand(true)
	1160	+	registry.SetDemand(true)
	1161	+	registry.SetDemand(false)
	1162	+
	1163	+	call := receiveCall(t, probe)
	1164	+	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
	1165	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1166	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
	1167	+	})
	1168	+	require.Equal(t, start.Add(500*time.Millisecond), host.NextDue, "one remaining demand unit must still use the faster cadence")
	1169	+
	1170	+	// The second, and last, subscription closes: demand drops back to zero.
	1171	+	registry.SetDemand(false)
	1172	+
	1173	+	clock.Advance(500 * time.Millisecond)
	1174	+	secondSuccess := start.Add(500 * time.Millisecond)
	1175	+	next := receiveCall(t, probe)
	1176	+	next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
	1177	+	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1178	+	return !host.Checking && host.LastSuccess.Equal(secondSuccess)
	1179	+	})
	1180	+	require.Equal(t, secondSuccess.Add(defaultFreshFor), host.NextDue, "no live demand must revert to the passive default cadence")
	1181	+	}
	1182	+
1075	1183		func TestRegistryDemandOverridesFreshSchedule(t *testing.T) {
1076	1184		start := time.Unix(100, 0)
1077	1185		clock := newManualClock(start)


@@ -1072,6 +1072,114 @@ func TestRegistrySchedulesFreshObservationAfterSuccess(t *testing.T) {
1072	require.True(t, next.registration.Equal(reg))	1072		require.True(t, next.registration.Equal(reg))
1073	}	1073		}
1074		1074
		1075	+	// TestRegistryAppliesSubscriptionDemandFreshness pins C2: apply/applyLocal
		1076	+	// schedule NextDue on RegistryConfig's demand cadence while Registry.SetDemand
		1077	+	// reports a live client subscription, and keep the passive defaultFreshFor
		1078	+	// once demand drops back to zero. It exercises both the remote and the local
		1079	+	// observation paths on the same registry, since service.go's Subscribe/Close
		1080	+	// wiring feeds one shared demand counter for both.
		1081	+	func TestRegistryAppliesSubscriptionDemandFreshness(t *testing.T) {
		1082	+	const (
		1083	+	demandLocal = 250 * time.Millisecond
		1084	+	demandRemote = 500 * time.Millisecond
		1085	+	)
		1086	+	tests := []struct {
		1087	+	name string
		1088	+	local bool
		1089	+	demand bool
		1090	+	wantAfter time.Duration
		1091	+	}{
		1092	+	{name: "remote no demand keeps default", local: false, demand: false, wantAfter: defaultFreshFor},
		1093	+	{name: "remote demand uses faster cadence", local: false, demand: true, wantAfter: demandRemote},
		1094	+	{name: "local no demand keeps default", local: true, demand: false, wantAfter: defaultFreshFor},
		1095	+	{name: "local demand uses faster cadence", local: true, demand: true, wantAfter: demandLocal},
		1096	+	}
		1097	+
		1098	+	for _, tt := range tests {
		1099	+	t.Run(tt.name, func(t *testing.T) {
		1100	+	start := time.Unix(100, 0)
		1101	+	clock := newManualClock(start)
		1102	+	remoteProbe := newTestProbe(1)
		1103	+	localProbe := newScriptedLocalProbe(1)
		1104	+	registry, err := NewRegistryWithConfig(1, newTestStore(), remoteProbe, clock, nil, RegistryConfig{
		1105	+	Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: localProbe},
		1106	+	DemandFreshForLocal: demandLocal,
		1107	+	DemandFreshForRemote: demandRemote,
		1108	+	})
		1109	+	require.NoError(t, err)
		1110	+	registry.jitter = identityJitter
		1111	+	reg := registration(t, "example.test", 1)
		1112	+	require.NoError(t, registry.setHosts(hostRecords(reg)))
		1113	+	startRegistry(t, registry)
		1114	+
		1115	+	if tt.demand {
		1116	+	registry.SetDemand(true)
		1117	+	t.Cleanup(func() { registry.SetDemand(false) })
		1118	+	}
		1119	+
		1120	+	if tt.local {
		1121	+	call := receiveLocalCall(t, localProbe)
		1122	+	call.result <- localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
		1123	+	Identity: "local-daemon",
		1124	+	Incarnation: ports.BrokerDaemonIncarnation{9},
		1125	+	ProtocolVersion: protocol.Version,
		1126	+	Availability: domain.RemoteAvailabilityReachable,
		1127	+	}}
		1128	+	host := waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool {
		1129	+	return !o.Checking && o.Availability == domain.RemoteAvailabilityReachable
		1130	+	})
		1131	+	require.Equal(t, start.Add(tt.wantAfter), host.NextDue)
		1132	+	return
		1133	+	}
		1134	+	call := receiveCall(t, remoteProbe)
		1135	+	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
		1136	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
		1137	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
		1138	+	})
		1139	+	require.Equal(t, start.Add(tt.wantAfter), host.NextDue)
		1140	+	})
		1141	+	}
		1142	+	}
		1143	+
		1144	+	// TestRegistrySetDemandIsCountedNotBoolean pins that SetDemand is a balanced
		1145	+	// counter: two openings require two closings before the schedule reverts to
		1146	+	// the passive default, exactly as two concurrent client subscriptions would
		1147	+	// drive it through service.go.
		1148	+	func TestRegistrySetDemandIsCountedNotBoolean(t *testing.T) {
		1149	+	start := time.Unix(100, 0)
		1150	+	clock := newManualClock(start)
		1151	+	probe := newTestProbe(2)
		1152	+	registry := newTestRegistry(t, 1, newTestStore(), probe, clock)
		1153	+	registry.demandFreshForRemote = 500 * time.Millisecond
		1154	+	reg := registration(t, "example.test", 1)
		1155	+	require.NoError(t, registry.setHosts(hostRecords(reg)))
		1156	+	startRegistry(t, registry)
		1157	+
		1158	+	// Two subscriptions open; only one closes, so demand must stay active.
		1159	+	registry.SetDemand(true)
		1160	+	registry.SetDemand(true)
		1161	+	registry.SetDemand(false)
		1162	+
		1163	+	call := receiveCall(t, probe)
		1164	+	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
		1165	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
		1166	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
		1167	+	})
		1168	+	require.Equal(t, start.Add(500*time.Millisecond), host.NextDue, "one remaining demand unit must still use the faster cadence")
		1169	+
		1170	+	// The second, and last, subscription closes: demand drops back to zero.
		1171	+	registry.SetDemand(false)
		1172	+
		1173	+	clock.Advance(500 * time.Millisecond)
		1174	+	secondSuccess := start.Add(500 * time.Millisecond)
		1175	+	next := receiveCall(t, probe)
		1176	+	next.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
		1177	+	host = waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
		1178	+	return !host.Checking && host.LastSuccess.Equal(secondSuccess)
		1179	+	})
		1180	+	require.Equal(t, secondSuccess.Add(defaultFreshFor), host.NextDue, "no live demand must revert to the passive default cadence")
		1181	+	}
		1182	+
1075	func TestRegistryDemandOverridesFreshSchedule(t *testing.T) {	1183		func TestRegistryDemandOverridesFreshSchedule(t *testing.T) {
1076	start := time.Unix(100, 0)	1184		start := time.Unix(100, 0)
1077	clock := newManualClock(start)	1185		clock := newManualClock(start)

M internal/usecase/broker/service.go +23 -9


@@ -216,7 +216,11 @@ func (s *Service) Snapshot() ports.BrokerSnapshot { return s.registry.Snapshot()
216	216
217	217		// Subscribe delegates to the registry and tracks the returned subscription so
218	218		// Close releases every subscription this connection still owns. A closed
219		-	// connection refuses new subscriptions.
	219	+	// connection refuses new subscriptions. Opening a subscription also records
	220	+	// one unit of registry demand (Registry.SetDemand(true)): while a client
	221	+	// keeps watching the snapshot, the registry re-probes on the faster demand
	222	+	// cadence instead of the passive default. serviceSubscription.Close balances
	223	+	// it with SetDemand(false).
220	224		func (s *Service) Subscribe() (ports.BrokerSubscription, error) {
221	225		s.mu.Lock()
222	226		if s.closed {

@@ -225,6 +229,7 @@ func (s *Service) Subscribe() (ports.BrokerSubscription, error) {
225	229		}
226	230		sub := &serviceSubscription{service: s, inner: s.registry.Subscribe()}
227	231		s.subs[sub] = struct{}{}
	232	+	s.registry.SetDemand(true)
228	233		s.mu.Unlock()
229	234		return sub, nil
230	235		}

@@ -259,6 +264,18 @@ func (s *Service) OpenStream(ctx context.Context, request ports.BrokerOpenStream
259	264		release()
260	265		return nil, err
261	266		}
	267	+	// A local control or attachment stream is a natural signal that the local
	268	+	// daemon's own catalogue may be about to change (a session request is
	269	+	// being issued, or an attachment claim is being taken): mark a local
	270	+	// re-probe pending as soon as the stream opens, not only once it ends, so
	271	+	// a session created by this very attach appears in `ls --all`/other
	272	+	// pickers promptly, even while the client stays attached. An observation
	273	+	// stream never triggers this: it is the probe traffic itself, and
	274	+	// re-arming from it would starve the schedule instead of catching up to
	275	+	// it.
	276	+	if request.Local && (request.Purpose == ports.BrokerStreamControl \|\| request.Purpose == ports.BrokerStreamAttachment) {
	277	+	s.registry.RequestProbe("")
	278	+	}
262	279		// Detach the connection-scoped context link once the pool reports the
263	280		// stream terminal, so a long-lived connection never accumulates one link
264	281		// per opened stream. Nothing here ends the stream: setup completion and the

@@ -266,14 +283,10 @@ func (s *Service) OpenStream(ctx context.Context, request ports.BrokerOpenStream
266	283		go func() {
267	284		<-stream.Done()
268	285		release()
269		-	// A completed local control or attachment stream is the natural signal
270		-	// that the local daemon's own catalogue may have just changed (a session
271		-	// was created, killed, or its attachment state flipped): mark a local
272		-	// re-probe pending so the next dispatch observes it ahead of the
273		-	// freshness window instead of leaving `ls --all`/the picker stale for up
274		-	// to the retry/fresh interval. An observation stream never triggers this:
275		-	// it is the probe traffic itself, and re-arming from it would starve the
276		-	// schedule instead of catching up to it.
	286	+	// The stream ending is the same signal again (a session was created,
	287	+	// killed, or its attachment state flipped over the stream's lifetime):
	288	+	// request one more local re-probe so a change made just before detach
	289	+	// is not left stale for up to the retry/fresh interval.
277	290		if request.Local && (request.Purpose == ports.BrokerStreamControl \|\| request.Purpose == ports.BrokerStreamAttachment) {
278	291		s.registry.RequestProbe("")
279	292		}

@@ -450,6 +463,7 @@ func (s *serviceSubscription) Close() {
450	463		s.inner.Close()
451	464		s.service.mu.Lock()
452	465		delete(s.service.subs, s)
	466	+	s.service.registry.SetDemand(false)
453	467		s.service.mu.Unlock()
454	468		})
455	469		}

 // Subscribe delegates to the registry and tracks the returned subscription so
 // Close releases every subscription this connection still owns. A closed
-// connection refuses new subscriptions.
+// connection refuses new subscriptions. Opening a subscription also records
+// one unit of registry demand (Registry.SetDemand(true)): while a client
+// keeps watching the snapshot, the registry re-probes on the faster demand
+// cadence instead of the passive default. serviceSubscription.Close balances
+// it with SetDemand(false).
 func (s *Service) Subscribe() (ports.BrokerSubscription, error) {
 	s.mu.Lock()
 	if s.closed {
 	}
 	sub := &serviceSubscription{service: s, inner: s.registry.Subscribe()}
 	s.subs[sub] = struct{}{}
+	s.registry.SetDemand(true)
 	s.mu.Unlock()
 	return sub, nil
 }
 		release()
 		return nil, err
 	}
+	// A local control or attachment stream is a natural signal that the local
+	// daemon's own catalogue may be about to change (a session request is
+	// being issued, or an attachment claim is being taken): mark a local
+	// re-probe pending as soon as the stream opens, not only once it ends, so
+	// a session created by this very attach appears in `ls --all`/other
+	// pickers promptly, even while the client stays attached. An observation
+	// stream never triggers this: it is the probe traffic itself, and
+	// re-arming from it would starve the schedule instead of catching up to
+	// it.
+	if request.Local && (request.Purpose == ports.BrokerStreamControl || request.Purpose == ports.BrokerStreamAttachment) {
+		s.registry.RequestProbe("")
+	}
 	// Detach the connection-scoped context link once the pool reports the
 	// stream terminal, so a long-lived connection never accumulates one link
 	// per opened stream. Nothing here ends the stream: setup completion and the
 	go func() {
 		<-stream.Done()
 		release()
-		// A completed local control or attachment stream is the natural signal
+		// The stream ending is the same signal again (a session was created,
-		// that the local daemon's own catalogue may have just changed (a session
+		// killed, or its attachment state flipped over the stream's lifetime):
-		// was created, killed, or its attachment state flipped): mark a local
+		// request one more local re-probe so a change made just before detach
-		// re-probe pending so the next dispatch observes it ahead of the
+		// is not left stale for up to the retry/fresh interval.
-		// freshness window instead of leaving `ls --all`/the picker stale for up
-		// to the retry/fresh interval. An observation stream never triggers this:
-		// it is the probe traffic itself, and re-arming from it would starve the
-		// schedule instead of catching up to it.
 		if request.Local && (request.Purpose == ports.BrokerStreamControl || request.Purpose == ports.BrokerStreamAttachment) {
 			s.registry.RequestProbe("")
 		}
 		s.inner.Close()
 		s.service.mu.Lock()
 		delete(s.service.subs, s)
+		s.service.registry.SetDemand(false)
 		s.service.mu.Unlock()
 	})
 }

M internal/usecase/broker/service_test.go +94 -0


@@ -12,6 +12,7 @@ import (
12	12
13	13		"github.com/bnema/vev/internal/domain"
14	14		"github.com/bnema/vev/internal/ports"
	15	+	"github.com/bnema/vev/internal/protocol"
15	16		"github.com/stretchr/testify/require"
16	17		)
17	18

@@ -134,6 +135,12 @@ func registrySubscriptionCount(r *Registry) int {
134	135		return len(r.subs)
135	136		}
136	137
	138	+	func registryDemandCount(r *Registry) int {
	139	+	r.mu.Lock()
	140	+	defer r.mu.Unlock()
	141	+	return r.demand
	142	+	}
	143	+
137	144		func TestAuthorityAdmissionRollback(t *testing.T) {
138	145		authority, _, pool, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
139	146

@@ -298,6 +305,59 @@ func TestServiceOpenStreamScopeFencing(t *testing.T) {
298	305		}
299	306		}
300	307
	308	+	// TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose pins the "Extra"
	309	+	// fix from Phase C2: OpenStream marks a local re-probe pending as soon as a
	310	+	// local control/attachment stream opens, not only once it ends, so a session
	311	+	// created or claimed over this very stream can appear in `ls --all`/the
	312	+	// picker while the client stays attached, instead of waiting for detach. Both
	313	+	// transitions are proven against the registry's actual scheduling with no
	314	+	// clock advance at all, exactly like
	315	+	// TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly proves the
	316	+	// registry-side half of the mechanism.
	317	+	func TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose(t *testing.T) {
	318	+	reachable := func() localProbeAnswer {
	319	+	return localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
	320	+	Identity: "local-daemon", Incarnation: ports.BrokerDaemonIncarnation{9},
	321	+	ProtocolVersion: protocol.Version, Availability: domain.RemoteAvailabilityReachable,
	322	+	}}
	323	+	}
	324	+
	325	+	clock := newManualClock(time.Unix(100, 0))
	326	+	store := newTestStore()
	327	+	local := newScriptedLocalProbe(3)
	328	+	registry, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{
	329	+	Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: local},
	330	+	})
	331	+	require.NoError(t, err)
	332	+	registry.jitter = identityJitter
	333	+	authority, _, _, _ := composeTestAuthority(t, 1, registry, immediateConnector, clock)
	334	+
	335	+	service, err := authority.AdmitClient(context.Background())
	336	+	require.NoError(t, err)
	337	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
	338	+
	339	+	// The registry's first dispatch (Run start) already probes the local
	340	+	// daemon because it has never been observed: settle it so NextDue lands
	341	+	// far in the future (defaultFreshFor, no live demand yet), and only an
	342	+	// explicit reconcile can explain any further attempt.
	343	+	call := receiveLocalCall(t, local)
	344	+	call.result <- reachable()
	345	+	waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
	346	+
	347	+	// Opening a local control stream must itself mark a re-probe pending.
	348	+	stream, err := service.OpenStream(context.Background(), poolRequest(service.ConnectionID(), 1))
	349	+	require.NoError(t, err)
	350	+	opened := receiveLocalCall(t, local)
	351	+	opened.result <- reachable()
	352	+	waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
	353	+
	354	+	// Closing the stream marks a second re-probe pending, exactly as it did
	355	+	// for stream completion before this fix.
	356	+	require.NoError(t, stream.Close())
	357	+	closed := receiveLocalCall(t, local)
	358	+	closed.result <- reachable()
	359	+	}
	360	+
301	361		func TestServiceNextStreamIDAllocation(t *testing.T) {
302	362		authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
303	363		service, err := authority.AdmitClient(context.Background())

@@ -414,6 +474,40 @@ func TestServiceSubscriptionCleanup(t *testing.T) {
414	474		requireBrokerClosed(t, err)
415	475		}
416	476
	477	+	// TestServiceSubscribeDrivesRegistryDemand pins the C2 wiring: opening a
	478	+	// service-level subscription (an attached client keeps one open the whole
	479	+	// time it watches the snapshot) increments the registry's demand count, and
	480	+	// closing it decrements it back. This exercises the actual service.go call
	481	+	// sites (Subscribe / serviceSubscription.Close), not Registry.SetDemand
	482	+	// directly, and proves the count is balanced across concurrent subscriptions
	483	+	// and idempotent Close.
	484	+	func TestServiceSubscribeDrivesRegistryDemand(t *testing.T) {
	485	+	authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
	486	+	service, err := authority.AdmitClient(context.Background())
	487	+	require.NoError(t, err)
	488	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
	489	+
	490	+	require.Zero(t, registryDemandCount(registry))
	491	+
	492	+	sub, err := service.Subscribe()
	493	+	require.NoError(t, err)
	494	+	require.Equal(t, 1, registryDemandCount(registry))
	495	+
	496	+	other, err := service.Subscribe()
	497	+	require.NoError(t, err)
	498	+	require.Equal(t, 2, registryDemandCount(registry), "a second live subscription must add a second unit of demand")
	499	+
	500	+	sub.Close()
	501	+	require.Equal(t, 1, registryDemandCount(registry), "one closed subscription must leave the other's demand live")
	502	+
	503	+	other.Close()
	504	+	require.Zero(t, registryDemandCount(registry))
	505	+
	506	+	// Close is idempotent: a repeated close must never double-release demand.
	507	+	other.Close()
	508	+	require.Zero(t, registryDemandCount(registry))
	509	+	}
	510	+
417	511		func TestServiceSnapshotDelegates(t *testing.T) {
418	512		authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
419	513		service, err := authority.AdmitClient(context.Background())


@@ -12,6 +12,7 @@ import (
12		12
13	"github.com/bnema/vev/internal/domain"	13		"github.com/bnema/vev/internal/domain"
14	"github.com/bnema/vev/internal/ports"	14		"github.com/bnema/vev/internal/ports"
		15	+	"github.com/bnema/vev/internal/protocol"
15	"github.com/stretchr/testify/require"	16		"github.com/stretchr/testify/require"
16	)	17		)
17		18

@@ -134,6 +135,12 @@ func registrySubscriptionCount(r *Registry) int {
134	return len(r.subs)	135		return len(r.subs)
135	}	136		}
136		137
		138	+	func registryDemandCount(r *Registry) int {
		139	+	r.mu.Lock()
		140	+	defer r.mu.Unlock()
		141	+	return r.demand
		142	+	}
		143	+
137	func TestAuthorityAdmissionRollback(t *testing.T) {	144		func TestAuthorityAdmissionRollback(t *testing.T) {
138	authority, _, pool, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)	145		authority, _, pool, supervisor, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
139		146

@@ -298,6 +305,59 @@ func TestServiceOpenStreamScopeFencing(t *testing.T) {
298	}	305		}
299	}	306		}
300		307
		308	+	// TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose pins the "Extra"
		309	+	// fix from Phase C2: OpenStream marks a local re-probe pending as soon as a
		310	+	// local control/attachment stream opens, not only once it ends, so a session
		311	+	// created or claimed over this very stream can appear in `ls --all`/the
		312	+	// picker while the client stays attached, instead of waiting for detach. Both
		313	+	// transitions are proven against the registry's actual scheduling with no
		314	+	// clock advance at all, exactly like
		315	+	// TestRegistryLocalPendingReprobeHonorsExplicitReconcileOnly proves the
		316	+	// registry-side half of the mechanism.
		317	+	func TestServiceOpenStreamRequestsLocalReprobeOnOpenAndClose(t *testing.T) {
		318	+	reachable := func() localProbeAnswer {
		319	+	return localProbeAnswer{snapshot: ports.BrokerDaemonObservation{
		320	+	Identity: "local-daemon", Incarnation: ports.BrokerDaemonIncarnation{9},
		321	+	ProtocolVersion: protocol.Version, Availability: domain.RemoteAvailabilityReachable,
		322	+	}}
		323	+	}
		324	+
		325	+	clock := newManualClock(time.Unix(100, 0))
		326	+	store := newTestStore()
		327	+	local := newScriptedLocalProbe(3)
		328	+	registry, err := NewRegistryWithConfig(1, store, nil, clock, nil, RegistryConfig{
		329	+	Local: &LocalObservation{DisplayOrigin: "local", Policy: poolPolicy(), Probe: local},
		330	+	})
		331	+	require.NoError(t, err)
		332	+	registry.jitter = identityJitter
		333	+	authority, _, _, _ := composeTestAuthority(t, 1, registry, immediateConnector, clock)
		334	+
		335	+	service, err := authority.AdmitClient(context.Background())
		336	+	require.NoError(t, err)
		337	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
		338	+
		339	+	// The registry's first dispatch (Run start) already probes the local
		340	+	// daemon because it has never been observed: settle it so NextDue lands
		341	+	// far in the future (defaultFreshFor, no live demand yet), and only an
		342	+	// explicit reconcile can explain any further attempt.
		343	+	call := receiveLocalCall(t, local)
		344	+	call.result <- reachable()
		345	+	waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
		346	+
		347	+	// Opening a local control stream must itself mark a re-probe pending.
		348	+	stream, err := service.OpenStream(context.Background(), poolRequest(service.ConnectionID(), 1))
		349	+	require.NoError(t, err)
		350	+	opened := receiveLocalCall(t, local)
		351	+	opened.result <- reachable()
		352	+	waitLocal(t, registry, func(o ports.BrokerDaemonObservation) bool { return !o.Checking })
		353	+
		354	+	// Closing the stream marks a second re-probe pending, exactly as it did
		355	+	// for stream completion before this fix.
		356	+	require.NoError(t, stream.Close())
		357	+	closed := receiveLocalCall(t, local)
		358	+	closed.result <- reachable()
		359	+	}
		360	+
301	func TestServiceNextStreamIDAllocation(t *testing.T) {	361		func TestServiceNextStreamIDAllocation(t *testing.T) {
302	authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)	362		authority, _, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
303	service, err := authority.AdmitClient(context.Background())	363		service, err := authority.AdmitClient(context.Background())

@@ -414,6 +474,40 @@ func TestServiceSubscriptionCleanup(t *testing.T) {
414	requireBrokerClosed(t, err)	474		requireBrokerClosed(t, err)
415	}	475		}
416		476
		477	+	// TestServiceSubscribeDrivesRegistryDemand pins the C2 wiring: opening a
		478	+	// service-level subscription (an attached client keeps one open the whole
		479	+	// time it watches the snapshot) increments the registry's demand count, and
		480	+	// closing it decrements it back. This exercises the actual service.go call
		481	+	// sites (Subscribe / serviceSubscription.Close), not Registry.SetDemand
		482	+	// directly, and proves the count is balanced across concurrent subscriptions
		483	+	// and idempotent Close.
		484	+	func TestServiceSubscribeDrivesRegistryDemand(t *testing.T) {
		485	+	authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
		486	+	service, err := authority.AdmitClient(context.Background())
		487	+	require.NoError(t, err)
		488	+	t.Cleanup(func() { require.NoError(t, service.Close()) })
		489	+
		490	+	require.Zero(t, registryDemandCount(registry))
		491	+
		492	+	sub, err := service.Subscribe()
		493	+	require.NoError(t, err)
		494	+	require.Equal(t, 1, registryDemandCount(registry))
		495	+
		496	+	other, err := service.Subscribe()
		497	+	require.NoError(t, err)
		498	+	require.Equal(t, 2, registryDemandCount(registry), "a second live subscription must add a second unit of demand")
		499	+
		500	+	sub.Close()
		501	+	require.Equal(t, 1, registryDemandCount(registry), "one closed subscription must leave the other's demand live")
		502	+
		503	+	other.Close()
		504	+	require.Zero(t, registryDemandCount(registry))
		505	+
		506	+	// Close is idempotent: a repeated close must never double-release demand.
		507	+	other.Close()
		508	+	require.Zero(t, registryDemandCount(registry))
		509	+	}
		510	+
417	func TestServiceSnapshotDelegates(t *testing.T) {	511		func TestServiceSnapshotDelegates(t *testing.T) {
418	authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)	512		authority, registry, _, _, _, _ := newTestAuthority(t, 1, nil, immediateConnector)
419	service, err := authority.AdmitClient(context.Background())	513		service, err := authority.AdmitClient(context.Background())