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