feat(app): compose the client offline behind a hidden command

Add the offline interface-parity composition: a hidden `_broker-client` runs the
real autonomous supervisor against the broker sandbox over broker IPC, so the
terminal, UI-driver and browser harnesses reach a real daemon session through a
broker logical stream. Production commands are untouched: the new composition is
reachable only through that hidden subcommand, with its own command kind and its
own connectivity-matrix row, and nothing it does leaves the operator-supplied
offline root.

- The sandbox now composes the broker-owned local observation when the config
  provisions a local binding, and stays observation-disabled when it does not, so
  a sandbox without a local binding observes nothing new and remote membership is
  still refused.
- Picker rendering samples the terminal geometry at render time and never
  consumes the terminal's resize channel, which keeps the supervisor's attachment
  geometry collector its only consumer.
- The UI-driver harness publishes Control false, because the supervisor cannot
  bind UI actions yet, and derives the generation from supervisor state. Its
  status stays reconnecting: that is the honest discovery state until autonomous
  UI action binding exists, and a test pins it so the limit cannot drift silently.

Review disposition: an independent review rejected the first version as "parity
proven" because three of the four harness tests passed with a Hello the daemon
rejects. The terminal and UI-driver tests now assert fixture-shell output on a
real terminal and fail under that mutation, verified independently. The browser
test proves authenticated transport, real logical-stream admission and live
frames, but not post-Hello session output: the adapter's transactional frames
carry no marker within the wait budget. That limit is stated in the test rather
than papered over, and the bounded marker attempt is recorded here so nobody
re-runs it expecting a different result.

commit: fc4cddcba5ab7c87f582749d075d9366bda2a586
author: brice · 33d ago · 2026-09-19 13:11 UTC
parent: 0e566b4
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

7 files · +596 -7

M internal/app/broker_offline.go +24 -4


@@ -34,9 +34,11 @@ import (
34	34		// config marker. It never reads or writes production runtime, state, or config
35	35		// and never changes an ordinary command, path, or factory.
36	36		//
37		-	// Nothing here spawns a detached process, connects or spawns a daemon, reports
38		-	// status, or reaches SSH or QUIC: those remain later slices. The composed
39		-	// process is foreground: it owns one lifetime lock for its whole run, serves
	37	+	// Nothing here spawns a detached process or daemon, reports status, or changes
	38	+	// a production path. When the sandbox configuration provisions a local binding,
	39	+	// the registry observes that existing Unix daemonmux route without opening a
	40	+	// logical stream; it owns no remote membership and never reaches SSH or QUIC.
	41	+	// The composed process is foreground: it owns one lifetime lock for its whole run, serves
40	42		// until a signal or its parent context ends it, or until the broker idles out,
41	43		// and then shuts down in a fixed order.
42	44

@@ -281,7 +283,11 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
281	283		}
282	284		defer startupLease.Release()
283	285
284		-	registry, err := broker.NewRegistryWithConfig(epoch, store, nil, clk, log, broker.RegistryConfig{ObservationDisabled: true})
	286	+	registryConfig, err := offlineRegistryConfig(config, log)
	287	+	if err != nil {
	288	+	return err
	289	+	}
	290	+	registry, err := broker.NewRegistryWithConfig(epoch, store, nil, clk, log, registryConfig)
285	291		if err != nil {
286	292		return err
287	293		}

@@ -355,6 +361,20 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
355	361		// is never logged; any other terminal accept failure is unexpected: it is logged
356	362		// as an error and reported on failed so the caller commits shutdown instead of
357	363		// leaving a bound broker that can no longer admit work.
	364	+	func offlineRegistryConfig(config brokerconfig.Config, log slog.Logger) (broker.RegistryConfig, error) {
	365	+	if config == nil {
	366	+	return broker.RegistryConfig{}, errors.New("vev: offline registry requires configuration")
	367	+	}
	368	+	if _, ok := config.LocalBinding(); !ok {
	369	+	return broker.RegistryConfig{ObservationDisabled: true}, nil
	370	+	}
	371	+	local, err := brokerLocalObservation(config, log)
	372	+	if err != nil {
	373	+	return broker.RegistryConfig{}, err
	374	+	}
	375	+	return broker.RegistryConfig{Local: local}, nil
	376	+	}
	377	+
358	378		func drainBrokerAccept(listener ports.BrokerListener, log *slog.Logger, started chan<- struct{}, drained chan<- struct{}, failed chan<- struct{}) {
359	379		defer close(drained)
360	380		close(started)

 // config marker. It never reads or writes production runtime, state, or config
 // and never changes an ordinary command, path, or factory.
 //
-// Nothing here spawns a detached process, connects or spawns a daemon, reports
+// Nothing here spawns a detached process or daemon, reports status, or changes
-// status, or reaches SSH or QUIC: those remain later slices. The composed
+// a production path. When the sandbox configuration provisions a local binding,
-// process is foreground: it owns one lifetime lock for its whole run, serves
+// the registry observes that existing Unix daemonmux route without opening a
+// logical stream; it owns no remote membership and never reaches SSH or QUIC.
+// The composed process is foreground: it owns one lifetime lock for its whole run, serves
 // until a signal or its parent context ends it, or until the broker idles out,
 // and then shuts down in a fixed order.
 	}
 	defer startupLease.Release()
-	registry, err := broker.NewRegistryWithConfig(epoch, store, nil, clk, log, broker.RegistryConfig{ObservationDisabled: true})
+	registryConfig, err := offlineRegistryConfig(config, log)
+	if err != nil {
+		return err
+	}
+	registry, err := broker.NewRegistryWithConfig(epoch, store, nil, clk, log, registryConfig)
 	if err != nil {
 		return err
 	}
 // is never logged; any other terminal accept failure is unexpected: it is logged
 // as an error and reported on failed so the caller commits shutdown instead of
 // leaving a bound broker that can no longer admit work.
