25ad966 fix(client,app): trust remote absence only from a fresh observation · vev

M internal/app/broker_client.go +7 -8


@@ -453,10 +453,11 @@ func remoteNamedCreateResolver(endpoint, name string) client.InitialNavigationRe
453	453
454	454		// remoteAttachOrCreateResolver resolves one remote session name to an exact
455	455		// attach when the endpoint's committed observation carries it, and to a named
456		-	// creation fenced to the same registration once the observed inventory proves
457		-	// it absent. Before the host's first inventory it asks to be resolved again. An endpoint the broker
458		-	// does not carry is refused rather than dialed.
459		-	func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigationResolver {
	456	+	// creation fenced to the same registration once an inventory observed at or
	457	+	// after requestedAt proves it absent. Before such an observation it asks the
	458	+	// supervisor to request one and resolve again. An endpoint the broker does not
	459	+	// carry is refused rather than dialed.
	460	+	func remoteAttachOrCreateResolver(endpoint, name string, requestedAt time.Time) client.InitialNavigationResolver {
460	461		attach := remoteExactAttachResolver(endpoint, name)
461	462		create := remoteNamedCreateResolver(endpoint, name)
462	463		return func(snapshot ports.BrokerSnapshot) (client.InitialNavigation, error) {

@@ -467,10 +468,8 @@ func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigatio
467	468		if _, exists := brokerObservationExactTarget(observation, name); exists {
468	469		return attach(snapshot)
469	470		}
470		-	if !observation.InventoryKnown {
471		-	// Absence is only known once the host's inventory was observed;
472		-	// creating before that would collide with a live remote name.
473		-	return client.InitialNavigation{}, client.ErrInitialNavigationNotObserved
	471	+	if !observation.InventoryKnown \|\| observation.LastSuccess.Before(requestedAt) {
	472	+	return client.InitialNavigation{}, client.InitialNavigationNotObserved{Endpoint: endpoint}
474	473		}
475	474		return create(snapshot)
476	475		}

 // remoteAttachOrCreateResolver resolves one remote session name to an exact
 // attach when the endpoint's committed observation carries it, and to a named
-// creation fenced to the same registration once the observed inventory proves
+// creation fenced to the same registration once an inventory observed at or
-// it absent. Before the host's first inventory it asks to be resolved again. An endpoint the broker
+// after requestedAt proves it absent. Before such an observation it asks the
-// does not carry is refused rather than dialed.
+// supervisor to request one and resolve again. An endpoint the broker does not
-func remoteAttachOrCreateResolver(endpoint, name string) client.InitialNavigationResolver {
+// carry is refused rather than dialed.
+func remoteAttachOrCreateResolver(endpoint, name string, requestedAt time.Time) client.InitialNavigationResolver {
 	attach := remoteExactAttachResolver(endpoint, name)
 	create := remoteNamedCreateResolver(endpoint, name)
 	return func(snapshot ports.BrokerSnapshot) (client.InitialNavigation, error) {
 		if _, exists := brokerObservationExactTarget(observation, name); exists {
 			return attach(snapshot)
 		}
-		if !observation.InventoryKnown {
+		if !observation.InventoryKnown || observation.LastSuccess.Before(requestedAt) {
-			// Absence is only known once the host's inventory was observed;
+			return client.InitialNavigation{}, client.InitialNavigationNotObserved{Endpoint: endpoint}
-			// creating before that would collide with a live remote name.
-			return client.InitialNavigation{}, client.ErrInitialNavigationNotObserved
 		}
 		return create(snapshot)
 	}

M internal/app/ui_driver.go +8 -6


@@ -12,6 +12,7 @@ import (
12	12		"strings"
13	13		"sync"
14	14		"syscall"
	15	+	"time"
15	16
16	17		"github.com/bnema/vev/internal/adapters/clock"
17	18		"github.com/bnema/vev/internal/adapters/ipc"

@@ -194,22 +195,23 @@ func runUIDriver(ctx context.Context, options uiDriverOptions) error {
194	195		// name creates, a bare local invocation creates ephemerally, and a remote
195	196		// target is only ever the registration and lifecycle a broker publication
196	197		// carries.
197		-	func uiDriverNavigation(options uiDriverOptions) (client.InitialNavigation, client.InitialNavigationResolver, error) {
	198	+	func uiDriverNavigation(options uiDriverOptions, requestedAt time.Time) (client.InitialNavigation, client.InitialNavigationResolver, error) {
198	199		if options.picker {
199	200		return client.InitialNavigation{}, nil, nil
200	201		}
201	202		if options.remote != "" && options.session != "" {
202	203		// `--remote HOST --session NAME` is attach-or-create: an existing
203	204		// remote session is attached by its exact lifecycle, an absent one is
204		-	// created. Creating unconditionally would collide with the remote
205		-	// daemon's live name and be refused.
	205	+	// created. Absence is only trusted from an observation newer than
	206	+	// this invocation; creating on an older inventory would collide with
	207	+	// a live remote name and be refused.
206	208		if err := domain.ValidateRemoteHostTarget(options.remote); err != nil {
207	209		return client.InitialNavigation{}, nil, err
208	210		}
209	211		if err := domain.ValidateSessionName(options.session); err != nil {
210	212		return client.InitialNavigation{}, nil, err
211	213		}
212		-	return client.InitialNavigation{}, remoteAttachOrCreateResolver(options.remote, options.session), nil
	214	+	return client.InitialNavigation{}, remoteAttachOrCreateResolver(options.remote, options.session, requestedAt), nil
213	215		}
214	216		intent := protocol.IntentEphemeral
215	217		if options.session != "" {

@@ -223,7 +225,8 @@ func uiDriverNavigation(options uiDriverOptions) (client.InitialNavigation, clie
223	225		// context, and the one-shot initial navigation the CLI intent names. It never
224	226		// provisions, reconfigures, or destroys an endpoint, and it owns no daemon.
225	227		func runHeadlessUIDriver(ctx context.Context, options uiDriverOptions) error {
226		-	navigation, resolver, err := uiDriverNavigation(options)
	228	+	clk := clock.New()
	229	+	navigation, resolver, err := uiDriverNavigation(options, clk.Now())
227	230		if err != nil {
228	231		return err
229	232		}

@@ -232,7 +235,6 @@ func runHeadlessUIDriver(ctx context.Context, options uiDriverOptions) error {
232	235		return err
233	236		}
234	237		defer func() { _ = logCloser.Close() }()
235		-	clk := clock.New()
236	238		terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: options.cols, Rows: options.rows}}, "")
237	239		if err != nil {
238	240		return fmt.Errorf("vev: create headless terminal: %w", err)

 	"strings"
 	"sync"
 	"syscall"
+	"time"
 	"github.com/bnema/vev/internal/adapters/clock"
 	"github.com/bnema/vev/internal/adapters/ipc"
 // name creates, a bare local invocation creates ephemerally, and a remote
 // target is only ever the registration and lifecycle a broker publication
 // carries.
-func uiDriverNavigation(options uiDriverOptions) (client.InitialNavigation, client.InitialNavigationResolver, error) {
+func uiDriverNavigation(options uiDriverOptions, requestedAt time.Time) (client.InitialNavigation, client.InitialNavigationResolver, error) {
 	if options.picker {
 		return client.InitialNavigation{}, nil, nil
 	}
 	if options.remote != "" && options.session != "" {
 		// `--remote HOST --session NAME` is attach-or-create: an existing
 		// remote session is attached by its exact lifecycle, an absent one is
-		// created. Creating unconditionally would collide with the remote
+		// created. Absence is only trusted from an observation newer than
-		// daemon's live name and be refused.
+		// this invocation; creating on an older inventory would collide with
+		// a live remote name and be refused.
 		if err := domain.ValidateRemoteHostTarget(options.remote); err != nil {
 			return client.InitialNavigation{}, nil, err
 		}
 		if err := domain.ValidateSessionName(options.session); err != nil {
 			return client.InitialNavigation{}, nil, err
 		}
-		return client.InitialNavigation{}, remoteAttachOrCreateResolver(options.remote, options.session), nil
+		return client.InitialNavigation{}, remoteAttachOrCreateResolver(options.remote, options.session, requestedAt), nil
 	}
 	intent := protocol.IntentEphemeral
 	if options.session != "" {
 // context, and the one-shot initial navigation the CLI intent names. It never
 // provisions, reconfigures, or destroys an endpoint, and it owns no daemon.
 func runHeadlessUIDriver(ctx context.Context, options uiDriverOptions) error {
-	navigation, resolver, err := uiDriverNavigation(options)
+	clk := clock.New()
+	navigation, resolver, err := uiDriverNavigation(options, clk.Now())
 	if err != nil {
 		return err
 	}
 		return err
 	}
 	defer func() { _ = logCloser.Close() }()
-	clk := clock.New()
 	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: options.cols, Rows: options.rows}}, "")
 	if err != nil {
 		return fmt.Errorf("vev: create headless terminal: %w", err)

M internal/app/ui_driver_acceptance_test.go +16 -6


@@ -193,25 +193,26 @@ func TestUIDriverDetachReturnsToPickerWithoutRecreating(t *testing.T) {
193	193		// committed publication (never an invented identity). The driver keeps no
194	194		// navigation vocabulary of its own.
195	195		func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
196		-	picker, resolver, err := uiDriverNavigation(uiDriverOptions{picker: true})
	196	+	picker, resolver, err := uiDriverNavigation(uiDriverOptions{picker: true}, time.Time{})
197	197		require.NoError(t, err)
198	198		require.Nil(t, resolver)
199	199		require.Equal(t, client.InitialNavigation{}, picker)
200	200		require.NoError(t, picker.Validate(), "the picker intent is the safe zero value")
201	201
202		-	local, resolver, err := uiDriverNavigation(uiDriverOptions{})
	202	+	local, resolver, err := uiDriverNavigation(uiDriverOptions{}, time.Time{})
203	203		require.NoError(t, err)
204	204		require.Nil(t, resolver)
205	205		require.Equal(t, client.InitialNavigationCreateEphemeral, local.Kind)
206	206		require.True(t, local.Destination.Local)
207	207
208		-	named, resolver, err := uiDriverNavigation(uiDriverOptions{session: "work"})
	208	+	named, resolver, err := uiDriverNavigation(uiDriverOptions{session: "work"}, time.Time{})
209	209		require.NoError(t, err)
210	210		require.Nil(t, resolver)
211	211		require.Equal(t, client.InitialNavigationCreateNamed, named.Kind)
212	212		require.Equal(t, "work", named.Name)
213	213		require.True(t, named.Destination.Local)
214	214
	215	+	requestedAt := time.Unix(2000, 0)
215	216		unobserved := ports.BrokerSnapshot{
216	217		Epoch: terminalCompositionEpoch,
217	218		Revision: 1,

@@ -221,7 +222,7 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
221	222		Registration: terminalCompositionRegistration("user@example.com"),
222	223		}},
223	224		}
224		-	remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot {
	225	+	observedAt := func(at time.Time, sessions ...string) ports.BrokerSnapshot {
225	226		return ports.BrokerSnapshot{
226	227		Epoch: terminalCompositionEpoch,
227	228		Revision: 1,

@@ -230,10 +231,14 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
230	231		DisplayOrigin: "user@example.com",
231	232		Registration: terminalCompositionRegistration("user@example.com"),
232	233		InventoryKnown: true,
	234	+	LastSuccess: at,
233	235		Sessions: terminalCompositionSessions(sessions),
234	236		}},
235	237		}
236	238		}
	239	+	remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot {
	240	+	return observedAt(requestedAt.Add(time.Second), sessions...)
	241	+	}
237	242		tests := []struct {
238	243		name string
239	244		options uiDriverOptions

@@ -243,6 +248,8 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
243	248		wantErr error
244	249		}{
245	250		{name: "remote named before first inventory waits", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: unobserved, wantErr: client.ErrInitialNavigationNotObserved},
	251	+	{name: "absence in an inventory older than the invocation waits", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: observedAt(requestedAt.Add(-time.Second), "work"), wantErr: client.ErrInitialNavigationNotObserved},
	252	+	{name: "presence in an older inventory attaches", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: observedAt(requestedAt.Add(-time.Second), "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"},
246	253		{name: "remote ephemeral", options: uiDriverOptions{remote: "user@example.com"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateEphemeral},
247	254		{name: "remote named existing attaches exact lifecycle", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: remoteSnapshot("other", "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"},
248	255		{name: "remote named missing creates", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateNamed, wantName: "fresh"},

@@ -250,7 +257,7 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
250	257		}
251	258		for _, tt := range tests {
252	259		t.Run(tt.name, func(t *testing.T) {
253		-	navigation, resolver, err := uiDriverNavigation(tt.options)
	260	+	navigation, resolver, err := uiDriverNavigation(tt.options, requestedAt)
254	261		require.NoError(t, err)
255	262		require.Equal(t, client.InitialNavigation{}, navigation, "a remote target is never decided before the connection")
256	263		require.NotNil(t, resolver, "a remote target resolves against the committed publication")

@@ -258,6 +265,9 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
258	265		resolved, resolveErr := resolver(tt.snapshot)
259	266		if tt.wantErr != nil {
260	267		require.ErrorIs(t, resolveErr, tt.wantErr)
	268	+	var pending client.InitialNavigationNotObserved
	269	+	require.ErrorAs(t, resolveErr, &pending)
	270	+	require.Equal(t, "user@example.com", pending.Endpoint, "the wait names the endpoint to reconcile")
261	271		return
262	272		}
263	273		require.NoError(t, resolveErr)

@@ -278,7 +288,7 @@ func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
278	288
279	289		// A target the committed publication does not carry is refused, never
280	290		// created implicitly.
281		-	_, resolver, err = uiDriverNavigation(uiDriverOptions{remote: "user@elsewhere.test"})
	291	+	_, resolver, err = uiDriverNavigation(uiDriverOptions{remote: "user@elsewhere.test"}, requestedAt)
282	292		require.NoError(t, err)
283	293		_, err = resolver(ports.BrokerSnapshot{Epoch: terminalCompositionEpoch, Revision: 1})
284	294		require.Error(t, err)

 // committed publication (never an invented identity). The driver keeps no
 // navigation vocabulary of its own.
 func TestUIDriverNavigationReusesTheSharedTerminalTranslation(t *testing.T) {
-	picker, resolver, err := uiDriverNavigation(uiDriverOptions{picker: true})
+	picker, resolver, err := uiDriverNavigation(uiDriverOptions{picker: true}, time.Time{})
 	require.NoError(t, err)
 	require.Nil(t, resolver)
 	require.Equal(t, client.InitialNavigation{}, picker)
 	require.NoError(t, picker.Validate(), "the picker intent is the safe zero value")
-	local, resolver, err := uiDriverNavigation(uiDriverOptions{})
+	local, resolver, err := uiDriverNavigation(uiDriverOptions{}, time.Time{})
 	require.NoError(t, err)
 	require.Nil(t, resolver)
 	require.Equal(t, client.InitialNavigationCreateEphemeral, local.Kind)
 	require.True(t, local.Destination.Local)
-	named, resolver, err := uiDriverNavigation(uiDriverOptions{session: "work"})
+	named, resolver, err := uiDriverNavigation(uiDriverOptions{session: "work"}, time.Time{})
 	require.NoError(t, err)
 	require.Nil(t, resolver)
 	require.Equal(t, client.InitialNavigationCreateNamed, named.Kind)
 	require.Equal(t, "work", named.Name)
 	require.True(t, named.Destination.Local)
+	requestedAt := time.Unix(2000, 0)
 	unobserved := ports.BrokerSnapshot{
 		Epoch:    terminalCompositionEpoch,
 		Revision: 1,
 			Registration:  terminalCompositionRegistration("user@example.com"),
 		}},
 	}
-	remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot {
+	observedAt := func(at time.Time, sessions ...string) ports.BrokerSnapshot {
 		return ports.BrokerSnapshot{
 			Epoch:    terminalCompositionEpoch,
 			Revision: 1,
 				DisplayOrigin:  "user@example.com",
 				Registration:   terminalCompositionRegistration("user@example.com"),
 				InventoryKnown: true,
+				LastSuccess:    at,
 				Sessions:       terminalCompositionSessions(sessions),
 			}},
 		}
 	}
+	remoteSnapshot := func(sessions ...string) ports.BrokerSnapshot {
+		return observedAt(requestedAt.Add(time.Second), sessions...)
+	}
 	tests := []struct {
 		name     string
 		options  uiDriverOptions
 		wantErr  error
 	}{
 		{name: "remote named before first inventory waits", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: unobserved, wantErr: client.ErrInitialNavigationNotObserved},
+		{name: "absence in an inventory older than the invocation waits", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: observedAt(requestedAt.Add(-time.Second), "work"), wantErr: client.ErrInitialNavigationNotObserved},
+		{name: "presence in an older inventory attaches", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: observedAt(requestedAt.Add(-time.Second), "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"},
 		{name: "remote ephemeral", options: uiDriverOptions{remote: "user@example.com"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateEphemeral},
 		{name: "remote named existing attaches exact lifecycle", options: uiDriverOptions{remote: "user@example.com", session: "work"}, snapshot: remoteSnapshot("other", "work"), wantKind: client.InitialNavigationAttachExact, wantName: "work"},
 		{name: "remote named missing creates", options: uiDriverOptions{remote: "user@example.com", session: "fresh"}, snapshot: remoteSnapshot("work"), wantKind: client.InitialNavigationCreateNamed, wantName: "fresh"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			navigation, resolver, err := uiDriverNavigation(tt.options)
+			navigation, resolver, err := uiDriverNavigation(tt.options, requestedAt)
 			require.NoError(t, err)
 			require.Equal(t, client.InitialNavigation{}, navigation, "a remote target is never decided before the connection")
 			require.NotNil(t, resolver, "a remote target resolves against the committed publication")
 			resolved, resolveErr := resolver(tt.snapshot)
 			if tt.wantErr != nil {
 				require.ErrorIs(t, resolveErr, tt.wantErr)
+				var pending client.InitialNavigationNotObserved
+				require.ErrorAs(t, resolveErr, &pending)
+				require.Equal(t, "user@example.com", pending.Endpoint, "the wait names the endpoint to reconcile")
 				return
 			}
 			require.NoError(t, resolveErr)
 	// A target the committed publication does not carry is refused, never
 	// created implicitly.
-	_, resolver, err = uiDriverNavigation(uiDriverOptions{remote: "user@elsewhere.test"})
+	_, resolver, err = uiDriverNavigation(uiDriverOptions{remote: "user@elsewhere.test"}, requestedAt)
 	require.NoError(t, err)
 	_, err = resolver(ports.BrokerSnapshot{Epoch: terminalCompositionEpoch, Revision: 1})
 	require.Error(t, err)

M internal/usecase/client/initial_navigation_test.go +13 -3


@@ -506,6 +506,7 @@ func TestInitialNavigationResolverFailureIsSelectionUnavailable(t *testing.T) {
506	506		// on the next committed publication, and only the bounded observation budget
507	507		// turns the wait into a selection-unavailable refusal.
508	508		func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
	509	+	late := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(5), SessionName: "late"}
509	510		tests := []struct {
510	511		name string
511	512		// observedAt is the first revision the resolver can decide on; zero

@@ -540,9 +541,11 @@ func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
540	541		ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) {
541	542		calls.Add(1)
542	543		if tt.observedAt == 0 \|\| snapshot.Revision < tt.observedAt {
543		-	return InitialNavigation{}, ErrInitialNavigationNotObserved
	544	+	return InitialNavigation{}, InitialNavigationNotObserved{Endpoint: "user@host"}
544	545		}
545		-	return InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}, nil
	546	+	// The target exists only in the later publication: the stream
	547	+	// request must resolve against the snapshot that decided.
	548	+	return InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: snapshot.Epoch, Destination: ports.BrokerEndpointFence{Local: true}, Target: late}, nil
546	549		},
547	550		})
548	551		done := make(chan error, 1)

@@ -562,10 +565,17 @@ func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
562	565		return false
563	566		}, 5*time.Second, time.Millisecond, "the wait is bounded by the observation budget")
564	567		require.Zero(t, opened.Load(), "an unobserved target never dials")
	568	+	service.mu.Lock()
	569	+	reconciles := append([]string(nil), service.reconciles...)
	570	+	service.mu.Unlock()
	571	+	require.Equal(t, []string{"user@host"}, reconciles, "the wait asks the broker for one fresh observation")
565	572
566	573		if tt.wantOpen {
567		-	service.publishSnapshot(localDaemonSnapshot(3, 2))
	574	+	service.publishSnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
	575	+	pickerTestLocalObservation(time.Unix(1000, 0), pickerTestSession("late", 5, catalogue_Up)),
	576	+	}})
568	577		require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond)
	578	+	require.Equal(t, late, service.openedRequests()[0].Target)
569	579		require.Equal(t, int64(2), calls.Load(), "the armed intent resolves once per publication")
570	580		} else {
571	581		budget.fire()

 // on the next committed publication, and only the bounded observation budget
 // turns the wait into a selection-unavailable refusal.
 func TestInitialNavigationWaitsForTargetObservation(t *testing.T) {
+	late := protocol.ExactSessionTarget{LifecycleID: pickerTestLifecycle(5), SessionName: "late"}
 	tests := []struct {
 		name string
 		// observedAt is the first revision the resolver can decide on; zero
 				ResolveInitialNavigation: func(snapshot ports.BrokerSnapshot) (InitialNavigation, error) {
 					calls.Add(1)
 					if tt.observedAt == 0 || snapshot.Revision < tt.observedAt {
-						return InitialNavigation{}, ErrInitialNavigationNotObserved
+						return InitialNavigation{}, InitialNavigationNotObserved{Endpoint: "user@host"}
 					}
-					return InitialNavigation{Kind: InitialNavigationCreateEphemeral, Destination: ports.BrokerEndpointFence{Local: true}}, nil
+					// The target exists only in the later publication: the stream
+					// request must resolve against the snapshot that decided.
+					return InitialNavigation{Kind: InitialNavigationAttachExact, Epoch: snapshot.Epoch, Destination: ports.BrokerEndpointFence{Local: true}, Target: late}, nil
 				},
 			})
 			done := make(chan error, 1)
 				return false
 			}, 5*time.Second, time.Millisecond, "the wait is bounded by the observation budget")
 			require.Zero(t, opened.Load(), "an unobserved target never dials")
+			service.mu.Lock()
+			reconciles := append([]string(nil), service.reconciles...)
+			service.mu.Unlock()
+			require.Equal(t, []string{"user@host"}, reconciles, "the wait asks the broker for one fresh observation")
 			if tt.wantOpen {
-				service.publishSnapshot(localDaemonSnapshot(3, 2))
+				service.publishSnapshot(ports.BrokerSnapshot{Epoch: 3, Revision: 2, Daemons: []ports.BrokerDaemonObservation{
+					pickerTestLocalObservation(time.Unix(1000, 0), pickerTestSession("late", 5, catalogue_Up)),
+				}})
 				require.Eventually(t, func() bool { return opened.Load() == 1 }, 5*time.Second, time.Millisecond)
+				require.Equal(t, late, service.openedRequests()[0].Target)
 				require.Equal(t, int64(2), calls.Load(), "the armed intent resolves once per publication")
 			} else {
 				budget.fire()

M internal/usecase/client/supervisor_attach.go +45 -13


@@ -393,38 +393,67 @@ func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (Initi
393	393		// later publications, bounded by initialNavigationObservationBudget.
394	394		var ErrInitialNavigationNotObserved = errors.New("vev: initial navigation target is not observed yet")
395	395
	396	+	// InitialNavigationNotObserved is the resolver's ErrInitialNavigationNotObserved
	397	+	// naming the configured endpoint whose fresh observation it needs; the
	398	+	// supervisor asks the broker to reconcile that endpoint once per wait.
	399	+	type InitialNavigationNotObserved struct {
	400	+	Endpoint string
	401	+	}
	402	+
	403	+	func (e InitialNavigationNotObserved) Error() string {
	404	+	return ErrInitialNavigationNotObserved.Error() + ": " + e.Endpoint
	405	+	}
	406	+
	407	+	func (e InitialNavigationNotObserved) Is(target error) bool {
	408	+	return target == ErrInitialNavigationNotObserved
	409	+	}
	410	+
396	411		// initialNavigationObservationBudget bounds how long an initial navigation
397	412		// waits for its target's first observation; it stays inside the 15-second
398	413		// handshake budget.
399	414		const initialNavigationObservationBudget = 10 * time.Second
400	415
	416	+	// initialNavigationTake is one consumption attempt of the armed initial
	417	+	// navigation against the committed publication that decided it.
	418	+	type initialNavigationTake struct {
	419	+	snapshot ports.BrokerSnapshot
	420	+	navigation InitialNavigation
	421	+	consumed bool
	422	+	err error
	423	+	terminated bool
	424	+	termErr error
	425	+	}
	426	+
401	427		// awaitInitialNavigationObservation re-resolves the armed initial navigation
402	428		// on each committed publication until the resolver can decide, the budget
403		-	// expires (a bounded refusal), or the run ends.
404		-	func (s Supervisor) awaitInitialNavigationObservation(ctx context.Context, input terminalInputLifetime, service ports.BrokerService) (InitialNavigation, bool, error, bool, error) {
	429	+	// expires (a bounded refusal), or the run ends. The returned snapshot is the
	430	+	// publication the decision was made on, so the stream request resolves against
	431	+	// the same authority.
	432	+	func (s Supervisor) awaitInitialNavigationObservation(ctx context.Context, input terminalInputLifetime, service ports.BrokerService) initialNavigationTake {
405	433		timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget)
406	434		defer timer.Stop()
407	435		for {
408	436		select {
409	437		case <-s.brokerChanged():
410		-	navigation, consumed, err := s.takeInitialNavigation(service.Snapshot())
	438	+	snapshot := service.Snapshot()
	439	+	navigation, consumed, err := s.takeInitialNavigation(snapshot)
411	440		if errors.Is(err, ErrInitialNavigationNotObserved) {
412	441		continue
413	442		}
414		-	return navigation, consumed, err, false, nil
	443	+	return initialNavigationTake{snapshot: snapshot, navigation: navigation, consumed: consumed, err: err}
415	444		case <-timer.C():
416	445		s.mu.Lock()
417	446		s.navigationConsumed = true
418	447		s.mu.Unlock()
419		-	return InitialNavigation{}, true, ErrInitialNavigationNotObserved, false, nil
	448	+	return initialNavigationTake{consumed: true, err: ErrInitialNavigationNotObserved}
420	449		case <-service.Done():
421	450		// The ready loop observes the loss; the intent stays armed for the
422	451		// next adopted connection.
423		-	return InitialNavigation{}, false, nil, false, nil
	452	+	return initialNavigationTake{}
424	453		case <-ctx.Done():
425		-	return InitialNavigation{}, false, nil, true, ctx.Err()
	454	+	return initialNavigationTake{terminated: true, termErr: ctx.Err()}
426	455		case err := <-input.EOF():
427		-	return InitialNavigation{}, false, nil, true, terminalReadCause(err)
	456	+	return initialNavigationTake{terminated: true, termErr: terminalReadCause(err)}
428	457		}
429	458		}
430	459		}

@@ -442,12 +471,15 @@ func (s Supervisor) runInitialNavigation(ctx context.Context, input terminalIn
442	471		snapshot := service.Snapshot()
443	472		navigation, consumed, err := s.takeInitialNavigation(snapshot)
444	473		if errors.Is(err, ErrInitialNavigationNotObserved) {
445		-	var terminated bool
446		-	var termErr error
447		-	navigation, consumed, err, terminated, termErr = s.awaitInitialNavigationObservation(ctx, input, service)
448		-	if terminated {
449		-	return true, termErr
	474	+	var pending InitialNavigationNotObserved
	475	+	if errors.As(err, &pending) && pending.Endpoint != "" {
	476	+	service.RequestReconcile(pending.Endpoint)
	477	+	}
	478	+	take := s.awaitInitialNavigationObservation(ctx, input, service)
	479	+	if take.terminated {
	480	+	return true, take.termErr
450	481		}
	482	+	snapshot, navigation, consumed, err = take.snapshot, take.navigation, take.consumed, take.err
451	483		}
452	484		if !consumed {
453	485		return false, nil

 // later publications, bounded by initialNavigationObservationBudget.
 var ErrInitialNavigationNotObserved = errors.New("vev: initial navigation target is not observed yet")
+// InitialNavigationNotObserved is the resolver's ErrInitialNavigationNotObserved
+// naming the configured endpoint whose fresh observation it needs; the
+// supervisor asks the broker to reconcile that endpoint once per wait.
+type InitialNavigationNotObserved struct {
+	Endpoint string
+}
+func (e InitialNavigationNotObserved) Error() string {
+	return ErrInitialNavigationNotObserved.Error() + ": " + e.Endpoint
+}
+func (e InitialNavigationNotObserved) Is(target error) bool {
+	return target == ErrInitialNavigationNotObserved
+}
 // initialNavigationObservationBudget bounds how long an initial navigation
 // waits for its target's first observation; it stays inside the 15-second
 // handshake budget.
 const initialNavigationObservationBudget = 10 * time.Second
+// initialNavigationTake is one consumption attempt of the armed initial
+// navigation against the committed publication that decided it.
+type initialNavigationTake struct {
+	snapshot   ports.BrokerSnapshot
+	navigation InitialNavigation
+	consumed   bool
+	err        error
+	terminated bool
+	termErr    error
+}
 // awaitInitialNavigationObservation re-resolves the armed initial navigation
 // on each committed publication until the resolver can decide, the budget
-// expires (a bounded refusal), or the run ends.
+// expires (a bounded refusal), or the run ends. The returned snapshot is the
-func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService) (InitialNavigation, bool, error, bool, error) {
+// publication the decision was made on, so the stream request resolves against
+// the same authority.
+func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerService) initialNavigationTake {
 	timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget)
 	defer timer.Stop()
 	for {
 		select {
 		case <-s.brokerChanged():
-			navigation, consumed, err := s.takeInitialNavigation(service.Snapshot())
+			snapshot := service.Snapshot()
+			navigation, consumed, err := s.takeInitialNavigation(snapshot)
 			if errors.Is(err, ErrInitialNavigationNotObserved) {
 				continue
 			}
-			return navigation, consumed, err, false, nil
+			return initialNavigationTake{snapshot: snapshot, navigation: navigation, consumed: consumed, err: err}
 		case <-timer.C():
 			s.mu.Lock()
 			s.navigationConsumed = true
 			s.mu.Unlock()
-			return InitialNavigation{}, true, ErrInitialNavigationNotObserved, false, nil
+			return initialNavigationTake{consumed: true, err: ErrInitialNavigationNotObserved}
 		case <-service.Done():
 			// The ready loop observes the loss; the intent stays armed for the
 			// next adopted connection.
-			return InitialNavigation{}, false, nil, false, nil
+			return initialNavigationTake{}
 		case <-ctx.Done():
-			return InitialNavigation{}, false, nil, true, ctx.Err()
+			return initialNavigationTake{terminated: true, termErr: ctx.Err()}
 		case err := <-input.EOF():
-			return InitialNavigation{}, false, nil, true, terminalReadCause(err)
+			return initialNavigationTake{terminated: true, termErr: terminalReadCause(err)}
 		}
 	}
 }
 	snapshot := service.Snapshot()
 	navigation, consumed, err := s.takeInitialNavigation(snapshot)
 	if errors.Is(err, ErrInitialNavigationNotObserved) {
-		var terminated bool
+		var pending InitialNavigationNotObserved
-		var termErr error
+		if errors.As(err, &pending) && pending.Endpoint != "" {
-		navigation, consumed, err, terminated, termErr = s.awaitInitialNavigationObservation(ctx, input, service)
+			service.RequestReconcile(pending.Endpoint)
-		if terminated {
+		}
-			return true, termErr
+		take := s.awaitInitialNavigationObservation(ctx, input, service)
+		if take.terminated {
+			return true, take.termErr
 		}
+		snapshot, navigation, consumed, err = take.snapshot, take.navigation, take.consumed, take.err
 	}
 	if !consumed {
 		return false, nil