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