5c34fff feat(broker): distinguish reachable hosts without a daemon · vev

M internal/adapters/brokerwire/codec_test.go +12 -0


@@ -561,6 +561,18 @@ func TestBrokerSnapshotDaemonByteForByte(t *testing.T) {
561	561		message SnapshotPart
562	562		}{
563	563		{"observed remote daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}},
	564	+	{"reachable remote without daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: func() ports.BrokerDaemonObservation {
	565	+	daemon := testDaemonObservation()
	566	+	daemon.Identity = ""
	567	+	daemon.Incarnation = ports.BrokerDaemonIncarnation{}
	568	+	daemon.ProtocolVersion = 0
	569	+	daemon.Capabilities = 0
	570	+	daemon.Availability = domain.RemoteAvailabilityNoDaemon
	571	+	daemon.LastFailure = domain.RemoteFailure{}
	572	+	daemon.LastSuccess = time.Time{}
	573	+	daemon.InventoryKnown = false
	574	+	return daemon
	575	+	}(), SessionCount: 0}}},
564	576		{"local daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 1}}},
565	577		{"local session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: true, Session: testCatalogSession()}}},
566	578		}


@@ -561,6 +561,18 @@ func TestBrokerSnapshotDaemonByteForByte(t *testing.T) {
561	message SnapshotPart	561		message SnapshotPart
562	}{	562		}{
563	{"observed remote daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}},	563		{"observed remote daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 1, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: testDaemonObservation(), SessionCount: 1}}},
		564	+	{"reachable remote without daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotDaemonPart{HostIndex: 1, Daemon: func() ports.BrokerDaemonObservation {
		565	+	daemon := testDaemonObservation()
		566	+	daemon.Identity = ""
		567	+	daemon.Incarnation = ports.BrokerDaemonIncarnation{}
		568	+	daemon.ProtocolVersion = 0
		569	+	daemon.Capabilities = 0
		570	+	daemon.Availability = domain.RemoteAvailabilityNoDaemon
		571	+	daemon.LastFailure = domain.RemoteFailure{}
		572	+	daemon.LastSuccess = time.Time{}
		573	+	daemon.InventoryKnown = false
		574	+	return daemon
		575	+	}(), SessionCount: 0}}},
564	{"local daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 1}}},	576		{"local daemon", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 0, Part: SnapshotDaemonPart{HostIndex: 0, Daemon: testLocalDaemonObservation(), SessionCount: 1}}},
565	{"local session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: true, Session: testCatalogSession()}}},	577		{"local session", SnapshotPart{Epoch: 7, Connection: connection, Generation: 2, Revision: 5, Index: 2, Part: SnapshotSessionPart{HostIndex: 0, SessionIndex: 0, Local: true, Session: testCatalogSession()}}},
566	}	578		}

M internal/adapters/brokerwire/convert.go +1 -1


