0ec9504 fix(broker): record client and broker lifecycle diagnostics · vev

M internal/app/broker_client.go +3 -0


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"errors"
6	6		"fmt"
	7	+	"log/slog"
7	8		"os"
8	9		"sync"
9	10		"time"

@@ -41,6 +42,7 @@ import (
41	42		// to the process clock); the remaining fields are optional and keep the
42	43		// supervisor defaults when they are left zero.
43	44		type brokerClientConfig struct {
	45	+	Logger *slog.Logger
44	46		Connector ports.BrokerConnector
45	47		Terminal ports.Terminal
46	48		Clock ports.Clock

@@ -277,6 +279,7 @@ func runBrokerClient(ctx context.Context, cfg brokerClientConfig) error {
277	279		picker := client.NewPicker(clk, 0, attachmentEnv.TrueColor)
278	280		presentation := &brokerClientPresentation{terminal: cfg.Terminal, picker: picker, ui: cfg.UI, onState: cfg.OnState, clock: clk}
279	281		supervisor, err := client.NewSupervisor(client.SupervisorConfig{
	282	+	Logger: cfg.Logger,
280	283		Connector: cfg.Connector,
281	284		Terminal: cfg.Terminal,
282	285		Clock: clk,


@@ -4,6 +4,7 @@ import (
4	"context"	4		"context"
5	"errors"	5		"errors"
6	"fmt"	6		"fmt"
		7	+	"log/slog"
7	"os"	8		"os"
8	"sync"	9		"sync"
9	"time"	10		"time"

@@ -41,6 +42,7 @@ import (
41	// to the process clock); the remaining fields are optional and keep the	42		// to the process clock); the remaining fields are optional and keep the
42	// supervisor defaults when they are left zero.	43		// supervisor defaults when they are left zero.
43	type brokerClientConfig struct {	44		type brokerClientConfig struct {
		45	+	Logger *slog.Logger
44	Connector ports.BrokerConnector	46		Connector ports.BrokerConnector
45	Terminal ports.Terminal	47		Terminal ports.Terminal
46	Clock ports.Clock	48		Clock ports.Clock

@@ -277,6 +279,7 @@ func runBrokerClient(ctx context.Context, cfg brokerClientConfig) error {
277	picker := client.NewPicker(clk, 0, attachmentEnv.TrueColor)	279		picker := client.NewPicker(clk, 0, attachmentEnv.TrueColor)
278	presentation := &brokerClientPresentation{terminal: cfg.Terminal, picker: picker, ui: cfg.UI, onState: cfg.OnState, clock: clk}	280		presentation := &brokerClientPresentation{terminal: cfg.Terminal, picker: picker, ui: cfg.UI, onState: cfg.OnState, clock: clk}
279	supervisor, err := client.NewSupervisor(client.SupervisorConfig{	281		supervisor, err := client.NewSupervisor(client.SupervisorConfig{
		282	+	Logger: cfg.Logger,
280	Connector: cfg.Connector,	283		Connector: cfg.Connector,
281	Terminal: cfg.Terminal,	284		Terminal: cfg.Terminal,
282	Clock: clk,	285		Clock: clk,

M internal/app/broker_offline.go +2 -2


@@ -56,7 +56,7 @@ const (
56	56		// refused rather than absorbed.
57	57		var brokerServePoolLimits = broker.PoolLimits{
58	58		Physical: 64,
59		-	Clients: brokeripc.DefaultMaxClients,
	59	+	Clients: 2 * brokeripc.DefaultMaxClients,
60	60		Streams: 256,
61	61		StreamsPerClient: 64,
62	62		Warm: brokerconfig.DefaultWarmTransports,

@@ -216,7 +216,7 @@ func runBrokerServeCommand(ctx context.Context, options brokerServeOptions) (ret
216	216		func sandboxLogging(dir string) (*slog.Logger, io.Closer, error) {
217	217		return logging.Setup(logging.Config{
218	218		Dir: dir,
219		-	Component: logging.Daemon,
	219	+	Component: logging.Broker,
220	220		Level: logging.EnvLevel(),
221	221		MaxBytes: logging.DefaultMaxBytes,
222	222		})


@@ -56,7 +56,7 @@ const (
56		// refused rather than absorbed.	56		// refused rather than absorbed.
57		var brokerServePoolLimits = broker.PoolLimits{	57		var brokerServePoolLimits = broker.PoolLimits{
58		Physical: 64,	58		Physical: 64,
59	-	Clients: brokeripc.DefaultMaxClients,	59	+	Clients: 2 * brokeripc.DefaultMaxClients,
60		Streams: 256,	60		Streams: 256,
61		StreamsPerClient: 64,	61		StreamsPerClient: 64,
62		Warm: brokerconfig.DefaultWarmTransports,	62		Warm: brokerconfig.DefaultWarmTransports,

@@ -216,7 +216,7 @@ func runBrokerServeCommand(ctx context.Context, options brokerServeOptions) (ret
216		func sandboxLogging(dir string) (*slog.Logger, io.Closer, error) {	216		func sandboxLogging(dir string) (*slog.Logger, io.Closer, error) {
217		return logging.Setup(logging.Config{	217		return logging.Setup(logging.Config{
218		Dir: dir,	218		Dir: dir,
219	-	Component: logging.Daemon,	219	+	Component: logging.Broker,
220		Level: logging.EnvLevel(),	220		Level: logging.EnvLevel(),
221		MaxBytes: logging.DefaultMaxBytes,	221		MaxBytes: logging.DefaultMaxBytes,
222		})	222		})

M internal/app/ui_driver.go +4 -2


@@ -230,7 +230,7 @@ func runHeadlessUIDriver(ctx context.Context, options uiDriverOptions) error {
230	230		if err != nil {
231	231		return err
232	232		}
233		-	_, logCloser, err := configureLogging(logging.Client, false)
	233	+	log, logCloser, err := configureLogging(logging.Client, false)
234	234		if err != nil {
235	235		return err
236	236		}

@@ -245,6 +245,7 @@ func runHeadlessUIDriver(ctx context.Context, options uiDriverOptions) error {
245	245		attachmentEnv := terminalAttachmentEnvironment()
246	246		attachmentEnv.Cwd = sessionEnv.Cwd
247	247		return runUIDriverClient(ctx, brokerClientConfig{
	248	+	Logger: log,
248	249		Connector: newProductionBrokerConnector(),
249	250		Terminal: terminal,
250	251		Clock: clk,

@@ -434,7 +435,7 @@ func runAttachWithOptions(ctx context.Context, intent uint8, name, remoteTarget
434	435		}
435	436		ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT)
436	437		defer stop()
437		-	_, logCloser, err := configureLogging(logging.Client, false)
	438	+	log, logCloser, err := configureLogging(logging.Client, false)
438	439		if err != nil {
439	440		return err
440	441		}

@@ -499,6 +500,7 @@ func runAttachWithOptions(ctx context.Context, intent uint8, name, remoteTarget
499	500		attachmentEnv := terminalAttachmentEnvironment()
500	501		attachmentEnv.Cwd = sessionEnv.Cwd
501	502		return runBrokerClient(ctx, brokerClientConfig{
	503	+	Logger: log,
502	504		Connector: withPreconnected(newProductionBrokerConnector(), preconnected),
503	505		Terminal: terminal,
504	506		Clock: clk,

 	if err != nil {
 		return err
 	}
-	_, logCloser, err := configureLogging(logging.Client, false)
+	log, logCloser, err := configureLogging(logging.Client, false)
 	if err != nil {
 		return err
 	}
 	attachmentEnv := terminalAttachmentEnvironment()
 	attachmentEnv.Cwd = sessionEnv.Cwd
 	return runUIDriverClient(ctx, brokerClientConfig{
+		Logger:                   log,
 		Connector:                newProductionBrokerConnector(),
 		Terminal:                 terminal,
 		Clock:                    clk,
 	}
 	ctx, stop := signal.NotifyContext(ctx, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGINT)
 	defer stop()
-	_, logCloser, err := configureLogging(logging.Client, false)
+	log, logCloser, err := configureLogging(logging.Client, false)
 	if err != nil {
 		return err
 	}
 	attachmentEnv := terminalAttachmentEnvironment()
 	attachmentEnv.Cwd = sessionEnv.Cwd
 	return runBrokerClient(ctx, brokerClientConfig{
+		Logger:                   log,
 		Connector:                withPreconnected(newProductionBrokerConnector(), preconnected),
 		Terminal:                 terminal,
 		Clock:                    clk,

M internal/logging/logging.go +1 -0


@@ -16,6 +16,7 @@ type Component string
16	16
17	17		const (
18	18		Daemon Component = "daemon"
	19	+	Broker Component = "broker"
19	20		Client Component = "client"
20	21		Stdio Component = "stdio"
21	22		)


@@ -16,6 +16,7 @@ type Component string
16		16
17	const (	17		const (
18	Daemon Component = "daemon"	18		Daemon Component = "daemon"
		19	+	Broker Component = "broker"
19	Client Component = "client"	20		Client Component = "client"
20	Stdio Component = "stdio"	21		Stdio Component = "stdio"
21	)	22		)

M internal/usecase/client/supervisor.go +39 -3


@@ -5,6 +5,7 @@ import (
5	5		"errors"
6	6		"fmt"
7	7		"io"
	8	+	"log/slog"
8	9		"math/rand/v2"
9	10		"reflect"
10	11		"sync"

@@ -290,6 +291,8 @@ type SpinnerPresentation interface {
290	291		}
291	292
292	293		type SupervisorConfig struct {
	294	+	// Logger receives bounded client lifecycle diagnostics. Optional.
	295	+	Logger *slog.Logger
293	296		// Connector establishes the broker connection. Required.
294	297		Connector ports.BrokerConnector
295	298		// Terminal is the controlling terminal the supervisor owns. Required.

@@ -382,7 +385,8 @@ type AttachmentLifecycleAction struct {
382	385		// Supervisor owns one autonomous client process: raw mode, one terminal input
383	386		// lifetime, and the broker-connectivity lifecycle.
384	387		type Supervisor struct {
385		-	cfg SupervisorConfig
	388	+	cfg SupervisorConfig
	389	+	logger *slog.Logger
386	390
387	391		// pendingPickerKey retains a commit observed while connecting. It is
388	392		// revalidated against the adopted service just like a ready-phase commit.

@@ -465,8 +469,13 @@ func NewSupervisor(cfg SupervisorConfig) (*Supervisor, error) {
465	469		} else if err := cfg.InitialNavigation.Validate(); err != nil {
466	470		return nil, fmt.Errorf("vev: supervisor initial navigation: %w", err)
467	471		}
	472	+	logger := cfg.Logger
	473	+	if logger == nil {
	474	+	logger = slog.New(slog.NewTextHandler(io.Discard, nil))
	475	+	}
468	476		supervisor := &Supervisor{
469	477		cfg: cfg,
	478	+	logger: logger,
470	479		state: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected},
471	480		clientID: newClientID(),
472	481		navigation: cfg.InitialNavigation,

@@ -732,9 +741,12 @@ func (s Supervisor) retryFailure(ctx context.Context, input terminalInputLifet
732	741		// and normalized to a typed unavailable failure so the run loop never observes
733	742		// an untyped cause.
734	743		func (s *Supervisor) connectAttempt(ctx context.Context, generation uint64) supervisorAttempt {
	744	+	s.logger.Debug("client_broker_connect_begin", "generation", generation)
735	745		service, err := s.cfg.Connector.Connect(ctx)
736	746		if err != nil {
737		-	return supervisorAttempt{generation: generation, err: normalizeUnavailable(err)}
	747	+	failure := normalizeUnavailable(err)
	748	+	s.logger.Warn("client_broker_connect_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
	749	+	return supervisorAttempt{generation: generation, err: failure}
738	750		}
739	751		if supervisorNil(service) {
740	752		return supervisorAttempt{generation: generation, err: ports.BrokerError{

@@ -745,7 +757,9 @@ func (s *Supervisor) connectAttempt(ctx context.Context, generation uint64) supe
745	757		sub, err := service.Subscribe()
746	758		if err != nil {
747	759		_ = service.Close()
748		-	return supervisorAttempt{generation: generation, err: normalizeUnavailable(err)}
	760	+	failure := normalizeUnavailable(err)
	761	+	s.logger.Warn("client_broker_subscribe_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
	762	+	return supervisorAttempt{generation: generation, err: failure}
749	763		}
750	764		if supervisorNil(sub) {
751	765		// A service that reports no subscription cannot ever become Ready, so it

@@ -766,6 +780,7 @@ func (s *Supervisor) connectAttempt(ctx context.Context, generation uint64) supe
766	780		continue
767	781		case <-service.Done():
768	782		cause := normalizeUnavailable(service.Err())
	783	+	s.logger.Warn("client_broker_initial_publication_lost", "generation", generation, "code", brokerErrorCode(cause), "error", cause, "cause", errors.Unwrap(cause))
769	784		sub.Close()
770	785		_ = service.Close()
771	786		return supervisorAttempt{generation: generation, err: cause}

@@ -1086,6 +1101,14 @@ func supervisorRetryable(err error) bool {
1086	1101		// policy) still selects the non-retryable path; every untyped failure, including
1087	1102		// a nil cause, becomes an unavailable BrokerError that preserves the original
1088	1103		// error as its local-only Cause.
	1104	+	func brokerErrorCode(err error) string {
	1105	+	var typed ports.BrokerError
	1106	+	if errors.As(err, &typed) {
	1107	+	return typed.Code.String()
	1108	+	}
	1109	+	return "unknown"
	1110	+	}
	1111	+
1089	1112		func normalizeUnavailable(err error) error {
1090	1113		if err == nil {
1091	1114		return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost"}

@@ -1094,17 +1117,27 @@ func normalizeUnavailable(err error) error {
1094	1117		if errors.As(err, &typed) {
1095	1118		return err
1096	1119		}
	1120	+	var admission ports.BrokerAdmissionError
	1121	+	if errors.As(err, &admission) {
	1122	+	return err
	1123	+	}
1097	1124		return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost", Cause: err}
1098	1125		}
1099	1126
1100	1127		// notify surfaces a connectivity failure through the optional notifier.
1101	1128		func (s *Supervisor) notifyLifecycle(kind LifecycleNoticeKind) {
	1129	+	if s != nil {
	1130	+	s.logger.Debug("client_lifecycle_notice", "kind", kind)
	1131	+	}
1102	1132		if s != nil && s.cfg.NotifyLifecycle != nil {
1103	1133		s.cfg.NotifyLifecycle(LifecycleNotice{Kind: kind})
1104	1134		}
1105	1135		}
1106	1136
1107	1137		func (s *Supervisor) notify(err error) {
	1138	+	if err != nil {
	1139	+	s.logger.Warn("client_connectivity_failure", "state", s.State().Connectivity.String(), "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
	1140	+	}
1108	1141		if s.cfg.Notify == nil \|\| err == nil {
1109	1142		return
1110	1143		}

@@ -1130,6 +1163,9 @@ func (s *Supervisor) transition(event supervisorEvent) {
1130	1163		s.state = reduceSupervisor(s.state, event)
1131	1164		state := s.state
1132	1165		s.mu.Unlock()
	1166	+	if state.Presentation != previous \|\| event.kind == supervisorBeginAttempt \|\| event.kind == supervisorTransientFailure \|\| event.kind == supervisorBrokerLoss {
	1167	+	s.logger.Debug("client_state_transition", "from", previous.String(), "to", state.Presentation.String(), "connectivity", state.Connectivity.String(), "attempt", state.Attempt, "generation", state.Generation, "error", event.err, "cause", errors.Unwrap(event.err))
	1168	+	}
1133	1169		if state.Presentation != previous {
1134	1170		s.publishPresentation(state.Presentation)
1135	1171		}

 	"errors"
 	"fmt"
 	"io"
+	"log/slog"
 	"math/rand/v2"
 	"reflect"
 	"sync"
 }
 type SupervisorConfig struct {
+	// Logger receives bounded client lifecycle diagnostics. Optional.
+	Logger *slog.Logger
 	// Connector establishes the broker connection. Required.
 	Connector ports.BrokerConnector
 	// Terminal is the controlling terminal the supervisor owns. Required.
 // Supervisor owns one autonomous client process: raw mode, one terminal input
 // lifetime, and the broker-connectivity lifecycle.
 type Supervisor struct {
-	cfg SupervisorConfig
+	cfg    SupervisorConfig
+	logger *slog.Logger
 	// pendingPickerKey retains a commit observed while connecting. It is
 	// revalidated against the adopted service just like a ready-phase commit.
 	} else if err := cfg.InitialNavigation.Validate(); err != nil {
 		return nil, fmt.Errorf("vev: supervisor initial navigation: %w", err)
 	}
+	logger := cfg.Logger
+	if logger == nil {
+		logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	}
 	supervisor := &Supervisor{
 		cfg:        cfg,
+		logger:     logger,
 		state:      State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected},
 		clientID:   newClientID(),
 		navigation: cfg.InitialNavigation,
 // and normalized to a typed unavailable failure so the run loop never observes
 // an untyped cause.
 func (s *Supervisor) connectAttempt(ctx context.Context, generation uint64) supervisorAttempt {
+	s.logger.Debug("client_broker_connect_begin", "generation", generation)
 	service, err := s.cfg.Connector.Connect(ctx)
 	if err != nil {
-		return supervisorAttempt{generation: generation, err: normalizeUnavailable(err)}
+		failure := normalizeUnavailable(err)
+		s.logger.Warn("client_broker_connect_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
+		return supervisorAttempt{generation: generation, err: failure}
 	}
 	if supervisorNil(service) {
 		return supervisorAttempt{generation: generation, err: ports.BrokerError{
 	sub, err := service.Subscribe()
 	if err != nil {
 		_ = service.Close()
-		return supervisorAttempt{generation: generation, err: normalizeUnavailable(err)}
+		failure := normalizeUnavailable(err)
+		s.logger.Warn("client_broker_subscribe_failed", "generation", generation, "code", brokerErrorCode(failure), "error", failure, "cause", errors.Unwrap(failure))
+		return supervisorAttempt{generation: generation, err: failure}
 	}
 	if supervisorNil(sub) {
 		// A service that reports no subscription cannot ever become Ready, so it
 			continue
 		case <-service.Done():
 			cause := normalizeUnavailable(service.Err())
+			s.logger.Warn("client_broker_initial_publication_lost", "generation", generation, "code", brokerErrorCode(cause), "error", cause, "cause", errors.Unwrap(cause))
 			sub.Close()
 			_ = service.Close()
 			return supervisorAttempt{generation: generation, err: cause}
 // policy) still selects the non-retryable path; every untyped failure, including
 // a nil cause, becomes an unavailable BrokerError that preserves the original
 // error as its local-only Cause.
+func brokerErrorCode(err error) string {
+	var typed ports.BrokerError
+	if errors.As(err, &typed) {
+		return typed.Code.String()
+	}
+	return "unknown"
+}
 func normalizeUnavailable(err error) error {
 	if err == nil {
 		return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost"}
 	if errors.As(err, &typed) {
 		return err
 	}
+	var admission ports.BrokerAdmissionError
+	if errors.As(err, &admission) {
+		return err
+	}
 	return ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker connection lost", Cause: err}
 }
 // notify surfaces a connectivity failure through the optional notifier.
 func (s *Supervisor) notifyLifecycle(kind LifecycleNoticeKind) {
+	if s != nil {
+		s.logger.Debug("client_lifecycle_notice", "kind", kind)
+	}
 	if s != nil && s.cfg.NotifyLifecycle != nil {
 		s.cfg.NotifyLifecycle(LifecycleNotice{Kind: kind})
 	}
 }
 func (s *Supervisor) notify(err error) {
+	if err != nil {
+		s.logger.Warn("client_connectivity_failure", "state", s.State().Connectivity.String(), "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
+	}
 	if s.cfg.Notify == nil || err == nil {
 		return
 	}
 	s.state = reduceSupervisor(s.state, event)
 	state := s.state
 	s.mu.Unlock()
+	if state.Presentation != previous || event.kind == supervisorBeginAttempt || event.kind == supervisorTransientFailure || event.kind == supervisorBrokerLoss {
+		s.logger.Debug("client_state_transition", "from", previous.String(), "to", state.Presentation.String(), "connectivity", state.Connectivity.String(), "attempt", state.Attempt, "generation", state.Generation, "error", event.err, "cause", errors.Unwrap(event.err))
+	}
 	if state.Presentation != previous {
 		s.publishPresentation(state.Presentation)
 	}