+func offlineRegistryConfig(config *brokerconfig.Config, log *slog.Logger) (broker.RegistryConfig, error) {
+	if config == nil {
+		return broker.RegistryConfig{}, errors.New("vev: offline registry requires configuration")
+	}
+	if _, ok := config.LocalBinding(); !ok {
+		return broker.RegistryConfig{ObservationDisabled: true}, nil
+	}
+	local, err := brokerLocalObservation(config, log)
+	if err != nil {
+		return broker.RegistryConfig{}, err
+	}
+	return broker.RegistryConfig{Local: local}, nil
+}
 func drainBrokerAccept(listener ports.BrokerListener, log *slog.Logger, started chan<- struct{}, drained chan<- struct{}, failed chan<- struct{}) {
 	defer close(drained)
 	close(started)

A internal/app/broker_offline_client.go +211 -0

@@ -0,0 +1,211 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"io"
	7	+	"log/slog"
	8	+	"os"
	9	+	"strings"
	10	+	"sync"
	11	+
	12	+	"github.com/bnema/vev/internal/adapters/brokeripc"
	13	+	"github.com/bnema/vev/internal/adapters/clock"
	14	+	"github.com/bnema/vev/internal/adapters/term"
	15	+	"github.com/bnema/vev/internal/adapters/uidriver"
	16	+	"github.com/bnema/vev/internal/adapters/uiterm"
	17	+	"github.com/bnema/vev/internal/domain"
	18	+	"github.com/bnema/vev/internal/ports"
	19	+	"github.com/bnema/vev/internal/usecase/client"
	20	+	)
	21	+
	22	+	// brokerClientCommand is the hidden P5.4b client entry point. It is deliberately
	23	+	// separate from every ordinary attach, UI-driver, and web command: only an
	24	+	// explicitly supplied offline root can select the broker sandbox.
	25	+	const brokerClientCommand = "_broker-client"
	26	+
	27	+	const (
	28	+	offlineClientTerminal = "terminal"
	29	+	offlineClientUIDriver = "ui-driver"
	30	+	)
	31	+
	32	+	type brokerClientOptions struct {
	33	+	offlineRoot string
	34	+	harness string
	35	+	}
	36	+
	37	+	func parseBrokerClientArgs(args []string) (command, error) {
	38	+	options := brokerClientOptions{harness: offlineClientTerminal}
	39	+	var rootSet, harnessSet bool
	40	+	for index := 0; index < len(args); {
	41	+	switch args[index] {
	42	+	case "--offline-root":
	43	+	if rootSet {
	44	+	return command{}, usagef("`%s` received duplicate --offline-root", brokerClientCommand)
	45	+	}
	46	+	if index+1 >= len(args) \|\| args[index+1] == "" {
	47	+	return command{}, usagef("`--offline-root` requires a path")
	48	+	}
	49	+	options.offlineRoot, rootSet = args[index+1], true
	50	+	index += 2
	51	+	case "--harness":
	52	+	if harnessSet {
	53	+	return command{}, usagef("`%s` received duplicate --harness", brokerClientCommand)
	54	+	}
	55	+	if index+1 >= len(args) {
	56	+	return command{}, usagef("`--harness` requires terminal or ui-driver")
	57	+	}
	58	+	options.harness, harnessSet = args[index+1], true
	59	+	if options.harness != offlineClientTerminal && options.harness != offlineClientUIDriver {
	60	+	return command{}, usagef("`--harness` requires terminal or ui-driver")
	61	+	}
	62	+	index += 2
	63	+	default:
	64	+	if strings.HasPrefix(args[index], "-") {
	65	+	return command{}, usagef("unknown flag %q for `%s`", args[index], brokerClientCommand)
	66	+	}
	67	+	return command{}, usagef("`%s` does not accept positional arguments", brokerClientCommand)
	68	+	}
	69	+	}
	70	+	if !rootSet {
	71	+	return command{}, usagef("`%s` requires --offline-root", brokerClientCommand)
	72	+	}
	73	+	return command{kind: kindBrokerClient, brokerClient: options}, nil
	74	+	}
	75	+
	76	+	func runBrokerClientCommand(ctx context.Context, options brokerClientOptions) error {
	77	+	layout, err := offlineLayout(options.offlineRoot)
	78	+	if err != nil {
	79	+	return err
	80	+	}
	81	+	log := slog.New(slog.NewTextHandler(io.Discard, nil))
	82	+	switch options.harness {
	83	+	case offlineClientTerminal:
	84	+	return runOfflineClient(ctx, brokeripc.SocketPath(layout.Runtime), term.New(), log, nil, nil)
	85	+	case offlineClientUIDriver:
	86	+	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: uiDriverDefaultColumns, Rows: uiDriverDefaultRows}}, "")
	87	+	if err != nil {
	88	+	return err
	89	+	}
	90	+	defer terminal.Close()
	91	+	ui := client.NewUI(terminal, clock.New())
	92	+	return runOfflineUIDriver(ctx, brokeripc.SocketPath(layout.Runtime), terminal, ui, log, &stdioStream{reader: os.Stdin, writer: os.Stdout})
	93	+	default:
	94	+	return usagef("unknown offline client harness %q", options.harness)
	95	+	}
	96	+	}
	97	+
	98	+	// runOfflineClient is the common offline terminal composition used by the
	99	+	// physical terminal, headless UI driver, and browser terminal harnesses. The
	100	+	// connector is the real broker IPC adapter and InitialNavigation opens a real
	101	+	// broker logical stream; no direct daemon dialer is present.
	102	+	func runOfflineClient(ctx context.Context, socket string, terminal ports.Terminal, log *slog.Logger, onState func(client.State), onFailure func(error)) error {
	103	+	if terminal == nil {
	104	+	return errors.New("vev: offline client requires a terminal")
	105	+	}
	106	+	clk := clock.New()
	107	+	picker := client.NewPicker(clk, 0)
	108	+	var writer sync.Mutex
	109	+	render := func(state client.State) {
	110	+	if onState != nil {
	111	+	onState(state)
	112	+	}
	113	+	if state.Presentation != client.PresentPicker {
	114	+	return
	115	+	}
	116	+	// ResizeEvents has exactly one consumer for the whole supervisor run:
	117	+	// its attachment geometry collector. Picker rendering samples Geometry
	118	+	// here instead of competing for that single event channel.
	119	+	geometry, err := terminal.Geometry()
	120	+	if err != nil {
	121	+	return
	122	+	}
	123	+	frame := picker.Render(geometry.Size)
	124	+	notice := picker.RenderNotice(geometry.Size)
	125	+	if len(frame) == 0 && len(notice) == 0 {
	126	+	return
	127	+	}
	128	+	writer.Lock()
	129	+	defer writer.Unlock()
	130	+	if len(frame) != 0 {
	131	+	_, _ = terminal.Out().Write(frame)
	132	+	}
	133	+	if len(notice) != 0 {
	134	+	_, _ = terminal.Out().Write(notice)
	135	+	}
	136	+	_ = terminal.Flush()
	137	+	}
	138	+	navigation := client.InitialNavigationCreateEphemeral
	139	+	supervisor, err := client.NewSupervisor(client.SupervisorConfig{
	140	+	Connector: brokeripc.NewConnector(socket, brokeripc.Config{}),
	141	+	Terminal: terminal,
	142	+	Clock: clk,
	143	+	Picker: picker,
	144	+	InitialNavigation: &navigation,
	145	+	AttachmentEnvironment: client.AttachmentEnvironment{
	146	+	TermEnv: os.Getenv("TERM"), Cwd: currentWorkingDirectory(), TrueColor: client.DetectTrueColor(os.Getenv("TERM"), os.Getenv("COLORTERM"), os.Environ()),
	147	+	},
	148	+	Render: render,
	149	+	Notify: func(_ client.State, err error) {
	150	+	if onFailure != nil {
	151	+	onFailure(err)
	152	+	}
	153	+	},
	154	+	})
	155	+	if err != nil {
	156	+	return err
	157	+	}
	158	+	if log != nil {
	159	+	log.Debug("broker_offline_client", "socket", socket)
	160	+	}
	161	+	return supervisor.Run(ctx)
	162	+	}
	163	+
	164	+	func currentWorkingDirectory() string {
	165	+	cwd, err := os.Getwd()
	166	+	if err != nil {
	167	+	return ""
	168	+	}
	169	+	return cwd
	170	+	}
	171	+
	172	+	func runOfflineUIDriver(ctx context.Context, socket string, terminal uiterm.Terminal, ui client.UI, log *slog.Logger, stream io.ReadWriteCloser) error {
	173	+	runCtx, cancel := context.WithCancel(ctx)
	174	+	defer cancel()
	175	+	runnerDone := make(chan error, 1)
	176	+	states := make(chan client.State, 1)
	177	+	go func() {
	178	+	runnerDone <- runOfflineClient(runCtx, socket, terminal, log, func(state client.State) {
	179	+	select {
	180	+	case states <- state:
	181	+	default:
	182	+	}
	183	+	}, nil)
	184	+	}()
	185	+
	186	+	var state client.State
	187	+	select {
	188	+	case state = <-states:
	189	+	case err := <-runnerDone:
	190	+	return err
	191	+	case <-runCtx.Done():
	192	+	return runCtx.Err()
	193	+	}
	194	+	// The autonomous supervisor publishes terminal transactions but does not own
	195	+	// the ordinary Runner's UI action-binding loop. Reconnecting is therefore
	196	+	// the honest discovery status, and Control is false: advertising control
	197	+	// would invite actions that no foreground binding can accept. Generation is
	198	+	// the real supervisor attempt generation rather than a fabricated constant.
	199	+	ready := uidriver.Ready{Attachment: ui.Handle(), Generation: state.Generation, Control: false, Status: ports.UIStatusReconnecting}
	200	+	server := uidriver.New(ui, clock.New())
	201	+	serveDone := make(chan error, 1)
	202	+	go func() { serveDone <- server.Serve(runCtx, stream, ready) }()
	203	+	select {
	204	+	case err := <-serveDone:
	205	+	cancel()
	206	+	return errors.Join(err, ignoreContextCancellation(<-runnerDone))
	207	+	case err := <-runnerDone:
	208	+	cancel()
	209	+	return errors.Join(err, ignoreContextCancellation(<-serveDone))
	210	+	}
	211	+	}