@@ -456,7 +456,7 @@ func validateSnapshotDaemonFields(daemon ports.BrokerDaemonObservation) error {
456	456		return errConvertRange
457	457		}
458	458		if daemon.Availability < domain.RemoteAvailabilityUnknown \|\|
459		-	daemon.Availability > domain.RemoteAvailabilityInvalidResponse {
	459	+	daemon.Availability > domain.RemoteAvailabilityNoDaemon {
460	460		return ErrInvalidMessage
461	461		}
462	462		if daemon.LastFailure.Kind > domain.RemoteFailureInvalidResponse {


@@ -456,7 +456,7 @@ func validateSnapshotDaemonFields(daemon ports.BrokerDaemonObservation) error {
456		return errConvertRange	456		return errConvertRange
457		}	457		}
458		if daemon.Availability < domain.RemoteAvailabilityUnknown \|\|	458		if daemon.Availability < domain.RemoteAvailabilityUnknown \|\|
459	-	daemon.Availability > domain.RemoteAvailabilityInvalidResponse {	459	+	daemon.Availability > domain.RemoteAvailabilityNoDaemon {
460		return ErrInvalidMessage	460		return ErrInvalidMessage
461		}	461		}
462		if daemon.LastFailure.Kind > domain.RemoteFailureInvalidResponse {	462		if daemon.LastFailure.Kind > domain.RemoteFailureInvalidResponse {

M internal/adapters/sshstdio/mux.go +3 -0


@@ -57,6 +57,9 @@ const (
57	57		// The capture is a diagnostic aid, so an unbounded or hostile stream can
58	58		// never grow the buffer without limit.
59	59		muxStderrLimit = 4 << 10
	60	+	// MuxExitNoDaemon is reserved for the hidden mux helper when its local
	61	+	// daemon carriage is absent. OpenSSH uses 255 for transport/auth failures.
	62	+	MuxExitNoDaemon = 73
60	63		// muxDiagnosticLimit bounds the sanitized diagnostic echoed to a logger.
61	64		muxDiagnosticLimit = 512
62	65		)


@@ -57,6 +57,9 @@ const (
57	// The capture is a diagnostic aid, so an unbounded or hostile stream can	57		// The capture is a diagnostic aid, so an unbounded or hostile stream can
58	// never grow the buffer without limit.	58		// never grow the buffer without limit.
59	muxStderrLimit = 4 << 10	59		muxStderrLimit = 4 << 10
		60	+	// MuxExitNoDaemon is reserved for the hidden mux helper when its local
		61	+	// daemon carriage is absent. OpenSSH uses 255 for transport/auth failures.
		62	+	MuxExitNoDaemon = 73
60	// muxDiagnosticLimit bounds the sanitized diagnostic echoed to a logger.	63		// muxDiagnosticLimit bounds the sanitized diagnostic echoed to a logger.
61	muxDiagnosticLimit = 512	64		muxDiagnosticLimit = 512
62	)	65		)

M internal/app/broker_offline_mux.go +6 -0


@@ -512,6 +512,9 @@ func brokerQUICUnavailable(action string, err error) error {
512	512		func runBrokerMuxStdioCommand(ctx context.Context, options brokerMuxOptions) error {
513	513		raw, err := dialBrokerMuxHelper(ctx, options)
514	514		if err != nil {
	515	+	if daemonCarriageAbsence(err) {
	516	+	return &exitCoded{code: sshstdio.MuxExitNoDaemon, err: err}
	517	+	}
515	518		return fmt.Errorf("vev: broker mux stdio: dial daemonmux carriage: %w", err)
516	519		}
517	520		bounded, ok := raw.(wire.BoundedTransport)

@@ -633,6 +636,9 @@ func runBrokerMuxQUICProxyCommand(ctx context.Context, options brokerMuxOptions)
633	636		}()
634	637		raw, err := dialBrokerMuxHelper(ctx, options)
635	638		if err != nil {
	639	+	if daemonCarriageAbsence(err) {
	640	+	return &exitCoded{code: sshstdio.MuxExitNoDaemon, err: err}
	641	+	}
636	642		return fmt.Errorf("vev: broker mux quic proxy: dial daemonmux carriage: %w", err)
637	643		}
638	644		bounded, ok := raw.(wire.BoundedTransport)


@@ -512,6 +512,9 @@ func brokerQUICUnavailable(action string, err error) error {
512	func runBrokerMuxStdioCommand(ctx context.Context, options brokerMuxOptions) error {	512		func runBrokerMuxStdioCommand(ctx context.Context, options brokerMuxOptions) error {
513	raw, err := dialBrokerMuxHelper(ctx, options)	513		raw, err := dialBrokerMuxHelper(ctx, options)
514	if err != nil {	514		if err != nil {
		515	+	if daemonCarriageAbsence(err) {
		516	+	return &exitCoded{code: sshstdio.MuxExitNoDaemon, err: err}
		517	+	}
515	return fmt.Errorf("vev: broker mux stdio: dial daemonmux carriage: %w", err)	518		return fmt.Errorf("vev: broker mux stdio: dial daemonmux carriage: %w", err)
516	}	519		}
517	bounded, ok := raw.(wire.BoundedTransport)	520		bounded, ok := raw.(wire.BoundedTransport)

@@ -633,6 +636,9 @@ func runBrokerMuxQUICProxyCommand(ctx context.Context, options brokerMuxOptions)
633	}()	636		}()
634	raw, err := dialBrokerMuxHelper(ctx, options)	637		raw, err := dialBrokerMuxHelper(ctx, options)
635	if err != nil {	638		if err != nil {
		639	+	if daemonCarriageAbsence(err) {
		640	+	return &exitCoded{code: sshstdio.MuxExitNoDaemon, err: err}
		641	+	}
636	return fmt.Errorf("vev: broker mux quic proxy: dial daemonmux carriage: %w", err)	642		return fmt.Errorf("vev: broker mux quic proxy: dial daemonmux carriage: %w", err)
637	}	643		}
638	bounded, ok := raw.(wire.BoundedTransport)	644		bounded, ok := raw.(wire.BoundedTransport)

M internal/app/broker_remote_probe.go +11 -0


@@ -105,6 +105,13 @@ func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.Remot
105	105		}
106	106		physical, err := p.connector.Connect(ctx, endpoint)
107	107		if err != nil {
	108	+	var brokerErr ports.BrokerError
	109	+	if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorNoDaemon {
	110	+	// This typed dial-time result means the SSH helper ran on the remote
	111	+	// machine but its local daemon carriage was absent. Do not infer this
	112	+	// from a later stream loss on an already authenticated connection.
	113	+	return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityNoDaemon}, nil
	114	+	}
108	115		return ports.BrokerDaemonObservation{}, err
109	116		}
110	117		adopted := false

@@ -146,6 +153,10 @@ func observeDaemonCatalogue(ctx context.Context, physical ports.BrokerPhysicalCo
146	153		}
147	154		logical, err := physical.OpenStream(ctx, request)
148	155		if err != nil {
	156	+	// A connection that completed the authenticated physical handshake had
	157	+	// a daemon. Losing its logical stream is a transport failure, not proof
	158	+	// that the daemon is absent; preserve the failure so pooled callers can
	159	+	// discard a dead connection and retry with a fresh dial.
149	160		return ports.BrokerDaemonObservation{}, err
150	161		}
151	162		if logical == nil {


@@ -105,6 +105,13 @@ func (p *brokerRemoteProbe) Probe(ctx context.Context, registration domain.Remot
105	}	105		}
106	physical, err := p.connector.Connect(ctx, endpoint)	106		physical, err := p.connector.Connect(ctx, endpoint)
107	if err != nil {	107		if err != nil {
		108	+	var brokerErr ports.BrokerError
		109	+	if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorNoDaemon {
		110	+	// This typed dial-time result means the SSH helper ran on the remote
		111	+	// machine but its local daemon carriage was absent. Do not infer this
		112	+	// from a later stream loss on an already authenticated connection.
		113	+	return ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityNoDaemon}, nil
		114	+	}
108	return ports.BrokerDaemonObservation{}, err	115		return ports.BrokerDaemonObservation{}, err
109	}	116		}
110	adopted := false	117		adopted := false

@@ -146,6 +153,10 @@ func observeDaemonCatalogue(ctx context.Context, physical ports.BrokerPhysicalCo
146	}	153		}
147	logical, err := physical.OpenStream(ctx, request)	154		logical, err := physical.OpenStream(ctx, request)
148	if err != nil {	155		if err != nil {
		156	+	// A connection that completed the authenticated physical handshake had
		157	+	// a daemon. Losing its logical stream is a transport failure, not proof
		158	+	// that the daemon is absent; preserve the failure so pooled callers can
		159	+	// discard a dead connection and retry with a fresh dial.
149	return ports.BrokerDaemonObservation{}, err	160		return ports.BrokerDaemonObservation{}, err
150	}	161		}
151	if logical == nil {	162		if logical == nil {

M internal/domain/remote_state.go +4 -0


@@ -16,6 +16,7 @@ const (
16	16		RemoteAvailabilityIncompatible
17	17		RemoteAvailabilityAuthFailed
18	18		RemoteAvailabilityInvalidResponse
	19	+	RemoteAvailabilityNoDaemon
19	20		)
20	21
21	22		// String returns the stable lowercase wire/display token for an availability.

@@ -33,6 +34,8 @@ func (a RemoteAvailability) String() string {
33	34		return "authentication_failed"
34	35		case RemoteAvailabilityInvalidResponse:
35	36		return "invalid_response"
	37	+	case RemoteAvailabilityNoDaemon:
	38	+	return "no_daemon"
36	39		default:
37	40		return "unknown"
38	41		}

@@ -155,4 +158,5 @@ const (
155	158		RemoteReasonTimeout = "timeout"
156	159		RemoteReasonMalformed = "malformed"
157	160		RemoteReasonAuthFailure = "auth_failure"
	161	+	RemoteReasonNoDaemon = "no_daemon"
158	162		)


@@ -16,6 +16,7 @@ const (
16	RemoteAvailabilityIncompatible	16		RemoteAvailabilityIncompatible
17	RemoteAvailabilityAuthFailed	17		RemoteAvailabilityAuthFailed
18	RemoteAvailabilityInvalidResponse	18		RemoteAvailabilityInvalidResponse
		19	+	RemoteAvailabilityNoDaemon
19	)	20		)
20		21
21	// String returns the stable lowercase wire/display token for an availability.	22		// String returns the stable lowercase wire/display token for an availability.

@@ -33,6 +34,8 @@ func (a RemoteAvailability) String() string {
33	return "authentication_failed"	34		return "authentication_failed"
34	case RemoteAvailabilityInvalidResponse:	35		case RemoteAvailabilityInvalidResponse:
35	return "invalid_response"	36		return "invalid_response"
		37	+	case RemoteAvailabilityNoDaemon:
		38	+	return "no_daemon"
36	default:	39		default:
37	return "unknown"	40		return "unknown"
38	}	41		}

@@ -155,4 +158,5 @@ const (
155	RemoteReasonTimeout = "timeout"	158		RemoteReasonTimeout = "timeout"
156	RemoteReasonMalformed = "malformed"	159		RemoteReasonMalformed = "malformed"
157	RemoteReasonAuthFailure = "auth_failure"	160		RemoteReasonAuthFailure = "auth_failure"
		161	+	RemoteReasonNoDaemon = "no_daemon"
158	)	162		)

M internal/ports/broker.go +4 -1


