1// Package app wires vev's runtime together via dependency injection. It is
2// the single place where usecases (client, daemon) are connected to concrete
3// adapters (ipc transport/listener, terminal); usecases themselves never
4// import adapters.
5package app
6
7import (
8 "context"
9 "crypto/rand"
10 "errors"
11 "fmt"
12 "io"
13 "log/slog"
14 "net"
15 "net/http"
16 _ "net/http/pprof"
17 "os"
18 "os/signal"
19 "path/filepath"
20 "strconv"
21 "strings"
22 "sync"
23 "syscall"
24 "text/tabwriter"
25 "time"
26
27 "github.com/bnema/vev/internal/adapters/clock"
28 "github.com/bnema/vev/internal/adapters/config"
29 "github.com/bnema/vev/internal/adapters/daemonidentity"
30 "github.com/bnema/vev/internal/adapters/daemonmux"
31 "github.com/bnema/vev/internal/adapters/ipc"
32 "github.com/bnema/vev/internal/adapters/lifecycle"
33 "github.com/bnema/vev/internal/adapters/noticefile"
34 "github.com/bnema/vev/internal/adapters/observability"
35 "github.com/bnema/vev/internal/adapters/pty"
36 "github.com/bnema/vev/internal/adapters/sessionwire"
37 "github.com/bnema/vev/internal/adapters/shellcmd"
38 snapshotadapter "github.com/bnema/vev/internal/adapters/snapshot"
39 "github.com/bnema/vev/internal/domain"
40 "github.com/bnema/vev/internal/logging"
41 "github.com/bnema/vev/internal/persist"
42 "github.com/bnema/vev/internal/platform"
43 "github.com/bnema/vev/internal/ports"
44 "github.com/bnema/vev/internal/protocol"
45 "github.com/bnema/vev/internal/protocol/wire"
46 "github.com/bnema/vev/internal/usecase/client"
47 "github.com/bnema/vev/internal/usecase/daemon"
48 "github.com/bnema/vev/internal/usecase/recovery"
49 "github.com/bnema/vev/pkg/safedir"
50)
51
52// cmdKind identifies which sub-command the CLI parsed.
53type cmdKind int
54
55const (
56 kindAttach cmdKind = iota // ephemeral/new/attach — distinguished by intent
57 kindList
58 kindHost
59 kindKill
60 kindCmd
61 kindEnv
62 kindDaemon
63 kindDaemonLauncher
64 kindUIDriver
65 kindUIRemoteCleanup
66 kindDaemonStop
67 kindWebDaemon
68 kindWebRenew
69 kindWebServe
70 kindBrokerMuxStdio
71 kindBrokerMuxQUICBootstrap
72 kindBrokerMuxQUICProxy
73 kindProductionBrokerServe
74 kindProductionBrokerLauncher
75 kindBrokerReady
76 kindHelp
77 kindVersion
78)
79
80// command is the parsed CLI invocation: what to do, plus the attach intent
81// and session name where relevant.
82type command struct {
83 kind cmdKind
84 intent uint8
85 name string
86 remoteTarget string
87 listHost string
88 listAll bool
89 hostAction string
90 hostTarget string
91 // hostTransport is the `host add` carriage: hostTransportQUIC or hostTransportSSH.
92 hostTransport string
93 killAll bool
94 killSessions bool
95 cmd cmdInvocation
96 env envInvocation
97 brokerServe brokerServeOptions
98 brokerMux brokerMuxOptions
99 brokerReady brokerReadyOptions
100 uiDriver uiDriverOptions
101 uiObserve bool
102 uiControl bool
103 uiSocket string
104 web webOptions
105}
106
107// usageError is a user-facing argument error; the app prints it (with usage)
108// rather than a stack of wrapped internals.
109type usageError struct{ msg string }
110
111func (e *usageError) Error() string { return "vev: " + e.msg }
112
113func usagef(format string, args ...any) error {
114 return &usageError{msg: fmt.Sprintf(format, args...)}
115}
116
117const usageText = `vev — a terminal multiplexer
118
119usage:
120 vev attach to (or create) an ephemeral session
121 vev new <name> create and attach to a named session
122 vev attach <name> attach to an existing session (alias: a)
123 vev attach user@host[:session]
124 attach through SSH to a remote vev daemon
125 vev ls list local sessions
126 vev ls <host> list sessions on a known remote host
127 vev ls --all list local and remote sessions
128 vev host add [--transport quic|ssh] <host>
129 add a pinned remote host (default transport: quic)
130 vev host rm <host> remove a pinned remote host
131 vev host list list known remote hosts
132 vev kill <name> kill a session
133 vev kill --sessions kill all sessions (vev keeps running)
134 vev kill --all stop everything: every vev window, the broker, and the
135 daemon (named sessions come back on the next start)
136 vev cmd <command> run a control command (vev cmd --help)
137 vev env <fish|sh> print shell code that refreshes desktop variables
138 vev --ui-observe expose passive observation for this interactive client
139 (optional: --ui-socket PATH)
140 vev --ui-control expose observation and input control for this client
141 (optional: --ui-socket PATH)
142 vev --web-daemon start the private web terminal or print its current link
143 vev --web-renew-token revoke web access and print a fresh link
144 Web options: --web-listen IP:PORT --web-origin http(s)://HOST[:PORT]
145 vev --help show this help
146 vev --version show version`
147
148// Build metadata. Defaults describe a plain `go build`; releases overwrite
149// them via -ldflags "-X github.com/bnema/vev/internal/app.version=..." (and
150// .commit / .date) — see .goreleaser.yaml.
151var (
152 version = "0.1.0-dev"
153 commit = "none"
154 date = "unknown"
155
156 openCatalogue = persist.OpenOrCreate
157 listenDaemon = func(dir string, observer ports.SerializedRuntimeObserver) (wire.Listener, error) {
158 return ipc.Listen(dir, ipc.WithRuntimeObserver(observer))
159 }
160)
161
162// versionLine renders the --version output.
163func versionLine() string {
164 return fmt.Sprintf("vev %s (commit %s, built %s)", version, commit, date)
165}
166
167// Run is the entry point invoked by main. It parses args into a command and
168// dispatches it.
169func Run(args []string) error {
170 if err := platform.ActivateDevelopmentEnvironment(); err != nil {
171 return err
172 }
173 cmd, err := parseArgs(args)
174 if err != nil {
175 return err
176 }
177 return dispatch(context.Background(), cmd)
178}
179
180// parseArgs turns the raw argv tail into a command. It is deliberately a
181// pure function so the full dispatch table can be unit-tested without any
182// I/O.
183func parseArgs(args []string) (command, error) {
184 if len(args) == 0 {
185 return command{kind: kindAttach, intent: protocol.IntentEphemeral}, nil
186 }
187 if args[0] == "--ui-driver" {
188 options, err := parseUIDriverArgs(args[1:])
189 if err != nil {
190 return command{}, err
191 }
192 return command{kind: kindUIDriver, uiDriver: options}, nil
193 }
194
195 var uiObserve, uiControl bool
196 var uiSocket string
197 for len(args) > 0 {
198 switch args[0] {
199 case "--ui-observe":
200 uiObserve = true
201 case "--ui-control":
202 uiControl = true
203 case "--ui-socket":
204 if len(args) < 2 || args[1] == "" {
205 return command{}, usagef("`--ui-socket` requires a path")
206 }
207 uiSocket = args[1]
208 args = args[2:]
209 continue
210 default:
211 goto parsedUIFlags
212 }
213 args = args[1:]
214 }
215parsedUIFlags:
216 interactive, err := parseInteractiveUIFlags(nil, uiObserve, uiControl, uiSocket)
217 if err != nil {
218 return command{}, err
219 }
220 uiObserve, uiControl, uiSocket = interactive.observe, interactive.control, interactive.socket
221 if len(args) == 0 {
222 return command{kind: kindAttach, intent: protocol.IntentEphemeral, uiObserve: uiObserve, uiControl: uiControl, uiSocket: uiSocket}, nil
223 }
224 if (uiObserve || uiControl || uiSocket != "") && args[0] != "new" && args[0] != "attach" && args[0] != "a" {
225 return command{}, usagef("UI flags are only valid for an attach command")
226 }
227
228 switch args[0] {
229 case "--web-daemon", "--web-serve":
230 return parseWebArgs(args)
231 case "--web-renew-token":
232 if len(args) != 1 {
233 return command{}, usagef("`%s` does not accept arguments", args[0])
234 }
235 return command{kind: kindWebRenew}, nil
236 case "--daemon":
237 return command{kind: kindDaemon}, nil
238 case "--daemon-launcher":
239 return command{kind: kindDaemonLauncher}, nil
240 case productionBrokerServeCommand:
241 return parseProductionBrokerServeArgs(args[1:])
242 case productionBrokerLauncherCommand:
243 return parseProductionBrokerLauncherArgs(args[1:])
244 case brokerReadyCommand:
245 options, err := parseBrokerReadyArgs(args[1:])
246 return command{kind: kindBrokerReady, brokerReady: options}, err
247 case brokerMuxStdioCommand:
248 return parseBrokerMuxArgs(brokerMuxStdioCommand, kindBrokerMuxStdio, args[1:])
249 case brokerMuxQUICBootstrapCommand:
250 return parseBrokerMuxArgs(brokerMuxQUICBootstrapCommand, kindBrokerMuxQUICBootstrap, args[1:])
251 case brokerMuxQUICProxyCommand:
252 return parseBrokerMuxArgs(brokerMuxQUICProxyCommand, kindBrokerMuxQUICProxy, args[1:])
253 case uiRemoteCleanupCommand:
254 if len(args) != 1 {
255 return command{}, usagef("`%s` does not accept arguments", uiRemoteCleanupCommand)
256 }
257 return command{kind: kindUIRemoteCleanup}, nil
258 case daemonStopCommand:
259 if len(args) != 1 {
260 return command{}, usagef("`%s` does not accept arguments", daemonStopCommand)
261 }
262 return command{kind: kindDaemonStop}, nil
263 case "new":
264 if len(args) < 2 || args[1] == "" {
265 return command{}, usagef("`new` requires a session name")
266 }
267 name := args[1]
268 var remoteTarget string
269 optionStart := 2
270 if len(args) > 2 && !strings.HasPrefix(args[2], "--") {
271 remoteTarget = args[2]
272 if err := domain.ValidateRemoteHostTarget(remoteTarget); err != nil {
273 return command{}, err
274 }
275 optionStart = 3
276 }
277 interactive, err := parseInteractiveUIFlags(args[optionStart:], uiObserve, uiControl, uiSocket)
278 if err != nil {
279 return command{}, err
280 }
281 if err := domain.ValidateSessionName(name); err != nil {
282 return command{}, err
283 }
284 return command{kind: kindAttach, intent: protocol.IntentNew, name: name, remoteTarget: remoteTarget, uiObserve: interactive.observe, uiControl: interactive.control, uiSocket: interactive.socket}, nil
285 case "attach", "a":
286 if len(args) < 2 || args[1] == "" {
287 return command{}, usagef("`attach` requires a session name")
288 }
289 interactive, err := parseInteractiveUIFlags(args[2:], uiObserve, uiControl, uiSocket)
290 if err != nil {
291 return command{}, err
292 }
293 cmd := command{kind: kindAttach, intent: protocol.IntentAttach, name: args[1]}
294 if target, session, ok := parseRemoteAttachTarget(args[1]); ok {
295 if err := domain.ValidateRemoteHostTarget(target); err != nil {
296 return command{}, err
297 }
298 if session != "" {
299 if err := domain.ValidateSessionName(session); err != nil {
300 return command{}, err
301 }
302 }
303 cmd.remoteTarget = target
304 cmd.name = session
305 if session == "" {
306 cmd.intent = protocol.IntentEphemeral
307 }
308 }
309 cmd.uiObserve = interactive.observe
310 cmd.uiControl = interactive.control
311 cmd.uiSocket = interactive.socket
312 return cmd, nil
313 case "ls", "list":
314 return parseListArgs(args[1:])
315 case "host":
316 return parseHostArgs(args[1:])
317 case "env":
318 invocation, err := parseEnvArgs(args[1:])
319 if err != nil {
320 return command{}, err
321 }
322 return command{kind: kindEnv, env: invocation}, nil
323 case "cmd":
324 invocation, err := parseCmdArgs(args[1:])
325 if err != nil {
326 return command{}, err
327 }
328 return command{kind: kindCmd, cmd: invocation}, nil
329 case "kill":
330 if len(args) < 2 || args[1] == "" {
331 return command{}, usagef("`kill` requires a session name, --sessions, or --all")
332 }
333 if args[1] == "--" {
334 if len(args) != 3 || args[2] == "" {
335 return command{}, usagef("`kill --` requires a session name")
336 }
337 return command{kind: kindKill, name: args[2]}, nil
338 }
339 if len(args) > 2 {
340 return command{}, usagef("`kill` accepts exactly one session name, --sessions, or --all")
341 }
342 switch args[1] {
343 case "--sessions":
344 return command{kind: kindKill, killSessions: true}, nil
345 case "--all":
346 return command{kind: kindKill, killAll: true}, nil
347 default:
348 if strings.HasPrefix(args[1], "-") {
349 return command{}, usagef("unknown flag %q for `kill`; use `kill -- NAME` for a dashed session name", args[1])
350 }
351 return command{kind: kindKill, name: args[1]}, nil
352 }
353 case "-h", "--help", "help":
354 return command{kind: kindHelp}, nil
355 case "--version", "version":
356 return command{kind: kindVersion}, nil
357 default:
358 return command{}, usagef("unknown command %q", args[0])
359 }
360}
361
362// dispatch runs the parsed command.
363func dispatch(ctx context.Context, cmd command) error {
364 switch cmd.kind {
365 case kindHelp:
366 fmt.Println(usageText)
367 return nil
368 case kindVersion:
369 fmt.Println(versionLine())
370 return nil
371 case kindWebRenew:
372 return renewWebToken(ctx)
373 case kindWebDaemon:
374 return launchWebDaemon(ctx, cmd.web)
375 case kindWebServe:
376 return runWebDaemon(ctx, cmd.web)
377 case kindDaemon:
378 return runDaemon()
379 case kindDaemonLauncher:
380 return runDaemonLauncher()
381 case kindBrokerMuxStdio:
382 return runBrokerMuxStdioCommand(ctx, cmd.brokerMux)
383 case kindBrokerMuxQUICBootstrap:
384 return runBrokerMuxQUICBootstrapCommand(ctx, cmd.brokerMux)
385 case kindBrokerMuxQUICProxy:
386 return runBrokerMuxQUICProxyCommand(ctx, cmd.brokerMux)
387 case kindProductionBrokerServe:
388 return runBrokerServeCommand(ctx, cmd.brokerServe)
389 case kindProductionBrokerLauncher:
390 return runProductionBrokerLauncherCommand(ctx)
391 case kindBrokerReady:
392 return runBrokerReady(ctx, cmd.brokerReady, os.Stdout)
393 case kindDaemonStop:
394 return requestDaemonStop(ctx)
395 case kindUIRemoteCleanup:
396 return runUIRemoteCleanup(ctx)
397 case kindUIDriver:
398 return runUIDriver(ctx, cmd.uiDriver)
399 case kindList:
400 return runList(ctx, cmd)
401 case kindHost:
402 return runHostCommand(ctx, cmd, defaultRemoteHostDeps())
403 case kindKill:
404 if cmd.killAll {
405 return runKillAll(ctx, os.Stdout)
406 }
407 return runKill(ctx, cmd.name, cmd.killSessions)
408 case kindCmd:
409 return runCmd(ctx, cmd.cmd)
410 case kindEnv:
411 return runEnv(ctx, cmd.env)
412 case kindAttach:
413 return runAttachWithOptions(ctx, cmd.intent, cmd.name, cmd.remoteTarget, interactiveUIOptions{observe: cmd.uiObserve, control: cmd.uiControl, socket: cmd.uiSocket})
414 default:
415 return usagef("unhandled command")
416 }
417}
418
419func parseListArgs(args []string) (command, error) {
420 if len(args) == 0 {
421 return command{kind: kindList}, nil
422 }
423 if len(args) > 1 {
424 return command{}, usagef("`ls` accepts at most one host or --all")
425 }
426 switch args[0] {
427 case "--all":
428 return command{kind: kindList, listAll: true}, nil
429 default:
430 if strings.HasPrefix(args[0], "-") {
431 return command{}, usagef("unknown flag %q for `ls`", args[0])
432 }
433 if err := domain.ValidateRemoteHostTarget(args[0]); err != nil {
434 return command{}, err
435 }
436 return command{kind: kindList, listHost: args[0]}, nil
437 }
438}
439
440func parseHostArgs(args []string) (command, error) {
441 if len(args) == 0 {
442 return command{}, usagef("`host` requires add, rm, or list")
443 }
444 switch args[0] {
445 case hostActionAdd:
446 return parseHostAddArgs(args[1:])
447 case hostActionRm:
448 if len(args) < 2 || args[1] == "" {
449 return command{}, usagef("`host rm` requires a host target")
450 }
451 if strings.HasPrefix(args[1], "-") {
452 return command{}, usagef("unknown flag %q for `host rm`", args[1])
453 }
454 if len(args) > 2 {
455 return command{}, usagef("`host rm` accepts exactly one host target")
456 }
457 if err := domain.ValidateRemoteHostTarget(args[1]); err != nil {
458 return command{}, err
459 }
460 return command{kind: kindHost, hostAction: hostActionRm, hostTarget: args[1]}, nil
461 case hostActionList:
462 if len(args) > 1 {
463 if strings.HasPrefix(args[1], "-") {
464 return command{}, usagef("unknown flag %q for `host list`", args[1])
465 }
466 return command{}, usagef("`host list` accepts no arguments")
467 }
468 return command{kind: kindHost, hostAction: hostActionList}, nil
469 default:
470 return command{}, usagef("unknown host action %q", args[0])
471 }
472}
473
474// parseHostAddArgs parses `host add [--transport quic|ssh] <host>`.
475func parseHostAddArgs(args []string) (command, error) {
476 transport, target := "", ""
477 for i := 0; i < len(args); i++ {
478 arg := args[i]
479 switch {
480 case arg == "--transport" || strings.HasPrefix(arg, "--transport="):
481 if transport != "" {
482 return command{}, usagef("`host add` accepts --transport once")
483 }
484 value, hasValue := strings.CutPrefix(arg, "--transport=")
485 if !hasValue {
486 if i+1 >= len(args) {
487 return command{}, usagef("`--transport` requires a value (%s or %s)", hostTransportQUIC, hostTransportSSH)
488 }
489 i++
490 value = args[i]
491 }
492 switch value {
493 case hostTransportQUIC, hostTransportSSH:
494 transport = value
495 default:
496 return command{}, usagef("invalid transport %q (want %q or %q)", value, hostTransportQUIC, hostTransportSSH)
497 }
498 case strings.HasPrefix(arg, "-"):
499 return command{}, usagef("unknown flag %q for `host add`", arg)
500 case target != "":
501 return command{}, usagef("`host add` accepts exactly one host target")
502 default:
503 target = arg
504 }
505 }
506 if target == "" {
507 return command{}, usagef("`host add` requires a host target")
508 }
509 if err := domain.ValidateRemoteHostTarget(target); err != nil {
510 return command{}, err
511 }
512 if transport == "" {
513 transport = hostTransportQUIC
514 }
515 return command{kind: kindHost, hostAction: hostActionAdd, hostTarget: target, hostTransport: transport}, nil
516}
517
518// performanceTrace creates one serialized timestamp owner for this process.
519// An empty trace environment leaves all production behavior and wire bytes
520// unchanged.
521// Composition-root factory seams keep observer propagation testable without
522// opening real transports.
523var connectBroker = connectProductionBroker
524var connectDaemonStopBroker = connectExistingBroker
525
526var newPerformanceTrace = performanceTrace
527
528func performanceTrace(clk ports.Clock) (ports.SerializedRuntimeObserver, io.Closer, error) {
529 return performanceTraceWithFactories(clk, observability.NewJSONL, ports.NewRuntimeCorrelationObserver)
530}
531
532// performanceTraceWithFactories keeps setup rollback behavior directly
533// testable without requiring an operating-system file close failure.
534func performanceTraceWithFactories(
535 clk ports.Clock,
536 newSink func(string, ports.Clock, string) (ports.RuntimeObserver, io.Closer, error),
537 newCorrelation func(ports.RuntimeObserver, ports.RuntimeCorrelationInputs) (ports.RuntimeObserver, error),
538) (ports.SerializedRuntimeObserver, io.Closer, error) {
539 path, processID := os.Getenv("VEV_PERF_TRACE"), os.Getenv("VEV_PERF_PROCESS_ID")
540 if path == "" {
541 return nil, nil, nil
542 }
543 observer, closer, err := newSink(path, clk, processID)
544 if err != nil {
545 return nil, nil, err
546 }
547 // The harness supplies these manifest fields to every launched role. Keep a
548 // valid standalone trace for operators that set only the original trace
549 // variables, while ensuring harness traces match their process mapping.
550 inputs := ports.RuntimeCorrelationInputs{Scenario: os.Getenv("VEV_PERF_SCENARIO"), Run: 1}
551 if inputs.Scenario == "" {
552 inputs.Scenario = "runtime"
553 }
554 if rawRun := os.Getenv("VEV_PERF_RUN"); rawRun != "" {
555 inputs.Run, err = strconv.ParseUint(rawRun, 10, 64)
556 if err != nil {
557 setupErr := fmt.Errorf("invalid VEV_PERF_RUN %q: %w", rawRun, err)
558 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
559 }
560 if inputs.Run == 0 {
561 setupErr := fmt.Errorf("invalid VEV_PERF_RUN %q", rawRun)
562 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
563 }
564 }
565 observer, err = newCorrelation(observer, inputs)
566 if err != nil {
567 setupErr := fmt.Errorf("configure runtime trace correlation: %w", err)
568 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
569 }
570 reporter := observability.NewSerialized(observer, runtimeTraceQueueDepth)
571 return reporter, &runtimeTraceCloser{reporter: reporter, closer: closer}, nil
572}
573
574func closeTraceAfterSetupFailure(closer io.Closer, setupErr error) error {
575 if closer == nil {
576 return setupErr
577 }
578 if closeErr := closer.Close(); closeErr != nil {
579 return errors.Join(setupErr, fmt.Errorf("close performance trace after setup failure: %w", closeErr))
580 }
581 return setupErr
582}
583
584// runtimeTraceQueueDepth bounds all process-local trace producer handoffs.
585// A full queue emits the serialized diagnostic gap rather than delaying a
586// terminal, transport, or ACK progress path.
587const runtimeTraceQueueDepth = 256
588
589// runtimeTraceCloser owns the two-stage trace shutdown: drain the reporter
590// before closing the concrete timestamp/file owner. sync.Once makes every
591// deferred and explicit close path safe without closing shared workers twice.
592type runtimeTraceCloser struct {
593 reporter ports.SerializedRuntimeObserver
594 closer io.Closer
595 once sync.Once
596 err error
597}
598
599func (c *runtimeTraceCloser) Close() error {
600 if c == nil {
601 return nil
602 }
603 c.once.Do(func() {
604 if c.reporter != nil {
605 c.reporter.Flush()
606 c.reporter.Close()
607 }
608 if c.closer != nil {
609 c.err = c.closer.Close()
610 }
611 })
612 return c.err
613}
614
615func configureLogging(component logging.Component, rotateAtRuntime bool) (*slog.Logger, io.Closer, error) {
616 return logging.Setup(logging.Config{
617 Dir: platform.StateDir(),
618 Component: component,
619 Level: logging.EnvLevel(),
620 MaxBytes: logging.DefaultMaxBytes,
621 RotateAtRuntime: rotateAtRuntime,
622 })
623}
624
625func logConfigWarnings(log *slog.Logger, warnings []domain.Warning) {
626 for _, warning := range warnings {
627 if warning.Line > 0 {
628 log.Warn("config warning", "line", warning.Line, "msg", warning.Msg)
629 continue
630 }
631 log.Warn("config warning", "msg", warning.Msg)
632 }
633}
634
635func loadConfigOrDefaults(log *slog.Logger, path string) domain.Config {
636 cfg, warnings, err := config.Load(path)
637 if err != nil {
638 if log != nil {
639 log.Warn("loading config failed; using defaults", "path", path, "err", err)
640 }
641 cfg = domain.Defaults()
642 }
643 if log != nil {
644 logConfigWarnings(log, warnings)
645 }
646 return cfg
647}
648
649func snapshotDir() string {
650 return filepath.Join(platform.StateDir(), "snapshots")
651}
652
653func developmentTempDirOption(ensurePrivate func(string) error) (daemon.Option, error) {
654 dir, active := platform.DevelopmentEnvironmentTempDir()
655 if !active {
656 return nil, nil
657 }
658 if err := ensurePrivate(dir); err != nil {
659 return nil, fmt.Errorf("vev: secure development temp directory: %w", err)
660 }
661 return daemon.WithTempDir(dir), nil
662}
663
664func pprofAddrIsLoopback(addr string) bool {
665 host, _, err := net.SplitHostPort(addr)
666 if err != nil || host == "" {
667 return false
668 }
669 if strings.EqualFold(host, "localhost") {
670 return true
671 }
672 ip := net.ParseIP(host)
673 return ip != nil && ip.IsLoopback()
674}
675
676type ownedDaemonStart func(context.Context) error
677
678type lifecycleOwnership interface {
679 Release() error
680}
681
682type lifecycleStartupDeps struct {
683 ensurePrivate func(string) error
684 acquire func(context.Context, string, time.Duration) (lifecycleOwnership, error)
685 log *slog.Logger
686}
687
688const lifecycleAcquireRetry = 25 * time.Millisecond
689
690func lifecycleStartupDependencies(log *slog.Logger) lifecycleStartupDeps {
691 return lifecycleStartupDeps{
692 ensurePrivate: safedir.EnsurePrivate,
693 acquire: func(ctx context.Context, runtimeDir string, retry time.Duration) (lifecycleOwnership, error) {
694 return lifecycle.Acquire(ctx, runtimeDir, retry)
695 },
696 log: log,
697 }
698}
699
700func runWithLifecycleOwner(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart) error {
701 return runWithLifecycleOwnerDeps(ctx, runtimeDir, stateDir, start, lifecycleStartupDependencies(nil))
702}
703
704func runWithLifecycleOwnerDeps(ctx context.Context, runtimeDir, stateDir string, start ownedDaemonStart, deps lifecycleStartupDeps) (retErr error) {
705 log := deps.log
706 if log == nil {
707 log = slog.Default()
708 }
709 if err := deps.ensurePrivate(runtimeDir); err != nil {
710 return fmt.Errorf("vev: secure runtime directory: %w", err)
711 }
712 if err := deps.ensurePrivate(stateDir); err != nil {
713 return fmt.Errorf("vev: secure state directory: %w", err)
714 }
715 lockPath := lifecycle.Path(runtimeDir)
716 log.Info("lifecycle_owner_wait", "path", lockPath)
717 owner, err := deps.acquire(ctx, runtimeDir, lifecycleAcquireRetry)
718 if err != nil {
719 log.Warn("lifecycle_owner_wait_failed", "path", lockPath, "reason_code", "acquire-failed")
720 return fmt.Errorf("vev: acquire lifecycle ownership: %w", err)
721 }
722 log.Info("lifecycle_owner_acquired", "path", lockPath)
723 defer func() {
724 releaseErr := owner.Release()
725 if releaseErr != nil {
726 log.Error("lifecycle_owner_release_failed", "path", lockPath, "reason_code", "release-failed")
727 } else {
728 log.Info("lifecycle_owner_released", "path", lockPath)
729 }
730 retErr = errors.Join(retErr, releaseErr)
731 }()
732 if start == nil {
733 return errors.New("vev: nil owned daemon startup")
734 }
735 return start(ctx)
736}
737
738// runDaemon runs the daemon in the foreground (the hidden --daemon path,
739// entered by an auto-spawned child) while holding exclusive lifecycle
740// ownership through complete teardown.
741func runDaemon() (retErr error) {
742 ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
743 defer stop()
744 log, logCloser, err := configureLogging(logging.Daemon, true)
745 if err != nil {
746 return err
747 }
748 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
749 return runWithLifecycleOwnerDeps(ctx, ipc.SocketDir(), platform.StateDir(), func(ctx context.Context) error {
750 return runDaemonOwnedWithLogger(ctx, log)
751 }, lifecycleStartupDependencies(log))
752}
753
754func logCatalogueRecovery(log *slog.Logger, records []domain.CatalogueRecord, recoveryMode string) {
755 log.Info("catalogue_validated", "records", len(records), "recovery", recoveryMode)
756}
757
758func logStartupRecoveryCounts(log *slog.Logger, records []domain.CatalogueRecord, restoring int) {
759 healthy, fresh, broken := 0, 0, 0
760 for _, record := range records {
761 switch {
762 case record.DegradedReason != "":
763 broken++
764 case record.Committed == nil:
765 fresh++
766 default:
767 healthy++
768 }
769 }
770 log.Info("daemon_startup_complete", "healthy", healthy, "fresh", fresh, "restoring", restoring, "broken", broken)
771}
772
773func runDaemonOwned(ctx context.Context) (retErr error) {
774 log, logCloser, err := configureLogging(logging.Daemon, true)
775 if err != nil {
776 return err
777 }
778 defer func() { retErr = errors.Join(retErr, logCloser.Close()) }()
779 return runDaemonOwnedWithLogger(ctx, log)
780}
781
782func runDaemonOwnedWithLogger(ctx context.Context, log *slog.Logger) (retErr error) {
783 clk := clock.New()
784 observer, observerCloser, err := newPerformanceTrace(clk)
785 if err != nil {
786 return fmt.Errorf("vev: performance trace: %w", err)
787 }
788 if observerCloser != nil {
789 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
790 }
791 if addr := os.Getenv("VEV_PPROF_ADDR"); addr != "" {
792 if !pprofAddrIsLoopback(addr) {
793 log.Warn("pprof bound to non-loopback address; /debug/pprof is unauthenticated", "addr", addr)
794 }
795 go func() {
796 if err := http.ListenAndServe(addr, nil); err != nil && !errors.Is(err, http.ErrServerClosed) {
797 log.Error("pprof server exited", "err", err)
798 }
799 }()
800 log.Info("pprof enabled", "addr", addr)
801 }
802
803 var daemonOpts []daemon.Option
804 developmentTempOpt, err := developmentTempDirOption(safedir.EnsurePrivate)
805 if err != nil {
806 return err
807 }
808 if developmentTempOpt != nil {
809 daemonOpts = append(daemonOpts, developmentTempOpt)
810 }
811 if observer != nil {
812 daemonOpts = append(daemonOpts, daemon.WithRuntimeObserver(observer))
813 }
814 configPath := platform.ConfigPath()
815 cfg := loadConfigOrDefaults(log, configPath)
816 daemonOpts = append(daemonOpts, daemon.WithConfig(cfg))
817 daemonOpts = append(daemonOpts, daemon.WithBarScriptCommandRunner(shellcmd.New()))
818 daemonOpts = append(daemonOpts, daemon.WithProcessInspector(platform.NewProcessInspector()), daemon.WithDirOrHome(platform.DirOrHome))
819 snapshotRepository := snapshotadapter.NewRepositoryWithLogger(snapshotDir(), log)
820 daemonOpts = append(daemonOpts, daemon.WithSnapshotRepository(snapshotRepository))
821 noticeStore := noticefile.New(platform.StateDir())
822 daemonOpts = append(daemonOpts, daemon.WithNoticeStore(noticeStore))
823 stateDir := platform.StateDir()
824 storePath := persist.StorePath(stateDir)
825 opened, err := openCatalogue(stateDir)
826 if err != nil {
827 log.Error("catalogue_validation_failed", "path", storePath, "reason_code", "open-failed")
828 if errors.Is(err, persist.ErrCatalogueUnreadable) {
829 return unreadableCatalogueError(stateDir)
830 }
831 return fmt.Errorf("vev: open durable session state %s: %w", storePath, err)
832 }
833 recoveryMode := "current"
834 if opened.NewInstall {
835 recoveryMode = "new-install"
836 }
837 coordinator := recovery.NewCoordinator(opened.Catalogue, snapshotRepository, rand.Reader)
838 if opened.Migration.Performed {
839 recoveryMode = "catalogue-migrated"
840 log.Info("catalogue_migrated", "source_formats", opened.Migration.SourceFormats, "target_format", opened.Migration.TargetFormat, "records", opened.Migration.RecordCount, "backup", opened.Migration.BackupPath)
841 }
842 logCatalogueRecovery(log, opened.Records, recoveryMode)
843 daemonOpts = append(daemonOpts, daemon.WithRecoveryCoordinator(coordinator))
844 daemonOpts = append(daemonOpts, daemon.WithSnapshotGarbageCollection())
845 log.Info("session persistence enabled", "path", storePath)
846 daemonOpts = append(daemonOpts, daemon.WithCatalogue(opened.Catalogue, opened.Records))
847
848 // Construct the catalogue-backed expected-session registry before socket
849 // publication. Phase 3 snapshot restoration remains asynchronous in Serve.
850 identity, err := daemonidentity.LoadOrCreate(stateDir)
851 if err != nil {
852 return fmt.Errorf("vev: establish daemon identity: %w", err)
853 }
854 incarnation, err := daemonidentity.NewIncarnation()
855 if err != nil {
856 return fmt.Errorf("vev: establish daemon incarnation: %w", err)
857 }
858 binding, err := daemonmux.NewServerBindings(identity, incarnation, []daemonmux.ServerPolicyAdmission{
859 {Policy: localDaemonPolicy(), Origin: ports.SessionOriginLocal},
860 {Policy: remoteBrokerPolicy(hostTransportQUIC), Origin: ports.SessionOriginRemote},
861 {Policy: remoteBrokerPolicy(hostTransportSSH), Origin: ports.SessionOriginRemote},
862 })
863 if err != nil {
864 return fmt.Errorf("vev: establish daemon mux authority: %w", err)
865 }
866 aggregate := daemonmux.NewAggregateListener()
867 muxSupervisor, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 0)
868 if err != nil {
869 return err
870 }
871 defer func() { retErr = errors.Join(retErr, muxSupervisor.Close()) }()
872 muxListener, err := ipc.ListenMux(daemonmux.SocketPath(ipc.SocketDir()))
873 if err != nil {
874 return fmt.Errorf("vev: listen daemon mux: %w", err)
875 }
876 defer func() { retErr = errors.Join(retErr, muxListener.Close()) }()
877 go func() {
878 for {
879 raw, acceptErr := muxListener.Accept()
880 if acceptErr != nil {
881 return
882 }
883 go func() { _ = muxSupervisor.Adopt(ctx, raw) }()
884 }
885 }()
886
887 // The catalogue-backed registry exists before the socket is published.
888 // Snapshot GC runs in the background once restoration finishes, so it never
889 // delays socket publication.
890 d := daemon.New(pty.NewFactory(), clk, log, daemonOpts...)
891 ln, err := listenDaemon(ipc.SocketDir(), observer)
892 if err != nil {
893 closeErr := opened.Catalogue.Close()
894 log.Error("daemon listen failed", "socket_dir", ipc.SocketDir(), "err", err)
895 return errors.Join(fmt.Errorf("vev: daemon listen: %w", err), closeErr)
896 }
897 defer func() { _ = ln.Close() }()
898 log.Info("daemon starting", "socket", ln.Addr())
899
900 watchCtx, stopWatch := context.WithCancel(ctx)
901 defer stopWatch()
902 go func() {
903 if err := config.Watch(watchCtx, clk, configPath, func(cfg domain.Config, warnings []domain.Warning) {
904 logConfigWarnings(log, warnings)
905 d.ApplyConfig(cfg)
906 }); err != nil && !errors.Is(err, context.Canceled) {
907 log.Warn("config watcher stopped", "path", configPath, "err", err)
908 }
909 }()
910 if err := aggregate.Register(sessionwire.NewServerListener(ln)); err != nil {
911 return fmt.Errorf("vev: register local session listener: %w", err)
912 }
913 if err := d.Serve(ctx, aggregate); err != nil {
914 log.Error("daemon exited", "err", err)
915 return err
916 }
917 log.Info("daemon exited cleanly")
918 return nil
919}
920
921// runAttach dials (auto-spawning the daemon if needed) and runs the client
922// attach loop. Logging goes to the shared file: the client must never write
923// to the console while the terminal is raw.
924func runAttach(ctx context.Context, intent uint8, name, remoteTarget string) (retErr error) {
925 if os.Getenv("VEV") != "" {
926 if remoteTarget == "" && intent == protocol.IntentNew {
927 return createDetachedTerminalSession(ctx, name)
928 }
929 return errors.New("vev: sessions should be nested with care; unset VEV to force")
930 }
931 ctx, stop := clientSignalContext(ctx)
932 defer stop()
933 log, logCloser, err := configureLogging(logging.Client, false)
934 if err != nil {
935 return err
936 }
937 defer func() { _ = logCloser.Close() }()
938 clk := clock.New()
939 observer, observerCloser, err := newPerformanceTrace(clk)
940 if err != nil {
941 return fmt.Errorf("vev: performance trace: %w", err)
942 }
943 if observerCloser != nil {
944 defer func() { retErr = errors.Join(retErr, observerCloser.Close()) }()
945 }
946 if observer != nil {
947 observer.ObserveRuntime(ports.NewRuntimeMark("client", ports.RuntimeTransportDiagnostic, 0, true))
948 }
949 terminal := terminalForAttach()
950 intent, preconnected, err := resolveAttachCreationIntent(ctx, intent, name, remoteTarget, terminal)
951 if err != nil {
952 return err
953 }
954 navigation, resolver, err := terminalBrokerNavigation(intent, name, remoteTarget)
955 if err != nil {
956 if preconnected != nil {
957 _ = preconnected.Close()
958 }
959 return err
960 }
961 callbacks := terminalBrokerCallbacks()
962 return runBrokerClient(ctx, brokerClientConfig{
963 Logger: log, Connector: withPreconnected(newProductionBrokerConnector(), preconnected), Terminal: terminal, Clock: clk,
964 InitialNavigation: navigation, ResolveInitialNavigation: resolver,
965 AttachmentEnvironment: outerTerminalAttachmentEnvironment(log), SessionEnvironment: terminalSessionEnvironment(),
966 OnState: callbacks.OnState, OnLifecycle: callbacks.OnLifecycle, OnFailure: callbacks.OnFailure,
967 })
968}
969
970const (
971 uiRemoteCleanupCommand = "_ui-cleanup"
972 // daemonStopCommand is the hidden, environment-scoped graceful stop of
973 // this environment's daemon, for harnesses and scripts. Users run
974 // `kill --all` instead.
975 daemonStopCommand = "_daemon-stop"
976)
977
978const daemonStopTimeout = 2 * time.Second
979
980var (
981 errDaemonNotRunning = errors.New("vev: no daemon running")
982 errKillOutcomeUnknown = errors.New("vev: kill outcome unknown")
983)
984
985func createDetachedLocalSession(ctx context.Context, name string) error {
986 service, err := connectBroker(ctx)
987 if err != nil {
988 return fmt.Errorf("%w: %v", errDaemonUnreachable, err)
989 }
990 defer func() { _ = service.Close() }()
991 operations, err := client.NewBrokerOperations(service, clock.New())
992 if err != nil {
993 return err
994 }
995 identity, _ := parseVEVEnv(os.Getenv("VEV"))
996 result, err := operations.Command(ctx, localBrokerOperationRoute(service.Snapshot()), protocol.CommandRequest{
997 Version: protocol.Version, Slug: "new-session", Args: []string{name}, TargetSession: identity.session,
998 })
999 if err != nil {
1000 return err
1001 }
1002 return renderDetachedCreationResult(result)
1003}
1004
1005func renderDetachedCreationResult(result protocol.CommandResult) error {
1006 return renderCommandResult(io.Discard, result)
1007}
1008
1009func runUIRemoteCleanup(ctx context.Context) error {
1010 root := os.Getenv("VEV_ENV_ROOT")
1011 if !filepath.IsAbs(root) || filepath.Clean(root) == string(filepath.Separator) {
1012 return errors.New("vev: remote UI cleanup requires a private absolute launch root")
1013 }
1014 if runtimeDir := os.Getenv("XDG_RUNTIME_DIR"); runtimeDir == "" || runtimeDir != filepath.Join(root, "runtime") {
1015 return errors.New("vev: remote UI cleanup requires the launch runtime directory")
1016 }
1017 if err := requestDaemonStop(ctx); err != nil && !errors.Is(err, errDaemonNotRunning) {
1018 return err
1019 }
1020 return nil
1021}
1022
1023func requestDaemonStop(ctx context.Context) error {
1024 ctx, cancel := context.WithTimeout(ctx, daemonStopTimeout)
1025 defer cancel()
1026 service, err := connectDaemonStopBroker(ctx)
1027 if err != nil {
1028 if errors.Is(err, errBrokerAbsent) {
1029 return stopDaemonWithoutBroker(ctx)
1030 }
1031 return unreachableBrokerError(err)
1032 }
1033 defer func() { _ = service.Close() }()
1034 operations, err := client.NewBrokerOperations(service, clock.New())
1035 if err != nil {
1036 return err
1037 }
1038 result, err := operations.StopDaemon(ctx, localBrokerOperationRoute(service.Snapshot()))
1039 if err != nil {
1040 return err
1041 }
1042 return brokerKillResultError(result)
1043}
1044
1045func unreachableBrokerError(err error) error {
1046 return &exitCoded{code: 3, err: fmt.Errorf("%w: %v", errDaemonUnreachable, err)}
1047}
1048
1049func brokerKillResultError(result protocol.KillResult) error {
1050 switch result.Outcome {
1051 case protocol.KillSucceeded:
1052 return nil
1053 case protocol.KillFailed:
1054 if result.Text == "" {
1055 result.Text = "kill failed"
1056 }
1057 return errors.New(result.Text)
1058 default:
1059 if result.Text == "" {
1060 result.Text = "daemon did not report a final outcome"
1061 }
1062 return &exitCoded{code: 3, err: fmt.Errorf("%w: %s", errKillOutcomeUnknown, result.Text)}
1063 }
1064}
1065
1066// gracefulTeardownWaiter is implemented by carriages (QUIC) whose Close defers
1067// a bounded connection close past its return. The broker mux QUIC proxy owns its
1068// process, so it must wait for that close before returning: exiting first
1069// would discard the final synchronous envelope Close already handed to the
1070// stream.
1071type gracefulTeardownWaiter interface {
1072 WaitGracefulTeardown(ctx context.Context) error
1073}
1074
1075// proxyTeardownTimeout bounds the proxy's wait for a deferred graceful close,
1076// leaving headroom beyond the adapter's own graceful window. A carriage that
1077// does not defer its close is not waited on.
1078const proxyTeardownTimeout = 2 * time.Second
1079
1080func waitGracefulTeardown(transport wire.Transport) error {
1081 waiter, ok := transport.(gracefulTeardownWaiter)
1082 if !ok {
1083 return nil
1084 }
1085 ctx, cancel := context.WithTimeout(context.Background(), proxyTeardownTimeout)
1086 defer cancel()
1087 return waiter.WaitGracefulTeardown(ctx)
1088}
1089
1090// runList reads existing state without starting either background process.
1091func runList(ctx context.Context, cmd command) error {
1092 service, err := connectListBroker(ctx)
1093 if errors.Is(err, errBrokerAbsent) {
1094 return runListWithoutBroker(ctx, cmd, os.Stdout)
1095 }
1096 if err != nil {
1097 return unreachableBrokerError(err)
1098 }
1099 defer func() { _ = service.Close() }()
1100 if cmd.listAll || cmd.listHost != "" {
1101 return runBrokerSnapshotList(cmd, service.Snapshot(), os.Stdout)
1102 }
1103 operations, err := client.NewBrokerOperations(service, clock.New())
1104 if err != nil {
1105 return err
1106 }
1107 sessions, err := operations.List(ctx, localBrokerOperationRoute(service.Snapshot()))
1108 if err != nil {
1109 var brokerErr ports.BrokerError
1110 if errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorNoDaemon {
1111 printSessions(os.Stdout, nil)
1112 return nil
1113 }
1114 return err
1115 }
1116 printSessions(os.Stdout, sessions)
1117 return nil
1118}
1119
1120var connectListBroker = connectExistingBroker
1121var dialListDaemon = ipc.DialContext
1122
1123// errNoLocalDaemon reports that no local daemon socket answers; nothing was
1124// started to find out.
1125var errNoLocalDaemon = errors.New("vev: no local daemon")
1126
1127// exchangeLocalDaemon sends one request directly to an existing local daemon
1128// and returns its reply. It is the broker-less path of read and stop
1129// commands: the broker being absent is the normal idle state, and these
1130// commands must stay true without starting it. It never spawns a daemon.
1131func exchangeLocalDaemon(ctx context.Context, request protocol.ClientMessage) (protocol.ServerMessage, error) {
1132 transport, err := dialListDaemon(ctx, ipc.SocketDir())
1133 if err != nil {
1134 if backendAbsent(err) {
1135 return nil, errNoLocalDaemon
1136 }
1137 return nil, err
1138 }
1139 connection := sessionwire.NewClientConnection(transport)
1140 defer func() { _ = connection.Close() }()
1141 if err := connection.SendClient(request); err != nil {
1142 return nil, err
1143 }
1144 return connection.ReceiveServer()
1145}
1146
1147func runListWithoutBroker(ctx context.Context, cmd command, out io.Writer) error {
1148 if cmd.listHost != "" {
1149 _, err := fmt.Fprintf(out, "%s: not connected (no broker running)\n", cmd.listHost)
1150 return err
1151 }
1152 reply, err := exchangeLocalDaemon(ctx, protocol.List{})
1153 switch {
1154 case errors.Is(err, errNoLocalDaemon):
1155 printSessions(out, nil)
1156 case err != nil:
1157 return fmt.Errorf("vev: list local sessions: %w", err)
1158 default:
1159 sessions, ok := reply.(protocol.Sessions)
1160 if !ok {
1161 return fmt.Errorf("vev: unexpected local list reply %T", reply)
1162 }
1163 printSessions(out, sessions.Sessions)
1164 }
1165 if cmd.listAll {
1166 return listDisconnectedHosts(out)
1167 }
1168 return nil
1169}
1170
1171// stopDaemonWithoutBroker asks an existing local daemon to stop directly. A
1172// daemon routinely outlives the idle broker, so "no broker" never means "no
1173// daemon".
1174func stopDaemonWithoutBroker(ctx context.Context) error {
1175 reply, err := exchangeLocalDaemon(ctx, protocol.Kill{RequestID: 1, Scope: protocol.KillDaemon})
1176 if errors.Is(err, errNoLocalDaemon) {
1177 return errDaemonNotRunning
1178 }
1179 if err != nil {
1180 return fmt.Errorf("vev: stop local daemon: %w", err)
1181 }
1182 result, ok := reply.(protocol.KillResult)
1183 if !ok {
1184 return fmt.Errorf("vev: unexpected daemon stop reply %T", reply)
1185 }
1186 return brokerKillResultError(result)
1187}
1188
1189// printSessions renders a session table (or a friendly note when empty).
1190func printSessions(w io.Writer, sessions []protocol.SessionInfo) {
1191 if len(sessions) == 0 {
1192 _, _ = fmt.Fprintln(w, "no sessions")
1193 return
1194 }
1195 tw := tabwriter.NewWriter(w, 0, 4, 2, ' ', 0)
1196 _, _ = fmt.Fprintln(tw, "NAME\tSTATE\tTABS\tATTACHED")
1197 for _, s := range sessions {
1198 state := "up"
1199 tabs := fmt.Sprintf("%d", s.Tabs)
1200 attached := "no"
1201 switch s.State {
1202 case protocol.SessionDown:
1203 state = "down"
1204 tabs = "-"
1205 case protocol.SessionBroken:
1206 state = "broken"
1207 tabs = "-"
1208 default:
1209 if s.Ephemeral {
1210 state = "temporary"
1211 }
1212 }
1213 if s.Attached {
1214 attached = "yes"
1215 }
1216 _, _ = fmt.Fprintf(tw, "%s\t%s\t%s\t%s\n", s.Name, state, tabs, attached)
1217 }
1218 _ = tw.Flush()
1219}
1220
1221func unreadableCatalogueError(stateDir string) error {
1222 return fmt.Errorf("%w: vev: durable session state at %s cannot be read and was left untouched.\n"+
1223 "vev does not erase it automatically. To start fresh, remove it:\n"+
1224 " rm -rf %s", persist.ErrCatalogueUnreadable, stateDir, stateDir)
1225}
1226
1227// runKill deletes one session or, with sessions set, every session through the
1228// broker, starting it if needed.
1229func runKill(ctx context.Context, name string, sessions bool) error {
1230 service, err := connectBroker(ctx)
1231 if err != nil {
1232 return unreachableBrokerError(err)
1233 }
1234 defer func() { _ = service.Close() }()
1235 operations, err := client.NewBrokerOperations(service, clock.New())
1236 if err != nil {
1237 return err
1238 }
1239 route := localBrokerOperationRoute(service.Snapshot())
1240 var result protocol.KillResult
1241 if sessions {
1242 result, err = operations.KillAll(ctx, route)
1243 } else {
1244 result, err = operations.Kill(ctx, route, name)
1245 }
1246 if err != nil {
1247 return err
1248 }
1249 if err := brokerKillResultError(result); err != nil {
1250 return err
1251 }
1252 if sessions {
1253 fmt.Println("killed all sessions")
1254 } else {
1255 fmt.Printf("killed %s\n", name)
1256 }
1257 return nil
1258}