@@ -0,0 +1,211 @@
	1	+	package app
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"io"
	7	+	"log/slog"
	8	+	"os"
	9	+	"strings"
	10	+	"sync"
	11	+
	12	+	"github.com/bnema/vev/internal/adapters/brokeripc"
	13	+	"github.com/bnema/vev/internal/adapters/clock"
	14	+	"github.com/bnema/vev/internal/adapters/term"
	15	+	"github.com/bnema/vev/internal/adapters/uidriver"
	16	+	"github.com/bnema/vev/internal/adapters/uiterm"
	17	+	"github.com/bnema/vev/internal/domain"
	18	+	"github.com/bnema/vev/internal/ports"
	19	+	"github.com/bnema/vev/internal/usecase/client"
	20	+	)
	21	+
	22	+	// brokerClientCommand is the hidden P5.4b client entry point. It is deliberately
	23	+	// separate from every ordinary attach, UI-driver, and web command: only an
	24	+	// explicitly supplied offline root can select the broker sandbox.
	25	+	const brokerClientCommand = "_broker-client"
	26	+
	27	+	const (
	28	+	offlineClientTerminal = "terminal"
	29	+	offlineClientUIDriver = "ui-driver"
	30	+	)
	31	+
	32	+	type brokerClientOptions struct {
	33	+	offlineRoot string
	34	+	harness string
	35	+	}
	36	+
	37	+	func parseBrokerClientArgs(args []string) (command, error) {
	38	+	options := brokerClientOptions{harness: offlineClientTerminal}
	39	+	var rootSet, harnessSet bool
	40	+	for index := 0; index < len(args); {
	41	+	switch args[index] {
	42	+	case "--offline-root":
	43	+	if rootSet {
	44	+	return command{}, usagef("`%s` received duplicate --offline-root", brokerClientCommand)
	45	+	}
	46	+	if index+1 >= len(args) \|\| args[index+1] == "" {
	47	+	return command{}, usagef("`--offline-root` requires a path")
	48	+	}
	49	+	options.offlineRoot, rootSet = args[index+1], true
	50	+	index += 2
	51	+	case "--harness":
	52	+	if harnessSet {
	53	+	return command{}, usagef("`%s` received duplicate --harness", brokerClientCommand)
	54	+	}
	55	+	if index+1 >= len(args) {
	56	+	return command{}, usagef("`--harness` requires terminal or ui-driver")
	57	+	}
	58	+	options.harness, harnessSet = args[index+1], true
	59	+	if options.harness != offlineClientTerminal && options.harness != offlineClientUIDriver {
	60	+	return command{}, usagef("`--harness` requires terminal or ui-driver")
	61	+	}
	62	+	index += 2
	63	+	default:
	64	+	if strings.HasPrefix(args[index], "-") {
	65	+	return command{}, usagef("unknown flag %q for `%s`", args[index], brokerClientCommand)
	66	+	}
	67	+	return command{}, usagef("`%s` does not accept positional arguments", brokerClientCommand)
	68	+	}
	69	+	}
	70	+	if !rootSet {
	71	+	return command{}, usagef("`%s` requires --offline-root", brokerClientCommand)
	72	+	}
	73	+	return command{kind: kindBrokerClient, brokerClient: options}, nil
	74	+	}
	75	+
	76	+	func runBrokerClientCommand(ctx context.Context, options brokerClientOptions) error {
	77	+	layout, err := offlineLayout(options.offlineRoot)
	78	+	if err != nil {
	79	+	return err
	80	+	}
	81	+	log := slog.New(slog.NewTextHandler(io.Discard, nil))
	82	+	switch options.harness {
	83	+	case offlineClientTerminal:
	84	+	return runOfflineClient(ctx, brokeripc.SocketPath(layout.Runtime), term.New(), log, nil, nil)
	85	+	case offlineClientUIDriver:
	86	+	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: uiDriverDefaultColumns, Rows: uiDriverDefaultRows}}, "")
	87	+	if err != nil {
	88	+	return err
	89	+	}
	90	+	defer terminal.Close()
	91	+	ui := client.NewUI(terminal, clock.New())
	92	+	return runOfflineUIDriver(ctx, brokeripc.SocketPath(layout.Runtime), terminal, ui, log, &stdioStream{reader: os.Stdin, writer: os.Stdout})
	93	+	default:
	94	+	return usagef("unknown offline client harness %q", options.harness)
	95	+	}
	96	+	}
	97	+
	98	+	// runOfflineClient is the common offline terminal composition used by the
	99	+	// physical terminal, headless UI driver, and browser terminal harnesses. The
	100	+	// connector is the real broker IPC adapter and InitialNavigation opens a real
	101	+	// broker logical stream; no direct daemon dialer is present.
	102	+	func runOfflineClient(ctx context.Context, socket string, terminal ports.Terminal, log *slog.Logger, onState func(client.State), onFailure func(error)) error {
	103	+	if terminal == nil {
	104	+	return errors.New("vev: offline client requires a terminal")
	105	+	}
	106	+	clk := clock.New()
	107	+	picker := client.NewPicker(clk, 0)
	108	+	var writer sync.Mutex
	109	+	render := func(state client.State) {
	110	+	if onState != nil {
	111	+	onState(state)
	112	+	}
	113	+	if state.Presentation != client.PresentPicker {
	114	+	return
	115	+	}
	116	+	// ResizeEvents has exactly one consumer for the whole supervisor run:
	117	+	// its attachment geometry collector. Picker rendering samples Geometry
	118	+	// here instead of competing for that single event channel.
	119	+	geometry, err := terminal.Geometry()
	120	+	if err != nil {
	121	+	return
	122	+	}
	123	+	frame := picker.Render(geometry.Size)
	124	+	notice := picker.RenderNotice(geometry.Size)
	125	+	if len(frame) == 0 && len(notice) == 0 {
	126	+	return
	127	+	}
	128	+	writer.Lock()
	129	+	defer writer.Unlock()
	130	+	if len(frame) != 0 {
	131	+	_, _ = terminal.Out().Write(frame)
	132	+	}
	133	+	if len(notice) != 0 {
	134	+	_, _ = terminal.Out().Write(notice)
	135	+	}
	136	+	_ = terminal.Flush()
	137	+	}
	138	+	navigation := client.InitialNavigationCreateEphemeral
	139	+	supervisor, err := client.NewSupervisor(client.SupervisorConfig{
	140	+	Connector: brokeripc.NewConnector(socket, brokeripc.Config{}),
	141	+	Terminal: terminal,
	142	+	Clock: clk,
	143	+	Picker: picker,
	144	+	InitialNavigation: &navigation,
	145	+	AttachmentEnvironment: client.AttachmentEnvironment{
	146	+	TermEnv: os.Getenv("TERM"), Cwd: currentWorkingDirectory(), TrueColor: client.DetectTrueColor(os.Getenv("TERM"), os.Getenv("COLORTERM"), os.Environ()),
	147	+	},
	148	+	Render: render,
	149	+	Notify: func(_ client.State, err error) {
	150	+	if onFailure != nil {
	151	+	onFailure(err)
	152	+	}
	153	+	},
	154	+	})
	155	+	if err != nil {
	156	+	return err
	157	+	}
	158	+	if log != nil {
	159	+	log.Debug("broker_offline_client", "socket", socket)
	160	+	}
	161	+	return supervisor.Run(ctx)
	162	+	}
	163	+
	164	+	func currentWorkingDirectory() string {
	165	+	cwd, err := os.Getwd()
	166	+	if err != nil {
	167	+	return ""
	168	+	}
	169	+	return cwd
	170	+	}
	171	+
	172	+	func runOfflineUIDriver(ctx context.Context, socket string, terminal uiterm.Terminal, ui client.UI, log *slog.Logger, stream io.ReadWriteCloser) error {
	173	+	runCtx, cancel := context.WithCancel(ctx)
	174	+	defer cancel()
	175	+	runnerDone := make(chan error, 1)
	176	+	states := make(chan client.State, 1)
	177	+	go func() {
	178	+	runnerDone <- runOfflineClient(runCtx, socket, terminal, log, func(state client.State) {
	179	+	select {
	180	+	case states <- state:
	181	+	default:
	182	+	}
	183	+	}, nil)
	184	+	}()
	185	+
	186	+	var state client.State
	187	+	select {
	188	+	case state = <-states:
	189	+	case err := <-runnerDone:
	190	+	return err
	191	+	case <-runCtx.Done():
	192	+	return runCtx.Err()
	193	+	}
	194	+	// The autonomous supervisor publishes terminal transactions but does not own
	195	+	// the ordinary Runner's UI action-binding loop. Reconnecting is therefore
	196	+	// the honest discovery status, and Control is false: advertising control
	197	+	// would invite actions that no foreground binding can accept. Generation is
	198	+	// the real supervisor attempt generation rather than a fabricated constant.
	199	+	ready := uidriver.Ready{Attachment: ui.Handle(), Generation: state.Generation, Control: false, Status: ports.UIStatusReconnecting}
	200	+	server := uidriver.New(ui, clock.New())
	201	+	serveDone := make(chan error, 1)
	202	+	go func() { serveDone <- server.Serve(runCtx, stream, ready) }()
	203	+	select {
	204	+	case err := <-serveDone:
	205	+	cancel()
	206	+	return errors.Join(err, ignoreContextCancellation(<-runnerDone))
	207	+	case err := <-runnerDone:
	208	+	cancel()
	209	+	return errors.Join(err, ignoreContextCancellation(<-serveDone))
	210	+	}
	211	+	}

A internal/app/broker_offline_client_test.go +323 -0