@@ -570,6 +570,7 @@ const (
570	570		BrokerErrorFatalTerminal
571	571		BrokerErrorHostConflict
572	572		BrokerErrorMembershipImmutable
	573	+	BrokerErrorNoDaemon
573	574		)
574	575
575	576		func (c BrokerErrorCode) String() string {

@@ -598,6 +599,8 @@ func (c BrokerErrorCode) String() string {
598	599		return "host_conflict"
599	600		case BrokerErrorMembershipImmutable:
600	601		return "membership_immutable"
	602	+	case BrokerErrorNoDaemon:
	603	+	return "no_daemon"
601	604		default:
602	605		return "unknown"
603	606		}

@@ -609,7 +612,7 @@ func (c BrokerErrorCode) Validate() error {
609	612		BrokerErrorCancelled, BrokerErrorOutcomeUnknown, BrokerErrorAttachmentLost,
610	613		BrokerErrorStaleEpoch, BrokerErrorConflictingPolicy,
611	614		BrokerErrorExplicitExit, BrokerErrorFatalTerminal,
612		-	BrokerErrorHostConflict, BrokerErrorMembershipImmutable:
	615	+	BrokerErrorHostConflict, BrokerErrorMembershipImmutable, BrokerErrorNoDaemon:
613	616		return nil
614	617		default:
615	618		return errors.New("ports: invalid broker error code")

 	BrokerErrorFatalTerminal
 	BrokerErrorHostConflict
 	BrokerErrorMembershipImmutable
+	BrokerErrorNoDaemon
 )
 func (c BrokerErrorCode) String() string {
 		return "host_conflict"
 	case BrokerErrorMembershipImmutable:
 		return "membership_immutable"
+	case BrokerErrorNoDaemon:
+		return "no_daemon"
 	default:
 		return "unknown"
 	}
 		BrokerErrorCancelled, BrokerErrorOutcomeUnknown, BrokerErrorAttachmentLost,
 		BrokerErrorStaleEpoch, BrokerErrorConflictingPolicy,
 		BrokerErrorExplicitExit, BrokerErrorFatalTerminal,
-		BrokerErrorHostConflict, BrokerErrorMembershipImmutable:
+		BrokerErrorHostConflict, BrokerErrorMembershipImmutable, BrokerErrorNoDaemon:
 		return nil
 	default:
 		return errors.New("ports: invalid broker error code")

M internal/ports/broker_observation.go +1 -1


@@ -107,7 +107,7 @@ func (o BrokerDaemonObservation) Validate() error {
107	107		if err := validateObservedDaemonState(o); err != nil {
108	108		return err
109	109		}
110		-	if o.Availability < domain.RemoteAvailabilityUnknown \|\| o.Availability > domain.RemoteAvailabilityInvalidResponse {
	110	+	if o.Availability < domain.RemoteAvailabilityUnknown \|\| o.Availability > domain.RemoteAvailabilityNoDaemon {
111	111		return errors.New("ports: broker daemon observation availability is out of range")
112	112		}
113	113		if o.LastFailure.Kind > domain.RemoteFailureInvalidResponse {


@@ -107,7 +107,7 @@ func (o BrokerDaemonObservation) Validate() error {
107		if err := validateObservedDaemonState(o); err != nil {	107		if err := validateObservedDaemonState(o); err != nil {
108		return err	108		return err
109		}	109		}
110	-	if o.Availability < domain.RemoteAvailabilityUnknown \|\| o.Availability > domain.RemoteAvailabilityInvalidResponse {	110	+	if o.Availability < domain.RemoteAvailabilityUnknown \|\| o.Availability > domain.RemoteAvailabilityNoDaemon {
111		return errors.New("ports: broker daemon observation availability is out of range")	111		return errors.New("ports: broker daemon observation availability is out of range")
112		}	112		}
113		if o.LastFailure.Kind > domain.RemoteFailureInvalidResponse {	113		if o.LastFailure.Kind > domain.RemoteFailureInvalidResponse {

M internal/ports/broker_store.go +1 -1


@@ -228,7 +228,7 @@ func ValidateDurableHostProjection(obs BrokerDaemonObservation) error {
228	228		if obs.Local {
229	229		return errors.New("ports: local daemon observation is never durable")
230	230		}
231		-	if obs.Availability < domain.RemoteAvailabilityUnknown \|\| obs.Availability > domain.RemoteAvailabilityInvalidResponse {
	231	+	if obs.Availability < domain.RemoteAvailabilityUnknown \|\| obs.Availability > domain.RemoteAvailabilityNoDaemon {
232	232		return errors.New("ports: broker host availability is out of range")
233	233		}
234	234		if obs.LastFailure.Kind > domain.RemoteFailureInvalidResponse {


@@ -228,7 +228,7 @@ func ValidateDurableHostProjection(obs BrokerDaemonObservation) error {
228		if obs.Local {	228		if obs.Local {
229		return errors.New("ports: local daemon observation is never durable")	229		return errors.New("ports: local daemon observation is never durable")
230		}	230		}
231	-	if obs.Availability < domain.RemoteAvailabilityUnknown \|\| obs.Availability > domain.RemoteAvailabilityInvalidResponse {	231	+	if obs.Availability < domain.RemoteAvailabilityUnknown \|\| obs.Availability > domain.RemoteAvailabilityNoDaemon {
232		return errors.New("ports: broker host availability is out of range")	232		return errors.New("ports: broker host availability is out of range")
233		}	233		}
234		if obs.LastFailure.Kind > domain.RemoteFailureInvalidResponse {	234		if obs.LastFailure.Kind > domain.RemoteFailureInvalidResponse {

M internal/ports/broker_test.go +1 -1


@@ -101,7 +101,7 @@ func TestValidateDurableHostProjection(t *testing.T) {
101	101		}, wantErr: true},
102	102		{name: "zero availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = 0 }, wantErr: true},
103	103		{name: "unknown availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityUnknown }},
104		-	{name: "availability past the closed range", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityInvalidResponse + 1 }, wantErr: true},
	104	+	{name: "availability past the closed range", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityNoDaemon + 1 }, wantErr: true},
105	105		{name: "failure kind past the closed range", mutate: func(o *BrokerDaemonObservation) { o.LastFailure.Kind = domain.RemoteFailureInvalidResponse + 1 }, wantErr: true},
106	106		{name: "missing lifecycle identity", mutate: func(o *BrokerDaemonObservation) {
107	107		o.InventoryKnown = true


@@ -101,7 +101,7 @@ func TestValidateDurableHostProjection(t *testing.T) {
101		}, wantErr: true},	101		}, wantErr: true},
102		{name: "zero availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = 0 }, wantErr: true},	102		{name: "zero availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = 0 }, wantErr: true},
103		{name: "unknown availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityUnknown }},	103		{name: "unknown availability", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityUnknown }},
104	-	{name: "availability past the closed range", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityInvalidResponse + 1 }, wantErr: true},	104	+	{name: "availability past the closed range", mutate: func(o *BrokerDaemonObservation) { o.Availability = domain.RemoteAvailabilityNoDaemon + 1 }, wantErr: true},
105		{name: "failure kind past the closed range", mutate: func(o *BrokerDaemonObservation) { o.LastFailure.Kind = domain.RemoteFailureInvalidResponse + 1 }, wantErr: true},	105		{name: "failure kind past the closed range", mutate: func(o *BrokerDaemonObservation) { o.LastFailure.Kind = domain.RemoteFailureInvalidResponse + 1 }, wantErr: true},
106		{name: "missing lifecycle identity", mutate: func(o *BrokerDaemonObservation) {	106		{name: "missing lifecycle identity", mutate: func(o *BrokerDaemonObservation) {
107		o.InventoryKnown = true	107		o.InventoryKnown = true

M internal/protocol/catalogue/catalogue.go +1 -1


@@ -265,7 +265,7 @@ func validText(value string) bool {
265	265
266	266		func validReason(reason string) bool {
267	267		switch reason {
268		-	case "", domain.RemoteReasonRefreshing, domain.RemoteReasonCatalogStale, domain.RemoteReasonHostUnreachable, domain.RemoteReasonVersionMismatch, domain.RemoteReasonSessionDown, domain.RemoteReasonSessionBroken, domain.RemoteReasonIdentityChanged, domain.RemoteReasonNotFound, domain.RemoteReasonTimeout, domain.RemoteReasonMalformed:
	268	+	case "", domain.RemoteReasonRefreshing, domain.RemoteReasonCatalogStale, domain.RemoteReasonHostUnreachable, domain.RemoteReasonVersionMismatch, domain.RemoteReasonSessionDown, domain.RemoteReasonSessionBroken, domain.RemoteReasonIdentityChanged, domain.RemoteReasonNotFound, domain.RemoteReasonTimeout, domain.RemoteReasonMalformed, domain.RemoteReasonNoDaemon:
269	269		return true
270	270		default:
271	271		return false


@@ -265,7 +265,7 @@ func validText(value string) bool {
265			265
266		func validReason(reason string) bool {	266		func validReason(reason string) bool {
267		switch reason {	267		switch reason {
268	-	case "", domain.RemoteReasonRefreshing, domain.RemoteReasonCatalogStale, domain.RemoteReasonHostUnreachable, domain.RemoteReasonVersionMismatch, domain.RemoteReasonSessionDown, domain.RemoteReasonSessionBroken, domain.RemoteReasonIdentityChanged, domain.RemoteReasonNotFound, domain.RemoteReasonTimeout, domain.RemoteReasonMalformed:	268	+	case "", domain.RemoteReasonRefreshing, domain.RemoteReasonCatalogStale, domain.RemoteReasonHostUnreachable, domain.RemoteReasonVersionMismatch, domain.RemoteReasonSessionDown, domain.RemoteReasonSessionBroken, domain.RemoteReasonIdentityChanged, domain.RemoteReasonNotFound, domain.RemoteReasonTimeout, domain.RemoteReasonMalformed, domain.RemoteReasonNoDaemon:
269		return true	269		return true
270		default:	270		default:
271		return false	271		return false

M internal/protocol/picker_interaction.go +2 -1


@@ -64,6 +64,7 @@ const (
64	64		PickerLineStatusStale
65	65		PickerLineStatusVersion
66	66		PickerLineStatusError
	67	+	PickerLineStatusNoDaemon
67	68		)
68	69
69	70		// PickerLineActions is the set of actions a row admits. The presenting client

@@ -255,7 +256,7 @@ func validPickerLineKind(kind PickerLineKind) bool {
255	256		}
256	257
257	258		func validPickerLineStatus(status PickerLineStatus) bool {
258		-	return status <= PickerLineStatusError
	259	+	return status <= PickerLineStatusNoDaemon
259	260		}
260	261
261	262		func validPickerLineActions(actions PickerLineActions) bool {

 	PickerLineStatusStale
 	PickerLineStatusVersion
 	PickerLineStatusError
+	PickerLineStatusNoDaemon
 )
 // PickerLineActions is the set of actions a row admits. The presenting client
 }
 func validPickerLineStatus(status PickerLineStatus) bool {
-	return status <= PickerLineStatusError
+	return status <= PickerLineStatusNoDaemon
 }
 func validPickerLineActions(actions PickerLineActions) bool {

M internal/usecase/broker/pool_warm_test.go +45 -0


@@ -241,6 +241,51 @@ func TestPoolCloseRetiresWarmTransports(t *testing.T) {
241	241		require.False(t, ok)
242	242		}
243	243
	244	+	func TestPoolLocalOpenDoesNotWaitForProbeAdoption(t *testing.T) {
	245	+	h := newWarmHarness(t, 2, time.Minute)
	246	+	h.visit(t, "local-daemon")
	247	+
	248	+	probe := &fakePhysical{
	249	+	endpoint: ports.BrokerDialTarget{Policy: poolPolicy()},
	250	+	identity: "remote-daemon",
	251	+	done: make(chan struct{}),
	252	+	}
	253	+	start := make(chan struct{})
	254	+	adopted := make(chan bool, 1)
	255	+	opened := make(chan error, 1)
	256	+	go func() {
	257	+	<-start
	258	+	adopted <- h.pool.AdoptPhysical(probe, poolPolicy())
	259	+	}()
	260	+	go func() {
	261	+	<-start
	262	+	h.mu.Lock()
	263	+	h.target = "local-daemon"
	264	+	h.stream++
	265	+	stream := h.stream
	266	+	h.mu.Unlock()
	267	+	logical, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
	268	+	if err == nil {
	269	+	err = logical.Close()
	270	+	}
	271	+	opened <- err
	272	+	}()
	273	+	close(start)
	274	+
	275	+	select {
	276	+	case err := <-opened:
	277	+	require.NoError(t, err, "local OpenStream must progress while another goroutine adopts a remote probe transport")
	278	+	case <-time.After(3 * time.Second):
	279	+	t.Fatal("local OpenStream blocked while probe adoption was in progress")
	280	+	}
	281	+	select {
	282	+	case ok := <-adopted:
	283	+	require.True(t, ok, "the authenticated remote physical should be adopted")
	284	+	case <-time.After(3 * time.Second):
	285	+	t.Fatal("probe transport adoption did not complete")
	286	+	}
	287	+	}
	288	+
244	289		func TestPoolSharedPhysical(t *testing.T) {
245	290		tests := []struct {
246	291		name string


@@ -241,6 +241,51 @@ func TestPoolCloseRetiresWarmTransports(t *testing.T) {
241	require.False(t, ok)	241		require.False(t, ok)
242	}	242		}
243		243
		244	+	func TestPoolLocalOpenDoesNotWaitForProbeAdoption(t *testing.T) {
		245	+	h := newWarmHarness(t, 2, time.Minute)
		246	+	h.visit(t, "local-daemon")
		247	+
		248	+	probe := &fakePhysical{
		249	+	endpoint: ports.BrokerDialTarget{Policy: poolPolicy()},
		250	+	identity: "remote-daemon",
		251	+	done: make(chan struct{}),
		252	+	}
		253	+	start := make(chan struct{})
		254	+	adopted := make(chan bool, 1)
		255	+	opened := make(chan error, 1)
		256	+	go func() {
		257	+	<-start
		258	+	adopted <- h.pool.AdoptPhysical(probe, poolPolicy())
		259	+	}()
		260	+	go func() {
		261	+	<-start
		262	+	h.mu.Lock()
		263	+	h.target = "local-daemon"
		264	+	h.stream++
		265	+	stream := h.stream
		266	+	h.mu.Unlock()
		267	+	logical, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
		268	+	if err == nil {
		269	+	err = logical.Close()
		270	+	}
		271	+	opened <- err
		272	+	}()
		273	+	close(start)
		274	+
		275	+	select {
		276	+	case err := <-opened:
		277	+	require.NoError(t, err, "local OpenStream must progress while another goroutine adopts a remote probe transport")
		278	+	case <-time.After(3 * time.Second):
		279	+	t.Fatal("local OpenStream blocked while probe adoption was in progress")
		280	+	}
		281	+	select {
		282	+	case ok := <-adopted:
		283	+	require.True(t, ok, "the authenticated remote physical should be adopted")
		284	+	case <-time.After(3 * time.Second):
		285	+	t.Fatal("probe transport adoption did not complete")
		286	+	}
		287	+	}
		288	+
244	func TestPoolSharedPhysical(t *testing.T) {	289		func TestPoolSharedPhysical(t *testing.T) {
245	tests := []struct {	290		tests := []struct {
246	name string	291		name string

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


@@ -1107,6 +1107,22 @@ func (r *Registry) apply(result probeResult) {
1107	1107		current.LastAttempt = result.at
1108	1108
1109	1109		kind, availability, failed := observationOutcome(result)
	1110	+	if !failed && availability == domain.RemoteAvailabilityNoDaemon {
	1111	+	current.Availability = availability
	1112	+	current.LastSuccess = result.at
	1113	+	current.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
	1114	+	current.ConsecutiveFailures = 0
	1115	+	current.LastFailure = domain.RemoteFailure{}
	1116	+	current.Identity = ""
	1117	+	current.Incarnation = ports.BrokerDaemonIncarnation{}
	1118	+	current.ProtocolVersion = 0
	1119	+	current.Capabilities = 0
	1120	+	current.InventoryKnown = false
	1121	+	current.Sessions = nil
	1122	+	r.hosts[result.endpoint] = current
	1123	+	r.publishLocked(true)
	1124	+	return
	1125	+	}
1110	1126		if !failed {
1111	1127		// Configured authority comes only from the registry's own projection:
1112	1128		// a probe result supplies observed state, never endpoint, registration,

@@ -1146,8 +1162,12 @@ func (r *Registry) apply(result probeResult) {
1146	1162		}
1147	1163		current.ConsecutiveFailures++
1148	1164		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1149		-	current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), result.endpoint, result.attempt))
1150		-	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err)
	1165	+	delay := r.retryDelay(current.ConsecutiveFailures)
	1166	+	if r.demand > 0 && delay > r.demandFreshForRemote {
	1167	+	delay = r.demandFreshForRemote
	1168	+	}
	1169	+	current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
	1170	+	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))
1151	1171		r.hosts[result.endpoint] = current
1152	1172		r.publishLocked(true)
1153	1173		}

@@ -1169,6 +1189,8 @@ func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.Re
1169	1189		switch result.snapshot.Availability {
1170	1190		case domain.RemoteAvailabilityReachable:
1171	1191		return domain.RemoteFailureNone, domain.RemoteAvailabilityReachable, false
	1192	+	case domain.RemoteAvailabilityNoDaemon:
	1193	+	return domain.RemoteFailureNone, domain.RemoteAvailabilityNoDaemon, false
1172	1194		case domain.RemoteAvailabilityIncompatible:
1173	1195		return domain.RemoteFailureIncompatible, domain.RemoteAvailabilityIncompatible, true
1174	1196		case domain.RemoteAvailabilityAuthFailed:


@@ -1107,6 +1107,22 @@ func (r *Registry) apply(result probeResult) {
1107		current.LastAttempt = result.at	1107		current.LastAttempt = result.at
1108			1108
1109		kind, availability, failed := observationOutcome(result)	1109		kind, availability, failed := observationOutcome(result)
			1110	+	if !failed && availability == domain.RemoteAvailabilityNoDaemon {
			1111	+	current.Availability = availability
			1112	+	current.LastSuccess = result.at
			1113	+	current.NextDue = result.at.Add(r.jitter(r.freshForLocked(false), result.endpoint, result.attempt))
			1114	+	current.ConsecutiveFailures = 0
			1115	+	current.LastFailure = domain.RemoteFailure{}
			1116	+	current.Identity = ""
			1117	+	current.Incarnation = ports.BrokerDaemonIncarnation{}
			1118	+	current.ProtocolVersion = 0
			1119	+	current.Capabilities = 0
			1120	+	current.InventoryKnown = false
			1121	+	current.Sessions = nil
			1122	+	r.hosts[result.endpoint] = current
			1123	+	r.publishLocked(true)
			1124	+	return
			1125	+	}
1110		if !failed {	1126		if !failed {
1111		// Configured authority comes only from the registry's own projection:	1127		// Configured authority comes only from the registry's own projection:
1112		// a probe result supplies observed state, never endpoint, registration,	1128		// a probe result supplies observed state, never endpoint, registration,

@@ -1146,8 +1162,12 @@ func (r *Registry) apply(result probeResult) {
1146		}	1162		}
1147		current.ConsecutiveFailures++	1163		current.ConsecutiveFailures++
1148		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}	1164		current.LastFailure = domain.RemoteFailure{Kind: kind, Err: result.err}
1149	-	current.NextDue = result.at.Add(r.jitter(r.retryDelay(current.ConsecutiveFailures), result.endpoint, result.attempt))	1165	+	delay := r.retryDelay(current.ConsecutiveFailures)
1150	-	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err)	1166	+	if r.demand > 0 && delay > r.demandFreshForRemote {
			1167	+	delay = r.demandFreshForRemote
			1168	+	}
			1169	+	current.NextDue = result.at.Add(r.jitter(delay, result.endpoint, result.attempt))
			1170	+	r.log.Debug("broker_probe_failed", "endpoint", result.endpoint, "kind", kind, "failures", current.ConsecutiveFailures, "err", result.err, "cause", errors.Unwrap(result.err))
1151		r.hosts[result.endpoint] = current	1171		r.hosts[result.endpoint] = current
1152		r.publishLocked(true)	1172		r.publishLocked(true)
1153		}	1173		}

@@ -1169,6 +1189,8 @@ func observationOutcome(result probeResult) (domain.RemoteFailureKind, domain.Re
1169		switch result.snapshot.Availability {	1189		switch result.snapshot.Availability {
1170		case domain.RemoteAvailabilityReachable:	1190		case domain.RemoteAvailabilityReachable:
1171		return domain.RemoteFailureNone, domain.RemoteAvailabilityReachable, false	1191		return domain.RemoteFailureNone, domain.RemoteAvailabilityReachable, false
			1192	+	case domain.RemoteAvailabilityNoDaemon:
			1193	+	return domain.RemoteFailureNone, domain.RemoteAvailabilityNoDaemon, false
1172		case domain.RemoteAvailabilityIncompatible:	1194		case domain.RemoteAvailabilityIncompatible:
1173		return domain.RemoteFailureIncompatible, domain.RemoteAvailabilityIncompatible, true	1195		return domain.RemoteFailureIncompatible, domain.RemoteAvailabilityIncompatible, true
1174		case domain.RemoteAvailabilityAuthFailed:	1196		case domain.RemoteAvailabilityAuthFailed:

M internal/usecase/broker/registry_test.go +78 -7


@@ -1320,6 +1320,12 @@ func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) {
1320	1320		wantAvailability: domain.RemoteAvailabilityInvalidResponse,
1321	1321		wantKind: domain.RemoteFailureInvalidResponse,
1322	1322		},
	1323	+	{
	1324	+	name: "no daemon after a prior success",
	1325	+	answer: probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityNoDaemon}},
	1326	+	wantAvailability: domain.RemoteAvailabilityNoDaemon,
	1327	+	wantKind: domain.RemoteFailureNone,
	1328	+	},
1323	1329		{
1324	1330		name: "transport error",
1325	1331		answer: probeAnswer{err: dialErr},

@@ -1344,20 +1350,55 @@ func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) {
1344	1350		reg := registration(t, "example.test", 1)
1345	1351		require.NoError(t, registry.setHosts(hostRecords(reg)))
1346	1352		startRegistry(t, registry)
	1353	+	if tt.wantAvailability == domain.RemoteAvailabilityNoDaemon {
	1354	+	reachable := probeAnswer{snapshot: ports.BrokerDaemonObservation{
	1355	+	Availability: domain.RemoteAvailabilityReachable, Identity: ports.BrokerDaemonIdentity("remote-daemon"),
	1356	+	Incarnation: ports.BrokerDaemonIncarnation{1}, ProtocolVersion: protocol.Version,
	1357	+	InventoryKnown: true, Sessions: []catalogue.RemoteCatalogSession{hostSession("old")},
	1358	+	}}
	1359	+	call := receiveCall(t, probe)
	1360	+	call.result <- reachable
	1361	+	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1362	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable && host.Identity == ports.BrokerDaemonIdentity("remote-daemon")
	1363	+	})
	1364	+	clock.Advance(5 * time.Second)
	1365	+	registry.RequestProbe(reg.Endpoint)
	1366	+	call = receiveCall(t, probe)
	1367	+	call.result <- reachable
	1368	+	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1369	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
	1370	+	})
	1371	+	clock.Advance(20 * time.Second)
	1372	+	registry.RequestProbe(reg.Endpoint)
	1373	+	}
1347	1374
1348	1375		call := receiveCall(t, probe)
1349	1376		call.result <- tt.answer
1350	1377		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1351		-	return !host.Checking && host.ConsecutiveFailures == 1
	1378	+	return !host.Checking && (host.ConsecutiveFailures == 1 \|\| tt.wantAvailability == domain.RemoteAvailabilityNoDaemon && host.Availability == domain.RemoteAvailabilityNoDaemon)
1352	1379		})
1353	1380		require.Equal(t, tt.wantAvailability, host.Availability)
1354	1381		require.Equal(t, tt.wantKind, host.LastFailure.Kind)
1355		-	require.Equal(t, uint(1), host.ConsecutiveFailures)
1356		-	require.Equal(t, uint64(1), host.FailureEpisode)
1357		-	// No observation succeeded yet, so there is no last success.
1358		-	require.True(t, host.LastSuccess.IsZero())
1359		-	// Every non-reachable outcome retries on the capped cadence.
1360		-	require.Equal(t, start.Add(defaultRetryBase), host.NextDue)
	1382	+	if tt.wantAvailability == domain.RemoteAvailabilityNoDaemon {
	1383	+	require.Zero(t, host.ConsecutiveFailures)
	1384	+	require.Zero(t, host.FailureEpisode)
	1385	+	require.Equal(t, start.Add(25*time.Second), host.LastSuccess)
	1386	+	require.Equal(t, start.Add(40*time.Second), host.NextDue)
	1387	+	require.Empty(t, host.Identity)
	1388	+	require.True(t, host.Incarnation.IsZero())
	1389	+	require.Zero(t, host.ProtocolVersion)
	1390	+	require.Zero(t, host.Capabilities)
	1391	+	require.False(t, host.InventoryKnown)
	1392	+	require.Empty(t, host.Sessions)
	1393	+	require.Equal(t, domain.RemoteFailureNone, host.LastFailure.Kind)
	1394	+	} else {
	1395	+	require.Equal(t, uint(1), host.ConsecutiveFailures)
	1396	+	require.Equal(t, uint64(1), host.FailureEpisode)
	1397	+	// No observation succeeded yet, so there is no last success.
	1398	+	require.True(t, host.LastSuccess.IsZero())
	1399	+	// Every non-reachable outcome retries on the capped cadence.
	1400	+	require.Equal(t, start.Add(defaultRetryBase), host.NextDue)
	1401	+	}
1361	1402		if tt.wantErr != nil {
1362	1403		require.ErrorIs(t, host.LastFailure, tt.wantErr)
1363	1404		} else {

@@ -1605,6 +1646,36 @@ func TestRegistryPreservesLastSuccessAndOwnsFailureEpisode(t *testing.T) {
1605	1646		require.Nil(t, host.LastFailure.Err)
1606	1647		}
1607	1648
	1649	+	func TestRegistryDemandCapsFailureBackoff(t *testing.T) {
	1650	+	start := time.Unix(100, 0)
	1651	+	clock := newManualClock(start)
	1652	+	probe := newTestProbe(1)
	1653	+	reg := registration(t, "demand.test", 1)
	1654	+	store := newTestStore()
	1655	+	require.NoError(t, store.ReplaceHosts(1, hostRecords(reg)))
	1656	+	registry := newTestRegistry(t, 1, store, probe, clock)
	1657	+	startRegistry(t, registry)
	1658	+	registry.SetDemand(true)
	1659	+	t.Cleanup(func() { registry.SetDemand(false) })
	1660	+
	1661	+	for failure := uint(1); failure <= 9; failure++ {
	1662	+	call := receiveCall(t, probe)
	1663	+	call.result <- probeAnswer{err: errors.New("daemon not available")}
	1664	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1665	+	return !host.Checking && host.ConsecutiveFailures == failure
	1666	+	})
	1667	+	require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), defaultDemandFreshForRemote, "demanded retry gap must never exceed the freshness interval")
	1668	+	clock.Advance(host.NextDue.Sub(clock.Now()))
	1669	+	registry.RequestProbe(reg.Endpoint)
	1670	+	}
	1671	+	call := receiveCall(t, probe)
	1672	+	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
	1673	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
	1674	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
	1675	+	})
	1676	+	require.Zero(t, host.ConsecutiveFailures)
	1677	+	}
	1678	+
1608	1679		func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) {
1609	1680		start := time.Unix(100, 0)
1610	1681		clock := newManualClock(start)


@@ -1320,6 +1320,12 @@ func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) {
1320		wantAvailability: domain.RemoteAvailabilityInvalidResponse,	1320		wantAvailability: domain.RemoteAvailabilityInvalidResponse,
1321		wantKind: domain.RemoteFailureInvalidResponse,	1321		wantKind: domain.RemoteFailureInvalidResponse,
1322		},	1322		},
			1323	+	{
			1324	+	name: "no daemon after a prior success",
			1325	+	answer: probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityNoDaemon}},
			1326	+	wantAvailability: domain.RemoteAvailabilityNoDaemon,
			1327	+	wantKind: domain.RemoteFailureNone,
			1328	+	},
1323		{	1329		{
1324		name: "transport error",	1330		name: "transport error",
1325		answer: probeAnswer{err: dialErr},	1331		answer: probeAnswer{err: dialErr},

@@ -1344,20 +1350,55 @@ func TestRegistryNonReachableOutcomesUseTypedFailureBackoff(t *testing.T) {
1344		reg := registration(t, "example.test", 1)	1350		reg := registration(t, "example.test", 1)
1345		require.NoError(t, registry.setHosts(hostRecords(reg)))	1351		require.NoError(t, registry.setHosts(hostRecords(reg)))
1346		startRegistry(t, registry)	1352		startRegistry(t, registry)
			1353	+	if tt.wantAvailability == domain.RemoteAvailabilityNoDaemon {
			1354	+	reachable := probeAnswer{snapshot: ports.BrokerDaemonObservation{
			1355	+	Availability: domain.RemoteAvailabilityReachable, Identity: ports.BrokerDaemonIdentity("remote-daemon"),
			1356	+	Incarnation: ports.BrokerDaemonIncarnation{1}, ProtocolVersion: protocol.Version,
			1357	+	InventoryKnown: true, Sessions: []catalogue.RemoteCatalogSession{hostSession("old")},
			1358	+	}}
			1359	+	call := receiveCall(t, probe)
			1360	+	call.result <- reachable
			1361	+	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
			1362	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable && host.Identity == ports.BrokerDaemonIdentity("remote-daemon")
			1363	+	})
			1364	+	clock.Advance(5 * time.Second)
			1365	+	registry.RequestProbe(reg.Endpoint)
			1366	+	call = receiveCall(t, probe)
			1367	+	call.result <- reachable
			1368	+	waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
			1369	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
			1370	+	})
			1371	+	clock.Advance(20 * time.Second)
			1372	+	registry.RequestProbe(reg.Endpoint)
			1373	+	}
1347			1374
1348		call := receiveCall(t, probe)	1375		call := receiveCall(t, probe)
1349		call.result <- tt.answer	1376		call.result <- tt.answer
1350		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {	1377		host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
1351	-	return !host.Checking && host.ConsecutiveFailures == 1	1378	+	return !host.Checking && (host.ConsecutiveFailures == 1 \|\| tt.wantAvailability == domain.RemoteAvailabilityNoDaemon && host.Availability == domain.RemoteAvailabilityNoDaemon)
1352		})	1379		})
1353		require.Equal(t, tt.wantAvailability, host.Availability)	1380		require.Equal(t, tt.wantAvailability, host.Availability)
1354		require.Equal(t, tt.wantKind, host.LastFailure.Kind)	1381		require.Equal(t, tt.wantKind, host.LastFailure.Kind)
1355	-	require.Equal(t, uint(1), host.ConsecutiveFailures)	1382	+	if tt.wantAvailability == domain.RemoteAvailabilityNoDaemon {
1356	-	require.Equal(t, uint64(1), host.FailureEpisode)	1383	+	require.Zero(t, host.ConsecutiveFailures)
1357	-	// No observation succeeded yet, so there is no last success.	1384	+	require.Zero(t, host.FailureEpisode)
1358	-	require.True(t, host.LastSuccess.IsZero())	1385	+	require.Equal(t, start.Add(25*time.Second), host.LastSuccess)
1359	-	// Every non-reachable outcome retries on the capped cadence.	1386	+	require.Equal(t, start.Add(40*time.Second), host.NextDue)
1360	-	require.Equal(t, start.Add(defaultRetryBase), host.NextDue)	1387	+	require.Empty(t, host.Identity)
			1388	+	require.True(t, host.Incarnation.IsZero())
			1389	+	require.Zero(t, host.ProtocolVersion)
			1390	+	require.Zero(t, host.Capabilities)
			1391	+	require.False(t, host.InventoryKnown)
			1392	+	require.Empty(t, host.Sessions)
			1393	+	require.Equal(t, domain.RemoteFailureNone, host.LastFailure.Kind)
			1394	+	} else {
			1395	+	require.Equal(t, uint(1), host.ConsecutiveFailures)
			1396	+	require.Equal(t, uint64(1), host.FailureEpisode)
			1397	+	// No observation succeeded yet, so there is no last success.
			1398	+	require.True(t, host.LastSuccess.IsZero())
			1399	+	// Every non-reachable outcome retries on the capped cadence.
			1400	+	require.Equal(t, start.Add(defaultRetryBase), host.NextDue)
			1401	+	}
1361		if tt.wantErr != nil {	1402		if tt.wantErr != nil {
1362		require.ErrorIs(t, host.LastFailure, tt.wantErr)	1403		require.ErrorIs(t, host.LastFailure, tt.wantErr)
1363		} else {	1404		} else {

@@ -1605,6 +1646,36 @@ func TestRegistryPreservesLastSuccessAndOwnsFailureEpisode(t *testing.T) {
1605		require.Nil(t, host.LastFailure.Err)	1646		require.Nil(t, host.LastFailure.Err)
1606		}	1647		}
1607			1648
			1649	+	func TestRegistryDemandCapsFailureBackoff(t *testing.T) {
			1650	+	start := time.Unix(100, 0)
			1651	+	clock := newManualClock(start)
			1652	+	probe := newTestProbe(1)
			1653	+	reg := registration(t, "demand.test", 1)
			1654	+	store := newTestStore()
			1655	+	require.NoError(t, store.ReplaceHosts(1, hostRecords(reg)))
			1656	+	registry := newTestRegistry(t, 1, store, probe, clock)
			1657	+	startRegistry(t, registry)
			1658	+	registry.SetDemand(true)
			1659	+	t.Cleanup(func() { registry.SetDemand(false) })
			1660	+
			1661	+	for failure := uint(1); failure <= 9; failure++ {
			1662	+	call := receiveCall(t, probe)
			1663	+	call.result <- probeAnswer{err: errors.New("daemon not available")}
			1664	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
			1665	+	return !host.Checking && host.ConsecutiveFailures == failure
			1666	+	})
			1667	+	require.LessOrEqual(t, host.NextDue.Sub(clock.Now()), defaultDemandFreshForRemote, "demanded retry gap must never exceed the freshness interval")
			1668	+	clock.Advance(host.NextDue.Sub(clock.Now()))
			1669	+	registry.RequestProbe(reg.Endpoint)
			1670	+	}
			1671	+	call := receiveCall(t, probe)
			1672	+	call.result <- probeAnswer{snapshot: ports.BrokerDaemonObservation{Availability: domain.RemoteAvailabilityReachable}}
			1673	+	host := waitHost(t, registry, reg.Endpoint, func(host ports.BrokerDaemonObservation) bool {
			1674	+	return !host.Checking && host.Availability == domain.RemoteAvailabilityReachable
			1675	+	})
			1676	+	require.Zero(t, host.ConsecutiveFailures)
			1677	+	}
			1678	+
1608		func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) {	1679		func TestRegistryBackoffIsExponentialAndCapped(t *testing.T) {
1609		start := time.Unix(100, 0)	1680		start := time.Unix(100, 0)
1610		clock := newManualClock(start)	1681		clock := newManualClock(start)

M internal/usecase/client/picker_catalogue_projection.go +26 -4


@@ -27,7 +27,8 @@ import (
27	27		// A session with tabs is a non-focusable header; its tab rows are the
28	28		// destinations, each resolving to its exact tab. A session without tab
29	29		// metadata is its own destination. A host status row is published only when
30		-	// it adds information: the host lists no session and is not reachable.
	30	+	// it adds information: the host lists no session and is either not reachable
	31	+	// or known to have no daemon.
31	32		//
32	33		// The cursor starts on the attachment the picker is presented over (its tab,
33	34		// else its session), otherwise on the most recent session's active tab.

@@ -100,6 +101,10 @@ func (c *pickerCatalogue) projectLocked() {
100	101		}
101	102		}
102	103		if len(observation.Sessions) == 0 && observation.Availability != domain.RemoteAvailabilityReachable {
	104	+	hostRef := pickerSelectionRef{
	105	+	kind: pickerSelectionCreateEphemeral, epoch: c.snapshot.Epoch,
	106	+	local: observation.Local, endpoint: observation.Endpoint, registration: observation.Registration,
	107	+	}
103	108		hostRows = append(hostRows, pickerRow{line: protocol.PickerLine{
104	109		Key: pickerHostRowKey(observation),
105	110		Kind: protocol.PickerLineHost,

@@ -108,9 +113,16 @@ func (c *pickerCatalogue) projectLocked() {
108	113		Status: pickerObservationStatus(observation, fresh),
109	114		StatusDetail: pickerObservationReason(observation, fresh),
110	115		Dim: true,
111		-	// A host row stays reachable for inspection and never commits.
	116	+	// A host row is focusable and offers creation only when no daemon
	117	+	// exists. The endpoint registration is revalidated at commit.
112	118		Focusable: true,
113		-	}})
	119	+	Actions: func() protocol.PickerLineActions {
	120	+	if observation.Availability == domain.RemoteAvailabilityNoDaemon {
	121	+	return protocol.PickerCanNavigate
	122	+	}
	123	+	return 0
	124	+	}(),
	125	+	}, ref: hostRef, hasRef: observation.Availability == domain.RemoteAvailabilityNoDaemon})
114	126		}
115	127		if observation.Local {
116	128		hasLocal = true

@@ -131,9 +143,17 @@ func (c *pickerCatalogue) projectLocked() {
131	143		localRows = append(localRows, block.rows...)
132	144		}
133	145		localRows = append(localRows, localHost...)
	146	+	for _, row := range localHost {
	147	+	if row.hasRef {
	148	+	refs[row.line.Key] = row.ref
	149	+	}
	150	+	}
134	151		recent = append(recent, localRows...)
135	152		for _, remote := range remotes {
136	153		for _, row := range remote.rows {
	154	+	if row.line.Kind == protocol.PickerLineHost && row.hasRef {
	155	+	refs[row.line.Key] = row.ref
	156	+	}
137	157		if row.line.Kind == protocol.PickerLineSession && row.line.Detail == "" {
138	158		row.line.Detail = "@" + remote.origin
139	159		} else if row.line.Kind == protocol.PickerLineSession {

@@ -430,7 +450,7 @@ func pickerResolveTab(ref pickerSelectionRef, session catalogue.RemoteCatalogSes
430	450		// last probe could not reach it, authenticate, or read its catalogue.
431	451		func pickerObservationFailing(observation ports.BrokerDaemonObservation) bool {
432	452		switch observation.Availability {
433		-	case domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityAuthFailed, domain.RemoteAvailabilityInvalidResponse:
	453	+	case domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityAuthFailed, domain.RemoteAvailabilityInvalidResponse, domain.RemoteAvailabilityNoDaemon:
434	454		return true
435	455		default:
436	456		return false

@@ -468,6 +488,8 @@ func pickerReasonText(reason string) string {
468	488		return "catalog malformed"
469	489		case domain.RemoteReasonAuthFailure:
470	490		return "authentication failed"
	491	+	case domain.RemoteReasonNoDaemon:
	492	+	return "no daemon — Enter to create a session"
471	493		case domain.RemoteReasonRefreshing:
472	494		return "refreshing"
473	495		case domain.RemoteReasonIdentityChanged:

 // A session with tabs is a non-focusable header; its tab rows are the
 // destinations, each resolving to its exact tab. A session without tab
 // metadata is its own destination. A host status row is published only when
-// it adds information: the host lists no session and is not reachable.
+// it adds information: the host lists no session and is either not reachable
+// or known to have no daemon.
 //
 // The cursor starts on the attachment the picker is presented over (its tab,
 // else its session), otherwise on the most recent session's active tab.
 			}
 		}
 		if len(observation.Sessions) == 0 && observation.Availability != domain.RemoteAvailabilityReachable {
+			hostRef := pickerSelectionRef{
+				kind: pickerSelectionCreateEphemeral, epoch: c.snapshot.Epoch,
+				local: observation.Local, endpoint: observation.Endpoint, registration: observation.Registration,
+			}
 			hostRows = append(hostRows, pickerRow{line: protocol.PickerLine{
 				Key:          pickerHostRowKey(observation),
 				Kind:         protocol.PickerLineHost,
 				Status:       pickerObservationStatus(observation, fresh),
 				StatusDetail: pickerObservationReason(observation, fresh),
 				Dim:          true,
-				// A host row stays reachable for inspection and never commits.
+				// A host row is focusable and offers creation only when no daemon
+				// exists. The endpoint registration is revalidated at commit.
 				Focusable: true,
-			}})
+				Actions: func() protocol.PickerLineActions {
+					if observation.Availability == domain.RemoteAvailabilityNoDaemon {
+						return protocol.PickerCanNavigate
+					}
+					return 0
+				}(),
+			}, ref: hostRef, hasRef: observation.Availability == domain.RemoteAvailabilityNoDaemon})
 		}
 		if observation.Local {
 			hasLocal = true
 		localRows = append(localRows, block.rows...)
 	}
 	localRows = append(localRows, localHost...)
+	for _, row := range localHost {
+		if row.hasRef {
+			refs[row.line.Key] = row.ref
+		}
+	}
 	recent = append(recent, localRows...)
 	for _, remote := range remotes {
 		for _, row := range remote.rows {
+			if row.line.Kind == protocol.PickerLineHost && row.hasRef {
+				refs[row.line.Key] = row.ref
+			}
 			if row.line.Kind == protocol.PickerLineSession && row.line.Detail == "" {
 				row.line.Detail = "@" + remote.origin
 			} else if row.line.Kind == protocol.PickerLineSession {
 // last probe could not reach it, authenticate, or read its catalogue.
 func pickerObservationFailing(observation ports.BrokerDaemonObservation) bool {
 	switch observation.Availability {
-	case domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityAuthFailed, domain.RemoteAvailabilityInvalidResponse:
+	case domain.RemoteAvailabilityUnreachable, domain.RemoteAvailabilityAuthFailed, domain.RemoteAvailabilityInvalidResponse, domain.RemoteAvailabilityNoDaemon:
 		return true
 	default:
 		return false
 		return "catalog malformed"
 	case domain.RemoteReasonAuthFailure:
 		return "authentication failed"
+	case domain.RemoteReasonNoDaemon:
+		return "no daemon — Enter to create a session"
 	case domain.RemoteReasonRefreshing:
 		return "refreshing"
 	case domain.RemoteReasonIdentityChanged:

M internal/usecase/client/picker_controller.go +16 -1


@@ -562,7 +562,22 @@ func (p *pickerController) ResolveKeyTarget(key string, base pickerResolveBase)
562	562		}
563	563		return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err
564	564		}
565		-	request, tab, err := p.catalogue.ResolveTarget(key, base)
	565	+	ref, ok := p.catalogue.Ref(key)
	566	+	if !ok {
	567	+	err := pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "picker row is not in the catalogue"}
	568	+	p.offerNotice("picker-refusal", pickerRefusalNotice(err))
	569	+	return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err
	570	+	}
	571	+	var request ports.BrokerOpenStreamRequest
	572	+	var tab attachmentTab
	573	+	var err error
	574	+	if ref.kind == pickerSelectionCreateNamed \|\| ref.kind == pickerSelectionCreateEphemeral {
	575	+	request, err = p.catalogue.Resolve(key, base)
	576	+	} else if ref.kind == pickerSelectionExact {
	577	+	request, tab, err = p.catalogue.ResolveTarget(key, base)
	578	+	} else {
	579	+	err = pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "this picker row is not a session destination"}
	580	+	}
566	581		if err != nil {
567	582		p.offerNotice("picker-refusal", pickerRefusalNotice(err))
568	583		return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err


@@ -562,7 +562,22 @@ func (p *pickerController) ResolveKeyTarget(key string, base pickerResolveBase)
562		}	562		}
563		return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err	563		return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err
564		}	564		}
565	-	request, tab, err := p.catalogue.ResolveTarget(key, base)	565	+	ref, ok := p.catalogue.Ref(key)
			566	+	if !ok {
			567	+	err := pickerCatalogueError{Code: pickerCatalogueUnknown, Text: "picker row is not in the catalogue"}
			568	+	p.offerNotice("picker-refusal", pickerRefusalNotice(err))
			569	+	return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err
			570	+	}
			571	+	var request ports.BrokerOpenStreamRequest
			572	+	var tab attachmentTab
			573	+	var err error
			574	+	if ref.kind == pickerSelectionCreateNamed \|\| ref.kind == pickerSelectionCreateEphemeral {
			575	+	request, err = p.catalogue.Resolve(key, base)
			576	+	} else if ref.kind == pickerSelectionExact {
			577	+	request, tab, err = p.catalogue.ResolveTarget(key, base)
			578	+	} else {
			579	+	err = pickerCatalogueError{Code: pickerCatalogueUnavailable, Text: "this picker row is not a session destination"}
			580	+	}
566		if err != nil {	581		if err != nil {
567		p.offerNotice("picker-refusal", pickerRefusalNotice(err))	582		p.offerNotice("picker-refusal", pickerRefusalNotice(err))
568		return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err	583		return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err

M internal/usecase/daemon/picker_lines.go +2 -0


@@ -404,6 +404,8 @@ func pickerStatusFor(view pickerSessionView, stopped bool) protocol.PickerLineSt
404	404		remote := pickerViewIsRemote(view)
405	405		if remote {
406	406		switch view.RemoteReason {
	407	+	case domain.RemoteReasonNoDaemon:
	408	+	return protocol.PickerLineStatusNoDaemon
407	409		case domain.RemoteReasonHostUnreachable:
408	410		return protocol.PickerLineStatusDown
409	411		case domain.RemoteReasonVersionMismatch:


@@ -404,6 +404,8 @@ func pickerStatusFor(view pickerSessionView, stopped bool) protocol.PickerLineSt
404	remote := pickerViewIsRemote(view)	404		remote := pickerViewIsRemote(view)
405	if remote {	405		if remote {
406	switch view.RemoteReason {	406		switch view.RemoteReason {
		407	+	case domain.RemoteReasonNoDaemon:
		408	+	return protocol.PickerLineStatusNoDaemon
407	case domain.RemoteReasonHostUnreachable:	409		case domain.RemoteReasonHostUnreachable:
408	return protocol.PickerLineStatusDown	410		return protocol.PickerLineStatusDown
409	case domain.RemoteReasonVersionMismatch:	411		case domain.RemoteReasonVersionMismatch:

M internal/usecase/picker/model.go +2 -0


@@ -477,6 +477,8 @@ func lineStatusBadge(status protocol.PickerLineStatus) string {
477	477		return "[version]"
478	478		case protocol.PickerLineStatusError:
479	479		return "[error]"
	480	+	case protocol.PickerLineStatusNoDaemon:
	481	+	return "[no daemon]"
480	482		default:
481	483		return ""
482	484		}


@@ -477,6 +477,8 @@ func lineStatusBadge(status protocol.PickerLineStatus) string {
477	return "[version]"	477		return "[version]"
478	case protocol.PickerLineStatusError:	478		case protocol.PickerLineStatusError:
479	return "[error]"	479		return "[error]"
		480	+	case protocol.PickerLineStatusNoDaemon:
		481	+	return "[no daemon]"
480	default:	482		default:
481	return ""	483		return ""
482	}	484		}