@@ -0,0 +1,323 @@
	1	+	//go:build linux
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"context"
	7	+	"encoding/json"
	8	+	"io"
	9	+	"net/http"
	10	+	"net/http/httptest"
	11	+	"os"
	12	+	"path/filepath"
	13	+	"strings"
	14	+	"testing"
	15	+	"time"
	16	+
	17	+	"github.com/coder/websocket"
	18	+	"github.com/stretchr/testify/require"
	19	+
	20	+	"github.com/bnema/vev/internal/adapters/brokerconfig"
	21	+	"github.com/bnema/vev/internal/adapters/clock"
	22	+	"github.com/bnema/vev/internal/adapters/daemonmux"
	23	+	"github.com/bnema/vev/internal/adapters/ipc"
	24	+	"github.com/bnema/vev/internal/adapters/pty"
	25	+	"github.com/bnema/vev/internal/adapters/uiterm"
	26	+	"github.com/bnema/vev/internal/adapters/webterm"
	27	+	"github.com/bnema/vev/internal/domain"
	28	+	"github.com/bnema/vev/internal/ports"
	29	+	"github.com/bnema/vev/internal/usecase/client"
	30	+	"github.com/bnema/vev/internal/usecase/daemon"
	31	+	"github.com/bnema/vev/pkg/safedir"
	32	+	)
	33	+
	34	+	type offlineClientFixture struct {
	35	+	root string
	36	+	socket string
	37	+	streams <-chan struct{}
	38	+	cancel context.CancelFunc
	39	+	prodRuntime string
	40	+	prodState string
	41	+	}
	42	+
	43	+	func startOfflineClientFixture(t *testing.T) offlineClientFixture {
	44	+	t.Helper()
	45	+	root, prodRuntime, prodState := isolateSandboxEnv(t)
	46	+	require.NoError(t, os.MkdirAll(root, 0o700))
	47	+	route := filepath.Join(root, "local-mux.sock")
	48	+	policy := brokerTestPolicy()
	49	+	writeSandboxConfig(t, root, map[string]any{
	50	+	"marker": brokerconfig.Marker, "registrations": []any{},
	51	+	"local": map[string]any{
	52	+	"identity": brokerLocalTestIdentity, "displayOrigin": "local", "route": route,
	53	+	"policy": map[string]any{"protocolVersion": policy.ProtocolVersion, "catalogSchemaVersion": policy.CatalogSchemaVersion, "environmentPolicy": "client-owned", "transport": policy.Transport, "trust": policy.Trust, "launch": policy.Launch, "isolation": policy.Isolation},
	54	+	},
	55	+	})
	56	+	layout, err := offlineLayout(root)
	57	+	require.NoError(t, err)
	58	+	for _, dir := range []string{layout.Root, layout.Runtime, layout.State, layout.Log} {
	59	+	require.NoError(t, safedir.EnsurePrivate(dir))
	60	+	}
	61	+	require.NoError(t, layout.VerifyCreated())
	62	+
	63	+	rawListener, err := ipc.ListenMux(route)
	64	+	require.NoError(t, err)
	65	+	aggregate := daemonmux.NewAggregateListener()
	66	+	binding, err := daemonmux.NewServerBinding(brokerLocalTestIdentity, ports.BrokerDaemonIncarnation{1}, policy)
	67	+	require.NoError(t, err)
	68	+	mux, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 8)
	69	+	require.NoError(t, err)
	70	+	ctx, cancel := context.WithCancel(context.Background())
	71	+	go func() {
	72	+	for {
	73	+	raw, acceptErr := rawListener.Accept()
	74	+	if acceptErr != nil {
	75	+	return
	76	+	}
	77	+	go func() { _ = mux.Adopt(ctx, raw) }()
	78	+	}
	79	+	}()
	80	+	streamSeen := make(chan struct{}, 16)
	81	+	counting := &countingServerListener{ServerListener: aggregate, seen: streamSeen}
	82	+	d := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithShell("/bin/sh", []string{"-c", "sleep 0.1; printf offline-ready; trap : TERM; while :; do read line \|\| exit; done"}))
	83	+	go func() { _ = d.Serve(ctx, counting) }()
	84	+
	85	+	// The client picker and registry must share wall-clock freshness semantics:
	86	+	// a manual clock rooted at Unix zero would correctly make the observation
	87	+	// stale to the real client clock before initial navigation resolves.
	88	+	deps, ready := testBrokerServeDeps(clock.New())
	89	+	brokerDone := runSandbox(ctx, brokerServeOptions{offlineRoot: root, idleGrace: time.Hour}, deps)
	90	+	var socket string
	91	+	select {
	92	+	case socket = <-ready:
	93	+	case serveErr := <-brokerDone:
	94	+	t.Fatalf("offline broker exited before readiness: %v", serveErr)
	95	+	case <-time.After(brokerTestWait):
	96	+	t.Fatal("offline broker did not become ready")
	97	+	}
	98	+	t.Cleanup(func() {
	99	+	cancel()
	100	+	_ = rawListener.Close()
	101	+	_ = mux.Close()
	102	+	_ = aggregate.Close()
	103	+	select {
	104	+	case <-brokerDone:
	105	+	case <-time.After(brokerTestWait):
	106	+	t.Error("offline broker did not stop")
	107	+	}
	108	+	requireProductionUntouched(t, prodRuntime, prodState)
	109	+	})
	110	+	return offlineClientFixture{root: root, socket: socket, streams: streamSeen, cancel: cancel, prodRuntime: prodRuntime, prodState: prodState}
	111	+	}
	112	+
	113	+	type countingServerListener struct {
	114	+	ports.ServerListener
	115	+	seen chan<- struct{}
	116	+	}
	117	+
	118	+	func (l *countingServerListener) Accept() (ports.ServerConnection, error) {
	119	+	connection, err := l.ServerListener.Accept()
	120	+	if err == nil {
	121	+	select {
	122	+	case l.seen <- struct{}{}:
	123	+	default:
	124	+	}
	125	+	}
	126	+	return connection, err
	127	+	}
	128	+
	129	+	func awaitLogicalStream(t *testing.T, streams <-chan struct{}) {
	130	+	t.Helper()
	131	+	select {
	132	+	case <-streams:
	133	+	case <-time.After(brokerTestWait):
	134	+	t.Fatal("client did not reach a session through a broker logical stream")
	135	+	}
	136	+	}
	137	+
	138	+	func offlineSnapshotText(snapshot ports.UISnapshot) string {
	139	+	var text strings.Builder
	140	+	for _, cell := range snapshot.Cells {
	141	+	if !cell.Continuation {
	142	+	text.WriteString(cell.Text)
	143	+	}
	144	+	}
	145	+	return text.String()
	146	+	}
	147	+
	148	+	func awaitUITerminalMarker(t testing.T, terminal uiterm.Terminal, marker string) {
	149	+	t.Helper()
	150	+	deadline := time.NewTimer(brokerTestWait)
	151	+	defer deadline.Stop()
	152	+	for {
	153	+	if snapshot, err := terminal.Snapshot(); err == nil && strings.Contains(offlineSnapshotText(snapshot), marker) {
	154	+	return
	155	+	}
	156	+	select {
	157	+	case <-terminal.Changes():
	158	+	case <-deadline.C:
	159	+	t.Fatalf("terminal never rendered session marker %q", marker)
	160	+	}
	161	+	}
	162	+	}
	163	+
	164	+	func TestParseBrokerClientArgsAndProductionDispatchIsolation(t *testing.T) {
	165	+	parsed, err := parseArgs([]string{brokerClientCommand, "--offline-root", "/tmp/offline", "--harness", offlineClientUIDriver})
	166	+	require.NoError(t, err)
	167	+	require.Equal(t, kindBrokerClient, parsed.kind)
	168	+	require.Equal(t, connectivityLocalOnly, mustConnectivityOwner(t, "kindBrokerClient"))
	169	+
	170	+	ordinary := []struct {
	171	+	args []string
	172	+	kind cmdKind
	173	+	}{{nil, kindAttach}, {[]string{"list"}, kindList}, {[]string{"--ui-driver"}, kindUIDriver}, {[]string{"--web-daemon"}, kindWebDaemon}}
	174	+	for _, test := range ordinary {
	175	+	command, parseErr := parseArgs(test.args)
	176	+	require.NoError(t, parseErr)
	177	+	require.Equal(t, test.kind, command.kind)
	178	+	require.NotEqual(t, kindBrokerClient, command.kind)
	179	+	}
	180	+
	181	+	rejections := []struct {
	182	+	name string
	183	+	args []string
	184	+	}{
	185	+	{name: "missing root", args: []string{brokerClientCommand}},
	186	+	{name: "missing root value", args: []string{brokerClientCommand, "--offline-root"}},
	187	+	{name: "duplicate root", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "--offline-root", "/tmp/b"}},
	188	+	{name: "unknown flag", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "--unknown"}},
	189	+	{name: "positional", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "extra"}},
	190	+	{name: "bad harness", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "--harness", "other"}},
	191	+	}
	192	+	for _, test := range rejections {
	193	+	t.Run(test.name, func(t *testing.T) {
	194	+	_, parseErr := parseArgs(test.args)
	195	+	require.Error(t, parseErr)
	196	+	})
	197	+	}
	198	+	}
	199	+
	200	+	func mustConnectivityOwner(t *testing.T, kind string) connectivityOwner {
	201	+	t.Helper()
	202	+	owner, ok := connectivityOwnerFor(kind)
	203	+	require.True(t, ok)
	204	+	return owner
	205	+	}
	206	+
	207	+	func TestOfflineClientTerminalReachesSessionThroughBrokerIPC(t *testing.T) {
	208	+	fixture := startOfflineClientFixture(t)
	209	+	ctx, cancel := context.WithCancel(t.Context())
	210	+	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, "")
	211	+	require.NoError(t, err)
	212	+	done := make(chan error, 1)
	213	+	failures := make(chan error, 1)
	214	+	go func() {
	215	+	done <- runOfflineClient(ctx, fixture.socket, terminal, discardLog(), nil, func(err error) {
	216	+	select {
	217	+	case failures <- err:
	218	+	default:
	219	+	}
	220	+	})
	221	+	}()
	222	+	select {
	223	+	case <-fixture.streams:
	224	+	case err := <-failures:
	225	+	t.Fatalf("offline client stream open failed: %v (%#v)", err, err)
	226	+	case <-time.After(brokerTestWait):
	227	+	t.Fatal("client did not reach a session through a broker logical stream")
	228	+	}
	229	+	awaitUITerminalMarker(t, terminal, "offline-ready")
	230	+	cancel()
	231	+	select {
	232	+	case err := <-done:
	233	+	require.NoError(t, ignoreContextCancellation(err))
	234	+	case <-time.After(brokerTestWait):
	235	+	t.Fatal("offline terminal client did not stop")
	236	+	}
	237	+	}
	238	+
	239	+	func TestOfflineClientUIDriverReachesSessionThroughBrokerIPC(t *testing.T) {
	240	+	fixture := startOfflineClientFixture(t)
	241	+	ctx, cancel := context.WithCancel(t.Context())
	242	+	defer cancel()
	243	+	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, "")
	244	+	require.NoError(t, err)
	245	+	ui := client.NewUI(terminal, clock.New())
	246	+	serverEnd, clientEnd := io.Pipe()
	247	+	writerReader, writerEnd := io.Pipe()
	248	+	stream := &stdioStream{reader: serverEnd, writer: writerEnd}
	249	+	done := make(chan error, 1)
	250	+	go func() { done <- runOfflineUIDriver(ctx, fixture.socket, terminal, ui, discardLog(), stream) }()
	251	+	awaitLogicalStream(t, fixture.streams)
	252	+	decoder := json.NewDecoder(writerReader)
	253	+	readyResult := make(chan map[string]any, 1)
	254	+	readyError := make(chan error, 1)
	255	+	go func() {
	256	+	var ready map[string]any
	257	+	if err := decoder.Decode(&ready); err != nil {
	258	+	readyError <- err
	259	+	return
	260	+	}
	261	+	readyResult <- ready
	262	+	}()
	263	+	select {
	264	+	case ready := <-readyResult:
	265	+	result, ok := ready["result"].(map[string]any)
	266	+	require.True(t, ok)
	267	+	require.Equal(t, false, result["control"], "autonomous supervisor has no UI action binding")
	268	+	require.Equal(t, string(ports.UIStatusReconnecting), result["status"], "reconnecting is the honest discovery status until autonomous UI binding exists")
	269	+	require.NotZero(t, result["generation"], "ready carries the real supervisor generation")
	270	+	case err := <-readyError:
	271	+	t.Fatalf("decode UI-driver ready: %v", err)
	272	+	case <-time.After(brokerTestWait):
	273	+	t.Fatal("offline UI driver did not publish ready")
	274	+	}
	275	+	awaitUITerminalMarker(t, terminal, "offline-ready")
	276	+	_ = clientEnd.Close()
	277	+	_ = writerReader.Close()
	278	+	cancel()
	279	+	select {
	280	+	case <-done:
	281	+	case <-time.After(brokerTestWait):
	282	+	t.Fatal("offline UI driver did not stop")
	283	+	}
	284	+	}
	285	+
	286	+	func TestOfflineClientBrowserReachesSessionThroughBrokerIPC(t *testing.T) {
	287	+	fixture := startOfflineClientFixture(t)
	288	+	ctx, cancel := context.WithCancel(t.Context())
	289	+	defer cancel()
	290	+	token := strings.Repeat("b", 43)
	291	+	handler, err := webterm.NewServer(ctx, webterm.Settings{}, token, func(runCtx context.Context, terminal *webterm.Terminal) error {
	292	+	return runOfflineClient(runCtx, fixture.socket, terminal, discardLog(), nil, nil)
	293	+	})
	294	+	require.NoError(t, err)
	295	+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { r.Host = webterm.Address; handler.ServeHTTP(w, r) }))
	296	+	defer server.Close()
	297	+	headers := http.Header{"Origin": {webterm.Origin}, "Cookie": {"vev-web-session=" + token}}
	298	+	connection, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(server.URL, "http")+"/ws", &websocket.DialOptions{HTTPHeader: headers})
	299	+	require.NoError(t, err)
	300	+	defer connection.CloseNow()
	301	+	connection.SetReadLimit(64 << 20)
	302	+	awaitLogicalStream(t, fixture.streams)
	303	+	readCtx, stopRead := context.WithTimeout(ctx, brokerTestWait)
	304	+	defer stopRead()
	305	+	// This proves authenticated browser transport, real broker logical-stream
	306	+	// admission, and live browser frames. It does not prove post-Hello session
	307	+	// output: webterm may publish its transactional full-screen snapshot before
	308	+	// the fixture marker and does not guarantee another frame containing it.
	309	+	_, frame, err := connection.Read(readCtx)
	310	+	require.NoError(t, err)
	311	+	require.NotEmpty(t, frame, "browser emitted no frame after broker logical-stream admission")
	312	+	require.NoError(t, connection.CloseNow())
	313	+	waited := make(chan struct{})
	314	+	go func() {
	315	+	handler.Wait()
	316	+	close(waited)
	317	+	}()
	318	+	select {
	319	+	case <-waited:
	320	+	case <-time.After(brokerTestWait):
	321	+	t.Fatal("browser handler did not stop")
	322	+	}
	323	+	}

@@ -0,0 +1,323 @@
	1	+	//go:build linux
	2	+
	3	+	package app
	4	+
	5	+	import (
	6	+	"context"
	7	+	"encoding/json"
	8	+	"io"
	9	+	"net/http"
	10	+	"net/http/httptest"
	11	+	"os"
	12	+	"path/filepath"
	13	+	"strings"
	14	+	"testing"
	15	+	"time"
	16	+
	17	+	"github.com/coder/websocket"
	18	+	"github.com/stretchr/testify/require"
	19	+
	20	+	"github.com/bnema/vev/internal/adapters/brokerconfig"
	21	+	"github.com/bnema/vev/internal/adapters/clock"
	22	+	"github.com/bnema/vev/internal/adapters/daemonmux"
	23	+	"github.com/bnema/vev/internal/adapters/ipc"
	24	+	"github.com/bnema/vev/internal/adapters/pty"
	25	+	"github.com/bnema/vev/internal/adapters/uiterm"
	26	+	"github.com/bnema/vev/internal/adapters/webterm"
	27	+	"github.com/bnema/vev/internal/domain"
	28	+	"github.com/bnema/vev/internal/ports"
	29	+	"github.com/bnema/vev/internal/usecase/client"
	30	+	"github.com/bnema/vev/internal/usecase/daemon"
	31	+	"github.com/bnema/vev/pkg/safedir"
	32	+	)
	33	+
	34	+	type offlineClientFixture struct {
	35	+	root string
	36	+	socket string
	37	+	streams <-chan struct{}
	38	+	cancel context.CancelFunc
	39	+	prodRuntime string
	40	+	prodState string
	41	+	}
	42	+
	43	+	func startOfflineClientFixture(t *testing.T) offlineClientFixture {
	44	+	t.Helper()
	45	+	root, prodRuntime, prodState := isolateSandboxEnv(t)
	46	+	require.NoError(t, os.MkdirAll(root, 0o700))
	47	+	route := filepath.Join(root, "local-mux.sock")
	48	+	policy := brokerTestPolicy()
	49	+	writeSandboxConfig(t, root, map[string]any{
	50	+	"marker": brokerconfig.Marker, "registrations": []any{},
	51	+	"local": map[string]any{
	52	+	"identity": brokerLocalTestIdentity, "displayOrigin": "local", "route": route,
	53	+	"policy": map[string]any{"protocolVersion": policy.ProtocolVersion, "catalogSchemaVersion": policy.CatalogSchemaVersion, "environmentPolicy": "client-owned", "transport": policy.Transport, "trust": policy.Trust, "launch": policy.Launch, "isolation": policy.Isolation},
	54	+	},
	55	+	})
	56	+	layout, err := offlineLayout(root)
	57	+	require.NoError(t, err)
	58	+	for _, dir := range []string{layout.Root, layout.Runtime, layout.State, layout.Log} {
	59	+	require.NoError(t, safedir.EnsurePrivate(dir))
	60	+	}
	61	+	require.NoError(t, layout.VerifyCreated())
	62	+
	63	+	rawListener, err := ipc.ListenMux(route)
	64	+	require.NoError(t, err)
	65	+	aggregate := daemonmux.NewAggregateListener()
	66	+	binding, err := daemonmux.NewServerBinding(brokerLocalTestIdentity, ports.BrokerDaemonIncarnation{1}, policy)
	67	+	require.NoError(t, err)
	68	+	mux, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 8)
	69	+	require.NoError(t, err)
	70	+	ctx, cancel := context.WithCancel(context.Background())
	71	+	go func() {
	72	+	for {
	73	+	raw, acceptErr := rawListener.Accept()
	74	+	if acceptErr != nil {
	75	+	return
	76	+	}
	77	+	go func() { _ = mux.Adopt(ctx, raw) }()
	78	+	}
	79	+	}()
	80	+	streamSeen := make(chan struct{}, 16)
	81	+	counting := &countingServerListener{ServerListener: aggregate, seen: streamSeen}
	82	+	d := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemon.WithShell("/bin/sh", []string{"-c", "sleep 0.1; printf offline-ready; trap : TERM; while :; do read line \|\| exit; done"}))
	83	+	go func() { _ = d.Serve(ctx, counting) }()
	84	+
	85	+	// The client picker and registry must share wall-clock freshness semantics:
	86	+	// a manual clock rooted at Unix zero would correctly make the observation
	87	+	// stale to the real client clock before initial navigation resolves.
	88	+	deps, ready := testBrokerServeDeps(clock.New())
	89	+	brokerDone := runSandbox(ctx, brokerServeOptions{offlineRoot: root, idleGrace: time.Hour}, deps)
	90	+	var socket string
	91	+	select {
	92	+	case socket = <-ready:
	93	+	case serveErr := <-brokerDone:
	94	+	t.Fatalf("offline broker exited before readiness: %v", serveErr)
	95	+	case <-time.After(brokerTestWait):
	96	+	t.Fatal("offline broker did not become ready")
	97	+	}
	98	+	t.Cleanup(func() {
	99	+	cancel()
	100	+	_ = rawListener.Close()
	101	+	_ = mux.Close()
	102	+	_ = aggregate.Close()
	103	+	select {
	104	+	case <-brokerDone:
	105	+	case <-time.After(brokerTestWait):
	106	+	t.Error("offline broker did not stop")
	107	+	}
	108	+	requireProductionUntouched(t, prodRuntime, prodState)
	109	+	})
	110	+	return offlineClientFixture{root: root, socket: socket, streams: streamSeen, cancel: cancel, prodRuntime: prodRuntime, prodState: prodState}
	111	+	}
	112	+
	113	+	type countingServerListener struct {
	114	+	ports.ServerListener
	115	+	seen chan<- struct{}
	116	+	}
	117	+
	118	+	func (l *countingServerListener) Accept() (ports.ServerConnection, error) {
	119	+	connection, err := l.ServerListener.Accept()
	120	+	if err == nil {
	121	+	select {
	122	+	case l.seen <- struct{}{}:
	123	+	default:
	124	+	}
	125	+	}
	126	+	return connection, err
	127	+	}
	128	+
	129	+	func awaitLogicalStream(t *testing.T, streams <-chan struct{}) {
	130	+	t.Helper()
	131	+	select {
	132	+	case <-streams:
	133	+	case <-time.After(brokerTestWait):
	134	+	t.Fatal("client did not reach a session through a broker logical stream")
	135	+	}
	136	+	}
	137	+
	138	+	func offlineSnapshotText(snapshot ports.UISnapshot) string {
	139	+	var text strings.Builder
	140	+	for _, cell := range snapshot.Cells {
	141	+	if !cell.Continuation {
	142	+	text.WriteString(cell.Text)
	143	+	}
	144	+	}
	145	+	return text.String()
	146	+	}
	147	+
	148	+	func awaitUITerminalMarker(t testing.T, terminal uiterm.Terminal, marker string) {
	149	+	t.Helper()
	150	+	deadline := time.NewTimer(brokerTestWait)
	151	+	defer deadline.Stop()
	152	+	for {
	153	+	if snapshot, err := terminal.Snapshot(); err == nil && strings.Contains(offlineSnapshotText(snapshot), marker) {
	154	+	return
	155	+	}
	156	+	select {
	157	+	case <-terminal.Changes():
	158	+	case <-deadline.C:
	159	+	t.Fatalf("terminal never rendered session marker %q", marker)
	160	+	}
	161	+	}
	162	+	}
	163	+
	164	+	func TestParseBrokerClientArgsAndProductionDispatchIsolation(t *testing.T) {
	165	+	parsed, err := parseArgs([]string{brokerClientCommand, "--offline-root", "/tmp/offline", "--harness", offlineClientUIDriver})
	166	+	require.NoError(t, err)
	167	+	require.Equal(t, kindBrokerClient, parsed.kind)
	168	+	require.Equal(t, connectivityLocalOnly, mustConnectivityOwner(t, "kindBrokerClient"))
	169	+
	170	+	ordinary := []struct {
	171	+	args []string
	172	+	kind cmdKind
	173	+	}{{nil, kindAttach}, {[]string{"list"}, kindList}, {[]string{"--ui-driver"}, kindUIDriver}, {[]string{"--web-daemon"}, kindWebDaemon}}
	174	+	for _, test := range ordinary {
	175	+	command, parseErr := parseArgs(test.args)
	176	+	require.NoError(t, parseErr)
	177	+	require.Equal(t, test.kind, command.kind)
	178	+	require.NotEqual(t, kindBrokerClient, command.kind)
	179	+	}
	180	+
	181	+	rejections := []struct {
	182	+	name string
	183	+	args []string
	184	+	}{
	185	+	{name: "missing root", args: []string{brokerClientCommand}},
	186	+	{name: "missing root value", args: []string{brokerClientCommand, "--offline-root"}},
	187	+	{name: "duplicate root", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "--offline-root", "/tmp/b"}},
	188	+	{name: "unknown flag", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "--unknown"}},
	189	+	{name: "positional", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "extra"}},
	190	+	{name: "bad harness", args: []string{brokerClientCommand, "--offline-root", "/tmp/a", "--harness", "other"}},
	191	+	}
	192	+	for _, test := range rejections {
	193	+	t.Run(test.name, func(t *testing.T) {
	194	+	_, parseErr := parseArgs(test.args)
	195	+	require.Error(t, parseErr)
	196	+	})
	197	+	}
	198	+	}
	199	+
	200	+	func mustConnectivityOwner(t *testing.T, kind string) connectivityOwner {
	201	+	t.Helper()
	202	+	owner, ok := connectivityOwnerFor(kind)
	203	+	require.True(t, ok)
	204	+	return owner
	205	+	}
	206	+
	207	+	func TestOfflineClientTerminalReachesSessionThroughBrokerIPC(t *testing.T) {
	208	+	fixture := startOfflineClientFixture(t)
	209	+	ctx, cancel := context.WithCancel(t.Context())
	210	+	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, "")
	211	+	require.NoError(t, err)
	212	+	done := make(chan error, 1)
	213	+	failures := make(chan error, 1)
	214	+	go func() {
	215	+	done <- runOfflineClient(ctx, fixture.socket, terminal, discardLog(), nil, func(err error) {
	216	+	select {
	217	+	case failures <- err:
	218	+	default:
	219	+	}
	220	+	})
	221	+	}()
	222	+	select {
	223	+	case <-fixture.streams:
	224	+	case err := <-failures:
	225	+	t.Fatalf("offline client stream open failed: %v (%#v)", err, err)
	226	+	case <-time.After(brokerTestWait):
	227	+	t.Fatal("client did not reach a session through a broker logical stream")
	228	+	}
	229	+	awaitUITerminalMarker(t, terminal, "offline-ready")
	230	+	cancel()
	231	+	select {
	232	+	case err := <-done:
	233	+	require.NoError(t, ignoreContextCancellation(err))
	234	+	case <-time.After(brokerTestWait):
	235	+	t.Fatal("offline terminal client did not stop")
	236	+	}
	237	+	}
	238	+
	239	+	func TestOfflineClientUIDriverReachesSessionThroughBrokerIPC(t *testing.T) {
	240	+	fixture := startOfflineClientFixture(t)
	241	+	ctx, cancel := context.WithCancel(t.Context())
	242	+	defer cancel()
	243	+	terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, "")
	244	+	require.NoError(t, err)
	245	+	ui := client.NewUI(terminal, clock.New())
	246	+	serverEnd, clientEnd := io.Pipe()
	247	+	writerReader, writerEnd := io.Pipe()
	248	+	stream := &stdioStream{reader: serverEnd, writer: writerEnd}
	249	+	done := make(chan error, 1)
	250	+	go func() { done <- runOfflineUIDriver(ctx, fixture.socket, terminal, ui, discardLog(), stream) }()
	251	+	awaitLogicalStream(t, fixture.streams)
	252	+	decoder := json.NewDecoder(writerReader)
	253	+	readyResult := make(chan map[string]any, 1)
	254	+	readyError := make(chan error, 1)
	255	+	go func() {
	256	+	var ready map[string]any
	257	+	if err := decoder.Decode(&ready); err != nil {
	258	+	readyError <- err
	259	+	return
	260	+	}
	261	+	readyResult <- ready
	262	+	}()
	263	+	select {
	264	+	case ready := <-readyResult:
	265	+	result, ok := ready["result"].(map[string]any)
	266	+	require.True(t, ok)
	267	+	require.Equal(t, false, result["control"], "autonomous supervisor has no UI action binding")
	268	+	require.Equal(t, string(ports.UIStatusReconnecting), result["status"], "reconnecting is the honest discovery status until autonomous UI binding exists")
	269	+	require.NotZero(t, result["generation"], "ready carries the real supervisor generation")
	270	+	case err := <-readyError:
	271	+	t.Fatalf("decode UI-driver ready: %v", err)
	272	+	case <-time.After(brokerTestWait):
	273	+	t.Fatal("offline UI driver did not publish ready")
	274	+	}
	275	+	awaitUITerminalMarker(t, terminal, "offline-ready")
	276	+	_ = clientEnd.Close()
	277	+	_ = writerReader.Close()
	278	+	cancel()
	279	+	select {
	280	+	case <-done:
	281	+	case <-time.After(brokerTestWait):
	282	+	t.Fatal("offline UI driver did not stop")
	283	+	}
	284	+	}
	285	+
	286	+	func TestOfflineClientBrowserReachesSessionThroughBrokerIPC(t *testing.T) {
	287	+	fixture := startOfflineClientFixture(t)
	288	+	ctx, cancel := context.WithCancel(t.Context())
	289	+	defer cancel()
	290	+	token := strings.Repeat("b", 43)
	291	+	handler, err := webterm.NewServer(ctx, webterm.Settings{}, token, func(runCtx context.Context, terminal *webterm.Terminal) error {
	292	+	return runOfflineClient(runCtx, fixture.socket, terminal, discardLog(), nil, nil)
	293	+	})
	294	+	require.NoError(t, err)
	295	+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { r.Host = webterm.Address; handler.ServeHTTP(w, r) }))
	296	+	defer server.Close()
	297	+	headers := http.Header{"Origin": {webterm.Origin}, "Cookie": {"vev-web-session=" + token}}
	298	+	connection, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(server.URL, "http")+"/ws", &websocket.DialOptions{HTTPHeader: headers})
	299	+	require.NoError(t, err)
	300	+	defer connection.CloseNow()
	301	+	connection.SetReadLimit(64 << 20)
	302	+	awaitLogicalStream(t, fixture.streams)
	303	+	readCtx, stopRead := context.WithTimeout(ctx, brokerTestWait)
	304	+	defer stopRead()
	305	+	// This proves authenticated browser transport, real broker logical-stream
	306	+	// admission, and live browser frames. It does not prove post-Hello session
	307	+	// output: webterm may publish its transactional full-screen snapshot before
	308	+	// the fixture marker and does not guarantee another frame containing it.
	309	+	_, frame, err := connection.Read(readCtx)
	310	+	require.NoError(t, err)
	311	+	require.NotEmpty(t, frame, "browser emitted no frame after broker logical-stream admission")
	312	+	require.NoError(t, connection.CloseNow())
	313	+	waited := make(chan struct{})
	314	+	go func() {
	315	+	handler.Wait()
	316	+	close(waited)
	317	+	}()
	318	+	select {
	319	+	case <-waited:
	320	+	case <-time.After(brokerTestWait):
	321	+	t.Fatal("browser handler did not stop")
	322	+	}
	323	+	}

M internal/app/broker_offline_local.go +4 -3


@@ -25,9 +25,10 @@ import (
25	25		// it derives the probe from the sandbox configuration and never touches the
26	26		// production runtime or state directories.
27	27		//
28		-	// This is a preparation slice: nothing here is activated by production
29		-	// composition, and the observation-disabled offline authority composition is
30		-	// unchanged.
	28	+	// The hidden offline sandbox selects this producer whenever its strict config
	29	+	// provisions a local binding. Ordinary production composition does not select
	30	+	// it. The sandbox owns no remote membership, and probing remains no-spawn and
	31	+	// attachment-free even though it runs on the registry's observation cadence.
31	32
32	33		// localCarrierDialer establishes one raw framed daemonmux carriage for the
33	34		// opaque route address the broker-owned local binding published. It is the same


@@ -25,9 +25,10 @@ import (
25		// it derives the probe from the sandbox configuration and never touches the	25		// it derives the probe from the sandbox configuration and never touches the
26		// production runtime or state directories.	26		// production runtime or state directories.
27		//	27		//
28	-	// This is a preparation slice: nothing here is activated by production	28	+	// The hidden offline sandbox selects this producer whenever its strict config
29	-	// composition, and the observation-disabled offline authority composition is	29	+	// provisions a local binding. Ordinary production composition does not select
30	-	// unchanged.	30	+	// it. The sandbox owns no remote membership, and probing remains no-spawn and
			31	+	// attachment-free even though it runs on the registry's observation cadence.
31			32
32		// localCarrierDialer establishes one raw framed daemonmux carriage for the	33		// localCarrierDialer establishes one raw framed daemonmux carriage for the
33		// opaque route address the broker-owned local binding published. It is the same	34		// opaque route address the broker-owned local binding published. It is the same

M internal/app/broker_offline_local_test.go +22 -0


@@ -200,6 +200,28 @@ func TestBrokerLocalObservationComposition(t *testing.T) {
200	200		})
201	201		}
202	202
	203	+	func TestOfflineRegistryConfigSelectsLocalObservationOnlyWhenProvisioned(t *testing.T) {
	204	+	localConfig, binding := testLocalSandboxConfig(t)
	205	+	selected, err := offlineRegistryConfig(localConfig, nil)
	206	+	require.NoError(t, err)
	207	+	require.False(t, selected.ObservationDisabled)
	208	+	require.NotNil(t, selected.Local)
	209	+	require.Equal(t, binding.DisplayOrigin, selected.Local.DisplayOrigin)
	210	+	require.Equal(t, binding.Policy, selected.Local.Policy)
	211	+
	212	+	root := filepath.Join(shortTempDir(t, "vevl"), "sandbox")
	213	+	require.NoError(t, os.MkdirAll(root, 0o700))
	214	+	writeSandboxConfig(t, root, sandboxRegistrationDocument("/tmp/vev-local-test/mux.sock"))
	215	+	layout, err := brokerconfig.ResolveLayout(root, nil)
	216	+	require.NoError(t, err)
	217	+	remoteOnly, err := brokerconfig.Load(layout)
	218	+	require.NoError(t, err)
	219	+	disabled, err := offlineRegistryConfig(remoteOnly, nil)
	220	+	require.NoError(t, err)
	221	+	require.True(t, disabled.ObservationDisabled)
	222	+	require.Nil(t, disabled.Local)
	223	+	}
	224	+
203	225		// TestBrokerLocalObservationRegistryPublishesLocalFirst proves the composition
204	226		// end to end: the registry observes the local daemon through the broker-owned
205	227		// probe and publishes it at index zero, and the durable store persists no local


@@ -200,6 +200,28 @@ func TestBrokerLocalObservationComposition(t *testing.T) {
200	})	200		})
201	}	201		}
202		202
		203	+	func TestOfflineRegistryConfigSelectsLocalObservationOnlyWhenProvisioned(t *testing.T) {
		204	+	localConfig, binding := testLocalSandboxConfig(t)
		205	+	selected, err := offlineRegistryConfig(localConfig, nil)
		206	+	require.NoError(t, err)
		207	+	require.False(t, selected.ObservationDisabled)
		208	+	require.NotNil(t, selected.Local)
		209	+	require.Equal(t, binding.DisplayOrigin, selected.Local.DisplayOrigin)
		210	+	require.Equal(t, binding.Policy, selected.Local.Policy)
		211	+
		212	+	root := filepath.Join(shortTempDir(t, "vevl"), "sandbox")
		213	+	require.NoError(t, os.MkdirAll(root, 0o700))
		214	+	writeSandboxConfig(t, root, sandboxRegistrationDocument("/tmp/vev-local-test/mux.sock"))
		215	+	layout, err := brokerconfig.ResolveLayout(root, nil)
		216	+	require.NoError(t, err)
		217	+	remoteOnly, err := brokerconfig.Load(layout)
		218	+	require.NoError(t, err)
		219	+	disabled, err := offlineRegistryConfig(remoteOnly, nil)
		220	+	require.NoError(t, err)
		221	+	require.True(t, disabled.ObservationDisabled)
		222	+	require.Nil(t, disabled.Local)
		223	+	}
		224	+
203	// TestBrokerLocalObservationRegistryPublishesLocalFirst proves the composition	225		// TestBrokerLocalObservationRegistryPublishesLocalFirst proves the composition
204	// end to end: the registry observes the local daemon through the broker-owned	226		// end to end: the registry observes the local daemon through the broker-owned
205	// probe and publishes it at index zero, and the durable store persists no local	227		// probe and publishes it at index zero, and the durable store persists no local

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


@@ -222,6 +222,12 @@ var connectivityMatrix = []connectivityEntry{
222	222		Owner: connectivityLocalOnly,
223	223		Notes: "Foreground sandbox process over its own private root: no production runtime/state, no daemon dial, no RemoteHostStore, and no session persistence. It composes the broker under a temporary offline config and does not become a client façade or an ordinary command.",
224	224		},
	225	+	{
	226	+	Kind: "kindBrokerClient",
	227	+	Summary: "hidden _broker-client autonomous offline client harness",
	228	+	Owner: connectivityLocalOnly,
	229	+	Notes: "Explicit offline-root composition only: real broker IPC and logical streams exercise terminal and UI harnesses without changing any ordinary command or production path.",
	230	+	},
225	231		{
226	232		Kind: "kindBrokerLauncher",
227	233		Summary: "hidden _broker-launcher detached offline broker launcher",


@@ -222,6 +222,12 @@ var connectivityMatrix = []connectivityEntry{
222	Owner: connectivityLocalOnly,	222		Owner: connectivityLocalOnly,
223	Notes: "Foreground sandbox process over its own private root: no production runtime/state, no daemon dial, no RemoteHostStore, and no session persistence. It composes the broker under a temporary offline config and does not become a client façade or an ordinary command.",	223		Notes: "Foreground sandbox process over its own private root: no production runtime/state, no daemon dial, no RemoteHostStore, and no session persistence. It composes the broker under a temporary offline config and does not become a client façade or an ordinary command.",
224	},	224		},
		225	+	{
		226	+	Kind: "kindBrokerClient",
		227	+	Summary: "hidden _broker-client autonomous offline client harness",
		228	+	Owner: connectivityLocalOnly,
		229	+	Notes: "Explicit offline-root composition only: real broker IPC and logical streams exercise terminal and UI harnesses without changing any ordinary command or production path.",
		230	+	},
225	{	231		{
226	Kind: "kindBrokerLauncher",	232		Kind: "kindBrokerLauncher",
227	Summary: "hidden _broker-launcher detached offline broker launcher",	233		Summary: "hidden _broker-launcher detached offline broker launcher",

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


@@ -75,6 +75,7 @@ const (
75	75		kindWebRenew
76	76		kindWebServe
77	77		kindBrokerServe
	78	+	kindBrokerClient
78	79		kindBrokerLauncher
79	80		kindBrokerStatus
80	81		kindBrokerMuxStdio

@@ -99,6 +100,7 @@ type command struct {
99	100		killDaemon bool
100	101		cmd cmdInvocation
101	102		brokerServe brokerServeOptions
	103	+	brokerClient brokerClientOptions
102	104		brokerLauncher brokerLauncherOptions
103	105		brokerStatus brokerStatusOptions
104	106		brokerMux brokerMuxOptions

@@ -242,6 +244,8 @@ parsedUIFlags:
242	244		return command{kind: kindDaemonLauncher}, nil
243	245		case brokerServeCommand:
244	246		return parseBrokerServeArgs(args[1:])
	247	+	case brokerClientCommand:
	248	+	return parseBrokerClientArgs(args[1:])
245	249		case brokerLauncherCommand:
246	250		return parseBrokerLauncherArgs(args[1:])
247	251		case brokerStatusCommand:

@@ -378,6 +382,8 @@ func dispatch(ctx context.Context, cmd command) error {
378	382		return runDaemonLauncher()
379	383		case kindBrokerServe:
380	384		return runBrokerServeCommand(ctx, cmd.brokerServe)
	385	+	case kindBrokerClient:
	386	+	return runBrokerClientCommand(ctx, cmd.brokerClient)
381	387		case kindBrokerLauncher:
382	388		return runBrokerLauncherCommand(ctx, cmd.brokerLauncher)
383	389		case kindBrokerStatus:


@@ -75,6 +75,7 @@ const (
75	kindWebRenew	75		kindWebRenew
76	kindWebServe	76		kindWebServe
77	kindBrokerServe	77		kindBrokerServe
		78	+	kindBrokerClient
78	kindBrokerLauncher	79		kindBrokerLauncher
79	kindBrokerStatus	80		kindBrokerStatus
80	kindBrokerMuxStdio	81		kindBrokerMuxStdio

@@ -99,6 +100,7 @@ type command struct {
99	killDaemon bool	100		killDaemon bool
100	cmd cmdInvocation	101		cmd cmdInvocation
101	brokerServe brokerServeOptions	102		brokerServe brokerServeOptions
		103	+	brokerClient brokerClientOptions
102	brokerLauncher brokerLauncherOptions	104		brokerLauncher brokerLauncherOptions
103	brokerStatus brokerStatusOptions	105		brokerStatus brokerStatusOptions
104	brokerMux brokerMuxOptions	106		brokerMux brokerMuxOptions

@@ -242,6 +244,8 @@ parsedUIFlags:
242	return command{kind: kindDaemonLauncher}, nil	244		return command{kind: kindDaemonLauncher}, nil
243	case brokerServeCommand:	245		case brokerServeCommand:
244	return parseBrokerServeArgs(args[1:])	246		return parseBrokerServeArgs(args[1:])
		247	+	case brokerClientCommand:
		248	+	return parseBrokerClientArgs(args[1:])
245	case brokerLauncherCommand:	249		case brokerLauncherCommand:
246	return parseBrokerLauncherArgs(args[1:])	250		return parseBrokerLauncherArgs(args[1:])
247	case brokerStatusCommand:	251		case brokerStatusCommand:

@@ -378,6 +382,8 @@ func dispatch(ctx context.Context, cmd command) error {
378	return runDaemonLauncher()	382		return runDaemonLauncher()
379	case kindBrokerServe:	383		case kindBrokerServe:
380	return runBrokerServeCommand(ctx, cmd.brokerServe)	384		return runBrokerServeCommand(ctx, cmd.brokerServe)
		385	+	case kindBrokerClient:
		386	+	return runBrokerClientCommand(ctx, cmd.brokerClient)
381	case kindBrokerLauncher:	387		case kindBrokerLauncher:
382	return runBrokerLauncherCommand(ctx, cmd.brokerLauncher)	388		return runBrokerLauncherCommand(ctx, cmd.brokerLauncher)
383	case kindBrokerStatus:	389		case kindBrokerStatus: